fix(client): the cast relay's header, lead and restart

The header is everything before the first moof, a receiver far behind
loses nothing, and a stop not waited for no longer clears the next start.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 319 insertions, 36 deletions

diff --git a/packages/meshbay-client/src/cast-relay.js b/packages/meshbay-client/src/cast-relay.js
index ed20b337..e51d52eb 100644
--- a/packages/meshbay-client/src/cast-relay.js
+++ b/packages/meshbay-client/src/cast-relay.js
@@ -56,6 +56,9 @@ const util = require('node:util');
 const debug = util.debuglog('cast-relay');
 
 const RING_CAP = 64;
+// A fragment is a whole segment, megabytes at a film's bitrate: 64 of them
+// can be hundreds of megabytes, so the backlog is bounded in bytes too.
+const RING_MAX_BYTES = 32 * 1024 * 1024;
 // What the receiver is ever told a photo is. The page re-encodes every photo
 // to JPEG, so this list is a guard, not a menu.
 const FILE_TYPES = new Set([
@@ -66,7 +69,16 @@ const FILE_TYPES = new Set([
 // A track is a few megabytes, a long lossless one a few hundred; this only
 // stops a page from filling the machine's memory.
 const FILE_MAX_BYTES = 1024 * 1024 * 1024;
-const BACKPRESSURE_HIGH = 8 * 1024 * 1024;
+// How far a receiver may fall behind before fragments are dropped for it.
+// The page runs ahead of the television by its whole read-ahead (up to 48 MB,
+// video-player.js) plus the receiver's start-up, and a fragment is a few
+// seconds of picture: the 8 MB this used to be dropped one every few seconds
+// from a receiver reading at playback speed, and each froze the picture for
+// its length. Node holds what waits in the response's own buffer.
+const CLIENT_LEAD_MAX = 512 * 1024 * 1024;
+// A receiver that connects before the first moof waits for the whole header.
+const HEADER_WAIT_MS = 10000;
+const FTYP = 0x66747970;
 const MOOF = 0x6d6f6f66;
 const PORT_BASE = 19550;
 const PORT_COUNT = 4;
@@ -97,6 +109,10 @@ class BoxAccumulator {
   constructor() {
     this._buf = Buffer.alloc(0);
     this._synced = false;
+    this._preambleSeen = false;
+    // Called once, with every byte before the first moof: the stream's
+    // header (ftyp, moov), however many pushes it came in.
+    this.onPreamble = null;
   }
 
   push(data) {
@@ -106,7 +122,11 @@ class BoxAccumulator {
     if (!this._synced) {
       const idx = this._findMoof();
       if (idx === -1) return fragments;
-      debug(`[cast-relay] box sync: found first moof at byte offset ${idx}, discarded ${idx} bytes`);
+      debug(`[cast-relay] box sync: found first moof at byte offset ${idx}`);
+      if (!this._preambleSeen) {
+        this._preambleSeen = true;
+        if (this.onPreamble) this.onPreamble(Buffer.from(this._buf.subarray(0, idx)));
+      }
       this._buf = this._buf.subarray(idx);
       this._synced = true;
     }
@@ -149,6 +169,7 @@ class BoxAccumulator {
   reset() {
     this._buf = Buffer.alloc(0);
     this._synced = false;
+    this._preambleSeen = false;
   }
 
   _findMoof() {
@@ -164,6 +185,29 @@ class BoxAccumulator {
   }
 }
 
+/**
+ * The header is what comes before the first moof, and only that.
+ *
+ * The page hands over the first chunk it decrypted, which is a slice of the
+ * stream and not a box: measured, it was by turns the 28-byte ftyp alone, the
+ * header exactly, or the header and the start of the first fragment,
+ * depending on when the node read ffmpeg's output. Served as the header, the
+ * first left the receiver without a moov and the third broke the box
+ * structure, so the same film cast one try in three. The pushed stream
+ * carries the whole header; this trims the page's chunk for the case where
+ * nothing came before the first moof.
+ */
+function headerOnly(init) {
+  let at = 0;
+  while (at + 8 <= init.length) {
+    if (init.readUInt32BE(at + 4) === MOOF) return init.subarray(0, at);
+    const size = init.readUInt32BE(at);
+    if (size < 8) break;
+    at += size;
+  }
+  return init;
+}
+
 class CastRelay {
   constructor() {
     this._server = null;
@@ -171,12 +215,16 @@ class CastRelay {
     this._token = null;
     this._lanIP = null;
     this._initSegment = null;
+    this._headerReady = false;
+    this._waiting = new Set();
     this._ring = [];
+    this._ringBytes = 0;
     this._clients = new Set();
     this._accum = new BoxAccumulator();
     this._fragCount = 0;
     this._chunkCount = 0;
     this._bytesSent = 0;
+    this._dropped = 0;
     this._subtitle = null;
     this._subtitleVersion = 0;
     this._file = null;
@@ -303,21 +351,20 @@ class CastRelay {
   }
 
   async start({ codec, initSegment, subtitle }) {
-    if (this._server) await this.stop();
+    // Everything the new stream needs is set before the first await: the
+    // page's pushes arrive while this waits for its port, and they belong to
+    // the new stream's accumulator, header first.
+    const closing = this._teardown();
 
     this._token = crypto.randomBytes(16).toString('hex');
     this._lanIP = lanAddress();
-    this._initSegment = initSegment ? Buffer.from(initSegment) : null;
-    this._ring = [];
-    this._clients = new Set();
-    this._accum = new BoxAccumulator();
-    this._fragCount = 0;
-    this._chunkCount = 0;
-    this._bytesSent = 0;
-    this._subtitle = null;
-    this._file = null;
-    this._incoming = null;
+    this._initSegment = initSegment ? headerOnly(Buffer.from(initSegment)) : null;
+    // Nothing to wait for without a first chunk: no header is coming.
+    this._headerReady = !initSegment;
+    this._accum.onPreamble = (bytes) => this._preamble(bytes);
     this.setSubtitle(subtitle);
+    const token = this._token;
+    await closing;
 
     const server = http.createServer((req, res) => this._handle(req, res));
 
@@ -333,19 +380,23 @@ class CastRelay {
           server.on('error', onError);
           server.listen(port, this._lanIP, () => {
             server.removeListener('error', onError);
-            this._port = port;
             resolve();
           });
         });
-        bound = true;
+        bound = port;
       } catch (err) {
         if (err.code !== 'EADDRINUSE') throw err;
         debug(`[cast-relay] port ${port} busy, trying next`);
       }
     }
     if (!bound) throw new Error('All cast relay ports are in use');
+    if (this._token !== token) {
+      server.close();
+      throw new Error('Cast relay stopped while starting');
+    }
 
     this._server = server;
+    this._port = bound;
     debug(`[cast-relay] started on ${this.url}`);
     debug(`[cast-relay] init segment: ${this._initSegment ? this._initSegment.length + ' bytes' : 'none'}`);
     return {
@@ -356,6 +407,18 @@ class CastRelay {
     };
   }
 
+  /**
+   * Every byte the page pushed before the first moof. Kept as the header when
+   * it starts with an ftyp; the page's own first chunk, trimmed, otherwise.
+   */
+  _preamble(bytes) {
+    if (bytes.length >= 8 && bytes.readUInt32BE(4) === FTYP) this._initSegment = bytes;
+    debug(`[cast-relay] header complete: ${this._initSegment ? this._initSegment.length : 0} bytes`);
+    this._headerReady = true;
+    for (const res of this._waiting) this._admit(res);
+    this._waiting.clear();
+  }
+
   pushSegment(data) {
     const buf = Buffer.isBuffer(data) ? data : Buffer.from(data);
     this._chunkCount++;
@@ -368,14 +431,22 @@ class CastRelay {
         debug(`[cast-relay] fragment #${this._fragCount}: ${frag.length} bytes (from ${this._chunkCount} chunks), ${this._clients.size} client(s)`);
       }
 
-      if (this._ring.length >= RING_CAP) {
-        this._ring.shift();
-      }
       this._ring.push(frag);
+      this._ringBytes += frag.length;
+      while (this._ring.length > RING_CAP
+             || (this._ringBytes > RING_MAX_BYTES && this._ring.length > 1)) {
+        this._ringBytes -= this._ring.shift().length;
+      }
 
       for (const res of this._clients) {
-        if (res.writableLength > BACKPRESSURE_HIGH) {
-          debug(`[cast-relay] backpressure: dropping fragment for slow client`);
+        if (res.writableLength > CLIENT_LEAD_MAX) {
+          // The picture on the television freezes for this fragment's length.
+          this._dropped++;
+          if (this._dropped <= 3 || this._dropped % 20 === 0) {
+            debug(`[cast-relay] dropped a fragment of ${frag.length} bytes for a`
+              + ` receiver ${(res.writableLength / 1048576).toFixed(0)} MB behind,`
+              + ` ${this._dropped} dropped so far`);
+          }
           continue;
         }
         res.write(frag);
@@ -386,35 +457,55 @@ class CastRelay {
 
   finish() {
     debug('[cast-relay] finishing stream');
+    for (const res of this._waiting) this._admit(res);
+    this._waiting.clear();
     for (const res of this._clients) {
       try { res.end(); } catch { /* already closed */ }
     }
   }
 
   async stop() {
-    debug(`[cast-relay] stopping — ${this._chunkCount} chunks, ${this._fragCount} fragments, ${(this._bytesSent / 1048576).toFixed(1)} MB sent`);
-    for (const res of this._clients) {
-      try { res.end(); } catch { /* already closed */ }
-    }
-    this._clients.clear();
+    await this._teardown();
+  }
 
-    if (this._server) {
-      const srv = this._server;
-      this._server = null;
-      await new Promise((resolve) => srv.close(resolve));
+  /**
+   * Everything is reset before anything is awaited, and the promise only says
+   * when the old port is closed. The page calls stop() without waiting and
+   * start() next: a stop that cleared its state after the close cleared the
+   * new stream's instead (its token, port and header) and the receiver was
+   * handed `:null/stream.mp4?t=null`. The close used to wait, too, for the
+   * receiver to let go of the old stream, which it does only once loaded with
+   * the new one; its connections are cut instead.
+   */
+  _teardown() {
+    debug(`[cast-relay] stopping: ${this._chunkCount} chunks, ${this._fragCount} fragments, ${(this._bytesSent / 1048576).toFixed(1)} MB sent, ${this._dropped} dropped`);
+    for (const res of [...this._clients, ...this._waiting]) {
+      try { res.end(); } catch { /* already closed */ }
     }
+    this._clients = new Set();
+    this._waiting = new Set();
 
+    const srv = this._server;
+    this._server = null;
     this._port = null;
     this._token = null;
     this._initSegment = null;
+    this._headerReady = false;
     this._subtitle = null;
     this._file = null;
     this._incoming = null;
     this._ring = [];
-    this._accum.reset();
+    this._ringBytes = 0;
+    this._accum = new BoxAccumulator();
     this._fragCount = 0;
     this._chunkCount = 0;
     this._bytesSent = 0;
+    this._dropped = 0;
+
+    if (!srv) return Promise.resolve();
+    const closed = new Promise((resolve) => srv.close(resolve));
+    srv.closeAllConnections();
+    return closed;
   }
 
   _handle(req, res) {
@@ -469,7 +560,6 @@ class CastRelay {
       return;
     }
 
-    let sent = 0;
     res.writeHead(200, {
       // A receiver that has been given a side-loaded track reads the media
       // through the same CORS-checked path as the track, so the headers go on
@@ -482,6 +572,31 @@ class CastRelay {
       'Connection': 'keep-alive',
     });
 
+    res.flushHeaders();
+    req.on('close', () => {
+      this._clients.delete(res);
+      this._waiting.delete(res);
+      debug(`[cast-relay] client disconnected, ${this._clients.size} remaining`);
+    });
+
+    if (this._headerReady) {
+      this._admit(res);
+      return;
+    }
+    // Before the first moof the header may be a piece of itself: the
+    // receiver waits for all of it, or for the page's chunk at worst.
+    this._waiting.add(res);
+    const token = this._token;
+    setTimeout(() => {
+      if (this._token !== token || !this._waiting.delete(res)) return;
+      debug('[cast-relay] no moof yet, serving the first chunk as the header');
+      this._admit(res);
+    }, HEADER_WAIT_MS).unref();
+  }
+
+  /** The header, the backlog, then live fragments from here on. */
+  _admit(res) {
+    let sent = 0;
     if (this._initSegment) {
       res.write(this._initSegment);
       sent += this._initSegment.length;
@@ -492,13 +607,9 @@ class CastRelay {
       sent += frag.length;
     }
 
-    debug(`[cast-relay] client connected from ${req.socket.remoteAddress} — sent init + ${this._ring.length} fragments (${(sent / 1024).toFixed(0)} KB)`);
+    debug(`[cast-relay] client connected from ${res.socket && res.socket.remoteAddress}: sent init + ${this._ring.length} fragments (${(sent / 1024).toFixed(0)} KB)`);
 
     this._clients.add(res);
-    req.on('close', () => {
-      this._clients.delete(res);
-      debug(`[cast-relay] client disconnected, ${this._clients.size} remaining`);
-    });
   }
 
   _handleSubtitle(res) {
@@ -573,3 +684,4 @@ class CastRelay {
 }
 
 module.exports = CastRelay;
+module.exports.headerOnly = headerOnly;
diff --git a/packages/meshbay-hub/tests/test_cast_relay_stream.py b/packages/meshbay-hub/tests/test_cast_relay_stream.py
new file mode 100644
index 00000000..a46f318c
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_cast_relay_stream.py
@@ -0,0 +1,171 @@
+"""
+The desktop cast relay's stream, run for real: plain Node, no dependencies.
+
+Three faults, each of which a receiver showed as a cast that hung or froze:
+
+* **The header is everything before the first moof.** The page hands the relay
+  its first decrypted chunk, which is a slice of the stream: the ftyp alone,
+  the header exactly, or the header and the start of a fragment, depending on
+  when the node read ffmpeg's output. Taken as the header, two of the three
+  broke the stream, so the same film cast one try in three.
+* **A receiver reading at playback speed loses nothing.** The page runs ahead
+  of the television by its whole read-ahead; dropped past 8 MB, each lost
+  fragment froze the picture for its few seconds.
+* **A stop the page does not wait for cannot clear the next start.** The old
+  stop cleared its state after the port closed, which was after the new
+  start, and the receiver was handed `:null/stream.mp4?t=null`.
+"""
+
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+PACKAGES = Path(__file__).resolve().parents[2]
+RELAY = PACKAGES / "meshbay-client" / "src" / "cast-relay.js"
+
+SCRIPT = r"""
+const http = require('node:http');
+const CastRelay = require(process.argv[2]);
+
+const box = (type, size) => {
+  const b = Buffer.alloc(size, type.charCodeAt(0));
+  b.writeUInt32BE(size, 0); b.write(type, 4, 'latin1');
+  return b;
+};
+const ftyp = box('ftyp', 28);
+const moov = box('moov', 1272);
+const header = Buffer.concat([ftyp, moov]);
+const fragment = (n, size) => Buffer.concat([box('moof', 100 + n), box('mdat', size)]);
+
+// Read a whole response, or only what it holds after `ms`.
+const read = (url, ms) => new Promise((resolve, reject) => {
+  const parts = [];
+  const req = http.get(url, { agent: false }, (res) => {
+    res.on('data', (c) => parts.push(c));
+    res.on('end', () => resolve(Buffer.concat(parts)));
+    if (ms) setTimeout(() => { req.destroy(); resolve(Buffer.concat(parts)); }, ms);
+  });
+  req.on('error', (e) => (ms ? null : reject(e)));
+});
+const sleep = (ms) => new Promise((ok) => setTimeout(ok, ms));
+
+// The page's order: start with its first chunk, then push every chunk,
+// that one included.
+async function cast(relay, chunks, { early } = {}) {
+  const started = await relay.start({ initSegment: chunks[0] });
+  const body = early ? read(started.url) : null;
+  if (early) await sleep(50);
+  for (const c of chunks) relay.pushSegment(c);
+  const served = early ? null : read(started.url);
+  await sleep(50);
+  relay.finish();
+  return (await (body || served)).toString('hex');
+}
+
+(async () => {
+  const out = {};
+  const frags = [fragment(1, 300000), fragment(2, 800000), fragment(3, 250000)];
+  const stream = Buffer.concat([header, ...frags]);
+  out.expected = stream.toString('hex');
+
+  const relay = new CastRelay();
+  // The three first chunks measured on a real film.
+  out.ftypAlone = await cast(relay, [ftyp, moov, ...frags]);
+  out.headerExactly = await cast(relay, [header, ...frags]);
+  const cut = header.length + 4000;
+  out.headerAndFilm = await cast(relay, [stream.subarray(0, cut), stream.subarray(cut)]);
+  // A receiver connected before the first moof waits for the whole header.
+  out.earlyReceiver = await cast(relay, [ftyp, moov, ...frags], { early: true });
+  await relay.stop();
+
+  // A receiver that reads nothing while 40 MB of film is pushed, then reads.
+  const started = await relay.start({ initSegment: header });
+  let slowRes = null;
+  const parts = [];
+  const done = new Promise((resolve) => {
+    http.get(started.url, { agent: false }, (res) => {
+      slowRes = res; res.pause();
+      res.on('data', (c) => parts.push(c));
+      res.on('end', resolve);
+    }).on('error', () => {});
+  });
+  await sleep(100);
+  out.headersBeforeTheHeader = slowRes !== null;
+  relay.pushSegment(header);
+  const big = [];
+  for (let i = 0; i < 8; i++) big.push(fragment(i, 5 * 1024 * 1024));
+  for (const f of big) relay.pushSegment(f);
+  relay.finish();
+  slowRes.resume();
+  await done;
+  out.slowReceiverGotAll = Buffer.concat(parts).equals(Buffer.concat([header, ...big]));
+  out.dropped = relay._dropped;
+
+  // stop() not awaited, then start(), while a receiver still holds the old
+  // stream: the new relay keeps its own token, port and header.
+  const holder = read(relay.url, 300);
+  await sleep(50);
+  relay.stop();
+  const next = await relay.start({ initSegment: header });
+  await holder;
+  await sleep(100);
+  out.afterRestart = { url: relay.url, returned: next.url, active: relay.active };
+  relay.pushSegment(header);
+  relay.pushSegment(frags[0]);
+  const after = read(relay.url);
+  await sleep(50);
+  relay.finish();
+  out.afterRestartBody = (await after).toString('hex');
+  out.afterRestartExpected = Buffer.concat([header, frags[0]]).toString('hex');
+  await relay.stop();
+
+  out.trimmed = CastRelay.headerOnly(stream.subarray(0, cut)).toString('hex');
+  out.header = header.toString('hex');
+  console.log(JSON.stringify(out));
+})().catch((e) => { console.error(e); process.exit(1); });
+"""
+
+
+@pytest.fixture(scope="module")
+def run(tmp_path_factory):
+    if shutil.which("node") is None or not RELAY.exists():
+        pytest.skip("node or the desktop client sources are not available")
+    script = tmp_path_factory.mktemp("relay") / "stream.cjs"
+    script.write_text(SCRIPT, encoding="utf-8")
+    proc = subprocess.run(["node", str(script), str(RELAY)],
+                          capture_output=True, text=True, encoding="utf-8", timeout=60)
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+@pytest.mark.parametrize("first_chunk", ["ftypAlone", "headerExactly", "headerAndFilm"])
+def test_the_header_is_everything_before_the_first_moof(run, first_chunk):
+    assert run[first_chunk] == run["expected"]
+
+
+def test_a_receiver_that_connects_first_waits_for_the_whole_header(run):
+    assert run["earlyReceiver"] == run["expected"]
+
+
+def test_a_waiting_receiver_is_answered_at_once(run):
+    assert run["headersBeforeTheHeader"] is True
+
+
+def test_the_pages_first_chunk_is_trimmed_to_its_header(run):
+    assert run["trimmed"] == run["header"]
+
+
+def test_a_receiver_far_behind_loses_no_fragment(run):
+    assert run["slowReceiverGotAll"] is True
+    assert run["dropped"] == 0
+
+
+def test_a_stop_not_waited_for_leaves_the_next_start_alone(run):
+    r = run["afterRestart"]
+    assert r["active"] is True
+    assert r["url"] == r["returned"]
+    assert ":null" not in r["url"] and "t=null" not in r["url"]
+    assert run["afterRestartBody"] == run["afterRestartExpected"]