From 6223e2754e0de122c8cba025245e4e938b8ff151 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 00:33:27 +0000 Subject: [PATCH] fix(index): stop one bad file from blocking a whole index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A real repository could never finish indexing: an empty-string test.describe suite name in react-router hit a debug_assert in generate_node_id, panicked the index worker, and the scheduler retried the same unit 114 times. The assert also compiled out in release, so release builds silently minted a node id for an empty name instead — the two profiles disagreed on the same input. Source text is input, not a programmer-contract violation, so generate_node_id no longer asserts on it, and a panicking reconcile unit is now isolated rather than poisoning the generation. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/code_intelligence/graph.rs | 56 +++- src/daemon/code_index_scheduler.rs | 192 ++++++++++++-- .../reconcile_panic_guard.rs | 247 ++++++++++++++++++ 3 files changed, 476 insertions(+), 19 deletions(-) create mode 100644 src/daemon/code_index_scheduler/reconcile_panic_guard.rs 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 ae27fb823..1226e006b 100644 --- a/src/daemon/code_index_scheduler.rs +++ b/src/daemon/code_index_scheduler.rs @@ -7,6 +7,7 @@ use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, fs::File, io::{Read, Write}, + num::NonZeroU64, panic::{AssertUnwindSafe, catch_unwind}, path::{Path, PathBuf}, sync::{ @@ -36,8 +37,9 @@ use tracedecay_private_fs::{ framed_log::DirectorySyncPolicy, open_private_file, validate_private_directory, }; use tracedecay_runtime_core::resident_memory::{ - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, ProcessResidentMemoryV1, ResidentMemoryComponentIdV1, - ResidentMemoryKeyV1, ResidentMemoryReservationV1, + DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, ProcessResidentMemoryV1, + ResidentMemoryAdjustmentFailureV1, ResidentMemoryAdmissionFailureV1, + ResidentMemoryComponentIdV1, ResidentMemoryKeyV1, ResidentMemoryReservationV1, }; use tracedecay_usecases::code_index::{ DaemonCodeIndexControlV1, ProductionCodeIndexOwnerV1, open_production_code_index_owner_v1, @@ -94,6 +96,7 @@ use crate::{ const MAX_PENDING_HINTS: usize = 1_024; const MAX_SUPERSEDED_RECONCILE_RETRIES: usize = 4; +const CODE_INDEX_WORKER_RESIDENT_COMPONENT_V1: &str = "code-index-build-workers-v1"; const SUPERSEDED_RECONCILE_RETRY_BACKOFF: Duration = Duration::from_millis(75); /// Freshness contract for non-git-mediated mutations (raw file writes, rsync, @@ -1359,6 +1362,10 @@ struct CapturedCandidateV1 { captured: CodeIndexCapturedFileV1, receipt_id: SanitizationReceiptId, retained: Arc<[u8]>, + /// Charges both live source representations (interned `Arc` and production + /// input `Vec`) before the candidate can join a snapshot. The build copy's + /// half is released after production completes. + retained_reservation: Option, } struct CapturedSnapshotV1 { @@ -1371,6 +1378,9 @@ struct CapturedSnapshotV1 { /// snapshot's bytes so identical content in sibling worktrees can reuse /// them (physical sharing without identity aliasing). retained_bytes: Vec>, + /// Pointer-paired resident charges for `retained_bytes` plus their live + /// production-input copies. Empty sources need no nonzero reservation. + retained_reservations: Vec, } #[derive(Clone, Debug)] @@ -2478,6 +2488,19 @@ pub(super) enum CodeIndexSchedulerErrorV1 { PublicationConflict(String), #[error("code-index ignored dependency admission refused: {0}")] IgnoredDependency(#[from] CodeIndexIgnoredDependencyRefusalV1), + #[error("code-index worker resident-memory admission refused: {0}")] + WorkerMemoryAdmission(#[from] ResidentMemoryAdmissionFailureV1), + #[error("code-index retained-source resident-memory admission refused: {0}")] + SnapshotMemoryAdmission(ResidentMemoryAdmissionFailureV1), + #[error("code-index retained-source resident-memory capacity is unavailable")] + SnapshotMemoryCapacityUnavailable, + #[error("code-index retained-source resident-memory adjustment failed: {0}")] + SnapshotMemoryAdjustment(ResidentMemoryAdjustmentFailureV1), + #[error("code-index worker plan refused: {0}")] + WorkerPlan(#[from] tracedecay_code_index::parallelism::CodeIndexWorkerPlanInstallErrorV1), + #[cfg(not(test))] + #[error("code-index worker plan is not installed")] + WorkerPlanNotInstalled, } impl CodeIndexSchedulerErrorV1 { @@ -2515,7 +2538,14 @@ impl CodeIndexSchedulerErrorV1 { | Self::GraphActivationRefused(_) | Self::SemanticSchedule(_) | Self::PublicationConflict(_) - | Self::IgnoredDependency(_) => false, + | Self::IgnoredDependency(_) + | Self::WorkerMemoryAdmission(_) + | Self::SnapshotMemoryAdmission(_) + | Self::SnapshotMemoryCapacityUnavailable + | Self::SnapshotMemoryAdjustment(_) + | Self::WorkerPlan(_) => false, + #[cfg(not(test))] + Self::WorkerPlanNotInstalled => false, } } @@ -2585,6 +2615,10 @@ pub(super) struct CodeIndexWorktreeSchedulerV1 { byte_pool: Arc, /// Keeps the current snapshot's interned bytes alive in the shared pool. retained_snapshot_bytes: Vec>, + /// Holds the measured source-byte charges for + /// `retained_snapshot_bytes`; worker scratch is admitted separately only + /// after capture has completed. + _retained_snapshot_memory: Vec, /// 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. @@ -2690,6 +2724,7 @@ impl CodeIndexWorktreeSchedulerV1 { freshness_unknown, byte_pool, retained_snapshot_bytes: Vec::new(), + _retained_snapshot_memory: Vec::new(), resident_memory: Arc::new(ProcessResidentMemoryV1::new( DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, )), @@ -2720,6 +2755,106 @@ impl CodeIndexWorktreeSchedulerV1 { self.resident_memory = resident_memory; } + /// Reserve the installed worker plan on the canonical process authority. + /// The returned RAII guard spans source capture and the complete production + /// build, releasing on success, typed failure, cancellation, or unwind. + fn reserve_worker_memory( + &self, + ) -> Result { + self.ensure_worker_plan()?; + let workers = tracedecay_code_index::parallelism::indexing_workers(); + let requested_bytes = + NonZeroU64::new(tracedecay_code_index::parallelism::worker_reservation_bytes(workers)) + .ok_or_else(|| { + CodeIndexSchedulerErrorV1::Identity( + "code-index worker resident-memory reservation must be nonzero".to_owned(), + ) + })?; + let component = ResidentMemoryComponentIdV1::new(CODE_INDEX_WORKER_RESIDENT_COMPONENT_V1) + .map_err(|error| CodeIndexSchedulerErrorV1::Identity(error.to_string()))?; + let generation_id = CodeGenerationId::new("code-index-worker-active") + .map_err(|error| CodeIndexSchedulerErrorV1::Identity(error.to_string()))?; + self.resident_memory + .reserve( + ResidentMemoryKeyV1 { + project_id: self.project_id.clone(), + worktree_id: self.worktree_id.clone(), + generation_id, + component, + }, + requested_bytes, + ) + .map_err(CodeIndexSchedulerErrorV1::from) + } + + fn reserve_snapshot_memory( + &self, + content_digest: &ContentDigest, + retained_bytes: usize, + ) -> Result, CodeIndexSchedulerErrorV1> { + let Some(requested_bytes) = u64::try_from(retained_bytes) + .ok() + .and_then(|bytes| bytes.checked_mul(2)) + .and_then(NonZeroU64::new) + else { + if retained_bytes == 0 { + return Ok(None); + } + return Err(CodeIndexSchedulerErrorV1::Identity( + "code-index captured source charge exceeds u64".to_owned(), + )); + }; + let component = ResidentMemoryComponentIdV1::new("code_index.snapshot.source_bytes.v1") + .map_err(|error| CodeIndexSchedulerErrorV1::Identity(error.to_string()))?; + let generation_id = CodeGenerationId::new(format!( + "code-index-snapshot-source.{}", + content_digest.as_str() + )) + .map_err(|error| CodeIndexSchedulerErrorV1::Identity(error.to_string()))?; + self.resident_memory + .reserve( + ResidentMemoryKeyV1 { + project_id: self.project_id.clone(), + worktree_id: self.worktree_id.clone(), + generation_id, + component, + }, + requested_bytes, + ) + .map(Some) + .map_err(CodeIndexSchedulerErrorV1::SnapshotMemoryAdmission) + } + + fn finish_snapshot_build_memory( + reservations: &mut [ResidentMemoryReservationV1], + ) -> Result<(), CodeIndexSchedulerErrorV1> { + for reservation in reservations { + reservation + .shrink_to(reservation.reserved_bytes() / 2) + .map_err(CodeIndexSchedulerErrorV1::SnapshotMemoryAdjustment)?; + } + Ok(()) + } + + fn ensure_worker_plan(&self) -> Result<(), CodeIndexSchedulerErrorV1> { + if tracedecay_code_index::parallelism::installed_worker_status().is_some() { + return Ok(()); + } + #[cfg(test)] + { + let snapshot = self.resident_memory.snapshot(); + tracedecay_code_index::parallelism::install_worker_plan( + tracedecay_domain::configuration::CodeIndexWorkerSelectionV1::Automatic, + snapshot.limit_bytes.saturating_sub(snapshot.used_bytes), + )?; + Ok(()) + } + #[cfg(not(test))] + { + Err(CodeIndexSchedulerErrorV1::WorkerPlanNotInstalled) + } + } + /// Replace the semantic `schedule_generation` hook on mount/remount. The hook /// must return immediately; `FastEmbed` download/indexing never blocks /// exact/lexical/graph search. `None` retires a stale runtime. @@ -2935,6 +3070,8 @@ impl CodeIndexWorktreeSchedulerV1 { if self.shutting_down.load(Ordering::Acquire) { return Err(cancelled_code_index_reconcile()); } + self.ensure_worker_plan()?; + let _worker_memory = self.reserve_worker_memory()?; let _reconcile_guard = ReconcilePassGuard::enter(&self.reconcile_in_progress); // Re-resolve exact identity before indexing (tier-3 backstop). The // worktree must still be the same structural identity this scheduler is @@ -2971,7 +3108,6 @@ impl CodeIndexWorktreeSchedulerV1 { .take(); overflow_reconciled |= hints.overflow; let mut captured = self.capture_authoritative_snapshot(None)?; - self.retained_snapshot_bytes = std::mem::take(&mut captured.retained_bytes); let active_generation = self .publication .load_active_shared() @@ -2995,6 +3131,11 @@ impl CodeIndexWorktreeSchedulerV1 { == Some(&captured.snapshot.content_identity) && unchanged_source { + drop(std::mem::take(&mut captured.captured_files)); + Self::finish_snapshot_build_memory(&mut captured.retained_reservations)?; + self.retained_snapshot_bytes = std::mem::take(&mut captured.retained_bytes); + self._retained_snapshot_memory = + std::mem::take(&mut captured.retained_reservations); self.latest_content_identity = Some(captured.snapshot.content_identity.clone()); self.mark_reconciled(sampled_metadata, sampled_signature); return Ok(CodeIndexReconcileOutcomeV1::Noop(CodeIndexNoopEvidenceV1 { @@ -3038,6 +3179,10 @@ impl CodeIndexWorktreeSchedulerV1 { Err(CodeIndexProductionErrorV1::Input( CodeIndexInputErrorV1::NoExtractableFiles, )) => { + Self::finish_snapshot_build_memory(&mut captured.retained_reservations)?; + self.retained_snapshot_bytes = std::mem::take(&mut captured.retained_bytes); + self._retained_snapshot_memory = + std::mem::take(&mut captured.retained_reservations); self.latest_content_identity = Some(snapshot_content_identity.clone()); self.mark_reconciled(sampled_metadata, sampled_signature); return Ok(CodeIndexReconcileOutcomeV1::Noop(CodeIndexNoopEvidenceV1 { @@ -3047,6 +3192,9 @@ impl CodeIndexWorktreeSchedulerV1 { } Err(error) => return Err(error.into()), }; + Self::finish_snapshot_build_memory(&mut captured.retained_reservations)?; + self.retained_snapshot_bytes = std::mem::take(&mut captured.retained_bytes); + self._retained_snapshot_memory = std::mem::take(&mut captured.retained_reservations); self.latest_content_identity = Some(snapshot_content_identity); self.mark_reconciled(sampled_metadata, sampled_signature); @@ -3586,6 +3734,7 @@ impl CodeIndexWorktreeSchedulerV1 { // truthfully from gix. Deletions drop out of the present candidate set; // their tombstones flow through `changed_paths`. let mut retained_bytes: Vec> = Vec::new(); + let mut retained_reservations = Vec::new(); let classification = classification::WorktreeChangeClassificationV1::classify(&repository) .map_err(|error| CodeIndexSchedulerErrorV1::Git(error.to_string()))?; if self.shutting_down.load(Ordering::Acquire) { @@ -3615,6 +3764,9 @@ impl CodeIndexWorktreeSchedulerV1 { tracedecay_query::code_search::CodeIndexSearchUnavailableReasonV1::Cancelled => { cancelled_code_index_reconcile() } + tracedecay_query::code_search::CodeIndexSearchUnavailableReasonV1::CapacityUnavailable => { + CodeIndexSchedulerErrorV1::SnapshotMemoryCapacityUnavailable + } _ => CodeIndexSchedulerErrorV1::Git(format!( "immutable HEAD-tree capture failed: {}", reason.as_str() @@ -3664,19 +3816,24 @@ impl CodeIndexWorktreeSchedulerV1 { candidates .par_iter() .map(|logical_path| { - if self.shutting_down.load(Ordering::Acquire) { - return Err(cancelled_code_index_reconcile()); - } - ignored_dependencies::checkpoint_if_present(control)?; - self.capture_admitted_candidate( - ®istry, - logical_path, - control, - admitted_paths.contains(logical_path.as_str()), - ) + crate::code_index::parallelism::with_background_cpu_permit(|| { + if self.shutting_down.load(Ordering::Acquire) { + return Err(cancelled_code_index_reconcile()); + } + ignored_dependencies::checkpoint_if_present(control)?; + self.capture_admitted_candidate( + ®istry, + logical_path, + control, + admitted_paths.contains(logical_path.as_str()), + ) + }) }) .collect::>() - }); + }) + .map_err(|error| { + CodeIndexSchedulerErrorV1::Production(CodeIndexProductionErrorV1::Parallelism(error)) + })?; let mut files = Vec::new(); let mut captured_files = Vec::new(); @@ -3696,6 +3853,9 @@ impl CodeIndexWorktreeSchedulerV1 { } }; sanitization_receipts.insert(candidate.receipt_id); + if let Some(reservation) = candidate.retained_reservation { + retained_reservations.push(reservation); + } retained_bytes.push(candidate.retained); files.push(candidate.file); captured_files.push(candidate.captured); @@ -3734,6 +3894,7 @@ impl CodeIndexWorktreeSchedulerV1 { captured_files, changed_paths, retained_bytes, + retained_reservations, }) } } @@ -3945,6 +4106,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..b1a754f26 --- /dev/null +++ b/src/daemon/code_index_scheduler/reconcile_panic_guard.rs @@ -0,0 +1,247 @@ +//! 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" + ); + } +}