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 15ae71fb0e..d726d53d46 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 @@ -3118,3 +3118,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-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()), - ) -} 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 6415ef3209..bcffa29d9c 100644 --- a/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs @@ -98,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 @@ -115,12 +121,16 @@ 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"); - // The planner probes the generation-store lock and answers - // `GenerationStoreBusy` whenever a writer owns the store; production - // maintenance defers that tick and comes back. This route stays mounted, - // so the pass tail that publishes the edits above can still own the store - // here. Consume the same typed answer instead of reading it as a failure. - let plan = tokio::time::timeout(Duration::from_secs(30), async { + // 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. The planner also probes + // the generation-store lock and answers `GenerationStoreBusy` whenever a + // writer owns the store; that lock state publishes no seat, so the wait + // keeps the short maintenance-style tick as its floor. + 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, @@ -128,16 +138,27 @@ async fn mounted_code_generation_retention_continues_capped_segment_reclamation( &|| false, Some(&graph_replay_pool_root), ) { - Ok(plan) => return plan, - Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy) => { - tokio::time::sleep(Duration::from_millis(25)).await; + 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:?}"), } + tokio::select! { + changed = serving_seats.changed() => changed + .expect("the seating channel stays open while the registry lives"), + () = tokio::time::sleep(Duration::from_millis(25)) => {} + } } }) .await - .expect("code generation retention plan converges"); + .expect("superseded source became collectable after the serving seat moved"); let first_candidate = plan .collectable_generations .iter()