Use BucketQueueView in the protocol. (#2807)

## Motivation

The `BucketQueueView` allows to use queues more efficiently.

## Proposal

It is known that using Queues on Key-Value stores is kind of an
antipattern for KeyValue stores.
Grouping entries into buckets can reduce this problem.

The problem is that grouping increases the memory cost and exposes to
possible Out Of Memory. Thus we did the following:
* For `QueueView` with `BlockHeight` being the type, we use a bucket
size of 1000 since BlockHeight is just 8 bytes.
* For `QueueView` with `TimestampedBundleInInbox` we use a bucket size
of 100 since this type is essentially 4 Cryptohashes, a few number and a
channel name which can be long but not too long.
* For `QueueView` with `MessageBundle` no change is made as this type
can be very large.

It is to be pointed out that `BucketQueueView` has two differences with
`QueueView`:
* Buckets indeed.
* The `front` is no longer an async operation, which means the front is
accessible just after the `load`. On the other hand the `delete_front`
becomes an async operation.

## Test Plan

(A) The CI is the recommended way.

(B) More tests would be needed for correctness.

(C) Runtime would be helpful to gauge the gain effectively obtained.

## Release Plan

This PR changes the way that data is stored in the storage, therefore
using it on existing TestNet / DevNet is impossible.

## Links

None.

Author: MathieuDutSik

linera-chain/src/chain.rs

@@ -31,10 +31,10 @@ use linera_execution::{
     ResourceTracker, ServiceRuntimeEndpoint, TransactionTracker,
 };
 use linera_views::{
+    bucket_queue_view::BucketQueueView,
     context::Context,
     log_view::LogView,
     map_view::MapView,
-    queue_view::QueueView,
     reentrant_collection_view::ReentrantCollectionView,
     register_view::RegisterView,
     set_view::SetView,
@@ -205,6 +205,10 @@ impl BundleInInbox {
     }
 }
 
+// The `TimestampedBundleInInbox` is a relatively small type, so a total
+// of 100 seems reasonable for the storing of the data.
+const TIMESTAMPBUNDLE_BUCKET_SIZE: usize = 100;
+
 /// A view accessing the state of a chain.
 #[derive(Debug, RootView, ClonableView, SimpleObject)]
 #[graphql(cache_control(no_cache))]
@@ -239,7 +243,8 @@ where
     /// Mailboxes used to receive messages indexed by their origin.
     pub inboxes: ReentrantCollectionView<C, Origin, InboxStateView<C>>,
     /// A queue of unskippable bundles, with the timestamp when we added them to the inbox.
-    pub unskippable_bundles: QueueView<C, TimestampedBundleInInbox>,
+    pub unskippable_bundles:
+        BucketQueueView<C, TimestampedBundleInInbox, TIMESTAMPBUNDLE_BUCKET_SIZE>,
     /// Unskippable bundles that have been removed but are still in the queue.
     pub removed_unskippable_bundles: SetView<C, BundleInInbox>,
     /// The heights of previous blocks that sent messages to the same recipients.
@@ -680,10 +685,10 @@ where
         }
         if !removed_unskippable.is_empty() {
             // Delete all removed bundles from the front of the unskippable queue.
-            let maybe_front = self.unskippable_bundles.front().await?;
+            let maybe_front = self.unskippable_bundles.front();
             if maybe_front.is_some_and(|ts_entry| removed_unskippable.remove(&ts_entry.entry)) {
-                self.unskippable_bundles.delete_front();
-                while let Some(ts_entry) = self.unskippable_bundles.front().await? {
+                self.unskippable_bundles.delete_front().await?;
+                while let Some(ts_entry) = self.unskippable_bundles.front() {
                     if !removed_unskippable.remove(&ts_entry.entry) {
                         if !self
                             .removed_unskippable_bundles
@@ -694,7 +699,7 @@ where
                         }
                         self.removed_unskippable_bundles.remove(&ts_entry.entry)?;
                     }
-                    self.unskippable_bundles.delete_front();
+                    self.unskippable_bundles.delete_front().await?;
                 }
             }
             for entry in removed_unskippable {

linera-chain/src/outbox.rs

@@ -8,8 +8,8 @@ use linera_base::data_types::{ArithmeticError, BlockHeight};
 #[cfg(with_testing)]
 use linera_views::context::MemoryContext;
 use linera_views::{
+    bucket_queue_view::BucketQueueView,
     context::Context,
-    queue_view::QueueView,
     register_view::RegisterView,
     views::{ClonableView, View, ViewError},
 };
@@ -34,6 +34,12 @@ static OUTBOX_SIZE: LazyLock<HistogramVec> = LazyLock::new(|| {
     )
 });
 
+// The number of block heights in a bucket
+// The `BlockHeight` has just 8 bytes so the size is constant.
+// This means that by choosing a size of 1000, we have a
+// reasonable size that will not create any memory issues.
+const BLOCK_HEIGHT_BUCKET_SIZE: usize = 1000;
+
 /// The state of an outbox
 /// * An outbox is used to send messages to another chain.
 /// * Internally, this is implemented as a FIFO queue of (increasing) block heights.
@@ -50,7 +56,7 @@ where
     pub next_height_to_schedule: RegisterView<C, BlockHeight>,
     /// Keep sending these certified blocks of ours until they are acknowledged by
     /// receivers.
-    pub queue: QueueView<C, BlockHeight>,
+    pub queue: BucketQueueView<C, BlockHeight, BLOCK_HEIGHT_BUCKET_SIZE>,
 }
 
 impl<C> OutboxStateView<C>
@@ -82,11 +88,11 @@ where
         height: BlockHeight,
     ) -> Result<Vec<BlockHeight>, ViewError> {
         let mut updates = Vec::new();
-        while let Some(h) = self.queue.front().await? {
+        while let Some(h) = self.queue.front().cloned() {
             if h > height {
                 break;
             }
-            self.queue.delete_front();
+            self.queue.delete_front().await?;
             updates.push(h);
         }
         #[cfg(with_metrics)]

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

@@ -570,7 +570,7 @@ where
         let chain = &mut self.state.chain;
         if let (Some(epoch), Some(entry)) = (
             chain.execution_state.system.epoch.get(),
-            chain.unskippable_bundles.front().await?,
+            chain.unskippable_bundles.front(),
         ) {
             let elapsed = self
                 .state

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

@@ -549,8 +549,8 @@ where
         let outboxes = self.chain.outboxes.try_load_entries(&targets).await?;
         for outbox in outboxes {
             let outbox = outbox.expect("Only existing outboxes should be referenced by `indices`");
-            let front = outbox.queue.front().await?;
-            if front.is_some_and(|key| key <= height) {
+            let front = outbox.queue.front();
+            if front.is_some_and(|key| *key <= height) {
                 return Ok(false);
             }
         }

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

@@ -207,6 +207,14 @@ A block height to identify blocks in a chain
 scalar BlockHeight
 
 
+type BucketQueueView_BlockHeight_e824a938 {
+	entries(count: Int): [BlockHeight!]!
+}
+
+type BucketQueueView_TimestampedBundleInInbox_5a630c55 {
+	entries(count: Int): [TimestampedBundleInInbox!]!
+}
+
 """
 An origin and cursor of a unskippable bundle that is no longer in our inbox.
 """
@@ -335,7 +343,7 @@ type ChainStateExtendedView {
 	"""
 	A queue of unskippable bundles, with the timestamp when we added them to the inbox.
 	"""
-	unskippableBundles: QueueView_TimestampedBundleInInbox_5a630c55!
+	unskippableBundles: BucketQueueView_TimestampedBundleInInbox_5a630c55!
 	"""
 	Unskippable bundles that have been removed but are still in the queue.
 	"""
@@ -895,7 +903,7 @@ type OutboxStateView {
 	Keep sending these certified blocks of ours until they are acknowledged by
 	receivers.
 	"""
-	queue: QueueView_BlockHeight_e824a938!
+	queue: BucketQueueView_BlockHeight_e824a938!
 }
 
 """
@@ -992,18 +1000,10 @@ type QueryRoot {
 	version: VersionInfo!
 }
 
-type QueueView_BlockHeight_e824a938 {
-	entries(count: Int): [BlockHeight!]!
-}
-
 type QueueView_MessageBundle_f4399f0b {
 	entries(count: Int): [MessageBundle!]!
 }
 
-type QueueView_TimestampedBundleInInbox_5a630c55 {
-	entries(count: Int): [TimestampedBundleInInbox!]!
-}
-
 """
 The recipient of a transfer
 """

linera-views/src/views/bucket_queue_view.rs

@@ -126,6 +126,7 @@ fn stored_indices<T>(stored_data: &VecDeque<(usize, Bucket<T>)>, position: usize
 /// The size `N` has to be chosen by taking into account the size of the type `T`
 /// and the basic size of a block. For example a total size of 100 bytes to 10 KB
 /// seems adequate.
+#[derive(Debug)]
 pub struct BucketQueueView<C, T, const N: usize> {
     context: C,
     /// The buckets of stored data. If missing, then it has not been loaded. The first index is always loaded.