fix(events): bound the mid-stream signal, and stop it moving existing alerts (BUG-2730, codex round 4)

Three findings from the operator-at-3am angle, all real.

A pub/sub outage on a workspace with a subscriber but NO replay buffer
yet was silent. dropWorkspaceCoverage returned early before telling
anyone, on the reasoning that there was no coverage to end — true of the
BUFFER, and beside the point for the SUBSCRIBER, which has the largest
possible hole and the least evidence of it. Live subscribers are now
signalled on that path while the reset metric stays suppressed: the
metric measures coverage endings, the signal measures clients who may
have missed something, and those are different questions.

The gap channel coalesces, which bounds the queue but not the loop: once
the handler consumes a signal the next drop re-arms it, so a slow client
could be answered with a delta sync, made slower, and answered again.
Both handlers now share a gapAnnouncer that allows one announcement per
connection per 5 seconds — a delta-sync round trip, not a tuning knob —
and LATCHES rather than drops, so a gap inside the window is announced
when the window closes. Suppressing it would be this fix's own defect
one layer up.

Folding mid-stream signals into pad_*_resume_gaps_total silently changed
what every existing alert on those counters measures, and a mixed-version
fleet would have reported two populations under one name for the length
of a rollout. They go back to counting resumes; the new population gets
pad_event_midstream_resyncs_total and pad_watchevents_midstream_resyncs_total,
which count CLIENTS TOLD rather than causes — one instance-wide coverage
loss moves them once per subscriber while the reset counter moves once,
and that ratio is the fan-out an operator wants when judging a storm.

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 |

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)
+	}
+}

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()

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)

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,

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.

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

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 {

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.
 //

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
+}

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")
+	}
+}