From 47b001f9a864dc829f7d0c911ef3ac7fd15ad2a1 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Fri, 10 Jul 2026 18:55:44 +0800
Subject: [PATCH] fix: clean up realtime asr finish failures
---
src/asr_realtime.rs | 115 ++++++++++++++++++++++++++++++++++++++++++++++++++-------
1 files changed, 101 insertions(+), 14 deletions(-)
diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index db28b8c..16ab4d1 100644
--- a/src/asr_realtime.rs
+++ b/src/asr_realtime.rs
@@ -15,8 +15,7 @@
const SAMPLE_RATE_16K: u32 = 16_000;
const CHANNELS_MONO: u32 = 1;
-const MIN_CHUNK_DURATION_MS: u64 = 20;
-const MAX_CHUNK_DURATION_MS: u64 = 1_000;
+const REQUIRED_CHUNK_DURATION_MS: u64 = 200;
const UPLOAD_QUEUE_CAPACITY: usize = 64;
const MAX_NDJSON_LINE_BYTES: usize = 24 * 1024;
const FINISH_RESPONSE_TIMEOUT: Duration = Duration::from_secs(15);
@@ -37,9 +36,7 @@
&& non_blank(&self.url)
&& non_blank(&self.runtime_token)
&& non_blank(&self.runtime_session_nonce)
- && self.chunk_duration_ms >= MIN_CHUNK_DURATION_MS
- && self.chunk_duration_ms <= MAX_CHUNK_DURATION_MS
- && self.chunk_duration_ms % MIN_CHUNK_DURATION_MS == 0
+ && self.chunk_duration_ms == REQUIRED_CHUNK_DURATION_MS
}
}
@@ -115,21 +112,27 @@
speech_duration_ms: u64,
end_reason: &str,
) -> Result<RealtimeAsrOutcome> {
- if let Some(chunk) = self.chunker.flush() {
- let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
- self.try_send_line(line)?;
+ let enqueue_result = (|| -> Result<()> {
+ if let Some(chunk) = self.chunker.flush() {
+ let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
+ self.try_send_line(line)?;
+ }
+ self.try_send_line(vad_speech_end_line(speech_duration_ms, end_reason)?)?;
+ self.try_send_line(finish_line()?)?;
+ Ok(())
+ })();
+ if let Err(error) = enqueue_result {
+ self.abort_and_wait().await;
+ return Err(error);
}
- self.try_send_line(vad_speech_end_line(speech_duration_ms, end_reason)?)?;
- self.try_send_line(finish_line()?)?;
self.sender.take();
let chunk_count = self.chunker.chunk_count;
let audio_bytes = self.chunker.audio_bytes;
- let mut task = self.task;
- let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut task).await {
+ let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut self.task).await {
Ok(joined) => joined.context("realtime asr upload task failed")??,
Err(_) => {
- task.abort();
+ self.abort_and_wait().await;
return Err(anyhow!("realtime asr finish response timeout"));
}
};
@@ -147,8 +150,14 @@
.await
.is_err()
{
- self.task.abort();
+ self.abort_and_wait().await;
}
+ }
+
+ async fn abort_and_wait(&mut self) {
+ self.sender.take();
+ self.task.abort();
+ let _ = timeout(CANCEL_RESPONSE_TIMEOUT, &mut self.task).await;
}
fn try_send_line(&self, line: Vec<u8>) -> Result<()> {
@@ -416,6 +425,48 @@
assert!(line.len() <= MAX_NDJSON_LINE_BYTES);
}
+ #[test]
+ fn realtime_config_rejects_chunk_duration_other_than_frozen_200ms() {
+ let mut config = fixture_config("http://127.0.0.1/realtime".to_string());
+ assert!(config.is_ready());
+
+ config.chunk_duration_ms = 500;
+ assert!(!config.is_ready());
+ config.chunk_duration_ms = 1_000;
+ assert!(!config.is_ready());
+ }
+
+ #[tokio::test]
+ async fn finish_when_tail_chunk_enqueue_fails_then_stops_upload_task() {
+ let (mut upload, task_active) = blocked_upload(true);
+ upload.chunker.push_48k(&vec![1; 480]);
+
+ let result = upload.finish(10, "silence").await;
+
+ assert!(result.is_err());
+ assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
+ }
+
+ #[tokio::test]
+ async fn finish_when_speech_end_enqueue_fails_then_stops_upload_task() {
+ let (upload, task_active) = blocked_upload(true);
+
+ let result = upload.finish(200, "silence").await;
+
+ assert!(result.is_err());
+ assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
+ }
+
+ #[tokio::test]
+ async fn finish_when_finish_enqueue_fails_then_stops_upload_task() {
+ let (upload, task_active) = blocked_upload(false);
+
+ let result = upload.finish(200, "silence").await;
+
+ assert!(result.is_err());
+ assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
+ }
+
#[tokio::test]
async fn upload_streams_canonical_ndjson_and_reads_final_response() {
let response = json!({
@@ -500,6 +551,42 @@
}
}
+ fn blocked_upload(
+ prefill_queue: bool,
+ ) -> (
+ RealtimeAsrUpload,
+ std::sync::Arc<std::sync::atomic::AtomicBool>,
+ ) {
+ let (sender, receiver) = mpsc::channel(1);
+ if prefill_queue {
+ sender.try_send(vec![b'x']).expect("prefill queue");
+ }
+ let task_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
+ let task_state = task_active.clone();
+ let active_guard = TestUploadTaskGuard(task_state);
+ let task = tokio::spawn(async move {
+ let _active_guard = active_guard;
+ let _receiver = receiver;
+ std::future::pending::<Result<RealtimeAsrOutcome>>().await
+ });
+ (
+ RealtimeAsrUpload {
+ sender: Some(sender),
+ task,
+ chunker: Pcm16kChunker::new(200),
+ },
+ task_active,
+ )
+ }
+
+ struct TestUploadTaskGuard(std::sync::Arc<std::sync::atomic::AtomicBool>);
+
+ impl Drop for TestUploadTaskGuard {
+ fn drop(&mut self) {
+ self.0.store(false, std::sync::atomic::Ordering::Release);
+ }
+ }
+
struct CapturedRequest {
headers: String,
body: String,
--
Gitblit v1.9.3