feat: exit signals, startup and runtime-drop semantics as in OTP (ranks 5e, 5f, 5g)

.cargo/audit.toml

@@ -5,6 +5,7 @@ ignore = [
     "RUSTSEC-2026-0194", # quick-xml via pprof/inferno (private profiling-experiments feature)
     "RUSTSEC-2026-0195", # quick-xml via pprof/inferno (private profiling-experiments feature)
     "RUSTSEC-2025-0057", # fxhash via monoio (private io-experiments feature)
+    "RUSTSEC-2026-0441", # pin-utils via loom's futures feature (Loom dev-dependency only)
 ]
 informational_warnings = ["unmaintained", "unsound"]
 

.github/workflows/ci.yml

@@ -37,7 +37,7 @@ jobs:
         run: cargo test --locked --verbose
       - name: Run capability examples
         run: |
-          for example in execution_groups execution_profiles supervision_recovery supervision_tree dedicated_thread runtime_isolation byte_budget_pipeline timeout_outcomes graceful_drain protocol_timeouts streams_and_recipients registrations_and_fakes native_lifecycle coordinated_shutdown test_kit; do
+          for example in execution_groups execution_profiles supervision_recovery supervision_tree dedicated_thread runtime_isolation byte_budget_pipeline timeout_outcomes graceful_drain protocol_timeouts streams_and_recipients registrations_and_fakes native_lifecycle coordinated_shutdown test_kit exit_signals; do
             timeout 60s cargo run --locked -p spawned-capability-examples --bin "$example"
           done
           timeout 60s cargo run --locked -p spawned-capability-examples --features metrics --bin framework_metrics

.github/workflows/verification.yml

@@ -62,7 +62,7 @@ jobs:
       - name: Record supervisor lifecycle traces
         env:
           LIFECYCLE_TRACE_DIR: ${{ github.workspace }}/lifecycle-traces
-        run: cargo test --locked -p spawned-concurrency --features lifecycle-trace,failpoints --test supervision_trace --test cohort_trace --test tree_trace --test nested_supervision --test coverage_trace
+        run: cargo test --locked -p spawned-concurrency --features lifecycle-trace,failpoints --test supervision_trace --test cohort_trace --test tree_trace --test nested_supervision --test coverage_trace --test job_trace --test reply_trace --test lifecycle_trace
       - name: Validate traces and negative controls
         run: nix run .#check-tla -- --traces-only lifecycle-traces --negative-controls
       # Which model actions, projected states and transitions the recorded
@@ -172,6 +172,10 @@ jobs:
         env:
           RUSTFLAGS: --cfg spawned_scheduler_loom
         run: cargo test --locked -p spawned-concurrency --no-default-features --test mailbox_admission_loom
+      - name: Model-check the reply slot, job core, link table and monitor fence
+        env:
+          RUSTFLAGS: --cfg spawned_scheduler_loom
+        run: cargo test --locked --release -p spawned-concurrency --no-default-features --test reply_slot_loom --test registry_turn_loom --test job_core_loom --test link_fence_loom
 
   # Data races in the runtime and its dependencies. The crates forbid
   # unsafe code, so ThreadSanitizer is the sanitizer that matters here.

README.md

@@ -144,7 +144,7 @@ The return type on each protocol method determines the message kind:
 - `Result<(), ActorError>` — fire-and-forget send (both modes), returns the send result
 - No return / `-> ()` — fire-and-forget send (both modes), discards the send result
 
-**Actors** implement message handlers with `#[actor]`. Each handler method is annotated with `#[request_handler]`, `#[send_handler]`, or `#[handler]` and receives a single message struct plus a `Context`. The macro generates the `Actor` trait impl and one `Handler<M>` impl per method.
+**Actors** implement message handlers with `#[actor]`. Each handler method is annotated with `#[request_handler]` (answers with its return value), `#[send_handler]` (returns `()`) or `#[deferred_handler]` (answers through its `PendingReply`, now or later) and receives a single message struct plus a `Context`. The macro generates the `Actor` trait impl and one `Handler<M>` impl per method.
 
 **Using an actor** is just calling trait methods on its `ActorRef`. Because `#[protocol]` generates `impl NameServerProtocol for ActorRef<NameServer>`, you write `ns.add(...)` and `ns.find(...)` — the macro handles message construction and mailbox dispatch behind the scenes.
 

ROADMAP.md

@@ -720,7 +720,7 @@ Small, independent of the scheduler and lifecycle design, and visible to users.
 |---|---|---|
 | 5d | One semantics contract | A single `docs/design/semantics.md` stating every guarantee Spawned gives (message ordering and delivery, request outcomes, exit propagation through links and monitors, panic handling, restart and budget rules, shutdown order, timers, what `panic=abort` disables), each linked to the TLA+ invariant, test or example that checks it, and listing what is explicitly not guaranteed. Today these are spread across the guide, design contracts, TESTING.md and the models. New designs are reviewed against it. Erlang's semantics fit on a few pages; ours should too. |
 | 5e | Exit signals as in OTP (v0.8) | Align every exit-signal rule with OTP, as one design with a TLA+ model (the planned base-lifecycle model: links, monitors, `trap_exit`) and negative controls. (1) A direct `kill` stays untrappable, but the exit it causes propagates to linked actors as a distinct `Killed` reason (OTP `killed`), which they can trap. (2) `exit(Pid, normal)` semantics: a `Normal` exit signal from another actor is ignored by an actor that is not trapping exits; other reasons, including `Shutdown`, are delivered as trappable `Exit` messages to a trapping actor, and only `Kill` cannot be trapped. Today every direct stop ignores `trap_exit`. (3) A supervisor's `Shutdown` to a trapping child runs its normal termination path, as gen_server calls `terminate` on an exit from its parent. (4) Linking to or monitoring an actor that has already exited reports `NoProc` (OTP `noproc`) instead of its earlier exit reason. Breaking for exhaustive matches on `ExitReason`: pair with `#[non_exhaustive]` (as rank 11 does for `ActorError`), and update the semantics contract, migration notes and trace mapping. |
-| 5f | Restart budget counts recovery rounds (OTP alignment, v0.8) | Today every rebuilt child counts against the budget, including the siblings OneForAll and RestForOne rebuild, so those strategies exhaust a budget N times faster than OTP users expect. Count one restart per recovery round, as OTP does. Also align initial
+| 5f | Restart budget counts recovery rounds (OTP alignment, v0.8) | **Done in the 0.8 stack (#240):** a OneForAll/RestForOne recovery round takes one grant when the coordinator selects it (OTP's `add_restart` once per restart action); its rebuilds take none; a failed replacement requests a new, charged round. Before, every rebuilt child counted, so those strategies exhausted a budget N times faster than OTP users expect. Remaining: `Cohorts.tla` still abstracts budgets (`Supervision.tla`'s `BudgetBounded` covers OneForOne, where a child retry is the restart action); modelling the per-round charge with a per-child negative control is follow-up work. Also align initial
 startup: OTP aborts `start_link` on any child start error, while Spawned retries
 a Permanent or Transient child's failed factory, panic or startup timeout during
 the first start (semantics §11; found reviewing #233). Behaviour change: extend the `Supervision`/`Cohorts` TLA+ models and their Apalache inductive checks (`BudgetBounded`) to the new rule, with a negative control for counting per child, and update the Kani restart-window proofs and the guide. |

concurrency/Cargo.toml

@@ -83,7 +83,7 @@ required-features = ["backend-experiments"]
 pprof = { version = "0.15", optional = true, features = ["flamegraph"] }
 
 [target.'cfg(spawned_scheduler_loom)'.dev-dependencies]
-loom = "0.7"
+loom = { version = "0.7", features = ["futures"] }
 
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(spawned_scheduler_loom)', 'cfg(spawned_sim)', 'cfg(kani)', 'cfg(kani_slow)'] }

concurrency/benches/baseline.rs

@@ -1,9 +1,10 @@
 //! Process-isolated baseline driver; methodology in benchmarks/README.md.
 #[path = "support/args.rs"]
 mod cli;
-use spawned_concurrency::{message::Message, tasks, threads, ChildHandle};
+use spawned_concurrency::{message::Message, tasks, threads, ChildHandle, Deadline, ReplyReceiver};
 use std::{
     alloc::{GlobalAlloc, Layout, System},
+    future::IntoFuture,
     hint::black_box,
     sync::{
         atomic::{AtomicBool, AtomicU64, Ordering},
@@ -98,14 +99,24 @@ fn process(work: Work) -> Ack {
 struct Sink;
 impl tasks::Actor for Sink {}
 impl tasks::Handler<Work> for Sink {
-    async fn handle(&mut self, work: Work, _: &tasks::Context<Self>) -> Ack {
-        process(work)
+    async fn handle(
+        &mut self,
+        work: Work,
+        reply: spawned_concurrency::PendingReply<'_, Ack>,
+        _: &tasks::Context<Self>,
+    ) {
+        reply.reply(process(work));
     }
 }
 impl threads::Actor for Sink {}
 impl threads::Handler<Work> for Sink {
-    fn handle(&mut self, work: Work, _: &threads::Context<Self>) -> Ack {
-        process(work)
+    fn handle(
+        &mut self,
+        work: Work,
+        reply: spawned_concurrency::PendingReply<'_, Ack>,
+        _: &threads::Context<Self>,
+    ) {
+        reply.reply(process(work));
     }
 }
 #[derive(Clone)]
@@ -115,35 +126,47 @@ enum Target {
     Tokio(tokio::sync::mpsc::UnboundedSender<(Work, tokio::sync::oneshot::Sender<Ack>)>),
 }
 enum Reply {
-    Tasks(tokio::sync::oneshot::Receiver<Ack>),
-    Threads(mpsc::Receiver<Ack>),
+    Tasks(ReplyReceiver<Ack>),
+    Tokio(tokio::sync::oneshot::Receiver<Ack>),
+    Threads(ReplyReceiver<Ack>),
 }
 impl Target {
+    /// A raw request with no deadline: the 0.8 form of 0.7's `request_raw`
+    /// (no timeout), so the comparison with 0.7 measures the same operation.
     fn request(&self, work: Work) -> Reply {
         match self {
-            Self::Tasks(actor) => Reply::Tasks(actor.request_raw(work).expect("actor stopped")),
-            Self::Threads(actor) => Reply::Threads(actor.request_raw(work).expect("actor stopped")),
+            Self::Tasks(actor) => Reply::Tasks(
+                actor
+                    .send_request(work, Deadline::Never)
+                    .expect("actor stopped"),
+            ),
+            Self::Threads(actor) => Reply::Threads(
+                actor
+                    .send_request(work, Deadline::Never)
+                    .expect("actor stopped"),
+            ),
             Self::Tokio(tx) => {
                 let (reply, rx) = tokio::sync::oneshot::channel();
                 tx.send((work, reply))
                     .unwrap_or_else(|_| panic!("baseline stopped"));
-                Reply::Tasks(rx)
+                Reply::Tokio(rx)
             }
         }
     }
 }
 impl Reply {
     fn wait(self, rt: &tokio::runtime::Handle) -> Ack {
         match self {
-            Self::Tasks(rx) => rt.block_on(async {
+            Self::Tasks(rx) => rt
+                .block_on(rx.into_future())
+                .expect("reply timeout or dropped"),
+            Self::Tokio(rx) => rt.block_on(async {
                 tokio::time::timeout(Duration::from_secs(10), rx)
                     .await
                     .expect("reply timeout")
                     .expect("reply dropped")
             }),
-            Self::Threads(rx) => rx
-                .recv_timeout(Duration::from_secs(10))
-                .expect("reply timeout/disconnected"),
+            Self::Threads(rx) => rx.recv().expect("reply timeout/disconnected"),
         }
     }
 }

concurrency/benches/designs.rs

@@ -123,41 +123,54 @@ impl Sink {
 }
 impl tasks::Actor for Sink {}
 impl tasks::Handler<Work> for Sink {
-    async fn handle(&mut self, work: Work, _: &tasks::Context<Self>) -> Ack {
-        let start = if work.timing {
-            Instant::now()
-        } else {
-            work.enqueued
-        };
-        match self.work.as_str() {
-            "timer" => tokio::time::sleep(self.wait).await,
-            "blocking" => std::thread::sleep(self.wait),
-            "tcp" => {
-                use tokio::io::{AsyncReadExt, AsyncWriteExt};
-                if self.socket.is_none() {
-                    let socket = tokio::net::TcpStream::connect(self.address.unwrap())
-                        .await
-                        .unwrap();
-                    socket.set_nodelay(true).unwrap();
-                    self.socket = Some(socket);
+    async fn handle(
+        &mut self,
+        work: Work,
+        reply: spawned_concurrency::PendingReply<'_, Ack>,
+        _: &tasks::Context<Self>,
+    ) {
+        let __reply_value: Ack = {
+            let start = if work.timing {
+                Instant::now()
+            } else {
+                work.enqueued
+            };
+            match self.work.as_str() {
+                "timer" => tokio::time::sleep(self.wait).await,
+                "blocking" => std::thread::sleep(self.wait),
+                "tcp" => {
+                    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+                    if self.socket.is_none() {
+                        let socket = tokio::net::TcpStream::connect(self.address.unwrap())
+                            .await
+                            .unwrap();
+                        socket.set_nodelay(true).unwrap();
+                        self.socket = Some(socket);
+                    }
+                    let socket = self.socket.as_mut().unwrap();
+                    let mut frame = [0u8; 64];
+                    frame[..8].copy_from_slice(&work.id.to_le_bytes());
+                    let expected = frame;
+                    socket.write_all(&frame).await.unwrap();
+                    socket.read_exact(&mut frame).await.unwrap();
+                    assert_eq!(frame, expected);
                 }
-                let socket = self.socket.as_mut().unwrap();
-                let mut frame = [0u8; 64];
-                frame[..8].copy_from_slice(&work.id.to_le_bytes());
-                let expected = frame;
-                socket.write_all(&frame).await.unwrap();
-                socket.read_exact(&mut frame).await.unwrap();
-                assert_eq!(frame, expected);
+                _ => {}
             }
-            _ => {}
-        }
-        process_at(work, start)
+            process_at(work, start)
+        };
+        reply.reply(__reply_value);
     }
 }
 impl threads::Actor for Sink {}
 impl threads::Handler<Work> for Sink {
-    fn handle(&mut self, work: Work, _: &threads::Context<Self>) -> Ack {
-        process(work)
+    fn handle(
+        &mut self,
+        work: Work,
+        reply: spawned_concurrency::PendingReply<'_, Ack>,
+        _: &threads::Context<Self>,
+    ) {
+        reply.reply(process(work));
     }
 }
 enum ReplySender {
@@ -188,13 +201,29 @@ impl Message for Probe {
     type Result = ();
 }
 impl tasks::Handler<Probe> for Sink {
-    async fn handle(&mut self, probe: Probe, _: &tasks::Context<Self>) {
-        probe.reply.send(process(probe.work));
+    async fn handle(
+        &mut self,
+        probe: Probe,
+        reply: spawned_concurrency::PendingReply<'_, ()>,
+        _: &tasks::Context<Self>,
+    ) {
+        {
+            probe.reply.send(process(probe.work));
+        }
+        reply.reply(());
     }
 }
 impl threads::Handler<Probe> for Sink {
-    fn handle(&mut self, probe: Probe, _: &threads::Context<Self>) {
-        probe.reply.send(process(probe.work));
+    fn handle(
+        &mut self,
+        probe: Probe,
+        reply: spawned_concurrency::PendingReply<'_, ()>,
+        _: &threads::Context<Self>,
+    ) {
+        {
+            probe.reply.send(process(probe.work));
+        }
+        reply.reply(());
     }
 }
 enum TokioMessage {
@@ -264,6 +293,9 @@ enum Target {
 enum Reply {
     Tasks(tokio::sync::oneshot::Receiver<Ack>),
     Threads(mpsc::Receiver<Ack>),
+    /// A framework request's receiver (its deadline is fixed at send).
+    FrameworkTasks(spawned_concurrency::ReplyReceiver<Ack>),
+    FrameworkThreads(spawned_concurrency::ReplyReceiver<Ack>),
 }
 impl Target {
     fn request(&self, work: Work, kind: &str) -> Reply {
@@ -282,8 +314,16 @@ impl Target {
             return rx;
         }
         match self {
-            Self::Tasks(actor) => Reply::Tasks(actor.request_raw(work).expect("actor stopped")),
-            Self::Threads(actor) => Reply::Threads(actor.request_raw(work).expect("actor stopped")),
+            Self::Tasks(actor) => Reply::FrameworkTasks(
+                actor
+                    .send_request(work, Duration::from_secs(10))
+                    .expect("actor stopped"),
+            ),
+            Self::Threads(actor) => Reply::FrameworkThreads(
+                actor
+                    .send_request(work, Duration::from_secs(10))
+                    .expect("actor stopped"),
+            ),
             Self::Tokio(tx) => {
                 let (reply, rx) = tokio::sync::oneshot::channel();
                 tx.send(TokioMessage::Request(work, reply))
@@ -300,6 +340,11 @@ impl Reply {
             return match self {
                 Self::Tasks(rx) => futures::executor::block_on(rx).expect("reply dropped"),
                 Self::Threads(rx) => rx.recv().expect("reply dropped"),
+                Self::FrameworkTasks(rx) => {
+                    futures::executor::block_on(std::future::IntoFuture::into_future(rx))
+                        .expect("reply failed")
+                }
+                Self::FrameworkThreads(rx) => rx.recv().expect("reply failed"),
             };
         }
         match self {
@@ -312,6 +357,8 @@ impl Reply {
             Self::Threads(rx) => rx
                 .recv_timeout(Duration::from_secs(10))
                 .expect("reply timeout/disconnected"),
+            Self::FrameworkTasks(rx) => rt.block_on(async { rx.await.expect("reply failed") }),
+            Self::FrameworkThreads(rx) => rx.recv().expect("reply failed"),
         }
     }
 }