1 files changed, 171 insertions, 0 deletions
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"]