fix(gateway): voice speaking handler, metadata cache, transmitter backpressure
- speakingHandler: check subscriptions.has() BEFORE voiceActiveUser broadcast - segment.ts: add LRU metadata cache (200 entries, avoids Discord API in hotpath) - transmitter.ts: add backpressure handling with drain queue/flush (PassThrough.write() return value was ignored — potential OOM under load) - cleanup backpressure queue on transmitter stop Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -13,11 +13,15 @@ import type { RecordingSession } from "./sessionRecording.js";
|
||||
import { uploadRecordingSegment } from "./uploader.js";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Logger
|
||||
// Logger & metadata cache
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const logger = createChildLogger("voice-segment");
|
||||
|
||||
/** LRU-ish cache: userId -> UserMetadata. Avoids Discord API calls in hotpath. */
|
||||
const metadataCache = new Map<string, UserMetadata>();
|
||||
const METADATA_CACHE_MAX = 200;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// collectUserMetadata (was metadata.ts)
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -27,7 +31,8 @@ export async function collectUserMetadata(
|
||||
userId: string,
|
||||
channel: VoiceChannel,
|
||||
): Promise<UserMetadata> {
|
||||
logger.debug({ userId }, "Collecting user metadata");
|
||||
const cached = metadataCache.get(userId);
|
||||
if (cached) return cached;
|
||||
|
||||
const user =
|
||||
client.users.cache.get(userId) ||
|
||||
@@ -52,7 +57,7 @@ export async function collectUserMetadata(
|
||||
position: role.position,
|
||||
})) ?? [];
|
||||
|
||||
return {
|
||||
const result: UserMetadata = {
|
||||
userId,
|
||||
username,
|
||||
tag: user?.tag ?? "Unknown#0000",
|
||||
@@ -76,7 +81,9 @@ export async function collectUserMetadata(
|
||||
highestRole: roles[0] ?? null,
|
||||
joinedTimestamp: member?.joinedTimestamp ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
cacheMetadata(userId, result);
|
||||
return result;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Path helpers (was segment.ts)
|
||||
|
||||
@@ -54,6 +54,10 @@ export function createSpeakingHandler(
|
||||
"Voice activity detected",
|
||||
);
|
||||
|
||||
// Skip if user already has an active stream subscription
|
||||
// (check BEFORE broadcast to avoid false positive events)
|
||||
if (receiver.subscriptions.has(userId)) return;
|
||||
|
||||
// Notify webserver / WebSocket clients
|
||||
eventBroadcaster?.voiceActiveUser(userId, {
|
||||
username: userMetadata.username,
|
||||
@@ -61,9 +65,6 @@ export function createSpeakingHandler(
|
||||
speaking: true,
|
||||
});
|
||||
|
||||
// Skip if user already has an active stream subscription
|
||||
if (receiver.subscriptions.has(userId)) return;
|
||||
|
||||
// Ensure per-user recording directory
|
||||
const userDir = path.join(recordingsDir, userId);
|
||||
await fsPromises.mkdir(userDir, { recursive: true }).catch(() => {
|
||||
|
||||
@@ -21,6 +21,9 @@ export class VoiceTransmitter {
|
||||
private ffmpegProcess: ReturnType<typeof spawn> | null = null;
|
||||
private isActive = false;
|
||||
private readonly TRANSMIT_CHANNEL = BACKEND_VOICE_TRANSMIT;
|
||||
/** Queue for PCM chunks when backpressure is active */
|
||||
private backpressureQueue: Buffer[] = [];
|
||||
private draining = false;
|
||||
|
||||
/**
|
||||
* Start listening for PCM audio data from Redis and stream to Discord
|
||||
@@ -130,8 +133,19 @@ export class VoiceTransmitter {
|
||||
const data = JSON.parse(message);
|
||||
if (data.type === "pcm" && data.buffer) {
|
||||
const pcmBuffer = Buffer.from(data.buffer, "base64");
|
||||
logger.debug({ bytes: pcmBuffer.length }, "Received PCM chunk");
|
||||
this.pcmStream.write(pcmBuffer);
|
||||
const canContinue = this.pcmStream.write(pcmBuffer);
|
||||
// Backpressure: queue until drain
|
||||
if (!canContinue) {
|
||||
this.draining = true;
|
||||
this.pcmStream.once("drain", () => {
|
||||
this.draining = false;
|
||||
// Flush queued chunks
|
||||
while (this.backpressureQueue.length > 0) {
|
||||
const queued = this.backpressureQueue.shift()!;
|
||||
if (!this.pcmStream.write(queued)) break;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error({ error: err }, "Failed to process PCM data");
|
||||
@@ -150,6 +164,9 @@ export class VoiceTransmitter {
|
||||
|
||||
this.isActive = false;
|
||||
|
||||
this.backpressureQueue = [];
|
||||
this.draining = false;
|
||||
|
||||
if (this.pcmStream) {
|
||||
this.pcmStream.end();
|
||||
this.pcmStream = null;
|
||||
|
||||
Reference in New Issue
Block a user