amadan

oficina / meet Public

stream: end on shutdown, require the EventSource Accept, tighter window

Review fix round 1. SIGTERM now cancels the context the app is built on (signal.NotifyContext in main; rastrillo's own handler still fires, since signal.Notify fans out) and a live stream watches it, so one open stream no longer holds the ten-second Shutdown and turns a clean stop into an exit 1. Ruling 15: the attaching GET requires an Accept of text/event-stream and is otherwise the usual 404 with boot, so a cross-site navigation cannot take the session's peer slot; the probe forms are exempt.

Minors: probe 204s carry Cache-Control no-store; the redundant Connection header is gone; ?last= is refused above 8 KiB before it is parsed; currentPerson's ok is checked; the session key binds the subject as well as the cookie; the stream timings have a floor so a zero cannot panic a ticker in a request; the hub refuses a resume past resume_until rather than waiting for the tick, so the window is thirty seconds and not thirty to forty-five; the SSE reader joins multi-line data, accepts fields without the space after the colon, and blocks with a bounded timeout instead of dropping an event; refusal tests use a two-second context-bounded GET.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
Paul Campbell pushed by paul@keymail.dev 096ee5553ad0070bb899adfe728ed479b2a09cd7 parent f2f5f3f
8 files changed, +344 −60
  • cmd/meet/main.go +14 −4
  • internal/meet/app.go +43 −10
  • internal/meet/hub/relay.go +10 −1
  • internal/meet/hub/relay_test.go +36 −0
  • internal/meet/stream.go +50 −6
  • internal/meettest/harness_test.go +18 −0
  • internal/meettest/sse_test.go +54 −24
  • internal/meettest/stream_test.go +119 −15

CI passed — run details

diff --git a/cmd/meet/main.go b/cmd/meet/main.go
index a2f1ef6..45fab21 100644
--- a/cmd/meet/main.go
+++ b/cmd/meet/main.go
@@ -5,7 +5,9 @@ import (
"fmt"
"log/slog"
"os"
+ "os/signal"
"strings"
+ "syscall"
"amadan.net/rastrillo/rastrillo"
"amadan.net/rastrillo/rastrillo/db"
@@ -60,10 +62,18 @@ func main() {
Logger: logger,
}
- // context.Background(): the hourly sweep this starts runs for the
- // life of the process, and nothing here ever needs to cancel it
- // early (review C2-3 — "acceptable" per the brief).
- mux, rt, err := meet.App(context.Background(), d, cfg)
+ // Cancelled on SIGTERM, which is how a live SSE stream learns the
+ // process is stopping: rastrillo.Serve handles the same signal with
+ // a ten-second Shutdown, and Shutdown waits for handlers without
+ // cancelling their request contexts, so a stream that watched
+ // nothing would hold that whole deadline and turn a clean stop into
+ // a failed one. signal.Notify fans out, so rastrillo's own handler
+ // still fires. The hourly sweep and the hub tick end here too, at
+ // the moment the process is going down anyway (review C2-3).
+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+ defer stop()
+
+ mux, rt, err := meet.App(ctx, d, cfg)
if err != nil {
logger.Error("build app", "err", err)
os.Exit(1)
diff --git a/internal/meet/app.go b/internal/meet/app.go
index d8129f3..9c2fc82 100644
--- a/internal/meet/app.go
+++ b/internal/meet/app.go
@@ -40,19 +40,43 @@ type app struct {
// (spec §4.4): thirty a second, burst sixty.
signalLimiter *rateLimiter
+ // done closes when the context App was given is cancelled — a
+ // SIGTERM, in the deployed shape. Live streams watch it and return,
+ // because http.Server.Shutdown waits for handlers without
+ // cancelling their contexts: one open stream would otherwise hold
+ // the whole shutdown deadline.
+ done <-chan struct{}
+
// 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()) }
+func (a *app) heartbeatEvery() time.Duration { return atLeast(a.heartbeat.Load()) }
+func (a *app) streamLife() time.Duration { return atLeast(a.ttl.Load()) }
+
+// atLeast keeps a timing above the floor: time.NewTicker panics on a
+// zero or negative duration, and that would be inside a request.
+func atLeast(ns int64) time.Duration {
+ if d := time.Duration(ns); d > minStreamTiming {
+ return d
+ }
+ return minStreamTiming
+}
// 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 }
+type TestHandle struct {
+ a *app
+ stop context.CancelFunc
+}
+
+// ShutdownForTest cancels the context the app was built on, which is
+// what a SIGTERM does to the deployed process: live streams see it and
+// return. It does not stop the handler serving requests.
+func (h *TestHandle) ShutdownForTest() { h.stop() }
// SetStreamTimings shortens the SSE heartbeat and TTL so a test can
// watch a stream expire without waiting five minutes. Set it before any
@@ -195,9 +219,10 @@ func (a *app) tickForever(ctx context.Context) {
// returns the *app alongside the mux so tests can reach into it (see
// NewForTest); production code uses App, which discards the *app.
//
-// ctx bounds the sweep at boot. The hourly ticker is not started here:
-// App starts it, NewForTest does not, so a test suite does not leak a
-// goroutine per app it builds.
+// ctx bounds the sweep at boot, and its cancellation is what tells live
+// streams the process is stopping (a.done). The hourly ticker is not
+// started here: App starts it, NewForTest does not, so a test suite does
+// not leak a goroutine per app it builds.
func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, error) {
rc, err := resolve(cfg)
if err != nil {
@@ -237,6 +262,7 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err
return nil, nil, err
}
a.hubSalt = hex.EncodeToString(salt)
+ a.done = ctx.Done()
a.heartbeat.Store(int64(defaultHeartbeat))
a.ttl.Store(int64(defaultStreamTTL))
// The hub reads standing back through the app, and never under its
@@ -347,8 +373,13 @@ func (a *app) nextDue() time.Time {
// returns a *http.ServeMux because rastrillo.Options.Mux is typed that
// way; the chi router mounts inside it, and a *Runtime carrying the
// platform wake function main.go wires to Options.NextDue. It also
-// starts the hourly sweep on ctx, which runs until ctx is cancelled or
-// the process exits.
+// starts the hourly sweep and the hub tick on ctx, which run until ctx
+// is cancelled or the process exits.
+//
+// Cancel ctx on SIGTERM: live SSE streams end when it is cancelled, and
+// http.Server.Shutdown waits for handlers without cancelling their
+// contexts, so a stream that does not watch this holds the shutdown
+// deadline open.
func App(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *Runtime, error) {
mux, a, err := build(ctx, d, cfg)
if err != nil {
@@ -363,9 +394,11 @@ func App(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *Runtime, e
// controls. No sweep ticker is started: see build. Tests only;
// production uses App.
func NewForTest(d *db.DB, cfg Config) (*http.ServeMux, *TestHandle, error) {
- mux, a, err := build(context.Background(), d, cfg)
+ ctx, cancel := context.WithCancel(context.Background())
+ mux, a, err := build(ctx, d, cfg)
if err != nil {
+ cancel()
return nil, nil, err
}
- return mux, &TestHandle{a: a}, nil
+ return mux, &TestHandle{a: a, stop: cancel}, nil
}
diff --git a/internal/meet/hub/relay.go b/internal/meet/hub/relay.go
index 63bc4d6..081115c 100644
--- a/internal/meet/hub/relay.go
+++ b/internal/meet/hub/relay.go
@@ -137,11 +137,20 @@ func (h *Hub) resumeRefusalLocked(sqid, session string, peer PeerID) *ResumeErro
if t, ok := h.tombstones[peer]; ok && t.room == sqid && t.session == session {
return &ResumeError{Status: 410, Reason: t.reason}
}
- if _, found := h.findLocked(sqid, session, peer); !found {
+ p, found := h.findLocked(sqid, session, peer)
+ if !found {
// Unknown here, unknown in another session, unknown in another
// room: one answer, so nothing leaks (spec §4.2).
return &ResumeError{Status: 404, Boot: h.boot}
}
+ // The window is thirty seconds, not thirty plus however long it is
+ // until the next tick: a resume that arrives after resume_until is
+ // refused now. Removing the participation and announcing the
+ // departure is still the tick's, so a refusal never has a room's
+ // events as a side effect.
+ if p.sub == nil && !p.resumeUntil.IsZero() && h.cfg.Now().After(p.resumeUntil) {
+ return &ResumeError{Status: 404, Boot: h.boot}
+ }
return nil
}
diff --git a/internal/meet/hub/relay_test.go b/internal/meet/hub/relay_test.go
index 050fbf9..80a79f1 100644
--- a/internal/meet/hub/relay_test.go
+++ b/internal/meet/hub/relay_test.go
@@ -326,3 +326,39 @@ func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) {
t.Fatalf("lapsed resume: %v", err)
}
}
+
+// Spec §4.5's window is thirty seconds, not thirty plus however long it
+// is until the next tick: a resume that arrives after resume_until is
+// refused on arrival. The participation is still the tick's to remove,
+// so a refusal never has a room's events as a side effect.
+func TestAResumePastItsWindowIsRefusedBeforeTheTickRemovesIt(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ h := testHub(&now)
+ a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
+ h.Detach("r1", a.Peer, a.Sub)
+
+ now = now.Add(ResumeWindow - time.Second)
+ if re := h.CanResume("r1", "sa", a.Peer); re != nil {
+ t.Fatalf("inside the window: %+v", re)
+ }
+ now = now.Add(2 * time.Second) // past resume_until, before any tick
+ re := h.CanResume("r1", "sa", a.Peer)
+ if re == nil || re.Status != 404 || re.Boot != h.Boot() {
+ t.Fatalf("past the window: %+v", re)
+ }
+ if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil {
+ t.Fatal("Resume attached a participation past its window")
+ }
+ // Refused, but still the tick's to remove and still this session's
+ // one peer in the room until then.
+ if _, ok := h.Find("r1", "sa", a.Peer); !ok {
+ t.Fatal("the refusal removed the participation")
+ }
+ if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != ErrSessionHasPeer {
+ t.Fatalf("a fresh stream before the tick: %v", err)
+ }
+ h.Tick(context.Background())
+ if _, ok := h.Find("r1", "sa", a.Peer); ok {
+ t.Fatal("the tick left the lapsed participation behind")
+ }
+}
diff --git a/internal/meet/stream.go b/internal/meet/stream.go
index 0566791..d00ca9d 100644
--- a/internal/meet/stream.go
+++ b/internal/meet/stream.go
@@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"net/http"
+ "strings"
"time"
"github.com/go-chi/chi/v5"
@@ -23,6 +24,12 @@ const (
defaultHeartbeat = 15 * time.Second
defaultStreamTTL = 5 * time.Minute
streamWriteWindow = 30 * time.Second
+ // minStreamTiming is the floor under both, so a zero or negative
+ // value from SetStreamTimings cannot panic a ticker in a request
+ // goroutine.
+ minStreamTiming = 10 * time.Millisecond
+ // maxLastQuery bounds the resume's ?last= before it is parsed.
+ maxLastQuery = 8 << 10
)
// sessionKey is the session's identity for the hub (spec §4.2): the
@@ -36,17 +43,30 @@ const (
// 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 == "" {
+ s, ok := sessions.Current(r)
+ if !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))
+ // The subject is bound in as well as the cookie: the key then
+ // cannot survive the person it belongs to, whatever a cookie value
+ // might come to be reused for.
+ sum := sha256.Sum256([]byte(a.hubSalt + "|" + s.Subject + "|" + c.Value))
return hex.EncodeToString(sum[:]), true
}
+// acceptsEventStream is ruling 15. The attaching stream is only ever
+// opened by EventSource, which always asks for text/event-stream; a
+// cross-site top-level navigation asks for HTML, and without this check
+// it would take the session's one peer slot in the room and hold it for
+// the resume window. The probe forms are exempt: they attach nothing.
+func acceptsEventStream(r *http.Request) bool {
+ return strings.Contains(strings.ToLower(r.Header.Get("Accept")), "text/event-stream")
+}
+
// 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
@@ -76,7 +96,12 @@ func (a *app) refuseResume(w http.ResponseWriter, re *hub.ResumeError) {
// 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)
+ person, ok := currentPerson(r)
+ if !ok {
+ a.logger.Error("meet: stream with no person in context", "path", r.URL.Path)
+ streamJSON(w, http.StatusInternalServerError, nil)
+ return
+ }
session, ok := a.sessionKey(r)
if !ok {
a.logger.Error("meet: stream with no session subject", "path", r.URL.Path)
@@ -84,6 +109,12 @@ func (a *app) stream(w http.ResponseWriter, r *http.Request) {
return
}
probe := r.URL.Query().Get("probe") == "1"
+ if !probe && !acceptsEventStream(r) {
+ // Not a stream client: the same 404 an unknown meeting gets, and
+ // no participation is created (ruling 15).
+ streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()})
+ return
+ }
var sub *hub.Sub
var peer hub.PeerID
@@ -116,11 +147,18 @@ func (a *app) stream(w http.ResponseWriter, r *http.Request) {
a.refuseResume(w, re)
return
}
- w.WriteHeader(http.StatusNoContent)
+ streamJSON(w, http.StatusNoContent, nil)
return
}
var last map[hub.PeerID]uint64
if raw := r.URL.Query().Get("last"); raw != "" {
+ // One seq per peer, sixteen peers to a room: a document this
+ // far past that is not a browser's, and is refused before it
+ // is parsed.
+ if len(raw) > maxLastQuery {
+ streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"})
+ return
+ }
if err := json.Unmarshal([]byte(raw), &last); err != nil {
streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"})
return
@@ -182,7 +220,7 @@ func (a *app) stream(w http.ResponseWriter, r *http.Request) {
streamJSON(w, http.StatusConflict, map[string]any{"peer": string(existing)})
return
}
- w.WriteHeader(http.StatusNoContent)
+ streamJSON(w, http.StatusNoContent, nil)
return
}
kind := hub.KindGuest
@@ -222,7 +260,6 @@ func (a *app) serveStream(w http.ResponseWriter, r *http.Request, sub *hub.Sub,
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")
@@ -243,6 +280,13 @@ func (a *app) serveStream(w http.ResponseWriter, r *http.Request, sub *hub.Sub,
select {
case <-r.Context().Done():
return
+ case <-a.done:
+ // The process is going down. Shutdown waits for handlers and
+ // does not cancel their contexts, so a live stream would
+ // hold the whole ten-second deadline and turn a clean stop
+ // into a failed one. The browser resumes, as it does at the
+ // TTL.
+ return
case <-ttl.C:
return
case <-heartbeat.C:
diff --git a/internal/meettest/harness_test.go b/internal/meettest/harness_test.go
index 020cbf6..e51ffd4 100644
--- a/internal/meettest/harness_test.go
+++ b/internal/meettest/harness_test.go
@@ -77,6 +77,11 @@ func (x *harness) advance(d time.Duration) {
// for the whole pass).
func (x *harness) sweep() { x.ctl.SweepForTest(x.now) }
+// shutdown is the SIGTERM the deployed process gets: the context the
+// app was built on is cancelled, and live streams end themselves so a
+// graceful shutdown is not held open by one.
+func (x *harness) shutdown() { x.ctl.ShutdownForTest() }
+
// authLinks counts rastrillo's unclicked magic links. It is the app's
// own table, not Meet's, which is the point: nothing deletes a row there
// but presenting the token or sweeping.
@@ -234,6 +239,19 @@ func (c *client) get(target string) *httptest.ResponseRecorder {
return c.do(httptest.NewRequest(http.MethodGet, target, nil))
}
+// getStream is get for the stream endpoint, for the answers that are
+// not a stream: it carries EventSource's Accept header (ruling 15) and
+// a two-second deadline, so a case that expects a refusal and gets a
+// stream fails in two seconds instead of holding the recorder open for
+// the whole TTL. A recorder cannot read a stream anyway.
+func (c *client) getStream(target string) *httptest.ResponseRecorder {
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ req := httptest.NewRequest(http.MethodGet, target, nil).WithContext(ctx)
+ req.Header.Set("Accept", "text/event-stream")
+ return c.do(req)
+}
+
func (c *client) post(target string, form url.Values) *httptest.ResponseRecorder {
req := httptest.NewRequest(http.MethodPost, target, strings.NewReader(form.Encode()))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
diff --git a/internal/meettest/sse_test.go b/internal/meettest/sse_test.go
index b9188ee..629d3e6 100644
--- a/internal/meettest/sse_test.go
+++ b/internal/meettest/sse_test.go
@@ -8,6 +8,7 @@ import (
"io"
"net/http"
"strings"
+ "sync"
"testing"
"time"
)
@@ -16,12 +17,14 @@ import (
// 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{}
+ t *testing.T
+ cancel context.CancelFunc
+ body io.ReadCloser
+ events chan sseEvent
+ hb chan struct{}
+ done chan struct{}
+ closing chan struct{}
+ once sync.Once
}
type sseEvent struct {
@@ -63,42 +66,66 @@ func (c *client) stream(target string) *sseReader {
}
s := &sseReader{
t: c.t, cancel: cancel, body: resp.Body,
- events: make(chan sseEvent, 64), hb: make(chan struct{}, 64), done: make(chan struct{}),
+ events: make(chan sseEvent, 64), hb: make(chan struct{}, 64),
+ done: make(chan struct{}), closing: make(chan struct{}),
}
c.t.Cleanup(s.close)
go s.read()
return s
}
+// read parses the wire format the way an EventSource does: a field is
+// "name:" with one optional space after the colon, several data lines
+// join with a newline, a line starting with a colon is a comment, and a
+// blank line dispatches what has been collected.
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
+ var hasData bool
for sc.Scan() {
line := sc.Text()
switch {
case line == "":
- if ev.name != "" {
- select {
- case s.events <- ev:
- default:
- }
+ if ev.name != "" || hasData {
+ s.dispatch(ev)
}
- ev = sseEvent{}
+ ev, hasData = sseEvent{}, false
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: "))
+ case strings.HasPrefix(line, "event:"):
+ ev.name = field(line, "event:")
+ case strings.HasPrefix(line, "data:"):
+ if hasData {
+ ev.data = append(ev.data, '\n')
+ }
+ ev.data = append(ev.data, field(line, "data:")...)
+ hasData = true
}
}
}
+func field(line, name string) string {
+ return strings.TrimPrefix(strings.TrimPrefix(line, name), " ")
+}
+
+// dispatch hands one event to the test goroutine. It blocks rather than
+// dropping: an overflowing buffer means a test is not reading what it
+// asked for, and that should fail out loud rather than quietly lose the
+// event that would have failed the assertion.
+func (s *sseReader) dispatch(ev sseEvent) {
+ select {
+ case s.events <- ev:
+ case <-s.closing:
+ case <-time.After(5 * time.Second):
+ s.t.Errorf("sse: nothing read %q for five seconds; the event buffer overflowed", ev.name)
+ }
+}
+
// next waits for one event.
func (s *sseReader) next(timeout time.Duration) (string, []byte, bool) {
select {
@@ -148,13 +175,16 @@ func (s *sseReader) waitClosed(timeout time.Duration) bool {
// 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")
- }
+ s.once.Do(func() {
+ close(s.closing)
+ 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
diff --git a/internal/meettest/stream_test.go b/internal/meettest/stream_test.go
index 2c02c24..0320afa 100644
--- a/internal/meettest/stream_test.go
+++ b/internal/meettest/stream_test.go
@@ -4,6 +4,8 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
+ "net/url"
+ "strings"
"testing"
"time"
)
@@ -36,7 +38,7 @@ func TestStreamSaysHelloWithPeerAndBootAndRefusesTheSecondFreshStream(t *testing
if len(peer) != 32 || len(boot) != 32 {
t.Fatalf("hello peer=%q boot=%q", peer, boot)
}
- rec := c.get("/m/" + sqid + "/stream")
+ rec := c.getStream("/m/" + sqid + "/stream")
if rec.Code != http.StatusConflict {
t.Fatalf("second fresh stream: %d, want 409", rec.Code)
}
@@ -73,10 +75,10 @@ func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) {
guest := x.browser(t)
guest.get("/j/" + tok)
guest.signIn("guest@example.com")
- if rec := guest.get("/m/" + elsewhere + "/stream"); rec.Code != http.StatusNotFound {
+ if rec := guest.getStream("/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)
+ rec := guest.getStream("/m/" + sqid + "/stream?peer=" + peer)
if rec.Code != http.StatusNotFound {
t.Fatalf("resume through another session: %d, want 404", rec.Code)
}
@@ -84,7 +86,7 @@ func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) {
if _, gotBoot, _ := answer(t, rec); gotBoot != boot {
t.Fatalf("404 boot %q, want %q", gotBoot, boot)
}
- unknown := guest.get("/m/" + sqid + "/stream?peer=0123456789abcdef0123456789abcdef")
+ unknown := guest.getStream("/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())
}
@@ -118,7 +120,7 @@ func TestResumeReattachesTheSameParticipation(t *testing.T) {
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 {
+ if rec := c.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusConflict {
t.Fatalf("fresh stream while resumed: %d, want 409", rec.Code)
}
}
@@ -167,7 +169,7 @@ func TestLapsedResumeIs404WithTheSameBootAndAFreshStreamGetsANewPeer(t *testing.
x.advance(31 * time.Second)
x.ctl.TickForTest()
- rec := c.get("/m/" + sqid + "/stream?peer=" + peer)
+ rec := c.getStream("/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)
@@ -188,8 +190,14 @@ func TestResumeProbeAnswersWithoutAttaching(t *testing.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)
+ // No Accept header: a probe attaches nothing, so ruling 15 exempts
+ // it. Its answer is still never cached.
+ rec0 := c.get("/m/" + sqid + "/stream?probe=1")
+ if rec0.Code != http.StatusNoContent {
+ t.Fatalf("probe with nothing open: %d, want 204", rec0.Code)
+ }
+ if cc := rec0.Header().Get("Cache-Control"); cc != "no-store" {
+ t.Fatalf("probe Cache-Control %q, want no-store", cc)
}
s := c.stream("/m/" + sqid + "/stream")
peer, boot := s.hello()
@@ -200,8 +208,12 @@ func TestResumeProbeAnswersWithoutAttaching(t *testing.T) {
}
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)
+ probed := c.get("/m/" + sqid + "/stream?probe=1&peer=" + peer)
+ if probed.Code != http.StatusNoContent {
+ t.Fatalf("probe of a resumable peer: %d, want 204", probed.Code)
+ }
+ if cc := probed.Header().Get("Cache-Control"); cc != "no-store" {
+ t.Fatalf("resume probe Cache-Control %q, want no-store", cc)
}
rec = c.get("/m/" + sqid + "/stream?probe=1&peer=0123456789abcdef0123456789abcdef")
_, gotBoot, _ := answer(t, rec)
@@ -243,12 +255,12 @@ func TestEndingTheMeetingClosesTheStreamAndRefusesWhatComesAfter(t *testing.T) {
if !s.waitClosed(2 * time.Second) {
t.Fatal("the stream stayed open after the meeting ended")
}
- rec := c.get("/m/" + sqid + "/stream?peer=" + peer)
+ rec := c.getStream("/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 {
+ if rec := c.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound {
t.Fatalf("fresh stream on an ended meeting: %d, want 404", rec.Code)
}
@@ -258,17 +270,109 @@ func TestEndingTheMeetingClosesTheStreamAndRefusesWhatComesAfter(t *testing.T) {
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 {
+ if rec := c.getStream("/m/" + quiet + "/stream"); rec.Code != http.StatusNotFound {
t.Fatalf("fresh stream on a meeting nobody opened: %d, want 404", rec.Code)
}
// The peer the ending tombstoned above, offered to a room it was
// never in: a tombstone answers only through its own room and
// session, so this is the plain 404 an unknown peer gets.
- rec = c.get("/m/" + quiet + "/stream?peer=" + peer)
+ rec = c.getStream("/m/" + quiet + "/stream?peer=" + peer)
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 {
+ if rec := c.getStream("/m/zzzzzzzzzz/stream"); rec.Code != http.StatusNotFound {
t.Fatalf("stream for a meeting that does not exist: %d", rec.Code)
}
}
+
+// Ruling 15: the attaching stream is opened by EventSource, which always
+// asks for text/event-stream. A cross-site top-level navigation asks for
+// HTML, and it must not be able to take the session's one peer slot in
+// the room and hold it for the resume window.
+func TestAStreamWithoutTheEventStreamAcceptTakesNoPeer(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+
+ rec := c.get("/m/" + sqid + "/stream") // a navigation, not EventSource
+ if rec.Code != http.StatusNotFound {
+ t.Fatalf("stream without the Accept header: %d, want 404", rec.Code)
+ }
+ // The 404 an unknown meeting gets, and nothing was allocated: the
+ // real stream that follows is a fresh peer, not a 409.
+ if _, boot, _ := answer(t, rec); boot == "" {
+ t.Fatalf("404 body %q, want the boot id", rec.Body.String())
+ }
+ if rec := c.get("/m/" + sqid + "/stream?peer=whatever"); rec.Code != http.StatusNotFound {
+ t.Fatalf("resume without the Accept header: %d, want 404", rec.Code)
+ }
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ if peer, _ := s.hello(); peer == "" {
+ t.Fatal("the stream that does ask for text/event-stream got no peer")
+ }
+}
+
+// A resume's ?last= is refused above eight kibibytes, before it is
+// parsed: one seq per peer and sixteen peers to a room, so nothing this
+// size is a browser's.
+func TestAnOversizeLastIsRefusedBeforeItIsParsed(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ s := c.stream("/m/" + sqid + "/stream")
+ peer, _ := s.hello()
+ s.close()
+
+ big := strings.Repeat("a", 9<<10)
+ rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer + "&last=" + big)
+ if rec.Code != http.StatusBadRequest {
+ t.Fatalf("oversize last: %d, want 400", rec.Code)
+ }
+ // Still resumable: a refused query is not a refused participation.
+ s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer + "&last=" + url.QueryEscape(`{"`+peer+`":3}`))
+ defer s2.close()
+ if p, _ := s2.hello(); p != peer {
+ t.Fatalf("resume after the refused last: %s, want %s", p, peer)
+ }
+}
+
+// A SIGTERM cancels the context the app was built on. rastrillo answers
+// the same signal with a ten-second Shutdown, and Shutdown waits for
+// handlers without cancelling their contexts, so a live stream that
+// watched nothing would hold that whole deadline and the process would
+// exit 1 instead of cleanly. The browser resumes, as at the TTL.
+func TestShutdownEndsLiveStreams(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()
+ s.hello()
+
+ x.shutdown()
+ if !s.waitClosed(2 * time.Second) {
+ t.Fatal("a live stream outlived the app's context")
+ }
+}
+
+// SetStreamTimings is a test knob, and a zero from it must not panic a
+// ticker inside a request goroutine: both timings have a floor.
+func TestZeroStreamTimingsAreClampedNotFatal(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ x.ctl.SetStreamTimings(0, 0)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ if peer, _ := s.hello(); peer == "" {
+ t.Fatal("no peer")
+ }
+ if !s.waitClosed(2 * time.Second) {
+ t.Fatal("a floored TTL did not end the stream")
+ }
+ // The app is still serving: nothing panicked.
+ if rec := c.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK {
+ t.Fatalf("holding screen after the floored stream: %d", rec.Code)
+ }
+}