MythEclipse/GMW
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(goLive): download screen-share media to file before play (not live pipe)

Browse Source 
The live pipe (yt-dlp -o - -> ffmpeg) delivers data at network speed with
unreliable PTS, which defeats ffmpeg -re and made x264 -r 30 force-duplicate
held frames -> ~1fps video (the patah-patah symptom). Per user suggestion,
download the FULL clip to a temp file first (downloadScreenInput), then feed
that FILE PATH to prepareStream. String inputs already get -re, so the
encoder now paces cleanly at 1x against a monotonic-PTS file — proven
reliable in local tests (vs the live pipe which always bursted). Temp file
is removed on stream end / stop.

- getDirectScreenInput -> downloadScreenInput (returns file path)
- resolveInputWithRetry now awaits a completed file + retries on failure
- screenShareController.stops/cleanup removes the per-run tmpdir
- screenShareInput.test.ts updated to the file-download contract
This commit is contained in:
asepharyana
2026-08-13 16:42:41 +07:00
parent 89f1097729
commit f156fc0c9e
3 changed files with 143 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/discord-gateway/src/modules/voice-recording/mediaSource.ts
+63 -50
View File
@@ -3,7 +3,9 @@ import {
chmodSync, chmodSync,
existsSync, existsSync,
mkdtempSync, mkdtempSync,
readdirSync,
rmSync, rmSync,
statSync,
writeFileSync, writeFileSync,
} from "node:fs"; } from "node:fs";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
@@ -478,27 +480,24 @@ export function resolveMediaUrl(
} }
/** /**
* Resolve a media URL to a single playable input stream for screen share / * Download the merged video+audio media for screen share to a TEMP FILE
* GoLive streaming. * first, then return the file path.
* *
* Streams the merged video+audio media directly from yt-dlp stdout (`-o -`). * Screen-share playback does NOT stream from yt-dlp stdout anymore: the
* merge feed is delivered at network speed (bursts + stalls), and a live
* pipe defeats ffmpeg's `-re` throttle (the input PTS timeline is
* unreliable), so the encoder force-duplicates held frames → ~1fps video.
* Downloading the FULL clip to a file first gives the encoder a clean,
* monotonic-PTS input, where `-re` (applied in prepareStream for string
* inputs) pacing is proven reliable.
* *
* This is deliberately NOT the old --dump-single-json + manual URL-fetch * @returns absolute path of the completed media file (caller should delete
* approach: YouTube signs DASH URLs for the extracting client and rejects * it via cleanup after playback ends).
* them with 403 when fetched raw by ffmpeg/curl (verified 2026-08-12: even
* curl with the EXACT http_headers from the yt-dlp dump got 403 on some
* videos, while yt-dlp's own downloader succeeded). Streaming from yt-dlp
* lets it handle auth, cookies and transient retries internally — the same
* mechanism resolveMediaUrl already uses for music playback.
*
* @returns a Readable of the merged media stream.
*/ */
export function getDirectScreenInput(url: string): Promise<Readable> { export function downloadScreenInput(url: string): Promise<string> {
return new Promise<Readable>((resolve) => { return new Promise<string>((resolve, reject) => {
// Merge fragments must NOT be written to the process CWD — the Nix // Temp dir per run (world-writable like /tmp) so parallel/retry runs
// store dir is read-only for the deployed gateway (EACCES). Use a // never collide on merge fragments.
// per-run temp dir (world-writable like /tmp) so parallel/retry runs
// never collide on merge fragments and any user can write to it.
const tmpDir = mkdtempSync(join(tmpdir(), "gmw-ytdlp-")); const tmpDir = mkdtempSync(join(tmpdir(), "gmw-ytdlp-"));
chmodSync(tmpDir, 0o1777); chmodSync(tmpDir, 0o1777);
@@ -506,18 +505,18 @@ export function getDirectScreenInput(url: string): Promise<Readable> {
const args = [ const args = [
"-f", "-f",
"bestvideo[protocol^=http]+bestaudio[protocol^=http]/best[protocol^=http]/best", "bestvideo[protocol^=http]+bestaudio[protocol^=http]/best[protocol^=http]/best",
// File output (NOT `-o -`): yt-dlp merges DASH fragments into a real
// container with clean timestamps, which is what -re needs to pace.
"-o", "-o",
"-", join(tmpDir, "media.%(ext)s"),
"--no-playlist", "--no-playlist",
"--no-warnings", "--no-warnings",
"--no-progress", "--no-progress",
...cookieArgs, ...cookieArgs,
"-P",
tmpDir,
url, url,
]; ];
logger.info({ url }, "Spawning yt-dlp for screen share input streaming"); logger.info({ url }, "Downloading full media for screen share input");
const proc = spawn("yt-dlp", args, { const proc = spawn("yt-dlp", args, {
stdio: ["ignore", "pipe", "pipe"], stdio: ["ignore", "pipe", "pipe"],
@@ -525,9 +524,6 @@ export function getDirectScreenInput(url: string): Promise<Readable> {
activeProcesses.add(proc); activeProcesses.add(proc);
const stream = new PassThrough();
proc.stdout.pipe(stream);
let stderrBuf = ""; let stderrBuf = "";
const MAX_STDERR = 4096; const MAX_STDERR = 4096;
proc.stderr?.on("data", (chunk: Buffer) => { proc.stderr?.on("data", (chunk: Buffer) => {
@@ -536,41 +532,58 @@ export function getDirectScreenInput(url: string): Promise<Readable> {
} }
}); });
let producedData = false; let settled = false;
stream.once("data", () => { const failOnce = (message: string) => {
producedData = true; if (settled) return;
}); settled = true;
activeProcesses.delete(proc);
rmSync(tmpDir, { recursive: true, force: true });
reject(new Error(message));
};
proc.on("error", (err: NodeJS.ErrnoException) => { proc.on("error", (err: NodeJS.ErrnoException) => {
activeProcesses.delete(proc); failOnce(`yt-dlp failed to start: ${err.message}`);
rmSync(tmpDir, { recursive: true, force: true });
if (err.code === "ENOENT") {
stream.destroy(buildNotInstalledError());
} else {
stream.destroy(new Error(`yt-dlp failed to start: ${err.message}`));
}
}); });
let procFinished = false;
proc.on("close", (code) => { proc.on("close", (code) => {
procFinished = true;
activeProcesses.delete(proc); activeProcesses.delete(proc);
rmSync(tmpDir, { recursive: true, force: true }); if (code !== 0) {
// Fail fast: a download that dies before producing ANY bytes (e.g.
// transient YouTube 403) cannot feed the encoder — destroy the stream
// so the caller retries with a fresh yt-dlp run instead of streaming
// a silent black tile.
if (code !== 0 && !producedData && !stream.destroyed) {
const detail = stderrBuf.trim() ? `: ${stderrBuf.trim()}` : ""; const detail = stderrBuf.trim() ? `: ${stderrBuf.trim()}` : "";
stream.destroy( failOnce(`yt-dlp download failed (exit ${code})${detail}`);
new Error( return;
`yt-dlp screen input stream failed (exit ${code})${detail}`, }
), // Find the media file yt-dlp wrote (skip .part / .ytdl temp state).
); let mediaPath: string | null = null;
for (const entry of readdirSync(tmpDir)) {
if (entry.endsWith(".part") || entry.endsWith(".ytdl")) continue;
mediaPath = join(tmpDir, entry);
break;
}
if (mediaPath && existsSync(mediaPath) && statSync(mediaPath).size > 0) {
settled = true;
resolve(mediaPath);
} else {
failOnce("yt-dlp finished but produced no media file");
} }
}); });
// Resolve immediately — data flows as yt-dlp downloads. The caller's // Safety net: a stalled download must not hang the gateway forever.
// resolveInputWithRetry validates the first byte and retries on failure. const timer = setTimeout(
resolve(stream); () => {
try {
proc.kill("SIGTERM");
} catch {
/* already dead */
}
if (!procFinished) {
failOnce("yt-dlp download timed out (10 min)");
}
},
10 * 60 * 1000,
);
proc.once("close", () => clearTimeout(timer));
}); });
} }
services/discord-gateway/src/modules/voice-recording/screenShareController.ts
+34 -79
View File
@@ -1,4 +1,5 @@
import { PassThrough, type Readable } from "node:stream"; import { rmSync } from "node:fs";
import { dirname } from "node:path";
import type { Client } from "discord.js-selfbot-v13"; import type { Client } from "discord.js-selfbot-v13";
import { createChildLogger } from "@/shared/logger/index"; import { createChildLogger } from "@/shared/logger/index";
import { import {
@@ -9,7 +10,7 @@ import {
Streamer, Streamer,
} from "../../goLive/index.js"; } from "../../goLive/index.js";
import { import {
getDirectScreenInput, downloadScreenInput,
INVIDIOUS_INSTANCES, INVIDIOUS_INSTANCES,
isYoutubeWatchUrl, isYoutubeWatchUrl,
toInvidiousUrl, toInvidiousUrl,
@@ -61,17 +62,17 @@ export class ScreenShareController {
} }
/** /**
* Resolve the screen-share input with retry + first-byte validation. * Resolve the screen-share input with retry (downloads the FULL media to a
* temp file; returns the file path).
* *
* Transient YouTube 403s kill the merge ffmpeg BEFORE it produces any * Transient YouTube 403s kill a download BEFORE completion; we validate
* output; without validation the stream would "start" with a dead input * the finished file and retry with a FRESH yt-dlp run (signed DASH URLs
* and show a black tile forever. So after getDirectScreenInput resolves we * expire quickly). Downloading to a file (instead of streaming the merge
* tee the stream through a PassThrough and wait for the FIRST readable * pipe) gives the encoder a monotonic-PTS input, so ffmpeg `-re` pacing
* byte (or an error / early EOF). On failure the whole resolution is * in prepareStream actually works (it does NOT on live pipes — the root
* retried with a FRESH yt-dlp run (signed DASH URLs expire quickly — the * of the ~1fps video).
* old URLs cannot simply be re-fetched).
*/ */
private async resolveInputWithRetry(source: string): Promise<Readable> { private async resolveInputWithRetry(source: string): Promise<string> {
const MAX_ATTEMPTS = 3; const MAX_ATTEMPTS = 3;
let lastError: Error | null = null; let lastError: Error | null = null;
@@ -99,73 +100,12 @@ export class ScreenShareController {
} }
try { try {
const input = await getDirectScreenInput(source); const mediaPath = await downloadScreenInput(source);
this.logger.info(
const tee = new PassThrough(); { mediaPath, attempt },
input.on("error", (err) => tee.destroy(err)); "Screen input downloaded to file",
input.on("end", () => tee.end()); );
input.pipe(tee); return mediaPath;
// 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();
// Listeners were just removed by cleanup() — destroying tee WITH
// an error would emit "error" on an unlistened PassThrough and
// surface as an unhandled 'error' event (crash). Destroy
// silently; the error lives in the rejection only.
tee.destroy();
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);
});
// Safety net: cleanup() removes the once() listeners on timeout/error,
// but a late error event from input.pipe(tee) can still fire on an
// unlistened PassThrough and crash the gateway (unhandled 'error').
// A permanent no-op listener guarantees the event is always swallowed.
tee.on("error", () => {});
// Pass the tee onward — the encoder consumes the same buffered
// stream, so no data from the merge is lost.
return tee;
} catch (err) { } catch (err) {
lastError = err instanceof Error ? err : new Error(String(err)); lastError = err instanceof Error ? err : new Error(String(err));
this.logger.warn( this.logger.warn(
@@ -174,7 +114,7 @@ export class ScreenShareController {
maxAttempts: MAX_ATTEMPTS, maxAttempts: MAX_ATTEMPTS,
error: lastError.message, error: lastError.message,
}, },
"Screen input resolution failed; retrying with fresh yt-dlp", "Screen input download failed; retrying with fresh yt-dlp",
); );
if (attempt < MAX_ATTEMPTS) { if (attempt < MAX_ATTEMPTS) {
await new Promise((r) => setTimeout(r, 1500 * attempt)); await new Promise((r) => setTimeout(r, 1500 * attempt));
@@ -250,6 +190,19 @@ export class ScreenShareController {
}); });
const { command } = prepared; const { command } = prepared;
// The downloaded temp media lives in a per-run tmpdir; remove it once
// playback is done (natural end, failure, or user stop). The tmpdir is
// the parent of the media file, so deleting it removes the file too.
const cleanupTempMedia = () => {
try {
if (typeof input === "string") {
rmSync(dirname(input), { recursive: true, force: true });
}
} catch {
/* best-effort — tmp dirs are world-writable, leak is bounded */
}
};
let stopped = false; let stopped = false;
// Restore the @discordjs/voice connection after the stream ends (both // Restore the @discordjs/voice connection after the stream ends (both
// natural end and failure), so the user can keep using audio/mic. // natural end and failure), so the user can keep using audio/mic.
@@ -262,6 +215,7 @@ export class ScreenShareController {
/* already dead */ /* already dead */
} }
} }
cleanupTempMedia();
try { try {
this.streamer?.voiceConnection?.stop(); this.streamer?.voiceConnection?.stop();
} catch { } catch {
@@ -312,6 +266,7 @@ export class ScreenShareController {
} catch { } catch {
/* already dead */ /* already dead */
} }
cleanupTempMedia();
// Leave the voice channel the Streamer joined (its own connection). // Leave the voice channel the Streamer joined (its own connection).
try { try {
this.streamer?.voiceConnection?.stop(); this.streamer?.voiceConnection?.stop();
services/discord-gateway/tests/screenShareInput.test.ts
+46 -62
View File
@@ -1,12 +1,14 @@
// ═══════════════════════════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════════════════════════
// Screen share input resolution tests // Screen share input resolution tests
// //
// getDirectScreenInput now streams the merged video+audio media straight from // downloadScreenInput downloads the FULL merged video+audio media to a temp
// yt-dlp stdout (`-o -`) — same auth-handling mechanism as resolveMediaUrl for // file first (yt-dlp `-o <tmpdir>/media.%(ext)s`), then returns the file
// music. There is no manual URL fetch or local ffmpeg merge anymore. // path. Feeding a FILE path (not a live stdout pipe) to prepareStream is
// what makes ffmpeg `-re` pacing reliable — a pipe has unreliable PTS and
// caused the ~1fps force-duplication symptom.
// //
// yt-dlp is faked via a PATH shim script so the test does not hit the network // yt-dlp is faked via a PATH shim script so the test does not hit the
// or need real binaries. // network or need real binaries.
// ═══════════════════════════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════════════════════════
import { import {
@@ -18,34 +20,41 @@ import {
} from "node:fs"; } from "node:fs";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
import { join } from "node:path"; import { join } from "node:path";
import { Readable } from "node:stream";
import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { getDirectScreenInput } from "../src/modules/voice-recording/mediaSource.js"; import { downloadScreenInput } from "../src/modules/voice-recording/mediaSource.js";
// ─── fake bin dir ────────────────────────────────────────────────────────────── // ─── fake bin dir ──────────────────────────────────────────────────────────────
let fakeBinDir: string | null = null; let fakeBinDir: string | null = null;
const realPath = process.env.PATH; const realPath = process.env.PATH;
beforeAll(() => { // Bash shim: find the -o pattern, substitute the extension, write 4096 bytes.
fakeBinDir = mkdtempSync(join(tmpdir(), "gmw-fake-bins-")); // Escaped as a separate string to keep the TS template literal simple.
const ytShimBody = `prev=""
// Fake yt-dlp: streams a few bytes to stdout (like `yt-dlp -o -` does). out=""
// Modes (env): for a in "$@"; do
// GMW_FAKE_YTDLP_FAIL=1 → stderr 403 + exit 8 WITHOUT stdout bytes if [ "$prev" = "-o" ]; then out="$a"; fi
// (mimics a download rejected by YouTube). prev="$a"
const ytShim = `#!/usr/bin/env bash done
if [ "$GMW_FAKE_YTDLP_FAIL" = "1" ]; then if [ "$GMW_FAKE_YTDLP_FAIL" = "1" ]; then
echo "ERROR: [youtube] ...: 403 Forbidden (access denied)" >&2 echo "ERROR: [youtube] ...: 403 Forbidden (access denied)" >&2
exit 8 exit 8
fi fi
# Fake yt-dlp — ignore args, emit a few bytes so consumers see a live stream. target=$(printf '%s' "$out" | sed 's/%(ext)s/.mp4/')
head -c 4096 /dev/urandom head -c 4096 /dev/urandom > "$target"
exit 0 exit 0
`; `;
const ytShim = `#!/usr/bin/env bash\n${ytShimBody}`;
const ytShimDump = `#!/usr/bin/env bash
printf '%s\\n' "$*" >> "$GMW_FAKE_YTDLP_DUMP_ARGS"
${ytShimBody}
`;
beforeAll(() => {
fakeBinDir = mkdtempSync(join(tmpdir(), "gmw-fake-bins-"));
writeFileSync(join(fakeBinDir, "yt-dlp"), ytShim); writeFileSync(join(fakeBinDir, "yt-dlp"), ytShim);
chmodSync(join(fakeBinDir, "yt-dlp"), 0o755); chmodSync(join(fakeBinDir, "yt-dlp"), 0o755);
process.env.PATH = `${fakeBinDir}:${process.env.PATH}`; process.env.PATH = `${fakeBinDir}:${process.env.PATH}`;
}); });
@@ -56,72 +65,47 @@ afterAll(() => {
process.env.PATH = realPath; process.env.PATH = realPath;
}); });
// ─── helpers ───────────────────────────────────────────────────────────────────
function consumeStream(stream: Readable): Promise<string> {
return new Promise<string>((resolve) => {
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();
});
}
// ─── tests ───────────────────────────────────────────────────────────────────── // ─── tests ─────────────────────────────────────────────────────────────────────
describe("getDirectScreenInput", () => { describe("downloadScreenInput", () => {
it("returns a live Readable and streams media bytes from yt-dlp stdout", async () => { it("downloads media to a temp file and returns its path", async () => {
const result = await getDirectScreenInput("https://youtu.be/abc"); const mediaPath = await downloadScreenInput("https://youtu.be/abc");
expect(Readable.isReadable(result)).toBe(true); expect(typeof mediaPath).toBe("string");
expect(mediaPath).toMatch(/gmw-ytdlp-/);
const outcome = await consumeStream(result); const size = readFileSync(mediaPath).length;
// The fake yt-dlp emits 4096 bytes → the stream must deliver them. expect(size).toBeGreaterThan(0);
expect(outcome).toMatch(/^(error|end)-after-[1-9]\d*B$/); // The fake yt-dlp writes 4096 bytes → the file must deliver them.
expect(size).toBe(4096);
}); });
it("destroys the stream with an error when yt-dlp fails before producing data (transient 403)", async () => { it("rejects when yt-dlp fails (transient 403) so the controller retries", async () => {
// Simulate the production failure: yt-dlp's downloader hits a transient
// YouTube 403 and exits non-zero WITHOUT emitting a single byte. The
// returned Readable must terminate with zero bytes (error OR end) so the
// controller's resolveInputWithRetry retries with a fresh run instead of
// streaming a silent black tile.
process.env.GMW_FAKE_YTDLP_FAIL = "1"; process.env.GMW_FAKE_YTDLP_FAIL = "1";
try { try {
const result = await getDirectScreenInput("https://youtu.be/abc"); await expect(downloadScreenInput("https://youtu.be/abc")).rejects.toThrow(
expect(Readable.isReadable(result)).toBe(true); /403|exit 8|failed/i,
);
const outcome = await consumeStream(result);
expect(outcome).toMatch(/^(error|end)-after-0B$/);
} finally { } finally {
delete process.env.GMW_FAKE_YTDLP_FAIL; delete process.env.GMW_FAKE_YTDLP_FAIL;
} }
}); });
it("passes -o - (stdout streaming) and a temp dir to yt-dlp", async () => { it("passes a file -o pattern (NOT `-o -`) to yt-dlp", async () => {
const argsDump = join( const argsDump = join(
tmpdir(), tmpdir(),
`gmw-ytargs-${process.pid}-${Date.now()}.txt`, `gmw-ytargs-${process.pid}-${Date.now()}.txt`,
); );
process.env.GMW_FAKE_YTDLP_DUMP_ARGS = argsDump; process.env.GMW_FAKE_YTDLP_DUMP_ARGS = argsDump;
// Augment the fake to dump its argv.
const shim = `#!/usr/bin/env bash
printf '%s\\n' "$*" >> "$GMW_FAKE_YTDLP_DUMP_ARGS"
head -c 4096 /dev/urandom
exit 0
`;
const realPath2 = process.env.PATH; const realPath2 = process.env.PATH;
const dir = fakeBinDir as unknown as string; const dir = fakeBinDir as unknown as string;
const existing = join(dir, "yt-dlp"); const existing = join(dir, "yt-dlp");
// Overwrite with the argv-dumping variant. writeFileSync(existing, ytShimDump);
writeFileSync(existing, shim);
chmodSync(existing, 0o755); chmodSync(existing, 0o755);
try { try {
const result = await getDirectScreenInput("https://youtu.be/abc"); const mediaPath = await downloadScreenInput("https://youtu.be/abc");
await consumeStream(result); expect(readFileSync(mediaPath).length).toBeGreaterThan(0);
await new Promise((r) => setTimeout(r, 100)); await new Promise((r) => setTimeout(r, 100));
const args = readFileSync(argsDump, "utf8").trim(); const args = readFileSync(argsDump, "utf8").trim();
expect(args).toContain("-o -"); expect(args).not.toContain("-o -");
expect(args).toContain("-o ");
expect(args).toMatch(/gmw-ytdlp-/); expect(args).toMatch(/gmw-ytdlp-/);
} finally { } finally {
delete process.env.GMW_FAKE_YTDLP_DUMP_ARGS; delete process.env.GMW_FAKE_YTDLP_DUMP_ARGS;