Add the core execution types, the borrowed context, and the seeded u64 fx hasher

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

@@ -0,0 +1,281 @@
+use std::cell::UnsafeCell;
+use std::marker::PhantomData;
+use std::mem::MaybeUninit;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+
+pub struct Arena {
+	generation: AtomicU64,
+	offset: AtomicUsize,
+	buf: Box<[UnsafeCell<MaybeUninit<u8>>]>,
+	drops: Mutex<Vec<DropEntry>>,
+}
+
+impl std::fmt::Debug for Arena {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("Arena").field("generation", &self.generation).field("size", &self.buf.len()).finish()
+	}
+}
+
+struct DropEntry {
+	offset: usize,
+	drop_fn: unsafe fn(*mut u8),
+}
+
+// SAFETY: disjoint regions are handed out by an atomic bump; a region is written
+// only by its allocating caller before publication, and cross-thread hand-off is
+// ordered by the Release/Acquire pair on the published handle word.
+unsafe impl Sync for Arena {}
+unsafe impl Send for Arena {}
+
+impl std::panic::UnwindSafe for Arena {}
+impl std::panic::RefUnwindSafe for Arena {}
+
+impl Arena {
+	pub fn new(capacity: usize) -> Self {
+		let buf = (0..capacity).map(|_| UnsafeCell::new(MaybeUninit::uninit())).collect();
+		Self {
+			// Starts at 1 so the null handle word (0) can never upgrade.
+			generation: AtomicU64::new(1),
+			offset: AtomicUsize::new(0),
+			buf,
+			drops: Mutex::new(Vec::new()),
+		}
+	}
+
+	pub fn generation(&self) -> u64 {
+		self.generation.load(Ordering::Acquire)
+	}
+
+	fn base(&self) -> *mut u8 {
+		self.buf.as_ptr() as *mut u8
+	}
+
+	fn reserve(&self, size: usize, align: usize) -> Option<usize> {
+		debug_assert!(align.is_power_of_two());
+		let base = self.base() as usize;
+		let mut start = 0;
+		self.offset
+			.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
+				// Alignment is computed on the absolute address; the backbone
+				// allocation itself has no alignment guarantee.
+				let addr = (base.checked_add(current)?.checked_add(align - 1)?) & !(align - 1);
+				start = addr - base;
+				let end = start.checked_add(size)?;
+				(end <= self.buf.len()).then_some(end)
+			})
+			.ok()?;
+		Some(start)
+	}
+
+	pub fn alloc<T>(&self, value: T) -> Option<(&T, ArenaWeak<T>)> {
+		let offset = self.reserve(size_of::<T>(), align_of::<T>())?;
+		let ptr = unsafe { self.base().add(offset) }.cast::<T>();
+		// SAFETY: freshly reserved, aligned, in-bounds, unaliased.
+		unsafe { ptr.write(value) };
+		if std::mem::needs_drop::<T>() {
+			unsafe fn glue<T>(p: *mut u8) {
+				unsafe { p.cast::<T>().drop_in_place() }
+			}
+			self.drops.lock().unwrap().push(DropEntry { offset, drop_fn: glue::<T> });
+		}
+		// SAFETY: initialized above; insert-only, so no `&mut` to it can exist.
+		Some((unsafe { &*ptr }, ArenaWeak::new(self.generation(), offset)))
+	}
+
+	pub fn alloc_slice_copy<T: Copy>(&self, src: &[T]) -> Option<&[T]> {
+		let buf = self.alloc_scratch::<T>(src.len())?;
+		for (slot, &value) in buf.iter_mut().zip(src) {
+			slot.write(value);
+		}
+		// SAFETY: every lane written above from `src`.
+		Some(unsafe { std::slice::from_raw_parts(buf.as_ptr().cast::<T>(), src.len()) })
+	}
+
+	// Not drop-tracked: callers must either consume every written lane or
+	// restrict themselves to `Copy` payloads (leak, not UB, otherwise).
+	#[allow(clippy::mut_from_ref)]
+	pub fn alloc_scratch<T>(&self, len: usize) -> Option<&mut [MaybeUninit<T>]> {
+		let size = size_of::<T>().checked_mul(len)?;
+		let offset = self.reserve(size, align_of::<T>())?;
+		let ptr = unsafe { self.base().add(offset) }.cast::<MaybeUninit<T>>();
+		// SAFETY: exclusive region; lifetime tied to `&self`, and `reset` takes
+		// `&mut self`, so the slice cannot outlive the generation.
+		Some(unsafe { std::slice::from_raw_parts_mut(ptr, len) })
+	}
+
+	pub fn reset(&mut self) {
+		let base = self.base();
+		for entry in self.drops.get_mut().unwrap().drain(..) {
+			// 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;
+		self.generation.fetch_add(1, Ordering::Release);
+	}
+}
+
+impl Drop for Arena {
+	fn drop(&mut self) {
+		self.reset();
+	}
+}
+
+pub struct ArenaWeak<T> {
+	word: u64,
+	_marker: PhantomData<*const T>,
+}
+
+impl<T> Clone for ArenaWeak<T> {
+	fn clone(&self) -> Self {
+		*self
+	}
+}
+impl<T> Copy for ArenaWeak<T> {}
+
+impl<T> ArenaWeak<T> {
+	pub const NULL: Self = ArenaWeak { word: 0, _marker: PhantomData };
+
+	fn new(generation: u64, offset: usize) -> Self {
+		debug_assert!(offset < u32::MAX as usize);
+		Self {
+			word: ((generation & 0xFFFF_FFFF) << 32) | offset as u64,
+			_marker: PhantomData,
+		}
+	}
+
+	pub fn upgrade(self, arena: &Arena) -> Option<&T> {
+		let generation = self.word >> 32;
+		if generation != arena.generation() & 0xFFFF_FFFF {
+			return None;
+		}
+		let offset = (self.word & 0xFFFF_FFFF) as usize;
+		// SAFETY: same generation means the entry was fully written before its
+		// word was published (Release) and cannot move or be overwritten within
+		// a generation (insert-only); the Acquire load that produced this word
+		// ordered the payload writes.
+		Some(unsafe { &*arena.base().add(offset).cast::<T>() })
+	}
+}
+
+pub struct ArenaCell<T> {
+	word: AtomicU64,
+	_marker: PhantomData<fn() -> T>,
+}
+
+impl<T> Clone for ArenaCell<T> {
+	fn clone(&self) -> Self {
+		Self {
+			word: AtomicU64::new(self.word.load(Ordering::Acquire)),
+			_marker: PhantomData,
+		}
+	}
+}
+
+impl<T> std::fmt::Debug for ArenaCell<T> {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("ArenaCell").field("word", &self.word.load(Ordering::Relaxed)).finish()
+	}
+}
+
+impl<T> Default for ArenaCell<T> {
+	fn default() -> Self {
+		Self {
+			word: AtomicU64::new(0),
+			_marker: PhantomData,
+		}
+	}
+}
+
+impl<T> ArenaCell<T> {
+	pub fn new() -> Self {
+		Self::default()
+	}
+
+	pub fn load<'e>(&self, arena: &'e Arena) -> Option<&'e T> {
+		let weak = ArenaWeak::<T> {
+			word: self.word.load(Ordering::Acquire),
+			_marker: PhantomData,
+		};
+		weak.upgrade(arena)
+	}
+
+	pub fn store(&self, weak: ArenaWeak<T>) {
+		self.word.store(weak.word, Ordering::Release);
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use std::sync::atomic::AtomicU32;
+
+	#[test]
+	fn alloc_upgrade_reset_miss() {
+		let mut arena = Arena::new(1024);
+		let cell = ArenaCell::new();
+		let (value, weak) = arena.alloc(41u32).unwrap();
+		assert_eq!(*value, 41);
+		cell.store(weak);
+		assert_eq!(cell.load(&arena), Some(&41));
+		arena.reset();
+		assert_eq!(cell.load(&arena), None, "stale handle must miss");
+	}
+
+	#[test]
+	fn capacity_survives_reset() {
+		let mut arena = Arena::new(64 + align_of::<u32>() - 1);
+		for _ in 0..10 {
+			for _ in 0..16 {
+				assert!(arena.alloc(0u32).is_some());
+			}
+			assert!(arena.alloc(0u32).is_none(), "exhausted within generation");
+			arena.reset();
+		}
+	}
+
+	#[test]
+	fn panics_leave_the_arena_coherent() {
+		fn assert_ref_unwind_safe<T: std::panic::RefUnwindSafe>() {}
+		assert_ref_unwind_safe::<Arena>();
+
+		static DROPS: AtomicU32 = AtomicU32::new(0);
+		struct Probe(#[allow(dead_code)] String);
+		impl Drop for Probe {
+			fn drop(&mut self) {
+				DROPS.fetch_add(1, Ordering::Relaxed);
+			}
+		}
+		let mut arena = Arena::new(1024);
+		let cell = ArenaCell::new();
+		let result = std::panic::catch_unwind(|| {
+			let (_, weak) = arena.alloc(Probe("pre-panic".into())).unwrap();
+			cell.store(weak);
+			panic!("mid-eval");
+		});
+		assert!(result.is_err());
+		assert!(cell.load(&arena).is_some(), "the generation is still live after the caught panic");
+		arena.reset();
+		assert_eq!(DROPS.load(Ordering::Relaxed), 1, "reset reclaims pre-panic allocations");
+		assert!(cell.load(&arena).is_none(), "the bump kills stale handles");
+		assert!(arena.alloc(0u32).is_some(), "the arena stays usable");
+	}
+
+	#[test]
+	fn drop_glue_runs_on_reset() {
+		static DROPS: AtomicU32 = AtomicU32::new(0);
+		struct Probe(#[allow(dead_code)] String);
+		impl Drop for Probe {
+			fn drop(&mut self) {
+				DROPS.fetch_add(1, Ordering::Relaxed);
+			}
+		}
+		let mut arena = Arena::new(1024);
+		arena.alloc(Probe("owns heap".into())).unwrap();
+		arena.alloc(Probe("me too".into())).unwrap();
+		assert_eq!(DROPS.load(Ordering::Relaxed), 0);
+		arena.reset();
+		assert_eq!(DROPS.load(Ordering::Relaxed), 2);
+	}
+}

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

@@ -0,0 +1,155 @@
+use crate::gpoll::Finality;
+use std::cell::UnsafeCell;
+use std::mem::MaybeUninit;
+use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
+
+const SLOT_EMPTY: u8 = 0;
+const SLOT_FINAL: u8 = 1;
+const SLOT_PARTIAL: u8 = 2;
+
+pub struct FrameTable<T, const CAP: usize> {
+	slots: [FrameSlot<T>; CAP],
+}
+
+struct FrameSlot<T> {
+	key: AtomicU64,
+	state: AtomicU8,
+	value: UnsafeCell<MaybeUninit<T>>,
+}
+
+pub enum Lookup<'t, T> {
+	Hit(Finality, &'t T),
+	Vacant(VacantSlot<'t, T>),
+	Full,
+}
+
+pub struct VacantSlot<'t, T> {
+	slot: &'t FrameSlot<T>,
+}
+
+impl<T, const CAP: usize> Default for FrameTable<T, CAP> {
+	fn default() -> Self {
+		Self::new()
+	}
+}
+
+impl<T, const CAP: usize> FrameTable<T, CAP> {
+	pub fn new() -> Self {
+		Self {
+			slots: std::array::from_fn(|_| FrameSlot {
+				key: AtomicU64::new(0),
+				state: AtomicU8::new(SLOT_EMPTY),
+				value: UnsafeCell::new(MaybeUninit::uninit()),
+			}),
+		}
+	}
+
+	pub fn lookup(&self, hash: u64) -> Lookup<'_, T> {
+		// Keys are forced odd so the empty sentinel (0) can never collide.
+		let key = hash | 1;
+		for probe in 0..CAP {
+			let slot = &self.slots[(key as usize).wrapping_add(probe) % CAP];
+			let stored = slot.key.load(Ordering::Acquire);
+			if stored == key {
+				return match slot.state.load(Ordering::Acquire) {
+					// SAFETY: a published state was stored with Release after the
+					// value write; the Acquire load above ordered that write.
+					SLOT_FINAL => Lookup::Hit(Finality::AllFinal, unsafe { (*slot.value.get()).assume_init_ref() }),
+					SLOT_PARTIAL => Lookup::Hit(Finality::Partial, unsafe { (*slot.value.get()).assume_init_ref() }),
+					_ => Lookup::Full,
+				};
+			}
+			if stored == 0 && slot.key.compare_exchange(0, key, Ordering::AcqRel, Ordering::Acquire).is_ok() {
+				return Lookup::Vacant(VacantSlot { slot });
+			}
+		}
+		Lookup::Full
+	}
+}
+
+impl<T, const CAP: usize> Drop for FrameTable<T, CAP> {
+	fn drop(&mut self) {
+		for slot in &self.slots {
+			if slot.state.load(Ordering::Acquire) != SLOT_EMPTY {
+				// SAFETY: a non-empty state is only ever stored after the value
+				// write in `publish`.
+				unsafe { (*slot.value.get()).assume_init_drop() }
+			}
+		}
+	}
+}
+
+impl<'t, T> VacantSlot<'t, T> {
+	pub fn publish(self, value: T, finality: Finality) -> &'t T {
+		// 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 state = match finality {
+			Finality::AllFinal => SLOT_FINAL,
+			Finality::Partial => SLOT_PARTIAL,
+		};
+		self.slot.state.store(state, Ordering::Release);
+		lent
+	}
+
+	pub fn release(self) {
+		self.slot.key.store(0, Ordering::Release);
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use std::sync::atomic::AtomicU32;
+
+	#[test]
+	fn publish_then_hit_with_finality() {
+		let table = FrameTable::<u32, 8>::new();
+		let Lookup::Vacant(slot) = table.lookup(7) else {
+			panic!("fresh table must be vacant");
+		};
+		assert_eq!(*slot.publish(41, Finality::Partial), 41);
+		let Lookup::Hit(finality, value) = table.lookup(7) else {
+			panic!("published key must hit");
+		};
+		assert_eq!((finality, *value), (Finality::Partial, 41));
+	}
+
+	#[test]
+	fn released_slot_is_vacant_again() {
+		let table = FrameTable::<u32, 8>::new();
+		let Lookup::Vacant(slot) = table.lookup(7) else { unreachable!() };
+		slot.release();
+		assert!(matches!(table.lookup(7), Lookup::Vacant(_)));
+	}
+
+	#[test]
+	fn distinct_keys_probe_past_collisions_until_full() {
+		let table = FrameTable::<u32, 4>::new();
+		for hash in [2, 4, 6, 8] {
+			let Lookup::Vacant(slot) = table.lookup(hash) else {
+				panic!("hash {hash} should find a vacant slot");
+			};
+			slot.publish(hash as u32, Finality::AllFinal);
+		}
+		assert!(matches!(table.lookup(100), Lookup::Full));
+	}
+
+	#[test]
+	fn drop_runs_glue_for_published_values_only() {
+		static DROPS: AtomicU32 = AtomicU32::new(0);
+		struct Probe;
+		impl Drop for Probe {
+			fn drop(&mut self) {
+				DROPS.fetch_add(1, Ordering::Relaxed);
+			}
+		}
+		let table = FrameTable::<Probe, 8>::new();
+		let Lookup::Vacant(slot) = table.lookup(1) else { unreachable!() };
+		slot.publish(Probe, Finality::AllFinal);
+		let Lookup::Vacant(reserved_but_unpublished) = table.lookup(2) else { unreachable!() };
+		reserved_but_unpublished.release();
+		drop(table);
+		assert_eq!(DROPS.load(Ordering::Relaxed), 1);
+	}
+}

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

@@ -0,0 +1,235 @@
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ErrorKind {
+	Node(&'static str),
+	ArenaExhausted,
+	Panic,
+}
+
+impl PartialEq<&str> for ErrorKind {
+	fn eq(&self, other: &&str) -> bool {
+		matches!(self, ErrorKind::Node(kind) if kind == other)
+	}
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct GraphError {
+	pub kind: ErrorKind,
+	pub trace: Vec<usize>,
+}
+
+impl GraphError {
+	pub fn new(kind: &'static str) -> Self {
+		Self {
+			kind: ErrorKind::Node(kind),
+			trace: Vec::new(),
+		}
+	}
+
+	pub fn traced(mut self, input_index: usize) -> Self {
+		self.trace.push(input_index);
+		self
+	}
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub enum GPoll<T> {
+	Pending,
+	Final(T),
+	Partial(T),
+	Fallback(Box<(T, GraphError)>),
+	Error(Box<GraphError>),
+}
+
+impl<T> GPoll<T> {
+	#[inline(always)]
+	pub fn map<U>(self, f: impl FnOnce(T) -> U) -> GPoll<U> {
+		match self {
+			GPoll::Pending => GPoll::Pending,
+			GPoll::Final(value) => GPoll::Final(f(value)),
+			GPoll::Partial(value) => GPoll::Partial(f(value)),
+			GPoll::Fallback(boxed) => {
+				let (value, e) = *boxed;
+				GPoll::Fallback(Box::new((f(value), e)))
+			}
+			GPoll::Error(e) => GPoll::Error(e),
+		}
+	}
+
+	#[inline(always)]
+	pub fn and_then<U>(self, f: impl FnOnce(T) -> GPoll<U>) -> GPoll<U> {
+		match self {
+			GPoll::Pending => GPoll::Pending,
+			GPoll::Final(value) => f(value),
+			GPoll::Partial(value) => match f(value) {
+				GPoll::Final(result) => GPoll::Partial(result),
+				other => other,
+			},
+			GPoll::Fallback(boxed) => {
+				let (value, e) = *boxed;
+				match f(value) {
+					GPoll::Pending => GPoll::Pending,
+					GPoll::Final(result) | GPoll::Partial(result) => GPoll::Fallback(Box::new((result, e))),
+					GPoll::Fallback(inner) => {
+						let (result, _) = *inner;
+						GPoll::Fallback(Box::new((result, e)))
+					}
+					GPoll::Error(inner) => GPoll::Error(inner),
+				}
+			}
+			GPoll::Error(e) => GPoll::Error(e),
+		}
+	}
+
+	#[inline(always)]
+	pub fn zip<U>(self, other: GPoll<U>) -> GPoll<(T, U)> {
+		match (self, other) {
+			(GPoll::Error(e), _) | (_, GPoll::Error(e)) => GPoll::Error(e),
+			(GPoll::Pending, _) | (_, GPoll::Pending) => GPoll::Pending,
+			(GPoll::Final(a), GPoll::Final(b)) => GPoll::Final((a, b)),
+			(GPoll::Fallback(boxed), GPoll::Final(b) | GPoll::Partial(b)) => {
+				let (a, e) = *boxed;
+				GPoll::Fallback(Box::new(((a, b), e)))
+			}
+			(GPoll::Final(a) | GPoll::Partial(a), GPoll::Fallback(boxed)) => {
+				let (b, e) = *boxed;
+				GPoll::Fallback(Box::new(((a, b), e)))
+			}
+			(GPoll::Fallback(first), GPoll::Fallback(second)) => {
+				let (a, e) = *first;
+				let (b, _) = *second;
+				GPoll::Fallback(Box::new(((a, b), e)))
+			}
+			(GPoll::Partial(a), GPoll::Final(b) | GPoll::Partial(b)) | (GPoll::Final(a), GPoll::Partial(b)) => GPoll::Partial((a, b)),
+		}
+	}
+
+	#[inline(always)]
+	pub fn trace(self, input: usize) -> Self {
+		match self {
+			GPoll::Fallback(mut boxed) => {
+				boxed.1.trace.push(input);
+				GPoll::Fallback(boxed)
+			}
+			GPoll::Error(mut e) => {
+				e.trace.push(input);
+				GPoll::Error(e)
+			}
+			other => other,
+		}
+	}
+
+	pub fn fallback(value: T, kind: &'static str) -> Self {
+		GPoll::Fallback(Box::new((value, GraphError::new(kind))))
+	}
+
+	pub fn error(kind: &'static str) -> Self {
+		GPoll::Error(Box::new(GraphError::new(kind)))
+	}
+
+	pub fn arena_exhausted() -> Self {
+		GPoll::Error(Box::new(GraphError {
+			kind: ErrorKind::ArenaExhausted,
+			trace: Vec::new(),
+		}))
+	}
+
+	pub fn panicked() -> Self {
+		GPoll::Error(Box::new(GraphError {
+			kind: ErrorKind::Panic,
+			trace: Vec::new(),
+		}))
+	}
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub enum Interrupt {
+	Pending,
+	Error(Box<GraphError>),
+}
+
+impl From<GraphError> for Interrupt {
+	fn from(error: GraphError) -> Self {
+		Interrupt::Error(Box::new(error))
+	}
+}
+
+impl<T> From<Interrupt> for GPoll<T> {
+	fn from(interrupt: Interrupt) -> Self {
+		match interrupt {
+			Interrupt::Pending => GPoll::Pending,
+			Interrupt::Error(e) => GPoll::Error(e),
+		}
+	}
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Extent {
+	Free,
+	Exactly(usize),
+}
+
+impl Extent {
+	pub fn meet(a: GPoll<Extent>, b: GPoll<Extent>) -> GPoll<Extent> {
+		a.zip(b).and_then(|(a, b)| match (a, b) {
+			(Extent::Free, other) | (other, Extent::Free) => GPoll::Final(other),
+			(Extent::Exactly(n), Extent::Exactly(m)) if n == m => GPoll::Final(Extent::Exactly(n)),
+			(Extent::Exactly(n), Extent::Exactly(m)) => GPoll::fallback(Extent::Exactly(n.min(m)), "extent mismatch"),
+		})
+	}
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Finality {
+	AllFinal,
+	Partial,
+}
+
+impl Finality {
+	pub fn meet(self, other: Finality) -> Finality {
+		match (self, other) {
+			(Finality::AllFinal, Finality::AllFinal) => Finality::AllFinal,
+			_ => Finality::Partial,
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn free_is_the_meet_identity() {
+		let meet = Extent::meet(GPoll::Final(Extent::Free), GPoll::Final(Extent::Exactly(4)));
+		assert_eq!(meet, GPoll::Final(Extent::Exactly(4)));
+	}
+
+	#[test]
+	fn extent_mismatch_truncates_and_reports() {
+		let meet = Extent::meet(GPoll::Final(Extent::Exactly(3)), GPoll::Final(Extent::Exactly(5)));
+		let GPoll::Fallback(boxed) = meet else {
+			panic!("expected fallback, got {meet:?}");
+		};
+		assert_eq!(boxed.0, Extent::Exactly(3));
+		assert!(boxed.1.kind == "extent mismatch");
+	}
+
+	#[test]
+	fn error_dominates_pending_in_zip() {
+		let zipped = GPoll::<u32>::error("boom").zip(GPoll::<u32>::Pending);
+		assert!(matches!(zipped, GPoll::Error(_)));
+	}
+
+	#[test]
+	fn trace_builds_root_to_source_path() {
+		let poll = GPoll::<u32>::error("boom").trace(2).trace(0);
+		let GPoll::Error(e) = poll else { unreachable!() };
+		assert_eq!(e.trace, vec![2, 0]);
+	}
+
+	#[test]
+	fn interrupt_round_trips_to_gpoll() {
+		assert_eq!(GPoll::<u32>::from(Interrupt::Pending), GPoll::Pending);
+		let interrupt = Interrupt::from(GraphError::new("boom"));
+		assert!(matches!(GPoll::<u32>::from(interrupt), GPoll::Error(e) if e.kind == "boom"));
+	}
+}

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

@@ -1,16 +1,21 @@
 extern crate log;
 
+pub mod arena;
 pub mod bounds;
 pub mod consts;
 pub mod context;
+pub mod frame_table;
 pub mod generic;
+pub mod gpoll;
 pub mod list;
 pub mod math;
 pub mod memo;
 pub mod misc;
+pub mod node;
 pub mod ops;
 pub mod registry;
 pub mod render_complexity;
+pub mod runtime;
 pub mod transform;
 pub mod uuid;
 pub mod value;

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

@@ -0,0 +1,372 @@
+use crate::context::InjectIndex;
+use crate::gpoll::{Extent, Finality, GPoll, GraphError, Interrupt};
+use std::cell::Cell;
+use std::mem::MaybeUninit;
+use std::ops::Range;
+
+#[derive(Debug)]
+pub enum BatchStatus<'a, T> {
+	Lent(&'a [T], Finality),
+	Filled(&'a mut [T], Finality),
+	Pending,
+	Error(GraphError),
+	NeedBuffer,
+	InvalidRange,
+}
+
+/// # Safety
+///
+/// The first `len` elements of `scratch` must be initialized, and `len` must not exceed `scratch.len()`.
+pub unsafe fn assume_init_prefix_mut<T>(scratch: &mut [MaybeUninit<T>], len: usize) -> &mut [T] {
+	debug_assert!(len <= scratch.len());
+	unsafe { std::slice::from_raw_parts_mut(scratch.as_mut_ptr().cast::<T>(), len) }
+}
+
+pub trait Node<Input> {
+	type Output;
+
+	fn eval(&self, input: &Input) -> GPoll<Self::Output>;
+
+	fn extent(&self, _input: &Input) -> GPoll<Extent> {
+		GPoll::Final(Extent::Free)
+	}
+
+	/// Introspection access to node-resident records; `None` for ordinary nodes.
+	fn serialize(&self) -> Option<std::sync::Arc<dyn std::any::Any + Send + Sync>> {
+		None
+	}
+
+	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,
+	{
+		let Some(scratch) = scratch else {
+			return BatchStatus::NeedBuffer;
+		};
+		let Some(len) = range.end.checked_sub(range.start).map(|len| len as usize) else {
+			return BatchStatus::InvalidRange;
+		};
+		if scratch.len() < len {
+			return BatchStatus::InvalidRange;
+		}
+		let mut local = *input;
+		let mut finality = Finality::AllFinal;
+		for offset in 0..len {
+			local.set_index(range.start + offset as u64);
+			let abort = match self.eval(&local) {
+				GPoll::Final(value) => {
+					scratch[offset].write(value);
+					None
+				}
+				GPoll::Partial(value) => {
+					scratch[offset].write(value);
+					finality = Finality::Partial;
+					None
+				}
+				GPoll::Pending => Some(BatchStatus::Pending),
+				GPoll::Fallback(boxed) => Some(BatchStatus::Error(boxed.1)),
+				GPoll::Error(e) => Some(BatchStatus::Error(*e)),
+			};
+			if let Some(status) = abort {
+				for written in scratch[..offset].iter_mut() {
+					// SAFETY: every lane before `offset` was written by this loop.
+					unsafe { written.assume_init_drop() };
+				}
+				return status;
+			}
+		}
+		// SAFETY: all `len` lanes were written by the loop above.
+		BatchStatus::Filled(unsafe { assume_init_prefix_mut(scratch, len) }, finality)
+	}
+}
+
+impl<Input, N> Node<Input> for &N
+where
+	N: Node<Input> + ?Sized,
+{
+	type Output = N::Output;
+
+	fn eval(&self, input: &Input) -> GPoll<Self::Output> {
+		(**self).eval(input)
+	}
+
+	fn extent(&self, input: &Input) -> GPoll<Extent> {
+		(**self).extent(input)
+	}
+
+	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,
+	{
+		(**self).eval_batch(input, range, scratch)
+	}
+}
+
+impl<Input, N> Node<Input> for Box<N>
+where
+	N: Node<Input> + ?Sized,
+{
+	type Output = N::Output;
+
+	fn eval(&self, input: &Input) -> GPoll<Self::Output> {
+		(**self).eval(input)
+	}
+
+	fn extent(&self, input: &Input) -> GPoll<Extent> {
+		(**self).extent(input)
+	}
+
+	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,
+	{
+		(**self).eval_batch(input, range, scratch)
+	}
+}
+
+impl<Input, N> Node<Input> for std::sync::Arc<N>
+where
+	N: Node<Input> + ?Sized,
+{
+	type Output = N::Output;
+
+	fn eval(&self, input: &Input) -> GPoll<Self::Output> {
+		(**self).eval(input)
+	}
+
+	fn extent(&self, input: &Input) -> GPoll<Extent> {
+		(**self).extent(input)
+	}
+
+	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,
+	{
+		(**self).eval_batch(input, range, scratch)
+	}
+}
+
+pub struct StatusCell {
+	finality: Cell<Finality>,
+	error: Cell<Option<GraphError>>,
+	no_partial: bool,
+}
+
+impl Default for StatusCell {
+	fn default() -> Self {
+		Self::new()
+	}
+}
+
+impl StatusCell {
+	pub fn new() -> Self {
+		Self {
+			finality: Cell::new(Finality::AllFinal),
+			error: Cell::new(None),
+			no_partial: false,
+		}
+	}
+
+	pub fn no_partial() -> Self {
+		Self { no_partial: true, ..Self::new() }
+	}
+
+	pub fn eval_input<Input, N: Node<Input>>(&self, input_index: usize, node: &N, input: &Input) -> Result<N::Output, Interrupt> {
+		match node.eval(input) {
+			GPoll::Final(value) => Ok(value),
+			GPoll::Partial(_) if self.no_partial => Err(Interrupt::Pending),
+			GPoll::Partial(value) => {
+				self.finality.set(Finality::Partial);
+				Ok(value)
+			}
+			GPoll::Fallback(boxed) => {
+				let (value, error) = *boxed;
+				let first = self.error.take();
+				self.error.set(first.or(Some(error.traced(input_index))));
+				Ok(value)
+			}
+			GPoll::Pending => Err(Interrupt::Pending),
+			GPoll::Error(mut error) => {
+				error.trace.push(input_index);
+				Err(Interrupt::Error(error))
+			}
+		}
+	}
+
+	pub fn finish<T>(self, value: T) -> GPoll<T> {
+		match (self.error.take(), self.finality.get()) {
+			(Some(error), _) => GPoll::Fallback(Box::new((value, error))),
+			(None, Finality::AllFinal) => GPoll::Final(value),
+			(None, Finality::Partial) => GPoll::Partial(value),
+		}
+	}
+
+	pub fn merge<T>(self, poll: GPoll<T>) -> GPoll<T> {
+		match poll {
+			GPoll::Final(value) => self.finish(value),
+			GPoll::Partial(_) if self.no_partial => GPoll::Pending,
+			GPoll::Partial(value) => match self.finish(value) {
+				GPoll::Final(value) => GPoll::Partial(value),
+				other => other,
+			},
+			GPoll::Fallback(boxed) => {
+				let (value, error) = *boxed;
+				let first = self.error.take().unwrap_or(error);
+				GPoll::Fallback(Box::new((value, first)))
+			}
+			interrupted => interrupted,
+		}
+	}
+}
+
+#[derive(Clone, Copy)]
+pub struct LazyInput<'a, N> {
+	node: &'a N,
+	cell: &'a StatusCell,
+	input_index: usize,
+}
+
+impl<'a, N> LazyInput<'a, N> {
+	pub fn new(node: &'a N, cell: &'a StatusCell, input_index: usize) -> Self {
+		Self { node, cell, input_index }
+	}
+
+	pub fn eval<Input>(&self, ctx: &Input) -> Result<N::Output, Interrupt>
+	where
+		N: Node<Input>,
+	{
+		self.cell.eval_input(self.input_index, self.node, ctx)
+	}
+}
+
+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::*;
+	use std::sync::atomic::{AtomicU32, Ordering};
+
+	#[derive(Clone, Copy)]
+	struct TestInput {
+		index: u64,
+	}
+
+	impl InjectIndex for TestInput {
+		fn set_index(&mut self, index: u64) {
+			self.index = index;
+		}
+	}
+
+	struct Double;
+
+	impl Node<TestInput> for Double {
+		type Output = u64;
+
+		fn eval(&self, input: &TestInput) -> GPoll<u64> {
+			GPoll::Final(input.index * 2)
+		}
+	}
+
+	#[test]
+	fn spec_loop_fills_scratch_per_lane() {
+		let input = TestInput { index: 0 };
+		let mut scratch = [const { MaybeUninit::uninit() }; 4];
+		let status = Double.eval_batch(&input, 2..6, Some(&mut scratch));
+		let BatchStatus::Filled(lanes, finality) = status else {
+			panic!("expected filled, got {status:?}");
+		};
+		assert_eq!(lanes, &[4, 6, 8, 10]);
+		assert_eq!(finality, Finality::AllFinal);
+	}
+
+	#[test]
+	fn probe_without_scratch_requests_a_buffer() {
+		let input = TestInput { index: 0 };
+		assert!(matches!(Double.eval_batch(&input, 0..4, None), BatchStatus::NeedBuffer));
+	}
+
+	#[test]
+	fn undersized_scratch_is_an_invalid_range() {
+		let input = TestInput { index: 0 };
+		let mut scratch = [const { MaybeUninit::uninit() }; 2];
+		assert!(matches!(Double.eval_batch(&input, 0..4, Some(&mut scratch)), BatchStatus::InvalidRange));
+	}
+
+	#[test]
+	fn partial_lane_downgrades_batch_finality() {
+		struct PartialAtThree;
+		impl Node<TestInput> for PartialAtThree {
+			type Output = u64;
+			fn eval(&self, input: &TestInput) -> GPoll<u64> {
+				match input.index {
+					3 => GPoll::Partial(input.index),
+					index => GPoll::Final(index),
+				}
+			}
+		}
+		let input = TestInput { index: 0 };
+		let mut scratch = [const { MaybeUninit::uninit() }; 4];
+		let status = PartialAtThree.eval_batch(&input, 0..4, Some(&mut scratch));
+		let BatchStatus::Filled(lanes, finality) = status else {
+			panic!("expected filled, got {status:?}");
+		};
+		assert_eq!(lanes, &[0, 1, 2, 3]);
+		assert_eq!(finality, Finality::Partial);
+	}
+
+	#[test]
+	fn abort_drops_already_written_lanes() {
+		static DROPS: AtomicU32 = AtomicU32::new(0);
+		struct Probe;
+		impl Drop for Probe {
+			fn drop(&mut self) {
+				DROPS.fetch_add(1, Ordering::Relaxed);
+			}
+		}
+		struct PendingAtTwo;
+		impl Node<TestInput> for PendingAtTwo {
+			type Output = Probe;
+			fn eval(&self, input: &TestInput) -> GPoll<Probe> {
+				match input.index {
+					2 => GPoll::Pending,
+					_ => GPoll::Final(Probe),
+				}
+			}
+		}
+		let input = TestInput { index: 0 };
+		let mut scratch = [const { MaybeUninit::uninit() }; 4];
+		let status = PendingAtTwo.eval_batch(&input, 0..4, Some(&mut scratch));
+		assert!(matches!(status, BatchStatus::Pending));
+		assert_eq!(DROPS.load(Ordering::Relaxed), 2);
+	}
+
+	#[test]
+	fn trait_is_object_safe_across_erased_edges() {
+		let erased: Box<dyn Node<TestInput, Output = u64>> = Box::new(Double);
+		let input = TestInput { index: 21 };
+		assert_eq!(erased.eval(&input), GPoll::Final(42));
+		let mut scratch = [const { MaybeUninit::uninit() }; 2];
+		let status = erased.eval_batch(&input, 0..2, Some(&mut scratch));
+		assert!(matches!(status, BatchStatus::Filled(_, Finality::AllFinal)));
+	}
+}

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

@@ -0,0 +1,148 @@
+use crate::SourceId;
+use std::collections::HashMap;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, PoisonError};
+
+#[cfg(not(target_family = "wasm"))]
+pub type SourceFuture<T = ()> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
+#[cfg(target_family = "wasm")]
+pub type SourceFuture<T = ()> = Pin<Box<dyn Future<Output = T> + 'static>>;
+
+#[cfg(not(target_family = "wasm"))]
+pub type DynRuntime = dyn Runtime + Send + Sync;
+#[cfg(target_family = "wasm")]
+pub type DynRuntime = dyn Runtime;
+
+pub trait Runtime {
+	fn spawn(&self, source: SourceId, future: SourceFuture);
+}
+
+#[derive(Clone)]
+pub struct RuntimeHandle(pub Arc<DynRuntime>);
+
+// SAFETY: wasm is single threaded, so the handle never actually crosses a thread.
+#[cfg(target_family = "wasm")]
+unsafe impl Send for RuntimeHandle {}
+// SAFETY: as in Send.
+#[cfg(target_family = "wasm")]
+unsafe impl Sync for RuntimeHandle {}
+
+impl std::fmt::Debug for RuntimeHandle {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("RuntimeHandle").finish_non_exhaustive()
+	}
+}
+
+impl graphene_hash::CacheHash for RuntimeHandle {
+	fn cache_hash<H: std::hash::Hasher>(&self, _state: &mut H) {}
+}
+
+pub trait Spawner {
+	fn spawn(&self, task: SourceFuture);
+}
+
+#[cfg(not(target_family = "wasm"))]
+pub type DynSpawner = dyn Spawner + Send + Sync;
+#[cfg(target_family = "wasm")]
+pub type DynSpawner = dyn Spawner;
+
+#[cfg(not(target_family = "wasm"))]
+pub type DynNotifier = dyn Fn() + Send + Sync;
+#[cfg(target_family = "wasm")]
+pub type DynNotifier = dyn Fn();
+
+impl<S: Spawner + ?Sized> Spawner for Box<S> {
+	fn spawn(&self, task: SourceFuture) {
+		(**self).spawn(task)
+	}
+}
+
+/// Dropped tasks never complete.
+pub struct NoopSpawner;
+
+impl Spawner for NoopSpawner {
+	fn spawn(&self, _task: SourceFuture) {
+		log::warn!("async source spawned before a host spawner is wired; the task is dropped");
+	}
+}
+
+pub type DynGraphRuntime = GraphRuntime<Box<DynSpawner>>;
+
+impl Default for RuntimeHandle {
+	fn default() -> Self {
+		Self(Arc::new(GraphRuntime::new(Box::new(NoopSpawner) as Box<DynSpawner>)))
+	}
+}
+
+pub struct GraphRuntime<S> {
+	generations: Arc<Mutex<HashMap<SourceId, u64>>>,
+	dirty: Arc<AtomicBool>,
+	notifier: Arc<Mutex<Arc<DynNotifier>>>,
+	spawner: S,
+}
+
+// SAFETY: wasm is single threaded, so the runtime never actually crosses a thread.
+#[cfg(target_family = "wasm")]
+unsafe impl<S> Send for GraphRuntime<S> {}
+// SAFETY: as in Send.
+#[cfg(target_family = "wasm")]
+unsafe impl<S> Sync for GraphRuntime<S> {}
+
+impl<S> GraphRuntime<S> {
+	pub fn new(spawner: S) -> Self {
+		Self {
+			generations: Arc::default(),
+			dirty: Arc::default(),
+			notifier: Arc::new(Mutex::new(Arc::new(|| {}))),
+			spawner,
+		}
+	}
+
+	pub fn set_notifier(&self, notifier: Arc<DynNotifier>) {
+		*self.notifier.lock().unwrap_or_else(PoisonError::into_inner) = notifier;
+	}
+
+	pub fn retain_sources(&self, live: &[SourceId]) {
+		let mut generations = self.generations.lock().unwrap_or_else(PoisonError::into_inner);
+		generations.retain(|source, _| live.contains(source));
+		for source in live {
+			generations.entry(*source).or_insert(0);
+		}
+	}
+
+	pub fn snapshot(&self) -> Vec<(SourceId, u64)> {
+		let generations = self.generations.lock().unwrap_or_else(PoisonError::into_inner);
+		let mut snapshot: Vec<_> = generations.iter().map(|(source, generation)| (*source, *generation)).collect();
+		snapshot.sort_unstable();
+		snapshot
+	}
+
+	pub fn take_dirty(&self) -> bool {
+		self.dirty.swap(false, Ordering::Acquire)
+	}
+
+	pub fn spawner(&self) -> &S {
+		&self.spawner
+	}
+}
+
+impl<S: Spawner> Runtime for GraphRuntime<S> {
+	fn spawn(&self, source: SourceId, future: SourceFuture) {
+		let generations = Arc::clone(&self.generations);
+		let dirty = Arc::clone(&self.dirty);
+		let notifier = Arc::clone(&self.notifier);
+		self.spawner.spawn(Box::pin(async move {
+			future.await;
+			let mut generations = generations.lock().unwrap_or_else(PoisonError::into_inner);
+			if let Some(generation) = generations.get_mut(&source) {
+				*generation += 1;
+				dirty.store(true, Ordering::Release);
+				drop(generations);
+				let notifier = Arc::clone(&notifier.lock().unwrap_or_else(PoisonError::into_inner));
+				notifier();
+			}
+		}));
+	}
+}

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

@@ -235,6 +235,7 @@ pub enum Type {
 	Fn(Box<Type>, Box<Type>),
 	/// Represents a future which promises to return the inner type.
 	Future(Box<Type>),
+	Ref(Box<Type>),
 }
 
 impl Default for Type {
@@ -308,6 +309,7 @@ impl Type {
 			Self::Concrete(ty) => Some(ty.size),
 			Self::Fn(_, _) => None,
 			Self::Future(_) => None,
+			Self::Ref(_) => None,
 		}
 	}
 
@@ -317,6 +319,7 @@ impl Type {
 			Self::Concrete(ty) => Some(ty.align),
 			Self::Fn(_, _) => None,
 			Self::Future(_) => None,
+			Self::Ref(_) => None,
 		}
 	}
 
@@ -326,6 +329,7 @@ impl Type {
 			Self::Concrete(_) => self,
 			Self::Fn(_, output) => output.nested_type(),
 			Self::Future(output) => output.nested_type(),
+			Self::Ref(inner) => inner.nested_type(),
 		}
 	}
 
@@ -338,6 +342,7 @@ impl Type {
 			Self::Concrete(_) => None,
 			Self::Fn(_, output) => output.replace_nested(f),
 			Self::Future(output) => output.replace_nested(f),
+			Self::Ref(inner) => inner.replace_nested(f),
 		}
 	}
 
@@ -347,6 +352,7 @@ impl Type {
 			Type::Concrete(ty) => simplify_identifier_name(&ty.name),
 			Type::Fn(call_arg, return_value) => format!("{} called with {}", return_value.identifier_name(), call_arg.identifier_name()),
 			Type::Future(ty) => ty.identifier_name(),
+			Type::Ref(ty) => ty.identifier_name(),
 		}
 	}
 }
@@ -441,6 +447,7 @@ impl std::fmt::Display for Type {
 			Type::Concrete(ty) => write!(f, "{ty}"),
 			Type::Fn(_, return_value) => write!(f, "{return_value}"),
 			Type::Future(ty) => write!(f, "{ty}"),
+			Type::Ref(ty) => write!(f, "{ty}"),
 		}
 	}
 }

node-graph/libraries/graphene-hash/src/lib.rs

@@ -68,6 +68,7 @@ impl_via_hash! {
 #[cfg(feature = "std")]
 impl_via_hash! {
 	String,
+	core::time::Duration,
 }
 
 impl<'a> CacheHash for std::borrow::Cow<'a, str> {
@@ -235,3 +236,142 @@ impl_tuple!(A, B, C);
 impl_tuple!(A, B, C, D);
 impl_tuple!(A, B, C, D, E);
 impl_tuple!(A, B, C, D, E, F);
+
+/// rustc-hash's polynomial hash with the state pinned to u64, so keys match across native and wasm targets.
+/// The state starts at a nonzero seed: zero-initialized fx absorbs leading zero words, which produced
+/// a real wrong-value memo hit in the prototype.
+#[derive(Clone)]
+pub struct FxHasher64 {
+	hash: u64,
+}
+
+const K: u64 = 0xf1357aea2e62a9c5;
+const SEED: u64 = 0x517cc1b727220a95;
+const SEED1: u64 = 0x243f6a8885a308d3;
+const SEED2: u64 = 0x13198a2e03707344;
+const PREVENT_TRIVIAL_ZERO_COLLAPSE: u64 = 0xa4093822299f31d0;
+
+impl Default for FxHasher64 {
+	fn default() -> Self {
+		Self::new()
+	}
+}
+
+impl FxHasher64 {
+	pub const fn new() -> Self {
+		Self { hash: SEED }
+	}
+
+	#[inline]
+	fn add_to_hash(&mut self, i: u64) {
+		self.hash = self.hash.wrapping_add(i).wrapping_mul(K);
+	}
+}
+
+impl core::hash::Hasher for FxHasher64 {
+	#[inline]
+	fn write(&mut self, bytes: &[u8]) {
+		self.add_to_hash(hash_bytes(bytes));
+	}
+
+	#[inline]
+	fn write_u8(&mut self, i: u8) {
+		self.add_to_hash(i as u64);
+	}
+
+	#[inline]
+	fn write_u16(&mut self, i: u16) {
+		self.add_to_hash(i as u64);
+	}
+
+	#[inline]
+	fn write_u32(&mut self, i: u32) {
+		self.add_to_hash(i as u64);
+	}
+
+	#[inline]
+	fn write_u64(&mut self, i: u64) {
+		self.add_to_hash(i);
+	}
+
+	#[inline]
+	fn write_u128(&mut self, i: u128) {
+		self.add_to_hash(i as u64);
+		self.add_to_hash((i >> 64) as u64);
+	}
+
+	#[inline]
+	fn write_usize(&mut self, i: usize) {
+		self.add_to_hash(i as u64);
+	}
+
+	#[inline]
+	fn finish(&self) -> u64 {
+		self.hash.rotate_left(26)
+	}
+}
+
+#[inline]
+fn multiply_mix(x: u64, y: u64) -> u64 {
+	let full = (x as u128) * (y as u128);
+	(full as u64) ^ ((full >> 64) as u64)
+}
+
+#[inline]
+fn hash_bytes(bytes: &[u8]) -> u64 {
+	let len = bytes.len();
+	let mut s0 = SEED1;
+	let mut s1 = SEED2;
+
+	if len <= 16 {
+		if len >= 8 {
+			s0 ^= u64::from_le_bytes(bytes[0..8].try_into().unwrap());
+			s1 ^= u64::from_le_bytes(bytes[len - 8..].try_into().unwrap());
+		} else if len >= 4 {
+			s0 ^= u32::from_le_bytes(bytes[0..4].try_into().unwrap()) as u64;
+			s1 ^= u32::from_le_bytes(bytes[len - 4..].try_into().unwrap()) as u64;
+		} else if len > 0 {
+			let lo = bytes[0];
+			let mid = bytes[len / 2];
+			let hi = bytes[len - 1];
+			s0 ^= lo as u64;
+			s1 ^= ((hi as u64) << 8) | mid as u64;
+		}
+	} else {
+		let mut off = 0;
+		while off < len - 16 {
+			let x = u64::from_le_bytes(bytes[off..off + 8].try_into().unwrap());
+			let y = u64::from_le_bytes(bytes[off + 8..off + 16].try_into().unwrap());
+			let t = multiply_mix(s0 ^ x, PREVENT_TRIVIAL_ZERO_COLLAPSE ^ y);
+			s0 = s1;
+			s1 = t;
+			off += 16;
+		}
+
+		let suffix = &bytes[len - 16..];
+		s0 ^= u64::from_le_bytes(suffix[0..8].try_into().unwrap());
+		s1 ^= u64::from_le_bytes(suffix[8..16].try_into().unwrap());
+	}
+
+	multiply_mix(s0, s1) ^ (len as u64)
+}
+
+#[cfg(test)]
+mod tests {
+	use super::FxHasher64;
+	use core::hash::Hasher;
+
+	#[test]
+	fn leading_zero_words_are_not_absorbed() {
+		let hash_words = |words: &[u64]| {
+			let mut hasher = FxHasher64::new();
+			for &word in words {
+				hasher.write_u64(word);
+			}
+			hasher.finish()
+		};
+		assert_ne!(hash_words(&[]), hash_words(&[0]), "a zero word must change the hash of the empty input");
+		assert_ne!(hash_words(&[0]), hash_words(&[0, 0]), "zero words must accumulate distinct states");
+		assert_ne!(hash_words(&[0, 7]), hash_words(&[7]), "a leading zero word must not be absorbed");
+	}
+}