[refactor] use block_waiter instead of batch_loader (MystenLabs#738)

This commit is swapping the batch_loader with the block_waiter in the executor. This allow us to use a common solution for fetching batches and take advantage of the existing payload sync capabilities. Also earlier fix code has been removed for swapping external worker addresses with internal ones.

executor/Cargo.toml

@@ -23,6 +23,7 @@ tokio-util = { version = "0.7.3", features = ["codec"] }
 tonic = "0.7.2"
 tracing = "0.1.36"
 prometheus = "0.13.1"
+backoff = { version = "0.4.0", features = ["tokio"] }
 
 types = { path = "../types" }
 worker = { path = "../worker" }
@@ -36,5 +37,7 @@ match_opt = "0.1.2"
 indexmap = { version = "1.9.1", features = ["serde"] }
 rand = "0.8.5"
 tempfile = "3.3.0"
+primary = { path = "../primary" }
 test_utils = { path = "../test_utils" }
 types = { path = "../types" }
+telemetry-subscribers = { git = "https://github.com/mystenlabs/mysten-infra.git", rev = "d965a5a795dcdb4d1c7964acf556bc249fdc58aa" }

executor/src/core.rs

@@ -14,11 +14,7 @@ use tokio::{
     task::JoinHandle,
 };
 use tracing::debug;
-use types::{
-    metered_channel, Batch, BatchDigest, ReconfigureNotification, SequenceNumber,
-    SerializedBatchMessage,
-};
-use worker::WorkerMessage;
+use types::{metered_channel, Batch, BatchDigest, ReconfigureNotification, SequenceNumber};
 
 #[cfg(test)]
 #[path = "tests/executor_tests.rs"]
@@ -30,7 +26,7 @@ pub mod executor_tests;
 /// not processes twice the same transaction (despite crash-recovery).
 pub struct Core<State: ExecutionState> {
     /// The temporary storage holding all transactions' data (that may be too big to hold in memory).
-    store: Store<BatchDigest, SerializedBatchMessage>,
+    store: Store<BatchDigest, Batch>,
     /// The (global) state to perform execution.
     execution_state: Arc<State>,
     /// Receive reconfiguration updates.
@@ -58,7 +54,7 @@ where
     /// Spawn a new executor in a dedicated tokio task.
     #[must_use]
     pub fn spawn(
-        store: Store<BatchDigest, SerializedBatchMessage>,
+        store: Store<BatchDigest, Batch>,
         execution_state: Arc<State>,
         rx_reconfigure: watch::Receiver<ReconfigureNotification>,
         rx_subscriber: metered_channel::Receiver<ConsensusOutput>,
@@ -138,8 +134,8 @@ where
         total_batches: usize,
     ) -> SubscriberResult<()> {
         // The store should now hold all transaction data referenced by the input certificate.
-        let batch = match self.store.read(batch_digest).await? {
-            Some(x) => x,
+        let transactions = match self.store.read(batch_digest).await? {
+            Some(x) => x.0,
             None => {
                 // If two certificates contain the exact same batch (eg. by the actions of a Byzantine
                 // consensus node), some correct client may already have deleted the batch from their
@@ -152,12 +148,6 @@ where
             }
         };
 
-        // Deserialize the consensus workers' batch message to retrieve a list of transactions.
-        let transactions = match bincode::deserialize(&batch)? {
-            WorkerMessage::Batch(Batch(x)) => x,
-            _ => bail!(SubscriberError::UnexpectedProtocolMessage),
-        };
-
         // Execute every transaction in the batch.
         let total_transactions = transactions.len();
         for (index, transaction) in transactions.into_iter().enumerate() {

executor/src/errors.rs

@@ -4,6 +4,7 @@ use config::WorkerId;
 use std::fmt::Debug;
 use store::StoreError;
 use thiserror::Error;
+use types::CertificateDigest;
 
 #[macro_export]
 macro_rules! bail {
@@ -28,6 +29,9 @@ pub enum SubscriberError {
     #[error("Storage failure: {0}")]
     StoreError(#[from] StoreError),
 
+    #[error("Error occurred while retrieving certificate {0} payload: {1}")]
+    PayloadRetrieveError(CertificateDigest, String),
+
     #[error("Consensus referenced unexpected worker id {0}")]
     UnexpectedWorkerId(WorkerId),