apollo_consensus_orchestrator: verify retrospective state commitment infos before proposing

crates/apollo_consensus_orchestrator/src/build_proposal.rs

@@ -56,6 +56,11 @@ use crate::utils::{
     RetrospectiveBlockHashError,
     StreamSender,
 };
+#[cfg(feature = "os_input")]
+use crate::utils::{
+    verify_retrospective_state_commitment_infos,
+    RetrospectiveStateCommitmentInfosError,
+};
 
 // Minimal wait time that avoids an immediate timeout.
 const MIN_WAIT_DURATION: Duration = Duration::from_millis(1);
@@ -101,6 +106,9 @@ pub(crate) enum BuildProposalError {
     Batcher(String, BatcherClientError),
     #[error(transparent)]
     RetrospectiveBlockHashError(#[from] RetrospectiveBlockHashError),
+    #[cfg(feature = "os_input")]
+    #[error(transparent)]
+    RetrospectiveStateCommitmentInfosError(#[from] RetrospectiveStateCommitmentInfosError),
     #[error("Failed to send proposal part: {0}")]
     SendError(String),
     #[error("ExchangeRateOracle error: {0}")]
@@ -225,6 +233,16 @@ async fn initiate_build(args: &mut ProposalBuildArguments) -> BuildProposalResul
     )
     .await?;
 
+    // Make sure the blob of this height will carry the next height's retrospective commitment
+    // infos.
+    #[cfg(feature = "os_input")]
+    verify_retrospective_state_commitment_infos(
+        args.deps.batcher.as_ref(),
+        args.deps.cende_ambassador.as_ref(),
+        init.height,
+    )
+    .await?;
+
     let build_proposal_input = ProposeBlockInput {
         proposal_id: args.proposal_id,
         deadline: args.batcher_deadline,

crates/apollo_consensus_orchestrator/src/utils.rs

@@ -35,6 +35,8 @@ use starknet_api::hash::StarkHash;
 use starknet_api::StarknetApiError;
 use tracing::{info, warn};
 
+#[cfg(feature = "os_input")]
+use crate::cende::{CendeAmbassadorError, CendeContext};
 use crate::metrics::{
     CONSENSUS_L1_GAS_PRICE_PROVIDER_ERROR,
     CONSENSUS_RETROSPECTIVE_BLOCK_HASH_MISMATCH,
@@ -69,6 +71,28 @@ pub(crate) enum RetrospectiveBlockHashError {
 
 pub(crate) type RetrospectiveBlockHashResult<T> = Result<T, RetrospectiveBlockHashError>;
 
+#[cfg(feature = "os_input")]
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum RetrospectiveStateCommitmentInfosError {
+    #[error(transparent)]
+    BatcherError(#[from] BatcherClientError),
+    #[error(transparent)]
+    CendeError(#[from] CendeAmbassadorError),
+    #[error(
+        "State commitment infos of the next height's retrospective block \
+         {retrospective_block_number} are not stored yet: the batcher doesn't have them, and the \
+         cende recorder commitment infos height offset is {cende_recorder_height_offset:?}"
+    )]
+    NotStored {
+        retrospective_block_number: BlockNumber,
+        cende_recorder_height_offset: Option<BlockNumber>,
+    },
+}
+
+#[cfg(feature = "os_input")]
+pub(crate) type RetrospectiveStateCommitmentInfosResult<T> =
+    Result<T, RetrospectiveStateCommitmentInfosError>;
+
 #[derive(Debug)]
 pub(crate) struct GasPriceParams {
     pub min_l1_gas_price_wei: GasPrice,
@@ -463,6 +487,51 @@ pub(crate) async fn wait_for_retrospective_block_hash(
     result
 }
 
+/// Verifies that the batcher or the cende recorder has stored the state commitment infos of the
+/// next height's retrospective block. Skipped when the batcher doesn't have them and the recorder
+/// has stored none at all, for pre-feature activation.
+#[cfg(feature = "os_input")]
+pub(crate) async fn verify_retrospective_state_commitment_infos(
+    batcher_client: &dyn BatcherClient,
+    cende_ambassador: &dyn CendeContext,
+    height: BlockNumber,
+) -> RetrospectiveStateCommitmentInfosResult<()> {
+    let Some(retrospective_block_number) =
+        height.unchecked_next().0.checked_sub(STORED_BLOCK_HASH_BUFFER)
+    else {
+        info!(
+            "The next height is less than {STORED_BLOCK_HASH_BUFFER}, no retrospective state \
+             commitment infos are required."
+        );
+        return Ok(());
+    };
+    let retrospective_block_number = BlockNumber(retrospective_block_number);
+
+    // The batcher is queried at the retrospective height itself since apollo storage doesn't
+    // guarantee the commitment infos of every height are stored (e.g. blocks added via sync);
+    // the cende recorder's storage is contiguous, so its height offset is enough.
+    if batcher_client.get_state_commitment_infos(retrospective_block_number).await?.is_some() {
+        return Ok(());
+    }
+
+    let cende_recorder_height_offset = cende_ambassador.commitment_infos_height_offset().await?;
+    if cende_recorder_height_offset.is_some_and(|offset| offset > retrospective_block_number) {
+        return Ok(());
+    }
+    if cende_recorder_height_offset.is_none() {
+        warn!(
+            "The batcher doesn't have the retrospective block's state commitment infos and the \
+             cende recorder hasn't stored any; skipping the retrospective state commitment infos \
+             validation."
+        );
+        return Ok(());
+    }
+    Err(RetrospectiveStateCommitmentInfosError::NotStored {
+        retrospective_block_number,
+        cende_recorder_height_offset,
+    })
+}
+
 pub(crate) fn truncate_to_executed_txs(
     content: &mut Vec<Vec<InternalConsensusTransaction>>,
     final_n_executed_txs: usize,

crates/apollo_consensus_orchestrator/src/utils_test.rs

@@ -1,24 +1,47 @@
 use apollo_batcher_types::communication::BatcherClientError;
+#[cfg(feature = "os_input")]
+use apollo_batcher_types::communication::MockBatcherClient;
 use apollo_batcher_types::errors::BatcherError;
 use apollo_protobuf::consensus::ProposalInit;
 use apollo_state_sync_types::communication::StateSyncClientError;
 use apollo_state_sync_types::errors::StateSyncError;
 use assert_matches::assert_matches;
 use blockifier::abi::constants::STORED_BLOCK_HASH_BUFFER;
+#[cfg(feature = "os_input")]
+use rstest::rstest;
 use starknet_api::block::{BlockHash, BlockHashAndNumber, BlockNumber};
+#[cfg(feature = "os_input")]
+use starknet_committer::patricia_merkle_tree::types::CompressedStateCommitmentInfos;
 use starknet_types_core::felt::Felt;
 
 use crate::build_proposal::ProposalBuildArguments;
+#[cfg(feature = "os_input")]
+use crate::cende::MockCendeContext;
 use crate::test_utils::create_proposal_build_arguments;
 use crate::utils::{
     get_l1_prices_in_fri_and_wei,
     retrospective_block_hash,
     wait_for_retrospective_block_hash,
     RetrospectiveBlockHashError,
 };
+#[cfg(feature = "os_input")]
+use crate::utils::{
+    verify_retrospective_state_commitment_infos,
+    RetrospectiveStateCommitmentInfosError,
+};
 
 const CURRENT_BLOCK_NUMBER: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER);
 const RETRO_BLOCK_NUMBER: BlockNumber = BlockNumber(0);
+#[cfg(feature = "os_input")]
+const NEXT_HEIGHT_RETRO_BLOCK_NUMBER: BlockNumber =
+    BlockNumber(CURRENT_BLOCK_NUMBER.0 + 1 - STORED_BLOCK_HASH_BUFFER);
+// A recorder offset above the next height's retrospective block number means its commitment infos
+// are stored; an offset equal to it means they are missing.
+#[cfg(feature = "os_input")]
+const STORED_HEIGHT_OFFSET: Option<BlockNumber> =
+    Some(BlockNumber(NEXT_HEIGHT_RETRO_BLOCK_NUMBER.0 + 1));
+#[cfg(feature = "os_input")]
+const BEHIND_HEIGHT_OFFSET: Option<BlockNumber> = Some(NEXT_HEIGHT_RETRO_BLOCK_NUMBER);
 const MUST_HAVE_BLOCK_HASH_FOR: BlockNumber = BlockNumber(1);
 const RETRO_BLOCK_HASH: BlockHash = BlockHash(Felt::from_hex_unchecked("0x1234567890abcdef"));
 
@@ -331,3 +354,70 @@ async fn wait_for_retrospective_block_hash_batcher_ready_after_a_while() {
         Some(BlockHashAndNumber { number: RETRO_BLOCK_NUMBER, hash: RETRO_BLOCK_HASH })
     );
 }
+
+#[cfg(feature = "os_input")]
+fn mock_batcher_commitment_infos(batcher_has_infos: bool) -> MockBatcherClient {
+    let mut batcher = MockBatcherClient::new();
+    batcher.expect_get_state_commitment_infos().times(1).returning(move |block_number| {
+        assert_eq!(block_number, NEXT_HEIGHT_RETRO_BLOCK_NUMBER);
+        Ok(batcher_has_infos
+            .then(|| CompressedStateCommitmentInfos(b"compressed-state-commitment-infos".to_vec())))
+    });
+    batcher
+}
+
+#[cfg(feature = "os_input")]
+fn mock_cende_recorder_height_offset(height_offset: Option<BlockNumber>) -> MockCendeContext {
+    let mut cende_ambassador = MockCendeContext::new();
+    cende_ambassador
+        .expect_commitment_infos_height_offset()
+        .times(1)
+        .returning(move || Ok(height_offset));
+    cende_ambassador
+}
+
+#[cfg(feature = "os_input")]
+#[rstest]
+#[case::stored_on_batcher(true, None, true)]
+#[case::stored_only_on_cende(false, STORED_HEIGHT_OFFSET, true)]
+#[case::both_sides_empty_skip_validation(false, None, true)]
+#[case::missing_on_batcher_and_recorder_behind(false, BEHIND_HEIGHT_OFFSET, false)]
+#[tokio::test]
+async fn retrospective_state_commitment_infos(
+    #[case] batcher_has_infos: bool,
+    #[case] cende_recorder_height_offset: Option<BlockNumber>,
+    #[case] validation_passes: bool,
+) {
+    // When the batcher has the commitment infos, the cende recorder must not be queried.
+    let cende_ambassador = if batcher_has_infos {
+        MockCendeContext::new()
+    } else {
+        mock_cende_recorder_height_offset(cende_recorder_height_offset)
+    };
+    let res = verify_retrospective_state_commitment_infos(
+        &mock_batcher_commitment_infos(batcher_has_infos),
+        &cende_ambassador,
+        CURRENT_BLOCK_NUMBER,
+    )
+    .await;
+
+    if validation_passes {
+        res.unwrap();
+    } else {
+        assert_matches!(res.unwrap_err(), RetrospectiveStateCommitmentInfosError::NotStored { .. });
+    }
+}
+
+#[cfg(feature = "os_input")]
+#[tokio::test]
+async fn retrospective_state_commitment_infos_next_height_below_buffer() {
+    // No queries are expected: heights whose next height is below the buffer have no
+    // retrospective block.
+    verify_retrospective_state_commitment_infos(
+        &MockBatcherClient::new(),
+        &MockCendeContext::new(),
+        BlockNumber(STORED_BLOCK_HASH_BUFFER - 2),
+    )
+    .await
+    .unwrap();
+}