amadan

oficina / docs Public

Plan 5, tasks 1-2: seq becomes a cursor, and a hub to announce it

ApplyUpdate returns the seq it assigned, GET /d/{sqid}/updates takes
?since=N, and both update routes answer with a Docs-Seq header. The
header is derived from the rows being written and never from a second
MAX(seq) query: reads go to the read pool and writes to the
single-connection write pool, so a row committing between the two
queries would let a client advance its cursor over content it was
never sent. cursor_test.go's TestEmptySliceEchoesSinceRatherThanThe
CurrentMax is written to fail against that implementation.

The hub is one process's fan-out, keyed by document. Its channels are
one deep and its sends are non-blocking, which is safe because the
seq it carries is advisory: a subscriber that misses ten
notifications still asks for ?since=its own cursor and gets all ten
updates. It lives beside the handlers rather than inside Store, so
Store stays persistence and doc_updates stays the plain append-only
table section 4.3 called for.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 2ab2df0419685f255d8d78eceb60d5a7dd920d29 parent 2787767
10 files changed, +635 −16
  • internal/docs/app.go +1 −1
  • internal/docs/cursor_test.go +266 −0
  • internal/docs/editor_view_test.go +1 −1
  • internal/docs/handlers.go +62 −2
  • internal/docs/hub.go +105 −0
  • internal/docs/hub_test.go +154 −0
  • internal/docs/render.go +9 −0
  • internal/docs/seed.go +6 −1
  • internal/docs/store.go +27 −7
  • internal/docs/store_test.go +4 −4

Intent

full ledger: worktree-live-sync

diff --git a/internal/docs/app.go b/internal/docs/app.go
index 07a1a02..c2e2cfc 100644
--- a/internal/docs/app.go
+++ b/internal/docs/app.go
@@ -400,7 +400,7 @@ func New(cfg Config) (*App, error) {
return nil, fmt.Errorf("docs: build auth: %w", err)
}
- h := &handlers{render: render, logger: logger, store: store}
+ h := &handlers{render: render, logger: logger, store: store, hub: newHub()}
r := chi.NewRouter()
// Mounted above every route, per rastrillo SKILL.md §5: refuses a
diff --git a/internal/docs/cursor_test.go b/internal/docs/cursor_test.go
new file mode 100644
index 0000000..f9606fe
--- /dev/null
+++ b/internal/docs/cursor_test.go
@@ -0,0 +1,266 @@
+package docs
+
+import (
+ "bytes"
+ "context"
+ "net/http"
+ "strconv"
+ "testing"
+)
+
+// The cursor protocol (spec 2026-09-01-live-sync-design.md §4) is what
+// lets a second tab catch up without reloading: ?since=N slices the
+// log, and Docs-Seq tells the client how far the bytes it just
+// received actually reach.
+//
+// These tests live in their own file rather than in updates_test.go
+// because they test a protocol rather than the update endpoints'
+// existing behaviour, and because the D2 test below is a regression
+// test for a specific rejected implementation — the kind that goes
+// stale silently if it ends up filed under a general heading.
+
+// postUpdate posts one real Yjs update as owner and returns the
+// response, so a test that only needs "another row exists now" does
+// not restate the whole request each time.
+func postUpdate(t *testing.T, app *App, sqid, address string, update []byte) *http.Response {
+ t.Helper()
+ rec := serve(app, app.TestRequest(t, address, http.MethodPost,
+ "/d/"+sqid+"/updates", bytes.NewReader(update)))
+ return rec.Result()
+}
+
+// docsSeq reads the Docs-Seq header off a response, failing the test
+// if it is missing or unparseable — every response from either update
+// route must carry one, so an absent header is a bug rather than a
+// case to tolerate.
+func docsSeq(t *testing.T, res *http.Response) int64 {
+ t.Helper()
+ raw := res.Header.Get("Docs-Seq")
+ if raw == "" {
+ return -1
+ }
+ n, err := strconv.ParseInt(raw, 10, 64)
+ if err != nil {
+ t.Fatalf("Docs-Seq: %q does not parse as an integer: %v", raw, err)
+ }
+ return n
+}
+
+// A POST answers with the seq of the row it just wrote, which is what
+// lets a tab advance its own cursor without a follow-up fetch (D3).
+func TestPostAnswersWithTheSeqItWrote(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Cursor")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ res := postUpdate(t, app, doc.Sqid, "owner@example.com", validUpdate(t))
+ if res.StatusCode != http.StatusNoContent {
+ t.Fatalf("POST = %d, want 204", res.StatusCode)
+ }
+ if got := docsSeq(t, res); got != 1 {
+ t.Errorf("Docs-Seq after the first POST = %d, want 1", got)
+ }
+
+ res = postUpdate(t, app, doc.Sqid, "owner@example.com", validUpdate(t))
+ if got := docsSeq(t, res); got != 2 {
+ t.Errorf("Docs-Seq after the second POST = %d, want 2", got)
+ }
+}
+
+// ?since=N returns only what came after N, and says so.
+func TestSinceSlicesTheLog(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Sliced")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ update := validUpdate(t)
+ postUpdate(t, app, doc.Sqid, "owner@example.com", update)
+ postUpdate(t, app, doc.Sqid, "owner@example.com", update)
+
+ full := serve(app, app.TestRequest(t, "owner@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/updates", nil))
+ if full.Code != http.StatusOK {
+ t.Fatalf("full GET = %d, want 200", full.Code)
+ }
+ if got := docsSeq(t, full.Result()); got != 2 {
+ t.Errorf("Docs-Seq on a full GET = %d, want 2 (the highest row returned)", got)
+ }
+ fullLen := full.Body.Len()
+
+ sliced := serve(app, app.TestRequest(t, "owner@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/updates?since=1", nil))
+ if sliced.Code != http.StatusOK {
+ t.Fatalf("sliced GET = %d, want 200", sliced.Code)
+ }
+ if sliced.Body.Len() >= fullLen {
+ t.Errorf("?since=1 returned %d bytes, the full log is %d — the slice was not applied",
+ sliced.Body.Len(), fullLen)
+ }
+ if sliced.Body.Len() == 0 {
+ t.Error("?since=1 returned nothing, but row 2 exists")
+ }
+ if got := docsSeq(t, sliced.Result()); got != 2 {
+ t.Errorf("Docs-Seq on ?since=1 = %d, want 2", got)
+ }
+}
+
+// TestEmptySliceEchoesSinceRatherThanTheCurrentMax is the regression
+// test for the design's D2, and it is written to fail against the
+// implementation the review rejected.
+//
+// An earlier draft defined Docs-Seq as "the highest seq included, or
+// the document's CURRENT MAX when the slice is empty". That second
+// clause needs its own MAX(seq) query, and reads go to the read pool
+// while writes go to the single-connection write pool — nothing
+// serializes the two. So a row could commit between the slice query
+// and the MAX query, and the client would receive an empty body
+// alongside a header naming a row it had just been told nothing
+// about. A client that trusts the header (which is the whole point of
+// the header) advances its cursor past that row and never fetches it
+// again: content present on the server, permanently invisible in that
+// tab until a manual reload.
+//
+// Deriving the header from the bytes in hand makes that
+// unrepresentable, which is what this asserts: with a row that the
+// slice deliberately excludes, the header must still echo `since`
+// rather than reporting the higher row.
+func TestEmptySliceEchoesSinceRatherThanTheCurrentMax(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Echo")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ // One row exists (seq 1). Asking for everything after seq 1 is
+ // empty *now*; the point of the test is what the header says.
+ postUpdate(t, app, doc.Sqid, "owner@example.com", validUpdate(t))
+
+ empty := serve(app, app.TestRequest(t, "owner@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/updates?since=1", nil))
+ if empty.Code != http.StatusOK {
+ t.Fatalf("empty slice GET = %d, want 200", empty.Code)
+ }
+ if empty.Body.Len() != 0 {
+ t.Fatalf("?since=1 returned %d bytes, want none", empty.Body.Len())
+ }
+ if got := docsSeq(t, empty.Result()); got != 1 {
+ t.Errorf("Docs-Seq on an empty slice = %d, want 1 (an echo of since)", got)
+ }
+
+ // Now the sharp case: a row the client has NOT been given exists,
+ // and the answer must still not name it. A MAX(seq) implementation
+ // returns 2 here and loses row 2 in any client that trusts it.
+ postUpdate(t, app, doc.Sqid, "owner@example.com", validUpdate(t))
+ stale := serve(app, app.TestRequest(t, "owner@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/updates?since=2", nil))
+ if stale.Body.Len() != 0 {
+ t.Fatalf("?since=2 returned %d bytes, want none", stale.Body.Len())
+ }
+ if got := docsSeq(t, stale.Result()); got != 2 {
+ t.Errorf("Docs-Seq on an empty slice = %d, want 2 (an echo of since, never a row the client did not receive)", got)
+ }
+}
+
+// An unparseable or negative since is the client's error, and saying
+// so beats guessing at an intent: silently treating it as 0 would
+// resend the whole log, which is the expensive answer to a question
+// nobody asked.
+func TestMalformedSinceIsBadRequest(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Malformed")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ for _, since := range []string{"banana", "-1", "1.5", "9999999999999999999999"} {
+ rec := serve(app, app.TestRequest(t, "owner@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/updates?since="+since, nil))
+ if rec.Code != http.StatusBadRequest {
+ t.Errorf("?since=%s = %d, want 400", since, rec.Code)
+ }
+ }
+}
+
+// A document nobody has written to yet answers 0, so a fresh editor
+// starts its cursor at 0 rather than at nothing.
+func TestDocsSeqOnAnEmptyDocumentIsZero(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Fresh")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ rec := serve(app, app.TestRequest(t, "owner@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/updates", nil))
+ if rec.Code != http.StatusOK {
+ t.Fatalf("GET = %d, want 200", rec.Code)
+ }
+ if got := docsSeq(t, rec.Result()); got != 0 {
+ t.Errorf("Docs-Seq on an empty document = %d, want 0", got)
+ }
+}
+
+// ApplyUpdate's seq is only meaningful when it returns no error
+// (design D4). A rejected update must not consume a seq, or a client
+// that trusted the number would skip a row that later takes it.
+func TestRejectedUpdateConsumesNoSeq(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Rejected")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ seq, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t))
+ if err != nil {
+ t.Fatalf("ApplyUpdate: %v", err)
+ }
+ if seq != 1 {
+ t.Fatalf("first ApplyUpdate returned seq %d, want 1", seq)
+ }
+
+ if _, err := app.Store().ApplyUpdate(ctx, doc, []byte{0xff, 0x00, 0x13, 0x37}); err == nil {
+ t.Fatal("ApplyUpdate accepted garbage")
+ }
+
+ seq, err = app.Store().ApplyUpdate(ctx, doc, validUpdate(t))
+ if err != nil {
+ t.Fatalf("ApplyUpdate after a rejection: %v", err)
+ }
+ if seq != 2 {
+ t.Errorf("ApplyUpdate after a rejection returned seq %d, want 2 — a refused update consumed a seq", seq)
+ }
+}
diff --git a/internal/docs/editor_view_test.go b/internal/docs/editor_view_test.go
index 52e5560..c2c1966 100644
--- a/internal/docs/editor_view_test.go
+++ b/internal/docs/editor_view_test.go
@@ -20,7 +20,7 @@ func TestShowRendersContentWithoutScript(t *testing.T) {
owner := app.TestPerson(t, "reader@example.com")
doc, _ := app.Store().Create(ctx, owner, "Readable")
update := readFixture(t, "../../fixtures/vectors/fragment/simple.bin")
- if err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
t.Fatalf("ApplyUpdate: %v", err)
}
diff --git a/internal/docs/handlers.go b/internal/docs/handlers.go
index 778790d..cc579d7 100644
--- a/internal/docs/handlers.go
+++ b/internal/docs/handlers.go
@@ -6,6 +6,7 @@ import (
"io"
"log/slog"
"net/http"
+ "strconv"
"strings"
"time"
@@ -35,6 +36,15 @@ type handlers struct {
render *renderer
logger *slog.Logger
store *Store
+
+ // hub is the live-update fan-out (hub.go). It lives here rather
+ // than on Store on purpose: Store is persistence and knows nothing
+ // about transports, and doc_updates stays the plain append-only
+ // table the design spec's §4.3 insisted on when it rejected
+ // rastrillo/eventlog. The consequence is that a writer which does
+ // not go through updatesPost does not notify — today that is only
+ // sample seeding, which runs before any editor can be open.
+ hub *hub
}
// listPage is GET /'s view data (spec §6.4). Title and the group
@@ -314,9 +324,14 @@ func (h *handlers) updatesGet(w http.ResponseWriter, r *http.Request) {
return
}
+ since, ok := parseSince(w, r)
+ if !ok {
+ return
+ }
+
var updates []DocUpdate
if err := h.store.db.WithContext(r.Context()).
- Where("doc_id = ?", doc.ID).
+ Where("doc_id = ? AND seq > ?", doc.ID, since).
Order("seq ASC").
Find(&updates).Error; err != nil {
h.logger.Error("docs: load update log", "sqid", doc.Sqid, "err", err)
@@ -334,7 +349,19 @@ func (h *handlers) updatesGet(w http.ResponseWriter, r *http.Request) {
return
}
+ // Docs-Seq is derived from the rows in hand — max(since, the
+ // highest seq actually being written) — and never from a second
+ // MAX(seq) query. See parseSince's own doc comment and the design's
+ // D2 for the update this loses if that is ever "simplified".
+ highest := since
+ for _, u := range updates {
+ if u.Seq > highest {
+ highest = u.Seq
+ }
+ }
+
w.Header().Set("Content-Type", "application/octet-stream")
+ w.Header().Set("Docs-Seq", strconv.FormatInt(highest, 10))
for _, u := range updates {
if _, err := w.Write(u.UpdateBin); err != nil {
// The client went away mid-response; nothing left to do
@@ -375,7 +402,8 @@ func (h *handlers) updatesPost(w http.ResponseWriter, r *http.Request) {
return
}
- if err := h.store.ApplyUpdate(r.Context(), doc, update); err != nil {
+ seq, err := h.store.ApplyUpdate(r.Context(), doc, update)
+ if err != nil {
if updateIsClientFault(err) {
// Info, not Error: a client sending bytes the core
// cleanly rejects is an ordinary, expected outcome of
@@ -394,9 +422,41 @@ func (h *handlers) updatesPost(w http.ResponseWriter, r *http.Request) {
return
}
+ // Both of these happen only on a nil error, which is what makes
+ // the seq trustworthy (design D4). An error here can be returned
+ // over an already-committed row — ApplyUpdate's trailing
+ // updated_at bump lives outside its transaction — and in that case
+ // the client retries, writes a duplicate row Yjs integrates
+ // idempotently, and the broadcast happens on the retry instead.
+ w.Header().Set("Docs-Seq", strconv.FormatInt(seq, 10))
+ h.hub.broadcast(doc.ID, seq)
+
w.WriteHeader(http.StatusNoContent)
}
+// parseSince reads the ?since= cursor off a request: the highest seq
+// the client already holds, so the response can carry only what came
+// after it. Absent means 0, which asks for the whole log — what the
+// editor's first load wants.
+//
+// A malformed or negative value is a 400 rather than a silent 0.
+// Treating "banana" as "send everything" would answer an expensive
+// question nobody asked, and would hide a client bug behind a
+// response that looks fine; a client that cannot say where it is
+// should be told so.
+func parseSince(w http.ResponseWriter, r *http.Request) (int64, bool) {
+ raw := r.URL.Query().Get("since")
+ if raw == "" {
+ return 0, true
+ }
+ since, err := strconv.ParseInt(raw, 10, 64)
+ if err != nil || since < 0 {
+ http.Error(w, rastrillo.T(r, "docs.error.badsince"), http.StatusBadRequest)
+ return 0, false
+ }
+ return since, true
+}
+
// updateIsClientFault classifies a Store.ApplyUpdate error: true means
// the core looked at the bytes the client just posted and refused
// them — a 400, the client's problem, and unremarkable enough to log
diff --git a/internal/docs/hub.go b/internal/docs/hub.go
new file mode 100644
index 0000000..7269fa7
--- /dev/null
+++ b/internal/docs/hub.go
@@ -0,0 +1,105 @@
+package docs
+
+import "sync"
+
+// hub is this instance's in-process fan-out: one document's accepted
+// update, announced to whoever currently has that document open. It
+// is the "broadcast primitive" rastrillo's own SSE handler
+// (jobs/events.go) says it deliberately did without — that one polls
+// its store once a second because it had nothing to wait on, and
+// this exists precisely so the stream handler here can wait instead
+// of poll.
+//
+// Scope, deliberately: one process. Docs is one instance per team
+// (the design spec's opening line, and its §3 "One SQLite file per
+// instance"), so there is no second process to reach and nothing here
+// needs a broker, a database channel, or a network hop. If Docs ever
+// runs more than one process per team, this type is the seam that has
+// to change, and nothing else does.
+//
+// What travels is a seq, and it is ADVISORY. It says "there is
+// something at least this new"; it never says "here is what changed"
+// and a subscriber must not treat it as a cursor assignment. That is
+// what makes the lossy one-slot channel below correct rather than
+// merely convenient: a subscriber that misses ten notifications still
+// fetches ?since=<its own cursor> and receives all ten updates.
+type hub struct {
+ mu sync.Mutex
+ subs map[int64]map[chan int64]struct{}
+}
+
+func newHub() *hub {
+ return &hub{subs: make(map[int64]map[chan int64]struct{})}
+}
+
+// subscribe registers interest in docID and returns the channel to
+// wait on plus the function that unregisters it. The channel is
+// buffered one deep: it carries "something happened", not a queue of
+// everything that did.
+//
+// The returned stop is idempotent, because its caller (the stream
+// handler) defers it and may also have reason to stop early. Calling
+// it twice, or after the hub has forgotten the document, is a no-op
+// rather than a panic on a closed channel — note that stop does not
+// close the channel at all. Closing it would race a broadcast that
+// has already passed the map lookup and is about to send; leaving it
+// to the garbage collector costs nothing, since nothing else holds a
+// reference once it is out of the map.
+func (h *hub) subscribe(docID int64) (<-chan int64, func()) {
+ ch := make(chan int64, 1)
+
+ h.mu.Lock()
+ if h.subs[docID] == nil {
+ h.subs[docID] = make(map[chan int64]struct{})
+ }
+ h.subs[docID][ch] = struct{}{}
+ h.mu.Unlock()
+
+ var once sync.Once
+ return ch, func() {
+ once.Do(func() {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ subs, ok := h.subs[docID]
+ if !ok {
+ return
+ }
+ delete(subs, ch)
+ // Drop the inner map with its last subscriber, or this
+ // map accumulates one empty entry per document ever
+ // opened for the life of the process.
+ if len(subs) == 0 {
+ delete(h.subs, docID)
+ }
+ })
+ }
+}
+
+// broadcast tells docID's subscribers that seq exists.
+//
+// It runs on the HTTP request goroutine that just accepted an update
+// (handlers.go's updatesPost), so it must never block: a browser tab
+// on a wedged connection would otherwise hold a stranger's POST open.
+// Hence the default case — a full slot already means "go and fetch",
+// and a second notification would tell the subscriber nothing the
+// first one does not.
+func (h *hub) broadcast(docID, seq int64) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+
+ for ch := range h.subs[docID] {
+ select {
+ case ch <- seq:
+ default:
+ }
+ }
+}
+
+// documentCount reports how many documents have at least one
+// subscriber. It exists for the tests that assert this map does not
+// grow forever; nothing in the app reads it.
+func (h *hub) documentCount() int {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return len(h.subs)
+}
diff --git a/internal/docs/hub_test.go b/internal/docs/hub_test.go
new file mode 100644
index 0000000..827d0b3
--- /dev/null
+++ b/internal/docs/hub_test.go
@@ -0,0 +1,154 @@
+package docs
+
+import (
+ "sync"
+ "testing"
+ "time"
+)
+
+// recv waits briefly for a seq on ch. The wait exists because a
+// broadcast and its delivery are not the same instant; it is not a
+// synchronisation mechanism the code under test relies on, only a
+// bound on how long a test is willing to be wrong.
+func recv(t *testing.T, ch <-chan int64) (int64, bool) {
+ t.Helper()
+ select {
+ case seq := <-ch:
+ return seq, true
+ case <-time.After(time.Second):
+ return 0, false
+ }
+}
+
+// quiet asserts nothing arrives. Shorter than recv's timeout because
+// a false negative here costs a slow test and a false positive costs
+// a wrong belief about isolation.
+func quiet(t *testing.T, ch <-chan int64) bool {
+ t.Helper()
+ select {
+ case seq := <-ch:
+ t.Errorf("received seq %d on a channel that should have been quiet", seq)
+ return false
+ case <-time.After(50 * time.Millisecond):
+ return true
+ }
+}
+
+func TestHubDeliversToItsOwnDocumentOnly(t *testing.T) {
+ h := newHub()
+
+ mine, stop := h.subscribe(1)
+ defer stop()
+ theirs, stopTheirs := h.subscribe(2)
+ defer stopTheirs()
+
+ h.broadcast(1, 7)
+
+ if seq, ok := recv(t, mine); !ok || seq != 7 {
+ t.Errorf("subscriber to document 1 got (%d, %v), want (7, true)", seq, ok)
+ }
+ quiet(t, theirs)
+}
+
+// A broadcast with nobody listening is not an error and not a leak.
+func TestHubBroadcastWithNoSubscribers(t *testing.T) {
+ h := newHub()
+ h.broadcast(42, 1) // must not panic or block
+
+ ch, stop := h.subscribe(42)
+ defer stop()
+ quiet(t, ch) // and must not have been queued up for a later arrival
+}
+
+// Unsubscribing stops delivery, and doing it twice is safe: the SSE
+// handler defers its stop and may also stop early on a write failure.
+func TestHubUnsubscribeStopsDeliveryAndIsIdempotent(t *testing.T) {
+ h := newHub()
+ ch, stop := h.subscribe(1)
+
+ stop()
+ stop()
+
+ h.broadcast(1, 3)
+ quiet(t, ch)
+}
+
+// TestHubSlowSubscriberNeverBlocksABroadcaster is the property the
+// whole design rests on: broadcast runs on the HTTP request goroutine
+// that just accepted an update, so a subscriber that is not reading —
+// a browser tab on a wedged connection, a handler between select
+// turns — must never be able to hold that request open. The slot is
+// one deep and sends are non-blocking, which is safe precisely
+// because the seq is advisory: a client that misses a notification
+// still fetches ?since=cursor and gets everything.
+func TestHubSlowSubscriberNeverBlocksABroadcaster(t *testing.T) {
+ h := newHub()
+ ch, stop := h.subscribe(1)
+ defer stop()
+
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ for seq := int64(1); seq <= 100; seq++ {
+ h.broadcast(1, seq)
+ }
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("broadcast blocked on a subscriber that was not reading")
+ }
+
+ // The subscriber missed most of them, which is fine and is the
+ // point — but it must still be told that something happened, or
+ // it would never fetch.
+ if _, ok := recv(t, ch); !ok {
+ t.Error("a subscriber that fell behind was left with no pending notification at all")
+ }
+}
+
+// The map must not grow forever: a document with no subscribers left
+// holds no entry.
+func TestHubForgetsDocumentsWithNoSubscribers(t *testing.T) {
+ h := newHub()
+
+ _, stopA := h.subscribe(1)
+ _, stopB := h.subscribe(1)
+
+ stopA()
+ if got := h.documentCount(); got != 1 {
+ t.Errorf("after one of two subscribers left, hub tracks %d documents, want 1", got)
+ }
+ stopB()
+ if got := h.documentCount(); got != 0 {
+ t.Errorf("after the last subscriber left, hub tracks %d documents, want 0", got)
+ }
+}
+
+// Run with -race, this is the test that catches a missing lock.
+func TestHubUnderConcurrentUse(t *testing.T) {
+ h := newHub()
+
+ var wg sync.WaitGroup
+ for i := 0; i < 8; i++ {
+ wg.Add(1)
+ go func(n int) {
+ defer wg.Done()
+ for j := 0; j < 50; j++ {
+ ch, stop := h.subscribe(int64(n % 3))
+ h.broadcast(int64(n%3), int64(j))
+ select {
+ case <-ch:
+ default:
+ }
+ stop()
+ }
+ }(i)
+ }
+ wg.Wait()
+
+ if got := h.documentCount(); got != 0 {
+ t.Errorf("hub tracks %d documents after every subscriber left, want 0", got)
+ }
+}
diff --git a/internal/docs/render.go b/internal/docs/render.go
index 5f221cd..5e09493 100644
--- a/internal/docs/render.go
+++ b/internal/docs/render.go
@@ -75,6 +75,15 @@ var Catalog = rastrillo.Catalog{
// browser sees for it (Task 5).
"docs.error.updatetoolarge": "That update is too large.",
+ // docs.error.badsince answers GET /d/{sqid}/updates when the
+ // ?since= cursor is not a non-negative integer (handlers.go's
+ // parseSince). Only a broken or hand-typed client can produce it —
+ // the editor's own cursor comes from a Docs-Seq header this app
+ // wrote — so this is a key nobody should ever see, kept a catalog
+ // key anyway because "nobody should see it" is exactly the
+ // reasoning that leaves an untranslated literal in the codebase.
+ "docs.error.badsince": "That update cursor is not valid.",
+
// docs.document.untitled is POST /documents' fallback title when
// the submitted one is empty or all whitespace — boilerplate copy
// this app writes, not the caller's own input, so it is a catalog
diff --git a/internal/docs/seed.go b/internal/docs/seed.go
index 7a4dbc4..dce1f72 100644
--- a/internal/docs/seed.go
+++ b/internal/docs/seed.go
@@ -91,7 +91,12 @@ func Seed(ctx context.Context, store *Store, owner int64) error {
if err != nil {
return fmt.Errorf("docs: seed: read update for %q: %w", sd.Title, err)
}
- if err := store.ApplyUpdate(ctx, doc, update); err != nil {
+ // The assigned seq is discarded here on purpose: seeding runs
+ // when a person's row is first created, before any editor of
+ // theirs can be open, so there is nobody subscribed to notify
+ // (design D5's accepted consequence — the broadcast lives in
+ // updatesPost, not in Store).
+ if _, err := store.ApplyUpdate(ctx, doc, update); err != nil {
return fmt.Errorf("docs: seed: apply update for %q: %w", sd.Title, err)
}
diff --git a/internal/docs/store.go b/internal/docs/store.go
index 2427f5e..8e0693b 100644
--- a/internal/docs/store.go
+++ b/internal/docs/store.go
@@ -491,22 +491,22 @@ func (e *updateRejection) Unwrap() error { return e.err }
// owner makes the Where match zero rows, First returns
// gorm.ErrRecordNotFound, and this method returns before replay is
// even called — no core touched, no row written.
-func (s *Store) ApplyUpdate(ctx context.Context, doc *Document, update []byte) error {
+func (s *Store) ApplyUpdate(ctx context.Context, doc *Document, update []byte) (int64, error) {
var ownerCheck Document
if err := owned(s.db.WithContext(ctx), doc.OwnerID).
Where("id = ?", doc.ID).
First(&ownerCheck).Error; err != nil {
- return fmt.Errorf("docs: apply update to document %d: verify owner %d: %w", doc.ID, doc.OwnerID, err)
+ return 0, fmt.Errorf("docs: apply update to document %d: verify owner %d: %w", doc.ID, doc.OwnerID, err)
}
h, err := s.replay(ctx, doc)
if err != nil {
- return err
+ return 0, err
}
defer h.Close(ctx)
if err := h.Apply(ctx, update); err != nil {
- return fmt.Errorf("docs: apply update to document %d: %w", doc.ID, &updateRejection{err})
+ return 0, fmt.Errorf("docs: apply update to document %d: %w", doc.ID, &updateRejection{err})
}
// MAX(seq) then INSERT used to be two separate statements, not one
@@ -536,6 +536,14 @@ func (s *Store) ApplyUpdate(ctx context.Context, doc *Document, update []byte) e
// updates_test.go's TestConcurrentUpdatesDoNotCollideOnSeq (21
// concurrent POSTs to one document) genuinely exhausted it on a
// real run before this transaction replaced it.
+ // assignedSeq is captured out of the transaction closure so this
+ // method can return it. It is only ever read after Transaction
+ // returns nil, so it names a row that is committed and durable —
+ // which is exactly the contract this method's callers rely on
+ // (design D4: the seq is meaningful only when err == nil). Nothing
+ // about the single-writer MAX(seq)->INSERT guarantee above changes
+ // by observing the number the closure already computed.
+ var assignedSeq int64
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var maxSeq int64
if err := tx.Model(&DocUpdate{}).
@@ -554,9 +562,10 @@ func (s *Store) ApplyUpdate(ctx context.Context, doc *Document, update []byte) e
if err := tx.Create(row).Error; err != nil {
return fmt.Errorf("docs: persist update for document %d: %w", doc.ID, err)
}
+ assignedSeq = row.Seq
return nil
}); err != nil {
- return err
+ return 0, err
}
// Bumping documents.updated_at here — not just appending the
@@ -581,7 +590,18 @@ func (s *Store) ApplyUpdate(ctx context.Context, doc *Document, update []byte) e
Model(&Document{}).
Where("id = ?", doc.ID).
Update("updated_at", time.Now().UTC()).Error; err != nil {
- return fmt.Errorf("docs: bump updated_at for document %d: %w", doc.ID, err)
+ // The doc_updates row above is already committed at this
+ // point, so this error is returned over durable content. That
+ // is deliberate and its consequences are named in the design
+ // (D4): the caller must not broadcast and must not report a
+ // seq, the client's save queue treats the resulting 500 as
+ // retryable, and the retry writes a duplicate row — which Yjs
+ // integrates idempotently, costing one row against
+ // maxRetainedUpdates rather than any content. Returning 0 here
+ // rather than assignedSeq is what makes "seq is meaningful
+ // only on a nil error" true by construction instead of by
+ // convention.
+ return 0, fmt.Errorf("docs: bump updated_at for document %d: %w", doc.ID, err)
}
- return nil
+ return assignedSeq, nil
}
diff --git a/internal/docs/store_test.go b/internal/docs/store_test.go
index 1a4cc97..886cb0e 100644
--- a/internal/docs/store_test.go
+++ b/internal/docs/store_test.go
@@ -67,7 +67,7 @@ func TestTextComesFromTheCore(t *testing.T) {
// but targets root "content", which Store.Text no longer reads —
// it would read back "", proving nothing about the CRDT.
update := docs.ReadFragmentFixture(t, "hello")
- if err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
t.Fatalf("ApplyUpdate: %v", err)
}
@@ -125,7 +125,7 @@ func TestListOrdersByUpdatedAt(t *testing.T) {
time.Sleep(5 * time.Millisecond)
update := docs.ReadFixture(t, "hello")
- if err := app.Store().ApplyUpdate(ctx, older, update); err != nil {
+ if _, err := app.Store().ApplyUpdate(ctx, older, update); err != nil {
t.Fatalf("ApplyUpdate(older): %v", err)
}
@@ -253,7 +253,7 @@ func TestApplyUpdateRefusesAnotherOwnersDocument(t *testing.T) {
// content assertions below go through Store.Text, which reads root
// "prosemirror" now — see TestTextComesFromTheCore's own comment.
update := docs.ReadFragmentFixture(t, "hello")
- if err := app.Store().ApplyUpdate(ctx, &tampered, update); err == nil {
+ if _, err := app.Store().ApplyUpdate(ctx, &tampered, update); err == nil {
t.Fatal("ApplyUpdate returned nil for a mismatched owner — want a real error, before anything is written")
}
@@ -275,7 +275,7 @@ func TestApplyUpdateRefusesAnotherOwnersDocument(t *testing.T) {
// Prove the fix does not break the working path: the legitimate
// owner's own ApplyUpdate must still apply end to end.
- if err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
t.Fatalf("ApplyUpdate (legitimate owner): %v", err)
}
legit, err := app.Store().Text(ctx, doc)