From 91c7a67d2fc1f6439cad332b80b11de188dcdfde Mon Sep 17 00:00:00 2001 From: asepharyana Date: Tue, 11 Aug 2026 21:43:16 +0700 Subject: [PATCH] fix(goLive): stream demux directly instead of spool-to-file (empty screen share) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of 'tile appears but content empty': demux() spooled the live NUT/H264 input to a temp file and awaited stream 'finish' — but the merge ffmpeg output never ends during playback, so demux deadlocked, no probe, no transcode, 0 frames sent. - Demuxer: pipe input straight into ffmpeg stdin (-i pipe:0), parse NAL frames live from stdout; parse video metadata from ffmpeg stderr with a 1.5s race (fall back to H264 defaults). No spool, no await-end. - screenShareController: pass width/height/frameRate (1280x720@30) to playStream — matches the prepareStream encode settings, so setVideoAttributes gets real dimensions even when ffmpeg can't report metadata on an open pipe. - Add tests/golive-demux-live-e2e.ts: proves frames flow while input is still open (regression test for the deadlock). --- .../discord-gateway/src/goLive/Demuxer.ts | 215 +++++++++--------- .../voice-recording/screenShareController.ts | 3 + .../tests/golive-demux-live-e2e.ts | 63 +++++ 3 files changed, 172 insertions(+), 109 deletions(-) create mode 100644 services/discord-gateway/tests/golive-demux-live-e2e.ts diff --git a/services/discord-gateway/src/goLive/Demuxer.ts b/services/discord-gateway/src/goLive/Demuxer.ts index 919b1e6..6e94dd9 100644 --- a/services/discord-gateway/src/goLive/Demuxer.ts +++ b/services/discord-gateway/src/goLive/Demuxer.ts @@ -13,9 +13,7 @@ */ 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"; @@ -140,9 +138,10 @@ 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, @@ -152,109 +151,18 @@ export async function demux( 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((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> = []; - 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", "-i", - effectiveInput, + isStream ? "pipe:0" : input, "-c:v", "copy", "-an", // no audio in this minimal demuxer @@ -262,8 +170,103 @@ export async function demux( "h264", "pipe:1", ]; + const proc = spawn(FFMPEG, args, { + stdio: isStream ? ["pipe", "pipe", "pipe"] : ["ignore", "pipe", "pipe"], + }); - 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()); + } + + // 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) => { + stderrBuf = (stderrBuf + d.toString()).slice(-16384); + 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((resolve) => { + const check = setInterval(() => { + if (parsedMeta) { + clearInterval(check); + resolve(); + } + }, 25); + }), + new Promise((resolve) => setTimeout(resolve, 1500)), + ]); // 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. @@ -356,11 +359,6 @@ export async function demux( }); } - if (proc.stderr) { - proc.stderr.on("data", () => { - /* errors swallowed */ - }); - } proc.on("close", () => { vPipe.end(); aPipe.end(); @@ -370,7 +368,6 @@ export async function demux( proc.kill("SIGTERM"); vPipe.end(); aPipe.end(); - cleanupSpool(); }; return { video: vInfo, audio: aInfo, close }; diff --git a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts index 540642d..2916aad 100644 --- a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts +++ b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts @@ -145,6 +145,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 diff --git a/services/discord-gateway/tests/golive-demux-live-e2e.ts b/services/discord-gateway/tests/golive-demux-live-e2e.ts new file mode 100644 index 0000000..613f9de --- /dev/null +++ b/services/discord-gateway/tests/golive-demux-live-e2e.ts @@ -0,0 +1,63 @@ +// 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. +// 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((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" }); +const { video, close } = await demuxPromise; + +let frames = 0; +let bytes = 0; +video.stream.on("data", (frame: { data: Buffer }) => { + frames++; + bytes += frame.data.length; +}); + +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)); + +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}, bytes: ${bytes}`); +if (frames === 0) { + console.error("FAIL: no frames emitted while input still open (deadlock)"); + close(); + process.exit(1); +} +input.end(); +await new Promise((r) => setTimeout(r, 300)); +close(); +console.log("PASS: live stream demux works"); +process.exit(0);