fix(registry): by_name stores the full holder Pid — a dead name heals under slot reuse

Root cause of soak20 signature 2 (refcount_test.exs 'watcher crash',
110235x fast {:error, :server_down} probes over the full await window):
by_name mapped name -> slot *index*, so a name whose holder died (no stop
path unregisters; prune is lazy) and whose slot was then re-tenanted read
as live-held: register failed NameTaken{holder: <unrelated tenant>} (which
the bridge macro's generated start() swallows -> start_server/1 reports :ok
for a server that never came up), while name resolution reached the
tenant's mailbox, missed on the message TypeId and failed fast WITHOUT
pruning — the wedge self-sustained for the tenant's lifetime. Name-
addressed send additionally judged liveness on the slot's *current*
mailbox pid, so a same-typed tenant would have received the message
(misdelivery) and a differently typed one a misleading NoChannel.

Fix: by_name: HashMap<&'static str, Pid> — every reader judges the
*stored* holder with the generation-checked live(), so a recycled slot's
tenant no longer impersonates a dead holder, and every touch (register /
whereis / resolve / send) prunes and heals a stale name. prune(index)
becomes prune_holder(pid): names bound to the holder go; the mailbox goes
only while still the holder's own (a tenant's replacement mailbox is left
untouched). Introspection matches names to mailboxes by full pid, so a
stale name never annotates a slot's new tenant.

Deterministic regression test added first and shown to fail pre-fix
(tests/stale_name_slot_reuse.rs: tiny slab forces re-tenanting; old slot
(1,0) died, tenant (1,1) took the index; register -> NameTaken pre-fix).
Post-fix it asserts the healed contract: whereis -> None (pruned), call ->
ServerDown, re-register -> Ok. Suite 33 ok-binaries, clippy gate clean.

In the wild the window opened at every splice_test teardown:
Splice.terminate -> exit_server('subtree') left the name bound; width 20
raised the re-tenant probability. Downstream (smarm_beam): install_child's
unregister-before-register workaround becomes dead code (removed there);
the #[smarm_server] macro's swallowed register error becomes truthful
idempotency (a NameTaken now really is a live holder).

src/registry.rs

@@ -217,14 +217,20 @@ pub(crate) struct MailboxInfo {
 }
 
 /// The directory. Invariant (held under the registry lock): every value in
-/// `by_name` is the index of a [`Mailbox`] present in `by_index`, and that
+/// `by_name` is the full [`Pid`] (index *and* generation) of an actor that
-/// mailbox's `pid.index()` equals the key. Stale entries (dead actors) violate
+/// published a [`Mailbox`] into `by_index` at registration time. Stale entries
-/// nothing — they are simply pruned on contact.
+/// (dead holders — including holders whose slot has since been re-tenanted by
+/// a different actor) violate nothing — they are pruned on contact, and the
+/// generation makes "dead" decidable even after slot reuse.
 pub(crate) struct Registry {
     /// `pid.index() -> the actor's mailbox`. The handle store.
     by_index: HashMap<u32, Mailbox>,
-    /// `name -> pid.index()`. Several names may map to one actor.
+    /// `name -> holder pid`. Several names may map to one actor. The full pid
-    by_name: HashMap<&'static str, u32>,
+    /// (not just the index) is load-bearing: an index alone cannot tell a dead
+    /// holder from the live actor now tenanting its recycled slot, which made
+    /// such a name read as live-held — unresolvable *and* unregisterable — and
+    /// would misdeliver to a same-typed tenant (soak20 signature 2).
+    by_name: HashMap<&'static str, Pid>,
 }
 
 impl Registry {
@@ -232,10 +238,15 @@ impl Registry {
         Self { by_index: HashMap::new(), by_name: HashMap::new() }
     }
 
-    /// Drop a dead actor's mailbox and every name that pointed at it.
+    /// Drop a dead holder's artifacts: every name bound to it, and its mailbox
-    fn prune(&mut self, index: u32) {
+    /// — but only while the mailbox is still *its own*. A recycled slot's
-        self.by_index.remove(&index);
+    /// mailbox belongs to the live tenant (publish replaces it wholesale on
-        self.by_name.retain(|_, idx| *idx != index);
+    /// pid mismatch) and is left untouched.
+    fn prune_holder(&mut self, holder: Pid) {
+        self.by_name.retain(|_, p| *p != holder);
+        if self.by_index.get(&holder.index()).is_some_and(|mb| mb.pid == holder) {
+            self.by_index.remove(&holder.index());
+        }
     }
 
     /// RFC 016 snapshot input: per-slot-index registry view — the actor's
@@ -244,12 +255,15 @@ impl Registry {
     /// under the registry Leaf; the per-channel `queued_len` takes a Channel
     /// lock, legal under the Leaf (Leaf → Channel). Carries each mailbox's full
     /// `pid` so the caller can discard a stale incarnation's entry against the
-    /// slab's live generation. Names that dangle (point at no mailbox) are
+    /// slab's live generation. Names are matched to mailboxes by *full pid*, so
-    /// dropped — they violate no invariant and get pruned on next contact.
+    /// a stale name (dead holder) still annotates the corpse's own mailbox if
+    /// that survives, but never a recycled slot's new tenant; names that attach
+    /// to no mailbox are dropped — they violate no invariant and get pruned on
+    /// next contact.
     pub(crate) fn introspect_map(&self) -> HashMap<u32, MailboxInfo> {
-        let mut names: HashMap<u32, Vec<&'static str>> = HashMap::new();
+        let mut names: HashMap<Pid, Vec<&'static str>> = HashMap::new();
-        for (&name, &idx) in &self.by_name {
+        for (&name, &pid) in &self.by_name {
-            names.entry(idx).or_default().push(name);
+            names.entry(pid).or_default().push(name);
         }
         let mut out: HashMap<u32, MailboxInfo> = HashMap::with_capacity(self.by_index.len());
         for (&idx, mb) in &self.by_index {
@@ -258,7 +272,7 @@ impl Registry {
                 idx,
                 MailboxInfo {
                     pid: mb.pid,
-                    names: names.remove(&idx).unwrap_or_default(),
+                    names: names.remove(&mb.pid).unwrap_or_default(),
                     depth: depth.min(u32::MAX as usize) as u32,
                 },
             );
@@ -276,7 +290,7 @@ impl Registry {
         let names = self
             .by_name
             .iter()
-            .filter_map(|(&n, &i)| (i == idx).then_some(n))
+            .filter_map(|(&n, &p)| (p == mb.pid).then_some(n))
             .collect();
         Some(MailboxInfo { pid: mb.pid, names, depth: depth.min(u32::MAX as usize) as u32 })
     }
@@ -315,21 +329,22 @@ pub(crate) fn register_with<M: Send + 'static>(
         if !live(inner, me) {
             return Err(RegisterError::NoProc);
         }
-        if let Some(&holder_idx) = reg.by_name.get(key) {
+        if let Some(&holder) = reg.by_name.get(key) {
-            match reg.by_index.get(&holder_idx).map(|m| m.pid) {
+            if holder == me {
-                Some(holder) if holder == me => {} // same actor: just add the channel below
+                // Same actor: just add the channel below.
-                Some(holder) if live(inner, holder) => {
+            } else if live(inner, holder) {
-                    return Err(RegisterError::NameTaken { holder });
+                return Err(RegisterError::NameTaken { holder });
-                }
+            } else {
-                Some(_) => reg.prune(holder_idx), // dead holder: free the name
+                // Dead holder: free the name (and its other stale artifacts).
-                None => {
+                // Liveness is judged against the *stored* pid, generation
-                    reg.by_name.remove(key); // dangling name: free it
+                // included — a recycled slot's live tenant no longer makes a
-                }
+                // dead name read as taken (soak20 signature 2).
+                reg.prune_holder(holder);
             }
         }
         // Publish (or extend) the mailbox with this channel, then bind the name.
         publish_channel::<M>(&mut reg, me, tx);
-        reg.by_name.insert(key, me.index());
+        reg.by_name.insert(key, me);
         Ok(())
     })
 }
@@ -394,17 +409,14 @@ pub(crate) fn install_for<M: Send + 'static>(pid: Pid, tx: Sender<M>) {
 pub fn whereis(name: &str) -> Option<Pid> {
     with_runtime(|inner| {
         let mut reg = inner.registry.lock();
-        let idx = *reg.by_name.get(name)?;
+        let pid = *reg.by_name.get(name)?;
-        match reg.by_index.get(&idx).map(|m| m.pid) {
+        if live(inner, pid) {
-            Some(pid) if live(inner, pid) => Some(pid),
+            Some(pid)
-            Some(_) => {
+        } else {
-                reg.prune(idx);
+            // Generation-checked against the stored holder: a recycled slot's
-                None
+            // live tenant reads dead here, and the stale name heals.
-            }
+            reg.prune_holder(pid);
-            None => {
+            None
-                reg.by_name.remove(name);
-                None
-            }
         }
     })
 }
@@ -431,19 +443,18 @@ pub fn lookup_as<A: Addressable>(name: &str) -> Option<Pid<A>> {
 pub(crate) fn resolve_named_sender<M: Send + 'static>(name: &str) -> Option<(Pid, Sender<M>)> {
     with_runtime(|inner| {
         let mut reg = inner.registry.lock();
-        let idx = *reg.by_name.get(name)?;
+        let pid = *reg.by_name.get(name)?;
-        let pid = match reg.by_index.get(&idx).map(|m| m.pid) {
+        if !live(inner, pid) {
-            Some(pid) if live(inner, pid) => pid,
+            // Stored-pid liveness, generation included: a name whose holder
-            Some(_) => {
+            // died is pruned (heals) even if the slot has a new tenant —
-                reg.prune(idx);
+            // previously the tenant's mailbox made the name unresolvable
-                return None;
+            // *without* pruning, wedging it for the tenant's lifetime.
-            }
+            reg.prune_holder(pid);
-            None => {
+            return None;
-                reg.by_name.remove(name);
+        }
-                return None;
+        // A live holder's mailbox is its own (publish replaces wholesale on
-            }
+        // pid mismatch, and one live actor per slot), so index lookup is safe.
-        };
+        let tx = reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>)?;
-        let tx = reg.by_index.get(&idx).and_then(Mailbox::clone_sender::<M>)?;
         Some((pid, tx))
     })
 }
@@ -454,11 +465,8 @@ pub(crate) fn resolve_named_sender<M: Send + 'static>(name: &str) -> Option<(Pid
 pub fn unregister(name: &str) -> Option<Pid> {
     with_runtime(|inner| {
         let mut reg = inner.registry.lock();
-        let idx = reg.by_name.remove(name)?;
+        let pid = reg.by_name.remove(name)?;
-        match reg.by_index.get(&idx).map(|m| m.pid) {
+        if live(inner, pid) { Some(pid) } else { None }
-            Some(pid) if live(inner, pid) => Some(pid),
-            _ => None,
-        }
     })
 }
 
@@ -476,22 +484,20 @@ pub fn send<M: Send + 'static>(name: Name<M>, msg: M) -> Result<(), SendError<M>
         // before sending (a send can unpark a receiver).
         let tx = {
             let mut reg = inner.registry.lock();
-            let idx = match reg.by_name.get(key) {
+            let pid = match reg.by_name.get(key) {
-                Some(&i) => i,
+                Some(&p) => p,
                 None => return Err(SendError::Unresolved(msg)),
             };
-            let pid = match reg.by_index.get(&idx).map(|m| m.pid) {
-                Some(pid) => pid,
-                None => {
-                    reg.by_name.remove(key);
-                    return Err(SendError::Unresolved(msg));
-                }
-            };
             if !live(inner, pid) {
-                reg.prune(idx);
+                // Stored-pid liveness (generation included). Previously this
+                // checked the slot's *current* mailbox pid, so a recycled
+                // slot's live tenant passed — and a same-typed tenant would
+                // have received the message (misdelivery), a differently
+                // typed one a misleading NoChannel.
+                reg.prune_holder(pid);
                 return Err(SendError::Unresolved(msg));
             }
-            match reg.by_index.get(&idx).and_then(Mailbox::clone_sender::<M>) {
+            match reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>) {
                 Some(tx) => tx,
                 None => return Err(SendError::NoChannel(msg)),
             }
@@ -526,7 +532,7 @@ fn send_to_pid<M: Send + 'static>(
             }
             // Our incarnation's mailbox, but the actor has died: prune + Dead.
             Some(stored) if stored == pid => {
-                reg.prune(pid.index());
+                reg.prune_holder(pid);
                 return Err(SendError::Dead(msg));
             }
             // A different incarnation (or nothing) occupies the slot: the actor

tests/stale_name_slot_reuse.rs

@@ -0,0 +1,169 @@
+//! Reproducer (soak20 signature 2, refcount_test.exs "watcher crash"):
+//! `by_name` stores only the slot *index*, so a name whose holder died — never
+//! unregistered, since no smarm stop path unregisters (prune is lazy) — and
+//! whose slot was then re-tenanted by an unrelated actor reads as *live-held*:
+//!
+//!   - `register` of the name fails `NameTaken { holder: <unrelated tenant> }`,
+//!     so the bridge's generated `start()` (a `let _ =`) silently no-ops and
+//!     `start_server/1` reports `:ok` for a server that never came up;
+//!   - a by-name `call` resolves the tenant's mailbox, misses on the message
+//!     `TypeId`, and fails `ServerDown` fast — and does NOT prune (only the
+//!     dead-holder and dangling-name arms prune), so the name never heals
+//!     while the tenant lives. The wedge is self-sustaining.
+//!
+//! Wild signature: 110235x fast `{:error, :server_down}` probes over the full
+//! 5 s await window after a swallowed restart (200-run width-20 soak, run 59).
+//!
+//! The test asserts the *contract*: after its holder dies, a name must be
+//! re-registrable regardless of what happened to the slot. Red pre-fix.
+
+use smarm::{
+    call, init, request_stop, whereis, CallError, Config, GenServer, GenServerBuilder,
+    GenServerName, RegisterError,
+};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+const TARGET: GenServerName<Target> = GenServerName::new("stale_reuse_target");
+
+/// The named server whose death opens the window. Trivial on purpose.
+struct Target;
+
+impl GenServer for Target {
+    type Call = ();
+    type Reply = ();
+    type Cast = ();
+    type Info = ();
+    type Timer = ();
+
+    fn handle_call(&mut self, _req: ()) {}
+    fn handle_cast(&mut self, _op: ()) {}
+}
+
+/// The unrelated tenant. A *different* server type, so its mailbox holds a
+/// different `Envelope` `TypeId` — a same-typed tenant would make the by-name
+/// `call` *deliver to the wrong server* instead of failing, which is the same
+/// root hole wearing a worse hat.
+struct Filler;
+
+impl GenServer for Filler {
+    type Call = ();
+    type Reply = ();
+    type Cast = ();
+    type Info = ();
+    type Timer = ();
+
+    fn handle_call(&mut self, _req: ()) {}
+    fn handle_cast(&mut self, _op: ()) {}
+}
+
+#[derive(Debug)]
+struct Observed {
+    old_slot: (u32, u32),
+    tenant_slot: (u32, u32),
+    /// `whereis` of the dead name after re-tenanting — `Some` is the misread.
+    whereis_after_reuse: Option<(u32, u32)>,
+    /// By-name call after re-tenanting — the wild `server_down` fast-fail.
+    call_after_reuse: Result<(), CallError>,
+    /// The contract under test: re-registering the dead name.
+    restart: Result<(), RegisterError>,
+}
+
+#[test]
+fn dead_name_with_reused_slot_must_be_re_registrable() {
+    let out: Arc<Mutex<Option<Observed>>> = Arc::new(Mutex::new(None));
+    let out_w = out.clone();
+
+    // A deliberately tiny slab forces prompt slot recycling: with every filler
+    // held alive, the freed slot is the only *recycled* one, so a filler lands
+    // on it deterministically well before the slab (a loud panic) runs out.
+    init(Config::exact(2).max_actors(32)).run(move || {
+        // 1. Named server up; record its slot.
+        let target = GenServerBuilder::new(Target)
+            .named(TARGET)
+            .start()
+            .expect("name should be free at test start");
+        let old_pid = target.pid();
+
+        // 2. Kill it WITHOUT unregistering (no stop path does). Death is
+        //    confirmed via the *ref*, never the name — a by-name resolve of a
+        //    dead-but-not-yet-reused holder takes the prune arm and heals the
+        //    name, destroying the precondition.
+        request_stop(old_pid);
+        loop {
+            match target.call(()) {
+                Err(CallError::ServerDown) => break,
+                Ok(()) => smarm::sleep(Duration::from_millis(5)),
+            }
+        }
+        drop(target);
+
+        // 3. Re-tenant the slot: spawn fillers (all kept alive) until one
+        //    lands on the old index.
+        let mut fillers = Vec::new();
+        let mut tenant = None;
+        for i in 0..24 {
+            let name: &'static str = Box::leak(format!("stale_filler_{i}").into_boxed_str());
+            let f = GenServerBuilder::new(Filler)
+                .named(GenServerName::<Filler>::new(name))
+                .start()
+                .expect("filler names are fresh");
+            let fp = f.pid();
+            fillers.push(f);
+            if fp.index() == old_pid.index() {
+                tenant = Some(fp);
+                break;
+            }
+        }
+        let tenant = tenant.expect(
+            "precondition: the freed slot must be re-tenanted within the tiny slab \
+             (slots are recycled; every filler is held alive)",
+        );
+
+        // 4. Observe the poisoned state through the same paths the bridge uses.
+        let whereis_after_reuse = whereis(TARGET.as_str()).map(|p| (p.index(), p.generation()));
+        let call_after_reuse = call(TARGET, ());
+        let restart = GenServerBuilder::new(Target)
+            .named(TARGET)
+            .start()
+            .map(|_fresh_ref| ());
+
+        *out_w.lock().unwrap() = Some(Observed {
+            old_slot: (old_pid.index(), old_pid.generation()),
+            tenant_slot: (tenant.index(), tenant.generation()),
+            whereis_after_reuse,
+            call_after_reuse,
+            restart,
+        });
+
+        drop(fillers);
+    });
+
+    let o = out.lock().unwrap().take().expect("run body completed");
+    eprintln!("observed: {o:?}");
+
+    assert!(
+        o.restart.is_ok(),
+        "re-registering '{}' after its holder died failed with {:?}: the dead name \
+         reads as held by the live, unrelated tenant {:?} because by_name kept only \
+         the slot index (old slot {:?}). This is the silent-no-op start_server path \
+         of soak20 signature 2.",
+        TARGET.as_str(),
+        o.restart,
+        o.tenant_slot,
+        o.old_slot,
+    );
+
+    // The healed semantics around the re-register: the stale name reads
+    // *unbound* (never the tenant), and a by-name call fails ServerDown rather
+    // than resolving anything of the tenant's.
+    assert_eq!(
+        o.whereis_after_reuse, None,
+        "whereis of a dead name must prune and report unbound, not the slot's new tenant",
+    );
+    assert_eq!(
+        o.call_after_reuse,
+        Err(CallError::ServerDown),
+        "a by-name call to a dead name must fail ServerDown",
+    );
+}