feat(lock): distributed lock seam + pg_advisory_lock for a remote backend (spec 21)

The concurrency spine's local gofrs/flock cannot serialize two developers on two
machines mutating the SAME remote backend's ledger rows / provisioning / port
allocation — the central unsolved problem for a team/cloud shared backend
(spec 21). This lands the correctness gate: a Locker seam whose remote
implementation is a session-scoped pg_advisory_lock on the shared cluster
Postgres (Q-REMOTE-LOCK RESOLVED — the DB we already run is the coordinator, so
zero new infra and no daemon; session-scoped means it auto-releases on a crash so
another machine reconciles cleanly).

- internal/lock/distlock.go: the Locker interface, FileLocker (today's flock,
  verbatim), and LockerFor(remote, subject, path, connect) — remote+connector →
  PGLocker, else the local flock (a remote backend with no reachable cluster yet
  degrades safely to single-machine correctness). internal/lock stays a
  dependency-free leaf (plain bool, no docker import → no cycle).
- internal/lock/pglock.go: PGLocker takes a session-scoped pg_advisory_lock keyed
  by AdvisoryKey (FNV-64a subject hash → deterministic bigint, so every client of
  one cluster hashes the same subject to the same key). It polls
  pg_try_advisory_lock (not the blocking form) to honor ctx cancel/timeout, and
  opens a fresh session per WithLock so the lock never outlives its section.
  AdvisoryConn is the injectable session seam (transaction-pooling pgbouncer
  breaks session advisory locks — documented).
- internal/provision/advisory.go: the pgx/v5-backed AdvisoryConn + PGLockConnector
  (the production session), kept out of internal/lock to preserve the leaf.

Tests (offline, no live PG): AdvisoryKey determinism, acquire/run/release order,
fn-error release, connect/try errors, ctx-timeout on a held key, LockerFor
selection, FileLocker, and a 50-goroutine mutual-exclusion test over an in-memory
advisory server (compare-and-set) run under -race. This is the foundation; wiring
the orchestrator's ~25 lock call sites through the selected Locker + reaching the
remote cluster over an SSH forward are the follow-ups.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: gustavobertoi

internal/lock/distlock.go

@@ -0,0 +1,52 @@
+package lock
+
+import "context"
+
+// Locker serializes every mutation of the machine-global ledger or the shared
+// Docker stack. It is the seam that lets the concurrency spine span a REMOTE
+// team backend (spec 21): the local implementation is the gofrs/flock advisory
+// lock (a single machine); the remote implementation is a session-scoped
+// pg_advisory_lock on the shared cluster Postgres, which serializes across
+// MACHINES against one backend.
+//
+// This exists because a local flock cannot serialize two developers on two
+// laptops mutating the SAME remote ledger rows / provisioning / port allocation
+// (spec 21 §"the central unsolved problem"). Q-REMOTE-LOCK is RESOLVED: use
+// pg_advisory_lock on the cluster DB as the coordinator — zero new infra,
+// crash-safe (session-scoped → auto-released on disconnect), reusing pgx.
+type Locker interface {
+	// WithLock runs fn while holding the lock, releasing it before returning
+	// (on success or error). It blocks until the lock is held or ctx is done.
+	WithLock(ctx context.Context, fn func() error) error
+}
+
+// FileLocker is the LOCAL Locker: the coarse gofrs/flock at a lockfile path
+// (today's behavior, verbatim). An empty Path runs fn unlocked (tests).
+type FileLocker struct{ Path string }
+
+// NewFileLocker returns a FileLocker bound to a lockfile path (under XDG_RUNTIME_DIR).
+func NewFileLocker(path string) FileLocker { return FileLocker{Path: path} }
+
+// WithLock takes the flock at Path, runs fn, and releases it.
+func (f FileLocker) WithLock(ctx context.Context, fn func() error) error {
+	if f.Path == "" {
+		return fn()
+	}
+	return WithLock(ctx, f.Path, fn)
+}
+
+// LockerFor selects the concurrency primitive for a backend. A remote backend
+// with a reachable cluster Postgres (connect != nil) serializes across machines
+// on a pg_advisory_lock keyed by subject; everything else uses the local flock
+// at path. connect is nil until remote host-reachability is wired, so a remote
+// backend degrades safely to the local flock (single-machine correctness holds;
+// cross-machine serialization is the follow-up that needs the cluster reachable).
+//
+// The parameter is a plain bool, not a docker.Backend, so internal/lock stays a
+// dependency-free leaf (no import cycle).
+func LockerFor(remote bool, subject, path string, connect PGConnector) Locker {
+	if remote && connect != nil {
+		return NewPGLocker(subject, connect)
+	}
+	return NewFileLocker(path)
+}

internal/lock/distlock_test.go

@@ -0,0 +1,45 @@
+package lock
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+)
+
+func TestFileLocker_RunsFn(t *testing.T) {
+	l := NewFileLocker(filepath.Join(t.TempDir(), "x.lock"))
+	ran := false
+	if err := l.WithLock(context.Background(), func() error { ran = true; return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if !ran {
+		t.Fatal("fn did not run under the file lock")
+	}
+}
+
+func TestFileLocker_EmptyPathRunsUnlocked(t *testing.T) {
+	ran := false
+	if err := (FileLocker{}).WithLock(context.Background(), func() error { ran = true; return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if !ran {
+		t.Fatal("empty-path FileLocker should run fn unlocked")
+	}
+}
+
+func TestLockerFor_Selects(t *testing.T) {
+	connect := func(context.Context) (AdvisoryConn, error) { return nil, nil }
+
+	// Remote + a connector → distributed pg lock.
+	if _, ok := LockerFor(true, "subj", "/tmp/x.lock", connect).(*PGLocker); !ok {
+		t.Error("remote + connector should select *PGLocker")
+	}
+	// Remote but no connector (reachability not wired) → safe local fallback.
+	if _, ok := LockerFor(true, "subj", "/tmp/x.lock", nil).(FileLocker); !ok {
+		t.Error("remote + nil connector should fall back to FileLocker")
+	}
+	// Local → file lock.
+	if _, ok := LockerFor(false, "subj", "/tmp/x.lock", connect).(FileLocker); !ok {
+		t.Error("local should select FileLocker")
+	}
+}

internal/lock/pglock.go

@@ -0,0 +1,108 @@
+package lock
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"hash/fnv"
+	"time"
+)
+
+// This file is the DISTRIBUTED lock (spec 21, Q-REMOTE-LOCK RESOLVED): a
+// session-scoped pg_advisory_lock on the shared cluster Postgres. It serializes
+// two developers on two machines against ONE remote backend — the guarantee a
+// local flock structurally cannot provide.
+//
+// Why pg_advisory_lock and not a coordinator daemon: the team Postgres we already
+// run IS the coordinator (zero new infra), the lock is session-scoped so it is
+// auto-released when the holder disconnects or crashes (a kill -9 on one machine
+// lets another machine's next command reconcile cleanly), and it reuses the pgx
+// path provisioning already uses. Gotcha (spec 21): advisory locks live on a
+// SESSION — a transaction-pooling pgbouncer silently breaks them, so the
+// connector must hand out a direct/session connection, not a pooled one.
+
+// advisoryRetry is how often PGLocker re-polls pg_try_advisory_lock while waiting,
+// mirroring the flock's TryLockContext cadence.
+const advisoryRetry = 100 * time.Millisecond
+
+// AdvisoryConn is the minimal session-connection surface PGLocker needs. It MUST
+// be a single dedicated Postgres session (never a pool / transaction-mode
+// pgbouncer). Injectable so the locker is unit-testable without a live server.
+type AdvisoryConn interface {
+	// TryLock runs `SELECT pg_try_advisory_lock($1)` and reports whether the lock
+	// was granted on this session (non-blocking).
+	TryLock(ctx context.Context, key int64) (bool, error)
+	// Unlock runs `SELECT pg_advisory_unlock($1)` for this session.
+	Unlock(ctx context.Context, key int64) error
+	// Close ends the session (a crash-safe backstop that releases any held lock).
+	Close(ctx context.Context) error
+}
+
+// PGConnector opens a fresh AdvisoryConn (a dedicated session). PGLocker opens
+// one per WithLock and closes it after, so the advisory lock never outlives the
+// critical section even if a release is missed.
+type PGConnector func(ctx context.Context) (AdvisoryConn, error)
+
+// PGLocker is the distributed Locker. It hashes Subject to a deterministic
+// bigint key and takes a session-scoped pg_advisory_lock on it.
+type PGLocker struct {
+	Subject string
+	Connect PGConnector
+	// retry is injectable for tests; 0 → advisoryRetry.
+	retry time.Duration
+}
+
+// NewPGLocker returns a PGLocker for a lock subject and a session connector.
+func NewPGLocker(subject string, connect PGConnector) *PGLocker {
+	return &PGLocker{Subject: subject, Connect: connect}
+}
+
+// AdvisoryKey maps a lock subject to the deterministic int64 pg_advisory_lock
+// key: FNV-64a over the subject, reinterpreted as a signed bigint. Deterministic
+// across machines and runs, so every client of one cluster hashes the same
+// subject to the same key. A hash collision only causes extra (safe)
+// serialization between two subjects — never a missed exclusion.
+func AdvisoryKey(subject string) int64 {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(subject))
+	return int64(h.Sum64())
+}
+
+// WithLock opens a dedicated session, polls pg_try_advisory_lock until the key is
+// held or ctx is done, runs fn, then releases the lock and closes the session.
+// Polling (rather than the blocking pg_advisory_lock) honors ctx cancellation and
+// timeouts instead of parking a backend indefinitely.
+func (p *PGLocker) WithLock(ctx context.Context, fn func() error) error {
+	if p.Connect == nil {
+		return errors.New("pg advisory lock: no session connector configured")
+	}
+	retry := p.retry
+	if retry == 0 {
+		retry = advisoryRetry
+	}
+	key := AdvisoryKey(p.Subject)
+
+	conn, err := p.Connect(ctx)
+	if err != nil {
+		return fmt.Errorf("open advisory-lock session: %w", err)
+	}
+	defer func() { _ = conn.Close(ctx) }()
+
+	for {
+		ok, err := conn.TryLock(ctx, key)
+		if err != nil {
+			return fmt.Errorf("acquire advisory lock: %w", err)
+		}
+		if ok {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("timed out waiting for the shared-backend lock (subject %q); another machine may be holding it: %w", p.Subject, ctx.Err())
+		case <-time.After(retry):
+		}
+	}
+	// Held. Release (Unlock) runs before Close because deferred calls are LIFO.
+	defer func() { _ = conn.Unlock(ctx, key) }()
+	return fn()
+}

internal/lock/pglock_test.go

@@ -0,0 +1,173 @@
+package lock
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestAdvisoryKeyDeterministic(t *testing.T) {
+	subject := "devstack-shared-ledger"
+	first, second := AdvisoryKey(subject), AdvisoryKey(subject)
+	if first != second {
+		t.Fatalf("same subject must hash to the same key (%d != %d)", first, second)
+	}
+	if AdvisoryKey("a") == AdvisoryKey("b") {
+		t.Fatal("different subjects should (almost surely) differ")
+	}
+}
+
+// fakeAdvisory is an in-memory model of ONE cluster's advisory-lock table: a
+// single map[key]bool guarded by a real mutex, so concurrent PGLockers exercise
+// genuine mutual exclusion (compare-and-set), like the real server session.
+type fakeAdvisory struct {
+	mu   sync.Mutex
+	held map[int64]bool
+}
+
+func newFakeAdvisory() *fakeAdvisory { return &fakeAdvisory{held: map[int64]bool{}} }
+
+// session is one connection to the fake server.
+type fakeSession struct {
+	srv    *fakeAdvisory
+	closed bool
+	// hooks to simulate failures
+	tryErr   error
+	closeErr error
+}
+
+func (s *fakeSession) TryLock(_ context.Context, key int64) (bool, error) {
+	if s.tryErr != nil {
+		return false, s.tryErr
+	}
+	s.srv.mu.Lock()
+	defer s.srv.mu.Unlock()
+	if s.srv.held[key] {
+		return false, nil
+	}
+	s.srv.held[key] = true
+	return true, nil
+}
+
+func (s *fakeSession) Unlock(_ context.Context, key int64) error {
+	s.srv.mu.Lock()
+	defer s.srv.mu.Unlock()
+	delete(s.srv.held, key)
+	return nil
+}
+
+func (s *fakeSession) Close(context.Context) error {
+	s.closed = true
+	return s.closeErr
+}
+
+func (f *fakeAdvisory) connector() PGConnector {
+	return func(context.Context) (AdvisoryConn, error) {
+		return &fakeSession{srv: f}, nil
+	}
+}
+
+func TestPGLocker_AcquireRunRelease(t *testing.T) {
+	srv := newFakeAdvisory()
+	l := &PGLocker{Subject: "s", Connect: srv.connector()}
+	ran := false
+	if err := l.WithLock(context.Background(), func() error {
+		ran = true
+		// While fn runs, the key must be held.
+		srv.mu.Lock()
+		defer srv.mu.Unlock()
+		if !srv.held[AdvisoryKey("s")] {
+			t.Error("key should be held while fn runs")
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if !ran {
+		t.Fatal("fn did not run")
+	}
+	// Released after WithLock returns.
+	if srv.held[AdvisoryKey("s")] {
+		t.Error("key should be released after WithLock")
+	}
+}
+
+func TestPGLocker_PropagatesFnError(t *testing.T) {
+	srv := newFakeAdvisory()
+	l := &PGLocker{Subject: "s", Connect: srv.connector()}
+	sentinel := errors.New("boom")
+	if err := l.WithLock(context.Background(), func() error { return sentinel }); !errors.Is(err, sentinel) {
+		t.Fatalf("want fn error propagated, got %v", err)
+	}
+	if srv.held[AdvisoryKey("s")] {
+		t.Error("key must be released even when fn errors")
+	}
+}
+
+func TestPGLocker_NoConnector(t *testing.T) {
+	l := &PGLocker{Subject: "s"}
+	if err := l.WithLock(context.Background(), func() error { return nil }); err == nil {
+		t.Fatal("want error when no connector is configured")
+	}
+}
+
+func TestPGLocker_ConnectError(t *testing.T) {
+	l := &PGLocker{Subject: "s", Connect: func(context.Context) (AdvisoryConn, error) {
+		return nil, errors.New("dial refused")
+	}}
+	ran := false
+	err := l.WithLock(context.Background(), func() error { ran = true; return nil })
+	if err == nil || ran {
+		t.Fatalf("connect error must abort before fn (err=%v ran=%v)", err, ran)
+	}
+}
+
+func TestPGLocker_CtxTimeoutWhenHeld(t *testing.T) {
+	srv := newFakeAdvisory()
+	// Pre-hold the key so TryLock always returns false.
+	srv.held[AdvisoryKey("s")] = true
+	l := &PGLocker{Subject: "s", Connect: srv.connector(), retry: time.Millisecond}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
+	defer cancel()
+	ran := false
+	err := l.WithLock(ctx, func() error { ran = true; return nil })
+	if err == nil || ran {
+		t.Fatalf("a permanently-held key must time out before fn (err=%v ran=%v)", err, ran)
+	}
+}
+
+func TestPGLocker_TryLockErrorSurfaces(t *testing.T) {
+	l := &PGLocker{Subject: "s", Connect: func(context.Context) (AdvisoryConn, error) {
+		return &fakeSession{srv: newFakeAdvisory(), tryErr: errors.New("conn reset")}, nil
+	}}
+	if err := l.WithLock(context.Background(), func() error { return nil }); err == nil {
+		t.Fatal("a TryLock error must surface")
+	}
+}
+
+// TestPGLocker_Serializes is the concurrency guarantee: N goroutines each take
+// the lock and do a deliberately non-atomic read-modify-write; mutual exclusion
+// must make the final count exact. Run under -race.
+func TestPGLocker_Serializes(t *testing.T) {
+	srv := newFakeAdvisory()
+	l := &PGLocker{Subject: "ledger", Connect: srv.connector(), retry: time.Millisecond}
+	const n = 50
+	counter := 0
+	var wg sync.WaitGroup
+	for range n {
+		wg.Go(func() {
+			_ = l.WithLock(context.Background(), func() error {
+				c := counter
+				time.Sleep(time.Microsecond) // widen the race window
+				counter = c + 1
+				return nil
+			})
+		})
+	}
+	wg.Wait()
+	if counter != n {
+		t.Fatalf("mutual exclusion failed: counter = %d, want %d", counter, n)
+	}
+}