fix: decouple livekit audio drain from turn processing

cai

2026-07-10 2776a3104f076b1dbf33452ad88cb60036c39702

 src/main.rs

@@ -3,6 +3,7 @@
mod service;

use std::{
    borrow::Cow,
    env, fs,
    path::{Path, PathBuf},
    sync::{
@@ -32,12 +33,18 @@
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::time::{sleep, sleep_until};
use tokio::{sync::mpsc::UnboundedReceiver, task::JoinHandle};
use tokio::time::{sleep, sleep_until, timeout};
use tokio::{
    sync::{mpsc, mpsc::UnboundedReceiver, watch},
    task::JoinHandle,
};
use tracing::{info, warn};

const USER_AUDIO_SAMPLE_RATE_HZ: u32 = 48_000;
const USER_AUDIO_NUM_CHANNELS: u16 = 1;
const INBOUND_AUDIO_QUEUE_CAPACITY: usize = 100;
const INBOUND_AUDIO_DROP_LOG_INTERVAL: u64 = 100;
const AUDIO_DRAIN_STOP_GRACE: Duration = Duration::from_secs(2);
const DEFAULT_BOT_AUDIO_PROFILE: &str = "pcm-16k";
const LIVEKIT_48K_SAMPLE_RATE_HZ: u32 = 48_000;
const PCM_16K_SAMPLE_RATE_HZ: u32 = 16_000;
@@ -2469,6 +2476,68 @@
            i32::from(USER_AUDIO_NUM_CHANNELS),
        );
        let started_at = Instant::now();
        let (frame_tx, mut frame_rx) =
            mpsc::channel::<DrainedUserAudioFrame>(INBOUND_AUDIO_QUEUE_CAPACITY);
        let (drain_shutdown_tx, mut drain_shutdown_rx) = watch::channel(false);
        let drain_call_id = call_id.clone();
        let drain_trace_id = trace_id.clone();
        let drain_task = tokio::spawn(async move {
            let mut received_frame_count: u64 = 0;
            let mut dropped_frame_count: u64 = 0;
            loop {
                tokio::select! {
                    changed = drain_shutdown_rx.changed() => {
                        match changed {
                            Ok(()) if *drain_shutdown_rx.borrow() => break,
                            Ok(()) => {}
                            Err(_) => break,
                        }
                    }
                    maybe_frame = stream.next() => {
                        let Some(frame) = maybe_frame else {
                            break;
                        };
                        received_frame_count = received_frame_count.saturating_add(1);
                        let drained = DrainedUserAudioFrame {
                            frame_index: received_frame_count,
                            captured_elapsed_ms: started_at.elapsed().as_millis() as u64,
                            frame: AudioFrame {
                                data: Cow::Owned(frame.data.as_ref().to_vec()),
                                sample_rate: frame.sample_rate,
                                num_channels: frame.num_channels,
                                samples_per_channel: frame.samples_per_channel,
                            },
                        };
                        match frame_tx.try_send(drained) {
                            Ok(()) => {}
                            Err(mpsc::error::TrySendError::Full(_)) => {
                                dropped_frame_count = dropped_frame_count.saturating_add(1);
                                if dropped_frame_count % INBOUND_AUDIO_DROP_LOG_INTERVAL == 1 {
                                    warn!(
                                        call_id = %drain_call_id,
                                        trace_id = %drain_trace_id,
                                        dropped_frame_count,
                                        received_frame_count,
                                        queue_capacity = INBOUND_AUDIO_QUEUE_CAPACITY,
                                        "runtime helper inbound_audio_queue_full_dropping_newest"
                                    );
                                }
                            }
                            Err(mpsc::error::TrySendError::Closed(_)) => break,
                        }
                    }
                }
            }
            stream.close();
            info!(
                call_id = %drain_call_id,
                trace_id = %drain_trace_id,
                received_frame_count,
                dropped_frame_count,
                queue_capacity = INBOUND_AUDIO_QUEUE_CAPACITY,
                "runtime helper user_audio_drain_ended"
            );
        });
        let mut frame_count: u64 = 0;
        let mut sample_count: u64 = 0;
        let mut simple_vad = if simple_vad_enabled {
@@ -2478,10 +2547,11 @@
        };
        let mut realtime_asr_upload: Option<RealtimeAsrUpload> = None;

        while let Some(frame) = stream.next().await {
            frame_count += 1;
        while let Some(drained) = frame_rx.recv().await {
            let frame = drained.frame;
            frame_count = drained.frame_index;
            sample_count += u64::from(frame.samples_per_channel) * u64::from(frame.num_channels);
            let elapsed_ms = started_at.elapsed().as_millis() as u64;
            let elapsed_ms = drained.captured_elapsed_ms;
            if frame_count == 1 {
                info!(
                    call_id = %call_id,
@@ -2641,6 +2711,21 @@
        if let Some(upload) = realtime_asr_upload.take() {
            upload.cancel("stream_end").await;
        }
        let _ = drain_shutdown_tx.send(true);
        let mut drain_task = drain_task;
        if timeout(AUDIO_DRAIN_STOP_GRACE, &mut drain_task)
            .await
            .is_err()
        {
            drain_task.abort();
            let _ = drain_task.await;
            warn!(
                call_id = %call_id,
                trace_id = %trace_id,
                stop_grace_ms = AUDIO_DRAIN_STOP_GRACE.as_millis() as u64,
                "runtime helper user_audio_drain_stop_timeout"
            );
        }

        info!(
            call_id = %call_id,
@@ -2655,6 +2740,12 @@
    })
}

struct DrainedUserAudioFrame {
    frame_index: u64,
    captured_elapsed_ms: u64,
    frame: AudioFrame<'static>,
}

async fn finish_realtime_asr_upload(
    upload: RealtimeAsrUpload,
    call_id: &str,