| 1 | package hub |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "io" |
| 8 | "log/slog" |
| 9 | "testing" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | func twoAdmitted(t *testing.T) (*Hub, *time.Time, OpenResult, OpenResult) { |
| 14 | t.Helper() |
| 15 | now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| 16 | h := testHub(&now) |
| 17 | ctx := context.Background() |
| 18 | a, _ := h.Open("r1", "sa", 1, "Ann", KindMember) |
| 19 | h.Join(ctx, "r1", "sa", a.Peer) |
| 20 | b, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| 21 | h.Join(ctx, "r1", "sb", b.Peer) |
| 22 | drain(a.Sub) |
| 23 | drain(b.Sub) |
| 24 | return h, &now, a, b |
| 25 | } |
| 26 | |
| 27 | func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T) { |
| 28 | h, _, a, b := twoAdmitted(t) |
| 29 | ctx := context.Background() |
| 30 | g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| 31 | h.Join(ctx, "r1", "sg", g.Peer) |
| 32 | drain(a.Sub) |
| 33 | drain(b.Sub) |
| 34 | 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"}`)}) |
| 35 | if err != nil { |
| 36 | t.Fatal(err) |
| 37 | } |
| 38 | bev := drain(b.Sub) |
| 39 | if len(bev) != 1 || bev[0].From != a.Peer || bev[0].Name != "Ann" || string(bev[0].Payload) != `{"sdp":"x"}` { |
| 40 | t.Fatalf("b got %+v", bev) |
| 41 | } |
| 42 | if ev := drain(a.Sub); len(ev) != 0 { |
| 43 | t.Fatalf("sender echoed %v", types(ev)) |
| 44 | } |
| 45 | if ev := drain(g.Sub); len(ev) != 0 { |
| 46 | t.Fatalf("waiting guest received a relay: %v", types(ev)) |
| 47 | } |
| 48 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: g.Peer, Seq: 2, Gen: 1}); err == nil { |
| 49 | t.Fatal("relay to a waiting peer accepted") |
| 50 | } |
| 51 | if err := h.Relay(testCtx, "r1", "sg", g.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrNotAdmitted { |
| 52 | t.Fatalf("waiting guest relaying: %v", err) |
| 53 | } |
| 54 | if err := h.Relay(testCtx, "r1", "sb", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer { |
| 55 | t.Fatalf("peer through another session: %v", err) |
| 56 | } |
| 57 | if err := h.Relay(testCtx, "r2", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer { |
| 58 | t.Fatalf("peer through another room: %v", err) |
| 59 | } |
| 60 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeChat, To: b.Peer, Seq: 4, Gen: 1}); !errors.Is(err, ErrNotTargeted) { |
| 61 | t.Fatalf("chat as a targeted relay: %v", err) |
| 62 | } |
| 63 | } |
| 64 | |
| 65 | // A peer is not its own target: a self-addressed relay or reset would |
| 66 | // otherwise ring a message the sender already has and open a pair with |
| 67 | // itself, which no roster ever names. |
| 68 | func TestAPeerIsNotItsOwnTarget(t *testing.T) { |
| 69 | h, _, a, _ := twoAdmitted(t) |
| 70 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrUnknownPeer { |
| 71 | t.Fatalf("self-addressed relay: %v", err) |
| 72 | } |
| 73 | if err := h.Reset(testCtx, "r1", "sa", a.Peer, a.Peer); err != ErrUnknownPeer { |
| 74 | t.Fatalf("self-addressed reset: %v", err) |
| 75 | } |
| 76 | if ev := drain(a.Sub); len(ev) != 0 { |
| 77 | t.Fatalf("a self target reached the sender: %v", types(ev)) |
| 78 | } |
| 79 | p, _ := h.Find("r1", "sa", a.Peer) |
| 80 | if len(p.ring) != 0 { |
| 81 | t.Fatalf("a self target filled the ring: %+v", p.ring) |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | func TestStaleGenIsRefusedAndResetConverges(t *testing.T) { |
| 86 | h, _, a, b := twoAdmitted(t) |
| 87 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen { |
| 88 | t.Fatalf("gen 2 before any reset: %v", err) |
| 89 | } |
| 90 | if err := h.Reset(testCtx, "r1", "sa", a.Peer, b.Peer); err != nil { |
| 91 | t.Fatal(err) |
| 92 | } |
| 93 | if err := h.Reset(testCtx, "r1", "sb", b.Peer, a.Peer); err != nil { // simultaneous resets converge on the higher gen |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | aev, bev := drain(a.Sub), drain(b.Sub) |
| 97 | if len(aev) != 2 || aev[1].Type != TypePair || aev[1].Gen != 3 || aev[1].Peer != b.Peer { |
| 98 | t.Fatalf("a got %+v", aev) |
| 99 | } |
| 100 | if len(bev) != 2 || bev[1].Gen != 3 || bev[1].Peer != a.Peer { |
| 101 | t.Fatalf("b got %+v", bev) |
| 102 | } |
| 103 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != ErrStaleGen { |
| 104 | t.Fatal("old gen accepted after reset") |
| 105 | } |
| 106 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) { |
| 112 | h, now, a, b := twoAdmitted(t) |
| 113 | ctx := context.Background() |
| 114 | for i := uint64(1); i <= 5; i++ { |
| 115 | h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)}) |
| 116 | } |
| 117 | drain(a.Sub) |
| 118 | h.Detach("r1", a.Peer, nil) |
| 119 | *now = now.Add(5 * time.Second) |
| 120 | // Two more from the same sender *while a is detached*: the point of |
| 121 | // the ring is that a message sent into the window is kept, not just |
| 122 | // one that was already delivered and re-read. Without these, the |
| 123 | // replay below would be satisfied by the ring alone and would say |
| 124 | // nothing about what happens during the thirty seconds a browser is |
| 125 | // away (Task 3 carry-over). |
| 126 | for i := uint64(6); i <= 7; i++ { |
| 127 | if err := h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)}); err != nil { |
| 128 | t.Fatalf("relay %d into the resume window: %v", i, err) |
| 129 | } |
| 130 | } |
| 131 | res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3}) |
| 132 | if err != nil { |
| 133 | t.Fatal(err) |
| 134 | } |
| 135 | if !res.Admitted || res.Roster == nil { |
| 136 | t.Fatalf("resume %+v", res) |
| 137 | } |
| 138 | var seqs []uint64 |
| 139 | for _, m := range res.Replay { |
| 140 | seqs = append(seqs, m.Seq) |
| 141 | } |
| 142 | if len(seqs) != 4 || seqs[0] != 4 || seqs[1] != 5 || seqs[2] != 6 || seqs[3] != 7 { |
| 143 | t.Fatalf("replay seqs %v, want 4 5 6 7 (everything above last, delivered and undelivered alike)", seqs) |
| 144 | } |
| 145 | if len(res.Gap) != 0 { |
| 146 | t.Fatalf("a ring that held everything reported a gap: %v", res.Gap) |
| 147 | } |
| 148 | if res.Sub == nil || res.Sub.Lost() { |
| 149 | t.Fatal("resume did not attach a fresh subscriber") |
| 150 | } |
| 151 | // The old window is over; the tick must not reap a participation |
| 152 | // that has resumed. |
| 153 | *now = now.Add(ResumeWindow + time.Second) |
| 154 | h.Tick(ctx) |
| 155 | if _, ok := h.Find("r1", "sa", a.Peer); !ok { |
| 156 | t.Fatal("the tick reaped a participation that had resumed") |
| 157 | } |
| 158 | // A waiting guest resumes and learns nothing. |
| 159 | g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| 160 | h.Join(ctx, "r1", "sg", g.Peer) |
| 161 | h.Detach("r1", g.Peer, nil) |
| 162 | gres, err := h.Resume(testCtx, "r1", "sg", g.Peer, nil) |
| 163 | if err != nil { |
| 164 | t.Fatal(err) |
| 165 | } |
| 166 | if gres.Admitted || gres.Roster != nil || len(gres.Replay) != 0 { |
| 167 | t.Fatalf("waiting guest resume leaked %+v", gres) |
| 168 | } |
| 169 | } |
| 170 | |
| 171 | func TestResumeGapWhenTheRingRolledOver(t *testing.T) { |
| 172 | h, now, a, b := twoAdmitted(t) |
| 173 | for i := uint64(1); i <= 40; i++ { |
| 174 | h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| 175 | } |
| 176 | drain(a.Sub) |
| 177 | h.Detach("r1", a.Peer, nil) |
| 178 | *now = now.Add(time.Second) |
| 179 | res, _ := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2}) |
| 180 | if len(res.Gap) != 1 || res.Gap[0] != b.Peer || len(res.Replay) != 0 { |
| 181 | t.Fatalf("expected a gap for b, got replay %d gap %v", len(res.Replay), res.Gap) |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | // One ring is shared by every sender, so one peer's burst can evict |
| 186 | // another's messages wholesale. The gap must still be reported, which |
| 187 | // the ring's own oldest seq per sender cannot see. |
| 188 | func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) { |
| 189 | h, now, a, b := twoAdmitted(t) |
| 190 | ctx := context.Background() |
| 191 | c, _ := h.Open("r1", "sc", 3, "Cat", KindMember) |
| 192 | h.Join(ctx, "r1", "sc", c.Peer) |
| 193 | drain(a.Sub) |
| 194 | for i := uint64(1); i <= 3; i++ { |
| 195 | h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| 196 | } |
| 197 | for i := uint64(1); i <= RingSize; i++ { |
| 198 | h.Relay(testCtx, "r1", "sc", c.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| 199 | } |
| 200 | drain(a.Sub) |
| 201 | h.Detach("r1", a.Peer, nil) |
| 202 | *now = now.Add(time.Second) |
| 203 | res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 1, c.Peer: RingSize}) |
| 204 | if err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | if len(res.Gap) != 1 || res.Gap[0] != b.Peer { |
| 208 | t.Fatalf("b's messages were evicted wholesale, gap %v", res.Gap) |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | // The ring and the eviction marks outlive the peer that filled them. |
| 213 | // A resume must not name a peer that has left: the roster the browser |
| 214 | // reconciles against says it is gone, so a gap for it would ask for a |
| 215 | // reset with nobody, and its replayed signalling would rebuild the |
| 216 | // connection the roster has just told the browser to close. |
| 217 | func TestResumeNamesNoPeerThatHasLeft(t *testing.T) { |
| 218 | h, now, a, b := twoAdmitted(t) |
| 219 | for i := uint64(1); i <= 40; i++ { |
| 220 | h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| 221 | } |
| 222 | p, _ := h.Find("r1", "sa", a.Peer) |
| 223 | if p.evicted[b.Peer] == 0 { |
| 224 | t.Fatal("the ring did not evict anything, so this proves nothing") |
| 225 | } |
| 226 | if err := h.Leave("r1", "sb", b.Peer); err != nil { |
| 227 | t.Fatal(err) |
| 228 | } |
| 229 | drain(a.Sub) |
| 230 | h.Detach("r1", a.Peer, nil) |
| 231 | *now = now.Add(time.Second) |
| 232 | res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2}) |
| 233 | if err != nil { |
| 234 | t.Fatal(err) |
| 235 | } |
| 236 | if len(res.Gap) != 0 { |
| 237 | t.Fatalf("gap names a peer that left: %v", res.Gap) |
| 238 | } |
| 239 | if len(res.Replay) != 0 { |
| 240 | t.Fatalf("replay carries a departed peer's signalling: %+v", res.Replay) |
| 241 | } |
| 242 | } |
| 243 | |
| 244 | // A second resume of an attached peer replaces the attachment: the old |
| 245 | // stream is told replaced, and the participation itself carries on. |
| 246 | func TestResumeReplacesAnExistingAttachment(t *testing.T) { |
| 247 | h, _, a, b := twoAdmitted(t) |
| 248 | res, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil) |
| 249 | if err != nil { |
| 250 | t.Fatal(err) |
| 251 | } |
| 252 | old := drain(a.Sub) |
| 253 | if len(old) != 1 || old[0].Type != TypeLost || old[0].Reason != ReasonReplaced { |
| 254 | t.Fatalf("old stream got %+v", old) |
| 255 | } |
| 256 | if !a.Sub.Lost() { |
| 257 | t.Fatal("the old subscriber was not closed") |
| 258 | } |
| 259 | if ev := drain(b.Sub); len(ev) != 0 { |
| 260 | t.Fatalf("the room was told the peer left: %v", types(ev)) |
| 261 | } |
| 262 | if !res.Admitted { |
| 263 | t.Fatal("the replaced attachment lost its admission") |
| 264 | } |
| 265 | if err := h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != nil { |
| 266 | t.Fatal(err) |
| 267 | } |
| 268 | if ev := drain(res.Sub); len(ev) != 1 || ev[0].Type != TypeOffer { |
| 269 | t.Fatalf("the new attachment got %v", types(ev)) |
| 270 | } |
| 271 | if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err != nil { |
| 272 | t.Fatalf("a replaced attachment left a tombstone: %v", err) |
| 273 | } |
| 274 | } |
| 275 | |
| 276 | func TestResumeAnswers(t *testing.T) { |
| 277 | h, now, a, _ := twoAdmitted(t) |
| 278 | if _, err := h.Resume(testCtx, "r1", "sb", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 || err.(*ResumeError).Boot != h.Boot() { |
| 279 | t.Fatalf("resume through another session: %v", err) |
| 280 | } |
| 281 | if _, err := h.Resume(testCtx, "r9", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 { |
| 282 | t.Fatalf("resume through another room: %v", err) |
| 283 | } |
| 284 | h.Takeover("r1", "sa") |
| 285 | if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 410 || err.(*ResumeError).Reason != ReasonReplaced { |
| 286 | t.Fatalf("resume after takeover: %v", err) |
| 287 | } |
| 288 | *now = now.Add(TombstoneLife + time.Minute) |
| 289 | h.Tick(context.Background()) |
| 290 | if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 { |
| 291 | t.Fatalf("resume after tombstone expiry: %v", err) |
| 292 | } |
| 293 | if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != nil { |
| 294 | t.Fatalf("fresh stream after tombstone: %v", err) |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | // Why a participation ended is account state, and a peer id is in every |
| 299 | // roster the room-mates hold. A tombstone is therefore answered only |
| 300 | // through the room and session it was bound to; anyone else gets the |
| 301 | // same 404 an unknown peer gets (spec §4.2). |
| 302 | func TestATombstoneIsAnsweredOnlyThroughItsOwnRoomAndSession(t *testing.T) { |
| 303 | h, _, a, b := twoAdmitted(t) |
| 304 | if _, err := h.Open("r2", "sother", 9, "Other", KindMember); err != nil { |
| 305 | t.Fatal(err) |
| 306 | } |
| 307 | if err := h.Remove(context.Background(), "r1", "sb", b.Peer, a.Peer); err != nil { |
| 308 | t.Fatal(err) |
| 309 | } |
| 310 | |
| 311 | // Ann's own session, in the room she was removed from. |
| 312 | _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil) |
| 313 | re, ok := err.(*ResumeError) |
| 314 | if !ok || re.Status != 410 || re.Reason != ReasonRemoved { |
| 315 | t.Fatalf("her own resume: %v", err) |
| 316 | } |
| 317 | if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re == nil || re.Status != 410 || re.Reason != ReasonRemoved { |
| 318 | t.Fatalf("her own probe: %v", re) |
| 319 | } |
| 320 | |
| 321 | // Bob, who has her peer id from the roster, learns nothing. |
| 322 | for _, probe := range []struct{ what, sqid, session string }{ |
| 323 | {"another session in the same room", "r1", "sb"}, |
| 324 | {"another room this session is in", "r2", "sother"}, |
| 325 | {"another room and another session", "r2", "sb"}, |
| 326 | } { |
| 327 | _, err := h.Resume(testCtx, probe.sqid, probe.session, a.Peer, nil) |
| 328 | re, ok := err.(*ResumeError) |
| 329 | if !ok || re.Status != 404 || re.Boot != h.Boot() || re.Reason != "" { |
| 330 | t.Fatalf("%s: %v", probe.what, err) |
| 331 | } |
| 332 | if re := h.CanResume(testCtx, probe.sqid, probe.session, a.Peer); re == nil || re.Status != 404 || re.Reason != "" { |
| 333 | t.Fatalf("%s, probed: %+v", probe.what, re) |
| 334 | } |
| 335 | } |
| 336 | } |
| 337 | |
| 338 | // Spec §15.3: a probe changes nothing. Not the participation, and not |
| 339 | // the kind either — recording a member's demotion to guest would send a |
| 340 | // fresh roster to every admitted peer in the room, which is a room-wide |
| 341 | // event caused by a request whose whole contract is that it has no |
| 342 | // effects (Codex S2-10). The transition is real, and is applied by the |
| 343 | // next thing this member actually does. |
| 344 | func TestAProbeNeverWritesAKindNorBroadcastsARoster(t *testing.T) { |
| 345 | now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| 346 | // Standing this test can move: a member taken off the team while |
| 347 | // holding a live guest grant is still someone — ok stays true — and |
| 348 | // their kind becomes guest. That is the branch a probe wrote through. |
| 349 | kind := KindMember |
| 350 | h := New(Config{ |
| 351 | Now: func() time.Time { return now }, |
| 352 | Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), |
| 353 | StandingFor: func(ctx context.Context, room string, personID int64) (Kind, bool, error) { |
| 354 | return kind, true, nil |
| 355 | }, |
| 356 | }) |
| 357 | a, _ := h.Open("r1", "sa", 1, "Ann", KindMember) |
| 358 | h.Join(testCtx, "r1", "sa", a.Peer) |
| 359 | b, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| 360 | h.Join(testCtx, "r1", "sb", b.Peer) |
| 361 | drain(a.Sub) |
| 362 | drain(b.Sub) |
| 363 | |
| 364 | kind = KindGuest // off the team, grant still live |
| 365 | |
| 366 | if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re != nil { |
| 367 | t.Fatalf("the probe refused a participation that is still there: %+v", re) |
| 368 | } |
| 369 | p, ok := h.Find("r1", "sa", a.Peer) |
| 370 | if !ok || p.Kind != KindMember { |
| 371 | t.Fatalf("the probe wrote the kind: %v", p.Kind) |
| 372 | } |
| 373 | if ev := drain(a.Sub); len(ev) != 0 { |
| 374 | t.Fatalf("the probe sent its own stream %v", types(ev)) |
| 375 | } |
| 376 | if ev := drain(b.Sub); len(ev) != 0 { |
| 377 | t.Fatalf("the probe sent the room %v", types(ev)) |
| 378 | } |
| 379 | |
| 380 | // The next mutating request applies it, and owes the room the roster |
| 381 | // a kind change earns (§2.4). |
| 382 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != nil { |
| 383 | t.Fatal(err) |
| 384 | } |
| 385 | if p, _ := h.Find("r1", "sa", a.Peer); p.Kind != KindGuest { |
| 386 | t.Fatalf("kind after a mutating request: %v, want guest", p.Kind) |
| 387 | } |
| 388 | if ev := types(drain(a.Sub)); len(ev) != 1 || ev[0] != TypeRoster { |
| 389 | t.Fatalf("the demoted member got %v, want one roster", ev) |
| 390 | } |
| 391 | if ev := types(drain(b.Sub)); len(ev) != 2 || ev[0] != TypeRoster || ev[1] != TypeOffer { |
| 392 | t.Fatalf("the room got %v, want the roster then the offer", ev) |
| 393 | } |
| 394 | } |
| 395 | |
| 396 | func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) { |
| 397 | h, now, a, _ := twoAdmitted(t) |
| 398 | h.Detach("r1", a.Peer, nil) |
| 399 | *now = now.Add(ResumeWindow + time.Second) |
| 400 | h.Tick(context.Background()) |
| 401 | _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil) |
| 402 | re, ok := err.(*ResumeError) |
| 403 | if !ok || re.Status != 404 || re.Boot != h.Boot() { |
| 404 | t.Fatalf("lapsed resume: %v", err) |
| 405 | } |
| 406 | } |
| 407 | |
| 408 | // Spec §4.5's window is thirty seconds, not thirty plus however long it |
| 409 | // is until the next tick: a resume that arrives after resume_until is |
| 410 | // refused on arrival (§15.7). The refusal itself still writes nothing — |
| 411 | // a probe changes nothing — but the fresh stream the browser opens next |
| 412 | // must not be refused 409 by the very participation it was just refused |
| 413 | // a resume of (Codex S2-2): a browser told "not here" and then "open in |
| 414 | // another tab" about the same peer has nowhere to go and shows a |
| 415 | // takeover screen for a tab that does not exist. Open ends it and takes |
| 416 | // its place. |
| 417 | func TestAResumePastItsWindowIsRefusedAndAFreshStreamTakesItsPlace(t *testing.T) { |
| 418 | now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| 419 | h := testHub(&now) |
| 420 | a, _ := h.Open("r1", "sa", 1, "Ann", KindMember) |
| 421 | h.Detach("r1", a.Peer, a.Sub) |
| 422 | |
| 423 | now = now.Add(ResumeWindow - time.Second) |
| 424 | if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re != nil { |
| 425 | t.Fatalf("inside the window: %+v", re) |
| 426 | } |
| 427 | now = now.Add(2 * time.Second) // past resume_until, before any tick |
| 428 | re := h.CanResume(testCtx, "r1", "sa", a.Peer) |
| 429 | if re == nil || re.Status != 404 || re.Boot != h.Boot() { |
| 430 | t.Fatalf("past the window: %+v", re) |
| 431 | } |
| 432 | if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil { |
| 433 | t.Fatal("Resume attached a participation past its window") |
| 434 | } |
| 435 | // Refused, and the refusal itself removed nothing: a probe and a |
| 436 | // refused resume are both defined as changing nothing. |
| 437 | if _, ok := h.Find("r1", "sa", a.Peer); !ok { |
| 438 | t.Fatal("the refusal removed the participation") |
| 439 | } |
| 440 | // Nor does the session still hold a peer here as far as anything |
| 441 | // that asks is concerned — the holding screen's "open in another |
| 442 | // tab" notice reads this. |
| 443 | if _, live := h.LiveForSession("r1", "sa"); live { |
| 444 | t.Fatal("a lapsed participation still counts as this session's live peer") |
| 445 | } |
| 446 | // The fresh stream, before any tick: it succeeds, with a new peer. |
| 447 | fresh, err := h.Open("r1", "sa", 1, "Ann", KindMember) |
| 448 | if err != nil { |
| 449 | t.Fatalf("a fresh stream before the tick: %v", err) |
| 450 | } |
| 451 | if fresh.Peer == a.Peer || fresh.Peer == "" { |
| 452 | t.Fatalf("fresh stream reused the lapsed peer: %q", fresh.Peer) |
| 453 | } |
| 454 | if _, ok := h.Find("r1", "sa", a.Peer); ok { |
| 455 | t.Fatal("the lapsed participation survived the fresh stream") |
| 456 | } |
| 457 | if id, live := h.LiveForSession("r1", "sa"); !live || id != fresh.Peer { |
| 458 | t.Fatalf("after the fresh stream the session holds %q (live=%v), want %q", id, live, fresh.Peer) |
| 459 | } |
| 460 | // And the tick that follows finds nothing left of it to remove. |
| 461 | h.Tick(context.Background()) |
| 462 | if _, ok := h.Find("r1", "sa", fresh.Peer); !ok { |
| 463 | t.Fatal("the tick removed the fresh participation") |
| 464 | } |
| 465 | } |
| 466 | |
| 467 | // Spec §4.6: a `pair` at or below the generation a browser already holds |
| 468 | // is ignored. The hub owns the half of that rule this package can gate — |
| 469 | // generations only ever rise, and only the newest is live — and the |
| 470 | // browser's half rests on it: two resets from one side yield 2 and 3, so |
| 471 | // the `pair` carrying 2 is superseded by the time it could be applied, |
| 472 | // and signalling that still carries 2 is refused. Ignoring a superseded |
| 473 | // `pair` is therefore safe rather than merely tidy. |
| 474 | func TestGenerationsOnlyRiseAndASupersededPairIsStale(t *testing.T) { |
| 475 | h, _, a, b := twoAdmitted(t) |
| 476 | for i := 0; i < 2; i++ { |
| 477 | if err := h.Reset(testCtx, "r1", "sa", a.Peer, b.Peer); err != nil { |
| 478 | t.Fatalf("reset %d: %v", i+1, err) |
| 479 | } |
| 480 | } |
| 481 | for _, side := range []struct { |
| 482 | name string |
| 483 | ms []Message |
| 484 | other PeerID |
| 485 | }{ |
| 486 | {"a", drain(a.Sub), b.Peer}, |
| 487 | {"b", drain(b.Sub), a.Peer}, |
| 488 | } { |
| 489 | if len(side.ms) != 2 { |
| 490 | t.Fatalf("%s got %v, want one pair per reset", side.name, types(side.ms)) |
| 491 | } |
| 492 | for i, want := range []uint64{2, 3} { |
| 493 | m := side.ms[i] |
| 494 | if m.Type != TypePair || m.Peer != side.other || m.Gen != want { |
| 495 | t.Fatalf("%s pair %d: %+v, want pair with %s at gen %d", side.name, i, m, side.other, want) |
| 496 | } |
| 497 | } |
| 498 | } |
| 499 | |
| 500 | // Gen 2 is exactly the superseded one: announced by the first reset, |
| 501 | // replaced by the second before either side could finish with it. |
| 502 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen { |
| 503 | t.Fatalf("signalling at the superseded generation: %v, want ErrStaleGen", err) |
| 504 | } |
| 505 | // A generation the hub never issued is refused too: only the current |
| 506 | // one is live, from either direction. |
| 507 | if err := h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 4}); err != ErrStaleGen { |
| 508 | t.Fatalf("signalling above the current generation: %v, want ErrStaleGen", err) |
| 509 | } |
| 510 | if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil { |
| 511 | t.Fatalf("signalling at the current generation: %v", err) |
| 512 | } |
| 513 | } |
| 514 | |
amadan