diff --git a/changelog.d/fixed-6218-sse-gap-signal.md b/changelog.d/fixed-6218-sse-gap-signal.md new file mode 100644 index 000000000..cb2f45062 --- /dev/null +++ b/changelog.d/fixed-6218-sse-gap-signal.md @@ -0,0 +1 @@ +- The contributor activity stream (`/api/contribute/events`) now tells a client when it missed events ([#6218](https://github.com/hivecommons/hive/issues/6218)). A subscriber whose channel fills has events discarded rather than back-pressuring the hub — that is deliberate and unchanged, since a slow observer must never stall worker assignment — but the loss was invisible: the connection stayed open, the 25-second heartbeat kept it looking healthy, and a long-lived client could not tell "nothing happened" from "I missed events". For an unattended monitor that is the dangerous failure, because stale data renders as current until some unrelated reconnect repairs it. Two additive fields fix it. Every frame now carries `seq`, a monotonic stream position assigned once at fan-out so the same event has the same number for every subscriber — a jump from 41 to 45 means three events never arrived, and the `hello` frame reports the position the client connected at so it has a baseline. And when events are discarded for a connection, the server sends a `gap` frame naming how many went and where the stream has reached. The connection stays open and reconnect-with-replay is still the recovery path; what changed is that a client now knows to take it. The gap is reported before the next activity frame, and on the heartbeat tick if none arrives, so a stream that has gone quiet still surfaces it rather than sitting on pings forever. Existing clients ignore both fields and are unaffected; the `/contribute` page uses the signal to re-sync immediately instead of waiting out its poll timer. diff --git a/dashboard/openapi.json b/dashboard/openapi.json index 991664269..f13fec933 100644 --- a/dashboard/openapi.json +++ b/dashboard/openapi.json @@ -10792,10 +10792,10 @@ "Contribute" ], "summary": "Live contributor activity event stream (SSE)", - "description": "Server-Sent Events stream (Content-Type: text/event-stream), NOT a single JSON response. Public, read-only \u2014 anonymous browsers may subscribe. On connect it sends one 'hello' frame carrying a bounded replay of recent activity plus a ready-work queue snapshot (and, when the convergence-diagnostics shadow-mode toggle is on, additive 'withheld'/'admission_coverage' fields), then forwards each subsequent activity event as an 'activity' frame. An idle connection receives a ': ping' comment roughly every 25s to keep intermediaries from timing out the stream. Each event line is JSON-encoded per the sseEvent shape ({type, activity?, replay?, queue?, withheld?, admission_coverage?}).", + "description": "Server-Sent Events stream (Content-Type: text/event-stream), NOT a single JSON response. Public, read-only \u2014 anonymous browsers may subscribe. On connect it sends one 'hello' frame carrying a bounded replay of recent activity plus a ready-work queue snapshot (and, when the convergence-diagnostics shadow-mode toggle is on, additive 'withheld'/'admission_coverage' fields), then forwards each subsequent activity event as an 'activity' frame. An idle connection receives a ': ping' comment roughly every 25s to keep intermediaries from timing out the stream. Each event line is JSON-encoded per the sseEvent shape ({type, activity?, replay?, queue?, seq?, dropped?, withheld?, admission_coverage?}).\n\nGap detection (#6218): every frame carries 'seq', a monotonic stream position assigned once at fan-out, so the same event has the same number for every subscriber. On 'hello' it is the position when this client connected, so its first 'activity' frame should be seq+1; a larger jump means events were lost. A subscriber whose channel fills has events DISCARDED rather than being back-pressured onto the hub (unchanged, and deliberate \u2014 a slow observer must not stall worker assignment), but the loss is no longer silent: the server then sends a 'gap' frame carrying 'dropped' (how many events this connection missed) and 'seq' (the position reached). The connection stays open; reconnecting replays the recent buffer, which is the recovery path. The gap frame is emitted before the next activity frame, and on the heartbeat tick if none arrives, so a client that has gone quiet still learns it is behind rather than reading pings as health.", "responses": { "200": { - "description": "text/event-stream of sseEvent frames (hello, then activity, with periodic heartbeat comments)", + "description": "text/event-stream of sseEvent frames (hello, then activity, a gap frame when events were dropped for this connection, with periodic heartbeat comments)", "content": { "text/event-stream": { "schema": { diff --git a/src/docs/api-reference.md b/src/docs/api-reference.md index e56f123e4..030723a1d 100644 --- a/src/docs/api-reference.md +++ b/src/docs/api-reference.md @@ -266,7 +266,7 @@ Read the result from `GET /api/kick/{agent}/status`, which returns `status` of ` | `GET` | `/api/contribute/status` | Public | Contribute Status | `pkg/dashboard/api_contribute.go:242` | | `GET` | `/api/contribute/activity` | Public | Contribute Activity | `pkg/dashboard/api_contribute.go:243` | | `GET` | `/api/contribute/fleet` | Public | Contribute Fleet | `pkg/dashboard/api_contribute.go:244` | -| `GET` | `/api/contribute/events` | Public | Contribute Events | `pkg/dashboard/api_contribute.go:248` | +| `GET` | `/api/contribute/events` | Public | Contribute Events (SSE). Every frame carries a monotonic `seq`; a connection whose channel filled gets a `gap` frame naming how many events it missed, so a long-lived client can tell "nothing happened" from "I missed events" ([#6218](https://github.com/hivecommons/hive/issues/6218)) | `pkg/dashboard/api_contribute.go:248` | | `GET` | `/api/contribute/queue` | Public | Contribute Queue | `pkg/dashboard/api_contribute.go:251` | | `GET` | `/api/contribute/opportunistic` | Public | Contribute Opportunistic | `pkg/dashboard/api_contribute.go:255` | | `GET` | `/api/contribute/limits` | Public | Contribute Limits | `pkg/dashboard/api_contribute.go:260` | diff --git a/src/pkg/dashboard/api_contribute.go b/src/pkg/dashboard/api_contribute.go index f8e2301b1..f27216a8a 100644 --- a/src/pkg/dashboard/api_contribute.go +++ b/src/pkg/dashboard/api_contribute.go @@ -5951,6 +5951,24 @@ function ccOnActivity(e){ } } +// ── A reported stream gap (#6218) ────────────────────────────────────────────── +// The server tells us when it discarded events for this connection because our +// channel was full. It is not an error and the stream stays open — what it means +// is that this page is now missing something, so the cheap repair is to re-read +// the reliable endpoints at once rather than wait out the 6s poll. +// +// This page already polled as a hedge, so the gap only makes it prompt. The +// clients this signal actually rescues are the headless ones that trusted the +// stream and had no way to learn they were behind. +function ccOnGap(ev){ + var n=(ev&&ev.dropped)||0; + console.warn('contribute SSE: missed '+n+' event(s) (stream at seq '+((ev&&ev.seq)||'?')+'); resyncing'); + try{ccPollActivity();}catch(e){} + try{fetch('/api/contribute/queue').then(function(r){return r.json();}).then(function(d){ + if(d&&d.queue){ccQueue=d.queue.slice();ccRenderQueue();} + }).catch(function(){});}catch(e){} +} + // ── SSE lifecycle with graceful fallback ─────────────────────────────────────── function ccHydrate(payload){ if(payload.queue){ccQueue=payload.queue.slice();ccRenderQueue();} @@ -6000,6 +6018,7 @@ function ccStart(){ try{var ev=JSON.parse(m.data);}catch(err){return;} if(ev.type==='hello')ccHydrate(ev); else if(ev.type==='activity'&&ev.activity)ccOnActivity(ev.activity); + else if(ev.type==='gap')ccOnGap(ev); }; ccEs.onerror=function(){ // Stream dropped. Show polling state, start the queue fallback, and let the diff --git a/src/pkg/dashboard/contribute_sse.go b/src/pkg/dashboard/contribute_sse.go index 02ce58360..652d88b65 100644 --- a/src/pkg/dashboard/contribute_sse.go +++ b/src/pkg/dashboard/contribute_sse.go @@ -7,6 +7,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/hivecommons/hive/pkg/config" @@ -58,10 +59,27 @@ const readyQueueDefaultLimit = 150 // initial hydration payload (queue + replay) from subsequent single activity // events so the client can render each without guessing. type sseEvent struct { - Type string `json:"type"` // "activity" | "hello" + Type string `json:"type"` // "activity" | "hello" | "gap" Activity *ActivityEntry `json:"activity,omitempty"` // set when Type=="activity" Replay []ActivityEntry `json:"replay,omitempty"` // set when Type=="hello" Queue []ReadyQueueItem `json:"queue,omitempty"` // set when Type=="hello" + // Seq is the stream position (#6218). It is a monotonic counter over every + // event broadcast, assigned once at fan-out so all subscribers see the SAME + // number for the same event. On "activity" it is that event's position; on + // "hello" it is the position the stream had when this subscriber registered, + // so the first activity it receives should be Seq+1; on "gap" it is the + // position reached so far. + // + // A client that tracks it detects loss on its own: a jump from 41 to 45 + // means three events never arrived. That is the primitive the "gap" frame + // below cannot give precisely — see the ordering note in flushGap. + // + // omitempty, so a stream that has broadcast nothing yet (Seq 0) serialises + // exactly as it did before this field existed. + Seq uint64 `json:"seq,omitempty"` + // Dropped is how many events were discarded for THIS subscriber because its + // channel was full. Set only on Type=="gap". + Dropped uint64 `json:"dropped,omitempty"` // Withheld and AdmissionCoverage are the #4246 convergence admission // diagnostics, set on the "hello" frame ONLY when the convergence toggle is // in shadow mode (default off → both absent, payload unchanged). They come @@ -132,6 +150,20 @@ func (it ReadyQueueItem) identityKey() string { // it. Guarded by the hub's sseMu. type sseSubscriber struct { events chan sseEvent + // startSeq is the stream position when this subscriber registered, captured + // under the registry lock so it cannot straddle a concurrent broadcast. The + // hello frame reports it, which is what lets a client tell an event it + // missed from one that simply had not happened yet. + // + // It is read at registration and never written again, so the handler reads + // it without the lock. + startSeq uint64 + // dropped counts events discarded for this subscriber because its channel + // was full. Atomic because broadcast increments it under the REGISTRY lock + // while the HTTP writer drains it holding nothing — and the writer must + // never take that lock, since the whole point of the non-blocking send is + // that a slow client cannot reach the hub's event path (#6218). + dropped atomic.Uint64 } // sseRegistry holds the live SSE subscribers. It is a small struct on the hub so @@ -139,6 +171,9 @@ type sseSubscriber struct { type sseRegistry struct { mu sync.Mutex subs map[*sseSubscriber]struct{} + // seq is the monotonic stream position, incremented once per broadcast so a + // single event carries one number for every subscriber. Guarded by mu. + seq uint64 } func newSSERegistry() *sseRegistry { @@ -150,11 +185,22 @@ func newSSERegistry() *sseRegistry { func (r *sseRegistry) subscribe() *sseSubscriber { sub := &sseSubscriber{events: make(chan sseEvent, sseSubscriberBuffer)} r.mu.Lock() + // Under the same lock as the registration, so the recorded position and the + // set of events this subscriber will receive cannot disagree: every event + // numbered above startSeq is one it was registered for. + sub.startSeq = r.seq r.subs[sub] = struct{}{} r.mu.Unlock() return sub } +// currentSeq returns the stream position reached so far. +func (r *sseRegistry) currentSeq() uint64 { + r.mu.Lock() + defer r.mu.Unlock() + return r.seq +} + // unsubscribe removes a subscriber and closes its channel. Idempotent: a second // call (e.g. handler cleanup after a broadcast already dropped a wedged sub) is a // no-op because the map delete guards the close. @@ -171,15 +217,32 @@ func (r *sseRegistry) unsubscribe(sub *sseSubscriber) { // subscriber whose buffer is full is skipped (its browser will reconnect and // replay) — the broadcast must never block the caller, which is the hub's event // path. This is the leak-safe, back-pressure-free contract the spec requires. +// +// Dropping rather than blocking stays exactly as it was: a slow observer must +// not stall worker assignment. What changed in #6218 is that the drop is no +// longer INVISIBLE. Each event carries a monotonic Seq, and a discarded one +// bumps that subscriber's dropped counter, which the HTTP writer turns into a +// "gap" frame. Before, a long-lived client could not tell "nothing happened" +// from "I missed events", so an unattended monitor rendered stale data as +// current until some unrelated reconnect repaired it. func (r *sseRegistry) broadcast(ev sseEvent) { r.mu.Lock() defer r.mu.Unlock() + // Numbered here, once, so every subscriber sees the same position for the + // same event — a per-subscriber counter would make the numbers unusable for + // comparing two clients or for reasoning about the stream as a whole. + r.seq++ + ev.Seq = r.seq for sub := range r.subs { select { case sub.events <- ev: default: // Subscriber is behind; drop this event for it rather than block the - // hub. The client's ring-buffer replay on reconnect recovers state. + // hub. Record the loss so the writer can tell the client about it — + // the client's ring-buffer replay on reconnect is still what + // RECOVERS the state, but it only gets a chance to if it knows it + // needs to. + sub.dropped.Add(1) } } } @@ -581,6 +644,11 @@ func (s *Server) handleContributeEvents(w http.ResponseWriter, r *http.Request) Type: "hello", Replay: replay, Queue: snap.queue, + // The position this subscriber started at (#6218). The first "activity" + // frame it receives should be Seq+1; anything higher means events were + // lost between hydration and delivery, which a client can see without + // waiting for a "gap" frame. + Seq: sub.startSeq, } if diag { hello.Withheld = snap.withheld @@ -598,6 +666,41 @@ func (s *Server) handleContributeEvents(w http.ResponseWriter, r *http.Request) ticker := time.NewTicker(sseHeartbeatInterval) defer ticker.Stop() + // flushGap emits a "gap" frame when this subscriber has had events discarded + // for a full channel (#6218), and reports whether writing succeeded. + // + // It runs on BOTH loop branches, and which one fires is worth being precise + // about. In practice it is the EVENT branch: a drop can only happen when the + // channel is full, so there are ~32 queued events behind it and this check + // runs before each of those pops. The heartbeat branch is the guarantee, not + // the usual path — it is what makes "reported" unconditional rather than + // contingent on another event ever arriving for this subscriber, which is + // the property the report asks for (a client that goes quiet must still + // learn it is behind, instead of reading heartbeats as health forever). + // + // Swap-to-zero, so a gap is reported exactly once and a drop that lands + // between the read and the write is carried into the next frame rather than + // lost. + // + // Ordering, stated honestly: the frame says "you are missing events", not + // "the events after this one are the ones you missed". A full channel means + // the DISCARDED event was newer than the ~32 still queued, so this fires + // while the client is still draining good ones. That is deliberate — telling + // a client early that it is behind is strictly better than telling it late — + // and it is exactly why every event carries Seq: the sequence numbers locate + // the discontinuity precisely, while this frame is the prompt to go looking. + flushGap := func() bool { + n := sub.dropped.Swap(0) + if n == 0 { + return true + } + return writeSSE(w, sseEvent{ + Type: "gap", + Dropped: n, + Seq: s.contributeHub.sse.currentSeq(), + }) + } + for { select { case <-ctx.Done(): @@ -606,11 +709,17 @@ func (s *Server) handleContributeEvents(w http.ResponseWriter, r *http.Request) if !open { return // registry closed our channel (server shutdown / forced drop) } + if !flushGap() { + return + } if !writeSSE(w, ev) { return } flusher.Flush() case <-ticker.C: + if !flushGap() { + return + } if _, err := fmt.Fprint(w, ": ping\n\n"); err != nil { return } @@ -622,7 +731,13 @@ func (s *Server) handleContributeEvents(w http.ResponseWriter, r *http.Request) // sseHeartbeatInterval is how often an idle stream emits an SSE comment ping. It is // short enough to keep proxies/load-balancers from closing an idle connection and // to detect a dead client promptly, without meaningful overhead. -const sseHeartbeatInterval = 25 * time.Second +// +// A var rather than a const solely so tests can shorten it (the same seam +// CACertPath uses in pkg/proxy); production never reassigns it. The heartbeat +// branch is where a gap gets reported on a stream that has gone quiet — the +// exact case #6218 describes — so leaving it reachable only by a 25-second wait +// would mean leaving it untested. +var sseHeartbeatInterval = 25 * time.Second // writeSSE marshals one event and writes it as a single SSE "data:" frame. Returns // false on a write/marshal error so the caller can tear down the subscriber. diff --git a/src/pkg/dashboard/contribute_sse_gap_test.go b/src/pkg/dashboard/contribute_sse_gap_test.go new file mode 100644 index 000000000..199fa98eb --- /dev/null +++ b/src/pkg/dashboard/contribute_sse_gap_test.go @@ -0,0 +1,374 @@ +package dashboard + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// ── Stream-gap observability (#6218) ──────────────────────────────────────── +// +// Dropping an event for a full subscriber is deliberate and stays: a slow +// observer must never back-pressure the hub's event path. What #6218 reported is +// that the loss was UNOBSERVABLE — the connection stayed open, the 25s heartbeat +// kept it looking healthy, and a long-lived client could not tell "nothing +// happened" from "I missed events". These tests pin the two things that now make +// it observable: a monotonic Seq on every frame, and a "gap" frame naming how +// many events went. + +// Every broadcast advances the position, and one event carries the SAME number +// for every subscriber — a per-subscriber counter would make the numbers useless +// for comparing two clients or reasoning about the stream as a whole. +func TestSSEBroadcastAssignsMonotonicSeq(t *testing.T) { + reg := newSSERegistry() + a := reg.subscribe() + b := reg.subscribe() + + if a.startSeq != 0 || b.startSeq != 0 { + t.Fatalf("fresh registry should start at 0, got a=%d b=%d", a.startSeq, b.startSeq) + } + + for i := 0; i < 3; i++ { + reg.broadcast(sseEvent{Type: "activity", Activity: &ActivityEntry{Username: "u"}}) + } + if got := reg.currentSeq(); got != 3 { + t.Errorf("currentSeq() = %d, want 3", got) + } + + var aSeqs, bSeqs []uint64 + for i := 0; i < 3; i++ { + aSeqs = append(aSeqs, (<-a.events).Seq) + bSeqs = append(bSeqs, (<-b.events).Seq) + } + for i, want := range []uint64{1, 2, 3} { + if aSeqs[i] != want { + t.Errorf("subscriber a event %d has seq %d, want %d", i, aSeqs[i], want) + } + if bSeqs[i] != aSeqs[i] { + t.Errorf("the same event was numbered %d for one subscriber and %d for another", aSeqs[i], bSeqs[i]) + } + } +} + +// A subscriber that joins mid-stream records where it came in, so its hello frame +// can tell the client which events it was never entitled to. +func TestSSESubscribeRecordsStartPosition(t *testing.T) { + reg := newSSERegistry() + reg.broadcast(sseEvent{Type: "activity"}) + reg.broadcast(sseEvent{Type: "activity"}) + + late := reg.subscribe() + if late.startSeq != 2 { + t.Fatalf("late subscriber startSeq = %d, want 2", late.startSeq) + } + reg.broadcast(sseEvent{Type: "activity"}) + + ev := <-late.events + if ev.Seq != late.startSeq+1 { + t.Errorf("first event after joining has seq %d, want startSeq+1 (%d)", ev.Seq, late.startSeq+1) + } + if len(late.events) != 0 { + t.Errorf("a late subscriber received %d event(s) from before it joined", len(late.events)) + } +} + +// The counter is what turns a silent discard into something reportable. This is +// the paired half of TestBroadcastDropsSlowSubscriber: that one pins that the +// event is dropped, this one pins that the drop is RECORDED. +func TestSSEBroadcastCountsDroppedEvents(t *testing.T) { + reg := newSSERegistry() + sub := reg.subscribe() + for i := 0; i < sseSubscriberBuffer; i++ { + sub.events <- sseEvent{Type: "activity"} + } + if got := sub.dropped.Load(); got != 0 { + t.Fatalf("dropped = %d before any overflow, want 0", got) + } + + for i := 0; i < 3; i++ { + reg.broadcast(sseEvent{Type: "activity", Activity: &ActivityEntry{Username: "overflow"}}) + } + + if got := sub.dropped.Load(); got != 3 { + t.Errorf("dropped = %d after three overflowed broadcasts, want 3", got) + } + // The position still advances: the events happened, this subscriber just did + // not get them. A client comparing its last seen seq against a later frame's + // is exactly how it notices. + if got := reg.currentSeq(); got != 3 { + t.Errorf("currentSeq() = %d, want 3 — a dropped event still occupies a position", got) + } + if len(sub.events) != sseSubscriberBuffer { + t.Errorf("channel length = %d, want %d (an overflow was enqueued anyway)", len(sub.events), sseSubscriberBuffer) + } +} + +// sseFrames splits a recorded SSE body into the decoded JSON data frames, +// ignoring `: ping` comments. +func sseFrames(t *testing.T, body string) []sseEvent { + t.Helper() + var out []sseEvent + for _, line := range strings.Split(body, "\n") { + line = strings.TrimSpace(line) + if !strings.HasPrefix(line, "data: ") { + continue + } + var ev sseEvent + if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err != nil { + t.Fatalf("undecodable SSE frame %q: %v", line, err) + } + out = append(out, ev) + } + return out +} + +// The end-to-end shape of the fix: a connected client whose channel overflows is +// told, over the SAME still-open connection, that it missed events — and the +// gap frame arrives BEFORE the next activity frame, so the client learns it is +// behind before it is handed anything to render. +func TestSSEHandlerEmitsGapFrameAfterDrops(t *testing.T) { + setupContributeEnv(t) + s := NewServer(0, slog.Default()) + s.registerContributeRoutes() + hub := s.contributeHub + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest(http.MethodGet, "/api/contribute/events", nil).WithContext(ctx) + rec := newSyncRecorder() + + done := make(chan struct{}) + go func() { + s.handleContributeEvents(rec, req) + close(done) + }() + waitFor(t, func() bool { return hub.sse.count() == 1 }, "subscriber to register") + + // Reach into the one registered subscriber and starve it: fill its channel so + // the next broadcasts have nowhere to go. This is the state a slow browser or + // a stalled monitor reaches on its own. + hub.sse.mu.Lock() + var sub *sseSubscriber + for candidate := range hub.sse.subs { + sub = candidate + } + hub.sse.mu.Unlock() + if sub == nil { + t.Fatal("no subscriber registered") + } + for len(sub.events) < cap(sub.events) { + sub.events <- sseEvent{Type: "activity", Activity: &ActivityEntry{Username: "filler"}} + } + + // Two events the client will never see. + hub.addActivity("lost-one", "picked up", "contributor", "claude", "sonnet", "", "acme/repo#1") + hub.addActivity("lost-two", "picked up", "contributor", "claude", "sonnet", "", "acme/repo#2") + waitFor(t, func() bool { return sub.dropped.Load() == 2 }, "both events to be recorded as dropped") + + // Drain one queued event so the writer wakes and reports the gap. (The + // heartbeat would do it too, but this test is not waiting 25 seconds for it — + // TestSSEGapIsReportedOnTheHeartbeatPath covers that branch directly.) + <-sub.events + + waitFor(t, func() bool { return strings.Contains(rec.BodyString(), `"gap"`) }, "a gap frame to be written") + + frames := sseFrames(t, rec.BodyString()) + var gap *sseEvent + var gapIndex, firstActivityAfterGap = -1, -1 + for i := range frames { + if frames[i].Type == "gap" && gap == nil { + gap = &frames[i] + gapIndex = i + } + if gap != nil && frames[i].Type == "activity" && firstActivityAfterGap < 0 && i > gapIndex { + firstActivityAfterGap = i + } + } + if gap == nil { + t.Fatalf("no gap frame in the stream:\n%s", rec.BodyString()) + } + if gap.Dropped != 2 { + t.Errorf("gap frame reports %d dropped, want 2", gap.Dropped) + } + if gap.Seq == 0 { + t.Error("gap frame carries no stream position, so a client cannot tell how far it is behind") + } + if frames[0].Type != "hello" { + t.Errorf("first frame is %q, want hello", frames[0].Type) + } + if gapIndex == 0 { + t.Error("the gap frame preceded the hello frame") + } + + // Reported once, not on every subsequent wake-up. + if got := sub.dropped.Load(); got != 0 { + t.Errorf("dropped = %d after the gap was reported, want 0 (it would repeat forever)", got) + } + + // And the connection is still open — the fix is a signal, not a disconnect. + select { + case <-done: + t.Fatal("the handler exited; a reported gap must not tear down the stream") + default: + } + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("handler did not return after context cancel") + } +} + +// The heartbeat branch, exercised on its own. +// +// A pending drop is marked DIRECTLY on the subscriber here rather than through +// broadcast, and that is deliberate: a real drop implies a full channel, so the +// event branch would pop one of the ~32 queued events and report the gap before +// the ticker ever fired. Driving broadcast would therefore test the event path +// again while looking like it tested this one — which is exactly what an earlier +// draft of this test did, until removing the heartbeat check left it green. +// +// What this pins is the branch's contract: with drops pending and NOTHING else +// arriving, the client still learns. That is the property the report asks for — +// a monitor that goes quiet must not read heartbeats as health forever — and it +// is the half the event path cannot provide. +// +// The heartbeat is shortened for the test rather than waited out; production +// stays at 25 seconds. +func TestSSEGapIsReportedOnAnIdleStream(t *testing.T) { + setupContributeEnv(t) + restore := sseHeartbeatInterval + sseHeartbeatInterval = 20 * time.Millisecond + t.Cleanup(func() { sseHeartbeatInterval = restore }) + + s := NewServer(0, slog.Default()) + s.registerContributeRoutes() + hub := s.contributeHub + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest(http.MethodGet, "/api/contribute/events", nil).WithContext(ctx) + rec := newSyncRecorder() + done := make(chan struct{}) + go func() { + s.handleContributeEvents(rec, req) + close(done) + }() + waitFor(t, func() bool { return hub.sse.count() == 1 }, "subscriber to register") + + hub.sse.mu.Lock() + var sub *sseSubscriber + for candidate := range hub.sse.subs { + sub = candidate + } + hub.sse.mu.Unlock() + if sub == nil { + t.Fatal("no subscriber registered") + } + // The stream is idle: the channel is empty and the writer is parked in its + // select. Mark a drop as pending — the state broadcast leaves behind — and + // then let nothing else happen at all. + if len(sub.events) != 0 { + t.Fatalf("expected an idle subscriber, found %d queued event(s)", len(sub.events)) + } + sub.dropped.Add(1) + + // Deliberately no broadcast and no drain from here on — only the heartbeat + // ticking against a client that is now behind. + waitFor(t, func() bool { return strings.Contains(rec.BodyString(), `"gap"`) }, + "an idle stream to report the gap on its heartbeat") + + var gap *sseEvent + for _, f := range sseFrames(t, rec.BodyString()) { + if f.Type == "gap" { + ev := f + gap = &ev + break + } + } + if gap == nil { + t.Fatalf("no gap frame:\n%s", rec.BodyString()) + } + if gap.Dropped != 1 { + t.Errorf("gap frame reports %d dropped, want 1", gap.Dropped) + } + // The heartbeat itself must keep going — the gap is a signal on a healthy + // connection, not a teardown. + if !strings.Contains(rec.BodyString(), ": ping") { + t.Error("no heartbeat comment in the stream; the ping branch stopped pinging") + } + select { + case <-done: + t.Fatal("the handler exited; a reported gap must not tear down the stream") + default: + } + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("handler did not return after context cancel") + } +} + +// A healthy stream must not gain a gap frame, and the hello frame has to report +// the position or a client cannot establish a baseline to compare against. +func TestSSEHealthyStreamCarriesSeqAndNoGap(t *testing.T) { + setupContributeEnv(t) + s := NewServer(0, slog.Default()) + s.registerContributeRoutes() + hub := s.contributeHub + + // Two events BEFORE anyone connects, so the hello frame has a non-zero + // position to report. + hub.addActivity("early", "joined", "contributor", "claude", "sonnet", "", "") + hub.addActivity("early", "left", "contributor", "claude", "sonnet", "", "") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest(http.MethodGet, "/api/contribute/events", nil).WithContext(ctx) + rec := newSyncRecorder() + done := make(chan struct{}) + go func() { + s.handleContributeEvents(rec, req) + close(done) + }() + waitFor(t, func() bool { return hub.sse.count() == 1 }, "subscriber to register") + + hub.addActivity("live", "picked up", "contributor", "claude", "sonnet", "", "acme/repo#7") + waitFor(t, func() bool { return strings.Contains(rec.BodyString(), "live") }, "the live event to stream") + + frames := sseFrames(t, rec.BodyString()) + if len(frames) < 2 { + t.Fatalf("want at least a hello and an activity frame, got %d:\n%s", len(frames), rec.BodyString()) + } + hello, activity := frames[0], frames[1] + if hello.Type != "hello" { + t.Fatalf("first frame is %q, want hello", hello.Type) + } + if hello.Seq != 2 { + t.Errorf("hello reports seq %d, want 2 (the position when this client joined)", hello.Seq) + } + if activity.Type != "activity" || activity.Seq != hello.Seq+1 { + t.Errorf("first activity frame is %q at seq %d, want activity at %d — a healthy stream must be contiguous", + activity.Type, activity.Seq, hello.Seq+1) + } + for _, f := range frames { + if f.Type == "gap" { + t.Errorf("a healthy stream emitted a gap frame: %+v", f) + } + } + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("handler did not return after context cancel") + } +}