amadan

oficina / meet Public

hub: standing on every member request, and two locks

S2-1 (P1): Relay, Reset, Resume and CanResume re-derive standing for a
participation whose kind is member, per §2.5's paragraph — the row that
said standing is not consulted is corrected in §15.12. The read happens
outside the hub lock; the participation is re-found under it and applied
through standingLostLocked, so an admitted guest is still exempt
(ruling 14). A member with no standing is ended with lost{standing} and
answered 404, byte-identical to the unknown-peer body. CanResume stays
non-mutating: it answers and leaves the ending to the next mutating
request or the tick.

S2-2 (P2): Open ends a participation whose resume window has passed
instead of refusing the fresh stream 409 until the tick, and
LiveForSession/LiveKindForSession no longer count one — so a browser
refused a resume is not then told the meeting is open in another tab.

S2-4 (P2): Sub.terminate's eviction receive is non-blocking and the
final send is retried once. The old unconditional receive could block on
a channel the reader had drained, with h.mu held.
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 3c8c6ac090638c8e2d724fd0246bc22a516f5e2a parent 67d0ead
11 files changed, +551 −70
  • internal/meet/hub/admission.go +59 −0
  • internal/meet/hub/hub.go +43 −3
  • internal/meet/hub/hub_test.go +7 −1
  • internal/meet/hub/relay.go +68 −5
  • internal/meet/hub/relay_test.go +74 −51
  • internal/meet/hub/subscriber.go +24 −5
  • internal/meet/hub/subscriber_test.go +50 −0
  • internal/meet/signal.go +2 −2
  • internal/meet/stream.go +3 −3
  • internal/meettest/standing_test.go +183 −0
  • internal/meettest/stream_test.go +38 −0
diff --git a/internal/meet/hub/admission.go b/internal/meet/hub/admission.go
index 544949c..160b281 100644
--- a/internal/meet/hub/admission.go
+++ b/internal/meet/hub/admission.go
@@ -134,6 +134,65 @@ func (h *Hub) standingLostLocked(r *room, p *Participation) bool {
return true
}
+// memberStandingGone re-derives standing for a participation whose kind
+// is `member`, which spec §2.5's paragraph — and §15.12, which corrects
+// the table row above it — require on **every** request that
+// participation makes, not only on the tick: "a member who has just been
+// removed is dropped by the next thing they do, not the next tick."
+// Relay, Reset, Resume and CanResume all go through here first.
+//
+// A guest is never touched and never costs a standing read: ruling 14
+// makes the participation the authority once someone is admitted, and a
+// waiting guest is the tick's to expire.
+//
+// end says whether this call may write. The mutating callers (Relay,
+// Reset, Resume) end a member who has lost standing here and now, with
+// lost{standing} and the departure the room is owed; the probe
+// (CanResume) passes false, answers, and leaves the ending to the next
+// mutating request or to the tick — a probe is defined as changing
+// nothing (§15.3).
+//
+// The standing read happens with the lock down, like every other one
+// (StandingFor is a database read), so the participation is re-found
+// under the lock afterwards: it may have ended while the read was in
+// flight.
+func (h *Hub) memberStandingGone(ctx context.Context, sqid, session string, peer PeerID, end bool) (bool, error) {
+ h.mu.Lock()
+ p, found := h.findLocked(sqid, session, peer)
+ if !found || p.Kind != KindMember {
+ // Not found is the caller's own lookup to answer, in its own
+ // words; a guest is not this function's business at all.
+ h.mu.Unlock()
+ return false, nil
+ }
+ personID := p.PersonID
+ h.mu.Unlock()
+
+ kind, ok, err := h.cfg.StandingFor(ctx, sqid, personID)
+ if err != nil {
+ return false, err
+ }
+
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ p, found = h.findLocked(sqid, session, peer)
+ if !found {
+ return false, nil
+ }
+ r := h.rooms[sqid]
+ if !ok {
+ if !end {
+ // The kind is read again rather than remembered: the tick may
+ // have made this an admitted guest while the read was out, and
+ // that one keeps everything (ruling 14).
+ return p.Kind == KindMember, nil
+ }
+ return h.standingLostLocked(r, p), nil
+ }
+ h.setKindLocked(r, p, kind)
+ return false, nil
+}
+
// 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)
diff --git a/internal/meet/hub/hub.go b/internal/meet/hub/hub.go
index b320f20..1d0ed84 100644
--- a/internal/meet/hub/hub.go
+++ b/internal/meet/hub/hub.go
@@ -184,7 +184,21 @@ func (h *Hub) Open(sqid, session string, personID int64, name string, kind Kind)
return OpenResult{}, ErrUnknownPeer
}
if existing, ok := h.bySession[sqid][session]; ok {
- return OpenResult{Existing: existing}, ErrSessionHasPeer
+ // A participation whose resume window has passed is not a
+ // conflict: §15.7 refuses its resume the moment the window
+ // passes, so answering 409 "open in another tab" until the tick
+ // caught up told a browser that had just been refused a resume
+ // that it was competing with itself (Codex S2-2). It is ended
+ // here, on the same terms the tick would have ended it on —
+ // nothing can be listening for the lost event, and the room is
+ // owed the departure.
+ if p := h.lapsedLocked(r, existing); p != nil {
+ wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer)
+ delete(h.tombstones, p.Peer)
+ h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
+ } else {
+ return OpenResult{Existing: existing}, ErrSessionHasPeer
+ }
}
if r != nil && len(r.peers) >= MaxParticipations {
return OpenResult{}, ErrRoomFull
@@ -226,13 +240,39 @@ func (h *Hub) findLocked(sqid, session string, peer PeerID) (*Participation, boo
return p, true
}
+// lapsedLocked is the participation this room holds under peer, when it
+// is detached and past its resume window — the state §15.7 has already
+// refused a resume in, and which only the tick used to clear. It is nil
+// for a live one, and for a peer this room does not hold.
+func (h *Hub) lapsedLocked(r *room, peer PeerID) *Participation {
+ if r == nil {
+ return nil
+ }
+ p, ok := r.peers[peer]
+ if !ok {
+ return nil
+ }
+ if p.sub != nil || p.resumeUntil.IsZero() || !h.cfg.Now().After(p.resumeUntil) {
+ return nil
+ }
+ return p
+}
+
// LiveForSession names the peer this session holds in the room, whether
// attached or within its resume window (spec §4.2).
+//
+// Past that window it holds nothing: the participation is still in the
+// hub until the tick, but a resume of it is already refused (§15.7) and
+// a fresh stream now ends it and takes its place (Open), so nothing that
+// asks "does this session hold a peer here" may still be told yes.
func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) {
h.mu.Lock()
defer h.mu.Unlock()
id, ok := h.bySession[sqid][session]
- return id, ok
+ if !ok || h.lapsedLocked(h.rooms[sqid], id) != nil {
+ return "", false
+ }
+ return id, true
}
// LiveKindForSession is LiveForSession plus the kind that participation
@@ -248,7 +288,7 @@ 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 {
+ if !ok || h.lapsedLocked(h.rooms[sqid], id) != nil {
return "", 0, false
}
p, ok := h.findLocked(sqid, session, id)
diff --git a/internal/meet/hub/hub_test.go b/internal/meet/hub/hub_test.go
index 59dee12..f7c3c05 100644
--- a/internal/meet/hub/hub_test.go
+++ b/internal/meet/hub/hub_test.go
@@ -8,6 +8,12 @@ import (
"time"
)
+// testCtx is the context the four request-path methods now take, so a
+// member's standing can be re-derived on every request (spec §15.12).
+// Nothing in these tests cancels it: the standing they exercise comes
+// from testHub's StandingFor, not from the context.
+var testCtx = context.Background()
+
func testHub(now *time.Time) *Hub {
return New(Config{
Now: func() time.Time { return *now },
@@ -151,7 +157,7 @@ func TestDetachOfADisplacedStreamLeavesTheReplacementAttached(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
- res, err := h.Resume("r1", "sa", a.Peer, nil)
+ res, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/meet/hub/relay.go b/internal/meet/hub/relay.go
index 081115c..7e5cdc7 100644
--- a/internal/meet/hub/relay.go
+++ b/internal/meet/hub/relay.go
@@ -1,6 +1,7 @@
package hub
import (
+ "context"
"fmt"
"time"
)
@@ -9,10 +10,24 @@ func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t =
// 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 {
+//
+// A member's standing is re-derived first (§15.12): the participation
+// authorizes everything after entry (§2.4), but a member who has lost
+// the team is dropped by the next thing they do, and this is that thing.
+// A member with no standing is ended here and the request answers
+// ErrNotYours, which the handler renders as the same not-found body an
+// unknown peer gets.
+func (h *Hub) Relay(ctx context.Context, sqid, session string, from PeerID, m Message) error {
if !isTargeted(m.Type) {
return fmt.Errorf("%w: %q", ErrNotTargeted, m.Type)
}
+ gone, err := h.memberStandingGone(ctx, sqid, session, from, true)
+ if err != nil {
+ return err
+ }
+ if gone {
+ return ErrNotYours
+ }
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
@@ -65,8 +80,16 @@ func (h *Hub) ringLocked(p *Participation, m Message) {
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 {
+// Reset bumps the pair's generation and tells both sides. A member's
+// standing is re-derived first, exactly as Relay does (§15.12).
+func (h *Hub) Reset(ctx context.Context, sqid, session string, from, to PeerID) error {
+ gone, err := h.memberStandingGone(ctx, sqid, session, from, true)
+ if err != nil {
+ return err
+ }
+ if gone {
+ return ErrNotYours
+ }
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
@@ -159,18 +182,58 @@ func (h *Hub) resumeRefusalLocked(sqid, session string, peer PeerID) *ResumeErro
// does not read or trim the ring. It is the probe behind
// GET /m/{sqid}/stream?peer=&probe=1, which exists because EventSource
// hides the status code of a refused stream from the page.
-func (h *Hub) CanResume(sqid, session string, peer PeerID) *ResumeError {
+// It re-derives a member's standing (§15.12) and answers 404 when it has
+// gone — the same body an unknown peer gets — but it does not end the
+// participation: that is the next mutating request's, or the tick's. A
+// probe changes nothing, and a browser that reads 404 here goes on to
+// open a fresh stream, which is what a dropped member should get.
+func (h *Hub) CanResume(ctx context.Context, sqid, session string, peer PeerID) *ResumeError {
+ gone, err := h.memberStandingGone(ctx, sqid, session, peer, false)
+ if err != nil {
+ // A standing read that failed is not evidence of anything. The
+ // participation answers for itself, as it did before §15.12, and
+ // the tick will re-read standing in fifteen seconds.
+ h.cfg.Logger.Warn("hub: standing lookup failed on a resume probe", "room", sqid, "peer", peer, "error", err)
+ }
h.mu.Lock()
defer h.mu.Unlock()
+ if gone {
+ return &ResumeError{Status: 404, Boot: h.boot}
+ }
return h.resumeRefusalLocked(sqid, session, peer)
}
// 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) {
+//
+// A member's standing is re-derived before anything is attached
+// (§15.12). The refusal ladder runs first, so a participation that
+// ended for a reason worth naming still answers 410 with that reason
+// rather than spending a database read; a member whose standing has
+// gone since is ended here and answered 404, which is the answer an
+// unknown peer gets, byte for byte.
+func (h *Hub) Resume(ctx context.Context, sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) {
+ h.mu.Lock()
+ if re := h.resumeRefusalLocked(sqid, session, peer); re != nil {
+ h.mu.Unlock()
+ return ResumeResult{}, re
+ }
+ h.mu.Unlock()
+
+ gone, err := h.memberStandingGone(ctx, sqid, session, peer, true)
+ if err != nil {
+ return ResumeResult{}, err
+ }
+ if gone {
+ return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot}
+ }
+
h.mu.Lock()
defer h.mu.Unlock()
+ // The lock was down for the standing read, so the ladder is walked
+ // again: the window may have passed, or the meeting ended, while it
+ // was out.
if re := h.resumeRefusalLocked(sqid, session, peer); re != nil {
return ResumeResult{}, re
}
diff --git a/internal/meet/hub/relay_test.go b/internal/meet/hub/relay_test.go
index 2a762e4..bddae33 100644
--- a/internal/meet/hub/relay_test.go
+++ b/internal/meet/hub/relay_test.go
@@ -29,7 +29,7 @@ func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T
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"}`)})
+ err := h.Relay(testCtx, "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)
}
@@ -43,19 +43,19 @@ func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T
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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Relay(testCtx, "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}); !errors.Is(err, ErrNotTargeted) {
+ if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeChat, To: b.Peer, Seq: 4, Gen: 1}); !errors.Is(err, ErrNotTargeted) {
t.Fatalf("chat as a targeted relay: %v", err)
}
}
@@ -65,10 +65,10 @@ func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T
// itself, which no roster ever names.
func TestAPeerIsNotItsOwnTarget(t *testing.T) {
h, _, a, _ := twoAdmitted(t)
- if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrUnknownPeer {
+ if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrUnknownPeer {
t.Fatalf("self-addressed relay: %v", err)
}
- if err := h.Reset("r1", "sa", a.Peer, a.Peer); err != ErrUnknownPeer {
+ if err := h.Reset(testCtx, "r1", "sa", a.Peer, a.Peer); err != ErrUnknownPeer {
t.Fatalf("self-addressed reset: %v", err)
}
if ev := drain(a.Sub); len(ev) != 0 {
@@ -82,13 +82,13 @@ func TestAPeerIsNotItsOwnTarget(t *testing.T) {
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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Reset(testCtx, "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
+ if err := h.Reset(testCtx, "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)
@@ -98,10 +98,10 @@ func TestStaleGenIsRefusedAndResetConverges(t *testing.T) {
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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil {
t.Fatal(err)
}
}
@@ -110,7 +110,7 @@ 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(`{}`)})
+ h.Relay(testCtx, "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, nil)
@@ -122,11 +122,11 @@ func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) {
// 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 {
+ if err := h.Relay(testCtx, "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})
+ res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3})
if err != nil {
t.Fatal(err)
}
@@ -157,7 +157,7 @@ func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) {
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
h.Detach("r1", g.Peer, nil)
- gres, err := h.Resume("r1", "sg", g.Peer, nil)
+ gres, err := h.Resume(testCtx, "r1", "sg", g.Peer, nil)
if err != nil {
t.Fatal(err)
}
@@ -169,12 +169,12 @@ func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) {
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})
+ h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
}
drain(a.Sub)
h.Detach("r1", a.Peer, nil)
*now = now.Add(time.Second)
- res, _ := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
+ res, _ := h.Resume(testCtx, "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)
}
@@ -190,15 +190,15 @@ func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) {
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})
+ h.Relay(testCtx, "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})
+ h.Relay(testCtx, "r1", "sc", c.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
}
drain(a.Sub)
h.Detach("r1", a.Peer, nil)
*now = now.Add(time.Second)
- res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 1, c.Peer: RingSize})
+ res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 1, c.Peer: RingSize})
if err != nil {
t.Fatal(err)
}
@@ -215,7 +215,7 @@ func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) {
func TestResumeNamesNoPeerThatHasLeft(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})
+ h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
}
p, _ := h.Find("r1", "sa", a.Peer)
if p.evicted[b.Peer] == 0 {
@@ -227,7 +227,7 @@ func TestResumeNamesNoPeerThatHasLeft(t *testing.T) {
drain(a.Sub)
h.Detach("r1", a.Peer, nil)
*now = now.Add(time.Second)
- res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
+ res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
if err != nil {
t.Fatal(err)
}
@@ -243,7 +243,7 @@ func TestResumeNamesNoPeerThatHasLeft(t *testing.T) {
// 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)
+ res, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
if err != nil {
t.Fatal(err)
}
@@ -260,32 +260,32 @@ func TestResumeReplacesAnExistingAttachment(t *testing.T) {
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 {
+ if err := h.Relay(testCtx, "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 {
+ if _, err := h.Resume(testCtx, "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() {
+ if _, err := h.Resume(testCtx, "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 {
+ if _, err := h.Resume(testCtx, "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 {
+ if _, err := h.Resume(testCtx, "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 {
+ if _, err := h.Resume(testCtx, "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 {
@@ -307,12 +307,12 @@ func TestATombstoneIsAnsweredOnlyThroughItsOwnRoomAndSession(t *testing.T) {
}
// Ann's own session, in the room she was removed from.
- _, err := h.Resume("r1", "sa", a.Peer, nil)
+ _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
re, ok := err.(*ResumeError)
if !ok || re.Status != 410 || re.Reason != ReasonRemoved {
t.Fatalf("her own resume: %v", err)
}
- if re := h.CanResume("r1", "sa", a.Peer); re == nil || re.Status != 410 || re.Reason != ReasonRemoved {
+ if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re == nil || re.Status != 410 || re.Reason != ReasonRemoved {
t.Fatalf("her own probe: %v", re)
}
@@ -322,12 +322,12 @@ func TestATombstoneIsAnsweredOnlyThroughItsOwnRoomAndSession(t *testing.T) {
{"another room this session is in", "r2", "sother"},
{"another room and another session", "r2", "sb"},
} {
- _, err := h.Resume(probe.sqid, probe.session, a.Peer, nil)
+ _, err := h.Resume(testCtx, probe.sqid, probe.session, a.Peer, nil)
re, ok := err.(*ResumeError)
if !ok || re.Status != 404 || re.Boot != h.Boot() || re.Reason != "" {
t.Fatalf("%s: %v", probe.what, err)
}
- if re := h.CanResume(probe.sqid, probe.session, a.Peer); re == nil || re.Status != 404 || re.Reason != "" {
+ if re := h.CanResume(testCtx, probe.sqid, probe.session, a.Peer); re == nil || re.Status != 404 || re.Reason != "" {
t.Fatalf("%s, probed: %+v", probe.what, re)
}
}
@@ -338,7 +338,7 @@ func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) {
h.Detach("r1", a.Peer, nil)
*now = now.Add(ResumeWindow + time.Second)
h.Tick(context.Background())
- _, err := h.Resume("r1", "sa", a.Peer, nil)
+ _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
re, ok := err.(*ResumeError)
if !ok || re.Status != 404 || re.Boot != h.Boot() {
t.Fatalf("lapsed resume: %v", err)
@@ -347,37 +347,60 @@ func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) {
// Spec §4.5's window is thirty seconds, not thirty plus however long it
// is until the next tick: a resume that arrives after resume_until is
-// refused on arrival. The participation is still the tick's to remove,
-// so a refusal never has a room's events as a side effect.
-func TestAResumePastItsWindowIsRefusedBeforeTheTickRemovesIt(t *testing.T) {
+// refused on arrival (§15.7). The refusal itself still writes nothing —
+// a probe changes nothing — but the fresh stream the browser opens next
+// must not be refused 409 by the very participation it was just refused
+// a resume of (Codex S2-2): a browser told "not here" and then "open in
+// another tab" about the same peer has nowhere to go and shows a
+// takeover screen for a tab that does not exist. Open ends it and takes
+// its place.
+func TestAResumePastItsWindowIsRefusedAndAFreshStreamTakesItsPlace(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
h.Detach("r1", a.Peer, a.Sub)
now = now.Add(ResumeWindow - time.Second)
- if re := h.CanResume("r1", "sa", a.Peer); re != nil {
+ if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re != nil {
t.Fatalf("inside the window: %+v", re)
}
now = now.Add(2 * time.Second) // past resume_until, before any tick
- re := h.CanResume("r1", "sa", a.Peer)
+ re := h.CanResume(testCtx, "r1", "sa", a.Peer)
if re == nil || re.Status != 404 || re.Boot != h.Boot() {
t.Fatalf("past the window: %+v", re)
}
- if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil {
+ if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil {
t.Fatal("Resume attached a participation past its window")
}
- // Refused, but still the tick's to remove and still this session's
- // one peer in the room until then.
+ // Refused, and the refusal itself removed nothing: a probe and a
+ // refused resume are both defined as changing nothing.
if _, ok := h.Find("r1", "sa", a.Peer); !ok {
t.Fatal("the refusal removed the participation")
}
- if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != ErrSessionHasPeer {
+ // Nor does the session still hold a peer here as far as anything
+ // that asks is concerned — the holding screen's "open in another
+ // tab" notice reads this.
+ if _, live := h.LiveForSession("r1", "sa"); live {
+ t.Fatal("a lapsed participation still counts as this session's live peer")
+ }
+ // The fresh stream, before any tick: it succeeds, with a new peer.
+ fresh, err := h.Open("r1", "sa", 1, "Ann", KindMember)
+ if err != nil {
t.Fatalf("a fresh stream before the tick: %v", err)
}
- h.Tick(context.Background())
+ if fresh.Peer == a.Peer || fresh.Peer == "" {
+ t.Fatalf("fresh stream reused the lapsed peer: %q", fresh.Peer)
+ }
if _, ok := h.Find("r1", "sa", a.Peer); ok {
- t.Fatal("the tick left the lapsed participation behind")
+ t.Fatal("the lapsed participation survived the fresh stream")
+ }
+ if id, live := h.LiveForSession("r1", "sa"); !live || id != fresh.Peer {
+ t.Fatalf("after the fresh stream the session holds %q (live=%v), want %q", id, live, fresh.Peer)
+ }
+ // And the tick that follows finds nothing left of it to remove.
+ h.Tick(context.Background())
+ if _, ok := h.Find("r1", "sa", fresh.Peer); !ok {
+ t.Fatal("the tick removed the fresh participation")
}
}
@@ -391,7 +414,7 @@ func TestAResumePastItsWindowIsRefusedBeforeTheTickRemovesIt(t *testing.T) {
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 {
+ if err := h.Reset(testCtx, "r1", "sa", a.Peer, b.Peer); err != nil {
t.Fatalf("reset %d: %v", i+1, err)
}
}
@@ -416,15 +439,15 @@ func TestGenerationsOnlyRiseAndASupersededPairIsStale(t *testing.T) {
// 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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Relay(testCtx, "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 {
+ if err := h.Relay(testCtx, "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/hub/subscriber.go b/internal/meet/hub/subscriber.go
index 97e8f2c..3bb68ae 100644
--- a/internal/meet/hub/subscriber.go
+++ b/internal/meet/hub/subscriber.go
@@ -41,11 +41,30 @@ func (s *Sub) terminate(final Message) {
// dropped to make room for it: a peer ended with a real reason
// still gets the events it had not read yet. Only a full buffer
// costs one, and by then the subscriber is lost anyway.
- select {
- case s.ch <- final:
- default:
- <-s.ch // reached only when full; no other sender holds h.mu
- s.ch <- final
+ //
+ // Every operation here is non-blocking, and that is the whole point
+ // (Codex S2-4). This runs with h.mu held, and a failed send proves
+ // the buffer was full at that instant only: the stream's own reader
+ // is not excluded by the lock and can drain the channel between the
+ // two statements. An unconditional receive would then block on an
+ // empty channel, holding the hub's one lock against every other
+ // room in the process, until the reader happened to send something
+ // nobody is sending. So: try, evict one if there is one, try again.
+ // Two attempts is the ceiling — the hub is the only sender and it
+ // holds the lock, so a send that fails after a successful eviction
+ // cannot happen, and dropping the final event is still better than
+ // deadlocking the hub if it ever does.
+ for i := 0; i < 2; i++ {
+ select {
+ case s.ch <- final:
+ close(s.ch)
+ return
+ default:
+ }
+ select {
+ case <-s.ch: // full a moment ago; one event makes room
+ default: // the reader drained it: the retry above will fit
+ }
}
close(s.ch)
}
diff --git a/internal/meet/hub/subscriber_test.go b/internal/meet/hub/subscriber_test.go
new file mode 100644
index 0000000..ca67b94
--- /dev/null
+++ b/internal/meet/hub/subscriber_test.go
@@ -0,0 +1,50 @@
+package hub
+
+import (
+ "testing"
+ "time"
+)
+
+// terminate runs with the hub's one lock held, and the stream's reader
+// is not excluded by that lock: it drains the channel on its own
+// goroutine, whenever it is scheduled. So a failed non-blocking send
+// proves the buffer was full at that instant and nothing more, and the
+// eviction that follows must not be a blocking receive — on an empty
+// channel it would hold h.mu against every room in the process, waiting
+// for an event that only h.mu's holder could send (Codex S2-4).
+//
+// The window is a few instructions wide, so this is a probabilistic
+// test run many times rather than a single deterministic one. Against
+// the unfixed code it hangs a few thousand rounds in, with and without
+// -race; against the fixed code the whole run costs a quarter of a
+// second (a second and a third under -race). A round that has not
+// finished within the deadline is the deadlock, and the test says so
+// rather than hanging the package.
+func TestTerminateNeverBlocksOnAChannelTheReaderDrained(t *testing.T) {
+ const rounds = 50000
+ for i := 0; i < rounds; i++ {
+ s := newSub()
+ for j := 0; j < subBuffer; j++ {
+ s.ch <- Message{Type: TypePeer}
+ }
+ // The reader: a stream handler, draining as fast as it is given
+ // events, exactly as serveStream does.
+ drained := make(chan struct{})
+ go func() {
+ defer close(drained)
+ for range s.C() {
+ }
+ }()
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ s.terminate(Message{Type: TypeLost, Reason: ReasonRemoved})
+ }()
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatalf("round %d: terminate blocked on a drained channel with the hub lock held", i)
+ }
+ <-drained
+ }
+}
diff --git a/internal/meet/signal.go b/internal/meet/signal.go
index 1789603..973b1be 100644
--- a/internal/meet/signal.go
+++ b/internal/meet/signal.go
@@ -157,11 +157,11 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) {
case hub.TypeLeave:
herr = a.hub.Leave(sqid, session, body.Peer)
case hub.TypeOffer, hub.TypeAnswer, hub.TypeCandidate:
- herr = a.hub.Relay(sqid, session, body.Peer, hub.Message{
+ herr = a.hub.Relay(r.Context(), sqid, session, body.Peer, hub.Message{
Type: body.Type, To: body.To, Seq: body.Seq, Gen: body.Gen, Payload: body.Payload,
})
case hub.TypeReset:
- herr = a.hub.Reset(sqid, session, body.Peer, body.To)
+ herr = a.hub.Reset(r.Context(), sqid, session, body.Peer, body.To)
case hub.TypeChat, hub.TypeShare:
// Slice 4.
a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type", map[string]any{"error": "bad-request"})
diff --git a/internal/meet/stream.go b/internal/meet/stream.go
index 4dae231..d9d561d 100644
--- a/internal/meet/stream.go
+++ b/internal/meet/stream.go
@@ -175,7 +175,7 @@ func (a *app) stream(w http.ResponseWriter, r *http.Request) {
// The hub answers first when it holds a tombstone, so a peer
// that was in the room when it ended is told why (410 ended)
// rather than sent to open a fresh stream.
- if re := a.hub.CanResume(sqid, session, resume); re != nil {
+ if re := a.hub.CanResume(r.Context(), sqid, session, resume); re != nil {
a.refuseResume(w, re)
return
}
@@ -183,7 +183,7 @@ func (a *app) stream(w http.ResponseWriter, r *http.Request) {
return
}
if probe {
- if re := a.hub.CanResume(sqid, session, resume); re != nil {
+ if re := a.hub.CanResume(r.Context(), sqid, session, resume); re != nil {
a.refuseResume(w, re)
return
}
@@ -204,7 +204,7 @@ func (a *app) stream(w http.ResponseWriter, r *http.Request) {
return
}
}
- res, err := a.hub.Resume(sqid, session, resume, last)
+ res, err := a.hub.Resume(r.Context(), sqid, session, resume, last)
if err != nil {
re, ok := err.(*hub.ResumeError)
if !ok {
diff --git a/internal/meettest/standing_test.go b/internal/meettest/standing_test.go
new file mode 100644
index 0000000..9f74ad7
--- /dev/null
+++ b/internal/meettest/standing_test.go
@@ -0,0 +1,183 @@
+package meettest
+
+import (
+ "encoding/json"
+ "net/http"
+ "testing"
+ "time"
+)
+
+// twoMembersInOneRoom is the shape every case below starts from: two
+// signed-in members of the same meeting, both joined, both streams
+// open, and each one's peer id.
+func twoMembersInOneRoom(t *testing.T) (x *harness, sqid string, a, b *client, sa, sb *sseReader, pa, pb string) {
+ t.Helper()
+ x, sqid, _ = memberWithMeeting(t)
+ a = x.browser(t)
+ a.signIn("paul@example.com")
+ sa = a.stream("/m/" + sqid + "/stream")
+ t.Cleanup(sa.close)
+ pa, _ = sa.hello()
+ a.signal(sqid, pa, `{"type":"join"}`)
+
+ b = newMemberBrowser(t, x, "paul@rslw.com")
+ sb = b.stream("/m/" + sqid + "/stream")
+ t.Cleanup(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)
+ }
+ return
+}
+
+// gone reads the `gone` event the room is owed when a participation
+// ends, and says whether it named this peer.
+func sawGone(t *testing.T, s *sseReader, peer string) bool {
+ t.Helper()
+ ev, data, ok := s.nextOf("gone", 2*time.Second)
+ if !ok {
+ t.Logf("no gone event (got %q)", ev)
+ return false
+ }
+ var g struct{ From string }
+ json.Unmarshal(data, &g)
+ return g.From == peer
+}
+
+// Spec §2.5's paragraph, corrected into the table by §15.12: a
+// participation whose kind is `member` re-derives standing on every
+// request, not only on the fifteen-second tick — "a member who has just
+// been removed is dropped by the next thing they do". Codex S2-1 found
+// every one of these accepting a removed member's traffic until the
+// tick caught up.
+//
+// Each case revokes B's membership and then makes exactly one request,
+// before any tick has run.
+func TestAMemberWithNoStandingIsRefusedByTheNextRequestNotTheNextTick(t *testing.T) {
+ t.Run("relay", func(t *testing.T) {
+ x, sqid, _, b, sa, _, pa, pb := twoMembersInOneRoom(t)
+ // The answer a peer that never existed gets, read before
+ // anything is revoked: a dropped member must be told exactly
+ // this and nothing more (§2.5, "404, never 403").
+ unknown, unknownBody := b.signal(sqid, pb, `{"type":"offer","to":"`+zeroPeer+`","seq":1,"gen":1,"payload":{}}`)
+ if unknown.StatusCode != http.StatusNotFound {
+ t.Fatalf("offer to an unknown peer: %d", unknown.StatusCode)
+ }
+
+ x.ctl.SetAllowlist("paul@example.com") // B is off the team
+ resp, body := b.signal(sqid, pb, `{"type":"offer","to":"`+pa+`","seq":2,"gen":1,"payload":{}}`)
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("relay by a removed member before the tick: %d %s", resp.StatusCode, body)
+ }
+ if string(body) != string(unknownBody) {
+ t.Fatalf("relay refusal body %s, want the unknown-peer body %s", body, unknownBody)
+ }
+ if !sawGone(t, sa, pb) {
+ t.Fatal("the room was not told the removed member had gone")
+ }
+ })
+
+ t.Run("reset", func(t *testing.T) {
+ x, sqid, _, b, sa, _, pa, pb := twoMembersInOneRoom(t)
+ x.ctl.SetAllowlist("paul@example.com")
+ resp, body := b.signal(sqid, pb, `{"type":"reset","to":"`+pa+`"}`)
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("reset by a removed member before the tick: %d %s", resp.StatusCode, body)
+ }
+ if !sawGone(t, sa, pb) {
+ t.Fatal("the room was not told the removed member had gone")
+ }
+ })
+
+ t.Run("resume", func(t *testing.T) {
+ x, sqid, _, b, sa, sb, _, pb := twoMembersInOneRoom(t)
+ sb.close()
+ x.ctl.SetAllowlist("paul@example.com")
+ rec := b.getStream("/m/" + sqid + "/stream?peer=" + pb)
+ _, boot, reason := answer(t, rec)
+ if rec.Code != http.StatusNotFound || boot == "" || reason != "" {
+ t.Fatalf("resume by a removed member before the tick: %d boot=%q reason=%q, want 404 with a boot and no reason",
+ rec.Code, boot, reason)
+ }
+ if !sawGone(t, sa, pb) {
+ t.Fatal("the room was not told the removed member had gone")
+ }
+ })
+
+ // The probe answers the same 404 and ends nothing: §15.3 defines it
+ // as changing nothing at all, so the ending is the next mutating
+ // request's, or the tick's.
+ t.Run("probe", func(t *testing.T) {
+ x, sqid, _, b, sa, _, pa, pb := twoMembersInOneRoom(t)
+ x.ctl.SetAllowlist("paul@example.com")
+ rec := b.get("/m/" + sqid + "/stream?probe=1&peer=" + pb)
+ _, boot, reason := answer(t, rec)
+ if rec.Code != http.StatusNotFound || boot == "" || reason != "" {
+ t.Fatalf("probe by a removed member before the tick: %d boot=%q reason=%q, want 404 with a boot and no reason",
+ rec.Code, boot, reason)
+ }
+ if ev, _, ok := sa.nextOf("gone", 500*time.Millisecond); ok {
+ t.Fatalf("the probe ended the participation (%q reached the room)", ev)
+ }
+ // Still there, and still the room's to lose on the next thing
+ // this member actually does.
+ if resp, body := b.signal(sqid, pb, `{"type":"reset","to":"`+pa+`"}`); resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("reset after the probe: %d %s", resp.StatusCode, body)
+ }
+ if !sawGone(t, sa, pb) {
+ t.Fatal("the room was not told the removed member had gone")
+ }
+ })
+}
+
+// The other half of §15.12, and ruling 14's guarantee: an admitted
+// guest is untouched by any of this. Their grant has expired, which is
+// exactly the state a member would be dropped for, and they go on
+// relaying, resetting and resuming — the participation is the authority
+// once someone is in.
+func TestAnAdmittedGuestWithNoStandingKeepsRelayingResettingAndResuming(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+`"}}`)
+ 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
+
+ if resp, body := g.signal(sqid, pg, `{"type":"offer","to":"`+pm+`","seq":1,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("relay past grant expiry: %d %s", resp.StatusCode, body)
+ }
+ if resp, body := g.signal(sqid, pg, `{"type":"reset","to":"`+pm+`"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("reset past grant expiry: %d %s", resp.StatusCode, body)
+ }
+ if rec := g.get("/m/" + sqid + "/stream?probe=1&peer=" + pg); rec.Code != http.StatusNoContent {
+ t.Fatalf("resume probe past grant expiry: %d, want 204", rec.Code)
+ }
+ 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 := sm.nextOf("gone", 500*time.Millisecond); ok {
+ t.Fatalf("the expired guest was dropped by their own requests (%q reached the member)", ev)
+ }
+}
+
+// zeroPeer is a peer id of the right shape that no room ever issued.
+const zeroPeer = "00000000000000000000000000000000"
diff --git a/internal/meettest/stream_test.go b/internal/meettest/stream_test.go
index 7005ef9..d9123b5 100644
--- a/internal/meettest/stream_test.go
+++ b/internal/meettest/stream_test.go
@@ -182,6 +182,44 @@ func TestLapsedResumeIs404WithTheSameBootAndAFreshStreamGetsANewPeer(t *testing.
}
}
+// The same lapse, with no tick at all in it (Codex S2-2). §15.7 refuses
+// the resume the moment the window passes; the browser's answer to a 404
+// is a fresh stream, and until now that fresh stream was refused 409 by
+// the very participation it had just been refused a resume of — so the
+// client showed "This meeting is open in another tab" about a tab that
+// did not exist, and the holding screen said the same. The tick is what
+// used to clear it, up to fifteen seconds later.
+func TestAFreshStreamAfterALapsedResumeSucceedsBeforeAnyTick(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ x.ctl.SetStreamTimings(50*time.Millisecond, 150*time.Millisecond)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ s := c.stream("/m/" + sqid + "/stream")
+ peer, boot := s.hello()
+ if !s.waitClosed(2 * time.Second) {
+ t.Fatal("stream did not close at the TTL")
+ }
+ x.advance(31 * time.Second) // past the window, and no tick is run
+
+ rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer)
+ if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot {
+ t.Fatalf("lapsed resume: %d boot %q, want 404 and %q", rec.Code, gotBoot, boot)
+ }
+ // The probe the client asks first says the same, and changes nothing.
+ if probed := c.get("/m/" + sqid + "/stream?probe=1"); probed.Code != http.StatusNoContent {
+ t.Fatalf("peerless probe after the lapse: %d, want 204", probed.Code)
+ }
+ if rec := c.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK ||
+ strings.Contains(rec.Body.String(), "already open in another tab") {
+ t.Fatalf("holding screen after the lapse: %d, and it still says the meeting is open elsewhere", rec.Code)
+ }
+ s3 := c.stream("/m/" + sqid + "/stream")
+ defer s3.close()
+ if p, b := s3.hello(); p == "" || p == peer || b != boot {
+ t.Fatalf("fresh stream after the lapse: peer %q boot %q", p, b)
+ }
+}
+
// The resume probe (§4.5, recorded for revision 9): EventSource hides
// the status of a refused stream from the page, so the browser asks
// first. The probe answers and does nothing else — it must not attach,