test: planted-bug follow-ups (restart, budget, backoff, link and timer oracles; Kani layer; bounded waits)

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)).

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;
     });
 }
 

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") {

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()

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

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;
     });
 }

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

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))