amadan

oficina / meet Public

hub: Fanout, an untargeted message to every other admitted peer

The test file's proposed events()/peerOf() helpers do not exist in the
package; used the existing drain() and a direct h.Find() lookup instead
(the brief names this as the fallback when peerOf is absent). The third
test adds a second admitted member (Carol) beyond the authorize_test.go
fixture, as a witness that Bob's post-departure chat actually reaches
somebody once Ann is gone.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
Paul Campbell pushed by paul@keymail.dev c52eba768efd3d46f6b5692fd81fd9ea68414dfa parent 9c406f5
2 files changed, +193 −0
  • internal/meet/hub/fanout.go +45 −0
  • internal/meet/hub/fanout_test.go +148 −0

CI passed — run details

diff --git a/internal/meet/hub/fanout.go b/internal/meet/hub/fanout.go
new file mode 100644
index 0000000..faa3c7c
--- /dev/null
+++ b/internal/meet/hub/fanout.go
@@ -0,0 +1,45 @@
+package hub
+
+import (
+ "context"
+ "errors"
+)
+
+// ErrNotFanout is a Fanout of a type that is targeted or server-raised.
+var ErrNotFanout = errors.New("hub: not a fan-out type")
+
+// Fanout is §4.3's chat and share: an admitted peer to every other
+// admitted peer, the envelope rebuilt from the sender's participation so
+// from and name can never be forged, nothing ringed or stored (§6).
+// The sender's authority is §2.5's last row, exactly as Relay's: a
+// member-kind participation re-derives standing first (§15.12), and one
+// whose standing is gone is ended here with lost{standing} and answered
+// ErrNotYours; a guest never consults standing at all (ruling 14).
+func (h *Hub) Fanout(ctx context.Context, sqid, session string, from PeerID, m Message) error {
+ if m.Type != TypeChat && m.Type != TypeShare {
+ return ErrNotFanout
+ }
+ gone, err := h.memberStandingGone(ctx, sqid, session, from, true)
+ if err != nil {
+ return err
+ }
+ if gone {
+ return ErrNotYours
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ r, ok := h.rooms[sqid]
+ if !ok {
+ return ErrUnknownPeer
+ }
+ p, found := h.findLocked(sqid, session, from)
+ if !found {
+ return ErrUnknownPeer
+ }
+ if !p.Admitted {
+ return ErrNotAdmitted
+ }
+ out := Message{Type: m.Type, From: p.Peer, Name: p.Name, Payload: m.Payload}
+ h.broadcastLocked(r, out, func(q *Participation) bool { return q.Admitted && q.Peer != p.Peer })
+ return nil
+}
diff --git a/internal/meet/hub/fanout_test.go b/internal/meet/hub/fanout_test.go
new file mode 100644
index 0000000..5c34de4
--- /dev/null
+++ b/internal/meet/hub/fanout_test.go
@@ -0,0 +1,148 @@
+package hub
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "log/slog"
+ "testing"
+ "time"
+)
+
+func TestFanoutReachesEveryOtherAdmittedPeerAndNobodyElse(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)
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer) // waiting
+ drain(a.Sub)
+ 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 {
+ 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 n := len(drain(a.Sub)); n != 0 {
+ t.Fatalf("the sender must not receive its own chat, got %d", n)
+ }
+ 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 {
+ t.Fatalf("chat is never ringed, ring has %d", n)
+ }
+}
+
+func TestFanoutNeedsAnAdmittedSenderBoundToThisSession(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)
+ msg := Message{Type: TypeChat, Payload: json.RawMessage(`{"text":"hi"}`)}
+ if err := h.Fanout(ctx, "r1", "sa", a.Peer, msg); !errors.Is(err, ErrNotAdmitted) {
+ t.Fatalf("unjoined: %v", err)
+ }
+ h.Join(ctx, "r1", "sa", a.Peer)
+ if err := h.Fanout(ctx, "r1", "sb", a.Peer, msg); !errors.Is(err, ErrUnknownPeer) {
+ t.Fatalf("another session: %v", err)
+ }
+ if err := h.Fanout(ctx, "r2", "sa", a.Peer, msg); !errors.Is(err, ErrUnknownPeer) {
+ t.Fatalf("another room: %v", err)
+ }
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ 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)
+ }
+}
+
+// TestFanoutDropsAMemberWhoseStandingIsGoneAndKeepsAnAdmittedGuest mirrors
+// authorize_test.go's TestAuthorizeDropsAMemberWhoseStandingIsGoneAndKeepsAnAdmittedGuest:
+// the same mutable-standing fixture, moved without a tick running, so the
+// only thing that could refuse Ann is Fanout's own re-derivation (§15.12).
+// Carol, a second admitted member, is the witness that Bob's later chat
+// actually reaches somebody once Ann is gone; Bob himself, an admitted
+// guest whose grant has since "expired", never even asks standing
+// (ruling 14) and fans out anyway.
+func TestFanoutDropsAMemberWhoseStandingIsGoneAndKeepsAnAdmittedGuest(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ standing := map[int64]bool{1: true, 3: true, 100: true}
+ guestStandingCalls := 0
+ h := New(Config{
+ Now: func() time.Time { return now },
+ Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
+ StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
+ if id == 100 {
+ guestStandingCalls++
+ return KindGuest, standing[id], nil
+ }
+ return KindMember, standing[id], nil
+ },
+ })
+ ctx := context.Background()
+ a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
+ if err := h.Join(ctx, "r1", "sa", a.Peer); err != nil {
+ t.Fatal(err)
+ }
+ c, _ := h.Open("r1", "sc", 3, "Carol", KindMember)
+ if err := h.Join(ctx, "r1", "sc", c.Peer); err != nil {
+ t.Fatal(err)
+ }
+ b, _ := h.Open("r1", "sb", 100, "Bob", KindGuest)
+ if err := h.Join(ctx, "r1", "sb", b.Peer); err != nil {
+ t.Fatal(err)
+ }
+ if err := h.Admit(ctx, "r1", "sa", a.Peer, b.Peer); err != nil {
+ t.Fatal(err)
+ }
+ drain(a.Sub)
+ drain(b.Sub)
+ drain(c.Sub)
+
+ standing[1] = false // Ann taken off the team, no tick has run
+ msg := Message{Type: TypeChat, Payload: json.RawMessage(`{"text":"hi"}`)}
+ if err := h.Fanout(ctx, "r1", "sa", a.Peer, msg); !errors.Is(err, ErrNotYours) {
+ t.Fatalf("member whose standing is gone: want ErrNotYours, got %v", err)
+ }
+ if _, found := h.Find("r1", "sa", a.Peer); found {
+ t.Fatal("Ann's participation should have ended")
+ }
+ bev := drain(b.Sub)
+ if len(bev) != 1 || bev[0].Type != TypeGone || bev[0].From != a.Peer {
+ t.Fatalf("Bob's stream got %v, want gone naming Ann and no chat", types(bev))
+ }
+ drain(c.Sub) // Carol also saw the departure; not this test's point
+
+ standing[100] = false // Bob's grant has "expired"
+ before := guestStandingCalls
+ if err := h.Fanout(ctx, "r1", "sb", b.Peer, msg); err != nil {
+ t.Fatalf("admitted guest with a lapsed grant: %v", err)
+ }
+ if guestStandingCalls != before {
+ t.Fatalf("Fanout asked StandingFor for a guest: %d calls during this Fanout, want 0", guestStandingCalls-before)
+ }
+ cev := drain(c.Sub)
+ if len(cev) != 1 || cev[0].Type != TypeChat || cev[0].From != b.Peer {
+ t.Fatalf("Carol should have received Bob's chat, got %v", types(cev))
+ }
+ if n := len(drain(b.Sub)); n != 0 {
+ t.Fatalf("Bob must not receive his own chat, got %d", n)
+ }
+}