race condition where gather transcribe is restarted after final transcript

lib/session/call-session.js

@@ -877,6 +877,7 @@ class CallSession extends Emitter {
       const taskNum = ++this.taskIdx;
       const stackNum = this.stackIdx;
       const task = this.tasks.shift();
+      task._stackNum = `${stackNum}:${taskNum}`;
       this.logger.info(`CallSession:exec starting task #${stackNum}:${taskNum}: ${task.name}`);
       this._notifyTaskStatus(task, {event: 'starting'});
       // Register verbhook span wait for end

lib/tasks/gather.js

@@ -107,7 +107,7 @@ class TaskGather extends SttTask {
   }
 
   async exec(cs, {ep}) {
-    this.logger.debug({options: this.data}, 'Gather:exec');
+    this.logger.debug({options: this.data}, `Gather:exec ${this.stackNum}`);
     await super.exec(cs, {ep});
     const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, cs.srf);
 
@@ -171,6 +171,7 @@ class TaskGather extends SttTask {
             this.logger.info('Gather:exec - task was quickly killed so do not transcribe');
             return;
           }
+          this.logger.debug('Gather:exec - going to start transcribing (startListening)');
           this._startTranscribing(ep);
           return updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid);
         } catch (e) {
@@ -235,9 +236,15 @@ class TaskGather extends SttTask {
 
       if (this.input.includes('speech') && this.listenDuringPrompt) {
         await this._setSpeechHandlers(cs, ep);
-        this._startTranscribing(ep);
+        if (!this.resolved && !this.killed) {
-        updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid)
+          this.logger.debug(`Gather:exec ${this.stackNum} - going to start transcribing (listenDuringPrompt)`);
-          .catch(() => {/*already logged error */});
+          this._startTranscribing(ep);
+          updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid)
+            .catch(() => {/*already logged error */});
+        }
+        else {
+          this.logger.info(`Gather:exec ${this.stackNum} - task was killed or resolved before starting transcription`);
+        }
       }
 
       if (this.input.includes('digits') || this.dtmfBargein || this.asrDtmfTerminationDigit) {
@@ -482,7 +489,7 @@ class TaskGather extends SttTask {
       locale: this.language,
       interim: this.interim,
       bugname: this.bugname
-    }, 'Gather:_startTranscribing');
+    }, `Gather:_startTranscribing ${this.stackNum}`);
 
     /**
      * Note: we don't need to ask deepgram for interim results, because they
@@ -700,7 +707,12 @@ class TaskGather extends SttTask {
     // make sure this is not a transcript from answering machine detection
     const bugname = fsEvent.getHeader('media-bugname');
     const finished = fsEvent.getHeader('transcription-session-finished');
-    this.logger.debug({evt, bugname, finished, vendor: this.vendor}, 'Gather:_onTranscription raw transcript');
+    this.logger.debug({
+      evt,
+      bugname,
+      finished,
+      vendor: this.vendor
+    }, `Gather:_onTranscription ${this.stackNum} raw transcript`);
     if (bugname && this.bugname !== bugname) return;
 
     if (this.vendor === 'ibm' && evt?.state === 'listening') return;
@@ -793,14 +805,21 @@ class TaskGather extends SttTask {
         this._startAsrTimer();
 
         /* some STT engines will keep listening after a final response, so no need to restart */
-        if (!['soniox', 'aws', 'microsoft', 'deepgram'].includes(this.vendor)) this._startTranscribing(ep);
+        if (!['soniox', 'aws', 'microsoft', 'deepgram'].includes(this.vendor)) {
+          this.logger.debug('Gather:_onTranscription - going to start transcribing again (continuous asr)');
+          this._startTranscribing(ep);
+        }
       }
       else {
         if (this.bargein && (words + bufferedWords) < this.minBargeinWordCount) {
           this.logger.debug({evt, words, bufferedWords},
             'TaskGather:_onTranscription - final transcript but < min barge words');
           this._bufferedTranscripts.push(evt);
-          if (!['soniox', 'aws', 'microsoft', 'deepgram'].includes(this.vendor)) this._startTranscribing(ep);
+          if (!['soniox', 'aws', 'microsoft', 'deepgram'].includes(this.vendor)) {
+            this.logger.debug(
+              `Gather:_onTranscription - start transcribing again (min bargein words=${this.minBargeinWordCount}`);
+            this._startTranscribing(ep);
+          }
           return;
         }
         else {
@@ -866,7 +885,7 @@ class TaskGather extends SttTask {
     }
   }
   _onEndOfUtterance(cs, ep) {
-    this.logger.debug('TaskGather:_onEndOfUtterance');
+    this.logger.debug(`TaskGather:_onEndOfUtterance ${this.stackNum}`);
     if (this.bargein && this.minBargeinWordCount === 0) {
       this._killAudio(cs);
     }
@@ -881,6 +900,7 @@ class TaskGather extends SttTask {
      * since we dont have a final transcript yet.
      */
     if (!this.resolved && !this.killed && !this._bufferedTranscripts.length && this.wantsSingleUtterance) {
+      this.logger.debug('Gather:_onEndOfUtterance - start transcribing again (end of utterance/wantsSingleUtterance)');
       this._startTranscribing(ep);
     }
   }
@@ -906,6 +926,7 @@ class TaskGather extends SttTask {
       try {
         await this._fallback();
         await this._initSpeech(cs, ep);
+        this.logger.debug('Gather:_onJambonzError - going to start transcribing again (jambonz error)');
         this._startTranscribing(ep);
         updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid);
         return;
@@ -955,7 +976,7 @@ class TaskGather extends SttTask {
         this.logger.debug('TaskGather:_onNoSpeechDetected for old gather, ignoring');
       }
       else {
-        this.logger.debug('TaskGather:_onNoSpeechDetected - listen again');
+        this.logger.debug('Gather:_onNoSpeechDetected - going to start transcribing again');
         this._startTranscribing(ep);
       }
       return;
@@ -963,7 +984,17 @@ class TaskGather extends SttTask {
   }
 
   async _resolve(reason, evt) {
-    this.logger.debug(`TaskGather:resolve with reason ${reason}`);
+    this.logger.debug(`TaskGather:resolve ${this.stackNum} with reason ${reason}`);
+    if (this.needsStt && this.ep && this.ep.connected) {
+      this.ep.stopTranscription({
+        vendor: this.vendor,
+        bugname: this.bugname
+      })
+        .catch((err) => {
+          if (this.resolved) return;
+          this.logger.error({err}, 'Error stopping transcription');
+        });
+    }
     if (this.resolved) return;
 
     this.resolved = true;
@@ -981,13 +1012,6 @@ class TaskGather extends SttTask {
       'stt.resolve': reason,
       'stt.result': JSON.stringify(evt)
     });
-    if (this.needsStt && this.ep && this.ep.connected) {
-      this.ep.stopTranscription({
-        vendor: this.vendor,
-        bugname: this.bugname
-      })
-        .catch((err) => this.logger.error({err}, 'Error stopping transcription'));
-    }
 
     if (this.callSession && this.callSession.callGone) {
       this.logger.debug('TaskGather:_resolve - call is gone, not invoking web callback');