1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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)
|