From 7e87137266d00556367d83d691cc2521b125677b Mon Sep 17 00:00:00 2001 From: Sasha Varlamov Date: Sun, 30 Aug 2026 22:58:04 +0000 Subject: [PATCH] fix(daemon): never run family side effects inline on the trace ingest worker A long write-op side-effect pass (e.g. a large checkout/commit in a monorepo) used to execute inline on the serial trace-ingest worker: apply_trace_payload_to_state took the family exec lock and drained the family sequencer in place, and the non-sequencer Applied branch ran maybe_apply_side_effects_for_applied_command directly. While one family ground through minutes of git work, the processed_trace_ingest_seq watermark froze, so checkpoint admission (which waits on that watermark) stalled for every repository family: checkpoints were "received into bounded ingress" but never admitted, queued trace payloads backed up, and the socket health check tripped processing_stalled restarts that were deferred while the very checkpoints the stall blocked remained outstanding (#2252). Trace ingestion is now sequencing-only: sequencer mutations are constant-time map operations, and drains run on detached tasks that serialize per family via the existing side_effect_exec_lock. Side effects of commands that do not participate in the sequencer (git am, filter-branch, ...) are sequenced too, as AppliedSideEffects entries ordered by command start time, so they keep the per-family ordering the inline worker used to guarantee. Because side-effect passes do blocking git work on the 4-thread daemon runtime, command passes are bounded by a dedicated 2-permit semaphore (checkpoint side effects keep their own: they run on the blocking pool and must not be starved by long command passes), so simultaneously grinding families can never occupy every worker thread. The watermark used to imply side-effect completion, so the fences that relied on it are restored explicitly: in-flight passes register (RAII, before entries are popped) in inflight_effects_by_family; sync.family drains every family's ready entries and waits for in-flight passes (side effects can write into other repositories - a push to an explicit path pushes authorship notes into the destination repo); and graceful shutdown plus the update/uptime restart paths defer while a pass is executing so process teardown cannot kill one mid-write. Entries fail-closed behind a still-open trace root stay queued without blocking any fence, exactly as before. Fixes #2252 Co-Authored-By: Claude Fable 5 --- src/daemon.rs | 643 ++++++++++++++++++++++++++++++++----------- tests/daemon_mode.rs | 325 ++++++++++++++++++++++ 2 files changed, 813 insertions(+), 155 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index 359de5a4df..df92de710b 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -104,6 +104,14 @@ const DAEMON_SOCKET_PROBE_TIMEOUT: Duration = Duration::from_millis(100); const CHECKPOINT_INGRESS_REQUEST_LIMIT: usize = 1_024; const CHECKPOINT_INGRESS_BYTE_LIMIT: usize = 64 * 1024 * 1024; const CHECKPOINT_FAMILY_DRAIN_CONCURRENCY: usize = 2; +/// Global bound on concurrently executing command side-effect passes. They +/// do blocking git work directly on the daemon runtime's 4 worker threads, +/// and with drains running on detached tasks an unbounded number of +/// simultaneously grinding families could otherwise occupy every worker and +/// starve the runtime (#2252). Checkpoint side effects keep their own +/// semaphore: they run on the blocking pool and must not be starved by +/// long command passes. +const COMMAND_SIDE_EFFECT_CONCURRENCY: usize = 2; // Trace2 frames are written synchronously by Git to the daemon's Unix socket. // With small kernel socket buffers (macOS defaults to ~8 KiB), a bursty trace2 // stream can fill the buffer and block the raw `git` process in `write()` until @@ -2648,6 +2656,14 @@ fn read_checkpoint_body( enum FamilySequencerEntry { PendingRoot, ReadyCommand(Box), + /// A command already applied to family state (it did not participate in + /// the sequencer, e.g. `git am`) whose side-effect pass is still pending. + /// Sequencing the pass keeps it ordered with, and serialized against, + /// the family's other passes (#2252). + AppliedSideEffects { + applied: Box, + commit_file_timestamp_snapshots: CommitFileTimestampSnapshotHandles, + }, Checkpoint { request: Box, receipt_seq: u64, @@ -2682,6 +2698,30 @@ const COMMIT_FILE_TIMESTAMP_SNAPSHOT_WAIT: Duration = Duration::from_millis(500) const SESSION_EVENT_RECOVERY_PREFLIGHT_WAIT: Duration = Duration::from_secs(2); const SESSION_EVENT_RECOVERY_PREFLIGHT_POLL: Duration = Duration::from_millis(100); +/// RAII registration of an in-flight family side-effect pass; see +/// [`ActorDaemonCoordinator::begin_family_effect_guarded`]. +struct FamilyEffectGuard<'a> { + coordinator: &'a ActorDaemonCoordinator, + family: String, +} + +impl Drop for FamilyEffectGuard<'_> { + fn drop(&mut self) { + let _ = self.coordinator.end_family_effect(&self.family); + } +} + +/// Extracts a printable message from a `catch_unwind` panic payload. +fn panic_payload_message(panic_payload: &(dyn std::any::Any + Send)) -> String { + if let Some(message) = panic_payload.downcast_ref::() { + message.clone() + } else if let Some(message) = panic_payload.downcast_ref::<&str>() { + (*message).to_string() + } else { + "unknown panic".to_string() + } +} + fn run_blocking_side_effect(operation: impl FnOnce() -> T) -> T { if tokio::runtime::Handle::try_current() .is_ok_and(|handle| handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread) @@ -2768,7 +2808,11 @@ pub struct ActorDaemonCoordinator { Mutex>>, side_effect_errors_by_family: Mutex>>, side_effect_exec_locks: Mutex>>>, + /// Families with a scheduled-or-running coalesced drain task; see + /// [`Self::schedule_family_drain`]. + scheduled_family_drains: Mutex>, checkpoint_side_effect_semaphore: Semaphore, + command_side_effect_semaphore: Semaphore, checkpoint_ingress_quota: Arc, checkpoint_ingress_tx: std::sync::OnceLock>, next_checkpoint_receipt_seq: AtomicUsize, @@ -2865,7 +2909,6 @@ impl DaemonExitAction { enum TracePayloadApplyOutcome { None, - Applied(Box), QueuedFamily, } @@ -2892,7 +2935,9 @@ impl ActorDaemonCoordinator { recent_replay_prerequisites_by_family: Mutex::new(HashMap::new()), side_effect_errors_by_family: Mutex::new(HashMap::new()), side_effect_exec_locks: Mutex::new(HashMap::new()), + scheduled_family_drains: Mutex::new(HashSet::new()), checkpoint_side_effect_semaphore: Semaphore::new(CHECKPOINT_FAMILY_DRAIN_CONCURRENCY), + command_side_effect_semaphore: Semaphore::new(COMMAND_SIDE_EFFECT_CONCURRENCY), checkpoint_ingress_quota: Arc::new(CheckpointIngressQuota::new( CHECKPOINT_INGRESS_REQUEST_LIMIT, CHECKPOINT_INGRESS_BYTE_LIMIT, @@ -3295,6 +3340,58 @@ impl ActorDaemonCoordinator { Ok(()) } + /// Registers an in-flight family side-effect pass for the guard's + /// lifetime. Sync, await, and graceful shutdown fence on this + /// registration, so it must be released on every exit path — early + /// returns, panics, and dropped tasks included — which only a Drop + /// impl can guarantee. + fn begin_family_effect_guarded<'a>(&'a self, family: &str) -> FamilyEffectGuard<'a> { + let _ = self.begin_family_effect(family); + FamilyEffectGuard { + coordinator: self, + family: family.to_string(), + } + } + + /// Attribution work an automatic restart would abandon mid-flight: + /// accepted checkpoints, queued trace payloads, actionable sequencer + /// entries, and executing side-effect passes. PendingRoot placeholders + /// are deliberately excluded so an idle interactive command (e.g. a + /// rebase waiting on an editor) cannot defer restarts forever (#2252). + fn has_pending_attribution_work(&self) -> bool { + if self.outstanding_checkpoint_state().0 > 0 { + return true; + } + if self.queued_trace_payloads.load(Ordering::Relaxed) > 0 { + return true; + } + if self.has_inflight_family_effects() { + return true; + } + if let Ok(map) = self.family_sequencers_by_family.lock() + && map.values().any(|state| { + state + .entries + .values() + .any(|entry| !matches!(entry, FamilySequencerEntry::PendingRoot)) + }) + { + return true; + } + false + } + + /// Whether any detached side-effect pass is currently in flight, in any + /// family. Passes register via `begin_family_effect` before the trace + /// ingest watermark advances, so this is a valid completion fence for + /// work the watermark no longer covers (#2252). + fn has_inflight_family_effects(&self) -> bool { + self.inflight_effects_by_family + .lock() + .map(|map| !map.is_empty()) + .unwrap_or(false) + } + fn end_family_effect(&self, family: &str) -> Result<(), GitAiError> { let mut map = self .inflight_effects_by_family @@ -3525,35 +3622,34 @@ impl ActorDaemonCoordinator { self.append_pending_root_entry(&family, root_sid, started_at_ns) } - async fn append_ready_command_entry( + /// Appends an entry to the family sequencer, ordered by the originating + /// command's start time. The caller is responsible for scheduling a + /// drain of the family afterwards — this must stay a constant-time map + /// insert because it runs on the serial trace ingest worker, whose + /// watermark checkpoint admission waits on (#2252). + fn append_family_sequencer_entry( &self, family: &str, - command: crate::daemon::domain::NormalizedCommand, + started_at_ns: u128, + entry: FamilySequencerEntry, ) -> Result<(), GitAiError> { - let exec_lock = self.side_effect_exec_lock(family)?; - let _guard = exec_lock.lock().await; - { - let mut sequencers = self.family_sequencers_by_family.lock().map_err(|_| { - GitAiError::Generic("family sequencer map lock poisoned".to_string()) - })?; - let state = - sequencers - .entry(family.to_string()) - .or_insert_with(|| FamilySequencerState { - next_ordinal: 1, - entries: BTreeMap::new(), - }); - let order = FamilySequencerOrder { - started_at_ns: command.started_at_ns, - ordinal: state.next_ordinal, - }; - state.next_ordinal = state.next_ordinal.saturating_add(1); - state - .entries - .insert(order, FamilySequencerEntry::ReadyCommand(Box::new(command))); - } - self.drain_ready_family_sequencer_entries_locked(family) - .await + let mut sequencers = self + .family_sequencers_by_family + .lock() + .map_err(|_| GitAiError::Generic("family sequencer map lock poisoned".to_string()))?; + let state = sequencers + .entry(family.to_string()) + .or_insert_with(|| FamilySequencerState { + next_ordinal: 1, + entries: BTreeMap::new(), + }); + let order = FamilySequencerOrder { + started_at_ns, + ordinal: state.next_ordinal, + }; + state.next_ordinal = state.next_ordinal.saturating_add(1); + state.entries.insert(order, entry); + Ok(()) } async fn drain_ready_family_sequencer_entries(&self, family: &str) -> Result<(), GitAiError> { @@ -3586,18 +3682,111 @@ impl ActorDaemonCoordinator { Ok(()) } - async fn drain_ready_family_sequencers_after_root_cleared( - &self, - family: Option, - ) -> Result<(), GitAiError> { + /// Schedules drains for sequencer entries unblocked by a cleared trace + /// root. Drains run detached: side-effect passes are unbounded-duration + /// git work and must never execute inline on the trace ingest worker, + /// whose watermark checkpoint admission waits on (#2252). + fn schedule_ready_family_drains_after_root_cleared(self: &Arc, family: Option) { if let Some(family) = family { - self.drain_ready_family_sequencer_entries(&family).await + self.schedule_family_drain(family); } else { - self.drain_all_ready_family_sequencers().await + self.schedule_all_ready_family_drains(); } } - async fn replace_pending_root_entry( + /// Schedules a detached drain for one family, coalescing to at most one + /// scheduled-or-running drain task per family. The marker is released + /// only after a pass that ends with no actionable front entry (checked + /// atomically with the release), so an entry appended before this call + /// is always covered: either the running task's next pass pops it, or + /// this call spawns a fresh task. + fn schedule_family_drain(self: &Arc, family: String) { + { + let Ok(mut scheduled) = self.scheduled_family_drains.lock() else { + return; + }; + if !scheduled.insert(family.clone()) { + return; + } + } + let coordinator = Arc::clone(self); + tokio::spawn(async move { + loop { + if let Err(error) = coordinator + .drain_ready_family_sequencer_entries(&family) + .await + { + tracing::error!( + component = "daemon", + phase = "checkpoint_processing", + reason = "family_drain_failed", + %family, + %error, + "failed draining family sequencer" + ); + if let Ok(mut scheduled) = coordinator.scheduled_family_drains.lock() { + scheduled.remove(&family); + } + return; + } + // Deregister atomically with the emptiness check: an entry + // appended after the pass but before deregistration must + // either be seen here (loop again) or by the fresh task its + // own schedule call spawns after we release the marker. + let Ok(mut scheduled) = coordinator.scheduled_family_drains.lock() else { + return; + }; + if !coordinator.family_has_actionable_front_entry(&family) { + scheduled.remove(&family); + return; + } + } + }); + } + + /// Whether the family's sequencer front would be popped by a drain right + /// now — mirrors the gates of `drain_ready_family_sequencer_entries_locked` + /// (unadmitted checkpoints, PendingRoot front, prior-open-root fencing). + /// Gated-but-present entries return false: the event that lifts their + /// gate (admission completion, root clear) schedules its own drain. + fn family_has_actionable_front_entry(&self, family: &str) -> bool { + if self.unadmitted_checkpoints.load(Ordering::Acquire) > 0 { + return false; + } + let Ok(map) = self.family_sequencers_by_family.lock() else { + return false; + }; + let Some(state) = map.get(family) else { + return false; + }; + let Some((order, entry)) = state.entries.first_key_value() else { + return false; + }; + if matches!(entry, FamilySequencerEntry::PendingRoot) { + return false; + } + let entry_root_sid = match entry { + FamilySequencerEntry::ReadyCommand(command) => Some(command.root_sid.as_str()), + FamilySequencerEntry::AppliedSideEffects { applied, .. } => { + Some(applied.command.root_sid.as_str()) + } + _ => None, + }; + !self + .family_entry_blocked_by_prior_open_trace_root( + family, + order.started_at_ns, + entry_root_sid, + ) + .unwrap_or(true) + } + + /// Replaces a root's PendingRoot sequencer entry with `replacement` and + /// returns the family whose sequencer changed. The caller is responsible + /// for scheduling a drain of that family afterwards — this must stay a + /// constant-time map mutation because it runs on the serial trace ingest + /// worker, whose watermark checkpoint admission waits on (#2252). + fn replace_pending_root_entry( &self, root_sid: &str, replacement: FamilySequencerEntry, @@ -3606,8 +3795,6 @@ impl ActorDaemonCoordinator { return Ok(None); }; let family = slot.family.clone(); - let exec_lock = self.side_effect_exec_lock(&family)?; - let _guard = exec_lock.lock().await; { let mut sequencers = self.family_sequencers_by_family.lock().map_err(|_| { GitAiError::Generic("family sequencer map lock poisoned".to_string()) @@ -3636,8 +3823,6 @@ impl ActorDaemonCoordinator { } } } - self.drain_ready_family_sequencer_entries_locked(&family) - .await?; Ok(Some(family)) } @@ -3842,20 +4027,27 @@ impl ActorDaemonCoordinator { .lock() .map_err(|_| GitAiError::Generic("trace ingress state lock poisoned".to_string()))?; for root_sid in roots { - if let Some(count) = ingress.root_open_connections.get_mut(root_sid) { - if *count > 1 { - *count -= 1; - continue; - } - ingress.root_open_connections.remove(root_sid); + if let Some(count) = ingress.root_open_connections.get_mut(root_sid) + && *count > 1 + { + *count -= 1; + continue; } if !Self::trace_root_needs_close_marker(&ingress, root_sid) { + ingress.root_open_connections.remove(root_sid); Self::clear_trace_ingress_root_locked(&mut ingress, root_sid); continue; } if ingress.root_close_markers_enqueued.contains(root_sid) { continue; } + // Keep the root registered as open until the ingest worker has + // processed its queued frames and this close marker + // (clear_trace_root_tracking removes the registration then). The + // reader runs ahead of the worker; clearing here would drop the + // open-root fence while the root's own command is still queued, + // letting a detached drain execute a later command's pass first + // and invert per-family side-effect order (#2252). ingress.root_close_markers_enqueued.insert(root_sid.clone()); close_marker_candidates.push(root_sid.clone()); } @@ -4109,14 +4301,7 @@ impl ActorDaemonCoordinator { Err(error) } Err(panic_payload) => { - let panic_msg = - if let Some(s) = panic_payload.downcast_ref::() { - s.clone() - } else if let Some(s) = panic_payload.downcast_ref::<&str>() { - s.to_string() - } else { - "unknown panic".to_string() - }; + let panic_msg = panic_payload_message(panic_payload.as_ref()); tracing::error!( component = "daemon", phase = "trace_ingest_worker", @@ -4251,14 +4436,7 @@ impl ActorDaemonCoordinator { } } Err(panic_payload) => { - let panic_msg = - if let Some(message) = panic_payload.downcast_ref::() { - message.clone() - } else if let Some(message) = panic_payload.downcast_ref::<&str>() { - message.to_string() - } else { - "unknown panic".to_string() - }; + let panic_msg = panic_payload_message(panic_payload.as_ref()); tracing::error!( component = "daemon", phase = "checkpoint_ingress_worker", @@ -4810,6 +4988,10 @@ impl ActorDaemonCoordinator { &self, family: &str, ) -> Result<(), GitAiError> { + // Register the in-flight pass BEFORE popping entries: completion + // fences must never observe an empty sequencer with no registered + // pass while popped entries are about to execute (#2252). + let _family_effect = self.begin_family_effect_guarded(family); let mut ready: Vec<(u64, FamilySequencerEntry)> = Vec::new(); let mut progressed = false; { @@ -4828,6 +5010,9 @@ impl ActorDaemonCoordinator { } let entry_root_sid = match first_entry.get() { FamilySequencerEntry::ReadyCommand(command) => Some(command.root_sid.as_str()), + FamilySequencerEntry::AppliedSideEffects { applied, .. } => { + Some(applied.command.root_sid.as_str()) + } _ => None, }; if self.family_entry_blocked_by_prior_open_trace_root( @@ -4854,7 +5039,6 @@ impl ActorDaemonCoordinator { return Ok(()); } - let _ = self.begin_family_effect(family); for (order, ready_entry) in ready { // Per-family drains must be strictly serialized; overlapping or // order-regressing exec windows in a wltrace capture indicate a @@ -4865,6 +5049,14 @@ impl ActorDaemonCoordinator { "command:{}", command.primary_command.as_deref().unwrap_or("unknown") ), + FamilySequencerEntry::AppliedSideEffects { applied, .. } => format!( + "applied:{}", + applied + .command + .primary_command + .as_deref() + .unwrap_or("unknown") + ), FamilySequencerEntry::Checkpoint { receipt_seq, .. } => { format!("checkpoint:seq={receipt_seq}") } @@ -4874,6 +5066,13 @@ impl ActorDaemonCoordinator { }); match ready_entry { FamilySequencerEntry::ReadyCommand(command) => { + let _side_effect_permit = self + .command_side_effect_semaphore + .acquire() + .await + .map_err(|_| { + GitAiError::Generic("command side-effect semaphore closed".to_string()) + })?; // Wrap the entire command + side-effect pipeline in catch_unwind // so that a panic (e.g. from UTF-8 boundary issues in diff parsing) // does not kill the daemon process. @@ -4931,14 +5130,7 @@ impl ActorDaemonCoordinator { ); } Err(panic_payload) => { - let panic_msg = if let Some(s) = panic_payload.downcast_ref::() - { - s.clone() - } else if let Some(s) = panic_payload.downcast_ref::<&str>() { - s.to_string() - } else { - "unknown panic".to_string() - }; + let panic_msg = panic_payload_message(panic_payload.as_ref()); let error = GitAiError::Generic(format!( "daemon command side effect panic: {}", panic_msg @@ -4956,6 +5148,59 @@ impl ActorDaemonCoordinator { } } } + FamilySequencerEntry::AppliedSideEffects { + applied, + mut commit_file_timestamp_snapshots, + } => { + let _side_effect_permit = self + .command_side_effect_semaphore + .acquire() + .await + .map_err(|_| { + GitAiError::Generic("command side-effect semaphore closed".to_string()) + })?; + let side_effect_result = { + let future = self.maybe_apply_side_effects_for_applied_command( + Some(family), + &applied, + &mut commit_file_timestamp_snapshots, + ); + let caught = std::panic::AssertUnwindSafe(future); + match futures::FutureExt::catch_unwind(caught).await { + Ok(result) => result, + Err(panic_payload) => { + let panic_msg = panic_payload_message(panic_payload.as_ref()); + Err(GitAiError::Generic(format!( + "daemon command side effect panic: {}", + panic_msg + ))) + } + } + }; + if let Err(error) = &side_effect_result { + let _ = self.record_side_effect_error(family, order, error); + tracing::error!( + %error, + %family, + seq = applied.seq, + "command side effect failed" + ); + } + if let Err(error) = self.append_command_completion_log( + family, + &applied, + &side_effect_result, + order, + ) { + let _ = self.record_side_effect_error(family, order, &error); + tracing::error!( + %error, + %family, + order, + "command completion log write failed" + ); + } + } FamilySequencerEntry::Checkpoint { mut request, receipt_seq, @@ -5080,14 +5325,7 @@ impl ActorDaemonCoordinator { let result = match checkpoint_request { Ok(inner) => inner, Err(panic_payload) => { - let panic_msg = if let Some(s) = panic_payload.downcast_ref::() - { - s.clone() - } else if let Some(s) = panic_payload.downcast_ref::<&str>() { - s.to_string() - } else { - "unknown panic".to_string() - }; + let panic_msg = panic_payload_message(panic_payload.as_ref()); tracing::error!( component = "daemon", phase = "checkpoint_side_effect", @@ -5232,7 +5470,6 @@ impl ActorDaemonCoordinator { FamilySequencerEntry::PendingRoot => {} } } - let _ = self.end_family_effect(family); let _ = progressed; Ok(()) @@ -5913,6 +6150,9 @@ impl ActorDaemonCoordinator { && let Ok(delay_ms) = delay_ms.parse::() && delay_ms > 0 { + // Lets tests poll for the grind having actually started + // instead of guessing with fixed sleeps. + tracing::info!(op = primary, delay_ms, "test side-effect delay started"); tokio::time::sleep(Duration::from_millis(delay_ms)).await; break; } @@ -6735,7 +6975,7 @@ impl ActorDaemonCoordinator { } async fn apply_trace_payload_to_state( - &self, + self: &Arc, payload: Value, ) -> Result { let payload_root_sid = Self::trace_payload_root_sid(&payload); @@ -6752,17 +6992,15 @@ impl ActorDaemonCoordinator { let mut normalizer = self.normalizer.lock().await; let _ = normalizer.sweep_orphans_for_roots(&[root_sid.to_string()]); } - let replaced_family = self - .replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled) - .await?; + let replaced_family = + self.replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled)?; let outcome = if replaced_family.is_some() { TracePayloadApplyOutcome::QueuedFamily } else { TracePayloadApplyOutcome::None }; self.clear_trace_root_tracking(root_sid)?; - self.drain_ready_family_sequencers_after_root_cleared(replaced_family) - .await?; + self.schedule_ready_family_drains_after_root_cleared(replaced_family); return Ok(outcome); } @@ -6780,13 +7018,11 @@ impl ActorDaemonCoordinator { .unwrap_or_default(), payload_root_sid.as_deref().unwrap_or_default(), ) && let Some(root_sid) = payload_root_sid.as_deref() - && let Some(family) = self - .replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled) - .await? + && let Some(family) = + self.replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled)? { self.clear_trace_root_tracking(root_sid)?; - self.drain_ready_family_sequencers_after_root_cleared(Some(family)) - .await?; + self.schedule_ready_family_drains_after_root_cleared(Some(family)); return Ok(TracePayloadApplyOutcome::QueuedFamily); } return Ok(TracePayloadApplyOutcome::None); @@ -6794,13 +7030,10 @@ impl ActorDaemonCoordinator { let root_sid = command.root_sid.clone(); let mut family_to_drain_after_clear = None; - let outcome = if let Some(family) = self - .replace_pending_root_entry( - &root_sid, - FamilySequencerEntry::ReadyCommand(Box::new(command.clone())), - ) - .await? - { + let outcome = if let Some(family) = self.replace_pending_root_entry( + &root_sid, + FamilySequencerEntry::ReadyCommand(Box::new(command.clone())), + )? { self.cache_commit_file_timestamp_snapshots_for_command(&command)?; family_to_drain_after_clear = Some(family); TracePayloadApplyOutcome::QueuedFamily @@ -6811,12 +7044,44 @@ impl ActorDaemonCoordinator { ) { self.cache_commit_file_timestamp_snapshots_for_command(&command)?; - self.append_ready_command_entry(&family, command).await?; + let started_at_ns = command.started_at_ns; + self.append_family_sequencer_entry( + &family, + started_at_ns, + FamilySequencerEntry::ReadyCommand(Box::new(command)), + )?; family_to_drain_after_clear = Some(family); TracePayloadApplyOutcome::QueuedFamily } else { match self.coordinator.route_command(command).await { - Ok(applied) => TracePayloadApplyOutcome::Applied(Box::new(applied)), + Ok(applied) => { + if let Some(family) = + applied.command.family_key.as_ref().map(|key| key.0.clone()) + { + // The command is applied to family state, but its + // side-effect pass is unbounded git work: sequence it + // so drains execute it off-worker, ordered by the + // command's start time and serialized with the + // family's other passes (#2252). + let commit_file_timestamp_snapshots = + Self::start_commit_file_timestamp_snapshots_for_command( + &applied.command, + ); + let started_at_ns = applied.command.started_at_ns; + self.append_family_sequencer_entry( + &family, + started_at_ns, + FamilySequencerEntry::AppliedSideEffects { + applied: Box::new(applied), + commit_file_timestamp_snapshots, + }, + )?; + family_to_drain_after_clear = Some(family); + TracePayloadApplyOutcome::QueuedFamily + } else { + TracePayloadApplyOutcome::None + } + } Err(error) => { let _ = self.clear_trace_root_tracking(&root_sid); return Err(error); @@ -6824,8 +7089,7 @@ impl ActorDaemonCoordinator { } }; self.clear_trace_root_tracking(&root_sid)?; - self.drain_ready_family_sequencers_after_root_cleared(family_to_drain_after_clear) - .await?; + self.schedule_ready_family_drains_after_root_cleared(family_to_drain_after_clear); Ok(outcome) } @@ -6833,45 +7097,7 @@ impl ActorDaemonCoordinator { if !is_trace_payload(&payload) { return Ok(()); } - match self.apply_trace_payload_to_state(payload).await? { - TracePayloadApplyOutcome::None | TracePayloadApplyOutcome::QueuedFamily => {} - TracePayloadApplyOutcome::Applied(applied) => { - if let Some(family) = applied.command.family_key.as_ref().map(|key| key.0.clone()) { - self.begin_family_effect(&family)?; - let mut commit_file_timestamp_snapshots = - Self::start_commit_file_timestamp_snapshots_for_command(&applied.command); - let result = self - .maybe_apply_side_effects_for_applied_command( - Some(&family), - &applied, - &mut commit_file_timestamp_snapshots, - ) - .await; - let _ = self.end_family_effect(&family); - if let Err(error) = &result { - let _ = self.record_side_effect_error(&family, applied.seq, error); - tracing::error!( - %error, - %family, - seq = applied.seq, - "async side-effect error" - ); - } - if let Err(error) = - self.append_command_completion_log(&family, &applied, &result, applied.seq) - { - let _ = self.record_side_effect_error(&family, applied.seq, &error); - tracing::error!( - %error, - %family, - seq = applied.seq, - "async completion log write failed" - ); - } - } - } - } - + let _ = self.apply_trace_payload_to_state(payload).await?; Ok(()) } @@ -6949,17 +7175,26 @@ impl ActorDaemonCoordinator { self.wait_for_trace_ingest_processed_through_family(&family.0) .await; - let exec_lock = self.side_effect_exec_lock(&family.0)?; loop { self.wait_for_no_unadmitted_checkpoints().await; - let guard = exec_lock.lock().await; - self.drain_ready_family_sequencer_entries_locked(&family.0) - .await?; - if self.unadmitted_checkpoints.load(Ordering::Acquire) == 0 { - drop(guard); + // Drain every family, not just the synced one: side-effect + // passes can write into other repositories (`git push ` + // pushes authorship notes into the destination repo), and before + // side effects ran detached the global ingest watermark implied + // all already-ingested passes had completed. Ready entries are + // executed here or by their detached drains (serialized per + // family by the exec lock); passes already handed off are + // visible via the effect registrations they take before the + // watermark advances (#2252). Entries fail-closed behind a + // still-open trace root stay queued without blocking this loop, + // exactly as they did pre-detachment. + self.drain_all_ready_family_sequencers().await?; + if self.unadmitted_checkpoints.load(Ordering::Acquire) == 0 + && !self.has_inflight_family_effects() + { break; } - drop(guard); + tokio::time::sleep(Duration::from_millis(25)).await; } self.status_for_family(repo_working_dir).await @@ -6974,7 +7209,11 @@ impl ActorDaemonCoordinator { self.wait_for_trace_ingest_processed_through().await; self.drain_all_ready_family_sequencers().await?; - if self.outstanding_checkpoint_state().0 == 0 { + // Detached side-effect passes (drains and non-sequencer + // commands) are invisible to the sequencer map once their + // entries are popped; exiting while one is in flight would let + // process teardown kill it mid-write (#2252). + if self.outstanding_checkpoint_state().0 == 0 && !self.has_inflight_family_effects() { return Ok(()); } tokio::time::sleep(Duration::from_millis(50)).await; @@ -9128,11 +9367,13 @@ fn daemon_update_check_loop(coordinator: Arc, started_at Ok(DaemonUpdateCheckResult::UpdateReady) => { let (outstanding_checkpoints, retained_checkpoint_bytes) = coordinator.outstanding_checkpoint_state(); - if outstanding_checkpoints > 0 { + // Also defer while queued or executing attribution work + // remains: process teardown would abandon it (#2252). + if coordinator.has_pending_attribution_work() { tracing::info!( outstanding_checkpoints, retained_checkpoint_bytes, - "update restart deferred while accepted checkpoints remain" + "update restart deferred while attribution work remains" ); } else { tracing::info!("update check: newer version available, requesting shutdown"); @@ -9152,11 +9393,13 @@ fn daemon_update_check_loop(coordinator: Arc, started_at if uptime_ns >= daemon_max_uptime_ns() { let (outstanding_checkpoints, retained_checkpoint_bytes) = coordinator.outstanding_checkpoint_state(); - if outstanding_checkpoints > 0 { + // Also defer while queued or executing attribution work + // remains: process teardown would abandon it (#2252). + if coordinator.has_pending_attribution_work() { tracing::info!( outstanding_checkpoints, retained_checkpoint_bytes, - "uptime restart deferred while accepted checkpoints remain" + "uptime restart deferred while attribution work remains" ); } else { tracing::info!("uptime exceeded max, requesting restart"); @@ -10274,6 +10517,54 @@ mod tests { ); } + #[tokio::test] + async fn pending_attribution_work_fence_ignores_idle_pending_roots() { + let coord = ActorDaemonCoordinator::new(); + assert!( + !coord.has_pending_attribution_work(), + "an idle daemon must not defer automatic restarts" + ); + + // An idle interactive command (PendingRoot placeholder, e.g. a rebase + // waiting on an editor) must not defer restarts forever. + coord + .append_pending_root_entry("family-a", "pending-root-1", 10) + .expect("append pending root"); + assert!( + !coord.has_pending_attribution_work(), + "a PendingRoot placeholder alone must not defer restarts" + ); + + // Actionable sequencer entries must defer: a restart would abandon + // the queued attribution pass before its drain registers an effect. + coord + .append_family_sequencer_entry("family-b", 20, FamilySequencerEntry::Canceled) + .expect("append actionable entry"); + assert!( + coord.has_pending_attribution_work(), + "queued sequencer entries must defer restarts" + ); + coord + .family_sequencers_by_family + .lock() + .unwrap() + .remove("family-b"); + assert!(!coord.has_pending_attribution_work()); + + // Executing side-effect passes must defer too. + { + let _effect = coord.begin_family_effect_guarded("family-c"); + assert!( + coord.has_pending_attribution_work(), + "in-flight side-effect passes must defer restarts" + ); + } + assert!( + !coord.has_pending_attribution_work(), + "the effect guard must release the fence on drop" + ); + } + #[tokio::test] async fn draining_an_unknown_family_does_not_retain_empty_sequencer_state() { let coord = ActorDaemonCoordinator::new(); @@ -10779,7 +11070,7 @@ mod tests { #[tokio::test] async fn mutating_pending_root_is_created_when_repo_and_argv_arrive_on_different_events() { - let coord = ActorDaemonCoordinator::new(); + let coord = Arc::new(ActorDaemonCoordinator::new()); let temp = tempfile::tempdir().unwrap(); let repo = temp.path().join("repo"); let init = std::process::Command::new("git") @@ -10914,12 +11205,26 @@ mod tests { coord .record_trace_connection_close(&[sid.to_string()]) .unwrap(); + // The reader-side close keeps a mutating root registered until the + // ingest worker processes its queued frames and close marker, so the + // fence must still hold across the reader/worker gap (#2252). + assert!( + tokio::time::timeout( + Duration::from_millis(50), + coord.wait_for_trace_ingest_processed_through() + ) + .await + .is_err(), + "checkpoint fence must hold until the worker processes the root's close marker" + ); + + coord.clear_trace_root_tracking(sid).unwrap(); tokio::time::timeout( Duration::from_secs(1), coord.wait_for_trace_ingest_processed_through(), ) .await - .expect("checkpoint fence should pass once the mutating trace root closes"); + .expect("checkpoint fence should pass once the worker has processed the root's close"); } #[tokio::test] @@ -10982,12 +11287,26 @@ mod tests { coord .record_trace_connection_close(&[sid.to_string()]) .unwrap(); + // The reader-side close keeps a mutating root registered until the + // ingest worker processes its queued frames and close marker, so the + // fence must still hold across the reader/worker gap (#2252). + assert!( + tokio::time::timeout( + Duration::from_millis(50), + coord.wait_for_trace_ingest_processed_through_family(&own_family) + ) + .await + .is_err(), + "own family fence must hold until the worker processes the root's close marker" + ); + + coord.clear_trace_root_tracking(sid).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"); + .expect("own family fence should pass once the worker has processed the root's close"); } #[tokio::test] @@ -11482,12 +11801,26 @@ mod tests { coord .record_trace_connection_close(&[sid.to_string()]) .unwrap(); + // The reader-side close keeps a mutating root registered until the + // ingest worker processes its queued frames and close marker, so the + // fence must still hold across the reader/worker gap (#2252). + assert!( + tokio::time::timeout( + Duration::from_millis(50), + coord.wait_for_trace_ingest_processed_through() + ) + .await + .is_err(), + "checkpoint fence must hold until the worker processes the root's close marker" + ); + + coord.clear_trace_root_tracking(sid).unwrap(); tokio::time::timeout( Duration::from_secs(1), coord.wait_for_trace_ingest_processed_through(), ) .await - .expect("checkpoint fence should pass once the branch mutation root closes"); + .expect("checkpoint fence should pass once the worker has processed the root's close"); } #[tokio::test] diff --git a/tests/daemon_mode.rs b/tests/daemon_mode.rs index 9a064f0f91..36ddc0bf65 100644 --- a/tests/daemon_mode.rs +++ b/tests/daemon_mode.rs @@ -171,6 +171,29 @@ fn write_trace_frames_to_stream(stream: &mut impl Write, payloads: &[Value]) { stream.flush().expect("failed to flush trace payloads"); } +/// Waits until the repo's dedicated daemon logs that a delayed side-effect +/// pass (GIT_AI_TEST_DELAY_SIDE_EFFECT_MS_FOR_COMMAND) has started executing, +/// so grind tests gate on the actual daemon-side precondition instead of +/// fixed sleeps that can pass vacuously or fail spuriously on loaded runners. +#[cfg(not(windows))] +fn wait_for_test_side_effect_delay_started(repo: &TestRepo) { + let started = std::time::Instant::now(); + loop { + if repo + .daemon_stderr_contents() + .contains("test side-effect delay started") + { + return; + } + assert!( + started.elapsed() < Duration::from_secs(20), + "daemon never entered the delayed side-effect pass; logs:\n{}", + repo.daemon_stderr_contents() + ); + thread::sleep(Duration::from_millis(25)); + } +} + fn repo_workdir_string(repo: &TestRepo) -> String { repo.path().to_string_lossy().to_string() } @@ -2126,6 +2149,308 @@ fn daemon_drains_independent_checkpoint_families_concurrently() { } } +/// Regression test for #2252: a long write-op side-effect pass in one family +/// (a large checkout/commit in a monorepo) must not stall the trace-ingest +/// watermark that checkpoint admission waits on. A checkpoint for an +/// independent repository family must be admitted and processed while the +/// other family's side effects are still grinding. +#[test] +#[cfg(not(windows))] +fn family_side_effect_grind_does_not_stall_independent_checkpoint_admission() { + let grinding_repo = TestRepo::new_with_daemon_env(&[( + "GIT_AI_TEST_DELAY_SIDE_EFFECT_MS_FOR_COMMAND", + "commit=10000", + )]); + let independent_repo = TestRepo::new_with_daemon_scope(DaemonTestScope::NoDaemon); + let control_socket = daemon_control_socket_path(&grinding_repo); + + // A traced commit whose side-effect pass grinds for 10s in its family. + fs::write(grinding_repo.path().join("grind.txt"), "grind\n").unwrap(); + grinding_repo + .git_without_test_sync_for_test(&["add", "grind.txt"], &[]) + .unwrap(); + grinding_repo + .git_without_test_sync_for_test(&["commit", "-m", "grind commit"], &[]) + .unwrap(); + + wait_for_test_side_effect_delay_started(&grinding_repo); + + fs::write( + independent_repo.path().join("independent.txt"), + "independent-grind-checkpoint\n", + ) + .unwrap(); + let checkpoint = CheckpointRequest { + trace_id: "independent-grind-checkpoint".to_string(), + checkpoint_kind: CheckpointKind::AiAgent, + agent_id: Some(AgentId { + tool: "mock_ai".to_string(), + id: "independent-grind-checkpoint".to_string(), + model: "test".to_string(), + }), + files: vec![CheckpointFile { + path: PathBuf::from("independent.txt"), + content: Some("independent-grind-checkpoint\n".to_string()), + repo_work_dir: independent_repo.path().to_path_buf(), + base_commit: BaseCommit::Initial, + }], + path_role: PreparedPathRole::Edited, + stream_source: None, + metadata: Default::default(), + }; + let response = + send_checkpoint_request_with_timeout(&control_socket, &checkpoint, Duration::from_secs(2)) + .expect("checkpoint should be acknowledged during an unrelated side-effect grind"); + assert!(response.ok, "checkpoint failed: {response:?}"); + + // The independent family must finish processing well before the 10s grind + // completes; admission queueing behind the grind is issue #2252. + let started = std::time::Instant::now(); + loop { + let checkpoint_count = find_repository_in_path(independent_repo.path().to_str().unwrap()) + .unwrap() + .storage + .working_log_for_base_commit("initial") + .unwrap() + .read_all_checkpoints() + .map(|checkpoints| checkpoints.len()) + .unwrap_or(0); + if checkpoint_count == 1 { + break; + } + assert!( + started.elapsed() < Duration::from_secs(5), + "independent checkpoint was not admitted/processed while another \ + family's side effects were running (#2252)" + ); + thread::sleep(Duration::from_millis(25)); + } +} + +/// Regression test for #2252's headline symptom: checkpoints "received into +/// bounded ingress" but never "prepared for family admission" while the same +/// family's write-op side effects grind. Admission is a global sequencing +/// stage and must not queue behind side-effect execution; only processing may +/// wait for the family's turn. +#[test] +#[cfg(not(windows))] +fn checkpoint_admission_proceeds_while_same_family_side_effects_grind() { + let repo = TestRepo::new_with_daemon_env(&[( + "GIT_AI_TEST_DELAY_SIDE_EFFECT_MS_FOR_COMMAND", + "commit=10000", + )]); + + fs::write(repo.path().join("grind.txt"), "grind\n").unwrap(); + repo.git_without_test_sync_for_test(&["add", "grind.txt"], &[]) + .unwrap(); + repo.git_without_test_sync_for_test(&["commit", "-m", "grind commit"], &[]) + .unwrap(); + + wait_for_test_side_effect_delay_started(&repo); + + fs::write(repo.path().join("same-family.txt"), "same-family\n").unwrap(); + let checkpoint = CheckpointRequest { + trace_id: "same-family-grind-checkpoint".to_string(), + checkpoint_kind: CheckpointKind::AiAgent, + agent_id: Some(AgentId { + tool: "mock_ai".to_string(), + id: "same-family-grind-checkpoint".to_string(), + model: "test".to_string(), + }), + files: vec![CheckpointFile { + path: PathBuf::from("same-family.txt"), + content: Some("same-family\n".to_string()), + repo_work_dir: repo.path().to_path_buf(), + base_commit: BaseCommit::Initial, + }], + path_role: PreparedPathRole::Edited, + stream_source: None, + metadata: Default::default(), + }; + let response = send_checkpoint_request_with_timeout( + &daemon_control_socket_path(&repo), + &checkpoint, + Duration::from_secs(2), + ) + .expect("checkpoint should be acknowledged during a same-family side-effect grind"); + assert!(response.ok, "checkpoint failed: {response:?}"); + + // Admission (not processing) must complete well before the 10s grind ends. + let started = std::time::Instant::now(); + loop { + let logs = repo.daemon_stderr_contents(); + if logs.lines().any(|line| { + line.contains("checkpoint prepared for family admission") + && line.contains("same-family-grind-checkpoint") + }) { + break; + } + assert!( + started.elapsed() < Duration::from_secs(5), + "checkpoint was received but not admitted while the same family's \ + side effects were running (#2252); daemon logs:\n{logs}" + ); + thread::sleep(Duration::from_millis(25)); + } +} + +/// #2252: side effects for commands that do not participate in the family +/// sequencer (e.g. `git am`) run on detached tasks. `sync.family` must still +/// cover them — it must not report a family as synced while such a pass is +/// in flight. +#[test] +#[cfg(not(windows))] +fn sync_family_waits_for_detached_non_sequencer_side_effects() { + let repo = TestRepo::new_with_daemon_env(&[( + "GIT_AI_TEST_DELAY_SIDE_EFFECT_MS_FOR_COMMAND", + "am=3000", + )]); + + // Base commit so `git am` has a parent to apply onto. + fs::write(repo.path().join("base.txt"), "base\n").unwrap(); + repo.git_without_test_sync_for_test(&["add", "base.txt"], &[]) + .unwrap(); + repo.git_without_test_sync_for_test(&["commit", "-m", "base"], &[]) + .unwrap(); + + let patch = "\ +From 1111111111111111111111111111111111111111 Mon Sep 17 00:00:00 2001 +From: Repro +Date: Mon, 1 Sep 2026 00:00:00 +0000 +Subject: [PATCH] add am file + +--- + am.txt | 1 + + 1 file changed, 1 insertion(+) + create mode 100644 am.txt + +diff --git a/am.txt b/am.txt +new file mode 100644 +index 0000000..7898192 +--- /dev/null ++++ b/am.txt +@@ -0,0 +1 @@ ++a +--\x20 +2.39.0 +"; + fs::write(repo.path().join("am-patch.mbox"), patch).unwrap(); + repo.git_without_test_sync_for_test(&["am", "am-patch.mbox"], &[]) + .unwrap(); + wait_for_test_side_effect_delay_started(&repo); + + let sync = send_control_request_with_timeout( + &daemon_control_socket_path(&repo), + &ControlRequest::SyncFamily { + repo_working_dir: repo_workdir_string(&repo), + }, + Duration::from_secs(30), + ) + .expect("sync.family should succeed"); + assert!(sync.ok, "sync.family failed: {sync:?}"); + + let am_entries = completion_entries_for_command(&repo, "am"); + assert!( + !am_entries.is_empty(), + "sync.family returned before the detached `git am` side-effect pass completed (#2252)" + ); +} + +/// #2252: with side-effect passes running on detached tasks, graceful +/// shutdown must not complete while such a pass is still in flight — the +/// process exit right after the shutdown response would kill it mid-write. +#[test] +#[cfg(not(windows))] +fn graceful_shutdown_waits_for_detached_side_effect_passes() { + let repo = TestRepo::new_with_daemon_env(&[( + "GIT_AI_TEST_DELAY_SIDE_EFFECT_MS_FOR_COMMAND", + "commit=3000", + )]); + + fs::write(repo.path().join("grind.txt"), "grind\n").unwrap(); + repo.git_without_test_sync_for_test(&["add", "grind.txt"], &[]) + .unwrap(); + repo.git_without_test_sync_for_test(&["commit", "-m", "grind commit"], &[]) + .unwrap(); + wait_for_test_side_effect_delay_started(&repo); + + let shutdown = send_control_request_with_timeout( + &daemon_control_socket_path(&repo), + &ControlRequest::Shutdown, + Duration::from_secs(30), + ) + .expect("graceful shutdown request should succeed"); + assert!(shutdown.ok, "graceful shutdown failed: {shutdown:?}"); + + let commit_entries = completion_entries_for_command(&repo, "commit"); + assert!( + !commit_entries.is_empty(), + "graceful shutdown completed while the commit side-effect pass was still running (#2252)" + ); +} + +/// #2252: a side-effect pass can write into a DIFFERENT repository — e.g. +/// `git push ` pushes authorship notes into the destination repo. +/// Before side effects ran detached, the global ingest watermark implied +/// they had completed, so `sync.family` on the destination covered them +/// transitively. That guarantee must survive detachment: sync.family must +/// not report until already-ingested side-effect work has finished, in any +/// family. +#[test] +#[cfg(not(windows))] +fn sync_family_covers_cross_repo_side_effects_of_other_families() { + let local = TestRepo::new_with_daemon_env(&[( + "GIT_AI_TEST_DELAY_SIDE_EFFECT_MS_FOR_COMMAND", + "push=3000", + )]); + let destination = TestRepo::new_bare_with_daemon_scope(DaemonTestScope::NoDaemon); + + fs::write(local.path().join("pushed.txt"), "pushed\n").unwrap(); + local.git_og(&["add", "pushed.txt"]).unwrap(); + local.git_og(&["commit", "-m", "pushed commit"]).unwrap(); + let commit_sha = local + .git_og(&["rev-parse", "HEAD"]) + .unwrap() + .trim() + .to_string(); + local + .git_og(&[ + "notes", + "--ref=ai", + "add", + "-m", + "cross-repo-note", + commit_sha.as_str(), + ]) + .unwrap(); + + let destination_path = destination.path().to_string_lossy().to_string(); + local + .git_without_test_sync_for_test( + &["push", destination_path.as_str(), "HEAD:refs/heads/pushed"], + &[], + ) + .unwrap(); + wait_for_test_side_effect_delay_started(&local); + + let sync = send_control_request_with_timeout( + &daemon_control_socket_path(&local), + &ControlRequest::SyncFamily { + repo_working_dir: destination_path.clone(), + }, + Duration::from_secs(30), + ) + .expect("sync.family for the destination should succeed"); + assert!(sync.ok, "sync.family failed: {sync:?}"); + + let pushed_note = destination.git_og(&["notes", "--ref=ai", "show", commit_sha.as_str()]); + assert!( + pushed_note.is_ok_and(|note| note.contains("cross-repo-note")), + "sync.family for the destination returned before the push side-effect \ + pass finished pushing authorship notes into it (#2252)" + ); +} + #[test] #[cfg(not(windows))] fn daemon_checkpoint_processing_failure_is_logged_after_receipt_ack() {