diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 141 |
1 files changed, 101 insertions, 40 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]}") |