Don't execute sender chains (#3989)

## Motivation

When the client verifies incoming messages to its own chains, it doesn't
need the sender chain's execution state, so computing the block
execution there is unnecessary work.

Eventually we want to go even further and avoid even downloading some of
the sender chain's blocks (see
#3969).

## Proposal

As a preparation for #3969, we introduce an `unexecuted_blocks` map to
the chain state: It maps block heights that are higher than
`next_block_height`, i.e. haven't been executed yet, to block hashes.

We don't purge empty outboxes anymore. When updating the outboxes, we
make sure that we haven't skipped any messages and compare to the next
height to schedule: That allows us to update some outboxes even for
loose blocks, ahead of execution.

For now we still download all blocks of the sender chain. Skipping any
will be done in a future PR.
However, we now add them as loose blocks and don't execute them.

## Test Plan

CI

## Release Plan

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

## Links

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

Author: afck

linera-base/src/data_types.rs

@@ -13,6 +13,7 @@ use std::{
     hash::Hash,
     io, iter,
     num::ParseIntError,
+    ops::{Bound, RangeBounds},
     path::Path,
     str::FromStr,
 };
@@ -486,6 +487,32 @@ impl TryFrom<BlockHeight> for usize {
     }
 }
 
+/// Allows converting [`BlockHeight`] ranges to inclusive tuples of bounds.
+pub trait BlockHeightRangeBounds {
+    /// Returns the range as a tuple of inclusive bounds.
+    /// If the range is empty, returns `None`.
+    fn to_inclusive(&self) -> Option<(BlockHeight, BlockHeight)>;
+}
+
+impl<T: RangeBounds<BlockHeight>> BlockHeightRangeBounds for T {
+    fn to_inclusive(&self) -> Option<(BlockHeight, BlockHeight)> {
+        let start = match self.start_bound() {
+            Bound::Included(height) => *height,
+            Bound::Excluded(height) => height.try_add_one().ok()?,
+            Bound::Unbounded => BlockHeight(0),
+        };
+        let end = match self.end_bound() {
+            Bound::Included(height) => *height,
+            Bound::Excluded(height) => height.try_sub_one().ok()?,
+            Bound::Unbounded => BlockHeight::MAX,
+        };
+        if start > end {
+            return None;
+        }
+        Some((start, end))
+    }
+}
+
 impl_wrapped_number!(Amount, u128);
 impl_wrapped_number!(BlockHeight, u64);
 impl_wrapped_number!(TimeDelta, u64);

linera-chain/src/chain.rs

@@ -3,6 +3,7 @@
 
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap, HashSet},
+    ops::RangeBounds,
     sync::Arc,
 };
 
@@ -11,7 +12,7 @@ use linera_base::{
     crypto::{CryptoHash, ValidatorPublicKey},
     data_types::{
         Amount, ApplicationDescription, ApplicationPermissions, ArithmeticError, Blob, BlockHeight,
-        Epoch, OracleResponse, Timestamp,
+        BlockHeightRangeBounds as _, Epoch, OracleResponse, Timestamp,
     },
     ensure,
     identifiers::{AccountOwner, ApplicationId, BlobType, ChainId, MessageId},
@@ -277,6 +278,9 @@ where
     /// Number of outgoing messages in flight for each block height.
     /// We use a `RegisterView` to prioritize speed for small maps.
     pub outbox_counters: RegisterView<C, BTreeMap<BlockHeight, u32>>,
+
+    /// Blocks that have been verified but not executed yet, and that may not be contiguous.
+    pub preprocessed_blocks: MapView<C, BlockHeight, CryptoHash>,
 }
 
 /// Block-chaining state.
@@ -419,9 +423,6 @@ where
                 self.outbox_counters.get_mut().remove(&update);
             }
         }
-        if outbox.queue.count() == 0 {
-            self.outboxes.remove_entry(target)?;
-        }
         #[cfg(with_metrics)]
         metrics::NUM_OUTBOXES
             .with_label_values(&[])
@@ -942,6 +943,20 @@ where
             &block.body.messages,
         )?;
         self.confirmed_log.push(hash);
+        self.preprocessed_blocks.remove(&block.header.height)?;
+        Ok(())
+    }
+
+    /// Adds a block to `preprocessed_blocks`, and updates the outboxes where possible.
+    pub async fn preprocess_block(&mut self, block: &ConfirmedBlock) -> Result<(), ChainError> {
+        let hash = block.inner().hash();
+        let block = block.inner().inner();
+        let height = block.header.height;
+        if height < self.tip_state.get().next_block_height {
+            return Ok(());
+        }
+        self.process_outgoing_messages(block).await?;
+        self.preprocessed_blocks.insert(&height, hash)?;
         Ok(())
     }
 
@@ -1067,6 +1082,37 @@ where
         Ok(())
     }
 
+    /// Returns the hashes of all blocks in the given range. Returns an error if we are missing
+    /// any of those blocks.
+    pub async fn block_hashes(
+        &self,
+        range: impl RangeBounds<BlockHeight>,
+    ) -> Result<Vec<CryptoHash>, ChainError> {
+        let next_height = self.tip_state.get().next_block_height;
+        // If the range is not empty, it can always be represented as start..=end.
+        let Some((start, end)) = range.to_inclusive() else {
+            return Ok(Vec::new());
+        };
+        let mut hashes = if let Ok(last_height) = next_height.try_sub_one() {
+            let usize_start = usize::try_from(start)?;
+            let usize_end = usize::try_from(end.min(last_height))?;
+            self.confirmed_log.read(usize_start..=usize_end).await?
+        } else {
+            Vec::new()
+        };
+        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())
+                    })?,
+            );
+        }
+        Ok(hashes)
+    }
+
     /// Resets the chain manager for the next block height.
     fn reset_chain_manager(
         &mut self,
@@ -1094,12 +1140,33 @@ where
         // application.
         let recipients = block.recipients();
         let block_height = block.header.height;
+        let next_height = self.tip_state.get().next_block_height;
 
         // Update the outboxes.
         let outbox_counters = self.outbox_counters.get_mut();
         let targets = recipients.into_iter().collect::<Vec<_>>();
         let outboxes = self.outboxes.try_load_entries_mut(&targets).await?;
-        for mut outbox in outboxes {
+        for (mut outbox, target) in outboxes.into_iter().zip(&targets) {
+            if block_height > next_height {
+                // Find the hash of the block that was most recently added to the outbox.
+                let prev_hash = match outbox.next_height_to_schedule.get().try_sub_one().ok() {
+                    Some(height) if height < next_height => {
+                        let index =
+                            usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
+                        Some(self.confirmed_log.get(index).await?.ok_or_else(|| {
+                            ChainError::InternalError("missing entry in confirmed_log".into())
+                        })?)
+                    }
+                    Some(height) => Some(self.preprocessed_blocks.get(&height).await?.ok_or_else(
+                        || ChainError::InternalError("missing entry in preprocessed_blocks".into()),
+                    )?),
+                    None => None,
+                };
+                // Schedule only if no block is missing that sent something to the same recipient.
+                if prev_hash.as_ref() != block.body.previous_message_blocks.get(target) {
+                    continue;
+                }
+            }
             if outbox.schedule_message(block_height)? {
                 *outbox_counters.entry(block_height).or_default() += 1;
             }

linera-core/src/chain_worker/actor.rs

@@ -121,6 +121,13 @@ where
         callback: oneshot::Sender<Result<(ChainInfoResponse, NetworkActions), WorkerError>>,
     },
 
+    /// Preprocess a block without executing it.
+    PreprocessCertificate {
+        certificate: ConfirmedBlockCertificate,
+        #[debug(skip)]
+        callback: oneshot::Sender<Result<NetworkActions, WorkerError>>,
+    },
+
     /// Process a cross-chain update.
     ProcessCrossChainUpdate {
         origin: ChainId,
@@ -392,6 +399,12 @@ where
                         .await,
                 )
                 .is_ok(),
+            ChainWorkerRequest::PreprocessCertificate {
+                certificate,
+                callback,
+            } => callback
+                .send(self.worker.preprocess_certificate(certificate).await)
+                .is_ok(),
             ChainWorkerRequest::ProcessCrossChainUpdate {
                 origin,
                 bundles,
@@ -480,6 +493,9 @@ where
             ChainWorkerRequest::ProcessConfirmedBlock { callback, .. } => {
                 callback.send(Err(error)).is_ok()
             }
+            ChainWorkerRequest::PreprocessCertificate { callback, .. } => {
+                callback.send(Err(error)).is_ok()
+            }
             ChainWorkerRequest::ProcessCrossChainUpdate { callback, .. } => {
                 callback.send(Err(error)).is_ok()
             }

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

@@ -414,6 +414,53 @@ where
         Ok((info, actions))
     }
 
+    /// Stores a block's blobs, and adds its messages to the outbox where possible.
+    /// Does not execute the block.
+    pub(super) async fn preprocess_certificate(
+        &mut self,
+        certificate: ConfirmedBlockCertificate,
+    ) -> Result<NetworkActions, WorkerError> {
+        let block = certificate.block();
+        // 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 > block.header.height {
+            // We already processed this block.
+            return self.state.create_network_actions().await;
+        }
+
+        let required_blob_ids = block.required_blob_ids();
+        let created_blobs: BTreeMap<_, _> = block.iter_created_blobs().collect();
+        let blobs_result = self
+            .state
+            .get_required_blobs(block.required_blob_ids(), &created_blobs)
+            .await
+            .map(|blobs| blobs.into_values().collect::<Vec<_>>());
+
+        if let Ok(blobs) = &blobs_result {
+            self.state
+                .storage
+                .write_blobs_and_certificate(blobs, &certificate)
+                .await?;
+        }
+
+        // Update the blob state with last used certificate hash.
+        let blob_state = certificate.value().to_blob_state(blobs_result.is_ok());
+        let blob_ids = required_blob_ids.into_iter().collect::<Vec<_>>();
+        self.state
+            .storage
+            .maybe_write_blob_states(&blob_ids, blob_state)
+            .await?;
+        blobs_result?;
+        // Update the outboxes.
+        self.state
+            .chain
+            .preprocess_block(certificate.value())
+            .await?;
+        // Persist chain.
+        self.save().await?;
+        self.state.create_network_actions().await
+    }
+
     /// Schedules a notification for when cross-chain messages are delivered up to the given
     /// `height`.
     #[instrument(level = "trace", skip(self, notify_when_messages_are_delivered))]

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

@@ -27,7 +27,7 @@ use linera_chain::{
 };
 use linera_execution::{ExecutionStateView, Query, QueryOutcome, ServiceRuntimeEndpoint};
 use linera_storage::{Clock as _, Storage};
-use linera_views::{views::ClonableView, ViewError};
+use linera_views::views::ClonableView;
 use tokio::sync::{oneshot, OwnedRwLockReadGuard, RwLock};
 
 #[cfg(test)]
@@ -261,6 +261,18 @@ where
             .await
     }
 
+    /// Preprocesses a block without executing it.
+    #[tracing::instrument(level = "debug", skip(self))]
+    pub(super) async fn preprocess_certificate(
+        &mut self,
+        certificate: ConfirmedBlockCertificate,
+    ) -> Result<NetworkActions, WorkerError> {
+        ChainWorkerStateWithAttemptedChanges::new(self)
+            .await
+            .preprocess_certificate(certificate)
+            .await
+    }
+
     /// Updates the chain's inboxes, receiving messages from a cross-chain update.
     #[tracing::instrument(level = "debug", skip(self))]
     pub(super) async fn process_cross_chain_update(
@@ -458,22 +470,38 @@ where
     ) -> Result<NetworkActions, WorkerError> {
         // Load all the certificates we will need, regardless of the medium.
         let heights = BTreeSet::from_iter(heights_by_recipient.values().flatten().copied());
-        let heights_usize = heights
-            .iter()
+        let next_block_height = self.chain.tip_state.get().next_block_height;
+        let log_heights = heights
+            .range(..next_block_height)
             .copied()
             .map(usize::try_from)
             .collect::<Result<Vec<_>, _>>()?;
-        let hashes = self
+        let mut hashes = self
             .chain
             .confirmed_log
-            .multi_get(heights_usize.clone())
+            .multi_get(log_heights)
             .await?
             .into_iter()
-            .zip(heights_usize)
+            .zip(&heights)
             .map(|(maybe_hash, height)| {
-                maybe_hash.ok_or_else(|| ViewError::not_found("confirmed log entry", height))
+                maybe_hash.ok_or_else(|| WorkerError::ConfirmedLogEntryNotFound {
+                    height: *height,
+                    chain_id: self.chain_id(),
+                })
             })
             .collect::<Result<Vec<_>, _>>()?;
+        for height in heights.range(next_block_height..) {
+            hashes.push(
+                self.chain
+                    .preprocessed_blocks
+                    .get(height)
+                    .await?
+                    .ok_or_else(|| WorkerError::PreprocessedBlocksEntryNotFound {
+                        height: *height,
+                        chain_id: self.chain_id(),
+                    })?,
+            );
+        }
         let certificates = self.storage.read_certificates(hashes).await?;
         let certificates = heights
             .into_iter()