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..b5f76713c9 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations.rs +++ b/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; diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/generation_scan.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/generation_scan.rs index 91b997173c..7c2923eaf2 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/generation_scan.rs +++ b/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 { - 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); diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs index 6bdc552abd..1d8d867a75 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs +++ b/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 { - 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 { @@ -112,5 +112,5 @@ fn open_lock_file(path: &Path) -> Result { .write(true) .truncate(false) .open(path) - .map_err(storage) + .map_err(super::deferred_if_absent) } 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..47d6d406f9 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 @@ -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:?}" + ); +} 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..b82f5dfe19 100644 --- a/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs +++ b/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:?}"),