fix(gateway): fail-fast + retry on screen share merge failure (black tile zombie)
Root cause (2026-08-12 11:50 test): merge ffmpeg hit a transient YouTube 403 and exited code 8 BEFORE prepareStream attached its input listeners (voice release+join takes ~10s). The input's end/error events fired into the void, the encoder stdin never received EOF, demux resolved with fallback 0x0 metadata, setSpeaking fired anyway → stream 'started' with zero frames for 8+ minutes (black tile, both ffmpeg processes hung). Fixes: - mediaSource: pass yt-dlp http_headers (UA/referer) to the merge ffmpeg via -headers to suppress transient 403s; destroy the returned stream with an error when the merge exits non-zero before producing bytes. - screenShareController: resolveInputWithRetry — tee the merge stream and wait for the first readable byte (12s timeout) before proceeding; on error/EOF/timeout retry the whole resolution with a FRESH yt-dlp run (signed DASH URLs expire fast) up to 3 attempts. Stuck merges get EPIPE via input.destroy() so no process leaks per attempt. - prepareStream: race guard — if the input already ended/destroyed before listeners attach, EOF the encoder stdin immediately; first-frame watchdog in playStream rejects 'started but nothing flowing' after 10s instead of resolving with a silent black stream. Tests: +2 (merge-fail zero-byte terminal state, -headers forwarding).
This commit is contained in:
asepharyana
@@ -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<void>((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<void>((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<void>((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);
|
||||
})();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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<string, string> | 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<string, string>,
|
||||
): 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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<string | Readable> {
|
||||
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<void>((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<ScreenSharePlayback> {
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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=<file> → 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<string>((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<string, unknown>).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<void>((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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user