diff --git a/src/pg.rs b/src/pg.rs index dcf257b..29ab8bb 100644 --- a/src/pg.rs +++ b/src/pg.rs @@ -17,8 +17,9 @@ //! group it touches first — draining each membership's monitor with a //! non-blocking `try_recv` — and on the first sign of death sweeps that pid out //! of *every* group via the predicate primitive. So the set is self-pruning on -//! contact rather than filtered on read; Phase 3 adds a generation-checked -//! liveness read as a belt-and-braces backstop on the read path. +//! contact rather than filtered on read; the read path additionally applies a +//! generation-checked liveness backstop so a member already dead but not yet +//! reaped is never *returned*, even though eviction remains the monitor's job. //! //! [reap]: ProcessGroups::reap_group //! @@ -231,9 +232,9 @@ impl ProcessGroups { self.remove_where(|m| dead.contains(&m.pid)) } - /// Raw enumeration of a group's members — no liveness filtering (Phase 3 - /// layers a generation-checked backstop on the public `members`). Absent - /// group → empty. + /// Raw enumeration of a group's members — no liveness filtering. Used by + /// tests to assert storage state independently of the read-path backstop. + #[cfg(test)] fn members_of(&self, group: &str) -> Vec { self.groups .get(group) @@ -241,10 +242,22 @@ impl ProcessGroups { .unwrap_or_default() } - /// The first member of `group`, if any — backing the stateless first-live - /// `pick` without cloning the whole group. - fn first_member(&self, group: &str) -> Option { - self.groups.get(group).and_then(|v| v.first().map(|e| e.member)) + /// Live members of `group`, in insertion order. The `is_live` oracle is the + /// read-path backstop (Phase 3): a member whose slot is already dead is + /// dropped from the *result* even if its `Down` has not been drained yet. + /// Backstop only — the entry stays in storage; eviction is the monitor's + /// job (`reap_group`). + fn members_where(&self, group: &str, mut is_live: impl FnMut(Pid) -> bool) -> Vec { + self.groups + .get(group) + .map(|v| v.iter().map(|e| e.member.pid).filter(|&p| is_live(p)).collect()) + .unwrap_or_default() + } + + /// The first live member of `group` in insertion order — stateless + /// first-live `pick`, with the same read-path backstop as `members_where`. + fn first_member_where(&self, group: &str, mut is_live: impl FnMut(Pid) -> bool) -> Option { + self.groups.get(group)?.iter().map(|e| e.member.pid).find(|&p| is_live(p)) } } @@ -253,6 +266,14 @@ fn member_for(inner: &crate::runtime::RuntimeInner, pid: Pid) -> Member { Member { node: inner.node_id, incarnation: inner.incarnation, pid } } +/// Is `pid` a live actor right now? Generation-checked atomic slot-word read, +/// no lock — identical to the registry's guard. The read-path backstop: a +/// generation is never reused, so a dead member is detectable independently of +/// whether its monitor `Down` has been drained yet. +fn live(inner: &crate::runtime::RuntimeInner, pid: Pid) -> bool { + inner.slot_at(pid).is_some_and(|s| s.is_live_for(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, `false` if it was already a member. @@ -316,15 +337,16 @@ pub fn leave(group: &str, pid: Pid) -> bool { /// Fan-out read: every live member of `group`. /// /// The touched group is reaped first, so dead members are evicted before the -/// read rather than filtered out of it. Phase 3 adds a generation-checked -/// liveness read as a belt-and-braces backstop. +/// read. The generation-checked liveness read is a belt-and-braces backstop: +/// even in the finalize window where a member is already dead but its `Down` +/// has not yet landed, it is dropped from the result. /// /// Panics if called outside `Runtime::run()`. pub fn members(group: &str) -> Vec { let (pids, reaped) = with_runtime(|inner| { let mut pg = inner.process_groups.lock(); let reaped = pg.reap_group(group); - let pids: Vec = pg.members_of(group).into_iter().map(|m| m.pid).collect(); + let pids = pg.members_where(group, |pid| live(inner, pid)); (pids, reaped) }); drop(reaped); @@ -332,16 +354,17 @@ pub fn members(group: &str) -> Vec { } /// Pool/discovery read: one live member of `group`, or `None` if empty. -/// Stateless first-live selection (the touched group is reaped first, so the -/// first surviving member is live; smarter routing is an explicitly later, -/// clustered concern per RFC 010). +/// Stateless first-live selection: the touched group is reaped first, and the +/// generation-checked liveness backstop skips any member already dead but not +/// yet reaped. Smarter routing is an explicitly later, clustered concern per +/// RFC 010. /// /// Panics if called outside `Runtime::run()`. pub fn pick(group: &str) -> Option { let (picked, reaped) = with_runtime(|inner| { let mut pg = inner.process_groups.lock(); let reaped = pg.reap_group(group); - let picked = pg.first_member(group).map(|m| m.pid); + let picked = pg.first_member_where(group, |pid| live(inner, pid)); (picked, reaped) }); drop(reaped); @@ -495,4 +518,26 @@ mod tests { assert_eq!(evicted.len(), 1); assert!(pg.members_of("a").is_empty()); } + + #[test] + fn read_backstop_hides_a_member_the_monitor_has_not_yet_reaped() { + let mut pg = ProcessGroups::new(); + // Both senders held: reap_group would see Ok(None) and evict neither. + let (a, _ta) = synth(1, 0); + let (b, _tb) = synth(2, 0); + pg.join("g", a); + pg.join("g", b); + + // The slot-word oracle already reports pid 1 dead (finalize window), + // ahead of any Down delivery. + let dead = Pid::new(1, 0); + let oracle = |pid: Pid| pid != dead; + + assert_eq!(pg.members_where("g", oracle), vec![Pid::new(2, 0)], "dead pid filtered from read"); + assert_eq!(pg.first_member_where("g", oracle), Some(Pid::new(2, 0)), "pick skips the dead first member"); + + // Backstop does not evict — that stays the monitor's job; raw storage + // still holds both until reap runs. + assert_eq!(pg.members_of("g"), vec![member(1, 0), member(2, 0)]); + } }