From 218ae5cbf3d1656125df0a6fd73784923d56678a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 06:52:19 +0000 Subject: [PATCH 1/2] fix(code-index): stamp continuations before the seat looks idle A seat waiter that sees reconcile_in_progress at zero treats the owner as finished. Noting BusyFollowUp after that drop sampled an empty slot and then raced the failure ceiling. The stamp now holds the pass guard for the note only. A store directory that does not exist yet is an unpublished plan, not Storage(NotFound). Planning against latest_generation_id before cold open created the scoped store was that error. The retention journey now re-reads the store on the serving-seat signal. Co-authored-by: Zack Jackson --- .../src/code_index_generations.rs | 38 +++++++ .../src/code_index_generations/tests.rs | 19 ++++ .../src/code_index_scheduler/registry.rs | 15 +++ .../code_index_scheduler/registry/mount.rs | 106 ++++++++++++++---- .../generation_retention_test.rs | 52 +++++++-- 5 files changed, 202 insertions(+), 28 deletions(-) diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations.rs b/crates/tracedecay-code-index-retention/src/code_index_generations.rs index b4976c6602..2525b4948d 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations.rs @@ -778,6 +778,25 @@ pub fn plan_code_generation_retention_with_verification( ) } +/// A store directory that does not exist yet has nothing to collect. The +/// sealer creates that directory on open; until then the census is the same +/// unpublished plan an empty directory with no pointer produces. +fn unpublished_store_plan( + vector_readable_sources: &BTreeSet, +) -> CodeGenerationRetentionPlanV1 { + CodeGenerationRetentionPlanV1 { + active_generation_id: None, + vector_readable_sources: vector_readable_sources.clone(), + superseded_generations: Vec::new(), + collectable_generations: Vec::new(), + collectable_text_artifacts: Vec::new(), + collectable_generation_segments: GenerationSegmentCensusV1::NoneFound, + text_artifact_inventory_bytes: 0, + verification: GenerationDigestVerificationV1::Full, + active_pointer: None, + } +} + /// Recover any bounded prior apply, then build the next fully verified /// collection unit while preserving the caller's cancellation authority. /// @@ -795,6 +814,25 @@ pub fn prepare_next_code_generation_retention_cancellable( if observe_cancel(is_cancelled) { return Err(CodeGenerationRetentionErrorV1::Cancelled); } + // The serving seat can name a generation before the scoped store directory + // exists: cold open creates it inside the worker, and a waiter that only + // saw `latest_generation_id` plans against a path canonicalize then + // reports as `Storage(NotFound)`. That is an unpublished store, the same + // typed state as a directory with no pointer, not a storage failure the + // failure ceiling then retries. + match std::fs::metadata(store_root) { + Ok(metadata) if metadata.is_dir() => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Ok(unpublished_store_plan(vector_readable_sources)); + } + Ok(_) => { + return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( + "code-generation store '{}' is not a directory", + store_root.display() + ))); + } + Err(error) => return Err(storage(error)), + } recover_code_generation_retention_cancellable( store_root, vector_readable_sources, diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs index 2dbe91a0c7..f403f1b703 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs @@ -3072,3 +3072,22 @@ fn recovery_completes_a_committed_rewrite_that_never_reached_the_pointer() { plan_code_generation_retention(fixture.store.path(), &BTreeSet::new()) .expect("a recovered store must stay plannable"); } + +#[test] +fn missing_store_is_an_unpublished_plan_not_a_storage_failure() { + let missing = std::env::temp_dir().join(format!( + "tracedecay-missing-code-store-{}", + std::process::id() + )); + assert!(!missing.exists()); + let plan = prepare_next_code_generation_retention_cancellable( + &missing, + &BTreeSet::new(), + &|| false, + None, + ) + .expect("a store that has not been opened is unpublished"); + assert_eq!(plan.active_generation_id, None); + assert!(plan.collectable_generations.is_empty()); + assert!(!plan.has_collectable_work()); +} diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs index a5392c83b1..8e4341a777 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs @@ -2109,6 +2109,21 @@ impl CodeIndexSchedulerRegistryV1 { } } + /// Stamp a continuation while `reconcile_in_progress` still reports this pass. + /// + /// A seat waiter that sees the pass counter at zero treats the owner as + /// idle. Noting `BusyFollowUp` after that drop is the race: the waiter + /// samples an empty slot, then loses to the stamp and burns the failure + /// ceiling. The guard lives only for the note. + fn note_visible_worker_continuation( + passes: &Arc, + pending_wake: &PendingWakeV1, + wake: &tokio::sync::Notify, + ) { + let _visible = super::ReconcilePassGuard::enter(passes); + Self::note_worker_continuation(pending_wake, wake); + } + /// Claim the pending wake as one reconcile's arrival, at the instant the /// scheduler dequeues it. /// diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs index ac239ce49c..a38c3733ad 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/mount.rs @@ -1168,7 +1168,20 @@ impl CodeIndexSchedulerRegistryV1 { // A successor-only retained projection holds no pass guard of // its own; keeping the worker's guard through graph seat would // report rebuild_in_flight for clone backfill that is not - // exact/lexical work. + // exact/lexical work. Stamp the continuation this projection + // already owes before that drop, so idle means the slot is set. + if let Some(outcome) = published_text_projection_outcome.as_ref() { + let schedule_continuation = match outcome { + PublishedTextProjectionOutcomeV1::Finished => graph_text + .as_ref() + .is_some_and(LatestCodeTextGenerationV1::text_projection_needs_work), + PublishedTextProjectionOutcomeV1::Unfinished => true, + PublishedTextProjectionOutcomeV1::Shutdown => false, + }; + if schedule_continuation { + Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + } + } if retained_text_projection.is_none() || retained_projection_successor_only { drop(reconcile_pass.take()); } @@ -1400,8 +1413,13 @@ impl CodeIndexSchedulerRegistryV1 { // all and never published the successor generation. The // `retained_graph_head_recovery_attempted` guard above is // now false for every later pass, so this cannot spin - // another retained-recovery Noop. - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + // another retained-recovery Noop. The optional-graph drop + // may already have happened; re-enter the pass for the stamp. + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } // A recovered revision-7 verified head already serves its // native graph from the retained text owner, and that owner @@ -1548,8 +1566,12 @@ impl CodeIndexSchedulerRegistryV1 { if roster_refusal_rebuild { // One pass, claimed from the scheduler, so // a refusal that keeps reproducing cannot - // spin this worker. - Self::note_worker_continuation( + // spin this worker. Stamp while a pass + // guard is held: the step guard above has + // already dropped, and an idle read must + // not sample the empty slot. + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, &worker_pending_wake, &worker_wake, ); @@ -1714,7 +1736,15 @@ impl CodeIndexSchedulerRegistryV1 { .as_ref() .is_some_and(LatestCodeTextGenerationV1::text_projection_needs_work) { - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + // Stamped before the optional-graph guard drop + // when this outcome was already known. Re-enter + // so a reader that cleared the slot during + // graph cannot sample the stamp as idle. + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } // Large text projections can outlive the bounded // source proof established before publication. The @@ -1784,7 +1814,11 @@ impl CodeIndexSchedulerRegistryV1 { "the publication's text owner did not finish its projection; \ the sealed generation stays unseated until it does" ); - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } } // Keep the pass lifetime around the post-projection source @@ -1965,7 +1999,11 @@ impl CodeIndexSchedulerRegistryV1 { if text_latest.text_projection_needs_work() && !text_latest.query_owners_are_ready() { - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } } Ok(Err(error)) => { @@ -1994,7 +2032,27 @@ impl CodeIndexSchedulerRegistryV1 { } // The source proof and serving witness are now published as // one lifecycle. Optional receipts do not keep source - // verification in flight. + // verification in flight. Stamp a clone-backfill continuation + // before the drop: a seat waiter that sees the counter at zero + // must already observe the slot, or it races the failure ceiling. + if clone_backfill_waiting_for_source + && matches!( + &result, + Ok((Ok(CodeIndexReconcileOutcomeV1::Noop(_)), _, _)) + ) + && worker_text_generation + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_some() + && worker_source_freshness + .ready_without_stat(&worker_project_root, &worker_shutting_down) + { + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); + } drop(reconcile_pass.take()); if let Ok((Ok(outcome), _, _)) = &result { // A pass that ran to a terminal outcome proves neither the @@ -2046,12 +2104,8 @@ impl CodeIndexSchedulerRegistryV1 { ); } worker_serving_generation_changed.send_replace(()); - // The retained slice was checked before reconciliation - // renewed this proof. Preserve its wake now that source - // is current, without requiring another query arrival. - if clone_backfill_waiting_for_source { - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); - } + // The clone-backfill continuation was stamped before + // this pass dropped `reconcile_in_progress`. } } else { // Surface bounded non-terminal failure without new project-path data. @@ -2269,7 +2323,6 @@ impl CodeIndexSchedulerRegistryV1 { PublishedTextProjectionOutcomeV1::Unfinished } }; - drop(reconcile_pass.take()); match outcome { PublishedTextProjectionOutcomeV1::Finished if !retained_head_recovered_without_complete_replay @@ -2281,7 +2334,11 @@ impl CodeIndexSchedulerRegistryV1 { // full replay can proceed without overlapping it. // A clone-fingerprint backfill changed no owner // the seat reads, so it owes no such pass. - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } PublishedTextProjectionOutcomeV1::Finished => { let installed_owner_still_needs_work = worker_text_generation @@ -2292,7 +2349,11 @@ impl CodeIndexSchedulerRegistryV1 { LatestCodeTextGenerationV1::text_projection_needs_work, ); if installed_owner_still_needs_work { - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } } PublishedTextProjectionOutcomeV1::Shutdown => { @@ -2313,9 +2374,16 @@ impl CodeIndexSchedulerRegistryV1 { // stopped short would sleep until an unrelated // arrival, exactly as the inline slice's own // follow-up notify prevented. - Self::note_worker_continuation(&worker_pending_wake, &worker_wake); + Self::note_visible_worker_continuation( + &worker_reconcile_in_progress, + &worker_pending_wake, + &worker_wake, + ); } } + // The continuation is already in the slot. This drop is the + // first moment the pass looks idle. + drop(reconcile_pass.take()); } if worker_shutting_down.load(Ordering::Acquire) { tracing::info!( diff --git a/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs b/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs index b0d224dfec..4e7da22927 100644 --- a/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs @@ -11,7 +11,8 @@ use super::journey_test_support::git; use super::*; use crate::daemon::maintenance::project_store_maintenance_lease; use tracedecay_code_index_retention::code_index_generations::{ - MAX_CODE_GENERATION_RETENTION_BATCH_V1, prepare_next_code_generation_retention_cancellable, + CodeGenerationRetentionErrorV1, MAX_CODE_GENERATION_RETENTION_BATCH_V1, + prepare_next_code_generation_retention_cancellable, }; use tracedecay_maintenance::tick::{MaintenanceContinuation, MaintenanceTickOutcome}; @@ -97,7 +98,13 @@ async fn mounted_code_generation_retention_continues_capped_segment_reclamation( .expect("project server") .cg() .await; - let canonical_root = graph.project_root().to_path_buf(); + // The scheduler hashes the canonical project root. A non-canonical + // `project_root()` names a store that is never created, and + // `latest_generation_id` still answers because it canonicalizes itself. + let canonical_root = graph + .project_root() + .canonicalize() + .expect("canonical project root"); let first_source = schedulers .latest_generation_id(&canonical_root) .await @@ -114,13 +121,40 @@ async fn mounted_code_generation_retention_continues_capped_segment_reclamation( &canonical_root, ); let graph_replay_pool_root = graph.db().database_path().with_extension("graph-replay"); - let plan = prepare_next_code_generation_retention_cancellable( - &code_store_root, - &BTreeSet::new(), - &|| false, - Some(&graph_replay_pool_root), - ) - .expect("code generation retention plan"); + // Text seating moves `latest_generation_id` before the scoped store + // exists and before the superseded sealed file is collectable. Planning + // once at that instant is the race: canonicalize returns NotFound, and + // a wall-clock retry of the same snapshot hits the failure ceiling. + // Wake on the serving seat and re-read the store. + let mut serving_seats = schedulers.subscribe_serving_seats(); + let plan = tokio::time::timeout(Duration::from_mins(2), async { + loop { + match prepare_next_code_generation_retention_cancellable( + &code_store_root, + &BTreeSet::new(), + &|| false, + Some(&graph_replay_pool_root), + ) { + Ok(plan) + if plan + .collectable_generations + .iter() + .any(|generation| generation.generation_id == first_source) => + { + return plan; + } + Ok(_) => {} + Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy) => {} + Err(error) => panic!("code generation retention plan: {error}"), + } + serving_seats + .changed() + .await + .expect("the seating channel stays open while the registry lives"); + } + }) + .await + .expect("superseded source became collectable after the serving seat moved"); let first_candidate = plan .collectable_generations .iter() From 543b6e253805e5c142e30675e0ddf4d12af2e536 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 08:22:00 +0000 Subject: [PATCH 2/2] style(session-temporal): keep the test module last and rustfmt master Master run 35431539771 failed Check formatting (projector.rs, query.rs) and Clippy (items_after_test_module in query.rs) after #1844/#1845 merged without CI. Co-Authored-By: Claude Fable 5.1 --- .../projector.rs | 8 ++--- .../src/query.rs | 29 +++++++++---------- 2 files changed, 18 insertions(+), 19 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs index 10818733c1..632898ca24 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs @@ -119,11 +119,11 @@ impl SessionTemporalRefreshProjector for CanonicalSessionTemporalProjector { // Empty remaining range is a durable no-op: terminalize with an // empty complete progress batch instead of deferring forever. Ok(None) => canonical_noop_complete_effect(&recovery), - Err(error) if error.is_storage() => Err( - SessionTemporalRefreshProjectorError::retryable(format!( + Err(error) if error.is_storage() => { + Err(SessionTemporalRefreshProjectorError::retryable(format!( "source_busy: {error}" - )), - ), + ))) + } Err(_) => Err(SessionTemporalRefreshProjectorError::terminal( "projector_failed", )), diff --git a/crates/tracedecay-session-temporal-store/src/query.rs b/crates/tracedecay-session-temporal-store/src/query.rs index 367b14bb79..ae0ab6b4de 100644 --- a/crates/tracedecay-session-temporal-store/src/query.rs +++ b/crates/tracedecay-session-temporal-store/src/query.rs @@ -246,8 +246,7 @@ pub(super) async fn read_observations( // ceiling. Split until a single observation remains; that // observation is then a typed storage failure, not a retry // that looks like a busy source. - if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 - { + if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 { let mid = start + chunk.len() / 2; pending.push((mid, end)); pending.push((start, mid)); @@ -289,11 +288,18 @@ fn observation_prefetch_exceeded_materialization_limit(error: &SessionStoreError } } +/// The error `read_observation` raises for an id the store does not hold, reused +/// by callers that resolve prefetched observations out of a batch map. +pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { + storage_message( + PERSIST_OPERATION, + format!("source observation {} is missing", observation_id.as_str()), + ) +} + #[cfg(test)] mod tests { - use super::{ - PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage, - }; + use super::{PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage}; #[test] fn materialization_limit_is_the_prefetch_split_signal() { @@ -308,15 +314,8 @@ mod tests { assert!(observation_prefetch_exceeded_materialization_limit( &exceeded )); - assert!(!observation_prefetch_exceeded_materialization_limit(&locked)); + assert!(!observation_prefetch_exceeded_materialization_limit( + &locked + )); } } - -/// The error `read_observation` raises for an id the store does not hold, reused -/// by callers that resolve prefetched observations out of a batch map. -pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { - storage_message( - PERSIST_OPERATION, - format!("source observation {} is missing", observation_id.as_str()), - ) -}