fix: wait for master playlist to be present when requesting m3u8

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

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<T> implements Promise<T> {
-  #promise: Promise<T>;
-  #resolve!: (v: T | PromiseLike<T>) => void;
-  #reject!: (reason?: unknown) => void;
-  #state: 'pending' | 'fulfilled' | 'rejected' = 'pending';
-  #value: T | undefined;
-  #err: unknown;
-
-  constructor() {
-    this.#promise = new Promise<T>((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<T>) {
-    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<TResult1 = T, TResult2 = never>(
-    onfulfilled?: ((value: T) => TResult1 | PromiseLike<TResult1>) | null,
-    onrejected?: ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
-  ): Promise<TResult1 | TResult2> {
-    return this.#promise.then(onfulfilled, onrejected);
-  }
-
-  catch<TResult = never>(
-    onrejected?: ((reason: unknown) => TResult | PromiseLike<TResult>) | null,
-  ): Promise<T | TResult> {
-    return this.#promise.catch(onrejected);
-  }
-
-  finally(onfinally?: (() => void) | null): Promise<T> {
-    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<number, string[]>();
   #startPromises: Promise<boolean>[] = [];
 
-  @InjectLogger() private declare readonly logger: Logger;
+  @InjectLogger() declare private readonly logger: Logger;
 
   constructor(
     @inject(TunarrSubprocessService)

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';

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() {

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<HlsSessionOptions> {
     return this.readPlaylist();
   }
 
-  async getMasterPlaylist(): Promise<Result<string | undefined>> {
+  async getMasterPlaylist({
+    wait = false,
+    maxWaitMs = 30_000,
+  }: GetMasterPlaylistOptions = {}): Promise<Result<string | undefined>> {
     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<HlsSessionOptions> {
 
         this.logger.trace(
           'No playlist for HLS sessions at %s',
-          this._m3u8PlaylistPath,
+          this._variantPlaylistPath,
         );
         return;
       });
@@ -410,11 +423,11 @@ export class HlsSession extends BaseHlsSession<HlsSessionOptions> {
   }
 
   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',
     });
 

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 extends EventEmitter, T>(
+  emitter: Emitter,
+  eventName: string,
+  timeoutMs: number,
+): Promise<T> {
+  return new Promise<T>((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);
+  });
+}

server/src/util/future.ts

@@ -0,0 +1,81 @@
+export class Future<T> implements Promise<T> {
+  #promise: Promise<T>;
+  #resolve!: (v: T | PromiseLike<T>) => void;
+  #reject!: (reason?: unknown) => void;
+  #state: 'pending' | 'fulfilled' | 'rejected' = 'pending';
+  #value: T | undefined;
+  #err: unknown;
+
+  constructor() {
+    this.#promise = new Promise<T>((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<T>) {
+    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<TResult1 = T, TResult2 = never>(
+    onfulfilled?: ((value: T) => TResult1 | PromiseLike<TResult1>) | null,
+    onrejected?: ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
+  ): Promise<TResult1 | TResult2> {
+    return this.#promise.then(onfulfilled, onrejected);
+  }
+
+  catch<TResult = never>(
+    onrejected?: ((reason: unknown) => TResult | PromiseLike<TResult>) | null,
+  ): Promise<T | TResult> {
+    return this.#promise.catch(onrejected);
+  }
+
+  finally(onfinally?: (() => void) | null): Promise<T> {
+    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;
+  }
+}