| 1 | package callstest |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "fmt" |
| 6 | "net/http" |
| 7 | "net/http/httptest" |
| 8 | "net/url" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | // answer reads a refused stream: the status and the JSON body spec §4.5 |
| 15 | // fixes the browser's behaviour on. A refusal never streams, so the |
| 16 | // in-process recorder is the right client for it. |
| 17 | func answer(t *testing.T, rec *httptest.ResponseRecorder) (peer, boot, reason string) { |
| 18 | t.Helper() |
| 19 | var body struct{ Peer, Boot, Reason string } |
| 20 | if rec.Body.Len() > 0 { |
| 21 | if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { |
| 22 | t.Fatalf("refusal body %q: %v", rec.Body.String(), err) |
| 23 | } |
| 24 | } |
| 25 | return body.Peer, body.Boot, body.Reason |
| 26 | } |
| 27 | |
| 28 | // Spec §4.1: the first event on a stream is hello, carrying the peer id |
| 29 | // the server issued and the boot id of this process. Spec §4.2: a |
| 30 | // session holds one peer per room, so a second fresh stream is refused |
| 31 | // with 409 naming the peer it already has. |
| 32 | func TestStreamSaysHelloWithPeerAndBootAndRefusesTheSecondFreshStream(t *testing.T) { |
| 33 | x, sqid, _ := memberWithMeeting(t) |
| 34 | c := x.browser(t) |
| 35 | c.signIn("paul@example.com") |
| 36 | s := c.stream("/m/" + sqid + "/stream") |
| 37 | defer s.close() |
| 38 | peer, boot := s.hello() |
| 39 | if len(peer) != 32 || len(boot) != 32 { |
| 40 | t.Fatalf("hello peer=%q boot=%q", peer, boot) |
| 41 | } |
| 42 | rec := c.getStream("/m/" + sqid + "/stream") |
| 43 | if rec.Code != http.StatusConflict { |
| 44 | t.Fatalf("second fresh stream: %d, want 409", rec.Code) |
| 45 | } |
| 46 | if named, _, _ := answer(t, rec); named != peer { |
| 47 | t.Fatalf("409 names %q, want %q", named, peer) |
| 48 | } |
| 49 | // A second browser of the same person is a different session, so it |
| 50 | // gets a peer of its own (§4.4 allows four streams per person). |
| 51 | c2 := x.browser(t) |
| 52 | c2.signIn("paul@example.com") |
| 53 | s2 := c2.stream("/m/" + sqid + "/stream") |
| 54 | defer s2.close() |
| 55 | if peer2, boot2 := s2.hello(); peer2 == peer || boot2 != boot { |
| 56 | t.Fatalf("second browser: peer %q boot %q", peer2, boot2) |
| 57 | } |
| 58 | } |
| 59 | |
| 60 | // Spec §2.5: a fresh stream is authorized by standing, a resume by the |
| 61 | // participation bound to this session. Neither ever answers 403. |
| 62 | func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) { |
| 63 | x, sqid, tok := memberWithMeeting(t) |
| 64 | stranger := x.browser(t) |
| 65 | if rec := stranger.get("/m/" + sqid + "/stream"); rec.Code != http.StatusSeeOther { |
| 66 | t.Fatalf("signed-out stream: %d, want the sign-in redirect", rec.Code) |
| 67 | } |
| 68 | c := x.browser(t) |
| 69 | c.signIn("paul@example.com") |
| 70 | // A second meeting, which the guest below is never given a link to. |
| 71 | elsewhere, _ := createMeeting(t, c, "Not yours", "") |
| 72 | s := c.stream("/m/" + sqid + "/stream") |
| 73 | peer, boot := s.hello() |
| 74 | s.close() |
| 75 | |
| 76 | guest := x.browser(t) |
| 77 | guest.get("/j/" + tok) |
| 78 | guest.signIn("guest@example.com") |
| 79 | if rec := guest.getStream("/m/" + elsewhere + "/stream"); rec.Code != http.StatusNotFound { |
| 80 | t.Fatalf("a meeting the guest has no standing for: %d, want 404", rec.Code) |
| 81 | } |
| 82 | rec := guest.getStream("/m/" + sqid + "/stream?peer=" + peer) |
| 83 | if rec.Code != http.StatusNotFound { |
| 84 | t.Fatalf("resume through another session: %d, want 404", rec.Code) |
| 85 | } |
| 86 | // The same answer an unknown peer gets, so nothing leaks (§4.2). |
| 87 | if _, gotBoot, _ := answer(t, rec); gotBoot != boot { |
| 88 | t.Fatalf("404 boot %q, want %q", gotBoot, boot) |
| 89 | } |
| 90 | unknown := guest.getStream("/m/" + sqid + "/stream?peer=0123456789abcdef0123456789abcdef") |
| 91 | if unknown.Code != http.StatusNotFound || unknown.Body.String() != rec.Body.String() { |
| 92 | t.Fatalf("unknown peer: %d %s", unknown.Code, unknown.Body.String()) |
| 93 | } |
| 94 | } |
| 95 | |
| 96 | // Spec §4.5: a closed stream leaves the participation in the hub for |
| 97 | // thirty seconds, and a resume reattaches the same one — the same peer, |
| 98 | // the same boot. This participation never joined, so it is told only |
| 99 | // that it is waiting, which is all a peer outside the room may learn. |
| 100 | func TestResumeReattachesTheSameParticipation(t *testing.T) { |
| 101 | x, sqid, _ := memberWithMeeting(t) |
| 102 | c := x.browser(t) |
| 103 | c.signIn("paul@example.com") |
| 104 | s := c.stream("/m/" + sqid + "/stream") |
| 105 | peer, boot := s.hello() |
| 106 | s.close() |
| 107 | |
| 108 | s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| 109 | defer s2.close() |
| 110 | p2, b2 := s2.hello() |
| 111 | if p2 != peer || b2 != boot { |
| 112 | t.Fatalf("resume hello %s %s, want %s %s", p2, b2, peer, boot) |
| 113 | } |
| 114 | ev, data, ok := s2.next(time.Second) |
| 115 | if !ok || ev != "state" { |
| 116 | t.Fatalf("after a resume: %s ok=%v", ev, ok) |
| 117 | } |
| 118 | var state struct{ Waiting bool } |
| 119 | json.Unmarshal(data, &state) |
| 120 | if !state.Waiting { |
| 121 | t.Fatalf("state %s, want waiting", data) |
| 122 | } |
| 123 | // The resumed participation is still the session's one peer. |
| 124 | if rec := c.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusConflict { |
| 125 | t.Fatalf("fresh stream while resumed: %d, want 409", rec.Code) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | // Spec §4.1: the stream heartbeats, and dies at its TTL so hibernation |
| 130 | // is never held off by one. The browser resumes and loses nothing. |
| 131 | func TestTTLClosesTheStreamAndResumeWorksWithinTheWindow(t *testing.T) { |
| 132 | x, sqid, _ := memberWithMeeting(t) |
| 133 | x.ctl.SetStreamTimings(50*time.Millisecond, 300*time.Millisecond) |
| 134 | c := x.browser(t) |
| 135 | c.signIn("paul@example.com") |
| 136 | s := c.stream("/m/" + sqid + "/stream") |
| 137 | peer, _ := s.hello() |
| 138 | if !s.waitHeartbeat(2 * time.Second) { |
| 139 | t.Fatal("no heartbeat on an idle stream") |
| 140 | } |
| 141 | if !s.waitClosed(2 * time.Second) { |
| 142 | t.Fatal("stream did not close at the TTL") |
| 143 | } |
| 144 | s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| 145 | defer s2.close() |
| 146 | if p, _ := s2.hello(); p != peer { |
| 147 | t.Fatalf("resume after the TTL: %s, want %s", p, peer) |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | // Spec §4.5: a resume that misses its window is 404 with the same boot |
| 152 | // the browser last saw — the process did not restart, the window |
| 153 | // lapsed — and the browser's answer is a fresh stream, which gets a |
| 154 | // peer of its own because the old participation is gone. |
| 155 | func TestLapsedResumeIs404WithTheSameBootAndAFreshStreamGetsANewPeer(t *testing.T) { |
| 156 | x, sqid, _ := memberWithMeeting(t) |
| 157 | // The stream is left to die at its own TTL rather than closed from |
| 158 | // this side: the handler detaches on its way out, before the |
| 159 | // response is finished, so an EOF here means the resume window has |
| 160 | // already been stamped — and the clock this test then moves is |
| 161 | // ahead of it, not behind. |
| 162 | x.ctl.SetStreamTimings(50*time.Millisecond, 150*time.Millisecond) |
| 163 | c := x.browser(t) |
| 164 | c.signIn("paul@example.com") |
| 165 | s := c.stream("/m/" + sqid + "/stream") |
| 166 | peer, boot := s.hello() |
| 167 | if !s.waitClosed(2 * time.Second) { |
| 168 | t.Fatal("stream did not close at the TTL") |
| 169 | } |
| 170 | x.advance(31 * time.Second) |
| 171 | x.ctl.TickForTest() |
| 172 | |
| 173 | rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer) |
| 174 | _, gotBoot, _ := answer(t, rec) |
| 175 | if rec.Code != http.StatusNotFound || gotBoot != boot { |
| 176 | t.Fatalf("lapsed resume: %d boot %q, want 404 and %q", rec.Code, gotBoot, boot) |
| 177 | } |
| 178 | s3 := c.stream("/m/" + sqid + "/stream") |
| 179 | defer s3.close() |
| 180 | if p, b := s3.hello(); p == "" || p == peer || b != boot { |
| 181 | t.Fatalf("fresh stream after the lapse: peer %q boot %q", p, b) |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | // The same lapse, with no tick at all in it (Codex S2-2). §15.7 refuses |
| 186 | // the resume the moment the window passes; the browser's answer to a 404 |
| 187 | // is a fresh stream, and until now that fresh stream was refused 409 by |
| 188 | // the very participation it had just been refused a resume of — so the |
| 189 | // client showed "This meeting is open in another tab" about a tab that |
| 190 | // did not exist, and the holding screen said the same. The tick is what |
| 191 | // used to clear it, up to fifteen seconds later. |
| 192 | func TestAFreshStreamAfterALapsedResumeSucceedsBeforeAnyTick(t *testing.T) { |
| 193 | x, sqid, _ := memberWithMeeting(t) |
| 194 | x.ctl.SetStreamTimings(50*time.Millisecond, 150*time.Millisecond) |
| 195 | c := x.browser(t) |
| 196 | c.signIn("paul@example.com") |
| 197 | s := c.stream("/m/" + sqid + "/stream") |
| 198 | peer, boot := s.hello() |
| 199 | if !s.waitClosed(2 * time.Second) { |
| 200 | t.Fatal("stream did not close at the TTL") |
| 201 | } |
| 202 | x.advance(31 * time.Second) // past the window, and no tick is run |
| 203 | |
| 204 | rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer) |
| 205 | if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot { |
| 206 | t.Fatalf("lapsed resume: %d boot %q, want 404 and %q", rec.Code, gotBoot, boot) |
| 207 | } |
| 208 | // The probe the client asks first says the same, and changes nothing. |
| 209 | if probed := c.get("/m/" + sqid + "/stream?probe=1"); probed.Code != http.StatusNoContent { |
| 210 | t.Fatalf("peerless probe after the lapse: %d, want 204", probed.Code) |
| 211 | } |
| 212 | if rec := c.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK || |
| 213 | strings.Contains(rec.Body.String(), "already open in another tab") { |
| 214 | t.Fatalf("holding screen after the lapse: %d, and it still says the meeting is open elsewhere", rec.Code) |
| 215 | } |
| 216 | s3 := c.stream("/m/" + sqid + "/stream") |
| 217 | defer s3.close() |
| 218 | if p, b := s3.hello(); p == "" || p == peer || b != boot { |
| 219 | t.Fatalf("fresh stream after the lapse: peer %q boot %q", p, b) |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | // The resume probe (§4.5, recorded for revision 9): EventSource hides |
| 224 | // the status of a refused stream from the page, so the browser asks |
| 225 | // first. The probe answers and does nothing else — it must not attach, |
| 226 | // must not clear the resume window and must not touch the ring. |
| 227 | func TestResumeProbeAnswersWithoutAttaching(t *testing.T) { |
| 228 | x, sqid, _ := memberWithMeeting(t) |
| 229 | c := x.browser(t) |
| 230 | c.signIn("paul@example.com") |
| 231 | |
| 232 | // No Accept header: a probe attaches nothing, so ruling 15 exempts |
| 233 | // it. Its answer is still never cached. |
| 234 | rec0 := c.get("/m/" + sqid + "/stream?probe=1") |
| 235 | if rec0.Code != http.StatusNoContent { |
| 236 | t.Fatalf("probe with nothing open: %d, want 204", rec0.Code) |
| 237 | } |
| 238 | if cc := rec0.Header().Get("Cache-Control"); cc != "no-store" { |
| 239 | t.Fatalf("probe Cache-Control %q, want no-store", cc) |
| 240 | } |
| 241 | s := c.stream("/m/" + sqid + "/stream") |
| 242 | peer, boot := s.hello() |
| 243 | rec := c.get("/m/" + sqid + "/stream?probe=1") |
| 244 | named, _, _ := answer(t, rec) |
| 245 | if rec.Code != http.StatusConflict || named != peer { |
| 246 | t.Fatalf("probe for a fresh stream while one is open: %d %q", rec.Code, named) |
| 247 | } |
| 248 | s.close() |
| 249 | |
| 250 | probed := c.get("/m/" + sqid + "/stream?probe=1&peer=" + peer) |
| 251 | if probed.Code != http.StatusNoContent { |
| 252 | t.Fatalf("probe of a resumable peer: %d, want 204", probed.Code) |
| 253 | } |
| 254 | if cc := probed.Header().Get("Cache-Control"); cc != "no-store" { |
| 255 | t.Fatalf("resume probe Cache-Control %q, want no-store", cc) |
| 256 | } |
| 257 | rec = c.get("/m/" + sqid + "/stream?probe=1&peer=0123456789abcdef0123456789abcdef") |
| 258 | _, gotBoot, _ := answer(t, rec) |
| 259 | if rec.Code != http.StatusNotFound || gotBoot != boot { |
| 260 | t.Fatalf("probe of an unknown peer: %d boot %q", rec.Code, gotBoot) |
| 261 | } |
| 262 | // Probed twice and still resumable: the probe changed nothing. |
| 263 | s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| 264 | defer s2.close() |
| 265 | if p, _ := s2.hello(); p != peer { |
| 266 | t.Fatalf("resume after two probes: %s, want %s", p, peer) |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | // Spec §2.4 and §4.5: ending the meeting closes every stream with |
| 271 | // lost{ended}, and the peer id is remembered so a resume is told why. |
| 272 | // A meeting nobody streamed has no headstone in the hub at all, so |
| 273 | // ended_at is what refuses the stream (the Task 2 carry). |
| 274 | func TestEndingTheMeetingClosesTheStreamAndRefusesWhatComesAfter(t *testing.T) { |
| 275 | x, sqid, _ := memberWithMeeting(t) |
| 276 | c := x.browser(t) |
| 277 | c.signIn("paul@example.com") |
| 278 | s := c.stream("/m/" + sqid + "/stream") |
| 279 | defer s.close() |
| 280 | peer, boot := s.hello() |
| 281 | |
| 282 | if rec := c.post("/meetings/"+sqid+"/end", nil); rec.Code != http.StatusSeeOther { |
| 283 | t.Fatalf("end meeting: %d", rec.Code) |
| 284 | } |
| 285 | ev, data, ok := s.next(2 * time.Second) |
| 286 | if !ok || ev != "lost" { |
| 287 | t.Fatalf("after the meeting ended: %s ok=%v", ev, ok) |
| 288 | } |
| 289 | var lost struct{ Reason string } |
| 290 | json.Unmarshal(data, &lost) |
| 291 | if lost.Reason != "ended" { |
| 292 | t.Fatalf("lost %s, want reason ended", data) |
| 293 | } |
| 294 | if !s.waitClosed(2 * time.Second) { |
| 295 | t.Fatal("the stream stayed open after the meeting ended") |
| 296 | } |
| 297 | rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer) |
| 298 | _, _, reason := answer(t, rec) |
| 299 | if rec.Code != http.StatusGone || reason != "ended" { |
| 300 | t.Fatalf("resume after the ending: %d %q, want 410 ended", rec.Code, reason) |
| 301 | } |
| 302 | if rec := c.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound { |
| 303 | t.Fatalf("fresh stream on an ended meeting: %d, want 404", rec.Code) |
| 304 | } |
| 305 | |
| 306 | // A meeting nobody opened a stream on: the hub has never heard of |
| 307 | // it, so only the row can refuse it. |
| 308 | quiet, _ := createMeeting(t, c, "Never opened", "") |
| 309 | if rec := c.post("/meetings/"+quiet+"/end", nil); rec.Code != http.StatusSeeOther { |
| 310 | t.Fatalf("end the quiet meeting: %d", rec.Code) |
| 311 | } |
| 312 | if rec := c.getStream("/m/" + quiet + "/stream"); rec.Code != http.StatusNotFound { |
| 313 | t.Fatalf("fresh stream on a meeting nobody opened: %d, want 404", rec.Code) |
| 314 | } |
| 315 | // The peer the ending tombstoned above, offered to a room it was |
| 316 | // never in: a tombstone answers only through its own room and |
| 317 | // session, so this is the plain 404 an unknown peer gets. |
| 318 | rec = c.getStream("/m/" + quiet + "/stream?peer=" + peer) |
| 319 | if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot { |
| 320 | t.Fatalf("resume into a meeting nobody opened: %d boot %q", rec.Code, gotBoot) |
| 321 | } |
| 322 | if rec := c.getStream("/m/zzzzzzzzzz/stream"); rec.Code != http.StatusNotFound { |
| 323 | t.Fatalf("stream for a meeting that does not exist: %d", rec.Code) |
| 324 | } |
| 325 | } |
| 326 | |
| 327 | // Ruling 15: the attaching stream is opened by EventSource, which always |
| 328 | // asks for text/event-stream. A cross-site top-level navigation asks for |
| 329 | // HTML, and it must not be able to take the session's one peer slot in |
| 330 | // the room and hold it for the resume window. |
| 331 | func TestAStreamWithoutTheEventStreamAcceptTakesNoPeer(t *testing.T) { |
| 332 | x, sqid, _ := memberWithMeeting(t) |
| 333 | c := x.browser(t) |
| 334 | c.signIn("paul@example.com") |
| 335 | |
| 336 | rec := c.getNav("/m/" + sqid + "/stream") // a navigation, not EventSource |
| 337 | if rec.Code != http.StatusNotAcceptable { |
| 338 | t.Fatalf("stream without the Accept header: %d, want 406", rec.Code) |
| 339 | } |
| 340 | // The check runs before any lookup, so the answer is the same for |
| 341 | // every meeting and every person, and says what is actually wrong. |
| 342 | if got := strings.TrimSpace(rec.Body.String()); got != `{"error":"accept"}` { |
| 343 | t.Fatalf("406 body %s", got) |
| 344 | } |
| 345 | if rec := c.getNav("/m/" + sqid + "/stream?peer=whatever"); rec.Code != http.StatusNotAcceptable { |
| 346 | t.Fatalf("resume without the Accept header: %d, want 406", rec.Code) |
| 347 | } |
| 348 | s := c.stream("/m/" + sqid + "/stream") |
| 349 | defer s.close() |
| 350 | if peer, _ := s.hello(); peer == "" { |
| 351 | t.Fatal("the stream that does ask for text/event-stream got no peer") |
| 352 | } |
| 353 | } |
| 354 | |
| 355 | // A resume's ?last= is refused above eight kibibytes, before it is |
| 356 | // parsed: one seq per peer and sixteen peers to a room, so nothing this |
| 357 | // size is a browser's. |
| 358 | func TestAnOversizeLastIsRefusedBeforeItIsParsed(t *testing.T) { |
| 359 | x, sqid, _ := memberWithMeeting(t) |
| 360 | c := x.browser(t) |
| 361 | c.signIn("paul@example.com") |
| 362 | s := c.stream("/m/" + sqid + "/stream") |
| 363 | peer, _ := s.hello() |
| 364 | s.close() |
| 365 | |
| 366 | // Valid JSON, and too big: a last that only the cap can refuse. Bad |
| 367 | // JSON would be refused by the parse whether the cap is there or |
| 368 | // not, and would prove nothing. |
| 369 | var big strings.Builder |
| 370 | big.WriteString("{") |
| 371 | for i := 0; i < 700; i++ { |
| 372 | if i > 0 { |
| 373 | big.WriteString(",") |
| 374 | } |
| 375 | fmt.Fprintf(&big, "%q:1", fmt.Sprintf("%032x", i)) |
| 376 | } |
| 377 | big.WriteString("}") |
| 378 | if !json.Valid([]byte(big.String())) || big.Len() <= 8<<10 { |
| 379 | t.Fatalf("the oversize last is %d bytes and valid=%v", big.Len(), json.Valid([]byte(big.String()))) |
| 380 | } |
| 381 | rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer + "&last=" + url.QueryEscape(big.String())) |
| 382 | if rec.Code != http.StatusBadRequest { |
| 383 | t.Fatalf("oversize last: %d, want 400", rec.Code) |
| 384 | } |
| 385 | // Still resumable: a refused query is not a refused participation. |
| 386 | s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer + "&last=" + url.QueryEscape(`{"`+peer+`":3}`)) |
| 387 | defer s2.close() |
| 388 | if p, _ := s2.hello(); p != peer { |
| 389 | t.Fatalf("resume after the refused last: %s, want %s", p, peer) |
| 390 | } |
| 391 | } |
| 392 | |
| 393 | // Spec §4.4: a subscriber that falls sixty-four events behind is |
| 394 | // declared lost and told so, and §4.5 makes lost{buffer} the one reason |
| 395 | // a browser answers with a resume instead of a screen. This is that path |
| 396 | // over real HTTP rather than against hub.Sub directly: a client that |
| 397 | // stops reading its socket backs the handler's write up, the hub's |
| 398 | // sixty-four-slot buffer fills behind the stalled handler, and the |
| 399 | // participation itself survives — so the resume that follows gets the |
| 400 | // same peer back. |
| 401 | func TestASlowClientIsLostOnTheBufferAndResumesTheSameParticipation(t *testing.T) { |
| 402 | x, sqid, _ := memberWithMeeting(t) |
| 403 | sender := x.browser(t) |
| 404 | sender.signIn("paul@example.com") |
| 405 | ss := sender.stream("/m/" + sqid + "/stream") |
| 406 | defer ss.close() |
| 407 | from, _ := ss.hello() |
| 408 | if resp, _ := sender.signal(sqid, from, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| 409 | t.Fatalf("sender join: %d", resp.StatusCode) |
| 410 | } |
| 411 | |
| 412 | // The victim reads its hello and then stops reading altogether. |
| 413 | victim := newMemberBrowser(t, x, "paul@rslw.com") |
| 414 | vs := victim.rawStream("/m/" + sqid + "/stream") |
| 415 | defer vs.close() |
| 416 | to := vs.hello() |
| 417 | if resp, _ := victim.signal(sqid, to, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| 418 | t.Fatalf("victim join: %d", resp.StatusCode) |
| 419 | } |
| 420 | if ev, _, ok := ss.nextOf("peer", 2*time.Second); !ok { |
| 421 | t.Fatalf("the sender never saw the victim admitted (got %q); nothing below would reach it", ev) |
| 422 | } |
| 423 | |
| 424 | // The arithmetic, and it is the host's sysctls that it deliberately |
| 425 | // does not depend on. Two things absorb events before the hub's own |
| 426 | // buffer starts filling: the socket, and the handler's one in hand. |
| 427 | // Both ends of the socket are pinned — the server's SO_SNDBUF by the |
| 428 | // harness's own listener (about 16 KiB once Linux has doubled it), |
| 429 | // the client's SO_RCVBUF by rawStream's dial (about 8 KiB) — so at |
| 430 | // 60 KiB an event the socket holds at most one, whatever tcp_wmem's |
| 431 | // ceiling happens to be on this machine. The hub's buffer is |
| 432 | // sixty-four, and the handler holds one more. So the overflow is at |
| 433 | // about sixty-six, and a hundred is comfortably past it. |
| 434 | // |
| 435 | // The limiter is thirty a second with a burst of sixty on the |
| 436 | // harness's frozen clock, so the burst is spent in batches with a |
| 437 | // refill between them rather than being throttled halfway through. |
| 438 | const flood = 100 |
| 439 | payload := strings.Repeat("x", 60*1024) |
| 440 | for i := 1; i <= flood; i++ { |
| 441 | if i%50 == 1 && i > 1 { |
| 442 | x.advance(2 * time.Second) // refills the whole burst |
| 443 | } |
| 444 | body := fmt.Sprintf(`{"type":"candidate","to":%q,"seq":%d,"gen":1,"payload":{"c":%q}}`, to, i, payload) |
| 445 | if resp, b := sender.signal(sqid, from, body); resp.StatusCode != http.StatusOK { |
| 446 | t.Fatalf("candidate %d: %d %s", i, resp.StatusCode, b) |
| 447 | } |
| 448 | } |
| 449 | |
| 450 | // Now the victim starts reading again. Whatever was buffered comes |
| 451 | // through first; the stream ends on lost{buffer}. |
| 452 | var seen int |
| 453 | for { |
| 454 | name, data, ok := vs.next(30 * time.Second) |
| 455 | if !ok { |
| 456 | t.Fatalf("the stream ended without lost{buffer} after %d of %d events; "+ |
| 457 | "if that is close to %d the socket absorbed the whole flood and the "+ |
| 458 | "harness's SO_SNDBUF did not take", seen, flood, flood) |
| 459 | } |
| 460 | seen++ |
| 461 | if name != "lost" { |
| 462 | continue |
| 463 | } |
| 464 | var lost struct{ Reason string } |
| 465 | json.Unmarshal(data, &lost) |
| 466 | if lost.Reason != "buffer" { |
| 467 | t.Fatalf("lost %s, want reason buffer", data) |
| 468 | } |
| 469 | break |
| 470 | } |
| 471 | // Events were actually lost, which is the whole claim: a stream that |
| 472 | // delivered every one of them and then said lost{buffer} would be |
| 473 | // something else entirely. The bound is the arithmetic above and not |
| 474 | // merely "fewer than the flood": sixty-six is what the buffer, the |
| 475 | // handler's one in hand and a socket that can hold at most one event |
| 476 | // add up to, so a number well past it means the socket absorbed the |
| 477 | // flood and the harness's pinned send buffer did not take. |
| 478 | const ceiling = 72 |
| 479 | if seen > ceiling { |
| 480 | t.Fatalf("the victim read %d of %d events before it was declared lost; "+ |
| 481 | "the buffer is sixty-four and the socket should hold at most one 60 KiB event, "+ |
| 482 | "so anything past %d means the send buffer was not pinned", seen, flood, ceiling) |
| 483 | } |
| 484 | |
| 485 | // The participation is not the subscriber: it is still there, and a |
| 486 | // resume gets the same peer and the roster back (§4.5). |
| 487 | vs.close() |
| 488 | vs2 := victim.stream("/m/" + sqid + "/stream?peer=" + to) |
| 489 | defer vs2.close() |
| 490 | if p, _ := vs2.hello(); p != to { |
| 491 | t.Fatalf("resume after lost{buffer}: %s, want %s", p, to) |
| 492 | } |
| 493 | if ev, _, ok := vs2.nextOf("roster", 2*time.Second); !ok { |
| 494 | t.Fatalf("the resumed peer is not admitted any more (got %q)", ev) |
| 495 | } |
| 496 | } |
| 497 | |
| 498 | // Spec §2.4's ruling 14, end to end: an admitted guest is never ended by |
| 499 | // their grant lapsing, because the participation is the authority once |
| 500 | // they are in. So signal keeps working, the stream keeps resuming — and |
| 501 | // fresh entry, which is standing's alone, is refused. |
| 502 | func TestAnAdmittedGuestCarriesOnPastGrantExpiryButGetsNoFreshEntry(t *testing.T) { |
| 503 | x, sqid, tok := memberWithMeeting(t) |
| 504 | m := x.browser(t) |
| 505 | m.signIn("paul@example.com") |
| 506 | sm := m.stream("/m/" + sqid + "/stream") |
| 507 | defer sm.close() |
| 508 | pm, _ := sm.hello() |
| 509 | m.signal(sqid, pm, `{"type":"join"}`) |
| 510 | |
| 511 | g := x.browser(t) |
| 512 | g.get("/j/" + tok) |
| 513 | g.signIn("guest@example.com") |
| 514 | sg := g.stream("/m/" + sqid + "/stream") |
| 515 | defer sg.close() |
| 516 | pg, _ := sg.hello() |
| 517 | g.signal(sqid, pg, `{"type":"join"}`) |
| 518 | m.signal(sqid, pm, `{"type":"admit","payload":{"peer":"`+pg+`"}}`) |
| 519 | // Proof of admission before anything is claimed about it: a waiting |
| 520 | // guest would also survive most of what follows, and would prove |
| 521 | // nothing about ruling 14. |
| 522 | if ev, _, ok := sg.nextOf("roster", 2*time.Second); !ok { |
| 523 | t.Fatalf("the guest was never admitted (got %q)", ev) |
| 524 | } |
| 525 | |
| 526 | x.advance(2*time.Hour + time.Minute) // the grant is over |
| 527 | |
| 528 | // signal: still theirs. join is idempotent for an admitted peer, and |
| 529 | // a targeted relay is something only an admitted peer may send, so |
| 530 | // both are worth asking. |
| 531 | if resp, b := g.signal(sqid, pg, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| 532 | t.Fatalf("join past grant expiry: %d %s", resp.StatusCode, b) |
| 533 | } |
| 534 | if resp, b := g.signal(sqid, pg, `{"type":"offer","to":"`+pm+`","seq":1,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK { |
| 535 | t.Fatalf("offer past grant expiry: %d %s", resp.StatusCode, b) |
| 536 | } |
| 537 | if ev, _, ok := sm.nextOf("offer", 2*time.Second); !ok { |
| 538 | t.Fatalf("the member never received the expired guest's offer (got %q)", ev) |
| 539 | } |
| 540 | |
| 541 | // stream?peer=: still theirs, roster and all. |
| 542 | sg.close() |
| 543 | sg2 := g.stream("/m/" + sqid + "/stream?peer=" + pg) |
| 544 | defer sg2.close() |
| 545 | if p, _ := sg2.hello(); p != pg { |
| 546 | t.Fatalf("resume past grant expiry: %s, want %s", p, pg) |
| 547 | } |
| 548 | if ev, _, ok := sg2.nextOf("roster", 2*time.Second); !ok { |
| 549 | t.Fatalf("the resumed guest is not admitted any more (got %q)", ev) |
| 550 | } |
| 551 | |
| 552 | // GET /m/{sqid}/ice is authorized by the guest's admitted |
| 553 | // participation (hub.Authorize), never by standing — ruling 14 |
| 554 | // again, on the one endpoint §13 names "on every endpoint": a |
| 555 | // lapsed grant does not cost this expired guest their ICE servers. |
| 556 | x.ctl.SetRelay(relayContract(t)) |
| 557 | rec := g.get("/m/" + sqid + "/ice?peer=" + pg) |
| 558 | if rec.Code != http.StatusOK { |
| 559 | t.Fatalf("GET /ice past grant expiry: %d, want 200", rec.Code) |
| 560 | } |
| 561 | var ice struct { |
| 562 | ICEServers []struct { |
| 563 | URLs []string `json:"urls"` |
| 564 | } `json:"iceServers"` |
| 565 | } |
| 566 | if err := json.Unmarshal(rec.Body.Bytes(), &ice); err != nil { |
| 567 | t.Fatal(err) |
| 568 | } |
| 569 | if len(ice.ICEServers) == 0 { |
| 570 | t.Fatal("GET /ice past grant expiry returned no servers") |
| 571 | } |
| 572 | |
| 573 | // Fresh entry is not theirs: a stream with no peer is authorized by |
| 574 | // standing, and standing is gone. 404, never 403 (§2.5). |
| 575 | if rec := g.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound { |
| 576 | t.Fatalf("fresh stream past grant expiry: %d, want 404", rec.Code) |
| 577 | } |
| 578 | // It is standing that refused that, not the one-peer-per-session |
| 579 | // rule: this session's peer is refused with 409 while it holds one, |
| 580 | // and 409 is what this would be if the standing check had passed. |
| 581 | // (A second browser for the same guest cannot be used to make the |
| 582 | // point: signing one in needs the join link again, and following the |
| 583 | // link mints a fresh grant, which is the very thing that has to be |
| 584 | // gone here.) |
| 585 | // |
| 586 | // The holding screen is *not* refused: roomAccess is standing OR a |
| 587 | // live participation (§2.5), and ruling 14 keeps this one's |
| 588 | // participation. TestRoomPageSurvivesGrantExpiryForAnAdmittedGuest |
| 589 | // is the room page's half of the same rule. |
| 590 | if rec := g.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK { |
| 591 | t.Fatalf("holding screen past grant expiry: %d, want 200", rec.Code) |
| 592 | } |
| 593 | } |
| 594 | |
| 595 | // Spec §2.4 and §13's "a member dropped on the tick", over HTTP: the |
| 596 | // team changes, the next tick re-derives standing for every joined |
| 597 | // participation, and the member who is no longer on it loses the room |
| 598 | // while the room is told they are gone. |
| 599 | func TestAMemberDroppedFromTheAllowlistIsLostOnTheTick(t *testing.T) { |
| 600 | x, sqid, _ := memberWithMeeting(t) |
| 601 | a := x.browser(t) |
| 602 | a.signIn("paul@example.com") |
| 603 | sa := a.stream("/m/" + sqid + "/stream") |
| 604 | defer sa.close() |
| 605 | pa, _ := sa.hello() |
| 606 | a.signal(sqid, pa, `{"type":"join"}`) |
| 607 | |
| 608 | b := newMemberBrowser(t, x, "paul@rslw.com") |
| 609 | sb := b.stream("/m/" + sqid + "/stream") |
| 610 | defer sb.close() |
| 611 | pb, _ := sb.hello() |
| 612 | b.signal(sqid, pb, `{"type":"join"}`) |
| 613 | if ev, _, ok := sa.nextOf("peer", 2*time.Second); !ok { |
| 614 | t.Fatalf("the two members never met (got %q)", ev) |
| 615 | } |
| 616 | |
| 617 | x.ctl.SetAllowlist("paul@example.com") // B is off the team |
| 618 | x.ctl.TickForTest() |
| 619 | |
| 620 | ev, data, ok := sb.nextOf("lost", 2*time.Second) |
| 621 | if !ok { |
| 622 | t.Fatalf("the dropped member's stream stayed open (got %q)", ev) |
| 623 | } |
| 624 | var lost struct{ Reason string } |
| 625 | json.Unmarshal(data, &lost) |
| 626 | if lost.Reason != "standing" { |
| 627 | t.Fatalf("dropped member got lost %s, want reason standing", data) |
| 628 | } |
| 629 | ev, data, ok = sa.nextOf("gone", 2*time.Second) |
| 630 | if !ok { |
| 631 | t.Fatalf("the room was never told the member left (got %q)", ev) |
| 632 | } |
| 633 | var gone struct{ From string } |
| 634 | json.Unmarshal(data, &gone) |
| 635 | if gone.From != pb { |
| 636 | t.Fatalf("gone names %q, want %q", gone.From, pb) |
| 637 | } |
| 638 | |
| 639 | // Their participation is gone, and the tombstone says why (§4.5). |
| 640 | rec := b.getStream("/m/" + sqid + "/stream?peer=" + pb) |
| 641 | if _, _, reason := answer(t, rec); rec.Code != http.StatusGone || reason != "standing" { |
| 642 | t.Fatalf("resume after the drop: %d %q, want 410 standing", rec.Code, reason) |
| 643 | } |
| 644 | // And so is entry: no standing, no fresh stream, no room page. |
| 645 | if rec := b.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound { |
| 646 | t.Fatalf("fresh stream after the drop: %d, want 404", rec.Code) |
| 647 | } |
| 648 | if rec := b.get("/m/" + sqid); rec.Code != http.StatusNotFound { |
| 649 | t.Fatalf("room page after the drop: %d, want 404", rec.Code) |
| 650 | } |
| 651 | } |
| 652 | |
| 653 | // A SIGTERM cancels the context the app was built on. rastrillo answers |
| 654 | // the same signal with a ten-second Shutdown, and Shutdown waits for |
| 655 | // handlers without cancelling their contexts, so a live stream that |
| 656 | // watched nothing would hold that whole deadline and the process would |
| 657 | // exit 1 instead of cleanly. The browser resumes, as at the TTL. |
| 658 | func TestShutdownEndsLiveStreams(t *testing.T) { |
| 659 | x, sqid, _ := memberWithMeeting(t) |
| 660 | c := x.browser(t) |
| 661 | c.signIn("paul@example.com") |
| 662 | s := c.stream("/m/" + sqid + "/stream") |
| 663 | defer s.close() |
| 664 | s.hello() |
| 665 | |
| 666 | x.shutdown() |
| 667 | if !s.waitClosed(2 * time.Second) { |
| 668 | t.Fatal("a live stream outlived the app's context") |
| 669 | } |
| 670 | } |
| 671 | |
| 672 | // SetStreamTimings is a test knob, and a zero from it must not panic a |
| 673 | // ticker inside a request goroutine: both timings have a floor. |
| 674 | func TestZeroStreamTimingsAreClampedNotFatal(t *testing.T) { |
| 675 | x, sqid, _ := memberWithMeeting(t) |
| 676 | // A zero heartbeat with a TTL that is not zero: the heartbeat is |
| 677 | // what proves the clamp. hello is written before the ticker exists, |
| 678 | // and net/http recovers a panic per connection, so a stream that |
| 679 | // panicked on time.NewTicker(0) would still say hello and still |
| 680 | // close — but it would never beat. |
| 681 | x.ctl.SetStreamTimings(0, 500*time.Millisecond) |
| 682 | c := x.browser(t) |
| 683 | c.signIn("paul@example.com") |
| 684 | s := c.stream("/m/" + sqid + "/stream") |
| 685 | defer s.close() |
| 686 | if peer, _ := s.hello(); peer == "" { |
| 687 | t.Fatal("no peer") |
| 688 | } |
| 689 | if !s.waitHeartbeat(2 * time.Second) { |
| 690 | t.Fatal("a floored heartbeat never beat") |
| 691 | } |
| 692 | if !s.waitClosed(2 * time.Second) { |
| 693 | t.Fatal("the stream did not end at its TTL") |
| 694 | } |
| 695 | // The app is still serving: nothing panicked. |
| 696 | if rec := c.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK { |
| 697 | t.Fatalf("holding screen after the floored stream: %d", rec.Code) |
| 698 | } |
| 699 | } |
| 700 | |
amadan