diff --git a/server/LIVE_PACKET.md b/server/LIVE_PACKET.md index 9ae7f5d..6af6347 100644 --- a/server/LIVE_PACKET.md +++ b/server/LIVE_PACKET.md @@ -2,12 +2,12 @@ Docker Compose sets `STREAMVAULT_LIVE_PACKET_AWARE=1` globally for current and future video channels. A direct server process still requires explicit `1`; any other value keeps the previous FFmpeg-per-session HLS implementation. `STREAMVAULT_LIVE_PACKET_AWARE=0` is the global emergency rollback, without touching raw archives or saved recordings. This does not alter archive/show capture or the legacy TS route. -The `/api/live/:id/authorize`, `/index.m3u8`, and `/segment/:segmentId.ts` handlers retain their existing app authentication, signed native-playback tickets, reader admission, and private cache. Per-channel admission happens before async setup. In packet mode, the worker accepts only a loopback `/api/stream/:id` URL and supplies an Authorization header when configured, without putting it on argv. The existing `/api/stream/:id` route itself remains publicly accessible as before; the new signed HLS routes do not change that route's access policy. One PyAV demux/mux plus one FFmpeg HLS mux spans clean upstream Content-Length and chunked EOFs. The final packet of BOTH H.264 and AAC remains private at each source boundary: a socket cut can make PyAV flush a truncated ADTS audio frame even with `is_corrupt=False`. A replay must exactly match the committed A/V history and replace each withheld packet with a same-timestamp, same-flags payload extending its prefix before any successor output. This prevents publishing a partial AAC frame or permanently rejecting its later complete counterpart. Source reads, seam probes, staged files, published cache, readers, and channel slots have bounds. FFmpeg retains a 64-entry staging manifest to survive accelerated replay; Node removes each staged TS after its private served copy is published, so staging media do not accumulate with the manifest. A failed seam or worker exit after accepting source media invalidates the cache and latches that channel unavailable until process restart; it never reconnects raw media into the same presentation. An initial HTTP/transport failure before accepting any media leaves no playable segment and permits a fresh worker on a later authorization (not a hot retry loop). Once media is established, an unavailable HTTP reopen receives at most three attempts inside the same worker, with cancellable 250 ms gaps. It does not refresh byte/publication watchdogs or replace the mux/timeline; any eventual media still undergoes full A/V seam proof. Exhaustion and unsafe bodies invalidate the presentation as before. Native players receive an unavailable response; the client retries 503/429 within a 25-second authorization budget before considering legacy fallback, honoring `Retry-After` without retrying earlier than requested and cancelling obsolete player/backend generations. Authentication rejections never permit fallback. Chromium uses Hls.js/MSE when supported even if it advertises native HLS; iPhone/Safari retain native HLS, and native-only browsers keep their fallback. This avoids Chromium's short native buffer and source-replay underruns seen in the ESPN browser soak. Unsupported tracks (including subtitles) and incompatible codecs fail closed in this mode. +The `/api/live/:id/authorize`, `/index.m3u8`, and `/segment/:segmentId.ts` handlers retain their existing app authentication, signed native-playback tickets, reader admission, and private cache. Per-channel admission happens before async setup. In packet mode, the worker accepts only a loopback `/api/stream/:id` URL and supplies an Authorization header when configured, without putting it on argv. The existing `/api/stream/:id` route itself remains publicly accessible as before; the new signed HLS routes do not change that route's access policy. One PyAV demux/mux plus one FFmpeg HLS mux spans clean upstream Content-Length and chunked EOFs. The final packet of BOTH H.264 and AAC remains private at each source boundary: a socket cut can make PyAV flush a truncated ADTS audio frame even with `is_corrupt=False`. A replay must exactly match the committed A/V history and replace each withheld packet with a same-timestamp, same-flags payload extending its prefix before any successor output. This prevents publishing a partial AAC frame or permanently rejecting its later complete counterpart. Source reads, seam probes, staged files, published cache, readers, and channel slots have bounds. FFmpeg retains a 64-entry staging manifest to survive accelerated replay; Node removes each staged TS after its private served copy is published, so staging media do not accumulate with the manifest. A failed seam or worker exit after accepting source media invalidates the cache and latches that channel unavailable until process restart; it never reconnects raw media into the same presentation. An initial HTTP/transport failure before accepting any media leaves no playable segment and permits a fresh worker on a later authorization (not a hot retry loop). Once media is established, an unavailable HTTP reopen receives at most three attempts inside the same worker, with cancellable one- and two-second cooldown gaps. It does not refresh byte/publication watchdogs or replace the mux/timeline; any eventual media still undergoes full A/V seam proof. Exhaustion and unsafe bodies invalidate the presentation as before. Native players receive an unavailable response; the client retries 503/429 within a 50-second authorization budget before considering legacy fallback, honoring `Retry-After` without retrying earlier than requested and cancelling obsolete player/backend generations. Authentication rejections never permit fallback. Chromium uses Hls.js/MSE when supported even if it advertises native HLS; iPhone/Safari retain native HLS, and native-only browsers keep their fallback. This avoids Chromium's short native buffer and source-replay underruns seen in the ESPN browser soak. Unsupported tracks (including subtitles) and incompatible codecs fail closed in this mode. Stream-copy cuts HLS at source keyframes, so a two-second requested cadence does not impose a four-second GOP limit. Published segments must have a finite duration greater than zero and at most 12 seconds, with the existing per-segment byte bound unchanged. Every playlist advertises a fixed `TARGETDURATION:12`; it does not change as short and long GOPs arrive. Latched unsafe channels return immediate 501 without `Retry-After`, while real warmup/capacity still returns retryable 503. This status distinction never clears a latch, resumes an unproved seam, or restores stale ticket media. Production diagnostics record only channel ID and fixed retirement category. -The runtime image installs Debian `python3` and `python3-av`. Node invokes `/usr/bin/python3` so PyAV is visible; Python's `/usr/local/bin/python3` may not see apt-installed modules. FFmpeg and Python diagnostics are suppressed because URLs can contain credentials. Upstream responses are streamed with a 16 KiB read cap, a bounded socket stall timeout, finite seam history/probe, and a staged/published disk ceiling; cumulative response bytes are **not** capped, since a legitimate live HTTP body may stay open for hours. Source-read progress is marked while consuming a repeated prefix, so the HLS publication watchdog does not kill an actively catching-up stitch; an actually silent source still times out. A separate 45-second cap without any NEW published segment kills and latches the channel even if bytes (such as TS null packets) keep arriving; bytes alone are not playable progress. The Node channel owner, not an arbitrary Python wall-clock timer, retires idle viewers. +The runtime image installs Debian `python3` and `python3-av`. Node invokes `/usr/bin/python3` so PyAV is visible; Python's `/usr/local/bin/python3` may not see apt-installed modules. FFmpeg and Python diagnostics are suppressed because URLs can contain credentials. Upstream responses are streamed with a 16 KiB read cap, a bounded 10-second socket timeout below the owner's unchanged 15-second source-silence deadline, finite seam history/probe, and a staged/published disk ceiling; cumulative response bytes are **not** capped, since a legitimate live HTTP body may stay open for hours. Source-read progress is marked while consuming a repeated prefix, so the HLS publication watchdog does not kill an actively catching-up stitch; an actually silent source still times out. A separate 45-second cap without any NEW published segment kills and latches the channel even if bytes (such as TS null packets) keep arriving; bytes alone are not playable progress. The Node channel owner, not an arbitrary Python wall-clock timer, retires idle viewers. Local integration tests exercise an authenticated loopback proxy, signed playlist **and** segment fetches, both clean EOF framings, an exact non-IDR continuation checked against a baseline packet inventory, and strict FFmpeg video/audio HLS decode. Negative cases include shifted AAC, a missing replay picture, incompatible video, false Content-Length, and an open silent source; previously cached segments are invalidated and the channel does not thrash-restart. Admission, reader cap, and shutdown cleanup are separately exercised. The original standalone matcher tests include ambiguous replay and bounded probing. A production-sized, natural-EOF fixture also exercises a fast two-session HLS manifest rollover on the Pi with the default poll interval and quotas. -**Acceptance scope:** CI builds and runs the actual x64 and ARM64 runtime images with Debian PyAV; x64 also runs the packaged server suite. Two natural live-proxy EOF responses (~84 MB each) were locally replayed across one non-IDR seam with strict HLS decode and once-only H.264/AAC boundary hashes. This does not prove sustained Pi throughput under concurrent viewers, replay catch-up against a physical player's buffered lead, or physical Samsung/iPhone playback. Audio-only retains its existing path; subtitle/codec layouts this worker cannot verify fail closed and can use the legacy route on a new authorization. A channel whose seam cannot be verified is never spliced with unproved media. +**Acceptance scope:** CI builds and runs the actual x64 and ARM64 runtime images with Debian PyAV; x64 also runs the packaged server suite. Two natural live-proxy EOF responses (~84 MB each) were locally replayed across one non-IDR seam with strict HLS decode and once-only H.264/AAC boundary hashes. This does not prove sustained Pi throughput under concurrent viewers, replay catch-up against a physical player's buffered lead, or physical Samsung/iPhone playback. Native Apple/WebKit clients request `/api/live-compatible` if the strict primary presentation is unavailable or fails. This independently signed, four-channel-capped FFmpeg stream-copy HLS buffer uses a sibling cache and new segment epochs, publishing explicit discontinuities rather than making a seamless-stitch claim. Primary unsafe latches and stale media remain invalid, and authorization rejection never downgrades to compatibility. Unavailable compatibility ends with an actionable error, not native raw MPEG-TS or an endless spinner. Audio-only retains its existing path. A channel whose seam cannot be verified is never spliced with unproved media. diff --git a/server/src/index.ts b/server/src/index.ts index 2a17e1e..5ce575f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -217,19 +217,29 @@ app.use(createArchiveRouter({ store: archiveStore, root: archiveRoot, secret: ar getChannel: getChannelById, getRecording, getPrograms: getArchivePrograms, start: id => archiveCapture.start(id), stop: id => archiveCapture.stopArchive(id), prioritize: (id, start, end) => archiveCapture.prioritizeWindow(id, start, end) })); -const liveBuffer = createLiveBuffer({ root: process.env.STREAMVAULT_LIVE_BUFFER_DIR || - path.join(process.env.TMPDIR || '/var/tmp', 'streamvault-live-buffer'), +const liveBufferRoot = path.resolve(process.env.STREAMVAULT_LIVE_BUFFER_DIR || + path.join(process.env.TMPDIR || '/var/tmp', 'streamvault-live-buffer')); +const liveBuffer = createLiveBuffer({ root: liveBufferRoot, packetAware: packetAwareLiveEnabled(process.env.STREAMVAULT_LIVE_PACKET_AWARE), // IDs are validated by packet ingestion; reasons are fixed server categories. // Never log source URLs, tickets, or worker stderr. onUnsafe: (reason, id) => console.warn('Live HLS unavailable', { channelId: id, reason }) }); -app.use('/api/live', createLiveRouter(liveBuffer, id => { +// Separate disposable cache and router secret: never reopen a latched primary +// presentation or make its old tickets/segment epochs valid in compatibility mode. +const compatibleLiveBuffer = createLiveBuffer({ + root: `${liveBufferRoot}-compatible`, + packetAware: false, maxChannels: 4, +}); +const liveSource = (id: string) => { + if (!/^[-A-Za-z0-9_]+$/.test(id)) return null; const channel = getChannelById(id); // The existing proxy validates upstream responses and hides credential-bearing // provider URLs from the FFmpeg process arguments. return channel?.content_type === 'livetv' && channel.url ? `http://127.0.0.1:${PORT}/api/stream/${encodeURIComponent(id)}?subs=1` : null; -})); +}; +app.use('/api/live', createLiveRouter(liveBuffer, liveSource)); +app.use('/api/live-compatible', createLiveRouter(compatibleLiveBuffer, liveSource, { basePath: '/api/live-compatible' })); // Request logging; include completion timing only when a request is slow. app.use((req, res, next) => { @@ -2414,7 +2424,7 @@ function shutdown(signal: string): void { }); const recorderAndAnalysisShutdown = (async () => { - await Promise.all([schedulerShutdown, stopAllRecordings(), archiveCapture.stopAll(), liveBuffer.stop()]); + await Promise.all([schedulerShutdown, stopAllRecordings(), archiveCapture.stopAll(), liveBuffer.stop(), compatibleLiveBuffer.stop()]); await stopCommercialAnalysisWorker(); })().catch(error => { logger.warn(`Recorder/analysis shutdown failed: ${error instanceof Error ? error.message : error}`); diff --git a/server/src/live-buffer.ts b/server/src/live-buffer.ts index 1c1cb29..fc6f231 100644 --- a/server/src/live-buffer.ts +++ b/server/src/live-buffer.ts @@ -357,7 +357,9 @@ export function createLiveBuffer(options: LiveBufferOptions = {}) { }; } -export function createLiveRouter(buffer: ReturnType, source: (id: string) => string | null | undefined) { +export function createLiveRouter(buffer: ReturnType, source: (id: string) => string | null | undefined, + options: { basePath?: '/api/live' | '/api/live-compatible' } = {}) { + const basePath = options.basePath ?? '/api/live'; const router = Router(); const secret = randomBytes(32); const ttl = 24 * 60 * 60_000; @@ -410,7 +412,7 @@ export function createLiveRouter(buffer: ReturnType, so if (buffer.isUnsafe(id)) { res.status(501).end(); return; } if (!ready) { res.set('Retry-After', '2').status(503).end(); return; } const { ticket, expiresAt } = ticketFor(id); - res.set('Cache-Control', 'private, no-store').json({ playlistUrl: `/api/live/${encodeURIComponent(id)}/index.m3u8?ticket=${ticket}`, expiresAt }); + res.set('Cache-Control', 'private, no-store').json({ playlistUrl: `${basePath}/${encodeURIComponent(id)}/index.m3u8?ticket=${ticket}`, expiresAt }); }); router.get('/:id/index.m3u8', async (req, res) => { const id = String(req.params.id); diff --git a/server/src/live-compatible.test.ts b/server/src/live-compatible.test.ts new file mode 100644 index 0000000..fd18669 --- /dev/null +++ b/server/src/live-compatible.test.ts @@ -0,0 +1,109 @@ +import { afterEach, expect, it, vi } from 'vitest'; +import { EventEmitter } from 'node:events'; +import express from 'express'; +import { createServer, type Server } from 'node:http'; +import { mkdtemp, readdir, rm, writeFile } from 'node:fs/promises'; +import path from 'node:path'; +import { tmpdir } from 'node:os'; +const mocks = vi.hoisted(() => ({ spawn: vi.fn() })); +vi.mock('node:child_process', () => ({ spawn: mocks.spawn })); +import { createLiveBuffer, createLiveRouter } from './live-buffer.js'; + +let root: string; +let server: Server; +const buffers: ReturnType[] = []; +const source = 'http://127.0.0.1:1/api/stream/channel-a?subs=1'; +const auth = { headers: { authorization: 'Bearer synthetic-compatible-test' } }; +afterEach(async () => { + await Promise.all(buffers.splice(0).map(buffer => buffer.stop())); + if (server) await new Promise(resolve => { server.close(() => resolve()); server.closeAllConnections(); }); + if (root) await rm(root, { recursive: true, force: true }); + delete process.env.STREAMVAULT_AUTH_TOKEN; +}); +it('serves native-compatible signed HLS with independent tickets, cache epochs and primary unsafe state', async () => { + process.env.STREAMVAULT_AUTH_TOKEN = 'synthetic-compatible-test'; + mocks.spawn.mockImplementation(() => { + const worker = Object.assign(new EventEmitter(), { exitCode: null as number | null, signalCode: null, + kill: vi.fn(() => { queueMicrotask(() => { worker.exitCode = 0; worker.emit('close', 0); }); return true; }) }); + return worker; + }); + root = await mkdtemp(path.join(process.env.TMPDIR || tmpdir(), 'sv-compatible-')); + const primary = createLiveBuffer({ root: path.join(root, 'primary'), packetAware: true, pollMs: 10 }); + const compatible = createLiveBuffer({ root: path.join(root, 'compatible'), packetAware: false, maxChannels: 4, pollMs: 10 }); + buffers.push(primary, compatible); + async function seed(buffer: typeof primary, dir: string) { + await buffer.playlist('channel-a', source); + let stage = ''; + await vi.waitFor(async () => { + const [channel] = await readdir(dir); + const ingest = (await readdir(path.join(dir, channel))).find(name => name.startsWith('ingest-')); + expect(ingest).toBeTruthy(); + stage = path.join(dir, channel, ingest!); + }); + await writeFile(path.join(stage, '0.ts'), Buffer.alloc(188)); + await writeFile(path.join(stage, 'index.m3u8'), '#EXTM3U\n#EXTINF:8.334,\n0.ts\n'); + await vi.waitFor(async () => expect(await buffer.playlist('channel-a', source)).toContain('segment/')); + return stage; + } + const primaryStage = await seed(primary, path.join(root, 'primary')); + await seed(compatible, path.join(root, 'compatible')); + expect(mocks.spawn.mock.calls.map(([command]) => command)).toContain('ffmpeg'); + const app = express(); + const getSource = (id: string) => /^channel-[a-e]$/.test(id) ? source : null; + app.use('/api/live', createLiveRouter(primary, getSource)); + app.use('/api/live-compatible', createLiveRouter(compatible, getSource, { basePath: '/api/live-compatible' })); + server = createServer(app); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + if (!address || typeof address === 'string') throw Error('No listener'); + const origin = `http://127.0.0.1:${address.port}`; + const primaryAuth = await fetch(`${origin}/api/live/channel-a/authorize`, auth); + const compatibleAuth = await fetch(`${origin}/api/live-compatible/channel-a/authorize`, auth); + expect(primaryAuth.status).toBe(200); + expect(compatibleAuth.status).toBe(200); + expect(compatibleAuth.headers.get('cache-control')).toBe('private, no-store'); + const primaryUrl = new URL((await primaryAuth.json()).playlistUrl, origin); + const compatibleUrl = new URL((await compatibleAuth.json()).playlistUrl, origin); + expect(compatibleUrl.pathname).toBe('/api/live-compatible/channel-a/index.m3u8'); + const primaryManifest = await (await fetch(primaryUrl)).text(); + const response = await fetch(compatibleUrl); + expect(response.status).toBe(200); + expect(response.headers.get('content-type')).toMatch(/application\/vnd.apple.mpegurl/); + const manifest = await response.text(); + const segmentOf = (text: string, url: URL) => new URL(text.split('\n').find(line => line.startsWith('segment/'))!, url); + const primarySegment = segmentOf(primaryManifest, primaryUrl); + const compatibleSegment = segmentOf(manifest, compatibleUrl); + expect(compatibleSegment.searchParams.get('epoch')).not.toBe(primarySegment.searchParams.get('epoch')); + expect((await fetch(compatibleSegment)).status).toBe(200); + for (const [url, other] of [[compatibleUrl, primaryUrl], [primaryUrl, compatibleUrl]]) { + const wrongTicket = new URL(url); + wrongTicket.searchParams.set('ticket', other.searchParams.get('ticket')!); + expect((await fetch(wrongTicket)).status).toBe(401); + } + expect((await fetch(`${origin}/api/live-compatible/channel-a/authorize`)).status).toBe(401); + expect((await fetch(`${origin}/api/live-compatible/missing/authorize`, auth)).status).toBe(404); + expect((await fetch(`${origin}/api/live-compatible/channel-a/index.m3u8?audio=1`, auth)).status).toBe(422); + const wrongEpoch = new URL(compatibleSegment); + wrongEpoch.searchParams.set('epoch', primarySegment.searchParams.get('epoch')!); + expect((await fetch(wrongEpoch)).status).toBe(404); + await writeFile(path.join(primaryStage, 'UNSAFE'), 'synthetic-marker'); + await vi.waitFor(() => expect(primary.isUnsafe('channel-a')).toBe(true)); + expect((await fetch(primaryUrl)).status).toBe(501); + expect((await fetch(primarySegment)).status).toBe(404); + expect((await fetch(`${origin}/api/live/channel-a/authorize`, auth)).status).toBe(501); + expect((await fetch(compatibleUrl)).status).toBe(200); + expect((await fetch(compatibleSegment)).status).toBe(200); + expect(compatible.isUnsafe('channel-a')).toBe(false); + const cold = await fetch(`${origin}/api/live-compatible/channel-b/index.m3u8`, auth); + expect(cold.status).toBe(503); + expect(cold.headers.get('retry-after')).toBe('2'); + await compatible.playlist('channel-c', source); + await compatible.playlist('channel-d', source); + expect(compatible.activeCount).toBe(4); + const capped = await fetch(`${origin}/api/live-compatible/channel-e/index.m3u8`, auth); + expect(capped.status).toBe(503); + expect(compatible.activeCount).toBe(4); + await Promise.all([primary.stop(), compatible.stop()]); + expect(primary.activeCount).toBe(0); + expect(compatible.activeCount).toBe(0); +}); diff --git a/server/src/live-packet.integration.test.ts b/server/src/live-packet.integration.test.ts index e3d7dfd..3db9c3d 100644 --- a/server/src/live-packet.integration.test.ts +++ b/server/src/live-packet.integration.test.ts @@ -650,7 +650,11 @@ with av.open(sys.argv[1]) as src, av.open(sys.argv[2], 'w', format='mpegts') as const firstSegment = manifest.split('\n').find(line => line.startsWith('segment/'))!; expect((await fetch(new URL(firstSegment, playlist))).status).toBe(200); release(); - await waitFor(async () => buffer.activeCount === 0 ? true : null, 10000); + // A silent body now has a bounded ten-second socket read, while the owner + // independently enforces fifteen seconds of source silence. Allow that + // existing owner deadline plus inspection/retirement margin, not forty-five + // seconds of publication grace, for this deliberately silent successor. + await waitFor(async () => buffer.activeCount === 0 ? true : null, failure === 'silent-source' ? 18000 : 10000); expect((await fetch(new URL(firstSegment, playlist))).status).toBe(404); const rejected = await fetch(playlist); expect(rejected.status).toBe(501); diff --git a/server/src/live_packet_worker.py b/server/src/live_packet_worker.py index 7b9178d..e7c4258 100644 --- a/server/src/live_packet_worker.py +++ b/server/src/live_packet_worker.py @@ -17,6 +17,10 @@ import av from live_packet_stitch import Stitcher, UnsafeSeam, ts_packets +# Long-GOP sources can deliver media in bursts. Keep socket waits below the +# owner's unchanged 15-second source-silence watchdog, not below a normal GOP. +DEFAULT_SOURCE_TIMEOUT_SECONDS = 10. + class InitialSourceUnavailable(Exception): """No source media was accepted; a later authorization may try again.""" @@ -53,7 +57,7 @@ def read(self, size): class Worker: def __init__(self, source: str, directory: Path, *, session_bytes=64 * 1024 * 1024, - disk_bytes=64 * 1024 * 1024, source_timeout=3., idle_seconds=180.): + disk_bytes=64 * 1024 * 1024, source_timeout=DEFAULT_SOURCE_TIMEOUT_SECONDS, idle_seconds=180.): parts = urlsplit(source) if (parts.scheme != 'http' or parts.hostname != '127.0.0.1' or not parts.port or parts.username or parts.password or parts.fragment or not parts.path.startswith('/api/stream/') @@ -127,7 +131,10 @@ def open_continuation_source(self): try: return self.open_source() except InitialSourceUnavailable: - if attempt + 1 == attempts or self.stop.wait(.25): + # A provider can reject immediate reopen while releasing the + # previous connection. Keep three attempts, but leave one and + # two seconds for that cooldown; never refresh media liveness. + if attempt + 1 == attempts or self.stop.wait(float(attempt + 1)): raise def monitor(self): @@ -184,7 +191,11 @@ def run(self): while not self.stop.is_set(): connection, body, length = self.open_continuation_source() try: - with av.open(body, 'r', format='mpegts') as demux: + # Discover supported TS tracks within bounded media + # time/bytes; a valid initial burst followed by silence + # must not wait for the longer socket timeout to demux. + with av.open(body, 'r', format='mpegts', + options={'analyzeduration': '1000000', 'probesize': '1048576'}) as demux: packets = ts_packets(demux) first = next(packets) self.received_media = True @@ -262,7 +273,7 @@ def main(): parser.add_argument('--directory', type=Path, required=True) parser.add_argument('--disk-bytes', type=int, required=True) parser.add_argument('--session-bytes', type=int, default=64 * 1024 * 1024) - parser.add_argument('--source-timeout', type=float, default=3.) + parser.add_argument('--source-timeout', type=float, default=DEFAULT_SOURCE_TIMEOUT_SECONDS) args = parser.parse_args() try: worker = Worker(args.source, args.directory, disk_bytes=args.disk_bytes, diff --git a/server/src/test_live_packet_worker.py b/server/src/test_live_packet_worker.py index 221868f..1493103 100644 --- a/server/src/test_live_packet_worker.py +++ b/server/src/test_live_packet_worker.py @@ -8,6 +8,7 @@ import time import unittest from unittest.mock import patch +import av from live_packet_worker import Body, InitialSourceUnavailable, Worker from live_packet_stitch import UnsafeSeam @@ -112,6 +113,28 @@ def test_transient_reopen_keeps_the_same_worker_and_returns_the_provable_body(se self.assertTrue(worker.received_media) self.assertFalse(worker.failure) + def test_reopen_waits_out_a_three_second_provider_close_cooldown(self): + with tempfile.TemporaryDirectory() as tmp: + worker = self.worker(Path(tmp)) + result = (object(), object(), None) + elapsed = 0. + def unavailable_until_cooldown(): + if elapsed < 3.: + raise InitialSourceUnavailable() + return result + def cancellable_wait(delay): + nonlocal elapsed + elapsed += delay + return False + with patch.object(worker, 'open_source', side_effect=unavailable_until_cooldown) as opened, \ + patch.object(worker.stop, 'wait', side_effect=cancellable_wait) as waited, \ + patch.object(worker, 'note_progress') as progress: + self.assertIs(worker.open_continuation_source(), result) + self.assertEqual(opened.call_count, 3) + self.assertEqual([call.args[0] for call in waited.call_args_list], [1., 2.]) + progress.assert_not_called() + self.assertFalse(worker.failure) + def test_reopen_budget_exhausts_after_three_attempts(self): with tempfile.TemporaryDirectory() as tmp: worker = self.worker(Path(tmp)) @@ -183,6 +206,54 @@ def assert_classification(self, response, length, marker, *, received_media=Fals self.assertEqual({item.name for item in stage.iterdir()}, {marker}, worker.failure_type) return worker, body + def test_default_read_timeout_survives_a_three_and_half_second_media_gap(self): + data = self.media['libx264'] + class Source(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + def do_GET(self): + self.send_response(200) + self.send_header('Content-Type', 'video/mp2t') + self.send_header('Content-Length', str(len(data))) + self.end_headers() + midpoint = len(data) // 2 + try: + self.wfile.write(data[:midpoint]) + self.wfile.flush() + time.sleep(3.5) + self.wfile.write(data[midpoint:]) + except (BrokenPipeError, ConnectionResetError): + pass + server = ThreadingHTTPServer(('127.0.0.1', 0), Source) + serve = threading.Thread(target=server.serve_forever, daemon=True) + serve.start() + try: + with tempfile.TemporaryDirectory() as tmp: + worker = Worker(f'http://127.0.0.1:{server.server_port}/api/stream/channel-a', Path(tmp)) + connection, body, length = worker.open_source() + try: + received = bytearray() + while chunk := body.read(16 * 1024): + received.extend(chunk) + self.assertFalse(body.failed) + self.assertEqual(bytes(received), data) + self.assertEqual(length, len(data)) + self.assertLess(worker.source_timeout, 15.) + finally: + connection.close() + finally: + server.shutdown() + server.server_close() + serve.join(timeout=2) + + def test_bounded_input_discovery_still_validates_real_media_and_truncation(self): + data = self.media['libx264'] + with patch('live_packet_worker.av.open', wraps=av.open) as opened: + worker, _ = self.assert_classification(ScriptedResponse(data), len(data) + 188, 'UNSAFE') + self.assertTrue(worker.received_media) + demux_call = next(call for call in opened.call_args_list if call.args[1] == 'r') + self.assertEqual(demux_call.kwargs['options'], {'analyzeduration': '1000000', 'probesize': '1048576'}) + def test_initial_body_timeout_without_media_is_retryable(self): worker, body = self.assert_classification(ScriptedResponse(error=TimeoutError()), None, 'RETRYABLE') self.assertTrue(body.failed) diff --git a/src/hooks/usePlayer.liveHls.test.tsx b/src/hooks/usePlayer.liveHls.test.tsx index 4706e9e..4f36000 100644 --- a/src/hooks/usePlayer.liveHls.test.tsx +++ b/src/hooks/usePlayer.liveHls.test.tsx @@ -3,6 +3,8 @@ import { createRoot, type Root } from 'react-dom/client'; import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { usePlayer } from './usePlayer'; import { usePlayerStore } from '../stores/playerStore'; +import type { getAuthorizedLiveHlsUrl } from '../services/livePlayback'; +import { ApiError, rotateBackendRequestScope } from '../services/api'; const mocks = vi.hoisted(() => { // The module-level recovery singleton captures Date.now in its constructor. @@ -12,7 +14,7 @@ const mocks = vi.hoisted(() => { hlsSupported: vi.fn(() => true), mpegtsPlayer: { on: vi.fn(), destroy: vi.fn(), load: vi.fn(), attachMediaElement: vi.fn((video: HTMLVideoElement) => { video.src = 'blob:legacy-ts-test'; }) }, - authorize: vi.fn(async () => '/api/live/live_test/index.m3u8?ticket=test-only'), + authorize: vi.fn(async () => '/api/live/live_test/index.m3u8?ticket=test-only'), instances: [] as Array<{ on: ReturnType; destroy: ReturnType }>, }; }); @@ -54,6 +56,7 @@ describe('signed live MSE startup lead', () => { vi.clearAllMocks(); mocks.instances.length = 0; mocks.hlsSupported.mockReturnValue(true); + mocks.authorize.mockReset(); mocks.authorize.mockResolvedValue('/api/live/live_test/index.m3u8?ticket=test-only'); localStorage.clear(); vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('{}', { status: 404 })); @@ -154,6 +157,265 @@ describe('signed live MSE startup lead', () => { }, ); + it.each(['iPhone', 'Safari'])('uses compatible signed native HLS after primary unavailability on %s', async platform => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue(platform === 'iPhone' + ? 'iPhone AppleWebKit/605.1.15 Safari/604.1' : 'Macintosh AppleWebKit/605.1.15 Safari/605.1.15'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValueOnce('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await ready(); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(mocks.authorize).toHaveBeenLastCalledWith(expect.any(String), 'live_test', window.location.origin, + expect.objectContaining({ delivery: 'compatible', isCurrent: expect.any(Function) })); + expect(new URL(video.src).pathname).toBe('/api/live-compatible/live_test/index.m3u8'); + expect(video.play).toHaveBeenCalledOnce(); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(mocks.instances).toHaveLength(0); + }); + + it('uses compatible HLS on a normal replacement play after native primary failure', async () => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + await ready(); + await act(async () => video.dispatchEvent(new Event('error'))); + mocks.authorize.mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await ready(); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(mocks.authorize).toHaveBeenLastCalledWith(expect.any(String), 'live_test', window.location.origin, + expect.objectContaining({ delivery: 'compatible' })); + expect(new URL(video.src).pathname).toBe('/api/live-compatible/live_test/index.m3u8'); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + }); + + it.each(['primary unavailable', 'primary watchdog'])( + 'stops compatible startup without verified media progress after %s', async primaryFailure => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + ranges = []; + if (primaryFailure === 'primary unavailable') mocks.authorize.mockResolvedValueOnce(null); + else mocks.authorize.mockResolvedValueOnce('/api/live/live_test/index.m3u8?ticket=test-only'); + mocks.authorize.mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await act(async () => hookRef.current?.play()); + if (primaryFailure === 'primary watchdog') { + await act(async () => vi.advanceTimersByTimeAsync(30_250)); + } + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(new URL(video.src).pathname).toBe('/api/live-compatible/live_test/index.m3u8'); + // Neither a zero-time clock event nor ready/play events prove media progress. + video.currentTime = 0; + await act(async () => video.dispatchEvent(new Event('timeupdate'))); + await act(async () => vi.advanceTimersByTimeAsync(29_999)); + expect(usePlayerStore.getState().status).toBe('loading'); + // Recovery preserves the primary failure's backoff (1s versus initial 250ms). + await act(async () => vi.advanceTimersByTimeAsync(primaryFailure === 'primary watchdog' ? 1_001 : 251)); + expect(usePlayerStore.getState().status).toBe('error'); + expect(usePlayerStore.getState().errorMessage).toMatch(/compatible.*connection.*retry/i); + await act(async () => { + video.dispatchEvent(new Event('error')); + video.dispatchEvent(new Event('ended')); + video.dispatchEvent(new Event('waiting')); + video.dispatchEvent(new Event('stalled')); + await vi.advanceTimersByTimeAsync(169_250); + }); + expect(usePlayerStore.getState().status).toBe('error'); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(mocks.instances).toHaveLength(0); + // Explicit retry starts a fresh attempt and can still become healthy. + await act(async () => hookRef.current?.retry()); + video.currentTime = 1; + await act(async () => { + video.dispatchEvent(new Event('loadeddata')); + video.dispatchEvent(new Event('canplay')); + video.dispatchEvent(new Event('timeupdate')); + }); + expect(mocks.authorize).toHaveBeenCalledTimes(3); + expect(usePlayerStore.getState().status).toBe('playing'); + }, + ); + + it('does not treat compatible readiness or a resolved play promise as media progress', async () => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await ready(); + expect(usePlayerStore.getState().status).toBe('playing'); + await act(async () => vi.advanceTimersByTimeAsync(30_250)); + expect(usePlayerStore.getState().status).toBe('error'); + expect(usePlayerStore.getState().errorMessage).toMatch(/compatible.*retry/i); + await act(async () => vi.advanceTimersByTimeAsync(65_000)); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + }); + + it('recovers later interruptions after compatible native media genuinely advances', async () => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await ready(); + await act(async () => video.dispatchEvent(new Event('timeupdate'))); + await act(async () => vi.advanceTimersByTimeAsync(12_250)); + expect(mocks.authorize).toHaveBeenCalledTimes(3); + expect(mocks.authorize).toHaveBeenLastCalledWith(expect.any(String), 'live_test', window.location.origin, + expect.objectContaining({ delivery: 'compatible' })); + expect(usePlayerStore.getState().status).toBe('loading'); + video.currentTime = 41; + await act(async () => { + video.dispatchEvent(new Event('loadeddata')); + video.dispatchEvent(new Event('canplay')); + video.dispatchEvent(new Event('timeupdate')); + await vi.advanceTimersByTimeAsync(12_250); + }); + expect(mocks.authorize).toHaveBeenCalledTimes(4); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(mocks.instances).toHaveLength(0); + }); + + it('retains playing status when a compatible recovery watchdog fires after backend invalidation', async () => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await ready(); + expect(usePlayerStore.getState().status).toBe('playing'); + const src = video.getAttribute('src'); + const pauses = vi.mocked(video.pause).mock.calls.length; + await act(async () => rotateBackendRequestScope()); + await act(async () => vi.advanceTimersByTimeAsync(65_000)); + expect(usePlayerStore.getState().status).toBe('playing'); + expect(usePlayerStore.getState().errorMessage).toBe(''); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(video.play).toHaveBeenCalledOnce(); + expect(video.pause).toHaveBeenCalledTimes(pauses); + expect(video.getAttribute('src')).toBe(src); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(mocks.instances).toHaveLength(0); + }); + + it('retains playing status when a compatible native media error fires after backend invalidation', async () => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await ready(); + expect(usePlayerStore.getState().status).toBe('playing'); + const src = video.getAttribute('src'); + const pauses = vi.mocked(video.pause).mock.calls.length; + await act(async () => rotateBackendRequestScope()); + await act(async () => video.dispatchEvent(new Event('error'))); + expect(usePlayerStore.getState().status).toBe('playing'); + expect(usePlayerStore.getState().errorMessage).toBe(''); + await act(async () => vi.advanceTimersByTimeAsync(65_000)); + expect(usePlayerStore.getState().status).toBe('playing'); + expect(usePlayerStore.getState().errorMessage).toBe(''); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(video.play).toHaveBeenCalledOnce(); + expect(video.pause).toHaveBeenCalledTimes(pauses); + expect(video.getAttribute('src')).toBe(src); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(mocks.instances).toHaveLength(0); + }); + + it.each(['stop', 'channel switch', 'backend switch'])( + 'ignores the compatible first-frame watchdog after %s', async cancellation => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValue('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await act(async () => hookRef.current?.play()); + await act(async () => { + if (cancellation === 'stop') hookRef.current?.stop(); + else if (cancellation === 'channel switch') usePlayerStore.setState({ currentChannel: { + ...usePlayerStore.getState().currentChannel!, id: 'live_next', + } }); + else rotateBackendRequestScope(); + }); + const status = usePlayerStore.getState().status; + await act(async () => vi.advanceTimersByTimeAsync(65_000)); + expect(usePlayerStore.getState().status).toBe(status); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + }, + ); + + it.each(['attachment', 'media'])('never sends a native Apple client to raw TS when compatible HLS %s fails', async failure => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + if (failure === 'attachment') vi.spyOn(video, 'load').mockImplementation(() => { + if (video.getAttribute('src')?.includes('/api/live-compatible/')) throw new Error('Native attachment failed'); + }); + mocks.authorize.mockResolvedValueOnce(null) + .mockResolvedValueOnce('/api/live-compatible/live_test/index.m3u8?ticket=compatible-test'); + await act(async () => hookRef.current?.play()); + if (failure === 'media') await act(async () => video.dispatchEvent(new Event('error'))); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(usePlayerStore.getState().status).toBe('error'); + expect(usePlayerStore.getState().errorMessage).toMatch(/compatible.*retry/i); + await act(async () => vi.advanceTimersByTimeAsync(65_000)); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + }); + + it.each([401, 403])('never tries compatible HLS or raw TS after primary auth %s', async status => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockRejectedValue(new ApiError(status, 'Synthetic rejection')); + await act(async () => hookRef.current?.play()); + expect(mocks.authorize).toHaveBeenCalledOnce(); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(usePlayerStore.getState().status).toBe('error'); + await act(async () => vi.advanceTimersByTimeAsync(65_000)); + expect(mocks.authorize).toHaveBeenCalledOnce(); + }); + + it('shows a bounded actionable error when compatible authorization remains unavailable', async () => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + mocks.authorize.mockResolvedValueOnce(null).mockImplementationOnce(() => + new Promise(resolve => setTimeout(() => resolve(null), 50_000))); + await act(async () => hookRef.current?.play()); + expect(usePlayerStore.getState().status).toBe('loading'); + await act(async () => vi.advanceTimersByTimeAsync(50_000)); + expect(usePlayerStore.getState().status).toBe('error'); + expect(usePlayerStore.getState().errorMessage).toMatch(/compatible.*connection.*retry/i); + await act(async () => { + video.dispatchEvent(new Event('error')); + video.dispatchEvent(new Event('waiting')); + await vi.advanceTimersByTimeAsync(65_000); + }); + expect(mocks.authorize).toHaveBeenCalledTimes(2); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + expect(video.getAttribute('src')).toBeNull(); + }); + + it.each(['stop', 'channel switch', 'backend switch', 'replacement play'])( + 'ignores a compatible authorization arriving after %s', async cancellation => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue('iPhone AppleWebKit/605.1.15 Safari/604.1'); + vi.mocked(video.canPlayType).mockReturnValue('maybe'); + let resolveCompatible!: (url: string) => void; + mocks.authorize.mockResolvedValueOnce(null).mockImplementationOnce(() => + new Promise(resolve => { resolveCompatible = resolve; })); + await act(async () => hookRef.current?.play()); + const isCurrent = mocks.authorize.mock.calls[1][3]!.isCurrent!; + await act(async () => { + if (cancellation === 'stop') hookRef.current?.stop(); + else if (cancellation === 'channel switch') usePlayerStore.setState({ currentChannel: { + ...usePlayerStore.getState().currentChannel!, id: 'live_next', + } }); + else if (cancellation === 'backend switch') rotateBackendRequestScope(); + else hookRef.current?.play(); + }); + expect(isCurrent()).toBe(false); + const status = usePlayerStore.getState().status; + const src = video.getAttribute('src'); + await act(async () => resolveCompatible('/api/live-compatible/live_test/index.m3u8?ticket=obsolete')); + expect(video.getAttribute('src')).toBe(src); + expect(usePlayerStore.getState().status).toBe(status); + expect(mocks.mpegtsPlayer.load).not.toHaveBeenCalled(); + }, + ); + it('does not gate a legacy TS fallback even when live authorization succeeded', async () => { mocks.hlsSupported.mockReturnValue(false); await act(async () => { @@ -255,8 +517,8 @@ describe('signed live MSE startup lead', () => { appended?.(); canplay.call(video, new Event('canplay')); progress?.call(video, new ProgressEvent('progress')); - // Cross both watchdog + retry deadlines, still inside authorization's 25s budget. - await vi.advanceTimersByTimeAsync(24_000); + // Cross both watchdog + retry deadlines, still inside authorization's 50s budget. + await vi.advanceTimersByTimeAsync(49_000); }); expect(video.play).not.toHaveBeenCalled(); if (resolveAuthorization) { diff --git a/src/hooks/usePlayer.ts b/src/hooks/usePlayer.ts index d8a3bc5..804447e 100644 --- a/src/hooks/usePlayer.ts +++ b/src/hooks/usePlayer.ts @@ -35,8 +35,9 @@ import { } from '../services/playbackSeek'; import { useRecordingStore } from '../stores/recordingStore'; import { attachFiniteHls } from '../services/finiteHls'; -import { attachLiveHls } from '../services/liveHls'; +import { attachLiveHls, prefersNativeLiveHls } from '../services/liveHls'; import { getAuthorizedLiveHlsUrl } from '../services/livePlayback'; +import { getBackendRequestScope } from '../services/api'; import { getAvPlayClockReading, getHtml5ClockReading, @@ -219,7 +220,6 @@ function startBrowserSubtitleTrack( const liveStreamRecovery = new LiveStreamRecovery((reason, attempt) => { log.warn(`Live stream: ${reason} — reconnecting (attempt ${attempt})`); - usePlayerStore.getState().setStatus('loading'); restartActiveLiveStream?.(); }); @@ -492,6 +492,7 @@ export function usePlayer(): { // browser live playback still carries captions so it can expose only tracks // that the media element actually detects, with Off enforced by track mode. const keepSubsRef = useRef(getSubtitlesEnabled()); + const nativePrimaryFailureRef = useRef<{ channelId: string; backendGeneration: number } | null>(null); const play = useCallback(function play(finiteRetryPosition?: number) { const channel = usePlayerStore.getState().currentChannel; @@ -524,6 +525,25 @@ export function usePlayer(): { const clockGeneration = playbackClock.begin(channel.duration ?? 0); const commercialGeneration = beginCommercialPlayback(channel); const authorizationGeneration = ++liveAuthorizationGeneration; + const backendGeneration = getBackendRequestScope().generation; + const isCurrentAuthorization = () => authorizationGeneration === liveAuthorizationGeneration && + backendGeneration === getBackendRequestScope().generation && + usePlayerStore.getState().currentChannel?.id === channel.id; + const nativeVideo = document.getElementById('av-player') as HTMLVideoElement | null; + const nativeLiveHls = channel.contentType === 'livetv' && !audioOnly && !channel.dvrHls && + !(typeof webapis !== 'undefined' && webapis.avplay) && !!nativeVideo && prefersNativeLiveHls(nativeVideo); + if (nativePrimaryFailureRef.current?.channelId !== channel.id || + nativePrimaryFailureRef.current?.backendGeneration !== backendGeneration) nativePrimaryFailureRef.current = null; + let compatibleLiveHls = nativeLiveHls && !!nativePrimaryFailureRef.current; + const noteNativePrimaryFailure = () => { + if (nativeLiveHls && !compatibleLiveHls && isCurrentAuthorization()) { + nativePrimaryFailureRef.current = { channelId: channel.id, backendGeneration }; + } + }; + const failCompatibleLiveHls = () => { + disableLiveStreamRecovery(); + setError('Compatible Live TV is unavailable. Check the server connection and tap to retry.'); + }; const startPlayback = (authorizedLiveUrl: string | null) => { if (authorizationGeneration !== liveAuthorizationGeneration || usePlayerStore.getState().currentChannel?.id !== channel.id || @@ -788,10 +808,21 @@ export function usePlayer(): { video.dataset.channelId = channel.id; const playbackGeneration = ++html5PlaybackGeneration; const isCurrentPlayback = () => playbackGeneration === html5PlaybackGeneration && - (!authorizedLiveUrl || (authorizationGeneration === liveAuthorizationGeneration && - usePlayerStore.getState().currentChannel?.id === channel.id)); + (!authorizedLiveUrl || isCurrentAuthorization()); + let hasLiveMediaProgress = false; if (isLiveTs) { - restartActiveLiveStream = play; + restartActiveLiveStream = () => { + if (!isCurrentPlayback() || !isCurrentAuthorization()) return; + setStatus('loading'); + // A successful authorization/attachment is not a decoded first frame. + // Stop the compatible startup watchdog instead of reauthorizing forever. + if (nativeLiveHls && compatibleLiveHls && !hasLiveMediaProgress) { + failCompatibleLiveHls(); + return; + } + noteNativePrimaryFailure(); + play(); + }; liveStreamRecovery.begin(channel.id); } else { restartActiveLiveStream = null; @@ -1023,6 +1054,7 @@ export function usePlayer(): { } if (!isLiveTs || !isCurrentPlayback() || video.currentTime <= lastMediaTime) return; lastMediaTime = video.currentTime; + hasLiveMediaProgress = true; liveStreamRecovery.progress(); recoverDrainedLiveStream(); }; @@ -1044,6 +1076,8 @@ export function usePlayer(): { return; } if (isLiveTs && isCurrentPlayback()) { + if (nativeLiveHls && compatibleLiveHls) { failCompatibleLiveHls(); return; } + noteNativePrimaryFailure(); setStatus('loading'); liveStreamRecovery.transportEnded('mpegts-error'); return; @@ -1056,6 +1090,7 @@ export function usePlayer(): { log.info('HTML5 event: ended'); if (isLiveTs && isCurrentPlayback()) { if (pendingLiveEof) { recoverDrainedLiveStream(true); return; } + noteNativePrimaryFailure(); setStatus('loading'); liveStreamRecovery.transportEnded('media-ended'); return; @@ -1223,6 +1258,7 @@ export function usePlayer(): { video.dataset.streamOffset = '0'; void attachLiveHls(video, playUrl, () => { if (!isCurrentPlayback()) return; + noteNativePrimaryFailure(); setStatus('loading'); liveStreamRecovery.transportEnded('hls-error'); }, isCurrentPlayback, undefined, attemptPlay).then(dispose => { @@ -1230,6 +1266,14 @@ export function usePlayer(): { disposeFiniteHls = dispose; }).catch(() => { if (!isCurrentPlayback()) return; + if (nativeLiveHls) { + if (compatibleLiveHls) failCompatibleLiveHls(); + else { + noteNativePrimaryFailure(); + liveStreamRecovery.transportEnded('hls-error'); + } + return; + } // An engine without native HLS or MSE support keeps the old TS path. log.warn('HTML5: live HLS unavailable; falling back to MPEG-TS'); startMpegTsPlayback(); @@ -1325,14 +1369,29 @@ export function usePlayer(): { // watchdog/queued retry without resetting backoff; begin() rearms recovery // only after this authorization succeeds and attaches its player. liveStreamRecovery.suspend(); - void getAuthorizedLiveHlsUrl(useChannelStore.getState().apiBaseUrl, channel.id, window.location.origin, { - isCurrent: () => authorizationGeneration === liveAuthorizationGeneration && - usePlayerStore.getState().currentChannel?.id === channel.id, + const apiBaseUrl = useChannelStore.getState().apiBaseUrl; + void getAuthorizedLiveHlsUrl(apiBaseUrl, channel.id, window.location.origin, { + isCurrent: isCurrentAuthorization, + ...(compatibleLiveHls ? { delivery: 'compatible' as const } : {}), }) - .then(startPlayback) + .then(async url => { + if (!isCurrentAuthorization()) return; + if (!url && nativeLiveHls && !compatibleLiveHls) { + noteNativePrimaryFailure(); + compatibleLiveHls = true; + url = await getAuthorizedLiveHlsUrl(apiBaseUrl, channel.id, window.location.origin, { + delivery: 'compatible', isCurrent: isCurrentAuthorization, + }); + if (!isCurrentAuthorization()) return; + } + if (!url && nativeLiveHls) { + failCompatibleLiveHls(); + return; + } + startPlayback(url); + }) .catch(() => { - if (authorizationGeneration !== liveAuthorizationGeneration || - usePlayerStore.getState().currentChannel?.id !== channel.id) return; + if (!isCurrentAuthorization()) return; setError('Live playback could not be authorized. Check the server connection.'); }); } else { @@ -1343,6 +1402,7 @@ export function usePlayer(): { }, []); const stop = useCallback(() => { + nativePrimaryFailureRef.current = null; playerRef.current = null; subtitleTracksRef.current = []; selectedSubtitleIndexRef.current = -1; diff --git a/src/services/liveHls.ts b/src/services/liveHls.ts index 4832f16..d9dd1cb 100644 --- a/src/services/liveHls.ts +++ b/src/services/liveHls.ts @@ -1,6 +1,12 @@ import type Hls from 'hls.js'; import { isAppleMobile } from '../utils/platform'; +/** WebKit's native path, shared with live authorization transport selection. */ +export function prefersNativeLiveHls(video: HTMLVideoElement): boolean { + const safari = /Safari\//.test(navigator.userAgent) && !/(Chrome|Chromium|Edg|OPR)\//.test(navigator.userAgent); + return !!video.canPlayType('application/vnd.apple.mpegurl') && (isAppleMobile() || safari); +} + /** Attach a rolling live HLS playlist without replacing the player at source EOF. */ export async function attachLiveHls( video: HTMLVideoElement, @@ -19,8 +25,7 @@ export async function attachLiveHls( }; // Chromium can advertise native HLS but keep too little lead to cover source // replay catch-up. Prefer the controlled MSE buffer there; retain WebKit HLS. - const safari = /Safari\//.test(navigator.userAgent) && !/(Chrome|Chromium|Edg|OPR)\//.test(navigator.userAgent); - if (nativeSupported && (isAppleMobile() || safari)) return nativeAttach(); + if (prefersNativeLiveHls(video)) return nativeAttach(); const { default: HlsPlayer } = await loadHls(); if (!isCurrent()) throw new Error('Live playback superseded'); diff --git a/src/services/livePlayback.test.ts b/src/services/livePlayback.test.ts index 0308ea7..4fed51a 100644 --- a/src/services/livePlayback.test.ts +++ b/src/services/livePlayback.test.ts @@ -79,7 +79,20 @@ describe('live HLS native-player authorization', () => { expect(fetchMock).toHaveBeenCalledTimes(2); }); - it('bounds a stalled cold retry including network time to 25 seconds', async () => { + it('keeps cold native HLS authorization alive through a forty-second first publication', async () => { + const fetchMock = vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(json({ error: 'Warming up' }, 503, '40')) + .mockResolvedValueOnce(json({ playlistUrl: '/api/live/live_future/index.m3u8?ticket=synthetic' })); + const result = getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test'); + await vi.advanceTimersByTimeAsync(39_999); + expect(fetchMock).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await expect(result).resolves.toContain('/api/live/live_future/index.m3u8'); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); + }); + + it('bounds a stalled cold retry including network time to 50 seconds', async () => { let retrySignal: AbortSignal | undefined; const fetchMock = vi.spyOn(globalThis, 'fetch') .mockResolvedValueOnce(json({ error: 'Warming up' }, 503, '3')) @@ -92,7 +105,7 @@ describe('live HLS native-player authorization', () => { .then(value => { settled = true; return value; }, error => { settled = true; return error; }); await vi.advanceTimersByTimeAsync(3000); expect(fetchMock).toHaveBeenCalledTimes(2); - await vi.advanceTimersByTimeAsync(21_999); + await vi.advanceTimersByTimeAsync(46_999); expect(settled).toBe(false); await vi.advanceTimersByTimeAsync(1); expect(settled).toBe(true); @@ -128,7 +141,7 @@ describe('live HLS native-player authorization', () => { expect(settled.mock.calls[0][0].error).toMatchObject({ status }); expect(text).not.toHaveBeenCalled(); expect(vi.getTimerCount()).toBe(0); - await vi.advanceTimersByTimeAsync(24_000); + await vi.advanceTimersByTimeAsync(49_000); await outcome; expect(settled).toHaveBeenCalledExactlyOnceWith({ error: expect.any(ApiError) }); for (const [, init] of fetchMock.mock.calls) { @@ -145,8 +158,8 @@ describe('live HLS native-player authorization', () => { expect(fetchMock).toHaveBeenCalledTimes(1); await vi.runAllTimersAsync(); await expect(result).resolves.toBeNull(); - expect(fetchMock.mock.calls.length).toBeLessThanOrEqual(25); - expect(Date.now() - start).toBeLessThanOrEqual(25_000); + expect(fetchMock.mock.calls.length).toBeLessThanOrEqual(50); + expect(Date.now() - start).toBeLessThanOrEqual(50_000); expect(vi.getTimerCount()).toBe(0); }); @@ -160,7 +173,7 @@ describe('live HLS native-player authorization', () => { it('does not treat an initial stalled transport as permission to downgrade authentication', async () => { vi.spyOn(globalThis, 'fetch').mockImplementation(() => new Promise(() => {})); const outcome = getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test').catch(error => error); - await vi.advanceTimersByTimeAsync(25_000); + await vi.advanceTimersByTimeAsync(50_000); expect(await outcome).toMatchObject({ name: 'TimeoutError' }); expect(vi.getTimerCount()).toBe(0); }); @@ -179,6 +192,80 @@ describe('live HLS native-player authorization', () => { expect(fetchMock).toHaveBeenCalledTimes(2); }); + it('can explicitly authorize compatibility after a real primary 501 response', async () => { + const fetchMock = vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(json({ error: 'Unsafe seam' }, 501)) + .mockResolvedValueOnce(json({ playlistUrl: '/api/live-compatible/live_future/index.m3u8?ticket=synthetic' })); + await expect(getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test')).resolves.toBeNull(); + await expect(getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test', { + delivery: 'compatible', + })).resolves.toContain('/api/live-compatible/live_future/index.m3u8'); + expect(fetchMock.mock.calls.map(([url]) => url)).toEqual([ + '/api/live/live_future/authorize', '/api/live-compatible/live_future/authorize', + ]); + }); + + it.each([401, 403])('never downgrades compatible authorization %s after 503', async status => { + const fetchMock = vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(json({}, 503, '1')).mockResolvedValueOnce(json({}, status)); + const outcome = getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test', { + delivery: 'compatible', + }).catch(error => error); + await vi.runAllTimersAsync(); + expect(await outcome).toMatchObject({ status }); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); + }); + + it('bounds compatible 503 retries by the same 50-second deadline', async () => { + const fetchMock = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => json({}, 503, '2')); + const started = Date.now(); + const outcome = getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test', { delivery: 'compatible' }); + await vi.runAllTimersAsync(); + await expect(outcome).resolves.toBeNull(); + expect(Date.now() - started).toBeLessThanOrEqual(50_000); + expect(fetchMock.mock.calls.length).toBeLessThanOrEqual(25); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each(['backend', 'player'])('cancels compatible retry on obsolete %s identity', async cancellation => { + let current = true; + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue(json({}, 503, '2')); + const outcome = getAuthorizedLiveHlsUrl('', 'live_future', 'https://app.example.test', { + delivery: 'compatible', isCurrent: () => current, + }).catch(error => error); + await vi.advanceTimersByTimeAsync(1); + if (cancellation === 'backend') rotateBackendRequestScope(); + else current = false; + await vi.runAllTimersAsync(); + expect(await outcome).toMatchObject({ name: cancellation === 'backend' ? 'StaleBackendRequestError' : 'AbortError' }); + expect(fetchMock).toHaveBeenCalledOnce(); + }); + + it.each([ + '/api/live/live_future/index.m3u8?ticket=synthetic', + '/api/live-compatible/another/index.m3u8?ticket=synthetic', + 'https://other.example.test/api/live-compatible/live_future/index.m3u8', + '//dvr.example.test/api/live-compatible/live_future/index.m3u8', + ])('strictly rejects unrelated compatible playlist %s', async playlistUrl => { + vi.spyOn(globalThis, 'fetch').mockResolvedValue(json({ playlistUrl })); + await expect(getAuthorizedLiveHlsUrl('https://dvr.example.test', 'live_future', 'https://app.example.test', { + delivery: 'compatible', + })).rejects.toThrow('unrelated playlist'); + }); + + it('explicitly authorizes compatible HLS with the same stored token and backend', async () => { + localStorage.setItem('streamvault_auth_token', JSON.stringify('test-secret')); + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue(json({ + playlistUrl: '/api/live-compatible/live_future/index.m3u8?ticket=synthetic', + })); + await expect(getAuthorizedLiveHlsUrl('https://dvr.example.test', 'live_future', 'https://app.example.test', { + delivery: 'compatible', + })).resolves.toBe('https://dvr.example.test/api/live-compatible/live_future/index.m3u8?ticket=synthetic'); + expect(fetchMock.mock.calls[0][0]).toBe('https://dvr.example.test/api/live-compatible/live_future/authorize'); + expect(new Headers(fetchMock.mock.calls[0][1]?.headers).get('x-streamvault-token')).toBe('test-secret'); + }); + it('fetches a signed feed with the stored token and keeps it on the configured backend', async () => { localStorage.setItem('streamvault_auth_token', JSON.stringify('test-secret')); const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue(json({ playlistUrl: '/api/live/live_future/index.m3u8?ticket=synthetic' })); diff --git a/src/services/livePlayback.ts b/src/services/livePlayback.ts index d15f3ff..8cc5b83 100644 --- a/src/services/livePlayback.ts +++ b/src/services/livePlayback.ts @@ -1,10 +1,15 @@ import { ApiError, apiFetch, getBackendRequestScope, StaleBackendRequestError } from './api'; import { resolveMediaUrl } from './recordingPlayback'; -const AUTHORIZATION_BUDGET_MS = 25_000; +// Allow the server's hard 45-second first-publication deadline to finish. +// The extra five seconds cover the bounded readiness request/response, not an +// extension of source/publication liveness or permission to ignore auth errors. +const AUTHORIZATION_BUDGET_MS = 50_000; let authorizationGeneration = 0; export interface LiveAuthorizationOptions { + /** Explicit native-HLS compatibility request; primary remains the default. */ + delivery?: 'primary' | 'compatible'; /** The player owns this generation; false cancels retries after stop/restart. */ isCurrent?: () => boolean; } @@ -14,6 +19,7 @@ export async function getAuthorizedLiveHlsUrl( apiBaseUrl: string, channelId: string, pageOrigin = window.location.origin, options: LiveAuthorizationOptions = {}, ): Promise { + const route = options.delivery === 'compatible' ? '/api/live-compatible' : '/api/live'; const generation = ++authorizationGeneration; const deadline = Date.now() + AUTHORIZATION_BUDGET_MS; const scope = getBackendRequestScope(); @@ -43,14 +49,14 @@ export async function getAuthorizedLiveHlsUrl( if (expired) throw timeoutError; try { const response = await apiFetch<{ playlistUrl: string }>( - apiBaseUrl, `/api/live/${encodeURIComponent(channelId)}/authorize`, { signal: controller.signal }, + apiBaseUrl, `${route}/${encodeURIComponent(channelId)}/authorize`, { signal: controller.signal }, ); assertCurrent(); if (expired) throw timeoutError; if (typeof response?.playlistUrl !== 'string' || !response.playlistUrl) { throw new Error('Live playback ticket did not include a URL'); } - const expected = new URL(`${apiBaseUrl}/api/live/${encodeURIComponent(channelId)}/index.m3u8`, pageOrigin); + const expected = new URL(`${apiBaseUrl}${route}/${encodeURIComponent(channelId)}/index.m3u8`, pageOrigin); const issued = new URL(response.playlistUrl, expected); if (!['http:', 'https:'].includes(issued.protocol) || issued.origin !== expected.origin || issued.pathname !== expected.pathname ||