RFC 016 Chunk 2b: per-actor messages-received counter

Tally each dequeued message against the receiving actor, surfaced as
ActorInfo.messages_received (the 'is this actor draining slower than its
mailbox fills' signal).

- messages-received, not sent (D4): the receiver counts on its own
  thread, so it's a single-writer Relaxed load+store on a hot Slot
  AtomicU64, no atomic RMW (D5); reuses the stashed *const Slot from 2a.
- Incremented at all six channel dequeue-success sites (recv,
  recv_timeout x2, recv_match, try_recv_match, try_recv) via
  preempt::note_message_received; no-op outside an actor (null slot).
- Resets with overruns in reset_counters across the three lifecycle
  sites (D7).

Matrix: debug default + rq-mpmc + rq-striped + release green; loom
slot_state/run_queue models pass (the 3 send_after_to loom failures are
pre-existing at fc014c4 — plain #[test]s under --cfg loom, not Chunk 2).

src/channel.rs

@@ -170,6 +170,7 @@ impl<T> Receiver<T> {
             {
                 let mut g = self.inner.lock();
                 if let Some(v) = g.queue.pop_front() {
+                    crate::preempt::note_message_received();
                     return Ok(v);
                 }
                 if g.senders == 0 {
@@ -229,6 +230,7 @@ impl<T> Receiver<T> {
         {
             let mut g = self.inner.lock();
             if let Some(v) = g.queue.pop_front() {
+                crate::preempt::note_message_received();
                 return Ok(v);
             }
             if g.senders == 0 {
@@ -255,6 +257,7 @@ impl<T> Receiver<T> {
         crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
         let mut g = self.inner.lock();
         if let Some(v) = g.queue.pop_front() {
+            crate::preempt::note_message_received();
             return Ok(v);
         }
         if g.senders == 0 {
@@ -282,6 +285,7 @@ impl<T> Receiver<T> {
                 let mut g = self.inner.lock();
                 if let Some(i) = g.queue.iter().position(|v| pred(v)) {
                     // position() found it, so remove() returns Some.
+                    crate::preempt::note_message_received();
                     return Ok(g.queue.remove(i).unwrap());
                 }
                 if g.senders == 0 {
@@ -313,6 +317,7 @@ impl<T> Receiver<T> {
     {
         let mut g = self.inner.lock();
         if let Some(i) = g.queue.iter().position(|v| pred(v)) {
+            crate::preempt::note_message_received();
             return Ok(Some(g.queue.remove(i).unwrap()));
         }
         if g.senders == 0 {
@@ -326,6 +331,7 @@ impl<T> Receiver<T> {
     pub fn try_recv(&self) -> Result<Option<T>, RecvError> {
         let mut g = self.inner.lock();
         if let Some(v) = g.queue.pop_front() {
+            crate::preempt::note_message_received();
             return Ok(Some(v));
         }
         if g.senders == 0 {

src/introspect.rs

@@ -87,6 +87,10 @@ pub struct ActorInfo {
     /// many times the actor was preempted for exceeding its slice. Resets on
     /// restart (per-incarnation, D7).
     pub overruns: u64,
+    /// Messages this actor has received (dequeued) this incarnation (RFC 016
+    /// Chunk 2) — answers "is this actor a hotspot / draining slower than its
+    /// mailbox fills." Counts received, not sent (D4). Per-incarnation (D7).
+    pub messages_received: u64,
 }
 
 /// A whole-runtime snapshot. See the module docs for the D2 tearing model.
@@ -165,6 +169,7 @@ fn read_slot(slot: &Slot, idx: u32, mail: Option<&MailboxInfo>) -> Option<ActorI
 
     // Counters are hot-region atomics, read lock-free (RFC 016 Chunk 2).
     let overruns = slot.overruns();
+    let messages_received = slot.messages_received();
 
     // Names + depth belong to this incarnation only if the registry mailbox's
     // pid matches the slab generation; a stale registry entry (dead prior
@@ -185,6 +190,7 @@ fn read_slot(slot: &Slot, idx: u32, mail: Option<&MailboxInfo>) -> Option<ActorI
         joiners,
         mailbox_depth,
         overruns,
+        messages_received,
     })
 }
 

src/preempt.rs

@@ -114,6 +114,19 @@ fn note_overrun() {
     }
 }
 
+/// Tally one received message against the on-CPU actor (RFC 016 Chunk 2),
+/// called from the channel receive path on each successful dequeue. A no-op
+/// outside an actor (null slot). One TLS load + one Relaxed load/store on a
+/// cache line the receiving thread already owns — no atomic RMW, no lock. Same
+/// slot-lifetime safety argument as `note_overrun`.
+#[inline]
+pub(crate) fn note_message_received() {
+    let p = CURRENT_SLOT.with(|c| c.get());
+    if !p.is_null() {
+        unsafe { (*p).record_message() };
+    }
+}
+
 /// Observation point for cooperative cancellation. If the on-CPU actor has
 /// been flagged for stop, raise the sentinel panic so the trampoline's
 /// `catch_unwind` tears the stack down (running Drop) and reports

src/runtime.rs

@@ -434,6 +434,11 @@ pub(crate) struct Slot {
     /// than `SlotCold` so the increment needs no lock; reset across reuse like
     /// every other slot field (`vacant` / `reclaim_slot` / `install_actor`).
     overruns: AtomicU64,
+    /// RFC 016 Chunk 2 — messages this actor has received (dequeued). Same
+    /// single-writer discipline as `overruns`: only the receiving actor, on its
+    /// own thread, increments it on the receive path (D4/D5), Relaxed load+store
+    /// with no RMW; snapshot reads Relaxed.
+    messages_received: AtomicU64,
     /// Cold lifecycle data. See [`SlotCold`].
     pub(crate) cold: RawMutex<SlotCold>,
 }
@@ -446,6 +451,7 @@ impl Slot {
             stop_ptr: AtomicPtr::new(std::ptr::null_mut()),
             closure: AtomicPtr::new(std::ptr::null_mut()),
             overruns: AtomicU64::new(0),
+            messages_received: AtomicU64::new(0),
             cold: RawMutex::new(SlotCold {
                 actor: None,
                 waiters: Vec::new(),
@@ -490,6 +496,21 @@ impl Slot {
         self.overruns.load(Ordering::Relaxed)
     }
 
+    /// Tally one received (dequeued) message. Same single-writer Relaxed
+    /// discipline as `record_overrun`; called by the receiving actor on its own
+    /// thread, so no RMW.
+    #[inline]
+    pub(crate) fn record_message(&self) {
+        let v = self.messages_received.load(Ordering::Relaxed);
+        self.messages_received.store(v.wrapping_add(1), Ordering::Relaxed);
+    }
+
+    /// Read the received-message tally (Relaxed; cross-thread snapshot read).
+    #[inline]
+    pub(crate) fn messages_received(&self) -> u64 {
+        self.messages_received.load(Ordering::Relaxed)
+    }
+
     /// Zero the per-actor introspection counters. Called at every point a slot
     /// is recycled or freshly occupied (`reclaim_slot`, `install_actor`) so a
     /// reused slot never carries a previous incarnation's counts — the standing
@@ -497,6 +518,7 @@ impl Slot {
     #[inline]
     pub(crate) fn reset_counters(&self) {
         self.overruns.store(0, Ordering::Relaxed);
+        self.messages_received.store(0, Ordering::Relaxed);
     }
 
     /// A pid's-eye snapshot of the slot. Cold paths re-read this under the

tests/introspect.rs

@@ -235,6 +235,7 @@ fn tree_from_nests_children_and_reroots_orphans() {
         joiners: 0,
         mailbox_depth: 0,
         overruns: 0,
+        messages_received: 0,
     };
 
     let snap = RuntimeSnapshot {
@@ -286,3 +287,37 @@ fn overrun_count_increments_on_forced_preemption() {
         h.join().unwrap();
     });
 }
+
+#[test]
+fn messages_received_counts_dequeues() {
+    const MQ: Name<u64> = Name::new("mq");
+    const N: u64 = 5;
+    run(|| {
+        let (ready_tx, ready_rx) = channel::<()>();
+        let (done_tx, done_rx) = channel::<()>();
+        let (gate_tx, gate_rx) = channel::<()>();
+        let (cmd_tx, cmd_rx) = channel::<u64>();
+
+        let h = spawn(move || {
+            register(MQ, cmd_tx).unwrap();
+            ready_tx.send(()).unwrap(); // sends don't count toward received
+            for _ in 0..N {
+                cmd_rx.recv().unwrap(); // each dequeue tallies one
+            }
+            done_tx.send(()).unwrap();
+            gate_rx.recv().unwrap(); // happens only after we've checked
+        });
+        ready_rx.recv().unwrap();
+
+        for i in 0..N {
+            send(MQ, i).unwrap();
+        }
+        done_rx.recv().unwrap(); // worker has drained all N
+
+        let info = actor_info(h.pid()).expect("worker present");
+        assert_eq!(info.messages_received, N, "one tally per dequeued message");
+
+        gate_tx.send(()).unwrap();
+        h.join().unwrap();
+    });
+}