From 4db615e0e35d70034253d2d4d61754d2ea8d9ff5 Mon Sep 17 00:00:00 2001 From: erhant Date: Wed, 8 Jul 2026 16:33:58 +0300 Subject: [PATCH 1/7] feat(chain_state): initial design doc --- lez/chain_state/DESIGN.md | 248 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 248 insertions(+) create mode 100644 lez/chain_state/DESIGN.md diff --git a/lez/chain_state/DESIGN.md b/lez/chain_state/DESIGN.md new file mode 100644 index 00000000..3bb993e3 --- /dev/null +++ b/lez/chain_state/DESIGN.md @@ -0,0 +1,248 @@ +# `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_core`: that is the RISC0 guest crate; a stateful host machine has + no place in the zkVM circuit and does not know what a `Block` is. +- **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 (decision: A).** `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. +A storage-trait abstraction (option B) is a possible later follow-up, not part of +this task. + +## 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 + stall: Option, +} + +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 record stall + park (head tip frozen). +- `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. +- `rollback_orphan(this_msg)` — drop from that entry forward, re-derive head. +- `status() -> { final_height, head_height, stall }` — for RPC/UI. + +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. + +## 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["record StallReason,
freeze head tip,
mark processed — do NOT apply"] + 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,
mirror into head"] + FIN -->|none| CP + MOVE --> CP + DIRECT --> 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: apply_block returns Parked(err) + CaughtUp --> Parked: invalid block adopted + Parked --> Parked: further non-chaining blocks (orphans_since++) + Parked --> Syncing: valid successor chains on frozen tip → stall cleared + note right of Parked + head tip frozen at last valid block. + Producer (on our turn) builds on this + frozen tip, same as honest peers, + so the invalid block is skipped. + 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** + +| # | Scenario | Handling | Expected | +| --- | ------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ----------------------------------------------- | +| 8 | Authorized sequencer posts a block with an invalid state transition | `apply_block` → `Parked(StateTransition)`; freeze head tip, record stall, mark processed | 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). From 8ab13de775e22d851e37455672e78575611f547a Mon Sep 17 00:00:00 2001 From: erhant Date: Wed, 8 Jul 2026 17:00:06 +0300 Subject: [PATCH 2/7] chore(chain_state): update DESIGN.md --- lez/chain_state/DESIGN.md | 103 ++++++++++++++++++++++++++++---------- 1 file changed, 76 insertions(+), 27 deletions(-) diff --git a/lez/chain_state/DESIGN.md b/lez/chain_state/DESIGN.md index 3bb993e3..3e0bc384 100644 --- a/lez/chain_state/DESIGN.md +++ b/lez/chain_state/DESIGN.md @@ -33,20 +33,16 @@ sequencer and indexer expose their exact current state. lee_core ← lee (owns V03State) ← common (owns Block, BedrockStatus, clock_invocation, recompute_hash) ← lez/chain_state ← { indexer/core, sequencer/core } ``` -- **Not** `lee_core`: that is the RISC0 guest crate; a stateful host machine has - no place in the zkVM circuit and does not know what a `Block` is. - **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 (decision: A).** `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. -A storage-trait abstraction (option B) is a possible later follow-up, not part of -this task. +**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 @@ -90,7 +86,8 @@ struct ChainState { final_tip: Option, head_state: V03State, // final_state + applied head blocks head_blocks: Vec, // ordered, above final_tip - stall: Option, + final_stall: Option, // persisted to RocksDB — see §4a + head_stall: Option, // in-memory only — recomputed from the stream on restart } struct HeadEntry { this_msg: MsgId, block: Block } @@ -105,8 +102,9 @@ 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 record stall + park (head tip frozen). + outbox, else `apply_block` on the head tip; on success push `HeadEntry`. On + failure, set **`head_stall`** (in-memory `StallReason`) and freeze the head tip + at the last valid block — do **not** persist. - `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. @@ -114,14 +112,59 @@ Operations: `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. -- `rollback_orphan(this_msg)` — drop from that entry forward, re-derive head. -- `status() -> { final_height, head_height, stall }` — for RPC/UI. + `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** + `StallReason` written to disk (see §4a). +- `rollback_orphan(this_msg)` — drop from that entry forward, re-derive head; + clears `head_stall` if the re-derived head is clean. +- `status() -> { final_height, head_height, head_stall, final_stall }` — for RPC/UI. 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. Two stalls — persisted vs in-memory + +Both tiers carry a `StallReason` (`final_stall`, `head_stall`); they are equally +informative. The difference is **persistence**, which follows from durability. L1 +finality is about inscription **canonicality**, not LEZ-block **content** +validity, so an authorized sequencer can get a content-invalid block finalized — +which is why both tiers can meet an invalid block in the first place. + +- **`head_stall` — in-memory only.** The head is reorg-able and re-derived from + every `channel_update`. An invalid adopted block is transient: it is either + orphaned (a competing valid block at the same height wins) or it finalizes. We + set `head_stall` for observability (RPC/UI can show "head blocked at `N`: + StateTransition at tx 3") and freeze the head tip at the last valid block, but + we do **not** write it to disk — on restart the head is rebuilt from the stream, + so a persisted head stall would be redundant and could go stale. Because the + next adopted block chains on the bad one, it fails validation too, so the head + stays frozen until a valid block (competing or post-reorg) is adopted. +- **`final_stall` — persisted.** The final tier is irreversible. If an invalid + block *finalizes*, the node is durably stuck until a valid successor (built by + honest sequencers on the last valid parent) finalizes. This must survive + restart: the startup chain-consistency / anchor check reads it, it is what we + surface as `Stalled`, and it is the signal the committee acts on to evict a bad + sequencer. + +An invalid block **migrates the problem head→final** when it finalizes: `head_stall` +is set at `N` first; once `N+1(bad)` finalizes, `apply_finalized` fails and records +the persisted `final_stall`. Both tiers end up stuck at `N` consistently, and both +recover when `N+1′` finalizes. The indexer (final tier only) never sets +`head_stall`, so it behaves exactly as it does today. + +### 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 (`head_stall` set) 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 }`: @@ -160,16 +203,20 @@ flowchart TD VAL --> OUT{"AcceptOutcome"} OUT -->|Applied| APP["append this_msg+block to head,
advance head tip, clear stall"] OUT -->|AlreadyApplied| SKIP - OUT -->|"Parked(err)"| PARK["record StallReason,
freeze head tip,
mark processed — do NOT apply"] + OUT -->|"Parked(err)"| PARK["set head_stall (in-memory),
freeze head tip — do NOT apply.
Not persisted"] 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,
mirror into head"] + 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 - DIRECT --> CP + MIRROR --> CP + FSTALL --> CP CP["persist checkpoint atomically"] ``` @@ -180,15 +227,15 @@ stateDiagram-v2 [*] --> Syncing Syncing --> CaughtUp: stream drained, no stall CaughtUp --> Syncing: new adopted / finalized arrives - Syncing --> Parked: apply_block returns Parked(err) - CaughtUp --> Parked: invalid block adopted - Parked --> Parked: further non-chaining blocks (orphans_since++) - Parked --> Syncing: valid successor chains on frozen tip → stall cleared + 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 - head tip frozen at last valid block. - Producer (on our turn) builds on this - frozen tip, same as honest peers, - so the invalid block is skipped. + final_stall — persisted, survives restart. + Head-tier bad blocks do NOT enter this state: + they set head_stall (in-memory) and self-heal + via reorg/finalization. Producer (on our turn) + builds on the last valid tip either way. end note ``` @@ -211,11 +258,13 @@ stateDiagram-v2 | 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** +**Invalid / bad block** — "stall" below means the **persisted `final_stall`** +(§4a). A bad block seen only in `adopted` sets the in-memory `head_stall` (not +persisted); it becomes a persisted `final_stall` only if it finalizes. | # | Scenario | Handling | Expected | | --- | ------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ----------------------------------------------- | -| 8 | Authorized sequencer posts a block with an invalid state transition | `apply_block` → `Parked(StateTransition)`; freeze head tip, record stall, mark processed | park-and-skip; no apply, no halt | +| 8 | Authorized sequencer posts a block with an invalid state transition | head: `apply_block` → `Parked`, set `head_stall` (in-memory), no persist. 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 | From 80e0af241b36aa2cbd7296b15068063c45ff5ae8 Mon Sep 17 00:00:00 2001 From: erhant Date: Wed, 8 Jul 2026 17:40:20 +0300 Subject: [PATCH 3/7] feat(chain_state): initial preps, one more design fix --- Cargo.lock | 14 ++ Cargo.toml | 2 + lez/chain_state/Cargo.toml | 21 +++ lez/chain_state/DESIGN.md | 99 +++++----- lez/chain_state/src/apply.rs | 280 ++++++++++++++++++++++++++++ lez/chain_state/src/ingest_error.rs | 67 +++++++ lez/chain_state/src/lib.rs | 17 ++ lez/chain_state/src/stall_reason.rs | 25 +++ 8 files changed, 480 insertions(+), 45 deletions(-) create mode 100644 lez/chain_state/Cargo.toml create mode 100644 lez/chain_state/src/apply.rs create mode 100644 lez/chain_state/src/ingest_error.rs create mode 100644 lez/chain_state/src/lib.rs create mode 100644 lez/chain_state/src/stall_reason.rs diff --git a/Cargo.lock b/Cargo.lock index 0bd194bb..954d98fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1349,6 +1349,20 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "chain_state" +version = "0.1.0" +dependencies = [ + "anyhow", + "common", + "lee", + "logos-blockchain-zone-sdk", + "serde", + "serde_json", + "testnet_initial_state", + "thiserror 2.0.18", +] + [[package]] name = "chkstk_stub" version = "0.1.0" 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..337367cc --- /dev/null +++ b/lez/chain_state/Cargo.toml @@ -0,0 +1,21 @@ +[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-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 index 3e0bc384..600fb23f 100644 --- a/lez/chain_state/DESIGN.md +++ b/lez/chain_state/DESIGN.md @@ -86,8 +86,7 @@ struct ChainState { final_tip: Option, head_state: V03State, // final_state + applied head blocks head_blocks: Vec, // ordered, above final_tip - final_stall: Option, // persisted to RocksDB — see §4a - head_stall: Option, // in-memory only — recomputed from the stream on restart + final_stall: Option, // the one stall — persisted to RocksDB. See §4a } struct HeadEntry { this_msg: MsgId, block: Block } @@ -103,8 +102,8 @@ 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, set **`head_stall`** (in-memory `StallReason`) and freeze the head tip - at the last valid block — do **not** persist. + 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. @@ -113,52 +112,62 @@ Operations: - `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** - `StallReason` written to disk (see §4a). -- `rollback_orphan(this_msg)` — drop from that entry forward, re-derive head; - clears `head_stall` if the re-derived head is clean. -- `status() -> { final_height, head_height, head_stall, final_stall }` — for RPC/UI. + 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. Two stalls — persisted vs in-memory +### 4a. One stall — `final_stall`, persisted -Both tiers carry a `StallReason` (`final_stall`, `head_stall`); they are equally -informative. The difference is **persistence**, which follows from durability. L1 -finality is about inscription **canonicality**, not LEZ-block **content** -validity, so an authorized sequencer can get a content-invalid block finalized — -which is why both tiers can meet an invalid block in the first place. +There is a single stall, `final_stall`, on the final tier. The head tier does +**not** carry its own stall, and this is deliberate. -- **`head_stall` — in-memory only.** The head is reorg-able and re-derived from - every `channel_update`. An invalid adopted block is transient: it is either - orphaned (a competing valid block at the same height wins) or it finalizes. We - set `head_stall` for observability (RPC/UI can show "head blocked at `N`: - StateTransition at tx 3") and freeze the head tip at the last valid block, but - we do **not** write it to disk — on restart the head is rebuilt from the stream, - so a persisted head stall would be redundant and could go stale. Because the - next adopted block chains on the bad one, it fails validation too, so the head - stays frozen until a valid block (competing or post-reorg) is adopted. -- **`final_stall` — persisted.** The final tier is irreversible. If an invalid - block *finalizes*, the node is durably stuck until a valid successor (built by - honest sequencers on the last valid parent) finalizes. This must survive - restart: the startup chain-consistency / anchor check reads it, it is what we - surface as `Stalled`, and it is the signal the committee acts on to evict a bad - sequencer. +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. -An invalid block **migrates the problem head→final** when it finalizes: `head_stall` -is set at `N` first; once `N+1(bad)` finalizes, `apply_finalized` fails and records -the persisted `final_stall`. Both tiers end up stuck at `N` consistently, and both -recover when `N+1′` finalizes. The indexer (final tier only) never sets -`head_stall`, so it behaves exactly as it does today. +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 (`head_stall` set) on a peer's bad block, we build on that frozen valid +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 @@ -203,7 +212,7 @@ flowchart TD VAL --> OUT{"AcceptOutcome"} OUT -->|Applied| APP["append this_msg+block to head,
advance head tip, clear stall"] OUT -->|AlreadyApplied| SKIP - OUT -->|"Parked(err)"| PARK["set head_stall (in-memory),
freeze head tip — do NOT apply.
Not persisted"] + OUT -->|"Parked(err)"| PARK["freeze head tip — do NOT apply.
No stall recorded (self-heals
via reorg/finalization)"] SKIP --> FIN APP --> FIN PARK --> FIN @@ -231,11 +240,11 @@ stateDiagram-v2 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 — persisted, survives restart. + final_stall — the one stall. Persisted, survives restart. Head-tier bad blocks do NOT enter this state: - they set head_stall (in-memory) and self-heal - via reorg/finalization. Producer (on our turn) - builds on the last valid tip either way. + 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 ``` @@ -258,13 +267,13 @@ stateDiagram-v2 | 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 **persisted `final_stall`** -(§4a). A bad block seen only in `adopted` sets the in-memory `head_stall` (not -persisted); it becomes a persisted `final_stall` only if it finalizes. +**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`, set `head_stall` (in-memory), no persist. If it finalizes: persisted `final_stall` | park-and-skip; no apply, no halt | +| 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 | diff --git a/lez/chain_state/src/apply.rs b/lez/chain_state/src/apply.rs new file mode 100644 index 00000000..38e49b53 --- /dev/null +++ b/lez/chain_state/src/apply.rs @@ -0,0 +1,280 @@ +//! 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 last successfully applied block: the parent the next block must chain on. +/// +/// Only what validation needs today (`block_id` + `hash`). The two-tier +/// `ChainState`'s `final_tip` will extend this with the inscription `l1_slot` +/// when the anchor layer lands; the slot is currently tracked separately. +#[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`. +/// +/// Validation runs first and touches nothing. Application then mutates `state` +/// in place and can fail partway, so callers pass a scratch clone and commit it +/// only when this returns `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/ingest_error.rs b/lez/chain_state/src/ingest_error.rs new file mode 100644 index 00000000..bd377a0e --- /dev/null +++ b/lez/chain_state/src/ingest_error.rs @@ -0,0 +1,67 @@ +use common::HashType; +use serde::{Deserialize, Serialize}; + +/// Why an L2 block from the channel could not be applied. +/// +/// 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}")] + /// Here we store the error string that is derived from [`borsh::from_slice`]'s [`Err`]. + Deserialize(String), + #[error("Unexpected block id: expected {expected}, got {got}")] + UnexpectedBlockId { expected: u64, got: u64 }, + #[error("Broken chain link: expected prev {expected_prev}, got {got_prev}")] + BrokenChainLink { + expected_prev: HashType, + got_prev: HashType, + }, + #[error("Block hash mismatch: computed {computed}, header {header}")] + HashMismatch { + computed: HashType, + header: HashType, + }, + #[error("Block has no transactions")] + EmptyBlock, + #[error("Last transaction must be the public clock invocation for the block timestamp")] + InvalidClockTransaction, + #[error("Genesis block must contain only public transactions")] + NonPublicGenesisTransaction, + #[error("State transition failed at transaction {tx_index}: {reason}")] + StateTransition { + /// Index of the failing transaction within the block body. + tx_index: u64, + /// Reason string from `lee::Error` to `anyhow::Error` to `{:#}`. + /// + /// This is required because `lee::Error` is not `Clone + Serialize + Deserialize`, so we + /// cannot store it directly. + reason: String, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn serializes_and_round_trips_externally_tagged() { + let err = BlockIngestError::UnexpectedBlockId { + expected: 5, + got: 7, + }; + let value = serde_json::to_value(&err).expect("serialize"); + assert_eq!( + value, + serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } }) + ); + let back: BlockIngestError = serde_json::from_value(value).expect("deserialize"); + assert!(matches!( + back, + BlockIngestError::UnexpectedBlockId { + expected: 5, + got: 7 + } + )); + } +} diff --git a/lez/chain_state/src/lib.rs b/lez/chain_state/src/lib.rs new file mode 100644 index 00000000..67efa146 --- /dev/null +++ b/lez/chain_state/src/lib.rs @@ -0,0 +1,17 @@ +//! Shared, storage-free chain-state core for the LEZ sequencer and indexer. +//! +//! Hosts the single validate-then-apply entry point ([`apply_block`]) plus the +//! shared types ([`BlockIngestError`], [`StallReason`], [`Tip`], +//! [`AcceptOutcome`]) that both the sequencer and the indexer build on. The +//! crate performs no I/O: callers own their storage and drive the +//! `scratch → persist → commit` ordering around these primitives. +//! +//! See `DESIGN.md` in this crate for the two-tier chain-state model this backs. + +pub mod apply; +pub mod ingest_error; +pub mod stall_reason; + +pub use apply::{AcceptOutcome, Tip, apply_block}; +pub use ingest_error::BlockIngestError; +pub use stall_reason::StallReason; diff --git a/lez/chain_state/src/stall_reason.rs b/lez/chain_state/src/stall_reason.rs new file mode 100644 index 00000000..c5118fcc --- /dev/null +++ b/lez/chain_state/src/stall_reason.rs @@ -0,0 +1,25 @@ +use common::HashType; +use logos_blockchain_zone_sdk::Slot; +use serde::{Deserialize, Serialize}; + +use crate::ingest_error::BlockIngestError; + +/// Diagnostic record of the first block that broke the L2 chain. +/// +/// The block-derived fields are `None` for a deserialize break (no header was +/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at. +/// `first_seen` is the breaking block's L2 timestamp (`None` for a deserialize break). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StallReason { + pub block_id: Option, + pub block_hash: Option, + pub prev_block_hash: Option, + pub l1_slot: Slot, + pub error: BlockIngestError, + pub first_seen: Option, + /// Number of later non-chaining blocks (orphans, since the tip is frozen). + /// + /// TODO: We could store a different "branch" of blocks following this break, but for now we + /// just count them. + pub orphans_since: u64, +} From fbaf50ed1ea0e533ce4b39bbc5d3955661cdb149 Mon Sep 17 00:00:00 2001 From: erhant Date: Wed, 8 Jul 2026 17:49:21 +0300 Subject: [PATCH 4/7] chore(chain_state): doc fix --- lez/chain_state/src/apply.rs | 24 +++++++++++++----------- lez/chain_state/src/lib.rs | 12 +++--------- 2 files changed, 16 insertions(+), 20 deletions(-) diff --git a/lez/chain_state/src/apply.rs b/lez/chain_state/src/apply.rs index 38e49b53..8a791a78 100644 --- a/lez/chain_state/src/apply.rs +++ b/lez/chain_state/src/apply.rs @@ -11,11 +11,8 @@ use lee::{GENESIS_BLOCK_ID, V03State}; use crate::ingest_error::BlockIngestError; -/// The last successfully applied block: the parent the next block must chain on. -/// -/// Only what validation needs today (`block_id` + `hash`). The two-tier -/// `ChainState`'s `final_tip` will extend this with the inscription `l1_slot` -/// when the anchor layer lands; the slot is currently tracked separately. +/// 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, @@ -34,9 +31,7 @@ pub enum AcceptOutcome { /// Validates `block` against `tip`, then applies it to `state`. /// -/// Validation runs first and touches nothing. Application then mutates `state` -/// in place and can fail partway, so callers pass a scratch clone and commit it -/// only when this returns `Ok`. +/// Mutates `state` in place, so callers pass a scratch clone and commit on `Ok`. pub fn apply_block( tip: Option<&Tip>, block: &Block, @@ -181,7 +176,10 @@ mod tests { let err = apply_block(None, &block, &mut state).expect_err("should reject"); assert!(matches!( err, - BlockIngestError::UnexpectedBlockId { expected: 1, got: 2 } + BlockIngestError::UnexpectedBlockId { + expected: 1, + got: 2 + } )); } @@ -193,10 +191,14 @@ mod tests { // 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"); + let err = + apply_block(Some(&tip_of(&genesis)), &bad, &mut state).expect_err("should reject"); assert!(matches!( err, - BlockIngestError::UnexpectedBlockId { expected: 2, got: 3 } + BlockIngestError::UnexpectedBlockId { + expected: 2, + got: 3 + } )); } diff --git a/lez/chain_state/src/lib.rs b/lez/chain_state/src/lib.rs index 67efa146..065f643d 100644 --- a/lez/chain_state/src/lib.rs +++ b/lez/chain_state/src/lib.rs @@ -1,12 +1,6 @@ -//! Shared, storage-free chain-state core for the LEZ sequencer and indexer. -//! -//! Hosts the single validate-then-apply entry point ([`apply_block`]) plus the -//! shared types ([`BlockIngestError`], [`StallReason`], [`Tip`], -//! [`AcceptOutcome`]) that both the sequencer and the indexer build on. The -//! crate performs no I/O: callers own their storage and drive the -//! `scratch → persist → commit` ordering around these primitives. -//! -//! See `DESIGN.md` in this crate for the two-tier chain-state model this backs. +//! 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 mod apply; pub mod ingest_error; From 75f2ab00cabf243cc6a8cf78173f5369d0ea6e03 Mon Sep 17 00:00:00 2001 From: erhant Date: Wed, 8 Jul 2026 18:35:10 +0300 Subject: [PATCH 5/7] refactor!(indexer): migr to `chain_state` --- Cargo.lock | 2 +- lez/indexer/core/Cargo.toml | 2 +- lez/indexer/core/src/block_store.rs | 132 +--------------------- lez/indexer/core/src/chain_consistency.rs | 3 +- lez/indexer/core/src/ingest_error.rs | 67 ----------- lez/indexer/core/src/lib.rs | 7 +- lez/indexer/core/src/stall_reason.rs | 25 ---- lez/indexer/core/src/status.rs | 6 +- 8 files changed, 11 insertions(+), 233 deletions(-) delete mode 100644 lez/indexer/core/src/ingest_error.rs delete mode 100644 lez/indexer/core/src/stall_reason.rs diff --git a/Cargo.lock b/Cargo.lock index 954d98fb..64f12ceb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3866,6 +3866,7 @@ dependencies = [ "async-stream", "authenticated_transfer_core", "borsh", + "chain_state", "common", "futures", "humantime-serde", @@ -3879,7 +3880,6 @@ dependencies = [ "storage", "tempfile", "testnet_initial_state", - "thiserror 2.0.18", "tokio", "url", ] 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/ingest_error.rs b/lez/indexer/core/src/ingest_error.rs deleted file mode 100644 index bda53497..00000000 --- a/lez/indexer/core/src/ingest_error.rs +++ /dev/null @@ -1,67 +0,0 @@ -use common::HashType; -use serde::{Deserialize, Serialize}; - -/// Why the indexer could not apply an L2 block from the channel. -/// -/// Persisted in `RocksDB`, so every variant must have the following -/// traits: `Clone + Serialize + Deserialize`. -#[derive(Debug, Clone, Serialize, Deserialize, thiserror::Error)] -pub enum BlockIngestError { - #[error("Failed to deserialize L2 block: {0}")] - /// Here we store the error string that is derived from [`borsh::from_slice`]'s [`Err`]. - Deserialize(String), - #[error("Unexpected block id: expected {expected}, got {got}")] - UnexpectedBlockId { expected: u64, got: u64 }, - #[error("Broken chain link: expected prev {expected_prev}, got {got_prev}")] - BrokenChainLink { - expected_prev: HashType, - got_prev: HashType, - }, - #[error("Block hash mismatch: computed {computed}, header {header}")] - HashMismatch { - computed: HashType, - header: HashType, - }, - #[error("Block has no transactions")] - EmptyBlock, - #[error("Last transaction must be the public clock invocation for the block timestamp")] - InvalidClockTransaction, - #[error("Genesis block must contain only public transactions")] - NonPublicGenesisTransaction, - #[error("State transition failed at transaction {tx_index}: {reason}")] - StateTransition { - /// Index of the failing transaction within the block body. - tx_index: u64, - /// Reason string from `lee::Error` to `anyhow::Error` to `{:#}`. - /// - /// This is required because `lee::Error` is not `Clone + Serialize + Deserialize`, so we - /// cannot store it directly. - reason: String, - }, -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn serializes_and_round_trips_externally_tagged() { - let err = BlockIngestError::UnexpectedBlockId { - expected: 5, - got: 7, - }; - let value = serde_json::to_value(&err).expect("serialize"); - assert_eq!( - value, - serde_json::json!({ "UnexpectedBlockId": { "expected": 5, "got": 7 } }) - ); - let back: BlockIngestError = serde_json::from_value(value).expect("deserialize"); - assert!(matches!( - back, - BlockIngestError::UnexpectedBlockId { - expected: 5, - got: 7 - } - )); - } -} 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/stall_reason.rs b/lez/indexer/core/src/stall_reason.rs deleted file mode 100644 index c5118fcc..00000000 --- a/lez/indexer/core/src/stall_reason.rs +++ /dev/null @@ -1,25 +0,0 @@ -use common::HashType; -use logos_blockchain_zone_sdk::Slot; -use serde::{Deserialize, Serialize}; - -use crate::ingest_error::BlockIngestError; - -/// Diagnostic record of the first block that broke the L2 chain. -/// -/// The block-derived fields are `None` for a deserialize break (no header was -/// ever parsed). `l1_slot` is the L1 slot the breaking inscription was read at. -/// `first_seen` is the breaking block's L2 timestamp (`None` for a deserialize break). -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StallReason { - pub block_id: Option, - pub block_hash: Option, - pub prev_block_hash: Option, - pub l1_slot: Slot, - pub error: BlockIngestError, - pub first_seen: Option, - /// Number of later non-chaining blocks (orphans, since the tip is frozen). - /// - /// TODO: We could store a different "branch" of blocks following this break, but for now we - /// just count them. - pub orphans_since: u64, -} 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), From 1fa5cb795cc7a3613c10a9f472484ae948f85980 Mon Sep 17 00:00:00 2001 From: erhant Date: Thu, 9 Jul 2026 17:47:48 +0300 Subject: [PATCH 6/7] feat(chain_state): implement two-tip `ChainState` struct --- Cargo.lock | 1 + lez/chain_state/Cargo.toml | 1 + lez/chain_state/src/chain.rs | 417 +++++++++++++++++++++++++++++++++++ lez/chain_state/src/lib.rs | 10 +- 4 files changed, 425 insertions(+), 4 deletions(-) create mode 100644 lez/chain_state/src/chain.rs diff --git a/Cargo.lock b/Cargo.lock index 64f12ceb..65e0cde7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1356,6 +1356,7 @@ dependencies = [ "anyhow", "common", "lee", + "logos-blockchain-core", "logos-blockchain-zone-sdk", "serde", "serde_json", diff --git a/lez/chain_state/Cargo.toml b/lez/chain_state/Cargo.toml index 337367cc..1fcc052e 100644 --- a/lez/chain_state/Cargo.toml +++ b/lez/chain_state/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] common.workspace = true lee.workspace = true +logos-blockchain-core.workspace = true logos-blockchain-zone-sdk.workspace = true anyhow.workspace = true 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/chain_state/src/lib.rs b/lez/chain_state/src/lib.rs index 065f643d..c4725ec3 100644 --- a/lez/chain_state/src/lib.rs +++ b/lez/chain_state/src/lib.rs @@ -2,10 +2,12 @@ //! the [`apply_block`] entry point plus [`BlockIngestError`], [`StallReason`], //! [`Tip`], and [`AcceptOutcome`]. See `DESIGN.md` for the two-tier model. -pub mod apply; -pub mod ingest_error; -pub mod stall_reason; - 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; From d90dbd83b2828feba470e442fd13341b15def653 Mon Sep 17 00:00:00 2001 From: erhant Date: Thu, 9 Jul 2026 18:06:08 +0300 Subject: [PATCH 7/7] feat!(sequencer): add `is_our_turn` --- lez/sequencer/core/src/block_publisher.rs | 19 +++++++++++++++++-- lez/sequencer/core/src/lib.rs | 6 ++++++ lez/sequencer/core/src/mock.rs | 4 ++++ lez/sequencer/service/src/lib.rs | 15 +++++++-------- 4 files changed, 34 insertions(+), 10 deletions(-) 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"); } }