Address a bug with the root_keys in linera-storage-service. (#4858)

## Motivation

There is a bug in the handling of the root_keys in the
`linera-storage-service`. The `root_key` is directly appended to the
`start_key`. The result is an object which is not prefix-free.

## Proposal

Replace
```rust
start_key.extend(root_key)
```

by
```rust
start_key.extend(bcs::to_bytes(&root_key)?)
```

by doing so, we no longer have ambiguity in the construction of
`start_key`.


## Test Plan

The CI.
Added a test that fails on main on that precise problem.

## Release Plan

Since, the `linera-storage-service` is not used in TestNet Conway, no
backport is needed.

## Links

None.

Author: MathieuDutSik

linera-storage-service/src/client.rs

@@ -304,14 +304,14 @@ impl WritableKeyValueStore for StorageServiceStoreInternal {
             chunk_size += self.start_key.len();
         }
 
-        let root_key_len = self.start_key.len() - self.prefix_len;
+        let bcs_root_key_len = self.start_key.len() - self.prefix_len;
         for operation in batch.operations {
             let (key_len, value_len) = match &operation {
                 WriteOperation::Delete { key } => (key.len(), 0),
                 WriteOperation::Put { key, value } => (key.len(), value.len()),
                 WriteOperation::DeletePrefix { key_prefix } => (key_prefix.len(), 0),
             };
-            let operation_size = key_len + value_len + root_key_len;
+            let operation_size = key_len + value_len + bcs_root_key_len;
             ensure!(
                 key_len <= MAX_KEY_SIZE,
                 StorageServiceStoreError::KeyTooLong
@@ -488,7 +488,7 @@ impl KeyValueDatabase for StorageServiceDatabaseInternal {
         let max_stream_queries = self.max_stream_queries;
         let mut start_key = vec![KeyPrefix::Key as u8];
         start_key.extend(&self.namespace);
-        start_key.extend(root_key);
+        start_key.extend(bcs::to_bytes(root_key)?);
         let prefix_len = self.namespace.len() + 1;
         Ok(StorageServiceStoreInternal {
             channel,

linera-storage-service/src/server.rs

@@ -186,7 +186,14 @@ impl StorageServer {
     pub async fn list_root_keys(&self, namespace: &[u8]) -> Result<Vec<Vec<u8>>, Status> {
         let mut full_key = vec![KeyPrefix::RootKey as u8];
         full_key.extend(namespace);
-        self.find_keys_by_prefix(&full_key).await
+        let bcs_root_keys = self.find_keys_by_prefix(&full_key).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)
+                .map_err(|e| Status::unknown(format!("Bcs error {e:?} at list_root_keys")))?;
+            root_keys.push(root_key);
+        }
+        Ok(root_keys)
     }
 
     pub async fn delete_all(&self) -> Result<(), Status> {

linera-storage-service/tests/store_test.rs

@@ -4,10 +4,13 @@
 #![cfg(feature = "storage-service")]
 
 use anyhow::Result;
-use linera_storage_service::client::StorageServiceDatabaseInternal;
+use linera_storage_service::client::{StorageServiceDatabase, StorageServiceDatabaseInternal};
 use linera_views::{
     batch::Batch,
-    store::TestKeyValueDatabase as _,
+    store::{
+        KeyValueDatabase as _, ReadableKeyValueStore as _, TestKeyValueDatabase as _,
+        WritableKeyValueStore 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,
@@ -60,3 +63,17 @@ 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(())
+}