Fix multi → single leader transition.

Author: afck

linera-chain/src/manager.rs

@@ -379,18 +379,16 @@ where
             return Ok(false); // We are not a validator.
         };
         ensure!(
-            round <= self.current_round(),
+            round == self.current_round(),
             ChainError::WrongRound(self.current_round())
         );
-        if round == self.current_round() {
-            let Some(round_timeout) = *self.round_timeout.get() else {
-                return Err(ChainError::RoundDoesNotTimeOut);
-            };
-            ensure!(
-                local_time >= round_timeout,
-                ChainError::NotTimedOutYet(round_timeout)
-            );
-        }
+        let Some(round_timeout) = *self.round_timeout.get() else {
+            return Err(ChainError::RoundDoesNotTimeOut);
+        };
+        ensure!(
+            local_time >= round_timeout,
+            ChainError::NotTimedOutYet(round_timeout)
+        );
         if let Some(vote) = self.timeout_vote.get() {
             if vote.round == round {
                 return Ok(false); // We already signed this timeout.
@@ -580,6 +578,12 @@ where
                     .iter()
                     .map(|proposal| proposal.content.round),
             )
+            .chain(
+                self.signed_proposal
+                    .get()
+                    .iter()
+                    .map(|proposal| proposal.content.round),
+            )
             .max()
             .unwrap_or_default()
             .max(self.ownership.get().first_round());
@@ -687,7 +691,7 @@ where
     /// Sets the signed proposal, if it is newer than the known one, and not from a single-leader
     /// round. Returns whether it was updated.
     pub fn update_signed_proposal(&mut self, proposal: &BlockProposal) -> bool {
-        if proposal.content.round > Round::MultiLeader(u32::MAX) {
+        if proposal.content.round > Round::SingleLeader(0) {
             return false;
         }
         if let Some(old_proposal) = self.signed_proposal.get() {
@@ -765,7 +769,6 @@ pub struct ChainManagerInfo {
     #[debug(skip_if = Option::is_none)]
     pub requested_locking: Option<Box<LockingBlock>>,
     /// Latest timeout certificate we have seen.
-    #[debug(skip_if = Option::is_none)]
     pub timeout: Option<Box<TimeoutCertificate>>,
     /// Latest vote we cast (either to validate or to confirm a block).
     #[debug(skip_if = Option::is_none)]

linera-core/src/client/mod.rs

@@ -993,7 +993,10 @@ impl<Env: Environment> Client<Env> {
         let query = ChainInfoQuery::new(chain_id).with_manager_values();
         let remote_info = remote_node.handle_chain_info_query(query).await?;
 
-        info!("{:?}", remote_info);
+        info!(
+            "Info from {}: {:?}",
+            remote_node.public_key, remote_info.manager
+        );
 
         if let Some(new_info) = self
             .download_certificates_from(remote_node, chain_id, remote_info.next_block_height)
@@ -1920,15 +1923,15 @@ impl<Env: Environment> ChainClient<Env> {
 
         self.synchronize_to_known_height().await?;
 
-//        if self.has_other_owners(&info.manager.ownership) {
-            // For chains with any owner other than ourselves, we could be missing recent
-            // certificates created by other owners. Further synchronize blocks from the network.
-            // This is a best-effort that depends on network conditions.
+        //        if self.has_other_owners(&info.manager.ownership) {
+        // For chains with any owner other than ourselves, we could be missing recent
+        // certificates created by other owners. Further synchronize blocks from the network.
+        // This is a best-effort that depends on network conditions.
         let info = self
             .client
             .maybe_synchronize_chain_state(self.chain_id)
             .await?;
-//        }
+        //        }
 
         if info.epoch > self.client.admin_committees().await?.0 {
             self.client
@@ -3035,7 +3038,10 @@ impl<Env: Environment> ChainClient<Env> {
             .any(|proposal| proposal.content.round == manager.current_round)
             || (manager.current_round.is_fast() && has_oracle_responses);
         info!("conflict: {}", conflict);
-        info!("next candidate round: {:?}", manager.ownership.next_round(manager.current_round));
+        info!(
+            "next candidate round: {:?}",
+            manager.ownership.next_round(manager.current_round)
+        );
 
         let round = if !conflict {
             manager.current_round
@@ -3052,7 +3058,12 @@ impl<Env: Environment> ChainClient<Env> {
                 "Conflicting proposal in the current round",
             ));
         };
-        info!("can {} propose in round {}? {}", identity, round, manager.can_propose(identity, round));
+        info!(
+            "can {} propose in round {}? {}",
+            identity,
+            round,
+            manager.can_propose(identity, round)
+        );
         if manager.can_propose(identity, round) {
             return Ok(Either::Left(round));
         }

linera-core/src/remote_node.rs

@@ -113,7 +113,7 @@ impl<N: ValidatorNode> RemoteNode<N> {
                 .await;
             match result {
                 Err(NodeError::MissingCertificateValue) => {
-                    warn!(
+                    tracing::debug!(
                         "Validator {} forgot a certificate value that they signed before",
                         self.public_key
                     );
@@ -135,7 +135,7 @@ impl<N: ValidatorNode> RemoteNode<N> {
                 .await;
             match result {
                 Err(NodeError::MissingCertificateValue) => {
-                    warn!(
+                    tracing::debug!(
                         "Validator {} forgot a certificate value that they signed before",
                         self.public_key
                     );

linera-core/src/updater.rs

@@ -22,6 +22,7 @@ use linera_base::{
 };
 use linera_chain::{
     data_types::{BlockProposal, LiteVote},
+    manager::LockingBlock,
     types::{ConfirmedBlock, GenericCertificate, ValidatedBlock, ValidatedBlockCertificate},
 };
 use linera_execution::{committee::Committee, system::EPOCH_STREAM_NAME};
@@ -451,6 +452,7 @@ where
             } else {
                 tracing::warn!("NOPE 0");
             }
+            // TODO: Deduplicate!!
             return Ok(());
         };
         // Figure out which certificates this validator is missing. In many cases, it's just the
@@ -517,7 +519,8 @@ where
             }
         }
         // If the remote node is missing a timeout certificate, send it as well.
-        let local_info = self.local_node.chain_info(chain_id).await?;
+        let query = ChainInfoQuery::new(chain_id).with_manager_values();
+        let local_info = self.local_node.handle_chain_info_query(query).await?.info;
         if let Some(cert) = local_info.manager.timeout {
             if (local_info.next_block_height, cert.round) >= (remote_height, remote_round) {
                 tracing::debug!("send_chain_information --> {:?}", cert);
@@ -528,6 +531,36 @@ where
         } else {
             tracing::debug!("send_chain_information NOPE 2");
         }
+        if let Some(proposal) = local_info.manager.requested_proposed {
+            tracing::info!("send_chain_information prop --> {:?}", proposal);
+            if let Err(err) = Box::pin(self.send_block_proposal(proposal, vec![])).await {
+                tracing::info!("Failed to propose: {err}");
+            }
+        }
+        if let Some(proposal) = local_info.manager.requested_signed_proposal {
+            tracing::info!("send_chain_information sig --> {:?}", proposal);
+            if let Err(err) = Box::pin(self.send_block_proposal(proposal, vec![])).await {
+                tracing::info!("Failed to propose: {err}");
+            }
+        }
+        match local_info.manager.requested_locking.map(|b| *b) {
+            Some(LockingBlock::Regular(validated)) => {
+                if let Err(err) = self
+                    .remote_node
+                    .handle_optimized_validated_certificate(
+                        &validated,
+                        CrossChainMessageDelivery::NonBlocking,
+                    )
+                    .await
+                {
+                    tracing::info!("Failed to send locking block: {err}");
+                }
+            }
+            Some(LockingBlock::Fast(_)) => {
+                tracing::error!("TODO: FAST")
+            }
+            None => {}
+        }
         Ok(())
     }
 
@@ -605,7 +638,50 @@ where
             }
             CommunicateAction::RequestTimeout { round, height, .. } => {
                 let query = ChainInfoQuery::new(chain_id).with_timeout(height, round);
-                let info = self.remote_node.handle_chain_info_query(query).await?;
+                let info = match self
+                    .remote_node
+                    .handle_chain_info_query(query.clone())
+                    .await
+                {
+                    Ok(info) => info,
+                    Err(NodeError::WrongRound(validator_round)) => {
+                        tracing::info!(
+                            "Failed to request timeout from validator {} because it sees \
+                            chain {} at round {} instead of {}.",
+                            self.remote_node.public_key,
+                            chain_id,
+                            validator_round,
+                            round
+                        );
+                        // The timeout is for a different round, so we need to update the validator.
+                        // TODO: this should probably be more specific as to which rounds are retried.
+                        self.send_chain_information(
+                            chain_id,
+                            height,
+                            CrossChainMessageDelivery::NonBlocking,
+                        )
+                        .await?;
+                        self.remote_node
+                            .handle_chain_info_query(query.clone())
+                            .await?
+                    }
+                    Err(NodeError::UnexpectedBlockHeight {
+                        expected_block_height,
+                        found_block_height,
+                    }) if expected_block_height < found_block_height => {
+                        // The proposal is for a later block height, so we need to update the validator.
+                        self.send_chain_information(
+                            chain_id,
+                            found_block_height,
+                            CrossChainMessageDelivery::NonBlocking,
+                        )
+                        .await?;
+                        self.remote_node
+                            .handle_chain_info_query(query.clone())
+                            .await?
+                    }
+                    Err(err) => Err(err)?,
+                };
                 info.manager.timeout_vote.ok_or_else(|| {
                     NodeError::MissingVoteInValidatorResponse("request a timeout".into())
                 })?

linera-rpc/src/grpc/client.rs

@@ -75,7 +75,7 @@ impl GrpcClient {
                 true
             }
             Code::Ok | Code::Cancelled | Code::ResourceExhausted => {
-                info!("Unexpected gRPC status: {}; retrying", status);
+                debug!("Unexpected gRPC status: {}; retrying", status);
                 true
             }
             Code::NotFound => false, // This code is used if e.g. the validator is missing blobs.