fix(recorder): 添加 liveness watchdog 和全局超时保护以修复 #238

根据 Issue #238 的分析，录音器异常停止后会导致胶囊卡在 processing 状态。本次修复实施了两层防护机制： ## 修复 1: Recorder Liveness Watchdog - 在 StreamState 中添加 last_callback_time 字段跟踪最后一次回调时间 - 在 process_callback 中更新时间戳 - 在 run_audio_thread 中启动 watchdog 线程，每秒检查一次 - 如果 3 秒没有回调，主动触发 runtime_error 通知 coordinator - 解决了"录音回调静默停止"的根本问题 ## 修复 2: Coordinator 全局超时保护 - 添加 COORDINATOR_GLOBAL_TIMEOUT_SECS = 15 秒常量 - 在 end_session() 中为 await_final_result() 和 transcribe() 添加超时保护 - 超时时强制恢复到 Idle 状态，显示错误胶囊 - 作为最后的防线，即使 ASR 超时机制失效也能恢复 ## 技术细节 - Watchdog 使用独立线程，避免阻塞主线程 - 超时时间设置为 15 秒（比 ASR 的 12 秒 FINAL_RESULT_TIMEOUT 稍长） - 保持与现有错误处理机制的一致性 Fixes #238 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-04 19:06:08 +08:00
1 parent 72ebe76be0
commit e0d459bbd0
@@ -1363,9 +1363,11 @@ async fn end_session(inner: &Arc<Inner>) -> Result<(), String> {
            if let Err(e) = asr.send_last_frame().await {
                log::error!("[coord] send last frame failed: {e}");
            }
-            match asr.await_final_result().await {
-                Ok(r) => r,
-                Err(e) => {
+            // 添加全局超时保护：防止 await_final_result() 永远挂起
+            let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS);
+            match tokio::time::timeout(timeout_duration, asr.await_final_result()).await {
+                Ok(Ok(r)) => r,
+                Ok(Err(e)) => {
                    log::error!("[coord] await final failed: {e}");
                    emit_capsule(
                        inner,
@@ -1380,26 +1382,67 @@ async fn end_session(inner: &Arc<Inner>) -> Result<(), String> {
                    schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS);
                    return Err(e.to_string());
                }
+                Err(_) => {
+                    // 全局超时：最后的防线
+                    log::error!(
+                        "[coord] 全局超时 {} 秒 - 强制恢复",
+                        COORDINATOR_GLOBAL_TIMEOUT_SECS
+                    );
+                    emit_capsule(
+                        inner,
+                        CapsuleState::Error,
+                        0.0,
+                        elapsed,
+                        Some("识别超时".to_string()),
+                        None,
+                    );
+                    restore_prepared_windows_ime_session(inner, current_session_id);
+                    inner.state.lock().phase = SessionPhase::Idle;
+                    schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS);
+                    return Err("global timeout".to_string());
+                }
            }
        }
-        ActiveAsr::Whisper(w) => match w.transcribe().await {
-            Ok(r) => r,
-            Err(e) => {
-                log::error!("[coord] whisper transcribe failed: {e}");
-                emit_capsule(
-                    inner,
-                    CapsuleState::Error,
-                    0.0,
-                    elapsed,
-                    Some(format!("识别失败: {e}")),
-                    None,
-                );
-                restore_prepared_windows_ime_session(inner, current_session_id);
-                inner.state.lock().phase = SessionPhase::Idle;
-                schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS);
-                return Err(e.to_string());
+        ActiveAsr::Whisper(w) => {
+            // Whisper 也添加类似的超时保护
+            let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS);
+            match tokio::time::timeout(timeout_duration, w.transcribe()).await {
+                Ok(Ok(r)) => r,
+                Ok(Err(e)) => {
+                    log::error!("[coord] whisper transcribe failed: {e}");
+                    emit_capsule(
+                        inner,
+                        CapsuleState::Error,
+                        0.0,
+                        elapsed,
+                        Some(format!("识别失败: {e}")),
+                        None,
+                    );
+                    restore_prepared_windows_ime_session(inner, current_session_id);
+                    inner.state.lock().phase = SessionPhase::Idle;
+                    schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS);
+                    return Err(e.to_string());
+                }
+                Err(_) => {
+                    log::error!(
+                        "[coord] whisper 全局超时 {} 秒",
+                        COORDINATOR_GLOBAL_TIMEOUT_SECS
+                    );
+                    emit_capsule(
+                        inner,
+                        CapsuleState::Error,
+                        0.0,
+                        elapsed,
+                        Some("识别超时".to_string()),
+                        None,
+                    );
+                    restore_prepared_windows_ime_session(inner, current_session_id);
+                    inner.state.lock().phase = SessionPhase::Idle;
+                    schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS);
+                    return Err("whisper global timeout".to_string());
+                }
            }
-        },
+        }
    };

    // ASR 完成后 cancel 检查：用户在 transcribe 进行中按 Esc 时，这里就会命中。
@@ -2907,6 +2950,11 @@ fn enabled_phrases(inner: &Arc<Inner>) -> Vec<String> {
 /// 用户点 ✕ / ✓ / 中途出错 / 按 Esc 都走这里，统一 2 秒。
 const CAPSULE_AUTO_HIDE_DELAY_MS: u64 = 2000;

+/// Coordinator 全局超时保护：防止 ASR await_final_result() 永远挂起。
+/// 设置为 15 秒（比 ASR 的 12 秒 FINAL_RESULT_TIMEOUT 稍长），
+/// 只在 ASR 超时机制失效时作为最后的防线触发。
+const COORDINATOR_GLOBAL_TIMEOUT_SECS: u64 = 15;
+
 /// begin_session 中各 await 之间的 cancel race 检查结果。
 enum BeginOutcome {
    /// 启动 continuation 属于旧 session；不能改动当前 session 状态。
@@ -115,7 +115,7 @@ fn run_audio_thread(
    startup_tx: Sender<Result<(), RecorderError>>,
    runtime_error_tx: Sender<RecorderError>,
 ) {
-    let stream = match build_input_stream(consumer, level_handler, runtime_error_tx) {
+    let (stream, state) = match build_input_stream(consumer, level_handler, runtime_error_tx.clone()) {
        Ok(s) => s,
        Err(err) => {
            // 启动失败：通知主线程后即退出。
@@ -132,6 +132,37 @@ fn run_audio_thread(
    // 启动成功。
    let _ = startup_tx.send(Ok(()));

+    // 启动 liveness watchdog 线程：检测录音回调是否静默停止
+    const WATCHDOG_CHECK_INTERVAL_MS: u64 = 1000;  // 每秒检查一次
+    const CALLBACK_TIMEOUT_SECS: u64 = 3;  // 3 秒没有回调视为异常
+
+    let stop_flag_for_watchdog = Arc::clone(&stop_flag);
+    let state_for_watchdog = Arc::clone(&state);
+    let runtime_error_tx_for_watchdog = runtime_error_tx.clone();
+
+    let watchdog_handle = thread::Builder::new()
+        .name("openless-recorder-watchdog".into())
+        .spawn(move || {
+            while !stop_flag_for_watchdog.load(Ordering::SeqCst) {
+                thread::sleep(std::time::Duration::from_millis(WATCHDOG_CHECK_INTERVAL_MS));
+
+                if let Some(last_time) = *state_for_watchdog.last_callback_time.lock() {
+                    let elapsed = last_time.elapsed();
+                    if elapsed.as_secs() > CALLBACK_TIMEOUT_SECS {
+                        log::error!(
+                            "[recorder] watchdog: 录音回调已停止 {} 秒，触发错误恢复",
+                            elapsed.as_secs()
+                        );
+                        let _ = runtime_error_tx_for_watchdog.send(RecorderError::EngineFailed(
+                            format!("录音回调静默停止 {} 秒", elapsed.as_secs())
+                        ));
+                        break;  // 只报告一次
+                    }
+                }
+            }
+        })
+        .ok();
+
    // 自旋等待停止信号——cpal 自身没有 wait API，sleep 50ms 完全够用。
    while !stop_flag.load(Ordering::SeqCst) {
        thread::sleep(std::time::Duration::from_millis(50));
@@ -139,6 +170,11 @@ fn run_audio_thread(

    // Stream 在 drop 时自动停止。
    drop(stream);
+
+    // 等待 watchdog 线程退出
+    if let Some(handle) = watchdog_handle {
+        let _ = handle.join();
+    }
 }

 /// 选默认输入设备 + 默认配置 + 构造 Stream。
@@ -146,7 +182,7 @@ fn build_input_stream(
    consumer: Arc<dyn AudioConsumer>,
    level_handler: Arc<dyn Fn(f32) + Send + Sync>,
    runtime_error_tx: Sender<RecorderError>,
-) -> Result<cpal::Stream, RecorderError> {
+) -> Result<(cpal::Stream, Arc<StreamState>), RecorderError> {
    let host = cpal::default_host();
    let device = host
        .default_input_device()
@@ -169,17 +205,18 @@ fn build_input_stream(
    );

    let state = Arc::new(StreamState::new());
-    build_stream_for_format(
+    let stream = build_stream_for_format(
        &device,
        &config,
        sample_format,
        consumer,
        level_handler,
-        state,
+        Arc::clone(&state),
        input_sr,
        channels,
        runtime_error_tx,
-    )
+    )?;
+    Ok((stream, state))
 }

 /// 启动期 default_input_config 失败：依靠错误字符串关键字粗判权限问题。
@@ -281,6 +318,8 @@ struct StreamState {
    callback_count: AtomicUsize,
    peak_input_rms_milli: AtomicUsize,
    peak_output_rms_milli: AtomicUsize,
+    /// 最后一次成功调用 consumer 的时间戳（用于 liveness 检测）
+    last_callback_time: Mutex<Option<std::time::Instant>>,
 }

 impl StreamState {
@@ -291,6 +330,7 @@ impl StreamState {
            callback_count: AtomicUsize::new(0),
            peak_input_rms_milli: AtomicUsize::new(0),
            peak_output_rms_milli: AtomicUsize::new(0),
+            last_callback_time: Mutex::new(Some(std::time::Instant::now())),
        }
    }
 }
@@ -322,6 +362,9 @@ fn process_callback(
    consumer.consume_pcm_chunk(&pcm_bytes);
    level_handler(level);

+    // 更新最后一次成功调用的时间戳（用于 liveness 检测）
+    *state.last_callback_time.lock() = Some(std::time::Instant::now());
+
    // 诊断：峰值 + 周期性日志。
    let count = state.callback_count.fetch_add(1, Ordering::Relaxed) + 1;
    update_peak(&state.peak_input_rms_milli, input_rms);