From e78a987e3e9d74ea72ff576b09bee62e6406eb4a Mon Sep 17 00:00:00 2001 From: Dennis Kobert Date: Tue, 8 Sep 2026 13:14:12 +0000 Subject: [PATCH] Bind a park forwarding to the destination arena generation --- node-graph/libraries/core-types/src/arena.rs | 44 ++++++++++++++++---- 1 file changed, 37 insertions(+), 7 deletions(-) diff --git a/node-graph/libraries/core-types/src/arena.rs b/node-graph/libraries/core-types/src/arena.rs index 936e135c3a..2ac9100ed6 100644 --- a/node-graph/libraries/core-types/src/arena.rs +++ b/node-graph/libraries/core-types/src/arena.rs @@ -27,11 +27,12 @@ pub struct Arena { /// a park costs one pointer in the arena and owns its content outside it. retained_heap: AtomicUsize, /// Where [`Arena::move_park`] sent each moved park, as its offset here to - /// the header's address in the receiving arena and the moved type's key, so - /// a payload two records share is moved once and a mistyped sharer is - /// refused. Cleared by [`Arena::reset`], so a forwarding holds for one - /// generation. - forwarded: Mutex>, + /// the header's address in the receiving arena, the moved type's key and + /// that arena's generation, so a payload two records share is moved once + /// while a mistyped sharer, a sharer naming another destination, and a + /// sharer whose destination has since flushed are all refused. Cleared by + /// [`Arena::reset`], so a forwarding holds for one generation. + forwarded: Mutex>, } impl std::fmt::Debug for Arena { @@ -264,6 +265,12 @@ impl Arena { /// fan-out sharer's second move reaches the one header and the obligation /// is never duplicated. /// + /// A forwarding is to one destination generation, since the returned + /// pointer is only the caller's to publish where it lands in the `dst` it + /// named: a second move to another arena, or to one that has flushed since, + /// is refused rather than answered with a foreign or freed header. The + /// refusal reads as a decline, so the caller takes its clone path. + /// /// `dst` is credited `retained`, which is what a clone of the payload would /// have credited it, and this arena is debited what its own park recorded, /// so neither counter reads worse than it did before the move. @@ -290,8 +297,11 @@ impl Arena { (offset < self.buf.len()).then_some(())?; let type_of = TypeId::of::(); let mut forwarded = self.forwarded.lock().unwrap(); - if let Some(&(moved, moved_type)) = forwarded.get(&offset) { + if let Some(&(moved, moved_type, moved_generation)) = forwarded.get(&offset) { (moved_type == type_of).then_some(())?; + // Generations are globally unique, so the match is `dst` itself and + // its current epoch; anything else declines. + (moved_generation == dst.generation()).then_some(())?; return Some(moved as *const T::Static); } let mut entries = self.drops.lock().unwrap(); @@ -316,7 +326,7 @@ impl Arena { }); dst.retained_heap.fetch_add(retained, Ordering::Relaxed); let _ = self.retained_heap.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(parked))); - forwarded.insert(offset, (target as usize, type_of)); + forwarded.insert(offset, (target as usize, type_of, dst.generation())); Some(target.cast_const()) } @@ -567,6 +577,26 @@ mod tests { assert!(arena.handle_at(unreserved).is_none(), "a byte past the watermark mints nothing"); } + #[test] + fn a_forwarding_answers_only_the_destination_it_recorded() { + let _guard = COUNTER_GUARD.lock().unwrap_or_else(PoisonError::into_inner); + struct Probe(#[allow(dead_code)] String); + unsafe impl dyn_any::StaticType for Probe { + type Static = Probe; + } + let transient = Arena::new(1024).unwrap(); + let mut first = Arena::new(1024).unwrap(); + let second = Arena::new(1024).unwrap(); + + let (parked, _) = transient.alloc_sized_keyed(Probe(String::from("shared by two records")), 21).unwrap(); + let src = std::ptr::from_ref(parked).cast::(); + unsafe { transient.move_park::(src, &first, 21) }.unwrap(); + + assert!(unsafe { transient.move_park::(src, &second, 21) }.is_none(), "a sharer naming another destination is refused"); + first.reset(); + assert!(unsafe { transient.move_park::(src, &first, 21) }.is_none(), "a destination that has flushed since is refused too"); + } + /// Held by every test that perturbs [`NEXT_GENERATION`], so a swapped-out counter /// is never observed by a concurrently constructing test. static COUNTER_GUARD: Mutex<()> = Mutex::new(());