diff --git a/crates/chainfold/benches/apply.rs b/crates/chainfold/benches/apply.rs index 2bf0be6..e89aa22 100644 --- a/crates/chainfold/benches/apply.rs +++ b/crates/chainfold/benches/apply.rs @@ -5,10 +5,8 @@ use std::hint::black_box; use chainfold::{ Batch, BlockRef, - BlockSpan, Engine, EngineConfig, - LogEvent, test_util::NoopFold, }; use criterion::{ @@ -24,9 +22,9 @@ const RING_CAPACITY: usize = 128; /// Blocks the timed batch spans. const SPAN_COUNT: u64 = 64; /// Events carried by each timed span. -const EVENTS_PER_SPAN: u64 = 64; +const EVENTS_PER_SPAN: u32 = 64; /// Total events the timed batch carries. -const EVENT_COUNT: u64 = SPAN_COUNT * EVENTS_PER_SPAN; +const EVENT_COUNT: u64 = SPAN_COUNT * EVENTS_PER_SPAN as u64; /// Builds a distinguishable header for a block number. fn block_ref(number: u64) -> BlockRef { @@ -43,18 +41,8 @@ fn warmed_engine() -> Engine { checkpoint_slots: 0, }; let mut engine = Engine::new(NoopFold, config).expect("engine config is valid"); - let warmup = Batch { - boundary: None, - spans: vec![BlockSpan { - block: block_ref(0), - start: 0, - end: 1, - }], - events: vec![LogEvent { - log_index: 0, - event: 0, - }], - }; + let mut warmup = Batch::new(); + warmup.push_block(block_ref(0), [(0u32, 0u64)]); engine .apply_batch(&warmup) .expect("warmup batch applies cleanly"); @@ -63,28 +51,15 @@ fn warmed_engine() -> Engine { /// Builds the batch under measurement: 4096 events over 64 spans past the warmup block. fn timed_batch() -> Batch { - let mut events = Vec::with_capacity(EVENT_COUNT as usize); - let mut spans = Vec::with_capacity(SPAN_COUNT as usize); + let mut batch = Batch::new(); + batch.boundary = Some(block_ref(0)); for block in 1..=SPAN_COUNT { - let start = events.len() as u32; - for log_index in 0..EVENTS_PER_SPAN { - events.push(LogEvent { - log_index, - event: log_index, - }); - } - let end = events.len() as u32; - spans.push(BlockSpan { - block: block_ref(block), - start, - end, - }); - } - Batch { - boundary: Some(block_ref(0)), - spans, - events, + batch.push_block( + block_ref(block), + (0..EVENTS_PER_SPAN).map(|log_index| (log_index, u64::from(log_index))), + ); } + batch } fn bench_apply(c: &mut Criterion) { diff --git a/crates/chainfold/benches/snapshot.rs b/crates/chainfold/benches/snapshot.rs index df3a20a..11f74ab 100644 --- a/crates/chainfold/benches/snapshot.rs +++ b/crates/chainfold/benches/snapshot.rs @@ -5,10 +5,8 @@ use std::hint::black_box; use chainfold::{ Batch, BlockRef, - BlockSpan, Engine, EngineConfig, - LogEvent, test_util::RecordingFold, }; use criterion::{ @@ -21,7 +19,7 @@ use criterion::{ /// Observed-block ring window; only one block is ever observed in this benchmark. const RING_CAPACITY: usize = 8; /// Recorded-fold entries the envelope carries. -const ENTRY_COUNT: u64 = 4096; +const ENTRY_COUNT: u32 = 4096; /// Builds a distinguishable header for a block number. fn block_ref(number: u64) -> BlockRef { @@ -38,23 +36,11 @@ fn recorded_engine() -> Engine { }; let mut engine = Engine::new(RecordingFold::default(), config).expect("engine config is valid"); - let mut events = Vec::with_capacity(ENTRY_COUNT as usize); - for log_index in 0..ENTRY_COUNT { - events.push(LogEvent { - log_index, - event: log_index, - }); - } - let end = events.len() as u32; - let batch = Batch { - boundary: None, - spans: vec![BlockSpan { - block: block_ref(1), - start: 0, - end, - }], - events, - }; + let mut batch = Batch::new(); + batch.push_block( + block_ref(1), + (0..ENTRY_COUNT).map(|log_index| (log_index, u64::from(log_index))), + ); engine.apply_batch(&batch).expect("apply_batch succeeds"); engine } diff --git a/crates/chainfold/src/batch.rs b/crates/chainfold/src/batch.rs index 29e896e..d84b521 100644 --- a/crates/chainfold/src/batch.rs +++ b/crates/chainfold/src/batch.rs @@ -7,35 +7,39 @@ use core::fmt; use crate::position::BlockRef; -/// One event with its log index; the block number lives on the owning span. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct LogEvent { - /// Index of the log within its block. - pub log_index: u64, - /// Consumer event decoded from the log. - pub event: E, +/// One block's events within a batch. +#[derive(Debug)] +pub struct SpanView<'a, E> { + /// Number of the block the events belong to. + pub number: u64, + /// Hash of the block the events belong to. + pub hash: &'a [u8; 32], + /// Log index of each event within the block, strictly ascending. + pub log_indices: &'a [u32], + /// Consumer events, parallel to `log_indices`. + pub events: &'a [E], } -/// Half-open range of events belonging to one observed block. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct BlockSpan { - /// Block the events belong to. - pub block: BlockRef, - /// First event index in the batch's event array. - pub start: u32, - /// One past the last event index in the batch's event array. - pub end: u32, +impl SpanView<'_, E> { + /// Owned header of the span's block. + pub fn block(&self) -> BlockRef { + BlockRef { + number: self.number, + hash: *self.hash, + } + } } -/// One poll's worth of events in flat layout, reusable across polls. +/// One poll's worth of events #[derive(Debug, Clone, PartialEq, Eq)] pub struct Batch { /// Refetched header of the cursor block; None means the source could not produce it. pub boundary: Option, - /// One span per observed block, ascending, each covering a contiguous event range. - pub spans: Vec, - /// Every event of the batch, ordered by block then log index. - pub events: Vec>, + blocks: Vec, + hashes: Vec<[u8; 32]>, + ends: Vec, + log_indices: Vec, + events: Vec, } impl Batch { @@ -43,69 +47,137 @@ impl Batch { pub fn new() -> Self { Self { boundary: None, - spans: Vec::new(), + blocks: Vec::new(), + hashes: Vec::new(), + ends: Vec::new(), + log_indices: Vec::new(), events: Vec::new(), } } - /// Empties spans, events, and boundary while keeping capacity. + /// Empties every lane and the boundary while keeping capacity. pub fn clear(&mut self) { self.boundary = None; - self.spans.clear(); + self.blocks.clear(); + self.hashes.clear(); + self.ends.clear(); + self.log_indices.clear(); self.events.clear(); } /// True when the batch carries no spans. pub fn is_empty(&self) -> bool { - self.spans.is_empty() + self.ends.is_empty() + } + + /// Spans the batch carries, one per observed block. + pub fn span_count(&self) -> usize { + self.ends.len() } - /// Validates the flat batch shape against the rules of record. + /// Appends one block's complete event set in log order. + pub fn push_block( + &mut self, + block: BlockRef, + events: impl IntoIterator, + ) { + self.blocks.push(block.number); + self.hashes.push(block.hash); + for (log_index, event) in events { + self.log_indices.push(log_index); + self.events.push(event); + } + self.ends.push(self.events.len()); + } + + /// Iterates spans oldest first with their event slices. + pub fn spans(&self) -> Spans<'_, E> { + Spans { + batch: self, + next: 0, + start: 0, + } + } + + /// Validates the orderings a source controls, before any event reaches the fold. pub fn validate(&self) -> Result<(), BatchShapeError> { - if u32::try_from(self.events.len()).is_err() { - return Err(BatchShapeError::TooManyEvents { - len: self.events.len(), - }); + if let Some(span) = first_descent(&self.blocks) { + return Err(BatchShapeError::BlocksNotAscending { span }); } - let mut previous: Option<&BlockSpan> = None; - for (span_index, span) in self.spans.iter().enumerate() { - if span.start >= span.end || span.end as usize > self.events.len() { - return Err(BatchShapeError::SpanBoundsInvalid { span: span_index }); - } - if span.start != previous.map_or(0, |previous| previous.end) { - return Err(BatchShapeError::SpansNotContiguous { span: span_index }); + let mut start = 0usize; + for (span, end) in self.ends.iter().copied().enumerate() { + if end <= start { + return Err(BatchShapeError::SpanEmpty { span }); } - if previous.is_some_and(|previous| span.block.number <= previous.block.number) - { - return Err(BatchShapeError::BlocksNotAscending { span: span_index }); - } - let events = &self.events[span.start as usize..span.end as usize]; - let ascending = events - .iter() - .zip(&events[1..]) - .fold(true, |acc, (a, b)| acc & (a.log_index < b.log_index)); - if !ascending { - let offset = events - .windows(2) - .position(|pair| pair[1].log_index <= pair[0].log_index) - .expect("a failed ascending sweep always has a locatable pair"); + if let Some(offset) = first_descent(&self.log_indices[start..end]) { return Err(BatchShapeError::LogIndexNotAscending { - span: span_index, - #[allow(clippy::cast_possible_truncation)] - index: span.start + 1 + offset as u32, + span, + index: start + offset, }); } - previous = Some(span); - } - if previous.map_or(0, |span| span.end) as usize != self.events.len() { - return Err(BatchShapeError::SpansNotContiguous { - span: self.spans.len().saturating_sub(1), - }); + start = end; } Ok(()) } } +/// Index of the first element that does not exceed its predecessor. +/// +/// The sweep is branchless so it vectorizes; the cold locate pass runs only on failure. +pub(crate) fn first_descent(values: &[T]) -> Option { + let tail = values.get(1..).unwrap_or_default(); + let ascending = values + .iter() + .zip(tail) + .fold(true, |acc, (a, b)| acc & (a < b)); + if ascending { + return None; + } + locate_descent(values) +} + +#[cold] +fn locate_descent(values: &[T]) -> Option { + values + .windows(2) + .position(|pair| pair[1] <= pair[0]) + .map(|offset| offset + 1) +} + +/// Oldest-first iterator over a batch's spans. +#[derive(Debug)] +pub struct Spans<'a, E> { + batch: &'a Batch, + next: usize, + start: usize, +} + +impl<'a, E> Iterator for Spans<'a, E> { + type Item = SpanView<'a, E>; + + fn next(&mut self) -> Option { + let end = *self.batch.ends.get(self.next)?; + let range = self.start..end; + let number = self.batch.blocks[self.next]; + let hash = &self.batch.hashes[self.next]; + self.next += 1; + self.start = end; + Some(SpanView { + number, + hash, + log_indices: &self.batch.log_indices[range.clone()], + events: &self.batch.events[range], + }) + } + + fn size_hint(&self) -> (usize, Option) { + let remaining = self.batch.ends.len() - self.next; + (remaining, Some(remaining)) + } +} + +impl ExactSizeIterator for Spans<'_, E> {} + impl Default for Batch { fn default() -> Self { Self::new() @@ -115,18 +187,8 @@ impl Default for Batch { /// Batch layout violation found before any event reaches the fold. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum BatchShapeError { - /// Batch carries more events than an index can address. - TooManyEvents { - /// Events the batch carries. - len: usize, - }, - /// Span is empty or reaches past the event array. - SpanBoundsInvalid { - /// Index of the offending span. - span: usize, - }, - /// Span does not start where its predecessor ended. - SpansNotContiguous { + /// Span carries no events; a block with none is left out of the batch. + SpanEmpty { /// Index of the offending span. span: usize, }, @@ -139,22 +201,16 @@ pub enum BatchShapeError { LogIndexNotAscending { /// Index of the offending span. span: usize, - /// Index of the offending event in the batch's event array. - index: u32, + /// Index of the offending event in the batch's event lane. + index: usize, }, } impl fmt::Display for BatchShapeError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - Self::TooManyEvents { len } => { - write!(f, "batch carries {len} events, exceeding u32::MAX") - } - Self::SpanBoundsInvalid { span } => { - write!(f, "span {span} is empty or has out-of-range bounds") - } - Self::SpansNotContiguous { span } => { - write!(f, "span {span} is not contiguous with its neighbor") + Self::SpanEmpty { span } => { + write!(f, "span {span} carries no events") } Self::BlocksNotAscending { span } => { write!(f, "span {span} block number is not strictly ascending") @@ -176,19 +232,12 @@ mod tests { use super::{ Batch, BatchShapeError, - BlockSpan, }; use crate::position::BlockRef; #[cfg(not(feature = "std"))] - use alloc::{ - vec, - vec::Vec, - }; + use alloc::vec::Vec; #[cfg(feature = "std")] - use std::{ - vec, - vec::Vec, - }; + use std::vec::Vec; fn block(number: u64) -> BlockRef { BlockRef { @@ -197,85 +246,32 @@ mod tests { } } - fn events(indices: &[u64]) -> Vec> { - indices - .iter() - .map(|&log_index| super::LogEvent { - log_index, - event: log_index, - }) - .collect() + /// Builds a batch from one (block number, log indices) pair per span. + fn batch_of(spans: &[(u64, &[u32])]) -> Batch { + let mut batch = Batch::new(); + for (number, indices) in spans { + batch.push_block( + block(*number), + indices.iter().map(|index| (*index, u64::from(*index))), + ); + } + batch } #[test] fn valid_batch_passes_validation() { - // given two spans covering four events contiguously - let batch = Batch { - boundary: None, - spans: vec![ - BlockSpan { - block: block(1), - start: 0, - end: 2, - }, - BlockSpan { - block: block(2), - start: 2, - end: 4, - }, - ], - events: events(&[0, 1, 0, 1]), - }; + // given two spans covering four events + let batch = batch_of(&[(1, &[0, 1]), (2, &[0, 1])]); // when validated let result = batch.validate(); // then Ok assert_eq!(result, Ok(())); } - #[test] - fn batch_with_gap_between_spans_fails() { - // given span end 2 and next start 3 - let batch = Batch { - boundary: None, - spans: vec![ - BlockSpan { - block: block(1), - start: 0, - end: 2, - }, - BlockSpan { - block: block(2), - start: 3, - end: 4, - }, - ], - events: events(&[0, 1, 0, 1]), - }; - // when validated - let result = batch.validate(); - // then SpansNotContiguous at span 1 - assert_eq!(result, Err(BatchShapeError::SpansNotContiguous { span: 1 })); - } - #[test] fn batch_with_descending_blocks_fails() { // given spans at blocks 7 then 5 - let batch = Batch { - boundary: None, - spans: vec![ - BlockSpan { - block: block(7), - start: 0, - end: 1, - }, - BlockSpan { - block: block(5), - start: 1, - end: 2, - }, - ], - events: events(&[0, 0]), - }; + let batch = batch_of(&[(7, &[0]), (5, &[0])]); // when validated let result = batch.validate(); // then BlocksNotAscending at span 1 @@ -285,15 +281,7 @@ mod tests { #[test] fn batch_with_repeated_log_index_fails() { // given one span with log indices 3, 3 - let batch = Batch { - boundary: None, - spans: vec![BlockSpan { - block: block(1), - start: 0, - end: 2, - }], - events: events(&[3, 3]), - }; + let batch = batch_of(&[(1, &[3, 3])]); // when validated let result = batch.validate(); // then LogIndexNotAscending at span 0, event index 1 @@ -305,64 +293,49 @@ mod tests { #[test] fn batch_with_an_eventless_span_fails() { - // given a second span whose start equals its end - let batch = Batch { - boundary: None, - spans: vec![ - BlockSpan { - block: block(1), - start: 0, - end: 1, - }, - BlockSpan { - block: block(2), - start: 1, - end: 1, - }, - ], - events: events(&[0]), - }; + // given a second span pushed with no events + let batch = batch_of(&[(1, &[0]), (2, &[])]); // when validated let result = batch.validate(); - // then SpanBoundsInvalid at span 1 - assert_eq!(result, Err(BatchShapeError::SpanBoundsInvalid { span: 1 })); + // then SpanEmpty at span 1 + assert_eq!(result, Err(BatchShapeError::SpanEmpty { span: 1 })); } #[test] - fn batch_with_events_but_no_spans_fails() { - // given one event and zero spans - let batch = Batch { - boundary: None, - spans: vec![], - events: events(&[0]), - }; - // when validated - let result = batch.validate(); - // then SpansNotContiguous at span 0 - assert_eq!(result, Err(BatchShapeError::SpansNotContiguous { span: 0 })); + fn spans_yield_their_own_event_slices() { + // given three spans of differing width + let batch = batch_of(&[(1, &[0, 1, 2]), (4, &[7]), (9, &[0, 5])]); + // when iterating the spans + let seen: Vec<(u64, Vec, Vec)> = batch + .spans() + .map(|span| (span.number, span.log_indices.to_vec(), span.events.to_vec())) + .collect(); + // then each span carries exactly the events pushed with its block + assert_eq!( + seen, + [ + (1, [0, 1, 2].to_vec(), [0u64, 1, 2].to_vec()), + (4, [7].to_vec(), [7u64].to_vec()), + (9, [0, 5].to_vec(), [0u64, 5].to_vec()), + ] + ); + assert_eq!(batch.spans().len(), batch.span_count()); } #[test] fn batch_clear_keeps_capacity() { // given a filled batch - let mut batch = Batch { - boundary: Some(block(1)), - spans: vec![BlockSpan { - block: block(1), - start: 0, - end: 1, - }], - events: events(&[0]), - }; - let spans_capacity = batch.spans.capacity(); + let mut batch = batch_of(&[(1, &[0])]); + batch.boundary = Some(block(1)); + let spans_capacity = batch.blocks.capacity(); let events_capacity = batch.events.capacity(); // when cleared batch.clear(); // then empty with prior capacities assert!(batch.is_empty()); assert_eq!(batch.boundary, None); - assert!(batch.events.is_empty()); - assert_eq!(batch.spans.capacity(), spans_capacity); + assert_eq!(batch.spans().count(), 0); + assert_eq!(batch.blocks.capacity(), spans_capacity); assert_eq!(batch.events.capacity(), events_capacity); } } diff --git a/crates/chainfold/src/checkpoint.rs b/crates/chainfold/src/checkpoint.rs index 08394a3..d45ba5f 100644 --- a/crates/chainfold/src/checkpoint.rs +++ b/crates/chainfold/src/checkpoint.rs @@ -19,7 +19,11 @@ pub(crate) struct CheckpointRing { pub(crate) struct Slot { pub(crate) fold: F, pub(crate) cursor: Option, - pub(crate) ring: BlockRing, +} + +/// True when the engine's ring can still serve the slot's window. +fn live(slot: &Slot, ring: &BlockRing) -> bool { + slot.cursor.is_none_or(|cursor| ring.observes(cursor.block)) } impl CheckpointRing { @@ -30,14 +34,21 @@ impl CheckpointRing { } } - /// Occupied slots oldest first, since the next write position holds the oldest. - fn oldest_first(&self) -> impl DoubleEndedIterator> { + /// Live slots oldest first, since the next write position holds the oldest. + fn oldest_first<'a>( + &'a self, + ring: &'a BlockRing, + ) -> impl DoubleEndedIterator> { let (newest, oldest) = self.slots.split_at(self.next); - oldest.iter().chain(newest).flatten() + oldest + .iter() + .chain(newest) + .flatten() + .filter(move |slot| live(slot, ring)) } - pub(crate) fn count(&self) -> usize { - self.oldest_first().count() + pub(crate) fn count(&self, ring: &BlockRing) -> usize { + self.oldest_first(ring).count() } /// Stores a slot at the next write position; overwrites the oldest when full. @@ -50,15 +61,20 @@ impl CheckpointRing { self.next = (self.next + 1) % len; } - /// Oldest retained slot; the mirror of best_at_or_below's newest-first scan. - pub(crate) fn oldest(&self) -> Option<&Slot> { - self.oldest_first().next() + /// Oldest live slot; the mirror of best_at_or_below's newest-first scan. + pub(crate) fn oldest<'a>(&'a self, ring: &'a BlockRing) -> Option<&'a Slot> { + self.oldest_first(ring).next() } - /// Newest slot with cursor block at or below the argument; empty-cursor slots always qualify. + /// Newest live slot with cursor block at or below the argument; empty-cursor slots + /// always qualify. #[cold] - pub(crate) fn best_at_or_below(&self, block: u64) -> Option<&Slot> { - self.oldest_first() + pub(crate) fn best_at_or_below<'a>( + &'a self, + block: u64, + ring: &'a BlockRing, + ) -> Option<&'a Slot> { + self.oldest_first(ring) .rev() .find(|slot| slot.cursor.is_none_or(|cursor| cursor.block <= block)) } diff --git a/crates/chainfold/src/driver.rs b/crates/chainfold/src/driver.rs index 5accbcf..2d1042e 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -49,6 +49,11 @@ const DEFAULT_BACKOFF_MAX: Duration = Duration::from_secs(30); /// Default anchor divergence rollbacks tolerated before the typed terminal state. const DEFAULT_MAX_DIVERGENCE_RETRIES: u32 = 1; +/// True when `block` has reached the next interval step past the last marked block. +fn due(last: Option, block: u64, interval: u64) -> bool { + last.is_none_or(|last| block >= last.saturating_add(interval)) +} + /// Poll loop state machine a harness drives; owns cadence, backoff, recovery. pub trait Tickable { /// Advances the loop by one poll, apply, and recovery step. @@ -128,9 +133,7 @@ pub struct DriverConfig { /// nothing. None means caller-driven only. pub checkpoint_interval: Option, /// Blocks of durable-point progress between snapshot offers; None disables - /// offers. The durability margin of an offered snapshot is checkpoint coverage, - /// checkpoint_slots * checkpoint_interval blocks; size both for the deepest - /// reorg the fold must survive without a resync. + /// offers. pub snapshot_interval: Option, /// Anchor divergence rollbacks tolerated before the typed terminal state. pub max_divergence_retries: u32, @@ -365,10 +368,11 @@ where fn auto_checkpoint(&mut self) -> Option { let interval = self.config.checkpoint_interval?; let cursor = self.engine.cursor()?; - let due = self - .last_checkpoint_block - .is_none_or(|last| cursor.block >= last.saturating_add(interval)); - if due { self.run_checkpoint() } else { None } + if due(self.last_checkpoint_block, cursor.block, interval) { + self.run_checkpoint() + } else { + None + } } /// Stores a checkpoint, records its block, then runs the anchor check. @@ -387,10 +391,7 @@ where } let interval = self.config.snapshot_interval?; let point = self.engine.durable_point()?; - let due = self - .last_snapshot_block - .is_none_or(|last| point.block >= last.saturating_add(interval)); - if !due { + if !due(self.last_snapshot_block, point.block, interval) { return None; } match self.sink.offer(&self.engine) { @@ -778,18 +779,12 @@ mod tests { vec::Vec, }; - use crate::{ - batch::{ - BlockSpan, - LogEvent, - }, - test_util::{ - FailKind, - PollFailure, - RecordingFold, - ScriptedChain, - WatermarkSink, - }, + use crate::test_util::{ + FailKind, + PollFailure, + RecordingFold, + ScriptedChain, + WatermarkSink, }; /// Wraps a scripted chain, counting probes and optionally failing the next few. @@ -857,15 +852,7 @@ mod tests { ) -> Result<(), PollFailure> { out.clear(); out.boundary = cursor.map(|_| self.block); - out.spans.push(BlockSpan { - block: self.block, - start: 0, - end: 1, - }); - out.events.push(LogEvent { - log_index: 0, - event: 1, - }); + out.push_block(self.block, [(0u32, 1u64)]); Ok(()) } @@ -1341,13 +1328,37 @@ mod tests { checkpoint_interval: Some(4), ..DriverConfig::default() }; - let mut driver = new_driver(chain, engine_config(8), config); + let engine = EngineConfig { + ring_capacity: 16, + checkpoint_slots: 8, + }; + let mut driver = new_driver(chain, engine, config); // when driven to the tip run_to_idle(&mut driver); // then checkpoint_count is at least 3 assert!(driver.engine().checkpoint_count() >= 3); } + #[test] + fn checkpoints_expire_once_their_block_leaves_the_ring() { + // given checkpoint_interval 4 over twelve blocks with a ring holding only 8 + let mut chain = ScriptedChain::new(1); + for value in 1..=12u64 { + chain.push_block(&[value]); + } + chain.set_batch_blocks(1); + let config = DriverConfig { + checkpoint_interval: Some(4), + ..DriverConfig::default() + }; + let mut driver = new_driver(chain, engine_config(8), config); + // when driven to the tip, leaving the block 4 checkpoint outside the window + run_to_idle(&mut driver); + // then only the checkpoints the ring still observes are retained + assert_eq!(driver.engine().checkpoint_count(), 2); + assert_eq!(driver.engine().durable_point(), Some(Position::new(5, 0))); + } + #[test] fn halt_is_terminal_and_recoverable_via_engine_mut() { // given a fold that halts at block 3 after a checkpoint taken at block 2 diff --git a/crates/chainfold/src/engine.rs b/crates/chainfold/src/engine.rs index bca9750..67c6d73 100644 --- a/crates/chainfold/src/engine.rs +++ b/crates/chainfold/src/engine.rs @@ -1,8 +1,7 @@ use crate::{ batch::{ Batch, - BlockSpan, - LogEvent, + SpanView, }, checkpoint::{ CheckpointRing, @@ -118,21 +117,25 @@ impl Engine { self.skips } - /// Count of checkpoints currently retained. + /// Count of checkpoints the ring can still serve; expired slots are not counted. pub fn checkpoint_count(&self) -> usize { - self.checkpoints.count() + self.checkpoints.count(&self.ring) } - /// Cursor of the oldest retained checkpoint; the reorg-safe durable point. - /// None without checkpoints or when the oldest slot predates any applied event. + /// Cursor of the oldest live checkpoint; the reorg-safe durable point. + /// + /// None without checkpoints, when the oldest live slot predates any applied event, or + /// once every slot's cursor block has left the observed window. pub fn durable_point(&self) -> Option { - self.checkpoints.oldest().and_then(|slot| slot.cursor) + self.checkpoints + .oldest(&self.ring) + .and_then(|slot| slot.cursor) } - /// Oldest retained checkpoint slot, for the snapshot codec. + /// Oldest live checkpoint slot, for the snapshot codec. #[cfg(feature = "wincode")] pub(crate) fn oldest_checkpoint(&self) -> Option<&Slot> { - self.checkpoints.oldest() + self.checkpoints.oldest(&self.ring) } /// Borrows the fold state. @@ -150,6 +153,12 @@ impl Engine { self.ring.iter() } + /// Iterates observed blocks at or below a number, oldest first; a checkpoint's window. + #[cfg(feature = "wincode")] + pub(crate) fn observed_at_or_below(&self, number: u64) -> Observed<'_> { + self.ring.iter_at_or_below(number) + } + /// Applies one poll's batch: total order, dedup, boundary recheck, fork detection. pub fn apply_batch( &mut self, @@ -182,23 +191,27 @@ impl Engine { } let mut summary = ApplySummary::default(); - for span in &batch.spans { + for span in batch.spans() { let redelivered = self .cursor - .is_some_and(|cursor| span.block.number <= cursor.block); + .is_some_and(|cursor| span.number <= cursor.block); if redelivered - && let Some(observed_hash) = self.ring.hash_at(span.block.number) - && observed_hash != span.block.hash + && let Some(observed_hash) = self.ring.hash_at(span.number) + && &observed_hash != span.hash { - return Err(fork_suspected(span.block.number, observed_hash, span.block)); + return Err(fork_suspected(span.number, observed_hash, span.block())); } - self.apply_span(span, batch, &mut summary)?; + self.apply_span(&span, &mut summary)?; } Ok(summary) } - /// Stores a checkpoint of the current fold, cursor, and ring. + /// Stores a checkpoint of the current fold and cursor. + /// + /// A slot carries no block history of its own; its window is the engine's ring + /// truncated to its cursor, so rollback depth is bounded by ring_capacity and a slot + /// expires once its cursor block leaves the window. /// /// A no-op with zero slots or a non-Active status, so every stored slot holds /// state the engine still trusts. @@ -212,15 +225,17 @@ impl Engine { self.checkpoints.store(Slot { fold: self.fold.clone(), cursor: self.cursor, - ring: self.ring.clone(), }); } - /// Restores the newest checkpoint whose cursor block is at or below the argument. + /// Restores the newest live checkpoint whose cursor block is at or below the argument, + /// truncating the ring to that cursor. + /// /// Clears Halted and Poisoned; drops checkpoints above the argument, the fork /// boundary, so checkpoints between it and the restored cursor stay valid. /// Freshness resets to None, since the restored cursor is unverified until the - /// next boundary check confirms it. + /// next boundary check confirms it. Slots whose cursor block has left the observed + /// window are expired, so NoCheckpointAtOrBelow also names an exhausted window. #[cold] pub fn rollback_at_or_below( &mut self, @@ -234,11 +249,14 @@ impl Engine { } let slot = self .checkpoints - .best_at_or_below(block) + .best_at_or_below(block, &self.ring) .ok_or(RollbackError::NoCheckpointAtOrBelow { block })?; self.fold = slot.fold.clone(); self.cursor = slot.cursor; - self.ring = slot.ring.clone(); + match self.cursor { + Some(cursor) => self.ring.truncate_above(cursor.block), + None => self.ring.clear(), + } self.status = EngineStatus::Active; self.last_verified = None; self.checkpoints.drop_above(block); @@ -276,62 +294,61 @@ impl Engine { /// Applies every event of one span, deduping positions at or below the cursor. fn apply_span( &mut self, - span: &BlockSpan, - batch: &Batch, + span: &SpanView<'_, F::Event>, summary: &mut ApplySummary, ) -> Result<(), ApplyError> { - let events = &batch.events[span.start as usize..span.end as usize]; - let pos = |entry: &LogEvent| { - Position::new(span.block.number, entry.log_index) - }; + let number = span.number; + let pos = |log_index: u32| Position::new(number, u64::from(log_index)); let deduped = match self.cursor { - Some(cursor) if span.block.number > cursor.block => 0, - Some(cursor) if span.block.number < cursor.block => events.len(), - Some(cursor) => { - events.partition_point(|entry| entry.log_index <= cursor.log_index) - } + Some(cursor) if number > cursor.block => 0, + Some(cursor) if number < cursor.block => span.log_indices.len(), + Some(cursor) => span + .log_indices + .partition_point(|index| u64::from(*index) <= cursor.log_index), None => 0, }; summary.deduped += deduped as u64; - let fresh = &events[deduped..]; - let Some(last) = fresh.last() else { + // lanes are equal length by construction, so the paired walk drops no event + let indices = &span.log_indices[deduped..]; + let events = &span.events[deduped..]; + let Some(last) = indices.last() else { return Ok(()); }; - for (index, entry) in fresh.iter().enumerate() { - let at = pos(entry); - match self.fold.apply(at, &entry.event) { + for (offset, (log_index, event)) in indices.iter().zip(events).enumerate() { + let at = pos(*log_index); + match self.fold.apply(at, event) { Ok(()) => summary.applied += 1, Err(FoldError::Skip(_)) => { self.skips += 1; summary.skipped += 1; } Err(FoldError::Halt(error)) => { - self.consumed_through(span.block, fresh, &pos, index); + self.consumed_through(span, indices, offset); return Err(self.halt(at, error)); } Err(FoldError::Poison(error)) => { - self.consumed_through(span.block, fresh, &pos, index); + self.consumed_through(span, indices, offset); return Err(self.poison(at, error)); } } } - self.advance(span.block, pos(last)); + self.advance(span.number, span.hash, pos(*last)); Ok(()) } - /// Places the cursor at the predecessor of `fresh[index]`; a no-op at index 0, since - /// nothing in this span was consumed yet. + /// Places the cursor at the predecessor of `indices[offset]`; a no-op at offset 0, + /// since nothing in this span was consumed yet. #[cold] fn consumed_through( &mut self, - block: BlockRef, - fresh: &[LogEvent], - pos: &impl Fn(&LogEvent) -> Position, - index: usize, + span: &SpanView<'_, F::Event>, + indices: &[u32], + offset: usize, ) { - if let Some(entry) = index.checked_sub(1).and_then(|i| fresh.get(i)) { - self.advance(block, pos(entry)); + if let Some(log_index) = offset.checked_sub(1).and_then(|i| indices.get(i)) { + let pos = Position::new(span.number, u64::from(*log_index)); + self.advance(span.number, span.hash, pos); } } @@ -339,13 +356,16 @@ impl Engine { /// /// Ring and cursor move together, so the ring's newest entry is the cursor block /// at every point a batch can return from. - #[inline] - fn advance(&mut self, block: BlockRef, pos: Position) { + fn advance(&mut self, number: u64, hash: &[u8; 32], pos: Position) { if self .ring - .newest() - .is_none_or(|newest| newest.number < block.number) + .newest_number() + .is_none_or(|newest| newest < number) { + let block = BlockRef { + number, + hash: *hash, + }; self.ring.push(block); self.last_verified = Some(block); } @@ -386,10 +406,7 @@ fn fork_suspected( mod tests { use super::*; use crate::{ - batch::{ - BatchShapeError, - LogEvent, - }, + batch::BatchShapeError, test_util::{ FailKind, RecordingFold, @@ -415,26 +432,19 @@ mod tests { fn batch_of( boundary: Option, - spans: Vec<(BlockRef, Vec)>, + spans: Vec<(BlockRef, Vec)>, ) -> Batch { - let mut events = Vec::new(); - let mut built_spans = Vec::new(); + let mut batch = Batch::new(); + batch.boundary = boundary; for (block, log_indices) in spans { - let start = events.len() as u32; - for log_index in log_indices { - events.push(LogEvent { - log_index, - event: log_index, - }); - } - let end = events.len() as u32; - built_spans.push(BlockSpan { block, start, end }); - } - Batch { - boundary, - spans: built_spans, - events, + batch.push_block( + block, + log_indices + .into_iter() + .map(|index| (index, u64::from(index))), + ); } + batch } fn new_engine() -> Engine { @@ -828,43 +838,15 @@ mod tests { #[test] fn invalid_shape_is_rejected_before_fold_runs() { - // given a gap batch + // given a batch whose second span moves the block number backwards let mut engine = new_engine(); - let batch = Batch { - boundary: None, - spans: vec![ - BlockSpan { - block: block(1, 0), - start: 0, - end: 1, - }, - BlockSpan { - block: block(2, 0), - start: 2, - end: 3, - }, - ], - events: vec![ - LogEvent { - log_index: 0, - event: 0u64, - }, - LogEvent { - log_index: 0, - event: 0u64, - }, - LogEvent { - log_index: 0, - event: 0u64, - }, - ], - }; + let batch = batch_of(None, vec![(block(2, 0), vec![0]), (block(1, 0), vec![0])]); // when applied let result = engine.apply_batch(&batch); // then Shape and the fold recorded nothing assert_eq!( result, - Err(ApplyError::Shape(BatchShapeError::SpansNotContiguous { + Err(ApplyError::Shape(BatchShapeError::BlocksNotAscending { span: 1 })) ); @@ -1171,6 +1153,32 @@ mod tests { assert_eq!(point, Some(Position::new(7, 0))); } + #[test] + fn rollback_refuses_a_checkpoint_whose_block_left_the_ring() { + // given a checkpoint at block 1, then twelve blocks through a ring holding 8 + let mut engine = engine_with_checkpoints(2); + engine + .apply_batch(&batch_of(None, vec![(block(1, 0), vec![0])])) + .unwrap(); + engine.checkpoint(); + for number in 2..=12u64 { + engine + .apply_batch(&batch_of( + Some(block(number - 1, 0)), + vec![(block(number, 0), vec![0])], + )) + .unwrap(); + } + // when rolling back to the expired checkpoint's block + let result = engine.rollback_at_or_below(1); + // then it is refused rather than restored against a window that cannot serve it + assert_eq!( + result, + Err(RollbackError::NoCheckpointAtOrBelow { block: 1 }) + ); + assert_eq!(engine.checkpoint_count(), 0); + } + #[test] fn durable_point_is_none_without_checkpoints() { // given a fresh engine with no retained slots diff --git a/crates/chainfold/src/error.rs b/crates/chainfold/src/error.rs index e1ee863..dbf4581 100644 --- a/crates/chainfold/src/error.rs +++ b/crates/chainfold/src/error.rs @@ -86,11 +86,6 @@ impl core::error::Error for EngineStatus {} pub enum DivergenceCause { /// Fork deeper than the oldest observed block in the ring. ForkBeyondWindow, - /// A canonical ancestor exists but no retained checkpoint sits at or below it. - NoCheckpointBelowAncestor { - /// Deepest still-canonical observed block. - ancestor: u64, - }, /// Replay is required but the source's horizon no longer covers the start block. HorizonExceeded { /// Block replay must start from. @@ -109,9 +104,6 @@ impl fmt::Display for DivergenceCause { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::ForkBeyondWindow => write!(f, "fork deeper than the observed window"), - Self::NoCheckpointBelowAncestor { ancestor } => { - write!(f, "no checkpoint at or below ancestor block {ancestor}") - } Self::HorizonExceeded { needed, horizon } => { write!( f, diff --git a/crates/chainfold/src/harness.rs b/crates/chainfold/src/harness.rs index 1f59017..520de65 100644 --- a/crates/chainfold/src/harness.rs +++ b/crates/chainfold/src/harness.rs @@ -96,22 +96,28 @@ impl Handle { *self.status.borrow() } - /// Returns when caught up or terminal, regardless of when the transition happened. - pub async fn wait_caught_up(&mut self) -> DriverStatus { + /// Returns the first published status the predicate accepts, or the last one + /// published once the loop thread has dropped its sender. + async fn settled( + &mut self, + accept: impl FnMut(&DriverStatus) -> bool, + ) -> DriverStatus { self.status - .wait_for(|s| s.caught_up || s.is_terminal()) + .wait_for(accept) .await .map(|status| *status) .unwrap_or_else(|_| *self.status.borrow()) } + /// Returns when caught up or terminal, regardless of when the transition happened. + pub async fn wait_caught_up(&mut self) -> DriverStatus { + self.settled(|s| s.caught_up || s.is_terminal()).await + } + /// Returns when the cursor reaches `pos` or the driver is terminal. pub async fn wait_past(&mut self, pos: Position) -> DriverStatus { - self.status - .wait_for(|s| s.is_terminal() || s.cursor.is_some_and(|cursor| cursor >= pos)) + self.settled(|s| s.is_terminal() || s.cursor.is_some_and(|c| c >= pos)) .await - .map(|status| *status) - .unwrap_or_else(|_| *self.status.borrow()) } /// Returns when the durable cursor reaches `pos` or the driver is terminal. @@ -119,13 +125,8 @@ impl Handle { /// A later resync lowers the durable cursor, so the answer holds for the /// instant it resolves. pub async fn wait_durable(&mut self, pos: Position) -> DriverStatus { - self.status - .wait_for(|s| { - s.is_terminal() || s.durable_cursor.is_some_and(|cursor| cursor >= pos) - }) + self.settled(|s| s.is_terminal() || s.durable_cursor.is_some_and(|c| c >= pos)) .await - .map(|status| *status) - .unwrap_or_else(|_| *self.status.borrow()) } /// Asks the loop to checkpoint before its next tick. diff --git a/crates/chainfold/src/lib.rs b/crates/chainfold/src/lib.rs index 350335f..5a13ad0 100644 --- a/crates/chainfold/src/lib.rs +++ b/crates/chainfold/src/lib.rs @@ -61,8 +61,8 @@ pub use anchor::{ pub use batch::{ Batch, BatchShapeError, - BlockSpan, - LogEvent, + SpanView, + Spans, }; pub use driver::{ Driver, diff --git a/crates/chainfold/src/ring.rs b/crates/chainfold/src/ring.rs index d308c34..8e57c90 100644 --- a/crates/chainfold/src/ring.rs +++ b/crates/chainfold/src/ring.rs @@ -9,8 +9,6 @@ use std::{ vec, }; -use core::cmp::Ordering; - use crate::position::BlockRef; /// Bounded history of observed blocks, structure of arrays, power-of-two capacity. @@ -42,11 +40,6 @@ impl BlockRing { self.numbers.len() } - #[inline] - pub(crate) fn is_empty(&self) -> bool { - self.len == 0 - } - #[inline] fn physical(&self, logical: usize) -> usize { (self.head + logical) & (self.capacity() - 1) @@ -68,13 +61,11 @@ impl BlockRing { } } + /// Number of the newest entry, without reading the hash lane. #[inline] - pub(crate) fn newest(&self) -> Option { - if self.is_empty() { - None - } else { - Some(self.get(self.len - 1)) - } + pub(crate) fn newest_number(&self) -> Option { + let newest = self.len.checked_sub(1)?; + Some(self.numbers[self.physical(newest)]) } /// Oldest-first access; index must be below len. @@ -89,18 +80,57 @@ impl BlockRing { /// Hash for an exact number if still observed. pub(crate) fn hash_at(&self, number: u64) -> Option<[u8; 32]> { + let logical = self.index_of(number)?; + Some(self.hashes[self.physical(logical)]) + } + + /// Logical index of an exact number. + /// + /// Numbers ascend by at least one per entry, so the tip-relative guess is the + /// highest index the number can sit at; a gapped chain falls back to the search. + fn index_of(&self, number: u64) -> Option { + let newest = self.len.checked_sub(1)?; + let behind = self.numbers[self.physical(newest)].checked_sub(number)?; + let guess = usize::try_from(behind) + .ok() + .and_then(|behind| newest.checked_sub(behind)) + .unwrap_or(0); + if self.numbers[self.physical(guess)] == number { + return Some(guess); + } + let logical = self.index_at_or_below(number)?; + (self.numbers[self.physical(logical)] == number).then_some(logical) + } + + /// Drops entries with a number strictly above the argument. + #[cold] + pub(crate) fn truncate_above(&mut self, number: u64) { + self.len = match self.index_at_or_below(number) { + Some(logical) => logical + 1, + None => 0, + }; + if self.len == 0 { + self.head = 0; + } + } + + /// Logical index of the newest entry at or below the argument. + fn index_at_or_below(&self, number: u64) -> Option { + let newest = self.len.checked_sub(1)?; + if self.numbers[self.physical(newest)] <= number { + return Some(newest); + } let mut low = 0usize; - let mut high = self.len; + let mut high = newest; while low < high { - let mid = low + (high - low) / 2; - let physical = self.physical(mid); - match self.numbers[physical].cmp(&number) { - Ordering::Less => low = mid + 1, - Ordering::Equal => return Some(self.hashes[physical]), - Ordering::Greater => high = mid, + let mid = low + (high - low).div_ceil(2); + if self.numbers[self.physical(mid)] <= number { + low = mid; + } else { + high = mid - 1; } } - None + (self.numbers[self.physical(low)] <= number).then_some(low) } /// Empties the ring, keeping the allocation; stale slots stay unreachable below len. @@ -116,6 +146,23 @@ impl BlockRing { remaining: self.len, } } + + /// Oldest-first iterator over the entries at or below the argument. + #[cfg(any(feature = "wincode", test))] + pub(crate) fn iter_at_or_below(&self, number: u64) -> Observed<'_> { + Observed { + ring: self, + next: 0, + remaining: self + .index_at_or_below(number) + .map_or(0, |logical| logical + 1), + } + } + + /// True when the exact number is still observed. + pub(crate) fn observes(&self, number: u64) -> bool { + self.index_of(number).is_some() + } } /// Oldest-first iterator over observed blocks. @@ -143,11 +190,7 @@ impl<'a> Iterator for Observed<'a> { } } -impl<'a> ExactSizeIterator for Observed<'a> { - fn len(&self) -> usize { - self.remaining - } -} +impl ExactSizeIterator for Observed<'_> {} #[cfg(test)] mod tests { @@ -178,7 +221,7 @@ mod tests { ring.push(block(2)); ring.push(block(3)); // when reading newest - let newest = ring.newest(); + let newest = ring.iter().last(); // then the last push returns assert_eq!(newest, Some(block(3))); } @@ -210,6 +253,100 @@ mod tests { assert_eq!(ring.hash_at(5), Some(block(5).hash)); } + #[test] + fn hash_at_finds_numbers_across_a_gapped_ring() { + // given a ring whose numbers skip, defeating the tip-relative guess + let mut ring = BlockRing::with_capacity(8); + for number in [1u64, 2, 5, 9, 10, 40] { + ring.push(block(number)); + } + // when querying each stored number and two absent ones + let found: Vec> = [1u64, 2, 5, 9, 10, 40, 3, 39] + .iter() + .map(|n| ring.hash_at(*n)) + .collect(); + // then every stored number resolves and the gaps stay absent + assert_eq!( + found[..6], + [1u64, 2, 5, 9, 10, 40].map(|n| Some(block(n).hash)) + ); + assert_eq!(found[6], None); + assert_eq!(found[7], None); + } + + #[test] + fn hash_at_rejects_numbers_above_the_tip() { + // given a ring topping out at block 3 + let mut ring = BlockRing::with_capacity(4); + for number in 1..=3 { + ring.push(block(number)); + } + // when querying above the tip and on an empty ring + let above = ring.hash_at(4); + let empty = BlockRing::with_capacity(4).hash_at(1); + // then both are absent + assert_eq!(above, None); + assert_eq!(empty, None); + } + + #[test] + fn truncate_above_keeps_the_prefix_at_or_below() { + // given a wrapped ring holding blocks 3 through 6 + let mut ring = BlockRing::with_capacity(4); + for number in 1..=6 { + ring.push(block(number)); + } + // when truncating above block 4 + ring.truncate_above(4); + // then only blocks 3 and 4 remain, newest first at 4 + assert_eq!(ring.iter().collect::>(), vec![block(3), block(4)]); + assert_eq!(ring.iter().last(), Some(block(4))); + } + + #[test] + fn truncate_above_empties_when_every_entry_is_higher() { + // given a ring holding blocks 3 through 6 + let mut ring = BlockRing::with_capacity(4); + for number in 1..=6 { + ring.push(block(number)); + } + // when truncating below everything observed + ring.truncate_above(2); + // then the ring is empty and accepts pushes again + assert_eq!(ring.iter().len(), 0); + ring.push(block(9)); + assert_eq!(ring.iter().last(), Some(block(9))); + } + + #[test] + fn iter_at_or_below_yields_the_matching_prefix() { + // given a gapped ring + let mut ring = BlockRing::with_capacity(8); + for number in [2u64, 4, 7, 11] { + ring.push(block(number)); + } + // when iterating at or below a number between entries + let prefix: Vec = ring.iter_at_or_below(9).collect(); + // then the entries up to and including 7 appear, oldest first + assert_eq!(prefix, vec![block(2), block(4), block(7)]); + assert_eq!(ring.iter_at_or_below(1).count(), 0); + } + + #[test] + fn observes_reports_exact_membership() { + // given a ring holding blocks 3 through 6 + let mut ring = BlockRing::with_capacity(4); + for number in 1..=6 { + ring.push(block(number)); + } + // when asking about a retained and an evicted number + let retained = ring.observes(4); + let evicted = ring.observes(2); + // then only the retained one is observed + assert!(retained); + assert!(!evicted); + } + #[test] fn clone_is_independent() { // given a cloned ring @@ -219,7 +356,7 @@ mod tests { // when pushing to the original ring.push(block(2)); // then the clone is unchanged - assert_eq!(clone.newest(), Some(block(1))); - assert_eq!(ring.newest(), Some(block(2))); + assert_eq!(clone.iter().last(), Some(block(1))); + assert_eq!(ring.iter().last(), Some(block(2))); } } diff --git a/crates/chainfold/src/snapshot.rs b/crates/chainfold/src/snapshot.rs index e841072..c2a01f4 100644 --- a/crates/chainfold/src/snapshot.rs +++ b/crates/chainfold/src/snapshot.rs @@ -8,6 +8,7 @@ use std::vec::Vec; use core::fmt; use crate::{ + batch::first_descent, engine::{ Engine, EngineConfig, @@ -269,11 +270,11 @@ impl Engine { encode_envelope(self.fold(), self.cursor(), self.observed(), out) } - /// Encodes the oldest retained checkpoint; Ok(None) means nothing is retained and - /// nothing was appended. + /// Encodes the oldest live checkpoint; Ok(None) means nothing is live and nothing + /// was appended. /// /// The returned point is the encoded snapshot's cursor, trailing the live cursor by - /// the ring's checkpoint coverage. + /// the checkpoint span. pub fn encode_durable_snapshot( &self, out: &mut Vec, @@ -284,7 +285,8 @@ impl Engine { let Some(point) = slot.cursor else { return Ok(None); }; - encode_envelope(&slot.fold, slot.cursor, slot.ring.iter(), out)?; + let window = self.observed_at_or_below(point.block); + encode_envelope(&slot.fold, slot.cursor, window, out)?; Ok(Some(point)) } @@ -329,11 +331,7 @@ impl Engine { capacity: config.ring_capacity, }); } - if envelope - .ring_numbers - .windows(2) - .any(|pair| pair[0] >= pair[1]) - { + if first_descent(&envelope.ring_numbers).is_some() { return Err(SnapshotError::RingNotAscending); } // The capacity check above bounds the count, so the product cannot overflow; @@ -354,17 +352,12 @@ impl Engine { let fold = F::decode_state(&envelope.state).map_err(SnapshotError::State)?; let mut ring = BlockRing::with_capacity(config.ring_capacity); - let lanes = envelope - .ring_numbers - .iter() - .zip(envelope.ring_hashes.chunks_exact(HASH_LEN)); - for (&number, hash) in lanes { - ring.push(BlockRef { - number, - hash: hash - .try_into() - .map_err(|_| SnapshotError::RingHashLenMismatch)?, - }); + let hashes = envelope.ring_hashes.chunks_exact(HASH_LEN); + for (&number, hash) in envelope.ring_numbers.iter().zip(hashes) { + let hash = hash + .try_into() + .map_err(|_| SnapshotError::RingHashLenMismatch)?; + ring.push(BlockRef { number, hash }); } let mut engine = Engine::new(fold, config).map_err(SnapshotError::Config)?; @@ -409,11 +402,7 @@ mod tests { encode_custom, }; use crate::{ - batch::{ - Batch, - BlockSpan, - LogEvent, - }, + batch::Batch, engine::{ Engine, EngineConfig, @@ -455,26 +444,19 @@ mod tests { fn batch_of( boundary: Option, - spans: Vec<(BlockRef, Vec)>, + spans: Vec<(BlockRef, Vec)>, ) -> Batch { - let mut events = Vec::new(); - let mut built_spans = Vec::new(); + let mut batch = Batch::new(); + batch.boundary = boundary; for (block, log_indices) in spans { - let start = events.len() as u32; - for log_index in log_indices { - events.push(LogEvent { - log_index, - event: log_index, - }); - } - let end = events.len() as u32; - built_spans.push(BlockSpan { block, start, end }); - } - Batch { - boundary, - spans: built_spans, - events, + batch.push_block( + block, + log_indices + .into_iter() + .map(|index| (index, u64::from(index))), + ); } + batch } fn test_engine() -> Engine { diff --git a/crates/chainfold/src/storage/flusher.rs b/crates/chainfold/src/storage/flusher.rs index 8728c4d..c4cc701 100644 --- a/crates/chainfold/src/storage/flusher.rs +++ b/crates/chainfold/src/storage/flusher.rs @@ -112,6 +112,37 @@ impl Admission { } } +#[repr(align(128))] +struct Lane(T); + +/// State both the apply thread and the flusher thread reach, one allocation per flusher. +struct Shared { + admission: Lane, + poisoned: Lane, + durable_cursor: Lane>>, +} + +impl Shared { + fn new(queue_depth: usize, cursor: Option) -> Self { + Self { + admission: Lane(Admission::new(queue_depth)), + poisoned: Lane(AtomicBool::new(false)), + durable_cursor: Lane(Mutex::new(cursor)), + } + } + + fn is_poisoned(&self) -> bool { + self.poisoned.0.load(Ordering::Acquire) + } + + /// Marks the flusher poisoned and completes the failing job's own token with its error. + #[cold] + fn poison_and_report(&self, token: &TokenState, error: StoreError) { + self.poisoned.0.store(true, Ordering::Release); + token.complete(Err(FlushError::from(error))); + } +} + /// Shared cell one durability token waits on. #[derive(Default)] struct TokenState { @@ -159,38 +190,30 @@ struct Job { snapshot: Vec, cursor: Option, token: Arc, - admission: Arc, + shared: Arc, } impl Drop for Job { fn drop(&mut self) { self.token.complete(Err(FlushError::Closed)); - self.admission.release(); + self.shared.admission.0.release(); } } -/// Marks the flusher poisoned and completes the failing job's own token with its error. -#[cold] -fn poison_and_report(poisoned: &AtomicBool, token: &TokenState, error: StoreError) { - poisoned.store(true, Ordering::Release); - token.complete(Err(FlushError::from(error))); -} - /// Drains jobs until the channel closes, committing each unless already poisoned. fn run_flusher( mut store: SnapshotStore, receiver: mpsc::Receiver, - durable_cursor: Arc>>, - poisoned: Arc, + shared: Arc, ) -> SnapshotStore { while let Ok(job) = receiver.recv() { - if poisoned.load(Ordering::Acquire) { + if shared.is_poisoned() { continue; } match store.commit(&job.snapshot, job.cursor) { Ok(()) => { let committed = store.durable_cursor(); - let mut watermark = lock_recovering(&durable_cursor); + let mut watermark = lock_recovering(&shared.durable_cursor.0); // A commit supersedes whatever the store held, so the watermark // tracks it down as well as up and always names what a reopen // recovers. A resync commits older state and lowers it. @@ -198,7 +221,7 @@ fn run_flusher( drop(watermark); job.token.complete(Ok(())); } - Err(error) => poison_and_report(&poisoned, &job.token, error), + Err(error) => shared.poison_and_report(&job.token, error), } } store @@ -207,9 +230,7 @@ fn run_flusher( /// Background fsync thread; the commit point is the fsync return, not the write. pub struct Flusher { sender: mpsc::SyncSender, - admission: Arc, - durable_cursor: Arc>>, - poisoned: Arc, + shared: Arc, handle: JoinHandle>, } @@ -217,19 +238,12 @@ impl Flusher { /// Spawns the background fsync thread over an opened store. pub fn spawn(store: SnapshotStore, queue_depth: usize) -> Self { let (sender, receiver) = mpsc::sync_channel::(queue_depth); - let durable_cursor = Arc::new(Mutex::new(store.durable_cursor())); - let poisoned = Arc::new(AtomicBool::new(false)); - let admission = Arc::new(Admission::new(queue_depth)); - let thread_cursor = Arc::clone(&durable_cursor); - let thread_poisoned = Arc::clone(&poisoned); - let handle = thread::spawn(move || { - run_flusher(store, receiver, thread_cursor, thread_poisoned) - }); + let shared = Arc::new(Shared::new(queue_depth, store.durable_cursor())); + let thread_shared = Arc::clone(&shared); + let handle = thread::spawn(move || run_flusher(store, receiver, thread_shared)); Self { sender, - admission, - durable_cursor, - poisoned, + shared, handle, } } @@ -240,16 +254,16 @@ impl Flusher { snapshot: Vec, cursor: Option, ) -> Result { - if self.poisoned.load(Ordering::Acquire) { + if self.shared.is_poisoned() { return Err(FlushError::Closed); } - self.admission.acquire(); + self.shared.admission.0.acquire(); let token = Arc::new(TokenState::default()); let job = Job { snapshot, cursor, token: Arc::clone(&token), - admission: Arc::clone(&self.admission), + shared: Arc::clone(&self.shared), }; if self.sender.send(job).is_err() { return Err(FlushError::Closed); @@ -260,7 +274,7 @@ impl Flusher { /// Cursor a reopen of the store would recover; falls when a resync commits /// older state. pub fn durable_cursor(&self) -> Option { - *lock_recovering(&self.durable_cursor) + *lock_recovering(&self.shared.durable_cursor.0) } /// Stops the thread and returns the store; pending jobs complete first. @@ -312,11 +326,7 @@ mod tests { vfs::Vfs, }; use crate::{ - batch::{ - Batch, - BlockSpan, - LogEvent, - }, + batch::Batch, engine::{ Engine, EngineConfig, @@ -357,18 +367,9 @@ mod tests { /// Advances the engine by one block carrying a single event. fn advance(engine: &mut Engine, number: u64) { let boundary = number.checked_sub(1).filter(|&n| n > 0).map(block_ref); - let batch = Batch { - boundary, - spans: vec![BlockSpan { - block: block_ref(number), - start: 0, - end: 1, - }], - events: vec![LogEvent { - log_index: 0, - event: number, - }], - }; + let mut batch = Batch::new(); + batch.boundary = boundary; + batch.push_block(block_ref(number), [(0u32, number)]); engine.apply_batch(&batch).unwrap(); } diff --git a/crates/chainfold/src/storage/manifest.rs b/crates/chainfold/src/storage/manifest.rs index 8f0dc7e..f4b8b76 100644 --- a/crates/chainfold/src/storage/manifest.rs +++ b/crates/chainfold/src/storage/manifest.rs @@ -40,27 +40,37 @@ pub(crate) struct SlotRecord { pub(crate) snapshot_id: u64, } -/// Splits a cursor into its slot flag and fields. -fn cursor_flag_and_fields(cursor: Option) -> (u8, u64, u64) { - match cursor { - Some(pos) => (1, pos.block, pos.log_index), - None => (0, 0, 0), - } +/// Writes a little-endian u64 field at a fixed offset within a slot. +#[inline] +fn put_u64(bytes: &mut [u8; SLOT_SIZE], offset: usize, value: u64) { + bytes[offset..offset + U64_LEN].copy_from_slice(&value.to_le_bytes()); +} + +/// Reads a little-endian u64 field at a fixed offset within a full-size slot. +#[inline] +fn get_u64(bytes: &[u8], offset: usize) -> u64 { + let field: [u8; U64_LEN] = bytes[offset..offset + U64_LEN] + .try_into() + .expect("slot is fixed size and every u64 field fits inside it"); + u64::from_le_bytes(field) } /// Encodes a slot into its fixed layout with a CRC32C trailer over bytes `0..60`. pub(crate) fn encode_slot(record: &SlotRecord) -> [u8; SLOT_SIZE] { let mut bytes = [0u8; SLOT_SIZE]; - bytes[VERSION_OFFSET..VERSION_OFFSET + U64_LEN] - .copy_from_slice(&record.version.to_le_bytes()); - let (flag, block, log_index) = cursor_flag_and_fields(record.cursor); - bytes[CURSOR_FLAG_OFFSET] = flag; - bytes[CURSOR_BLOCK_OFFSET..CURSOR_BLOCK_OFFSET + U64_LEN] - .copy_from_slice(&block.to_le_bytes()); - bytes[CURSOR_LOG_INDEX_OFFSET..CURSOR_LOG_INDEX_OFFSET + U64_LEN] - .copy_from_slice(&log_index.to_le_bytes()); - bytes[SNAPSHOT_ID_OFFSET..SNAPSHOT_ID_OFFSET + U64_LEN] - .copy_from_slice(&record.snapshot_id.to_le_bytes()); + put_u64(&mut bytes, VERSION_OFFSET, record.version); + bytes[CURSOR_FLAG_OFFSET] = u8::from(record.cursor.is_some()); + put_u64( + &mut bytes, + CURSOR_BLOCK_OFFSET, + record.cursor.map_or(0, |pos| pos.block), + ); + put_u64( + &mut bytes, + CURSOR_LOG_INDEX_OFFSET, + record.cursor.map_or(0, |pos| pos.log_index), + ); + put_u64(&mut bytes, SNAPSHOT_ID_OFFSET, record.snapshot_id); let crc = crc32c(&bytes[..CRC_COVERED_LEN]); bytes[CRC_OFFSET..CRC_OFFSET + CRC_LEN].copy_from_slice(&crc.to_le_bytes()); bytes @@ -71,44 +81,24 @@ pub(crate) fn decode_slot(bytes: &[u8]) -> Option { if bytes.len() != SLOT_SIZE { return None; } - let expected_crc = u32::from_le_bytes( - bytes[CRC_OFFSET..CRC_OFFSET + CRC_LEN] - .try_into() - .expect("slice length matches the CRC field width"), - ); - let computed_crc = crc32c(&bytes[..CRC_COVERED_LEN]); - if computed_crc != expected_crc { + let stored: [u8; CRC_LEN] = bytes[CRC_OFFSET..CRC_OFFSET + CRC_LEN] + .try_into() + .expect("slice length matches the CRC field width"); + if crc32c(&bytes[..CRC_COVERED_LEN]) != u32::from_le_bytes(stored) { return None; } - let version = u64::from_le_bytes( - bytes[VERSION_OFFSET..VERSION_OFFSET + U64_LEN] - .try_into() - .expect("slice length matches a u64 field width"), - ); - let block = u64::from_le_bytes( - bytes[CURSOR_BLOCK_OFFSET..CURSOR_BLOCK_OFFSET + U64_LEN] - .try_into() - .expect("slice length matches a u64 field width"), - ); - let log_index = u64::from_le_bytes( - bytes[CURSOR_LOG_INDEX_OFFSET..CURSOR_LOG_INDEX_OFFSET + U64_LEN] - .try_into() - .expect("slice length matches a u64 field width"), - ); let cursor = match bytes[CURSOR_FLAG_OFFSET] { 0 => None, - 1 => Some(Position::new(block, log_index)), + 1 => Some(Position::new( + get_u64(bytes, CURSOR_BLOCK_OFFSET), + get_u64(bytes, CURSOR_LOG_INDEX_OFFSET), + )), _ => return None, }; - let snapshot_id = u64::from_le_bytes( - bytes[SNAPSHOT_ID_OFFSET..SNAPSHOT_ID_OFFSET + U64_LEN] - .try_into() - .expect("slice length matches a u64 field width"), - ); Some(SlotRecord { - version, + version: get_u64(bytes, VERSION_OFFSET), cursor, - snapshot_id, + snapshot_id: get_u64(bytes, SNAPSHOT_ID_OFFSET), }) } diff --git a/crates/chainfold/src/test_util.rs b/crates/chainfold/src/test_util.rs index 029544b..6676874 100644 --- a/crates/chainfold/src/test_util.rs +++ b/crates/chainfold/src/test_util.rs @@ -8,11 +8,7 @@ use std::vec::Vec; use core::fmt; use crate::{ - batch::{ - Batch, - BlockSpan, - LogEvent, - }, + batch::Batch, engine::Engine, error::{ DurabilityLost, @@ -101,38 +97,19 @@ impl crate::snapshot::Persist for RecordingFold { } fn decode_state(bytes: &[u8]) -> Result { - const COUNT_LEN: usize = 8; const ENTRY_LEN: usize = 24; - if bytes.len() < COUNT_LEN { - return Err(()); - } - let count_bytes: [u8; COUNT_LEN] = - bytes[..COUNT_LEN].try_into().map_err(|_| ())?; - let count = usize::try_from(u64::from_le_bytes(count_bytes)).map_err(|_| ())?; - let entries_len = count.checked_mul(ENTRY_LEN).ok_or(())?; - let total_len = COUNT_LEN.checked_add(entries_len).ok_or(())?; - if bytes.len() != total_len { + let (count, entries) = bytes.split_first_chunk::<8>().ok_or(())?; + let count = usize::try_from(u64::from_le_bytes(*count)).map_err(|_| ())?; + if entries.len() != count.checked_mul(ENTRY_LEN).ok_or(())? { return Err(()); } - let mut applied = Vec::with_capacity(count); - let mut offset = COUNT_LEN; - for _ in 0..count { - let block_end = offset.checked_add(8).ok_or(())?; - let log_index_end = block_end.checked_add(8).ok_or(())?; - let event_end = log_index_end.checked_add(8).ok_or(())?; - let block = - u64::from_le_bytes(bytes[offset..block_end].try_into().map_err(|_| ())?); - let log_index = u64::from_le_bytes( - bytes[block_end..log_index_end].try_into().map_err(|_| ())?, - ); - let event = u64::from_le_bytes( - bytes[log_index_end..event_end].try_into().map_err(|_| ())?, - ); - applied.push((Position::new(block, log_index), event)); - offset = event_end; - } + // The length check above makes every chunk exactly three 8-byte lanes. + let lane = |e: &[u8], i: usize| { + u64::from_le_bytes(e[i * 8..][..8].try_into().expect("three lanes")) + }; + let entry = |e: &[u8]| (Position::new(lane(e, 0), lane(e, 1)), lane(e, 2)); Ok(Self { - applied, + applied: entries.chunks_exact(ENTRY_LEN).map(entry).collect(), fail_at: None, }) } @@ -353,23 +330,15 @@ impl EventSource for ScriptedChain { if served >= self.batch_blocks { break; } - let start = u32::try_from(out.events.len()).expect("event count fits in u32"); - for (log_index, event) in block.events.iter().enumerate() { - let log_index = u64::try_from(log_index).expect("log index fits in u64"); - out.events.push(LogEvent { - log_index, - event: *event, - }); - } - let end = u32::try_from(out.events.len()).expect("event count fits in u32"); - out.spans.push(BlockSpan { - block: BlockRef { + out.push_block( + BlockRef { number: block.number, hash: block.hash, }, - start, - end, - }); + block.events.iter().enumerate().map(|(index, event)| { + (u32::try_from(index).expect("log index fits in u32"), *event) + }), + ); served += 1; } Ok(()) @@ -479,6 +448,28 @@ impl CrashVfs { } } + /// Consumes budget and reports how much of a `len`-byte effect this call keeps. + fn torn_write(&mut self, len: usize) -> (bool, usize) { + let crashing = self.consume_budget(); + ( + crashing, + if crashing { + len.min(self.torn_len) + } else { + len + }, + ) + } + + /// Maps a crash flag onto the io result every mutating op returns. + fn outcome(crashing: bool) -> io::Result<()> { + if crashing { + Err(crash_budget_error()) + } else { + Ok(()) + } + } + /// Resolves the inode a write targets, allocating a fresh one for a new name. fn inode_for_write(&mut self, path: &Path) -> u64 { if let Some(&inode) = self.volatile_names.get(path) { @@ -510,47 +501,29 @@ impl Vfs for CrashVfs { fn write(&mut self, path: &Path, bytes: &[u8]) -> io::Result<()> { let inode = self.inode_for_write(path); - let crashing = self.consume_budget(); - let keep = if crashing { - bytes.len().min(self.torn_len) - } else { - bytes.len() - }; + let (crashing, keep) = self.torn_write(bytes.len()); self.files .get_mut(&inode) .expect("write always registers its inode first") .volatile = bytes[..keep].to_vec(); - if crashing { - Err(crash_budget_error()) - } else { - Ok(()) - } + Self::outcome(crashing) } fn write_at(&mut self, path: &Path, offset: u64, bytes: &[u8]) -> io::Result<()> { let inode = self.inode_for_write(path); - let crashing = self.consume_budget(); - let apply_len = if crashing { - bytes.len().min(self.torn_len) - } else { - bytes.len() - }; + let (crashing, keep) = self.torn_write(bytes.len()); let offset = usize::try_from(offset).expect("offset fits in memory on this platform"); let file = self .files .get_mut(&inode) .expect("write_at always registers its inode first"); - let needed = offset + apply_len; + let needed = offset + keep; if file.volatile.len() < needed { file.volatile.resize(needed, 0); } - file.volatile[offset..offset + apply_len].copy_from_slice(&bytes[..apply_len]); - if crashing { - Err(crash_budget_error()) - } else { - Ok(()) - } + file.volatile[offset..needed].copy_from_slice(&bytes[..keep]); + Self::outcome(crashing) } fn fsync_file(&mut self, path: &Path) -> io::Result<()> { @@ -558,26 +531,17 @@ impl Vfs for CrashVfs { .volatile_names .get(path) .ok_or_else(crash_not_found_error)?; - let crashing = self.consume_budget(); + let target_len = self.files[&inode].volatile.len(); + let (crashing, keep) = self.torn_write(target_len); let file = self .files .get_mut(&inode) .expect("fsync_file always resolves a registered inode"); - let target_len = file.volatile.len(); if file.durable.len() < target_len { file.durable.resize(target_len, 0); } - let keep = if crashing { - self.torn_len.min(target_len) - } else { - target_len - }; file.durable[..keep].copy_from_slice(&file.volatile[..keep]); - if crashing { - Err(crash_budget_error()) - } else { - Ok(()) - } + Self::outcome(crashing) } fn rename(&mut self, from: &Path, to: &Path) -> io::Result<()> { @@ -586,13 +550,9 @@ impl Vfs for CrashVfs { .remove(from) .ok_or_else(crash_not_found_error)?; let crashing = self.consume_budget(); - if crashing { - self.volatile_names.insert(from.to_path_buf(), inode); - Err(crash_budget_error()) - } else { - self.volatile_names.insert(to.to_path_buf(), inode); - Ok(()) - } + let name = if crashing { from } else { to }; + self.volatile_names.insert(name.to_path_buf(), inode); + Self::outcome(crashing) } fn remove(&mut self, path: &Path) -> io::Result<()> { @@ -603,10 +563,8 @@ impl Vfs for CrashVfs { let crashing = self.consume_budget(); if crashing { self.volatile_names.insert(path.to_path_buf(), inode); - Err(crash_budget_error()) - } else { - Ok(()) } + Self::outcome(crashing) } fn list(&mut self, dir: &Path) -> io::Result> { @@ -744,9 +702,10 @@ mod tests { // when polled chain.next_batch(Some(cursor), &mut batch).unwrap(); // then one span for block 9 with the complete event set - assert_eq!(batch.spans.len(), 1); - assert_eq!(batch.spans[0].block.number, 9); - let events: Vec = batch.events.iter().map(|entry| entry.event).collect(); + assert_eq!(batch.span_count(), 1); + let span = batch.spans().next().expect("one span"); + assert_eq!(span.number, 9); + let events: Vec = span.events.to_vec(); assert_eq!(events, vec![30, 31]); } @@ -795,7 +754,7 @@ mod tests { // when polled chain.next_batch(None, &mut batch).unwrap(); // then two spans - assert_eq!(batch.spans.len(), 2); + assert_eq!(batch.span_count(), 2); } #[test] @@ -809,7 +768,7 @@ mod tests { // when polled chain.next_batch(Some(tip), &mut batch).unwrap(); // then no spans - assert!(batch.spans.is_empty()); + assert!(batch.is_empty()); } #[test] diff --git a/crates/chainfold/tests/apply_alloc.rs b/crates/chainfold/tests/apply_alloc.rs index 1b66d49..2995b50 100644 --- a/crates/chainfold/tests/apply_alloc.rs +++ b/crates/chainfold/tests/apply_alloc.rs @@ -21,10 +21,8 @@ use std::{ use chainfold::{ Batch, BlockRef, - BlockSpan, Engine, EngineConfig, - LogEvent, test_util::NoopFold, }; @@ -59,21 +57,13 @@ fn block_ref(number: u64) -> BlockRef { /// Builds a single-span batch of `count` events over one block. fn block_batch(boundary: Option, number: u64, count: u32) -> Batch { - let events: Vec> = (0..count as u64) - .map(|log_index| LogEvent { - log_index, - event: log_index, - }) - .collect(); - Batch { - boundary, - spans: vec![BlockSpan { - block: block_ref(number), - start: 0, - end: count, - }], - events, - } + let mut batch = Batch::new(); + batch.boundary = boundary; + batch.push_block( + block_ref(number), + (0..count).map(|log_index| (log_index, u64::from(log_index))), + ); + batch } #[test] diff --git a/crates/chainfold/tests/restart_identity.rs b/crates/chainfold/tests/restart_identity.rs index e904413..2c19bea 100644 --- a/crates/chainfold/tests/restart_identity.rs +++ b/crates/chainfold/tests/restart_identity.rs @@ -4,10 +4,8 @@ use chainfold::{ Batch, BlockRef, - BlockSpan, Engine, EngineConfig, - LogEvent, test_util::RecordingFold, }; use proptest::prelude::*; @@ -36,22 +34,12 @@ fn apply_block(engine: &mut Engine, number: u64, event_count: usi return; } let boundary = engine.cursor().map(|cursor| block_ref(cursor.block)); - let events: Vec> = (0..event_count as u64) - .map(|log_index| LogEvent { - log_index, - event: log_index, - }) - .collect(); - let end = events.len() as u32; - let batch = Batch { - boundary, - spans: vec![BlockSpan { - block: block_ref(number), - start: 0, - end, - }], - events, - }; + let mut batch = Batch::new(); + batch.boundary = boundary; + batch.push_block( + block_ref(number), + (0..event_count as u32).map(|log_index| (log_index, u64::from(log_index))), + ); engine.apply_batch(&batch).unwrap(); } diff --git a/crates/chainfold/tests/storage_crash.rs b/crates/chainfold/tests/storage_crash.rs index a37b192..be87e73 100644 --- a/crates/chainfold/tests/storage_crash.rs +++ b/crates/chainfold/tests/storage_crash.rs @@ -7,12 +7,10 @@ use std::path::PathBuf; use chainfold::{ Batch, BlockRef, - BlockSpan, Driver, DriverConfig, Engine, EngineConfig, - LogEvent, Position, Tickable, storage::{ @@ -67,18 +65,9 @@ fn block_ref(number: u64) -> BlockRef { /// Advances the reference engine by one block carrying a single event. fn advance(engine: &mut Engine, number: u64) { let boundary = number.checked_sub(1).filter(|&n| n > 0).map(block_ref); - let batch = Batch { - boundary, - spans: vec![BlockSpan { - block: block_ref(number), - start: 0, - end: 1, - }], - events: vec![LogEvent { - log_index: 0, - event: number, - }], - }; + let mut batch = Batch::new(); + batch.boundary = boundary; + batch.push_block(block_ref(number), [(0u32, number)]); engine.apply_batch(&batch).unwrap(); }