From 3a00783557eddb01aa05ecc1bec6f719133e4dec Mon Sep 17 00:00:00 2001 From: xarmian Date: Sun, 23 Aug 2026 01:54:43 +0000 Subject: [PATCH] fix(sse): queued events go out before the gap announcement (BUG-2730, codex round 13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reading both handlers as state machines: the event channel and the gap channel are two arms of one select, so with both ready Go picks at random. Announcing first and 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 immediately handed IDs that re-establish one, below the hole; and on an ID-space change those queued events belong to the space that was just abandoned. The gap is now latched and the announcement waits for an empty channel. Nothing is discarded to achieve it, and that restraint is the load-bearing part on the watch stream: a queued one-shot PUSH cannot be recovered by any reconcile, so dropping it to make the cursor tidy would destroy the only copy. Draining cannot starve the announcement either — one event per iteration, re-checked at the top, so it lands on the first iteration with nothing queued, immediately when the channel was already empty. Pinned by asserting the ORDER of the frames with twenty events queued ahead of the gap. Without the barrier that fails on the first or third event, roughly half the time per run. --- internal/server/handlers_events.go | 37 ++++++++++--- internal/server/handlers_stream_gap_test.go | 58 +++++++++++++++++++++ internal/server/handlers_watch_events.go | 37 ++++++++++--- 3 files changed, 120 insertions(+), 12 deletions(-) diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index 310a354c..81c5f54f 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -407,8 +407,36 @@ 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 + 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 until the channel is empty. 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. + // + // Draining cannot starve the announcement — the loop delivers one + // event per iteration and re-checks, so the announcement lands on the + // first iteration where nothing is queued, immediately if the channel + // was empty when the gap arrived. + if gapSeen && len(ch) == 0 { + gapSeen = false + if gapAnn.observe() && !announceGap() { + return + } + } + select { case <-ctx.Done(): // Client disconnected @@ -465,12 +493,9 @@ 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). - if !gapAnn.observe() { - continue - } - if !announceGap() { - return - } + // Latched, not announced here: the barrier at the top of the + // loop decides when, so anything already queued goes out first. + gapSeen = true 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 156bee29..1d71f3dc 100644 --- a/internal/server/handlers_stream_gap_test.go +++ b/internal/server/handlers_stream_gap_test.go @@ -440,3 +440,61 @@ 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) + } + } +} diff --git a/internal/server/handlers_watch_events.go b/internal/server/handlers_watch_events.go index 12f870c1..e5ccfc30 100644 --- a/internal/server/handlers_watch_events.go +++ b/internal/server/handlers_watch_events.go @@ -312,8 +312,36 @@ 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 + 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 until the channel is empty. 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. + // + // Draining cannot starve the announcement — the loop delivers one + // event per iteration and re-checks, so the announcement lands on the + // first iteration where nothing is queued, immediately if the channel + // was empty when the gap arrived. + if gapSeen && len(ch) == 0 { + gapSeen = false + if gapAnn.observe() && !announceGap() { + return + } + } + select { case <-ctx.Done(): return @@ -355,12 +383,9 @@ 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. - if !gapAnn.observe() { - continue - } - if !announceGap() { - return - } + // Latched, not announced here: the barrier at the top of the + // loop decides when, so anything already queued goes out first. + gapSeen = true case <-gapAnn.cool(): if gapAnn.flush() && !announceGap() {