amadan
hub: terminate keeps queued events unless the buffer is full
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
2 files changed,
+29
−3
internal/meet/hub/hub_test.go+22 −0internal/meet/hub/subscriber.go+7 −3
diff --git a/internal/meet/hub/hub_test.go b/internal/meet/hub/hub_test.go| index a9ec43b..7706991 100644 |
| --- a/internal/meet/hub/hub_test.go |
| +++ b/internal/meet/hub/hub_test.go |
| @@ -120,3 +120,25 @@ func TestSlowSubscriberIsDeclaredLost(t *testing.T) { |
| t.Fatalf("last event %+v, want lost/buffer", last) |
| } |
| } |
| + |
| +func TestTerminateDoesNotEvictWhenNotFull(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember) |
| + h.deliver("r1", a.Peer, Message{Type: TypeGone, From: "someone"}) |
| + a.Sub.terminate(Message{Type: TypeLost, Reason: ReasonRemoved}) |
| + |
| + var got []Message |
| + for m := range a.Sub.C() { |
| + got = append(got, m) |
| + } |
| + if len(got) != 2 { |
| + t.Fatalf("got %d events %+v, want the queued one and the lost", len(got), got) |
| + } |
| + if got[0].Type != TypeGone || got[0].From != "someone" { |
| + t.Fatalf("first event %+v, want the queued gone", got[0]) |
| + } |
| + if got[1].Type != TypeLost || got[1].Reason != ReasonRemoved { |
| + t.Fatalf("second event %+v, want lost/removed", got[1]) |
| + } |
| +} |
diff --git a/internal/meet/hub/subscriber.go b/internal/meet/hub/subscriber.go| index d2feccf..97e8f2c 100644 |
| --- a/internal/meet/hub/subscriber.go |
| +++ b/internal/meet/hub/subscriber.go |
| @@ -37,12 +37,16 @@ func (s *Sub) terminate(final Message) { |
| return |
| } |
| close(s.done) |
| - // Make room for the final event, then close so the range ends. |
| + // The final event must land, but nothing already queued may be |
| + // dropped to make room for it: a peer ended with a real reason |
| + // still gets the events it had not read yet. Only a full buffer |
| + // costs one, and by then the subscriber is lost anyway. |
| select { |
| - case <-s.ch: |
| + case s.ch <- final: |
| default: |
| + <-s.ch // reached only when full; no other sender holds h.mu |
| + s.ch <- final |
| } |
| - s.ch <- final |
| close(s.ch) |
| } |