Correct a bug in root_key handling in RocksDb. (#4859)

## Motivation

The bug addressed in PR #4858 is also present in `RocksDb`.

## Proposal

The same solution is applied. The test code inserted in
`linera-storage-service` is added to `root_key_admin_test` so that all
storage are now tested on this possible issue.

## Test Plan

The CI.

## Release Plan

This PR should NOT be merged into TestNet Conway.

The reason is that while the bug being addressed is real, correcting it
means that we lose access to the
RocksDB storage, through the standard API.

Backport in TestNet Conway after main.

## Links

None.

Author: MathieuDutSik

linera-storage-service/tests/store_test.rs

@@ -4,13 +4,10 @@
 #![cfg(feature = "storage-service")]
 
 use anyhow::Result;
-use linera_storage_service::client::{StorageServiceDatabase, StorageServiceDatabaseInternal};
+use linera_storage_service::client::StorageServiceDatabaseInternal;
 use linera_views::{
     batch::Batch,
-    store::{
-        KeyValueDatabase as _, ReadableKeyValueStore as _, TestKeyValueDatabase as _,
-        WritableKeyValueStore as _,
-    },
+    store::TestKeyValueDatabase as _,
     test_utils::{
         get_random_byte_vector, get_random_test_scenarios, namespace_admin_test,
         root_key_admin_test, run_reads, run_test_batch_from_blank, run_writes_from_blank,
@@ -63,17 +60,3 @@ async fn test_storage_service_big_raw_write() -> Result<()> {
     run_test_batch_from_blank(&store, key_prefix, batch).await;
     Ok(())
 }
-
-#[tokio::test]
-async fn test_storage_service_open_shared() -> Result<()> {
-    let database = StorageServiceDatabase::connect_test_namespace().await?;
-    let store1 = database.open_shared(&[2, 3, 4, 5])?;
-    let mut batch = Batch::new();
-    batch.put_key_value_bytes(vec![6, 7], vec![123, 135]);
-    store1.write_batch(batch).await?;
-
-    let store2 = database.open_shared(&[])?;
-    let key_values = store2.find_key_values_by_prefix(&[2]).await?;
-    assert_eq!(key_values.len(), 0);
-    Ok(())
-}

linera-views/src/backends/rocks_db.rs

@@ -547,7 +547,7 @@ impl KeyValueDatabase for RocksDbDatabaseInternal {
 
     fn open_shared(&self, root_key: &[u8]) -> Result<Self::Store, RocksDbStoreInternalError> {
         let mut start_key = ROOT_KEY_DOMAIN.to_vec();
-        start_key.extend(root_key);
+        start_key.extend(bcs::to_bytes(root_key)?);
         let mut executor = self.executor.clone();
         executor.start_key = start_key;
         Ok(RocksDbStoreInternal {
@@ -585,7 +585,13 @@ impl KeyValueDatabase for RocksDbDatabaseInternal {
     async fn list_root_keys(&self) -> Result<Vec<Vec<u8>>, RocksDbStoreInternalError> {
         let mut store = self.open_shared(&[])?;
         store.executor.start_key = vec![STORED_ROOT_KEYS_PREFIX];
-        store.find_keys_by_prefix(&[]).await
+        let bcs_root_keys = store.find_keys_by_prefix(&[]).await?;
+        let mut root_keys = Vec::new();
+        for bcs_root_key in bcs_root_keys {
+            let root_key = bcs::from_bytes::<Vec<u8>>(&bcs_root_key)?;
+            root_keys.push(root_key);
+        }
+        Ok(root_keys)
     }
 
     async fn delete_all(config: &Self::Config) -> Result<(), RocksDbStoreInternalError> {

linera-views/src/test_utils/mod.rs

@@ -834,6 +834,21 @@ where
         }
     }
     assert_eq!(keys, read_keys);
+
+    // Checking prefix freeness of the (root_key, key). This is a
+    // common problem that needs to be tested.
+    let database = D::connect_test_namespace().await.expect("database");
+    let store1 = database.open_shared(&[2, 3, 4, 5]).expect("store1");
+    let mut batch = Batch::new();
+    batch.put_key_value_bytes(vec![6, 7], vec![123, 135]);
+    store1.write_batch(batch).await.expect("write_batch");
+
+    let store2 = database.open_shared(&[]).expect("store2");
+    let key_values = store2
+        .find_key_values_by_prefix(&[2])
+        .await
+        .expect("key_values");
+    assert_eq!(key_values.len(), 0);
 }
 
 /// A store can be in exclusive access where it stores the absence of values