From dd963993eb30fb1b5d55dd10c01ed96419538783 Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Mon, 2 Nov 2026 18:17:37 -0500 Subject: [PATCH] fix(sync): scan events keep instance-wide fan-out as the sanctioned BroadcastToAll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Verifying the per-user chokepoint end to end surfaced a UI regression it would have caused: web/src/header.ts runs the sitewide scan indicator on scan_progress/scan_complete events in every browser, and routing those to the triggering identity alone froze the indicator for everyone else at its page-load state (active-summary stays the load-time fallback, but live movement died). The source's own comment declares the indicator sitewide by design, and scan payloads carry no user data — job ids, file counts, progress percentages only. Resolution honors both truths: BroadcastToAll is re-added as the ONE sanctioned instance-wide fan-out, its doc comment making the decision explicit (no user-derived data; anything carrying books, notes, or annotations belongs in BroadcastToUser), and the two scan emitters use it. TestBroadcastToAllReachesEveryone pins the exception alongside TestBroadcastIsPerUser — audiences are now named decisions the compiler and the tests both enforce. --- docs/developer/websocket-api.md | 15 +++++++------ internal/services/worker.go | 15 ++++++++----- internal/sync/websocket.go | 22 +++++++++++++++++++ internal/sync/websocket_test.go | 38 ++++++++++++++++++++++++++++++++- 4 files changed, 77 insertions(+), 13 deletions(-) diff --git a/docs/developer/websocket-api.md b/docs/developer/websocket-api.md index e8856731..9fd51669 100644 --- a/docs/developer/websocket-api.md +++ b/docs/developer/websocket-api.md @@ -75,13 +75,14 @@ nothing to implement. A connection that stops answering is removed. ## Fan-out and backpressure — read this before building a client -- **Every event is delivered only to the acting user's connections.** - Progress, annotation, and conflict events go to the user whose sync - activity produced them; scan events go to the identity that triggered - the job; collection events to the collection's owner. The - ConnectionManager has no untargeted broadcast — the compiler enforces - an audience on every emitter (2026-11-02, after cross-user delivery - was found and fixed). +- **Every event carries an explicit audience.** Progress, annotation, + conflict, and collection events are delivered only to the acting + user's connections — the ConnectionManager has no accidental + fan-out, and the compiler forces every emitter to name its audience + (2026-11-02, after cross-user delivery was found and fixed). The one + sanctioned instance-wide fan-out is **scan events** + (`BroadcastToAll`): they carry job ids and counts, never user data, + and the sitewide scan indicator in every browser runs on them. - Every connection has a bounded send queue. **If a client falls behind, messages for that connection are skipped, not queued.** Clients must therefore treat the socket as a *hint* — on reconnect (or any gap), diff --git a/internal/services/worker.go b/internal/services/worker.go index 35e65a1c..fcbba363 100644 --- a/internal/services/worker.go +++ b/internal/services/worker.go @@ -402,15 +402,18 @@ func (w *Worker) processJob(job *Job) { Errors: errors, } - if job.Type == JobTypeScan && w.connManager != nil && job.UserID != "" { + if job.Type == JobTypeScan && w.connManager != nil { // Cancelled scans announce scan_cancelled so clients reset // their scanning indicators; a "scan_complete" would read as // a finished scan. + // Scan events are instance activity, not user data (job ids and + // counts only) — they fan out to every client so the sitewide + // scan indicator works for everyone. See BroadcastToAll. msgType := wsync.MessageTypeScanComplete if status == JobStatusCancelled { msgType = wsync.MessageTypeScanCancelled } - w.connManager.BroadcastToUser(job.UserID, wsync.BroadcastMessage{ + w.connManager.BroadcastToAll(wsync.BroadcastMessage{ Type: msgType, Data: map[string]interface{}{ "job_id": job.ID, @@ -464,8 +467,10 @@ func (w *Worker) processScanJob(job *Job) (interface{}, error) { result.NewItems = newItems result.Errors = errors } - // Broadcast via WebSocket to user - if w.connManager != nil && job.UserID != "" { + // Broadcast scan progress to every connected client — instance + // activity, no user data (see BroadcastToAll). The sitewide + // header indicator runs on these events. + if w.connManager != nil { msg := wsync.BroadcastMessage{ Type: wsync.MessageTypeScanProgress, Data: map[string]interface{}{ @@ -477,7 +482,7 @@ func (w *Worker) processScanJob(job *Job) (interface{}, error) { }, } - w.connManager.BroadcastToUser(job.UserID, msg) + w.connManager.BroadcastToAll(msg) } } diff --git a/internal/sync/websocket.go b/internal/sync/websocket.go index 06aa21be..9362366c 100644 --- a/internal/sync/websocket.go +++ b/internal/sync/websocket.go @@ -71,6 +71,28 @@ func NewConnectionManager() *ConnectionManager { } } +// BroadcastToAll delivers an event to every connected client on the +// instance. This is the ONE sanctioned instance-wide fan-out, and using +// it is a decision: the payload must contain no user-derived data. Its +// only caller is the scan-event emitter — scan activity is instance +// activity (job ids, file counts, progress percentages), and the +// sitewide scan indicator in web/src/header.ts is built on receiving it +// in every browser. Anything that carries a user's books, notes, or +// annotations belongs in BroadcastToUser. +func (m *ConnectionManager) BroadcastToAll(msg BroadcastMessage) { + m.mu.RLock() + defer m.mu.RUnlock() + + for _, conn := range m.connections { + select { + case conn.Send <- msg: + default: + // Channel full, skip this connection + log.Printf("WebSocket: Channel full for %s, skipping broadcast", conn.DeviceName) + } + } +} + // BroadcastToUser sends a message to all connections for a specific user func (m *ConnectionManager) BroadcastToUser(userID string, msg BroadcastMessage) { m.mu.RLock() diff --git a/internal/sync/websocket_test.go b/internal/sync/websocket_test.go index 7d851d97..26226594 100644 --- a/internal/sync/websocket_test.go +++ b/internal/sync/websocket_test.go @@ -68,8 +68,44 @@ func TestBroadcastIsPerUser(t *testing.T) { } } - // ...and Bob received nothing. Any message here is a cross-user leak. + // ...and Bob received nothing user-derived from Alice's sync + // activity. Scan events are the one sanctioned instance-wide fan-out + // (BroadcastToAll — job ids and counts, no user data); everything + // carrying a user's books or annotations stops at that user's edge. if leaked := drain(b1); len(leaked) != 0 { t.Fatalf("bob's client received %d message(s) from alice's sync activity — cross-user leak", len(leaked)) } } + +// TestBroadcastToAllReachesEveryone pins the sanctioned exception: scan +// activity is instance-wide (the sitewide header indicator runs on it), +// while everything user-derived stays per-user (TestBroadcastIsPerUser). +func TestBroadcastToAllReachesEveryone(t *testing.T) { + cm := NewConnectionManager() + + var conns []*DeviceConnection + for _, user := range []string{"alice", "bob", "carol"} { + conn := &DeviceConnection{ + ID: uuid.NewString(), + UserID: user, + DeviceType: "test", + DeviceName: user + "-device", + Send: make(chan BroadcastMessage, 4), + } + cm.AddConnection(conn) + conns = append(conns, conn) + } + + cm.BroadcastToAll(BroadcastMessage{Type: MessageTypeScanProgress}) + + for _, conn := range conns { + select { + case msg := <-conn.Send: + if msg.Type != MessageTypeScanProgress { + t.Errorf("%s: got %q, want scan_progress", conn.DeviceName, msg.Type) + } + default: + t.Errorf("%s: received nothing from BroadcastToAll", conn.DeviceName) + } + } +}