fix(retention): defer a census that races a missing path

crates/tracedecay-code-index-retention/src/code_index_generations.rs

@@ -885,6 +885,9 @@ fn plan_code_generation_retention_with_verification_cancellable(
         Err(error) if error.kind() == std::io::ErrorKind::NotFound && active_pointer.is_none() => {
             None
         }
+        // A pointer is only durable once its generation directory is, so a
+        // live pointer over an absent directory is loss, not a publisher
+        // race, and must stay loud.
         Err(error) => return Err(storage(error)),
     };
     let mut generations = BTreeMap::new();
@@ -1218,7 +1221,7 @@ fn sweep_unreferenced_generation_segments(
                 continue;
             }
             let mut reader = CancellableGenerationManifestReaderV1 {
-                file: File::open(&path).map_err(storage)?,
+                file: File::open(&path).map_err(deferred_if_absent)?,
                 hasher: Sha256::new(),
                 is_cancelled,
                 cancelled: false,
@@ -1281,7 +1284,7 @@ fn sweep_unreferenced_generation_segments(
         if live_segments.contains(&format!("sha256:{digest}")) {
             continue;
         }
-        let metadata = path.symlink_metadata().map_err(storage)?;
+        let metadata = path.symlink_metadata().map_err(deferred_if_absent)?;
         if !metadata.file_type().is_file() {
             return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
                 "generation segment '{}' is not a regular file",
@@ -2033,5 +2036,18 @@ fn storage(error: impl std::fmt::Display) -> CodeGenerationRetentionErrorV1 {
     CodeGenerationRetentionErrorV1::Storage(error.to_string())
 }
 
+/// A path that is not there yet, or that a peer unlinked after this census
+/// listed it, is not a broken disk. The publisher creates the scope root and
+/// the sealed files under the store lock, then drops that lock; a census that
+/// does not hold the lock can observe the gap. The next tick sees a stable
+/// tree. Every other I/O failure stays a storage error.
+pub(super) fn deferred_if_absent(error: std::io::Error) -> CodeGenerationRetentionErrorV1 {
+    if error.kind() == std::io::ErrorKind::NotFound {
+        CodeGenerationRetentionErrorV1::GenerationStoreBusy
+    } else {
+        storage(error)
+    }
+}
+
 #[cfg(test)]
 mod tests;

crates/tracedecay-code-index-retention/src/code_index_generations/generation_scan.rs

@@ -7,7 +7,8 @@ use tracedecay_domain::canonical_text::{encode_tagged_lowercase_hex, is_lowercas
 
 use super::{
     CodeGenerationRetentionErrorV1, GenerationDigestVerificationV1,
-    MAX_GENERATION_METADATA_PREFIX_BYTES, SealedGenerationManifestMetadataV1, storage,
+    MAX_GENERATION_METADATA_PREFIX_BYTES, SealedGenerationManifestMetadataV1, deferred_if_absent,
+    storage,
 };
 
 const MAX_FORMAT_REVISION_PREFIX_BYTES: usize = 4 * 1024;
@@ -16,7 +17,7 @@ pub(super) fn read_generation_format_revision(
     path: &Path,
     is_cancelled: &dyn Fn() -> bool,
 ) -> Result<u32, CodeGenerationRetentionErrorV1> {
-    let mut file = File::open(path).map_err(storage)?;
+    let mut file = File::open(path).map_err(deferred_if_absent)?;
     let mut prefix = vec![0_u8; MAX_FORMAT_REVISION_PREFIX_BYTES];
     let bytes_read = file.read(&mut prefix).map_err(storage)?;
     crate::hotpath_observe::retention_inspected(bytes_read as u64);
@@ -40,7 +41,7 @@ pub(super) fn read_generation_metadata(
     is_cancelled: &dyn Fn() -> bool,
 ) -> Result<(u32, SealedGenerationManifestMetadataV1, String, u64), CodeGenerationRetentionErrorV1>
 {
-    let mut file = File::open(path).map_err(storage)?;
+    let mut file = File::open(path).map_err(deferred_if_absent)?;
     let size_bytes = file.metadata().map_err(storage)?.len();
     let mut hasher = Sha256::new();
     let mut prefix = Vec::with_capacity(MAX_GENERATION_METADATA_PREFIX_BYTES);

crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs

@@ -102,7 +102,7 @@ fn lock_file(
 }
 
 fn canonical_store_root(store_root: &Path) -> Result<PathBuf, CodeGenerationRetentionErrorV1> {
-    std::fs::canonicalize(store_root).map_err(storage)
+    std::fs::canonicalize(store_root).map_err(super::deferred_if_absent)
 }
 
 fn open_lock_file(path: &Path) -> Result<File, CodeGenerationRetentionErrorV1> {
@@ -112,5 +112,5 @@ fn open_lock_file(path: &Path) -> Result<File, CodeGenerationRetentionErrorV1> {
         .write(true)
         .truncate(false)
         .open(path)
-        .map_err(storage)
+        .map_err(super::deferred_if_absent)
 }

crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs

@@ -1547,6 +1547,23 @@ fn idle_maintenance_preparation_stays_metadata_only() {
     );
 }
 
+#[test]
+fn preparation_defers_when_the_scope_root_does_not_exist_yet() {
+    let parent = tempfile::TempDir::new().expect("parent");
+    let missing = parent.path().join("not-created");
+    let error = prepare_next_code_generation_retention_cancellable(
+        &missing,
+        &BTreeSet::new(),
+        &|| false,
+        None,
+    )
+    .expect_err("an unpublished scope root has no census");
+    assert!(
+        matches!(error, CodeGenerationRetentionErrorV1::GenerationStoreBusy),
+        "a missing scope root is the publisher's create window, not a storage failure: {error:?}"
+    );
+}
+
 #[test]
 fn metadata_only_segment_census_observes_at_most_one_directory_entry() {
     let store = tempfile::TempDir::new().expect("create unpublished store");
@@ -3118,3 +3135,28 @@ 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");
 }
+
+/// The census opens every name `read_dir` just returned. Publication can
+/// unlink that name first. `NotFound` is the same deferral as a held writer,
+/// not a storage failure. Any other open failure stays storage.
+#[test]
+fn vanished_listed_generation_open_defers_instead_of_storage_loss() {
+    let root = tempfile::tempdir().expect("census root");
+    let missing = root.path().join(format!("generation-{:064x}.json", 1));
+    let error = super::generation_scan::read_generation_format_revision(&missing, &|| false)
+        .expect_err("a vanished listed generation defers the census");
+    assert!(
+        matches!(error, CodeGenerationRetentionErrorV1::GenerationStoreBusy),
+        "a missing listed generation is a publisher race, not a storage failure: {error:?}"
+    );
+
+    let directory = root.path().join("not-a-generation-file");
+    std::fs::create_dir(&directory).expect("directory where a file was listed");
+    let storage_error =
+        super::generation_scan::read_generation_format_revision(&directory, &|| false)
+            .expect_err("a directory is not a vanished file");
+    assert!(
+        matches!(storage_error, CodeGenerationRetentionErrorV1::Storage(_)),
+        "non-NotFound census I/O stays a storage failure: {storage_error:?}"
+    );
+}

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",
                 )),

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()),
-    )
-}

crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs

@@ -116,10 +116,12 @@ async fn mounted_code_generation_retention_continues_capped_segment_reclamation(
         );
     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.
+    // `GenerationStoreBusy` whenever a writer owns the store, and the same
+    // probe over the graph replay pool answers `GraphReplayPoolBusy`;
+    // production maintenance defers both and comes back. This route stays
+    // mounted, so the pass tail that publishes the edits above can still own
+    // either lock here. Consume the same typed answers instead of reading
+    // them as failures.
     let plan = tokio::time::timeout(Duration::from_secs(30), async {
         loop {
             match prepare_next_code_generation_retention_cancellable(
@@ -129,7 +131,10 @@ async fn mounted_code_generation_retention_continues_capped_segment_reclamation(
                 Some(&graph_replay_pool_root),
             ) {
                 Ok(plan) => return plan,
-                Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy) => {
+                Err(
+                    CodeGenerationRetentionErrorV1::GenerationStoreBusy
+                    | CodeGenerationRetentionErrorV1::GraphReplayPoolBusy,
+                ) => {
                     tokio::time::sleep(Duration::from_millis(25)).await;
                 }
                 Err(error) => panic!("code generation retention plan: {error:?}"),