diff --git a/docs/deployment.md b/docs/deployment.md index 91994af4..882f94a2 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -252,6 +252,17 @@ differently and a third-party consumer cannot infer this from the frame: - **Keep the connection open.** The frame is not a close and does not ask for a reconnect. The server keeps streaming; a client that tears down and redials on every `sync_required` turns one delta into a reconnect storm. +- **Expect events after it, possibly with IDs below the hole.** A mid-stream + `sync_required` is not ordered against events the server had already queued + for that connection, so a client can receive the frame and then events that + predate the gap. Their IDs re-establish a cursor at a position the server has + just disclaimed. This is deliberate and bounded: reconciling is what the + frame asked for, and a later reconnect from such a cursor is refused by the + coverage check and answered with `sync_required` again. Holding the + announcement back until those events drained was tried and removed — every + version of it could defer the announcement indefinitely while a busy + workspace kept the queue full, and an unbounded silence is worse than a + redundant resync. - **Stop trusting your cursor.** The empty `id:` retires it, so a compliant SSE client stops sending `Last-Event-ID` on its next reconnect. Do not re-send the old value: the server has just said it cannot vouch for that position. diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index f3721c0d..86a654d2 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -407,46 +407,8 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { return true } - // gapSeen latches a gap the bus reported until the event channel is - // empty — see the ordering barrier at the top of the loop. - gapSeen := false - // gapDrainBudget is how many already-queued events still owe delivery - // before the latched gap may be announced — see the barrier below. - gapDrainBudget := 0 - ctx := r.Context() for { - // ORDERING BARRIER (BUG-2730). The event channel and the gap channel - // are two arms of one select, so when both are ready Go picks at - // random — and announcing first, then draining events the subscriber - // queued BEFORE the hole, is the wrong order twice over. The client - // is told its position is untrustworthy and then handed IDs that - // re-establish one, below the hole; and on an ID-space change those - // queued events belong to the abandoned space. - // - // So the announcement waits for the events that were ALREADY QUEUED - // when the gap arrived. Nothing is discarded to achieve it, which - // matters most on the watch stream: a queued one-shot PUSH is not - // recoverable by any reconcile, so dropping it to make the cursor - // tidy would lose the only copy. - // - // THE WAIT IS BOUNDED BY A COUNT, NOT BY THE CHANNEL EMPTYING. An - // earlier version waited for len(ch) == 0 and claimed it could not - // starve; that is false under continuous refill, and the subscriber - // this signal exists for — a slow one on a busy workspace — is - // precisely the one whose channel never empties. gapDrainBudget is - // the queue depth captured at the moment the gap was latched, so once - // that many events have gone out, every event that was queued before - // the hole has been delivered and anything still waiting arrived - // after it. Terminating by construction, and exact rather than a - // timeout. - if gapReadyToAnnounce(gapSeen, len(ch), gapDrainBudget) { - gapSeen = false - if gapAnn.observe() && !announceGap() { - return - } - } - select { case <-ctx.Done(): // Client disconnected @@ -457,13 +419,6 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { // Channel closed (unsubscribed) return } - // Counts against the latched gap's drain budget whether or not - // this one is VISIBLE to the caller: the budget measures how many - // queued events predate the hole, and a filtered event occupied a - // slot exactly like a delivered one. - if gapDrainBudget > 0 { - gapDrainBudget-- - } if sseEventVisible(event) { // Broken-pipe / EOF on the live event path means the // client has gone away. Exit the handler so the bus @@ -510,12 +465,28 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { // named on BUG-2730 (replay from the buffer rather than resync; // the dropped events are still in it, 1024 deep against the // channel's 64). - // Latched, not announced here: the barrier at the top of the - // loop decides when, so anything already queued goes out first. - // The depth is captured NOW — it is the exact number of events - // that predate the hole, and it is what bounds the wait. - gapSeen = true - gapDrainBudget = len(ch) + // ORDERING AGAINST QUEUED EVENTS IS DELIBERATELY NOT GUARANTEED. + // This and the event channel are two arms of one select, so with + // both ready Go picks at random: a client can receive + // sync_required and then events that were queued BEFORE the hole, + // whose IDs re-establish the cursor this frame just retired, at a + // position below the hole. + // + // That is bounded and self-correcting — the client was told to + // reconcile, and a later reconnect from a below-the-hole cursor + // is refused by the coverage check and told again. A barrier that + // held the announcement until the pre-hole events had drained was + // built and REMOVED: every version of it could defer the + // announcement indefinitely under a producer that refills faster + // than a slow client drains, and an unbounded silence is a worse + // failure than a redundant resync. Documented in + // docs/deployment.md rather than left for a reader to discover. + if !gapAnn.observe() { + continue + } + if !announceGap() { + return + } case <-gapAnn.cool(): if gapAnn.flush() && !announceGap() { diff --git a/internal/server/handlers_stream_gap_test.go b/internal/server/handlers_stream_gap_test.go index 85844a42..156bee29 100644 --- a/internal/server/handlers_stream_gap_test.go +++ b/internal/server/handlers_stream_gap_test.go @@ -440,108 +440,3 @@ func assertGapBurstYieldsOneThenOne(t *testing.T, frames <-chan string, gaps cha // silently discarded. waitForFrameWithEvent(t, frames, "sync_required") } - -// TestQueuedEventsGoOutBeforeTheGapAnnouncement pins the ordering barrier. -// -// Both channels are arms of one select, so with both ready Go picks at random. -// Announcing first and then draining pre-hole events hands the client IDs that -// re-establish the cursor the announcement just retired, below the hole. The -// barrier makes the announcement wait for an empty channel. -// -// The test raises the gap and publishes in the same breath, then asserts the -// ORDER of the frames. Without the barrier this fails roughly half the time, -// which -count makes reliable. -func TestQueuedEventsGoOutBeforeTheGapAnnouncement(t *testing.T) { - srv := testServerWithEvents(t) - inner := events.New() - bus := &gapEventBus{EventBus: inner, gaps: make(chan struct{}, 1)} - srv.SetEventBus(bus) - ts := httptest.NewServer(srv) - defer ts.Close() - - slug := createTestWorkspace(t, ts.URL, "Test") - wsID := workspaceIDForSlug(t, srv, slug) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - frames := readRawSSEFrames(t, ctx, ts.URL+"/api/v1/events?workspace="+slug, "") - waitForFrameWithEvent(t, frames, "connected") - - // Queue events FIRST so they are sitting in the subscriber's channel, then - // raise the gap. Both select arms are now ready. - const queued = 20 - for range queued { - inner.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: wsID, Collection: "tasks"}) - } - bus.gaps <- struct{}{} - - seenEvents := 0 - deadline := time.After(5 * time.Second) - for { - select { - case f, ok := <-frames: - if !ok { - t.Fatalf("stream closed after %d/%d events and no announcement", seenEvents, queued) - } - switch { - case strings.Contains(f, "event: sync_required"): - if seenEvents != queued { - t.Fatalf("the gap was announced after %d of %d queued events; "+ - "the remaining ones would re-establish a cursor below the hole", seenEvents, queued) - } - return - case strings.Contains(f, "event: "+events.ItemCreated): - seenEvents++ - } - case <-deadline: - t.Fatalf("no announcement arrived; saw %d/%d events", seenEvents, queued) - } - } -} - -// TestGapAnnouncementIsNotStarvedByContinuousRefill is the counterfactual to -// the ordering barrier's own bound. -// -// The barrier's first version waited for an empty channel and its comment -// claimed that could not starve. It can: a publisher refilling faster than the -// client drains keeps len(ch) above zero forever, and the subscriber this -// signal exists for — a slow one on a busy workspace — is exactly the one -// whose channel never empties. The wait is bounded by the queue depth captured -// at latch time instead, so it terminates whatever the publisher does. -func TestGapAnnouncementIsNotStarvedByContinuousRefill(t *testing.T) { - srv := testServerWithEvents(t) - inner := events.New() - bus := &gapEventBus{EventBus: inner, gaps: make(chan struct{}, 1)} - srv.SetEventBus(bus) - ts := httptest.NewServer(srv) - defer ts.Close() - - slug := createTestWorkspace(t, ts.URL, "Test") - wsID := workspaceIDForSlug(t, srv, slug) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - frames := readRawSSEFrames(t, ctx, ts.URL+"/api/v1/events?workspace="+slug, "") - waitForFrameWithEvent(t, frames, "connected") - - // Refill continuously for the whole test, so the handler's channel is - // never observed empty. - stop := make(chan struct{}) - defer close(stop) - go func() { - for { - select { - case <-stop: - return - default: - inner.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: wsID, Collection: "tasks"}) - } - } - }() - - bus.gaps <- struct{}{} - - // The announcement must still arrive. Under the unbounded version this - // times out, because the emptiness condition is never satisfied. - waitForFrameWithEvent(t, frames, "sync_required") -} diff --git a/internal/server/handlers_watch_events.go b/internal/server/handlers_watch_events.go index 9414e5d4..9b74d16e 100644 --- a/internal/server/handlers_watch_events.go +++ b/internal/server/handlers_watch_events.go @@ -312,46 +312,8 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) return true } - // gapSeen latches a gap until the notification channel is empty — see the - // ordering barrier at the top of the loop. - gapSeen := false - // gapDrainBudget is how many already-queued events still owe delivery - // before the latched gap may be announced — see the barrier below. - gapDrainBudget := 0 - ctx := r.Context() for { - // ORDERING BARRIER (BUG-2730). The event channel and the gap channel - // are two arms of one select, so when both are ready Go picks at - // random — and announcing first, then draining events the subscriber - // queued BEFORE the hole, is the wrong order twice over. The client - // is told its position is untrustworthy and then handed IDs that - // re-establish one, below the hole; and on an ID-space change those - // queued events belong to the abandoned space. - // - // So the announcement waits for the events that were ALREADY QUEUED - // when the gap arrived. Nothing is discarded to achieve it, which - // matters most on the watch stream: a queued one-shot PUSH is not - // recoverable by any reconcile, so dropping it to make the cursor - // tidy would lose the only copy. - // - // THE WAIT IS BOUNDED BY A COUNT, NOT BY THE CHANNEL EMPTYING. An - // earlier version waited for len(ch) == 0 and claimed it could not - // starve; that is false under continuous refill, and the subscriber - // this signal exists for — a slow one on a busy workspace — is - // precisely the one whose channel never empties. gapDrainBudget is - // the queue depth captured at the moment the gap was latched, so once - // that many events have gone out, every event that was queued before - // the hole has been delivered and anything still waiting arrived - // after it. Terminating by construction, and exact rather than a - // timeout. - if gapReadyToAnnounce(gapSeen, len(ch), gapDrainBudget) { - gapSeen = false - if gapAnn.observe() && !announceGap() { - return - } - } - select { case <-ctx.Done(): return @@ -360,13 +322,6 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) if !ok { return } - // Counts against the latched gap's drain budget whether or not - // this one is VISIBLE to the caller: the budget measures how many - // queued events predate the hole, and a filtered event occupied a - // slot exactly like a delivered one. - if gapDrainBudget > 0 { - gapDrainBudget-- - } if !watchNotificationVisible(watches, visCache.forWorkspace(n.WorkspaceID), user.ID, sessionID, armed, n) { continue } @@ -400,12 +355,28 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) // // Rate-limited per connection with nothing dropped — a gap inside // the window is latched, not discarded. See gapAnnouncer. - // Latched, not announced here: the barrier at the top of the - // loop decides when, so anything already queued goes out first. - // The depth is captured NOW — it is the exact number of events - // that predate the hole, and it is what bounds the wait. - gapSeen = true - gapDrainBudget = len(ch) + // ORDERING AGAINST QUEUED EVENTS IS DELIBERATELY NOT GUARANTEED. + // This and the event channel are two arms of one select, so with + // both ready Go picks at random: a client can receive + // sync_required and then events that were queued BEFORE the hole, + // whose IDs re-establish the cursor this frame just retired, at a + // position below the hole. + // + // That is bounded and self-correcting — the client was told to + // reconcile, and a later reconnect from a below-the-hole cursor + // is refused by the coverage check and told again. A barrier that + // held the announcement until the pre-hole events had drained was + // built and REMOVED: every version of it could defer the + // announcement indefinitely under a producer that refills faster + // than a slow client drains, and an unbounded silence is a worse + // failure than a redundant resync. Documented in + // docs/deployment.md rather than left for a reader to discover. + if !gapAnn.observe() { + continue + } + if !announceGap() { + return + } case <-gapAnn.cool(): if gapAnn.flush() && !announceGap() { diff --git a/internal/server/stream_gap_announcer.go b/internal/server/stream_gap_announcer.go index d75954e0..bf3580ee 100644 --- a/internal/server/stream_gap_announcer.go +++ b/internal/server/stream_gap_announcer.go @@ -40,28 +40,6 @@ func (s *Server) gapCooldown() time.Duration { return midStreamGapCooldown } -// gapReadyToAnnounce is the SSE handlers' ordering barrier, as a predicate. -// -// latched says a gap is waiting; queued is the subscriber channel's current -// depth; budget is the depth captured when the gap was latched, decremented -// once per event taken off the channel since. -// -// The announcement waits for the events that were already queued when the gap -// arrived, so a client is not told its position is untrustworthy and then -// immediately handed IDs that re-establish one, below the hole. It waits for -// them BY COUNT rather than for the channel to empty: an emptiness-only -// condition never fires under continuous refill, and the subscriber this -// signal exists for — a slow one on a busy workspace — is exactly the one -// whose channel never empties. -// -// A function rather than an expression inlined in two select loops because the -// starvation case is not reliably reproducible end-to-end (the channel does -// briefly empty under most schedulings), so the bound needs somewhere it can -// be asserted directly. -func gapReadyToAnnounce(latched bool, queued, budget int) bool { - return latched && (queued == 0 || budget <= 0) -} - // gapAnnouncer rate-limits one connection's mid-stream gap announcements // without losing any. // diff --git a/internal/server/stream_gap_announcer_test.go b/internal/server/stream_gap_announcer_test.go index a9312ee2..48795737 100644 --- a/internal/server/stream_gap_announcer_test.go +++ b/internal/server/stream_gap_announcer_test.go @@ -78,42 +78,3 @@ func TestGapAnnouncerDoesNotAnnounceAnEmptyWindow(t *testing.T) { t.Error("a gap after a quiet window was bounded; the window had already closed") } } - -// TestGapReadyToAnnounceTerminatesUnderContinuousRefill is the bound the -// handlers' barrier rests on, asserted where it can be. -// -// The first version of that barrier waited for an empty channel and its -// comment claimed that could not starve. It can — a publisher refilling faster -// than a slow client drains keeps the depth above zero indefinitely — and an -// end-to-end test does not reliably reproduce it, because under most -// schedulings the channel does briefly empty. Which is exactly why the bound -// is a predicate: the starvation case can be stated here even though it cannot -// be provoked there. -func TestGapReadyToAnnounceTerminatesUnderContinuousRefill(t *testing.T) { - cases := []struct { - name string - latched bool - queued, budget int - want bool - }{ - {"no gap latched, nothing to announce", false, 0, 0, false}, - {"no gap latched, and a drained budget does not invent one", false, 5, 0, false}, - {"latched with an empty channel announces at once", true, 0, 0, true}, - {"latched with queued events waits for them", true, 5, 5, false}, - {"latched, partway through the queue, still waits", true, 3, 2, false}, - // The starvation case. The channel is STILL not empty — a publisher - // has refilled it — but every event that predated the hole has gone - // out, so the wait is over. An emptiness-only condition answers false - // here, forever. - {"latched, budget spent, channel refilled: announces anyway", true, 40, 0, true}, - {"latched, budget overspent", true, 1, -3, true}, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - if got := gapReadyToAnnounce(tc.latched, tc.queued, tc.budget); got != tc.want { - t.Errorf("gapReadyToAnnounce(%v, %d, %d) = %v, want %v", - tc.latched, tc.queued, tc.budget, got, tc.want) - } - }) - } -}