diff --git a/crates/tracedecay-application/src/git_intelligence.rs b/crates/tracedecay-application/src/git_intelligence.rs index 5046bbf93d..ed8ff8af72 100644 --- a/crates/tracedecay-application/src/git_intelligence.rs +++ b/crates/tracedecay-application/src/git_intelligence.rs @@ -1762,6 +1762,14 @@ mod tests { "user.email=fixture@example.com", "-c", "commit.gpgsign=false", + // `git commit` spawns a detached `git maintenance run --auto` + // that holds `.git/objects/maintenance.lock` after the commit + // returns; the byte-identical snapshot must not see it appear + // or vanish between its two walks. + "-c", + "maintenance.auto=false", + "-c", + "gc.auto=0", ]) .args(args) .current_dir(self.path()) diff --git a/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs b/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs index 4df4e34dd5..827cdaece2 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs @@ -17,6 +17,10 @@ fn shipped_binary_stops_quietly_when_a_pipeline_reader_exits() { let output = Command::new("sh") .args(["-c", r#""$TRACEDECAY_BIN" tool | head -n 4"#]) .env("TRACEDECAY_BIN", env!("CARGO_BIN_EXE_tracedecay")) + // A hotpath-enabled binary binds its metrics port on start; when a + // sibling test's daemon already holds it, the bind failure lands on + // stderr and breaks the quiet-pipeline assertion below. + .env("HOTPATH_METRICS_SERVER_OFF", "true") .output() .expect("tracedecay tool pipeline should run"); 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..060c11aca6 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations.rs @@ -899,8 +899,11 @@ fn plan_code_generation_retention_with_verification_cancellable( let Some(file_name) = generation_file_name(&path) else { continue; }; - let (format_revision, manifest, raw_state_digest, size_bytes) = - read_generation_metadata(&path, verification, is_cancelled)?; + let Some((format_revision, manifest, raw_state_digest, size_bytes)) = + read_generation_metadata(&path, verification, is_cancelled)? + else { + continue; + }; let expected_file = format!( "generation-{}.json", sha256_hex_suffix(&raw_state_digest).unwrap_or(&raw_state_digest) @@ -1212,13 +1215,18 @@ fn sweep_unreferenced_generation_segments( })? .to_owned() }; - if read_generation_format_revision(&path, is_cancelled)? - != SEALED_GENERATION_FORMAT_REVISION_V1 - { + let Some(revision) = read_generation_format_revision(&path, is_cancelled)? else { + continue; + }; + if revision != SEALED_GENERATION_FORMAT_REVISION_V1 { continue; } let mut reader = CancellableGenerationManifestReaderV1 { - file: File::open(&path).map_err(storage)?, + file: match File::open(&path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(storage(error)), + }, hasher: Sha256::new(), is_cancelled, cancelled: false, @@ -1281,7 +1289,11 @@ fn sweep_unreferenced_generation_segments( if live_segments.contains(&format!("sha256:{digest}")) { continue; } - let metadata = path.symlink_metadata().map_err(storage)?; + let metadata = match path.symlink_metadata() { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(storage(error)), + }; if !metadata.file_type().is_file() { return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( "generation segment '{}' is not a regular file", 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..2778860571 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 @@ -15,8 +15,14 @@ const MAX_FORMAT_REVISION_PREFIX_BYTES: usize = 4 * 1024; pub(super) fn read_generation_format_revision( path: &Path, is_cancelled: &dyn Fn() -> bool, -) -> Result { - let mut file = File::open(path).map_err(storage)?; +) -> Result, CodeGenerationRetentionErrorV1> { + let mut file = match File::open(path) { + Ok(file) => file, + // The directory entry was removed between listing and open. That is + // concurrent publication, not a broken store. + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(storage(error)), + }; 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); @@ -25,12 +31,14 @@ pub(super) fn read_generation_format_revision( return Err(CodeGenerationRetentionErrorV1::Cancelled); } prefix.truncate(bytes_read); - parse_json_u32_field(&prefix, b"format_revision").ok_or_else(|| { - CodeGenerationRetentionErrorV1::UnsafeState(format!( - "generation file '{}' has no readable format revision in its bounded prefix", - path.display() - )) - }) + parse_json_u32_field(&prefix, b"format_revision") + .ok_or_else(|| { + CodeGenerationRetentionErrorV1::UnsafeState(format!( + "generation file '{}' has no readable format revision in its bounded prefix", + path.display() + )) + }) + .map(Some) } #[hotpath::measure(label = "usecases.retention.read_metadata")] @@ -38,9 +46,15 @@ pub(super) fn read_generation_metadata( path: &Path, verification: GenerationDigestVerificationV1, is_cancelled: &dyn Fn() -> bool, -) -> Result<(u32, SealedGenerationManifestMetadataV1, String, u64), CodeGenerationRetentionErrorV1> -{ - let mut file = File::open(path).map_err(storage)?; +) -> Result< + Option<(u32, SealedGenerationManifestMetadataV1, String, u64)>, + CodeGenerationRetentionErrorV1, +> { + let mut file = match File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(storage(error)), + }; 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); @@ -91,7 +105,7 @@ pub(super) fn read_generation_metadata( } GenerationDigestVerificationV1::MetadataOnly => named_state_digest(path)?, }; - Ok((format_revision, manifest, state_digest, size_bytes)) + Ok(Some((format_revision, manifest, state_digest, size_bytes))) } fn named_state_digest(path: &Path) -> Result { diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs index dc00b2bbb6..a87577aeee 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs @@ -2,14 +2,14 @@ //! //! Quarantined generations are journaled, then hard-linked into the replay pool before the receipt is durable. +#[cfg(test)] +use std::cell::Cell; use std::collections::BTreeSet; use std::fs::File; use std::io::Read; #[cfg(unix)] use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; -#[cfg(test)] -use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Instant; use sha2::{Digest, Sha256}; @@ -264,23 +264,30 @@ pub(super) fn acquire_graph_replay_pool_lock_checked( GraphReplayPoolLockV1::acquire_exclusive(pool_root, deadline, is_cancelled) } +// Per-thread, not process-wide: an acquire runs on its caller's thread, and +// the test harness runs the other acquire tests in parallel on their own +// threads. Shared statics let any concurrent acquire land between a test's +// reset and its read, which is what turned the exact `(1, 0)` proof into an +// occasional `(5, 3)`. #[cfg(test)] -static GRAPH_REPLAY_POOL_ACQUIRE_TRIES: AtomicUsize = AtomicUsize::new(0); -#[cfg(test)] -static GRAPH_REPLAY_POOL_ACQUIRE_WAITS: AtomicUsize = AtomicUsize::new(0); +thread_local! { + static GRAPH_REPLAY_POOL_ACQUIRE_TRIES: Cell = const { Cell::new(0) }; + static GRAPH_REPLAY_POOL_ACQUIRE_WAITS: Cell = const { Cell::new(0) }; +} #[cfg(test)] pub(super) fn reset_graph_replay_pool_acquire_observation() { - GRAPH_REPLAY_POOL_ACQUIRE_TRIES.store(0, Ordering::SeqCst); - GRAPH_REPLAY_POOL_ACQUIRE_WAITS.store(0, Ordering::SeqCst); + GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(|tries| tries.set(0)); + GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(|waits| waits.set(0)); } -/// `(non_blocking_tries, wait_for_exclusive_calls)` since the last reset. +/// `(non_blocking_tries, wait_for_exclusive_calls)` on this thread since the +/// last reset. #[cfg(test)] pub(super) fn graph_replay_pool_acquire_observation() -> (usize, usize) { ( - GRAPH_REPLAY_POOL_ACQUIRE_TRIES.load(Ordering::SeqCst), - GRAPH_REPLAY_POOL_ACQUIRE_WAITS.load(Ordering::SeqCst), + GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(Cell::get), + GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(Cell::get), ) } @@ -307,7 +314,7 @@ impl GraphReplayPoolLockV1 { // the budget is gone. Windows lock-conflict is `Ok(None)` via // `is_lock_contended`, not Storage. #[cfg(test)] - GRAPH_REPLAY_POOL_ACQUIRE_TRIES.fetch_add(1, Ordering::SeqCst); + GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(|tries| tries.set(tries.get() + 1)); match try_acquire_code_generation_store_lock(pool_root)? { Some(guard) => { crate::hotpath_observe::retention_replay_pool_acquired(); @@ -327,7 +334,7 @@ impl GraphReplayPoolLockV1 { fn wait_for_exclusive(deadline: Instant) { #[cfg(test)] - GRAPH_REPLAY_POOL_ACQUIRE_WAITS.fetch_add(1, Ordering::SeqCst); + GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(|waits| waits.set(waits.get() + 1)); crate::hotpath_observe::retention_replay_pool_acquire_wait(); let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { 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..d97a8ecd26 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 @@ -1,7 +1,11 @@ use std::fs::{File, OpenOptions}; use std::path::{Path, PathBuf}; +use std::time::Instant; -use super::{CodeGenerationRetentionErrorV1, SCOPE_RETENTION_LOCK_FILE, STORE_LOCK_FILE, storage}; +use super::{ + CodeGenerationRetentionErrorV1, GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + GRAPH_REPLAY_POOL_ACQUIRE_POLL, SCOPE_RETENTION_LOCK_FILE, STORE_LOCK_FILE, storage, +}; pub struct CodeGenerationStoreLockV1 { file: File, @@ -36,7 +40,33 @@ impl Drop for CodeGenerationStoreLockV1 { pub fn acquire_code_generation_store_lock( store_root: &Path, ) -> Result { - lock_file(store_root, STORE_LOCK_FILE, true) + acquire_code_generation_store_lock_checked( + store_root, + Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + &|| false, + ) +} + +/// Exclusive generation-store lock that stops at `deadline` or cancellation. +/// +/// A free lock is taken even when the deadline has already elapsed, so a +/// caller that only needs one uncontended critical section is not refused. +/// A held lock returns [`CodeGenerationRetentionErrorV1::GenerationStoreBusy`] +/// or [`CodeGenerationRetentionErrorV1::Cancelled`] instead of blocking in +/// `File::lock`, which cannot observe either signal. +pub fn acquire_code_generation_store_lock_checked( + store_root: &Path, + deadline: Instant, + is_cancelled: &dyn Fn() -> bool, +) -> Result { + lock_file_checked( + store_root, + STORE_LOCK_FILE, + true, + deadline, + is_cancelled, + CodeGenerationRetentionErrorV1::GenerationStoreBusy, + ) } /// Try to hold the generation store as a reader for one bounded read of @@ -81,24 +111,63 @@ pub fn try_acquire_code_generation_store_lock( pub(super) fn acquire_scope_retention_lock( store_root: &Path, ) -> Result { - lock_file(store_root, SCOPE_RETENTION_LOCK_FILE, false) + acquire_scope_retention_lock_checked( + store_root, + Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + &|| false, + ) +} + +pub(super) fn acquire_scope_retention_lock_checked( + store_root: &Path, + deadline: Instant, + is_cancelled: &dyn Fn() -> bool, +) -> Result { + lock_file_checked( + store_root, + SCOPE_RETENTION_LOCK_FILE, + false, + deadline, + is_cancelled, + CodeGenerationRetentionErrorV1::GenerationStoreBusy, + ) } #[hotpath::measure(label = "code_index_retention.lock")] -fn lock_file( +fn lock_file_checked( store_root: &Path, lock_file: &str, generation_store: bool, + deadline: Instant, + is_cancelled: &dyn Fn() -> bool, + busy: CodeGenerationRetentionErrorV1, ) -> Result { let store_root = canonical_store_root(store_root)?; - let lock = open_lock_file(&store_root.join(lock_file))?; - lock.lock().map_err(storage)?; - Ok(CodeGenerationStoreLockV1 { - file: lock, - store_root, - generation_store, - shared: false, - }) + let deadline = deadline.min(Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET); + loop { + if is_cancelled() { + return Err(CodeGenerationRetentionErrorV1::Cancelled); + } + let lock = open_lock_file(&store_root.join(lock_file))?; + match lock.try_lock().map_err(std::io::Error::from) { + Ok(()) => { + return Ok(CodeGenerationStoreLockV1 { + file: lock, + store_root, + generation_store, + shared: false, + }); + } + Err(error) if tracedecay_private_fs::is_lock_contended(&error) => { + if Instant::now() >= deadline { + return Err(busy); + } + let remaining = deadline.saturating_duration_since(Instant::now()); + std::thread::park_timeout(remaining.min(GRAPH_REPLAY_POOL_ACQUIRE_POLL)); + } + Err(error) => return Err(storage(error)), + } + } } fn canonical_store_root(store_root: &Path) -> Result { 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..1c02bcc0ac 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 @@ -997,6 +997,52 @@ fn text_artifact_retention_collects_staging_database_sidecars_with_their_owner() ); } +/// The inventory scans the artifact root without the generation-store lock, so +/// the text-artifact builder can retire a `.staging` family between the +/// directory listing and the stat. A vanished entry is already reclaimed and +/// must leave the plan intact rather than failing it with a storage error. +#[test] +fn text_artifact_inventory_skips_an_entry_reclaimed_during_the_scan() { + let store = tempfile::TempDir::new().expect("artifact store"); + let artifacts_root = code_text_artifacts_root(store.path()); + std::fs::create_dir_all(&artifacts_root).expect("create artifact root"); + let staging_family = ["a", "b", "c"] + .into_iter() + .map(|seed| { + let path = artifacts_root.join(format!(".text-artifact-{}.staging", seed.repeat(64))); + std::fs::write(&path, b"staging").expect("write staging evidence"); + path + }) + .collect::>(); + + // The scan probes cancellation once on entry and once per directory entry, + // before it takes that entry. Retiring from the third probe on leaves the + // listing already taken and one entry already inspected, so every further + // name the scan holds names a file that is gone from disk. + let probes = std::sync::atomic::AtomicUsize::new(0); + let retire_during_the_scan = || { + if probes.fetch_add(1, std::sync::atomic::Ordering::Relaxed) >= 2 { + for path in &staging_family { + let _ = std::fs::remove_file(path); + } + } + false + }; + + let inventory = plan_collectable_text_artifacts_cancellable( + store.path(), + None, + GenerationDigestVerificationV1::Full, + &retire_during_the_scan, + ) + .expect("an entry reclaimed mid-scan leaves the store plannable"); + assert!( + inventory.candidates.len() < staging_family.len(), + "an entry that vanished before its stat is reclaimed, not planned: {:?}", + inventory.candidates + ); +} + #[test] fn applied_retention_refuses_a_busy_generation_store_and_retries() { let (store, _) = fixture_store(2); @@ -3072,3 +3118,26 @@ 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 absence, not a storage failure the +/// maintenance tick must treat as a broken store. Any other open failure +/// stays storage. +#[test] +fn vanished_listed_generation_open_is_absent_not_storage_loss() { + let root = tempfile::tempdir().expect("census root"); + let missing = root.path().join(format!("generation-{:064x}.json", 1)); + let opened = super::generation_scan::read_generation_format_revision(&missing, &|| false) + .expect("a vanished listed generation is absent, not a storage failure"); + assert_eq!(opened, None); + + 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-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs index 44f0b28e87..9a0e6f9fb9 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs @@ -1,6 +1,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; +use super::super::locking::acquire_code_generation_store_lock_checked; use super::*; fn ensure_replay_pool(pool_root: &std::path::Path) { @@ -306,6 +307,52 @@ fn checked_acquire_returns_busy_when_the_carried_deadline_has_elapsed() { drop(publisher); } +#[test] +fn store_lock_returns_cancelled_without_waiting_out_the_budget() { + let (_root, store) = isolated_pool(); + let holder = hold_replay_pool(&store); + let started = Instant::now(); + let error = match acquire_code_generation_store_lock_checked( + &store, + Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + &|| true, + ) { + Ok(_) => panic!("cancellation must win a held store lock"), + Err(error) => error, + }; + + assert!(matches!(error, CodeGenerationRetentionErrorV1::Cancelled)); + assert!( + started.elapsed() < Duration::from_millis(20), + "cancelled store lock must not poll the budget, took {:?}", + started.elapsed() + ); + drop(holder); +} + +#[test] +fn store_lock_returns_busy_when_the_carried_deadline_has_elapsed() { + let (_root, store) = isolated_pool(); + let holder = hold_replay_pool(&store); + let started = Instant::now(); + let error = match acquire_code_generation_store_lock_checked(&store, Instant::now(), &|| false) + { + Ok(_) => panic!("an elapsed deadline must defer a held store lock"), + Err(error) => error, + }; + + assert!(matches!( + error, + CodeGenerationRetentionErrorV1::GenerationStoreBusy + )); + assert!( + started.elapsed() < Duration::from_millis(20), + "an expired held store lock must not poll, took {:?}", + started.elapsed() + ); + drop(holder); +} + #[test] fn checked_acquire_takes_a_free_pool_and_releases_without_leak() { let (_root, pool) = isolated_pool(); diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs index c62e73dc92..28a3eed779 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs @@ -394,7 +394,22 @@ pub(super) fn plan_collectable_text_artifacts_cancellable( ) })?; let path = entry.path(); - let metadata = std::fs::symlink_metadata(&path).map_err(storage)?; + // This inventory reads the artifact root without the generation-store + // lock, so an entry the listing just named can already be gone: the + // text-artifact builder retires a `.staging` family (the staging + // database and its `-journal`/`-wal`/`-shm` sidecars) under that lock + // while this scan runs. A vanished entry is reclaimed, which is what + // this inventory would have planned anyway, so it is not a candidate + // and not a failure. Failing the plan here turned every publish that + // raced a maintenance tick into a loud `retention_plan_failed` pass + // (master run 35422072661, `Storage("No such file or directory")`). + // A completed artifact the durable index *references* is verified + // above, before this scan, and stays fail-closed if it disappears. + let metadata = match std::fs::symlink_metadata(&path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(storage(error)), + }; if !metadata.file_type().is_file() { return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( "code text artifact inventory path '{}' is not a regular file", @@ -418,13 +433,15 @@ pub(super) fn plan_collectable_text_artifacts_cancellable( } else { verification }; - verify_unreferenced_completed_text_artifact( + if !verify_unreferenced_completed_text_artifact( &path, digest, metadata.len(), candidate_verification, is_cancelled, - )?; + )? { + continue; + } Some(CodeTextArtifactRetentionCandidateV1 { artifact_file: file_name, kind: CodeTextArtifactRetentionKindV1::Completed, @@ -530,13 +547,19 @@ pub(super) fn verify_completed_text_artifact( is_cancelled: &dyn Fn() -> bool, ) -> Result<(), CodeGenerationRetentionErrorV1> { let digest = sha256_file_component(&descriptor.artifact_digest, "text artifact")?; - verify_unreferenced_completed_text_artifact( + if !verify_unreferenced_completed_text_artifact( path, digest, descriptor.artifact_size_bytes, verification, is_cancelled, - ) + )? { + return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( + "code text artifact '{}' disappeared while its identity was being verified", + path.display() + ))); + } + Ok(()) } /// A content-addressed path is trusted only after the open file and its path @@ -549,15 +572,23 @@ pub(super) fn verify_unreferenced_completed_text_artifact( expected_size_bytes: u64, verification: GenerationDigestVerificationV1, is_cancelled: &dyn Fn() -> bool, -) -> Result<(), CodeGenerationRetentionErrorV1> { - let before = std::fs::symlink_metadata(path).map_err(storage)?; +) -> Result { + let before = match std::fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(storage(error)), + }; if !before.file_type().is_file() || before.len() != expected_size_bytes { return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( "code text artifact '{}' has an invalid regular-file identity", path.display() ))); } - let file = File::open(path).map_err(storage)?; + let file = match File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(storage(error)), + }; if !path_still_names_open_file(path, &file, &before)? { return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( "code text artifact '{}' changed while its identity was being verified", @@ -578,7 +609,7 @@ pub(super) fn verify_unreferenced_completed_text_artifact( path.display() ))); } - Ok(()) + Ok(true) } /// `active_pointer` is the pointer the store carries *now*, which is not @@ -837,13 +868,18 @@ pub(super) fn stage_collectable_text_artifacts_cancellable( } else { GenerationDigestVerificationV1::Full }; - verify_unreferenced_completed_text_artifact( + if !verify_unreferenced_completed_text_artifact( &source, digest, candidate.size_bytes, candidate_verification, is_cancelled, - )?; + )? { + return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( + "text-artifact candidate '{}' disappeared before quarantine", + candidate.artifact_file + ))); + } } if observe_cancel(is_cancelled) { return Err(CodeGenerationRetentionErrorV1::Cancelled); diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs index 138157469d..0ac4e3a71f 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs @@ -4022,7 +4022,9 @@ async fn diagnostics_change_generation_advances_for_out_of_band_git_drift() { async fn elapsed_freshness_window_alone_does_not_make_dashboard_state_stale() { let fixture = GitFixture::new(&[("src/main.rs", "fn main() {}\n")]); let store = TempDir::new().expect("store root"); - let registry = CodeIndexSchedulerRegistryV1::new(1); + // Single-permit admission: holding it below parks the background worker, + // which the host's default bound cannot do. + let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); registry .mount_worktree( test_project_id(), @@ -4033,18 +4035,33 @@ async fn elapsed_freshness_window_alone_does_not_make_dashboard_state_stale() { .expect("mount daemon-owned scheduler"); wait_for_initial_generation(®istry, fixture.path()).await; wait_for_dashboard_ready(®istry, fixture.path()).await; + // The mount leaves clone backfill behind, and the wakes that drain it + // leave a banked permit whose no-op pass projects `Verifying` instead of + // `Fresh` (CI run 35425541839). Settle the mount-era chain, hold the + // admission so no pass can start under the sample, and prove the + // pending-wake slot stays empty, exactly as the text-progress test does. + drain_clone_backfill(®istry, fixture.path()).await; + settled_owner_with_idle_admission(®istry, fixture.path()).await; + let _quiet_owner = quiesced_background_reconcile_admission(®istry, fixture.path()).await; let canonical = fixture.path().canonicalize().expect("canonical fixture"); - { + let scope = { let mounted = registry.mounted.lock().await; - mounted - .get(&canonical) - .expect("mounted worktree") + let worktree = mounted.get(&canonical).expect("mounted worktree"); + worktree .scheduler .lock() .expect("scheduler") .policy .staleness_threshold = Duration::ZERO; - } + tracedecay_contracts::ResolvedScope::new( + test_project_id(), + worktree.repository_id.clone(), + worktree.worktree_id.clone(), + None, + ) + .expect("resolved scope") + }; + clear_pending_wake_until_quiet(®istry, &scope).await; let projected = registry .dashboard_freshness(fixture.path()) diff --git a/crates/tracedecay-daemon-control/src/service/probe.rs b/crates/tracedecay-daemon-control/src/service/probe.rs index 3cec2582bd..c974b6a920 100644 --- a/crates/tracedecay-daemon-control/src/service/probe.rs +++ b/crates/tracedecay-daemon-control/src/service/probe.rs @@ -215,7 +215,9 @@ fn classify_daemon_protocol_identity( match identity { Ok((name, version)) if name.as_deref() == Some("tracedecay") - && version.as_deref() == Some(expected_version) => + && version.as_deref().is_some_and(|version| { + tracedecay_daemon_protocol::versions_name_same_build(version, expected_version) + }) => { DaemonProtocolState::Ready } @@ -638,6 +640,56 @@ fn missing_loopback_authority() -> TraceDecayError { } } +#[cfg(test)] +mod identity_classification_tests { + use super::{DaemonProtocolState, classify_daemon_protocol_identity}; + + const SHA: &str = "84598a0b9c841b914565f46b20bb6c765706e8e5"; + + /// The identity a `tracedecay` daemon reporting `version` answers with. + fn identity(version: &str) -> (Option, Option) { + (Some("tracedecay".to_owned()), Some(version.to_owned())) + } + + /// `tracedecay update` installs a release and then waits for the daemon + /// that binary starts. The release path knows the version it installed as + /// the bare release, while the daemon names the commit it was built from, + /// so readiness used to refuse the very binary it had just installed. + #[test] + fn a_daemon_naming_its_commit_is_ready_against_its_bare_release() { + assert_eq!( + classify_daemon_protocol_identity( + Ok(identity(&format!("0.1.0-beta.47+{SHA}"))), + "0.1.0-beta.47", + ), + DaemonProtocolState::Ready + ); + } + + /// A genuinely stale daemon is still refused, whichever side names a + /// commit. + #[test] + fn a_different_build_is_still_an_identity_mismatch() { + let stale = format!("0.1.0-beta.46+{SHA}"); + assert_eq!( + classify_daemon_protocol_identity(Ok(identity(&stale)), "0.1.0-beta.47"), + DaemonProtocolState::IdentityMismatch { + name: Some("tracedecay".to_owned()), + version: Some(stale), + expected_version: "0.1.0-beta.47".to_owned(), + } + ); + let other_commit = format!("0.1.0-beta.47+{}", "b".repeat(40)); + assert!(matches!( + classify_daemon_protocol_identity( + Ok(identity(&other_commit)), + &format!("0.1.0-beta.47+{SHA}"), + ), + DaemonProtocolState::IdentityMismatch { .. } + )); + } +} + #[cfg(test)] mod timeout_classification_tests { use std::io::{self, Cursor, Read, Write}; diff --git a/crates/tracedecay-daemon-protocol/src/handshake.rs b/crates/tracedecay-daemon-protocol/src/handshake.rs index 231c3ffa85..9bc27c73c6 100644 --- a/crates/tracedecay-daemon-protocol/src/handshake.rs +++ b/crates/tracedecay-daemon-protocol/src/handshake.rs @@ -146,12 +146,41 @@ impl DaemonHandshakeRefusal { /// Old clients send no version (empty string); that is indistinguishable from /// "same version before this field existed", so it never counts as skew. pub fn client_version_skew(client_version: &str, daemon_version: &str) -> Option { - if client_version.is_empty() || client_version == daemon_version { + if client_version.is_empty() || versions_name_same_build(client_version, daemon_version) { return None; } Some(client_version.to_string()) } +/// Whether two reported versions name the same binary, the one comparison +/// every version identity check in the product runs. +/// +/// A version is `"{release}"` or `"{release}+{full sha}[.dirty]"`, and `SemVer` +/// requires build metadata to be ignored for precedence. A side that reports +/// only the release is therefore **less specific**, not different: the release +/// tag `v0.1.0-beta.47` and the binary that names itself +/// `0.1.0-beta.47+` are one identity, and treating them as a mismatch is +/// what failed `tracedecay update`'s own readiness wait against the daemon it +/// had just installed. +/// +/// When both sides do name a commit they must name the same one, which keeps +/// the skew this comparison was added to catch (367a44ad00): two checkout +/// builds of one release differ only by commit, and a daemon left running from +/// the previous build is exactly that case. +#[must_use] +pub fn versions_name_same_build(left: &str, right: &str) -> bool { + let (Some(left_release), Some(right_release)) = (release_version(left), release_version(right)) + else { + // Neither side is a version this comparison understands, so refuse to + // guess and fall back to the literal texts. + return left == right; + }; + left_release.cmp_precedence(&right_release) == std::cmp::Ordering::Equal + && (left_release.build == right_release.build + || left_release.build.is_empty() + || right_release.build.is_empty()) +} + fn release_version(version: &str) -> Option { semver::Version::parse(version.strip_prefix('v').unwrap_or(version)).ok() } @@ -227,6 +256,52 @@ mod handshake_refusal_tests { ); } + /// The observed `tracedecay update` failure: the release path reports the + /// bare release it installed while the daemon that binary starts names its + /// own commit, so readiness compared `0.1.0-beta.47` against + /// `0.1.0-beta.47+` and refused the daemon it had just installed. + #[test] + fn a_bare_release_and_its_own_build_are_one_identity() { + let build = "0.1.0-beta.47+84598a0b9c841b914565f46b20bb6c765706e8e5"; + assert!(versions_name_same_build("0.1.0-beta.47", build)); + assert!(versions_name_same_build(build, "0.1.0-beta.47")); + assert!( + versions_name_same_build("v0.1.0-beta.47", build), + "the GitHub release tag names the same identity as the binary it ships" + ); + assert_eq!(client_version_skew("0.1.0-beta.47", build), None); + assert_eq!(client_version_skew(build, "0.1.0-beta.47"), None); + } + + /// Build metadata still separates two builds of one release, the skew this + /// comparison exists to catch. + #[test] + fn two_commits_of_one_release_stay_distinguishable() { + let older = "0.1.0-beta.47+aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let newer = "0.1.0-beta.47+bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + assert!(!versions_name_same_build(older, newer)); + assert_eq!(client_version_skew(older, newer), Some(older.to_owned())); + assert!(!versions_name_same_build( + "0.1.0-beta.46", + "0.1.0-beta.47+aaaa" + )); + assert!( + !versions_name_same_build( + older, + "0.1.0-beta.47+aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.dirty" + ), + "a dirty worktree is not the commit it was built from" + ); + } + + /// Nothing that fails to parse may be declared a match by accident. + #[test] + fn unparseable_versions_compare_literally() { + assert!(versions_name_same_build("not-a-version", "not-a-version")); + assert!(!versions_name_same_build("not-a-version", "0.1.0-beta.47")); + assert!(!versions_name_same_build("", "0.1.0-beta.47")); + } + #[test] fn foreign_lines_never_parse_as_refusal_frames() { assert_eq!(DaemonHandshakeRefusal::from_line("{}"), None); diff --git a/crates/tracedecay-daemon-protocol/src/lib.rs b/crates/tracedecay-daemon-protocol/src/lib.rs index bc9112b66a..10b4fe309b 100644 --- a/crates/tracedecay-daemon-protocol/src/lib.rs +++ b/crates/tracedecay-daemon-protocol/src/lib.rs @@ -99,6 +99,7 @@ pub use contract::{ pub use handshake::{ DAEMON_HANDSHAKE_REFUSAL_PROTOCOL, DaemonHandshake, DaemonHandshakeRefusal, DaemonHandshakeRefusalReason, MovedStoreAdoption, client_version_skew, version_skew_action, + versions_name_same_build, }; pub use lsp_wire::{ ConnectionLocalRequestSequence, FramePoll, FrameSend, LspFrame, LspSessionAccess, diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index c02f99ca30..4fd33d97cc 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -56,9 +56,9 @@ use tracedecay_store::observation::ObservationIdentityCollisionDispositionV1; use tracedecay_store::{ AnchoredObservationWrite, CursorAdvanceLedgerReasonV1, CursorAdvanceLedgerReceiptIdV1, CursorAdvanceOutcome, ObservationCoverageReason, ObservationCursorAdvance, - ObservationPersistOutcome, - ObservationProjectionStore, ObservationStore, ObservationStoreError, ObservationWrite, - ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, + ObservationPersistOutcome, ObservationProjectionStore, ObservationStore, ObservationStoreError, + ObservationWrite, ProjectionPersistOutcome, ProjectionSkipReason, + SESSION_MESSAGE_PROJECTOR_VERSION, }; use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs index 39200507c3..5072da10d1 100644 --- a/crates/tracedecay-global-db/src/tests.rs +++ b/crates/tracedecay-global-db/src/tests.rs @@ -739,12 +739,19 @@ async fn single_analytics_append_commits_in_one_writer_dispatch() { metadata_json: None, }; - let append = harness.registered.append_analytics_event(&event); - tokio::pin!(append); - assert!(matches!( - futures_util::poll!(&mut append), - std::task::Poll::Pending - )); + // One poll hands the INSERT to the writer thread, which autocommits it and + // only afterwards answers the reply channel. Whether that answer has already + // arrived when the poll returns is a race with that thread, so the poll's own + // result is not the contract; abandoning the future before anyone reads the + // reply is. A single-dispatch autocommit survives that; a transaction-framed + // append would roll back and never reach the inspection connection below. + { + let append = harness.registered.append_analytics_event(&event); + tokio::pin!(append); + if let std::task::Poll::Ready(result) = futures_util::poll!(&mut append) { + result.expect("single analytics append"); + } + } let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1); loop { @@ -758,7 +765,7 @@ async fn single_analytics_append_commits_in_one_writer_dispatch() { } assert!( std::time::Instant::now() < deadline, - "single append did not autocommit while its future remained unpolled" + "single append did not autocommit after its future was abandoned" ); std::thread::sleep(std::time::Duration::from_millis(5)); } diff --git a/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs b/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs index 99d43cfcc3..fa15cd71fc 100644 --- a/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs +++ b/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs @@ -91,13 +91,12 @@ pub async fn handle_field_sites( for site in sites { let line_text = line_at(&source, site.byte).unwrap_or(""); - let enclosing = nodes - .iter() - .filter(|n| { - let line = site.line.saturating_sub(1); - n.metadata.start_line <= line && line <= n.end_line() - }) - .min_by_key(|n| n.metadata.line_span); + // Attribute by byte containment: a read and a write of the + // same field can share one line, and two declarations can + // too, so a line number cannot say which declaration a + // site is inside. Masking preserves byte layout, so the + // offset the scan reports indexes `source` unchanged. + let enclosing = enclosing_declaration(nodes, site.byte as u64); if let Some(scope) = &qualified_scope { if !scope.target_exists { continue; @@ -648,3 +647,79 @@ fn line_is_comment(source: &str, byte: usize) -> bool { let trimmed = line.trim_start(); trimmed.starts_with("//") } + +#[cfg(test)] +mod field_site_attribution_tests { + use super::*; + use tracedecay_domain::{ComplexityAnalysisV1, SourceSpan}; + + fn digest(byte: char) -> T + where + T: TryFrom, + >::Error: std::fmt::Debug, + { + T::try_from(format!("sha256:{}", byte.to_string().repeat(64))).expect("digest") + } + + fn declaration(name: &str, span: std::ops::Range) -> VerifiedAnalysisSymbol { + VerifiedAnalysisSymbol { + occurrence: SymbolOccurrenceId::new(format!("occurrence.{name}")).expect("occurrence"), + path: "src/lib.rs".to_owned(), + source_span: Some(SourceSpan { + start_byte: span.start as u64, + end_byte: span.end as u64, + }), + metadata: LineageSymbolRecordV1 { + occurrence: SymbolOccurrenceId::new(format!("occurrence.{name}")) + .expect("occurrence"), + identity: digest('1'), + qualified_name: name.to_owned(), + simple_name: name.to_owned(), + kind: "function".to_owned(), + visibility: "private".to_owned(), + branches: 0, + loops: 0, + max_nesting: 0, + complexity_analysis: ComplexityAnalysisV1::Complete, + // Both declarations live on line 1: the shape that made + // line-based attribution a coin flip. + line_span: 1, + start_line: 0, + signature: None, + docstring: None, + is_async: false, + derives: Vec::new(), + skip_test_coverage: false, + file_identity: digest('2'), + content_digest: digest('3'), + }, + } + } + + /// A read and a write of one field, inside two functions that share a + /// line, each belong to the function whose bytes contain them. Every + /// candidate has `line_span == 1` here, so the old smallest-line-span + /// selection had nothing to break the tie with and returned whichever + /// symbol the graph page happened to yield first. + #[test] + fn attributes_a_read_and_a_write_sharing_one_line() { + let source = "fn r(s: &S) -> u32 { s.count } fn w(s: &mut S) { s.count = 1; }"; + let write_start = source.find("fn w").expect("second function"); + let nodes = vec![ + declaration("r", 0..write_start), + declaration("w", write_start..source.len()), + ]; + + let sites = find_field_references(source, "count"); + assert_eq!(sites.len(), 2, "one read and one write: {sites:?}"); + assert!(matches!(sites[0].kind, FieldRefKind::Read)); + assert!(matches!(sites[1].kind, FieldRefKind::Write)); + assert_eq!(sites[0].line, sites[1].line, "both sites share one line"); + + for (site, expected) in sites.iter().zip(["r", "w"]) { + let enclosing = enclosing_declaration(&nodes, site.byte as u64) + .map(|node| node.metadata.qualified_name.as_str()); + assert_eq!(enclosing, Some(expected), "site {site:?}"); + } + } +} diff --git a/crates/tracedecay-mcp/src/handlers/analysis/mod.rs b/crates/tracedecay-mcp/src/handlers/analysis/mod.rs index e3f3ffe05b..dcd9a84d82 100644 --- a/crates/tracedecay-mcp/src/handlers/analysis/mod.rs +++ b/crates/tracedecay-mcp/src/handlers/analysis/mod.rs @@ -43,7 +43,7 @@ use tracedecay_code_index::graph_projection::CodeGraphSemanticEdgeV1; use tracedecay_code_index::lineage::LineageSymbolRecordV1; use tracedecay_domain::code_intelligence::NodeKind; use tracedecay_domain::errors::{Result, TraceDecayError}; -use tracedecay_domain::{RelationEdgeKindV1, SymbolOccurrenceId}; +use tracedecay_domain::{RelationEdgeKindV1, SourceSpan, SymbolOccurrenceId}; use tracedecay_graph_query::VerifiedGraphQuery; fn path_is_rust(path: &str) -> bool { @@ -63,6 +63,9 @@ const ANALYSIS_RELATION_BUDGET: usize = 2_000_000; struct VerifiedAnalysisSymbol { occurrence: SymbolOccurrenceId, path: String, + /// Byte range the declaration occupies in its file. Line numbers cannot + /// separate two declarations that share one line; this can. + source_span: Option, metadata: LineageSymbolRecordV1, } @@ -74,6 +77,25 @@ impl VerifiedAnalysisSymbol { } } +/// Innermost declaration whose source range covers `match_byte`. +/// +/// Byte containment, not line containment: an attribute such as `#[test]` and +/// the two functions on `#[test] fn a() {…} fn b() {…}` all sit on one line, +/// and only the byte range says which of them the site is inside. Selecting by +/// line also had no stable order to break ties with, since symbols arrive in +/// occurrence order and occurrence ids are per-project digests. +fn enclosing_declaration( + nodes: &[VerifiedAnalysisSymbol], + match_byte: u64, +) -> Option<&VerifiedAnalysisSymbol> { + nodes + .iter() + .filter_map(|node| node.source_span.map(|span| (node, span))) + .filter(|(_, span)| span.start_byte <= match_byte && match_byte < span.end_byte) + .min_by_key(|(_, span)| span.end_byte.saturating_sub(span.start_byte)) + .map(|(node, _)| node) +} + fn verified_analysis_symbols( graph: &VerifiedGraphQuery, scope_prefix: Option<&str>, @@ -89,6 +111,10 @@ fn verified_analysis_symbols( page.symbols .into_iter() .map(|symbol| { + let source_span = symbol + .binding + .as_ref() + .and_then(|binding| binding.source_span); let path = symbol .binding .and_then(|binding| binding.logical_path) @@ -109,6 +135,7 @@ fn verified_analysis_symbols( Ok(VerifiedAnalysisSymbol { occurrence: symbol.occurrence, path, + source_span, metadata, }) }) diff --git a/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs b/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs index 920b28a21c..c0d587c1e8 100644 --- a/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs +++ b/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs @@ -8,7 +8,8 @@ use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_graph_query::VerifiedGraphQuery; use super::{ - VerifiedAnalysisSymbol, path_is_rust, verified_analysis_symbols, verified_analysis_unavailable, + VerifiedAnalysisSymbol, enclosing_declaration, path_is_rust, verified_analysis_symbols, + verified_analysis_unavailable, }; use crate::ToolResult; use crate::handlers::support::{effective_path, rendered_tool_result}; @@ -41,23 +42,28 @@ fn source_may_contain_unsafe_kind(source: &str, kind: &str) -> bool { } } -fn line_matches_unsafe_kind(line: &str, kind: &str) -> bool { +/// Byte offset of the risky construct within `line`, when the line has one. +/// +/// The offset is what lets a match be attributed to the declaration that +/// actually contains it: two declarations can share a line, so a line number +/// alone cannot say which one a site belongs to. +fn line_matches_unsafe_kind(line: &str, kind: &str) -> Option { let trimmed = line.trim_start(); if trimmed.starts_with("//") || trimmed.starts_with("///") { - return false; + return None; } match kind { "unwrap" => contains_method_call(line, "unwrap", true), "expect" => contains_method_call(line, "expect", false), - "panic" => line.contains("panic!("), - "todo" => line.contains("todo!("), - "unimplemented" => line.contains("unimplemented!("), + "panic" => line.find("panic!("), + "todo" => line.find("todo!("), + "unimplemented" => line.find("unimplemented!("), "unsafe_block" => contains_unsafe_block_start(line), - _ => false, + _ => None, } } -fn contains_method_call(line: &str, method: &str, empty_parens: bool) -> bool { +fn contains_method_call(line: &str, method: &str, empty_parens: bool) -> Option { let needle = format!(".{method}"); let bytes = line.as_bytes(); let mut start = 0usize; @@ -69,18 +75,18 @@ fn contains_method_call(line: &str, method: &str, empty_parens: bool) -> bool { if is_word_boundary && next == Some(b'(') { if empty_parens { if line[after + 1..].trim_start().starts_with(')') { - return true; + return Some(abs); } } else { - return true; + return Some(abs); } } start = abs + needle.len(); } - false + None } -fn contains_unsafe_block_start(line: &str) -> bool { +fn contains_unsafe_block_start(line: &str) -> Option { let bytes = line.as_bytes(); let mut start = 0usize; while let Some(pos) = line[start..].find("unsafe") { @@ -97,12 +103,12 @@ fn contains_unsafe_block_start(line: &str) -> bool { || rest.starts_with("impl ") || rest.starts_with("trait ") { - return true; + return Some(abs); } } start = abs + "unsafe".len(); } - false + None } fn path_looks_like_test(path: &str) -> bool { @@ -216,7 +222,16 @@ pub async fn handle_unsafe_patterns( // Masking can erase every raw hit (all of them in comments or // string literals), so the file's nodes are fetched only once a // real match survives. - for (idx, (line, masked_line)) in source.lines().zip(masked.lines()).enumerate() { + // Split inclusively so each line keeps its own byte offset; + // masking preserves byte layout, so the two sides stay aligned. + let mut line_start = 0usize; + for (idx, (line, masked_line)) in source + .split_inclusive('\n') + .zip(masked.split_inclusive('\n')) + .enumerate() + { + let line_offset = line_start; + line_start += line.len(); let line_no = (idx as u32) + 1; // A mixed test/production line is not wholly test scope, // so keep its production risk visible. @@ -225,16 +240,11 @@ pub async fn handle_unsafe_patterns( continue; } for kind in &kinds { - if line_matches_unsafe_kind(masked_line, kind) { + if let Some(column) = line_matches_unsafe_kind(masked_line, kind) { let nodes = symbols_by_file.get(file).map_or(&[][..], Vec::as_slice); - let enclosing = nodes - .iter() - .filter(|n| { - n.metadata.start_line.saturating_add(1) <= line_no - && line_no <= n.end_line().saturating_add(1) - }) - .min_by_key(|n| n.metadata.line_span) - .map(|n| n.metadata.qualified_name.clone()); + let enclosing = + enclosing_declaration(nodes, (line_offset + column) as u64) + .map(|node| node.metadata.qualified_name.clone()); *by_kind.entry(kind.clone()).or_insert(0) += 1; matches.push(json!({ "kind": kind, @@ -314,7 +324,7 @@ mod unsafe_pattern_detection_tests { for line in lines { for kind in kinds { - if line_matches_unsafe_kind(line, kind) { + if line_matches_unsafe_kind(line, kind).is_some() { assert!( source_may_contain_unsafe_kind(line, kind), "prefilter would drop a real {kind} site: {line:?}" @@ -337,47 +347,46 @@ mod unsafe_pattern_detection_tests { fn detects_unsafe_block_inside_safe_fn() { // An `unsafe { }` block living inside an otherwise-safe function, the // exact shape the audit fixture plants. - assert!(line_matches_unsafe_kind( - " unsafe { *ptr as usize }", - "unsafe_block" - )); - assert!(contains_unsafe_block_start(" unsafe { *ptr as usize }")); + assert!(line_matches_unsafe_kind(" unsafe { *ptr as usize }", "unsafe_block").is_some()); + assert!(contains_unsafe_block_start(" unsafe { *ptr as usize }").is_some()); } #[test] fn detects_unsafe_fn_impl_and_trait() { - assert!(line_matches_unsafe_kind( - "pub unsafe fn raw(&self) {", - "unsafe_block" - )); - assert!(line_matches_unsafe_kind( - "unsafe impl Send for Foo {}", - "unsafe_block" - )); - assert!(line_matches_unsafe_kind( - "unsafe trait Zeroable {}", - "unsafe_block" - )); + assert!(line_matches_unsafe_kind("pub unsafe fn raw(&self) {", "unsafe_block").is_some()); + assert!(line_matches_unsafe_kind("unsafe impl Send for Foo {}", "unsafe_block").is_some()); + assert!(line_matches_unsafe_kind("unsafe trait Zeroable {}", "unsafe_block").is_some()); } #[test] fn ignores_safe_code_and_comments() { // Plain safe code has no unsafe markers. - assert!(!line_matches_unsafe_kind( - "let x = total as usize;", - "unsafe_block" - )); + assert!(line_matches_unsafe_kind("let x = total as usize;", "unsafe_block").is_none()); // The word appears only in a comment/doc line: not a real unsafe site. - assert!(!line_matches_unsafe_kind( - "// this is not unsafe { } really", - "unsafe_block" - )); - assert!(!line_matches_unsafe_kind( - "/// drop the needless unsafe block", - "unsafe_block" - )); + assert!( + line_matches_unsafe_kind("// this is not unsafe { } really", "unsafe_block").is_none() + ); + assert!( + line_matches_unsafe_kind("/// drop the needless unsafe block", "unsafe_block") + .is_none() + ); // A substring of a longer identifier must not trip the word-boundary check. - assert!(!contains_unsafe_block_start("let unsafely = 1;")); - assert!(!contains_unsafe_block_start("let make_unsafe_thing = 2;")); + assert!(contains_unsafe_block_start("let unsafely = 1;").is_none()); + assert!(contains_unsafe_block_start("let make_unsafe_thing = 2;").is_none()); + } + + /// The reported offset is what attributes a site to a declaration, so it + /// has to point at the construct itself, not at the start of the line. + #[test] + fn reports_where_on_the_line_the_site_is() { + let line = "#[test] fn a() { Some(5).unwrap(); } pub fn b() { panic!(); }"; + assert_eq!( + line_matches_unsafe_kind(line, "unwrap"), + Some(line.find(".unwrap()").expect("unwrap call")) + ); + assert_eq!( + line_matches_unsafe_kind(line, "panic"), + Some(line.find("panic!(").expect("panic call")) + ); } } diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs index 922ea3552b..bb09cccdfa 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs @@ -1,3 +1,4 @@ +use std::collections::{HashMap, HashSet}; use std::io; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -60,6 +61,7 @@ impl CodeLexicalCloneSuccessorV1 { memory_budget_bytes: usize, ) -> Result { let connection = open_builder_connection(staging_path, memory_budget_bytes)?; + ensure_clone_occurrence_indexes(&connection)?; let mutation_gate = register_builder_mutation_gate(&connection)?; let (prior_digest, format_revision): (String, i64) = connection .query_row( @@ -430,6 +432,24 @@ fn reset_clone_tables(connection: &Connection) -> Result<(), CodeLexicalArtifact .map_err(sqlite_error) } +/// Lookup indexes for resume verification, which reads postings by +/// occurrence. Both postings tables are keyed from `class`/`language`, so +/// without these every per-occurrence read is a full table scan; replaying N +/// committed pages after a restart then costs N scans of every posting the +/// repository has, and a daemon sat inside that replay for hours. A prior +/// artifact copied from a build that predates the indexes gains them here, +/// and nothing digests or enumerates the index schema. +fn ensure_clone_occurrence_indexes( + connection: &Connection, +) -> Result<(), CodeLexicalArtifactErrorV1> { + connection + .execute_batch( + "CREATE INDEX IF NOT EXISTS clone_exact_postings_by_occurrence ON clone_exact_postings(symbol_occurrence_id); + CREATE INDEX IF NOT EXISTS clone_fingerprint_postings_by_occurrence ON clone_fingerprint_postings(symbol_occurrence_id);", + ) + .map_err(sqlite_error) +} + fn append_clone_rows( transaction: &rusqlite::Transaction<'_>, page: &VerifiedSealedLexicalPageV1, @@ -559,11 +579,106 @@ fn verify_copied_source_page( Ok(()) } +type CloneExactRowV1 = (i64, i64, String, String); + +/// Postings for one page's occurrences, read through the occurrence indexes +/// `ensure_clone_occurrence_indexes` installs and bucketed by occurrence. +/// +/// The postings tables are keyed from `class`/`language`; before those +/// indexes existed a per-occurrence read scanned the whole table, and a +/// daemon spent eleven hours replaying committed pages after a restart. +struct ClonePagePostingsV1 { + exact: HashMap>, + fingerprints: HashMap>, +} + +impl ClonePagePostingsV1 { + fn read( + connection: &Connection, + occurrences: &HashSet<&str>, + control: &dyn CodeIndexExecutionControlV1, + ) -> Result { + let mut exact: HashMap> = HashMap::new(); + let mut statement = connection + .prepare( + "SELECT class, normalization_revision, digest, payload_digest FROM clone_exact_postings WHERE symbol_occurrence_id = ?1", + ) + .map_err(sqlite_error)?; + for occurrence in occurrences { + checkpoint(control)?; + let mut rows = statement.query([occurrence]).map_err(sqlite_error)?; + while let Some(row) = rows.next().map_err(sqlite_error)? { + exact.entry((*occurrence).to_owned()).or_default().push(( + row.get(0).map_err(sqlite_error)?, + row.get(1).map_err(sqlite_error)?, + row.get(2).map_err(sqlite_error)?, + row.get(3).map_err(sqlite_error)?, + )); + } + } + drop(statement); + + let mut fingerprints: HashMap> = HashMap::new(); + let mut statement = connection + .prepare( + "SELECT language, class, normalization_revision, fingerprint, token_position, payload_digest, body_digest FROM clone_fingerprint_postings WHERE symbol_occurrence_id = ?1", + ) + .map_err(sqlite_error)?; + for occurrence in occurrences { + checkpoint(control)?; + let mut rows = statement.query([occurrence]).map_err(sqlite_error)?; + while let Some(row) = rows.next().map_err(sqlite_error)? { + fingerprints + .entry((*occurrence).to_owned()) + .or_default() + .push(( + row.get(0).map_err(sqlite_error)?, + row.get(1).map_err(sqlite_error)?, + row.get(2).map_err(sqlite_error)?, + row.get(3).map_err(sqlite_error)?, + row.get(4).map_err(sqlite_error)?, + row.get(5).map_err(sqlite_error)?, + row.get(6).map_err(sqlite_error)?, + )); + } + } + drop(statement); + checkpoint(control)?; + + // Each table's primary key is unique within one occurrence, so sorting + // a bucket reproduces the `ORDER BY` the per-body queries used. + for rows in exact.values_mut() { + rows.sort(); + } + for rows in fingerprints.values_mut() { + rows.sort(); + } + Ok(Self { + exact, + fingerprints, + }) + } + + fn exact_for(&self, occurrence: &str) -> &[CloneExactRowV1] { + self.exact.get(occurrence).map_or(&[], Vec::as_slice) + } + + fn fingerprints_for(&self, occurrence: &str) -> &[CloneFingerprintRowV1] { + self.fingerprints.get(occurrence).map_or(&[], Vec::as_slice) + } +} + fn verify_clone_page_rows( connection: &Connection, page: &VerifiedSealedLexicalPageV1, control: &dyn CodeIndexExecutionControlV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { + let occurrences = page + .clone_bodies() + .iter() + .map(|body| body.occurrence.symbol_occurrence_id.as_str()) + .collect::>(); + let postings = ClonePagePostingsV1::read(connection, &occurrences, control)?; for body in page.clone_bodies() { checkpoint(control)?; let expected_payload = serde_json::to_vec(&body.payload) @@ -631,24 +746,12 @@ fn verify_clone_page_rows( ) }) .collect::>(); - let mut statement = connection - .prepare( - "SELECT class, normalization_revision, digest, payload_digest FROM clone_exact_postings WHERE symbol_occurrence_id = ?1 ORDER BY class, normalization_revision, digest", - ) - .map_err(sqlite_error)?; - let stored_postings = statement - .query_map([body.occurrence.symbol_occurrence_id.as_str()], |row| { - Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; - if stored_postings != expected_postings { + if postings.exact_for(body.occurrence.symbol_occurrence_id.as_str()) != expected_postings { return Err(CodeLexicalArtifactErrorV1::Corrupt( "resumed clone postings differ from their sealed source page".to_owned(), )); } - verify_clone_fingerprint_page_rows(connection, body)?; + verify_clone_fingerprint_page_rows(&postings, body)?; } Ok(()) } @@ -656,7 +759,7 @@ fn verify_clone_page_rows( type CloneFingerprintRowV1 = (String, i64, i64, i64, i64, String, String); fn verify_clone_fingerprint_page_rows( - connection: &Connection, + postings: &ClonePagePostingsV1, body: &CodeIndexCloneBodyV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { let mut expected = Vec::new(); @@ -679,26 +782,7 @@ fn verify_clone_fingerprint_page_rows( } } expected.sort(); - let mut statement = connection - .prepare( - "SELECT language, class, normalization_revision, fingerprint, token_position, payload_digest, body_digest FROM clone_fingerprint_postings WHERE symbol_occurrence_id = ?1 ORDER BY language, class, normalization_revision, fingerprint, token_position", - ) - .map_err(sqlite_error)?; - let stored = statement - .query_map([body.occurrence.symbol_occurrence_id.as_str()], |row| { - Ok(( - row.get(0)?, - row.get(1)?, - row.get(2)?, - row.get(3)?, - row.get(4)?, - row.get(5)?, - row.get(6)?, - )) - }) - .map_err(sqlite_error)? - .collect::, _>>() - .map_err(sqlite_error)?; + let stored = postings.fingerprints_for(body.occurrence.symbol_occurrence_id.as_str()); if stored != expected { return Err(CodeLexicalArtifactErrorV1::Corrupt( "resumed clone fingerprints differ from their sealed source page".to_owned(), diff --git a/crates/tracedecay-runtime-core/src/db/engine/error.rs b/crates/tracedecay-runtime-core/src/db/engine/error.rs index 2955a7610e..d2570b249a 100644 --- a/crates/tracedecay-runtime-core/src/db/engine/error.rs +++ b/crates/tracedecay-runtime-core/src/db/engine/error.rs @@ -95,6 +95,12 @@ impl From for Error { ExactSqlError::RequestLimitExceeded => { Self::InvalidOperation("SQL request exceeds migration limits".to_owned()) } + // A materialization ceiling is a property of the submitted + // statement, not of the engine: the untyped `Runtime` fallback + // made callers read it as a transient storage fault and replay it. + ExactSqlError::QueryLimitExceeded => Self::InvalidOperation( + "exact SQL query materialization exceeded its limit".to_owned(), + ), ExactSqlError::AuthorityDenied(message) => Self::InvalidOperation(message), ExactSqlError::Sqlite { operation, @@ -142,4 +148,23 @@ impl Error { _ => None, } } + + /// True when replaying this exact statement can never succeed. + /// + /// A `SQLITE_CONSTRAINT` abort is a schema-contract trigger or constraint + /// refusing this exact row, and `InvalidOperation` is an admission or + /// materialization ceiling refusing this exact statement. Neither is a + /// transient engine condition, so a caller that retries one spins until + /// something else changes the durable state. + #[hotpath::skip] + pub const fn is_deterministic_refusal(&self) -> bool { + match self { + Self::InvalidOperation(_) => true, + Self::StatementBatch { source, .. } => source.is_deterministic_refusal(), + _ => matches!(self.sqlite_code(), Some(SQLITE_CONSTRAINT)), + } + } } + +/// `SQLITE_CONSTRAINT`: a constraint or `RAISE(ABORT)` trigger refused the row. +const SQLITE_CONSTRAINT: i32 = 19; diff --git a/crates/tracedecay-runtime-core/src/db/engine/tests.rs b/crates/tracedecay-runtime-core/src/db/engine/tests.rs index a3c02d8f47..5e51bd0a70 100644 --- a/crates/tracedecay-runtime-core/src/db/engine/tests.rs +++ b/crates/tracedecay-runtime-core/src/db/engine/tests.rs @@ -386,3 +386,43 @@ async fn transaction_statement_batch_reports_the_exact_failed_statement() { } mod async_writer; + +#[test] +fn deterministic_refusals_are_distinguished_from_transient_engine_faults() { + use tracedecay_rusqlite_runtime::exact_sql::ExactSqlError; + + // A schema-contract trigger abort refuses this exact row for good. + let constraint = Error::Sqlite { + operation: "execute", + code: Some(19), + extended_code: Some(1811), + message: "invalid session refresh progress".to_owned(), + }; + assert!(constraint.is_deterministic_refusal()); + assert!( + Error::StatementBatch { + index: 0, + source: Box::new(constraint), + } + .is_deterministic_refusal() + ); + + // A materialization ceiling refuses this exact statement for good, and + // must not arrive as an untyped `Runtime` message. + let limit = Error::from(ExactSqlError::QueryLimitExceeded); + assert!(matches!(limit, Error::InvalidOperation(_))); + assert!(limit.is_deterministic_refusal()); + + // Contention and I/O faults stay retryable. + assert!(!Error::Busy.is_deterministic_refusal()); + assert!(!Error::Runtime("writer restarted".to_owned()).is_deterministic_refusal()); + assert!( + !Error::Sqlite { + operation: "execute", + code: Some(5), + extended_code: Some(5), + message: "database is locked".to_owned(), + } + .is_deterministic_refusal() + ); +} diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs b/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs index c101e90f35..a8a90bf26d 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs @@ -1,5 +1,5 @@ use std::collections::HashMap; -use std::time::Duration; +use std::time::{Duration, Instant}; use tracedecay_store::{ GraphDependencyGenerationIdentityV1, GraphGenerationIdV1, GraphNamespaceV1, @@ -28,8 +28,6 @@ use super::{ REPLAY_READER_ACQUIRE_SLICE, TOMBSTONE_COLUMNS, }; -const BEGIN_BUSY_ATTEMPT_BUDGET: u32 = 64; - /// Maximum owner sequences bound into one `IN (...)` dependency lookup. /// /// This is **not** `REFERENCED_ANCHOR_BATCH` from @@ -59,37 +57,66 @@ const BEGIN_BUSY_ATTEMPT_BUDGET: u32 = 64; /// exceeded the row cap in one chunk. const GRAPH_REPLAY_DEPENDENCY_BATCH: usize = 38; -#[hotpath::measure(label = "rusqlite.graph_publication.begin")] -pub(super) fn begin( - handle: &ExactSqlHandle, +/// Wall-clock budget for one begin acquisition, retries included. +/// +/// `ExactSqlError::Busy` answers two different questions with one variant: the +/// exact-SQL command queue was full (`map_writer_send_error`), or the writer's +/// own `EXACT_SQL_WRITE_LOCK_ACQUIRE_LIMIT` lock loop was exhausted. Only the +/// first is worth another attempt. A wall-clock budget separates them without a +/// second error variant: an exhausted lock attempt has already spent this whole +/// window inside `begin_immediate`, so it gets exactly one attempt and its +/// 64ms answer is never multiplied into seconds, while a queue refusal returns +/// at once and still has the window to drain in. +const BEGIN_ACQUIRE_BUDGET: Duration = Duration::from_millis(64); + +/// Pause between admission retries, matching the writer's own busy pause. +const BEGIN_BUSY_RETRY_PAUSE: Duration = Duration::from_millis(1); + +/// Runs `attempt` until it answers, the budget is spent, or the caller is +/// interrupted. `Ok(None)` means no answer within the budget; the caller owns +/// what that means for its own operation. +fn acquire_within_begin_budget( context: &GraphPublicationOperationContextV1<'_>, -) -> GraphPublicationStoreResultV1 { - let mut busy_attempts = 0_u32; + mut attempt: impl FnMut() -> Result, +) -> GraphPublicationStoreResultV1> { + let deadline = Instant::now() + BEGIN_ACQUIRE_BUDGET; loop { ensure_not_interrupted(context)?; - match hotpath::measure_block!("rusqlite.graph_publication.begin_immediate", { - handle.begin_immediate() - }) { - Ok(transaction) => { + match attempt() { + Ok(value) => { ensure_not_interrupted(context)?; - return Ok(transaction); + return Ok(Some(value)); } Err(ExactSqlError::Busy) => { - busy_attempts = busy_attempts.saturating_add(1); - if busy_attempts >= BEGIN_BUSY_ATTEMPT_BUDGET { - return Err(GraphPublicationStoreErrorV1::Infrastructure); + if let Some(reason) = context.interruption() { + return Err(GraphPublicationStoreErrorV1::Interrupted(reason)); } - std::thread::sleep(Duration::from_millis(1)); - ensure_not_interrupted(context)?; + if Instant::now() >= deadline { + return Ok(None); + } + std::thread::sleep(BEGIN_BUSY_RETRY_PAUSE); } Err(_) => { ensure_not_interrupted(context)?; - return Err(GraphPublicationStoreErrorV1::Infrastructure); + return Ok(None); } } } } +#[hotpath::measure(label = "rusqlite.graph_publication.begin")] +pub(super) fn begin( + handle: &ExactSqlHandle, + context: &GraphPublicationOperationContextV1<'_>, +) -> GraphPublicationStoreResultV1 { + acquire_within_begin_budget(context, || { + hotpath::measure_block!("rusqlite.graph_publication.begin_immediate", { + handle.begin_immediate() + }) + })? + .ok_or(GraphPublicationStoreErrorV1::Infrastructure) +} + pub(super) fn ensure_owner( handle: &ExactSqlHandle, projection: &GraphProjectionIdentityV1, @@ -128,30 +155,16 @@ pub(super) fn begin_read( handle: &ExactSqlHandle, context: &GraphPublicationOperationContextV1<'_>, ) -> GraphPublicationStoreResultV1 { - let mut busy_attempts = 0_u32; - loop { - ensure_not_interrupted(context)?; - match hotpath::measure_block!("rusqlite.graph_publication.begin_read_snapshot", { + if let Some(snapshot) = acquire_within_begin_budget(context, || { + hotpath::measure_block!("rusqlite.graph_publication.begin_read_snapshot", { handle.begin_read_snapshot(REPLAY_READER_ACQUIRE_SLICE) - }) { - Ok(snapshot) => { - ensure_not_interrupted(context)?; - return Ok(ExactPublicationRead::Snapshot(snapshot)); - } - Err(ExactSqlError::Busy) => { - busy_attempts = busy_attempts.saturating_add(1); - if busy_attempts >= BEGIN_BUSY_ATTEMPT_BUDGET { - break; - } - std::thread::sleep(Duration::from_millis(1)); - ensure_not_interrupted(context)?; - } - Err(_) => { - ensure_not_interrupted(context)?; - break; - } - } + }) + })? { + return Ok(ExactPublicationRead::Snapshot(snapshot)); } + // The deferred fallback waits the writer without consulting `context`, so + // this is the last point that can answer a cancelled or expired caller. + ensure_not_interrupted(context)?; hotpath::measure_block!("rusqlite.graph_publication.begin_deferred", { handle .begin_deferred() diff --git a/crates/tracedecay-session-memory/src/session/retrieval.rs b/crates/tracedecay-session-memory/src/session/retrieval.rs index 2d35c62ea6..da4653155a 100644 --- a/crates/tracedecay-session-memory/src/session/retrieval.rs +++ b/crates/tracedecay-session-memory/src/session/retrieval.rs @@ -565,6 +565,19 @@ fn map_execution_error( SessionTemporalExecutionError::Denied => SessionRetrievalOutcome::Denied, SessionTemporalExecutionError::Unavailable => SessionRetrievalOutcome::Unavailable, SessionTemporalExecutionError::ResetRequired => SessionRetrievalOutcome::ResetRequired, + // A partial generation is a projection still converging, not an + // authoritative empty root: answering `CompleteZero` there publishes + // "nothing exists, and that is final" for rows the store has already + // committed but not yet published. `map_report` refuses that for an + // empty ranked page; the execution-error path owes the same refusal, + // so the caller re-reads instead of believing the zero. + SessionTemporalExecutionError::Empty { + freshness: freshness @ SessionDataFreshness::Partial { generation_lag }, + } => SessionRetrievalOutcome::Partial { + items: Vec::new(), + freshness, + omitted: generation_lag.max(1), + }, SessionTemporalExecutionError::Empty { freshness } => { SessionRetrievalOutcome::CompleteZero { freshness } } @@ -1039,6 +1052,19 @@ mod tests { ); } + #[test] + fn partial_generation_empty_execution_is_never_an_authoritative_zero() { + let freshness = SessionDataFreshness::Partial { generation_lag: 1 }; + assert_eq!( + map_execution_error(SessionTemporalExecutionError::Empty { freshness }), + SessionRetrievalOutcome::Partial { + items: Vec::new(), + freshness, + omitted: 1, + } + ); + } + #[test] fn persisted_reset_and_unavailable_remain_distinct() { assert_eq!( diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs index ed7fcde8e0..365f4d76f1 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs @@ -25,6 +25,7 @@ use super::wake::{ SessionTemporalRefreshWakeState, TerminalAttemptGuard, }; use tracedecay_global_db::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1}; +use tracedecay_runtime_core::db::engine::Error as EngineError; use tracedecay_session_temporal_store::{ SessionRefreshRecoveryV1, SessionRefreshRestartStateV1, SessionTemporalStore, }; @@ -634,11 +635,34 @@ async fn session_projection_refresh( run_session_temporal_refresh_pass(database, state, projector, policy).await } -fn classify_store_error(error: &SessionStoreError) -> SessionTemporalRefreshRetryClass { - if error.is_storage() { - SessionTemporalRefreshRetryClass::Storage - } else { - SessionTemporalRefreshRetryClass::Projector +/// True when replaying this store failure unchanged could still succeed. +/// +/// `is_storage` only says the failure came from the storage adapter; it does +/// not say the failure is transient. A schema-contract trigger refusing the +/// submitted row, or an exact-SQL ceiling refusing the submitted statement, is +/// deterministic: the worker resubmits the identical request every pass, so +/// treating it as retryable is an unbounded spin at the backoff cap rather +/// than a recovery. Those are terminal, and the caller durably fails the +/// refresh instead of retrying it. +fn is_retryable_storage(error: &SessionStoreError) -> bool { + matches!(error, SessionStoreError::Storage { .. }) && !is_deterministic_refusal(error) +} + +/// True when the durable contract refused the exact submitted row or +/// statement: a typed store refusal, or an engine failure that replays +/// identically. Only such a refusal retires a running refresh. An +/// interrupted pass (cancelled control, deadline, budget) and transient +/// storage leave the operation for the next pass, which may hold a +/// different control. +fn is_deterministic_refusal(error: &SessionStoreError) -> bool { + match error { + SessionStoreError::Cancelled + | SessionStoreError::DeadlineExceeded + | SessionStoreError::BudgetExceeded { .. } => false, + SessionStoreError::Storage { source, .. } => source + .downcast_ref::() + .is_some_and(EngineError::is_deterministic_refusal), + _ => true, } } @@ -667,7 +691,7 @@ pub async fn process_refresh_begin_requests( tracedecay_store::SessionRefreshDispositionV1::Joined => report.joined += 1, } } - Err(error) if error.is_storage() => { + Err(error) if is_retryable_storage(&error) => { report.last_error = Some(format!("{error:?}")); report.retryable_errors += 1; report.observe_retry(SessionTemporalRefreshRetryClass::Storage); @@ -709,7 +733,7 @@ pub async fn begin_admitted_session_refreshes( { Ok(page) => page, Err(error) => { - if classify_store_error(&error) == SessionTemporalRefreshRetryClass::Storage { + if is_retryable_storage(&error) { report.last_error = Some(format!("{error:?}")); report.retryable_errors += 1; report.observe_retry(SessionTemporalRefreshRetryClass::Storage); @@ -767,7 +791,7 @@ async fn complete_ready_refresh( Ok(_) => { report.completed += 1; } - Err(error) if error.is_storage() => { + Err(error) if is_retryable_storage(&error) => { report.last_error = Some(format!("{error:?}")); report.retryable_errors += 1; report.observe_retry(SessionTemporalRefreshRetryClass::Storage); @@ -796,6 +820,48 @@ fn record_projector_error( } } +/// Typed failure recorded when the durable contract refuses the projected +/// progress row. It is not a projector fault: the row was well formed for the +/// state the projector read, and the durable state disagrees. +const REFRESH_PROGRESS_REFUSED: &str = "refresh_progress_refused"; + +/// Builds the durable failure request that retires one running refresh. +fn durable_failure_request( + recovery: &SessionRefreshRecoveryV1, + failure_code: String, +) -> Option { + let (frontier, coverage) = match recovery.progress() { + Some(progress) => (progress.frontier(), *progress.coverage()), + None => ( + SessionRefreshFrontierV1::new( + recovery.target_frontier().observed_through(), + recovery.source_frontier(), + ) + .ok()?, + zero_refresh_coverage(), + ), + }; + let request = SessionRefreshFailureRequestV1::new( + recovery.operation_id().clone(), + recovery.session_id().clone(), + frontier, + coverage, + failure_code, + ) + .ok()?; + Some( + match recovery + .progress() + .and_then(SessionRefreshProgressV1::source_coverage) + .cloned() + .or_else(|| recovery.source_coverage(frontier.committed_through()).ok()) + { + Some(source_coverage) => request.with_source_coverage(source_coverage), + None => request, + }, + ) +} + pub async fn apply_refresh_effect( store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>, state: &SessionTemporalRefreshWakeState, @@ -814,43 +880,73 @@ pub async fn apply_refresh_effect( .await { Ok(_) => report.projected_batches += 1, - Err(error) if error.is_storage() => { + Err(error) if is_retryable_storage(&error) => { report.last_error = Some(format!("{error:?}")); report.retryable_errors += 1; report.observe_retry(SessionTemporalRefreshRetryClass::Storage); } - Err(error) => { - report.last_error = Some(format!("{error:?}")); - report.terminal_errors += 1; - } - } - } - SessionTemporalRefreshEffect::Fail(request) => { - if !state.claim_terminal_attempt(recovery) { - return; - } - let mut attempt = TerminalAttemptGuard::new(state, recovery); - match store.fail_session_refresh(request).await { - Ok(_) => { - report.failed += 1; - state.record_terminal_discovery_failure(recovery); - } - Err(error) if error.is_storage() => { + Err(error) if is_deterministic_refusal(&error) => { + // A refused progress row is not work the next pass can + // finish: rediscovery hands the projector the same durable + // state and the same row comes back refused. Retire the + // operation so it leaves `running` and a fresh refresh can + // be admitted, instead of resubmitting it forever. report.last_error = Some(format!("{error:?}")); - report.retryable_errors += 1; - report.observe_retry(SessionTemporalRefreshRetryClass::Storage); + match durable_failure_request( + recovery, + durable_projector_failure_code(REFRESH_PROGRESS_REFUSED), + ) { + Some(request) => { + apply_fail_effect(store, state, recovery, request, report).await; + } + None => report.terminal_errors += 1, + } } Err(error) => { - attempt.retain(); + // Cancelled control, budget ceiling: this pass could not + // persist, but the row itself was not refused, so the + // operation stays `running` for the next pass. report.last_error = Some(format!("{error:?}")); report.terminal_errors += 1; } } } + SessionTemporalRefreshEffect::Fail(request) => { + apply_fail_effect(store, state, recovery, request, report).await; + } SessionTemporalRefreshEffect::Deferred => report.deferred += 1, } } +async fn apply_fail_effect( + store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>, + state: &SessionTemporalRefreshWakeState, + recovery: &SessionRefreshRecoveryV1, + request: SessionRefreshFailureRequestV1, + report: &mut SessionTemporalRefreshPassReport, +) { + if !state.claim_terminal_attempt(recovery) { + return; + } + let mut attempt = TerminalAttemptGuard::new(state, recovery); + match store.fail_session_refresh(request).await { + Ok(_) => { + report.failed += 1; + state.record_terminal_discovery_failure(recovery); + } + Err(error) if is_retryable_storage(&error) => { + report.last_error = Some(format!("{error:?}")); + report.retryable_errors += 1; + report.observe_retry(SessionTemporalRefreshRetryClass::Storage); + } + Err(error) => { + attempt.retain(); + report.last_error = Some(format!("{error:?}")); + report.terminal_errors += 1; + } + } +} + async fn project_running_refresh( database: &RegisteredGlobalDbLeaseV1, store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>, @@ -894,35 +990,7 @@ async fn project_running_refresh( Err(error) => { let failure_code = durable_projector_failure_code(&error.code); report.last_error = Some(failure_code.clone()); - let (frontier, coverage) = if let Some(progress) = recovery.progress() { - (progress.frontier(), *progress.coverage()) - } else { - let Ok(frontier) = SessionRefreshFrontierV1::new( - recovery.target_frontier().observed_through(), - recovery.source_frontier(), - ) else { - report.terminal_errors += 1; - return; - }; - (frontier, zero_refresh_coverage()) - }; - let request = if let Ok(request) = SessionRefreshFailureRequestV1::new( - recovery.operation_id().clone(), - recovery.session_id().clone(), - frontier, - coverage, - failure_code, - ) { - match recovery - .progress() - .and_then(SessionRefreshProgressV1::source_coverage) - .cloned() - .or_else(|| recovery.source_coverage(frontier.committed_through()).ok()) - { - Some(source_coverage) => request.with_source_coverage(source_coverage), - None => request, - } - } else { + let Some(request) = durable_failure_request(recovery, failure_code) else { report.terminal_errors += 1; return; }; @@ -960,7 +1028,7 @@ async fn running_refreshes( Ok(recoveries) => Some(recoveries), Err(error) => { report.last_error = Some(format!("{error:?}")); - if classify_store_error(&error) == SessionTemporalRefreshRetryClass::Storage { + if is_retryable_storage(&error) { report.retryable_errors += 1; report.observe_retry(SessionTemporalRefreshRetryClass::Storage); } else { @@ -1121,6 +1189,36 @@ mod tests { use tracedecay_sessions::runtime::{SessionMessageRecord, SessionRecord}; use tracedecay_store::ParseOffset; + #[test] + fn deterministic_storage_refusals_are_not_retryable() { + // The schema-contract trigger that refused eleven hours of identical + // progress rows in #1794: transport-level `Storage`, but replaying it + // can never succeed. + let refused = SessionStoreError::storage( + "persist session refresh progress", + EngineError::Sqlite { + operation: "execute", + code: Some(19), + extended_code: Some(1811), + message: "invalid session refresh progress".to_owned(), + }, + ); + assert!(!is_retryable_storage(&refused)); + + // Contention is the transient case the retry loop exists for. + assert!(is_retryable_storage(&SessionStoreError::storage( + "persist session refresh progress", + EngineError::Busy, + ))); + + // Typed contract failures were already terminal and stay terminal. + assert!(!is_retryable_storage( + &SessionStoreError::InvalidStateTransition { + context: "refresh progress successor", + } + )); + } + #[test] fn dropping_worker_instrumentation_clears_pending_state_once() { let state = SessionTemporalRefreshWakeState::default(); diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs index cd3c8ed6c0..f119795e41 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs @@ -1,16 +1,21 @@ use std::sync::Arc; use std::time::Duration; -use tracedecay_domain::SessionId; +use tracedecay_domain::{SessionId, UtcMicros}; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; use tracedecay_session_temporal_store::SessionTemporalStore; use tracedecay_store::{ - SessionRefreshBeginOrJoinRequestV1, SessionRefreshFrontierV1, SessionRefreshStore, + SessionRefreshBeginOrJoinRequestV1, SessionRefreshFrontierV1, SessionRefreshProgressV1, + SessionRefreshStore, SessionTemporalProjectionBatchV1, }; -use super::projector::{CanonicalSessionTemporalProjector, SessionTemporalRefreshPolicy}; +use super::projector::{ + CanonicalSessionTemporalProjector, SessionTemporalRefreshEffect, SessionTemporalRefreshPolicy, + zero_refresh_coverage, +}; +use super::registry::SessionTemporalRefreshPassReport; use super::wake::{SessionTemporalRefreshRetryClass, SessionTemporalRefreshWakeState}; -use super::worker::run_session_temporal_refresh_pass; +use super::worker::{apply_refresh_effect, run_session_temporal_refresh_pass}; async fn begin_empty_refreshes( store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>, @@ -80,3 +85,66 @@ async fn retryable_recovery_stops_the_pass_and_preserves_unattempted_work() { assert_eq!(report.backlog, Some(2)); assert_eq!(state.pending_recovery_operations().len(), 1); } + +#[tokio::test] +async fn refused_projection_progress_retires_the_refresh_instead_of_retrying() { + let harness = RegisteredGlobalDbHarness::open("refresh-refused-progress-retires").await; + let store = SessionTemporalStore::new(harness.registered.as_ref()); + begin_empty_refreshes(&store, ["refused-progress"]).await; + let recovery = store + .running_session_refreshes() + .await + .expect("recoveries") + .pop() + .expect("one running recovery"); + let state = SessionTemporalRefreshWakeState::default(); + let mut report = SessionTemporalRefreshPassReport::default(); + + // Progress that claims a second committed batch while submitting the + // first one. The durable contract refuses it, and every later pass would + // hand the projector the same state and rebuild the same refused row. + let progress = SessionRefreshProgressV1::new( + recovery.operation_id().clone(), + recovery.session_id().clone(), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + zero_refresh_coverage(), + 2, + 0, + UtcMicros(1), + ); + let batch = SessionTemporalProjectionBatchV1::new( + recovery.session_id().clone(), + recovery.candidate_generation(), + recovery.frozen_watermarks().clone(), + vec![], + vec![], + vec![], + ) + .expect("batch") + .with_checkpoint(0, 0, 0) + .expect("checkpoint"); + + apply_refresh_effect( + &store, + &state, + &recovery, + SessionTemporalRefreshEffect::Projection { progress, batch }, + &mut report, + ) + .await; + + assert_eq!( + report.failed, 1, + "a refused progress row must retire the refresh, not stay running" + ); + assert_eq!(report.retryable_errors, 0); + assert_eq!(report.terminal_errors, 0); + assert!( + store + .running_session_refreshes() + .await + .expect("recoveries") + .is_empty(), + "the retired refresh must not be rediscovered" + ); +} diff --git a/crates/tracedecay-session-temporal-store/src/doctor_health.rs b/crates/tracedecay-session-temporal-store/src/doctor_health.rs index 8a76c5a39a..9bdf989170 100644 --- a/crates/tracedecay-session-temporal-store/src/doctor_health.rs +++ b/crates/tracedecay-session-temporal-store/src/doctor_health.rs @@ -1362,7 +1362,6 @@ mod registered_tests { ) .await .expect("drop required index"); - drop(writer); let database = SessionTemporalRegisteredDb::db_path(&harness.registered); std::fs::OpenOptions::new() diff --git a/crates/tracedecay-sessions/src/admission/mod.rs b/crates/tracedecay-sessions/src/admission/mod.rs index dc51b835a1..092d05457c 100644 --- a/crates/tracedecay-sessions/src/admission/mod.rs +++ b/crates/tracedecay-sessions/src/admission/mod.rs @@ -924,6 +924,7 @@ pub(crate) mod test_support { projection_failure: Arc>>, cancel_on_discovery_queue_read: Arc>>, session_backfill_page_pause: Arc>>, + deterministic_capture_refusal: Arc>>, } impl MemoryHostAdmission { @@ -940,6 +941,16 @@ pub(crate) mod test_support { self.store.state().capture_failures_remaining = 1; } + /// Refuse every capture the way a deterministic content refusal does: + /// the same record fails identically on every retry, so callers must + /// converge past it rather than re-attempt the source forever. + pub(crate) fn refuse_captures_deterministically(&self, reason: &'static str) { + *self + .deterministic_capture_refusal + .lock() + .unwrap_or_else(|error| error.into_inner()) = Some(reason); + } + /// Make the next `count` session-message lookups report the store as /// unavailable, the way reader-pool saturation does. pub(crate) fn fail_next_session_message_lookups(&self, count: usize) { @@ -1042,6 +1053,13 @@ pub(crate) mod test_support { request: CaptureObservationRequest, ) -> AdmissionFuture<'a, CaptureObservationOutcome> { Box::pin(async move { + if let Some(reason) = *self + .deterministic_capture_refusal + .lock() + .unwrap_or_else(|error| error.into_inner()) + { + return Err(HostAdmissionOutcome::deterministic_content_refusal(reason)); + } { let mut state = self.store.state(); state.scalar_capture_calls = state.scalar_capture_calls.saturating_add(1); diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index 67439f9621..b663217469 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -269,17 +269,23 @@ impl Default for CodexReplayIndex { } #[cfg(test)] -static CODEX_REPLAY_INDEX_ENTRIES_VISITED: std::sync::atomic::AtomicU64 = - std::sync::atomic::AtomicU64::new(0); +thread_local! { + /// Per-thread, because `indexed_replay_pass` runs entirely on its caller's + /// thread and the one test that measures B-tree traversal shares the + /// process with every other test replaying an index in parallel. A global + /// counter measures the whole suite's traversal, not this pass's. + static CODEX_REPLAY_INDEX_ENTRIES_VISITED: std::cell::Cell = + const { std::cell::Cell::new(0) }; +} #[cfg(test)] fn reset_replay_index_entries_visited_for_test() { - CODEX_REPLAY_INDEX_ENTRIES_VISITED.store(0, std::sync::atomic::Ordering::Release); + CODEX_REPLAY_INDEX_ENTRIES_VISITED.with(|visited| visited.set(0)); } #[cfg(test)] fn replay_index_entries_visited_for_test() -> u64 { - CODEX_REPLAY_INDEX_ENTRIES_VISITED.load(std::sync::atomic::Ordering::Acquire) + CODEX_REPLAY_INDEX_ENTRIES_VISITED.with(std::cell::Cell::get) } fn indexed_replay_pass( @@ -295,7 +301,8 @@ fn indexed_replay_pass( let lower = position.map_or(Bound::Unbounded, Bound::Excluded); for indexed in index.paths.range((lower, Bound::Unbounded)) { #[cfg(test)] - CODEX_REPLAY_INDEX_ENTRIES_VISITED.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + CODEX_REPLAY_INDEX_ENTRIES_VISITED + .with(|visited| visited.set(visited.get().saturating_add(1))); let path_bytes = u64::try_from(crate::runtime::source::path_byte_len(&indexed.path)).unwrap_or(u64::MAX); if paths.len() >= bounds.max_files.max(1) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs index 117e0d70ca..8982f1a1e2 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs @@ -15,6 +15,7 @@ use tracedecay_store::observation::{ObservationCoverageReason, ObservationCursor use crate::admission::{HostAdmission, HostAdmissionOutcome}; use crate::observation::{CaptureObservationOutcome, ObservationCancellation}; +use crate::runtime::jsonl_observation_admission::is_deterministic_content_refusal; use crate::runtime::shared::TranscriptIngestStats; use tracedecay_runtime_core::db::{SqliteFileIdentityOperation, sqlite_generation_identity}; @@ -89,8 +90,29 @@ async fn advance_coverage( .map_err(host_admission_error) } +/// The admission's own verdict, verbatim. +/// +/// The status alone names a family ("degraded"), not a cause: every +/// deterministic refusal, cursor mismatch and contract violation collapsed +/// into one indistinguishable sentence, so a sweep that skipped the same +/// `state.db` every five seconds forever gave an operator nothing to act on. +/// Carry the reason code, retryability and storage cause the outcome already +/// holds. fn host_admission_error(outcome: HostAdmissionOutcome) -> String { - crate::runtime::snapshot_observation::host_admission_status_message("Hermes", outcome.status) + let mut message = crate::runtime::snapshot_observation::host_admission_status_message( + "Hermes", + outcome.status, + ); + if let Some(reason) = outcome.reason_code { + message.push_str(&format!( + " (reason_code={reason}, retryable={})", + outcome.retryable + )); + } + if let Some(cause) = outcome.storage_cause { + message.push_str(&format!(": {cause}")); + } + message } pub(super) async fn drain_hermes_projections_with_admission( @@ -221,11 +243,44 @@ pub(super) async fn admit_rows_with_admission_and_cancellation( .await?; } HermesAdmissionAction::Capture(request) => { - match facade - .capture_observation(*request) - .await - .map_err(host_admission_error)? - { + let captured = match facade.capture_observation(*request).await { + Ok(captured) => captured, + // A deterministic content refusal re-fails identically on + // every pass. Without a durable skip the source's cursor + // never clears the offending row, so the whole `state.db` + // is abandoned every sweep, forever, with one WARN each + // time. Cover past it with a typed reason exactly as the + // shared JSONL path does so the stream converges. + Err(outcome) if is_deterministic_content_refusal(&outcome) => { + tracing::warn!( + provider = PROVIDER, + row = row.id, + reason = outcome.reason_code.unwrap_or("host_admission_refused"), + "admission refused a Hermes row; covering past it" + ); + advance_coverage( + facade, + source, + range, + expected_cursor, + scope.clone(), + generation, + if outcome.reason_code == Some("observation_identity_collision") { + ObservationCoverageReason::ObservationIdentityCollision + } else { + ObservationCoverageReason::AdmissionRefused + }, + None, + file_identity, + resume_fingerprint, + cancellation, + ) + .await?; + continue; + } + Err(outcome) => return Err(host_admission_error(outcome)), + }; + match captured { CaptureObservationOutcome::Persisted { outcome, .. } | CaptureObservationOutcome::AcceptedForReplay { outcome, .. } => { if matches!(*outcome, ObservationPersistOutcome::Committed(_)) { diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs index 88476f6bf4..3d2ca7642f 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs @@ -26,6 +26,38 @@ fn new_sweep_budget(max_new_bytes: Option) -> IngestByteBudget { IngestByteBudget::bounded(max_new_bytes.unwrap_or(DEFAULT_HERMES_SWEEP_BYTES)) } +/// Whether this sweep pass should report a source outcome, given what the last +/// pass reported for the same `state.db`. +/// +/// A source that cannot be admitted stays unadmittable until something about +/// the store or the file changes, and the sweep runs every few seconds. Logging +/// the identical line each pass buried every other daemon warning without +/// telling an operator anything the first line did not. Report a failure when +/// it is new or its reason changed; `None` records a recovered source so its +/// next failure is reported again. The map is keyed by discovered Hermes +/// profile, so it is bounded by the number of profiles on disk. +fn hermes_source_outcome_is_new(state_db: &Path, error: Option<&str>) -> bool { + use std::collections::BTreeMap; + use std::sync::{LazyLock, Mutex, PoisonError}; + + static REPORTED: LazyLock>> = + LazyLock::new(|| Mutex::new(BTreeMap::new())); + let mut reported = REPORTED.lock().unwrap_or_else(PoisonError::into_inner); + match error { + Some(error) => { + if reported.get(state_db).is_some_and(|last| last == error) { + return false; + } + reported.insert(state_db.to_path_buf(), error.to_owned()); + true + } + None => { + reported.remove(state_db); + false + } + } +} + /// Default Hermes profile homes under the resolved user home. /// /// Missing home is a typed absence (`None`), never an empty successful sweep. @@ -283,14 +315,19 @@ pub(super) async fn ingest_homes_capped_with_admission_and_cancellation( ) .await { - Ok(source_stats) => outcome.stats = outcome.stats.merge(source_stats), + Ok(source_stats) => { + hermes_source_outcome_is_new(&source.state_db, None); + outcome.stats = outcome.stats.merge(source_stats); + } Err(error) => { outcome.source_failures = outcome.source_failures.saturating_add(1); - tracing::warn!( - state_db = %source.state_db.display(), - error, - "skipping Hermes transcript source" - ); + if hermes_source_outcome_is_new(&source.state_db, Some(&error)) { + tracing::warn!( + state_db = %source.state_db.display(), + error, + "skipping Hermes transcript source" + ); + } } } } @@ -403,14 +440,19 @@ async fn ingest_user_homes_capped_with_admission( ) .await { - Ok(source_stats) => outcome.stats = outcome.stats.merge(source_stats), + Ok(source_stats) => { + hermes_source_outcome_is_new(&source.state_db, None); + outcome.stats = outcome.stats.merge(source_stats); + } Err(error) => { outcome.source_failures = outcome.source_failures.saturating_add(1); - tracing::warn!( - state_db = %source.state_db.display(), - error, - "skipping projectless Hermes transcript source" - ); + if hermes_source_outcome_is_new(&source.state_db, Some(&error)) { + tracing::warn!( + state_db = %source.state_db.display(), + error, + "skipping projectless Hermes transcript source" + ); + } } } } diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs index 7d5f30d72c..d69d457c28 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs @@ -1688,3 +1688,59 @@ async fn unreadable_state_db_is_a_counted_source_failure_not_a_clean_sweep() { ); assert!(admission.observations().is_empty()); } + +/// A deterministic admission refusal is permanent: the same row fails the same +/// way on every sweep. Without a durable skip the source cursor never clears +/// it, so the whole profile `state.db` is abandoned every pass forever, which +/// is what produced an endless "skipping projectless Hermes transcript source" +/// WARN on a live daemon. Cover past it, exactly as the shared JSONL path +/// does, so the source converges. +mod deterministic_refusal_recovery { + use super::*; + + async fn admit_one_refused_row(reason: &'static str) -> MemoryHostAdmission { + let admission = MemoryHostAdmission::default(); + admission.refuse_captures_deterministically(reason); + let stats = admit_rows_with_admission_and_cancellation( + &admission, + &[fixture(1)], + ObservationScopeV1::Profile, + ObservationSourceGenerationV1::new(1).unwrap(), + 1, + 1, + |_| Some(fixture_projection()), + &ObservationCancellation::default(), + ) + .await + .expect("a permanently refused row must not abandon the whole source"); + assert_eq!(stats.messages_upserted, 0); + admission + } + + #[tokio::test] + async fn refused_row_is_covered_past_instead_of_skipping_the_source() { + let admission = admit_one_refused_row("privacy_boundary_failed").await; + + let advances = admission.non_durable_advances(); + assert_eq!( + advances.len(), + 1, + "the refused row must be covered exactly once" + ); + assert_eq!( + advances[0].reason(), + ObservationCoverageReason::AdmissionRefused + ); + assert_eq!(advances[0].next_cursor().position(), 1); + } + + #[tokio::test] + async fn identity_collision_keeps_its_own_coverage_reason() { + let admission = admit_one_refused_row("observation_identity_collision").await; + + assert_eq!( + admission.non_durable_advances()[0].reason(), + ObservationCoverageReason::ObservationIdentityCollision + ); + } +} diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index 468022fd84..2f40a44a2f 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -483,10 +483,18 @@ pub(in crate::runtime) fn install_test_shared_jsonl_preparation_authority() { use std::num::NonZeroUsize; use tracedecay_runtime_core::resident_memory::ProcessResidentMemoryV1; + // One process-wide budget serves the whole suite, so every test thread + // holding a `SHARED_JSONL_WORKER_RESERVATION_BYTES` page charges it at + // once. At 32 GiB a wide harness drove the derived preparation capacity + // down to two entries, which is the shared metadata cache's degraded mode, + // not the product's: a production process meters one ingest workload + // against the machine. Size the budget past what the harness's own + // parallelism can reserve so capacity stays CPU-bound, the way the + // composition root installs it. static MEMORY: OnceLock> = OnceLock::new(); let memory = Arc::clone(MEMORY.get_or_init(|| { Arc::new(ProcessResidentMemoryV1::new( - NonZeroU64::new(32 * 1024 * 1024 * 1024).unwrap(), + NonZeroU64::new(1024 * 1024 * 1024 * 1024).unwrap(), )) })); let background_cpu = Arc::new(ProcessBackgroundCpuV1::new(NonZeroUsize::new(48).unwrap())); @@ -1849,10 +1857,19 @@ impl ActiveAdmission<'_> { .with_resume_checkpoint(self.file_identity, checkpoint.resume_fingerprint); hotpath::gauge!("jsonl_admission_coverage_frames").inc(1.0); hotpath::gauge!("jsonl_admission_writer_submits").inc(1.0); - self.admission + if let Err(outcome) = self + .admission .advance_non_durable_source_cursor(advance, self.cancellation.clone()) .await - .map_err(|outcome| { + { + if is_lost_cursor_cas(&outcome) + && self + .peer_already_covered(expected_cursor, checkpoint.end_offset) + .await + { + return Ok(()); + } + return Err({ if is_admission_cancellation(&outcome, &self.cancellation) { TranscriptIngestError::Cancelled { provider: self.provider, @@ -1874,12 +1891,47 @@ impl ActiveAdmission<'_> { .unwrap_or("non_durable_cursor_advance_failed"), } } - })?; + }); + } *expected_cursor = Some(self.cursor_at(checkpoint.end_offset, checkpoint.resume_fingerprint)?); Ok(()) } + /// Whether the peer that won a cursor CAS already covered this range. + /// + /// Live hook ingest and the catch-up sweep own the same `(source, scope)` + /// cursor and routinely read the same transcript at once; the store's + /// compare-and-swap is what keeps them honest, so one of them loses. The + /// loser's frames are almost always already durable behind the winner's + /// cursor, and re-reading that cursor is enough to prove it. Adopt the + /// winner's cursor and let the pass continue instead of failing the whole + /// source over work that is already committed. + /// + /// A read failure, a different generation, or a cursor short of this frame + /// all answer "not covered", which keeps the caller's typed block. + #[hotpath::skip] + async fn peer_already_covered( + &self, + expected_cursor: &mut Option, + end_offset: u64, + ) -> bool { + let Ok(actual) = self + .admission + .get_source_cursor(&self.source, &self.scope) + .await + else { + return false; + }; + let covered = actual.as_ref().is_some_and(|cursor| { + cursor.generation() == self.generation && cursor.position() >= end_offset + }); + if covered { + *expected_cursor = actual; + } + covered + } + fn capture_request( &self, expected_cursor: Option, @@ -1987,6 +2039,13 @@ impl ActiveAdmission<'_> { if outcome.status == HostAdmissionStatus::Backpressured { hotpath::gauge!("jsonl_admission_backpressure_writer").inc(1.0); } + if is_lost_cursor_cas(&outcome) + && self + .peer_already_covered(expected_cursor, checkpoint.end_offset) + .await + { + return Ok(DurableFrameDisposition::AlreadyDurable); + } if is_admission_cancellation(&outcome, &self.cancellation) { Err(TranscriptIngestError::Cancelled { provider: self.provider, @@ -2172,6 +2231,21 @@ impl ActiveAdmission<'_> { } } } + // The batch is atomic: nothing in this window committed. When + // the peer that won the CAS is already past the window's last + // frame, every frame in it is durable behind the winner's + // cursor, so this is a no-op rather than a failed source pass. + if is_lost_cursor_cas(&outcome) + && let Some(last) = checkpoints.last() + && self + .peer_already_covered(expected_cursor, last.end_offset) + .await + { + progress.frames_skipped = progress + .frames_skipped + .saturating_add(checkpoints.len() as u64); + return Ok(()); + } if is_admission_cancellation(&outcome, &self.cancellation) { Err(CaptureWindowError::Ingest( TranscriptIngestError::Cancelled { @@ -2852,7 +2926,14 @@ pub(in crate::runtime) async fn admit_jsonl_observations( /// unbound authorities, retryable races, says nothing about the record and /// must surface as a typed block instead of writing coverage over a commit /// that never landed (or one that already landed and advanced the cursor). -fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool { +/// A cursor compare-and-swap lost to a peer that owns the same +/// `(source, scope)` cursor. Retryable by construction; whether it is a +/// failure at all depends on what the winner already covered. +fn is_lost_cursor_cas(outcome: &HostAdmissionOutcome) -> bool { + outcome.reason_code == Some("cursor_conflict") +} + +pub(in crate::runtime) fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool { matches!( outcome.recovery, Some(HostAdmissionRecovery::DeterministicContentRefusal) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index cb9b053e7d..3915ac089b 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -59,6 +59,10 @@ struct SeamSpyAdmission { capture_calls: AtomicU64, capture_collision_dispositions: Mutex>, cover_past_advances: Mutex>, + /// Commit the next capture through the shared store and then report the + /// cursor CAS as lost, the way a live hook ingest wins the race a sweep + /// was still trying to write. + peer_wins_next_cursor_cas: AtomicBool, } #[tokio::test] @@ -651,6 +655,14 @@ impl SeamSpyAdmission { *self.scripted_capture_error.lock().unwrap() = Some(outcome); } + fn script_peer_wins_next_cursor_cas(&self) { + self.peer_wins_next_cursor_cas.store(true, Ordering::SeqCst); + } + + fn peer_won_cursor_cas(&self) -> bool { + self.peer_wins_next_cursor_cas.swap(false, Ordering::SeqCst) + } + fn script_batch_error(&self, outcome: HostAdmissionOutcome) { *self.scripted_batch_error.lock().unwrap() = Some(outcome); } @@ -683,6 +695,12 @@ impl HostAdmission for SeamSpyAdmission { .lock() .unwrap() .push(request.identity_collision_disposition()); + if self.peer_won_cursor_cas() { + let _ = self.inner.capture_observation(request).await; + return Err(HostAdmissionOutcome::retained_backpressured( + "cursor_conflict", + )); + } if let Some(outcome) = self.scripted_capture_error_once.lock().unwrap().take() { return Err(outcome); } @@ -703,6 +721,12 @@ impl HostAdmission for SeamSpyAdmission { .iter() .map(CaptureObservationRequest::identity_collision_disposition), ); + if self.peer_won_cursor_cas() { + let _ = self.inner.capture_observations(requests).await; + return Err(HostAdmissionOutcome::retained_backpressured( + "cursor_conflict", + )); + } if let Some(outcome) = self.scripted_batch_error.lock().unwrap().take() { return Err(outcome); } @@ -898,6 +922,68 @@ async fn retryable_admission_failures_keep_their_own_verdict() { assert!(stored_cursor(&spy).await.is_none()); } +/// Live hook ingest and the catch-up sweep own the same `(source, scope)` +/// cursor, so one of them loses the store's compare-and-swap. When the winner +/// already covered the range the loser was writing, the loser's work is +/// durable and the pass is a no-op, not a failed source: reporting it as a +/// failure produced a "Cursor transcript catch-up failed" WARN roughly every +/// ten seconds on a live daemon for work that was already committed. +#[tokio::test] +async fn cursor_cas_lost_to_a_peer_that_covered_the_range_is_a_no_op() { + let (_temp, path, len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + spy.script_peer_wins_next_cursor_cas(); + + let stats = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("a CAS the peer already covered must not fail the source pass"); + + assert_eq!( + stored_cursor(&spy).await.map(|cursor| cursor.position()), + Some(len), + "the pass must adopt the winner's frontier" + ); + assert!( + !spy.inner.observations().is_empty(), + "the peer's commit is the durable record this pass stopped duplicating" + ); + assert_eq!( + stats.frames_accepted, 0, + "the loser accepts nothing of its own" + ); + assert!( + stats.frames_skipped > 0, + "the covered frames are counted as skipped, not lost" + ); +} + +/// The same lost CAS with nothing behind it stays a typed retryable block: +/// adopting a frontier the winner never reached would skip real records. +#[tokio::test] +async fn cursor_cas_lost_without_peer_coverage_stays_a_typed_block() { + let (_temp, path, _len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + spy.script_capture_error(HostAdmissionOutcome::retained_backpressured( + "cursor_conflict", + )); + + let error = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect_err("an uncovered race must surface for another pass"); + + assert!(matches!( + error, + TranscriptIngestError::HostAdmission { + reason: "cursor_conflict", + retryable: true, + .. + } + )); + assert!(stored_cursor(&spy).await.is_none()); +} + #[tokio::test] async fn eligible_identity_collision_retries_once_with_normalizer_fallback() { super::install_test_shared_jsonl_preparation_authority(); @@ -1175,6 +1261,11 @@ async fn content_refusals_cover_past_so_the_stream_converges() { #[tokio::test] async fn codex_session_meta_prefix_is_decoded_once_across_consumers() { + // The shared metadata cache retains entries up to + // `shared_jsonl_preparation_capacity()`, so this test only observes the + // shared decode once the preparation authority is installed: without it the + // capacity is the degraded fallback of one entry. + super::install_test_shared_jsonl_preparation_authority(); let (_temp, path, _) = rollout_fixture(); let first = SeamSpyAdmission::default(); let second = SeamSpyAdmission::default(); diff --git a/crates/tracedecay-store/src/session/refresh.rs b/crates/tracedecay-store/src/session/refresh.rs index 33b52a7e8d..c6c8696dc9 100644 --- a/crates/tracedecay-store/src/session/refresh.rs +++ b/crates/tracedecay-store/src/session/refresh.rs @@ -282,8 +282,13 @@ impl SessionRefreshProgressV1 { } let current = self.coverage; let candidate = next.coverage; + // The durable guard admits a successor only when it strictly advances + // the committed frontier. Accepting an equal frontier here let a + // producer submit a row the trigger then refused as a SQLite + // constraint abort, which the worker read as transient storage and + // resubmitted forever. Refuse it as typed state instead. if self.frontier.observed_through != next.frontier.observed_through - || next.frontier.committed_through < self.frontier.committed_through + || next.frontier.committed_through <= self.frontier.committed_through || next.committed_batches < self.committed_batches || next.committed_records < self.committed_records || candidate.visible < current.visible diff --git a/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs b/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs index f4ffc7aeaa..0de934cb97 100644 --- a/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs +++ b/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs @@ -225,6 +225,25 @@ fn refresh_frontiers_and_progress_are_monotonic_and_terminal() { }) )); + // The durable guard admits a successor only when it strictly advances the + // committed frontier, so a stalled successor must be refused here rather + // than deferred to a SQLite constraint abort the caller reads as storage. + let stalled = SessionRefreshProgressV1::new( + operation_id(), + session_id.clone(), + SessionRefreshFrontierV1::new(10, 8).unwrap(), + coverage(), + 2, + 8, + UtcMicros(101), + ); + assert!(matches!( + initial.validate_successor(&stalled), + Err(SessionStoreError::InvalidStateTransition { + context: "refresh progress successor" + }) + )); + let terminal = SessionRefreshReceiptV1::completed( SessionRefreshCompletionRequestV1::new( operation_id(), diff --git a/crates/tracedecay/src/daemon/branch_add.rs b/crates/tracedecay/src/daemon/branch_add.rs index 4e2d3793f7..0b7b5f17e2 100644 --- a/crates/tracedecay/src/daemon/branch_add.rs +++ b/crates/tracedecay/src/daemon/branch_add.rs @@ -1,4 +1,4 @@ -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use tracedecay_application::pr_tracking::{ @@ -153,6 +153,9 @@ async fn activate_and_track_manual_branch( let graph = Arc::clone(graph); let schedulers = schedulers.clone(); let branch = branch.to_owned(); + let published_data_root = data_root.clone(); + let published_schedulers = schedulers.clone(); + let published_registries = administration.session_runtime_registries(); administration .admit_manual_branch_publication(|cancellation, admitted| async move { @@ -215,6 +218,15 @@ async fn activate_and_track_manual_branch( tracked } .await; + if matches!(&result, Ok(outcome) if *outcome != BranchAddOutcome::Deferred) { + mount_published_branch_query_authority( + published_registries.as_ref(), + &published_schedulers, + &published_data_root, + &branch, + ) + .await; + } match &result { Ok(outcome) => log_daemon_event( "manual_branch_publication", @@ -237,6 +249,101 @@ async fn activate_and_track_manual_branch( .await } +/// Mounts the checked-in core query authority on the branch worktree this +/// publication sealed, from the project's own durable cursor-key authority. +/// +/// An explicitly published branch worktree is never a project-open route, so +/// nothing else mounts its query authority: an exact branch read could only +/// borrow one already mounted on a peer checkout of the same repository +/// (`mount_query_authority_from_project_peer`), and that peer's own mount is +/// deferred until it seats a text generation. A read taken right after this +/// publication sealed its provenance therefore failed closed with a +/// non-retryable `authority_unavailable` even though the branch generation was +/// published and servable. Mounting here makes the generation this journey +/// publishes queryable without depending on an unrelated worktree's +/// activation order. +/// +/// Runs inside the daemon-owned publication task, after the generation it +/// serves is committed: the admitting `branch add` caller returns at admission +/// and never waits for this, and the profile's session-registry lock is taken +/// here rather than on that caller's path. +/// +/// Best effort by design: the branch generation is already committed, so a +/// missing session mount or cursor key must not retract it. The exact branch +/// read falls back to borrowing a peer authority when this could not run. +#[cfg(unix)] +#[hotpath::measure(label = "daemon.branch_add.query_authority", future = true)] +async fn mount_published_branch_query_authority( + registries: Option<&(super::branch_admin::SharedSessionRuntimeRegistries, PathBuf)>, + schedulers: &CodeIndexSchedulerRegistryV1, + data_root: &Path, + branch: &str, +) { + let Some((registries, profile_root)) = registries else { + return; + }; + let Some(source) = + tracedecay_runtime_core::branch_meta::load_branch_meta(data_root).and_then(|meta| { + meta.branches + .get(branch) + .and_then(|entry| entry.graph_source.clone()) + }) + else { + return; + }; + let worktree_root = PathBuf::from(&source.worktree_root); + let Ok(project_id) = tracedecay_domain::ProjectId::new(source.project_id.clone()) else { + return; + }; + let Ok(scope) = + tracedecay_code_index_runtime::resolved_scope_for_project(&worktree_root, &project_id) + else { + return; + }; + let sessions = { + let registries = registries.lock().await; + registries + .get(profile_root) + .map(|entry| Arc::clone(&entry.registry)) + }; + let Some(sessions) = sessions.and_then(|registry| registry.get().cloned()) else { + return; + }; + let Some(session_db) = sessions.mounted_project_sessions(&project_id).await else { + return; + }; + let cursor_keys = match session_db.load_session_cursor_key_provider_result().await { + Ok(cursor_keys) => cursor_keys, + Err(error) => { + tracing::debug!( + event = "branch_query_authority_mount", + outcome = "unavailable", + branch = %branch, + reason = %error, + "durable query cursor key is unavailable for the published branch" + ); + return; + } + }; + if let Err(error) = + tracedecay_code_index_runtime::code_index_scheduler::query_runtime::mount_core_query_authority_on_project_open( + schedulers, + &worktree_root, + &scope, + &cursor_keys, + ) + .await + { + tracing::debug!( + event = "branch_query_authority_mount", + outcome = "unavailable", + branch = %branch, + reason = %error, + "published branch query authority is unavailable; exact reads fall back to a peer" + ); + } +} + #[cfg(unix)] #[hotpath::measure(label = "daemon.branch_add.owner", future = true)] pub(super) async fn activate_and_track_manual_branch_owned( diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs index f65a2c3924..899dc6fcfc 100644 --- a/crates/tracedecay/src/daemon/branch_admin.rs +++ b/crates/tracedecay/src/daemon/branch_admin.rs @@ -944,6 +944,24 @@ impl StoreAdministration { registry.mounted_session_databases().await } + /// The profile's session-runtime registry map and its canonical root, + /// taken without locking either. + /// + /// Branch publication resolves the project's durable cursor-key authority + /// through this inside its own background task, so an explicitly published + /// branch can mount its own query authority without the admitting caller + /// paying for a registry lock it never reads. + #[cfg(unix)] + pub(super) fn session_runtime_registries( + &self, + ) -> Option<(SharedSessionRuntimeRegistries, std::path::PathBuf)> { + let profile_root = self + .profile_identity() + .and_then(|identity| authority::canonical_identity_path(identity.profile_root())) + .ok()?; + Some((Arc::clone(&self.session_runtime_registries), profile_root)) + } + #[hotpath::measure(label = "daemon.branch_admin.mounted_project_servers", future = true)] pub(super) async fn mounted_project_servers(&self) -> Vec> { let Ok(profile_root) = self diff --git a/crates/tracedecay/src/daemon/core_proxy.rs b/crates/tracedecay/src/daemon/core_proxy.rs index d9b3a4a712..025ef26848 100644 --- a/crates/tracedecay/src/daemon/core_proxy.rs +++ b/crates/tracedecay/src/daemon/core_proxy.rs @@ -909,7 +909,7 @@ fn daemon_version_skew_warning_for_request( client_version: &str, ) -> Option { let daemon_version = proxy_initialize_metadata_for_request(request, responses).daemon_version?; - if daemon_version == client_version { + if tracedecay_daemon_protocol::versions_name_same_build(&daemon_version, client_version) { return None; } let action = version_skew_action(&daemon_version, client_version); 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..6415ef3209 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}; @@ -114,13 +115,29 @@ 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"); + // 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 { + loop { + match prepare_next_code_generation_retention_cancellable( + &code_store_root, + &BTreeSet::new(), + &|| false, + Some(&graph_replay_pool_root), + ) { + Ok(plan) => return plan, + Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy) => { + tokio::time::sleep(Duration::from_millis(25)).await; + } + Err(error) => panic!("code generation retention plan: {error:?}"), + } + } + }) + .await + .expect("code generation retention plan converges"); let first_candidate = plan .collectable_generations .iter() diff --git a/crates/tracedecay/src/daemon/store_runtime_tests.rs b/crates/tracedecay/src/daemon/store_runtime_tests.rs index 3b58371f8a..8799d13b61 100644 --- a/crates/tracedecay/src/daemon/store_runtime_tests.rs +++ b/crates/tracedecay/src/daemon/store_runtime_tests.rs @@ -30,9 +30,9 @@ use tracedecay_session_memory::memory::{ }; use tracedecay_store::{ CursorAdvanceOutcome, FactReadControl, FactWriteControl, ObservationCoverageReason, - ObservationCursorAdvance, ObservationStore, ObservationStoreError, ProjectId, - ProjectMemoryFactHistoryQueryV1, ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1, - RetainedGraphStoreLeaseV1, StoreShardIdV1, + ObservationCursorAdvance, ObservationStore, ProjectId, ProjectMemoryFactHistoryQueryV1, + ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1, RetainedGraphStoreLeaseV1, + StoreShardIdV1, }; use tracedecay_store_runtime::{ DaemonSessionRuntimeRegistryV1, RegisteredSchemaConvergenceStatus, process_runtime_generation, @@ -1193,19 +1193,22 @@ async fn retained_runtime_ledger_replays_during_bounded_background_convergence() .expect("replay retained cursor while convergence is pending"), CursorAdvanceOutcome::ExactDuplicate ); - let conflicting_advance = runtime_cursor_advance( + // The same range under a different coverage reason finds the retained + // cursor already at its `next_cursor`, so it is a duplicate of the + // applied coverage rather than a collision (#1842). + let rereasoned_advance = runtime_cursor_advance( &project_id, "retired", ObservationCoverageReason::BlankFrame, ); - assert!(matches!( + assert_eq!( database .observation_store() - .advance_source_cursor(conflicting_advance) + .advance_source_cursor(rereasoned_advance) .await - .expect_err("classify retained cursor collision while convergence is pending"), - ObservationStoreError::CursorAdvanceCollision - )); + .expect("classify a re-reasoned retained cursor while convergence is pending"), + CursorAdvanceOutcome::ExactDuplicate + ); let fresh_advance = runtime_cursor_advance(&project_id, "fresh", ObservationCoverageReason::OutOfScope); diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs index 2d409392bd..28f292af5f 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs @@ -82,6 +82,51 @@ fn write_production_codex_rollouts(home: &Path, project: &Path, count: usize) { async fn production_codex_message_search( harness: &ProductionProjectCompositionHarnessV1, project: &Path, +) -> Value { + // A `partial` generation is the store saying "still converging", the same + // not-ready contract as `stale`: re-read it. Every other outcome answers + // now, so an empty `complete_zero` still fails the assertions below. + let payload = tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + let payload = production_codex_message_search_once(harness, project).await; + if payload["outcome"] != "partial" + || payload["results"] + .as_array() + .is_some_and(|results| !results.is_empty()) + { + break payload; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("production Codex message search convergence deadline"); + assert!( + payload["results"].as_array().is_some_and(|results| { + results.iter().any(|result| { + result["message"]["text"] + .as_str() + .is_some_and(|text| text.contains("cobalt orchard scheduler migration")) + }) + }), + "production Codex message search was empty after completed ingest: {payload}" + ); + assert!( + payload["results"].as_array().is_some_and(|results| { + results.iter().any(|result| { + result["message"]["text"].as_str() + == Some("The cobalt orchard scheduler migration is ready for review") + }) + }), + "production Codex message search did not hydrate the exact assistant message: {payload}" + ); + payload +} + +#[cfg(feature = "test-transport")] +async fn production_codex_message_search_once( + harness: &ProductionProjectCompositionHarnessV1, + project: &Path, ) -> Value { let response = harness .call_tool( @@ -108,30 +153,10 @@ async fn production_codex_message_search( .expect("production message search JSON"); // Retained tools respond with the full evidence envelope; the search // payload the assertions consume lives under `outcome.value.payload`. - let payload = envelope + envelope .pointer("/outcome/value/payload") .cloned() - .unwrap_or(envelope); - assert!( - payload["results"].as_array().is_some_and(|results| { - results.iter().any(|result| { - result["message"]["text"] - .as_str() - .is_some_and(|text| text.contains("cobalt orchard scheduler migration")) - }) - }), - "production Codex message search was empty after completed ingest: {payload}" - ); - assert!( - payload["results"].as_array().is_some_and(|results| { - results.iter().any(|result| { - result["message"]["text"].as_str() - == Some("The cobalt orchard scheduler migration is ready for review") - }) - }), - "production Codex message search did not hydrate the exact assistant message: {payload}" - ); - payload + .unwrap_or(envelope) } #[cfg(feature = "test-transport")] diff --git a/crates/tracedecay/tests/session_suite/observation_store/mod.rs b/crates/tracedecay/tests/session_suite/observation_store/mod.rs index c4d27abd67..eeeb421313 100644 --- a/crates/tracedecay/tests/session_suite/observation_store/mod.rs +++ b/crates/tracedecay/tests/session_suite/observation_store/mod.rs @@ -1476,7 +1476,11 @@ async fn cursor_only_progress_persists_non_payload_receipt_and_retries_idempoten } #[tokio::test] -async fn cursor_only_retry_rejects_same_cursor_with_different_reason() { +/// A second owner replaying the same range with a different coverage reason +/// finds the durable cursor already at its `next_cursor`: the coverage it +/// wanted to record is applied, so the replay is a duplicate, not a +/// collision that blocks ingest (#1842). The committed reason stays. +async fn cursor_only_retry_with_same_cursor_and_different_reason_is_a_duplicate() { let tmp = TempDir::new().unwrap(); let runtime = profile_runtime(&tmp).await; let store = runtime @@ -1493,7 +1497,7 @@ async fn cursor_only_retry_rejects_same_cursor_with_different_reason() { .await .unwrap(); - assert!(matches!( + assert_eq!( store .advance_source_cursor(cursor_advance( None, @@ -1501,9 +1505,10 @@ async fn cursor_only_retry_rejects_same_cursor_with_different_reason() { 10, NonDurableFrameReason::OutOfScope, )) - .await, - Err(ObservationStoreError::CursorAdvanceCollision) - )); + .await + .unwrap(), + CursorAdvanceOutcome::ExactDuplicate + ); assert_eq!( store.get_source_cursor(&source(), &scope()).await.unwrap(), Some(cursor(10))