Fix missing committee blob handling. (#4978) (#4993)

Partial backport of #4978.

## Motivation

The current `Committee` implementation is a bit complicated, and
sometimes sets the quorum threshold higher than necessary.

## Proposal

#4978 revealed an issue with the certificate handling logic in the
worker: `committees_for` returns `ViewErrors` instead of
`BlobsNotFound`/`EventsNotFound`, so the client fails to send the
validators the admin chain if needed.

This was fixed; I also kept minor cleanups I made while debugging this.
I got a few stack overflows, so I doubled the client's Tokio thread
stack size.

## Test Plan

CI

## Release Plan

- The fixes (but maybe not the quorum change) should be
    - released in a new SDK and
    - released in a validator hotfix.

## Links

- PR to main: #4978
- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: afck

linera-core/src/chain_worker/state.rs

@@ -18,7 +18,7 @@ use linera_base::{
     },
     ensure,
     hashed::Hashed,
-    identifiers::{AccountOwner, ApplicationId, BlobId, BlobType, ChainId, EventId, StreamId},
+    identifiers::{AccountOwner, ApplicationId, BlobId, BlobType, ChainId},
 };
 use linera_chain::{
     data_types::{
@@ -30,7 +30,7 @@ use linera_chain::{
     ChainError, ChainExecutionContext, ChainStateView, ExecutionResultExt as _,
 };
 use linera_execution::{
-    system::EPOCH_STREAM_NAME, Committee, ExecutionStateView, Query, QueryContext, QueryOutcome,
+    Committee, ExecutionRuntimeContext as _, ExecutionStateView, Query, QueryContext, QueryOutcome,
     ServiceRuntimeEndpoint,
 };
 use linera_storage::{Clock as _, ResultReadCertificates, Storage};
@@ -659,6 +659,14 @@ where
         // Check that the chain is active and ready for this validated block.
         // Verify the certificate. Returns a catch-all error to make client code more robust.
         self.initialize_and_save_if_needed().await?;
+        let tip_state = self.chain.tip_state.get();
+        ensure!(
+            header.height == tip_state.next_block_height,
+            ChainError::UnexpectedBlockHeight {
+                expected_block_height: tip_state.next_block_height,
+                found_block_height: header.height,
+            }
+        );
         let (epoch, committee) = self.chain.current_committee()?;
         check_block_epoch(epoch, header.chain_id, header.epoch)?;
         certificate.check(committee)?;
@@ -747,21 +755,23 @@ where
         {
             certificate.check(committee)?;
         } else {
-            let committees = self.storage.committees_for(epoch..=epoch).await?;
-            let Some(committee) = committees.get(&epoch) else {
-                let net_description = self
-                    .storage
-                    .read_network_description()
-                    .await?
-                    .ok_or_else(|| WorkerError::MissingNetworkDescription)?;
-                return Err(WorkerError::EventsNotFound(vec![EventId {
-                    chain_id: net_description.admin_chain_id,
-                    stream_id: StreamId::system(EPOCH_STREAM_NAME),
-                    index: epoch.0,
-                }]));
-            };
-            // This line is duplicated, but this avoids cloning and a lifetimes error.
-            certificate.check(committee)?;
+            let committee = self
+                .chain
+                .execution_state
+                .context()
+                .extra()
+                .get_committees(epoch..=epoch)
+                .await
+                .map_err(|error| {
+                    ChainError::ExecutionError(Box::new(error), ChainExecutionContext::Block)
+                })?
+                .remove(&epoch)
+                .ok_or_else(|| {
+                    ChainError::InternalError(format!(
+                        "missing committee for epoch {epoch}; this is a bug"
+                    ))
+                })?;
+            certificate.check(&committee)?;
         }
 
         // Certificate check passed - which means the blobs the block requires are legitimate and
@@ -836,17 +846,6 @@ where
             .filter_map(|blob_id| blobs.remove(blob_id))
             .collect::<Vec<_>>();
 
-        // If height is zero, we haven't initialized the chain state or verified the epoch before -
-        // do it now.
-        // This will fail if the chain description blob is still missing - but that's alright,
-        // because we already wrote the blob state above, so the client can now upload the
-        // blob, which will get accepted, and retry.
-        if height == BlockHeight::ZERO {
-            self.initialize_and_save_if_needed().await?;
-            let (epoch, _) = self.chain.current_committee()?;
-            check_block_epoch(epoch, chain_id, block.header.epoch)?;
-        }
-
         // Execute the block and update inboxes.
         let local_time = self.storage.clock().current_time();
         let chain = &mut self.chain;
@@ -1382,6 +1381,8 @@ where
         } = &proposal;
         let block = &content.block;
         let chain = &self.chain;
+        // Check if the chain is ready for this new block proposal.
+        chain.tip_state.get().verify_block_chaining(block)?;
         // Check the epoch.
         let (epoch, committee) = chain.current_committee()?;
         check_block_epoch(epoch, block.chain_id, block.epoch)?;
@@ -1431,8 +1432,6 @@ where
                 original_proposal.check_signature()?;
             }
         }
-        // Check if the chain is ready for this new block proposal.
-        chain.tip_state.get().verify_block_chaining(block)?;
         if chain.manager.check_proposed_block(&proposal)? == manager::Outcome::Skip {
             // We already voted for this block.
             return Ok((self.chain_info_response(), NetworkActions::default()));

linera-core/src/updater.rs

@@ -263,7 +263,7 @@ where
         let mut sent_admin_chain = false;
         let mut sent_blobs = false;
         loop {
-            result = match result {
+            match result {
                 Err(NodeError::EventsNotFound(event_ids))
                     if !sent_admin_chain
                         && certificate.inner().chain_id() != self.admin_id
@@ -275,9 +275,6 @@ where
                     // The validator doesn't have the committee that signed the certificate.
                     self.update_admin_chain().await?;
                     sent_admin_chain = true;
-                    self.remote_node
-                        .handle_confirmed_certificate(certificate.clone(), delivery)
-                        .await
                 }
                 Err(NodeError::BlobsNotFound(blob_ids)) if !sent_blobs => {
                     // The validator is missing the blobs required by the certificate.
@@ -292,12 +289,13 @@ where
                     let blobs = maybe_blobs.ok_or(NodeError::BlobsNotFound(blob_ids))?;
                     self.remote_node.node.upload_blobs(blobs).await?;
                     sent_blobs = true;
-                    self.remote_node
-                        .handle_confirmed_certificate(certificate.clone(), delivery)
-                        .await
                 }
                 result => return Ok(result?),
-            };
+            }
+            result = self
+                .remote_node
+                .handle_confirmed_certificate(certificate.clone(), delivery)
+                .await;
         }
     }
 
@@ -312,7 +310,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)?;
@@ -325,9 +323,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(
@@ -337,12 +332,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.
@@ -658,22 +654,15 @@ where
                     .ok_or_else(|| ChainClientError::MissingConfirmedBlock(hash))?
             };
             let info = match self.send_confirmed_certificate(certificate, delivery).await {
-                Err(ChainClientError::RemoteNodeError(NodeError::EventsNotFound(event_ids)))
-                    if event_ids.iter().all(|event_id| {
-                        event_id.stream_id == StreamId::system(EPOCH_STREAM_NAME)
-                            && event_id.chain_id == self.admin_id
-                    }) =>
-                {
-                    if chain_id != self.admin_id {
-                        tracing::error!(
-                            "Missing epochs were not handled by send_confirmed_certificate."
-                        );
-                    }
+                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?
                 }
-                Err(err) => return Err(err),
-                Ok(info) => info,
             };
             // 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);

linera-core/src/worker.rs

@@ -171,6 +171,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,
@@ -278,7 +281,8 @@ impl WorkerError {
             | WorkerError::UnexpectedBlob
             | WorkerError::TooManyPublishedBlobs(_)
             | WorkerError::ViewError(ViewError::NotFound(_)) => false,
-            WorkerError::InvalidCrossChainRequest
+            WorkerError::BcsError(_)
+            | WorkerError::InvalidCrossChainRequest
             | WorkerError::ViewError(_)
             | WorkerError::ConfirmedLogEntryNotFound { .. }
             | WorkerError::PreprocessedBlocksEntryNotFound { .. }
@@ -296,16 +300,14 @@ impl From<ChainError> for WorkerError {
     #[instrument(level = "trace", skip(chain_error))]
     fn from(chain_error: ChainError) -> Self {
         match chain_error {
-            ChainError::ExecutionError(execution_error, context) => {
-                if let ExecutionError::BlobsNotFound(blob_ids) = *execution_error {
-                    Self::BlobsNotFound(blob_ids)
-                } else {
-                    Self::ChainError(Box::new(ChainError::ExecutionError(
-                        execution_error,
-                        context,
-                    )))
-                }
-            }
+            ChainError::ExecutionError(execution_error, context) => match *execution_error {
+                ExecutionError::BlobsNotFound(blob_ids) => Self::BlobsNotFound(blob_ids),
+                ExecutionError::EventsNotFound(event_ids) => Self::EventsNotFound(event_ids),
+                _ => Self::ChainError(Box::new(ChainError::ExecutionError(
+                    execution_error,
+                    context,
+                ))),
+            },
             error => Self::ChainError(Box::new(error)),
         }
     }

linera-execution/src/committee.rs

@@ -68,7 +68,8 @@ pub struct Committee {
     total_votes: u64,
     /// The threshold to form a quorum.
     quorum_threshold: u64,
-    /// The threshold to prove the validity of a statement.
+    /// The threshold to prove the validity of a statement. I.e. the assumption is that strictly
+    /// less than `validity_threshold` are faulty.
     validity_threshold: u64,
     /// The policy agreed on for this epoch.
     policy: ResourceControlPolicy,