amadan

oficina / sheets Public

feat: SSE fan-out, with an idle policy written down rather than emergent

A streamed 200 carrying text/event-stream, not a 101 — SSE is an
ordinary response with a streaming body, and a transport check written
against the WebSocket upgrade would never pass. The source document
this design came from conflated the two; the test asserts the right one
and says why.

The test needs a real server. A ResponseRecorder buffers everything and
returns when the handler does, so a handler that built its whole body
and returned would satisfy every assertion a recorder can make while
being exactly the bug this endpoint must not have.

Spec 7 words the idle policy as "no edits and no presence heartbeat".
There is no heartbeat, and the reason is in the code rather than left
as a gap: a heartbeat POST arrives as a separate request with no way to
reach the goroutine holding the stream, so it would have to be routed
through the hub by connection id — machinery to reproduce what the
connection already tells us. The client closes its stream when the tab
is hidden instead, which is stronger: a hidden tab holds no socket at
all rather than an idle one. The server cannot enforce that —
EventSource reconnects after any close — so the timeouts here are a
backstop for clients that vanished without saying so, and the comment
says exactly that.

X-Accel-Buffering: no is asserted, because an edge that buffers a
stream into a complete response turns live sync into "everything
arrives when you close the tab", and no in-process test can see it.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018RN4HhHR4rHH3Fxh4LMds5
Paul Campbell pushed by paul@keymail.dev 4b94b94827229852a725979fe4b37781333b4f47 parent 8cf9363
3 files changed, +357 −0
  • internal/sheets/app.go +1 −0
  • internal/sheets/handlers_stream.go +155 −0
  • internal/sheets/handlers_stream_test.go +201 −0
diff --git a/internal/sheets/app.go b/internal/sheets/app.go
index 95f9b2a..61c1d8b 100644
--- a/internal/sheets/app.go
+++ b/internal/sheets/app.go
@@ -137,6 +137,7 @@ func (a *App) Router() http.Handler {
// ask for it in.
r.Get("/api/sheets/{sheetID}/tabs/{tabID}/cells", a.handleCellsGet)
r.Post("/api/sheets/{sheetID}/tabs/{tabID}/cells", a.handleCellPost)
+ r.Get("/api/sheets/{sheetID}/tabs/{tabID}/stream", a.handleStream)
})
return r
}
diff --git a/internal/sheets/handlers_stream.go b/internal/sheets/handlers_stream.go
new file mode 100644
index 0000000..9c0ee0c
--- /dev/null
+++ b/internal/sheets/handlers_stream.go
@@ -0,0 +1,155 @@
+package sheets
+
+import (
+ "net/http"
+ "time"
+)
+
+// The idle policy, written down rather than emergent (spec §7). Held
+// sockets and hibernation pull against each other, and this is where
+// that is resolved.
+//
+// pingEvery keeps the connection alive through proxies that drop a
+// silent stream, and gives the client something to notice when the
+// server has gone away.
+//
+// idleTimeout closes a stream that has carried no edit. maxStreamLife
+// closes one regardless of activity, so a tab left open all weekend
+// cannot hold an instance awake and billable indefinitely.
+//
+// Spec §7 words the idle policy as "no edits and no presence
+// heartbeat". There is no heartbeat here, and the reason is worth
+// stating rather than leaving as a gap: a heartbeat POST arrives as a
+// separate request with no way to reach the specific goroutine holding
+// the stream, so it would have had to be routed through the hub by
+// connection id — real machinery to reproduce something the connection
+// itself already tells us. Instead the client CLOSES its stream when
+// the tab is hidden and opens a new one when it comes back, which is
+// strictly stronger than a heartbeat that stops: a hidden tab holds no
+// socket at all rather than an idle one.
+//
+// That only works with the client's cooperation, and it is worth being
+// plain that the server cannot enforce it: EventSource reconnects on
+// its own after any close, so a client that ignores the "bye" event
+// below will simply come straight back. The server's timeouts are
+// therefore a backstop against clients that vanished without saying so
+// — a sleeping laptop, a dropped network — and not the whole policy.
+const (
+ pingEvery = 25 * time.Second
+ idleTimeout = 15 * time.Minute
+ maxStreamLife = 4 * time.Hour
+)
+
+// handleStream is the live connection: a streamed 200 carrying
+// text/event-stream, one message per committed write on this tab.
+//
+// A streamed 200, NOT a 101 (spec §7). SSE is an ordinary response with
+// a streaming body; 101 is the WebSocket upgrade, and a transport check
+// written against 101 would never pass here. The conflation is worth
+// naming because the source document this design came from made it.
+func (a *App) handleStream(w http.ResponseWriter, r *http.Request) {
+ sheetID, tabID := sheetIDFrom(r), tabIDFrom(r)
+ if sheetID == 0 || tabID == 0 || !a.tabBelongsTo(sheetID, tabID) {
+ a.notFound(w, r)
+ return
+ }
+
+ rc := http.NewResponseController(w)
+
+ w.Header().Set("Content-Type", "text/event-stream")
+ w.Header().Set("Cache-Control", "no-store")
+ w.Header().Set("Connection", "keep-alive")
+ // The edge will happily buffer a streaming response into a complete
+ // one, which turns live sync into "everything arrives when you
+ // close the tab". This header is what stops it, and its absence is
+ // invisible in every test that does not run through a real proxy.
+ w.Header().Set("X-Accel-Buffering", "no")
+ w.WriteHeader(http.StatusOK)
+
+ // Flushed before anything is subscribed, so the client's onopen
+ // fires immediately rather than on the first edit. A grid that says
+ // "connecting" until somebody types looks broken for as long as
+ // nobody does.
+ if err := rc.Flush(); err != nil {
+ return
+ }
+
+ sub := a.hub.Subscribe(tabID)
+ defer sub.Close()
+
+ ping := time.NewTicker(pingEvery)
+ defer ping.Stop()
+ idle := time.NewTimer(idleTimeout)
+ defer idle.Stop()
+ life := time.NewTimer(maxStreamLife)
+ defer life.Stop()
+
+ for {
+ select {
+ case <-r.Context().Done():
+ return
+
+ case <-life.C:
+ // The ceiling. Say so rather than dropping silently: the
+ // client reconnects on a close either way, but a named
+ // reason is the difference between a debuggable log and a
+ // mystery.
+ writeEvent(w, rc, "bye", `{"reason":"max-life"}`)
+ return
+
+ case <-idle.C:
+ writeEvent(w, rc, "bye", `{"reason":"idle"}`)
+ return
+
+ case <-ping.C:
+ // A comment line: valid SSE, ignored by EventSource, and
+ // enough to keep an intermediary from calling the
+ // connection dead.
+ if _, err := w.Write([]byte(": ping\n\n")); err != nil {
+ return
+ }
+ if err := rc.Flush(); err != nil {
+ return
+ }
+
+ case m := <-sub.C():
+ if sub.Lost() {
+ // This subscriber overran its buffer and has missed at
+ // least one message, so everything after it would paint
+ // a grid with holes in it. Tell the client to refetch
+ // and stop: a reload is cheap, and a silently wrong
+ // spreadsheet is not.
+ writeEvent(w, rc, "resync", `{"reason":"overrun"}`)
+ return
+ }
+ if !writeEvent(w, rc, "cells", string(m.Data)) {
+ return
+ }
+ // A delivered edit resets the idle clock: this tab is
+ // demonstrably part of a live session.
+ resetTimer(idle, idleTimeout)
+ }
+ }
+}
+
+// writeEvent sends one SSE event and flushes it. It returns false once
+// the connection is gone, which is the only error worth distinguishing:
+// there is nobody left to tell.
+//
+// The data is written as a single line. Every payload this app sends is
+// compact JSON, which contains no newline — and a newline inside an SSE
+// data field would split one message into two, producing a truncated
+// JSON parse on the client rather than an error anybody could read.
+func writeEvent(w http.ResponseWriter, rc *http.ResponseController, event, data string) bool {
+ if _, err := w.Write([]byte("event: " + event + "\ndata: " + data + "\n\n")); err != nil {
+ return false
+ }
+ return rc.Flush() == nil
+}
+
+// resetTimer restarts a timer that may or may not have fired. Stop then
+// Reset is the shape that is correct for both.
+func resetTimer(t *time.Timer, d time.Duration) {
+ t.Stop()
+ t.Reset(d)
+}
diff --git a/internal/sheets/handlers_stream_test.go b/internal/sheets/handlers_stream_test.go
new file mode 100644
index 0000000..9f02343
--- /dev/null
+++ b/internal/sheets/handlers_stream_test.go
@@ -0,0 +1,201 @@
+package sheets
+
+import (
+ "bufio"
+ "context"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+)
+
+// liveServer starts a real HTTP server for the app.
+//
+// httptest.ResponseRecorder cannot test this: it buffers everything and
+// returns when the handler does, so a handler that built its whole body
+// and returned would pass every assertion a recorder can make — while
+// being exactly the bug this endpoint must not have.
+func liveServer(t *testing.T, app *App) *httptest.Server {
+ t.Helper()
+ srv := httptest.NewServer(app.Router())
+ t.Cleanup(srv.Close)
+ return srv
+}
+
+// openStream connects to a tab's stream and returns the response.
+func openStream(t *testing.T, srv *httptest.Server, s SheetSummary, cookies []*http.Cookie) *http.Response {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ t.Cleanup(cancel)
+
+ url := fmt.Sprintf("%s/api/sheets/%d/tabs/%d/stream", srv.URL, s.ID, s.TabID)
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+ if err != nil {
+ t.Fatalf("build stream request: %v", err)
+ }
+ for _, c := range cookies {
+ req.AddCookie(c)
+ }
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("open stream: %v", err)
+ }
+ t.Cleanup(func() { resp.Body.Close() })
+ return resp
+}
+
+// readEventWithin reads SSE lines until a data line arrives, or fails.
+func readEventWithin(t *testing.T, body io.Reader, d time.Duration) string {
+ t.Helper()
+ type result struct{ line string }
+ ch := make(chan result, 1)
+ go func() {
+ sc := bufio.NewScanner(body)
+ for sc.Scan() {
+ if strings.HasPrefix(sc.Text(), "data: ") {
+ ch <- result{sc.Text()}
+ return
+ }
+ }
+ ch <- result{""}
+ }()
+ select {
+ case r := <-ch:
+ if r.line == "" {
+ t.Fatal("the stream ended without sending a data line")
+ }
+ return r.line
+ case <-time.After(d):
+ t.Fatalf("no event within %s", d)
+ return ""
+ }
+}
+
+// Spec §7: the SSE transport check is a streamed 200, not a 101. SSE is
+// text/event-stream over a normal response; 101 is the WebSocket
+// upgrade, and a check written against 101 would never pass.
+func TestStreamIsAStreamed200NotAnUpgrade(t *testing.T) {
+ app, cookies := signIn(t)
+ s, _ := newSheet(t, app, cookies, "Budget")
+ srv := liveServer(t, app)
+
+ resp := openStream(t, srv, s, cookies)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("stream = %d, want a streamed 200", resp.StatusCode)
+ }
+ if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+ t.Fatalf("Content-Type = %q, want text/event-stream", ct)
+ }
+ // The header that stops an edge buffering a stream into a complete
+ // response — the failure that turns live sync into "everything
+ // arrives when you close the tab", and which no in-process test can
+ // otherwise see.
+ if got := resp.Header.Get("X-Accel-Buffering"); got != "no" {
+ t.Errorf("X-Accel-Buffering = %q, want %q", got, "no")
+ }
+}
+
+// The real assertion: an edit made after the stream was opened arrives
+// on it, without the response having to end. A handler that builds its
+// body and returns passes every status check above and is useless.
+func TestAnEditArrivesOnAnOpenStream(t *testing.T) {
+ app, cookies := signIn(t)
+ s, cells := newSheet(t, app, cookies, "Budget")
+ srv := liveServer(t, app)
+
+ resp := openStream(t, srv, s, cookies)
+
+ // Wait until the handler has actually subscribed. Posting before
+ // that races the subscription and loses the message — the test
+ // would be flaky rather than wrong, which is worse.
+ waitForSubscriber(t, app, s.TabID)
+
+ postCell(t, app, cookies, cells, `{"r":0,"c":0,"raw":"=6*7"}`)
+
+ line := readEventWithin(t, resp.Body, 5*time.Second)
+ if !strings.Contains(line, `"display":"42"`) {
+ t.Fatalf("first streamed event = %q, want the computed value", line)
+ }
+}
+
+// Two tabs on one sheet: the whole feature, at the layer below the
+// browser.
+func TestTwoStreamsBothReceiveAnEdit(t *testing.T) {
+ app, cookies := signIn(t)
+ s, cells := newSheet(t, app, cookies, "Budget")
+ srv := liveServer(t, app)
+
+ first := openStream(t, srv, s, cookies)
+ second := openStream(t, srv, s, cookies)
+ waitForSubscriberCount(t, app, s.TabID, 2)
+
+ postCell(t, app, cookies, cells, `{"r":0,"c":0,"raw":"hello"}`)
+
+ for i, resp := range []*http.Response{first, second} {
+ if line := readEventWithin(t, resp.Body, 5*time.Second); !strings.Contains(line, "hello") {
+ t.Errorf("stream %d received %q", i, line)
+ }
+ }
+}
+
+func TestStreamOfAnUnknownTabIs404(t *testing.T) {
+ app, cookies := signIn(t)
+ s, _ := newSheet(t, app, cookies, "Budget")
+
+ path := fmt.Sprintf("/api/sheets/%d/tabs/999999/stream", s.ID)
+ if rec := do(t, app, as(httptest.NewRequest(http.MethodGet, path, nil), cookies)); rec.Code != http.StatusNotFound {
+ t.Fatalf("streaming an unknown tab = %d, want 404", rec.Code)
+ }
+}
+
+func TestStreamRequiresASession(t *testing.T) {
+ app, cookies := signIn(t)
+ s, _ := newSheet(t, app, cookies, "Budget")
+
+ path := fmt.Sprintf("/api/sheets/%d/tabs/%d/stream", s.ID, s.TabID)
+ if rec := do(t, app, httptest.NewRequest(http.MethodGet, path, nil)); rec.Code == http.StatusOK {
+ t.Fatal("the stream served a request with no session")
+ }
+}
+
+// Closing the connection must unsubscribe, or an instance leaks one
+// subscription per browser tab that has ever been opened.
+func TestClosingAStreamUnsubscribes(t *testing.T) {
+ app, cookies := signIn(t)
+ s, _ := newSheet(t, app, cookies, "Budget")
+ srv := liveServer(t, app)
+
+ resp := openStream(t, srv, s, cookies)
+ waitForSubscriber(t, app, s.TabID)
+ resp.Body.Close()
+
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if app.hub.Subscribers(s.TabID) == 0 {
+ return
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ t.Fatalf("after the client closed, %d subscribers remain", app.hub.Subscribers(s.TabID))
+}
+
+func waitForSubscriber(t *testing.T, app *App, tabID uint64) {
+ t.Helper()
+ waitForSubscriberCount(t, app, tabID, 1)
+}
+
+func waitForSubscriberCount(t *testing.T, app *App, tabID uint64, want int) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if app.hub.Subscribers(tabID) >= want {
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ t.Fatalf("only %d subscribers on tab %d after 5s, want %d",
+ app.hub.Subscribers(tabID), tabID, want)
+}