Fix updater.

Author: afck

linera-core/src/chain_worker/state.rs

@@ -762,61 +762,35 @@ where
         {
             certificate.check(committee)?;
         } else {
-            // Inline committees_for logic with specific error handling.
             let net_description = self
                 .storage
                 .read_network_description()
                 .await?
                 .ok_or_else(|| WorkerError::MissingNetworkDescription)?;
-            let admin_chain_id = net_description.admin_chain_id;
-
-            let committee = if epoch == Epoch::ZERO {
+            let committee_hash = if epoch == Epoch::ZERO {
                 // Genesis epoch is stored in NetworkDescription.
-                let blob_id = BlobId::new(
-                    net_description.genesis_committee_blob_hash,
-                    BlobType::Committee,
-                );
-                let committee_blob = self
-                    .storage
-                    .read_blob(blob_id)
-                    .await?
-                    .ok_or_else(|| WorkerError::BlobsNotFound(vec![blob_id]))?;
-                bcs::from_bytes::<Committee>(committee_blob.bytes())
-                    .map_err(|e| WorkerError::from(linera_views::ViewError::from(e)))?
+                net_description.genesis_committee_blob_hash
             } else {
                 // Read the epoch creation event.
-                let epoch_creation_events = self
-                    .storage
-                    .read_events_from_index(
-                        &admin_chain_id,
-                        &StreamId::system(EPOCH_STREAM_NAME),
-                        epoch.0,
-                    )
-                    .await?;
-
-                let event = epoch_creation_events
-                    .into_iter()
-                    .find(|index_and_event| index_and_event.index == epoch.0)
-                    .ok_or_else(|| {
-                        WorkerError::EventsNotFound(vec![EventId {
-                            chain_id: admin_chain_id,
-                            stream_id: StreamId::system(EPOCH_STREAM_NAME),
-                            index: epoch.0,
-                        }])
-                    })?;
-
-                let committee_hash: CryptoHash = bcs::from_bytes(&event.event)
-                    .map_err(|e| WorkerError::from(linera_views::ViewError::from(e)))?;
-                let blob_id = BlobId::new(committee_hash, BlobType::Committee);
-                let committee_blob = self
+                let event_id = EventId {
+                    chain_id: net_description.admin_chain_id,
+                    stream_id: StreamId::system(EPOCH_STREAM_NAME),
+                    index: epoch.0,
+                };
+                let event = self
                     .storage
-                    .read_blob(blob_id)
+                    .read_event(event_id.clone())
                     .await?
-                    .ok_or_else(|| WorkerError::BlobsNotFound(vec![blob_id]))?;
-                bcs::from_bytes::<Committee>(committee_blob.bytes())
-                    .map_err(|e| WorkerError::from(linera_views::ViewError::from(e)))?
+                    .ok_or_else(|| WorkerError::EventsNotFound(vec![event_id]))?;
+                bcs::from_bytes(&event)?
             };
-            // This line is duplicated, but this avoids cloning and a lifetimes error.
+            let blob_id = BlobId::new(committee_hash, BlobType::Committee);
+            let committee_blob = self
+                .storage
+                .read_blob(blob_id)
+                .await?
+                .ok_or_else(|| WorkerError::BlobsNotFound(vec![blob_id]))?;
+            let committee = bcs::from_bytes(committee_blob.bytes())?;
             certificate.check(&committee)?;
         }
 

linera-core/src/updater.rs

@@ -248,8 +248,7 @@ where
         loop {
             match result {
                 Err(NodeError::EventsNotFound(event_ids))
-                    if self.admin_id != certificate.block().header.chain_id
-                        && !sent_admin_chain
+                    if !sent_admin_chain
                         && certificate.inner().chain_id() != self.admin_id
                         && event_ids.iter().all(|event_id| {
                             event_id.stream_id == StreamId::system(EPOCH_STREAM_NAME)
@@ -260,18 +259,6 @@ where
                     self.update_admin_chain().await?;
                     sent_admin_chain = true;
                 }
-                Err(NodeError::BlobsNotFound(blob_ids))
-                    if blob_ids
-                        .iter()
-                        .all(|id| id.blob_type == BlobType::Committee) =>
-                {
-                    if sent_admin_chain || self.admin_id == certificate.block().header.chain_id {
-                        return Err(NodeError::BlobsNotFound(blob_ids).into());
-                    }
-                    // The validator doesn't have the committee that signed the certificate.
-                    self.update_admin_chain().await?;
-                    sent_admin_chain = true;
-                }
                 Err(NodeError::BlobsNotFound(blob_ids)) if !sent_blobs => {
                     // The validator is missing the blobs required by the certificate.
                     self.remote_node
@@ -306,7 +293,7 @@ where
             .await;
 
         let chain_id = certificate.inner().chain_id();
-        Ok(match &result {
+        match &result {
             Err(original_err @ NodeError::BlobsNotFound(blob_ids)) => {
                 self.remote_node
                     .check_blobs_not_found(&certificate, blob_ids)?;
@@ -319,9 +306,6 @@ where
                     .await?
                     .ok_or_else(|| original_err.clone())?;
                 self.remote_node.send_pending_blobs(chain_id, blobs).await?;
-                self.remote_node
-                    .handle_validated_certificate(certificate)
-                    .await
             }
             Err(error) => {
                 self.sync_if_needed(
@@ -331,12 +315,13 @@ where
                     error,
                 )
                 .await?;
-                self.remote_node
-                    .handle_validated_certificate(certificate)
-                    .await
             }
-            _ => result,
-        }?)
+            _ => return Ok(result?),
+        }
+        Ok(self
+            .remote_node
+            .handle_validated_certificate(certificate)
+            .await?)
     }
 
     /// Requests a vote for a timeout certificate for the given round from the remote node.
@@ -616,11 +601,7 @@ where
         target_block_height: BlockHeight,
         delivery: CrossChainMessageDelivery,
     ) -> Result<(), chain_client::Error> {
-        let info = if let Some(height) = target_block_height
-            .try_sub_one()
-            .ok()
-            .filter(|_| chain_id != self.admin_id)
-        {
+        let info = if let Ok(height) = target_block_height.try_sub_one() {
             // Figure out which certificates this validator is missing. In many cases, it's just the
             // last one, so we optimistically send that one right away.
             let hash = self
@@ -644,9 +625,17 @@ where
                 .read_certificate(hash)
                 .await?
                 .ok_or_else(|| chain_client::Error::MissingConfirmedBlock(hash))?;
-            let info = self
-                .send_confirmed_certificate(certificate, delivery)
-                .await?;
+            let info = match self.send_confirmed_certificate(certificate, delivery).await {
+                Ok(info) => info,
+                Err(error) => {
+                    tracing::debug!(
+                        address = self.remote_node.address(), %error,
+                        "validator failed to handle confirmed certificate; sending whole chain",
+                    );
+                    let query = ChainInfoQuery::new(chain_id);
+                    self.remote_node.handle_chain_info_query(query).await?
+                }
+            };
             // Obtain the missing blocks and the manager state from the local node.
             let heights = (info.next_block_height.0..target_block_height.0).map(BlockHeight);
             let validator_missing_hashes = self

linera-core/src/worker.rs

@@ -172,6 +172,9 @@ pub enum WorkerError {
     #[error(transparent)]
     ChainError(#[from] Box<ChainError>),
 
+    #[error(transparent)]
+    BcsError(#[from] bcs::Error),
+
     // Chain access control
     #[error("Block was not signed by an authorized owner")]
     InvalidOwner,
@@ -279,7 +282,8 @@ impl WorkerError {
             | WorkerError::UnexpectedBlob
             | WorkerError::TooManyPublishedBlobs(_)
             | WorkerError::ViewError(ViewError::NotFound(_)) => false,
-            WorkerError::InvalidCrossChainRequest
+            WorkerError::BcsError(_)
+            | WorkerError::InvalidCrossChainRequest
             | WorkerError::ViewError(_)
             | WorkerError::ConfirmedLogEntryNotFound { .. }
             | WorkerError::PreprocessedBlocksEntryNotFound { .. }

linera-service/src/cli/main.rs

@@ -2200,6 +2200,7 @@ fn main() -> anyhow::Result<process::ExitCode> {
         builder
     };
 
+    runtime.thread_stack_size(4 << 20);
     if let Some(blocking_threads) = options.tokio_blocking_threads {
         runtime.max_blocking_threads(blocking_threads);
     }