diff --git a/crates/tinyflows-module/src/run_manager/actor_recovery.rs b/crates/tinyflows-module/src/run_manager/actor_recovery.rs index 24de6755..c5336a87 100644 --- a/crates/tinyflows-module/src/run_manager/actor_recovery.rs +++ b/crates/tinyflows-module/src/run_manager/actor_recovery.rs @@ -83,9 +83,14 @@ impl Drop for ActorRecoveryGuard { // environment, so even cancellation before its first poll releases // this entry without removing a replacement actor. let active_cleanup = ActiveCleanupGuard(Arc::clone(&self.active)); + #[cfg(test)] + let run_id = self.run_id.clone(); runtime.spawn(async move { let _active_cleanup = active_cleanup; recovery.reconcile().await; + drop(_active_cleanup); + #[cfg(test)] + super::actor_recovery_tests::notify_recovery_finished_for_test(&run_id); }); } } diff --git a/crates/tinyflows-module/src/run_manager_actor_recovery_tests.rs b/crates/tinyflows-module/src/run_manager_actor_recovery_tests.rs index 093b4a9a..cbc1056c 100644 --- a/crates/tinyflows-module/src/run_manager_actor_recovery_tests.rs +++ b/crates/tinyflows-module/src/run_manager_actor_recovery_tests.rs @@ -31,7 +31,12 @@ async fn dropping_an_unpolled_actor_future_reconciles_only_its_live_run() { // Tokio can drop a spawned future before its first poll during shutdown. // The actor factory must have captured recovery ownership before it // returned this future, so the admitted row cannot be left running. + let (reconciled_tx, reconciled_rx) = oneshot::channel(); + install_recovery_completion(&fixture.run_id, reconciled_tx); drop(actor); + reconciled_rx + .await + .expect("actor recovery task must finish"); assert_reconciled( &fixture.catalog, @@ -60,9 +65,14 @@ async fn aborting_an_admitted_actor_reconciles_its_running_row() { )); admitted_tx.send(()).unwrap(); arrived_rx.await.unwrap(); + let (reconciled_tx, reconciled_rx) = oneshot::channel(); + install_recovery_completion(&fixture.run_id, reconciled_tx); task.abort(); assert!(task.await.unwrap_err().is_cancelled()); drop(release_tx); + reconciled_rx + .await + .expect("actor recovery task must finish"); assert_reconciled( &fixture.catalog, @@ -468,6 +478,29 @@ fn install_after_admission_gate( static AFTER_ADMISSION_GATES: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +#[cfg(test)] +static RECOVERY_COMPLETIONS: LazyLock>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +#[cfg(test)] +fn install_recovery_completion(run_id: &str, finished: oneshot::Sender<()>) { + RECOVERY_COMPLETIONS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(run_id.to_owned(), finished); +} + +#[cfg(test)] +pub(super) fn notify_recovery_finished_for_test(run_id: &str) { + if let Some(finished) = RECOVERY_COMPLETIONS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(run_id) + { + let _ = finished.send(()); + } +} + #[cfg(test)] pub(super) async fn pause_after_admission_for_test(run_id: &str) { let gate = AFTER_ADMISSION_GATES diff --git a/crates/tinyflows-schedule/src/runtime/mod.rs b/crates/tinyflows-schedule/src/runtime/mod.rs index 614ee958..95c1fe04 100644 --- a/crates/tinyflows-schedule/src/runtime/mod.rs +++ b/crates/tinyflows-schedule/src/runtime/mod.rs @@ -198,8 +198,9 @@ impl Scheduler { let config = self.config.clone(); let cancellation = self.cancellation.subscribe(); let permits = Arc::clone(&self.permits); + let job_id = claim.job.id.clone(); self.tasks.spawn(async move { - run_claim_with_renewal( + let outcome = run_claim_with_renewal( &ports, &config, claim, @@ -208,8 +209,17 @@ impl Scheduler { Some(permits), cancellation, ) - .await - .map(|_| ()) + .await?; + // The run record holds the result; a failed delivery has no + // other trace in the poll loop. + if !outcome.success { + tracing::warn!( + %job_id, + delivery_status = ?outcome.delivery_status, + "scheduled run did not succeed" + ); + } + Ok(()) }); count += 1; }