diff options
Diffstat (limited to 'packages/meshbay-node')
5 files changed, 199 insertions, 176 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/__init__.py new file mode 100644 index 0000000..a2e212a --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/__init__.py @@ -0,0 +1 @@ +"""What the node does for one app only: Videos, Music.""" diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py new file mode 100644 index 0000000..bd2ab52 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py @@ -0,0 +1,170 @@ +"""Subtitle tracks for the Videos app: extracted once, whole-file, to WebVTT.""" + +import asyncio +import logging +import time + +import blake3 +from meshbay_common import MNP_VERSION +from meshbay_common.protocol import MNP + +from meshbay_node.media_probe import probe_video as _probe_video +from meshbay_node.roots import off_disk +from meshbay_node.transport.webrtc.disk import _locate +from meshbay_node.transport.webrtc.media_tools import ( + _extract_subtitle_to_webvtt, + _subtitle_timeout_for, +) + +log = logging.getLogger("meshbay_node.transport.webrtc_server") + + +class SubtitlesMixin: + async def _do_subtitle_request(self, msg: dict) -> None: + """ + One embedded subtitle track, extracted whole-file to WebVTT and served + back through the ordinary file_req/chunk path — the same indirection + as `_do_audio_transcode_request`, and cached the same way, so a + film's subtitles are extracted once in the life of the file rather + than once per viewing. + + The ordinal is validated against `probe_video`'s *filtered* list and + then used as the ffmpeg `-map 0:s:<n>` argument, which is only correct + because `SubtitleTrack.ordinal` counts every subtitle stream including + the bitmap ones the list omits (see media_probe.py). Checking + membership rather than range is what makes that hold: a bitmap + ordinal is in range and is not in the list, and extracting it would + produce an empty WebVTT — a subtitle track with no subtitles in it, + which reports no error anywhere. + """ + ctx = self._group_ctx() + file_id = msg.get("file_id", "") + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) + return + + media_cache = self._ctx.get("media_cache") + if media_cache is None: + self._send({"type": "error", "detail": "Subtitles unavailable"}) + return + + try: + ordinal = int(msg.get("track", 0) or 0) + except (TypeError, ValueError): + ordinal = -1 + + # Logged before the cache is consulted, and that ordering is the point: + # a cached track used to answer without leaving a line, so the journal + # could not say whether a viewer had asked for subtitles at all. That + # turned "no request in the log" into evidence it was never entitled + # to be — the second time in this feature that a silent success was + # read as an absence. + log.info("subtitle: req file=%s track=%d size=%.1fMB slots_free=%s", + file_id[:12], ordinal, entry.size / 1e6, + getattr(self._ctx.get("_transcode_sem"), "_value", "?")) + t0 = time.monotonic() + + synthetic_id = f"subtitle:{entry.id}:{ordinal}" + cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) + if cached_hash is not None: + blob = await media_cache.get_thumb(cached_hash) + if blob is not None: + log.info("subtitle: served file=%s track=%d from cache, %d bytes", + file_id[:12], ordinal, len(blob)) + self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION, + "file_id": file_id, "track": ordinal, + "hash": cached_hash, "size": len(blob), + "mime": "text/vtt"}) + return + # Cached hash but the blob was pruned: fall through and extract + # again, same as a cold cache. + + 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 + + # 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 + + 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) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py new file mode 100644 index 0000000..fd3184c --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py @@ -0,0 +1,23 @@ +"""Blocking disk work the session hands to `off_disk`.""" + +from pathlib import Path + +from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path + + +def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: + """ + Where an entry is, and whether it is readable — or the refusal to send. + + Both halves are syscalls: `resolve()` walks the path and `exists()` stats + it, and a stat is what *wakes* a sleeping disk. Leaving either on the event + loop and offloading only the read would move the stall rather than remove + it, and the read would then find the disk already awake. Blocking; called + through `off_disk`. + """ + path = entry_abs_path(roots, entry) + if path is None: + return None, ROOT_NOT_SERVED + if not path.exists(): + return None, "File not on disk" + return path, None 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 66a0f3c..53840f6 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -141,12 +141,12 @@ from meshbay_node.roots import ( SAFE_UPLOAD_NAME, RootSet, _free_name, - entry_abs_path, off_disk, safe_subdir, ) from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK from meshbay_node.transfers import TransferSlots +from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin from meshbay_node.transport.webrtc.channel import ( _REPLY_TO, _DataChannelBuffer, @@ -154,11 +154,10 @@ from meshbay_node.transport.webrtc.channel import ( _get_remote_ip, _pack, ) +from meshbay_node.transport.webrtc.disk import _locate from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG from meshbay_node.transport.webrtc.media_tools import ( - _extract_subtitle_to_webvtt, _seek_lands_at, - _subtitle_timeout_for, _transcode_audio_to_aac, ) from meshbay_node.transport.wire import index_sync_message @@ -378,7 +377,7 @@ _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 -class WebRTCPeerSession: +class WebRTCPeerSession(SubtitlesMixin): """One WebRTC peer connection, handling MNP over a DataChannel.""" def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""): @@ -4079,155 +4078,6 @@ class WebRTCPeerSession: "file_id": file_id, "hash": transcode_hash, "size": len(blob), "mime": "audio/mp4"}) - async def _do_subtitle_request(self, msg: dict) -> None: - """ - One embedded subtitle track, extracted whole-file to WebVTT and served - back through the ordinary file_req/chunk path — the same indirection - as `_do_audio_transcode_request` above, and cached the same way, so a - film's subtitles are extracted once in the life of the file rather - than once per viewing. - - The ordinal is validated against `probe_video`'s *filtered* list and - then used as the ffmpeg `-map 0:s:<n>` argument, which is only correct - because `SubtitleTrack.ordinal` counts every subtitle stream including - the bitmap ones the list omits (see media_probe.py). Checking - membership rather than range is what makes that hold: a bitmap - ordinal is in range and is not in the list, and extracting it would - produce an empty WebVTT — a subtitle track with no subtitles in it, - which reports no error anywhere. - """ - ctx = self._group_ctx() - file_id = msg.get("file_id", "") - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send({"type": "error", "detail": "File not found"}) - return - file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) - if refusal is not None: - self._send({"type": "error", "detail": refusal}) - return - - media_cache = self._ctx.get("media_cache") - if media_cache is None: - self._send({"type": "error", "detail": "Subtitles unavailable"}) - return - - try: - ordinal = int(msg.get("track", 0) or 0) - except (TypeError, ValueError): - ordinal = -1 - - # Logged before the cache is consulted, and that ordering is the point: - # a cached track used to answer without leaving a line, so the journal - # could not say whether a viewer had asked for subtitles at all. That - # turned "no request in the log" into evidence it was never entitled - # to be — the second time in this feature that a silent success was - # read as an absence. - log.info("subtitle: req file=%s track=%d size=%.1fMB slots_free=%s", - file_id[:12], ordinal, entry.size / 1e6, - getattr(self._ctx.get("_transcode_sem"), "_value", "?")) - t0 = time.monotonic() - - synthetic_id = f"subtitle:{entry.id}:{ordinal}" - cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) - if cached_hash is not None: - blob = await media_cache.get_thumb(cached_hash) - if blob is not None: - log.info("subtitle: served file=%s track=%d from cache, %d bytes", - file_id[:12], ordinal, len(blob)) - self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION, - "file_id": file_id, "track": ordinal, - "hash": cached_hash, "size": len(blob), - "mime": "text/vtt"}) - return - # Cached hash but the blob was pruned: fall through and extract - # again, same as a cold cache. - - 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 - - # 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 - - 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: """ docs/MESHBAY_DESIGN.md §9.8: MusicBrainz metadata for one track, resolved @@ -6772,24 +6622,6 @@ def _rmdir_if_empty(target: Path) -> bool: return True -def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: - """ - Where an entry is, and whether it is readable — or the refusal to send. - - Both halves are syscalls: `resolve()` walks the path and `exists()` stats - it, and a stat is what *wakes* a sleeping disk. Leaving either on the event - loop and offloading only the read would move the stall rather than remove - it, and the read would then find the disk already awake. Blocking; called - through `off_disk`. - """ - path = entry_abs_path(roots, entry) - if path is None: - return None, ROOT_NOT_SERVED - if not path.exists(): - return None, "File not on disk" - return path, None - - def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" with open(tmp_path, "wb" if first else "ab") as f: diff --git a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py index 7a27a37..32b503b 100644 --- a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py +++ b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py @@ -37,11 +37,8 @@ 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, - _subtitle_timeout_for, -) +from meshbay_node.transport.webrtc.media_tools import _subtitle_timeout_for +from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video from conftest import needs_subprocess, one_root |