amadan
tests: the remaining slice-2 Go gates, and four carry-overs
Gates from §13 that Tasks 1 to 5 did not cover:
- concurrent admissions: five members race five waiting guests into one
free mesh slot and exactly one gets in
- generations only rise: two resets from one side yield 2 and 3, and
signalling at the superseded 2 is refused, which is what makes the
browser's "ignore a pair at or below the local generation" safe
- an admitted guest carries on past grant expiry on signal and on
stream?peer=, and is refused fresh entry
- the 64-event subscriber buffer over real HTTP: a client that stops
reading is told lost{buffer} and resumes the same participation
- a member dropped from the team is lost on the tick with reason
standing, and the room is told gone
Carry-overs:
- Task 2: Admit's expired-target path goes through standingLostLocked,
so a waiting *member* who lost standing is told standing, not expired
- Task 3: the resume replay test now sends into the window as well as
before it; a note beside End that any future room reaping must outlive
TombstoneLife
- Task 5: refuseBadRoom's body asserted byte-equal to the other 404s; a
validSqid table for the alphabet branch; a signal from a
detached-but-resumable session inside its window; the 429 log line's
peer= asserted to be the hub's, not the request's; the comment on
TestSignalRefusesAPeerFromAnotherMeeting corrected
- Task 6: roomAccess's participation fallback is ruling 14's alone, so
an ex-member whose participation still says member gets 404 rather
than being relabelled a guest; confirmRemove also requires fresh
member standing; its Cancel and back link go to the holding screen;
the guest tests prove admission; a guest is refused the confirm page;
TestRoomPageIsScriptlessHonestAndStandingGated renamed for what it
tests
New: TestHandle.SetAllowlist, with the app's allowlist behind an atomic
so a test can change the team while the tick goroutine reads it.
Deviation: §13's "refused fresh entry" is asserted on the fresh stream,
not on GET /m/{sqid}/join — Task 6's roomAccess deliberately admits an
admitted guest's participation to the holding screen (ruling 14), and
TestRoomPageSurvivesGrantExpiryForAnAdmittedGuest pins that. GET /ice is
slice 3's and is asserted absent, with a comment saying what that line
must become when it lands.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
13 files changed,
+996
−14
internal/meet/app.go+22 −0internal/meet/auth.go+2 −2internal/meet/handlers_room.go+27 −3internal/meet/hub/admission.go+15 −2internal/meet/hub/admission_test.go+172 −0internal/meet/hub/hub.go+23 −0internal/meet/hub/relay_test.go+67 −1internal/meet/ids_test.go+40 −0internal/meet/standing.go+17 −1internal/meettest/room_test.go+141 −1internal/meettest/signal_test.go+110 −4internal/meettest/sse_test.go+121 −0internal/meettest/stream_test.go+239 −0
diff --git a/internal/meet/app.go b/internal/meet/app.go| index 0aeecc7..81fd24b 100644 |
| --- a/internal/meet/app.go |
| +++ b/internal/meet/app.go |
| @@ -51,6 +51,12 @@ type app struct { |
| // them (SetStreamTimings) while stream handlers are reading them. |
| heartbeat atomic.Int64 |
| ttl atomic.Int64 |
| + |
| + // allow is the team, held as an atomic for the same reason: a test |
| + // drops a member from it (SetAllowlist) while the tick goroutine is |
| + // re-deriving standing. Nil means "as resolved at boot"; nothing in |
| + // production ever stores into it. Read it with allowNow(). |
| + allow atomic.Pointer[allowlist] |
| } |
| func (a *app) heartbeatEvery() time.Duration { return atLeast(a.heartbeat.Load()) } |
| @@ -86,6 +92,22 @@ func (h *TestHandle) SetStreamTimings(heartbeat, ttl time.Duration) { |
| h.a.ttl.Store(int64(ttl)) |
| } |
| +// SetAllowlist replaces the team, so a test can drop a member and watch |
| +// the next tick apply spec §2.4's transition to everyone that member |
| +// holds a participation for. The addresses are normalized exactly as |
| +// resolveAllowlist normalizes $MEET_ALLOWED_ADDRESSES, and the list is |
| +// always non-nil: SetAllowlist() with no addresses means "nobody is a |
| +// member", never development's "everybody is". |
| +func (h *TestHandle) SetAllowlist(addrs ...string) { |
| + list := allowlist{addrs: map[string]struct{}{}} |
| + for _, addr := range addrs { |
| + if addr = normalizeAddress(addr); addr != "" { |
| + list.addrs[addr] = struct{}{} |
| + } |
| + } |
| + h.a.allow.Store(&list) |
| +} |
| + |
| // TickForTest runs one pass of the hub housekeeping App runs every |
| // fifteen seconds: unjoined and lapsed participations dropped, |
| // tombstones expired, every joined kind re-derived from standing. |
diff --git a/internal/meet/auth.go b/internal/meet/auth.go| index 8891699..67fba9b 100644 |
| --- a/internal/meet/auth.go |
| +++ b/internal/meet/auth.go |
| @@ -33,7 +33,7 @@ func (a *app) newAuthPlugin(writer *sql.DB) (*rauth.Auth, error) { |
| } |
| func (a *app) authorize(address string) bool { |
| - if a.cfg.allow.allows(address) { |
| + if a.allowNow().allows(address) { |
| return true |
| } |
| live, err := a.store.pendingLive(context.Background(), address) |
| @@ -79,7 +79,7 @@ func (a *app) gateBegin(next http.HandlerFunc) http.HandlerFunc { |
| // written first, for the member too (it carries their destination). |
| func (a *app) mayRequestLink(w http.ResponseWriter, r *http.Request, address string) bool { |
| address = normalizeAddress(address) |
| - member := a.cfg.allow.allows(address) |
| + member := a.allowNow().allows(address) |
| ck, hasCookie := a.readJoinCookie(r) |
| if !hasCookie { |
| return member |
diff --git a/internal/meet/handlers_room.go b/internal/meet/handlers_room.go| index dc9bec5..bfc3b0e 100644 |
| --- a/internal/meet/handlers_room.go |
| +++ b/internal/meet/handlers_room.go |
| @@ -37,7 +37,16 @@ func (a *app) roomAccess(w http.ResponseWriter, r *http.Request) (Meeting, Perso |
| ErrorPage(w, r, http.StatusInternalServerError, "") |
| return Meeting{}, Person{}, StandingNone, false |
| } |
| - if _, live := a.hub.LiveForSession(sqid, session); !live { |
| + // The fallback is ruling 14's and nothing wider: an admitted |
| + // guest survives their own grant expiring, because §2.4 makes |
| + // the participation the authority once they are in. A member who |
| + // lost standing is not that case — §2.5 drops them by the next |
| + // thing they do — so a participation whose kind is member is no |
| + // longer a way into this page. Relabelling one as a guest would |
| + // hand an ex-member the room for as long as the hub happened to |
| + // hold their participation, which is the opposite of the rule. |
| + _, kind, live := a.hub.LiveKindForSession(sqid, session) |
| + if !live || kind == hub.KindMember { |
| ErrorPage(w, r, http.StatusNotFound, "") |
| return Meeting{}, Person{}, StandingNone, false |
| } |
| @@ -79,10 +88,20 @@ func hubRefusal(err error) bool { |
| // can never name — or offer to remove — a peer the actor could not |
| // actually remove. |
| func (a *app) confirmRemove(w http.ResponseWriter, r *http.Request) { |
| - m, _, _, ok := a.roomAccess(w, r) |
| + m, _, st, ok := a.roomAccess(w, r) |
| if !ok { |
| return |
| } |
| + // Fresh standing as well as the participation PeerNameFor checks: |
| + // §2.5's current kind, re-derived on this request. PeerNameFor reads |
| + // the kind the last tick wrote, and a member demoted since then |
| + // would otherwise still be shown a page offering to remove someone — |
| + // a page whose POST hub.Remove would then refuse. The two checks |
| + // answer the same 404, so nothing distinguishes them from outside. |
| + if st != StandingMember { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| target := chi.URLParam(r, "peer") |
| if !validPeerID(hub.PeerID(target)) { |
| ErrorPage(w, r, http.StatusNotFound, "") |
| @@ -107,7 +126,12 @@ func (a *app) confirmRemove(w http.ResponseWriter, r *http.Request) { |
| // Task 11 copy catalogue). |
| "Body": "They will be disconnected and can rejoin only if they are still allowed in.", |
| "Action": "/m/" + m.Sqid + "/remove/" + target, "Label": "Remove", |
| - "CancelHref": "/m/" + m.Sqid, "BackLabel": a.store.title(m), "Sqid": m.Sqid, |
| + // Cancel and the back link both lead to the holding screen, the |
| + // same place a confirmed removal lands. This page opens in its |
| + // own tab while the call holds the one behind it, and /m/{sqid} |
| + // would open a second stream in this one — a 409 and a takeover |
| + // offer, for a person who only pressed Cancel (issue 11). |
| + "CancelHref": "/m/" + m.Sqid + "/join", "BackLabel": a.store.title(m), "Sqid": m.Sqid, |
| })) |
| } |
diff --git a/internal/meet/hub/admission.go b/internal/meet/hub/admission.go| index d2192cc..544949c 100644 |
| --- a/internal/meet/hub/admission.go |
| +++ b/internal/meet/hub/admission.go |
| @@ -285,8 +285,14 @@ func (h *Hub) Admit(ctx context.Context, sqid, session string, actor, target Pee |
| return ErrNotWaiting |
| } |
| if !targetOK { |
| - wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonExpired) |
| - h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting) |
| + // Through standingLostLocked, not endLocked: a loss of standing |
| + // costs what the kind held earns, and there is exactly one place |
| + // that decides which (spec §2.4). The target is always waiting |
| + // here — the guard above refuses an admitted one — but it is not |
| + // always a guest: the tick promotes a waiting guest to member |
| + // and leaves it waiting when the mesh is full, and that one is |
| + // owed reason "standing", not "expired". |
| + h.standingLostLocked(r, tp) |
| return nil |
| } |
| h.setKindLocked(r, tp, targetKind) |
| @@ -392,4 +398,11 @@ func (h *Hub) End(sqid string) { |
| } |
| r.ended = true |
| delete(h.bySession, sqid) |
| + // Nothing reaps rooms today, so a headstone lives as long as the |
| + // process. If a future round ever does reap them, it must leave an |
| + // ended room in place for at least TombstoneLife: the tombstones |
| + // written just above are answered only through their own room |
| + // (resumeRefusalLocked), and a room reaped ahead of them turns spec |
| + // §4.5's terminal 410 "ended" into a 404, which sends a browser off |
| + // to open a fresh stream on a meeting that is over. |
| } |
diff --git a/internal/meet/hub/admission_test.go b/internal/meet/hub/admission_test.go| index b70744e..ce914ce 100644 |
| --- a/internal/meet/hub/admission_test.go |
| +++ b/internal/meet/hub/admission_test.go |
| @@ -2,6 +2,8 @@ package hub |
| import ( |
| "context" |
| + "fmt" |
| + "sync" |
| "testing" |
| "time" |
| ) |
| @@ -510,3 +512,173 @@ func TestEndSendsEndedToEveryoneAndDropsTheRoom(t *testing.T) { |
| t.Fatal("ended room accepted a stream") |
| } |
| } |
| + |
| +// Spec §13's "the cap at join and at admit including four queued guests |
| +// and concurrent admissions": the mesh cap is checked under the hub's |
| +// own lock, inside admitLocked, so five members racing to fill the one |
| +// remaining slot cannot between them put five people in a room of four. |
| +// |
| +// Every call here is a legitimate request that the hub either performed |
| +// or refused for the cap, so none of them is an error: the loser learns |
| +// it lost from the `full` event on its own stream (§4.3), which is why |
| +// this asserts on the room's state rather than on a returned error. |
| +func TestConcurrentAdmissionsFillTheLastSlotExactlyOnce(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + |
| + // Three admitted members: three of the four mesh slots taken, one |
| + // free. Each admits from its own session, so five goroutines are |
| + // five real callers and not one caller five times. |
| + actors := make([]struct { |
| + session string |
| + peer PeerID |
| + }, 5) |
| + for i := range actors { |
| + session := fmt.Sprintf("sm%d", i) |
| + m, err := h.Open("r1", session, int64(1+i), "Mia", KindMember) |
| + if err != nil { |
| + t.Fatalf("open member %d: %v", i, err) |
| + } |
| + actors[i].session, actors[i].peer = session, m.Peer |
| + if i < 3 { |
| + if err := h.Join(ctx, "r1", session, m.Peer); err != nil { |
| + t.Fatalf("join member %d: %v", i, err) |
| + } |
| + } |
| + } |
| + if got := admittedNow(t, h, "r1"); got != 3 { |
| + t.Fatalf("setup left %d admitted, want 3", got) |
| + } |
| + // The two members that did not join are still members with a live |
| + // participation, but they are not admitted, so they may not admit: |
| + // only the three that joined are used as actors below. |
| + |
| + guests := make([]struct { |
| + session string |
| + peer PeerID |
| + }, 5) |
| + for i := range guests { |
| + session := fmt.Sprintf("sg%d", i) |
| + g, err := h.Open("r1", session, int64(100+i), "Gus", KindGuest) |
| + if err != nil { |
| + t.Fatalf("open guest %d: %v", i, err) |
| + } |
| + if err := h.Join(ctx, "r1", session, g.Peer); err != nil { |
| + t.Fatalf("guest %d knock: %v", i, err) |
| + } |
| + guests[i].session, guests[i].peer = session, g.Peer |
| + } |
| + |
| + var wg sync.WaitGroup |
| + errs := make([]error, len(guests)) |
| + start := make(chan struct{}) |
| + for i := range guests { |
| + wg.Add(1) |
| + go func(i int) { |
| + defer wg.Done() |
| + a := actors[i%3] // the three admitted members, round-robin |
| + <-start |
| + errs[i] = h.Admit(ctx, "r1", a.session, a.peer, guests[i].peer) |
| + }(i) |
| + } |
| + close(start) |
| + wg.Wait() |
| + |
| + for i, err := range errs { |
| + if err != nil { |
| + t.Fatalf("admit %d: %v", i, err) |
| + } |
| + } |
| + admitted := 0 |
| + for _, g := range guests { |
| + p, ok := h.Find("r1", g.session, g.peer) |
| + if !ok { |
| + t.Fatalf("a guest vanished: %s", g.peer) |
| + } |
| + if p.Admitted { |
| + admitted++ |
| + } |
| + } |
| + if admitted != 1 { |
| + t.Fatalf("%d of five guests were let into one free slot", admitted) |
| + } |
| + if got := admittedNow(t, h, "r1"); got != MaxAdmitted { |
| + t.Fatalf("the room holds %d admitted, want %d", got, MaxAdmitted) |
| + } |
| +} |
| + |
| +// admittedNow counts the room's admitted participations through the |
| +// hub's own lock, without reaching past its API for a field. |
| +func admittedNow(t *testing.T, h *Hub, sqid string) int { |
| + t.Helper() |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return 0 |
| + } |
| + return admittedCountLocked(r) |
| +} |
| + |
| +// Spec §2.4 and ruling 14: a loss of standing costs what the kind held |
| +// earns, and standingLostLocked is the one place that decides which. |
| +// Admit's expired-target path used to call endLocked(ReasonExpired) |
| +// directly, which is right for a guest and wrong for the one target |
| +// that is waiting and a member — the tick promotes a waiting guest to |
| +// member and leaves it waiting when the mesh is full (§2.4's retry), and |
| +// that participation is owed reason "standing". |
| +func TestAdmitOfAWaitingMemberWhoLostStandingEndsItAsAMember(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + kinds := map[int64]Kind{100: KindGuest} |
| + 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 kinds[id], authority[id], nil |
| + }}) |
| + ctx := context.Background() |
| + |
| + // Four admitted members: the mesh is full, so the promotion below |
| + // cannot admit and the guest waits on as a member. |
| + var actor OpenResult |
| + for i := 0; i < MaxAdmitted; i++ { |
| + session := fmt.Sprintf("sm%d", i) |
| + m, _ := h.Open("r1", session, int64(1+i), "Mia", KindMember) |
| + if err := h.Join(ctx, "r1", session, m.Peer); err != nil { |
| + t.Fatalf("join member %d: %v", i, err) |
| + } |
| + if i == 0 { |
| + actor = m |
| + } |
| + } |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + if err := h.Join(ctx, "r1", "sg", g.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + kinds[100] = KindMember // promoted while waiting |
| + h.Tick(ctx) |
| + p, ok := h.Find("r1", "sg", g.Peer) |
| + if !ok || p.Admitted || p.Kind != KindMember { |
| + t.Fatalf("setup: waiting member is %+v (admitted=%v)", p, ok && p.Admitted) |
| + } |
| + drain(g.Sub) |
| + |
| + // Standing goes before anyone gets round to admitting them. |
| + authority[100] = false |
| + if err := h.Admit(ctx, "r1", "sm0", actor.Peer, g.Peer); err != nil { |
| + t.Fatalf("admit: %v", err) |
| + } |
| + if _, ok := h.Find("r1", "sg", g.Peer); ok { |
| + t.Fatal("a target with no standing was left in the room") |
| + } |
| + ev := drain(g.Sub) |
| + if len(ev) == 0 || ev[len(ev)-1].Type != TypeLost { |
| + t.Fatalf("the dropped target got %v", types(ev)) |
| + } |
| + if got := ev[len(ev)-1].Reason; got != ReasonStanding { |
| + t.Fatalf("a waiting member who lost standing was told %q, want %q", got, ReasonStanding) |
| + } |
| +} |
diff --git a/internal/meet/hub/hub.go b/internal/meet/hub/hub.go| index c6e9a31..b320f20 100644 |
| --- a/internal/meet/hub/hub.go |
| +++ b/internal/meet/hub/hub.go |
| @@ -235,6 +235,29 @@ func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) { |
| return id, ok |
| } |
| +// LiveKindForSession is LiveForSession plus the kind that participation |
| +// currently carries — the kind the last tick or the last request wrote, |
| +// never a fresh standing read, which is the app's to do. |
| +// |
| +// It exists for the page authority of spec §2.5: a live participation |
| +// stands in for standing only where §2.4 says the participation is now |
| +// the authority, and that is an admitted **guest** whose grant lapsed |
| +// (ruling 14). A member who lost the room is dropped by the next thing |
| +// they do, so their participation must not let them keep the page. |
| +func (h *Hub) LiveKindForSession(sqid, session string) (PeerID, Kind, bool) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + id, ok := h.bySession[sqid][session] |
| + if !ok { |
| + return "", 0, false |
| + } |
| + p, ok := h.findLocked(sqid, session, id) |
| + if !ok { |
| + return "", 0, false |
| + } |
| + return id, p.Kind, true |
| +} |
| + |
| // PeerNameFor answers target's display name exactly when the caller |
| // (this session, in this room) could go on to remove it: an admitted |
| // member participation, and target present in the same room and not |
diff --git a/internal/meet/hub/relay_test.go b/internal/meet/hub/relay_test.go| index 80a79f1..2a762e4 100644 |
| --- a/internal/meet/hub/relay_test.go |
| +++ b/internal/meet/hub/relay_test.go |
| @@ -115,13 +115,31 @@ func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) { |
| drain(a.Sub) |
| h.Detach("r1", a.Peer, nil) |
| *now = now.Add(5 * time.Second) |
| + // Two more from the same sender *while a is detached*: the point of |
| + // the ring is that a message sent into the window is kept, not just |
| + // one that was already delivered and re-read. Without these, the |
| + // replay below would be satisfied by the ring alone and would say |
| + // nothing about what happens during the thirty seconds a browser is |
| + // away (Task 3 carry-over). |
| + for i := uint64(6); i <= 7; i++ { |
| + if err := h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)}); err != nil { |
| + t.Fatalf("relay %d into the resume window: %v", i, err) |
| + } |
| + } |
| 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 { |
| + if !res.Admitted || res.Roster == nil { |
| t.Fatalf("resume %+v", res) |
| } |
| + var seqs []uint64 |
| + for _, m := range res.Replay { |
| + seqs = append(seqs, m.Seq) |
| + } |
| + if len(seqs) != 4 || seqs[0] != 4 || seqs[1] != 5 || seqs[2] != 6 || seqs[3] != 7 { |
| + t.Fatalf("replay seqs %v, want 4 5 6 7 (everything above last, delivered and undelivered alike)", seqs) |
| + } |
| if len(res.Gap) != 0 { |
| t.Fatalf("a ring that held everything reported a gap: %v", res.Gap) |
| } |
| @@ -362,3 +380,51 @@ func TestAResumePastItsWindowIsRefusedBeforeTheTickRemovesIt(t *testing.T) { |
| t.Fatal("the tick left the lapsed participation behind") |
| } |
| } |
| + |
| +// Spec §4.6: a `pair` at or below the generation a browser already holds |
| +// is ignored. The hub owns the half of that rule this package can gate — |
| +// generations only ever rise, and only the newest is live — and the |
| +// browser's half rests on it: two resets from one side yield 2 and 3, so |
| +// the `pair` carrying 2 is superseded by the time it could be applied, |
| +// and signalling that still carries 2 is refused. Ignoring a superseded |
| +// `pair` is therefore safe rather than merely tidy. |
| +func TestGenerationsOnlyRiseAndASupersededPairIsStale(t *testing.T) { |
| + h, _, a, b := twoAdmitted(t) |
| + for i := 0; i < 2; i++ { |
| + if err := h.Reset("r1", "sa", a.Peer, b.Peer); err != nil { |
| + t.Fatalf("reset %d: %v", i+1, err) |
| + } |
| + } |
| + for _, side := range []struct { |
| + name string |
| + ms []Message |
| + other PeerID |
| + }{ |
| + {"a", drain(a.Sub), b.Peer}, |
| + {"b", drain(b.Sub), a.Peer}, |
| + } { |
| + if len(side.ms) != 2 { |
| + t.Fatalf("%s got %v, want one pair per reset", side.name, types(side.ms)) |
| + } |
| + for i, want := range []uint64{2, 3} { |
| + m := side.ms[i] |
| + if m.Type != TypePair || m.Peer != side.other || m.Gen != want { |
| + t.Fatalf("%s pair %d: %+v, want pair with %s at gen %d", side.name, i, m, side.other, want) |
| + } |
| + } |
| + } |
| + |
| + // Gen 2 is exactly the superseded one: announced by the first reset, |
| + // replaced by the second before either side could finish with it. |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen { |
| + t.Fatalf("signalling at the superseded generation: %v, want ErrStaleGen", err) |
| + } |
| + // A generation the hub never issued is refused too: only the current |
| + // one is live, from either direction. |
| + if err := h.Relay("r1", "sb", b.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 4}); err != ErrStaleGen { |
| + t.Fatalf("signalling above the current generation: %v, want ErrStaleGen", err) |
| + } |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil { |
| + t.Fatalf("signalling at the current generation: %v", err) |
| + } |
| +} |
diff --git a/internal/meet/ids_test.go b/internal/meet/ids_test.go| index b1c41f7..0108063 100644 |
| --- a/internal/meet/ids_test.go |
| +++ b/internal/meet/ids_test.go |
| @@ -17,6 +17,46 @@ func TestSqidUsesOnlyTheReadableAlphabet(t *testing.T) { |
| } |
| } |
| +// validSqid is what stands between a room id straight off the request |
| +// line and a rate-limiter key, a hub lookup and a log line (review Task |
| +// 5 item 1), so both of its branches are worth a table: the length, and |
| +// the alphabet. The alphabet branch is the one no other test reaches — |
| +// every id newSqid produces is in it by construction. |
| +func TestValidSqidChecksLengthAndAlphabet(t *testing.T) { |
| + good, err := newSqid() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + table := []struct { |
| + name string |
| + in string |
| + want bool |
| + }{ |
| + {"a minted id", good, true}, |
| + {"ten from the alphabet", "23456789ab", true}, |
| + {"the last character of the alphabet", "zzzzzzzzzz", true}, |
| + {"empty", "", false}, |
| + {"nine characters", "23456789a", false}, |
| + {"eleven characters", "23456789abc", false}, |
| + {"the letter o, which the alphabet drops", "2345678oab", false}, |
| + {"the letter l, which the alphabet drops", "2345678lab", false}, |
| + {"the digit one, which the alphabet drops", "23456781ab", false}, |
| + {"the digit zero, which the alphabet drops", "23456780ab", false}, |
| + {"uppercase", "23456789AB", false}, |
| + {"a hyphen", "2345678-ab", false}, |
| + {"a dot", "23456789.b", false}, |
| + {"a space", "23456789 b", false}, |
| + {"ten bytes of five multi-byte runes", strings.Repeat("é", 5), false}, |
| + } |
| + for _, tc := range table { |
| + t.Run(tc.name, func(t *testing.T) { |
| + if got := validSqid(tc.in); got != tc.want { |
| + t.Fatalf("validSqid(%q) = %v, want %v", tc.in, got, tc.want) |
| + } |
| + }) |
| + } |
| +} |
| + |
| func TestLinkTokenHashIsSHA256Hex(t *testing.T) { |
| tok, hash, err := newLinkToken() |
| if err != nil { |
diff --git a/internal/meet/standing.go b/internal/meet/standing.go| index e00a96a..23758f8 100644 |
| --- a/internal/meet/standing.go |
| +++ b/internal/meet/standing.go |
| @@ -63,9 +63,25 @@ func (s Standing) String() string { |
| return "none" |
| } |
| +// allowNow is the team as it stands right now. It is read through an |
| +// atomic rather than straight off a.cfg because TestHandle.SetAllowlist |
| +// swaps it while the tick goroutine and live handlers are reading it — |
| +// a member dropped from the team mid-call is what spec §2.4's tick has |
| +// to notice, and there is no other way to make that happen in a test |
| +// without rewriting an environment variable and restarting the process. |
| +// Production never calls the setter, so this is one atomic load on a |
| +// path that already does a database read. |
| +func (a *app) allowNow() allowlist { |
| + if l := a.allow.Load(); l != nil { |
| + return *l |
| + } |
| + return a.cfg.allow |
| +} |
| + |
| // isMember is the lone-shape roster read: the allowlist. |
| func (a *app) isMember(p Person) bool { |
| - return a.cfg.allow.addrs != nil && a.cfg.allow.allows(p.Email) |
| + l := a.allowNow() |
| + return l.addrs != nil && l.allows(p.Email) |
| } |
| func (a *app) standingFor(ctx context.Context, p Person, m Meeting) (Standing, error) { |
diff --git a/internal/meettest/room_test.go b/internal/meettest/room_test.go| index bd91cec..d99c5dc 100644 |
| --- a/internal/meettest/room_test.go |
| +++ b/internal/meettest/room_test.go |
| @@ -8,7 +8,13 @@ import ( |
| "time" |
| ) |
| -func TestRoomPageIsScriptlessHonestAndStandingGated(t *testing.T) { |
| +// Renamed from TestRoomPageIsScriptlessHonestAndStandingGated: the |
| +// standing gate is not tested here and never was — fix round 1's |
| +// Important 1 moved it to |
| +// TestRoomAndTakeoverAreGatedByStandingOrParticipation, which has a real |
| +// second meeting to check it against. What is left is what this test |
| +// actually asserts. |
| +func TestRoomPageRendersItsHeaderStandingAndScriptlessFallback(t *testing.T) { |
| x, sqid, _ := memberWithMeeting(t) |
| c := x.browser(t) |
| c.signIn("paul@example.com") |
| @@ -100,6 +106,13 @@ func TestRoomPageSurvivesGrantExpiryForAnAdmittedGuest(t *testing.T) { |
| pg, _ := sg.hello() |
| g.signal(sqid, pg, `{"type":"join"}`) |
| m.signal(sqid, pm, `{"type":"admit","payload":{"peer":"`+pg+`"}}`) |
| + // The admission itself, proved rather than assumed: a guest still |
| + // only waiting would survive the expiry below on the waiting |
| + // participation alone, and the test would say nothing about ruling |
| + // 14. The roster is what admission sends and waiting never does. |
| + if ev, _, ok := sg.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("the guest was never admitted (got %q)", ev) |
| + } |
| x.advance(2*time.Hour + time.Minute) |
| if rec := g.get("/m/" + sqid); rec.Code != http.StatusOK { |
| @@ -107,6 +120,117 @@ func TestRoomPageSurvivesGrantExpiryForAnAdmittedGuest(t *testing.T) { |
| } |
| } |
| +// The other side of ruling 14, and Task 6's issue 6: roomAccess falls |
| +// back to a live participation only where §2.4 makes the participation |
| +// the authority, which is an admitted **guest**. A member who has lost |
| +// the team is dropped by the next thing they do (§2.5), so their own |
| +// participation is not a way to keep the page — relabelling it as a |
| +// guest's would hand an ex-member the room for as long as the hub |
| +// happened to still hold it. |
| +func TestAnExMemberCannotKeepTheRoomThroughTheirOwnParticipation(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + a := x.browser(t) |
| + a.signIn("paul@example.com") |
| + sa := a.stream("/m/" + sqid + "/stream") |
| + defer sa.close() |
| + pa, _ := sa.hello() |
| + a.signal(sqid, pa, `{"type":"join"}`) |
| + |
| + b := newMemberBrowser(t, x, "paul@rslw.com") |
| + sb := b.stream("/m/" + sqid + "/stream") |
| + defer sb.close() |
| + pb, _ := sb.hello() |
| + if resp, _ := b.signal(sqid, pb, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("b's join: %d", resp.StatusCode) |
| + } |
| + if ev, _, ok := sb.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("b was never admitted (got %q)", ev) |
| + } |
| + for _, path := range []string{"/m/" + sqid, "/m/" + sqid + "/join"} { |
| + if rec := b.get(path); rec.Code != http.StatusOK { |
| + t.Fatalf("GET %s while on the team: %d", path, rec.Code) |
| + } |
| + } |
| + |
| + // Off the team, and deliberately no tick: the participation is still |
| + // there, still admitted, and still says member. Standing is the only |
| + // thing that has changed. |
| + x.ctl.SetAllowlist("paul@example.com") |
| + for _, path := range []string{"/m/" + sqid, "/m/" + sqid + "/join"} { |
| + if rec := b.get(path); rec.Code != http.StatusNotFound { |
| + t.Fatalf("GET %s after leaving the team: %d, want 404", path, rec.Code) |
| + } |
| + } |
| + // The member who is still on it is untouched. |
| + if rec := a.get("/m/" + sqid); rec.Code != http.StatusOK { |
| + t.Fatalf("the remaining member's room page: %d", rec.Code) |
| + } |
| +} |
| + |
| +// Task 6's issue 3, finished: the removal confirm page needs current |
| +// member standing as well as the admitted member participation |
| +// hub.PeerNameFor checks. PeerNameFor reads the kind the last tick |
| +// wrote, so a member who has just left the team — and who still reaches |
| +// the page through a guest grant of their own — would otherwise be |
| +// offered a page whose POST hub.Remove refuses. Both refusals are the |
| +// same 404, never a 403 (§2.5). |
| +func TestTheRemovalConfirmPageNeedsFreshMemberStanding(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + a := x.browser(t) |
| + a.signIn("paul@example.com") |
| + sa := a.stream("/m/" + sqid + "/stream") |
| + defer sa.close() |
| + pa, _ := sa.hello() |
| + a.signal(sqid, pa, `{"type":"join"}`) |
| + |
| + b := newMemberBrowser(t, x, "paul@rslw.com") |
| + sb := b.stream("/m/" + sqid + "/stream") |
| + defer sb.close() |
| + pb, _ := sb.hello() |
| + b.signal(sqid, pb, `{"type":"join"}`) |
| + if ev, _, ok := sb.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("b was never admitted (got %q)", ev) |
| + } |
| + if rec := b.get("/m/" + sqid + "/remove/" + pa); rec.Code != http.StatusOK { |
| + t.Fatalf("confirm page for a member on the team: %d", rec.Code) |
| + } |
| + |
| + // Off the team, and then in through the front door as an outsider: |
| + // the link mints a guest grant, so roomAccess answers guest standing |
| + // while the hub still has the participation down as a member. That |
| + // gap is the one the standing check closes. |
| + x.ctl.SetAllowlist("paul@example.com") |
| + rec := b.get("/j/" + tok) |
| + if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/m/"+sqid+"/join" { |
| + t.Fatalf("the ex-member's walk through the join link: %d %s", rec.Code, rec.Header().Get("Location")) |
| + } |
| + if rec := b.get("/m/" + sqid + "/remove/" + pa); rec.Code != http.StatusNotFound { |
| + t.Fatalf("confirm page for an ex-member holding a guest grant: %d, want 404", rec.Code) |
| + } |
| + // And the POST agrees, which is the whole point: a page that offers |
| + // an action its own POST would refuse is a page that lies. |
| + if rec := b.post("/m/"+sqid+"/remove/"+pa, nil); rec.Code != http.StatusNotFound { |
| + t.Fatalf("POST remove for an ex-member holding a guest grant: %d, want 404", rec.Code) |
| + } |
| + |
| + // A plain guest, admitted and all, never gets the page either — not |
| + // even for a peer that really is in the room. |
| + g := x.browser(t) |
| + g.get("/j/" + tok) |
| + g.signIn("guest@example.com") |
| + sg := g.stream("/m/" + sqid + "/stream") |
| + defer sg.close() |
| + pg, _ := sg.hello() |
| + g.signal(sqid, pg, `{"type":"join"}`) |
| + a.signal(sqid, pa, `{"type":"admit","payload":{"peer":"`+pg+`"}}`) |
| + if ev, _, ok := sg.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("the guest was never admitted (got %q)", ev) |
| + } |
| + if rec := g.get("/m/" + sqid + "/remove/" + pa); rec.Code != http.StatusNotFound { |
| + t.Fatalf("confirm page for an admitted guest: %d, want 404", rec.Code) |
| + } |
| +} |
| + |
| func TestRemoveNeedsAConfirmPageAndAnAdmittedMemberActor(t *testing.T) { |
| x, sqid, tok := memberWithMeeting(t) |
| m := x.browser(t) |
| @@ -135,6 +259,17 @@ func TestRemoveNeedsAConfirmPageAndAnAdmittedMemberActor(t *testing.T) { |
| if !strings.Contains(rec.Body.String(), "Board meeting") { |
| t.Fatalf("confirm page back-nav: %s", rec.Body.String()) |
| } |
| + // Cancel and the back link both lead to the holding screen, where a |
| + // confirmed removal also lands (Important 11's other half). Never |
| + // /m/{sqid}: this page opens in its own tab while the call holds the |
| + // one behind it, and the room would open a second stream in this |
| + // one — a 409 and a takeover offer, for someone who pressed Cancel. |
| + if !strings.Contains(rec.Body.String(), `href="/m/`+sqid+`/join"`) { |
| + t.Fatalf("confirm page must lead back to the holding screen:\n%s", rec.Body.String()) |
| + } |
| + if strings.Contains(rec.Body.String(), `href="/m/`+sqid+`"`) { |
| + t.Fatalf("confirm page still offers the room, which would open a second stream:\n%s", rec.Body.String()) |
| + } |
| // A peer that does not exist in the room at all must not render a |
| // working confirm page: PeerNameFor answers nothing to remove. |
| if rec := m.get("/m/" + sqid + "/remove/" + strings.Repeat("0", 32)); rec.Code != http.StatusNotFound { |
| @@ -254,6 +389,11 @@ func TestTakeoverRequiresStandingNotJustParticipation(t *testing.T) { |
| pg, _ := sg.hello() |
| g.signal(sqid, pg, `{"type":"join"}`) |
| m.signal(sqid, pm, `{"type":"admit","payload":{"peer":"`+pg+`"}}`) |
| + // Admission proved, not assumed: the point of this test is what an |
| + // *admitted* guest may and may not do past the expiry. |
| + if ev, _, ok := sg.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("the guest was never admitted (got %q)", ev) |
| + } |
| x.advance(2*time.Hour + time.Minute) |
| if rec := g.get("/m/" + sqid); rec.Code != http.StatusOK { |
diff --git a/internal/meettest/signal_test.go b/internal/meettest/signal_test.go| index e047781..c6ce4f1 100644 |
| --- a/internal/meettest/signal_test.go |
| +++ b/internal/meettest/signal_test.go |
| @@ -60,9 +60,13 @@ func TestSignalNeedsAPeerBoundToThisSessionAndRoom(t *testing.T) { |
| } |
| // A peer that is perfectly real, and bound to this very session, but in |
| -// a different room, is the same 404: lookups start from the room named |
| -// in the URL (spec §4.2's "cross-room delivery is impossible by |
| -// construction"). |
| +// a different room, is the same 404 — and from Task 5's fix round 2 it |
| +// is refused earlier than it used to be: the LiveForSession gate answers |
| +// it before the limiter, the body read or any hub dispatch, because this |
| +// session holds no participation in the room named in the URL. Either |
| +// way the rule is §4.2's "cross-room delivery is impossible by |
| +// construction": every lookup starts from the room in the URL, so a peer |
| +// id from elsewhere names nothing here. |
| func TestSignalRefusesAPeerFromAnotherMeeting(t *testing.T) { |
| x, sqid, _ := memberWithMeeting(t) |
| c := x.browser(t) |
| @@ -220,7 +224,7 @@ func TestSignalRefusesAnOversizedRoomIDAndNeverLogsIt(t *testing.T) { |
| c.signIn("paul@example.com") |
| huge := strings.Repeat("a", 1024) |
| - resp, _ := c.signalRaw(huge, `{"peer":"0123456789abcdef0123456789abcdef","type":"join"}`) |
| + resp, badRoomBody := c.signalRaw(huge, `{"peer":"0123456789abcdef0123456789abcdef","type":"join"}`) |
| if resp.StatusCode != http.StatusNotFound { |
| t.Fatalf("1 KB room id: %d, want 404", resp.StatusCode) |
| } |
| @@ -230,6 +234,108 @@ func TestSignalRefusesAnOversizedRoomIDAndNeverLogsIt(t *testing.T) { |
| if strings.Contains(buf.String(), huge) { |
| t.Fatal("the oversized room id reached the log") |
| } |
| + |
| + // Byte for byte the not-found body every other refusal here carries. |
| + // A malformed room id is refused earlier and for a different reason |
| + // than a peer that is not yours, and neither of those is anybody's |
| + // business from outside: one answer, or the difference between them |
| + // is a way to ask which meetings exist (§2.5). |
| + sqid, _ := createMeeting(t, c, "A room this session has never opened a stream in", "") |
| + resp, unboundBody := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`) |
| + if resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("a room with no live participation: %d, want 404", resp.StatusCode) |
| + } |
| + if !bytes.Equal(badRoomBody, unboundBody) { |
| + t.Fatalf("404 bodies differ: bad room %q, no participation %q", badRoomBody, unboundBody) |
| + } |
| +} |
| + |
| +// Task 5's fix round 2 ruling, and §4.5's window: the LiveForSession |
| +// gate in front of the limiter is about a session that holds nothing in |
| +// this room, not about a browser whose stream happens to be down. A |
| +// detached participation inside its resume window is still this |
| +// session's peer, so its signalling is still accepted — which is what |
| +// lets a browser post `leave`, or answer an offer, in the gap between |
| +// two streams. Once the window lapses and the tick clears it, the same |
| +// request is the same 404 a stranger gets. |
| +func TestSignalFromADetachedSessionInsideItsWindowIsAccepted(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, _ := s.hello() |
| + if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("join: %d", resp.StatusCode) |
| + } |
| + if ev, _, ok := s.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("never admitted (got %q)", ev) |
| + } |
| + |
| + s.close() // the stream is gone; the participation has thirty seconds |
| + if resp, b := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("signal from a detached session inside its window: %d %s", resp.StatusCode, b) |
| + } |
| + if n := x.ctl.SignalLimiterSize(); n != 1 { |
| + t.Fatalf("limiter buckets: %d, want 1 — the detached session reached the limiter", n) |
| + } |
| + |
| + x.advance(31 * time.Second) |
| + x.ctl.TickForTest() |
| + if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("signal after the window lapsed: %d, want 404", resp.StatusCode) |
| + } |
| +} |
| + |
| +// Task 5's fix round 2 item 3: the 429 line names the peer the hub says |
| +// this session holds, never the one the request claimed. A flood that |
| +// rotates its claimed peer id must leave one log line's worth of truth |
| +// behind — otherwise the rate-limit log is written by the attacker. |
| +func TestTheRateLimitLogNamesTheSessionsRealPeer(t *testing.T) { |
| + var buf bytes.Buffer |
| + x := newHarnessWithLogger(t, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com") |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + sqid, _ := createMeeting(t, c, "x", "") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, _ := s.hello() |
| + c.signal(sqid, peer, `{"type":"join"}`) |
| + |
| + var claimed string |
| + var got429 bool |
| + for i := 0; i < 100; i++ { |
| + claimed = fmt.Sprintf("%032x", i) |
| + resp, _ := c.signal(sqid, claimed, `{"type":"join"}`) |
| + if resp.StatusCode == http.StatusTooManyRequests { |
| + got429 = true |
| + break |
| + } |
| + } |
| + if !got429 { |
| + t.Fatal("the burst never hit the rate limit") |
| + } |
| + |
| + var rateLine string |
| + for _, line := range strings.Split(buf.String(), "\n") { |
| + if strings.Contains(line, "outcome=rate") { |
| + rateLine = line |
| + break |
| + } |
| + } |
| + if rateLine == "" { |
| + t.Fatalf("no rate-limit line in the log:\n%s", buf.String()) |
| + } |
| + if !strings.Contains(rateLine, "peer="+peer) { |
| + t.Fatalf("the 429 line does not name this session's own peer: %s", rateLine) |
| + } |
| + if strings.Contains(rateLine, "peer="+claimed) { |
| + t.Fatalf("the 429 line names the peer the request claimed: %s", rateLine) |
| + } |
| + // And nothing else: type is unknown, because the body was never |
| + // read (the limiter runs first), and the room is the one in the URL. |
| + if !strings.Contains(rateLine, "type=unknown") || !strings.Contains(rateLine, "room="+sqid) { |
| + t.Fatalf("the 429 line: %s", rateLine) |
| + } |
| } |
| // Task 5 review minor 8: the body shapes signal.go must classify |
diff --git a/internal/meettest/sse_test.go b/internal/meettest/sse_test.go| index 3a21940..a4e1d83 100644 |
| --- a/internal/meettest/sse_test.go |
| +++ b/internal/meettest/sse_test.go |
| @@ -6,9 +6,11 @@ import ( |
| "context" |
| "encoding/json" |
| "io" |
| + "net" |
| "net/http" |
| "strings" |
| "sync" |
| + "syscall" |
| "testing" |
| "time" |
| ) |
| @@ -205,6 +207,125 @@ func (s *sseReader) close() { |
| }) |
| } |
| +// rawStream is the stream with nobody draining it in the background — |
| +// the one thing sseReader cannot be, because its read goroutine is |
| +// always reading. A test that wants the server's own 64-event |
| +// subscriber buffer to overflow (spec §4.4) has to stop reading the |
| +// socket: that is what backs the handler's write up, and only a handler |
| +// that is not draining lets the buffer behind it fill. |
| +// |
| +// Nothing here parses more of the wire format than the tests need: an |
| +// event name, its data, and heartbeat comments skipped. |
| +type rawStream struct { |
| + t *testing.T |
| + cancel context.CancelFunc |
| + body io.ReadCloser |
| + r *bufio.Reader |
| + once sync.Once |
| +} |
| + |
| +// tinyWindowClient dials with a deliberately small receive buffer. It is |
| +// what makes "the client stopped reading" mean something: with the |
| +// kernel's own defaults, loopback will happily absorb megabytes on the |
| +// reader's behalf, and the server's write never blocks, so the hub's |
| +// buffer never fills and the test proves nothing. SO_RCVBUF also turns |
| +// receive-window autotuning off, which is the point — the window must |
| +// stay small rather than growing to meet the flood. |
| +func tinyWindowClient() *http.Client { |
| + d := &net.Dialer{ |
| + Control: func(network, address string, c syscall.RawConn) error { |
| + var setErr error |
| + if err := c.Control(func(fd uintptr) { |
| + setErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_RCVBUF, 4096) |
| + }); err != nil { |
| + return err |
| + } |
| + return setErr |
| + }, |
| + } |
| + return &http.Client{Transport: &http.Transport{DialContext: d.DialContext}} |
| +} |
| + |
| +func (c *client) rawStream(target string) *rawStream { |
| + c.t.Helper() |
| + srv := c.x.server() |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+target, nil) |
| + if err != nil { |
| + cancel() |
| + c.t.Fatal(err) |
| + } |
| + for _, ck := range c.cookies { |
| + req.AddCookie(ck) |
| + } |
| + req.Header.Set("Accept", "text/event-stream") |
| + resp, err := tinyWindowClient().Do(req) |
| + if err != nil { |
| + cancel() |
| + c.t.Fatalf("GET %s: %v", target, err) |
| + } |
| + if resp.StatusCode != http.StatusOK { |
| + b, _ := io.ReadAll(resp.Body) |
| + resp.Body.Close() |
| + cancel() |
| + c.t.Fatalf("GET %s: %d\n%s", target, resp.StatusCode, b) |
| + } |
| + // A small reader on purpose: the socket must stop being drained the |
| + // moment the test stops calling next, not a buffer's worth later. |
| + s := &rawStream{t: c.t, cancel: cancel, body: resp.Body, r: bufio.NewReaderSize(resp.Body, 512)} |
| + c.t.Cleanup(s.close) |
| + return s |
| +} |
| + |
| +// next reads one event, blocking until the server sends one or the |
| +// timeout cancels the request out from under the read. |
| +func (s *rawStream) next(timeout time.Duration) (string, []byte, bool) { |
| + s.t.Helper() |
| + stop := time.AfterFunc(timeout, s.cancel) |
| + defer stop.Stop() |
| + var name string |
| + var data []byte |
| + for { |
| + raw, err := s.r.ReadString('\n') |
| + if err != nil { |
| + return "", nil, false |
| + } |
| + line := strings.TrimRight(raw, "\r\n") |
| + switch { |
| + case line == "": |
| + if name != "" || data != nil { |
| + return name, data, true |
| + } |
| + case strings.HasPrefix(line, ":"): // heartbeat |
| + case strings.HasPrefix(line, "event:"): |
| + name = field(line, "event:") |
| + case strings.HasPrefix(line, "data:"): |
| + data = append(data, field(line, "data:")...) |
| + } |
| + } |
| +} |
| + |
| +// hello is next's first event, which is always hello (§4.1). |
| +func (s *rawStream) hello() string { |
| + s.t.Helper() |
| + name, data, ok := s.next(2 * time.Second) |
| + if !ok || name != "hello" { |
| + s.t.Fatalf("first event is %q (ok=%v), want hello", name, ok) |
| + } |
| + var body struct{ Peer string } |
| + if err := json.Unmarshal(data, &body); err != nil { |
| + s.t.Fatalf("hello %s: %v", data, err) |
| + } |
| + return body.Peer |
| +} |
| + |
| +func (s *rawStream) close() { |
| + s.once.Do(func() { |
| + s.cancel() |
| + s.body.Close() |
| + }) |
| +} |
| + |
| // signal is POST /m/{sqid}/signal, the upstream half of §4.1: a JSON |
| // body carrying the peer and the message, under CSRF. body is decoded |
| // into a map and given "peer", rather than spliced in as a string — a |
diff --git a/internal/meettest/stream_test.go b/internal/meettest/stream_test.go| index edf5528..c983ce8 100644 |
| --- a/internal/meettest/stream_test.go |
| +++ b/internal/meettest/stream_test.go |
| @@ -352,6 +352,245 @@ func TestAnOversizeLastIsRefusedBeforeItIsParsed(t *testing.T) { |
| } |
| } |
| +// Spec §4.4: a subscriber that falls sixty-four events behind is |
| +// declared lost and told so, and §4.5 makes lost{buffer} the one reason |
| +// a browser answers with a resume instead of a screen. This is that path |
| +// over real HTTP rather than against hub.Sub directly: a client that |
| +// stops reading its socket backs the handler's write up, the hub's |
| +// sixty-four-slot buffer fills behind the stalled handler, and the |
| +// participation itself survives — so the resume that follows gets the |
| +// same peer back. |
| +func TestASlowClientIsLostOnTheBufferAndResumesTheSameParticipation(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + sender := x.browser(t) |
| + sender.signIn("paul@example.com") |
| + ss := sender.stream("/m/" + sqid + "/stream") |
| + defer ss.close() |
| + from, _ := ss.hello() |
| + if resp, _ := sender.signal(sqid, from, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("sender join: %d", resp.StatusCode) |
| + } |
| + |
| + // The victim reads its hello and then stops reading altogether. |
| + victim := newMemberBrowser(t, x, "paul@rslw.com") |
| + vs := victim.rawStream("/m/" + sqid + "/stream") |
| + defer vs.close() |
| + to := vs.hello() |
| + if resp, _ := victim.signal(sqid, to, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("victim join: %d", resp.StatusCode) |
| + } |
| + if ev, _, ok := ss.nextOf("peer", 2*time.Second); !ok { |
| + t.Fatalf("the sender never saw the victim admitted (got %q); nothing below would reach it", ev) |
| + } |
| + |
| + // The arithmetic, because it is what makes this deterministic rather |
| + // than lucky. Two things absorb events before the hub's own buffer |
| + // starts filling: the socket, and the handler's one in hand. The |
| + // socket's share is bounded by the kernel's send buffer — about |
| + // 1.8 MiB on a default Linux (tcp_wmem's 4 MiB ceiling, minus skb |
| + // overhead), which at 60 KiB an event is roughly thirty of them — |
| + // and the receive side is pinned small by rawStream's own dial. The |
| + // buffer is sixty-four. So a flood well past ninety-five is past |
| + // every one of those, and the count below leaves room for a kernel |
| + // twice as generous as this one. |
| + // |
| + // The limiter is thirty a second with a burst of sixty on the |
| + // harness's frozen clock, so the burst is spent in batches with a |
| + // refill between them rather than being throttled halfway through. |
| + const flood = 160 |
| + payload := strings.Repeat("x", 60*1024) |
| + for i := 1; i <= flood; i++ { |
| + if i%50 == 1 && i > 1 { |
| + x.advance(2 * time.Second) // refills the whole burst |
| + } |
| + body := fmt.Sprintf(`{"type":"candidate","to":%q,"seq":%d,"gen":1,"payload":{"c":%q}}`, to, i, payload) |
| + if resp, b := sender.signal(sqid, from, body); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("candidate %d: %d %s", i, resp.StatusCode, b) |
| + } |
| + } |
| + |
| + // Now the victim starts reading again. Whatever was buffered comes |
| + // through first; the stream ends on lost{buffer}. |
| + var seen int |
| + for { |
| + name, data, ok := vs.next(30 * time.Second) |
| + if !ok { |
| + t.Fatalf("the stream ended without lost{buffer} after %d events", seen) |
| + } |
| + seen++ |
| + if name != "lost" { |
| + continue |
| + } |
| + var lost struct{ Reason string } |
| + json.Unmarshal(data, &lost) |
| + if lost.Reason != "buffer" { |
| + t.Fatalf("lost %s, want reason buffer", data) |
| + } |
| + break |
| + } |
| + // Events were actually lost, which is the whole claim: a stream that |
| + // delivered every one of them and then said lost{buffer} would be |
| + // something else entirely. |
| + if seen > flood { |
| + t.Fatalf("the victim read %d of %d events and lost none of them", seen, flood) |
| + } |
| + |
| + // The participation is not the subscriber: it is still there, and a |
| + // resume gets the same peer and the roster back (§4.5). |
| + vs.close() |
| + vs2 := victim.stream("/m/" + sqid + "/stream?peer=" + to) |
| + defer vs2.close() |
| + if p, _ := vs2.hello(); p != to { |
| + t.Fatalf("resume after lost{buffer}: %s, want %s", p, to) |
| + } |
| + if ev, _, ok := vs2.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("the resumed peer is not admitted any more (got %q)", ev) |
| + } |
| +} |
| + |
| +// Spec §2.4's ruling 14, end to end: an admitted guest is never ended by |
| +// their grant lapsing, because the participation is the authority once |
| +// they are in. So signal keeps working, the stream keeps resuming — and |
| +// fresh entry, which is standing's alone, is refused. |
| +func TestAnAdmittedGuestCarriesOnPastGrantExpiryButGetsNoFreshEntry(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + m := x.browser(t) |
| + m.signIn("paul@example.com") |
| + sm := m.stream("/m/" + sqid + "/stream") |
| + defer sm.close() |
| + pm, _ := sm.hello() |
| + m.signal(sqid, pm, `{"type":"join"}`) |
| + |
| + g := x.browser(t) |
| + g.get("/j/" + tok) |
| + g.signIn("guest@example.com") |
| + sg := g.stream("/m/" + sqid + "/stream") |
| + defer sg.close() |
| + pg, _ := sg.hello() |
| + g.signal(sqid, pg, `{"type":"join"}`) |
| + m.signal(sqid, pm, `{"type":"admit","payload":{"peer":"`+pg+`"}}`) |
| + // Proof of admission before anything is claimed about it: a waiting |
| + // guest would also survive most of what follows, and would prove |
| + // nothing about ruling 14. |
| + if ev, _, ok := sg.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("the guest was never admitted (got %q)", ev) |
| + } |
| + |
| + x.advance(2*time.Hour + time.Minute) // the grant is over |
| + |
| + // signal: still theirs. join is idempotent for an admitted peer, and |
| + // a targeted relay is something only an admitted peer may send, so |
| + // both are worth asking. |
| + if resp, b := g.signal(sqid, pg, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("join past grant expiry: %d %s", resp.StatusCode, b) |
| + } |
| + if resp, b := g.signal(sqid, pg, `{"type":"offer","to":"`+pm+`","seq":1,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("offer past grant expiry: %d %s", resp.StatusCode, b) |
| + } |
| + if ev, _, ok := sm.nextOf("offer", 2*time.Second); !ok { |
| + t.Fatalf("the member never received the expired guest's offer (got %q)", ev) |
| + } |
| + |
| + // stream?peer=: still theirs, roster and all. |
| + sg.close() |
| + sg2 := g.stream("/m/" + sqid + "/stream?peer=" + pg) |
| + defer sg2.close() |
| + if p, _ := sg2.hello(); p != pg { |
| + t.Fatalf("resume past grant expiry: %s, want %s", p, pg) |
| + } |
| + if ev, _, ok := sg2.nextOf("roster", 2*time.Second); !ok { |
| + t.Fatalf("the resumed guest is not admitted any more (got %q)", ev) |
| + } |
| + |
| + // GET /m/{sqid}/ice is slice 3's, and is not mounted yet, so it |
| + // answers chi's own 404 for everybody. When slice 3 mounts it, this |
| + // line becomes the assertion that the expired guest is served like |
| + // every other admitted peer — the endpoint is named in §13's "on |
| + // every endpoint" and this is where it lands. |
| + if rec := g.get("/m/" + sqid + "/ice"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("GET /ice: %d — slice 3 has mounted it; give this test its real arm", rec.Code) |
| + } |
| + |
| + // Fresh entry is not theirs: a stream with no peer is authorized by |
| + // standing, and standing is gone. 404, never 403 (§2.5). |
| + if rec := g.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("fresh stream past grant expiry: %d, want 404", rec.Code) |
| + } |
| + // It is standing that refused that, not the one-peer-per-session |
| + // rule: this session's peer is refused with 409 while it holds one, |
| + // and 409 is what this would be if the standing check had passed. |
| + // (A second browser for the same guest cannot be used to make the |
| + // point: signing one in needs the join link again, and following the |
| + // link mints a fresh grant, which is the very thing that has to be |
| + // gone here.) |
| + // |
| + // The holding screen is *not* refused: roomAccess is standing OR a |
| + // live participation (§2.5), and ruling 14 keeps this one's |
| + // participation. TestRoomPageSurvivesGrantExpiryForAnAdmittedGuest |
| + // is the room page's half of the same rule. |
| + if rec := g.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK { |
| + t.Fatalf("holding screen past grant expiry: %d, want 200", rec.Code) |
| + } |
| +} |
| + |
| +// Spec §2.4 and §13's "a member dropped on the tick", over HTTP: the |
| +// team changes, the next tick re-derives standing for every joined |
| +// participation, and the member who is no longer on it loses the room |
| +// while the room is told they are gone. |
| +func TestAMemberDroppedFromTheAllowlistIsLostOnTheTick(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + a := x.browser(t) |
| + a.signIn("paul@example.com") |
| + sa := a.stream("/m/" + sqid + "/stream") |
| + defer sa.close() |
| + pa, _ := sa.hello() |
| + a.signal(sqid, pa, `{"type":"join"}`) |
| + |
| + b := newMemberBrowser(t, x, "paul@rslw.com") |
| + sb := b.stream("/m/" + sqid + "/stream") |
| + defer sb.close() |
| + pb, _ := sb.hello() |
| + b.signal(sqid, pb, `{"type":"join"}`) |
| + if ev, _, ok := sa.nextOf("peer", 2*time.Second); !ok { |
| + t.Fatalf("the two members never met (got %q)", ev) |
| + } |
| + |
| + x.ctl.SetAllowlist("paul@example.com") // B is off the team |
| + x.ctl.TickForTest() |
| + |
| + ev, data, ok := sb.nextOf("lost", 2*time.Second) |
| + if !ok { |
| + t.Fatalf("the dropped member's stream stayed open (got %q)", ev) |
| + } |
| + var lost struct{ Reason string } |
| + json.Unmarshal(data, &lost) |
| + if lost.Reason != "standing" { |
| + t.Fatalf("dropped member got lost %s, want reason standing", data) |
| + } |
| + ev, data, ok = sa.nextOf("gone", 2*time.Second) |
| + if !ok { |
| + t.Fatalf("the room was never told the member left (got %q)", ev) |
| + } |
| + var gone struct{ From string } |
| + json.Unmarshal(data, &gone) |
| + if gone.From != pb { |
| + t.Fatalf("gone names %q, want %q", gone.From, pb) |
| + } |
| + |
| + // Their participation is gone, and the tombstone says why (§4.5). |
| + rec := b.getStream("/m/" + sqid + "/stream?peer=" + pb) |
| + if _, _, reason := answer(t, rec); rec.Code != http.StatusGone || reason != "standing" { |
| + t.Fatalf("resume after the drop: %d %q, want 410 standing", rec.Code, reason) |
| + } |
| + // And so is entry: no standing, no fresh stream, no room page. |
| + if rec := b.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("fresh stream after the drop: %d, want 404", rec.Code) |
| + } |
| + if rec := b.get("/m/" + sqid); rec.Code != http.StatusNotFound { |
| + t.Fatalf("room page after the drop: %d, want 404", rec.Code) |
| + } |
| +} |
| + |
| // A SIGTERM cancels the context the app was built on. rastrillo answers |
| // the same signal with a ten-second Shutdown, and Shutdown waits for |
| // handlers without cancelling their contexts, so a live stream that |