Forward serialize through node wrappers and free abandoned frame table reservations

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

@@ -1,6 +1,6 @@
 use crate::gpoll::Finality;
 use std::cell::UnsafeCell;
-use std::mem::MaybeUninit;
+use std::mem::{ManuallyDrop, MaybeUninit};
 use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
 
 const SLOT_EMPTY: u8 = 0;
@@ -81,18 +81,27 @@ impl<T, const CAP: usize> Drop for FrameTable<T, CAP> {
 
 impl<'t, T> VacantSlot<'t, T> {
 	pub fn publish(self, value: T, finality: Finality) -> &'t T {
+		let slot = ManuallyDrop::new(self).slot;
 		// SAFETY: the CAS in `lookup` reserved this slot exclusively for us and
 		// its state is still SLOT_EMPTY, so nobody reads the value yet.
-		let lent = unsafe { &*(*self.slot.value.get()).write(value) };
+		let lent = unsafe { &*(*slot.value.get()).write(value) };
 		let state = match finality {
 			Finality::AllFinal => SLOT_FINAL,
 			Finality::Partial => SLOT_PARTIAL,
 		};
-		self.slot.state.store(state, Ordering::Release);
+		slot.state.store(state, Ordering::Release);
 		lent
 	}
 
 	pub fn release(self) {
+		drop(self);
+	}
+}
+
+/// Frees the reservation, so an early return or panic between `lookup` and
+/// `publish` cannot retire the slot for the rest of the table's life.
+impl<T> Drop for VacantSlot<'_, T> {
+	fn drop(&mut self) {
 		self.slot.key.store(0, Ordering::Release);
 	}
 }
@@ -123,6 +132,14 @@ mod tests {
 		assert!(matches!(table.lookup(7), Lookup::Vacant(_)));
 	}
 
+	#[test]
+	fn a_dropped_reservation_frees_the_slot() {
+		let table = FrameTable::<u32, 8>::new();
+		let Lookup::Vacant(slot) = table.lookup(7) else { unreachable!() };
+		drop(slot);
+		assert!(matches!(table.lookup(7), Lookup::Vacant(_)), "an abandoned reservation must not retire the slot");
+	}
+
 	#[test]
 	fn neighboring_hashes_do_not_share_an_entry() {
 		let table = FrameTable::<u32, 8>::new();

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

@@ -94,6 +94,10 @@ where
 		(**self).extent(input)
 	}
 
+	fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
+		(**self).serialize()
+	}
+
 	fn eval_batch<'a>(&self, input: &'a Input, range: Range<u64>, scratch: Option<&'a mut [MaybeUninit<Self::Output>]>) -> BatchStatus<'a, Self::Output>
 	where
 		Input: InjectIndex + Copy,
@@ -116,6 +120,10 @@ where
 		(**self).extent(input)
 	}
 
+	fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
+		(**self).serialize()
+	}
+
 	fn eval_batch<'a>(&self, input: &'a Input, range: Range<u64>, scratch: Option<&'a mut [MaybeUninit<Self::Output>]>) -> BatchStatus<'a, Self::Output>
 	where
 		Input: InjectIndex + Copy,
@@ -138,6 +146,10 @@ where
 		(**self).extent(input)
 	}
 
+	fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
+		(**self).serialize()
+	}
+
 	fn eval_batch<'a>(&self, input: &'a Input, range: Range<u64>, scratch: Option<&'a mut [MaybeUninit<Self::Output>]>) -> BatchStatus<'a, Self::Output>
 	where
 		Input: InjectIndex + Copy,
@@ -239,28 +251,6 @@ impl<'a, N> LazyInput<'a, N> {
 	}
 }
 
-impl<'a, Input, N> Node<Input> for LazyInput<'a, N>
-where
-	N: Node<Input>,
-{
-	type Output = N::Output;
-
-	fn eval(&self, input: &Input) -> GPoll<Self::Output> {
-		self.node.eval(input)
-	}
-
-	fn extent(&self, input: &Input) -> GPoll<Extent> {
-		self.node.extent(input)
-	}
-
-	fn eval_batch<'b>(&self, input: &'b Input, range: Range<u64>, scratch: Option<&'b mut [MaybeUninit<Self::Output>]>) -> BatchStatus<'b, Self::Output>
-	where
-		Input: InjectIndex + Copy,
-	{
-		self.node.eval_batch(input, range, scratch)
-	}
-}
-
 #[cfg(test)]
 mod tests {
 	use super::*;