fix(gateway): EPIPE crash on media stop — stream error handlers + no shutdown on transient stream errors

This commit is contained in:
asepharyana
2026-08-11 20:10:05 +07:00
parent ff554fcff2
commit 10d7ecd405
3 changed files with 85 additions and 3 deletions
@@ -279,6 +279,25 @@ export async function initializeDiscordGateway() {
}); });
process.on("uncaughtException", (err) => { 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"); logger.error({ error: err }, "Uncaught exception");
gracefulShutdown("uncaughtException"); gracefulShutdown("uncaughtException");
}); });
@@ -79,6 +79,22 @@ export function transcodeToHighQualityOgg(
); );
input.pipe(proc.stdin); 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); activeProcesses.add(proc);
const cleanup = () => { const cleanup = () => {
@@ -248,6 +264,20 @@ export function resolveMediaUrl(
// `--print` headers to stderr — pipe stdout immediately so the child // `--print` headers to stderr — pipe stdout immediately so the child
// never blocks on a full pipe while we wait for the headers on stderr. // never blocks on a full pipe while we wait for the headers on stderr.
const mediaStream = new PassThrough(); 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); proc.stdout.pipe(mediaStream);
let stderrBuf = ""; let stderrBuf = "";
@@ -56,6 +56,24 @@ export class VoiceTransmitter {
// Create PCM input stream // Create PCM input stream
this.pcmStream = new PassThrough(); this.pcmStream = new PassThrough();
this.pcmStream.setMaxListeners(32); // drain listeners accumulate during backpressure 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 // Spawn FFmpeg to encode 24kHz mono PCM → OggOpus
// Input: 24kHz mono s16le (raw PCM) // Input: 24kHz mono s16le (raw PCM)
@@ -146,7 +164,12 @@ export class VoiceTransmitter {
); );
this.redisSub.on("message", (channel, message) => { 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 { try {
const data = JSON.parse(message); const data = JSON.parse(message);
@@ -161,11 +184,21 @@ export class VoiceTransmitter {
this.draining = false; this.draining = false;
// Re-acquire stream reference (could have been replaced by restart) // Re-acquire stream reference (could have been replaced by restart)
const currentStream = this.pcmStream; const currentStream = this.pcmStream;
if (!currentStream) return; if (!currentStream || !this.isActive) return;
// Flush queued chunks // Flush queued chunks
while (this.backpressureQueue.length > 0) { while (this.backpressureQueue.length > 0) {
const queued = this.backpressureQueue.shift()!; 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;
}
} }
}); });
} }