linera_views/views/

reentrant_collection_view.rs

// Copyright (c) Zefchain Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use std::{
    borrow::Borrow,
    collections::{btree_map, BTreeMap},
    io::Write,
    marker::PhantomData,
    mem,
    ops::Deref,
    sync::Arc,
};

use allocative::{Allocative, Key, Visitor};
use async_lock::{RwLock, RwLockReadGuardArc, RwLockWriteGuardArc};
#[cfg(with_metrics)]
use linera_base::prometheus_util::MeasureLatency as _;
use serde::{de::DeserializeOwned, Serialize};

use crate::{
    batch::Batch,
    common::{CustomSerialize, HasherOutput, SliceExt as _, Update},
    context::{BaseKey, Context},
    hashable_wrapper::WrappedHashableContainerView,
    historical_hash_wrapper::HistoricallyHashableView,
    store::ReadableKeyValueStore as _,
    views::{ClonableView, HashableView, Hasher, ReplaceContext, View, ViewError, MIN_VIEW_TAG},
};

#[cfg(with_metrics)]
mod metrics {
    use std::sync::LazyLock;

    use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram_vec};
    use prometheus::HistogramVec;

    /// The runtime of hash computation
    pub static REENTRANT_COLLECTION_VIEW_HASH_RUNTIME: LazyLock<HistogramVec> =
        LazyLock::new(|| {
            register_histogram_vec(
                "reentrant_collection_view_hash_runtime",
                "ReentrantCollectionView hash runtime",
                &[],
                exponential_bucket_latencies(5.0),
            )
        });
}

/// A read-only accessor for a particular subview in a [`ReentrantCollectionView`].
#[derive(Debug)]
pub struct ReadGuardedView<T>(RwLockReadGuardArc<T>);

impl<T> std::ops::Deref for ReadGuardedView<T> {
    type Target = T;
    fn deref(&self) -> &T {
        self.0.deref()
    }
}

/// A read-write accessor for a particular subview in a [`ReentrantCollectionView`].
#[derive(Debug)]
pub struct WriteGuardedView<T>(RwLockWriteGuardArc<T>);

impl<T> std::ops::Deref for WriteGuardedView<T> {
    type Target = T;
    fn deref(&self) -> &T {
        self.0.deref()
    }
}

impl<T> std::ops::DerefMut for WriteGuardedView<T> {
    fn deref_mut(&mut self) -> &mut T {
        self.0.deref_mut()
    }
}

/// A view that supports accessing a collection of views of the same kind, indexed by `Vec<u8>`,
/// possibly several subviews at a time.
#[derive(Debug)]
pub struct ReentrantByteCollectionView<C, W> {
    /// The view [`Context`].
    context: C,
    /// If the current persisted data will be completely erased and replaced on the next flush.
    delete_storage_first: bool,
    /// Entries that may have staged changes.
    updates: BTreeMap<Vec<u8>, Update<Arc<RwLock<W>>>>,
}

impl<C, W: Allocative> Allocative for ReentrantByteCollectionView<C, W> {
    fn visit<'a, 'b: 'a>(&self, visitor: &'a mut Visitor<'b>) {
        let name = Key::new("ReentrantByteCollectionView");
        let size = mem::size_of::<Self>();
        let mut visitor = visitor.enter(name, size);

        for (k, v) in &self.updates {
            let key_name = Key::new("key");
            visitor.visit_field(key_name, k);
            match v {
                Update::Removed => {
                    let key = Key::new("update_removed");
                    visitor.visit_field(key, &());
                }
                Update::Set(v) => {
                    if let Some(v) = v.try_read() {
                        let key = Key::new("update_set");
                        visitor.visit_field(key, v.deref());
                    }
                }
            }
        }
        visitor.exit();
    }
}

impl<W, C2> ReplaceContext<C2> for ReentrantByteCollectionView<W::Context, W>
where
    W: View + ReplaceContext<C2>,
    C2: Context,
{
    type Target = ReentrantByteCollectionView<C2, <W as ReplaceContext<C2>>::Target>;

    async fn with_context(
        &mut self,
        ctx: impl FnOnce(&Self::Context) -> C2 + Clone,
    ) -> Self::Target {
        let mut updates: BTreeMap<_, Update<Arc<RwLock<W::Target>>>> = BTreeMap::new();
        for (key, update) in &self.updates {
            let new_value = match update {
                Update::Removed => Update::Removed,
                Update::Set(x) => Update::Set(Arc::new(RwLock::new(
                    x.write().await.with_context(ctx.clone()).await,
                ))),
            };
            updates.insert(key.clone(), new_value);
        }
        ReentrantByteCollectionView {
            context: ctx(&self.context),
            delete_storage_first: self.delete_storage_first,
            updates,
        }
    }
}

/// We need to find new base keys in order to implement the collection view.
/// We do this by appending a value to the base key.
///
/// Sub-views in a collection share a common key prefix, like in other view types. However,
/// just concatenating the shared prefix with sub-view keys makes it impossible to distinguish if a
/// given key belongs to a child sub-view or a grandchild sub-view (consider for example if a
/// collection is stored inside the collection).
#[repr(u8)]
enum KeyTag {
    /// Prefix for specifying an index and serves to indicate the existence of an entry in the collection.
    Index = MIN_VIEW_TAG,
    /// Prefix for specifying as the prefix for the sub-view.
    Subview,
}

impl<W: View> View for ReentrantByteCollectionView<W::Context, W> {
    const NUM_INIT_KEYS: usize = 0;

    type Context = W::Context;

    fn context(&self) -> Self::Context {
        self.context.clone()
    }

    fn pre_load(_context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
        Ok(Vec::new())
    }

    fn post_load(context: Self::Context, _values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
        Ok(Self {
            context,
            delete_storage_first: false,
            updates: BTreeMap::new(),
        })
    }

    fn rollback(&mut self) {
        self.delete_storage_first = false;
        self.updates.clear();
    }

    async fn has_pending_changes(&self) -> bool {
        if self.delete_storage_first {
            return true;
        }
        !self.updates.is_empty()
    }

    fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
        let mut delete_view = false;
        if self.delete_storage_first {
            delete_view = true;
            batch.delete_key_prefix(self.context.base_key().bytes.clone());
            for (index, update) in &self.updates {
                if let Update::Set(view) = update {
                    let view = view
                        .try_read()
                        .ok_or_else(|| ViewError::TryLockError(index.clone()))?;
                    view.pre_save(batch)?;
                    self.add_index(batch, index);
                    delete_view = false;
                }
            }
        } else {
            for (index, update) in &self.updates {
                match update {
                    Update::Set(view) => {
                        let view = view
                            .try_read()
                            .ok_or_else(|| ViewError::TryLockError(index.clone()))?;
                        view.pre_save(batch)?;
                        self.add_index(batch, index);
                    }
                    Update::Removed => {
                        let key_subview = self.get_subview_key(index);
                        let key_index = self.get_index_key(index);
                        batch.delete_key(key_index);
                        batch.delete_key_prefix(key_subview);
                    }
                }
            }
        }
        Ok(delete_view)
    }

    fn post_save(&mut self) {
        for (_index, update) in mem::take(&mut self.updates) {
            if let Update::Set(view) = update {
                let mut view = view.try_write().expect("pre_save was called before");
                view.post_save();
            }
        }
        self.delete_storage_first = false;
    }

    fn clear(&mut self) {
        self.delete_storage_first = true;
        self.updates.clear();
    }
}

impl<W: ClonableView> ClonableView for ReentrantByteCollectionView<W::Context, W> {
    fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
        let cloned_updates = self
            .updates
            .iter()
            .map(|(key, value)| {
                let cloned_value = match value {
                    Update::Removed => Update::Removed,
                    Update::Set(view_lock) => {
                        let mut view = view_lock
                            .try_write()
                            .ok_or_else(|| ViewError::TryLockError(key.clone()))?;
                        Update::Set(Arc::new(RwLock::new(view.clone_unchecked()?)))
                    }
                };
                Ok::<_, ViewError>((key.clone(), cloned_value))
            })
            .collect::<Result<_, _>>()?;

        Ok(ReentrantByteCollectionView {
            context: self.context.clone(),
            delete_storage_first: self.delete_storage_first,
            updates: cloned_updates,
        })
    }
}

impl<C: Context, W> ReentrantByteCollectionView<C, W> {
    fn get_index_key(&self, index: &[u8]) -> Vec<u8> {
        self.context
            .base_key()
            .base_tag_index(KeyTag::Index as u8, index)
    }

    fn get_subview_key(&self, index: &[u8]) -> Vec<u8> {
        self.context
            .base_key()
            .base_tag_index(KeyTag::Subview as u8, index)
    }

    fn add_index(&self, batch: &mut Batch, index: &[u8]) {
        let key = self.get_index_key(index);
        batch.put_key_value_bytes(key, vec![]);
    }
}

impl<W: View> ReentrantByteCollectionView<W::Context, W> {
    /// Reads the view and if missing returns the default view
    async fn wrapped_view(
        context: &W::Context,
        delete_storage_first: bool,
        short_key: &[u8],
    ) -> Result<Arc<RwLock<W>>, ViewError> {
        let key = context
            .base_key()
            .base_tag_index(KeyTag::Subview as u8, short_key);
        let context = context.clone_with_base_key(key);
        // Obtain a view and set its pending state to the default (e.g. empty) state
        let view = if delete_storage_first {
            W::new(context)?
        } else {
            W::load(context).await?
        };
        Ok(Arc::new(RwLock::new(view)))
    }

    /// Load the view and insert it into the updates if needed.
    /// If the entry is missing, then it is set to default.
    async fn try_load_view_mut(&mut self, short_key: &[u8]) -> Result<Arc<RwLock<W>>, ViewError> {
        use btree_map::Entry::*;
        Ok(match self.updates.entry(short_key.to_owned()) {
            Occupied(mut entry) => match entry.get_mut() {
                Update::Set(view) => view.clone(),
                entry @ Update::Removed => {
                    let wrapped_view = Self::wrapped_view(&self.context, true, short_key).await?;
                    *entry = Update::Set(wrapped_view.clone());
                    wrapped_view
                }
            },
            Vacant(entry) => {
                let wrapped_view =
                    Self::wrapped_view(&self.context, self.delete_storage_first, short_key).await?;
                entry.insert(Update::Set(wrapped_view.clone()));
                wrapped_view
            }
        })
    }

    /// Load the view from the update is available.
    /// If missing, then the entry is loaded from storage and if
    /// missing there an error is reported.
    async fn try_load_view(&self, short_key: &[u8]) -> Result<Option<Arc<RwLock<W>>>, ViewError> {
        Ok(if let Some(entry) = self.updates.get(short_key) {
            match entry {
                Update::Set(view) => Some(view.clone()),
                _entry @ Update::Removed => None,
            }
        } else if self.delete_storage_first {
            None
        } else {
            let key_index = self
                .context
                .base_key()
                .base_tag_index(KeyTag::Index as u8, short_key);
            if self.context.store().contains_key(&key_index).await? {
                let view = Self::wrapped_view(&self.context, false, short_key).await?;
                Some(view)
            } else {
                None
            }
        })
    }

    /// Loads a subview for the data at the given index in the collection. If an entry
    /// is absent then a default entry is added to the collection. The resulting view
    /// can be modified.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// let subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// let value = subview.get();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub async fn try_load_entry_mut(
        &mut self,
        short_key: &[u8],
    ) -> Result<WriteGuardedView<W>, ViewError> {
        Ok(WriteGuardedView(
            self.try_load_view_mut(short_key)
                .await?
                .try_write_arc()
                .ok_or_else(|| ViewError::TryLockError(short_key.to_vec()))?,
        ))
    }

    /// Loads a subview at the given index in the collection and gives read-only access to the data.
    /// If an entry is absent then `None` is returned.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// }
    /// let subview = view.try_load_entry(&[0, 1]).await.unwrap().unwrap();
    /// let value = subview.get();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub async fn try_load_entry(
        &self,
        short_key: &[u8],
    ) -> Result<Option<ReadGuardedView<W>>, ViewError> {
        match self.try_load_view(short_key).await? {
            None => Ok(None),
            Some(view) => Ok(Some(ReadGuardedView(
                view.try_read_arc()
                    .ok_or_else(|| ViewError::TryLockError(short_key.to_vec()))?,
            ))),
        }
    }

    /// Returns `true` if the collection contains a value for the specified key.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// assert!(view.contains_key(&[0, 1]).await.unwrap());
    /// assert!(!view.contains_key(&[0, 2]).await.unwrap());
    /// # })
    /// ```
    pub async fn contains_key(&self, short_key: &[u8]) -> Result<bool, ViewError> {
        Ok(if let Some(entry) = self.updates.get(short_key) {
            match entry {
                Update::Set(_view) => true,
                Update::Removed => false,
            }
        } else if self.delete_storage_first {
            false
        } else {
            let key_index = self
                .context
                .base_key()
                .base_tag_index(KeyTag::Index as u8, short_key);
            self.context.store().contains_key(&key_index).await?
        })
    }

    /// Removes an entry. If absent then nothing happens.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// let value = subview.get_mut();
    /// assert_eq!(*value, String::default());
    /// view.remove_entry(vec![0, 1]);
    /// let keys = view.keys().await.unwrap();
    /// assert_eq!(keys.len(), 0);
    /// # })
    /// ```
    pub fn remove_entry(&mut self, short_key: Vec<u8>) {
        if self.delete_storage_first {
            // Optimization: No need to mark `short_key` for deletion as we are going to remove all the keys at once.
            self.updates.remove(&short_key);
        } else {
            self.updates.insert(short_key, Update::Removed);
        }
    }

    /// Marks the entry so that it is removed in the next flush.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    ///     let value = subview.get_mut();
    ///     *value = String::from("Hello");
    /// }
    /// view.try_reset_entry_to_default(&[0, 1]).unwrap();
    /// let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// let value = subview.get_mut();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub fn try_reset_entry_to_default(&mut self, short_key: &[u8]) -> Result<(), ViewError> {
        let key = self
            .context
            .base_key()
            .base_tag_index(KeyTag::Subview as u8, short_key);
        let context = self.context.clone_with_base_key(key);
        let view = W::new(context)?;
        let view = Arc::new(RwLock::new(view));
        let view = Update::Set(view);
        self.updates.insert(short_key.to_vec(), view);
        Ok(())
    }

    /// Gets the extra data.
    pub fn extra(&self) -> &<W::Context as Context>::Extra {
        self.context.extra()
    }
}

impl<W: View> ReentrantByteCollectionView<W::Context, W> {
    /// Loads multiple entries for writing at once.
    /// The entries in `short_keys` have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    ///     *subview.get_mut() = "Bonjour".to_string();
    /// }
    /// let short_keys = vec![vec![0, 1], vec![2, 3]];
    /// let subviews = view.try_load_entries_mut(short_keys).await.unwrap();
    /// let value1 = subviews[0].get();
    /// let value2 = subviews[1].get();
    /// assert_eq!(*value1, "Bonjour".to_string());
    /// assert_eq!(*value2, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_mut(
        &mut self,
        short_keys: Vec<Vec<u8>>,
    ) -> Result<Vec<WriteGuardedView<W>>, ViewError> {
        let mut short_keys_to_load = Vec::new();
        let mut keys = Vec::new();
        for short_key in &short_keys {
            let key = self
                .context
                .base_key()
                .base_tag_index(KeyTag::Subview as u8, short_key);
            let context = self.context.clone_with_base_key(key);
            match self.updates.entry(short_key.to_vec()) {
                btree_map::Entry::Occupied(mut entry) => {
                    if let Update::Removed = entry.get() {
                        let view = W::new(context)?;
                        let view = Arc::new(RwLock::new(view));
                        entry.insert(Update::Set(view));
                    }
                }
                btree_map::Entry::Vacant(entry) => {
                    if self.delete_storage_first {
                        let view = W::new(context)?;
                        let view = Arc::new(RwLock::new(view));
                        entry.insert(Update::Set(view));
                    } else {
                        keys.extend(W::pre_load(&context)?);
                        short_keys_to_load.push(short_key.to_vec());
                    }
                }
            }
        }
        let values = self.context.store().read_multi_values_bytes(&keys).await?;
        for (loaded_values, short_key) in values
            .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
            .zip(short_keys_to_load)
        {
            let key = self
                .context
                .base_key()
                .base_tag_index(KeyTag::Subview as u8, &short_key);
            let context = self.context.clone_with_base_key(key);
            let view = W::post_load(context, loaded_values)?;
            let wrapped_view = Arc::new(RwLock::new(view));
            self.updates
                .insert(short_key.to_vec(), Update::Set(wrapped_view));
        }

        short_keys
            .into_iter()
            .map(|short_key| {
                let Some(Update::Set(view)) = self.updates.get(&short_key) else {
                    unreachable!(
                        "Entry should have been inserted as Update::Set by try_load_view_mut"
                    )
                };
                Ok(WriteGuardedView(
                    view.clone()
                        .try_write_arc()
                        .ok_or_else(|| ViewError::TryLockError(short_key))?,
                ))
            })
            .collect()
    }

    /// Loads multiple entries for writing at once with their keys.
    /// The entries in short_keys have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let mut subview = view.try_load_entry_mut(&vec![0, 1]).await.unwrap();
    ///     *subview.get_mut() = "Bonjour".to_string();
    /// }
    /// let short_keys = vec![vec![0, 1], vec![2, 3]];
    /// let subviews = view.try_load_entries_pairs_mut(short_keys).await.unwrap();
    /// let value1 = subviews[0].1.get();
    /// let value2 = subviews[1].1.get();
    /// assert_eq!(*value1, "Bonjour".to_string());
    /// assert_eq!(*value2, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_pairs_mut(
        &mut self,
        short_keys: Vec<Vec<u8>>,
    ) -> Result<Vec<(Vec<u8>, WriteGuardedView<W>)>, ViewError> {
        let values = self.try_load_entries_mut(short_keys.clone()).await?;
        Ok(short_keys.into_iter().zip(values).collect())
    }

    /// Loads multiple entries for reading at once.
    /// The entries in `short_keys` have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// }
    /// let short_keys = vec![vec![0, 1], vec![2, 3]];
    /// let subviews = view.try_load_entries(short_keys).await.unwrap();
    /// assert!(subviews[1].is_none());
    /// let value0 = subviews[0].as_ref().unwrap().get();
    /// assert_eq!(*value0, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries(
        &self,
        short_keys: Vec<Vec<u8>>,
    ) -> Result<Vec<Option<ReadGuardedView<W>>>, ViewError> {
        let mut results = vec![None; short_keys.len()];
        let mut keys_to_check = Vec::new();
        let mut keys_to_check_metadata = Vec::new();

        for (position, short_key) in short_keys.into_iter().enumerate() {
            if let Some(update) = self.updates.get(&short_key) {
                if let Update::Set(view) = update {
                    results[position] = Some((short_key, view.clone()));
                }
            } else if !self.delete_storage_first {
                let key_index = self
                    .context
                    .base_key()
                    .base_tag_index(KeyTag::Index as u8, &short_key);
                keys_to_check.push(key_index);
                keys_to_check_metadata.push((position, short_key));
            }
        }

        let found_keys = self.context.store().contains_keys(&keys_to_check).await?;
        let entries_to_load = keys_to_check_metadata
            .into_iter()
            .zip(found_keys)
            .filter_map(|(metadata, found)| found.then_some(metadata))
            .map(|(position, short_key)| {
                let subview_key = self
                    .context
                    .base_key()
                    .base_tag_index(KeyTag::Subview as u8, &short_key);
                let subview_context = self.context.clone_with_base_key(subview_key);
                (position, short_key.to_owned(), subview_context)
            })
            .collect::<Vec<_>>();
        if !entries_to_load.is_empty() {
            let mut keys_to_load = Vec::with_capacity(entries_to_load.len() * W::NUM_INIT_KEYS);
            for (_, _, context) in &entries_to_load {
                keys_to_load.extend(W::pre_load(context)?);
            }
            let values = self
                .context
                .store()
                .read_multi_values_bytes(&keys_to_load)
                .await?;
            for (loaded_values, (position, short_key, context)) in values
                .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
                .zip(entries_to_load)
            {
                let view = W::post_load(context, loaded_values)?;
                let wrapped_view = Arc::new(RwLock::new(view));
                results[position] = Some((short_key, wrapped_view));
            }
        }

        results
            .into_iter()
            .map(|maybe_view| match maybe_view {
                Some((short_key, view)) => Ok(Some(ReadGuardedView(
                    view.try_read_arc()
                        .ok_or_else(|| ViewError::TryLockError(short_key))?,
                ))),
                None => Ok(None),
            })
            .collect()
    }

    /// Loads multiple entries for reading at once with their keys.
    /// The entries in short_keys have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&vec![0, 1]).await.unwrap();
    /// }
    /// let short_keys = vec![vec![0, 1], vec![0, 2]];
    /// let subviews = view.try_load_entries_pairs(short_keys).await.unwrap();
    /// assert!(subviews[1].1.is_none());
    /// let value0 = subviews[0].1.as_ref().unwrap().get();
    /// assert_eq!(*value0, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_pairs(
        &self,
        short_keys: Vec<Vec<u8>>,
    ) -> Result<Vec<(Vec<u8>, Option<ReadGuardedView<W>>)>, ViewError> {
        let values = self.try_load_entries(short_keys.clone()).await?;
        Ok(short_keys.into_iter().zip(values).collect())
    }

    /// Loads all the entries for reading at once.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// }
    /// let subviews = view.try_load_all_entries().await.unwrap();
    /// assert_eq!(subviews.len(), 1);
    /// # })
    /// ```
    pub async fn try_load_all_entries(
        &self,
    ) -> Result<Vec<(Vec<u8>, ReadGuardedView<W>)>, ViewError> {
        let short_keys = self.keys().await?;
        let mut loaded_views = vec![None; short_keys.len()];

        // Load views that are not in updates and not deleted
        if !self.delete_storage_first {
            let mut keys = Vec::new();
            let mut short_keys_and_indexes = Vec::new();
            for (index, short_key) in short_keys.iter().enumerate() {
                if !self.updates.contains_key(short_key) {
                    let key = self
                        .context
                        .base_key()
                        .base_tag_index(KeyTag::Subview as u8, short_key);
                    let context = self.context.clone_with_base_key(key);
                    keys.extend(W::pre_load(&context)?);
                    short_keys_and_indexes.push((short_key.to_vec(), index));
                }
            }
            let values = self.context.store().read_multi_values_bytes(&keys).await?;
            for (loaded_values, (short_key, index)) in values
                .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
                .zip(short_keys_and_indexes)
            {
                let key = self
                    .context
                    .base_key()
                    .base_tag_index(KeyTag::Subview as u8, &short_key);
                let context = self.context.clone_with_base_key(key);
                let view = W::post_load(context, loaded_values)?;
                let wrapped_view = Arc::new(RwLock::new(view));
                loaded_views[index] = Some(wrapped_view);
            }
        }

        // Create result from updates and loaded views
        short_keys
            .into_iter()
            .zip(loaded_views)
            .map(|(short_key, loaded_view)| {
                let view = if let Some(Update::Set(view)) = self.updates.get(&short_key) {
                    view.clone()
                } else if let Some(view) = loaded_view {
                    view
                } else {
                    unreachable!("All entries should have been loaded into memory");
                };
                let guard = ReadGuardedView(
                    view.try_read_arc()
                        .ok_or_else(|| ViewError::TryLockError(short_key.clone()))?,
                );
                Ok((short_key, guard))
            })
            .collect()
    }

    /// Loads all the entries for writing at once.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// }
    /// let subviews = view.try_load_all_entries_mut().await.unwrap();
    /// assert_eq!(subviews.len(), 1);
    /// # })
    /// ```
    pub async fn try_load_all_entries_mut(
        &mut self,
    ) -> Result<Vec<(Vec<u8>, WriteGuardedView<W>)>, ViewError> {
        let short_keys = self.keys().await?;
        if !self.delete_storage_first {
            let mut keys = Vec::new();
            let mut short_keys_to_load = Vec::new();

            for short_key in &short_keys {
                if !self.updates.contains_key(short_key) {
                    let key = self
                        .context
                        .base_key()
                        .base_tag_index(KeyTag::Subview as u8, short_key);
                    let context = self.context.clone_with_base_key(key);
                    keys.extend(W::pre_load(&context)?);
                    short_keys_to_load.push(short_key.to_vec());
                }
            }

            let values = self.context.store().read_multi_values_bytes(&keys).await?;
            for (loaded_values, short_key) in values
                .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
                .zip(short_keys_to_load)
            {
                let key = self
                    .context
                    .base_key()
                    .base_tag_index(KeyTag::Subview as u8, &short_key);
                let context = self.context.clone_with_base_key(key);
                let view = W::post_load(context, loaded_values)?;
                let wrapped_view = Arc::new(RwLock::new(view));
                self.updates
                    .insert(short_key.to_vec(), Update::Set(wrapped_view));
            }
        }
        short_keys
            .into_iter()
            .map(|short_key| {
                let Some(Update::Set(view)) = self.updates.get(&short_key) else {
                    unreachable!("All entries should have been loaded into `updates`")
                };
                let guard = WriteGuardedView(
                    view.clone()
                        .try_write_arc()
                        .ok_or_else(|| ViewError::TryLockError(short_key.clone()))?,
                );
                Ok((short_key, guard))
            })
            .collect()
    }
}

impl<W: View> ReentrantByteCollectionView<W::Context, W> {
    /// Returns the list of indices in the collection in lexicographic order.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
    /// let keys = view.keys().await.unwrap();
    /// assert_eq!(keys, vec![vec![0, 1], vec![0, 2]]);
    /// # })
    /// ```
    pub async fn keys(&self) -> Result<Vec<Vec<u8>>, ViewError> {
        let mut keys = Vec::new();
        self.for_each_key(|key| {
            keys.push(key.to_vec());
            Ok(())
        })
        .await?;
        Ok(keys)
    }

    /// Returns the number of indices of the collection.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
    /// assert_eq!(view.iterative_count().await.unwrap(), 2);
    /// # })
    /// ```
    pub async fn iterative_count(&self) -> Result<usize, ViewError> {
        let mut count = 0;
        self.for_each_key(|_key| {
            count += 1;
            Ok(())
        })
        .await?;
        Ok(count)
    }

    /// Applies a function f on each index (aka key). Keys are visited in a
    /// lexicographic order. If the function returns false then the loop
    /// ends prematurely.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
    /// let mut count = 0;
    /// view.for_each_key_while(|_key| {
    ///     count += 1;
    ///     Ok(count < 1)
    /// })
    /// .await
    /// .unwrap();
    /// assert_eq!(count, 1);
    /// # })
    /// ```
    pub async fn for_each_key_while<F>(&self, mut f: F) -> Result<(), ViewError>
    where
        F: FnMut(&[u8]) -> Result<bool, ViewError> + Send,
    {
        let mut updates = self.updates.iter();
        let mut update = updates.next();
        if !self.delete_storage_first {
            let base = self.get_index_key(&[]);
            for index in self.context.store().find_keys_by_prefix(&base).await? {
                loop {
                    match update {
                        Some((key, value)) if key <= &index => {
                            if let Update::Set(_) = value {
                                if !f(key)? {
                                    return Ok(());
                                }
                            }
                            update = updates.next();
                            if key == &index {
                                break;
                            }
                        }
                        _ => {
                            if !f(&index)? {
                                return Ok(());
                            }
                            break;
                        }
                    }
                }
            }
        }
        while let Some((key, value)) = update {
            if let Update::Set(_) = value {
                if !f(key)? {
                    return Ok(());
                }
            }
            update = updates.next();
        }
        Ok(())
    }

    /// Applies a function f on each index (aka key). Keys are visited in a
    /// lexicographic order.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
    ///     ReentrantByteCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
    /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
    /// let mut count = 0;
    /// view.for_each_key(|_key| {
    ///     count += 1;
    ///     Ok(())
    /// })
    /// .await
    /// .unwrap();
    /// assert_eq!(count, 2);
    /// # })
    /// ```
    pub async fn for_each_key<F>(&self, mut f: F) -> Result<(), ViewError>
    where
        F: FnMut(&[u8]) -> Result<(), ViewError> + Send,
    {
        self.for_each_key_while(|key| {
            f(key)?;
            Ok(true)
        })
        .await
    }
}

impl<W: HashableView> HashableView for ReentrantByteCollectionView<W::Context, W> {
    type Hasher = sha3::Sha3_256;

    async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
        #[cfg(with_metrics)]
        let _hash_latency = metrics::REENTRANT_COLLECTION_VIEW_HASH_RUNTIME.measure_latency();
        let mut hasher = sha3::Sha3_256::default();
        let keys = self.keys().await?;
        let count = keys.len() as u32;
        hasher.update_with_bcs_bytes(&count)?;
        for key in keys {
            hasher.update_with_bytes(&key)?;
            let hash = if let Some(entry) = self.updates.get_mut(&key) {
                let Update::Set(view) = entry else {
                    unreachable!("Loaded entries in updates should always be Update::Set");
                };
                let mut view = view
                    .try_write_arc()
                    .ok_or_else(|| ViewError::TryLockError(key))?;
                view.hash_mut().await?
            } else {
                let key = self
                    .context
                    .base_key()
                    .base_tag_index(KeyTag::Subview as u8, &key);
                let context = self.context.clone_with_base_key(key);
                let mut view = W::load(context).await?;
                view.hash_mut().await?
            };
            hasher.write_all(hash.as_ref())?;
        }
        Ok(hasher.finalize())
    }

    async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
        #[cfg(with_metrics)]
        let _hash_latency = metrics::REENTRANT_COLLECTION_VIEW_HASH_RUNTIME.measure_latency();
        let mut hasher = sha3::Sha3_256::default();
        let keys = self.keys().await?;
        let count = keys.len() as u32;
        hasher.update_with_bcs_bytes(&count)?;
        for key in keys {
            hasher.update_with_bytes(&key)?;
            let hash = if let Some(entry) = self.updates.get(&key) {
                let Update::Set(view) = entry else {
                    unreachable!("Loaded entries in updates should always be Update::Set");
                };
                let view = view
                    .try_read_arc()
                    .ok_or_else(|| ViewError::TryLockError(key))?;
                view.hash().await?
            } else {
                let key = self
                    .context
                    .base_key()
                    .base_tag_index(KeyTag::Subview as u8, &key);
                let context = self.context.clone_with_base_key(key);
                let view = W::load(context).await?;
                view.hash().await?
            };
            hasher.write_all(hash.as_ref())?;
        }
        Ok(hasher.finalize())
    }
}

/// A view that supports accessing a collection of views of the same kind, indexed by keys,
/// possibly several subviews at a time.
#[derive(Debug, Allocative)]
#[allocative(bound = "C, I, W: Allocative")]
pub struct ReentrantCollectionView<C, I, W> {
    collection: ReentrantByteCollectionView<C, W>,
    #[allocative(skip)]
    _phantom: PhantomData<I>,
}

impl<I, W, C2> ReplaceContext<C2> for ReentrantCollectionView<W::Context, I, W>
where
    W: View + ReplaceContext<C2>,
    I: Send + Sync + Serialize + DeserializeOwned,
    C2: Context,
{
    type Target = ReentrantCollectionView<C2, I, <W as ReplaceContext<C2>>::Target>;

    async fn with_context(
        &mut self,
        ctx: impl FnOnce(&Self::Context) -> C2 + Clone,
    ) -> Self::Target {
        ReentrantCollectionView {
            collection: self.collection.with_context(ctx).await,
            _phantom: self._phantom,
        }
    }
}

impl<I, W> View for ReentrantCollectionView<W::Context, I, W>
where
    W: View,
    I: Send + Sync + Serialize + DeserializeOwned,
{
    const NUM_INIT_KEYS: usize = ReentrantByteCollectionView::<W::Context, W>::NUM_INIT_KEYS;

    type Context = W::Context;

    fn context(&self) -> Self::Context {
        self.collection.context()
    }

    fn pre_load(context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
        ReentrantByteCollectionView::<W::Context, W>::pre_load(context)
    }

    fn post_load(context: Self::Context, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
        let collection = ReentrantByteCollectionView::post_load(context, values)?;
        Ok(ReentrantCollectionView {
            collection,
            _phantom: PhantomData,
        })
    }

    fn rollback(&mut self) {
        self.collection.rollback()
    }

    async fn has_pending_changes(&self) -> bool {
        self.collection.has_pending_changes().await
    }

    fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
        self.collection.pre_save(batch)
    }

    fn post_save(&mut self) {
        self.collection.post_save()
    }

    fn clear(&mut self) {
        self.collection.clear()
    }
}

impl<I, W> ClonableView for ReentrantCollectionView<W::Context, I, W>
where
    W: ClonableView,
    I: Send + Sync + Serialize + DeserializeOwned,
{
    fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
        Ok(ReentrantCollectionView {
            collection: self.collection.clone_unchecked()?,
            _phantom: PhantomData,
        })
    }
}

impl<I, W> ReentrantCollectionView<W::Context, I, W>
where
    W: View,
    I: Sync + Send + Serialize + DeserializeOwned,
{
    /// Loads a subview for the data at the given index in the collection. If an entry
    /// is absent then a default entry is put on the collection. The obtained view can
    /// then be modified.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// let subview = view.try_load_entry_mut(&23).await.unwrap();
    /// let value = subview.get();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub async fn try_load_entry_mut<Q>(
        &mut self,
        index: &Q,
    ) -> Result<WriteGuardedView<W>, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + ?Sized,
    {
        let short_key = BaseKey::derive_short_key(index)?;
        self.collection.try_load_entry_mut(&short_key).await
    }

    /// Loads a subview at the given index in the collection and gives read-only access to the data.
    /// If an entry is absent then `None` is returned.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
    /// let value = subview.get();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub async fn try_load_entry<Q>(
        &self,
        index: &Q,
    ) -> Result<Option<ReadGuardedView<W>>, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + ?Sized,
    {
        let short_key = BaseKey::derive_short_key(index)?;
        self.collection.try_load_entry(&short_key).await
    }

    /// Returns `true` if the collection contains a value for the specified key.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// assert!(view.contains_key(&23).await.unwrap());
    /// assert!(!view.contains_key(&24).await.unwrap());
    /// # })
    /// ```
    pub async fn contains_key<Q>(&self, index: &Q) -> Result<bool, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + ?Sized,
    {
        let short_key = BaseKey::derive_short_key(index)?;
        self.collection.contains_key(&short_key).await
    }

    /// Marks the entry so that it is removed in the next flush.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
    /// let value = subview.get_mut();
    /// assert_eq!(*value, String::default());
    /// view.remove_entry(&23);
    /// let keys = view.indices().await.unwrap();
    /// assert_eq!(keys.len(), 0);
    /// # })
    /// ```
    pub fn remove_entry<Q>(&mut self, index: &Q) -> Result<(), ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + ?Sized,
    {
        let short_key = BaseKey::derive_short_key(index)?;
        self.collection.remove_entry(short_key);
        Ok(())
    }

    /// Marks the entry so that it is removed in the next flush.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// {
    ///     let mut subview = view.try_load_entry_mut(&23).await.unwrap();
    ///     let value = subview.get_mut();
    ///     *value = String::from("Hello");
    /// }
    /// view.try_reset_entry_to_default(&23).unwrap();
    /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
    /// let value = subview.get_mut();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub fn try_reset_entry_to_default<Q>(&mut self, index: &Q) -> Result<(), ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + ?Sized,
    {
        let short_key = BaseKey::derive_short_key(index)?;
        self.collection.try_reset_entry_to_default(&short_key)
    }

    /// Gets the extra data.
    pub fn extra(&self) -> &<W::Context as Context>::Extra {
        self.collection.extra()
    }
}

impl<I, W> ReentrantCollectionView<W::Context, I, W>
where
    W: View,
    I: Sync + Send + Serialize + DeserializeOwned,
{
    /// Load multiple entries for writing at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// let indices = vec![23, 42];
    /// let subviews = view.try_load_entries_mut(&indices).await.unwrap();
    /// let value1 = subviews[0].get();
    /// let value2 = subviews[1].get();
    /// assert_eq!(*value1, String::default());
    /// assert_eq!(*value2, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_mut<'a, Q>(
        &'a mut self,
        indices: impl IntoIterator<Item = &'a Q>,
    ) -> Result<Vec<WriteGuardedView<W>>, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + 'a,
    {
        let short_keys = indices
            .into_iter()
            .map(|index| BaseKey::derive_short_key(index))
            .collect::<Result<_, _>>()?;
        self.collection.try_load_entries_mut(short_keys).await
    }

    /// Loads multiple entries for writing at once with their keys.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// let indices = [23, 42];
    /// let subviews = view.try_load_entries_pairs_mut(indices).await.unwrap();
    /// let value1 = subviews[0].1.get();
    /// let value2 = subviews[1].1.get();
    /// assert_eq!(*value1, String::default());
    /// assert_eq!(*value2, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_pairs_mut<Q>(
        &mut self,
        indices: impl IntoIterator<Item = Q>,
    ) -> Result<Vec<(Q, WriteGuardedView<W>)>, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + Clone,
    {
        let indices_vec: Vec<Q> = indices.into_iter().collect();
        let values = self.try_load_entries_mut(indices_vec.iter()).await?;
        Ok(indices_vec.into_iter().zip(values).collect())
    }

    /// Load multiple entries for reading at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let indices = vec![23, 42];
    /// let subviews = view.try_load_entries(&indices).await.unwrap();
    /// assert!(subviews[1].is_none());
    /// let value0 = subviews[0].as_ref().unwrap().get();
    /// assert_eq!(*value0, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries<'a, Q>(
        &'a self,
        indices: impl IntoIterator<Item = &'a Q>,
    ) -> Result<Vec<Option<ReadGuardedView<W>>>, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + 'a,
    {
        let short_keys = indices
            .into_iter()
            .map(|index| BaseKey::derive_short_key(index))
            .collect::<Result<_, _>>()?;
        self.collection.try_load_entries(short_keys).await
    }

    /// Loads multiple entries for reading at once with their keys.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let indices = [23, 42];
    /// let subviews = view.try_load_entries_pairs(indices).await.unwrap();
    /// assert!(subviews[1].1.is_none());
    /// let value0 = subviews[0].1.as_ref().unwrap().get();
    /// assert_eq!(*value0, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_pairs<Q>(
        &self,
        indices: impl IntoIterator<Item = Q>,
    ) -> Result<Vec<(Q, Option<ReadGuardedView<W>>)>, ViewError>
    where
        I: Borrow<Q>,
        Q: Serialize + Clone,
    {
        let indices_vec: Vec<Q> = indices.into_iter().collect();
        let values = self.try_load_entries(indices_vec.iter()).await?;
        Ok(indices_vec.into_iter().zip(values).collect())
    }

    /// Loads all entries for writing at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subviews = view.try_load_all_entries_mut().await.unwrap();
    /// assert_eq!(subviews.len(), 1);
    /// # })
    /// ```
    pub async fn try_load_all_entries_mut(
        &mut self,
    ) -> Result<Vec<(I, WriteGuardedView<W>)>, ViewError> {
        let results = self.collection.try_load_all_entries_mut().await?;
        results
            .into_iter()
            .map(|(short_key, view)| {
                let index = BaseKey::deserialize_value(&short_key)?;
                Ok((index, view))
            })
            .collect()
    }

    /// Load multiple entries for reading at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subviews = view.try_load_all_entries().await.unwrap();
    /// assert_eq!(subviews.len(), 1);
    /// # })
    /// ```
    pub async fn try_load_all_entries(&self) -> Result<Vec<(I, ReadGuardedView<W>)>, ViewError> {
        let results = self.collection.try_load_all_entries().await?;
        results
            .into_iter()
            .map(|(short_key, view)| {
                let index = BaseKey::deserialize_value(&short_key)?;
                Ok((index, view))
            })
            .collect()
    }
}

impl<I, W> ReentrantCollectionView<W::Context, I, W>
where
    W: View,
    I: Sync + Send + Serialize + DeserializeOwned,
{
    /// Returns the list of indices in the collection in an order determined
    /// by serialization.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// view.try_load_entry_mut(&25).await.unwrap();
    /// let indices = view.indices().await.unwrap();
    /// assert_eq!(indices.len(), 2);
    /// # })
    /// ```
    pub async fn indices(&self) -> Result<Vec<I>, ViewError> {
        let mut indices = Vec::new();
        self.for_each_index(|index| {
            indices.push(index);
            Ok(())
        })
        .await?;
        Ok(indices)
    }

    /// Returns the number of indices in the collection.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// view.try_load_entry_mut(&25).await.unwrap();
    /// assert_eq!(view.iterative_count().await.unwrap(), 2);
    /// # })
    /// ```
    pub async fn iterative_count(&self) -> Result<usize, ViewError> {
        self.collection.iterative_count().await
    }

    /// Applies a function f on each index. Indices are visited in an order
    /// determined by the serialization. If the function f returns false then
    /// the loop ends prematurely.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// view.try_load_entry_mut(&24).await.unwrap();
    /// let mut count = 0;
    /// view.for_each_index_while(|_key| {
    ///     count += 1;
    ///     Ok(count < 1)
    /// })
    /// .await
    /// .unwrap();
    /// assert_eq!(count, 1);
    /// # })
    /// ```
    pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
    where
        F: FnMut(I) -> Result<bool, ViewError> + Send,
    {
        self.collection
            .for_each_key_while(|key| {
                let index = BaseKey::deserialize_value(key)?;
                f(index)
            })
            .await?;
        Ok(())
    }

    /// Applies a function f on each index. Indices are visited in an order
    /// determined by the serialization.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
    ///     ReentrantCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// view.try_load_entry_mut(&28).await.unwrap();
    /// let mut count = 0;
    /// view.for_each_index(|_key| {
    ///     count += 1;
    ///     Ok(())
    /// })
    /// .await
    /// .unwrap();
    /// assert_eq!(count, 2);
    /// # })
    /// ```
    pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
    where
        F: FnMut(I) -> Result<(), ViewError> + Send,
    {
        self.collection
            .for_each_key(|key| {
                let index = BaseKey::deserialize_value(key)?;
                f(index)
            })
            .await?;
        Ok(())
    }
}

impl<I, W> HashableView for ReentrantCollectionView<W::Context, I, W>
where
    W: HashableView,
    I: Send + Sync + Serialize + DeserializeOwned,
{
    type Hasher = sha3::Sha3_256;

    async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
        self.collection.hash_mut().await
    }

    async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
        self.collection.hash().await
    }
}

/// A view that supports accessing a collection of views of the same kind, indexed by an ordered key,
/// possibly several subviews at a time.
#[derive(Debug, Allocative)]
#[allocative(bound = "C, I, W: Allocative")]
pub struct ReentrantCustomCollectionView<C, I, W> {
    collection: ReentrantByteCollectionView<C, W>,
    #[allocative(skip)]
    _phantom: PhantomData<I>,
}

impl<I, W> View for ReentrantCustomCollectionView<W::Context, I, W>
where
    W: View,
    I: Send + Sync + CustomSerialize,
{
    const NUM_INIT_KEYS: usize = ReentrantByteCollectionView::<W::Context, W>::NUM_INIT_KEYS;

    type Context = W::Context;

    fn context(&self) -> Self::Context {
        self.collection.context()
    }

    fn pre_load(context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
        ReentrantByteCollectionView::<_, W>::pre_load(context)
    }

    fn post_load(context: Self::Context, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
        let collection = ReentrantByteCollectionView::post_load(context, values)?;
        Ok(ReentrantCustomCollectionView {
            collection,
            _phantom: PhantomData,
        })
    }

    fn rollback(&mut self) {
        self.collection.rollback()
    }

    async fn has_pending_changes(&self) -> bool {
        self.collection.has_pending_changes().await
    }

    fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
        self.collection.pre_save(batch)
    }

    fn post_save(&mut self) {
        self.collection.post_save()
    }

    fn clear(&mut self) {
        self.collection.clear()
    }
}

impl<I, W> ClonableView for ReentrantCustomCollectionView<W::Context, I, W>
where
    W: ClonableView,
    Self: View,
{
    fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
        Ok(ReentrantCustomCollectionView {
            collection: self.collection.clone_unchecked()?,
            _phantom: PhantomData,
        })
    }
}

impl<I, W> ReentrantCustomCollectionView<W::Context, I, W>
where
    W: View,
    I: Sync + Send + CustomSerialize,
{
    /// Loads a subview for the data at the given index in the collection. If an entry
    /// is absent then a default entry is put in the collection on this index.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// let subview = view.try_load_entry_mut(&23).await.unwrap();
    /// let value = subview.get();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub async fn try_load_entry_mut<Q>(
        &mut self,
        index: &Q,
    ) -> Result<WriteGuardedView<W>, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize,
    {
        let short_key = index.to_custom_bytes()?;
        self.collection.try_load_entry_mut(&short_key).await
    }

    /// Loads a subview at the given index in the collection and gives read-only access to the data.
    /// If an entry is absent then `None` is returned.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
    /// let value = subview.get();
    /// assert_eq!(*value, String::default());
    /// # })
    /// ```
    pub async fn try_load_entry<Q>(
        &self,
        index: &Q,
    ) -> Result<Option<ReadGuardedView<W>>, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize,
    {
        let short_key = index.to_custom_bytes()?;
        self.collection.try_load_entry(&short_key).await
    }

    /// Returns `true` if the collection contains a value for the specified key.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// assert!(view.contains_key(&23).await.unwrap());
    /// assert!(!view.contains_key(&24).await.unwrap());
    /// # })
    /// ```
    pub async fn contains_key<Q>(&self, index: &Q) -> Result<bool, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize,
    {
        let short_key = index.to_custom_bytes()?;
        self.collection.contains_key(&short_key).await
    }

    /// Removes an entry. If absent then nothing happens.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
    /// let value = subview.get_mut();
    /// assert_eq!(*value, String::default());
    /// view.remove_entry(&23);
    /// let keys = view.indices().await.unwrap();
    /// assert_eq!(keys.len(), 0);
    /// # })
    /// ```
    pub fn remove_entry<Q>(&mut self, index: &Q) -> Result<(), ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize,
    {
        let short_key = index.to_custom_bytes()?;
        self.collection.remove_entry(short_key);
        Ok(())
    }

    /// Marks the entry so that it is removed in the next flush.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// {
    ///     let mut subview = view.try_load_entry_mut(&23).await.unwrap();
    ///     let value = subview.get_mut();
    ///     *value = String::from("Hello");
    /// }
    /// {
    ///     view.try_reset_entry_to_default(&23).unwrap();
    ///     let subview = view.try_load_entry(&23).await.unwrap().unwrap();
    ///     let value = subview.get();
    ///     assert_eq!(*value, String::default());
    /// }
    /// # })
    /// ```
    pub fn try_reset_entry_to_default<Q>(&mut self, index: &Q) -> Result<(), ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize,
    {
        let short_key = index.to_custom_bytes()?;
        self.collection.try_reset_entry_to_default(&short_key)
    }

    /// Gets the extra data.
    pub fn extra(&self) -> &<W::Context as Context>::Extra {
        self.collection.extra()
    }
}

impl<I, W: View> ReentrantCustomCollectionView<W::Context, I, W>
where
    I: Sync + Send + CustomSerialize,
{
    /// Load multiple entries for writing at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// let subviews = view.try_load_entries_mut(&[23, 42]).await.unwrap();
    /// let value1 = subviews[0].get();
    /// let value2 = subviews[1].get();
    /// assert_eq!(*value1, String::default());
    /// assert_eq!(*value2, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_mut<'a, Q>(
        &mut self,
        indices: impl IntoIterator<Item = &'a Q>,
    ) -> Result<Vec<WriteGuardedView<W>>, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize + 'a,
    {
        let short_keys = indices
            .into_iter()
            .map(|index| index.to_custom_bytes())
            .collect::<Result<_, _>>()?;
        self.collection.try_load_entries_mut(short_keys).await
    }

    /// Loads multiple entries for writing at once with their keys.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// let indices = [23, 42];
    /// let subviews = view.try_load_entries_pairs_mut(indices).await.unwrap();
    /// let value1 = subviews[0].1.get();
    /// let value2 = subviews[1].1.get();
    /// assert_eq!(*value1, String::default());
    /// assert_eq!(*value2, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_pairs_mut<Q>(
        &mut self,
        indices: impl IntoIterator<Item = Q>,
    ) -> Result<Vec<(Q, WriteGuardedView<W>)>, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize + Clone,
    {
        let indices_vec: Vec<Q> = indices.into_iter().collect();
        let values = self.try_load_entries_mut(indices_vec.iter()).await?;
        Ok(indices_vec.into_iter().zip(values).collect())
    }

    /// Load multiple entries for reading at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subviews = view.try_load_entries(&[23, 42]).await.unwrap();
    /// assert!(subviews[1].is_none());
    /// let value0 = subviews[0].as_ref().unwrap().get();
    /// assert_eq!(*value0, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries<'a, Q>(
        &self,
        indices: impl IntoIterator<Item = &'a Q>,
    ) -> Result<Vec<Option<ReadGuardedView<W>>>, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize + 'a,
    {
        let short_keys = indices
            .into_iter()
            .map(|index| index.to_custom_bytes())
            .collect::<Result<_, _>>()?;
        self.collection.try_load_entries(short_keys).await
    }

    /// Loads multiple entries for reading at once with their keys.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let indices = [23, 42];
    /// let subviews = view.try_load_entries_pairs(indices).await.unwrap();
    /// assert!(subviews[1].1.is_none());
    /// let value0 = subviews[0].1.as_ref().unwrap().get();
    /// assert_eq!(*value0, String::default());
    /// # })
    /// ```
    pub async fn try_load_entries_pairs<Q>(
        &self,
        indices: impl IntoIterator<Item = Q>,
    ) -> Result<Vec<(Q, Option<ReadGuardedView<W>>)>, ViewError>
    where
        I: Borrow<Q>,
        Q: CustomSerialize + Clone,
    {
        let indices_vec: Vec<Q> = indices.into_iter().collect();
        let values = self.try_load_entries(indices_vec.iter()).await?;
        Ok(indices_vec.into_iter().zip(values).collect())
    }

    /// Loads all entries for writing at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subviews = view.try_load_all_entries_mut().await.unwrap();
    /// assert_eq!(subviews.len(), 1);
    /// # })
    /// ```
    pub async fn try_load_all_entries_mut(
        &mut self,
    ) -> Result<Vec<(I, WriteGuardedView<W>)>, ViewError> {
        let results = self.collection.try_load_all_entries_mut().await?;
        results
            .into_iter()
            .map(|(short_key, view)| {
                let index = I::from_custom_bytes(&short_key)?;
                Ok((index, view))
            })
            .collect()
    }

    /// Load multiple entries for reading at once.
    /// The entries in indices have to be all distinct.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// {
    ///     let _subview = view.try_load_entry_mut(&23).await.unwrap();
    /// }
    /// let subviews = view.try_load_all_entries().await.unwrap();
    /// assert_eq!(subviews.len(), 1);
    /// # })
    /// ```
    pub async fn try_load_all_entries(&self) -> Result<Vec<(I, ReadGuardedView<W>)>, ViewError> {
        let results = self.collection.try_load_all_entries().await?;
        results
            .into_iter()
            .map(|(short_key, view)| {
                let index = I::from_custom_bytes(&short_key)?;
                Ok((index, view))
            })
            .collect()
    }
}

impl<I, W> ReentrantCustomCollectionView<W::Context, I, W>
where
    W: View,
    I: Sync + Send + CustomSerialize,
{
    /// Returns the list of indices in the collection. The order is determined by
    /// the custom serialization.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// view.try_load_entry_mut(&25).await.unwrap();
    /// let indices = view.indices().await.unwrap();
    /// assert_eq!(indices, vec![23, 25]);
    /// # })
    /// ```
    pub async fn indices(&self) -> Result<Vec<I>, ViewError> {
        let mut indices = Vec::new();
        self.for_each_index(|index| {
            indices.push(index);
            Ok(())
        })
        .await?;
        Ok(indices)
    }

    /// Returns the number of entries in the collection.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// view.try_load_entry_mut(&25).await.unwrap();
    /// assert_eq!(view.iterative_count().await.unwrap(), 2);
    /// # })
    /// ```
    pub async fn iterative_count(&self) -> Result<usize, ViewError> {
        self.collection.iterative_count().await
    }

    /// Applies a function f on each index. Indices are visited in an order
    /// determined by the custom serialization. If the function f returns false
    /// then the loop ends prematurely.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&28).await.unwrap();
    /// view.try_load_entry_mut(&24).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// let mut part_indices = Vec::new();
    /// view.for_each_index_while(|index| {
    ///     part_indices.push(index);
    ///     Ok(part_indices.len() < 2)
    /// })
    /// .await
    /// .unwrap();
    /// assert_eq!(part_indices, vec![23, 24]);
    /// # })
    /// ```
    pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
    where
        F: FnMut(I) -> Result<bool, ViewError> + Send,
    {
        self.collection
            .for_each_key_while(|key| {
                let index = I::from_custom_bytes(key)?;
                f(index)
            })
            .await?;
        Ok(())
    }

    /// Applies a function f on each index. Indices are visited in an order
    /// determined by the custom serialization.
    /// ```rust
    /// # tokio_test::block_on(async {
    /// # use linera_views::context::MemoryContext;
    /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
    /// # use linera_views::register_view::RegisterView;
    /// # use linera_views::views::View;
    /// # let context = MemoryContext::new_for_testing(());
    /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
    ///     ReentrantCustomCollectionView::load(context).await.unwrap();
    /// view.try_load_entry_mut(&28).await.unwrap();
    /// view.try_load_entry_mut(&24).await.unwrap();
    /// view.try_load_entry_mut(&23).await.unwrap();
    /// let mut indices = Vec::new();
    /// view.for_each_index(|index| {
    ///     indices.push(index);
    ///     Ok(())
    /// })
    /// .await
    /// .unwrap();
    /// assert_eq!(indices, vec![23, 24, 28]);
    /// # })
    /// ```
    pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
    where
        F: FnMut(I) -> Result<(), ViewError> + Send,
    {
        self.collection
            .for_each_key(|key| {
                let index = I::from_custom_bytes(key)?;
                f(index)
            })
            .await?;
        Ok(())
    }
}

impl<I, W> HashableView for ReentrantCustomCollectionView<W::Context, I, W>
where
    W: HashableView,
    I: Send + Sync + CustomSerialize,
{
    type Hasher = sha3::Sha3_256;

    async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
        self.collection.hash_mut().await
    }

    async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
        self.collection.hash().await
    }
}

/// Type wrapping `ReentrantByteCollectionView` while memoizing the hash.
pub type HashedReentrantByteCollectionView<C, W> =
    WrappedHashableContainerView<C, ReentrantByteCollectionView<C, W>, HasherOutput>;

/// Type wrapping `ReentrantCollectionView` while memoizing the hash.
pub type HashedReentrantCollectionView<C, I, W> =
    WrappedHashableContainerView<C, ReentrantCollectionView<C, I, W>, HasherOutput>;

/// Type wrapping `ReentrantCustomCollectionView` while memoizing the hash.
pub type HashedReentrantCustomCollectionView<C, I, W> =
    WrappedHashableContainerView<C, ReentrantCustomCollectionView<C, I, W>, HasherOutput>;

/// Wrapper around `ReentrantByteCollectionView` to compute hashes based on the history of changes.
pub type HistoricallyHashedReentrantByteCollectionView<C, W> =
    HistoricallyHashableView<C, ReentrantByteCollectionView<C, W>>;

/// Wrapper around `ReentrantCollectionView` to compute hashes based on the history of changes.
pub type HistoricallyHashedReentrantCollectionView<C, I, W> =
    HistoricallyHashableView<C, ReentrantCollectionView<C, I, W>>;

/// Wrapper around `ReentrantCustomCollectionView` to compute hashes based on the history of changes.
pub type HistoricallyHashedReentrantCustomCollectionView<C, I, W> =
    HistoricallyHashableView<C, ReentrantCustomCollectionView<C, I, W>>;

#[cfg(with_graphql)]
mod graphql {
    use std::borrow::Cow;

    use super::{ReadGuardedView, ReentrantCollectionView};
    use crate::{
        graphql::{hash_name, mangle, missing_key_error, Entry, MapInput},
        views::View,
    };

    impl<T: async_graphql::OutputType> async_graphql::OutputType for ReadGuardedView<T> {
        fn type_name() -> Cow<'static, str> {
            T::type_name()
        }

        fn create_type_info(registry: &mut async_graphql::registry::Registry) -> String {
            T::create_type_info(registry)
        }

        async fn resolve(
            &self,
            ctx: &async_graphql::ContextSelectionSet<'_>,
            field: &async_graphql::Positioned<async_graphql::parser::types::Field>,
        ) -> async_graphql::ServerResult<async_graphql::Value> {
            (**self).resolve(ctx, field).await
        }
    }

    impl<C: Send + Sync, K: async_graphql::OutputType, V: async_graphql::OutputType>
        async_graphql::TypeName for ReentrantCollectionView<C, K, V>
    {
        fn type_name() -> Cow<'static, str> {
            format!(
                "ReentrantCollectionView_{}_{}_{:08x}",
                mangle(K::type_name()),
                mangle(V::type_name()),
                hash_name::<(K, V)>(),
            )
            .into()
        }
    }

    #[async_graphql::Object(cache_control(no_cache), name_type)]
    impl<K, V> ReentrantCollectionView<V::Context, K, V>
    where
        K: async_graphql::InputType
            + async_graphql::OutputType
            + serde::ser::Serialize
            + serde::de::DeserializeOwned
            + std::fmt::Debug,
        V: View + async_graphql::OutputType,
    {
        async fn keys(&self) -> Result<Vec<K>, async_graphql::Error> {
            Ok(self.indices().await?)
        }

        #[graphql(derived(name = "count"))]
        async fn count_(&self) -> Result<u32, async_graphql::Error> {
            Ok(self.iterative_count().await? as u32)
        }

        async fn entry(
            &self,
            key: K,
        ) -> Result<Entry<K, ReadGuardedView<V>>, async_graphql::Error> {
            let value = self
                .try_load_entry(&key)
                .await?
                .ok_or_else(|| missing_key_error(&key))?;
            Ok(Entry { value, key })
        }

        async fn entries(
            &self,
            input: Option<MapInput<K>>,
        ) -> Result<Vec<Entry<K, ReadGuardedView<V>>>, async_graphql::Error> {
            let keys = if let Some(keys) = input
                .and_then(|input| input.filters)
                .and_then(|filters| filters.keys)
            {
                keys
            } else {
                self.indices().await?
            };

            let values = self.try_load_entries(&keys).await?;
            Ok(values
                .into_iter()
                .zip(keys)
                .filter_map(|(value, key)| value.map(|value| Entry { value, key }))
                .collect())
        }
    }

    use crate::reentrant_collection_view::ReentrantCustomCollectionView;
    impl<C: Send + Sync, K: async_graphql::OutputType, V: async_graphql::OutputType>
        async_graphql::TypeName for ReentrantCustomCollectionView<C, K, V>
    {
        fn type_name() -> Cow<'static, str> {
            format!(
                "ReentrantCustomCollectionView_{}_{}_{:08x}",
                mangle(K::type_name()),
                mangle(V::type_name()),
                hash_name::<(K, V)>(),
            )
            .into()
        }
    }

    #[async_graphql::Object(cache_control(no_cache), name_type)]
    impl<K, V> ReentrantCustomCollectionView<V::Context, K, V>
    where
        K: async_graphql::InputType
            + async_graphql::OutputType
            + crate::common::CustomSerialize
            + std::fmt::Debug,
        V: View + async_graphql::OutputType,
    {
        async fn keys(&self) -> Result<Vec<K>, async_graphql::Error> {
            Ok(self.indices().await?)
        }

        async fn entry(
            &self,
            key: K,
        ) -> Result<Entry<K, ReadGuardedView<V>>, async_graphql::Error> {
            let value = self
                .try_load_entry(&key)
                .await?
                .ok_or_else(|| missing_key_error(&key))?;
            Ok(Entry { value, key })
        }

        async fn entries(
            &self,
            input: Option<MapInput<K>>,
        ) -> Result<Vec<Entry<K, ReadGuardedView<V>>>, async_graphql::Error> {
            let keys = if let Some(keys) = input
                .and_then(|input| input.filters)
                .and_then(|filters| filters.keys)
            {
                keys
            } else {
                self.indices().await?
            };

            let values = self.try_load_entries(&keys).await?;
            Ok(values
                .into_iter()
                .zip(keys)
                .filter_map(|(value, key)| value.map(|value| Entry { value, key }))
                .collect())
        }
    }
}