From 52c908ef35acbf33fdbbfb05f828f38e178e622b Mon Sep 17 00:00:00 2001 From: Sasha Varlamov Date: Wed, 29 Jul 2026 15:13:33 +0000 Subject: [PATCH] perf(daemon): scope sync.family trace fence to requested family sync.family waited on the global trace fence, so an open mutating trace root in repository A delayed synchronization and test drains for unrelated repository B. Use the existing trace2-captured root family attribution to ignore roots already assigned to another family. Unattributed roots continue to fail closed, the requested family still waits for its own roots, and global consumers such as await and shutdown retain the global fence. Add internal fence coverage plus a two-TestRepo regression proving an unrelated family completes while the originating family remains blocked until the trace root closes. Re-lands the per-family fence idea from #1980 on the current architecture. Co-Authored-By: Claude Fable 5 --- src/daemon.rs | 116 ++++++++++++++++++++++++++++++++++++++++++- tests/daemon_mode.rs | 77 ++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+), 1 deletion(-) diff --git a/src/daemon.rs b/src/daemon.rs index 774d4162e4..aef688932f 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -3796,6 +3796,25 @@ impl ActorDaemonCoordinator { }) } + /// As [`Self::has_open_trace_roots_that_may_mutate_refs`], but scoped to + /// one family: roots already attributed to a DIFFERENT family (via their + /// `def_repo` worktree) cannot mutate this family's refs and are ignored. + /// Roots with no family attribution yet fail closed and block everyone. + fn has_open_trace_roots_that_may_mutate_family(&self, family: &str) -> bool { + let Ok(ingress) = self.trace_ingress_state.lock() else { + return false; + }; + ingress.root_open_connections.iter().any(|(root, count)| { + *count > 0 + && !ingress.root_definitely_read_only.contains(root) + && ingress.root_mutating.get(root).copied().unwrap_or(true) + && ingress + .root_families + .get(root) + .is_none_or(|root_family| root_family == family) + }) + } + fn next_trace_ingest_seq(&self) -> u64 { // Relaxed: we only need fetch_add atomicity (unique monotone values), // not ordering w.r.t. any other atomic. @@ -4363,6 +4382,32 @@ impl ActorDaemonCoordinator { } } + /// As [`Self::wait_for_trace_ingest_processed_through`], but scoped to one + /// family: open mutating roots already attributed to a different family do + /// not hold this fence, so a long-running git command in one repository no + /// longer delays `sync.family` for every other repository. Unattributed + /// roots still fail closed and block until their `def_repo` arrives. + async fn wait_for_trace_ingest_processed_through_family(&self, family: &str) { + loop { + let target = self.next_trace_ingest_seq.load(Ordering::Acquire) as u64; + self.wait_for_trace_ingest_seq(target).await; + + // Enroll before checking (see wait_for_trace_ingest_seq): the + // notify_waiters fired by the root's close/def_repo must not race + // the condition load. + let progress = self.trace_ingest_progress_notify.notified(); + tokio::pin!(progress); + progress.as_mut().enable(); + if !self.has_open_trace_roots_that_may_mutate_family(family) { + return; + } + tokio::select! { + _ = &mut progress => {} + _ = self.wait_for_shutdown() => return, + } + } + } + /// Prepares `payload` for ingestion and returns whether it should be /// enqueued. /// @@ -6590,7 +6635,8 @@ impl ActorDaemonCoordinator { .await; self.wait_for_no_unadmitted_checkpoints().await; let family = self.backend.resolve_family(Path::new(&repo_working_dir))?; - self.wait_for_trace_ingest_processed_through().await; + self.wait_for_trace_ingest_processed_through_family(&family.0) + .await; let exec_lock = self.side_effect_exec_lock(&family.0)?; loop { @@ -9924,6 +9970,74 @@ mod tests { .expect("checkpoint fence should pass once the mutating trace root closes"); } + #[tokio::test] + async fn family_fence_ignores_open_mutating_roots_of_other_families() { + let coord = Arc::new(ActorDaemonCoordinator::new()); + let temp = tempfile::tempdir().unwrap(); + let other_repo = temp.path().join("other-repo"); + std::fs::create_dir_all(other_repo.join(".git")).unwrap(); + std::fs::write( + other_repo.join(".git").join("HEAD"), + "ref: refs/heads/main\n", + ) + .unwrap(); + + let sid = "20260411T120000.000000-Psid1"; + coord.trace_root_connection_opened(sid).unwrap(); + let mut start = make_start_payload(&["git", "commit", "-m", "other repo commit"]); + assert!(coord.prepare_trace_payload_for_ingest(&mut start)); + + // Before the root is attributed to a repository, it must block every + // family (fail closed: it could belong to any of them). + assert!( + tokio::time::timeout( + Duration::from_millis(50), + coord.wait_for_trace_ingest_processed_through_family("/some/unrelated/family") + ) + .await + .is_err(), + "an unattributed mutating root must block all family fences" + ); + + // def_repo attributes the root to other-repo's family; unrelated + // families must no longer wait on it. + let mut def_repo = serde_json::json!({ + "event": "def_repo", + "sid": sid, + "worktree": other_repo.to_string_lossy(), + }); + assert!(coord.prepare_trace_payload_for_ingest(&mut def_repo)); + + tokio::time::timeout( + Duration::from_millis(250), + coord.wait_for_trace_ingest_processed_through_family("/some/unrelated/family"), + ) + .await + .expect("a mutating root attributed to another family must not block this fence"); + + // The root's own family still waits until the connection closes. + let own_family = coord.backend.resolve_family(&other_repo).unwrap().0; + assert!( + tokio::time::timeout( + Duration::from_millis(50), + coord.wait_for_trace_ingest_processed_through_family(&own_family) + ) + .await + .is_err(), + "the root's own family fence must still wait for the open root" + ); + + coord + .record_trace_connection_close(&[sid.to_string()]) + .unwrap(); + tokio::time::timeout( + Duration::from_secs(1), + coord.wait_for_trace_ingest_processed_through_family(&own_family), + ) + .await + .expect("own family fence should pass once the root closes"); + } + #[tokio::test] async fn trace_connection_close_without_atexit_cancels_pending_root() { let coord = Arc::new(ActorDaemonCoordinator::new()); diff --git a/tests/daemon_mode.rs b/tests/daemon_mode.rs index 98cc816836..6573722937 100644 --- a/tests/daemon_mode.rs +++ b/tests/daemon_mode.rs @@ -2408,6 +2408,83 @@ fn daemon_stalled_unidentified_trace_connection_does_not_block_sync_control_requ ); } +#[test] +#[cfg(not(windows))] +fn daemon_sync_family_ignores_open_mutating_root_from_other_family() { + let first_repo = TestRepo::new_dedicated_daemon(); + let second_repo = TestRepo::new_with_daemon_scope(DaemonTestScope::NoDaemon); + let trace_socket = daemon_trace_socket_path(&first_repo); + let control_socket = daemon_control_socket_path(&first_repo); + let first_worktree = repo_workdir_string(&first_repo); + let second_worktree = repo_workdir_string(&second_repo); + let first_git_dir = first_repo.path().join(".git").to_string_lossy().to_string(); + let sid = "cross-family-open-mutating-root"; + + let mut open_trace = + 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 open_trace, + &[ + json!({ + "event": "start", + "sid": sid, + "argv": ["git", "commit", "-m", "long-running commit"], + "time_ns": 1_000u64, + }), + json!({ + "event": "def_repo", + "sid": sid, + "worktree": first_worktree, + "repo": first_git_dir, + "time_ns": 1_001u64, + }), + ], + ); + thread::sleep(Duration::from_millis(150)); + + let own_control_socket = control_socket.clone(); + let own_worktree = repo_workdir_string(&first_repo); + let (own_sync_tx, own_sync_rx) = mpsc::channel(); + let own_sync = thread::spawn(move || { + let response = send_control_request_with_timeout( + &own_control_socket, + &ControlRequest::SyncFamily { + repo_working_dir: own_worktree, + }, + Duration::from_secs(5), + ); + let _ = own_sync_tx.send(response); + }); + assert!( + own_sync_rx + .recv_timeout(Duration::from_millis(250)) + .is_err(), + "sync.family must still wait for an open mutating root in its own family" + ); + + let unrelated_sync = send_control_request_with_timeout( + &control_socket, + &ControlRequest::SyncFamily { + repo_working_dir: second_worktree, + }, + Duration::from_secs(1), + ) + .expect("an open mutating root from another family must not block sync.family"); + assert!( + unrelated_sync.ok, + "unrelated sync.family request failed: {unrelated_sync:?}" + ); + + write_trace_frames_to_stream(&mut open_trace, &[trace_atexit_frame(sid, 0, 1_002)]); + let own_sync_response = own_sync_rx + .recv_timeout(Duration::from_secs(1)) + .expect("own-family sync should complete after the trace root closes") + .expect("own-family sync request failed"); + assert!(own_sync_response.ok, "own-family sync failed"); + own_sync.join().unwrap(); +} + #[test] #[cfg(not(windows))] fn daemon_partial_trace_line_does_not_block_checkpoint_control_request() {