diff --git a/crates/tracedecay-domain/src/code_intelligence/graph.rs b/crates/tracedecay-domain/src/code_intelligence/graph.rs index d75c556f0..e2c4c4530 100644 --- a/crates/tracedecay-domain/src/code_intelligence/graph.rs +++ b/crates/tracedecay-domain/src/code_intelligence/graph.rs @@ -896,11 +896,19 @@ pub struct CodeBlock { /// /// The ID format is `"kind:32hexchars"` where the hex portion is the first 32 /// characters of the SHA-256 hash of the input components. +/// +/// `name` is extractor output derived from arbitrary user source, not a +/// programmer-supplied invariant, so an empty name is accepted input rather +/// than a contract violation: legal constructs such as `test.describe("", …)`, +/// `export default () => {}`, and IIFEs are genuinely unnamed. Identity does +/// not depend on the name being non-empty — file path, kind, and line already +/// separate two anonymous constructs — so an empty name hashes like any other +/// and behaves identically in debug and release builds. A `debug_assert!` here +/// previously panicked the indexing pool on such files in debug/perf profiles +/// while release silently produced this same id; callers that want a readable +/// display name must synthesize one (see the `` convention used +/// across the extractors) instead of relying on an assertion here. pub fn generate_node_id(file_path: &str, kind: &NodeKind, name: &str, line: u32) -> String { - debug_assert!( - !name.is_empty(), - "generate_node_id called with empty name for {file_path}:{line}" - ); let input = format!("{}:{}:{}:{}", file_path, kind.as_str(), name, line); let mut hasher = Sha256::new(); hasher.update(input.as_bytes()); @@ -924,3 +932,43 @@ pub struct ResolvedRef { pub confidence: f64, pub resolved_by: String, } + +#[cfg(test)] +mod empty_name_node_id_tests { + use super::{NodeKind, generate_node_id}; + + /// Real source contains anonymous constructs (`test.describe("", …)`, + /// `export default () => {}`). Extracted names are parser output, not a + /// programmer contract, so an empty name must produce an id in every + /// build profile instead of tripping a debug-only assertion. + #[test] + fn empty_name_yields_a_deterministic_id_in_every_profile() { + let first = generate_node_id( + "integration/fs-routes-test.ts", + &NodeKind::Function, + "", + 286, + ); + let second = generate_node_id( + "integration/fs-routes-test.ts", + &NodeKind::Function, + "", + 286, + ); + assert_eq!(first, second, "empty-name ids must be deterministic"); + assert!( + first.starts_with("function:"), + "unexpected id shape: {first}" + ); + } + + /// Uniqueness of an empty-name id comes from file path, kind, and line, + /// so distinct anonymous constructs never collide onto one node. + #[test] + fn empty_name_ids_stay_distinct_per_file_kind_and_line() { + let base = generate_node_id("a.ts", &NodeKind::Function, "", 286); + assert_ne!(base, generate_node_id("b.ts", &NodeKind::Function, "", 286)); + assert_ne!(base, generate_node_id("a.ts", &NodeKind::Class, "", 286)); + assert_ne!(base, generate_node_id("a.ts", &NodeKind::Function, "", 287)); + } +} diff --git a/src/daemon/code_index_scheduler.rs b/src/daemon/code_index_scheduler.rs index 97158bf3a..765497e86 100644 --- a/src/daemon/code_index_scheduler.rs +++ b/src/daemon/code_index_scheduler.rs @@ -2558,6 +2558,36 @@ impl CodeIndexSchedulerErrorV1 { } } + /// A refusal that is transient *by construction*: this pass was turned away + /// because a bounded shared resource was already fully held, and it is + /// released by whoever holds it rather than by anything about this input. + /// + /// The background worker schedules its own delayed retry for exactly these, + /// because releasing shared capacity emits no wake: a sibling worktree or + /// artifact build finishing does not notify this worktree, so without a + /// self-scheduled retry it stayed stale until an unrelated query or edit + /// happened to wake it. + /// + /// The distinction the admission failure carries is the whole point. A + /// request that exceeds the *entire* process limit is shaped like a + /// capacity refusal and is not one — no other holder can release enough for + /// it — so it is classified permanent and never self-retried. Identity + /// failures, git and IO faults, production and privacy refusals, adjustment + /// invariant breaks, an uninstalled worker plan, and publication conflicts + /// likewise reproduce over the same input or already have an owner that + /// re-drives them; self-scheduling those is precisely the unbounded-retry + /// failure this module exists to stop. + pub(super) fn is_transient_capacity_failure(&self) -> bool { + match self { + Self::WorkerMemoryAdmission(failure) | Self::SnapshotMemoryAdmission(failure) => { + failure.requested_bytes <= failure.limit_bytes + } + Self::SnapshotMemoryCapacityUnavailable => true, + Self::GraphProjection(CodeGraphProjectionError::BudgetExhausted { .. }) => true, + _ => false, + } + } + pub(super) fn is_graph_activation_refusal(&self) -> bool { matches!(self, Self::GraphActivationRefused(_)) || matches!( @@ -2628,6 +2658,10 @@ pub(super) struct CodeIndexWorktreeSchedulerV1 { /// `retained_snapshot_bytes`; worker scratch is admitted separately only /// after capture has completed. _retained_snapshot_memory: Vec, + /// Deterministic reconcile fault used only by the worker-loop isolation + /// tests; production never installs one. + #[cfg(test)] + reconcile_fault: Option>, /// Process resident-memory authority artifact builds and readers reserve /// through. Standalone opens get a private default-limit authority; the /// registry rebinds its shared process authority at mount. @@ -2734,6 +2768,8 @@ impl CodeIndexWorktreeSchedulerV1 { byte_pool, retained_snapshot_bytes: Vec::new(), _retained_snapshot_memory: Vec::new(), + #[cfg(test)] + reconcile_fault: None, resident_memory: Arc::new(ProcessResidentMemoryV1::new( DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, )), @@ -3056,10 +3092,25 @@ impl CodeIndexWorktreeSchedulerV1 { Some(self.bind_latest_complete(generation)) } + /// Install a deterministic reconcile fault for one mounted worktree so a + /// test can drive the real background worker loop over a pass that panics + /// or fails, and count the attempts the loop actually makes. + #[cfg(test)] + pub(in crate::daemon::code_index_scheduler) fn install_reconcile_fault_for_test( + &mut self, + fault: Arc, + ) { + self.reconcile_fault = Some(fault); + } + /// Retained-owner activation entry point. Foreground reads never call this. pub(super) fn activate_or_reconcile( &mut self, ) -> Result { + #[cfg(test)] + if let Some(fault) = self.reconcile_fault.clone() { + fault.arrive()?; + } // The in-progress signal must cover the retained-activation branch // too: the worker has already claimed the pending wake, so without it // a failing activation pass would leave query admission unable to see @@ -4114,6 +4165,7 @@ pub(in crate::daemon) mod observability; mod privacy; pub(in crate::daemon) mod queries; pub(in crate::daemon) mod query_runtime; +mod reconcile_panic_guard; mod registry; pub(crate) mod semantic_query_runtime; pub(crate) mod semantic_vector_graph; diff --git a/src/daemon/code_index_scheduler/reconcile_panic_guard.rs b/src/daemon/code_index_scheduler/reconcile_panic_guard.rs new file mode 100644 index 000000000..902dcc6c6 --- /dev/null +++ b/src/daemon/code_index_scheduler/reconcile_panic_guard.rs @@ -0,0 +1,419 @@ +//! Bounded retry policy for a background reconcile pass whose blocking task +//! panicked. +//! +//! A reconcile fans per-file work across the indexing pool over arbitrary user +//! source. A panic there aborts the pass and surfaces as an opaque `JoinError` +//! on the worker loop. The loop then restored the pending arrival and waited +//! for the next wake — and because the offending input is still on disk, the +//! next wake reproduced the identical panic. One malformed file therefore +//! blocked a project's entire code index indefinitely, retrying forever with +//! no backoff and no terminal state. +//! +//! This policy makes that failure degrade instead of loop: repeated panics +//! back off exponentially, and after a bounded number of consecutive panics +//! the worker stops re-attempting until the input actually changes (the +//! code-index control epoch advances) or a pass makes progress. The shape +//! mirrors the sealed-generation activation backoff already used by the +//! registry worker; tests shrink the clock, not the shape. + +use std::time::Duration; + +use tokio::time::Instant; + +/// Bounded exponential backoff between reconcile retries after a panicking +/// pass. The floor keeps a transient panic from hot-looping the pool; the +/// ceiling keeps a persistently panicking input from being retried more than a +/// few times an hour. +pub(super) const RECONCILE_PANIC_BACKOFF_FLOOR: Duration = if cfg!(test) { + Duration::from_millis(50) +} else { + Duration::from_secs(30) +}; +pub(super) const RECONCILE_PANIC_BACKOFF_CEILING: Duration = if cfg!(test) { + Duration::from_millis(400) +} else { + Duration::from_mins(10) +}; + +/// Consecutive panics over unchanged input after which retrying is pointless: +/// the same bytes reproduce the same panic. Further attempts are suppressed +/// until the input changes. +pub(super) const MAX_CONSECUTIVE_RECONCILE_PANICS_V1: u32 = 4; + +/// What the worker loop should do after a reconcile pass panicked. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum ReconcilePanicDecisionV1 { + /// Re-arm a wake after this delay and try the same input again. + RetryAfter(Duration), + /// The bound is exhausted. Stop re-arming; only changed input or a + /// progressing pass resumes reconciles. + Quarantine, +} + +/// Consecutive-panic accounting for one mounted worktree's worker loop. +#[derive(Debug)] +pub(super) struct ReconcilePanicGuardV1 { + consecutive_panics: u32, + backoff: Duration, + next_attempt_at: Option, + quarantined: bool, + /// Control epoch observed when the guard last quarantined. An advance + /// means new input, which is worth one more attempt. + quarantined_at_epoch: u64, +} + +impl Default for ReconcilePanicGuardV1 { + fn default() -> Self { + Self::new() + } +} + +impl ReconcilePanicGuardV1 { + pub(super) const fn new() -> Self { + Self { + consecutive_panics: 0, + backoff: RECONCILE_PANIC_BACKOFF_FLOOR, + next_attempt_at: None, + quarantined: false, + quarantined_at_epoch: 0, + } + } + + /// Any pass that completed without panicking clears the accounting: the + /// next panic starts again at the floor. + pub(super) fn record_progress(&mut self) { + self.consecutive_panics = 0; + self.backoff = RECONCILE_PANIC_BACKOFF_FLOOR; + self.next_attempt_at = None; + self.quarantined = false; + } + + /// Record a panicking pass and decide whether the same input is worth + /// another attempt. + pub(super) fn record_panic(&mut self, now: Instant, epoch: u64) -> ReconcilePanicDecisionV1 { + self.consecutive_panics = self.consecutive_panics.saturating_add(1); + if self.consecutive_panics >= MAX_CONSECUTIVE_RECONCILE_PANICS_V1 { + self.quarantined = true; + self.quarantined_at_epoch = epoch; + self.next_attempt_at = None; + return ReconcilePanicDecisionV1::Quarantine; + } + let delay = self.backoff; + self.next_attempt_at = Some(now + delay); + self.backoff = self + .backoff + .saturating_mul(2) + .min(RECONCILE_PANIC_BACKOFF_CEILING); + ReconcilePanicDecisionV1::RetryAfter(delay) + } + + /// Consecutive panics observed since the last progressing pass. Reported + /// on the warn path so an operator sees a bounded counter rather than an + /// undifferentiated repeating line. + pub(super) const fn consecutive_panics(&self) -> u32 { + self.consecutive_panics + } + + /// True while this wake must be skipped: either the backoff window is + /// still open, or the guard is quarantined and the input has not changed. + pub(super) fn suppresses_pass(&mut self, now: Instant, epoch: u64) -> bool { + if self.quarantined { + if epoch != self.quarantined_at_epoch { + // New input is not the input that panicked; allow one attempt + // and let it re-quarantine if it panics again. + self.quarantined = false; + self.consecutive_panics = 0; + self.backoff = RECONCILE_PANIC_BACKOFF_FLOOR; + self.next_attempt_at = None; + return false; + } + return true; + } + self.next_attempt_at.is_some_and(|at| now < at) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn origin() -> Instant { + Instant::now() + } + + #[test] + fn repeated_panics_back_off_and_then_quarantine() { + let mut guard = ReconcilePanicGuardV1::new(); + let now = origin(); + + let mut delays = Vec::new(); + let mut decisions = Vec::new(); + for _ in 0..MAX_CONSECUTIVE_RECONCILE_PANICS_V1 { + let decision = guard.record_panic(now, 7); + decisions.push(decision); + if let ReconcilePanicDecisionV1::RetryAfter(delay) = decision { + delays.push(delay); + } + } + + assert_eq!( + decisions.last(), + Some(&ReconcilePanicDecisionV1::Quarantine), + "an unchanged panicking input must reach a terminal state, not retry forever" + ); + assert_eq!( + delays.len() as u32, + MAX_CONSECUTIVE_RECONCILE_PANICS_V1 - 1, + "every attempt before the bound is a delayed retry" + ); + assert!( + delays.windows(2).all(|pair| pair[1] > pair[0]), + "retries must back off, not repeat identically: {delays:?}" + ); + assert!( + delays + .iter() + .all(|delay| *delay >= RECONCILE_PANIC_BACKOFF_FLOOR + && *delay <= RECONCILE_PANIC_BACKOFF_CEILING), + "delays stay inside the bounded window: {delays:?}" + ); + } + + #[test] + fn a_quarantined_guard_suppresses_further_passes_over_unchanged_input() { + let mut guard = ReconcilePanicGuardV1::new(); + let now = origin(); + for _ in 0..MAX_CONSECUTIVE_RECONCILE_PANICS_V1 { + guard.record_panic(now, 7); + } + + let far_future = now + Duration::from_secs(86_400); + assert!( + guard.suppresses_pass(far_future, 7), + "unchanged input must stay quarantined however long the wait" + ); + } + + #[test] + fn changed_input_lifts_the_quarantine() { + let mut guard = ReconcilePanicGuardV1::new(); + let now = origin(); + for _ in 0..MAX_CONSECUTIVE_RECONCILE_PANICS_V1 { + guard.record_panic(now, 7); + } + assert!(guard.suppresses_pass(now, 7)); + + assert!( + !guard.suppresses_pass(now, 8), + "an advanced control epoch is new input and earns another attempt" + ); + assert_eq!(guard.consecutive_panics(), 0, "accounting restarts"); + } + + #[test] + fn the_backoff_window_suppresses_only_until_it_elapses() { + let mut guard = ReconcilePanicGuardV1::new(); + let now = origin(); + let ReconcilePanicDecisionV1::RetryAfter(delay) = guard.record_panic(now, 7) else { + panic!("the first panic must schedule a retry"); + }; + + assert!(guard.suppresses_pass(now, 7), "window is open"); + assert!( + !guard.suppresses_pass(now + delay, 7), + "the pass runs once the window elapses" + ); + } + + #[test] + fn a_progressing_pass_clears_the_accounting() { + let mut guard = ReconcilePanicGuardV1::new(); + let now = origin(); + for _ in 0..MAX_CONSECUTIVE_RECONCILE_PANICS_V1 { + guard.record_panic(now, 7); + } + assert!(guard.suppresses_pass(now, 7)); + + guard.record_progress(); + + assert!(!guard.suppresses_pass(now, 7), "progress lifts suppression"); + assert_eq!(guard.consecutive_panics(), 0); + assert_eq!( + guard.record_panic(now, 7), + ReconcilePanicDecisionV1::RetryAfter(RECONCILE_PANIC_BACKOFF_FLOOR), + "the next panic restarts at the floor" + ); + } +} + +/// Bounded delayed retry for a reconcile pass that failed because shared +/// process capacity was momentarily exhausted. +/// +/// A reconcile reserves against one process-wide resident-memory authority. A +/// sibling worktree or artifact build can hold that budget when this pass asks +/// for it, and the request is refused before any indexing work starts. Nothing +/// wakes this worker when the competing holder releases: the failure path only +/// restored the pending arrival, so the worktree stayed stale until an +/// unrelated query or edit happened to wake it. +/// +/// The retry is deliberately narrow. A panicking or permanently refused pass +/// reproduces on every attempt, so retrying it forever is the failure this +/// module exists to prevent; only a failure that is *transient by construction* +/// — capacity another holder will release — earns a re-arm, and even that is +/// capped so a genuinely undersized budget degrades to stale instead of +/// spinning. +pub(super) const RECONCILE_CAPACITY_RETRY_FLOOR: Duration = if cfg!(test) { + Duration::from_millis(40) +} else { + Duration::from_secs(2) +}; +pub(super) const RECONCILE_CAPACITY_RETRY_CEILING: Duration = if cfg!(test) { + Duration::from_millis(320) +} else { + Duration::from_secs(60) +}; + +/// Consecutive capacity refusals after which the shared budget is not +/// momentarily contended but structurally too small for this worktree. Further +/// self-scheduled wakes only burn the pool; the next real hint still retries. +pub(super) const MAX_CONSECUTIVE_CAPACITY_RETRIES_V1: u32 = 5; + +/// Consecutive-capacity-refusal accounting for one mounted worktree's worker. +#[derive(Debug)] +pub(super) struct ReconcileCapacityRetryV1 { + consecutive: u32, + backoff: Duration, +} + +impl Default for ReconcileCapacityRetryV1 { + fn default() -> Self { + Self::new() + } +} + +impl ReconcileCapacityRetryV1 { + pub(super) const fn new() -> Self { + Self { + consecutive: 0, + backoff: RECONCILE_CAPACITY_RETRY_FLOOR, + } + } + + /// Any pass that did not fail on capacity clears the accounting. + pub(super) fn record_progress(&mut self) { + self.consecutive = 0; + self.backoff = RECONCILE_CAPACITY_RETRY_FLOOR; + } + + /// `Some(delay)` arms exactly one delayed wake; `None` means the bound is + /// spent and this worker stops self-scheduling until real input arrives. + pub(super) fn record_capacity_failure(&mut self) -> Option { + if self.consecutive >= MAX_CONSECUTIVE_CAPACITY_RETRIES_V1 { + return None; + } + self.consecutive = self.consecutive.saturating_add(1); + let delay = self.backoff; + self.backoff = self + .backoff + .saturating_mul(2) + .min(RECONCILE_CAPACITY_RETRY_CEILING); + Some(delay) + } + + /// Consecutive capacity refusals since the last non-capacity pass. + pub(super) const fn consecutive(&self) -> u32 { + self.consecutive + } +} + +/// Deterministic reconcile fault installed by the worker-loop isolation tests. +/// +/// The guard types above are pure state machines; on their own they prove +/// nothing about whether the background worker consults them. These tests +/// therefore drive the real registry worker over a real mounted worktree and +/// count the passes it actually attempts, which is only observable if the +/// scheduler can be made to fail on demand. +#[cfg(test)] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum ReconcileFaultKindV1 { + /// Unwinds inside the blocking reconcile task, exactly as a malformed + /// source file did through `generate_node_id`. + Panic, + /// Refused before any indexing work because a sibling worktree or artifact + /// build was holding the shared resident-memory budget. The request fits + /// the limit, so releasing that budget makes it admissible. + TransientCapacity, + /// Refused because the request is larger than the whole process limit. It + /// is shaped like a capacity refusal and is not one: no release by any + /// other holder can ever admit it. + OversizedCapacity, + /// A refusal the same input reproduces forever. + Permanent, +} + +#[cfg(test)] +#[derive(Debug)] +pub(super) struct ReconcileFaultInjectionV1 { + kind: ReconcileFaultKindV1, + /// Passes to fault before behaving normally; `usize::MAX` never recovers. + faulting_passes: usize, + attempts: std::sync::atomic::AtomicUsize, +} + +#[cfg(test)] +impl ReconcileFaultInjectionV1 { + pub(super) const fn new(kind: ReconcileFaultKindV1, faulting_passes: usize) -> Self { + Self { + kind, + faulting_passes, + attempts: std::sync::atomic::AtomicUsize::new(0), + } + } + + /// Reconcile passes the worker actually dispatched, faulting or not. + pub(super) fn attempts(&self) -> usize { + self.attempts + .load(std::sync::atomic::Ordering::Acquire) + } + + /// Called at the top of every real reconcile pass. + pub(super) fn arrive(&self) -> Result<(), super::CodeIndexSchedulerErrorV1> { + let seen = self + .attempts + .fetch_add(1, std::sync::atomic::Ordering::AcqRel); + if seen >= self.faulting_passes { + return Ok(()); + } + match self.kind { + ReconcileFaultKindV1::Panic => { + panic!("injected reconcile panic (pass {})", seen + 1) + } + // Exactly the shape `ProcessResidentMemoryV1::reserve` returns when + // the process budget is already spoken for by another holder. + ReconcileFaultKindV1::TransientCapacity => { + Err(super::CodeIndexSchedulerErrorV1::WorkerMemoryAdmission( + tracedecay_runtime_core::resident_memory::ResidentMemoryAdmissionFailureV1 { + used_bytes: 900, + requested_bytes: 200, + limit_bytes: 1_000, + }, + )) + } + // Same variant, but the request alone exceeds the whole limit. + ReconcileFaultKindV1::OversizedCapacity => { + Err(super::CodeIndexSchedulerErrorV1::WorkerMemoryAdmission( + tracedecay_runtime_core::resident_memory::ResidentMemoryAdmissionFailureV1 { + used_bytes: 0, + requested_bytes: 4_000, + limit_bytes: 1_000, + }, + )) + } + ReconcileFaultKindV1::Permanent => Err( + super::CodeIndexSchedulerErrorV1::Identity( + "injected permanent reconcile refusal".to_owned(), + ), + ), + } + } +} diff --git a/src/daemon/code_index_scheduler/registry.rs b/src/daemon/code_index_scheduler/registry.rs index f7f41aadd..2506edf46 100644 --- a/src/daemon/code_index_scheduler/registry.rs +++ b/src/daemon/code_index_scheduler/registry.rs @@ -26,6 +26,9 @@ use tracedecay_domain::{CodeGenerationId, ManifestDigest, ProjectId, RepositoryI use tracedecay_lsp::LspRuntimeFailure; use super::graph_activation::{CodeGraphActivationAuthorityV1, CodeGraphActivationPolicyV1}; +use super::reconcile_panic_guard::{ + ReconcileCapacityRetryV1, ReconcilePanicDecisionV1, ReconcilePanicGuardV1, +}; use super::{ CodeIndexArrivalV1, CodeIndexCadenceOutcomeV1, CodeIndexCadenceTelemetryV1, CodeIndexCadenceTriggerV1, CodeIndexEventToReadyReceiptV1, CodeIndexNoopEvidenceV1, @@ -42,6 +45,8 @@ mod cold_read_wake_tests; mod ignored_dependencies; mod lsp_projection; #[cfg(test)] +mod reconcile_failure_isolation_tests; +#[cfg(test)] mod runtime_generation_census_tests; mod scope_identity; @@ -2209,6 +2214,11 @@ impl CodeIndexSchedulerRegistryV1 { let worker_serving_generation = Arc::clone(&serving_generation); let worker_serving_generation_epoch = Arc::clone(&serving_generation_epoch); let worker_wake = Arc::clone(&wake); + // The code-index control epoch. It advances exactly when new input is + // announced (hook hints, watch paths, overflow), so it is the signal a + // quarantined worker uses to decide that the bytes which panicked it + // are no longer the bytes it is being asked to index. + let worker_control_epoch = Arc::clone(&epoch); let worker_pending_wake = Arc::clone(&pending_wake); let worker_cadence_telemetry = Arc::clone(&self.cadence_telemetry); let worker_shutting_down = Arc::clone(&shutting_down); @@ -2289,11 +2299,37 @@ impl CodeIndexSchedulerRegistryV1 { // rebuild+reseal of an equivalent generation. let mut seat_retry_backoff = ACTIVATION_RETRY_BACKOFF_FLOOR; let mut next_seat_attempt_at: Option = None; + // Bounded retry state for a reconcile whose blocking task unwound. + // Arbitrary user source runs through the indexing pool, so a panic + // there is an input fault, not a programmer-contract break: it + // reproduces on every pass over the same bytes. Without this the + // loop re-dispatched the identical unit on every wake forever. + let mut panic_guard = ReconcilePanicGuardV1::new(); + // Bounded retry state for a reconcile refused because shared + // process capacity was momentarily held by a sibling worktree or + // artifact build. Releasing that capacity emits no wake, so this + // worker must schedule its own. + let mut capacity_retry = ReconcileCapacityRetryV1::new(); loop { worker_wake.notified().await; if worker_shutting_down.load(Ordering::Acquire) { return; } + // A quarantined or backing-off panic unit must not consume the + // pending arrival: the wake stays outstanding so a later + // eligible pass still measures its full queue wait. + if panic_guard.suppresses_pass( + tokio::time::Instant::now(), + worker_control_epoch.load(Ordering::Acquire), + ) { + tracing::debug!( + event = "code_index_reconcile_panic_suppressed", + path = "background_worker", + consecutive_panics = panic_guard.consecutive_panics(), + "code-index reconcile is suppressed after repeated panics over unchanged input" + ); + continue; + } let _semantic_evaluation_publication = worker_semantic_evaluation_publication_gate.lock().await; let Ok(_background_reconcile_admission) = @@ -2600,6 +2636,11 @@ impl CodeIndexSchedulerRegistryV1 { } } if let Ok((Ok(outcome), _, _)) = &result { + // A pass that ran to a terminal outcome proves neither the + // panicking input nor the capacity contention is still + // reproducing, so both bounded retry states restart. + panic_guard.record_progress(); + capacity_retry.record_progress(); if let CodeIndexReconcileOutcomeV1::Published(evidence) = outcome { Self::publish_generation( &worker_generation_publications, @@ -2636,12 +2677,81 @@ impl CodeIndexSchedulerRegistryV1 { } else { // Surface bounded non-terminal failure without new project-path data. match &result { - Ok((Err(error), _, _)) => tracing::warn!( - event = "code_index_reconcile_failed", - path = "background_worker", - error = %error, - "code-index background reconcile failed; the served generation stays stale" - ), + Ok((Err(error), _, _)) => { + // The pass completed; whatever refused it was not an + // unwind, so panic accounting restarts. + panic_guard.record_progress(); + let transient_capacity = error.is_transient_capacity_failure(); + tracing::warn!( + event = "code_index_reconcile_failed", + path = "background_worker", + transient_capacity, + error = %error, + "code-index background reconcile failed; the served generation stays stale" + ); + if transient_capacity { + // Shared process capacity was held by another + // holder when this pass asked for it. Releasing + // it emits no wake, so without a self-scheduled + // retry this worktree stayed stale until some + // unrelated query or edit happened to wake it. + // Permanent refusals deliberately never reach + // here: retrying those forever is the failure + // this loop already had. + match capacity_retry.record_capacity_failure() { + Some(delay) => { + let retry_wake = Arc::clone(&worker_wake); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + retry_wake.notify_one(); + }); + } + None => tracing::warn!( + event = "code_index_reconcile_capacity_retry_exhausted", + path = "background_worker", + consecutive = capacity_retry.consecutive(), + "code-index reconcile stopped retrying a capacity refusal; the next hint retries" + ), + } + } else { + capacity_retry.record_progress(); + } + } + Err(error) if error.is_panic() => { + // Arbitrary user source runs through the indexing + // pool, so an unwind here is malformed input that + // reproduces byte-for-byte on every later pass. + // Bound it instead of re-dispatching the identical + // unit on every wake. + capacity_retry.record_progress(); + let decision = panic_guard.record_panic( + tokio::time::Instant::now(), + worker_control_epoch.load(Ordering::Acquire), + ); + match decision { + ReconcilePanicDecisionV1::RetryAfter(delay) => { + tracing::warn!( + event = "code_index_reconcile_panicked", + path = "background_worker", + consecutive_panics = panic_guard.consecutive_panics(), + error = %error, + "code-index background reconcile panicked; retrying the same input with backoff" + ); + let retry_wake = Arc::clone(&worker_wake); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + retry_wake.notify_one(); + }); + } + ReconcilePanicDecisionV1::Quarantine => tracing::warn!( + event = "code_index_reconcile_quarantined", + path = "background_worker", + consecutive_panics = panic_guard.consecutive_panics(), + error = %error, + "code-index background reconcile is quarantined after repeated panics; changed input or a progressing pass resumes it" + ), + } + } Err(error) => tracing::warn!( event = "code_index_reconcile_failed", path = "background_worker", @@ -2656,7 +2766,6 @@ impl CodeIndexSchedulerRegistryV1 { if worker_shutting_down.load(Ordering::Acquire) { return; } - // The next coalesced hint wakes this worker after a contained panic. let _ = result; } }; diff --git a/src/daemon/code_index_scheduler/registry/reconcile_failure_isolation_tests.rs b/src/daemon/code_index_scheduler/registry/reconcile_failure_isolation_tests.rs new file mode 100644 index 000000000..715c61ba1 --- /dev/null +++ b/src/daemon/code_index_scheduler/registry/reconcile_failure_isolation_tests.rs @@ -0,0 +1,371 @@ +//! The background reconcile worker's failure isolation, asserted through the +//! real worker loop rather than against the policy types in isolation. +//! +//! Both defects these tests pin were only observable *at the loop*: a policy +//! object can behave perfectly while nothing consults it. Every assertion here +//! counts reconcile passes the worker actually dispatched — never elapsed time, +//! which swings run to run on a shared machine. + +use std::fs; +use std::path::Path; +use std::process::Command; +use std::sync::Arc; +use std::time::Duration; + +use tempfile::TempDir; + +use super::super::reconcile_panic_guard::{ + MAX_CONSECUTIVE_CAPACITY_RETRIES_V1, MAX_CONSECUTIVE_RECONCILE_PANICS_V1, + ReconcileFaultInjectionV1, ReconcileFaultKindV1, +}; +use super::CodeIndexSchedulerRegistryV1; + +/// Wake rounds driven from outside the worker. Each stands for the ordinary +/// wake traffic a live daemon produces (cadence ticks, queries, sibling +/// activity) over input that has not changed. +const EXTERNAL_WAKE_ROUNDS: usize = 12; + +/// Spacing between external wakes. `Notify::notify_one` stores a single permit, +/// so back-to-back notifies would collapse into one pass and understate the +/// unbounded-retry behaviour these tests are meant to catch. +const WAKE_ROUND_SPACING: Duration = Duration::from_millis(120); + +/// Ceiling on how long a test waits for the worker to settle. Nothing is +/// asserted about this number; it only stops a hung worker from hanging CI. +const SETTLE_DEADLINE: Duration = Duration::from_secs(20); + +/// Idle window that means "no pass is pending" at mount, before any policy is +/// self-scheduling anything. +const MOUNT_QUIET_WINDOW: Duration = Duration::from_millis(500); + +/// Idle window that means "the worker has stopped self-scheduling". An order of +/// magnitude above the policy's own retry ceiling so a retry still queued +/// behind a loaded machine is never mistaken for termination. +const TERMINATION_QUIET_WINDOW: Duration = Duration::from_secs(3); + +struct Fixture { + _root: TempDir, + project: std::path::PathBuf, + registry: CodeIndexSchedulerRegistryV1, +} + +impl Fixture { + async fn mount(project_id: &str) -> Self { + let root = TempDir::new().expect("fixture root"); + let project = root.path().join("project"); + fs::create_dir_all(project.join("src")).expect("create source root"); + fs::write(project.join("src/main.rs"), "fn main() {}\n").expect("write source"); + run_git_in(&project, &["init", "-q", "-b", "main"]); + run_git_in(&project, &["add", "."]); + run_git_in(&project, &["commit", "-qm", "fixture"]); + + let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); + registry + .mount_worktree( + tracedecay_domain::ProjectId::new(project_id).expect("project identity"), + &project, + root.path().join("store"), + None, + ) + .await + .expect("mount scheduler"); + + let fixture = Self { + _root: root, + project, + registry, + }; + // Mount itself can drive a pass. Let the worker go quiet before a fault + // is installed, so every pass a test counts is one the test caused. + // This is setup, not an assertion: nothing is claimed about how long it + // takes, only that counting starts from rest. + fixture.settle_for(MOUNT_QUIET_WINDOW).await; + fixture + } + + /// Block until the worker has had no pass in flight for `window`. + async fn settle_for(&self, window: Duration) { + let deadline = tokio::time::Instant::now() + SETTLE_DEADLINE; + let mut quiet_since: Option = None; + while tokio::time::Instant::now() < deadline { + if self + .registry + .reconcile_in_progress_for_test(&self.project) + .await + { + quiet_since = None; + } else { + match quiet_since { + None => quiet_since = Some(tokio::time::Instant::now()), + Some(since) if since.elapsed() >= window => return, + Some(_) => {} + } + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + } + + /// Install the fault and hand back the shared counter of dispatched passes. + async fn install_fault( + &self, + kind: ReconcileFaultKindV1, + faulting_passes: usize, + ) -> Arc { + let fault = Arc::new(ReconcileFaultInjectionV1::new(kind, faulting_passes)); + let canonical = self.project.canonicalize().expect("canonical project"); + let mounted = self.registry.mounted.lock().await; + let worktree = mounted.get(&canonical).expect("mounted worktree"); + worktree + .scheduler + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .install_reconcile_fault_for_test(Arc::clone(&fault)); + fault + } + + /// One wake that carries no new input: the control epoch does not advance, + /// exactly as it does not when a cadence tick or query wakes the worker + /// over bytes nobody touched. + async fn wake_without_new_input(&self) { + let canonical = self.project.canonicalize().expect("canonical project"); + let mounted = self.registry.mounted.lock().await; + let worktree = mounted.get(&canonical).expect("mounted worktree"); + worktree.wake.notify_one(); + } + + /// Drive `EXTERNAL_WAKE_ROUNDS` spaced wakes over unchanged input. + async fn drive_external_wakes(&self) { + for _ in 0..EXTERNAL_WAKE_ROUNDS { + self.wake_without_new_input().await; + tokio::time::sleep(WAKE_ROUND_SPACING).await; + } + } +} + +/// Poll until `attempts` reaches `target`, or the deadline expires. Returns the +/// last observed count so the caller asserts on the count, not on the wait. +async fn wait_for_attempts(fault: &ReconcileFaultInjectionV1, target: usize) -> usize { + let deadline = tokio::time::Instant::now() + SETTLE_DEADLINE; + loop { + let seen = fault.attempts(); + if seen >= target || tokio::time::Instant::now() >= deadline { + return seen; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// FINDING 1. A reconcile unit that panics on every pass must stop being +/// retried. +/// +/// The reported symptom was one malformed source file panicking the indexing +/// pool and the scheduler re-dispatching the identical unit 114 times, leaving +/// the project index permanently stale. Before the guard was wired into this +/// loop, every wake produced another attempt: the count tracked the wakes. +/// Bounded means the count stops at the policy bound however many wakes +/// arrive. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_reconcile_that_panics_every_pass_stops_being_retried() { + let fixture = Fixture::mount("project.reconcile-panic-isolation").await; + let fault = fixture + .install_fault(ReconcileFaultKindV1::Panic, usize::MAX) + .await; + + let bound = MAX_CONSECUTIVE_RECONCILE_PANICS_V1 as usize; + + // One wake starts it. The guard's own backoff drives the retries. + fixture.wake_without_new_input().await; + wait_for_attempts(&fault, bound).await; + + // Now behave like a live daemon: keep waking the worker over the same + // bytes. A wired guard suppresses every one of these. + fixture.drive_external_wakes().await; + let attempts = fault.attempts(); + + assert!( + attempts <= bound, + "a panicking unit must reach a terminal state, not one retry per wake: \ + {attempts} passes after 1 + {EXTERNAL_WAKE_ROUNDS} wakes (bound {bound})" + ); + assert!( + attempts >= 1, + "the first wake must actually dispatch a pass; {attempts} means the harness never ran" + ); + assert!( + attempts < 1 + EXTERNAL_WAKE_ROUNDS, + "unbounded retry: attempts ({attempts}) still scale with wakes ({})", + 1 + EXTERNAL_WAKE_ROUNDS + ); + + fixture.registry.shutdown().await; +} + +/// FINDING 1, other half. Quarantine must not be permanent: input that +/// actually changed advances the code-index control epoch and earns another +/// attempt, or a fixed file would never be indexed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn changed_input_lifts_a_quarantined_reconcile() { + let fixture = Fixture::mount("project.reconcile-panic-epoch").await; + let fault = fixture + .install_fault(ReconcileFaultKindV1::Panic, usize::MAX) + .await; + let bound = MAX_CONSECUTIVE_RECONCILE_PANICS_V1 as usize; + + fixture.wake_without_new_input().await; + let quarantined_at = wait_for_attempts(&fault, bound).await; + fixture.drive_external_wakes().await; + assert_eq!( + fault.attempts(), + quarantined_at, + "unchanged input must stay quarantined across every external wake" + ); + + // A real hook hint advances the control epoch: these are not the bytes + // that panicked. + assert!( + fixture + .registry + .notify_hook_paths(&fixture.project, &["src/main.rs".to_owned()]) + .await, + "the hint must reach the mounted scheduler" + ); + let after_hint = wait_for_attempts(&fault, quarantined_at + 1).await; + + assert!( + after_hint > quarantined_at, + "changed input must earn another attempt; stayed at {quarantined_at}" + ); + + fixture.registry.shutdown().await; +} + +/// FINDING 2. A reconcile refused because shared process capacity was +/// momentarily held must be retried by this worker on its own. +/// +/// Nothing wakes this worktree when the competing holder releases the budget, +/// so before the retry was wired the single failing pass was the last pass: +/// the worktree stayed stale until an unrelated query or edit happened to wake +/// it. The assertion is that a second pass happens with **no** further external +/// wake, and that it then succeeds. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_transient_capacity_refusal_is_retried_without_an_external_wake() { + let fixture = Fixture::mount("project.reconcile-capacity-retry").await; + // Refuse exactly the first pass, as a sibling holder would while it holds + // the shared budget; every later pass finds capacity. + let fault = fixture + .install_fault(ReconcileFaultKindV1::TransientCapacity, 1) + .await; + + // Exactly one wake, and never another from outside. + fixture.wake_without_new_input().await; + let attempts = wait_for_attempts(&fault, 2).await; + + assert!( + attempts >= 2, + "a transient capacity refusal must schedule its own retry; only {attempts} pass(es) \ + ran after a single wake, so the worktree stays stale until unrelated traffic arrives" + ); + + fixture.registry.shutdown().await; +} + +/// FINDING 2, guard rail. The retry must not become the bug it fixes: a +/// capacity refusal that never clears is bounded, not retried forever. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_capacity_refusal_that_never_clears_is_bounded() { + let fixture = Fixture::mount("project.reconcile-capacity-bound").await; + let fault = fixture + .install_fault(ReconcileFaultKindV1::TransientCapacity, usize::MAX) + .await; + let bound = 1 + MAX_CONSECUTIVE_CAPACITY_RETRIES_V1 as usize; + + fixture.wake_without_new_input().await; + fixture.settle_for(TERMINATION_QUIET_WINDOW).await; + let settled = fault.attempts(); + // The decisive property: self-scheduling has *stopped*. An unbounded retry + // keeps producing passes here however long the wait. + fixture.settle_for(TERMINATION_QUIET_WINDOW).await; + + assert_eq!( + fault.attempts(), + settled, + "self-scheduled capacity retries must terminate, not keep re-arming" + ); + assert!( + settled <= bound, + "self-scheduled capacity retries must respect the policy bound: \ + {settled} passes from one wake (bound {bound})" + ); + assert!( + settled >= 2, + "the bound must still allow at least one retry; saw {settled}" + ); + + fixture.registry.shutdown().await; +} + +/// FINDING 2, the distinction that matters most. A refusal that *is* a +/// resident-memory admission failure but can never be admitted — the request +/// alone exceeds the whole process limit — must not be self-retried. No other +/// holder releasing anything makes it fit, so retrying it is the unbounded +/// retry this PR exists to remove, wearing a capacity label. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_capacity_refusal_that_can_never_fit_is_not_retried() { + let fixture = Fixture::mount("project.reconcile-oversized-capacity").await; + let fault = fixture + .install_fault(ReconcileFaultKindV1::OversizedCapacity, usize::MAX) + .await; + + fixture.wake_without_new_input().await; + wait_for_attempts(&fault, 1).await; + fixture.settle_for(TERMINATION_QUIET_WINDOW).await; + + assert_eq!( + fault.attempts(), + 1, + "an admission failure whose request exceeds the process limit must not \ + be treated as transient capacity" + ); + + fixture.registry.shutdown().await; +} + +/// FINDING 2, guard rail. A refusal the same input reproduces forever must +/// **not** be self-retried either. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_permanent_refusal_is_never_self_retried() { + let fixture = Fixture::mount("project.reconcile-permanent-refusal").await; + let fault = fixture + .install_fault(ReconcileFaultKindV1::Permanent, usize::MAX) + .await; + + fixture.wake_without_new_input().await; + wait_for_attempts(&fault, 1).await; + // Any self-scheduled retry would land inside the quiet window. + fixture.settle_for(TERMINATION_QUIET_WINDOW).await; + + assert_eq!( + fault.attempts(), + 1, + "a permanent refusal must not schedule its own retry" + ); + + fixture.registry.shutdown().await; +} + +fn run_git_in(root: &Path, args: &[&str]) { + let output = Command::new("git") + .args(args) + .current_dir(root) + .env("GIT_AUTHOR_NAME", "TraceDecay Test") + .env("GIT_AUTHOR_EMAIL", "test@tracedecay.invalid") + .env("GIT_COMMITTER_NAME", "TraceDecay Test") + .env("GIT_COMMITTER_EMAIL", "test@tracedecay.invalid") + .output() + .expect("git command should run"); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); +}