diff --git a/.alfonso/release-notes/next/batch-3-startup-fix.md b/.alfonso/release-notes/next/batch-3-startup-fix.md new file mode 100644 index 0000000000..9ea6feb636 --- /dev/null +++ b/.alfonso/release-notes/next/batch-3-startup-fix.md @@ -0,0 +1 @@ +- AFT no longer stops answering for several seconds after a restart when its database holds a long history of shell commands. diff --git a/.alfonso/release-notes/next/rust-1-100-layout.md b/.alfonso/release-notes/next/rust-1-100-layout.md new file mode 100644 index 0000000000..0551063bc1 --- /dev/null +++ b/.alfonso/release-notes/next/rust-1-100-layout.md @@ -0,0 +1 @@ +Rust diagnostics: completed Cargo checks can be saved and restored with Rust 1.100's new intermediate-artifact layout, including projects with build scripts. Rust 1.99's layout remains supported. diff --git a/Cargo.lock b/Cargo.lock index ba850eac61..b4da7eb795 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -84,6 +84,7 @@ dependencies = [ "cortexkit-role-tool-display", "cortexkit-role-tool-provider", "cortexkit-role-tool-provider-conformance", + "cortexkit-test-support", "cortexkit-tree-sitter-scss", "crc32fast", "crossbeam-channel", @@ -929,6 +930,19 @@ dependencies = [ "serde", ] +[[package]] +name = "cortexkit-test-support" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50773b2a265a84af63d31ed52653cf07cf5690a85b0b23d65a3d8f1e776ef2e9" +dependencies = [ + "file-id", + "rustix", + "sha2", + "subc-os", + "tokio", +] + [[package]] name = "cortexkit-tree-sitter-scss" version = "1.0.1" @@ -1370,6 +1384,15 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "file-id" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1fc6a637b6dc58414714eddd9170ff187ecb0933d4c7024d1abbd23a3cc26e9" +dependencies = [ + "windows-sys 0.60.2", +] + [[package]] name = "filedescriptor" version = "0.8.3" diff --git a/benchmarks/aft-search/slice-descriptors/ci-speed-batch-3-startup-fix.json b/benchmarks/aft-search/slice-descriptors/ci-speed-batch-3-startup-fix.json new file mode 100644 index 0000000000..ede178619b --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/ci-speed-batch-3-startup-fix.json @@ -0,0 +1,7 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": ["harness-goldens"], + "note": "Follow-up to post-0.64 batch 3: startup held-task recovery reads in bounded indexed pages, database sections of the permission-card and shells paths move off the subc runtime thread, and Rust 1.100 Cargo layout support for completed Rust checks and native test launchers. No ranking-fence file changes; search is untouched." +} diff --git a/crates/aft/Cargo.toml b/crates/aft/Cargo.toml index be8350b7ee..47a66355ee 100644 --- a/crates/aft/Cargo.toml +++ b/crates/aft/Cargo.toml @@ -297,6 +297,7 @@ windows-sys = { version = "0.61", features = [ ] } [dev-dependencies] +cortexkit-test-support = "=0.1.14" graphql-parser = "0.4.1" cortexkit-role-tool-provider-conformance = "=0.15.6" cortexkit-role-tool-provider = { version = "=0.15.5", features = ["test-vectors"] } diff --git a/crates/aft/src/bash_background/secrets_tests.rs b/crates/aft/src/bash_background/secrets_tests.rs index 2b2b64f6b3..74dfa70a0e 100644 --- a/crates/aft/src/bash_background/secrets_tests.rs +++ b/crates/aft/src/bash_background/secrets_tests.rs @@ -597,19 +597,27 @@ fn secret_bash_windows_refuses_all_profiles_without_a_job() { } #[cfg(unix)] -fn native_launcher(root: &Path) -> impl Drop { - // The test executable runs from `/deps/`, and `aft` is built - // into `/`. With Cargo's `build.build-dir` set, test - // executables move under the build folder while binaries stay in the - // target folder, so the same profile folder there is looked in too. - let executable = std::env::current_exe().unwrap(); - let profile_dir = executable.parent().unwrap().parent().unwrap(); +fn native_launcher_candidates( + executable: &Path, + provided: Option, + target: Option, + manifest: Option, +) -> Vec { + if let Some(provided) = provided { + return vec![provided]; + } + // Nextest only supplies BIN_EXE variables to integration tests and benches, + // not library tests. Locate the profile above Cargo's old `deps` or new + // `build//` intermediate tree in the extracted archive. + let Some(profile_dir) = cortexkit_test_support::cargo_profile_dir_of(executable) else { + return Vec::new(); + }; let profile = profile_dir.file_name().unwrap(); let mut candidates = vec![profile_dir.join("aft")]; - if let Some(target) = std::env::var_os("CARGO_TARGET_DIR") { + if let Some(target) = target { candidates.push(std::path::PathBuf::from(target).join(profile).join("aft")); } - if let Some(manifest) = std::env::var_os("CARGO_MANIFEST_DIR") { + if let Some(manifest) = manifest { candidates.push( std::path::PathBuf::from(manifest) .join("../../target") @@ -617,6 +625,20 @@ fn native_launcher(root: &Path) -> impl Drop { .join("aft"), ); } + candidates +} + +#[cfg(unix)] +fn native_launcher(root: &Path) -> impl Drop { + let candidates = native_launcher_candidates( + &std::env::current_exe().unwrap(), + std::env::var_os("AFT_TEST_AFT_BINARY") + .or_else(|| std::env::var_os("NEXTEST_BIN_EXE_aft")) + .or_else(|| std::env::var_os("CARGO_BIN_EXE_aft")) + .map(std::path::PathBuf::from), + std::env::var_os("CARGO_TARGET_DIR"), + std::env::var_os("CARGO_MANIFEST_DIR"), + ); let launcher = candidates .iter() .find(|path| path.is_file()) @@ -627,6 +649,48 @@ fn native_launcher(root: &Path) -> impl Drop { crate::sandbox_spawn::install_sandbox_launcher_for_test(root, launcher) } +#[cfg(unix)] +#[test] +fn native_launcher_candidates_support_both_cargo_layouts() { + for executable in [ + "/archive/target/debug/deps/agent_file_tools-hash", + "/archive/target/debug/build/agent-file-tools/hash/aft", + "/archive/target/debug/build/agent-file-tools/hash/out/aft", + ] { + let candidates = native_launcher_candidates( + Path::new(executable), + None, + Some("/custom-target".into()), + Some("/checkout/crates/aft".into()), + ); + assert_eq!( + candidates, + vec![ + std::path::PathBuf::from("/archive/target/debug/aft"), + std::path::PathBuf::from("/custom-target/debug/aft"), + std::path::PathBuf::from("/checkout/crates/aft/../../target/debug/aft"), + ], + "{executable}" + ); + } +} + +#[cfg(unix)] +#[test] +fn native_launcher_candidates_respect_explicit_binary_even_if_missing() { + let binary = std::path::PathBuf::from("/missing/explicit-aft"); + assert_eq!( + native_launcher_candidates( + Path::new("/archive/target/debug/build/agent-file-tools/hash/aft"), + Some(binary.clone()), + Some("/custom-target".into()), + None, + ), + vec![binary], + "a missing explicit binary must fail rather than use a stale fallback" + ); +} + #[cfg(unix)] #[test] fn secret_bash_real_plan_precedence_and_untrusted_request_scope() { diff --git a/crates/aft/src/list_surfaces.rs b/crates/aft/src/list_surfaces.rs index fb229cabde..80cfb2b0a0 100644 --- a/crates/aft/src/list_surfaces.rs +++ b/crates/aft/src/list_surfaces.rs @@ -436,6 +436,12 @@ pub static EXCLUSIONS: &[ExclusionEntry] = &[ location_or_primitive: "SQL LIMIT, PER_TICK (4)", reason: "Keyset pagination bounds background lifecycle probes, not permissions.list; every live grant remains listed and later maintenance ticks continue after the saved cursor.", }, + ExclusionEntry { + file: "subc/held_task_recovery.rs", + enclosing_item: "sessions", + location_or_primitive: "RECOVERY_CHUNK (64), SQL LIMIT, keyset pagination", + reason: "Internal startup recovery exhausts every candidate source in index-seeking pages; the chunk bounds one shared database mutex hold, not an agent-visible result list.", + }, ExclusionEntry { file: "subc/shells/runtime.rs", enclosing_item: "list, project_row, output", diff --git a/crates/aft/src/lsp/completed_rust_check.rs b/crates/aft/src/lsp/completed_rust_check.rs index b267ceb805..bcc9274ab3 100644 --- a/crates/aft/src/lsp/completed_rust_check.rs +++ b/crates/aft/src/lsp/completed_rust_check.rs @@ -886,43 +886,59 @@ impl Runtime { profiles.push(triple.path().join("release/build")); } } - let prefix = format!("{}-", config.get("name")?.as_str()?); + let package_name = config.get("name")?.as_str()?; + let prefix = format!("{package_name}-"); let mut declared = false; for profile in profiles { - let mut entries = match fs::read_dir(profile) { - Ok(entries) => entries, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, - Err(_) => return None, - }; - loop { - expired(deadline)?; - let Some(entry) = entries.next() else { - break; + // Cargo 1.100 groups build units by package instead of keeping + // build-script runs in package-hash directories. Their captured + // stdout moved from `output` to `//run/stdout`. + let mut directories = vec![(profile, false)]; + while let Some((directory, package_units)) = directories.pop() { + let mut entries = match fs::read_dir(directory) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(_) => return None, }; - let entry = entry.ok()?; - if !entry.file_name().to_str()?.starts_with(&prefix) { - continue; - } - let output = entry.path().join("output"); - if !output.is_file() { - continue; - } - let bytes = read_bounded(&output, deadline)?; - for line in std::str::from_utf8(&bytes).ok()?.lines() { + loop { expired(deadline)?; - if let Some(key) = line - .strip_prefix("cargo:rerun-if-env-changed=") - .or_else(|| line.strip_prefix("cargo::rerun-if-env-changed=")) - { - declared = true; - env_keys.push(key.to_string()); + let Some(entry) = entries.next() else { + break; + }; + let entry = entry.ok()?; + let name = entry.file_name(); + let name = name.to_str()?; + if !package_units && name == package_name { + directories.push((entry.path(), true)); + continue; } - if let Some(path) = line - .strip_prefix("cargo:rerun-if-changed=") - .or_else(|| line.strip_prefix("cargo::rerun-if-changed=")) - { - declared = true; - outputs.push(canonical(&package.join(path))?); + let output = if package_units { + entry.path().join("run/stdout") + } else if name.starts_with(&prefix) { + entry.path().join("output") + } else { + continue; + }; + if !output.is_file() { + continue; + } + let bytes = read_bounded(&output, deadline)?; + for line in std::str::from_utf8(&bytes).ok()?.lines() { + expired(deadline)?; + if let Some(key) = line + .strip_prefix("cargo:rerun-if-env-changed=") + .or_else(|| line.strip_prefix("cargo::rerun-if-env-changed=")) + { + declared = true; + env_keys.push(key.to_string()); + } + if let Some(path) = line + .strip_prefix("cargo:rerun-if-changed=") + .or_else(|| line.strip_prefix("cargo::rerun-if-changed=")) + { + declared = true; + outputs.push(canonical(&package.join(path))?); + } } } } @@ -1894,6 +1910,52 @@ mod tests { cache.validated_with_runtime("v1".into(), Instant::now() + BUDGET) } + #[test] + fn build_inputs_support_legacy_and_unit_layouts() { + for profile in ["debug", "release", "some-target/debug"] { + for output in ["member-control/output", "member/control/run/stdout"] { + let (_temp, mut cache) = fixture(); + cache.runtime.env.insert( + "CARGO_TARGET_DIR".into(), + cache.root.join("target").to_string_lossy().into_owned(), + ); + cache.runtime.options = Some(serde_json::json!({"cargo": {"targetDir": true}})); + write(&cache.root, "member/input.txt", "build input"); + let inputs = &cache.saved.as_ref().unwrap().fingerprint; + let collect = || { + cache + .runtime + .build_inputs(&cache.root, inputs, Instant::now() + BUDGET) + }; + assert!( + collect().is_none(), + "missing build output must not certify inputs" + ); + write( + &cache.root, + &format!("target/rust-analyzer/{profile}/build/{output}"), + "cargo:rerun-if-changed=input.txt\ncargo::rerun-if-env-changed=BUILD_INPUT\n", + ); + // Other packages must not add declarations to this package's + // certificate, even when their names share its prefix. + write( + &cache.root, + &format!( + "target/rust-analyzer/{profile}/build/member-other/control/run/stdout" + ), + "cargo:rerun-if-changed=nonexistent.txt\n", + ); + let (paths, env) = collect().expect("discover declared build inputs"); + assert_eq!( + paths, + vec![canonical(&cache.root.join("member/input.txt")).unwrap()], + "{profile}/{output}" + ); + assert_eq!(env, vec!["BUILD_INPUT"], "{profile}/{output}"); + } + } + } + #[test] fn next_write_removes_record_for_deleted_checkout() { let (temp, cache) = fixture(); diff --git a/crates/aft/src/subc/consent.rs b/crates/aft/src/subc/consent.rs index ecfd871708..47b00c8862 100644 --- a/crates/aft/src/subc/consent.rs +++ b/crates/aft/src/subc/consent.rs @@ -3562,7 +3562,7 @@ fn macos_request_params(folder: &str, program: &str, steps: &str) -> Value { json!({"kind":"macos_permission","folder":folder,"program":program,"steps":steps}) } -#[cfg(target_os = "macos")] +#[cfg(any(target_os = "macos", test))] pub(super) async fn withdraw_macos_permission_cards( db: Arc>, ) -> Result { @@ -3570,8 +3570,11 @@ pub(super) async fn withdraw_macos_permission_cards( let transport = transport().ok_or_else(|| ConsentError::Unavailable("Cingulate is unavailable".into()))?; let _serial = MACOS_CARDS_LOCK.lock().await; - let rows = { - let conn = db + let read_db = Arc::clone(&db); + // The subc runtime has one thread. Even waiting for the shared database + // mutex here would stop frame polling while startup recovery reads a page. + let rows = tokio::task::spawn_blocking(move || { + let conn = read_db .lock() .map_err(|e| ConsentError::Unavailable(e.to_string()))?; let mut statement = conn.prepare("SELECT folder,program,selector,project_root,session,elicitation_id FROM macos_permission_cards") @@ -3589,8 +3592,10 @@ pub(super) async fn withdraw_macos_permission_cards( }) .map_err(|e| ConsentError::Unavailable(e.to_string()))?; rows.collect::, _>>() - .map_err(|e| ConsentError::Unavailable(e.to_string()))? - }; + .map_err(|e| ConsentError::Unavailable(e.to_string())) + }) + .await + .map_err(|e| ConsentError::Unavailable(e.to_string()))??; let mut withdrawn = 0; for (folder, program, scope, root, session, id) in rows { let had_id = id.is_some(); @@ -3629,13 +3634,19 @@ pub(super) async fn withdraw_macos_permission_cards( Err(error) => return Err(error), } } - db.lock() - .map_err(|e| ConsentError::Unavailable(e.to_string()))? - .execute( - "DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2", - params![folder, program], - ) - .map_err(|e| ConsentError::Unavailable(e.to_string()))?; + let delete_db = Arc::clone(&db); + let delete_folder = folder.clone(); + tokio::task::spawn_blocking(move || { + delete_db.lock() + .map_err(|e| ConsentError::Unavailable(e.to_string()))? + .execute( + "DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2", + params![delete_folder, program], + ) + .map_err(|e| ConsentError::Unavailable(e.to_string())) + }) + .await + .map_err(|e| ConsentError::Unavailable(e.to_string()))??; Ok(()) } .await; @@ -3670,19 +3681,26 @@ pub(super) async fn macos_permissions( let program = "ck-aft"; for observation in observations { let folder = observation.folder.to_string_lossy().into_owned(); - let row: Option<(String, String, String, Option)> = { - let conn = db + let read_db = Arc::clone(&db); + let read_folder = folder.clone(); + let read_stamp = stamp.clone(); + let read_identity = identity.clone(); + let readable = observation.readable; + let row: Option<(String, String, String, Option)> = tokio::task::spawn_blocking(move || { + let conn = read_db .lock() .map_err(|e| ConsentError::Unavailable(e.to_string()))?; - if !observation.readable { + if !readable { // The existing database column is not a wire idempotency key: // informational cards use only the folder/program pair. - conn.execute("INSERT OR IGNORE INTO macos_permission_cards(folder,program,call_key,selector,project_root,session) VALUES (?1,?2,'',?3,?4,?5)", params![folder,program,serde_json::to_string(&selector(stamp)).expect("selector serializes"),identity.project_root.to_string_lossy(),identity.session]) + conn.execute("INSERT OR IGNORE INTO macos_permission_cards(folder,program,call_key,selector,project_root,session) VALUES (?1,?2,'',?3,?4,?5)", params![read_folder,program,serde_json::to_string(&selector(&read_stamp)).expect("selector serializes"),read_identity.project_root.to_string_lossy(),read_identity.session]) .map_err(|e| ConsentError::Unavailable(e.to_string()))?; } - conn.query_row("SELECT selector,project_root,session,elicitation_id FROM macos_permission_cards WHERE folder=?1 AND program=?2", params![folder,program], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).optional() - .map_err(|e| ConsentError::Unavailable(e.to_string()))? - }; + conn.query_row("SELECT selector,project_root,session,elicitation_id FROM macos_permission_cards WHERE folder=?1 AND program=?2", params![read_folder,program], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).optional() + .map_err(|e| ConsentError::Unavailable(e.to_string())) + }) + .await + .map_err(|e| ConsentError::Unavailable(e.to_string()))??; let Some((scope, root, session, id)) = row else { continue; }; @@ -3718,15 +3736,28 @@ pub(super) async fn macos_permissions( // produce a permission answer. Remove the unpublished // local row and preserve the named refusal for logging, // without changing the tool's macos_permission_denied result. - db.lock().map_err(|e| ConsentError::Unavailable(e.to_string()))?.execute("DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2 AND elicitation_id IS NULL", params![folder,program]) - .map_err(|e| ConsentError::Unavailable(e.to_string()))?; + let delete_db = Arc::clone(&db); + let delete_folder = folder.clone(); + tokio::task::spawn_blocking(move || { + delete_db.lock().map_err(|e| ConsentError::Unavailable(e.to_string()))?.execute("DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2 AND elicitation_id IS NULL", params![delete_folder,program]) + .map_err(|e| ConsentError::Unavailable(e.to_string())) + }) + .await + .map_err(|e| ConsentError::Unavailable(e.to_string()))??; return Err(error); } result => result?, }; let id = decode_elicitation_id(&result)?; - db.lock().map_err(|e| ConsentError::Unavailable(e.to_string()))?.execute("UPDATE macos_permission_cards SET elicitation_id=?3 WHERE folder=?1 AND program=?2", params![folder,program,id]) - .map_err(|e| ConsentError::Unavailable(e.to_string()))?; + let update_db = Arc::clone(&db); + let update_folder = folder.clone(); + let update_id = id.clone(); + tokio::task::spawn_blocking(move || { + update_db.lock().map_err(|e| ConsentError::Unavailable(e.to_string()))?.execute("UPDATE macos_permission_cards SET elicitation_id=?3 WHERE folder=?1 AND program=?2", params![update_folder,program,update_id]) + .map_err(|e| ConsentError::Unavailable(e.to_string())) + }) + .await + .map_err(|e| ConsentError::Unavailable(e.to_string()))??; id } }; @@ -3746,13 +3777,19 @@ pub(super) async fn macos_permissions( .map_err(|_| { ConsentError::Unavailable("macOS permission card withdrawal timed out".into()) })??; - db.lock() - .map_err(|e| ConsentError::Unavailable(e.to_string()))? - .execute( - "DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2", - params![folder, program], - ) - .map_err(|e| ConsentError::Unavailable(e.to_string()))?; + let delete_db = Arc::clone(&db); + tokio::task::spawn_blocking(move || { + delete_db + .lock() + .map_err(|e| ConsentError::Unavailable(e.to_string()))? + .execute( + "DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2", + params![folder, program], + ) + .map_err(|e| ConsentError::Unavailable(e.to_string())) + }) + .await + .map_err(|e| ConsentError::Unavailable(e.to_string()))??; } } Ok(()) diff --git a/crates/aft/src/subc/consent_tests.rs b/crates/aft/src/subc/consent_tests.rs index 9fdbb72c2b..b4a345abc3 100644 --- a/crates/aft/src/subc/consent_tests.rs +++ b/crates/aft/src/subc/consent_tests.rs @@ -6296,7 +6296,237 @@ async fn macos_permission_cards_deduplicate_and_withdraw_on_readable_probe() { ); } -#[cfg(target_os = "macos")] +// macOS permission cards exist only on Unix hosts, and the fake Cingulate checks +// folders as POSIX absolute paths, so these card tests do not run on Windows. +#[cfg(unix)] +#[tokio::test(flavor = "current_thread")] +async fn macos_permissions_db_contention_does_not_block_runtime_polling() { + #[derive(Clone, Copy, Debug)] + enum Stage { + Read, + RefusalDelete, + Update, + WithdrawalDelete, + } + + struct LockBeforeWrite { + stage: Stage, + fake: Arc, + acquire: std::sync::mpsc::Sender<()>, + acquired: StdMutex>>, + waiting: StdMutex>>, + } + + impl ConsentTransport for LockBeforeWrite { + fn call(&self, call: ConsentCall) -> ConsentFuture { + let expected = if matches!(self.stage, Stage::WithdrawalDelete) { + "consent.withdraw" + } else { + "consent.request" + }; + assert_eq!(call.method, expected); + let response = if matches!(self.stage, Stage::RefusalDelete) { + Err(fake_refusal("access_macos_only", "macOS is required")) + } else { + self.fake + .reply(call.method, &call.params) + .map(Option::unwrap) + }; + self.acquire.send(()).unwrap(); + let acquired = self.acquired.lock().unwrap().take().unwrap(); + let waiting = self.waiting.lock().unwrap().take().unwrap(); + Box::pin(async move { + acquired.await.unwrap(); + waiting.send(()).unwrap(); + response + }) + } + } + + let _lock = CONSENT_TEST_LOCK.lock().await; + for stage in [ + Stage::Read, + Stage::RefusalDelete, + Stage::Update, + Stage::WithdrawalDelete, + ] { + let f = fixture(FakeCingulate::up()); + let db = f.ctx.db().unwrap(); + let folder = f.dir.path().join("protected"); + if matches!(stage, Stage::WithdrawalDelete) { + macos_permissions( + Arc::clone(&db), + f.head.clone(), + vec![crate::bounded_io::PermissionObservation { + folder: folder.clone(), + readable: false, + }], + ) + .await + .unwrap(); + } + let (acquire, requested) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + let (locked, acquired) = tokio::sync::oneshot::channel(); + let holder_db = Arc::clone(&db); + let holder = std::thread::spawn(move || { + requested.recv_timeout(Duration::from_secs(2)).unwrap(); + let _guard = holder_db.lock().unwrap(); + locked.send(()).unwrap(); + // Bound a runtime-blocking regression even when its timer cannot run. + released.recv_timeout(Duration::from_secs(2)).is_ok() + }); + let waiting = if matches!(stage, Stage::Read) { + acquire.send(()).unwrap(); + acquired.await.unwrap(); + None + } else { + let (waiting, started) = tokio::sync::oneshot::channel(); + install_test_transport(Some(Arc::new(LockBeforeWrite { + stage, + fake: Arc::clone(&f.fake), + acquire, + acquired: StdMutex::new(Some(acquired)), + waiting: StdMutex::new(Some(waiting)), + }))); + Some(started) + }; + let start = Instant::now(); + let mut permission = tokio::spawn(macos_permissions( + Arc::clone(&db), + f.head.clone(), + vec![crate::bounded_io::PermissionObservation { + folder, + readable: matches!(stage, Stage::WithdrawalDelete), + }], + )); + if let Some(waiting) = waiting { + tokio::time::timeout(Duration::from_secs(3), waiting) + .await + .expect("permission transport reached the database write") + .unwrap(); + } + tokio::select! { + _ = tokio::time::sleep(Duration::from_millis(50)) => {}, + result = &mut permission => panic!("permission finished while db was held ({stage:?}): {result:?}"), + } + let _ = release.send(()); + assert!( + holder.join().unwrap(), + "runtime did not release the lock holder in time ({stage:?})" + ); + assert!( + start.elapsed() < Duration::from_secs(1), + "runtime stopped polling while waiting for the database ({stage:?})" + ); + let result = permission.await.unwrap(); + if matches!(stage, Stage::RefusalDelete) { + assert!( + matches!(result, Err(ConsentError::Refused { code, message }) if code == "access_macos_only" && message == "macOS is required") + ); + } else { + result.unwrap(); + } + let ids = db + .lock() + .unwrap() + .prepare("SELECT elicitation_id FROM macos_permission_cards") + .unwrap() + .query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::, _>>() + .unwrap(); + if matches!(stage, Stage::Read | Stage::Update) { + assert_eq!(ids, vec!["acc_1"]); + assert_eq!(f.fake.calls("consent.request").len(), 1); + assert!(f.fake.calls("consent.withdraw").is_empty()); + } else { + assert!(ids.is_empty()); + if matches!(stage, Stage::WithdrawalDelete) { + assert_eq!(f.fake.calls("consent.request").len(), 1); + assert_eq!(f.fake.calls("consent.withdraw").len(), 1); + } + } + assert!(f.fake.calls("consent.await").is_empty()); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn macos_permission_startup_db_contention_does_not_block_runtime_polling() { + struct LockBeforeDelete { + acquire: std::sync::mpsc::Sender<()>, + acquired: StdMutex>>, + } + impl ConsentTransport for LockBeforeDelete { + fn call(&self, call: ConsentCall) -> ConsentFuture { + assert_eq!(call.method, "consent.withdraw"); + self.acquire.send(()).unwrap(); + let acquired = self.acquired.lock().unwrap().take().unwrap(); + Box::pin(async move { + acquired.await.unwrap(); + Ok(json!({})) + }) + } + } + + let _lock = CONSENT_TEST_LOCK.lock().await; + for before_delete in [false, true] { + let f = fixture(FakeCingulate::up()); + let db = f.ctx.db().unwrap(); + db.lock().unwrap().execute( + "INSERT INTO macos_permission_cards(folder,program,call_key,selector,project_root,session,elicitation_id) + VALUES ('folder','program','key',?1,'/root','session','id')", + [serde_json::to_string(&selector(f.head.scope.as_ref().unwrap())).unwrap()], + ).unwrap(); + let (acquire, requested) = std::sync::mpsc::channel(); + let (release, released) = std::sync::mpsc::channel(); + let (locked, acquired) = tokio::sync::oneshot::channel(); + let holder_db = Arc::clone(&db); + let holder = std::thread::spawn(move || { + requested.recv().unwrap(); + let _guard = holder_db.lock().unwrap(); + locked.send(()).unwrap(); + // A broken cleanup must fail rather than deadlock the test runtime. + released.recv_timeout(Duration::from_secs(2)).is_ok() + }); + if before_delete { + install_test_transport(Some(Arc::new(LockBeforeDelete { + acquire, + acquired: StdMutex::new(Some(acquired)), + }))); + } else { + acquire.send(()).unwrap(); + acquired.await.unwrap(); + } + let mut cleanup = tokio::spawn(withdraw_macos_permission_cards(Arc::clone(&db))); + let start = std::time::Instant::now(); + let kept_polling = tokio::select! { + _ = tokio::time::sleep(Duration::from_millis(50)) => true, + result = &mut cleanup => { panic!("cleanup finished while db was held: {result:?}"); } + }; + release.send(()).unwrap(); + assert!( + holder.join().unwrap(), + "runtime did not release the lock holder in time" + ); + assert!( + kept_polling && start.elapsed() < Duration::from_secs(1), + "runtime stopped polling while waiting for the database (delete={before_delete})" + ); + assert_eq!(cleanup.await.unwrap().unwrap(), 1); + assert_eq!( + db.lock() + .unwrap() + .query_row("SELECT count(*) FROM macos_permission_cards", [], |row| row + .get::<_, usize>(0)) + .unwrap(), + 0 + ); + install_test_transport(None); + } +} + +#[cfg(unix)] struct StartupMacosWithdrawal { fake: Arc, failed_id: String, @@ -6305,7 +6535,7 @@ struct StartupMacosWithdrawal { calls: StdMutex>, } -#[cfg(target_os = "macos")] +#[cfg(unix)] impl ConsentTransport for StartupMacosWithdrawal { fn call(&self, call: ConsentCall) -> ConsentFuture { assert_eq!(call.method, "consent.withdraw"); @@ -6332,7 +6562,7 @@ impl ConsentTransport for StartupMacosWithdrawal { } } -#[cfg(target_os = "macos")] +#[cfg(unix)] #[tokio::test] async fn macos_permission_startup_withdraws_ids_deletes_unpublished_and_keeps_failed_rows() { let _lock = CONSENT_TEST_LOCK.lock().await; @@ -6399,7 +6629,7 @@ async fn macos_permission_startup_withdraws_ids_deletes_unpublished_and_keeps_fa /// A card filed by a session that has since ended can't be withdrawn under its /// scope; Cingulate settles it, so startup drops the record instead of /// retrying it at every start. -#[cfg(target_os = "macos")] +#[cfg(unix)] #[tokio::test] async fn macos_permission_startup_drops_cards_whose_scope_ended() { let _lock = CONSENT_TEST_LOCK.lock().await; diff --git a/crates/aft/src/subc/held_task_recovery.rs b/crates/aft/src/subc/held_task_recovery.rs index 304535adab..e48beacc86 100644 --- a/crates/aft/src/subc/held_task_recovery.rs +++ b/crates/aft/src/subc/held_task_recovery.rs @@ -11,34 +11,69 @@ struct Session { task_ids: Vec, } -fn sessions(db: &crate::db::TrackedConnection) -> rusqlite::Result> { - let mut stmt = db.prepare( - "SELECT t.harness,t.session_id, - COALESCE(json_extract(t.metadata,'$.project_root'),t.cwd), - t.task_id,t.metadata, - EXISTS(SELECT 1 FROM call_ledger c - WHERE c.task_id=t.task_id AND c.recovered=1 AND c.state='DispatchStarted' - AND json_extract(t.metadata,'$.call_key.requester')=c.carrier - AND json_extract(t.metadata,'$.call_key.key')=c.call_key) AS keyed - FROM bash_tasks t - WHERE keyed OR t.status NOT IN ('completed','failed','killed','timed_out','fate_unknown') - OR t.completion_delivered=0 - ORDER BY 1,2,3,t.task_id", - )?; - let rows = stmt.query_map([], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - row.get::<_, String>(4)?, - row.get::<_, bool>(5)?, - )) - })?; - let mut pending = std::collections::BTreeMap::<_, Vec>::new(); - for row in rows { - let (harness, session, root, task, metadata, keyed) = row?; +// Each page is an index seek, not an OFFSET scan. The existing replay index +// includes legacy metadata/column disagreements, which are filtered after reading. +const RECOVERY_CHUNK: usize = 64; +const RUNNER_PAGE: &str = "SELECT harness,session_id,cwd,task_id,metadata, + project_key,started_at,rowid,status,completion_delivered + FROM bash_tasks INDEXED BY idx_bash_tasks_project_replay + WHERE (status NOT IN ('completed','failed','killed','timed_out','fate_unknown') + OR completion_delivered=0 + OR CASE WHEN json_valid(metadata) + THEN json_type(metadata,'$.completion_delivered')='false' ELSE 0 END) + AND (harness,project_key,started_at,task_id,rowid)>(?1,?2,?3,?4,?5) + ORDER BY harness,project_key,started_at,task_id,rowid LIMIT ?6"; + +#[cfg(test)] +thread_local! { + // The offline benchmark starts cleanup only after recovery owns the mutex. + static FIRST_RECOVERY_LOCK: std::cell::RefCell>> = const { std::cell::RefCell::new(None) }; +} + +struct RecoveryReader<'a> { + db: &'a Arc>, + #[cfg(test)] + on_chunk: &'a mut dyn FnMut(usize, Duration), +} + +impl RecoveryReader<'_> { + fn read( + &mut self, + query: impl FnOnce(&crate::db::TrackedConnection) -> rusqlite::Result>, + ) -> rusqlite::Result> { + let conn = self.db.lock().map_err(|_| rusqlite::Error::InvalidQuery)?; + #[cfg(test)] + let started = Instant::now(); + #[cfg(test)] + FIRST_RECOVERY_LOCK.with(|probe| { + if let Some(observer) = probe.borrow_mut().take() { + observer(); + } + }); + let rows = query(&conn)?; + #[cfg(test)] + let held = started.elapsed(); + drop(conn); + #[cfg(test)] + (self.on_chunk)(rows.len(), held); + Ok(rows) + } +} + +fn sessions( + db: &Arc>, + #[cfg(test)] on_chunk: &mut dyn FnMut(usize, Duration), +) -> rusqlite::Result> { + use rusqlite::params; + let mut reader = RecoveryReader { + db, + #[cfg(test)] + on_chunk, + }; + let mut pending = std::collections::BTreeMap::<_, std::collections::BTreeSet>::new(); + let mut include = |harness: String, session, cwd: String, task, metadata: String, keyed| { let metadata: Value = serde_json::from_str(&metadata).unwrap_or(Value::Null); + let root = metadata["project_root"].as_str().unwrap_or(&cwd).to_owned(); let runner = [ metadata["launching_identity"]["principal"].as_str(), metadata["call_key"]["requester"].as_str(), @@ -61,7 +96,133 @@ fn sessions(db: &crate::db::TrackedConnection) -> rusqlite::Result> pending .entry((harness, session, root)) .or_default() - .push(task); + .insert(task); + } + }; + let mut cursor = ( + String::new(), + String::new(), + i64::MIN, + String::new(), + i64::MIN, + ); + loop { + let rows = reader.read(|conn| { + conn.prepare(RUNNER_PAGE)? + .query_map( + params![ + cursor.0, + cursor.1, + cursor.2, + cursor.3, + cursor.4, + RECOVERY_CHUNK + ], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, String>(5)?, + row.get::<_, i64>(6)?, + row.get::<_, i64>(7)?, + row.get::<_, String>(8)?, + row.get::<_, bool>(9)?, + )) + }, + )? + .collect::>>() + })?; + if rows.is_empty() { + break; + } + for (harness, session, cwd, task, metadata, project, started, rowid, status, delivered) in + rows + { + cursor = (harness.clone(), project, started, task.clone(), rowid); + if !delivered + || !matches!( + status.as_str(), + "completed" | "failed" | "killed" | "timed_out" | "fate_unknown" + ) + { + include(harness, session, cwd, task, metadata, false); + } + } + } + // Start with the small DispatchStarted range of the retention index. Walk + // NULL and non-NULL settled_at separately so every page has an index bound. + for null_settled in [true, false] { + let mut cursor = (i64::MIN, i64::MIN); + loop { + let keys = reader.read(|conn| { + let sql = if null_settled { + "SELECT settled_at,rowid,carrier,call_key,task_id,recovered FROM call_ledger + WHERE state='DispatchStarted' AND settled_at IS NULL AND rowid>?2 + ORDER BY settled_at,rowid LIMIT ?3" + } else { + "SELECT settled_at,rowid,carrier,call_key,task_id,recovered FROM call_ledger + WHERE state='DispatchStarted' AND (settled_at,rowid)>(?1,?2) + ORDER BY settled_at,rowid LIMIT ?3" + }; + let mut stmt = conn.prepare(sql)?; + let rows = stmt + .query_map(params![cursor.0, cursor.1, RECOVERY_CHUNK], |row| { + Ok(( + row.get::<_, Option>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, Option>(4)?, + row.get::<_, bool>(5)?, + )) + })? + .collect::>>(); + rows + })?; + if keys.is_empty() { + break; + } + for (settled, rowid, carrier, key, task, recovered) in keys { + cursor = (settled.unwrap_or(i64::MIN), rowid); + let Some(task) = task.filter(|_| recovered) else { + continue; + }; + let mut owner = (String::new(), String::new()); + loop { + let rows = reader.read(|conn| { + conn.prepare( + "SELECT harness,session_id,cwd,task_id,metadata FROM bash_tasks + WHERE task_id=?1 AND (harness,session_id)>(?2,?3) + ORDER BY harness,session_id LIMIT ?4", + )? + .query_map(params![task, owner.0, owner.1, RECOVERY_CHUNK], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + )) + })? + .collect::>>() + })?; + if rows.is_empty() { + break; + } + for (harness, session, cwd, task, metadata) in rows { + owner = (harness.clone(), session.clone()); + let parsed: Value = serde_json::from_str(&metadata).unwrap_or(Value::Null); + if parsed["call_key"]["requester"].as_str() == Some(carrier.as_str()) + && parsed["call_key"]["key"].as_str() == Some(key.as_str()) + { + include(harness, session, cwd, task, metadata, true); + } + } + } + } } } Ok(pending @@ -70,7 +231,7 @@ fn sessions(db: &crate::db::TrackedConnection) -> rusqlite::Result> harness, session_id, project_root: root.into(), - task_ids, + task_ids: task_ids.into_iter().collect(), }) .collect()) } @@ -104,10 +265,12 @@ fn recover( .open_db(&storage.join("aft.db")) .map_err(|error| error.to_string())?; crate::db::call_ledger::recover_pool_once(&db).map_err(|error| error.to_string())?; - let pending = { - let conn = db.lock().map_err(|error| error.to_string())?; - sessions(&conn).map_err(|error| error.to_string())? - }; + let pending = sessions( + &db, + #[cfg(test)] + &mut |_, _| {}, + ) + .map_err(|error| error.to_string())?; let mut replayed = std::collections::BTreeMap::>::new(); let mut missing_roots = std::collections::BTreeMap::::new(); for session in pending { @@ -257,6 +420,10 @@ fn replay( } } +#[cfg(test)] +#[path = "held_task_recovery_benchmark.rs"] +mod benchmark; + #[cfg(test)] pub(crate) mod tests { use super::*; @@ -579,6 +746,104 @@ pub(crate) mod tests { received.await.unwrap(); } + #[test] + fn recovery_selection_uses_bounded_index_pages_and_releases_shared_mutex() { + let dir = tempfile::tempdir().unwrap(); + let ctx = context(dir.path(), &dir.path().join("storage")); + let db = ctx.db().unwrap(); + { + let conn = db.lock().unwrap(); + conn.execute_batch(r#"BEGIN; + WITH RECURSIVE n(i) AS (VALUES(0) UNION ALL SELECT i+1 FROM n WHERE i<4095) + INSERT INTO bash_tasks(harness,session_id,task_id,project_key,command,cwd,status, + started_at,completion_delivered,metadata) + SELECT 'runner','history',printf('old-%05d',i),'root','done','/root','completed',0,1,'{}' FROM n; + WITH RECURSIVE n(i) AS (VALUES(0) UNION ALL SELECT i+1 FROM n WHERE i<129) + INSERT INTO bash_tasks(harness,session_id,task_id,project_key,command,cwd,status, + started_at,completion_delivered,metadata) + SELECT 'runner',printf('active-%03d',i),'same-task','root','running','/root','running',0,1,'{}' FROM n; + INSERT INTO bash_tasks(harness,session_id,task_id,project_key,command,cwd,status, + started_at,completion_delivered,metadata) VALUES + ('runner','undelivered','undelivered','root','','/root','completed',0,0,'{}'), + ('opencode','keyed','keyed','root','','/root','completed',0,1, + '{"project_root":"/project","call_key":{"requester":"direct","key":"held"}}'), + ('opencode','wrong-owner','keyed','root','','/root','completed',0,1,'{}'), + ('runner','legacy-disagreement','legacy','root','','/root','completed',0,1,'{"completion_delivered":false}'), + ('opencode','not-runner','not-runner','root','','/root','running',0,0,'{}'), + ('opencode','not-recovered','not-recovered','root','','/root','completed',0,1, + '{"call_key":{"requester":"direct","key":"not-recovered"}}'), + ('opencode','settled','settled','root','','/root','completed',0,1, + '{"call_key":{"requester":"direct","key":"settled"}}'); + INSERT INTO call_ledger(carrier,call_key,digest,state,event_id,task_id,recovered) VALUES + ('direct','held','','DispatchStarted','','keyed',1), + ('direct','not-recovered','','DispatchStarted','','not-recovered',0), + ('direct','settled','','Settled','','settled',1); + WITH RECURSIVE n(i) AS (VALUES(0) UNION ALL SELECT i+1 FROM n WHERE i<129) + INSERT INTO call_ledger(carrier,call_key,digest,state,event_id,settled_at,recovered) + SELECT 'direct',printf('unlinked-%03d',i),'','DispatchStarted','', + CASE WHEN i<65 THEN NULL ELSE 1 END,1 FROM n; + COMMIT;"#).unwrap(); + let plan = conn + .prepare(&format!("EXPLAIN QUERY PLAN {RUNNER_PAGE}")) + .unwrap() + .query_map( + rusqlite::params!["", "", i64::MIN, "", i64::MIN, RECOVERY_CHUNK], + |row| row.get::<_, String>(3), + ) + .unwrap() + .collect::>>() + .unwrap() + .join("\n"); + assert!( + plan.contains("SEARCH bash_tasks USING INDEX idx_bash_tasks_project_replay"), + "{plan}" + ); + assert!(!plan.contains("TEMP B-TREE"), "{plan}"); + } + let mut chunks = Vec::new(); + let selected = sessions(&db, &mut |rows, held| { + assert!( + db.try_lock().is_ok(), + "chunk observer must run after unlocking" + ); + assert!(rows <= RECOVERY_CHUNK, "read {rows} rows in one lock hold"); + chunks.push((rows, held)); + }) + .unwrap(); + let mut expected = (0..130) + .map(|i| Session { + harness: "runner".into(), + session_id: format!("active-{i:03}"), + project_root: "/root".into(), + task_ids: vec!["same-task".into()], + }) + .collect::>(); + expected.insert( + 0, + Session { + harness: "opencode".into(), + session_id: "keyed".into(), + project_root: "/project".into(), + task_ids: vec!["keyed".into()], + }, + ); + expected.push(Session { + harness: "runner".into(), + session_id: "undelivered".into(), + project_root: "/root".into(), + task_ids: vec!["undelivered".into()], + }); + assert_eq!(selected, expected); + assert!( + chunks + .iter() + .filter(|(rows, _)| *rows == RECOVERY_CHUNK) + .count() + >= 4, + "both candidate sources must cross page boundaries: {chunks:?}" + ); + } + #[test] fn runner_saved_selection_includes_carriers_and_parsed_harnesses_only() { let dir = tempfile::tempdir().unwrap(); @@ -630,7 +895,8 @@ pub(crate) mod tests { ) .unwrap(); } - let selected = sessions(&conn).unwrap(); + drop(conn); + let selected = sessions(&db, &mut |_, _| {}).unwrap(); assert_eq!( selected .iter() diff --git a/crates/aft/src/subc/held_task_recovery_benchmark.rs b/crates/aft/src/subc/held_task_recovery_benchmark.rs new file mode 100644 index 0000000000..690bdb71d2 --- /dev/null +++ b/crates/aft/src/subc/held_task_recovery_benchmark.rs @@ -0,0 +1,275 @@ +//! Opt-in measurements on an offline snapshot. Commands are selected from the +//! database, never restored into a running registry or attached to processes. + +use super::*; +use crate::subc::consent; + +// The selection function from the pre-pagination implementation is retained here +// so one binary can compare both lock scopes against the same offline database. +fn baseline_sessions(db: &crate::db::TrackedConnection) -> rusqlite::Result> { + let mut stmt = db.prepare( + "SELECT t.harness,t.session_id, + COALESCE(json_extract(t.metadata,'$.project_root'),t.cwd), + t.task_id,t.metadata, + EXISTS(SELECT 1 FROM call_ledger c + WHERE c.task_id=t.task_id AND c.recovered=1 AND c.state='DispatchStarted' + AND json_extract(t.metadata,'$.call_key.requester')=c.carrier + AND json_extract(t.metadata,'$.call_key.key')=c.call_key) AS keyed + FROM bash_tasks t + WHERE keyed OR t.status NOT IN ('completed','failed','killed','timed_out','fate_unknown') + OR t.completion_delivered=0 + ORDER BY 1,2,3,t.task_id", + )?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, bool>(5)?, + )) + })?; + let mut pending = std::collections::BTreeMap::<_, Vec>::new(); + for row in rows { + let (harness, session, root, task, metadata, keyed) = row?; + let metadata: Value = serde_json::from_str(&metadata).unwrap_or(Value::Null); + let runner = [ + metadata["launching_identity"]["principal"].as_str(), + metadata["call_key"]["requester"].as_str(), + ] + .into_iter() + .any(|carrier| { + matches!( + carrier, + Some("reserved:thalamus-harness" | "reserved:broca") + ) + }) || [ + Some(harness.as_str()), + metadata["launching_raw_harness"].as_str(), + metadata["launching_identity"]["harness"].as_str(), + ] + .into_iter() + .flatten() + .any(|raw| raw.parse::() == Ok(crate::harness::Harness::Runner)); + if keyed || runner { + pending + .entry((harness, session, root)) + .or_default() + .push(task); + } + } + Ok(pending + .into_iter() + .map(|((harness, session_id, root), task_ids)| Session { + harness, + session_id, + project_root: root.into(), + task_ids, + }) + .collect()) +} + +#[derive(Clone, Copy, Debug)] +struct Measurement { + total: Duration, + max_hold: Duration, + holds: usize, + tick_delay: Duration, +} + +struct NoCingulate; +impl consent::ConsentTransport for NoCingulate { + fn call(&self, _: consent::ConsentCall) -> consent::ConsentFuture { + panic!("the offline benchmark must never call Cingulate") + } +} + +// Only the synchronous SELECT/DELETE sections are modelled for the old cleanup. +// The fixed case below calls the production async cleanup with an unpublished +// card, which exercises both database sections without making a transport call. +fn baseline_cleanup(db: &Arc>) { + let rows = { + let conn = db.lock().unwrap(); + let mut stmt = conn.prepare("SELECT folder,program,selector,project_root,session,elicitation_id FROM macos_permission_cards").unwrap(); + let rows = stmt + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, Option>(5)?, + )) + }) + .unwrap() + .collect::>>() + .unwrap(); + rows + }; + for (folder, program, _, _, _, id) in rows { + assert!(id.is_none()); + db.lock() + .unwrap() + .execute( + "DELETE FROM macos_permission_cards WHERE folder=?1 AND program=?2", + rusqlite::params![folder, program], + ) + .unwrap(); + } +} + +fn measure( + db: &Arc>, + fixed: bool, +) -> (Vec, Measurement) { + let mut max_hold = Duration::ZERO; + let mut holds = 0; + let mut observe = |_: usize, held: Duration| { + holds += 1; + max_hold = max_hold.max(held); + }; + // The shared wrapper measures from acquisition until immediately before + // unlock, including query iteration; time waiting to acquire is excluded. + let begin = Instant::now(); + RecoveryReader { + db, + on_chunk: &mut observe, + } + .read(|conn| { + crate::db::call_ledger::recover(conn)?; + Ok(Vec::<()>::new()) + }) + .unwrap(); + db.lock().unwrap().execute("INSERT INTO macos_permission_cards(folder,program,call_key,selector,project_root,session,elicitation_id) VALUES ('offline-benchmark','program','','unused','unused','unused',NULL)", []).unwrap(); + let (start_tx, start_rx) = tokio::sync::oneshot::channel(); + let (finish_tx, finish_rx) = tokio::sync::oneshot::channel(); + let runtime_db = Arc::clone(db); + let runtime = std::thread::spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async move { + start_rx.await.unwrap(); + let mut next = Instant::now() + Duration::from_millis(10); + let cleanup = tokio::spawn(async move { + if fixed { + consent::withdraw_macos_permission_cards(runtime_db) + .await + .unwrap(); + } else { + baseline_cleanup(&runtime_db); + } + }); + let mut finish = finish_rx; + let mut worst = Duration::ZERO; + loop { + tokio::select! { + _ = tokio::time::sleep_until(next.into()) => { + let now = Instant::now(); + worst = worst.max(now.saturating_duration_since(next)); + next = now + Duration::from_millis(10); + } + _ = &mut finish => { + worst = worst.max(Instant::now().saturating_duration_since(next)); + cleanup.await.unwrap(); + return worst.max(Instant::now().saturating_duration_since(next)); + } + } + } + }) + }); + FIRST_RECOVERY_LOCK.with(|probe| { + *probe.borrow_mut() = Some(Box::new(move || { + let _ = start_tx.send(()); + })) + }); + let selected = if fixed { + sessions(db, &mut observe).unwrap() + } else { + RecoveryReader { + db, + on_chunk: &mut observe, + } + .read(baseline_sessions) + .unwrap() + }; + let total = begin.elapsed(); + finish_tx.send(()).unwrap(); + let tick_delay = runtime.join().unwrap(); + ( + selected, + Measurement { + total, + max_hold, + holds, + tick_delay, + }, + ) +} + +#[test] +#[ignore = "requires AFT_RECOVERY_BENCH_DB naming an offline snapshot copy under target/scratch"] +fn startup_recovery_snapshot_benchmark() { + let Some(path) = std::env::var_os("AFT_RECOVERY_BENCH_DB") else { + return; + }; + // Read at run time so the compiled test does not embed this checkout's path. + let manifest = std::env::var_os("CARGO_MANIFEST_DIR").expect("cargo sets CARGO_MANIFEST_DIR"); + let scratch = PathBuf::from(manifest) + .join("../../target/scratch") + .canonicalize() + .unwrap(); + let path = PathBuf::from(path).canonicalize().unwrap(); + assert!( + path.starts_with(&scratch), + "only scratch snapshot copies are allowed" + ); + assert!(std::env::var_os("AFT_STORAGE_DIR").is_none()); + let db = Arc::new(StdMutex::new(crate::db::open(&path).unwrap())); + assert_eq!( + db.lock() + .unwrap() + .query_row("SELECT count(*) FROM macos_permission_cards", [], |row| row + .get::<_, usize>(0)) + .unwrap(), + 0, + "snapshot must not contain published cards" + ); + consent::install_test_transport(Some(Arc::new(NoCingulate))); + let expected = baseline_sessions(&db.lock().unwrap()).unwrap(); + println!( + "warmup selected {} sessions / {} tasks", + expected.len(), + expected.iter().map(|s| s.task_ids.len()).sum::() + ); + for fixed in [false, true] { + let mut runs = Vec::new(); + for run in 1..=3 { + let (selected, measured) = measure(&db, fixed); + assert_eq!(selected, expected); + println!( + "{} run={run} total_ms={:.3} max_hold_ms={:.3} holds={} worst_tick_delay_ms={:.3}", + if fixed { "fixed" } else { "base" }, + measured.total.as_secs_f64() * 1000.0, + measured.max_hold.as_secs_f64() * 1000.0, + measured.holds, + measured.tick_delay.as_secs_f64() * 1000.0 + ); + runs.push(measured); + } + let mut totals = runs.iter().map(|r| r.total).collect::>(); + totals.sort(); + let mut holds = runs.iter().map(|r| r.max_hold).collect::>(); + holds.sort(); + let mut delays = runs.iter().map(|r| r.tick_delay).collect::>(); + delays.sort(); + println!("{} median/worst: total_ms={:.3}/{:.3} max_hold_ms={:.3}/{:.3} tick_delay_ms={:.3}/{:.3}", + if fixed { "fixed" } else { "base" }, totals[1].as_secs_f64()*1000.0, totals[2].as_secs_f64()*1000.0, + holds[1].as_secs_f64()*1000.0, holds[2].as_secs_f64()*1000.0, + delays[1].as_secs_f64()*1000.0, delays[2].as_secs_f64()*1000.0); + } + consent::install_test_transport(None); +} diff --git a/crates/aft/src/subc/mod.rs b/crates/aft/src/subc/mod.rs index eeff255474..0312fbd0a2 100644 --- a/crates/aft/src/subc/mod.rs +++ b/crates/aft/src/subc/mod.rs @@ -5468,17 +5468,24 @@ where // route loop, never before it: each withdrawal waits on Cingulate, and a // slow or absent Cingulate must not delay route binds. #[cfg(target_os = "macos")] - match shared_app.open_db(&storage_dir.join("aft.db")) { - Ok(db) => { - tokio::spawn(async move { - if let Err(error) = consent::withdraw_macos_permission_cards(db).await { - log::warn!("macOS permission card startup cleanup failed: {error:?}"); + { + let app = Arc::clone(&shared_app); + let path = storage_dir.join("aft.db"); + tokio::spawn(async move { + match tokio::task::spawn_blocking(move || app.open_db(&path)).await { + Ok(Ok(db)) => { + if let Err(error) = consent::withdraw_macos_permission_cards(db).await { + log::warn!("macOS permission card startup cleanup failed: {error:?}"); + } } - }); - } - Err(error) => { - log::warn!("macOS permission card startup cleanup could not open database: {error}") - } + Ok(Err(error)) => log::warn!( + "macOS permission card startup cleanup could not open database: {error}" + ), + Err(error) => { + log::warn!("macOS permission card startup cleanup worker failed: {error}") + } + } + }); } rich_ui::display::start(connection_file_path.to_path_buf()); runner_notices::install(connection_file_path.to_path_buf()); diff --git a/crates/aft/src/subc/shells/runtime.rs b/crates/aft/src/subc/shells/runtime.rs index fd094bee8b..22a704cbe5 100644 --- a/crates/aft/src/subc/shells/runtime.rs +++ b/crates/aft/src/subc/shells/runtime.rs @@ -315,13 +315,16 @@ impl Service { drop(state); let unknown_agent_ids = if let Some(agents) = params.agent_ids { let db = self.app.db(); - let known = db - .and_then(|db| { + let known = tokio::task::spawn_blocking(move || { + db.and_then(|db| { db.lock() .ok() .and_then(|conn| crate::db::access_grant_rows::known_agent_ids(&conn).ok()) }) - .unwrap_or_default(); + .unwrap_or_default() + }) + .await + .unwrap_or_default(); let unknown: Vec<_> = agents .into_iter() .filter(|id| !known.contains(id)) diff --git a/crates/aft/src/subc/shells/runtime_tests.rs b/crates/aft/src/subc/shells/runtime_tests.rs index f5cb4ae4f5..15ba5bc074 100644 --- a/crates/aft/src/subc/shells/runtime_tests.rs +++ b/crates/aft/src/subc/shells/runtime_tests.rs @@ -1156,6 +1156,57 @@ async fn shells_v1_output_pty_screen_uses_masked_renderer_and_cursor() { service.stop(); } +#[tokio::test(flavor = "current_thread")] +async fn shells_list_agent_db_contention_does_not_block_runtime_polling() { + let (_dir, ctx, executor, app) = status_context(); + let db = ctx.db().unwrap(); + { + let conn = db.lock().unwrap(); + let key = ScopeKey::from_stamp(&stamp("head", false)); + crate::db::access_grants::register_scope(&conn, &key, None, &Default::default()).unwrap(); + crate::db::access_grants::register_agent(&conn, &key, "known-agent", true).unwrap(); + } + let service = Service::new(app, executor); + let (release, released) = std::sync::mpsc::channel(); + let (locked, acquired) = tokio::sync::oneshot::channel(); + let holder_db = Arc::clone(&db); + let holder = std::thread::spawn(move || { + let _guard = holder_db.lock().unwrap(); + locked.send(()).unwrap(); + // The external deadline still runs if the current-thread runtime blocks. + released.recv_timeout(Duration::from_secs(2)).is_ok() + }); + acquired.await.unwrap(); + let list_service = Arc::clone(&service); + let mut list = tokio::spawn(async move { + list_service + .list(ListParams { + agent_ids: Some(vec!["unknown-agent".into(), "known-agent".into()]), + ..Default::default() + }) + .await + }); + let start = std::time::Instant::now(); + tokio::select! { + _ = tokio::time::sleep(Duration::from_millis(50)) => {}, + result = &mut list => panic!("list finished while db was held: {result:?}"), + } + let _ = release.send(()); + assert!( + holder.join().unwrap(), + "runtime did not release the lock holder in time" + ); + assert!( + start.elapsed() < Duration::from_secs(1), + "runtime stopped polling while waiting for the database" + ); + let reply = list.await.unwrap().unwrap(); + assert!(reply.rows.is_empty()); + assert!(reply.next_cursor.is_none()); + assert_eq!(reply.unknown_agent_ids, Some(vec!["unknown-agent".into()])); + service.stop(); +} + #[cfg(target_os = "linux")] #[tokio::test] async fn shells_v1_list_pages_fifty_running_first_and_cursor_is_filter_bound() { diff --git a/crates/aft/tests/integration/lsp_diagnostics_test.rs b/crates/aft/tests/integration/lsp_diagnostics_test.rs index 4e461096b0..48cac836fe 100644 --- a/crates/aft/tests/integration/lsp_diagnostics_test.rs +++ b/crates/aft/tests/integration/lsp_diagnostics_test.rs @@ -4371,9 +4371,31 @@ fn main() { "{mode}: check artifacts at {check_out}, expected {}", analyzer_target.display() ); + // Cargo's intermediate directory layout is not stable. Require this + // crate's compiler artifact under the isolated target, not a `deps` folder. + let mut directories = vec![analyzer_target.clone()]; + let mut found_compiler_artifact = false; + while let Some(directory) = directories.pop() { + for entry in fs::read_dir(directory).expect("read analyzer artifacts") { + let entry = entry.unwrap(); + let kind = entry.file_type().unwrap(); + if kind.is_dir() { + directories.push(entry.path()); + } else if kind.is_file() + && entry + .file_name() + .to_string_lossy() + .starts_with("liblock_isolation") + && entry.path().extension().is_some_and(|ext| ext == "rmeta") + { + found_compiler_artifact = true; + } + } + } assert!( - analyzer_target.join("debug/deps").is_dir(), - "check artifacts missing" + found_compiler_artifact, + "{mode}: check compiler artifacts missing under {}", + analyzer_target.display() ); eprintln!( "{mode}: analyzer artifacts under {}; cargo build stderr:\n{stderr}",