Error the audio stream when the connection closes before turn.end

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<WebSocketServer> {
+        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<string>((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")
+    })
+})

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}