amadan

oficina / meet Public

hub: join, admit, leave, remove, takeover, end, the cap and the tick

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 d3d7e848ad051b184f4c3e0750e5c78535793c3c parent 3ea35db
3 files changed, +672 −3
  • internal/meet/hub/admission.go +304 −0
  • internal/meet/hub/admission_test.go +294 −0
  • internal/meet/hub/hub.go +74 −3
diff --git a/internal/meet/hub/admission.go b/internal/meet/hub/admission.go
new file mode 100644
index 0000000..7f2ed0e
--- /dev/null
+++ b/internal/meet/hub/admission.go
@@ -0,0 +1,304 @@
+package hub
+
+import "context"
+
+func admittedCountLocked(r *room) int {
+ n := 0
+ for _, p := range r.peers {
+ if p.Admitted {
+ n++
+ }
+ }
+ return n
+}
+
+func waitingCountLocked(r *room) int {
+ n := 0
+ for _, p := range r.peers {
+ if p.Joined && !p.Admitted {
+ n++
+ }
+ }
+ return n
+}
+
+// rosterFor is the authoritative snapshot for one viewer: admitted peers
+// with the viewer's pair generation, and the waiting queue only when the
+// viewer is a member (spec §4.3).
+func (h *Hub) rosterFor(r *room, viewer *Participation) *Roster {
+ ro := &Roster{}
+ for _, p := range r.peers {
+ if !p.Admitted {
+ continue
+ }
+ rp := RosterPeer{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String()}
+ if p.Peer != viewer.Peer {
+ rp.Gen = h.genLocked(r, viewer.Peer, p.Peer)
+ }
+ ro.Peers = append(ro.Peers, rp)
+ }
+ if viewer.Kind == KindMember {
+ for _, p := range r.peers {
+ if p.Joined && !p.Admitted {
+ ro.Waiting = append(ro.Waiting, RosterPeer{Peer: p.Peer, Name: p.Name, Kind: "guest"})
+ }
+ }
+ }
+ return ro
+}
+
+// genLocked returns the pair's generation, starting it at one on first
+// sight (spec §4.6).
+func (h *Hub) genLocked(r *room, a, b PeerID) uint64 {
+ k := keyFor(a, b)
+ if r.gens[k] == 0 {
+ r.gens[k] = 1
+ }
+ return r.gens[k]
+}
+
+// admitLocked is the one transition into the admitted state: the cap
+// under the lock, roster to the newcomer, peer to the others, knock-gone
+// to members if they were waiting. Returns false when the room is full.
+func (h *Hub) admitLocked(r *room, p *Participation) bool {
+ if admittedCountLocked(r) >= MaxAdmitted {
+ return false
+ }
+ wasWaiting := p.Joined && !p.Admitted
+ p.Admitted = true
+ p.Joined = true
+ p.JoinedAt = h.cfg.Now()
+ if p.sub != nil {
+ p.sub.send(Message{Type: TypeRoster, Roster: h.rosterFor(r, p)})
+ }
+ for _, q := range r.peers {
+ if q.Admitted && q.Peer != p.Peer && q.sub != nil {
+ // The one-entry roster carries this recipient's own pair
+ // generation with the newcomer; browsers read it from
+ // roster.peers[0].
+ q.sub.send(Message{Type: TypePeer, Peer: p.Peer, Name: p.Name,
+ Roster: &Roster{Peers: []RosterPeer{{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String(), Gen: h.genLocked(r, q.Peer, p.Peer)}}}})
+ }
+ }
+ if wasWaiting {
+ h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers)
+ }
+ return true
+}
+
+// refreshRostersLocked sends the authoritative snapshot to every
+// admitted peer, which is what a kind change owes them (spec §2.4).
+func (h *Hub) refreshRostersLocked(r *room) {
+ for _, q := range r.peers {
+ if q.Admitted && q.sub != nil {
+ q.sub.send(Message{Type: TypeRoster, Roster: h.rosterFor(r, q)})
+ }
+ }
+}
+
+// Join is the entry, so standing is re-derived now (spec §2.5).
+func (h *Hub) Join(ctx context.Context, sqid, session string, peer PeerID) error {
+ kind, ok, err := h.standing(ctx, sqid, session, peer)
+ if err != nil {
+ return err
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ p, found := h.findLocked(sqid, session, peer)
+ if !found {
+ return ErrUnknownPeer // it ended while standing was being read
+ }
+ r := h.rooms[sqid]
+ if !ok {
+ h.endLocked(r, p, ReasonStanding)
+ return ErrNotYours
+ }
+ p.Kind = kind
+ if p.Joined {
+ return nil // idempotent
+ }
+ if kind == KindMember {
+ if !h.admitLocked(r, p) {
+ if p.sub != nil {
+ p.sub.send(Message{Type: TypeFull})
+ }
+ return ErrRoomFull
+ }
+ return nil
+ }
+ if waitingCountLocked(r) >= MaxWaiting {
+ if p.sub != nil {
+ p.sub.send(Message{Type: TypeFull})
+ }
+ return ErrTooManyWaiting
+ }
+ p.Joined = true
+ p.JoinedAt = h.cfg.Now()
+ h.broadcastLocked(r, Message{Type: TypeKnock, From: p.Peer, Name: p.Name}, admittedMembers)
+ return nil
+}
+
+// standing looks the peer up to find its person, then asks the app
+// without holding the lock, because it is a database read.
+func (h *Hub) standing(ctx context.Context, sqid, session string, peer PeerID) (Kind, bool, error) {
+ h.mu.Lock()
+ p, found := h.findLocked(sqid, session, peer)
+ var personID int64
+ if found {
+ personID = p.PersonID
+ }
+ h.mu.Unlock()
+ if !found {
+ return 0, false, ErrUnknownPeer
+ }
+ return h.cfg.StandingFor(ctx, sqid, personID)
+}
+
+// Admit: the actor must be admitted with current member kind; the target
+// waiting with entry authority now; the cap under the lock (spec §4.3).
+func (h *Hub) Admit(ctx context.Context, sqid, session string, actor, target PeerID) error {
+ actorKind, actorOK, err := h.standing(ctx, sqid, session, actor)
+ if err != nil {
+ return err
+ }
+ h.mu.Lock()
+ var targetPerson int64
+ found := false
+ if r, ok := h.rooms[sqid]; ok {
+ var tp *Participation
+ if tp, found = r.peers[target]; found {
+ targetPerson = tp.PersonID
+ }
+ }
+ h.mu.Unlock()
+ if !found {
+ return ErrNotWaiting
+ }
+ targetKind, targetOK, err := h.cfg.StandingFor(ctx, sqid, targetPerson)
+ if err != nil {
+ return err
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ // Both participations are re-found: either may have ended while the
+ // two standing reads were in flight.
+ a, ok := h.findLocked(sqid, session, actor)
+ if !ok {
+ return ErrUnknownPeer
+ }
+ r := h.rooms[sqid]
+ if !actorOK {
+ h.endLocked(r, a, ReasonStanding)
+ return ErrNotYours
+ }
+ a.Kind = actorKind
+ if !a.Admitted {
+ return ErrNotAdmitted
+ }
+ if a.Kind != KindMember {
+ return ErrNotMember
+ }
+ tp, ok := r.peers[target]
+ if !ok || !tp.Joined || tp.Admitted {
+ return ErrNotWaiting
+ }
+ if !targetOK {
+ wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonExpired)
+ h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting)
+ return nil
+ }
+ tp.Kind = targetKind
+ if !h.admitLocked(r, tp) {
+ if a.sub != nil {
+ a.sub.send(Message{Type: TypeFull})
+ }
+ }
+ return nil
+}
+
+// Leave ends a participation in any state at its own request. It is not
+// a tombstoned ending: a leave is not something to resume from.
+func (h *Hub) Leave(sqid, session string, peer PeerID) error {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ r, ok := h.rooms[sqid]
+ if !ok {
+ return ErrUnknownPeer
+ }
+ p, found := h.findLocked(sqid, session, peer)
+ if !found {
+ return ErrUnknownPeer
+ }
+ wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer)
+ delete(h.tombstones, p.Peer)
+ h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
+ return nil
+}
+
+// Remove is spec §3.5: the actor holds an admitted member participation
+// in this room (kind re-derived), the target is any other participation.
+func (h *Hub) Remove(ctx context.Context, sqid, session string, actor, target PeerID) error {
+ kind, ok, err := h.standing(ctx, sqid, session, actor)
+ if err != nil {
+ return err
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ a, found := h.findLocked(sqid, session, actor)
+ if !found || !ok {
+ return ErrUnknownPeer
+ }
+ r := h.rooms[sqid]
+ a.Kind = kind
+ if !a.Admitted {
+ return ErrNotAdmitted
+ }
+ if a.Kind != KindMember {
+ return ErrNotMember
+ }
+ tp, found := r.peers[target]
+ if !found || tp.Peer == a.Peer {
+ return ErrUnknownPeer
+ }
+ wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonRemoved)
+ h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting)
+ return nil
+}
+
+// Takeover ends the session's live participation in the room with the
+// state-dependent departure (spec §4.2), leaving a replaced tombstone so
+// the displaced stream can be told why.
+func (h *Hub) Takeover(sqid, session string) (PeerID, error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ r, ok := h.rooms[sqid]
+ if !ok {
+ return "", ErrUnknownPeer
+ }
+ id, ok := h.bySession[sqid][session]
+ if !ok {
+ return "", ErrUnknownPeer
+ }
+ p, ok := r.peers[id]
+ if !ok {
+ delete(h.bySession[sqid], session)
+ return "", ErrUnknownPeer
+ }
+ wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonReplaced)
+ h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
+ return id, nil
+}
+
+// End is the meeting ending: lost{ended} to every stream, every
+// participation ended, and the room kept only as a headstone so a later
+// Open on an ended meeting is refused (spec §2.4).
+func (h *Hub) End(sqid string) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ r := h.roomLocked(sqid)
+ for _, p := range r.peers {
+ h.endLocked(r, p, ReasonEnded)
+ }
+ r.ended = true
+ delete(h.bySession, sqid)
+}
diff --git a/internal/meet/hub/admission_test.go b/internal/meet/hub/admission_test.go
new file mode 100644
index 0000000..4b7a6e8
--- /dev/null
+++ b/internal/meet/hub/admission_test.go
@@ -0,0 +1,294 @@
+package hub
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+// drain returns every event currently buffered on a sub, non-blocking.
+func drain(s *Sub) []Message {
+ var out []Message
+ for {
+ select {
+ case m, ok := <-s.C():
+ if !ok {
+ return out
+ }
+ out = append(out, m)
+ default:
+ return out
+ }
+ }
+}
+
+func types(ms []Message) []string {
+ out := make([]string, len(ms))
+ for i, m := range ms {
+ out[i] = m.Type
+ }
+ return out
+}
+
+func TestMemberJoinsAndGuestWaits(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ if err := h.Join(ctx, "r1", "sm", m.Peer); err != nil {
+ t.Fatal(err)
+ }
+ ev := drain(m.Sub)
+ if len(ev) != 1 || ev[0].Type != TypeRoster || len(ev[0].Roster.Peers) != 1 {
+ t.Fatalf("member join events %v", types(ev))
+ }
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ if err := h.Join(ctx, "r1", "sg", g.Peer); err != nil {
+ t.Fatal(err)
+ }
+ if ev := drain(g.Sub); len(ev) != 0 {
+ t.Fatalf("waiting guest received %v", types(ev))
+ }
+ ev = drain(m.Sub)
+ if len(ev) != 1 || ev[0].Type != TypeKnock || ev[0].From != g.Peer || ev[0].Name != "Gus" {
+ t.Fatalf("member did not get the knock: %v", types(ev))
+ }
+}
+
+func TestAdmitSendsRosterToGuestPeerToOthersAndKnockGone(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ drain(m.Sub)
+ if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil {
+ t.Fatal(err)
+ }
+ gev := drain(g.Sub)
+ if len(gev) != 1 || gev[0].Type != TypeRoster || len(gev[0].Roster.Peers) != 2 || gev[0].Roster.Waiting != nil {
+ t.Fatalf("guest got %v (waiting queue must not reach a guest)", types(gev))
+ }
+ mev := types(drain(m.Sub))
+ if len(mev) != 2 || mev[0] != TypePeer || mev[1] != TypeKnockGone {
+ t.Fatalf("member got %v", mev)
+ }
+}
+
+func TestOnlyAnAdmittedMemberMayAdmit(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ g1, _ := h.Open("r1", "s1", 100, "G1", KindGuest)
+ h.Join(ctx, "r1", "s1", g1.Peer)
+ g2, _ := h.Open("r1", "s2", 101, "G2", KindGuest)
+ h.Join(ctx, "r1", "s2", g2.Peer)
+ h.Admit(ctx, "r1", "sm", m.Peer, g1.Peer)
+ if err := h.Admit(ctx, "r1", "s1", g1.Peer, g2.Peer); err != ErrNotMember {
+ t.Fatalf("admitted guest admitting: %v", err)
+ }
+ if err := h.Admit(ctx, "r1", "s2", g2.Peer, g2.Peer); err != ErrNotAdmitted {
+ t.Fatalf("waiting guest admitting: %v", err)
+ }
+}
+
+func TestCapAtJoinAndAtAdmit(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ var members []OpenResult
+ for i := 0; i < 5; i++ {
+ o, _ := h.Open("r1", "s"+string(rune('a'+i)), int64(1+i), "M", KindMember)
+ members = append(members, o)
+ h.Join(ctx, "r1", "s"+string(rune('a'+i)), o.Peer)
+ }
+ fifth := drain(members[4].Sub)
+ if len(fifth) != 1 || fifth[0].Type != TypeFull {
+ t.Fatalf("fifth member got %v, want full", types(fifth))
+ }
+ // One leaves; four queued guests and the freed slot yield one admission.
+ h.Leave("r1", "sd", members[3].Peer)
+ var guests []OpenResult
+ for i := 0; i < 4; i++ {
+ o, _ := h.Open("r1", "g"+string(rune('a'+i)), int64(100+i), "G", KindGuest)
+ guests = append(guests, o)
+ h.Join(ctx, "r1", "g"+string(rune('a'+i)), o.Peer)
+ }
+ drain(members[0].Sub)
+ if err := h.Admit(ctx, "r1", "sa", members[0].Peer, guests[0].Peer); err != nil {
+ t.Fatal(err)
+ }
+ if err := h.Admit(ctx, "r1", "sa", members[0].Peer, guests[1].Peer); err != nil {
+ t.Fatal(err) // Admit itself returns nil; the outcome is the `full` event
+ }
+ ev := types(drain(members[0].Sub))
+ if ev[len(ev)-1] != TypeFull {
+ t.Fatalf("second admit over the cap: %v, want full last", ev)
+ }
+ if p, _ := h.Find("r1", "gb", guests[1].Peer); p.Admitted {
+ t.Fatal("guest admitted over the cap")
+ }
+}
+
+func TestEightWaitingThenFull(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ for i := 0; i < 8; i++ {
+ o, _ := h.Open("r1", "g"+string(rune('a'+i)), int64(100+i), "G", KindGuest)
+ h.Join(ctx, "r1", "g"+string(rune('a'+i)), o.Peer)
+ }
+ o, _ := h.Open("r1", "gz", 200, "G", KindGuest)
+ if err := h.Join(ctx, "r1", "gz", o.Peer); err != ErrTooManyWaiting {
+ t.Fatalf("ninth waiting: %v", err)
+ }
+ if ev := drain(o.Sub); len(ev) != 1 || ev[0].Type != TypeFull {
+ t.Fatalf("ninth got %v", types(ev))
+ }
+}
+
+func TestGuestFirstArrivalIsInTheMembersRoster(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ ev := drain(m.Sub)
+ if len(ev) != 1 || ev[0].Type != TypeRoster || len(ev[0].Roster.Waiting) != 1 || ev[0].Roster.Waiting[0].Peer != g.Peer {
+ t.Fatalf("member roster %+v", ev)
+ }
+}
+
+func TestAdmitRefusesAnExpiredGrantBeforeTheTick(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ authority := map[int64]bool{100: true}
+ h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
+ if id < 100 {
+ return KindMember, true, nil
+ }
+ return KindGuest, authority[id], nil
+ }})
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ authority[100] = false // grant expired, no tick yet
+ drain(m.Sub)
+ if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil {
+ t.Fatal(err)
+ }
+ if p, ok := h.Find("r1", "sg", g.Peer); ok && p.Admitted {
+ t.Fatal("expired guest was admitted")
+ }
+ gev := drain(g.Sub)
+ if len(gev) == 0 || gev[len(gev)-1].Type != TypeLost || gev[len(gev)-1].Reason != ReasonExpired {
+ t.Fatalf("expired guest got %v", types(gev))
+ }
+ if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeKnockGone {
+ t.Fatalf("member got %v", ev)
+ }
+}
+
+func TestTickDropsAMemberWhoLostStandingAndPromotesAWaitingGuest(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ kinds := map[int64]Kind{1: KindMember, 100: KindGuest}
+ ok := map[int64]bool{1: true, 100: true}
+ h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
+ return kinds[id], ok[id], nil
+ }})
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ drain(m.Sub)
+ kinds[100] = KindMember // promoted while waiting
+ h.Tick(ctx)
+ if p, _ := h.Find("r1", "sg", g.Peer); !p.Admitted || p.Kind != KindMember {
+ t.Fatal("waiting guest promoted to member was not admitted on the tick")
+ }
+ ok[1] = false // member removed from the allowlist
+ drain(g.Sub)
+ h.Tick(ctx)
+ if _, found := h.Find("r1", "sm", m.Peer); found {
+ t.Fatal("removed member still in the room")
+ }
+ mev := drain(m.Sub)
+ if len(mev) == 0 || mev[len(mev)-1].Reason != ReasonStanding {
+ t.Fatalf("removed member got %v", types(mev))
+ }
+ if ev := types(drain(g.Sub)); len(ev) == 0 || ev[len(ev)-1] != TypeGone {
+ t.Fatalf("remaining peer got %v, want gone last", ev)
+ }
+}
+
+func TestTakeoverDepartureDependsOnState(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ ag, _ := h.Open("r1", "sag", 100, "Adm", KindGuest)
+ h.Join(ctx, "r1", "sag", ag.Peer)
+ h.Admit(ctx, "r1", "sm", m.Peer, ag.Peer)
+ w, _ := h.Open("r1", "sw", 101, "Wait", KindGuest)
+ h.Join(ctx, "r1", "sw", w.Peer)
+ u, _ := h.Open("r1", "su", 102, "Unj", KindGuest)
+ drain(m.Sub)
+ drain(ag.Sub)
+ // waiting takeover: knock-gone to members only
+ if _, err := h.Takeover("r1", "sw"); err != nil {
+ t.Fatal(err)
+ }
+ if ev := types(drain(ag.Sub)); len(ev) != 0 {
+ t.Fatalf("admitted guest saw a waiting departure: %v", ev)
+ }
+ if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeKnockGone {
+ t.Fatalf("member got %v", ev)
+ }
+ // unjoined takeover: nothing
+ h.Takeover("r1", "su")
+ _ = u
+ if ev := types(drain(m.Sub)); len(ev) != 0 {
+ t.Fatalf("unjoined takeover announced %v", ev)
+ }
+ // admitted takeover: gone to admitted peers, old stream lost/replaced
+ h.Takeover("r1", "sag")
+ agev := drain(ag.Sub)
+ if len(agev) == 0 || agev[len(agev)-1].Reason != ReasonReplaced {
+ t.Fatalf("replaced peer got %v", types(agev))
+ }
+ if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeGone {
+ t.Fatalf("member got %v", ev)
+ }
+ if _, err := h.Open("r1", "sag", 100, "Adm", KindGuest); err != nil {
+ t.Fatalf("fresh stream after takeover: %v", err)
+ }
+}
+
+func TestEndSendsEndedToEveryoneAndDropsTheRoom(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ ctx := context.Background()
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer)
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer)
+ h.End("r1")
+ for _, s := range []*Sub{m.Sub, g.Sub} {
+ ev := drain(s)
+ if len(ev) == 0 || ev[len(ev)-1].Type != TypeLost || ev[len(ev)-1].Reason != ReasonEnded {
+ t.Fatalf("got %v, want lost/ended last", types(ev))
+ }
+ }
+ if _, err := h.Open("r1", "sx", 5, "X", KindMember); err == nil {
+ t.Fatal("ended room accepted a stream")
+ }
+}
diff --git a/internal/meet/hub/hub.go b/internal/meet/hub/hub.go
index 5cc7f8b..5a7328d 100644
--- a/internal/meet/hub/hub.go
+++ b/internal/meet/hub/hub.go
@@ -265,12 +265,22 @@ func (h *Hub) endLocked(r *room, p *Participation, reason Reason) (wasAdmitted,
return
}
+// standingCheck is one joined participation whose standing the tick must
+// re-read, and the answer it came back with.
+type standingCheck struct {
+ room string
+ peer PeerID
+ personID int64
+ kind Kind
+ ok bool
+}
+
// Tick runs the fifteen-second housekeeping: unjoined and lapsed
-// participations are dropped, tombstones expire. Task 2 adds kind
-// re-derivation and promotion.
+// participations are dropped, tombstones expire, and every joined
+// participation's kind is re-derived from standing, applying the
+// transition table of spec §2.4.
func (h *Hub) Tick(ctx context.Context) {
h.mu.Lock()
- defer h.mu.Unlock()
now := h.cfg.Now()
for _, r := range h.rooms {
for _, p := range r.peers {
@@ -290,6 +300,67 @@ func (h *Hub) Tick(ctx context.Context) {
delete(h.tombstones, id)
}
}
+ var checks []standingCheck
+ for _, r := range h.rooms {
+ for _, p := range r.peers {
+ if p.Joined {
+ checks = append(checks, standingCheck{room: r.sqid, peer: p.Peer, personID: p.PersonID})
+ }
+ }
+ }
+ h.mu.Unlock()
+
+ // Standing is a database read, so it happens with the lock down.
+ answered := make([]standingCheck, 0, len(checks))
+ for _, c := range checks {
+ kind, ok, err := h.cfg.StandingFor(ctx, c.room, c.personID)
+ if err != nil {
+ h.cfg.Logger.Warn("hub: standing lookup failed on tick",
+ "room", c.room, "peer", c.peer, "error", err)
+ continue // this participation waits for the next tick
+ }
+ c.kind, c.ok = kind, ok
+ answered = append(answered, c)
+ }
+
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ for _, c := range answered {
+ r, found := h.rooms[c.room]
+ if !found {
+ continue
+ }
+ p, found := r.peers[c.peer]
+ if !found || !p.Joined {
+ continue // it ended while standing was being read
+ }
+ switch {
+ case !c.ok && p.Kind == KindMember:
+ // A member who lost standing loses the room, admitted or not.
+ wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonStanding)
+ h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
+ case !c.ok && !p.Admitted:
+ // A waiting guest's grant lapsed before anyone admitted them.
+ wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonExpired)
+ h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
+ case !c.ok:
+ // An admitted guest whose grant expired: nothing. The
+ // participation is the authority now (ruling 14).
+ case p.Admitted:
+ if p.Kind != c.kind {
+ p.Kind = c.kind
+ h.refreshRostersLocked(r)
+ }
+ default:
+ // Waiting. A promotion to member is admission under the cap,
+ // exactly as an admit would be; a refusal leaves them waiting
+ // as a member, free to retry with join.
+ p.Kind = c.kind
+ if c.kind == KindMember && !h.admitLocked(r, p) && p.sub != nil {
+ p.sub.send(Message{Type: TypeFull})
+ }
+ }
+ }
}
// announceDepartureLocked tells the room that a participation is gone,