From c44aeff670ff74554ddee266b2e9255f7b688d83 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 14 Oct 2026 21:47:11 +0000 Subject: [PATCH 1/4] refactor(daemon): run background owners on the native runtime --- crates/tracedecay/src/daemon/bootstrap.rs | 13 +- crates/tracedecay/src/daemon/branch_admin.rs | 120 ++++++++---- .../daemon/branch_admin/project_retirement.rs | 155 +++++++++++---- .../branch_admin/remote_recovery_lifecycle.rs | 3 +- crates/tracedecay/src/daemon/engine.rs | 4 +- crates/tracedecay/src/daemon/maintenance.rs | 184 +++++++++++++----- .../src/daemon/pr_autotrack/runtime.rs | 36 ++-- .../src/daemon/pr_autotrack/tests.rs | 98 ++++++---- crates/tracedecay/src/daemon/scheduler.rs | 29 ++- crates/tracedecay/src/daemon/tests.rs | 69 ++++++- .../tracedecay/src/daemon/tests/bootstrap.rs | 2 +- .../tracedecay/src/daemon/tests/ownership.rs | 13 +- .../src/daemon/tests/scheduler_config.rs | 107 ++++++++-- .../src/daemon/tests/scheduler_shutdown.rs | 158 +++++++++++++-- crates/tracedecay/src/test_support/mod.rs | 4 + 15 files changed, 763 insertions(+), 232 deletions(-) diff --git a/crates/tracedecay/src/daemon/bootstrap.rs b/crates/tracedecay/src/daemon/bootstrap.rs index 319b3ab594..be9dfa9467 100644 --- a/crates/tracedecay/src/daemon/bootstrap.rs +++ b/crates/tracedecay/src/daemon/bootstrap.rs @@ -713,7 +713,18 @@ async fn run_foreground_unix( let pr_autotrack_task = pr_autotrack::spawn_with_administration( engine.store_administration.clone(), engine.invocation.code_index_schedulers.clone(), - ); + ) + .inspect_err(|error| { + log_daemon_event( + "pr_autotrack", + &[ + ("action", "start".to_string()), + ("outcome", "runtime_unavailable".to_string()), + ("reason", error.clone()), + ], + ); + }) + .ok(); let engine = engine .with_git_watcher(git_watcher) .with_maintenance_coordinator(maintenance) diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs index e6bbbf6874..05fb0ad82b 100644 --- a/crates/tracedecay/src/daemon/branch_admin.rs +++ b/crates/tracedecay/src/daemon/branch_admin.rs @@ -124,9 +124,8 @@ struct MaintenanceReaperKey { #[cfg(unix)] struct MaintenanceReaperHandle { - retired_task: tokio::task::AbortHandle, + retired_task: Arc>, termination: Arc, - _task: tokio::task::JoinHandle<()>, } #[cfg(unix)] @@ -498,7 +497,7 @@ struct ManualBranchPublicationTasks { closed: AtomicBool, join_failed: AtomicBool, cancellation: CancellationToken, - tasks: tokio::sync::Mutex>, + tasks: tokio::sync::Mutex>, } #[cfg(unix)] @@ -508,11 +507,44 @@ impl Default for ManualBranchPublicationTasks { closed: AtomicBool::new(false), join_failed: AtomicBool::new(false), cancellation: CancellationToken::new(), - tasks: tokio::sync::Mutex::new(tokio::task::JoinSet::new()), + tasks: tokio::sync::Mutex::new(Vec::new()), } } } +/// One admitted publication; dropping it aborts the task, as a dropped +/// `JoinSet` did. +#[cfg(unix)] +struct ManualBranchPublication(native_runtime::NativeTask<()>); + +#[cfg(unix)] +impl Drop for ManualBranchPublication { + fn drop(&mut self) { + self.0.abort(); + } +} + +#[cfg(unix)] +impl ManualBranchPublicationTasks { + /// Logs and records a publication that ended without completing. + fn record_join_failure(&self, action: &str, outcome: native_runtime::NativeTaskOutcome<()>) { + let reason = match outcome { + native_runtime::NativeTaskOutcome::Completed(()) => return, + native_runtime::NativeTaskOutcome::Cancelled => "task cancelled".to_owned(), + native_runtime::NativeTaskOutcome::Panicked(panic) => format!("task panicked: {panic}"), + }; + self.join_failed.store(true, Ordering::Release); + log_daemon_event( + "manual_branch_publication", + &[ + ("action", action.to_owned()), + ("outcome", "task_join_failed".to_owned()), + ("reason", reason), + ], + ); + } +} + /// Waitable receipt for the durable account-deletion tombstone persist. /// /// Subscribe before starting deletion. `wait` fails closed if the @@ -645,25 +677,26 @@ impl StoreAdministration { "manual branch publication admission is closed", )); } - while let Some(result) = tasks.try_join_next() { - if let Err(error) = result { + tasks.retain(|task| match task.0.try_finished() { + Some(outcome) => { self.manual_branch_publications - .join_failed - .store(true, Ordering::Release); - log_daemon_event( - "manual_branch_publication", - &[ - ("action", "reap".to_owned()), - ("outcome", "task_join_failed".to_owned()), - ("reason", error.to_string()), - ], - ); + .record_join_failure("reap", outcome); + false } - } + None => true, + }); let cancellation = self.manual_branch_publications.cancellation.clone(); - tasks.spawn(async move { + let task = native_runtime::NativeTask::spawn(async move { let _ = result_sender.send(publication(cancellation).await); - }); + }) + .map_err(|error| { + TraceDecayError::project_route( + "branch_tracking_failed", + true, + format!("manual branch publication could not start: {error}"), + ) + })?; + tasks.push(ManualBranchPublication(task)); } Ok(result_receiver) } @@ -733,24 +766,14 @@ impl StoreAdministration { &self, ) -> std::result::Result<(), String> { self.cancel_manual_branch_publications(); - let mut tasks = { + let tasks = { let mut owned = self.manual_branch_publications.tasks.lock().await; std::mem::take(&mut *owned) }; - while let Some(result) = tasks.join_next().await { - if let Err(error) = result { - self.manual_branch_publications - .join_failed - .store(true, Ordering::Release); - log_daemon_event( - "manual_branch_publication", - &[ - ("action", "shutdown".to_owned()), - ("outcome", "task_join_failed".to_owned()), - ("reason", error.to_string()), - ], - ); - } + for task in tasks { + let outcome = task.0.wait().await; + self.manual_branch_publications + .record_join_failure("shutdown", outcome); } if self .manual_branch_publications @@ -1444,7 +1467,7 @@ impl StoreAdministration { mut reservation: MaintenanceReaperReservation, kind: MaintenanceReaperKind, owner: ProjectServerKey, - task: tokio::task::JoinHandle<()>, + task: native_runtime::NativeTask<()>, termination: Arc, cleanup: F, ) where @@ -1463,31 +1486,44 @@ impl StoreAdministration { barrier.block(); } - let retired_task = task.abort_handle(); task.abort(); - let mut state = self.retirement_reapers.state(); - let key = MaintenanceReaperRegistry::next_key(&mut state, kind, &owner); + let retired_task = Arc::new(task); + let key = + MaintenanceReaperRegistry::next_key(&mut self.retirement_reapers.state(), kind, &owner); let finalizer = MaintenanceReaperFinalizer { registry: Arc::clone(&self.retirement_reapers), key: key.clone(), termination: Arc::clone(&termination), }; let (start, registered) = tokio::sync::oneshot::channel(); - let reaper = tokio::spawn(tracing::Instrument::instrument( + let reaped_task = Arc::clone(&retired_task); + // Spawned before the registry lock: a refused spawn drops the + // finalizer, which takes that lock. + if let Err(error) = native_runtime::NativeTask::spawn(tracing::Instrument::instrument( async move { let _finalizer = finalizer; let _ = registered.await; - let _ = task.await; + let _ = reaped_task.wait().await; cleanup.await; }, tracing::trace_span!("daemon.branch_admin.retirement_reaper"), - )); + )) { + log_daemon_event( + "retirement_reaper", + &[ + ("owner", format!("{:?}", owner.owner.project_id)), + ("outcome", "runtime_unavailable".to_owned()), + ("error", error), + ], + ); + return; + } + let mut state = self.retirement_reapers.state(); let replaced = state.reapers.insert( key, MaintenanceReaperHandle { retired_task, termination, - _task: reaper, }, ); debug_assert!(replaced.is_none()); diff --git a/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs b/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs index 73cf27b2bc..83f6ea19e7 100644 --- a/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs +++ b/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs @@ -1,7 +1,7 @@ use super::{StoreAdministration, StoreOwnerKey}; use std::time::Instant; -use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; +use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome, TimeoutError}; use tracedecay_store_runtime::{ShutdownStatus, ShutdownTaskOutcome, ShutdownTaskReceipt}; /// Drops the process-global Context Scout owner when this project's other @@ -24,7 +24,6 @@ pub(crate) fn retire_registered_context_scout_owner( pub(super) struct ProjectServerRetirement { pub(super) owner: StoreOwnerKey, completion: tokio::sync::watch::Receiver, - _task: tokio::task::JoinHandle<()>, } pub(crate) struct ProjectServerCapacityRetirementCompletion { @@ -95,7 +94,11 @@ impl ProjectServerRetirementAdmission<'_> { Task: std::future::Future + Send + 'static, { let task = tokio::spawn(retirement); - track_project_server_retirement_after_admission(&mut self.retirements, owner, task, false); + track_project_server_retirement_after_admission( + &mut self.retirements, + owner, + tokio_retirement_status(task, false), + ); } /// Track one capacity retirement whose outcome the retiring open awaits. @@ -114,11 +117,10 @@ impl ProjectServerRetirementAdmission<'_> { ProjectServerRetirementStatus::Failed(_) => tracked.owner != owner, ProjectServerRetirementStatus::Pending => true, }); - let (task, completion) = spawn_fallible_retirement(retirement); + let completion = spawn_fallible_retirement(retirement); self.retirements.push(ProjectServerRetirement { owner, completion: completion.clone(), - _task: task, }); ProjectServerCapacityRetirementCompletion { completion } } @@ -126,10 +128,7 @@ impl ProjectServerRetirementAdmission<'_> { fn spawn_fallible_retirement( retirement: Task, -) -> ( - tokio::task::JoinHandle<()>, - tokio::sync::watch::Receiver, -) +) -> tokio::sync::watch::Receiver where Task: std::future::Future> + Send + 'static, { @@ -139,14 +138,14 @@ where completion: task_completion, terminal: false, }; - let task = tokio::spawn(async move { + tokio::spawn(async move { let status = match retirement.await { Ok(()) => ProjectServerRetirementStatus::Clean, Err(error) => ProjectServerRetirementStatus::Failed(error.to_string()), }; finalizer.complete(status); }); - (task, completion) + completion } #[derive(Clone, Debug, Eq, PartialEq)] @@ -202,11 +201,39 @@ async fn wait_for_project_server_retirement( } } +async fn tokio_retirement_status( + task: tokio::task::JoinHandle<()>, + cancelled_is_clean: bool, +) -> ProjectServerRetirementStatus { + match task.await { + Ok(()) => ProjectServerRetirementStatus::Clean, + Err(error) if cancelled_is_clean && error.is_cancelled() => { + ProjectServerRetirementStatus::Clean + } + Err(error) => ProjectServerRetirementStatus::Failed(error.to_string()), + } +} + +/// The receipt of a native owner aborted for retirement: its cancellation is +/// the clean outcome. +async fn aborted_native_retirement_status(task: NativeTask<()>) -> ProjectServerRetirementStatus { + match task.wait().await { + NativeTaskOutcome::Completed(()) | NativeTaskOutcome::Cancelled => { + ProjectServerRetirementStatus::Clean + } + NativeTaskOutcome::Panicked(panic) => { + ProjectServerRetirementStatus::Failed(format!("task panicked: {panic}")) + } + } +} + +/// Records one retirement and publishes `joined`'s status from a native +/// task. When the native runtime refuses that task, its finalizer publishes +/// a failed receipt. fn track_project_server_retirement_after_admission( retirements: &mut Vec, owner: StoreOwnerKey, - task: tokio::task::JoinHandle<()>, - cancelled_is_clean: bool, + joined: impl std::future::Future + Send + 'static, ) { retirements.retain(|retirement| { !matches!( @@ -220,36 +247,20 @@ fn track_project_server_retirement_after_admission( completion: task_completion, terminal: false, }; - let task = tokio::spawn(async move { - let status = match task.await { - Ok(()) => ProjectServerRetirementStatus::Clean, - Err(error) if cancelled_is_clean && error.is_cancelled() => { - ProjectServerRetirementStatus::Clean - } - Err(error) => ProjectServerRetirementStatus::Failed(error.to_string()), - }; - finalizer.complete(status); - }); - retirements.push(ProjectServerRetirement { - owner, - completion, - _task: task, - }); + // Detached: the receipt, not the task, is what callers wait on. + if let Err(error) = NativeTask::spawn(async move { finalizer.complete(joined.await) }) { + tracing::error!(%error, "project server retirement tracker was refused"); + } + retirements.push(ProjectServerRetirement { owner, completion }); } async fn track_project_server_retirement( retirements: &tokio::sync::Mutex>, owner: StoreOwnerKey, - task: tokio::task::JoinHandle<()>, - cancelled_is_clean: bool, + joined: impl std::future::Future + Send + 'static, ) { let mut retirements = retirements.lock().await; - track_project_server_retirement_after_admission( - &mut retirements, - owner, - task, - cancelled_is_clean, - ); + track_project_server_retirement_after_admission(&mut retirements, owner, joined); } pub(super) async fn track_retirement_task( @@ -257,7 +268,7 @@ pub(super) async fn track_retirement_task( owner: StoreOwnerKey, task: tokio::task::JoinHandle<()>, ) { - track_project_server_retirement(retirements, owner, task, false).await; + track_project_server_retirement(retirements, owner, tokio_retirement_status(task, false)).await; } pub(super) async fn track_aborted_retirement_task( @@ -265,7 +276,16 @@ pub(super) async fn track_aborted_retirement_task( owner: StoreOwnerKey, task: tokio::task::JoinHandle<()>, ) { - track_project_server_retirement(retirements, owner, task, true).await; + track_project_server_retirement(retirements, owner, tokio_retirement_status(task, true)).await; +} + +pub(super) async fn track_aborted_native_retirement_task( + retirements: &tokio::sync::Mutex>, + owner: StoreOwnerKey, + task: NativeTask<()>, +) { + track_project_server_retirement(retirements, owner, aborted_native_retirement_status(task)) + .await; } async fn shutdown_receipt( @@ -350,7 +370,12 @@ impl StoreAdministration { owner: StoreOwnerKey, task: tokio::task::JoinHandle<()>, ) { - track_project_server_retirement(&self.project_server_retirements, owner, task, false).await; + track_project_server_retirement( + &self.project_server_retirements, + owner, + tokio_retirement_status(task, false), + ) + .await; } // pub(crate): the test-transport production harness joins retirements from @@ -1049,4 +1074,56 @@ mod tests { "joined retirement must release the exact evicted server" ); } + + /// The tracker of a Tokio-owned retirement runs on the native runtime, so + /// a caller with no Tokio runtime can track and settle it. + #[test] + fn a_tokio_retirement_is_tracked_and_settled_without_a_tokio_runtime() { + let tokio = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .build() + .expect("tokio runtime"); + let (release, released) = tokio::sync::oneshot::channel::<()>(); + let task = tokio.spawn(async move { + let _ = released.await; + }); + let hazards = crate::test_support::native_runtime_hazards(); + let receipt = native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + let administration = StoreAdministration::default(); + let retirements = &administration.project_server_retirements; + track_retirement_task(retirements, owner("project-a"), task).await; + let _ = release.send(()); + settle_project_retirements( + retirements, + std::path::Path::new("/profile"), + "project-a", + Instant::now() + std::time::Duration::from_secs(5), + ) + .await + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + assert_eq!(receipt.status(), ShutdownStatus::Clean); + } + + #[test] + fn an_aborted_native_owner_settles_its_retirement_clean() { + let receipt = native_runtime::block_on(async { + let administration = StoreAdministration::default(); + let retirements = &administration.project_server_retirements; + let task = NativeTask::spawn(std::future::pending::<()>()).expect("native task"); + task.abort(); + track_aborted_native_retirement_task(retirements, owner("project-a"), task).await; + settle_project_retirements( + retirements, + std::path::Path::new("/profile"), + "project-a", + Instant::now() + std::time::Duration::from_secs(5), + ) + .await + }) + .expect("native runtime"); + assert_eq!(receipt.status(), ShutdownStatus::Clean); + } } diff --git a/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs b/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs index e855a17c48..9435c2615d 100644 --- a/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs +++ b/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs @@ -321,7 +321,8 @@ async fn retire_maintenance_tasks( task.abort(); } for (owner, task) in tasks { - super::project_retirement::track_aborted_retirement_task(retirements, owner, task).await; + super::project_retirement::track_aborted_native_retirement_task(retirements, owner, task) + .await; } } diff --git a/crates/tracedecay/src/daemon/engine.rs b/crates/tracedecay/src/daemon/engine.rs index 7c5e41f9cc..6c97fd1d06 100644 --- a/crates/tracedecay/src/daemon/engine.rs +++ b/crates/tracedecay/src/daemon/engine.rs @@ -208,9 +208,9 @@ impl DaemonEngine { pub(super) async fn with_pr_autotrack_task( self, - task: crate::daemon::pr_autotrack::PrAutotrackTask, + task: Option, ) -> Self { - *self.pr_autotrack_task.lock().await = Some(task); + *self.pr_autotrack_task.lock().await = task; self } diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 3268c094f8..75f69e5663 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -4,7 +4,6 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::Mutex; -use tokio::task::JoinHandle; use tracedecay_code_index::parallelism::{ collect_idle_installed_worker_heaps, collect_installed_worker_heaps_when_idle, }; @@ -28,7 +27,7 @@ use super::branch_admin::StoreAdministration; use tracedecay_daemon_service::shutdown::DAEMON_TASK_ABORT_DEADLINE; use tracedecay_domain::process_heap::request_idle_thread_collection_v1; use tracedecay_runtime_core::logging::log_daemon_event; -use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; +use tracedecay_runtime_core::native_runtime::{self, NativeTask, TimeoutError}; const MAINTENANCE_STORE_PAGE_LIMIT: usize = 8; const REGISTERED_PROJECT_PAGE_LIMIT: usize = 64; @@ -69,6 +68,22 @@ async fn registered_project_stores( } } +/// Starts one maintenance owner on the native runtime. A refusal is logged +/// with the owner it leaves unstarted. +fn spawn_maintenance_task( + owner: &'static str, + work: impl std::future::Future + Send + 'static, +) -> Option> { + NativeTask::spawn(work) + .inspect_err(|error| { + log_daemon_event( + "maintenance_task_unavailable", + &[("owner", owner.to_string()), ("reason", error.clone())], + ); + }) + .ok() +} + /// Joins a cancelled maintenance task. A tick notices cancellation only /// between store units, and the unit it is in may be awaiting blocking store /// work that holds a database client. Aborting drops only the awaiting future: @@ -76,14 +91,14 @@ async fn registered_project_stores( /// then leaves the store's WAL untruncated. The task therefore gets the abort /// deadline to finish cooperatively and is aborted only past it. async fn join_cancelled_maintenance_task( - task: Option>, + task: Option>, owner: &'static str, abort_deadline: Duration, ) { - let Some(mut task) = task else { + let Some(task) = task else { return; }; - let reason = match native_runtime::timeout(abort_deadline, &mut task).await { + let reason = match native_runtime::timeout(abort_deadline, task.wait()).await { Ok(_) => return, Err(TimeoutError::Elapsed) => "cooperative_join_exceeded_abort_deadline".to_string(), Err(error @ TimeoutError::TimerUnavailable(_)) => error.to_string(), @@ -97,8 +112,8 @@ async fn join_cancelled_maintenance_task( ], ); task.abort(); - let reason = match native_runtime::timeout(abort_deadline, task).await { - Ok(Ok(()) | Err(_)) => return, + let reason = match native_runtime::timeout(abort_deadline, task.wait()).await { + Ok(_) => return, Err(TimeoutError::Elapsed) => "join_exceeded_abort_deadline".to_string(), Err(error @ TimeoutError::TimerUnavailable(_)) => error.to_string(), }; @@ -363,7 +378,7 @@ pub(super) struct MaintenanceCoordinator { cancellation: tracedecay_runtime_core::cancellation::CancellationToken, background_cpu: Option>, wake: Arc, - task: Arc>>>, + task: Arc>>>, metrics: Arc>, /// Round-robin fairness cursor over mounted stores: the sort key of the /// last store processed. The next tick resumes immediately after it so no @@ -373,7 +388,7 @@ pub(super) struct MaintenanceCoordinator { /// mounted project. `None` keeps the daily cadence retry-eligible. last_branch_gc: Arc>>, /// The resident-memory sampler task; see [`Self::spawn`]. - resident_memory_sampler: Arc>>>, + resident_memory_sampler: Arc>>>, /// Last resident-memory verdict logged, shared by the sampler and the /// retention tick so a sustained over-budget state logs once per /// transition rather than once per sample. @@ -489,13 +504,15 @@ impl MaintenanceCoordinator { ); } let sampler_owner = coordinator.clone(); - let sampler = tokio::spawn(tracing::Instrument::instrument( - async move { - sampler_owner.run_resident_memory_sampler().await; - }, - tracing::trace_span!("daemon.maintenance.resident_memory_sampler"), - )); - *coordinator.resident_memory_sampler.lock().await = Some(sampler); + *coordinator.resident_memory_sampler.lock().await = spawn_maintenance_task( + "resident_memory_sampler", + tracing::Instrument::instrument( + async move { + sampler_owner.run_resident_memory_sampler().await; + }, + tracing::trace_span!("daemon.maintenance.resident_memory_sampler"), + ), + ); if !retention_maintenance_enabled(&retention) { return coordinator; } @@ -505,41 +522,48 @@ impl MaintenanceCoordinator { // letting a day of publications pile up behind the daily cadence. let publication_owner = coordinator.clone(); let mut publications = code_index_schedulers.subscribe_generation_publications(); - tokio::spawn(tracing::Instrument::instrument( - async move { - loop { - tokio::select! { - biased; - () = publication_owner.cancellation.cancelled() => break, - received = publications.recv() => match received { - Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { - publication_owner.request_due(); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - }, + // Detached: it ends with the coordinator's cancellation or the + // publication channel. + drop(spawn_maintenance_task( + "publication_due_requests", + tracing::Instrument::instrument( + async move { + loop { + tokio::select! { + biased; + () = publication_owner.cancellation.cancelled() => break, + received = publications.recv() => match received { + Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + publication_owner.request_due(); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }, + } } - } - }, - tracing::trace_span!("daemon.maintenance.publication_due_requests"), + }, + tracing::trace_span!("daemon.maintenance.publication_due_requests"), + ), )); let task_owner = coordinator.clone(); let interval = Duration::from_secs(retention.interval_hours.max(1).saturating_mul(3_600)); - let handle = tokio::spawn(tracing::Instrument::instrument( - async move { - task_owner - .run( - profile_root, - administration, - code_index_schedulers, - retention, - branch_gc, - interval, - ) - .await; - }, - tracing::trace_span!("daemon.maintenance.retention_loop"), - )); - *coordinator.task.lock().await = Some(handle); + *coordinator.task.lock().await = spawn_maintenance_task( + "retention_tick", + tracing::Instrument::instrument( + async move { + task_owner + .run( + profile_root, + administration, + code_index_schedulers, + retention, + branch_gc, + interval, + ) + .await; + }, + tracing::trace_span!("daemon.maintenance.retention_loop"), + ), + ); coordinator } @@ -1241,7 +1265,7 @@ mod tests { }; use tracedecay_domain::UtcMicros; - use super::{MaintenanceCoordinator, run_resident_memory_sampler_loop}; + use super::{MaintenanceCoordinator, NativeTask, run_resident_memory_sampler_loop}; use tracedecay_maintenance::telemetry::{ StoreTelemetrySamplingRegistry, TableGrowthObservation, compare_table_growth, }; @@ -1663,8 +1687,18 @@ mod tests { /// A registered project nothing has mounted since the daemon started is /// collected by the first full tick and converges on its continuation, /// instead of waiting for a client to mount it. - #[tokio::test] + #[test] + fn a_full_tick_collects_an_unmounted_registered_project_without_a_tokio_runtime() { + let hazards = crate::test_support::native_runtime_hazards(); + tracedecay_runtime_core::native_runtime::block_on( + a_full_tick_collects_an_unmounted_registered_project(), + ) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + } + async fn a_full_tick_collects_an_unmounted_registered_project() { + assert!(tokio::runtime::Handle::try_current().is_err()); let temp = tempfile::TempDir::new().expect("temp root"); let profile_root = temp.path().join("profile"); let checkout = temp.path().join("checkout"); @@ -1752,11 +1786,12 @@ mod tests { let held_state = Arc::clone(&tick_state); let started = Arc::new(Notify::new()); let task_started = Arc::clone(&started); - let handle = tokio::spawn(async move { + let handle = NativeTask::spawn(async move { task_started.notify_one(); std::future::pending::<()>().await; drop(held_state); - }); + }) + .expect("native task"); started.notified().await; super::join_cancelled_maintenance_task( @@ -1784,14 +1819,15 @@ mod tests { let blocking_client = Arc::clone(&store_client); let started = Arc::new(Notify::new()); let task_started = Arc::clone(&started); - let handle = tokio::spawn(async move { - let _ = tokio::task::spawn_blocking(move || { + let handle = NativeTask::spawn(async move { + let _ = tracedecay_runtime_core::native_runtime::spawn_blocking(move || { task_started.notify_one(); std::thread::sleep(Duration::from_millis(300)); drop(blocking_client); }) .await; - }); + }) + .expect("native task"); *coordinator.task.lock().await = Some(handle); started.notified().await; @@ -1803,4 +1839,44 @@ mod tests { "the maintenance join returned while the tick's blocking store work held its client" ); } + + /// The coordinator starts its sampler, publication and retention owners + /// on the native runtime and joins them at shutdown, with no Tokio + /// runtime entered. + #[test] + fn the_coordinator_runs_and_shuts_down_without_a_tokio_runtime() { + let hazards = crate::test_support::native_runtime_hazards(); + tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + let temp = tempfile::TempDir::new().expect("temp root"); + let retention = tracedecay_configuration::RetentionConfig { + orphan_store_gc_days: Some(30), + ..tracedecay_configuration::RetentionConfig::default() + }; + assert!(super::retention_maintenance_enabled(&retention)); + let coordinator = MaintenanceCoordinator::spawn( + temp.path().join("profile"), + super::StoreAdministration::default(), + tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1::new(1), + retention, + super::BranchStoreGcCadenceV1 { branch_gc_days: 30 }, + ) + .await; + assert!(coordinator.task.lock().await.is_some(), "the retention owner started"); + assert!( + coordinator.resident_memory_sampler.lock().await.is_some(), + "the sampler started" + ); + tracedecay_runtime_core::native_runtime::timeout( + Duration::from_secs(10), + coordinator.shutdown(), + ) + .await + .expect("shutdown joins every maintenance owner"); + assert!(coordinator.task.lock().await.is_none()); + assert!(coordinator.resident_memory_sampler.lock().await.is_none()); + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + } } diff --git a/crates/tracedecay/src/daemon/pr_autotrack/runtime.rs b/crates/tracedecay/src/daemon/pr_autotrack/runtime.rs index 62943302eb..41c35af77c 100644 --- a/crates/tracedecay/src/daemon/pr_autotrack/runtime.rs +++ b/crates/tracedecay/src/daemon/pr_autotrack/runtime.rs @@ -25,7 +25,7 @@ const BASE_TICK: Duration = Duration::from_mins(1); /// before joining the task. pub(crate) struct PrAutotrackTask { cancellation: CancellationToken, - task: tokio::task::JoinHandle<()>, + task: native_runtime::NativeTask<()>, } impl PrAutotrackTask { @@ -33,18 +33,21 @@ impl PrAutotrackTask { self.cancellation.clone() } - pub(crate) async fn shutdown(mut self) { + pub(crate) async fn shutdown(self) { self.cancellation.cancel(); - if let Err(error) = (&mut self.task).await { - log_daemon_event( - "pr_autotrack", - &[ - ("action", "shutdown".to_string()), - ("outcome", "task_join_failed".to_string()), - ("reason", error.to_string()), - ], - ); - } + let reason = match self.task.wait().await { + native_runtime::NativeTaskOutcome::Completed(()) => return, + native_runtime::NativeTaskOutcome::Cancelled => "task cancelled".to_string(), + native_runtime::NativeTaskOutcome::Panicked(panic) => panic.to_string(), + }; + log_daemon_event( + "pr_autotrack", + &[ + ("action", "shutdown".to_string()), + ("outcome", "task_join_failed".to_string()), + ("reason", reason), + ], + ); } } @@ -56,19 +59,20 @@ impl Drop for PrAutotrackTask { /// Spawns the PR-autotrack poll loop with the daemon's shared store coordinator /// and the retained code-index scheduler used for PR-head worktree activation. +/// The loop runs on the native runtime; the error is that runtime's refusal. pub(crate) fn spawn_with_administration( administration: StoreAdministration, schedulers: CodeIndexSchedulerRegistryV1, -) -> PrAutotrackTask { +) -> Result { let cancellation = CancellationToken::new(); let task_cancellation = cancellation.clone(); - let task = tokio::spawn(tracing::Instrument::instrument( + let task = native_runtime::NativeTask::spawn(tracing::Instrument::instrument( async move { run(administration, schedulers, task_cancellation).await; }, tracing::trace_span!("daemon.pr_autotrack.loop"), - )); - PrAutotrackTask { cancellation, task } + ))?; + Ok(PrAutotrackTask { cancellation, task }) } async fn run( diff --git a/crates/tracedecay/src/daemon/pr_autotrack/tests.rs b/crates/tracedecay/src/daemon/pr_autotrack/tests.rs index e748970d27..cdf283c9e7 100644 --- a/crates/tracedecay/src/daemon/pr_autotrack/tests.rs +++ b/crates/tracedecay/src/daemon/pr_autotrack/tests.rs @@ -3,17 +3,41 @@ use std::time::Duration; use super::*; use crate::test_support::git::git; +use tracedecay_runtime_core::native_runtime; -#[tokio::test] -async fn spawned_loop_is_cancellable_and_joinable() { - let schedulers = CodeIndexSchedulerRegistryV1::new(1); - let task = spawn_with_administration(StoreAdministration::default(), schedulers); +#[test] +fn spawned_loop_is_cancellable_and_joinable_without_a_tokio_runtime() { + let hazards = crate::test_support::native_runtime_hazards(); + native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + let schedulers = CodeIndexSchedulerRegistryV1::new(1); + let task = spawn_with_administration(StoreAdministration::default(), schedulers) + .expect("the native runtime admits the loop"); - assert!( - tokio::time::timeout(Duration::from_secs(1), task.shutdown()) - .await - .is_ok() - ); + assert!( + native_runtime::timeout(Duration::from_secs(1), task.shutdown()) + .await + .is_ok() + ); + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); +} + +/// The tick's reconcile mounts the PR head through the code-index scheduler +/// with no Tokio runtime entered. +#[test] +fn reconcile_activates_a_pr_head_without_a_tokio_runtime() { + let repo = tempfile::tempdir().unwrap(); + let origin = tempfile::tempdir().unwrap(); + init_repo_with_pr_11_head(repo.path(), origin.path()); + let hazards = crate::test_support::native_runtime_hazards(); + native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + assert_pr_11_head_reconciles(repo.path()).await; + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); } // ---- Reconcile: removal + idempotency (no index required) ------------------- @@ -334,48 +358,52 @@ async fn reconcile_does_not_prepare_new_pr_without_scheduler_activation() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn reconcile_activates_discovered_pr_head_when_scheduler_is_injected() { - use tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1; - let repo = tempfile::tempdir().unwrap(); let origin = tempfile::tempdir().unwrap(); - git(repo.path(), &["init", "-q", "-b", "main"]); - git(repo.path(), &["config", "user.name", "TraceDecay Test"]); + init_repo_with_pr_11_head(repo.path(), origin.path()); + assert_pr_11_head_reconciles(repo.path()).await; +} + +/// A repository whose origin holds `refs/pull/11/head`, with the PR branch +/// deleted locally. +fn init_repo_with_pr_11_head(repo: &Path, origin: &Path) { + git(repo, &["init", "-q", "-b", "main"]); + git(repo, &["config", "user.name", "TraceDecay Test"]); git( - repo.path(), + repo, &["config", "user.email", "tracedecay@example.invalid"], ); - std::fs::create_dir_all(repo.path().join("src")).unwrap(); - std::fs::write(repo.path().join("src/lib.rs"), "pub fn on_main() {}\n").unwrap(); - git(repo.path(), &["add", "."]); - git(repo.path(), &["commit", "-qm", "initial"]); - git(origin.path(), &["init", "-q", "--bare", "-b", "main"]); - git( - repo.path(), - &["remote", "add", "origin", origin.path().to_str().unwrap()], - ); - git(repo.path(), &["push", "-q", "origin", "main"]); - git(repo.path(), &["checkout", "-q", "-b", "feature-11", "main"]); - std::fs::write(repo.path().join("src/pr_11.rs"), "pub fn pr_eleven() {}\n").unwrap(); - git(repo.path(), &["add", "."]); - git(repo.path(), &["commit", "-qm", "PR 11 content"]); - git(repo.path(), &["push", "-q", "origin", "feature-11"]); + std::fs::create_dir_all(repo.join("src")).unwrap(); + std::fs::write(repo.join("src/lib.rs"), "pub fn on_main() {}\n").unwrap(); + git(repo, &["add", "."]); + git(repo, &["commit", "-qm", "initial"]); + git(origin, &["init", "-q", "--bare", "-b", "main"]); + git(repo, &["remote", "add", "origin", origin.to_str().unwrap()]); + git(repo, &["push", "-q", "origin", "main"]); + git(repo, &["checkout", "-q", "-b", "feature-11", "main"]); + std::fs::write(repo.join("src/pr_11.rs"), "pub fn pr_eleven() {}\n").unwrap(); + git(repo, &["add", "."]); + git(repo, &["commit", "-qm", "PR 11 content"]); + git(repo, &["push", "-q", "origin", "feature-11"]); git( - origin.path(), + origin, &["update-ref", "refs/pull/11/head", "refs/heads/feature-11"], ); - git(repo.path(), &["checkout", "-q", "main"]); - git(repo.path(), &["branch", "-q", "-D", "feature-11"]); + git(repo, &["checkout", "-q", "main"]); + git(repo, &["branch", "-q", "-D", "feature-11"]); +} +async fn assert_pr_11_head_reconciles(repo: &Path) { let graph = Arc::new( tracedecay_project::project::TraceDecay::open_with_options_for_test( - repo.path(), + repo, tracedecay_project::project::TraceDecayOpenOptions::default(), ) .await .expect("open project graph"), ); let data_root = graph.store_layout().data_root.clone(); - let discovery = discover_open_prs_with_control(repo.path(), default_pr_command_control()) + let discovery = discover_open_prs_with_control(repo, default_pr_command_control()) .expect("discover PR head"); assert_eq!(discovery.open.len(), 1); assert_eq!(discovery.open[0].number, 11); @@ -383,7 +411,7 @@ async fn reconcile_activates_discovered_pr_head_when_scheduler_is_injected() { let schedulers = CodeIndexSchedulerRegistryV1::new(2); let command_control = PrCommandControl::default(); let report = reconcile_project_with_administration( - repo.path(), + repo, &data_root, &discovery, 10, diff --git a/crates/tracedecay/src/daemon/scheduler.rs b/crates/tracedecay/src/daemon/scheduler.rs index 291b52a3dc..07245b2555 100644 --- a/crates/tracedecay/src/daemon/scheduler.rs +++ b/crates/tracedecay/src/daemon/scheduler.rs @@ -2,11 +2,11 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; -use tokio::task::JoinHandle; use tracedecay_automation_runtime::automation::AutomationRunControl; use tracedecay_automation_runtime::automation::backend::{AgentTaskBackend, AgentTaskKind}; use tracedecay_automation_runtime::automation::maintenance_termination::MaintenanceTaskTermination; use tracedecay_automation_runtime::automation::scheduler_stop::AutomationSchedulerStop; +use tracedecay_runtime_core::native_runtime::NativeTask; use tracedecay_automation_runtime::automation::effect_runtime::settlement::{ AutomationEffectAdmission, AutomationEffectAuthority, RetainedAutomationSettlementOutcome, @@ -261,7 +261,7 @@ pub(super) enum AutomationSchedulerLifecycle { } pub(super) struct AutomationSchedulerHandle { - pub(super) task: Option>, + pub(super) task: Option>, pub(super) wake: Arc, pub(super) completion: Arc<()>, pub(super) generation: Arc, @@ -272,7 +272,7 @@ pub(super) struct AutomationSchedulerHandle { #[cfg(test)] impl AutomationSchedulerHandle { - pub(super) fn for_test(task: JoinHandle<()>) -> Self { + pub(super) fn for_test(task: NativeTask<()>) -> Self { Self { task: Some(task), wake: Arc::new(tokio::sync::Notify::new()), @@ -541,7 +541,7 @@ impl DaemonEngine { if let Some(mut handle) = finished && let Some(task) = handle.task.take() { - let _ = task.await; + let _ = task.wait().await; } #[cfg(test)] @@ -635,7 +635,7 @@ impl DaemonEngine { if let Some(mut handle) = finished && let Some(task) = handle.task.take() { - let _ = task.await; + let _ = task.wait().await; } debug_assert!(matches!( lifecycle, @@ -771,10 +771,23 @@ impl DaemonEngine { || handle.lifecycle == AutomationSchedulerLifecycle::Retiring }); }; - let task = tokio::spawn(tracing::Instrument::instrument( + let task = match NativeTask::spawn(tracing::Instrument::instrument( scheduler_loop, tracing::trace_span!("daemon.scheduler.loop"), - )); + )) { + Ok(task) => task, + Err(error) => { + log_daemon_event( + "scheduler_start", + &[ + ("project", key.project_root.display().to_string()), + ("outcome", "runtime_unavailable".to_string()), + ("error", error), + ], + ); + return AutomationSchedulerReconcileOutcome::OwnerUnavailable; + } + }; { let mut signals = self .store_administration @@ -1062,7 +1075,7 @@ pub(super) fn same_scheduler_owner(left: &ProjectServerKey, right: &ProjectServe fn observed_scheduler_lifecycle( handle: &AutomationSchedulerHandle, ) -> AutomationSchedulerLifecycle { - if handle.task.as_ref().is_some_and(JoinHandle::is_finished) { + if handle.task.as_ref().is_some_and(NativeTask::is_finished) { AutomationSchedulerLifecycle::Finished } else { handle.lifecycle diff --git a/crates/tracedecay/src/daemon/tests.rs b/crates/tracedecay/src/daemon/tests.rs index e6df9a309d..58afdacae2 100644 --- a/crates/tracedecay/src/daemon/tests.rs +++ b/crates/tracedecay/src/daemon/tests.rs @@ -11,6 +11,8 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; #[cfg(unix)] use tokio::task::JoinHandle; use tracedecay_query::code_search; +#[cfg(unix)] +use tracedecay_runtime_core::native_runtime::NativeTask; use tracedecay_tool_catalog::ApplicationSurfaceOperation; #[cfg(unix)] @@ -518,10 +520,35 @@ fn an_unservable_search_reports_every_lane_down() { } #[cfg(unix)] -fn test_automation_scheduler_handle(task: JoinHandle<()>) -> AutomationSchedulerHandle { +fn test_automation_scheduler_handle(task: NativeTask<()>) -> AutomationSchedulerHandle { AutomationSchedulerHandle::for_test(task) } +/// Reports whether a native test task's future has finished or been dropped. +#[cfg(unix)] +struct NativeTaskProbe(std::sync::Weak<()>); + +#[cfg(unix)] +impl NativeTaskProbe { + fn is_finished(&self) -> bool { + self.0.strong_count() == 0 + } +} + +#[cfg(unix)] +fn probed_native_test_task( + work: impl std::future::Future + Send + 'static, +) -> (NativeTask<()>, NativeTaskProbe) { + let alive = Arc::new(()); + let probe = NativeTaskProbe(Arc::downgrade(&alive)); + let task = NativeTask::spawn(async move { + let _alive = alive; + work.await; + }) + .expect("native test task"); + (task, probe) +} + #[cfg(unix)] async fn wait_for_automation_scheduler_state( engine: &DaemonEngine, @@ -555,7 +582,7 @@ async fn wait_for_automation_scheduler_state( #[cfg(unix)] async fn wait_for_finished_task( - task: &JoinHandle<()>, + task: &NativeTask<()>, deadline: tokio::time::Instant, description: &str, ) { @@ -598,6 +625,44 @@ impl Drop for NoncooperativeTaskRelease { } } +/// A native owner stuck in synchronous work until released: an abort cannot +/// drop it before it yields. It runs on its own one-worker runtime so the +/// blocked worker is never one the process runtime needs. +#[cfg(unix)] +fn spawn_noncooperative_native_test_task() -> ( + NativeTask<()>, + NativeTaskProbe, + tokio::sync::oneshot::Receiver<()>, + tokio::sync::oneshot::Receiver<()>, + NoncooperativeTaskRelease, +) { + let release = NoncooperativeTaskRelease { + state: Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new())), + }; + let task_release = release.clone(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (completed_tx, completed_rx) = tokio::sync::oneshot::channel(); + let runtime: &'static tracedecay_runtime_core::native_runtime::Runtime = Box::leak(Box::new( + tracedecay_runtime_core::native_runtime::RuntimeBuilder::multi_thread() + .worker_threads(1) + .build() + .expect("dedicated native runtime"), + )); + let (task, probe) = runtime.block_on(async move { + probed_native_test_task(async move { + let _ = started_tx.send(()); + let (released, changed) = &*task_release.state; + let mut ready = released.lock().unwrap_or_else(PoisonError::into_inner); + while !*ready { + ready = changed.wait(ready).unwrap_or_else(PoisonError::into_inner); + } + drop(ready); + let _ = completed_tx.send(()); + }) + }); + (task, probe, started_rx, completed_rx, release) +} + #[cfg(unix)] fn spawn_noncooperative_test_task() -> ( JoinHandle<()>, diff --git a/crates/tracedecay/src/daemon/tests/bootstrap.rs b/crates/tracedecay/src/daemon/tests/bootstrap.rs index 533a5aa03d..44515e9706 100644 --- a/crates/tracedecay/src/daemon/tests/bootstrap.rs +++ b/crates/tracedecay/src/daemon/tests/bootstrap.rs @@ -1030,7 +1030,7 @@ async fn remote_project_deletion_stays_settling_until_transferred_reaper_joins() project_root: project.canonicalize().expect("canonical project"), scope_prefix: None, }; - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, _, started_rx, completed_rx, release) = spawn_noncooperative_native_test_task(); started_rx.await.expect("maintenance task started"); engine .store_administration diff --git a/crates/tracedecay/src/daemon/tests/ownership.rs b/crates/tracedecay/src/daemon/tests/ownership.rs index eb345feacc..1ee2b92cca 100644 --- a/crates/tracedecay/src/daemon/tests/ownership.rs +++ b/crates/tracedecay/src/daemon/tests/ownership.rs @@ -1325,8 +1325,7 @@ async fn project_rekey_awaits_stale_automation_owner_before_replacement() { }; let mut new = old.clone(); new.owner.graph_db_path = PathBuf::from("/store/new.db"); - let task = tokio::spawn(std::future::pending::<()>()); - let stale_task = task.abort_handle(); + let (task, stale_task) = probed_native_test_task(std::future::pending::<()>()); engine .store_administration .automation_schedulers() @@ -1366,11 +1365,11 @@ async fn shutdown_waits_for_blocked_automation_retirement_reaper_and_is_idempote project_root: PathBuf::from("/project"), scope_prefix: None, }; - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, stale_task, started_rx, completed_rx, release) = + spawn_noncooperative_native_test_task(); started_rx .await .expect("noncooperative automation owner started"); - let stale_task = task.abort_handle(); engine .store_administration .automation_schedulers() @@ -1454,9 +1453,9 @@ async fn automation_retirement_timeout_retains_owner_tombstone_until_join_finish }; let mut new = old.clone(); new.owner.graph_db_path = PathBuf::from("/store/new.db"); - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, stale_task, started_rx, completed_rx, release) = + spawn_noncooperative_native_test_task(); started_rx.await.expect("noncooperative owner started"); - let stale_task = task.abort_handle(); engine .store_administration .automation_schedulers() @@ -1595,7 +1594,7 @@ async fn released_automation_tombstone_allows_one_eventual_replacement() { let mut old = new.clone(); old.owner.graph_db_path = old.owner.graph_db_path.with_extension("retiring.db"); - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, _, started_rx, completed_rx, release) = spawn_noncooperative_native_test_task(); tokio::time::timeout(std::time::Duration::from_secs(2), started_rx) .await .expect("noncooperative owner start timed out") diff --git a/crates/tracedecay/src/daemon/tests/scheduler_config.rs b/crates/tracedecay/src/daemon/tests/scheduler_config.rs index a79c8faea8..bb53398507 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_config.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_config.rs @@ -311,7 +311,7 @@ async fn disabled_finished_scheduler_reenables_with_a_fresh_owner() { let server = save_scheduled_automation(&engine, &handshake, false).await; let cg = server.cg().await; let key = ProjectServerKey::from_open_project(&cg, &handshake).expect("owner key"); - let finished = tokio::spawn(async {}); + let finished = NativeTask::spawn(async {}).expect("native task"); wait_for_finished_task( &finished, tokio::time::Instant::now() + tokio::time::Duration::from_secs(2), @@ -371,9 +371,10 @@ async fn concurrent_reenable_creates_one_live_scheduler_owner() { let cg = server.cg().await; let key = ProjectServerKey::from_open_project(&cg, &handshake).expect("owner key"); let (finished_tx, finished_rx) = tokio::sync::oneshot::channel(); - let finished = tokio::spawn(async move { + let finished = NativeTask::spawn(async move { let _ = finished_tx.send(()); - }); + }) + .expect("native task"); tokio::time::timeout(tokio::time::Duration::from_secs(2), finished_rx) .await .expect("finished owner barrier timed out") @@ -750,7 +751,7 @@ async fn cached_project_reconciles_cli_enabled_automation_without_cache_probe() ) .await; - let finished = tokio::spawn(async {}); + let finished = NativeTask::spawn(async {}).expect("native task"); wait_for_finished_task( &finished, tokio::time::Instant::now() + tokio::time::Duration::from_secs(2), @@ -1004,20 +1005,33 @@ async fn cancel_stops_the_automation_loop_while_the_scheduler_map_is_held() { .automation_schedulers() .lock() .await; - let automation_loop = schedulers - .get(&key) - .and_then(|handle| handle.task.as_ref()) - .expect("started automation loop") - .abort_handle(); + assert!( + schedulers + .get(&key) + .and_then(|handle| handle.task.as_ref()) + .is_some(), + "started automation loop" + ); engine.cancel_automation_schedulers(); tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; drop(schedulers); + // An ended loop removes its own registration. let mut waited = tokio::time::Duration::ZERO; - while !automation_loop.is_finished() && waited < tokio::time::Duration::from_secs(2) { + let stopped_before_join = loop { + let stopped = engine + .store_administration + .automation_schedulers() + .lock() + .await + .get(&key) + .and_then(|handle| handle.task.as_ref()) + .is_none_or(NativeTask::is_finished); + if stopped || waited >= tokio::time::Duration::from_secs(2) { + break stopped; + } tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; waited += tokio::time::Duration::from_millis(10); - } - let stopped_before_join = automation_loop.is_finished(); + }; engine.shutdown_all().await; assert!( @@ -1025,3 +1039,72 @@ async fn cancel_stops_the_automation_loop_while_the_scheduler_map_is_held() { "a cancel issued while the scheduler map was held must still end the loop" ); } + +/// A reconcile from a caller with no Tokio runtime starts the automation +/// loop, and the loop reads its disabled configuration and commits its exit +/// on the native runtime. +#[cfg(unix)] +#[test] +fn the_automation_loop_starts_and_exits_without_a_tokio_runtime() { + let tokio = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("tokio runtime"); + let dir = TempDir::new().expect("temp dir"); + let project = dir.path().join("project"); + let client_identity = test_client_identity_for(dir.path().join("profile")); + std::fs::create_dir_all(project.join("src")).expect("src dir"); + std::fs::write(project.join("src/main.rs"), "fn main() {}\n").expect("source file"); + tokio.block_on(initialize_test_project(&project, &client_identity)); + let handshake = DaemonHandshake { + project_path: Some(project.clone()), + client_identity, + ..test_handshake_defaults() + }; + let engine = test_daemon_engine_for_profile(&handshake.client_identity.profile_root); + let _database_scope = enter_test_daemon_database_scope( + &handshake.client_identity.profile_root, + "scheduler-off-tokio-test", + ); + let server = tokio.block_on(save_scheduled_automation(&engine, &handshake, false)); + let cg = tokio.block_on(server.cg()); + let key = ProjectServerKey::from_open_project(&cg, &handshake).expect("owner key"); + let barrier = Arc::new(AutomationSchedulerExitBarrier::new()); + let _barrier_release = AutomationExitBarrierRelease(Arc::clone(&barrier)); + tokio.block_on(async { + *engine.automation_scheduler_exit_barrier.lock().await = Some(Arc::clone(&barrier)); + }); + engine + .automation_configured_override + .store(true, std::sync::atomic::Ordering::Relaxed); + + let hazards = crate::test_support::native_runtime_hazards(); + let (outcome, decision) = tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + let outcome = engine + .ensure_automation_scheduler(key.clone(), project.clone(), handshake.clone()) + .await; + tracedecay_runtime_core::native_runtime::timeout( + std::time::Duration::from_secs(20), + barrier.wait_until_reached(), + ) + .await + .expect("the loop read its disabled configuration"); + barrier.release(); + let decision = tracedecay_runtime_core::native_runtime::timeout( + std::time::Duration::from_secs(20), + barrier.wait_for_decision(), + ) + .await + .expect("the loop decided whether to exit"); + (outcome, decision) + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + assert_eq!( + outcome, + tracedecay_contracts::retrieval::AutomationSchedulerReconcileOutcome::Started + ); + assert_eq!(decision, AutomationSchedulerExitBarrier::EXIT); + tokio.block_on(engine.shutdown_all()); +} diff --git a/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs b/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs index 899a6e2144..0a42180c3d 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs @@ -11,9 +11,10 @@ async fn pr_autotrack_is_cancelled_before_invocation_join() { let task = crate::daemon::pr_autotrack::spawn_with_administration( engine.store_administration.clone(), tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1::new(1), - ); + ) + .expect("the native runtime admits the loop"); let cancellation = task.cancellation(); - let engine = engine.with_pr_autotrack_task(task).await; + let engine = engine.with_pr_autotrack_task(Some(task)).await; let owner_phases = engine.shutdown_owner_phases().await; assert!(!cancellation.is_cancelled()); @@ -166,7 +167,7 @@ async fn daemon_scheduler_shutdown_aborts_and_joins_every_loop() { project_root: PathBuf::from("/projects/shutdown-test"), scope_prefix: None, }; - let task = tokio::spawn(std::future::pending::<()>()); + let task = NativeTask::spawn(std::future::pending::<()>()).expect("native task"); engine .store_administration .automation_schedulers() @@ -208,11 +209,11 @@ async fn automation_shutdown_timeout_keeps_unfinished_task_tracked() { project_root: PathBuf::from("/projects/automation-shutdown-timeout-test"), scope_prefix: None, }; - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, stale_task, started_rx, completed_rx, release) = + spawn_noncooperative_native_test_task(); started_rx .await .expect("noncooperative automation owner started"); - let stale_task = task.abort_handle(); engine .store_administration .automation_schedulers() @@ -324,7 +325,7 @@ async fn automation_retirement_reserves_only_after_scheduler_lock_admission() { project_root: PathBuf::from("/projects/retirement-lock-admission"), scope_prefix: None, }; - let task = tokio::spawn(std::future::pending::<()>()); + let task = NativeTask::spawn(std::future::pending::<()>()).expect("native task"); engine .store_administration .automation_schedulers() @@ -383,7 +384,7 @@ async fn cancelled_contended_automation_retirement_remains_shutdown_owned() { let mut replacement = key.clone(); replacement.owner.graph_db_path = PathBuf::from("/stores/automation-registration-cancel-test/new.db"); - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, _, started_rx, completed_rx, release) = spawn_noncooperative_native_test_task(); tokio::time::timeout(MAINTENANCE_TEST_DEADLINE, started_rx) .await .expect("noncooperative automation owner start timed out") @@ -536,7 +537,7 @@ async fn deletion_drain_retains_already_transferred_maintenance_reaper_until_ret project_root: PathBuf::from("/projects/deletion-reaper-test"), scope_prefix: None, }; - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, _, started_rx, completed_rx, release) = spawn_noncooperative_native_test_task(); started_rx.await.expect("maintenance owner started"); engine .store_administration @@ -589,7 +590,7 @@ async fn project_reaper_settle_ignores_unrelated_pending_registration() { project_root: PathBuf::from("/projects/unrelated-pending-reaper"), scope_prefix: None, }; - let (task, started_rx, completed_rx, release) = spawn_noncooperative_test_task(); + let (task, _, started_rx, completed_rx, release) = spawn_noncooperative_native_test_task(); started_rx .await .expect("unrelated maintenance owner started"); @@ -654,9 +655,10 @@ async fn panicked_retired_task_releases_scheduler_registration() { project_root: PathBuf::from("/projects/panicked-automation-retirement-test"), scope_prefix: None, }; - let automation_task = tokio::spawn(async { + let automation_task = NativeTask::spawn(async { panic!("panicked automation owner"); - }); + }) + .expect("native task"); engine .store_administration .automation_schedulers() @@ -713,7 +715,9 @@ async fn scheduler_shutdown_does_not_wait_for_contended_administration_gate() { .await .insert( key, - test_automation_scheduler_handle(tokio::spawn(std::future::pending::<()>())), + test_automation_scheduler_handle( + NativeTask::spawn(std::future::pending::<()>()).expect("native task"), + ), ); let (gate_entered_tx, gate_entered_rx) = tokio::sync::oneshot::channel(); let (gate_release_tx, gate_release_rx) = tokio::sync::oneshot::channel(); @@ -852,3 +856,133 @@ async fn engine_shutdown_with_code_index_reconcile_in_flight_stays_inside_term_g receipt.background.unfinished() ); } + +/// Retirement hands the aborted loop to a reaper on the native runtime, which +/// joins it and releases its registration with no Tokio runtime entered. +#[cfg(unix)] +#[test] +fn automation_retirement_reaps_without_a_tokio_runtime() { + let hazards = crate::test_support::native_runtime_hazards(); + tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + let engine = DaemonEngine::default(); + let key = ProjectServerKey { + owner: StoreOwnerKey { + profile_root: PathBuf::from("/profiles/native-reaper-test"), + global_db_path: PathBuf::from("/profiles/native-reaper-test/global.db"), + project_id: Some("native-reaper-test".to_string()), + store_root: PathBuf::from("/stores/native-reaper-test"), + graph_db_path: PathBuf::from("/stores/native-reaper-test/graph.db"), + }, + project_root: PathBuf::from("/projects/native-reaper-test"), + scope_prefix: None, + }; + let (task, stale_task) = probed_native_test_task(std::future::pending::<()>()); + engine + .store_administration + .automation_schedulers() + .lock() + .await + .insert(key.clone(), test_automation_scheduler_handle(task)); + + let retirement = engine + .retire_automation_scheduler_locked(&key) + .await + .expect("automation owner retirement"); + tracedecay_runtime_core::native_runtime::timeout( + MAINTENANCE_TEST_DEADLINE, + retirement.wait(), + ) + .await + .expect("the reaper joins the retired loop"); + tracedecay_runtime_core::native_runtime::timeout( + MAINTENANCE_TEST_DEADLINE, + engine + .store_administration + .wait_for_retirement_reaper_count_for_test(0), + ) + .await + .expect("the reaper releases its registration"); + + assert!(stale_task.is_finished(), "retirement aborts the loop"); + assert!( + engine + .store_administration + .automation_schedulers() + .lock() + .await + .is_empty() + ); + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); +} + +/// Branch add admits its publication onto the native runtime and shutdown +/// joins it, with no Tokio runtime entered. +#[cfg(unix)] +#[test] +fn manual_branch_publication_runs_and_joins_without_a_tokio_runtime() { + let hazards = crate::test_support::native_runtime_hazards(); + tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + let administration = DaemonEngine::default().store_administration.clone(); + let (release_sender, release_receiver) = tokio::sync::oneshot::channel(); + let admission = administration + .admit_manual_branch_publication(|_, admitted| async move { + let _ = admitted.send(()); + let _ = release_receiver.await; + Ok(tracedecay_runtime_core::branch::BranchAddOutcome::Added) + }) + .await + .expect("manual branch publication is admitted"); + assert_eq!( + admission, + tracedecay_runtime_core::branch::BranchAddOutcome::Deferred + ); + let _ = release_sender.send(()); + assert_eq!( + tracedecay_runtime_core::native_runtime::timeout( + MAINTENANCE_TEST_DEADLINE, + administration.shutdown_manual_branch_publications(), + ) + .await, + Ok(Ok(())), + "shutdown joins the completed publication" + ); + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); +} + +/// Dropping the administration's last handle aborts a publication it still +/// owns, as the `JoinSet` it replaced did. +#[cfg(unix)] +#[tokio::test] +async fn dropping_the_administration_aborts_an_owned_manual_branch_publication() { + let administration = DaemonEngine::default().store_administration.clone(); + let alive = Arc::new(()); + let held = Arc::downgrade(&alive); + let (started_sender, started_receiver) = tokio::sync::oneshot::channel(); + administration + .admit_manual_branch_publication(|_, admitted| async move { + let _alive = alive; + let _ = admitted.send(()); + let _ = started_sender.send(()); + std::future::pending::<()>().await; + Ok(tracedecay_runtime_core::branch::BranchAddOutcome::Added) + }) + .await + .expect("manual branch publication is admitted"); + started_receiver.await.expect("publication started"); + assert_eq!(held.strong_count(), 1); + + drop(administration); + tokio::time::timeout(MAINTENANCE_TEST_DEADLINE, async { + while held.strong_count() > 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the dropped owner aborts its publication"); +} diff --git a/crates/tracedecay/src/test_support/mod.rs b/crates/tracedecay/src/test_support/mod.rs index 3047ba4acc..4cf1d7bf92 100644 --- a/crates/tracedecay/src/test_support/mod.rs +++ b/crates/tracedecay/src/test_support/mod.rs @@ -24,6 +24,10 @@ pub(crate) fn native_runtime_hazards() -> u64 { "daemon-service.observability.mount_drain", "daemon-service.observability.settle_drain", "daemon-service.observability.producer_drop", + "daemon-service.work_attempt.spawn_for_worktree", + "daemon-service.host_admission.bootstrap", + "daemon-service.shutdown.runner", + "daemon-service.shutdown_coordination.runner", ] .into_iter() .map(tracedecay_runtime_core::native_runtime::tokio_runtime_missing_count) From 66d4273a55c015a68adfcc55675d08826ce54e08 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 14 Oct 2026 22:14:05 +0000 Subject: [PATCH 2/4] test(daemon): prove background owners run without Tokio --- .../daemon/branch_admin/project_retirement.rs | 18 +++----------- crates/tracedecay/src/daemon/tests.rs | 2 +- .../src/daemon/tests/scheduler_config.rs | 24 ++++++++++++------- .../src/daemon/tests/scheduler_shutdown.rs | 2 +- 4 files changed, 21 insertions(+), 25 deletions(-) diff --git a/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs b/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs index 83f6ea19e7..4d4f28025d 100644 --- a/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs +++ b/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs @@ -97,7 +97,7 @@ impl ProjectServerRetirementAdmission<'_> { track_project_server_retirement_after_admission( &mut self.retirements, owner, - tokio_retirement_status(task, false), + tokio_retirement_status(task), ); } @@ -203,13 +203,9 @@ async fn wait_for_project_server_retirement( async fn tokio_retirement_status( task: tokio::task::JoinHandle<()>, - cancelled_is_clean: bool, ) -> ProjectServerRetirementStatus { match task.await { Ok(()) => ProjectServerRetirementStatus::Clean, - Err(error) if cancelled_is_clean && error.is_cancelled() => { - ProjectServerRetirementStatus::Clean - } Err(error) => ProjectServerRetirementStatus::Failed(error.to_string()), } } @@ -268,15 +264,7 @@ pub(super) async fn track_retirement_task( owner: StoreOwnerKey, task: tokio::task::JoinHandle<()>, ) { - track_project_server_retirement(retirements, owner, tokio_retirement_status(task, false)).await; -} - -pub(super) async fn track_aborted_retirement_task( - retirements: &tokio::sync::Mutex>, - owner: StoreOwnerKey, - task: tokio::task::JoinHandle<()>, -) { - track_project_server_retirement(retirements, owner, tokio_retirement_status(task, true)).await; + track_project_server_retirement(retirements, owner, tokio_retirement_status(task)).await; } pub(super) async fn track_aborted_native_retirement_task( @@ -373,7 +361,7 @@ impl StoreAdministration { track_project_server_retirement( &self.project_server_retirements, owner, - tokio_retirement_status(task, false), + tokio_retirement_status(task), ) .await; } diff --git a/crates/tracedecay/src/daemon/tests.rs b/crates/tracedecay/src/daemon/tests.rs index 58afdacae2..6ddcd04158 100644 --- a/crates/tracedecay/src/daemon/tests.rs +++ b/crates/tracedecay/src/daemon/tests.rs @@ -542,8 +542,8 @@ fn probed_native_test_task( let alive = Arc::new(()); let probe = NativeTaskProbe(Arc::downgrade(&alive)); let task = NativeTask::spawn(async move { - let _alive = alive; work.await; + drop(alive); }) .expect("native test task"); (task, probe) diff --git a/crates/tracedecay/src/daemon/tests/scheduler_config.rs b/crates/tracedecay/src/daemon/tests/scheduler_config.rs index bb53398507..ac70cb4188 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_config.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_config.rs @@ -1071,7 +1071,16 @@ fn the_automation_loop_starts_and_exits_without_a_tokio_runtime() { let key = ProjectServerKey::from_open_project(&cg, &handshake).expect("owner key"); let barrier = Arc::new(AutomationSchedulerExitBarrier::new()); let _barrier_release = AutomationExitBarrierRelease(Arc::clone(&barrier)); + // Project open started its own loop; retire it so the reconcile below + // starts the one under test. tokio.block_on(async { + if let Some(retirement) = engine.retire_automation_scheduler_locked(&key).await { + retirement.wait().await; + } + engine + .store_administration + .wait_for_retirement_reaper_count_for_test(0) + .await; *engine.automation_scheduler_exit_barrier.lock().await = Some(Arc::clone(&barrier)); }); engine @@ -1079,11 +1088,15 @@ fn the_automation_loop_starts_and_exits_without_a_tokio_runtime() { .store(true, std::sync::atomic::Ordering::Relaxed); let hazards = crate::test_support::native_runtime_hazards(); - let (outcome, decision) = tracedecay_runtime_core::native_runtime::block_on(async { + let decision = tracedecay_runtime_core::native_runtime::block_on(async { assert!(tokio::runtime::Handle::try_current().is_err()); let outcome = engine .ensure_automation_scheduler(key.clone(), project.clone(), handshake.clone()) .await; + assert_eq!( + outcome, + tracedecay_contracts::retrieval::AutomationSchedulerReconcileOutcome::Started + ); tracedecay_runtime_core::native_runtime::timeout( std::time::Duration::from_secs(20), barrier.wait_until_reached(), @@ -1091,20 +1104,15 @@ fn the_automation_loop_starts_and_exits_without_a_tokio_runtime() { .await .expect("the loop read its disabled configuration"); barrier.release(); - let decision = tracedecay_runtime_core::native_runtime::timeout( + tracedecay_runtime_core::native_runtime::timeout( std::time::Duration::from_secs(20), barrier.wait_for_decision(), ) .await - .expect("the loop decided whether to exit"); - (outcome, decision) + .expect("the loop decided whether to exit") }) .expect("native runtime"); assert_eq!(crate::test_support::native_runtime_hazards(), hazards); - assert_eq!( - outcome, - tracedecay_contracts::retrieval::AutomationSchedulerReconcileOutcome::Started - ); assert_eq!(decision, AutomationSchedulerExitBarrier::EXIT); tokio.block_on(engine.shutdown_all()); } diff --git a/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs b/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs index 0a42180c3d..f9d95bd673 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs @@ -966,10 +966,10 @@ async fn dropping_the_administration_aborts_an_owned_manual_branch_publication() let (started_sender, started_receiver) = tokio::sync::oneshot::channel(); administration .admit_manual_branch_publication(|_, admitted| async move { - let _alive = alive; let _ = admitted.send(()); let _ = started_sender.send(()); std::future::pending::<()>().await; + drop(alive); Ok(tracedecay_runtime_core::branch::BranchAddOutcome::Added) }) .await From f9f5212b2e97a39388898f879177fd4dea6727ce Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 14 Oct 2026 22:57:05 +0000 Subject: [PATCH 3/4] test(daemon): prove the PR autotrack owner aborts on drop --- .../src/daemon/tests/scheduler_shutdown.rs | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs b/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs index f9d95bd673..54d3351642 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_shutdown.rs @@ -986,3 +986,34 @@ async fn dropping_the_administration_aborts_an_owned_manual_branch_publication() .await .expect("the dropped owner aborts its publication"); } + +/// Dropping the PR auto-track owner aborts its loop, as a timed-out shutdown +/// join does. The parked loop holds one administration clone until then. +#[cfg(unix)] +#[tokio::test] +async fn dropping_the_pr_autotrack_owner_aborts_its_parked_loop() { + let dir = TempDir::new().expect("temp dir"); + let profile_root = dir.path().join("profile"); + let engine = test_daemon_engine_for_profile(&profile_root); + let _database_scope = enter_test_daemon_database_scope(&profile_root, "pr-autotrack-drop-test"); + let administration = engine.store_administration.clone(); + let owners = Arc::strong_count(administration.automation_schedulers()); + let task = crate::daemon::pr_autotrack::spawn_with_administration( + administration.clone(), + tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1::new(1), + ) + .expect("the native runtime admits the loop"); + assert_eq!( + Arc::strong_count(administration.automation_schedulers()), + owners + 1 + ); + + drop(task); + tokio::time::timeout(MAINTENANCE_TEST_DEADLINE, async { + while Arc::strong_count(administration.automation_schedulers()) > owners { + tokio::task::yield_now().await; + } + }) + .await + .expect("the dropped owner aborts its loop"); +} From f7ad6ee2ed0528288e1caa9e0eb60ee5fd003eab Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 14 Oct 2026 23:46:55 +0000 Subject: [PATCH 4/4] test(daemon): wait on the cancelled loop's end, not a poll --- .../src/daemon/tests/scheduler_config.rs | 40 +++++++++---------- 1 file changed, 19 insertions(+), 21 deletions(-) diff --git a/crates/tracedecay/src/daemon/tests/scheduler_config.rs b/crates/tracedecay/src/daemon/tests/scheduler_config.rs index ac70cb4188..b0b6bf5c56 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_config.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_config.rs @@ -1000,38 +1000,36 @@ async fn cancel_stops_the_automation_loop_while_the_scheduler_map_is_held() { .ensure_automation_scheduler(key.clone(), project, handshake) .await; - let schedulers = engine + let mut schedulers = engine .store_administration .automation_schedulers() .lock() .await; - assert!( - schedulers - .get(&key) - .and_then(|handle| handle.task.as_ref()) - .is_some(), - "started automation loop" - ); + let automation_loop = schedulers + .get_mut(&key) + .and_then(|handle| handle.task.take()) + .expect("started automation loop"); engine.cancel_automation_schedulers(); tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; drop(schedulers); - // An ended loop removes its own registration. - let mut waited = tokio::time::Duration::ZERO; - let stopped_before_join = loop { - let stopped = engine + // The loop's tick is 60 s: only the cancel can end it inside this bound. + let stopped_before_join = tracedecay_runtime_core::native_runtime::timeout( + std::time::Duration::from_secs(20), + automation_loop.wait(), + ) + .await + .is_ok(); + if !stopped_before_join + && let Some(handle) = engine .store_administration .automation_schedulers() .lock() .await - .get(&key) - .and_then(|handle| handle.task.as_ref()) - .is_none_or(NativeTask::is_finished); - if stopped || waited >= tokio::time::Duration::from_secs(2) { - break stopped; - } - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - waited += tokio::time::Duration::from_millis(10); - }; + .get_mut(&key) + { + // Hand the live loop back so shutdown can abort it. + handle.task = Some(automation_loop); + } engine.shutdown_all().await; assert!(