fix(runtime): recover initial-process final-slice signals

CHANGELOG.md

@@ -13,6 +13,19 @@ All notable milestones for the nera project are documented in this file.
   covering message and timeout wakes, tail position, deeper nesting and
   successive inner receives.
 
+## Initial process handles signals queued during its final slice
+
+- When the initial entry function returns with links still held, the scheduler
+  now checks its inbox after publishing the wait. A message or exit signal sent
+  while the process was still running no longer remains stranded and hangs
+  program termination.
+- EXIT fragments carry their signal classification in the same inbox push.
+  A worker that wakes itself before the sender's wake attempt now handles a
+  fatal exit as an exit, rather than completing normally.
+- Runtime regressions exercise both deliveries through the worker loop,
+  including a sender paused after its push, and both inbox drain paths. An
+  empty-inbox control retains the initial process's wait and links.
+
 ## Oversized binary copies raise Erlang errors
 
 - `binary:copy/2` checks result-size multiplication and the binary header's

crates/nera-runtime/src/ffi/process.rs

@@ -1893,8 +1893,7 @@ pub unsafe extern "C" fn nera_rt_exit_2(ctx: *mut ProcessContext, pid_term: u64,
                     heap.alloc_tuple(&[exit_atom, from_pid, ErlTerm(reason)])
                 });
                 let fragment = crate::message::MessageFragment::from_term(exit_msg);
-                entry.inbox.push(fragment);
-                entry.has_exit_signals.store(true, std::sync::atomic::Ordering::Release);
+                entry.inbox.push(crate::message::MessageFragment::ExitSignal(Box::new(fragment)));
                 // Try waking the process. try_wake() handles receive-waiting
                 // (ASTATUS_WAITING), try_wake_io() handles IO-waiting
                 // (ASTATUS_IO_WAIT). Only one CAS can succeed.

crates/nera-runtime/src/message.rs

@@ -161,6 +161,10 @@ pub struct HeapFragment {
 /// inbox (`SegQueue`). The receiving worker drains fragments into the process
 /// mailbox under its own Mutex.
 pub enum MessageFragment {
+    /// EXIT classification travels with the payload in the same queue push.
+    /// The worker marks the mailbox for signal handling when it drains this,
+    /// before a self-wake can treat the payload as ordinary mail.
+    ExitSignal(Box<MessageFragment>),
     /// Immediate value (atom, small int, pid, nil) — no heap needed.
     Immediate(ImmediateTerm),
     /// Heap-backed term with its own mini-heap (boxed to shrink the enum).
@@ -309,6 +313,7 @@ impl MessageFragment {
     /// Consumes the fragment.
     pub fn copy_to_heap(self, target: &mut ProcessHeap) -> ErlTerm {
         match self {
+            MessageFragment::ExitSignal(signal) => signal.copy_to_heap(target),
             MessageFragment::Immediate(term) => term.get(),
             MessageFragment::RefcBinary(range) => {
                 // Move the validated window and its Arc without another check or clone.

crates/nera-runtime/src/process.rs

@@ -320,9 +320,9 @@ pub struct ProcessEntry {
     /// Atomic mirror of `ProcessStatus` for lock-free wake detection.
     /// Senders CAS this from `WAITING → READY` to wake a process.
     atomic_status: AtomicU32,
-    /// Set when an EXIT signal is pushed to the inbox. The worker checks
-    /// this before scanning the mailbox for EXIT tuples, avoiding O(N)
-    /// scans on every wake when no signals are pending.
+    /// Set by the worker when it drains a classified EXIT fragment. The
+    /// classification is queued with the payload, never published separately
+    /// by the sender. Avoids O(N) mailbox scans when no signals are pending.
     pub has_exit_signals: AtomicBool,
     /// The actual process state, protected by a Mutex.
     proc: Mutex<Process>,
@@ -580,8 +580,10 @@ fn drain_inbox_into(entry: &ProcessEntry, proc: &mut Process) {
                 }
             }
             other => {
+                let is_exit = matches!(&other, MessageFragment::ExitSignal(_));
                 let msg = other.copy_to_heap(&mut proc.heap);
                 proc.mailbox.send(msg);
+                if is_exit { entry.has_exit_signals.store(true, Ordering::Relaxed); }
             }
         }
         return;
@@ -610,8 +612,10 @@ fn drain_inbox_into(entry: &ProcessEntry, proc: &mut Process) {
                 }
             }
             other => {
+                let is_exit = matches!(&other, MessageFragment::ExitSignal(_));
                 let msg = other.copy_to_heap(&mut proc.heap);
                 proc.mailbox.send(msg);
+                if is_exit { entry.has_exit_signals.store(true, Ordering::Relaxed); }
             }
         }
     }
@@ -3593,6 +3597,39 @@ mod tests {
     // ASTATUS_IO_WAIT and try_wake / try_wake_io separation
     // ---------------------------------------------------------------
 
+    // EXIT publication must be indivisible even if its sender stops before
+    // waking the target. Exercise both the single and batched drain paths.
+    #[test]
+    fn inbox_drain_keeps_exit_classification_with_its_payload() {
+        use crate::message::{FragmentElem, ImmediateTerm};
+        for ordinary_count in [0, 1, 2] {
+            let entry = ProcessEntry::new(Process::new(ProcessId::new(1)));
+            let ordinary: Vec<_> = (0..ordinary_count).map(ErlTerm::small_int).collect();
+            for &msg in &ordinary {
+                entry.inbox.push(MessageFragment::Immediate(ImmediateTerm::new(msg)));
+            }
+            entry.inbox.push(MessageFragment::ExitSignal(Box::new(
+                MessageFragment::SimpleTuple(vec![
+                    FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom(
+                        crate::atom_table::ATOM_EXIT_SIGNAL))),
+                    FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::pid(ProcessId::new(2)))),
+                    FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom(
+                        crate::atom_table::ATOM_BADARG))),
+                ]))));
+            // No sender-side flag store or wake has occurred.
+            assert!(!entry.has_exit_signals.load(Ordering::Relaxed));
+            let mut proc = entry.lock_for_runtime();
+            proc.drain_inbox();
+            assert!(entry.has_exit_signals.swap(false, Ordering::Acquire));
+            // SAFETY: the guard exclusively owns every copied mailbox term;
+            // no collection or mutation intervenes before signal processing.
+            assert_eq!(unsafe { process_exit_signals(&mut proc) },
+                Some(ErlTerm::atom(crate::atom_table::ATOM_BADARG)));
+            assert_eq!(proc.mailbox.messages().collect::<Vec<_>>(), ordinary);
+            assert!(entry.inbox.is_empty());
+        }
+    }
+
     #[test]
     fn guard_drains_and_publishes_only_its_own_entry() {
         use crate::message::ImmediateTerm;

crates/nera-runtime/src/scheduler.rs

@@ -724,8 +724,7 @@ fn propagate_exit_signals(
             };
             drop(proc);
 
-            entry.inbox.push(fragment);
-            entry.has_exit_signals.store(true, Ordering::Release);
+            entry.inbox.push(MessageFragment::ExitSignal(Box::new(fragment)));
             // Try waking the process. try_wake() handles receive-waiting,
             // try_wake_io() handles IO-waiting. Only one CAS can succeed.
             // The worker that runs the process cancels its receive timer;
@@ -1594,12 +1593,15 @@ fn worker_loop(
                     // from a linked process or when the runtime shuts down.
                     let is_initial = proc.id == initial_pid.load(Ordering::Relaxed);
                     if is_initial && !proc.otp.links.is_empty() {
-                        proc.transition(ProcessState::Waiting(WaitReason::Receive {
-                            timeout_deadline_ms: None,
-                        }));
-                        runnable_count.fetch_sub(1, Ordering::AcqRel);
-                        drop(proc);
-                        // Do not re-inject — the process sleeps until an EXIT
+                        // A signal can arrive during the final Running slice
+                        // and fail its wake. Publish the wait before checking
+                        // the inbox, just as for an explicit receive.
+                        if publish_wait(proc, &proc_entry, ProcessState::Waiting(
+                            WaitReason::Receive { timeout_deadline_ms: None }),
+                            &timer_wheel, &runnable_count) {
+                            local.push(proc_entry);
+                        }
+                        // Otherwise the process sleeps until an EXIT
                         // signal or message wakes it, at which point it will
                         // exit normally (no more code to run).
                         continue;
@@ -2423,6 +2425,149 @@ mod tests {
         expired.len()
     }
 
+    #[derive(Clone, Copy, Debug)]
+    enum FinalSliceDelivery {
+        Message,
+        Exit,
+        ExitBeforeSenderWake,
+        Empty,
+    }
+
+    // The initial entry returns while it still owns a link. Deliver in its
+    // actual Running slice, after the worker's inbox drain, so the sender's
+    // wake fails. The worker must publish its wait and recover that wake.
+    fn check_initial_return_with_links(delivery: FinalSliceDelivery) {
+        use crate::message::{FragmentElem, ImmediateTerm, MessageFragment};
+        use std::cell::RefCell;
+
+        thread_local! {
+            static FINAL_SLICE_SEND: RefCell<Option<Box<dyn FnOnce()>>> = const { RefCell::new(None) };
+        }
+        unsafe extern "C" fn returning_entry(
+            _ctx: *mut crate::process::ProcessContext, _args: *const u64, _nargs: u32,
+        ) {
+            FINAL_SLICE_SEND.with(|send| {
+                if let Some(send) = send.borrow_mut().take() { send(); }
+            });
+        }
+
+        let scheduler = Scheduler::new(1);
+        let initial_pid = scheduler.next_pid();
+        let linked_pid = scheduler.next_pid();
+        let mut initial = Process::with_entry(initial_pid, returning_entry, vec![]);
+        initial.add_link(linked_pid);
+        scheduler.spawn_process(initial);
+        let Steal::Success(initial) = scheduler.injector.steal() else {
+            panic!("initial process must be queued");
+        };
+        let mut linked = Process::new(linked_pid);
+        linked.add_link(initial_pid);
+        scheduler.spawn_process(linked);
+        let Steal::Success(linked) = scheduler.injector.steal() else {
+            panic!("linked process must be queued");
+        };
+        assert!(!publish_wait(linked.lock_for_runtime(), &linked,
+            ProcessState::Waiting(WaitReason::Receive { timeout_deadline_ms: None }),
+            &scheduler.timer_wheel, &scheduler.runnable_count));
+        scheduler.injector.push(initial.clone());
+
+        let observed = Arc::new(AtomicU32::new(0));
+        let observation = observed.clone();
+        let target = initial.clone();
+        FINAL_SLICE_SEND.with(|send| {
+            *send.borrow_mut() = Some(Box::new(move || {
+                observation.store(target.published_status(), Ordering::Release);
+                let fragment = match delivery {
+                    FinalSliceDelivery::Message => Some(MessageFragment::Immediate(
+                        ImmediateTerm::new(ErlTerm::atom(crate::atom_table::ATOM_OK)))),
+                    FinalSliceDelivery::Exit | FinalSliceDelivery::ExitBeforeSenderWake => Some(
+                        MessageFragment::ExitSignal(Box::new(MessageFragment::SimpleTuple(vec![
+                        FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom(
+                            crate::atom_table::ATOM_EXIT_SIGNAL))),
+                        FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::pid(linked_pid))),
+                        FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom(
+                            crate::atom_table::ATOM_BADARG))),
+                    ])))),
+                    FinalSliceDelivery::Empty => None,
+                };
+                if let Some(fragment) = fragment {
+                    target.inbox.push(fragment);
+                    // Record an unexpected successful sender wake without
+                    // panicking across the native entry boundary.
+                    if !matches!(delivery, FinalSliceDelivery::ExitBeforeSenderWake)
+                        && target.try_wake() { observation.store(u32::MAX, Ordering::Release); }
+                }
+            }));
+        });
+        let worker = Worker::new_fifo();
+        let stealers = Arc::new(vec![worker.stealer()]);
+        // Controlled parks bound the old hang without sleeps or retries.
+        let parks = std::cell::Cell::new(0);
+        let forced_stop = std::cell::Cell::new(false);
+        #[cfg(mio_supported)]
+        let (reactor, join) = crate::reactor::spawn_reactor();
+        worker_loop(0, worker, scheduler.injector.clone(), stealers,
+            scheduler.process_table.clone(), scheduler.shutdown.clone(),
+            scheduler.runnable_count.clone(), scheduler.alive_count.clone(),
+            scheduler.any_crash.clone(), scheduler.initial_pid.clone(), 1,
+            scheduler.registered_names.clone(), scheduler.timer_wheel.clone(),
+            scheduler.io_pool.clone(), scheduler.worker_stats[0].clone(),
+            scheduler.worker_inboxes[0].clone(), Arc::new(vec![]),
+            |_| {
+                parks.set(parks.get() + 1);
+                if parks.get() > 16 {
+                    forced_stop.set(true);
+                    scheduler.shutdown.store(true, Ordering::SeqCst);
+                }
+            },
+            #[cfg(mio_supported)]
+            reactor.clone());
+        #[cfg(mio_supported)]
+        { reactor.shutdown(); join.join().unwrap(); }
+        FINAL_SLICE_SEND.with(|send| assert!(send.borrow().is_none()));
+        eprintln!("initial return: delivery={delivery:?} state_at_send={} inbox_empty={} \
+            runnable={} forced_stop={}", observed.load(Ordering::Acquire),
+            initial.inbox.is_empty(), scheduler.runnable_count.load(Ordering::Acquire),
+            forced_stop.get());
+        assert_eq!(observed.load(Ordering::Acquire), ASTATUS_RUNNING);
+        assert!(!forced_stop.get(), "the final-slice delivery must not strand the worker");
+        assert!(initial.inbox.is_empty());
+        assert_eq!(scheduler.runnable_count.load(Ordering::Acquire), 0);
+        assert!(scheduler.shutdown.load(Ordering::Acquire));
+        assert_eq!(scheduler.any_crash.load(Ordering::Acquire),
+            matches!(delivery, FinalSliceDelivery::Exit | FinalSliceDelivery::ExitBeforeSenderWake));
+        if matches!(delivery, FinalSliceDelivery::Empty) {
+            assert_eq!(initial.published_status(), ASTATUS_WAITING);
+            assert_eq!(initial.lock_for_runtime().otp.links.len(), 1);
+        } else {
+            assert_eq!(initial.published_status(), ASTATUS_EXITED);
+            assert!(scheduler.get_process(initial_pid).is_none());
+        }
+        assert!(processes_are_blocked(&scheduler.process_table));
+    }
+
+    #[test]
+    fn initial_return_with_links_observes_final_slice_message() {
+        check_initial_return_with_links(FinalSliceDelivery::Message);
+    }
+
+    #[test]
+    fn initial_return_with_links_observes_final_slice_exit() {
+        check_initial_return_with_links(FinalSliceDelivery::Exit);
+    }
+
+    #[test]
+    fn initial_return_with_links_observes_exit_before_sender_wake() {
+        // Pause the sender after inbox publication, before its wake attempt.
+        // Classification must already be visible to the receiver's self-wake.
+        check_initial_return_with_links(FinalSliceDelivery::ExitBeforeSenderWake);
+    }
+
+    #[test]
+    fn initial_return_with_links_without_delivery_stays_waiting() {
+        check_initial_return_with_links(FinalSliceDelivery::Empty);
+    }
+
     #[test]
     fn stale_receive_timer_cannot_replace_a_later_one() {
         let scheduler = Scheduler::new(1);
@@ -2580,14 +2725,13 @@ mod tests {
                         use crate::message::{FragmentElem, ImmediateTerm, MessageFragment};
                         // exit/2 can wake the target before another process's
                         // linked exit kills it; the queued kill is now stale.
-                        linked.inbox.push(MessageFragment::SimpleTuple(vec![
+                        linked.inbox.push(MessageFragment::ExitSignal(Box::new(MessageFragment::SimpleTuple(vec![
                             FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom(
                                 crate::atom_table::ATOM_EXIT_SIGNAL))),
                             FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::pid(dead_pid))),
                             FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom(
                                 crate::atom_table::ATOM_KILL))),
-                        ]));
-                        linked.has_exit_signals.store(true, Ordering::Release);
+                        ]))));
                     }
                     assert!(if state == io { linked.try_wake_io() } else { linked.try_wake() });
                     scheduler.runnable_count.fetch_add(1, Ordering::AcqRel);