[testnet] Simplify and generalize block_hashes. (#4897) (#4901)

Backport of #4897.

## Motivation

After introducing sparse chains, we use `block_hashes` not just for
ranges, but for sparse lists of blocks, too.

## Proposal

Use `LogView::multi_get` (list of heights) instead of `read` (range).
Make `block_hashes` accept a list of heights instead of a range.

## Test Plan

CI should catch regressions; no new functionality was added.

## Release Plan

- These changes should:
    - be released in the next SDK version, (low priority!)
    - be released in the next validator hotfix. (low priority!)

## Links

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

Author: afck

linera-base/src/data_types.rs

@@ -12,7 +12,6 @@ use std::{
     hash::Hash,
     io, iter,
     num::ParseIntError,
-    ops::{Bound, RangeBounds},
     path::Path,
     str::FromStr,
     sync::Arc,
@@ -498,32 +497,6 @@ 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,15 +3,14 @@
 
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap, HashSet},
-    ops::RangeBounds,
     sync::Arc,
 };
 
 use linera_base::{
     crypto::{CryptoHash, ValidatorPublicKey},
     data_types::{
-        ApplicationDescription, ApplicationPermissions, ArithmeticError, Blob, BlockHeight,
-        BlockHeightRangeBounds as _, Epoch, OracleResponse, Timestamp,
+        ApplicationDescription, ApplicationPermissions, ArithmeticError, Blob, BlockHeight, Epoch,
+        OracleResponse, Timestamp,
     },
     ensure,
     identifiers::{AccountOwner, ApplicationId, BlobType, ChainId, StreamId},
@@ -1056,43 +1055,37 @@ where
     }
 
     /// Returns the hashes of all blocks we have in the given range.
+    ///
+    /// If the input heights are in ascending order, the hashes will be in the same order.
+    /// Otherwise they may be unordered.
     #[instrument(skip_all, fields(
         chain_id = %self.chain_id(),
         next_block_height = %self.tip_state.get().next_block_height,
-        start_height = ?range.start_bound(),
-        end_height = ?range.end_bound()
     ))]
     pub async fn block_hashes(
         &self,
-        range: impl RangeBounds<BlockHeight>,
+        heights: impl IntoIterator<Item = 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());
-        };
         // Everything up to (excluding) next_height is in confirmed_log.
-        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()
-        };
+        let (confirmed_heights, unconfirmed_heights) = heights
+            .into_iter()
+            .partition::<Vec<_>, _>(|height| *height < next_height);
+        let confirmed_indices = confirmed_heights
+            .into_iter()
+            .map(|height| usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow))
+            .collect::<Result<_, _>>()?;
+        let confirmed_hashes = self.confirmed_log.multi_get(confirmed_indices).await?;
         // Everything after (including) next_height in preprocessed_blocks if we have it.
-        let block_heights = (start.max(next_height).0..=end.0)
-            .map(BlockHeight)
-            .collect::<Vec<_>>();
-        for hash in self
+        let unconfirmed_hashes = self
             .preprocessed_blocks
-            .multi_get(&block_heights)
-            .await?
+            .multi_get(&unconfirmed_heights)
+            .await?;
+        Ok(confirmed_hashes
             .into_iter()
+            .chain(unconfirmed_hashes)
             .flatten()
-        {
-            hashes.push(hash);
-        }
-        Ok(hashes)
+            .collect())
     }
 
     /// Resets the chain manager for the next block height.

linera-core/src/chain_worker/state.rs

@@ -7,7 +7,6 @@ use std::{
     borrow::Cow,
     collections::{BTreeMap, BTreeSet, HashMap, HashSet},
     iter,
-    ops::RangeInclusive,
     sync::{self, Arc},
 };
 
@@ -1561,11 +1560,9 @@ where
             bundles.sort_by_key(|b| b.bundle.timestamp);
             info.requested_pending_message_bundles = bundles;
         }
-        let mut hashes = Vec::new();
-        let height_ranges = into_ranges(query.request_sent_certificate_hashes_by_heights);
-        for height_range in height_ranges {
-            hashes.extend(chain.block_hashes(height_range).await?);
-        }
+        let hashes = chain
+            .block_hashes(query.request_sent_certificate_hashes_by_heights)
+            .await?;
         info.requested_sent_certificate_hashes = hashes;
         if let Some(start) = query.request_received_log_excluding_first_n {
             let start = usize::try_from(start).map_err(|_| ArithmeticError::Overflow)?;
@@ -1644,21 +1641,6 @@ where
     }
 }
 
-/// Returns an iterator of inclusive ranges that exactly cover the list of block heights.
-fn into_ranges(
-    values: impl IntoIterator<Item = BlockHeight>,
-) -> impl Iterator<Item = RangeInclusive<BlockHeight>> {
-    let mut values_iter = values.into_iter().peekable();
-    iter::from_fn(move || {
-        let start = values_iter.next()?;
-        let mut end = start;
-        while values_iter.peek() == end.try_add_one().ok().as_ref() {
-            end = values_iter.next()?;
-        }
-        Some(start..=end)
-    })
-}
-
 /// Returns the keys whose value is `None`.
 fn missing_blob_ids(maybe_blobs: &BTreeMap<BlobId, Option<Blob>>) -> Vec<BlobId> {
     maybe_blobs
@@ -1771,21 +1753,3 @@ impl<'a> CrossChainUpdateHelper<'a> {
         Ok(bundles)
     }
 }
-
-#[test]
-fn test_into_ranges() {
-    assert_eq!(
-        into_ranges(vec![
-            BlockHeight(2),
-            BlockHeight(3),
-            BlockHeight(4),
-            BlockHeight(6)
-        ],)
-        .collect::<Vec<_>>(),
-        vec![
-            BlockHeight(2)..=BlockHeight(4),
-            BlockHeight(6)..=BlockHeight(6)
-        ]
-    );
-    assert_eq!(into_ranges(vec![]).collect::<Vec<_>>(), vec![]);
-}

linera-core/src/updater.rs

@@ -522,7 +522,7 @@ where
                 .local_node
                 .chain_state_view(chain_id)
                 .await?
-                .block_hashes(height..=height)
+                .block_hashes([height])
                 .await?
                 .into_iter()
                 .next()
@@ -556,12 +556,12 @@ where
                 Ok(info) => info,
             };
             // Obtain the missing blocks and the manager state from the local node.
-            let range = info.next_block_height..target_block_height;
+            let heights = (info.next_block_height.0..target_block_height.0).map(BlockHeight);
             let validator_missing_hashes = self
                 .local_node
                 .chain_state_view(chain_id)
                 .await?
-                .block_hashes(range)
+                .block_hashes(heights)
                 .await?;
             if !validator_missing_hashes.is_empty() {
                 // Send the requested certificates in order.

linera-service/src/cli/main.rs

@@ -1830,7 +1830,7 @@ impl Runnable for Job {
                     .await
                     .context("Failed to load chain")?;
                 let block_hash = chain_state_view
-                    .block_hashes(height..=height)
+                    .block_hashes([height])
                     .await
                     .context("Failed to find a block hash for the given height")?[0];
                 let block = context