diff --git a/Cargo.lock b/Cargo.lock index 0bd194bb..65e0cde7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1349,6 +1349,21 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "chain_state" +version = "0.1.0" +dependencies = [ + "anyhow", + "common", + "lee", + "logos-blockchain-core", + "logos-blockchain-zone-sdk", + "serde", + "serde_json", + "testnet_initial_state", + "thiserror 2.0.18", +] + [[package]] name = "chkstk_stub" version = "0.1.0" @@ -3852,6 +3867,7 @@ dependencies = [ "async-stream", "authenticated_transfer_core", "borsh", + "chain_state", "common", "futures", "humantime-serde", @@ -3865,7 +3881,6 @@ dependencies = [ "storage", "tempfile", "testnet_initial_state", - "thiserror 2.0.18", "tokio", "url", ] diff --git a/Cargo.toml b/Cargo.toml index 3e71f293..c9471681 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ members = [ "lez", "lez/system_accounts", + "lez/chain_state", "lez/sequencer/core", "lez/sequencer/service", "lez/sequencer/service/protocol", @@ -64,6 +65,7 @@ members = [ lee = { path = "lee/state_machine" } lee_core = { path = "lee/state_machine/core" } common = { path = "lez/common" } +chain_state = { path = "lez/chain_state" } mempool = { path = "lez/mempool" } storage = { path = "lez/storage" } key_protocol = { path = "lee/key_protocol" } diff --git a/lez/chain_state/Cargo.toml b/lez/chain_state/Cargo.toml new file mode 100644 index 00000000..1fcc052e --- /dev/null +++ b/lez/chain_state/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "chain_state" +version = "0.1.0" +edition = "2024" +license = { workspace = true } + +[lints] +workspace = true + +[dependencies] +common.workspace = true +lee.workspace = true +logos-blockchain-core.workspace = true +logos-blockchain-zone-sdk.workspace = true + +anyhow.workspace = true +serde.workspace = true +thiserror.workspace = true + +[dev-dependencies] +testnet_initial_state.workspace = true +serde_json.workspace = true diff --git a/lez/chain_state/DESIGN.md b/lez/chain_state/DESIGN.md new file mode 100644 index 00000000..600fb23f --- /dev/null +++ b/lez/chain_state/DESIGN.md @@ -0,0 +1,306 @@ +# `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. + +Branch: `erhant/lez-two-tip-chain-state` (off `erhant/indexer-recoverable-invalid-blocks`). + +--- + +## 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, l1_slot: Slot } + +enum AcceptOutcome { Applied, AlreadyApplied, Parked(BlockIngestError) } +``` + +`Tip` carries `l1_slot` (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/apply.rs b/lez/chain_state/src/apply.rs new file mode 100644 index 00000000..8a791a78 --- /dev/null +++ b/lez/chain_state/src/apply.rs @@ -0,0 +1,282 @@ +//! The single validate-then-apply entry point shared by the sequencer and the +//! indexer. Pure and storage-free: callers apply on a scratch clone of state and +//! commit only on `Ok`. + +use common::{ + HashType, + block::Block, + transaction::{LeeTransaction, clock_invocation}, +}; +use lee::{GENESIS_BLOCK_ID, V03State}; + +use crate::ingest_error::BlockIngestError; + +/// The parent the next block must chain on. +// `l1_slot` will be added here when the `ChainState` anchor layer lands. +#[derive(Debug, Clone)] +pub struct Tip { + pub block_id: u64, + pub hash: HashType, +} + +/// Outcome of feeding a parsed L2 block to a validated tip. +pub enum AcceptOutcome { + /// Chained and applied; the tip advances. + Applied, + /// A duplicate re-delivery of an already-applied block. No state change. + AlreadyApplied, + /// Did not chain or failed to apply; the tip stays frozen. + Parked(BlockIngestError), +} + +/// Validates `block` against `tip`, then applies it to `state`. +/// +/// Mutates `state` in place, so callers pass a scratch clone and commit on `Ok`. +pub fn apply_block( + tip: Option<&Tip>, + block: &Block, + state: &mut V03State, +) -> Result<(), BlockIngestError> { + validate_against_tip(tip, block)?; + apply_block_to_state(block, state)?; + Ok(()) +} + +/// Checks that `block` is the valid continuation of `tip`: hash integrity, +/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip +/// (cold state) expects the genesis block. +fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> { + let computed = block.recompute_hash(); + if computed != block.header.hash { + return Err(BlockIngestError::HashMismatch { + computed, + header: block.header.hash, + }); + } + + match tip { + None => { + if block.header.block_id != GENESIS_BLOCK_ID { + return Err(BlockIngestError::UnexpectedBlockId { + expected: GENESIS_BLOCK_ID, + got: block.header.block_id, + }); + } + } + Some(tip) => { + let expected = tip + .block_id + .checked_add(1) + .expect("block id should not overflow"); + if block.header.block_id != expected { + return Err(BlockIngestError::UnexpectedBlockId { + expected, + got: block.header.block_id, + }); + } + if block.header.prev_block_hash != tip.hash { + return Err(BlockIngestError::BrokenChainLink { + expected_prev: tip.hash, + got_prev: block.header.prev_block_hash, + }); + } + } + } + Ok(()) +} + +/// Applies a block's transactions to `state`, mapping every failure to a +/// [`BlockIngestError`] so the caller can park rather than crash. Operates in +/// place; the caller commits only on `Ok`. +fn apply_block_to_state(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> { + let (clock_tx, user_txs) = block + .body + .transactions + .split_last() + .ok_or(BlockIngestError::EmptyBlock)?; + + let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp)); + if *clock_tx != expected_clock { + return Err(BlockIngestError::InvalidClockTransaction); + } + + let is_genesis = block.header.block_id == GENESIS_BLOCK_ID; + for (tx_index, transaction) in user_txs.iter().enumerate() { + let state_transition = |err: anyhow::Error| BlockIngestError::StateTransition { + tx_index: tx_index.try_into().expect("tx index fits in u64"), + reason: format!("{err:#}"), + }; + if is_genesis { + let LeeTransaction::Public(public_tx) = transaction else { + return Err(BlockIngestError::NonPublicGenesisTransaction); + }; + state + .transition_from_public_transaction( + public_tx, + block.header.block_id, + block.header.timestamp, + ) + .map_err(|err| state_transition(err.into()))?; + } else { + transaction + .clone() + .execute_on_state(state, block.header.block_id, block.header.timestamp) + .map_err(|err| state_transition(err.into()))?; + } + } + + let LeeTransaction::Public(clock_public_tx) = clock_tx else { + return Err(BlockIngestError::InvalidClockTransaction); + }; + state + .transition_from_public_transaction( + clock_public_tx, + block.header.block_id, + block.header.timestamp, + ) + .map_err(|err| BlockIngestError::StateTransition { + tx_index: user_txs.len().try_into().expect("tx index fits in u64"), + reason: format!("{:#}", anyhow::Error::from(err)), + })?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use common::{ + block::HashableBlockData, + test_utils::{ + create_transaction_native_token_transfer, produce_dummy_block, + produce_dummy_empty_transaction, sequencer_sign_key_for_testing, + }, + }; + use testnet_initial_state::{initial_pub_accounts_private_keys, initial_state}; + + use super::*; + + fn tip_of(block: &Block) -> Tip { + Tip { + block_id: block.header.block_id, + hash: block.header.hash, + } + } + + #[test] + fn genesis_applies_on_empty_tip() { + let mut state = initial_state(); + let genesis = produce_dummy_block(1, None, vec![]); + apply_block(None, &genesis, &mut state).expect("genesis applies"); + } + + #[test] + fn non_genesis_first_block_is_unexpected_id() { + let mut state = initial_state(); + let block = produce_dummy_block(2, None, vec![]); + let err = apply_block(None, &block, &mut state).expect_err("should reject"); + assert!(matches!( + err, + BlockIngestError::UnexpectedBlockId { + expected: 1, + got: 2 + } + )); + } + + #[test] + fn skip_ahead_block_is_unexpected_id() { + let mut state = initial_state(); + let genesis = produce_dummy_block(1, None, vec![]); + apply_block(None, &genesis, &mut state).expect("genesis applies"); + + // Tip is at 1; a block with id 3 skips ahead. + let bad = produce_dummy_block(3, Some(genesis.header.hash), vec![]); + let err = + apply_block(Some(&tip_of(&genesis)), &bad, &mut state).expect_err("should reject"); + assert!(matches!( + err, + BlockIngestError::UnexpectedBlockId { + expected: 2, + got: 3 + } + )); + } + + #[test] + fn broken_chain_link_detected() { + let mut state = initial_state(); + let genesis = produce_dummy_block(1, None, vec![]); + apply_block(None, &genesis, &mut state).expect("genesis applies"); + + // Correct id (2), wrong parent hash. + let block2 = produce_dummy_block(2, Some(HashType([9_u8; 32])), vec![]); + let err = + apply_block(Some(&tip_of(&genesis)), &block2, &mut state).expect_err("should reject"); + assert!(matches!(err, BlockIngestError::BrokenChainLink { .. })); + } + + #[test] + fn hash_mismatch_detected() { + let mut state = initial_state(); + let mut genesis = produce_dummy_block(1, None, vec![]); + // Tampering with the header invalidates the stored hash. + genesis.header.timestamp = 999; + let err = apply_block(None, &genesis, &mut state).expect_err("should reject"); + assert!(matches!(err, BlockIngestError::HashMismatch { .. })); + } + + #[test] + fn empty_block_rejected() { + let mut state = initial_state(); + // A block with no transactions at all (not even the mandatory clock tx). + let block = HashableBlockData { + block_id: 1, + prev_block_hash: HashType([0_u8; 32]), + timestamp: 0, + transactions: vec![], + } + .into_pending_block(&sequencer_sign_key_for_testing()); + let err = apply_block(None, &block, &mut state).expect_err("should reject"); + assert!(matches!(err, BlockIngestError::EmptyBlock)); + } + + #[test] + fn missing_clock_tail_is_invalid_clock() { + let mut state = initial_state(); + // Last tx is not the expected clock invocation for the timestamp. + let block = HashableBlockData { + block_id: 1, + prev_block_hash: HashType([0_u8; 32]), + timestamp: 50, + transactions: vec![produce_dummy_empty_transaction()], + } + .into_pending_block(&sequencer_sign_key_for_testing()); + let err = apply_block(None, &block, &mut state).expect_err("should reject"); + assert!(matches!(err, BlockIngestError::InvalidClockTransaction)); + } + + #[test] + fn applies_transfers_and_advances_state() { + let mut state = initial_state(); + 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(); + + // Genesis (block 1): clock-only. + let genesis = produce_dummy_block(1, None, vec![]); + apply_block(None, &genesis, &mut state).expect("genesis applies"); + let mut tip = tip_of(&genesis); + + // Blocks 2..=11: one native transfer of 10 each (nonces 0..=9). + for i in 0..10_u64 { + let tx = create_transaction_native_token_transfer(from, i.into(), to, 10, &sign_key); + let block = produce_dummy_block(i + 2, Some(tip.hash), vec![tx]); + apply_block(Some(&tip), &block, &mut state).expect("transfer applies"); + tip = tip_of(&block); + } + + assert_eq!(state.get_account_by_id(from).balance, 9900); + assert_eq!(state.get_account_by_id(to).balance, 20100); + } +} diff --git a/lez/chain_state/src/chain.rs b/lez/chain_state/src/chain.rs new file mode 100644 index 00000000..f0b4eddb --- /dev/null +++ b/lez/chain_state/src/chain.rs @@ -0,0 +1,417 @@ +//! 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. + +use common::block::Block; +use lee::V03State; +use logos_blockchain_core::mantle::ops::channel::MsgId; +use logos_blockchain_zone_sdk::Slot; + +use crate::{ + AcceptOutcome, BlockIngestError, StallReason, + apply::{Tip, apply_block}, +}; + +/// A head block plus the channel message that carried it. +pub struct HeadEntry { + pub this_msg: MsgId, + pub block: Block, +} + +/// The head tier (reorg-able, from `adopted`/`orphaned`) over the final tier +/// (irreversible, from `finalized`). `head_state == final_state` replayed +/// through `head_blocks`. +pub struct ChainState { + final_state: V03State, + final_tip: Option, + head_state: V03State, + head_blocks: Vec, + final_stall: Option, +} + +impl ChainState { + /// Fresh state anchored at the genesis/initial state, no blocks applied. + #[must_use] + pub fn new(initial_state: V03State) -> Self { + Self::from_final(initial_state, None) + } + + /// State restored from a persisted final tier; head mirrors final. + #[must_use] + pub fn from_final(final_state: V03State, final_tip: Option) -> Self { + Self { + head_state: final_state.clone(), + final_state, + final_tip, + head_blocks: Vec::new(), + final_stall: None, + } + } + + /// State the sequencer builds its next block on. + #[must_use] + pub const fn head_state(&self) -> &V03State { + &self.head_state + } + + #[must_use] + pub const fn final_state(&self) -> &V03State { + &self.final_state + } + + /// Parent the next produced block must chain on. + #[must_use] + pub fn head_tip(&self) -> Option { + self.head_blocks + .last() + .map(|entry| tip_of(&entry.block)) + .or_else(|| self.final_tip.clone()) + } + + #[must_use] + pub fn final_tip(&self) -> Option { + self.final_tip.clone() + } + + #[must_use] + pub const fn final_stall(&self) -> Option<&StallReason> { + self.final_stall.as_ref() + } + + /// 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. + 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) + { + return AcceptOutcome::AlreadyApplied; + } + + let mut scratch = self.head_state.clone(); + match apply_block(tip.as_ref(), block, &mut scratch) { + Ok(()) => { + self.head_state = scratch; + self.head_blocks.push(HeadEntry { + this_msg, + block: block.clone(), + }); + AcceptOutcome::Applied + } + Err(err) => AcceptOutcome::Parked(err), + } + } + + /// 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) + { + self.head_blocks.truncate(idx); + self.rederive_head(); + } + } + + /// One channel update: revert every `orphaned`, then apply every `adopted`. + pub fn apply_channel_update( + &mut self, + orphaned: &[MsgId], + adopted: &[(MsgId, Block)], + ) -> Vec { + let earliest = orphaned + .iter() + .filter_map(|msg| { + self.head_blocks + .iter() + .position(|entry| entry.this_msg == *msg) + }) + .min(); + if let Some(idx) = earliest { + self.head_blocks.truncate(idx); + self.rederive_head(); + } + adopted + .iter() + .map(|(msg, block)| self.apply_adopted(*msg, block)) + .collect() + } + + /// 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`. + pub fn apply_finalized( + &mut self, + this_msg: MsgId, + 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), + } + } + + /// Moves `head_blocks[0..=idx]` into the final tier (already validated in head). + fn finalize_through(&mut self, idx: usize) { + let finalized: Vec = self.head_blocks.drain(0..=idx).collect(); + for entry in finalized { + apply_block(self.final_tip.as_ref(), &entry.block, &mut self.final_state) + .expect("a validated head block must apply to the final tier"); + self.final_tip = Some(tip_of(&entry.block)); + } + self.final_stall = None; + } + + /// 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 { + let mut scratch = self.final_state.clone(); + match apply_block(self.final_tip.as_ref(), block, &mut scratch) { + Ok(()) => { + self.final_state = scratch; + self.final_tip = Some(tip_of(block)); + self.final_stall = None; + self.head_blocks.clear(); + self.head_state = self.final_state.clone(); + AcceptOutcome::Applied + } + Err(err) => { + self.record_final_stall(block, l1_slot, err.clone()); + AcceptOutcome::Parked(err) + } + } + } + + /// Rebuilds `head_state` from the final tier plus the current `head_blocks`. + fn rederive_head(&mut self) { + let mut state = self.final_state.clone(); + let mut tip = self.final_tip.clone(); + for entry in &self.head_blocks { + apply_block(tip.as_ref(), &entry.block, &mut state) + .expect("validated head blocks must replay"); + tip = Some(tip_of(&entry.block)); + } + self.head_state = state; + } + + /// First stall is stored verbatim; later ones only bump `orphans_since`. + fn record_final_stall(&mut self, block: &Block, l1_slot: Slot, error: BlockIngestError) { + self.final_stall = Some(match self.final_stall.take() { + Some(mut existing) => { + existing.orphans_since = existing.orphans_since.saturating_add(1); + existing + } + None => stall_for(block, l1_slot, error), + }); + } +} + +const fn tip_of(block: &Block) -> Tip { + Tip { + block_id: block.header.block_id, + hash: block.header.hash, + } +} + +const fn stall_for(block: &Block, l1_slot: Slot, error: BlockIngestError) -> StallReason { + StallReason { + block_id: Some(block.header.block_id), + block_hash: Some(block.header.hash), + prev_block_hash: Some(block.header.prev_block_hash), + l1_slot, + error, + first_seen: Some(block.header.timestamp), + orphans_since: 0, + } +} + +#[cfg(test)] +mod tests { + use common::test_utils::{create_transaction_native_token_transfer, produce_dummy_block}; + use testnet_initial_state::{initial_pub_accounts_private_keys, initial_state}; + + use super::*; + + fn msg(n: u8) -> MsgId { + MsgId::from([n; 32]) + } + + fn slot(n: u64) -> Slot { + Slot::from(n) + } + + #[test] + fn adopted_blocks_advance_head() { + let mut chain = ChainState::new(initial_state()); + + let genesis = produce_dummy_block(1, None, vec![]); + assert!(matches!( + chain.apply_adopted(msg(1), &genesis), + AcceptOutcome::Applied + )); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + assert!(matches!( + chain.apply_adopted(msg(2), &block2), + AcceptOutcome::Applied + )); + + assert_eq!(chain.head_tip().expect("head tip").block_id, 2); + // Nothing finalized yet. + assert!(chain.final_tip().is_none()); + } + + #[test] + fn adopted_bad_block_freezes_head_without_stall() { + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + chain.apply_adopted(msg(1), &genesis); + + // Skips ahead (id 3 while head tip is 1). + let bad = produce_dummy_block(3, Some(genesis.header.hash), vec![]); + assert!(matches!( + chain.apply_adopted(msg(3), &bad), + AcceptOutcome::Parked(BlockIngestError::UnexpectedBlockId { + expected: 2, + got: 3 + }) + )); + assert_eq!(chain.head_tip().expect("head tip").block_id, 1); + assert!( + chain.final_stall().is_none(), + "head freeze records no stall" + ); + } + + #[test] + fn adopted_is_idempotent() { + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + chain.apply_adopted(msg(1), &genesis); + + assert!(matches!( + chain.apply_adopted(msg(1), &genesis), + AcceptOutcome::AlreadyApplied + )); + assert_eq!(chain.head_tip().expect("head tip").block_id, 1); + } + + #[test] + fn orphan_reverts_head() { + 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![]); + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + chain.apply_adopted(msg(1), &genesis); + chain.apply_adopted(msg(2), &block2); + chain.apply_adopted(msg(3), &block3); + + chain.revert_orphan(msg(3)); + assert_eq!(chain.head_tip().expect("head tip").block_id, 2); + + // A competing block 3 now applies cleanly on block 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_tip().expect("head tip").block_id, 3); + } + + #[test] + fn channel_update_reverts_then_applies() { + 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![]); + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + chain.apply_adopted(msg(1), &genesis); + chain.apply_adopted(msg(2), &block2); + 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)]); + assert!(matches!(outcomes.as_slice(), [AcceptOutcome::Applied])); + assert_eq!(chain.head_tip().expect("head tip").block_id, 3); + } + + #[test] + fn finalize_moves_head_into_final() { + 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![]); + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + chain.apply_adopted(msg(1), &genesis); + chain.apply_adopted(msg(2), &block2); + chain.apply_adopted(msg(3), &block3); + + // Finalize through block 2. + assert!(matches!( + chain.apply_finalized(msg(2), &block2, slot(100)), + AcceptOutcome::Applied + )); + assert_eq!(chain.final_tip().expect("final tip").block_id, 2); + // Head tip unchanged; head still ends at 3. + assert_eq!(chain.head_tip().expect("head tip").block_id, 3); + } + + #[test] + fn backfill_applies_directly_to_final() { + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + assert!(matches!( + chain.apply_finalized(msg(1), &genesis, slot(10)), + AcceptOutcome::Applied + )); + assert_eq!(chain.final_tip().expect("final tip").block_id, 1); + // Head mirrors final during backfill. + assert_eq!(chain.head_tip().expect("head tip").block_id, 1); + } + + #[test] + fn invalid_finalized_block_sets_final_stall() { + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + chain.apply_finalized(msg(1), &genesis, slot(10)); + + // Skip-ahead finalized block, not in head: parks the final tier. + let bad = produce_dummy_block(3, Some(genesis.header.hash), vec![]); + assert!(matches!( + chain.apply_finalized(msg(3), &bad, slot(20)), + AcceptOutcome::Parked(_) + )); + let stall = chain.final_stall().expect("final stall recorded"); + assert_eq!(stall.block_id, Some(3)); + } + + #[test] + fn head_state_reflects_applied_transfers() { + 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(); + + let mut chain = ChainState::new(initial_state()); + let genesis = produce_dummy_block(1, None, vec![]); + chain.apply_adopted(msg(1), &genesis); + + let tx = create_transaction_native_token_transfer(from, 0, to, 10, &sign_key); + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]); + chain.apply_adopted(msg(2), &block2); + + assert_eq!(chain.head_state().get_account_by_id(from).balance, 9990); + assert_eq!(chain.head_state().get_account_by_id(to).balance, 20010); + } +} diff --git a/lez/indexer/core/src/ingest_error.rs b/lez/chain_state/src/ingest_error.rs similarity index 92% rename from lez/indexer/core/src/ingest_error.rs rename to lez/chain_state/src/ingest_error.rs index bda53497..bd377a0e 100644 --- a/lez/indexer/core/src/ingest_error.rs +++ b/lez/chain_state/src/ingest_error.rs @@ -1,10 +1,10 @@ use common::HashType; use serde::{Deserialize, Serialize}; -/// Why the indexer could not apply an L2 block from the channel. +/// Why an L2 block from the channel could not be applied. /// -/// Persisted in `RocksDB`, so every variant must have the following -/// traits: `Clone + Serialize + Deserialize`. +/// Persisted in `RocksDB` (as part of [`crate::StallReason`]), so every variant +/// must be `Clone + Serialize + Deserialize`. #[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)] pub enum BlockIngestError { #[error("Failed to deserialize L2 block: {0}")] diff --git a/lez/chain_state/src/lib.rs b/lez/chain_state/src/lib.rs new file mode 100644 index 00000000..c4725ec3 --- /dev/null +++ b/lez/chain_state/src/lib.rs @@ -0,0 +1,13 @@ +//! 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. + +pub use apply::{AcceptOutcome, Tip, apply_block}; +pub use chain::{ChainState, HeadEntry}; +pub use ingest_error::BlockIngestError; +pub use stall_reason::StallReason; + +pub mod apply; +pub mod chain; +pub mod ingest_error; +pub mod stall_reason; diff --git a/lez/indexer/core/src/stall_reason.rs b/lez/chain_state/src/stall_reason.rs similarity index 100% rename from lez/indexer/core/src/stall_reason.rs rename to lez/chain_state/src/stall_reason.rs diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index b3877639..1518b540 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -12,6 +12,7 @@ default = [] testnet = [] [dependencies] +chain_state.workspace = true common.workspace = true logos-blockchain-zone-sdk.workspace = true lee.workspace = true @@ -29,7 +30,6 @@ futures.workspace = true url.workspace = true logos-blockchain-core.workspace = true serde_json.workspace = true -thiserror.workspace = true async-stream.workspace = true tokio.workspace = true diff --git a/lez/indexer/core/src/block_store.rs b/lez/indexer/core/src/block_store.rs index 2a2bfbce..4bfb2644 100644 --- a/lez/indexer/core/src/block_store.rs +++ b/lez/indexer/core/src/block_store.rs @@ -1,12 +1,12 @@ use std::{path::Path, sync::Arc}; use anyhow::{Context as _, Result}; +use chain_state::{AcceptOutcome, BlockIngestError, StallReason, Tip, apply_block}; use common::{ - HashType, block::{BedrockStatus, Block, BlockHeader}, - transaction::{LeeTransaction, clock_invocation}, + transaction::LeeTransaction, }; -use lee::{Account, AccountId, GENESIS_BLOCK_ID, V03State}; +use lee::{Account, AccountId, V03State}; use lee_core::BlockId; use log::warn; use logos_blockchain_core::header::HeaderId; @@ -14,23 +14,6 @@ use logos_blockchain_zone_sdk::Slot; use storage::indexer::RocksDBIO; use tokio::sync::RwLock; -use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; - -struct Tip { - block_id: u64, - hash: HashType, -} - -/// Outcome of feeding a parsed L2 block to the validated tip. -pub enum AcceptOutcome { - /// Chained and applied; tip and L1 read cursor both advance. - Applied, - /// A duplicate re-delivery of the current tip. Just L2 advances. - AlreadyApplied, - /// Did not chain or failed to apply; tip stays frozen, stall recorded. - Parked(BlockIngestError), -} - #[derive(Clone)] pub struct IndexerStore { dbio: Arc, @@ -246,14 +229,9 @@ impl IndexerStore { return Ok(AcceptOutcome::AlreadyApplied); } - 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)); - } - // TODO: we use scratch state to be atomic, but need to revisit how expensive a clone is let mut scratch = self.current_state.read().await.clone(); - if let Err(err) = apply_block_to_scratch(block, &mut scratch) { + if let Err(err) = apply_block(tip.as_ref(), block, &mut scratch) { self.record_stall(Some(&block.header), l1_slot, err.clone())?; return Ok(AcceptOutcome::Parked(err)); } @@ -275,112 +253,11 @@ impl IndexerStore { } } -/// Checks that `block` is the valid continuation of `tip`: hash integrity, -/// then block-id continuity, then `prev_block_hash` linkage. A `None` tip -/// (cold store) expects the genesis block. -fn validate_against_tip(tip: Option<&Tip>, block: &Block) -> Result<(), BlockIngestError> { - let computed = block.recompute_hash(); - if computed != block.header.hash { - return Err(BlockIngestError::HashMismatch { - computed, - header: block.header.hash, - }); - } - - match tip { - None => { - if block.header.block_id != GENESIS_BLOCK_ID { - return Err(BlockIngestError::UnexpectedBlockId { - expected: GENESIS_BLOCK_ID, - got: block.header.block_id, - }); - } - } - Some(tip) => { - let expected = tip - .block_id - .checked_add(1) - .expect("block id should not overflow"); - if block.header.block_id != expected { - return Err(BlockIngestError::UnexpectedBlockId { - expected, - got: block.header.block_id, - }); - } - if block.header.prev_block_hash != tip.hash { - return Err(BlockIngestError::BrokenChainLink { - expected_prev: tip.hash, - got_prev: block.header.prev_block_hash, - }); - } - } - } - Ok(()) -} - -/// Applies a block's transactions to `state`, mapping every failure to a -/// [`BlockIngestError`] so the caller can park rather than crash. Operates on a -/// scratch state; the caller commits only on `Ok`. -fn apply_block_to_scratch(block: &Block, state: &mut V03State) -> Result<(), BlockIngestError> { - let (clock_tx, user_txs) = block - .body - .transactions - .split_last() - .ok_or(BlockIngestError::EmptyBlock)?; - - let expected_clock = LeeTransaction::Public(clock_invocation(block.header.timestamp)); - if *clock_tx != expected_clock { - return Err(BlockIngestError::InvalidClockTransaction); - } - - let is_genesis = block.header.block_id == GENESIS_BLOCK_ID; - for (tx_index, transaction) in user_txs.iter().enumerate() { - let state_transition = |err: anyhow::Error| BlockIngestError::StateTransition { - tx_index: tx_index.try_into().expect("tx index fits in u64"), - reason: format!("{err:#}"), - }; - if is_genesis { - let LeeTransaction::Public(public_tx) = transaction else { - return Err(BlockIngestError::NonPublicGenesisTransaction); - }; - state - .transition_from_public_transaction( - public_tx, - block.header.block_id, - block.header.timestamp, - ) - .map_err(|err| state_transition(err.into()))?; - } else { - transaction - .clone() - .execute_on_state(state, block.header.block_id, block.header.timestamp) - .map_err(|err| state_transition(err.into()))?; - } - } - - let LeeTransaction::Public(clock_public_tx) = clock_tx else { - return Err(BlockIngestError::InvalidClockTransaction); - }; - state - .transition_from_public_transaction( - clock_public_tx, - block.header.block_id, - block.header.timestamp, - ) - .map_err(|err| BlockIngestError::StateTransition { - tx_index: user_txs.len().try_into().expect("tx index fits in u64"), - reason: format!("{:#}", anyhow::Error::from(err)), - })?; - - Ok(()) -} - #[cfg(test)] mod stall_reason_tests { use common::HashType; use super::*; - use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; #[tokio::test] async fn stall_reason_roundtrips_and_clears() { @@ -529,7 +406,6 @@ mod accept_tests { use common::{HashType, block::HashableBlockData, test_utils::produce_dummy_block}; use super::*; - use crate::ingest_error::BlockIngestError; fn signing_key() -> lee::PrivateKey { lee::PrivateKey::try_new([7_u8; 32]).expect("valid key") diff --git a/lez/indexer/core/src/chain_consistency.rs b/lez/indexer/core/src/chain_consistency.rs index 00f28822..30c3d552 100644 --- a/lez/indexer/core/src/chain_consistency.rs +++ b/lez/indexer/core/src/chain_consistency.rs @@ -305,8 +305,7 @@ mod tests { use super::*; use crate::{ - BlockIngestError, - block_store::AcceptOutcome, + AcceptOutcome, BlockIngestError, config::{ChannelId, ClientConfig, IndexerConfig}, }; diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index 77839296..600630c6 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -2,18 +2,17 @@ use std::{path::Path, sync::Arc}; use anyhow::Result; use arc_swap::ArcSwap; +pub use chain_state::{AcceptOutcome, BlockIngestError, StallReason}; use common::block::Block; // TODO: Remove after testnet use futures::StreamExt as _; -pub use ingest_error::BlockIngestError; use log::{error, info, warn}; use logos_blockchain_zone_sdk::{ CommonHttpClient, Slot, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, }; -pub use stall_reason::StallReason; use crate::{ - block_store::{AcceptOutcome, IndexerStore}, + block_store::IndexerStore, chain_consistency::ChainConsistency, config::IndexerConfig, status::{IndexerStatus, IndexerSyncStatus}, @@ -21,8 +20,6 @@ use crate::{ pub mod block_store; pub mod chain_consistency; pub mod config; -pub mod ingest_error; -pub mod stall_reason; pub mod status; #[derive(Clone)] diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index a483fde6..aa182c23 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -1,7 +1,6 @@ +use chain_state::StallReason; use serde::Serialize; -use crate::stall_reason::StallReason; - /// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell /// "still catching up" apart from "something went wrong". #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] @@ -117,10 +116,9 @@ mod tests { #[test] fn stalled_status_serializes_with_stall_reason() { + use chain_state::{BlockIngestError, StallReason}; use logos_blockchain_zone_sdk::Slot; - use crate::{ingest_error::BlockIngestError, stall_reason::StallReason}; - let status = IndexerStatus { sync: IndexerSyncStatus::stalled("broken chain link".to_owned()), indexed_block_id: Some(41), diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 21551131..f8a64c5d 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -12,10 +12,14 @@ use logos_blockchain_zone_sdk::{ adapter::NodeHttpClient, sequencer::{ DepositInfo, Event, FinalizedOp, InscriptionInfo, - SequencerConfig as ZoneSdkSequencerConfig, WithdrawArg, WithdrawInfo, ZoneSequencer, + SequencerConfig as ZoneSdkSequencerConfig, TurnNotification, WithdrawArg, WithdrawInfo, + ZoneSequencer, }, }; -use tokio::{sync::mpsc, task::JoinHandle}; +use tokio::{ + sync::{mpsc, watch}, + task::JoinHandle, +}; use crate::config::BedrockConfig; @@ -63,6 +67,9 @@ pub trait BlockPublisherTrait: Clone { async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result<()>; fn channel_id(&self) -> ChannelId; + + /// Whether this sequencer is currently authorized to write to the channel. + fn is_our_turn(&self) -> bool; } /// Real block publisher backed by zone-sdk's `ZoneSequencer`. @@ -70,6 +77,7 @@ pub trait BlockPublisherTrait: Clone { pub struct ZoneSdkPublisher { channel_id: ChannelId, publish_tx: mpsc::Sender<(Inscription, Vec)>, + turn_rx: watch::Receiver, // Aborts the drive task when the last clone is dropped. _drive_task: Arc, } @@ -112,6 +120,8 @@ impl BlockPublisherTrait for ZoneSdkPublisher { // Grab readiness receiver before moving the sequencer into the drive // task so we can await cold-start completion below. let mut ready_rx = sequencer.subscribe_ready(); + // Grab the turn watch before the move; the sdk actor keeps it current. + let turn_rx = sequencer.subscribe_turn_to_write(); let (publish_tx, mut publish_rx) = mpsc::channel::<(Inscription, Vec)>(PUBLISH_INBOX_CAPACITY); @@ -197,6 +207,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher { Ok(Self { channel_id: config.channel_id, publish_tx, + turn_rx, _drive_task: Arc::new(DriveTaskGuard(drive_task)), }) } @@ -218,6 +229,10 @@ impl BlockPublisherTrait for ZoneSdkPublisher { fn channel_id(&self) -> ChannelId { self.channel_id } + + fn is_our_turn(&self) -> bool { + self.turn_rx.borrow().our_turn_to_write + } } /// Deserialize inscription payload as a `Block` and return it's`block_id`. diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 75dcc1a9..c7e7e4a7 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -534,6 +534,12 @@ impl SequencerCore { self.block_publisher.clone() } + /// Whether this sequencer is currently authorized to write to the channel. + #[must_use] + pub fn is_our_turn(&self) -> bool { + self.block_publisher.is_our_turn() + } + fn next_block_id(&self) -> u64 { self.chain_height .checked_add(1) diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index 39f635f9..81508f7c 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -48,4 +48,8 @@ impl BlockPublisherTrait for MockBlockPublisher { fn channel_id(&self) -> ChannelId { self.channel_id } + + fn is_our_turn(&self) -> bool { + true + } } diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index 687ee424..d6779102 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -172,16 +172,15 @@ async fn main_loop(seq_core: Arc>, block_timeout: Duration) loop { tokio::time::sleep(block_timeout).await; - info!("Collecting transactions from mempool, block creation"); + let mut state = seq_core.lock().await; - let id = { - let mut state = seq_core.lock().await; - - state.produce_new_block().await? - }; + // Only produce on our turn. + if !state.is_our_turn() { + continue; + } + info!("Our turn: collecting transactions from mempool, creating block"); + let id = state.produce_new_block().await?; info!("Block with id {id} created"); - - info!("Waiting for new transactions"); } }