apollo_batcher: perform commitment tasks

Author: yoavGrs

crates/apollo_batcher/src/batcher.rs

@@ -1159,6 +1159,7 @@ impl Batcher {
         let commitment_results = self.commitment_manager.get_commitment_results().await;
         for commitment_task_output in commitment_results.into_iter() {
             let height = commitment_task_output.height;
+            info!("Writing commitment results to storage for height {}.", height);
 
             // Decide whether to finalize the block hash based on the config.
             let should_finalize_block_hash =

crates/apollo_batcher/src/commitment_manager/commitment_manager_impl.rs

@@ -263,12 +263,12 @@ impl<S: StateCommitterTrait> CommitmentManager<S> {
                 if height == BlockNumber::ZERO {
                     parent_hash = Some(BlockHash::GENESIS_PARENT_HASH);
                 }
-                let parent_hash = parent_hash.ok_or(CommitmentManagerError::MissingBlockHash(
-                    height.prev().expect(
+                let parent_hash = parent_hash.ok_or_else(|| {
+                    CommitmentManagerError::MissingBlockHash(height.prev().expect(
                         "For the genesis block, the block hash is constant and should not be \
                          fetched from storage.",
-                    ),
-                ))?;
+                    ))
+                })?;
                 let partial_block_hash_components = partial_block_hash_components
                     .ok_or(CommitmentManagerError::MissingPartialBlockHashComponents(height))?;
                 let block_hash =

crates/apollo_batcher/src/commitment_manager/state_committer.rs

@@ -1,10 +1,19 @@
 #![allow(dead_code, unused_variables, unused_mut)]
 
-use apollo_committer_types::communication::SharedCommitterClient;
+use apollo_committer_types::committer_types::{CommitBlockRequest, RevertBlockRequest};
+use apollo_committer_types::communication::{CommitterRequest, SharedCommitterClient};
+use apollo_committer_types::errors::{CommitterClientError, CommitterClientResult};
+use tokio::sync::mpsc::error::TrySendError;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::task::JoinHandle;
+use tracing::{error, info};
 
-use crate::commitment_manager::types::{CommitterTaskInput, CommitterTaskOutput};
+use crate::commitment_manager::types::{
+    CommitmentTaskOutput,
+    CommitterTaskInput,
+    CommitterTaskOutput,
+    RevertTaskOutput,
+};
 
 /// Commits state changes by calling the committer.
 pub(crate) trait StateCommitterTrait {
@@ -29,7 +38,7 @@ impl StateCommitterTrait for StateCommitter {
         committer_client: SharedCommitterClient,
     ) -> Self {
         let handle = tokio::spawn(async move {
-            Self::perform_commitment_tasks(tasks_receiver, results_sender, committer_client).await;
+            Self::perform_tasks(tasks_receiver, results_sender, committer_client).await;
         });
         Self { task_performer_handle: handle }
     }
@@ -39,13 +48,84 @@ impl StateCommitterTrait for StateCommitter {
 }
 
 impl StateCommitter {
-    pub(crate) async fn perform_commitment_tasks(
+    /// Performs the tasks in the channel. Retries at recoverable errors.
+    pub(crate) async fn perform_tasks(
         mut tasks_receiver: Receiver<CommitterTaskInput>,
         mut results_sender: Sender<CommitterTaskOutput>,
         committer_client: SharedCommitterClient,
     ) {
-        // Placeholder: simply drain the receiver and do nothing.
-        // TODO(Amos): Implement the actual commitment tasks logic.
-        while let Some(_task) = tasks_receiver.recv().await {}
+        // TODO(Yoav): Test this function.
+        while let Some(CommitterTaskInput(request)) = tasks_receiver.recv().await {
+            let output = perform_task(request, &committer_client).await;
+            let height = output.height();
+            match results_sender.try_send(output.clone()) {
+                Ok(_) => {
+                    info!(
+                        "Successfully sent the committer result to the results channel: \
+                         {output:?}."
+                    );
+                }
+                Err(TrySendError::Full(_)) => {
+                    panic!("Results channel is full for height {height}.")
+                }
+                Err(err) => panic!("Failed to send results for height {height}. error: {err}"),
+            }
+        }
+    }
+}
+
+/// Performs a commitment task by calling the committer.
+/// Retries at recoverable errors.
+async fn perform_task(
+    request: CommitterRequest,
+    committer_client: &SharedCommitterClient,
+) -> CommitterTaskOutput {
+    loop {
+        let result = match &request {
+            CommitterRequest::CommitBlock(commit_block_request) => {
+                perform_commit_block_task(commit_block_request.clone(), committer_client).await
+            }
+            CommitterRequest::RevertBlock(revert_block_request) => {
+                perform_revert_block_task(revert_block_request.clone(), committer_client).await
+            }
+        };
+        match result {
+            Ok(output) => return output,
+            Err(err) => {
+                log_error_and_maybe_panic(err).await;
+                continue;
+            }
+        }
+    }
+}
+
+async fn perform_commit_block_task(
+    commit_block_request: CommitBlockRequest,
+    committer_client: &SharedCommitterClient,
+) -> CommitterClientResult<CommitterTaskOutput> {
+    let height = commit_block_request.height;
+    let response = committer_client.commit_block(commit_block_request).await?;
+    Ok(CommitterTaskOutput::Commit(CommitmentTaskOutput { response, height }))
+}
+
+async fn perform_revert_block_task(
+    revert_block_request: RevertBlockRequest,
+    committer_client: &SharedCommitterClient,
+) -> CommitterClientResult<CommitterTaskOutput> {
+    let height = revert_block_request.height;
+    let response = committer_client.revert_block(revert_block_request).await?;
+    Ok(CommitterTaskOutput::Revert(RevertTaskOutput { response, height }))
+}
+
+/// Panics on unrecoverable errors.
+async fn log_error_and_maybe_panic(error: CommitterClientError) {
+    // TODO(Amos): Handle errors.
+    match error {
+        CommitterClientError::ClientError(client_error) => {
+            error!("Committer client error: {client_error}. retrying...");
+        }
+        CommitterClientError::CommitterError(committer_error) => {
+            panic!("Committer internal error: {committer_error}");
+        }
     }
 }

crates/apollo_batcher/src/commitment_manager/types.rs

@@ -33,6 +33,13 @@ impl CommitterTaskOutput {
             Self::Revert(_) => panic!("Got revert output: {self:?}"),
         }
     }
+
+    pub(crate) fn height(&self) -> BlockNumber {
+        match self {
+            Self::Commit(CommitmentTaskOutput { height, .. })
+            | Self::Revert(RevertTaskOutput { height, .. }) => *height,
+        }
+    }
 }
 
 pub(crate) struct FinalBlockCommitment {

crates/apollo_integration_tests/src/end_to_end_flow_utils.rs

@@ -82,7 +82,7 @@ pub async fn end_to_end_flow(args: EndToEndFlowArgs) {
         .install_recorder()
         .expect("Should be able to install global prometheus recorder");
 
-    const TEST_SCENARIO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(50);
+    const TEST_SCENARIO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(100);
     // Setup.
     let mock_running_system = FlowTestSetup::new_from_tx_generator(
         &tx_generator,