fix(rb-06): native capture worker is joined on stop
The accumulator task was fire-and-forget — `tokio::spawn` without
retaining the JoinHandle. `stop_native_capture` sent a stop signal,
slept 50ms, and returned; the worker could still be running its
final flush and appending to `all_samples` when the next
`start_native_capture` cleared it. Rapid start→stop→start could
leak tail samples from one session into another.
Replace `NativeCaptureState.stop_tx` with `worker:
AsyncMutex<Option<CaptureWorker>>`, where CaptureWorker owns both
the stop sender and the spawned task's JoinHandle. New helper
`stop_worker(worker)` sends stop, drops the sender, and `.await`s
the join. Both the prior-worker tear-down in `start_native_capture`
and `stop_native_capture` itself go through the helper, so the
worker is always fully terminated before any downstream read or
next-session cleanup.
AsyncMutex (not std::sync::Mutex) because the stop path awaits
while holding the lock. Also drops the 50ms sleep from
stop_native_capture — the join is an exact barrier.
Two regression tests:
- stop_worker_awaits_full_termination_no_writes_after_join:
synthetic worker with an atomic counter and a flush marker.
After stop_worker the flush must have run and no further
writes may appear.
- stop_worker_is_idempotent_on_a_worker_that_has_already_exited:
tasks that stop themselves must still join cleanly.
A full cpal-backed start→stop→start integration test is not
feasible in Linux CI without an audio device. The component tests
cover the invariant the real flow depends on.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -2,7 +2,8 @@ use std::path::PathBuf;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use tauri::{Emitter, Manager};
|
||||
use tokio::sync::mpsc as tokio_mpsc;
|
||||
use tokio::sync::{mpsc as tokio_mpsc, Mutex as AsyncMutex};
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use kon_audio::{DeviceInfo, MicrophoneCapture};
|
||||
use kon_core::constants::WHISPER_SAMPLE_RATE;
|
||||
@@ -19,10 +20,37 @@ pub async fn list_audio_devices() -> Result<Vec<DeviceInfo>, String> {
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// A running native-capture accumulator worker, held so the command
|
||||
/// layer can both signal it to stop and `await` its termination. RB-06
|
||||
/// replaced a fire-and-forget `tokio::spawn` that let the previous
|
||||
/// worker keep flushing and appending samples after `stop_native_capture`
|
||||
/// returned — a rapid start → stop → start could contaminate the new
|
||||
/// session's samples vector with tail writes from the old one.
|
||||
struct CaptureWorker {
|
||||
stop_tx: tokio_mpsc::Sender<()>,
|
||||
join: JoinHandle<()>,
|
||||
}
|
||||
|
||||
/// Send the stop signal and await full worker termination. Consumes
|
||||
/// `CaptureWorker` because the contained handles are single-use.
|
||||
/// Errors from `join.await` (task panicked or was cancelled) are
|
||||
/// logged and swallowed — the caller only needs the synchronisation
|
||||
/// barrier, not the worker's return value.
|
||||
async fn stop_worker(worker: CaptureWorker) {
|
||||
let _ = worker.stop_tx.send(()).await;
|
||||
drop(worker.stop_tx);
|
||||
if let Err(e) = worker.join.await {
|
||||
eprintln!("[native-capture] worker task did not terminate cleanly: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared state for native microphone capture.
|
||||
pub struct NativeCaptureState {
|
||||
/// Stop signal sender — dropping this stops the accumulator task.
|
||||
stop_tx: Mutex<Option<tokio_mpsc::Sender<()>>>,
|
||||
/// The running accumulator worker, if any. `tokio::sync::Mutex`
|
||||
/// because the fastest-moving consumer (`stop_worker`) awaits while
|
||||
/// holding the lock — a `std::sync::Mutex` would have to be released
|
||||
/// and reacquired around each await point.
|
||||
worker: AsyncMutex<Option<CaptureWorker>>,
|
||||
/// All captured samples (16kHz mono) for save_audio.
|
||||
all_samples: Arc<Mutex<Vec<f32>>>,
|
||||
}
|
||||
@@ -30,7 +58,7 @@ pub struct NativeCaptureState {
|
||||
impl NativeCaptureState {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
stop_tx: Mutex::new(None),
|
||||
worker: AsyncMutex::new(None),
|
||||
all_samples: Arc::new(Mutex::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
@@ -53,13 +81,12 @@ pub async fn start_native_capture(
|
||||
device_name.as_deref().unwrap_or("<auto>")
|
||||
);
|
||||
|
||||
// Stop any existing capture: send an explicit stop signal first, then
|
||||
// drop the sender. The accumulator task watches for `Disconnected` too,
|
||||
// but signalling explicitly avoids the brief race window.
|
||||
// (Codex review 2026/04/17 D1)
|
||||
if let Some(tx) = state.stop_tx.lock().unwrap().take() {
|
||||
let _ = tx.try_send(());
|
||||
drop(tx);
|
||||
// Stop any in-flight worker and AWAIT its termination before opening
|
||||
// a new capture. Without the join we would race a draining worker
|
||||
// against the `all_samples.clear()` below, leaving old-session
|
||||
// samples in the new-session vector (RB-06).
|
||||
if let Some(existing) = state.worker.lock().await.take() {
|
||||
stop_worker(existing).await;
|
||||
}
|
||||
|
||||
// `MicrophoneCapture::start()` is synchronous and may spend up to
|
||||
@@ -91,13 +118,13 @@ pub async fn start_native_capture(
|
||||
all_samples.lock().unwrap().clear();
|
||||
|
||||
let (stop_tx, mut stop_rx) = tokio_mpsc::channel::<()>(1);
|
||||
*state.stop_tx.lock().unwrap() = Some(stop_tx);
|
||||
|
||||
let all_samples_clone = all_samples.clone();
|
||||
|
||||
// Spawn a task that reads cpal chunks, downsamples to 16kHz mono,
|
||||
// and emits events to the frontend
|
||||
tokio::spawn(async move {
|
||||
// and emits events to the frontend. The JoinHandle is retained in
|
||||
// `state.worker` so `stop_native_capture` can await full termination.
|
||||
let join = tokio::spawn(async move {
|
||||
let mut pcm_buffer: Vec<f32> = Vec::new();
|
||||
let chunk_size = 8000_usize; // ~0.5s at 16kHz
|
||||
|
||||
@@ -203,23 +230,24 @@ pub async fn start_native_capture(
|
||||
}
|
||||
});
|
||||
|
||||
*state.worker.lock().await = Some(CaptureWorker { stop_tx, join });
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Stop native microphone capture. Returns all captured samples (16kHz mono).
|
||||
///
|
||||
/// Awaits full worker termination before reading `all_samples`, so the
|
||||
/// returned vector contains every sample the worker flushed — and
|
||||
/// nothing from a worker that technically outlived the call (RB-06).
|
||||
#[tauri::command]
|
||||
pub async fn stop_native_capture(
|
||||
state: tauri::State<'_, NativeCaptureState>,
|
||||
) -> Result<Vec<f32>, String> {
|
||||
// Extract the stop sender without holding the guard across an await
|
||||
let stop_tx = state.stop_tx.lock().unwrap().take();
|
||||
if let Some(tx) = stop_tx {
|
||||
let _ = tx.send(()).await;
|
||||
if let Some(worker) = state.worker.lock().await.take() {
|
||||
stop_worker(worker).await;
|
||||
}
|
||||
|
||||
// Brief delay to let the accumulator flush
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
|
||||
let samples = {
|
||||
let mut all = state.all_samples.lock().unwrap();
|
||||
std::mem::take(&mut *all)
|
||||
@@ -285,7 +313,10 @@ static RECORDING_COUNTER: std::sync::atomic::AtomicU64 =
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::recording_filename;
|
||||
use super::{recording_filename, stop_worker, CaptureWorker};
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
#[test]
|
||||
fn recording_filenames_are_unique_across_rapid_calls() {
|
||||
@@ -328,6 +359,76 @@ mod tests {
|
||||
assert!(parts[2].len() >= 4, "counter component is zero-padded to >=4 digits");
|
||||
assert!(parts[2].chars().all(|c| c.is_ascii_digit()));
|
||||
}
|
||||
|
||||
// RB-06 regression: after `stop_worker(worker).await` completes, the
|
||||
// underlying task must have exited — no lingering writes to shared
|
||||
// state can leak past the stop point. The real native-capture
|
||||
// worker drains a capture queue and appends to `all_samples`; this
|
||||
// test swaps that for a synthetic worker that bumps an atomic
|
||||
// counter in a loop and applies a distinct "flush" marker at exit.
|
||||
// The assertions mirror the real-world invariant a caller needs:
|
||||
// (a) after stop_worker returns, the worker has run its flush;
|
||||
// (b) subsequent sleeps see the counter frozen — no writes occur
|
||||
// after the join barrier.
|
||||
// Pre-fix behaviour (fire-and-forget `tokio::spawn`) failed both:
|
||||
// a start→stop→start cycle could observe tail writes from the
|
||||
// previous worker in the new session's vector.
|
||||
#[tokio::test]
|
||||
async fn stop_worker_awaits_full_termination_no_writes_after_join() {
|
||||
let counter = Arc::new(AtomicU32::new(0));
|
||||
let counter_task = counter.clone();
|
||||
let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1);
|
||||
|
||||
let join = tokio::spawn(async move {
|
||||
loop {
|
||||
if stop_rx.try_recv().is_ok() {
|
||||
break;
|
||||
}
|
||||
counter_task.fetch_add(1, Ordering::SeqCst);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
|
||||
}
|
||||
// Flush marker — mirrors the final pcm_buffer drain in the
|
||||
// real worker. Setting a value with a distinctive high bit
|
||||
// so the test can prove the flush ran.
|
||||
counter_task.fetch_or(0x8000_0000, Ordering::SeqCst);
|
||||
});
|
||||
|
||||
// Let the worker accumulate a few bumps before we signal stop.
|
||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||
|
||||
stop_worker(CaptureWorker { stop_tx, join }).await;
|
||||
|
||||
let after_stop = counter.load(Ordering::SeqCst);
|
||||
assert!(
|
||||
after_stop & 0x8000_0000 != 0,
|
||||
"flush marker must be set post-stop (got {after_stop:#x})"
|
||||
);
|
||||
|
||||
// Post-join, no further writes are possible because the task
|
||||
// has ended. Sleep briefly and re-read to confirm.
|
||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||
let later = counter.load(Ordering::SeqCst);
|
||||
assert_eq!(
|
||||
later, after_stop,
|
||||
"no writes must happen after stop_worker returns"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stop_worker_is_idempotent_on_a_worker_that_has_already_exited() {
|
||||
// A worker that stops itself (channel disconnected, capture
|
||||
// dead, etc.) must still be join-able cleanly by stop_worker —
|
||||
// the helper should swallow any expected "task already done"
|
||||
// condition without panicking.
|
||||
let (stop_tx, _stop_rx) = mpsc::channel::<()>(1);
|
||||
let join = tokio::spawn(async { /* exit immediately */ });
|
||||
|
||||
// Give the task a tick to finish.
|
||||
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
|
||||
|
||||
// This must not hang or panic.
|
||||
stop_worker(CaptureWorker { stop_tx, join }).await;
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn persist_audio_samples(
|
||||
|
||||
Reference in New Issue
Block a user