Avoid loading empty outboxes when creating network actions. (#4119)

## Motivation

In #3989 I had to
remove the code that deletes empty outboxes. However, we can actually
still do that in some cases, and we can keep a separate list of nonempty
outboxes to avoid reading empty ones.

## Proposal

Delete empty outboxes if that outbox is not ahead of chain execution.

Keep a list of nonempty outboxes in the chain state to avoid loading
empty ones.

## Test Plan

This improved the benchmark for me again:

```
same_chain_native_token_transfers
                        time:   [152.54 ms 155.17 ms 158.41 ms]
                        change: [-27.026% -24.383% -21.712%] (p = 0.00 < 0.05)
                        Performance has improved.
```

CI should catch regressions.

## Release Plan

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

## Links

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

Author: afck

linera-chain/src/chain.rs

@@ -28,7 +28,7 @@ use linera_views::{
     context::Context,
     log_view::LogView,
     map_view::MapView,
-    reentrant_collection_view::ReentrantCollectionView,
+    reentrant_collection_view::{ReadGuardedView, ReentrantCollectionView},
     register_view::RegisterView,
     set_view::SetView,
     store::ReadableKeyValueStore as _,
@@ -278,6 +278,8 @@ where
     /// Number of outgoing messages in flight for each block height.
     /// We use a `RegisterView` to prioritize speed for small maps.
     pub outbox_counters: RegisterView<C, BTreeMap<BlockHeight, u32>>,
+    /// 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>,
@@ -414,15 +416,22 @@ where
                 .outbox_counters
                 .get_mut()
                 .get_mut(&update)
-                .expect("message counter should be present");
-            *counter = counter
-                .checked_sub(1)
-                .expect("message counter should not underflow");
+                .ok_or_else(|| {
+                    ChainError::InternalError("message counter should be present".into())
+                })?;
+            *counter = counter.checked_sub(1).ok_or(ArithmeticError::Underflow)?;
             if *counter == 0 {
                 // Important for the test in `all_messages_delivered_up_to`.
                 self.outbox_counters.get_mut().remove(&update);
             }
         }
+        if outbox.queue.count() == 0 {
+            self.nonempty_outboxes.get_mut().remove(target);
+            // If the outbox is empty and not ahead of the executed blocks, remove it.
+            if *outbox.next_height_to_schedule.get() <= self.tip_state.get().next_block_height {
+                self.outboxes.remove_entry(target)?;
+            }
+        }
         #[cfg(with_metrics)]
         metrics::NUM_OUTBOXES
             .with_label_values(&[])
@@ -696,6 +705,21 @@ where
         Ok(())
     }
 
+    /// Returns the chain IDs of all recipients for which a message is waiting in the outbox.
+    pub fn nonempty_outbox_chain_ids(&self) -> Vec<ChainId> {
+        self.nonempty_outboxes.get().iter().copied().collect()
+    }
+
+    /// Returns the outboxes for the given targets, or an error if any of them are missing.
+    pub async fn load_outboxes(
+        &self,
+        targets: &[ChainId],
+    ) -> Result<Vec<ReadGuardedView<OutboxStateView<C>>>, ChainError> {
+        let vec_of_options = self.outboxes.try_load_entries(targets).await?;
+        let optional_vec = vec_of_options.into_iter().collect::<Option<Vec<_>>>();
+        optional_vec.ok_or_else(|| ChainError::InternalError("Missing outboxes".into()))
+    }
+
     /// Executes a block: first the incoming messages, then the main operation.
     /// Does not update chain state other than the execution state.
     #[expect(clippy::too_many_arguments)]
@@ -1146,6 +1170,7 @@ where
 
         // Update the outboxes.
         let outbox_counters = self.outbox_counters.get_mut();
+        let nonempty_outboxes = self.nonempty_outboxes.get_mut();
         let targets = recipients.into_iter().collect::<Vec<_>>();
         let outboxes = self.outboxes.try_load_entries_mut(&targets).await?;
         for (mut outbox, target) in outboxes.into_iter().zip(&targets) {
@@ -1176,6 +1201,7 @@ where
             }
             if outbox.schedule_message(block_height)? {
                 *outbox_counters.entry(block_height).or_default() += 1;
+                nonempty_outboxes.insert(*target);
             }
         }
 

linera-core/src/chain_worker/state/mod.rs

@@ -448,16 +448,15 @@ where
     /// Loads pending cross-chain requests.
     async fn create_network_actions(&self) -> Result<NetworkActions, WorkerError> {
         let mut heights_by_recipient = BTreeMap::<_, Vec<_>>::new();
-        let mut targets = self.chain.outboxes.indices().await?;
+        let mut targets = self.chain.nonempty_outbox_chain_ids();
         if let Some(tracked_chains) = self.tracked_chains.as_ref() {
             let tracked_chains = tracked_chains
                 .read()
                 .expect("Panics should not happen while holding a lock to `tracked_chains`");
             targets.retain(|target| tracked_chains.contains(target));
         }
-        let outboxes = self.chain.outboxes.try_load_entries(&targets).await?;
+        let outboxes = self.chain.load_outboxes(&targets).await?;
         for (target, outbox) in targets.into_iter().zip(outboxes) {
-            let outbox = outbox.expect("Only existing outboxes should be referenced by `indices`");
             let heights = outbox.queue.elements().await?;
             heights_by_recipient.insert(target, heights);
         }
@@ -539,14 +538,13 @@ where
         let Some(tracked_chains) = self.tracked_chains.as_ref() else {
             return Ok(false);
         };
-        let mut targets = self.chain.outboxes.indices().await?;
+        let mut targets = self.chain.nonempty_outbox_chain_ids();
         {
             let tracked_chains = tracked_chains.read().unwrap();
             targets.retain(|target| tracked_chains.contains(target));
         }
-        let outboxes = self.chain.outboxes.try_load_entries(&targets).await?;
+        let outboxes = self.chain.load_outboxes(&targets).await?;
         for outbox in outboxes {
-            let outbox = outbox.expect("Only existing outboxes should be referenced by `indices`");
             let front = outbox.queue.front();
             if front.is_some_and(|key| *key <= height) {
                 return Ok(false);

linera-service-graphql-client/gql/service_schema.graphql

@@ -361,6 +361,10 @@ type ChainStateExtendedView {
 	"""
 	outboxCounters: JSONObject!
 	"""
+	Outboxes with at least one pending message. This allows us to avoid loading all outboxes.
+	"""
+	nonemptyOutboxes: [ChainId!]!
+	"""
 	Blocks that have been verified but not executed yet, and that may not be contiguous.
 	"""
 	preprocessedBlocks: MapView_BlockHeight_CryptoHash_1bae6d76!