fix(node): size the subtitle budget to the file, and extract each track once

Verifying the feature across a library turned up two faults, one of which
broke it outright.

**The extraction budget was a constant and the cost is not.** Extracting
demuxes the whole container, so the file sets the price: measured at 9.8 s per
GB on a library held on an external disk — 36 s for a 3.9 GB title, 71 s for a
7.3 GB one. Against a flat 60 s that worked on most of a library and failed on
the big films, which to whoever is watching one is indistinguishable from a
broken feature. `Avatar.The.Way.of.Water` timed out every time. The budget is
now three times the measured rate per gigabyte, floored at the old 60 s and
capped so no container can pin a transcode slot indefinitely. The client's
own timeout follows it: the node always answers, so that one is a backstop
against a silent peer rather than a deadline for the work.

**Two clicks ran two extractions.** The cache is consulted on the way in, so a
second request arriving while the first was still running missed it and ran
its own — seen in the log as two identical extractions of one 4.3 GB file
overlapping, each holding a transcode slot and reading the file end to end.
Latecomers now wait on the answer the first is producing.

The in-flight entry is registered *before* the first await, not after. The
first version registered it after the probe, two concurrent requests both got
past the check while neither had registered, and the test for it failed —
which is the only reason this note can be written from the right side.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UGY17EPph5LsLzePPXhUVc

2 files changed, 155 insertions, 41 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 47cffcc..540c174 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -292,6 +292,23 @@ AUDIO_TRANSCODE_TIMEOUT_SECS = 120
 # pathological container rather than against the work itself, and it is short
 # next to the audio one because nothing here decodes a media stream.
 SUBTITLE_EXTRACT_TIMEOUT_SECS = 60
+# Extracting a subtitle demuxes the whole container, so the cost is set by the
+# file and not by the subtitle: measured at **9.8 s per GB** on a library held
+# on an external disk — 36 s for a 3.9 GB title, 71 s for a 7.3 GB one. A flat
+# 60 s therefore worked on most of a library and failed on the big films, with
+# nothing to distinguish that from a broken feature. The allowance is three
+# times the measured rate so a slower disk, or one being read by a stream at
+# the same time, still finishes.
+SUBTITLE_EXTRACT_SECS_PER_GB = 30
+SUBTITLE_EXTRACT_TIMEOUT_MAX_SECS = 900
+
+
+def _subtitle_timeout_for(size_bytes: int) -> float:
+    """How long this file is allowed to take. See the constants above."""
+    gb = max(0.0, size_bytes) / 1_000_000_000
+    return min(SUBTITLE_EXTRACT_TIMEOUT_MAX_SECS,
+               max(SUBTITLE_EXTRACT_TIMEOUT_SECS,
+                   SUBTITLE_EXTRACT_SECS_PER_GB * gb))
 # Probing where a copied seek lands costs 0.06–0.07 s on a real title, so this
 # bounds a pathological container rather than the work itself.
 SEEK_PROBE_TIMEOUT_SECS = 10
@@ -4101,47 +4118,90 @@ class WebRTCPeerSession:
             # Cached hash but the blob was pruned: fall through and extract
             # again, same as a cold cache.
 
-        probe = await _probe_video(str(file_path))
-        if not any(tr.ordinal == ordinal for tr in probe.subtitle_tracks):
-            # Not a range check — see this method's docstring.
-            log.warning("subtitle: file=%s has no text track %d", file_id[:12], ordinal)
-            self._send({"type": "error", "detail": "No such subtitle track"})
+        inflight = self._ctx.setdefault("_subtitle_inflight", {})
+        running = inflight.get(synthetic_id)
+        if running is not None:
+            log.info("subtitle: file=%s track=%d already extracting, waiting",
+                     file_id[:12], ordinal)
+            # Shielded: this waiter being cancelled must not cancel the
+            # extraction every other waiter is also relying on.
+            answer = await asyncio.shield(running)
+            if answer is None:
+                self._send({"type": "error", "detail": "Subtitle extraction failed"})
+                return
+            subtitle_hash, size = answer
+            log.info("subtitle: served file=%s track=%d from a shared extraction, %d bytes",
+                     file_id[:12], ordinal, size)
+            self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                        "file_id": file_id, "track": ordinal,
+                        "hash": subtitle_hash, "size": size, "mime": "text/vtt"})
             return
 
-        sem = self._transcode_semaphore()
-        if sem.locked() and sem._value <= 0:
-            log.info("subtitle: refused, no transcode slot free")
-            self._send({"type": "error", "detail": "Server busy, retry shortly"})
-            return
-        async with sem:
-            log.info("subtitle: extracting file=%s track=%d (slot taken)",
-                     file_id[:12], ordinal)
-            try:
-                blob = await _extract_subtitle_to_webvtt(file_path, ordinal)
-            # BaseException, not Exception: a cancelled task — the peer went
-            # away, the session is being torn down — raises CancelledError,
-            # which is not an Exception and would otherwise leave this handler
-            # with no reply sent and no line in the log. The client is then
-            # waiting on something nothing will ever answer, which is exactly
-            # the shape that is impossible to report.
-            except BaseException as e:
-                log.warning("subtitle: extract failed file=%s track=%d after %.1fs: %r",
-                            file_id[:12], ordinal, time.monotonic() - t0, e)
-                self._send({"type": "error",
-                            "detail": f"Subtitle extraction failed: {e}"})
-                if isinstance(e, asyncio.CancelledError):
-                    raise
+        # One extraction per file and track at a time, across every viewer.
+        # The cache is only consulted on the way in, so two clicks seconds
+        # apart both missed it and both ran — seen in the log as two identical
+        # extractions of one 4.3 GB file overlapping, each holding a transcode
+        # slot and reading the whole file. Latecomers wait for the answer the
+        # first one is already producing.
+        #
+        # **Registered before the first `await` below**, never after. The
+        # first version registered it after the probe, and two requests
+        # arriving together both got past the check above while neither had
+        # registered yet — so both extracted, which is the race this exists to
+        # close. The probe and the slot check sit behind it for that reason.
+        fut = asyncio.get_running_loop().create_future()
+        inflight[synthetic_id] = fut
+        try:
+            probe = await _probe_video(str(file_path))
+            if not any(tr.ordinal == ordinal for tr in probe.subtitle_tracks):
+                # Not a range check — see this method's docstring.
+                log.warning("subtitle: file=%s has no text track %d",
+                            file_id[:12], ordinal)
+                self._send({"type": "error", "detail": "No such subtitle track"})
+                return
+
+            sem = self._transcode_semaphore()
+            if sem.locked() and sem._value <= 0:
+                log.info("subtitle: refused, no transcode slot free")
+                self._send({"type": "error", "detail": "Server busy, retry shortly"})
                 return
 
-        subtitle_hash = blake3.blake3(blob).hexdigest()
-        await media_cache.put_thumb(subtitle_hash, synthetic_id, blob)
-        self._audit("subtitle_extract", f"{entry.name} [{ordinal}]")
-        log.info("subtitle: extracted file=%s track=%d in %.1fs, %d bytes",
-                 file_id[:12], ordinal, time.monotonic() - t0, len(blob))
-        self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
-                    "file_id": file_id, "track": ordinal,
-                    "hash": subtitle_hash, "size": len(blob),
-                    "mime": "text/vtt"})
+            budget = _subtitle_timeout_for(entry.size)
+            async with sem:
+                log.info("subtitle: extracting file=%s track=%d (slot taken, up to %.0fs)",
+                         file_id[:12], ordinal, budget)
+                blob = await _extract_subtitle_to_webvtt(file_path, ordinal, budget)
+
+            subtitle_hash = blake3.blake3(blob).hexdigest()
+            await media_cache.put_thumb(subtitle_hash, synthetic_id, blob)
+            self._audit("subtitle_extract", f"{entry.name} [{ordinal}]")
+            log.info("subtitle: extracted file=%s track=%d in %.1fs, %d bytes",
+                     file_id[:12], ordinal, time.monotonic() - t0, len(blob))
+            fut.set_result((subtitle_hash, len(blob)))
+            self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
+                        "file_id": file_id, "track": ordinal,
+                        "hash": subtitle_hash, "size": len(blob),
+                        "mime": "text/vtt"})
+        # BaseException, not Exception: a cancelled task — the peer went away,
+        # the session is being torn down — raises CancelledError, which is not
+        # an Exception and would otherwise leave this handler with no reply
+        # sent and no line in the log. The client is then waiting on something
+        # nothing will ever answer, which is the shape nobody can report.
+        except BaseException as e:
+            log.warning("subtitle: extract failed file=%s track=%d after %.1fs: %r",
+                        file_id[:12], ordinal, time.monotonic() - t0, e)
+            self._send({"type": "error",
+                        "detail": f"Subtitle extraction failed: {e}"})
+            if isinstance(e, asyncio.CancelledError):
+                raise
+        finally:
+            # A result either way, never an exception: a future nobody is
+            # waiting on yet would be an unretrieved-exception warning, and a
+            # cancelled task would otherwise leave every latecomer waiting on
+            # a future that is never resolved at all.
+            if not fut.done():
+                fut.set_result(None)
+            inflight.pop(synthetic_id, None)
 
     async def _do_music_meta_request(self, msg: dict) -> None:
         """
@@ -6597,7 +6657,8 @@ async def _seek_lands_at(file_path: Path, t: float, map_args: list[str]) -> floa
     return landed
 
 
-async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int) -> bytes:
+async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int,
+                                      timeout: float = SUBTITLE_EXTRACT_TIMEOUT_SECS) -> bytes:
     """
     One subtitle track out of a container, whole, as WebVTT.
 
@@ -6631,11 +6692,11 @@ async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int) -> bytes:
         )
         try:
             _, stderr = await asyncio.wait_for(
-                proc.communicate(), timeout=SUBTITLE_EXTRACT_TIMEOUT_SECS)
+                proc.communicate(), timeout=timeout)
         except asyncio.TimeoutError:
             proc.kill()
             await proc.wait()
-            raise RuntimeError(f"ffmpeg timed out after {SUBTITLE_EXTRACT_TIMEOUT_SECS}s")
+            raise RuntimeError(f"ffmpeg timed out after {timeout:.0f}s")
         if proc.returncode != 0:
             raise RuntimeError(
                 f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}")
diff --git a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py
index 9a09edb..d3bcc8b 100644
--- a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py
+++ b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py
@@ -38,7 +38,11 @@ from meshbay_common.crypto import generate_gek
 from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
 from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.media_probe import TEXT_SUBTITLE_CODECS
-from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video
+from meshbay_node.transport.webrtc_server import (
+    WebRTCPeerSession,
+    _probe_video,
+    _subtitle_timeout_for,
+)
 
 from conftest import needs_subprocess, one_root
 
@@ -356,3 +360,52 @@ async def test_the_result_is_fetched_through_the_ordinary_chunk_path(tmp_path):
     plain = decrypt_chunk_aes(key, chunk["nonce"], chunk["ct"])
     assert plain.decode("utf-8").startswith("WEBVTT")
     assert _CUE_WORD[2] in plain.decode("utf-8")
+
+
+def test_the_budget_grows_with_the_file_not_with_the_subtitle():
+    """Extraction demuxes the whole container, so the file sets the cost.
+
+    Measured on a library held on an external disk: 9.8 s per GB — 36 s for a
+    3.9 GB title and 71 s for a 7.3 GB one. A flat 60 s therefore worked on
+    most of a library and failed on the big films, which is indistinguishable
+    from a broken feature to whoever is watching one. The allowance is three
+    times the measured rate, so a slower disk still finishes.
+    """
+    assert _subtitle_timeout_for(500_000_000) == 60           # small file, the floor
+    assert _subtitle_timeout_for(7_310_000_000) > 71 * 2      # the film that timed out
+    assert _subtitle_timeout_for(3_900_000_000) > 36 * 2
+    # Bounded: a pathological container must not pin a transcode slot for ever.
+    assert _subtitle_timeout_for(500_000_000_000) == 900
+    # Monotonic, or a bigger file could be given less time than a smaller one.
+    budgets = [_subtitle_timeout_for(int(gb * 1e9)) for gb in (1, 4, 8, 20, 100)]
+    assert budgets == sorted(budgets)
+
+
+@pytest.mark.asyncio
+async def test_two_requests_for_one_track_extract_once(tmp_path):
+    """Two clicks seconds apart used to run two whole extractions.
+
+    The cache is consulted on the way in, so the second request missed it
+    while the first was still running: seen in the log as two identical
+    extractions of one 4.3 GB file overlapping, each holding a transcode slot
+    and reading the file end to end. The latecomer waits for the answer the
+    first is already producing, and both are answered.
+    """
+    clip = tmp_path / "clip.mp4"
+    _make_subtitled_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    import asyncio
+    await asyncio.gather(
+        session._do_subtitle_request({"file_id": file_id, "track": 1}),
+        session._do_subtitle_request({"file_id": file_id, "track": 1}),
+    )
+
+    replies = [m for m in session.sent if m.get("type") == "subtitle_resp"]
+    assert len(replies) == 2, f"both callers must be answered: {session.sent}"
+    assert replies[0]["hash"] == replies[1]["hash"]
+    assert session._ctx["media_cache"].puts == 1, (
+        "the second request ran its own extraction instead of joining the first")
+    assert not session._ctx["_subtitle_inflight"], (
+        "the in-flight entry outlived the extraction and would block the next one")