rebase

lib/tasks/dial.js

@@ -21,7 +21,7 @@ const {parseUri} = require('drachtio-srf');
 const {ANCHOR_MEDIA_ALWAYS,
   JAMBONZ_DIAL_PAI_HEADER,
   JAMBONES_DIAL_SBC_FOR_REGISTERED_USER} = require('../config');
-const { isOnhold, isOpusFirst } = require('../utils/sdp-utils');
+const { isOnhold, isOpusFirst, getLeadingCodec } = require('../utils/sdp-utils');
 const { normalizeJambones } = require('@jambonz/verb-specifications');
 const { selectHostPort } = require('../utils/network');
 const { sleepFor } = require('../utils/helpers');
@@ -158,6 +158,7 @@ class TaskDial extends Task {
 
   get canReleaseMedia() {
     const keepAnchor = this.data.anchorMedia ||
+      this.weAreTranscoding ||
       this.cs.isBackGroundListen ||
       this.cs.onHoldMusic ||
       ANCHOR_MEDIA_ALWAYS ||
@@ -929,7 +930,13 @@ class TaskDial extends Task {
         this.logger.info({err}, 'Dial:_selectSingleDial - Error boosting audio signal');
       }
     }
-
+    /* basic determination to see if call is being transcoded */
+    const codecA = getLeadingCodec(this.epOther.local.sdp);
+    const codecB = getLeadingCodec(this.ep.remote.sdp);
+    this.weAreTranscoding = (codecA !== codecB);
+    if (this.weAreTranscoding) {
+      this.logger.info(`Dial:_selectSingleDial - transcoding from ${codecA} (A leg) to ${codecB} (B leg)`);
+    }
     /* if we can release the media back to the SBC, do so now */
     if (this.canReleaseMedia || this.shouldExitMediaPathEntirely) {
       setTimeout(this._releaseMedia.bind(this, cs, sd, this.shouldExitMediaPathEntirely), 200);

lib/tasks/gather.js

@@ -259,7 +259,7 @@ class TaskGather extends SttTask {
             startDtmfListener();
           }
           this._stopVad();
-          if (!this.killed) {
+          if (!this.killed && !this.resolved) {
             startListening(cs, ep);
             if (this.input.includes('speech') && this.vendor === 'nuance' && this.listenDuringPrompt) {
               this.logger.debug('Gather:exec - starting transcription timers after say completes');
@@ -297,7 +297,7 @@ class TaskGather extends SttTask {
             startDtmfListener();
           }
           this._stopVad();
-          if (!this.killed) {
+          if (!this.killed && !this.resolved) {
             startListening(cs, ep);
             if (this.input.includes('speech') && this.vendor === 'nuance' && this.listenDuringPrompt) {
               this.logger.debug('Gather:exec - starting transcription timers after play completes');
@@ -1173,7 +1173,7 @@ class TaskGather extends SttTask {
   }
 
   async _startFallback(cs, ep, evt) {
-    if (this.canFallback) {
+    if (this.canFallback()) {
       this._stopTranscribing(ep);
       try {
         this.logger.debug('gather:_startFallback');

lib/tasks/say.js

@@ -2,8 +2,9 @@ const assert = require('assert');
 const TtsTask = require('./tts-task');
 const {TaskName, TaskPreconditions} = require('../utils/constants');
 const pollySSMLSplit = require('polly-ssml-split');
-const { SpeechCredentialError } = require('../utils/error');
+const { SpeechCredentialError, NonFatalTaskError } = require('../utils/error');
 const { sleepFor } = require('../utils/helpers');
+const { NON_FANTAL_ERRORS } = require('../utils/constants.json');
 
 /**
  * Discard unmatching responses:
@@ -402,11 +403,19 @@ class TaskSay extends TtsTask {
             this._playResolve = resolve;
             this._playReject = reject;
           });
-          const r = await ep.play(filename);
-          this.logger.debug({r}, 'Say:exec play result');
-          if (r.playbackSeconds == null && r.playbackMilliseconds == null && r.playbackLastOffsetPos == null) {
-            this._playReject(new Error('Playback failed to start'));
+          try {
+            const r = await ep.play(filename);
+            this.logger.debug({r}, 'Say:exec play result');
+            if (r.playbackSeconds == null && r.playbackMilliseconds == null && r.playbackLastOffsetPos == null) {
+              this._playReject(new Error('Playback failed to start'));
+            }
+          } catch (err) {
+            if (NON_FANTAL_ERRORS.includes(err.message)) {
+              throw new NonFatalTaskError(err.message);
+            }
+            throw err;
           }
+
           try {
             // wait for playback-stop event received to confirm if the playback is successful
             await this._playPromise;

lib/tasks/stt-task.js

@@ -171,7 +171,7 @@ class SttTask extends Task {
       try {
         this.sttCredentials = await this._initSpeechCredentials(this.cs, this.vendor, this.label);
       } catch (error) {
-        if (this.canFallback) {
+        if (this.canFallback()) {
           this.notifyError(
             {
               msg: 'ASR error', details:`Invalid vendor ${this.vendor}, Error: ${error}`,
@@ -260,8 +260,19 @@ class SttTask extends Task {
     ep.addCustomEventListener(event, handler);
   }
 
-  removeCustomEventListeners() {
-    this.eventHandlers.forEach((h) => h.ep.removeCustomEventListener(h.event, h.handler));
+  removeCustomEventListeners(ep) {
+    if (ep) {
+      // for specific endpoint
+      this.eventHandlers.filter((h) => h.ep === ep).forEach((h) => {
+        h.ep.removeCustomEventListener(h.event, h.handler);
+      });
+      this.eventHandlers = this.eventHandlers.filter((h) => h.ep !== ep);
+      return;
+    } else {
+      // for all endpoints
+      this.eventHandlers.forEach((h) => h.ep.removeCustomEventListener(h.event, h.handler));
+      this.eventHandlers = [];
+    }
   }
 
   async _initSpeechCredentials(cs, vendor, label) {
@@ -329,11 +340,13 @@ class SttTask extends Task {
     return credentials;
   }
 
-  get canFallback() {
+  canFallback() {
     return this.fallbackVendor && this.isHandledByPrimaryProvider && !this.cs.hasFallbackAsr;
   }
 
-  async _initFallback() {
+  // ep is optional for gather or any verb that have single ep,
+  // but transcribe does need as it might has 2 eps
+  async _initFallback(ep) {
     assert(this.fallbackVendor, 'fallback failed without fallbackVendor configuration');
     this.logger.info(`Failed to use primary STT provider, fallback to ${this.fallbackVendor}`);
     this.isHandledByPrimaryProvider = false;
@@ -346,7 +359,7 @@ class SttTask extends Task {
     this.data.recognizer.label = this.label;
     this.sttCredentials = await this._initSpeechCredentials(this.cs, this.vendor, this.label);
     // cleanup previous listener from previous vendor
-    this.removeCustomEventListeners();
+    this.removeCustomEventListeners(ep);
   }
 
   async compileHintsForCobalt(ep, hostport, model, token, hints) {

lib/tasks/transcribe.js

@@ -70,6 +70,9 @@ class TaskTranscribe extends SttTask {
     this._bufferedTranscripts = [ [], [] ];  // for channel 1 and 2
     this.bugname_prefix = 'transcribe_';
     this.paused = false;
+    // fallback flags
+    this.isHandledByPrimaryProviderForEp1 = true;
+    this.isHandledByPrimaryProviderForEp2 = true;
   }
 
   get name() { return TaskName.Transcribe; }
@@ -776,7 +779,7 @@ class TaskTranscribe extends SttTask {
   }
 
   async _startFallback(cs, _ep, evt) {
-    if (this.canFallback) {
+    if (this.canFallback(_ep)) {
       _ep.stopTranscription({
         vendor: this.vendor,
         bugname: this.bugname,
@@ -786,7 +789,7 @@ class TaskTranscribe extends SttTask {
       try {
         this.notifyError({ msg: 'ASR error',
           details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'in progress'});
-        await this._initFallback();
+        await this._initFallback(_ep);
         let channel = 1;
         if (this.ep !== _ep) {
           channel = 2;
@@ -895,6 +898,41 @@ class TaskTranscribe extends SttTask {
     if (this._asrTimer) clearTimeout(this._asrTimer);
     this._asrTimer = null;
   }
+
+  // We need to keep track the fallback is happened for each endpoint
+  // override the canFallback and _initFallback methods to make sure that
+  // we only fallback once per endpoint
+  // we want to keep track this on task level instead of endpoint level
+  // because the endpoint instance is used across multiple tasks.
+  canFallback(ep) {
+    let isHandledByPrimaryProvider = this.isHandledByPrimaryProvider;
+    if (ep === this.ep) {
+      isHandledByPrimaryProvider = this.isHandledByPrimaryProviderForEp1;
+    } else if (ep === this.ep2) {
+      isHandledByPrimaryProvider = this.isHandledByPrimaryProviderForEp2;
+    }
+
+    const isOneOfEndpointAlreadyFallenBack = !!this.ep && !!this.ep2 &&
+      this.isHandledByPrimaryProviderForEp1 !== this.isHandledByPrimaryProviderForEp2;
+
+    // fallback is configured
+    return this.fallbackVendor &&
+      // has this endpoint already fallen back
+      isHandledByPrimaryProvider &&
+      // in global level, is there any fallback is already happened
+      // one fallen endpoint will mark cs.hasFallbackAsr to true,
+      // so if one endpoint was fallen, the other endpoint would be able to fallback.
+      (isOneOfEndpointAlreadyFallenBack || !this.cs.hasFallbackAsr);
+  }
+
+  _initFallback(ep) {
+    if (ep === this.ep) {
+      this.isHandledByPrimaryProviderForEp1 = false;
+    } else if (ep === this.ep2) {
+      this.isHandledByPrimaryProviderForEp2 = false;
+    }
+    return super._initFallback(ep);
+  }
 }
 
 module.exports = TaskTranscribe;

lib/tasks/tts-task.js

@@ -279,9 +279,9 @@ class TtsTask extends Task {
       }
 
       /* produce an audio segment from the provided text */
-      const generateAudio = async(text) => {
-        if (this.killed) return;
-        if (text.startsWith('silence_stream://')) return text;
+      const generateAudio = async(text, index) => {
+        if (this.killed) return {index, filePath: null};
+        if (text.startsWith('silence_stream://')) return {index, filePath: text};
 
         /* otel: trace time for tts */
         if (!preCache && !this._disableTracing)  {
@@ -310,7 +310,6 @@ class TtsTask extends Task {
             renderForCaching: preCache
           });
           if (!filePath.startsWith('say:')) {
-            this.playbackIds.push(null);
             this.logger.debug(`Say: file ${filePath}, served from cache ${servedFromCache}`);
             if (filePath) cs.trackTmpFile(filePath);
             if (this.otelSpan) {
@@ -338,10 +337,11 @@ class TtsTask extends Task {
                 'id': this.id
               });
             }
+            return {index, filePath, playbackId: null};
           }
           else {
-            this.playbackIds.push(extractPlaybackId(filePath));
-            this.logger.debug({playbackIds: this.playbackIds}, 'Say: a streaming tts api will be used');
+            const playbackId = extractPlaybackId(filePath);
+            this.logger.debug('Say: a streaming tts api will be used');
             const modifiedPath = filePath.replace('say:{', `say:{session-uuid=${ep.uuid},`);
             this.notifyStatus({
               event: 'synthesized-audio',
@@ -350,9 +350,8 @@ class TtsTask extends Task {
               servedFromCache,
               'id': this.id
             });
-            return modifiedPath;
+            return {index, filePath: modifiedPath, playbackId};
           }
-          return filePath;
         } catch (err) {
           this.logger.info({err}, 'Error synthesizing tts');
           if (this.otelSpan) this.otelSpan.end();
@@ -367,8 +366,20 @@ class TtsTask extends Task {
         }
       };
 
-      const arr = this.text.map((t) => (this._validateURL(t) ? t : generateAudio(t)));
-      return (await Promise.all(arr)).filter((fp) => fp && fp.length);
+      // process all text segments in parallel will cause ordering issue
+      // so we attach index to each promise result and sort them later
+
+      const arr = this.text.map((t, index) => (this._validateURL(t) ?
+        Promise.resolve({index, filePath: t, playbackId: null}) : generateAudio(t, index)));
+      const results = await Promise.all(arr);
+      const sorted = results.sort((a, b) => a.index - b.index);
+
+      return sorted
+        .filter((fp) => fp.filePath && fp.filePath.length)
+        .map((r) => {
+          this.playbackIds.push(r.playbackId);
+          return r.filePath;
+        });
     } catch (err) {
       this.logger.info(err, 'TaskSay:exec error');
       throw err;