linera_views/views/
key_value_store_view.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! We implement two types:
5//! 1) The first type `KeyValueStoreView` implements View and the function of `KeyValueStore`.
6//!
7//! 2) The second type `ViewContainer` encapsulates `KeyValueStoreView` and provides the following functionalities:
8//!    * The `Clone` trait
9//!    * a `write_batch` that takes a `&self` instead of a `&mut self`
10//!    * a `write_batch` that writes in the context instead of writing of the view.
11//!
12//! Currently, that second type is only used for tests.
13//!
14//! Key tags to create the sub-keys of a `KeyValueStoreView` on top of the base key.
15
16use std::{collections::BTreeMap, fmt::Debug, ops::Bound::Included, sync::Mutex};
17
18use allocative::Allocative;
19#[cfg(with_metrics)]
20use linera_base::prometheus_util::MeasureLatency as _;
21use linera_base::{ensure, visit_allocative_simple};
22
23use crate::{
24    batch::{Batch, WriteOperation},
25    common::{
26        from_bytes_option, get_key_range_for_prefix, get_upper_bound, DeletionSet, HasherOutput,
27        SuffixClosedSetIterator, Update,
28    },
29    context::Context,
30    hashable_wrapper::WrappedHashableContainerView,
31    historical_hash_wrapper::HistoricallyHashableView,
32    store::ReadableKeyValueStore,
33    views::{ClonableView, HashableView, Hasher, ReplaceContext, View, ViewError, MIN_VIEW_TAG},
34};
35
36#[cfg(with_metrics)]
37pub(crate) mod metrics {
38    use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram_vec};
39    use prometheus::HistogramVec;
40
41    linera_base::declare_metrics! {
42        /// The latency of hash computation
43        pub static KEY_VALUE_STORE_VIEW_HASH_LATENCY: HistogramVec =
44            register_histogram_vec(
45                "key_value_store_view_hash_latency",
46                "KeyValueStoreView hash latency",
47                &[],
48                exponential_bucket_latencies(5.0),
49            );
50
51        /// The latency of get operation
52        pub static KEY_VALUE_STORE_VIEW_GET_LATENCY: HistogramVec =
53            register_histogram_vec(
54                "key_value_store_view_get_latency",
55                "KeyValueStoreView get latency",
56                &[],
57                exponential_bucket_latencies(5.0),
58            );
59
60        /// The latency of multi get
61        pub static KEY_VALUE_STORE_VIEW_MULTI_GET_LATENCY: HistogramVec =
62            register_histogram_vec(
63                "key_value_store_view_multi_get_latency",
64                "KeyValueStoreView multi get latency",
65                &[],
66                exponential_bucket_latencies(5.0),
67            );
68
69        /// The latency of contains key
70        pub static KEY_VALUE_STORE_VIEW_CONTAINS_KEY_LATENCY: HistogramVec =
71            register_histogram_vec(
72                "key_value_store_view_contains_key_latency",
73                "KeyValueStoreView contains key latency",
74                &[],
75                exponential_bucket_latencies(5.0),
76            );
77
78        /// The latency of contains keys
79        pub static KEY_VALUE_STORE_VIEW_CONTAINS_KEYS_LATENCY: HistogramVec =
80            register_histogram_vec(
81                "key_value_store_view_contains_keys_latency",
82                "KeyValueStoreView contains keys latency",
83                &[],
84                exponential_bucket_latencies(5.0),
85            );
86
87        /// The latency of find keys by prefix operation
88        pub static KEY_VALUE_STORE_VIEW_FIND_KEYS_BY_PREFIX_LATENCY: HistogramVec =
89            register_histogram_vec(
90                "key_value_store_view_find_keys_by_prefix_latency",
91                "KeyValueStoreView find keys by prefix latency",
92                &[],
93                exponential_bucket_latencies(5.0),
94            );
95
96        /// The latency of find key values by prefix operation
97        pub static KEY_VALUE_STORE_VIEW_FIND_KEY_VALUES_BY_PREFIX_LATENCY: HistogramVec =
98            register_histogram_vec(
99                "key_value_store_view_find_key_values_by_prefix_latency",
100                "KeyValueStoreView find key values by prefix latency",
101                &[],
102                exponential_bucket_latencies(5.0),
103            );
104
105        /// The latency of write batch operation
106        pub static KEY_VALUE_STORE_VIEW_WRITE_BATCH_LATENCY: HistogramVec =
107            register_histogram_vec(
108                "key_value_store_view_write_batch_latency",
109                "KeyValueStoreView write batch latency",
110                &[],
111                exponential_bucket_latencies(5.0),
112            );
113    }
114}
115
116#[cfg(with_testing)]
117use {
118    crate::store::{KeyValueStoreError, WithError, WritableKeyValueStore},
119    async_lock::RwLock,
120    std::sync::Arc,
121    thiserror::Error,
122};
123
124#[repr(u8)]
125enum KeyTag {
126    /// Prefix for the indices of the view.
127    Index = MIN_VIEW_TAG,
128    /// Prefix for the hash.
129    Hash,
130}
131
132/// A view that represents the functions of `KeyValueStore`.
133///
134/// Comment on the data set:
135/// In order to work, the view needs to store the updates and deleted prefixes.
136/// The updates and deleted prefixes have to be coherent. This means:
137/// * If an index is deleted by one in deleted prefixes then it should not be present
138///   in the updates at all.
139/// * [`DeletePrefix::key_prefix`][entry1] should not dominate anyone. That is if we have `[0,2]`
140///   then we should not have `[0,2,3]` since it would be dominated by the preceding.
141///
142/// With that we have:
143/// * in order to test if an `index` is deleted by a prefix we compute the highest deleted prefix `dp`
144///   such that `dp <= index`.
145///   If `dp` is indeed a prefix then we conclude that `index` is deleted, otherwise not.
146///   The no domination is essential here.
147///
148/// [entry1]: crate::batch::WriteOperation::DeletePrefix
149#[derive(Debug, Allocative)]
150#[allocative(bound = "C")]
151pub struct KeyValueStoreView<C> {
152    /// The view context.
153    #[allocative(skip)]
154    context: C,
155    /// Tracks deleted key prefixes.
156    deletion_set: DeletionSet,
157    /// Pending changes not yet persisted to storage.
158    updates: BTreeMap<Vec<u8>, Update<Vec<u8>>>,
159    /// The hash persisted in storage.
160    #[allocative(visit = visit_allocative_simple)]
161    stored_hash: Option<HasherOutput>,
162    /// Memoized hash, if any.
163    #[allocative(visit = visit_allocative_simple)]
164    hash: Mutex<Option<HasherOutput>>,
165}
166
167impl<C: Context, C2: Context> ReplaceContext<C2> for KeyValueStoreView<C> {
168    type Target = KeyValueStoreView<C2>;
169
170    async fn with_context(
171        &mut self,
172        ctx: impl FnOnce(&Self::Context) -> C2 + Clone,
173    ) -> Self::Target {
174        let hash = *self.hash.lock().unwrap();
175        KeyValueStoreView {
176            context: ctx.clone()(&self.context),
177            deletion_set: self.deletion_set.clone(),
178            updates: self.updates.clone(),
179            stored_hash: self.stored_hash,
180            hash: Mutex::new(hash),
181        }
182    }
183}
184
185impl<C: Context> View for KeyValueStoreView<C> {
186    const NUM_INIT_KEYS: usize = 1;
187
188    type Context = C;
189
190    fn context(&self) -> C {
191        self.context.clone()
192    }
193
194    fn pre_load(context: &C) -> Result<Vec<Vec<u8>>, ViewError> {
195        let key_hash = context.base_key().base_tag(KeyTag::Hash as u8);
196        Ok(vec![key_hash])
197    }
198
199    fn post_load(context: C, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
200        let hash = from_bytes_option(values.first().ok_or(ViewError::PostLoadValuesError)?)?;
201        Ok(Self {
202            context,
203            deletion_set: DeletionSet::new(),
204            updates: BTreeMap::new(),
205            stored_hash: hash,
206            hash: Mutex::new(hash),
207        })
208    }
209
210    fn rollback(&mut self) {
211        self.deletion_set.rollback();
212        self.updates.clear();
213        *self.hash.get_mut().unwrap() = self.stored_hash;
214    }
215
216    async fn has_pending_changes(&self) -> bool {
217        if self.deletion_set.has_pending_changes() {
218            return true;
219        }
220        if !self.updates.is_empty() {
221            return true;
222        }
223        let hash = self.hash.lock().unwrap();
224        self.stored_hash != *hash
225    }
226
227    fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
228        let mut delete_view = false;
229        if self.deletion_set.delete_storage_first {
230            delete_view = true;
231            batch.delete_key_prefix(self.context.base_key().bytes.clone());
232            for (index, update) in self.updates.iter() {
233                if let Update::Set(value) = update {
234                    let key = self
235                        .context
236                        .base_key()
237                        .base_tag_index(KeyTag::Index as u8, index);
238                    batch.put_key_value_bytes(key, value.clone());
239                    delete_view = false;
240                }
241            }
242        } else {
243            for index in self.deletion_set.deleted_prefixes.iter() {
244                let key = self
245                    .context
246                    .base_key()
247                    .base_tag_index(KeyTag::Index as u8, index);
248                batch.delete_key_prefix(key);
249            }
250            for (index, update) in self.updates.iter() {
251                let key = self
252                    .context
253                    .base_key()
254                    .base_tag_index(KeyTag::Index as u8, index);
255                match update {
256                    Update::Removed => batch.delete_key(key),
257                    Update::Set(value) => batch.put_key_value_bytes(key, value.clone()),
258                }
259            }
260        }
261        let hash = *self.hash.lock().unwrap();
262        if self.stored_hash != hash {
263            let key = self.context.base_key().base_tag(KeyTag::Hash as u8);
264            match hash {
265                None => batch.delete_key(key),
266                Some(hash) => batch.put_key_value(key, &hash)?,
267            }
268        }
269        Ok(delete_view)
270    }
271
272    fn post_save(&mut self) {
273        self.deletion_set.delete_storage_first = false;
274        self.deletion_set.deleted_prefixes.clear();
275        self.updates.clear();
276        let hash = *self.hash.lock().unwrap();
277        self.stored_hash = hash;
278    }
279
280    fn clear(&mut self) {
281        self.deletion_set.clear();
282        self.updates.clear();
283        *self.hash.get_mut().unwrap() = None;
284    }
285}
286
287impl<C: Context> ClonableView for KeyValueStoreView<C> {
288    fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
289        Ok(KeyValueStoreView {
290            context: self.context.clone(),
291            deletion_set: self.deletion_set.clone(),
292            updates: self.updates.clone(),
293            stored_hash: self.stored_hash,
294            hash: Mutex::new(*self.hash.get_mut().unwrap()),
295        })
296    }
297}
298
299impl<C: Context> KeyValueStoreView<C> {
300    fn max_key_size(&self) -> usize {
301        let prefix_len = self.context.base_key().bytes.len();
302        <C::Store as ReadableKeyValueStore>::MAX_KEY_SIZE - 1 - prefix_len
303    }
304
305    /// Applies the function f over all indices. If the function f returns
306    /// false, then the loop ends prematurely.
307    /// ```rust
308    /// # tokio_test::block_on(async {
309    /// # use linera_views::context::MemoryContext;
310    /// # use linera_views::key_value_store_view::KeyValueStoreView;
311    /// # use linera_views::views::View;
312    /// # let context = MemoryContext::new_for_testing(());
313    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
314    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
315    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
316    /// view.insert(vec![0, 3], vec![0]).await.unwrap();
317    /// let mut count = 0;
318    /// view.for_each_index_while(|_key| {
319    ///     count += 1;
320    ///     Ok(count < 2)
321    /// })
322    /// .await
323    /// .unwrap();
324    /// assert_eq!(count, 2);
325    /// # })
326    /// ```
327    pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
328    where
329        F: FnMut(&[u8]) -> Result<bool, ViewError> + Send,
330    {
331        let key_prefix = self.context.base_key().base_tag(KeyTag::Index as u8);
332        let mut updates = self.updates.iter();
333        let mut update = updates.next();
334        if !self.deletion_set.delete_storage_first {
335            let mut suffix_closed_set =
336                SuffixClosedSetIterator::new(0, self.deletion_set.deleted_prefixes.iter());
337            for index in self
338                .context
339                .store()
340                .find_keys_by_prefix(&key_prefix)
341                .await?
342            {
343                loop {
344                    match update {
345                        Some((key, value)) if key <= &index => {
346                            if let Update::Set(_) = value {
347                                if !f(key)? {
348                                    return Ok(());
349                                }
350                            }
351                            update = updates.next();
352                            if key == &index {
353                                break;
354                            }
355                        }
356                        _ => {
357                            if !suffix_closed_set.find_key(&index) && !f(&index)? {
358                                return Ok(());
359                            }
360                            break;
361                        }
362                    }
363                }
364            }
365        }
366        while let Some((key, value)) = update {
367            if let Update::Set(_) = value {
368                if !f(key)? {
369                    return Ok(());
370                }
371            }
372            update = updates.next();
373        }
374        Ok(())
375    }
376
377    /// Applies the function f over all indices.
378    /// ```rust
379    /// # tokio_test::block_on(async {
380    /// # use linera_views::context::MemoryContext;
381    /// # use linera_views::key_value_store_view::KeyValueStoreView;
382    /// # use linera_views::views::View;
383    /// # let context = MemoryContext::new_for_testing(());
384    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
385    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
386    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
387    /// view.insert(vec![0, 3], vec![0]).await.unwrap();
388    /// let mut count = 0;
389    /// view.for_each_index(|_key| {
390    ///     count += 1;
391    ///     Ok(())
392    /// })
393    /// .await
394    /// .unwrap();
395    /// assert_eq!(count, 3);
396    /// # })
397    /// ```
398    pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
399    where
400        F: FnMut(&[u8]) -> Result<(), ViewError> + Send,
401    {
402        self.for_each_index_while(|key| {
403            f(key)?;
404            Ok(true)
405        })
406        .await
407    }
408
409    /// Applies the function f over all index/value pairs.
410    /// If the function f returns false then the loop ends prematurely.
411    /// ```rust
412    /// # tokio_test::block_on(async {
413    /// # use linera_views::context::MemoryContext;
414    /// # use linera_views::key_value_store_view::KeyValueStoreView;
415    /// # use linera_views::views::View;
416    /// # let context = MemoryContext::new_for_testing(());
417    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
418    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
419    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
420    /// let mut values = Vec::new();
421    /// view.for_each_index_value_while(|_key, value| {
422    ///     values.push(value.to_vec());
423    ///     Ok(values.len() < 1)
424    /// })
425    /// .await
426    /// .unwrap();
427    /// assert_eq!(values, vec![vec![0]]);
428    /// # })
429    /// ```
430    pub async fn for_each_index_value_while<F>(&self, mut f: F) -> Result<(), ViewError>
431    where
432        F: FnMut(&[u8], &[u8]) -> Result<bool, ViewError> + Send,
433    {
434        let key_prefix = self.context.base_key().base_tag(KeyTag::Index as u8);
435        let mut updates = self.updates.iter();
436        let mut update = updates.next();
437        if !self.deletion_set.delete_storage_first {
438            let mut suffix_closed_set =
439                SuffixClosedSetIterator::new(0, self.deletion_set.deleted_prefixes.iter());
440            for entry in self
441                .context
442                .store()
443                .find_key_values_by_prefix(&key_prefix)
444                .await?
445            {
446                let (index, index_val) = entry;
447                loop {
448                    match update {
449                        Some((key, value)) if key <= &index => {
450                            if let Update::Set(value) = value {
451                                if !f(key, value)? {
452                                    return Ok(());
453                                }
454                            }
455                            update = updates.next();
456                            if key == &index {
457                                break;
458                            }
459                        }
460                        _ => {
461                            if !suffix_closed_set.find_key(&index) && !f(&index, &index_val)? {
462                                return Ok(());
463                            }
464                            break;
465                        }
466                    }
467                }
468            }
469        }
470        while let Some((key, value)) = update {
471            if let Update::Set(value) = value {
472                if !f(key, value)? {
473                    return Ok(());
474                }
475            }
476            update = updates.next();
477        }
478        Ok(())
479    }
480
481    /// Applies the function f over all index/value pairs.
482    /// ```rust
483    /// # tokio_test::block_on(async {
484    /// # use linera_views::context::MemoryContext;
485    /// # use linera_views::key_value_store_view::KeyValueStoreView;
486    /// # use linera_views::views::View;
487    /// # let context = MemoryContext::new_for_testing(());
488    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
489    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
490    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
491    /// let mut part_keys = Vec::new();
492    /// view.for_each_index_while(|key| {
493    ///     part_keys.push(key.to_vec());
494    ///     Ok(part_keys.len() < 1)
495    /// })
496    /// .await
497    /// .unwrap();
498    /// assert_eq!(part_keys, vec![vec![0, 1]]);
499    /// # })
500    /// ```
501    pub async fn for_each_index_value<F>(&self, mut f: F) -> Result<(), ViewError>
502    where
503        F: FnMut(&[u8], &[u8]) -> Result<(), ViewError> + Send,
504    {
505        self.for_each_index_value_while(|key, value| {
506            f(key, value)?;
507            Ok(true)
508        })
509        .await
510    }
511
512    /// Returns the list of indices in lexicographic order.
513    /// ```rust
514    /// # tokio_test::block_on(async {
515    /// # use linera_views::context::MemoryContext;
516    /// # use linera_views::key_value_store_view::KeyValueStoreView;
517    /// # use linera_views::views::View;
518    /// # let context = MemoryContext::new_for_testing(());
519    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
520    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
521    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
522    /// let indices = view.indices().await.unwrap();
523    /// assert_eq!(indices, vec![vec![0, 1], vec![0, 2]]);
524    /// # })
525    /// ```
526    pub async fn indices(&self) -> Result<Vec<Vec<u8>>, ViewError> {
527        let mut indices = Vec::new();
528        self.for_each_index(|index| {
529            indices.push(index.to_vec());
530            Ok(())
531        })
532        .await?;
533        Ok(indices)
534    }
535
536    /// Returns the list of indices and values in lexicographic order.
537    /// ```rust
538    /// # tokio_test::block_on(async {
539    /// # use linera_views::context::MemoryContext;
540    /// # use linera_views::key_value_store_view::KeyValueStoreView;
541    /// # use linera_views::views::View;
542    /// # let context = MemoryContext::new_for_testing(());
543    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
544    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
545    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
546    /// let key_values = view.indices().await.unwrap();
547    /// assert_eq!(key_values, vec![vec![0, 1], vec![0, 2]]);
548    /// # })
549    /// ```
550    pub async fn index_values(&self) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ViewError> {
551        let mut index_values = Vec::new();
552        self.for_each_index_value(|index, value| {
553            index_values.push((index.to_vec(), value.to_vec()));
554            Ok(())
555        })
556        .await?;
557        Ok(index_values)
558    }
559
560    /// Returns the number of entries.
561    /// ```rust
562    /// # tokio_test::block_on(async {
563    /// # use linera_views::context::MemoryContext;
564    /// # use linera_views::key_value_store_view::KeyValueStoreView;
565    /// # use linera_views::views::View;
566    /// # let context = MemoryContext::new_for_testing(());
567    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
568    /// view.insert(vec![0, 1], vec![0]).await.unwrap();
569    /// view.insert(vec![0, 2], vec![0]).await.unwrap();
570    /// let count = view.iterative_count().await.unwrap();
571    /// assert_eq!(count, 2);
572    /// # })
573    /// ```
574    pub async fn iterative_count(&self) -> Result<usize, ViewError> {
575        let mut count = 0;
576        self.for_each_index(|_index| {
577            count += 1;
578            Ok(())
579        })
580        .await?;
581        Ok(count)
582    }
583
584    /// Obtains the value at the given index, if any.
585    /// ```rust
586    /// # tokio_test::block_on(async {
587    /// # use linera_views::context::MemoryContext;
588    /// # use linera_views::key_value_store_view::KeyValueStoreView;
589    /// # use linera_views::views::View;
590    /// # let context = MemoryContext::new_for_testing(());
591    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
592    /// view.insert(vec![0, 1], vec![42]).await.unwrap();
593    /// assert_eq!(view.get(&[0, 1]).await.unwrap(), Some(vec![42]));
594    /// assert_eq!(view.get(&[0, 2]).await.unwrap(), None);
595    /// # })
596    /// ```
597    pub async fn get(&self, index: &[u8]) -> Result<Option<Vec<u8>>, ViewError> {
598        #[cfg(with_metrics)]
599        let _latency = metrics::KEY_VALUE_STORE_VIEW_GET_LATENCY.measure_latency();
600        ensure!(index.len() <= self.max_key_size(), ViewError::KeyTooLong);
601        if let Some(update) = self.updates.get(index) {
602            let value = match update {
603                Update::Removed => None,
604                Update::Set(value) => Some(value.clone()),
605            };
606            return Ok(value);
607        }
608        if self.deletion_set.contains_prefix_of(index) {
609            return Ok(None);
610        }
611        let key = self
612            .context
613            .base_key()
614            .base_tag_index(KeyTag::Index as u8, index);
615        Ok(self.context.store().read_value_bytes(&key).await?)
616    }
617
618    /// Tests whether the store contains a specific index.
619    /// ```rust
620    /// # tokio_test::block_on(async {
621    /// # use linera_views::context::MemoryContext;
622    /// # use linera_views::key_value_store_view::KeyValueStoreView;
623    /// # use linera_views::views::View;
624    /// # let context = MemoryContext::new_for_testing(());
625    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
626    /// view.insert(vec![0, 1], vec![42]).await.unwrap();
627    /// assert!(view.contains_key(&[0, 1]).await.unwrap());
628    /// assert!(!view.contains_key(&[0, 2]).await.unwrap());
629    /// # })
630    /// ```
631    pub async fn contains_key(&self, index: &[u8]) -> Result<bool, ViewError> {
632        #[cfg(with_metrics)]
633        let _latency = metrics::KEY_VALUE_STORE_VIEW_CONTAINS_KEY_LATENCY.measure_latency();
634        ensure!(index.len() <= self.max_key_size(), ViewError::KeyTooLong);
635        if let Some(update) = self.updates.get(index) {
636            let test = match update {
637                Update::Removed => false,
638                Update::Set(_value) => true,
639            };
640            return Ok(test);
641        }
642        if self.deletion_set.contains_prefix_of(index) {
643            return Ok(false);
644        }
645        let key = self
646            .context
647            .base_key()
648            .base_tag_index(KeyTag::Index as u8, index);
649        Ok(self.context.store().contains_key(&key).await?)
650    }
651
652    /// Tests whether the view contains a range of indices
653    /// ```rust
654    /// # tokio_test::block_on(async {
655    /// # use linera_views::context::MemoryContext;
656    /// # use linera_views::key_value_store_view::KeyValueStoreView;
657    /// # use linera_views::views::View;
658    /// # let context = MemoryContext::new_for_testing(());
659    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
660    /// view.insert(vec![0, 1], vec![42]).await.unwrap();
661    /// let keys = vec![vec![0, 1], vec![0, 2]];
662    /// let results = view.contains_keys(&keys).await.unwrap();
663    /// assert_eq!(results, vec![true, false]);
664    /// # })
665    /// ```
666    pub async fn contains_keys(&self, indices: &[Vec<u8>]) -> Result<Vec<bool>, ViewError> {
667        #[cfg(with_metrics)]
668        let _latency = metrics::KEY_VALUE_STORE_VIEW_CONTAINS_KEYS_LATENCY.measure_latency();
669        let mut results = Vec::with_capacity(indices.len());
670        let mut missed_indices = Vec::new();
671        let mut vector_query = Vec::new();
672        for (i, index) in indices.iter().enumerate() {
673            ensure!(index.len() <= self.max_key_size(), ViewError::KeyTooLong);
674            if let Some(update) = self.updates.get(index) {
675                let value = match update {
676                    Update::Removed => false,
677                    Update::Set(_) => true,
678                };
679                results.push(value);
680            } else {
681                results.push(false);
682                if !self.deletion_set.contains_prefix_of(index) {
683                    missed_indices.push(i);
684                    let key = self
685                        .context
686                        .base_key()
687                        .base_tag_index(KeyTag::Index as u8, index);
688                    vector_query.push(key);
689                }
690            }
691        }
692        let values = self.context.store().contains_keys(&vector_query).await?;
693        for (i, value) in missed_indices.into_iter().zip(values) {
694            results[i] = value;
695        }
696        Ok(results)
697    }
698
699    /// Obtains the values of a range of indices
700    /// ```rust
701    /// # tokio_test::block_on(async {
702    /// # use linera_views::context::MemoryContext;
703    /// # use linera_views::key_value_store_view::KeyValueStoreView;
704    /// # use linera_views::views::View;
705    /// # let context = MemoryContext::new_for_testing(());
706    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
707    /// view.insert(vec![0, 1], vec![42]).await.unwrap();
708    /// assert_eq!(
709    ///     view.multi_get(&[vec![0, 1], vec![0, 2]]).await.unwrap(),
710    ///     vec![Some(vec![42]), None]
711    /// );
712    /// # })
713    /// ```
714    pub async fn multi_get(&self, indices: &[Vec<u8>]) -> Result<Vec<Option<Vec<u8>>>, ViewError> {
715        #[cfg(with_metrics)]
716        let _latency = metrics::KEY_VALUE_STORE_VIEW_MULTI_GET_LATENCY.measure_latency();
717        let mut result = Vec::with_capacity(indices.len());
718        let mut missed_indices = Vec::new();
719        let mut vector_query = Vec::new();
720        for (i, index) in indices.iter().enumerate() {
721            ensure!(index.len() <= self.max_key_size(), ViewError::KeyTooLong);
722            if let Some(update) = self.updates.get(index) {
723                let value = match update {
724                    Update::Removed => None,
725                    Update::Set(value) => Some(value.clone()),
726                };
727                result.push(value);
728            } else {
729                result.push(None);
730                if !self.deletion_set.contains_prefix_of(index) {
731                    missed_indices.push(i);
732                    let key = self
733                        .context
734                        .base_key()
735                        .base_tag_index(KeyTag::Index as u8, index);
736                    vector_query.push(key);
737                }
738            }
739        }
740        let values = self
741            .context
742            .store()
743            .read_multi_values_bytes(&vector_query)
744            .await?;
745        for (i, value) in missed_indices.into_iter().zip(values) {
746            result[i] = value;
747        }
748        Ok(result)
749    }
750
751    /// Applies the given batch of `crate::common::WriteOperation`.
752    /// ```rust
753    /// # tokio_test::block_on(async {
754    /// # use linera_views::context::MemoryContext;
755    /// # use linera_views::key_value_store_view::KeyValueStoreView;
756    /// # use linera_views::batch::Batch;
757    /// # use linera_views::views::View;
758    /// # let context = MemoryContext::new_for_testing(());
759    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
760    /// view.insert(vec![0, 1], vec![34]).await.unwrap();
761    /// view.insert(vec![3, 4], vec![42]).await.unwrap();
762    /// let mut batch = Batch::new();
763    /// batch.delete_key_prefix(vec![0]);
764    /// view.write_batch(batch).unwrap();
765    /// let key_values = view.find_key_values_by_prefix(&[0]).await.unwrap();
766    /// assert_eq!(key_values, vec![]);
767    /// # })
768    /// ```
769    pub fn write_batch(&mut self, batch: Batch) -> Result<(), ViewError> {
770        #[cfg(with_metrics)]
771        let _latency = metrics::KEY_VALUE_STORE_VIEW_WRITE_BATCH_LATENCY.measure_latency();
772        *self.hash.get_mut().unwrap() = None;
773        let max_key_size = self.max_key_size();
774        for operation in batch.operations {
775            match operation {
776                WriteOperation::Delete { key } => {
777                    ensure!(key.len() <= max_key_size, ViewError::KeyTooLong);
778                    if self.deletion_set.contains_prefix_of(&key) {
779                        // Optimization: No need to mark `short_key` for deletion as we are going to remove all the keys at once.
780                        self.updates.remove(&key);
781                    } else {
782                        self.updates.insert(key, Update::Removed);
783                    }
784                }
785                WriteOperation::Put { key, value } => {
786                    ensure!(key.len() <= max_key_size, ViewError::KeyTooLong);
787                    self.updates.insert(key, Update::Set(value));
788                }
789                WriteOperation::DeletePrefix { key_prefix } => {
790                    ensure!(key_prefix.len() <= max_key_size, ViewError::KeyTooLong);
791                    let key_list = self
792                        .updates
793                        .range(get_key_range_for_prefix(key_prefix.clone()))
794                        .map(|x| x.0.to_vec())
795                        .collect::<Vec<_>>();
796                    for key in key_list {
797                        self.updates.remove(&key);
798                    }
799                    self.deletion_set.insert_key_prefix(key_prefix);
800                }
801            }
802        }
803        Ok(())
804    }
805
806    /// Sets or inserts a value.
807    /// ```rust
808    /// # tokio_test::block_on(async {
809    /// # use linera_views::context::MemoryContext;
810    /// # use linera_views::key_value_store_view::KeyValueStoreView;
811    /// # use linera_views::views::View;
812    /// # let context = MemoryContext::new_for_testing(());
813    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
814    /// view.insert(vec![0, 1], vec![34]).await.unwrap();
815    /// assert_eq!(view.get(&[0, 1]).await.unwrap(), Some(vec![34]));
816    /// # })
817    /// ```
818    pub async fn insert(&mut self, index: Vec<u8>, value: Vec<u8>) -> Result<(), ViewError> {
819        let mut batch = Batch::new();
820        batch.put_key_value_bytes(index, value);
821        self.write_batch(batch)
822    }
823
824    /// Removes a value. If absent then the action has no effect.
825    /// ```rust
826    /// # tokio_test::block_on(async {
827    /// # use linera_views::context::MemoryContext;
828    /// # use linera_views::key_value_store_view::KeyValueStoreView;
829    /// # use linera_views::views::View;
830    /// # let context = MemoryContext::new_for_testing(());
831    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
832    /// view.insert(vec![0, 1], vec![34]).await.unwrap();
833    /// view.remove(vec![0, 1]).await.unwrap();
834    /// assert_eq!(view.get(&[0, 1]).await.unwrap(), None);
835    /// # })
836    /// ```
837    pub async fn remove(&mut self, index: Vec<u8>) -> Result<(), ViewError> {
838        let mut batch = Batch::new();
839        batch.delete_key(index);
840        self.write_batch(batch)
841    }
842
843    /// Deletes a key prefix.
844    /// ```rust
845    /// # tokio_test::block_on(async {
846    /// # use linera_views::context::MemoryContext;
847    /// # use linera_views::key_value_store_view::KeyValueStoreView;
848    /// # use linera_views::views::View;
849    /// # let context = MemoryContext::new_for_testing(());
850    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
851    /// view.insert(vec![0, 1], vec![34]).await.unwrap();
852    /// view.remove_by_prefix(vec![0]).await.unwrap();
853    /// assert_eq!(view.get(&[0, 1]).await.unwrap(), None);
854    /// # })
855    /// ```
856    pub async fn remove_by_prefix(&mut self, key_prefix: Vec<u8>) -> Result<(), ViewError> {
857        let mut batch = Batch::new();
858        batch.delete_key_prefix(key_prefix);
859        self.write_batch(batch)
860    }
861
862    /// Iterates over all the keys matching the given prefix. The prefix is not included in the returned keys.
863    /// ```rust
864    /// # tokio_test::block_on(async {
865    /// # use linera_views::context::MemoryContext;
866    /// # use linera_views::key_value_store_view::KeyValueStoreView;
867    /// # use linera_views::views::View;
868    /// # let context = MemoryContext::new_for_testing(());
869    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
870    /// view.insert(vec![0, 1], vec![34]).await.unwrap();
871    /// view.insert(vec![3, 4], vec![42]).await.unwrap();
872    /// let keys = view.find_keys_by_prefix(&[0]).await.unwrap();
873    /// assert_eq!(keys, vec![vec![1]]);
874    /// # })
875    /// ```
876    pub async fn find_keys_by_prefix(&self, key_prefix: &[u8]) -> Result<Vec<Vec<u8>>, ViewError> {
877        #[cfg(with_metrics)]
878        let _latency = metrics::KEY_VALUE_STORE_VIEW_FIND_KEYS_BY_PREFIX_LATENCY.measure_latency();
879        ensure!(
880            key_prefix.len() <= self.max_key_size(),
881            ViewError::KeyTooLong
882        );
883        let len = key_prefix.len();
884        let key_prefix_full = self
885            .context
886            .base_key()
887            .base_tag_index(KeyTag::Index as u8, key_prefix);
888        let mut keys = Vec::new();
889        let key_prefix_upper = get_upper_bound(key_prefix);
890        let mut updates = self
891            .updates
892            .range((Included(key_prefix.to_vec()), key_prefix_upper));
893        let mut update = updates.next();
894        if !self.deletion_set.delete_storage_first {
895            let mut suffix_closed_set =
896                SuffixClosedSetIterator::new(0, self.deletion_set.deleted_prefixes.iter());
897            for key in self
898                .context
899                .store()
900                .find_keys_by_prefix(&key_prefix_full)
901                .await?
902            {
903                loop {
904                    match update {
905                        Some((update_key, update_value))
906                            if &update_key[len..] <= key.as_slice() =>
907                        {
908                            if let Update::Set(_) = update_value {
909                                keys.push(update_key[len..].to_vec());
910                            }
911                            update = updates.next();
912                            if update_key[len..] == key[..] {
913                                break;
914                            }
915                        }
916                        _ => {
917                            let mut key_with_prefix = key_prefix.to_vec();
918                            key_with_prefix.extend_from_slice(&key);
919                            if !suffix_closed_set.find_key(&key_with_prefix) {
920                                keys.push(key);
921                            }
922                            break;
923                        }
924                    }
925                }
926            }
927        }
928        while let Some((update_key, update_value)) = update {
929            if let Update::Set(_) = update_value {
930                let update_key = update_key[len..].to_vec();
931                keys.push(update_key);
932            }
933            update = updates.next();
934        }
935        Ok(keys)
936    }
937
938    /// Iterates over all the key-value pairs, for keys matching the given prefix. The
939    /// prefix is not included in the returned keys.
940    /// ```rust
941    /// # tokio_test::block_on(async {
942    /// # use linera_views::context::MemoryContext;
943    /// # use linera_views::key_value_store_view::KeyValueStoreView;
944    /// # use linera_views::views::View;
945    /// # let context = MemoryContext::new_for_testing(());
946    /// let mut view = KeyValueStoreView::load(context).await.unwrap();
947    /// view.insert(vec![0, 1], vec![34]).await.unwrap();
948    /// view.insert(vec![3, 4], vec![42]).await.unwrap();
949    /// let key_values = view.find_key_values_by_prefix(&[0]).await.unwrap();
950    /// assert_eq!(key_values, vec![(vec![1], vec![34])]);
951    /// # })
952    /// ```
953    pub async fn find_key_values_by_prefix(
954        &self,
955        key_prefix: &[u8],
956    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ViewError> {
957        #[cfg(with_metrics)]
958        let _latency =
959            metrics::KEY_VALUE_STORE_VIEW_FIND_KEY_VALUES_BY_PREFIX_LATENCY.measure_latency();
960        ensure!(
961            key_prefix.len() <= self.max_key_size(),
962            ViewError::KeyTooLong
963        );
964        let len = key_prefix.len();
965        let key_prefix_full = self
966            .context
967            .base_key()
968            .base_tag_index(KeyTag::Index as u8, key_prefix);
969        let mut key_values = Vec::new();
970        let key_prefix_upper = get_upper_bound(key_prefix);
971        let mut updates = self
972            .updates
973            .range((Included(key_prefix.to_vec()), key_prefix_upper));
974        let mut update = updates.next();
975        if !self.deletion_set.delete_storage_first {
976            let mut suffix_closed_set =
977                SuffixClosedSetIterator::new(0, self.deletion_set.deleted_prefixes.iter());
978            for entry in self
979                .context
980                .store()
981                .find_key_values_by_prefix(&key_prefix_full)
982                .await?
983            {
984                let (key, value) = entry;
985                loop {
986                    match update {
987                        Some((update_key, update_value)) if update_key[len..] <= key[..] => {
988                            if let Update::Set(update_value) = update_value {
989                                let key_value = (update_key[len..].to_vec(), update_value.to_vec());
990                                key_values.push(key_value);
991                            }
992                            update = updates.next();
993                            if update_key[len..] == key[..] {
994                                break;
995                            }
996                        }
997                        _ => {
998                            let mut key_with_prefix = key_prefix.to_vec();
999                            key_with_prefix.extend_from_slice(&key);
1000                            if !suffix_closed_set.find_key(&key_with_prefix) {
1001                                key_values.push((key, value));
1002                            }
1003                            break;
1004                        }
1005                    }
1006                }
1007            }
1008        }
1009        while let Some((update_key, update_value)) = update {
1010            if let Update::Set(update_value) = update_value {
1011                let key_value = (update_key[len..].to_vec(), update_value.to_vec());
1012                key_values.push(key_value);
1013            }
1014            update = updates.next();
1015        }
1016        Ok(key_values)
1017    }
1018
1019    async fn compute_hash(&self) -> Result<<sha3::Sha3_256 as Hasher>::Output, ViewError> {
1020        #[cfg(with_metrics)]
1021        let _hash_latency = metrics::KEY_VALUE_STORE_VIEW_HASH_LATENCY.measure_latency();
1022        let mut hasher = sha3::Sha3_256::default();
1023        let mut count = 0u32;
1024        self.for_each_index_value(|index, value| -> Result<(), ViewError> {
1025            count += 1;
1026            hasher.update_with_bytes(index)?;
1027            hasher.update_with_bytes(value)?;
1028            Ok(())
1029        })
1030        .await?;
1031        hasher.update_with_bcs_bytes(&count)?;
1032        Ok(hasher.finalize())
1033    }
1034}
1035
1036impl<C: Context> HashableView for KeyValueStoreView<C> {
1037    type Hasher = sha3::Sha3_256;
1038
1039    async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1040        let hash = *self.hash.get_mut().unwrap();
1041        match hash {
1042            Some(hash) => Ok(hash),
1043            None => {
1044                let new_hash = self.compute_hash().await?;
1045                let hash = self.hash.get_mut().unwrap();
1046                *hash = Some(new_hash);
1047                Ok(new_hash)
1048            }
1049        }
1050    }
1051
1052    async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1053        let hash = *self.hash.lock().unwrap();
1054        match hash {
1055            Some(hash) => Ok(hash),
1056            None => {
1057                let new_hash = self.compute_hash().await?;
1058                let mut hash = self.hash.lock().unwrap();
1059                *hash = Some(new_hash);
1060                Ok(new_hash)
1061            }
1062        }
1063    }
1064}
1065
1066/// Type wrapping `KeyValueStoreView` while memoizing the hash.
1067pub type HashedKeyValueStoreView<C> =
1068    WrappedHashableContainerView<C, KeyValueStoreView<C>, HasherOutput>;
1069
1070/// Wrapper around `KeyValueStoreView` to compute hashes based on the history of changes.
1071pub type HistoricallyHashedKeyValueStoreView<C> = HistoricallyHashableView<C, KeyValueStoreView<C>>;
1072
1073/// A virtual DB client using a `KeyValueStoreView` as a backend (testing only).
1074#[cfg(with_testing)]
1075#[derive(Debug, Clone)]
1076pub struct ViewContainer<C> {
1077    view: Arc<RwLock<KeyValueStoreView<C>>>,
1078}
1079
1080#[cfg(with_testing)]
1081impl<C> WithError for ViewContainer<C> {
1082    type Error = ViewContainerError;
1083}
1084
1085#[cfg(with_testing)]
1086/// The error type for [`ViewContainer`] operations.
1087#[derive(Error, Debug)]
1088pub enum ViewContainerError {
1089    /// View error.
1090    #[error(transparent)]
1091    ViewError(#[from] ViewError),
1092
1093    /// BCS serialization error.
1094    #[error(transparent)]
1095    BcsError(#[from] bcs::Error),
1096}
1097
1098#[cfg(with_testing)]
1099impl KeyValueStoreError for ViewContainerError {
1100    const BACKEND: &'static str = "view_container";
1101}
1102
1103#[cfg(with_testing)]
1104impl<C: Context> ReadableKeyValueStore for ViewContainer<C> {
1105    const MAX_KEY_SIZE: usize = <C::Store as ReadableKeyValueStore>::MAX_KEY_SIZE;
1106
1107    fn root_key(&self) -> Result<Vec<u8>, ViewContainerError> {
1108        Ok(Vec::new())
1109    }
1110
1111    async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, ViewContainerError> {
1112        let view = self.view.read().await;
1113        Ok(view.get(key).await?)
1114    }
1115
1116    async fn contains_key(&self, key: &[u8]) -> Result<bool, ViewContainerError> {
1117        let view = self.view.read().await;
1118        Ok(view.contains_key(key).await?)
1119    }
1120
1121    async fn contains_keys(&self, keys: &[Vec<u8>]) -> Result<Vec<bool>, ViewContainerError> {
1122        let view = self.view.read().await;
1123        Ok(view.contains_keys(keys).await?)
1124    }
1125
1126    async fn read_multi_values_bytes(
1127        &self,
1128        keys: &[Vec<u8>],
1129    ) -> Result<Vec<Option<Vec<u8>>>, ViewContainerError> {
1130        let view = self.view.read().await;
1131        Ok(view.multi_get(keys).await?)
1132    }
1133
1134    async fn find_keys_by_prefix(
1135        &self,
1136        key_prefix: &[u8],
1137    ) -> Result<Vec<Vec<u8>>, ViewContainerError> {
1138        let view = self.view.read().await;
1139        Ok(view.find_keys_by_prefix(key_prefix).await?)
1140    }
1141
1142    async fn find_key_values_by_prefix(
1143        &self,
1144        key_prefix: &[u8],
1145    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ViewContainerError> {
1146        let view = self.view.read().await;
1147        Ok(view.find_key_values_by_prefix(key_prefix).await?)
1148    }
1149}
1150
1151#[cfg(with_testing)]
1152impl<C: Context> WritableKeyValueStore for ViewContainer<C> {
1153    const MAX_VALUE_SIZE: usize = <C::Store as WritableKeyValueStore>::MAX_VALUE_SIZE;
1154
1155    async fn write_batch(&self, batch: Batch) -> Result<(), ViewContainerError> {
1156        let mut view = self.view.write().await;
1157        view.write_batch(batch)?;
1158        let mut batch = Batch::new();
1159        view.pre_save(&mut batch)?;
1160        view.post_save();
1161        view.context()
1162            .store()
1163            .write_batch(batch)
1164            .await
1165            .map_err(ViewError::from)?;
1166        Ok(())
1167    }
1168
1169    async fn clear_journal(&self) -> Result<(), ViewContainerError> {
1170        Ok(())
1171    }
1172}
1173
1174#[cfg(with_testing)]
1175impl<C: Context> ViewContainer<C> {
1176    /// Creates a [`ViewContainer`].
1177    pub async fn new(context: C) -> Result<Self, ViewError> {
1178        let view = KeyValueStoreView::load(context).await?;
1179        let view = Arc::new(RwLock::new(view));
1180        Ok(Self { view })
1181    }
1182}
linera_views/views/key_value_store_view.rs

linera_views/views/
key_value_store_view.rs
