fix(gateway): voice module critical bugs
- shutdown ordering: DB pool closed AFTER voice disconnect (was before) - muxer: amix (mix) -> concat (sequential) filter for OGG segments - upload retries: 0->3 with proper backoff 1-10s - teleUpload retry backoff: minTimeout 0->1000, maxTimeout 0->10000 - mediaSource: 64KB buffer bound on readFirstTwoLines, 1MB on extractMediaInfo - ffmpegProcess: pipe stderr to structured logger instead of inherit Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -36,19 +36,23 @@ export function createGracefulShutdown(options: GracefulShutdownOptions) {
|
||||
try {
|
||||
options.stopMetricsServer?.();
|
||||
stopMuxerWorker();
|
||||
options.logger.info("Closing database...");
|
||||
await options.closeDatabase();
|
||||
options.logger.info("Database closed");
|
||||
|
||||
// 1. Voice disconnect (may write final DB records)
|
||||
options.logger.info("Stopping voice connection...");
|
||||
await options.voiceController.disconnect();
|
||||
|
||||
// 2. Close Redis/pubsub after voice is done
|
||||
options.logger.info("Closing event broadcaster...");
|
||||
await options.eventBroadcaster.close();
|
||||
|
||||
options.logger.info("Closing command handler...");
|
||||
await options.commandHandler.close();
|
||||
|
||||
// 3. DB pool LAST – voice disconnect may finalize recordings
|
||||
options.logger.info("Closing database...");
|
||||
await options.closeDatabase();
|
||||
options.logger.info("Database closed");
|
||||
|
||||
options.logger.info("Destroying Discord client...");
|
||||
try {
|
||||
options.client.destroy();
|
||||
|
||||
@@ -48,7 +48,12 @@ export function runFfmpeg(args: string[]): Promise<void> {
|
||||
logger.debug({ args }, "Starting ffmpeg");
|
||||
|
||||
const proc = spawn("ffmpeg", args, {
|
||||
stdio: ["ignore", "inherit", "inherit"],
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
|
||||
let stderrBuf = "";
|
||||
proc.stderr?.on("data", (chunk: Buffer) => {
|
||||
stderrBuf += chunk.toString("utf8");
|
||||
});
|
||||
|
||||
proc.on("close", (code) => {
|
||||
@@ -56,8 +61,9 @@ export function runFfmpeg(args: string[]): Promise<void> {
|
||||
logger.debug("ffmpeg completed successfully");
|
||||
resolve();
|
||||
} else {
|
||||
logger.warn({ exitCode: code }, "ffmpeg exited with non-zero code");
|
||||
reject(new Error(`ffmpeg exited with code ${code}`));
|
||||
const detail = stderrBuf.trim().slice(0, 2000);
|
||||
logger.warn({ exitCode: code, stderr: detail }, "ffmpeg exited with non-zero code");
|
||||
reject(new Error(`ffmpeg exited with code ${code}: ${detail}`));
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -61,7 +61,12 @@ function parseSeconds(value: string): number {
|
||||
* Returns the parsed header and a new Readable that contains all remaining
|
||||
* data (the audio stream).
|
||||
*/
|
||||
function readFirstTwoLines(stdout: Readable): Promise<{
|
||||
const MAX_HEADER_BUFFER = 65536; // 64KB safety limit for metadata headers
|
||||
|
||||
function readFirstTwoLines(
|
||||
stdout: Readable,
|
||||
maxBufferSize: number = MAX_HEADER_BUFFER,
|
||||
): Promise<{
|
||||
title: string;
|
||||
duration: number;
|
||||
remaining: Readable;
|
||||
@@ -82,6 +87,11 @@ function readFirstTwoLines(stdout: Readable): Promise<{
|
||||
function onData(chunk: Buffer) {
|
||||
if (stage === "done") return;
|
||||
buffer = Buffer.concat([buffer, chunk]);
|
||||
if (buffer.length > maxBufferSize) {
|
||||
cleanup();
|
||||
reject(new Error(`Metadata header exceeded ${maxBufferSize} bytes`));
|
||||
return;
|
||||
}
|
||||
processBuffer();
|
||||
}
|
||||
|
||||
@@ -297,10 +307,15 @@ export async function extractMediaInfo(url: string): Promise<MediaInfo> {
|
||||
let stdoutBuf = "";
|
||||
let stderrBuf = "";
|
||||
const MAX_STDERR = 4096;
|
||||
const MAX_STDOUT = 1_048_576; // 1MB safety limit
|
||||
|
||||
if (proc.stdout) {
|
||||
proc.stdout.on("data", (chunk: Buffer) => {
|
||||
stdoutBuf += chunk.toString("utf8");
|
||||
if (stdoutBuf.length < MAX_STDOUT) {
|
||||
stdoutBuf += chunk
|
||||
.toString("utf8")
|
||||
.slice(0, MAX_STDOUT - stdoutBuf.length);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -134,10 +134,10 @@ async function processJob(
|
||||
"Processing muxer job",
|
||||
);
|
||||
|
||||
// Build filter: concat all inputs with a crossfade or simple concat
|
||||
// We use amix for mixing multiple audio streams (not sequential concat)
|
||||
const inputLabels = data.inputs.map((_, i) => `[${i}:a]`);
|
||||
const filterComplex = `${inputLabels.join("")}amix=inputs=${data.inputs.length}:duration=first:dropout_transition=2[out]`;
|
||||
// Sequential concat of all OGG segments (amix = simultaneous mix, wrong for this)
|
||||
const inputLabels = data.inputs.map((_, i) => `[${i}:a:0]`);
|
||||
const n = data.inputs.length;
|
||||
const filterComplex = `${inputLabels.join("")}concat=n=${n}:v=0:a=1[out]`;
|
||||
|
||||
const args = buildMuxFfmpegArgs({
|
||||
inputs: data.inputs,
|
||||
|
||||
@@ -60,7 +60,7 @@ export async function uploadRecordingSegment(input: {
|
||||
filename: fileName,
|
||||
contentType: "audio/ogg",
|
||||
uploadUrl: config.TELE_UPLOAD_URL,
|
||||
retries: 0,
|
||||
retries: 3,
|
||||
});
|
||||
const downloadUrl = uploadResult.url;
|
||||
|
||||
|
||||
@@ -71,8 +71,8 @@ export async function uploadToTele(input: {
|
||||
},
|
||||
{
|
||||
retries,
|
||||
minTimeout: 0,
|
||||
maxTimeout: 0,
|
||||
minTimeout: 1000,
|
||||
maxTimeout: 10000,
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user