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..69bfce445e 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations.rs @@ -1745,7 +1745,27 @@ fn read_active_pointer( store_root: &Path, ) -> Result { let path = store_root.join(ACTIVE_POINTER_FILE); - let bytes = std::fs::read(&path).map_err(storage)?; + // A directory in the pointer slot makes `read(2)` return EISDIR. That is + // the same corrupt authority the publication store refuses; do not let the + // OS error replace the typed unsafe-state. + match std::fs::metadata(&path) { + Ok(metadata) if metadata.file_type().is_file() => {} + Ok(_) => { + return Err(CodeGenerationRetentionErrorV1::UnsafeState( + "active code-generation pointer is not a regular file".to_owned(), + )); + } + Err(error) => return Err(storage(error)), + } + let bytes = std::fs::read(&path).map_err(|error| { + if error.kind() == std::io::ErrorKind::IsADirectory { + CodeGenerationRetentionErrorV1::UnsafeState( + "active code-generation pointer is not a regular file".to_owned(), + ) + } else { + storage(error) + } + })?; serde_json::from_slice(&bytes).map_err(|error| { CodeGenerationRetentionErrorV1::UnsafeState(format!( "active pointer '{}' is corrupt: {error}", 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..15ae71fb0e 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); 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..bc4bfe5588 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", diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs index 3422489f5e..636f838d89 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs @@ -871,6 +871,33 @@ impl DaemonCodeIndexPublicationStoreV1 { CodeIndexPublicationStoreErrorV1::CorruptionResetRequired(error.to_string()) } + /// A pointer slot that is not a regular file is a corrupt authority. + /// + /// `read(2)` and `rename(2)` both report that shape as `EISDIR`. Mapping + /// the OS error to `Unavailable` (or letting it surface as a raw I/O + /// fault) misclassifies a broken publication pointer. Callers in the + /// scheduler publication family must see reset-required corruption. + fn corrupt_non_file_pointer() -> CodeIndexPublicationStoreErrorV1 { + Self::corruption("active code-generation pointer is not a regular file") + } + + fn map_pointer_io(error: std::io::Error) -> CodeIndexPublicationStoreErrorV1 { + if error.kind() == std::io::ErrorKind::IsADirectory { + Self::corrupt_non_file_pointer() + } else { + Self::unavailable(error) + } + } + + fn require_regular_pointer_slot(&self) -> Result<(), CodeIndexPublicationStoreErrorV1> { + match std::fs::metadata(&self.active_path) { + Ok(metadata) if metadata.file_type().is_file() => Ok(()), + Ok(_) => Err(Self::corrupt_non_file_pointer()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(Self::map_pointer_io(error)), + } + } + fn acquire_generation_read_lock( &self, ) -> Result { @@ -1089,8 +1116,11 @@ impl DaemonCodeIndexPublicationStoreV1 { .unwrap_or_else(PoisonError::into_inner) = None; return Ok(None); } - Err(error) => return Err(Self::unavailable(error)), + Err(error) => return Err(Self::map_pointer_io(error)), }; + if !metadata.file_type().is_file() { + return Err(Self::corrupt_non_file_pointer()); + } if metadata.len() > MAX_DURABLE_PUBLICATION_POINTER_BYTES { return Err(Self::corruption( "durable code-generation index exceeds its byte bound", @@ -1102,7 +1132,7 @@ impl DaemonCodeIndexPublicationStoreV1 { // a fixed-width pointer through another path, and a 1-second mtime // filesystem can leave both unchanged while the bytes move. The memo // is reused only when the file digest matches. - let bytes = std::fs::read(&self.active_path).map_err(Self::unavailable)?; + let bytes = std::fs::read(&self.active_path).map_err(Self::map_pointer_io)?; let digest = Self::state_digest(&bytes); { let mut memo = self @@ -2558,8 +2588,11 @@ impl CodeIndexAtomicPublicationPort for DaemonCodeIndexPublicationStoreV1 { std::fs::remove_file(&temporary).map_err(Self::unavailable)?; } hotpath::measure_block!("code_index.generation.publish.pointer_commit", { + // Refuse a directory (or any non-file) before `rename(2)`. Replacing + // one returns EISDIR, which is not a publication-family fault. + self.require_regular_pointer_slot()?; Self::write_durable(&temporary, &bytes)?; - std::fs::rename(&temporary, &self.active_path).map_err(Self::unavailable)?; + std::fs::rename(&temporary, &self.active_path).map_err(Self::map_pointer_io)?; Self::sync_directory( self.active_path .parent() 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/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-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-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/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs index bc1388df7f..e4c972bc6b 100644 --- a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs +++ b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs @@ -1,7 +1,10 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Condvar, Mutex}; -use tracedecay_code_index::production::CodeIndexProductionErrorV1; +use tracedecay_code_index::production::{ + CodeIndexProductionErrorV1, CodeIndexPublicationStoreErrorV1, +}; +use tracedecay_code_index_retention::code_index_generations::try_acquire_code_generation_store_lock; use super::*; @@ -148,15 +151,50 @@ export function GenerationAnchor(value: PublicWidget) { return value; } } fn assert_publication_error(error: CodeIndexSchedulerErrorV1) { + let CodeIndexSchedulerErrorV1::Production(CodeIndexProductionErrorV1::Publication( + CodeIndexPublicationStoreErrorV1::CorruptionResetRequired(detail), + )) = &error + else { + panic!( + "coalesced failure must stay in the scheduler publication family, not an EISDIR misclass: {error:?}" + ); + }; + assert!( + detail.contains("not a regular file"), + "a directory pointer slot is publication corruption, got {detail}" + ); assert!( - matches!( - error, - CodeIndexSchedulerErrorV1::Production(CodeIndexProductionErrorV1::Publication(_)) - ), - "coalesced failure must preserve the production publication error family" + !detail.contains("Is a directory") && !detail.contains("os error 21"), + "publication corruption must not carry the raw EISDIR OS error: {detail}" ); } +/// Hold the only background permit once no pass is in flight. +/// +/// Text seating keeps `reconcile_in_progress` after it drops the scheduler +/// mutex, and that pass can still rename a valid active pointer. A truncated +/// pointer written in that window is not a closed fault. Occupying the permit +/// while the owner has not entered its pass stops that rewrite. +async fn hold_idle_background_admission( + registry: &CodeIndexSchedulerRegistryV1, +) -> tokio::sync::OwnedSemaphorePermit { + let admission = registry.background_reconcile_admission(); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + if registry.memory_stats().await.reconciling_worktrees == 0 + && let Ok(permit) = admission.clone().try_acquire_owned() + && registry.memory_stats().await.reconciling_worktrees == 0 + { + return permit; + } + assert!( + std::time::Instant::now() <= deadline, + "background reconcile did not go idle before publication fault injection" + ); + tokio::time::sleep(Duration::from_millis(2)).await; + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn aborted_flight_owner_wakes_follower_and_allows_a_fresh_owner() { let fixture = fixture(); @@ -242,6 +280,8 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() { let registry = Arc::new(mount(fixture.path(), &store, 1).await); let baseline = latest(®istry, fixture.path()).await; let request = request_for(&baseline, "pkg"); + let idle_admission = hold_idle_background_admission(®istry).await; + registry.clear_pending_wake_for_scope(&request.scope).await; let hold = SchedulerHold::acquire(®istry, fixture.path()).await; let (owner_control, owner_entered) = BlockingNthControl::new(4); @@ -283,22 +323,12 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() { &fixture.path().canonicalize().expect("canonical fixture"), ); let pointer_path = scoped_store.join("active-code-generation-v1.json"); - // Every production writer of the active pointer reads it, edits it in - // memory and renames a temporary over it while holding the exclusive - // generation-store lock. Corrupting the file without that lock races an - // in-flight read-modify-write whose rename then restores a valid pointer, - // and this owner publishes instead of failing closed. The racer is the - // background pass tail: it releases the background admission permit this - // owner then takes (registry/mount.rs, "release the background admission - // permit before HeadOpening / graph work") and keeps attaching the - // generation's text artifact afterwards, so neither the held admission - // nor the held scheduler mutex proves the store is quiet. Taking the - // store lock does: being granted it means no writer is mid-transaction, - // and any writer that starts after it is released reads the corruption - // under the lock and refuses instead of overwriting it. + // Writers rename a temporary over the active pointer while holding the + // generation-store lock. A truncated file is not a closed fault: a pass + // that already read a valid pointer can rename it back. A directory cannot + // be renamed over. Taking the lock first means no writer is mid-transaction + // when the slot stops being a regular file. let pointer_bytes = { - use tracedecay_code_index_retention::code_index_generations::try_acquire_code_generation_store_lock; - let store_lock = tokio::time::timeout(Duration::from_secs(5), async { loop { if let Some(lock) = try_acquire_code_generation_store_lock(&scoped_store) @@ -312,10 +342,16 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() { .await .expect("no generation-store writer is mid-transaction"); let pointer_bytes = std::fs::read(&pointer_path).expect("read active pointer"); - std::fs::write(&pointer_path, b"{").expect("corrupt active pointer"); + // `rename(2)` replaces a truncated file with a valid pointer. A + // directory cannot be renamed over, so the fault stays closed. The + // scheduler must report publication corruption, not the EISDIR that + // read and rename return for that directory. + std::fs::remove_file(&pointer_path).expect("remove active pointer"); + std::fs::create_dir(&pointer_path).expect("replace active pointer with a directory"); drop(store_lock); pointer_bytes }; + drop(idle_admission); owner_control.release(); hold.release(); @@ -332,6 +368,7 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() { assert_publication_error(owner_error); assert_publication_error(follower_error); + std::fs::remove_dir_all(&pointer_path).expect("remove faulted pointer node"); std::fs::write(pointer_path, pointer_bytes).expect("restore active pointer"); registry.shutdown().await; } 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")]