diff --git a/docs/deployment.md b/docs/deployment.md index 74fc5cac..540f94e8 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -235,7 +235,15 @@ an empty `id:` field, exactly as the resume case does, so a client stops resending a position the server has just disclaimed. There is no separate event name for the mid-stream case, deliberately: every -client acts on the two identically. +client acts on the two identically. There IS a separate metric — see +`pad_event_midstream_resyncs_total` below — so the two populations stay +distinguishable to an operator without changing what any existing alert means. + +**A connection is told at most once every 5 seconds**, and nothing is lost to +that bound: a gap arriving inside the window is remembered and announced when +the window closes. The bound exists because the subscriber most likely to be +signalled is a slow one, and answering "you could not keep up" with "now fetch +a delta" can feed back into more drops. #### Redis health and metrics @@ -256,12 +264,14 @@ Alert on these instead: | `pad_redis_up` | `0` when the last probe (every 15s) failed. Exported only when Redis is configured — absence means "no Redis", not "down" | | `pad_stream_connections_active` | Held streaming connections on this instance, across both SSE endpoints — the population the limits bound | | `pad_watchevents_sequence_gaps_total` | This instance missed notifications — a delivery fault | -| `pad_watchevents_resume_gaps_total` | `sync_required` signals sent on the watch stream — from a hole, a cold start, an epoch change, or a shared-counter disagreement. Counts SIGNALS, not resumes: since BUG-2730 it also covers a subscriber told MID-STREAM that it missed notifications, which is not a resume at all. The user-visible one | +| `pad_watchevents_resume_gaps_total` | Resumes this instance could not serve — from a hole, a cold start, an epoch change, or a shared-counter disagreement. Each sends a client `sync_required`. RESUME-TIME ONLY; a subscriber told mid-stream is counted separately, so an alert on this keeps the meaning it had | +| `pad_watchevents_midstream_resyncs_total` | Watch-stream subscribers told MID-STREAM that they missed notifications, on a connection that stayed open. New in BUG-2730 | | `pad_watchevents_notifications_missed_total` | How many notifications those gaps spanned | | `pad_watchevents_notifications_dropped_total` | Received but not delivered to a local subscriber — that connection's buffer was full. Since BUG-2730 that subscriber is told (`sync_required`, mid-stream) rather than silently under-served, so a rise here produces a rise in `pad_watchevents_resume_gaps_total`, one client at a time | | `pad_watchevents_sequence_resets_total` | The Redis counter or epoch changed; replay buffers dropped | | `pad_watchevents_receive_loop_exits_total` | Non-zero outside shutdown means an instance publishes but receives nothing | -| `pad_event_resume_gaps_total` | The ACTIVITY stream's (`/api/v1/events`) twin of the watch counter above. **Expect a step around a deploy, with the RATE settling back to baseline** (the counter itself only ever increases) — each instance starts with no replay coverage, so an early resume against a workspace it has not seen yet is a warranted resync. It counts SIGNALS, not clients: a client that reconnects several times is counted several times. Since BUG-2730 the population is wider than resumes — a live subscriber told mid-stream that it has a hole is counted here too, so a deploy with no reconnects at all can still move it | +| `pad_event_resume_gaps_total` | The ACTIVITY stream's (`/api/v1/events`) twin of the watch resume counter above. **Expect a step around a deploy, with the RATE settling back to baseline** (the counter itself only ever increases) — each instance starts with no replay coverage, so an early resume against a workspace it has not seen yet is a warranted resync. It counts RESUMES, not clients: a deploy with no reconnects does not move it at all, and a client that reconnects several times is counted several times. A rate that does not settle is the thing to alert on | +| `pad_event_midstream_resyncs_total` | Activity-stream subscribers told MID-STREAM that they missed events, on a connection that stayed open. New in BUG-2730, and the counter to watch when judging whether that fix is costing more resyncs than it is worth. It counts CLIENTS TOLD, not causes: one instance-wide coverage loss moves this once per live subscriber while `pad_event_sequence_resets_total` moves once, and the ratio is the fan-out | | `pad_event_sequence_resets_total` | Activity replay coverage dropped, by reason. `subscription_resumed` — a pub/sub connection dropped and resubscribed, dropping that workspace's buffer; expect it during a Redis failover and expect it to stop afterwards. `epoch_change` — the shared counter's ID space changed generation, dropping every buffer; expect a handful per cutover. `counter_backward` — an ID arrived at or below a buffer's high-water mark with no generation change; see *Event ID-space migration* for what to expect per phase. `epoch_regressed` — a LOWER generation was seen, so this instance stopped vouching for its buffers. One alongside an `epoch_change` is a message that was in flight when the generation rotated; a RUN of them means the counter itself went backwards, i.e. Redis lost writes. `undecodable_message` — a message on these channels could not be parsed, so that workspace's coverage ended; expect zero, and suspect a namespace collision | | `pad_event_events_dropped_total` | Activity events not delivered to a live subscriber, by reason — today only `slow_subscriber` (that connection's 64-deep channel was full). Per-SUBSCRIBER: every subscriber that was keeping up received the event. New in BUG-2730, along with the fix that stops the drop being silent, so a deploy that starts reporting these is not necessarily a regression — it may be the first time they were countable | | `pad_event_receive_loop_exits_total` | A workspace's activity subscription loop stopped. Unlike the watch stream's twin this does **not** stay at zero — it is expected at shutdown and whenever a workspace's last local subscriber leaves. Read it as a rate against a stable subscriber count | diff --git a/internal/events/gap_signal_test.go b/internal/events/gap_signal_test.go index d7424ef3..79c826bb 100644 --- a/internal/events/gap_signal_test.go +++ b/internal/events/gap_signal_test.go @@ -318,3 +318,34 @@ func TestIDSpaceResetSignalsEveryWorkspace(t *testing.T) { t.Error("an ID-space reset invalidates every buffer, so every subscriber must be told") } } + +// TestCoverageDropWithNoBufferStillSignals is codex round 4's P1. A subscriber +// that connected and then sat through a pub/sub outage before ANY event was +// received for its workspace has the largest possible hole and the least +// evidence of it — and the early return that (correctly) suppresses the reset +// metric was returning before the subscribers were told. +// +// The control is the metric: the reset must stay unreported, because there was +// no coverage to end. Without that leg the obvious "fix" of deleting the early +// return would pass. +func TestCoverageDropWithNoBufferStillSignals(t *testing.T) { + b := newTestRedisBus(t) + obs := &recordingObserver{} + b.SetObserver(obs) + + ch, gaps, ok := b.SubscribeIfAllowed("ws-1", 0) + if !ok { + t.Fatal("subscribe refused") + } + defer b.Unsubscribe(ch) + + // Deliberately no fanOutLocally: this workspace has no replay buffer. + b.dropWorkspaceCoverage("ws-1", ResetReasonSubscriptionResumed, b.currentSubGen("ws-1")) + + if !raised(gaps) { + t.Error("a subscriber that sat through an outage before any event arrived was not told") + } + if got := obs.resetReasons(); len(got) != 0 { + t.Errorf("a reset was reported for a workspace with no coverage to end: %v", got) + } +} diff --git a/internal/events/observer_test.go b/internal/events/observer_test.go index 0d9c7ee7..3dac498e 100644 --- a/internal/events/observer_test.go +++ b/internal/events/observer_test.go @@ -37,6 +37,12 @@ func (o *recordingObserver) EventDropped(reason string) { o.drops = append(o.drops, reason) } +func (o *recordingObserver) resetReasons() []string { + o.mu.Lock() + defer o.mu.Unlock() + return append([]string(nil), o.resets...) +} + func (o *recordingObserver) dropped() []string { o.mu.Lock() defer o.mu.Unlock() diff --git a/internal/events/redis_bus.go b/internal/events/redis_bus.go index 95bec10b..008786ed 100644 --- a/internal/events/redis_bus.go +++ b/internal/events/redis_bus.go @@ -895,9 +895,24 @@ func (b *RedisBus) dropWorkspaceCoverage(workspaceID, reason string, gen int64) } if _, ok := b.replayBuffers[workspaceID]; !ok { - // Nothing buffered: there is no coverage to end and no client that - // could have been told it was current. Reporting here would give the - // counter a baseline on every reconnect of an idle workspace. + // Nothing buffered: there is no coverage to END, so no buffer to drop + // and nothing to report. Reporting here would give the reset counter + // a baseline on every reconnect of an idle workspace. + // + // THE LIVE SUBSCRIBERS STILL GET TOLD, and that is not a contradiction + // (codex round 4). "No buffer" says nothing was RECEIVED for this + // workspace on this instance; it does not say nothing was PUBLISHED. + // A subscriber that connected and then sat through a pub/sub outage + // has exactly the hole this signal exists for, and it is the case + // where the instance has the least idea what it missed. Answering + // only when a buffer happens to exist would make the honesty + // conditional on having already received something, which is + // backwards. + // + // The asymmetry is deliberate: the metric measures COVERAGE ENDINGS + // (there was none) and the signal measures CLIENTS WHO MAY HAVE + // MISSED SOMETHING (there are some). + b.signalWorkspaceLocked(workspaceID) return } delete(b.replayBuffers, workspaceID) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index b7ca262d..105758a7 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -136,6 +136,23 @@ type Metrics struct { // there one client at a time, and that is the fix working, not a fault. EventEventsDroppedTotal *prometheus.CounterVec + // EventMidstreamResyncsTotal and WatchMidstreamResyncsTotal count clients + // told MID-STREAM that they have a hole (BUG-2730) — a signal that did not + // exist before, on connections that stay open. + // + // SEPARATE FROM THE RESUME COUNTERS ON PURPOSE. Folding them in would have + // silently changed what every existing alert on + // pad_*_resume_gaps_total measures, and mixed-version fleets would report + // two different populations under one name during a rollout. + // + // Counts CLIENTS TOLD, not causes: one instance-wide coverage loss moves + // this once per live subscriber while pad_*_sequence_resets_total moves + // once. The ratio between them is the fan-out, which is the number an + // operator actually wants when deciding whether a resync storm is + // underway. + EventMidstreamResyncsTotal prometheus.Counter + WatchMidstreamResyncsTotal prometheus.Counter + // EventSequenceResetsTotal counts activity-stream coverage resets by // reason. Three reasons: // @@ -388,7 +405,7 @@ func New() *Metrics { watchResumeGapsTotal := prometheus.NewCounter(prometheus.CounterOpts{ Name: "pad_watchevents_resume_gaps_total", - Help: "sync_required signals sent on the watch stream: a resume this instance could not serve, an unreadable cursor, or (since BUG-2730) a live subscriber told mid-stream that it missed notifications.", + Help: "Resumes this instance could not serve, each sending a client sync_required. Resume-time only — a live subscriber told mid-stream is counted by pad_watchevents_midstream_resyncs_total instead, so existing alerts on this counter keep their meaning.", }) watchSequenceResetsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ @@ -403,7 +420,17 @@ func New() *Metrics { eventResumeGapsTotal := prometheus.NewCounter(prometheus.CounterOpts{ Name: "pad_event_resume_gaps_total", - Help: "sync_required signals sent on the activity stream: a resume this instance could not serve, an unreadable cursor, or (since BUG-2730) a live subscriber told mid-stream that it missed events. Counts signals, not clients. Expect a step around a deploy (cold buffers) returning to baseline; a rate that does not settle is the signal.", + Help: "Activity-stream resumes this instance could not serve, each sending a client sync_required. Counts resumes, not clients. Resume-time only — a live subscriber told mid-stream is counted by pad_event_midstream_resyncs_total instead, so existing alerts on this counter keep their meaning. Expect a step around a deploy (cold buffers) returning to baseline; a rate that does not settle is the signal.", + }) + + eventMidstreamResyncsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "pad_event_midstream_resyncs_total", + Help: "Activity-stream subscribers told MID-STREAM that they missed events, on a connection that stayed open (BUG-2730). Counts clients told, not causes — one instance-wide coverage loss tells every subscriber and increments this once per subscriber, while pad_event_sequence_resets_total counts it once. New in BUG-2730; a fresh counter rather than folding into pad_event_resume_gaps_total, so alerts on that one keep their meaning.", + }) + + watchMidstreamResyncsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "pad_watchevents_midstream_resyncs_total", + Help: "Watch-stream subscribers told MID-STREAM that they missed notifications, on a connection that stayed open (BUG-2730). Counts clients told, not causes. See pad_event_midstream_resyncs_total.", }) eventEventsDroppedTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ @@ -435,6 +462,8 @@ func New() *Metrics { watchReceiveLoopExitsTotal, eventResumeGapsTotal, eventEventsDroppedTotal, + eventMidstreamResyncsTotal, + watchMidstreamResyncsTotal, eventSequenceResetsTotal, eventReceiveLoopExitsTotal, sessionPresenceFailuresTotal, @@ -465,6 +494,8 @@ func New() *Metrics { WatchSequenceResetsTotal: watchSequenceResetsTotal, EventResumeGapsTotal: eventResumeGapsTotal, EventEventsDroppedTotal: eventEventsDroppedTotal, + EventMidstreamResyncsTotal: eventMidstreamResyncsTotal, + WatchMidstreamResyncsTotal: watchMidstreamResyncsTotal, EventSequenceResetsTotal: eventSequenceResetsTotal, EventReceiveLoopExitsTotal: eventReceiveLoopExitsTotal, WatchReceiveLoopExitsTotal: watchReceiveLoopExitsTotal, diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index 9f450e64..eb666f09 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -378,6 +378,29 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { membershipCheck := time.NewTimer(firstDelay) defer membershipCheck.Stop() + // Bounds how often this connection can be told mid-stream that it has a + // hole — see gapAnnouncer for why a slow subscriber needs bounding and + // why nothing is dropped to achieve it. + gapAnn := newGapAnnouncer(midStreamGapCooldown) + defer gapAnn.stop() + + // announceGap writes the mid-stream signal. Returns false when the write + // failed, which means the client is gone and the handler must return. + announceGap := func() bool { + slog.Info("SSE: live subscriber missed events, sending sync_required mid-stream", + "workspace", ws.Slug, "user_id", sseUserID) + s.countMidStreamResync(true) + if err := writeSSEResetCursorEvent(w, "sync_required", map[string]string{ + "reason": "This stream missed events it cannot replay. Full sync required.", + }); err != nil { + slog.Debug("SSE: mid-stream sync_required write failed, closing", + "workspace", ws.Slug, "error", err) + return false + } + flusher.Flush() + return true + } + ctx := r.Context() for { select { @@ -429,28 +452,24 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { // reason the resume path retires it: the position we would be // confirming is one we have just said we cannot vouch for. // - // LOAD POSTURE, stated because it is the cost side and someone - // will want to see it: for a slow subscriber this answers "you - // could not keep up" with "now also fetch a delta". The gap - // channel coalesces (capacity 1, non-blocking send), so the - // signal's rate is bounded by how fast this loop reads it, never - // by the drop rate, and the web client defers a sync_required - // that arrives mid-sync rather than stacking them (BUG-2508). If - // that still feeds back in production, the escape hatch named on - // BUG-2730 is to replay from the buffer instead of resyncing — - // the dropped events are still in it, since it holds 1024 against - // the channel's 64. - slog.Info("SSE: live subscriber missed events, sending sync_required mid-stream", - "workspace", ws.Slug, "user_id", sseUserID) - s.countResumeGap(true) - if err := writeSSEResetCursorEvent(w, "sync_required", map[string]string{ - "reason": "This stream missed events it cannot replay. Full sync required.", - }); err != nil { - slog.Debug("SSE: mid-stream sync_required write failed, closing", - "workspace", ws.Slug, "error", err) + // Rate-limited per connection, and nothing is lost by it: a gap + // inside the cooldown is latched and announced when the window + // closes. See gapAnnouncer for the feedback loop that makes the + // bound necessary — the escape hatch if it proves insufficient is + // 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 + } + + case <-gapAnn.cool(): + if gapAnn.flush() && !announceGap() { return } - flusher.Flush() case <-keepalive.C: // Send keepalive comment to prevent proxy/LB timeouts. diff --git a/internal/server/handlers_stream_gap_test.go b/internal/server/handlers_stream_gap_test.go index b5668736..e4b815dd 100644 --- a/internal/server/handlers_stream_gap_test.go +++ b/internal/server/handlers_stream_gap_test.go @@ -84,14 +84,19 @@ func TestActivityStreamAnnouncesAGapMidStream(t *testing.T) { frames := readRawSSEFrames(t, ctx, ts.URL+"/api/v1/events?workspace="+slug, "") waitForFrameWithEvent(t, frames, "connected") - before := counterValue(t, m.EventResumeGapsTotal) + resumeBefore := counterValue(t, m.EventResumeGapsTotal) bus.gaps <- struct{}{} frame := waitForFrameWithEvent(t, frames, "sync_required") assertRetiresCursor(t, frame) - if got := counterValue(t, m.EventResumeGapsTotal); got <= before { - t.Errorf("a mid-stream gap did not move the sync_required counter: %v -> %v", before, got) + if got := counterValue(t, m.EventMidstreamResyncsTotal); got != 1 { + t.Errorf("pad_event_midstream_resyncs_total = %v, want 1", got) + } + // The RESUME counter must not move: existing alerts are written against + // it and this was not a resume (codex round 4). + if got := counterValue(t, m.EventResumeGapsTotal); got != resumeBefore { + t.Errorf("a mid-stream gap moved the resume counter: %v -> %v", resumeBefore, got) } // THE STREAM MUST STAY OPEN (codex round 3). sync_required tells the diff --git a/internal/server/handlers_watch_events.go b/internal/server/handlers_watch_events.go index 8e722863..b014939d 100644 --- a/internal/server/handlers_watch_events.go +++ b/internal/server/handlers_watch_events.go @@ -294,6 +294,24 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) revalC = *ch } + // Same per-connection bound as the activity stream — see gapAnnouncer. + gapAnn := newGapAnnouncer(midStreamGapCooldown) + defer gapAnn.stop() + + announceGap := func() bool { + slog.Info("watch-events: live subscriber missed notifications, sending sync_required mid-stream", + "user_id", user.ID) + s.countMidStreamResync(false) + if err := writeSSEResetCursorEvent(w, "sync_required", map[string]string{ + "reason": "This stream missed notifications it cannot replay. Full sync required.", + }); err != nil { + slog.Debug("watch-events: mid-stream sync_required write failed, closing", "user_id", user.ID, "error", err) + return false + } + flusher.Flush() + return true + } + ctx := r.Context() for { select { @@ -334,16 +352,20 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) // The empty `id:` retires the cursor, so a generic SSE client // does not go on resending a position we have just said we // cannot vouch for. - slog.Info("watch-events: live subscriber missed notifications, sending sync_required mid-stream", - "user_id", user.ID) - s.countResumeGap(false) - if err := writeSSEResetCursorEvent(w, "sync_required", map[string]string{ - "reason": "This stream missed notifications it cannot replay. Full sync required.", - }); err != nil { - slog.Debug("watch-events: mid-stream sync_required write failed, closing", "user_id", user.ID, "error", err) + // + // 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 + } + + case <-gapAnn.cool(): + if gapAnn.flush() && !announceGap() { return } - flusher.Flush() case <-keepalive.C: if _, err := fmt.Fprintf(w, ": keepalive\n\n"); err != nil { diff --git a/internal/server/server.go b/internal/server/server.go index e44987e4..ed19fd60 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -962,6 +962,25 @@ func (s *Server) countResumeGap(activity bool) { s.metrics.WatchResumeGapsTotal.Inc() } +// countMidStreamResync records a client told MID-STREAM that it missed +// events, on a connection that stayed open (BUG-2730). +// +// Deliberately NOT countResumeGap: that counter's population is resumes this +// instance could not serve, and existing alerts are written against it. +// Widening it in place would have changed what those alerts measure without +// changing their name, and a mixed-version fleet would report two populations +// under one metric for the length of a rollout. +func (s *Server) countMidStreamResync(activity bool) { + if s.metrics == nil { + return + } + if activity { + s.metrics.EventMidstreamResyncsTotal.Inc() + return + } + s.metrics.WatchMidstreamResyncsTotal.Inc() +} + // SetMetrics attaches Prometheus metrics to the server. // Must be called before the first request is served. // diff --git a/internal/server/stream_gap_announcer.go b/internal/server/stream_gap_announcer.go new file mode 100644 index 00000000..e6821491 --- /dev/null +++ b/internal/server/stream_gap_announcer.go @@ -0,0 +1,89 @@ +package server + +import "time" + +// midStreamGapCooldown is the minimum interval between two mid-stream +// sync_required signals on ONE connection. +// +// WHY A COOLDOWN AT ALL. The signal's remedy is a delta sync, and the +// subscriber most likely to be signalled is a slow one — so answering "you +// could not keep up" with "now also fetch a delta" can feed back: more sync +// work, slower draining, more drops, more signals. The bus's gap channel +// coalesces (capacity 1), which bounds the QUEUE but not the LOOP: once the +// handler consumes a signal, the next drop re-arms it immediately. +// +// WHY THIS LENGTH. It is a round trip, not a tuning knob pulled out of the +// air: a second signal is only useful once the client has had time to act on +// the first, and telling it again mid-delta is pure waste — the web client +// already defers a sync_required that arrives while one is in flight +// (BUG-2508). Five seconds is comfortably longer than a delta sync against a +// busy workspace and far shorter than the 30s keepalive, so a genuinely new +// hole is never sat on for a noticeable time. +// +// WHY A LATCH RATHER THAN A DROP. Suppressing a gap that arrives inside the +// window would be the same dishonesty this whole fix exists to remove, one +// layer up. A gap inside the window is REMEMBERED and announced when the +// window closes, so the client always learns — at most once per window. +const midStreamGapCooldown = 5 * time.Second + +// gapAnnouncer rate-limits one connection's mid-stream gap announcements +// without losing any. +// +// Used by both SSE handlers, which have separate select loops that would +// otherwise each grow their own copy of this logic — and a rate limiter +// implemented twice is a rate limiter that behaves two ways. +// +// Not safe for concurrent use: it belongs to one handler goroutine. +type gapAnnouncer struct { + cooldown time.Duration + timer *time.Timer + // cooling is the timer's channel while a window is open, and nil + // otherwise. A nil channel blocks forever in a select, which is how the + // case switches itself off without a second flag. + cooling <-chan time.Time + pending bool +} + +func newGapAnnouncer(cooldown time.Duration) *gapAnnouncer { + t := time.NewTimer(cooldown) + if !t.Stop() { + <-t.C + } + return &gapAnnouncer{cooldown: cooldown, timer: t} +} + +// stop releases the timer. Safe to call more than once. +func (g *gapAnnouncer) stop() { g.timer.Stop() } + +// cool is the channel a handler's select must watch alongside the bus's gap +// channel. It fires when a window closes; the handler then calls flush. +func (g *gapAnnouncer) cool() <-chan time.Time { return g.cooling } + +// observe records a gap the bus just reported and says whether to announce it +// NOW. False means the announcement is latched and will come back through +// cool()/flush() when the window closes. +func (g *gapAnnouncer) observe() bool { + if g.cooling != nil { + g.pending = true + return false + } + g.open() + return true +} + +// flush is called when cool() fires. It reports whether a latched gap is owed +// an announcement now, and reopens the window if so. +func (g *gapAnnouncer) flush() bool { + g.cooling = nil + if !g.pending { + return false + } + g.pending = false + g.open() + return true +} + +func (g *gapAnnouncer) open() { + g.timer.Reset(g.cooldown) + g.cooling = g.timer.C +} diff --git a/internal/server/stream_gap_announcer_test.go b/internal/server/stream_gap_announcer_test.go new file mode 100644 index 00000000..48795737 --- /dev/null +++ b/internal/server/stream_gap_announcer_test.go @@ -0,0 +1,80 @@ +package server + +import ( + "testing" + "time" +) + +// The announcer answers codex round 4's P1: a slow subscriber answered with a +// delta sync can be made slower, drop more, and be answered again. The bound +// has to hold WITHOUT losing a gap, because dropping one would be the same +// dishonesty this fix exists to remove, one layer up. +// +// Both halves are asserted here, and each is the other's control: the second +// gap inside the window must NOT announce immediately, and it must NOT be +// forgotten. +func TestGapAnnouncerBoundsTheRateWithoutLosingAGap(t *testing.T) { + g := newGapAnnouncer(40 * time.Millisecond) + defer g.stop() + + if !g.observe() { + t.Fatal("the first gap on a fresh connection must announce immediately") + } + if g.observe() { + t.Fatal("a second gap inside the window announced immediately; the rate is unbounded") + } + if g.observe() { + t.Fatal("a third gap inside the window announced immediately") + } + + select { + case <-g.cool(): + case <-time.After(2 * time.Second): + t.Fatal("the cooldown never fired, so the latched gap would never be announced") + } + + if !g.flush() { + t.Fatal("the gaps that arrived inside the window were forgotten rather than latched") + } + + // Latching reopens the window: the announcement just made is subject to + // the same bound as any other. + if g.observe() { + t.Error("a gap immediately after a latched announcement was not bounded") + } +} + +// The other direction: a window that closes with nothing latched must not +// announce. Without this, the announcer would emit a spurious sync_required +// every cooldown for the life of any connection that ever saw one gap. +func TestGapAnnouncerDoesNotAnnounceAnEmptyWindow(t *testing.T) { + g := newGapAnnouncer(20 * time.Millisecond) + defer g.stop() + + if !g.observe() { + t.Fatal("the first gap must announce") + } + + select { + case <-g.cool(): + case <-time.After(2 * time.Second): + t.Fatal("the cooldown never fired") + } + if g.flush() { + t.Error("a window that closed with no latched gap announced anyway") + } + + // And the case is now off: cool() must not fire again on its own, or the + // handler's select would spin. + select { + case <-g.cool(): + t.Error("the cooldown re-armed itself with nothing to announce") + case <-time.After(60 * time.Millisecond): + } + + // A gap arriving after a quiet window announces immediately — the bound + // applies to bursts, not to a connection that has settled. + if !g.observe() { + t.Error("a gap after a quiet window was bounded; the window had already closed") + } +}