From f0370a3c209736c3097ccc7a7becd4bbcb1d404a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 4 Jun 2026 13:35:10 +0700 Subject: [PATCH] Fix long dictation retry backpressure Treat Node IPC backpressure as a queue signal instead of a failed send, and pace buffered dictation replay so finish waits for the backlog to flush. --- .../dictation/dictation-stream-sender.test.ts | 26 ++++++ .../src/dictation/dictation-stream-sender.ts | 59 +++++++++++- .../local/test-fixtures/paused-ipc-worker.cjs | 23 +++++ .../providers/local/worker-client.test.ts | 89 +++++++++++++++++++ .../speech/providers/local/worker-client.ts | 12 +-- 5 files changed, 197 insertions(+), 12 deletions(-) create mode 100644 packages/server/src/server/speech/providers/local/test-fixtures/paused-ipc-worker.cjs diff --git a/packages/app/src/dictation/dictation-stream-sender.test.ts b/packages/app/src/dictation/dictation-stream-sender.test.ts index 152329420..0684dd2b1 100644 --- a/packages/app/src/dictation/dictation-stream-sender.test.ts +++ b/packages/app/src/dictation/dictation-stream-sender.test.ts @@ -137,4 +137,30 @@ describe("DictationStreamSender", () => { expect(client.chunks.map((c) => c.seq)).toEqual([0, 1]); }); + + it("does not replay long buffered native dictation in one synchronous burst", async () => { + const client = new FakeDaemonClient(); + client.isConnected = false; + const sender = new DictationStreamSender({ + client: client as unknown as DaemonClient, + format: "audio/pcm;rate=16000;bits=16", + createDictationId: () => "d1", + }); + + for (let seq = 0; seq < 480; seq += 1) { + sender.enqueueSegment(`native-frame-${seq}`); + } + + client.isConnected = true; + const finish = sender.finish(sender.getFinalSeq()); + + await tick(); + + expect(client.chunks.length).toBeLessThanOrEqual(128); + await expect(finish).resolves.toEqual({ + dictationId: "d1", + text: "ok", + }); + expect(client.finishes).toEqual([{ dictationId: "d1", finalSeq: 479 }]); + }); }); diff --git a/packages/app/src/dictation/dictation-stream-sender.ts b/packages/app/src/dictation/dictation-stream-sender.ts index 272482c74..08a748c46 100644 --- a/packages/app/src/dictation/dictation-stream-sender.ts +++ b/packages/app/src/dictation/dictation-stream-sender.ts @@ -1,6 +1,10 @@ import { generateMessageId } from "@/types/stream"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +const MAX_CHUNKS_PER_FLUSH_TURN = 128; + +const waitForNextFlushTurn = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); + export interface DictationStreamSenderParams { client: DaemonClient | null; format: string; @@ -33,6 +37,8 @@ export class DictationStreamSender { private sendSeq = 0; private segments: string[] = []; private streamReady = false; + private flushTimer: ReturnType | null = null; + private drainWaiters: Array<() => void> = []; private startGeneration = 0; private startPromise: Promise | null = null; @@ -64,6 +70,7 @@ export class DictationStreamSender { } clearAll(): void { + this.clearScheduledFlush(); this.dictationId = null; this.sendSeq = 0; this.segments = []; @@ -73,6 +80,7 @@ export class DictationStreamSender { } resetStreamForReplay(): void { + this.clearScheduledFlush(); this.dictationId = null; this.sendSeq = 0; this.streamReady = false; @@ -108,13 +116,18 @@ export class DictationStreamSender { } let sent = 0; - while (this.sendSeq < this.segments.length) { + while (this.sendSeq < this.segments.length && sent < MAX_CHUNKS_PER_FLUSH_TURN) { const seq = this.sendSeq; const audio = this.segments[seq]; client.sendDictationStreamChunk(dictationId, seq, audio, this.format); this.sendSeq = seq + 1; sent += 1; } + if (this.hasPendingSegments()) { + this.scheduleFlush(); + } else { + this.resolveDrainWaiters(); + } return sent; } @@ -184,6 +197,7 @@ export class DictationStreamSender { } this.flush(); + await this.waitForFlushDrain(); return client.finishDictationStream(dictationId, finalSeq); } @@ -195,4 +209,47 @@ export class DictationStreamSender { } this.resetStreamForReplay(); } + + private hasPendingSegments(): boolean { + return this.sendSeq < this.segments.length; + } + + private scheduleFlush(): void { + if (this.flushTimer) { + return; + } + this.flushTimer = setTimeout(() => { + this.flushTimer = null; + this.flush(); + }, 0); + } + + private clearScheduledFlush(): void { + if (!this.flushTimer) { + return; + } + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + + private async waitForFlushDrain(): Promise { + while (this.hasPendingSegments()) { + const client = this.client; + if (!client?.isConnected || !this.dictationId || !this.streamReady) { + throw new Error("Failed to flush dictation stream"); + } + await new Promise((resolve) => { + this.drainWaiters.push(resolve); + }); + await waitForNextFlushTurn(); + } + } + + private resolveDrainWaiters(): void { + const waiters = this.drainWaiters; + this.drainWaiters = []; + for (const resolve of waiters) { + resolve(); + } + } } diff --git a/packages/server/src/server/speech/providers/local/test-fixtures/paused-ipc-worker.cjs b/packages/server/src/server/speech/providers/local/test-fixtures/paused-ipc-worker.cjs new file mode 100644 index 000000000..784cb17b3 --- /dev/null +++ b/packages/server/src/server/speech/providers/local/test-fixtures/paused-ipc-worker.cjs @@ -0,0 +1,23 @@ +process.on("message", (message) => { + if (!message || message.type !== "session.create") { + if (message && message.requestId && process.send) { + process.send({ type: "response", requestId: message.requestId, ok: true }); + } + return; + } + + process.send( + { + type: "response", + requestId: message.requestId, + ok: true, + result: { requiredSampleRate: 16000 }, + }, + () => { + const lock = new Int32Array(new SharedArrayBuffer(4)); + Atomics.wait(lock, 0, 0, 500); + }, + ); +}); + +setInterval(() => undefined, 1000); diff --git a/packages/server/src/server/speech/providers/local/worker-client.test.ts b/packages/server/src/server/speech/providers/local/worker-client.test.ts index 27b37bd15..b42ff4adc 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.test.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.test.ts @@ -1,5 +1,7 @@ import { EventEmitter } from "node:events"; import { once } from "node:events"; +import { fork, type ChildProcess } from "node:child_process"; +import { fileURLToPath } from "node:url"; import pino from "pino"; import { describe, expect, it } from "vitest"; @@ -54,6 +56,50 @@ class FakeLocalSpeechWorker extends EventEmitter { } } +class PausedIpcWorker { + private readonly child: ChildProcess; + + constructor() { + this.child = fork( + fileURLToPath(new URL("./test-fixtures/paused-ipc-worker.cjs", import.meta.url)), + [], + { serialization: "advanced", stdio: ["ignore", "ignore", "ignore", "ipc"] }, + ); + } + + get connected(): boolean { + return this.child.connected; + } + + get killed(): boolean { + return this.child.killed; + } + + send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean { + return this.child.send(message, (error) => callback(error ?? null)); + } + + disconnect(): void { + this.child.disconnect(); + } + + kill(): boolean { + return this.child.kill(); + } + + on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this; + on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this; + on( + event: "message" | "exit", + listener: + | ((message: LocalSpeechWorkerToParentMessage) => void) + | ((code: number | null, signal: NodeJS.Signals | null) => void), + ): this { + this.child.on(event, listener as (...args: unknown[]) => void); + return this; + } +} + function createClient(options?: { idleTtlMs?: number }) { const workers: FakeLocalSpeechWorker[] = []; const client = new LocalSpeechWorkerClient({ @@ -168,6 +214,49 @@ describe("LocalSpeechWorkerClient", () => { ]); }); + it("does not surface real IPC backpressure when replaying native-sized dictation frames", async () => { + const workers: PausedIpcWorker[] = []; + const client = new LocalSpeechWorkerClient({ + config: { + modelsDir: "/tmp/models", + voiceSttModel: "parakeet-tdt-0.6b-v2-int8", + dictationSttModel: "parakeet-tdt-0.6b-v2-int8", + voiceTtsModel: "kokoro-en-v0_19", + }, + requestTimeoutMs: 30_000, + idleTtlMs: 30_000, + forkWorker: () => { + const worker = new PausedIpcWorker(); + workers.push(worker); + return worker; + }, + }); + const provider = new WorkerBackedSpeechToTextProvider(client, "dictationStt"); + const session = provider.createSession({ logger: pino({ level: "silent" }) }); + let observedError: Error | null = null; + (session as EventEmitter).on("error", (error: Error) => { + observedError = error; + }); + + try { + await session.connect(); + const nativeFrame = Buffer.alloc(1024, 1); + + for (let seq = 0; seq < 480; seq += 1) { + session.appendPcm16(nativeFrame); + } + session.commit(); + await waitForMicrotasks(); + + expect(observedError?.message).not.toBe("Local speech worker IPC channel is not writable"); + } finally { + client.shutdown(); + for (const worker of workers) { + worker.kill(); + } + } + }); + it("forwards VAD session events through the shared worker", async () => { const { client, workers } = createClient(); const provider = new WorkerBackedTurnDetectionProvider(client); diff --git a/packages/server/src/server/speech/providers/local/worker-client.ts b/packages/server/src/server/speech/providers/local/worker-client.ts index c6a6f95bb..88c0d46fb 100644 --- a/packages/server/src/server/speech/providers/local/worker-client.ts +++ b/packages/server/src/server/speech/providers/local/worker-client.ts @@ -248,7 +248,7 @@ export class LocalSpeechWorkerClient { timeout, }); - const sent = worker.send(message, (error) => { + worker.send(message, (error) => { if (!error) { return; } @@ -262,16 +262,6 @@ export class LocalSpeechWorkerClient { this.scheduleIdleShutdownIfReady(); pending.reject(error); }); - if (!sent) { - const pending = this.pendingRequests.get(requestId); - if (pending) { - clearTimeout(pending.timeout); - this.pendingRequests.delete(requestId); - this.inFlightRequests = Math.max(0, this.inFlightRequests - 1); - this.scheduleIdleShutdownIfReady(); - pending.reject(new Error("Local speech worker IPC channel is not writable")); - } - } }); }