diff --git a/lez/chain_state/src/chain.rs b/lez/chain_state/src/chain.rs index f0b4eddb..fe4e36c5 100644 --- a/lez/chain_state/src/chain.rs +++ b/lez/chain_state/src/chain.rs @@ -53,6 +53,13 @@ impl ChainState { &self.head_state } + /// Mutable access to the head state. Bypasses the `head_blocks` invariant, so + /// it is meant for tests and low-level callers. + #[must_use] + pub const fn head_state_mut(&mut self) -> &mut V03State { + &mut self.head_state + } + #[must_use] pub const fn final_state(&self) -> &V03State { &self.final_state diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 45467d49..d0c9aaa0 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -65,14 +65,12 @@ impl DepositMetadata { } pub struct SequencerCore { - state: lee::V03State, - /// Two-tier follow state fed by the publisher's `on_follow` sink. Shared with - /// the drive task. Passive today; production moves onto its head next. + /// Two-tier chain state: production builds on its head; the publisher's + /// `on_follow` sink feeds adopted/orphaned/finalized peer blocks into it. chain: Arc>, store: SequencerStore, mempool: MemPool<(TransactionOrigin, LeeTransaction)>, sequencer_config: SequencerConfig, - chain_height: u64, block_publisher: BP, } @@ -143,7 +141,7 @@ impl SequencerCore { .expect("Failed to read latest block meta from store"); let chain = Arc::new(Mutex::new(ChainState::from_final( - state.clone(), + state, Some(Tip { block_id: latest_block_meta.id, hash: latest_block_meta.hash, @@ -202,11 +200,9 @@ impl SequencerCore { } let sequencer_core = Self { - state, chain, store, mempool, - chain_height: latest_block_meta.id, sequencer_config: config, block_publisher, }; @@ -345,7 +341,10 @@ impl SequencerCore { } /// Feed one channel delta into the follow state: revert orphaned, apply - /// adopted, then finalize. Passive today — production still uses `self.state`. + /// adopted, then finalize. Production builds on this same head. + /// + /// TODO: persist adopted/finalized peer blocks to the store here so restart and RPC see the full canonical chain, not just our own blocks. + /// Open question: will peers communicate in another way? fn on_follow(chain: Arc>) -> block_publisher::OnFollowSink { Box::new(move |update: block_publisher::FollowUpdate| { let chain = Arc::clone(&chain); @@ -390,26 +389,27 @@ impl SequencerCore { // Apply our own block to the head with the MsgId the publish assigned it, // so the head advances and the later adopted redelivery dedups. - self.chain - .lock() - .expect("chain state mutex poisoned") - .apply_adopted(this_msg, &block); + let head_state = { + let mut chain = self.chain.lock().expect("chain state mutex poisoned"); + chain.apply_adopted(this_msg, &block); + chain.head_state().clone() + }; self.store.update( &block, &deposit_event_ids, withdrawal_reconciliation_keys, - &self.state, + &head_state, )?; - Ok(self.chain_height) + Ok(block.header.block_id) } /// Validates and applies a single mempool transaction to the current state. /// Returns `Ok(true)` if the transaction was valid and applied, `Ok(false)` if /// it was skipped due to validation failure. fn apply_mempool_transaction( - &mut self, + state: &mut lee::V03State, origin: TransactionOrigin, tx: &LeeTransaction, block_height: u64, @@ -420,11 +420,7 @@ impl SequencerCore { let tx_hash = tx.hash(); match origin { TransactionOrigin::User => { - let validated_diff = match tx.validate_on_state( - &self.state, - block_height, - timestamp, - ) { + let validated_diff = match tx.validate_on_state(state, block_height, timestamp) { Ok(diff) => diff, Err(err) => { error!( @@ -438,7 +434,7 @@ impl SequencerCore { withdrawals.push(withdraw_data); } - self.state.apply_state_diff(validated_diff); + state.apply_state_diff(validated_diff); } TransactionOrigin::Sequencer => { let LeeTransaction::Public(public_tx) = tx else { @@ -449,7 +445,7 @@ impl SequencerCore { deposit_event_ids.push(deposit_op_id); } - self.state + state .transition_from_public_transaction(public_tx, block_height, timestamp) .context("Failed to execute sequencer-generated transaction")?; } @@ -462,7 +458,18 @@ impl SequencerCore { fn build_block_from_mempool(&mut self) -> Result { let now = Instant::now(); - let new_block_height = self.next_block_id(); + // Build on the head: its tip is the parent, its state the validation base. + let (prev_block_hash, new_block_height, mut working_state) = { + let chain = self.chain.lock().expect("chain state mutex poisoned"); + let tip = chain.head_tip(); + let height = tip.as_ref().map_or(GENESIS_BLOCK_ID, |head| { + head.block_id + .checked_add(1) + .expect("block id should not overflow") + }); + let prev = tip.map_or(HashType([0; 32]), |head| head.hash); + (prev, height, chain.head_state().clone()) + }; let mut valid_transactions = Vec::new(); let mut deposit_event_ids = Vec::new(); @@ -471,11 +478,6 @@ impl SequencerCore { let max_block_size = usize::try_from(self.sequencer_config.max_block_size.as_u64()) .expect("`max_block_size` should fit into usize"); - let latest_block_meta = self - .store - .latest_block_meta() - .context("Failed to get latest block meta from store")?; - let new_block_timestamp = u64::try_from(chrono::Utc::now().timestamp_millis()) .expect("Timestamp must be positive"); @@ -494,7 +496,7 @@ impl SequencerCore { let temp_hashable_data = HashableBlockData { block_id: new_block_height, transactions: temp_valid_transactions, - prev_block_hash: latest_block_meta.hash, + prev_block_hash, timestamp: new_block_timestamp, }; @@ -511,7 +513,8 @@ impl SequencerCore { break; } - if self.apply_mempool_transaction( + if Self::apply_mempool_transaction( + &mut working_state, origin, &tx, new_block_height, @@ -527,7 +530,7 @@ impl SequencerCore { } } - self.state + working_state .transition_from_public_transaction(&clock_tx, new_block_height, new_block_timestamp) .context("Clock transaction failed. Aborting block production.")?; valid_transactions.push(clock_lee_tx); @@ -535,7 +538,7 @@ impl SequencerCore { let hashable_data = HashableBlockData { block_id: new_block_height, transactions: valid_transactions, - prev_block_hash: latest_block_meta.hash, + prev_block_hash, timestamp: new_block_timestamp, }; @@ -543,8 +546,6 @@ impl SequencerCore { .clone() .into_pending_block(self.store.signing_key()); - self.chain_height = new_block_height; - log::info!( "Created block with {} transactions in {} seconds", hashable_data.transactions.len(), @@ -558,16 +559,30 @@ impl SequencerCore { }) } - pub const fn state(&self) -> &lee::V03State { - &self.state + /// A clone of the current head state. + /// + /// TODO: cloning the whole state per call is wasteful; add targeted + /// account/nonce/proof accessors so RPC reads don't clone the whole state. + #[must_use] + pub fn state(&self) -> lee::V03State { + self.chain + .lock() + .expect("chain state mutex poisoned") + .head_state() + .clone() } pub const fn block_store(&self) -> &SequencerStore { &self.store } - pub const fn chain_height(&self) -> u64 { - self.chain_height + #[must_use] + pub fn chain_height(&self) -> u64 { + self.chain + .lock() + .expect("chain state mutex poisoned") + .head_tip() + .map_or(0, |tip| tip.block_id) } pub const fn sequencer_config(&self) -> &SequencerConfig { @@ -615,12 +630,6 @@ impl SequencerCore { pub fn chain(&self) -> Arc> { Arc::clone(&self.chain) } - - fn next_block_id(&self) -> u64 { - self.chain_height - .checked_add(1) - .unwrap_or_else(|| panic!("Max block height reached: {}", self.chain_height)) - } } struct BlockWithMeta { diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 4b5ebfbc..c1d7027a 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -133,14 +133,14 @@ async fn start_from_config() { let (sequencer, _mempool_handle) = SequencerCoreWithMockClients::start_from_config(config.clone()).await; - assert_eq!(sequencer.chain_height, 1); + assert_eq!(sequencer.chain_height(), 1); assert_eq!(sequencer.sequencer_config.max_num_tx_in_block, 10); let acc1_account_id = initial_public_user_accounts()[0].account_id; let acc2_account_id = initial_public_user_accounts()[1].account_id; - let balance_acc_1 = sequencer.state.get_account_by_id(acc1_account_id).balance; - let balance_acc_2 = sequencer.state.get_account_by_id(acc2_account_id).balance; + let balance_acc_1 = sequencer.state().get_account_by_id(acc1_account_id).balance; + let balance_acc_2 = sequencer.state().get_account_by_id(acc2_account_id).balance; assert_eq!(10000, balance_acc_1); assert_eq!(20000, balance_acc_2); @@ -173,7 +173,7 @@ async fn start_from_config_opens_existing_db_if_it_exists() { let (sequencer, _mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await; - assert_eq!(sequencer.chain_height, 1); + assert_eq!(sequencer.chain_height(), 1); assert!(sequencer.store.latest_block_meta().is_ok()); } @@ -282,7 +282,7 @@ async fn transaction_pre_check_native_transfer_valid() { #[tokio::test] async fn transaction_pre_check_native_transfer_other_signature() { - let (mut sequencer, _mempool_handle) = common_setup().await; + let (sequencer, _mempool_handle) = common_setup().await; let acc1 = initial_public_user_accounts()[0].account_id; let acc2 = initial_public_user_accounts()[1].account_id; @@ -296,7 +296,15 @@ async fn transaction_pre_check_native_transfer_other_signature() { let tx = tx.transaction_stateless_check().unwrap(); // Signature is not from sender. Execution fails - let result = tx.execute_check_on_state(&mut sequencer.state, 0, 0); + let result = tx.execute_check_on_state( + sequencer + .chain() + .lock() + .expect("chain mutex poisoned") + .head_state_mut(), + 0, + 0, + ); assert!(matches!( result, @@ -306,7 +314,7 @@ async fn transaction_pre_check_native_transfer_other_signature() { #[tokio::test] async fn transaction_pre_check_native_transfer_sent_too_much() { - let (mut sequencer, _mempool_handle) = common_setup().await; + let (sequencer, _mempool_handle) = common_setup().await; let acc1 = initial_public_user_accounts()[0].account_id; let acc2 = initial_public_user_accounts()[1].account_id; @@ -322,9 +330,15 @@ async fn transaction_pre_check_native_transfer_sent_too_much() { // Passed pre-check assert!(result.is_ok()); - let result = result - .unwrap() - .execute_check_on_state(&mut sequencer.state, 0, 0); + let result = result.unwrap().execute_check_on_state( + sequencer + .chain() + .lock() + .expect("chain mutex poisoned") + .head_state_mut(), + 0, + 0, + ); let is_failed_at_balance_mismatch = matches!( result.err().unwrap(), lee::error::LeeError::ProgramExecutionFailed(_) @@ -335,7 +349,7 @@ async fn transaction_pre_check_native_transfer_sent_too_much() { #[tokio::test] async fn transaction_execute_native_transfer() { - let (mut sequencer, _mempool_handle) = common_setup().await; + let (sequencer, _mempool_handle) = common_setup().await; let acc1 = initial_public_user_accounts()[0].account_id; let acc2 = initial_public_user_accounts()[1].account_id; @@ -346,11 +360,19 @@ async fn transaction_execute_native_transfer() { acc1, 0, acc2, 100, &sign_key1, ); - tx.execute_check_on_state(&mut sequencer.state, 0, 0) - .unwrap(); + tx.execute_check_on_state( + sequencer + .chain() + .lock() + .expect("chain mutex poisoned") + .head_state_mut(), + 0, + 0, + ) + .unwrap(); - let bal_from = sequencer.state.get_account_by_id(acc1).balance; - let bal_to = sequencer.state.get_account_by_id(acc2).balance; + let bal_from = sequencer.state().get_account_by_id(acc1).balance; + let bal_to = sequencer.state().get_account_by_id(acc2).balance; assert_eq!(bal_from, 9900); assert_eq!(bal_to, 20100); @@ -387,7 +409,7 @@ async fn push_tx_into_mempool_blocks_until_mempool_is_full() { #[tokio::test] async fn build_block_from_mempool() { let (mut sequencer, mempool_handle) = common_setup().await; - let genesis_height = sequencer.chain_height; + let genesis_height = sequencer.chain_height(); let tx = common::test_utils::produce_dummy_empty_transaction(); mempool_handle @@ -397,7 +419,7 @@ async fn build_block_from_mempool() { let result = sequencer.build_block_from_mempool(); assert!(result.is_ok()); - assert_eq!(sequencer.chain_height, genesis_height + 1); + assert_eq!(sequencer.chain_height(), genesis_height + 1); } #[tokio::test] @@ -429,7 +451,7 @@ async fn replay_transactions_are_rejected_in_the_same_block() { sequencer.produce_new_block().await.unwrap(); let block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); @@ -464,7 +486,7 @@ async fn replay_transactions_are_rejected_in_different_blocks() { sequencer.produce_new_block().await.unwrap(); let block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); assert_eq!( @@ -483,7 +505,7 @@ async fn replay_transactions_are_rejected_in_different_blocks() { sequencer.produce_new_block().await.unwrap(); let block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); // The replay is rejected, so only the clock tx is in the block. @@ -525,7 +547,7 @@ async fn restart_from_storage() { sequencer.produce_new_block().await.unwrap(); let block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); assert_eq!( @@ -541,8 +563,8 @@ async fn restart_from_storage() { // with the above transaction and update the state to reflect that. let (sequencer, _mempool_handle) = SequencerCoreWithMockClients::start_from_config(config.clone()).await; - let balance_acc_1 = sequencer.state.get_account_by_id(acc1_account_id).balance; - let balance_acc_2 = sequencer.state.get_account_by_id(acc2_account_id).balance; + let balance_acc_1 = sequencer.state().get_account_by_id(acc1_account_id).balance; + let balance_acc_2 = sequencer.state().get_account_by_id(acc2_account_id).balance; // Balances should be consistent with the stored block assert_eq!( @@ -640,7 +662,7 @@ async fn produce_block_with_correct_prev_meta_after_restart() { // Step 5: Verify the new block has correct previous block metadata let new_block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); @@ -692,7 +714,7 @@ async fn transactions_touching_clock_account_are_dropped_from_block() { let block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); @@ -747,7 +769,7 @@ async fn user_tx_that_chain_calls_clock_is_dropped() { let block = sequencer .store - .get_block_at_id(sequencer.chain_height) + .get_block_at_id(sequencer.chain_height()) .unwrap() .unwrap(); @@ -766,10 +788,13 @@ async fn block_production_aborts_when_clock_account_data_is_corrupted() { // Corrupt the clock 01 account data so the clock program panics on deserialization. let clock_account_id = system_accounts::clock_account_ids()[0]; - let mut corrupted = sequencer.state.get_account_by_id(clock_account_id); + let mut corrupted = sequencer.state().get_account_by_id(clock_account_id); corrupted.data = vec![0xff; 3].try_into().unwrap(); sequencer - .state + .chain() + .lock() + .expect("chain mutex poisoned") + .head_state_mut() .force_insert_account(clock_account_id, corrupted); // Push a dummy transaction so the mempool is non-empty.