Park the arena during drop glue, guard filled batches, and normalize deserialized sources

node-graph/libraries/core-types/src/arena.rs

@@ -152,16 +152,15 @@ impl Arena {
 	/// where every handle misses and further allocation is refused.
 	pub fn reset(&mut self) -> bool {
 		let base = self.base();
-		for entry in self.drops.get_mut().unwrap().drain(..).rev() {
+		let entries = std::mem::take(self.drops.get_mut().unwrap());
+		self.generation.store(PARKED_GENERATION, Ordering::Release);
+		for entry in entries.into_iter().rev() {
 			// SAFETY: registered at alloc time; insert-only means the region was
 			// never overwritten within this generation.
 			unsafe { (entry.drop_fn)(base.add(entry.offset)) }
 		}
 		*self.offset.get_mut() = 0;
-		let Some(generation) = next_generation() else {
-			self.generation.store(PARKED_GENERATION, Ordering::Release);
-			return false;
-		};
+		let Some(generation) = next_generation() else { return false };
 		self.generation.store(generation, Ordering::Release);
 		true
 	}
@@ -320,6 +319,25 @@ mod tests {
 		assert!(arena.alloc(0u32).is_some(), "the arena stays usable");
 	}
 
+	#[test]
+	fn a_panicking_destructor_leaves_no_resolvable_handle() {
+		let _guard = COUNTER_GUARD.lock().unwrap_or_else(PoisonError::into_inner);
+		struct Bomb;
+		impl Drop for Bomb {
+			fn drop(&mut self) {
+				panic!("payload destructor");
+			}
+		}
+		let mut arena = Arena::new(1024).unwrap();
+		let cell = ArenaCell::new();
+		let (_, weak) = arena.alloc(Bomb).unwrap();
+		cell.store(weak);
+
+		let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| arena.reset()));
+		assert!(unwound.is_err(), "the panic must propagate out of reset");
+		assert!(cell.load(&arena).is_none(), "a half-dropped generation must resolve no handle");
+	}
+
 	/// 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(());

node-graph/libraries/core-types/src/context.rs

@@ -200,7 +200,7 @@ pub struct ContextDependencies {
 	pub extract: ContextFeatures,
 	pub inject: ContextFeatures,
 	/// Must stay sorted.
-	#[cfg_attr(feature = "serde", serde(default))]
+	#[cfg_attr(feature = "serde", serde(default, deserialize_with = "deserialize_sorted_sources"))]
 	pub sources: Vec<SourceId>,
 }
 
@@ -209,9 +209,21 @@ pub struct ContextDependencies {
 pub struct ContextModification {
 	pub features: ContextFeatures,
 	/// Must stay sorted.
+	#[cfg_attr(feature = "serde", serde(deserialize_with = "deserialize_sorted_sources"))]
 	pub sources: Vec<SourceId>,
 }
 
+/// Restores the sorted-and-deduplicated invariant that `contains` and `difference`
+/// rely on for binary search, which arbitrary serialized input can violate.
+#[cfg(feature = "serde")]
+fn deserialize_sorted_sources<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<Vec<SourceId>, D::Error> {
+	use serde::Deserialize;
+	let mut sources = Vec::deserialize(deserializer)?;
+	sources.sort_unstable();
+	sources.dedup();
+	Ok(sources)
+}
+
 impl core::ops::BitOrAssign<&ContextModification> for ContextModification {
 	fn bitor_assign(&mut self, other: &Self) {
 		self.features |= other.features;
@@ -958,8 +970,8 @@ pub struct CtxSnapshot {
 	real_time: Option<f64>,
 	animation_time: Option<f64>,
 	pointer_position: Option<DVec2>,
-	index: Vec<usize>,
-	positions: Vec<DVec2>,
+	index: Option<Vec<usize>>,
+	positions: Option<Vec<DVec2>>,
 	varargs: Vec<Vec<OwnedSlot>>,
 	generations: Vec<(SourceId, u64)>,
 }
@@ -974,8 +986,8 @@ impl CtxSnapshot {
 			real_time: ctx.try_real_time(),
 			animation_time: ctx.try_animation_time(),
 			pointer_position: ctx.try_pointer_position(),
-			index: ctx.try_index().map(|levels| levels.collect()).unwrap_or_default(),
-			positions: ctx.try_position().map(|positions| positions.collect()).unwrap_or_default(),
+			index: ctx.try_index().map(|levels| levels.collect()),
+			positions: ctx.try_position().map(|positions| positions.collect()),
 			varargs: std::iter::successors(ctx.varargs_head(), |link| link.outer)
 				.map(|link| link.args.iter().map(|slot| slot.clone_slot()).collect())
 				.collect(),
@@ -1018,13 +1030,13 @@ impl ExtractPointerPosition for CtxSnapshot {
 
 impl ExtractIndex for CtxSnapshot {
 	fn try_index(&self) -> Option<impl Iterator<Item = usize>> {
-		Some(self.index.iter().copied())
+		self.index.as_ref().map(|levels| levels.iter().copied())
 	}
 }
 
 impl ExtractPosition for CtxSnapshot {
 	fn try_position(&self) -> Option<impl Iterator<Item = DVec2>> {
-		Some(self.positions.iter().copied())
+		self.positions.as_ref().map(|positions| positions.iter().copied())
 	}
 }
 
@@ -1549,6 +1561,21 @@ mod context_impl_tests {
 		assert!(matches!(empty.vararg(0), Err(VarArgsResult::NoVarArgs)));
 	}
 
+	#[test]
+	fn a_snapshot_preserves_an_absent_position_axis() {
+		let arena = Arena::new(64).unwrap();
+		let generations = [];
+		let scope = scope_fixture(&generations, &arena);
+		let root = ContextImpl::root(&scope);
+
+		let absent = CtxSnapshot::capture(&root);
+		assert!(absent.try_position().is_none(), "capturing an unpositioned context must not invent a position stack");
+
+		let position = DVec2::new(1., 2.);
+		let positioned = CtxSnapshot::capture(&root.with_position(&PositionLink { position, outer: None }));
+		assert_eq!(positioned.try_position().map(|p| p.collect::<Vec<_>>()), Some(vec![position]));
+	}
+
 	#[test]
 	fn scope_arena_reaches_kernels_through_extract_arena() {
 		let arena = Arena::new(1024).unwrap();

node-graph/libraries/core-types/src/node.rs

@@ -7,13 +7,48 @@ use std::ops::Range;
 #[derive(Debug)]
 pub enum BatchStatus<'a, T> {
 	Lent(&'a [T], Finality),
-	Filled(&'a mut [T], Finality),
+	Filled(FilledBatch<'a, T>, Finality),
 	Pending,
 	Error(GraphError),
 	NeedBuffer,
 	InvalidRange,
 }
 
+/// Owns the initialized prefix of a caller-supplied scratch buffer, dropping every
+/// lane unless [`FilledBatch::into_values`] hands the obligation back to the caller.
+#[derive(Debug)]
+pub struct FilledBatch<'a, T> {
+	values: &'a mut [T],
+}
+
+impl<'a, T> FilledBatch<'a, T> {
+	/// # Safety
+	///
+	/// The first `len` elements of `scratch` must be initialized, and `len` must not exceed `scratch.len()`.
+	pub unsafe fn new(scratch: &'a mut [MaybeUninit<T>], len: usize) -> Self {
+		Self {
+			values: unsafe { assume_init_prefix_mut(scratch, len) },
+		}
+	}
+
+	pub fn values(&self) -> &[T] {
+		self.values
+	}
+
+	pub fn into_values(self) -> &'a mut [T] {
+		let mut guard = std::mem::ManuallyDrop::new(self);
+		std::mem::take(&mut guard.values)
+	}
+}
+
+impl<T> Drop for FilledBatch<'_, T> {
+	fn drop(&mut self) {
+		// SAFETY: every lane was initialized when the guard was built and none has
+		// been moved out, since `into_values` consumes the guard instead.
+		unsafe { std::ptr::drop_in_place(self.values as *mut [T]) }
+	}
+}
+
 /// # Safety
 ///
 /// The first `len` elements of `scratch` must be initialized, and `len` must not exceed `scratch.len()`.
@@ -76,7 +111,7 @@ pub trait Node<Input> {
 			}
 		}
 		// SAFETY: all `len` lanes were written by the loop above.
-		BatchStatus::Filled(unsafe { assume_init_prefix_mut(scratch, len) }, finality)
+		BatchStatus::Filled(unsafe { FilledBatch::new(scratch, len) }, finality)
 	}
 }
 
@@ -285,10 +320,37 @@ mod tests {
 		let BatchStatus::Filled(lanes, finality) = status else {
 			panic!("expected filled, got {status:?}");
 		};
-		assert_eq!(lanes, &[4, 6, 8, 10]);
+		assert_eq!(lanes.values(), &[4, 6, 8, 10]);
 		assert_eq!(finality, Finality::AllFinal);
 	}
 
+	#[test]
+	fn a_dropped_filled_batch_reclaims_every_lane() {
+		static DROPS: AtomicU32 = AtomicU32::new(0);
+		#[derive(Clone)]
+		struct Probe;
+		impl Drop for Probe {
+			fn drop(&mut self) {
+				DROPS.fetch_add(1, Ordering::Relaxed);
+			}
+		}
+		struct Probes;
+		impl Node<TestInput> for Probes {
+			type Output = Probe;
+
+			fn eval(&self, _input: &TestInput) -> GPoll<Probe> {
+				GPoll::Final(Probe)
+			}
+		}
+
+		let input = TestInput { index: 0 };
+		let mut scratch = [const { MaybeUninit::uninit() }; 3];
+		let status = Probes.eval_batch(&input, 0..3, Some(&mut scratch));
+		assert!(matches!(status, BatchStatus::Filled(..)));
+		drop(status);
+		assert_eq!(DROPS.load(Ordering::Relaxed), 3, "an unconsumed batch must not leak its lanes");
+	}
+
 	#[test]
 	fn probe_without_scratch_requests_a_buffer() {
 		let input = TestInput { index: 0 };
@@ -320,7 +382,7 @@ mod tests {
 		let BatchStatus::Filled(lanes, finality) = status else {
 			panic!("expected filled, got {status:?}");
 		};
-		assert_eq!(lanes, &[0, 1, 2, 3]);
+		assert_eq!(lanes.values(), &[0, 1, 2, 3]);
 		assert_eq!(finality, Finality::Partial);
 	}