logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-11 00:09:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge d90dbd83b2828feba470e442fd13341b15def653 into 9b2b4a3d8ca714fa089df634a230d1e0764dcdd7

Browse Source
This commit is contained in:
erhant 2026-07-09 15:06:14 +00:00 committed by GitHub
commit 08bc914d1d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 1105 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
Cargo.lock generated
View File

@ -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",
]

2
Cargo.toml
View File

@ -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" }

22
lez/chain_state/Cargo.toml Normal file
View File

@ -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

306
lez/chain_state/DESIGN.md Normal file
View File

@ -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<Tip>,
head_state: V03State, // final_state + applied head blocks
head_blocks: Vec<HeadEntry>, // ordered, above final_tip
final_stall: Option<StallReason>, // the one stall — persisted to RocksDB. See §4a
}
struct HeadEntry { this_msg: MsgId, block: Block }
```
The **head** tier is a MsgId-keyed chain (adopted/orphaned reference
`this_msg`/`parent_msg`); the **final** tier is block-id-keyed. `apply_block`
validation stays LEZ-level (`block_id` + `prev_block_hash`) — the two chains run
in parallel and must agree, so we validate via `apply_block` _and_ track `MsgId`
for revert correlation.
Operations:
- `apply_adopted(inscription) -> AcceptOutcome` — dedup by `this_msg` against our
outbox, else `apply_block` on the head tip; on success push `HeadEntry`. On
failure, **do nothing durable**: the head tip simply stays at the last valid
block. No stall is recorded (see §4a) — the head self-heals from the stream.
- `apply_channel_update(orphaned, adopted)` — revert every `orphaned` by
`this_msg`, re-derive `head_state` (clone `final_state`, replay survivors),
then apply every `adopted` in order. Atomic per event.
- `finalize_up_to(block_id)` — move `head_blocks` up to `block_id` into
`final_state` (already validated; a move, not a re-apply).
- `apply_finalized(inscription)` — steady state: if present in head by
`this_msg`, `finalize_up_to`; cold-start backfill (not in head):
`apply_block` directly to `final_state`, mirror into head. If a finalized block
fails to apply, set **`final_stall`** and persist it — this is the **only** stall
(see §4a).
- `rollback_orphan(this_msg)` — drop from that entry forward, re-derive head.
- `status() -> { final_height, head_height, final_stall }` — for RPC/UI. A derived
"head blocked" indicator can be computed on demand (see §4a) without persisting.
For the **indexer** (finalized-only `next_messages` stream), `head_blocks` stays
empty and `head == final`; it exercises only `apply_finalized`. The **sequencer**
uses both tiers from day one.
### 4a. One stall — `final_stall`, persisted
There is a single stall, `final_stall`, on the final tier. The head tier does
**not** carry its own stall, and this is deliberate.
The head and the final tier never represent two independent problems: a block
always reaches the head first (as `adopted`) and only later the final tier (as
`finalized`), so a would-be "head stall" is just the earlier, provisional sighting
of the exact block that `final_stall` records durably if it finalizes — the same
event modeled twice.
And the head does not need a recorded reason to freeze. **The tip-freeze is
intrinsic**: not applying a bad block is what freezes the tip; no marker is
required. The head's freeze is also **transient and self-healing** — the bad block
is either orphaned (a competing valid block at the same height wins and applies on
its own) or it finalizes. Subsequent adopted blocks that chain on the bad one fail
validation by themselves, and the producer builds on the head tip regardless of
any marker. So a persisted head stall would be redundant (re-derived from the
stream on restart) and buys no behavior.
`final_stall` is the stall that does real work:
- The **indexer already requires it** and ships it today. The indexer has only a
final tier (finalized-only stream, no head); its startup chain-consistency /
anchor check reads the persisted stall to know where it is parked. The shared
`final_stall` serves that unchanged.
- It **survives restart** and is what we surface as `Stalled`.
- "A bad block **finalized**" is the only irreversible, actionable condition — the
signal the committee acts on to evict a bad sequencer. A provisional head block
that may vanish on the next reorg is not something to evict over.
The one thing we forgo is an *early warning* that a sequencer is posting garbage
before it finalizes. That condition frequently self-heals via reorg, so alarming on
it is mostly noise; if wanted, it is a **derived, non-persisted** indicator (e.g.
"k adopted inscriptions above the head tip remain unapplied"), computed on demand —
not a second `StallReason` in the struct.
So the sequencer and the indexer share exactly one stall concept, keeping the two
consumers uniform.
### 4b. Producer contract — write on turn, build on last valid
The sequencer publishes **only on its own turn** (the SDK queues out-of-turn
publishes). When it is our turn we build the next block on the **current head
tip**, which is by construction the last validly-applied block. So if the head is
frozen on a peer's bad block, we build on that frozen valid
tip — the same parent every honest sequencer chooses — and thereby skip the bad
block rather than extend it. A parked node keeps following peers' valid blocks as
they arrive; the moment it also gets a turn, it produces the next valid block on
its last valid tip. Net: parking never stops us from producing correctly on our
turn.
## 5. Event → tier mapping
`Event::BlocksProcessed { checkpoint, channel_update: { orphaned, adopted }, finalized }`:
| Input | Source | Effect |
| --------------------- | ---------------------------------------------- | --------------------------------------------------- |
| adopted inscription | `channel_update.adopted` | validate + apply to **head** |
| orphaned inscription | `channel_update.orphaned` | revert from **head**, return txs to mempool |
| finalized inscription | `finalized[].ops` (`FinalizedOp::Inscription`) | move head→**final**, or apply directly on backfill |
| own publish | publish-return | optimistically apply to **head**, record `this_msg` |
**Golden rules:** (1) validation is deterministic, so every honest node makes the
same accept/park decision. (2) An invalid block is _processed but discarded_ —
never applied, never halts the node. (3) Finalized is never reverted. (4) We
rebuild orphaned blocks ourselves; we do not trust the SDK's republish (it keeps
stale LEZ contents — prev-hash, tx selection, and resulting state were all
computed against the old parent).
---
## 6. Scenarios
### Processing one `BlocksProcessed` event
```mermaid
flowchart TD
EV["Event::BlocksProcessed"] --> ORPH{"orphaned<br/>non-empty?"}
ORPH -->|yes| REV["For each orphaned by this_msg:<br/>drop from head_blocks,<br/>return its txs to mempool"]
REV --> RED["Re-derive head_state:<br/>clone final_state, replay survivors"]
ORPH -->|no| ADO
RED --> ADO{"adopted<br/>non-empty?"}
ADO -->|"yes, in order"| DEDUP{"this_msg in<br/>our outbox?"}
ADO -->|no| FIN
DEDUP -->|"yes (our own)"| SKIP["skip: already applied optimistically"]
DEDUP -->|no| VAL["apply_block on head tip"]
VAL --> OUT{"AcceptOutcome"}
OUT -->|Applied| APP["append this_msg+block to head,<br/>advance head tip, clear stall"]
OUT -->|AlreadyApplied| SKIP
OUT -->|"Parked(err)"| PARK["freeze head tip — do NOT apply.<br/>No stall recorded (self-heals<br/>via reorg/finalization)"]
SKIP --> FIN
APP --> FIN
PARK --> FIN
FIN{"finalized<br/>inscriptions?"}
FIN -->|"already in head (steady state)"| MOVE["finalize_up_to:<br/>move head→final, trim head_blocks"]
FIN -->|"not in head (cold-start backfill)"| DIRECT["apply_block directly to final"]
DIRECT --> DOK{"applied?"}
DOK -->|yes| MIRROR["mirror into head"]
DOK -->|"no (invalid finalized)"| FSTALL["record StallReason on FINAL,<br/>freeze final tip"]
FIN -->|none| CP
MOVE --> CP
MIRROR --> CP
FSTALL --> CP
CP["persist checkpoint atomically"]
```
### Park / recovery status
```mermaid
stateDiagram-v2
[*] --> Syncing
Syncing --> CaughtUp: stream drained, no stall
CaughtUp --> Syncing: new adopted / finalized arrives
Syncing --> Parked: invalid block FINALIZED (apply_finalized fails)
Parked --> Parked: further non-chaining finalized blocks (orphans_since++)
Parked --> Syncing: valid successor finalizes on frozen final tip → stall cleared
note right of Parked
final_stall — the one stall. Persisted, survives restart.
Head-tier bad blocks do NOT enter this state:
the head tip freezes intrinsically (no stall) and
self-heals via reorg/finalization. Producer (on our
turn) builds on the last valid tip either way.
end note
```
### Scenario table
**Normal flow**
| # | Scenario | Handling | Expected |
| --- | ---------------------------------------- | ------------------------------------------------------------- | ----------------------------------- |
| 1 | Adopted block chains cleanly on head tip | `apply_block` → `Applied`; append `{this_msg, block}` to head | head advances; converges with peers |
| 2 | Our own block comes back in `adopted` | dedup by `this_msg` against outbox → skip | no double-apply |
| 3 | Adopted block later finalizes | `finalize_up_to` moves head→final, trims `head_blocks` | final advances; no re-apply |
| 4 | Re-delivery of an already-applied block | id ≤ tip & stored hash matches → `AlreadyApplied` | idempotent, no state change |
**Reorg / orphan**
| # | Scenario | Handling | Expected |
| --- | ---------------------------------------------------------- | -------------------------------------------------------------------------------------------- | ----------------------------------------------- |
| 5 | Our block orphaned at turn handoff (stale-parent race) | revert by `this_msg`, return txs to mempool, **rebuild** on new head tip (not SDK republish) | our txs re-queued; next block on correct parent |
| 6 | Batch reorg: some `orphaned` + some `adopted` in one event | revert all orphaned, re-derive head, then apply all adopted in order | deterministic convergence |
| 7 | Orphan chain (parent transitively off canonical) | SDK surfaces all affected as `orphaned`; revert each, replay survivors | head_state matches new canonical branch |
**Invalid / bad block** — "stall" below means the one **persisted `final_stall`**
(§4a). A bad block seen only in `adopted` records **no** stall — the head tip just
freezes and self-heals; it becomes a `final_stall` only if it finalizes.
| # | Scenario | Handling | Expected |
| --- | ------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ----------------------------------------------- |
| 8 | Authorized sequencer posts a block with an invalid state transition | head: `apply_block` → `Parked`, freeze head tip, no stall recorded. If it finalizes: persisted `final_stall` | park-and-skip; no apply, no halt |
| 9 | Broken chain link / hash mismatch / unexpected id in adopted | `Parked(BrokenChainLink / HashMismatch / UnexpectedBlockId)`; same park | frozen tip; peers park identically |
| 10 | Undeserializable inscription payload | park with `Deserialize` (no header); processing advances | recover when a valid block chains on frozen tip |
| 11 | Valid successor after a park (recovery) | block chaining on frozen tip → `Applied` → clear stall | head resumes automatically; no divergence |
| 12 | Further non-chaining blocks while parked | keep first `StallReason`, bump `orphans_since` | original cause preserved; still parked |
**Producing while parked**
| # | Scenario | Handling | Expected |
| --- | ----------------------------------------------- | ----------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ |
| 13 | It's our turn but head is parked on a bad block | producer builds on the **frozen valid tip** (head tip = last valid), skipping the invalid block | we emit the next valid block on the same parent honest peers use — chain moves on our turn |
**Startup / backfill**
| # | Scenario | Handling | Expected |
| --- | --------------------------------------------------- | -------------------------------------------------------------------------------------------------- | ------------------------------------- |
| 14 | Cold start / reconnect backfill | history via `finalized`, empty `channel_update`; apply directly to final + mirror head | head == final until live deltas start |
| 15 | Local store belongs to a different chain (L1 reset) | anchor-based `chain_consistency` check at startup: wipe+reindex if `allow_chain_reset`, else error | no silent divergence |
## 7. Invariants
Should-never-happen conditions — assert/log, don't silently absorb:
- An `orphaned` entry never references a block at or below the **final** tip —
finalized is irreversible. If seen, it is a bug.
- `head` tip ≥ `final` tip at all times; `head_blocks` holds exactly the blocks
between them.
- After processing any event, `head_state == final_state` replayed through
`head_blocks` (the re-derivation is the source of truth).
- A parked node's frozen tip is identical across all honest nodes for the same
invalid block (deterministic validation).

282
lez/chain_state/src/apply.rs Normal file
View File

@ -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);
}
}

417
lez/chain_state/src/chain.rs Normal file
View File

@ -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<Tip>,
head_state: V03State,
head_blocks: Vec<HeadEntry>,
final_stall: Option<StallReason>,
}
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<Tip>) -> 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<Tip> {
self.head_blocks
.last()
.map(|entry| tip_of(&entry.block))
.or_else(|| self.final_tip.clone())
}
#[must_use]
pub fn final_tip(&self) -> Option<Tip> {
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<AcceptOutcome> {
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<HeadEntry> = 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);
}
}

6
lez/indexer/core/src/ingest_error.rs → lez/chain_state/src/ingest_error.rs
View File

@ -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}")]

13
lez/chain_state/src/lib.rs Normal file
View File

@ -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;

0
lez/indexer/core/src/stall_reason.rs → lez/chain_state/src/stall_reason.rs
View File

2
lez/indexer/core/Cargo.toml
View File

@ -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

132
lez/indexer/core/src/block_store.rs
View File

@ -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<RocksDBIO>,
@ -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")

3
lez/indexer/core/src/chain_consistency.rs
View File

@ -305,8 +305,7 @@ mod tests {
use super::*;
use crate::{
BlockIngestError,
block_store::AcceptOutcome,
AcceptOutcome, BlockIngestError,
config::{ChannelId, ClientConfig, IndexerConfig},
};

7
lez/indexer/core/src/lib.rs
View File

@ -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)]

6
lez/indexer/core/src/status.rs
View File

@ -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),

19
lez/sequencer/core/src/block_publisher.rs
View File

@ -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<WithdrawArg>) -> 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<WithdrawArg>)>,
turn_rx: watch::Receiver<TurnNotification>,
// Aborts the drive task when the last clone is dropped.
_drive_task: Arc<DriveTaskGuard>,
}
@ -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<WithdrawArg>)>(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`.

6
lez/sequencer/core/src/lib.rs
View File

@ -534,6 +534,12 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
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)

4
lez/sequencer/core/src/mock.rs
View File

@ -48,4 +48,8 @@ impl BlockPublisherTrait for MockBlockPublisher {
fn channel_id(&self) -> ChannelId {
self.channel_id
}
fn is_our_turn(&self) -> bool {
true
}
}

15
lez/sequencer/service/src/lib.rs
View File

@ -172,16 +172,15 @@ async fn main_loop(seq_core: Arc<Mutex<SequencerCore>>, 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");
}
}