fix(mcpx): make events replay lossless past the subscriber buffer (port of dezren39/nix#193)

pkgs/mcpx/internal/events/events.go

@@ -207,28 +207,44 @@ func (b *Bus) Subscribe(f Filter, since uint64) (sub *Subscription, gap bool) {
 // knows it missed something and can resynchronise instead of trusting a
 // stream with a hole in it.
 func (b *Bus) SubscribeFrom(f Filter, since uint64, replay bool) (sub *Subscription, gap bool) {
-	ch := make(chan Event, defaults.EventSubscriberBuf)
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	sub = &Subscription{C: ch, ch: ch, filter: f, id: b.next, bus: b}
-	b.next++
-
+	// Collect the replay first, under the lock, so its size is known
+	// before the channel exists.
+	var replayEvents []Event
 	if replay {
 		h := b.retained()
 		if since > 0 && len(h) > 0 && h[0].Seq > since+1 {
 			gap = true
 		}
 		for _, e := range h {
 			if e.Seq > since && f.Matches(e) {
-				select {
-				case ch <- e:
-				default:
-					sub.Dropped.Add(1)
-				}
+				replayEvents = append(replayEvents, e)
 			}
 		}
 	}
+
+	// The replay is owed, not live lag: size the channel for the whole
+	// replay plus the usual live buffer. The old code replayed into the
+	// live buffer with non-blocking sends while holding the lock -- before
+	// the subscriber could read anything -- so a replay larger than
+	// EventSubscriberBuf silently dropped everything past the buffer and
+	// reported no gap (#193). The replay is bounded by the retained
+	// history, so the extra capacity is bounded too.
+	ch := make(chan Event, defaults.EventSubscriberBuf+len(replayEvents))
+	sub = &Subscription{C: ch, ch: ch, filter: f, id: b.next, bus: b}
+	b.next++
+	for _, e := range replayEvents {
+		select {
+		case ch <- e:
+		default:
+			// Unreachable while the channel is sized for the replay, but
+			// if it ever happens the loss must be visible, not silent.
+			sub.Dropped.Add(1)
+			gap = true
+		}
+	}
 	b.subs[sub.id] = sub
 	return sub, gap
 }

pkgs/mcpx/internal/events/events_test.go

@@ -84,6 +84,31 @@ func TestAReconnectingSubscriberMissesNothing(t *testing.T) {
 	}
 }
 
+func TestAReplayLargerThanTheSubscriberBufferMissesNothing(t *testing.T) {
+	// Regression for #193: the retained history (1024 by default) is
+	// larger than the subscriber buffer (256), and the replay used to fill
+	// that buffer before the subscriber could read anything, silently
+	// dropping the rest with no gap reported.
+	b := events.New(0)
+	const n = 300 // > EventSubscriberBuf (256), < EventHistory (1024)
+	for i := 0; i < n; i++ {
+		b.Publish(events.Event{Kind: events.CallFinished})
+	}
+	s, gap := b.SubscribeFrom(events.Filter{}, 0, true)
+	defer s.Close()
+	if gap {
+		t.Error("everything asked for is still retained, so there should be no gap")
+	}
+	for want := uint64(1); want <= n; want++ {
+		if e := recv(t, s); e.Seq != want {
+			t.Fatalf("replay out of order or incomplete: got seq %d, want %d", e.Seq, want)
+		}
+	}
+	if d := s.Dropped.Load(); d != 0 {
+		t.Errorf("replay dropped %d events before the subscriber could read", d)
+	}
+}
+
 func TestAGapBeyondHistoryIsReportedRatherThanHidden(t *testing.T) {
 	// A subscriber gone longer than the history covers is told so, so it can
 	// resynchronise rather than trust a stream with a hole in it.