diff --git a/internal/collab/manager.go b/internal/collab/manager.go index a600c32a..d854000c 100644 --- a/internal/collab/manager.go +++ b/internal/collab/manager.go @@ -200,6 +200,32 @@ func (m *RoomManager) LastRestoreSeq(itemID string) (int64, bool) { return v, ok } +// invalidateRestoreFences drops the item's IN-MEMORY restore-fence fast-path +// entries — both lastRestoreSeqs and restoreBoundaries — so the next Join +// stale-seed check (manager.go ~L388) and collab-snapshot flush gate +// (handlers_items.go ~L1328) fall through to the DURABLE columns +// (items.last_restore_seq / items.restore_boundary_op_id). +// +// Used by ForceRefreshRoom's UNCERTAIN commit-outcome path (BUG-2276 residual 1). +// After a Postgres restore commit whose landing we couldn't confirm, the caches may +// still hold a PRIOR restore's generation. If THIS restore actually LANDED, a +// cursor-0 peer seeded at the prior generation that reconnects — or a stale +// collab-snapshot PATCH — would be admitted against the cached prior fence and +// clobber the newly-landed content, because both consumers TRUST an in-memory hit +// and skip the durable read. Clearing the fast-path forces the durable read, which +// reflects this restore if it committed (and both consumers already FAIL CLOSED on a +// durable read error). delete() on a nil map is a no-op. MUST be called under +// itemLock so it can't race a concurrent restore's SetRestoreBoundary/SetLastRestoreSeq. +func (m *RoomManager) invalidateRestoreFences(itemID string) { + m.lastRestoreSeqMu.Lock() + delete(m.lastRestoreSeqs, itemID) + m.lastRestoreSeqMu.Unlock() + + m.restoreBoundaryMu.Lock() + delete(m.restoreBoundaries, itemID) + m.restoreBoundaryMu.Unlock() +} + // itemLock returns the lazily-allocated mutex guarding setup-phase // operations on itemID. Locks live in the manager for the lifetime of // the process — for a workspace with many items this is at most a few @@ -987,17 +1013,28 @@ func (m *RoomManager) PruneAndApply(itemID string, applyFn func() error) error { // mid-session auth revalidation loop (which writes canWrite) can neither thaw the // freeze mid-restore nor get its viewer/editor decision clobbered by it. // -// RESIDUAL (BUG-2276, deferred — commit-outcome ambiguity): a commit ERROR is -// treated as "rolled back". On SQLite (the self-host shape) that holds — a commit -// either fsyncs or it doesn't. On Postgres a commit that DURABLY lands but whose -// acknowledgement is lost (connection drop at the commit boundary) surfaces as an -// error here, so we unfreeze+resume peers on a stale Y.Doc even though the DB now -// holds restored content + a pruned op-log — a stale flush could then clobber. -// Un-freezing is the RIGHT call for the common genuine-rollback error (a real -// rollback must not discard peers' unflushed edits), so the fix is not "reseed on -// every error" (that regresses the common case) but commit-outcome reconciliation -// (re-read after a commit error to learn what actually happened). Narrow + -// Postgres-only; tracked in BUG-2276. +// COMMIT-OUTCOME RECONCILIATION (BUG-2276 residual 1): a commit ERROR is +// ambiguous on Postgres but not on SQLite. On SQLite (the self-host shape) a +// commit either fsyncs or it doesn't, so when `reconcile` is nil an error is +// taken as a rollback verbatim (un-freeze + bail, above). On Postgres a commit +// that DURABLY lands but whose acknowledgement is lost (connection drop at the +// commit boundary) ALSO surfaces here as an error; blindly un-freezing would +// resume peers on a stale Y.Doc even though the DB now holds restored content + a +// pruned op-log, and a subsequent stale flush could then clobber. Un-freezing is +// still the RIGHT call for a genuine rollback (a real rollback must not discard +// peers' unflushed edits), so the fix is not "reseed on every error" (that +// regresses the common case) but commit-outcome reconciliation: the Postgres +// caller supplies a `reconcile` callback that RE-READS after a commit error to +// learn what actually happened, yielding three outcomes — (a) LANDED → treat as +// success (publish the boundary + reseed using the durably-stamped boundary/seq; +// do NOT un-freeze); (b) DEFINITELY rolled back → un-freeze + bail (the +// genuine-rollback case); (c) the reconcile read itself failed / outcome +// UNCERTAIN → keep the conns FROZEN and return (never un-freeze onto a +// possibly-stale doc — a frozen peer can't persist and converges cleanly once it +// reconnects/times out, the safe degraded mode). The whole mechanism is gated to +// Postgres by the caller passing `reconcile` ONLY when the store dialect is +// Postgres; with `reconcile == nil` the SQLite/self-host path stays byte-for-byte +// the pre-fix behavior. Tracked in BUG-2276. // // `commit` returns (pre-prune MAX(op-log), restored item.seq): both captured // INSIDE its transaction, so the boundary can't fail-open on a MAX read error @@ -1015,7 +1052,7 @@ func (m *RoomManager) PruneAndApply(itemID string, applyFn func() error) error { // the prune (so it isn't in forceRefreshAll), and (3) a cursor-0 pre-restore tab // that reconnects AFTER a server restart (fenced off the durable column since the // in-memory fast-path is empty then). See the lastRestoreSeqs field doc. -func (m *RoomManager) ForceRefreshRoom(itemID string, commit func() (int64, int64, error)) error { +func (m *RoomManager) ForceRefreshRoom(itemID string, commit func() (int64, int64, error), reconcile func() (RestoreReconcileResult, error)) error { lock := m.itemLock(itemID) lock.Lock() defer lock.Unlock() @@ -1042,34 +1079,87 @@ func (m *RoomManager) ForceRefreshRoom(itemID string, commit func() (int64, int6 // The commit reads the pre-prune MAX(op-log), writes items.content=restored + // the version, and wipes the op-log in ONE transaction, returning (MAX, seq). - // On failure nothing changed on disk; un-freeze the room and bail without - // publishing the boundary or reseeding. - var maxID, restoredSeq int64 + // boundary/restoredSeq are the values the success path publishes: on a clean + // commit they are (MAX+1, seq) straight from the return; on a Postgres commit + // whose ack was lost but whose tx reconciliation proves DID land, they are + // recovered from the durable stamps by `reconcile` (BUG-2276 residual 1). The + // defaults cover the (prod-unused) commit==nil path — the historical + // SetRestoreBoundary(1) / SetLastRestoreSeq(0). + boundary, restoredSeq := int64(1), int64(0) if commit != nil { - m, seq, err := commit() - if err != nil { - if room != nil { - room.mu.Lock() - for _, rc := range room.conns { - rc.frozen.Store(false) - } - room.mu.Unlock() - room.appendMu.Unlock() - } + maxID, seq, err := commit() + switch { + case err == nil: + boundary, restoredSeq = maxID+1, seq + case reconcile == nil: + // SQLite / self-host: a commit error is unambiguous — the ONE tx rolled + // back (items.content, version, op-log wipe, boundary read all together), + // so nothing changed on disk. Un-freeze the room and bail without + // publishing the boundary or reseeding; the room is left exactly as it + // was. (Byte-for-byte the pre-BUG-2276 behavior.) + m.unfreezeAndReleaseAppend(room) return err + default: + // Postgres: the commit reported an error, but a durably-landed commit + // whose ack was lost at the connection boundary ALSO surfaces here. Re-read + // to learn what actually happened before discarding peers' state. + res, rerr := reconcile() + switch { + case rerr != nil: + // (c) UNCERTAIN — the reconcile read itself failed (or a not-found + // re-read: the item may have been archived AFTER a durable-but-ack-lost + // commit), so we cannot tell a genuine rollback from an ack-lost-but- + // landed commit. Un-freezing onto a possibly-stale Y.Doc could let a + // stale flush clobber content that may in fact be committed; but simply + // leaving the conns frozen-and-open would silently drop every subsequent + // edit forever (the collab read path has no WS read-deadline/heartbeat). + // SAFEST: release appendMu (no socket I/O under appendMu), then + // PLAIN-CLOSE the sockets — NOT force_refresh (we don't know + // items.content is authoritative). Each client reconnects and + // re-evaluates the durable restore fences fresh through Join, which is + // correct whichever way the commit actually went. Leave rc.frozen set so + // any already-read frame is still dropped as the readLoop unwinds. + // + // FIRST, still under itemLock (and before releasing appendMu), drop the + // item's IN-MEMORY restore fences: if this restore LANDED, the caches + // still hold a PRIOR restore's generation, and a reconnecting cursor-0 + // peer / stale collab-snapshot would be admitted against that stale cache + // and clobber the landed content. Clearing them forces Join + the snapshot + // gate to consult the DURABLE columns (which reflect this restore if it + // committed, and fail closed on a read error). (BUG-2276 residual 1, P1.) + m.invalidateRestoreFences(itemID) + if room != nil { + room.appendMu.Unlock() + room.closeAllConnsPlain() + } + slog.Error("collab: version-restore commit outcome uncertain after ack loss; froze + plain-closed peers to force a safe reconnect", + "item_id", itemID, "commit_err", err, "reconcile_err", rerr) + return errors.Join(err, rerr) + case !res.Landed: + // (b) definitely rolled back — un-freeze + bail, identical to the + // self-host path. Peers keep their unflushed edits. + m.unfreezeAndReleaseAppend(room) + return err + default: + // (a) landed despite the lost ack — treat as SUCCESS. Recover the + // boundary + restored seq from the durable stamps and fall through to + // the success path (publish + reseed); do NOT un-freeze. + boundary, restoredSeq = res.Boundary, res.Seq + slog.Warn("collab: version-restore commit ack lost but effects landed; reconciled to success", + "item_id", itemID, "boundary", boundary, "restored_seq", restoredSeq, "commit_err", err) + } } - maxID, restoredSeq = m, seq } - // Commit succeeded: items.content=restored, op-log empty. Publish the - // stale-flush boundary = pre-prune MAX+1 (or 1 when empty). IDs are - // AUTOINCREMENT/BIGSERIAL-monotonic across prunes, so every in-flight + // Commit succeeded (or reconciled to landed): items.content=restored, op-log + // empty. Publish the stale-flush boundary = pre-prune MAX+1 (or 1 when empty). + // IDs are AUTOINCREMENT/BIGSERIAL-monotonic across prunes, so every in-flight // snapshot's cursor (≤ pre-prune MAX) is below the boundary and rejected by // the collab-snapshot gate, while every genuine post-refresh op gets an id ≥ // MAX+1 and is accepted. The gate runs under this same itemLock, so no // in-flight snapshot can slip a write between the prune (committed above) and // the boundary becoming visible. - m.SetRestoreBoundary(itemID, maxID+1) + m.SetRestoreBoundary(itemID, boundary) // Record the restored content generation so Join can force_refresh any peer // whose ?content_seq seed predates it (the stale-SEED clobber the op-log-id @@ -1089,6 +1179,39 @@ func (m *RoomManager) ForceRefreshRoom(itemID string, commit func() (int64, int6 return nil } +// RestoreReconcileResult reports what a post-commit-error re-read learned about +// whether a version-restore's transaction DURABLY landed despite a lost commit +// acknowledgement (BUG-2276 residual 1, Postgres-only). Boundary/Seq are valid +// ONLY when Landed is true — they are recovered from the durable restore stamps +// (items.restore_boundary_op_id = pre-prune MAX(op-log.id)+1, and +// items.last_restore_seq = the restored item.seq) so ForceRefreshRoom can reuse +// its normal success path. The producer returns a non-nil error instead (leaving +// this zero) when it cannot tell landed from rolled-back — the UNCERTAIN outcome, +// which keeps the room frozen. +type RestoreReconcileResult struct { + Landed bool + Boundary int64 + Seq int64 +} + +// unfreezeAndReleaseAppend reverses the freeze applied at the top of +// ForceRefreshRoom on a genuine rollback: it clears every conn's frozen flag +// (peers resume editing their live Y.Doc, viewers keep read-only) and releases +// appendMu, leaving the room exactly as it was before the restore attempt. A nil +// room is a no-op (nothing was frozen). MUST be called with appendMu held; it +// preserves the room.mu-inside-appendMu lock order. +func (m *RoomManager) unfreezeAndReleaseAppend(room *Room) { + if room == nil { + return + } + room.mu.Lock() + for _, rc := range room.conns { + rc.frozen.Store(false) + } + room.mu.Unlock() + room.appendMu.Unlock() +} + // closeFrameDeadline is the absolute time budget for sending a // CloseMessage frame via WriteControl before falling through to a // plain Close. Generous enough that a healthy connection always diff --git a/internal/collab/manager_test.go b/internal/collab/manager_test.go index 3687ab78..c5a25a03 100644 --- a/internal/collab/manager_test.go +++ b/internal/collab/manager_test.go @@ -526,7 +526,7 @@ func TestForceRefreshRoom(t *testing.T) { return 0, 0, err } return maxID, 42, nil - }); err != nil { + }, nil); err != nil { t.Fatalf("ForceRefreshRoom: %v", err) } if !committed { @@ -575,6 +575,253 @@ func TestForceRefreshRoom(t *testing.T) { } } +// TestForceRefreshRoomReconcile covers the Postgres commit-outcome reconciliation +// (BUG-2276 residual 1): when the restore commit returns an error, the supplied +// reconcile callback disambiguates three outcomes — landed (take the success +// path), definitely rolled back (un-freeze + bail), and uncertain (stay frozen + +// bail). The SQLite/self-host path (reconcile == nil) is exercised by +// TestForceRefreshRoom above. +func TestForceRefreshRoomReconcile(t *testing.T) { + // (a) The commit's tx DURABLY landed but its ack was lost (Postgres), so the + // commit closure prunes the op-log then returns an error. reconcile reports + // Landed → ForceRefreshRoom must take the SUCCESS path: publish the recovered + // boundary + seq and reseed peers, NOT un-freeze onto a stale doc. + t.Run("landed_takes_success_path", func(t *testing.T) { + bus := NewMemoryOpBus() + defer bus.Close() + store := &fakeOpLog{} + mgr := NewRoomManager(store, bus) + defer mgr.Close() + + srv := newCollabTestServer(t, mgr) + defer srv.Close() + + conn := dialWS(t, srv, "item-a") + defer conn.Close() + for i := 0; i < 200; i++ { + if mgr.RoomCount() == 1 && bus.SubscriberCount("item-a") == 1 { + break + } + time.Sleep(5 * time.Millisecond) + } + + if _, err := store.AppendYjsUpdate("item-a", []byte{0x00, 0x01}, "1"); err != nil { + t.Fatalf("seed: %v", err) + } + + const wantBoundary = int64(8) + const wantSeq = int64(99) + reconciled := false + err := mgr.ForceRefreshRoom("item-a", + func() (int64, int64, error) { + // The tx pruned the op-log and committed durably, then the ack was + // lost at the commit boundary — so it surfaces here as an error. + if _, perr := store.PruneYjsUpdatesBefore("item-a", distantFuture); perr != nil { + return 0, 0, perr + } + return 0, 0, errors.New("commit: driver: bad connection") + }, + func() (RestoreReconcileResult, error) { + reconciled = true + return RestoreReconcileResult{Landed: true, Boundary: wantBoundary, Seq: wantSeq}, nil + }) + if err != nil { + t.Fatalf("ForceRefreshRoom (landed) returned error, want success: %v", err) + } + if !reconciled { + t.Fatal("reconcile callback did not run") + } + if b, ok := mgr.RestoreBoundary("item-a"); !ok || b != wantBoundary { + t.Fatalf("restore boundary = (%d, %v), want (%d, true)", b, ok, wantBoundary) + } + if s, ok := mgr.LastRestoreSeq("item-a"); !ok || s != wantSeq { + t.Fatalf("last restore seq = (%d, %v), want (%d, true)", s, ok, wantSeq) + } + // The success path ran the reseed: the peer got a force_refresh frame. + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + sawForceRefresh := false + for { + mt, data, rerr := conn.ReadMessage() + if rerr != nil { + break + } + if mt == websocket.TextMessage { + var ctl ControlMessage + if json.Unmarshal(data, &ctl) == nil && ctl.Type == ControlMessageForceRefresh { + sawForceRefresh = true + } + } + } + if !sawForceRefresh { + t.Fatal("landed reconcile did not reseed peers (no force_refresh frame)") + } + }) + + // (b) The commit genuinely rolled back. reconcile reports !Landed → + // ForceRefreshRoom un-freezes the conn and returns the commit error; no + // boundary/seq published, peers keep their live Y.Doc. + t.Run("rolled_back_unfreezes", func(t *testing.T) { + bus := NewMemoryOpBus() + defer bus.Close() + mgr := NewRoomManager(&fakeOpLog{}, bus) + defer mgr.Close() + + room := mgr.getOrCreate("item-a") + rc := &roomConn{id: 1, conn: &websocket.Conn{}} + rc.canWrite.Store(true) + if err := room.addConn(rc); err != nil { + t.Fatalf("addConn: %v", err) + } + // Drop the fake conn before Close so closeAll doesn't Close() a nil socket. + defer func() { + room.mu.Lock() + delete(room.conns, rc.conn) + room.mu.Unlock() + }() + + commitErr := errors.New("commit: rolled back") + err := mgr.ForceRefreshRoom("item-a", + func() (int64, int64, error) { return 0, 0, commitErr }, + func() (RestoreReconcileResult, error) { + return RestoreReconcileResult{Landed: false}, nil + }) + if !errors.Is(err, commitErr) { + t.Fatalf("ForceRefreshRoom (rolled back) err = %v, want the commit error", err) + } + if rc.frozen.Load() { + t.Fatal("rolled-back reconcile must UN-freeze the conn, but frozen is still set") + } + if _, ok := mgr.RestoreBoundary("item-a"); ok { + t.Fatal("rolled-back reconcile must NOT publish a restore boundary") + } + if _, ok := mgr.LastRestoreSeq("item-a"); ok { + t.Fatal("rolled-back reconcile must NOT publish a last restore seq") + } + }) + + // (c) The commit errored AND the reconcile read itself failed → outcome + // UNCERTAIN. ForceRefreshRoom must return an error wrapping both causes, must + // NOT publish a boundary/seq, and must PLAIN-CLOSE the peer's socket (NOT + // force_refresh) so it reconnects and re-evaluates the durable fences fresh — + // a frozen-but-open socket would silently drop every subsequent edit forever. + t.Run("uncertain_plain_closes_peers", func(t *testing.T) { + bus := NewMemoryOpBus() + defer bus.Close() + mgr := NewRoomManager(&fakeOpLog{}, bus) + defer mgr.Close() + + srv := newCollabTestServer(t, mgr) + defer srv.Close() + + conn := dialWS(t, srv, "item-a") + defer conn.Close() + for i := 0; i < 200; i++ { + if mgr.RoomCount() == 1 && bus.SubscriberCount("item-a") == 1 { + break + } + time.Sleep(5 * time.Millisecond) + } + + commitErr := errors.New("commit: bad connection") + reconcileErr := errors.New("reconcile: read failed") + err := mgr.ForceRefreshRoom("item-a", + func() (int64, int64, error) { return 0, 0, commitErr }, + func() (RestoreReconcileResult, error) { + return RestoreReconcileResult{}, reconcileErr + }) + if err == nil { + t.Fatal("uncertain reconcile must return an error") + } + if !errors.Is(err, commitErr) || !errors.Is(err, reconcileErr) { + t.Fatalf("uncertain err = %v, want it to wrap BOTH the commit and reconcile errors", err) + } + if _, ok := mgr.RestoreBoundary("item-a"); ok { + t.Fatal("uncertain reconcile must NOT publish a restore boundary") + } + if _, ok := mgr.LastRestoreSeq("item-a"); ok { + t.Fatal("uncertain reconcile must NOT publish a last restore seq") + } + // The peer's socket must be PLAIN-closed: no force_refresh frame arrives + // before the read errors out. (force_refresh would falsely assert + // items.content is authoritative, which is exactly what's unknown here.) + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + closed := false + for { + mt, data, rerr := conn.ReadMessage() + if rerr != nil { + closed = true + break // socket closed by closeAllConnsPlain + } + if mt == websocket.TextMessage { + var ctl ControlMessage + if json.Unmarshal(data, &ctl) == nil && ctl.Type == ControlMessageForceRefresh { + t.Fatal("uncertain reconcile plain-closes; it must NOT send a force_refresh frame") + } + } + } + if !closed { + t.Fatal("uncertain reconcile must close the peer's socket") + } + }) +} + +// TestForceRefreshRoomUncertainInvalidatesInMemoryFences is the BUG-2276 +// residual-1 P1 guard: on the UNCERTAIN commit-outcome path, ForceRefreshRoom must +// CLEAR the item's in-memory restore fences (lastRestoreSeqs + restoreBoundaries) +// so a reconnecting stale-seed peer is judged against the DURABLE column — which +// reflects a landed-but-ack-lost restore — not a PRIOR restore's cached generation. +// Otherwise a cursor-0 peer seeded at the prior generation would be admitted against +// the stale cache and clobber the landed content. +func TestForceRefreshRoomUncertainInvalidatesInMemoryFences(t *testing.T) { + bus := NewMemoryOpBus() + defer bus.Close() + // The DURABLE column reflects the LANDED-but-ack-lost restore R2 (seq=20). The + // in-memory caches still hold the PRIOR restore R1's generation (seq=10 / + // boundary=100) — the stale-cache trap this guards against. + store := &fakeOpLog{lastRestoreSeqs: map[string]int64{"item-a": 20}} + mgr := NewRoomManager(store, bus) + defer mgr.Close() + + mgr.SetLastRestoreSeq("item-a", 10) // R1 in-memory fast-path + mgr.SetRestoreBoundary("item-a", 100) // R1 in-memory boundary + if s, ok := mgr.LastRestoreSeq("item-a"); !ok || s != 10 { + t.Fatalf("precondition: R1 last restore seq = (%d,%v), want (10,true)", s, ok) + } + if b, ok := mgr.RestoreBoundary("item-a"); !ok || b != 100 { + t.Fatalf("precondition: R1 boundary = (%d,%v), want (100,true)", b, ok) + } + + // R2's commit errors AND its reconcile read errors → UNCERTAIN. + err := mgr.ForceRefreshRoom("item-a", + func() (int64, int64, error) { return 0, 0, errors.New("commit: bad connection") }, + func() (RestoreReconcileResult, error) { + return RestoreReconcileResult{}, errors.New("reconcile: read failed") + }) + if err == nil { + t.Fatal("uncertain reconcile must return an error") + } + + // Both in-memory fences must be CLEARED so the next check reads durable. + if s, ok := mgr.LastRestoreSeq("item-a"); ok { + t.Fatalf("uncertain must invalidate the in-memory last-restore-seq fence; still cached (%d)", s) + } + if b, ok := mgr.RestoreBoundary("item-a"); ok { + t.Fatalf("uncertain must invalidate the in-memory restore boundary; still cached (%d)", b) + } + + // A peer seeded at R1's generation (content_seq=10) now reconnecting is judged + // against the DURABLE column (R2=20): 10 < 20 → force_refresh. With the STALE + // cache it would have compared 10 vs cached-10 (equal) and been admitted — the + // clobber this fix prevents. + srv := newCollabTestServer(t, mgr) + defer srv.Close() + stale := dialWSContentSeq(t, srv, "item-a", 10) + defer stale.Close() + if ctl := readControlWithin(t, stale, time.Second); ctl.Type != ControlMessageForceRefresh { + t.Fatalf("stale R1-generation seed after uncertain: want %q (durable fence), got %q", ControlMessageForceRefresh, ctl.Type) + } +} + // TestJoinDurableRestoreBoundaryFencesStaleSeedAfterRestart is the BUG-2264 // restart-durability guard (residual #1 closed): the stale-seed Join fence must // survive a server restart. A restore in a PRIOR process stamped the DURABLE diff --git a/internal/collab/room.go b/internal/collab/room.go index 2403f3f5..64a8a488 100644 --- a/internal/collab/room.go +++ b/internal/collab/room.go @@ -611,6 +611,35 @@ func (r *Room) closeAll() { } } +// closeAllConnsPlain force-closes every connection on this room with a PLAIN +// socket Close — no force_refresh frame, and the room is left ALIVE (not marked +// closing) so a reconnect rejoins normally. Used by ForceRefreshRoom's UNCERTAIN +// commit-outcome path (BUG-2276 residual 1): when a Postgres restore commit +// errors and reconciliation cannot tell whether the tx landed, we must NOT tell +// peers to reseed from items.content (a force_refresh would assert the content is +// authoritative, which is exactly what we don't know). We also must not leave the +// peers frozen-but-open: the collab read path has no WS read-deadline/heartbeat, +// so a frozen conn would silently drop every subsequent edit forever. A plain +// close makes each client reconnect and re-evaluate the DURABLE restore fences +// (items.last_restore_seq / restore_boundary_op_id) fresh through Join, which is +// safe whichever way the commit actually went. conn.Close is concurrency-safe +// with an in-flight WriteMessage (gorilla), so no writeMu handshake is needed; +// each conn's frozen flag is left set so any frame already read is still dropped +// before the readLoop unwinds. Mirrors closeAll's collect-under-lock / +// close-outside-lock discipline. +func (r *Room) closeAllConnsPlain() { + r.mu.Lock() + conns := make([]*websocket.Conn, 0, len(r.conns)) + for c := range r.conns { + conns = append(conns, c) + } + r.mu.Unlock() + + for _, c := range conns { + _ = c.Close() + } +} + // forceRefreshAll sends a force_refresh control frame to every connected client // and then closes each connection, so every peer discards its in-memory Y.Doc // and rebuilds from the canonical items.content on reconnect (BUG-2264). Used by diff --git a/internal/server/handlers_item_versions.go b/internal/server/handlers_item_versions.go index c7644a65..7d147ee6 100644 --- a/internal/server/handlers_item_versions.go +++ b/internal/server/handlers_item_versions.go @@ -3,12 +3,15 @@ package server import ( "database/sql" "errors" + "fmt" "net/http" "github.com/go-chi/chi/v5" + "github.com/PerpetualSoftware/pad/internal/collab" "github.com/PerpetualSoftware/pad/internal/events" "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" ) // errRestoreItemGone signals that the item was concurrently deleted during a @@ -175,10 +178,51 @@ func (s *Server) handleRestoreItemVersion(w http.ResponseWriter, r *http.Request // atomically (no out-of-tx fail-open) and a failed commit rolls back all of // it. ForceRefreshRoom then publishes the boundary (maxID+1) + content // generation (seq) and reseeds under the per-item lock. + // + // Postgres-only commit-outcome reconciliation (BUG-2276 residual 1): on + // Postgres a commit that DURABLY lands but whose ack is lost at the + // connection boundary surfaces to ForceRefreshRoom as an error; supply a + // reconcile callback that re-reads to distinguish that from a genuine + // rollback. On SQLite a commit error is unambiguous, so we pass nil and keep + // the verbatim rollback-on-error path. `content` is the exact restored + // version content. + // + // baselineSeq is the pre-restore item.seq the reconcile compares + // last_restore_seq against. It is captured INSIDE the commit's precheck — + // i.e. UNDER ForceRefreshRoom's per-item lock + the workspace seq lock, + // immediately before the mutation — NOT from the pre-lock `item.Seq` read at + // the top of the handler. A restore that completed while this request waited + // for the per-item lock would otherwise leave its advanced last_restore_seq + // visible against a stale baseline and make reconcile falsely classify a + // genuine rollback of THIS attempt as landed (BUG-2276 P2). + var ( + baselineSeq int64 + baselineCaptured bool + reconcile func() (collab.RestoreReconcileResult, error) + ) + if s.store.D().Driver() == store.DriverPostgres { + reconcile = func() (collab.RestoreReconcileResult, error) { + res, fresh, rerr := s.reconcileRestoreCommit(item.ID, content, baselineSeq, baselineCaptured) + if rerr == nil && res.Landed { + // Reconciled LANDED despite the lost ack: surface the freshly-read + // restored item so the handler returns it AND emits the item SSE + // event, instead of the false 404 it would hit with updated==nil + // (the commit's UpdateItemWithPreCheck returned (nil, ackErr) on this + // path). BUG-2276 P2. + updated = fresh + } + return res, rerr + } + } werr := s.collab.ForceRefreshRoom(item.ID, func() (int64, int64, error) { var maxID int64 u, uerr := s.store.UpdateItemWithPreCheck(item.ID, input, - func(tx *sql.Tx, _ *models.Item) error { + func(tx *sql.Tx, existing *models.Item) error { + // Capture the pre-restore seq under the per-item + workspace seq + // lock, before any mutation (BUG-2276 P2 — see the baselineSeq note + // above). `existing` is the row as read at the top of the update tx. + baselineSeq = existing.Seq + baselineCaptured = true m, _, merr := s.store.MaxOpLogIDTx(tx, item.ID) if merr != nil { return merr @@ -201,11 +245,20 @@ func (s *Server) handleRestoreItemVersion(w http.ResponseWriter, r *http.Request // or reseeding; the handler surfaces it below. return 0, 0, errRestoreItemGone } + if s.restoreAckFault != nil { + if fe := s.restoreAckFault(); fe != nil { + // TEST SEAM (BUG-2276 residual 1): the tx above committed DURABLY, + // but return an error exactly as UpdateItemWithPreCheck would on a + // lost ack — and do NOT set `updated`, so reconcile must recover the + // restored item from the durable state. + return 0, 0, fe + } + } updated = u // (pre-prune MAX for the stale-flush boundary, restored seq for the // content generation Join uses to force_refresh stale-seeded peers). return maxID, u.Seq, nil - }) + }, reconcile) if werr != nil { if errors.Is(werr, errRestoreItemGone) { writeError(w, http.StatusNotFound, "not_found", "Item not found") @@ -246,3 +299,75 @@ func (s *Server) handleRestoreItemVersion(w http.ResponseWriter, r *http.Request writeJSON(w, http.StatusOK, updated) } + +// reconcileRestoreCommit re-reads an item after a version-restore commit reported +// an error, to determine whether the restore's transaction DURABLY landed anyway +// (a Postgres commit whose ack was lost at the connection boundary — BUG-2276 +// residual 1). It is supplied to ForceRefreshRoom ONLY on Postgres; SQLite passes +// nil, keeping the unambiguous rollback-on-error behavior. +// +// A restore's defining durable effects, all written in ONE tx, are: items.content +// = the target version's content, items.last_restore_seq = the restore's new +// item.seq (strictly greater than the pre-restore seq), and +// items.restore_boundary_op_id = pre-prune MAX(op-log.id)+1. We read all three and +// require the two INDEPENDENT signals to AGREE: +// +// - contentMatches: fresh items.content == the exact restored version content +// (a restore sets content to an EXACT prior version). +// - seqAdvanced: items.last_restore_seq > baselineSeq (the pre-restore seq +// captured UNDER the per-item lock, immediately before the tx). last_restore_seq +// is written ONLY by restores, which are serialised under the collab per-item +// lock, so during this reconcile it holds either the pre-restore value (rolled +// back) or this restore's new seq (landed) — never a concurrent writer's. Gated +// on baselineCaptured: if the tx failed before the precheck ran (nothing +// mutated), the baseline is unknown, so seqAdvanced is forced false. +// +// Outcomes: +// - Both true → LANDED (recover Boundary = restore_boundary_op_id and Seq = +// last_restore_seq so ForceRefreshRoom's success path publishes the same fences +// it would have on a clean commit, and return the freshly-read item so the +// caller can respond with it + emit the SSE event). +// - Both false → DEFINITELY rolled back (un-freeze, the genuine-rollback path); +// returns a nil item. +// - Disagree, read error, or NOT-FOUND → ambiguous; return an error so +// ForceRefreshRoom keeps the room frozen and plain-closes it (the safe degraded +// mode). A not-found re-read is UNCERTAIN, NOT rolled-back: the restore may have +// durably landed and a concurrent archive then soft-deleted the item — treating +// that as rolled-back would un-freeze stale peers onto the archived item and let +// them poison its op-log for a later unarchive (BUG-2276 P1). Also covers the +// rare restore-to-identical-content tx that ALSO ack-lost (contentMatches +// coincidentally true while seqAdvanced is false): staying frozen is safe. +func (s *Server) reconcileRestoreCommit(itemID, targetContent string, baselineSeq int64, baselineCaptured bool) (collab.RestoreReconcileResult, *models.Item, error) { + fresh, err := s.store.GetItem(itemID) + if err != nil { + return collab.RestoreReconcileResult{}, nil, fmt.Errorf("reconcile restore: read item: %w", err) + } + if fresh == nil { + // NOT-FOUND is UNCERTAIN, not rolled-back (BUG-2276 P1): a durable restore + // could have landed and a concurrent archive then soft-deleted the item. + // Return an error → ForceRefreshRoom stays frozen + plain-closes, so no stale + // peer resumes onto the archived item's op-log. + return collab.RestoreReconcileResult{}, nil, fmt.Errorf("reconcile restore: item %s not found on re-read (archived mid-restore?)", itemID) + } + lastRestoreSeq, lrOK, err := s.store.ItemLastRestoreSeq(itemID) + if err != nil { + return collab.RestoreReconcileResult{}, nil, fmt.Errorf("reconcile restore: read last_restore_seq: %w", err) + } + boundaryOpID, bOK, err := s.store.ItemRestoreBoundaryOpID(itemID) + if err != nil { + return collab.RestoreReconcileResult{}, nil, fmt.Errorf("reconcile restore: read restore_boundary_op_id: %w", err) + } + + contentMatches := fresh.Content == targetContent + seqAdvanced := baselineCaptured && lrOK && lastRestoreSeq > baselineSeq + switch { + case contentMatches && seqAdvanced && bOK: + return collab.RestoreReconcileResult{Landed: true, Boundary: boundaryOpID, Seq: lastRestoreSeq}, fresh, nil + case !contentMatches && !seqAdvanced: + return collab.RestoreReconcileResult{Landed: false}, nil, nil + default: + return collab.RestoreReconcileResult{}, nil, fmt.Errorf( + "reconcile restore: ambiguous outcome (contentMatches=%v seqAdvanced=%v boundaryStamped=%v baselineCaptured=%v)", + contentMatches, seqAdvanced, bOK, baselineCaptured) + } +} diff --git a/internal/server/handlers_item_versions_test.go b/internal/server/handlers_item_versions_test.go new file mode 100644 index 00000000..c8059d69 --- /dev/null +++ b/internal/server/handlers_item_versions_test.go @@ -0,0 +1,433 @@ +package server + +import ( + "database/sql" + "errors" + "net/http" + "sync/atomic" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/collab" + "github.com/PerpetualSoftware/pad/internal/events" + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" + "github.com/PerpetualSoftware/pad/internal/store/storetest" +) + +const ( + reconcileOriginal = "original body\n" + reconcileRestored = "restored-from-an-older-version body\n" +) + +// errSimAckLoss simulates a Postgres commit whose tx durably landed but whose +// acknowledgement was lost at the connection boundary — the exact case BUG-2276 +// residual 1's reconciliation exists for. +var errSimAckLoss = errors.New("sim: commit ack lost after durable landing") + +func reconcileNewItem(t *testing.T, srv *Server, slug, content string) *models.Item { + t.Helper() + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+slug+"/collections/tasks/items", map[string]interface{}{ + "title": "reconcile subject", + "content": content, + "source": "cli", + "fields": `{"status":"open"}`, + }) + if rr.Code != http.StatusCreated { + t.Fatalf("create item: %d: %s", rr.Code, rr.Body.String()) + } + var it models.Item + parseJSON(t, rr, &it) + return &it +} + +// runRealRestoreTx drives the REAL restore commit transaction against the store — +// the exact MaxOpLogIDTx + StampRestoreBoundaryOpIDTx + PruneItemOpLogTx + +// MarkRestoreBoundary sequence handleRestoreItemVersion runs — so reconcile is +// exercised against genuine ATOMIC durable stamps, not columns written by hand. +// Returns the restored item + the pre-prune MAX (the boundary is MAX+1). +func runRealRestoreTx(t *testing.T, srv *Server, itemID, content string) (*models.Item, int64) { + t.Helper() + var maxID int64 + u, err := srv.store.UpdateItemWithPreCheck(itemID, models.ItemUpdate{ + Content: &content, + ChangeSummary: "Restored from an older version", + LastModifiedBy: "user", + Source: "web", + ForceVersion: true, + MarkRestoreBoundary: true, + }, func(tx *sql.Tx, _ *models.Item) error { + m, _, merr := srv.store.MaxOpLogIDTx(tx, itemID) + if merr != nil { + return merr + } + maxID = m + if serr := srv.store.StampRestoreBoundaryOpIDTx(tx, itemID, m+1); serr != nil { + return serr + } + return srv.store.PruneItemOpLogTx(tx, itemID) + }) + if err != nil { + t.Fatalf("real restore tx: %v", err) + } + if u == nil { + t.Fatal("real restore tx: item vanished") + } + return u, maxID +} + +// TestReconcileRestoreCommit exercises the Postgres commit-outcome reconciliation +// signal logic (BUG-2276 residual 1) against a real store, driving genuine restore +// tx stamps. After a version-restore commit reports an error, reconcileRestoreCommit +// re-reads the item and decides — from two independent durable signals (content == +// the restored version AND last_restore_seq advanced past the UNDER-LOCK baseline +// seq) — whether the restore actually LANDED, DEFINITELY rolled back, or is +// UNCERTAIN (signals disagree / read error / not-found), the last of which returns +// an error so ForceRefreshRoom keeps the room frozen and plain-closes it. +// +// (Runs on the SQLite test store; the helper's LOGIC is dialect-independent — it is +// only WIRED IN on Postgres, where a durably-landed-but-ack-lost commit can occur. +// The store gate lives in handleRestoreItemVersion.) +func TestReconcileRestoreCommit(t *testing.T) { + srv := testServer(t) + slug := createWSWithCollections(t, srv) + + // (a) The restore's durable effects are all present on a fresh read → LANDED, + // with Boundary/Seq recovered from the durable stamps AND the fresh item + // returned (so the caller can respond with it + emit the SSE event). + t.Run("landed_returns_item", func(t *testing.T) { + it := reconcileNewItem(t, srv, slug, reconcileOriginal) + baseline := it.Seq + u, maxID := runRealRestoreTx(t, srv, it.ID, reconcileRestored) + + res, fresh, err := srv.reconcileRestoreCommit(it.ID, reconcileRestored, baseline, true) + if err != nil { + t.Fatalf("reconcile (landed) err: %v", err) + } + if !res.Landed { + t.Fatal("reconcile: want Landed=true for a durably-landed restore") + } + if res.Boundary != maxID+1 { + t.Fatalf("reconcile Boundary = %d, want %d (from restore_boundary_op_id)", res.Boundary, maxID+1) + } + if res.Seq != u.Seq { + t.Fatalf("reconcile Seq = %d, want %d (from last_restore_seq)", res.Seq, u.Seq) + } + if fresh == nil || fresh.Content != reconcileRestored { + t.Fatalf("reconcile must return the freshly-read restored item; got %+v", fresh) + } + }) + + // (b) Nothing landed: content unchanged, last_restore_seq never set → both + // signals say "not landed" → Landed=false, nil item (the genuine-rollback path). + t.Run("rolled_back", func(t *testing.T) { + it := reconcileNewItem(t, srv, slug, reconcileOriginal) + res, fresh, err := srv.reconcileRestoreCommit(it.ID, reconcileRestored, it.Seq, true) + if err != nil { + t.Fatalf("reconcile (rolled back) err: %v", err) + } + if res.Landed { + t.Fatal("reconcile: want Landed=false when nothing landed") + } + if fresh != nil { + t.Fatal("reconcile: rolled-back must return a nil item") + } + }) + + // (c) Signals disagree: content already equals the restore target but + // last_restore_seq did NOT advance → UNCERTAIN → error → caller stays frozen. + t.Run("uncertain_signals_disagree", func(t *testing.T) { + it := reconcileNewItem(t, srv, slug, reconcileRestored) + res, _, err := srv.reconcileRestoreCommit(it.ID, reconcileRestored, it.Seq, true) + if err == nil { + t.Fatalf("reconcile: want an error for ambiguous signals, got Landed=%v", res.Landed) + } + }) + + // (d) BUG-2276 P1: a NOT-FOUND re-read is UNCERTAIN (error), NOT rolled-back. + // The restore may have durably landed and a concurrent archive then soft-deleted + // the item; classifying that as rolled-back would un-freeze stale peers onto the + // archived item and poison its op-log. + t.Run("not_found_is_uncertain", func(t *testing.T) { + res, fresh, err := srv.reconcileRestoreCommit("nonexistent-item-id", reconcileRestored, 0, true) + if err == nil { + t.Fatalf("reconcile: a not-found re-read must be UNCERTAIN (error), got Landed=%v", res.Landed) + } + if fresh != nil { + t.Fatal("reconcile: not-found must return a nil item") + } + }) + + // (e) BUG-2276 P2: a prior restore advanced last_restore_seq + stamped the + // boundary; THIS attempt then genuinely rolls back choosing identical content. + // With the baseline captured UNDER the lock (= the post-prior-restore seq), + // reconcile must NOT falsely classify this rollback as LANDED. (A stale pre-lock + // baseline below the prior restore's seq WOULD have made seqAdvanced true.) + t.Run("stale_baseline_prior_restore_not_false_landed", func(t *testing.T) { + it := reconcileNewItem(t, srv, slug, reconcileOriginal) + u, _ := runRealRestoreTx(t, srv, it.ID, reconcileRestored) // prior restore R1 + // R2 rolled back: nothing changed since R1 (content==restored, + // last_restore_seq==u.Seq). The under-lock baseline == u.Seq. + res, _, err := srv.reconcileRestoreCommit(it.ID, reconcileRestored, u.Seq, true) + if err == nil && res.Landed { + t.Fatal("stale-baseline guard: a rollback after a prior restore must NOT be classified LANDED") + } + }) +} + +// TestForceRefreshRoomAckLossReconciledReturnsRestoredItem drives the REAL restore +// tx + REAL reconcile through ForceRefreshRoom with a synthetic ACK loss (the +// commit's tx durably lands, then the commit closure returns an error as if the +// ack were lost at the connection boundary). It proves the corrected BUG-2276 P2 +// no-false-404 wiring: on reconciled-LANDED, the reconcile closure captures the +// freshly-read restored item into `updated`, so the handler returns it (200) and +// emits the item_updated SSE — instead of the false 404 it hit before +// (UpdateItemWithPreCheck returns (nil, ackErr) on this path). +func TestForceRefreshRoomAckLossReconciledReturnsRestoredItem(t *testing.T) { + srv := testServerWithCollab(t) + slug := createWSWithCollections(t, srv) + it := reconcileNewItem(t, srv, slug, reconcileOriginal) + + // Mirror handleRestoreItemVersion's wiring: baseline captured under the lock in + // the precheck; reconcile surfaces the restored item into `updated`. + var ( + updated *models.Item + baselineSeq int64 + baselineCaptured bool + ) + content := reconcileRestored + input := models.ItemUpdate{ + Content: &content, + ChangeSummary: "Restored from an older version", + LastModifiedBy: "user", + Source: "web", + ForceVersion: true, + MarkRestoreBoundary: true, + } + reconcile := func() (collab.RestoreReconcileResult, error) { + res, fresh, rerr := srv.reconcileRestoreCommit(it.ID, content, baselineSeq, baselineCaptured) + if rerr == nil && res.Landed { + updated = fresh + } + return res, rerr + } + + werr := srv.collab.ForceRefreshRoom(it.ID, func() (int64, int64, error) { + _, uerr := srv.store.UpdateItemWithPreCheck(it.ID, input, func(tx *sql.Tx, existing *models.Item) error { + baselineSeq = existing.Seq + baselineCaptured = true + m, _, merr := srv.store.MaxOpLogIDTx(tx, it.ID) + if merr != nil { + return merr + } + if serr := srv.store.StampRestoreBoundaryOpIDTx(tx, it.ID, m+1); serr != nil { + return serr + } + return srv.store.PruneItemOpLogTx(tx, it.ID) + }) + if uerr != nil { + return 0, 0, uerr + } + // The tx committed durably; now simulate the lost ACK. Return an error WITHOUT + // setting `updated`, exactly as the real UpdateItemWithPreCheck would return + // (nil, err) on ack loss. + return 0, 0, errSimAckLoss + }, reconcile) + + if werr != nil { + t.Fatalf("ForceRefreshRoom must reconcile the ack-lost-but-landed commit to success, got: %v", werr) + } + if updated == nil { + t.Fatal("BUG-2276 P2: reconciled-LANDED must surface the restored item into `updated` (else the handler false-404s)") + } + if updated.Content != reconcileRestored { + t.Fatalf("updated.Content = %q, want the restored content", updated.Content) + } + if b, ok := srv.collab.RestoreBoundary(it.ID); !ok || b <= 0 { + t.Fatalf("reconciled-LANDED must publish the restore boundary; got (%d, %v)", b, ok) + } + if s, ok := srv.collab.LastRestoreSeq(it.ID); !ok || s != updated.Seq { + t.Fatalf("reconciled-LANDED last restore seq = (%d, %v), want (%d, true)", s, ok, updated.Seq) + } +} + +// TestRestoreItemVersionEmitsItemUpdatedSSE proves the restore handler emits the +// item_updated SSE whenever it produces an updated item — the event path the +// BUG-2276 P2 fix relies on (reconciled-LANDED sets `updated`, so this same emit +// fires). Driven through the real HTTP handler on the normal restore path. +func TestRestoreItemVersionEmitsItemUpdatedSSE(t *testing.T) { + srv := testServer(t) + bus := events.New() + srv.SetEventBus(bus) + slug := createWSWithCollections(t, srv) + + ws, err := srv.store.GetWorkspaceBySlug(slug) + if err != nil || ws == nil { + t.Fatalf("resolve workspace: %v", err) + } + + // Create v1, then update to v2 so a version bracketing v1 exists to restore. + it := reconcileNewItem(t, srv, slug, reconcileOriginal) + time.Sleep(1100 * time.Millisecond) + rr := doRequest(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+it.Slug, map[string]interface{}{ + "content": reconcileRestored, + }) + if rr.Code != http.StatusOK { + t.Fatalf("update item: %d: %s", rr.Code, rr.Body.String()) + } + + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items/"+it.Slug+"/versions", nil) + if rr.Code != http.StatusOK { + t.Fatalf("list versions: %d: %s", rr.Code, rr.Body.String()) + } + var versions []models.Version + parseJSON(t, rr, &versions) + var versionID string + for _, v := range versions { + if v.Content == reconcileOriginal { + versionID = v.ID + break + } + } + if versionID == "" { + t.Fatalf("no version with the original content to restore; got %d versions", len(versions)) + } + + // Subscribe right before the restore so the next item_updated is the restore's. + ch := bus.Subscribe(ws.ID) + defer bus.Unsubscribe(ch) + + rr = doRequest(srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+it.Slug+"/versions/"+versionID+"/restore", nil) + if rr.Code != http.StatusOK { + t.Fatalf("restore version: %d: %s", rr.Code, rr.Body.String()) + } + + select { + case ev := <-ch: + if ev.Type != "item_updated" { + t.Fatalf("restore emitted SSE type %q, want item_updated", ev.Type) + } + if ev.ItemID != it.ID { + t.Fatalf("restore SSE ItemID = %q, want %q", ev.ItemID, it.ID) + } + case <-time.After(2 * time.Second): + t.Fatal("restore did not emit an item_updated SSE event") + } + + // Exactly one: no second event follows within a short window. + select { + case ev := <-ch: + t.Fatalf("restore must emit EXACTLY one event; got a second: type=%q", ev.Type) + case <-time.After(300 * time.Millisecond): + } +} + +// testServerPostgres returns a *Server backed by an ISOLATED PostgreSQL database +// (t.Skip unless PAD_TEST_POSTGRES_URL is set) with collab + an event bus wired. +// Unlike testServer (always SQLite), this lets a test exercise the Postgres-gated +// restore reconciliation end-to-end under `make test-pg`. +func testServerPostgres(t *testing.T) (*Server, *events.MemoryBus) { + t.Helper() + s := storetest.NewPostgres(t) // skips if PAD_TEST_POSTGRES_URL is unset + srv := New(s) + t.Cleanup(func() { srv.Stop() }) + + obus := collab.NewMemoryOpBus() + t.Cleanup(obus.Close) + rm := collab.NewRoomManager(srv.store, obus) + t.Cleanup(rm.Close) + srv.SetCollabRoomManager(rm) + + ebus := events.New() + srv.SetEventBus(ebus) + return srv, ebus +} + +// TestRestoreVersionHandlerAckLossReconciledEndToEnd drives the ACTUAL restore +// ENDPOINT through the real Postgres-gated handler with a commit-ack-loss seam +// (BUG-2276 residual 1). The restore tx durably commits, then the seam returns an +// error as if the ack were lost; the handler's Postgres-gated reconcile must +// re-read, classify LANDED, and respond 200 with the restored item AND emit exactly +// one item_updated event — proving the P2 no-false-404 fix end-to-end on a real +// Postgres store. Skips unless PAD_TEST_POSTGRES_URL is set (runs under make test-pg). +func TestRestoreVersionHandlerAckLossReconciledEndToEnd(t *testing.T) { + srv, ebus := testServerPostgres(t) + if srv.store.D().Driver() != store.DriverPostgres { + t.Fatalf("expected a Postgres store, got %s", srv.store.D().Driver()) + } + slug := createWSWithCollections(t, srv) + ws, err := srv.store.GetWorkspaceBySlug(slug) + if err != nil || ws == nil { + t.Fatalf("resolve workspace: %v", err) + } + + // Create v1, update to v2 so a version bracketing v1 exists to restore. + it := reconcileNewItem(t, srv, slug, reconcileOriginal) + time.Sleep(1100 * time.Millisecond) + rr := doRequest(srv, "PATCH", "/api/v1/workspaces/"+slug+"/items/"+it.Slug, map[string]interface{}{ + "content": reconcileRestored, + }) + if rr.Code != http.StatusOK { + t.Fatalf("update item: %d: %s", rr.Code, rr.Body.String()) + } + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items/"+it.Slug+"/versions", nil) + if rr.Code != http.StatusOK { + t.Fatalf("list versions: %d: %s", rr.Code, rr.Body.String()) + } + var versions []models.Version + parseJSON(t, rr, &versions) + var versionID string + for _, v := range versions { + if v.Content == reconcileOriginal { + versionID = v.ID + break + } + } + if versionID == "" { + t.Fatalf("no version with the original content to restore; got %d versions", len(versions)) + } + + // Arm the ack-loss seam to fire ONCE (this restore), so the tx durably lands but + // the commit closure returns an error — exactly the Postgres commit-ack-loss case. + var faultFired int32 + srv.restoreAckFault = func() error { + if atomic.AddInt32(&faultFired, 1) == 1 { + return errSimAckLoss + } + return nil + } + + ch := ebus.Subscribe(ws.ID) + defer ebus.Unsubscribe(ch) + + rr = doRequest(srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+it.Slug+"/versions/"+versionID+"/restore", nil) + if rr.Code != http.StatusOK { + t.Fatalf("restore under ack-loss must RECONCILE to 200 (no false 404); got %d: %s", rr.Code, rr.Body.String()) + } + if atomic.LoadInt32(&faultFired) == 0 { + t.Fatal("the ack-loss seam never fired; the reconcile path wasn't exercised") + } + var restored models.Item + parseJSON(t, rr, &restored) + if restored.Content != reconcileOriginal { + t.Fatalf("reconciled restore body content = %q, want the restored version content", restored.Content) + } + if restored.ID != it.ID { + t.Fatalf("reconciled restore body ID = %q, want %q", restored.ID, it.ID) + } + + // Exactly one item_updated event fires on the reconciled path. + select { + case ev := <-ch: + if ev.Type != "item_updated" || ev.ItemID != it.ID { + t.Fatalf("reconciled restore SSE: type=%q item=%q, want item_updated for %q", ev.Type, ev.ItemID, it.ID) + } + case <-time.After(2 * time.Second): + t.Fatal("reconciled restore did not emit an item_updated event") + } + select { + case ev := <-ch: + t.Fatalf("reconciled restore must emit EXACTLY one event; got a second: type=%q", ev.Type) + case <-time.After(300 * time.Millisecond): + } +} diff --git a/internal/server/server.go b/internal/server/server.go index a8c1b62a..9b9f9688 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -247,6 +247,14 @@ type Server struct { // homelabs behind a firewall) typically prefer this; operators with // public exposure should leave it off and use the logs-token path. bypassSetupToken bool + + // restoreAckFault is a TEST SEAM (always nil in production). When non-nil, + // handleRestoreItemVersion's collab commit closure invokes it AFTER the restore + // transaction has durably committed; a non-nil return simulates a Postgres commit + // whose acknowledgement was lost at the connection boundary (the tx landed, but + // the driver surfaces an error), exercising BUG-2276 residual 1's commit-outcome + // reconciliation end-to-end through the real handler. + restoreAckFault func() error } // goAsync spawns fn in a goroutine that's tracked by s.bg, so Stop() can diff --git a/internal/store/storetest/postgres.go b/internal/store/storetest/postgres.go new file mode 100644 index 00000000..4a5fffe6 --- /dev/null +++ b/internal/store/storetest/postgres.go @@ -0,0 +1,80 @@ +package storetest + +import ( + "database/sql" + "os" + "strings" + "testing" + + "github.com/google/uuid" + + "github.com/PerpetualSoftware/pad/internal/store" +) + +// NewPostgres returns a *store.Store backed by an ISOLATED PostgreSQL database +// when PAD_TEST_POSTGRES_URL is set, and t.Skip()s the test otherwise. It lets +// tests OUTSIDE the store package (e.g. internal/server) exercise Postgres-gated +// code paths under `make test-pg`, where the store-package white-box helper +// (internal/store/store_test.go::testStorePostgres) isn't importable. +// +// It mirrors that helper: a uniquely-named database is CREATEd off the base URL, +// opened via store.NewPostgres (which runs the full migration chain), and DROPped +// on cleanup. The pgx driver is already registered transitively via the store +// import. KEEP IN SYNC with store_test.go's testStorePostgres (duplicated for the +// same import-cycle reason as NewSQLite — see the package doc). +func NewPostgres(t *testing.T) *store.Store { + t.Helper() + + baseURL := os.Getenv("PAD_TEST_POSTGRES_URL") + if baseURL == "" { + t.Skip("PAD_TEST_POSTGRES_URL not set; Postgres-backed test skipped") + } + + dbName := "pad_test_" + strings.ReplaceAll(uuid.New().String()[:8], "-", "") + + admin, err := sql.Open("pgx", baseURL) + if err != nil { + t.Fatalf("storetest: open pg admin conn: %v", err) + } + // CREATE DATABASE cannot run inside a transaction. + if _, err := admin.Exec("CREATE DATABASE " + dbName); err != nil { + _ = admin.Close() + t.Fatalf("storetest: create test database %s: %v", dbName, err) + } + _ = admin.Close() + + s, err := store.NewPostgres(replaceDBName(baseURL, dbName)) + if err != nil { + dropPostgresDB(baseURL, dbName) + t.Fatalf("storetest: open test postgres store: %v", err) + } + t.Cleanup(func() { + _ = s.Close() + dropPostgresDB(baseURL, dbName) + }) + return s +} + +func dropPostgresDB(baseURL, dbName string) { + admin, err := sql.Open("pgx", baseURL) + if err != nil { + return + } + defer admin.Close() + _, _ = admin.Exec("DROP DATABASE IF EXISTS " + dbName + " WITH (FORCE)") +} + +// replaceDBName swaps the database name in a postgres connection URL, preserving +// any query string. Mirrors store_test.go::replaceDBName. +func replaceDBName(connStr, newDB string) string { + query := "" + base := connStr + if qIdx := strings.IndexByte(connStr, '?'); qIdx >= 0 { + query = connStr[qIdx:] + base = connStr[:qIdx] + } + if lastSlash := strings.LastIndexByte(base, '/'); lastSlash >= 0 { + return base[:lastSlash+1] + newDB + query + } + return connStr +}