From 698fe2ff6eb2bc84d21af7d22afa4e7a4b525801 Mon Sep 17 00:00:00 2001 From: xarmian Date: Wed, 7 Oct 2026 19:45:14 -0400 Subject: [PATCH] fix(server): a collab revalidation tick that errors makes the conn read-only (BUG-3474) (#1866) * test(server): a revalidation tick that errors leaves the collab conn writable (BUG-3474, red) A real fault: workspace_members is renamed, so the tick's membership lookup fails with a plain error. After the first failing tick, a frame still persists. Red 3/3 on SQLite. * fix(server): a collab revalidation tick that errors makes the conn read-only, not closed (BUG-3474) Lead ruling: fail closed on WRITE, keep the connection. Both error branches of the tick (GetItem, and a plain error from authorizeCollabAccess) now call SetConnWritable(false) before logging. The next clean tick restores write if the role still allows it. The test faults each branch for real (renamed workspace_members, renamed items). * test(server): a credential re-check store error keeps the collab conn writable; recovery is checked by content (BUG-3474, codex r1, red) * fix(server): an unknown credential re-check makes the collab conn read-only and skips the tick (BUG-3474, codex r1) streamCredentialStillValid reads a store error as valid (right for SSE, which has no write), so the collab tick went on to set write from the role. The collab loop now switches on credentialLiveness: invalid closes as before; unknown sets read-only, logs at warn and re-arms. --- .../server/bug3474_reval_fail_closed_test.go | 214 ++++++++++++++++++ internal/server/handlers_collab.go | 34 ++- 2 files changed, 245 insertions(+), 3 deletions(-) create mode 100644 internal/server/bug3474_reval_fail_closed_test.go diff --git a/internal/server/bug3474_reval_fail_closed_test.go b/internal/server/bug3474_reval_fail_closed_test.go new file mode 100644 index 000000000..47e6bcc4d --- /dev/null +++ b/internal/server/bug3474_reval_fail_closed_test.go @@ -0,0 +1,214 @@ +package server + +import ( + "bytes" + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +// revalWarnCounter counts the collab revalidation-error warnings, passing +// every record on to the handler it wraps. +type revalWarnCounter struct { + slog.Handler + mu *sync.Mutex + count *int +} + +func (h revalWarnCounter) Handle(ctx context.Context, rec slog.Record) error { + // The stream helper's message too: before BUG-3474 the collab tick's + // credential re-check logged through it, so the sessions leg reaches its + // real assertion (does the frame persist?) on the unfixed code. + if strings.HasPrefix(rec.Message, "collab: revalidation") || strings.HasPrefix(rec.Message, "stream: credential re-check failed") { + h.mu.Lock() + *h.count++ + h.mu.Unlock() + } + return h.Handler.Handle(ctx, rec) +} + +func (h revalWarnCounter) WithAttrs(a []slog.Attr) slog.Handler { + return revalWarnCounter{Handler: h.Handler.WithAttrs(a), mu: h.mu, count: h.count} +} + +func (h revalWarnCounter) WithGroup(n string) slog.Handler { + return revalWarnCounter{Handler: h.Handler.WithGroup(n), mu: h.mu, count: h.count} +} + +// BUG-3474 (lead ruling): a revalidation tick that ERRORS must not leave the +// connection with the write permission it last had. It used to keep both the +// connection and its write flag, so an editor demoted while the store was +// failing kept writing until the first clean tick. Now the connection is kept +// but goes read-only, and the next clean tick restores write if the role +// still allows it. +// +// The faults are real, one per error branch of the tick: renaming sessions +// makes the credential re-check UNKNOWN (codex r1: it used to read as valid, +// so that tick went on and set write from the role); renaming +// workspace_members fails the membership lookup inside authorizeCollabAccess +// with a plain (non-denial) error; renaming items fails the tick's own +// GetItem. Frames persist to item_yjs_updates, which no rename touches. +func TestBUG3474_RevalErrorFailsClosedOnWrite(t *testing.T) { + for _, table := range []string{"sessions", "workspace_members", "items"} { + t.Run(table, func(t *testing.T) { runBUG3474FaultLeg(t, table) }) + } +} + +func runBUG3474FaultLeg(t *testing.T, table string) { + origInterval := collabMembershipRevalInterval + collabMembershipRevalInterval = 25 * time.Millisecond + defer func() { collabMembershipRevalInterval = origInterval }() + + var mu sync.Mutex + warns := 0 + // A fresh handler, not slog.Default().Handler(): the initial default + // handler writes through the log package, which routes back into slog + // once SetDefault replaces it, and the wrapped call deadlocks on log's + // own mutex. + prevLogger := slog.Default() + slog.SetDefault(slog.New(revalWarnCounter{Handler: slog.NewTextHandler(io.Discard, nil), mu: &mu, count: &warns})) + defer slog.SetDefault(prevLogger) + warnCount := func() int { + mu.Lock() + defer mu.Unlock() + return warns + } + + srv := testServerWithCollab(t) + bootstrapFirstUser(t, srv, "admin@test.com", "Admin") + ts := httptest.NewServer(srv) + defer ts.Close() + + ws, err := srv.store.CreateWorkspace(models.WorkspaceCreate{Name: "Reval fault"}) + if err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + col, err := srv.store.CreateCollection(ws.ID, models.CollectionCreate{Name: "Tasks", Schema: `{"fields":[]}`}) + if err != nil { + t.Fatalf("CreateCollection: %v", err) + } + item, err := srv.store.CreateItem(ws.ID, col.ID, models.ItemCreate{Title: "Doc", Fields: `{}`}) + if err != nil { + t.Fatalf("CreateItem: %v", err) + } + u, err := srv.store.CreateUser(models.UserCreate{ + Email: "fault@test.com", Name: "Fault", Password: "correct-horse-battery-staple", Role: "member", + }) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + if err := srv.store.AddWorkspaceMember(ws.ID, u.ID, "editor"); err != nil { + t.Fatalf("AddWorkspaceMember: %v", err) + } + token, err := srv.store.CreateSession(u.ID, "go-test", "127.0.0.1", "go-test", 24*time.Hour) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + cookies := []*http.Cookie{{Name: "pad_session", Value: token}} + + conn, resp, err := dialCollab(t, ts.URL, item.ID, cookies, "go-test") + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if resp.StatusCode != http.StatusSwitchingProtocols { + t.Fatalf("expected 101, got %d", resp.StatusCode) + } + + // The control: an editor's frame persists. + if err := conn.WriteMessage(websocket.BinaryMessage, []byte{0x00, 0xE1}); err != nil { + t.Fatalf("editor write: %v", err) + } + waitForOpLogRows(t, srv, item.ID, 1, 3*time.Second) + + // The fault, and its premise: a tick has met it. The warning is logged + // AFTER the connection is made read-only, so from here a frame must drop. + before := warnCount() + if _, err := srv.store.DB().Exec(`ALTER TABLE ` + table + ` RENAME TO ` + table + `_bug3474`); err != nil { + t.Fatalf("rename: %v", err) + } + restored := false + restore := func() { + if restored { + return + } + restored = true + if _, err := srv.store.DB().Exec(`ALTER TABLE ` + table + `_bug3474 RENAME TO ` + table); err != nil { + t.Fatalf("restore: %v", err) + } + } + defer restore() + deadline := time.Now().Add(3 * time.Second) + for warnCount() == before { + if time.Now().After(deadline) { + t.Fatal("no revalidation tick met the fault within 3s") + } + time.Sleep(5 * time.Millisecond) + } + + faulted := []byte{0x00, 0x22} + if err := conn.WriteMessage(websocket.BinaryMessage, faulted); err != nil { + t.Fatalf("faulted write: %v", err) + } + // In-process persistence is sub-millisecond, so a leaked frame usually + // shows inside this window. The decisive check is the one after recovery. + settle := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(settle) { + rows, err := srv.store.LoadYjsUpdatesSince(item.ID, 0) + if err != nil { + t.Fatalf("LoadYjsUpdatesSince: %v", err) + } + if len(rows) > 1 { + t.Fatalf("a frame sent while revalidation was failing persisted (%d rows): the connection kept write", len(rows)) + } + time.Sleep(10 * time.Millisecond) + } + + // Write comes back after a clean tick: still an editor, so the next + // successful revalidation restores it. Frames are re-sent until one of + // THEM lands (each distinct, so none is deduplicated), because the clean + // tick's timing is not observable from here. One connection's frames are + // handled in order, so once a recovery frame has landed the faulted frame + // has been handled too, and its absence is then a fact rather than a + // window (codex r1: a count could not tell the two apart). + restore() + sent := map[string]bool{} + landed := false + resend := time.Now().Add(5 * time.Second) + var rows []models.YjsUpdate + for i := byte(0); time.Now().Before(resend) && !landed; i++ { + frame := []byte{0x00, 0x30, i} + sent[string(frame)] = true + if err := conn.WriteMessage(websocket.BinaryMessage, frame); err != nil { + t.Fatalf("recovered write: %v", err) + } + time.Sleep(50 * time.Millisecond) + rows, err = srv.store.LoadYjsUpdatesSince(item.ID, 0) + if err != nil { + t.Fatalf("LoadYjsUpdatesSince: %v", err) + } + for _, row := range rows { + if sent[string(row.UpdateData)] { + landed = true + } + } + } + if !landed { + t.Fatal("write never came back after the store recovered") + } + for _, row := range rows { + if bytes.Equal(row.UpdateData, faulted) { + t.Fatal("the frame sent while revalidation was failing persisted: the connection kept write") + } + } +} diff --git a/internal/server/handlers_collab.go b/internal/server/handlers_collab.go index 9323c10bb..c2b9c026a 100644 --- a/internal/server/handlers_collab.go +++ b/internal/server/handlers_collab.go @@ -366,7 +366,8 @@ func (s *Server) collabRevalidationLoop( // `CloseConn` with a policy violation, matching the // item-disappeared branch below rather than inventing a fourth way // to end a collab connection. - if !s.streamCredentialStillValid(r) { + switch s.credentialLiveness(r) { + case credentialInvalid: slog.Info("collab: credential invalidated mid-stream, closing connection", "item_id", itemID, "user_id", userID, @@ -377,6 +378,19 @@ func (s *Server) collabRevalidationLoop( "Your session has ended.", ) return + case credentialUnknown: + // The re-check itself failed (a store error). Not a + // revocation, so the conn stays open, but it is an errored + // tick, so it goes READ-ONLY and skips authorization until a + // tick succeeds (BUG-3474, codex r1: this used to read as + // valid, and the tick went on to set write from the role). + s.collab.SetConnWritable(itemID, conn, false) + slog.Warn("collab: revalidation credential check failed; keeping connection open, read-only", + "item_id", itemID, + "user_id", userID, + ) + timer.Reset(interval) + continue } // Re-fetch the item every tick so a mid-session move // (item collection changed to one the user can't see) @@ -384,7 +398,10 @@ func (s *Server) collabRevalidationLoop( // snapshot captured at upgrade time isn't enough. fresh, ferr := s.store.GetItem(itemID) if ferr != nil { - slog.Warn("collab: revalidation GetItem failed; keeping connection open", + // FAIL CLOSED ON WRITE (BUG-3474): kept open, made read-only + // until a clean tick; see the default branch below. + s.collab.SetConnWritable(itemID, conn, false) + slog.Warn("collab: revalidation GetItem failed; keeping connection open, read-only", "item_id", itemID, "user_id", userID, "error", ferr, @@ -440,8 +457,19 @@ func (s *Server) collabRevalidationLoop( // operator notices a sustained pattern, but we // MUST NOT close the conn — a single failed // query shouldn't punt every active editor. + // + // FAIL CLOSED ON WRITE (BUG-3474, lead ruling): the conn + // stays open but goes READ-ONLY until a tick succeeds. It + // used to keep the write permission it last had, so an + // editor demoted while the store was failing kept writing + // until the first clean tick. The next successful tick + // restores write through SetConnWritable above if the role + // still allows it. The cost is an editor briefly read-only + // during a DB blip, which is cheaper than a demoted editor + // writing through one. + s.collab.SetConnWritable(itemID, conn, false) // Re-arm and try again on the next tick. - slog.Warn("collab: revalidation error; keeping connection open", + slog.Warn("collab: revalidation error; keeping connection open, read-only", "item_id", itemID, "user_id", userID, "error", err,