| 1 | package docs |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "bytes" |
| 6 | "context" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | // These tests run against a real httptest.NewServer rather than the |
| 15 | // httptest.NewRecorder every other test in this package uses, and the |
| 16 | // reason is structural rather than stylistic: a recorder collects |
| 17 | // what a handler wrote and hands it over once the handler RETURNS. A |
| 18 | // stream handler does not return until the connection is over, so a |
| 19 | // recorder can express "what did this stream eventually say" and can |
| 20 | // never express "what does this stream say while it is open", which |
| 21 | // is the only interesting question about it. A real server and an |
| 22 | // incrementally-read body can. |
| 23 | |
| 24 | // streamDurations shortens the handler's own timings for a test. The |
| 25 | // production values (five minutes, fifteen seconds) are correct for a |
| 26 | // browser and useless for a test suite. |
| 27 | func streamDurations(t *testing.T, app *App, heartbeat, ttl time.Duration) { |
| 28 | t.Helper() |
| 29 | h := app.testHandlers() |
| 30 | h.streamHeartbeat = heartbeat |
| 31 | h.streamTTL = ttl |
| 32 | } |
| 33 | |
| 34 | // openStream starts a GET on path as address and returns the response |
| 35 | // plus a reader positioned at its body, having already checked the |
| 36 | // status. The caller closes the body. |
| 37 | func openStream(t *testing.T, srv *httptest.Server, app *App, address, path string) (*http.Response, *bufio.Reader) { |
| 38 | t.Helper() |
| 39 | |
| 40 | req := app.TestRequest(t, address, http.MethodGet, path, nil) |
| 41 | live, err := http.NewRequest(http.MethodGet, srv.URL+path, nil) |
| 42 | if err != nil { |
| 43 | t.Fatalf("build stream request: %v", err) |
| 44 | } |
| 45 | for _, c := range req.Cookies() { |
| 46 | live.AddCookie(c) |
| 47 | } |
| 48 | |
| 49 | res, err := srv.Client().Do(live) |
| 50 | if err != nil { |
| 51 | t.Fatalf("open stream %s: %v", path, err) |
| 52 | } |
| 53 | return res, bufio.NewReader(res.Body) |
| 54 | } |
| 55 | |
| 56 | // nextEvent reads SSE lines until it has one complete event, and |
| 57 | // returns its name and data. A comment heartbeat (": ping") is |
| 58 | // returned as the name "ping" so a test can assert on it. |
| 59 | func nextEvent(t *testing.T, r *bufio.Reader, within time.Duration) (name, data string) { |
| 60 | t.Helper() |
| 61 | |
| 62 | type parsed struct{ name, data string } |
| 63 | got := make(chan parsed, 1) |
| 64 | errs := make(chan error, 1) |
| 65 | |
| 66 | go func() { |
| 67 | var ev, payload string |
| 68 | for { |
| 69 | line, err := r.ReadString('\n') |
| 70 | if err != nil { |
| 71 | errs <- err |
| 72 | return |
| 73 | } |
| 74 | line = strings.TrimRight(line, "\r\n") |
| 75 | switch { |
| 76 | case strings.HasPrefix(line, ": "): |
| 77 | got <- parsed{"ping", strings.TrimPrefix(line, ": ")} |
| 78 | return |
| 79 | case strings.HasPrefix(line, "event: "): |
| 80 | ev = strings.TrimPrefix(line, "event: ") |
| 81 | case strings.HasPrefix(line, "data: "): |
| 82 | payload = strings.TrimPrefix(line, "data: ") |
| 83 | case line == "": |
| 84 | if ev != "" { |
| 85 | got <- parsed{ev, payload} |
| 86 | return |
| 87 | } |
| 88 | } |
| 89 | } |
| 90 | }() |
| 91 | |
| 92 | select { |
| 93 | case p := <-got: |
| 94 | return p.name, p.data |
| 95 | case err := <-errs: |
| 96 | t.Fatalf("reading the stream: %v", err) |
| 97 | case <-time.After(within): |
| 98 | t.Fatalf("no event within %s", within) |
| 99 | } |
| 100 | return "", "" |
| 101 | } |
| 102 | |
| 103 | // The stream is owner-scoped exactly as the update routes are, and a |
| 104 | // stranger must get 404 rather than 403 — a 403 would confirm the |
| 105 | // document exists. |
| 106 | func TestStreamIsOwnerScoped(t *testing.T) { |
| 107 | app, err := NewForTest(t) |
| 108 | if err != nil { |
| 109 | t.Fatalf("NewForTest: %v", err) |
| 110 | } |
| 111 | ctx := context.Background() |
| 112 | owner := app.TestPerson(t, "owner@example.com") |
| 113 | app.TestPerson(t, "stranger@example.com") |
| 114 | doc, err := app.Store().Create(ctx, owner, "Private") |
| 115 | if err != nil { |
| 116 | t.Fatalf("Create: %v", err) |
| 117 | } |
| 118 | |
| 119 | rec := serve(app, app.TestRequest(t, "stranger@example.com", http.MethodGet, |
| 120 | "/d/"+doc.Sqid+"/stream", nil)) |
| 121 | if rec.Code != http.StatusNotFound { |
| 122 | t.Errorf("stranger GET /stream = %d, want 404", rec.Code) |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | // The connect-time event is load-bearing rather than an optimisation: |
| 127 | // it is what heals a tab whose cursor legitimately failed to advance |
| 128 | // (design D3), and what covers the gap between the page rendering and |
| 129 | // the stream opening. Without it, a tab that missed a notification |
| 130 | // during that gap waits for the next edit before it learns anything. |
| 131 | func TestStreamAnnouncesTheCurrentSeqOnConnect(t *testing.T) { |
| 132 | app, err := NewForTest(t) |
| 133 | if err != nil { |
| 134 | t.Fatalf("NewForTest: %v", err) |
| 135 | } |
| 136 | ctx := context.Background() |
| 137 | owner := app.TestPerson(t, "owner@example.com") |
| 138 | doc, err := app.Store().Create(ctx, owner, "Connect") |
| 139 | if err != nil { |
| 140 | t.Fatalf("Create: %v", err) |
| 141 | } |
| 142 | if _, err := app.Store().ApplyUpdate(ctx, doc, app.TestActorFor(t, doc), validUpdate(t)); err != nil { |
| 143 | t.Fatalf("ApplyUpdate: %v", err) |
| 144 | } |
| 145 | |
| 146 | srv := httptest.NewServer(app.Handler()) |
| 147 | defer srv.Close() |
| 148 | |
| 149 | res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream") |
| 150 | defer res.Body.Close() |
| 151 | |
| 152 | if res.StatusCode != http.StatusOK { |
| 153 | t.Fatalf("stream = %d, want 200", res.StatusCode) |
| 154 | } |
| 155 | if ct := res.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") { |
| 156 | t.Errorf("Content-Type = %q, want text/event-stream", ct) |
| 157 | } |
| 158 | if cc := res.Header.Get("Cache-Control"); cc != "no-store" { |
| 159 | t.Errorf("Cache-Control = %q, want no-store", cc) |
| 160 | } |
| 161 | |
| 162 | name, data := nextEvent(t, body, 2*time.Second) |
| 163 | if name != "update" || data != "1" { |
| 164 | t.Errorf("connect event = (%q, %q), want (update, 1)", name, data) |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | // The feature itself: an update accepted from one request reaches a |
| 169 | // stream opened by another. |
| 170 | func TestStreamDeliversAnUpdateFromAnotherRequest(t *testing.T) { |
| 171 | app, err := NewForTest(t) |
| 172 | if err != nil { |
| 173 | t.Fatalf("NewForTest: %v", err) |
| 174 | } |
| 175 | ctx := context.Background() |
| 176 | owner := app.TestPerson(t, "owner@example.com") |
| 177 | doc, err := app.Store().Create(ctx, owner, "Two tabs") |
| 178 | if err != nil { |
| 179 | t.Fatalf("Create: %v", err) |
| 180 | } |
| 181 | |
| 182 | srv := httptest.NewServer(app.Handler()) |
| 183 | defer srv.Close() |
| 184 | |
| 185 | res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream") |
| 186 | defer res.Body.Close() |
| 187 | |
| 188 | // Drain the connect-time event (seq 0 — nothing written yet). |
| 189 | if name, data := nextEvent(t, body, 2*time.Second); name != "update" || data != "0" { |
| 190 | t.Fatalf("connect event = (%q, %q), want (update, 0)", name, data) |
| 191 | } |
| 192 | |
| 193 | post := app.TestRequest(t, "owner@example.com", http.MethodPost, |
| 194 | "/d/"+doc.Sqid+"/updates", bytes.NewReader(validUpdate(t))) |
| 195 | if rec := serve(app, post); rec.Code != http.StatusNoContent { |
| 196 | t.Fatalf("POST = %d, want 204", rec.Code) |
| 197 | } |
| 198 | |
| 199 | name, data := nextEvent(t, body, 2*time.Second) |
| 200 | if name != "update" || data != "1" { |
| 201 | t.Errorf("event after a POST = (%q, %q), want (update, 1)", name, data) |
| 202 | } |
| 203 | } |
| 204 | |
| 205 | // A quiet stream still writes, which is what makes the write deadline |
| 206 | // able to fire and a dead client discoverable. Without the heartbeat |
| 207 | // a wedged connection is only noticed at the next real edit, which |
| 208 | // may be never. |
| 209 | func TestStreamHeartbeatsWhenQuiet(t *testing.T) { |
| 210 | app, err := NewForTest(t) |
| 211 | if err != nil { |
| 212 | t.Fatalf("NewForTest: %v", err) |
| 213 | } |
| 214 | ctx := context.Background() |
| 215 | owner := app.TestPerson(t, "owner@example.com") |
| 216 | doc, err := app.Store().Create(ctx, owner, "Quiet") |
| 217 | if err != nil { |
| 218 | t.Fatalf("Create: %v", err) |
| 219 | } |
| 220 | streamDurations(t, app, 50*time.Millisecond, 5*time.Second) |
| 221 | |
| 222 | srv := httptest.NewServer(app.Handler()) |
| 223 | defer srv.Close() |
| 224 | |
| 225 | res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream") |
| 226 | defer res.Body.Close() |
| 227 | |
| 228 | nextEvent(t, body, 2*time.Second) // the connect event |
| 229 | |
| 230 | if name, _ := nextEvent(t, body, 2*time.Second); name != "ping" { |
| 231 | t.Errorf("quiet stream produced %q, want a ping heartbeat", name) |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | // No connection outlives the TTL. This is half of the hibernation |
| 236 | // story (design §6): the server's hard bound on how long one stream |
| 237 | // may hold a goroutine and keep an instance from sleeping. |
| 238 | func TestStreamEndsAtItsTTL(t *testing.T) { |
| 239 | app, err := NewForTest(t) |
| 240 | if err != nil { |
| 241 | t.Fatalf("NewForTest: %v", err) |
| 242 | } |
| 243 | ctx := context.Background() |
| 244 | owner := app.TestPerson(t, "owner@example.com") |
| 245 | doc, err := app.Store().Create(ctx, owner, "Bounded") |
| 246 | if err != nil { |
| 247 | t.Fatalf("Create: %v", err) |
| 248 | } |
| 249 | streamDurations(t, app, time.Second, 150*time.Millisecond) |
| 250 | |
| 251 | srv := httptest.NewServer(app.Handler()) |
| 252 | defer srv.Close() |
| 253 | |
| 254 | res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream") |
| 255 | defer res.Body.Close() |
| 256 | |
| 257 | nextEvent(t, body, 2*time.Second) // the connect event |
| 258 | |
| 259 | done := make(chan error, 1) |
| 260 | go func() { |
| 261 | _, err := body.ReadString('\n') |
| 262 | done <- err |
| 263 | }() |
| 264 | |
| 265 | select { |
| 266 | case err := <-done: |
| 267 | if err == nil { |
| 268 | t.Error("stream produced more output after its TTL instead of ending") |
| 269 | } |
| 270 | case <-time.After(2 * time.Second): |
| 271 | t.Error("stream was still open well past its TTL") |
| 272 | } |
| 273 | } |
| 274 | |
| 275 | // A client that goes away must not leave the handler running. This is |
| 276 | // what keeps a graceful drain's budget safe with streams open. |
| 277 | func TestStreamEndsWhenTheClientGoesAway(t *testing.T) { |
| 278 | app, err := NewForTest(t) |
| 279 | if err != nil { |
| 280 | t.Fatalf("NewForTest: %v", err) |
| 281 | } |
| 282 | ctx := context.Background() |
| 283 | owner := app.TestPerson(t, "owner@example.com") |
| 284 | doc, err := app.Store().Create(ctx, owner, "Abandoned") |
| 285 | if err != nil { |
| 286 | t.Fatalf("Create: %v", err) |
| 287 | } |
| 288 | |
| 289 | srv := httptest.NewServer(app.Handler()) |
| 290 | defer srv.Close() |
| 291 | |
| 292 | res, body := openStream(t, srv, app, "owner@example.com", "/d/"+doc.Sqid+"/stream") |
| 293 | nextEvent(t, body, 2*time.Second) |
| 294 | |
| 295 | // Closing the body is the client going away. |
| 296 | res.Body.Close() |
| 297 | |
| 298 | // The handler's subscriber must be gone. Polling the hub is the |
| 299 | // observable consequence; the handler itself has no return value |
| 300 | // to wait on. |
| 301 | deadline := time.Now().Add(2 * time.Second) |
| 302 | for time.Now().Before(deadline) { |
| 303 | if app.testHandlers().hub.documentCount() == 0 { |
| 304 | return |
| 305 | } |
| 306 | time.Sleep(10 * time.Millisecond) |
| 307 | } |
| 308 | t.Error("a subscriber outlived the client that opened it") |
| 309 | } |
| 310 | |
amadan