Leave gaps in sender chains on the validator side (#4181)

## Motivation

If a client creates a block receiving messages from a chain with gaps,
and a validator doesn't have that chain yet, the client currently has no
way of convincing the validator that the proposal is correct, because
the validator expects the whole sender chain as proof of the sent
messages.

## Proposal

Make validators accept confirmed certificates for blocks on chains with
gaps. If there is a gap, blocks should only be preprocessed, instead of
fully processed - like on the clients.

## Test Plan

CI with an added test exercising the problematic scenario

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

## Links

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

Author: bart-linera

linera-chain/src/block.rs

@@ -345,7 +345,7 @@ pub struct BlockBody {
     /// The list of outgoing messages for each transaction.
     pub messages: Vec<Vec<OutgoingMessage>>,
     /// The hashes of previous blocks that sent messages to the same recipients.
-    pub previous_message_blocks: BTreeMap<ChainId, CryptoHash>,
+    pub previous_message_blocks: BTreeMap<ChainId, (CryptoHash, BlockHeight)>,
     /// The record of oracle responses for each transaction.
     pub oracle_responses: Vec<Vec<OracleResponse>>,
     /// The list of events produced by each transaction.
@@ -586,7 +586,7 @@ impl BcsHashable<'_> for Block {}
 
 #[derive(Serialize, Deserialize)]
 pub struct PreviousMessageBlocksMap<'a> {
-    inner: Cow<'a, BTreeMap<ChainId, CryptoHash>>,
+    inner: Cow<'a, BTreeMap<ChainId, (CryptoHash, BlockHeight)>>,
 }
 
 impl<'de> BcsHashable<'de> for PreviousMessageBlocksMap<'de> {}

linera-chain/src/chain.rs

@@ -788,7 +788,7 @@ where
                     .ok_or_else(|| {
                         ChainError::InternalError("missing entry in confirmed_log".into())
                     })?;
-                previous_message_blocks.insert(recipient, hash);
+                previous_message_blocks.insert(recipient, (hash, height));
             }
         }
 
@@ -969,8 +969,7 @@ where
         Ok(())
     }
 
-    /// Returns the hashes of all blocks in the given range. Returns an error if we are missing
-    /// any of those blocks.
+    /// Returns the hashes of all blocks we have in the given range.
     pub async fn block_hashes(
         &self,
         range: impl RangeBounds<BlockHeight>,
@@ -990,14 +989,9 @@ where
         };
         // Everything after (including) next_height in in preprocessed_blocks if we have it.
         for height in start.max(next_height).0..=end.0 {
-            hashes.push(
-                self.preprocessed_blocks
-                    .get(&BlockHeight(height))
-                    .await?
-                    .ok_or_else(|| {
-                        ChainError::InternalError("missing entry in preprocessed_blocks".into())
-                    })?,
-            );
+            if let Some(hash) = self.preprocessed_blocks.get(&BlockHeight(height)).await? {
+                hashes.push(hash);
+            }
         }
         Ok(hashes)
     }
@@ -1040,7 +1034,8 @@ where
             if block_height > next_height {
                 // There may be a gap in the chain before this block. We can only add it to this
                 // outbox if the previous message to the same recipient has already been added.
-                let prev_hash = match outbox.next_height_to_schedule.get().try_sub_one().ok() {
+                let maybe_prev_hash = match outbox.next_height_to_schedule.get().try_sub_one().ok()
+                {
                     // The block with the last added message has already been executed; look up its
                     // hash in the confirmed_log.
                     Some(height) if height < next_height => {
@@ -1058,8 +1053,38 @@ where
                     None => None, // No message to that sender was added yet.
                 };
                 // Only schedule if this block contains the next message for that recipient.
-                if prev_hash.as_ref() != block.body.previous_message_blocks.get(target) {
-                    continue;
+                match (
+                    maybe_prev_hash,
+                    block.body.previous_message_blocks.get(target),
+                ) {
+                    (None, None) => {
+                        // No previous message block expected and none indicated by the outbox -
+                        // all good
+                    }
+                    (Some(_), None) => {
+                        // Outbox indicates there was a previous message block, but
+                        // previous_message_blocks has no idea about it - possible bug
+                        return Err(ChainError::InternalError(
+                            "block indicates no previous message block,\
+                            but we have one in the outbox"
+                                .into(),
+                        ));
+                    }
+                    (None, Some((_, prev_msg_block_height))) => {
+                        // We have no previously processed block in the outbox, but we are
+                        // expecting one - this could be due to an empty outbox having been pruned.
+                        // Only process the outbox if the height of the previous message block is
+                        // lower than the tip
+                        if *prev_msg_block_height >= next_height {
+                            continue;
+                        }
+                    }
+                    (Some(ref prev_hash), Some((prev_msg_block_hash, _))) => {
+                        // Only process the outbox if the hashes match.
+                        if prev_hash != prev_msg_block_hash {
+                            continue;
+                        }
+                    }
                 }
             }
             if outbox.schedule_message(block_height)? {

linera-chain/src/data_types.rs

@@ -296,7 +296,7 @@ pub struct BlockExecutionOutcome {
     /// The list of outgoing messages for each transaction.
     pub messages: Vec<Vec<OutgoingMessage>>,
     /// The hashes of previous blocks that sent messages to the same recipients.
-    pub previous_message_blocks: BTreeMap<ChainId, CryptoHash>,
+    pub previous_message_blocks: BTreeMap<ChainId, (CryptoHash, BlockHeight)>,
     /// The hash of the chain's execution state after this block.
     pub state_hash: CryptoHash,
     /// The record of oracle responses for each transaction.

linera-core/src/chain_worker/state/attempted_changes.rs

@@ -290,11 +290,6 @@ where
 
         // Check that the chain is active and ready for this confirmation.
         let tip = self.state.chain.tip_state.get().clone();
-        if tip.next_block_height < height {
-            return Err(WorkerError::MissingEarlierBlocks {
-                current_block_height: tip.next_block_height,
-            });
-        }
         if tip.next_block_height > height {
             // We already processed this block.
             let actions = self.state.create_network_actions().await?;
@@ -304,6 +299,7 @@ where
             return Ok((info, actions));
         }
 
+        // We haven't processed the block - verify the certificate first
         if height == BlockHeight::ZERO {
             // For the initial proposal on a chain, we read the committee directly from
             // storage. If the certificate is good for this committee, we will accept it -
@@ -325,6 +321,17 @@ where
             check_block_epoch(epoch, chain_id, block.header.epoch)?;
             certificate.check(committee)?;
         }
+
+        // If this block is higher than the next expected block in this chain, we're going
+        // to have a gap: do not process this block fully, only preprocess it.
+        if tip.next_block_height < height {
+            let actions = self.preprocess_certificate(certificate).await?;
+            self.register_delivery_notifier(height, &actions, notify_when_messages_are_delivered)
+                .await;
+            let info = ChainInfoResponse::new(&self.state.chain, self.state.config.key_pair());
+            return Ok((info, actions));
+        }
+
         // This should always be true for valid certificates.
         ensure!(
             tip.block_hash == block.header.previous_block_hash,

linera-core/src/client/mod.rs

@@ -507,7 +507,7 @@ impl<Env: Environment> Client<Env> {
     }
 
     #[instrument(level = "trace", skip_all)]
-    pub async fn preprocess_certificate(
+    async fn preprocess_certificate(
         &self,
         certificate: Box<ConfirmedBlockCertificate>,
     ) -> Result<(), LocalNodeError> {