diff --git a/src/daemon.rs b/src/daemon.rs index aef688932f..4d5c91ec34 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -2725,9 +2725,13 @@ pub struct ActorDaemonCoordinator { backend: Arc, coordinator: Arc>, - normalizer: AsyncMutex< - crate::daemon::trace_normalizer::TraceNormalizer< - crate::daemon::git_backend::SystemGitBackend, + // std Mutex (not AsyncMutex) so the normalizer can move into + // spawn_blocking closures; it is only ever locked from blocking context. + normalizer: Arc< + Mutex< + crate::daemon::trace_normalizer::TraceNormalizer< + crate::daemon::git_backend::SystemGitBackend, + >, >, >, pending_rebase_original_head_by_worktree: Mutex)>>, @@ -2822,8 +2826,8 @@ impl ActorDaemonCoordinator { coordinator: Arc::new(crate::daemon::coordinator::Coordinator::new( backend.clone(), )), - normalizer: AsyncMutex::new(crate::daemon::trace_normalizer::TraceNormalizer::new( - backend.clone(), + normalizer: Arc::new(Mutex::new( + crate::daemon::trace_normalizer::TraceNormalizer::new(backend.clone()), )), backend, pending_rebase_original_head_by_worktree: Mutex::new(HashMap::new()), @@ -6433,13 +6437,34 @@ impl ActorDaemonCoordinator { .and_then(Value::as_str) .unwrap_or_default() .to_string(); + let terminal_root_event = is_terminal_root_trace_event( + &event, + payload + .get("sid") + .and_then(Value::as_str) + .unwrap_or_default(), + payload_root_sid.as_deref().unwrap_or_default(), + ); if event == TRACE_CONNECTION_CLOSED_EVENT { let Some(root_sid) = payload_root_sid.as_deref() else { return Ok(TracePayloadApplyOutcome::None); }; { - let mut normalizer = self.normalizer.lock().await; - let _ = normalizer.sweep_orphans_for_roots(&[root_sid.to_string()]); + // Orphan sweeps read reflogs and repo state; run them on a + // blocking worker so filesystem I/O does not block a Tokio worker. + // Recover from poisoning: the ingest worker catches per-payload + // panics and continues, so a poisoned lock must not disable + // normalization for the daemon's remaining lifetime. + let normalizer = Arc::clone(&self.normalizer); + let root_sid_owned = root_sid.to_string(); + crate::tokio_runtime::spawn_blocking_result(move || { + let mut normalizer = normalizer + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let _ = normalizer.sweep_orphans_for_roots(&[root_sid_owned]); + Ok(()) + }) + .await?; } let replaced_family = self .replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled) @@ -6456,19 +6481,27 @@ impl ActorDaemonCoordinator { } self.maybe_append_pending_root_from_trace_payload(&payload)?; + // Normalization can touch the filesystem (alias resolution reads git + // config files on exit events); run it on a blocking worker so a slow + // disk or cold cache does not block a Tokio worker. The ingest task + // still awaits each result to preserve trace ordering. + // Recover from poisoning: the ingest worker catches per-payload panics + // and continues, so a poisoned lock must not disable normalization for + // the daemon's remaining lifetime. let emitted = { - let mut normalizer = self.normalizer.lock().await; - normalizer.ingest_payload(&payload)? + let normalizer = Arc::clone(&self.normalizer); + let payload_for_ingest = payload; + crate::tokio_runtime::spawn_blocking_result(move || { + normalizer + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .ingest_payload(&payload_for_ingest) + }) + .await? }; let Some(command) = emitted else { - if is_terminal_root_trace_event( - &event, - payload - .get("sid") - .and_then(Value::as_str) - .unwrap_or_default(), - payload_root_sid.as_deref().unwrap_or_default(), - ) && let Some(root_sid) = payload_root_sid.as_deref() + if terminal_root_event + && let Some(root_sid) = payload_root_sid.as_deref() && let Some(family) = self .replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled) .await? diff --git a/src/tokio_runtime.rs b/src/tokio_runtime.rs index a22710dda9..d0e0ff6f73 100644 --- a/src/tokio_runtime.rs +++ b/src/tokio_runtime.rs @@ -65,10 +65,16 @@ fn build_bounded_runtime( } pub(crate) fn build_daemon_runtime() -> Result { - build_bounded_runtime( - DAEMON_RUNTIME_WORKER_THREADS, - DAEMON_RUNTIME_MAX_BLOCKING_THREADS, - ) + #[cfg(feature = "test-support")] + let worker_threads = std::env::var("GIT_AI_TEST_DAEMON_RUNTIME_WORKER_THREADS") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DAEMON_RUNTIME_WORKER_THREADS); + #[cfg(not(feature = "test-support"))] + let worker_threads = DAEMON_RUNTIME_WORKER_THREADS; + + build_bounded_runtime(worker_threads, DAEMON_RUNTIME_MAX_BLOCKING_THREADS) } // Post-commit attribution calls this helper from inside the daemon runtime. diff --git a/tests/daemon_mode.rs b/tests/daemon_mode.rs index 6573722937..756cef992c 100644 --- a/tests/daemon_mode.rs +++ b/tests/daemon_mode.rs @@ -34,6 +34,8 @@ use std::fs; use std::io::{BufRead, BufReader}; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; +#[cfg(not(windows))] +use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; use std::sync::atomic::{AtomicBool, Ordering}; @@ -2676,6 +2678,155 @@ fn daemon_trace_connection_close_without_atexit_does_not_block_later_trace() { panic!("daemon did not process a later trace after a mutating root closed before atexit"); } +#[test] +#[cfg(not(windows))] +fn daemon_trace_normalization_offload_keeps_single_worker_runtime_responsive() { + let repo = TestRepo::new_with_daemon_scope(DaemonTestScope::NoDaemon); + let mut daemon = DaemonGuard::start_with_env( + &repo, + &[ + ("GIT_AI_TEST_DAEMON_RUNTIME_WORKER_THREADS", "1"), + ("GIT_AI_DAEMON_UPDATE_CHECK_INTERVAL", "86400"), + ("GIT_AI_DAEMON_MAX_UPTIME_SECS", "86400"), + ], + ); + let trace_socket = daemon_trace_socket_path(&repo); + let control_socket = daemon_control_socket_path(&repo); + let worktree = repo_workdir_string(&repo); + let git_dir = repo.path().join(".git").to_string_lossy().to_string(); + + // A FIFO-backed repository config gives the real SystemGitBackend a + // deterministic blocking read on the first alias lookup, without adding a + // delay hook to the latency-sensitive ingestion path. + let repo_config_path = repo.path().join(".git/config"); + let original_config = + fs::read_to_string(&repo_config_path).expect("failed to read original repository config"); + fs::remove_file(&repo_config_path).expect("failed to replace repository config with FIFO"); + let mkfifo = Command::new("mkfifo") + .arg(&repo_config_path) + .status() + .expect("failed to invoke mkfifo"); + assert!(mkfifo.success(), "mkfifo failed: {mkfifo}"); + + let sid = "blocking-alias-normalization"; + let mut trace_stream = + open_local_socket_stream_with_timeout(&trace_socket, DAEMON_TEST_PROBE_TIMEOUT) + .expect("failed to connect to trace socket"); + write_trace_frames_to_stream( + &mut trace_stream, + &[ + json!({ + "event": "start", + "sid": sid, + "argv": ["git", "ci", "-m", "synthetic"], + "time_ns": 11_000u64, + }), + json!({ + "event": "def_repo", + "sid": sid, + "worktree": worktree, + "repo": git_dir, + "time_ns": 11_001u64, + }), + ], + ); + thread::sleep(Duration::from_millis(150)); + + let response = send_control_request_with_timeout( + &control_socket, + &ControlRequest::StatusFamily { + repo_working_dir: repo_workdir_string(&repo), + }, + Duration::from_millis(500), + ) + .expect("blocking trace normalization must not occupy the only Tokio worker"); + assert!(response.ok, "status.family failed: {response:?}"); + + let deadline = std::time::Instant::now() + Duration::from_secs(2); + let mut alias_writer = loop { + match fs::OpenOptions::new() + .write(true) + .custom_flags(libc::O_NONBLOCK) + .open(&repo_config_path) + { + Ok(writer) => break writer, + Err(error) + if error.raw_os_error() == Some(libc::ENXIO) + && std::time::Instant::now() < deadline => + { + thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("failed to open blocking alias config: {error}"), + } + }; + alias_writer + .write_all(format!("{original_config}\n[alias]\n\tci = commit\n").as_bytes()) + .expect("failed to release blocking alias config read"); + drop(alias_writer); + fs::remove_file(&repo_config_path).expect("failed to remove blocking repository config FIFO"); + fs::write( + &repo_config_path, + format!("{original_config}\n[alias]\n\tci = commit\n"), + ) + .expect("failed to restore repository config"); + + let session = repos::test_repo::new_daemon_test_sync_session_id(); + let session_arg = format!("git-ai.testSyncSession={session}"); + let followup_sid = "trace-after-blocking-alias-normalization"; + write_trace_frames_to_stream( + &mut trace_stream, + &[ + json!({ + "event": "exit", + "sid": sid, + "code": 0, + "time_ns": 11_100u64, + }), + trace_atexit_frame(sid, 0, 11_101u64), + json!({ + "event": "start", + "sid": followup_sid, + "argv": ["git", "-c", session_arg, "commit", "-m", "followup"], + "time_ns": 12_000u64, + }), + json!({ + "event": "def_repo", + "sid": followup_sid, + "worktree": repo_workdir_string(&repo), + "repo": repo.path().join(".git").to_string_lossy().to_string(), + "time_ns": 12_001u64, + }), + json!({ + "event": "exit", + "sid": followup_sid, + "code": 0, + "time_ns": 12_100u64, + }), + trace_atexit_frame(followup_sid, 0, 12_101u64), + ], + ); + let sync = send_control_request_with_timeout( + &control_socket, + &ControlRequest::SyncFamily { + repo_working_dir: repo_workdir_string(&repo), + }, + Duration::from_secs(2), + ) + .expect("sync.family should complete after the blocking config read is released"); + assert!(sync.ok, "sync.family failed: {sync:?}"); + let matching_completions = repo + .daemon_completion_entries() + .into_iter() + .filter(|entry| entry.test_sync_session.as_deref() == Some(session.as_str())) + .collect::>(); + assert_eq!( + matching_completions.len(), + 1, + "a later trace must normalize and complete after the blocking read" + ); + daemon.shutdown(); +} + #[test] #[cfg(not(windows))] fn daemon_control_listener_stalled_connection_does_not_block_later_control_requests() {