amadan

oficina / meet Public

hub: targeted relay, pair generations, replay ring and resume

Relay, Reset and Resume as the brief lays them out, adapted to the
shape Task 2 left on the branch: sendLocked instead of a hand-rolled
nil check on the target's sub.

One deviation. The brief detects ring eviction from the ring's own
oldest seq per sender, which cannot see a sender whose messages were
evicted wholesale by another sender's burst — one ring is shared by
every sender, and three admitted peers share thirty-two slots. The
ring now remembers the highest seq it has dropped per sender, and a
gap is that mark standing above the browser's last. That rule subsumes
the brief's (a ring whose oldest is above last+1 has dropped last+1)
and adds no false positives: a message the browser did receive is at
or below its own last.
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 8849742bd14bba80904ccb3289f3a8826c40620c parent 69f5aff
3 files changed, +429 −3
  • internal/meet/hub/hub.go +6 −3
  • internal/meet/hub/relay.go +188 −0
  • internal/meet/hub/relay_test.go +235 −0

CI passed — run details

diff --git a/internal/meet/hub/hub.go b/internal/meet/hub/hub.go
index 11f7adb..9749dfb 100644
--- a/internal/meet/hub/hub.go
+++ b/internal/meet/hub/hub.go
@@ -88,7 +88,8 @@ type Participation struct {
resumeUntil time.Time // zero while a stream is attached
seqOut map[PeerID]uint64 // next seq per target (unused by the hub; kept by browsers)
lastSeen map[PeerID]uint64
- ring []Message // targeted messages delivered to this peer, newest last, cap RingSize
+ ring []Message // targeted messages delivered to this peer, newest last, cap RingSize
+ evicted map[PeerID]uint64 // highest seq per sender the ring has dropped
}
type pairKey struct{ a, b PeerID }
@@ -393,8 +394,10 @@ func (h *Hub) broadcastLocked(r *room, m Message, who audience) {
}
}
-// deliver is the one place an event reaches a subscriber; targeted
-// messages are also kept in the recipient's replay ring (Task 3).
+// deliver is one event to one peer found by room and id, for the events
+// the hub raises about the room. A targeted message never comes this
+// way: Relay holds the recipient's participation already, because it
+// must also ring the message (see relay.go).
func (h *Hub) deliver(sqid string, to PeerID, m Message) {
h.mu.Lock()
defer h.mu.Unlock()
diff --git a/internal/meet/hub/relay.go b/internal/meet/hub/relay.go
new file mode 100644
index 0000000..ffcb8f9
--- /dev/null
+++ b/internal/meet/hub/relay.go
@@ -0,0 +1,188 @@
+package hub
+
+import (
+ "fmt"
+ "time"
+)
+
+func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t == TypeCandidate }
+
+// Relay delivers one targeted message. The envelope is rebuilt from the
+// binding; the payload is never read.
+func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error {
+ if !isTargeted(m.Type) {
+ return fmt.Errorf("hub: %q is not a targeted type", m.Type)
+ }
+ 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
+ }
+ tp, ok := r.peers[m.To]
+ if !ok || !tp.Admitted {
+ return ErrUnknownPeer
+ }
+ if m.Gen != h.genLocked(r, p.Peer, tp.Peer) {
+ return ErrStaleGen
+ }
+ out := Message{Type: m.Type, From: p.Peer, Name: p.Name, To: tp.Peer, Seq: m.Seq, Gen: m.Gen, Payload: m.Payload}
+ h.ringLocked(tp, out)
+ h.sendLocked(tp, out)
+ return nil
+}
+
+// ringLocked keeps a copy for the resume window, newest last. What falls
+// out of the ring is remembered as a high-water mark per sender, which
+// is the only way a resume can tell that a sender's messages were
+// evicted wholesale by another sender's burst.
+func (h *Hub) ringLocked(p *Participation, m Message) {
+ p.ring = append(p.ring, m)
+ if len(p.ring) <= RingSize {
+ return
+ }
+ for _, old := range p.ring[:len(p.ring)-RingSize] {
+ if !isTargeted(old.Type) {
+ continue
+ }
+ if p.evicted == nil {
+ p.evicted = map[PeerID]uint64{}
+ }
+ if old.Seq > p.evicted[old.From] {
+ p.evicted[old.From] = old.Seq
+ }
+ }
+ p.ring = p.ring[len(p.ring)-RingSize:]
+}
+
+// Reset bumps the pair's generation and tells both sides.
+func (h *Hub) Reset(sqid, session string, from, to PeerID) error {
+ 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
+ }
+ tp, ok := r.peers[to]
+ if !ok || !tp.Admitted {
+ return ErrUnknownPeer
+ }
+ k := keyFor(p.Peer, tp.Peer)
+ h.genLocked(r, p.Peer, tp.Peer) // starts the pair at one if it is new
+ r.gens[k]++
+ gen := r.gens[k]
+ for _, pair := range [][2]*Participation{{p, tp}, {tp, p}} {
+ m := Message{Type: TypePair, Peer: pair[1].Peer, Gen: gen, To: pair[0].Peer}
+ h.ringLocked(pair[0], m)
+ h.sendLocked(pair[0], m)
+ }
+ return nil
+}
+
+// ResumeResult is what the stream handler owes a browser that came back
+// inside its window: a fresh subscriber, and, when the participation is
+// admitted, the roster, the replay and the senders it must reset with.
+type ResumeResult struct {
+ Sub *Sub
+ Admitted bool
+ Roster *Roster
+ Replay []Message
+ Gap []PeerID
+}
+
+// ResumeError is a refused resume in the shape spec §4.5's table asks
+// for: a status, a reason on the terminal 410s, and the boot id on the
+// 404s so a browser can tell a restart from a lapsed window.
+type ResumeError struct {
+ Status int
+ Reason Reason
+ Boot string
+}
+
+func (e *ResumeError) Error() string {
+ return fmt.Sprintf("hub: resume refused %d %s", e.Status, e.Reason)
+}
+
+// Resume reattaches a stream to a participation in its window. It is
+// never a transition into the admitted state: a resume does not consume
+// a cap slot and cannot admit anyone (spec §4.3).
+func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if t, ok := h.tombstones[peer]; ok {
+ return ResumeResult{}, &ResumeError{Status: 410, Reason: t.reason}
+ }
+ if _, ok := h.rooms[sqid]; !ok {
+ return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot}
+ }
+ p, found := h.findLocked(sqid, session, peer)
+ if !found {
+ // Unknown here, unknown in another session, unknown in another
+ // room: one answer, so nothing leaks (spec §4.2).
+ return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot}
+ }
+ r := h.rooms[sqid]
+ if p.sub != nil {
+ // A second stream for the same peer replaces the attachment only:
+ // no tombstone, no departure, the participation carries on.
+ p.sub.terminate(Message{Type: TypeLost, Reason: ReasonReplaced})
+ }
+ p.sub = newSub()
+ p.resumeUntil = time.Time{}
+ res := ResumeResult{Sub: p.sub, Admitted: p.Admitted}
+ if !p.Admitted {
+ // A waiting guest learns nothing by resuming (ruling 9).
+ return res, nil
+ }
+ res.Roster = h.rosterFor(r, p)
+ // Replay is per sender: everything above last, unless the ring has
+ // dropped something above last, in which case that sender gets a gap
+ // and no partial replay (spec §4.5).
+ gapped := map[PeerID]bool{}
+ for from, high := range p.evicted {
+ if high > last[from] {
+ gapped[from] = true
+ res.Gap = append(res.Gap, from)
+ }
+ }
+ for _, m := range p.ring {
+ switch {
+ case isTargeted(m.Type):
+ if !gapped[m.From] && m.Seq > last[m.From] {
+ res.Replay = append(res.Replay, m)
+ }
+ case m.Type == TypePair:
+ // The roster already carries the current generation for every
+ // admitted peer, so this never fires for a current roster;
+ // the branch is kept because the ring is the authority for
+ // what was sent, and the browser reconciles from both (§4.6).
+ if m.Gen > h.currentGenSeenLocked(res.Roster, m.Peer) {
+ res.Replay = append(res.Replay, m)
+ }
+ }
+ }
+ return res, nil
+}
+
+func (h *Hub) currentGenSeenLocked(ro *Roster, peer PeerID) uint64 {
+ for _, rp := range ro.Peers {
+ if rp.Peer == peer {
+ return rp.Gen
+ }
+ }
+ return 0
+}
diff --git a/internal/meet/hub/relay_test.go b/internal/meet/hub/relay_test.go
new file mode 100644
index 0000000..1dd532a
--- /dev/null
+++ b/internal/meet/hub/relay_test.go
@@ -0,0 +1,235 @@
+package hub
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+)
+
+func twoAdmitted(t *testing.T) (*Hub, *time.Time, OpenResult, OpenResult) {
+ t.Helper()
+ 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)
+ return h, &now, a, b
+}
+
+func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T) {
+ h, _, a, b := twoAdmitted(t)
+ ctx := context.Background()
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ drain(a.Sub)
+ drain(b.Sub)
+ err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1, From: "forged", Name: "Forged", Payload: json.RawMessage(`{"sdp":"x"}`)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ bev := drain(b.Sub)
+ if len(bev) != 1 || bev[0].From != a.Peer || bev[0].Name != "Ann" || string(bev[0].Payload) != `{"sdp":"x"}` {
+ t.Fatalf("b got %+v", bev)
+ }
+ if ev := drain(a.Sub); len(ev) != 0 {
+ t.Fatalf("sender echoed %v", types(ev))
+ }
+ if ev := drain(g.Sub); len(ev) != 0 {
+ t.Fatalf("waiting guest received a relay: %v", types(ev))
+ }
+ if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: g.Peer, Seq: 2, Gen: 1}); err == nil {
+ t.Fatal("relay to a waiting peer accepted")
+ }
+ if err := h.Relay("r1", "sg", g.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrNotAdmitted {
+ t.Fatalf("waiting guest relaying: %v", err)
+ }
+ if err := h.Relay("r1", "sb", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer {
+ t.Fatalf("peer through another session: %v", err)
+ }
+ if err := h.Relay("r2", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer {
+ t.Fatalf("peer through another room: %v", err)
+ }
+ if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeChat, To: b.Peer, Seq: 4, Gen: 1}); err == nil {
+ t.Fatal("chat accepted as a targeted relay")
+ }
+}
+
+func TestStaleGenIsRefusedAndResetConverges(t *testing.T) {
+ h, _, a, b := twoAdmitted(t)
+ if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen {
+ t.Fatalf("gen 2 before any reset: %v", err)
+ }
+ if err := h.Reset("r1", "sa", a.Peer, b.Peer); err != nil {
+ t.Fatal(err)
+ }
+ if err := h.Reset("r1", "sb", b.Peer, a.Peer); err != nil { // simultaneous resets converge on the higher gen
+ t.Fatal(err)
+ }
+ aev, bev := drain(a.Sub), drain(b.Sub)
+ if len(aev) != 2 || aev[1].Type != TypePair || aev[1].Gen != 3 || aev[1].Peer != b.Peer {
+ t.Fatalf("a got %+v", aev)
+ }
+ if len(bev) != 2 || bev[1].Gen != 3 || bev[1].Peer != a.Peer {
+ t.Fatalf("b got %+v", bev)
+ }
+ if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != ErrStaleGen {
+ t.Fatal("old gen accepted after reset")
+ }
+ if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) {
+ h, now, a, b := twoAdmitted(t)
+ ctx := context.Background()
+ for i := uint64(1); i <= 5; i++ {
+ h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)})
+ }
+ drain(a.Sub)
+ h.Detach("r1", a.Peer)
+ *now = now.Add(5 * time.Second)
+ res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !res.Admitted || res.Roster == nil || len(res.Replay) != 2 || res.Replay[0].Seq != 4 || res.Replay[1].Seq != 5 {
+ t.Fatalf("resume %+v", res)
+ }
+ if len(res.Gap) != 0 {
+ t.Fatalf("a ring that held everything reported a gap: %v", res.Gap)
+ }
+ if res.Sub == nil || res.Sub.Lost() {
+ t.Fatal("resume did not attach a fresh subscriber")
+ }
+ // The old window is over; the tick must not reap a participation
+ // that has resumed.
+ *now = now.Add(ResumeWindow + time.Second)
+ h.Tick(ctx)
+ if _, ok := h.Find("r1", "sa", a.Peer); !ok {
+ t.Fatal("the tick reaped a participation that had resumed")
+ }
+ // A waiting guest resumes and learns nothing.
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ h.Detach("r1", g.Peer)
+ gres, err := h.Resume("r1", "sg", g.Peer, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if gres.Admitted || gres.Roster != nil || len(gres.Replay) != 0 {
+ t.Fatalf("waiting guest resume leaked %+v", gres)
+ }
+}
+
+func TestResumeGapWhenTheRingRolledOver(t *testing.T) {
+ h, now, a, b := twoAdmitted(t)
+ for i := uint64(1); i <= 40; i++ {
+ h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
+ }
+ drain(a.Sub)
+ h.Detach("r1", a.Peer)
+ *now = now.Add(time.Second)
+ res, _ := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
+ if len(res.Gap) != 1 || res.Gap[0] != b.Peer || len(res.Replay) != 0 {
+ t.Fatalf("expected a gap for b, got replay %d gap %v", len(res.Replay), res.Gap)
+ }
+}
+
+// One ring is shared by every sender, so one peer's burst can evict
+// another's messages wholesale. The gap must still be reported, which
+// the ring's own oldest seq per sender cannot see.
+func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) {
+ h, now, a, b := twoAdmitted(t)
+ ctx := context.Background()
+ c, _ := h.Open("r1", "sc", 3, "Cat", KindMember)
+ h.Join(ctx, "r1", "sc", c.Peer)
+ drain(a.Sub)
+ for i := uint64(1); i <= 3; i++ {
+ h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
+ }
+ for i := uint64(1); i <= RingSize; i++ {
+ h.Relay("r1", "sc", c.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
+ }
+ drain(a.Sub)
+ h.Detach("r1", a.Peer)
+ *now = now.Add(time.Second)
+ res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 1, c.Peer: RingSize})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(res.Gap) != 1 || res.Gap[0] != b.Peer {
+ t.Fatalf("b's messages were evicted wholesale, gap %v", res.Gap)
+ }
+}
+
+// A second resume of an attached peer replaces the attachment: the old
+// stream is told replaced, and the participation itself carries on.
+func TestResumeReplacesAnExistingAttachment(t *testing.T) {
+ h, _, a, b := twoAdmitted(t)
+ res, err := h.Resume("r1", "sa", a.Peer, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ old := drain(a.Sub)
+ if len(old) != 1 || old[0].Type != TypeLost || old[0].Reason != ReasonReplaced {
+ t.Fatalf("old stream got %+v", old)
+ }
+ if !a.Sub.Lost() {
+ t.Fatal("the old subscriber was not closed")
+ }
+ if ev := drain(b.Sub); len(ev) != 0 {
+ t.Fatalf("the room was told the peer left: %v", types(ev))
+ }
+ if !res.Admitted {
+ t.Fatal("the replaced attachment lost its admission")
+ }
+ if err := h.Relay("r1", "sb", b.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != nil {
+ t.Fatal(err)
+ }
+ if ev := drain(res.Sub); len(ev) != 1 || ev[0].Type != TypeOffer {
+ t.Fatalf("the new attachment got %v", types(ev))
+ }
+ if _, err := h.Resume("r1", "sa", a.Peer, nil); err != nil {
+ t.Fatalf("a replaced attachment left a tombstone: %v", err)
+ }
+}
+
+func TestResumeAnswers(t *testing.T) {
+ h, now, a, _ := twoAdmitted(t)
+ if _, err := h.Resume("r1", "sb", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 || err.(*ResumeError).Boot != h.Boot() {
+ t.Fatalf("resume through another session: %v", err)
+ }
+ if _, err := h.Resume("r9", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 {
+ t.Fatalf("resume through another room: %v", err)
+ }
+ h.Takeover("r1", "sa")
+ if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 410 || err.(*ResumeError).Reason != ReasonReplaced {
+ t.Fatalf("resume after takeover: %v", err)
+ }
+ *now = now.Add(TombstoneLife + time.Minute)
+ h.Tick(context.Background())
+ if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 {
+ t.Fatalf("resume after tombstone expiry: %v", err)
+ }
+ if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != nil {
+ t.Fatalf("fresh stream after tombstone: %v", err)
+ }
+}
+
+func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) {
+ h, now, a, _ := twoAdmitted(t)
+ h.Detach("r1", a.Peer)
+ *now = now.Add(ResumeWindow + time.Second)
+ h.Tick(context.Background())
+ _, err := h.Resume("r1", "sa", a.Peer, nil)
+ re, ok := err.(*ResumeError)
+ if !ok || re.Status != 404 || re.Boot != h.Boot() {
+ t.Fatalf("lapsed resume: %v", err)
+ }
+}