perf(daemon): scope sync.family's trace fence to the requested family

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());

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() {