Address a bug in try_load_entries of CollectionView. (#4753)

## Motivation

In `CollectionView`, we have `KeyTag::Index` / `KeyTag::Subview`, but
they were wrongly
assigned in the function `try_load_entries`.

## Proposal

The correction was straightforward, once it was found.

## Test Plan

A random test has been added. This test concerns both the
`CollectionView` and `ReentrantCollectionView`.
This test covers both the found bug, the preceding one with
`W::NUM_INIT_KEYS == 0`. Therefore, the two
added tests in the `view_test.rs` are removed.

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

## Links

A test case was created in
[linera_end_to_end_tests/graphql_induced_crash](https://github.com/MathieuDutSik/linera_end_to_end_tests).
It could be put as
an additional integration test that exercises the GraphQL capabilities
of Linera.

Author: MathieuDutSik

linera-views/src/views/collection_view.rs

@@ -367,11 +367,15 @@ impl<W: View> ByteCollectionView<W::Context, W> {
                 None => {
                     results.push(None); // Placeholder, may be updated later
                     if !self.delete_storage_first {
+                        let key = self
+                            .context
+                            .base_key()
+                            .base_tag_index(KeyTag::Subview as u8, &short_key);
+                        let subview_context = self.context.clone_with_base_key(key);
                         let key = self
                             .context
                             .base_key()
                             .base_tag_index(KeyTag::Index as u8, &short_key);
-                        let subview_context = self.context.clone_with_base_key(key.clone());
                         keys_to_check.push(key);
                         keys_to_check_metadata.push((position, subview_context));
                     }

linera-views/tests/random_container_tests.rs

@@ -6,15 +6,15 @@ use std::collections::{BTreeMap, BTreeSet};
 use anyhow::Result;
 use linera_views::{
     bucket_queue_view::HashedBucketQueueView,
-    collection_view::HashedCollectionView,
+    collection_view::{CollectionView, HashedCollectionView},
     context::{Context, MemoryContext},
     key_value_store_view::{KeyValueStoreView, SizeData},
-    map_view::HashedByteMapView,
+    map_view::{HashedByteMapView, MapView},
     queue_view::HashedQueueView,
     random::make_deterministic_rng,
-    reentrant_collection_view::HashedReentrantCollectionView,
+    reentrant_collection_view::{HashedReentrantCollectionView, ReentrantCollectionView},
     register_view::RegisterView,
-    views::{CryptoHashRootView, CryptoHashView, RootView, View},
+    views::{CryptoHashRootView, CryptoHashView, HashableView as _, RootView, View},
 };
 use rand::{distributions::Uniform, Rng, RngCore};
 
@@ -508,6 +508,125 @@ async fn bucket_queue_view_mutability_check() -> Result<()> {
     Ok(())
 }
 
+#[derive(CryptoHashRootView)]
+pub struct NestedCollectionMapView<C> {
+    pub map1: CollectionView<C, String, MapView<C, String, u64>>,
+    pub map2: ReentrantCollectionView<C, String, MapView<C, String, u64>>,
+}
+
+impl<C: Context> NestedCollectionMapView<C> {
+    async fn read_maps_nested_collection_map_view(
+        &self,
+    ) -> Result<BTreeMap<String, BTreeMap<String, u64>>> {
+        let indices1 = self.map1.indices().await?;
+        let indices2 = self.map2.indices().await?;
+        assert_eq!(indices1, indices2, "Different set of indices");
+
+        let subviews1 = self.map1.try_load_entries(&indices1).await?;
+        let subviews2 = self.map2.try_load_entries(&indices1).await?;
+        let mut state_map = BTreeMap::new();
+        for ((subview1, subview2), index) in subviews1.into_iter().zip(subviews2).zip(indices1) {
+            let key_values1 = subview1.unwrap().index_values().await?;
+            let key_values2 = subview2.unwrap().index_values().await?;
+            assert_eq!(key_values1, key_values2, "key-values should be equal");
+            let key_values = key_values1.into_iter().collect::<BTreeMap<String, u64>>();
+            state_map.insert(index, key_values);
+        }
+        Ok(state_map)
+    }
+}
+
+#[tokio::test]
+async fn nested_collection_map_view_check() -> Result<()> {
+    let context = MemoryContext::new_for_testing(());
+    let mut rng = make_deterministic_rng();
+    let mut state_map: BTreeMap<String, BTreeMap<String, u64>> = BTreeMap::new();
+    let n = 20;
+    for _ in 0..n {
+        let mut view = NestedCollectionMapView::load(context.clone()).await?;
+        let hash = view.crypto_hash().await?;
+        let save = rng.gen::<bool>();
+
+        let count_oper = rng.gen_range(0..25);
+        let mut new_state_map = state_map.clone();
+        for _ in 0..count_oper {
+            let keys: Vec<String> = new_state_map.keys().cloned().collect::<Vec<_>>();
+            let count = new_state_map.len();
+            let choice = rng.gen_range(0..5);
+            if choice >= 2 {
+                let key1 = rng.gen_range::<u8, _>(0..10);
+                let key1 = format!("key1_{key1}");
+                let key2 = rng.gen_range::<u8, _>(0..10);
+                let key2 = format!("key2_{key2}");
+                let value = rng.gen_range::<u64, _>(0..100);
+                // insert into maps.
+                let subview1 = view.map1.load_entry_mut(&key1).await?;
+                subview1.insert(&key2, value)?;
+                let mut subview2 = view.map2.try_load_entry_mut(&key1).await?;
+                subview2.insert(&key2, value)?;
+                // insert into control
+                let mut map = new_state_map.get(&key1).cloned().unwrap_or_default();
+                map.insert(key2, value);
+                new_state_map.insert(key1, map);
+            }
+            if choice == 1 && count > 0 {
+                let pos = rng.gen_range(0..count) as usize;
+                let key = keys[pos].clone();
+                view.map1.remove_entry(&key)?;
+                view.map2.remove_entry(&key)?;
+                new_state_map.remove(&key);
+            }
+            if choice == 2 && count > 0 {
+                let pos = rng.gen_range(0..count);
+                let key1 = keys[pos].clone();
+                let submap = new_state_map.get_mut(&key1).unwrap();
+                let count = submap.len();
+                if count > 0 {
+                    let subkeys = submap
+                        .iter()
+                        .map(|(key, _)| key.clone())
+                        .collect::<Vec<_>>();
+                    let pos = rng.gen_range(0..count);
+                    let key2 = subkeys[pos].clone();
+                    submap.remove(&key2);
+                    // Removing some entries from the view
+                    let subview1 = view.map1.load_entry_mut(&key1).await?;
+                    subview1.remove(&key2)?;
+                    let mut subview2 = view.map2.try_load_entry_mut(&key1).await?;
+                    subview2.remove(&key2)?;
+                }
+            }
+            let state_view = view.read_maps_nested_collection_map_view().await?;
+            assert_eq!(
+                state_view, new_state_map,
+                "state_view should match new_state_map"
+            );
+            let new_hash = view.crypto_hash().await?;
+            if state_map == new_state_map {
+                assert_eq!(new_hash, hash);
+            } else {
+                // If equal it is a bug or a hash collision (unlikely)
+                assert_ne!(new_hash, hash);
+            }
+            let hash1 = view.map1.hash().await?;
+            let hash2 = view.map2.hash().await?;
+            assert_eq!(
+                hash1, hash2,
+                "hash for CollectionView / ReentrantCollectionView should match"
+            );
+        }
+        if save {
+            if state_map != new_state_map {
+                assert!(view.has_pending_changes().await);
+            }
+            state_map.clone_from(&new_state_map);
+            view.save().await?;
+            assert!(!view.has_pending_changes().await);
+        }
+    }
+    Ok(())
+}
+
 #[derive(CryptoHashRootView)]
 pub struct QueueStateView<C> {
     pub queue: HashedQueueView<C, u8>,

linera-views/tests/views_tests.rs

@@ -15,16 +15,16 @@ use linera_views::{
         Batch, WriteOperation,
         WriteOperation::{Delete, DeletePrefix, Put},
     },
-    collection_view::{CollectionView, HashedCollectionView},
+    collection_view::HashedCollectionView,
     context::{Context, MemoryContext, ViewContext},
     key_value_store_view::{KeyValueStoreView, ViewContainer},
     log_view::HashedLogView,
     lru_caching::LruCachingMemoryDatabase,
-    map_view::{ByteMapView, HashedMapView, MapView},
+    map_view::{ByteMapView, HashedMapView},
     memory::MemoryDatabase,
     queue_view::HashedQueueView,
     random::make_deterministic_rng,
-    reentrant_collection_view::{HashedReentrantCollectionView, ReentrantCollectionView},
+    reentrant_collection_view::HashedReentrantCollectionView,
     register_view::HashedRegisterView,
     set_view::HashedSetView,
     store::{KeyValueDatabase, TestKeyValueDatabase as _, WritableKeyValueStore as _},
@@ -614,52 +614,6 @@ where
     .await
 }
 
-#[derive(RootView)]
-pub struct NestedCollectionMapView<C> {
-    pub map: CollectionView<C, String, MapView<C, String, u64>>,
-}
-
-// This test exercise the case W::NUM_INIT_KEYS == 0
-// in CollectionView.
-#[tokio::test]
-async fn test_nested_collection_map_view() -> anyhow::Result<()> {
-    let context = MemoryContext::new_for_testing(());
-    {
-        let mut view = NestedCollectionMapView::load(context.clone()).await?;
-        let subview = view.map.load_entry_mut("Bonjour").await?;
-        subview.insert("A bientot", 49)?;
-        view.save().await?;
-    }
-    let view = NestedCollectionMapView::load(context).await?;
-    let keys = vec!["Bonjour".to_string()];
-    let subviews = view.map.try_load_entries(&keys).await?;
-    assert!(subviews[0].is_some());
-    Ok(())
-}
-
-#[derive(RootView)]
-pub struct NestedReentrantCollectionMapView<C> {
-    pub map: ReentrantCollectionView<C, String, MapView<C, String, u64>>,
-}
-
-#[tokio::test]
-async fn test_nested_reentrant_collection_map_view() -> anyhow::Result<()> {
-    let context = MemoryContext::new_for_testing(());
-    {
-        let mut view = NestedReentrantCollectionMapView::load(context.clone()).await?;
-        {
-            let mut subview = view.map.try_load_entry_mut("Bonjour").await?;
-            subview.insert("A bientot", 49)?;
-        }
-        view.save().await?;
-    }
-    let view = NestedReentrantCollectionMapView::load(context).await?;
-    let keys = vec!["Bonjour".to_string()];
-    let subviews = view.map.try_load_entries(&keys).await?;
-    assert!(subviews[0].is_some());
-    Ok(())
-}
-
 #[derive(CryptoHashRootView)]
 pub struct ByteMapStateView<C> {
     pub map: ByteMapView<C, u8>,