fix(runtime): stop immediately on fatal self-exit signals

CHANGELOG.md

@@ -2,6 +2,20 @@
 
 All notable milestones for the nera project are documented in this file.
 
+## Fatal self-exit signals stop execution immediately
+
+- `exit(self(), Reason)` now stops before the next expression when the signal
+  is fatal: `kill` even while trapping exits, or any reason while not trapping.
+  Catch handlers and `after` bodies do not run for an exit signal, as on BEAM.
+- A self-sent `normal` signal terminates a process that is not trapping exits;
+  `normal` from another process is still ignored. Trapped non-`kill` signals
+  remain messages, and `kill` still becomes the monitor reason `killed`.
+- Pending signals retain their provenance separately from message contents.
+  Ordinary or already-trapped `{'EXIT', Pid, Reason}` messages cannot become
+  fatal signals when a later exit signal arrives.
+- The `exit_self` regression runs unchanged natively and on BEAM, covering
+  direct and dynamic calls, tail calls, handlers and receive resumption.
+
 ## Stored bitstrings keep their bytes and exact bit length
 
 - ETS and `persistent_term` now own copies of non-byte-aligned bitstrings,

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

@@ -1874,7 +1874,7 @@ pub unsafe extern "C" fn nera_rt_is_process_alive(ctx: *mut ProcessContext, term
     })
 }
 
-/// erlang:exit/2 — send an exit signal to another process.
+/// erlang:exit/2 — send an exit signal, yielding immediately for a fatal self-signal.
 #[no_mangle]
 pub unsafe extern "C" fn nera_rt_exit_2(ctx: *mut ProcessContext, pid_term: u64, reason: u64) -> u64 {
     let target_pid = match ErlTerm(pid_term).as_pid_id() {
@@ -1894,6 +1894,15 @@ pub unsafe extern "C" fn nera_rt_exit_2(ctx: *mut ProcessContext, pid_term: u64,
                 });
                 let fragment = crate::message::MessageFragment::from_term(exit_msg);
                 entry.inbox.push(crate::message::MessageFragment::ExitSignal(Box::new(fragment)));
+                if target_pid == self_pid && !ctx.is_null() {
+                    let trap_exit = unsafe { (*(*ctx).otp_ptr).trap_exit };
+                    if !trap_exit || ErlTerm(reason).as_atom_id() == Some(crate::atom_table::ATOM_KILL) {
+                        // Return to the scheduler before the next expression.
+                        // It consumes the queued signal before resuming any
+                        // caller, so neither catch nor after can intercept it.
+                        unsafe { (*ctx).status = crate::process::STATUS_YIELD; }
+                    }
+                }
                 // 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,9 +161,9 @@ 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.
+    /// Pending exit/2 signal, distinct from an ordinary {'EXIT', Pid, Reason}
+    /// message. Classification travels with the payload in one queue push;
+    /// inbox draining retains it until signal processing.
     ExitSignal(Box<MessageFragment>),
     /// Immediate value (atom, small int, pid, nil) — no heap needed.
     Immediate(ImmediateTerm),
@@ -313,7 +313,6 @@ 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.
@@ -335,8 +334,8 @@ impl MessageFragment {
                 target.adopt_fragment(frag.heap);
                 frag.root
             }
-            MessageFragment::ActiveReadReady { .. } => {
-                panic!("ActiveReadReady must be intercepted by drain_inbox, not copy_to_heap");
+            MessageFragment::ActiveReadReady { .. } | MessageFragment::ExitSignal(_) => {
+                panic!("runtime notifications must be intercepted by drain_inbox, not copy_to_heap");
             }
         }
     }

crates/nera-runtime/src/process.rs

@@ -579,11 +579,14 @@ fn drain_inbox_into(entry: &ProcessEntry, proc: &mut Process) {
                     handle_active_read_on_worker(proc, stream_handle, &in_flight);
                 }
             }
+            MessageFragment::ExitSignal(signal) => {
+                let msg = signal.copy_to_heap(&mut proc.heap);
+                proc.mailbox.send_exit_signal(msg);
+                entry.has_exit_signals.store(true, Ordering::Release);
+            }
             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;
@@ -611,11 +614,14 @@ fn drain_inbox_into(entry: &ProcessEntry, proc: &mut Process) {
                     handle_active_read_on_worker(proc, stream_handle, &in_flight);
                 }
             }
+            MessageFragment::ExitSignal(signal) => {
+                let msg = signal.copy_to_heap(&mut proc.heap);
+                proc.mailbox.send_exit_signal(msg);
+                entry.has_exit_signals.store(true, Ordering::Release);
+            }
             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); }
             }
         }
     }
@@ -2137,13 +2143,13 @@ impl Process {
 // Exit signal processing
 // ---------------------------------------------------------------------------
 
-/// Scan the process mailbox for `{'EXIT', FromPid, Reason}` tuples and
-/// process them according to BEAM semantics:
+/// Process pending exit/2 signals in mailbox order, never ordinary or already
+/// trapped `{'EXIT', FromPid, Reason}` messages, according to BEAM semantics:
 ///
 /// - `kill` reason: always kill (untrappable) — returns `Some(ATOM_KILLED)`
 /// - `trap_exit=true` + non-kill: leave in mailbox as regular message
-/// - `trap_exit=false` + `normal`: silently remove (BEAM ignores normal exits)
-/// - `trap_exit=false` + non-normal, non-kill: kill — returns `Some(reason)`
+/// - `trap_exit=false` + `normal` from another process: silently remove
+/// - `trap_exit=false` + self-sent `normal` or any other reason: kill
 ///
 /// Returns `Some(exit_reason)` if the process should die.
 ///
@@ -2178,6 +2184,11 @@ pub(crate) unsafe fn process_exit_signals(proc: &mut Process) -> Option<ErlTerm>
     let buffer = proc.mailbox.buf;
     for offset in 0..slots {
         let physical = (head.wrapping_add(offset) & mask) as usize;
+        // Provenance belongs to the mailbox slot, not the tuple's contents.
+        // Once a signal is trapped it remains an ordinary message on later scans.
+        if !proc.mailbox.exit_signals.remove(&physical) {
+            continue;
+        }
         // SAFETY: each original occupied slot was initialized. This loop only
         // tombstones/removes slots: no resize or append invalidates the buffer.
         let msg = ErlTerm(unsafe { *buffer.add(physical) });
@@ -2223,8 +2234,11 @@ pub(crate) unsafe fn process_exit_signals(proc: &mut Process) -> Option<ErlTerm>
             continue;
         }
 
-        // trap_exit=false
-        if reason_atom == Some(ATOM_NORMAL) {
+        // A normal signal from another process is ignored, but an explicit
+        // exit(self(), normal) terminates a process that is not trapping exits.
+        // SAFETY: the same live three-element tuple contains its sender PID.
+        let from_pid = unsafe { ErlTerm(*ptr.add(2)) };
+        if reason_atom == Some(ATOM_NORMAL) && from_pid != ErlTerm::pid(proc.id) {
             // SAFETY: this live slot was just read from the stable scan buffer;
             // no intervening operation changes storage or removes this slot.
             unsafe {
@@ -2233,7 +2247,7 @@ pub(crate) unsafe fn process_exit_signals(proc: &mut Process) -> Option<ErlTerm>
             continue;
         }
 
-        // Non-normal, non-kill: kill the process with this reason
+        // Fatal signal, including self-sent normal: exit with this reason
         // SAFETY: this live slot belongs to the unchanged scan buffer.
         unsafe {
             proc.mailbox.remove_scanned_slot(physical);
@@ -2269,6 +2283,9 @@ pub struct Mailbox {
     scan_pos: u32,
     /// Number of tombstoned (removed) slots within `[head, head+len)`.
     tombstone_count: u32,
+    /// Physical slots containing pending exit/2 signals, not ordinary messages.
+    /// Empty before compiled receive code runs; GC preserves slot positions.
+    exit_signals: HashSet<usize>,
 }
 
 // Field offsets for codegen inlining.
@@ -2313,6 +2330,7 @@ impl Mailbox {
             len: 0,
             scan_pos: 0,
             tombstone_count: 0,
+            exit_signals: HashSet::new(),
         }
     }
 
@@ -2356,6 +2374,7 @@ impl Mailbox {
         assert!(!new_buf.is_null(), "allocation failed");
 
         let mut write = 0u32;
+        let mut exit_signals = HashSet::new();
         if self.len > 0 && !self.buf.is_null() {
             for i in 0..self.len {
                 let src_idx = self.physical_index(i);
@@ -2366,6 +2385,9 @@ impl Mailbox {
                 unsafe {
                     *new_buf.add(write as usize) = val;
                 }
+                if self.exit_signals.contains(&src_idx) {
+                    exit_signals.insert(write as usize);
+                }
                 write += 1;
             }
         }
@@ -2381,6 +2403,7 @@ impl Mailbox {
         }
 
         self.buf = new_buf;
+        self.exit_signals = exit_signals;
         self.capacity = new_cap;
         self.head = 0;
         self.len = kept;
@@ -2422,6 +2445,12 @@ impl Mailbox {
         self.len += 1;
     }
 
+    /// Append an exit/2 signal with provenance that ordinary sends cannot forge.
+    fn send_exit_signal(&mut self, msg: ErlTerm) {
+        self.send(msg);
+        self.exit_signals.insert(self.physical_index(self.len - 1));
+    }
+
     /// Number of live messages currently in the mailbox.
     #[inline]
     pub fn len(&self) -> usize {
@@ -2472,6 +2501,7 @@ impl Mailbox {
         let msg = ErlTerm(unsafe { *self.buf.add(phys) });
         assert!(!msg.is_tombstone(), "remove_at: double-remove of tombstone");
 
+        self.exit_signals.remove(&phys);
         // Write tombstone sentinel.
         unsafe {
             *self.buf.add(phys) = crate::term::TOMBSTONE_VALUE;
@@ -2493,6 +2523,7 @@ impl Mailbox {
         debug_assert!(physical < self.capacity as usize);
         // SAFETY: caller establishes initialized, live storage within capacity.
         debug_assert!(!ErlTerm(unsafe { *self.buf.add(physical) }).is_tombstone());
+        self.exit_signals.remove(&physical);
         // SAFETY: caller establishes the writable slot; &mut self is exclusive.
         unsafe { *self.buf.add(physical) = crate::term::TOMBSTONE_VALUE; }
         self.tombstone_count += 1;
@@ -2535,12 +2566,16 @@ impl Mailbox {
         debug_assert_eq!(self.scan_pos, 0, "compact: must not run during active scan");
         let alive = self.len - self.tombstone_count;
         let mut write_pos: u32 = 0;
+        let mut exit_signals = HashSet::new();
         for read_pos in 0..self.len {
             let phys_r = self.physical_index(read_pos);
             let val = unsafe { *self.buf.add(phys_r) };
             if val != crate::term::TOMBSTONE_VALUE {
+                let phys_w = self.physical_index(write_pos);
+                if self.exit_signals.contains(&phys_r) {
+                    exit_signals.insert(phys_w);
+                }
                 if write_pos != read_pos {
-                    let phys_w = self.physical_index(write_pos);
                     unsafe {
                         *self.buf.add(phys_w) = val;
                     }
@@ -2549,6 +2584,7 @@ impl Mailbox {
             }
         }
         debug_assert_eq!(write_pos, alive);
+        self.exit_signals = exit_signals;
         self.len = alive;
         self.tombstone_count = 0;
     }
@@ -2879,7 +2915,7 @@ mod tests {
                 ErlTerm::pid(crate::process::ProcessId::new(2)),
                 ErlTerm::atom(reason),
             ]);
-            proc.mailbox.send(signal);
+            proc.mailbox.send_exit_signal(signal);
         }
         // Fill the ring so the pre-fix stale-length read cannot read uninitialized
         // slots; it still incorrectly skips the kill after removing the head.
@@ -2908,7 +2944,7 @@ mod tests {
         let signal =
             proc.heap
                 .alloc_tuple(&[ErlTerm::atom(ATOM_EXIT_SIGNAL), ErlTerm::pid(crate::process::ProcessId::new(2)), reason]);
-        proc.mailbox.send(signal);
+        proc.mailbox.send_exit_signal(signal);
         // SAFETY: all roots are immediate or initialized on this process heap,
         // held exclusively; the mailbox roots are updated by the collector.
         unsafe { proc.collect_garbage_full() };
@@ -2982,11 +3018,11 @@ mod tests {
             for _ in 0..14 {
                 proc.mailbox.remove_at(0);
             }
-            proc.mailbox.send(normal);
+            proc.mailbox.send_exit_signal(normal);
             proc.mailbox.send(ErlTerm::small_int(99));
-            proc.mailbox.send(normal);
-            proc.mailbox.send(normal);
-            proc.mailbox.send(kill);
+            proc.mailbox.send_exit_signal(normal);
+            proc.mailbox.send_exit_signal(normal);
+            proc.mailbox.send_exit_signal(kill);
             proc.mailbox.remove_at(3); // a tombstone between EXIT signals
             proc.mailbox.remove_at(0);
             proc.mailbox.remove_at(0); // head wraps during the next signal removal
@@ -3005,12 +3041,72 @@ mod tests {
             ErlTerm::atom(ATOM_NORMAL),
         ]);
         for _ in 0..3 {
-            proc.mailbox.send(normal);
+            proc.mailbox.send_exit_signal(normal);
         }
         // SAFETY: every queued tuple is initialized on this live, unmoving heap.
         assert_eq!(unsafe { process_exit_signals(&mut proc) }, None);
         assert!(proc.mailbox.is_empty());
     }
+    #[test]
+    fn exit_signal_provenance_survives_mailbox_growth_and_compaction() {
+        use crate::atom_table::{ATOM_EXIT_SIGNAL, ATOM_KILL, ATOM_KILLED, ATOM_NORMAL};
+        let mut proc = Process::new(ProcessId::new(1));
+        let ordinary = proc.heap.alloc_tuple(&[
+            ErlTerm::atom(ATOM_EXIT_SIGNAL), ErlTerm::pid(proc.id), ErlTerm::atom(ATOM_NORMAL),
+        ]);
+        let signal = proc.heap.alloc_tuple(&[
+            ErlTerm::atom(ATOM_EXIT_SIGNAL), ErlTerm::pid(proc.id), ErlTerm::atom(ATOM_KILL),
+        ]);
+        // Start with a wrapped ring so growing moves the signal's physical slot.
+        for i in 0..16 { proc.mailbox.send(ErlTerm::small_int(i)); }
+        for _ in 0..14 { proc.mailbox.remove_at(0); }
+        proc.mailbox.send(ordinary);
+        for i in 0..10 { proc.mailbox.send(ErlTerm::small_int(i)); }
+        proc.mailbox.send_exit_signal(signal);
+        for i in 0..64 { proc.mailbox.send(ErlTerm::small_int(i)); }
+        // Remove the leading padding and leave tombstones before and after the
+        // signal. Compaction must move its provenance with its message.
+        proc.mailbox.remove_at(0);
+        proc.mailbox.remove_at(0);
+        for i in 1..11 { proc.mailbox.remove_at(i); }
+        for i in 12..60 { proc.mailbox.remove_at(i); }
+        proc.mailbox.reset_scan();
+        // SAFETY: all terms are owned by this live, exclusively accessed heap.
+        assert_eq!(unsafe { process_exit_signals(&mut proc) }, Some(ErlTerm::atom(ATOM_KILLED)));
+        assert_eq!(proc.mailbox.messages().next(), Some(ordinary));
+    }
+
+    #[test]
+    fn exit_message_provenance_survives_single_and_batch_inbox_drain() {
+        use crate::atom_table::{ATOM_EXIT_SIGNAL, ATOM_KILL, ATOM_KILLED, ATOM_NORMAL, ATOM_TRUE};
+        for single_first in [false, true] {
+            let entry = ProcessEntry::new(Process::new(ProcessId::new(1)));
+            let mut proc = entry.lock_for_runtime();
+            let self_pid = ErlTerm::pid(proc.id);
+            let ordinary = proc.heap.alloc_tuple(&[
+                ErlTerm::atom(ATOM_EXIT_SIGNAL), self_pid, ErlTerm::atom(ATOM_NORMAL),
+            ]);
+            let signal = proc.heap.alloc_tuple(&[
+                ErlTerm::atom(ATOM_EXIT_SIGNAL), self_pid, ErlTerm::atom(ATOM_KILL),
+            ]);
+            // SAFETY: both source graphs remain live and unmoving on this heap.
+            entry.inbox.push(unsafe { MessageFragment::from_term(ordinary) });
+            if single_first { proc.drain_inbox(); }
+            entry.inbox.push(MessageFragment::ExitSignal(Box::new(
+                unsafe { MessageFragment::from_term(signal) }
+            )));
+            proc.drain_inbox();
+            // SAFETY: draining transferred both graphs into this locked process.
+            assert_eq!(unsafe { process_exit_signals(&mut proc) }, Some(ErlTerm::atom(ATOM_KILLED)));
+            let kept = proc.mailbox.messages().next().unwrap();
+            // SAFETY: the original and delivered tuple are still owned and live.
+            assert_eq!(unsafe {
+                crate::ffi::compare::nera_rt_exact_eq(std::ptr::null_mut(), kept.0, ordinary.0)
+            }, ErlTerm::atom(ATOM_TRUE).0);
+            assert_eq!(proc.mailbox.len(), 1);
+        }
+    }
+
     #[test]
     fn mailbox_rejects_overflowing_growth_without_mutation() {
         for active_scan in [false, true] {

crates/nera-runtime/src/scheduler.rs

@@ -724,7 +724,9 @@ fn propagate_exit_signals(
             };
             drop(proc);
 
-            entry.inbox.push(MessageFragment::ExitSignal(Box::new(fragment)));
+            // Trapping has already converted this link signal into a message.
+            // Do not reinterpret it if trap_exit changes before delivery.
+            entry.inbox.push(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;