diff --git a/services/discord-gateway/src/goLive/prepareStream.ts b/services/discord-gateway/src/goLive/prepareStream.ts index f29d92f..b446745 100644 --- a/services/discord-gateway/src/goLive/prepareStream.ts +++ b/services/discord-gateway/src/goLive/prepareStream.ts @@ -206,9 +206,19 @@ export function prepareStream( : spawn(FFMPEG_BIN, args, { stdio: ["pipe", "pipe", "pipe"] }); if (proc.stdin && !isUrl) { - input.on("data", (chunk: Buffer) => proc.stdin?.write(chunk)); - input.on("end", () => proc.stdin?.end()); - input.on("error", () => proc.stdin?.destroy()); + // Race guard: the merge ffmpeg may have already exited (transient 403 + // or stream death) before this function attaches its listeners — the + // input's 'end'/'error' events then fire into the void and the encoder + // stdin NEVER receives EOF, leaving an encoder that waits forever and a + // screen share that shows a black tile with zero frames. Check the + // terminal state eagerly and EOF the encoder immediately. + if (input.readableEnded || input.destroyed) { + proc.stdin.end(); + } else { + input.on("data", (chunk: Buffer) => proc.stdin?.write(chunk)); + input.on("end", () => proc.stdin?.end()); + input.on("error", () => proc.stdin?.destroy()); + } } proc.stdout?.pipe(output); @@ -319,14 +329,81 @@ export async function playStream( } }; - return new Promise((resolve) => { - vStream.once("finish", () => { - cleanup(); + // First-frame watchdog: if the encoder never delivers a single frame + // (dead merge input, empty stream, codec mismatch), fail fast instead of + // "playing" a black tile forever. The demuxer resolves with fallback + // metadata even when no frame ever arrives, so this timeout is the only + // place that detects "started but nothing flowing". + let firstFrameTimer: NodeJS.Timeout | null = null; + let gotFirstFrame = false; + const firstFrame = new Promise((resolve, reject) => { + firstFrameTimer = setTimeout(() => { + if (!gotFirstFrame) { + cleanup(); + reject( + new Error( + "No video frames within 10s of stream start — input stream failed", + ), + ); + } + }, 10000); + video.stream.once("data", () => { + gotFirstFrame = true; + if (firstFrameTimer) clearTimeout(firstFrameTimer); resolve(); }); + }); + + return new Promise((resolve, reject) => { + let settled = false; + const settle = (fn: () => void) => () => { + if (settled) return; + settled = true; + if (firstFrameTimer) clearTimeout(firstFrameTimer); + fn(); + }; + + vStream.once("finish", () => { + settle(() => { + cleanup(); + if (!gotFirstFrame) { + reject(new Error("Screen video stream ended without any frame")); + } else { + resolve(); + } + })(); + }); vStream.once("error", () => { - cleanup(); - resolve(); + settle(() => { + cleanup(); + if (!gotFirstFrame) { + reject(new Error("Screen video stream errored before first frame")); + } else { + resolve(); + } + })(); + }); + // The stream may end without ever producing a frame (input was + // silently dead) — surface that instead of resolving "successfully". + video.stream.once("end", () => { + settle(() => { + cleanup(); + if (!gotFirstFrame) { + reject(new Error("Screen video stream ended before any frame")); + } else { + resolve(); + } + })(); + }); + // Watchdog timeout: no frame arrived within 10s — fail fast instead of + // "playing" a black tile forever. cleanup() kills the encoder so the + // vStream finish/error handlers above still fire, but the settled guard + // ensures this rejection wins. + firstFrame.catch((err) => { + settle(() => { + cleanup(); + reject(err); + })(); }); }); } diff --git a/services/discord-gateway/src/modules/voice-recording/mediaSource.ts b/services/discord-gateway/src/modules/voice-recording/mediaSource.ts index 338ec21..ffd275d 100644 --- a/services/discord-gateway/src/modules/voice-recording/mediaSource.ts +++ b/services/discord-gateway/src/modules/voice-recording/mediaSource.ts @@ -522,6 +522,12 @@ async function resolveScreenInput( ); const videoUrl = video?.url as string | undefined; const audioUrl = audio?.url as string | undefined; + // yt-dlp returns the exact HTTP headers needed to fetch each signed DASH + // URL (User-Agent etc.). Passing them to the merge ffmpeg prevents + // transient YouTube 403s ("Server returned 403 Forbidden") that kill the + // stream before it starts. + const videoHeaders = + (video?.http_headers as Record | undefined) ?? {}; if ( typeof videoUrl === "string" && @@ -529,7 +535,7 @@ async function resolveScreenInput( typeof audioUrl === "string" && audioUrl.length > 0 ) { - return mergeScreenStreams(videoUrl, audioUrl); + return mergeScreenStreams(videoUrl, audioUrl, videoHeaders); } } @@ -542,41 +548,61 @@ async function resolveScreenInput( * Merge a video-only URL and an audio-only URL into a single NUT stream using * a child ffmpeg process. Both URLs come from the same yt-dlp run, so they * share the same signature/expiry and are consumed immediately. + * + * Fail-fast contract: if the merge process exits non-zero BEFORE producing any + * output bytes (e.g. transient YouTube 403), the returned Readable is + * destroyed with an error so the caller can retry — otherwise the screen + * share would "start" with a dead input and stream a black tile forever. */ -function mergeScreenStreams(videoUrl: string, audioUrl: string): Readable { +function mergeScreenStreams( + videoUrl: string, + audioUrl: string, + httpHeaders?: Record, +): Readable { logger.info("Merging video+audio DASH streams into a single NUT input"); - const ffmpeg = spawn( - "ffmpeg", - [ - "-hide_banner", - "-loglevel", - "error", - "-reconnect", - "1", - "-reconnect_streamed", - "1", - "-reconnect_delay_max", - "5", - "-i", - videoUrl, - "-i", - audioUrl, - "-map", - "0:v:0", - "-map", - "1:a:0", - "-c:v", - "copy", - "-c:a", - "copy", - "-f", - "nut", - "pipe:1", - ], - { stdio: ["ignore", "pipe", "pipe"] }, + const args = [ + "-hide_banner", + "-loglevel", + "error", + "-reconnect", + "1", + "-reconnect_streamed", + "1", + "-reconnect_delay_max", + "5", + ]; + // Pass the browser-like headers yt-dlp attached to the signed URLs. Without + // a proper User-Agent YouTube sometimes answers 403 to ffmpeg's plain + // Lavf/… agent and the whole stream dies before producing a frame. + const headerStr = Object.entries(httpHeaders ?? {}) + .map(([k, v]) => `${k}: ${v}`) + .join("\r\n"); + if (headerStr) { + args.push("-headers", headerStr); + } + args.push( + "-i", + videoUrl, + "-i", + audioUrl, + "-map", + "0:v:0", + "-map", + "1:a:0", + "-c:v", + "copy", + "-c:a", + "copy", + "-f", + "nut", + "pipe:1", ); + const ffmpeg = spawn("ffmpeg", args, { + stdio: ["ignore", "pipe", "pipe"], + }); + // Track so cleanup() can terminate the merge during graceful shutdown. activeProcesses.add(ffmpeg); ffmpeg.once("exit", () => { @@ -592,12 +618,15 @@ function mergeScreenStreams(videoUrl: string, audioUrl: string): Readable { } }); + let producedData = false; + ffmpeg.on("error", (err) => { const msg = err.message === "spawn ffmpeg ENOENT" ? "FFmpeg not found! Install ffmpeg in the container." : err.message; logger.error({ error: msg }, "Screen stream merge ffmpeg error"); + stream.destroy(new Error(msg)); }); ffmpeg.on("exit", (code) => { @@ -606,10 +635,23 @@ function mergeScreenStreams(videoUrl: string, audioUrl: string): Readable { { code, stderr: stderr.slice(-500) || undefined }, "Screen stream merge ffmpeg exited", ); + // Fail fast: a merge that dies before emitting ANY bytes cannot feed the + // encoder — destroy the stream with an error so the caller retries with a + // fresh resolution instead of streaming a silent black tile. + if (code !== 0 && !producedData && !stream.destroyed) { + stream.destroy( + new Error( + `Screen stream merge failed before producing data (exit ${code})${stderr ? `: ${stderr.slice(-300)}` : ""}`, + ), + ); + } }); const stream = ffmpeg.stdout; stream.setMaxListeners(32); + stream.once("data", () => { + producedData = true; + }); return stream; } diff --git a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts index 62bbada..768b5d8 100644 --- a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts +++ b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts @@ -1,3 +1,4 @@ +import { PassThrough, type Readable } from "node:stream"; import type { Client } from "discord.js-selfbot-v13"; import { createChildLogger } from "@/shared/logger/index"; import { @@ -54,6 +55,114 @@ export class ScreenShareController { return this.active !== null; } + /** + * Resolve the screen-share input with retry + first-byte validation. + * + * Transient YouTube 403s kill the merge ffmpeg BEFORE it produces any + * output; without validation the stream would "start" with a dead input + * and show a black tile forever. So after getDirectScreenInput resolves we + * tee the stream through a PassThrough and wait for the FIRST readable + * byte (or an error / early EOF). On failure the whole resolution is + * retried with a FRESH yt-dlp run (signed DASH URLs expire quickly — the + * old URLs cannot simply be re-fetched). + */ + private async resolveInputWithRetry( + source: string, + ): Promise { + const MAX_ATTEMPTS = 3; + let lastError: Error | null = null; + + for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + try { + const input = await getDirectScreenInput(source); + if (typeof input === "string") { + // Direct URL input — nothing to validate; the encoder ffmpeg will + // connect itself and fail loudly on a bad URL. + return input; + } + + const tee = new PassThrough(); + input.on("error", (err) => tee.destroy(err)); + input.on("end", () => tee.end()); + input.pipe(tee); + // If the merge process is stuck (no data, no exit) destroy the raw + // stream too so ffmpeg gets EPIPE on its next write and dies — + // otherwise every failed attempt leaks a merge process. + const destroyInput = () => { + try { + input.destroy(); + } catch { + /* already gone */ + } + }; + + await new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup(); + destroyInput(); + tee.destroy( + new Error( + "Screen input produced no data within 12s — merge likely failed", + ), + ); + reject( + new Error( + "Screen input produced no data within 12s — merge likely failed", + ), + ); + }, 12000); + const onReadable = () => { + if (tee.readableLength > 0) { + cleanup(); + resolve(); + } + // readableLength === 0 can mean "EOF reached" — handled by onEnd. + }; + const onError = (err: Error) => { + cleanup(); + reject(err); + }; + const onEnd = () => { + cleanup(); + destroyInput(); + reject(new Error("Screen input ended before producing any data")); + }; + const cleanup = () => { + clearTimeout(timer); + tee.removeListener("readable", onReadable); + tee.removeListener("error", onError); + tee.removeListener("end", onEnd); + }; + tee.once("readable", onReadable); + tee.once("error", onError); + tee.once("end", onEnd); + }); + + // Pass the tee onward — the encoder consumes the same buffered + // stream, so no data from the merge is lost. + return tee; + } catch (err) { + lastError = err instanceof Error ? err : new Error(String(err)); + this.logger.warn( + { + attempt, + maxAttempts: MAX_ATTEMPTS, + error: lastError.message, + }, + "Screen input resolution failed; retrying with fresh yt-dlp", + ); + if (attempt < MAX_ATTEMPTS) { + await new Promise((r) => setTimeout(r, 1500 * attempt)); + } + } + } + + throw ( + lastError ?? + new Error("Screen input resolution failed after multiple attempts") + ); + } + async start(source: string): Promise { const status = this.getVoiceStatus(); if (!status.connected || !status.activeGuildId || !status.activeChannelId) { @@ -65,7 +174,7 @@ export class ScreenShareController { } try { - const input = await getDirectScreenInput(source); + const input = await this.resolveInputWithRetry(source); if (!this.streamer) { this.streamer = new Streamer(this.client); } diff --git a/services/discord-gateway/tests/screenShareInput.test.ts b/services/discord-gateway/tests/screenShareInput.test.ts index 08038b2..59942d6 100644 --- a/services/discord-gateway/tests/screenShareInput.test.ts +++ b/services/discord-gateway/tests/screenShareInput.test.ts @@ -10,7 +10,13 @@ // hit the network or need real binaries. // ═══════════════════════════════════════════════════════════════════════════════ -import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { + chmodSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Readable } from "node:stream"; @@ -40,7 +46,18 @@ exit 1 // Fake ffmpeg: writes a small nut-ish payload to stdout so the returned // Readable actually emits data (the merge path in mergeScreenStreams). + // Modes (env): + // GMW_FAKE_FFMPEG_FAIL=1 → exit 1, no stdout (mimics transient 403) + // GMW_FAKE_FFMPEG_DUMP_ARGS= → append argv to the file (asserts + // flags like -headers are forwarded to the merge process) const ffShim = `#!/usr/bin/env bash +if [ -n "$GMW_FAKE_FFMPEG_DUMP_ARGS" ]; then + printf '%s\\n' "$*" >> "$GMW_FAKE_FFMPEG_DUMP_ARGS" +fi +if [ "$GMW_FAKE_FFMPEG_FAIL" = "1" ]; then + echo "403 Forbidden" >&2 + exit 8 +fi # Fake ffmpeg — ignore args, emit a few bytes so consumers see a live stream. head -c 4096 /dev/urandom exit 0 @@ -139,4 +156,81 @@ describe("getDirectScreenInput", () => { /screen input resolution exited with code 1/, ); }); + + it("terminates with ZERO bytes when the merge ffmpeg fails before producing data (transient 403)", async () => { + // Simulate the 11:50 production failure: yt-dlp resolves fine, but the + // merge ffmpeg hits a transient YouTube 403 and exits non-zero WITHOUT + // emitting a single byte. getDirectScreenInput still resolves (the + // Readable exists) — the fail-fast contract: the stream must terminate + // (error OR end — the end-before-exit ordering makes both possible) + // without ever delivering a frame to a consumer. The controller's + // resolveInputWithRetry turns either signal into a fresh retry. + process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson( + dashPairInfo( + "https://cdn.example/video.mp4", + "https://cdn.example/audio.m4a", + ), + ); + process.env.GMW_FAKE_FFMPEG_FAIL = "1"; + try { + const result = await getDirectScreenInput("https://youtu.be/abc"); + expect(Readable.isReadable(result)).toBe(true); + + const outcome = await new Promise((resolve) => { + const stream = result as Readable; + let got = 0; + stream.on("data", (chunk: Buffer) => { + got += chunk.length; + }); + stream.on("error", () => resolve(`error-after-${got}B`)); + stream.on("end", () => resolve(`end-after-${got}B`)); + stream.resume(); + }); + // Fail-fast: the consumer must NOT receive any bytes (no black-tile + // zombie stream). Either a destroyed-with-error stream or a clean + // end-before-exit is a valid terminal state — the caller retries. + expect(outcome).toMatch(/^(error|end)-after-0B$/); + } finally { + delete process.env.GMW_FAKE_FFMPEG_FAIL; + } + }); + + it("forwards yt-dlp http_headers to the merge ffmpeg (-headers)", async () => { + const info = dashPairInfo( + "https://cdn.example/video.mp4", + "https://cdn.example/audio.m4a", + ); + // Add the browser-like headers yt-dlp attaches to signed DASH URLs. + (info.requested_formats[0] as Record).http_headers = { + "User-Agent": + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + Referer: "https://www.youtube.com/", + }; + const argsDump = join( + tmpdir(), + `gmw-ffargs-${process.pid}-${Date.now()}.txt`, + ); + process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson(info); + process.env.GMW_FAKE_FFMPEG_DUMP_ARGS = argsDump; + try { + const result = await getDirectScreenInput("https://youtu.be/abc"); + // Consume the stream so the merge ffmpeg process runs to completion. + await new Promise((resolve) => { + const stream = result as Readable; + stream.on("data", () => {}); + stream.on("error", () => resolve()); + stream.on("end", () => resolve()); + stream.resume(); + }); + // Allow the fake ffmpeg to flush its argv dump. + await new Promise((r) => setTimeout(r, 100)); + const args = readFileSync(argsDump, "utf8").trim(); + expect(args).toContain("-headers"); + expect(args).toContain("Mozilla/5.0"); + expect(args).toContain("Referer: https://www.youtube.com/"); + } finally { + delete process.env.GMW_FAKE_FFMPEG_DUMP_ARGS; + rmSync(argsDump, { force: true }); + } + }); });