| 1 | // fixtures/live-updates.mjs — the editor's live-update subscriber, |
| 2 | // factored out of fixtures/editor-entry.js so its cursor arithmetic |
| 3 | // and connection policy can be exercised under plain Node with no |
| 4 | // DOM, no EventSource and no real Y.Doc (fixtures/ |
| 5 | // live-updates.test.mjs). Same shape as save-queue.mjs and |
| 6 | // resync-controller.mjs, and for the same reason: the control flow |
| 7 | // being tested does not depend on any of those being real. |
| 8 | // |
| 9 | // What it does: an SSE stream tells this tab that the document has |
| 10 | // content through some seq; this module fetches whatever is newer |
| 11 | // than its own cursor through the ordinary binary updates endpoint |
| 12 | // and hands the bytes to its caller. The stream carries no document |
| 13 | // bytes (see internal/docs/stream.go's own doc comment for why), so |
| 14 | // everything here is cursor bookkeeping and connection policy. |
| 15 | // |
| 16 | // It deliberately owns no DOM. The caller wires visibility and |
| 17 | // activity events to noteHidden/noteActivity, which keeps this file |
| 18 | // testable and keeps every listener registration in one place in |
| 19 | // editor-entry.js. |
| 20 | |
| 21 | // IDLE_MS is how long a visible tab with nothing happening keeps its |
| 22 | // stream open. Platform hibernation is default-on and this app holds |
| 23 | // one connection per open editor, so a document left open on Friday |
| 24 | // must not pin its instance awake all weekend (the design spec's §6, |
| 25 | // and the docs spec's §7.3). Two minutes is well under the server's |
| 26 | // own five-minute TTL, so an idle tab closes on its own terms rather |
| 27 | // than waiting to be cut off. |
| 28 | const IDLE_MS = 120000 |
| 29 | |
| 30 | // HIDDEN_MS is the same thing for a tab that has been switched away |
| 31 | // from, and it exists because closing on `hidden` immediately — which |
| 32 | // is what this module did first — is what a person sees as a flash. |
| 33 | // |
| 34 | // The sequence, measured in a real browser rather than reasoned about: |
| 35 | // tab away, the stream closes; tab back, and the browser paints the |
| 36 | // document THEN, from what is already in the DOM, which is the version |
| 37 | // from before you left. The reconnect and its catch-up fetch land 18ms |
| 38 | // later on a loopback address, and the paragraph visibly rewrites |
| 39 | // itself. Every tab switch cost one frame of stale text, and no amount |
| 40 | // of making the reconnect faster removes it, because the paint happens |
| 41 | // before any round trip can possibly return. |
| 42 | // |
| 43 | // So the fix is not to reconnect faster but not to have disconnected. |
| 44 | // Thirty seconds covers switching between two tabs, which is the whole |
| 45 | // of the case that flashed, and it is still four times more aggressive |
| 46 | // than the visible-idle close: a document left open on Friday drops its |
| 47 | // stream half a minute after you look away, which serves §7.3's |
| 48 | // hibernation requirement strictly better than the 120s a merely idle |
| 49 | // visible tab is already allowed. |
| 50 | const HIDDEN_MS = 30000 |
| 51 | |
| 52 | // updatesURL is the one place either query string this module (or its |
| 53 | // caller in editor-entry.js) can ask for gets spelled, so the spelling |
| 54 | // is pinned by a plain-Node test instead of living twice, inline, in a |
| 55 | // DOM-bound file this project has no test harness for. handlers.go |
| 56 | // matches the raw flag as the literal string "1" and nothing else — |
| 57 | // `raw=true`, `raw=on`, or a bare `raw` all fail silently, with no |
| 58 | // error and no visible symptom beyond the resync controller refetching |
| 59 | // the same corrupt snapshot forever — so this is deliberately a pure |
| 60 | // string builder with no fetch and no DOM, testable by asserting exact |
| 61 | // output rather than by loading a browser. |
| 62 | export function updatesURL(base, { since, raw } = {}) { |
| 63 | const params = [] |
| 64 | if (since !== undefined) params.push(`since=${since}`) |
| 65 | if (raw) params.push('raw=1') |
| 66 | return params.length ? `${base}?${params.join('&')}` : base |
| 67 | } |
| 68 | |
| 69 | export function createLiveUpdates({ |
| 70 | initialSeq = 0, |
| 71 | // openStream(onEvent, onError) -> { close() }. onEvent receives the |
| 72 | // seq an "update" event carried. |
| 73 | openStream, |
| 74 | // fetchSince(since, { raw }) -> { bytes, seq }. bytes may be empty; |
| 75 | // seq is what the server's Docs-Seq header said. raw, when passed |
| 76 | // as { raw: true }, is the decode-failure escape hatch: it must |
| 77 | // reach the server as the literal query string `raw=1` (handlers.go |
| 78 | // matches nothing else), and asks for exactly the doc_updates rows |
| 79 | // after `since` with the snapshot lookup skipped entirely — see the |
| 80 | // comment on the retry in drain() below for why this exists. |
| 81 | fetchSince, |
| 82 | onUpdates, |
| 83 | onError = () => {}, |
| 84 | setTimeoutFn = setTimeout, |
| 85 | clearTimeoutFn = clearTimeout, |
| 86 | idleMs = IDLE_MS, |
| 87 | hiddenMs = HIDDEN_MS, |
| 88 | }) { |
| 89 | let cursor = initialSeq |
| 90 | let stream = null |
| 91 | let idleTimer = null |
| 92 | let stopped = false |
| 93 | |
| 94 | // fetching/pending are the whole of the serialization: one request |
| 95 | // in flight at a time, and a single bit remembering that something |
| 96 | // arrived while it was. Without this a burst of notifications |
| 97 | // starts a fetch per event, and their responses race each other |
| 98 | // into the Y.Doc. |
| 99 | let fetching = false |
| 100 | let pending = false |
| 101 | |
| 102 | // One timer, two deadlines. A hidden tab and an idle visible one are |
| 103 | // closed by the same mechanism and differ only in how long they get, |
| 104 | // so whichever is armed last wins and there is no way for the two to |
| 105 | // be pending at once and race each other. |
| 106 | function armClose(ms) { |
| 107 | if (idleTimer !== null) clearTimeoutFn(idleTimer) |
| 108 | idleTimer = setTimeoutFn(() => { |
| 109 | idleTimer = null |
| 110 | closeStream() |
| 111 | }, ms) |
| 112 | } |
| 113 | |
| 114 | function armIdle() { |
| 115 | armClose(idleMs) |
| 116 | } |
| 117 | |
| 118 | function closeStream() { |
| 119 | if (idleTimer !== null) { |
| 120 | clearTimeoutFn(idleTimer) |
| 121 | idleTimer = null |
| 122 | } |
| 123 | if (!stream) return |
| 124 | stream.close() |
| 125 | stream = null |
| 126 | } |
| 127 | |
| 128 | // openIfClosed is the ONLY place a connection is opened, and every |
| 129 | // close above is explicit. Native EventSource reconnection is never |
| 130 | // relied on anywhere in this module, and that is load-bearing |
| 131 | // rather than fastidious: the browser retries on any server close, |
| 132 | // including the idle close this module just chose and the server's |
| 133 | // own TTL, so a parked tab would reopen its stream forever and go |
| 134 | // on holding a hibernating instance awake. rastrillo's own SSE |
| 135 | // handler (jobs/events.go) does rely on native reconnection, which |
| 136 | // is right for it — it has no close policy of its own — and is the |
| 137 | // one piece of that recipe whose reason does not survive being |
| 138 | // moved here. |
| 139 | function openIfClosed() { |
| 140 | if (stopped || stream) return |
| 141 | stream = openStream( |
| 142 | (seq) => { |
| 143 | armIdle() |
| 144 | // Advisory, never an assignment: this says "there is |
| 145 | // something at least this new", and the fetch is what |
| 146 | // determines the cursor. A tab that has already seen this |
| 147 | // seq — its own write, echoed back — has nothing to do. |
| 148 | if (seq > cursor) drain() |
| 149 | }, |
| 150 | (err) => { |
| 151 | // A dead stream is a freshness failure, never a save failure: |
| 152 | // POST is a different path and edits are still saving, so |
| 153 | // this must not reach the save indicator. The connection is |
| 154 | // dropped rather than retried on a timer, and the next |
| 155 | // activity brings it back. |
| 156 | closeStream() |
| 157 | onError(err) |
| 158 | }, |
| 159 | ) |
| 160 | armIdle() |
| 161 | } |
| 162 | |
| 163 | async function drain() { |
| 164 | if (fetching) { |
| 165 | pending = true |
| 166 | return |
| 167 | } |
| 168 | fetching = true |
| 169 | try { |
| 170 | const from = cursor |
| 171 | let res = await fetchSince(from) |
| 172 | if (stopped) return |
| 173 | try { |
| 174 | if (res && res.bytes && res.bytes.length) onUpdates(res.bytes) |
| 175 | } catch (decodeErr) { |
| 176 | // onUpdates threw: Y.readUpdate hit bytes it could not decode. |
| 177 | // The server serves a snapshot without ever calling into the |
| 178 | // CRDT core (handlers.go's own comment on raw=1), so it cannot |
| 179 | // detect a snapshot corrupted on disk after it was written and |
| 180 | // verified — the browser, via this exact throw, is the only |
| 181 | // reader that can. Without this retry the resync controller |
| 182 | // would refetch the same corrupt bytes forever and the |
| 183 | // document would never load. The retry asks for the SAME |
| 184 | // range (`from`, unchanged) with the snapshot lookup skipped |
| 185 | // entirely, which is the one thing the server has always been |
| 186 | // able to serve correctly. A second throw here — from the |
| 187 | // retried fetch itself, or from onUpdates on the retried |
| 188 | // bytes — is not caught a second time; it falls through to |
| 189 | // the outer catch below, which reports it and leaves the |
| 190 | // cursor untouched. |
| 191 | res = await fetchSince(from, { raw: true }) |
| 192 | if (stopped) return |
| 193 | if (res && res.bytes && res.bytes.length) onUpdates(res.bytes) |
| 194 | } |
| 195 | // max, never assignment: a response that overtakes another must |
| 196 | // not drag the cursor backwards and cause everything above it |
| 197 | // to be fetched and applied a second time. |
| 198 | if (res && typeof res.seq === 'number' && res.seq > cursor) cursor = res.seq |
| 199 | } catch (err) { |
| 200 | // Leave the cursor where it is. Either the bytes were never |
| 201 | // received, or they were received twice (plain, then raw) and |
| 202 | // neither applied — either way anything else would skip them, |
| 203 | // so the next notification (or the connect-time event on the |
| 204 | // next reopen) retries from the same place. |
| 205 | onError(err) |
| 206 | } finally { |
| 207 | fetching = false |
| 208 | if (pending && !stopped) { |
| 209 | pending = false |
| 210 | drain() |
| 211 | } |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | return { |
| 216 | start() { |
| 217 | stopped = false |
| 218 | openIfClosed() |
| 219 | }, |
| 220 | |
| 221 | // stop is called on teardown and on every resync — the Y.Doc this |
| 222 | // was feeding is about to be discarded, so a response still in |
| 223 | // flight must not be applied to its replacement. |
| 224 | stop() { |
| 225 | stopped = true |
| 226 | closeStream() |
| 227 | }, |
| 228 | |
| 229 | // noteOwnSeq records the seq the server assigned to this tab's own |
| 230 | // POST, and is where the round trip is saved in the common case. |
| 231 | // |
| 232 | // The guard is the whole of it: adopt the number only when it is |
| 233 | // exactly the next row. If this tab is at 10 and its own write |
| 234 | // landed at 43, then rows 11-42 came from somewhere else and have |
| 235 | // not been applied here — taking 43 would skip them permanently, |
| 236 | // with no later notification able to bring them back, because |
| 237 | // every future fetch would ask for what came after 43. Staying at |
| 238 | // 10 costs one fetch and loses nothing. |
| 239 | noteOwnSeq(seq) { |
| 240 | if (typeof seq !== 'number' || !Number.isFinite(seq)) return |
| 241 | if (seq === cursor + 1) cursor = seq |
| 242 | }, |
| 243 | |
| 244 | // noteActivity is any sign of a person: a pointer moving, a |
| 245 | // scroll, a key going down, a tab becoming visible, a local edit. |
| 246 | // It both defers the idle close and brings a closed stream back. |
| 247 | // |
| 248 | // "Any activity" is deliberately broad. Visibility, focus and |
| 249 | // local edits alone leave the commonest case unhandled — someone |
| 250 | // READING a long document on a tab that is already visible and |
| 251 | // already focused fires none of them, so the stream would close |
| 252 | // on idle and nothing would ever reopen it. That tab is exactly |
| 253 | // the watching half of "type in one tab, watch the other", and it |
| 254 | // would have gone quietly stale forever. |
| 255 | noteActivity() { |
| 256 | if (stopped) return |
| 257 | if (stream) armIdle() |
| 258 | else openIfClosed() |
| 259 | }, |
| 260 | |
| 261 | // noteHidden is the tab going into the background. It shortens the |
| 262 | // deadline rather than closing, so a switch away and back inside |
| 263 | // the grace never drops the stream and the tab is already current |
| 264 | // when it is painted — see HIDDEN_MS for what the immediate close |
| 265 | // looked like to a person. |
| 266 | // |
| 267 | // Nothing is opened here: a hidden tab with a closed stream stays |
| 268 | // closed, because coming back is activity and noteActivity is what |
| 269 | // reopens it. |
| 270 | noteHidden() { |
| 271 | if (stopped || !stream) return |
| 272 | armClose(hiddenMs) |
| 273 | }, |
| 274 | |
| 275 | get cursor() { return cursor }, |
| 276 | get streaming() { return stream !== null }, |
| 277 | } |
| 278 | } |
| 279 | |
amadan