| 1 | package docs |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "io" |
| 6 | "net/http" |
| 7 | "strconv" |
| 8 | "time" |
| 9 | |
| 10 | "amadan.net/oficina/home/roster" |
| 11 | "amadan.net/rastrillo/rastrillo" |
| 12 | ) |
| 13 | |
| 14 | // The stream's timings. Lifted from rastrillo's own SSE handler |
| 15 | // (jobs/events.go), and each one is carried for its own reason rather |
| 16 | // than by imitation — see the notes on each. |
| 17 | const ( |
| 18 | // defaultStreamWriteDeadline is the per-write idle deadline, |
| 19 | // re-armed before every event and every heartbeat through |
| 20 | // http.NewResponseController. This is what lets a stream coexist |
| 21 | // with rastrillo serve's zero WriteTimeout default: a total |
| 22 | // response deadline would cut a healthy long-lived stream off |
| 23 | // alongside the stalled ones, so the framework leaves WriteTimeout |
| 24 | // off and expects a streaming handler to bound its own writes. |
| 25 | defaultStreamWriteDeadline = 30 * time.Second |
| 26 | |
| 27 | // defaultStreamHeartbeat is the comment-ping cadence on a quiet |
| 28 | // stream. It keeps intermediaries from timing out a connection |
| 29 | // that looks idle, and — more importantly here — it is what makes |
| 30 | // the write deadline above ever fire: without a periodic write, a |
| 31 | // client that has gone away without closing is discovered at the |
| 32 | // next real edit, which may be never. |
| 33 | defaultStreamHeartbeat = 15 * time.Second |
| 34 | |
| 35 | // defaultStreamTTL bounds any one connection. The client reopens |
| 36 | // (see fixtures/live-updates.mjs), so this costs a browser nothing |
| 37 | // and is half of the hibernation story the design spec's §6 |
| 38 | // describes: platform hibernation is default-on, and a document |
| 39 | // left open on Friday must not pin its instance awake all weekend. |
| 40 | // No timer here exists outside a live connection, so nothing in |
| 41 | // this file is a ticker that dies at hibernation. |
| 42 | defaultStreamTTL = 5 * time.Minute |
| 43 | ) |
| 44 | |
| 45 | // stream serves GET /d/{sqid}/stream: a Server-Sent Events feed that |
| 46 | // says "this document has content through seq N" and nothing else. |
| 47 | // The bytes themselves are fetched by the client through |
| 48 | // GET /d/{sqid}/updates?since=N — the same binary endpoint it already |
| 49 | // used at mount, through the same replay path. |
| 50 | // |
| 51 | // Keeping document bytes off this stream is deliberate, and mirrors |
| 52 | // the reasoning in rastrillo's own SSE handler for keeping HTML off |
| 53 | // its one ("SSE never carries HTML, so the one rendering path |
| 54 | // stays"). Two consequences here: there is exactly one code path that |
| 55 | // turns stored updates into a browser's Y.Doc, rather than two that |
| 56 | // can drift; and SSE's text-only framing costs nothing, where putting |
| 57 | // opaque binary Yjs updates on it would have meant base64 and about a |
| 58 | // third more bytes on every keystroke batch. |
| 59 | // |
| 60 | // Reader-scoped through getDocumentForUpdates, so a stranger gets the |
| 61 | // same 404 the update routes give — never a 403, which would confirm |
| 62 | // the document exists. |
| 63 | func (h *handlers) stream(w http.ResponseWriter, r *http.Request) { |
| 64 | doc, ok := h.getDocumentForUpdates(w, r) |
| 65 | if !ok { |
| 66 | return |
| 67 | } |
| 68 | |
| 69 | fl, ok := w.(http.Flusher) |
| 70 | if !ok { |
| 71 | h.logger.Error("docs: stream: ResponseWriter does not flush", "sqid", doc.Sqid) |
| 72 | http.Error(w, rastrillo.T(r, "docs.error.generic"), h.internalFailureStatus()) |
| 73 | return |
| 74 | } |
| 75 | |
| 76 | // Subscribe BEFORE reading the current seq, never after. An update |
| 77 | // committing between the two must produce a redundant |
| 78 | // notification, which costs one empty fetch; the other order would |
| 79 | // let it fall into the gap and produce nothing at all, and the tab |
| 80 | // would sit there current-looking and stale until the next edit. |
| 81 | ch, stop := h.hub.subscribe(doc.ID) |
| 82 | defer stop() |
| 83 | |
| 84 | var guard *teamStreamGuard |
| 85 | var leaseTimer *time.Timer |
| 86 | var leaseDeadline <-chan time.Time |
| 87 | if h.store.team != nil { |
| 88 | a, ok := r.Context().Value(accessContextKey{}).(*requestAccess) |
| 89 | if !ok || a == nil || a.runtime != h.store.team { |
| 90 | http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable) |
| 91 | return |
| 92 | } |
| 93 | guard = &teamStreamGuard{store: h.store, access: a, request: r, docID: doc.ID} |
| 94 | if a.decision.State == roster.Member { |
| 95 | guard.deadline = time.Now().Add(a.decision.Lease.Remaining()) |
| 96 | } |
| 97 | if _, ok := guard.check(); !ok { |
| 98 | http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable) |
| 99 | return |
| 100 | } |
| 101 | if a.decision.State == roster.Member { |
| 102 | remaining := guard.remaining() |
| 103 | if remaining <= 0 { |
| 104 | http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable) |
| 105 | return |
| 106 | } |
| 107 | leaseTimer = time.NewTimer(remaining) |
| 108 | defer leaseTimer.Stop() |
| 109 | leaseDeadline = leaseTimer.C |
| 110 | } |
| 111 | } |
| 112 | seqContext := r.Context() |
| 113 | if guard != nil { |
| 114 | var cancel context.CancelFunc |
| 115 | seqContext, cancel = context.WithTimeout(seqContext, min(guard.remaining(), defaultStreamWriteDeadline)) |
| 116 | defer cancel() |
| 117 | } |
| 118 | current, err := h.store.currentSeq(seqContext, doc.ID) |
| 119 | if err != nil { |
| 120 | h.logger.Error("docs: stream: read current seq", "sqid", doc.Sqid, "err", err) |
| 121 | http.Error(w, rastrillo.T(r, "docs.error.generic"), h.internalFailureStatus()) |
| 122 | return |
| 123 | } |
| 124 | |
| 125 | rc := http.NewResponseController(w) |
| 126 | if guard != nil { |
| 127 | limit, ok := guard.check() |
| 128 | if !ok || rc.SetWriteDeadline(time.Now().Add(limit)) != nil { |
| 129 | http.Error(w, rastrillo.T(r, "docs.error.generic"), http.StatusServiceUnavailable) |
| 130 | return |
| 131 | } |
| 132 | } |
| 133 | w.Header().Set("Content-Type", "text/event-stream") |
| 134 | w.Header().Set("Cache-Control", "no-store") |
| 135 | w.WriteHeader(http.StatusOK) |
| 136 | fl.Flush() |
| 137 | |
| 138 | // write is every byte's one exit: deadline armed, then the bytes, |
| 139 | // then the flush. The deadline error is ignored deliberately — a |
| 140 | // ResponseWriter that cannot set deadlines still streams, just |
| 141 | // unbounded — while a write error ends the handler, because it |
| 142 | // means the client is gone. |
| 143 | write := func(text string) bool { |
| 144 | // Team streams already revalidate through their original bounded lease. |
| 145 | if guard == nil && !h.nativeStreamAllowed(r, doc.Sqid) { |
| 146 | return false |
| 147 | } |
| 148 | limit := defaultStreamWriteDeadline |
| 149 | if guard != nil { |
| 150 | var ok bool |
| 151 | limit, ok = guard.check() |
| 152 | if !ok { |
| 153 | return false |
| 154 | } |
| 155 | } |
| 156 | if err := rc.SetWriteDeadline(time.Now().Add(limit)); err != nil && guard != nil { |
| 157 | return false |
| 158 | } |
| 159 | if _, err := io.WriteString(w, text); err != nil { |
| 160 | return false |
| 161 | } |
| 162 | fl.Flush() |
| 163 | return true |
| 164 | } |
| 165 | event := func(seq int64) bool { |
| 166 | return write("event: update\ndata: " + strconv.FormatInt(seq, 10) + "\n\n") |
| 167 | } |
| 168 | |
| 169 | // The connect-time event is load-bearing, not an optimisation. It |
| 170 | // is what heals a tab whose cursor legitimately failed to advance |
| 171 | // (a POST whose seq was not cursor+1, design D3), and it covers |
| 172 | // the gap between the page rendering server-side and this stream |
| 173 | // opening. Sending the current seq unconditionally costs a client |
| 174 | // that is already current exactly one empty ?since= fetch. |
| 175 | if !event(current) { |
| 176 | return |
| 177 | } |
| 178 | |
| 179 | heartbeat := time.NewTimer(h.heartbeatInterval()) |
| 180 | defer heartbeat.Stop() |
| 181 | ttl := time.NewTimer(h.ttlInterval()) |
| 182 | defer ttl.Stop() |
| 183 | |
| 184 | for { |
| 185 | select { |
| 186 | case <-leaseDeadline: |
| 187 | return |
| 188 | case <-r.Context().Done(): |
| 189 | // The client went away, or the server is draining. Either |
| 190 | // way this handler must return so a graceful shutdown's |
| 191 | // budget is not held by an open stream. |
| 192 | return |
| 193 | case <-ttl.C: |
| 194 | // A clean close, not an error: the client reopens if it |
| 195 | // still wants one, and a parked tab does not. |
| 196 | return |
| 197 | case seq := <-ch: |
| 198 | if !event(seq) { |
| 199 | return |
| 200 | } |
| 201 | heartbeat.Reset(h.heartbeatInterval()) |
| 202 | case <-heartbeat.C: |
| 203 | if !write(": ping\n\n") { |
| 204 | return |
| 205 | } |
| 206 | heartbeat.Reset(h.heartbeatInterval()) |
| 207 | } |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | // heartbeatInterval and ttlInterval read the handler's own overrides, |
| 212 | // falling back to the constants above. The overrides exist for |
| 213 | // stream_test.go, whose whole subject is behaviour over time: the |
| 214 | // production values are right for a browser and useless for a test |
| 215 | // suite, and the alternative — a test that waits five minutes to |
| 216 | // prove a five-minute TTL — is not a test anyone runs. |
| 217 | func (h *handlers) heartbeatInterval() time.Duration { |
| 218 | if h.streamHeartbeat > 0 { |
| 219 | return h.streamHeartbeat |
| 220 | } |
| 221 | return defaultStreamHeartbeat |
| 222 | } |
| 223 | |
| 224 | func (h *handlers) ttlInterval() time.Duration { |
| 225 | if h.streamTTL > 0 { |
| 226 | return h.streamTTL |
| 227 | } |
| 228 | return defaultStreamTTL |
| 229 | } |
| 230 | |
amadan