Prune block_hashes at every checkpoint

linera-chain/src/chain.rs

@@ -556,8 +556,10 @@ where
     }
 
     /// Inserts `(height, hash)` into `block_hashes` and updates the
-    /// `next_height_to_preprocess` register accordingly. Every write to
+    /// `next_height_to_preprocess` register accordingly. Every insertion into
     /// `block_hashes` must go through this helper so the register stays in sync.
+    /// Removals are confined to `prune_block_hashes`, which only drops heights below the
+    /// highest one and so leaves the register untouched.
     fn insert_block_hash(
         &mut self,
         height: BlockHeight,
@@ -1518,10 +1520,68 @@ where
         self.insert_block_hash(block.header.height, hash)?;
         if block.body.starts_with_checkpoint() {
             self.latest_checkpoint_height.set(Some(block.header.height));
+            self.prune_block_hashes(block).await?;
         }
         Ok(updated_streams)
     }
 
+    /// Drops the entries of `block_hashes` below the checkpoint block that the chain no
+    /// longer needs, so that the map does not grow without bound as the chain advances.
+    ///
+    /// Two kinds of pre-checkpoint block survive:
+    ///
+    /// * The blocks the checkpoint vouches for in its `outbox_block_hashes`, i.e. the ones
+    ///   that still carry unacknowledged messages. These are the blocks a node bootstrapping
+    ///   from the checkpoint is given, and the only ones a surviving `previous_message_blocks`
+    ///   anchor can point at, since the checkpoint drops the anchor of every fully
+    ///   acknowledged recipient.
+    /// * The blocks still queued in an outbox. Delivery is confirmed by the recipient's
+    ///   validators, which is a separate path from the acknowledgement that empties
+    ///   `unfinalized_message_blocks`, so a queue can outlive the checkpoint's guarantee.
+    ///   Dropping such a block would leave the outbox unable to build a cross-chain request
+    ///   for it, and hence unable to ever drain. Outbox queues are local state, so unlike the
+    ///   vouched-for set this one differs between nodes: what remains is bounded, not
+    ///   identical everywhere.
+    ///
+    /// Only heights below the checkpoint are removed, so `next_height_to_preprocess` — which
+    /// tracks the highest known height — stays accurate.
+    async fn prune_block_hashes(&mut self, block: &Block) -> Result<(), ChainError> {
+        let Some(OracleResponse::Checkpoint {
+            outbox_block_hashes,
+            ..
+        }) = block.body.oracle_responses.first().and_then(|r| r.first())
+        else {
+            return Err(ChainError::InternalError(
+                "checkpoint block is missing its OracleResponse::Checkpoint".into(),
+            ));
+        };
+        let vouched_for = outbox_block_hashes.iter().collect::<HashSet<_>>();
+        // Every outbox, not just the ones in `nonempty_outboxes`: on a client that index only
+        // covers tracked targets, but an untracked target's queue is resurrected by
+        // `reconcile_outbox_index` if the chain is tracked later on. An outbox is removed once
+        // its queue drains, so this loads no more entries than there are pending recipients.
+        let mut queued = BTreeSet::new();
+        for (_, outbox) in self.outboxes.try_load_all_entries().await? {
+            queued.extend(outbox.queue.elements().await?);
+        }
+        let obsolete = self
+            .block_hashes
+            .index_values()
+            .await?
+            .into_iter()
+            .filter(|(height, hash)| {
+                *height < block.header.height
+                    && !vouched_for.contains(hash)
+                    && !queued.contains(height)
+            })
+            .map(|(height, _)| height)
+            .collect::<Vec<_>>();
+        for height in obsolete {
+            self.block_hashes.remove(&height)?;
+        }
+        Ok(())
+    }
+
     /// 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(
@@ -1695,13 +1755,14 @@ where
                 if *outbox.next_height_to_schedule.get() > block_height {
                     continue; // We already added this recipient's messages to the outbox.
                 }
+                // A missing entry means a checkpoint pruned the block: the recipient had
+                // acknowledged everything we sent it below the checkpoint, so there is no
+                // predecessor left to chain to. That is the same state a node that
+                // bootstrapped from the checkpoint is in, where the outbox is rebuilt from
+                // scratch and starts out at height zero.
                 let maybe_prev_hash = match outbox.next_height_to_schedule.get().try_sub_one().ok()
                 {
-                    Some(height) => {
-                        Some(self.block_hashes.get(&height).await?.ok_or_else(|| {
-                            ChainError::CorruptedChainState("missing entry in block_hashes".into())
-                        })?)
-                    }
+                    Some(height) => self.block_hashes.get(&height).await?,
                     None => None, // No message to that sender was added yet.
                 };
                 // Only schedule if this block contains the next message for that recipient.
@@ -1715,12 +1776,14 @@ where
                     }
                     (Some(_), None) => {
                         // The outbox already has a previous height for this recipient,
-                        // but this block's body recorded no predecessor — that means
-                        // this is the first non-`CheckpointAck` send to this recipient,
-                        // even though earlier `CheckpointAck`-only blocks have already
-                        // been added to the off-chain outbox. We can still schedule:
-                        // the bundle will carry `previous_height = None`, which the
-                        // receiver accepts as "first ever".
+                        // but this block's body recorded no predecessor. Either earlier
+                        // `CheckpointAck`-only blocks were added to the off-chain outbox
+                        // while being skipped from `body.previous_message_blocks`, or a
+                        // checkpoint dropped the recipient's anchor because it had
+                        // acknowledged everything we sent it. Both mean the recipient has
+                        // no unreceived message from us below this block, so we can
+                        // schedule: the bundle carries `previous_height = None`, which
+                        // only tells the receiver to skip its gap check.
                     }
                     (None, Some((_, prev_msg_block_height))) => {
                         // We have no previously processed block in the outbox, but we are
@@ -1733,8 +1796,8 @@ where
                     }
                     (Some(ref prev_hash), Some((prev_msg_block_hash, _))) => {
                         // Only process the outbox if the hashes match. A mismatch can
-                        // arise legitimately when intermediate `CheckpointAck`-only
-                        // blocks sit in the off-chain outbox but are skipped from
+                        // arise legitimately when intermediate blocks sit in the
+                        // off-chain outbox but are skipped from
                         // `body.previous_message_blocks`; same fallback as the
                         // `(Some, None)` arm above.
                         if prev_hash != prev_msg_block_hash {

linera-core/src/unit_tests/client_tests.rs

@@ -4251,6 +4251,22 @@ where
     // Producer.1: checkpoint #1. unfinalized_message_blocks[recipient] = {(0, 0)}.
     producer.checkpoint().await.unwrap().unwrap();
 
+    // The transfer is unacknowledged, so checkpoint #1 vouches for its block and must keep
+    // it in `block_hashes` — that is the block a node bootstrapping from the checkpoint is
+    // handed, and the one the recipient's `previous_message_blocks` anchor points at.
+    {
+        let producer_state = producer
+            .client
+            .local_node
+            .chain_state_view(producer_id)
+            .await?;
+        assert!(producer_state
+            .block_hashes
+            .get(&BlockHeight::ZERO)
+            .await?
+            .is_some());
+    }
+
     // Recipient.0: consume the transfer. Now the inbox's `next_cursor_to_remove[producer]`
     // is past the transfer's cursor and `pending_checkpoint_ack_targets` contains the
     // producer.
@@ -4338,6 +4354,60 @@ where
         "cycle should terminate: producer's next checkpoint must not notify the recipient",
     );
 
+    // Checkpoint #2 vouches for nothing, so it prunes the blocks below it from
+    // `block_hashes` — checkpoint #1 at height 1 and the ack-consuming block at height 2.
+    //
+    // The transfer at height 0 stays, because this client's outbox still queues it: the
+    // recipient's acknowledgement came back as a message, whereas an outbox is only drained
+    // by a delivery confirmation from the recipient's validators, which this client has not
+    // seen. Pruning height 0 would leave the outbox unable to build its cross-chain request,
+    // and hence unable to ever drain.
+    {
+        let producer_state = producer
+            .client
+            .local_node
+            .chain_state_view(producer_id)
+            .await?;
+        let outbox = producer_state
+            .outboxes
+            .try_load_entry(&recipient_id)
+            .await?
+            .expect("the producer has an outbox for the recipient");
+        assert_eq!(outbox.queue.elements().await?, vec![BlockHeight::ZERO]);
+        assert_eq!(
+            producer_state.block_hashes.indices().await?,
+            vec![BlockHeight::ZERO, checkpoint_2.block().header.height],
+        );
+    }
+
+    // The producer→recipient lane must stay usable after the completed cycle, even for a
+    // node that bootstrapped from checkpoint #2 and therefore holds no pre-checkpoint
+    // blocks. Simulate one by resetting the producer to its latest checkpoint. Checkpoint
+    // #2 dropped the recipient's `previous_message_blocks` anchor (everything sent to it
+    // was acknowledged), so the new block carries no predecessor: the reset producer can
+    // execute it without the height-0 entry of `block_hashes`, and the checkpointed
+    // recipient accepts the bundle.
+    producer
+        .client
+        .local_node
+        .reset_and_reexecute_chain(producer_id)
+        .await?;
+    producer
+        .transfer_to_account(
+            AccountOwner::CHAIN,
+            Amount::ONE,
+            Account::chain(recipient_id),
+        )
+        .await
+        .unwrap_ok_committed();
+    recipient.synchronize_from_validators().await?;
+    recipient.process_inbox().await?;
+    assert_eq!(
+        recipient.local_balance().await?,
+        Amount::from_tokens(2),
+        "the post-cycle transfer must reach the checkpointed recipient",
+    );
+
     Ok(())
 }
 

linera-execution/src/execution.rs

@@ -54,7 +54,9 @@ pub struct ExecutionStateViewInner<C> {
     pub system: SystemExecutionStateView<C>,
     /// User applications.
     pub users: ReentrantCollectionView<C, ApplicationId, KeyValueStoreView<C>>,
-    /// The heights of previous blocks that sent messages to the same recipients.
+    /// The heights of previous blocks that sent messages to the same recipients. A
+    /// checkpoint drops the entry of every recipient that has acknowledged all of them,
+    /// so an entry never points at a block a checkpoint stopped guaranteeing.
     pub previous_message_blocks: MapView<C, ChainId, BlockHeight>,
     /// The heights of previous blocks that published events to the same streams.
     pub previous_event_blocks: MapView<C, StreamId, BlockHeight>,
@@ -202,6 +204,43 @@ where
         // reset the set so the next own checkpoint only fires for origins that send
         // us a fresh non-`Checkpoint` message in the meantime.
         self.system.pending_checkpoint_ack_targets.clear();
+        self.prune_message_anchors().await?;
+        Ok(())
+    }
+
+    /// Drops the `previous_message_blocks` anchor of every recipient that has acknowledged
+    /// all the messages we sent it, i.e. every recipient absent from
+    /// `unfinalized_message_blocks`.
+    ///
+    /// A checkpoint only guarantees the availability of the blocks listed in the
+    /// checkpoint's `outbox_block_hashes`, which are exactly the ones still referenced by
+    /// `unfinalized_message_blocks`. An anchor pointing below the checkpoint at any other
+    /// block would dangle: a node that bootstrapped from the checkpoint never saw that
+    /// block, so resolving the anchor to a hash would fail.
+    ///
+    /// Only fully acknowledged recipients are dropped. A recipient with unacknowledged
+    /// bundles keeps its anchor, so the chain of `previous_message_blocks` links that lets
+    /// it detect and fetch skipped message blocks stays intact — unlike
+    /// `previous_event_blocks`, which the checkpoint clears outright because the summary
+    /// events supersede the pruned ones. Its anchor is also the highest height in its
+    /// `unfinalized_message_blocks` entry — both are written together for every
+    /// non-`CheckpointAck` bundle, and an acknowledgement only trims a prefix of the
+    /// cursors — so the anchor is covered by `outbox_block_hashes`.
+    async fn prune_message_anchors(&mut self) -> Result<(), ExecutionError> {
+        let mut acknowledged = Vec::new();
+        for recipient in self.previous_message_blocks.indices().await? {
+            if !self
+                .system
+                .unfinalized_message_blocks
+                .contains_key(&recipient)
+                .await?
+            {
+                acknowledged.push(recipient);
+            }
+        }
+        for recipient in acknowledged {
+            self.previous_message_blocks.remove(&recipient)?;
+        }
         Ok(())
     }
 

linera-execution/src/unit_tests/system_tests.rs

@@ -290,14 +290,33 @@ async fn checkpoint_notifies_origins_and_receive_records_finalization() -> anyho
     };
 
     // Producer side: apply a checkpoint with non-empty origin_cursors and check that
-    // it queues one `SystemMessage::CheckpointAck` per origin.
+    // it queues one `SystemMessage::CheckpointAck` per origin, and that it prunes the
+    // `previous_message_blocks` anchors of recipients that acknowledged everything.
     use linera_views::{batch::Batch, store::WritableKeyValueStore as _, views::View as _};
     let mut view = SystemExecutionState {
         description: Some(dummy_chain_description(0)),
         ..SystemExecutionState::default()
     }
     .into_view()
     .await;
+    // `recipient_acked` has no `unfinalized_message_blocks` entry, so the checkpoint stops
+    // guaranteeing the availability of its anchor block and must drop the anchor.
+    // `recipient_pending` still owes us an acknowledgement at its anchor height, so its
+    // anchor is covered by `outbox_block_hashes` and must survive.
+    let recipient_acked = dummy_chain_description(4).id();
+    let recipient_pending = dummy_chain_description(5).id();
+    let pending_anchor = BlockHeight::from(6);
+    view.previous_message_blocks
+        .insert(&recipient_acked, BlockHeight::from(4))?;
+    view.previous_message_blocks
+        .insert(&recipient_pending, pending_anchor)?;
+    view.system.unfinalized_message_blocks.insert(
+        &recipient_pending,
+        std::collections::BTreeSet::from([Cursor {
+            height: pending_anchor,
+            index: 0,
+        }]),
+    )?;
     let mut batch = Batch::new();
     view.pre_save(&mut batch)?;
     view.context().store().write_batch(batch).await?;
@@ -328,6 +347,16 @@ async fn checkpoint_notifies_origins_and_receive_records_finalization() -> anyho
             })
         );
     }
+    assert!(
+        !view
+            .previous_message_blocks
+            .contains_key(&recipient_acked)
+            .await?
+    );
+    assert_eq!(
+        view.previous_message_blocks.get(&recipient_pending).await?,
+        Some(pending_anchor)
+    );
 
     // Receiver side: pre-populate `unfinalized_message_blocks` with three heights
     // from origin_a (the block-end hook would normally do this), then deliver the