linera_views/views/
log_view.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    collections::BTreeMap,
6    ops::{Bound, Range, RangeBounds},
7};
8
9use allocative::Allocative;
10use linera_base::data_types::ArithmeticError;
11#[cfg(with_metrics)]
12use linera_base::prometheus_util::MeasureLatency as _;
13use serde::{de::DeserializeOwned, Serialize};
14
15use crate::{
16    batch::Batch,
17    common::{from_bytes_option_or_default, HasherOutput},
18    context::Context,
19    hashable_wrapper::WrappedHashableContainerView,
20    historical_hash_wrapper::HistoricallyHashableView,
21    store::ReadableKeyValueStore as _,
22    views::{ClonableView, HashableView, Hasher, View, ViewError, MIN_VIEW_TAG},
23};
24
25#[cfg(with_metrics)]
26pub(crate) mod metrics {
27    use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram_vec};
28    use prometheus::HistogramVec;
29
30    linera_base::declare_metrics! {
31        /// The runtime of hash computation
32        pub static LOG_VIEW_HASH_RUNTIME: HistogramVec =
33            register_histogram_vec(
34                "log_view_hash_runtime",
35                "LogView hash runtime",
36                &[],
37                exponential_bucket_latencies(5.0),
38            );
39    }
40}
41
42/// Key tags to create the sub-keys of a `LogView` on top of the base key.
43#[repr(u8)]
44enum KeyTag {
45    /// Prefix for the storing of the variable `stored_count`.
46    Count = MIN_VIEW_TAG,
47    /// Prefix for the indices of the log.
48    Index,
49}
50
51/// A view that supports logging values of type `T`.
52#[derive(Debug, Allocative)]
53#[allocative(bound = "C, T: Allocative")]
54pub struct LogView<C, T> {
55    /// The view context.
56    #[allocative(skip)]
57    context: C,
58    /// Whether to clear storage before applying updates.
59    delete_storage_first: bool,
60    /// The number of entries persisted in storage.
61    stored_count: u32,
62    /// New values not yet persisted to storage.
63    new_values: Vec<T>,
64}
65
66impl<C, T> View for LogView<C, T>
67where
68    C: Context,
69    T: Send + Sync + Serialize,
70{
71    const NUM_INIT_KEYS: usize = 1;
72
73    type Context = C;
74
75    fn context(&self) -> C {
76        self.context.clone()
77    }
78
79    fn pre_load(context: &C) -> Result<Vec<Vec<u8>>, ViewError> {
80        Ok(vec![context.base_key().base_tag(KeyTag::Count as u8)])
81    }
82
83    fn post_load(context: C, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
84        let stored_count =
85            from_bytes_option_or_default(values.first().ok_or(ViewError::PostLoadValuesError)?)?;
86        Ok(Self {
87            context,
88            delete_storage_first: false,
89            stored_count,
90            new_values: Vec::new(),
91        })
92    }
93
94    fn rollback(&mut self) {
95        self.delete_storage_first = false;
96        self.new_values.clear();
97    }
98
99    async fn has_pending_changes(&self) -> bool {
100        if self.delete_storage_first {
101            return true;
102        }
103        !self.new_values.is_empty()
104    }
105
106    fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
107        let mut delete_view = false;
108        if self.delete_storage_first {
109            batch.delete_key_prefix(self.context.base_key().bytes.clone());
110            delete_view = true;
111        }
112        if !self.new_values.is_empty() {
113            delete_view = false;
114            let new_values_len =
115                u32::try_from(self.new_values.len()).map_err(|_| ArithmeticError::Overflow)?;
116            let new_count = self
117                .stored_count
118                .checked_add(new_values_len)
119                .ok_or(ArithmeticError::Overflow)?;
120            for (index, value) in (self.stored_count..).zip(&self.new_values) {
121                let key = self
122                    .context
123                    .base_key()
124                    .derive_tag_key(KeyTag::Index as u8, &index)?;
125                batch.put_key_value(key, value)?;
126            }
127            let key = self.context.base_key().base_tag(KeyTag::Count as u8);
128            batch.put_key_value(key, &new_count)?;
129        }
130        Ok(delete_view)
131    }
132
133    fn post_save(&mut self) {
134        if self.delete_storage_first {
135            self.stored_count = 0;
136        }
137        self.stored_count += u32::try_from(self.new_values.len()).expect("verified in pre_save");
138        self.new_values.clear();
139        self.delete_storage_first = false;
140    }
141
142    fn clear(&mut self) {
143        self.delete_storage_first = true;
144        self.new_values.clear();
145    }
146}
147
148impl<C, T> ClonableView for LogView<C, T>
149where
150    C: Context,
151    T: Clone + Send + Sync + Serialize,
152{
153    fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
154        Ok(LogView {
155            context: self.context.clone(),
156            delete_storage_first: self.delete_storage_first,
157            stored_count: self.stored_count,
158            new_values: self.new_values.clone(),
159        })
160    }
161}
162
163impl<C, T> LogView<C, T>
164where
165    C: Context,
166{
167    /// Pushes a value to the end of the log.
168    /// ```rust
169    /// # tokio_test::block_on(async {
170    /// # use linera_views::context::MemoryContext;
171    /// # use linera_views::log_view::LogView;
172    /// # use linera_views::views::View;
173    /// # let context = MemoryContext::new_for_testing(());
174    /// let mut log = LogView::load(context).await.unwrap();
175    /// log.push(34);
176    /// # })
177    /// ```
178    pub fn push(&mut self, value: T) {
179        self.new_values.push(value);
180    }
181
182    /// Reads the size of the log.
183    /// ```rust
184    /// # tokio_test::block_on(async {
185    /// # use linera_views::context::MemoryContext;
186    /// # use linera_views::log_view::LogView;
187    /// # use linera_views::views::View;
188    /// # let context = MemoryContext::new_for_testing(());
189    /// let mut log = LogView::load(context).await.unwrap();
190    /// log.push(34);
191    /// log.push(42);
192    /// assert_eq!(log.count(), 2);
193    /// # })
194    /// ```
195    pub fn count(&self) -> usize {
196        if self.delete_storage_first {
197            self.new_values.len()
198        } else {
199            self.stored_count as usize + self.new_values.len()
200        }
201    }
202
203    /// Obtains the extra data.
204    pub fn extra(&self) -> &C::Extra {
205        self.context.extra()
206    }
207}
208
209impl<C, T> LogView<C, T>
210where
211    C: Context,
212    T: Clone + DeserializeOwned + Serialize + Send + Sync,
213{
214    /// Reads the logged value with the given index (including staged ones).
215    /// ```rust
216    /// # tokio_test::block_on(async {
217    /// # use linera_views::context::MemoryContext;
218    /// # use linera_views::log_view::LogView;
219    /// # use linera_views::views::View;
220    /// # let context = MemoryContext::new_for_testing(());
221    /// let mut log = LogView::load(context).await.unwrap();
222    /// log.push(34);
223    /// assert_eq!(log.get(0).await.unwrap(), Some(34));
224    /// # })
225    /// ```
226    pub async fn get(&self, index: usize) -> Result<Option<T>, ViewError> {
227        let stored_count = self.stored_count as usize;
228        let value = if self.delete_storage_first {
229            self.new_values.get(index).cloned()
230        } else if index < stored_count {
231            let index = u32::try_from(index).map_err(|_| ArithmeticError::Overflow)?;
232            let key = self
233                .context
234                .base_key()
235                .derive_tag_key(KeyTag::Index as u8, &index)?;
236            self.context.store().read_value(&key).await?
237        } else {
238            self.new_values.get(index - stored_count).cloned()
239        };
240        Ok(value)
241    }
242
243    /// Reads several logged keys (including staged ones)
244    /// ```rust
245    /// # tokio_test::block_on(async {
246    /// # use linera_views::context::MemoryContext;
247    /// # use linera_views::log_view::LogView;
248    /// # use linera_views::views::View;
249    /// # let context = MemoryContext::new_for_testing(());
250    /// let mut log = LogView::load(context).await.unwrap();
251    /// log.push(34);
252    /// log.push(42);
253    /// assert_eq!(
254    ///     log.multi_get(vec![0, 1]).await.unwrap(),
255    ///     vec![Some(34), Some(42)]
256    /// );
257    /// # })
258    /// ```
259    pub async fn multi_get(&self, indices: Vec<usize>) -> Result<Vec<Option<T>>, ViewError> {
260        let mut result = Vec::new();
261        if self.delete_storage_first {
262            for index in indices {
263                result.push(self.new_values.get(index).cloned());
264            }
265        } else {
266            let stored_count = self.stored_count as usize;
267            let mut index_to_positions = BTreeMap::<usize, Vec<usize>>::new();
268            for (pos, index) in indices.into_iter().enumerate() {
269                if index < stored_count {
270                    index_to_positions.entry(index).or_default().push(pos);
271                    result.push(None);
272                } else {
273                    result.push(self.new_values.get(index - stored_count).cloned());
274                }
275            }
276            let mut keys = Vec::new();
277            let mut vec_positions = Vec::new();
278            for (index, positions) in index_to_positions {
279                let index = u32::try_from(index).map_err(|_| ArithmeticError::Overflow)?;
280                let key = self
281                    .context
282                    .base_key()
283                    .derive_tag_key(KeyTag::Index as u8, &index)?;
284                keys.push(key);
285                vec_positions.push(positions);
286            }
287            let values = self.context.store().read_multi_values(&keys).await?;
288            for (positions, value) in vec_positions.into_iter().zip(values) {
289                if let Some((&last, rest)) = positions.split_last() {
290                    for &position in rest {
291                        *result.get_mut(position).unwrap() = value.clone();
292                    }
293                    *result.get_mut(last).unwrap() = value;
294                }
295            }
296        }
297        Ok(result)
298    }
299
300    /// Reads the index-value pairs at the given positions.
301    /// ```rust
302    /// # tokio_test::block_on(async {
303    /// # use linera_views::context::MemoryContext;
304    /// # use linera_views::log_view::LogView;
305    /// # use linera_views::views::View;
306    /// # let context = MemoryContext::new_for_testing(());
307    /// let mut log = LogView::load(context).await.unwrap();
308    /// log.push(34);
309    /// log.push(42);
310    /// assert_eq!(
311    ///     log.multi_get_pairs(vec![0, 1, 5]).await.unwrap(),
312    ///     vec![(0, Some(34)), (1, Some(42)), (5, None)]
313    /// );
314    /// # })
315    /// ```
316    pub async fn multi_get_pairs(
317        &self,
318        indices: Vec<usize>,
319    ) -> Result<Vec<(usize, Option<T>)>, ViewError> {
320        let values = self.multi_get(indices.clone()).await?;
321        Ok(indices.into_iter().zip(values).collect())
322    }
323
324    async fn read_context(&self, range: Range<usize>) -> Result<Vec<T>, ViewError> {
325        let count = range.len();
326        let mut keys = Vec::with_capacity(count);
327        for index in range {
328            let index = u32::try_from(index).map_err(|_| ArithmeticError::Overflow)?;
329            let key = self
330                .context
331                .base_key()
332                .derive_tag_key(KeyTag::Index as u8, &index)?;
333            keys.push(key);
334        }
335        let mut values = Vec::with_capacity(count);
336        for entry in self.context.store().read_multi_values(&keys).await? {
337            match entry {
338                None => {
339                    return Err(ViewError::MissingEntries("LogView".into()));
340                }
341                Some(value) => values.push(value),
342            }
343        }
344        Ok(values)
345    }
346
347    /// Reads the logged values in the given range (including staged ones).
348    /// ```rust
349    /// # tokio_test::block_on(async {
350    /// # use linera_views::context::MemoryContext;
351    /// # use linera_views::log_view::LogView;
352    /// # use linera_views::views::View;
353    /// # let context = MemoryContext::new_for_testing(());
354    /// let mut log = LogView::load(context).await.unwrap();
355    /// log.push(34);
356    /// log.push(42);
357    /// log.push(56);
358    /// assert_eq!(log.read(0..2).await.unwrap(), vec![34, 42]);
359    /// # })
360    /// ```
361    pub async fn read<R>(&self, range: R) -> Result<Vec<T>, ViewError>
362    where
363        R: RangeBounds<usize>,
364    {
365        let effective_stored_count = if self.delete_storage_first {
366            0
367        } else {
368            self.stored_count as usize
369        };
370        let end = match range.end_bound() {
371            Bound::Included(end) => *end + 1,
372            Bound::Excluded(end) => *end,
373            Bound::Unbounded => self.count(),
374        }
375        .min(self.count());
376        let start = match range.start_bound() {
377            Bound::Included(start) => *start,
378            Bound::Excluded(start) => *start + 1,
379            Bound::Unbounded => 0,
380        };
381        if start >= end {
382            return Ok(Vec::new());
383        }
384        if start < effective_stored_count {
385            if end <= effective_stored_count {
386                self.read_context(start..end).await
387            } else {
388                let mut values = self.read_context(start..effective_stored_count).await?;
389                values.extend(
390                    self.new_values[0..(end - effective_stored_count)]
391                        .iter()
392                        .cloned(),
393                );
394                Ok(values)
395            }
396        } else {
397            Ok(
398                self.new_values[(start - effective_stored_count)..(end - effective_stored_count)]
399                    .to_vec(),
400            )
401        }
402    }
403}
404
405impl<C, T> HashableView for LogView<C, T>
406where
407    C: Context,
408    T: Send + Sync + Clone + Serialize + DeserializeOwned,
409{
410    type Hasher = sha3::Sha3_256;
411
412    async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
413        self.hash().await
414    }
415
416    async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
417        #[cfg(with_metrics)]
418        let _hash_latency = metrics::LOG_VIEW_HASH_RUNTIME.measure_latency();
419        let elements = self.read(..).await?;
420        let mut hasher = sha3::Sha3_256::default();
421        hasher.update_with_bcs_bytes(&elements)?;
422        Ok(hasher.finalize())
423    }
424}
425
426/// Type wrapping `LogView` while memoizing the hash.
427pub type HashedLogView<C, T> = WrappedHashableContainerView<C, LogView<C, T>, HasherOutput>;
428
429/// Wrapper around `LogView` to compute hashes based on the history of changes.
430pub type HistoricallyHashedLogView<C, T> = HistoricallyHashableView<C, LogView<C, T>>;
431
432#[cfg(not(web))]
433mod graphql {
434    use std::borrow::Cow;
435
436    use linera_base::data_types::ArithmeticError;
437
438    use super::LogView;
439    use crate::{
440        context::Context,
441        graphql::{hash_name, mangle},
442    };
443
444    impl<C: Send + Sync, T: async_graphql::OutputType> async_graphql::TypeName for LogView<C, T> {
445        fn type_name() -> Cow<'static, str> {
446            format!(
447                "LogView_{}_{:08x}",
448                mangle(T::type_name()),
449                hash_name::<T>()
450            )
451            .into()
452        }
453    }
454
455    #[async_graphql::Object(cache_control(no_cache), name_type)]
456    impl<C: Context, T: async_graphql::OutputType> LogView<C, T>
457    where
458        T: serde::ser::Serialize + serde::de::DeserializeOwned + Clone + Send + Sync,
459    {
460        #[graphql(derived(name = "count"))]
461        async fn count_(&self) -> Result<u32, async_graphql::Error> {
462            Ok(u32::try_from(self.count()).map_err(|_| ArithmeticError::Overflow)?)
463        }
464
465        async fn entries(
466            &self,
467            start: Option<usize>,
468            end: Option<usize>,
469        ) -> async_graphql::Result<Vec<T>> {
470            Ok(self
471                .read(start.unwrap_or_default()..end.unwrap_or_else(|| self.count()))
472                .await?)
473        }
474    }
475}
linera_views/views/log_view.rs

linera_views/views/
log_view.rs
