amadan

oficina / sheets Public

live: an in-process fan-out hub that never blocks on a wedged tab

The hub moves opaque payloads from one publisher to every subscriber on
a tab. It knows nothing about cells, sheets, HTTP or storage, and that
boundary is load-bearing rather than tidy: AGENTS.md forbids cell
content in any log line, and a fan-out package that could see inside
its payloads is one whose first debugging session breaks that rule.

Every send is non-blocking. The publisher is a request handler that has
already committed a database write, so waiting on a browser that
stopped reading would hold a writer connection open for as long as that
browser felt like it. An overrun subscriber is marked lost and told to
resync — a grid with holes in it is worse than a reload.

live joins store on the -race target, for the same reason store is
there: it is where the goroutines are.
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 3b3cfe2df233362105e5c4e80e5aecee8b0fec69 parent 031665f
3 files changed, +318 −7
  • Makefile +8 −7
  • internal/sheets/live/hub.go +145 −0
  • internal/sheets/live/hub_test.go +165 −0

CI passed — run details

diff --git a/Makefile b/Makefile
index 0e99e22..a142a80 100644
--- a/Makefile
+++ b/Makefile
@@ -74,12 +74,13 @@ test-migrations:
test-go:
CGO_ENABLED=0 go test -count=1 ./...
-# The concurrency-bearing package: the one-connection writer pool and
-# its handle gate (AGENTS.md: "Scope tx, never the outer handle, inside
-# transactions, or the one-connection writer pool deadlocks"). Scoped to
-# store rather than ./... because -race's slowdown is worth paying where
-# the concurrency actually lives, not repeated across packages that
-# don't touch goroutines.
+# The concurrency-bearing packages: store's one-connection writer pool
+# and its handle gate (AGENTS.md: "Scope tx, never the outer handle,
+# inside transactions, or the one-connection writer pool deadlocks"),
+# and live's fan-out hub, where one publisher and every open browser
+# tab meet. Scoped to those two rather than ./... because -race's
+# slowdown is worth paying where the concurrency actually lives, not
+# repeated across packages that don't touch goroutines.
#
# Deliberately NOT CGO_ENABLED=0: the race detector requires cgo (`go:
# -race requires cgo`), so this one target runs with the ambient
@@ -88,4 +89,4 @@ test-go:
# so it does not violate the cgo-free binary rule, which is about what
# gets deployed.
test-race:
- go test -race -count=1 ./internal/sheets/store/...
+ go test -race -count=1 ./internal/sheets/store/... ./internal/sheets/live/...
diff --git a/internal/sheets/live/hub.go b/internal/sheets/live/hub.go
new file mode 100644
index 0000000..bdc86d2
--- /dev/null
+++ b/internal/sheets/live/hub.go
@@ -0,0 +1,145 @@
+// Package live is the fan-out between one person's edit and everybody
+// else's open tab.
+//
+// It knows nothing about cells, sheets, HTTP or storage: a Message
+// carries an opaque, already-marshalled payload and a version, and the
+// hub moves it from one publisher to every subscriber on a tab. That
+// boundary is deliberate and load-bearing. AGENTS.md forbids cell
+// content in any log line, and a fan-out package that could see inside
+// its payloads is a package whose first debugging session breaks that
+// rule. This one cannot.
+package live
+
+import (
+ "sync"
+ "sync/atomic"
+)
+
+// subBuffer is how many messages a subscriber may fall behind by before
+// it is declared lost.
+//
+// Deep enough that an ordinary tab under a burst — a paste that
+// recalculates a column, several people typing at once — never overruns,
+// and shallow enough that a thousand wedged tabs cost kilobytes rather
+// than megabytes. The alternative, an unbounded buffer, turns one
+// browser that stopped reading into an instance-wide memory leak with no
+// upper bound and no symptom until the process dies.
+const subBuffer = 64
+
+// Message is one fan-out payload.
+//
+// Data is already-marshalled bytes. The hub never looks inside it, and
+// nothing here ever logs it.
+type Message struct {
+ Version uint64
+ Data []byte
+}
+
+// Sub is one open connection's subscription. Get one from
+// Hub.Subscribe; always Close it.
+type Sub struct {
+ hub *Hub
+ tabID uint64
+ ch chan Message
+ lost atomic.Bool
+ once sync.Once
+}
+
+// C is the channel to select on. It is never closed — Close removes the
+// subscription from the hub instead — so a reader distinguishes "no
+// messages" from "gone" through its own context, not through a channel
+// close it would have to race the publisher for.
+func (s *Sub) C() <-chan Message { return s.ch }
+
+// Lost reports whether this subscriber overran subBuffer and has missed
+// at least one message. A lost subscriber must not keep painting: its
+// grid would have holes in it that nothing later fills. The SSE handler
+// tells the client to resync instead.
+func (s *Sub) Lost() bool { return s.lost.Load() }
+
+// Close unsubscribes. It is idempotent: an SSE handler defers it and may
+// also reach it on a write error, and doing the removal twice must not
+// panic or disturb a subscription some later Subscribe handed out.
+func (s *Sub) Close() {
+ s.once.Do(func() { s.hub.remove(s) })
+}
+
+// Hub fans messages out to the subscribers of a tab. The zero value is
+// not usable; call NewHub. One per process.
+type Hub struct {
+ mu sync.Mutex
+ tabs map[uint64]map[*Sub]struct{}
+}
+
+// NewHub builds an empty hub.
+func NewHub() *Hub {
+ return &Hub{tabs: map[uint64]map[*Sub]struct{}{}}
+}
+
+// Subscribe registers a new subscriber for a tab.
+func (h *Hub) Subscribe(tabID uint64) *Sub {
+ s := &Sub{hub: h, tabID: tabID, ch: make(chan Message, subBuffer)}
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ subs, ok := h.tabs[tabID]
+ if !ok {
+ subs = map[*Sub]struct{}{}
+ h.tabs[tabID] = subs
+ }
+ subs[s] = struct{}{}
+ return s
+}
+
+// Publish delivers a message to every current subscriber of a tab.
+//
+// The send is non-blocking, always. A subscriber whose buffer is full is
+// marked lost and skipped — never waited for — because the caller is a
+// request handler that has already committed a database write, and
+// blocking it on a browser that stopped reading would hold a writer
+// connection open for as long as that browser felt like it.
+//
+// Publish is safe to call with no subscribers, which is the common case:
+// one person editing alone.
+func (h *Hub) Publish(tabID uint64, m Message) {
+ h.mu.Lock()
+ subs := make([]*Sub, 0, len(h.tabs[tabID]))
+ for s := range h.tabs[tabID] {
+ subs = append(subs, s)
+ }
+ h.mu.Unlock()
+
+ // Sent outside the lock: a slow select here would otherwise block
+ // every other tab's Subscribe and Close as well.
+ for _, s := range subs {
+ select {
+ case s.ch <- m:
+ default:
+ s.lost.Store(true)
+ }
+ }
+}
+
+// Subscribers reports how many subscriptions a tab currently has.
+// Exposed for tests and for a presence count later; it deliberately
+// counts connections rather than people.
+func (h *Hub) Subscribers(tabID uint64) int {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return len(h.tabs[tabID])
+}
+
+// remove drops a subscription, and the tab's map entry with it once the
+// last subscriber leaves — so an instance that has served ten thousand
+// tabs over a week does not keep ten thousand empty maps.
+func (h *Hub) remove(s *Sub) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ subs, ok := h.tabs[s.tabID]
+ if !ok {
+ return
+ }
+ delete(subs, s)
+ if len(subs) == 0 {
+ delete(h.tabs, s.tabID)
+ }
+}
diff --git a/internal/sheets/live/hub_test.go b/internal/sheets/live/hub_test.go
new file mode 100644
index 0000000..6665d53
--- /dev/null
+++ b/internal/sheets/live/hub_test.go
@@ -0,0 +1,165 @@
+package live
+
+import (
+ "sync"
+ "testing"
+ "time"
+)
+
+// recv waits for one message, or fails. Every test wants this and none
+// of them vary the timeout.
+func recv(t *testing.T, s *Sub, what string) Message {
+ t.Helper()
+ select {
+ case m := <-s.C():
+ return m
+ case <-time.After(2 * time.Second):
+ t.Fatalf("%s: received nothing", what)
+ return Message{}
+ }
+}
+
+func TestPublishReachesEverySubscriberOnTheTab(t *testing.T) {
+ h := NewHub()
+ a := h.Subscribe(7)
+ b := h.Subscribe(7)
+ defer a.Close()
+ defer b.Close()
+
+ h.Publish(7, Message{Version: 3, Data: []byte(`{"v":3}`)})
+
+ for i, s := range []*Sub{a, b} {
+ if m := recv(t, s, "subscriber"); m.Version != 3 {
+ t.Errorf("subscriber %d got version %d, want 3", i, m.Version)
+ }
+ }
+}
+
+// The tab id is the whole routing key. If it leaks, one team's edits
+// paint into another sheet's grid.
+func TestPublishDoesNotReachAnotherTab(t *testing.T) {
+ h := NewHub()
+ here := h.Subscribe(7)
+ elsewhere := h.Subscribe(8)
+ defer here.Close()
+ defer elsewhere.Close()
+
+ h.Publish(7, Message{Version: 3})
+
+ recv(t, here, "the subscribed tab")
+ select {
+ case m := <-elsewhere.C():
+ t.Fatalf("a subscriber on tab 8 received tab 7's message: %+v", m)
+ case <-time.After(50 * time.Millisecond):
+ }
+}
+
+func TestPublishWithNoSubscribersIsFine(t *testing.T) {
+ // One person editing alone is the common case, not an edge case.
+ NewHub().Publish(7, Message{Version: 1})
+}
+
+func TestCloseUnsubscribes(t *testing.T) {
+ h := NewHub()
+ s := h.Subscribe(7)
+ if got := h.Subscribers(7); got != 1 {
+ t.Fatalf("Subscribers(7) = %d after one Subscribe, want 1", got)
+ }
+ s.Close()
+ if got := h.Subscribers(7); got != 0 {
+ t.Fatalf("after Close, Subscribers(7) = %d, want 0", got)
+ }
+ // An SSE handler defers Close and may also reach it on a write
+ // error. A double close must not panic in a goroutine no request
+ // owns.
+ s.Close()
+}
+
+func TestClosingOneSubscriberLeavesTheOthers(t *testing.T) {
+ h := NewHub()
+ a := h.Subscribe(7)
+ b := h.Subscribe(7)
+ defer b.Close()
+ a.Close()
+
+ h.Publish(7, Message{Version: 9})
+ if m := recv(t, b, "the surviving subscriber"); m.Version != 9 {
+ t.Fatalf("surviving subscriber got version %d, want 9", m.Version)
+ }
+}
+
+// A tab that has stopped reading must never block the writer: the
+// publisher is a request handler that has already committed a database
+// write.
+func TestASlowSubscriberIsDroppedNotBlocking(t *testing.T) {
+ h := NewHub()
+ s := h.Subscribe(7)
+ defer s.Close()
+
+ for i := 0; i < subBuffer+10; i++ {
+ done := make(chan struct{})
+ go func(i int) {
+ h.Publish(7, Message{Version: uint64(i)})
+ close(done)
+ }(i)
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatalf("Publish blocked on message %d — one wedged browser would hold a writer open", i)
+ }
+ }
+ if !s.Lost() {
+ t.Fatal("an overrun subscriber is not marked lost; it would paint a grid with holes in it")
+ }
+}
+
+func TestAKeepingUpSubscriberIsNotLost(t *testing.T) {
+ h := NewHub()
+ s := h.Subscribe(7)
+ defer s.Close()
+ for i := 0; i < subBuffer*3; i++ {
+ h.Publish(7, Message{Version: uint64(i)})
+ recv(t, s, "keeping up")
+ }
+ if s.Lost() {
+ t.Fatal("a subscriber that read every message was declared lost")
+ }
+}
+
+// Subscribe, Publish and Close all race in production: a tab opening
+// while another closes while an edit lands.
+func TestConcurrentSubscribePublishAndClose(t *testing.T) {
+ h := NewHub()
+ var wg sync.WaitGroup
+ for i := 0; i < 20; i++ {
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ s := h.Subscribe(7)
+ s.Close()
+ }()
+ go func() {
+ defer wg.Done()
+ h.Publish(7, Message{Version: 1, Data: []byte("x")})
+ }()
+ }
+ wg.Wait()
+ if got := h.Subscribers(7); got != 0 {
+ t.Fatalf("Subscribers(7) = %d after every subscriber closed, want 0", got)
+ }
+}
+
+// An instance that has served many tabs over a week must not keep one
+// empty map per tab it has ever seen.
+func TestAnEmptiedTabIsForgotten(t *testing.T) {
+ h := NewHub()
+ for id := uint64(1); id <= 50; id++ {
+ h.Subscribe(id).Close()
+ }
+ h.mu.Lock()
+ n := len(h.tabs)
+ h.mu.Unlock()
+ if n != 0 {
+ t.Fatalf("the hub still holds %d tab entries after every subscriber left", n)
+ }
+}