Compare commits
15
Commits
3acb03391a
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
67ab289caa | ||
|
|
ef9e243609 | ||
|
|
42a503c206 | ||
|
|
652974e23a | ||
|
|
407e003399 | ||
|
|
968a43b0f4 | ||
|
|
91c7a67d2f | ||
|
|
8615383829 | ||
|
|
10d7ecd405 | ||
|
|
ff554fcff2 | ||
|
|
f8b253ba5e | ||
|
|
c8473b0610 | ||
|
|
3deca91ffe | ||
|
|
edec2edf82 | ||
|
|
17013fe1e5 |
@@ -11,15 +11,10 @@
|
||||
let
|
||||
pkgs = import nixpkgs { inherit system; };
|
||||
|
||||
# libdatachannel 0.24.0 — the version the GoLive N-API binding links
|
||||
# against. nixpkgs 0.24.1 was built against a newer glibc (ABI
|
||||
# GLIBC_ABI_GNU2_TLS missing on this host), so pin 0.24.0 explicitly.
|
||||
libdatachannel-src = pkgs.fetchFromGitHub {
|
||||
owner = "paullouisageneau";
|
||||
repo = "libdatachannel";
|
||||
rev = "v0.24.0";
|
||||
sha256 = "1jk53qsrihg1bsc0dmr5rajkgp3hi7d98pvpr0qnpk15b7ilb6fy";
|
||||
};
|
||||
# libdatachannel for the GoLive N-API binding. nixpkgs 0.24.1 is built
|
||||
# against this host's glibc and ships both lib + dev headers, so the
|
||||
# binding links cleanly inside the Nix sandbox (no manual cmake build).
|
||||
libdatachannel = pkgs.libdatachannel;
|
||||
|
||||
# Source filter: `path:` literals do NOT respect .gitignore by default,
|
||||
# so a dirty local out/ (stale chunks from previous builds) leaks into
|
||||
@@ -172,6 +167,7 @@ WRAPPER
|
||||
pkgs.pkg-config
|
||||
pkgs.openssl
|
||||
pkgs.openssl.dev
|
||||
libdatachannel.dev # rtc/rtc.hpp headers for the GoLive binding
|
||||
pkgs.git # libdatachannel FetchContent clones from GitHub
|
||||
pkgs.cacert
|
||||
];
|
||||
@@ -201,24 +197,21 @@ WRAPPER
|
||||
done
|
||||
echo "=== Building libdatachannel-min N-API binding ==="
|
||||
# The GoLive screen-share stack uses a minimal N-API binding
|
||||
# (native/libdatachannel-min) over libdatachannel 0.24.0 built from
|
||||
# source. node-gyp links against the libdatachannel .so.
|
||||
# (native/libdatachannel-min) over nixpkgs libdatachannel.
|
||||
(
|
||||
cd native/libdatachannel-min
|
||||
# libdatachannel 0.24.0 fetched from GitHub (see flake inputs) —
|
||||
# build with CMake, then node-gyp links against the .so.
|
||||
mkdir -p build/ldc
|
||||
cd build/ldc
|
||||
cmake -DCMAKE_BUILD_TYPE=Release \
|
||||
-DNO_EXAMPLES=ON -DNO_TESTS=ON -DNO_WEBSOCKET=ON \
|
||||
-DNO_MEDIA=OFF \
|
||||
-DCMAKE_INSTALL_PREFIX=$PWD/install \
|
||||
"${opensslDevEnv}" ${libdatachannel-src} 2>&1 || true
|
||||
make -j"$NIX_BUILD_CORES" 2>&1 || true
|
||||
cd ..
|
||||
LD_LIBRARY_PATH=$PWD/ldc node-gyp rebuild 2>&1 || true
|
||||
cp -r build/Release/datachannel_min.node . 2>/dev/null || true
|
||||
echo "libdatachannel-min binding: $(ls -la datachannel_min.node 2>/dev/null | awk '{print $5}') bytes"
|
||||
# binding.gyp resolves include/lib from env (LDC_INCLUDE = .dev
|
||||
# include root, LDC_LIB = lib output dir, NAPI_INCLUDE =
|
||||
# node-addon-api include root).
|
||||
NAPI_INCLUDE=$(find ../../node_modules/.pnpm -maxdepth 3 \
|
||||
-type d -path "*node_modules/node-addon-api" | head -1)
|
||||
echo "NAPI_INCLUDE=$NAPI_INCLUDE"
|
||||
LDC_INCLUDE=${libdatachannel.dev} LDC_LIB=${libdatachannel.out}/lib/libdatachannel.so.0.24.1 \
|
||||
NAPI_INCLUDE=$NAPI_INCLUDE \
|
||||
npx node-gyp rebuild 2>&1 || true
|
||||
ls -la build/Release/datachannel_min.node 2>/dev/null \
|
||||
&& echo "libdatachannel-min binding OK: $(stat -c%s build/Release/datachannel_min.node) bytes" \
|
||||
|| echo "WARN: libdatachannel-min binding build FAILED (screen share disabled)"
|
||||
)
|
||||
echo "=== Compiling TypeScript ===="
|
||||
npx tsc 2>&1
|
||||
@@ -254,6 +247,22 @@ WRAPPER
|
||||
mkdir -p $out/lib/gmw-discord-gateway
|
||||
cp -r dist node_modules package.json tsconfig.json $out/lib/gmw-discord-gateway/
|
||||
|
||||
# GoLive native binding — loadNative resolves it relative to
|
||||
# dist/goLive/native.js, i.e. <root>/native/libdatachannel-min/
|
||||
# build/Release/datachannel_min.node; libdatachannel .so must sit
|
||||
# next to it and be on LD_LIBRARY_PATH at runtime.
|
||||
mkdir -p $out/lib/gmw-discord-gateway/native/libdatachannel-min/build/Release
|
||||
cp native/libdatachannel-min/build/Release/datachannel_min.node \
|
||||
$out/lib/gmw-discord-gateway/native/libdatachannel-min/build/Release/ 2>/dev/null || true
|
||||
mkdir -p $out/lib/gmw-discord-gateway/native/libdatachannel-min/build/ldc
|
||||
cp -rL native/libdatachannel-min/build/ldc/libdatachannel.so* \
|
||||
$out/lib/gmw-discord-gateway/native/libdatachannel-min/build/ldc/ 2>/dev/null || true
|
||||
# If the binding failed to build, screen share is simply disabled —
|
||||
# the gateway itself must still start.
|
||||
if [ ! -f $out/lib/gmw-discord-gateway/native/libdatachannel-min/build/Release/datachannel_min.node ]; then
|
||||
echo "WARN: datachannel_min.node missing — GoLive screen share disabled in this build"
|
||||
fi
|
||||
|
||||
# Also include drizzle migrations if they exist
|
||||
cp -r drizzle $out/lib/gmw-discord-gateway/ 2>/dev/null || true
|
||||
|
||||
@@ -262,6 +271,7 @@ WRAPPER
|
||||
#!${pkgs.runtimeShell}
|
||||
cd $out/lib/gmw-discord-gateway
|
||||
export PATH=${pkgs.ffmpeg-headless}/bin:${pkgs.yt-dlp}/bin:\$PATH
|
||||
export LD_LIBRARY_PATH=${libdatachannel.out}/lib:\$LD_LIBRARY_PATH
|
||||
exec ${nodejs}/bin/node dist/index.js
|
||||
WRAPPER
|
||||
chmod +x $out/bin/gmw-discord-gateway
|
||||
|
||||
@@ -4,11 +4,11 @@
|
||||
"target_name": "libdatachannel_min",
|
||||
"sources": ["binding.cpp"],
|
||||
"include_dirs": [
|
||||
"<!@(node -p \"require('node-addon-api').include\")",
|
||||
"/home/code/GMW/services/discord-gateway/node_modules/.pnpm/@lng2004+node-datachannel@0.32.0-20260202/node_modules/@lng2004/node-datachannel/build/_deps/libdatachannel-src/include"
|
||||
"<!(node -e \"console.log(process.env.NAPI_INCLUDE || (() => { try { return require('node-addon-api').include; } catch { return '/nonexistent'; } })())\")",
|
||||
"<!(node -e \"const s=process.env.LDC_INCLUDE||'/nix/store/39a85gpfjqy3h3k8jwrwh7m9yc3inqw7-source';console.log(s+'/include')\")"
|
||||
],
|
||||
"libraries": [
|
||||
"/tmp/ldc-build/libdatachannel.so.0.24.0"
|
||||
"<!(node -e \"console.log(process.env.LDC_LIB || '/tmp/ldc-build/libdatachannel.so.0.24.0')\")"
|
||||
],
|
||||
"cflags": ["-std=c++17", "-fexceptions"],
|
||||
"cflags_cc": ["-std=c++17", "-fexceptions"],
|
||||
|
||||
@@ -279,12 +279,51 @@ export async function initializeDiscordGateway() {
|
||||
});
|
||||
|
||||
process.on("uncaughtException", (err) => {
|
||||
const code =
|
||||
typeof (err as NodeJS.ErrnoException).code === "string"
|
||||
? (err as NodeJS.ErrnoException).code
|
||||
: "";
|
||||
// Transient stream-teardown errors (voice stop/disconnect races, child
|
||||
// process stdin closed while we still write) are NOT fatal — crashing the
|
||||
// gateway on EPIPE takes the whole bot offline mid-music. Log + continue.
|
||||
if (
|
||||
code === "EPIPE" ||
|
||||
code === "ERR_STREAM_DESTROYED" ||
|
||||
code === "ERR_STREAM_WRITE_AFTER_END" ||
|
||||
code === "ECONNRESET"
|
||||
) {
|
||||
logger.warn(
|
||||
{ error: err },
|
||||
"Uncaught transient stream error — continuing",
|
||||
);
|
||||
return;
|
||||
}
|
||||
logger.error({ error: err }, "Uncaught exception");
|
||||
gracefulShutdown("uncaughtException");
|
||||
});
|
||||
|
||||
process.on("unhandledRejection", (reason, promise) => {
|
||||
logger.error({ reason, promise }, "Unhandled rejection");
|
||||
const err =
|
||||
reason instanceof Error ? reason : new Error(String(reason ?? "unknown"));
|
||||
const code = (err as NodeJS.ErrnoException).code ?? "";
|
||||
// Same transient-teardown policy as uncaughtException: a rejection that
|
||||
// fires while a stream is being torn down (EPIPE after ffmpeg stdin
|
||||
// closes, write-after-destroy, socket reset) must NOT take the whole
|
||||
// gateway offline. Log detail + continue. Everything else still shuts
|
||||
// down so real bugs surface.
|
||||
if (
|
||||
code === "EPIPE" ||
|
||||
code === "ERR_STREAM_DESTROYED" ||
|
||||
code === "ERR_STREAM_WRITE_AFTER_END" ||
|
||||
code === "ECONNRESET"
|
||||
) {
|
||||
logger.warn(
|
||||
{ error: err },
|
||||
"Unhandled rejection transient stream error — continuing",
|
||||
);
|
||||
return;
|
||||
}
|
||||
logger.error({ error: err, reason: String(reason) }, "Unhandled rejection");
|
||||
gracefulShutdown("unhandledRejection");
|
||||
});
|
||||
|
||||
|
||||
@@ -262,6 +262,9 @@ a=ice-lite
|
||||
[audioSection, videoSection, videoRtpMap].join("\n"),
|
||||
"answer",
|
||||
);
|
||||
console.log(
|
||||
`[goLive:${this.constructor.name}] SELECT_PROTOCOL_ACK processed — remote answer set (${[audioSection, videoSection].join("\n").length}B)`,
|
||||
);
|
||||
this.emit("select_protocol_ack");
|
||||
}
|
||||
|
||||
@@ -330,7 +333,15 @@ a=ice-lite
|
||||
if (seq) this._sequenceNumber = seq;
|
||||
if (op === VoiceOpCodes.READY) {
|
||||
this.handleReady(d);
|
||||
this.setProtocols().then(() => this.ready?.(this._webRtcWrapper));
|
||||
this.setProtocols()
|
||||
.then(() => this.ready?.(this._webRtcWrapper))
|
||||
.catch((err: unknown) => {
|
||||
// PC can be closed while setProtocols is in flight (stream
|
||||
// teardown) — don't let that become an unhandledRejection.
|
||||
console.log(
|
||||
`[goLive:${this.constructor.name}] setProtocols rejected during teardown: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
});
|
||||
this.setVideoAttributes(false);
|
||||
} else if (op >= 4000) {
|
||||
console.error(`${this.constructor.name} connection error`, d);
|
||||
@@ -519,10 +530,14 @@ a=ice-lite
|
||||
const reconnect = () => {
|
||||
const webRtcConn = this._webRtcWrapper.initWebRtc();
|
||||
webRtcConn.onStateChange((state) => {
|
||||
console.log(`[goLive:${this.constructor.name}] pc state => ${state}`);
|
||||
if (state === "closed" && !this._closed) reconnect();
|
||||
});
|
||||
this._webRtcWrapper.onLocalDescription = (sdp) => {
|
||||
const rtc_connection_id = randomUUID();
|
||||
console.log(
|
||||
`[goLive:${this.constructor.name}] sending SELECT_PROTOCOL (offer ${sdp.length}B, rtc_connection_id=${rtc_connection_id.slice(0, 8)})`,
|
||||
);
|
||||
this.sendOpcode(VoiceOpCodes.SELECT_PROTOCOL, {
|
||||
protocol: "webrtc",
|
||||
codecs: Object.values(CodecPayloadType),
|
||||
@@ -532,9 +547,18 @@ a=ice-lite
|
||||
});
|
||||
};
|
||||
// createOffer (binding resolves full SDP incl. candidates after gathering)
|
||||
void webRtcConn.createOffer().then((sdp) => {
|
||||
this._webRtcWrapper.onLocalDescription?.(sdp);
|
||||
});
|
||||
void webRtcConn
|
||||
.createOffer()
|
||||
.then((sdp) => {
|
||||
this._webRtcWrapper.onLocalDescription?.(sdp);
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
// PC closed while offer is gathering (stream teardown / reconnect) —
|
||||
// swallow, the reconnect loop will start a fresh offer.
|
||||
console.log(
|
||||
`[goLive:${this.constructor.name}] createOffer rejected: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
});
|
||||
};
|
||||
reconnect();
|
||||
return new Promise((resolve) => {
|
||||
|
||||
@@ -13,12 +13,13 @@
|
||||
*/
|
||||
|
||||
import { spawn } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createWriteStream, existsSync, readdirSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { existsSync, readdirSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { PassThrough } from "node:stream";
|
||||
|
||||
/** 4-byte AnnexB start code (00 00 00 01) used when building access units. */
|
||||
const startCode4 = Buffer.from([0, 0, 0, 1]);
|
||||
|
||||
/**
|
||||
* Resolve ffmpeg/ffprobe binary. Prefers explicit env override, then PATH,
|
||||
* then a Nix-store ffmpeg-headless (the GMW flake provides it in the service
|
||||
@@ -140,121 +141,36 @@ export async function probeStreams(
|
||||
|
||||
/**
|
||||
* Demux input (URL string or readable stream) into video frames on a
|
||||
* PassThrough. Uses ffmpeg -f h264 -c copy for video-only AnnexB output.
|
||||
* Returns stream info + the video pipe. Audio is not extracted (GoLive
|
||||
* screen share sends silence / uses Discord's mixed audio).
|
||||
* PassThrough. Streams input DIRECTLY into ffmpeg (no spool-to-file — the
|
||||
* live NUT/H264 source never ends, so spooling deadlocks). ffmpeg emits
|
||||
* AnnexB H264 on stdout; NAL units are split into frames on the fly.
|
||||
* Video metadata is parsed from ffmpeg stderr during init.
|
||||
*/
|
||||
export async function demux(
|
||||
input: string | PassThrough,
|
||||
_opts: { format: string },
|
||||
opts: { format: string; frameRate?: number },
|
||||
): Promise<{
|
||||
video: DemuxedStream | undefined;
|
||||
audio: DemuxedStream | undefined;
|
||||
close: () => void;
|
||||
}> {
|
||||
const _label = randomUUID();
|
||||
const vPipe = new PassThrough({ objectMode: true, highWaterMark: 128 });
|
||||
const aPipe = new PassThrough({ objectMode: true, highWaterMark: 128 });
|
||||
|
||||
// For stream input, spool to a temp file first so ffprobe can inspect it
|
||||
// (ffprobe needs a seekable file; pipes can't be re-read). The stream is
|
||||
// fully consumed before ffmpeg starts — acceptable for screen-share
|
||||
// sources which are already fully buffered by yt-dlp in practice.
|
||||
let spoolPath: string | null = null;
|
||||
const cleanupSpool = () => {
|
||||
if (spoolPath) {
|
||||
import("node:fs").then(({ unlink }) => unlink(spoolPath!, () => {}));
|
||||
spoolPath = null;
|
||||
}
|
||||
};
|
||||
|
||||
let effectiveInput: string;
|
||||
if (typeof input === "string") {
|
||||
effectiveInput = input;
|
||||
} else {
|
||||
spoolPath = join(tmpdir(), `golive-demux-${_label}.h264`);
|
||||
const ws = createWriteStream(spoolPath);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
input.pipe(ws);
|
||||
input.on("error", reject);
|
||||
ws.on("finish", resolve);
|
||||
ws.on("error", reject);
|
||||
});
|
||||
effectiveInput = spoolPath;
|
||||
}
|
||||
|
||||
// Probe for codec + dimensions
|
||||
let streams: Array<Record<string, unknown>> = [];
|
||||
try {
|
||||
streams = await probeStreams(effectiveInput);
|
||||
} catch (_e) {
|
||||
// probe failed (e.g. raw h264 without container) — infer h264 default
|
||||
streams = [];
|
||||
}
|
||||
|
||||
const v = streams.find((s) => s.codec_type === "video");
|
||||
const a = streams.find((s) => s.codec_type === "audio");
|
||||
let vInfo: DemuxedStream | undefined;
|
||||
let aInfo: DemuxedStream | undefined;
|
||||
|
||||
if (v) {
|
||||
const codecName = (v.codec_name as string) ?? "h264";
|
||||
const rFrame = (v.r_frame_rate as string) ?? "0/1";
|
||||
const [num, den] = rFrame.split("/").map((n) => Number(n));
|
||||
vInfo = {
|
||||
codec:
|
||||
AVCodecID[
|
||||
(codecName.toUpperCase() as keyof typeof AVCodecID) ??
|
||||
"AV_CODEC_ID_H264"
|
||||
] ?? AVCodecID.AV_CODEC_ID_H264,
|
||||
codecName,
|
||||
width: (v.width as number) ?? 0,
|
||||
height: (v.height as number) ?? 0,
|
||||
framerate_num: num ?? 0,
|
||||
framerate_den: den ?? 1,
|
||||
sample_rate: 0,
|
||||
stream: vPipe,
|
||||
};
|
||||
} else {
|
||||
// Probe failed (e.g. raw AnnexB h264 input) — still emit frames on the
|
||||
// video pipe; playStream infers dimensions from the first frame.
|
||||
vInfo = {
|
||||
codec: AVCodecID.AV_CODEC_ID_H264,
|
||||
codecName: "h264",
|
||||
width: 0,
|
||||
height: 0,
|
||||
framerate_num: 0,
|
||||
framerate_den: 1,
|
||||
sample_rate: 0,
|
||||
stream: vPipe,
|
||||
};
|
||||
}
|
||||
|
||||
if (a) {
|
||||
const codecName = (a.codec_name as string) ?? "opus";
|
||||
aInfo = {
|
||||
codec:
|
||||
AVCodecID[
|
||||
(codecName.toUpperCase() as keyof typeof AVCodecID) ??
|
||||
"AV_CODEC_ID_OPUS"
|
||||
],
|
||||
codecName,
|
||||
width: 0,
|
||||
height: 0,
|
||||
framerate_num: 0,
|
||||
framerate_den: 0,
|
||||
sample_rate: Number(a.sample_rate) ?? 0,
|
||||
stream: aPipe,
|
||||
};
|
||||
}
|
||||
|
||||
// Spawn ffmpeg — extract raw video (AnnexB for H264) to stdout
|
||||
const isStream = typeof input !== "string";
|
||||
const args: string[] = [
|
||||
"-hide_banner",
|
||||
// info level: stream init lines ("Stream #0:0: Video: h264...") go to
|
||||
// stderr and are parsed for dimensions/fps.
|
||||
"-loglevel",
|
||||
"error",
|
||||
"info",
|
||||
// Input format hint: prepareStream always emits raw AnnexB H264 on
|
||||
// pipe:0. Raw H264 has NO magic header, so ffmpeg's auto-detection
|
||||
// fails with "Invalid data found when processing input" whenever the
|
||||
// first bytes arrive late/buffered. Pin the demuxer input format.
|
||||
...(isStream ? ["-f", "h264"] : []),
|
||||
"-i",
|
||||
effectiveInput,
|
||||
isStream ? "pipe:0" : input,
|
||||
"-c:v",
|
||||
"copy",
|
||||
"-an", // no audio in this minimal demuxer
|
||||
@@ -262,25 +178,165 @@ export async function demux(
|
||||
"h264",
|
||||
"pipe:1",
|
||||
];
|
||||
const proc = spawn(FFMPEG, args, {
|
||||
stdio: isStream ? ["pipe", "pipe", "pipe"] : ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
console.log(
|
||||
`[goLive:Demuxer] spawn ffmpeg pid=${proc.pid} input=${isStream ? "stream" : input} args=${args.join(" ")}`,
|
||||
);
|
||||
|
||||
const proc = spawn(FFMPEG, args, { stdio: ["ignore", "pipe", "pipe"] });
|
||||
// Pipe live input straight into ffmpeg stdin — never await stream end.
|
||||
if (isStream && proc.stdin) {
|
||||
input.pipe(proc.stdin);
|
||||
input.on("error", () => proc.stdin?.destroy());
|
||||
}
|
||||
|
||||
// Scan stdout for NAL units. Each NAL unit (between start codes) is one frame
|
||||
// payload. We emit them individually; the packetizer chain handles FU-A.
|
||||
// Set true once stderr metadata has been parsed (see handler below).
|
||||
let parsedMeta = false;
|
||||
|
||||
// Parse stream metadata from ffmpeg stderr as it arrives (first chunk has
|
||||
// the init lines). Fall back to H264 defaults if parsing fails.
|
||||
let vInfo: DemuxedStream = {
|
||||
codec: AVCodecID.AV_CODEC_ID_H264,
|
||||
codecName: "h264",
|
||||
width: 0,
|
||||
height: 0,
|
||||
framerate_num: 0,
|
||||
framerate_den: 1,
|
||||
sample_rate: 0,
|
||||
stream: vPipe,
|
||||
};
|
||||
let aInfo: DemuxedStream | undefined;
|
||||
let stderrBuf = "";
|
||||
if (proc.stderr) {
|
||||
proc.stderr.on("data", (d: Buffer) => {
|
||||
const text = d.toString();
|
||||
stderrBuf = (stderrBuf + text).slice(-16384);
|
||||
// Surface actionable lines: ffmpeg errors + stream init lines
|
||||
if (/error|invalid|no such|failed|cannot|not found|unable/i.test(text)) {
|
||||
console.log(
|
||||
`[goLive:Demuxer] ffmpeg stderr: ${text.trim().split("\n").slice(0, 4).join(" | ")}`,
|
||||
);
|
||||
}
|
||||
if (parsedMeta) return;
|
||||
const streamRe = /Stream #0:(\d+): (Video|Audio): ([^,]+)/g;
|
||||
let m: RegExpExecArray | null;
|
||||
const found: Array<{ kind: string; codecRaw: string }> = [];
|
||||
// biome-ignore lint/suspicious/noAssignInExpressions: regex loop idiom
|
||||
while ((m = streamRe.exec(stderrBuf)) !== null) {
|
||||
found.push({ kind: m[2], codecRaw: m[3] });
|
||||
}
|
||||
const v = found.find((s) => s.kind === "Video");
|
||||
const a = found.find((s) => s.kind === "Audio");
|
||||
if (!v && !a) return;
|
||||
parsedMeta = true;
|
||||
if (v) {
|
||||
const codecName = v.codecRaw.split(" ")[0].toLowerCase();
|
||||
const dim = /(\d{2,5})x(\d{2,5})/.exec(stderrBuf);
|
||||
const fps = /(\d+(?:\.\d+)?) fps/.exec(stderrBuf);
|
||||
vInfo = {
|
||||
codec:
|
||||
AVCodecID[
|
||||
(codecName.toUpperCase() as keyof typeof AVCodecID) ??
|
||||
"AV_CODEC_ID_H264"
|
||||
] ?? AVCodecID.AV_CODEC_ID_H264,
|
||||
codecName,
|
||||
width: dim ? Number(dim[1]) : 0,
|
||||
height: dim ? Number(dim[2]) : 0,
|
||||
framerate_num: fps ? Math.round(Number(fps[1]) * 1000) : 0,
|
||||
framerate_den: fps ? 1000 : 1,
|
||||
sample_rate: 0,
|
||||
stream: vPipe,
|
||||
};
|
||||
}
|
||||
if (a) {
|
||||
const codecName = a.codecRaw.split(" ")[0].toLowerCase();
|
||||
const sr = /(\d+) Hz/.exec(stderrBuf);
|
||||
aInfo = {
|
||||
codec:
|
||||
AVCodecID[
|
||||
(codecName.toUpperCase() as keyof typeof AVCodecID) ??
|
||||
"AV_CODEC_ID_OPUS"
|
||||
] ?? AVCodecID.AV_CODEC_ID_OPUS,
|
||||
codecName,
|
||||
width: 0,
|
||||
height: 0,
|
||||
framerate_num: 0,
|
||||
framerate_den: 0,
|
||||
sample_rate: sr ? Number(sr[1]) : 0,
|
||||
stream: aPipe,
|
||||
};
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Wait (briefly) for ffmpeg to print its stream init lines on stderr so
|
||||
// vInfo carries real dimensions/fps. The lines arrive with the first chunk
|
||||
// — a short timeout covers slow starts; callers fall back to sensible
|
||||
// defaults when width/height are 0 anyway.
|
||||
await Promise.race([
|
||||
new Promise<void>((resolve) => {
|
||||
const check = setInterval(() => {
|
||||
if (parsedMeta) {
|
||||
clearInterval(check);
|
||||
resolve();
|
||||
}
|
||||
}, 25);
|
||||
}),
|
||||
new Promise<void>((resolve) => setTimeout(resolve, 1500)),
|
||||
]);
|
||||
|
||||
// Scan stdout for AnnexB NAL units and group them into ACCESS UNITS
|
||||
// (one picture). Discord's H264 decoder requires a complete access unit —
|
||||
// parameter sets + slice — inside a single RTP frame. Emitting each NAL
|
||||
// as its own frame (SPS/PPS/SEI separate from the slice) makes the decoder
|
||||
// unable to produce ANY picture: production showed a black GoLive tile
|
||||
// despite frames flowing (5892B slices + 4B PPS + 33B SPS as separate
|
||||
// frames, each with a near-zero RTP timestamp delta). We therefore buffer
|
||||
// NALs and flush one frame per slice, prepending the parameter sets that
|
||||
// precede it, and timestamp it as ONE frame at the video frame rate.
|
||||
let videoBuf = Buffer.alloc(0);
|
||||
let frameCount = 0;
|
||||
let pendingNals: Buffer[] = [];
|
||||
let pendingHasSlice = false;
|
||||
let pendingIsKey = false;
|
||||
// Raw H264 streams carry no timing info — ffmpeg's h264 demuxer guesses
|
||||
// 25fps on stderr. Prefer the caller's explicit frameRate (the encode
|
||||
// setting); it drives both RTP timestamp advance and pacing.
|
||||
const videoFps =
|
||||
opts.frameRate ?? (vInfo.framerate_num / vInfo.framerate_den || 30);
|
||||
|
||||
const emitFrame = (nal: Uint8Array, isKeyFrame: boolean) => {
|
||||
const flushAccessUnit = () => {
|
||||
if (pendingNals.length === 0) return;
|
||||
// AnnexB access unit: 00 00 00 01 + NAL for every buffered NAL. The
|
||||
// packetizer (H264RtpPacketizer, StartSequence separator) needs the
|
||||
// start codes to find NAL boundaries inside the frame.
|
||||
const parts: Buffer[] = [];
|
||||
for (const n of pendingNals) parts.push(startCode4, n);
|
||||
const au = Buffer.concat(parts);
|
||||
const isKey = pendingIsKey;
|
||||
pendingNals = [];
|
||||
pendingHasSlice = false;
|
||||
pendingIsKey = false;
|
||||
vPipe.write({
|
||||
data: Buffer.from(nal),
|
||||
data: au,
|
||||
// One frame at videoFps: duration=1 in a 1/fps timebase →
|
||||
// BaseMediaStream computes frametime=1000/fps ms → the RTP timestamp
|
||||
// advances clockRate/fps per frame (3000 @ 30fps / 90kHz), which is
|
||||
// what Discord's receiver expects for real-time video.
|
||||
pts: frameCount,
|
||||
duration: 1,
|
||||
timeBase: { num: 1, den: 90000 },
|
||||
flags: isKeyFrame ? AV_PKT_FLAG_KEY : 0,
|
||||
timeBase: { num: 1, den: videoFps },
|
||||
flags: isKey ? AV_PKT_FLAG_KEY : 0,
|
||||
streamIndex: 0,
|
||||
free: () => {},
|
||||
});
|
||||
frameCount++;
|
||||
if (frameCount === 1 || frameCount % 30 === 0) {
|
||||
console.log(
|
||||
`[goLive:Demuxer] frames=${frameCount} last=${au.length}B key=${isKey}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
if (proc.stdout) {
|
||||
@@ -327,8 +383,21 @@ export async function demux(
|
||||
while (end > 0 && nal[end - 1] === 0) end--;
|
||||
if (end > 0) {
|
||||
const nalTrimmed = nal.subarray(0, end);
|
||||
const isIdr = (nalTrimmed[0] & 0x1f) === 5; // IDR
|
||||
emitFrame(nalTrimmed, isIdr);
|
||||
const nalType = nalTrimmed[0] & 0x1f;
|
||||
const isSlice = nalType === 1 || nalType === 5;
|
||||
if (isSlice) {
|
||||
// A new slice while one is pending closes the previous
|
||||
// access unit (x264 emits one slice per frame).
|
||||
if (pendingHasSlice) flushAccessUnit();
|
||||
pendingNals.push(Buffer.from(nalTrimmed));
|
||||
pendingHasSlice = true;
|
||||
if (nalType === 5) pendingIsKey = true;
|
||||
} else {
|
||||
// Parameter-set / SEI / AUD / filler NAL. After a slice these
|
||||
// belong to the NEXT access unit — flush the completed frame.
|
||||
if (pendingHasSlice) flushAccessUnit();
|
||||
pendingNals.push(Buffer.from(nalTrimmed));
|
||||
}
|
||||
}
|
||||
}
|
||||
// Skip the 00 00 01 at scPos-3 to find next
|
||||
@@ -346,21 +415,12 @@ export async function demux(
|
||||
}
|
||||
});
|
||||
proc.stdout.on("end", () => {
|
||||
if (videoBuf.length > 0) {
|
||||
let end = videoBuf.length;
|
||||
while (end > 0 && videoBuf[end - 1] === 0) end--;
|
||||
if (end > 0) emitFrame(videoBuf.subarray(0, end), false);
|
||||
}
|
||||
flushAccessUnit();
|
||||
vPipe.end();
|
||||
aPipe.end();
|
||||
});
|
||||
}
|
||||
|
||||
if (proc.stderr) {
|
||||
proc.stderr.on("data", () => {
|
||||
/* errors swallowed */
|
||||
});
|
||||
}
|
||||
proc.on("close", () => {
|
||||
vPipe.end();
|
||||
aPipe.end();
|
||||
@@ -370,7 +430,6 @@ export async function demux(
|
||||
proc.kill("SIGTERM");
|
||||
vPipe.end();
|
||||
aPipe.end();
|
||||
cleanupSpool();
|
||||
};
|
||||
|
||||
return { video: vInfo, audio: aInfo, close };
|
||||
|
||||
@@ -26,13 +26,24 @@ export function software(
|
||||
} = {},
|
||||
): () => EncoderSet {
|
||||
const { x264, x265 } = opts;
|
||||
const { preset: x264Preset = "superfast", tune: x264Tune = "film" } =
|
||||
const { preset: x264Preset = "superfast", tune: x264Tune = "zerolatency" } =
|
||||
x264 ?? {};
|
||||
const { preset: x265Preset = "superfast", tune: x265Tune } = x265 ?? {};
|
||||
return () => ({
|
||||
H264: {
|
||||
name: "libx264",
|
||||
options: ["-forced-idr 1", `-tune ${x264Tune}`, `-preset ${x264Preset}`],
|
||||
// -profile:v baseline is REQUIRED: the SDP advertises
|
||||
// profile-level-id=42e01f (constrained baseline) and Discord's
|
||||
// receiver decodes with that profile. x264's default is High — a
|
||||
// High-profile bitstream against a baseline SDP negotiation fails to
|
||||
// decode → black GoLive tile (production bug, fixed 2026-08-12).
|
||||
// zerolatency matches @dank074 (no lookahead — correct for live).
|
||||
options: [
|
||||
"-forced-idr 1",
|
||||
"-profile:v baseline",
|
||||
`-tune ${x264Tune}`,
|
||||
`-preset ${x264Preset}`,
|
||||
],
|
||||
},
|
||||
H265: {
|
||||
name: "libx265",
|
||||
|
||||
@@ -48,7 +48,19 @@ export class Streamer {
|
||||
this._client = client;
|
||||
// listen for gateway dispatch events
|
||||
this.client.on("raw", (packet) => {
|
||||
this._gatewayEmitter.emit(packet.t, packet.d);
|
||||
const t = packet.t as string;
|
||||
if (
|
||||
t === "STREAM_CREATE" ||
|
||||
t === "STREAM_SERVER_UPDATE" ||
|
||||
t === "VOICE_STATE_UPDATE" ||
|
||||
t === "VOICE_SERVER_UPDATE"
|
||||
) {
|
||||
console.log(
|
||||
`[goLive:Streamer] raw dispatch ${t}`,
|
||||
JSON.stringify(packet.d).slice(0, 220),
|
||||
);
|
||||
}
|
||||
this._gatewayEmitter.emit(t, packet.d);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -65,6 +77,11 @@ export class Streamer {
|
||||
}
|
||||
|
||||
sendOpcode(code: number, data: unknown): void {
|
||||
// Direct instrumentation — bypasses the bootstrap debug filter (which
|
||||
// drops messages without [VOICE / [ffmpeg / error / stream).
|
||||
console.log(
|
||||
`[goLive:Streamer] sendOpcode op=${code} d=${JSON.stringify(data)}`,
|
||||
);
|
||||
this.client.ws.broadcast({ op: code, d: data });
|
||||
}
|
||||
|
||||
@@ -141,7 +158,6 @@ export class Streamer {
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.signalStream();
|
||||
const {
|
||||
guildId: clientGuildId,
|
||||
channelId: clientChannelId,
|
||||
@@ -155,40 +171,85 @@ export class Streamer {
|
||||
clientUserId,
|
||||
clientChannelId,
|
||||
(conn) => {
|
||||
clearTimeout(streamTimeout);
|
||||
clearInterval(retryInterval);
|
||||
resolve(conn);
|
||||
},
|
||||
);
|
||||
this.voiceConnection.streamConnection = streamConn;
|
||||
this._gatewayEmitter.on(
|
||||
"STREAM_CREATE",
|
||||
(d: { stream_key: string; rtc_server_id: string }) => {
|
||||
const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
|
||||
if (
|
||||
clientGuildId !== guildId ||
|
||||
clientChannelId !== channelId ||
|
||||
clientUserId !== userId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
streamConn.serverId = d.rtc_server_id;
|
||||
streamConn.streamKey = d.stream_key;
|
||||
streamConn.setSession(session_id);
|
||||
},
|
||||
);
|
||||
this._gatewayEmitter.on(
|
||||
"STREAM_SERVER_UPDATE",
|
||||
(d: { stream_key: string; endpoint: string; token: string }) => {
|
||||
const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
|
||||
if (
|
||||
clientGuildId !== guildId ||
|
||||
clientChannelId !== channelId ||
|
||||
clientUserId !== userId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
streamConn.setTokens(d.endpoint, d.token);
|
||||
},
|
||||
|
||||
// Attach listeners BEFORE the first signal so a fast dispatch can't
|
||||
// be lost between signalStream() and listener registration.
|
||||
const onStreamCreate = (d: {
|
||||
stream_key: string;
|
||||
rtc_server_id: string;
|
||||
}) => {
|
||||
const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
|
||||
if (
|
||||
clientGuildId !== guildId ||
|
||||
clientChannelId !== channelId ||
|
||||
clientUserId !== userId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
streamConn.serverId = d.rtc_server_id;
|
||||
streamConn.streamKey = d.stream_key;
|
||||
streamConn.setSession(session_id);
|
||||
};
|
||||
const onStreamServerUpdate = (d: {
|
||||
stream_key: string;
|
||||
endpoint: string;
|
||||
token: string;
|
||||
}) => {
|
||||
const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
|
||||
if (
|
||||
clientGuildId !== guildId ||
|
||||
clientChannelId !== channelId ||
|
||||
clientUserId !== userId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
streamConn.setTokens(d.endpoint, d.token);
|
||||
};
|
||||
this._gatewayEmitter.on("STREAM_CREATE", onStreamCreate);
|
||||
this._gatewayEmitter.on("STREAM_SERVER_UPDATE", onStreamServerUpdate);
|
||||
|
||||
const cleanup = () => {
|
||||
clearTimeout(streamTimeout);
|
||||
clearInterval(retryInterval);
|
||||
this._gatewayEmitter.removeListener("STREAM_CREATE", onStreamCreate);
|
||||
this._gatewayEmitter.removeListener(
|
||||
"STREAM_SERVER_UPDATE",
|
||||
onStreamServerUpdate,
|
||||
);
|
||||
};
|
||||
const streamTimeout = setTimeout(() => {
|
||||
cleanup();
|
||||
reject(
|
||||
new Error(
|
||||
"Timed out waiting for STREAM_CREATE/STREAM_SERVER_UPDATE from Discord (stream handshake) — voice media session may not be active",
|
||||
),
|
||||
);
|
||||
}, 12_000);
|
||||
|
||||
// Discord sometimes drops the STREAM_CREATE request silently (upstream
|
||||
// issue #217/#219) — resend a few times instead of giving up after one.
|
||||
let attempt = 0;
|
||||
const retryInterval = setInterval(() => {
|
||||
attempt += 1;
|
||||
if (attempt >= 4) {
|
||||
clearInterval(retryInterval);
|
||||
return;
|
||||
}
|
||||
console.log(
|
||||
`[goLive:Streamer] createStream: retrying STREAM_CREATE (attempt ${attempt + 1}/4)`,
|
||||
);
|
||||
this.signalStream();
|
||||
}, 3_000);
|
||||
console.log(
|
||||
`[goLive:Streamer] createStream: sending STREAM_CREATE (attempt 1/4)`,
|
||||
);
|
||||
this.signalStream();
|
||||
});
|
||||
}
|
||||
|
||||
@@ -241,6 +302,16 @@ export class Streamer {
|
||||
channelId: channel_id,
|
||||
botId: user_id,
|
||||
} = this.voiceConnection;
|
||||
// Un-deafen before requesting the stream (mimic real client). Do NOT
|
||||
// set self_video: true — that flips on the bot's camera in Discord
|
||||
// (visible to everyone); screen share should not enable the camera.
|
||||
this.sendOpcode(GatewayOpCodes.VOICE_STATE_UPDATE, {
|
||||
guild_id,
|
||||
channel_id,
|
||||
self_mute: false,
|
||||
self_deaf: false,
|
||||
self_video: false,
|
||||
});
|
||||
this.sendOpcode(GatewayOpCodes.STREAM_CREATE, {
|
||||
type,
|
||||
guild_id,
|
||||
|
||||
@@ -68,6 +68,7 @@ export class WebRtcConnWrapper {
|
||||
private _audioTrack: NativeTrack | null = null;
|
||||
private _videoTrack: NativeTrack | null = null;
|
||||
private _videoCodec: WebRtcVideoCodec | null = null;
|
||||
private _videoFrameLog = 0;
|
||||
/** Assigned by BaseMediaConnection to send the gathered SDP to Discord. */
|
||||
onLocalDescription: ((sdp: string) => void) | null = null;
|
||||
|
||||
@@ -114,7 +115,15 @@ export class WebRtcConnWrapper {
|
||||
}
|
||||
|
||||
sendVideoFrame(frame: Buffer, frametime: number): void {
|
||||
if (!this.ready || !this._videoTrack) return;
|
||||
if (!this.ready || !this._videoTrack) {
|
||||
if (this._videoFrameLog === 0) {
|
||||
console.log(
|
||||
`[goLive:WebRtc] sendVideoFrame DROPPED ready=${this.ready} track=${this._videoTrack !== null}`,
|
||||
);
|
||||
this._videoFrameLog++;
|
||||
}
|
||||
return;
|
||||
}
|
||||
const clockRate = CodecPayloadType[this._videoCodec ?? "H264"].clockRate;
|
||||
if (this._videoCodec === "H264") {
|
||||
let spsRewritten = false;
|
||||
@@ -157,6 +166,12 @@ export class WebRtcConnWrapper {
|
||||
}
|
||||
this._videoTrack.sendFrame(frame);
|
||||
this._videoTrack.addTimestamp(Math.round((frametime * clockRate) / 1000));
|
||||
this._videoFrameLog++;
|
||||
if (this._videoFrameLog === 1 || this._videoFrameLog % 30 === 0) {
|
||||
console.log(
|
||||
`[goLive:WebRtc] sendVideoFrame #${this._videoFrameLog} bytes=${frame.length} ready=${this.ready}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
setPacketizer(videoCodec: string): void {
|
||||
|
||||
@@ -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);
|
||||
@@ -262,15 +272,24 @@ export async function playStream(
|
||||
options: PlayStreamOptions = {},
|
||||
): Promise<void> {
|
||||
const conn = await streamer.createStream();
|
||||
console.log("[goLive:playStream] createStream resolved");
|
||||
|
||||
const { video, close: demuxClose } = await demux(prepared.output, {
|
||||
format: options.format ?? "nut",
|
||||
frameRate:
|
||||
typeof options.frameRate === "number" ? options.frameRate : undefined,
|
||||
});
|
||||
console.log(
|
||||
`[goLive:playStream] demux done codec=${video?.codecName ?? "?"} ${video?.width ?? 0}x${video?.height ?? 0} fps=${video ? video.framerate_num / video.framerate_den || 30 : 30}`,
|
||||
);
|
||||
|
||||
if (!video) throw new Error("No video stream in media");
|
||||
|
||||
conn.setPacketizer(video.codecName);
|
||||
conn.mediaConnection.setSpeaking(true);
|
||||
console.log(
|
||||
`[goLive:playStream] setPacketizer(${video.codecName}) + setSpeaking done`,
|
||||
);
|
||||
|
||||
const w =
|
||||
typeof options.width === "function"
|
||||
@@ -310,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);
|
||||
})();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -79,6 +79,22 @@ export function transcodeToHighQualityOgg(
|
||||
);
|
||||
|
||||
input.pipe(proc.stdin);
|
||||
// ffmpeg teardown closes stdin while the upstream source may still write —
|
||||
// swallow EPIPE / destroyed-stream errors so they don't crash the gateway.
|
||||
proc.stdin.on("error", (err: NodeJS.ErrnoException) => {
|
||||
if (
|
||||
err.code === "EPIPE" ||
|
||||
err.code === "ERR_STREAM_DESTROYED" ||
|
||||
err.code === "ERR_STREAM_WRITE_AFTER_END"
|
||||
) {
|
||||
logger.debug(
|
||||
{ code: err.code },
|
||||
"Transcode stdin closed during teardown",
|
||||
);
|
||||
} else {
|
||||
logger.error({ error: err.message }, "Transcode stdin error");
|
||||
}
|
||||
});
|
||||
activeProcesses.add(proc);
|
||||
|
||||
const cleanup = () => {
|
||||
@@ -248,6 +264,20 @@ export function resolveMediaUrl(
|
||||
// `--print` headers to stderr — pipe stdout immediately so the child
|
||||
// never blocks on a full pipe while we wait for the headers on stderr.
|
||||
const mediaStream = new PassThrough();
|
||||
// Teardown (player stop / ffmpeg exit) destroys this stream while
|
||||
// yt-dlp may still push bytes — without a listener an EPIPE /
|
||||
// ERR_STREAM_DESTROYED surfaces as an uncaughtException.
|
||||
mediaStream.on("error", (err: NodeJS.ErrnoException) => {
|
||||
if (
|
||||
err.code === "EPIPE" ||
|
||||
err.code === "ERR_STREAM_DESTROYED" ||
|
||||
err.code === "ERR_STREAM_WRITE_AFTER_END"
|
||||
) {
|
||||
logger.debug({ code: err.code }, "Media stream closed during teardown");
|
||||
} else {
|
||||
logger.error({ error: err.message }, "Media stream error");
|
||||
}
|
||||
});
|
||||
proc.stdout.pipe(mediaStream);
|
||||
|
||||
let stderrBuf = "";
|
||||
@@ -492,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" &&
|
||||
@@ -499,7 +535,7 @@ async function resolveScreenInput(
|
||||
typeof audioUrl === "string" &&
|
||||
audioUrl.length > 0
|
||||
) {
|
||||
return mergeScreenStreams(videoUrl, audioUrl);
|
||||
return mergeScreenStreams(videoUrl, audioUrl, videoHeaders);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -512,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", () => {
|
||||
@@ -562,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) => {
|
||||
@@ -576,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);
|
||||
}
|
||||
@@ -105,7 +214,11 @@ export class ScreenShareController {
|
||||
frameRate: 30,
|
||||
bitrateVideo: 2500,
|
||||
bitrateVideoMax: 4000,
|
||||
includeAudio: true,
|
||||
// Video-only GoLive: the -f h264 output muxer cannot carry audio
|
||||
// ("h264 muxer does not support any stream of type audio" → header
|
||||
// write fails → empty stdout → demux 'Invalid data' → black tile).
|
||||
// Audio is not delivered by the GoLive demux path anyway.
|
||||
includeAudio: false,
|
||||
videoCodec: normalizeVideoCodec("H264"),
|
||||
});
|
||||
const { command } = prepared;
|
||||
@@ -145,6 +258,9 @@ export class ScreenShareController {
|
||||
};
|
||||
const done = playStream(prepared, this.streamer, {
|
||||
type: "go-live",
|
||||
width: 1280,
|
||||
height: 720,
|
||||
frameRate: 30,
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
// Never let a stream failure become an unhandledRejection — that
|
||||
|
||||
@@ -56,6 +56,24 @@ export class VoiceTransmitter {
|
||||
// Create PCM input stream
|
||||
this.pcmStream = new PassThrough();
|
||||
this.pcmStream.setMaxListeners(32); // drain listeners accumulate during backpressure
|
||||
// Voice teardown (stop / disconnect / ffmpeg exit) destroys this stream
|
||||
// while Redis PCM messages may still be in flight. Without a listener,
|
||||
// EPIPE / ERR_STREAM_DESTROYED / ERR_STREAM_WRITE_AFTER_END surface as
|
||||
// an uncaughtException and crash the whole gateway.
|
||||
this.pcmStream.on("error", (err: NodeJS.ErrnoException) => {
|
||||
if (
|
||||
err.code === "EPIPE" ||
|
||||
err.code === "ERR_STREAM_DESTROYED" ||
|
||||
err.code === "ERR_STREAM_WRITE_AFTER_END"
|
||||
) {
|
||||
logger.debug(
|
||||
{ code: err.code },
|
||||
"PCM stream closed during voice teardown — ignoring",
|
||||
);
|
||||
} else {
|
||||
logger.error({ error: err.message }, "PCM stream error");
|
||||
}
|
||||
});
|
||||
|
||||
// Spawn FFmpeg to encode 24kHz mono PCM → OggOpus
|
||||
// Input: 24kHz mono s16le (raw PCM)
|
||||
@@ -146,7 +164,12 @@ export class VoiceTransmitter {
|
||||
);
|
||||
|
||||
this.redisSub.on("message", (channel, message) => {
|
||||
if (channel !== this.TRANSMIT_CHANNEL || !this.pcmStream) return;
|
||||
if (
|
||||
!this.isActive ||
|
||||
channel !== this.TRANSMIT_CHANNEL ||
|
||||
!this.pcmStream
|
||||
)
|
||||
return;
|
||||
|
||||
try {
|
||||
const data = JSON.parse(message);
|
||||
@@ -161,11 +184,21 @@ export class VoiceTransmitter {
|
||||
this.draining = false;
|
||||
// Re-acquire stream reference (could have been replaced by restart)
|
||||
const currentStream = this.pcmStream;
|
||||
if (!currentStream) return;
|
||||
if (!currentStream || !this.isActive) return;
|
||||
// Flush queued chunks
|
||||
while (this.backpressureQueue.length > 0) {
|
||||
const queued = this.backpressureQueue.shift()!;
|
||||
if (!currentStream.write(queued)) break;
|
||||
try {
|
||||
if (!currentStream.write(queued)) break;
|
||||
} catch (err) {
|
||||
logger.debug(
|
||||
{
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
},
|
||||
"PCM flush write failed during teardown — ignoring",
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -22,11 +22,14 @@ describe("goLive port: codec + encoders", () => {
|
||||
expect(normalizeVideoCodec("av1")).toBe("AV1");
|
||||
});
|
||||
|
||||
it("software encoder exposes x264 libx264 superfast film", () => {
|
||||
it("software encoder exposes x264 libx264 baseline zerolatency", () => {
|
||||
const enc = Encoders.software()();
|
||||
expect(enc.H264.name).toBe("libx264");
|
||||
expect(enc.H264.options).toContain("-preset superfast");
|
||||
expect(enc.H264.options).toContain("-tune film");
|
||||
expect(enc.H264.options).toContain("-tune zerolatency");
|
||||
// Baseline profile is REQUIRED to match the SDP's profile-level-id=42e01f
|
||||
// (constrained baseline) — High-profile bitstreams fail to decode → black
|
||||
expect(enc.H264.options).toContain("-profile:v baseline");
|
||||
});
|
||||
|
||||
it("CodecPayloadType has opus + H264 entries", () => {
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
// Regression test: demux must emit frames from a LIVE stream that never
|
||||
// ends (the NUT/H264 merge output during playback). The old implementation
|
||||
// spooled the whole stream to a file first → deadlocked forever → 0 frames.
|
||||
// v2: also validates ACCESS-UNIT grouping — each emitted frame must be a
|
||||
// complete picture (parameter sets + slice), never a bare SPS/PPS/SEI NAL,
|
||||
// and must be timestamped at the video frame rate (RTP +clockRate/fps).
|
||||
// Run: npx tsx tests/golive-demux-live-e2e.ts [ffmpeg-path]
|
||||
import { spawn } from "node:child_process";
|
||||
import { PassThrough } from "node:stream";
|
||||
import { demux } from "../src/goLive/Demuxer.js";
|
||||
|
||||
const FFMPEG = process.argv[2] ?? "ffmpeg";
|
||||
|
||||
// 1) Generate a 2s H264 test clip to a temp file
|
||||
const clip = "/tmp/golive-live-test.h264";
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const p = spawn(
|
||||
FFMPEG,
|
||||
[
|
||||
"-hide_banner", "-loglevel", "error",
|
||||
"-f", "lavfi", "-i", "testsrc=size=640x360:rate=30:duration=2",
|
||||
"-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p",
|
||||
"-f", "h264", clip,
|
||||
],
|
||||
{ stdio: ["ignore", "ignore", "pipe"] },
|
||||
);
|
||||
let err = "";
|
||||
p.stderr?.on("data", (d: Buffer) => (err += d.toString()));
|
||||
p.on("close", (code) => (code === 0 ? resolve() : reject(new Error(err))));
|
||||
});
|
||||
|
||||
// 2) Feed the clip through a PassThrough but DON'T end it (live semantics),
|
||||
// with a small pause after the first chunk so demux has time to emit.
|
||||
const input = new PassThrough();
|
||||
const demuxPromise = demux(input, { format: "h264", frameRate: 30 });
|
||||
const { video, close } = await demuxPromise;
|
||||
if (!video) {
|
||||
console.error("FAIL: demux returned no video stream");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
interface Emitted {
|
||||
data: Buffer;
|
||||
duration: number;
|
||||
timeBase: { num: number; den: number };
|
||||
flags: number;
|
||||
}
|
||||
const frames: Emitted[] = [];
|
||||
video.stream.on("data", (frame: Emitted) => {
|
||||
frames.push(frame);
|
||||
});
|
||||
|
||||
const fs = await import("node:fs");
|
||||
const buf = fs.readFileSync(clip);
|
||||
const chunkSize = 16384;
|
||||
for (let i = 0; i < buf.length; i += chunkSize) {
|
||||
input.write(buf.subarray(i, i + chunkSize));
|
||||
if (i === 0) await new Promise((r) => setTimeout(r, 1500));
|
||||
}
|
||||
// Stream still open — if the old spool logic was here we'd never emit.
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
|
||||
// 3) Validate access-unit structure
|
||||
const nalTypes = (frame: Buffer): number[] => {
|
||||
const out: number[] = [];
|
||||
let i = 0;
|
||||
while (i < frame.length - 3) {
|
||||
if (frame[i] === 0 && frame[i + 1] === 0 && frame[i + 2] === 1) {
|
||||
const start = i;
|
||||
let j = i + 3;
|
||||
if (frame[j - 4] === 0 && j >= 4) {
|
||||
// 4-byte start code already consumed by i pointing at the 3-byte tail
|
||||
}
|
||||
while (j < frame.length - 3) {
|
||||
if (frame[j] === 0 && frame[j + 1] === 0 && frame[j + 2] === 1) break;
|
||||
j++;
|
||||
}
|
||||
const nal = frame.subarray(start + 3, j);
|
||||
if (nal.length > 0) out.push(nal[0] & 0x1f);
|
||||
i = j;
|
||||
} else {
|
||||
i++;
|
||||
}
|
||||
}
|
||||
return out;
|
||||
};
|
||||
|
||||
let bareParamSetFrames = 0;
|
||||
let framesWithoutSlice = 0;
|
||||
let keyframesWithParamSets = 0;
|
||||
let keyframesWithoutParamSets = 0;
|
||||
for (const f of frames) {
|
||||
const types = nalTypes(f.data);
|
||||
const hasSlice = types.some((t) => t === 1 || t === 5);
|
||||
const hasParams = types.some((t) => t === 7 || t === 8);
|
||||
const isKey = (f.flags & 1) !== 0;
|
||||
if (!hasSlice) framesWithoutSlice++;
|
||||
if (types.length === 1 && (types[0] === 7 || types[0] === 8 || types[0] === 6)) {
|
||||
bareParamSetFrames++;
|
||||
}
|
||||
if (isKey && hasParams) keyframesWithParamSets++;
|
||||
if (isKey && !hasParams) keyframesWithoutParamSets++;
|
||||
}
|
||||
|
||||
console.log(
|
||||
`metadata: ${video.codecName} ${video.width}x${video.height} fps=${video.framerate_num}/${video.framerate_den}`,
|
||||
);
|
||||
console.log(`frames while stream OPEN (not ended): ${frames.length}`);
|
||||
console.log(`frames w/o slice NAL: ${framesWithoutSlice}, bare param-set frames: ${bareParamSetFrames}`);
|
||||
console.log(`keyframes with SPS/PPS: ${keyframesWithParamSets}, without: ${keyframesWithoutParamSets}`);
|
||||
if (frames.length === 0) {
|
||||
console.error("FAIL: no frames emitted while input still open (deadlock)");
|
||||
close();
|
||||
process.exit(1);
|
||||
}
|
||||
if (bareParamSetFrames > 0 || framesWithoutSlice > 0) {
|
||||
console.error("FAIL: demux emitted bare parameter-set frames (must group into access units)");
|
||||
close();
|
||||
process.exit(1);
|
||||
}
|
||||
if (frames.some((f) => f.duration !== 1 || f.timeBase.den !== 30)) {
|
||||
console.error("FAIL: frame duration/timeBase not 1/30 (RTP timestamp advance wrong)");
|
||||
close();
|
||||
process.exit(1);
|
||||
}
|
||||
input.end();
|
||||
await new Promise((r) => setTimeout(r, 300));
|
||||
close();
|
||||
console.log("PASS: live stream demux works + access units grouped correctly");
|
||||
process.exit(0);
|
||||
@@ -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