diff --git a/CHANGELOG.md b/CHANGELOG.md index 676edf3b..257c6a53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- Tests: planted-bug follow-ups. New scenario-fuzzer oracles for restart + decisions, RestForOne cohorts, Transient restarts, required restarts versus + budget exhaustion, backoff delays in virtual time, linked exit reasons, + `unlink` and cancelled timers; bounded waits so tests that used to hang under + a bug fail with a named assertion (`within!`); a Kani layer in the + planted-bug runner; the planted-bug matrix in `docs/TESTING.md` re-run with + them. `concurrent_failures_coalesce_into_one_round` no longer flakes on one + CPU (it read `current()` after the first crash's round had begun). - ROADMAP: a ranked determinism backlog (D1-D10) under priority tier 3, from an audit of what escapes the simulator in Spawned, Ethrex and Ethlambda ([docs/research/determinism-audit.md](docs/research/determinism-audit.md)). diff --git a/concurrency/tests/backend_experiments.rs b/concurrency/tests/backend_experiments.rs index 068dfb14..1da83eef 100644 --- a/concurrency/tests/backend_experiments.rs +++ b/concurrency/tests/backend_experiments.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] #![cfg(feature = "backend-experiments")] //! Run each experiment mode in a separate process; never mutate env inside tests. +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, tasks::{Actor, ActorStart, Backend, Context, Handler}, @@ -96,7 +99,7 @@ fn experiment_io_reactor() { let actor = Worker.start_with_backend(Backend::Thread); assert_eq!(actor.request(Network).await.unwrap(), 42); actor.child_handle().stop(); - actor.join().await; + within!(actor.join()).await; }); } diff --git a/concurrency/tests/backend_resource_limits.rs b/concurrency/tests/backend_resource_limits.rs index 6df9f59f..73690a58 100644 --- a/concurrency/tests/backend_resource_limits.rs +++ b/concurrency/tests/backend_resource_limits.rs @@ -2,6 +2,9 @@ //! Deterministic characterization of executor capacity; no machine-wide tuning. // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, tasks::{Actor, ActorStart, Backend, Context, Handler}, @@ -211,7 +214,7 @@ fn descriptor_limit_child() { cleanup.0.push(recovered.child_handle()); recovered.request(Ping).await.unwrap(); recovered.child_handle().stop(); - recovered.join().await; + within!(recovered.join()).await; } }); if mode.starts_with("sharded") { diff --git a/concurrency/tests/canary_feedback.rs b/concurrency/tests/canary_feedback.rs index 20f58821..91aeb89c 100644 --- a/concurrency/tests/canary_feedback.rs +++ b/concurrency/tests/canary_feedback.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, supervision::{ @@ -54,7 +57,7 @@ async fn unfinished_report_identifies_child_and_active_turn() { Ok::<_, String>(Worker) })) .unwrap(); - let _supervisor = builder.start().await.unwrap(); + let _supervisor = within!(builder.start()).await.unwrap(); let actor = child.current().unwrap(); let (entered, arrival) = oneshot::channel(); let (release, gate) = oneshot::channel(); @@ -105,7 +108,7 @@ async fn completed_turn_is_not_reported_while_helper_cleanup_remains() { let system = ActorSystem::new().unwrap(); let group = system.async_group("canary-helper", 1).unwrap(); let actor = Worker.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); actor.context().register_name("canary-standalone").unwrap(); let (entered, arrival) = oneshot::channel(); let (release, gate) = std::sync::mpsc::channel(); @@ -141,7 +144,7 @@ async fn completed_turn_is_not_reported_while_helper_cleanup_remains() { .iter() .any(|e| e.id == actor.id() && e.reason == ExitReason::Shutdown)); release.send(()).unwrap(); - helper.join().await.unwrap(); + within!(helper.join()).await.unwrap(); assert!(report.wait(LIMIT).await.complete); } @@ -208,7 +211,7 @@ async fn supervision_logs_restart_budget_exhaustion_and_give_up() { .restart(RestartMode::Transient), ) .unwrap(); - let supervisor = builder.start().await.unwrap(); + let supervisor = within!(builder.start()).await.unwrap(); child .current() .unwrap() @@ -236,7 +239,7 @@ async fn supervision_logs_restart_budget_exhaustion_and_give_up() { .restart(RestartMode::Temporary), ) .unwrap(); - assert!(failing.start().await.is_err()); + assert!(within!(failing.start()).await.is_err()); // Not simulated: the cohort case blocks a spawn_blocking helper. #[cfg(not(spawned_sim))] { @@ -284,7 +287,7 @@ async fn supervision_logs_restart_budget_exhaustion_and_give_up() { Ok::<_, String>(Worker) })) .unwrap(); - let cohort = cohort.start().await.unwrap(); + let cohort = within!(cohort.start()).await.unwrap(); origin .current() .unwrap() @@ -408,7 +411,7 @@ async fn panicked_turn_is_cleared_before_stopped_hook() { } .start_on(&group) .unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); assert!(actor.request(Quit(ExitReason::Kill)).await.is_err()); spawned_rt::tasks::timeout(LIMIT, arrival) .await @@ -456,9 +459,9 @@ async fn trapped_exit_diagnostics_use_the_public_signal_name() { } .start_on(&group) .unwrap(); - observer.ready().await.unwrap(); + within!(observer.ready()).await.unwrap(); let source = Worker.start_on(&group).unwrap(); - source.ready().await.unwrap(); + within!(source.ready()).await.unwrap(); observer.context().link(&source.child_handle()); source .child_handle() diff --git a/concurrency/tests/cancellation_safety.rs b/concurrency/tests/cancellation_safety.rs index 20311750..fb37f519 100644 --- a/concurrency/tests/cancellation_safety.rs +++ b/concurrency/tests/cancellation_safety.rs @@ -6,6 +6,9 @@ //! deadline. See docs/TESTING.md for the per-operation contract. #![cfg(feature = "async")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, protocol, registry, @@ -671,7 +674,7 @@ async fn supervisor_shutdown_and_child_waits_are_cancel_safe() { async move { Ok::<_, String>(probe) } })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let first = child.current().unwrap(); let finished = match variant { 0 => { @@ -688,7 +691,7 @@ async fn supervisor_shutdown_and_child_waits_are_cancel_safe() { spawned_rt::tasks::timeout(LIMIT, supervisor.join()) .await .unwrap_or_else(|_| panic!("{op}: N={n} tree wedged")); - assert_eq!(first.wait_exit().await, ExitReason::Shutdown); + assert_eq!(within!(first.wait_exit()).await, ExitReason::Shutdown); outcome.finished() } 1 => { @@ -745,7 +748,7 @@ async fn system_shutdown_is_cancel_safe() { let mut probe = fx.probe(); probe.name = Some(name.clone()); let actor = fx.spawn(probe); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let helper_gate = Arc::new(Notify::new()); actor.request(Cmd::Helper(helper_gate)).await.unwrap(); let system = fx.system.as_ref().unwrap(); @@ -760,7 +763,7 @@ async fn system_shutdown_is_cancel_safe() { if n > 0 { assert!(report.complete, "{op}: N={n}"); assert!(fx.group.as_ref().unwrap().is_closed()); - assert_eq!(actor.wait_exit().await, ExitReason::Shutdown); + assert_eq!(within!(actor.wait_exit()).await, ExitReason::Shutdown); assert!(!registry::registered().contains(&name), "{op}: leaked name"); } fx.finish(&[&actor]).await; @@ -781,7 +784,7 @@ async fn system_shutdown_is_cancel_safe() { let mut probe = fx.probe(); probe.stop_gate = Some(gate.clone()); let actor = fx.spawn(probe); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let report = fx.system.as_ref().unwrap().shutdown(Duration::ZERO).await; assert!(!report.complete); let outcome = attempt(report.wait(LIMIT), n, PARK).await; @@ -814,7 +817,7 @@ async fn helper_join_is_cancel_safe() { async move { let fx = Fx::new(place); let actor = fx.spawn(fx.probe()); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let gate = Arc::new(Notify::new()); let waiting = gate.clone(); let task = actor diff --git a/concurrency/tests/cfg_gated_protocol.rs b/concurrency/tests/cfg_gated_protocol.rs index e1045d25..7bf8afc5 100644 --- a/concurrency/tests/cfg_gated_protocol.rs +++ b/concurrency/tests/cfg_gated_protocol.rs @@ -2,6 +2,9 @@ //! Integration test: verifies that #[cfg]-gated protocol methods compile and work //! correctly, including blanket impls and type-erased references. +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::tasks::{Actor, ActorStart as _, Context, Handler}; use spawned_concurrency::{protocol, Response}; use spawned_rt::tasks as rt; @@ -59,7 +62,7 @@ fn cfg_gated_method_works_via_blanket_impl() { let configured = actor.with_request_timeout(std::time::Duration::from_secs(1)); assert_eq!(configured.test_only_increment().await.unwrap(), 12); actor.context().stop(); - actor.join().await; + within!(actor.join()).await; }); } @@ -81,6 +84,6 @@ fn cfg_gated_method_works_via_protocol_ref() { .to_mixed_ref(); assert_eq!(configured.test_only_increment().await.unwrap(), 2); actor.context().stop(); - actor.join().await; + within!(actor.join()).await; }); } diff --git a/concurrency/tests/cohort_trace.rs b/concurrency/tests/cohort_trace.rs index 5dc9aba1..99657cd8 100644 --- a/concurrency/tests/cohort_trace.rs +++ b/concurrency/tests/cohort_trace.rs @@ -1,4 +1,7 @@ #![cfg(feature = "lifecycle-trace")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ lifecycle_trace, message::Message, @@ -105,7 +108,7 @@ async fn trace_cohort_selection_and_temporary_retirement() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[1].current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, children[1].wait_for_generation(1)) .await @@ -169,7 +172,7 @@ async fn trace_superseded_round_with_parked_temporary_tail() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[1].current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, entered.notified()) .await @@ -231,7 +234,7 @@ async fn trace_normally_retired_transient_is_skipped_by_later_cohorts() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); children[2].current().unwrap().context().stop(); spawned_rt::tasks::timeout(LIMIT, async { while children[2].snapshot().state @@ -310,7 +313,7 @@ async fn trace_failed_replacement_starts_a_new_round() { .restart(RestartMode::Transient), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(b.current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, b.wait_for_generation(2)) .await diff --git a/concurrency/tests/coverage_trace.rs b/concurrency/tests/coverage_trace.rs index 3042ee2d..d22a32ae 100644 --- a/concurrency/tests/coverage_trace.rs +++ b/concurrency/tests/coverage_trace.rs @@ -11,6 +11,9 @@ //! holds a `FailScenario`, which serializes them. #![cfg(all(feature = "lifecycle-trace", feature = "failpoints"))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ failpoints::{self, Action, FailPoint, FailScenario}, lifecycle_trace, @@ -216,7 +219,7 @@ async fn restart_in_construction_when_shutdown_begins_and_deadline_expires() { .shutdown_timeout(Duration::from_millis(100)), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); crash(&restarting).await; until("the replacement under construction", || { let s = restarting.snapshot(); @@ -362,7 +365,7 @@ async fn nested_leaf_restarts_inside_its_subtree() { }, )) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let leaf = leaf.lock().unwrap().take().unwrap(); crash(&leaf).await; tokio::time::timeout(LIMIT, leaf.wait_for_generation(1)) diff --git a/concurrency/tests/execution_groups.rs b/concurrency/tests/execution_groups.rs index 5ab8ac0f..294671e5 100644 --- a/concurrency/tests/execution_groups.rs +++ b/concurrency/tests/execution_groups.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, system::{ActorSystem, SystemError}, @@ -43,8 +46,8 @@ async fn owned_groups_communicate_and_close_with_stale_handles() { let b = system.async_group("general", 2).unwrap(); let first = Worker.start_on(&a).unwrap(); let second = Worker.start_on(&b).unwrap(); - first.ready().await.unwrap(); - second.ready().await.unwrap(); + within!(first.ready()).await.unwrap(); + within!(second.ready()).await.unwrap(); assert!(first .request(Ping) .await @@ -137,7 +140,7 @@ async fn expired_shutdown_retains_closed_runtime_until_startup_cleanup_finishes( assert!(Worker.start_on(&group).is_err()); release.send(()).unwrap(); assert!(report.wait(LIMIT).await.complete); - assert!(actor.ready().await.is_err()); + assert!(within!(actor.ready()).await.is_err()); assert_eq!(stopped.load(Ordering::SeqCst), 1); } @@ -179,7 +182,7 @@ async fn shutdown_joins_helper_drops_and_generation_owned_registration() { .start_on(&group) .unwrap(); rx.await.unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); assert!(spawned_concurrency::registry::whereis::< spawned_concurrency::tasks::ActorRef, >("execution-group-owned-helper-test") @@ -201,7 +204,7 @@ async fn dropping_system_requests_shutdown_without_an_ambient_runtime() { let group = system.async_group("drop", 1).unwrap(); assert!(system.async_group("drop", 2).is_err()); let actor = Worker.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); std::thread::spawn(move || drop(system)).join().unwrap(); assert_eq!( spawned_rt::tasks::timeout(LIMIT, actor.child_handle().wait_exit_async()) @@ -218,7 +221,7 @@ async fn registered_blocking_work_keeps_runtime_owned_until_native_completion() let system = ActorSystem::new().unwrap(); let group = system.async_group("blocking-helper", 1).unwrap(); let actor = Worker.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let (entered_tx, entered) = tokio::sync::oneshot::channel(); let (release, release_rx) = std::sync::mpsc::channel(); let dropped = Arc::new(AtomicUsize::new(0)); @@ -252,7 +255,7 @@ async fn scoped_helpers_report_panics_and_framework_abort_drops_payload() { let system = ActorSystem::new().unwrap(); let group = system.async_group("helper-errors", 1).unwrap(); let actor = Worker.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let async_error = actor .context() .spawn_helper(async { panic!("async helper panic") }) @@ -293,7 +296,7 @@ async fn ambient_actor_started_inside_group_is_unmanaged_and_aborted_at_teardown let system = ActorSystem::new().unwrap(); let group = system.async_group("ambient-child", 1).unwrap(); let managed = Worker.start_on(&group).unwrap(); - managed.ready().await.unwrap(); + within!(managed.ready()).await.unwrap(); let ambient = managed .context() .spawn_helper(async { Worker.start() }) @@ -303,7 +306,7 @@ async fn ambient_actor_started_inside_group_is_unmanaged_and_aborted_at_teardown .unwrap(); assert!(ambient.context().execution_group().is_none()); assert!(matches!( - ambient.ready().await, + within!(ambient.ready()).await, Err(SystemError::UnmanagedActor) )); assert!(ambient.context().spawn_helper(async {}).is_err()); @@ -313,8 +316,8 @@ async fn ambient_actor_started_inside_group_is_unmanaged_and_aborted_at_teardown .unwrap() .starts_with("spawned-ambient-child")); assert!(system.shutdown(LIMIT).await.complete); - assert_eq!(ambient.wait_exit().await, ExitReason::Kill); - assert_eq!(managed.wait_exit().await, ExitReason::Shutdown); + assert_eq!(within!(ambient.wait_exit()).await, ExitReason::Kill); + assert_eq!(within!(managed.wait_exit()).await, ExitReason::Shutdown); } struct Queued(Probe); @@ -392,7 +395,7 @@ async fn old_scope_cleanup_cannot_remove_a_replacement_registration() { let system = ActorSystem::new().unwrap(); let group = system.async_group("registry-replacement", 1).unwrap(); let actor = Worker.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let name = "execution-groups-stale-registration"; actor.context().register_name(name).unwrap(); assert!(matches!( @@ -436,7 +439,7 @@ async fn generated_protocols_work_unchanged_with_explicit_group_placement() { let system = ActorSystem::new().unwrap(); let group = system.async_group("generated-protocol", 1).unwrap(); let actor = ManagedProtocolActor(42).start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); assert_eq!(actor.value().await.unwrap(), 42); assert!(system.shutdown(LIMIT).await.complete); assert!(actor.value().await.is_err()); @@ -447,7 +450,7 @@ async fn stuck_standalone_cleanup_does_not_leave_earlier_actor_running_forever() let system = ActorSystem::new().unwrap(); let group = system.async_group("stuck-standalone-order", 1).unwrap(); let first = Worker.start_on(&group).unwrap(); - first.ready().await.unwrap(); + within!(first.ready()).await.unwrap(); let (entered_tx, entered) = tokio::sync::oneshot::channel(); let (release, release_rx) = tokio::sync::oneshot::channel(); struct Stuck { @@ -466,7 +469,7 @@ async fn stuck_standalone_cleanup_does_not_leave_earlier_actor_running_forever() } .start_on(&group) .unwrap(); - last.ready().await.unwrap(); + within!(last.ready()).await.unwrap(); let report = system.shutdown(Duration::from_millis(20)).await; assert!(!report.complete); spawned_rt::tasks::timeout(LIMIT, entered) @@ -514,7 +517,10 @@ async fn readiness_reports_actual_panic_after_startup_stop_instead_of_early_norm ready, Err(SystemError::StartupFailed(ExitReason::Panic(_))) )); - assert!(matches!(actor.wait_exit().await, ExitReason::Panic(_))); + assert!(matches!( + within!(actor.wait_exit()).await, + ExitReason::Panic(_) + )); } } assert!(system.shutdown(LIMIT).await.complete); @@ -538,7 +544,7 @@ async fn concurrent_zero_deadline_reports_include_scopes_during_shutdown_initial } .start_on(&group) .unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let mut reporters = Vec::new(); for _ in 0..64 { let owner = system.clone(); @@ -581,7 +587,7 @@ async fn ordered_shutdown_preserves_ordinary_calls_but_closes_new_helper_admissi let system = ActorSystem::new().unwrap(); let group = system.async_group("shutdown-admission", 1).unwrap(); let earlier = HelperAware.start_on(&group).unwrap(); - earlier.ready().await.unwrap(); + within!(earlier.ready()).await.unwrap(); let observed = Arc::new(AtomicUsize::new(0)); Dependent { earlier, @@ -610,7 +616,7 @@ async fn aborted_scoped_helper_joins_as_cancelled() { let system = ActorSystem::new().unwrap(); let group = system.async_group("helper-abort", 1).unwrap(); let actor = Worker.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let helper = actor .context() .spawn_helper(std::future::pending::<()>()) diff --git a/concurrency/tests/execution_profiles.rs b/concurrency/tests/execution_profiles.rs index fac66149..819f9201 100644 --- a/concurrency/tests/execution_profiles.rs +++ b/concurrency/tests/execution_profiles.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, supervision::{Backoff, ChildSpec, SupervisorConfig, SupervisorStrategy}, @@ -65,9 +68,9 @@ async fn native_context_placement_uses_a_free_slot_and_validates_helper_budgets( }) ); let actor = Probe.start_on(&group.shard(0).unwrap()).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let peer = actor.request(StartPeer).await.unwrap().unwrap(); - peer.ready().await.unwrap(); + within!(peer.ready()).await.unwrap(); assert!(peer.request(Audit).await.unwrap().0.ends_with("actor-1")); assert!(matches!( actor.request(StartPeer).await.unwrap(), @@ -114,8 +117,8 @@ async fn persistent_native_slots_bound_admission_and_helpers_progress_during_blo ); let a = Probe.start_on(&group.shard(0).unwrap()).unwrap(); let b = Probe.start_on(&group.shard(1).unwrap()).unwrap(); - a.ready().await.unwrap(); - b.ready().await.unwrap(); + within!(a.ready()).await.unwrap(); + within!(b.ready()).await.unwrap(); assert!(matches!(Probe.start_on(&group), Err(SystemError::Capacity))); assert!(a.request(Audit).await.unwrap().0.ends_with("actor-0")); let (entered_tx, entered) = tokio::sync::oneshot::channel(); @@ -166,7 +169,7 @@ async fn native_capacity_is_released_only_after_registered_helper_cleanup() { let system = ActorSystem::new().unwrap(); let group = system.thread_group("slot-cleanup", 1).unwrap(); let actor = Probe.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let (entered_tx, entered) = tokio::sync::oneshot::channel(); let (release, release_rx) = std::sync::mpsc::channel(); let helper = actor @@ -196,7 +199,7 @@ async fn native_capacity_is_released_only_after_registered_helper_cleanup() { }) .await .unwrap(); - replacement.ready().await.unwrap(); + within!(replacement.ready()).await.unwrap(); assert!(system.shutdown(LIMIT).await.complete); } // Not simulated: native execution groups need OS threads. @@ -239,7 +242,7 @@ async fn each_strategy_recovers_on_a_full_native_pool_without_overlapping_slots( .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[1].current().unwrap().request(Crash).await.is_err()); for (i, child) in children.iter().enumerate() { if strategy == SupervisorStrategy::OneForAll @@ -293,9 +296,9 @@ async fn pinned_shards_preserve_explicit_cpu_placement_and_cross_shard_requests( .map(|i| Probe.start_on(&group.shard(i).unwrap()).unwrap()) .collect(); for (i, actor) in actors.iter().enumerate() { - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let peer_on_same_shard = actor.request(StartPeer).await.unwrap().unwrap(); - peer_on_same_shard.ready().await.unwrap(); + within!(peer_on_same_shard.ready()).await.unwrap(); assert_eq!( peer_on_same_shard.request(Audit).await.unwrap().1, Some(cpus[i]) @@ -315,7 +318,7 @@ async fn pinned_shards_preserve_explicit_cpu_placement_and_cross_shard_requests( }) .join() .unwrap(); - assert_eq!(helper.join().await.unwrap(), cpus[i]); + assert_eq!(within!(helper.join()).await.unwrap(), cpus[i]); } assert!(system.async_group("async", 1).unwrap().shard(0).is_err()); let report = system.shutdown(LIMIT).await; @@ -354,7 +357,7 @@ mod tls_join { let cpu = spawned_rt::tasks::affinity::available_cpus().unwrap()[0]; let group = system.pinned_group("tls-owner", vec![cpu]).unwrap(); let actor = Probe.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let (entered_tx, entered) = tokio::sync::oneshot::channel(); let (release, release_rx) = std::sync::mpsc::channel(); actor @@ -489,7 +492,7 @@ fn partial_pinned_construction_rolls_back_before_returning_error() { let runtime = spawned_rt::tasks::control_runtime().unwrap(); runtime.block_on(async { let actor = Probe.start_on(&group).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); assert!(actor.request(Audit).await.is_ok()); assert!(system.shutdown(LIMIT).await.complete); }); diff --git a/concurrency/tests/failpoints.rs b/concurrency/tests/failpoints.rs index 5142132b..4e5a759b 100644 --- a/concurrency/tests/failpoints.rs +++ b/concurrency/tests/failpoints.rs @@ -7,6 +7,9 @@ //! serializes the tests in this binary and prints the seed on failure. #![cfg(all(feature = "failpoints", feature = "async"))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ failpoints::{self, Action, FailPoint, FailScenario}, message::Message, @@ -122,7 +125,7 @@ async fn factory_errors_restart_within_budget_and_count_generations() { } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); let current = child.wait_ready().await.unwrap(); // Three injected failures, each its own generation, then one success. assert_eq!(failpoints::stats("child.factory").fired, 3); @@ -151,7 +154,7 @@ async fn persistent_factory_failure_exhausts_budget_and_fails_startup() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - match tree.start().await { + match within!(tree.start()).await { Err(SupervisorError::Failed(error)) => { assert!(error.contains("restart budget exhausted"), "{error}") } @@ -179,7 +182,7 @@ async fn factory_panic_is_a_failed_start_not_a_controller_failure() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap(); let snapshot = child.snapshot(); assert_eq!(snapshot.generation, 2); @@ -211,12 +214,14 @@ async fn seeded_probabilistic_factory_failures_match_generation_count() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); let mut actor = child.wait_ready().await.unwrap(); for _ in 0..8 { let generation = child.snapshot().generation; actor.send(Crash).unwrap(); - actor = child.wait_for_generation(generation).await.unwrap(); + actor = within!(child.wait_for_generation(generation)) + .await + .unwrap(); } // Every generation is either an injected failure or a started actor. let fired = failpoints::stats("child.factory").fired as usize; @@ -251,7 +256,7 @@ async fn ready_generation_discarded_before_publication_is_never_current() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); let current = child.wait_ready().await.unwrap(); let handles = started.handles(); assert_eq!(handles.len(), 2); @@ -326,7 +331,7 @@ async fn stop_while_restart_is_pending(site: &str) { } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap().send(Crash).unwrap(); reached(site, 1).await; let report = tree.shutdown(LIMIT).await; @@ -368,7 +373,7 @@ async fn controller_panic_before_restart_fails_supervisor_and_shuts_down() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap().send(Crash).unwrap(); tokio::time::timeout(LIMIT, tree.join()).await.unwrap(); assert!(tree @@ -402,9 +407,9 @@ async fn slow_controller_after_completed_cleanup_is_not_a_cleanup_failure() { .shutdown_timeout(Duration::from_millis(100)), ) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap().send(Crash).unwrap(); - child.wait_for_generation(1).await.unwrap(); + within!(child.wait_for_generation(1)).await.unwrap(); assert_eq!(failpoints::stats("child.before_cleanup_join").fired, 1); assert!(tree.failure().is_none(), "{:?}", tree.failure()); assert!(tree.shutdown(LIMIT).await.complete); @@ -432,7 +437,7 @@ async fn slow_stop_turn_reports_incomplete_at_deadline_then_joins() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap(); let began = Instant::now(); let report = tree.shutdown(Duration::from_millis(100)).await; @@ -460,8 +465,8 @@ async fn slow_system_coordinator_reports_then_completes() { let started = Started::default(); let a = group.start(Worker::new(&started)).unwrap(); let b = group.start(Worker::new(&started)).unwrap(); - a.ready().await.unwrap(); - b.ready().await.unwrap(); + within!(a.ready()).await.unwrap(); + within!(b.ready()).await.unwrap(); let report = system.shutdown(Duration::from_millis(50)).await; assert!(!report.complete); let report = report.wait(LIMIT).await; @@ -522,7 +527,7 @@ async fn placement_failure_is_reported_and_fails_the_supervisor() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - match tree.start().await { + match within!(tree.start()).await { Err(SupervisorError::Failed(error)) => { assert!(error.contains("placement failed"), "{error}") } @@ -571,7 +576,7 @@ async fn registration_racing_shutdown_leaves_no_entry() { assert!(system.shutdown(LIMIT).await.complete); assert_eq!(*registration.lock().unwrap(), Some(false)); assert!(registry::whereis::>("fp-register-race").is_none()); - assert!(actor.ready().await.is_err()); + assert!(within!(actor.ready()).await.is_err()); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -615,9 +620,9 @@ async fn replacement_registers_only_after_old_entry_is_removed() { } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap().send(Crash).unwrap(); - let current = child.wait_for_generation(1).await.unwrap(); + let current = within!(child.wait_for_generation(1)).await.unwrap(); assert!(failpoints::stats("registry.before_unregister").fired >= 1); let flags: Vec<_> = registrations .lock() @@ -680,7 +685,10 @@ async fn peer_exit_during_link_install_still_propagates() { linker.send(LinkTo(target.child_handle())).unwrap(); reached("link.before_install", 1).await; target.send(Crash).unwrap(); - assert!(matches!(target.wait_exit().await, ExitReason::Panic(_))); + assert!(matches!( + within!(target.wait_exit()).await, + ExitReason::Panic(_) + )); // The link completes after the peer died; the exit must still arrive. tokio::time::timeout(LIMIT, linker.join()).await.unwrap(); } @@ -699,14 +707,14 @@ async fn target_exit_during_monitor_install_still_delivers_down() { watcher.send(Watch(target.child_handle())).unwrap(); reached("monitor.before_install", 1).await; target.send(Crash).unwrap(); - target.wait_exit().await; + within!(target.wait_exit()).await; let reason = tokio::time::timeout(LIMIT, received.recv()) .await .unwrap() .unwrap(); assert!(matches!(reason, ExitReason::Panic(_))); watcher.context().stop(); - watcher.join().await; + within!(watcher.join()).await; } // ------------------------------------------------------------- actor.* (tasks) @@ -726,7 +734,7 @@ async fn injected_started_panic_is_an_abnormal_exit_seen_by_monitors() { let started = Started::default(); let target = Worker::new(&started).start(); watcher.send(Watch(target.child_handle())).unwrap(); - match target.wait_exit().await { + match within!(target.wait_exit()).await { ExitReason::Panic(message) => assert!(message.contains("actor.before_started")), other => panic!("expected panic exit, got {other:?}"), } @@ -737,7 +745,7 @@ async fn injected_started_panic_is_an_abnormal_exit_seen_by_monitors() { .unwrap(); assert!(matches!(reason, ExitReason::Panic(_))); watcher.context().stop(); - watcher.join().await; + within!(watcher.join()).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -753,7 +761,7 @@ async fn readiness_racing_shutdown_is_not_published() { let actor = group.start(Worker::new(&Started::default())).unwrap(); reached("actor.after_started_before_ready", 1).await; assert!(system.shutdown(LIMIT).await.complete); - assert!(actor.ready().await.is_err()); + assert!(within!(actor.ready()).await.is_err()); assert!(actor.exit_reason().is_some()); } @@ -768,7 +776,7 @@ async fn slow_stopped_reports_at_deadline_then_completes() { let system = ActorSystem::new().unwrap(); let group = system.async_group("fp-stopped", 1).unwrap(); let actor = group.start(Worker::new(&Started::default())).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let report = system.shutdown(Duration::from_millis(100)).await; assert!(!report.complete); assert_eq!(report.unfinished_actors, [actor.id()]); @@ -799,10 +807,10 @@ async fn injected_stopped_panic_restarts_transient_child() { .restart(spawned_concurrency::supervision::RestartMode::Transient), ) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); // A normal stop whose stopped() panics is abnormal, so Transient restarts. child.wait_ready().await.unwrap().context().stop(); - child.wait_for_generation(1).await.unwrap(); + within!(child.wait_for_generation(1)).await.unwrap(); assert!(matches!( child.snapshot().last_exit, Some(ExitReason::Panic(_)) @@ -836,7 +844,7 @@ async fn supervisor_coordinator_panic_is_reported_and_contained() { async move { Ok::<_, String>(Worker::new(&s)) } })) .unwrap(); - let tree = tree.start().await.unwrap(); + let tree = within!(tree.start()).await.unwrap(); child.wait_ready().await.unwrap(); let report = tree.shutdown(LIMIT).await; let failure = report.failure.as_ref().expect("no failure reported"); @@ -856,7 +864,7 @@ async fn system_shutdown_coordinator_panic_is_reported_and_contained() { let system = ActorSystem::new().unwrap(); let group = system.async_group("fp-gap-system", 1).unwrap(); let actor = group.start(Worker::new(&Started::default())).unwrap(); - actor.ready().await.unwrap(); + within!(actor.ready()).await.unwrap(); let report = system.shutdown(LIMIT).await; assert!(report.complete); assert!( diff --git a/concurrency/tests/full_actor_shards.rs b/concurrency/tests/full_actor_shards.rs index ba6bd413..5e00450b 100644 --- a/concurrency/tests/full_actor_shards.rs +++ b/concurrency/tests/full_actor_shards.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] #![cfg(feature = "backend-experiments")] //! Same actor semantics on default Tokio, batched Tokio and the opt-in shard pool. +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, tasks::{Actor, ActorStart, Context, Handler}, @@ -64,7 +67,7 @@ impl Handler for Worker { let child = Worker(Default::default()).start(); let _ = child.request(Thread).await.unwrap(); child.child_handle().stop(); - child.join().await; + within!(child.join()).await; let helper = tokio::spawn(async { tokio::time::sleep(Duration::from_millis(1)).await; 42u64 diff --git a/concurrency/tests/leaks.rs b/concurrency/tests/leaks.rs index 4491a0ca..b56c4cc8 100644 --- a/concurrency/tests/leaks.rs +++ b/concurrency/tests/leaks.rs @@ -16,6 +16,9 @@ //! `ActorRef`/`Context`, link tables, and native monitor waiter threads. // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{message::Message, registry, threads}; use std::cell::RefCell; use std::sync::{Arc, Weak}; @@ -236,7 +239,7 @@ mod async_leaks { } let timer = tasks::send_interval(Duration::from_secs(3600), actor.context(), Ping); actor.context().stop(); - actor.join().await; + within!(actor.join()).await; drop((actor, timer)); }); tasks_drain_to(&rt, 0, &format!("{backend:?}: runtime tasks after stop")); @@ -268,18 +271,18 @@ mod async_leaks { owner.context().monitor(&target.child_handle()); tasks_drain_to(&rt, baseline + 2, "two monitor waiters spawned"); owner.context().stop(); - owner.join().await; + within!(owner.join()).await; tasks_drain_to(&rt, baseline - 1, "monitor waiters after owner exit"); // Target exit delivers Down and releases the waiter. let watcher = Probe::new(Arc::new(())).start(); watcher.context().monitor(&target.child_handle()); target.request(Crash).await.unwrap_err(); - target.join().await; + within!(target.join()).await; watcher.request(Ping).await.unwrap(); tasks_drain_to(&rt, baseline - 1, "monitor waiter after target exit"); watcher.context().stop(); - watcher.join().await; + within!(watcher.join()).await; }); tasks_drain_to(&rt, 0, "runtime tasks after all actors stopped"); } @@ -356,8 +359,8 @@ mod async_leaks { a.context().link(&b.child_handle()); a.context().stop(); b.context().stop(); - a.join().await; - b.join().await; + within!(a.join()).await; + within!(b.join()).await; weaks }); tasks_drain_to(&rt, 0, &format!("{backend:?}: cycle runtime tasks")); @@ -451,7 +454,7 @@ mod async_leaks { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); // Several generations of the first child. for generation in 1..=3 { let current = children[0] @@ -641,9 +644,9 @@ mod sim_leaks { probe.context().stop(); a.context().stop(); b.context().stop(); - probe.join().await; - a.join().await; - b.join().await; + within!(probe.join()).await; + within!(a.join()).await; + within!(b.join()).await; drop((probe, timer, a, b)); settle(&format!("seed {seed}: actor state"), || { weak.iter().all(|w| w.strong_count() == 0) @@ -686,14 +689,14 @@ mod sim_leaks { owner.context().monitor(&target.child_handle()); owner.context().monitor(&target.child_handle()); owner.context().stop(); - owner.join().await; + within!(owner.join()).await; // The owner's own task is gone too. settle(&format!("seed {seed}: waiters after owner exit"), || { rt.stats().tasks < baseline }) .await; target.context().stop(); - target.join().await; + within!(target.join()).await; }); } } @@ -763,7 +766,7 @@ mod sim_leaks { let baseline = rt.stats().tasks; watcher.context().monitor(&target.child_handle()); target.request(Crash).await.unwrap_err(); - target.join().await; + within!(target.join()).await; watcher.request(Ping).await.unwrap(); // The target's task and the waiter are both gone. settle(&format!("seed {seed}: waiter after target exit"), || { @@ -771,7 +774,7 @@ mod sim_leaks { }) .await; watcher.context().stop(); - watcher.join().await; + within!(watcher.join()).await; }); } } @@ -814,11 +817,15 @@ mod sim_leaks { } })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); for generation in 1..=3 { - let current = child.wait_for_generation(generation - 1).await.unwrap(); + let current = within!(child.wait_for_generation(generation - 1)) + .await + .unwrap(); current.request(Crash).await.unwrap_err(); - child.wait_for_generation(generation).await.unwrap(); + within!(child.wait_for_generation(generation)) + .await + .unwrap(); } assert!(registry::registered().iter().any(|n| n == name)); assert!(supervisor.shutdown(Duration::from_secs(5)).await.complete); diff --git a/concurrency/tests/linked_exit_macro.rs b/concurrency/tests/linked_exit_macro.rs index cee8d125..ddec401d 100644 --- a/concurrency/tests/linked_exit_macro.rs +++ b/concurrency/tests/linked_exit_macro.rs @@ -1,4 +1,7 @@ //! Public #[actor] lifecycle wiring for trapped exits in both modes. +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{actor, Exit, ExitReason}; #[cfg(feature = "async")] @@ -41,8 +44,8 @@ mod asynchronous { .unwrap(); assert_eq!(event.from, child.id()); assert!(matches!(event.reason, ExitReason::Panic(_))); - child.wait_exit().await; - assert_eq!(observer.wait_exit().await, ExitReason::Normal); + within!(child.wait_exit()).await; + assert_eq!(within!(observer.wait_exit()).await, ExitReason::Normal); }); } } diff --git a/concurrency/tests/metrics.rs b/concurrency/tests/metrics.rs index a3615e4e..e381fe06 100644 --- a/concurrency/tests/metrics.rs +++ b/concurrency/tests/metrics.rs @@ -1,5 +1,8 @@ #![cfg(feature = "async")] #![cfg(feature = "metrics")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ metrics, tasks::{Actor, ActorStart, Backend, Context, Handler}, @@ -109,7 +112,7 @@ fn metrics_conserve_work_across_all_lifecycle_paths() { .await .unwrap(); actor.context().stop(); - assert_eq!(actor.wait_exit().await, ExitReason::Normal); + assert_eq!(within!(actor.wait_exit()).await, ExitReason::Normal); assert_eq!(actor.mailbox_depth(), 0); assert!(actor.send(Ping).is_err()); assert!(matches!( @@ -127,7 +130,7 @@ fn metrics_conserve_work_across_all_lifecycle_paths() { release.send(()).unwrap(); assert!(matches!(pending.await, Err(ActorError::ActorStopped))); assert!(raw.await.is_err()); - actor.join().await; + within!(actor.join()).await; assert_eq!(actor.mailbox_depth(), 0); let actor = Probe { gate: None }.start(); @@ -135,7 +138,10 @@ fn metrics_conserve_work_across_all_lifecycle_paths() { actor.request(Crash).await, Err(ActorError::ActorStopped) )); - assert!(matches!(actor.wait_exit().await, ExitReason::Panic(_))); + assert!(matches!( + within!(actor.wait_exit()).await, + ExitReason::Panic(_) + )); // Managed restart/shutdown must conserve the same aggregate accounting. let system = spawned_concurrency::system::ActorSystem::new().unwrap(); let group = system.async_group("metrics-recovery", 1).unwrap(); @@ -149,7 +155,7 @@ fn metrics_conserve_work_across_all_lifecycle_paths() { |_| async { Ok::<_, String>(Probe { gate: None }) }, )) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let first = child.current().unwrap(); assert_eq!(first.query().await.unwrap(), 9); assert!(first.request(Crash).await.is_err()); @@ -197,7 +203,7 @@ fn metrics_conserve_work_across_all_lifecycle_paths() { let actor = Probe { gate: None }.start_with_backend(backend); actor.request(Ping).await.unwrap(); actor.context().stop(); - actor.join().await; + within!(actor.join()).await; } }); drop(rt); diff --git a/concurrency/tests/metrics_sampling.rs b/concurrency/tests/metrics_sampling.rs index 23e4b1f4..5504acd0 100644 --- a/concurrency/tests/metrics_sampling.rs +++ b/concurrency/tests/metrics_sampling.rs @@ -1,5 +1,8 @@ #![cfg(feature = "async")] #![cfg(feature = "metrics")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, metrics, @@ -30,7 +33,7 @@ fn sampling_retains_every_counter_and_scales_only_timing_observations() { actor.request(Ping).await.unwrap(); } actor.context().stop(); - actor.join().await; + within!(actor.join()).await; }); let s = metrics::snapshot(); assert_eq!(s.messages_handled_total, 16); diff --git a/concurrency/tests/nested_supervision.rs b/concurrency/tests/nested_supervision.rs index 344bb6c3..f3180c86 100644 --- a/concurrency/tests/nested_supervision.rs +++ b/concurrency/tests/nested_supervision.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, supervision::{ @@ -85,7 +88,7 @@ async fn nested_budget_failure_reconstructs_fresh_subtree_and_preserves_parent_s Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let old_tree = subtree.current().unwrap(); let old_slot = subtree.snapshot(); let old_leaf = published @@ -106,7 +109,7 @@ async fn nested_budget_failure_reconstructs_fresh_subtree_and_preserves_parent_s assert_ne!(subtree.snapshot().actor_id, old_slot.actor_id); assert_eq!(calls.load(Ordering::SeqCst), 2); assert!(old_tree.failure().unwrap().contains("budget exhausted")); - old_tree.join().await; + within!(old_tree.join()).await; let new_leaf = published .lock() .unwrap() @@ -164,7 +167,7 @@ async fn temporary_subtree_failure_is_not_rebuilt() { Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let old_tree = subtree.current().unwrap(); let leaf = published .lock() @@ -368,7 +371,7 @@ async fn native_leaf_helper_blocks_subtree_replacement_and_remains_owned_after_a .shutdown_timeout(Duration::from_millis(20)), ) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let leaf = published .lock() .unwrap() @@ -438,7 +441,7 @@ async fn late_nested_cleanup_failures_are_retained_in_parent_and_system_reports( .shutdown_timeout(Duration::from_secs(2)), ) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let inner = nested.current().unwrap(); supervisor.stop(); spawned_rt::tasks::timeout(LIMIT, async { @@ -486,7 +489,7 @@ async fn aborted_initial_startup_rolls_back_before_a_later_stuck_shutdown_target let release = Arc::new(tokio::sync::Notify::new()); use spawned_concurrency::tasks::ActorStart; let later = HeldStop(release.clone()).start_on(&group).unwrap(); - later.ready().await.unwrap(); + within!(later.ready()).await.unwrap(); let owner = system.clone(); let shutdown = spawned_rt::tasks::spawn(async move { owner.shutdown(Duration::from_secs(30)).await }); @@ -525,7 +528,7 @@ async fn wait_ready_resolves_the_running_subtree() { }, )) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let ready = spawned_rt::tasks::timeout(LIMIT, subtree.wait_ready()) .await .unwrap() @@ -617,7 +620,7 @@ async fn nested_failed_replacement_retries_then_escalates_on_exhaustion() { }, )) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let old_leaf = published.lock().unwrap().take().unwrap(); assert!(old_leaf.current().unwrap().request(Crash).await.is_err()); if escalates { diff --git a/concurrency/tests/request_histories.rs b/concurrency/tests/request_histories.rs index 9cc436f8..3bd6e522 100644 --- a/concurrency/tests/request_histories.rs +++ b/concurrency/tests/request_histories.rs @@ -10,6 +10,9 @@ mod actor; mod bounded; #[path = "support/store_actor.rs"] mod store; +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use actor::*; use bounded::{Action, Operation as Bounded, Outcome}; use proptest::prelude::*; @@ -513,7 +516,7 @@ fn run(rt: &tokio::runtime::Runtime, mode: Mode, bug: Bug, program: &Program) -> task.await.unwrap(); } actor.child_handle().stop(); - actor.join().await; + within!(actor.join()).await; }), Mode::Threads => { let _ = rt; diff --git a/concurrency/tests/resource_cleanup.rs b/concurrency/tests/resource_cleanup.rs index 1642777d..a483fe61 100644 --- a/concurrency/tests/resource_cleanup.rs +++ b/concurrency/tests/resource_cleanup.rs @@ -1,5 +1,8 @@ // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; #[cfg(feature = "async")] use spawned_concurrency::tasks; use spawned_concurrency::{message::Message, threads}; @@ -99,7 +102,7 @@ fn pending_async_listener_is_dropped_on_actor_exit() { }); let listener = tasks::spawn_listener(actor.context(), stream); actor.child_handle().stop(); - actor.join().await; + within!(actor.join()).await; tokio::time::timeout(Duration::from_secs(2), listener) .await .unwrap() diff --git a/concurrency/tests/semantics_gaps.rs b/concurrency/tests/semantics_gaps.rs index ae0d6f3b..0043e87c 100644 --- a/concurrency/tests/semantics_gaps.rs +++ b/concurrency/tests/semantics_gaps.rs @@ -1839,9 +1839,16 @@ mod supervision_contract { .collect(); let supervisor = tree.start().await.unwrap(); let generations: Vec = children.iter().map(|c| c.snapshot().generation).collect(); - for child in &children { - child.current().unwrap().send(Crash).unwrap(); - } + // Take every generation before crashing any: on a multi-threaded + // runtime the first crash's round can already be stopping the others, + // so `current()` after a crash may report `Stopping`. A crash that + // reaches a sibling the round already stopped is absorbed by it; the + // check below holds either way. + let actors: Vec<_> = children.iter().map(|c| c.current().unwrap()).collect(); + for actor in &actors { + let _ = actor.send(Crash); + } + drop(actors); for (child, generation) in children.iter().zip(&generations) { next_generation(child, *generation).await; } diff --git a/concurrency/tests/sim_ordering.rs b/concurrency/tests/sim_ordering.rs index 13ccef97..1eea3469 100644 --- a/concurrency/tests/sim_ordering.rs +++ b/concurrency/tests/sim_ordering.rs @@ -14,6 +14,9 @@ //! ``` #![cfg(feature = "async")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, tasks::{Actor, ActorRef, ActorStart, Context, Handler, Recipient}, @@ -145,7 +148,7 @@ async fn per_sender_fifo_body() { ); } sink.child_handle().stop(); - sink.join().await; + within!(sink.join()).await; } #[spawned_macros::sim_test(explore = "pct:d=3", iterations = 1_000)] @@ -235,7 +238,7 @@ async fn turns_body() { outcomes.extend(task.await.unwrap()); } stopper.await.unwrap(); - actor.join().await; + within!(actor.join()).await; // L7: join() returned, so stopped() has completed and new work is refused. let events = snapshot(&events); @@ -416,7 +419,7 @@ async fn stop_discards_body() { stopper.await.unwrap(); // The stop was requested while Hold was running: nothing queued may run. release_tx.send(()).unwrap(); - actor.join().await; + within!(actor.join()).await; // L7: by the time join() returns every queued payload has been dropped // and every queued reply sender released. @@ -436,7 +439,7 @@ async fn stop_discards_body() { vec![Seen::Hold, Seen::Stopped], "D4: a queued message was dispatched after the stop" ); - assert_eq!(actor.wait_exit().await, ExitReason::Normal); + assert_eq!(within!(actor.wait_exit()).await, ExitReason::Normal); assert!( matches!( actor.send(Queued(Payload(drops.clone()))), @@ -584,7 +587,7 @@ async fn exit_after_last_message_body(end: End) { source.child_handle().stop(); } noise.await.unwrap(); - let reason = source.wait_exit().await; + let reason = within!(source.wait_exit()).await; for _ in 0..2 { timeout(LIMIT, notified.recv()) .await @@ -622,7 +625,7 @@ async fn exit_after_last_message_body(end: End) { End::External => assert_eq!(reason, ExitReason::Normal), } observer.child_handle().stop(); - observer.join().await; + within!(observer.join()).await; } async fn exit_after_last_message_all() { diff --git a/concurrency/tests/supervision.rs b/concurrency/tests/supervision.rs index 18192715..4e308ff3 100644 --- a/concurrency/tests/supervision.rs +++ b/concurrency/tests/supervision.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, supervision::{Backoff, ChildSpec, RestartMode, SupervisorConfig}, @@ -86,7 +89,7 @@ async fn one_for_one_uses_fresh_state_same_group_and_instance_bound_refs() { Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let old = child.current().unwrap(); let sibling_id = sibling.current().unwrap().id(); assert_eq!(old.request(Next).await.unwrap(), 1); @@ -130,7 +133,7 @@ async fn restart_modes_distinguish_normal_shutdown_and_panic() { .restart(mode), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let old = child.current().unwrap(); let generation = child.snapshot().generation; match &exit { @@ -176,7 +179,7 @@ async fn restart_budget_is_shared_between_siblings_and_exhaustion_closes_tree() Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(a.current().unwrap().request(Crash).await.is_err()); let a2 = spawned_rt::tasks::timeout(LIMIT, a.wait_for_generation(1)) .await @@ -268,7 +271,7 @@ async fn shutdown_during_backoff_prevents_new_factory_invocation() { async { Ok::<_, String>(Counter(0)) } })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(child.current().unwrap().request(Crash).await.is_err()); assert!(supervisor.shutdown(LIMIT).await.complete); assert_eq!(attempts.load(Ordering::SeqCst), 1); @@ -326,7 +329,7 @@ async fn cleanup_deadline_escalates_without_overlapping_a_live_generation() { .shutdown_timeout(Duration::from_millis(20)), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); spawned_rt::tasks::timeout(LIMIT, entered) .await .unwrap() @@ -393,7 +396,7 @@ async fn stuck_stopped_hook_triggers_cleanup_deadline_before_terminal_publicatio .shutdown_timeout(Duration::from_millis(20)), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let old = child.current().unwrap(); let response = old.request_raw(Crash).unwrap(); spawned_rt::tasks::timeout(LIMIT, entered) @@ -450,7 +453,7 @@ async fn initial_startup_and_shutdown_follow_dependency_order_across_groups() { })) .unwrap(); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let report = system.shutdown(LIMIT).await; assert!(report.complete); assert!(report.supervisor_failures.is_empty()); @@ -575,7 +578,7 @@ async fn repeated_crash_producer_and_shutdown_races_never_overlap_generations() async move { Ok::<_, String>(RaceActor { live, count: 0 }) } })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let old = child.current().unwrap(); let mut producers = Vec::new(); for _ in 0..8 { @@ -666,7 +669,7 @@ async fn system_shutdown_orders_standalone_actors_and_supervisor_trees_together( } .start_on(&group) .unwrap(); - a.ready().await.unwrap(); + within!(a.ready()).await.unwrap(); let mut trees = Vec::new(); for name in ["b", "d"] { let mut tree = system.supervisor(config(5)).unwrap(); @@ -676,7 +679,7 @@ async fn system_shutdown_orders_standalone_actors_and_supervisor_trees_together( async move { Ok::<_, String>(Ordered { name, log }) } })) .unwrap(); - trees.push(tree.start().await.unwrap()); + trees.push(within!(tree.start()).await.unwrap()); if name == "b" { Ordered { name: "c", @@ -721,7 +724,7 @@ async fn generation_waiters_wake_at_admission_close_before_their_shutdown_turn() } })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let mut waiter = Box::pin(first.wait_for_generation(99)); assert!(futures::poll!(waiter.as_mut()).is_pending()); supervisor.stop(); @@ -744,7 +747,7 @@ async fn stuck_supervisor_does_not_prevent_earlier_standalone_shutdown_after_sys let system = ActorSystem::new().unwrap(); let group = system.async_group("stuck-tree-order", 1).unwrap(); let earlier = Counter(0).start_on(&group).unwrap(); - earlier.ready().await.unwrap(); + within!(earlier.ready()).await.unwrap(); let (entered_tx, entered) = tokio::sync::oneshot::channel(); let (release, release_rx) = tokio::sync::oneshot::channel(); let args = Arc::new(std::sync::Mutex::new(Some((entered_tx, release_rx)))); @@ -767,7 +770,7 @@ async fn stuck_supervisor_does_not_prevent_earlier_standalone_shutdown_after_sys .shutdown_timeout(Duration::from_secs(60)), ) .unwrap(); - let _supervisor = tree.start().await.unwrap(); + let _supervisor = within!(tree.start()).await.unwrap(); let inner_actor = inside.current().unwrap(); let report = system.shutdown(Duration::from_millis(20)).await; assert!(!report.complete); @@ -859,7 +862,7 @@ async fn transient_replacement_exiting_before_readiness_is_retried() { .restart(RestartMode::Transient), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(child.current().unwrap().request(Crash).await.is_err()); let replacement = spawned_rt::tasks::timeout(LIMIT, child.wait_for_generation(2)) .await @@ -930,7 +933,7 @@ async fn failed_replacements_exhaust_the_restart_budget_exactly() { .restart(RestartMode::Transient), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(child.current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, supervisor.join()) .await @@ -1038,7 +1041,7 @@ async fn supervisor_report_deadline_signals_remaining_siblings() { .shutdown_timeout(Duration::from_secs(60)), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let actor = first.current().unwrap(); let report = supervisor.shutdown(Duration::from_millis(20)).await; assert!(!report.complete); @@ -1102,7 +1105,7 @@ async fn backoff_exponent_resets_only_after_a_stable_generation() { Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let snapshots = supervisor.snapshots(); assert_eq!(snapshots.len(), 1); assert_eq!(snapshots[0].name, "flappy"); @@ -1208,7 +1211,7 @@ async fn system_shutdown_reports_supervisor_failures() { Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(child.current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, supervisor.join()) .await @@ -1253,7 +1256,7 @@ async fn system_deadline_signals_remaining_supervised_siblings() { .shutdown_timeout(Duration::from_secs(60)), ) .unwrap(); - let _supervisor = tree.start().await.unwrap(); + let _supervisor = within!(tree.start()).await.unwrap(); let actor = first.current().unwrap(); let report = system.shutdown(Duration::from_millis(50)).await; assert!(!report.complete); @@ -1320,7 +1323,7 @@ async fn owned_registration_is_released_before_the_replacement_registers() { Ok::<_, String>(Named("supervision-gap-registered")) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); for generation in 1..=3 { let actor = child.current().unwrap(); assert!(actor.request(Crash).await.is_err()); @@ -1393,7 +1396,7 @@ async fn one_for_all_restart_waits_for_its_backoff() { Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); // A lower bound only: a slow machine makes the restart later, not earlier. let latency = restart_latency(&child).await; assert!( @@ -1485,7 +1488,7 @@ fn native_restart_window_expires_on_the_monotonic_clock() { Ok::<_, String>(Counter(0)) })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let child = &child; let crash = |generation| { let current = child.current().unwrap(); diff --git a/concurrency/tests/supervision_strategies.rs b/concurrency/tests/supervision_strategies.rs index 1cf5842f..714c56bf 100644 --- a/concurrency/tests/supervision_strategies.rs +++ b/concurrency/tests/supervision_strategies.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] // Tests that need OS threads are compiled out under spawned_sim. #![cfg_attr(spawned_sim, allow(unused_imports, dead_code))] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, supervision::{Backoff, ChildSpec, RestartMode, SupervisorConfig, SupervisorStrategy}, @@ -132,7 +135,7 @@ async fn strategies_select_affected_children_and_reconstruct_in_dependency_order .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let old: Vec<_> = children .iter() .map(|child| child.current().unwrap()) @@ -233,7 +236,7 @@ async fn temporary_collateral_child_stops_without_reconstruction() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[0].current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, children[0].wait_for_generation(1)) .await @@ -291,7 +294,7 @@ async fn failed_cohort_construction_rebuilds_the_whole_cohort_like_otp() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[0].current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, children[2].wait_for_generation(1)) .await @@ -347,7 +350,7 @@ async fn cohort_reconstruction_consumes_shared_factory_attempt_budget() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[0].current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, supervisor.join()) .await @@ -401,7 +404,7 @@ async fn simultaneous_failures_and_shutdown_do_not_overlap_cohorts_or_generation .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let barrier = Arc::new(tokio::sync::Barrier::new(2)); for child in &children { child @@ -464,7 +467,7 @@ async fn blocked_collateral_helper_prevents_entire_cohort_reconstruction() { .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let (entered_tx, entered) = tokio::sync::oneshot::channel(); let (release, release_rx) = std::sync::mpsc::channel(); let collateral = children[1].current().unwrap(); @@ -553,7 +556,7 @@ async fn upstream_failure_supersedes_a_round_while_a_dependent_factory_is_pendin .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(children[1].current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, pending.notified()) .await @@ -673,7 +676,7 @@ async fn replacement_stopping_before_readiness_restarts_the_strategy_scope() { .restart(RestartMode::Transient), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); // B crashes: OneForAll rebuilds A and B (A2 ready, B2 stops Normal // before readiness); RestForOne rebuilds only B. assert!(b.current().unwrap().request(CrashPlain).await.is_err()); diff --git a/concurrency/tests/supervision_trace.rs b/concurrency/tests/supervision_trace.rs index 171e4a29..9d33e28b 100644 --- a/concurrency/tests/supervision_trace.rs +++ b/concurrency/tests/supervision_trace.rs @@ -4,6 +4,9 @@ //! `verification/tla/trace_check.py` checks them against Supervision.tla. #![cfg(feature = "lifecycle-trace")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ lifecycle_trace, message::Message, @@ -126,7 +129,7 @@ async fn trace_queued_destructor_panic_restarts_transient_child() { let child = tree .add(worker("drain", &group).restart(RestartMode::Transient)) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let actor = child.current().unwrap(); let (entered, arrival) = tokio::sync::oneshot::channel(); let (release, gate) = tokio::sync::oneshot::channel(); @@ -168,7 +171,7 @@ async fn trace_queued_destructor_failure_preserves_shutdown() { let child = tree .add(worker("drain", &group).restart(RestartMode::Transient)) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let actor = child.current().unwrap(); let (entered, arrival) = tokio::sync::oneshot::channel(); let (release, gate) = tokio::sync::oneshot::channel(); @@ -223,7 +226,7 @@ async fn trace_context_shutdown_and_abnormal_exit() { let failed = tree .add(worker("failed", &group).restart(modes[1])) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let actor = clean.current().unwrap(); actor.request(ExitWith(ExitReason::Shutdown)).await.unwrap(); wait_state(&clean, ChildState::Stopped).await; @@ -277,7 +280,7 @@ async fn trace_normal_start_and_stop() { tree.add(worker(&format!("c{k}"), &group).restart(*mode)) .unwrap(); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(supervisor.shutdown(LIMIT).await.complete); write_trace("normal_start_stop", supervisor.trace_id(), &modes, 5); assert!(system.shutdown(LIMIT).await.complete); @@ -291,7 +294,7 @@ async fn trace_permanent_panic_restarts() { let mut tree = system.supervisor(config(5)).unwrap(); let a = tree.add(worker("a", &group).restart(modes[0])).unwrap(); let b = tree.add(worker("b", &group).restart(modes[1])).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); crash(&a).await; spawned_rt::tasks::timeout(LIMIT, a.wait_for_generation(1)) .await @@ -315,7 +318,7 @@ async fn trace_transient_normal_exit_is_not_restarted() { let mut tree = system.supervisor(config(5)).unwrap(); tree.add(worker("a", &group).restart(modes[0])).unwrap(); let b = tree.add(worker("b", &group).restart(modes[1])).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); b.current().unwrap().send(Quit).unwrap(); wait_state(&b, ChildState::Stopped).await; assert_eq!(b.snapshot().generation, 1); @@ -332,7 +335,7 @@ async fn trace_temporary_exit_is_not_restarted() { let mut tree = system.supervisor(config(5)).unwrap(); let a = tree.add(worker("a", &group).restart(modes[0])).unwrap(); tree.add(worker("b", &group).restart(modes[1])).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); crash(&a).await; wait_state(&a, ChildState::Stopped).await; assert!(supervisor.shutdown(LIMIT).await.complete); @@ -350,7 +353,7 @@ async fn trace_external_shutdown_of_permanent_child_restarts() { let mut tree = system.supervisor(config(5)).unwrap(); let a = tree.add(worker("a", &group).restart(modes[0])).unwrap(); let b = tree.add(worker("b", &group).restart(modes[1])).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); a.current().unwrap().child_handle().shutdown(); spawned_rt::tasks::timeout(LIMIT, a.wait_for_generation(1)) .await @@ -382,7 +385,7 @@ async fn trace_factory_error_then_success() { } })) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert_eq!(calls.load(Ordering::SeqCst), 2); assert!(supervisor.shutdown(LIMIT).await.complete); write_trace( @@ -425,7 +428,7 @@ async fn trace_restart_budget_exhaustion() { let mut tree = system.supervisor(config(2)).unwrap(); let a = tree.add(worker("a", &group)).unwrap(); let b = tree.add(worker("b", &group)).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); crash(&a).await; spawned_rt::tasks::timeout(LIMIT, a.wait_for_generation(1)) .await @@ -457,7 +460,7 @@ async fn trace_disabled_retries() { let modes = [RestartMode::Permanent]; let mut tree = system.supervisor(config(0)).unwrap(); let a = tree.add(worker("a", &group)).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); crash(&a).await; spawned_rt::tasks::timeout(LIMIT, supervisor.join()) .await @@ -477,7 +480,7 @@ async fn trace_shutdown_during_backoff() { policy.backoff.maximum = Duration::from_secs(60); let mut tree = system.supervisor(policy).unwrap(); let a = tree.add(worker("a", &group)).unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); crash(&a).await; wait_state(&a, ChildState::Backoff).await; assert!(supervisor.shutdown(LIMIT).await.complete); @@ -536,7 +539,7 @@ async fn trace_replacement_exiting_before_readiness_is_retried() { .restart(mode), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); assert!(child.current().unwrap().request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, child.wait_for_generation(2)) .await @@ -613,7 +616,7 @@ async fn trace_shutdown_deadline_with_slow_stop() { .shutdown_timeout(Duration::from_millis(100)), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let report = supervisor.shutdown(Duration::from_millis(20)).await; assert!(!report.complete); spawned_rt::tasks::sleep(Duration::from_millis(300)).await; @@ -667,7 +670,7 @@ async fn trace_randomized_lifecycles() { .unwrap() }) .collect(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); for _ in 0..rng.next(6) { let child = &children[rng.next(3) as usize]; let op = rng.next(3); diff --git a/concurrency/tests/supervisor_policies.rs b/concurrency/tests/supervisor_policies.rs index ea810f94..bda79d4b 100644 --- a/concurrency/tests/supervisor_policies.rs +++ b/concurrency/tests/supervisor_policies.rs @@ -1,6 +1,9 @@ #![cfg(feature = "async")] //! Generated policy scripts against a separate generation/retirement oracle. //! Action boundaries are synchronized; this does not control Tokio/OS schedules. +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use proptest::prelude::*; use spawned_concurrency::{ message::Message, @@ -224,7 +227,7 @@ async fn scenario( .unwrap(), ); } - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); let mut expected = vec![ Expected { generation: 1, @@ -257,18 +260,18 @@ async fn scenario( actor.request_with_timeout(Crash, LIMIT).await, Err(ActorError::ActorStopped) )); - let reason = actor.wait_exit().await; + let reason = within!(actor.wait_exit()).await; assert!(matches!(reason, ExitReason::Panic(_))); true } Action::Normal => { actor.context().stop(); - assert_eq!(actor.wait_exit().await, ExitReason::Normal); + assert_eq!(within!(actor.wait_exit()).await, ExitReason::Normal); false } Action::Shutdown => { actor.child_handle().shutdown(); - assert_eq!(actor.wait_exit().await, ExitReason::Shutdown); + assert_eq!(within!(actor.wait_exit()).await, ExitReason::Shutdown); false } }; @@ -371,7 +374,7 @@ async fn an_already_retired_child_reports_a_generation_error_without_spinning() .restart(RestartMode::Temporary), ) .unwrap(); - let supervisor = tree.start().await.unwrap(); + let supervisor = within!(tree.start()).await.unwrap(); child.current().unwrap().context().stop(); assert!( spawned_rt::tasks::timeout(LIMIT, child.wait_for_generation(1)) diff --git a/concurrency/tests/support/deadline.rs b/concurrency/tests/support/deadline.rs new file mode 100644 index 00000000..7c14ab61 --- /dev/null +++ b/concurrency/tests/support/deadline.rs @@ -0,0 +1,34 @@ +//! Bounded waits for integration tests. A wait that a bug turns into a hang +//! (a supervisor that never fails, an exit that is never published, a message +//! that never arrives) fails the test with the expression and line it was +//! waiting on, instead of hanging the job until CI's timeout. The bound is far +//! above any legitimate wait, which are all handshakes, never fixed sleeps: on +//! the simulator it is virtual time, on Tokio it is only reached by a bug. +//! +//! ```ignore +//! #[macro_use] +//! #[path = "support/deadline.rs"] +//! mod deadline; +//! let tree = within!(tree.start()).await.unwrap(); +//! ``` +#![allow(dead_code, unused_macros)] +use std::time::Duration; + +pub const WAIT: Duration = Duration::from_secs(120); + +#[cfg(feature = "async")] +pub async fn within(what: &str, future: F) -> F::Output { + match spawned_rt::tasks::timeout(WAIT, future).await { + Ok(output) => output, + Err(_) => panic!("timed out after {WAIT:?} waiting for {what}"), + } +} + +macro_rules! within { + ($future:expr) => { + $crate::deadline::within( + concat!("`", stringify!($future), "` at ", file!(), ":", line!()), + $future, + ) + }; +} diff --git a/concurrency/tests/tree_trace.rs b/concurrency/tests/tree_trace.rs index a9f8d323..28d6feaa 100644 --- a/concurrency/tests/tree_trace.rs +++ b/concurrency/tests/tree_trace.rs @@ -1,4 +1,7 @@ #![cfg(feature = "lifecycle-trace")] +#[macro_use] +#[path = "support/deadline.rs"] +mod deadline; use spawned_concurrency::{ message::Message, supervision::{ @@ -83,7 +86,7 @@ async fn mixed_strategy_three_level_shutdown_and_subtree_replacement() { }, )) .unwrap(); - let supervisor = root.start().await.unwrap(); + let supervisor = within!(root.start()).await.unwrap(); let old = leaf.lock().unwrap().as_ref().unwrap().current().unwrap(); assert!(old.request(Crash).await.is_err()); spawned_rt::tasks::timeout(LIMIT, subtree.wait_for_generation(1)) diff --git a/docs/TESTING.md b/docs/TESTING.md index 8b01a34e..608de8bd 100644 --- a/docs/TESTING.md +++ b/docs/TESTING.md @@ -688,8 +688,15 @@ oracle's name, so shrinking can match it): | X3, X5 | `Exit` carries the peer's published reason; a trapping, still-linked actor gets it (no more often than links were made); a non-trapping one does not survive a peer's abnormal exit | | X7 | No `Note` from a source after its `Down`/`Exit` at the same receiver | | X8 | An actor whose only possible stop cause was one driver `stop`, `shutdown` or `stop_with_reason` on it while alive publishes exactly that reason (skipped if a hook or the state destructor panicked) | +| X3 (reason) | A standalone, non-trapping actor whose only possible stop cause was its link to a peer survives the peer's `Normal` exit, and after `Shutdown`, `Panic` or `Kill` publishes the peer's reason, so the crash keeps propagating | +| X5 (unlink) | After the driver unlinks two running actors (and does not link them again), neither receives an `Exit` from the other, and a standalone actor whose only stop causes were links to that peer outlives it | +| T1 (cancel) | A timer cancelled at time `c` delivers at most the ticks due by `c` (`send_after`: at most one, none if cancelled before its period; `send_interval`: one per elapsed period) | | L4, M | An actor that panicked exits abnormally; a published reason never changes; an actor nothing could have stopped is still alive | | V1, V3, V5 | A Temporary child is built at most once per supervisor instance; no factory runs while an earlier generation of its slot (or subtree) is alive; retries per rolling window never exceed `max_restarts` | +| V1, V2 (restart decisions) | In a flat root tree, every replacement follows either the slot's own attempt ending in a way its mode restarts (failed or unpublished attempt, any exit for Permanent, an abnormal one for Transient), or an upstream sibling's attempt ending after the previous attempt began (any sibling for OneForAll, an earlier one for RestForOne). An instance nothing but its supervisor could have stopped needs the second | +| V1, V2, V5 (required restarts) | In a flat root tree the driver never closed and that did not fail, once teardown finds it settled: every published Permanent instance, and every published Transient one that exited abnormally, has a successor, and so does every running, non-Temporary sibling in its cohort (later siblings for RestForOne, all for OneForAll). A missing restart is an exhausted budget that failed to fail the supervisor | +| V5 (exhaustion) | A "restart budget exhausted" failure happens only when, at some moment between the exhausting child's last attempt and the tree's last grant, the grants in the window plus one reservation per sibling reach `max_restarts` | +| V6 | OneForOne: the virtual time from the end of an attempt to the next factory call is the backoff bound for the child's failure streak (from zero through it with jitter, plus 10 ms and a quarter of the bound for clock creep); the streak counts restarts and resets after a generation that stayed ready for `stable_after` (an interval when stability is within 1 ms of the threshold) | | V4, V9 | `current()` never returns an exited generation; a finished supervisor has every child `Stopped` or `Failed`, no live generation, and runs no factory afterwards; children stop in reverse declaration order | | S1-S4 | A complete system shutdown leaves no managed actor running and no coordinator failure; a shut-down system admits nothing; targets stop in reverse admission order | | M5, S3, S5 | After teardown: every actor state dropped, no registered name left, and the simulator's task count back to its starting value | @@ -779,6 +786,7 @@ input and, for random schedules, its tape: | V9 order | under the canonical schedule a yielding driver starved a child that was already stopping; quiescing now sleeps (`stopping_child_settles_before_order`) | | X3 | `ctx.link` from an exited actor's context is a no-op (`register_link` sees its own table closed), while linking to an exited peer delivers its reason (X5); found by the generated-input test before the campaign. The asymmetry is not stated in the semantics contract. | | livelock | a nested subtree that restarts more slowly than its parent's budget refills retries forever; the start is now bounded at 30 s | +| V5 exhaustion | the oracle measured the window at the end of the exhausting child's last attempt, but a collateral OneForAll child reserves only after its round's cleanup, and a superseding round (a sibling's failed rebuild) makes it reserve again later; the oracle now takes the most grants at any moment up to the tree's last grant (`budget_reserved_after_round`) | ## Scope @@ -1130,7 +1138,7 @@ single-use macros. Mutation testing breaks one operator at a time. To measure whether the stack catches the bugs a reviewer would actually miss, 24 realistic bugs were written by hand (`verification/planted/patches/`, one patch each, listed in -`INDEX.tsv`) and each was run, alone, against five layers: +`INDEX.tsv`) and each was run, alone, against six layers: | Layer | What runs | Budget | |---|---|---| @@ -1139,10 +1147,11 @@ by hand (`verification/planted/patches/`, one patch each, listed in | C | Fuzz harness tests on stable (`cd fuzz && cargo test --release`: 64 seeds, 64 generated inputs, the regression inputs) | none | | F | `cargo fuzz run` from `fuzz/seeds`, one process | 600 s | | V | TLA+ trace validation of the lifecycles recorded by the dedicated trace tests (`Supervision`, `Cohorts`, `NestedSupervision`) | none | +| K | Kani proofs of the supervisor arithmetic (`cargo kani`, every non-slow harness in `concurrency/src/kani_proofs.rs`; `KANI_ENV` names a file to source for a private install) | 3600 s | ```sh -verification/planted/run.sh -l TSCFV -f 600 -o /tmp/planted baseline # control: every layer must miss -verification/planted/run.sh -l TSCFV -f 600 -o /tmp/planted # all 24 patches +verification/planted/run.sh -l TSCFVK -f 900 -o /tmp/planted baseline # control: every layer must miss +verification/planted/run.sh -l TSCFVK -f 600 -o /tmp/planted # all 24 patches verification/planted/run.sh -l TS -o /tmp/planted verification/planted/patches/09-*.patch ``` @@ -1152,69 +1161,99 @@ seed files the run created (a seed found under one bug must not help the next). Simulator and fuzzer tapes go to `OUT/tmp`. Patches touch `concurrency/src` only; they are never applied in CI. -**Results** (2026-11-11, `cc60801d`, the 30-core shared server at `nice 19`, -four patches in parallel). Times are seconds for the whole layer (T, S, C, V) -or to the first failing input (F, from the seed corpus; "<1" means a seed or -one of the first inputs). A dash is a miss. The unpatched control missed in -every layer, including 600 s (590,000 executions) of fuzzing. - -| # | Bug | T | S | C | F | V | -|---|---|---|---|---|---|---| -| 01 | Restart window admits `max + 1` restarts | 53 (1 unit test) | - | - | - | - | -| 02 | A restart exactly `window` old still counts | 46 (same unit test) | - | - | - | - | -| 03 | RestForOne rebuilds the failed child and the ones *before* it | 98 | 21 | - | - | 18 | -| 04 | Transient restarts after `Shutdown` | 72 | 28 | - | - | 16 | -| 05 | Temporary sibling rebuilt as OneForAll/RestForOne collateral | 48 | 14 | - | 8 (V1) | - | -| 06 | Backoff failure streak never resets after a stable generation | 43 | 29 | - | - | - | -| 07 | Supervisor stops children in declaration order | 71 | 17 | - | 8 (V9) | 15 | -| 08 | Exhausted budget retires the child instead of failing the supervisor | hang (8 tests) | 16 | - | - | 14 | -| 09 | Nested supervisor's restart mode ignored (always Permanent) | - | - | 13 | <1 (V1) | - | -| 10 | `Normal` exit kills a non-trapping linked peer | 97 | 11 | - | - | - | -| 11 | `unlink` removes only the caller's edge | 64 (2 unit tests) | - | - | - | - | -| 12 | Linking to a dead peer reports `Normal`, not its reason | 72 | - | 20 | <1 (X3) | - | -| 13 | Actor killed through a link exits `Normal` (no transitive propagation) | 71 | 24 | - | - | - | -| 14 | `ChildHandle::shutdown` commits `Normal` instead of `Shutdown` | 55 | 11 | - | - | - | -| 15 | Monitor waiter delivers `Down` after `demonitor` removed it | - | - | - | - | - | -| 16 | Monitoring an already-dead actor gives no `Down` | 155 | - | 13 | 2 (X1) | - | -| 17 | Exiting observer does not cancel its monitor waiters | 59 | 14 | - | - | - | -| 18 | Managed actor's registered name never removed | 72 | 27 | 19 | <1 (S5) | - | -| 19 | Run loop keeps handling queued messages after stop (D4) | 81 | - | - | - | - | -| 20 | `ActorRef::send` enqueues from a spawned task (breaks D1) | hang (1 test) | 23 | 14 | <1 (D1) | - | -| 21 | `send_after` ignores its own cancellation token | 91 (1 unit test) | - | - | - | - | -| 22 | `send_interval` keeps ticking after its actor stops | 107 | 13 | - | - | - | -| 23 | System shutdown stops targets in admission order | 96 | 14 | 15 | <1 (S2) | - | -| 24 | System shutdown ignores admission order between actors and trees | 76 | 19 | - | 13 (S2) | - | -| | **Caught** | **22** | **15** | **6** | **9** | **4** | - -23 of 24 were caught by at least one layer; T alone caught 22. One bug (15) -escaped everything. What the matrix says about each layer: - -- **T is the backstop.** Its misses were 09 (caught only by the fuzzer) and - 15. Five catches rest on a single test each (unit tests for 01, 02, 11 - and 21; `backoff_exponent_resets_only_after_a_stable_generation` for 06): - delete that test and the bug ships. Two bugs (08, 20) make tests hang - instead of fail; in CI that is a job timeout, not a named failure. -- **S** only runs `sim_test`s and the `sim.rs` sweeps; the many plain - `#[test]`s that need OS threads are compiled out, so it misses most link, - timer and mailbox bugs (11, 12, 16, 19, 21) that T catches. -- **F found 9 bugs, all in under 15 s**, and nothing in the remaining 600 s. - It has no oracle for *which* children a restart rebuilds (03, 04), for the - restart budget's lower bound or window edge (01, 02, 08), for backoff - timing (06), or for the reason a stopped actor publishes when it agrees with - what its peers saw (13, 14): X3 and X1 compare a notification with the - published reason, so a wrong published reason that is consistently wrong - passes. Longer runs would not help without new oracles. -- **V** only sees supervisor lifecycles, and caught 4 of the 9 supervision - bugs (it cannot see restart timing, the budget window or Temporary rules - for nested trees). +**Results** (2026-11-12, after the follow-ups below, `53c0995d`, the 30-core +shared server at `nice 19`, one patch at a time). Times are seconds for the +whole layer (T, S, C, V, K) or to the first failing input (F, from the seed +corpus; "<1" means a seed or one of the first inputs), with the fuzzer oracle +that fired. A dash is a miss. The unpatched control missed in every layer, +including 900 s (440,000 executions) of fuzzing; a second 900 s control (770,000 executions) after +the last oracle change (X3 for `Normal` peers) also missed. Row 10's C and F +cells are from that rerun; the first matrix (without that oracle) missed both. + +| # | Bug | T | S | C | F | V | K | +|---|---|---|---|---|---|---|---| +| 01 | Restart window admits `max + 1` restarts | 47 (1 unit test) | - | - | - | - | 109 | +| 02 | A restart exactly `window` old still counts | 52 (same unit test) | - | - | - | - | 114 | +| 03 | RestForOne rebuilds the failed child and the ones *before* it | 67 | 20 | - | 10 (V1/V2) | 17 | - | +| 04 | Transient restarts after `Shutdown` | 56 | 17 | - | 6 (V1/V2) | 16 | - | +| 05 | Temporary sibling rebuilt as OneForAll/RestForOne collateral | 46 | 17 | - | 2 (V1) | - | - | +| 06 | Backoff failure streak never resets after a stable generation | 48 (1 test) | 16 | - | 36 (V6) | - | - | +| 07 | Supervisor stops children in declaration order | 44 | 19 | - | 3 (V9) | 15 | - | +| 08 | Exhausted budget retires the child instead of failing the supervisor | 212 (19 tests) | 14 | 8 | 2 (V5) | 14 | - | +| 09 | Nested supervisor's restart mode ignored (always Permanent) | 44 (1 test) | 14 | 8 | <1 (V1) | - | - | +| 10 | `Normal` exit kills a non-trapping linked peer | 46 | 15 | 8 | 1 (X3) | - | - | +| 11 | `unlink` removes only the caller's edge | 43 (2 unit tests) | - | - | 17 (X5) | - | - | +| 12 | Linking to a dead peer reports `Normal`, not its reason | 54 | - | 8 | <1 (X3) | - | - | +| 13 | Actor killed through a link exits `Normal` (no transitive propagation) | 43 | 17 | 9 | <1 (X3) | - | - | +| 14 | `ChildHandle::shutdown` commits `Normal` instead of `Shutdown` | 44 | 17 | 9 | <1 (X8) | - | - | +| 15 | Monitor waiter delivers `Down` after `demonitor` removed it | - | - | - | - | - | - | +| 16 | Monitoring an already-dead actor gives no `Down` | 137 | - | 9 | 5 (X1) | - | - | +| 17 | Exiting observer does not cancel its monitor waiters | 59 | 14 | - | - | - | - | +| 18 | Managed actor's registered name never removed | 64 | 12 | 11 | <1 (S5) | - | - | +| 19 | Run loop keeps handling queued messages after stop (D4) | 47 | 13 | - | 244 (X3) | - | - | +| 20 | `ActorRef::send` enqueues from a spawned task (breaks D1) | hang (1 unit test) | 13 | 8 | <1 (D1) | - | - | +| 21 | `send_after` ignores its own cancellation token | 41 (1 unit test) | - | 8 | <1 (T1) | - | - | +| 22 | `send_interval` keeps ticking after its actor stops | 55 (2 tests) | 13 | - | - | - | - | +| 23 | System shutdown stops targets in admission order | 65 | 12 | 8 | <1 (S2) | - | - | +| 24 | System shutdown ignores admission order between actors and trees | 40 | 13 | - | 11 (S2) | - | - | +| | **Caught** | **23** | **17** | **11** | **19** | **4** | **2** | + +Before the follow-ups (2026-11-11, `cc60801d`, five layers, no K): caught +T 22, S 15, C 6, F 9, V 4; bugs 08 and 20 hung T; F missed every +restart-decision, budget, backoff and exit-reason bug (01-04, 06, 08, 13, 14). +Now 23 of 24 are caught by at least one layer and every caught bug by at least +two; 15 still escapes everything (below). + +- **F now catches 19 of 24** (was 9), all but one within 40 s. The new oracles + (see "Scenario fuzzing") check which children a restart rebuilds (03, 04: + V1/V2), that every required restart happens or the supervisor fails (08: V5), + backoff delays in virtual time (06: V6), linked exit reasons (13, and 10 + after the `Normal` case was added: X3), `unlink` (11: X5) and cancelled + timers (21: T1). 19 fell to X3 after 244 s. Still missed: 01 and 02 (see + below), 15, 17 (an exiting observer's lingering monitor waiters) and 22 (an interval + that outlives its actor): no fuzzer oracle observes either yet; T and S catch + both. +- **01 and 02 are not observable through the public API.** `admit_in_window` + is only reached after `reserve_retry` has pruned the same window with `>=` + and checked `RetryBudget::available` with `<`, so the invariant + `grants in window + reservations <= max` already holds when it runs; a + looser comparison or window edge there admits nothing the reservation did + not. Their second check is the Kani harness + `restart_window_admits_exactly_within_budget` (layer K), which proves the + function against an independent log and fails under each patch. +- **Single-test catches now have a second, independent check:** 01 and 02 + (Kani), 06 (fuzzer V6), 11 (fuzzer X5), 21 (fuzzer T1). 09 is caught by its + `sim_test` and by the fuzzer. +- **No hangs from supervisor bugs.** Under 08, T used to hang in + `failpoints::persistent_factory_failure_exhausts_budget_and_fails_startup` + (and seven tests queued behind its failpoint lock); with bounded waits + (`within!`, "Bounded waits" above) 19 tests fail by name in 212 s. 20 still + hangs T, in the unit test `tasks::actor::tests::backend_blocking_handles_send_and_request` + inside `concurrency/src/tasks/actor.rs`, which these follow-ups leave + untouched while that file is being rewritten; the runner now names the test + that ran past 60 s. Earlier mutation runs show the same pattern for other + unit tests in `concurrency/src` (`policy_tests`, `nested::*_tests`, + `tasks::actor::tests` link tests, `child_handle::tests`): they await without + a deadline and should use one when those files next change. +- **S, C, V** are unchanged in kind; C now also runs the new oracles over the + seeds and regression inputs, which catches 08, 10, 13, 14 and 21 from committed + inputs. **Oracles added for the gaps** (each fails with its bug planted and passes on `main`; checked with `run.sh`): | Bug | Oracle | Layers that now catch it | |---|---|---| -| 09 | `nested_supervision::temporary_subtree_failure_is_not_rebuilt` (a `sim_test`) | T, S (was C, F only) | -| 14 | Fuzzer oracle **X8**: an actor whose only possible stop cause was the driver's `stop`, `shutdown` or `stop_with_reason` on a live actor publishes exactly that reason (instances whose hooks or state destructor panicked are skipped) | F < 1 s, C (was T, S only) | +| 01, 02 | Kani `restart_window_admits_exactly_within_budget`, now a runner layer (K) | T, K (was T only) | +| 03, 04 | Fuzzer **V1/V2** (restart decisions): every replacement is required by the slot's own exit and mode, or by an upstream sibling's failure | F, plus T, S, V (was T, S, V) | +| 06 | Fuzzer **V6**: restart delays match the backoff bound for the failure streak in virtual time | F, T, S (was T, S) | +| 08 | Fuzzer **V5** (required restarts); bounded waits in the tests that hung | F, C, T named (was a T hang, S, V) | +| 09 | `nested_supervision::temporary_subtree_failure_is_not_rebuilt` (a `sim_test`) | T, S, C, F | +| 10 | Fuzzer **X3**: a `Normal` peer exit spares a non-trapping actor linked only to it | F, C, T, S (was T, S) | +| 11 | Fuzzer **X5** (unlink): no `Exit` and no linked death after both sides unlinked | F, T (was T only) | +| 13 | Fuzzer **X3** (reason): an actor stopped through a link publishes its peer's reason | F, C, T, S (was T, S) | +| 14 | Fuzzer **X8**: an actor whose only possible stop cause was the driver's `stop`, `shutdown` or `stop_with_reason` on a live actor publishes exactly that reason (instances whose hooks or state destructor panicked are skipped) | F, C, T, S (was T, S) | +| 21 | Fuzzer **T1** (cancel): a cancelled timer delivers no tick due after the cancel | F, C, T (was T only) | | 15 | None possible through the public API: see below | - | Bug 15 is not observable. `demonitor` is documented best-effort ("the message @@ -1448,6 +1487,21 @@ unfixed code before pushing, and name it in the PR (the [Bugs reproduced by these regressions](#bugs-reproduced-by-these-regressions) follows this rule. +### Bounded waits + +A test that awaits an outcome a bug can suppress (a supervisor's `start()` +that never fails, a `join()` or `wait_exit()` whose exit is never published, a +generation that never comes, a reply that never arrives) hangs instead of +failing, and CI then reports a job timeout rather than a test name. Integration +tests wrap those awaits in `within!(..)` (`concurrency/tests/support/deadline.rs`), +which fails with the awaited expression, file and line after 120 s: virtual +time on the simulator, and only reached by a bug on Tokio, since every such wait +is a handshake, never a fixed sleep. Tests on paused Tokio time +(`protocol_timeouts`, `semantics_gaps`) are left alone: paused time +auto-advances while work runs outside Tokio, which could fire the bound early. +Unit tests inside `concurrency/src` do not use it yet (see "Planted-bug +sensitivity" for the ones that still hang under a planted bug). + ### Single-CPU runs The `single-cpu` job in `ci.yml` builds the tests normally, then runs diff --git a/fuzz/regressions/budget_reserved_after_round.input b/fuzz/regressions/budget_reserved_after_round.input new file mode 100644 index 00000000..51ead054 Binary files /dev/null and b/fuzz/regressions/budget_reserved_after_round.input differ diff --git a/fuzz/regressions/budget_reserved_after_round.tape b/fuzz/regressions/budget_reserved_after_round.tape new file mode 100644 index 00000000..c6fd7ac9 --- /dev/null +++ b/fuzz/regressions/budget_reserved_after_round.tape @@ -0,0 +1,82 @@ +# spawned-sim tape v2 +seed 8875277784326192161 +spin 10000 +c 0 +c 1 +c 2 +p 4 +c 3 +c 5 +p 6 +c 4 +c 2 +c 7 +c 8 +c 7 +c 9 +c 7 +c 2 +c 0 +c 6 +c 0 +p 9 +p 4 +p 3 +p 4 +c 0 +c 7 +c 9 +t 1000000 +c 0 +t 51000000 +c 0 +t 101000000 +c 0 +t 151000000 +c 0 +t 201000000 +c 0 +t 251000000 +c 0 +t 300000000 +c 9 +c 7 +c 3 +c 4 +c 3 +c 4 +t 301000000 +c 0 +t 322689475 +c 4 +c 10 +c 4 +c 3 +c 4 +c 7 +c 3 +c 4 +c 3 +c 4 +t 351000000 +c 0 +t 379924451 +c 4 +c 11 +p 12 +c 4 +c 3 +c 7 +p 3 +p 4 +c 2 +p 12 +c 4 +c 2 +t 401000000 +c 0 +t 402000000 +c 0 +c 13 +c 1 +c 0 diff --git a/fuzz/src/driver.rs b/fuzz/src/driver.rs index 8c53c7c3..66aab672 100644 --- a/fuzz/src/driver.rs +++ b/fuzz/src/driver.rs @@ -4,7 +4,9 @@ //! so a run is a function of the input, the seed and the strategy. use crate::scenario::*; -use crate::world::{factory_call, now, Act, Do, SupInst, Worker, World}; +use crate::world::{ + factory_call, factory_outcome, now, Act, Do, Outcome, SupInst, TimerRec, Worker, World, +}; use spawned_concurrency::{ registry, supervision::{ @@ -66,6 +68,8 @@ enum Slot { struct Root { sup: Option, + /// Its supervisor instance in [`crate::world::State::sups`]. + inst: usize, sys: usize, /// Declared child paths, with the declared child count of nested slots. paths: Vec<(String, usize)>, @@ -111,7 +115,8 @@ struct Driver { actors: Vec, roots: Vec, units: Vec<(usize, Unit)>, - timers: Vec, + /// Live timers with their index in [`crate::world::State::timers`]. + timers: Vec<(TimerHandle, usize)>, pending: Vec>, links: Vec, /// Trap flag set by the driver and the op index of its last change. @@ -122,6 +127,16 @@ struct Driver { requested: HashMap, /// Total handler sleep ever sent: a fence may wait behind all of it. queued_sleep: Duration, + /// Driver unlinks, for the X5 unlink oracle. + unlinks: Vec, +} + +/// `a.unlink(b)` by the driver while both were running. +struct UnlinkRec { + a: ActorRef, + b: ActorRef, + /// Op index (a later `LinkRec` for the pair re-links it). + at: u64, } /// Run one scenario in the current simulated world; panics with the first @@ -157,8 +172,18 @@ pub async fn run_body(scenario: Scenario, seed: u64) { driver.teardown().await; trace(|| "leak checks".into()); drop(driver); + // Published exit reasons, then release the handles before leak checks. + { + let mut s = world.lock(); + for inst in &mut s.insts { + if let Some(handle) = inst.handle.take() { + inst.reason = handle.exit_reason(); + } + } + } leaks(&world, baseline).await; budget_checks(&world); + crate::oracles::check(&world); let violations = std::mem::take(&mut world.lock().violations); if let Some(first) = violations.first() { LAST_DETAILS.with(|d| { @@ -281,6 +306,13 @@ fn new_sup_inst(world: &World, root: usize, cfg: &SupervisorConfig) -> usize { root, max_restarts: cfg.max_restarts, window: cfg.restart_window, + strategy: cfg.strategy, + backoff: cfg.backoff.clone(), + flat: false, + children: 0, + closed_at: None, + settled: None, + failure: None, }); s.sups.len() - 1 } @@ -292,18 +324,20 @@ fn worker_spec( def: WorkerDef, group: &ExecutionGroup, ) -> ChildSpec { - let temporary = def.restart % 3 == 2; + let mode = restart_mode(def.restart); let name = path.clone(); ChildSpec::new(name, group, move |fctx: FactoryContext| { let world = world.clone(); let path = path.clone(); async move { - factory_call(&world, inst, &path, temporary); + let call = factory_call(&world, inst, &path, mode); let bit = 1u8 << (fctx.generation() % 8); if def.fail_mask & bit != 0 { + factory_outcome(&world, call, Outcome::Error); return Err("fuzz: injected factory error".to_string()); } if def.panic_mask & bit != 0 { + factory_outcome(&world, call, Outcome::Panic); panic!("fuzz: injected factory panic"); } let hooks = if def.hook_mask & bit != 0 { @@ -313,10 +347,11 @@ fn worker_spec( }; let worker = Worker::new(&world, Some(path), hooks); Worker::arm(&world, worker.instance()); + factory_outcome(&world, call, Outcome::Instance(worker.instance())); Ok(worker) } }) - .restart(restart_mode(def.restart)) + .restart(mode) .startup_timeout(Duration::from_millis(limit_ms(def.startup_timeout))) .shutdown_timeout(Duration::from_millis(limit_ms(def.shutdown_timeout))) } @@ -346,6 +381,7 @@ impl Driver { first_exit: HashMap::new(), requested: HashMap::new(), queued_sleep: Duration::ZERO, + unlinks: Vec::new(), }; driver.new_system(); driver @@ -406,9 +442,36 @@ impl Driver { .entry(a.id()) .or_insert_with(|| a.child_handle()); } + self.mark_published(&out); out } + /// Generations `current()` returned were published, so ready (V4). + fn mark_published(&self, refs: &[ActorRef]) { + let mut s = self.world.lock(); + for a in refs { + if let Some(i) = s.instance_of(a.id()) { + if s.insts[i].slot.is_some() { + s.insts[i].published = true; + } + } + } + } + + fn cancel_timer(&mut self, i: usize) { + let (timer, id) = self.timers.remove(i); + timer.cancellation_token.cancel(); + let mut s = self.world.lock(); + let rec = &mut s.timers[id]; + rec.cancelled.get_or_insert(now()); + } + + /// The driver closed root `root` (stop, shutdown or its system's). + fn closing(&self, root: usize) { + let inst = self.roots[root].inst; + self.world.lock().sups[inst].closed_at.get_or_insert(now()); + } + fn pick(&mut self, index: u8) -> Option> { let targets = self.targets(); (!targets.is_empty()).then(|| targets[index as usize % targets.len()].clone()) @@ -525,6 +588,7 @@ impl Driver { let msg = Do { nonce: self.next_nonce(), lane: 0, + timer: 0, act, }; let _ = if via_recipient { @@ -549,7 +613,12 @@ impl Driver { 0 }; let ms = timeout.map_or(5_000, |t| t as u64); - let msg = Do { nonce, lane, act }; + let msg = Do { + nonce, + lane, + timer: 0, + act, + }; let request = async move { let start = now(); let reply = match timeout { @@ -596,6 +665,23 @@ impl Driver { let (Some(pa), Some(pb)) = (self.pick(a), self.pick(b)) else { return; }; + // Both running and not yet stopping: no Exit between them can + // be in flight, so none may arrive after this (X5). + let running = |d: &Self, r: &ActorRef| { + let s = d.world.lock(); + r.child_handle().is_alive() + && s.instance_of(r.id()).is_some_and(|i| { + let x = &s.insts[i]; + x.started_end.is_some() && x.stopped_begin.is_none() + }) + }; + if pa.id() != pb.id() && running(self, &pa) && running(self, &pb) { + self.unlinks.push(UnlinkRec { + a: pa.clone(), + b: pb.clone(), + at: self.op, + }); + } pa.context().unlink(&pb.child_handle()); for l in &mut self.links { let ids = (l.a.id(), l.b.id()); @@ -684,23 +770,35 @@ impl Driver { cmd => cmd, }; let act = self.act(cmd, &aref); + let period = Duration::from_millis(1 + ms as u64 % 5_000); + let id = { + let mut s = self.world.lock(); + s.timers.push(TimerRec { + created: now(), + period, + interval, + cancelled: None, + handled: 0, + }); + s.timers.len() - 1 + }; let msg = Do { nonce: 0, lane: 0, + timer: 1 + id, act, }; - let period = Duration::from_millis(1 + ms as u64 % 5_000); let timer = if interval { send_interval(period, aref.context(), msg) } else { send_after(period, aref.context(), msg) }; - self.timers.push(timer); + self.timers.push((timer, id)); } Op::CancelTimer { which } => { if !self.timers.is_empty() { let i = which as usize % self.timers.len(); - self.timers[i].cancellation_token.cancel(); + self.cancel_timer(i); } } Op::Advance { ms } => rt::sleep(Duration::from_millis(ms as u64 % 5_000)).await, @@ -729,6 +827,7 @@ impl Driver { let msg = Do { nonce: self.next_nonce(), lane: 0, + timer: 0, act, }; drop(aref.request_raw(msg)); @@ -868,7 +967,7 @@ impl Driver { root, inst, path.clone(), - *restart % 3 == 2, + restart_mode(*restart), children, groups.clone(), nested_cfg.clone(), @@ -900,8 +999,14 @@ impl Driver { Err(_) => None, }; trace(|| format!("supervisor r{root}: started {}", sup.is_some())); + { + let mut s = self.world.lock(); + s.sups[inst].flat = model.is_some() && sup.is_some(); + s.sups[inst].children = paths.len(); + } self.roots.push(Root { sup, + inst, sys, paths, slots, @@ -953,18 +1058,23 @@ impl Driver { let Ok(aref) = workers[i].current() else { return; }; + self.mark_published(std::slice::from_ref(&aref)); if !aref.child_handle().is_alive() { self.violate( "V4: current() returned a generation that already exited", format!("{:?}", workers[i].snapshot().state), ); } + // The restart oracles tell a child's own exit from its supervisor's + // stop by these causes. + self.may_stop(aref.id()); match how { ChildCrash::Panic => { let nonce = self.next_nonce(); let _ = aref.send(Do { nonce, lane: 0, + timer: 0, act: Act::Panic, }); } @@ -1133,6 +1243,7 @@ impl Driver { return; }; let was_closing = std::mem::replace(&mut self.roots[root].closing, true); + self.closing(root); match mode % 3 { 0 => s.stop(), 1 => { @@ -1162,6 +1273,11 @@ impl Driver { /// Mark a system's actors as stoppable: its shutdown began. fn system_stopping(&mut self, sys: usize) { + for root in 0..self.roots.len() { + if self.roots[root].sys == sys { + self.closing(root); + } + } let ids: Vec = self .actors .iter() @@ -1424,6 +1540,7 @@ impl Driver { let msg = Do { nonce, lane: 0, + timer: 0, act: Act::Noop, }; match aref @@ -1495,8 +1612,8 @@ impl Driver { /// until nothing is in flight. False if supervisors did not settle. async fn quiesce(&mut self) -> bool { trace(|| "quiesce".into()); - for t in self.timers.drain(..) { - t.cancellation_token.cancel(); + while !self.timers.is_empty() { + self.cancel_timer(0); } self.collect_pending().await; for _ in 0..2 { @@ -1642,6 +1759,9 @@ impl Driver { rt::sleep(Duration::from_millis(10)).await; } if !alive { + if !trap { + self.linked_reason(wid, &published); + } continue; } if trap { @@ -1665,6 +1785,8 @@ impl Driver { } } + self.unlink_checks(); + let standalone: Vec<(ActorId, usize, Option)> = self .actors .iter() @@ -1715,8 +1837,112 @@ impl Driver { } } + /// A standalone, non-trapping actor whose only possible stop cause was + /// its link to a peer: a `Normal` peer exit must not have stopped it, and + /// any other publishes the peer's reason (X3: "exits with the same reason + /// ... propagates further"). + fn linked_reason(&self, id: ActorId, peer: &ExitReason) { + let Some(a) = self.actors.iter().find(|a| a.aref.id() == id) else { + return; + }; + let Some(exit) = a.aref.child_handle().exit_reason() else { + return; + }; + let s = self.world.lock(); + let x = &s.insts[a.instance]; + if s.causes.get(&id) != Some(&1) || x.panicked || x.drop_panicked { + return; + } + let detail = format!("w{} published {exit:?}, peer {peer:?}", a.instance); + drop(s); + if *peer == ExitReason::Normal { + self.violate( + "X3: a non-trapping actor was stopped by a linked peer's Normal exit", + detail, + ); + } else if exit != *peer { + self.violate( + "X3: an actor stopped through a link published another reason than its peer", + detail, + ); + } + } + + /// After `a.unlink(b)` between two running actors (and no later link of + /// the pair), neither may hear of the other's exit (X5: unlink removes + /// the link on both sides). + fn unlink_checks(&self) { + for u in &self.unlinks { + let (aid, bid) = (u.a.id(), u.b.id()); + let pair = |l: &LinkRec| { + (l.a.id(), l.b.id()) == (aid, bid) || (l.a.id(), l.b.id()) == (bid, aid) + }; + if self.links.iter().any(|l| pair(l) && l.at > u.at) { + continue; + } + { + let s = self.world.lock(); + if s.uncertain.contains(&aid) || s.uncertain.contains(&bid) { + continue; + } + } + for (me, other) in [(&u.a, &u.b), (&u.b, &u.a)] { + let (mid, oid) = (me.id(), other.id()); + if other.child_handle().exit_reason().is_none() { + continue; + } + let s = self.world.lock(); + let Some(i) = s.instance_of(mid) else { continue }; + if s.insts[i].exits.iter().any(|(from, _)| *from == oid) { + let detail = format!("w{i} from {}", s.name(oid)); + drop(s); + self.violate("X5: an Exit arrived from a peer after unlink", detail); + continue; + } + // A standalone actor whose only stop causes were its links to + // this peer must outlive the peer. + let links = self + .links + .iter() + .filter(|l| l.a.id() == mid || l.b.id() == mid) + .collect::>(); + let only_peer = links.iter().all(|l| pair(l)); + let standalone = self.actors.iter().any(|a| a.aref.id() == mid); + let x = &s.insts[i]; + if standalone + && only_peer + && s.causes.get(&mid).copied().unwrap_or(0) as usize == links.len() + && !x.panicked + && !x.drop_panicked + && me.child_handle().exit_reason().is_some() + { + let detail = format!( + "w{i} {:?} after {} {:?}", + me.child_handle().exit_reason(), + s.name(oid), + other.child_handle().exit_reason() + ); + drop(s); + self.violate("X5: an actor died through a link it had removed", detail); + } + } + } + } + async fn teardown(&mut self) { if self.quiesce().await { + // Every tree the driver never closed is settled: a restart the + // contract requires has happened by now (V1, V2, V5 oracles). + let tick = self.world.lock().tick(); + for root in &self.roots { + let failure = root.sup.as_ref().and_then(|s| s.failure()); + let mut s = self.world.lock(); + let sup = &mut s.sups[root.inst]; + if sup.closed_at.is_none() { + sup.settled = Some(tick); + sup.failure = failure; + } + } self.eventual().await; } for root in 0..self.roots.len() { @@ -1788,7 +2014,7 @@ fn budget_checks(world: &World) { let n = seen.entry((c.inst, c.slot.clone())).or_default(); *n += 1; if *n > 1 { - if c.temporary { + if c.mode == RestartMode::Temporary { found.push(( "V1: a Temporary child was constructed twice", c.slot.clone(), @@ -1828,13 +2054,13 @@ fn nested_configure( root: usize, parent_inst: usize, path: String, - temporary: bool, + mode: RestartMode, children: Vec, groups: Vec, cfg: SupervisorConfig, ) -> impl Fn(&mut SupervisorBuilder) -> Result<(), SupervisorError> + Send + Sync + 'static { move |builder: &mut SupervisorBuilder| { - factory_call(&world, parent_inst, &path, temporary); + factory_call(&world, parent_inst, &path, mode); let inst = new_sup_inst(&world, root, &cfg); for (j, w) in children.iter().enumerate() { let p = format!("{path}/c{j}"); diff --git a/fuzz/src/lib.rs b/fuzz/src/lib.rs index f624df3c..d49a0a3d 100644 --- a/fuzz/src/lib.rs +++ b/fuzz/src/lib.rs @@ -18,6 +18,7 @@ compile_error!( ); mod driver; +mod oracles; pub mod scenario; mod world; diff --git a/fuzz/src/oracles.rs b/fuzz/src/oracles.rs new file mode 100644 index 00000000..8729325c --- /dev/null +++ b/fuzz/src/oracles.rs @@ -0,0 +1,374 @@ +//! End-of-run oracles over the factory log and the instance log: restart +//! decisions (V1, V2), required restarts against budget exhaustion (V5), +//! backoff delays in virtual time (V6) and cancelled timers. They judge only +//! what the log determines; anything a race could explain is skipped, so a +//! violation is a contract breach, never a schedule the harness misread. +//! +//! Restart and backoff checks cover flat root supervisors (every declared +//! child a worker). A worker instance "could stop by itself" when something +//! other than its supervisor may have stopped it: a driver or handler stop, +//! an injected panic or destructor panic, a link, or a startup it may not +//! have finished. An instance that could not was stopped by its supervisor +//! (a cohort round, a close or a failure). + +use crate::world::{Call, Inst, Outcome, State, SupInst, World}; +use spawned_concurrency::{ + supervision::{RestartMode, SupervisorStrategy}, + ExitReason, +}; +use std::time::Duration; + +/// Slack on the upper bound of an observed backoff delay: virtual time +/// creeps 1 ms per busy stretch of polls. +fn slack(bound: Duration) -> Duration { + Duration::from_millis(10) + bound / 4 +} + +pub fn check(world: &World) { + let mut s = world.lock(); + let mut found: Vec<(&'static str, String)> = Vec::new(); + for (inst, sup) in s.sups.iter().enumerate() { + if !sup.flat { + continue; + } + let calls: Vec<(usize, &Call)> = s + .calls + .iter() + .enumerate() + .filter(|(_, c)| c.inst == inst) + .collect(); + let slots: Vec> = (0..sup.children) + .map(|j| { + calls + .iter() + .filter(|(_, c)| c.index == j) + .map(|(_, c)| *c) + .collect() + }) + .collect(); + restart_justified(&s, sup, &slots, &mut found); + required_restarts(&s, sup, &slots, &mut found); + budget_exhaustion(&s, sup, &calls, &mut found); + if sup.strategy == SupervisorStrategy::OneForOne { + backoff_delays(&s, sup, &slots, &mut found); + } + } + timers(&s, &mut found); + for (kind, detail) in found { + s.violate(kind, detail); + } +} + +fn instance<'a>(s: &'a State, c: &Call) -> Option<(usize, &'a Inst)> { + match c.outcome { + Outcome::Instance(i) => Some((i, &s.insts[i])), + _ => None, + } +} + +/// Something other than its supervisor may have stopped this instance. +fn could_stop_itself(s: &State, x: &Inst) -> bool { + !x.published + || x.panicked + || x.drop_panicked + || x.id.is_none_or(|id| s.may_stop.contains(&id)) +} + +/// Tick at which an attempt ended: the factory's return, or the drop of the +/// instance's state (before its exit is published, L6/L7). +fn ended(s: &State, c: &Call) -> Option { + match c.outcome { + Outcome::Error | Outcome::Panic => Some(c.seq), + Outcome::Instance(i) => s.insts[i].dropped, + Outcome::Pending => None, + } +} + +fn ended_at(s: &State, c: &Call) -> Option { + match c.outcome { + Outcome::Error | Outcome::Panic => Some(c.at), + Outcome::Instance(i) => s.insts[i].dropped_at, + Outcome::Pending => None, + } +} + +/// Siblings whose failure or stop can make `j` collateral: every other slot +/// for OneForAll, the earlier ones for RestForOne (V2). +fn upstream(strategy: SupervisorStrategy, j: usize, k: usize) -> bool { + match strategy { + SupervisorStrategy::OneForOne => false, + SupervisorStrategy::OneForAll => k != j, + SupervisorStrategy::RestForOne => k < j, + } +} + +fn abnormal(reason: &Option) -> bool { + reason.as_ref().is_none_or(ExitReason::is_abnormal) +} + +/// V1/V2: every replacement is one the restart table or the strategy asks +/// for. A replacement of slot `j` follows an attempt that ended by itself +/// in a way its restart mode restarts (a failed or unpublished attempt +/// unless Temporary, any exit for Permanent, an abnormal one for +/// Transient), or an attempt of an upstream sibling that ended after the +/// previous attempt of `j` began (`j` was collateral of that sibling's +/// round). Temporary replacements are V1's own check. +fn restart_justified( + s: &State, + sup: &SupInst, + slots: &[Vec<&Call>], + found: &mut Vec<(&'static str, String)>, +) { + for (j, attempts) in slots.iter().enumerate() { + for pair in attempts.windows(2) { + let (prev, next) = (pair[0], pair[1]); + if next.mode == RestartMode::Temporary { + continue; + } + let own = match instance(s, prev) { + None => true, + Some((_, x)) => { + if x.dropped.is_none_or(|d| d > next.seq) { + continue; // V3 reports an overlap. + } + could_stop_itself(s, x) + && match next.mode { + RestartMode::Permanent => true, + RestartMode::Transient => !x.published || abnormal(&x.reason), + RestartMode::Temporary => false, + } + } + }; + if own { + continue; + } + let collateral = slots.iter().enumerate().any(|(k, other)| { + upstream(sup.strategy, j, k) + && other.iter().any(|c| { + c.seq < next.seq + && ended(s, c).is_some_and(|e| e > prev.seq && e < next.seq) + }) + }); + if !collateral { + let reason = instance(s, prev).map(|(i, x)| (i, x.reason.clone())); + found.push(( + "V1/V2: a child was restarted although neither its exit nor its strategy requires it", + format!( + "{} ({:?}, {:?}) after {reason:?}", + next.slot, next.mode, sup.strategy + ), + )); + return; + } + } + } +} + +/// V1/V2/V5: in a tree that neither the driver closed nor failed, every +/// restart the contract requires happened by the time teardown found it +/// settled: a published Permanent instance, or a published Transient one +/// that exited abnormally, has a successor; so does every live, +/// non-Temporary downstream sibling (its cohort). A missing restart means +/// the budget was exhausted, which must fail the supervisor (V5). +fn required_restarts( + s: &State, + sup: &SupInst, + slots: &[Vec<&Call>], + found: &mut Vec<(&'static str, String)>, +) { + let (Some(settled), None) = (sup.settled, &sup.failure) else { + return; + }; + let after = |j: usize, tick: u64| slots[j].iter().any(|c| c.seq > tick); + for (k, attempts) in slots.iter().enumerate() { + for c in attempts { + let Some((i, x)) = instance(s, c) else { + continue; + }; + let Some(dropped) = x.dropped.filter(|d| *d < settled) else { + continue; + }; + let required = x.published + && match c.mode { + RestartMode::Permanent => true, + RestartMode::Transient => abnormal(&x.reason), + RestartMode::Temporary => false, + }; + if !required { + continue; + } + if !after(k, dropped) { + found.push(( + "V5: a required restart did not happen and the supervisor did not fail", + format!("{} w{i} {:?} ({:?})", c.slot, x.reason, c.mode), + )); + return; + } + // Downstream siblings running untouched when this one ended. + // (If this instance was itself collateral, its round began + // upstream and already stopped them: they are not running.) + for (j, others) in slots.iter().enumerate() { + if !(sup.strategy == SupervisorStrategy::RestForOne && j > k + || sup.strategy == SupervisorStrategy::OneForAll && j != k) + { + continue; + } + let live = others.iter().any(|o| { + o.mode != RestartMode::Temporary + && o.seq < dropped + && instance(s, o).is_some_and(|(_, y)| { + y.published + && y.stopped_begin.is_none_or(|b| b > dropped) + && y.dropped.is_none_or(|d| d > dropped) + && (o.mode == RestartMode::Permanent || !could_stop_itself(s, y)) + }) + }); + if live && !after(j, dropped) { + found.push(( + "V2: a sibling in the failed child's cohort was not restarted", + format!("{} after {} w{i} ({:?})", slot_name(sup, j), c.slot, sup.strategy), + )); + return; + } + } + } + } +} + +fn slot_name(sup: &SupInst, j: usize) -> String { + format!("r{}/c{j}", sup.root) +} + +/// V5: a supervisor fails for its restart budget only when the window was +/// full. The exhausting reservation was refused at some moment between the +/// end of that child's last attempt and the supervisor's last factory call +/// (a collateral child reserves only after its round's cleanup, and a +/// superseding round can make it reserve again): at one of those moments the +/// grants in the window plus the reservations siblings can hold (at most one +/// each) reach `max_restarts`. +fn budget_exhaustion( + s: &State, + sup: &SupInst, + calls: &[(usize, &Call)], + found: &mut Vec<(&'static str, String)>, +) { + let Some(failure) = &sup.failure else { return }; + let Some(rest) = failure.split("restart budget exhausted by ").nth(1) else { + return; + }; + let name: String = rest + .chars() + .take_while(|c| !c.is_whitespace() && *c != ',' && *c != ')') + .collect(); + let Some(last) = calls.iter().rev().find(|(_, c)| c.slot == name) else { + return; + }; + let Some(end) = ended_at(s, last.1) else { + return; + }; + let mut first = std::collections::HashSet::new(); + let grants: Vec = calls + .iter() + .filter(|(_, c)| !first.insert(c.index)) + .map(|(_, c)| c.at) + .collect(); + let in_window = |t: Duration| { + grants + .iter() + .filter(|g| **g <= t && t < **g + sup.window) + .count() + }; + let most = std::iter::once(end) + .chain(grants.iter().copied().filter(|g| *g >= end)) + .map(in_window) + .max() + .unwrap_or(0); + let siblings = sup.children.saturating_sub(1); + if most + siblings < sup.max_restarts { + found.push(( + "V5: the restart budget was reported exhausted before the window was full", + format!( + "{name}: at most {most} grants + {siblings} siblings < {} in {:?}", + sup.max_restarts, sup.window + ), + )); + } +} + +/// V6, OneForOne only: the delay between the end of an attempt and the next +/// factory call is the backoff bound for the child's failure streak (from +/// zero through it with jitter). The streak counts restarts and resets after +/// a generation that stayed ready for `stable_after`; it is tracked as an +/// interval where stability is too close to call. +fn backoff_delays( + s: &State, + sup: &SupInst, + slots: &[Vec<&Call>], + found: &mut Vec<(&'static str, String)>, +) { + let b = &sup.backoff; + let bound = |failures: u32| { + b.initial + .saturating_mul(1u32 << failures.min(31)) + .min(b.maximum) + }; + let margin = Duration::from_millis(1); + for attempts in slots { + let (mut lo, mut hi) = (0u32, 0u32); + for pair in attempts.windows(2) { + let (prev, next) = (pair[0], pair[1]); + let Some(end) = ended_at(s, prev) else { break }; + match instance(s, prev).map(|(_, x)| (x.started_end_at, x.stopped_begin_at)) { + Some((Some(ready), Some(stop))) => { + let up = stop.saturating_sub(ready); + if up >= b.stable_after + margin { + (lo, hi) = (0, 0); + } else if up + margin > b.stable_after { + lo = 0; + } + } + Some((Some(_), None)) => lo = 0, + _ => {} + } + let delay = next.at.saturating_sub(end); + let min = if b.jitter { Duration::ZERO } else { bound(lo) }; + let max = bound(hi); + if delay < min || delay > max + slack(max) { + found.push(( + "V6: a restart delay is outside the backoff bound for the failure streak", + format!( + "{}: {delay:?} not in [{min:?}, {max:?}] (streak {lo}..={hi}, {:?})", + next.slot, b + ), + )); + return; + } + (lo, hi) = (lo.saturating_add(1), hi.saturating_add(1)); + } + } +} + +/// A timer cancelled before a tick was due never delivers that tick: a +/// one-shot timer at most once if cancelled at or after `created + period`, +/// an interval at most once per period elapsed before the cancel. +fn timers(s: &State, found: &mut Vec<(&'static str, String)>) { + for (i, t) in s.timers.iter().enumerate() { + let Some(cancelled) = t.cancelled else { + continue; + }; + let due = (cancelled.saturating_sub(t.created).as_nanos() / t.period.as_nanos()) as u64; + let allowed = if t.interval { due } else { due.min(1) }; + if t.handled > allowed { + found.push(( + "T: a cancelled timer delivered a message after its cancellation", + format!( + "timer {i} ({}, {:?}): {} handled, {allowed} due by the cancel", + if t.interval { "interval" } else { "once" }, + t.period, + t.handled + ), + )); + return; + } + } +} diff --git a/fuzz/src/world.rs b/fuzz/src/world.rs index ec41f309..d4627b6a 100644 --- a/fuzz/src/world.rs +++ b/fuzz/src/world.rs @@ -6,7 +6,7 @@ use spawned_concurrency::{ message::Message, - supervision::SupervisedChild, + supervision::{Backoff, RestartMode, SupervisedChild, SupervisorStrategy}, tasks::{Actor, Context, Handler, Recipient}, ActorId, ChildHandle, Down, Exit, ExitReason, MonitorRef, }; @@ -52,6 +52,20 @@ pub struct Inst { pub exits: Vec<(ActorId, ExitReason)>, /// Sources whose `Down`/`Exit` this instance already handled. pub ended: HashSet, + /// Virtual times of the same hooks, for the restart and backoff oracles + /// (`ready_at` and `ended_at` of the supervisor's stability check are + /// read in the same polls as `started_end` and `stopped_begin`). + pub started_end_at: Option, + pub stopped_begin_at: Option, + pub dropped_at: Option, + /// Factory call that constructed this instance (supervised only). + pub call: Option, + /// Returned by `SupervisedChild::current()`: the supervisor published it, + /// so it was ready (V4). + pub published: bool, + /// Taken in `started()`; the published reason is read after teardown. + pub handle: Option, + pub reason: Option, } #[derive(Debug, Clone)] @@ -70,6 +84,31 @@ pub struct SupInst { pub root: usize, pub max_restarts: usize, pub window: Duration, + pub strategy: SupervisorStrategy, + pub backoff: Backoff, + /// A root whose declared children are all workers (the restart oracles + /// only judge these; nested subtrees are checked by V1, V3 and V5). + pub flat: bool, + /// Children declared (slot paths `r{root}/c{i}`). + pub children: usize, + /// Virtual time of the first stop, shutdown or system close the driver + /// requested before teardown. + pub closed_at: Option, + /// Teardown found the tree settled (no child starting, backing off or + /// stopping) at this tick, before shutting it down. + pub settled: Option, + /// `Supervisor::failure()` when teardown began. + pub failure: Option, +} + +/// How a factory call ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + /// The call has not returned (or its future was dropped). + Pending, + Error, + Panic, + Instance(usize), } /// One factory invocation (worker factory or nested `configure` callback). @@ -77,8 +116,23 @@ pub struct SupInst { pub struct Call { pub inst: usize, pub slot: String, + /// Declaration index of the slot in its supervisor. + pub index: usize, pub at: Duration, - pub temporary: bool, + pub seq: u64, + pub mode: RestartMode, + pub outcome: Outcome, +} + +/// A `send_after` or `send_interval` the driver created (bug 21's oracle). +#[derive(Debug, Clone)] +pub struct TimerRec { + pub created: Duration, + pub period: Duration, + pub interval: bool, + pub cancelled: Option, + /// Timer messages a handler ran. + pub handled: u64, } #[derive(Default)] @@ -106,6 +160,7 @@ pub struct State { pub roots_done: Vec, /// Latest `SupervisedChild` of each nested worker slot. pub nested: BTreeMap>, + pub timers: Vec, pub violations: Vec, } @@ -136,7 +191,7 @@ impl State { self.may_stop.insert(id); *self.causes.entry(id).or_default() += 1; } - fn tick(&mut self) -> u64 { + pub fn tick(&mut self) -> u64 { self.seq += 1; self.seq } @@ -174,6 +229,8 @@ pub enum Act { pub struct Do { pub nonce: u64, pub lane: u64, + /// `1 + index` into [`State::timers`] for timer messages, else 0. + pub timer: usize, pub act: Act, } impl Message for Do { @@ -234,6 +291,7 @@ impl Drop for Worker { let mut s = self.world.lock(); let seq = s.tick(); s.insts[self.instance].dropped = Some(seq); + s.insts[self.instance].dropped_at = Some(now()); s.live_total -= 1; if let Some(slot) = &self.slot { *s.live.entry(slot.clone()).or_default() -= 1; @@ -261,6 +319,7 @@ impl Actor for Worker { let twice = inst.started_begin.is_some(); inst.id = Some(id); inst.started_begin = Some(seq); + inst.handle = Some(ctx.actor_ref().child_handle()); if twice { s.violate("L1: started() ran twice", format!("w{}", self.instance)); } @@ -279,6 +338,7 @@ impl Actor for Worker { let mut s = self.world.lock(); let seq = s.tick(); s.insts[self.instance].started_end = Some(seq); + s.insts[self.instance].started_end_at = Some(now()); } async fn stopped(&mut self, _ctx: &Context) { @@ -289,6 +349,7 @@ impl Actor for Worker { let (twice, after_started_panic) = (inst.stopped_begin.is_some(), inst.started_panicked); inst.stopped_begin = Some(seq); + inst.stopped_begin_at = Some(now()); if twice { s.violate("L1: stopped() ran twice", format!("w{}", self.instance)); } @@ -341,6 +402,9 @@ impl Handler for Worker { { let mut s = self.world.lock(); self.turn(&mut s, "Do"); + if msg.timer != 0 { + s.timers[msg.timer - 1].handled += 1; + } if msg.nonce != 0 { if let Some(other) = s.handled.insert(msg.nonce, self.instance) { let detail = format!("nonce {} by w{other} and w{}", msg.nonce, self.instance); @@ -473,7 +537,7 @@ pub fn now() -> Duration { /// Record a factory call for `slot` under supervisor instance `inst` and /// check that no earlier generation of the slot is still alive (V3). -pub fn factory_call(world: &World, inst: usize, slot: &str, temporary: bool) { +pub fn factory_call(world: &World, inst: usize, slot: &str, mode: RestartMode) -> usize { let mut s = world.lock(); let live = s.live.get(slot).copied().unwrap_or(0); let prefix = format!("{slot}/"); @@ -497,10 +561,28 @@ pub fn factory_call(world: &World, inst: usize, slot: &str, temporary: bool) { ); } let at = now(); + let seq = s.tick(); + let index = slot + .rsplit_once("/c") + .and_then(|(_, i)| i.parse().ok()) + .unwrap_or(0); s.calls.push(Call { inst, slot: slot.to_string(), + index, at, - temporary, + seq, + mode, + outcome: Outcome::Pending, }); + s.calls.len() - 1 +} + +/// Record how factory call `call` ended. +pub fn factory_outcome(world: &World, call: usize, outcome: Outcome) { + let mut s = world.lock(); + s.calls[call].outcome = outcome; + if let Outcome::Instance(i) = outcome { + s.insts[i].call = Some(call); + } } diff --git a/fuzz/tests/budget_reserved_after_round.rs b/fuzz/tests/budget_reserved_after_round.rs new file mode 100644 index 00000000..4091601b --- /dev/null +++ b/fuzz/tests/budget_reserved_after_round.rs @@ -0,0 +1,87 @@ +//! Generated by `cargo run --example repro -- --emit budget_reserved_after_round`. +//! +//! Was: harness false positive (fixed in the harness): `V5: the restart budget +//! was reported exhausted before the window was full`. OneForAll, budget 2: c0 +//! crashes, the round rebuilds both, c0's rebuild fails in its factory and a +//! new round supersedes the first. c1 reserved again only after c0's grant, so +//! the window at the end of c1's last attempt held no grant. The oracle now +//! takes the most grants at any moment up to the tree's last grant. +//! Strategy random, seed 8875277784326192161. +//! +//! ```text +//! Scenario { +//! seed: 8875277784326192161, +//! sched: Random, +//! ops: [ +//! Supervisor( +//! SupSpec { +//! system: 37, +//! policy: Policy { +//! strategy: 37, +//! max_restarts: 37, +//! window: 9509, +//! backoff: 37, +//! jitter: true, +//! stable: 37, +//! }, +//! children: [ +//! Worker( +//! WorkerDef { +//! restart: 37, +//! group: 37, +//! fail_mask: 37, +//! panic_mask: 37, +//! hook_mask: 37, +//! hooks: Hooks { +//! flags: 37, +//! started_sleep: 37, +//! stopped_sleep: 37, +//! }, +//! startup_timeout: 37, +//! shutdown_timeout: 37, +//! }, +//! ), +//! Worker( +//! WorkerDef { +//! restart: 37, +//! group: 37, +//! fail_mask: 37, +//! panic_mask: 37, +//! hook_mask: 138, +//! hooks: Hooks { +//! flags: 0, +//! started_sleep: 24, +//! stopped_sleep: 206, +//! }, +//! startup_timeout: 9, +//! shutdown_timeout: 170, +//! }, +//! ), +//! ], +//! }, +//! ), +//! CrashChild { +//! sup: 0, +//! child: 0, +//! how: Panic, +//! }, +//! ], +//! } +//! ``` +#![cfg(spawned_sim)] + +const INPUT: &[u8] = include_bytes!("../regressions/budget_reserved_after_round.input"); +const TAPE: &str = include_str!("../regressions/budget_reserved_after_round.tape"); + +/// The input under its own schedule and under its recorded tape. +#[test] +fn budget_reserved_after_round_tape() { + spawned_fuzz::assert_passes(INPUT, Some(TAPE)); +} + +/// The same operations under other simulator schedules. +#[spawned_macros::sim_test(seeds = 32)] +async fn budget_reserved_after_round_seeds(seed: u64) { + spawned_fuzz::init(); + spawned_fuzz::run_body(spawned_fuzz::decode(INPUT).unwrap(), seed).await; +} diff --git a/verification/planted/run.sh b/verification/planted/run.sh index 0949407a..775bbb15 100755 --- a/verification/planted/run.sh +++ b/verification/planted/run.sh @@ -19,6 +19,9 @@ # FUZZ_SECONDS (default 300); time is to the first failing input # V TLA+ trace validation of the dedicated trace tests' supervisor # lifecycles (needs nix; only a TLC rejection counts) +# K Kani proofs of the supervisor arithmetic (concurrency/src/kani_proofs.rs; +# needs cargo-kani 0.68 on PATH, or KANI_ENV naming a file to source; +# only a failed proof counts) # # Output: OUT/results.tsv (patch, layer, verdict, seconds, detail) and one log # per patch and layer. Verdicts: caught, missed, build (the patch does not @@ -34,7 +37,7 @@ while getopts "l:f:o:" opt; do l) layers=$OPTARG ;; f) fuzz_secs=$OPTARG ;; o) out=$OPTARG ;; - *) sed -n '2,26p' "$0"; exit 2 ;; + *) sed -n '2,29p' "$0"; exit 2 ;; esac done shift $((OPTIND - 1)) @@ -54,8 +57,8 @@ mkdir -p "$TMPDIR" if [[ $# -eq 0 ]]; then set -- "$here"/patches/*.patch fi -if ! git -C "$root" diff --quiet -- concurrency; then - echo "concurrency/ has local changes; refusing to plant on top of them" >&2 +if ! git -C "$root" diff --quiet -- concurrency/src; then + echo "concurrency/src has local changes; refusing to plant on top of them" >&2 exit 2 fi @@ -77,7 +80,11 @@ layer_T() { local failed failed=$(grep -E '^ [A-Za-z0-9_:]+$' "$log" | sed 's/^ *//' | sort -u) if [[ $status -eq 124 ]]; then - verdict=caught; detail="hang (killed after 900 s)" + # Name what hung: libtest reports tests running past 60 s. + local slow + slow=$(grep -oE '^test [A-Za-z0-9_:]+ has been running' "$log" | cut -d' ' -f2 | sort -u) + verdict=caught + detail="hang (killed after 900 s): $(echo "$slow" | head -4 | tr '\n' ' ')" elif [[ $status -ne 0 ]]; then verdict=caught detail="$(echo "$failed" | grep -c .) failing: $(echo "$failed" | head -4 | tr '\n' ' ')" @@ -162,6 +169,21 @@ layer_V() { fi } +layer_K() { + # KANI_ENV: a file to source first (a private Kani install's PATH, etc.). + timed timeout 3600 bash -c "${KANI_ENV:+. '$KANI_ENV' && }cd '$root/concurrency' && cargo kani --output-format terse" + if grep -q 'VERIFICATION:- FAILED' "$log"; then + verdict=caught + detail="$(awk '/Checking harness/ {h = $3; sub(/[.]+$/, "", h)} /VERIFICATION:- FAILED/ {print h}' "$log" | sort -u | head -4 | tr '\n' ' ')" + elif grep -q 'Complete - [0-9]* successfully verified harnesses, 0 failures' "$log"; then + verdict=missed; detail="" + elif grep -qE '^error(\[E[0-9]+\])?:' "$log"; then + verdict=build; detail="" + else + verdict=error; detail="cargo kani failed without a failed proof" + fi +} + for patch in "$@"; do name=$(basename "$patch" .patch) if [[ $patch == baseline ]]; then