From f6304c55ce77f105434833e43f848f3bff258d3f Mon Sep 17 00:00:00 2001 From: Christian Benincasa Date: Sun, 31 May 2026 08:17:42 -0400 Subject: [PATCH] fix: wait for master playlist to be present when requesting m3u8 --- server/src/api/streamApi.ts | 9 ++- server/src/services/TunarrWorkerPool.ts | 85 +------------------------ server/src/stream/Session.ts | 5 +- server/src/stream/hls/BaseHlsSession.ts | 13 ++-- server/src/stream/hls/HlsSession.ts | 25 ++++++-- server/src/types/events.ts | 25 ++++++++ server/src/util/future.ts | 81 +++++++++++++++++++++++ 7 files changed, 145 insertions(+), 98 deletions(-) create mode 100644 server/src/types/events.ts create mode 100644 server/src/util/future.ts diff --git a/server/src/api/streamApi.ts b/server/src/api/streamApi.ts index 643ff8ecc..9ac8a7449 100644 --- a/server/src/api/streamApi.ts +++ b/server/src/api/streamApi.ts @@ -391,7 +391,9 @@ export const streamApi: RouterPluginAsyncCallback = async (fastify) => { .then((result) => result.mapAsync(async (session) => { session.recordHeartbeat(req.ip); - const masterResult = await session.getMasterPlaylist(); + const masterResult = await session.getMasterPlaylist({ + wait: true, + }); if (masterResult.isFailure()) { logger.error(masterResult.error); @@ -409,8 +411,9 @@ export const streamApi: RouterPluginAsyncCallback = async (fastify) => { channelId, session.masterPlaylistPath, ); - logger.error(fmtError); - throw new Error(fmtError); + logger.info(fmtError); + return res.status(404).send(); + // throw new Error(fmtError); } return res diff --git a/server/src/services/TunarrWorkerPool.ts b/server/src/services/TunarrWorkerPool.ts index a4199fd2e..193c921a5 100644 --- a/server/src/services/TunarrWorkerPool.ts +++ b/server/src/services/TunarrWorkerPool.ts @@ -17,6 +17,7 @@ import { WorkerRequestToResponse, } from '../types/worker_schemas.ts'; import { getNumericEnvVar, WORKER_POOL_SIZE_ENV_VAR } from '../util/env.ts'; +import { Future } from '../util/future.ts'; import { timeoutPromise } from '../util/index.ts'; import { InjectLogger } from '../util/inject.ts'; import { Logger } from '../util/logging/LoggerFactory.ts'; @@ -35,88 +36,6 @@ interface PooledWorker { ready: boolean; } -class Future implements Promise { - #promise: Promise; - #resolve!: (v: T | PromiseLike) => void; - #reject!: (reason?: unknown) => void; - #state: 'pending' | 'fulfilled' | 'rejected' = 'pending'; - #value: T | undefined; - #err: unknown; - - constructor() { - this.#promise = new Promise((resolve, reject) => { - this.#resolve = resolve; - this.#reject = reject; - }); - - this.#promise.then( - (v) => { - this.#state = 'fulfilled'; - this.#value = v; - }, - (err) => { - this.#state = 'rejected'; - this.#err = err; - }, - ); - } - - [Symbol.toStringTag]!: string; - - resolve(value: T | PromiseLike) { - if (this.#state === 'pending') { - this.#resolve(value); - } else { - throw new Error( - 'Resolving already fulfilled future with state ' + this.#state, - ); - } - } - - reject(e: unknown) { - if (this.#state === 'pending') { - this.#reject(e); - } else { - throw new Error( - 'Rejecting already fulfilled future with state ' + this.#state, - ); - } - } - - then( - onfulfilled?: ((value: T) => TResult1 | PromiseLike) | null, - onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | null, - ): Promise { - return this.#promise.then(onfulfilled, onrejected); - } - - catch( - onrejected?: ((reason: unknown) => TResult | PromiseLike) | null, - ): Promise { - return this.#promise.catch(onrejected); - } - - finally(onfinally?: (() => void) | null): Promise { - return this.#promise.finally(onfinally); - } - - get promise() { - return this.#promise; - } - - get state() { - return this.#state; - } - - get value() { - return this.#value; - } - - get error() { - return this.#err; - } -} - type State = 'pending' | 'started' | 'terminating'; @injectable() @@ -129,7 +48,7 @@ export class TunarrWorkerPool implements IWorkerPool { #outstandingByIndex = new Map(); #startPromises: Promise[] = []; - @InjectLogger() private declare readonly logger: Logger; + @InjectLogger() declare private readonly logger: Logger; constructor( @inject(TunarrSubprocessService) diff --git a/server/src/stream/Session.ts b/server/src/stream/Session.ts index 578441d06..d0b220435 100644 --- a/server/src/stream/Session.ts +++ b/server/src/stream/Session.ts @@ -37,6 +37,9 @@ export type SessionType = ChannelStreamMode | ChannelConcatStreamMode; // TODO: sort these all out.... and write docs type StreamSessionEvents = { state: [newState: SessionState, oldState: SessionState]; + // Emitted when a stream is initializing + init: []; + // Emitted when a stream has started (wait for ready is complete) start: []; stop: []; cleanup: []; @@ -134,7 +137,7 @@ export abstract class Session< try { this.state = 'starting'; - this.emit('start'); + this.emit('init'); await this.startInternal(); await this.waitForStreamReadyInternal(); this.state = 'started'; diff --git a/server/src/stream/hls/BaseHlsSession.ts b/server/src/stream/hls/BaseHlsSession.ts index bfd9d2193..90d8b6d0b 100644 --- a/server/src/stream/hls/BaseHlsSession.ts +++ b/server/src/stream/hls/BaseHlsSession.ts @@ -25,7 +25,7 @@ export abstract class BaseHlsSession< // Working directory for m3u8 playlists and fragments protected _workingDirectory: string; // Absolute path to the stream directory - protected _m3u8PlaylistPath: string; + protected _variantPlaylistPath: string; // Absolute path to the HLS master playlist protected _masterPlaylistPath: string; // The path to request streaming assets from the server @@ -53,7 +53,10 @@ export abstract class BaseHlsSession< this.baseDirectory, `stream_${this.channel.uuid}`, ); - this._m3u8PlaylistPath = path.join(this._workingDirectory, 'stream.m3u8'); + this._variantPlaylistPath = path.join( + this._workingDirectory, + 'stream.m3u8', + ); this._masterPlaylistPath = path.join( this._workingDirectory, 'playlist.m3u8', @@ -76,7 +79,7 @@ export abstract class BaseHlsSession< } get streamPath() { - return this._m3u8PlaylistPath; + return this._variantPlaylistPath; } get serverPath() { @@ -188,7 +191,7 @@ export abstract class BaseHlsSession< const playlistExists = some( workingDirectoryFiles.get(), - (f) => f === basename(this._m3u8PlaylistPath), + (f) => f === basename(this._variantPlaylistPath), ); const additionalRequired = this.getAdditionalRequiredFiles(); @@ -233,7 +236,7 @@ export abstract class BaseHlsSession< } get m3uPlaylistPath() { - return this._m3u8PlaylistPath; + return this._variantPlaylistPath; } get masterPlaylistPath() { diff --git a/server/src/stream/hls/HlsSession.ts b/server/src/stream/hls/HlsSession.ts index 9682925df..a2ea10959 100644 --- a/server/src/stream/hls/HlsSession.ts +++ b/server/src/stream/hls/HlsSession.ts @@ -26,7 +26,8 @@ import { filter, isEmpty, last, maxBy, sortBy } from 'lodash-es'; import fs from 'node:fs/promises'; import path, { basename, dirname, extname } from 'node:path'; import type { DeepRequired } from 'ts-essentials'; -import { ProgramStreamFactory } from '../ProgramStreamFactory.ts'; +import { waitForEvent } from '../../types/events.ts'; +import type { ProgramStreamFactory } from '../ProgramStreamFactory.ts'; import type { BaseHlsSessionOptions } from './BaseHlsSession.js'; import { BaseHlsSession } from './BaseHlsSession.js'; import { HlsMasterPlaylistMutator } from './HlsMasterPlaylistMutator.js'; @@ -47,6 +48,11 @@ export interface HlsSessionOptions extends BaseHlsSessionOptions { streamMode: 'hls' | 'hls_direct_v2'; } +type GetMasterPlaylistOptions = { + wait?: boolean; + maxWaitMs?: number; +}; + /** * Initializes an ffmpeg process that concatenates via the /playlist * endpoint and outputs an HLS format + segments @@ -80,10 +86,17 @@ export class HlsSession extends BaseHlsSession { return this.readPlaylist(); } - async getMasterPlaylist(): Promise> { + async getMasterPlaylist({ + wait = false, + maxWaitMs = 30_000, + }: GetMasterPlaylistOptions = {}): Promise> { return Result.attemptAsync(async () => { if (!(await fileExists(this._masterPlaylistPath))) { - return undefined; + if (!wait) return; + return waitForEvent(this, 'start', maxWaitMs).then(async () => { + const result = await this.getMasterPlaylist({ wait: false }); + return result.getOrThrow(); + }); } const content = await fs.readFile(this._masterPlaylistPath, 'utf-8'); const rendition = this.#currentSubtitleRendition; @@ -152,7 +165,7 @@ export class HlsSession extends BaseHlsSession { this.logger.trace( 'No playlist for HLS sessions at %s', - this._m3u8PlaylistPath, + this._variantPlaylistPath, ); return; }); @@ -410,11 +423,11 @@ export class HlsSession extends BaseHlsSession { } private async readPlaylist() { - if (!(await fileExists(this._m3u8PlaylistPath))) { + if (!(await fileExists(this._variantPlaylistPath))) { return; } - const playlistContents = await fs.readFile(this._m3u8PlaylistPath, { + const playlistContents = await fs.readFile(this._variantPlaylistPath, { encoding: 'utf-8', }); diff --git a/server/src/types/events.ts b/server/src/types/events.ts new file mode 100644 index 000000000..d1580a929 --- /dev/null +++ b/server/src/types/events.ts @@ -0,0 +1,25 @@ +import type { EventEmitter } from 'node:events'; + +/** + * Waits for a specific event to fire, or rejects if the timeout is reached. + * @template T The expected type array of the event payload data. + */ +export function waitForEvent( + emitter: Emitter, + eventName: string, + timeoutMs: number, +): Promise { + return new Promise((resolve, reject) => { + const timer: NodeJS.Timeout = setTimeout(() => { + emitter.off(eventName, eventHandler); + reject(new Error(`Event '${eventName}' timed out after ${timeoutMs}ms`)); + }, timeoutMs); + + const eventHandler = (...args: unknown[]) => { + clearTimeout(timer); + resolve(args as T); // Cast the spread array to our generic type tuple + }; + + emitter.once(eventName, eventHandler); + }); +} diff --git a/server/src/util/future.ts b/server/src/util/future.ts new file mode 100644 index 000000000..bc417f60d --- /dev/null +++ b/server/src/util/future.ts @@ -0,0 +1,81 @@ +export class Future implements Promise { + #promise: Promise; + #resolve!: (v: T | PromiseLike) => void; + #reject!: (reason?: unknown) => void; + #state: 'pending' | 'fulfilled' | 'rejected' = 'pending'; + #value: T | undefined; + #err: unknown; + + constructor() { + this.#promise = new Promise((resolve, reject) => { + this.#resolve = resolve; + this.#reject = reject; + }); + + this.#promise.then( + (v) => { + this.#state = 'fulfilled'; + this.#value = v; + }, + (err) => { + this.#state = 'rejected'; + this.#err = err; + }, + ); + } + + [Symbol.toStringTag]!: string; + + resolve(value: T | PromiseLike) { + if (this.#state === 'pending') { + this.#resolve(value); + } else { + throw new Error( + 'Resolving already fulfilled future with state ' + this.#state, + ); + } + } + + reject(e: unknown) { + if (this.#state === 'pending') { + this.#reject(e); + } else { + throw new Error( + 'Rejecting already fulfilled future with state ' + this.#state, + ); + } + } + + then( + onfulfilled?: ((value: T) => TResult1 | PromiseLike) | null, + onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | null, + ): Promise { + return this.#promise.then(onfulfilled, onrejected); + } + + catch( + onrejected?: ((reason: unknown) => TResult | PromiseLike) | null, + ): Promise { + return this.#promise.catch(onrejected); + } + + finally(onfinally?: (() => void) | null): Promise { + return this.#promise.finally(onfinally); + } + + get promise() { + return this.#promise; + } + + get state() { + return this.#state; + } + + get value() { + return this.#value; + } + + get error() { + return this.#err; + } +}