Use stream metadata request to detect when stream is ready (#2306)

* Choose whether to wait for stream start based on queryparam * debug log * call js endpoint * Create new func for metadata call * remove debug lines * small refactor from review comment
livepeer · Sep 17, 2024 · dd62fb0 · dd62fb0
1 parent 36c85c8
commit dd62fb0
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 1 deletion.
diff --git a/packages/api/src/controllers/helpers.ts b/packages/api/src/controllers/helpers.ts
@@ -699,6 +699,37 @@ export const triggerCatalystPullStart =
         throw new Error(`failed to trigger catalyst pull`);
       };
 
+export const waitCatalystStreamReady =
+  process.env.NODE_ENV === "test"
+    ? async () => {} // noop in case of tests
+    : async (stream: DBStream, playbackUrl: string) => {
+        const metadataUrl = new URL(playbackUrl);
+        metadataUrl.pathname = "/json_video+" + stream.playbackId + ".js";
+        const url = metadataUrl.toString();
+
+        const deadline = Date.now() + 2 * PULL_START_TIMEOUT;
+        while (Date.now() < deadline) {
+          const res = await fetchWithTimeoutAndRedirects(url, {
+            method: "GET",
+            timeout: PULL_START_TIMEOUT,
+            maxRedirects: 10,
+          });
+          const body = await res.text();
+          if (res.ok && body.includes(`"meta":`)) {
+            return;
+          }
+
+          logger.warn(
+            `getCatalystStreamMetadata failed for playbackUrl=${url} status=${
+              res.status
+            } error=${JSON.stringify(body)}`,
+          );
+          await sleep(250);
+        }
+
+        throw new Error(`failed to trigger catalyst pull`);
+      };
+
 export const triggerCatalystStreamStopSessions = (
   req: Request,
   playback_id: string,

diff --git a/packages/api/src/controllers/stream.ts b/packages/api/src/controllers/stream.ts
@@ -59,6 +59,7 @@ import {
   triggerCatalystPullStart,
   triggerCatalystStreamStopSessions,
   triggerCatalystStreamUpdated,
+  waitCatalystStreamReady,
 } from "./helpers";
 import { toExternalSession } from "./session";
 import wowzaHydrate from "./wowza-hydrate";
@@ -1256,7 +1257,8 @@ app.put(
       ? pathJoin(streamPullUrl + stream.playbackId, `index.m3u8`)
       : getHLSPlaybackUrl(ingest, stream);
     if (!stream.isActive || streamExisted) {
-      await triggerCatalystPullStart(stream, playbackUrl);
+      triggerCatalystPullStart(stream, playbackUrl);
+      await waitCatalystStreamReady(stream, playbackUrl);
     }
 
     res.status(streamExisted ? 200 : 201);