Relax the inbox check. (#4791)

afck · web-flow · commit 4f9e497372a3 · 2025-10-14T08:43:50.000Z
## Motivation Currently a block proposal is always rejected (by a validator, or even by the local node) if any of the incoming messages included on this chain up to and including the proposal itself have not been received from the sender chain yet. For example, if: - block A0 on chain A has included a message from block B0 on chain B, - and now a proposal for block A1 is made that includes a message from block C0 on chain C, - and a validator has block C0 but not B0, it would currently reject that proposal. That is unnecessarily restrictive, because block A0 has already been signed by a quorum of validators, and it causes more round-trips and download work for clients. ## Proposal Relax the inbox check, so that only incoming messages in the proposal itself have to be received yet. The validator in the above example would accept the proposal. In other words, don't reject a new block proposal just because you have an old, already confirmed block in that chain with an incoming message whose sender block you have not seen yet. ## Test Plan CI; some tests have been updated. ## Release Plan - These changes should be backported to `testnet_conway`, then - be released in a new SDK, - be released in a validator hotfix, ideally at the same time. ## Links - Closes #413. - [reviewer checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)
diff --git a/linera-chain/src/chain.rs b/linera-chain/src/chain.rs
@@ -7,7 +7,6 @@ use std::{
     sync::Arc,
 };
 
-use futures::stream::{self, StreamExt, TryStreamExt};
 use linera_base::{
     crypto::{CryptoHash, ValidatorPublicKey},
     data_types::{
@@ -31,7 +30,6 @@ use linera_views::{
     reentrant_collection_view::{ReadGuardedView, ReentrantCollectionView},
     register_view::RegisterView,
     set_view::SetView,
-    store::ReadableKeyValueStore as _,
     views::{ClonableView, CryptoHashView, RootView, View},
 };
 use serde::{Deserialize, Serialize};
@@ -517,58 +515,6 @@ where
         Ok(())
     }
 
-    /// Verifies that this chain is up-to-date and all the messages executed ahead of time
-    /// have been properly received by now.
-    #[instrument(target = "telemetry_only", skip_all, fields(
-        chain_id = %self.chain_id()
-    ))]
-    pub async fn validate_incoming_bundles(&self) -> Result<(), ChainError> {
-        let chain_id = self.chain_id();
-        let pairs = self.inboxes.try_load_all_entries().await?;
-        let max_stream_queries = self.context().store().max_stream_queries();
-        let stream = stream::iter(pairs)
-            .map(|(origin, inbox)| async move {
-                if let Some(bundle) = inbox.removed_bundles.front().await? {
-                    return Err(ChainError::MissingCrossChainUpdate {
-                        chain_id,
-                        origin,
-                        height: bundle.height,
-                    });
-                }
-                Ok::<(), ChainError>(())
-            })
-            .buffer_unordered(max_stream_queries);
-        stream.try_collect::<Vec<_>>().await?;
-        Ok(())
-    }
-
-    /// Collects all missing sender blocks from removed bundles across all inboxes.
-    /// Returns a map of origin chain IDs to their respective missing block heights.
-    #[instrument(target = "telemetry_only", skip_all, fields(
-        chain_id = %self.chain_id()
-    ))]
-    pub async fn collect_missing_sender_blocks(
-        &self,
-    ) -> Result<BTreeMap<ChainId, Vec<BlockHeight>>, ChainError> {
-        let pairs = self.inboxes.try_load_all_entries().await?;
-        let max_stream_queries = self.context().store().max_stream_queries();
-        let stream = stream::iter(pairs)
-            .map(|(origin, inbox)| async move {
-                let mut missing_heights = Vec::new();
-                let bundles = inbox.removed_bundles.elements().await?;
-                for bundle in bundles {
-                    missing_heights.push(bundle.height);
-                }
-                Ok::<(ChainId, Vec<BlockHeight>), ChainError>((origin, missing_heights))
-            })
-            .buffer_unordered(max_stream_queries);
-        let results: Vec<(ChainId, Vec<BlockHeight>)> = stream.try_collect().await?;
-        Ok(results
-            .into_iter()
-            .filter(|(_, heights)| !heights.is_empty())
-            .collect())
-    }
-
     pub async fn next_block_height_to_receive(
         &self,
         origin: &ChainId,
@@ -715,6 +661,7 @@ where
     pub async fn remove_bundles_from_inboxes(
         &mut self,
         timestamp: Timestamp,
+        must_be_present: bool,
         incoming_bundles: impl IntoIterator<Item = &IncomingBundle>,
     ) -> Result<(), ChainError> {
         let chain_id = self.chain_id();
@@ -749,6 +696,16 @@ where
                     .remove_bundle(bundle)
                     .await
                     .map_err(|error| (chain_id, origin, error))?;
+                if must_be_present {
+                    ensure!(
+                        was_present,
+                        ChainError::MissingCrossChainUpdate {
+                            chain_id,
+                            origin,
+                            height: bundle.height,
+                        }
+                    );
+                }
                 if was_present && !bundle.is_skippable() {
                     removed_unskippable.insert(BundleInInbox::new(origin, bundle));
                 }
diff --git a/linera-core/src/chain_worker/state.rs b/linera-core/src/chain_worker/state.rs
@@ -839,7 +839,11 @@ where
         let local_time = self.storage.clock().current_time();
         let chain = &mut self.chain;
         chain
-            .remove_bundles_from_inboxes(block.header.timestamp, block.body.incoming_bundles())
+            .remove_bundles_from_inboxes(
+                block.header.timestamp,
+                false,
+                block.body.incoming_bundles(),
+            )
             .await?;
         let oracle_responses = Some(block.body.oracle_responses.clone());
         let (proposed_block, outcome) = block.clone().into_proposal();
@@ -1310,7 +1314,7 @@ where
         let local_time = self.storage.clock().current_time();
 
         self.chain
-            .remove_bundles_from_inboxes(block.timestamp, block.incoming_bundles())
+            .remove_bundles_from_inboxes(block.timestamp, true, block.incoming_bundles())
             .await?;
         let outcome = if let Some(outcome) = outcome {
             outcome.clone()
@@ -1329,8 +1333,6 @@ where
             .tip_state
             .get_mut()
             .update_counters(&block.transactions, &outcome.messages)?;
-        // Verify that the resulting chain would have no unconfirmed incoming messages.
-        chain.validate_incoming_bundles().await?;
         // Don't save the changes since the block is not confirmed yet.
         chain.rollback();
 
diff --git a/linera-core/src/client/mod.rs b/linera-core/src/client/mod.rs
@@ -1028,68 +1028,6 @@ impl<Env: Environment> Client<Env> {
         Ok(remote_log)
     }
 
-    /// Downloads only the specific sender blocks needed for missing cross-chain messages.
-    /// This is a targeted alternative to `find_received_certificates` that only downloads
-    /// the exact sender blocks we're missing, rather than searching through all received
-    /// certificates.
-    async fn download_missing_sender_blocks(
-        &self,
-        receiver_chain_id: ChainId,
-        missing_blocks: BTreeMap<ChainId, Vec<BlockHeight>>,
-    ) -> Result<(), ChainClientError> {
-        if missing_blocks.is_empty() {
-            return Ok(());
-        }
-
-        let (_, committee) = self.admin_committee().await?;
-        let nodes = self.make_nodes(&committee)?;
-
-        // Download certificates for each sender chain at the specific heights.
-        stream::iter(missing_blocks.into_iter())
-            .map(|(sender_chain_id, heights)| {
-                let height = heights.into_iter().max();
-                let mut shuffled_nodes = nodes.clone();
-                shuffled_nodes.shuffle(&mut rand::thread_rng());
-                async move {
-                    let Some(height) = height else {
-                        return Ok(());
-                    };
-                    // Try to download from any node.
-                    for node in &shuffled_nodes {
-                        if let Err(err) = self
-                            .download_sender_block_with_sending_ancestors(
-                                receiver_chain_id,
-                                sender_chain_id,
-                                height,
-                                node,
-                            )
-                            .await
-                        {
-                            tracing::debug!(
-                                %height,
-                                %receiver_chain_id,
-                                %sender_chain_id,
-                                %err,
-                                validator = %node.public_key,
-                                "Failed to fetch sender block",
-                            );
-                        } else {
-                            return Ok::<_, ChainClientError>(());
-                        }
-                    }
-                    // If all nodes fail, return an error.
-                    Err(ChainClientError::CannotDownloadMissingSenderBlock {
-                        chain_id: sender_chain_id,
-                        height,
-                    })
-                }
-            })
-            .buffer_unordered(self.options.max_joined_tasks)
-            .try_collect::<Vec<_>>()
-            .await?;
-        Ok(())
-    }
-
     /// Downloads a specific sender block and recursively downloads any earlier blocks
     /// that also sent a message to our chain, based on `previous_message_blocks`.
     ///
@@ -2177,7 +2115,7 @@ impl<Env: Environment> ChainClient<Env> {
     }
 
     /// Prepares the chain for the next operation, i.e. makes sure we have synchronized it up to
-    /// its current height and are not missing any received messages from the inbox.
+    /// its current height.
     #[instrument(level = "trace")]
     pub async fn prepare_chain(&self) -> Result<Box<ChainInfo>, ChainClientError> {
         #[cfg(with_metrics)]
@@ -2203,16 +2141,6 @@ impl<Env: Environment> ChainClient<Env> {
                 .await?;
         }
 
-        // Check if we're missing any sender blocks for cross-chain messages.
-        let missing_blocks = self
-            .chain_state_view()
-            .await?
-            .collect_missing_sender_blocks()
-            .await?;
-        // Download any sender blocks we're missing.
-        self.client
-            .download_missing_sender_blocks(self.chain_id, missing_blocks)
-            .await?;
         self.client.update_from_info(&info);
         Ok(info)
     }
diff --git a/linera-core/src/unit_tests/client_tests.rs b/linera-core/src/unit_tests/client_tests.rs
@@ -2845,19 +2845,15 @@ where
         )
         .await?;
 
-    // Test that prepare_chain downloads only the sender blocks for acknowledged messages.
-    // The new client has the inbox state showing that a message from sender was processed,
-    // but needs to download the sender block. The message from sender2 hasn't been
-    // processed yet, so its block should NOT be downloaded.
+    // Test that prepare_chain does not download any sender chain blocks.
     let info = receiver2.prepare_chain().await?;
     assert_eq!(info.next_block_height, BlockHeight::from(1));
 
-    // Verify that sender's block WAS downloaded.
     let local_node = &receiver2.client.local_node;
     let sender_info = local_node.chain_info(sender.chain_id()).await?;
     assert_eq!(
         sender_info.next_block_height,
-        BlockHeight::from(1),
+        BlockHeight::ZERO,
         "prepare_chain should download acknowledged sender blocks"
     );
 
diff --git a/linera-core/src/unit_tests/worker_tests.rs b/linera-core/src/unit_tests/worker_tests.rs
@@ -36,7 +36,7 @@ use linera_chain::{
         CertificateKind, CertificateValue, ConfirmedBlock, ConfirmedBlockCertificate,
         GenericCertificate, Timeout, ValidatedBlock,
     },
-    ChainError, ChainExecutionContext,
+    ChainError, ChainExecutionContext, ChainStateView,
 };
 use linera_execution::{
     committee::Committee,
@@ -47,11 +47,12 @@ use linera_execution::{
     test_utils::{
         dummy_chain_description, ExpectedCall, RegisterMockApplication, SystemExecutionState,
     },
-    ExecutionError, Message, MessageKind, OutgoingMessage, Query, QueryContext, QueryOutcome,
-    QueryResponse, SystemQuery, SystemResponse,
+    ExecutionError, ExecutionRuntimeContext, Message, MessageKind, OutgoingMessage, Query,
+    QueryContext, QueryOutcome, QueryResponse, SystemQuery, SystemResponse,
 };
 use linera_storage::{DbStorage, Storage, TestClock};
 use linera_views::{
+    context::Context,
     memory::MemoryDatabase,
     random::generate_test_namespace,
     store::TestKeyValueDatabase as _,
@@ -474,6 +475,18 @@ where
     }
 }
 
+/// Asserts that there are no "removed" bundles in the inbox, that have been included as
+/// incoming in a block but not received from the sender chain yet.
+async fn assert_no_removed_bundles<C>(chain: &ChainStateView<C>)
+where
+    C: Context + Clone + Send + Sync + 'static,
+    C::Extra: ExecutionRuntimeContext,
+{
+    for (_, inbox) in chain.inboxes.try_load_all_entries().await.unwrap() {
+        assert_eq!(inbox.removed_bundles.front().await.unwrap(), None);
+    }
+}
+
 fn direct_outgoing_message(
     recipient: ChainId,
     kind: MessageKind,
@@ -2478,7 +2491,7 @@ where
     {
         let chain = env.worker().chain_state_view(chain_2).await?;
         assert!(chain.is_active());
-        chain.validate_incoming_bundles().await?;
+        assert_no_removed_bundles(&chain).await;
     }
 
     // Process the bounced message and try to use the refund.
@@ -2519,7 +2532,7 @@ where
     {
         let chain = env.worker.chain_state_view(chain_1).await?;
         assert!(chain.is_active());
-        chain.validate_incoming_bundles().await?;
+        assert_no_removed_bundles(&chain).await;
     }
     Ok(())
 }
@@ -2575,7 +2588,7 @@ where
     {
         let admin_chain = env.worker().chain_state_view(admin_id).await?;
         assert!(admin_chain.is_active());
-        admin_chain.validate_incoming_bundles().await?;
+        assert_no_removed_bundles(&admin_chain).await;
         assert_eq!(
             BlockHeight::from(1),
             admin_chain.tip_state.get().next_block_height
@@ -2656,7 +2669,7 @@ where
             *user_chain.execution_state.system.admin_id.get(),
             Some(admin_id)
         );
-        user_chain.validate_incoming_bundles().await?;
+        assert_no_removed_bundles(&user_chain).await;
         matches!(
             &user_chain
                 .inboxes
@@ -2739,7 +2752,7 @@ where
             Some(admin_id)
         );
         assert_eq!(user_chain.execution_state.system.committees.get().len(), 2);
-        user_chain.validate_incoming_bundles().await?;
+        assert_no_removed_bundles(&user_chain).await;
         Ok(())
     }
 }
@@ -3033,10 +3046,15 @@ where
         // The admin chain has an anticipated message.
         let admin_chain = env.worker().chain_state_view(admin_id).await?;
         assert!(admin_chain.is_active());
-        assert_matches!(
-            admin_chain.validate_incoming_bundles().await,
-            Err(ChainError::MissingCrossChainUpdate { .. })
-        );
+        assert!(admin_chain
+            .inboxes
+            .try_load_entry(&user_id)
+            .await?
+            .unwrap()
+            .removed_bundles
+            .front()
+            .await?
+            .is_some());
     }
 
     // Try again to execute the transfer from the user chain to the admin chain.
@@ -3049,7 +3067,7 @@ where
         // The admin chain has no more anticipated messages.
         let admin_chain = env.worker().chain_state_view(admin_id).await?;
         assert!(admin_chain.is_active());
-        admin_chain.validate_incoming_bundles().await?;
+        assert_no_removed_bundles(&admin_chain).await;
     }
 
     // Let's make a certificate for a block creating another epoch.
diff --git a/linera-service/tests/local_net_tests.rs b/linera-service/tests/local_net_tests.rs
@@ -1028,13 +1028,12 @@ async fn test_update_validator_sender_gaps(config: LocalNetConfig) -> Result<()>
     sender_client
         .transfer(Amount::from_tokens(2), sender_chain, sender_chain)
         .await?;
+    receiver_client.process_inbox(receiver_chain).await?;
     // transfer some more to create a gap in the chain from the recipient's perspective
     sender_client
         .transfer(Amount::from_tokens(3), sender_chain, receiver_chain)
         .await?;
 
-    receiver_client.process_inbox(receiver_chain).await?;
-
     // Restart the stopped validator and stop another one.
     net.restart_validator(UNAWARE_VALIDATOR_INDEX).await?;
     net.stop_validator(STOPPED_VALIDATOR_INDEX).await?;
@@ -1057,11 +1056,9 @@ async fn test_update_validator_sender_gaps(config: LocalNetConfig) -> Result<()>
         BlockHeight::ZERO
     );
 
-    // Try to send tokens from receiver to sender. Receiver should have a gap in the
-    // sender chain at this point.
-    receiver_client
-        .transfer(Amount::from_tokens(4), receiver_chain, sender_chain)
-        .await?;
+    // Process the last sender block. The client has a gap in the sender chain and does
+    // not update the unaware validator about block 1.
+    receiver_client.process_inbox(receiver_chain).await?;
 
     // Synchronize the validator
     let validator_address = net.validator_address(UNAWARE_VALIDATOR_INDEX);
