diff --git a/.alfonso/release-notes/next/queued-outside-edit-diagnostics.md b/.alfonso/release-notes/next/queued-outside-edit-diagnostics.md new file mode 100644 index 0000000000..2cf354a146 --- /dev/null +++ b/.alfonso/release-notes/next/queued-outside-edit-diagnostics.md @@ -0,0 +1 @@ +- `aft_inspect` no longer reports Rust files as clean right after another program changes them while AFT's language-server connection is busy; it now waits for the `cargo check` that change needs, or says the check is still running. diff --git a/.alfonso/release-notes/next/views-eviction-waits-for-cleanup-read.md b/.alfonso/release-notes/next/views-eviction-waits-for-cleanup-read.md new file mode 100644 index 0000000000..26431549db --- /dev/null +++ b/.alfonso/release-notes/next/views-eviction-waits-for-cleanup-read.md @@ -0,0 +1 @@ +Disk-limit eviction no longer keeps an idle view because a background cleanup pass happened to be reading its pointer at that moment; it waits for that short read to finish. diff --git a/.alfonso/release-notes/next/views-semantic-ready-snapshot.md b/.alfonso/release-notes/next/views-semantic-ready-snapshot.md new file mode 100644 index 0000000000..dacaff6c83 --- /dev/null +++ b/.alfonso/release-notes/next/views-semantic-ready-snapshot.md @@ -0,0 +1 @@ +- With views on, semantic search no longer briefly answers with text matches only (while reporting the semantic lane as ready) right after new embeddings are published. diff --git a/.github/workflows/_unit-suite.yml b/.github/workflows/_unit-suite.yml index ece771897d..b19064d601 100644 --- a/.github/workflows/_unit-suite.yml +++ b/.github/workflows/_unit-suite.yml @@ -274,7 +274,13 @@ jobs: - name: Publication input gate run: | set -euo pipefail - git fetch --force origin refs/tags/v0.48.1:refs/tags/v0.48.1 + # A runner that briefly cannot reach github.com must not fail the + # merge-queue group: retry the tag fetch before giving up. + for attempt in 1 2 3; do + git fetch --force origin refs/tags/v0.48.1:refs/tags/v0.48.1 && break + if [ "$attempt" = 3 ]; then exit 1; fi + sleep 20 + done node scripts/release-gate-v049.mjs --check-inputs - name: Cargo build (debug — needed by plugin e2e tests) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a312e48341..294d6ab858 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -777,7 +777,15 @@ jobs: working-directory: packages/aft-cli - name: Fetch final v0.48.1 activation baseline - run: git fetch --force origin refs/tags/v0.48.1:refs/tags/v0.48.1 + run: | + set -euo pipefail + # Retry so a brief network failure reaching github.com does not + # fail the release. + for attempt in 1 2 3; do + git fetch --force origin refs/tags/v0.48.1:refs/tags/v0.48.1 && break + if [ "$attempt" = 3 ]; then exit 1; fi + sleep 20 + done - name: Prepare publication records from the tagged candidate run: node scripts/release-gate-v049.mjs --prepare --snapshot-dir "${{ runner.temp }}/governed-publication" --candidate diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 20e289db63..b4f6b15fdb 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -103,7 +103,14 @@ jobs: run: bun run --cwd packages/aft-bridge build - name: Governed-docs aggregate input validator run: | - git fetch --no-tags origin refs/tags/v0.48.1:refs/tags/v0.48.1 + set -euo pipefail + # A runner that briefly cannot reach github.com must not fail the + # check: retry the tag fetch before giving up. + for attempt in 1 2 3; do + git fetch --no-tags origin refs/tags/v0.48.1:refs/tags/v0.48.1 && break + if [ "$attempt" = 3 ]; then exit 1; fi + sleep 20 + done node scripts/release-gate-v049.mjs --check-inputs pr-checks: diff --git a/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_5bc274728f6be348-scoped-inspect-certifies-a-stale-clean-result-af.json b/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_5bc274728f6be348-scoped-inspect-certifies-a-stale-clean-result-af.json new file mode 100644 index 0000000000..ebd574711e --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_5bc274728f6be348-scoped-inspect-certifies-a-stale-clean-result-af.json @@ -0,0 +1,8 @@ +{ + "targeted_mechanism": "none", + "kind": "harness", + "slice_class": "non_ranking", + "fixtures": [ + "harness-goldens" + ] +} diff --git a/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_ccd0193466-replay-tests-complete-pass-wide-budget.json b/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_ccd0193466-replay-tests-complete-pass-wide-budget.json new file mode 100644 index 0000000000..9353fe29e7 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_ccd0193466-replay-tests-complete-pass-wide-budget.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "bash-background-project-replay-tests" + ], + "note": "Test-only: project replay tests in crates/aft/src/bash_background/registry.rs that expect a single replay pass to reach every task now set the per-pass scan time budget to 60 s instead of the 25 ms production limit. No aft_search ranking, fusion, lane admission or truncation change." +} diff --git a/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_cd00f3761a4abf2b-two-queue-flakes-callgraph-store-pin-order-exter.json b/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_cd00f3761a4abf2b-two-queue-flakes-callgraph-store-pin-order-exter.json new file mode 100644 index 0000000000..5f155aa688 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/alfonso-task-bg_cd00f3761a4abf2b-two-queue-flakes-callgraph-store-pin-order-exter.json @@ -0,0 +1,10 @@ +{ + "slice_class": "engine_unwired", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "crates/aft/tests/integration/tool_call_parity_test.rs", + "crates/aft/src/commands/semantic_search/mod.rs" + ], + "note": "The ranking-fenced semantic_search mod.rs change is test-only: the split-request ranking test runs its borrowed-load, disk-check and first-search waits generous so a loaded runner cannot truncate the file list, and a missing host now prints the response. Enumeration, lane selection, scores, fusion, ordering and page cuts are unchanged; benchmark rows must remain byte-identical." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-bash-test-races.json b/benchmarks/aft-search/slice-descriptors/land-bash-test-races.json new file mode 100644 index 0000000000..0b3d504a83 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-bash-test-races.json @@ -0,0 +1,8 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": ["harness-goldens"], + "slices": ["bash-test-race-waits"], + "note": "Test-only: three bash tests now wait on an observable signal instead of racing it (a complete pid in the pid file, the aborted command's start marker, and a replay time budget wide enough to reach the injected recording failure). No product code changed and no changed path matches RANKING_FENCE_PREFIXES." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-borrowed-reload-entry-count.json b/benchmarks/aft-search/slice-descriptors/land-borrowed-reload-entry-count.json new file mode 100644 index 0000000000..65dc880815 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-borrowed-reload-entry-count.json @@ -0,0 +1,14 @@ +{ + "slice_class": "engine_unwired", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "crates/aft/tests/integration/tool_call_parity_test.rs", + "crates/aft/src/search_index.rs", + "crates/aft/src/commands/semantic_search/mod.rs" + ], + "slices": [ + "borrowed-reload-entry-count" + ], + "note": "Test-only. search_index.rs gains a #[cfg(test)] per-root counter of records read on the borrowed-load path; semantic_search/mod.rs and readonly_artifacts.rs tests assert the reload read between 1 and the entry budget instead of timing it. No product code in the fence changes, so ranked rows must remain byte-identical to the committed reference." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-eviction-cleanup-read.json b/benchmarks/aft-search/slice-descriptors/land-eviction-cleanup-read.json new file mode 100644 index 0000000000..cdc1285081 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-eviction-cleanup-read.json @@ -0,0 +1,8 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": ["harness-goldens"], + "slices": ["views-eviction-cleanup-read"], + "note": "Views storage only: disk-limit eviction now waits for this process's background cleanup reads of a view's pointer to close (bounded by its deadline or the registry busy timeout) instead of refusing an idle view as open. No changed path matches RANKING_FENCE_PREFIXES." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-flake-batch-1027.json b/benchmarks/aft-search/slice-descriptors/land-flake-batch-1027.json new file mode 100644 index 0000000000..1bed635c3c --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-flake-batch-1027.json @@ -0,0 +1,15 @@ +{ + "slice_class": "engine_unwired", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "crates/aft/tests/integration/tool_call_parity_test.rs", + "crates/aft/src/commands/semantic_search/mod.rs", + "crates/aft/src/commands/semantic_search/external_disk_check.rs", + "crates/aft/src/search_index.rs" + ], + "slices": [ + "flake-batch-1027" + ], + "note": "Batch of seven merge-queue fixes landed together so their groups stop failing on each other's flakes: #1028 (views-on readiness samples the installed snapshot's index; views-off unchanged), #1029, #1030, #1031, #1032, #1033, #1037. Fence hits are #1028's views-only readiness sample and test-only code in semantic_search and search_index; views-off ranked rows stay byte-identical to the committed reference." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-stderr-exit-hold-test.json b/benchmarks/aft-search/slice-descriptors/land-stderr-exit-hold-test.json new file mode 100644 index 0000000000..df864659fb --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-stderr-exit-hold-test.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "stderr-exit-hold-test" + ], + "note": "Test-only hook (debug builds) that holds the stderr mirror until the exit fence starts, so the slow-writer exit test proves the fence delivers the terminal marker to stderr. No search, ranking or routing code changes." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-tag-fetch-retry.json b/benchmarks/aft-search/slice-descriptors/land-tag-fetch-retry.json new file mode 100644 index 0000000000..98635d5713 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-tag-fetch-retry.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "tag-fetch-retry" + ], + "note": "CI only: the v0.48.1 tag fetch before the publication input gate retries three times, so a runner that briefly cannot reach github.com does not fail a merge-queue group. No engine or ranking files change." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-timing-flakes-batch.json b/benchmarks/aft-search/slice-descriptors/land-timing-flakes-batch.json new file mode 100644 index 0000000000..0709ec08bb --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-timing-flakes-batch.json @@ -0,0 +1,14 @@ +{ + "slice_class": "engine_unwired", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "crates/aft/tests/integration/tool_call_parity_test.rs", + "crates/aft/src/commands/semantic_search/mod.rs", + "crates/aft/src/commands/semantic_search/external_disk_check.rs" + ], + "slices": [ + "timing-flakes-batch" + ], + "note": "The ranking fence hits are test modules only: commands/semantic_search/mod.rs and external_disk_check.rs change only inside #[cfg(test)] code (a contended-search test bounds its wait well above the product budget; a capped disk-check test waits for the borrowed index to load and widens the time limits the test is not about). No product code in the fence changes, so ranked rows must remain byte-identical to the committed reference." +} diff --git a/benchmarks/aft-search/slice-descriptors/land-views-semantic-ready-snapshot.json b/benchmarks/aft-search/slice-descriptors/land-views-semantic-ready-snapshot.json new file mode 100644 index 0000000000..e8ed330e01 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-views-semantic-ready-snapshot.json @@ -0,0 +1,11 @@ +{ + "slice_class": "engine_unwired", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "crates/aft/tests/integration/tool_call_parity_test.rs", + "crates/aft/tests/integration/per_checkout_semantic_runtime.rs", + "crates/aft/src/commands/semantic_search/mod.rs" + ], + "note": "The readiness sample for a views-on root now asks the checkout view for the index it serves for the installed snapshot (building it only when that snapshot has none yet) instead of the owner's scope-wide cache, which a publication resets. Previously a search between a fill's publication and the worker's index rebuild saw no semantic snapshot, planned no semantic lane and answered with lexical hits only. Views-off roots have no checkout view and take the unchanged path, so their ranked rows stay byte-identical; scoring, fusion and paging are untouched." +} diff --git a/crates/aft/src/bash_background/registry.rs b/crates/aft/src/bash_background/registry.rs index 75c2aa13b1..f1c24652fc 100644 --- a/crates/aft/src/bash_background/registry.rs +++ b/crates/aft/src/bash_background/registry.rs @@ -17746,6 +17746,12 @@ mod tests { let db_task_id = "bash-4444444444444444"; let disk_task_id = "bash-5555555555555555"; let (registry, db, _) = registry_with_db_and_frames(storage.path()); + // The assertion below claims one pass reaches every session. A duration + // limit lets a loaded machine stop the scan before it gets there. + registry.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); write_running_project_task(storage.path(), project.path(), db_session, db_task_id); write_running_project_task(storage.path(), project.path(), disk_session, disk_task_id); @@ -17775,6 +17781,12 @@ mod tests { let session = "delivered-project-task"; let task_id = "bash-6666666666666666"; let (registry, _db, _) = registry_with_db_and_frames(storage.path()); + // This asserts the first pass already decided the delivered bundle is not + // replayed. A duration limit would defer that decision to a later pass. + registry.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); write_running_project_task(storage.path(), project.path(), session, task_id); let paths = task_paths(storage.path(), session, task_id).unwrap(); let mut metadata = read_task(&paths.json).unwrap(); @@ -17802,6 +17814,12 @@ mod tests { let project = tempfile::tempdir().unwrap(); let storage = tempfile::tempdir().unwrap(); let (registry, _db, _) = registry_with_db_and_frames(storage.path()); + // The batch is expected to reach the healthy session after parking the + // blocked one. A duration limit could stop the scan in between. + registry.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); let blocked_session = "unreadable-project-session"; let healthy_session = "readable-project-session"; let healthy_task = "bash-000000000000beef"; @@ -18033,6 +18051,8 @@ mod tests { let (registry, db, _) = registry_with_db_and_frames(storage.path()); let (session, mut live) = project_replay_history_fixture(storage.path(), project.path(), &db); + // The pass after the drain is expected to restore the live tasks on its + // own. A duration limit could stop it before reaching their session. registry.set_project_replay_scan_budget_for_test(64, Duration::from_secs(60)); let (mut merged, reads, passes) = drain_project_replay_fixture(®istry, storage.path(), project.path(), &session); @@ -18117,6 +18137,12 @@ mod tests { let storage = tempfile::tempdir().unwrap(); let project = tempfile::tempdir().unwrap(); let (registry, db, _) = registry_with_db_and_frames(storage.path()); + // The first pass is expected to reach the task directory of this session; + // a duration limit could postpone it to a later pass. + registry.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); let session = "cross-harness-project-session"; let task_id = "bash-000000000000cafe"; write_running_project_task(storage.path(), project.path(), session, task_id); @@ -18242,6 +18268,14 @@ mod tests { db.lock().unwrap().execute_batch( "CREATE TRIGGER reject_discovery BEFORE INSERT ON bash_tasks BEGIN SELECT RAISE(FAIL, 'fixture recording refused'); END;" ).unwrap(); + // The product's 25 ms time budget can expire on a slow runner after + // the task is read but before the session directory is exhausted, and + // a budget stop is reported ahead of the recording failure. A wide + // time budget keeps each pass checking the recording-failure report. + registry.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); for _ in 0..2 { let batch = registry.replay_session_from_disk_for_project( storage.path(), @@ -25545,6 +25579,12 @@ mod tests { .persisted_gc_started .store(true, Ordering::SeqCst); if project_scoped { + // This asserts the first project pass already replayed the task. + // A duration limit could postpone that to a later pass. + replay.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); replay .replay_session_for_project(storage.path(), "session", storage.path()) .unwrap(); @@ -25628,6 +25668,12 @@ mod tests { .inner .persisted_gc_started .store(true, Ordering::SeqCst); + // This expects a single pass to reach the task and record the foreign + // acknowledgement. A duration limit could postpone it to a later pass. + replay.set_project_replay_scan_budget_for_test( + PROJECT_REPLAY_SCAN_ENTRY_BUDGET, + Duration::from_secs(60), + ); replay .replay_session_for_project(storage.path(), "session", storage.path()) .unwrap(); diff --git a/crates/aft/src/blob_store/v3/replay.rs b/crates/aft/src/blob_store/v3/replay.rs index d34b1e290d..33ba4b46f1 100644 --- a/crates/aft/src/blob_store/v3/replay.rs +++ b/crates/aft/src/blob_store/v3/replay.rs @@ -7,7 +7,8 @@ //! byte is read or written outside SQLite. //! //! A run may cut power before its N-th mutating call (write, truncate, sync, -//! create or delete). From then on every mutating call fails with an I/O +//! create or delete; creating an empty WAL file is not counted, see +//! `open_file`). From then on every mutating call fails with an I/O //! error and changes nothing, as if the device had gone away; reads still //! see what reached the files before the cut. //! @@ -90,6 +91,7 @@ struct State { dir_ops: Vec, dir_syncs: Vec, asides: u64, + trace: Vec, } /// What one replayed run did before its power cut. @@ -100,6 +102,9 @@ pub(super) struct Report { pub crashed: bool, /// Directories synced before the cut, in order. pub dir_syncs: Vec, + /// Each counted call in order (`write members.sqlite-wal @0+32`, ...), so + /// a test can name where two runs of the same operation diverged. + pub trace: Vec, } fn state() -> MutexGuard<'static, State> { @@ -200,6 +205,7 @@ pub(super) fn finish() -> Report { ops: state.ops, crashed: cut, dir_syncs: std::mem::take(&mut state.dir_syncs), + trace: std::mem::take(&mut state.trace), } } @@ -262,12 +268,14 @@ fn drain_dir_syncs(state: &mut State) { } } -/// Counts one mutating call. `false` means power is already off. -fn step(state: &mut State) -> bool { +/// Counts one mutating call, recording `label` in the run's trace. `false` +/// means power is already off. +fn step(state: &mut State, label: impl FnOnce() -> String) -> bool { drain_dir_syncs(state); if state.crashed { return false; } + state.trace.push(label()); state.ops += 1; if state.crash_at == Some(state.ops) { state.crashed = true; @@ -277,6 +285,13 @@ fn step(state: &mut State) -> bool { true } +fn file_name(path: &Path) -> String { + path.file_name().map_or_else( + || path.display().to_string(), + |name| name.to_string_lossy().into_owned(), + ) +} + fn c_path(name: *const c_char) -> PathBuf { // SAFETY: SQLite passes a NUL-terminated file name. PathBuf::from( @@ -299,7 +314,23 @@ unsafe extern "C" fn open_file( let mut creating = false; if let Some(path) = &path { if flags & ffi::SQLITE_OPEN_CREATE != 0 && !path.exists() { - if !step(&mut state()) { + if flags & ffi::SQLITE_OPEN_WAL != 0 { + // SQLite creates a WAL file empty, even for a read-only + // connection, and a read-only close cannot remove it. Whether + // one is already there depends on what earlier connections to + // the same database did, and on a loaded CI runner it has + // differed between runs of the same test. Counting the + // creation made the number of cut points depend on that + // leftover file. + // An empty file holds no durable state, so no cut point is + // lost: every write to it is still counted, and the power cut + // still removes the file unless its directory was synced. + let mut state = state(); + drain_dir_syncs(&mut state); + if state.crashed { + return ffi::SQLITE_CANTOPEN; + } + } else if !step(&mut state(), || format!("create {}", file_name(path))) { return ffi::SQLITE_CANTOPEN; } creating = true; @@ -354,7 +385,7 @@ unsafe extern "C" fn delete_file( ) -> c_int { let path = c_path(name); let mut state = state(); - if !step(&mut state) { + if !step(&mut state, || format!("delete {}", file_name(&path))) { return ffi::SQLITE_IOERR_DELETE; } if !path.exists() { @@ -437,7 +468,9 @@ unsafe extern "C" fn write( ) -> c_int { let mut state = state(); let (methods, path) = methods_of(&state, file); - if !step(&mut state) { + if !step(&mut state, || { + format!("write {} @{offset}+{amount}", file_name(&path)) + }) { return ffi::SQLITE_IOERR_WRITE; } // SAFETY: the file is open and its original methods are valid. @@ -458,7 +491,9 @@ unsafe extern "C" fn write( unsafe extern "C" fn truncate(file: *mut ffi::sqlite3_file, size: i64) -> c_int { let mut state = state(); let (methods, path) = methods_of(&state, file); - if !step(&mut state) { + if !step(&mut state, || { + format!("truncate {} {size}", file_name(&path)) + }) { return ffi::SQLITE_IOERR_TRUNCATE; } // SAFETY: the file is open and its original methods are valid. @@ -477,7 +512,7 @@ unsafe extern "C" fn truncate(file: *mut ffi::sqlite3_file, size: i64) -> c_int unsafe extern "C" fn sync(file: *mut ffi::sqlite3_file, flags: c_int) -> c_int { let mut state = state(); let (methods, path) = methods_of(&state, file); - if !step(&mut state) { + if !step(&mut state, || format!("sync {}", file_name(&path))) { return ffi::SQLITE_IOERR_FSYNC; } // SAFETY: forward the original SQLite-owned file and synchronization flags. diff --git a/crates/aft/src/blob_store/v3/replay_tests.rs b/crates/aft/src/blob_store/v3/replay_tests.rs index 958bd012e3..61970567e0 100644 --- a/crates/aft/src/blob_store/v3/replay_tests.rs +++ b/crates/aft/src/blob_store/v3/replay_tests.rs @@ -224,7 +224,12 @@ fn enumerate(reuse: bool, prepare: impl Fn(&Fixture)) -> (replay::Report, PathBu let fixture = fixture(reuse); prepare(&fixture); let report = replayed(&fixture, Some(cut)); - assert_eq!(report.crashed, cut <= whole.ops, "cut {cut}"); + assert_eq!( + report.crashed, + cut <= whole.ops, + "cut {cut}: this run made other file calls than the uncut run; {}", + first_divergence(&whole.trace, &report.trace) + ); observe(&fixture) }) .collect::>(); @@ -237,6 +242,27 @@ fn enumerate(reuse: bool, prepare: impl Fn(&Fixture)) -> (replay::Report, PathBu (whole, storage, seen) } +/// Where a cut run's file calls first differ from the uncut run's, with a +/// few calls of context from each. +fn first_divergence(uncut: &[String], cut: &[String]) -> String { + let at = uncut + .iter() + .zip(cut) + .position(|(a, b)| a != b) + .unwrap_or(uncut.len().min(cut.len())); + let window = |trace: &[String]| { + trace[at.saturating_sub(3).min(trace.len())..(at + 5).min(trace.len())].to_vec() + }; + format!( + "first difference at call {} (uncut {} calls, cut run {} calls before its cut): uncut {:?}, cut run {:?}", + at + 1, + uncut.len(), + cut.len(), + window(uncut), + window(cut) + ) +} + fn assert_never_torn(seen: &[Seen]) { let torn: Vec = seen .iter() @@ -314,3 +340,41 @@ fn red_control_without_the_closure_flush_a_power_cut_tears_the_copy() { let (_, _, seen) = result.unwrap(); assert!(seen.contains(&Seen::Torn), "{seen:?}"); } + +/// A plane WAL that an earlier read-only reader left behind (SQLite creates +/// it even for a reader and cannot remove it on a read-only close) must not +/// change how many calls the copy is cut at. Whether that file is still there +/// when a copy starts has differed between runs on a loaded CI runner, and +/// counting its creation then made a cut run disagree with the uncut run. +#[test] +fn replayed_copy_counts_the_same_calls_with_or_without_a_leftover_reader_wal() { + let _guard = serial(); + let wal = |fixture: &Fixture| { + plane_path(&fixture.storage, FAMILY, FamilyPlane::Semantic) + .unwrap() + .with_file_name("semantic.sqlite-wal") + }; + let without = fixture(false); + assert!( + !wal(&without).exists(), + "the fixture's writer removed its WAL" + ); + let without = replayed(&without, None); + let with = fixture(false); + PlaneReader::open_existing(&with.storage, FAMILY, FamilyPlane::Semantic) + .unwrap() + .get(&with.key) + .unwrap(); + assert!( + wal(&with).exists(), + "the read-only reader left its WAL behind" + ); + let with = replayed(&with, None); + assert!(!without.crashed && !with.crashed); + assert_eq!( + without.ops, + with.ops, + "{}", + first_divergence(&without.trace, &with.trace) + ); +} diff --git a/crates/aft/src/commands/bash_kill_scope_tests.rs b/crates/aft/src/commands/bash_kill_scope_tests.rs index a005616771..bea57dcd2d 100644 --- a/crates/aft/src/commands/bash_kill_scope_tests.rs +++ b/crates/aft/src/commands/bash_kill_scope_tests.rs @@ -95,15 +95,8 @@ fn bash_kill_scope_kills_promoted_background_and_grandchildren() { ); let foreground = tasks.start(&stamp, "foreground", "sleep 30", false); assert!(tasks.registry.promote(&foreground, "foreground").unwrap()); - wait_for(|| pid_file.exists()); - let pid: i32 = std::fs::read_to_string(&pid_file) - .unwrap() - .trim() - .parse() - .unwrap(); - assert!(crate::bash_background::process::is_process_alive( - pid as u32 - )); + let pid = read_pid(&pid_file); + assert!(crate::bash_background::process::is_process_alive(pid)); let reply = tasks.cancel(&stamp); assert!(reply.success, "{reply:?}"); let killed = reply.data["killed"].as_array().unwrap(); @@ -132,7 +125,7 @@ fn bash_kill_scope_kills_promoted_background_and_grandchildren() { .take_cancelled_call_notices(session) .is_empty()); } - wait_for(|| !crate::bash_background::process::is_process_alive(pid as u32)); + wait_for(|| !crate::bash_background::process::is_process_alive(pid)); } #[test] @@ -343,12 +336,27 @@ fn bash_kill_scope_next_bash_family_reply_names_cancellation_once() { assert!(!again.contains("cancelled by owner")); } +/// Waits for a pid file written by `echo $! > file` and returns the pid. +/// +/// The shell creates the file before `echo` writes into it, so the file can +/// exist while still empty or partly written. Only a newline-terminated line +/// that parses counts as complete; the bound is generous for slow runners. fn read_pid(path: &std::path::Path) -> u32 { - std::fs::read_to_string(path) - .unwrap() - .trim() - .parse() - .unwrap() + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let contents = std::fs::read_to_string(path).unwrap_or_default(); + if contents.ends_with('\n') { + if let Ok(pid) = contents.trim().parse() { + return pid; + } + } + assert!( + Instant::now() < deadline, + "pid file {} never held a complete pid: {contents:?}", + path.display() + ); + std::thread::sleep(Duration::from_millis(10)); + } } /// A module restart leaves a project's saved tasks out of every live registry @@ -390,7 +398,6 @@ fn bash_kill_scope_stops_saved_task_before_its_project_binds() { let other_pid_file = project.path().join("other"); let task = start(&stamp, "session", &paused_pid_file); let other_task = start(&other, "other-session", &other_pid_file); - wait_for(|| paused_pid_file.exists() && other_pid_file.exists()); let paused_pid = read_pid(&paused_pid_file); let other_pid = read_pid(&other_pid_file); // The module exits: its registry lets go of the tasks without stopping diff --git a/crates/aft/src/commands/glob.rs b/crates/aft/src/commands/glob.rs index 8f7b82e05a..ea79d39925 100644 --- a/crates/aft/src/commands/glob.rs +++ b/crates/aft/src/commands/glob.rs @@ -901,7 +901,14 @@ mod tests { "id": "glob-bulk-lower-bound", "command": "glob", "pattern": "bulk/*.txt" })) .unwrap(); - let response = handle_glob(&request, &ctx); + // This test checks the reported lower bound, not deadline handling. + // A slow CI runner (Windows) can spend the default 5 s request + // budget on 140 per-path checks, so give the request a generous + // budget of its own. + let response = crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(60))), + || handle_glob(&request, &ctx), + ); assert!(response.success, "{response:?}"); let body = &response.data; let lower_bound = body["total_lower_bound"] diff --git a/crates/aft/src/commands/lsp_diagnostics.rs b/crates/aft/src/commands/lsp_diagnostics.rs index 341979d8bc..64c921ec3a 100644 --- a/crates/aft/src/commands/lsp_diagnostics.rs +++ b/crates/aft/src/commands/lsp_diagnostics.rs @@ -95,6 +95,12 @@ pub fn handle_lsp_diagnostics(req: &RawRequest, ctx: &AppContext) -> Response { Err(message) => return Response::error(&req.id, "invalid_request", message), }; + // Watcher changes still queued for the manager lock (an edit made by + // another program, say) have not reached the servers. Until they do, + // rust-analyzer owes no `cargo check` for them, and its old results + // would be read as describing the files as they are now. + ctx.lsp_forward_queued_watcher_changes(); + match (¶ms.file, ¶ms.directory) { (Some(file), None) => handle_file_mode(req, ctx, file, severity_filter, wait_ms), (None, Some(directory)) => { diff --git a/crates/aft/src/commands/semantic_search/external_disk_check.rs b/crates/aft/src/commands/semantic_search/external_disk_check.rs index ac7db07ea4..092b4c5dbb 100644 --- a/crates/aft/src/commands/semantic_search/external_disk_check.rs +++ b/crates/aft/src/commands/semantic_search/external_disk_check.rs @@ -1022,19 +1022,29 @@ mod tests { .unwrap(); } + // The work limit is what cuts each pass short here, so the time + // limits stay out of the way: the check's own budgets are wide, and the + // first search waits for the borrowed index to finish loading on its + // background thread. Without that wait a slow runner answers the first + // search from a degraded empty index, which never runs the disk check. let pass = || { - with_budgets_for_test(Duration::from_secs(5), Duration::from_secs(5), || { - with_work_limit_for_test(2, || { - let before = reindexes_for_test(); - let response = prose_search(&ctx, &root); - (response, reindexes_for_test() - before) - }) - }) + super::super::with_first_search_index_load_wait_budget_for_test( + Duration::from_secs(60), + || { + with_budgets_for_test(Duration::from_secs(60), Duration::from_secs(60), || { + with_work_limit_for_test(2, || { + let before = reindexes_for_test(); + let response = prose_search(&ctx, &root); + (response, reindexes_for_test() - before) + }) + }) + }, + ) }; let (first, first_reads) = pass(); let (second, second_reads) = pass(); - assert_eq!(first_reads, 2); + assert_eq!(first_reads, 2, "{first:#?}"); assert_eq!(first["complete"], false, "{first:#?}"); assert_eq!(first["saved_index_check"]["added"], 4); assert_eq!(first["saved_index_check"]["reread"], 2); @@ -1042,7 +1052,7 @@ mod tests { assert_eq!( second_reads, 2, - "the second pass reads only what the first left" + "the second pass reads only what the first left: {second:#?}" ); assert_eq!(second["saved_index_check"]["complete"], true, "{second:#?}"); assert_eq!(second["saved_index_check"]["reused"], false); diff --git a/crates/aft/src/commands/semantic_search/mod.rs b/crates/aft/src/commands/semantic_search/mod.rs index 7ee71f76ed..322ffd721e 100644 --- a/crates/aft/src/commands/semantic_search/mod.rs +++ b/crates/aft/src/commands/semantic_search/mod.rs @@ -520,16 +520,19 @@ impl extensions::ReadinessSource for RuntimeReadinessSource<'_> { ) { Some(status) => { let status = status.clone(); - // A views-on root reports its checkout view's last built - // index. A readiness sample runs on the search path, so it - // never builds one: the fill worker builds it before the - // lane is served and after every fill, and a search builds - // it if it is still out of date. Views-off roots never have - // a checkout view. + // A views-on root reports the index its checkout view serves + // for the installed snapshot. The fill worker builds it + // before the lane is served and after every fill, so a sample + // normally finds it built, possibly out of date (the search + // itself rebuilds an out-of-date one). Between a publication + // and the worker's rebuild there is none; `serving_index` + // then builds it, because without a snapshot the semantic + // lane would silently drop out of this search. Views-off + // roots never have a checkout view. let checkout_index = matches!(status, SemanticIndexStatus::Ready { .. }) .then(|| self.ctx.checkout_semantic_runtime()) .flatten() - .and_then(|runtime| runtime.cached_index()); + .and_then(|runtime| runtime.serving_index()); if let Some(index) = checkout_index { SemanticReadiness { status, @@ -9912,8 +9915,8 @@ mod tests { } /// A readiness sample runs on the search path, so it reports the views-on - /// checkout index the worker last built and never builds one, even when - /// that index is out of date for the installed snapshot. + /// checkout index the worker last built for the installed snapshot and + /// does not rebuild it, even when that index is out of date. #[test] fn readiness_sample_never_builds_the_checkout_semantic_index() { use extensions::ReadinessSource as _; @@ -9968,6 +9971,72 @@ mod tests { ); } + /// After a fill is folded into a new generation, the worker installs that + /// generation before it rebuilds the scored index. A search in that window + /// must still see the views-on semantic lane as ready: with no snapshot the + /// engine plans no semantic lane and answers lexical hits only, with no + /// symbol names or previews, while the reply still says `semantic: ready`. + #[test] + fn readiness_sample_serves_a_freshly_published_checkout_view() { + use extensions::ReadinessSource as _; + let project = tempfile::tempdir().expect("project"); + let storage = tempfile::tempdir().expect("storage"); + let root = std::fs::canonicalize(project.path()).unwrap(); + std::fs::write(root.join("lib.rs"), "pub fn needle() -> u8 {\n 1\n}\n").unwrap(); + let ctx = test_context(&root); + let slot = ctx.checkout_semantic(); + let (epoch, _wake) = slot.begin(); + let runtime = Arc::new( + crate::views::semantic_runtime::CheckoutSemantic::new( + &crate::root_cache::StoragePaths::same(storage.path()), + "family", + "scope", + &root, + crate::views::semantic::SemanticProducer::current( + "readiness-model", + crate::semantic_index::EmbedTextCaps::default(), + ), + Arc::downgrade(slot), + ) + .unwrap(), + ); + runtime.load().unwrap(); + // Fill and fold, which publishes and installs a new generation, without + // the scored-index rebuild the worker runs afterwards. + let published_before = runtime.installed().generation().name().to_owned(); + runtime + .refresh( + crate::views::semantic::FillBudget::default(), + &mut |texts: Vec| Ok(texts.iter().map(|_| vec![1.0, 0.0]).collect()), + ) + .unwrap(); + assert_ne!( + runtime.installed().generation().name(), + published_before, + "the fill must publish a new generation" + ); + assert!(slot.install(epoch, Arc::clone(&runtime))); + *ctx.semantic_index_status().write().unwrap() = SemanticIndexStatus::ready(); + + let source = RuntimeReadinessSource { ctx: &ctx }; + let observation = source.sample(); + let sampled = observation + .semantic + .snapshot + .as_ref() + .and_then(extensions::SemanticSnapshot::index) + .map(|index| index as *const crate::semantic_index::SemanticIndex); + assert!( + sampled.is_some(), + "a freshly published checkout view must report as ready" + ); + assert_eq!( + sampled, + Some(Arc::as_ptr(&runtime.index().unwrap())), + "readiness must hold the index that search scores for the installed snapshot" + ); + } + /// Mark the semantic lane ready with an empty resident index, so a test /// can exercise the walk routes aft_search takes while the trigram lane /// is not ready. @@ -13025,7 +13094,7 @@ mod tests { .expect("initialize external git fixture"); assert!(git_status.success()); let storage = tempfile::tempdir().expect("storage"); - for file_index in 0..64 { + for file_index in 0..8 { let file = external_root.join(format!( "packages/pkg_{}/src/module_{file_index}.rs", file_index % 8 @@ -13060,7 +13129,7 @@ mod tests { ); assert_eq!(first["success"], true, "initial borrow failed: {first:?}"); - for file_index in 0..40 { + for file_index in 0..8 { let file = external_root.join(format!( "packages/pkg_{}/src/module_{file_index}.rs", file_index % 8 @@ -13074,16 +13143,30 @@ mod tests { let mut rebuilt = SearchIndex::build(&external_root); rebuilt.write_to_disk(&cache_dir, None); - let started = Instant::now(); + // The churned generation holds 8 files, inside the 10-entry budget + // below, while its trigram lookup table holds far more entries, so + // the reload reads the 8 file entries and then stops at the entry + // budget. The time budget is generous on purpose: with time unable + // to stop the reload, a count inside 1..=10 proves the entry budget + // is what stopped it. The lower bound matters too: a probe that + // never saw the reload (for example a root-key mismatch) reads 0 + // and fails loudly instead of passing vacuously. Elapsed time + // cannot show any of this on a loaded runner, where the whole + // search can exceed 1 s even when the reload itself stopped + // immediately. + let records = + crate::search_index::observe_borrowed_load_records_read_for_test(&external_root); let degraded = crate::readonly_artifacts::with_borrowed_search_load_limits_for_test( 10, - Duration::from_secs(5), + Duration::from_secs(60), || response_value(handle_semantic_search(&req, &ctx)), ); + crate::search_index::unobserve_borrowed_load_records_read_for_test(&external_root); assert!( - started.elapsed() < Duration::from_secs(1), - "churned generation must stop at the borrowed-load budget" + (1..=10).contains(&records.load(Ordering::SeqCst)), + "churned generation must stop at the borrowed-load entry budget, read {} records", + records.load(Ordering::SeqCst) ); assert_eq!(degraded["success"], true); assert_ne!(degraded["fully_degraded"], true); @@ -15076,15 +15159,28 @@ mod tests { .write() .unwrap_or_else(std::sync::PoisonError::into_inner); let started = std::time::Instant::now(); - let response = response_value(handle_semantic_search( - &semantic_request("contention_marker", 5), - &ctx, - )); + // A first-search wait far longer than the time bound below: if the + // search ever waited out that budget on the writer-held index instead + // of degrading at once, the bound fails. + let response = + with_first_search_index_load_wait_budget_for_test(Duration::from_secs(120), || { + response_value(handle_semantic_search( + &semantic_request("contention_marker", 5), + &ctx, + )) + }); drop(writer_guard); + // The property is that a contended index degrades with a disclosure + // instead of blocking: this thread holds the write lock until the + // search returns, so a search that blocked on it would never return, + // and one that waited out the first-search budget would take 120 s. + // The bound sits far above the product's per-read 250 ms artifact + // budget so that a loaded runner's slowness cannot fail it. + let elapsed = started.elapsed(); assert!( - started.elapsed() < Duration::from_secs(2), - "contended search exceeded bounded response time" + elapsed < Duration::from_secs(30), + "contended search took {elapsed:?}, far past its bounded response time: {response:#?}" ); assert_eq!(response["success"], true); assert_eq!(response["lexical_only_fallback"], true); @@ -15871,27 +15967,45 @@ mod split_request_tests { save_external_index(&root, storage.path()); } let ctx = external_context(session.path(), storage.path(), false); - let response = search( - &ctx, - serde_json::json!({ - "query": "the scheduler drains the work queue", "pattern": "drain_now", "path": root, - }), - ); + // This test ranks a pattern definition after its host; the + // borrowed-load, disk-check and first-search waits it does not + // test run generous here so a loaded runner cannot truncate the + // file list being ranked. + let search_split = |params: serde_json::Value| { + crate::readonly_artifacts::with_borrowed_search_load_limits_for_test( + 1_000_000, + Duration::from_secs(30), + || { + external_disk_check::with_budgets_for_test( + Duration::from_secs(30), + Duration::from_secs(30), + || { + with_first_search_index_load_wait_budget_for_test( + Duration::from_secs(30), + || search(&ctx, params), + ) + }, + ) + }, + ) + }; + let response = search_split(serde_json::json!({ + "query": "the scheduler drains the work queue", "pattern": "drain_now", "path": root, + })); assert_eq!(response["success"], true, "{response}"); let files = file_names(&response); - let host = files.iter().position(|name| name == "p3.rs").unwrap(); + let Some(host) = files.iter().position(|name| name == "p3.rs") else { + panic!("host p3.rs missing from results: {response}"); + }; assert_eq!(files[host + 1], "defines.rs", "{response}"); assert!(response["results"][host + 1]["supports"] .as_str() .unwrap() .ends_with("p3.rs")); - let page = search( - &ctx, - serde_json::json!({ - "query": "the scheduler drains the work queue", "pattern": "drain_now", "path": root, - "offset": host, "topK": 2, - }), - ); + let page = search_split(serde_json::json!({ + "query": "the scheduler drains the work queue", "pattern": "drain_now", "path": root, + "offset": host, "topK": 2, + })); assert_eq!(file_names(&page), ["p3.rs", "defines.rs"], "{page}"); } } diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index 48e3b14360..94ee894817 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -4686,6 +4686,10 @@ pub struct AppContext { lsp_watcher_forward_slot: crate::lsp::manager::WatcherForwardSlot, /// How many of those helper threads were started, for tests. lsp_watcher_forward_helpers_spawned: AtomicUsize, + /// Held by tests to keep that helper thread from forwarding anything + /// (see [`Self::pause_lsp_watcher_forwarding_for_test`]); never held + /// otherwise. + lsp_watcher_forward_pause: Arc>, #[cfg(test)] lsp_classified_watcher_forwards: parking_lot::Mutex>>>, /// Document and configuration-file changes waiting for the LSP manager @@ -5475,6 +5479,7 @@ impl AppContext { lsp_post_edit_worker_running: Arc::new(AtomicBool::new(false)), lsp_watcher_forward_slot: Default::default(), lsp_watcher_forward_helpers_spawned: AtomicUsize::new(0), + lsp_watcher_forward_pause: Arc::default(), #[cfg(test)] lsp_classified_watcher_forwards: Default::default(), lsp_pending_change_slot: Default::default(), @@ -14490,10 +14495,12 @@ impl AppContext { drop(slot); let manager = Arc::clone(&self.lsp_manager); let forward_slot = Arc::clone(&self.lsp_watcher_forward_slot); + let pause = Arc::clone(&self.lsp_watcher_forward_pause); match std::thread::Builder::new() .name("aft-lsp-watched-files".into()) - .spawn(move || crate::lsp::manager::run_watcher_forward_helper(&manager, &forward_slot)) - { + .spawn(move || { + crate::lsp::manager::run_watcher_forward_helper(&manager, &forward_slot, &pause) + }) { Ok(_) => { self.lsp_watcher_forward_helpers_spawned .fetch_add(1, Ordering::Relaxed); @@ -14506,6 +14513,45 @@ impl AppContext { } } + /// Forward the watcher changes still queued for the LSP manager (see + /// [`Self::lsp_forward_or_queue_watcher_changes`]) now, waiting for its + /// lock, instead of leaving them to the helper thread. + /// + /// Diagnostics readers call this first. A change that is still queued has + /// not reached the language servers: rust-analyzer has neither heard of + /// it nor been asked for the `cargo check` it needs, so nothing marks its + /// published results as describing the files before the change, and a + /// reader would certify them as current. Once forwarded, the check the + /// change asked for counts as running until it reports. + /// + /// The backlog is taken while holding the manager lock, the order the + /// helper uses, so changes reach the servers in the order they were + /// queued. The slot itself stays in place: the helper still owns it, + /// finds it empty, and clears it. + pub(crate) fn lsp_forward_queued_watcher_changes(&self) { + let reloads = { + let mut lsp = self.lsp_manager.lock(); + let Some(backlog) = self + .lsp_watcher_forward_slot + .lock() + .as_mut() + .map(std::mem::take) + else { + return; + }; + backlog.apply(&mut lsp) + }; + crate::lsp::manager::spawn_watcher_rust_workspace_reloads(&self.lsp_manager, reloads); + } + + /// Keep the helper thread that forwards queued watcher changes from + /// forwarding anything until the returned guard is dropped, so a test + /// can act while a change waits in the queue. + #[doc(hidden)] + pub fn pause_lsp_watcher_forwarding_for_test(&self) -> parking_lot::MutexGuard<'_, ()> { + self.lsp_watcher_forward_pause.lock() + } + /// How many helper threads have waited for the LSP manager lock on behalf /// of contended watcher drains. #[doc(hidden)] diff --git a/crates/aft/src/inspect/diagnostics_category.rs b/crates/aft/src/inspect/diagnostics_category.rs index b6eb6335fd..13fec3f42e 100644 --- a/crates/aft/src/inspect/diagnostics_category.rs +++ b/crates/aft/src/inspect/diagnostics_category.rs @@ -604,6 +604,9 @@ fn collect_warm_working_set( .contains_key(&(file.to_path_buf(), server.clone())) }) }; + // Watcher changes still queued for the manager lock have not reached the + // servers; deliver them first so the state read below accounts for them. + ctx.lsp_forward_queued_watcher_changes(); { let mut lsp = ctx.lsp(); // Live language-server diagnostics come from queued events and the diff --git a/crates/aft/src/inspect/scoped_diagnostics_sweep.rs b/crates/aft/src/inspect/scoped_diagnostics_sweep.rs index 5816d179d2..b4423106b4 100644 --- a/crates/aft/src/inspect/scoped_diagnostics_sweep.rs +++ b/crates/aft/src/inspect/scoped_diagnostics_sweep.rs @@ -222,6 +222,10 @@ pub(crate) fn sweep_scoped_files( return sweep; } } + // Watcher changes still queued for the manager lock (an edit made by + // another program, say) have not reached the servers, which would + // otherwise be read as if they had analyzed the files as they are now. + ctx.lsp_forward_queued_watcher_changes(); // A check an earlier request expected and that never began is asked for // again now, so it can run while the files are opened. { diff --git a/crates/aft/src/logging.rs b/crates/aft/src/logging.rs index f06a598d86..39faa7be9b 100644 --- a/crates/aft/src/logging.rs +++ b/crates/aft/src/logging.rs @@ -1616,6 +1616,16 @@ pub fn flush_stderr_on_exit(timeout: std::time::Duration) -> bool { .is_none_or(|mirror| mirror.drain_on_exit(timeout)) } +/// Test-only (debug builds): stop the stderr mirror from writing until +/// [`flush_stderr_on_exit`] starts, so the lines logged in between are still +/// queued when the exit fence runs. Without the fence they would be lost. +#[cfg(debug_assertions)] +pub fn hold_stderr_mirror_until_exit_fence() { + if let Some(mirror) = STDERR_MIRROR.get() { + mirror.hold_until_exit_fence(); + } +} + /// Persist the last process-exit line without depending on the async writer. /// Taking its active sink prevents later rotation or reconfiguration from /// moving the marker. Queued diagnostics may be abandoned once their bounded diff --git a/crates/aft/src/logging/stderr_mirror.rs b/crates/aft/src/logging/stderr_mirror.rs index 9f8e7f2e30..2bcfc2be18 100644 --- a/crates/aft/src/logging/stderr_mirror.rs +++ b/crates/aft/src/logging/stderr_mirror.rs @@ -49,10 +49,44 @@ enum Message { Drain(SyncSender), } +/// Test-only (debug builds): parks the worker so that records logged during +/// process exit are still queued when the exit fence starts. That is the +/// state a busy host can leave them in when the process exits right after +/// logging them, and the fence must still deliver them. +#[cfg(debug_assertions)] +#[derive(Default)] +struct ExitHold { + armed: std::sync::Mutex, + released: std::sync::Condvar, +} + +#[cfg(debug_assertions)] +impl ExitHold { + fn arm(&self) { + *self.armed.lock().unwrap_or_else(|e| e.into_inner()) = true; + } + + fn release(&self) { + *self.armed.lock().unwrap_or_else(|e| e.into_inner()) = false; + self.released.notify_all(); + } + + fn wait(&self) { + let armed = self.armed.lock().unwrap_or_else(|e| e.into_inner()); + // Bounded so a hold that is never released cannot keep a test + // process's worker parked forever. + let _ = self + .released + .wait_timeout_while(armed, Duration::from_secs(10), |armed| *armed); + } +} + #[derive(Clone)] pub(super) struct Mirror { tx: Option>, health: Arc, + #[cfg(debug_assertions)] + exit_hold: Arc, } impl Mirror { @@ -63,10 +97,16 @@ impl Mirror { ) -> io::Result<(Self, JoinHandle<()>)> { let (tx, rx) = mpsc::sync_channel(capacity); let worker_health = Arc::clone(&health); + #[cfg(debug_assertions)] + let exit_hold = Arc::new(ExitHold::default()); + #[cfg(debug_assertions)] + let worker_hold = Arc::clone(&exit_hold); let worker = thread::Builder::new() .name("aft-stderr-mirror".into()) .spawn(move || { while let Ok(message) = rx.recv() { + #[cfg(debug_assertions)] + worker_hold.wait(); // No channel or file-sink lock is held during stderr I/O. let result = match &message { Message::Record(bytes) => sink.write_all(bytes), @@ -92,6 +132,8 @@ impl Mirror { Self { tx: Some(tx), health, + #[cfg(debug_assertions)] + exit_hold, }, worker, )) @@ -99,7 +141,12 @@ impl Mirror { pub(super) fn disabled(health: Arc) -> Self { health.start_failures.fetch_add(1, Ordering::Relaxed); - Self { tx: None, health } + Self { + tx: None, + health, + #[cfg(debug_assertions)] + exit_hold: Arc::default(), + } } pub(super) fn record(&self, bytes: &[u8]) { @@ -121,6 +168,8 @@ impl Mirror { // Only process teardown waits for this fence. Ordinary logger flushes stay // nonblocking, including when stderr is full or its sink is wedged. pub(super) fn drain_on_exit(&self, timeout: Duration) -> bool { + #[cfg(debug_assertions)] + self.exit_hold.release(); let Some(tx) = self.tx.as_ref() else { return false; }; @@ -148,6 +197,13 @@ impl Mirror { .unwrap_or(false) } + /// Test-only (debug builds): stop writing records until `drain_on_exit` + /// starts. See [`ExitHold`]. + #[cfg(debug_assertions)] + pub(super) fn hold_until_exit_fence(&self) { + self.exit_hold.arm(); + } + fn enqueue(&self, message: Message) { let Some(tx) = self.tx.as_ref() else { self.health.drop_message(&message); @@ -227,6 +283,36 @@ mod tests { worker.join().unwrap(); } + #[cfg(debug_assertions)] + #[test] + fn exit_hold_keeps_records_queued_until_the_exit_fence_starts() { + let bytes = Arc::new(Mutex::new(Vec::new())); + let (mirror, worker) = Mirror::spawn( + GateWriter { + bytes: Arc::clone(&bytes), + write_gate: None, + flush_gate: None, + }, + 8, + Arc::new(MirrorHealth::default()), + ) + .unwrap(); + mirror.hold_until_exit_fence(); + mirror.record(b"process exit marker\n"); + thread::sleep(Duration::from_millis(50)); + let written_while_held = bytes.lock().unwrap().clone(); + let drained = mirror.drain_on_exit(Duration::from_secs(5)); + drop(mirror); + worker.join().unwrap(); + assert!( + written_while_held.is_empty(), + "held worker wrote {:?}", + String::from_utf8_lossy(&written_while_held) + ); + assert!(drained, "exit fence did not release the held worker"); + assert_eq!(&*bytes.lock().unwrap(), b"process exit marker\n"); + } + #[test] fn full_queue_drops_whole_records_and_flushes_without_waiting() { let bytes = Arc::new(Mutex::new(Vec::new())); @@ -301,6 +387,8 @@ mod tests { let mirror = Mirror { tx: Some(tx), health: Arc::clone(&health), + #[cfg(debug_assertions)] + exit_hold: Arc::default(), }; mirror.record(&vec![b'x'; MAX_RECORD_BYTES + 1]); assert!(matches!(rx.try_recv(), Err(mpsc::TryRecvError::Empty))); diff --git a/crates/aft/src/lsp/manager.rs b/crates/aft/src/lsp/manager.rs index 914741c28f..45a316cd0a 100644 --- a/crates/aft/src/lsp/manager.rs +++ b/crates/aft/src/lsp/manager.rs @@ -8181,12 +8181,18 @@ impl WatcherForwardBacklog { /// the manager lock, forward everything queued, and repeat while drains /// queued more meanwhile. Clearing the slot under its lock when nothing is /// left is what lets the next contended drain start a new helper, so no -/// change is stranded and at most one helper exists at a time. +/// change is stranded and at most one helper exists at a time. A diagnostics +/// reader can empty the slot first (see +/// `AppContext::lsp_forward_queued_watcher_changes`); the helper then finds +/// nothing left and clears it. `pause` is held only by tests, to keep the +/// helper waiting. pub(crate) fn run_watcher_forward_helper( manager: &Arc>, slot: &WatcherForwardSlot, + pause: &parking_lot::Mutex<()>, ) { loop { + drop(pause.lock()); let mut lsp = manager.lock(); let backlog = { let mut guard = slot.lock(); diff --git a/crates/aft/src/main.rs b/crates/aft/src/main.rs index d3857d0d42..94816d70ab 100644 --- a/crates/aft/src/main.rs +++ b/crates/aft/src/main.rs @@ -406,6 +406,14 @@ fn main() { aft::subc::exit_elapsed().as_millis() ); let flushed = aft::logging::flush_durable_log(flush_wait); + // AFT_TEST_STDERR_EXIT_HOLD keeps every stderr line from here + // on queued until the stderr exit fence below starts, so a test + // can prove the fence delivers them rather than rely on the + // mirror thread happening to run first. Release builds ignore it. + #[cfg(debug_assertions)] + if std::env::var_os("AFT_TEST_STDERR_EXIT_HOLD").is_some() { + aft::logging::hold_stderr_mirror_until_exit_fence(); + } aft::slog_info!( "subc exit phase=log_flush_done flushed={} elapsed_ms={}", flushed, diff --git a/crates/aft/src/readonly_artifacts.rs b/crates/aft/src/readonly_artifacts.rs index 6e0019f092..4f30fd222c 100644 --- a/crates/aft/src/readonly_artifacts.rs +++ b/crates/aft/src/readonly_artifacts.rs @@ -1932,7 +1932,7 @@ mod tests { fn churned_borrowed_generation_stops_at_load_budget() { let _git_env = crate::test_env::hermetic_git_env_guard(); let (_project, root) = fixture_project(); - for file_index in 0..64 { + for file_index in 0..8 { fs::write( root.join(format!("src/module_{file_index}.rs")), format!("pub fn borrowed_budget_{file_index}() {{}}\n"), @@ -1951,7 +1951,7 @@ mod tests { ReadOnlyArtifact::Fresh(_) )); - for file_index in 0..40 { + for file_index in 0..8 { fs::write( root.join(format!("src/module_{file_index}.rs")), format!("pub fn churned_borrowed_budget_{file_index}() {{}}\n"), @@ -1960,17 +1960,27 @@ mod tests { } build_search_artifact(&root, storage.path()); - let started = std::time::Instant::now(); + // The churned generation holds 8 files, inside the 10-entry budget + // below, while its trigram lookup table holds far more entries, so + // the reload reads the 8 file entries and then stops at the entry + // budget. The time budget is generous on purpose: with time unable + // to stop the reload, a count inside 1..=10 proves the entry budget + // is what stopped it. The lower bound matters too: a probe that + // never saw the reload reads 0 and fails loudly instead of passing + // vacuously. Elapsed time cannot show any of this on a loaded + // runner, where even a stopped-early reload can exceed 1 s. + let records = crate::search_index::observe_borrowed_load_records_read_for_test(&root); let opened = open_search_index_from_cache_dir_with_budget( &root, cache_dir, 10, - Duration::from_secs(5), + Duration::from_secs(60), ); - assert_eq!( - started.elapsed() < Duration::from_secs(1), - true, - "record budget must stop before rerooting the churned corpus" + crate::search_index::unobserve_borrowed_load_records_read_for_test(&root); + assert!( + (1..=10).contains(&records.load(std::sync::atomic::Ordering::SeqCst)), + "record budget must stop before rerooting the churned corpus, read {} records", + records.load(std::sync::atomic::Ordering::SeqCst) ); assert!(matches!( opened, diff --git a/crates/aft/src/search_index.rs b/crates/aft/src/search_index.rs index e96cafd0c7..7d1e5b2633 100644 --- a/crates/aft/src/search_index.rs +++ b/crates/aft/src/search_index.rs @@ -314,6 +314,47 @@ impl<'a> BorrowedIndexLoadBudget<'a> { } } +/// Borrowed-load record reads observed per fixture root. A borrowed reload +/// parses on a background thread, so a thread-local counter on the searching +/// thread cannot see it; each test registers its own temporary root and the +/// read loop bumps only that root's counter. Asserting the count stayed at or +/// below the entry budget proves the reload stopped at the entry budget, +/// which a wall-clock bound cannot show on a loaded machine. +#[cfg(test)] +static BORROWED_LOAD_RECORDS_READ_PROBES: OnceLock>>> = + OnceLock::new(); + +/// Register `root` for borrowed-load record observation and return its +/// counter. Callers remove the probe with +/// [`unobserve_borrowed_load_records_read_for_test`] after the assertion. +#[cfg(test)] +pub(crate) fn observe_borrowed_load_records_read_for_test(root: &Path) -> Arc { + let counter = Arc::new(AtomicUsize::new(0)); + BORROWED_LOAD_RECORDS_READ_PROBES + .get_or_init(Mutex::default) + .lock() + .unwrap() + .insert(root.to_path_buf(), Arc::clone(&counter)); + counter +} + +/// Stop observing borrowed-load record reads for `root`. +#[cfg(test)] +pub(crate) fn unobserve_borrowed_load_records_read_for_test(root: &Path) { + if let Some(probes) = BORROWED_LOAD_RECORDS_READ_PROBES.get() { + probes.lock().unwrap().remove(root); + } +} + +#[cfg(test)] +fn note_borrowed_load_record_read_for_test(project_root: &Path) { + if let Some(probes) = BORROWED_LOAD_RECORDS_READ_PROBES.get() { + if let Some(counter) = probes.lock().unwrap().get(project_root) { + counter.fetch_add(1, Ordering::SeqCst); + } + } +} + /// Read an artifact pointer without allowing a writer to strand an interactive request. /// /// Index refreshes publish through `RwLock`s and can legitimately hold a write guard while @@ -2680,6 +2721,10 @@ impl SearchIndex { if unindexed[0] == 1 { unindexed_files.insert(file_id_u32); } + #[cfg(test)] + if borrowed_load_budget.is_some() { + note_borrowed_load_record_read_for_test(&project_root); + } } if !reader_has_remaining(&mut reader, postings_body_end, 8).ok()? { @@ -2764,6 +2809,10 @@ impl SearchIndex { offset, count, }); + #[cfg(test)] + if borrowed_load_budget.is_some() { + note_borrowed_load_record_read_for_test(&project_root); + } } let base = BasePostings { diff --git a/crates/aft/src/subc/mod.rs b/crates/aft/src/subc/mod.rs index b709cee33f..1f83372394 100644 --- a/crates/aft/src/subc/mod.rs +++ b/crates/aft/src/subc/mod.rs @@ -12795,10 +12795,17 @@ pub(crate) mod test_support { if short_deadline { assert_eq!(response.data["status"], "running", "{response:?}"); assert_eq!(response.data["waited"]["reason"], "timeout", "{response:?}"); + // Check which deadline the watch applied rather than how fast a + // loaded runner delivered it. The client's 1 s deadline (received + // 400 ms late) leaves a wait limit under a second or two, while + // the default worker window with timeoutMs=1_800_000 is tens of + // minutes, so 10 s separates the two with room for scheduling lag. + let limit_ms = response.data["waited"]["limit_ms"] + .as_u64() + .unwrap_or_else(|| panic!("watch reply names its wait limit: {response:?}")); assert!( - started.elapsed() < Duration::from_millis(650), - "queued watch missed shortened reply cutoff: {:?}", - started.elapsed() + limit_ms < 10_000, + "queued watch ignored the client deadline: wait limit {limit_ms} ms" ); return; } diff --git a/crates/aft/src/synapse_embed.rs b/crates/aft/src/synapse_embed.rs index 8e2868642a..887ffcaf90 100644 --- a/crates/aft/src/synapse_embed.rs +++ b/crates/aft/src/synapse_embed.rs @@ -582,7 +582,10 @@ pub(crate) mod pause_test_support { let recorded = requests.clone(); let server = std::thread::spawn(move || { tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async move { - tokio::time::timeout(Duration::from_secs(10), async move { + // Safety net only: stops a hung test from blocking forever. The + // session lasts as long as the test's own work, which a loaded + // CI runner (Windows) has stretched past 10 s, so keep it generous. + tokio::time::timeout(Duration::from_secs(60), async move { let listener = tokio::net::TcpListener::from_std(listener).unwrap(); let (mut stream, _) = listener.accept().await.unwrap(); subc_transport::authenticate_server(&mut stream, &key, &daemon_id, "synapse-test", Duration::from_secs(2)).await.unwrap(); diff --git a/crates/aft/src/views/eviction.rs b/crates/aft/src/views/eviction.rs index de87445e4a..32a0eeb063 100644 --- a/crates/aft/src/views/eviction.rs +++ b/crates/aft/src/views/eviction.rs @@ -667,6 +667,20 @@ fn open_in_process(view_dir: &Path, deadline: Option) -> Result, +) -> Option { + let bound = Instant::now() + Duration::from_millis(crate::blob_store::BUSY_TIMEOUT_MS); + let until = deadline.map_or(bound, |deadline| deadline.min(bound)); + super::generation::exclude_cleanup_reads(view_dir, until) +} + /// Why `member` cannot be evicted, read inside the registry barrier `tx` so /// that no registration can land between this check and the deletion. fn hold_inside_barrier( @@ -682,6 +696,9 @@ fn hold_inside_barrier( Ok(true) => return Ok(Some(Hold::Protected)), Err(reason) => return Ok(Some(Hold::Uncertain(reason))), } + let Some(_cleanup_reads) = quiesce_cleanup_reads(&member.view_dir, deadline) else { + return Ok(Some(Hold::OpenInProcess)); + }; match open_in_process(&member.view_dir, deadline) { Ok(false) => Ok(None), Ok(true) => Ok(Some(Hold::OpenInProcess)), @@ -1017,6 +1034,16 @@ pub fn evict_view( )?; return Ok(EvictionOutcome::Held(hold)); } + // No cleanup read may open the pointer from here until the deletion + // is done. Taken before the file-identity guard: a read counts as in + // progress before it takes that guard to open its connection. + let Some(_cleanup_reads) = quiesce_cleanup_reads(¤t.view_dir, deadline) else { + tx.execute( + "UPDATE members SET state = 'active' WHERE scope = ?1", + params![scope], + )?; + return Ok(EvictionOutcome::Held(Hold::OpenInProcess)); + }; // No connection may open a file of this set between the check above // and the deletion below. let _files = crate::db::file_identity::filesystem_guard(); @@ -2482,6 +2509,78 @@ mod tests { assert!(report.over_hard.is_empty()); } + /// After every publication a background cleanup pass reads the view's + /// pointer database to learn whether any older generation is left to + /// delete, without the family registry barrier. Eviction refuses a view + /// while this process has one of its SQLite files open, so it must wait + /// for such a read to close rather than keep a view nobody is using. + #[test] + fn a_cleanup_pass_reading_an_idle_view_never_holds_off_its_eviction() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + #[derive(Debug, PartialEq)] + enum Event { + CleanupReading, + EvictionWaits, + EnforcementDone, + } + + let storage = tempfile::tempdir().unwrap(); + let registry = FamilyRegistry::open(storage.path(), FAMILY).unwrap(); + idle_view(®istry, "a", 1, 4); + let view_dir = crate::views::registry::view_dir(registry.storage(), "a").unwrap(); + // The cleanup pass the publication started is finished; the one below + // is the only cleanup running. + super::super::generation::wait_for_generation_sweep_for_test(&view_dir); + let (events_tx, events) = std::sync::mpsc::channel(); + let (go_tx, go_rx) = std::sync::mpsc::channel::<()>(); + let go_rx = Mutex::new(go_rx); + let once = AtomicBool::new(false); + let hook_events = events_tx.clone(); + let hook = crate::storage_retention::observe_test_hook( + &view_dir, + Arc::new(move |step| { + if step == "pointer-census-open" && !once.swap(true, Ordering::SeqCst) { + // Hold the cleanup read open until the test lets it go. + hook_events.send(Event::CleanupReading).unwrap(); + go_rx + .lock() + .unwrap() + .recv_timeout(Duration::from_secs(15)) + .unwrap(); + } else if step == "eviction-awaits-cleanup-read" { + hook_events.send(Event::EvictionWaits).unwrap(); + } + }), + ); + let view = crate::views::ViewStore::existing_dir(view_dir.clone()) + .unwrap() + .with_sweep_registry(registry.clone()); + super::super::generation::schedule_generation_sweep(view); + let first = events.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(first, Event::CleanupReading); + assert!(open_in_process(&view_dir, None).unwrap()); + + registry.set_disk_budget(budget(0, u64::MAX)); + let enforcing = registry.clone(); + let enforcement = std::thread::spawn(move || { + let report = enforce(&enforcing, EnforceOptions::default()); + events_tx.send(Event::EnforcementDone).unwrap(); + report + }); + // Either eviction waits for the read, or it finished while the read + // was still open and so could not have evicted the view. + let next = events.recv_timeout(Duration::from_secs(10)).unwrap(); + go_tx.send(()).unwrap(); + let report = enforcement.join().unwrap().unwrap(); + super::super::generation::wait_for_generation_sweep_for_test(&view_dir); + drop(hook); + assert_eq!(next, Event::EvictionWaits, "{:?}", report.held); + assert_eq!(evicted(&report), ["a"], "{:?}", report.held); + assert!(!view_dir.exists()); + } + /// A view whose session is alive stays even when it is the least /// recently bound and the limit is not met without it. #[test] diff --git a/crates/aft/src/views/generation.rs b/crates/aft/src/views/generation.rs index 0520263bf2..876628b886 100644 --- a/crates/aft/src/views/generation.rs +++ b/crates/aft/src/views/generation.rs @@ -190,6 +190,154 @@ pub(super) fn with_publication_probe(view_dir: &Path, read: impl FnOnce() -> read() } +/// Disk-limit eviction refuses a view while this process has any of its +/// SQLite files open. Cleanup reads a view's pointer on a short-lived +/// connection, and the read before it knows whether there is anything to +/// clean runs without the family registry barrier eviction checks under (a +/// pass with nothing to clean leaves the registry untouched). This gate lets +/// eviction wait for such reads to close instead of mistaking an idle view +/// for an open one, and keeps new reads out until eviction is done. +#[derive(Default)] +struct CleanupReadState { + reading: usize, + excluded: bool, +} + +#[derive(Default)] +struct CleanupReadGate { + state: Mutex, + changed: std::sync::Condvar, +} + +fn cleanup_read_gate(view_dir: &Path) -> Arc { + type Key = Vec; + static GATES: OnceLock>>> = OnceLock::new(); + let key = crate::production_storage::comparison_key(view_dir, cfg!(windows)); + let mut gates = GATES + .get_or_init(Default::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + gates.retain(|_, gate| gate.strong_count() != 0); + if let Some(gate) = gates.get(&key).and_then(Weak::upgrade) { + return gate; + } + let gate = Arc::new(CleanupReadGate::default()); + gates.insert(key, Arc::downgrade(&gate)); + gate +} + +/// One cleanup read of a view's pointer, from before its connection opens +/// until after it closes. +struct CleanupRead { + gate: Arc, +} + +impl CleanupRead { + fn enter(view_dir: &Path) -> Self { + let gate = cleanup_read_gate(view_dir); + { + let mut state = gate + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + while state.excluded { + state = gate + .changed + .wait(state) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + state.reading += 1; + } + Self { gate } + } +} + +impl Drop for CleanupRead { + fn drop(&mut self) { + let mut state = self + .gate + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.reading -= 1; + drop(state); + self.gate.changed.notify_all(); + } +} + +/// Keeps cleanup reads of one view out while held. See [`exclude_cleanup_reads`]. +pub(super) struct CleanupReadsExcluded { + gate: Arc, +} + +impl Drop for CleanupReadsExcluded { + fn drop(&mut self) { + let mut state = self + .gate + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.excluded = false; + drop(state); + self.gate.changed.notify_all(); + } +} + +/// Waits until no cleanup read of `view_dir` is in progress in this process, +/// and keeps new ones from starting while the returned guard lives. `None` +/// when a read (or another exclusion) remains at `until`. Call it while +/// holding the family registry barrier and before taking the file-identity +/// guard: a read counts as in progress before it takes that guard to open +/// its connection, and it never needs the barrier, so in this order every +/// read being waited for can finish. +pub(super) fn exclude_cleanup_reads( + view_dir: &Path, + until: Instant, +) -> Option { + let gate = cleanup_read_gate(view_dir); + let mut state = gate + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + // The registry barrier already lets only one eviction check of a view + // run at a time in this process; waiting here keeps that true even so. + while state.excluded { + let Some(left) = until.checked_duration_since(Instant::now()) else { + return None; + }; + state = gate + .changed + .wait_timeout(state, left) + .unwrap_or_else(std::sync::PoisonError::into_inner) + .0; + } + state.excluded = true; + #[cfg(test)] + if state.reading != 0 { + drop(state); + crate::storage_retention::test_hook(view_dir, "eviction-awaits-cleanup-read"); + state = gate + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + while state.reading != 0 { + let Some(left) = until.checked_duration_since(Instant::now()) else { + break; + }; + state = gate + .changed + .wait_timeout(state, left) + .unwrap_or_else(std::sync::PoisonError::into_inner) + .0; + } + let closed = state.reading == 0; + drop(state); + let excluded = CleanupReadsExcluded { gate }; + // Dropping the guard lets waiting reads go on when this one gives up. + closed.then_some(excluded) +} + pub(super) fn schedule_generation_sweep(view: ViewStore) { let path = view.view_dir().to_path_buf(); let jobs = GENERATION_SWEEPS.get_or_init(|| Mutex::new(HashMap::new())); @@ -832,10 +980,13 @@ impl ViewStore { /// read-only connection that creates no WAL or SHM file and is closed /// before this returns. fn pointer_census(&self) -> Result { + let _read = CleanupRead::enter(self.view_dir()); let connection = crate::db::file_identity::IdentityConnection::open_census( &self.pointer_path(), "views::generation::census_pointer", )?; + #[cfg(test)] + crate::storage_retention::test_hook(self.view_dir(), "pointer-census-open"); Ok(read_pointer_census(&connection)?) } diff --git a/crates/aft/src/views/semantic_runtime.rs b/crates/aft/src/views/semantic_runtime.rs index 9baa87f782..d5580e38d1 100644 --- a/crates/aft/src/views/semantic_runtime.rs +++ b/crates/aft/src/views/semantic_runtime.rs @@ -622,11 +622,30 @@ impl CheckoutSemantic { .map_err(|error| error.to_string()) } - /// The index this checkout last built, possibly for an older snapshot, - /// without building anything. Readiness reports use it: they run on the - /// search path and only need to know that an index is being served. - pub fn cached_index(&self) -> Option> { - self.plane.cached_index(&self.access) + /// The index searches score for the installed snapshot, for the search + /// engine's readiness snapshot. Returns the one already built for that + /// snapshot, even when later fills made it out of date, so a sample + /// normally builds nothing. It builds one only when the installed snapshot + /// has none yet: right after a fill is published and installed, before the + /// worker rebuilds it, or for a borrower's fresh reader binding. Reporting + /// no index there would make the engine plan no semantic lane and answer + /// with lexical hits alone, and the search that follows would build the + /// same index anyway. + /// + /// The lookup goes through the installed snapshot's reader binding, which + /// is where `search` caches its index. The owner's scope-wide state can + /// differ: a publication replaces it while held snapshots keep the state + /// they were bound to. + pub fn serving_index(&self) -> Option> { + let installed = self.installed(); + self.plane + .cached_index_for(&self.access, &installed) + .or_else(|| { + self.plane + .overlay(&self.access, &self.root, &installed) + .ok() + .map(|overlay| overlay.index) + }) } /// Resident bytes this checkout accounts for: the family arena (shared diff --git a/crates/aft/tests/integration/inspect_command_test.rs b/crates/aft/tests/integration/inspect_command_test.rs index 404361f7b2..e47491a984 100644 --- a/crates/aft/tests/integration/inspect_command_test.rs +++ b/crates/aft/tests/integration/inspect_command_test.rs @@ -9287,6 +9287,134 @@ fn scoped_rust_inspect_reports_a_removed_field_after_an_outside_edit_with_real_r assert_removed_field_reported(&after); } +/// When a watcher drain forwards an outside edit while another thread holds +/// the LSP manager, the change waits in a queue for a helper thread, and +/// until that helper runs rust-analyzer has not heard of the edit, so no +/// `cargo check` is expected for it. The helper is held here, the way a +/// loaded machine can delay it, so the inspect runs while the change is still +/// queued. It must deliver the change itself and then report the compiler +/// error of the check the change starts, or name that check as still +/// running; the stale answer it used to give is a complete result without +/// that error. +#[test] +fn scoped_rust_inspect_forwards_a_queued_outside_edit_before_certifying() { + let (_temp_dir, root, lib) = single_crate_fixture("queued-outside-edit"); + let ctx = configured_context_with_diagnostics_timeout(&root, 40_000); + configure_fake_rust_lsp(&ctx); + ctx.lsp().set_extra_env("AFT_FAKE_LSP_CHECK_ON_SAVE", "0"); + let warm = scoped_diagnostics_inspect(&ctx, "queued-outside-edit-warm", "src/lib.rs"); + assert!(!reported_check_still_running(&warm), "{warm:#}"); + // The fake's own analysis always reports a test error and warning; only + // a check of a file containing `fake_compile_error` adds this one. + assert!( + !diagnostic_messages_for(&warm, "src/lib.rs") + .iter() + .any(|message| message == "fake compile error"), + "control: the file compiles before the outside edit: {warm:#}" + ); + + // Another program adds a compiler error. + fs::write(&lib, "// fake_compile_error\npub fn f() {}\n").unwrap(); + let paused = queue_watcher_change_for_held_helper(&ctx, &lib); + + let after = scoped_diagnostics_inspect(&ctx, "queued-outside-edit-after", "src/lib.rs"); + drop(paused); + if reported_check_still_running(&after) { + return; + } + let text = after["text"].as_str().expect("rendered text"); + assert!( + diagnostic_messages_for(&after, "src/lib.rs") + .iter() + .any(|message| message == "fake compile error"), + "the compiler error of the outside edit is missing: {text}\n{after:#}" + ); +} + +/// Forward the watcher's change to `path` while another thread holds the LSP +/// manager, so it is queued for the helper thread, and keep that helper from +/// delivering it until the returned guard is dropped. +fn queue_watcher_change_for_held_helper<'a>( + ctx: &'a AppContext, + path: &Path, +) -> parking_lot::MutexGuard<'a, ()> { + let paused = ctx.pause_lsp_watcher_forwarding_for_test(); + let (held_tx, held_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + thread::scope(|scope| { + scope.spawn(move || { + let _busy = ctx.lsp(); + held_tx.send(()).unwrap(); + let _ = release_rx.recv(); + }); + held_rx.recv().unwrap(); + ctx.lsp_forward_watcher_file_events(&[path.to_path_buf()]); + release_tx.send(()).unwrap(); + }); + assert!( + ctx.lsp_watcher_forward_pending_for_test(), + "control: the change must still be queued" + ); + paused +} + +/// `lsp_diagnostics` reads the same check state as inspect, so a change +/// still queued for the held helper thread must not be answered from +/// rust-analyzer's results for the file before it: the answer reports the +/// compiler error of the check the change starts, or is incomplete with the +/// note that the check is still running. +#[test] +fn rust_lsp_diagnostics_forwards_a_queued_outside_edit_before_answering() { + let (_temp_dir, root, lib) = single_crate_fixture("queued-outside-edit-lsp"); + let ctx = configured_context_with_diagnostics_timeout(&root, 40_000); + configure_fake_rust_lsp(&ctx); + ctx.lsp().set_extra_env("AFT_FAKE_LSP_CHECK_ON_SAVE", "0"); + let lsp_diagnostics = |id: &str| -> Value { + serde_json::to_value(aft::commands::lsp_diagnostics::handle_lsp_diagnostics( + &request(json!({ + "id": id, + "command": "lsp_diagnostics", + "file": lib.display().to_string(), + "wait_ms": 10_000, + })), + &ctx, + )) + .expect("response serializes") + }; + let compiler_error_reported = |response: &Value| { + response["diagnostics"] + .as_array() + .into_iter() + .flatten() + .any(|diagnostic| diagnostic["message"] == "fake compile error") + }; + let warm = lsp_diagnostics("queued-outside-edit-lsp-warm"); + assert_eq!(warm["success"], true, "{warm:#}"); + assert!( + !compiler_error_reported(&warm), + "control: the file compiles before the outside edit: {warm:#}" + ); + + // Another program adds a compiler error. + fs::write(&lib, "// fake_compile_error\npub fn f() {}\n").unwrap(); + let paused = queue_watcher_change_for_held_helper(&ctx, &lib); + + let after = lsp_diagnostics("queued-outside-edit-lsp-after"); + drop(paused); + assert_eq!(after["success"], true, "{after:#}"); + if after["complete"] == false && after["note"].as_str().is_some() { + assert_eq!( + after["note"], "rust-analyzer: cargo check still running; retry", + "{after:#}" + ); + return; + } + assert!( + compiler_error_reported(&after), + "the compiler error of the outside edit is missing: {after:#}" + ); +} + /// The opposite direction: a compiler error present when rust-analyzer /// starts (its first `cargo check` reports it) is then fixed with an AFT /// edit. `use of moved value` comes only from the compiler, never from diff --git a/crates/aft/tests/integration/per_checkout_callgraph.rs b/crates/aft/tests/integration/per_checkout_callgraph.rs index ac6b17949d..d97bc524c5 100644 --- a/crates/aft/tests/integration/per_checkout_callgraph.rs +++ b/crates/aft/tests/integration/per_checkout_callgraph.rs @@ -84,6 +84,10 @@ fn query_fixture(with_graph: bool) -> QueryFixture { }; let views = view.view_store().unwrap(); let name = GenerationName::for_manifest(&manifest).unwrap(); + // Pin the generation before materializing its derived database, as + // production publication does: the post-publication sweep scheduled + // by an earlier commit would otherwise reclaim the unpinned file. + let prepared = views.prepare_v2(&name, None, &manifest, None).unwrap(); let database = views.derived_path(&name.to_string()).unwrap(); std::fs::create_dir_all(database.parent().unwrap()).unwrap(); let store = view.open_store(FamilyPlane::Callgraph).unwrap(); @@ -93,7 +97,6 @@ fn query_fixture(with_graph: bool) -> QueryFixture { &BlobReader(store.reader()), ) .unwrap(); - let prepared = views.prepare_v2(&name, None, &manifest, None).unwrap(); views.commit_v2(prepared, None).unwrap(); let marker = aft::root_cache::ReadMarker::create(view.view_dir(), &name.to_string()).unwrap(); @@ -329,6 +332,12 @@ impl FirstLoadDriver for Driver { } let name = GenerationName::for_manifest(&manifest).map_err(error)?; let views = view.view_store().map_err(error)?; + // Pin the generation before materializing its derived database, as + // production publication does: the post-publication sweep scheduled + // by an earlier commit would otherwise reclaim the unpinned file. + let prepared = views + .prepare_v2(&name, None, &manifest, None) + .map_err(error)?; let database = views.derived_path(&name.to_string()).map_err(error)?; std::fs::create_dir_all(database.parent().unwrap()).map_err(error)?; callgraph::materialize( @@ -336,9 +345,6 @@ impl FirstLoadDriver for Driver { &OpenGeneration::new("projected", manifest.clone(), None), &BlobReader(store.reader()), )?; - let prepared = views - .prepare_v2(&name, None, &manifest, None) - .map_err(error)?; views.commit_v2(prepared, None).map_err(error)?; let marker = aft::root_cache::ReadMarker::create(view.view_dir(), &name.to_string()) .map_err(error)?; @@ -486,6 +492,12 @@ fn store_clone_pins_old_blobs_after_swap_and_gc() { ); let views = view.view_store().unwrap(); let name = GenerationName::for_manifest(&empty).unwrap(); + // Pin the generation before materializing its derived database, as + // production publication does: the post-publication sweep scheduled by + // the old generation's commit would otherwise reclaim the unpinned file. + let prepared = views + .prepare_v2(&name, Some(&old_name), &empty, None) + .unwrap(); let database = views.derived_path(&name.to_string()).unwrap(); let store = view.open_store(FamilyPlane::Callgraph).unwrap(); callgraph::materialize( @@ -494,9 +506,6 @@ fn store_clone_pins_old_blobs_after_swap_and_gc() { &BlobReader(store.reader()), ) .unwrap(); - let prepared = views - .prepare_v2(&name, Some(&old_name), &empty, None) - .unwrap(); views.commit_v2(prepared, None).unwrap(); let marker = aft::root_cache::ReadMarker::create(view.view_dir(), &name.to_string()).unwrap(); let successor = LiveDelta::new(Arc::new(OpenGeneration::new( diff --git a/crates/aft/tests/integration/subc_bridge_test.rs b/crates/aft/tests/integration/subc_bridge_test.rs index 2e4a555d7f..04ee6a6824 100644 --- a/crates/aft/tests/integration/subc_bridge_test.rs +++ b/crates/aft/tests/integration/subc_bridge_test.rs @@ -5233,20 +5233,27 @@ async fn drive_bash_abort_inflight_daemon(input: FakeDaemonInput) { } = open_fake_daemon_session(input).await; bind_route1(&mut stream, &root1).await; + // The command writes the marker once it is running. The release file is + // never created, so only the abort (or the 5 s command timeout) ends it. + let marker = root1.join("abort-inflight-started"); + let release = root1.join("abort-inflight-release"); send_tool_call( &mut stream, 1, 130, "bash", json!({ - "command": "sleep 30", + "command": hold_bash_until_release_command(&marker, &release, "abort-done"), "foreground_orchestrate": true, "wait": true, "timeout": 5_000, }), ) .await; - tokio::time::sleep(Duration::from_millis(300)).await; + // An abort only kills calls already registered as in flight, and a call + // registers when its command starts. Waiting a fixed time lost that race + // on a slow Windows runner; wait for the running command's marker instead. + wait_for_path_event(&marker, "aborted bash command start").await; send_tool_call( &mut stream, 1, diff --git a/crates/aft/tests/integration/subc_detach_test.rs b/crates/aft/tests/integration/subc_detach_test.rs index 0bfaa2afa3..324e2e88c3 100644 --- a/crates/aft/tests/integration/subc_detach_test.rs +++ b/crates/aft/tests/integration/subc_detach_test.rs @@ -847,6 +847,11 @@ fn drain_with_live_lsp_servers_and_writer( if writer_delay == "1000" { assert!(log.contains("phase=log_flush_done flushed=false"), "slow writer must exhaust the async flush budget: {}", log_tail(&log)); + // The terminal marker is mirrored to stderr after the queued + // flush; the stderr exit fence must deliver it there as well as + // to the durable file checked below. + assert!(log.contains("subc exit phase=process_exit mode="), + "stderr exit fence must deliver the terminal marker: {}", log_tail(&log)); } // Print phase and LSP summary lines. The bound below uses the LSP // phase's own elapsed value, never the outer process-exit observation. @@ -1185,6 +1190,11 @@ impl ModuleProcess { // A writer this slow cannot acknowledge a flush inside the exit // budget, wherever in its batch delay the request lands. command.env("AFT_TEST_LOG_FLUSH_HOLD_MS", hold); + // Keep the stderr mirror from writing the exit's last lines until + // the stderr exit fence starts, so the stderr assertions prove the + // fence delivers them instead of depending on the mirror thread + // being scheduled before the process exits. + command.env("AFT_TEST_STDERR_EXIT_HOLD", "1"); } if let Some(bin_dir) = bin_dir { // The test supplies its language servers through `bin_dir`. An diff --git a/packages/aft-bridge/src/__tests__/lsp-github-installer-regressions.test.ts b/packages/aft-bridge/src/__tests__/lsp-github-installer-regressions.test.ts index 6fe190c4f9..a5bfe118be 100644 --- a/packages/aft-bridge/src/__tests__/lsp-github-installer-regressions.test.ts +++ b/packages/aft-bridge/src/__tests__/lsp-github-installer-regressions.test.ts @@ -96,7 +96,8 @@ describe("archive prechecks", () => { execFileSync("tar", ["-cf", archive, "-C", root, name]); expect(await installerFixture()._precheckTarForTesting(archive, "tar")).toBe(4); }); - }); + // Spawns tar twice; a loaded CI runner can exceed bun's 5 s default. + }, 30_000); }); describe("cache publication already on main", () => { @@ -155,6 +156,8 @@ describe("cache publication already on main", () => { expect(existsSync(join(root, "lsp-binaries", ".quarantine"))).toBe(false); }); }, + // Spawns tar and runs an install; a loaded CI runner can exceed bun's 5 s default. + 30_000, ); test("cache validation cannot quarantine a leased GitHub installation", async () => {