Leave gaps between sender blocks. (#4130)

## Motivation

It's unnecessary to fully download all chains that sent a message to our
own chain(s).

## Proposal

Only download the relevant blocks.

## Test Plan

A new client test verifies that we are leaving gaps.

## Release Plan

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

## Links

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

Author: afck

linera-core/src/client/mod.rs

@@ -768,8 +768,8 @@ impl<Env: Environment> Client<Env> {
         Ok(())
     }
 
-    /// Downloads and processes all confirmed block certificates that sent any message to this
-    /// chain, including their ancestors.
+    /// Downloads and preprocesses all confirmed block certificates that sent any message to this
+    /// chain.
     #[instrument(level = "trace", skip(self))]
     async fn synchronize_received_certificates_from_validator(
         &self,
@@ -792,12 +792,12 @@ impl<Env: Environment> Client<Env> {
         let query = ChainInfoQuery::new(chain_id).with_received_log_excluding_first_n(tracker);
         let info = remote_node.handle_chain_info_query(query).await?;
         let remote_log = info.requested_received_log;
-        let remote_max_heights = Self::max_height_per_chain(&remote_log);
+        let remote_heights = Self::heights_per_chain(&remote_log);
 
         // Obtain the next block height we need in the local node, for each chain.
         let local_next_heights = self
             .local_node
-            .next_block_heights(remote_max_heights.keys(), chain_worker_limit)
+            .next_outbox_heights(remote_heights.keys(), chain_worker_limit, chain_id)
             .await?;
 
         // We keep track of the height we've successfully downloaded and checked, per chain.
@@ -806,15 +806,17 @@ impl<Env: Environment> Client<Env> {
         // put all their sent messages into the inbox.
         let mut other_sender_chains = Vec::new();
 
-        let certificate_hashes = future::try_join_all(remote_max_heights.into_iter().filter_map(
-            |(sender_chain_id, remote_height)| {
-                let local_next = *local_next_heights.get(&sender_chain_id)?;
+        let certificate_hashes = future::try_join_all(remote_heights.into_iter().filter_map(
+            |(sender_chain_id, remote_heights)| {
+                let height0 = *remote_heights.first()?;
+                let local_next = (*local_next_heights.get(&sender_chain_id)?).max(height0);
                 if let Ok(height) = local_next.try_sub_one() {
                     downloaded_heights.insert(sender_chain_id, height);
                 }
 
-                let Some(diff) = remote_height.0.checked_sub(local_next.0) else {
-                    // Our highest, locally-known block is higher than any block height
+                let height1 = *remote_heights.last()?;
+                let Some(diff) = height1.0.checked_sub(local_next.0) else {
+                    // Our highest, locally executed block is higher than any block height
                     // from the current batch. Skip this batch, but remember to wait for
                     // the messages to be delivered to the inboxes.
                     other_sender_chains.push(sender_chain_id);
@@ -823,20 +825,61 @@ impl<Env: Environment> Client<Env> {
 
                 // Find the hashes of the blocks we need.
                 let range = BlockHeightRange::multi(local_next, diff.saturating_add(1));
-                Some(remote_node.fetch_sent_certificate_hashes(sender_chain_id, range))
+                Some(async move {
+                    let hashes = remote_node
+                        .fetch_sent_certificate_hashes(sender_chain_id, range)
+                        .await?;
+                    Ok::<_, ChainClientError>(
+                        remote_heights
+                            .into_iter()
+                            .map(move |h| hashes[(h.0 - local_next.0) as usize]),
+                    )
+                })
             },
         ))
         .await?
         .into_iter()
         .flatten()
-        .collect();
+        .collect::<Vec<_>>();
+
+        let local_certificates =
+            future::try_join_all(certificate_hashes.iter().map(|hash| async move {
+                match self.storage_client().read_certificate(*hash).await {
+                    Ok(certificate) => Ok(Some(certificate)),
+                    Err(ViewError::NotFound(_)) => Ok(None),
+                    Err(error) => Err(error),
+                }
+            }))
+            .await?
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+        let local_certificate_hashes = local_certificates
+            .iter()
+            .map(|cert| cert.hash())
+            .collect::<HashSet<_>>();
 
         // Download the block certificates.
         let remote_certificates = remote_node
-            .download_certificates(certificate_hashes)
+            .download_certificates(
+                certificate_hashes
+                    .into_iter()
+                    .filter(|hash| !local_certificate_hashes.contains(hash))
+                    .collect(),
+            )
             .await?;
         let mut certificates_by_height_by_chain = BTreeMap::new();
 
+        for confirmed_block_certificate in local_certificates {
+            let block_header = &confirmed_block_certificate.inner().block().header;
+            let sender_chain_id = block_header.chain_id;
+            let height = block_header.height;
+            certificates_by_height_by_chain
+                .entry(sender_chain_id)
+                .or_insert_with(BTreeMap::new)
+                .insert(height, confirmed_block_certificate);
+        }
+
         // Check the signatures and keep only the ones that are valid.
         for confirmed_block_certificate in remote_certificates {
             let block_header = &confirmed_block_certificate.inner().block().header;
@@ -863,7 +906,7 @@ impl<Env: Environment> Client<Env> {
                     certificates_by_height_by_chain
                         .entry(sender_chain_id)
                         .or_insert_with(BTreeMap::new)
-                        .insert(height, confirmed_block_certificate.clone());
+                        .insert(height, confirmed_block_certificate);
                 }
             }
         }
@@ -881,10 +924,9 @@ impl<Env: Environment> Client<Env> {
         }
 
         for (sender_chain_id, certs) in &mut certificates_by_height_by_chain {
-            if !certs
+            if certs
                 .values()
-                .last()
-                .is_some_and(|cert| cert.block().recipients().contains(&chain_id))
+                .any(|cert| !cert.block().recipients().contains(&chain_id))
             {
                 warn!(
                     "Skipping received certificates from chain {sender_chain_id:.8}:
@@ -931,15 +973,17 @@ impl<Env: Environment> Client<Env> {
     }
 
     /// Given a set of chain ID-block height pairs, returns a map that assigns to each chain ID
-    /// the highest height seen.
-    fn max_height_per_chain(remote_log: &[ChainAndHeight]) -> BTreeMap<ChainId, BlockHeight> {
+    /// the set of heights. The returned map contains no empty values.
+    fn heights_per_chain(
+        remote_log: &[ChainAndHeight],
+    ) -> BTreeMap<ChainId, BTreeSet<BlockHeight>> {
         remote_log.iter().fold(
-            BTreeMap::<ChainId, BlockHeight>::new(),
+            BTreeMap::<ChainId, BTreeSet<_>>::new(),
             |mut chain_to_info, entry| {
                 chain_to_info
                     .entry(entry.chain_id)
-                    .and_modify(|h| *h = entry.height.max(*h))
-                    .or_insert(entry.height);
+                    .or_default()
+                    .insert(entry.height);
                 chain_to_info
             },
         )

linera-core/src/local_node.rs

@@ -311,23 +311,23 @@ where
     }
 
     /// Given a list of chain IDs, returns a map that assigns to each of them the next block
-    /// height, i.e. the lowest block height that we have not processed in the local node yet.
+    /// height to schedule, i.e. the lowest block height for which we haven't added the messages
+    /// to `receiver_id` to the outbox yet.
     ///
     /// It makes at most `chain_worker_limit` requests to the local node in parallel.
-    pub async fn next_block_heights(
+    pub async fn next_outbox_heights(
         &self,
         chain_ids: impl IntoIterator<Item = &ChainId>,
         chain_worker_limit: usize,
+        receiver_id: ChainId,
     ) -> Result<BTreeMap<ChainId, BlockHeight>, LocalNodeError> {
         let futures = chain_ids
             .into_iter()
             .map(|chain_id| async move {
                 let chain = self.chain_state_view(*chain_id).await?;
                 let mut next_height = chain.tip_state.get().next_block_height;
-                // TODO(#3969): This is not great for performance, but the whole function will
-                // probably go away with #3969 anyway.
-                while chain.preprocessed_blocks.contains_key(&next_height).await? {
-                    next_height.try_add_assign_one()?;
+                if let Some(outbox) = chain.outboxes.try_load_entry(&receiver_id).await? {
+                    next_height = next_height.max(*outbox.next_height_to_schedule.get());
                 }
                 Ok::<_, LocalNodeError>((*chain_id, next_height))
             })

linera-core/src/unit_tests/client_tests.rs

@@ -28,6 +28,7 @@ use linera_execution::{
     ExecutionError, Message, MessageKind, Operation, QueryOutcome, ResourceControlPolicy,
     SystemMessage, SystemQuery, SystemResponse,
 };
+use linera_storage::Storage;
 use rand::Rng;
 use test_case::test_case;
 use test_helpers::{
@@ -1289,6 +1290,70 @@ where
     Ok(())
 }
 
+#[test_case(MemoryStorageBuilder::default(); "memory")]
+#[cfg_attr(feature = "storage-service", test_case(ServiceStorageBuilder::new().await; "storage_service"))]
+#[test_log::test(tokio::test)]
+async fn test_sparse_sender_chain<B>(storage_builder: B) -> anyhow::Result<()>
+where
+    B: StorageBuilder,
+{
+    let signer = InMemorySigner::new(None);
+    let mut builder = TestBuilder::new(storage_builder, 4, 1, signer).await?;
+    let sender = builder.add_root_chain(1, Amount::from_tokens(4)).await?;
+    let receiver = builder.add_root_chain(2, Amount::ZERO).await?;
+    let receiver_id = receiver.chain_id();
+
+    let cert0 = sender
+        .transfer_to_account(
+            AccountOwner::CHAIN,
+            Amount::ONE,
+            Account::chain(receiver_id),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+    let cert1 = sender
+        .burn(AccountOwner::CHAIN, Amount::ONE)
+        .await
+        .unwrap()
+        .unwrap();
+    let cert2 = sender
+        .transfer_to_account(
+            AccountOwner::CHAIN,
+            Amount::ONE,
+            Account::chain(receiver_id),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+    receiver.synchronize_from_validators().await?;
+    receiver.process_inbox().await?;
+
+    // The first and last blocks sent something to the receiver. The middle one didn't.
+    // So the sender chain should have a gap.
+    assert!(
+        receiver
+            .storage_client()
+            .contains_certificate(cert0.hash())
+            .await?
+    );
+    assert!(
+        !receiver
+            .storage_client()
+            .contains_certificate(cert1.hash())
+            .await?
+    );
+    assert!(
+        receiver
+            .storage_client()
+            .contains_certificate(cert2.hash())
+            .await?
+    );
+
+    Ok(())
+}
+
 #[test_case(MemoryStorageBuilder::default(); "memory")]
 #[cfg_attr(feature = "storage-service", test_case(ServiceStorageBuilder::new().await; "storage_service"))]
 #[cfg_attr(feature = "rocksdb", test_case(RocksDbStorageBuilder::new().await; "rocks_db"))]