From d6480c1f02e5cec87498477d5c51d41295d38e02 Mon Sep 17 00:00:00 2001 From: xarmian Date: Sun, 23 Aug 2026 02:16:35 +0000 Subject: [PATCH] revert(sse): remove the ordering barrier; its failure mode is worse than the problem (BUG-2730, codex round 16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 16 found the third defect in a row inside the previous round's fix: the gap branch reset gapDrainBudget to the CURRENT queue depth on every signal, so a producer refilling faster than a slow client drains could re-raise the coalesced gap before the budget reached zero and the announcement would never fire — the exact starvation the budget was introduced to prevent, one level up. Rounds 13, 15 and 16 each found a defect in the fix from the round before. That pattern is the signal to stop patching and reassess, so I reassessed the barrier itself rather than fixing it a third time. What it prevented: a client receiving sync_required and then events queued before the hole, whose IDs re-establish a cursor below it. Bounded and self-correcting — the client was told to reconcile, and a later reconnect from such a cursor is refused by the coverage check and told again. What it risked: never announcing at all, on the connection type this whole unit exists for. Unbounded silence. A mechanism whose own failure class is worse than the one it fixes should not ship, so the barrier, its drain budget and its predicate are gone. The announcer and its cooldown stay: they answer a real feedback loop and they latch rather than drop, and their binding to both handlers is tested. The residual ordering behaviour is now documented in docs/deployment.md under what a client should do with sync_required, and in a comment at the gap branch — stated rather than left for a reader to find, which is the same posture as the rest of this unit. --- docs/deployment.md | 11 ++ internal/server/handlers_events.go | 73 ++++--------- internal/server/handlers_stream_gap_test.go | 105 ------------------- internal/server/handlers_watch_events.go | 73 ++++--------- internal/server/stream_gap_announcer.go | 22 ---- internal/server/stream_gap_announcer_test.go | 39 ------- 6 files changed, 55 insertions(+), 268 deletions(-) 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) - } - }) - } -}