mirror of
https://github.com/logos-blockchain/logos-execution-zone.git
synced 2026-07-18 11:49:31 +00:00
fix(chain_state, sequencer, storage): rebuild the two-tier state on restart
- 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
This commit is contained in:
erhant
parent
f8bc1581ce
commit
0135049710
1
Cargo.lock
generated
1
Cargo.lock
generated
@ -1376,6 +1376,7 @@ dependencies = [
|
||||
"borsh",
|
||||
"common",
|
||||
"lee",
|
||||
"log",
|
||||
"logos-blockchain-core",
|
||||
"logos-blockchain-zone-sdk",
|
||||
"serde",
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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<Tip>,
|
||||
head_state: V03State, // final_state + applied head blocks
|
||||
head_blocks: Vec<HeadEntry>, // ordered, above final_tip
|
||||
final_stall: Option<StallReason>, // 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<br/>non-empty?"}
|
||||
ORPH -->|yes| REV["For each orphaned by this_msg:<br/>drop from head_blocks,<br/>return its txs to mempool"]
|
||||
REV --> RED["Re-derive head_state:<br/>clone final_state, replay survivors"]
|
||||
ORPH -->|no| ADO
|
||||
RED --> ADO{"adopted<br/>non-empty?"}
|
||||
ADO -->|"yes, in order"| DEDUP{"this_msg in<br/>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,<br/>advance head tip, clear stall"]
|
||||
OUT -->|AlreadyApplied| SKIP
|
||||
OUT -->|"Parked(err)"| PARK["freeze head tip — do NOT apply.<br/>No stall recorded (self-heals<br/>via reorg/finalization)"]
|
||||
SKIP --> FIN
|
||||
APP --> FIN
|
||||
PARK --> FIN
|
||||
FIN{"finalized<br/>inscriptions?"}
|
||||
FIN -->|"already in head (steady state)"| MOVE["finalize_up_to:<br/>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,<br/>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).
|
||||
@ -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<Tip>,
|
||||
@ -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<usize> {
|
||||
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<HashType> {
|
||||
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<AcceptOutcome> {
|
||||
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
|
||||
|
||||
@ -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};
|
||||
|
||||
@ -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));
|
||||
|
||||
@ -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<BP: BlockPublisherTrait> SequencerCore<BP> {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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::<Result<Vec<_>, _>>()
|
||||
.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<BP: BlockPublisherTrait> SequencerCore<BP> {
|
||||
|
||||
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<LeeTransaction> = 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<LeeTransaction>) {
|
||||
/// 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<LeeTrans
|
||||
state.insert_genesis_account(config_id, config_account);
|
||||
}
|
||||
|
||||
state
|
||||
}
|
||||
|
||||
/// Builds the initial genesis state from [`build_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<LeeTransaction>) {
|
||||
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)
|
||||
}
|
||||
|
||||
|
||||
@ -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();
|
||||
|
||||
@ -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<Option<(V03State, BlockMeta)>> {
|
||||
let Some(meta) = self.get_opt::<FinalBlockMetaCellOwned>(())? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let state = self.get::<FinalLeeStateCellOwned>(())?;
|
||||
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<V03State> {
|
||||
self.get::<LEEStateCellOwned>(()).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(
|
||||
|
||||
@ -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<Vec<u8>> {
|
||||
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<Vec<u8>> {
|
||||
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<u64>);
|
||||
|
||||
|
||||
@ -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();
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user