From 59b21b30dd497f78d78a836935fc5bc8322b2f3e Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:01:08 +0530 Subject: [PATCH 1/8] fix(chainfold): guard the trust boundaries - decode refuses a cursorless snapshot that carries a ring, a state the engine never produces and cannot fold from - EngineConfig rejects more than 1 << 16 checkpoint slots with a typed error; Engine::new and decode_snapshot panicked on a huge count - Flusher::submit is private and DurabilityToken is gone, so only an engine's own offer moves the durable watermark; a caller holding the sink through Driver::sink could push it past the engine cursor - Flusher::spawn counts a queue depth of 0 as 1; 0 blocked the first offer forever --- crates/chainfold/src/engine.rs | 37 +++++++++++++- crates/chainfold/src/error.rs | 8 +++ crates/chainfold/src/snapshot.rs | 9 ++-- crates/chainfold/src/storage.rs | 1 - crates/chainfold/src/storage/flusher.rs | 65 +++++++++++++++++-------- 5 files changed, 92 insertions(+), 28 deletions(-) diff --git a/crates/chainfold/src/engine.rs b/crates/chainfold/src/engine.rs index 8c4a13b..6b7ae6c 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(()) } } @@ -883,6 +890,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 diff --git a/crates/chainfold/src/error.rs b/crates/chainfold/src/error.rs index 409ebe9..cc3675c 100644 --- a/crates/chainfold/src/error.rs +++ b/crates/chainfold/src/error.rs @@ -139,6 +139,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 +162,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/snapshot.rs b/crates/chainfold/src/snapshot.rs index 4ae37e9..047b36d 100644 --- a/crates/chainfold/src/snapshot.rs +++ b/crates/chainfold/src/snapshot.rs @@ -344,7 +344,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); } @@ -757,19 +756,19 @@ mod tests { } #[test] - fn cursorless_envelope_with_a_ring_is_refused() { - // given a hand-built envelope with no cursor but one observed block + 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/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..a0d0887 100644 --- a/crates/chainfold/src/storage/flusher.rs +++ b/crates/chainfold/src/storage/flusher.rs @@ -143,7 +143,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,30 +159,17 @@ 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. @@ -236,7 +223,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 +239,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); } @@ -268,7 +277,7 @@ impl Flusher { 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 @@ -642,4 +651,20 @@ 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(())))); + } } From 2fa3408e0300893f225086cbfe2cf7f38473707f Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:16:33 +0530 Subject: [PATCH 2/8] fix(chainfold): poison the engine before the fold runs A panic out of Fold::apply that the caller caught left the engine Active with a half-applied span. The next batch then folded its events twice, a snapshot of it decoded as a trusted engine, the harness loop died without a terminal status, and the driver kept reporting caught up. - apply_batch marks the engine Poisoned once per batch, before the span loop, and restores Active on every exit that is not a panic - encode_snapshot refuses a Poisoned engine with SnapshotError::Poisoned; a Halted engine still encodes - the harness catches a fold panic, publishes the Poisoned status, ends the loop, and shutdown still returns the driver; a panic that leaves the engine Active is rethrown - is_caught_up and DriverStatus::caught_up are false while the engine is not Active, so a Halt, a Poison and a caught panic stop reading as caught up --- crates/chainfold/src/driver.rs | 60 ++++++++++++++++- crates/chainfold/src/engine.rs | 36 +++++++++- crates/chainfold/src/harness.rs | 106 ++++++++++++++++++++++++++++-- crates/chainfold/src/snapshot.rs | 58 +++++++++++++++- crates/chainfold/src/test_util.rs | 6 ++ 5 files changed, 256 insertions(+), 10 deletions(-) diff --git a/crates/chainfold/src/driver.rs b/crates/chainfold/src/driver.rs index 6bb0529..e355dde 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -304,7 +304,7 @@ where /// True once the most recent poll returned no new blocks. pub fn is_caught_up(&self) -> bool { - self.caught_up + self.caught_up && self.engine.status().is_active() } /// Runs the interval-based checkpoint rule. @@ -538,7 +538,7 @@ 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(), durability_lost: self.durability_lost, @@ -1255,6 +1255,62 @@ mod tests { assert_eq!(driver.engine().cursor(), Some(Position::new(5, 0))); } + #[test] + 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); + } + + #[cfg(feature = "std")] + #[test] + 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]); + 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 generation_increments_every_tick() { // given a driver over an empty chain diff --git a/crates/chainfold/src/engine.rs b/crates/chainfold/src/engine.rs index 6b7ae6c..f5b19fe 100644 --- a/crates/chainfold/src/engine.rs +++ b/crates/chainfold/src/engine.rs @@ -211,6 +211,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 @@ -220,11 +225,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) } @@ -621,7 +628,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 { @@ -629,6 +636,7 @@ mod tests { refetched: block(5, 1), }) ); + assert_eq!(engine.status(), EngineStatus::Active); } #[test] @@ -837,6 +845,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 diff --git a/crates/chainfold/src/harness.rs b/crates/chainfold/src/harness.rs index 006beb8..88e6f2d 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() @@ -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/snapshot.rs b/crates/chainfold/src/snapshot.rs index 047b36d..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) } @@ -412,7 +430,10 @@ mod tests { Position, }, snapshot::Persist, - test_util::RecordingFold, + test_util::{ + FailKind, + RecordingFold, + }, }; #[cfg(not(feature = "std"))] use alloc::{ @@ -755,6 +776,39 @@ 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 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 diff --git a/crates/chainfold/src/test_util.rs b/crates/chainfold/src/test_util.rs index 98f531e..4e2dd6d 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)); From 27cc4c232a69424baffd3b7346157962c254bbd4 Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:50:44 +0530 Subject: [PATCH 3/8] fix(chainfold)!: keep the bookkeeping in step through a rollback or a reset - a rollback drops every checkpoint newer than the restored cursor, not only those above the fork block, so an expired checkpoint cannot come back to life when the refold reaches its block again. It also moves the write position back over the dropped slots, so the refold refills them before it evicts a live checkpoint, and a head that flaps between two siblings no longer wears the slots down to a resync - both cadence marks are clamped to the cursor block at the start of every tick, so a rollback, whoever ran it, no longer suppresses the next checkpoint or offer; a resync clears them the same way - SnapshotSink gains a required reset. After it, durable_cursor is None until an offer made after the call commits. The driver calls it when the engine was reset, whether by a resync or by hand, and when it takes over a sink that is ahead of its engine. DriverStatus::durable_cursor is None while a reset is unseen. The Flusher keeps a reset epoch beside its watermark, so a job queued before a reset commits without raising it. NoSink does nothing and WatermarkSink forgets its offers - a driver over a recovered engine that holds no checkpoint takes one at the recovery point, so a one-block reorg of the first tip rolls back instead of resyncing from the start block - chainfold 0.4.0, since SnapshotSink::reset is required --- crates/chainfold/Cargo.toml | 2 +- crates/chainfold/README.md | 3 +- crates/chainfold/src/checkpoint.rs | 12 +- crates/chainfold/src/driver.rs | 241 ++++++++++++++++++++++-- crates/chainfold/src/engine.rs | 47 ++++- crates/chainfold/src/harness.rs | 4 +- crates/chainfold/src/sink.rs | 8 +- crates/chainfold/src/storage/flusher.rs | 113 ++++++++++- crates/chainfold/src/test_util.rs | 6 +- 9 files changed, 400 insertions(+), 36 deletions(-) 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..696aaf7 100644 --- a/crates/chainfold/README.md +++ b/crates/chainfold/README.md @@ -25,7 +25,8 @@ durable, a reorg noticed only when a proof stops verifying. 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 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 e355dde..3b8a3af 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -84,8 +84,9 @@ pub struct DriverStatus { 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, @@ -162,6 +163,7 @@ where generation: u64, last_checkpoint_block: Option, last_snapshot_block: Option, + seen_resets: u64, durability_lost: bool, advanced: bool, } @@ -246,7 +248,7 @@ where fn around( engine: Engine, source: S, - sink: K, + mut sink: K, initial: F, driver_config: DriverConfig, ) -> Result { @@ -258,7 +260,12 @@ 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, @@ -272,9 +279,16 @@ where generation: 0, last_checkpoint_block: None, last_snapshot_block: None, + seen_resets, durability_lost: false, advanced: false, - }) + }; + // 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. @@ -354,13 +368,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 +384,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(), @@ -403,13 +417,17 @@ where 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(); + } 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. @@ -540,7 +558,11 @@ where engine: self.engine.status(), 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, } @@ -918,6 +940,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 @@ -1218,6 +1272,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 @@ -1387,6 +1496,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 diff --git a/crates/chainfold/src/engine.rs b/crates/chainfold/src/engine.rs index f5b19fe..e8ce6bf 100644 --- a/crates/chainfold/src/engine.rs +++ b/crates/chainfold/src/engine.rs @@ -96,6 +96,7 @@ pub struct Engine { status: EngineStatus, last_verified: Option, skips: u64, + resets: u64, } impl Engine { @@ -110,6 +111,7 @@ impl Engine { status: EngineStatus::Active, last_verified: None, skips: 0, + resets: 0, }) } @@ -133,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) @@ -259,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. @@ -287,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) } @@ -300,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. @@ -1299,6 +1306,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/harness.rs b/crates/chainfold/src/harness.rs index 88e6f2d..99687b9 100644 --- a/crates/chainfold/src/harness.rs +++ b/crates/chainfold/src/harness.rs @@ -145,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 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/storage/flusher.rs b/crates/chainfold/src/storage/flusher.rs index a0d0887..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))), } } @@ -176,6 +177,7 @@ impl TokenState { struct Job { snapshot: Vec, cursor: Option, + epoch: u64, token: Arc, shared: Arc, } @@ -202,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(())); } @@ -271,6 +275,7 @@ 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), }; @@ -280,10 +285,10 @@ impl Flusher { 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. @@ -313,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)] @@ -336,6 +347,10 @@ mod tests { }; use crate::{ batch::Batch, + driver::{ + Driver, + DriverConfig, + }, engine::{ Engine, EngineConfig, @@ -352,6 +367,7 @@ mod tests { test_util::{ CrashVfs, RecordingFold, + ScriptedChain, }, }; @@ -667,4 +683,85 @@ mod tests { // 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 4e2dd6d..efd36c3 100644 --- a/crates/chainfold/src/test_util.rs +++ b/crates/chainfold/src/test_util.rs @@ -134,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, @@ -155,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. From 5bd2bcec0b80b26fd220c0d65df3486ed887cc32 Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:02:25 +0530 Subject: [PATCH 4/8] fix(chainfold)!: fold only what one chain vouches for The scan read the cursor block's header before the head and the events, so a reorg landing between the reads let new-chain events fold on an old-chain boundary. It also folded whatever it was handed: events a reorg skipped between two windows, a torn answer, entries outside the window, two hashes at the cursor block, and the rest of a block the cursor stopped inside were folded wrongly or lost without a signal. A head below the cursor resynced the whole fold, and a head at u64::MAX never ended the walk. `Source::head` now returns the head block's header. The scan reads it first and vouches for everything after it: - the walk starts on the cursor block, so the answer carries the boundary header; two hashes there, or any entry outside its window, refuse the answer - entries at or before the cursor are dropped, so the rest of a partly folded cursor block is folded - a batch with events re-reads the head block and is dropped when it moved - a head below the cursor on the chain the ring observed is Idle, not a fork - `group_into` splits spans on any block change, so two hashes for one height are a source error - the walk stops at u64::MAX The dead `scanned_to` mark is gone. BREAKING: `Source::head` returns `BlockRef`, not `u64`. --- crates/chainfold/README.md | 6 +- crates/chainfold/benches/recover.rs | 2 +- crates/chainfold/src/driver.rs | 553 ++++++++++++++++++-- crates/chainfold/src/source.rs | 4 +- crates/chainfold/src/test_util.rs | 17 +- crates/chainfold/tests/storage_crash.rs | 4 +- crates/examples/examples/shielded_wallet.rs | 17 +- 7 files changed, 535 insertions(+), 68 deletions(-) diff --git a/crates/chainfold/README.md b/crates/chainfold/README.md index 696aaf7..e1370ff 100644 --- a/crates/chainfold/README.md +++ b/crates/chainfold/README.md @@ -60,9 +60,9 @@ 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. 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/driver.rs b/crates/chainfold/src/driver.rs index 3b8a3af..a550948 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -71,6 +71,16 @@ pub enum Tick { Terminal(EngineStatus), } +/// What a scan leaves for the tick to act on. +enum Scan { + /// `batch` is ready to apply. + Ready, + /// The source's head trails the cursor on the chain the ring observed; 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 { @@ -156,7 +166,6 @@ where config: DriverConfig, batch: Batch, scratch: Vec<(BlockRef, u32, F::Event)>, - scanned_to: Option, initial: F, consecutive_errors: u32, caught_up: bool, @@ -272,7 +281,6 @@ where config: driver_config, batch: Batch::new(), scratch: Vec::new(), - scanned_to: None, initial, consecutive_errors: 0, caught_up: false, @@ -416,7 +424,6 @@ where self.engine.reset(self.initial.clone()); self.caught_up = false; self.consecutive_errors = 0; - self.scanned_to = None; Tick::Resynced } @@ -444,9 +451,18 @@ 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; + match self.scan() { + Ok(Scan::Ready) => {} + 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; match self.engine.apply_batch(&self.batch) { @@ -483,38 +499,79 @@ 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. + 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 node behind the cursor on the chain the ring observed is lagging, not forked + if let Some(cursor) = cursor + && head.number < cursor.block + && self + .engine + .observed() + .all(|seen| seen.number != head.number || seen.hash == head.hash) + { + 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); + while from <= head.number { + let to = head.number.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 head.number == cursor.block { + Some(head) + } else { + self.source.header_at(cursor.block)? + }; } - Ok(()) + // 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(Scan::Ready) } /// Bisects the observed ring for the deepest still-canonical block, then rolls back. @@ -596,7 +653,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); } @@ -657,7 +714,7 @@ mod tests { type Event = u64; type Error = PollFailure; - fn head(&mut self) -> Result { + fn head(&mut self) -> Result { self.inner.head() } @@ -689,7 +746,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, } @@ -698,9 +755,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> { @@ -718,6 +774,86 @@ 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() + } + } + fn engine_config(checkpoint_slots: usize) -> EngineConfig { EngineConfig { ring_capacity: 8, @@ -725,6 +861,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, @@ -1085,7 +1241,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 { @@ -1104,8 +1260,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() @@ -1117,7 +1272,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, @@ -1130,7 +1285,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(); @@ -1145,14 +1300,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)); } @@ -1681,7 +1829,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 @@ -1821,4 +1969,315 @@ 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); + } } 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/test_util.rs b/crates/chainfold/src/test_util.rs index efd36c3..425c768 100644 --- a/crates/chainfold/src/test_util.rs +++ b/crates/chainfold/src/test_util.rs @@ -305,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> { @@ -748,11 +749,11 @@ mod tests { // 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] @@ -781,6 +782,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/storage_crash.rs b/crates/chainfold/tests/storage_crash.rs index dec364f..3ca2474 100644 --- a/crates/chainfold/tests/storage_crash.rs +++ b/crates/chainfold/tests/storage_crash.rs @@ -201,8 +201,8 @@ fn flusher_sink_persists_and_resumes() { } // then the durable cursor trails the tip, the resume starts there, and the view converges - assert_eq!(durable_cursor, Some(Position::new(8, 0))); - assert_eq!(resume_cursor, Some(Position::new(8, 0))); + assert_eq!(durable_cursor, Some(Position::new(7, 0))); + assert_eq!(resume_cursor, Some(Position::new(7, 0))); assert_eq!(resumed.engine().cursor(), Some(Position::new(10, 0))); let expected: Vec<(Position, u64)> = (1..=10u64) .map(|value| (Position::new(value, 0), value)) 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> { From be79d9ef7b2d1f1310045841b6ebb72954cee4a4 Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Fri, 9 Oct 2026 22:35:06 +0530 Subject: [PATCH 5/8] fix(chainfold)!: correct the tuning docs and two test helpers - the README said a fork deeper than the ring is ForkBeyondWindow and that with zero slots Halt and Poison escalate to a resync; such a fork resyncs, and Halt and Poison are terminal for every slot count - DivergenceCause::ForkBeyondWindow was never built, so it is gone - the apply_alloc counters were process wide, so a measurement could count the test harness's own threads and fail under cargo test; they count per thread now - ScriptedChain::events_in left out the block numbered u64::MAX from a range that ends there BREAKING: DivergenceCause::ForkBeyondWindow is removed. --- crates/chainfold/README.md | 10 +++---- crates/chainfold/src/engine.rs | 17 +++++------ crates/chainfold/src/error.rs | 3 -- crates/chainfold/src/test_util.rs | 22 +++++++++++++- crates/chainfold/tests/apply_alloc.rs | 43 +++++++++++++-------------- 5 files changed, 52 insertions(+), 43 deletions(-) diff --git a/crates/chainfold/README.md b/crates/chainfold/README.md index e1370ff..30bfafa 100644 --- a/crates/chainfold/README.md +++ b/crates/chainfold/README.md @@ -66,8 +66,7 @@ and recovery), and the adapters (`std`: tokio harness, snapshot store, flusher). - 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, @@ -183,11 +182,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/src/engine.rs b/crates/chainfold/src/engine.rs index e8ce6bf..f0a79a6 100644 --- a/crates/chainfold/src/engine.rs +++ b/crates/chainfold/src/engine.rs @@ -1240,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])])); @@ -1249,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] diff --git a/crates/chainfold/src/error.rs b/crates/chainfold/src/error.rs index cc3675c..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, diff --git a/crates/chainfold/src/test_util.rs b/crates/chainfold/src/test_util.rs index 425c768..f153a9a 100644 --- a/crates/chainfold/src/test_util.rs +++ b/crates/chainfold/src/test_util.rs @@ -331,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, @@ -744,6 +750,20 @@ 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 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 { From 8e5006d3118476afaf5f639517d883e8a4b23327 Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Fri, 9 Oct 2026 22:35:19 +0530 Subject: [PATCH 6/8] fix(chainfold): cap a tick's walk and keep what a long empty walk found A walk that found nothing ran to the head in one tick, however long, and any failed call dropped everything it had read. A first sync over a long empty stretch never finished under a rate limit or a few failed calls. - a tick reads at most `span` windows. The span starts at 64, halves after a source error and doubles after any other tick - a walk that stops at the cap with nothing to fold remembers the block it ended on, read between two reads of the head block so it lies on the chain the head names. The next tick checks that block against the head and goes on above it, with the ring's cursor block as the boundary. A mark off the head's chain is dropped, and the walk starts at the cursor in the same tick - the mark is set only once the tick's boundary check passes, and it never moves to the head. Folding a batch, a fork, a rollback, a reset and `source_mut` clear it - a head below the mark is Idle, as a head below the cursor is, unless it contradicts the ring - a capped tick is `Progressed` with a zero summary, is not caught up, and asks for no delay - a head that a folding or capped tick read before and after its walk, once its boundary check passed, names the same chain at the next tick. That tick takes the engine's own cursor block as the boundary and walks above it, and above the mark without checking it, so a catch-up at a small window reads no more than before walks started on the cursor block - `engine_mut` clears nothing. The driver records the engine's cursor and reset count at the end of each tick that leaves it active; a different pair at the next tick means the caller moved the engine, which drops the mark and the remembered head and reads as not caught up. A caller that only reaches the fold, as the merkle_log example does to prune it, keeps all of it A walk of up to 64 windows, 32,000 blocks at the default window, costs what it did. --- crates/chainfold/README.md | 7 +- crates/chainfold/src/driver.rs | 1165 ++++++++++++++++++++++- crates/chainfold/tests/storage_crash.rs | 4 +- 3 files changed, 1145 insertions(+), 31 deletions(-) diff --git a/crates/chainfold/README.md b/crates/chainfold/README.md index 30bfafa..ee31374 100644 --- a/crates/chainfold/README.md +++ b/crates/chainfold/README.md @@ -17,9 +17,10 @@ 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. diff --git a/crates/chainfold/src/driver.rs b/crates/chainfold/src/driver.rs index a550948..5f07d73 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, @@ -73,9 +77,13 @@ pub enum Tick { /// What a scan leaves for the tick to act on. enum Scan { - /// `batch` is ready to apply. - Ready, - /// The source's head trails the cursor on the chain the ring observed; nothing to do. + /// `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, on the chain the ring + /// observed; nothing to do. Lagging, /// The source contradicted itself; its answer is dropped. Refused, @@ -90,7 +98,7 @@ 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, @@ -175,6 +183,20 @@ where 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 @@ -290,6 +312,10 @@ where 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. @@ -320,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.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. @@ -435,13 +473,25 @@ where 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 } @@ -451,8 +501,13 @@ where if !self.engine.status().is_active() { return Tick::Terminal(self.engine.status()); } - match self.scan() { - Ok(Scan::Ready) => {} + 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; @@ -463,17 +518,27 @@ where 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) @@ -500,14 +565,21 @@ where } /// Fills `batch` with the events after the cursor, plus the cursor block's header, all - /// as the chain `head` names reports them. + /// 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 head = self.source.head()?; let cursor = self.engine.cursor(); - // a node behind the cursor on the chain the ring observed is lagging, not forked - if let Some(cursor) = cursor - && head.number < cursor.block + // a node behind the mark, or behind the cursor without one, on the chain the ring + // observed is lagging, not forked + if let Some(floor) = self + .walked + .map(|walked| walked.number) + .or(cursor.map(|cursor| cursor.block)) + && head.number < floor && self .engine .observed() @@ -518,8 +590,31 @@ where let window = self.source.window().max(1); let mut from = cursor.map_or(self.config.start_block, |cursor| cursor.block); - while from <= head.number { - let to = head.number.min(from.saturating_add(window - 1)); + // 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)?; if self @@ -559,24 +654,40 @@ where if let Some(cursor) = cursor && self.batch.boundary.is_none() { - self.batch.boundary = if head.number == cursor.block { + 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(Scan::Ready) + 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(); @@ -675,6 +786,10 @@ mod tests { vec, vec::Vec, }; + use core::cell::{ + Cell, + RefCell, + }; #[cfg(feature = "std")] use std::{ vec, @@ -854,6 +969,127 @@ mod tests { } } + /// 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, @@ -902,6 +1138,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, @@ -1536,6 +1787,9 @@ mod tests { ); 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")] @@ -1568,6 +1822,81 @@ mod tests { assert!(!driver.status().caught_up); } + #[test] + 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..=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 @@ -2280,4 +2609,788 @@ mod tests { ); 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 stops, and nothing it holds moves + assert_eq!((tick, read), (Tick::Idle, 1)); + 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 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); + } + + #[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/tests/storage_crash.rs b/crates/chainfold/tests/storage_crash.rs index 3ca2474..dec364f 100644 --- a/crates/chainfold/tests/storage_crash.rs +++ b/crates/chainfold/tests/storage_crash.rs @@ -201,8 +201,8 @@ fn flusher_sink_persists_and_resumes() { } // then the durable cursor trails the tip, the resume starts there, and the view converges - assert_eq!(durable_cursor, Some(Position::new(7, 0))); - assert_eq!(resume_cursor, Some(Position::new(7, 0))); + assert_eq!(durable_cursor, Some(Position::new(8, 0))); + assert_eq!(resume_cursor, Some(Position::new(8, 0))); assert_eq!(resumed.engine().cursor(), Some(Position::new(10, 0))); let expected: Vec<(Position, u64)> = (1..=10u64) .map(|value| (Position::new(value, 0), value)) From 9cb78f8edb33cb1f3b43f94345e3afa66af2ca55 Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Tue, 13 Oct 2026 15:29:58 +0530 Subject: [PATCH 7/8] fix: lagging behaviour --- crates/chainfold/src/driver.rs | 38 +++++++--- crates/chainfold/tests/lagging.rs | 111 ++++++++++++++++++++++++++++++ 2 files changed, 139 insertions(+), 10 deletions(-) create mode 100644 crates/chainfold/tests/lagging.rs diff --git a/crates/chainfold/src/driver.rs b/crates/chainfold/src/driver.rs index 5f07d73..9e90ef2 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -82,8 +82,8 @@ enum Scan { /// 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, on the chain the ring - /// observed; nothing to do. + /// 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, @@ -573,19 +573,32 @@ where self.batch.clear(); let head = self.source.head()?; let cursor = self.engine.cursor(); - // a node behind the mark, or behind the cursor without one, on the chain the ring - // observed is lagging, not forked + // 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 - && self + { + let lagging = match self .engine .observed() - .all(|seen| seen.number != head.number || seen.hash == head.hash) - { - return Ok(Scan::Lagging); + .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); @@ -2895,8 +2908,9 @@ mod tests { 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 stops, and nothing it holds moves - assert_eq!((tick, read), (Tick::Idle, 1)); + // 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))); @@ -2958,6 +2972,10 @@ mod tests { 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] 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]); +} From 16fd89c9996e0f81696b31a42c452683244ada1e Mon Sep 17 00:00:00 2001 From: rymnc <43716372+rymnc@users.noreply.github.com> Date: Tue, 13 Oct 2026 15:45:33 +0530 Subject: [PATCH 8/8] fix: add one more check for edge case --- crates/chainfold/src/driver.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/chainfold/src/driver.rs b/crates/chainfold/src/driver.rs index 9e90ef2..eb4ca6b 100644 --- a/crates/chainfold/src/driver.rs +++ b/crates/chainfold/src/driver.rs @@ -2916,6 +2916,12 @@ mod tests { 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;