diff --git a/crates/chainfold/Cargo.toml b/crates/chainfold/Cargo.toml index 047b090..1e9a41f 100644 --- a/crates/chainfold/Cargo.toml +++ b/crates/chainfold/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "chainfold" -version = "0.3.0" +version = "0.4.0" edition.workspace = true rust-version.workspace = true license.workspace = true diff --git a/crates/chainfold/README.md b/crates/chainfold/README.md index 7fdf073..ee31374 100644 --- a/crates/chainfold/README.md +++ b/crates/chainfold/README.md @@ -17,15 +17,17 @@ durable, a reorg noticed only when a proof stops verifying. - **apply**: a poll fills one `Batch`; the `Engine` enforces the total order, drops everything at or below the cursor as already applied, and counts the events the fold declares are not its own. -- **detect**: each batch carries the current header of the cursor block. A block hash - commits to its whole ancestry, so a reorg touching anything at or below the cursor - changes that one hash. One header per poll, one 40-byte compare. +- **detect**: each batch carries the current header of the cursor block, or reuses the one + last checked while the head has not moved. A block hash commits to its whole ancestry, + so a reorg touching anything at or below the cursor changes that one hash. One 40-byte + compare per poll. - **recover**: on a mismatch, the driver bisects the ring of observed blocks with `header_at` for the deepest still-canonical block, rolls back to the newest checkpoint at or below it, and refolds forward. `O(log W)` probes, then deterministic replay. - **persist**: the driver offers the oldest retained checkpoint to a sink; a flusher thread fsyncs it. The durable cursor trails the applied cursor and names what a - restart recovers, so the apply path never fsyncs and replay closes the gap. + restart recovers, so the apply path never fsyncs and replay closes the gap. After a + reset it is None until a later offer commits. ```mermaid flowchart LR @@ -59,14 +61,13 @@ and recovery), and the adapters (`std`: tokio harness, snapshot store, flusher). counted and stepped over), `Halt` (clean below this position, rollback or resync), `Poison` (partially mutated, untrusted until a restore). - `Source` is three reads: `head`, `header_at`, `events_in`. The driver owns the window - walk, the scan mark, the boundary refetch, and the grouping into spans, so a source - cannot get the batch shape wrong and every source gets fork detection. `Ok(None)` from - a probe means suspected fork. + walk, the boundary refetch, and the grouping into spans, so a source cannot get the + batch shape wrong and every source gets fork detection. `Ok(None)` from a probe means + suspected fork. - rollback is bounded to K retained checkpoints. Over-rollback past still-canonical but unobserved blocks is intended; the discarded events replay deterministically. - the recovery ladder is rollback, then resync from genesis, then a typed terminal state. - `DivergenceCause` names which rung ran out: fork deeper than the window, no checkpoint - below the ancestor, source horizon short of the start block. + `DivergenceCause` names which rung ran out: source horizon short of the start block. - `EngineStatus::Unrecoverable` is only left by a `reset`. Automated progress stops rather than folding on untrusted state. - the snapshot envelope is engine-owned: magic, format version, fold identity tag, @@ -182,11 +183,10 @@ dependency stack that a guest or wasm build has no use for. ### Tuning - `EngineConfig::ring_capacity` (W): observed-block window, a power of two in `[2, 1 << 20]`. - It bounds fork detection; a fork deeper than the oldest ring entry is - `ForkBeyondWindow`. + It bounds fork detection; a fork deeper than the oldest ring entry escalates to a resync. - `EngineConfig::checkpoint_slots` (K): retained rollback points. Zero disables rollback, - so `Halt` and `Poison` escalate straight to resync and anchor checks never fire. That is - the default shape of most consumers today, not an edge case. + so every fork escalates to a resync. That is the default shape of most consumers today, + not an edge case. - `DriverConfig::checkpoint_interval`: blocks of cursor progress between automatic checkpoints. The reorg an offered snapshot survives without a resync is `checkpoint_slots * checkpoint_interval` blocks; size both against the deepest reorg diff --git a/crates/chainfold/benches/recover.rs b/crates/chainfold/benches/recover.rs index e439957..980f417 100644 --- a/crates/chainfold/benches/recover.rs +++ b/crates/chainfold/benches/recover.rs @@ -56,7 +56,7 @@ impl Source for Probed { type Error = PollFailure; type Event = u64; - fn head(&mut self) -> Result { + fn head(&mut self) -> Result { self.inner.head() } diff --git a/crates/chainfold/src/checkpoint.rs b/crates/chainfold/src/checkpoint.rs index d45ba5f..ddadff4 100644 --- a/crates/chainfold/src/checkpoint.rs +++ b/crates/chainfold/src/checkpoint.rs @@ -79,11 +79,17 @@ impl CheckpointRing { .find(|slot| slot.cursor.is_none_or(|cursor| cursor.block <= block)) } - /// Drops slots with cursor block strictly above the argument. + /// Drops slots with a cursor after the argument, and moves the write position back + /// over them so the next stores refill the holes before any live slot is evicted. + /// + /// Cursors only rise between rollbacks, so the dropped slots are the newest stored. #[cold] - pub(crate) fn drop_above(&mut self, block: u64) { + pub(crate) fn drop_newer(&mut self, cursor: Option) { + let len = self.slots.len(); for slot in &mut self.slots { - slot.take_if(|slot| slot.cursor.is_some_and(|cursor| cursor.block > block)); + if slot.take_if(|slot| slot.cursor > cursor).is_some() { + self.next = (self.next + len - 1) % len; + } } } diff --git a/crates/chainfold/src/driver.rs b/crates/chainfold/src/driver.rs index 6bb0529..eb4ca6b 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -43,6 +43,8 @@ const DEFAULT_BACKOFF_BASE: Duration = Duration::from_millis(200); const DEFAULT_BACKOFF_MAX: Duration = Duration::from_secs(30); /// Default blocks between checkpoints. const DEFAULT_CHECKPOINT_INTERVAL: u64 = 64; +/// Most windows one tick reads; a longer walk goes on in the next tick. +const MAX_SPAN: u64 = 64; /// True when `block` has reached the next interval step past the last marked block. fn due(last: Option, block: u64, interval: u64) -> bool { @@ -52,7 +54,9 @@ fn due(last: Option, block: u64, interval: u64) -> bool { /// Outcome of one driver tick. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Tick { - /// Batch applied; the summary counts what the fold saw. + /// Batch applied; the summary counts what the fold saw. A zero summary means the walk + /// stopped at its per-tick cap with nothing to fold yet; the next tick goes on from + /// there. Progressed(ApplySummary), /// Poll returned nothing new. Idle, @@ -71,6 +75,20 @@ pub enum Tick { Terminal(EngineStatus), } +/// What a scan leaves for the tick to act on. +enum Scan { + /// `batch` is ready to apply; the walk ran under this head. + Ready(BlockRef), + /// The walk stopped at its cap with nothing to fold; `batch` holds only the boundary, + /// `anchor` is the block the walk ended on, and the walk ran under `head`. + Capped { anchor: BlockRef, head: BlockRef }, + /// The source's head trails the cursor, or the walked mark, and the newest ring block + /// at or below it is on its chain; nothing to do. + Lagging, + /// The source contradicted itself; its answer is dropped. + Refused, +} + /// Point-in-time snapshot of driver and engine state for external observers. #[derive(Debug, Clone, Copy, PartialEq)] pub struct DriverStatus { @@ -80,12 +98,13 @@ pub struct DriverStatus { pub last_verified: Option, /// Engine status behind this driver. pub engine: EngineStatus, - /// True once the most recent poll returned no new blocks. + /// True once the most recent poll reached the head and found no new blocks. pub caught_up: bool, /// Events the fold declared not its own. pub skips: u64, - /// Cursor the sink reports a restart would recover; None without a sink or a - /// flush. A resync lowers it, so it holds for the instant it was read. + /// Cursor the sink reports a restart would recover; None without a sink, before a + /// flush, and after a reset until an offer made since then commits. A resync clears + /// it, so it holds for the instant it was read. pub durable_cursor: Option, /// True once the sink refused an offer; folding continues unpersisted. pub durability_lost: bool, @@ -155,15 +174,29 @@ where config: DriverConfig, batch: Batch, scratch: Vec<(BlockRef, u32, F::Event)>, - scanned_to: Option, initial: F, consecutive_errors: u32, caught_up: bool, generation: u64, last_checkpoint_block: Option, last_snapshot_block: Option, + seen_resets: u64, durability_lost: bool, advanced: bool, + /// Windows the next tick may read, up to `MAX_SPAN`: halves after a source error, + /// doubles after any other tick. + span: u64, + /// The block the last walk ended on when it stopped at its cap: the chain through it + /// has nothing to fold above the cursor, up to that block. + walked: Option, + /// The head a tick read before and after a walk that folded or stopped at its cap, + /// once its boundary check passed. The chain it names holds the whole cursor block, + /// so a tick that finds the same head need not read that block again. + vouched: Option, + /// Cursor and reset count the last tick left the engine at; None when that tick left + /// it halted or poisoned, or never ended. A different pair now means the caller moved + /// the engine. + left_at: Option<(Option, u64)>, } impl Driver @@ -246,7 +279,7 @@ where fn around( engine: Engine, source: S, - sink: K, + mut sink: K, initial: F, driver_config: DriverConfig, ) -> Result { @@ -258,23 +291,38 @@ where horizon, }); } - Ok(Self { + // A sink ahead of the engine holds a cursor from a run this engine has not folded. + if sink.durable_cursor() > engine.cursor() { + sink.reset(); + } + let seen_resets = engine.resets(); + let mut driver = Self { engine, source, sink, config: driver_config, batch: Batch::new(), scratch: Vec::new(), - scanned_to: None, initial, consecutive_errors: 0, caught_up: false, generation: 0, last_checkpoint_block: None, last_snapshot_block: None, + seen_resets, durability_lost: false, advanced: false, - }) + span: MAX_SPAN, + walked: None, + vouched: None, + left_at: None, + }; + // A recovered engine holds no checkpoint. Its first poll would take one at the tip, + // so a depth-1 reorg there would resync; keep one at the recovery point. + if driver.engine.checkpoint_count() == 0 { + driver.auto_checkpoint(); + } + Ok(driver) } /// Borrows the durability sink. @@ -298,13 +346,25 @@ where } /// Mutable access to the underlying event source. + /// + /// A call drops the scan's progress, since the caller may change what it returns. pub fn source_mut(&mut self) -> &mut S { + self.walked = None; + self.vouched = None; &mut self.source } - /// True once the most recent poll returned no new blocks. + /// True once the most recent poll reached the head and found no new blocks. pub fn is_caught_up(&self) -> bool { self.caught_up + && self.engine.status().is_active() + && self.left_at == Some(self.engine_point()) + } + + /// The cursor and reset count, which change only when a tick or the caller moves the + /// engine. + fn engine_point(&self) -> (Option, u64) { + (self.engine.cursor(), self.engine.resets()) } /// Runs the interval-based checkpoint rule. @@ -354,13 +414,14 @@ where Tick::DurabilityLost } - /// Lowers the snapshot mark to the restore point so the next offer is not suppressed. - #[cold] - fn clamp_snapshot_mark(&mut self, to: Option) { - self.last_snapshot_block = self - .last_snapshot_block - .zip(to) - .map(|(last, point)| last.min(point.block)); + /// Lowers both cadence marks to the cursor block, and clears them without a cursor, so + /// a rollback or a reset suppresses neither the next checkpoint nor the next offer. + fn clamp_marks(&mut self) { + let block = self.engine.cursor().map(|cursor| cursor.block); + let clamp = + |mark: Option| mark.zip(block).map(|(mark, block)| mark.min(block)); + self.last_checkpoint_block = clamp(self.last_checkpoint_block); + self.last_snapshot_block = clamp(self.last_snapshot_block); } /// Rolls back to the newest checkpoint at or below the ancestor, else escalates. @@ -369,7 +430,6 @@ where match self.engine.rollback_at_or_below(ancestor) { Ok(to) => { self.caught_up = false; - self.clamp_snapshot_mark(to); Tick::RolledBack { to } } Err(RollbackError::NoCheckpointAtOrBelow { .. }) => self.resync_or_terminal(), @@ -402,21 +462,36 @@ where self.engine.reset(self.initial.clone()); self.caught_up = false; self.consecutive_errors = 0; - self.scanned_to = None; - self.last_checkpoint_block = None; - self.last_snapshot_block = None; Tick::Resynced } /// Runs one poll-apply step, then records whether the cursor moved forward. fn step(&mut self) -> Tick { + // The caller can roll back or reset through `engine_mut`, so both are caught here. + self.clamp_marks(); + if self.seen_resets != self.engine.resets() { + self.sink.reset(); + self.seen_resets = self.engine.resets(); + } + // An engine away from where the last tick left it was moved by the caller, so + // what was walked above its old cursor says nothing about the new one. + if self.left_at.take() != Some(self.engine_point()) { + self.walked = None; + self.vouched = None; + } let tick = self.poll_apply(); - // Only forward cursor movement earns an immediate re-poll; a batch the - // engine fully deduped leaves the loop on its poll interval. - self.advanced = matches!( - tick, - Tick::Progressed(summary) if summary.applied > 0 || summary.skipped > 0 - ); + // A tick that folded, or walked to its cap, earns an immediate re-poll. + self.advanced = matches!(tick, Tick::Progressed(_)); + self.span = if tick == Tick::SourceError { + (self.span / 2).max(1) + } else { + self.span.saturating_mul(2).min(MAX_SPAN) + }; + self.left_at = self + .engine + .status() + .is_active() + .then(|| self.engine_point()); tick } @@ -426,20 +501,44 @@ where if !self.engine.status().is_active() { return Tick::Terminal(self.engine.status()); } - if self.scan().is_err() { - self.consecutive_errors = self.consecutive_errors.saturating_add(1); - return Tick::SourceError; - } + let mut capped_at = None; + let head = match self.scan() { + Ok(Scan::Ready(head)) => head, + Ok(Scan::Capped { anchor, head }) => { + capped_at = Some(anchor); + head + } + Ok(Scan::Lagging) => { + self.consecutive_errors = 0; + self.caught_up = true; + return Tick::Idle; + } + Ok(Scan::Refused) | Err(_) => { + self.batch.clear(); + self.consecutive_errors = self.consecutive_errors.saturating_add(1); + return Tick::SourceError; + } + }; self.consecutive_errors = 0; + // The mark survives only a tick that folds nothing and passes its boundary check. + let walked = self.walked.take(); match self.engine.apply_batch(&self.batch) { Ok(summary) => { - self.caught_up = self.batch.is_empty(); + let idle = self.batch.is_empty() && capped_at.is_none(); + if self.batch.is_empty() { + self.walked = capped_at.or(walked); + } + // An idle walk read the head only once, so it vouches for no new head. + if !idle { + self.vouched = Some(head); + } + self.caught_up = idle; // A snapshot refusal overrides progress; a checkpoint is silent. self.auto_checkpoint(); if let Some(tick) = self.offer_snapshot() { return tick; } - if self.batch.is_empty() { + if idle { Tick::Idle } else { Tick::Progressed(summary) @@ -465,43 +564,143 @@ where } } - /// Fills `batch` with the blocks strictly after the cursor, plus the cursor block's - /// header as the source reports it now. - fn scan(&mut self) -> Result<(), S::Error> { + /// Fills `batch` with the events after the cursor, plus the cursor block's header, all + /// as the chain `head` names reports them. A walk stops at `span` windows; when it + /// has found nothing by then, the block it ended on is the mark the next walk starts + /// above. A head the last folding or capped tick vouched for is the same chain, so + /// the walk starts above the cursor block, and above the mark without checking it. + fn scan(&mut self) -> Result { self.batch.clear(); - let cursor = self.engine.cursor(); - self.batch.boundary = match cursor { - Some(cursor) => self.source.header_at(cursor.block)?, - None => None, - }; - let head = self.source.head()?; - let mut from = match cursor { - Some(cursor) => self - .scanned_to - .map_or(cursor.block + 1, |to| to.saturating_add(1)) - .min(cursor.block + 1), - None => self.config.start_block, - }; + let cursor = self.engine.cursor(); + // a head below the floor on another chain falls through to the walk, which fails + // the boundary check and bisects + if let Some(floor) = self + .walked + .map(|walked| walked.number) + .or(cursor.map(|cursor| cursor.block)) + && head.number < floor + { + let lagging = match self + .engine + .observed() + .take_while(|seen| seen.number <= head.number) + .last() + { + None => true, + Some(seen) if seen.number == head.number => seen == head, + // a source with no header at or below its own head contradicts itself; + // only another hash proves a fork + Some(seen) => self + .source + .header_at(seen.number)? + .is_none_or(|header| header == seen), + }; + if lagging { + return Ok(Scan::Lagging); + } + } let window = self.source.window().max(1); - while from <= head { - let to = head.min(from.saturating_add(window - 1)); + let mut from = cursor.map_or(self.config.start_block, |cursor| cursor.block); + // The same head names the same chain, which still holds the cursor block that + // tick checked, so a walk can start above it. + let vouched = self.vouched == Some(head); + // A mark on the head's chain holds nothing to fold below it, and that chain holds + // the cursor block the mark was set over. Any other mark is stale. + let mut resumed = false; + if let Some(walked) = self.walked { + if vouched + || head == walked + || self.source.header_at(walked.number)? == Some(walked) + { + from = walked.number.saturating_add(1); + resumed = true; + } else { + self.walked = None; + } + } else if vouched && let Some(cursor) = cursor { + from = cursor.block.saturating_add(1); + resumed = true; + } + let end = head + .number + .min(from.saturating_add(self.span.saturating_mul(window) - 1)); + while from <= end { + let to = end.min(from.saturating_add(window - 1)); self.scratch.clear(); self.source.events_in(from, to, &mut self.scratch)?; - self.scanned_to = Some(to); - from = to.saturating_add(1); + if self + .scratch + .iter() + .any(|(block, ..)| !(from..=to).contains(&block.number)) + { + return Ok(Scan::Refused); + } + if let Some(cursor) = cursor + && from == cursor.block + { + let mut at_cursor = self + .scratch + .iter() + .map(|(block, ..)| block) + .filter(|block| block.number == cursor.block); + let boundary = at_cursor.next().copied(); + if at_cursor.any(|block| Some(*block) != boundary) { + return Ok(Scan::Refused); + } + self.batch.boundary = boundary; + // the rest of a partly folded cursor block is still owed + self.scratch.retain(|(block, log_index, _)| { + Position::new(block.number, u64::from(*log_index)) > cursor + }); + } if !self.scratch.is_empty() { group_into(&mut self.batch, &mut self.scratch); break; } + let Some(next) = to.checked_add(1) else { + break; + }; + from = next; + } + if let Some(cursor) = cursor + && self.batch.boundary.is_none() + { + self.batch.boundary = if resumed { + self.engine.observed().last() + } else if head.number == cursor.block { + Some(head) + } else { + self.source.header_at(cursor.block)? + }; + } + if end < head.number && self.batch.is_empty() { + // The head reads before and after the anchor read bracket it, so the anchor + // lies on the chain `head` names. + let anchor = self.source.header_at(end)?; + let unmoved = self.source.header_at(head.number)? == Some(head); + return Ok(match anchor { + Some(anchor) if unmoved && anchor.number == end => { + Scan::Capped { anchor, head } + } + _ => Scan::Refused, + }); + } + // What was read since `head` came from its chain only while the source still has + // `head`. An answer that ends on `head` is no exception: it may still carry blocks + // of another fork below it. + if !self.batch.is_empty() && self.source.header_at(head.number)? != Some(head) { + return Ok(Scan::Refused); } - Ok(()) + Ok(Scan::Ready(head)) } /// Bisects the observed ring for the deepest still-canonical block, then rolls back. #[cold] fn recover_via_bisection(&mut self) -> Tick { + // The engine is about to leave the cursor block the head vouched for. + self.vouched = None; let observed: Vec = self.engine.observed().collect(); let mut lo = 0usize; let mut hi = observed.len(); @@ -538,9 +737,13 @@ where cursor: self.engine.cursor(), last_verified: self.engine.last_verified(), engine: self.engine.status(), - caught_up: self.caught_up, + caught_up: self.is_caught_up(), skips: self.engine.skip_count(), - durable_cursor: self.sink.durable_cursor(), + durable_cursor: if self.seen_resets == self.engine.resets() { + self.sink.durable_cursor() + } else { + None + }, durability_lost: self.durability_lost, generation: self.generation, } @@ -574,7 +777,7 @@ fn group_into(batch: &mut Batch, entries: &mut Vec<(BlockRef, u32, E)>) { let mut current: Option = None; for (block, log_index, event) in entries.drain(..) { match current { - Some(open) if open.number != block.number => { + Some(open) if open != block => { batch.push_block(open, span.drain(..)); current = Some(block); } @@ -596,6 +799,10 @@ mod tests { vec, vec::Vec, }; + use core::cell::{ + Cell, + RefCell, + }; #[cfg(feature = "std")] use std::{ vec, @@ -635,7 +842,7 @@ mod tests { type Event = u64; type Error = PollFailure; - fn head(&mut self) -> Result { + fn head(&mut self) -> Result { self.inner.head() } @@ -667,7 +874,7 @@ mod tests { } /// Source that re-serves the same one-event block whatever range is asked for, - /// so the engine dedupes every poll after the first. + /// so every poll after the first finds it already applied. struct Stuck { block: BlockRef, } @@ -676,9 +883,8 @@ mod tests { type Event = u64; type Error = PollFailure; - fn head(&mut self) -> Result { - // above the block, so the scan keeps querying and keeps being re-served it - Ok(self.block.number + 1) + fn head(&mut self) -> Result { + Ok(self.block) } fn header_at(&mut self, _number: u64) -> Result, PollFailure> { @@ -696,6 +902,207 @@ mod tests { } } + /// Scripted chain that misbehaves on cue: it reorgs just before a chosen read, and it + /// appends `extra` to the next range answer. + struct Hostile { + inner: ScriptedChain, + reads: u32, + reorg: Option<(u32, usize, Vec>)>, + extra: Vec<(BlockRef, u32, u64)>, + } + + impl Hostile { + fn new(inner: ScriptedChain) -> Self { + Self { + inner, + reads: 0, + reorg: None, + extra: Vec::new(), + } + } + + /// Reorgs the chain just before the `nth` read from now, whichever method it is. + fn reorg_before_read( + &mut self, + nth: u32, + depth: usize, + new_blocks: Vec>, + ) { + self.reorg = Some((self.reads + nth, depth, new_blocks)); + } + + fn read(&mut self) { + self.reads += 1; + assert!(self.reads < 1_000, "the scan never stopped reading"); + if self + .reorg + .as_ref() + .is_some_and(|(at, ..)| *at == self.reads) + { + let (_, depth, new_blocks) = self.reorg.take().expect("a reorg is due"); + let new_blocks: Vec<&[u64]> = + new_blocks.iter().map(Vec::as_slice).collect(); + self.inner.reorg(depth, &new_blocks); + } + } + } + + impl Source for Hostile { + type Event = u64; + type Error = PollFailure; + + fn head(&mut self) -> Result { + self.read(); + self.inner.head() + } + + fn header_at(&mut self, number: u64) -> Result, PollFailure> { + self.read(); + self.inner.header_at(number) + } + + fn events_in( + &mut self, + from: u64, + to: u64, + out: &mut Vec<(BlockRef, u32, u64)>, + ) -> Result<(), PollFailure> { + self.read(); + self.inner.events_in(from, to, out)?; + out.append(&mut self.extra); + Ok(()) + } + + fn horizon(&self) -> ReplayHorizon { + self.inner.horizon() + } + + fn window(&self) -> u64 { + self.inner.window() + } + } + + /// Scripted chain read through shared handles, so a test can move the chain between + /// ticks without `source_mut`, which drops the walk's mark. `gate` is asked once per + /// call and fails it by returning true; `hide` drops the events of one block; `lie_at` + /// answers the header request for that block with the next block's header. + struct Remote<'a, G> { + chain: &'a RefCell, + calls: &'a Cell, + gate: G, + hide: Option, + lie_at: Option, + } + + impl bool> Remote<'_, G> { + fn enter(&mut self) -> Result<(), PollFailure> { + self.calls.set(self.calls.get() + 1); + if (self.gate)() { + Err(PollFailure) + } else { + Ok(()) + } + } + } + + impl bool> Source for Remote<'_, G> { + type Event = u64; + type Error = PollFailure; + + fn head(&mut self) -> Result { + self.enter()?; + self.chain.borrow_mut().head() + } + + fn header_at(&mut self, number: u64) -> Result, PollFailure> { + self.enter()?; + let asked = if self.lie_at == Some(number) { + number + 1 + } else { + number + }; + self.chain.borrow_mut().header_at(asked) + } + + fn events_in( + &mut self, + from: u64, + to: u64, + out: &mut Vec<(BlockRef, u32, u64)>, + ) -> Result<(), PollFailure> { + self.enter()?; + self.chain.borrow_mut().events_in(from, to, out)?; + if let Some(hidden) = self.hide { + out.retain(|(block, ..)| block.number != hidden); + } + Ok(()) + } + + fn window(&self) -> u64 { + self.chain.borrow().window() + } + } + + fn remote_driver<'a, G: FnMut() -> bool>( + chain: &'a RefCell, + calls: &'a Cell, + gate: G, + engine: EngineConfig, + config: DriverConfig, + ) -> Driver> { + let source = Remote { + chain, + calls, + gate, + hide: None, + lie_at: None, + }; + Driver::new(RecordingFold::default(), source, engine, config).unwrap() + } + + /// Runs one tick and returns it with the source calls it made. + fn tick_counted bool>( + driver: &mut Driver>, + calls: &Cell, + ) -> (Tick, u32) { + let before = calls.get(); + let tick = driver.tick(); + (tick, calls.get() - before) + } + + /// Chain of `blocks` blocks read `window` at a time; the block numbered `n` carries + /// the event `n` when `n` is listed and none otherwise. + fn sparse_chain(blocks: u64, window: u64, events: &[u64]) -> ScriptedChain { + let mut chain = ScriptedChain::new(1); + for number in 1..=blocks { + let carried: &[u64] = if events.contains(&number) { + &[number] + } else { + &[] + }; + chain.push_block(carried); + } + chain.set_window(window); + chain + } + + /// Replaces the blocks `first..` of `chain` with blocks up to `last`, carrying the + /// listed events as `sparse_chain` does. + fn replace_from(chain: &mut ScriptedChain, first: u64, last: u64, events: &[u64]) { + let depth = chain.tip().map_or(0, |tip| tip.number) + 1 - first; + let blocks: Vec> = (first..=last) + .map(|number| { + if events.contains(&number) { + vec![number] + } else { + vec![] + } + }) + .collect(); + let blocks: Vec<&[u64]> = blocks.iter().map(Vec::as_slice).collect(); + chain.reorg(usize::try_from(depth).unwrap(), &blocks); + } + fn engine_config(checkpoint_slots: usize) -> EngineConfig { EngineConfig { ring_capacity: 8, @@ -703,6 +1110,26 @@ mod tests { } } + fn hostile_driver( + chain: ScriptedChain, + checkpoint_slots: usize, + config: DriverConfig, + ) -> Driver { + Driver::new( + RecordingFold::default(), + Hostile::new(chain), + engine_config(checkpoint_slots), + config, + ) + .unwrap() + } + + /// Every event the fold holds, in fold order. + fn folded>(driver: &Driver) -> Vec { + let applied = &driver.engine().fold().applied; + applied.iter().map(|(_, event)| *event).collect() + } + fn new_driver( chain: ScriptedChain, engine: EngineConfig, @@ -724,6 +1151,21 @@ mod tests { outcome } + /// Ticks until the first `Idle`; a thousand ticks without one fail the test. + fn settle(driver: &mut Driver) + where + F: Fold + Clone, + S: Source, + K: SnapshotSink, + { + for _ in 0..1_000 { + if driver.tick() == Tick::Idle { + return; + } + } + panic!("the driver never went idle"); + } + fn collect_to_idle(driver: &mut Driver) -> Vec where F: Fold + Clone, @@ -918,6 +1360,38 @@ mod tests { assert!(after < before); } + #[test] + fn a_reset_clears_the_durable_cursor_until_a_later_offer_commits() { + // given two drivers that offered through block 8, so each sink holds block 5 + fn offered() -> SinkDriver { + let mut driver = sink_driver(8, 4, cadence_config(1, 1)); + run_to_idle(&mut driver); + assert_eq!(driver.status().durable_cursor, Some(Position::new(5, 0))); + driver + } + let mut resynced = offered(); + let mut reset = offered(); + // when one resyncs after a reorg below its ring and the other is reset by hand + resynced + .source_mut() + .reorg(8, &[&[10], &[20], &[30], &[40]]); + let outcome = resynced.tick(); + reset.engine_mut().reset(RecordingFold::default()); + // then neither reports the durable cursor of the chain it left + assert_eq!(outcome, Tick::Resynced); + assert_eq!(resynced.status().durable_cursor, None); + assert_eq!(reset.status().durable_cursor, None); + for driver in [&mut resynced, &mut reset] { + // and a poll that offers nothing leaves it None + driver.source_mut().fail_next_polls(1); + assert_eq!(driver.tick(), Tick::SourceError); + assert_eq!(driver.status().durable_cursor, None); + // until the first offer made after the reset commits + driver.tick(); + assert_eq!(driver.status().durable_cursor, Some(Position::new(1, 0))); + } + } + #[test] fn driver_folds_to_tip_and_reports_caught_up() { // given a ten-block chain with one event per block @@ -1031,7 +1505,7 @@ mod tests { } #[test] - fn rollback_replays_from_the_restored_cursor_not_the_scan_mark() { + fn rollback_replays_from_the_restored_cursor() { // given a driver caught up on ten blocks, checkpointing every block let mut chain = ScriptedChain::new(1); for value in 1..=10u64 { @@ -1050,8 +1524,7 @@ mod tests { // when the last three blocks are replaced driver.source_mut().reorg(3, &[&[80], &[90], &[100]]); run_to_idle(&mut driver); - // then the scan restarted at the restored cursor, so the replacements folded; - // resuming at the high-water mark instead would have skipped them entirely + // then the scan restarted at the restored cursor, so the replacements folded let applied: Vec = driver .engine() .fold() @@ -1063,7 +1536,7 @@ mod tests { } #[test] - fn a_fully_deduped_batch_keeps_the_poll_interval() { + fn an_applied_block_served_again_keeps_the_poll_interval() { // given a source that re-serves the same one-event block on every poll let block = BlockRef { number: 1, @@ -1076,7 +1549,7 @@ mod tests { DriverConfig::default(), ) .unwrap(); - // when the first tick applies the block and the second dedupes it + // when the first tick applies the block and the second drops it as applied let applying = driver.tick(); let after_apply = driver.next_delay(); let deduping = driver.tick(); @@ -1091,14 +1564,7 @@ mod tests { }) ); assert_eq!(after_apply, Duration::ZERO); - assert_eq!( - deduping, - Tick::Progressed(ApplySummary { - applied: 0, - deduped: 1, - skipped: 0, - }) - ); + assert_eq!(deduping, Tick::Idle); assert_eq!(after_dedup, Duration::from_secs(1)); } @@ -1218,6 +1684,61 @@ mod tests { assert_eq!(driver.engine().durable_point(), Some(Position::new(5, 0))); } + /// Twenty one-event blocks folded with a checkpoint every 4, at blocks 1, 5, 9, 13, 17. + fn cadence_driver() -> Driver { + let config = DriverConfig { + checkpoint_interval: Some(4), + snapshot_interval: None, + ..DriverConfig::from_block(1) + }; + let engine = EngineConfig { + ring_capacity: 64, + checkpoint_slots: 16, + }; + let mut driver = new_driver(one_event_chain(20), engine, config); + run_to_idle(&mut driver); + assert_eq!(driver.engine().checkpoint_count(), 5); + driver + } + + #[test] + fn a_rollback_restarts_the_checkpoint_cadence_at_the_restore_point() { + // given twenty blocks folded with a checkpoint every 4, the last at block 17 + let mut driver = cadence_driver(); + // when blocks 15 to 20 are replaced and the refold runs on to block 24 + driver.source_mut().reorg( + 6, + &[ + &[91], + &[92], + &[93], + &[94], + &[95], + &[96], + &[97], + &[98], + &[99], + &[100], + ], + ); + run_to_idle(&mut driver); + // then the cadence restarted at the restore point, block 13, so the replaced 17 + // and the new 21 each got a checkpoint + assert_eq!(driver.engine().checkpoint_count(), 6); + } + + #[test] + fn a_manual_rollback_restarts_the_cadence_at_the_restore_point() { + // given twenty blocks folded with a checkpoint every 4, the last at block 17 + let mut driver = cadence_driver(); + // when the caller rolls back through engine_mut and the driver refolds + let restored = driver.engine_mut().rollback_at_or_below(12).unwrap(); + run_to_idle(&mut driver); + // then blocks 13 and 17 got their checkpoints again + assert_eq!(restored, Some(Position::new(9, 0))); + assert_eq!(driver.engine().checkpoint_count(), 5); + } + #[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 @@ -1256,42 +1777,176 @@ mod tests { } #[test] - fn generation_increments_every_tick() { - // given a driver over an empty chain - let chain = ScriptedChain::new(1); - let mut driver = new_driver(chain, engine_config(0), DriverConfig::default()); - let start = driver.status().generation; - // when three ticks of any outcome run - driver.tick(); - driver.tick(); - driver.tick(); - // then status generation rose by three - assert_eq!(driver.status().generation, start + 3); + fn a_terminal_tick_clears_caught_up() { + // given a caught-up driver whose fold halts on the next block's event + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + let halt_pos = Position::new(2, 0); + let fold = RecordingFold { + applied: Vec::new(), + fail_at: Some((halt_pos, FailKind::Halt)), + }; + let mut driver = + Driver::new(fold, chain, engine_config(0), DriverConfig::default()).unwrap(); + run_to_idle(&mut driver); + assert!(driver.is_caught_up()); + // when that block arrives and the next tick hits the halt + driver.source_mut().push_block(&[2]); + let outcome = driver.tick(); + // then the tick is Terminal and neither the driver nor its status claims caught up + assert_eq!( + outcome, + Tick::Terminal(EngineStatus::Halted { at: halt_pos }) + ); + assert!(!driver.is_caught_up()); + assert!(!driver.status().caught_up); + // and a manual recovery through engine_mut is not caught up before the next poll + driver.engine_mut().reset(RecordingFold::default()); + assert!(!driver.is_caught_up()); } + #[cfg(feature = "std")] #[test] - fn status_snapshot_reflects_engine() { - // given a driven driver over a two-block chain + fn a_caught_fold_panic_clears_caught_up() { + // given a caught-up driver whose fold panics on the next block's event let mut chain = ScriptedChain::new(1); chain.push_block(&[1]); - chain.push_block(&[2]); - let mut driver = new_driver(chain, engine_config(0), DriverConfig::default()); - driver.tick(); - // when reading status - let status = driver.status(); - // then cursor, skips, caught_up, engine status all match the engine accessors - assert_eq!(status.cursor, driver.engine().cursor()); - assert_eq!(status.last_verified, driver.engine().last_verified()); - assert_eq!(status.engine, driver.engine().status()); - assert_eq!(status.caught_up, driver.is_caught_up()); - assert_eq!(status.skips, driver.engine().skip_count()); + let fold = RecordingFold { + applied: Vec::new(), + fail_at: Some((Position::new(2, 0), FailKind::Panic)), + }; + let mut driver = + Driver::new(fold, chain, engine_config(0), DriverConfig::default()).unwrap(); + run_to_idle(&mut driver); + assert!(driver.is_caught_up()); + // when that block arrives and the panic out of the next tick is caught + driver.source_mut().push_block(&[2]); + let caught = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| driver.tick())); + // then the engine is Poisoned and neither the driver nor its status claims caught up + assert!(caught.is_err()); + assert_eq!( + driver.engine().status(), + EngineStatus::Poisoned { + at: Position::new(1, 0), + } + ); + assert!(!driver.is_caught_up()); + assert!(!driver.status().caught_up); } #[test] - fn reorged_content_produces_typed_fork_then_recovery() { - // given an applied chain with a checkpoint below the fork + fn fold_maintenance_between_ticks_keeps_caught_up() { + // given a driver over five one-event blocks let mut chain = ScriptedChain::new(1); - for value in 1..=6u64 { + for value in 1..=5u64 { + chain.push_block(&[value]); + } + let mut driver = new_driver(chain, engine_config(0), DriverConfig::default()); + // when it ticks and then touches the fold, as a loop that prunes it does, until + // it reports caught up + let mut ticks = 0; + while !driver.is_caught_up() { + driver.tick(); + driver.engine_mut().fold_mut(); + ticks += 1; + assert!(ticks < 50, "the driver never reported caught up"); + } + // then it gets there after one tick that folds and one that finds the head + assert_eq!(ticks, 2); + assert_eq!(folded(&driver), vec![1, 2, 3, 4, 5]); + } + + #[test] + fn a_halted_engine_rolled_back_to_its_own_cursor_is_not_caught_up() { + // given a caught-up driver with a checkpoint at its cursor, whose fold halts on + // the next block's event + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + let fold = RecordingFold { + applied: Vec::new(), + fail_at: Some((Position::new(2, 0), FailKind::Halt)), + }; + let mut driver = + Driver::new(fold, chain, engine_config(1), DriverConfig::default()).unwrap(); + run_to_idle(&mut driver); + assert!(driver.is_caught_up()); + // when that block arrives, the tick hits the halt, which leaves the cursor where + // it was, and the caller rolls back to the checkpoint there + driver.source_mut().push_block(&[2]); + driver.tick(); + let restored = driver.engine_mut().rollback_at_or_below(1).unwrap(); + // then the cursor is as it was, and the driver does not call that caught up + assert_eq!(restored, Some(Position::new(1, 0))); + assert!(!driver.is_caught_up()); + assert!(!driver.status().caught_up); + } + + #[cfg(feature = "std")] + #[test] + fn a_poisoned_engine_rolled_back_to_its_own_cursor_is_not_caught_up() { + // given a caught-up driver with a checkpoint at its cursor, whose fold panics on + // the next block's event + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + let fold = RecordingFold { + applied: Vec::new(), + fail_at: Some((Position::new(2, 0), FailKind::Panic)), + }; + let mut driver = + Driver::new(fold, chain, engine_config(1), DriverConfig::default()).unwrap(); + run_to_idle(&mut driver); + assert!(driver.is_caught_up()); + // when that block arrives, the panic out of the next tick is caught, and the + // caller rolls back to the checkpoint at the cursor, which the panic left alone + driver.source_mut().push_block(&[2]); + let caught = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| driver.tick())); + let restored = driver.engine_mut().rollback_at_or_below(1).unwrap(); + // then the cursor is as it was, and the driver does not call that caught up + assert!(caught.is_err()); + assert_eq!(restored, Some(Position::new(1, 0))); + assert!(!driver.is_caught_up()); + assert!(!driver.status().caught_up); + } + + #[test] + fn generation_increments_every_tick() { + // given a driver over an empty chain + let chain = ScriptedChain::new(1); + let mut driver = new_driver(chain, engine_config(0), DriverConfig::default()); + let start = driver.status().generation; + // when three ticks of any outcome run + driver.tick(); + driver.tick(); + driver.tick(); + // then status generation rose by three + assert_eq!(driver.status().generation, start + 3); + } + + #[test] + fn status_snapshot_reflects_engine() { + // given a driven driver over a two-block chain + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + chain.push_block(&[2]); + let mut driver = new_driver(chain, engine_config(0), DriverConfig::default()); + driver.tick(); + // when reading status + let status = driver.status(); + // then cursor, skips, caught_up, engine status all match the engine accessors + assert_eq!(status.cursor, driver.engine().cursor()); + assert_eq!(status.last_verified, driver.engine().last_verified()); + assert_eq!(status.engine, driver.engine().status()); + assert_eq!(status.caught_up, driver.is_caught_up()); + assert_eq!(status.skips, driver.engine().skip_count()); + } + + #[test] + fn reorged_content_produces_typed_fork_then_recovery() { + // given an applied chain with a checkpoint below the fork + let mut chain = ScriptedChain::new(1); + for value in 1..=6u64 { chain.push_block(&[value]); } chain.set_window(1); @@ -1331,6 +1986,106 @@ mod tests { assert_eq!(driver.engine().fold().applied, expected); } + #[test] + fn a_flapping_head_does_not_wear_the_checkpoints_down_to_a_resync() { + // given one event every 100 blocks, so each event block is a checkpoint, folded to + // block 700 at the defaults: four slots, one checkpoint per 64 blocks + let mut x = ScriptedChain::new(1); + for number in 1..=700u64 { + if number % 100 == 0 { + x.push_block(&[number]); + } else { + x.push_block(&[]); + } + } + x.set_window(1); + // and a second view of it whose head Y is X's sibling, with an event of its own + let mut y = x.clone(); + y.reorg(1, &[&[7001]]); + let mut driver = new_driver( + x.clone(), + EngineConfig::default(), + DriverConfig::from_block(1), + ); + run_to_idle(&mut driver); + assert_eq!(driver.engine().checkpoint_count(), 4); + // when the head flaps between X and Y, and the driver settles after each flap + let mut ticks = Vec::new(); + for flap in 0..6 { + *driver.source_mut() = if flap % 2 == 0 { y.clone() } else { x.clone() }; + ticks.extend(collect_to_idle(&mut driver)); + } + // then no flap resyncs, and the four checkpoints are all still there + assert!(!ticks.contains(&Tick::Resynced), "{ticks:?}"); + assert_eq!(driver.engine().checkpoint_count(), 4); + } + + /// Folds blocks `from..=to` of the chain, one event each, straight into the engine. + fn fold_blocks( + engine: &mut Engine, + chain: &ScriptedChain, + from: u64, + to: u64, + ) { + let mut batch = Batch::new(); + batch.boundary = from.checked_sub(1).and_then(|number| chain.header(number)); + for number in from..=to { + batch.push_block(chain.header(number).unwrap(), [(0, number)]); + } + engine.apply_batch(&batch).unwrap(); + } + + #[test] + fn a_resumed_driver_rolls_back_a_reorg_of_its_first_tip() { + // given an engine recovered at block 2 with no checkpoints, as a decode leaves it + let mut chain = ScriptedChain::new(1); + for value in 1..=6u64 { + chain.push_block(&[value]); + } + let mut engine = Engine::new(RecordingFold::default(), engine_config(2)).unwrap(); + fold_blocks(&mut engine, &chain, 1, 2); + let config = DriverConfig { + checkpoint_interval: Some(2), + ..DriverConfig::from_block(1) + }; + // when the driver resumes, folds to the tip in one poll, and the tip is replaced + let mut driver = + Driver::resume(engine, chain, RecordingFold::default(), config).unwrap(); + driver.tick(); + driver.source_mut().reorg(1, &[&[60]]); + let outcome = driver.tick(); + // then it rolls back to the recovery point instead of resyncing from genesis + assert_eq!( + outcome, + Tick::RolledBack { + to: Some(Position::new(2, 0)), + } + ); + } + + #[test] + fn a_driver_takes_no_extra_checkpoint_over_an_engine_that_has_one() { + // given an engine at block 4 whose one slot holds a checkpoint at block 2 + let mut chain = ScriptedChain::new(1); + for value in 1..=4u64 { + chain.push_block(&[value]); + } + let mut engine = Engine::new(RecordingFold::default(), engine_config(1)).unwrap(); + fold_blocks(&mut engine, &chain, 1, 2); + engine.checkpoint(); + fold_blocks(&mut engine, &chain, 3, 4); + // when a driver takes it over + let driver = Driver::resume( + engine, + chain, + RecordingFold::default(), + DriverConfig::from_block(1), + ) + .unwrap(); + // then the checkpoint at block 2 is still the only one + assert_eq!(driver.engine().durable_point(), Some(Position::new(2, 0))); + } + #[test] fn eventless_fork_point_is_still_detected() { // given events only on blocks 2 and 7 with cursor at 7 @@ -1416,7 +2171,7 @@ mod tests { } // when the chain reorgs at the sixth block and the tick surfaces the fork driver.source_mut().inner.reorg(3, &[&[60], &[70], &[80]]); - // the scan spends one probe per poll on the boundary; count only the bisection + // the earlier ticks spent probes too; count only the bisection driver.source_mut().calls = 0; let outcome = driver.tick(); // then rollback lands at or below the fifth and probe count is at most four @@ -1556,4 +2311,1110 @@ mod tests { ]; assert_eq!(driver.engine().fold().applied, expected); } + + #[test] + fn a_reorg_between_the_first_two_reads_leaves_no_orphan_in_the_fold() { + // given a driver folded to block 3, one block per poll + let mut chain = ScriptedChain::new(1); + for value in 1..=3u64 { + chain.push_block(&[value]); + } + chain.set_window(1); + let config = DriverConfig { + checkpoint_interval: Some(1), + ..DriverConfig::default() + }; + let mut driver = hostile_driver(chain, 4, config); + run_to_idle(&mut driver); + // when the chain replaces block 3 and adds a block right after the tick's first read + driver + .source_mut() + .reorg_before_read(2, 1, vec![vec![30], vec![40]]); + run_to_idle(&mut driver); + // then the fold holds exactly the canonical events + assert_eq!(folded(&driver), vec![1, 2, 30, 40]); + } + + #[test] + fn two_hashes_for_one_height_are_a_source_error() { + // given a one-block chain whose first answer also lists a rival block 1 + let mut chain = ScriptedChain::new(1); + chain.push_block(&[10]); + let rival = BlockRef { + number: 1, + hash: [9; 32], + }; + let mut source = Hostile::new(chain); + source.extra = vec![(rival, 1, 11)]; + let mut driver = Driver::new( + RecordingFold::default(), + source, + engine_config(0), + DriverConfig::from_block(1), + ) + .unwrap(); + // when ticked + let tick = driver.tick(); + // then the answer is refused and nothing folds + assert_eq!(tick, Tick::SourceError); + assert!(folded(&driver).is_empty()); + } + + #[test] + fn a_head_at_the_last_block_number_ends_the_walk() { + // given three empty blocks ending at u64::MAX, read two blocks per query + let first = u64::MAX - 2; + let mut chain = ScriptedChain::new(first); + for _ in 0..3 { + chain.push_block(&[]); + } + chain.set_window(2); + let mut driver = hostile_driver(chain, 0, DriverConfig::from_block(first)); + // when polled + let tick = driver.tick(); + // then the walk stops after the last window instead of reading it forever + assert_eq!(tick, Tick::Idle); + } + + #[test] + fn a_reorg_between_windows_does_not_skip_its_events() { + // given a driver folded to the only event, at block 5 of ten, two blocks per query + let mut chain = ScriptedChain::new(1); + for block in 1..=10u64 { + let events: &[u64] = if block == 5 { &[50] } else { &[] }; + chain.push_block(events); + } + chain.set_window(2); + let mut driver = hostile_driver(chain, 0, DriverConfig::from_block(1)); + run_to_idle(&mut driver); + // when blocks 6 to 10 are replaced after the scan read block 6 empty, with events + // at the new blocks 6 and 8 + let new_blocks = vec![vec![60], vec![], vec![80], vec![], vec![]]; + driver.source_mut().reorg_before_read(3, 5, new_blocks); + run_to_idle(&mut driver); + // then both new events fold + assert_eq!(folded(&driver), vec![50, 60, 80]); + } + + #[test] + fn a_torn_answer_is_not_folded() { + // given a six-block chain with events at blocks 1, 3 and 6 + let mut chain = ScriptedChain::new(1); + for events in [&[1][..], &[], &[30], &[], &[], &[60]] { + chain.push_block(events); + } + let old_block_3 = chain.header(3).unwrap(); + let mut source = Hostile::new(chain); + // when the chain reorgs after the head was read, and the one range answer still + // carries block 3 from the old chain + source.reorg_before_read(2, 4, vec![vec![], vec![], vec![], vec![61]]); + source.extra = vec![(old_block_3, 0, 30)]; + let mut driver = Driver::new( + RecordingFold::default(), + source, + engine_config(0), + DriverConfig::from_block(1), + ) + .unwrap(); + let first = driver.tick(); + run_to_idle(&mut driver); + // then the tick is refused and the fold ends with the new chain's events only + assert_eq!(first, Tick::SourceError); + assert_eq!(folded(&driver), vec![1, 61]); + } + + #[test] + fn a_torn_answer_ending_on_the_head_is_not_folded() { + // given a six-block chain with events at blocks 1, 3 and 6 + let mut chain = ScriptedChain::new(1); + for events in [&[1][..], &[], &[30], &[], &[], &[60]] { + chain.push_block(events); + } + let old_head = chain.header(6).unwrap(); + let mut source = Hostile::new(chain); + // when the chain reorgs after the head was read, and the one range answer still + // carries the old head block above a block of the new chain + source.reorg_before_read(2, 4, vec![vec![31], vec![], vec![], vec![]]); + source.extra = vec![(old_head, 0, 60)]; + let mut driver = Driver::new( + RecordingFold::default(), + source, + engine_config(0), + DriverConfig::from_block(1), + ) + .unwrap(); + let first = driver.tick(); + let after_first = folded(&driver); + run_to_idle(&mut driver); + // then the tick is refused and the fold ends with the new chain's events only + assert_eq!(first, Tick::SourceError); + assert!(after_first.is_empty()); + assert_eq!(folded(&driver), vec![1, 31]); + } + + #[test] + fn an_entry_outside_the_window_is_a_source_error() { + // given a three-block chain read two blocks per query, whose first answer also + // lists a block above its window + let mut chain = ScriptedChain::new(1); + for value in 1..=3u64 { + chain.push_block(&[value]); + } + chain.set_window(2); + let above = BlockRef { + number: 3, + hash: [7; 32], + }; + let mut source = Hostile::new(chain); + source.extra = vec![(above, 0, 777)]; + let mut driver = Driver::new( + RecordingFold::default(), + source, + engine_config(0), + DriverConfig::from_block(1), + ) + .unwrap(); + // when ticked + let tick = driver.tick(); + // then the answer is refused and nothing folds + assert_eq!(tick, Tick::SourceError); + assert!(folded(&driver).is_empty()); + } + + #[test] + fn an_entry_below_the_cursor_block_is_a_source_error() { + // given a driver folded to block 3 + let mut chain = ScriptedChain::new(1); + for value in 1..=3u64 { + chain.push_block(&[value]); + } + let mut driver = hostile_driver(chain, 0, DriverConfig::default()); + driver.tick(); + // when the next answer also lists an older block + let below = BlockRef { + number: 2, + hash: [7; 32], + }; + driver.source_mut().extra = vec![(below, 0, 555)]; + let tick = driver.tick(); + // then the answer is refused + assert_eq!(tick, Tick::SourceError); + assert_eq!(folded(&driver), vec![1, 2, 3]); + } + + #[test] + fn a_head_below_the_cursor_on_the_same_chain_is_idle() { + // given a driver folded to block 3 with a checkpoint there, over events at 1 and 3 + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + chain.push_block(&[]); + chain.push_block(&[3]); + let mut at_one = chain.clone(); + at_one.reorg(2, &[]); + let mut at_two = chain.clone(); + at_two.reorg(1, &[]); + let mut driver = new_driver(chain, engine_config(2), DriverConfig::default()); + driver.tick(); + driver.source_mut().fail_next_polls(1); + driver.tick(); + // when polls land on a node still at block 1, which the ring holds, and on one at + // block 2, which it does not + *driver.source_mut() = at_one; + let first = driver.tick(); + *driver.source_mut() = at_two; + let second = driver.tick(); + // then nothing moves: no rollback and no resync, and the error backoff ends + assert_eq!([first, second], [Tick::Idle; 2]); + assert_eq!(driver.next_delay(), Duration::from_secs(1)); + assert!(driver.is_caught_up()); + assert_eq!(driver.engine().cursor(), Some(Position::new(3, 0))); + assert_eq!(folded(&driver), vec![1, 3]); + } + + #[test] + fn a_cursor_inside_a_block_folds_the_rest_of_it() { + // given an engine that took the first event of block 2 and no more + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + chain.push_block(&[20, 21, 22]); + let mut engine = Engine::new(RecordingFold::default(), engine_config(0)).unwrap(); + let mut partial = Batch::new(); + partial.push_block(chain.header(1).unwrap(), [(0, 1)]); + partial.push_block(chain.header(2).unwrap(), [(0, 20)]); + engine.apply_batch(&partial).unwrap(); + // when a driver resumes over the whole chain + let mut driver = Driver::resume( + engine, + chain, + RecordingFold::default(), + DriverConfig::from_block(1), + ) + .unwrap(); + run_to_idle(&mut driver); + // then the rest of block 2 folds too + assert_eq!(folded(&driver), vec![1, 20, 21, 22]); + } + + #[test] + fn two_hashes_at_the_cursor_block_are_a_source_error() { + // given a driver folded to block 3 + let mut chain = ScriptedChain::new(1); + for value in 1..=3u64 { + chain.push_block(&[value]); + } + let mut driver = hostile_driver(chain, 0, DriverConfig::default()); + driver.tick(); + // when the next answer also lists a rival block 3 at an applied position + let rival = BlockRef { + number: 3, + hash: [7; 32], + }; + driver.source_mut().extra = vec![(rival, 0, 666)]; + let tick = driver.tick(); + // then the answer is refused + assert_eq!(tick, Tick::SourceError); + assert_eq!(folded(&driver), vec![1, 2, 3]); + } + + #[test] + fn only_a_poll_that_folds_re_reads_the_head_block() { + // given a driver folded to the head of a chain with an event in every block + let mut chain = ScriptedChain::new(1); + for value in 1..=3u64 { + chain.push_block(&[value]); + } + let mut driver = hostile_driver(chain, 0, DriverConfig::default()); + run_to_idle(&mut driver); + let reads = |driver: &mut Driver| { + let before = driver.source_mut().reads; + driver.tick(); + driver.source_mut().reads - before + }; + // when polled with no change, after an empty block, and after a block with an event + let idle = reads(&mut driver); + driver.source_mut().inner.push_block(&[]); + let empty_block = reads(&mut driver); + driver.source_mut().inner.push_block(&[9]); + let event = reads(&mut driver); + // then a poll reads the head and one range, and a fold also re-reads the head block + assert_eq!([idle, empty_block, event], [2, 2, 3]); + } + + #[test] + fn an_emptied_head_block_is_a_fork_found_without_reading_its_header() { + // given a driver folded to block 3, the head, with a checkpoint at each block + let config = DriverConfig { + checkpoint_interval: Some(1), + ..DriverConfig::default() + }; + let mut driver = hostile_driver(one_event_chain(3), 4, config); + run_to_idle(&mut driver); + // when block 3 is replaced by an empty block + driver.source_mut().inner.reorg(1, &[&[]]); + let before = driver.source_mut().reads; + let tick = driver.tick(); + // then the rollback costs the head, one range, and two bisection probes + assert_eq!( + tick, + Tick::RolledBack { + to: Some(Position::new(2, 0)), + } + ); + assert_eq!(driver.source_mut().reads - before, 4); + } + + #[test] + fn a_long_empty_walk_that_keeps_failing_still_finishes() { + // given an event, 1,000 empty blocks read four at a time, and another event, from + // a source that fails every 25th call, which is fewer calls than the walk needs + let chain = RefCell::new(sparse_chain(1_002, 4, &[1, 1_002])); + let calls = Cell::new(0); + let mut seen = 0; + let flaky = move || { + seen += 1; + seen % 25 == 0 + }; + let mut driver = remote_driver( + &chain, + &calls, + flaky, + engine_config(0), + DriverConfig::from_block(1), + ); + // when ticked until the second event folds + for _ in 0..1_000 { + driver.tick(); + if folded(&driver).len() == 2 { + break; + } + } + // then the walk got through the gap + assert_eq!(folded(&driver), vec![1, 1_002]); + } + + #[test] + fn a_token_bucket_rate_limit_still_finishes() { + // given 2,000 empty blocks read four at a time, then an event, from a source + // that allows 10 calls a second with a burst of 20, takes 20 ms a call, and fails + // the rest + let chain = RefCell::new(sparse_chain(2_001, 4, &[2_001])); + let calls = Cell::new(0); + let clock = Cell::new(0u64); + let (mut tokens, mut last) = (20_000u64, 0u64); + let limiter = { + let clock = &clock; + move || { + clock.set(clock.get() + 20); + // a millisecond at 10 calls a second earns ten thousandths of a call + tokens = (tokens + (clock.get() - last) * 10).min(20_000); + last = clock.get(); + let refused = tokens < 1_000; + if !refused { + tokens -= 1_000; + } + refused + } + }; + let mut driver = remote_driver( + &chain, + &calls, + limiter, + engine_config(0), + DriverConfig::from_block(1), + ); + // when ticked, sleeping each tick's delay, until the event folds + for _ in 0..10_000 { + driver.tick(); + if !folded(&driver).is_empty() { + break; + } + let delay = driver.next_delay(); + clock.set( + clock.get() + delay.as_secs() * 1_000 + u64::from(delay.subsec_millis()), + ); + } + // then the walk finished inside the limit + assert_eq!(folded(&driver), vec![2_001]); + } + + #[test] + fn a_reorg_that_adds_events_below_the_mark_drops_it() { + // given a driver folded to the event at block 1, whose next walk stopped at its + // cap over 256 empty blocks of a 600-block chain read four at a time + let chain = RefCell::new(sparse_chain(600, 4, &[1, 600])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + let capped = driver.tick(); + assert_eq!(capped, Tick::Progressed(ApplySummary::default())); + // when blocks 60 to 600 are replaced, with an event at 60 below the walk's end + replace_from(&mut chain.borrow_mut(), 60, 600, &[60, 600]); + settle(&mut driver); + // then the new event folds with the rest + assert_eq!(folded(&driver), vec![1, 60, 600]); + } + + #[test] + fn source_mut_drops_the_mark() { + // given a driver folded to the event at block 1, whose next walk stopped at its + // cap over a source that hides the event at block 60 + let chain = RefCell::new(sparse_chain(600, 4, &[1, 60, 600])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.source_mut().hide = Some(60); + driver.tick(); + let capped = driver.tick(); + assert_eq!(capped, Tick::Progressed(ApplySummary::default())); + // when the source starts to answer for block 60 + driver.source_mut().hide = None; + settle(&mut driver); + // then the walk starts again at the cursor and folds it + assert_eq!(folded(&driver), vec![1, 60, 600]); + } + + #[test] + fn engine_mut_drops_the_mark() { + // given a driver folded to block 3 one block per query, with a checkpoint at each + // event, whose next walk stopped at its cap over empty blocks + let chain = RefCell::new(sparse_chain(250, 1, &[1, 2, 3, 250])); + let calls = Cell::new(0); + let config = DriverConfig { + checkpoint_interval: Some(1), + ..DriverConfig::from_block(1) + }; + let mut driver = + remote_driver(&chain, &calls, || false, engine_config(4), config); + for _ in 0..3 { + driver.tick(); + } + let capped = driver.tick(); + assert_eq!(capped, Tick::Progressed(ApplySummary::default())); + // when the engine is rolled back to block 2 by hand + driver.engine_mut().rollback_at_or_below(2).unwrap(); + settle(&mut driver); + // then the walk starts again at the restored cursor and folds block 3 once more + assert_eq!(folded(&driver), vec![1, 2, 3, 250]); + } + + #[test] + fn a_long_empty_walk_finishes_through_fold_maintenance_between_its_ticks() { + // given an event, 600 empty blocks read four at a time, and another event + let chain = RefCell::new(sparse_chain(602, 4, &[1, 602])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + // when it ticks, touching the fold after each tick, until the second event folds + for _ in 0..100 { + driver.tick(); + driver.engine_mut().fold_mut(); + if folded(&driver).len() == 2 { + break; + } + } + // then the walk got through the gap, which takes more than one capped tick + assert_eq!(folded(&driver), vec![1, 602]); + } + + #[test] + fn a_manual_rollback_clears_caught_up_and_the_mark() { + // given a driver folded to block 3 one block per query, with a checkpoint at each + // event, that went idle after walks over empty blocks left a mark + let chain = RefCell::new(sparse_chain(250, 1, &[1, 2, 3])); + let calls = Cell::new(0); + let config = DriverConfig { + checkpoint_interval: Some(1), + ..DriverConfig::from_block(1) + }; + let mut driver = + remote_driver(&chain, &calls, || false, engine_config(4), config); + settle(&mut driver); + assert!(driver.is_caught_up()); + assert!(driver.walked.is_some()); + // when the engine is rolled back to block 2 by hand + driver.engine_mut().rollback_at_or_below(2).unwrap(); + // then it is not caught up + assert!(!driver.is_caught_up()); + assert!(!driver.status().caught_up); + // and the walk starts again at the restored cursor and folds block 3 once more + settle(&mut driver); + assert_eq!(folded(&driver), vec![1, 2, 3]); + assert!(driver.is_caught_up()); + } + + #[test] + fn a_manual_reset_of_an_engine_with_no_cursor_is_not_caught_up() { + // given a caught-up driver over an empty chain, so its engine has no cursor + let mut driver = new_driver( + ScriptedChain::new(1), + engine_config(0), + DriverConfig::default(), + ); + driver.tick(); + assert!(driver.is_caught_up()); + // when the engine is reset by hand + driver.engine_mut().reset(RecordingFold::default()); + // then the driver is not caught up before its next poll + assert!(!driver.is_caught_up()); + driver.tick(); + assert!(driver.is_caught_up()); + } + + #[test] + fn a_manual_reset_clears_caught_up_and_the_mark() { + // given a driver folded to the event at block 1 of 250 blocks read one at a time, + // that went idle after walks over empty blocks left a mark + let chain = RefCell::new(sparse_chain(250, 1, &[1])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + settle(&mut driver); + assert!(driver.is_caught_up()); + assert!(driver.walked.is_some()); + // when the engine is reset by hand + driver.engine_mut().reset(RecordingFold::default()); + // then it is not caught up + assert!(!driver.is_caught_up()); + assert!(!driver.status().caught_up); + // and it folds the chain again from the start instead of going on from the mark + settle(&mut driver); + assert_eq!(folded(&driver), vec![1]); + assert!(driver.is_caught_up()); + } + + #[test] + fn a_mark_off_the_head_chain_is_dropped_for_good() { + // given a fresh driver whose first walk stopped at its cap at block 256 of a + // 600-block chain with no events + let chain = RefCell::new(sparse_chain(600, 4, &[])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + // when blocks 2 and up are replaced by empty blocks up to 256, and the driver polls + // twice + replace_from(&mut chain.borrow_mut(), 2, 256, &[]); + let first = tick_counted(&mut driver, &calls); + let second = tick_counted(&mut driver, &calls); + // then the first poll finds the mark off the head's chain and walks from the + // start instead, 64 windows, and the second has no mark to check + assert_eq!([first, second], [(Tick::Idle, 66), (Tick::Idle, 65)]); + } + + #[test] + fn a_head_below_the_mark_is_idle_and_changes_nothing() { + // given a driver folded to the event at block 1, whose next walk stopped at its + // cap at block 257 of a 600-block chain + let chain = RefCell::new(sparse_chain(600, 4, &[1])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + driver.tick(); + let mark = driver.walked; + assert_eq!(mark, chain.borrow().header(257)); + // when the source answers from a node that is only at block 100 + let full = chain.borrow().clone(); + chain.borrow_mut().reorg(500, &[]); + let (tick, read) = tick_counted(&mut driver, &calls); + // then the tick reads the head and the cursor block under it, and stops, and + // nothing it holds moves + assert_eq!((tick, read), (Tick::Idle, 2)); + assert!(driver.is_caught_up()); + assert_eq!(driver.next_delay(), Duration::from_secs(1)); + assert_eq!(driver.engine().cursor(), Some(Position::new(1, 0))); + assert_eq!(folded(&driver), vec![1]); + assert_eq!(driver.walked, mark); + // and a node at the cursor block, which the ring holds, costs only the head read + let mut at_cursor = full.clone(); + at_cursor.reorg(599, &[]); + *chain.borrow_mut() = at_cursor; + assert_eq!(tick_counted(&mut driver, &calls), (Tick::Idle, 1)); + assert_eq!(driver.walked, mark); + // and the walk goes on from the mark once the node catches up, without a check of + // the mark, since the head is the one the capped walk read twice + *chain.borrow_mut() = full; + let (tick, read) = tick_counted(&mut driver, &calls); + assert_eq!( + (tick, read), + (Tick::Progressed(ApplySummary::default()), 67) + ); + } + + #[test] + fn a_first_sync_behind_the_mark_waits_for_the_node() { + // given a fresh driver whose first walk stopped at its cap at block 256 of a + // 600-block chain with no events + let chain = RefCell::new(sparse_chain(600, 4, &[])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + let mark = driver.walked; + assert_eq!(mark, chain.borrow().header(256)); + // when the source answers from a node that is only at block 100 + chain.borrow_mut().reorg(500, &[]); + let (tick, read) = tick_counted(&mut driver, &calls); + // then the tick reads the head and stops, and the mark stays + assert_eq!((tick, read), (Tick::Idle, 1)); + assert_eq!(driver.walked, mark); + } + + #[test] + fn a_head_that_contradicts_the_ring_is_a_fork_even_below_the_mark() { + // given a driver folded to the event at block 1, whose next walk stopped at its + // cap at block 257 of a 600-block chain + let chain = RefCell::new(sparse_chain(600, 4, &[1])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + driver.tick(); + // when the source answers from a one-block chain whose block 1 is another block + let mut other = sparse_chain(1, 4, &[]); + other.reorg(1, &[&[1]]); + *chain.borrow_mut() = other; + let tick = driver.tick(); + // then the tick is no wait for a node to catch up: the engine resyncs + assert_eq!(tick, Tick::Resynced); + // and the next tick folds the other chain's event at block 1 + assert_eq!(driver.walked, None); + driver.tick(); + assert_eq!(folded(&driver), vec![1]); + } + + #[test] + fn a_failed_tick_keeps_the_mark() { + // given a driver whose first walk stopped at its cap at block 256 of a 600-block + // chain with no events, and whose source fails its 70th call, the third of the + // next tick + let chain = RefCell::new(sparse_chain(600, 4, &[])); + let calls = Cell::new(0); + let mut seen = 0; + let gate = move || { + seen += 1; + seen == 70 + }; + let mut driver = remote_driver( + &chain, + &calls, + gate, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + // when that tick fails, and the next one runs + let failed = driver.tick(); + let mark = driver.walked; + let next = tick_counted(&mut driver, &calls); + // then the failure left the mark, and the next walk, half as long, goes on from it + assert_eq!(failed, Tick::SourceError); + assert_eq!(mark, chain.borrow().header(256)); + assert_eq!(next, (Tick::Progressed(ApplySummary::default()), 35)); + assert_eq!(driver.walked, chain.borrow().header(384)); + } + + #[test] + fn a_fold_inside_the_cap_costs_what_it_did() { + // given a fresh driver over a 600-block chain read four at a time, with an event + // in block 10 + let chain = RefCell::new(sparse_chain(600, 4, &[10])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + // when ticked once + let (tick, read) = tick_counted(&mut driver, &calls); + // then the walk stops at the window with the event, and the tick reads the head, + // three windows, and the head block again + let applied = ApplySummary { + applied: 1, + deduped: 0, + skipped: 0, + }; + assert_eq!((tick, read), (Tick::Progressed(applied), 5)); + } + + #[test] + fn a_head_on_the_mark_costs_one_call() { + // given a driver folded to the event at block 1, whose next walk stopped at its + // cap at block 257 of a 600-block chain + let chain = RefCell::new(sparse_chain(600, 4, &[1])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + driver.tick(); + // when the source answers from a node whose head is that block + chain.borrow_mut().reorg(343, &[]); + let (tick, read) = tick_counted(&mut driver, &calls); + // then the tick reads the head and nothing else + assert_eq!((tick, read), (Tick::Idle, 1)); + } + + #[test] + fn a_long_walk_goes_on_from_the_block_its_last_tick_ended_on() { + // given a fresh driver over 600 blocks read four at a time, whose only event is + // in the last block + let chain = RefCell::new(sparse_chain(600, 4, &[600])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + // when ticked four times, counting each tick's calls + let mut ticks = vec![tick_counted(&mut driver, &calls)]; + let first_delay = driver.next_delay(); + let first_caught_up = driver.is_caught_up(); + ticks.extend((0..3).map(|_| tick_counted(&mut driver, &calls))); + // then two ticks stop at 64 windows with nothing to fold, and ask for no delay + // and no caught-up; each costs the head, its windows, the block it ended on and + // the head block again, and the second starts above the mark without checking it, + // since the head is the one the first read twice + let empty = Tick::Progressed(ApplySummary::default()); + let applied = Tick::Progressed(ApplySummary { + applied: 1, + deduped: 0, + skipped: 0, + }); + assert_eq!( + ticks, + vec![(empty, 67), (empty, 67), (applied, 24), (Tick::Idle, 1)] + ); + assert_eq!(first_delay, Duration::ZERO); + assert!(!first_caught_up); + assert_eq!(folded(&driver), vec![600]); + } + + #[test] + fn a_failed_walk_is_halved_and_a_good_one_doubles_the_next() { + // given 400 empty blocks read one at a time, from a source that fails its first + // eight calls + let chain = RefCell::new(sparse_chain(400, 1, &[])); + let calls = Cell::new(0); + let mut seen = 0; + let gate = move || { + seen += 1; + seen <= 8 + }; + let mut driver = remote_driver( + &chain, + &calls, + gate, + engine_config(0), + DriverConfig::from_block(1), + ); + // when ticked sixteen times + let read: Vec = (0..16) + .map(|_| tick_counted(&mut driver, &calls).1) + .collect(); + // then the span runs 64 down to 1 and stays there, then 1 up to 64 and stays + // there: a tick costs its windows, the head, the block it ended on and the head + // block again, with no check of the mark, since the head is the one the tick + // before read twice + assert_eq!( + read, + vec![1, 1, 1, 1, 1, 1, 1, 1, 4, 5, 7, 11, 19, 35, 67, 67] + ); + } + + #[test] + fn a_head_reorg_does_not_cost_a_walk_its_progress() { + // given a driver whose two capped walks over 600 empty blocks read four at a time + // ended at blocks 256 and 512 + let chain = RefCell::new(sparse_chain(600, 4, &[])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + driver.tick(); + // when the head block is replaced by two blocks, and later one more block follows + replace_from(&mut chain.borrow_mut(), 600, 601, &[]); + let replaced = tick_counted(&mut driver, &calls); + chain.borrow_mut().push_block(&[]); + let followed = tick_counted(&mut driver, &calls); + // then each walk starts above block 512 and reaches the head: the head, the check + // of the mark, and 89 or 90 blocks in 23 windows + assert_eq!([replaced, followed], [(Tick::Idle, 25); 2]); + } + + #[test] + fn a_walk_that_fails_its_boundary_check_leaves_no_mark() { + // given a driver folded to the event at block 1 of a 600-block chain read four at + // a time + let chain = RefCell::new(sparse_chain(600, 4, &[1, 600])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.tick(); + // when the whole chain is replaced, so the next walk stops at its cap over empty + // blocks and finds block 1 changed + { + let mut chain = chain.borrow_mut(); + let mut blocks = vec![vec![11]]; + blocks.resize(599, vec![]); + blocks.push(vec![600]); + let blocks: Vec<&[u64]> = blocks.iter().map(Vec::as_slice).collect(); + chain.reorg(600, &blocks); + } + let recovery = driver.tick(); + settle(&mut driver); + // then the engine resyncs and folds the new chain from the start + assert_eq!(recovery, Tick::Resynced); + assert_eq!(folded(&driver), vec![11, 600]); + } + + #[test] + fn an_anchor_that_is_not_the_block_asked_for_is_a_source_error() { + // given a source that answers the request for block 256 with block 257's header + let chain = RefCell::new(sparse_chain(600, 4, &[])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + driver.source_mut().lie_at = Some(256); + // when a walk stops at its cap and asks for the block it ended on + let tick = driver.tick(); + // then the answer is refused + assert_eq!(tick, Tick::SourceError); + } + + /// A driver over `chain`, which is read one block at a time, folded to the head, with + /// a checkpoint at each block. + fn head_cache_driver<'a>( + chain: &'a RefCell, + calls: &'a Cell, + ) -> Driver bool>> { + let config = DriverConfig { + checkpoint_interval: Some(1), + ..DriverConfig::from_block(1) + }; + let mut driver = remote_driver(chain, calls, || false, engine_config(8), config); + settle(&mut driver); + driver + } + + #[test] + fn an_unchanged_head_skips_the_read_of_the_cursor_block() { + // given a driver over five one-event blocks read one at a time + let chain = RefCell::new(sparse_chain(5, 1, &[1, 2, 3, 4, 5])); + let calls = Cell::new(0); + let mut driver = remote_driver( + &chain, + &calls, + || false, + engine_config(0), + DriverConfig::from_block(1), + ); + // when it ticks to the head and once more, counting each tick's calls + let reads: Vec = (0..6) + .map(|_| tick_counted(&mut driver, &calls).1) + .collect(); + // then a tick that folds reads the head, the next block and the head block again, + // and the poll after the last reads the head and nothing else + assert_eq!(reads, vec![3, 3, 3, 3, 3, 1]); + } + + #[test] + fn a_reorg_below_the_cursor_block_is_caught_when_the_head_changes() { + // given a driver folded to block 5 of five one-event blocks + let chain = RefCell::new(sparse_chain(5, 1, &[1, 2, 3, 4, 5])); + let calls = Cell::new(0); + let mut driver = head_cache_driver(&chain, &calls); + // when blocks 3 to 5 are replaced by blocks with other events, so the head is + // another block at the same height + chain.borrow_mut().reorg(3, &[&[30], &[40], &[50]]); + let found = driver.tick(); + settle(&mut driver); + // then the next poll finds the fork and rolls back, and the new events fold + assert_eq!( + found, + Tick::RolledBack { + to: Some(Position::new(2, 0)), + } + ); + assert_eq!(folded(&driver), vec![1, 2, 30, 40, 50]); + } + + #[test] + fn source_mut_drops_the_head_cache() { + // given a driver folded to the head, whose poll with the head unchanged reads the + // head and nothing else + let chain = RefCell::new(sparse_chain(5, 1, &[1, 2, 3, 4, 5])); + let calls = Cell::new(0); + let mut driver = head_cache_driver(&chain, &calls); + assert_eq!(tick_counted(&mut driver, &calls), (Tick::Idle, 1)); + // when the source is borrowed, which may change what it returns + driver.source_mut(); + // then the next poll reads the cursor block again + assert_eq!(tick_counted(&mut driver, &calls), (Tick::Idle, 2)); + } + + #[test] + fn a_manual_rollback_drops_the_head_cache() { + // given a driver folded to the head, whose poll with the head unchanged reads the + // head and nothing else + let chain = RefCell::new(sparse_chain(5, 1, &[1, 2, 3, 4, 5])); + let calls = Cell::new(0); + let mut driver = head_cache_driver(&chain, &calls); + assert_eq!(tick_counted(&mut driver, &calls), (Tick::Idle, 1)); + // when the engine is rolled back to block 3 by hand + driver.engine_mut().rollback_at_or_below(3).unwrap(); + let refold = tick_counted(&mut driver, &calls); + // then the next tick reads the cursor block again, the next block, and the head + // block, and folds block 4 + let applied = ApplySummary { + applied: 1, + deduped: 0, + skipped: 0, + }; + assert_eq!(refold, (Tick::Progressed(applied), 4)); + } + + #[test] + fn a_rollback_drops_the_head_cache() { + // given a driver folded to block 5 of five one-event blocks read one at a time, + // with a checkpoint at each + let chain = RefCell::new(sparse_chain(5, 1, &[1, 2, 3, 4, 5])); + let calls = Cell::new(0); + let mut driver = head_cache_driver(&chain, &calls); + let original = chain.borrow().clone(); + // when block 5 is replaced, the next poll rolls back to block 4, and the chain + // goes back to the original block 5 + chain.borrow_mut().reorg(1, &[&[50]]); + let found = driver.tick(); + *chain.borrow_mut() = original; + let refold = tick_counted(&mut driver, &calls); + // then the poll after the rollback reads the cursor block again, the next block, + // and the head block, though the head is the one the driver vouched for before it + let applied = ApplySummary { + applied: 1, + deduped: 0, + skipped: 0, + }; + assert_eq!( + found, + Tick::RolledBack { + to: Some(Position::new(4, 0)), + } + ); + assert_eq!(refold, (Tick::Progressed(applied), 4)); + } + + #[test] + fn an_answer_that_fails_the_shape_check_vouches_for_no_head() { + // given a driver folded to block 5 of five one-event blocks read two at a time, + // with a checkpoint at each + let mut chain = one_event_chain(5); + chain.set_window(2); + let config = DriverConfig { + checkpoint_interval: Some(1), + ..DriverConfig::from_block(1) + }; + let mut driver = hostile_driver(chain, 8, config); + run_to_idle(&mut driver); + // when block 5 is replaced and a block 6 follows it, and the first answer from + // that chain also lists a rival block 6 + let rival = BlockRef { + number: 6, + hash: [7; 32], + }; + let source = driver.source_mut(); + source.inner.reorg(1, &[&[50], &[60]]); + source.extra = vec![(rival, 0, 666)]; + let refused = driver.tick(); + // and the next answer is clean + let found = driver.tick(); + // then the refused tick folded nothing, and the next one still checks the cursor + // block against the new chain + assert_eq!(refused, Tick::SourceError); + assert_eq!( + found, + Tick::RolledBack { + to: Some(Position::new(4, 0)), + } + ); + } + + #[test] + fn an_idle_poll_on_a_new_head_does_not_vouch_for_it() { + // given a driver folded to the head of five one-event blocks read one at a time + let chain = RefCell::new(sparse_chain(5, 1, &[1, 2, 3, 4, 5])); + let calls = Cell::new(0); + let mut driver = head_cache_driver(&chain, &calls); + // when a block with no events arrives and the driver polls twice + chain.borrow_mut().push_block(&[]); + let first = tick_counted(&mut driver, &calls); + let second = tick_counted(&mut driver, &calls); + // then each poll reads the head, the cursor block and the new block, since the + // walk that found nothing read the head only once + assert_eq!([first, second], [(Tick::Idle, 3); 2]); + } + + #[test] + fn a_resumed_driver_reads_the_rest_of_its_cursor_block() { + // given an engine that took the first event of block 2 and no more, over a chain + // read one block at a time + let mut scripted = ScriptedChain::new(1); + scripted.push_block(&[1]); + scripted.push_block(&[20, 21, 22]); + scripted.push_block(&[30]); + scripted.set_window(1); + let chain = RefCell::new(scripted); + let mut engine = Engine::new(RecordingFold::default(), engine_config(0)).unwrap(); + let mut partial = Batch::new(); + partial.push_block(chain.borrow().header(1).unwrap(), [(0, 1)]); + partial.push_block(chain.borrow().header(2).unwrap(), [(0, 20)]); + engine.apply_batch(&partial).unwrap(); + let calls = Cell::new(0); + let source = Remote { + chain: &chain, + calls: &calls, + gate: || false, + hide: None, + lie_at: None, + }; + let mut driver = Driver::resume( + engine, + source, + RecordingFold::default(), + DriverConfig::from_block(1), + ) + .unwrap(); + // when it is ticked to the head and once more, counting each tick's calls + let reads: Vec = (0..3) + .map(|_| tick_counted(&mut driver, &calls).1) + .collect(); + // then the first tick reads the cursor block and folds the rest of it, and only + // the ticks after it find a head they have vouched for + assert_eq!(folded(&driver), vec![1, 20, 21, 22, 30]); + assert_eq!(reads, vec![3, 3, 1]); + } } diff --git a/crates/chainfold/src/engine.rs b/crates/chainfold/src/engine.rs index 8c4a13b..f0a79a6 100644 --- a/crates/chainfold/src/engine.rs +++ b/crates/chainfold/src/engine.rs @@ -30,13 +30,15 @@ use crate::{ const MIN_RING_CAPACITY: usize = 2; /// Largest allowed observed-block ring capacity. const MAX_RING_CAPACITY: usize = 1 << 20; +/// Largest allowed checkpoint slot count. +const MAX_CHECKPOINT_SLOTS: usize = 1 << 16; /// Fixed engine construction parameters. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct EngineConfig { /// Observed-block window W; power of two between 2 and 1 << 20. pub ring_capacity: usize, - /// Retained checkpoint slots K; zero disables rollback. + /// Retained checkpoint slots K, at most 1 << 16; zero disables rollback. pub checkpoint_slots: usize, } @@ -50,7 +52,7 @@ impl Default for EngineConfig { } impl EngineConfig { - /// Accepts a power-of-two ring capacity within the allowed range. + /// Accepts a power-of-two ring capacity and a slot count within the allowed ranges. pub(crate) fn validate(&self) -> Result<(), ConfigError> { if !self.ring_capacity.is_power_of_two() { return Err(ConfigError::RingCapacityNotPowerOfTwo { @@ -62,6 +64,11 @@ impl EngineConfig { got: self.ring_capacity, }); } + if self.checkpoint_slots > MAX_CHECKPOINT_SLOTS { + return Err(ConfigError::CheckpointSlotsOutOfRange { + got: self.checkpoint_slots, + }); + } Ok(()) } } @@ -89,6 +96,7 @@ pub struct Engine { status: EngineStatus, last_verified: Option, skips: u64, + resets: u64, } impl Engine { @@ -103,6 +111,7 @@ impl Engine { status: EngineStatus::Active, last_verified: None, skips: 0, + resets: 0, }) } @@ -126,6 +135,11 @@ impl Engine { self.skips } + /// Count of `reset` calls, wrapping; how a driver notices a restart it did not run. + pub(crate) fn resets(&self) -> u64 { + self.resets + } + /// Count of checkpoints the ring can still serve; expired slots are not counted. pub fn checkpoint_count(&self) -> usize { self.checkpoints.count(&self.ring) @@ -204,6 +218,11 @@ impl Engine { self.last_verified = Some(boundary); } + // A panic in the fold unwinds past this mark and leaves its span half applied, so + // the engine stays Poisoned. Every other exit restores Active or sets its own. + self.status = EngineStatus::Poisoned { + at: self.cursor.unwrap_or(Position::new(0, 0)), + }; let mut summary = ApplySummary::default(); for span in batch.spans() { let redelivered = self @@ -213,11 +232,13 @@ impl Engine { && let Some(observed_hash) = self.ring.hash_at(span.number) && &observed_hash != span.hash { + self.status = EngineStatus::Active; return Err(fork_suspected(span.number, observed_hash, span.block())); } self.apply_span(&span, &mut summary)?; } + self.status = EngineStatus::Active; Ok(summary) } @@ -245,8 +266,7 @@ impl Engine { /// 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. + /// Clears Halted and Poisoned; drops checkpoints newer than the restored cursor. /// Freshness resets to None, since the restored cursor is unverified until the /// next boundary check confirms it. Slots whose cursor block has left the observed /// window are expired, so NoCheckpointAtOrBelow also names an exhausted window. @@ -273,7 +293,7 @@ impl Engine { } self.status = EngineStatus::Active; self.last_verified = None; - self.checkpoints.drop_above(block); + self.checkpoints.drop_newer(self.cursor); Ok(self.cursor) } @@ -286,6 +306,7 @@ impl Engine { self.status = EngineStatus::Active; self.last_verified = None; self.skips = 0; + self.resets = self.resets.wrapping_add(1); } /// Terminal for automated paths; only reset leaves this state. @@ -614,7 +635,7 @@ mod tests { // when a span at that number returns a different hash let next = batch_of(Some(block(5, 0)), vec![(block(5, 1), vec![0])]); let result = engine.apply_batch(&next); - // then ForkSuspected + // then ForkSuspected, and the engine stays Active since no fold failed assert_eq!( result, Err(ApplyError::ForkSuspected { @@ -622,6 +643,7 @@ mod tests { refetched: block(5, 1), }) ); + assert_eq!(engine.status(), EngineStatus::Active); } #[test] @@ -830,6 +852,32 @@ mod tests { assert_eq!(engine.fold().applied, vec![(poison_pos, 0)]); } + #[cfg(feature = "std")] + #[test] + fn a_fold_panic_leaves_the_engine_poisoned() { + // given a fold that panics on the second event of a three-event block + let mut engine = scripted_engine(Position::new(1, 1), FailKind::Panic); + let batch = batch_of(None, vec![(block(1, 0), vec![0, 1, 2])]); + // when the panic is caught + let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + engine.apply_batch(&batch) + })); + // then the engine is Poisoned, so redelivering the batch folds nothing twice + let poisoned = EngineStatus::Poisoned { + at: Position::new(0, 0), + }; + assert!(caught.is_err()); + assert_eq!(engine.status(), poisoned); + assert_eq!( + engine.apply_batch(&batch), + Err(ApplyError::NotActive { status: poisoned }) + ); + assert_eq!( + engine.fold().applied, + vec![(Position::new(1, 0), 0), (Position::new(1, 1), 1)] + ); + } + #[test] fn ring_records_each_observed_block_once() { // given two batches over four blocks @@ -883,6 +931,32 @@ mod tests { ); } + #[test] + fn config_rejects_checkpoint_slots_past_the_cap() { + // given slot counts at the cap, one past it, and far past it + let config = |checkpoint_slots| EngineConfig { + ring_capacity: 8, + checkpoint_slots, + }; + // when constructing + let at_cap = Engine::new(RecordingFold::default(), config(MAX_CHECKPOINT_SLOTS)); + let past_cap = + Engine::new(RecordingFold::default(), config(MAX_CHECKPOINT_SLOTS + 1)); + let far_past = Engine::new(RecordingFold::default(), config(usize::MAX)); + // then only the cap is accepted, and the rest are typed errors, never a panic + assert!(at_cap.is_ok()); + assert_eq!( + past_cap.err(), + Some(ConfigError::CheckpointSlotsOutOfRange { + got: MAX_CHECKPOINT_SLOTS + 1, + }) + ); + assert_eq!( + far_past.err(), + Some(ConfigError::CheckpointSlotsOutOfRange { got: usize::MAX }) + ); + } + #[test] fn checkpoint_then_rollback_restores_view() { // given a checkpoint at block 3 and applies through block 6 @@ -1166,7 +1240,11 @@ mod tests { fn unrecoverable_refuses_apply_and_rollback() { // given mark_unrecoverable let mut engine = new_engine(); - engine.mark_unrecoverable(DivergenceCause::ForkBeyondWindow); + let cause = DivergenceCause::HorizonExceeded { + needed: 0, + horizon: 1, + }; + engine.mark_unrecoverable(cause); // when applying or rolling back let apply_result = engine.apply_batch(&batch_of(None, vec![(block(1, 0), vec![0])])); @@ -1175,17 +1253,10 @@ mod tests { assert_eq!( apply_result, Err(ApplyError::NotActive { - status: EngineStatus::Unrecoverable { - cause: DivergenceCause::ForkBeyondWindow - }, - }) - ); - assert_eq!( - rollback_result, - Err(RollbackError::Unrecoverable { - cause: DivergenceCause::ForkBeyondWindow, + status: EngineStatus::Unrecoverable { cause }, }) ); + assert_eq!(rollback_result, Err(RollbackError::Unrecoverable { cause })); } #[test] @@ -1232,6 +1303,40 @@ mod tests { assert_eq!(engine.checkpoint_count(), 0); } + #[test] + fn rollback_to_a_cursorless_checkpoint_drops_the_expired_ones_above_it() { + // given a cursorless checkpoint and one at block 1, which a ring of two then forgets + let mut engine = Engine::new( + RecordingFold::default(), + EngineConfig { + ring_capacity: 2, + checkpoint_slots: 2, + }, + ) + .unwrap(); + engine.checkpoint(); + engine + .apply_batch(&batch_of(None, vec![(block(1, 0), vec![0, 1])])) + .unwrap(); + engine.checkpoint(); + engine + .apply_batch(&batch_of(Some(block(1, 0)), vec![(block(2, 0), vec![0])])) + .unwrap(); + engine + .apply_batch(&batch_of(Some(block(2, 0)), vec![(block(3, 0), vec![0])])) + .unwrap(); + // when it rolls back to the cursorless one, and a replaced block 1 is folded + engine.rollback_at_or_below(2).unwrap(); + engine + .apply_batch(&batch_of(None, vec![(block(1, 1), vec![0])])) + .unwrap(); + engine.checkpoint(); + // then the old block 1 checkpoint stays gone instead of living again, so the + // durable point is the cursorless one rather than a position past the cursor + assert_eq!(engine.cursor(), Some(Position::new(1, 0))); + assert_eq!(engine.durable_point(), None); + } + #[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 409ebe9..6dd4f2a 100644 --- a/crates/chainfold/src/error.rs +++ b/crates/chainfold/src/error.rs @@ -84,8 +84,6 @@ impl core::error::Error for EngineStatus {} /// Terminal divergence reason the engine cannot recover from automatically. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DivergenceCause { - /// Fork deeper than the oldest observed block in the ring. - ForkBeyondWindow, /// Replay is required but the source's horizon no longer covers the start block. HorizonExceeded { /// Block replay must start from. @@ -98,7 +96,6 @@ pub enum DivergenceCause { 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::HorizonExceeded { needed, horizon } => { write!( f, @@ -139,6 +136,11 @@ pub enum ConfigError { /// Capacity the caller asked for. got: usize, }, + /// Checkpoint slot count is above the allowed maximum. + CheckpointSlotsOutOfRange { + /// Slot count the caller asked for. + got: usize, + }, /// Source cannot replay from the configured start block. HorizonExceedsStart { /// Block the driver folds from. @@ -157,6 +159,9 @@ impl fmt::Display for ConfigError { Self::RingCapacityOutOfRange { got } => { write!(f, "ring capacity {got} is out of the allowed range") } + Self::CheckpointSlotsOutOfRange { got } => { + write!(f, "checkpoint slot count {got} is out of the allowed range") + } Self::HorizonExceedsStart { start, horizon } => { write!(f, "replay horizon {horizon} exceeds start block {start}") } diff --git a/crates/chainfold/src/harness.rs b/crates/chainfold/src/harness.rs index 006beb8..99687b9 100644 --- a/crates/chainfold/src/harness.rs +++ b/crates/chainfold/src/harness.rs @@ -1,6 +1,11 @@ //! Tokio harness: runs a `Driver` on its own thread and publishes status into a watch. use std::{ + panic::{ + AssertUnwindSafe, + catch_unwind, + resume_unwind, + }, sync::{ Arc, Condvar, @@ -53,13 +58,22 @@ where loop_state.0.lock().expect("harness state mutex poisoned"); std::mem::take(&mut guard.checkpoint_requests) }; - // Requests coalesce into one checkpoint, keeping the K slots at K cursors. - if requests > 0 { - driver.checkpoint(); - } - driver.tick(); + // A fold panic leaves the engine Poisoned: publish that terminal status and end + // the loop. A panic that leaves the engine active is not the fold's, so rethrow. + let outcome = catch_unwind(AssertUnwindSafe(|| { + // Requests coalesce into one checkpoint, keeping the K slots at K cursors. + if requests > 0 { + driver.checkpoint(); + } + driver.tick(); + })); let status = driver.status(); let _ = status_tx.send(status); + if let Err(panic) = outcome + && !status.is_terminal() + { + resume_unwind(panic); + } let stop = loop_state .0 .lock() @@ -131,8 +145,8 @@ impl Handle { /// Returns when the durable cursor reaches `pos` or the driver is terminal. /// - /// A later resync lowers the durable cursor, so the answer holds for the - /// instant it resolves. + /// A later resync clears the durable cursor until an offer made after it commits, so + /// the answer holds for the instant it resolves. pub async fn wait_durable(&mut self, pos: Position) -> DriverStatus { self.settled(|s| s.is_terminal() || s.durable_cursor.is_some_and(|c| c >= pos)) .await @@ -166,6 +180,11 @@ impl Handle { #[cfg(test)] mod tests { use std::{ + convert::Infallible, + sync::atomic::{ + AtomicBool, + Ordering, + }, time::Duration, vec, vec::Vec, @@ -178,6 +197,10 @@ mod tests { DriverConfig, }, engine::EngineConfig, + error::{ + EngineStatus, + FoldError, + }, test_util::{ FailKind, RecordingFold, @@ -337,6 +360,79 @@ mod tests { assert!(status.is_terminal()); } + #[tokio::test] + async fn a_fold_panic_publishes_a_poisoned_status_and_keeps_the_driver() { + // given a driver whose fold panics on the second block's event + let mut chain = ScriptedChain::new(1); + chain.push_block(&[1]); + chain.push_block(&[2]); + let fold = RecordingFold { + applied: Vec::new(), + fail_at: Some((Position::new(2, 0), FailKind::Panic)), + }; + let driver = Driver::new(fold, chain, engine_config(0), fast_config()).unwrap(); + let mut handle = spawn(driver); + // when awaiting caught up + let status = timeout(handle.wait_caught_up()).await; + // then the wait resolves with the poisoned status instead of a stale active one + assert!(matches!(status.engine, EngineStatus::Poisoned { .. })); + assert!(status.is_terminal()); + // and shutdown returns the driver with each event folded once + let driver = handle.shutdown().await; + let expected = vec![(Position::new(1, 0), 1), (Position::new(2, 0), 2)]; + assert_eq!(driver.engine().fold().applied, expected); + } + + /// Fold whose clone panics once armed, as a bug outside `apply` would. + struct CloneBomb(Arc); + + impl Clone for CloneBomb { + fn clone(&self) -> Self { + assert!(!self.0.load(Ordering::Relaxed), "armed clone panicked"); + Self(Arc::clone(&self.0)) + } + } + + impl Fold for CloneBomb { + type Event = u64; + type Error = Infallible; + + fn apply( + &mut self, + _pos: Position, + _event: &u64, + ) -> Result<(), FoldError> { + Ok(()) + } + } + + #[tokio::test] + #[should_panic(expected = "harness shutdown task panicked")] + async fn a_panic_outside_the_fold_still_unwinds_the_loop() { + // given a caught-up driver whose fold clone panics once armed + let armed = Arc::new(AtomicBool::new(false)); + let driver = Driver::new( + CloneBomb(Arc::clone(&armed)), + ScriptedChain::new(1), + engine_config(2), + fast_config(), + ) + .unwrap(); + let mut handle = spawn(driver); + timeout(handle.wait_caught_up()).await; + // when a checkpoint request makes the loop clone the fold + armed.store(true, Ordering::Relaxed); + handle.request_checkpoint(); + timeout(async { + while !handle.thread.is_finished() { + tokio::time::sleep(FAST_POLL).await; + } + }) + .await; + // then the panic was rethrown, so the loop thread is dead and shutdown says so + handle.shutdown().await; + } + #[tokio::test] async fn request_checkpoint_is_executed_by_the_loop() { // given a spawned driver diff --git a/crates/chainfold/src/sink.rs b/crates/chainfold/src/sink.rs index b99cd0f..b931e7d 100644 --- a/crates/chainfold/src/sink.rs +++ b/crates/chainfold/src/sink.rs @@ -10,8 +10,12 @@ use crate::{ pub trait SnapshotSink { /// Offers the engine's durable point for persistence; must not fsync. fn offer(&mut self, engine: &Engine) -> Result<(), DurabilityLost>; - /// Cursor a restart would recover; falls when a resync persists older state. + /// Cursor a restart would recover; None after a `reset` until a later offer commits. fn durable_cursor(&self) -> Option; + /// Forgets the offers made so far: the engine restarted from genesis, or the sink holds a + /// cursor the engine has not reached. Before it returns `durable_cursor` is None, and it + /// stays None until an offer made after this call commits. + fn reset(&mut self); } /// Sink that persists nothing; the default for drivers without durability. @@ -26,6 +30,8 @@ impl SnapshotSink for NoSink { fn durable_cursor(&self) -> Option { None } + + fn reset(&mut self) {} } #[cfg(test)] diff --git a/crates/chainfold/src/snapshot.rs b/crates/chainfold/src/snapshot.rs index 4ae37e9..9b3abd3 100644 --- a/crates/chainfold/src/snapshot.rs +++ b/crates/chainfold/src/snapshot.rs @@ -13,7 +13,10 @@ use crate::{ Engine, EngineConfig, }, - error::ConfigError, + error::{ + ConfigError, + EngineStatus, + }, fold::Fold, position::{ BlockRef, @@ -188,6 +191,11 @@ pub enum SnapshotError { /// Longest field length the codec accepts, in bytes. limit: usize, }, + /// Engine is poisoned, so its fold state is untrusted and is not encoded. + Poisoned { + /// Position the engine poisoned at. + at: Position, + }, /// Stored ring holds more entries than the configured capacity. RingExceedsCapacity { /// Entries the stored ring holds. @@ -229,6 +237,13 @@ impl fmt::Display for SnapshotError { Self::TooLarge { limit } => { write!(f, "snapshot envelope field exceeds the {limit} byte limit") } + Self::Poisoned { at } => { + write!( + f, + "engine poisoned at block {} log index {}; its state is not encoded", + at.block, at.log_index + ) + } Self::RingExceedsCapacity { len, capacity } => { write!( f, @@ -268,6 +283,9 @@ impl Engine { &self, out: &mut Vec, ) -> Result<(), SnapshotError> { + if let EngineStatus::Poisoned { at } = self.status() { + return Err(SnapshotError::Poisoned { at }); + } encode_envelope(self.fold(), self.cursor(), self.observed(), out) } @@ -344,7 +362,6 @@ impl Engine { } let cursor = (envelope.cursor_set != 0) .then(|| Position::new(envelope.cursor_block, envelope.cursor_log_index)); - // A cursorless engine has an empty ring, so None must pair with no entries. if envelope.ring_numbers.last().copied() != cursor.map(|pos| pos.block) { return Err(SnapshotError::RingCursorMismatch); } @@ -413,7 +430,10 @@ mod tests { Position, }, snapshot::Persist, - test_util::RecordingFold, + test_util::{ + FailKind, + RecordingFold, + }, }; #[cfg(not(feature = "std"))] use alloc::{ @@ -756,20 +776,53 @@ mod tests { assert_eq!(result.unwrap_err(), SnapshotError::RingCursorMismatch); } + /// Engine whose fold fails with `kind` on the second event of block 1. + fn engine_failed_with(kind: FailKind) -> Engine { + let fold = RecordingFold { + applied: Vec::new(), + fail_at: Some((Position::new(1, 1), kind)), + }; + let mut engine = Engine::new(fold, test_config()).unwrap(); + let result = engine.apply_batch(&batch_of(None, vec![(block(1, 0), vec![0, 1])])); + assert!(result.is_err()); + engine + } + #[test] - fn cursorless_envelope_with_a_ring_is_refused() { - // given a hand-built envelope with no cursor but one observed block + fn a_poisoned_engine_refuses_to_encode_but_a_halted_one_encodes() { + // given an engine poisoned by its fold and one halted by it + let poisoned = engine_failed_with(FailKind::Poison); + let halted = engine_failed_with(FailKind::Halt); + // when each encodes into its own buffer + let (mut poisoned_bytes, mut halted_bytes) = (Vec::new(), Vec::new()); + let refused = poisoned.encode_snapshot(&mut poisoned_bytes); + let accepted = halted.encode_snapshot(&mut halted_bytes); + // then only the poisoned engine is refused, and it appends nothing + assert_eq!( + refused, + Err(SnapshotError::Poisoned { + at: Position::new(1, 1), + }) + ); + assert!(poisoned_bytes.is_empty()); + assert_eq!(accepted, Ok(())); + assert!(!halted_bytes.is_empty()); + } + + #[test] + fn cursorless_snapshot_with_a_ring_is_refused() { + // given a hand-built envelope with no cursor and a one-block ring let mut state = Vec::new(); RecordingFold::default().encode_state(&mut state); let bytes = encode_custom( RecordingFold::STATE_TAG.as_bytes(), None, - &[block(5, 0)], + &[block(3, 0)], state, ); // when decoded let result = Engine::::decode_snapshot(&bytes, test_config()); - // then RingCursorMismatch; no engine reaches a cursorless state with a ring + // then RingCursorMismatch, since an engine without a cursor has observed nothing assert_eq!(result.unwrap_err(), SnapshotError::RingCursorMismatch); } } diff --git a/crates/chainfold/src/source.rs b/crates/chainfold/src/source.rs index 3a7bb7e..b80bb9c 100644 --- a/crates/chainfold/src/source.rs +++ b/crates/chainfold/src/source.rs @@ -24,8 +24,8 @@ pub trait Source { /// Failure reading the source. type Error; - /// Highest block this source will serve. - fn head(&mut self) -> Result; + /// Header of the highest block this source will serve. + fn head(&mut self) -> Result; /// Header of one block, or None when it is not on the current chain. fn header_at(&mut self, number: u64) -> Result, Self::Error>; diff --git a/crates/chainfold/src/storage.rs b/crates/chainfold/src/storage.rs index 9ff021b..27ce045 100644 --- a/crates/chainfold/src/storage.rs +++ b/crates/chainfold/src/storage.rs @@ -6,7 +6,6 @@ mod store; mod vfs; pub use flusher::{ - DurabilityToken, FlushError, Flusher, }; diff --git a/crates/chainfold/src/storage/flusher.rs b/crates/chainfold/src/storage/flusher.rs index c4cc701..e7c8e18 100644 --- a/crates/chainfold/src/storage/flusher.rs +++ b/crates/chainfold/src/storage/flusher.rs @@ -119,7 +119,8 @@ struct Lane(T); struct Shared { admission: Lane, poisoned: Lane, - durable_cursor: Lane>>, + /// Reset epoch and watermark; a commit moves the watermark only in its own epoch. + durable_cursor: Lane)>>, } impl Shared { @@ -127,7 +128,7 @@ impl Shared { Self { admission: Lane(Admission::new(queue_depth)), poisoned: Lane(AtomicBool::new(false)), - durable_cursor: Lane(Mutex::new(cursor)), + durable_cursor: Lane(Mutex::new((0, cursor))), } } @@ -143,7 +144,7 @@ impl Shared { } } -/// Shared cell one durability token waits on. +/// Cell the flusher thread settles with one submitted snapshot's outcome. #[derive(Default)] struct TokenState { result: Mutex>>, @@ -159,36 +160,24 @@ impl TokenState { self.ready.notify_all(); } } -} - -/// One submitted snapshot's durability outcome. -#[derive(Clone)] -pub struct DurabilityToken { - state: Arc, -} -impl DurabilityToken { /// Blocks until the flush completes or fails. - pub fn wait(&self) -> Result<(), FlushError> { - let guard = lock_recovering(&self.state.result); + #[cfg(test)] + fn wait(&self) -> Result<(), FlushError> { + let guard = lock_recovering(&self.result); let guard = self - .state .ready .wait_while(guard, |result| result.is_none()) .unwrap_or_else(|error| error.into_inner()); (*guard).expect("condvar wakes only after the result is set") } - - /// Non-blocking read of the flush outcome, if it has completed. - pub fn try_result(&self) -> Option> { - *lock_recovering(&self.state.result) - } } /// One snapshot handed from the apply thread to the flusher thread. struct Job { snapshot: Vec, cursor: Option, + epoch: u64, token: Arc, shared: Arc, } @@ -215,9 +204,11 @@ fn run_flusher( let committed = store.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. - *watermark = committed; + // tracks it down as well as up. A job queued before a reset commits + // without moving it. + if watermark.0 == job.epoch { + watermark.1 = committed; + } drop(watermark); job.token.complete(Ok(())); } @@ -236,7 +227,10 @@ pub struct Flusher { impl Flusher { /// Spawns the background fsync thread over an opened store. + /// + /// A `queue_depth` of 0 counts as 1. pub fn spawn(store: SnapshotStore, queue_depth: usize) -> Self { + let queue_depth = queue_depth.max(1); let (sender, receiver) = mpsc::sync_channel::(queue_depth); let shared = Arc::new(Shared::new(queue_depth, store.durable_cursor())); let thread_shared = Arc::clone(&shared); @@ -249,11 +243,30 @@ impl Flusher { } /// Fails fast once a prior flush failed; the watermark never moves after that. - pub fn submit( + /// + /// Private, so only an engine's own offer can move the watermark: + /// + /// ```compile_fail,E0624 + /// use chainfold::{ + /// Position, + /// storage::{ + /// Flusher, + /// RealVfs, + /// }, + /// }; + /// + /// // given the flusher a driver hands out through `Driver::sink` + /// fn given(flusher: &Flusher) { + /// // when a caller submits a snapshot at a cursor the engine never reached + /// flusher.submit(Vec::new(), Some(Position::new(1_000_000, 0))); + /// } + /// // then it does not compile + /// ``` + fn submit( &self, snapshot: Vec, cursor: Option, - ) -> Result { + ) -> Result, FlushError> { if self.shared.is_poisoned() { return Err(FlushError::Closed); } @@ -262,19 +275,20 @@ impl Flusher { let job = Job { snapshot, cursor, + epoch: lock_recovering(&self.shared.durable_cursor.0).0, token: Arc::clone(&token), shared: Arc::clone(&self.shared), }; if self.sender.send(job).is_err() { return Err(FlushError::Closed); } - Ok(DurabilityToken { state: token }) + Ok(token) } - /// Cursor a reopen of the store would recover; falls when a resync commits - /// older state. + /// Cursor a reopen of the store would recover; None after a `reset` until an + /// offer made after it commits. pub fn durable_cursor(&self) -> Option { - *lock_recovering(&self.shared.durable_cursor.0) + lock_recovering(&self.shared.durable_cursor.0).1 } /// Stops the thread and returns the store; pending jobs complete first. @@ -304,6 +318,12 @@ impl SnapshotSink for Flusher { fn durable_cursor(&self) -> Option { Flusher::durable_cursor(self) } + + /// Starts a new epoch, so jobs already queued commit without moving the watermark. + fn reset(&mut self) { + let mut watermark = lock_recovering(&self.shared.durable_cursor.0); + *watermark = (watermark.0.wrapping_add(1), None); + } } #[cfg(test)] @@ -327,6 +347,10 @@ mod tests { }; use crate::{ batch::Batch, + driver::{ + Driver, + DriverConfig, + }, engine::{ Engine, EngineConfig, @@ -343,6 +367,7 @@ mod tests { test_util::{ CrashVfs, RecordingFold, + ScriptedChain, }, }; @@ -642,4 +667,101 @@ mod tests { let second = handle.join().unwrap().unwrap(); assert_eq!(second.wait(), Ok(())); } + + #[test] + fn zero_queue_depth_counts_as_one() { + // given a flusher spawned with queue_depth 0 + let dir = PathBuf::from("/flusher"); + let (store, _) = SnapshotStore::open(config(dir, 1), CrashVfs::new()).unwrap(); + let flusher = Flusher::spawn(store, 0); + // when a snapshot is submitted from another thread, so a stall cannot hang the test + let (done_tx, done_rx) = mpsc::channel(); + thread::spawn(move || { + let token = flusher.submit(b"snapshot".to_vec(), Some(Position::new(1, 0))); + let _ = done_tx.send(token.map(|token| token.wait())); + }); + // then the flush completes instead of blocking the first submit forever + assert_eq!(done_rx.recv_timeout(Duration::from_secs(5)), Ok(Ok(Ok(())))); + } + + #[test] + fn a_job_queued_before_reset_commits_without_moving_the_watermark() { + // given a watermark at (2, 0) and a job for (5, 0) held at the commit gate + let dir = PathBuf::from("/flusher-reset"); + let (release_tx, release_rx) = mpsc::channel(); + let vfs = GatedVfs::new(CrashVfs::new(), release_rx, 3); + let (store, _) = SnapshotStore::open(config(dir, 1), vfs).unwrap(); + let mut flusher = Flusher::spawn(store, 4); + let settled = flusher + .submit(b"settled".to_vec(), Some(Position::new(2, 0))) + .unwrap(); + assert_eq!(settled.wait(), Ok(())); + assert_eq!(flusher.durable_cursor(), Some(Position::new(2, 0))); + let queued = flusher + .submit(b"queued".to_vec(), Some(Position::new(5, 0))) + .unwrap(); + // when the flusher is reset and the held job then commits + SnapshotSink::::reset(&mut flusher); + let cleared = flusher.durable_cursor(); + release_tx.send(()).unwrap(); + assert_eq!(queued.wait(), Ok(())); + // then the watermark stayed None, and only a job submitted after the reset moves it + assert_eq!(cleared, None); + assert_eq!(flusher.durable_cursor(), None); + let fresh = flusher + .submit(b"fresh".to_vec(), Some(Position::new(1, 0))) + .unwrap(); + assert_eq!(fresh.wait(), Ok(())); + assert_eq!(flusher.durable_cursor(), Some(Position::new(1, 0))); + } + + #[test] + fn with_sink_over_a_store_ahead_of_the_engine_reports_no_durable_cursor() { + // given a flusher over a store that already holds a snapshot at (9, 0) + let dir = PathBuf::from("/flusher-ahead"); + let (mut store, _) = + SnapshotStore::open(config(dir, 1), CrashVfs::new()).unwrap(); + store + .commit(b"earlier run", Some(Position::new(9, 0))) + .unwrap(); + let flusher = Flusher::spawn(store, 4); + assert_eq!(flusher.durable_cursor(), Some(Position::new(9, 0))); + // when a driver over a fresh engine takes it as its sink + let driver = Driver::with_sink( + RecordingFold::default(), + ScriptedChain::new(1), + flusher, + engine_config(), + DriverConfig::default(), + ) + .unwrap(); + // then neither the driver's status nor the sink reports a durable cursor + assert_eq!(driver.status().durable_cursor, None); + assert_eq!(driver.sink().durable_cursor(), None); + } + + #[test] + fn resuming_over_a_flusher_at_the_engine_cursor_keeps_its_watermark() { + // given an engine recovered at (2, 0) and a flusher whose store holds that snapshot + let dir = PathBuf::from("/flusher-resume"); + let (mut store, _) = + SnapshotStore::open(config(dir, 1), CrashVfs::new()).unwrap(); + store + .commit(b"recovered", Some(Position::new(2, 0))) + .unwrap(); + let mut engine = Engine::new(RecordingFold::default(), engine_config()).unwrap(); + advance(&mut engine, 1); + advance(&mut engine, 2); + // when a driver resumes the engine over that flusher + let driver = Driver::resume_with_sink( + engine, + ScriptedChain::new(1), + Flusher::spawn(store, 4), + RecordingFold::default(), + DriverConfig::default(), + ) + .unwrap(); + // then the watermark is not reset + assert_eq!(driver.status().durable_cursor, Some(Position::new(2, 0))); + } } diff --git a/crates/chainfold/src/test_util.rs b/crates/chainfold/src/test_util.rs index 98f531e..f153a9a 100644 --- a/crates/chainfold/src/test_util.rs +++ b/crates/chainfold/src/test_util.rs @@ -43,6 +43,8 @@ pub enum FailKind { Halt, /// Records the event, then refuses it. Poison, + /// Records the event, then panics. + Panic, } impl Fold for RecordingFold { @@ -64,6 +66,10 @@ impl Fold for RecordingFold { self.applied.push((pos, *event)); Err(FoldError::Poison(kind)) } + FailKind::Panic => { + self.applied.push((pos, *event)); + panic!("scripted fold panic at {pos:?}") + } }; } self.applied.push((pos, *event)); @@ -128,7 +134,7 @@ impl Fold for NoopFold { /// Sink recording each accepted offer's durable point; fails scripted offers. #[derive(Debug, Clone, Default, PartialEq)] pub struct WatermarkSink { - /// Durable point of every accepted offer, in offer order. + /// Durable point of every accepted offer since the last reset, in offer order. pub offered: Vec, /// Offers still scripted to fail before the sink accepts again. pub fail_next_offers: u32, @@ -149,6 +155,10 @@ impl SnapshotSink for WatermarkSink { fn durable_cursor(&self) -> Option { self.offered.last().copied() } + + fn reset(&mut self) { + self.offered.clear(); + } } /// Number of splitmix64 lanes composing a block hash. @@ -295,15 +305,16 @@ impl Source for ScriptedChain { type Event = u64; type Error = PollFailure; - fn head(&mut self) -> Result { + fn head(&mut self) -> Result { if self.pending_failures > 0 { self.pending_failures -= 1; return Err(PollFailure); } - Ok(self - .blocks - .last() - .map_or(self.first_block.saturating_sub(1), |block| block.number)) + // An empty chain heads at the parent of its first block. + Ok(self.tip().unwrap_or(BlockRef { + number: self.first_block.saturating_sub(1), + hash: [0u8; 32], + })) } fn header_at(&mut self, number: u64) -> Result, PollFailure> { @@ -320,7 +331,13 @@ impl Source for ScriptedChain { usize::try_from(number.saturating_sub(self.first_block)) .map_or(self.blocks.len(), |i| i.min(self.blocks.len())) }; - let (start, end) = (index(from), index(to.saturating_add(1))); + // One past `to`, without the `to + 1` that saturates at u64::MAX. + let end = if to < self.first_block { + 0 + } else { + index(to).saturating_add(1).min(self.blocks.len()) + }; + let start = index(from); for block in &self.blocks[start.min(end)..end] { let header = BlockRef { number: block.number, @@ -733,16 +750,30 @@ mod tests { assert_eq!(numbers, vec![2, 3]); } + #[test] + fn events_in_includes_the_block_numbered_u64_max() { + // given blocks numbered u64::MAX - 1 and u64::MAX + let mut chain = ScriptedChain::new(u64::MAX - 1); + chain.push_block(&[1]); + chain.push_block(&[2]); + let mut out = Vec::new(); + // when reading a range that ends at u64::MAX + chain.events_in(u64::MAX - 1, u64::MAX, &mut out).unwrap(); + // then both blocks' events + let numbers: Vec = out.iter().map(|(block, ..)| block.number).collect(); + assert_eq!(numbers, vec![u64::MAX - 1, u64::MAX]); + } + #[test] fn head_reports_the_tip_and_underflows_to_before_first_on_empty() { // given an empty chain starting at 1 let mut chain = ScriptedChain::new(1); // then head is below the first block, so nothing is in range - assert_eq!(chain.head(), Ok(0)); + assert_eq!(chain.head().map(|head| head.number), Ok(0)); // and once blocks exist it is the tip chain.push_block(&[1]); chain.push_block(&[2]); - assert_eq!(chain.head(), Ok(2)); + assert_eq!(chain.head(), Ok(chain.header(2).unwrap())); } #[test] @@ -771,6 +802,6 @@ mod tests { // then two errors then success assert_eq!(first, Err(PollFailure)); assert_eq!(second, Err(PollFailure)); - assert_eq!(third, Ok(1)); + assert_eq!(third.map(|head| head.number), Ok(1)); } } diff --git a/crates/chainfold/tests/apply_alloc.rs b/crates/chainfold/tests/apply_alloc.rs index 2995b50..a16f348 100644 --- a/crates/chainfold/tests/apply_alloc.rs +++ b/crates/chainfold/tests/apply_alloc.rs @@ -9,13 +9,7 @@ use std::{ Layout, System, }, - sync::{ - Mutex, - atomic::{ - AtomicU64, - Ordering::Relaxed, - }, - }, + cell::Cell, }; use chainfold::{ @@ -26,17 +20,21 @@ use chainfold::{ test_util::NoopFold, }; -static ALLOCATIONS: AtomicU64 = AtomicU64::new(0); -static ALLOCATED_BYTES: AtomicU64 = AtomicU64::new(0); -/// Serializes measured regions, since the counters are process wide. -static MEASURING: Mutex<()> = Mutex::new(()); +thread_local! { + /// Allocations and bytes this thread asked for. Per thread, so the test harness's own + /// threads never land in a measurement. + static COUNTS: Cell<(u64, u64)> = const { Cell::new((0, 0)) }; +} struct Counting; unsafe impl GlobalAlloc for Counting { unsafe fn alloc(&self, layout: Layout) -> *mut u8 { - ALLOCATIONS.fetch_add(1, Relaxed); - ALLOCATED_BYTES.fetch_add(layout.size() as u64, Relaxed); + // A thread can still allocate while its locals are torn down; those go uncounted. + let _ = COUNTS.try_with(|counts| { + let (allocations, bytes) = counts.get(); + counts.set((allocations + 1, bytes + layout.size() as u64)); + }); unsafe { System.alloc(layout) } } @@ -48,6 +46,11 @@ unsafe impl GlobalAlloc for Counting { #[global_allocator] static GLOBAL: Counting = Counting; +/// This thread's allocation count and allocated bytes so far. +fn counts() -> (u64, u64) { + COUNTS.with(Cell::get) +} + /// Deterministic block header: the block number embedded directly in the hash bytes. fn block_ref(number: u64) -> BlockRef { let mut hash = [0u8; 32]; @@ -69,9 +72,6 @@ fn block_batch(boundary: Option, number: u64, count: u32) -> Batch u64 { ring_capacity: 8, checkpoint_slots: 0, }; - let before = ALLOCATED_BYTES.load(Relaxed); + let (_, before) = counts(); let result = Engine::::decode_snapshot(bytes, config); - let after = ALLOCATED_BYTES.load(Relaxed); + let (_, after) = counts(); assert!(result.is_err(), "corrupt input decoded successfully"); after - before } @@ -114,9 +114,6 @@ fn corrupt_snapshot_decode_allocates_within_the_input_length() { // given every truncation of a valid snapshot plus one bit-flipped copy use chainfold::test_util::RecordingFold; - let _measuring = MEASURING - .lock() - .unwrap_or_else(|poison| poison.into_inner()); let mut engine = Engine::new( RecordingFold::default(), EngineConfig { diff --git a/crates/chainfold/tests/lagging.rs b/crates/chainfold/tests/lagging.rs new file mode 100644 index 0000000..a9ac0c3 --- /dev/null +++ b/crates/chainfold/tests/lagging.rs @@ -0,0 +1,111 @@ +use chainfold::{ + BlockRef, + Driver, + DriverConfig, + EngineConfig, + Position, + Source, + Tick, + test_util::{ + PollFailure, + RecordingFold, + ScriptedChain, + }, +}; + +/// Answers `head` from one node and every other read from another. +struct TwoNodes { + head: ScriptedChain, + reads: ScriptedChain, +} + +impl Source for TwoNodes { + type Event = u64; + type Error = PollFailure; + + fn head(&mut self) -> Result { + self.head.head() + } + + fn header_at(&mut self, number: u64) -> Result, PollFailure> { + self.reads.header_at(number) + } + + fn events_in( + &mut self, + from: u64, + to: u64, + out: &mut Vec<(BlockRef, u32, u64)>, + ) -> Result<(), PollFailure> { + self.reads.events_in(from, to, out) + } +} + +/// A driver folded to the tip of a `len`-block chain with one event at each of `events`. +fn folded(len: u64, events: &[u64]) -> Driver { + let mut chain = ScriptedChain::new(1); + for n in 1..=len { + let event = [n]; + chain.push_block(if events.contains(&n) { &event } else { &[] }); + } + let source = TwoNodes { + head: chain.clone(), + reads: chain, + }; + let config = EngineConfig { + ring_capacity: 64, + checkpoint_slots: 4, + }; + let mut driver = Driver::new( + RecordingFold::default(), + source, + config, + DriverConfig::from_block(1), + ) + .unwrap(); + while driver.tick() != Tick::Idle {} + driver +} + +fn applied(driver: &Driver) -> Vec { + driver + .engine() + .fold() + .applied + .iter() + .map(|(_, event)| *event) + .collect() +} + +#[test] +fn a_shorter_fork_with_no_ring_entry_at_its_head_is_found() { + // given a driver folded to the events at blocks 10 and 20 + let mut driver = folded(20, &[10, 20]); + // when both nodes move to a fork from block 9 that ends at 15, with an event at 12 + let fork: [&[u64]; 6] = [&[], &[], &[12], &[], &[], &[]]; + let source = driver.source_mut(); + source.head.reorg(11, &fork); + source.reads.reorg(11, &fork); + let ticks: Vec = (0..3).map(|_| driver.tick()).collect(); + // then the orphaned ring entry at 10 sends the driver into recovery, and it folds the + // fork's event instead of waiting on the head + assert_eq!(ticks[0], Tick::Resynced); + assert_eq!(ticks[2], Tick::Idle); + assert_eq!(applied(&driver), vec![12]); +} + +#[test] +fn a_node_with_no_header_under_its_own_head_is_lagging_not_forked() { + // given a driver folded to the events at blocks 10, 20 and 30 + let mut driver = folded(30, &[10, 20, 30]); + // when the head comes from a node at 25 and every other read from a node at 5, both + // on the same chain + let source = driver.source_mut(); + source.head.reorg(5, &[]); + source.reads.reorg(25, &[]); + let tick = driver.tick(); + // then the missing header at 20 proves no fork, and nothing is thrown away + assert_eq!(tick, Tick::Idle); + assert_eq!(driver.engine().cursor(), Some(Position::new(30, 0))); + assert_eq!(applied(&driver), vec![10, 20, 30]); +} diff --git a/crates/examples/examples/shielded_wallet.rs b/crates/examples/examples/shielded_wallet.rs index 94a788e..a0aec88 100644 --- a/crates/examples/examples/shielded_wallet.rs +++ b/crates/examples/examples/shielded_wallet.rs @@ -157,11 +157,18 @@ impl Source for Pool { type Error = core::convert::Infallible; type Event = u64; - fn head(&mut self) -> Result { - Ok(self - .blocks - .last() - .map_or(self.first_block.saturating_sub(1), |block| block.number)) + fn head(&mut self) -> Result { + // An empty pool heads at the parent of its first block. + Ok(self.blocks.last().map_or( + BlockRef { + number: self.first_block.saturating_sub(1), + hash: [0u8; 32], + }, + |block| BlockRef { + number: block.number, + hash: block.hash, + }, + )) } fn header_at(&mut self, number: u64) -> Result, Self::Error> {