Fix race condition causing "Writer is not bound to a WritableStream" error in Silero VAD (#786)

Author: toubatbrian

.changeset/soft-memes-drive.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents-plugin-silero': patch
+---
+
+Fix race condition causing "Writer is not bound to a WritableStream" error in Silero VAD

agents/src/vad.ts

@@ -177,6 +177,24 @@ export abstract class VADStream implements AsyncIterableIterator<VADEvent> {
     }
   }
 
+  /**
+   * Safely send a VAD event to the output stream, handling writer release errors during shutdown.
+   * @returns true if the event was sent, false if the stream is closing
+   * @throws Error if an unexpected error occurs
+   */
+  protected sendVADEvent(event: VADEvent): boolean {
+    if (this.closed) {
+      return false;
+    }
+
+    try {
+      this.outputWriter.write(event);
+      return true;
+    } catch (e) {
+      throw e;
+    }
+  }
+
   updateInputStream(audioStream: ReadableStream<AudioFrame>) {
     this.deferredInputStream.setSource(audioStream);
   }

plugins/silero/src/vad.ts

@@ -260,26 +260,30 @@ export class VADStream extends baseStream {
             pubSilenceDuration += windowDuration;
           }
 
-          this.outputWriter.write({
-            type: VADEventType.INFERENCE_DONE,
-            samplesIndex: pubCurrentSample,
-            timestamp: pubTimestamp,
-            silenceDuration: pubSilenceDuration,
-            speechDuration: pubSpeechDuration,
-            probability: p,
-            inferenceDuration,
-            frames: [
-              new AudioFrame(
-                inputFrame.data.subarray(0, toCopyInt),
-                this.#inputSampleRate,
-                1,
-                toCopyInt,
-              ),
-            ],
-            speaking: pubSpeaking,
-            rawAccumulatedSilence: silenceThresholdDuration,
-            rawAccumulatedSpeech: speechThresholdDuration,
-          });
+          if (
+            !this.sendVADEvent({
+              type: VADEventType.INFERENCE_DONE,
+              samplesIndex: pubCurrentSample,
+              timestamp: pubTimestamp,
+              silenceDuration: pubSilenceDuration,
+              speechDuration: pubSpeechDuration,
+              probability: p,
+              inferenceDuration,
+              frames: [
+                new AudioFrame(
+                  inputFrame.data.subarray(0, toCopyInt),
+                  this.#inputSampleRate,
+                  1,
+                  toCopyInt,
+                ),
+              ],
+              speaking: pubSpeaking,
+              rawAccumulatedSilence: silenceThresholdDuration,
+              rawAccumulatedSpeech: speechThresholdDuration,
+            })
+          ) {
+            continue;
+          }
 
           const resetWriteCursor = () => {
             if (!this.#speechBuffer) throw new Error('speechBuffer is empty');
@@ -314,19 +318,23 @@ export class VADStream extends baseStream {
               pubSilenceDuration = 0;
               pubSpeechDuration = speechThresholdDuration;
 
-              this.outputWriter.write({
-                type: VADEventType.START_OF_SPEECH,
-                samplesIndex: pubCurrentSample,
-                timestamp: pubTimestamp,
-                silenceDuration: pubSilenceDuration,
-                speechDuration: pubSpeechDuration,
-                probability: p,
-                inferenceDuration,
-                frames: [copySpeechBuffer()],
-                speaking: pubSpeaking,
-                rawAccumulatedSilence: 0,
-                rawAccumulatedSpeech: 0,
-              });
+              if (
+                !this.sendVADEvent({
+                  type: VADEventType.START_OF_SPEECH,
+                  samplesIndex: pubCurrentSample,
+                  timestamp: pubTimestamp,
+                  silenceDuration: pubSilenceDuration,
+                  speechDuration: pubSpeechDuration,
+                  probability: p,
+                  inferenceDuration,
+                  frames: [copySpeechBuffer()],
+                  speaking: pubSpeaking,
+                  rawAccumulatedSilence: 0,
+                  rawAccumulatedSpeech: 0,
+                })
+              ) {
+                continue;
+              }
             }
           } else {
             silenceThresholdDuration += windowDuration;
@@ -341,19 +349,23 @@ export class VADStream extends baseStream {
               pubSpeechDuration = 0;
               pubSilenceDuration = silenceThresholdDuration;
 
-              this.outputWriter.write({
-                type: VADEventType.END_OF_SPEECH,
-                samplesIndex: pubCurrentSample,
-                timestamp: pubTimestamp,
-                silenceDuration: pubSilenceDuration,
-                speechDuration: pubSpeechDuration,
-                probability: p,
-                inferenceDuration,
-                frames: [copySpeechBuffer()],
-                speaking: pubSpeaking,
-                rawAccumulatedSilence: 0,
-                rawAccumulatedSpeech: 0,
-              });
+              if (
+                !this.sendVADEvent({
+                  type: VADEventType.END_OF_SPEECH,
+                  samplesIndex: pubCurrentSample,
+                  timestamp: pubTimestamp,
+                  silenceDuration: pubSilenceDuration,
+                  speechDuration: pubSpeechDuration,
+                  probability: p,
+                  inferenceDuration,
+                  frames: [copySpeechBuffer()],
+                  speaking: pubSpeaking,
+                  rawAccumulatedSilence: 0,
+                  rawAccumulatedSpeech: 0,
+                })
+              ) {
+                continue;
+              }
 
               resetWriteCursor();
             }