blockifier,apollo_consensus_orchestrator: avoid state-diff clone in synced-block accessed keys

crates/apollo_consensus_orchestrator/src/cende/cende_test.rs

@@ -6,7 +6,6 @@ use apollo_class_manager_types::MockClassManagerClient;
 use blockifier::blockifier_versioned_constants::VersionedConstants;
 use blockifier::execution::call_info::{CallInfo, StorageAccessTracker};
 use blockifier::execution::entry_point::CallEntryPoint;
-use blockifier::state::cached_state::CommitmentStateDiff;
 use blockifier::state::stateful_compression::ALIAS_COUNTER_STORAGE_KEY;
 use blockifier::transaction::objects::TransactionExecutionInfo;
 use metrics_exporter_prometheus::PrometheusBuilder;
@@ -15,7 +14,7 @@ use rstest::rstest;
 use shared_execution_objects::central_objects::CentralTransactionExecutionInfo;
 use starknet_api::block::{BlockInfo, BlockNumber};
 use starknet_api::core::{ContractAddress, BLOCK_HASH_TABLE_ADDRESS};
-use starknet_api::state::StorageKey;
+use starknet_api::state::{StorageKey, ThinStateDiff};
 use starknet_api::test_utils::read_json_file;
 use starknet_api::transaction::fields::{snos_block_number_from_proof_facts, ProofFacts};
 use starknet_api::versioned_constants_logic::VersionedConstantsTrait;
@@ -365,8 +364,8 @@ fn compute_accessed_keys() {
     // A state diff writing to a storage key of another contract.
     let written_address = ContractAddress::from(0x500_u16);
     let written_key = StorageKey::from(0x600_u16);
-    let mut state_diff = CommitmentStateDiff::default();
-    state_diff.storage_updates.entry(written_address).or_default().insert(written_key, Felt::ONE);
+    let mut state_diff = ThinStateDiff::default();
+    state_diff.storage_diffs.entry(written_address).or_default().insert(written_key, Felt::ONE);
 
     let block_accessed_keys_data = BlockAccessedKeysData {
         transactions: vec![transaction],

crates/apollo_consensus_orchestrator/src/cende/mod.rs

@@ -155,8 +155,10 @@ impl BlockAccessedKeysData {
     /// Computes the block's `AccessedKeys` from the recorder-supplied transactions and execution
     /// infos, the synced block's `state_diff`, and the latest versioned constants. Mirrors
     /// `AccessedKeys::new`, but sources the call infos from the central execution infos and the
-    /// proof-facts block numbers from the transactions.
-    pub fn compute_accessed_keys(&self, state_diff: &CommitmentStateDiff) -> AccessedKeys {
+    /// proof-facts block numbers from the transactions. Takes the synced block's `state_diff` by
+    /// reference (a `ThinStateDiff`, not a `CommitmentStateDiff`) so the caller isn't forced to
+    /// clone it just to compute the accessed keys.
+    pub fn compute_accessed_keys(&self, state_diff: &ThinStateDiff) -> AccessedKeys {
         let proof_facts_block_numbers: Vec<BlockNumber> = self
             .transactions
             .iter()

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context.rs

@@ -1133,6 +1133,33 @@ impl ConsensusContext for SequencerConsensusContext {
             );
             return false;
         }
+        // Fetch the block's transactions and execution infos from the centralized recorder and
+        // compute the accessed keys locally, so the batcher can build the block's state commitment
+        // infos. Done before updating any consensus state so that a recorder failure leaves the
+        // context unchanged for the retried sync attempt.
+        let accessed_keys = if self.config.static_config.fetch_accessed_keys_from_centralized {
+            match self.deps.cende_ambassador.get_accessed_keys_input(block_number).await {
+                Ok(Some(block_data)) => {
+                    info!("Fetched accessed-keys data for synced block {block_number}.");
+                    Some(block_data.compute_accessed_keys(&sync_block.state_diff))
+                }
+                Ok(None) => {
+                    panic!(
+                        "The accessed-keys data for synced block {block_number} is expected to be \
+                         ready."
+                    );
+                }
+                Err(e) => {
+                    error!(
+                        "Failed to fetch accessed-keys data for synced block {block_number}: {e:?}"
+                    );
+                    return false;
+                }
+            }
+        } else {
+            None
+        };
+
         self.record_fee_proposal(height, sync_block.block_header_without_hash.fee_proposal_fri);
         self.previous_proposal_init =
             Some(previous_proposal_init_from_block_header(&sync_block.block_header_without_hash));
@@ -1142,7 +1169,7 @@ impl ConsensusContext for SequencerConsensusContext {
             "Adding sync block to Batcher for height {}",
             sync_block.block_header_without_hash.block_number,
         );
-        if let Err(e) = self.deps.batcher.add_sync_block(sync_block, None).await {
+        if let Err(e) = self.deps.batcher.add_sync_block(sync_block, accessed_keys).await {
             error!("Failed to add sync block to Batcher: {e:?}");
             return false;
         }

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context_test.rs

@@ -65,6 +65,7 @@ use starknet_committer::patricia_merkle_tree::types::CompressedStateCommitmentIn
 use tracing_test::traced_test;
 
 use crate::cende::{
+    BlockAccessedKeysData,
     CendeAmbassadorError,
     MockCendeContext,
     StateCommitmentInfosAndNumber,
@@ -2065,3 +2066,105 @@ async fn test_compute_proposer_fee_proposal_converges_to_oracle_target() {
         );
     }
 }
+
+// When `fetch_accessed_keys_from_centralized` is enabled, `try_sync` fetches the block's accessed
+// keys from the recorder and forwards them to the batcher's `add_sync_block`.
+#[tokio::test]
+async fn try_sync_forwards_accessed_keys_from_centralized() {
+    const SYNC_HEIGHT: BlockNumber = BlockNumber(7);
+
+    let (mut deps, _network) = create_test_and_network_deps();
+    // Specific get_block expectation must be registered before setup_default_expectations, which
+    // installs a catch-all handler.
+    deps.state_sync_client.expect_get_block().times(1).return_once(|height| {
+        let mut sync_block = SyncBlock::default();
+        sync_block.block_header_without_hash.block_number = height;
+        Ok(sync_block)
+    });
+    deps.setup_default_expectations();
+
+    // The recorder returns the block's transactions and execution infos; `try_sync` computes the
+    // accessed keys from them (plus the synced block's state diff and the latest versioned
+    // constants) and forwards them to `add_sync_block`. Empty data exercises the
+    // fetch -> compute -> forward path.
+    let block_data = BlockAccessedKeysData { transactions: vec![], execution_infos: vec![] };
+    deps.cende_ambassador
+        .expect_get_accessed_keys_input()
+        .times(1)
+        .withf(|block_number| *block_number == SYNC_HEIGHT)
+        .return_once(move |_| Ok(Some(block_data)));
+    deps.batcher
+        .expect_add_sync_block()
+        .times(1)
+        .withf(|_sync_block, accessed_keys| accessed_keys.is_some())
+        .return_once(|_, _| Ok(()));
+
+    let mut context = deps.build_context_with_config(ContextConfig {
+        static_config: ContextStaticConfig {
+            chain_id: CHAIN_ID,
+            fetch_accessed_keys_from_centralized: true,
+            ..Default::default()
+        },
+        ..Default::default()
+    });
+
+    assert!(context.try_sync(SYNC_HEIGHT).await);
+}
+
+// When `fetch_accessed_keys_from_centralized` is enabled, the block's accessed keys are required;
+// the data is expected to be ready at sync time, so a missing entry in the recorder is a bug.
+#[tokio::test]
+#[should_panic(expected = "The accessed-keys data for synced block 7 is expected to be ready.")]
+async fn try_sync_panics_when_recorder_has_no_accessed_keys() {
+    const SYNC_HEIGHT: BlockNumber = BlockNumber(7);
+
+    let (mut deps, _network) = create_test_and_network_deps();
+    // Specific get_block expectation must be registered before setup_default_expectations, which
+    // installs a catch-all handler.
+    deps.state_sync_client.expect_get_block().times(1).return_once(|height| {
+        let mut sync_block = SyncBlock::default();
+        sync_block.block_header_without_hash.block_number = height;
+        Ok(sync_block)
+    });
+    deps.setup_default_expectations();
+
+    deps.cende_ambassador.expect_get_accessed_keys_input().times(1).return_once(|_| Ok(None));
+    deps.batcher.expect_add_sync_block().times(0);
+
+    let mut context = deps.build_context_with_config(ContextConfig {
+        static_config: ContextStaticConfig {
+            chain_id: CHAIN_ID,
+            fetch_accessed_keys_from_centralized: true,
+            ..Default::default()
+        },
+        ..Default::default()
+    });
+
+    context.try_sync(SYNC_HEIGHT).await;
+}
+
+// With the opt-in disabled (the default), `try_sync` must not query the recorder and passes `None`
+// accessed keys to the batcher.
+#[tokio::test]
+async fn try_sync_skips_accessed_keys_when_disabled() {
+    const SYNC_HEIGHT: BlockNumber = BlockNumber(7);
+
+    let (mut deps, _network) = create_test_and_network_deps();
+    deps.state_sync_client.expect_get_block().times(1).return_once(|height| {
+        let mut sync_block = SyncBlock::default();
+        sync_block.block_header_without_hash.block_number = height;
+        Ok(sync_block)
+    });
+    deps.setup_default_expectations();
+
+    deps.cende_ambassador.expect_get_accessed_keys_input().times(0);
+    deps.batcher
+        .expect_add_sync_block()
+        .times(1)
+        .withf(|_sync_block, accessed_keys| accessed_keys.is_none())
+        .return_once(|_, _| Ok(()));
+
+    let mut context = deps.build_context();
+
+    assert!(context.try_sync(SYNC_HEIGHT).await);
+}

crates/blockifier/src/state/accessed_keys.rs

@@ -7,7 +7,7 @@ use starknet_api::block::BlockNumber;
 use starknet_api::core::{ClassHash, ContractAddress, BLOCK_HASH_TABLE_ADDRESS};
 use starknet_api::state::StorageKey;
 
-use super::cached_state::{CommitmentStateDiff, StateChangesKeys, StorageEntry};
+use super::cached_state::{StateChangesKeys, StateDiffView, StorageEntry};
 use super::stateful_compression::predicted_alias_storage_entries;
 use crate::blockifier_versioned_constants::VersionedConstants;
 use crate::execution::call_info::CallInfo;
@@ -73,7 +73,7 @@ impl AccessedKeys {
     pub fn new<'a>(
         execution_infos: impl IntoIterator<Item = &'a TransactionExecutionInfo>,
         proof_facts_block_numbers: impl IntoIterator<Item = &'a BlockNumber>,
-        state_diff: &CommitmentStateDiff,
+        state_diff: &impl StateDiffView,
         versioned_constants: &VersionedConstants,
     ) -> Self {
         Self::from_call_infos(
@@ -94,7 +94,7 @@ impl AccessedKeys {
     pub fn from_call_infos<'a>(
         call_infos: impl IntoIterator<Item = &'a CallInfo>,
         proof_facts_block_numbers: impl IntoIterator<Item = &'a BlockNumber>,
-        state_diff: &CommitmentStateDiff,
+        state_diff: &impl StateDiffView,
         versioned_constants: &VersionedConstants,
     ) -> Self {
         let mut storage_keys: BTreeSet<StorageEntry> = BTreeSet::new();
@@ -116,7 +116,7 @@ impl AccessedKeys {
         }
 
         // Storage entries written in the state diff.
-        for (address, inner) in &state_diff.storage_updates {
+        for (address, inner) in state_diff.storage_diffs() {
             storage_keys.extend(inner.keys().map(|key| (*address, *key)));
         }
         // Add the block hash table entries for the proof facts.
@@ -133,10 +133,11 @@ impl AccessedKeys {
         }
 
         accessed_contracts.extend(storage_keys.iter().map(|(address, _)| *address));
-        accessed_contracts.extend(state_diff.get_contract_addresses().into_iter().copied());
+        accessed_contracts.extend(state_diff.contract_addresses().copied());
 
-        accessed_class_hashes.extend(state_diff.address_to_class_hash.values().copied());
-        accessed_class_hashes.extend(state_diff.class_hash_to_compiled_class_hash.keys().copied());
+        accessed_class_hashes.extend(state_diff.deployed_contracts().values().copied());
+        accessed_class_hashes
+            .extend(state_diff.class_hash_to_compiled_class_hash().keys().copied());
 
         Self { storage_keys, accessed_contracts, accessed_class_hashes }
     }

crates/blockifier/src/state/cached_state.rs

@@ -724,14 +724,63 @@ pub struct CommitmentStateDiff {
     pub class_hash_to_compiled_class_hash: IndexMap<ClassHash, CompiledClassHash>,
 }
 
-impl CommitmentStateDiff {
-    /// Union of addresses with storage updates, nonce changes, or class hash changes.
-    pub fn get_contract_addresses(&self) -> HashSet<&ContractAddress> {
-        self.storage_updates
+/// A read-only view of the state-diff fields `AccessedKeys` and the alias-compression prediction
+/// need, implemented for both `CommitmentStateDiff` (built while executing a block) and
+/// `ThinStateDiff` (as received from state sync), so both can be computed from either state diff
+/// without an intermediate clone.
+pub trait StateDiffView {
+    fn storage_diffs(&self) -> &IndexMap<ContractAddress, IndexMap<StorageKey, Felt>>;
+    fn nonces(&self) -> &IndexMap<ContractAddress, Nonce>;
+    fn deployed_contracts(&self) -> &IndexMap<ContractAddress, ClassHash>;
+    fn class_hash_to_compiled_class_hash(&self) -> &IndexMap<ClassHash, CompiledClassHash>;
+
+    /// Union of addresses with storage updates, nonce changes, or class hash changes. May yield
+    /// the same address more than once; callers that need a deduplicated set should collect into
+    /// one.
+    fn contract_addresses(&self) -> impl Iterator<Item = &ContractAddress> {
+        self.storage_diffs()
             .keys()
-            .chain(self.address_to_nonce.keys())
-            .chain(self.address_to_class_hash.keys())
-            .collect()
+            .chain(self.nonces().keys())
+            .chain(self.deployed_contracts().keys())
+    }
+}
+
+impl StateDiffView for CommitmentStateDiff {
+    fn storage_diffs(&self) -> &IndexMap<ContractAddress, IndexMap<StorageKey, Felt>> {
+        &self.storage_updates
+    }
+
+    fn nonces(&self) -> &IndexMap<ContractAddress, Nonce> {
+        &self.address_to_nonce
+    }
+
+    fn deployed_contracts(&self) -> &IndexMap<ContractAddress, ClassHash> {
+        &self.address_to_class_hash
+    }
+
+    fn class_hash_to_compiled_class_hash(&self) -> &IndexMap<ClassHash, CompiledClassHash> {
+        &self.class_hash_to_compiled_class_hash
+    }
+}
+
+/// Drops `deprecated_declared_classes: Vec<ClassHash>`, which has no counterpart here: Cairo-0
+/// classes have no contract-class-trie leaf (mirrors `From<ThinStateDiff> for
+/// CommitmentStateDiff`).
+impl StateDiffView for ThinStateDiff {
+    fn storage_diffs(&self) -> &IndexMap<ContractAddress, IndexMap<StorageKey, Felt>> {
+        &self.storage_diffs
+    }
+
+    fn nonces(&self) -> &IndexMap<ContractAddress, Nonce> {
+        &self.nonces
+    }
+
+    fn deployed_contracts(&self) -> &IndexMap<ContractAddress, ClassHash> {
+        &self.deployed_contracts
+    }
+
+    fn class_hash_to_compiled_class_hash(&self) -> &IndexMap<ClassHash, CompiledClassHash> {
+        &self.class_hash_to_compiled_class_hash
     }
 }
 

crates/blockifier/src/state/stateful_compression.rs

@@ -6,7 +6,7 @@ use starknet_api::StarknetApiError;
 use starknet_types_core::felt::Felt;
 use thiserror::Error;
 
-use super::cached_state::{CachedState, CommitmentStateDiff, StateMaps, StorageEntry};
+use super::cached_state::{CachedState, StateDiffView, StateMaps, StorageEntry};
 use super::errors::StateError;
 use super::state_api::{State, StateReader, StateResult};
 
@@ -84,17 +84,17 @@ pub fn allocate_aliases_in_storage<S: StateReader>(
 /// access pattern of `AliasUpdater::new` (counter read) plus each `AliasUpdater::insert_alias`
 /// call (one read per qualifying key).
 pub fn predicted_alias_storage_entries(
-    state_diff: &CommitmentStateDiff,
+    state_diff: &impl StateDiffView,
     alias_contract_address: ContractAddress,
 ) -> HashSet<StorageEntry> {
     let mut entries = HashSet::new();
     entries.insert((alias_contract_address, ALIAS_COUNTER_STORAGE_KEY));
-    for address in state_diff.get_contract_addresses() {
+    for address in state_diff.contract_addresses() {
         if should_compress_address(address) {
             entries.insert((alias_contract_address, StorageKey(address.0)));
         }
     }
-    for (address, inner) in &state_diff.storage_updates {
+    for (address, inner) in state_diff.storage_diffs() {
         for storage_key in inner.keys() {
             if should_compress_storage_key(storage_key, address) {
                 entries.insert((alias_contract_address, *storage_key));