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