Merge confirmed_log and preprocessed_blocks into block_hashes

linera-chain/src/chain.rs

@@ -26,7 +26,7 @@ use linera_execution::{
 use linera_views::{
     context::Context,
     log_view::LogView,
-    map_view::MapView,
+    map_view::{CustomMapView, MapView},
     reentrant_collection_view::{ReadGuardedView, ReentrantCollectionView},
     register_view::RegisterView,
     views::{ClonableView, RootView, View},
@@ -212,9 +212,10 @@ where
     /// The incomplete sets of blobs for upcoming proposals.
     pub pending_proposed_blobs: ReentrantCollectionView<C, AccountOwner, PendingBlobsView<C>>,
 
-    /// Hashes of all certified blocks for this sender.
-    /// This ends with `block_hash` and has length `usize::from(next_block_height)`.
-    pub confirmed_log: LogView<C, CryptoHash>,
+    /// Hashes of all known blocks in this chain, indexed by their height. A block at
+    /// `height < next_block_height` is executed; a block at `height >= next_block_height`
+    /// is preprocessed (verified but not yet executed) and may not be contiguous.
+    pub block_hashes: CustomMapView<C, BlockHeight, CryptoHash>,
     /// Sender chain and height of all certified blocks known as a receiver (local ordering).
     pub received_log: LogView<C, ChainAndHeight>,
     /// The number of `received_log` entries we have synchronized, for each validator.
@@ -233,15 +234,23 @@ where
     /// Outboxes with at least one pending message. This allows us to avoid loading all outboxes.
     pub nonempty_outboxes: RegisterView<C, BTreeSet<ChainId>>,
 
-    /// Blocks that have been verified but not executed yet, and that may not be contiguous.
-    pub preprocessed_blocks: MapView<C, BlockHeight, CryptoHash>,
     /// Inboxes with at least one pending added bundle. This allows us to avoid loading all inboxes.
     pub nonempty_inboxes: RegisterView<C, BTreeSet<ChainId>>,
 
     /// The local wall-clock time when block 0 was last executed. Used to prevent
     /// reset-on-incorrect-outcome from looping: if not enough time has elapsed since
     /// the last reset, the error is returned instead.
     pub block_zero_executed_at: RegisterView<C, Timestamp>,
+
+    /// The height at which the next block can be preprocessed: one past the highest
+    /// height in `block_hashes` (executed or preprocessed), or `next_block_height` if
+    /// `block_hashes` is empty.
+    ///
+    /// Maintained as an O(1) shortcut for `next_height_to_preprocess`, since
+    /// `CustomMapView` does not yet expose a `last_index` lookup. Once
+    /// `linera-views` gains efficient first/last key support, this field can be
+    /// removed in favor of `block_hashes.last_index()`.
+    pub next_height_to_preprocess: RegisterView<C, BlockHeight>,
 }
 
 /// Block-chaining state.
@@ -478,14 +487,20 @@ where
         Ok(())
     }
 
-    /// Returns the height of the highest block we have, plus one. Includes preprocessed blocks.
-    ///
-    /// The "+ 1" is so that it can be used in the same places as `next_block_height`.
-    pub async fn next_height_to_preprocess(&self) -> Result<BlockHeight, ChainError> {
-        if let Some(height) = self.preprocessed_blocks.indices().await?.last() {
-            return Ok(height.saturating_add(BlockHeight(1)));
+    /// Inserts `(height, hash)` into `block_hashes` and updates the
+    /// `next_height_to_preprocess` register accordingly. Every write to
+    /// `block_hashes` must go through this helper so the register stays in sync.
+    fn insert_block_hash(
+        &mut self,
+        height: BlockHeight,
+        hash: CryptoHash,
+    ) -> Result<(), ChainError> {
+        self.block_hashes.insert(&height, hash)?;
+        let next = self.next_height_to_preprocess.get_mut();
+        if *next <= height {
+            *next = height.try_add_one()?;
         }
-        Ok(self.tip_state.get().next_block_height)
+        Ok(())
     }
 
     /// Attempts to process a new `bundle` of messages from the given `origin`. Returns an
@@ -674,7 +689,7 @@ where
     ))]
     async fn execute_block_inner(
         chain: &mut ExecutionStateView<C>,
-        confirmed_log: &LogView<C, CryptoHash>,
+        block_hashes: &CustomMapView<C, BlockHeight, CryptoHash>,
         block: &mut ProposedBlock,
         local_time: Timestamp,
         round: Option<u32>,
@@ -875,7 +890,6 @@ where
 
         let recipients = block_execution_tracker.recipients();
         let mut recipient_heights = Vec::new();
-        let mut indices = Vec::new();
         for (recipient, height) in chain
             .previous_message_blocks
             .multi_get_pairs(recipients)
@@ -885,36 +899,35 @@ where
                 .previous_message_blocks
                 .insert(&recipient, block.height)?;
             if let Some(height) = height {
-                let index = usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
-                indices.push(index);
                 recipient_heights.push((recipient, height));
             }
         }
-        let hashes = confirmed_log.multi_get(indices).await?;
+        let hashes = block_hashes
+            .multi_get(recipient_heights.iter().map(|(_, height)| height))
+            .await?;
         let mut previous_message_blocks = BTreeMap::new();
         for (hash, (recipient, height)) in hashes.into_iter().zip(recipient_heights) {
             let hash = hash.ok_or_else(|| {
-                ChainError::CorruptedChainState("missing entry in confirmed_log".into())
+                ChainError::CorruptedChainState("missing entry in block_hashes".into())
             })?;
             previous_message_blocks.insert(recipient, (hash, height));
         }
 
         let streams = block_execution_tracker.event_streams();
         let mut stream_heights = Vec::new();
-        let mut indices = Vec::new();
         for (stream, height) in chain.previous_event_blocks.multi_get_pairs(streams).await? {
             chain.previous_event_blocks.insert(&stream, block.height)?;
             if let Some(height) = height {
-                let index = usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
-                indices.push(index);
                 stream_heights.push((stream, height));
             }
         }
-        let hashes = confirmed_log.multi_get(indices).await?;
+        let hashes = block_hashes
+            .multi_get(stream_heights.iter().map(|(_, height)| height))
+            .await?;
         let mut previous_event_blocks = BTreeMap::new();
         for (hash, (stream, height)) in hashes.into_iter().zip(stream_heights) {
             let hash = hash.ok_or_else(|| {
-                ChainError::CorruptedChainState("missing entry in confirmed_log".into())
+                ChainError::CorruptedChainState("missing entry in block_hashes".into())
             })?;
             previous_event_blocks.insert(stream, (hash, height));
         }
@@ -1035,7 +1048,7 @@ where
 
         Self::execute_block_inner(
             &mut self.execution_state,
-            &self.confirmed_log,
+            &self.block_hashes,
             &mut block,
             local_time,
             round,
@@ -1079,12 +1092,11 @@ where
         tip.block_hash = Some(hash);
         tip.next_block_height.try_add_assign_one()?;
         tip.update_counters(&block.body.transactions, &block.body.messages)?;
-        self.confirmed_log.push(hash);
-        self.preprocessed_blocks.remove(&block.header.height)?;
+        self.insert_block_hash(block.header.height, hash)?;
         Ok(updated_streams)
     }
 
-    /// Adds a block to `preprocessed_blocks`, and updates the outboxes where possible.
+    /// Adds a block to `block_hashes` as preprocessed, and updates the outboxes where possible.
     /// Returns the set of streams that were updated as a result of preprocessing the block.
     #[instrument(skip_all, fields(
         chain_id = %self.chain_id(),
@@ -1102,7 +1114,7 @@ where
         }
         self.process_outgoing_messages(block).await?;
         let updated_streams = self.process_emitted_events(block).await?;
-        self.preprocessed_blocks.insert(&height, hash)?;
+        self.insert_block_hash(height, hash)?;
         Ok(updated_streams)
     }
 
@@ -1157,36 +1169,22 @@ where
         Ok(())
     }
 
-    /// 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.
+    /// Returns the hashes of all blocks we have at the given heights, in input order.
+    /// Unknown heights are skipped.
     #[instrument(skip_all, fields(
         chain_id = %self.chain_id(),
         next_block_height = %self.tip_state.get().next_block_height,
     ))]
-    pub async fn block_hashes(
+    pub async fn block_hashes_for_heights(
         &self,
         heights: impl IntoIterator<Item = BlockHeight>,
     ) -> Result<Vec<CryptoHash>, ChainError> {
-        let next_height = self.tip_state.get().next_block_height;
-        // Everything up to (excluding) next_height is in confirmed_log.
-        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 unconfirmed_hashes = self
-            .preprocessed_blocks
-            .multi_get(&unconfirmed_heights)
-            .await?;
-        Ok(confirmed_hashes
+        let heights = heights.into_iter().collect::<Vec<_>>();
+        Ok(self
+            .block_hashes
+            .multi_get(&heights)
+            .await?
             .into_iter()
-            .chain(unconfirmed_hashes)
             .flatten()
             .collect())
     }
@@ -1244,24 +1242,11 @@ where
                 }
                 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 => {
-                        let index =
-                            usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
-                        Some(self.confirmed_log.get(index).await?.ok_or_else(|| {
-                            ChainError::CorruptedChainState("missing entry in confirmed_log".into())
+                    Some(height) => {
+                        Some(self.block_hashes.get(&height).await?.ok_or_else(|| {
+                            ChainError::CorruptedChainState("missing entry in block_hashes".into())
                         })?)
                     }
-                    // The block with last added message has not been executed yet. If we have it,
-                    // it's in preprocessed_blocks.
-                    Some(height) => Some(self.preprocessed_blocks.get(&height).await?.ok_or_else(
-                        || {
-                            ChainError::CorruptedChainState(
-                                "missing entry in preprocessed_blocks".into(),
-                            )
-                        },
-                    )?),
                     None => None, // No message to that sender was added yet.
                 };
                 // Only schedule if this block contains the next message for that recipient.

linera-chain/src/unit_tests/chain_tests.rs

@@ -884,3 +884,20 @@ async fn prepare_test_with_dummy_mock_application(
 
     Ok((application, application_id, chain, block, time))
 }
+
+/// The `next_height_to_preprocess` register is read directly by callers rather
+/// than being computed from `block_hashes.indices()`. This pins down that the
+/// register is consulted (no fallback to `indices()` that would mis-sort heights
+/// spanning a byte boundary under BCS little-endian `u64` encoding).
+#[tokio::test]
+async fn test_next_height_to_preprocess_register() {
+    let chain_id = TestEnvironment::new().admin_chain_id();
+    let mut chain = ChainStateView::new(chain_id).await;
+
+    // Empty chain.
+    assert_eq!(*chain.next_height_to_preprocess.get(), BlockHeight(0));
+
+    // Set the register to a height crossing the first byte boundary.
+    chain.next_height_to_preprocess.set(BlockHeight(257));
+    assert_eq!(*chain.next_height_to_preprocess.get(), BlockHeight(257));
+}

linera-core/src/chain_worker/state.rs

@@ -462,7 +462,10 @@ where
             .flatten()
             .copied()
             .collect::<BTreeSet<_>>();
-        let hashes = self.chain.block_hashes(heights.iter().copied()).await?;
+        let hashes = self
+            .chain
+            .block_hashes_for_heights(heights.iter().copied())
+            .await?;
 
         let blocks = self.read_confirmed_blocks(&hashes).await?;
 
@@ -1155,13 +1158,17 @@ where
         while current_height >= retransmit_from {
             // We arrived at current_height via previous_message_blocks links, starting from the
             // chain state and following the links downwards. So these blocks should all be in
-            // confirmed_log or preprocessed_blocks already.
+            // `block_hashes` already.
             heights_to_re_add.push(current_height);
             // Load the block at current_height to find the previous message block
-            let hash = match &*self.chain.block_hashes([current_height]).await? {
+            let hash = match &*self
+                .chain
+                .block_hashes_for_heights([current_height])
+                .await?
+            {
                 [hash] => *hash,
                 _ => {
-                    return Err(WorkerError::ConfirmedBlockHashNotFound {
+                    return Err(WorkerError::BlockHashNotFound {
                         height: current_height,
                         chain_id: self.chain_id(),
                     })
@@ -1263,8 +1270,7 @@ where
 
         // 1. Collect all sender chain IDs and block hashes before clearing.
         let sender_ids = self.chain.inboxes.indices().await?;
-        let hashes = self.chain.confirmed_log.read(..).await?;
-        let preprocessed = self.chain.preprocessed_blocks.index_values().await?;
+        let block_hashes = self.chain.block_hashes.index_values().await?;
 
         // 2. Snapshot safety-critical manager state so that we cannot be tricked
         //    into double-signing if the reset wipes votes we already cast.
@@ -1287,17 +1293,7 @@ where
         );
 
         // 4. Re-load certificates one at a time by hash and re-process them.
-        for (index, hash) in hashes.into_iter().enumerate() {
-            let height = BlockHeight(index as u64);
-            let cert = self
-                .storage
-                .read_certificate(hash)
-                .await?
-                .map(Arc::unwrap_or_clone)
-                .ok_or_else(|| WorkerError::LocalBlockNotFound { height, chain_id })?;
-            Box::pin(self.process_confirmed_block(cert, None)).await?;
-        }
-        for (height, hash) in preprocessed {
+        for (height, hash) in block_hashes {
             let cert = self
                 .storage
                 .read_certificate(hash)
@@ -1373,7 +1369,7 @@ where
         let mut hashes = Vec::new();
         let mut height = start;
         while height < end {
-            match self.chain.preprocessed_blocks.get(&height).await? {
+            match self.chain.block_hashes.get(&height).await? {
                 Some(hash) => hashes.push(hash),
                 None => break,
             }
@@ -1421,7 +1417,7 @@ where
         &self,
         heights: Vec<BlockHeight>,
     ) -> Result<Vec<CryptoHash>, WorkerError> {
-        Ok(self.chain.block_hashes(heights).await?)
+        Ok(self.chain.block_hashes_for_heights(heights).await?)
     }
 
     /// Gets proposed blobs from the manager for specified blob IDs.
@@ -1510,8 +1506,8 @@ where
     }
 
     /// Gets the next height to preprocess.
-    pub(crate) async fn get_next_height_to_preprocess(&self) -> Result<BlockHeight, WorkerError> {
-        Ok(self.chain.next_height_to_preprocess().await?)
+    pub(crate) fn get_next_height_to_preprocess(&self) -> BlockHeight {
+        *self.chain.next_height_to_preprocess.get()
     }
 
     /// Attempts to vote for a leader timeout, if possible.
@@ -1643,7 +1639,7 @@ where
         &self,
         height: BlockHeight,
     ) -> Result<Option<Arc<ConfirmedBlockCertificate>>, WorkerError> {
-        let certificate_hash = match self.chain.confirmed_log.get(height.try_into()?).await? {
+        let certificate_hash = match self.chain.block_hashes.get(&height).await? {
             Some(hash) => hash,
             None => return Ok(None),
         };
@@ -2020,7 +2016,7 @@ where
             info.requested_pending_message_bundles = bundles;
         }
         let hashes = chain
-            .block_hashes(query.request_sent_certificate_hashes_by_heights)
+            .block_hashes_for_heights(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 {
@@ -2041,18 +2037,18 @@ where
                 .previous_event_blocks
                 .multi_get(&stream_ids)
                 .await?;
-            let mut indices = Vec::new();
             let mut streams_with_heights = Vec::new();
             for (stream_id, height) in stream_ids.into_iter().zip(heights) {
                 if let Some(height) = height {
-                    let index = usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
-                    indices.push(index);
                     streams_with_heights.push((stream_id, height));
                 }
             }
-            let hashes = chain.confirmed_log.multi_get(indices).await?;
+            let hashes = chain
+                .block_hashes
+                .multi_get(streams_with_heights.iter().map(|(_, height)| height))
+                .await?;
             for (maybe_hash, (stream_id, height)) in hashes.into_iter().zip(streams_with_heights) {
-                let hash = maybe_hash.ok_or_else(|| WorkerError::ConfirmedLogEntryNotFound {
+                let hash = maybe_hash.ok_or_else(|| WorkerError::BlockHashNotFound {
                     height,
                     chain_id: info.chain_id,
                 })?;