Unique ID generator

// Next issues the next ID. Under Wait and Fail it never issues an ID at or below the last one.
func (g *Generator) Next() (int64, error) {
  g.mu.Lock()
  defer g.mu.Unlock()
  now := g.clock.NowMS()
  if now < g.last && g.policy != Naive {
    behind := g.last - now
    if g.policy == Fail || behind > g.maxWait {
      return 0, fmt.Errorf("%w: %d ms behind the last ID", ErrClockBackwards, behind)
    }
    for now < g.last { // Wait: sleep until the clock catches up
      g.clock.Sleep(g.last - now)
      g.slept += g.last - now
      now = g.clock.NowMS()
    }
  }
  if now == g.last {
    g.seq++
    if g.seq > MaxSeq { // 4,096 IDs in this ms: wait for the next ms
      for now <= g.last {
        g.clock.Sleep(1)
        g.slept++
        now = g.clock.NowMS()
      }
      g.seq = 0
    }
  } else {
    g.seq = 0
  }
  g.last = now
  return (now-g.epoch)<<(NodeBits+SeqBits) | g.node<<SeqBits | g.seq, nil
}

UPDATE worker_leases w
SET holder = $1, epoch = w.epoch + 1, expires_at = now() + make_interval(secs => $2::float8 / 1000)
FROM (
  SELECT worker_id FROM worker_leases
  WHERE expires_at < now()
  ORDER BY worker_id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
) free
WHERE w.worker_id = free.worker_id
RETURNING w.worker_id, w.epoch, w.ceiling_ms;

UPDATE worker_leases
SET expires_at = now() + make_interval(secs => $3::float8 / 1000),
    ceiling_ms = GREATEST(ceiling_ms, $4)
WHERE worker_id = $1 AND epoch = $2
RETURNING ceiling_ms;

// Acquire takes the first free worker ID: one whose lease has expired. The update adds 1 to the
// epoch and returns the previous holder's ceiling, which becomes this holder's floor. Under the
// fenced mode it waits until its clock passes the floor, then renews to set its own ceiling.
func Acquire(ctx context.Context, db *pgxpool.Pool, c clocks.Clock, holder string, mode Mode, ttlMS, maxWait, epochMS int64) (*Leased, error) {
  l := &Leased{db: db, clock: c, mode: mode, holder: holder, ttlMS: ttlMS, maxWait: maxWait, epochMS: epochMS}
  err := db.QueryRow(ctx, leaseQ["acquire"], holder, ttlMS).Scan(&l.Worker, &l.Epoch, &l.Floor)
  if errors.Is(err, pgx.ErrNoRows) {
    return nil, ErrNoFreeWorker
  }
  if err != nil {
    return nil, fmt.Errorf("acquire a worker ID for %s: %w", holder, err)
  }
  if mode == Fenced {
    if behind := l.Floor - c.NowMS(); behind > maxWait {
      return nil, fmt.Errorf("worker %d: %w: %d ms", l.Worker, ErrFloorTooFar, behind)
    }
    for c.NowMS() <= l.Floor {
      c.Sleep(l.Floor - c.NowMS() + 1)
    }
  }
  if l.gen, err = clocks.NewGenerator(c, l.Worker, clocks.Wait, maxWait, epochMS); err != nil {
    return nil, fmt.Errorf("generator for worker %d: %w", l.Worker, err)
  }
  if err := l.Renew(ctx); err != nil {
    return nil, err
  }
  return l, nil
}

// Renew extends the lease and raises the ceiling to the clock plus the lease time. It matches no
// row once another holder has acquired the worker ID, because that holder raised the epoch.
func (l *Leased) Renew(ctx context.Context) error {
  want := l.clock.NowMS() + l.ttlMS
  var stored int64
  err := l.db.QueryRow(ctx, leaseQ["renew"], l.Worker, l.Epoch, l.ttlMS, want).Scan(&stored)
  if errors.Is(err, pgx.ErrNoRows) {
    return fmt.Errorf("renew worker %d epoch %d: %w", l.Worker, l.Epoch, ErrLeaseLost)
  }
  if err != nil {
    return fmt.Errorf("renew worker %d: %w", l.Worker, err)
  }
  l.mu.Lock()
  l.Ceiling = max(l.Ceiling, want)
  l.mu.Unlock()
  return nil
}

// Next issues an ID. Under the fenced mode an ID whose time is past the ceiling is never
// returned: the holder must renew first, and a renew fails once the worker ID has a new holder.
func (l *Leased) Next() (int64, error) {
  l.mu.Lock()
  defer l.mu.Unlock()
  id, err := l.gen.Next()
  if err != nil {
    return 0, fmt.Errorf("worker %d: %w", l.Worker, err)
  }
  if l.mode == Fenced {
    ms, _, _ := clocks.Decode(id)
    if at := ms + l.epochMS; at > l.Ceiling || at <= l.Floor {
      return 0, fmt.Errorf("worker %d at %d ms, ceiling %d: %w", l.Worker, at, l.Ceiling, ErrPastCeiling)
    }
  }
  return id, nil
}

// Next returns the next ULID. A new millisecond draws 80 fresh random bits; the same or an
// earlier millisecond adds 1 to the last ID, so IDs from one generator always increase.
func (g *ULIDGen) Next() (ULID, error) {
  g.mu.Lock()
  defer g.mu.Unlock()
  now := g.clock.NowMS()
  if now > g.ms {
    var u ULID
    binary.BigEndian.PutUint16(u[0:2], uint16(now>>32))
    binary.BigEndian.PutUint32(u[2:6], uint32(now))
    binary.BigEndian.PutUint16(u[6:8], uint16(g.rng.Uint32()))
    binary.BigEndian.PutUint64(u[8:16], g.rng.Uint64())
    g.last, g.ms = u, now
    return u, nil
  }
  for i := 15; i >= 6; i-- { // add 1 to the 80 random bits
    g.last[i]++
    if g.last[i] != 0 {
      return g.last, nil
    }
  }
  return ULID{}, ErrULIDOverflow
}