Introduce the "find_keys_in_interval" / "find_key_values_in_interval".

Author: MathieuDutSik

linera-execution/src/execution_state_actor.rs

@@ -24,7 +24,7 @@ use linera_base::{
     ownership::ChainOwnership,
     time::Instant,
 };
-use linera_views::{batch::Batch, context::Context, views::View};
+use linera_views::{batch::Batch, context::Context, store::KeyInterval, views::View};
 use oneshot::Sender;
 use reqwest::{header::HeaderMap, Client, Url};
 use tracing::{info_span, instrument, Instrument as _};
@@ -35,9 +35,10 @@ use crate::{
     system::{CreateApplicationResult, OpenChainConfig},
     util::{OracleResponseExt as _, RespondExt as _},
     ApplicationDescription, ApplicationId, ExecutionError, ExecutionRuntimeContext,
-    ExecutionStateView, JsVec, Message, MessageContext, MessageKind, ModuleId, Operation,
-    OperationContext, OutgoingMessage, ProcessStreamsContext, QueryContext, QueryOutcome,
-    ResourceController, SystemMessage, TransactionTracker, UserContractCode, UserServiceCode,
+    ExecutionStateView, IntervalKeyValues, IntervalKeys, JsVec, Message, MessageContext,
+    MessageKind, ModuleId, Operation, OperationContext, OutgoingMessage, ProcessStreamsContext,
+    QueryContext, QueryOutcome, ResourceController, SystemMessage, TransactionTracker,
+    UserContractCode, UserServiceCode,
 };
 
 /// Actor for handling requests to the execution state.
@@ -405,6 +406,32 @@ where
                 callback.respond(result);
             }
 
+            FindKeysInInterval {
+                id,
+                key_interval,
+                callback,
+            } => {
+                let view = self.state.users.try_load_entry(&id).await?;
+                let result = match view {
+                    Some(view) => view.find_keys_in_interval(key_interval).await?,
+                    None => (Vec::new(), true),
+                };
+                callback.respond(result);
+            }
+
+            FindKeyValuesInInterval {
+                id,
+                key_interval,
+                callback,
+            } => {
+                let view = self.state.users.try_load_entry(&id).await?;
+                let result = match view {
+                    Some(view) => view.find_key_values_in_interval(key_interval).await?,
+                    None => (Vec::new(), true),
+                };
+                callback.respond(result);
+            }
+
             WriteBatch {
                 id,
                 batch,
@@ -1400,6 +1427,20 @@ pub enum ExecutionRequest {
         callback: Sender<Vec<(Vec<u8>, Vec<u8>)>>,
     },
 
+    FindKeysInInterval {
+        id: ApplicationId,
+        key_interval: KeyInterval,
+        #[debug(skip)]
+        callback: Sender<IntervalKeys>,
+    },
+
+    FindKeyValuesInInterval {
+        id: ApplicationId,
+        key_interval: KeyInterval,
+        #[debug(skip)]
+        callback: Sender<IntervalKeyValues>,
+    },
+
     WriteBatch {
         id: ApplicationId,
         batch: Batch,

linera-execution/src/lib.rs

@@ -45,9 +45,12 @@ use linera_base::{
     ownership::ChainOwnership,
     vm::VmRuntime,
 };
-use linera_views::{batch::Batch, ViewError};
+use linera_views::{batch::Batch, store::KeyInterval, ViewError};
 use serde::{Deserialize, Serialize};
 use system::AdminOperation;
+
+pub type IntervalKeys = (Vec<Vec<u8>>, bool);
+pub type IntervalKeyValues = (Vec<(Vec<u8>, Vec<u8>)>, bool);
 use thiserror::Error;
 pub use web_thread_pool::Pool as ThreadPool;
 use web_thread_select as web_thread;
@@ -724,6 +727,8 @@ pub trait BaseRuntime {
     type ReadValueBytes: fmt::Debug + Send + Sync;
     type FindKeysByPrefix: fmt::Debug + Send + Sync;
     type FindKeyValuesByPrefix: fmt::Debug + Send + Sync;
+    type FindKeysInInterval: fmt::Debug + Send + Sync;
+    type FindKeyValuesInInterval: fmt::Debug + Send + Sync;
 
     /// The current chain ID.
     fn chain_id(&mut self) -> Result<ChainId, ExecutionError>;
@@ -864,6 +869,18 @@ pub trait BaseRuntime {
         promise: &Self::FindKeysByPrefix,
     ) -> Result<Vec<Vec<u8>>, ExecutionError>;
 
+    /// Creates the promise to access keys in a specific interval.
+    fn find_keys_in_interval_new(
+        &mut self,
+        key_interval: KeyInterval,
+    ) -> Result<Self::FindKeysInInterval, ExecutionError>;
+
+    /// Resolves the promise to access keys in a specific interval.
+    fn find_keys_in_interval_wait(
+        &mut self,
+        promise: &Self::FindKeysInInterval,
+    ) -> Result<IntervalKeys, ExecutionError>;
+
     /// Reads the data from the key/values having a specific prefix.
     #[cfg(feature = "test")]
     #[expect(clippy::type_complexity)]
@@ -888,6 +905,18 @@ pub trait BaseRuntime {
         promise: &Self::FindKeyValuesByPrefix,
     ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ExecutionError>;
 
+    /// Creates the promise to access key/values in a specific interval.
+    fn find_key_values_in_interval_new(
+        &mut self,
+        key_interval: KeyInterval,
+    ) -> Result<Self::FindKeyValuesInInterval, ExecutionError>;
+
+    /// Resolves the promise to access key/values in a specific interval.
+    fn find_key_values_in_interval_wait(
+        &mut self,
+        promise: &Self::FindKeyValuesInInterval,
+    ) -> Result<IntervalKeyValues, ExecutionError>;
+
     /// Makes an HTTP request to the given URL and returns the answer, if any.
     fn perform_http_request(
         &mut self,

linera-execution/src/runtime.rs

@@ -22,7 +22,7 @@ use linera_base::{
     time::Instant,
     vm::VmRuntime,
 };
-use linera_views::batch::Batch;
+use linera_views::{batch::Batch, store::KeyInterval};
 use oneshot::Receiver;
 use tracing::instrument;
 
@@ -251,6 +251,8 @@ impl<T> QueryManager<T> {
 type Keys = Vec<Vec<u8>>;
 type Value = Vec<u8>;
 type KeyValues = Vec<(Vec<u8>, Vec<u8>)>;
+type IntervalKeys = (Vec<Vec<u8>>, bool);
+type IntervalKeyValues = (Vec<(Vec<u8>, Vec<u8>)>, bool);
 
 #[derive(Debug, Default)]
 struct ViewUserState {
@@ -266,6 +268,10 @@ struct ViewUserState {
     find_keys_queries: QueryManager<Keys>,
     /// The find-key-values queries in progress.
     find_key_values_queries: QueryManager<KeyValues>,
+    /// The find-keys-in-interval queries in progress.
+    find_keys_in_interval_queries: QueryManager<IntervalKeys>,
+    /// The find-key-values-in-interval queries in progress.
+    find_key_values_in_interval_queries: QueryManager<IntervalKeyValues>,
 }
 
 impl ViewUserState {
@@ -276,6 +282,8 @@ impl ViewUserState {
         self.read_multi_values_queries.force_all()?;
         self.find_keys_queries.force_all()?;
         self.find_key_values_queries.force_all()?;
+        self.find_keys_in_interval_queries.force_all()?;
+        self.find_key_values_in_interval_queries.force_all()?;
         Ok(())
     }
 }
@@ -599,6 +607,8 @@ where
     type ReadMultiValuesBytes = u32;
     type FindKeysByPrefix = u32;
     type FindKeyValuesByPrefix = u32;
+    type FindKeysInInterval = u32;
+    type FindKeyValuesInInterval = u32;
 
     fn chain_id(&mut self) -> Result<ChainId, ExecutionError> {
         let mut this = self.inner();
@@ -930,6 +940,77 @@ where
         Ok(key_values)
     }
 
+    fn find_keys_in_interval_new(
+        &mut self,
+        key_interval: KeyInterval,
+    ) -> Result<Self::FindKeysInInterval, ExecutionError> {
+        let mut this = self.inner();
+        let id = this.current_application().id;
+        this.resource_controller.track_read_operation()?;
+        let receiver = this.execution_state_sender.send_request(move |callback| {
+            ExecutionRequest::FindKeysInInterval {
+                id,
+                key_interval,
+                callback,
+            }
+        })?;
+        let state = this.view_user_states.entry(id).or_default();
+        state.find_keys_in_interval_queries.register(receiver)
+    }
+
+    fn find_keys_in_interval_wait(
+        &mut self,
+        promise: &Self::FindKeysInInterval,
+    ) -> Result<(Vec<Vec<u8>>, bool), ExecutionError> {
+        let mut this = self.inner();
+        let id = this.current_application().id;
+        let (keys, is_finished) = {
+            let state = this.view_user_states.entry(id).or_default();
+            state.find_keys_in_interval_queries.wait(*promise)?
+        };
+        let read_size = keys.iter().map(Vec::len).sum::<usize>();
+        this.resource_controller
+            .track_bytes_read(read_size as u64)?;
+        Ok((keys, is_finished))
+    }
+
+    fn find_key_values_in_interval_new(
+        &mut self,
+        key_interval: KeyInterval,
+    ) -> Result<Self::FindKeyValuesInInterval, ExecutionError> {
+        let mut this = self.inner();
+        let id = this.current_application().id;
+        this.resource_controller.track_read_operation()?;
+        let receiver = this.execution_state_sender.send_request(move |callback| {
+            ExecutionRequest::FindKeyValuesInInterval {
+                id,
+                key_interval,
+                callback,
+            }
+        })?;
+        let state = this.view_user_states.entry(id).or_default();
+        state.find_key_values_in_interval_queries.register(receiver)
+    }
+
+    fn find_key_values_in_interval_wait(
+        &mut self,
+        promise: &Self::FindKeyValuesInInterval,
+    ) -> Result<(Vec<(Vec<u8>, Vec<u8>)>, bool), ExecutionError> {
+        let mut this = self.inner();
+        let id = this.current_application().id;
+        let (key_values, is_finished) = {
+            let state = this.view_user_states.entry(id).or_default();
+            state.find_key_values_in_interval_queries.wait(*promise)?
+        };
+        let read_size = key_values
+            .iter()
+            .map(|(key, value)| key.len() + value.len())
+            .sum::<usize>();
+        this.resource_controller
+            .track_bytes_read(read_size as u64)?;
+        Ok((key_values, is_finished))
+    }
+
     fn perform_http_request(
         &mut self,
         request: http::Request,