diff --git a/crates/tracedecay-application/src/diagnostics_producer.rs b/crates/tracedecay-application/src/diagnostics_producer.rs index c42994382f..82c87e939d 100644 --- a/crates/tracedecay-application/src/diagnostics_producer.rs +++ b/crates/tracedecay-application/src/diagnostics_producer.rs @@ -23,14 +23,13 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, RwLock}; use futures_util::FutureExt; -use futures_util::future::join_all; +use futures_util::StreamExt; use sha2::{Digest, Sha256}; -use tokio::sync::Semaphore; use tracedecay_domain::{CodeGenerationId, ContentDigest, UtcMicros}; use tracedecay_lsp::{ - SearchedTsconfig, TypeScriptFileOwner, TypeScriptProject, TypeScriptProjectConfigInputs, - run_typescript_compiler, typescript_build_info_program, typescript_file_owner, - typescript_install_command, typescript_projects, + SearchedTsconfig, TypeScriptCompilerAdmission, TypeScriptFileOwner, TypeScriptProject, + TypeScriptProjectConfigInputs, run_typescript_compiler, typescript_build_info_program, + typescript_file_owner, typescript_install_command, typescript_projects, }; use tracedecay_runtime_core::native_runtime; @@ -136,6 +135,7 @@ impl TypeScriptProducerStatusHandleV1 { /// digests of the generation it last checked and how each project fared. pub struct TypeScriptProducerStateV1 { status: TypeScriptProducerStatusHandleV1, + admission: TypeScriptCompilerAdmission, checked_files: Option>, projects: BTreeMap, } @@ -144,9 +144,13 @@ impl TypeScriptProducerStateV1 { /// State for one project owner's producer, publishing its status into /// `status`. #[must_use] - pub fn new(status: TypeScriptProducerStatusHandleV1) -> Self { + pub fn new( + status: TypeScriptProducerStatusHandleV1, + admission: TypeScriptCompilerAdmission, + ) -> Self { Self { status, + admission, checked_files: None, projects: BTreeMap::new(), } @@ -166,10 +170,6 @@ enum ProjectResult { }, } -/// Projects checked at once. Each check is one mostly single-threaded tsc -/// process, and a large project holds over a gigabyte while it runs. -const CONCURRENT_CHECKS: usize = 4; - /// Publishes the merged findings of each project's own TypeScript compiler as /// the compiler pillar's clean-generation snapshot for `identity`. /// @@ -261,7 +261,7 @@ async fn produce( } let reused = results.len(); let checked = stale.len(); - results.extend(check_projects(project_root, build_info_dir, stale).await); + results.extend(check_projects(project_root, build_info_dir, stale, &state.admission).await); state.checked_files = Some(files); let mut parsed = Vec::new(); @@ -375,12 +375,13 @@ fn needs_check( }) } -/// Checks `stale` projects, at most [`CONCURRENT_CHECKS`] at a time, each -/// against its own build-info in `build_info_dir`. +/// The lazy window bounds waiting futures; the shared admission accounts for +/// children until supervision has killed and reaped them. async fn check_projects( project_root: &Path, build_info_dir: &Path, stale: Vec<(PathBuf, PathBuf)>, + admission: &TypeScriptCompilerAdmission, ) -> BTreeMap { if stale.is_empty() { return BTreeMap::new(); @@ -406,22 +407,20 @@ async fn check_projects( }) .collect(); } - let permits = Arc::new(Semaphore::new(CONCURRENT_CHECKS)); - let checks = stale - .into_iter() + futures_util::stream::iter(stale) .map(|(tsconfig, compiler)| { let build_info = build_info_dir.join(build_info_file_name(&tsconfig)); let root = project_root.to_path_buf(); - let permits = Arc::clone(&permits); async move { let check = async { - let Ok(_permit) = permits.acquire().await else { - return ProjectResult::Failed { - reason: "the TypeScript check admission closed".to_owned(), - }; - }; - match run_typescript_compiler(&compiler, &root, &tsconfig, Some(&build_info)) - .await + match run_typescript_compiler( + admission, + &compiler, + &root, + &tsconfig, + Some(&build_info), + ) + .await { Ok(reported) => ProjectResult::Checked { reported: reported.into_iter().map(compiler_diagnostic).collect(), @@ -441,8 +440,9 @@ async fn check_projects( (tsconfig, result) } }) - .collect::>(); - join_all(checks).await.into_iter().collect() + .buffer_unordered(admission.concurrency()) + .collect() + .await } /// One build-info file per tsconfig path, so linked worktrees sharing a store @@ -557,13 +557,17 @@ mod tests { #[cfg(unix)] use std::path::PathBuf; + #[cfg(unix)] + use std::sync::Arc; #[cfg(unix)] use tracedecay_domain::UtcMicros; #[cfg(unix)] use tracedecay_domain::test_fixtures::id; #[cfg(unix)] - use tracedecay_lsp::typescript_projects; + use tracedecay_lsp::{TypeScriptCompilerAdmission, typescript_projects}; + #[cfg(unix)] + use tracedecay_runtime_core::background_cpu::ProcessBackgroundCpuV1; #[cfg(unix)] use tracedecay_runtime_core::test_executable::write_executable_script; @@ -580,6 +584,13 @@ mod tests { #[cfg(unix)] use crate::diagnostics_store::DiagnosticsStore; + #[cfg(unix)] + fn admission() -> TypeScriptCompilerAdmission { + TypeScriptCompilerAdmission::new(Arc::new(ProcessBackgroundCpuV1::new( + std::num::NonZeroUsize::new(4).unwrap(), + ))) + } + #[test] fn tsc_report_lines_become_compiler_pillar_diagnostics() { let diagnostic = compiler_diagnostic(tracedecay_lsp::Diagnostic { @@ -832,7 +843,10 @@ mod tests { #[tokio::test] async fn only_projects_a_generation_changed_are_rechecked() { let fixture = ProducerFixture::new(); - let mut state = TypeScriptProducerStateV1::new(TypeScriptProducerStatusHandleV1::default()); + let mut state = TypeScriptProducerStateV1::new( + TypeScriptProducerStatusHandleV1::default(), + admission(), + ); let both = vec![ ( "file.packages.app.src.index.ts".to_owned(), @@ -891,7 +905,7 @@ mod tests { async fn a_successor_owner_does_not_inherit_the_predecessors_status() { let fixture = ProducerFixture::new(); let predecessor = TypeScriptProducerStatusHandleV1::default(); - let mut state = TypeScriptProducerStateV1::new(predecessor.clone()); + let mut state = TypeScriptProducerStateV1::new(predecessor.clone(), admission()); let run = fixture.run("generation.producer.1", &mut state).await; assert_eq!( predecessor.current().map(|status| status.run().clone()), @@ -914,4 +928,75 @@ mod tests { }; assert_eq!(check, Some(TypeScriptProjectCheckV1::Checked)); } + + #[cfg(unix)] + #[tokio::test] + async fn isolated_producers_share_compiler_admission_and_keep_partial_findings() { + let fixtures = [ + ProducerFixture::new(), + ProducerFixture::new(), + ProducerFixture::new(), + ]; + let counter = tempfile::tempdir().unwrap(); + for fixture in &fixtures { + let compiler = fixture.root.join("node_modules/.bin/tsc"); + let script = std::fs::read_to_string(&compiler).unwrap(); + let script = script.replacen( + "#!/bin/sh\n", + &format!( + "#!/bin/sh\n\ + d='{}'\n\ + mkdir \"$d/active.$$\"\n\ + ls -d \"$d\"/active.* | wc -l >> \"$d/seen\"\n\ + sleep 0.3\n\ + rmdir \"$d/active.$$\"\n", + counter.path().display(), + ), + 1, + ); + write_executable_script(&compiler, script).unwrap(); + } + let admission = admission(); + let mut states = fixtures + .iter() + .map(|_| { + TypeScriptProducerStateV1::new( + TypeScriptProducerStatusHandleV1::default(), + admission.clone(), + ) + }) + .collect::>(); + let runs = futures_util::future::join_all( + fixtures + .iter() + .zip(states.iter_mut()) + .map(|(fixture, state)| fixture.run("generation.shared", state)), + ) + .await; + assert!( + runs.iter() + .all(|run| run == &published("generation.shared", 3, 0)), + "{runs:?}" + ); + for fixture in &fixtures { + assert_eq!(fixture.checked(), ["app", "broken", "lib"]); + assert_eq!(fixture.findings("generation.shared").await.len(), 2); + } + let counts = std::fs::read_to_string(counter.path().join("seen")).unwrap(); + let counts = counts + .lines() + .map(|line| line.trim().parse::().unwrap()) + .collect::>(); + assert_eq!(counts.len(), 9); + assert!( + counts.iter().all(|count| *count <= admission.concurrency()), + "{counts:?}" + ); + assert!(counts.iter().any(|count| *count > 1), "{counts:?}"); + assert_eq!( + std::fs::read_dir(counter.path()).unwrap().count(), + 1, + "every child removed its active marker", + ); + } } diff --git a/crates/tracedecay-automation-runtime/src/automation/effect_runtime/settlement/tests.rs b/crates/tracedecay-automation-runtime/src/automation/effect_runtime/settlement/tests.rs index dbaf6f7d6a..93dcb4552a 100644 --- a/crates/tracedecay-automation-runtime/src/automation/effect_runtime/settlement/tests.rs +++ b/crates/tracedecay-automation-runtime/src/automation/effect_runtime/settlement/tests.rs @@ -389,19 +389,10 @@ async fn retained_disabled_user_job_run( .await } -fn now_secs() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("wall clock") - .as_secs() as i64 -} - async fn task_lock_is_denied(dashboard_root: &std::path::Path, job_id: &str) -> bool { crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed( dashboard_root, &format!("user_job_{job_id}"), - None, - now_secs(), ) .await .expect("competing task lock") @@ -2237,8 +2228,6 @@ async fn retained_projector_panic_finishes_recovery_before_releasing_task_lock() if crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed( dashboard_root, &format!("user_job_{job_id}"), - None, - now_secs(), ) .await .expect("post-projector-panic task lock") @@ -2581,8 +2570,6 @@ async fn retained_pair_attempts_second_leg_and_keeps_both_guards_until_both_fini let reacquired = crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed( dashboard_root, &format!("user_job_{job_id}"), - None, - now_secs(), ) .await .expect("post-pair lock probe"); @@ -3010,8 +2997,6 @@ async fn retained_user_job_rebinds_and_recovery_retires_only_terminal_corrupt_sp crate::automation::scheduler::AutomationTaskLock::try_acquire_keyed( dashboard_root, &format!("user_job_{job_id}"), - None, - now_secs(), ) .await .expect("post-terminal lock") diff --git a/crates/tracedecay-automation-runtime/src/automation/jobs.rs b/crates/tracedecay-automation-runtime/src/automation/jobs.rs index f93d40e115..fe01cb1fe6 100644 --- a/crates/tracedecay-automation-runtime/src/automation/jobs.rs +++ b/crates/tracedecay-automation-runtime/src/automation/jobs.rs @@ -44,7 +44,6 @@ const JOB_COMMAND_TIMEOUT_SECS: u64 = 30; const JOB_COMMAND_OUTPUT_CAP_CHARS: usize = 16 * 1024; const JOB_SKILL_BODY_CAP_CHARS: usize = 4_000; const DEFAULT_JOB_FAILURE_COOLDOWN_SECS: u64 = 300; -const DEFAULT_JOB_STALE_LOCK_SECS: u64 = 6 * 60 * 60; const WEBHOOK_TIMEOUT_SECS: u64 = 10; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -179,15 +178,8 @@ fn job_lock_key(job_id: &str) -> String { pub(super) async fn try_acquire_job_task_lock( dashboard_root: &Path, job_id: &str, - now_secs: i64, ) -> Result> { - AutomationTaskLock::try_acquire_keyed( - dashboard_root, - &job_lock_key(job_id), - Some(DEFAULT_JOB_STALE_LOCK_SECS), - now_secs, - ) - .await + AutomationTaskLock::try_acquire_keyed(dashboard_root, &job_lock_key(job_id)).await } #[tracing::instrument(name = "automation.jobs.load", level = "trace", skip_all)] @@ -546,9 +538,7 @@ async fn run_retained_user_job( return ctx.skipped(reason).await.map_err(Into::into); } - let now_secs = current_timestamp(); - let Some(task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id, now_secs).await? - else { + let Some(task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id).await? else { let reason = if trigger == AutomationTrigger::Scheduler { AutomationSkipReasonV1::SchedulerLockActive } else { diff --git a/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_gate.rs b/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_gate.rs index 3df7f41ab6..c5ff84d17d 100644 --- a/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_gate.rs +++ b/crates/tracedecay-automation-runtime/src/automation/jobs/scheduler_gate.rs @@ -57,9 +57,7 @@ pub async fn evaluate_and_record_scheduler_skip( .map(Some); } - let lock_time = current_timestamp(); - let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id, lock_time).await? - else { + let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id).await? else { crate::automation::scheduler_trace::observe_skip_reason( AutomationSkipReasonV1::SchedulerLockActive, ); @@ -123,8 +121,7 @@ pub(super) async fn evaluate_and_record_scheduler_skip_at( .map(Some); } - let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id, now_secs).await? - else { + let Some(_task_lock) = try_acquire_job_task_lock(dashboard_root, &job.id).await? else { return record_scheduler_lock_skip( dashboard_root, config, diff --git a/crates/tracedecay-automation-runtime/src/automation/lifecycle.rs b/crates/tracedecay-automation-runtime/src/automation/lifecycle.rs index 16a8fc4e7b..657decd45b 100644 --- a/crates/tracedecay-automation-runtime/src/automation/lifecycle.rs +++ b/crates/tracedecay-automation-runtime/src/automation/lifecycle.rs @@ -26,8 +26,7 @@ use super::run_ledger::{ AutomationTrigger, append_run_record, load_run_ledger_task_summary, }; use super::scheduler::{ - AutomationTaskLock, load_session_activity, schedule_decision, stale_lock_secs, - task_disabled_skip, + AutomationTaskLock, load_session_activity, schedule_decision, task_disabled_skip, }; use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_global_db::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1}; @@ -529,16 +528,7 @@ fn scheduler_gate_with_lock_retention<'a>( trigger, AutomationTrigger::Scheduler | AutomationTrigger::HostReceipt ); - - let lock_now_secs = current_timestamp(); - let Some(lock) = AutomationTaskLock::try_acquire( - dashboard_root, - task, - stale_lock_secs(config, task), - lock_now_secs, - ) - .await? - else { + let Some(lock) = AutomationTaskLock::try_acquire(dashboard_root, task).await? else { super::scheduler_trace::observe_skip_reason( AutomationSkipReasonV1::SchedulerLockActive, ); @@ -1434,15 +1424,11 @@ mod recorded_failure_tests { let root = tempfile::tempdir().expect("temporary dashboard root"); let settlement_guard = AutomationRunSettlementGuard::new(); let retention = settlement_guard.retention(); - let task_lock = AutomationTaskLock::try_acquire( - root.path(), - AgentTaskKind::SessionReflector, - Some(3_600), - current_timestamp(), - ) - .await - .expect("first lock acquisition") - .expect("first lock"); + let task_lock = + AutomationTaskLock::try_acquire(root.path(), AgentTaskKind::SessionReflector) + .await + .expect("first lock acquisition") + .expect("first lock"); assert!( retain_task_lock(task_lock, Some(&retention)) .expect("retain lock") @@ -1459,28 +1445,34 @@ mod recorded_failure_tests { assert!(matches!(result, Err(AutomationRunError::Runtime(_)))); assert!( - AutomationTaskLock::try_acquire( - root.path(), - AgentTaskKind::SessionReflector, - Some(3_600), - current_timestamp(), - ) - .await - .expect("competing lock acquisition") - .is_none() + AutomationTaskLock::try_acquire(root.path(), AgentTaskKind::SessionReflector,) + .await + .expect("competing lock acquisition") + .is_none() ); + let old_coordination = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open( + root.path() + .join("automation_locks/session_reflector.lock.lock"), + ) + .expect("old publisher coordination"); + assert!(matches!( + old_coordination.try_lock(), + Err(std::fs::TryLockError::WouldBlock) + )); drop(settlement_guard); + old_coordination + .try_lock() + .expect("settlement releases old publisher coordination"); + old_coordination.unlock().expect("release old publisher"); assert!( - AutomationTaskLock::try_acquire( - root.path(), - AgentTaskKind::SessionReflector, - Some(3_600), - current_timestamp(), - ) - .await - .expect("post-settlement lock acquisition") - .is_some() + AutomationTaskLock::try_acquire(root.path(), AgentTaskKind::SessionReflector,) + .await + .expect("post-settlement lock acquisition") + .is_some() ); } } diff --git a/crates/tracedecay-automation-runtime/src/automation/scheduler.rs b/crates/tracedecay-automation-runtime/src/automation/scheduler.rs index ad8f3059ee..602996d84c 100644 --- a/crates/tracedecay-automation-runtime/src/automation/scheduler.rs +++ b/crates/tracedecay-automation-runtime/src/automation/scheduler.rs @@ -1,4 +1,4 @@ -use std::io::{Read, Write}; +use std::io::Read; use std::num::NonZeroU64; use std::path::{Path, PathBuf}; @@ -7,7 +7,6 @@ use cap_std::ambient_authority; use cap_std::fs::Dir; #[cfg(unix)] use cap_std::fs::OpenOptionsExt; -use cap_std::time::SystemClock; use serde::{Deserialize, Serialize}; pub use tracedecay_automation::config::{AutomationSchedule, CronSchedule, parse_schedule}; use tracedecay_automation::evidence_budget::{ @@ -32,10 +31,7 @@ use tracedecay_global_db::RegisteredGlobalDb; use tracedecay_private_fs::FileLease; const DEFAULT_FAILURE_COOLDOWN_SECS: u64 = 300; -const DEFAULT_STALE_LOCK_SECS: u64 = 6 * 60 * 60; -const AUTOMATION_TASK_LOCK_TOKEN_BYTES: usize = 32; -const MAX_AUTOMATION_TASK_LOCK_BYTES: u64 = 1024; -const AUTOMATION_TASK_LOCK_CLEANUP_ATTEMPTS: u32 = 3; +const MAX_PRE_LEASE_TASK_LOCK_BYTES: u64 = 1024; const SCHEDULER_CONTROL_FILENAME: &str = "automation_scheduler_control.json"; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] @@ -162,41 +158,13 @@ pub(crate) fn task_disabled_skip(task: AgentTaskKind) -> AutomationSkipReasonV1 } } +/// Exclusive admission for one automation task key. The task owns this value +/// for its whole lifetime, including retained effect settlement; dropping it +/// releases the cross-process lease. #[derive(Debug)] pub struct AutomationTaskLock { - path: PathBuf, - ownership_token: String, - staging_path: Option, -} - -#[derive(Debug)] -enum TaskLockPublicationError { - Definite(std::io::Error), - CommitUncertain { - error: std::io::Error, - cleanup_owner: AutomationTaskLock, - }, -} - -enum TaskLockStagingCreationError { - Definite(std::io::Error), - #[cfg(windows)] - CommitUncertain { - error: std::io::Error, - staging_path: PathBuf, - }, -} - -impl From for TaskLockPublicationError { - fn from(error: std::io::Error) -> Self { - Self::Definite(error) - } -} - -impl From for TaskLockStagingCreationError { - fn from(error: std::io::Error) -> Self { - Self::Definite(error) - } + _lease: FileLease, + _publication_coordination: FileLease, } pub fn scheduler_control_path(dashboard_root: &Path) -> PathBuf { @@ -264,36 +232,25 @@ fn save_scheduler_control_blocking(dashboard_root: &Path, bytes: &[u8]) -> std:: } impl AutomationTaskLock { - pub async fn try_acquire( - dashboard_root: &Path, - task: AgentTaskKind, - stale_after_secs: Option, - now_secs: i64, - ) -> Result> { - Self::try_acquire_keyed(dashboard_root, task_key(task), stale_after_secs, now_secs).await + pub async fn try_acquire(dashboard_root: &Path, task: AgentTaskKind) -> Result> { + Self::try_acquire_keyed(dashboard_root, task_key(task)).await } /// Acquires a lock under an arbitrary key. User-defined jobs lock per /// job (`user_job_`) so concurrent jobs never serialize on the shared /// fixed-task lock name. #[tracing::instrument(name = "automation.scheduler.task_lock", level = "trace", skip_all)] - pub async fn try_acquire_keyed( - dashboard_root: &Path, - key: &str, - stale_after_secs: Option, - now_secs: i64, - ) -> Result> { + pub async fn try_acquire_keyed(dashboard_root: &Path, key: &str) -> Result> { // cap-std ambient opens walk each component. On macOS `/var` is a // firmlink to `/private/var`; opening the unresolved tempfile spelling // can ENOENT under concurrent create. Resolve the existing prefix first. let lock_dir = tracedecay_runtime_core::path_safety::canonicalize_path_or_existing_parent( &dashboard_root.join("automation_locks"), ); - let path = lock_dir.join(format!("{key}.lock")); - let ownership_token = new_automation_task_lock_token()?; - let error_path = path.clone(); + let key = key.to_owned(); + let error_dir = lock_dir.clone(); tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - try_acquire_task_lock_blocking(&path, &ownership_token, stale_after_secs, now_secs) + try_acquire_task_lock_blocking(&lock_dir, &key) }) .await .map_err(|error| TraceDecayError::Config { @@ -301,91 +258,13 @@ impl AutomationTaskLock { })? .map_err(|error| TraceDecayError::Config { message: format!( - "failed to acquire automation lock '{}': {error}", - error_path.display() + "failed to acquire automation lock under '{}': {error}", + error_dir.display() ), }) } } -/// Releases the lock file synchronously, without ever calling into the tokio -/// runtime. -/// -/// Task-lock release must be synchronous: callers (and tests such as -/// `retained_settlement_guard_owns_task_lock_until_drop`) rely on the lock file -/// being gone the instant `drop` returns. The cleanup itself is genuinely -/// blocking, a file coordination lock, `sync_all`/parent-directory fsyncs, and -/// `std::thread::sleep` backoff between retries, so it is tempting to hand the -/// owning worker's run queue away with `tokio::task::block_in_place`. -/// -/// That is not sound here, because this guard is reachable from inside tokio's -/// own blocking-pool spawn path. When a runtime has begun shutting down, -/// `blocking::pool::Spawner::spawn_task` (tokio 1.53.1) shuts a refused task -/// down *while holding* the pool's non-reentrant `parking_lot` mutex, which -/// drops the task's future, and with it any `AutomationTaskLock` the future -/// owned, such as the `Arc` captured by -/// `start_retained_automation_settlement_inner`. `block_in_place` re-enters -/// `spawn_task` on that same thread to hand off the worker core, so the release -/// self-deadlocks on the mutex it is already under; the runtime's -/// `BlockingPool::shutdown` then waits for that thread forever. -/// -/// Blocking a worker for one unlink plus its fsyncs is the cost of being -/// correct in a destructor. Callers that care about worker latency release the -/// guard from `spawn_blocking` instead of dropping it on a worker. -impl Drop for AutomationTaskLock { - fn drop(&mut self) { - let path = &self.path; - let ownership_token = &self.ownership_token; - let staging_slot = &mut self.staging_path; - match retry_exact_task_lock_cleanup(|| { - remove_owned_task_lock_blocking(path, ownership_token) - }) { - Ok(()) => { - if let Some(staging_path) = staging_slot.take() - && let Err(error) = retry_exact_task_lock_cleanup(|| { - tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable( - &staging_path, - ) - .map(|_| ()) - }) - { - tracing::warn!( - path = %path.display(), - staging_path = %staging_path.display(), - error = %error, - "failed to retire exact automation task-lock staging ownership" - ); - } - } - Err(error) => { - tracing::warn!( - path = %path.display(), - staging_path = ?staging_slot.as_deref(), - error = %error, - "failed to release exact automation task-lock ownership; preserving retained staging evidence" - ); - } - } - } -} - -fn retry_exact_task_lock_cleanup(mut operation: Operation) -> std::io::Result<()> -where - Operation: FnMut() -> std::io::Result<()>, -{ - let mut attempt = 1_u32; - loop { - match operation() { - Ok(()) => return Ok(()), - Err(_) if attempt < AUTOMATION_TASK_LOCK_CLEANUP_ATTEMPTS => { - std::thread::sleep(std::time::Duration::from_millis(5_u64 << (attempt - 1))); - attempt += 1; - } - Err(error) => return Err(error), - } - } -} - #[tracing::instrument(name = "automation.scheduler.decision", level = "trace", skip_all)] /// Decides whether `task` is due under `config`, given the ledger `records` /// and the executable the backend now in force would spawn @@ -801,12 +680,6 @@ pub fn cron_is_due(cron: &CronSchedule, last_completed_at: Option, now_secs } } -pub fn stale_lock_secs(config: &AutomationConfig, task: AgentTaskKind) -> Option { - task_config(config, task) - .and_then(|task_config| task_config.stale_lock_secs) - .or(Some(DEFAULT_STALE_LOCK_SECS)) -} - /// User jobs carry their own schedule/enabled state (see /// `automation::jobs`), so the fixed-task config lookup falls back to a /// disabled default that makes the fixed-task gates skip them. @@ -870,414 +743,49 @@ pub(crate) fn elapsed_secs(completed_at: i64, now_secs: i64) -> u64 { (now_secs - completed_at) as u64 } -fn new_automation_task_lock_token() -> Result { - let mut random = [0_u8; AUTOMATION_TASK_LOCK_TOKEN_BYTES]; - getrandom::fill(&mut random).map_err(|error| TraceDecayError::Config { - message: format!("failed to generate automation task-lock ownership token: {error}"), - })?; - Ok(hex::encode(random)) -} - -fn resolve_task_lock_parent(path: &Path) -> std::io::Result { - let (parent, name) = path.parent().zip(path.file_name()).ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "automation task-lock path requires a parent and file name", - ) - })?; - // Resolve OS directory aliases without hiding a final symlink from no-follow checks. - Ok( - tracedecay_runtime_core::path_safety::canonicalize_path_or_existing_parent(parent) - .join(name), - ) -} - +/// Keeps both stable coordination files locked through task settlement. Older +/// daemons lock `.lock.lock` only around owner publication and cleanup: +/// checking their owner record under that lock closes the pre-publication +/// race, and retaining it prevents an old daemon publishing during this task. +/// Neither coordination inode may be unlinked, even when retiring a dead owner. fn try_acquire_task_lock_blocking( - path: &Path, - ownership_token: &str, - stale_after_secs: Option, - now_secs: i64, + lock_dir: &Path, + key: &str, ) -> std::io::Result> { - let path = resolve_task_lock_parent(path)?; - if let Some(parent) = path.parent() { - tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(parent)?; - } - let coordination = acquire_task_lock_coordination(&path)?; - for attempt in 0..2 { - match create_task_lock_file(&path, ownership_token, now_secs) { - Ok(task_lock) => return Ok(Some(task_lock)), - Err(TaskLockPublicationError::Definite(error)) - if error.kind() == std::io::ErrorKind::AlreadyExists => - { - let Some(snapshot) = read_task_lock_snapshot(&path)? else { - continue; - }; - if attempt == 0 && task_lock_is_reclaimable(&snapshot, stale_after_secs, now_secs) { - tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(&path)?; - continue; - } - return Ok(None); - } - Err(TaskLockPublicationError::Definite(error)) => return Err(error), - Err(TaskLockPublicationError::CommitUncertain { - error, - cleanup_owner, - }) => { - drop(coordination); - drop(cleanup_owner); - return Err(error); - } - } - } - Ok(None) -} - -fn create_task_lock_file( - path: &Path, - ownership_token: &str, - now_secs: i64, -) -> std::result::Result { - let (task_lock, staging_path) = publish_task_lock_file(path, ownership_token, now_secs)?; - let cleanup_result = - tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(&staging_path) - .map(|_| ()); - let sync_result = tracedecay_private_fs::framed_log::sync_parent_directory( - path, - tracedecay_private_fs::framed_log::DirectorySyncPolicy::Strict, - ); - Ok(task_lock.settle_published_staging(&staging_path, cleanup_result, sync_result)) -} - -#[cfg(test)] -fn create_task_lock_file_with_post_publish( - path: &Path, - ownership_token: &str, - now_secs: i64, - cleanup_published_staging: Cleanup, - sync_published_parent: SyncParent, -) -> std::result::Result -where - Cleanup: FnOnce(&Path) -> std::io::Result<()>, - SyncParent: FnOnce(&Path) -> std::io::Result<()>, -{ - let (task_lock, staging_path) = publish_task_lock_file(path, ownership_token, now_secs)?; - let cleanup_result = cleanup_published_staging(&staging_path); - let sync_result = sync_published_parent(path); - Ok(task_lock.settle_published_staging(&staging_path, cleanup_result, sync_result)) -} - -fn publish_task_lock_file( - path: &Path, - ownership_token: &str, - now_secs: i64, -) -> std::result::Result<(AutomationTaskLock, PathBuf), TaskLockPublicationError> { - let prepared = - prepare_task_lock_publication(path, ownership_token, now_secs).map_err(|failure| { - match failure { - TaskLockStagingCreationError::Definite(error) => { - TaskLockPublicationError::Definite(error) - } - #[cfg(windows)] - TaskLockStagingCreationError::CommitUncertain { - error, - staging_path, - } => { - let cleanup_error = cleanup_unpublished_task_lock_staging( - &staging_path, - "automation task-lock staging creation failed", - error, - ); - TaskLockPublicationError::Definite(cleanup_error) - } - } - })?; - let link_result = prepared.parent.hard_link( - &prepared.staging_name, - &prepared.parent, - &prepared.destination_name, - ); - settle_task_lock_link_result(path, ownership_token, prepared.staging_path, link_result) -} - -#[cfg(test)] -fn publish_task_lock_file_after_committed_link_error( - path: &Path, - ownership_token: &str, - now_secs: i64, -) -> std::result::Result<(AutomationTaskLock, PathBuf), TaskLockPublicationError> { - let prepared = - prepare_task_lock_publication(path, ownership_token, now_secs).map_err(|failure| { - match failure { - TaskLockStagingCreationError::Definite(error) => { - TaskLockPublicationError::Definite(error) - } - #[cfg(windows)] - TaskLockStagingCreationError::CommitUncertain { - error, - staging_path, - } => TaskLockPublicationError::Definite(cleanup_unpublished_task_lock_staging( - &staging_path, - "automation task-lock staging creation failed", - error, - )), - } - })?; - prepared - .parent - .hard_link( - &prepared.staging_name, - &prepared.parent, - &prepared.destination_name, - ) - .map_err(TaskLockPublicationError::Definite)?; - settle_task_lock_link_result( - path, - ownership_token, - prepared.staging_path, - Err(std::io::Error::other( - "injected commit-uncertain hard-link result", - )), - ) -} - -struct PreparedTaskLockPublication { - parent: Dir, - destination_name: std::ffi::OsString, - staging_name: std::ffi::OsString, - staging_path: PathBuf, -} - -fn prepare_task_lock_publication( - path: &Path, - ownership_token: &str, - now_secs: i64, -) -> std::result::Result { - let path = resolve_task_lock_parent(path)?; - tracedecay_runtime_core::storage::reject_symlink_components(&path, "automation task lock")?; - let parent_path = path.parent().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "automation task-lock path has no parent", - ) - })?; - let destination_name = path - .file_name() - .map(std::ffi::OsStr::to_os_string) - .ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "automation task-lock path has no file name", - ) - })?; - let parent = Dir::open_ambient_dir(parent_path, ambient_authority())?; - let staging_path = task_lock_staging_path(&path, ownership_token)?; - let staging_name = staging_path - .file_name() - .map(std::ffi::OsStr::to_os_string) - .ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "automation task-lock staging path has no file name", - ) - })?; - tracedecay_runtime_core::storage::reject_symlink_components( - &staging_path, - "automation task-lock staging", - )?; - #[cfg(windows)] - let mut file = - match tracedecay_runtime_core::windows_security::create_private_file(&staging_path) { - Ok(file) => file, - Err(error) => { - return if staging_path.try_exists().unwrap_or(true) { - Err(TaskLockStagingCreationError::CommitUncertain { - error, - staging_path, - }) - } else { - Err(TaskLockStagingCreationError::Definite(error)) - }; - } - }; - #[cfg(not(windows))] - let mut file = { - let mut options = cap_std::fs::OpenOptions::new(); - options - .create_new(true) - .write(true) - .follow(FollowSymlinks::No); - #[cfg(unix)] - options.mode(0o600); - parent.open_with(&staging_name, &options)? - }; - let started_at = - tracedecay_runtime_core::lifecycle_lease::process_start_time(std::process::id()); - let payload = match started_at { - Some(started_at) => format!( - "pid={}\ncreated_at={now_secs}\nstarted_at={started_at}\ntoken={ownership_token}\n", - std::process::id() - ), - None => format!( - "pid={}\ncreated_at={now_secs}\ntoken={ownership_token}\n", - std::process::id() - ), - }; - let write_result = file - .write_all(payload.as_bytes()) - .and_then(|()| file.sync_all()); - if let Err(error) = write_result { - drop(file); - return Err(TaskLockStagingCreationError::Definite( - cleanup_unpublished_task_lock_staging( - &staging_path, - "automation task-lock staging write failed", - error, - ), - )); - } - drop(file); - - Ok(PreparedTaskLockPublication { - parent, - destination_name, - staging_name, - staging_path, - }) -} - -fn settle_task_lock_link_result( - path: &Path, - ownership_token: &str, - staging_path: PathBuf, - link_result: std::io::Result<()>, -) -> std::result::Result<(AutomationTaskLock, PathBuf), TaskLockPublicationError> { - let owned_task_lock = || AutomationTaskLock { - path: path.to_path_buf(), - ownership_token: ownership_token.to_owned(), - staging_path: None, + tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(lock_dir)?; + let Some(lease) = open_task_lease(&lock_dir.join(format!("{key}.lease")))? else { + return Ok(None); }; - match link_result { - Ok(()) => Ok((owned_task_lock(), staging_path)), - Err(link_error) => match read_task_lock_snapshot(path) { - Ok(Some(snapshot)) if snapshot.ownership_token.as_deref() == Some(ownership_token) => { - tracing::warn!( - path = %path.display(), - error = %link_error, - "automation task-lock hard-link result was uncertain, but exact ownership is visible" - ); - Ok((owned_task_lock(), staging_path)) - } - Ok(_) => Err(TaskLockPublicationError::Definite( - cleanup_unpublished_task_lock_staging( - &staging_path, - "automation task-lock publication failed", - link_error, - ), - )), - Err(inspect_error) => { - let mut cleanup_owner = owned_task_lock(); - cleanup_owner.staging_path = Some(staging_path); - Err(TaskLockPublicationError::CommitUncertain { - error: std::io::Error::new( - inspect_error.kind(), - format!( - "automation task-lock hard-link result was uncertain: {link_error}; exact ownership inspection failed: {inspect_error}" - ), - ), - cleanup_owner, - }) - } - }, - } -} - -fn task_lock_staging_path(path: &Path, ownership_token: &str) -> std::io::Result { - let file_name = path.file_name().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "automation task-lock path has no file name", - ) - })?; - let mut staging_name = std::ffi::OsString::from("."); - staging_name.push(file_name); - staging_name.push(format!(".{ownership_token}.tmp")); - Ok(path.with_file_name(staging_name)) -} - -fn cleanup_unpublished_task_lock_staging( - staging_path: &Path, - context: &str, - error: std::io::Error, -) -> std::io::Error { - match tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(staging_path) { - Ok(_) => std::io::Error::new(error.kind(), format!("{context}: {error}")), - Err(cleanup_error) => std::io::Error::other(format!( - "{context}: {error}; staging cleanup failed: {cleanup_error}" - )), - } -} - -impl AutomationTaskLock { - fn settle_published_staging( - mut self, - staging_path: &Path, - cleanup_result: std::io::Result<()>, - sync_result: std::io::Result<()>, - ) -> Self { - if let Err(error) = cleanup_result { - tracing::warn!( - path = %self.path.display(), - staging_path = %staging_path.display(), - error = %error, - "automation task lock is owned, but its published staging link could not be retired" - ); - self.staging_path = Some(staging_path.to_path_buf()); - } - if let Err(error) = sync_result { - tracing::warn!( - path = %self.path.display(), - error = %error, - "automation task lock is owned, but its publication durability is uncertain" - ); - } - self - } -} - -fn remove_owned_task_lock_blocking(path: &Path, ownership_token: &str) -> std::io::Result<()> { - let _coordination = acquire_task_lock_coordination(path)?; - let Some(snapshot) = read_task_lock_snapshot(path)? else { - return Ok(()); + let lock_path = lock_dir.join(format!("{key}.lock")); + let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&lock_path); + let Some(publication_coordination) = open_task_lease(&coordination_path)? else { + return Ok(None); }; - if snapshot.ownership_token.as_deref() != Some(ownership_token) { - return Ok(()); + if !retire_pre_lease_owner(&lock_path)? { + return Ok(None); } - tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(path).map(|_| ()) + Ok(Some(AutomationTaskLock { + _lease: lease, + _publication_coordination: publication_coordination, + })) } -fn acquire_task_lock_coordination(path: &Path) -> std::io::Result { - let path = resolve_task_lock_parent(path)?; - let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&path); - tracedecay_runtime_core::storage::reject_symlink_components( - &coordination_path, - "automation task-lock coordination", - )?; +fn open_task_lease(path: &Path) -> std::io::Result> { + tracedecay_runtime_core::storage::reject_symlink_components(path, "automation task lease")?; #[cfg(windows)] - let file = tracedecay_runtime_core::windows_security::open_or_create_private_lock_file( - &coordination_path, - )?; + let file = tracedecay_runtime_core::windows_security::open_or_create_private_lock_file(path)?; #[cfg(not(windows))] let file = { - let parent_path = coordination_path.parent().ok_or_else(|| { + let parent_path = path.parent().ok_or_else(|| { std::io::Error::new( std::io::ErrorKind::InvalidInput, - "automation task-lock coordination path has no parent", + "automation task lease path has no parent", ) })?; - let file_name = coordination_path.file_name().ok_or_else(|| { + let file_name = path.file_name().ok_or_else(|| { std::io::Error::new( std::io::ErrorKind::InvalidInput, - "automation task-lock coordination path has no file name", + "automation task lease path has no file name", ) })?; let parent = Dir::open_ambient_dir(parent_path, ambient_authority())?; @@ -1289,8 +797,8 @@ fn acquire_task_lock_coordination(path: &Path) -> std::io::Result { .follow(FollowSymlinks::No); #[cfg(unix)] options.mode(0o600); - // Concurrent acquirers race the first creation of this coordination - // file; the shared helper absorbs the spurious Darwin `ENOENT`. + // Concurrent acquirers race the first creation of this file; the + // shared helper absorbs the spurious Darwin `ENOENT`. let file = tracedecay_private_fs::capability_dir::open_or_create_with( &parent, file_name, &options, )? @@ -1302,41 +810,60 @@ fn acquire_task_lock_coordination(path: &Path) -> std::io::Result { } file }; - file.lock()?; - let file = FileLease::held(file, "automation.task_lock.coordination"); - file.sync_all()?; - tracedecay_private_fs::framed_log::sync_parent_directory( - &coordination_path, - tracedecay_private_fs::framed_log::DirectorySyncPolicy::Strict, - )?; - Ok(file) + match file.try_lock() { + Ok(()) => Ok(Some(FileLease::held(file, "automation.task_lock"))), + Err(std::fs::TryLockError::WouldBlock) => Ok(None), + Err(std::fs::TryLockError::Error(error)) => Err(error), + } +} + +/// Before the lease, a task lock was a `.lock` file naming its owner's +/// pid and process start time, published beside a `.lock.lock` +/// coordination file. A live old-format holder still owns the key, so the +/// lease alone must not admit a second task. Returns `false` while such a +/// holder lives. Records of a confirmed-dead or pid-reused owner are retired; +/// ownership that cannot be verified is a typed error, never abandonment. +/// The caller must own the stable publication coordination through settlement. +fn retire_pre_lease_owner(lock_path: &Path) -> std::io::Result { + if let Some(owner) = read_pre_lease_owner(lock_path)? { + let Some(pid) = owner.pid else { + return Err(unverifiable_pre_lease_owner(lock_path)); + }; + match process_state(pid) { + ProcessState::Live if !recorded_owner_was_reused(pid, owner.started_at) => { + return Ok(false); + } + ProcessState::Unknown => return Err(unverifiable_pre_lease_owner(lock_path)), + ProcessState::Live | ProcessState::Dead => { + tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(lock_path)?; + } + } + } + Ok(true) +} + +fn unverifiable_pre_lease_owner(lock_path: &Path) -> std::io::Error { + std::io::Error::other(format!( + "pre-lease automation task lock '{}' has unverifiable ownership; stop older TraceDecay processes and delete it", + lock_path.display() + )) } -struct AutomationTaskLockSnapshot { +struct PreLeaseOwner { pid: Option, - created_at: Option, - /// Process start time recorded when the lock was published. Absent on - /// locks written before start-time fencing, and when the publisher could - /// not read its own start time. + /// Absent on locks written before start-time fencing, and when the + /// publisher could not read its own start time. started_at: Option, - ownership_token: Option, } -fn read_task_lock_snapshot(path: &Path) -> std::io::Result> { - let path = resolve_task_lock_parent(path)?; - tracedecay_runtime_core::storage::reject_symlink_components(&path, "automation task lock")?; - let parent_path = path.parent().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "automation task-lock path has no parent", - ) - })?; - let file_name = path.file_name().ok_or_else(|| { - std::io::Error::new( +fn read_pre_lease_owner(path: &Path) -> std::io::Result> { + tracedecay_runtime_core::storage::reject_symlink_components(path, "automation task lock")?; + let (Some(parent_path), Some(file_name)) = (path.parent(), path.file_name()) else { + return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, - "automation task-lock path has no file name", - ) - })?; + "automation task-lock path requires a parent and file name", + )); + }; let parent = Dir::open_ambient_dir(parent_path, ambient_authority())?; let mut options = cap_std::fs::OpenOptions::new(); options.read(true).follow(FollowSymlinks::No); @@ -1345,40 +872,18 @@ fn read_task_lock_snapshot(path: &Path) -> std::io::Result return Ok(None), Err(error) => return Err(error), }; - let metadata = file.metadata()?; - let mut bytes = Vec::with_capacity(MAX_AUTOMATION_TASK_LOCK_BYTES as usize); + let mut bytes = Vec::with_capacity(MAX_PRE_LEASE_TASK_LOCK_BYTES as usize); (&mut file) - .take(MAX_AUTOMATION_TASK_LOCK_BYTES + 1) + .take(MAX_PRE_LEASE_TASK_LOCK_BYTES + 1) .read_to_end(&mut bytes)?; - let contents = (bytes.len() as u64 <= MAX_AUTOMATION_TASK_LOCK_BYTES) + let contents = (bytes.len() as u64 <= MAX_PRE_LEASE_TASK_LOCK_BYTES) .then(|| std::str::from_utf8(&bytes).ok()) .flatten(); - let pid = contents - .and_then(|contents| parse_unique_lock_field(contents, "pid=")) - .and_then(|value| value.parse::().ok()); - let payload_created_at = contents - .and_then(|contents| parse_unique_lock_field(contents, "created_at=")) - .and_then(|value| value.parse::().ok()); - let started_at = contents - .and_then(|contents| parse_unique_lock_field(contents, "started_at=")) - .and_then(|value| value.parse::().ok()); - let ownership_token = contents - .and_then(|contents| parse_unique_lock_field(contents, "token=")) - .filter(|value| valid_automation_task_lock_token(value)) - .map(str::to_owned); - let created_at = payload_created_at.or_else(|| { - metadata - .modified() - .ok()? - .duration_since(SystemClock::UNIX_EPOCH) - .ok() - .and_then(|duration| i64::try_from(duration.as_secs()).ok()) - }); - Ok(Some(AutomationTaskLockSnapshot { - pid, - created_at, - started_at, - ownership_token, + let field = + |prefix: &str| contents.and_then(|contents| parse_unique_lock_field(contents, prefix)); + Ok(Some(PreLeaseOwner { + pid: field("pid=").and_then(|value| value.parse().ok()), + started_at: field("started_at=").and_then(|value| value.parse().ok()), })) } @@ -1390,63 +895,6 @@ fn parse_unique_lock_field<'a>(contents: &'a str, prefix: &str) -> Option<&'a st values.next().is_none().then_some(value) } -fn valid_automation_task_lock_token(token: &str) -> bool { - token.len() == AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2 - && token - .bytes() - .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) -} - -fn task_lock_is_reclaimable( - snapshot: &AutomationTaskLockSnapshot, - stale_after_secs: Option, - now_secs: i64, -) -> bool { - match snapshot.pid { - Some(pid) => match process_state(pid) { - // A confirmed-dead owner cannot still be inside the critical - // section. Age is irrelevant: a reused PID would look live, not - // dead, so waiting `stale_after_secs` only delays the retry. - ProcessState::Dead => true, - // Unknown liveness stays on the age gate. A missing age bound - // means the caller disabled that fallback. - ProcessState::Unknown => { - age_lock_is_reclaimable(snapshot.created_at, stale_after_secs, now_secs) - } - // A live PID is the owner only when its start time matches the - // one published with the lock. A mismatch is PID reuse (including - // a process this user cannot signal). A live PID with no recorded - // start time is a legacy lock and is not stolen. - ProcessState::Live => recorded_owner_was_reused(pid, snapshot.started_at), - }, - // The payload could not yield a pid (missing, oversized, non-UTF-8, - // or duplicate `pid=` lines). Fall back to age-based staleness using - // `created_at`, which `read_task_lock_snapshot` already backfills - // from the file's mtime when the payload has no parseable - // `created_at=` field. - None => age_lock_is_reclaimable(snapshot.created_at, stale_after_secs, now_secs), - } -} - -fn age_lock_is_reclaimable( - created_at: Option, - stale_after_secs: Option, - now_secs: i64, -) -> bool { - let Some(stale_after_secs) = stale_after_secs else { - return false; - }; - match created_at { - Some(created_at) => elapsed_secs(created_at, now_secs) >= stale_after_secs, - // Crash-debris escape hatch: no parseable pid AND no readable - // creation time (payload and mtime both unavailable) means the - // lock can never be aged by any other path, so treat it as - // reclaimable rather than permanently wedging the scheduler - // tick behind garbage lock contents. - None => true, - } -} - fn recorded_owner_was_reused(pid: u32, recorded_started_at: Option) -> bool { let Some(recorded_started_at) = recorded_started_at else { return false; @@ -1559,6 +1007,8 @@ fn windows_process_state(pid: u32) -> ProcessState { #[cfg(test)] mod tests { + use std::io::{BufRead, BufReader, Write}; + use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; use std::sync::atomic::{AtomicBool, Ordering}; use super::super::backend::{AgentTaskFailureClass, AgentTaskRetryAttempt}; @@ -2427,518 +1877,514 @@ evidence about it", assert!(!cron_is_due(&cron, Some(3_600), 3_600)); } + async fn acquire(root: &Path, key: &str) -> Option { + AutomationTaskLock::try_acquire_keyed(root, key) + .await + .expect("task-lock acquisition") + } + + /// Replays the shipped owner's publication/cleanup coordination in a real + /// process, with a pipe barrier before publication rather than a timing race. + #[test] + fn task_lock_process_participant() { + let Some(root) = std::env::var_os("TRACEDECAY_TEST_TASK_LOCK_ROOT") else { + return; + }; + let lock_dir = PathBuf::from(root); + let old = std::env::var("TRACEDECAY_TEST_TASK_LOCK_OWNER").unwrap() == "old"; + let lock_path = lock_dir.join("memory_curator.lock"); + let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&lock_path); + let mut coordination = None; + let mut task = None; + for command in std::io::stdin().lock().lines() { + let answer = match command.unwrap().as_str() { + "prepare" if old => { + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&coordination_path) + .unwrap(); + match file.try_lock() { + Ok(()) => { + coordination = Some(FileLease::held(file, "test.old-task")); + "held" + } + Err(std::fs::TryLockError::WouldBlock) => "busy", + Err(error) => panic!("old coordination: {error}"), + } + } + "publish" if old => { + assert!(coordination.is_some()); + assert!(!lock_path.exists()); + // The production publisher hard-linked a synced staging + // file under this lock. Keep that publication ordering. + let staging = lock_dir.join("old-owner-staging"); + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&staging) + .unwrap(); + writeln!(file, "pid={}\ncreated_at=0", std::process::id()).unwrap(); + file.sync_all().unwrap(); + std::fs::hard_link(&staging, &lock_path).unwrap(); + std::fs::remove_file(staging).unwrap(); + drop(coordination.take()); + "executing" + } + "finish" if old => { + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&coordination_path) + .unwrap(); + file.lock().unwrap(); + let guard = FileLease::held(file, "test.old-task-cleanup"); + tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable( + &lock_path, + ) + .unwrap(); + drop(guard); + "released" + } + "acquire" if !old => { + assert!(task.is_none()); + task = try_acquire_task_lock_blocking(&lock_dir, "memory_curator").unwrap(); + if task.is_some() { "held" } else { "busy" } + } + "release" if !old => { + assert!(task.take().is_some()); + "released" + } + command => panic!("unexpected task-lock process command: {command}"), + }; + println!("task-lock:{answer}"); + std::io::stdout().flush().unwrap(); + } + } + + struct TaskLockProcess { + child: Child, + input: ChildStdin, + output: BufReader, + } + + impl TaskLockProcess { + fn spawn(lock_dir: &Path, owner: &str) -> Self { + let mut child = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "automation::scheduler::tests::task_lock_process_participant", + "--nocapture", + ]) + .env("TRACEDECAY_TEST_TASK_LOCK_ROOT", lock_dir) + .env("TRACEDECAY_TEST_TASK_LOCK_OWNER", owner) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + Self { + input: child.stdin.take().unwrap(), + output: BufReader::new(child.stdout.take().unwrap()), + child, + } + } + + fn command(&mut self, command: &str) -> String { + writeln!(self.input, "{command}").unwrap(); + self.input.flush().unwrap(); + loop { + let mut line = String::new(); + assert_ne!(self.output.read_line(&mut line).unwrap(), 0, "child exited"); + if let Some((_, answer)) = line.split_once("task-lock:") { + return answer.trim().to_owned(); + } + } + } + } + + impl Drop for TaskLockProcess { + fn drop(&mut self) { + // Also reap the child if a regression assertion unwinds before + // the parent's explicit release. Never leave fixture owners alive. + let _ = self.child.kill(); + self.child.wait().unwrap(); + } + } + #[test] - fn published_task_lock_uncertainty_returns_owner_and_drop_retries_staging() { + fn old_and_new_processes_exclude_publication_and_execution_until_release() { let temp = tempdir().unwrap(); let lock_dir = temp.path().join("automation_locks"); tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(&lock_dir) .unwrap(); let lock_path = lock_dir.join("memory_curator.lock"); - let ownership_token = "a".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); - let staging_path = task_lock_staging_path(&lock_path, &ownership_token).unwrap(); - let task_lock = create_task_lock_file_with_post_publish( - &lock_path, - &ownership_token, - 100, - |_| { - Err(std::io::Error::other( - "injected staging cleanup uncertainty", - )) - }, - |_| Err(std::io::Error::other("injected parent sync uncertainty")), - ) - .unwrap(); + let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&lock_path); + let mut old = TaskLockProcess::spawn(&lock_dir, "old"); + let mut new = TaskLockProcess::spawn(&lock_dir, "new"); - assert!(lock_path.exists()); - assert!(staging_path.exists()); - let contender_token = "b".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); + assert_eq!(old.command("prepare"), "held"); + let stable_inode = std::fs::metadata(&coordination_path).unwrap(); assert!( - try_acquire_task_lock_blocking(&lock_path, &contender_token, Some(10), 200) - .unwrap() - .is_none(), - "publication uncertainty must still return a live token owner" + !lock_path.exists(), + "old holder is paused before publication" ); + let new_admission = new.command("acquire"); + assert_eq!(old.command("publish"), "executing"); + assert_eq!( + new_admission, "busy", + "old task published and executes: admitting new before publication runs both tasks" + ); + assert_eq!(new.command("acquire"), "busy", "old task still executes"); + assert_eq!(old.command("finish"), "released"); - drop(task_lock); + assert_eq!(new.command("acquire"), "held"); assert!(!lock_path.exists()); - assert!( - !staging_path.exists(), - "guard drop must retry its exact staging residue" + assert_eq!( + old.command("prepare"), + "busy", + "new task must retain old publication coordination for its whole lifetime" + ); + let mut next = TaskLockProcess::spawn(&lock_dir, "new"); + assert_eq!(next.command("acquire"), "busy"); + assert_eq!(new.command("release"), "released"); + + assert_eq!(old.command("prepare"), "held"); + assert_eq!(old.command("publish"), "executing"); + assert_eq!(next.command("acquire"), "busy"); + assert_eq!(old.command("finish"), "released"); + assert_eq!(next.command("acquire"), "held"); + assert_eq!(next.command("release"), "released"); + assert_eq!(new.command("acquire"), "held"); + drop(new); + assert_eq!( + old.command("prepare"), + "held", + "process death releases legacy coordination too" ); + assert_eq!(old.command("publish"), "executing"); + assert_eq!(next.command("acquire"), "busy"); + assert_eq!(old.command("finish"), "released"); + assert_eq!(next.command("acquire"), "held"); + assert_eq!(next.command("release"), "released"); + assert!(coordination_path.exists(), "coordination is never unlinked"); + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + assert_eq!( + std::fs::metadata(coordination_path).unwrap().ino(), + stable_inode.ino(), + "all generations must coordinate on the same inode" + ); + } + #[cfg(not(unix))] + drop(stable_inode); } - #[test] - fn exact_task_lock_cleanup_retries_before_abandoning_owner_evidence() { - let mut attempts = 0_u32; - retry_exact_task_lock_cleanup(|| { - attempts += 1; - if attempts < AUTOMATION_TASK_LOCK_CLEANUP_ATTEMPTS { - Err(std::io::Error::other("injected transient cleanup failure")) - } else { - Ok(()) - } - }) - .unwrap(); + #[tokio::test] + async fn same_key_admits_one_holder_until_release_and_other_keys_are_independent() { + let root = tempdir().unwrap(); + let first = acquire(root.path(), "memory_curator") + .await + .expect("first holder"); + assert!(acquire(root.path(), "memory_curator").await.is_none()); + let other = acquire(root.path(), "session_reflector") + .await + .expect("a different key is independent"); - assert_eq!(attempts, AUTOMATION_TASK_LOCK_CLEANUP_ATTEMPTS); + drop(first); + let next = acquire(root.path(), "memory_curator") + .await + .expect("explicit release admits the next holder"); + drop((other, next)); } - #[test] - fn committed_hard_link_error_adopts_visible_exact_owner() { - let temp = tempdir().unwrap(); - let lock_dir = temp.path().join("automation_locks"); - tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(&lock_dir) + #[tokio::test] + async fn cancellation_releases_both_task_coordination_locks() { + let root = tempdir().unwrap(); + let holder_root = root.path().to_path_buf(); + let (held_tx, held_rx) = tokio::sync::oneshot::channel(); + let holder = tokio::spawn(async move { + let _guard = acquire(&holder_root, "cancelled").await.unwrap(); + held_tx.send(()).unwrap(); + std::future::pending::<()>().await; + }); + held_rx.await.unwrap(); + assert!(acquire(root.path(), "cancelled").await.is_none()); + let coordination_path = root.path().join("automation_locks/cancelled.lock.lock"); + let old = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(coordination_path) .unwrap(); - let lock_path = lock_dir.join("session_reflector.lock"); - let ownership_token = "e".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); + assert!(matches!( + old.try_lock(), + Err(std::fs::TryLockError::WouldBlock) + )); + holder.abort(); + assert!(holder.await.unwrap_err().is_cancelled()); + old.try_lock() + .expect("cancellation releases old coordination"); + old.unlock().unwrap(); + assert!(acquire(root.path(), "cancelled").await.is_some()); + } - let (task_lock, staging_path) = - publish_task_lock_file_after_committed_link_error(&lock_path, &ownership_token, 100) - .unwrap(); - let cleanup_result = - tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(&staging_path) - .map(|_| ()); - let task_lock = task_lock.settle_published_staging(&staging_path, cleanup_result, Ok(())); + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn concurrent_same_key_admission_has_exactly_one_winner() { + let root = tempdir().unwrap(); + let attempts = (0..16).map(|_| { + let root = root.path().to_path_buf(); + tokio::spawn(async move { acquire(&root, "contended").await }) + }); + let mut winners = 0; + let mut held = Vec::new(); + for attempt in attempts { + if let Some(lock) = attempt.await.unwrap() { + winners += 1; + held.push(lock); + } + } + assert_eq!(winners, 1); + } - assert_eq!( - read_task_lock_snapshot(&lock_path) - .unwrap() - .unwrap() - .ownership_token - .as_deref(), - Some(ownership_token.as_str()), + #[tokio::test] + async fn release_keeps_the_lease_file_so_exclusion_never_splits_across_inodes() { + let root = tempdir().unwrap(); + let lease_path = root.path().join("automation_locks/memory_curator.lease"); + drop(acquire(root.path(), "memory_curator").await.unwrap()); + assert!( + lease_path.exists(), + "release must not unlink the lease file" ); - let contender_token = "f".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); + #[cfg(unix)] + let before = { + use std::os::unix::fs::MetadataExt; + std::fs::metadata(&lease_path).unwrap().ino() + }; + let held = acquire(root.path(), "memory_curator").await.unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + assert_eq!(std::fs::metadata(&lease_path).unwrap().ino(), before); + } + let foreign_open = std::fs::File::open(&lease_path).unwrap(); assert!( - try_acquire_task_lock_blocking(&lock_path, &contender_token, Some(10), 200) - .unwrap() - .is_none(), - "a commit-uncertain hard link with visible exact ownership must deny contenders" + matches!( + foreign_open.try_lock_shared(), + Err(std::fs::TryLockError::WouldBlock) + ), + "the lease is a kernel lock any other process observes" ); + drop(held); + foreign_open + .try_lock_shared() + .expect("release unlocks for other open file descriptions"); + } - drop(task_lock); - assert!(!lock_path.exists()); + #[test] + fn release_needs_no_tokio_runtime() { + let temp = tempdir().unwrap(); + let lock_dir = temp.path().join("automation_locks"); + let guard = try_acquire_task_lock_blocking(&lock_dir, "no_runtime") + .unwrap() + .expect("fresh key"); + drop(guard); + try_acquire_task_lock_blocking(&lock_dir, "no_runtime") + .unwrap() + .expect("drop released synchronously"); } #[cfg(unix)] #[test] - fn task_lock_coordination_rejects_final_symlink() { + fn task_coordination_symlinks_are_rejected_without_touching_their_targets() { let temp = tempdir().unwrap(); let lock_dir = temp.path().join("automation_locks"); tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(&lock_dir) .unwrap(); - let lock_path = lock_dir.join("memory_curator.lock"); - let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&lock_path); let outside = temp.path().join("outside-lock-target"); std::fs::write(&outside, b"unchanged").unwrap(); - std::os::unix::fs::symlink(&outside, &coordination_path).unwrap(); + for (key, suffix) in [("lease_symlink", "lease"), ("old_symlink", "lock.lock")] { + let path = lock_dir.join(format!("{key}.{suffix}")); + std::os::unix::fs::symlink(&outside, &path).unwrap(); - let error = try_acquire_task_lock_blocking( - &lock_path, - &"1".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2), - Some(10), - 200, - ) - .unwrap_err(); + let error = try_acquire_task_lock_blocking(&lock_dir, key).unwrap_err(); - assert!(!lock_path.exists()); - assert_eq!(std::fs::read(&outside).unwrap(), b"unchanged"); - assert!(coordination_path.is_symlink()); - assert_ne!(error.kind(), std::io::ErrorKind::NotFound); + assert_ne!(error.kind(), std::io::ErrorKind::NotFound); + assert_eq!(std::fs::read(&outside).unwrap(), b"unchanged"); + assert!(path.is_symlink()); + } } #[cfg(unix)] #[test] - fn task_lock_rejects_final_symlink_without_touching_target() { - let temp = tempdir().unwrap(); - let real = temp.path().join("real"); - let alias = temp.path().join("alias"); - std::fs::create_dir(&real).unwrap(); - std::os::unix::fs::symlink(&real, &alias).unwrap(); - let lock_path = alias.join("task.lock"); - let outside = temp.path().join("outside"); - let token = "1".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); - let contents = format!("pid=invalid\ncreated_at=1\ntoken={token}\n"); - std::fs::write(&outside, &contents).unwrap(); - std::os::unix::fs::symlink(&outside, &lock_path).unwrap(); - - assert!(read_task_lock_snapshot(&lock_path).is_err()); - assert!(prepare_task_lock_publication(&lock_path, &token, 200).is_err()); - assert!(remove_owned_task_lock_blocking(&lock_path, &token).is_err()); - assert!(try_acquire_task_lock_blocking(&lock_path, &token, Some(0), 200).is_err()); - assert!(lock_path.is_symlink()); - assert_eq!(std::fs::read_to_string(&outside).unwrap(), contents); - assert!(!tracedecay_runtime_core::storage::append_lock_path(&outside).exists()); - } - - #[test] - fn uncertain_old_task_lock_guard_preserves_replacement_token() { + fn pre_lease_owner_symlink_denial_releases_both_coordination_locks() { let temp = tempdir().unwrap(); let lock_dir = temp.path().join("automation_locks"); tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(&lock_dir) .unwrap(); - let lock_path = lock_dir.join("skill_writer.lock"); - let old_token = "c".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); - let old_staging_path = task_lock_staging_path(&lock_path, &old_token).unwrap(); - let old_lock = create_task_lock_file_with_post_publish( - &lock_path, - &old_token, - 100, - |_| { - Err(std::io::Error::other( - "injected staging cleanup uncertainty", - )) - }, - |_| Ok(()), - ) - .unwrap(); - tracedecay_runtime_core::storage::PrivateStoreIo::remove_file_durable(&lock_path).unwrap(); - - let replacement_token = "d".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); - let replacement = create_task_lock_file(&lock_path, &replacement_token, 200).unwrap(); - let replacement_record = std::fs::read_to_string(&lock_path).unwrap(); + let outside = temp.path().join("outside-owner"); + std::fs::write(&outside, b"unchanged").unwrap(); + let lock_path = lock_dir.join("memory_curator.lock"); + std::os::unix::fs::symlink(&outside, &lock_path).unwrap(); - drop(old_lock); - assert_eq!( - std::fs::read_to_string(&lock_path).unwrap(), - replacement_record, - "the uncertain prior guard must not unlink a newer owner token" - ); + assert!(try_acquire_task_lock_blocking(&lock_dir, "memory_curator").is_err()); + assert_eq!(std::fs::read(&outside).unwrap(), b"unchanged"); + assert!(lock_path.is_symlink()); + // Remove only the fixture symlink; the denied acquisition must leave + // both stable lock files reusable, without touching the outside target. + std::fs::remove_file(lock_path).unwrap(); assert!( - !old_staging_path.exists(), - "the prior guard must retire only its exact staging residue" + try_acquire_task_lock_blocking(&lock_dir, "memory_curator") + .unwrap() + .is_some() ); - - drop(replacement); - assert!(!lock_path.exists()); } - #[test] - fn task_lock_is_reclaimable_with_unparseable_pid_and_stale_created_at() { - let snapshot = AutomationTaskLockSnapshot { - pid: None, - created_at: Some(100), - started_at: None, - ownership_token: None, - }; - assert!( - task_lock_is_reclaimable(&snapshot, Some(10), 200), - "an unparseable pid must fall back to created_at staleness" - ); + fn write_pre_lease_lock(lock_dir: &Path, key: &str, payload: &[u8]) -> PathBuf { + tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(lock_dir).unwrap(); + let lock_path = lock_dir.join(format!("{key}.lock")); + std::fs::write(&lock_path, payload).unwrap(); + std::fs::write( + tracedecay_runtime_core::storage::append_lock_path(&lock_path), + b"", + ) + .unwrap(); + lock_path } #[test] - fn task_lock_is_reclaimable_with_unparseable_pid_and_fresh_created_at() { - let snapshot = AutomationTaskLockSnapshot { - pid: None, - created_at: Some(195), - started_at: None, - ownership_token: None, - }; - assert!( - !task_lock_is_reclaimable(&snapshot, Some(10), 200), - "a fresh created_at must not be reclaimable even without a pid" - ); - } + fn live_pre_lease_holder_keeps_the_key_and_its_files() { + let temp = tempdir().unwrap(); + let lock_dir = temp.path().join("automation_locks"); + let payload = format!("pid={}\ncreated_at=0\n", std::process::id()); + let lock_path = write_pre_lease_lock(&lock_dir, "memory_curator", payload.as_bytes()); - #[test] - fn task_lock_is_reclaimable_with_no_pid_and_no_created_at_is_crash_debris() { - let snapshot = AutomationTaskLockSnapshot { - pid: None, - created_at: None, - started_at: None, - ownership_token: None, - }; assert!( - task_lock_is_reclaimable(&snapshot, Some(10), 200), - "a lock with neither a parseable pid nor any created_at (payload \ - and mtime both unavailable) must not be permanently wedged" + try_acquire_task_lock_blocking(&lock_dir, "memory_curator") + .unwrap() + .is_none(), + "a live old-format holder must keep exclusion however old its lock is" ); - } - - #[test] - fn task_lock_is_reclaimable_denies_live_process_regardless_of_age() { - let snapshot = AutomationTaskLockSnapshot { - pid: Some(std::process::id()), - created_at: Some(0), - started_at: None, - ownership_token: None, - }; + assert!(lock_path.exists()); assert!( - !task_lock_is_reclaimable(&snapshot, Some(10), 200), - "a live owner must never be reclaimed, no matter how old the lock is" + try_acquire_task_lock_blocking(&lock_dir, "other_key") + .unwrap() + .is_some() ); } - fn reapable_process_program() -> &'static str { - if cfg!(windows) { "cmd" } else { "true" } - } - - fn reapable_process_args() -> &'static [&'static str] { - if cfg!(windows) { &["/C", "exit"] } else { &[] } - } - - #[test] - fn task_lock_is_reclaimable_for_a_fresh_confirmed_dead_pid() { - let mut child = std::process::Command::new(reapable_process_program()) - .args(reapable_process_args()) + fn reaped_pid() -> u32 { + let (program, args): (&str, &[&str]) = if cfg!(windows) { + ("cmd", &["/C", "exit"]) + } else { + ("true", &[]) + }; + let mut child = std::process::Command::new(program) + .args(args) .spawn() .expect("spawn a process to reap"); - let dead_pid = child.id(); - child.wait().expect("reap lock owner"); - assert!( - matches!(process_state(dead_pid), ProcessState::Dead), - "the reaped child must be a confirmed-dead owner" - ); - let snapshot = AutomationTaskLockSnapshot { - pid: Some(dead_pid), - created_at: Some(200), - started_at: None, - ownership_token: None, - }; - assert!( - task_lock_is_reclaimable(&snapshot, Some(6 * 60 * 60), 200), - "a confirmed-dead owner must be reclaimed immediately, not after the age gate" - ); - assert!( - task_lock_is_reclaimable(&snapshot, None, 200), - "disabling the age fallback must not keep a dead owner's lock" - ); - } - - #[test] - fn task_lock_is_reclaimable_when_a_live_pid_start_time_does_not_match() { - let Some(live_started_at) = - tracedecay_runtime_core::lifecycle_lease::process_start_time(std::process::id()) - else { - return; - }; - let mismatched = live_started_at.saturating_add(1); - let snapshot = AutomationTaskLockSnapshot { - pid: Some(std::process::id()), - created_at: Some(200), - started_at: Some(mismatched), - ownership_token: None, - }; - assert!( - task_lock_is_reclaimable(&snapshot, None, 200), - "a reused PID must not keep a lock published by a different process start" - ); - let same_owner = AutomationTaskLockSnapshot { - started_at: Some(live_started_at), - ..snapshot - }; - assert!( - !task_lock_is_reclaimable(&same_owner, Some(0), 10_000), - "a matching start time is the same owner even when the lock is old" - ); + let pid = child.id(); + child.wait().expect("reap process"); + assert!(matches!(process_state(pid), ProcessState::Dead)); + pid } #[test] - fn task_lock_is_reclaimed_end_to_end_when_payload_is_garbage_and_stale_after_is_zero() { + fn dead_pre_lease_owner_is_retired_without_replacing_coordination() { let temp = tempdir().unwrap(); let lock_dir = temp.path().join("automation_locks"); - tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable(&lock_dir) - .unwrap(); - let lock_path = lock_dir.join("garbage_payload.lock"); - - // A payload larger than MAX_AUTOMATION_TASK_LOCK_BYTES (and not even - // valid `key=value` text) means read_task_lock_snapshot can parse no - // pid at all. With stale_after_secs = Some(0), the created_at - // fallback (the file's own mtime) always qualifies as stale, so the - // lock must still be reclaimable on the retry attempt inside - // try_acquire_task_lock_blocking rather than wedging forever. - let garbage = vec![0xFFu8; 2000]; - assert!(garbage.len() as u64 > MAX_AUTOMATION_TASK_LOCK_BYTES); - std::fs::write(&lock_path, &garbage).unwrap(); - - let contender_token = "9".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2); - let acquired = try_acquire_task_lock_blocking(&lock_path, &contender_token, Some(0), 200) - .unwrap() - .expect("a garbage-payload lock with stale_after_secs=0 must be reclaimable"); + let payload = format!("pid={}\ncreated_at=0\n", reaped_pid()); + let lock_path = write_pre_lease_lock(&lock_dir, "memory_curator", payload.as_bytes()); + let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&lock_path); - let snapshot = read_task_lock_snapshot(&lock_path).unwrap().unwrap(); - assert_eq!( - snapshot.ownership_token.as_deref(), - Some(contender_token.as_str()), - "the contender's fresh token must now own the lock" - ); + let held = try_acquire_task_lock_blocking(&lock_dir, "memory_curator") + .unwrap() + .expect("a confirmed-dead old holder no longer owns the key"); - drop(acquired); assert!(!lock_path.exists()); + assert!(coordination_path.exists()); + let old = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&coordination_path) + .unwrap(); + assert!(matches!( + old.try_lock(), + Err(std::fs::TryLockError::WouldBlock) + )); + drop(held); + old.try_lock() + .expect("release admits the old coordination inode"); + old.unlock().unwrap(); } - /// Acquires a real task lock off-runtime so each release-path test starts - /// from the same live-owner state. - fn acquire_lock_for_release_test(lock_path: &Path) -> AutomationTaskLock { - tracedecay_runtime_core::storage::PrivateStoreIo::create_dir_all_durable( - lock_path.parent().unwrap(), - ) - .unwrap(); - try_acquire_task_lock_blocking( - lock_path, - &"c".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2), - Some(10), - 100, - ) - .unwrap() - .expect("a fresh lock path must be acquirable") - } - - /// A contender must be able to take the lock the instant the guard's drop - /// returns. - fn assert_lock_released(lock_path: &Path) { - assert!( - !lock_path.exists(), - "guard drop must remove the lock file synchronously" - ); - let contender = try_acquire_task_lock_blocking( - lock_path, - &"d".repeat(AUTOMATION_TASK_LOCK_TOKEN_BYTES * 2), - Some(10), - 200, - ) - .unwrap() - .expect("a released lock must be immediately acquirable by a contender"); - drop(contender); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn task_lock_release_completes_on_a_multi_thread_worker() { - let temp = tempdir().unwrap(); - let lock_path = temp - .path() - .join("automation_locks") - .join("multi_thread_worker.lock"); - - // Acquire the way production does (spawn_blocking), then move the guard - // back onto an async worker so its drop runs on the runtime. - let acquire_path = lock_path.clone(); - let guard = tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - acquire_lock_for_release_test(&acquire_path) - }) - .await - .unwrap(); - assert!(lock_path.exists()); - - drop(guard); - assert_lock_released(&lock_path); - } - - /// The ordering that wedged `daemon::tests::rmcp_route` for a full drain - /// bound: a worker offers a blocking task to a pool that has already begun - /// shutting down, and tokio's `blocking::pool::Spawner::spawn_task` shuts - /// the refused task down *while holding* the pool's non-reentrant mutex. - /// The refused closure owns this guard, so its release runs under that - /// mutex, and a release that re-enters the runtime never returns, leaving - /// `BlockingPool::shutdown` waiting for the thread forever. - /// - /// Ordering is fixed by channels, not timing: the worker is parked until - /// after `shutdown_timeout` has returned, so the pool is provably closed - /// before `spawn_blocking` is offered and the refusal branch is the only - /// branch reachable. The `shutdown_timeout` budget only bounds tokio's - /// attempt to join the deliberately parked worker. #[test] - fn task_lock_release_survives_a_drop_inside_a_refused_spawn_blocking() { + fn reused_pid_pre_lease_lock_is_retired() { + let live_started_at = + tracedecay_runtime_core::lifecycle_lease::process_start_time(std::process::id()) + .expect("current process start-time evidence"); let temp = tempdir().unwrap(); - let lock_path = temp - .path() - .join("automation_locks") - .join("refused_spawn_blocking.lock"); - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(1) - .enable_all() - .build() - .unwrap(); - let guard = acquire_lock_for_release_test(&lock_path); - assert!(lock_path.exists()); - - let (on_worker_tx, on_worker_rx) = std::sync::mpsc::channel(); - let (pool_closed_tx, pool_closed_rx) = std::sync::mpsc::channel(); - let (offered_tx, offered_rx) = std::sync::mpsc::channel(); - runtime.spawn(async move { - // Reached only from a worker thread, which is the sole context - // where the release can re-enter the blocking spawner. - on_worker_tx.send(()).unwrap(); - pool_closed_rx.recv().unwrap(); - let _refused = tokio::task::spawn_blocking(move || drop(guard)); - offered_tx.send(()).unwrap(); - }); - on_worker_rx.recv().unwrap(); - - runtime.shutdown_timeout(std::time::Duration::from_millis(50)); - pool_closed_tx.send(()).unwrap(); + let lock_dir = temp.path().join("automation_locks"); + let payload = format!( + "pid={}\nstarted_at={}\n", + std::process::id(), + live_started_at.saturating_add(1) + ); + let lock_path = write_pre_lease_lock(&lock_dir, "memory_curator", payload.as_bytes()); - offered_rx - .recv_timeout(std::time::Duration::from_secs(30)) - .expect("releasing the guard under the blocking pool's own mutex must not deadlock"); - assert_lock_released(&lock_path); + try_acquire_task_lock_blocking(&lock_dir, "memory_curator") + .unwrap() + .expect("a pid reused by a different process start is not the old owner"); + assert!(!lock_path.exists()); } #[test] - fn task_lock_release_works_with_no_tokio_runtime() { + fn unverifiable_pre_lease_lock_is_an_error_not_abandonment() { let temp = tempdir().unwrap(); - let lock_path = temp.path().join("automation_locks").join("no_runtime.lock"); + let lock_dir = temp.path().join("automation_locks"); + let lock_path = write_pre_lease_lock(&lock_dir, "memory_curator", &[0xFF; 2000]); + + assert!(try_acquire_task_lock_blocking(&lock_dir, "memory_curator").is_err()); + assert!(lock_path.exists(), "unverified ownership must be preserved"); assert!( - tokio::runtime::Handle::try_current().is_err(), - "this test must run without a runtime handle so the inline path is exercised" + try_acquire_task_lock_blocking(&lock_dir, "memory_curator").is_err(), + "the lease taken by the failed attempt must have been released" ); - - let guard = acquire_lock_for_release_test(&lock_path); - assert!(lock_path.exists()); - drop(guard); - assert_lock_released(&lock_path); } - #[tokio::test] - async fn task_lock_release_completes_on_a_current_thread_runtime() { + #[test] + fn pre_lease_owner_checks_preserve_live_and_unknown_records() { let temp = tempdir().unwrap(); - let lock_path = temp - .path() - .join("automation_locks") - .join("current_thread.lock"); - assert_eq!( - tokio::runtime::Handle::current().runtime_flavor(), - tokio::runtime::RuntimeFlavor::CurrentThread, - "#[tokio::test] must default to the current-thread flavor for this case" + let lock_dir = temp.path().join("automation_locks"); + let pid = std::process::id(); + let started_at = tracedecay_runtime_core::lifecycle_lease::process_start_time(pid) + .expect("current process start-time evidence"); + let live = format!("pid={pid}\nstarted_at={started_at}\ncreated_at=0\n"); + let live_path = write_pre_lease_lock(&lock_dir, "live", live.as_bytes()); + assert!( + try_acquire_task_lock_blocking(&lock_dir, "live") + .unwrap() + .is_none() ); - - let guard = acquire_lock_for_release_test(&lock_path); - assert!(lock_path.exists()); - drop(guard); - assert_lock_released(&lock_path); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn task_lock_release_from_spawn_blocking_owner_does_not_panic() { - let temp = tempdir().unwrap(); - let lock_path = temp - .path() - .join("automation_locks") - .join("settlement_owner.lock"); - - // The settlement-owner pattern in daemon::automation_effect acquires and - // drops the guard entirely inside spawn_blocking, which is where callers - // that cannot afford to block a worker are expected to release it. - let owned_path = lock_path.clone(); - tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - let guard = acquire_lock_for_release_test(&owned_path); - assert!(owned_path.exists()); - drop(guard); - assert!(!owned_path.exists()); - }) - .await - .expect("dropping the guard inside spawn_blocking must not panic"); - - assert_lock_released(&lock_path); + assert_eq!(std::fs::read(live_path).unwrap(), live.as_bytes()); + for (key, payload) in [ + ("unknown", "pid=0\ncreated_at=0\n".to_owned()), + ("duplicate", format!("pid={pid}\npid={pid}\ncreated_at=0\n")), + ("missing", "created_at=0\n".to_owned()), + ] { + let path = write_pre_lease_lock(&lock_dir, key, payload.as_bytes()); + assert!(try_acquire_task_lock_blocking(&lock_dir, key).is_err()); + assert_eq!(std::fs::read(path).unwrap(), payload.as_bytes()); + } } #[cfg(unix)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn concurrent_task_locks_acquire_through_symlinked_dashboard_prefix() { - // cap-std ambient opens walk each component. A prefix symlink - // (`tmp/link` → `tmp/real`, or macOS `/var` → `/private/var`) must - // resolve before Dir::open_ambient_dir or concurrent acquires ENOENT. + // A prefix symlink (`tmp/link` -> `tmp/real`, or macOS `/var` -> + // `/private/var`) must resolve before cap-std walks the components or + // concurrent acquires ENOENT. let temp = tempdir().unwrap(); let real = temp.path().join("real"); let link = temp.path().join("link"); @@ -2948,22 +2394,10 @@ evidence about it", let first_root = dashboard_root.clone(); let second_root = dashboard_root; - let first = tokio::spawn(async move { - AutomationTaskLock::try_acquire_keyed(&first_root, "concurrent-a", Some(10), 100).await - }); - let second = tokio::spawn(async move { - AutomationTaskLock::try_acquire_keyed(&second_root, "concurrent-b", Some(10), 100).await - }); - let first = first - .await - .expect("join first lock") - .expect("first lock acquire through prefix symlink"); - let second = second - .await - .expect("join second lock") - .expect("second lock acquire through prefix symlink"); + let first = tokio::spawn(async move { acquire(&first_root, "concurrent-a").await }); + let second = tokio::spawn(async move { acquire(&second_root, "concurrent-b").await }); assert!( - first.is_some() && second.is_some(), + first.await.unwrap().is_some() && second.await.unwrap().is_some(), "distinct keys under a shared lock dir must both acquire" ); } diff --git a/crates/tracedecay-cli/BUILD.bazel b/crates/tracedecay-cli/BUILD.bazel index 63c85c0adc..54557ecf2a 100644 --- a/crates/tracedecay-cli/BUILD.bazel +++ b/crates/tracedecay-cli/BUILD.bazel @@ -152,6 +152,7 @@ rust_test( rust_test( name = "refresh_memory_journey", + size = "large", srcs = glob(["tests/**/*.rs"], exclude = ["tests/fixtures/**"]) + glob(["build-support/**/*.rs", "src/**/*.rs"]) + ["//crates/tracedecay:tests_common_sources", "//tests:support_sources"], crate_root = "tests/refresh_memory_journey.rs", edition = crate_edition(), diff --git a/crates/tracedecay-cli/Cargo.toml b/crates/tracedecay-cli/Cargo.toml index 7f885fff17..a8175f2c44 100644 --- a/crates/tracedecay-cli/Cargo.toml +++ b/crates/tracedecay-cli/Cargo.toml @@ -7,6 +7,12 @@ description = "TraceDecay command-line interface and daemon binary" license = "MIT" repository = "https://github.com/ScriptedAlchemy/tracedecay" +# Five complete 48,000-function generations take 53–65 seconds each on the +# four-vCPU fleet. Keep the memory workload intact instead of letting the +# default five-minute medium deadline interrupt its final verification. +[package.metadata.bazel.test_sizes] +refresh_memory_journey = "large" + [[bin]] name = "tracedecay" path = "src/main.rs" diff --git a/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs index 62666fee20..5c5dbe1708 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs @@ -1439,12 +1439,15 @@ fn observe_http_connections(action: impl FnOnce(SocketAddr)) -> usize { let observer = scope.spawn(move || { let mut accepted = 0; loop { - match listener.accept() { + // Read the flag before accepting: a WouldBlock that follows a + // set flag proves the backlog was already empty once the + // action finished, so no completed connection is left behind. + let stopping = stop.load(Ordering::SeqCst); + let connection = listener.accept(); + match connection { Ok((_stream, _peer)) => accepted += 1, - // Stop only after all connections already in the backlog - // have been observed, regardless of thread scheduling. Err(error) if error.kind() == ErrorKind::WouldBlock => { - if stop.load(Ordering::SeqCst) { + if stopping { return Ok::<_, std::io::Error>(accepted); } std::thread::sleep(Duration::from_millis(5)); diff --git a/crates/tracedecay-daemon-service/src/automation_effect/journal/tests.rs b/crates/tracedecay-daemon-service/src/automation_effect/journal/tests.rs index 2e85d97d33..d0a386a93f 100644 --- a/crates/tracedecay-daemon-service/src/automation_effect/journal/tests.rs +++ b/crates/tracedecay-daemon-service/src/automation_effect/journal/tests.rs @@ -365,13 +365,6 @@ async fn retained_disabled_user_job_run( .await } -fn now_secs() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("wall clock") - .as_secs() as i64 -} - async fn retained_recovery_project( temp: &tempfile::TempDir, name: &str, @@ -399,8 +392,6 @@ async fn fixed_task_lock_is_denied( tracedecay_automation_runtime::automation::scheduler::AutomationTaskLock::try_acquire( dashboard_root, task, - None, - now_secs(), ) .await .expect("competing fixed-task lock") @@ -1292,8 +1283,6 @@ async fn reused_scheduler_skip_abandons_current_effect_before_observing_exact_pr tracedecay_automation_runtime::automation::scheduler::AutomationTaskLock::try_acquire( dashboard_root, AgentTaskKind::MemoryCurator, - None, - now_secs(), ) .await .expect("post-abandonment task lock") diff --git a/crates/tracedecay-lsp/src/compile_diagnostics/typescript.rs b/crates/tracedecay-lsp/src/compile_diagnostics/typescript.rs index c07b50e817..5eee4f57a1 100644 --- a/crates/tracedecay-lsp/src/compile_diagnostics/typescript.rs +++ b/crates/tracedecay-lsp/src/compile_diagnostics/typescript.rs @@ -21,17 +21,64 @@ use std::path::Path; use std::process::Command; +use std::sync::Arc; use std::time::{Duration, Instant}; +use tokio::sync::Semaphore; + use super::{Diagnostic, is_diagnostic_level}; use tracedecay_domain::errors::{Result, TraceDecayError}; +use tracedecay_runtime_core::background_cpu::ProcessBackgroundCpuV1; use tracedecay_runtime_core::cancellation::CancellationToken; use tracedecay_runtime_core::git::{GitCommandBounds, GitCommandError, bounded_command_output}; use tracedecay_runtime_core::native_runtime; -/// tsc has no deadline of its own: a check runs until it finishes or its -/// caller drops it. This only satisfies the bounded runner's required field. -const UNBOUNDED_CHECK: Duration = Duration::from_hours(365 * 24); +/// Limit simultaneous compiler children as well as their background CPU demand. +const MAX_CONCURRENT_COMPILERS: usize = 4; + +/// Shared by project producers in one daemon generation. +#[derive(Clone)] +pub struct TypeScriptCompilerAdmission { + permits: Arc, + background_cpu: Arc, + cancellation: CancellationToken, +} + +impl TypeScriptCompilerAdmission { + #[must_use] + pub fn new(background_cpu: Arc) -> Self { + Self { + permits: Arc::new(Semaphore::new(MAX_CONCURRENT_COMPILERS)), + background_cpu, + cancellation: CancellationToken::new(), + } + } + + #[must_use] + pub fn concurrency(&self) -> usize { + MAX_CONCURRENT_COMPILERS + } + + pub fn cancel(&self) { + self.permits.close(); + self.cancellation.cancel(); + } +} + +/// What one compiler child may consume once it holds a permit. A check that +/// exceeds any bound fails; it never reports a clean project. +#[derive(Clone, Copy)] +struct CheckLimits { + deadline: Duration, + max_stdout_bytes: usize, + max_stderr_bytes: usize, +} + +const CHECK_LIMITS: CheckLimits = CheckLimits { + deadline: Duration::from_mins(20), + max_stdout_bytes: 64 * 1024 * 1024, + max_stderr_bytes: 1024 * 1024, +}; /// Installs the project's own compiler where the producer looks for it; the /// command for a package with no workspace lockfile to install from. @@ -47,6 +94,9 @@ pub const TYPESCRIPT_INSTALL_COMMAND: &str = "npm install --save-dev typescript" /// other exit, or a diagnostic without a file location means the project was /// not checked, and that is a typed failure rather than a clean page. /// +/// Producers sharing `admission` share its compiler capacity. A running check is +/// bounded by a deadline and output limits, and exceeding one is a failure. +/// /// With `build_info`, the check is incremental against that /// `--tsBuildInfoFile`: tsc rechecks only what changed since the check that /// wrote it and records the program files it read there. @@ -56,10 +106,30 @@ pub const TYPESCRIPT_INSTALL_COMMAND: &str = "npm install --save-dev typescript" skip_all )] pub async fn run_compiler( + admission: &TypeScriptCompilerAdmission, compiler: &Path, project_root: &Path, tsconfig: &Path, build_info: Option<&Path>, +) -> Result> { + run_compiler_within( + admission, + compiler, + project_root, + tsconfig, + build_info, + CHECK_LIMITS, + ) + .await +} + +async fn run_compiler_within( + admission: &TypeScriptCompilerAdmission, + compiler: &Path, + project_root: &Path, + tsconfig: &Path, + build_info: Option<&Path>, + limits: CheckLimits, ) -> Result> { let mut cmd = Command::new(compiler); cmd.arg("-p") @@ -75,30 +145,62 @@ pub async fn run_compiler( cmd.current_dir(project_root); // Dropping this future cancels the token, and the runner then kills and // reaps tsc. - let cancel = CancellationToken::new(); + let cancel = admission.cancellation.child_token(); let _cancel_on_drop = CancelOnDrop(cancel.clone()); + let permit = Arc::clone(&admission.permits) + .acquire_owned() + .await + .map_err(|_| TraceDecayError::Config { + message: "the TypeScript compiler admission closed".to_owned(), + })?; let bounds = GitCommandBounds { - deadline: Instant::now() + UNBOUNDED_CHECK, + deadline: Instant::now() + limits.deadline, cancel: Some(cancel), - max_stdout_bytes: usize::MAX, - max_stderr_bytes: usize::MAX, + max_stdout_bytes: limits.max_stdout_bytes, + max_stderr_bytes: limits.max_stderr_bytes, }; - let output = native_runtime::spawn_blocking(move || bounded_command_output(cmd, None, &bounds)) - .await - .map_err(|error| TraceDecayError::Config { - message: format!("failed to run `{}`: {error}", compiler.display()), - })? - .map_err(|error| TraceDecayError::Config { - message: match error { - GitCommandError::Unavailable(error) => { - format!("failed to spawn `{}`: {error}", compiler.display()) - } - GitCommandError::ReadOutput { stream, source } => { - format!("failed to read `{}` {stream}: {source}", compiler.display()) + let background_cpu = Arc::clone(&admission.background_cpu); + // The permit travels with the blocking runner, which kills and reaps the + // child before returning, so it is held for the child's whole lifetime + // even when this future is dropped first. + let output = native_runtime::spawn_blocking(move || { + let _permit = permit; + let _cpu = background_cpu + .acquire_cancellable(|| { + bounds + .cancel + .as_ref() + .is_some_and(CancellationToken::is_cancelled) + || Instant::now() >= bounds.deadline + }) + .ok_or_else(|| { + if bounds + .cancel + .as_ref() + .is_some_and(CancellationToken::is_cancelled) + { + GitCommandError::Cancelled + } else { + GitCommandError::DeadlineExceeded } - other => format!("`{}` did not finish: {other}", compiler.display()), - }, - })?; + })?; + bounded_command_output(cmd, None, &bounds) + }) + .await + .map_err(|error| TraceDecayError::Config { + message: format!("failed to run `{}`: {error}", compiler.display()), + })? + .map_err(|error| TraceDecayError::Config { + message: match error { + GitCommandError::Unavailable(error) => { + format!("failed to spawn `{}`: {error}", compiler.display()) + } + GitCommandError::ReadOutput { stream, source } => { + format!("failed to read `{}` {stream}: {source}", compiler.display()) + } + other => format!("`{}` did not finish: {other}", compiler.display()), + }, + })?; let stdout = String::from_utf8_lossy(&output.stdout); if let Some(global) = stdout .lines() @@ -368,6 +470,13 @@ src/b.ts(2,2): warning TS6133: Second. native_runtime::block_on(future).expect("native runtime") } + #[cfg(unix)] + fn admission() -> TypeScriptCompilerAdmission { + TypeScriptCompilerAdmission::new(Arc::new(ProcessBackgroundCpuV1::new( + std::num::NonZeroUsize::new(MAX_CONCURRENT_COMPILERS).unwrap(), + ))) + } + #[cfg(unix)] #[test] fn run_compiler_accepts_every_diagnostics_present_exit_code() { @@ -383,8 +492,14 @@ src/b.ts(2,2): warning TS6133: Second. ), ).unwrap(); - let diagnostics = block(run_compiler(&compiler, project.path(), &tsconfig, None)) - .unwrap_or_else(|error| panic!("exit {exit} is a checked project: {error}")); + let diagnostics = block(run_compiler( + &admission(), + &compiler, + project.path(), + &tsconfig, + None, + )) + .unwrap_or_else(|error| panic!("exit {exit} is a checked project: {error}")); assert_eq!(diagnostics.len(), 1, "exit {exit}"); assert_eq!(diagnostics[0].file, "packages/app/src/(auth)/index.ts"); assert_eq!(diagnostics[0].code, "TS4023"); @@ -409,6 +524,7 @@ src/b.ts(2,2): warning TS6133: Second. ) .unwrap(); let result = block(run_compiler( + &admission(), &compiler, project.path(), &project.path().join("tsconfig.json"), @@ -425,6 +541,7 @@ src/b.ts(2,2): warning TS6133: Second. let compiler = project.path().join("tsc"); write_executable_script(&compiler, "#!/bin/sh\nexit 0\n").unwrap(); let diagnostics = block(run_compiler( + &admission(), &compiler, project.path(), &project.path().join("tsconfig.json"), @@ -446,6 +563,7 @@ src/b.ts(2,2): warning TS6133: Second. .unwrap(); let error = block(run_compiler( + &admission(), &compiler, project.path(), &project.path().join("tsconfig.json"), @@ -461,6 +579,17 @@ src/b.ts(2,2): warning TS6133: Second. #[cfg(unix)] #[test] fn dropping_a_check_kills_the_compiler() { + assert_running_check_settles(false); + } + + #[cfg(unix)] + #[test] + fn closing_admission_kills_the_running_compiler() { + assert_running_check_settles(true); + } + + #[cfg(unix)] + fn assert_running_check_settles(close_admission: bool) { let project = tempfile::tempdir().unwrap(); let compiler = project.path().join("tsc"); let pid_file = project.path().join("tsc.pid"); @@ -471,8 +600,15 @@ src/b.ts(2,2): warning TS6133: Second. .unwrap(); let tsconfig = project.path().join("tsconfig.json"); + let admission = admission(); let pid = block(async { - let check = std::pin::pin!(run_compiler(&compiler, project.path(), &tsconfig, None)); + let check = std::pin::pin!(run_compiler( + &admission, + &compiler, + project.path(), + &tsconfig, + None + )); let started = std::pin::pin!(async { loop { if let Ok(pid) = std::fs::read_to_string(&pid_file) { @@ -487,15 +623,22 @@ src/b.ts(2,2): warning TS6133: Second. futures_util::future::Either::Left((result, _)) => { panic!("the check finished before it was dropped: {result:?}") } - futures_util::future::Either::Right((pid, _check)) => pid, + futures_util::future::Either::Right((pid, check)) => { + if close_admission { + admission.cancel(); + assert!(check.await.is_err()); + } + pid + } } }); let deadline = Instant::now() + Duration::from_secs(10); while Command::new("kill") .args(["-0", &pid]) - .status() + .output() .unwrap() + .status .success() { assert!( @@ -504,5 +647,280 @@ src/b.ts(2,2): warning TS6133: Second. ); std::thread::sleep(Duration::from_millis(10)); } + let deadline = Instant::now() + Duration::from_secs(10); + while admission.permits.available_permits() != MAX_CONCURRENT_COMPILERS { + assert!( + Instant::now() < deadline, + "child settlement retained a permit" + ); + std::thread::yield_now(); + } + assert_eq!(admission.background_cpu.active_units(), 0); + } + + #[cfg(unix)] + #[test] + fn dropping_a_waiting_check_never_spawns_and_releases_its_demand() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script(&compiler, "#!/bin/sh\necho ran >> ran\nexit 0\n").unwrap(); + let tsconfig = project.path().join("tsconfig.json"); + let admission = admission(); + block(async { + let held = admission + .permits + .acquire_many(MAX_CONCURRENT_COMPILERS as u32) + .await + .unwrap(); + let mut waiting = Box::pin(run_compiler( + &admission, + &compiler, + project.path(), + &tsconfig, + None, + )); + assert!(futures_util::poll!(waiting.as_mut()).is_pending()); + assert!(!project.path().join("ran").exists()); + drop(waiting); + drop(held); + assert!( + run_compiler(&admission, &compiler, project.path(), &tsconfig, None) + .await + .unwrap() + .is_empty() + ); + }); + assert_eq!( + std::fs::read_to_string(project.path().join("ran")).unwrap(), + "ran\n" + ); + assert_eq!( + admission.permits.available_permits(), + MAX_CONCURRENT_COMPILERS + ); + } + + #[cfg(unix)] + #[test] + fn closed_admission_refuses_waiters_and_successor_checks() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script(&compiler, "#!/bin/sh\necho ran > ran\nexit 0\n").unwrap(); + let tsconfig = project.path().join("tsconfig.json"); + let admission = admission(); + block(async { + let held = admission + .permits + .acquire_many(MAX_CONCURRENT_COMPILERS as u32) + .await + .unwrap(); + let mut waiting = Box::pin(run_compiler( + &admission, + &compiler, + project.path(), + &tsconfig, + None, + )); + assert!(futures_util::poll!(waiting.as_mut()).is_pending()); + admission.cancel(); + assert!( + waiting + .await + .unwrap_err() + .to_string() + .contains("admission closed") + ); + drop(held); + assert!( + run_compiler(&admission, &compiler, project.path(), &tsconfig, None) + .await + .is_err() + ); + }); + assert!(!project.path().join("ran").exists()); + } + + #[cfg(unix)] + #[test] + fn compiler_waits_for_the_existing_background_cpu_authority() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script(&compiler, "#!/bin/sh\necho ran > ran\nexit 0\n").unwrap(); + let tsconfig = project.path().join("tsconfig.json"); + let cpu = Arc::new(ProcessBackgroundCpuV1::new( + std::num::NonZeroUsize::new(1).unwrap(), + )); + let admission = TypeScriptCompilerAdmission::new(Arc::clone(&cpu)); + let held = cpu.acquire(); + let release = std::thread::spawn({ + let cpu = Arc::clone(&cpu); + let root = project.path().to_owned(); + move || { + let deadline = Instant::now() + Duration::from_secs(10); + while cpu.waiting_work_units() == 0 { + assert!(Instant::now() < deadline); + std::thread::yield_now(); + } + assert!(!root.join("ran").exists()); + drop(held); + } + }); + assert!( + block(run_compiler( + &admission, + &compiler, + project.path(), + &tsconfig, + None + )) + .unwrap() + .is_empty() + ); + release.join().unwrap(); + assert!(project.path().join("ran").exists()); + assert_eq!(cpu.active_units(), 0); + assert_eq!(cpu.waiting_work_units(), 0); + } + + #[cfg(unix)] + fn tiny_limits() -> CheckLimits { + CheckLimits { + deadline: Duration::from_millis(300), + max_stdout_bytes: 4096, + max_stderr_bytes: 4096, + } + } + + #[cfg(unix)] + #[test] + fn endless_compiler_output_fails_the_check() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script( + &compiler, + "#!/bin/sh\nexec yes 'a.ts(1,1): error TS1: x.'\n", + ) + .unwrap(); + let tsconfig = project.path().join("tsconfig.json"); + let error = block(run_compiler_within( + &admission(), + &compiler, + project.path(), + &tsconfig, + None, + CheckLimits { + deadline: Duration::from_secs(30), + ..tiny_limits() + }, + )) + .expect_err("output beyond the bound is a failed check, not a report"); + assert!(error.to_string().contains("exceeded"), "{error}"); + } + + #[cfg(unix)] + #[test] + fn endless_compiler_stderr_fails_the_check() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script(&compiler, "#!/bin/sh\nexec yes 'failure' >&2\n").unwrap(); + let error = block(run_compiler_within( + &admission(), + &compiler, + project.path(), + &project.path().join("tsconfig.json"), + None, + CheckLimits { + deadline: Duration::from_secs(30), + ..tiny_limits() + }, + )) + .unwrap_err(); + assert!( + error + .to_string() + .contains("stderr output exceeded 4096 bytes"), + "{error}" + ); + } + + #[cfg(unix)] + #[test] + fn a_stalled_compiler_fails_at_its_deadline() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script(&compiler, "#!/bin/sh\nexec sleep 60\n").unwrap(); + let tsconfig = project.path().join("tsconfig.json"); + let started = Instant::now(); + let error = block(run_compiler_within( + &admission(), + &compiler, + project.path(), + &tsconfig, + None, + tiny_limits(), + )) + .expect_err("a stalled compiler never reports clean"); + assert!(error.to_string().contains("deadline"), "{error}"); + assert!(started.elapsed() < Duration::from_secs(20)); + } + + #[cfg(unix)] + #[test] + fn a_large_within_policy_report_retains_every_finding() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script( + &compiler, + "#!/bin/sh\ni=0; while [ \"$i\" -lt 1000 ]; do echo 'a.ts(1,1): error TS1: x.'; i=$((i+1)); done\nexit 2\n", + ).unwrap(); + let diagnostics = block(run_compiler_within( + &admission(), + &compiler, + project.path(), + &project.path().join("tsconfig.json"), + None, + CheckLimits { + deadline: Duration::from_secs(30), + max_stdout_bytes: 64 * 1024, + max_stderr_bytes: 4096, + }, + )) + .unwrap(); + assert_eq!(diagnostics.len(), 1000); + } + + #[cfg(unix)] + #[test] + fn concurrent_checks_never_exceed_the_shared_compiler_capacity() { + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + write_executable_script( + &compiler, + "#!/bin/sh\n\ + d=\"$(dirname \"$0\")\"\n\ + mkdir \"$d/active.$$\"\n\ + ls -d \"$d\"/active.* | wc -l >> \"$d/seen\"\n\ + sleep 0.3\n\ + rmdir \"$d/active.$$\"\n\ + exit 0\n", + ) + .unwrap(); + let tsconfig = project.path().join("tsconfig.json"); + let admission = admission(); + let results = block(futures_util::future::join_all( + (0..MAX_CONCURRENT_COMPILERS * 3) + .map(|_| run_compiler(&admission, &compiler, project.path(), &tsconfig, None)), + )); + assert!(results.iter().all(Result::is_ok), "{results:?}"); + let seen = std::fs::read_to_string(project.path().join("seen")).unwrap(); + let peak = seen + .lines() + .map(|line| line.trim().parse::().unwrap()) + .max() + .unwrap(); + assert!( + (2..=MAX_CONCURRENT_COMPILERS).contains(&peak), + "peak concurrent compilers {peak}" + ); } } diff --git a/crates/tracedecay-lsp/src/lib.rs b/crates/tracedecay-lsp/src/lib.rs index 1c67a047e0..25e49daafe 100644 --- a/crates/tracedecay-lsp/src/lib.rs +++ b/crates/tracedecay-lsp/src/lib.rs @@ -62,7 +62,8 @@ pub use compile_diagnostics::tsconfig::{ typescript_projects, }; pub use compile_diagnostics::typescript::{ - TYPESCRIPT_INSTALL_COMMAND, run_compiler as run_typescript_compiler, + TYPESCRIPT_INSTALL_COMMAND, TypeScriptCompilerAdmission, + run_compiler as run_typescript_compiler, }; pub use context::{ CanonicalContextProjectionAuthority, ContextCoverage, ContextExpansionEnvelope, diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs index 901c61ed04..7599d4812a 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs @@ -247,7 +247,8 @@ pub(super) async fn run_session_temporal_refresh_scheduler( policy, }; let _instrumentation = SessionTemporalRefreshWorkerInstrumentation::new(&state); - state.mark_running(); + // The registry marks the first worker running. A replacement must retain + // its panic blocker until a pass succeeds, just like other retry classes. loop { if state.cancelled.load(Ordering::Acquire) { return; diff --git a/crates/tracedecay/src/daemon/invocation_state.rs b/crates/tracedecay/src/daemon/invocation_state.rs index 0b5ac7eefc..4568956e7b 100644 --- a/crates/tracedecay/src/daemon/invocation_state.rs +++ b/crates/tracedecay/src/daemon/invocation_state.rs @@ -4,13 +4,13 @@ //! Request cancellation stays threaded through the invocation boundary //! explicitly, including the `multi_root_family_allows` kill-switch. -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use tracedecay_runtime_core::path_safety::canonical_existing_identity; use serde_json::Value; use tracedecay_code_index_runtime::code_index_scheduler; use tracedecay_daemon_identity::profile_identity; -use tracedecay_lsp::LspSessionRegistry; +use tracedecay_lsp::{LspSessionRegistry, TypeScriptCompilerAdmission}; use tracedecay_runtime_core::cancellation::CancellationToken; use tracedecay_runtime_core::resident_memory::{ ProcessResidentMemoryV1, detected_process_resident_memory_limit_v1, @@ -49,6 +49,7 @@ pub(crate) struct DaemonInvocationState { pub(super) github_credential_lifecycle: github_credential_lifecycle::DaemonGitHubReadOnlyCredentialLifecycleV1, pub(super) code_index_schedulers: code_index_scheduler::CodeIndexSchedulerRegistryV1, + pub(super) compiler_admission: Arc>, work_federated_query_authority: Arc, } @@ -85,6 +86,7 @@ impl DaemonInvocationState { github_credential_lifecycle: github_credential_lifecycle::DaemonGitHubReadOnlyCredentialLifecycleV1::default(), code_index_schedulers, + compiler_admission: Arc::new(OnceLock::new()), work_federated_query_authority, } } @@ -168,6 +170,9 @@ impl DaemonInvocationState { .map_err(|error| TraceDecayError::Config { message: format!("failed to configure Codex preparation resources: {error}"), })?; + self.compiler_admission.get_or_init(|| { + TypeScriptCompilerAdmission::new(Arc::clone(&installed.background_cpu)) + }); Ok(installed) } @@ -1064,6 +1069,9 @@ impl DaemonInvocationState { pub(super) fn cancel_admissions(&self) { self.service.cancel_admissions(); self.github_credential_lifecycle.shutdown(); + if let Some(admission) = self.compiler_admission.get() { + admission.cancel(); + } // Code-index workers only observe `shutting_down` / closed admission // once cancel runs. Leaving this until the join lets an in-flight // reconcile (graph seat, follow-up pass) keep the worker alive for @@ -1239,6 +1247,62 @@ impl WorkFederatedQueryAuthorityPortV1 for DaemonWorkFederatedQueryAuthorityV1 { mod shutdown_tests { use super::*; + #[cfg(unix)] + #[tokio::test] + async fn daemon_shutdown_closes_compiler_admission_for_every_project_owner() { + let state = DaemonInvocationState::default(); + assert!(state.compiler_admission.get().is_none()); + state + .install_worker_selection( + &StoreAdministration::default(), + tracedecay_domain::configuration::CodeIndexWorkerSelectionV1::default(), + ) + .unwrap(); + let sibling = state.clone(); + let admission = sibling.compiler_admission.get().unwrap().clone(); + let project = tempfile::tempdir().unwrap(); + let compiler = project.path().join("tsc"); + tracedecay_runtime_core::test_executable::write_executable_script( + &compiler, + "#!/bin/sh\necho ran >> ran\nexit 0\n", + ) + .unwrap(); + let config = project.path().join("tsconfig.json"); + assert!( + tracedecay_lsp::run_typescript_compiler( + &admission, + &compiler, + project.path(), + &config, + None, + ) + .await + .unwrap() + .is_empty() + ); + state.cancel_admissions(); + let error = tracedecay_lsp::run_typescript_compiler( + &admission, + &compiler, + project.path(), + &config, + None, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("admission closed"), "{error}"); + assert_eq!( + std::fs::read_to_string(project.path().join("ran")).unwrap(), + "ran\n" + ); + assert!( + DaemonInvocationState::default() + .compiler_admission + .get() + .is_none() + ); + } + #[tokio::test] async fn failed_lsp_lease_is_preserved_in_the_invocation_owner_receipt() { let state = DaemonInvocationState::default(); diff --git a/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs b/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs index bb242141d1..a06c9ae7d0 100644 --- a/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs @@ -16,6 +16,17 @@ use super::ProductionProjectCompositionHarnessV1; use super::journey_test_support::{initialize_project, tool_payload}; use tracedecay_runtime_core::storage::default_profile_sharded_layout; +/// True while another open file description holds the task lease exclusively. +/// The shared probe lock lasts only for the call. +fn task_lease_is_held(lease: &std::path::Path) -> bool { + std::fs::File::open(lease).is_ok_and(|file| { + matches!( + file.try_lock_shared(), + Err(std::fs::TryLockError::WouldBlock) + ) + }) +} + const RELEASE_DEADLINE: Duration = Duration::from_secs(60); #[derive(Default)] @@ -110,9 +121,9 @@ async fn shutdown_waits_for_accepted_curator_settlement_and_lock_release() { tokio::time::timeout(RELEASE_DEADLINE, pause.entered.notified()) .await .expect("the accepted curator reaches durable publication"); - let task_lock = dashboard_root.join("automation_locks/memory_curator.lock"); + let task_lock = dashboard_root.join("automation_locks/memory_curator.lease"); assert!( - task_lock.exists(), + task_lease_is_held(&task_lock), "the pending settlement still owns its task lock" ); @@ -130,7 +141,7 @@ async fn shutdown_waits_for_accepted_curator_settlement_and_lock_release() { } assert!(!shutdown.is_finished()); assert!( - task_lock.exists(), + task_lease_is_held(&task_lock), "shutdown must preserve the settlement's lock" ); release.send(()).expect("release curator settlement"); @@ -140,7 +151,7 @@ async fn shutdown_waits_for_accepted_curator_settlement_and_lock_release() { .expect("shutdown task"); assert!( - !task_lock.exists(), + !task_lease_is_held(&task_lock), "shutdown must join exact task-lock release" ); let ledger = std::fs::read_to_string(dashboard_root.join("automation_runs.jsonl")) diff --git a/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs b/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs index 1c76d13fbc..ddafe49b9f 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/compiler_diagnostics_producer.rs @@ -49,6 +49,10 @@ pub(super) fn spawn_typescript_diagnostics_producer( log_producer_event(&project_root, "code_index_disabled"); return false; } + let Some(admission) = invocation.compiler_admission.get().cloned() else { + log_producer_event(&project_root, "compiler_admission_unavailable"); + return false; + }; owner.spawn_background_task(tracing::Instrument::instrument( async move { let Some(projects) = discover_projects(&project_root).await else { @@ -74,7 +78,7 @@ pub(super) fn spawn_typescript_diagnostics_producer( let schedulers = invocation.code_index_schedulers.clone(); let mut signals = CodeIndexOwnerSignalsV1::subscribe(&schedulers, &project_root).await; let build_info_dir = graph.store_layout().data_root.join(BUILD_INFO_DIR); - let mut state = TypeScriptProducerStateV1::new(status); + let mut state = TypeScriptProducerStateV1::new(status, admission); // One producer run per generation, whatever its outcome: the root's // signals also wake on serving and pass changes, and a project // whose compiler keeps failing must not re-run it on each of them. diff --git a/crates/tracedecay/src/daemon/scheduler/combined_effect.rs b/crates/tracedecay/src/daemon/scheduler/combined_effect.rs index 1fcc6e644f..a85ffe60d0 100644 --- a/crates/tracedecay/src/daemon/scheduler/combined_effect.rs +++ b/crates/tracedecay/src/daemon/scheduler/combined_effect.rs @@ -1020,13 +1020,6 @@ mod tests { paths } - fn now_secs() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("wall clock") - .as_secs() as i64 - } - struct RecordingEarlyGateBackend { calls: AtomicUsize, } @@ -1379,14 +1372,19 @@ mod tests { let current_skill_task_lock = fixture .dashboard_root .join("automation_locks") - .join("skill_writer.lock"); + .join("skill_writer.lease"); tokio::time::timeout(std::time::Duration::from_secs(5), async { - while !current_skill_task_lock.is_file() { + while !std::fs::File::open(¤t_skill_task_lock).is_ok_and(|file| { + matches!( + file.try_lock_shared(), + Err(std::fs::TryLockError::WouldBlock) + ) + }) { tokio::select! { outcome = &mut effect => { panic!("partial replay returned before current abandonment: {outcome:?}") } - () = tokio::task::yield_now() => {} + () = tokio::time::sleep(std::time::Duration::from_millis(1)) => {} } } }) @@ -1396,8 +1394,6 @@ mod tests { tracedecay_automation_runtime::automation::scheduler::AutomationTaskLock::try_acquire( &fixture.dashboard_root, AgentTaskKind::SkillWriter, - None, - now_secs(), ) .await .expect("competing skill task lock") @@ -1448,8 +1444,6 @@ mod tests { tracedecay_automation_runtime::automation::scheduler::AutomationTaskLock::try_acquire( &fixture.dashboard_root, AgentTaskKind::SkillWriter, - None, - now_secs(), ) .await .expect("post-abandonment skill task lock") diff --git a/crates/tracedecay/tests/automation_runner_test/combined_review.rs b/crates/tracedecay/tests/automation_runner_test/combined_review.rs index 6f66a14262..cc12bb1f08 100644 --- a/crates/tracedecay/tests/automation_runner_test/combined_review.rs +++ b/crates/tracedecay/tests/automation_runner_test/combined_review.rs @@ -323,10 +323,9 @@ async fn retained_combined_review_defers_both_ledgers_and_holds_both_task_locks( "retained dispatch must not publish either admitted ledger before settlement" ); - let now = current_timestamp(); for task in [AgentTaskKind::SessionReflector, AgentTaskKind::SkillWriter] { assert!( - AutomationTaskLock::try_acquire(&cg.store_layout().dashboard_root, task, None, now,) + AutomationTaskLock::try_acquire(&cg.store_layout().dashboard_root, task) .await .unwrap() .is_none(), @@ -336,10 +335,9 @@ async fn retained_combined_review_defers_both_ledgers_and_holds_both_task_locks( drop(settlement_guards); for task in [AgentTaskKind::SessionReflector, AgentTaskKind::SkillWriter] { - let reacquired = - AutomationTaskLock::try_acquire(&cg.store_layout().dashboard_root, task, None, now) - .await - .unwrap(); + let reacquired = AutomationTaskLock::try_acquire(&cg.store_layout().dashboard_root, task) + .await + .unwrap(); assert!( reacquired.is_some(), "{task:?} lock must release with its guard" @@ -554,15 +552,11 @@ async fn combined_review_active_task_locks_skip_before_retrieval_or_backend() { let cg = init_project(temp.path()).await; seed_project_session_activity(&cg).await; let config = scheduler_config(Some(3600), None); - let active_lock = AutomationTaskLock::try_acquire( - &cg.store_layout().dashboard_root, - locked_task, - None, - current_timestamp(), - ) - .await - .unwrap() - .expect("fixture must own the task lock"); + let active_lock = + AutomationTaskLock::try_acquire(&cg.store_layout().dashboard_root, locked_task) + .await + .unwrap() + .expect("fixture must own the task lock"); let backend = CombinedJsonBackend::new(combined_output_fixture()); let retrieval = CountingAutomationSessionRetrieval::new(&cg); diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index bad308cc63..3c693a72df 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -18,7 +18,7 @@ use tracedecay_automation_runtime::automation::jobs::{ run_user_job_with_backend_for_retained_settlement, save_jobs, validate_job, }; use tracedecay_automation_runtime::automation::scheduler::{ - AutomationSchedule, cron_is_due, parse_schedule, + AutomationSchedule, AutomationTaskLock, cron_is_due, parse_schedule, }; use tracedecay_contracts::retained_surfaces::AutomationSkipReasonV1; use tracedecay_domain::errors::TraceDecayError; @@ -1056,9 +1056,10 @@ async fn manual_job_trigger_respects_existing_per_job_lock() { let dashboard_root = temp.path().join("dashboard"); let profile_root = temp.path().join("profile"); fs::create_dir_all(&profile_root).unwrap(); - let lock_dir = dashboard_root.join("automation_locks"); - fs::create_dir_all(&lock_dir).unwrap(); - fs::write(lock_dir.join("user_job_locked-job.lock"), b"9999999999").unwrap(); + let _held = AutomationTaskLock::try_acquire_keyed(&dashboard_root, "user_job_locked-job") + .await + .unwrap() + .expect("fixture owns the per-job lock"); let job = sample_job("locked-job"); let config = enabled_job_config(); @@ -1091,9 +1092,10 @@ async fn disabled_automation_skip_precedes_existing_per_job_lock() { let dashboard_root = temp.path().join("dashboard"); let profile_root = temp.path().join("profile"); fs::create_dir_all(&profile_root).unwrap(); - let lock_dir = dashboard_root.join("automation_locks"); - fs::create_dir_all(&lock_dir).unwrap(); - fs::write(lock_dir.join("user_job_disabled-job.lock"), b"9999999999").unwrap(); + let _held = AutomationTaskLock::try_acquire_keyed(&dashboard_root, "user_job_disabled-job") + .await + .unwrap() + .expect("fixture owns the per-job lock"); let job = sample_job("disabled-job"); let mut config = enabled_job_config(); @@ -1295,22 +1297,11 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .is_none(), "a new interval job is advisory-due before effect admission" ); - let lock_path = dashboard_root - .join("automation_locks") - .join("user_job_prefilter-gap.lock"); - assert!( - !lock_path.exists(), - "the advisory prefilter must release its reservation lock" - ); - fs::write( - &lock_path, - format!( - "pid={}\ncreated_at={}\n", - std::process::id(), - current_timestamp() - ), - ) - .unwrap(); + let _raced_in = + AutomationTaskLock::try_acquire_keyed(&dashboard_root, "user_job_prefilter-gap") + .await + .unwrap() + .expect("the advisory prefilter must release its reservation lock"); let backend = ContentBackend::new("unused"); let retained = run_user_job_with_backend_for_retained_settlement( diff --git a/crates/tracedecay/tests/automation_runner_test/memory_curator.rs b/crates/tracedecay/tests/automation_runner_test/memory_curator.rs index 85443d1886..3bd4094548 100644 --- a/crates/tracedecay/tests/automation_runner_test/memory_curator.rs +++ b/crates/tracedecay/tests/automation_runner_test/memory_curator.rs @@ -101,11 +101,13 @@ async fn memory_curator_empty_store_skips_without_a_backend_attempt_and_releases assert_eq!(run.ledger_record.backend_attempt_count, 0); assert!(run.ledger_record.backend_attempts.is_empty()); assert!( - !cg.store_layout() - .dashboard_root - .join("automation_locks") - .join("memory_curator.lock") - .exists(), + tracedecay_automation_runtime::automation::scheduler::AutomationTaskLock::try_acquire( + &cg.store_layout().dashboard_root, + tracedecay_automation_runtime::automation::backend::AgentTaskKind::MemoryCurator, + ) + .await + .unwrap() + .is_some(), "the empty run must release its scheduler lock" ); } diff --git a/crates/tracedecay/tests/automation_runner_test/runner.rs b/crates/tracedecay/tests/automation_runner_test/runner.rs index 2c46979761..10e22dd5d9 100644 --- a/crates/tracedecay/tests/automation_runner_test/runner.rs +++ b/crates/tracedecay/tests/automation_runner_test/runner.rs @@ -238,8 +238,7 @@ async fn memory_curator_runner_recovers_stale_scheduler_lock_file() { let lock_path = lock_dir.join("memory_curator.lock"); fs::write(&lock_path, "pid=999999\ncreated_at=100\n").unwrap(); let backend = JsonBackend::new(json!({"ops": []})); - let mut config = scheduler_config(None, None); - config.tasks.memory_curator.stale_lock_secs = Some(1); + let config = scheduler_config(None, None); let run = run_memory_curator( &cg, diff --git a/crates/tracedecay/tests/automation_runner_test/scheduler.rs b/crates/tracedecay/tests/automation_runner_test/scheduler.rs index 99b104f86b..8e9018d580 100644 --- a/crates/tracedecay/tests/automation_runner_test/scheduler.rs +++ b/crates/tracedecay/tests/automation_runner_test/scheduler.rs @@ -1315,14 +1315,13 @@ async fn task_lock_reclaims_stale_dead_pid_lock_file() { let dead_pid = reaped_task_lock_child_pid(); std::fs::write(&lock_path, format!("pid={dead_pid}\ncreated_at=100\n")).unwrap(); - let lock = - AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator, Some(10), 200) - .await - .unwrap(); + let lock = AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator) + .await + .unwrap(); assert!(lock.is_some()); + assert!(!lock_path.exists(), "a dead owner's lock file is retired"); drop(lock); - assert!(!lock_path.exists()); } #[cfg(any(unix, windows))] @@ -1339,17 +1338,16 @@ async fn task_lock_reclaims_a_fresh_dead_pid_without_waiting_out_the_age_gate() ) .unwrap(); - let lock = - AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator, Some(10), 200) - .await - .unwrap(); + let lock = AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator) + .await + .unwrap(); assert!( lock.is_some(), "a confirmed-dead owner must not block the next scheduler tick for stale_lock_secs" ); + assert!(!lock_path.exists(), "a dead owner's lock file is retired"); drop(lock); - assert!(!lock_path.exists()); } #[tokio::test] @@ -1364,10 +1362,9 @@ async fn task_lock_keeps_live_pid_lock_file() { ) .unwrap(); - let lock = - AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::SkillWriter, Some(10), 200) - .await - .unwrap(); + let lock = AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::SkillWriter) + .await + .unwrap(); assert!(lock.is_none()); assert!(lock_path.exists()); @@ -1393,14 +1390,9 @@ async fn task_lock_keeps_foreign_live_pid_past_stale_threshold() { ) .unwrap(); - let lock = AutomationTaskLock::try_acquire( - temp.path(), - AgentTaskKind::SessionReflector, - Some(10), - 200, - ) - .await - .unwrap(); + let lock = AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::SessionReflector) + .await + .unwrap(); assert!(lock.is_none()); assert!(lock_path.exists()); @@ -1408,123 +1400,6 @@ async fn task_lock_keeps_foreign_live_pid_past_stale_threshold() { child.wait().unwrap(); } -#[cfg(any(unix, windows))] -#[tokio::test] -async fn old_task_lock_guard_cannot_remove_replacement_token() { - let temp = tempdir().unwrap(); - let lock_path = temp - .path() - .join("automation_locks") - .join("memory_curator.lock"); - let first = - AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator, Some(10), 100) - .await - .unwrap() - .unwrap(); - let first_record = std::fs::read_to_string(&lock_path).unwrap(); - let first_token = task_lock_token(&first_record); - let dead_pid = reaped_task_lock_child_pid(); - std::fs::write( - &lock_path, - format!("pid={dead_pid}\ncreated_at=100\ntoken={first_token}\n"), - ) - .unwrap(); - - let replacement = - AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator, Some(10), 200) - .await - .unwrap() - .unwrap(); - let replacement_record = std::fs::read_to_string(&lock_path).unwrap(); - assert_ne!(task_lock_token(&replacement_record), first_token); - - drop(first); - assert_eq!( - std::fs::read_to_string(&lock_path).unwrap(), - replacement_record, - "the prior guard must not unlink a newer owner's exact token" - ); - let third = - AutomationTaskLock::try_acquire(temp.path(), AgentTaskKind::MemoryCurator, Some(10), 300) - .await - .unwrap(); - assert!(third.is_none()); - - drop(replacement); - assert!(!lock_path.exists()); -} - -#[tokio::test] -async fn cancelled_task_lock_acquisition_cleans_detached_owner() { - use std::future::Future; - - let temp = tempdir().unwrap(); - let root = temp.path().to_path_buf(); - let lock_dir = root.join("automation_locks"); - std::fs::create_dir_all(&lock_dir).unwrap(); - let lock_path = lock_dir.join("skill_writer.lock"); - let coordination_path = tracedecay_runtime_core::storage::append_lock_path(&lock_path); - #[cfg(windows)] - let coordination = { - let file = tracedecay_runtime_core::windows_security::open_or_create_private_lock_file( - &coordination_path, - ) - .unwrap(); - file.lock().unwrap(); - file - }; - #[cfg(not(windows))] - let coordination = - tracedecay_runtime_core::storage::acquire_sidecar_lock_blocking(&coordination_path) - .unwrap(); - let (submitted_tx, submitted_rx) = tokio::sync::oneshot::channel(); - let mut submitted_tx = Some(submitted_tx); - let acquire = tokio::spawn(async move { - let mut future = Box::pin(AutomationTaskLock::try_acquire( - &root, - AgentTaskKind::SkillWriter, - Some(10), - 200, - )); - std::future::poll_fn(move |context| { - let poll = future.as_mut().poll(context); - if poll.is_pending() - && let Some(submitted_tx) = submitted_tx.take() - { - let _ = submitted_tx.send(()); - } - poll - }) - .await - }); - submitted_rx.await.unwrap(); - acquire.abort(); - assert!(acquire.await.unwrap_err().is_cancelled()); - coordination.unlock().unwrap(); - drop(coordination); - - let acquired = tokio::time::timeout(std::time::Duration::from_secs(10), async { - loop { - if let Some(lock) = AutomationTaskLock::try_acquire( - temp.path(), - AgentTaskKind::SkillWriter, - Some(10), - 200, - ) - .await - .unwrap() - { - break lock; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }) - .await - .expect("detached acquisition must release its exact owner token"); - drop(acquired); - assert!(!lock_path.exists()); -} - #[cfg(any(unix, windows))] const TASK_LOCK_LIVENESS_CHILD_ENV: &str = "TRACEDECAY_TASK_LOCK_LIVENESS_CHILD"; #[cfg(any(unix, windows))] @@ -1570,14 +1445,6 @@ fn reaped_task_lock_child_pid() -> u32 { pid } -#[cfg(any(unix, windows))] -fn task_lock_token(record: &str) -> &str { - record - .lines() - .find_map(|line| line.strip_prefix("token=")) - .unwrap() -} - #[cfg(any(unix, windows))] #[test] fn task_lock_liveness_child() { diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs index 97d6af1c22..12a03e3eb6 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/runtime_surface_acceptance.rs @@ -448,6 +448,45 @@ async fn send_lsp(session: &mut DaemonLspSessionClient, value: Value) { ); } +async fn request_lsp_with_retrigger( + session: &mut DaemonLspSessionClient, + request_id: u64, + method: &str, + params: Value, +) -> Value { + // A warming analyzer replies with ServerCancelled and retriggerRequest. + // Respect that client-facing signal after reconnect as well as initial + // startup, while keeping every attempt inside one 90-second budget. + tokio::time::timeout(std::time::Duration::from_secs(90), async { + let mut attempt = 0_u64; + loop { + // The gateway rejects duplicate request ids, including retriggers. + let attempt_id = request_id + attempt * 10_000; + send_lsp( + session, + serde_json::json!({ + "jsonrpc": "2.0", + "id": attempt_id, + "method": method, + "params": params.clone(), + }), + ) + .await; + let response = poll_lsp_response(session, attempt_id).await; + if !response["error"]["data"]["retriggerRequest"] + .as_bool() + .unwrap_or(false) + { + return response; + } + attempt += 1; + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("daemon LSP {method} request {request_id} timed out after 90s")) +} + async fn shutdown_lsp(session: &mut DaemonLspSessionClient, request_id: u64) { let shutdown_request = serde_json::json!({ "jsonrpc": "2.0", @@ -2623,35 +2662,7 @@ async fn production_lsp_negotiates_and_projects_canonical_context() { }), ), ] { - // A warming analyzer is answered with ServerCancelled carrying - // retriggerRequest, which is the gateway's "ask again" signal rather - // than a failure. A compliant client re-sends on it, under one outer - // budget, instead of treating the first not-ready reply as the answer. - // Each attempt needs its own id; the gateway rejects a duplicate. - let retrigger_deadline = std::time::Instant::now() + std::time::Duration::from_secs(90); - let mut attempt = 0_u64; - let navigation = loop { - let attempt_id = request_id + attempt * 10_000; - send_lsp( - &mut session, - serde_json::json!({ - "jsonrpc": "2.0", - "id": attempt_id, - "method": method, - "params": params.clone(), - }), - ) - .await; - let response = poll_lsp_response(&mut session, attempt_id).await; - let retrigger = response["error"]["data"]["retriggerRequest"] - .as_bool() - .unwrap_or(false); - if !retrigger || std::time::Instant::now() >= retrigger_deadline { - break response; - } - attempt += 1; - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - }; + let navigation = request_lsp_with_retrigger(&mut session, request_id, method, params).await; assert!( navigation.get("result").is_some(), "{method} must return a standard LSP result: {navigation}" @@ -3222,17 +3233,13 @@ async fn production_lsp_negotiates_and_projects_canonical_context() { .reconnect(deadline, cancellation) .await .expect("reconnect production LSP session"); - send_lsp( + let after_reconnect = request_lsp_with_retrigger( &mut session, - serde_json::json!({ - "jsonrpc": "2.0", - "id": 701, - "method": "textDocument/documentSymbol", - "params": { "textDocument": { "uri": document_uri } }, - }), + 701, + "textDocument/documentSymbol", + serde_json::json!({ "textDocument": { "uri": document_uri } }), ) .await; - let after_reconnect = poll_lsp_response(&mut session, 701).await; assert!( after_reconnect.get("result").is_some(), "reconnected client must retain standard navigation: {after_reconnect}" diff --git a/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs b/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs index 9fb34b48f3..6ac16dc392 100644 --- a/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs +++ b/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs @@ -17,7 +17,7 @@ use tracedecay_session_temporal_store::{ use std::collections::HashSet; use std::sync::Arc; use std::sync::PoisonError; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::Ordering; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use serde_json::{Value, json}; @@ -1178,7 +1178,9 @@ async fn project_retirement_cancels_and_awaits_an_inflight_projector() { } struct PanicOnceProjector { - panicked: AtomicBool, + calls: std::sync::atomic::AtomicUsize, + retry_started: Arc, + resume_retry: Arc, } impl SessionTemporalRefreshProjector for PanicOnceProjector { @@ -1187,9 +1189,15 @@ impl SessionTemporalRefreshProjector for PanicOnceProjector { _database: &'a RegisteredGlobalDbLeaseV1, recovery: SessionRefreshRecoveryV1, ) -> SessionTemporalRefreshProjectionFuture<'a> { - let should_panic = !self.panicked.swap(true, Ordering::AcqRel); Box::pin(async move { - assert!(!should_panic, "injected refresh worker panic"); + let call = self.calls.fetch_add(1, Ordering::AcqRel); + assert_ne!(call, 0, "injected refresh worker panic"); + // Hold the first recovery invocation for observation; subsequent + // storage retries must not wait for a second test-owned release. + if call == 1 { + self.retry_started.notify_one(); + self.resume_retry.notified().await; + } Ok(empty_projection_effect(&recovery)) }) } @@ -1201,11 +1209,15 @@ async fn worker_recovery_exposes_blocker_and_drains_backlog() { let authority = registered_test_database(&temp, "worker-recovery", HostAdmissionScope::Profile).await; let db = authority.database(); + let retry_started = Arc::new(tokio::sync::Notify::new()); + let resume_retry = Arc::new(tokio::sync::Notify::new()); let mut registry = SessionTemporalRefreshSchedulerRegistry::default(); configure_scheduler( &mut registry, Arc::new(PanicOnceProjector { - panicked: AtomicBool::new(false), + calls: std::sync::atomic::AtomicUsize::new(0), + retry_started: Arc::clone(&retry_started), + resume_retry: Arc::clone(&resume_retry), }), SessionTemporalRefreshPolicy::default(), ); @@ -1215,17 +1227,22 @@ async fn worker_recovery_exposes_blocker_and_drains_backlog() { .await .unwrap(); assert!(wake.wake()); - let recovering = tokio::time::timeout(Duration::from_secs(1), async { - loop { - let status = registry.profile_worker_status(db.db_path()).await; - if status.unavailable_reason == Some(SessionTemporalRefreshUnavailableReason::Stalled) { - break status; - } - tokio::task::yield_now().await; - } - }) + let restarted = tokio::time::timeout(Duration::from_secs(1), retry_started.notified()).await; + // Inspect a held recovery pass, not the short backoff before it starts. + // Replacing a task has not recovered its durable backlog yet. + let recovering = registry.profile_worker_status(db.db_path()).await; + resume_retry.notify_one(); + restarted.expect("the supervisor should restart the panicked worker"); + tokio::time::timeout( + Duration::from_secs(1), + registry.wait_profile_idle(db.db_path()), + ) .await - .expect("worker should expose its stalled pre-progress recovery"); + .expect("the restarted worker should drain its recovered backlog"); + assert_eq!( + recovering.unavailable_reason, + Some(SessionTemporalRefreshUnavailableReason::Stalled) + ); assert_eq!(recovering.backlog, 1); assert_eq!( recovering.blocker, @@ -1235,7 +1252,6 @@ async fn worker_recovery_exposes_blocker_and_drains_backlog() { recovering.retry_class, Some(SessionTemporalRefreshRetryClass::Projector) ); - registry.wait_profile_idle(db.db_path()).await; let store = tracedecay_session_temporal_store::SessionTemporalStore::new(db); assert!(store.running_session_refreshes().await.unwrap().is_empty()); let recovered = registry.profile_worker_status(db.db_path()).await;