diff options
Diffstat (limited to 'packages')
3 files changed, 162 insertions, 42 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 63c71ff..1a28fb0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -1619,8 +1619,14 @@ class MeshBayTransport { * a position in the list this client received. */ async requestSubtitle(fileId, track) { + // Generous because the node's own bound is, and for the same reason: the + // extraction demuxes the whole container, measured at 9.8 s per GB on an + // external disk — 71 s for a 7.3 GB film, and more for a 4K one. The node + // always answers, with a refusal if its own budget runs out, so this is a + // backstop against a peer that has gone silent rather than a deadline for + // the work. It is paid once per film: every later viewing is cached. const msg = await this._sendAndWait( - { type: 'subtitle_req', v: '0.9', file_id: fileId, track }, 90000); + { type: 'subtitle_req', v: '0.9', file_id: fileId, track }, 900000); if (msg.type === 'error') throw new Error(msg.detail); return msg; } 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") |