From 7b969fded994cb7452735630d32e42ea72854df7 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Wed, 1 Jul 2026 15:51:45 -0300 Subject: [PATCH] feat(lock): distributed lock seam + pg_advisory_lock for a remote backend (spec 21) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- internal/lock/distlock.go | 52 ++++++++++ internal/lock/distlock_test.go | 45 +++++++++ internal/lock/pglock.go | 108 ++++++++++++++++++++ internal/lock/pglock_test.go | 173 +++++++++++++++++++++++++++++++++ internal/provision/advisory.go | 59 +++++++++++ 5 files changed, 437 insertions(+) create mode 100644 internal/lock/distlock.go create mode 100644 internal/lock/distlock_test.go create mode 100644 internal/lock/pglock.go create mode 100644 internal/lock/pglock_test.go create mode 100644 internal/provision/advisory.go diff --git a/internal/lock/distlock.go b/internal/lock/distlock.go new file mode 100644 index 0000000..82a9cee --- /dev/null +++ b/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) +} diff --git a/internal/lock/distlock_test.go b/internal/lock/distlock_test.go new file mode 100644 index 0000000..2cfa875 --- /dev/null +++ b/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") + } +} diff --git a/internal/lock/pglock.go b/internal/lock/pglock.go new file mode 100644 index 0000000..8098ac1 --- /dev/null +++ b/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() +} diff --git a/internal/lock/pglock_test.go b/internal/lock/pglock_test.go new file mode 100644 index 0000000..f6f0dc8 --- /dev/null +++ b/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) + } +} diff --git a/internal/provision/advisory.go b/internal/provision/advisory.go new file mode 100644 index 0000000..cc754e4 --- /dev/null +++ b/internal/provision/advisory.go @@ -0,0 +1,59 @@ +package provision + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" + + "github.com/open-source-cloud/devstack/internal/lock" +) + +// AdvisoryConn is a pgx/v5 session implementing lock.AdvisoryConn for the +// distributed pg_advisory_lock (spec 21, Q-REMOTE-LOCK). It is the production +// backing for internal/lock's PGLocker — internal/lock stays a pgx-free leaf, and +// the concrete session lives here where pgx already does. +// +// It MUST be a DIRECT session: pg_advisory_lock is session-scoped, so a +// transaction-pooling pgbouncer in front of the cluster silently defeats it (the +// lock and its release can land on different pooled backends). Point the DSN at a +// direct/session endpoint. +type AdvisoryConn struct{ conn *pgx.Conn } + +// ConnectAdvisory opens a dedicated pgx session to dsn for advisory locking. +func ConnectAdvisory(ctx context.Context, dsn string) (*AdvisoryConn, error) { + c, err := pgx.Connect(ctx, dsn) + if err != nil { + return nil, fmt.Errorf("connect advisory session: %w", err) + } + return &AdvisoryConn{conn: c}, nil +} + +// TryLock runs `SELECT pg_try_advisory_lock($1)` and reports whether the session +// now holds the lock (non-blocking). +func (a *AdvisoryConn) TryLock(ctx context.Context, key int64) (bool, error) { + var ok bool + if err := a.conn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", key).Scan(&ok); err != nil { + return false, err + } + return ok, nil +} + +// Unlock releases the session's advisory lock for key. +func (a *AdvisoryConn) Unlock(ctx context.Context, key int64) error { + _, err := a.conn.Exec(ctx, "SELECT pg_advisory_unlock($1)", key) + return err +} + +// Close ends the session (releasing any lock it still holds — the crash-safe +// backstop that makes advisory locks self-heal on a dead holder). +func (a *AdvisoryConn) Close(ctx context.Context) error { return a.conn.Close(ctx) } + +// PGLockConnector returns a lock.PGConnector that opens a FRESH advisory session +// per acquisition against dsn — the session-per-critical-section discipline +// PGLocker relies on so a lock never outlives its section. +func PGLockConnector(dsn string) lock.PGConnector { + return func(ctx context.Context) (lock.AdvisoryConn, error) { + return ConnectAdvisory(ctx, dsn) + } +}