Merge pull request #13395 from starkware-libs/dori/merge-main-v0.14.1-committer-into-main-v0.14.2-1774177856

Merge main-v0.14.1-committer into main-v0.14.2

Author: Yoni-Starkware

Cargo.lock

@@ -445,7 +445,9 @@ impl<ContextT: ConsensusContext> MultiHeightManager<ContextT> {
             },
 
             Err(err) => match err {
-                e @ ConsensusError::BatcherError(_) | e @ ConsensusError::CommitteeError(_) => {
+                e @ ConsensusError::BatcherError(_)
+                | e @ ConsensusError::CommitteeError(_)
+                | e @ ConsensusError::CendeAheadOfProposalHeight(_) => {
                     error!(
                         "Error while running consensus for height {height}, fallback to sync: {e}"
                     );

crates/apollo_consensus/src/manager.rs

@@ -142,6 +142,8 @@ pub enum ConsensusError {
     BatcherError(#[from] BatcherClientError),
     #[error("Committee error: {0}")]
     CommitteeError(String),
+    #[error("Cende recorder height >= proposal height ({0}); failing proposal")]
+    CendeAheadOfProposalHeight(BlockNumber),
     // Indicates an error in communication between consensus and the node's networking component.
     // As opposed to an error between this node and peer nodes.
     #[error("{0}")]

crates/apollo_consensus/src/types.rs

@@ -33,7 +33,7 @@ use reqwest::Response;
 use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, RequestBuilder};
 use reqwest_retry::policies::ExponentialBackoff;
 use reqwest_retry::{Jitter, RetryTransientMiddleware};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use shared_execution_objects::central_objects::CentralTransactionExecutionInfo;
 use starknet_api::block::{BlockHashAndNumber, BlockInfo, BlockNumber, StarknetVersion};
 use starknet_api::consensus_transaction::InternalConsensusTransaction;
@@ -100,6 +100,10 @@ pub trait CendeContext: Send + Sync {
     /// This function should return false if the previous height blob is not available.
     fn write_prev_height_blob(&self, current_height: BlockNumber) -> JoinHandle<bool>;
 
+    /// Returns the latest received block number from the recorder, or `None` if no blocks or on
+    /// error.
+    async fn get_latest_received_block(&self) -> Option<BlockNumber>;
+
     // Prepares the previous height blob that will be written in the next height.
     async fn prepare_blob_for_next_height(
         &self,
@@ -113,13 +117,23 @@ pub struct CendeAmbassador {
     // `None` indicates that there is no blob to write, and therefore, the node can't be the
     // proposer.
     prev_height_blob: Arc<Mutex<Option<AerospikeBlob>>>,
-    url: Url,
+    write_blob_url: Url,
+    get_latest_received_block_url: Url,
     client: ClientWithMiddleware,
     class_manager: SharedClassManagerClient,
 }
 
 /// The path to write blob in the Recorder.
 pub const RECORDER_WRITE_BLOB_PATH: &str = "/cende_recorder/write_blob";
+/// The path to get the latest received block from the Recorder (the next block that will be written
+/// to DB. returns null when no blocks exist).
+pub const RECORDER_GET_LATEST_RECEIVED_BLOCK_PATH: &str =
+    "/cende_recorder/get_latest_received_block";
+
+#[derive(Debug, Deserialize)]
+struct GetLatestReceivedBlockResponse {
+    block_number: Option<u64>,
+}
 
 impl CendeAmbassador {
     pub fn new(cende_config: CendeConfig, class_manager: SharedClassManagerClient) -> Self {
@@ -130,12 +144,14 @@ impl CendeAmbassador {
 
         CendeAmbassador {
             prev_height_blob: Arc::new(Mutex::new(None)),
-            url: {
-                let mut recorder_url = cende_config.recorder_url;
-                recorder_url =
-                    recorder_url.join(RECORDER_WRITE_BLOB_PATH).expect("Failed to construct URL");
-                recorder_url
-            },
+            write_blob_url: cende_config
+                .recorder_url
+                .join(RECORDER_WRITE_BLOB_PATH)
+                .expect("Failed to construct write blob URL"),
+            get_latest_received_block_url: cende_config
+                .recorder_url
+                .join(RECORDER_GET_LATEST_RECEIVED_BLOCK_PATH)
+                .expect("Failed to construct get latest received block URL"),
             client: ClientBuilder::new(reqwest::Client::new())
                 .with(RetryTransientMiddleware::new_with_policy(retry_policy))
                 .build(),
@@ -150,7 +166,7 @@ impl CendeContext for CendeAmbassador {
         info!("Start writing to Aerospike previous height blob for height {current_height}.");
 
         let prev_height_blob = self.prev_height_blob.clone();
-        let request_builder = self.client.post(self.url.clone());
+        let request_builder = self.client.post(self.write_blob_url.clone());
 
         task::spawn(
             async move {
@@ -189,6 +205,33 @@ impl CendeContext for CendeAmbassador {
         )
     }
 
+    async fn get_latest_received_block(&self) -> Option<BlockNumber> {
+        match self.client.get(self.get_latest_received_block_url.clone()).send().await {
+            Ok(response) if response.status().is_success() => {
+                match response.json::<GetLatestReceivedBlockResponse>().await {
+                    Ok(resp) => resp.block_number.map(BlockNumber),
+                    Err(e) => {
+                        warn!("Failed to parse recorder get_latest_received_block response: {e}");
+                        None
+                    }
+                }
+            }
+            // Response is not successful: log and return None.
+            Ok(response) => {
+                warn!(
+                    "Recorder get_latest_received_block returned error status {}: {}",
+                    response.status(),
+                    response.text().await.unwrap_or_else(|_| "unparseable".to_string())
+                );
+                None
+            }
+            Err(e) => {
+                warn!("Failed to request recorder get_latest_received_block: {e}");
+                None
+            }
+        }
+    }
+
     #[sequencer_latency_histogram(CENDE_PREPARE_BLOB_FOR_NEXT_HEIGHT_LATENCY, false)]
     async fn prepare_blob_for_next_height(
         &self,

crates/apollo_consensus_orchestrator/src/cende/mod.rs

@@ -114,6 +114,17 @@ type ProposalContent =
 type HeightToRoundToProposal =
     BTreeMap<BlockNumber, BTreeMap<Round, (ProposalCommitment, ProposalContent)>>;
 
+/// Cende status relative to the block we're building.
+#[derive(Debug)]
+pub(crate) enum CendeStatus {
+    /// Cende has current block (or higher) written; fail the proposal round.
+    CurrentBlob,
+    /// Cende has prev block written; skip writing, proceed with build.
+    PrevBlob,
+    /// Prev blob not yet written; must write before building.
+    PrevBlobMissing,
+}
+
 pub(crate) struct BuiltProposals {
     // {height: {proposal_commitment: (init, content, proposal_id, finished_info)}}
     // Note that multiple proposals IDs can be associated with the same content, but we only need
@@ -289,26 +300,21 @@ impl SequencerConsensusContext {
         StreamSender { proposal_sender }
     }
 
-    async fn get_latest_sync_height(&self) -> Option<BlockNumber> {
-        match self.deps.state_sync_client.get_latest_block_number().await {
-            Ok(height) => height,
-            Err(e) => {
-                error!("Failed to get latest sync height: {e:?}");
-                None
-            }
-        }
-    }
-
-    async fn can_skip_write_prev_height_blob(&self, height: BlockNumber) -> bool {
+    async fn get_cende_status(&self, height: BlockNumber) -> CendeStatus {
         if height == BlockNumber(0) {
-            return true;
+            return CendeStatus::PrevBlob;
         }
-        match self.get_latest_sync_height().await {
-            Some(latest_sync_height) => {
-                latest_sync_height
-                    >= height.prev().expect("Height should be greater than 0. Checked above.")
+        match self.deps.cende_ambassador.get_latest_received_block().await {
+            Some(latest_cende_block) => {
+                if latest_cende_block >= height {
+                    CendeStatus::CurrentBlob
+                } else if latest_cende_block >= height.prev().expect("height > 0 has predecessor") {
+                    CendeStatus::PrevBlob
+                } else {
+                    CendeStatus::PrevBlobMissing
+                }
             }
-            None => false,
+            None => CendeStatus::PrevBlobMissing,
         }
     }
 
@@ -535,18 +541,20 @@ impl ConsensusContext for SequencerConsensusContext {
         build_param: BuildParam,
         timeout: Duration,
     ) -> Result<oneshot::Receiver<ProposalCommitment>, ConsensusError> {
-        let cende_write_success = if self.can_skip_write_prev_height_blob(build_param.height).await
-        {
-            // cende_write_success is a AbortOnDropHandle. To get the actual handle we need to
-            // spawn the task.
-            AbortOnDropHandle::new(tokio::spawn(ready(true)))
-        } else {
-            // TODO(dvir): consider start writing the blob in `decision_reached`, to reduce
-            // transactions finality time. Use this option only for one special
-            // sequencer that is the same cluster as the recorder.
-            AbortOnDropHandle::new(
-                self.deps.cende_ambassador.write_prev_height_blob(build_param.height),
-            )
+        let status = self.get_cende_status(build_param.height).await;
+        let cende_write_success = match status {
+            CendeStatus::CurrentBlob => {
+                return Err(ConsensusError::CendeAheadOfProposalHeight(build_param.height));
+            }
+            CendeStatus::PrevBlob => AbortOnDropHandle::new(tokio::spawn(ready(true))),
+            CendeStatus::PrevBlobMissing => {
+                // TODO(dvir): consider start writing the blob in `decision_reached`, to reduce
+                // transactions finality time. Use this option only for one special
+                // sequencer that is the same cluster as the recorder.
+                AbortOnDropHandle::new(
+                    self.deps.cende_ambassador.write_prev_height_blob(build_param.height),
+                )
+            }
         };
 
         // Handles interrupting an active proposal from a previous height/round