Add the root_key to the MissingEntries error. (#4916)

## Motivation

We have some error of `MissingEntries` occurring in the logs. We do not
know what is the relevant root key.
So, we add it here.

## Proposal

The following has been done:
* The `fn root_key(&self) -> Result<Vec<u8>, Self::Error>` is added to
the trait definition. We need a `Result` because of the deserialization
of the keys in RocksDB and StorageService.
* For RocksDB, ScyllaDB, DynamoDB, and StorageService, this did not add
any new data.
* The key is not added to the error since this would force some clone
operation.



## Test Plan

The CI.

## Release Plan

The change could be backported to TestNet Conway.

## Links

None.

Author: MathieuDutSik

linera-sdk/src/views/system_api.rs

@@ -111,6 +111,10 @@ impl ReadableKeyValueStore for KeyValueStore {
         1
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, KeyValueStoreError> {
+        Ok(Vec::new())
+    }
+
     async fn contains_key(&self, key: &[u8]) -> Result<bool, KeyValueStoreError> {
         ensure!(
             key.len() <= Self::MAX_KEY_SIZE,

linera-service/src/proxy/grpc.rs

@@ -369,7 +369,7 @@ where
             ViewError::KeyTooLong | ViewError::ArithmeticError(_) => {
                 Status::out_of_range(err.to_string())
             }
-            ViewError::NotFound(_) | ViewError::MissingEntries => {
+            ViewError::NotFound(_) | ViewError::MissingEntries(_) => {
                 Status::not_found(err.to_string())
             }
         };

linera-storage-service/src/client.rs

@@ -98,6 +98,11 @@ impl ReadableKeyValueStore for StorageServiceStoreInternal {
         self.max_stream_queries
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, StorageServiceStoreError> {
+        let root_key = bcs::from_bytes(&self.start_key[self.prefix_len..])?;
+        Ok(root_key)
+    }
+
     async fn read_value_bytes(
         &self,
         key: &[u8],

linera-views/src/backends/dual.rs

@@ -96,6 +96,13 @@ where
         }
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, Self::Error> {
+        Ok(match self {
+            Self::First(store) => store.root_key().map_err(DualStoreError::First)?,
+            Self::Second(store) => store.root_key().map_err(DualStoreError::Second)?,
+        })
+    }
+
     async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
         let result = match self {
             Self::First(store) => store

linera-views/src/backends/dynamo_db.rs

@@ -803,6 +803,11 @@ impl ReadableKeyValueStore for DynamoDbStoreInternal {
         self.max_stream_queries
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, DynamoDbStoreInternalError> {
+        assert!(self.start_key.starts_with(EMPTY_ROOT_KEY));
+        Ok(self.start_key[EMPTY_ROOT_KEY.len()..].to_vec())
+    }
+
     async fn read_value_bytes(
         &self,
         key: &[u8],

linera-views/src/backends/indexed_db.rs

@@ -119,6 +119,12 @@ impl ReadableKeyValueStore for IndexedDbStore {
         self.max_stream_queries
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, IndexedDbStoreError> {
+        assert!(self.start_key.starts_with(&ROOT_KEY_DOMAIN));
+        let root_key = bcs::from_bytes(&self.start_key[ROOT_KEY_DOMAIN.len()..])?;
+        Ok(root_key)
+    }
+
     async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, IndexedDbStoreError> {
         let key = self.full_key(key);
         let key = js_sys::Uint8Array::from(key.as_slice());
@@ -285,7 +291,7 @@ impl KeyValueDatabase for IndexedDbDatabase {
 
     fn open_shared(&self, root_key: &[u8]) -> Result<Self::Store, IndexedDbStoreError> {
         let mut start_key = ROOT_KEY_DOMAIN.to_vec();
-        start_key.extend(root_key);
+        start_key.extend(bcs::to_bytes(&root_key)?);
         Ok(self.open_internal(start_key))
     }
 

linera-views/src/backends/journaling.rs

@@ -123,6 +123,10 @@ where
         self.store.max_stream_queries()
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, Self::Error> {
+        self.store.root_key()
+    }
+
     async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
         self.store.read_value_bytes(key).await
     }

linera-views/src/backends/lru_caching.rs

@@ -157,6 +157,10 @@ where
         self.store.max_stream_queries()
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, Self::Error> {
+        self.store.root_key()
+    }
+
     async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
         let Some(cache) = &self.cache else {
             return self.store.read_value_bytes(key).await;

linera-views/src/backends/memory.rs

@@ -73,6 +73,7 @@ impl MemoryDatabases {
         let map = store.clone();
         Ok(MemoryStore {
             map,
+            root_key: root_key.to_vec(),
             max_stream_queries,
         })
     }
@@ -111,6 +112,8 @@ static MEMORY_DATABASES: LazyLock<Mutex<MemoryDatabases>> =
 pub struct MemoryStore {
     /// The map used for storing the data.
     map: Arc<RwLock<MemoryStoreMap>>,
+    /// The root key.
+    root_key: Vec<u8>,
     /// The maximum number of queries used for a stream.
     max_stream_queries: usize,
 }
@@ -130,6 +133,10 @@ impl ReadableKeyValueStore for MemoryStore {
         self.max_stream_queries
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, MemoryStoreError> {
+        Ok(self.root_key.clone())
+    }
+
     async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, MemoryStoreError> {
         let map = self
             .map
@@ -247,6 +254,7 @@ impl MemoryStore {
     pub fn new_for_testing() -> Self {
         Self {
             map: Arc::default(),
+            root_key: Vec::new(),
             max_stream_queries: TEST_MEMORY_MAX_STREAM_QUERIES,
         }
     }

linera-views/src/backends/metering.rs

@@ -312,6 +312,10 @@ where
         self.store.max_stream_queries()
     }
 
+    fn root_key(&self) -> Result<Vec<u8>, Self::Error> {
+        self.store.root_key()
+    }
+
     async fn read_value_bytes(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
         let _latency = self.counter.read_value_bytes_latency.measure_latency();
         self.counter