diff --git a/docs/issues/README.md b/docs/issues/README.md index 75aafee..961c571 100644 --- a/docs/issues/README.md +++ b/docs/issues/README.md @@ -18,13 +18,12 @@ should be mirrored as real GitHub issues on `jakejars/kon`. | RB-01 | [c1-live-session-race.md](c1-live-session-race.md) | `src-tauri/commands/live.rs` | large | | RB-03 | [c4-transcript-profile-fk.md](c4-transcript-profile-fk.md) | `crates/storage/migrations.rs` + `database.rs` | large | -## MAJOR (6 open, 3 resolved) +## MAJOR (5 open, 4 resolved) | # | File | Area | Fix scope | |---|---|---|---| | RB-04 | [run-live-session-monolith.md](run-live-session-monolith.md) | `src-tauri/commands/live.rs` | large | | RB-05 | [poll-inference-channel-fatality.md](poll-inference-channel-fatality.md) | `src-tauri/commands/live.rs` | medium | -| RB-06 | [native-capture-worker-join.md](native-capture-worker-join.md) | `src-tauri/commands/audio.rs` | medium | | RB-08 | [power-assertion-macos-objc2.md](power-assertion-macos-objc2.md) | `src-tauri/commands/power.rs` | medium | | RB-10 | [llm-prompt-preflight.md](llm-prompt-preflight.md) | `crates/llm/lib.rs` | medium | | RB-11 | [keystore-thread-safety.md](keystore-thread-safety.md) | `crates/cloud-providers/keystore.rs` | medium | @@ -34,6 +33,7 @@ should be mirrored as real GitHub issues on `jakejars/kon`. | # | File | Area | Resolution | |---|---|---|---| | RB-02 | [c3-migrations-atomicity.md](c3-migrations-atomicity.md) | `crates/storage/migrations.rs` | Each migration now runs inside a `pool.begin()` / `tx.commit()` transaction alongside its `schema_version` insert. Regression test injects a poisoned v9 migration and asserts neither the partial schema change nor the version row persists. DRY'd `run_migrations_up_to` test helper onto the same code path. | +| RB-06 | [native-capture-worker-join.md](native-capture-worker-join.md) | `src-tauri/commands/audio.rs` | `NativeCaptureState.stop_tx` replaced by `worker: AsyncMutex>`. `CaptureWorker` bundles the stop sender and the spawned task's `JoinHandle`; `stop_worker(worker)` sends stop then `await`s termination. Both `start_native_capture` (prior-worker stop) and `stop_native_capture` use the helper. Removed the 50ms sleep — the join barrier is exact. Two regression tests cover the lifecycle guarantee and the already-exited case. | | RB-07 | [runtime-capabilities-accelerators.md](runtime-capabilities-accelerators.md) | `src-tauri/commands/models.rs` | Introduced `compose_accelerators(whisper_enabled, loader_available, target)` as a pure helper; `supported_accelerators()` reads `cfg(feature = "whisper")`, `vulkan_loader_available()`, and target OS then delegates. `get_runtime_capabilities` uses it in place of the hard-coded `["cpu", "vulkan"]`. Whisper's `supports_gpu` now follows the feature flag. Five regression tests cover all permutations. | | RB-09 | [decoder-partial-audio-on-error.md](decoder-partial-audio-on-error.md) | `crates/audio/decode.rs` | Packet-loop now propagates all non-EOF `SymphoniaError`s as `AudioDecodeFailed`; per-packet decode errors bubble via `?`. Mock-`MediaSource` regression test confirms mid-stream I/O errors surface instead of returning partial audio. | | RB-12 | [hotkey-linux-device-filter.md](hotkey-linux-device-filter.md) | `crates/hotkey/linux.rs` | Extracted `device_supports_combo` helper; `try_attach_device` now reads the configured `HotkeyCombo` from the watch channel and checks support for that trigger key. Four regression tests land in `linux::tests`. | diff --git a/docs/issues/native-capture-worker-join.md b/docs/issues/native-capture-worker-join.md index a270621..fcddfaf 100644 --- a/docs/issues/native-capture-worker-join.md +++ b/docs/issues/native-capture-worker-join.md @@ -4,6 +4,47 @@ **Path:** `src-tauri/src/commands/audio.rs:46-228` **Source:** [2026-04-22 code review](../code-review-2026-04-22.md) **Labels:** release-blocker, major, concurrency, audio +**Status:** RESOLVED (2026-04-22) + +## Resolution + +Introduced `CaptureWorker { stop_tx, join: JoinHandle<()> }` as the +single handle type retained in state. `NativeCaptureState.stop_tx` +(a `std::sync::Mutex>`) became `worker: +tokio::sync::Mutex>` — the async mutex is +required because the stop path awaits the join while holding the +lock, and holding a blocking mutex across an await is a bug pattern +we don't want to ship. + +New helper `stop_worker(worker)` sends the stop signal, drops the +sender, then `join.await`s the task. Errors from join (panic / +cancellation) are logged and swallowed; the caller needs the +synchronisation barrier, not the task's return value. + +Both lifecycle paths route through the helper: + +- `start_native_capture` — before opening a new capture, if a + previous worker is resident, stop it and await termination. + This removes the race where the old worker's final flush could + append to `all_samples` after the new path cleared it. +- `stop_native_capture` — take the worker, stop_worker, then read + `all_samples`. The previous 50ms sleep is no longer needed — the + join barrier is exact. + +Two regression tests in `commands::audio::tests`: + +- `stop_worker_awaits_full_termination_no_writes_after_join` — + synthetic worker bumps an atomic counter in a loop, applies a + flush marker at exit. Post-stop-worker the flush marker must be + set and no further writes must appear on a subsequent sleep. +- `stop_worker_is_idempotent_on_a_worker_that_has_already_exited` — + a task that finished on its own must still be join-able without + hang or panic. + +The full cpal-backed start→stop→start integration test the original +issue asks for is not feasible in a Linux CI without an audio +device. The component test above covers the underlying invariant +the real workflow depends on. ## Problem diff --git a/src-tauri/src/commands/audio.rs b/src-tauri/src/commands/audio.rs index 8a34ccd..f7b3360 100644 --- a/src-tauri/src/commands/audio.rs +++ b/src-tauri/src/commands/audio.rs @@ -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, 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>>, + /// 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>, /// All captured samples (16kHz mono) for save_audio. all_samples: Arc>>, } @@ -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("") ); - // 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 = 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, 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(