pg: Phase 1 — identity & storage substrate (RFC 012)

Add a process-groups module (name -> multiset<Member>) sibling to the
registry, plus the cluster-shaped identity it is keyed on.

- NodeId/Incarnation u32 newtypes; Member { node, incarnation, pid } at
  final (BEAM NEW_PID_EXT-shaped) layout.
- Config::node_id/incarnation builder setters beside wake_slot, defaulting
  to a fixed single-node value; threaded through RuntimeInner::new.
- process_groups: RawMutex<ProcessGroups> on RuntimeInner, Leaf class,
  mirroring the registry field.
- remove_where: the one dumb predicate-eviction primitive (no liveness wired
  yet; first caller is the Phase 2 death hook).
- Public surface (join/leave/members/pick) in pre-liveness/pre-monitor form,
  Pid-shaped; node/incarnation filled from runtime identity.
- Structural unit tests on synthetic members: idempotent join, multiset
  semantics across groups, generation-distinct members, leave + empty-group
  pruning, predicate sweep, incarnation-sweep shape.

src/lib.rs

@@ -24,6 +24,7 @@ pub mod io;
 pub mod mutex;
 pub mod monitor;
 pub mod registry;
+pub mod pg;
 pub mod link;
 pub mod gen_server;
 pub mod runtime;
@@ -54,6 +55,7 @@ pub use link::{link, trap_exit, unlink, ExitSignal};
 pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId};
 pub use mutex::{LockTimeout, Mutex, MutexGuard};
 pub use pid::Pid;
+pub use pg::{join, leave, members, pick, Incarnation, Member, NodeId};
 pub use registry::{name_of, register, unregister, whereis, RegisterError};
 pub use runtime::{init, Config, Runtime};
 pub use scheduler::{

src/pg.rs

@@ -0,0 +1,315 @@
+//! Process groups — a `name → multiset<Member>` map (RFC 012).
+//!
+//! Sits parallel to [`registry`](crate::registry), not on top of it. The
+//! registry is a *bimap*: at most one pid per name. A group is the opposite —
+//! many pids per name, the same pid in many groups — so it cannot be a
+//! generalization of the bimap; it is its own keyspace sharing only the
+//! liveness helper and the lock *class*.
+//!
+//! ## Cleanup is eager (unlike the registry)
+//!
+//! The registry prunes a stale binding lazily, on contact, because it only
+//! ever resolves one binding at a time. A group is *iterated* — `members`
+//! fans out to every member — so it must not carry dead members across a
+//! broadcast. Removal is therefore eager and monitor-driven: each `join`
+//! installs a `monitor(pid)` and the one-shot `Down` evicts the member
+//! (wired in Phase 2). A generation-checked liveness read stays on the read
+//! path as a belt-and-braces backstop (Phase 3), matching the registry's
+//! prune-on-contact guard.
+//!
+//! ## Eviction is one dumb primitive
+//!
+//! [`ProcessGroups::remove_where`] removes every member matching a predicate
+//! from every group. The primitive never knows *why* a member leaves — that
+//! is the caller's concern. Its first caller is the monitor death hook (this
+//! RFC); the later `evict_incarnation(node, inc)` sweep (RFC 010) reuses the
+//! same predicate path, which is the whole reason to shape it as a predicate.
+//!
+//! ## Identity is cluster-shaped from the first commit
+//!
+//! A [`Member`] is not a bare [`Pid`]: it is `(NodeId, Incarnation, Pid)`, a
+//! field-for-field image of a modern BEAM pid (`NEW_PID_EXT`) so the eventual
+//! ETF codec (RFC 011) is a near-identity mapping. In this RFC `NodeId` and
+//! `Incarnation` are runtime-init constants — one node, one fixed incarnation
+//! — threaded through storage and eviction anyway so the public surface never
+//! has to change to acquire them once clustering (RFC 010) supplies real
+//! values. No cluster types leak out of the public functions: callers pass and
+//! receive [`Pid`]; the node/incarnation are filled from runtime identity.
+//!
+//! ## Locking
+//!
+//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry`
+//! exactly: never held with any other lock, and the eviction path keeps it
+//! off the send path (it does not send under the lock). Liveness checks under
+//! it read only the atomic slot word, so the leaf rule holds trivially.
+
+use crate::pid::Pid;
+use crate::scheduler::with_runtime;
+use std::collections::HashMap;
+
+/// A cluster node handle. A `u32` integer handle, *not* an interned atom — the
+/// single deliberate divergence from the BEAM wire shape (RFC 011 names it).
+#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
+pub struct NodeId(u32);
+
+impl NodeId {
+    #[inline]
+    pub const fn new(v: u32) -> Self {
+        Self(v)
+    }
+    #[inline]
+    pub const fn get(self) -> u32 {
+        self.0
+    }
+}
+
+impl From<u32> for NodeId {
+    #[inline]
+    fn from(v: u32) -> Self {
+        Self(v)
+    }
+}
+
+/// A node's incarnation epoch — the BEAM `Creation` field adopted verbatim: it
+/// separates a crashed node from its restart. Fixed for the life of a run
+/// until clustering supplies a real one.
+#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
+pub struct Incarnation(u32);
+
+impl Incarnation {
+    #[inline]
+    pub const fn new(v: u32) -> Self {
+        Self(v)
+    }
+    #[inline]
+    pub const fn get(self) -> u32 {
+        self.0
+    }
+}
+
+impl From<u32> for Incarnation {
+    #[inline]
+    fn from(v: u32) -> Self {
+        Self(v)
+    }
+}
+
+/// The fixed single-node identity used until clustering (RFC 010) supplies
+/// real values. Carried like `wake_slot` so the public API never has to change
+/// to acquire it.
+pub const DEFAULT_NODE_ID: NodeId = NodeId(0);
+/// The fixed incarnation for the single-node default. Non-zero so it never
+/// collides with a BEAM "any creation" wildcard at interop time.
+pub const DEFAULT_INCARNATION: Incarnation = Incarnation(1);
+
+/// A group member's full identity: `(node, incarnation, pid)`.
+///
+/// Deliberately a field-for-field image of a modern BEAM pid (`NEW_PID_EXT`).
+/// In memory it is a plain struct — no wire packing; the packed representation
+/// belongs to the `RemoteRef` boundary (RFC 010/011), not here.
+#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
+pub struct Member {
+    /// Which node the pid lives on. `DEFAULT_NODE_ID` while single-node.
+    pub node: NodeId,
+    /// The node's incarnation epoch at the time of joining.
+    pub incarnation: Incarnation,
+    /// Pure local slot identity — unchanged; cluster identity is layered
+    /// *around* it here rather than overloading `Pid::generation`.
+    pub pid: Pid,
+}
+
+/// The store: `name → multiset<Member>`. Within a single group a `Member`
+/// appears at most once (`join` is idempotent); the *multiset* framing is for
+/// cluster-readiness — the same pid is freely a member of many groups, and the
+/// width admits multiples in general. Held under one Leaf-class `RawMutex`.
+pub(crate) struct ProcessGroups {
+    groups: HashMap<String, Vec<Member>>,
+}
+
+impl ProcessGroups {
+    pub(crate) fn new() -> Self {
+        Self { groups: HashMap::new() }
+    }
+
+    /// Add `member` to `group`. Idempotent within a group: returns `true` if
+    /// it was newly inserted, `false` if it was already present.
+    pub(crate) fn join(&mut self, group: &str, member: Member) -> bool {
+        let v = self.groups.entry(group.to_owned()).or_default();
+        if v.contains(&member) {
+            return false;
+        }
+        v.push(member);
+        true
+    }
+
+    /// Remove `member` from `group`. Returns whether anything was removed.
+    /// An emptied group is pruned so the keyspace does not leak empty vecs.
+    pub(crate) fn leave(&mut self, group: &str, member: Member) -> bool {
+        let Some(v) = self.groups.get_mut(group) else {
+            return false;
+        };
+        let before = v.len();
+        v.retain(|m| *m != member);
+        let removed = v.len() != before;
+        if v.is_empty() {
+            self.groups.remove(group);
+        }
+        removed
+    }
+
+    /// The one dumb eviction primitive: drop every member matching `pred` from
+    /// every group, pruning emptied groups. The primitive does not know *why*
+    /// a member leaves. First caller: the monitor death hook (Phase 2). Later:
+    /// `evict_incarnation` (RFC 010) over the same path.
+    //
+    // No non-test caller until Phase 2 wires the death hook to it.
+    #[allow(dead_code)]
+    pub(crate) fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) {
+        self.groups.retain(|_, v| {
+            v.retain(|m| !pred(m));
+            !v.is_empty()
+        });
+    }
+
+    /// Raw enumeration of a group's members — no liveness filtering (the
+    /// public `members` layers that on in Phase 3). Absent group → empty.
+    pub(crate) fn members_of(&self, group: &str) -> Vec<Member> {
+        self.groups.get(group).cloned().unwrap_or_default()
+    }
+}
+
+/// Build the full member identity for `pid` from runtime identity.
+fn member_for(inner: &crate::runtime::RuntimeInner, pid: Pid) -> Member {
+    Member { node: inner.node_id, incarnation: inner.incarnation, pid }
+}
+
+/// Add `pid` to `group`. The same pid may join many groups; within one group a
+/// pid is a member at most once (idempotent). Returns `true` if this call
+/// newly added the membership.
+///
+/// Panics if called outside `Runtime::run()`.
+//
+// Phase 2 will install `monitor(pid)` on first join (registered under the
+// shared mutex, racing `finalize_actor` exactly as every other monitor does)
+// and route its one-shot `Down` to `remove_where(|m| m.pid == pid)`.
+pub fn join(group: impl Into<String>, pid: Pid) -> bool {
+    let group = group.into();
+    with_runtime(|inner| {
+        let member = member_for(inner, pid);
+        inner.process_groups.lock().join(&group, member)
+    })
+}
+
+/// Drop `pid`'s membership of `group`. Returns whether a membership was
+/// removed.
+///
+/// Panics if called outside `Runtime::run()`.
+pub fn leave(group: &str, pid: Pid) -> bool {
+    with_runtime(|inner| {
+        let member = member_for(inner, pid);
+        inner.process_groups.lock().leave(group, member)
+    })
+}
+
+/// Fan-out read: every member of `group`.
+///
+/// The set is self-pruning once the death hook lands (Phase 2), so liveness is
+/// maintained by eviction rather than filtered on read; Phase 3 adds a
+/// generation-checked liveness read as the belt-and-braces backstop.
+///
+/// Panics if called outside `Runtime::run()`.
+pub fn members(group: &str) -> Vec<Pid> {
+    with_runtime(|inner| {
+        inner.process_groups.lock().members_of(group).into_iter().map(|m| m.pid).collect()
+    })
+}
+
+/// Pool/discovery read: one member of `group`, or `None` if empty. Stateless
+/// first-live selection (the first-*live* refinement lands in Phase 3; smarter
+/// routing is an explicitly later, clustered concern).
+///
+/// Panics if called outside `Runtime::run()`.
+pub fn pick(group: &str) -> Option<Pid> {
+    with_runtime(|inner| {
+        inner.process_groups.lock().members_of(group).into_iter().next().map(|m| m.pid)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn m(index: u32, generation: u32) -> Member {
+        Member {
+            node: DEFAULT_NODE_ID,
+            incarnation: DEFAULT_INCARNATION,
+            pid: Pid::new(index, generation),
+        }
+    }
+
+    #[test]
+    fn join_is_idempotent_within_a_group() {
+        let mut pg = ProcessGroups::new();
+        assert!(pg.join("workers", m(1, 0)), "first join is new");
+        assert!(!pg.join("workers", m(1, 0)), "second identical join is a no-op");
+        assert_eq!(pg.members_of("workers"), vec![m(1, 0)]);
+    }
+
+    #[test]
+    fn same_pid_in_many_groups_is_independent() {
+        let mut pg = ProcessGroups::new();
+        pg.join("a", m(1, 0));
+        pg.join("b", m(1, 0));
+        pg.join("b", m(2, 0));
+        assert_eq!(pg.members_of("a"), vec![m(1, 0)]);
+        assert_eq!(pg.members_of("b"), vec![m(1, 0), m(2, 0)]);
+    }
+
+    #[test]
+    fn distinct_generations_are_distinct_members() {
+        // ABA guard: same slot index, different generation = different actor.
+        let mut pg = ProcessGroups::new();
+        assert!(pg.join("g", m(1, 0)));
+        assert!(pg.join("g", m(1, 1)), "different generation is a distinct member");
+        assert_eq!(pg.members_of("g"), vec![m(1, 0), m(1, 1)]);
+    }
+
+    #[test]
+    fn leave_removes_one_membership_and_prunes_empty_groups() {
+        let mut pg = ProcessGroups::new();
+        pg.join("g", m(1, 0));
+        pg.join("g", m(2, 0));
+        assert!(pg.leave("g", m(1, 0)));
+        assert_eq!(pg.members_of("g"), vec![m(2, 0)]);
+        assert!(!pg.leave("g", m(1, 0)), "second leave finds nothing");
+        assert!(pg.leave("g", m(2, 0)));
+        assert!(pg.members_of("g").is_empty(), "group is now empty");
+        // Leaving a never-joined group is a clean no-op.
+        assert!(!pg.leave("never", m(9, 0)));
+    }
+
+    #[test]
+    fn remove_where_sweeps_every_group() {
+        let mut pg = ProcessGroups::new();
+        pg.join("a", m(1, 0));
+        pg.join("a", m(2, 0));
+        pg.join("b", m(1, 0));
+        pg.join("c", m(3, 0));
+        // Death of pid index 1 (any generation) evicts it everywhere.
+        pg.remove_where(|mem| mem.pid.index() == 1);
+        assert_eq!(pg.members_of("a"), vec![m(2, 0)]);
+        assert!(pg.members_of("b").is_empty(), "b held only pid 1; pruned");
+        assert_eq!(pg.members_of("c"), vec![m(3, 0)]);
+    }
+
+    #[test]
+    fn remove_where_can_match_an_incarnation_sweep() {
+        // Shape check for the later evict_incarnation(node, inc) caller.
+        let mut pg = ProcessGroups::new();
+        let dead = Member { incarnation: Incarnation::new(7), ..m(1, 0) };
+        pg.join("g", dead);
+        pg.join("g", m(2, 0));
+        pg.remove_where(|mem| mem.incarnation == Incarnation::new(7));
+        assert_eq!(pg.members_of("g"), vec![m(2, 0)]);
+    }
+}

src/runtime.rs

@@ -155,6 +155,8 @@ pub struct Config {
     stack_pool_cap: usize,
     max_actors: usize,
     wake_slot: bool,
+    node_id: crate::pg::NodeId,
+    incarnation: crate::pg::Incarnation,
 }
 
 impl Config {
@@ -168,6 +170,8 @@ impl Config {
             stack_pool_cap: n * 4,
             max_actors: DEFAULT_MAX_ACTORS,
             wake_slot: false,
+            node_id: crate::pg::DEFAULT_NODE_ID,
+            incarnation: crate::pg::DEFAULT_INCARNATION,
         }
     }
 
@@ -185,6 +189,8 @@ impl Config {
             stack_pool_cap: max * 4,
             max_actors: DEFAULT_MAX_ACTORS,
             wake_slot: false,
+            node_id: crate::pg::DEFAULT_NODE_ID,
+            incarnation: crate::pg::DEFAULT_INCARNATION,
         }
     }
 
@@ -241,6 +247,23 @@ impl Config {
         self
     }
 
+    /// This runtime's node identity (RFC 012). Defaults to a fixed single-node
+    /// value; clustering (RFC 010) will supply a real one. Threaded through pg
+    /// storage/eviction so the process-group public API never changes to
+    /// acquire it.
+    pub fn node_id(mut self, id: impl Into<crate::pg::NodeId>) -> Self {
+        self.node_id = id.into();
+        self
+    }
+
+    /// This runtime's incarnation epoch (RFC 012) — the BEAM `Creation` analogue
+    /// that separates a crashed node from its restart. Defaults to a fixed
+    /// single-node value; constant for the life of a run.
+    pub fn incarnation(mut self, inc: impl Into<crate::pg::Incarnation>) -> Self {
+        self.incarnation = inc.into();
+        self
+    }
+
     /// The number of scheduler threads this config resolves to.
     pub fn resolved_thread_count(&self) -> usize {
         if let Some(e) = self.exact {
@@ -265,6 +288,8 @@ impl Default for Config {
             stack_pool_cap: avail * 4,
             max_actors: DEFAULT_MAX_ACTORS,
             wake_slot: false,
+            node_id: crate::pg::DEFAULT_NODE_ID,
+            incarnation: crate::pg::DEFAULT_INCARNATION,
         }
     }
 }
@@ -507,6 +532,17 @@ pub(crate) struct RuntimeInner {
     /// with any other lock; liveness checks under it read only the atomic
     /// slot word.
     pub(crate) registry: RawMutex<crate::registry::Registry>,
+    /// Runtime identity (RFC 012). Read-only after init — one node, one fixed
+    /// incarnation until clustering (RFC 010) supplies real values. Carried
+    /// like `wake_slot`; pg fills these into every `Member` so the public
+    /// surface stays Pid-shaped.
+    pub(crate) node_id: crate::pg::NodeId,
+    pub(crate) incarnation: crate::pg::Incarnation,
+    /// Process groups: `name -> multiset<Member>` (RFC 012). RawMutex Leaf,
+    /// exactly like `registry`: never held with any other lock; liveness
+    /// checks under it read only the atomic slot word, and the eviction path
+    /// keeps it off the send path.
+    pub(crate) process_groups: RawMutex<crate::pg::ProcessGroups>,
     /// Recycled stacks waiting to be reused by the next spawn.
     pub(crate) stack_pool: RawMutex<Vec<crate::stack::Stack>>,
     /// Maximum number of stacks to retain in the pool.
@@ -521,6 +557,8 @@ impl RuntimeInner {
         stack_pool_cap: usize,
         max_actors: usize,
         wake_slot: bool,
+        node_id: crate::pg::NodeId,
+        incarnation: crate::pg::Incarnation,
     ) -> Arc<Self> {
         let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
         let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
@@ -542,6 +580,9 @@ impl RuntimeInner {
             timeslice_cycles,
             wake_slot,
             registry: RawMutex::new(crate::registry::Registry::new()),
+            node_id,
+            incarnation,
+            process_groups: RawMutex::new(crate::pg::ProcessGroups::new()),
             stack_pool: RawMutex::new(Vec::new()),
             stack_pool_cap,
         })
@@ -712,6 +753,8 @@ pub fn init(config: Config) -> Runtime {
             config.stack_pool_cap,
             config.max_actors,
             config.wake_slot,
+            config.node_id,
+            config.incarnation,
         ),
         thread_count: n,
     }