fix(desktop): 修尾部音频截断——控制帧改由转发线程排空音频后补发
ci / server (push) Failing after 11s
ci / design-system (push) Failing after 12s

松开时 capture.take() 异步停采集,通道里可能还缓冲着最后几帧未发;原 stop() 主线程立即 ws.send(Stop) 会抢在这几帧前入队,网关收到 stop 即关会话、丢弃晚到音频,表现为最后约半秒偶发不识别。改为:stop() 只写 canceled 标志并停采集,由音频转发线程排空通道后再发 Stop/Cancel,保证控制帧排在所有音频帧之后(同一 WS 写入队列,FIFO)。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-07-12 19:36:27 +08:00
parent 4b50ff2ce6
commit f8d3e0d1e4
+33 -15
View File
@@ -4,8 +4,9 @@
use parking_lot::Mutex; use parking_lot::Mutex;
use serde_json::{json, Value}; use serde_json::{json, Value};
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc; use std::sync::mpsc;
use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use tauri::{AppHandle, Emitter, EventTarget, Manager}; use tauri::{AppHandle, Emitter, EventTarget, Manager};
use tokio::sync::Notify; use tokio::sync::Notify;
@@ -24,12 +25,14 @@ pub struct DictationState {
} }
struct Session { struct Session {
id: String,
/// 会话代际(18A/18B):每次 start 自增,on_server_msg / 收尾任务校验后才生效。 /// 会话代际(18A/18B):每次 start 自增,on_server_msg / 收尾任务校验后才生效。
epoch: u64, epoch: u64,
capture: Option<crate::audio::Capture>, capture: Option<crate::audio::Capture>,
started: Instant, started: Instant,
got_first_partial: bool, got_first_partial: bool,
/// 收尾语义(true=取消):stop() 在停采集前写入,转发线程排空音频后据此
/// 发 Stop/Cancel 帧,保证控制帧在全部音频帧之后到网关(尾部不丢,见 stop)。
canceled: Arc<AtomicBool>,
} }
pub fn start(app: &AppHandle) { pub fn start(app: &AppHandle) {
@@ -71,23 +74,42 @@ pub fn start(app: &AppHandle) {
None None
} }
}; };
// 收尾语义标志:stop() 停采集前写入;转发线程排空音频后读取并发控制帧。
let canceled = Arc::new(AtomicBool::new(false));
{ {
let ws = ws.clone(); let ws = ws.clone();
let sid = session_id.clone();
let canceled = canceled.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
while let Ok(frame) = rx.recv() { // 排空音频通道:capture 被 dropstop 里)→ tx 关闭 → 这里 recv 收完
if ws.send(crate::ws::WsCmd::Audio(frame)).is_err() { // 剩余缓冲帧后返回 Err。ws 写入失败则直接退出,不再补发控制帧。
break; loop {
match rx.recv() {
Ok(frame) => {
if ws.send(crate::ws::WsCmd::Audio(frame)).is_err() {
return;
}
}
Err(_) => break,
} }
} }
// 音频已全部入队,最后才发 Stop/Cancel——保证控制帧排在所有音频帧之后,
// 网关先收完音频再收 stop,尾部半秒不再被截断(竞态修复)。
let cmd = if canceled.load(Ordering::SeqCst) {
crate::ws::WsCmd::Cancel { session_id: sid }
} else {
crate::ws::WsCmd::Stop { session_id: sid }
};
let _ = ws.send(cmd);
}); });
} }
*state.inner.lock() = Some(Session { *state.inner.lock() = Some(Session {
id: session_id,
epoch, epoch,
capture, capture,
started: Instant::now(), started: Instant::now(),
got_first_partial: false, got_first_partial: false,
canceled,
}); });
} }
@@ -104,7 +126,11 @@ pub fn stop(app: &AppHandle, canceled: bool) {
let epoch = sess.epoch; let epoch = sess.epoch;
emit_overlay(app, "hotkey", json!({"state": "up"})); emit_overlay(app, "hotkey", json!({"state": "up"}));
set_tray_tooltip(app, "dudu — 就绪"); set_tray_tooltip(app, "dudu — 就绪");
sess.capture.take(); // 停止采集
// 先写收尾语义,再停采集:转发线程排空音频后据此发 Stop/Cancel
// 保证控制帧排在所有音频帧之后(尾部不被截断)。不在此处直接发控制帧。
sess.canceled.store(canceled, Ordering::SeqCst);
sess.capture.take(); // 停止采集 → 关音频通道 → 转发线程排空后补发控制帧
let buf = app.state::<CommitBuffer>(); let buf = app.state::<CommitBuffer>();
if canceled { if canceled {
@@ -112,14 +138,6 @@ pub fn stop(app: &AppHandle, canceled: bool) {
buf.discard(epoch); buf.discard(epoch);
} }
let ws = (*app.state::<crate::ws::WsHandle>()).clone();
let cmd = if canceled {
crate::ws::WsCmd::Cancel { session_id: sess.id.clone() }
} else {
crate::ws::WsCmd::Stop { session_id: sess.id.clone() }
};
let _ = ws.send(cmd);
// 取尾部 final 完成信号(事件驱动注入,18C)。 // 取尾部 final 完成信号(事件驱动注入,18C)。
let finalized = buf.finalize_signal(); let finalized = buf.finalize_signal();