amadan
hub: fanout tests, prove allowlist, envelope strip, no replay
Fix round 1 from review: 1. The type-guard check now loops offer/roster/lost/join/"" through Fanout, each wanting ErrNotFanout, so a denylist of the targeted types (which would wrongly accept roster/lost/join/"") cannot pass. 2. The delivery test now builds the input with every field a forged or stale client could set (From, Name, To, Seq, Gen, Reason, Roster, Boot, Peer, Waiting) and asserts none of it survives into the delivered envelope, so a pass-through implementation that merely patches From/Name onto the input cannot pass. 3. New TestFanoutIsNeverReplayedOnResume: a chat sent to a peer whose stream is Detached must still not be ringed, and a subsequent Resume must report neither Replay nor Gap. The old ring check only ever ran against an attached recipient, so an implementation that rings a message specifically when the target is detached (arguably 'so it can be delivered later') passed the old suite; verified by mutation that this new test catches it while the old delivery test still passes unchanged. Ring reads now go through a new ringLen(h, sqid, session, peer) helper that takes h.mu itself, rather than through the *Participation pointer h.Find returns, matching the hub's own lock discipline. Each of the three fixes was checked by applying its named mutation to a throwaway copy of the repo under /home/paulca/.cache/tmp/claude-1001 and confirming the affected test fails while the others do not.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
1 file changed,
+88
−11
internal/meet/hub/fanout_test.go+88 −11
CI passed — run details
diff --git a/internal/meet/hub/fanout_test.go b/internal/meet/hub/fanout_test.go| index 5c34de4..3dc397e 100644 |
| --- a/internal/meet/hub/fanout_test.go |
| +++ b/internal/meet/hub/fanout_test.go |
| @@ -10,6 +10,19 @@ import ( |
| "time" |
| ) |
| +// ringLen reads a participation's ring length under the hub's own lock, |
| +// the way every other reader of ring must (hub.go's lock discipline): |
| +// never through a pointer a caller holds unsynchronized. |
| +func ringLen(h *Hub, sqid, session string, peer PeerID) int { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + p, ok := h.findLocked(sqid, session, peer) |
| + if !ok { |
| + return -1 |
| + } |
| + return len(p.ring) |
| +} |
| + |
| func TestFanoutReachesEveryOtherAdmittedPeerAndNobodyElse(t *testing.T) { |
| now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| h := testHub(&now) |
| @@ -24,13 +37,38 @@ func TestFanoutReachesEveryOtherAdmittedPeerAndNobodyElse(t *testing.T) { |
| drain(b.Sub) |
| drain(g.Sub) |
| - err := h.Fanout(ctx, "r1", "sa", a.Peer, Message{Type: TypeChat, Payload: json.RawMessage(`{"text":"hi"}`)}) |
| - if err != nil { |
| + // Every field a forged or stale client could smuggle in: the |
| + // envelope the hub rebuilds must carry none of it through. |
| + in := Message{ |
| + Type: TypeChat, |
| + From: "forged", |
| + Name: "Admin", |
| + To: b.Peer, |
| + Seq: 99, |
| + Gen: 42, |
| + Payload: json.RawMessage(`{"text":"hi"}`), |
| + Reason: ReasonEnded, |
| + Roster: &Roster{}, |
| + Boot: "x", |
| + Peer: "y", |
| + Waiting: true, |
| + } |
| + if err := h.Fanout(ctx, "r1", "sa", a.Peer, in); err != nil { |
| t.Fatal(err) |
| } |
| got := drain(b.Sub) |
| - if len(got) != 1 || got[0].Type != TypeChat || got[0].From != a.Peer || got[0].Name != "Ann" || string(got[0].Payload) != `{"text":"hi"}` || got[0].To != "" || got[0].Seq != 0 || got[0].Gen != 0 { |
| - t.Fatalf("Bob got %+v", got) |
| + if len(got) != 1 { |
| + t.Fatalf("Bob got %d events, want 1: %+v", len(got), got) |
| + } |
| + gm := got[0] |
| + if gm.Type != TypeChat || gm.From != a.Peer || gm.Name != "Ann" || string(gm.Payload) != `{"text":"hi"}` { |
| + t.Fatalf("Bob got %+v, want chat from Ann with the same payload", gm) |
| + } |
| + if gm.To != "" || gm.Seq != 0 || gm.Gen != 0 { |
| + t.Fatalf("Bob got %+v, want no To/Seq/Gen", gm) |
| + } |
| + if gm.Reason != "" || gm.Roster != nil || gm.Boot != "" || gm.Peer != "" || gm.Waiting { |
| + t.Fatalf("Bob got %+v, want none of the input's other fields carried through", gm) |
| } |
| if n := len(drain(a.Sub)); n != 0 { |
| t.Fatalf("the sender must not receive its own chat, got %d", n) |
| @@ -38,15 +76,48 @@ func TestFanoutReachesEveryOtherAdmittedPeerAndNobodyElse(t *testing.T) { |
| if n := len(drain(g.Sub)); n != 0 { |
| t.Fatalf("a waiting guest must receive nothing, got %d", n) |
| } |
| - bp, ok := h.Find("r1", "sb", b.Peer) |
| - if !ok { |
| - t.Fatal("Bob's participation should still exist") |
| - } |
| - if n := len(bp.ring); n != 0 { |
| + if n := ringLen(h, "r1", "sb", b.Peer); n != 0 { |
| t.Fatalf("chat is never ringed, ring has %d", n) |
| } |
| } |
| +// TestFanoutIsNeverReplayedOnResume is §6's "gone with the tab" through |
| +// the resume path, not just through a direct ring read: a chat sent |
| +// while the recipient is detached must still be neither ringed nor |
| +// replayed when that recipient resumes, and must not manufacture a gap |
| +// either (a gap means the ring evicted something above the resuming |
| +// peer's last-seen, and nothing was ever added to evict). |
| +func TestFanoutIsNeverReplayedOnResume(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + a, _ := h.Open("r1", "sa", 1, "Ann", KindMember) |
| + h.Join(ctx, "r1", "sa", a.Peer) |
| + b, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| + h.Join(ctx, "r1", "sb", b.Peer) |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + |
| + h.Detach("r1", b.Peer, b.Sub) |
| + msg := Message{Type: TypeChat, Payload: json.RawMessage(`{"text":"hi"}`)} |
| + if err := h.Fanout(ctx, "r1", "sa", a.Peer, msg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := ringLen(h, "r1", "sb", b.Peer); n != 0 { |
| + t.Fatalf("chat sent to a detached peer must still not be ringed, ring has %d", n) |
| + } |
| + res, err := h.Resume(ctx, "r1", "sb", b.Peer, map[PeerID]uint64{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(res.Replay) != 0 { |
| + t.Fatalf("chat must never be replayed on resume, got %d", len(res.Replay)) |
| + } |
| + if len(res.Gap) != 0 { |
| + t.Fatalf("nothing was ringed, so resume must report no gap either, got %d", len(res.Gap)) |
| + } |
| +} |
| + |
| func TestFanoutNeedsAnAdmittedSenderBoundToThisSession(t *testing.T) { |
| now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| h := testHub(&now) |
| @@ -68,8 +139,14 @@ func TestFanoutNeedsAnAdmittedSenderBoundToThisSession(t *testing.T) { |
| if err := h.Fanout(ctx, "r1", "sg", g.Peer, msg); !errors.Is(err, ErrNotAdmitted) { |
| t.Fatalf("waiting guest: %v", err) |
| } |
| - if err := h.Fanout(ctx, "r1", "sa", a.Peer, Message{Type: TypeOffer}); !errors.Is(err, ErrNotFanout) { |
| - t.Fatalf("a targeted type is not a fan-out: %v", err) |
| + // The guard is an allowlist of {chat, share}, not a denylist of the |
| + // targeted types: a server-raised type and the empty type must be |
| + // refused too, or a denylist that merely excludes offer/answer/ |
| + // candidate would pass this check by accident. |
| + for _, typ := range []string{TypeOffer, TypeRoster, TypeLost, TypeJoin, ""} { |
| + if err := h.Fanout(ctx, "r1", "sa", a.Peer, Message{Type: typ}); !errors.Is(err, ErrNotFanout) { |
| + t.Fatalf("type %q is not a fan-out: %v", typ, err) |
| + } |
| } |
| } |