From 013504971001b8c8eb971075ce2e9e1021c405b3 Mon Sep 17 00:00:00 2001 From: erhant Date: Thu, 16 Jul 2026 18:11:44 +0300 Subject: [PATCH] fix(chain_state, sequencer, storage): rebuild the two-tier state on restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - persist a final-tier snapshot (state + meta) atomically with the follow update that advanced it; on boot, anchor `final` on it and replay stored blocks above it as `head`, so a post-restart orphan of a not-yet-finalized block still reverts instead of silently diverging - correlate orphan/finalize events by block hash: restored head entries carry hash-derived sentinel MsgIds (the real ones are not persisted) - a finalized block chaining on an unfinalized head entry finalizes that prefix (finality is prefix-monotone) - warn on (and keep ignoring) a same-height adopted competitor that arrives without its orphan sibling — an SDK-contract breach worth surfacing - follow path reverts orphans via one batched `apply_channel_update` (single truncate + head re-derivation) instead of a re-derivation per orphan - remove DESIGN.md, move to task docs instead - trim very long comments / docstrings --- Cargo.lock | 1 + lez/chain_state/Cargo.toml | 1 + lez/chain_state/DESIGN.md | 308 ------------------- lez/chain_state/src/chain.rs | 228 +++++++++++--- lez/chain_state/src/lib.rs | 2 +- lez/indexer/core/src/block_store.rs | 5 +- lez/sequencer/core/src/lib.rs | 206 +++++++++---- lez/sequencer/core/src/tests.rs | 110 +++++++ lez/storage/src/sequencer/mod.rs | 65 ++-- lez/storage/src/sequencer/sequencer_cells.rs | 73 ++++- lez/storage/src/sequencer/tests.rs | 33 ++ 11 files changed, 592 insertions(+), 440 deletions(-) delete mode 100644 lez/chain_state/DESIGN.md diff --git a/Cargo.lock b/Cargo.lock index 06a3e08e..792f3856 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1376,6 +1376,7 @@ dependencies = [ "borsh", "common", "lee", + "log", "logos-blockchain-core", "logos-blockchain-zone-sdk", "serde", diff --git a/lez/chain_state/Cargo.toml b/lez/chain_state/Cargo.toml index ca9876ab..f788f55a 100644 --- a/lez/chain_state/Cargo.toml +++ b/lez/chain_state/Cargo.toml @@ -14,6 +14,7 @@ logos-blockchain-core.workspace = true logos-blockchain-zone-sdk.workspace = true anyhow.workspace = true +log.workspace = true serde.workspace = true thiserror.workspace = true diff --git a/lez/chain_state/DESIGN.md b/lez/chain_state/DESIGN.md deleted file mode 100644 index cb8595aa..00000000 --- a/lez/chain_state/DESIGN.md +++ /dev/null @@ -1,308 +0,0 @@ -# `lez/chain_state` — Two-Tier Chain State - -Design doc for the shared block-apply engine and two-tier chain state that -backs decentralized sequencing. Status: **interface freeze** — the -`apply_block` signature and the `ChainState` tip/state shape below are the -contract the produce-on-turn and follow-blocks tracks build against. Changing -them after the tracks split forces rework in both. - -## 1. Motivation - -Decentralized sequencing requires every honest node — sequencer or indexer — to -converge on the same chain and the same state by running one deterministic -_validate-then-apply_ path over blocks pulled from the channel. That path today -lives only inside the indexer (`lez/indexer/core/src/block_store.rs`), where the -recoverability work built a park-and-skip ingest: `accept_block` validates a -block against the current tip, applies it to a scratch clone of state atomically, -and on any failure records a `StallReason`, freezes the tip, and marks the bad -block _processed_ without applying it. The sequencer has no equivalent — it only -produces blocks and reads peer inscriptions for finalization; it never executes -peer blocks into its own state. - -This crate lifts that logic into a shared home and generalizes it into a -**two-tier** state machine so the sequencer can produce on the head while both -sequencer and indexer expose their exact current state. - -## 2. Crate placement & layering - -``` -lee_core ← lee (owns V03State) ← common (owns Block, BedrockStatus, clock_invocation, recompute_hash) ← lez/chain_state ← { indexer/core, sequencer/core } -``` - -- **Not** `lee`: the apply logic needs `Block`/`BedrockStatus`/`clock_invocation` - from `common`, and `common` depends on `lee` — putting it in `lee` inverts the - layer. -- `lez/chain_state` sits above `common`, depends on `common` + `lee`, and is - consumed by both `indexer/core` and `sequencer/core`. - -**Persistence boundary.** `chain_state` holds the in-memory state machine and the -pure logic; it performs **no I/O**. Each consumer keeps its own `RocksDBIO` and -drives the `scratch → put_block → commit` ordering, exactly as `accept_block` does -today. This keeps the crate fully unit-testable without a DB. - -## 3. The `apply_block` entry point - -A single pure function, called identically whether the block was produced by us, -adopted from a peer, or read finalized from the channel: - -```rust -/// Validate `block` against `tip`, then apply it to `state`. Pure: no I/O. -/// Mutates `state` only on success; on failure `state` is untouched and the -/// caller parks. -fn apply_block( - tip: Option<&Tip>, - block: &Block, - state: &mut V03State, -) -> Result<(), BlockIngestError>; -``` - -Validation order (unchanged from the indexer): hash integrity -(`recompute_hash`) → block-id continuity → `prev_block_hash` linkage, with a -`None` tip expecting the genesis block. Application splits off the mandatory -trailing clock tx, executes user txs (genesis = public-only), applies the clock -last. - -Shared types moved into the crate: `AcceptOutcome`, `BlockIngestError`, -`StallReason`, and `Tip`. - -```rust -struct Tip { block_id: u64, hash: HashType } - -enum AcceptOutcome { - Applied, - AlreadyApplied, - Parked(BlockIngestError), - RetryableFailure(BlockIngestError), -} -``` - -`Tip` will additionally carry `l1_slot` once the anchor layer lands (recorded -atomically with the tip), because the anchor / chain-consistency logic keys on -the inscription slot, not just `(id, hash)`. - -## 4. The two-tier `ChainState` - -```rust -struct ChainState { - final_state: V03State, // driven by finalized channel ops - final_tip: Option, - head_state: V03State, // final_state + applied head blocks - head_blocks: Vec, // ordered, above final_tip - final_stall: Option, // the one stall — persisted to RocksDB. See §4a -} - -struct HeadEntry { this_msg: MsgId, block: Block } -``` - -The **head** tier is a MsgId-keyed chain (adopted/orphaned reference -`this_msg`/`parent_msg`); the **final** tier is block-id-keyed. `apply_block` -validation stays LEZ-level (`block_id` + `prev_block_hash`) — the two chains run -in parallel and must agree, so we validate via `apply_block` _and_ track `MsgId` -for revert correlation. - -Operations: - -- `apply_adopted(inscription) -> AcceptOutcome` — dedup by `this_msg` against our - outbox, else `apply_block` on the head tip; on success push `HeadEntry`. On - failure, **do nothing durable**: the head tip simply stays at the last valid - block. No stall is recorded (see §4a) — the head self-heals from the stream. -- `apply_channel_update(orphaned, adopted)` — revert every `orphaned` by - `this_msg`, re-derive `head_state` (clone `final_state`, replay survivors), - then apply every `adopted` in order. Atomic per event. -- `finalize_up_to(block_id)` — move `head_blocks` up to `block_id` into - `final_state` (already validated; a move, not a re-apply). -- `apply_finalized(inscription)` — steady state: if present in head by - `this_msg`, `finalize_up_to`; cold-start backfill (not in head): - `apply_block` directly to `final_state`, mirror into head. If a finalized block - fails to apply, set **`final_stall`** and persist it — this is the **only** stall - (see §4a). -- `rollback_orphan(this_msg)` — drop from that entry forward, re-derive head. -- `status() -> { final_height, head_height, final_stall }` — for RPC/UI. A derived - "head blocked" indicator can be computed on demand (see §4a) without persisting. - -For the **indexer** (finalized-only `next_messages` stream), `head_blocks` stays -empty and `head == final`; it exercises only `apply_finalized`. The **sequencer** -uses both tiers from day one. - -### 4a. One stall — `final_stall`, persisted - -There is a single stall, `final_stall`, on the final tier. The head tier does -**not** carry its own stall, and this is deliberate. - -The head and the final tier never represent two independent problems: a block -always reaches the head first (as `adopted`) and only later the final tier (as -`finalized`), so a would-be "head stall" is just the earlier, provisional sighting -of the exact block that `final_stall` records durably if it finalizes — the same -event modeled twice. - -And the head does not need a recorded reason to freeze. **The tip-freeze is -intrinsic**: not applying a bad block is what freezes the tip; no marker is -required. The head's freeze is also **transient and self-healing** — the bad block -is either orphaned (a competing valid block at the same height wins and applies on -its own) or it finalizes. Subsequent adopted blocks that chain on the bad one fail -validation by themselves, and the producer builds on the head tip regardless of -any marker. So a persisted head stall would be redundant (re-derived from the -stream on restart) and buys no behavior. - -`final_stall` is the stall that does real work: - -- The **indexer already requires it** and ships it today. The indexer has only a - final tier (finalized-only stream, no head); its startup chain-consistency / - anchor check reads the persisted stall to know where it is parked. The shared - `final_stall` serves that unchanged. -- It **survives restart** and is what we surface as `Stalled`. -- "A bad block **finalized**" is the only irreversible, actionable condition — the - signal the committee acts on to evict a bad sequencer. A provisional head block - that may vanish on the next reorg is not something to evict over. - -The one thing we forgo is an *early warning* that a sequencer is posting garbage -before it finalizes. That condition frequently self-heals via reorg, so alarming on -it is mostly noise; if wanted, it is a **derived, non-persisted** indicator (e.g. -"k adopted inscriptions above the head tip remain unapplied"), computed on demand — -not a second `StallReason` in the struct. - -So the sequencer and the indexer share exactly one stall concept, keeping the two -consumers uniform. - -### 4b. Producer contract — write on turn, build on last valid - -The sequencer publishes **only on its own turn** (the SDK queues out-of-turn -publishes). When it is our turn we build the next block on the **current head -tip**, which is by construction the last validly-applied block. So if the head is -frozen on a peer's bad block, we build on that frozen valid -tip — the same parent every honest sequencer chooses — and thereby skip the bad -block rather than extend it. A parked node keeps following peers' valid blocks as -they arrive; the moment it also gets a turn, it produces the next valid block on -its last valid tip. Net: parking never stops us from producing correctly on our -turn. - -## 5. Event → tier mapping - -`Event::BlocksProcessed { checkpoint, channel_update: { orphaned, adopted }, finalized }`: - -| Input | Source | Effect | -| --------------------- | ---------------------------------------------- | --------------------------------------------------- | -| adopted inscription | `channel_update.adopted` | validate + apply to **head** | -| orphaned inscription | `channel_update.orphaned` | revert from **head**, return txs to mempool | -| finalized inscription | `finalized[].ops` (`FinalizedOp::Inscription`) | move head→**final**, or apply directly on backfill | -| own publish | publish-return | optimistically apply to **head**, record `this_msg` | - -**Golden rules:** (1) validation is deterministic, so every honest node makes the -same accept/park decision. (2) An invalid block is _processed but discarded_ — -never applied, never halts the node. (3) Finalized is never reverted. (4) We -rebuild orphaned blocks ourselves; we do not trust the SDK's republish (it keeps -stale LEZ contents — prev-hash, tx selection, and resulting state were all -computed against the old parent). - ---- - -## 6. Scenarios - -### Processing one `BlocksProcessed` event - -```mermaid -flowchart TD - EV["Event::BlocksProcessed"] --> ORPH{"orphaned
non-empty?"} - ORPH -->|yes| REV["For each orphaned by this_msg:
drop from head_blocks,
return its txs to mempool"] - REV --> RED["Re-derive head_state:
clone final_state, replay survivors"] - ORPH -->|no| ADO - RED --> ADO{"adopted
non-empty?"} - ADO -->|"yes, in order"| DEDUP{"this_msg in
our outbox?"} - ADO -->|no| FIN - DEDUP -->|"yes (our own)"| SKIP["skip: already applied optimistically"] - DEDUP -->|no| VAL["apply_block on head tip"] - VAL --> OUT{"AcceptOutcome"} - OUT -->|Applied| APP["append this_msg+block to head,
advance head tip, clear stall"] - OUT -->|AlreadyApplied| SKIP - OUT -->|"Parked(err)"| PARK["freeze head tip — do NOT apply.
No stall recorded (self-heals
via reorg/finalization)"] - SKIP --> FIN - APP --> FIN - PARK --> FIN - FIN{"finalized
inscriptions?"} - FIN -->|"already in head (steady state)"| MOVE["finalize_up_to:
move head→final, trim head_blocks"] - FIN -->|"not in head (cold-start backfill)"| DIRECT["apply_block directly to final"] - DIRECT --> DOK{"applied?"} - DOK -->|yes| MIRROR["mirror into head"] - DOK -->|"no (invalid finalized)"| FSTALL["record StallReason on FINAL,
freeze final tip"] - FIN -->|none| CP - MOVE --> CP - MIRROR --> CP - FSTALL --> CP - CP["persist checkpoint atomically"] -``` - -### Park / recovery status - -```mermaid -stateDiagram-v2 - [*] --> Syncing - Syncing --> CaughtUp: stream drained, no stall - CaughtUp --> Syncing: new adopted / finalized arrives - Syncing --> Parked: invalid block FINALIZED (apply_finalized fails) - Parked --> Parked: further non-chaining finalized blocks (orphans_since++) - Parked --> Syncing: valid successor finalizes on frozen final tip → stall cleared - note right of Parked - final_stall — the one stall. Persisted, survives restart. - Head-tier bad blocks do NOT enter this state: - the head tip freezes intrinsically (no stall) and - self-heals via reorg/finalization. Producer (on our - turn) builds on the last valid tip either way. - end note -``` - -### Scenario table - -**Normal flow** - -| # | Scenario | Handling | Expected | -| --- | ---------------------------------------- | ------------------------------------------------------------- | ----------------------------------- | -| 1 | Adopted block chains cleanly on head tip | `apply_block` → `Applied`; append `{this_msg, block}` to head | head advances; converges with peers | -| 2 | Our own block comes back in `adopted` | dedup by `this_msg` against outbox → skip | no double-apply | -| 3 | Adopted block later finalizes | `finalize_up_to` moves head→final, trims `head_blocks` | final advances; no re-apply | -| 4 | Re-delivery of an already-applied block | id ≤ tip & stored hash matches → `AlreadyApplied` | idempotent, no state change | - -**Reorg / orphan** - -| # | Scenario | Handling | Expected | -| --- | ---------------------------------------------------------- | -------------------------------------------------------------------------------------------- | ----------------------------------------------- | -| 5 | Our block orphaned at turn handoff (stale-parent race) | revert by `this_msg`, return txs to mempool, **rebuild** on new head tip (not SDK republish) | our txs re-queued; next block on correct parent | -| 6 | Batch reorg: some `orphaned` + some `adopted` in one event | revert all orphaned, re-derive head, then apply all adopted in order | deterministic convergence | -| 7 | Orphan chain (parent transitively off canonical) | SDK surfaces all affected as `orphaned`; revert each, replay survivors | head_state matches new canonical branch | - -**Invalid / bad block** — "stall" below means the one **persisted `final_stall`** -(§4a). A bad block seen only in `adopted` records **no** stall — the head tip just -freezes and self-heals; it becomes a `final_stall` only if it finalizes. - -| # | Scenario | Handling | Expected | -| --- | ------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ----------------------------------------------- | -| 8 | Authorized sequencer posts a block with an invalid state transition | head: `apply_block` → `Parked`, freeze head tip, no stall recorded. If it finalizes: persisted `final_stall` | park-and-skip; no apply, no halt | -| 9 | Broken chain link / hash mismatch / unexpected id in adopted | `Parked(BrokenChainLink / HashMismatch / UnexpectedBlockId)`; same park | frozen tip; peers park identically | -| 10 | Undeserializable inscription payload | park with `Deserialize` (no header); processing advances | recover when a valid block chains on frozen tip | -| 11 | Valid successor after a park (recovery) | block chaining on frozen tip → `Applied` → clear stall | head resumes automatically; no divergence | -| 12 | Further non-chaining blocks while parked | keep first `StallReason`, bump `orphans_since` | original cause preserved; still parked | - -**Producing while parked** - -| # | Scenario | Handling | Expected | -| --- | ----------------------------------------------- | ----------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | -| 13 | It's our turn but head is parked on a bad block | producer builds on the **frozen valid tip** (head tip = last valid), skipping the invalid block | we emit the next valid block on the same parent honest peers use — chain moves on our turn | - -**Startup / backfill** - -| # | Scenario | Handling | Expected | -| --- | --------------------------------------------------- | -------------------------------------------------------------------------------------------------- | ------------------------------------- | -| 14 | Cold start / reconnect backfill | history via `finalized`, empty `channel_update`; apply directly to final + mirror head | head == final until live deltas start | -| 15 | Local store belongs to a different chain (L1 reset) | anchor-based `chain_consistency` check at startup: wipe+reindex if `allow_chain_reset`, else error | no silent divergence | - -## 7. Invariants - -Should-never-happen conditions — assert/log, don't silently absorb: - -- An `orphaned` entry never references a block at or below the **final** tip — - finalized is irreversible. If seen, it is a bug. -- `head` tip ≥ `final` tip at all times; `head_blocks` holds exactly the blocks - between them. -- After processing any event, `head_state == final_state` replayed through - `head_blocks` (the re-derivation is the source of truth). -- A parked node's frozen tip is identical across all honest nodes for the same - invalid block (deterministic validation). diff --git a/lez/chain_state/src/chain.rs b/lez/chain_state/src/chain.rs index c3a810c0..35dc458b 100644 --- a/lez/chain_state/src/chain.rs +++ b/lez/chain_state/src/chain.rs @@ -1,8 +1,9 @@ //! Two-tier chain state: a reorg-able `head` the sequencer builds on, plus an -//! irreversible `final` tier. See `DESIGN.md` for the model and rationale. +//! irreversible `final` tier. -use common::block::Block; +use common::{HashType, block::Block}; use lee::V03State; +use log::warn; use logos_blockchain_core::mantle::ops::channel::MsgId; use logos_blockchain_zone_sdk::Slot; @@ -21,6 +22,9 @@ pub struct HeadEntry { /// (irreversible, from `finalized`). /// /// `head_state` is given by `final_state` replayed through `head_blocks`. +/// +/// Only the final tier stalls: an invalid `adopted` block just freezes the +/// head tip and self-heals via reorg or finalization. pub struct ChainState { final_state: V03State, final_tip: Option, @@ -86,18 +90,50 @@ impl ChainState { self.final_stall.as_ref() } + /// Position of a head entry, matched by `MsgId` or block hash (restored + /// entries carry sentinel `MsgId`s; re-inscriptions arrive under fresh ones). + fn head_position_of(&self, this_msg: MsgId, block_hash: HashType) -> Option { + self.head_blocks + .iter() + .position(|entry| entry.this_msg == this_msg || entry.block.header.hash == block_hash) + } + + /// Hash of the block we hold at `block_id` (final tip or head entry), if any. + fn hash_at(&self, block_id: u64) -> Option { + if let Some(tip) = &self.final_tip + && tip.block_id == block_id + { + return Some(tip.hash); + } + + self.head_blocks + .iter() + .find(|entry| entry.block.header.block_id == block_id) + .map(|entry| entry.block.header.hash) + } + /// Applies an adopted head block. On failure the head tip stays at the last - /// valid block and no stall is recorded (§4a); the head self-heals. + /// valid block and no stall is recorded; the head self-heals. pub fn apply_adopted(&mut self, this_msg: MsgId, block: &Block) -> AcceptOutcome { let tip = self.head_tip(); - if self - .head_blocks - .iter() - .any(|entry| entry.this_msg == this_msg) - || tip - .as_ref() - .is_some_and(|current| block.header.block_id <= current.block_id) + if self.head_position_of(this_msg, block.header.hash).is_some() { + return AcceptOutcome::AlreadyApplied; + } + if tip + .as_ref() + .is_some_and(|current| block.header.block_id <= current.block_id) { + // Re-delivery of a block we already hold; a different hash means an + // adopted competitor arrived without its orphan sibling — warn, keep ours. + if let Some(known) = self.hash_at(block.header.block_id) + && known != block.header.hash + { + warn!( + "Ignoring adopted block {} with hash {} — a different block ({}) \ + holds this height and no orphan event preceded it", + block.header.block_id, block.header.hash, known + ); + } return AcceptOutcome::AlreadyApplied; } @@ -116,30 +152,23 @@ impl ChainState { } /// Reverts an orphaned head block and everything after it, then re-derives head. - pub fn revert_orphan(&mut self, this_msg: MsgId) { - if let Some(idx) = self - .head_blocks - .iter() - .position(|entry| entry.this_msg == this_msg) - { + pub fn revert_orphan(&mut self, this_msg: MsgId, block: &Block) { + if let Some(idx) = self.head_position_of(this_msg, block.header.hash) { self.head_blocks.truncate(idx); self.rederive_head(); } } - /// One channel update: revert every `orphaned`, then apply every `adopted`. + /// One channel update: revert every `orphaned` (one truncate + re-derive), + /// then apply every `adopted` in order. Outcomes align with `adopted`. pub fn apply_channel_update( &mut self, - orphaned: &[MsgId], + orphaned: &[(MsgId, Block)], adopted: &[(MsgId, Block)], ) -> Vec { let earliest = orphaned .iter() - .filter_map(|msg| { - self.head_blocks - .iter() - .position(|entry| entry.this_msg == *msg) - }) + .filter_map(|(msg, block)| self.head_position_of(*msg, block.header.hash)) .min(); if let Some(idx) = earliest { self.head_blocks.truncate(idx); @@ -151,6 +180,20 @@ impl ChainState { .collect() } + /// Rebuilds one head entry from a persisted block, applying it in place (the + /// caller treats `Err` as fatal). + /// + /// Pass a hash-derived sentinel for `this_msg`; orphan/finalize events then correlate by block hash. + pub fn restore_head_block( + &mut self, + this_msg: MsgId, + block: Block, + ) -> Result<(), BlockIngestError> { + apply_block(self.head_tip().as_ref(), &block, &mut self.head_state)?; + self.head_blocks.push(HeadEntry { this_msg, block }); + Ok(()) + } + /// A finalized inscription. In steady state the block is already in head and is /// moved into `final`; on backfill (not in head) it is applied directly and may /// set `final_stall`. @@ -160,18 +203,22 @@ impl ChainState { block: &Block, l1_slot: Slot, ) -> AcceptOutcome { - // Match by [`MsgId`] or by block identity to handle a re-inscribed but - // identical block arriving under a fresh [`MsgId`]. - let in_head = self.head_blocks.iter().position(|entry| { - entry.this_msg == this_msg || entry.block.header.hash == block.header.hash - }); - match in_head { - Some(idx) => { - self.finalize_through(idx); - AcceptOutcome::Applied - } - None => self.apply_finalized_direct(block, l1_slot), + // Match by `MsgId` or block hash (re-inscriptions, restored entries). + if let Some(idx) = self.head_position_of(this_msg, block.header.hash) { + self.finalize_through(idx); + return AcceptOutcome::Applied; } + + // Finality is prefix-monotone: a finalized block chaining on an + // unfinalized head entry finalizes that prefix too. + if let Some(idx) = self + .head_blocks + .iter() + .position(|entry| entry.block.header.hash == block.header.prev_block_hash) + { + self.finalize_through(idx); + } + self.apply_finalized_direct(block, l1_slot) } /// Moves `head_blocks[0..=idx]` into the final tier (already validated in head). @@ -188,12 +235,9 @@ impl ChainState { /// Applies a finalized block straight to the final tier. On success the /// finalized chain is authoritative, so head rebases onto it. fn apply_finalized_direct(&mut self, block: &Block, l1_slot: Slot) -> AcceptOutcome { - // A finalized block at or below the final tip is a re-delivery — e.g. - // the finalization of blocks restored from the store at boot, which - // are the final tier from the start and so never sit in `head_blocks`. - // Idempotent, no stall. A *different* block at the tip height falls - // through to validation and parks: finalized is irreversible, so a - // conflicting finalized block at final height is a genuine stall. + // A finalized block at or below the final tip is a re-delivery: + // idempotent. A *different* block at the tip height falls through + // to validation and parks. if let Some(tip) = &self.final_tip && (block.header.block_id < tip.block_id || (block.header.block_id == tip.block_id && block.header.hash == tip.hash)) @@ -359,7 +403,7 @@ mod tests { chain.apply_adopted(msg(2), &block2); chain.apply_adopted(msg(3), &block3); - chain.revert_orphan(msg(3)); + chain.revert_orphan(msg(3), &block3); assert_eq!(chain.head_tip().expect("head tip").block_id, 2); // A competing block 3 now applies cleanly on block 2. @@ -382,7 +426,7 @@ mod tests { chain.apply_adopted(msg(3), &block3); let block3_prime = produce_dummy_block(3, Some(block2.header.hash), vec![]); - let outcomes = chain.apply_channel_update(&[msg(3)], &[(msg(13), block3_prime)]); + let outcomes = chain.apply_channel_update(&[(msg(3), block3)], &[(msg(13), block3_prime)]); assert!(matches!(outcomes.as_slice(), [AcceptOutcome::Applied])); assert_eq!(chain.head_tip().expect("head tip").block_id, 3); } @@ -458,7 +502,7 @@ mod tests { chain.apply_adopted(msg(4), &block4); // Orphaning block 3 drops the whole suffix (3 and 4). - chain.revert_orphan(msg(3)); + chain.revert_orphan(msg(3), &block3); assert_eq!(chain.head_tip().expect("head tip").block_id, 2); assert_eq!(chain.head_state().get_account_by_id(from).balance, 9990); @@ -494,7 +538,7 @@ mod tests { let block4_prime = produce_dummy_block(4, Some(block3_prime.header.hash), vec![tx4_prime]); let outcomes = chain.apply_channel_update( - &[msg(4), msg(3)], + &[(msg(4), block4), (msg(3), block3)], &[(msg(13), block3_prime), (msg(14), block4_prime)], ); @@ -517,13 +561,84 @@ mod tests { chain.apply_adopted(msg(2), &block2); let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); - let outcomes = chain.apply_channel_update(&[msg(99)], &[(msg(3), block3)]); + let unknown = produce_dummy_block(9, Some(HashType([7; 32])), vec![]); + let outcomes = chain.apply_channel_update(&[(msg(99), unknown)], &[(msg(3), block3)]); assert!(matches!(outcomes.as_slice(), [AcceptOutcome::Applied])); assert_eq!(chain.head_tip().expect("head tip").block_id, 3); assert_head_matches_replay(&chain); } + #[test] + fn same_height_adopted_competitor_without_orphan_is_ignored() { + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + chain.apply_adopted(msg(1), &genesis); + chain.apply_adopted(msg(2), &block2); + + // Same height, different hash, no preceding orphan event: an SDK + // contract breach. We keep the block we already applied (and warn). + let block2_prime = produce_dummy_block(2, Some(HashType([9; 32])), vec![]); + assert!(matches!( + chain.apply_adopted(msg(12), &block2_prime), + AcceptOutcome::AlreadyApplied + )); + assert_eq!(chain.head_tip().expect("head tip").hash, block2.header.hash); + assert_head_matches_replay(&chain); + } + + #[test] + fn restore_head_block_rebuilds_head_and_correlates_by_hash() { + let accounts = initial_pub_accounts_private_keys(); + let from = accounts[0].account_id; + let to = accounts[1].account_id; + let sign_key = accounts[0].pub_sign_key.clone(); + + // Restart shape: final tier from a persisted snapshot, head rebuilt from + // stored blocks under hash-derived sentinel MsgIds. + let mut state = initial_state(); + let genesis = produce_dummy_block(1, None, vec![]); + apply_block(None, &genesis, &mut state).expect("genesis applies"); + let mut chain = ChainState::from_final(state, Some(tip_of(&genesis))); + + let tx2 = create_transaction_native_token_transfer(from, 0, to, 10, &sign_key); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx2]); + let tx3 = create_transaction_native_token_transfer(from, 1, to, 10, &sign_key); + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![tx3]); + for block in [&block2, &block3] { + chain + .restore_head_block(MsgId::from(block.header.hash.0), block.clone()) + .expect("stored blocks must replay"); + } + assert_eq!(chain.head_tip().expect("head tip").block_id, 3); + assert_head_matches_replay(&chain); + + // The L1 orphans restored block 3 under its real (unknown-to-us) MsgId: + // correlated by hash, the revert works and a competitor applies. + chain.revert_orphan(msg(33), &block3); + assert_eq!(chain.head_tip().expect("head tip").block_id, 2); + + let block3_prime = produce_dummy_block(3, Some(block2.header.hash), vec![]); + assert!(matches!( + chain.apply_adopted(msg(13), &block3_prime), + AcceptOutcome::Applied + )); + assert_eq!(chain.head_state().get_account_by_id(to).balance, 20010); + assert_head_matches_replay(&chain); + } + + #[test] + fn restore_head_block_rejects_non_chaining_block() { + let mut chain = ChainState::new(initial_state()); + let skipped = produce_dummy_block(3, Some(HashType([9; 32])), vec![]); + assert!( + chain + .restore_head_block(MsgId::from(skipped.header.hash.0), skipped) + .is_err() + ); + } + #[test] fn finalized_reinscription_matches_by_block_hash() { let mut chain = ChainState::new(initial_state()); @@ -637,6 +752,29 @@ mod tests { assert_head_matches_replay(&chain); } + #[test] + fn finalized_successor_of_head_entry_finalizes_the_prefix() { + // Head holds unfinalized blocks 1..=2 (e.g. restored after a restart); + // a peer block 3 we never saw adopted arrives finalized. Its ancestry + // finalizes our prefix implicitly, then 3 applies to final directly. + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + chain.apply_adopted(msg(1), &genesis); + chain.apply_adopted(msg(2), &block2); + assert!(chain.final_tip().is_none()); + + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + assert!(matches!( + chain.apply_finalized(msg(3), &block3, slot(10)), + AcceptOutcome::Applied + )); + assert_eq!(chain.final_tip().expect("final tip").block_id, 3); + assert_eq!(chain.head_tip().expect("head tip").block_id, 3); + assert!(chain.final_stall().is_none()); + assert_head_matches_replay(&chain); + } + #[test] fn finalized_redelivery_at_or_below_final_tip_is_already_applied() { // Restart shape: the store's tip (incl. not-yet-finalized blocks) is diff --git a/lez/chain_state/src/lib.rs b/lez/chain_state/src/lib.rs index 6398ce77..5a9a026b 100644 --- a/lez/chain_state/src/lib.rs +++ b/lez/chain_state/src/lib.rs @@ -1,6 +1,6 @@ //! Storage-free chain-state core shared by the LEZ sequencer and indexer: //! the [`apply_block`] entry point plus [`BlockIngestError`], [`StallReason`], -//! [`Tip`], and [`AcceptOutcome`]. See `DESIGN.md` for the two-tier model. +//! [`Tip`], and [`AcceptOutcome`]. See [`ChainState`] for the two-tier model. pub use apply::{AcceptOutcome, Tip, apply_block, apply_block_to_state, validate_against_tip}; pub use chain::{ChainState, HeadEntry}; diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 0a6efe28..4722bf26 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -238,9 +238,8 @@ impl IndexerStore { return Ok(AcceptOutcome::AlreadyApplied); } - // Fail fast on validation before paying for the scratch clone — while - // parked, every re-delivered non-chaining block takes this path. - // Validation failures are never retryable, so parking here is exact. + // Validate before paying for the scratch clone; validation failures + // are never retryable, so parking immediately is exact. if let Err(err) = validate_against_tip(tip.as_ref(), block) { self.record_stall(Some(&block.header), l1_slot, err.clone())?; return Ok(AcceptOutcome::Parked(err)); diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 1fb52b1a..580edc44 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -9,7 +9,7 @@ use borsh::BorshDeserialize; use chain_state::{AcceptOutcome, ChainState, Tip}; use common::{ HashType, - block::{BedrockStatus, Block, HashableBlockData}, + block::{BedrockStatus, Block, BlockMeta, HashableBlockData}, transaction::{LeeTransaction, clock_invocation}, }; use config::{GenesisAction, SequencerConfig}; @@ -128,6 +128,67 @@ impl SequencerCore { } } + /// Rebuilds the two-tier [`ChainState`]: the final tier from the persisted + /// final snapshot (pre-genesis state when absent), the head tier by replaying + /// every stored block above it, so a post-restart orphan can still revert. + fn restore_chain_state( + config: &SequencerConfig, + store: &SequencerStore, + stored_head_state: &lee::V03State, + ) -> ChainState { + let final_snapshot = store + .dbio() + .get_final_snapshot() + .expect("Failed to read final snapshot from store"); + let (final_state, final_tip) = match final_snapshot { + Some((state, meta)) => ( + state, + Some(Tip { + block_id: meta.id, + hash: meta.hash, + }), + ), + // Nothing finalized yet: replay the whole stored chain. + None => (build_initial_state(config), None), + }; + let boundary = final_tip.as_ref().map_or(0, |tip| tip.block_id); + + let mut head_blocks = store + .get_all_blocks() + .filter_ok(|block| block.header.block_id > boundary) + .collect::, _>>() + .expect("Failed to read blocks from store while restoring chain state"); + head_blocks.sort_unstable_by_key(|block| block.header.block_id); + + let mut chain = ChainState::from_final(final_state, final_tip); + for block in head_blocks { + let block_id = block.header.block_id; + // NOTE: sentinel for the real (unpersisted) `MsgId`; never leaves the + // process, events correlate by hash. Persisting the real one needs a + // sidecar cell (block_id -> MsgId), since growing the shared `Block` + // type would ripple through every consumer's serialization. + let sentinel = MsgId::from(block.header.hash.0); + chain + .restore_head_block(sentinel, block) + .unwrap_or_else(|err| { + panic!( + "Stored block {block_id} does not replay while restoring chain state: {err}" + ) + }); + } + + // The replayed head must reproduce the persisted state byte-for-byte, + // else store and config disagree (e.g. edited genesis actions). + let replayed = borsh::to_vec(chain.head_state()).expect("state serializes"); + let stored = borsh::to_vec(stored_head_state).expect("state serializes"); + assert_eq!( + replayed, stored, + "Persisted state does not match the replayed chain; reset the store or restore the original config" + ); + + chain + } + pub async fn start_from_config( config: SequencerConfig, ) -> (Self, MemPoolHandle<(TransactionOrigin, LeeTransaction)>) { @@ -141,16 +202,8 @@ impl SequencerCore { let (store, state, _genesis_block) = Self::open_or_create_store(&config); - let latest_block_meta = store - .latest_block_meta() - .expect("Failed to read latest block meta from store"); - - let chain = Arc::new(Mutex::new(ChainState::from_final( - state, - Some(Tip { - block_id: latest_block_meta.id, - hash: latest_block_meta.hash, - }), + let chain = Arc::new(Mutex::new(Self::restore_chain_state( + &config, &store, &state, ))); let initial_checkpoint = store @@ -722,25 +775,27 @@ fn apply_follow_update( finalized, } = update; - // Apply under the lock and collect what to persist; take a single - // head snapshot. Release the lock before touching disk so the - // producer is never blocked on the follow path's I/O. - let (to_persist, resubmit_txs, head_snapshot) = { + // Apply under the lock and collect what to persist; release it before + // touching disk so the producer is never blocked on follow I/O. + let (to_persist, resubmit_txs, head_snapshot, final_snapshot) = { let mut chain = chain.lock().expect("chain state mutex poisoned"); - let mut resubmit_txs = Vec::new(); - for (this_msg, block) in &orphaned { - chain.revert_orphan(*this_msg); - resubmit_txs.extend(resubmittable_txs(block)); - } - let mut to_persist = Vec::new(); - for (this_msg, block) in &adopted { - if matches!( - chain.apply_adopted(*this_msg, block), - AcceptOutcome::Applied - ) { - to_persist.push((block, false)); - } - } + + // User txs of orphaned blocks, returned to the mempool below. + let resubmit_txs: Vec = orphaned + .iter() + .flat_map(|(_, block)| resubmittable_txs(block)) + .collect(); + + // Outcomes align with `adopted`. + let outcomes = chain.apply_channel_update(&orphaned, &adopted); + let mut to_persist: Vec<(&Block, bool)> = adopted + .iter() + .zip(&outcomes) + .filter(|(_, outcome)| matches!(outcome, AcceptOutcome::Applied)) + .map(|((_, block), _)| (block, false)) + .collect(); + + let mut final_advanced = false; for (this_msg, block) in &finalized { // FIXME: thread the finalized inscription's L1 slot once the // sdk surfaces it; only used for the invalid-finalized stall. @@ -749,9 +804,27 @@ fn apply_follow_update( AcceptOutcome::Applied ) { to_persist.push((block, true)); + final_advanced = true; } } - (to_persist, resubmit_txs, chain.head_state().clone()) + // Snapshot the advanced final tier so a restart re-anchors on it. + let final_snapshot = final_advanced.then(|| { + let tip = chain.final_tip().expect("advanced final tier has a tip"); + ( + chain.final_state().clone(), + BlockMeta { + id: tip.block_id, + hash: tip.hash, + }, + ) + }); + + ( + to_persist, + resubmit_txs, + chain.head_state().clone(), + final_snapshot, + ) }; // One atomic write for the whole update: blocks, tip meta and the state @@ -763,7 +836,11 @@ fn apply_follow_update( // ever landing in the store. Full `BlocksProcessed` atomicity (checkpoint // + blocks + state in one batch, per the sdk's event contract) is a // follow-up. - if let Err(err) = dbio.store_followed_blocks(&to_persist, &head_snapshot) { + if let Err(err) = dbio.store_followed_blocks( + &to_persist, + &head_snapshot, + final_snapshot.as_ref().map(|(state, meta)| (state, meta)), + ) { error!("Failed to persist followed blocks: {err:#}"); } @@ -830,41 +907,15 @@ fn replay_unfulfilled_deposit_events( }); } -/// Builds the initial genesis state from `testnet_initial_state` plus configured genesis -/// transactions. Returns the final state and the list of [`LeeTransaction`]s that should be -/// committed to the genesis block so external observers can replay them. -fn build_genesis_state(config: &SequencerConfig) -> (lee::V03State, Vec) { +/// The pre-genesis state: `testnet_initial_state` plus accounts seeded outside +/// any transaction (bridge-lock holdings, the cross-zone inbox config). +fn build_initial_state(config: &SequencerConfig) -> lee::V03State { #[cfg(not(feature = "testnet"))] let mut state = testnet_initial_state::initial_state(); #[cfg(feature = "testnet")] let mut state = testnet_initial_state::initial_state_testnet(); - let genesis_txs = config - .genesis - .iter() - .filter_map(|genesis_tx| match genesis_tx { - GenesisAction::SupplyAccount { - account_id, - balance, - } => Some(build_supply_account_genesis_transaction( - account_id, *balance, - )), - GenesisAction::SupplyBridgeAccount { balance } => { - Some(build_supply_bridge_account_genesis_transaction(*balance)) - } - // Force-inserted below: bridge_lock has no mint transaction. - GenesisAction::SupplyBridgeLockHolding { .. } => None, - }) - .chain(std::iter::once(clock_invocation(0))) - .inspect(|tx| { - state - .transition_from_public_transaction(tx, GENESIS_BLOCK_ID, 0) - .expect("Failed to execute genesis transaction"); - }) - .map(LeeTransaction::Public) - .collect(); - // Seed bridge-lock holder balances directly: they are not produced by any tx. for action in &config.genesis { if let GenesisAction::SupplyBridgeLockHolding { holder, amount } = action { @@ -882,6 +933,41 @@ fn build_genesis_state(config: &SequencerConfig) -> (lee::V03State, Vec (lee::V03State, Vec) { + let mut state = build_initial_state(config); + + let genesis_txs = config + .genesis + .iter() + .filter_map(|genesis_tx| match genesis_tx { + GenesisAction::SupplyAccount { + account_id, + balance, + } => Some(build_supply_account_genesis_transaction( + account_id, *balance, + )), + GenesisAction::SupplyBridgeAccount { balance } => { + Some(build_supply_bridge_account_genesis_transaction(*balance)) + } + // Force-inserted in `build_initial_state`: bridge_lock has no mint transaction. + GenesisAction::SupplyBridgeLockHolding { .. } => None, + }) + .chain(std::iter::once(clock_invocation(0))) + .inspect(|tx| { + state + .transition_from_public_transaction(tx, GENESIS_BLOCK_ID, 0) + .expect("Failed to execute genesis transaction"); + }) + .map(LeeTransaction::Public) + .collect(); + (state, genesis_txs) } diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 46af7019..ca35fe7f 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -1540,6 +1540,116 @@ async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() { assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized)); } +#[tokio::test] +async fn restart_restores_head_tier_and_recovers_from_orphan() { + let config = setup_sequencer_config(); + let acc1 = initial_public_user_accounts()[0].account_id; + let acc2 = initial_public_user_accounts()[1].account_id; + + // Produce block 2 (a user transfer), then "crash" before it finalizes. + let (tx, block2) = { + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config.clone()).await; + let tx = common::test_utils::create_transaction_native_token_transfer( + acc1, + 0, + acc2, + 10, + &create_signing_key_for_account1(), + ); + mempool_handle + .push((TransactionOrigin::User, tx.clone())) + .await + .unwrap(); + sequencer.produce_new_block().await.unwrap(); + (tx, sequencer.store.get_block_at_id(2).unwrap().unwrap()) + }; + + // Restart: nothing is finalized, so block 2 must come back as *head*, not + // final — the L1 can still orphan it. + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config.clone()).await; + assert_eq!(sequencer.chain_height(), 2); + + // The L1 orphans block 2 under its real MsgId (which we never persisted) + // and adopts a competing empty block 2'. + let genesis = sequencer.store.get_block_at_id(1).unwrap().unwrap(); + let block2_prime = + common::test_utils::produce_dummy_block(2, Some(genesis.header.hash), vec![]); + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + adopted: vec![(MsgId::from([21; 32]), block2_prime.clone())], + orphaned: vec![(MsgId::from([20; 32]), block2)], + finalized: vec![], + }, + ); + + // The head reorged onto 2': transfer reverted, store overwritten, and the + // orphaned user tx returned to the mempool. + assert_eq!(sequencer.chain_height(), 2); + let head_tip = sequencer + .chain() + .lock() + .expect("chain mutex poisoned") + .head_tip() + .expect("head tip set"); + assert_eq!(head_tip.hash, block2_prime.header.hash); + assert_eq!( + sequencer.with_state(|s| s.get_account_by_id(acc1).balance), + 10000, + "the orphaned transfer must be reverted" + ); + let stored = sequencer.store.get_block_at_id(2).unwrap().unwrap(); + assert_eq!(stored.header.hash, block2_prime.header.hash); + let (origin, requeued) = sequencer + .mempool + .pop() + .expect("orphaned user tx should be requeued"); + assert!(matches!(origin, TransactionOrigin::User)); + assert_eq!(requeued, tx); +} + +#[tokio::test] +async fn restart_reanchors_on_the_persisted_final_snapshot() { + let config = setup_sequencer_config(); + + // Produce block 2 and follow its finalization, which persists the final + // snapshot; then "crash". + { + let (mut sequencer, mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config.clone()).await; + let tx = common::test_utils::produce_dummy_empty_transaction(); + mempool_handle + .push((TransactionOrigin::User, tx)) + .await + .unwrap(); + sequencer.produce_new_block().await.unwrap(); + let block2 = sequencer.store.get_block_at_id(2).unwrap().unwrap(); + apply_follow_update( + &sequencer.store.dbio(), + &sequencer.chain(), + &mempool_handle, + FollowUpdate { + adopted: vec![], + orphaned: vec![], + finalized: vec![(MsgId::from(block2.header.hash.0), block2)], + }, + ); + } + + // Restart: the final tier re-anchors on the snapshot instead of treating + // the whole stored chain as final. + let (sequencer, _mempool_handle) = + SequencerCoreWithMockClients::start_from_config(config.clone()).await; + let chain = sequencer.chain(); + let chain = chain.lock().expect("chain mutex poisoned"); + assert_eq!(chain.final_tip().expect("final tip set").block_id, 2); + assert_eq!(chain.head_tip().expect("head tip set").block_id, 2); +} + #[tokio::test] async fn configure_channel_delegates_to_publisher() { let config = setup_sequencer_config(); diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index d7775e20..8d3a467d 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -19,10 +19,11 @@ use crate::{ }, error::DbError, sequencer::sequencer_cells::{ - LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, LatestBlockMetaCellOwned, - LatestBlockMetaCellRef, PendingDepositEventRecord, PendingDepositEventsCellOwned, - PendingDepositEventsCellRef, UnseenWithdrawCountCell, WithdrawalReconciliationKey, - ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, + FinalBlockMetaCellOwned, FinalBlockMetaCellRef, FinalLeeStateCellOwned, + FinalLeeStateCellRef, LEEStateCellOwned, LEEStateCellRef, LastFinalizedBlockIdCell, + LatestBlockMetaCellOwned, LatestBlockMetaCellRef, PendingDepositEventRecord, + PendingDepositEventsCellOwned, PendingDepositEventsCellRef, UnseenWithdrawCountCell, + WithdrawalReconciliationKey, ZoneSdkCheckpointCellOwned, ZoneSdkCheckpointCellRef, }, }; @@ -42,6 +43,10 @@ pub const DB_META_UNSEEN_WITHDRAW_COUNT_KEY: &str = "unseen_withdraw_count"; /// Key base for storing the LEE state. pub const DB_LEE_STATE_KEY: &str = "lee_state"; +/// Key base for storing the LEE state at the last L1-finalized block. +pub const DB_FINAL_LEE_STATE_KEY: &str = "final_lee_state"; +/// Key base for storing `(id, hash)` of the last L1-finalized block. +pub const DB_FINAL_BLOCK_META_KEY: &str = "final_block_meta"; /// Name of state column family. pub const CF_LEE_STATE_NAME: &str = "cf_lee_state"; @@ -522,6 +527,26 @@ impl RocksDBIO { .map(|opt| opt.map(|val| val.0)) } + /// `(state, meta)` at the last L1-finalized block; `None` until the first + /// finalization is observed. + pub fn get_final_snapshot(&self) -> DbResult> { + let Some(meta) = self.get_opt::(())? else { + return Ok(None); + }; + let state = self.get::(())?; + Ok(Some((state.0, meta.0))) + } + + fn put_final_snapshot_batch( + &self, + state: &V03State, + meta: &BlockMeta, + batch: &mut WriteBatch, + ) -> DbResult<()> { + self.put_batch(&FinalLeeStateCellRef(state), (), batch)?; + self.put_batch(&FinalBlockMetaCellRef(meta), (), batch) + } + pub fn get_lee_state(&self) -> DbResult { self.get::(()).map(|val| val.0) } @@ -619,36 +644,33 @@ impl RocksDBIO { } /// Persists a followed (peer) block so the store mirrors the canonical chain. - /// One-block form of [`Self::store_followed_blocks`]. + /// One-block form of [`Self::store_followed_blocks`], without a final snapshot. pub fn store_followed_block( &self, block: &Block, state: &V03State, finalized: bool, ) -> DbResult<()> { - self.store_followed_blocks(&[(block, finalized)], state) + self.store_followed_blocks(&[(block, finalized)], state, None) } - /// Persists a **batch** of followed (peer) blocks and the state after the last - /// of them in one atomic write, so a crash can never leave the stored state - /// ahead of the stored blocks and tip meta. + /// Persists a batch of followed blocks, the caller's head-tip `state`, and + /// the optional final-tier snapshot in one atomic write. /// - /// Per block, it: - /// - skips the payload write when the store already holds it (matched by id and hash) thus - /// covering our own blocks and re-deliveries - /// - stores it as finalized when its `finalized` flag is set - /// - /// When nothing needs writing, the state is left untouched too. + /// Per block: skips the payload write when the store already holds it (by + /// id and hash), unless `finalized` is set, which rewrites it with the + /// finalized status. When nothing needs writing, nothing is written — + /// an orphan-only update keeps the orphaned block and pre-revert state on + /// disk until the replacement overwrites them. /// /// TODO: the zone-sdk checkpoint is persisted by `on_checkpoint` *before* - /// this write; a crash in between resumes past these blocks without them - /// ever landing in the store. - /// Full `BlocksProcessed` atomicity by writing the checkpoint, blocks and - /// state in one batch, per the sdk's event contract is a follow-up. + /// this write. Full `BlocksProcessed` atomicity (checkpoint, blocks, state + /// and orphan reverts in one batch) is a follow-up. pub fn store_followed_blocks( &self, blocks: &[(&Block, bool)], state: &V03State, + final_snapshot: Option<(&V03State, &BlockMeta)>, ) -> DbResult<()> { let last_block_in_db = self.get_meta_last_block_in_db()?; let mut batch = WriteBatch::default(); @@ -683,7 +705,7 @@ impl RocksDBIO { } } - if batch.is_empty() { + if batch.is_empty() && final_snapshot.is_none() { return Ok(()); } // Meta is staged once for the batch's highest block. @@ -695,6 +717,9 @@ impl RocksDBIO { self.put_meta_latest_block_meta_batch(&tip, &mut batch)?; } self.put_lee_state_in_db_batch(state, &mut batch)?; + if let Some((final_state, final_meta)) = final_snapshot { + self.put_final_snapshot_batch(final_state, final_meta, &mut batch)?; + } self.db.write(batch).map_err(|rerr| { DbError::rocksdb_cast_message( diff --git a/lez/storage/src/sequencer/sequencer_cells.rs b/lez/storage/src/sequencer/sequencer_cells.rs index 7672e271..1d22e0f5 100644 --- a/lez/storage/src/sequencer/sequencer_cells.rs +++ b/lez/storage/src/sequencer/sequencer_cells.rs @@ -7,9 +7,10 @@ use crate::{ cells::{SimpleReadableCell, SimpleStorableCell, SimpleWritableCell}, error::DbError, sequencer::{ - CF_LEE_STATE_NAME, DB_LEE_STATE_KEY, DB_META_LAST_FINALIZED_BLOCK_ID, - DB_META_LATEST_BLOCK_META_KEY, DB_META_PENDING_DEPOSIT_EVENTS_KEY, - DB_META_UNSEEN_WITHDRAW_COUNT_KEY, DB_META_ZONE_SDK_CHECKPOINT_KEY, + CF_LEE_STATE_NAME, DB_FINAL_BLOCK_META_KEY, DB_FINAL_LEE_STATE_KEY, DB_LEE_STATE_KEY, + DB_META_LAST_FINALIZED_BLOCK_ID, DB_META_LATEST_BLOCK_META_KEY, + DB_META_PENDING_DEPOSIT_EVENTS_KEY, DB_META_UNSEEN_WITHDRAW_COUNT_KEY, + DB_META_ZONE_SDK_CHECKPOINT_KEY, }, }; @@ -43,6 +44,72 @@ impl SimpleWritableCell for LEEStateCellRef<'_> { } } +/// State at the last L1-finalized block, written atomically with +/// [`FinalBlockMetaCellRef`]. +#[derive(BorshDeserialize)] +pub struct FinalLeeStateCellOwned(pub V03State); + +impl SimpleStorableCell for FinalLeeStateCellOwned { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_FINAL_LEE_STATE_KEY; + const CF_NAME: &'static str = CF_LEE_STATE_NAME; +} + +impl SimpleReadableCell for FinalLeeStateCellOwned {} + +#[derive(BorshSerialize)] +pub struct FinalLeeStateCellRef<'state>(pub &'state V03State); + +impl SimpleStorableCell for FinalLeeStateCellRef<'_> { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_FINAL_LEE_STATE_KEY; + const CF_NAME: &'static str = CF_LEE_STATE_NAME; +} + +impl SimpleWritableCell for FinalLeeStateCellRef<'_> { + fn value_constructor(&self) -> DbResult> { + borsh::to_vec(&self).map_err(|err| { + DbError::borsh_cast_message(err, Some("Failed to serialize final state".to_owned())) + }) + } +} + +/// `(id, hash)` of the last L1-finalized block, paired with [`FinalLeeStateCellRef`]. +#[derive(BorshDeserialize)] +pub struct FinalBlockMetaCellOwned(pub BlockMeta); + +impl SimpleStorableCell for FinalBlockMetaCellOwned { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_FINAL_BLOCK_META_KEY; + const CF_NAME: &'static str = CF_META_NAME; +} + +impl SimpleReadableCell for FinalBlockMetaCellOwned {} + +#[derive(BorshSerialize)] +pub struct FinalBlockMetaCellRef<'blockmeta>(pub &'blockmeta BlockMeta); + +impl SimpleStorableCell for FinalBlockMetaCellRef<'_> { + type KeyParams = (); + + const CELL_NAME: &'static str = DB_FINAL_BLOCK_META_KEY; + const CF_NAME: &'static str = CF_META_NAME; +} + +impl SimpleWritableCell for FinalBlockMetaCellRef<'_> { + fn value_constructor(&self) -> DbResult> { + borsh::to_vec(&self).map_err(|err| { + DbError::borsh_cast_message( + err, + Some("Failed to serialize final block meta".to_owned()), + ) + }) + } +} + #[derive(Debug, BorshSerialize, BorshDeserialize)] pub struct LastFinalizedBlockIdCell(pub Option); diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index 2d86c494..1938684e 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -102,6 +102,7 @@ fn store_followed_blocks_batch_lands_meta_and_state_on_last_block() { dbio.store_followed_blocks( &[(&block2, true), (&block3, false)], &state_with_balance(300), + None, ) .unwrap(); @@ -117,6 +118,38 @@ fn store_followed_blocks_batch_lands_meta_and_state_on_last_block() { assert_eq!(stored_balance(&dbio), 300); } +#[test] +fn final_snapshot_round_trips_and_is_absent_on_fresh_store() { + let temp_dir = tempdir().unwrap(); + let (dbio, genesis) = dbio_with_genesis(temp_dir.path()); + + // Fresh store: no finalization observed yet. + assert!(dbio.get_final_snapshot().unwrap().is_none()); + + // A follow update that finalizes block 2 lands the snapshot in the same batch. + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + let final_meta = BlockMeta { + id: 2, + hash: block2.header.hash, + }; + dbio.store_followed_blocks( + &[(&block2, true)], + &state_with_balance(300), + Some((&state_with_balance(200), &final_meta)), + ) + .unwrap(); + + let (final_state, meta) = dbio + .get_final_snapshot() + .unwrap() + .expect("final snapshot is stored"); + assert_eq!(meta.id, 2); + assert_eq!(meta.hash, block2.header.hash); + assert_eq!(final_state.get_account_by_id(marker_id()).balance, 200); + // The head state is stored independently of the final snapshot. + assert_eq!(stored_balance(&dbio), 300); +} + #[test] fn store_followed_block_overwrites_competing_block_at_same_id() { let temp_dir = tempdir().unwrap();