From acdc818de8b4bbb2baefe2911139ab70bd37d235 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Fri, 15 May 2026 19:30:08 +0530 Subject: [PATCH] fix(voice): atomic install-start guard for Whisper/Piper install RPCs (#1787) Co-authored-by: Claude Opus 4.7 (1M context) --- src/openhuman/composio/client.rs | 72 +++---- src/openhuman/local_ai/schemas.rs | 68 ++++-- src/openhuman/local_ai/schemas_tests.rs | 127 +++++++++++ .../local_ai/voice_install_common.rs | 203 +++++++++++++++++- 4 files changed, 406 insertions(+), 64 deletions(-) diff --git a/src/openhuman/composio/client.rs b/src/openhuman/composio/client.rs index 0a07d8aa3..d3cd100f1 100644 --- a/src/openhuman/composio/client.rs +++ b/src/openhuman/composio/client.rs @@ -150,6 +150,38 @@ impl ComposioClient { // ── Execute ───────────────────────────────────────────────────── + /// `POST /agent-integrations/composio/execute` — single, non-retrying + /// HTTP round-trip. Use this when the caller owns the retry loop + /// (e.g. `auth_retry`) to avoid double-retry. + pub(crate) async fn execute_tool_once( + &self, + tool: &str, + arguments: Option, + ) -> Result { + let tool = tool.trim(); + if tool.is_empty() { + anyhow::bail!("composio.execute_tool_once: tool slug must not be empty"); + } + let arguments = arguments.unwrap_or(serde_json::Value::Object(Default::default())); + tracing::debug!(tool = %tool, "[composio] execute_tool_once (no built-in retry)"); + let body = json!({ "tool": tool, "arguments": arguments }); + let result = self.post_execute_tool(&body).await; + match &result { + Ok(resp) => tracing::debug!( + tool = %tool, + successful = resp.successful, + has_error = resp.error.is_some(), + "[composio] execute_tool_once completed" + ), + Err(err) => tracing::error!( + tool = %tool, + error = %err, + "[composio] execute_tool_once failed" + ), + } + result + } + /// `POST /agent-integrations/composio/execute` — run a Composio /// action and return the provider result + cost. pub async fn execute_tool( @@ -233,46 +265,6 @@ impl ComposioClient { .await } - /// Single-shot `execute_tool` — same body construction and slug validation - /// as [`Self::execute_tool`], but **without** the inner post-OAuth retry - /// that [`Self::execute_tool_with_post_oauth_retry`] performs. Reserved - /// for callers that already own a higher-level retry policy and would - /// otherwise stack two retry layers (4 hits to the gateway instead of 2). - /// In particular, [`super::auth_retry::execute_with_auth_retry`] uses - /// this entry point so its `must retry exactly once` contract still - /// holds after PR #1707 introduced the inner retry. - pub(crate) async fn execute_tool_once( - &self, - tool: &str, - arguments: Option, - ) -> Result { - let tool = tool.trim(); - if tool.is_empty() { - anyhow::bail!("composio.execute_tool_once: tool slug must not be empty"); - } - let arguments = arguments.unwrap_or(serde_json::Value::Object(Default::default())); - tracing::debug!( - tool = %tool, - "[composio] execute_tool_once start" - ); - let body = json!({ "tool": tool, "arguments": arguments }); - let result = self.post_execute_tool(&body).await; - match &result { - Ok(resp) => tracing::debug!( - tool = %tool, - successful = resp.successful, - has_error = resp.error.is_some(), - "[composio] execute_tool_once completed" - ), - Err(err) => tracing::debug!( - tool = %tool, - error = %err, - "[composio] execute_tool_once failed" - ), - } - result - } - /// `GET /agent-integrations/composio/github/repos` — list repositories /// available via the user's authorized GitHub connected account. pub async fn list_github_repos( diff --git a/src/openhuman/local_ai/schemas.rs b/src/openhuman/local_ai/schemas.rs index 1ca677a63..a99e4437a 100644 --- a/src/openhuman/local_ai/schemas.rs +++ b/src/openhuman/local_ai/schemas.rs @@ -1036,20 +1036,29 @@ fn handle_local_ai_install_whisper(params: Map) -> ControllerFutu let config = config_rpc::load_config_with_timeout().await?; let force = p.force.unwrap_or(false); - // Idempotency: a duplicate click while an install is already in - // flight should be a no-op, not a second concurrent download. - let current = crate::openhuman::local_ai::voice_install_common::read_status( + // Atomic install-start guard. A duplicate click while an install + // is already in flight (or a parallel auto-install firing + // alongside a manual click) must be a no-op — not a second + // concurrent download racing on the same `.part` file inside + // `download_to_file`. The previous read_status -> check -> + // write_status sequence was non-atomic and let two callers slip + // through; `try_acquire_install_slot` does the check-and-claim + // under a single mutex acquisition. + let slot = match crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot( crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER, - ); - if current.state - == crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing - { - tracing::debug!( - "[voice-install:whisper] already installing — returning current status" - ); - return serde_json::to_value(current) - .map_err(|e| format!("serialize whisper status: {e}")); - } + ) { + Some(slot) => slot, + None => { + tracing::debug!( + "[voice-install:whisper] slot already held — returning current status" + ); + let current = crate::openhuman::local_ai::voice_install_common::read_status( + crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER, + ); + return serde_json::to_value(current) + .map_err(|e| format!("serialize whisper status: {e}")); + } + }; // Mark "installing" before the spawn so the very next status poll // (≤ 2s away) reflects the new state without a stale read. @@ -1073,7 +1082,12 @@ fn handle_local_ai_install_whisper(params: Map) -> ControllerFutu "[voice-install:whisper] spawning background install" ); let model_size = p.model_size.clone(); + // Move the slot into the spawned task so it lives for the actual + // install duration (download + extract + validate), not just the + // RPC handler's lifetime. The slot's Drop releases the + // single-writer guard on task exit, including via panic. tokio::spawn(async move { + let _slot = slot; if let Err(e) = crate::openhuman::local_ai::install_whisper::install_whisper( &config, model_size, force, ) @@ -1096,16 +1110,23 @@ fn handle_local_ai_install_piper(params: Map) -> ControllerFuture let config = config_rpc::load_config_with_timeout().await?; let force = p.force.unwrap_or(false); - let current = crate::openhuman::local_ai::voice_install_common::read_status( + // See the whisper handler above for why this is an atomic slot + // acquisition rather than a read_status / write_status pair. + let slot = match crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot( crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER, - ); - if current.state - == crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing - { - tracing::debug!("[voice-install:piper] already installing — returning current status"); - return serde_json::to_value(current) - .map_err(|e| format!("serialize piper status: {e}")); - } + ) { + Some(slot) => slot, + None => { + tracing::debug!( + "[voice-install:piper] slot already held — returning current status" + ); + let current = crate::openhuman::local_ai::voice_install_common::read_status( + crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER, + ); + return serde_json::to_value(current) + .map_err(|e| format!("serialize piper status: {e}")); + } + }; crate::openhuman::local_ai::voice_install_common::write_status( crate::openhuman::local_ai::voice_install_common::VoiceInstallStatus { @@ -1126,7 +1147,10 @@ fn handle_local_ai_install_piper(params: Map) -> ControllerFuture "[voice-install:piper] spawning background install" ); let voice_id = p.voice_id.clone(); + // Move the slot into the spawned task — same rationale as the + // whisper handler. tokio::spawn(async move { + let _slot = slot; if let Err(e) = crate::openhuman::local_ai::install_piper::install_piper(&config, voice_id, force) .await diff --git a/src/openhuman/local_ai/schemas_tests.rs b/src/openhuman/local_ai/schemas_tests.rs index 957c18447..d898f1f56 100644 --- a/src/openhuman/local_ai/schemas_tests.rs +++ b/src/openhuman/local_ai/schemas_tests.rs @@ -269,3 +269,130 @@ async fn handle_set_ollama_path_accepts_empty_string_to_clear() { std::env::remove_var("OPENHUMAN_WORKSPACE"); } } + +/// Regression test for the CodeRabbit #7 race on PR #1755: when two +/// concurrent RPC calls (e.g. a double-click, or the auto-install firing +/// alongside a manual click) hit `handle_local_ai_install_whisper` at +/// the same time, only one of them must spawn a real install task. The +/// other must short-circuit and return the in-flight status without +/// starting a second download that would race on the same `.part` file. +/// +/// We exercise the actual handler — not just the slot primitive — so +/// the wiring at the call site is also covered. +#[tokio::test] +async fn install_whisper_handler_serializes_concurrent_calls() { + let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let tmp = TempDir::new().unwrap(); + unsafe { + std::env::set_var("OPENHUMAN_WORKSPACE", tmp.path()); + } + + // Pre-acquire the install slot from the test so we're guaranteed to + // observe the "already in flight" code path. Holding the slot here + // also means the handler under test will short-circuit immediately + // rather than spawning a real install task that would try to hit + // the network in CI. + let slot = crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot( + crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER, + ) + .expect("test should be able to claim the slot first"); + + // Mark the status table as `Installing` so the handler's + // short-circuit branch (which reads current status to return) sees + // a coherent snapshot. + crate::openhuman::local_ai::voice_install_common::write_status( + crate::openhuman::local_ai::voice_install_common::VoiceInstallStatus { + engine: crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER.to_string(), + state: crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing, + progress: Some(0), + downloaded_bytes: None, + total_bytes: None, + stage: Some("queued".to_string()), + error_detail: None, + }, + ); + + // Fire two handler calls in parallel. Both must succeed and both + // must return the existing `Installing` status — neither must + // mutate or re-spawn. This is exactly the double-click / auto-fire + // shape described in CodeRabbit #7. + let (r1, r2) = tokio::join!( + handle_local_ai_install_whisper(Map::new()), + handle_local_ai_install_whisper(Map::new()) + ); + + unsafe { + std::env::remove_var("OPENHUMAN_WORKSPACE"); + } + drop(slot); + // Clean up so other tests see Missing. + crate::openhuman::local_ai::voice_install_common::reset_status( + crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER, + ); + + let v1 = r1.expect("first call ok"); + let v2 = r2.expect("second call ok"); + // Both calls must report the engine is already installing — proving + // the handler short-circuited rather than running the spawn path. + for (label, v) in [("first", &v1), ("second", &v2)] { + let state = v.get("state").and_then(|s| s.as_str()); + assert_eq!( + state, + Some("installing"), + "{label} concurrent call should see Installing, got {v:?}" + ); + } +} + +/// Same regression for Piper. The two handlers share the slot +/// infrastructure but live in separate code paths, so the wiring needs +/// independent coverage. +#[tokio::test] +async fn install_piper_handler_serializes_concurrent_calls() { + let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let tmp = TempDir::new().unwrap(); + unsafe { + std::env::set_var("OPENHUMAN_WORKSPACE", tmp.path()); + } + + let slot = crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot( + crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER, + ) + .expect("test should be able to claim the slot first"); + + crate::openhuman::local_ai::voice_install_common::write_status( + crate::openhuman::local_ai::voice_install_common::VoiceInstallStatus { + engine: crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER.to_string(), + state: crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing, + progress: Some(0), + downloaded_bytes: None, + total_bytes: None, + stage: Some("queued".to_string()), + error_detail: None, + }, + ); + + let (r1, r2) = tokio::join!( + handle_local_ai_install_piper(Map::new()), + handle_local_ai_install_piper(Map::new()) + ); + + unsafe { + std::env::remove_var("OPENHUMAN_WORKSPACE"); + } + drop(slot); + crate::openhuman::local_ai::voice_install_common::reset_status( + crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER, + ); + + let v1 = r1.expect("first call ok"); + let v2 = r2.expect("second call ok"); + for (label, v) in [("first", &v1), ("second", &v2)] { + let state = v.get("state").and_then(|s| s.as_str()); + assert_eq!( + state, + Some("installing"), + "{label} concurrent call should see Installing, got {v:?}" + ); + } +} diff --git a/src/openhuman/local_ai/voice_install_common.rs b/src/openhuman/local_ai/voice_install_common.rs index 86dbb8588..ac0f7b2aa 100644 --- a/src/openhuman/local_ai/voice_install_common.rs +++ b/src/openhuman/local_ai/voice_install_common.rs @@ -20,9 +20,9 @@ //! shared harness here streams the body chunks itself and updates a //! singleton state map keyed by engine id. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; +use std::sync::{Mutex, OnceLock}; use std::time::{Duration, Instant}; use futures_util::StreamExt; @@ -163,6 +163,88 @@ pub(crate) fn reset_status(engine: &str) { table.remove(engine); } +/// Set of engines that currently have an install task in flight. Acts as a +/// true single-writer guard around the install-start critical section so +/// two concurrent RPC calls (a double-click, or the auto-install-on-change +/// firing in parallel with a manual button click) can't both pass an +/// `is-Installing?` snapshot check and then both spawn duplicate install +/// tasks that race on the same `.part` file inside `download_to_file`. +/// +/// `STATUS_TABLE` advertises lifecycle state to the polling RPC; this set +/// owns the *start* decision. They are deliberately separate locks: the +/// status table is read on every status poll (cheap, frequent), while +/// this set is only touched on install start / end (rare). +static IN_FLIGHT: OnceLock>> = OnceLock::new(); + +fn in_flight() -> &'static Mutex> { + IN_FLIGHT.get_or_init(|| Mutex::new(HashSet::new())) +} + +/// RAII guard returned by [`try_acquire_install_slot`]. Holding one of +/// these proves the caller has exclusive ownership of the install-start +/// slot for `engine`. Dropping it (including via panic unwind) releases +/// the slot so a subsequent install attempt can proceed. +/// +/// The handler is expected to **move** the slot into the spawned tokio +/// task so the slot lives for the install's actual duration, not just +/// the RPC handler's lifetime. Releasing the slot when the handler +/// returns (instead of when the install finishes) would re-open the +/// race the slot was added to close. +pub(crate) struct InstallSlot { + engine: &'static str, +} + +impl Drop for InstallSlot { + fn drop(&mut self) { + match in_flight().lock() { + Ok(mut guard) => { + let removed = guard.remove(self.engine); + log::debug!( + "[voice-install] install slot released engine={} was_present={}", + self.engine, + removed + ); + } + Err(_) => { + // Lock poisoned — another thread panicked while holding + // it. The set is in an unknown state; the best we can do + // is log and let the process continue. Subsequent + // acquire attempts will also hit the poisoned lock and + // surface the failure to the user. + log::error!( + "[voice-install] install slot lock poisoned on drop for engine={}", + self.engine + ); + } + } + } +} + +/// Atomically try to claim the install-start slot for `engine`. Returns +/// `Some(InstallSlot)` if no install is currently in flight for this +/// engine, or `None` if one is already running. +/// +/// This replaces the previous non-atomic `read_status` -> check -> +/// `write_status` sequence in the install RPC handlers. The +/// check-and-insert happens under a single mutex acquisition, so two +/// concurrent callers cannot both observe "not installing" and both +/// spawn tasks. +pub(crate) fn try_acquire_install_slot(engine: &'static str) -> Option { + let mut guard = in_flight() + .lock() + .expect("voice install in-flight lock poisoned"); + if guard.contains(engine) { + log::debug!( + "[voice-install] install slot denied engine={} (already in flight)", + engine + ); + return None; + } + guard.insert(engine); + log::debug!("[voice-install] install slot acquired engine={}", engine); + Some(InstallSlot { engine }) +} + /// Download `url` to `dest` with atomic rename. Streams bytes through /// SHA256 if `expected_sha256` is provided, otherwise validates that the /// final size is at least `min_bytes`. @@ -406,6 +488,123 @@ mod tests { assert_eq!(read_status(&engine).state, VoiceInstallState::Missing); } + // Engine ids used by the slot tests below. The slot map is keyed by + // `&'static str`, so we can't use uuid-suffixed names like the + // status-table tests; we use these dedicated keys instead. Production + // engine ids (ENGINE_WHISPER / ENGINE_PIPER) are deliberately avoided + // so tests can't deadlock against a real install in another test. + const TEST_SLOT_ENGINE_A: &str = "__test_slot_engine_a__"; + const TEST_SLOT_ENGINE_B: &str = "__test_slot_engine_b__"; + + /// Best-effort drain of a test slot so the global set is clean across + /// runs. Tests that leave a slot held (e.g. by forgetting it) would + /// pollute subsequent runs in the same `cargo test` invocation. + fn drain_test_slot(engine: &'static str) { + if let Ok(mut g) = in_flight().lock() { + g.remove(engine); + } + } + + #[test] + fn try_acquire_install_slot_grants_then_blocks_then_releases() { + drain_test_slot(TEST_SLOT_ENGINE_A); + + // First caller gets the slot. + let slot = try_acquire_install_slot(TEST_SLOT_ENGINE_A); + assert!(slot.is_some(), "first acquire should succeed"); + + // Concurrent caller is rejected while the first slot lives. + let second = try_acquire_install_slot(TEST_SLOT_ENGINE_A); + assert!( + second.is_none(), + "second acquire must be rejected while slot is held" + ); + + // Releasing the first slot reopens the door for a fresh caller. + drop(slot); + let third = try_acquire_install_slot(TEST_SLOT_ENGINE_A); + assert!( + third.is_some(), + "acquire after drop should succeed (Drop must release)" + ); + + drop(third); + drain_test_slot(TEST_SLOT_ENGINE_A); + } + + #[test] + fn install_slot_keys_are_independent_per_engine() { + drain_test_slot(TEST_SLOT_ENGINE_A); + drain_test_slot(TEST_SLOT_ENGINE_B); + + let slot_a = try_acquire_install_slot(TEST_SLOT_ENGINE_A).expect("A acquire"); + // Holding the A slot must not block the B slot — installs for + // whisper and piper run independently. + let slot_b = try_acquire_install_slot(TEST_SLOT_ENGINE_B) + .expect("B acquire must succeed independently"); + // Acquiring A again must still fail though. + assert!( + try_acquire_install_slot(TEST_SLOT_ENGINE_A).is_none(), + "A slot is still held" + ); + + drop(slot_a); + drop(slot_b); + drain_test_slot(TEST_SLOT_ENGINE_A); + drain_test_slot(TEST_SLOT_ENGINE_B); + } + + /// Race-path test — the whole reason the slot exists. Spawn many + /// concurrent tasks that all try to acquire the slot for the same + /// engine; exactly one must succeed, all others must be rejected. + /// This is the unit-level analogue of "two RPC handlers fire at the + /// same time and both spawn install tasks" — the bug CodeRabbit + /// flagged on PR #1755. + #[tokio::test] + async fn concurrent_acquire_grants_exactly_one_slot() { + drain_test_slot(TEST_SLOT_ENGINE_A); + + // 32 concurrent acquirers — high enough to make a non-atomic + // implementation almost certainly fail, low enough to stay + // hermetic and fast. + const N: usize = 32; + let mut handles = Vec::with_capacity(N); + for _ in 0..N { + handles.push(tokio::spawn(async move { + try_acquire_install_slot(TEST_SLOT_ENGINE_A) + })); + } + let mut winners = 0usize; + let mut losers = 0usize; + // Collect outcomes *before* any slot is dropped — winners must + // hold their slot alive past every other acquirer's attempt. + let mut held = Vec::new(); + for h in handles { + match h.await.expect("task panicked") { + Some(slot) => { + winners += 1; + held.push(slot); + } + None => losers += 1, + } + } + assert_eq!( + winners, 1, + "exactly one concurrent acquirer must win (got {winners})" + ); + assert_eq!(losers, N - 1, "all other acquirers must lose"); + + // Now drop the winner — the slot becomes available again. + held.clear(); + let after = try_acquire_install_slot(TEST_SLOT_ENGINE_A); + assert!( + after.is_some(), + "slot must be reacquirable once the winner drops" + ); + drop(after); + drain_test_slot(TEST_SLOT_ENGINE_A); + } + #[tokio::test] async fn download_to_file_rejects_oversize_min_bytes() { // 4xx-like guard: a non-existent host fails before we can write