amadan
stream: SSE per peer with hello, heartbeat, TTL and resume answers
Three departures from the brief. The hub's session id is a salted hash of the session cookie rather than the subject plus the session's creation instant: that instant is stored to the second, so two sign-ins of one person in one second collide and the second browser is refused the room. Detach takes the caller's own Sub and detaches only while that is still the attachment, so a stream displaced by a resume does not detach its replacement. And the tests do not drive join, because POST /m/{sqid}/signal arrives with the signal handler; entry, refusal, resume, the probe, the TTL and the ending are proved here, and admission end to end belongs to that task.
Also: ?probe=1 answers 204, 404, 409 or 410 without attaching, through a new read-only hub.CanResume, because EventSource hides the status of a refused stream from the page. The store's clock moved behind an atomic so a test can move it while a stream is open, and so the hub reads the moved clock at all. Refusals on this endpoint answer JSON, not the error page: its client is a script and spec 4.5 fixes the browser's behaviour per status and body.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
15 files changed,
+998
−28
internal/meet/app.go+69 −2internal/meet/auth_test.go+2 −2internal/meet/handlers_meetings.go+3 −0internal/meet/hub/hub.go+11 −1internal/meet/hub/hub_test.go+31 −1internal/meet/hub/relay.go+32 −11internal/meet/hub/relay_test.go+5 −5internal/meet/joincookie_test.go+1 −1internal/meet/participation.go+46 −0internal/meet/store.go+34 −3internal/meet/store_test.go+1 −1internal/meet/stream.go+285 −0internal/meettest/harness_test.go+22 −1internal/meettest/sse_test.go+181 −0internal/meettest/stream_test.go+275 −0
CI passed — run details
diff --git a/internal/meet/app.go b/internal/meet/app.go| index 57a1450..11088d6 100644 |
| --- a/internal/meet/app.go |
| +++ b/internal/meet/app.go |
| @@ -2,9 +2,12 @@ package meet |
| import ( |
| "context" |
| + "crypto/rand" |
| + "encoding/hex" |
| "fmt" |
| "log/slog" |
| "net/http" |
| + "sync/atomic" |
| "time" |
| "github.com/go-chi/chi/v5" |
| @@ -15,6 +18,7 @@ import ( |
| "amadan.net/rastrillo/rastrillo/db" |
| "amadan.net/rastrillo/rastrillo/migrate" |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| "amadan.net/oficina/meet/internal/meet/seal" |
| ) |
| @@ -28,16 +32,45 @@ type app struct { |
| auth *rauth.Auth |
| linkBudget *budget |
| outsiderBudget *budget |
| + hub *hub.Hub |
| + // hubSalt keys the session cookie into the hub's session id, so the |
| + // hub holds no session material it did not have to. |
| + hubSalt string |
| + |
| + // The stream's two timings, held as atomics because a test moves |
| + // them (SetStreamTimings) while stream handlers are reading them. |
| + heartbeat atomic.Int64 |
| + ttl atomic.Int64 |
| } |
| +func (a *app) heartbeatEvery() time.Duration { return time.Duration(a.heartbeat.Load()) } |
| +func (a *app) streamLife() time.Duration { return time.Duration(a.ttl.Load()) } |
| + |
| // TestHandle is the test-only control surface NewForTest hands back: |
| // the app's clock, and the sweep that would otherwise only run on a |
| // timer. Tests only; nothing in production holds one. |
| type TestHandle struct{ a *app } |
| +// SetStreamTimings shortens the SSE heartbeat and TTL so a test can |
| +// watch a stream expire without waiting five minutes. Set it before any |
| +// stream opens: a stream reads both once, when it starts. |
| +func (h *TestHandle) SetStreamTimings(heartbeat, ttl time.Duration) { |
| + h.a.heartbeat.Store(int64(heartbeat)) |
| + h.a.ttl.Store(int64(ttl)) |
| +} |
| + |
| +// TickForTest runs one pass of the hub housekeeping App runs every |
| +// fifteen seconds: unjoined and lapsed participations dropped, |
| +// tombstones expired, every joined kind re-derived from standing. |
| +func (h *TestHandle) TickForTest() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| + defer cancel() |
| + h.a.hub.Tick(ctx) |
| +} |
| + |
| // SetClock replaces the store's clock, so a test can expire grants, |
| // pending rows and magic links without sleeping. |
| -func (h *TestHandle) SetClock(now func() time.Time) { h.a.store.now = now } |
| +func (h *TestHandle) SetClock(now func() time.Time) { h.a.store.clk.set(now) } |
| // SweepForTest runs one bounded pass of what the hourly ticker runs, |
| // using now for both store.sweep and auth.Sweep — Internal 5's one |
| @@ -135,6 +168,26 @@ func (a *app) sweepForever(ctx context.Context) { |
| } |
| } |
| +// tickEvery is the hub's housekeeping interval (spec §2.4): the quiet |
| +// case, for participations nothing else is touching. |
| +const tickEvery = 15 * time.Second |
| + |
| +// tickForever is started by App and not by NewForTest, for the reason |
| +// sweepForever is: a ticker per test app is a goroutine per test that |
| +// outlives it. Tests drive it with TestHandle.TickForTest. |
| +func (a *app) tickForever(ctx context.Context) { |
| + t := time.NewTicker(tickEvery) |
| + defer t.Stop() |
| + for { |
| + select { |
| + case <-ctx.Done(): |
| + return |
| + case <-t.C: |
| + a.hub.Tick(ctx) |
| + } |
| + } |
| +} |
| + |
| // build wires the whole app: refusals, schema, router, static files. It |
| // returns the *app alongside the mux so tests can reach into it (see |
| // NewForTest); production code uses App, which discards the *app. |
| @@ -170,12 +223,24 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err |
| rc.Logger.Info("migrate", "applied", migrated.Applied, |
| "skipped", migrated.Skipped, "adopted", migrated.Adopted) |
| - st := &store{g: d.G, sealer: sealer, now: time.Now, logger: rc.Logger} |
| + st := &store{g: d.G, sealer: sealer, clk: newClock(time.Now), logger: rc.Logger} |
| a := &app{ |
| db: d, cfg: rc, sealer: sealer, store: st, logger: rc.Logger, |
| linkBudget: newBudget(5, 15*time.Minute), |
| outsiderBudget: newBudget(10, 15*time.Minute), |
| } |
| + salt := make([]byte, 16) |
| + if _, err := rand.Read(salt); err != nil { |
| + return nil, nil, err |
| + } |
| + a.hubSalt = hex.EncodeToString(salt) |
| + a.heartbeat.Store(int64(defaultHeartbeat)) |
| + a.ttl.Store(int64(defaultStreamTTL)) |
| + // The hub reads standing back through the app, and never under its |
| + // own lock: hubStanding is a database read. |
| + a.hub = hub.New(hub.Config{ |
| + Now: st.now, Logger: rc.Logger, StandingFor: a.hubStanding, |
| + }) |
| writer, err := d.G.DB() |
| if err != nil { |
| @@ -222,6 +287,7 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err |
| r.Get("/meetings/{sqid}/end", a.confirmEnd) |
| r.Post("/meetings/{sqid}/end", a.endMeeting) |
| }) |
| + r.Get("/m/{sqid}/stream", a.stream) |
| }) |
| mux := http.NewServeMux() |
| @@ -284,6 +350,7 @@ func App(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *Runtime, e |
| return nil, nil, err |
| } |
| go a.sweepForever(ctx) |
| + go a.tickForever(ctx) |
| return mux, &Runtime{NextDue: a.nextDue}, nil |
| } |
diff --git a/internal/meet/auth_test.go b/internal/meet/auth_test.go| index fc65421..53f3c5e 100644 |
| --- a/internal/meet/auth_test.go |
| +++ b/internal/meet/auth_test.go |
| @@ -31,7 +31,7 @@ func testApp(t *testing.T, members ...string) *app { |
| t.Fatal(err) |
| } |
| now := time.Date(2026, 9, 8, 10, 0, 0, 0, time.UTC) |
| - a.store.now = func() time.Time { return now } |
| + a.store.clk.set(func() time.Time { return now }) |
| return a |
| } |
| @@ -111,7 +111,7 @@ func TestStandingTable(t *testing.T) { |
| } |
| if tc.expired { |
| after := a.store.now().Add(grantLife + time.Minute) |
| - a.store.now = func() time.Time { return after } |
| + a.store.clk.set(func() time.Time { return after }) |
| } |
| got, err := a.standingFor(ctx, p, m) |
| if err != nil { |
diff --git a/internal/meet/handlers_meetings.go b/internal/meet/handlers_meetings.go| index 8375f2e..c55d544 100644 |
| --- a/internal/meet/handlers_meetings.go |
| +++ b/internal/meet/handlers_meetings.go |
| @@ -198,5 +198,8 @@ func (a *app) endMeeting(w http.ResponseWriter, r *http.Request) { |
| ErrorPage(w, r, http.StatusInternalServerError, "") |
| return |
| } |
| + // The row is the authority; the hub is told so every open stream |
| + // gets lost{ended} now rather than at the next tick (spec §2.4). |
| + a.hub.End(m.Sqid) |
| http.Redirect(w, r, "/", http.StatusSeeOther) |
| } |
diff --git a/internal/meet/hub/hub.go b/internal/meet/hub/hub.go| index 9749dfb..10a7f89 100644 |
| --- a/internal/meet/hub/hub.go |
| +++ b/internal/meet/hub/hub.go |
| @@ -229,7 +229,14 @@ func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) { |
| // Detach starts the resume window for a participation whose stream |
| // closed. Its subscriber is closed; a resume attaches a new one. |
| -func (h *Hub) Detach(sqid string, peer PeerID) { |
| +// |
| +// sub is the subscriber the caller was reading, and the detach happens |
| +// only while that is still the participation's attachment. A stream |
| +// handler calls this on every exit, including the exit a Resume from a |
| +// second stream caused (lost{replaced}), and that displaced handler |
| +// must not detach the replacement that took its place. A nil sub |
| +// detaches whatever is attached, for a caller that holds no subscriber. |
| +func (h *Hub) Detach(sqid string, peer PeerID, sub *Sub) { |
| h.mu.Lock() |
| defer h.mu.Unlock() |
| r, ok := h.rooms[sqid] |
| @@ -240,6 +247,9 @@ func (h *Hub) Detach(sqid string, peer PeerID) { |
| if !ok { |
| return |
| } |
| + if sub != nil && p.sub != sub { |
| + return |
| + } |
| if p.sub != nil { |
| p.sub.terminate(Message{Type: TypeLost, Reason: ReasonBuffer}) |
| p.sub = nil |
diff --git a/internal/meet/hub/hub_test.go b/internal/meet/hub/hub_test.go| index 7706991..d6d287e 100644 |
| --- a/internal/meet/hub/hub_test.go |
| +++ b/internal/meet/hub/hub_test.go |
| @@ -51,7 +51,7 @@ func TestOneLivePeerPerSessionPerRoom(t *testing.T) { |
| if err != ErrSessionHasPeer { |
| t.Fatalf("second stream: %v", err) |
| } |
| - h.Detach("r1", a.Peer) |
| + h.Detach("r1", a.Peer, nil) |
| if _, err := h.Open("r1", "sessA", 1, "Ann", KindMember); err != ErrSessionHasPeer { |
| t.Fatal("a detached peer inside its resume window should still block a fresh stream") |
| } |
| @@ -142,3 +142,33 @@ func TestTerminateDoesNotEvictWhenNotFull(t *testing.T) { |
| t.Fatalf("second event %+v, want lost/removed", got[1]) |
| } |
| } |
| + |
| +// A second stream that resumes an attached participation displaces the |
| +// first (lost{replaced}, ruling above). The displaced handler then does |
| +// what every stream handler does on its way out — Detach — and that |
| +// must not take the replacement's attachment down with it. |
| +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) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if ev := drain(a.Sub); len(ev) != 1 || ev[0].Reason != ReasonReplaced { |
| + t.Fatalf("displaced stream got %+v", ev) |
| + } |
| + h.Detach("r1", a.Peer, a.Sub) // the displaced handler, on its way out |
| + if res.Sub.Lost() { |
| + t.Fatal("the replacement stream was closed by the displaced one's Detach") |
| + } |
| + p, ok := h.Find("r1", "sa", a.Peer) |
| + if !ok || p.sub != res.Sub || !p.resumeUntil.IsZero() { |
| + t.Fatalf("participation lost its attachment: %+v", p) |
| + } |
| + // The replacement's own Detach still detaches. |
| + h.Detach("r1", a.Peer, res.Sub) |
| + if !res.Sub.Lost() || !p.resumeUntil.Equal(now.Add(ResumeWindow)) { |
| + t.Fatalf("the attached stream's Detach did nothing: %+v", p) |
| + } |
| +} |
diff --git a/internal/meet/hub/relay.go b/internal/meet/hub/relay.go| index ffcb8f9..614decf 100644 |
| --- a/internal/meet/hub/relay.go |
| +++ b/internal/meet/hub/relay.go |
| @@ -117,24 +117,45 @@ func (e *ResumeError) Error() string { |
| return fmt.Sprintf("hub: resume refused %d %s", e.Status, e.Reason) |
| } |
| -// 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) { |
| - h.mu.Lock() |
| - defer h.mu.Unlock() |
| +// resumeRefusalLocked is spec §4.5's refusal ladder, and the only |
| +// place it is written: Resume applies it before attaching, and |
| +// CanResume applies it and stops there. |
| +func (h *Hub) resumeRefusalLocked(sqid, session string, peer PeerID) *ResumeError { |
| if t, ok := h.tombstones[peer]; ok { |
| - return ResumeResult{}, &ResumeError{Status: 410, Reason: t.reason} |
| + return &ResumeError{Status: 410, Reason: t.reason} |
| } |
| if _, ok := h.rooms[sqid]; !ok { |
| - return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot} |
| + return &ResumeError{Status: 404, Boot: h.boot} |
| } |
| - p, found := h.findLocked(sqid, session, peer) |
| - if !found { |
| + if _, found := h.findLocked(sqid, session, peer); !found { |
| // Unknown here, unknown in another session, unknown in another |
| // room: one answer, so nothing leaks (spec §4.2). |
| - return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot} |
| + return &ResumeError{Status: 404, Boot: h.boot} |
| + } |
| + return nil |
| +} |
| + |
| +// CanResume answers what a resume would answer, and does nothing else: |
| +// it does not attach a subscriber, does not clear the resume window and |
| +// 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 { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + 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) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + if re := h.resumeRefusalLocked(sqid, session, peer); re != nil { |
| + return ResumeResult{}, re |
| } |
| + p, _ := h.findLocked(sqid, session, peer) |
| r := h.rooms[sqid] |
| if p.sub != nil { |
| // A second stream for the same peer replaces the attachment only: |
diff --git a/internal/meet/hub/relay_test.go b/internal/meet/hub/relay_test.go| index 1dd532a..8ba5a0b 100644 |
| --- a/internal/meet/hub/relay_test.go |
| +++ b/internal/meet/hub/relay_test.go |
| @@ -92,7 +92,7 @@ func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) { |
| h.Relay("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) |
| + h.Detach("r1", a.Peer, nil) |
| *now = now.Add(5 * time.Second) |
| res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3}) |
| if err != nil { |
| @@ -117,7 +117,7 @@ func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) { |
| // A waiting guest resumes and learns nothing. |
| g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| h.Join(ctx, "r1", "sg", g.Peer) |
| - h.Detach("r1", g.Peer) |
| + h.Detach("r1", g.Peer, nil) |
| gres, err := h.Resume("r1", "sg", g.Peer, nil) |
| if err != nil { |
| t.Fatal(err) |
| @@ -133,7 +133,7 @@ func TestResumeGapWhenTheRingRolledOver(t *testing.T) { |
| h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| } |
| drain(a.Sub) |
| - h.Detach("r1", a.Peer) |
| + h.Detach("r1", a.Peer, nil) |
| *now = now.Add(time.Second) |
| res, _ := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2}) |
| if len(res.Gap) != 1 || res.Gap[0] != b.Peer || len(res.Replay) != 0 { |
| @@ -157,7 +157,7 @@ func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) { |
| h.Relay("r1", "sc", c.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| } |
| drain(a.Sub) |
| - h.Detach("r1", a.Peer) |
| + 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}) |
| if err != nil { |
| @@ -224,7 +224,7 @@ func TestResumeAnswers(t *testing.T) { |
| func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) { |
| h, now, a, _ := twoAdmitted(t) |
| - h.Detach("r1", a.Peer) |
| + h.Detach("r1", a.Peer, nil) |
| *now = now.Add(ResumeWindow + time.Second) |
| h.Tick(context.Background()) |
| _, err := h.Resume("r1", "sa", a.Peer, nil) |
diff --git a/internal/meet/joincookie_test.go b/internal/meet/joincookie_test.go| index 32dc66b..194c5d3 100644 |
| --- a/internal/meet/joincookie_test.go |
| +++ b/internal/meet/joincookie_test.go |
| @@ -9,7 +9,7 @@ import ( |
| ) |
| func cookieApp(now time.Time) *app { |
| - return &app{cfg: resolved{instanceKey: "k", Config: Config{Origin: "http://app.test"}}, store: &store{now: func() time.Time { return now }}} |
| + return &app{cfg: resolved{instanceKey: "k", Config: Config{Origin: "http://app.test"}}, store: &store{clk: newClock(func() time.Time { return now })}} |
| } |
| func TestJoinCookieRoundTripsAndRejectsTampering(t *testing.T) { |
diff --git a/internal/meet/participation.go b/internal/meet/participation.go| new file mode 100644 |
| index 0000000..cd94c80 |
| --- /dev/null |
| +++ b/internal/meet/participation.go |
| @@ -0,0 +1,46 @@ |
| +package meet |
| + |
| +import ( |
| + "context" |
| + |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| +) |
| + |
| +// hubStanding is hub.Config.StandingFor: the app's standing (spec §2.3) |
| +// in the two facts the hub keeps — the kind a participation holds, and |
| +// whether the person still has entry authority for the room at all. |
| +// |
| +// It is never called under the hub's lock: the hub drops the lock |
| +// before every standing read, because this reads the database. |
| +// |
| +// A meeting that has ended, or that no longer exists, answers ok=false |
| +// with no error, which is what a lapsed grant answers too: the tick |
| +// then applies §2.4's transitions rather than treating it as a failure |
| +// to retry on the next tick. |
| +func (a *app) hubStanding(ctx context.Context, sqid string, personID int64) (hub.Kind, bool, error) { |
| + m, ok, err := a.store.meetingBySqid(ctx, sqid) |
| + if err != nil { |
| + return hub.KindGuest, false, err |
| + } |
| + if !ok || m.EndedAt != nil { |
| + return hub.KindGuest, false, nil |
| + } |
| + p, ok, err := a.store.personByID(ctx, personID) |
| + if err != nil { |
| + return hub.KindGuest, false, err |
| + } |
| + if !ok { |
| + return hub.KindGuest, false, nil |
| + } |
| + st, err := a.standingFor(ctx, p, m) |
| + if err != nil { |
| + return hub.KindGuest, false, err |
| + } |
| + switch st { |
| + case StandingMember: |
| + return hub.KindMember, true, nil |
| + case StandingGuest: |
| + return hub.KindGuest, true, nil |
| + } |
| + return hub.KindGuest, false, nil |
| +} |
diff --git a/internal/meet/store.go b/internal/meet/store.go| index f6833a8..7866fe0 100644 |
| --- a/internal/meet/store.go |
| +++ b/internal/meet/store.go |
| @@ -7,6 +7,7 @@ import ( |
| "fmt" |
| "log/slog" |
| "strings" |
| + "sync/atomic" |
| "time" |
| "gorm.io/gorm" |
| @@ -22,16 +23,35 @@ const grantLife = 2 * time.Hour |
| // pendingLife matches the pending-join cookie: fifteen minutes. |
| const pendingLife = 15 * time.Minute |
| +// clock is the store's time source. It is a value tests move (see |
| +// TestHandle.SetClock) while request goroutines are reading it — a |
| +// stream handler outlives the request that opened it — so the function |
| +// lives behind an atomic rather than in a plain field. |
| +type clock struct{ fn atomic.Value } |
| + |
| +func newClock(f func() time.Time) *clock { |
| + c := &clock{} |
| + c.set(f) |
| + return c |
| +} |
| + |
| +func (c *clock) set(f func() time.Time) { c.fn.Store(f) } |
| + |
| +func (c *clock) now() time.Time { return c.fn.Load().(func() time.Time)() } |
| + |
| // store is every database access in the app. Sealing happens here and |
| // nowhere else, so a handler never holds a plaintext title it did not |
| // get from the store. |
| type store struct { |
| g *gorm.DB |
| sealer *seal.Sealer |
| - now func() time.Time |
| + clk *clock |
| logger *slog.Logger |
| } |
| +// now is the app's clock, the one a test can move. |
| +func (s *store) now() time.Time { return s.clk.now() } |
| + |
| func actorFor(p Person) string { return fmt.Sprintf("person:%d", p.ID) } |
| // ensurePerson reads first and writes only when the read is empty: |
| @@ -142,6 +162,17 @@ func (s *store) meetingBySqid(ctx context.Context, sqid string) (Meeting, bool, |
| return m, err == nil, err |
| } |
| +// personByID is the hub's half of standing: the tick and the entry |
| +// checks hold a person id, not a request, so the row is read back here. |
| +func (s *store) personByID(ctx context.Context, id int64) (Person, bool, error) { |
| + var p Person |
| + err := s.g.WithContext(ctx).Where("id = ?", id).Take(&p).Error |
| + if errors.Is(err, gorm.ErrRecordNotFound) { |
| + return Person{}, false, nil |
| + } |
| + return p, err == nil, err |
| +} |
| + |
| // liveLink answers one false for a token that never existed, one that |
| // was revoked, and one whose meeting has ended (spec §3.2's one 404). |
| func (s *store) liveLink(ctx context.Context, token string) (MeetingLink, Meeting, bool, error) { |
| @@ -267,7 +298,7 @@ func (s *store) mintGrantViaLink(ctx context.Context, linkID, personID int64) (G |
| var g GuestGrant |
| live := false |
| err := s.g.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| - ts := &store{g: tx, sealer: s.sealer, now: s.now, logger: s.logger} |
| + ts := &store{g: tx, sealer: s.sealer, clk: s.clk, logger: s.logger} |
| _, m, ok, err := ts.liveLinkByID(ctx, linkID) |
| if err != nil || !ok { |
| return err |
| @@ -350,7 +381,7 @@ func (s *store) deletePending(ctx context.Context, address string) error { |
| func (s *store) consumePending(ctx context.Context, p Person, pend PendingJoin, member bool) (string, error) { |
| dest := "" |
| err := s.g.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| - ts := &store{g: tx, sealer: s.sealer, now: s.now, logger: s.logger} |
| + ts := &store{g: tx, sealer: s.sealer, clk: s.clk, logger: s.logger} |
| current, ok, err := ts.latestPending(ctx, p.Email) |
| if err != nil || !ok || current.ID != pend.ID { |
| return err |
diff --git a/internal/meet/store_test.go b/internal/meet/store_test.go| index e235894..7841f1f 100644 |
| --- a/internal/meet/store_test.go |
| +++ b/internal/meet/store_test.go |
| @@ -28,7 +28,7 @@ func testStore(t *testing.T) (*store, *time.Time) { |
| } |
| sl, _ := seal.New(bytes.Repeat([]byte{9}, 32)) |
| now := time.Date(2026, 9, 8, 10, 0, 0, 0, time.UTC) |
| - return &store{g: d.G, sealer: sl, now: func() time.Time { return now }, logger: logger}, &now |
| + return &store{g: d.G, sealer: sl, clk: newClock(func() time.Time { return now }), logger: logger}, &now |
| } |
| func TestCreateMeetingSealsTitleAndMintsOneLiveLink(t *testing.T) { |
diff --git a/internal/meet/stream.go b/internal/meet/stream.go| new file mode 100644 |
| index 0000000..0566791 |
| --- /dev/null |
| +++ b/internal/meet/stream.go |
| @@ -0,0 +1,285 @@ |
| +package meet |
| + |
| +import ( |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "fmt" |
| + "net/http" |
| + "time" |
| + |
| + "github.com/go-chi/chi/v5" |
| + |
| + "amadan.net/rastrillo/rastrillo/sessions" |
| + |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| +) |
| + |
| +const ( |
| + // defaultHeartbeat and defaultStreamTTL are Docs' live-sync numbers |
| + // (spec §4.1): a comment line every fifteen seconds so a dead stream |
| + // is noticed, and a five-minute life so hibernation is never held |
| + // off by one. The browser resumes; nothing is lost at the TTL. |
| + defaultHeartbeat = 15 * time.Second |
| + defaultStreamTTL = 5 * time.Minute |
| + streamWriteWindow = 30 * time.Second |
| +) |
| + |
| +// sessionKey is the session's identity for the hub (spec §4.2): the |
| +// session cookie this request presented, keyed through a salt this |
| +// process chose at boot, so what the hub holds is neither the token nor |
| +// the hash rastrillo stores. Two tabs of one browser share it, which is |
| +// what one peer per room and its takeover need; two browsers of one |
| +// person do not, which §4.4's four streams per person allows. |
| +// |
| +// The plan expected the subject plus the session's creation instant. |
| +// That is not an identity: two sign-ins of one person within one second |
| +// collide, and the second browser is then refused the room. |
| +func (a *app) sessionKey(r *http.Request) (string, bool) { |
| + if s, ok := sessions.Current(r); !ok || s.Subject == "" { |
| + return "", false |
| + } |
| + c, err := r.Cookie(a.auth.SessionCookie()) |
| + if err != nil || c.Value == "" { |
| + return "", false |
| + } |
| + sum := sha256.Sum256([]byte(a.hubSalt + "|" + c.Value)) |
| + return hex.EncodeToString(sum[:]), true |
| +} |
| + |
| +// streamJSON is every refusal this endpoint makes. The room's client is |
| +// a script, not a page: spec §4.5 fixes the browser's behaviour per |
| +// status and body, so a refusal answers JSON rather than the error page |
| +// a screen would get. |
| +func streamJSON(w http.ResponseWriter, status int, body map[string]any) { |
| + w.Header().Set("Content-Type", "application/json") |
| + w.Header().Set("Cache-Control", "no-store") |
| + w.WriteHeader(status) |
| + if body != nil { |
| + _ = json.NewEncoder(w).Encode(body) |
| + } |
| +} |
| + |
| +func (a *app) refuseResume(w http.ResponseWriter, re *hub.ResumeError) { |
| + body := map[string]any{} |
| + if re.Reason != "" { |
| + body["reason"] = string(re.Reason) |
| + } |
| + if re.Boot != "" { |
| + body["boot"] = re.Boot |
| + } |
| + streamJSON(w, re.Status, body) |
| +} |
| + |
| +// stream is GET /m/{sqid}/stream: one SSE stream per peer (spec §4.1), |
| +// opened fresh on standing or resumed on a participation bound to this |
| +// session (§2.5). ?probe=1 answers the same statuses and stops there. |
| +func (a *app) stream(w http.ResponseWriter, r *http.Request) { |
| + sqid := chi.URLParam(r, "sqid") |
| + person, _ := currentPerson(r) |
| + session, ok := a.sessionKey(r) |
| + if !ok { |
| + a.logger.Error("meet: stream with no session subject", "path", r.URL.Path) |
| + streamJSON(w, http.StatusInternalServerError, nil) |
| + return |
| + } |
| + probe := r.URL.Query().Get("probe") == "1" |
| + |
| + var sub *hub.Sub |
| + var peer hub.PeerID |
| + var first []hub.Message |
| + |
| + if resume := hub.PeerID(r.URL.Query().Get("peer")); resume != "" { |
| + // A resume is authorized by the participation, not by standing |
| + // (§2.5), but the meeting itself must still be live: End writes |
| + // no headstone for a room nobody opened, so ended_at is the only |
| + // authority a fresh process has. |
| + m, found, err := a.store.meetingBySqid(r.Context(), sqid) |
| + if err != nil { |
| + a.logger.Error("meet: stream: load meeting", "err", err) |
| + streamJSON(w, http.StatusInternalServerError, nil) |
| + return |
| + } |
| + if !found || m.EndedAt != nil { |
| + // 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 { |
| + a.refuseResume(w, re) |
| + return |
| + } |
| + a.refuseResume(w, &hub.ResumeError{Status: http.StatusNotFound, Boot: a.hub.Boot()}) |
| + return |
| + } |
| + if probe { |
| + if re := a.hub.CanResume(sqid, session, resume); re != nil { |
| + a.refuseResume(w, re) |
| + return |
| + } |
| + w.WriteHeader(http.StatusNoContent) |
| + return |
| + } |
| + var last map[hub.PeerID]uint64 |
| + if raw := r.URL.Query().Get("last"); raw != "" { |
| + if err := json.Unmarshal([]byte(raw), &last); err != nil { |
| + streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"}) |
| + return |
| + } |
| + } |
| + res, err := a.hub.Resume(sqid, session, resume, last) |
| + if err != nil { |
| + re, ok := err.(*hub.ResumeError) |
| + if !ok { |
| + a.logger.Error("meet: stream: resume", "err", err) |
| + streamJSON(w, http.StatusInternalServerError, nil) |
| + return |
| + } |
| + a.refuseResume(w, re) |
| + return |
| + } |
| + sub, peer = res.Sub, resume |
| + // Order matters: the roster, then the replay, then the gaps, all |
| + // written before anything the live subscriber carries. The |
| + // subscriber is live from the moment Resume returned, and the |
| + // browser discards what is at or below the seq it already has, |
| + // so a fresh candidate written ahead of the replay would make |
| + // the replay look stale. |
| + first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot()}) |
| + if res.Admitted { |
| + first = append(first, hub.Message{Type: hub.TypeRoster, Roster: res.Roster}) |
| + first = append(first, res.Replay...) |
| + for _, from := range res.Gap { |
| + first = append(first, hub.Message{Type: hub.TypeGap, From: from}) |
| + } |
| + } else { |
| + // Waiting or unjoined: nothing about the room (§4.5). |
| + first = append(first, hub.Message{Type: hub.TypeState, Waiting: true}) |
| + } |
| + } else { |
| + m, found, err := a.store.meetingBySqid(r.Context(), sqid) |
| + if err != nil { |
| + a.logger.Error("meet: stream: load meeting", "err", err) |
| + streamJSON(w, http.StatusInternalServerError, nil) |
| + return |
| + } |
| + if !found || m.EndedAt != nil { |
| + streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()}) |
| + return |
| + } |
| + st, err := a.standingFor(r.Context(), person, m) |
| + if err != nil { |
| + a.logger.Error("meet: stream: standing", "err", err) |
| + streamJSON(w, http.StatusInternalServerError, nil) |
| + return |
| + } |
| + if st == StandingNone { |
| + // Not this person's meeting to enter: 404, never 403 (§2.5). |
| + streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()}) |
| + return |
| + } |
| + if probe { |
| + if existing, live := a.hub.LiveForSession(sqid, session); live { |
| + streamJSON(w, http.StatusConflict, map[string]any{"peer": string(existing)}) |
| + return |
| + } |
| + w.WriteHeader(http.StatusNoContent) |
| + return |
| + } |
| + kind := hub.KindGuest |
| + if st == StandingMember { |
| + kind = hub.KindMember |
| + } |
| + res, err := a.hub.Open(sqid, session, person.ID, person.Name, kind) |
| + switch err { |
| + case nil: |
| + case hub.ErrSessionHasPeer: |
| + streamJSON(w, http.StatusConflict, map[string]any{"peer": string(res.Existing)}) |
| + return |
| + case hub.ErrRoomFull, hub.ErrPersonStreams: |
| + streamJSON(w, http.StatusServiceUnavailable, map[string]any{"error": "full"}) |
| + return |
| + default: |
| + // ErrUnknownPeer here is the room's headstone: the meeting |
| + // ended between the read above and the hub lock. |
| + streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()}) |
| + return |
| + } |
| + sub, peer = res.Sub, res.Peer |
| + first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot()}) |
| + } |
| + |
| + // From here the participation exists and is this handler's to hold |
| + // open. Every exit detaches it — this handler's own subscriber, so a |
| + // stream displaced by a second one does not detach its replacement — |
| + // which starts the thirty-second resume window (§4.5). |
| + defer a.hub.Detach(sqid, peer, sub) |
| + a.serveStream(w, r, sub, first) |
| +} |
| + |
| +// serveStream writes the events already decided, then the subscriber's, |
| +// until the client goes, the TTL expires, or the stream is lost. |
| +func (a *app) serveStream(w http.ResponseWriter, r *http.Request, sub *hub.Sub, first []hub.Message) { |
| + rc := http.NewResponseController(w) |
| + w.Header().Set("Content-Type", "text/event-stream") |
| + w.Header().Set("Cache-Control", "no-store") |
| + w.Header().Set("Connection", "keep-alive") |
| + // Nothing between this process and the browser should buffer an |
| + // event stream; nginx and its imitators read this header. |
| + w.Header().Set("X-Accel-Buffering", "no") |
| + w.WriteHeader(http.StatusOK) |
| + if err := rc.Flush(); err != nil { |
| + return |
| + } |
| + for _, m := range first { |
| + if !writeEvent(w, rc, m) { |
| + return |
| + } |
| + } |
| + heartbeat := time.NewTicker(a.heartbeatEvery()) |
| + defer heartbeat.Stop() |
| + ttl := time.NewTimer(a.streamLife()) |
| + defer ttl.Stop() |
| + for { |
| + select { |
| + case <-r.Context().Done(): |
| + return |
| + case <-ttl.C: |
| + return |
| + case <-heartbeat.C: |
| + if !writeRaw(w, rc, ": hb\n\n") { |
| + return |
| + } |
| + case m, ok := <-sub.C(): |
| + if !ok { |
| + return |
| + } |
| + if !writeEvent(w, rc, m) { |
| + return |
| + } |
| + if m.Type == hub.TypeLost || m.Type == hub.TypeEnded { |
| + // Terminal for this stream either way: lost{buffer} is |
| + // followed by a resume, the rest by a screen that says |
| + // why (§4.5). |
| + return |
| + } |
| + } |
| + } |
| +} |
| + |
| +func writeEvent(w http.ResponseWriter, rc *http.ResponseController, m hub.Message) bool { |
| + b, err := json.Marshal(m) |
| + if err != nil { |
| + return false |
| + } |
| + return writeRaw(w, rc, fmt.Sprintf("event: %s\ndata: %s\n\n", m.Type, b)) |
| +} |
| + |
| +// writeRaw writes one SSE frame under a deadline, so a client that has |
| +// stopped reading cannot hold a goroutine and a participation open. |
| +func writeRaw(w http.ResponseWriter, rc *http.ResponseController, s string) bool { |
| + _ = rc.SetWriteDeadline(time.Now().Add(streamWriteWindow)) |
| + if _, err := fmt.Fprint(w, s); err != nil { |
| + return false |
| + } |
| + return rc.Flush() == nil |
| +} |
diff --git a/internal/meettest/harness_test.go b/internal/meettest/harness_test.go| index ecb8576..2ce22b6 100644 |
| --- a/internal/meettest/harness_test.go |
| +++ b/internal/meettest/harness_test.go |
| @@ -40,6 +40,27 @@ type harness struct { |
| mail *recordingMailer |
| ctl *meet.TestHandle |
| now time.Time |
| + |
| + // t and srv are the streaming half: an SSE stream cannot be read |
| + // from an httptest.ResponseRecorder, which only hands its bytes |
| + // back when the handler has returned, so the tests that read one |
| + // go through a real server over a real socket. It is started on |
| + // first use and closed after every stream this test opened. |
| + t *testing.T |
| + srv *httptest.Server |
| +} |
| + |
| +// server is the real HTTP server the stream tests talk to. Its origin |
| +// is not testOrigin — it is 127.0.0.1 on whatever port the kernel gave |
| +// it — which does not matter: csrf.Protect compares the Origin header |
| +// the client sends against the configured origin, and the client sends |
| +// testOrigin, exactly as a browser on the app's own origin would. |
| +func (x *harness) server() *httptest.Server { |
| + if x.srv == nil { |
| + x.srv = httptest.NewServer(x.h) |
| + x.t.Cleanup(x.srv.Close) |
| + } |
| + return x.srv |
| } |
| func (x *harness) sqlDB() *sql.DB { s, _ := x.db.G.DB(); return s } |
| @@ -133,7 +154,7 @@ func newHarness(t *testing.T, members ...string) *harness { |
| // test can move; rastrillo stamps its auth_links rows from the real |
| // one. A fixed base put the two most of a day apart, so nothing that |
| // compares them could mean anything — see the auth_links sweep test. |
| - x := &harness{h: mux, db: d, dbPath: dbPath, mail: rec, ctl: ctl, now: time.Now().UTC()} |
| + x := &harness{h: mux, db: d, dbPath: dbPath, mail: rec, ctl: ctl, now: time.Now().UTC(), t: t} |
| x.advance(0) |
| return x |
| } |
diff --git a/internal/meettest/sse_test.go b/internal/meettest/sse_test.go| new file mode 100644 |
| index 0000000..4ecdef9 |
| --- /dev/null |
| +++ b/internal/meettest/sse_test.go |
| @@ -0,0 +1,181 @@ |
| +package meettest |
| + |
| +import ( |
| + "bufio" |
| + "context" |
| + "encoding/json" |
| + "io" |
| + "net/http" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// sseReader is a browser's EventSource, near enough: one open GET |
| +// against the harness's real server, its lines parsed into events on a |
| +// channel while the test goroutine reads them one at a time. |
| +type sseReader struct { |
| + t *testing.T |
| + cancel context.CancelFunc |
| + body io.ReadCloser |
| + events chan sseEvent |
| + hb chan struct{} |
| + done chan struct{} |
| +} |
| + |
| +type sseEvent struct { |
| + name string |
| + data []byte |
| +} |
| + |
| +// stream opens an SSE stream and fails the test unless the server |
| +// accepted it. A refusal is read with c.get instead, which needs no |
| +// socket: the status and the JSON body are the whole answer (§4.5). |
| +func (c *client) stream(target string) *sseReader { |
| + c.t.Helper() |
| + srv := c.x.server() |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+target, nil) |
| + if err != nil { |
| + cancel() |
| + c.t.Fatal(err) |
| + } |
| + for _, ck := range c.cookies { |
| + req.AddCookie(ck) |
| + } |
| + req.Header.Set("Accept", "text/event-stream") |
| + resp, err := srv.Client().Do(req) |
| + if err != nil { |
| + cancel() |
| + c.t.Fatalf("GET %s: %v", target, err) |
| + } |
| + if resp.StatusCode != http.StatusOK { |
| + b, _ := io.ReadAll(resp.Body) |
| + resp.Body.Close() |
| + cancel() |
| + c.t.Fatalf("GET %s: %d\n%s", target, resp.StatusCode, b) |
| + } |
| + if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") { |
| + resp.Body.Close() |
| + cancel() |
| + c.t.Fatalf("GET %s: Content-Type %q", target, ct) |
| + } |
| + s := &sseReader{ |
| + t: c.t, cancel: cancel, body: resp.Body, |
| + events: make(chan sseEvent, 64), hb: make(chan struct{}, 64), done: make(chan struct{}), |
| + } |
| + c.t.Cleanup(s.close) |
| + go s.read() |
| + return s |
| +} |
| + |
| +func (s *sseReader) read() { |
| + defer close(s.done) |
| + sc := bufio.NewScanner(s.body) |
| + sc.Buffer(make([]byte, 0, 64*1024), 1<<20) |
| + var ev sseEvent |
| + for sc.Scan() { |
| + line := sc.Text() |
| + switch { |
| + case line == "": |
| + if ev.name != "" { |
| + select { |
| + case s.events <- ev: |
| + default: |
| + } |
| + } |
| + ev = sseEvent{} |
| + case strings.HasPrefix(line, ":"): |
| + select { |
| + case s.hb <- struct{}{}: |
| + default: |
| + } |
| + case strings.HasPrefix(line, "event: "): |
| + ev.name = strings.TrimPrefix(line, "event: ") |
| + case strings.HasPrefix(line, "data: "): |
| + ev.data = []byte(strings.TrimPrefix(line, "data: ")) |
| + } |
| + } |
| +} |
| + |
| +// next waits for one event. |
| +func (s *sseReader) next(timeout time.Duration) (string, []byte, bool) { |
| + select { |
| + case ev := <-s.events: |
| + return ev.name, ev.data, true |
| + case <-time.After(timeout): |
| + return "", nil, false |
| + } |
| +} |
| + |
| +// hello reads the first event, which is always hello (spec §4.1), and |
| +// returns the peer id and the boot id it carries. |
| +func (s *sseReader) hello() (peer, boot string) { |
| + s.t.Helper() |
| + name, data, ok := s.next(2 * time.Second) |
| + if !ok || name != "hello" { |
| + s.t.Fatalf("first event is %q (ok=%v), want hello", name, ok) |
| + } |
| + var body struct{ Peer, Boot string } |
| + if err := json.Unmarshal(data, &body); err != nil { |
| + s.t.Fatalf("hello %s: %v", data, err) |
| + } |
| + return body.Peer, body.Boot |
| +} |
| + |
| +// waitHeartbeat waits for one SSE comment line, which is the only thing |
| +// the server sends on an idle stream. |
| +func (s *sseReader) waitHeartbeat(timeout time.Duration) bool { |
| + select { |
| + case <-s.hb: |
| + return true |
| + case <-time.After(timeout): |
| + return false |
| + } |
| +} |
| + |
| +// waitClosed waits for the server to end the stream. |
| +func (s *sseReader) waitClosed(timeout time.Duration) bool { |
| + select { |
| + case <-s.done: |
| + return true |
| + case <-time.After(timeout): |
| + return false |
| + } |
| +} |
| + |
| +// close is what a browser navigating away does. It waits for the read |
| +// goroutine to finish so the test never races the server's own exit. |
| +func (s *sseReader) close() { |
| + s.cancel() |
| + s.body.Close() |
| + select { |
| + case <-s.done: |
| + case <-time.After(2 * time.Second): |
| + s.t.Error("the stream reader did not stop") |
| + } |
| +} |
| + |
| +// signal is POST /m/{sqid}/signal, the upstream half of §4.1: a JSON |
| +// body carrying the peer and the message, under CSRF. The endpoint |
| +// itself arrives with the signal handler; this is the client for it. |
| +func (c *client) signal(sqid, peer, body string) *http.Response { |
| + c.t.Helper() |
| + payload := `{"peer":"` + peer + `",` + strings.TrimPrefix(body, "{") |
| + req, err := http.NewRequest(http.MethodPost, c.x.server().URL+"/m/"+sqid+"/signal", strings.NewReader(payload)) |
| + if err != nil { |
| + c.t.Fatal(err) |
| + } |
| + for _, ck := range c.cookies { |
| + req.AddCookie(ck) |
| + } |
| + req.Header.Set("Content-Type", "application/json") |
| + req.Header.Set("Origin", testOrigin) |
| + resp, err := c.x.server().Client().Do(req) |
| + if err != nil { |
| + c.t.Fatalf("POST signal: %v", err) |
| + } |
| + io.Copy(io.Discard, resp.Body) |
| + resp.Body.Close() |
| + return resp |
| +} |
diff --git a/internal/meettest/stream_test.go b/internal/meettest/stream_test.go| new file mode 100644 |
| index 0000000..f75384d |
| --- /dev/null |
| +++ b/internal/meettest/stream_test.go |
| @@ -0,0 +1,275 @@ |
| +package meettest |
| + |
| +import ( |
| + "encoding/json" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// answer reads a refused stream: the status and the JSON body spec §4.5 |
| +// fixes the browser's behaviour on. A refusal never streams, so the |
| +// in-process recorder is the right client for it. |
| +func answer(t *testing.T, rec *httptest.ResponseRecorder) (peer, boot, reason string) { |
| + t.Helper() |
| + var body struct{ Peer, Boot, Reason string } |
| + if rec.Body.Len() > 0 { |
| + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { |
| + t.Fatalf("refusal body %q: %v", rec.Body.String(), err) |
| + } |
| + } |
| + return body.Peer, body.Boot, body.Reason |
| +} |
| + |
| +// Spec §4.1: the first event on a stream is hello, carrying the peer id |
| +// the server issued and the boot id of this process. Spec §4.2: a |
| +// session holds one peer per room, so a second fresh stream is refused |
| +// with 409 naming the peer it already has. |
| +func TestStreamSaysHelloWithPeerAndBootAndRefusesTheSecondFreshStream(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, boot := s.hello() |
| + if len(peer) != 32 || len(boot) != 32 { |
| + t.Fatalf("hello peer=%q boot=%q", peer, boot) |
| + } |
| + rec := c.get("/m/" + sqid + "/stream") |
| + if rec.Code != http.StatusConflict { |
| + t.Fatalf("second fresh stream: %d, want 409", rec.Code) |
| + } |
| + if named, _, _ := answer(t, rec); named != peer { |
| + t.Fatalf("409 names %q, want %q", named, peer) |
| + } |
| + // A second browser of the same person is a different session, so it |
| + // gets a peer of its own (§4.4 allows four streams per person). |
| + c2 := x.browser(t) |
| + c2.signIn("paul@example.com") |
| + s2 := c2.stream("/m/" + sqid + "/stream") |
| + defer s2.close() |
| + if peer2, boot2 := s2.hello(); peer2 == peer || boot2 != boot { |
| + t.Fatalf("second browser: peer %q boot %q", peer2, boot2) |
| + } |
| +} |
| + |
| +// Spec §2.5: a fresh stream is authorized by standing, a resume by the |
| +// participation bound to this session. Neither ever answers 403. |
| +func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + stranger := x.browser(t) |
| + if rec := stranger.get("/m/" + sqid + "/stream"); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("signed-out stream: %d, want the sign-in redirect", rec.Code) |
| + } |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + // A second meeting, which the guest below is never given a link to. |
| + elsewhere, _ := createMeeting(t, c, "Not yours", "") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, boot := s.hello() |
| + s.close() |
| + |
| + guest := x.browser(t) |
| + guest.get("/j/" + tok) |
| + guest.signIn("guest@example.com") |
| + if rec := guest.get("/m/" + elsewhere + "/stream"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("a meeting the guest has no standing for: %d, want 404", rec.Code) |
| + } |
| + rec := guest.get("/m/" + sqid + "/stream?peer=" + peer) |
| + if rec.Code != http.StatusNotFound { |
| + t.Fatalf("resume through another session: %d, want 404", rec.Code) |
| + } |
| + // The same answer an unknown peer gets, so nothing leaks (§4.2). |
| + if _, gotBoot, _ := answer(t, rec); gotBoot != boot { |
| + t.Fatalf("404 boot %q, want %q", gotBoot, boot) |
| + } |
| + unknown := guest.get("/m/" + sqid + "/stream?peer=0123456789abcdef0123456789abcdef") |
| + if unknown.Code != http.StatusNotFound || unknown.Body.String() != rec.Body.String() { |
| + t.Fatalf("unknown peer: %d %s", unknown.Code, unknown.Body.String()) |
| + } |
| +} |
| + |
| +// Spec §4.5: a closed stream leaves the participation in the hub for |
| +// thirty seconds, and a resume reattaches the same one — the same peer, |
| +// the same boot. This participation never joined, so it is told only |
| +// that it is waiting, which is all a peer outside the room may learn. |
| +func TestResumeReattachesTheSameParticipation(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, boot := s.hello() |
| + s.close() |
| + |
| + s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| + defer s2.close() |
| + p2, b2 := s2.hello() |
| + if p2 != peer || b2 != boot { |
| + t.Fatalf("resume hello %s %s, want %s %s", p2, b2, peer, boot) |
| + } |
| + ev, data, ok := s2.next(time.Second) |
| + if !ok || ev != "state" { |
| + t.Fatalf("after a resume: %s ok=%v", ev, ok) |
| + } |
| + var state struct{ Waiting bool } |
| + json.Unmarshal(data, &state) |
| + if !state.Waiting { |
| + t.Fatalf("state %s, want waiting", data) |
| + } |
| + // The resumed participation is still the session's one peer. |
| + if rec := c.get("/m/" + sqid + "/stream"); rec.Code != http.StatusConflict { |
| + t.Fatalf("fresh stream while resumed: %d, want 409", rec.Code) |
| + } |
| +} |
| + |
| +// Spec §4.1: the stream heartbeats, and dies at its TTL so hibernation |
| +// is never held off by one. The browser resumes and loses nothing. |
| +func TestTTLClosesTheStreamAndResumeWorksWithinTheWindow(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + x.ctl.SetStreamTimings(50*time.Millisecond, 300*time.Millisecond) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, _ := s.hello() |
| + if !s.waitHeartbeat(2 * time.Second) { |
| + t.Fatal("no heartbeat on an idle stream") |
| + } |
| + if !s.waitClosed(2 * time.Second) { |
| + t.Fatal("stream did not close at the TTL") |
| + } |
| + s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| + defer s2.close() |
| + if p, _ := s2.hello(); p != peer { |
| + t.Fatalf("resume after the TTL: %s, want %s", p, peer) |
| + } |
| +} |
| + |
| +// Spec §4.5: a resume that misses its window is 404 with the same boot |
| +// the browser last saw — the process did not restart, the window |
| +// lapsed — and the browser's answer is a fresh stream, which gets a |
| +// peer of its own because the old participation is gone. |
| +func TestLapsedResumeIs404WithTheSameBootAndAFreshStreamGetsANewPeer(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + // The stream is left to die at its own TTL rather than closed from |
| + // this side: the handler detaches on its way out, before the |
| + // response is finished, so an EOF here means the resume window has |
| + // already been stamped — and the clock this test then moves is |
| + // ahead of it, not behind. |
| + 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) |
| + x.ctl.TickForTest() |
| + |
| + rec := c.get("/m/" + sqid + "/stream?peer=" + peer) |
| + _, gotBoot, _ := answer(t, rec) |
| + if rec.Code != http.StatusNotFound || gotBoot != boot { |
| + t.Fatalf("lapsed resume: %d boot %q, want 404 and %q", rec.Code, gotBoot, boot) |
| + } |
| + 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, |
| +// must not clear the resume window and must not touch the ring. |
| +func TestResumeProbeAnswersWithoutAttaching(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + |
| + if rec := c.get("/m/" + sqid + "/stream?probe=1"); rec.Code != http.StatusNoContent { |
| + t.Fatalf("probe with nothing open: %d, want 204", rec.Code) |
| + } |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, boot := s.hello() |
| + rec := c.get("/m/" + sqid + "/stream?probe=1") |
| + named, _, _ := answer(t, rec) |
| + if rec.Code != http.StatusConflict || named != peer { |
| + t.Fatalf("probe for a fresh stream while one is open: %d %q", rec.Code, named) |
| + } |
| + s.close() |
| + |
| + if rec := c.get("/m/" + sqid + "/stream?probe=1&peer=" + peer); rec.Code != http.StatusNoContent { |
| + t.Fatalf("probe of a resumable peer: %d, want 204", rec.Code) |
| + } |
| + rec = c.get("/m/" + sqid + "/stream?probe=1&peer=0123456789abcdef0123456789abcdef") |
| + _, gotBoot, _ := answer(t, rec) |
| + if rec.Code != http.StatusNotFound || gotBoot != boot { |
| + t.Fatalf("probe of an unknown peer: %d boot %q", rec.Code, gotBoot) |
| + } |
| + // Probed twice and still resumable: the probe changed nothing. |
| + s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| + defer s2.close() |
| + if p, _ := s2.hello(); p != peer { |
| + t.Fatalf("resume after two probes: %s, want %s", p, peer) |
| + } |
| +} |
| + |
| +// Spec §2.4 and §4.5: ending the meeting closes every stream with |
| +// lost{ended}, and the peer id is remembered so a resume is told why. |
| +// A meeting nobody streamed has no headstone in the hub at all, so |
| +// ended_at is what refuses the stream (the Task 2 carry). |
| +func TestEndingTheMeetingClosesTheStreamAndRefusesWhatComesAfter(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, boot := s.hello() |
| + |
| + if rec := c.post("/meetings/"+sqid+"/end", nil); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("end meeting: %d", rec.Code) |
| + } |
| + ev, data, ok := s.next(2 * time.Second) |
| + if !ok || ev != "lost" { |
| + t.Fatalf("after the meeting ended: %s ok=%v", ev, ok) |
| + } |
| + var lost struct{ Reason string } |
| + json.Unmarshal(data, &lost) |
| + if lost.Reason != "ended" { |
| + t.Fatalf("lost %s, want reason ended", data) |
| + } |
| + if !s.waitClosed(2 * time.Second) { |
| + t.Fatal("the stream stayed open after the meeting ended") |
| + } |
| + rec := c.get("/m/" + sqid + "/stream?peer=" + peer) |
| + _, _, reason := answer(t, rec) |
| + if rec.Code != http.StatusGone || reason != "ended" { |
| + t.Fatalf("resume after the ending: %d %q, want 410 ended", rec.Code, reason) |
| + } |
| + if rec := c.get("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("fresh stream on an ended meeting: %d, want 404", rec.Code) |
| + } |
| + |
| + // A meeting nobody opened a stream on: the hub has never heard of |
| + // it, so only the row can refuse it. |
| + quiet, _ := createMeeting(t, c, "Never opened", "") |
| + if rec := c.post("/meetings/"+quiet+"/end", nil); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("end the quiet meeting: %d", rec.Code) |
| + } |
| + if rec := c.get("/m/" + quiet + "/stream"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("fresh stream on a meeting nobody opened: %d, want 404", rec.Code) |
| + } |
| + // A peer id the hub has never heard of, rather than the tombstoned |
| + // one above: a tombstone is not scoped to its room yet (the Task 3 |
| + // review's first Important), so the ended peer would be answered |
| + // 410 here, in a room it was never in. |
| + rec = c.get("/m/" + quiet + "/stream?peer=fedcba9876543210fedcba9876543210") |
| + if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot { |
| + t.Fatalf("resume into a meeting nobody opened: %d boot %q", rec.Code, gotBoot) |
| + } |
| + if rec := c.get("/m/zzzzzzzzzz/stream"); rec.Code != http.StatusNotFound { |
| + t.Fatalf("stream for a meeting that does not exist: %d", rec.Code) |
| + } |
| +} |