From 62490049b7d874a2abb777fcc3cbefe294eb08b5 Mon Sep 17 00:00:00 2001 From: 1letme <1letme@users.noreply.github.com> Date: Wed, 8 Jul 2026 21:18:13 +0200 Subject: [PATCH] Error the audio stream when the connection closes before turn.end --- src/MsEdgeTTS.spec.ts | 59 +++++++++++++++++++++++++++++++++++++++++++ src/MsEdgeTTS.ts | 20 +++++++++++++-- 2 files changed, 77 insertions(+), 2 deletions(-) diff --git a/src/MsEdgeTTS.spec.ts b/src/MsEdgeTTS.spec.ts index f0c84bb..772d489 100644 --- a/src/MsEdgeTTS.spec.ts +++ b/src/MsEdgeTTS.spec.ts @@ -4,6 +4,8 @@ import {OUTPUT_EXTENSIONS, OUTPUT_FORMAT} from "./Output" import {mkdirSync, readFileSync, rmSync} from "fs" import {existsSync} from "node:fs" import {join} from "path" +import {AddressInfo} from "net" +import {WebSocketServer} from "ws" describe("MsEdgeTTS onerror", () => { it("should reject with an Error instance containing diagnostic info when the socket errors", async () => { @@ -128,3 +130,60 @@ describe("MsEdgeTTS", () => { }) }) + +describe("MsEdgeTTS truncated connection", () => { + // A local server standing in for Edge: it sends turn.start and one audio frame, then either + // sends turn.end (complete) or closes without it (a dropped/proxied connection mid-synthesis). + function startServer(sendTurnEnd: boolean): Promise { + return new Promise((resolve) => { + const wss = new WebSocketServer({port: 0}) + wss.on("listening", () => resolve(wss)) + wss.on("connection", (socket) => { + socket.on("message", (raw) => { + const match = /X-RequestId:(.*?)\r\n/.exec(raw.toString()) + if (!match) return // speech.config handshake, no request id + const requestId = match[1] + socket.send(`X-RequestId:${requestId}\r\nContent-Type:application/json; charset=utf-8\r\nPath:turn.start\r\n\r\n{}`) + socket.send(Buffer.concat([Buffer.from(`X-RequestId:${requestId}\r\nPath:audio\r\n`), Buffer.from([0, 1, 2, 3])])) + if (sendTurnEnd) { + socket.send(`X-RequestId:${requestId}\r\nContent-Type:application/json; charset=utf-8\r\nPath:turn.end\r\n\r\n{}`) + } + setTimeout(() => socket.close(), 20) + }) + }) + }) + } + + async function synth(sendTurnEnd: boolean): Promise<{ outcome: string, bytes: number }> { + const server = await startServer(sendTurnEnd) + const {port} = server.address() as AddressInfo + jest.spyOn(MsEdgeTTS as any, "getSynthUrl").mockResolvedValue(`ws://127.0.0.1:${port}`) + const tts = new MsEdgeTTS() + try { + await tts.setMetadata("en-US-AriaNeural", OUTPUT_FORMAT.WEBM_24KHZ_16BIT_MONO_OPUS) + const {audioStream} = tts.toStream("Some text that gets cut short.") + let bytes = 0 + const outcome = await new Promise((resolve) => { + audioStream.on("data", (c: Buffer) => bytes += c.length) + audioStream.on("end", () => resolve("end")) + audioStream.on("error", () => resolve("error")) + }) + return {outcome, bytes} + } finally { + tts.close() + server.close() + jest.restoreAllMocks() + } + } + + it("errors the audio stream when the socket closes before turn.end", async () => { + const {outcome, bytes} = await synth(false) + expect(bytes).toBeGreaterThan(0) // partial audio arrived... + expect(outcome).toBe("error") // ...but the truncation is surfaced, not hidden + }) + + it("ends the audio stream normally when turn.end is received", async () => { + const {outcome} = await synth(true) + expect(outcome).toBe("end") + }) +}) diff --git a/src/MsEdgeTTS.ts b/src/MsEdgeTTS.ts index 854ba53..9731b30 100644 --- a/src/MsEdgeTTS.ts +++ b/src/MsEdgeTTS.ts @@ -69,7 +69,7 @@ export class MsEdgeTTS { private _voice private _outputFormat private _metadataOptions: MetadataOptions = new MetadataOptions() - private _streams: { [key: string]: { audio: Readable, metadata: Readable } } = {} + private _streams: { [key: string]: { audio: Readable, metadata: Readable, turnEnded: boolean } } = {} private _startTime = 0 private readonly _agent: Agent @@ -140,6 +140,7 @@ export class MsEdgeTTS { } else if (message.includes(`Path:${messageTypes.TURN_END}`)) { // end of turn, close stream this._log("->", message) + this._streams[requestId].turnEnded = true this._streams[requestId].audio.push(null) } else if (message.includes(`Path:${messageTypes.RESPONSE}`)) { // context response, ignore @@ -165,7 +166,15 @@ export class MsEdgeTTS { this._ws.onclose = () => { this._log("disconnected after:", (Date.now() - this._startTime) / 1000, "seconds") for (const requestId in this._streams) { - this._streams[requestId].audio.push(null) + const stream = this._streams[requestId] + if (stream.turnEnded) { + // synthesis finished normally, just close the stream + stream.audio.push(null) + } else { + // socket closed before turn.end: the audio is truncated, so surface it as an + // error instead of silently ending the stream as if it were complete + stream.audio.destroy(new Error("Stream closed before the synthesis completed (no turn.end received). The audio is likely truncated.")) + } } } this._ws.onerror = (event: any) => { @@ -373,6 +382,12 @@ export class MsEdgeTTS { await Promise.all([ new Promise((resolve, reject) => { const writableAudioFile = audioStream.pipe(fs.createWriteStream(audioFilePath)) + // .pipe() doesn't forward source errors to the writable, so a truncated stream + // would hang here — watch the audio stream directly and reject on its error + audioStream.once("error", (e) => { + writableAudioFile.destroy() + reject(e) + }) writableAudioFile.once("close", async () => { if (writableAudioFile.bytesWritten > 0) { resolve(audioFilePath) @@ -457,6 +472,7 @@ export class MsEdgeTTS { this._streams[requestId] = { audio: audioStream, metadata: metadataStream, + turnEnded: false, } this._send(request).then() return {audioStream, metadataStream, requestId}