path: root/packages/meshbay-hub/tests/test_streamed_download_reliability.py

"""
The service-worker download path, which on Firefox and Safari is the only
unbounded way to write a file to disk.

Neither of those browsers has the File System Access API, and OPFS is not a
substitute: measured on Firefox 154, its quota is exactly 10% of the volume's
size (389,233,459 bytes on a 3,892,334,592-byte volume, refused to the byte),
which a film exceeds. So when this path declines, a large download has nowhere
left to go — there is no floor under it that can hold a film. That is what makes
its reliability a correctness property rather than a nicety.

The real module is imported under Node with the browser pieces it reaches
stubbed — `navigator.serviceWorker`, a document that "navigates" an iframe, and
Node's own TransformStream and MessageChannel, which are the real ones. What is
modelled is the environment; `serviceWorker()` and `openStreamedDownload()` are
executed, never reimplemented.

Three failures are pinned, all of which shipped:

  - registration happened inside the first click, so that click paid install,
    activate and claim while somebody watched a button do nothing;
  - a null result was cached for the life of the page, so one slow first click
    left the tab unable to stream anything again, curable only by a reload
    nobody knew to do;
  - one missed navigation fell straight through instead of retrying.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
DOWNLOADS = STATIC / "downloads.js"

pytestmark = pytest.mark.skipif(
    shutil.which("node") is None or not DOWNLOADS.exists(),
    reason="node or the SPA sources are not available")

# The stub browser. `plan` decides how the fake worker behaves, so one harness
# covers every case below.
PRELUDE = """
const store = new Map();
globalThis.localStorage = {
  getItem: k => (store.has(k) ? store.get(k) : null),
  setItem: (k, v) => store.set(k, String(v)),
  removeItem: k => store.delete(k),
};
const PLAN = %(plan)s;
const log = { registers: 0, claims: 0, navigations: 0, served: 0,
              unregisters: 0, wakes: 0 };

// The worker as the page sees it: something with postMessage. It answers a
// navigation by posting mbdl-serving back on the port it was handed, which is
// exactly the confirmation the real sw.js sends from its fetch handler.
let controller = null;
const pendingByFrame = new Map();
// Set before the controller exists, because the declaration below is what
// the temporal dead zone protects.
let asleep = PLAN.workerAsleep;
const makeController = () => ({
  postMessage: (msg, transfer) => {
    // A worker with nothing to do is terminated, and `pending` goes with it.
    // A ping wakes it; anything else posted while it sleeps is simply lost,
    // which is what makes this failure silent.
    if (asleep) {
      if (msg.type === 'mbdl-ping') {
        asleep = false;
        log.wakes += 1;
        if (msg.ports || (transfer && transfer[0])) {
          const port = (transfer && transfer[0]) || null;
          if (port) setTimeout(() => port.postMessage({type: 'mbdl-pong'}), 0);
        }
      }
      return;
    }
    if (msg.type === 'mbdl-ping') {
      const port = (transfer && transfer[0]) || null;
      if (port) setTimeout(() => port.postMessage({type: 'mbdl-pong'}), 0);
      return;
    }
    if (msg.type === 'mbdl-claim') {
      log.claims += 1;
      // A worker that actually claims when asked, which is what sw.js does.
      if (PLAN.controlOnClaim) {
        controller = makeController();
        for (const fn of listeners) fn();
      }
      return;
    }
    if (msg.type !== 'mbdl') return;
    pendingByFrame.set('/_mbdl/' + msg.id, msg.port);
    // The worker says it has it, which is what the page waits for.
    if (msg.port) setTimeout(() => msg.port.postMessage({type: 'mbdl-ready',
                                                         id: msg.id}), 0);
  },
});

const listeners = new Set();
// `globalThis.navigator` is read-only from Node 22 -- assigning to it is the
// mistake CLAUDE.md already records against test_locales.py. Define it.
Object.defineProperty(globalThis, 'navigator', {
  configurable: true,
  value: {
    serviceWorker: {
      get controller() { return controller; },
      // What the document started with, which is the whole of the repair's
      // evidence now. `getRegistration` is asked before anything registers.
      getRegistration: async () => (PLAN.registeredAtLoad
                                    ? {active: makeController()} : undefined),
      register: async () => {
        log.registers += 1;
        if (PLAN.registerThrows) throw new Error('registration blocked');
        // A registration that never answers at all. Distinct from one that
        // rejects: nothing is reported, nothing fails, the caller just waits.
        if (PLAN.registerHangs) await new Promise(() => {});
        // A worker that only becomes installable once the stuck registration
        // has been thrown away -- the browser this was reported from.
        const healed = PLAN.activeAfterUnregister && log.unregisters > 0;
        if (PLAN.controlAfterMs !== null || healed) {
          setTimeout(() => {
            controller = makeController();
            for (const fn of listeners) fn();
          }, healed ? 0 : PLAN.controlAfterMs);
        }
        return {
          active: (PLAN.active || healed) ? makeController() : null,
          unregister: async () => { log.unregisters += 1; return true; },
        };
      },
      // `register()` resolves as soon as the registration object exists, with
      // nothing but an installing worker; `ready` is what waits for an active
      // one. Measured on Firefox 154: an install handler that rejects leaves
      // `ready` unsettled past ten seconds while `register()` returns in 7 ms.
      get ready() {
        const healed = PLAN.activeAfterUnregister && log.unregisters > 0;
        return (PLAN.readySettles || healed)
          ? Promise.resolve({}) : new Promise(() => {});
      },
      addEventListener: (type, fn) => { if (type === 'controllerchange') listeners.add(fn); },
      removeEventListener: (type, fn) => { listeners.delete(fn); },
    },
  },
});

globalThis.window = globalThis;
globalThis.isSecureContext = true;
// Set before the module is imported, because it reads it at evaluation.
controller = %(controlled)s ? makeController() : null;
// The self-test's repair reloads once and remembers it for the tab; both have
// to exist here or priming the worker throws instead of repairing.
const session = new Map();
globalThis.sessionStorage = {
  getItem: k => (session.has(k) ? session.get(k) : null),
  setItem: (k, v) => session.set(k, String(v)),
  removeItem: k => session.delete(k),
};
log.reloads = 0;
globalThis.location = { reload: () => { log.reloads += 1; } };
globalThis.document = {
  createElement: () => ({ hidden: false, src: '', remove() {} }),
  body: {
    appendChild: (frame) => {
      log.navigations += 1;
      const port = pendingByFrame.get(frame.src);
      const answer = PLAN.serveOnNavigation === 'always'
        || (PLAN.serveOnNavigation === 'second' && log.navigations >= 2);
      if (port && answer) {
        log.served += 1;
        setTimeout(() => {
          port.postMessage({type: 'mbdl-serving', id: frame.src});
          // The worker's own copy of the port, dropped once answered. sw.js
          // drops it with the pending entry; here it has to be explicit or the
          // harness process never exits.
          port.close();
        }, 0);
      }
    },
  },
};

const M = await import('%(module)s');
// Production waits 15 s for each; these cases are about which branch runs.
const FAST = {controlMs: %(control)d, servedMs: 400};
const out = {};
"""


def _run(tmp_path, body, *, control_after_ms=0, active=True,
         serve="always", register_throws=False, control_budget_ms=800,
         ready_settles=True, register_hangs=False,
         active_after_unregister=False, control_on_claim=False,
         controlled_at_load=False, registered_at_load=False,
         worker_asleep=False):
    module = tmp_path / "downloads.mjs"
    module.write_text(DOWNLOADS.read_text())
    (tmp_path / "package.json").write_text('{"type":"module"}')
    plan = {
        "controlAfterMs": control_after_ms,
        "active": active,
        "serveOnNavigation": serve,
        "registerThrows": register_throws,
        "readySettles": ready_settles,
        "registerHangs": register_hangs,
        "activeAfterUnregister": active_after_unregister,
        "controlOnClaim": control_on_claim,
        "registeredAtLoad": registered_at_load,
        "workerAsleep": worker_asleep,
    }
    script = tmp_path / "case.mjs"
    script.write_text(
        (PRELUDE % {"plan": json.dumps(plan), "module": module.as_posix(),
                    "control": control_budget_ms,
                    "controlled": json.dumps(controlled_at_load)})
        + body
        + "\nout.log = log;\nconsole.log(JSON.stringify(out));\n")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True,
                          timeout=120)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


# ── A failure must never be cached ──────────────────────────────────────────

def test_a_missed_claim_does_not_poison_the_page(tmp_path):
    """
    The bug: `_swReady` held the null, so every later download in that tab got
    it back without trying. One slow first click and the tab could not stream
    again — on Firefox, that is every large download for the rest of the visit.

    Here the worker never takes control, so the first call fails; the second
    must register again rather than return a remembered null.
    """
    r = _run(tmp_path, """
      out.first = await M.openStreamedDownload('a.bin', 10, FAST) !== null;
      const after = log.registers;
      out.second = await M.openStreamedDownload('b.bin', 10, FAST) !== null;
      out.registeredAgain = log.registers > after;
    """, control_after_ms=None)
    assert r["first"] is False and r["second"] is False
    assert r["registeredAgain"] is True, "a failed attempt was cached"


def test_a_success_is_reused_rather_than_re_registered(tmp_path):
    """The other half: once controlled, it must not re-register per download."""
    r = _run(tmp_path, """
      // Closed, like a real caller: an open target holds a keep-alive interval
      // for the worker, and a test that leaks one never lets Node exit.
      for (const name of ['a.bin', 'b.bin']) {
        const t = await M.openStreamedDownload(name, 10, FAST);
        out[name[0]] = t !== null;
        if (t) await t.writable.close();
      }
    """)
    assert r["a"] and r["b"]
    assert r["log"]["registers"] <= 1, "re-registered on a page already controlled"


# ── Waiting for control, rather than giving up ──────────────────────────────

def test_control_arriving_late_is_still_used(tmp_path):
    """
    Control used to be waited for with a 3 s cap, inside the click. A cold
    worker on a busy machine can take longer, and the old code called that a
    browser that cannot stream. Scaled down here — the budget is a parameter, so
    what is pinned is that a claim arriving after the first check is still used,
    not the particular number of seconds.
    """
    r = _run(tmp_path, """
      const t0 = Date.now();
      const target = await M.openStreamedDownload('film.mkv', 20e9, FAST);
      out.ok = target !== null;
      out.waitedMs = Date.now() - t0;
      if (target) await target.writable.close();
    """, control_after_ms=1200, control_budget_ms=6000)
    assert r["ok"] is True, "gave up on a claim that arrived late"
    assert r["waitedMs"] >= 1100, "did not actually wait for the claim"


def test_an_uncontrolled_page_asks_the_worker_to_claim_again(tmp_path):
    """
    Active but not controlling — a page loaded before any worker existed, whose
    claim was missed. Rather than declare the path unavailable, ask again.
    """
    r = _run(tmp_path, """
      out.ok = await M.openStreamedDownload('a.bin', 10, FAST) !== null;
    """, control_after_ms=None, active=True)
    assert r["log"]["claims"] >= 1, "never asked the active worker to claim"


# ── Retrying a missed navigation ────────────────────────────────────────────

def test_a_missed_navigation_is_retried(tmp_path):
    """
    The worker takes the stream and is then never asked for the URL. The page
    used to give up at once; on Firefox that sends a film to the in-memory
    floor. It gets a second go, with a fresh id and a fresh iframe.
    """
    r = _run(tmp_path, """
      const t = await M.openStreamedDownload('film.mkv', 20e9, FAST);
      out.ok = t !== null;
      if (t) await t.writable.close();
    """, serve="second")
    assert r["ok"] is True, "one missed navigation ended the download"
    assert r["log"]["navigations"] == 2


def test_giving_up_says_why(tmp_path):
    """
    A silent null is what made the original defect invisible. Whatever happens,
    the reason has to be readable afterwards — it is what the refusal quotes.
    """
    r = _run(tmp_path, """
      out.target = await M.openStreamedDownload('a.bin', 10, FAST);
      out.why = M.lastStreamFailure();
    """, control_after_ms=None)
    assert r["target"] is None
    assert r["why"], "declined with no stated reason"


def test_a_registration_that_throws_is_reported_not_swallowed(tmp_path):
    r = _run(tmp_path, """
      out.target = await M.openStreamedDownload('a.bin', 10, FAST);
      out.why = M.lastStreamFailure();
    """, register_throws=True)
    assert r["target"] is None
    assert "registration" in r["why"]


# ── Wiring that the behavioural cases cannot see ────────────────────────────

def test_the_worker_is_primed_at_boot_not_at_the_first_click(tmp_path):
    """
    Registration inside the first download is the whole reason the claim was
    ever raced. `primeServiceWorker` has to be called where the app starts, and
    from a module that actually imports it — `node --check` would not notice a
    missing import, which is a mistake this repo has already shipped once.
    """
    app = (STATIC / "app.js").read_text()
    assert "downloads.primeServiceWorker()" in app, "nothing primes the worker"
    assert "import * as downloads from './downloads.js'" in app, (
        "app.js calls downloads.primeServiceWorker() without importing downloads")
    # In mount(), which runs at start-up — not inside a component or a handler.
    mount = app[app.index("const mount = () => {"):]
    assert "downloads.primeServiceWorker()" in mount[:mount.index("\n};")]


def test_the_worker_answers_a_re_claim(tmp_path):
    """The page's last resort before declaring the path unavailable only works
    if sw.js implements the other half."""
    sw = (STATIC / "sw.js").read_text()
    assert "mbdl-claim" in sw and "clients.claim()" in sw


# ── Nothing on this path may wait for ever ──────────────────────────────────

def test_a_worker_that_never_installs_does_not_hang_every_download(tmp_path):
    """The one that reached a person: four downloads stuck at "preparing", for
    ever, with nothing in the node's journal because no transfer had been asked
    for yet.

    `register()` resolves as soon as the registration object exists — with
    nothing but an *installing* worker — and `ready` waits for an active one.
    Measured on Firefox 154: an install handler that rejects leaves `ready`
    unsettled past ten seconds while `register()` returns in seven
    milliseconds. Neither had a deadline, and `_swPromise` is shared, so every
    download on the page waited on the same promise that would never settle.
    """
    out = _run(tmp_path, """
const t0 = Date.now();
out.worker = await M.openStreamedDownload('film.mkv', 1, FAST);
out.ms = Date.now() - t0;
out.why = M.lastStreamFailure();
""", ready_settles=False, active=False, control_after_ms=None,
       control_budget_ms=300)
    assert out["worker"] is None
    assert out["ms"] < 8000, (
        f"gave up after {out['ms']}ms — a budget that is not enforced is not a "
        "budget, and the row above it says 'preparing' the whole time")
    assert "active" in out["why"], out["why"]


def test_a_stuck_ready_does_not_throw_away_a_working_worker(tmp_path):
    """`ready` can be waiting on a *newer* worker that cannot install while an
    older one is perfectly able to serve. Giving up then would cost Firefox the
    only unbounded way it has to write a download to disk — a deadline must
    bound the waiting, never remove the capability."""
    out = _run(tmp_path, """
const target = await M.openStreamedDownload('film.mkv', 1, FAST);
out.target = target !== null;
// Closing stops the keep-alive; left open, its interval keeps this process
// alive well past the test's own timeout.
if (target) await target.writable.close();
""", ready_settles=False, active=True, control_budget_ms=300)
    assert out["target"] is True


def test_a_registration_that_never_answers_gives_up_too(tmp_path):
    """The other unbounded await. It rejects loudly in the case above; this is
    the case where it says nothing at all."""
    out = _run(tmp_path, """
const t0 = Date.now();
out.worker = await M.openStreamedDownload('film.mkv', 1, FAST);
out.ms = Date.now() - t0;
out.why = M.lastStreamFailure();
""", register_hangs=True, active=False, control_after_ms=None,
       control_budget_ms=300)
    assert out["worker"] is None
    assert out["ms"] < 8000, f"gave up after {out['ms']}ms"
    assert "register" in out["why"], out["why"]


def test_a_registration_stuck_installing_is_discarded_and_asked_for_again(tmp_path):
    """A deadline turns an invisible hang into a named failure, which is better
    but is not a fix: a registration stuck with nothing but an installing worker
    does not heal on its own. Every later visit finds the same registration and
    waits on the same `ready`, so the browser stays unable to stream a download
    until somebody opens developer tools — and on Firefox there is nothing else
    that can write a film to disk.

    So the stuck registration is thrown away and asked for once more.
    """
    out = _run(tmp_path, """
const target = await M.openStreamedDownload('film.mkv', 1, FAST);
out.target = target !== null;
if (target) await target.writable.close();
""", ready_settles=False, active=False, control_after_ms=None,
       active_after_unregister=True, control_budget_ms=300)
    assert out["log"]["unregisters"] == 1, (
        "the stuck registration was left in place")
    assert out["target"] is True, (
        "discarding it did not get the page a worker it could stream to")


# ── A page loaded with the worker bypassed ──────────────────────────────────


def test_a_hard_reloaded_page_reloads_itself_once(tmp_path):
    """Uncontrolled at load while an active registration already exists is a
    document fetched by a hard reload — Ctrl+F5, Ctrl+Shift+R — and nothing
    else. Measured on Chrome at document start: a first visit has neither, an
    ordinary reload has both, a hard reload has the registration and no
    controller.

    Such a page can still be claimed, so every control check passes; but the
    navigations it starts keep missing the worker, and the hidden iframe a
    streamed download needs is one. On Firefox and Safari that is the only way
    to write a file too large to hold in memory. An ordinary reload undoes it.
    """
    out = _run(tmp_path, """
      M.primeServiceWorker();
      await new Promise((r) => setTimeout(r, 200));
      out.reloads = log.reloads;
    """, controlled_at_load=False, registered_at_load=True)
    assert out["reloads"] == 1


def test_a_first_visit_is_not_a_bypass(tmp_path):
    """Also uncontrolled at load, and perfectly healthy: the worker is being
    installed right now and will claim the page in a moment. Reloading here
    would be a flicker on everybody's first visit — and it was, taking the
    group's WebRTC session down with it when it landed mid-connection."""
    out = _run(tmp_path, """
      M.primeServiceWorker();
      await new Promise((r) => setTimeout(r, 200));
      out.reloads = log.reloads;
    """, controlled_at_load=False, registered_at_load=False)
    assert out["reloads"] == 0


def test_a_controlled_page_does_not_reload(tmp_path):
    """The ordinary case, which must cost nothing at all: no reload, and no
    download spent asking. Chrome rations the downloads a page may start
    without a user gesture to about three, and the first version of this check
    asked its question by performing one — competing with the person's own
    downloads for that budget."""
    out = _run(tmp_path, """
      M.primeServiceWorker();
      await new Promise((r) => setTimeout(r, 200));
      out.reloads = log.reloads;
      out.navigations = log.navigations;
    """, controlled_at_load=True, registered_at_load=True)
    assert out["reloads"] == 0
    assert out["navigations"] == 0, (
        "priming performed a download; that budget belongs to the person")


def test_the_repair_happens_at_most_once(tmp_path):
    """The flag is in sessionStorage rather than a variable because the point is
    to survive the reload it triggers, and because a page that is still bypassed
    afterwards must stop rather than reload again, and again."""
    out = _run(tmp_path, """
      sessionStorage.setItem('meshbay.sw-repaired', '1');
      M.primeServiceWorker();
      await new Promise((r) => setTimeout(r, 200));
      out.reloads = log.reloads;
    """, controlled_at_load=False, registered_at_load=True)
    assert out["reloads"] == 0


# ── The claim is asked for, not waited for ──────────────────────────────────

def test_an_uncontrolled_page_asks_at_once_rather_than_after_the_budget(tmp_path):
    """A page that is uncontrolled while an active worker exists will not be
    claimed on its own — a document fetched by a hard reload is exactly that
    shape. Waiting the whole control budget first spends it on something that
    is not coming: about thirty seconds, measured, during which the person
    clicks download and watches four rows hang before the page repairs itself.
    """
    out = _run(tmp_path, """
      const t0 = Date.now();
      const target = await M.openStreamedDownload('film.mkv', 20e9, FAST);
      out.ms = Date.now() - t0;
      out.target = target !== null;
      out.claims = log.claims;
      // Closing stops the keep-alive; left open, its interval outlives the test.
      if (target) await target.writable.close();
    """, control_after_ms=None, control_on_claim=True, control_budget_ms=6000)
    assert out["target"] is True
    assert out["claims"] >= 1
    assert out["ms"] < 3000, (
        f"took {out['ms']}ms of a 6000ms budget — the claim was asked for only "
        "after the wait, not before it")


def test_a_download_waits_for_priming(tmp_path):
    """A click that lands while priming is still running must not race it: on a
    page about to reload, the attempt would fail for nothing."""
    out = _run(tmp_path, """
      M.primeServiceWorker();
      const t0 = Date.now();
      const target = await M.openStreamedDownload('film.mkv', 20e9, FAST);
      out.ms = Date.now() - t0;
      out.target = target !== null;
      if (target) await target.writable.close();
    """)
    assert out["target"] is True
    assert out["ms"] >= 1, "the download did not wait for priming at all"


def test_the_streamed_target_says_it_cannot_be_paused(tmp_path):
    """The value the widget's pause button is drawn from, read off the real
    module rather than a stub of it.

    It is false for a reason that is not about this code: the browser is already
    writing an HTTP response into its own download folder, so not feeding the
    stream stalls a download we can neither see nor resume, and an idle worker
    is terminated within seconds. Firefox and Safari therefore get cancel and no
    pause; Chrome gets one as soon as a download folder has been granted, which
    yields a held-open file instead of this.
    """
    out = _run(tmp_path, """
      const target = await M.openStreamedDownload('film.mkv', 20e9, FAST);
      out.pausable = target && target.pausable;
      if (target) await target.writable.close();
    """)
    assert out["pausable"] is False


# ── a worker that was asleep when we posted ─────────────────────────────────

def test_a_sleeping_worker_is_woken_before_it_is_handed_a_stream(tmp_path):
    """Reported from Chrome: a download started while an upload was running took
    thirty seconds to begin, every time.

    `pending` lives in the worker's memory and a worker with nothing to do is
    terminated — which is what a long upload leaves it, for minutes, since a
    WebRTC transfer gives it no events at all. The stream posted to it was lost;
    the iframe then woke it with nothing to find and the request fell through to
    the network, measured in the console as a 404 from the hub and fifteen
    seconds of silence, twice.

    `mbdl-ping` already existed — sent every ten seconds *while* writing, for
    the same reason. Nothing sent one before *starting*.
    """
    out = _run(tmp_path, """
      const target = await M.openStreamedDownload('film.mkv', 20e9, FAST);
      out.target = target !== null;
      out.wakes = log.wakes;
      out.navigations = log.navigations;
      if (target) await target.writable.close();
    """, worker_asleep=True)
    assert out["target"] is True, "the download never started"
    assert out["wakes"] == 1, "the worker was handed a stream while asleep"
    assert out["navigations"] == 1, (
        f"took {out['navigations']} attempts — the first one was wasted on a "
        "worker that had not been woken")