apollo_batcher: add ReadPathsAndCommitBlock committer task type

crates/apollo_batcher/src/commitment_manager/commitment_manager_impl.rs

@@ -220,6 +220,10 @@ impl<S: StateCommitterTrait> CommitmentManager<S> {
                 CommitterTaskOutput::Commit(commitment_task_result) => {
                     commitment_results.push(commitment_task_result)
                 }
+                #[cfg(feature = "os_input")]
+                CommitterTaskOutput::ReadPathsAndCommitBlock(read_path_and_commit_task_result) => {
+                    commitment_results.push(read_path_and_commit_task_result)
+                }
                 CommitterTaskOutput::Revert(revert_task_result) => {
                     return (commitment_results, revert_task_result);
                 }
@@ -548,7 +552,6 @@ impl<S: StateCommitterTrait> CommitmentManager<S> {
                     // TODO(Ariel): Add dedicated metrics once we use os_input in prod.
                     record_commit_block_metric(task_duration, height, task_type)
                 }
-
             }
         }
     }

crates/apollo_batcher/src/commitment_manager/state_committer.rs

@@ -1,6 +1,12 @@
 use std::time::Duration;
 
 use apollo_committer_types::committer_types::{CommitBlockRequest, RevertBlockRequest};
+#[cfg(feature = "os_input")]
+use apollo_committer_types::committer_types::{
+    CommitBlockResponse,
+    ReadPathsAndCommitBlockRequest,
+    ReadPathsAndCommitBlockResponse,
+};
 use apollo_committer_types::communication::SharedCommitterClient;
 use apollo_committer_types::errors::CommitterClientResult;
 use tokio::sync::mpsc::{Receiver, Sender};
@@ -90,6 +96,14 @@ async fn perform_task(
             CommitterTaskInput::Commit(commit_block_request) => {
                 perform_commit_block_task(commit_block_request.clone(), committer_client).await
             }
+            #[cfg(feature = "os_input")]
+            CommitterTaskInput::ReadPathsAndCommitBlock(read_paths_and_commit_block_request) => {
+                perform_read_paths_and_commit_block_task(
+                    read_paths_and_commit_block_request.clone(),
+                    committer_client,
+                )
+                .await
+            }
             CommitterTaskInput::Revert(revert_block_request) => {
                 perform_revert_block_task(revert_block_request.clone(), committer_client).await
             }
@@ -113,6 +127,22 @@ async fn perform_commit_block_task(
     Ok(CommitterTaskOutput::Commit(CommitmentTaskOutput { response, height }))
 }
 
+/// Commits the block and fetches the Patricia witnesses for the accessed keys via
+/// `ReadPathsAndCommitBlock`.
+#[cfg(feature = "os_input")]
+async fn perform_read_paths_and_commit_block_task(
+    request: ReadPathsAndCommitBlockRequest,
+    committer_client: &SharedCommitterClient,
+) -> CommitterClientResult<CommitterTaskOutput> {
+    let height = request.commit.height;
+    let ReadPathsAndCommitBlockResponse { global_root, patricia_proofs: _ } =
+        committer_client.read_paths_and_commit_block(request).await?;
+    Ok(CommitterTaskOutput::ReadPathsAndCommitBlock(CommitmentTaskOutput {
+        response: CommitBlockResponse { global_root },
+        height,
+    }))
+}
+
 async fn perform_revert_block_task(
     revert_block_request: RevertBlockRequest,
     committer_client: &SharedCommitterClient,

crates/apollo_batcher/src/commitment_manager/types.rs

@@ -4,6 +4,8 @@ use std::collections::HashMap;
 use std::fmt::Display;
 use std::time::Instant;
 
+#[cfg(feature = "os_input")]
+use apollo_committer_types::committer_types::ReadPathsAndCommitBlockRequest;
 use apollo_committer_types::committer_types::{
     CommitBlockRequest,
     CommitBlockResponse,
@@ -20,20 +22,27 @@ use tracing::warn;
 #[cfg_attr(test, derive(Clone))]
 pub(crate) enum CommitterTaskInput {
     Commit(CommitBlockRequest),
+    #[cfg(feature = "os_input")]
+    ReadPathsAndCommitBlock(ReadPathsAndCommitBlockRequest),
     Revert(RevertBlockRequest),
 }
 
 impl CommitterTaskInput {
     pub(crate) fn height(&self) -> BlockNumber {
         match self {
             Self::Commit(request) => request.height,
+            #[cfg(feature = "os_input")]
+            Self::ReadPathsAndCommitBlock(request) => request.commit.height,
             Self::Revert(request) => request.height,
         }
     }
 
+    /// The committer endpoint this task will use.
     pub(crate) fn task_type(&self) -> CommitterRequestLabelValue {
         match self {
             Self::Commit(_) => CommitterRequestLabelValue::CommitBlock,
+            #[cfg(feature = "os_input")]
+            Self::ReadPathsAndCommitBlock(_) => CommitterRequestLabelValue::ReadPathsAndCommitBlock,
             Self::Revert(_) => CommitterRequestLabelValue::RevertBlock,
         }
     }
@@ -47,6 +56,15 @@ impl Display for CommitterTaskInput {
                 "Commit(height={}, state_diff_commitment={:?})",
                 request.height, request.state_diff_commitment
             ),
+            #[cfg(feature = "os_input")]
+            Self::ReadPathsAndCommitBlock(request) => write!(
+                f,
+                "ReadPathsAndCommitBlock(height={}, state_diff_commitment={:?}, \
+                 num_accessed_keys={})",
+                request.commit.height,
+                request.commit.state_diff_commitment,
+                request.accessed_keys.len()
+            ),
             Self::Revert(request) => write!(f, "Revert(height={})", request.height),
         }
     }
@@ -67,28 +85,35 @@ pub(crate) struct RevertTaskOutput {
 #[derive(Clone, Debug)]
 pub(crate) enum CommitterTaskOutput {
     Commit(CommitmentTaskOutput),
+    #[cfg(feature = "os_input")]
+    ReadPathsAndCommitBlock(CommitmentTaskOutput),
     Revert(RevertTaskOutput),
 }
 
 impl CommitterTaskOutput {
     pub(crate) fn expect_commitment(self) -> CommitmentTaskOutput {
         match self {
             Self::Commit(commitment_task_output) => commitment_task_output,
+            #[cfg(feature = "os_input")]
+            Self::ReadPathsAndCommitBlock(commitment_task_output) => commitment_task_output,
             Self::Revert(_) => panic!("Got revert output: {self:?}"),
         }
     }
 
     pub(crate) fn height(&self) -> BlockNumber {
         match self {
             Self::Commit(output) => output.height,
+            #[cfg(feature = "os_input")]
+            Self::ReadPathsAndCommitBlock(output) => output.height,
             Self::Revert(output) => output.height,
         }
     }
 
-    /// The committer endpoint that produced this output.
     pub(crate) fn task_label(&self) -> CommitterRequestLabelValue {
         match self {
             Self::Commit(_) => CommitterRequestLabelValue::CommitBlock,
+            #[cfg(feature = "os_input")]
+            Self::ReadPathsAndCommitBlock(_) => CommitterRequestLabelValue::ReadPathsAndCommitBlock,
             Self::Revert(_) => CommitterRequestLabelValue::RevertBlock,
         }
     }

crates/blockifier/src/state/accessed_keys.rs

@@ -59,6 +59,15 @@ impl From<StateChangesKeys> for AccessedKeys {
 }
 
 impl AccessedKeys {
+    /// The total number of accessed trie leaves across all three tries.
+    pub fn len(&self) -> usize {
+        self.storage_keys.len() + self.accessed_contracts.len() + self.accessed_class_hashes.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
     /// Builds the [`AccessedKeys`] the OS needs to read at the execution of a block.
     pub fn new<'a>(
         execution_infos: impl IntoIterator<Item = &'a TransactionExecutionInfo>,