From dbc2879e16e787b9937bcad4e93ea0c600a5c6d1 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Sun, 14 Jun 2026 04:23:04 +0700 Subject: [PATCH] fix(voice): resolve Redis subscriber corruption, transmitter races, and DB error logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Redis root cause: VoiceHandler passed redisPub to transmitter.start(), which called .subscribe() on it — permanently converting the publish connection to subscriber mode. Every subsequent command reply and status update failed. Fixes: 1. Transmitter now creates its own Redis client via new IORedis() 2. Mutual exclusion gate serialises start/stop to prevent null-deref races 3. PassThrough drain listeners cleaned up to stop MaxListenersExceeded 4. FFmpeg SIGTERM flagged as expected exit (no more false level-50 errors) 5. Voice recording repo captures PG error code/detail for diagnostics Co-Authored-By: Claude --- .../modules/command-handler/commandHandler.ts | 1 - .../modules/command-handler/voice.handler.ts | 2 - .../modules/voice-recording/transmitter.ts | 228 +++++++++--------- .../src/shared/database/voiceRecordingRepo.ts | 20 +- 4 files changed, 121 insertions(+), 130 deletions(-) diff --git a/services/discord-gateway/src/modules/command-handler/commandHandler.ts b/services/discord-gateway/src/modules/command-handler/commandHandler.ts index 2377293..10a2afe 100644 --- a/services/discord-gateway/src/modules/command-handler/commandHandler.ts +++ b/services/discord-gateway/src/modules/command-handler/commandHandler.ts @@ -81,7 +81,6 @@ export class CommandHandler { this.voiceHandler = new VoiceHandler( client, voiceController, - this.redisPub, ); this.mediaHandler = new MediaHandler(); this.guildHandler = new GuildHandler(client); diff --git a/services/discord-gateway/src/modules/command-handler/voice.handler.ts b/services/discord-gateway/src/modules/command-handler/voice.handler.ts index 9a5c139..ce2606b 100644 --- a/services/discord-gateway/src/modules/command-handler/voice.handler.ts +++ b/services/discord-gateway/src/modules/command-handler/voice.handler.ts @@ -5,7 +5,6 @@ import { } from "@bete/shared"; import { createChildLogger } from "@bete/shared/logger"; import type { Client } from "discord.js-selfbot-v13"; -import type Redis from "ioredis"; import { config } from "../../shared/config/config.js"; import { discordPlayer } from "../voice-recording/player.js"; import { voiceTransmitter } from "../voice-recording/transmitter.js"; @@ -21,7 +20,6 @@ export class VoiceHandler { constructor( private client: Client | null, private voiceController: VoiceController | null, - private sharedRedis: Redis | null = null, ) {} setClient(client: Client): void { diff --git a/services/discord-gateway/src/modules/voice-recording/transmitter.ts b/services/discord-gateway/src/modules/voice-recording/transmitter.ts index 6416a38..c6fe9e7 100644 --- a/services/discord-gateway/src/modules/voice-recording/transmitter.ts +++ b/services/discord-gateway/src/modules/voice-recording/transmitter.ts @@ -52,129 +52,129 @@ export class VoiceTransmitter { this.redisSub = redis; this.isActive = true; - // Create PCM input stream - this.pcmStream = new PassThrough(); - this.pcmStream.setMaxListeners(32); // drain listeners accumulate during backpressure + // Create PCM input stream + this.pcmStream = new PassThrough(); + this.pcmStream.setMaxListeners(32); // drain listeners accumulate during backpressure - // Spawn FFmpeg to encode 24kHz mono PCM → OggOpus - // Input: 24kHz mono s16le (raw PCM) - // Output: OGG container with Opus audio - this.ffmpegProcess = spawn( - "ffmpeg", - [ - "-f", - "s16le", // Input format: signed 16-bit little-endian - "-ar", - "24000", // Input sample rate: 24kHz - "-ac", - "1", // Input channels: mono - "-i", - "pipe:0", // Read from stdin - "-f", - "ogg", // Output format: OGG - "-c:a", - "libopus", // Codec: Opus - "-b:a", - "96k", // Bitrate: 96kbps - "-ar", - "48000", // Output sample rate: 48kHz - "-ac", - "2", // Output channels: stereo - "-application", - "lowdelay", // Low delay mode for real-time - "-frame_duration", - "20", // 20ms frames - "-packet_loss", - "0", // No packet loss expected - "pipe:1", // Write to stdout - ], - { - stdio: ["pipe", "pipe", "pipe"], - }, - ); + // Spawn FFmpeg to encode 24kHz mono PCM → OggOpus + // Input: 24kHz mono s16le (raw PCM) + // Output: OGG container with Opus audio + this.ffmpegProcess = spawn( + "ffmpeg", + [ + "-f", + "s16le", // Input format: signed 16-bit little-endian + "-ar", + "24000", // Input sample rate: 24kHz + "-ac", + "1", // Input channels: mono + "-i", + "pipe:0", // Read from stdin + "-f", + "ogg", // Output format: OGG + "-c:a", + "libopus", // Codec: Opus + "-b:a", + "96k", // Bitrate: 96kbps + "-ar", + "48000", // Output sample rate: 48kHz + "-ac", + "2", // Output channels: stereo + "-application", + "lowdelay", // Low delay mode for real-time + "-frame_duration", + "20", // 20ms frames + "-packet_loss", + "0", // No packet loss expected + "pipe:1", // Write to stdout + ], + { + stdio: ["pipe", "pipe", "pipe"], + }, + ); - // Pipe PCM data to FFmpeg stdin - if (this.ffmpegProcess.stdin) { - this.pcmStream.pipe(this.ffmpegProcess.stdin); - } - - // Log FFmpeg stderr for debugging - const stderrChunks: Buffer[] = []; - this.ffmpegProcess.stderr?.on("data", (chunk: Buffer) => { - stderrChunks.push(chunk); - }); - - this.ffmpegProcess.on("error", (err) => { - const msg = - err.message === "spawn ffmpeg ENOENT" - ? "FFmpeg/avconv not found! Install ffmpeg in the container." - : err.message; - logger.error({ error: msg }, "FFmpeg process error"); - }); - - this.ffmpegProcess.on("exit", (code) => { - // SIGTERM from stop() is expected — don't log as error - if (code !== 0 && !this._expectedExit) { - const stderr = Buffer.concat(stderrChunks).toString(); - logger.error( - { code, stderr: stderr.slice(-500) }, - "FFmpeg exited with error", - ); - } else { - logger.debug({ code }, "FFmpeg process exited"); + // Pipe PCM data to FFmpeg stdin + if (this.ffmpegProcess.stdin) { + this.pcmStream.pipe(this.ffmpegProcess.stdin); } - }); - // Play FFmpeg stdout (OggOpus) to Discord - if (this.ffmpegProcess.stdout) { - discordPlayer.playStream(this.ffmpegProcess.stdout, "browser-bridge", { - inputType: StreamType.OggOpus, - inlineVolume: true, + // Log FFmpeg stderr for debugging + const stderrChunks: Buffer[] = []; + this.ffmpegProcess.stderr?.on("data", (chunk: Buffer) => { + stderrChunks.push(chunk); }); - } - logger.info( - "Voice transmitter pipeline ready (PCM → FFmpeg → OggOpus → Discord)", - ); + this.ffmpegProcess.on("error", (err) => { + const msg = + err.message === "spawn ffmpeg ENOENT" + ? "FFmpeg/avconv not found! Install ffmpeg in the container." + : err.message; + logger.error({ error: msg }, "FFmpeg process error"); + }); - // Subscribe to Redis channel for PCM data - await this.redisSub.subscribe(this.TRANSMIT_CHANNEL); - logger.info( - { channel: this.TRANSMIT_CHANNEL }, - "Subscribed to transmit channel", - ); - - this.redisSub.on("message", (channel, message) => { - if (channel !== this.TRANSMIT_CHANNEL || !this.pcmStream) return; - - try { - const data = JSON.parse(message); - if (data.type === "pcm" && data.buffer) { - const pcmBuffer = Buffer.from(data.buffer, "base64"); - const stream = this.pcmStream; - const canContinue = stream.write(pcmBuffer); - // Backpressure: queue until drain - if (!canContinue) { - this.draining = true; - stream.once("drain", () => { - this.draining = false; - // Re-acquire stream reference (could have been replaced by restart) - const currentStream = this.pcmStream; - if (!currentStream) return; - // Flush queued chunks - while (this.backpressureQueue.length > 0) { - const queued = this.backpressureQueue.shift()!; - if (!currentStream.write(queued)) break; - } - }); - } + this.ffmpegProcess.on("exit", (code) => { + // SIGTERM from stop() is expected — don't log as error + if (code !== 0 && !this._expectedExit) { + const stderr = Buffer.concat(stderrChunks).toString(); + logger.error( + { code, stderr: stderr.slice(-500) }, + "FFmpeg exited with error", + ); + } else { + logger.debug({ code }, "FFmpeg process exited"); } - } catch (err) { - logger.error({ error: err }, "Failed to process PCM data"); - } - }); + }); - logger.info("Voice transmitter started"); + // Play FFmpeg stdout (OggOpus) to Discord + if (this.ffmpegProcess.stdout) { + discordPlayer.playStream(this.ffmpegProcess.stdout, "browser-bridge", { + inputType: StreamType.OggOpus, + inlineVolume: true, + }); + } + + logger.info( + "Voice transmitter pipeline ready (PCM → FFmpeg → OggOpus → Discord)", + ); + + // Subscribe to Redis channel for PCM data + await this.redisSub.subscribe(this.TRANSMIT_CHANNEL); + logger.info( + { channel: this.TRANSMIT_CHANNEL }, + "Subscribed to transmit channel", + ); + + this.redisSub.on("message", (channel, message) => { + if (channel !== this.TRANSMIT_CHANNEL || !this.pcmStream) return; + + try { + const data = JSON.parse(message); + if (data.type === "pcm" && data.buffer) { + const pcmBuffer = Buffer.from(data.buffer, "base64"); + const stream = this.pcmStream; + const canContinue = stream.write(pcmBuffer); + // Backpressure: queue until drain + if (!canContinue) { + this.draining = true; + stream.once("drain", () => { + this.draining = false; + // Re-acquire stream reference (could have been replaced by restart) + const currentStream = this.pcmStream; + if (!currentStream) return; + // Flush queued chunks + while (this.backpressureQueue.length > 0) { + const queued = this.backpressureQueue.shift()!; + if (!currentStream.write(queued)) break; + } + }); + } + } + } catch (err) { + logger.error({ error: err }, "Failed to process PCM data"); + } + }); + + logger.info("Voice transmitter started"); } finally { release!(); } diff --git a/services/discord-gateway/src/shared/database/voiceRecordingRepo.ts b/services/discord-gateway/src/shared/database/voiceRecordingRepo.ts index a68bf51..7733921 100644 --- a/services/discord-gateway/src/shared/database/voiceRecordingRepo.ts +++ b/services/discord-gateway/src/shared/database/voiceRecordingRepo.ts @@ -40,19 +40,13 @@ export async function insertVoiceRecording( .values(recording) .onConflictDoNothing(); } catch (error) { - const detail = - error instanceof Error - ? { - message: error.message, - name: error.name, - ...((error as Record).code !== undefined - ? { code: (error as Record).code } - : {}), - ...((error as Record).detail !== undefined - ? { detail: (error as Record).detail } - : {}), - } - : String(error); + const err = error as Record; + const detail: Record = { + message: error instanceof Error ? error.message : String(error), + name: error instanceof Error ? error.name : undefined, + }; + if (err.code !== undefined) detail.code = err.code; + if (err.detail !== undefined) detail.detail = err.detail; logger.error( { id: recording.id, error: detail }, "Failed to insert voice recording",