amadan

oficina / docs Public

Plan 5, task 3: the stream endpoint

GET /d/{sqid}/stream is a Server-Sent Events feed that says "this
document has content through seq N" and carries no document bytes at
all. The client fetches those through the binary endpoint it already
used at mount, so there is exactly one path that turns stored updates
into a browser's Y.Doc rather than two that can drift, and SSE's
text-only framing costs no base64.

It follows rastrillo's own jobs/events.go recipe, with each piece
carried for its own reason: the write deadline is what lets a stream
coexist with serve's zero WriteTimeout, the heartbeat is what makes
that deadline ever fire on a quiet connection, and the TTL is half of
the hibernation story. The one deliberate divergence is that
handler's one-second database tick, which existed because it had no
broadcast primitive to wait on; this waits on the hub.

Subscribing happens before the current seq is read, never after. The
other order lets an update commit into the gap between the two and
produce no notification at all, leaving a tab current-looking and
stale until the next edit.

Stream tests run against a real httptest.NewServer: a recorder hands
over what a handler wrote once it returns, and a stream handler does
not return until the connection is over, so a recorder cannot express
the only interesting question about one.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 1d17ae1824b57850ea37a4132e0dd28fd76851ff parent 2ab2df0
6 files changed, +525 −1
  • internal/docs/app.go +7 −1
  • internal/docs/handlers.go +7 −0
  • internal/docs/store.go +21 −0
  • internal/docs/stream.go +170 −0
  • internal/docs/stream_test.go +309 −0
  • internal/docs/testing.go +11 −0

Intent

full ledger: worktree-live-sync

diff --git a/internal/docs/app.go b/internal/docs/app.go
index c2e2cfc..c09dd85 100644
--- a/internal/docs/app.go
+++ b/internal/docs/app.go
@@ -273,6 +273,7 @@ type App struct {
logger *slog.Logger
core *core.Runtime
store *Store
+ handlers *handlers
mailRecorder *recordingMailer // nil outside development; see auth.go
}
@@ -451,12 +452,17 @@ func New(cfg Config) (*App, error) {
// to a document through this route.
r.Get("/d/{sqid}/updates", h.updatesGet)
r.Post("/d/{sqid}/updates", h.updatesPost)
+ // Plan 5: the live-update stream (stream.go). A GET, so
+ // csrf.Protect passes it through untouched, and owner-scoped
+ // through the same getDocumentForUpdates helper as the two
+ // routes above — a stranger gets 404, never 403.
+ r.Get("/d/{sqid}/stream", h.stream)
})
mux := http.NewServeMux()
mux.Handle("/", r)
- return &App{cfg: cfg, db: d, mux: mux, logger: logger, core: rt, store: store, mailRecorder: mailRecorder}, nil
+ return &App{cfg: cfg, db: d, mux: mux, logger: logger, core: rt, store: store, handlers: h, mailRecorder: mailRecorder}, nil
}
// Store returns the app's document store — models.go's Document,
diff --git a/internal/docs/handlers.go b/internal/docs/handlers.go
index cc579d7..26df63e 100644
--- a/internal/docs/handlers.go
+++ b/internal/docs/handlers.go
@@ -45,6 +45,13 @@ type handlers struct {
// not go through updatesPost does not notify — today that is only
// sample seeding, which runs before any editor can be open.
hub *hub
+
+ // streamHeartbeat and streamTTL override stream.go's own defaults.
+ // Zero means "use the default", so nothing but a test ever sets
+ // them — see heartbeatInterval/ttlInterval for why the seam
+ // exists at all.
+ streamHeartbeat time.Duration
+ streamTTL time.Duration
}
// listPage is GET /'s view data (spec §6.4). Title and the group
diff --git a/internal/docs/store.go b/internal/docs/store.go
index 8e0693b..8d4dcfa 100644
--- a/internal/docs/store.go
+++ b/internal/docs/store.go
@@ -450,6 +450,27 @@ type updateRejection struct{ err error }
func (e *updateRejection) Error() string { return e.err.Error() }
func (e *updateRejection) Unwrap() error { return e.err }
+// currentSeq reports the highest seq in doc's update log, or 0 for a
+// document nothing has been written to yet.
+//
+// This is the stream handler's connect-time answer (stream.go) and
+// nothing else reads it. Note what it is NOT for: filling the
+// Docs-Seq header on GET /d/{sqid}/updates. That header is derived
+// from the rows being returned, precisely so that no query like this
+// one can observe a row the response does not carry — see the design
+// spec's D2 for the update that would otherwise go missing.
+func (s *Store) currentSeq(ctx context.Context, docID int64) (int64, error) {
+ var seq int64
+ if err := s.db.WithContext(ctx).
+ Model(&DocUpdate{}).
+ Where("doc_id = ?", docID).
+ Select("COALESCE(MAX(seq), 0)").
+ Scan(&seq).Error; err != nil {
+ return 0, fmt.Errorf("docs: current seq for document %d: %w", docID, err)
+ }
+ return seq, nil
+}
+
// ApplyUpdate applies a new Yjs update to doc: it replays the
// existing log to rebuild current state, applies update against that
// live core instance, and only once the core has accepted it does it
diff --git a/internal/docs/stream.go b/internal/docs/stream.go
new file mode 100644
index 0000000..35ce48e
--- /dev/null
+++ b/internal/docs/stream.go
@@ -0,0 +1,170 @@
+package docs
+
+import (
+ "io"
+ "net/http"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/rastrillo"
+)
+
+// The stream's timings. Lifted from rastrillo's own SSE handler
+// (jobs/events.go), and each one is carried for its own reason rather
+// than by imitation — see the notes on each.
+const (
+ // defaultStreamWriteDeadline is the per-write idle deadline,
+ // re-armed before every event and every heartbeat through
+ // http.NewResponseController. This is what lets a stream coexist
+ // with rastrillo serve's zero WriteTimeout default: a total
+ // response deadline would cut a healthy long-lived stream off
+ // alongside the stalled ones, so the framework leaves WriteTimeout
+ // off and expects a streaming handler to bound its own writes.
+ defaultStreamWriteDeadline = 30 * time.Second
+
+ // defaultStreamHeartbeat is the comment-ping cadence on a quiet
+ // stream. It keeps intermediaries from timing out a connection
+ // that looks idle, and — more importantly here — it is what makes
+ // the write deadline above ever fire: without a periodic write, a
+ // client that has gone away without closing is discovered at the
+ // next real edit, which may be never.
+ defaultStreamHeartbeat = 15 * time.Second
+
+ // defaultStreamTTL bounds any one connection. The client reopens
+ // (see fixtures/live-updates.mjs), so this costs a browser nothing
+ // and is half of the hibernation story the design spec's §6
+ // describes: platform hibernation is default-on, and a document
+ // left open on Friday must not pin its instance awake all weekend.
+ // No timer here exists outside a live connection, so nothing in
+ // this file is a ticker that dies at hibernation.
+ defaultStreamTTL = 5 * time.Minute
+)
+
+// stream serves GET /d/{sqid}/stream: a Server-Sent Events feed that
+// says "this document has content through seq N" and nothing else.
+// The bytes themselves are fetched by the client through
+// GET /d/{sqid}/updates?since=N — the same binary endpoint it already
+// used at mount, through the same replay path.
+//
+// Keeping document bytes off this stream is deliberate, and mirrors
+// the reasoning in rastrillo's own SSE handler for keeping HTML off
+// its one ("SSE never carries HTML, so the one rendering path
+// stays"). Two consequences here: there is exactly one code path that
+// turns stored updates into a browser's Y.Doc, rather than two that
+// can drift; and SSE's text-only framing costs nothing, where putting
+// opaque binary Yjs updates on it would have meant base64 and about a
+// third more bytes on every keystroke batch.
+//
+// Owner-scoped through getDocumentForUpdates, so a stranger gets the
+// same 404 the update routes give — never a 403, which would confirm
+// the document exists.
+func (h *handlers) stream(w http.ResponseWriter, r *http.Request) {
+ doc, ok := h.getDocumentForUpdates(w, r)
+ if !ok {
+ return
+ }
+
+ fl, ok := w.(http.Flusher)
+ if !ok {
+ h.logger.Error("docs: stream: ResponseWriter does not flush", "sqid", doc.Sqid)
+ http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusInternalServerError)
+ return
+ }
+
+ // Subscribe BEFORE reading the current seq, never after. An update
+ // committing between the two must produce a redundant
+ // notification, which costs one empty fetch; the other order would
+ // let it fall into the gap and produce nothing at all, and the tab
+ // would sit there current-looking and stale until the next edit.
+ ch, stop := h.hub.subscribe(doc.ID)
+ defer stop()
+
+ current, err := h.store.currentSeq(r.Context(), doc.ID)
+ if err != nil {
+ h.logger.Error("docs: stream: read current seq", "sqid", doc.Sqid, "err", err)
+ http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusInternalServerError)
+ return
+ }
+
+ rc := http.NewResponseController(w)
+ w.Header().Set("Content-Type", "text/event-stream")
+ w.Header().Set("Cache-Control", "no-store")
+ w.WriteHeader(http.StatusOK)
+ fl.Flush()
+
+ // write is every byte's one exit: deadline armed, then the bytes,
+ // then the flush. The deadline error is ignored deliberately — a
+ // ResponseWriter that cannot set deadlines still streams, just
+ // unbounded — while a write error ends the handler, because it
+ // means the client is gone.
+ write := func(text string) bool {
+ _ = rc.SetWriteDeadline(time.Now().Add(defaultStreamWriteDeadline))
+ if _, err := io.WriteString(w, text); err != nil {
+ return false
+ }
+ fl.Flush()
+ return true
+ }
+ event := func(seq int64) bool {
+ return write("event: update\ndata: " + strconv.FormatInt(seq, 10) + "\n\n")
+ }
+
+ // The connect-time event is load-bearing, not an optimisation. It
+ // is what heals a tab whose cursor legitimately failed to advance
+ // (a POST whose seq was not cursor+1, design D3), and it covers
+ // the gap between the page rendering server-side and this stream
+ // opening. Sending the current seq unconditionally costs a client
+ // that is already current exactly one empty ?since= fetch.
+ if !event(current) {
+ return
+ }
+
+ heartbeat := time.NewTimer(h.heartbeatInterval())
+ defer heartbeat.Stop()
+ ttl := time.NewTimer(h.ttlInterval())
+ defer ttl.Stop()
+
+ for {
+ select {
+ case <-r.Context().Done():
+ // The client went away, or the server is draining. Either
+ // way this handler must return so a graceful shutdown's
+ // budget is not held by an open stream.
+ return
+ case <-ttl.C:
+ // A clean close, not an error: the client reopens if it
+ // still wants one, and a parked tab does not.
+ return
+ case seq := <-ch:
+ if !event(seq) {
+ return
+ }
+ heartbeat.Reset(h.heartbeatInterval())
+ case <-heartbeat.C:
+ if !write(": ping\n\n") {
+ return
+ }
+ heartbeat.Reset(h.heartbeatInterval())
+ }
+ }
+}
+
+// heartbeatInterval and ttlInterval read the handler's own overrides,
+// falling back to the constants above. The overrides exist for
+// stream_test.go, whose whole subject is behaviour over time: the
+// production values are right for a browser and useless for a test
+// suite, and the alternative — a test that waits five minutes to
+// prove a five-minute TTL — is not a test anyone runs.
+func (h *handlers) heartbeatInterval() time.Duration {
+ if h.streamHeartbeat > 0 {
+ return h.streamHeartbeat
+ }
+ return defaultStreamHeartbeat
+}
+
+func (h *handlers) ttlInterval() time.Duration {
+ if h.streamTTL > 0 {
+ return h.streamTTL
+ }
+ return defaultStreamTTL
+}
diff --git a/internal/docs/stream_test.go b/internal/docs/stream_test.go
new file mode 100644
index 0000000..81dd058
--- /dev/null
+++ b/internal/docs/stream_test.go
@@ -0,0 +1,309 @@
+package docs
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+)
+
+// These tests run against a real httptest.NewServer rather than the
+// httptest.NewRecorder every other test in this package uses, and the
+// reason is structural rather than stylistic: a recorder collects
+// what a handler wrote and hands it over once the handler RETURNS. A
+// stream handler does not return until the connection is over, so a
+// recorder can express "what did this stream eventually say" and can
+// never express "what does this stream say while it is open", which
+// is the only interesting question about it. A real server and an
+// incrementally-read body can.
+
+// streamDurations shortens the handler's own timings for a test. The
+// production values (five minutes, fifteen seconds) are correct for a
+// browser and useless for a test suite.
+func streamDurations(t *testing.T, app *App, heartbeat, ttl time.Duration) {
+ t.Helper()
+ h := app.testHandlers()
+ h.streamHeartbeat = heartbeat
+ h.streamTTL = ttl
+}
+
+// openStream starts a GET on path as address and returns the response
+// plus a reader positioned at its body, having already checked the
+// status. The caller closes the body.
+func openStream(t *testing.T, srv *httptest.Server, app *App, address, path string) (*http.Response, *bufio.Reader) {
+ t.Helper()
+
+ req := app.TestRequest(t, address, http.MethodGet, path, nil)
+ live, err := http.NewRequest(http.MethodGet, srv.URL+path, nil)
+ if err != nil {
+ t.Fatalf("build stream request: %v", err)
+ }
+ for _, c := range req.Cookies() {
+ live.AddCookie(c)
+ }
+
+ res, err := srv.Client().Do(live)
+ if err != nil {
+ t.Fatalf("open stream %s: %v", path, err)
+ }
+ return res, bufio.NewReader(res.Body)
+}
+
+// nextEvent reads SSE lines until it has one complete event, and
+// returns its name and data. A comment heartbeat (": ping") is
+// returned as the name "ping" so a test can assert on it.
+func nextEvent(t *testing.T, r *bufio.Reader, within time.Duration) (name, data string) {
+ t.Helper()
+
+ type parsed struct{ name, data string }
+ got := make(chan parsed, 1)
+ errs := make(chan error, 1)
+
+ go func() {
+ var ev, payload string
+ for {
+ line, err := r.ReadString('\n')
+ if err != nil {
+ errs <- err
+ return
+ }
+ line = strings.TrimRight(line, "\r\n")
+ switch {
+ case strings.HasPrefix(line, ": "):
+ got <- parsed{"ping", strings.TrimPrefix(line, ": ")}
+ return
+ case strings.HasPrefix(line, "event: "):
+ ev = strings.TrimPrefix(line, "event: ")
+ case strings.HasPrefix(line, "data: "):
+ payload = strings.TrimPrefix(line, "data: ")
+ case line == "":
+ if ev != "" {
+ got <- parsed{ev, payload}
+ return
+ }
+ }
+ }
+ }()
+
+ select {
+ case p := <-got:
+ return p.name, p.data
+ case err := <-errs:
+ t.Fatalf("reading the stream: %v", err)
+ case <-time.After(within):
+ t.Fatalf("no event within %s", within)
+ }
+ return "", ""
+}
+
+// The stream is owner-scoped exactly as the update routes are, and a
+// stranger must get 404 rather than 403 — a 403 would confirm the
+// document exists.
+func TestStreamIsOwnerScoped(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")
+ app.TestPerson(t, "stranger@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Private")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ rec := serve(app, app.TestRequest(t, "stranger@example.com", http.MethodGet,
+ "/d/"+doc.Sqid+"/stream", nil))
+ if rec.Code != http.StatusNotFound {
+ t.Errorf("stranger GET /stream = %d, want 404", rec.Code)
+ }
+}
+
+// The connect-time event is load-bearing rather than an optimisation:
+// it is what heals a tab whose cursor legitimately failed to advance
+// (design D3), and what covers the gap between the page rendering and
+// the stream opening. Without it, a tab that missed a notification
+// during that gap waits for the next edit before it learns anything.
+func TestStreamAnnouncesTheCurrentSeqOnConnect(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, "Connect")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+ if _, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t)); err != nil {
+ t.Fatalf("ApplyUpdate: %v", err)
+ }
+
+ srv := httptest.NewServer(app.Handler())
+ defer srv.Close()
+
+ res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream")
+ defer res.Body.Close()
+
+ if res.StatusCode != http.StatusOK {
+ t.Fatalf("stream = %d, want 200", res.StatusCode)
+ }
+ if ct := res.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+ t.Errorf("Content-Type = %q, want text/event-stream", ct)
+ }
+ if cc := res.Header.Get("Cache-Control"); cc != "no-store" {
+ t.Errorf("Cache-Control = %q, want no-store", cc)
+ }
+
+ name, data := nextEvent(t, body, 2*time.Second)
+ if name != "update" || data != "1" {
+ t.Errorf("connect event = (%q, %q), want (update, 1)", name, data)
+ }
+}
+
+// The feature itself: an update accepted from one request reaches a
+// stream opened by another.
+func TestStreamDeliversAnUpdateFromAnotherRequest(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, "Two tabs")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ srv := httptest.NewServer(app.Handler())
+ defer srv.Close()
+
+ res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream")
+ defer res.Body.Close()
+
+ // Drain the connect-time event (seq 0 — nothing written yet).
+ if name, data := nextEvent(t, body, 2*time.Second); name != "update" || data != "0" {
+ t.Fatalf("connect event = (%q, %q), want (update, 0)", name, data)
+ }
+
+ post := app.TestRequest(t, "owner@example.com", http.MethodPost,
+ "/d/"+doc.Sqid+"/updates", bytes.NewReader(validUpdate(t)))
+ if rec := serve(app, post); rec.Code != http.StatusNoContent {
+ t.Fatalf("POST = %d, want 204", rec.Code)
+ }
+
+ name, data := nextEvent(t, body, 2*time.Second)
+ if name != "update" || data != "1" {
+ t.Errorf("event after a POST = (%q, %q), want (update, 1)", name, data)
+ }
+}
+
+// A quiet stream still writes, which is what makes the write deadline
+// able to fire and a dead client discoverable. Without the heartbeat
+// a wedged connection is only noticed at the next real edit, which
+// may be never.
+func TestStreamHeartbeatsWhenQuiet(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, "Quiet")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+ streamDurations(t, app, 50*time.Millisecond, 5*time.Second)
+
+ srv := httptest.NewServer(app.Handler())
+ defer srv.Close()
+
+ res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream")
+ defer res.Body.Close()
+
+ nextEvent(t, body, 2*time.Second) // the connect event
+
+ if name, _ := nextEvent(t, body, 2*time.Second); name != "ping" {
+ t.Errorf("quiet stream produced %q, want a ping heartbeat", name)
+ }
+}
+
+// No connection outlives the TTL. This is half of the hibernation
+// story (design §6): the server's hard bound on how long one stream
+// may hold a goroutine and keep an instance from sleeping.
+func TestStreamEndsAtItsTTL(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, "Bounded")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+ streamDurations(t, app, time.Second, 150*time.Millisecond)
+
+ srv := httptest.NewServer(app.Handler())
+ defer srv.Close()
+
+ res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream")
+ defer res.Body.Close()
+
+ nextEvent(t, body, 2*time.Second) // the connect event
+
+ done := make(chan error, 1)
+ go func() {
+ _, err := body.ReadString('\n')
+ done <- err
+ }()
+
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Error("stream produced more output after its TTL instead of ending")
+ }
+ case <-time.After(2 * time.Second):
+ t.Error("stream was still open well past its TTL")
+ }
+}
+
+// A client that goes away must not leave the handler running. This is
+// what keeps a graceful drain's budget safe with streams open.
+func TestStreamEndsWhenTheClientGoesAway(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, "Abandoned")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ srv := httptest.NewServer(app.Handler())
+ defer srv.Close()
+
+ res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream")
+ nextEvent(t, body, 2*time.Second)
+
+ // Closing the body is the client going away.
+ res.Body.Close()
+
+ // The handler's subscriber must be gone. Polling the hub is the
+ // observable consequence; the handler itself has no return value
+ // to wait on.
+ deadline := time.Now().Add(2 * time.Second)
+ for time.Now().Before(deadline) {
+ if app.testHandlers().hub.documentCount() == 0 {
+ return
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ t.Error("a subscriber outlived the client that opened it")
+}
diff --git a/internal/docs/testing.go b/internal/docs/testing.go
index 5d81d5c..bb6d1fa 100644
--- a/internal/docs/testing.go
+++ b/internal/docs/testing.go
@@ -166,6 +166,17 @@ func (a *App) TestRequest(t *testing.T, address, method, path string, body io.Re
return req
}
+// testHandlers exposes the app's handler struct so a test can reach
+// the two seams that only exist for testing: the stream's heartbeat
+// and TTL overrides (stream_test.go, whose subject is behaviour over
+// time), and the hub, whose subscriber count is the only observable
+// consequence of a stream handler having returned.
+//
+// Unexported, and in this file rather than a _test.go one for the
+// same reason NewForTest is: it is scaffolding this package's own
+// tests share, not an interface the app has.
+func (a *App) testHandlers() *handlers { return a.handlers }
+
// ReadFixture loads fixtures/data/<name>.update.bin — a real Yjs
// update produced by browser yjs (fixtures/generate.mjs), never
// hand-edited. Not in a _test.go file for the same reason as