linera_views/views/collection_view.rs
1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5 borrow::Borrow,
6 collections::{btree_map, BTreeMap},
7 io::Write,
8 marker::PhantomData,
9 mem,
10 ops::Deref,
11};
12
13use allocative::{Allocative, Key, Visitor};
14use async_lock::{RwLock, RwLockReadGuard};
15use linera_base::data_types::ArithmeticError;
16#[cfg(with_metrics)]
17use linera_base::prometheus_util::MeasureLatency as _;
18use serde::{de::DeserializeOwned, Serialize};
19
20use crate::{
21 batch::Batch,
22 common::{CustomSerialize, HasherOutput, SliceExt as _, Update},
23 context::{BaseKey, Context},
24 hashable_wrapper::WrappedHashableContainerView,
25 historical_hash_wrapper::HistoricallyHashableView,
26 store::ReadableKeyValueStore as _,
27 views::{collection_entry, ClonableView, HashableView, Hasher, View, ViewError},
28};
29
30#[cfg(with_metrics)]
31pub(crate) mod metrics {
32 use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram_vec};
33 use prometheus::HistogramVec;
34
35 linera_base::declare_metrics! {
36 /// The runtime of hash computation
37 pub static COLLECTION_VIEW_HASH_RUNTIME: HistogramVec =
38 register_histogram_vec(
39 "collection_view_hash_runtime",
40 "CollectionView hash runtime",
41 &[],
42 exponential_bucket_latencies(5.0),
43 );
44 }
45}
46
47/// A view that supports accessing a collection of views of the same kind, indexed by a
48/// `Vec<u8>`, one subview at a time.
49#[derive(Debug)]
50pub struct ByteCollectionView<C, W> {
51 /// The view context.
52 context: C,
53 /// Whether to clear storage before applying updates.
54 delete_storage_first: bool,
55 /// Entries that may have staged changes.
56 updates: RwLock<BTreeMap<Vec<u8>, Update<W>>>,
57}
58
59impl<C, W: Allocative> Allocative for ByteCollectionView<C, W> {
60 fn visit<'a, 'b: 'a>(&self, visitor: &'a mut Visitor<'b>) {
61 let name = Key::new("ByteCollectionView");
62 let size = mem::size_of::<Self>();
63 let mut visitor = visitor.enter(name, size);
64 if let Some(updates) = self.updates.try_read() {
65 updates.deref().visit(&mut visitor);
66 }
67 visitor.exit();
68 }
69}
70
71/// A read-only accessor for a particular subview in a [`CollectionView`].
72pub enum ReadGuardedView<'a, W> {
73 /// The view is loaded in the updates
74 Loaded {
75 /// The guard for the updates.
76 updates: RwLockReadGuard<'a, BTreeMap<Vec<u8>, Update<W>>>,
77 /// The key in question.
78 short_key: Vec<u8>,
79 },
80 /// The view is not loaded in the updates
81 NotLoaded {
82 /// The guard for the updates. It is needed so that it prevents
83 /// opening the view as write separately.
84 _updates: RwLockReadGuard<'a, BTreeMap<Vec<u8>, Update<W>>>,
85 /// The view obtained from the storage
86 view: W,
87 },
88}
89
90impl<W> std::ops::Deref for ReadGuardedView<'_, W> {
91 type Target = W;
92
93 fn deref(&self) -> &W {
94 match self {
95 ReadGuardedView::Loaded { updates, short_key } => {
96 let Update::Set(view) = updates.get(short_key).unwrap() else {
97 unreachable!("ReadGuardedView should only reference Update::Set entries");
98 };
99 view
100 }
101 ReadGuardedView::NotLoaded { _updates, view } => view,
102 }
103 }
104}
105
106impl<W: View> View for ByteCollectionView<W::Context, W> {
107 const NUM_INIT_KEYS: usize = 0;
108
109 type Context = W::Context;
110
111 fn context(&self) -> Self::Context {
112 self.context.clone()
113 }
114
115 fn pre_load(_context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
116 Ok(vec![])
117 }
118
119 fn post_load(context: Self::Context, _values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
120 Ok(Self {
121 context,
122 delete_storage_first: false,
123 updates: RwLock::new(BTreeMap::new()),
124 })
125 }
126
127 fn rollback(&mut self) {
128 self.delete_storage_first = false;
129 self.updates.get_mut().clear();
130 }
131
132 async fn has_pending_changes(&self) -> bool {
133 if self.delete_storage_first {
134 return true;
135 }
136 let updates = self.updates.read().await;
137 !updates.is_empty()
138 }
139
140 fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
141 let mut delete_view = false;
142 let updates = self
143 .updates
144 .try_read()
145 .ok_or_else(|| ViewError::TryLockError(vec![]))?;
146 if self.delete_storage_first {
147 delete_view = true;
148 batch.delete_key_prefix(self.context.base_key().bytes.clone());
149 for (index, update) in updates.iter() {
150 if let Update::Set(view) = update {
151 view.pre_save(batch)?;
152 self.add_index(batch, index);
153 delete_view = false;
154 }
155 }
156 } else {
157 for (index, update) in updates.iter() {
158 match update {
159 Update::Set(view) => {
160 view.pre_save(batch)?;
161 self.add_index(batch, index);
162 }
163 Update::Removed => {
164 let key_subview = self.get_subview_key(index);
165 let key_index = self.get_index_key(index);
166 batch.delete_key(key_index);
167 batch.delete_key_prefix(key_subview);
168 }
169 }
170 }
171 }
172 Ok(delete_view)
173 }
174
175 fn post_save(&mut self) {
176 for update in self.updates.get_mut().values_mut() {
177 if let Update::Set(view) = update {
178 view.post_save();
179 }
180 }
181 self.delete_storage_first = false;
182 self.updates.get_mut().clear();
183 }
184
185 fn clear(&mut self) {
186 self.delete_storage_first = true;
187 self.updates.get_mut().clear();
188 }
189}
190
191impl<W: ClonableView> ClonableView for ByteCollectionView<W::Context, W> {
192 fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
193 let cloned_updates = self
194 .updates
195 .get_mut()
196 .iter_mut()
197 .map(|(key, value)| {
198 let cloned_value: Result<_, ViewError> = match value {
199 Update::Removed => Ok(Update::Removed),
200 Update::Set(view) => Ok(Update::Set(view.clone_unchecked()?)),
201 };
202 cloned_value.map(|v| (key.clone(), v))
203 })
204 .collect::<Result<_, ViewError>>()?;
205
206 Ok(ByteCollectionView {
207 context: self.context.clone(),
208 delete_storage_first: self.delete_storage_first,
209 updates: RwLock::new(cloned_updates),
210 })
211 }
212}
213
214impl<W: View> ByteCollectionView<W::Context, W> {
215 fn get_index_key(&self, index: &[u8]) -> Vec<u8> {
216 collection_entry::index_key(&self.context, index)
217 }
218
219 fn get_subview_key(&self, index: &[u8]) -> Vec<u8> {
220 collection_entry::subview_key(&self.context, index)
221 }
222
223 fn add_index(&self, batch: &mut Batch, index: &[u8]) {
224 let key = self.get_index_key(index);
225 batch.put_key_value_bytes(key, vec![]);
226 }
227
228 /// Loads a subview for the data at the given index in the collection. If an entry
229 /// is absent then a default entry is added to the collection. The resulting view
230 /// can be modified.
231 /// ```rust
232 /// # tokio_test::block_on(async {
233 /// # use linera_views::context::MemoryContext;
234 /// # use linera_views::collection_view::ByteCollectionView;
235 /// # use linera_views::register_view::RegisterView;
236 /// # use linera_views::views::View;
237 /// # let context = MemoryContext::new_for_testing(());
238 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
239 /// ByteCollectionView::load(context).await.unwrap();
240 /// let subview = view.load_entry_mut(&[0, 1]).await.unwrap();
241 /// let value = subview.get();
242 /// assert_eq!(*value, String::default());
243 /// # })
244 /// ```
245 pub async fn load_entry_mut(&mut self, short_key: &[u8]) -> Result<&mut W, ViewError> {
246 match self.updates.get_mut().entry(short_key.to_vec()) {
247 btree_map::Entry::Occupied(entry) => {
248 let entry = entry.into_mut();
249 match entry {
250 Update::Set(view) => Ok(view),
251 Update::Removed => {
252 let key = collection_entry::subview_key(&self.context, short_key);
253 let context = self.context.clone_with_base_key(key);
254 // Obtain a view and set its pending state to the default (e.g. empty) state
255 let view = W::new(context)?;
256 *entry = Update::Set(view);
257 let Update::Set(view) = entry else {
258 unreachable!("Entry was just set to Update::Set");
259 };
260 Ok(view)
261 }
262 }
263 }
264 btree_map::Entry::Vacant(entry) => {
265 let key = collection_entry::subview_key(&self.context, short_key);
266 let context = self.context.clone_with_base_key(key);
267 let view = if self.delete_storage_first {
268 W::new(context)?
269 } else {
270 W::load(context).await?
271 };
272 let Update::Set(view) = entry.insert(Update::Set(view)) else {
273 unreachable!("Entry was just inserted as Update::Set");
274 };
275 Ok(view)
276 }
277 }
278 }
279
280 /// Loads a subview for the data at the given index in the collection. If an entry
281 /// is absent then `None` is returned. The resulting view cannot be modified.
282 /// May fail if one subview is already being visited.
283 /// ```rust
284 /// # tokio_test::block_on(async {
285 /// # use linera_views::context::MemoryContext;
286 /// # use linera_views::collection_view::ByteCollectionView;
287 /// # use linera_views::register_view::RegisterView;
288 /// # use linera_views::views::View;
289 /// # let context = MemoryContext::new_for_testing(());
290 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
291 /// ByteCollectionView::load(context).await.unwrap();
292 /// {
293 /// let _subview = view.load_entry_mut(&[0, 1]).await.unwrap();
294 /// }
295 /// {
296 /// let subview = view.try_load_entry(&[0, 1]).await.unwrap().unwrap();
297 /// let value = subview.get();
298 /// assert_eq!(*value, String::default());
299 /// }
300 /// assert!(view.try_load_entry(&[0, 2]).await.unwrap().is_none());
301 /// # })
302 /// ```
303 pub async fn try_load_entry(
304 &self,
305 short_key: &[u8],
306 ) -> Result<Option<ReadGuardedView<'_, W>>, ViewError> {
307 let updates = self.updates.read().await;
308 match updates.get(short_key) {
309 Some(update) => match update {
310 Update::Removed => Ok(None),
311 Update::Set(_) => Ok(Some(ReadGuardedView::Loaded {
312 updates,
313 short_key: short_key.to_vec(),
314 })),
315 },
316 None => {
317 if self.delete_storage_first {
318 return Ok(None);
319 }
320 // The index marker and the subview's initialization keys are read together, so
321 // that loading an entry costs a single round trip whether or not it exists.
322 let (subview_context, keys) =
323 collection_entry::entry_keys::<W>(&self.context, short_key)?;
324 let values = self.context.store().read_multi_values_bytes(&keys).await?;
325 let subview = collection_entry::post_load_entry::<W>(subview_context, &values)?;
326 let entry = subview.map(|view| ReadGuardedView::NotLoaded {
327 _updates: updates,
328 view,
329 });
330 Ok(entry)
331 }
332 }
333 }
334
335 /// Load multiple entries for reading at once.
336 /// The entries in `short_keys` have to be all distinct.
337 /// ```rust
338 /// # tokio_test::block_on(async {
339 /// # use linera_views::context::MemoryContext;
340 /// # use linera_views::collection_view::ByteCollectionView;
341 /// # use linera_views::register_view::RegisterView;
342 /// # use linera_views::views::View;
343 /// # let context = MemoryContext::new_for_testing(());
344 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
345 /// ByteCollectionView::load(context).await.unwrap();
346 /// {
347 /// let _subview = view.load_entry_mut(&[0, 1]).await.unwrap();
348 /// }
349 /// let short_keys = vec![vec![0, 1], vec![2, 3]];
350 /// let subviews = view.try_load_entries(short_keys).await.unwrap();
351 /// let value0 = subviews[0].as_ref().unwrap().get();
352 /// assert_eq!(*value0, String::default());
353 /// # })
354 /// ```
355 pub async fn try_load_entries(
356 &self,
357 short_keys: Vec<Vec<u8>>,
358 ) -> Result<Vec<Option<ReadGuardedView<'_, W>>>, ViewError> {
359 let mut results = Vec::with_capacity(short_keys.len());
360 let mut entries_to_load = Vec::new();
361 let updates = self.updates.read().await;
362
363 for (position, short_key) in short_keys.into_iter().enumerate() {
364 match updates.get(&short_key) {
365 Some(update) => match update {
366 Update::Removed => {
367 results.push(None);
368 }
369 Update::Set(_) => {
370 let updates = self.updates.read().await;
371 results.push(Some(ReadGuardedView::Loaded {
372 updates,
373 short_key: short_key.clone(),
374 }));
375 }
376 },
377 None => {
378 results.push(None); // Placeholder, may be updated later
379 if !self.delete_storage_first {
380 entries_to_load.push((position, short_key));
381 }
382 }
383 }
384 }
385
386 if !entries_to_load.is_empty() {
387 // The index markers and the subviews' initialization keys are read together, so
388 // that loading entries costs a single round trip whether or not they exist.
389 let entry_len = collection_entry::entry_len::<W>();
390 let mut keys = Vec::with_capacity(entries_to_load.len() * entry_len);
391 let mut subview_contexts = Vec::with_capacity(entries_to_load.len());
392 for (_, short_key) in &entries_to_load {
393 let (subview_context, entry) =
394 collection_entry::entry_keys::<W>(&self.context, short_key)?;
395 keys.extend(entry);
396 subview_contexts.push(subview_context);
397 }
398 let values = self.context.store().read_multi_values_bytes(&keys).await?;
399 for ((position, _), (entry_values, subview_context)) in entries_to_load
400 .into_iter()
401 .zip(values.chunks_exact(entry_len).zip(subview_contexts))
402 {
403 if let Some(view) =
404 collection_entry::post_load_entry::<W>(subview_context, entry_values)?
405 {
406 results[position] = Some(ReadGuardedView::NotLoaded {
407 _updates: self.updates.read().await,
408 view,
409 });
410 }
411 }
412 }
413
414 Ok(results)
415 }
416
417 /// Loads multiple entries for reading at once with their keys.
418 /// ```rust
419 /// # tokio_test::block_on(async {
420 /// # use linera_views::context::MemoryContext;
421 /// # use linera_views::collection_view::ByteCollectionView;
422 /// # use linera_views::register_view::RegisterView;
423 /// # use linera_views::views::View;
424 /// # let context = MemoryContext::new_for_testing(());
425 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
426 /// ByteCollectionView::load(context).await.unwrap();
427 /// {
428 /// let subview = view.load_entry_mut(&vec![0, 1]).await.unwrap();
429 /// subview.set("Bonjour".into());
430 /// }
431 /// let short_keys = vec![vec![0, 1], vec![0, 2]];
432 /// let pairs = view.try_load_entries_pairs(short_keys).await.unwrap();
433 /// assert_eq!(pairs[0].0, vec![0, 1]);
434 /// assert_eq!(pairs[1].0, vec![0, 2]);
435 /// let value0 = pairs[0].1.as_ref().unwrap().get();
436 /// assert_eq!(*value0, "Bonjour".to_string());
437 /// assert!(pairs[1].1.is_none());
438 /// # })
439 /// ```
440 pub async fn try_load_entries_pairs(
441 &self,
442 short_keys: Vec<Vec<u8>>,
443 ) -> Result<Vec<(Vec<u8>, Option<ReadGuardedView<'_, W>>)>, ViewError> {
444 let values = self.try_load_entries(short_keys.clone()).await?;
445 Ok(short_keys.into_iter().zip(values).collect())
446 }
447
448 /// Load all entries for reading at once.
449 /// ```rust
450 /// # tokio_test::block_on(async {
451 /// # use linera_views::context::MemoryContext;
452 /// # use linera_views::collection_view::ByteCollectionView;
453 /// # use linera_views::register_view::RegisterView;
454 /// # use linera_views::views::View;
455 /// # let context = MemoryContext::new_for_testing(());
456 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
457 /// ByteCollectionView::load(context).await.unwrap();
458 /// {
459 /// let _subview = view.load_entry_mut(&[0, 1]).await.unwrap();
460 /// }
461 /// let subviews = view.try_load_all_entries().await.unwrap();
462 /// assert_eq!(subviews.len(), 1);
463 /// # })
464 /// ```
465 pub async fn try_load_all_entries(
466 &self,
467 ) -> Result<Vec<(Vec<u8>, ReadGuardedView<'_, W>)>, ViewError> {
468 let updates = self.updates.read().await; // Acquire the read lock to prevent writes.
469 let short_keys = self.keys().await?;
470 let mut results = Vec::with_capacity(short_keys.len());
471
472 let mut keys_to_load = Vec::new();
473 let mut keys_to_load_metadata = Vec::new();
474 for (position, short_key) in short_keys.iter().enumerate() {
475 match updates.get(short_key) {
476 Some(update) => {
477 let Update::Set(_) = update else {
478 unreachable!("Loaded entries in updates should always be Update::Set");
479 };
480 let updates = self.updates.read().await;
481 let view = ReadGuardedView::Loaded {
482 updates,
483 short_key: short_key.clone(),
484 };
485 results.push((short_key.clone(), Some(view)));
486 }
487 None => {
488 // If a key is not in `updates`, then it is in storage.
489 // The key exists since otherwise it would not be in `short_keys`.
490 // Therefore we have `self.delete_storage_first = false`.
491 assert!(!self.delete_storage_first);
492 results.push((short_key.clone(), None));
493 let key = collection_entry::subview_key(&self.context, short_key);
494 let subview_context = self.context.clone_with_base_key(key);
495 keys_to_load.extend(W::pre_load(&subview_context)?);
496 keys_to_load_metadata.push((position, subview_context, short_key.clone()));
497 }
498 }
499 }
500
501 let values = self
502 .context
503 .store()
504 .read_multi_values_bytes(&keys_to_load)
505 .await?;
506
507 for (loaded_values, (position, context, short_key)) in values
508 .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
509 .zip(keys_to_load_metadata)
510 {
511 let view = W::post_load(context, loaded_values)?;
512 let updates = self.updates.read().await;
513 let guarded_view = ReadGuardedView::NotLoaded {
514 _updates: updates,
515 view,
516 };
517 results[position] = (short_key, Some(guarded_view));
518 }
519
520 Ok(results
521 .into_iter()
522 .map(|(short_key, view)| (short_key, view.unwrap()))
523 .collect::<Vec<_>>())
524 }
525
526 /// Resets an entry to the default value.
527 /// ```rust
528 /// # tokio_test::block_on(async {
529 /// # use linera_views::context::MemoryContext;
530 /// # use linera_views::collection_view::ByteCollectionView;
531 /// # use linera_views::register_view::RegisterView;
532 /// # use linera_views::views::View;
533 /// # let context = MemoryContext::new_for_testing(());
534 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
535 /// ByteCollectionView::load(context).await.unwrap();
536 /// let subview = view.load_entry_mut(&[0, 1]).await.unwrap();
537 /// let value = subview.get_mut();
538 /// *value = String::from("Hello");
539 /// view.reset_entry_to_default(&[0, 1]).unwrap();
540 /// let subview = view.load_entry_mut(&[0, 1]).await.unwrap();
541 /// let value = subview.get_mut();
542 /// assert_eq!(*value, String::default());
543 /// # })
544 /// ```
545 pub fn reset_entry_to_default(&mut self, short_key: &[u8]) -> Result<(), ViewError> {
546 let key = collection_entry::subview_key(&self.context, short_key);
547 let context = self.context.clone_with_base_key(key);
548 let view = W::new(context)?;
549 self.updates
550 .get_mut()
551 .insert(short_key.to_vec(), Update::Set(view));
552 Ok(())
553 }
554
555 /// Tests if the collection contains a specified key and returns a boolean.
556 /// ```rust
557 /// # tokio_test::block_on(async {
558 /// # use linera_views::context::MemoryContext;
559 /// # use linera_views::collection_view::ByteCollectionView;
560 /// # use linera_views::register_view::RegisterView;
561 /// # use linera_views::views::View;
562 /// # let context = MemoryContext::new_for_testing(());
563 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
564 /// ByteCollectionView::load(context).await.unwrap();
565 /// {
566 /// let _subview = view.load_entry_mut(&[0, 1]).await.unwrap();
567 /// }
568 /// assert!(view.contains_key(&[0, 1]).await.unwrap());
569 /// assert!(!view.contains_key(&[0, 2]).await.unwrap());
570 /// # })
571 /// ```
572 pub async fn contains_key(&self, short_key: &[u8]) -> Result<bool, ViewError> {
573 let updates = self.updates.read().await;
574 let contains = match updates.get(short_key) {
575 Some(entry) => match entry {
576 Update::Set(_view) => true,
577 _entry @ Update::Removed => false,
578 },
579 None => {
580 let key_index = collection_entry::index_key(&self.context, short_key);
581 !self.delete_storage_first && self.context.store().contains_key(&key_index).await?
582 }
583 };
584 Ok(contains)
585 }
586
587 /// Marks the entry as removed. If absent then nothing is done.
588 /// ```rust
589 /// # tokio_test::block_on(async {
590 /// # use linera_views::context::MemoryContext;
591 /// # use linera_views::collection_view::ByteCollectionView;
592 /// # use linera_views::register_view::RegisterView;
593 /// # use linera_views::views::View;
594 /// # let context = MemoryContext::new_for_testing(());
595 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
596 /// ByteCollectionView::load(context).await.unwrap();
597 /// let subview = view.load_entry_mut(&[0, 1]).await.unwrap();
598 /// let value = subview.get_mut();
599 /// assert_eq!(*value, String::default());
600 /// view.remove_entry(vec![0, 1]);
601 /// let keys = view.keys().await.unwrap();
602 /// assert_eq!(keys.len(), 0);
603 /// # })
604 /// ```
605 pub fn remove_entry(&mut self, short_key: Vec<u8>) {
606 if self.delete_storage_first {
607 // Optimization: No need to mark `short_key` for deletion as we are going to remove all the keys at once.
608 self.updates.get_mut().remove(&short_key);
609 } else {
610 self.updates.get_mut().insert(short_key, Update::Removed);
611 }
612 }
613
614 /// Gets the extra data.
615 pub fn extra(&self) -> &<W::Context as Context>::Extra {
616 self.context.extra()
617 }
618}
619
620impl<W: View> ByteCollectionView<W::Context, W> {
621 /// Applies a function f on each index (aka key). Keys are visited in the
622 /// lexicographic order. If the function returns false, then the loop
623 /// ends prematurely.
624 /// ```rust
625 /// # tokio_test::block_on(async {
626 /// # use linera_views::context::MemoryContext;
627 /// # use linera_views::collection_view::ByteCollectionView;
628 /// # use linera_views::register_view::RegisterView;
629 /// # use linera_views::views::View;
630 /// # let context = MemoryContext::new_for_testing(());
631 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
632 /// ByteCollectionView::load(context).await.unwrap();
633 /// view.load_entry_mut(&[0, 1]).await.unwrap();
634 /// view.load_entry_mut(&[0, 2]).await.unwrap();
635 /// let mut count = 0;
636 /// view.for_each_key_while(|_key| {
637 /// count += 1;
638 /// Ok(count < 1)
639 /// })
640 /// .await
641 /// .unwrap();
642 /// assert_eq!(count, 1);
643 /// # })
644 /// ```
645 pub async fn for_each_key_while<F>(&self, mut f: F) -> Result<(), ViewError>
646 where
647 F: FnMut(&[u8]) -> Result<bool, ViewError> + Send,
648 {
649 let updates = self.updates.read().await;
650 let mut updates = updates.iter();
651 let mut update = updates.next();
652 if !self.delete_storage_first {
653 let base = self.get_index_key(&[]);
654 for index in self.context.store().find_keys_by_prefix(&base).await? {
655 loop {
656 match update {
657 Some((key, value)) if key <= &index => {
658 if let Update::Set(_) = value {
659 if !f(key)? {
660 return Ok(());
661 }
662 }
663 update = updates.next();
664 if key == &index {
665 break;
666 }
667 }
668 _ => {
669 if !f(&index)? {
670 return Ok(());
671 }
672 break;
673 }
674 }
675 }
676 }
677 }
678 while let Some((key, value)) = update {
679 if let Update::Set(_) = value {
680 if !f(key)? {
681 return Ok(());
682 }
683 }
684 update = updates.next();
685 }
686 Ok(())
687 }
688
689 /// Applies a function f on each index (aka key). Keys are visited in a
690 /// lexicographic order.
691 /// ```rust
692 /// # tokio_test::block_on(async {
693 /// # use linera_views::context::MemoryContext;
694 /// # use linera_views::collection_view::ByteCollectionView;
695 /// # use linera_views::register_view::RegisterView;
696 /// # use linera_views::views::View;
697 /// # let context = MemoryContext::new_for_testing(());
698 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
699 /// ByteCollectionView::load(context).await.unwrap();
700 /// view.load_entry_mut(&[0, 1]).await.unwrap();
701 /// view.load_entry_mut(&[0, 2]).await.unwrap();
702 /// let mut count = 0;
703 /// view.for_each_key(|_key| {
704 /// count += 1;
705 /// Ok(())
706 /// })
707 /// .await
708 /// .unwrap();
709 /// assert_eq!(count, 2);
710 /// # })
711 /// ```
712 pub async fn for_each_key<F>(&self, mut f: F) -> Result<(), ViewError>
713 where
714 F: FnMut(&[u8]) -> Result<(), ViewError> + Send,
715 {
716 self.for_each_key_while(|key| {
717 f(key)?;
718 Ok(true)
719 })
720 .await
721 }
722
723 /// Returns the list of keys in the collection. The order is lexicographic.
724 /// ```rust
725 /// # tokio_test::block_on(async {
726 /// # use linera_views::context::MemoryContext;
727 /// # use linera_views::collection_view::ByteCollectionView;
728 /// # use linera_views::register_view::RegisterView;
729 /// # use linera_views::views::View;
730 /// # let context = MemoryContext::new_for_testing(());
731 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
732 /// ByteCollectionView::load(context).await.unwrap();
733 /// view.load_entry_mut(&[0, 1]).await.unwrap();
734 /// view.load_entry_mut(&[0, 2]).await.unwrap();
735 /// let keys = view.keys().await.unwrap();
736 /// assert_eq!(keys, vec![vec![0, 1], vec![0, 2]]);
737 /// # })
738 /// ```
739 pub async fn keys(&self) -> Result<Vec<Vec<u8>>, ViewError> {
740 let mut keys = Vec::new();
741 self.for_each_key(|key| {
742 keys.push(key.to_vec());
743 Ok(())
744 })
745 .await?;
746 Ok(keys)
747 }
748
749 /// Returns the number of entries in the collection.
750 /// ```rust
751 /// # tokio_test::block_on(async {
752 /// # use linera_views::context::MemoryContext;
753 /// # use linera_views::collection_view::ByteCollectionView;
754 /// # use linera_views::register_view::RegisterView;
755 /// # use linera_views::views::View;
756 /// # let context = MemoryContext::new_for_testing(());
757 /// let mut view: ByteCollectionView<_, RegisterView<_, String>> =
758 /// ByteCollectionView::load(context).await.unwrap();
759 /// view.load_entry_mut(&[0, 1]).await.unwrap();
760 /// view.load_entry_mut(&[0, 2]).await.unwrap();
761 /// assert_eq!(view.iterative_count().await.unwrap(), 2);
762 /// # })
763 /// ```
764 pub async fn iterative_count(&self) -> Result<usize, ViewError> {
765 let mut count = 0;
766 self.for_each_key(|_key| {
767 count += 1;
768 Ok(())
769 })
770 .await?;
771 Ok(count)
772 }
773}
774
775impl<W: HashableView> HashableView for ByteCollectionView<W::Context, W> {
776 type Hasher = sha3::Sha3_256;
777
778 async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
779 #[cfg(with_metrics)]
780 let _hash_latency = metrics::COLLECTION_VIEW_HASH_RUNTIME.measure_latency();
781 let mut hasher = sha3::Sha3_256::default();
782 let keys = self.keys().await?;
783 let count = u32::try_from(keys.len()).map_err(|_| ArithmeticError::Overflow)?;
784 hasher.update_with_bcs_bytes(&count)?;
785 let updates = self.updates.get_mut();
786 for key in keys {
787 hasher.update_with_bytes(&key)?;
788 let hash = match updates.get_mut(&key) {
789 Some(entry) => {
790 let Update::Set(view) = entry else {
791 unreachable!("Loaded entries in updates should always be Update::Set");
792 };
793 view.hash_mut().await?
794 }
795 None => {
796 let key = collection_entry::subview_key(&self.context, &key);
797 let context = self.context.clone_with_base_key(key);
798 let mut view = W::load(context).await?;
799 view.hash_mut().await?
800 }
801 };
802 hasher.write_all(hash.as_ref())?;
803 }
804 Ok(hasher.finalize())
805 }
806
807 async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
808 #[cfg(with_metrics)]
809 let _hash_latency = metrics::COLLECTION_VIEW_HASH_RUNTIME.measure_latency();
810 let mut hasher = sha3::Sha3_256::default();
811 let updates = self.updates.read().await; // Acquire the lock to prevent writes.
812 let keys = self.keys().await?;
813 let count = u32::try_from(keys.len()).map_err(|_| ArithmeticError::Overflow)?;
814 hasher.update_with_bcs_bytes(&count)?;
815 for key in keys {
816 hasher.update_with_bytes(&key)?;
817 let hash = match updates.get(&key) {
818 Some(entry) => {
819 let Update::Set(view) = entry else {
820 unreachable!("Loaded entries in updates should always be Update::Set");
821 };
822 view.hash().await?
823 }
824 None => {
825 let key = collection_entry::subview_key(&self.context, &key);
826 let context = self.context.clone_with_base_key(key);
827 let view = W::load(context).await?;
828 view.hash().await?
829 }
830 };
831 hasher.write_all(hash.as_ref())?;
832 }
833 Ok(hasher.finalize())
834 }
835}
836
837/// A view that supports accessing a collection of views of the same kind, indexed by a
838/// key, one subview at a time.
839#[derive(Debug, Allocative)]
840#[allocative(bound = "C, I, W: Allocative")]
841pub struct CollectionView<C, I, W> {
842 collection: ByteCollectionView<C, W>,
843 #[allocative(skip)]
844 _phantom: PhantomData<I>,
845}
846
847impl<W: View, I> View for CollectionView<W::Context, I, W>
848where
849 I: Send + Sync + Serialize + DeserializeOwned,
850{
851 const NUM_INIT_KEYS: usize = ByteCollectionView::<W::Context, W>::NUM_INIT_KEYS;
852
853 type Context = W::Context;
854
855 fn context(&self) -> Self::Context {
856 self.collection.context()
857 }
858
859 fn pre_load(context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
860 ByteCollectionView::<W::Context, W>::pre_load(context)
861 }
862
863 fn post_load(context: Self::Context, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
864 let collection = ByteCollectionView::post_load(context, values)?;
865 Ok(CollectionView {
866 collection,
867 _phantom: PhantomData,
868 })
869 }
870
871 fn rollback(&mut self) {
872 self.collection.rollback()
873 }
874
875 async fn has_pending_changes(&self) -> bool {
876 self.collection.has_pending_changes().await
877 }
878
879 fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
880 self.collection.pre_save(batch)
881 }
882
883 fn post_save(&mut self) {
884 self.collection.post_save()
885 }
886
887 fn clear(&mut self) {
888 self.collection.clear()
889 }
890}
891
892impl<I, W: ClonableView> ClonableView for CollectionView<W::Context, I, W>
893where
894 I: Send + Sync + Serialize + DeserializeOwned,
895{
896 fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
897 Ok(CollectionView {
898 collection: self.collection.clone_unchecked()?,
899 _phantom: PhantomData,
900 })
901 }
902}
903
904impl<I: Serialize, W: View> CollectionView<W::Context, I, W> {
905 /// Loads a subview for the data at the given index in the collection. If an entry
906 /// is absent then a default entry is added to the collection. The resulting view
907 /// can be modified.
908 /// ```rust
909 /// # tokio_test::block_on(async {
910 /// # use linera_views::context::MemoryContext;
911 /// # use linera_views::collection_view::CollectionView;
912 /// # use linera_views::register_view::RegisterView;
913 /// # use linera_views::views::View;
914 /// # let context = MemoryContext::new_for_testing(());
915 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
916 /// CollectionView::load(context).await.unwrap();
917 /// let subview = view.load_entry_mut(&23).await.unwrap();
918 /// let value = subview.get();
919 /// assert_eq!(*value, String::default());
920 /// # })
921 /// ```
922 pub async fn load_entry_mut<Q>(&mut self, index: &Q) -> Result<&mut W, ViewError>
923 where
924 I: Borrow<Q>,
925 Q: Serialize + ?Sized,
926 {
927 let short_key = BaseKey::derive_short_key(index)?;
928 self.collection.load_entry_mut(&short_key).await
929 }
930
931 /// Loads a subview for the data at the given index in the collection. If an entry
932 /// is absent then `None` is returned. The resulting view cannot be modified.
933 /// May fail if one subview is already being visited.
934 /// ```rust
935 /// # tokio_test::block_on(async {
936 /// # use linera_views::context::MemoryContext;
937 /// # use linera_views::collection_view::CollectionView;
938 /// # use linera_views::register_view::RegisterView;
939 /// # use linera_views::views::View;
940 /// # let context = MemoryContext::new_for_testing(());
941 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
942 /// CollectionView::load(context).await.unwrap();
943 /// {
944 /// let _subview = view.load_entry_mut(&23).await.unwrap();
945 /// }
946 /// {
947 /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
948 /// let value = subview.get();
949 /// assert_eq!(*value, String::default());
950 /// }
951 /// assert!(view.try_load_entry(&24).await.unwrap().is_none());
952 /// # })
953 /// ```
954 pub async fn try_load_entry<Q>(
955 &self,
956 index: &Q,
957 ) -> Result<Option<ReadGuardedView<'_, W>>, ViewError>
958 where
959 I: Borrow<Q>,
960 Q: Serialize + ?Sized,
961 {
962 let short_key = BaseKey::derive_short_key(index)?;
963 self.collection.try_load_entry(&short_key).await
964 }
965
966 /// Load multiple entries for reading at once.
967 /// The entries in indices have to be all distinct.
968 /// ```rust
969 /// # tokio_test::block_on(async {
970 /// # use linera_views::context::MemoryContext;
971 /// # use linera_views::collection_view::CollectionView;
972 /// # use linera_views::register_view::RegisterView;
973 /// # use linera_views::views::View;
974 /// # let context = MemoryContext::new_for_testing(());
975 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
976 /// CollectionView::load(context).await.unwrap();
977 /// {
978 /// let _subview = view.load_entry_mut(&23).await.unwrap();
979 /// }
980 /// let indices = vec![23, 24];
981 /// let subviews = view.try_load_entries(&indices).await.unwrap();
982 /// let value0 = subviews[0].as_ref().unwrap().get();
983 /// assert_eq!(*value0, String::default());
984 /// # })
985 /// ```
986 pub async fn try_load_entries<'a, Q>(
987 &self,
988 indices: impl IntoIterator<Item = &'a Q>,
989 ) -> Result<Vec<Option<ReadGuardedView<'_, W>>>, ViewError>
990 where
991 I: Borrow<Q>,
992 Q: Serialize + 'a,
993 {
994 let short_keys = indices
995 .into_iter()
996 .map(|index| BaseKey::derive_short_key(index))
997 .collect::<Result<_, _>>()?;
998 self.collection.try_load_entries(short_keys).await
999 }
1000
1001 /// Loads multiple entries for reading at once with their keys.
1002 /// The entries in indices have to be all distinct.
1003 /// ```rust
1004 /// # tokio_test::block_on(async {
1005 /// # use linera_views::context::MemoryContext;
1006 /// # use linera_views::collection_view::CollectionView;
1007 /// # use linera_views::register_view::RegisterView;
1008 /// # use linera_views::views::View;
1009 /// # let context = MemoryContext::new_for_testing(());
1010 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1011 /// CollectionView::load(context).await.unwrap();
1012 /// {
1013 /// let _subview = view.load_entry_mut(&23).await.unwrap();
1014 /// }
1015 /// let indices = [23, 24];
1016 /// let subviews = view.try_load_entries_pairs(indices).await.unwrap();
1017 /// let value0 = subviews[0].1.as_ref().unwrap().get();
1018 /// assert_eq!(*value0, String::default());
1019 /// # })
1020 /// ```
1021 pub async fn try_load_entries_pairs<Q>(
1022 &self,
1023 indices: impl IntoIterator<Item = Q>,
1024 ) -> Result<Vec<(Q, Option<ReadGuardedView<'_, W>>)>, ViewError>
1025 where
1026 I: Borrow<Q>,
1027 Q: Serialize + Clone,
1028 {
1029 let indices_vec: Vec<Q> = indices.into_iter().collect();
1030 let values = self.try_load_entries(indices_vec.iter()).await?;
1031 Ok(indices_vec.into_iter().zip(values).collect())
1032 }
1033
1034 /// Load all entries for reading at once.
1035 /// ```rust
1036 /// # tokio_test::block_on(async {
1037 /// # use linera_views::context::MemoryContext;
1038 /// # use linera_views::collection_view::CollectionView;
1039 /// # use linera_views::register_view::RegisterView;
1040 /// # use linera_views::views::View;
1041 /// # let context = MemoryContext::new_for_testing(());
1042 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1043 /// CollectionView::load(context).await.unwrap();
1044 /// {
1045 /// let _subview = view.load_entry_mut(&23).await.unwrap();
1046 /// }
1047 /// let subviews = view.try_load_all_entries().await.unwrap();
1048 /// assert_eq!(subviews.len(), 1);
1049 /// # })
1050 /// ```
1051 pub async fn try_load_all_entries(&self) -> Result<Vec<(I, ReadGuardedView<'_, W>)>, ViewError>
1052 where
1053 I: DeserializeOwned,
1054 {
1055 let results = self.collection.try_load_all_entries().await?;
1056 results
1057 .into_iter()
1058 .map(|(short_key, view)| {
1059 let index = BaseKey::deserialize_value(&short_key)?;
1060 Ok((index, view))
1061 })
1062 .collect()
1063 }
1064
1065 /// Resets an entry to the default value.
1066 /// ```rust
1067 /// # tokio_test::block_on(async {
1068 /// # use linera_views::context::MemoryContext;
1069 /// # use linera_views::collection_view::CollectionView;
1070 /// # use linera_views::register_view::RegisterView;
1071 /// # use linera_views::views::View;
1072 /// # let context = MemoryContext::new_for_testing(());
1073 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1074 /// CollectionView::load(context).await.unwrap();
1075 /// let subview = view.load_entry_mut(&23).await.unwrap();
1076 /// let value = subview.get_mut();
1077 /// *value = String::from("Hello");
1078 /// view.reset_entry_to_default(&23).unwrap();
1079 /// let subview = view.load_entry_mut(&23).await.unwrap();
1080 /// let value = subview.get_mut();
1081 /// assert_eq!(*value, String::default());
1082 /// # })
1083 /// ```
1084 pub fn reset_entry_to_default<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1085 where
1086 I: Borrow<Q>,
1087 Q: Serialize + ?Sized,
1088 {
1089 let short_key = BaseKey::derive_short_key(index)?;
1090 self.collection.reset_entry_to_default(&short_key)
1091 }
1092
1093 /// Removes an entry from the `CollectionView`. If absent nothing happens.
1094 /// ```rust
1095 /// # tokio_test::block_on(async {
1096 /// # use linera_views::context::MemoryContext;
1097 /// # use linera_views::collection_view::CollectionView;
1098 /// # use linera_views::register_view::RegisterView;
1099 /// # use linera_views::views::View;
1100 /// # let context = MemoryContext::new_for_testing(());
1101 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1102 /// CollectionView::load(context).await.unwrap();
1103 /// let subview = view.load_entry_mut(&23).await.unwrap();
1104 /// let value = subview.get_mut();
1105 /// assert_eq!(*value, String::default());
1106 /// view.remove_entry(&23);
1107 /// let keys = view.indices().await.unwrap();
1108 /// assert_eq!(keys.len(), 0);
1109 /// # })
1110 /// ```
1111 pub fn remove_entry<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1112 where
1113 I: Borrow<Q>,
1114 Q: Serialize + ?Sized,
1115 {
1116 let short_key = BaseKey::derive_short_key(index)?;
1117 self.collection.remove_entry(short_key);
1118 Ok(())
1119 }
1120
1121 /// Gets the extra data.
1122 pub fn extra(&self) -> &<W::Context as Context>::Extra {
1123 self.collection.extra()
1124 }
1125}
1126
1127impl<I, W: View> CollectionView<W::Context, I, W>
1128where
1129 I: Sync + Send + Serialize + DeserializeOwned,
1130{
1131 /// Returns the list of indices in the collection in the order determined by
1132 /// the serialization.
1133 /// ```rust
1134 /// # tokio_test::block_on(async {
1135 /// # use linera_views::context::MemoryContext;
1136 /// # use linera_views::collection_view::CollectionView;
1137 /// # use linera_views::register_view::RegisterView;
1138 /// # use linera_views::views::View;
1139 /// # let context = MemoryContext::new_for_testing(());
1140 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1141 /// CollectionView::load(context).await.unwrap();
1142 /// view.load_entry_mut(&23).await.unwrap();
1143 /// view.load_entry_mut(&25).await.unwrap();
1144 /// let indices = view.indices().await.unwrap();
1145 /// assert_eq!(indices.len(), 2);
1146 /// # })
1147 /// ```
1148 pub async fn indices(&self) -> Result<Vec<I>, ViewError> {
1149 let mut indices = Vec::new();
1150 self.for_each_index(|index| {
1151 indices.push(index);
1152 Ok(())
1153 })
1154 .await?;
1155 Ok(indices)
1156 }
1157
1158 /// Returns the number of entries in the collection.
1159 /// ```rust
1160 /// # tokio_test::block_on(async {
1161 /// # use linera_views::context::MemoryContext;
1162 /// # use linera_views::collection_view::CollectionView;
1163 /// # use linera_views::register_view::RegisterView;
1164 /// # use linera_views::views::View;
1165 /// # let context = MemoryContext::new_for_testing(());
1166 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1167 /// CollectionView::load(context).await.unwrap();
1168 /// view.load_entry_mut(&23).await.unwrap();
1169 /// view.load_entry_mut(&25).await.unwrap();
1170 /// assert_eq!(view.iterative_count().await.unwrap(), 2);
1171 /// # })
1172 /// ```
1173 pub async fn iterative_count(&self) -> Result<usize, ViewError> {
1174 self.collection.iterative_count().await
1175 }
1176}
1177
1178impl<I: DeserializeOwned, W: View> CollectionView<W::Context, I, W> {
1179 /// Applies a function f on each index. Indices are visited in an order
1180 /// determined by the serialization. If the function returns false then
1181 /// the loop ends prematurely.
1182 /// ```rust
1183 /// # tokio_test::block_on(async {
1184 /// # use linera_views::context::MemoryContext;
1185 /// # use linera_views::collection_view::CollectionView;
1186 /// # use linera_views::register_view::RegisterView;
1187 /// # use linera_views::views::View;
1188 /// # let context = MemoryContext::new_for_testing(());
1189 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1190 /// CollectionView::load(context).await.unwrap();
1191 /// view.load_entry_mut(&23).await.unwrap();
1192 /// view.load_entry_mut(&24).await.unwrap();
1193 /// let mut count = 0;
1194 /// view.for_each_index_while(|_key| {
1195 /// count += 1;
1196 /// Ok(count < 1)
1197 /// })
1198 /// .await
1199 /// .unwrap();
1200 /// assert_eq!(count, 1);
1201 /// # })
1202 /// ```
1203 pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
1204 where
1205 F: FnMut(I) -> Result<bool, ViewError> + Send,
1206 {
1207 self.collection
1208 .for_each_key_while(|key| {
1209 let index = BaseKey::deserialize_value(key)?;
1210 f(index)
1211 })
1212 .await?;
1213 Ok(())
1214 }
1215
1216 /// Applies a function f on each index. Indices are visited in an order
1217 /// determined by the serialization.
1218 /// ```rust
1219 /// # tokio_test::block_on(async {
1220 /// # use linera_views::context::MemoryContext;
1221 /// # use linera_views::collection_view::CollectionView;
1222 /// # use linera_views::register_view::RegisterView;
1223 /// # use linera_views::views::View;
1224 /// # let context = MemoryContext::new_for_testing(());
1225 /// let mut view: CollectionView<_, u64, RegisterView<_, String>> =
1226 /// CollectionView::load(context).await.unwrap();
1227 /// view.load_entry_mut(&23).await.unwrap();
1228 /// view.load_entry_mut(&28).await.unwrap();
1229 /// let mut count = 0;
1230 /// view.for_each_index(|_key| {
1231 /// count += 1;
1232 /// Ok(())
1233 /// })
1234 /// .await
1235 /// .unwrap();
1236 /// assert_eq!(count, 2);
1237 /// # })
1238 /// ```
1239 pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
1240 where
1241 F: FnMut(I) -> Result<(), ViewError> + Send,
1242 {
1243 self.collection
1244 .for_each_key(|key| {
1245 let index = BaseKey::deserialize_value(key)?;
1246 f(index)
1247 })
1248 .await?;
1249 Ok(())
1250 }
1251}
1252
1253impl<I, W: HashableView> HashableView for CollectionView<W::Context, I, W>
1254where
1255 I: Send + Sync + Serialize + DeserializeOwned,
1256{
1257 type Hasher = sha3::Sha3_256;
1258
1259 async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1260 self.collection.hash_mut().await
1261 }
1262
1263 async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1264 self.collection.hash().await
1265 }
1266}
1267
1268/// A map view that serializes the indices.
1269#[derive(Debug, Allocative)]
1270#[allocative(bound = "C, I, W: Allocative")]
1271pub struct CustomCollectionView<C, I, W> {
1272 collection: ByteCollectionView<C, W>,
1273 #[allocative(skip)]
1274 _phantom: PhantomData<I>,
1275}
1276
1277impl<I: Send + Sync, W: View> View for CustomCollectionView<W::Context, I, W> {
1278 const NUM_INIT_KEYS: usize = ByteCollectionView::<W::Context, W>::NUM_INIT_KEYS;
1279
1280 type Context = W::Context;
1281
1282 fn context(&self) -> Self::Context {
1283 self.collection.context()
1284 }
1285
1286 fn pre_load(context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
1287 ByteCollectionView::<_, W>::pre_load(context)
1288 }
1289
1290 fn post_load(context: Self::Context, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
1291 let collection = ByteCollectionView::post_load(context, values)?;
1292 Ok(CustomCollectionView {
1293 collection,
1294 _phantom: PhantomData,
1295 })
1296 }
1297
1298 fn rollback(&mut self) {
1299 self.collection.rollback()
1300 }
1301
1302 async fn has_pending_changes(&self) -> bool {
1303 self.collection.has_pending_changes().await
1304 }
1305
1306 fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
1307 self.collection.pre_save(batch)
1308 }
1309
1310 fn post_save(&mut self) {
1311 self.collection.post_save()
1312 }
1313
1314 fn clear(&mut self) {
1315 self.collection.clear()
1316 }
1317}
1318
1319impl<I: Send + Sync, W: ClonableView> ClonableView for CustomCollectionView<W::Context, I, W> {
1320 fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
1321 Ok(CustomCollectionView {
1322 collection: self.collection.clone_unchecked()?,
1323 _phantom: PhantomData,
1324 })
1325 }
1326}
1327
1328impl<I: CustomSerialize, W: View> CustomCollectionView<W::Context, I, W> {
1329 /// Loads a subview for the data at the given index in the collection. If an entry
1330 /// is absent then a default entry is added to the collection. The resulting view
1331 /// can be modified.
1332 /// ```rust
1333 /// # tokio_test::block_on(async {
1334 /// # use linera_views::context::MemoryContext;
1335 /// # use linera_views::collection_view::CustomCollectionView;
1336 /// # use linera_views::register_view::RegisterView;
1337 /// # use linera_views::views::View;
1338 /// # let context = MemoryContext::new_for_testing(());
1339 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1340 /// CustomCollectionView::load(context).await.unwrap();
1341 /// let subview = view.load_entry_mut(&23).await.unwrap();
1342 /// let value = subview.get();
1343 /// assert_eq!(*value, String::default());
1344 /// # })
1345 /// ```
1346 pub async fn load_entry_mut<Q>(&mut self, index: &Q) -> Result<&mut W, ViewError>
1347 where
1348 I: Borrow<Q>,
1349 Q: CustomSerialize,
1350 {
1351 let short_key = index.to_custom_bytes()?;
1352 self.collection.load_entry_mut(&short_key).await
1353 }
1354
1355 /// Loads a subview for the data at the given index in the collection. If an entry
1356 /// is absent then `None` is returned. The resulting view cannot be modified.
1357 /// May fail if one subview is already being visited.
1358 /// ```rust
1359 /// # tokio_test::block_on(async {
1360 /// # use linera_views::context::MemoryContext;
1361 /// # use linera_views::collection_view::CustomCollectionView;
1362 /// # use linera_views::register_view::RegisterView;
1363 /// # use linera_views::views::View;
1364 /// # let context = MemoryContext::new_for_testing(());
1365 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1366 /// CustomCollectionView::load(context).await.unwrap();
1367 /// {
1368 /// let _subview = view.load_entry_mut(&23).await.unwrap();
1369 /// }
1370 /// {
1371 /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
1372 /// let value = subview.get();
1373 /// assert_eq!(*value, String::default());
1374 /// }
1375 /// assert!(view.try_load_entry(&24).await.unwrap().is_none());
1376 /// # })
1377 /// ```
1378 pub async fn try_load_entry<Q>(
1379 &self,
1380 index: &Q,
1381 ) -> Result<Option<ReadGuardedView<'_, W>>, ViewError>
1382 where
1383 I: Borrow<Q>,
1384 Q: CustomSerialize,
1385 {
1386 let short_key = index.to_custom_bytes()?;
1387 self.collection.try_load_entry(&short_key).await
1388 }
1389
1390 /// Load multiple entries for reading at once.
1391 /// The entries in indices have to be all distinct.
1392 /// ```rust
1393 /// # tokio_test::block_on(async {
1394 /// # use linera_views::context::MemoryContext;
1395 /// # use linera_views::collection_view::CustomCollectionView;
1396 /// # use linera_views::register_view::RegisterView;
1397 /// # use linera_views::views::View;
1398 /// # let context = MemoryContext::new_for_testing(());
1399 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1400 /// CustomCollectionView::load(context).await.unwrap();
1401 /// {
1402 /// let _subview = view.load_entry_mut(&23).await.unwrap();
1403 /// }
1404 /// let subviews = view.try_load_entries(&[23, 42]).await.unwrap();
1405 /// let value0 = subviews[0].as_ref().unwrap().get();
1406 /// assert_eq!(*value0, String::default());
1407 /// # })
1408 /// ```
1409 pub async fn try_load_entries<'a, Q>(
1410 &self,
1411 indices: impl IntoIterator<Item = &'a Q>,
1412 ) -> Result<Vec<Option<ReadGuardedView<'_, W>>>, ViewError>
1413 where
1414 I: Borrow<Q>,
1415 Q: CustomSerialize + 'a,
1416 {
1417 let short_keys = indices
1418 .into_iter()
1419 .map(|index| index.to_custom_bytes())
1420 .collect::<Result<_, _>>()?;
1421 self.collection.try_load_entries(short_keys).await
1422 }
1423
1424 /// Loads multiple entries for reading at once with their keys.
1425 /// The entries in indices have to be all distinct.
1426 /// ```rust
1427 /// # tokio_test::block_on(async {
1428 /// # use linera_views::context::MemoryContext;
1429 /// # use linera_views::collection_view::CustomCollectionView;
1430 /// # use linera_views::register_view::RegisterView;
1431 /// # use linera_views::views::View;
1432 /// # let context = MemoryContext::new_for_testing(());
1433 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1434 /// CustomCollectionView::load(context).await.unwrap();
1435 /// {
1436 /// let _subview = view.load_entry_mut(&23).await.unwrap();
1437 /// }
1438 /// let indices = [23, 42];
1439 /// let subviews = view.try_load_entries_pairs(indices).await.unwrap();
1440 /// let value0 = subviews[0].1.as_ref().unwrap().get();
1441 /// assert_eq!(*value0, String::default());
1442 /// # })
1443 /// ```
1444 pub async fn try_load_entries_pairs<Q>(
1445 &self,
1446 indices: impl IntoIterator<Item = Q>,
1447 ) -> Result<Vec<(Q, Option<ReadGuardedView<'_, W>>)>, ViewError>
1448 where
1449 I: Borrow<Q>,
1450 Q: CustomSerialize + Clone,
1451 {
1452 let indices_vec: Vec<Q> = indices.into_iter().collect();
1453 let values = self.try_load_entries(indices_vec.iter()).await?;
1454 Ok(indices_vec.into_iter().zip(values).collect())
1455 }
1456
1457 /// Load all entries for reading at once.
1458 /// ```rust
1459 /// # tokio_test::block_on(async {
1460 /// # use linera_views::context::MemoryContext;
1461 /// # use linera_views::collection_view::CustomCollectionView;
1462 /// # use linera_views::register_view::RegisterView;
1463 /// # use linera_views::views::View;
1464 /// # let context = MemoryContext::new_for_testing(());
1465 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1466 /// CustomCollectionView::load(context).await.unwrap();
1467 /// {
1468 /// let _subview = view.load_entry_mut(&23).await.unwrap();
1469 /// }
1470 /// let subviews = view.try_load_all_entries().await.unwrap();
1471 /// assert_eq!(subviews.len(), 1);
1472 /// # })
1473 /// ```
1474 pub async fn try_load_all_entries(&self) -> Result<Vec<(I, ReadGuardedView<'_, W>)>, ViewError>
1475 where
1476 I: CustomSerialize,
1477 {
1478 let results = self.collection.try_load_all_entries().await?;
1479 results
1480 .into_iter()
1481 .map(|(short_key, view)| {
1482 let index = I::from_custom_bytes(&short_key)?;
1483 Ok((index, view))
1484 })
1485 .collect()
1486 }
1487
1488 /// Marks the entry so that it is removed in the next flush.
1489 /// ```rust
1490 /// # tokio_test::block_on(async {
1491 /// # use linera_views::context::MemoryContext;
1492 /// # use linera_views::collection_view::CustomCollectionView;
1493 /// # use linera_views::register_view::RegisterView;
1494 /// # use linera_views::views::View;
1495 /// # let context = MemoryContext::new_for_testing(());
1496 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1497 /// CustomCollectionView::load(context).await.unwrap();
1498 /// let subview = view.load_entry_mut(&23).await.unwrap();
1499 /// let value = subview.get_mut();
1500 /// *value = String::from("Hello");
1501 /// view.reset_entry_to_default(&23).unwrap();
1502 /// let subview = view.load_entry_mut(&23).await.unwrap();
1503 /// let value = subview.get_mut();
1504 /// assert_eq!(*value, String::default());
1505 /// # })
1506 /// ```
1507 pub fn reset_entry_to_default<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1508 where
1509 I: Borrow<Q>,
1510 Q: CustomSerialize,
1511 {
1512 let short_key = index.to_custom_bytes()?;
1513 self.collection.reset_entry_to_default(&short_key)
1514 }
1515
1516 /// Removes an entry from the `CollectionView`. If absent nothing happens.
1517 /// ```rust
1518 /// # tokio_test::block_on(async {
1519 /// # use linera_views::context::MemoryContext;
1520 /// # use linera_views::collection_view::CustomCollectionView;
1521 /// # use linera_views::register_view::RegisterView;
1522 /// # use linera_views::views::View;
1523 /// # let context = MemoryContext::new_for_testing(());
1524 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1525 /// CustomCollectionView::load(context).await.unwrap();
1526 /// let subview = view.load_entry_mut(&23).await.unwrap();
1527 /// let value = subview.get_mut();
1528 /// assert_eq!(*value, String::default());
1529 /// view.remove_entry(&23);
1530 /// let keys = view.indices().await.unwrap();
1531 /// assert_eq!(keys.len(), 0);
1532 /// # })
1533 /// ```
1534 pub fn remove_entry<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1535 where
1536 I: Borrow<Q>,
1537 Q: CustomSerialize,
1538 {
1539 let short_key = index.to_custom_bytes()?;
1540 self.collection.remove_entry(short_key);
1541 Ok(())
1542 }
1543
1544 /// Gets the extra data.
1545 pub fn extra(&self) -> &<W::Context as Context>::Extra {
1546 self.collection.extra()
1547 }
1548}
1549
1550impl<I: CustomSerialize + Send, W: View> CustomCollectionView<W::Context, I, W> {
1551 /// Returns the list of indices in the collection in the order determined by the custom serialization.
1552 /// ```rust
1553 /// # tokio_test::block_on(async {
1554 /// # use linera_views::context::MemoryContext;
1555 /// # use linera_views::collection_view::CustomCollectionView;
1556 /// # use linera_views::register_view::RegisterView;
1557 /// # use linera_views::views::View;
1558 /// # let context = MemoryContext::new_for_testing(());
1559 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1560 /// CustomCollectionView::load(context).await.unwrap();
1561 /// view.load_entry_mut(&23).await.unwrap();
1562 /// view.load_entry_mut(&25).await.unwrap();
1563 /// let indices = view.indices().await.unwrap();
1564 /// assert_eq!(indices, vec![23, 25]);
1565 /// # })
1566 /// ```
1567 pub async fn indices(&self) -> Result<Vec<I>, ViewError> {
1568 let mut indices = Vec::new();
1569 self.for_each_index(|index| {
1570 indices.push(index);
1571 Ok(())
1572 })
1573 .await?;
1574 Ok(indices)
1575 }
1576
1577 /// Returns the number of entries in the collection.
1578 /// ```rust
1579 /// # tokio_test::block_on(async {
1580 /// # use linera_views::context::MemoryContext;
1581 /// # use linera_views::collection_view::CustomCollectionView;
1582 /// # use linera_views::register_view::RegisterView;
1583 /// # use linera_views::views::View;
1584 /// # let context = MemoryContext::new_for_testing(());
1585 /// let mut view = CustomCollectionView::<_, u128, RegisterView<_, String>>::load(context)
1586 /// .await
1587 /// .unwrap();
1588 /// view.load_entry_mut(&(23 as u128)).await.unwrap();
1589 /// view.load_entry_mut(&(25 as u128)).await.unwrap();
1590 /// assert_eq!(view.iterative_count().await.unwrap(), 2);
1591 /// # })
1592 /// ```
1593 pub async fn iterative_count(&self) -> Result<usize, ViewError> {
1594 self.collection.iterative_count().await
1595 }
1596}
1597
1598impl<I: CustomSerialize, W: View> CustomCollectionView<W::Context, I, W> {
1599 /// Applies a function f on each index. Indices are visited in an order
1600 /// determined by the custom serialization. If the function f returns false,
1601 /// then the loop ends prematurely.
1602 /// ```rust
1603 /// # tokio_test::block_on(async {
1604 /// # use linera_views::context::MemoryContext;
1605 /// # use linera_views::collection_view::CustomCollectionView;
1606 /// # use linera_views::register_view::RegisterView;
1607 /// # use linera_views::views::View;
1608 /// # let context = MemoryContext::new_for_testing(());
1609 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1610 /// CustomCollectionView::load(context).await.unwrap();
1611 /// view.load_entry_mut(&28).await.unwrap();
1612 /// view.load_entry_mut(&24).await.unwrap();
1613 /// view.load_entry_mut(&23).await.unwrap();
1614 /// let mut part_indices = Vec::new();
1615 /// view.for_each_index_while(|index| {
1616 /// part_indices.push(index);
1617 /// Ok(part_indices.len() < 2)
1618 /// })
1619 /// .await
1620 /// .unwrap();
1621 /// assert_eq!(part_indices, vec![23, 24]);
1622 /// # })
1623 /// ```
1624 pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
1625 where
1626 F: FnMut(I) -> Result<bool, ViewError> + Send,
1627 {
1628 self.collection
1629 .for_each_key_while(|key| {
1630 let index = I::from_custom_bytes(key)?;
1631 f(index)
1632 })
1633 .await?;
1634 Ok(())
1635 }
1636
1637 /// Applies a function on each index. Indices are visited in an order
1638 /// determined by the custom serialization.
1639 /// ```rust
1640 /// # tokio_test::block_on(async {
1641 /// # use linera_views::context::MemoryContext;
1642 /// # use linera_views::collection_view::CustomCollectionView;
1643 /// # use linera_views::register_view::RegisterView;
1644 /// # use linera_views::views::View;
1645 /// # let context = MemoryContext::new_for_testing(());
1646 /// let mut view: CustomCollectionView<_, u128, RegisterView<_, String>> =
1647 /// CustomCollectionView::load(context).await.unwrap();
1648 /// view.load_entry_mut(&28).await.unwrap();
1649 /// view.load_entry_mut(&24).await.unwrap();
1650 /// view.load_entry_mut(&23).await.unwrap();
1651 /// let mut indices = Vec::new();
1652 /// view.for_each_index(|index| {
1653 /// indices.push(index);
1654 /// Ok(())
1655 /// })
1656 /// .await
1657 /// .unwrap();
1658 /// assert_eq!(indices, vec![23, 24, 28]);
1659 /// # })
1660 /// ```
1661 pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
1662 where
1663 F: FnMut(I) -> Result<(), ViewError> + Send,
1664 {
1665 self.collection
1666 .for_each_key(|key| {
1667 let index = I::from_custom_bytes(key)?;
1668 f(index)
1669 })
1670 .await?;
1671 Ok(())
1672 }
1673}
1674
1675impl<I, W: HashableView> HashableView for CustomCollectionView<W::Context, I, W>
1676where
1677 Self: View,
1678{
1679 type Hasher = sha3::Sha3_256;
1680
1681 async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1682 self.collection.hash_mut().await
1683 }
1684
1685 async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1686 self.collection.hash().await
1687 }
1688}
1689
1690/// Type wrapping `ByteCollectionView` while memoizing the hash.
1691pub type HashedByteCollectionView<C, W> =
1692 WrappedHashableContainerView<C, ByteCollectionView<C, W>, HasherOutput>;
1693
1694/// Wrapper around `ByteCollectionView` to compute hashes based on the history of changes.
1695pub type HistoricallyHashedByteCollectionView<C, W> =
1696 HistoricallyHashableView<C, ByteCollectionView<C, W>>;
1697
1698/// Type wrapping `CollectionView` while memoizing the hash.
1699pub type HashedCollectionView<C, I, W> =
1700 WrappedHashableContainerView<C, CollectionView<C, I, W>, HasherOutput>;
1701
1702/// Wrapper around `CollectionView` to compute hashes based on the history of changes.
1703pub type HistoricallyHashedCollectionView<C, I, W> =
1704 HistoricallyHashableView<C, CollectionView<C, I, W>>;
1705
1706/// Type wrapping `CustomCollectionView` while memoizing the hash.
1707pub type HashedCustomCollectionView<C, I, W> =
1708 WrappedHashableContainerView<C, CustomCollectionView<C, I, W>, HasherOutput>;
1709
1710/// Wrapper around `CustomCollectionView` to compute hashes based on the history of changes.
1711pub type HistoricallyHashedCustomCollectionView<C, I, W> =
1712 HistoricallyHashableView<C, CustomCollectionView<C, I, W>>;
1713
1714#[cfg(with_graphql)]
1715mod graphql {
1716 use std::borrow::Cow;
1717
1718 use super::{CollectionView, CustomCollectionView, ReadGuardedView};
1719 use crate::{
1720 graphql::{hash_name, mangle, missing_key_error, Entry, MapInput},
1721 views::View,
1722 };
1723
1724 impl<T: async_graphql::OutputType> async_graphql::OutputType for ReadGuardedView<'_, T> {
1725 fn type_name() -> Cow<'static, str> {
1726 T::type_name()
1727 }
1728
1729 fn create_type_info(registry: &mut async_graphql::registry::Registry) -> String {
1730 T::create_type_info(registry)
1731 }
1732
1733 async fn resolve(
1734 &self,
1735 ctx: &async_graphql::ContextSelectionSet<'_>,
1736 field: &async_graphql::Positioned<async_graphql::parser::types::Field>,
1737 ) -> async_graphql::ServerResult<async_graphql::Value> {
1738 (**self).resolve(ctx, field).await
1739 }
1740 }
1741
1742 impl<C: Send + Sync, K: async_graphql::OutputType, V: async_graphql::OutputType>
1743 async_graphql::TypeName for CollectionView<C, K, V>
1744 {
1745 fn type_name() -> Cow<'static, str> {
1746 format!(
1747 "CollectionView_{}_{}_{:08x}",
1748 mangle(K::type_name()),
1749 mangle(V::type_name()),
1750 hash_name::<(K, V)>(),
1751 )
1752 .into()
1753 }
1754 }
1755
1756 #[async_graphql::Object(cache_control(no_cache), name_type)]
1757 impl<K, V> CollectionView<V::Context, K, V>
1758 where
1759 K: async_graphql::InputType
1760 + async_graphql::OutputType
1761 + serde::ser::Serialize
1762 + serde::de::DeserializeOwned
1763 + std::fmt::Debug,
1764 V: View + async_graphql::OutputType,
1765 {
1766 async fn keys(&self) -> Result<Vec<K>, async_graphql::Error> {
1767 Ok(self.indices().await?)
1768 }
1769
1770 #[graphql(derived(name = "count"))]
1771 async fn count_(&self) -> Result<u32, async_graphql::Error> {
1772 let count = self.iterative_count().await?;
1773 u32::try_from(count).map_err(|_| async_graphql::Error::new("count exceeds u32"))
1774 }
1775
1776 async fn entry(
1777 &self,
1778 key: K,
1779 ) -> Result<Entry<K, ReadGuardedView<'_, V>>, async_graphql::Error> {
1780 let value = self
1781 .try_load_entry(&key)
1782 .await?
1783 .ok_or_else(|| missing_key_error(&key))?;
1784 Ok(Entry { value, key })
1785 }
1786
1787 async fn entries(
1788 &self,
1789 input: Option<MapInput<K>>,
1790 ) -> Result<Vec<Entry<K, ReadGuardedView<'_, V>>>, async_graphql::Error> {
1791 let keys = if let Some(keys) = input
1792 .and_then(|input| input.filters)
1793 .and_then(|filters| filters.keys)
1794 {
1795 keys
1796 } else {
1797 self.indices().await?
1798 };
1799
1800 let values = self.try_load_entries(&keys).await?;
1801 Ok(values
1802 .into_iter()
1803 .zip(keys)
1804 .filter_map(|(value, key)| value.map(|value| Entry { value, key }))
1805 .collect())
1806 }
1807 }
1808
1809 impl<C: Send + Sync, K: async_graphql::InputType, V: async_graphql::OutputType>
1810 async_graphql::TypeName for CustomCollectionView<C, K, V>
1811 {
1812 fn type_name() -> Cow<'static, str> {
1813 format!(
1814 "CustomCollectionView_{}_{}_{:08x}",
1815 mangle(K::type_name()),
1816 mangle(V::type_name()),
1817 hash_name::<(K, V)>(),
1818 )
1819 .into()
1820 }
1821 }
1822
1823 #[async_graphql::Object(cache_control(no_cache), name_type)]
1824 impl<K, V> CustomCollectionView<V::Context, K, V>
1825 where
1826 K: async_graphql::InputType
1827 + async_graphql::OutputType
1828 + crate::common::CustomSerialize
1829 + std::fmt::Debug,
1830 V: View + async_graphql::OutputType,
1831 {
1832 async fn keys(&self) -> Result<Vec<K>, async_graphql::Error> {
1833 Ok(self.indices().await?)
1834 }
1835
1836 #[graphql(derived(name = "count"))]
1837 async fn count_(&self) -> Result<u32, async_graphql::Error> {
1838 let count = self.iterative_count().await?;
1839 u32::try_from(count).map_err(|_| async_graphql::Error::new("count exceeds u32"))
1840 }
1841
1842 async fn entry(
1843 &self,
1844 key: K,
1845 ) -> Result<Entry<K, ReadGuardedView<'_, V>>, async_graphql::Error> {
1846 let value = self
1847 .try_load_entry(&key)
1848 .await?
1849 .ok_or_else(|| missing_key_error(&key))?;
1850 Ok(Entry { value, key })
1851 }
1852
1853 async fn entries(
1854 &self,
1855 input: Option<MapInput<K>>,
1856 ) -> Result<Vec<Entry<K, ReadGuardedView<'_, V>>>, async_graphql::Error> {
1857 let keys = if let Some(keys) = input
1858 .and_then(|input| input.filters)
1859 .and_then(|filters| filters.keys)
1860 {
1861 keys
1862 } else {
1863 self.indices().await?
1864 };
1865
1866 let values = self.try_load_entries(&keys).await?;
1867 Ok(values
1868 .into_iter()
1869 .zip(keys)
1870 .filter_map(|(value, key)| value.map(|value| Entry { value, key }))
1871 .collect())
1872 }
1873 }
1874}