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

fix(voice): resolve Redis subscriber corruption, transmitter races, and DB error logging

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
MythEclipse
2026-06-14 04:23:04 +07:00
co-authored by Claude
parent 32b0b20ca7
commit dbc2879e16
4 changed files with 121 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/discord-gateway/src/modules/command-handler/commandHandler.ts
-1
View File
@@ -81,7 +81,6 @@ export class CommandHandler {
this.voiceHandler = new VoiceHandler( this.voiceHandler = new VoiceHandler(
client, client,
voiceController, voiceController,
this.redisPub,
); );
this.mediaHandler = new MediaHandler(); this.mediaHandler = new MediaHandler();
this.guildHandler = new GuildHandler(client); this.guildHandler = new GuildHandler(client);
services/discord-gateway/src/modules/command-handler/voice.handler.ts
-2
View File
@@ -5,7 +5,6 @@ import {
} from "@bete/shared"; } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger"; import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13"; import type { Client } from "discord.js-selfbot-v13";
import type Redis from "ioredis";
import { config } from "../../shared/config/config.js"; import { config } from "../../shared/config/config.js";
import { discordPlayer } from "../voice-recording/player.js"; import { discordPlayer } from "../voice-recording/player.js";
import { voiceTransmitter } from "../voice-recording/transmitter.js"; import { voiceTransmitter } from "../voice-recording/transmitter.js";
@@ -21,7 +20,6 @@ export class VoiceHandler {
constructor( constructor(
private client: Client | null, private client: Client | null,
private voiceController: VoiceController | null, private voiceController: VoiceController | null,
private sharedRedis: Redis | null = null,
) {} ) {}
setClient(client: Client): void { setClient(client: Client): void {
services/discord-gateway/src/modules/voice-recording/transmitter.ts
+114 -114
View File
@@ -52,129 +52,129 @@ export class VoiceTransmitter {
this.redisSub = redis; this.redisSub = redis;
this.isActive = true; this.isActive = true;
// 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
// 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)
// Output: OGG container with Opus audio // Output: OGG container with Opus audio
this.ffmpegProcess = spawn( this.ffmpegProcess = spawn(
"ffmpeg", "ffmpeg",
[ [
"-f", "-f",
"s16le", // Input format: signed 16-bit little-endian "s16le", // Input format: signed 16-bit little-endian
"-ar", "-ar",
"24000", // Input sample rate: 24kHz "24000", // Input sample rate: 24kHz
"-ac", "-ac",
"1", // Input channels: mono "1", // Input channels: mono
"-i", "-i",
"pipe:0", // Read from stdin "pipe:0", // Read from stdin
"-f", "-f",
"ogg", // Output format: OGG "ogg", // Output format: OGG
"-c:a", "-c:a",
"libopus", // Codec: Opus "libopus", // Codec: Opus
"-b:a", "-b:a",
"96k", // Bitrate: 96kbps "96k", // Bitrate: 96kbps
"-ar", "-ar",
"48000", // Output sample rate: 48kHz "48000", // Output sample rate: 48kHz
"-ac", "-ac",
"2", // Output channels: stereo "2", // Output channels: stereo
"-application", "-application",
"lowdelay", // Low delay mode for real-time "lowdelay", // Low delay mode for real-time
"-frame_duration", "-frame_duration",
"20", // 20ms frames "20", // 20ms frames
"-packet_loss", "-packet_loss",
"0", // No packet loss expected "0", // No packet loss expected
"pipe:1", // Write to stdout "pipe:1", // Write to stdout
], ],
{ {
stdio: ["pipe", "pipe", "pipe"], stdio: ["pipe", "pipe", "pipe"],
}, },
); );
// Pipe PCM data to FFmpeg stdin // Pipe PCM data to FFmpeg stdin
if (this.ffmpegProcess.stdin) { if (this.ffmpegProcess.stdin) {
this.pcmStream.pipe(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");
} }
});
// Play FFmpeg stdout (OggOpus) to Discord // Log FFmpeg stderr for debugging
if (this.ffmpegProcess.stdout) { const stderrChunks: Buffer[] = [];
discordPlayer.playStream(this.ffmpegProcess.stdout, "browser-bridge", { this.ffmpegProcess.stderr?.on("data", (chunk: Buffer) => {
inputType: StreamType.OggOpus, stderrChunks.push(chunk);
inlineVolume: true,
}); });
}
logger.info( this.ffmpegProcess.on("error", (err) => {
"Voice transmitter pipeline ready (PCM → FFmpeg → OggOpus → Discord)", 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 this.ffmpegProcess.on("exit", (code) => {
await this.redisSub.subscribe(this.TRANSMIT_CHANNEL); // SIGTERM from stop() is expected — don't log as error
logger.info( if (code !== 0 && !this._expectedExit) {
{ channel: this.TRANSMIT_CHANNEL }, const stderr = Buffer.concat(stderrChunks).toString();
"Subscribed to transmit channel", logger.error(
); { code, stderr: stderr.slice(-500) },
"FFmpeg exited with error",
this.redisSub.on("message", (channel, message) => { );
if (channel !== this.TRANSMIT_CHANNEL || !this.pcmStream) return; } else {
logger.debug({ code }, "FFmpeg process exited");
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"); // 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 { } finally {
release!(); release!();
} }
services/discord-gateway/src/shared/database/voiceRecordingRepo.ts
+7 -13
View File
@@ -40,19 +40,13 @@ export async function insertVoiceRecording(
.values(recording) .values(recording)
.onConflictDoNothing(); .onConflictDoNothing();
} catch (error) { } catch (error) {
const detail = const err = error as Record<string, unknown>;
error instanceof Error const detail: Record<string, unknown> = {
? { message: error instanceof Error ? error.message : String(error),
message: error.message, name: error instanceof Error ? error.name : undefined,
name: error.name, };
...((error as Record<string, unknown>).code !== undefined if (err.code !== undefined) detail.code = err.code;
? { code: (error as Record<string, unknown>).code } if (err.detail !== undefined) detail.detail = err.detail;
: {}),
...((error as Record<string, unknown>).detail !== undefined
? { detail: (error as Record<string, unknown>).detail }
: {}),
}
: String(error);
logger.error( logger.error(
{ id: recording.id, error: detail }, { id: recording.id, error: detail },
"Failed to insert voice recording", "Failed to insert voice recording",