diff --git a/crates/cgka-engine/src/openmls_projection.rs b/crates/cgka-engine/src/openmls_projection.rs index e7237af2..93f81b39 100644 --- a/crates/cgka-engine/src/openmls_projection.rs +++ b/crates/cgka-engine/src/openmls_projection.rs @@ -132,14 +132,71 @@ struct CandidatePathProbe { messages: Vec, digests: Vec<[u8; 32]>, tip_epoch: u64, + /// The materialized candidate from the probe that produced this node (folding only + /// pending proposals — not applications). `None` only for the empty seed node, which is + /// never completed. Retained so a completed path can be reused by the canonicalization + /// core without a second full replay (#635), on the app-free path where it is + /// byte-identical to a fresh materialize. + materialized: Option, } #[derive(Clone, Debug)] struct StoredOpenMlsCandidatePathResult { candidate_paths: Vec, + /// Terminal materialized candidates for `candidate_paths`, in the same order. Reusable by + /// the canonicalization core only when the pass has no pending application messages (the + /// BFS folds proposals but not applications), otherwise the core re-materializes. Empty + /// (or shorter than `candidate_paths`) means "do not reuse". + materialized: Vec, invalid_commit_drops: Vec, } +/// Bounds the total OpenMLS replay round-trips a single convergence pass may perform, so +/// attacker-driven same-epoch commit branching cannot amplify into unbounded CPU/IO (#635). +/// The pass fails closed (`ReplayBudgetExceeded`) rather than returning a partial result. +struct ReplayBudget { + remaining: u64, +} + +/// Multiplicative slack over the linear `commits × (max_rewind_commits + 1)` probe estimate. +/// Generous enough that legitimate forks never trip the budget; small enough that pathological +/// `B^D` branching fails closed well before it materializes. +const CANDIDATE_REPLAY_BUDGET_SLACK: u64 = 4; +/// Floor so tiny passes (few commits, shallow rewind) always have headroom. +const CANDIDATE_REPLAY_BUDGET_FLOOR: u64 = 32; + +impl ReplayBudget { + fn new(limit: u64) -> Self { + Self { remaining: limit } + } + + /// Unlimited budget for callers outside the bounded convergence BFS (e.g. the public + /// `materialize_openmls_candidate_paths`, used directly by conformance vectors). + fn unlimited() -> Self { + Self { + remaining: u64::MAX, + } + } + + /// Derive the per-pass replay ceiling from the number of competing commits and the rewind + /// horizon. Saturating throughout so a hostile large input cannot overflow into a small cap. + fn for_pass(commit_count: usize, max_rewind_commits: u64) -> Self { + let limit = (commit_count as u64) + .saturating_mul(max_rewind_commits.saturating_add(1)) + .saturating_mul(CANDIDATE_REPLAY_BUDGET_SLACK) + .saturating_add(CANDIDATE_REPLAY_BUDGET_FLOOR); + Self::new(limit) + } + + fn consume(&mut self) -> Result<(), OpenMlsProjectionError> { + if self.remaining == 0 { + return Err(OpenMlsProjectionError::ReplayBudgetExceeded); + } + self.remaining -= 1; + Ok(()) + } +} + #[derive(Clone, Debug)] enum CandidatePathProbeResult { Materialized(Option), @@ -207,6 +264,7 @@ pub enum OpenMlsProjectionError { Serialize(String), Storage(String), InvalidPolicy(String), + ReplayBudgetExceeded, } impl std::fmt::Display for OpenMlsProjectionError { @@ -236,6 +294,9 @@ impl std::fmt::Display for OpenMlsProjectionError { OpenMlsProjectionError::InvalidPolicy(e) => { write!(f, "invalid convergence policy: {e}") } + OpenMlsProjectionError::ReplayBudgetExceeded => { + write!(f, "convergence replay budget exceeded") + } } } } @@ -294,6 +355,20 @@ pub fn materialize_openmls_candidate_paths( storage: &S, group_id: &GroupId, paths: &[OpenMlsCandidatePath], +) -> Result, OpenMlsProjectionError> { + materialize_openmls_candidate_paths_budgeted( + storage, + group_id, + paths, + &mut ReplayBudget::unlimited(), + ) +} + +fn materialize_openmls_candidate_paths_budgeted( + storage: &S, + group_id: &GroupId, + paths: &[OpenMlsCandidatePath], + budget: &mut ReplayBudget, ) -> Result, OpenMlsProjectionError> { let mut candidates = Vec::with_capacity(paths.len()); for path in paths { @@ -302,6 +377,7 @@ pub fn materialize_openmls_candidate_paths( path.branch_id.clone(), )); } + budget.consume()?; let observations = replay_openmls_messages(storage, group_id, &path.messages)?; let mut fork_epoch: Option = None; let mut tip_epoch: Option = None; @@ -374,6 +450,20 @@ pub fn canonicalize_openmls_batch( &batch.pending_messages, )?; let materialized = materialize_openmls_candidate_paths(storage, group_id, &candidate_paths)?; + canonicalize_openmls_batch_with_materialized(group_id, batch, materialized) +} + +/// Canonicalization core over already-materialized candidates. Split out of +/// [`canonicalize_openmls_batch`] so the stored-message hot path can reuse the candidates the +/// BFS already materialized instead of replaying every completed path a second time (#635). +/// `materialized` MUST equal what `materialize_openmls_candidate_paths` would produce for +/// `batch.candidate_paths` folded with `batch.pending_messages` (same order, sorted by +/// `branch_id`); the caller guarantees this only on the app-free reuse path. +fn canonicalize_openmls_batch_with_materialized( + group_id: &GroupId, + batch: OpenMlsCanonicalizationBatch, + materialized: Vec, +) -> Result { let proposal_id_by_ref = proposal_id_by_ref(&materialized); let materialized_candidates: Vec<_> = materialized .iter() @@ -576,20 +666,33 @@ fn canonicalize_stored_openmls_messages_from_current( work.commit_messages, &work.pending_messages, work.replay_start_epoch, + work.policy.convergence.max_rewind_commits, )?; - let mut result = canonicalize_openmls_batch( - storage, - group_id, - OpenMlsCanonicalizationBatch { - state: work.state, - candidate_paths: path_result.candidate_paths, - pending_messages: work.pending_messages, - outbound_intents: work.outbound_intents, - policy: work.policy, - now_ms: work.now_ms, - }, - )?; + // Reuse the candidates the BFS already materialized instead of replaying every completed + // path a second time — but ONLY when the pass has no pending application messages. The BFS + // probes fold pending proposals but not applications, so with pending apps the reused + // candidate would be missing the ApplicationProcessed observations the canonicalization + // core consumes; that case falls back to a fresh full materialize (#635). + let has_pending_apps = pending_messages_contain_application(&work.pending_messages)?; + let can_reuse_materialized = + !has_pending_apps && path_result.materialized.len() == path_result.candidate_paths.len(); + + let batch = OpenMlsCanonicalizationBatch { + state: work.state, + candidate_paths: path_result.candidate_paths, + pending_messages: work.pending_messages, + outbound_intents: work.outbound_intents, + policy: work.policy, + now_ms: work.now_ms, + }; + let mut result = if can_reuse_materialized { + let mut materialized = path_result.materialized; + materialized.sort_by(|a, b| a.branch_id.cmp(&b.branch_id)); + canonicalize_openmls_batch_with_materialized(group_id, batch, materialized)? + } else { + canonicalize_openmls_batch(storage, group_id, batch)? + }; append_dropped_messages(&mut result, path_result.invalid_commit_drops); Ok(result) } @@ -646,6 +749,7 @@ fn build_stored_openmls_candidate_paths( mut commits: Vec, pending_messages: &[TransportMessage], starting_epoch: u64, + max_rewind_commits: u64, ) -> Result { commits.sort_by(|a, b| { a.source_epoch @@ -654,11 +758,13 @@ fn build_stored_openmls_candidate_paths( .then_with(|| a.message.payload.cmp(&b.message.payload)) }); + let mut budget = ReplayBudget::for_pass(commits.len(), max_rewind_commits); let pending_proposals = pending_proposal_messages(pending_messages)?; let mut frontier = vec![CandidatePathProbe { messages: Vec::new(), digests: Vec::new(), tip_epoch: starting_epoch, + materialized: None, }]; let mut completed = Vec::new(); let mut invalid_commit_drops = Vec::new(); @@ -688,6 +794,7 @@ fn build_stored_openmls_candidate_paths( messages.clone(), &digests, &pending_proposals, + &mut budget, )? { CandidatePathProbeResult::Materialized(Some(candidate)) => candidate, CandidatePathProbeResult::Materialized(None) => continue, @@ -710,10 +817,12 @@ fn build_stored_openmls_candidate_paths( }; extended = true; + let tip_epoch = candidate.tip_epoch; next_frontier.push(CandidatePathProbe { messages, digests, - tip_epoch: candidate.tip_epoch, + tip_epoch, + materialized: Some(candidate), }); } @@ -725,18 +834,41 @@ fn build_stored_openmls_candidate_paths( frontier = next_frontier; } + let mut candidate_paths = Vec::with_capacity(completed.len()); + let mut materialized = Vec::with_capacity(completed.len()); + for path in completed { + candidate_paths.push(OpenMlsCandidatePath { + branch_id: branch_id_for_path_digests(&path.digests), + messages: path.messages, + }); + // Every completed path is a non-empty node created via a probe, so `materialized` is + // always `Some`. Push in lockstep with `candidate_paths`; if the invariant is ever + // broken the length mismatch makes the caller fall back to a fresh materialize. + if let Some(candidate) = path.materialized { + materialized.push(candidate); + } + } + Ok(StoredOpenMlsCandidatePathResult { - candidate_paths: completed - .into_iter() - .map(|path| OpenMlsCandidatePath { - branch_id: branch_id_for_path_digests(&path.digests), - messages: path.messages, - }) - .collect(), + candidate_paths, + materialized, invalid_commit_drops, }) } +/// True if any pending message is an application message (used to decide whether the BFS's +/// proposal-only materialized candidates can be reused by the canonicalization core — #635). +fn pending_messages_contain_application( + pending_messages: &[TransportMessage], +) -> Result { + for message in pending_messages { + if project_mls_message(&message.payload)?.kind == OpenMlsContentKind::Application { + return Ok(true); + } + } + Ok(false) +} + fn pending_proposal_messages( pending_messages: &[TransportMessage], ) -> Result, OpenMlsProjectionError> { @@ -755,13 +887,14 @@ fn probe_candidate_path( messages: Vec, digests: &[[u8; 32]], pending_proposals: &[TransportMessage], + budget: &mut ReplayBudget, ) -> Result { let path = OpenMlsCandidatePath { branch_id: branch_id_for_path_digests(digests), messages, }; let replay_paths = candidate_paths_with_pending_replay_messages(&[path], pending_proposals)?; - match materialize_openmls_candidate_paths(storage, group_id, &replay_paths) { + match materialize_openmls_candidate_paths_budgeted(storage, group_id, &replay_paths, budget) { Ok(mut candidates) => Ok(CandidatePathProbeResult::Materialized(candidates.pop())), Err(OpenMlsProjectionError::UnauthorizedCommit { message_id }) => { Ok(CandidatePathProbeResult::UnauthorizedCommit { message_id }) @@ -1713,3 +1846,60 @@ fn tls_hex(value: &T) -> Result .map(hex::encode) .map_err(|e| OpenMlsProjectionError::Serialize(format!("{e:?}"))) } + +#[cfg(test)] +mod replay_budget_tests { + use super::{ + CANDIDATE_REPLAY_BUDGET_FLOOR, CANDIDATE_REPLAY_BUDGET_SLACK, OpenMlsProjectionError, + ReplayBudget, + }; + + #[test] + fn for_pass_scales_with_commits_and_rewind_and_keeps_a_floor() { + // Zero commits still gets the floor so tiny passes always have headroom. + let mut empty = ReplayBudget::for_pass(0, 5); + assert_eq!(empty.remaining, CANDIDATE_REPLAY_BUDGET_FLOOR); + for _ in 0..CANDIDATE_REPLAY_BUDGET_FLOOR { + empty.consume().expect("floor budget covers the floor"); + } + assert!(matches!( + empty.consume(), + Err(OpenMlsProjectionError::ReplayBudgetExceeded) + )); + + // commits × (max_rewind + 1) × slack + floor. + let budget = ReplayBudget::for_pass(3, 5); + assert_eq!( + budget.remaining, + 3 * (5 + 1) * CANDIDATE_REPLAY_BUDGET_SLACK + CANDIDATE_REPLAY_BUDGET_FLOOR + ); + } + + #[test] + fn for_pass_saturates_instead_of_overflowing() { + // A hostile huge input must not wrap to a small cap. + let budget = ReplayBudget::for_pass(usize::MAX, u64::MAX); + assert_eq!(budget.remaining, u64::MAX); + } + + #[test] + fn consume_fails_closed_when_exhausted() { + let mut budget = ReplayBudget::new(2); + budget.consume().expect("first replay within budget"); + budget.consume().expect("second replay within budget"); + assert!(matches!( + budget.consume(), + Err(OpenMlsProjectionError::ReplayBudgetExceeded) + )); + } + + #[test] + fn unlimited_budget_never_trips() { + let mut budget = ReplayBudget::unlimited(); + for _ in 0..10_000 { + budget + .consume() + .expect("unlimited budget never fails closed"); + } + } +} diff --git a/crates/cgka-engine/tests/distributed_convergence.rs b/crates/cgka-engine/tests/distributed_convergence.rs index e82e9260..dda9f452 100644 --- a/crates/cgka-engine/tests/distributed_convergence.rs +++ b/crates/cgka-engine/tests/distributed_convergence.rs @@ -1151,6 +1151,90 @@ async fn engine_materializes_multi_commit_path_from_stored_commits() { ); } +/// Reuse-path sibling of `engine_materializes_multi_commit_path_from_stored_commits`: the same +/// multi-commit chain but with NO pending application message buffered, so canonicalization takes +/// the #635 reuse branch (BFS-materialized candidates are reused instead of re-materialized). The +/// canonical commits and resulting epoch must match the fresh path exactly. +#[tokio::test] +async fn engine_reuses_bfs_materialized_candidates_when_no_pending_app_messages() { + let (mut alice, _alice_storage) = build_client(b"alice"); + let (mut carol, carol_storage) = build_client(b"carol"); + let (mut david, _david_storage) = build_client(b"david"); + let (mut eve, _eve_storage) = build_client(b"eve"); + + let carol_kp = carol.fresh_key_package().await.unwrap(); + let (group_id, create) = alice + .create_group(CreateGroupRequest { + name: "engine-convergence-chain-app-free".into(), + description: "".into(), + members: vec![carol_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let (pending, welcomes) = match create { + SendResult::GroupCreated { pending, welcomes } => (pending, welcomes), + other => panic!("expected GroupCreated, got {other:?}"), + }; + alice.confirm_published(pending).await.unwrap(); + carol + .join_welcome(welcome_for(&welcomes, b"carol")) + .await + .unwrap(); + + let david_kp = david.fresh_key_package().await.unwrap(); + let invite_david = alice + .send(SendIntent::Invite { + group_id: group_id.clone(), + key_packages: vec![david_kp], + }) + .await + .unwrap(); + let (commit_david, pending_david) = evolution(invite_david); + alice.confirm_published(pending_david).await.unwrap(); + + let eve_kp = eve.fresh_key_package().await.unwrap(); + let invite_eve = alice + .send(SendIntent::Invite { + group_id: group_id.clone(), + key_packages: vec![eve_kp], + }) + .await + .unwrap(); + let (commit_eve, pending_eve) = evolution(invite_eve); + alice.confirm_published(pending_eve).await.unwrap(); + + let commit_eve = route(commit_eve, &group_id); + let commit_david = route(commit_david, &group_id); + // Buffer the child before the parent, and crucially NO app message — this keeps the + // canonicalization pass free of pending application messages so the reuse branch fires. + carol + .buffer_openmls_convergence_message(&group_id, commit_eve.clone(), 1_000) + .expect("child commit buffered first"); + carol + .buffer_openmls_convergence_message(&group_id, commit_david.clone(), 1_000) + .expect("parent commit buffered second"); + + let result = carol + .converge_stored_openmls_messages(&group_id, 1_000_000) + .expect("stored parent and child commits converge as one reused path"); + + assert_eq!(result.convergence_status, ConvergenceStatus::Settled); + assert_eq!(carol.epoch(&group_id).unwrap(), EpochId(3)); + assert_eq!( + result.accepted_commits, + vec![content_hex(&commit_david), content_hex(&commit_eve)] + ); + assert!(result.accepted_app_messages.is_empty()); + assert_message_state(&carol_storage, &commit_david, MessageState::Processed); + assert_message_state(&carol_storage, &commit_eve, MessageState::Processed); + let members = carol.members(&group_id).unwrap(); + assert!(members.iter().any(|member| member.id == david.self_id())); + assert!(members.iter().any(|member| member.id == eve.self_id())); +} + #[tokio::test] async fn engine_keeps_child_commit_pending_until_parent_arrives() { let (mut alice, _alice_storage) = build_client(b"alice"); diff --git a/crates/cli/src/daemon/mod.rs b/crates/cli/src/daemon/mod.rs index 831312b8..ea3dfe8a 100644 --- a/crates/cli/src/daemon/mod.rs +++ b/crates/cli/src/daemon/mod.rs @@ -325,28 +325,19 @@ async fn handle_daemon_connection( let _ = handle_group_state_subscription(&mut stream, &defaults, runtime, *cli).await; } DaemonRequest::StreamWatch { cli } => { - let mut workers_guard = workers.lock().await; let _ = handle_stream_watch_connection( cli, &mut stream, &defaults, state, events, - &mut workers_guard, + &workers, ) .await; } DaemonRequest::Execute { cli } => { - let mut workers_guard = workers.lock().await; - let _ = handle_execute_connection( - cli, - &mut stream, - &defaults, - state, - events, - &mut workers_guard, - ) - .await; + let _ = handle_execute_connection(cli, &mut stream, &defaults, state, events, &workers) + .await; } } } @@ -377,23 +368,21 @@ async fn handle_stream_watch_connection( defaults: &DaemonDefaults, state: Arc>, events: DaemonEventHub, - workers: &mut DaemonWorkers, + workers: &SharedDaemonWorkers, ) -> Result<(), Box> { apply_defaults(&mut cli, defaults); - reconcile_app_runtime( - defaults, - state.clone(), - events.clone(), - &mut workers.runtime, - ) - .await; - let output = start_stream_watch( - *cli, - defaults, - workers.runtime.runtime.as_ref(), - &workers.runtime.stream_watch, - ) - .await; + // Hold the lock only for the host-mutating reconcile; clone the runtime handle and the + // (interior-mutable) stream-watch registry, then spawn the watch + open the broker + // connection off the lock (#633). + let (runtime, stream_watch) = { + let mut guard = workers.lock().await; + reconcile_app_runtime(defaults, state.clone(), events.clone(), &mut guard.runtime).await; + ( + guard.runtime.runtime.clone(), + guard.runtime.stream_watch.clone(), + ) + }; + let output = start_stream_watch(*cli, defaults, runtime.as_ref(), &stream_watch).await; write_daemon_output(stream, &output).await; Ok(()) @@ -405,25 +394,44 @@ async fn handle_execute_connection( defaults: &DaemonDefaults, state: Arc>, events: DaemonEventHub, - workers: &mut DaemonWorkers, + workers: &SharedDaemonWorkers, ) -> Result<(), Box> { apply_defaults(&mut cli, defaults); if let Some(output) = blocked_daemon_execute_output(cli.as_ref()) { write_daemon_output(stream, &output).await; return Ok(()); } - if let Some(output) = handle_stream_compose_request( - &cli, - defaults, - state.clone(), - events.clone(), - &mut workers.runtime, - &mut workers.stream_compose, - ) - .await - { - write_daemon_output(stream, &output).await; - return Ok(()); + // Stream-compose commands mutate the compose session map and the runtime host together, so + // that low-traffic QUIC-preview path keeps the lock for its (short) duration. Gate on the + // compose subcommands specifically: non-compose stream commands (start/finish/watch/send) are + // hosted-runtime commands and must not take the workers lock here, or they would block behind + // an unrelated busy workers mutex before falling through to the off-lock hosted path. + if matches!( + cli.command, + crate::Command::Stream { + command: crate::StreamCommand::ComposeOpen { .. } + | crate::StreamCommand::ComposeAppend { .. } + | crate::StreamCommand::ComposeFinish { .. } + | crate::StreamCommand::ComposeCancel { .. }, + } + ) { + let compose_output = { + let mut guard = workers.lock().await; + let guard = &mut *guard; + handle_stream_compose_request( + &cli, + defaults, + state.clone(), + events.clone(), + &mut guard.runtime, + &mut guard.stream_compose, + ) + .await + }; + if let Some(output) = compose_output { + write_daemon_output(stream, &output).await; + return Ok(()); + } } let refresh = app_runtime_refresh_after_execute(&cli); if let Some(output) = handle_app_runtime_account_setup_request( @@ -431,35 +439,25 @@ async fn handle_execute_connection( defaults, state.clone(), events.clone(), - &mut workers.runtime, + workers, ) .await { write_daemon_output(stream, &output).await; return Ok(()); } - if let Some(output) = handle_app_runtime_command_request( - &cli, - defaults, - state.clone(), - events.clone(), - &mut workers.runtime, - ) - .await + if let Some(output) = + handle_app_runtime_command_request(&cli, defaults, state.clone(), events.clone(), workers) + .await { write_daemon_output(stream, &output).await; return Ok(()); } + // run_cli_local opens its own account/session and touches no shared daemon state, so it runs + // entirely off the workers lock — the core head-of-line fix (#633). let output = crate::run_cli_local(*cli).await; if output.code == 0 { - refresh_app_runtime( - defaults, - state.clone(), - events.clone(), - &mut workers.runtime, - refresh, - ) - .await; + refresh_app_runtime(defaults, state.clone(), events.clone(), workers, refresh).await; } write_daemon_output(stream, &output).await; diff --git a/crates/cli/src/daemon/runtime_host.rs b/crates/cli/src/daemon/runtime_host.rs index b3c6d499..9e249876 100644 --- a/crates/cli/src/daemon/runtime_host.rs +++ b/crates/cli/src/daemon/runtime_host.rs @@ -54,12 +54,28 @@ pub(crate) fn app_runtime_enabled(defaults: &DaemonDefaults) -> bool { defaults.relay.is_some() } +/// Reconcile the app runtime under the workers lock and return a cloned runtime handle so the +/// caller can perform relay I/O WITHOUT holding the lock. The lock is held only for the +/// host-mutating reconcile (runtime create / bridge spawn / account reconcile), matching the +/// subscription handlers and fixing the head-of-line blocking in #633. Returns `None` when no +/// runtime could be brought up (missing relay / open error). +pub(crate) async fn reconcile_and_clone_runtime( + defaults: &DaemonDefaults, + state: Arc>, + events: DaemonEventHub, + workers: &SharedDaemonWorkers, +) -> Option { + let mut guard = workers.lock().await; + reconcile_app_runtime(defaults, state, events, &mut guard.runtime).await; + guard.runtime.runtime.clone() +} + pub(crate) async fn handle_app_runtime_account_setup_request( cli: &Cli, defaults: &DaemonDefaults, state: Arc>, events: DaemonEventHub, - host: &mut AppRuntimeHost, + workers: &SharedDaemonWorkers, ) -> Option { let request = match app_runtime_account_setup_request(cli) { Ok(Some(request)) => request, @@ -69,13 +85,14 @@ pub(crate) async fn handle_app_runtime_account_setup_request( if !app_runtime_enabled(defaults) { return None; } - reconcile_app_runtime(defaults, state.clone(), events, host).await; - let Some(runtime) = &host.runtime else { + let Some(runtime) = reconcile_and_clone_runtime(defaults, state, events, workers).await else { return Some(crate::command_output_result( cli.json, Err(crate::DmError::MissingRelay), )); }; + // create_or_import_account drives relay I/O through the cloned runtime handle (internally + // synchronized), so it runs off the workers lock. let output = runtime .create_or_import_account(request) .await @@ -84,7 +101,31 @@ pub(crate) async fn handle_app_runtime_account_setup_request( Some(crate::command_output_result(cli.json, output)) } +/// Execute-path entry: reconcile under the lock, then dispatch the hosted command off the lock +/// against a cloned runtime handle (#633). This is the common `dm group|message|chats|…` path. pub(crate) async fn handle_app_runtime_command_request( + cli: &Cli, + defaults: &DaemonDefaults, + state: Arc>, + events: DaemonEventHub, + workers: &SharedDaemonWorkers, +) -> Option { + if !app_runtime_enabled(defaults) || !is_hosted_runtime_command(cli) { + return None; + } + let Some(runtime) = reconcile_and_clone_runtime(defaults, state, events, workers).await else { + return Some(crate::command_output_result( + cli.json, + Err(crate::DmError::MissingRelay), + )); + }; + dispatch_hosted_runtime_command(cli, defaults, &runtime).await +} + +/// Host-based entry used by the stream-compose path (`run_hosted_stream_marker_cli_json`), which +/// already holds the workers lock and owns a `&mut AppRuntimeHost`. Reconciles in place and +/// dispatches against the host's runtime — no re-entrant lock (which would deadlock). +pub(crate) async fn handle_hosted_runtime_command_with_host( cli: &Cli, defaults: &DaemonDefaults, state: Arc>, @@ -94,14 +135,24 @@ pub(crate) async fn handle_app_runtime_command_request( if !app_runtime_enabled(defaults) || !is_hosted_runtime_command(cli) { return None; } - reconcile_app_runtime(defaults, state.clone(), events, host).await; + reconcile_app_runtime(defaults, state, events, host).await; let Some(runtime) = &host.runtime else { return Some(crate::command_output_result( cli.json, Err(crate::DmError::MissingRelay), )); }; + dispatch_hosted_runtime_command(cli, defaults, runtime).await +} +/// Dispatch a hosted runtime command against an already-resolved runtime handle. Performs the +/// relay-backed command work and touches no shared daemon state, so callers run it off the +/// workers lock. +async fn dispatch_hosted_runtime_command( + cli: &Cli, + defaults: &DaemonDefaults, + runtime: &marmot_app::MarmotAppRuntime, +) -> Option { let secret_store = match crate::resolve_secret_store(defaults.secret_store) { Ok(secret_store) => secret_store, Err(err) => return Some(crate::command_output_result(cli.json, Err(err))), @@ -352,7 +403,7 @@ pub(crate) async fn refresh_app_runtime( defaults: &DaemonDefaults, state: Arc>, events: DaemonEventHub, - host: &mut AppRuntimeHost, + workers: &SharedDaemonWorkers, refresh: AppRuntimeRefresh, ) { if !app_runtime_enabled(defaults) { @@ -361,26 +412,45 @@ pub(crate) async fn refresh_app_runtime( match refresh { AppRuntimeRefresh::None => {} AppRuntimeRefresh::Reconcile => { - reconcile_app_runtime(defaults, state, events, host).await; + let _ = reconcile_and_clone_runtime(defaults, state, events, workers).await; } AppRuntimeRefresh::RestartSelected(selector) => { - if host.runtime.is_none() { - reconcile_app_runtime(defaults, state, events, host).await; + // Bring the runtime up under the lock if it is missing (matching the original + // "reconcile + return, no restart" for a cold host); otherwise clone the handle and + // restart off the lock. + let runtime = { + let mut guard = workers.lock().await; + if guard.runtime.runtime.is_none() { + reconcile_app_runtime( + defaults, + state.clone(), + events.clone(), + &mut guard.runtime, + ) + .await; + None + } else { + guard.runtime.runtime.clone() + } + }; + let Some(runtime) = runtime else { return; - } + }; if let Some(account_id) = resolve_app_runtime_account_id(defaults, selector).await { - if let Some(runtime) = &host.runtime - && let Err(err) = runtime.restart_account(&account_id).await - { + if let Err(err) = runtime.restart_account(&account_id).await { record_runtime_activity_error(&state, err.to_string()); } } else { - reconcile_app_runtime(defaults, state, events, host).await; + // Account not resolvable → reconcile as the original fallback did. + let mut guard = workers.lock().await; + reconcile_app_runtime(defaults, state.clone(), events.clone(), &mut guard.runtime) + .await; } } AppRuntimeRefresh::CatchUpAll => { - reconcile_app_runtime(defaults, state.clone(), events, host).await; - if let Some(runtime) = &host.runtime + let runtime = + reconcile_and_clone_runtime(defaults, state.clone(), events, workers).await; + if let Some(runtime) = runtime && let Err(err) = runtime.catch_up_accounts().await { record_runtime_activity_error(&state, err.to_string()); diff --git a/crates/cli/src/daemon/stream_workers.rs b/crates/cli/src/daemon/stream_workers.rs index 3139c270..eb07d46e 100644 --- a/crates/cli/src/daemon/stream_workers.rs +++ b/crates/cli/src/daemon/stream_workers.rs @@ -703,7 +703,7 @@ pub(crate) async fn run_hosted_stream_marker_cli_json( runtime_host: &mut AppRuntimeHost, ) -> Result { let Some(output) = - handle_app_runtime_command_request(cli, defaults, state, events, runtime_host).await + handle_hosted_runtime_command_with_host(cli, defaults, state, events, runtime_host).await else { return Err("stream marker command did not use the daemon runtime".to_owned()); }; diff --git a/crates/cli/src/daemon/tests.rs b/crates/cli/src/daemon/tests.rs index 16aeccfd..ac51808d 100644 --- a/crates/cli/src/daemon/tests.rs +++ b/crates/cli/src/daemon/tests.rs @@ -612,6 +612,50 @@ async fn daemon_status_response_does_not_wait_for_busy_workers() { assert_eq!(status.pid, Some(42)); } +#[tokio::test] +async fn daemon_execute_local_command_runs_without_holding_worker_lock() { + // A relay-less local command takes the `run_cli_local` path, which opens its own + // account/session and touches no shared daemon state. It must run WITHOUT acquiring the + // workers lock, so a concurrent lock holder (e.g. another command mid-reconcile against a + // slow relay) cannot head-of-line block it (#633). We hold the workers lock for the entire + // call: if the handler tried to lock it, this would deadlock and hit the timeout. + let defaults = DaemonDefaults { + home: PathBuf::from("/tmp/dm-daemon-home-hol"), + socket: PathBuf::from("/tmp/dm-daemon-hol.sock"), + pid_path: PathBuf::from("/tmp/dm-daemon-hol.pid"), + log_path: PathBuf::from("/tmp/dm-daemon-hol.log"), + relay: None, + discovery_relays: Vec::new(), + default_account_relays: Vec::new(), + secret_store: Some(crate::SecretStoreKind::File), + keychain_service: Some("daemon-keychain".to_owned()), + }; + let state = Arc::new(Mutex::new(DaemonState { + pid: 1, + started_at: 0, + last_runtime_activity: None, + })); + let events = DaemonEventHub::new(); + let workers = SharedDaemonWorkers::default(); + + // Hold the workers lock for the whole Execute. + let busy = workers.lock().await; + + let (mut server, client) = UnixStream::pair().expect("unix stream pair"); + let cli = Box::new(daemon_test_cli(crate::Command::Whoami)); + + let result = tokio::time::timeout( + Duration::from_secs(5), + handle_execute_connection(cli, &mut server, &defaults, state, events, &workers), + ) + .await + .expect("a relay-less Execute must not block on the held workers lock"); + result.expect("handle_execute_connection should succeed off the lock"); + + drop(busy); + drop(client); +} + #[tokio::test] async fn daemon_request_reader_within_returns_request_before_timeout() { let (mut server, mut client) = UnixStream::pair().expect("unix stream pair");