Show useful local speech crash details (#1379)
* Improve local speech worker crash diagnostics * Fix speech worker stderr tail diagnostics * Drain speech worker stderr before reporting exit * Avoid false speech worker crash logs on shutdown
This commit is contained in:
@@ -156,6 +156,7 @@ export async function initializeLocalSpeechServices(params: {
|
||||
|
||||
const workerClient = localConfig
|
||||
? new LocalSpeechWorkerClient({
|
||||
logger,
|
||||
config: {
|
||||
modelsDir: localConfig.modelsDir,
|
||||
voiceSttModel: localModels.voiceLocalSttModel,
|
||||
|
||||
@@ -20,6 +20,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js";
|
||||
class FakeLocalSpeechWorker extends EventEmitter {
|
||||
public connected = true;
|
||||
public killed = false;
|
||||
public pid = 12345;
|
||||
public readonly stderr = new EventEmitter() as NodeJS.ReadableStream;
|
||||
public readonly sent: LocalSpeechWorkerRequest[] = [];
|
||||
public disconnects = 0;
|
||||
public kills = 0;
|
||||
@@ -88,9 +90,9 @@ class PausedIpcWorker {
|
||||
}
|
||||
|
||||
on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this;
|
||||
on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
|
||||
on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
|
||||
on(
|
||||
event: "message" | "exit",
|
||||
event: "message" | "close",
|
||||
listener:
|
||||
| ((message: LocalSpeechWorkerToParentMessage) => void)
|
||||
| ((code: number | null, signal: NodeJS.Signals | null) => void),
|
||||
@@ -103,6 +105,7 @@ class PausedIpcWorker {
|
||||
function createClient(options?: { idleTtlMs?: number }) {
|
||||
const workers: FakeLocalSpeechWorker[] = [];
|
||||
const client = new LocalSpeechWorkerClient({
|
||||
logger: pino({ level: "silent" }),
|
||||
config: {
|
||||
modelsDir: "/tmp/models",
|
||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
@@ -120,6 +123,19 @@ function createClient(options?: { idleTtlMs?: number }) {
|
||||
return { client, workers };
|
||||
}
|
||||
|
||||
function createCapturingLogger(): { logger: pino.Logger; records: Array<Record<string, unknown>> } {
|
||||
const records: Array<Record<string, unknown>> = [];
|
||||
const logger = pino(
|
||||
{ level: "trace" },
|
||||
{
|
||||
write(line: string) {
|
||||
records.push(JSON.parse(line) as Record<string, unknown>);
|
||||
},
|
||||
},
|
||||
);
|
||||
return { logger, records };
|
||||
}
|
||||
|
||||
async function waitForMicrotasks(): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
@@ -217,6 +233,7 @@ describe("LocalSpeechWorkerClient", () => {
|
||||
it("does not surface real IPC backpressure when replaying native-sized dictation frames", async () => {
|
||||
const workers: PausedIpcWorker[] = [];
|
||||
const client = new LocalSpeechWorkerClient({
|
||||
logger: pino({ level: "silent" }),
|
||||
config: {
|
||||
modelsDir: "/tmp/models",
|
||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
@@ -257,6 +274,80 @@ describe("LocalSpeechWorkerClient", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("logs worker exit details and includes actionable context in the surfaced error", async () => {
|
||||
const { logger, records } = createCapturingLogger();
|
||||
const workers: FakeLocalSpeechWorker[] = [];
|
||||
const client = new LocalSpeechWorkerClient({
|
||||
logger,
|
||||
config: {
|
||||
modelsDir: "/tmp/models",
|
||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
voiceTtsModel: "kokoro-en-v0_19",
|
||||
},
|
||||
forkWorker: () => {
|
||||
const worker = new FakeLocalSpeechWorker();
|
||||
workers.push(worker);
|
||||
return worker;
|
||||
},
|
||||
});
|
||||
const provider = new WorkerBackedSpeechToTextProvider(client, "dictationStt");
|
||||
const session = provider.createSession({ logger: pino({ level: "silent" }) });
|
||||
|
||||
const connect = session.connect();
|
||||
workers[0].emit("exit", null, "SIGABRT");
|
||||
workers[0].stderr.emit("data", "dyld: Library not loaded: libsherpa-onnx-c-api.dylib");
|
||||
workers[0].emit("close", null, "SIGABRT");
|
||||
|
||||
await expect(connect).rejects.toThrow(
|
||||
"Local speech worker exited (signal SIGABRT) while handling session.create (dictationStt). Last stderr: dyld: Library not loaded: libsherpa-onnx-c-api.dylib",
|
||||
);
|
||||
const exitRecord = records.find((record) => record.msg === "Local speech worker exited");
|
||||
expect(exitRecord).toMatchObject({
|
||||
workerPid: 12345,
|
||||
signal: "SIGABRT",
|
||||
stderrTail: "dyld: Library not loaded: libsherpa-onnx-c-api.dylib",
|
||||
pendingRequests: [expect.objectContaining({ type: "session.create", kind: "dictationStt" })],
|
||||
});
|
||||
});
|
||||
|
||||
it("does not log intentional shutdowns as worker crashes", async () => {
|
||||
const { logger, records } = createCapturingLogger();
|
||||
const workers: FakeLocalSpeechWorker[] = [];
|
||||
const client = new LocalSpeechWorkerClient({
|
||||
logger,
|
||||
config: {
|
||||
modelsDir: "/tmp/models",
|
||||
voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
|
||||
voiceTtsModel: "kokoro-en-v0_19",
|
||||
},
|
||||
forkWorker: () => {
|
||||
const worker = new FakeLocalSpeechWorker();
|
||||
workers.push(worker);
|
||||
return worker;
|
||||
},
|
||||
});
|
||||
|
||||
const synthesize = client.synthesizeSpeech("hello");
|
||||
workers[0].respond(workers[0].sent[0], {
|
||||
audio: bufferToWorkerBytes(Buffer.from([1])),
|
||||
format: "pcm;rate=24000",
|
||||
});
|
||||
await synthesize;
|
||||
|
||||
client.shutdown();
|
||||
workers[0].emit("close", null, "SIGTERM");
|
||||
|
||||
expect(records.find((record) => record.msg === "Local speech worker exited")).toBeUndefined();
|
||||
expect(
|
||||
records.find((record) => record.msg === "Local speech worker closed after shutdown"),
|
||||
).toMatchObject({
|
||||
workerPid: 12345,
|
||||
signal: "SIGTERM",
|
||||
});
|
||||
});
|
||||
|
||||
it("forwards VAD session events through the shared worker", async () => {
|
||||
const { client, workers } = createClient();
|
||||
const provider = new WorkerBackedTurnDetectionProvider(client);
|
||||
|
||||
@@ -28,6 +28,8 @@ import { bufferToWorkerBytes, workerBytesToBuffer } from "./worker-bytes.js";
|
||||
const DEFAULT_REQUEST_TIMEOUT_MS = 30000;
|
||||
const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000;
|
||||
const DEFAULT_LOCAL_SAMPLE_RATE = 16000;
|
||||
const STDERR_TAIL_MAX_CHARS = 8000;
|
||||
const USER_ERROR_STDERR_MAX_CHARS = 1000;
|
||||
|
||||
type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Request
|
||||
? Request extends LocalSpeechWorkerRequest
|
||||
@@ -38,21 +40,27 @@ type LocalSpeechWorkerRequestInput = LocalSpeechWorkerRequest extends infer Requ
|
||||
interface LocalSpeechWorkerProcess {
|
||||
connected: boolean;
|
||||
killed: boolean;
|
||||
pid?: number;
|
||||
stderr?: Readable | null;
|
||||
send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean;
|
||||
disconnect(): void;
|
||||
kill(): boolean;
|
||||
on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this;
|
||||
on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
|
||||
on(event: "close", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
|
||||
}
|
||||
|
||||
interface PendingRequest {
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (error: Error) => void;
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
type: LocalSpeechWorkerRequest["type"];
|
||||
summary: Record<string, unknown>;
|
||||
startedAt: number;
|
||||
}
|
||||
|
||||
interface LocalSpeechWorkerClientOptions {
|
||||
config: LocalSpeechWorkerConfig;
|
||||
logger: pino.Logger;
|
||||
requestTimeoutMs?: number;
|
||||
idleTtlMs?: number;
|
||||
forkWorker?: () => LocalSpeechWorkerProcess;
|
||||
@@ -90,7 +98,7 @@ function forkLocalSpeechWorker(): LocalSpeechWorkerProcess {
|
||||
env,
|
||||
execArgv: resolveWorkerExecArgv(),
|
||||
serialization: "advanced",
|
||||
stdio: ["ignore", "ignore", "inherit", "ipc"],
|
||||
stdio: ["ignore", "ignore", "pipe", "ipc"],
|
||||
}) as LocalSpeechWorkerProcess;
|
||||
}
|
||||
|
||||
@@ -100,8 +108,75 @@ function isResponse(
|
||||
return message.type === "response";
|
||||
}
|
||||
|
||||
function truncateStart(value: string, maxChars: number): string {
|
||||
if (value.length <= maxChars) {
|
||||
return value;
|
||||
}
|
||||
return value.slice(value.length - maxChars);
|
||||
}
|
||||
|
||||
function summarizeWorkerRequest(message: LocalSpeechWorkerRequest): Record<string, unknown> {
|
||||
switch (message.type) {
|
||||
case "session.create":
|
||||
return { kind: message.kind, sessionId: message.sessionId };
|
||||
case "session.append":
|
||||
return { sessionId: message.sessionId, audioBytes: message.audio.byteLength };
|
||||
case "session.commit":
|
||||
case "session.clear":
|
||||
case "session.flush":
|
||||
case "session.reset":
|
||||
case "session.close":
|
||||
return { sessionId: message.sessionId };
|
||||
case "stt.transcribe":
|
||||
return {
|
||||
model: message.model,
|
||||
audioBytes: message.audio.byteLength,
|
||||
format: message.format,
|
||||
};
|
||||
case "tts.synthesize":
|
||||
return { textLength: message.text.length };
|
||||
}
|
||||
}
|
||||
|
||||
function formatExitStatus(code: number | null, signal: NodeJS.Signals | null): string {
|
||||
const parts: string[] = [];
|
||||
if (code !== null) {
|
||||
parts.push(`code ${code}`);
|
||||
}
|
||||
if (signal) {
|
||||
parts.push(`signal ${signal}`);
|
||||
}
|
||||
return parts.length > 0 ? parts.join(", ") : "unknown exit status";
|
||||
}
|
||||
|
||||
function formatPendingRequestForMessage(request: Record<string, unknown>): string {
|
||||
const type = typeof request.type === "string" ? request.type : "unknown request";
|
||||
const kind = typeof request.kind === "string" ? ` (${request.kind})` : "";
|
||||
return `${type}${kind}`;
|
||||
}
|
||||
|
||||
function buildWorkerExitMessage(params: {
|
||||
code: number | null;
|
||||
signal: NodeJS.Signals | null;
|
||||
pendingRequests: Array<Record<string, unknown>>;
|
||||
stderrTail: string;
|
||||
}): string {
|
||||
const pending =
|
||||
params.pendingRequests.length > 0
|
||||
? ` while handling ${params.pendingRequests
|
||||
.slice(0, 3)
|
||||
.map(formatPendingRequestForMessage)
|
||||
.join(", ")}`
|
||||
: "";
|
||||
const stderr = params.stderrTail
|
||||
? ` Last stderr: ${truncateStart(params.stderrTail, USER_ERROR_STDERR_MAX_CHARS)}`
|
||||
: " Check daemon.log and macOS DiagnosticReports for Paseo Voice crash details.";
|
||||
return `Local speech worker exited (${formatExitStatus(params.code, params.signal)})${pending}.${stderr}`;
|
||||
}
|
||||
|
||||
export class LocalSpeechWorkerClient {
|
||||
private readonly config: LocalSpeechWorkerConfig;
|
||||
private readonly logger: pino.Logger;
|
||||
private readonly requestTimeoutMs: number;
|
||||
private readonly idleTtlMs: number;
|
||||
private readonly forkWorker: () => LocalSpeechWorkerProcess;
|
||||
@@ -109,11 +184,16 @@ export class LocalSpeechWorkerClient {
|
||||
private readonly activeSessionIds = new Set<string>();
|
||||
private readonly sessionEmitters = new Map<string, EventEmitter>();
|
||||
private worker: LocalSpeechWorkerProcess | null = null;
|
||||
private workerPid: number | null = null;
|
||||
private stderrTail = "";
|
||||
private stderrLineBuffer = "";
|
||||
private inFlightRequests = 0;
|
||||
private idleTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private readonly intentionalWorkerCloses = new WeakSet<LocalSpeechWorkerProcess>();
|
||||
|
||||
constructor(options: LocalSpeechWorkerClientOptions) {
|
||||
this.config = options.config;
|
||||
this.logger = options.logger.child({ component: "local-speech-worker-client" });
|
||||
this.requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
|
||||
this.idleTtlMs = options.idleTtlMs ?? DEFAULT_IDLE_TTL_MS;
|
||||
this.forkWorker = options.forkWorker ?? forkLocalSpeechWorker;
|
||||
@@ -214,7 +294,9 @@ export class LocalSpeechWorkerClient {
|
||||
this.sessionEmitters.clear();
|
||||
const worker = this.worker;
|
||||
this.worker = null;
|
||||
this.workerPid = null;
|
||||
if (worker && !worker.killed) {
|
||||
this.intentionalWorkerCloses.add(worker);
|
||||
try {
|
||||
worker.disconnect();
|
||||
} catch {
|
||||
@@ -232,6 +314,7 @@ export class LocalSpeechWorkerClient {
|
||||
const worker = this.ensureWorker();
|
||||
const requestId = randomUUID();
|
||||
const message = { ...input, requestId } as LocalSpeechWorkerRequest;
|
||||
const requestSummary = summarizeWorkerRequest(message);
|
||||
this.inFlightRequests++;
|
||||
this.clearIdleTimer();
|
||||
|
||||
@@ -246,6 +329,9 @@ export class LocalSpeechWorkerClient {
|
||||
resolve: (value) => resolve(value as T),
|
||||
reject,
|
||||
timeout,
|
||||
type: message.type,
|
||||
summary: requestSummary,
|
||||
startedAt: Date.now(),
|
||||
});
|
||||
|
||||
worker.send(message, (error) => {
|
||||
@@ -271,11 +357,44 @@ export class LocalSpeechWorkerClient {
|
||||
}
|
||||
const worker = this.forkWorker();
|
||||
this.worker = worker;
|
||||
this.workerPid = worker.pid ?? null;
|
||||
this.stderrTail = "";
|
||||
this.stderrLineBuffer = "";
|
||||
this.logger.info(
|
||||
{
|
||||
workerPid: this.workerPid,
|
||||
modelsDir: this.config.modelsDir,
|
||||
voiceSttModel: this.config.voiceSttModel,
|
||||
dictationSttModel: this.config.dictationSttModel,
|
||||
voiceTtsModel: this.config.voiceTtsModel,
|
||||
},
|
||||
"Local speech worker spawned",
|
||||
);
|
||||
worker.stderr?.on("data", (chunk: Buffer | string) => this.handleWorkerStderr(chunk));
|
||||
worker.on("message", (message) => this.handleWorkerMessage(message));
|
||||
worker.on("exit", () => this.handleWorkerExit());
|
||||
worker.on("close", (code, signal) => this.handleWorkerExit(worker, code, signal));
|
||||
return worker;
|
||||
}
|
||||
|
||||
private handleWorkerStderr(chunk: Buffer | string): void {
|
||||
const text = Buffer.isBuffer(chunk) ? chunk.toString("utf8") : chunk;
|
||||
this.stderrTail = truncateStart(this.stderrTail + text, STDERR_TAIL_MAX_CHARS);
|
||||
|
||||
this.stderrLineBuffer += text;
|
||||
const lines = this.stderrLineBuffer.split(/\r?\n/);
|
||||
this.stderrLineBuffer = lines.pop() ?? "";
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) {
|
||||
continue;
|
||||
}
|
||||
this.logger.warn(
|
||||
{ workerPid: this.workerPid, stderr: trimmed },
|
||||
"Local speech worker stderr",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private handleWorkerMessage(message: LocalSpeechWorkerToParentMessage): void {
|
||||
if (isResponse(message)) {
|
||||
const pending = this.pendingRequests.get(message.requestId);
|
||||
@@ -317,13 +436,84 @@ export class LocalSpeechWorkerClient {
|
||||
}
|
||||
}
|
||||
|
||||
private handleWorkerExit(): void {
|
||||
this.worker = null;
|
||||
private handleWorkerExit(
|
||||
worker: LocalSpeechWorkerProcess,
|
||||
code: number | null,
|
||||
signal: NodeJS.Signals | null,
|
||||
): void {
|
||||
const wasCurrentWorker = this.worker === worker;
|
||||
const wasIntentionalClose = this.intentionalWorkerCloses.has(worker);
|
||||
this.intentionalWorkerCloses.delete(worker);
|
||||
const workerPid = worker.pid ?? (wasCurrentWorker ? this.workerPid : null);
|
||||
const pendingRequests = this.describePendingRequests();
|
||||
const activeSessionCount = this.activeSessionIds.size;
|
||||
const stderrTail = this.getStderrTail();
|
||||
|
||||
if (wasIntentionalClose) {
|
||||
this.logger.info(
|
||||
{
|
||||
workerPid,
|
||||
code,
|
||||
signal,
|
||||
pendingRequests,
|
||||
activeSessionCount,
|
||||
stderrTail: stderrTail || null,
|
||||
},
|
||||
"Local speech worker closed after shutdown",
|
||||
);
|
||||
if (wasCurrentWorker) {
|
||||
this.worker = null;
|
||||
this.workerPid = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (!wasCurrentWorker) {
|
||||
this.logger.warn(
|
||||
{
|
||||
workerPid,
|
||||
code,
|
||||
signal,
|
||||
pendingRequests,
|
||||
activeSessionCount,
|
||||
stderrTail: stderrTail || null,
|
||||
},
|
||||
"Stale local speech worker closed",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const error = new Error(
|
||||
buildWorkerExitMessage({
|
||||
code,
|
||||
signal,
|
||||
pendingRequests,
|
||||
stderrTail,
|
||||
}),
|
||||
);
|
||||
|
||||
this.logger.error(
|
||||
{
|
||||
err: error,
|
||||
workerPid,
|
||||
code,
|
||||
signal,
|
||||
pendingRequests,
|
||||
activeSessionCount,
|
||||
stderrTail: stderrTail || null,
|
||||
},
|
||||
"Local speech worker exited",
|
||||
);
|
||||
|
||||
if (wasCurrentWorker) {
|
||||
this.worker = null;
|
||||
this.workerPid = null;
|
||||
}
|
||||
this.clearIdleTimer();
|
||||
this.rejectAllPending(new Error("Local speech worker exited"));
|
||||
this.rejectAllPending(error);
|
||||
for (const [sessionId, emitter] of this.sessionEmitters) {
|
||||
if (this.activeSessionIds.has(sessionId)) {
|
||||
emitter.emit("error", new Error("Local speech worker exited"));
|
||||
this.emitErrorIfObserved(sessionId, emitter, error, workerPid);
|
||||
}
|
||||
}
|
||||
this.activeSessionIds.clear();
|
||||
@@ -331,6 +521,24 @@ export class LocalSpeechWorkerClient {
|
||||
this.inFlightRequests = 0;
|
||||
}
|
||||
|
||||
private describePendingRequests(): Array<Record<string, unknown>> {
|
||||
const now = Date.now();
|
||||
return Array.from(this.pendingRequests.entries()).map(([requestId, request]) =>
|
||||
Object.assign(
|
||||
{
|
||||
requestId,
|
||||
type: request.type,
|
||||
ageMs: Math.max(0, now - request.startedAt),
|
||||
},
|
||||
request.summary,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
private getStderrTail(): string {
|
||||
return truncateStart(this.stderrTail.trim(), STDERR_TAIL_MAX_CHARS);
|
||||
}
|
||||
|
||||
private rejectAllPending(error: Error): void {
|
||||
for (const [requestId, pending] of this.pendingRequests) {
|
||||
clearTimeout(pending.timeout);
|
||||
@@ -344,7 +552,27 @@ export class LocalSpeechWorkerClient {
|
||||
if (!emitter) {
|
||||
return;
|
||||
}
|
||||
emitter.emit("error", error instanceof Error ? error : new Error(String(error)));
|
||||
this.emitErrorIfObserved(
|
||||
sessionId,
|
||||
emitter,
|
||||
error instanceof Error ? error : new Error(String(error)),
|
||||
);
|
||||
}
|
||||
|
||||
private emitErrorIfObserved(
|
||||
sessionId: string,
|
||||
emitter: EventEmitter,
|
||||
error: Error,
|
||||
workerPid: number | null = this.workerPid,
|
||||
): void {
|
||||
if (emitter.listenerCount("error") > 0) {
|
||||
emitter.emit("error", error);
|
||||
return;
|
||||
}
|
||||
this.logger.warn(
|
||||
{ err: error, workerPid, sessionId },
|
||||
"Local speech worker session error had no listener",
|
||||
);
|
||||
}
|
||||
|
||||
private scheduleIdleShutdownIfReady(): void {
|
||||
|
||||
Reference in New Issue
Block a user