Remove CommonStoreInternalConfig and CommonStoreConfig (#4115)

## Motivation

Storage backends should only receive configuration options that they
actually use.

## Proposal

* Remove `CommonStoreInternalConfig` and `CommonStoreConfig` step by
step
* Add missing CLI options to  `linera-storage-service`

Nothing changes from the user point of view.

This temporary creates a little more code duplication between
`linera-service` and the binaries in `linera-indexer` and
`linera-storage-service` but I think that's fine and should be fixed
over time.

Future work:
* We may re-evaluate whether `max_stream_queries` is best implemented as
a method of `ReadableKeyValueStore`.
* I'm still conflicted about `max_concurrent_queries` and would very
much like to remove it.
* It's not clear what happens if we pass the wrong replication factor
for ScyllaDb.
* We may re-evaluate whether `linera-indexer` and
`linera-storage-service` belong to separate crates.

Follows #4111

## Test Plan

CI

Author: ma2bd

linera-indexer/lib/src/rocks_db.rs

@@ -6,8 +6,11 @@ use std::path::PathBuf;
 use clap::Parser as _;
 use linera_views::{
     lru_caching::StorageCacheConfig,
-    rocks_db::{PathWithGuard, RocksDbSpawnMode, RocksDbStore, RocksDbStoreConfig},
-    store::{AdminKeyValueStore, CommonStoreConfig},
+    rocks_db::{
+        PathWithGuard, RocksDbSpawnMode, RocksDbStore, RocksDbStoreConfig,
+        RocksDbStoreInternalConfig,
+    },
+    store::AdminKeyValueStore as _,
 };
 
 use crate::{
@@ -56,18 +59,20 @@ impl RocksDbRunner {
             max_entry_size: config.client.max_entry_size,
             max_cache_entries: config.client.max_cache_entries,
         };
-        let common_config = CommonStoreConfig {
-            max_concurrent_queries: config.client.max_concurrent_queries,
-            max_stream_queries: config.client.max_stream_queries,
-            storage_cache_config,
-            replication_factor: 1,
-        };
         let path_buf = config.client.storage.as_path().to_path_buf();
         let path_with_guard = PathWithGuard::new(path_buf);
         // The tests are run in single threaded mode, therefore we need
         // to use the safe default value of SpawnBlocking.
         let spawn_mode = RocksDbSpawnMode::SpawnBlocking;
-        let store_config = RocksDbStoreConfig::new(spawn_mode, path_with_guard, common_config);
+        let inner_config = RocksDbStoreInternalConfig {
+            spawn_mode,
+            path_with_guard,
+            max_stream_queries: config.client.max_stream_queries,
+        };
+        let store_config = RocksDbStoreConfig {
+            inner_config,
+            storage_cache_config,
+        };
         let namespace = config.client.namespace.clone();
         let store = RocksDbStore::maybe_create_and_connect(&store_config, &namespace).await?;
         Self::new(config, store).await

linera-indexer/lib/src/scylla_db.rs

@@ -3,8 +3,8 @@
 
 use linera_views::{
     lru_caching::StorageCacheConfig,
-    scylla_db::{ScyllaDbStore, ScyllaDbStoreConfig},
-    store::{AdminKeyValueStore, CommonStoreConfig},
+    scylla_db::{ScyllaDbStore, ScyllaDbStoreConfig, ScyllaDbStoreInternalConfig},
+    store::AdminKeyValueStore,
 };
 
 use crate::{
@@ -57,14 +57,17 @@ impl ScyllaDbRunner {
             max_entry_size: config.client.max_entry_size,
             max_cache_entries: config.client.max_cache_entries,
         };
-        let common_config = CommonStoreConfig {
-            max_concurrent_queries: config.client.max_concurrent_queries,
+        let inner_config = ScyllaDbStoreInternalConfig {
+            uri: config.client.uri.clone(),
             max_stream_queries: config.client.max_stream_queries,
-            storage_cache_config,
+            max_concurrent_queries: config.client.max_concurrent_queries,
             replication_factor: config.client.replication_factor,
         };
+        let store_config = ScyllaDbStoreConfig {
+            inner_config,
+            storage_cache_config,
+        };
         let namespace = config.client.table.clone();
-        let store_config = ScyllaDbStoreConfig::new(config.client.uri.clone(), common_config);
         let store = ScyllaDbStore::connect(&store_config, &namespace).await?;
         Self::new(config, store).await
     }

linera-service/src/storage.rs

@@ -14,13 +14,15 @@ use linera_storage_service::{
     common::{ServiceStoreConfig, ServiceStoreInternalConfig},
 };
 #[cfg(feature = "dynamodb")]
-use linera_views::dynamo_db::{DynamoDbStore, DynamoDbStoreConfig};
+use linera_views::dynamo_db::{DynamoDbStore, DynamoDbStoreConfig, DynamoDbStoreInternalConfig};
 #[cfg(feature = "rocksdb")]
-use linera_views::rocks_db::{PathWithGuard, RocksDbSpawnMode, RocksDbStore, RocksDbStoreConfig};
+use linera_views::rocks_db::{
+    PathWithGuard, RocksDbSpawnMode, RocksDbStore, RocksDbStoreConfig, RocksDbStoreInternalConfig,
+};
 use linera_views::{
     lru_caching::StorageCacheConfig,
     memory::{MemoryStore, MemoryStoreConfig},
-    store::{CommonStoreConfig, KeyValueStore},
+    store::KeyValueStore,
 };
 use serde::{Deserialize, Serialize};
 use tracing::error;
@@ -32,7 +34,7 @@ use {
 };
 #[cfg(feature = "scylladb")]
 use {
-    linera_views::scylla_db::{ScyllaDbStore, ScyllaDbStoreConfig},
+    linera_views::scylla_db::{ScyllaDbStore, ScyllaDbStoreConfig, ScyllaDbStoreInternalConfig},
     std::num::NonZeroU16,
     tracing::debug,
 };
@@ -65,36 +67,30 @@ pub struct CommonStorageOptions {
 }
 
 impl CommonStorageOptions {
-    pub fn common_store_config(&self) -> CommonStoreConfig {
-        let storage_cache_config = StorageCacheConfig {
+    pub fn storage_cache_config(&self) -> StorageCacheConfig {
+        StorageCacheConfig {
             max_cache_size: self.storage_max_cache_size,
             max_entry_size: self.storage_max_entry_size,
             max_cache_entries: self.storage_max_cache_entries,
-        };
-        CommonStoreConfig {
-            storage_cache_config,
-            max_concurrent_queries: self.storage_max_concurrent_queries,
-            max_stream_queries: self.storage_max_stream_queries,
-            replication_factor: self.storage_replication_factor,
         }
     }
 }
 
 /// The configuration of the key value store in use.
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub enum StoreConfig {
-    /// The storage service key-value store
-    #[cfg(feature = "storage-service")]
-    Service {
-        config: ServiceStoreConfig,
-        namespace: String,
-    },
     /// The memory key value store
     Memory {
         config: MemoryStoreConfig,
         namespace: String,
         genesis_path: PathBuf,
     },
+    /// The storage service key-value store
+    #[cfg(feature = "storage-service")]
+    Service {
+        config: ServiceStoreConfig,
+        namespace: String,
+    },
     /// The RocksDB key value store
     #[cfg(feature = "rocksdb")]
     RocksDb {
@@ -124,18 +120,18 @@ pub enum StoreConfig {
 #[derive(Clone, Debug)]
 #[cfg_attr(any(test), derive(Eq, PartialEq))]
 pub enum InnerStorageConfig {
-    /// The storage service description.
-    #[cfg(feature = "storage-service")]
-    Service {
-        /// The endpoint used.
-        endpoint: String,
-    },
     /// The memory description.
     Memory {
         /// The path to the genesis configuration. This is needed because we reinitialize
         /// memory databases from the genesis config everytime.
         genesis_path: PathBuf,
     },
+    /// The storage service description.
+    #[cfg(feature = "storage-service")]
+    Service {
+        /// The endpoint used.
+        endpoint: String,
+    },
     /// The RocksDB description.
     #[cfg(feature = "rocksdb")]
     RocksDb {
@@ -430,25 +426,11 @@ impl StorageConfig {
         &self,
         options: &CommonStorageOptions,
     ) -> Result<StoreConfig, anyhow::Error> {
-        let common_config = options.common_store_config();
         let namespace = self.namespace.clone();
         match &self.inner_storage_config {
-            #[cfg(feature = "storage-service")]
-            InnerStorageConfig::Service { endpoint } => {
-                let endpoint = endpoint.clone();
-                let inner_config = ServiceStoreInternalConfig {
-                    endpoint,
-                    common_config: common_config.reduced(),
-                };
-                let config = ServiceStoreConfig {
-                    inner_config,
-                    storage_cache_config: common_config.storage_cache_config,
-                };
-                Ok(StoreConfig::Service { config, namespace })
-            }
             InnerStorageConfig::Memory { genesis_path } => {
                 let config = MemoryStoreConfig {
-                    common_config: common_config.reduced(),
+                    max_stream_queries: options.storage_max_stream_queries,
                 };
                 let genesis_path = genesis_path.clone();
                 Ok(StoreConfig::Memory {
@@ -457,21 +439,58 @@ impl StorageConfig {
                     genesis_path,
                 })
             }
+            #[cfg(feature = "storage-service")]
+            InnerStorageConfig::Service { endpoint } => {
+                let inner_config = ServiceStoreInternalConfig {
+                    endpoint: endpoint.clone(),
+                    max_concurrent_queries: options.storage_max_concurrent_queries,
+                    max_stream_queries: options.storage_max_stream_queries,
+                };
+                let config = ServiceStoreConfig {
+                    inner_config,
+                    storage_cache_config: options.storage_cache_config(),
+                };
+                Ok(StoreConfig::Service { config, namespace })
+            }
             #[cfg(feature = "rocksdb")]
             InnerStorageConfig::RocksDb { path, spawn_mode } => {
-                let path_buf = path.to_path_buf();
-                let path_with_guard = PathWithGuard::new(path_buf);
-                let config = RocksDbStoreConfig::new(*spawn_mode, path_with_guard, common_config);
+                let path_with_guard = PathWithGuard::new(path.to_path_buf());
+                let inner_config = RocksDbStoreInternalConfig {
+                    spawn_mode: *spawn_mode,
+                    path_with_guard,
+                    max_stream_queries: options.storage_max_stream_queries,
+                };
+                let config = RocksDbStoreConfig {
+                    inner_config,
+                    storage_cache_config: options.storage_cache_config(),
+                };
                 Ok(StoreConfig::RocksDb { config, namespace })
             }
             #[cfg(feature = "dynamodb")]
             InnerStorageConfig::DynamoDb { use_dynamodb_local } => {
-                let config = DynamoDbStoreConfig::new(*use_dynamodb_local, common_config);
+                let inner_config = DynamoDbStoreInternalConfig {
+                    use_dynamodb_local: *use_dynamodb_local,
+                    max_concurrent_queries: options.storage_max_concurrent_queries,
+                    max_stream_queries: options.storage_max_stream_queries,
+                };
+                let config = DynamoDbStoreConfig {
+                    inner_config,
+                    storage_cache_config: options.storage_cache_config(),
+                };
                 Ok(StoreConfig::DynamoDb { config, namespace })
             }
             #[cfg(feature = "scylladb")]
             InnerStorageConfig::ScyllaDb { uri } => {
-                let config = ScyllaDbStoreConfig::new(uri.to_string(), common_config);
+                let inner_config = ScyllaDbStoreInternalConfig {
+                    uri: uri.clone(),
+                    max_stream_queries: options.storage_max_stream_queries,
+                    max_concurrent_queries: options.storage_max_concurrent_queries,
+                    replication_factor: options.storage_replication_factor,
+                };
+                let config = ScyllaDbStoreConfig {
+                    inner_config,
+                    storage_cache_config: options.storage_cache_config(),
+                };
                 Ok(StoreConfig::ScyllaDb { config, namespace })
             }
             #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
@@ -480,12 +499,27 @@ impl StorageConfig {
                 spawn_mode,
                 uri,
             } => {
-                let first_config = RocksDbStoreConfig::new(
-                    *spawn_mode,
-                    path_with_guard.clone(),
-                    common_config.clone(),
-                );
-                let second_config = ScyllaDbStoreConfig::new(uri.to_string(), common_config);
+                let inner_config = RocksDbStoreInternalConfig {
+                    spawn_mode: *spawn_mode,
+                    path_with_guard: path_with_guard.clone(),
+                    max_stream_queries: options.storage_max_stream_queries,
+                };
+                let first_config = RocksDbStoreConfig {
+                    inner_config,
+                    storage_cache_config: options.storage_cache_config(),
+                };
+
+                let inner_config = ScyllaDbStoreInternalConfig {
+                    uri: uri.clone(),
+                    max_stream_queries: options.storage_max_stream_queries,
+                    max_concurrent_queries: options.storage_max_concurrent_queries,
+                    replication_factor: options.storage_replication_factor,
+                };
+                let second_config = ScyllaDbStoreConfig {
+                    inner_config,
+                    storage_cache_config: options.storage_cache_config(),
+                };
+
                 let config = DualStoreConfig {
                     first_config,
                     second_config,
@@ -578,9 +612,8 @@ impl StoreConfig {
                 namespace,
                 genesis_path,
             } => {
-                let store_config = MemoryStoreConfig::new(config.common_config.max_stream_queries);
                 let mut storage = DbStorage::<MemoryStore, _>::maybe_create_and_connect(
-                    &store_config,
+                    &config,
                     &namespace,
                     wasm_runtime,
                 )

linera-storage-service/src/client.rs

@@ -19,10 +19,7 @@ use linera_views::store::TestKeyValueStore;
 use linera_views::{
     batch::{Batch, WriteOperation},
     lru_caching::LruCachingStore,
-    store::{
-        AdminKeyValueStore, CommonStoreInternalConfig, ReadableKeyValueStore, WithError,
-        WritableKeyValueStore,
-    },
+    store::{AdminKeyValueStore, ReadableKeyValueStore, WithError, WritableKeyValueStore},
     FutureSyncExt,
 };
 use serde::de::DeserializeOwned;
@@ -430,11 +427,10 @@ impl AdminKeyValueStore for ServiceStoreClientInternal {
 
     async fn connect(config: &Self::Config, namespace: &str) -> Result<Self, ServiceStoreError> {
         let semaphore = config
-            .common_config
             .max_concurrent_queries
             .map(|n| Arc::new(Semaphore::new(n)));
         let namespace = bcs::to_bytes(namespace)?;
-        let max_stream_queries = config.common_config.max_stream_queries;
+        let max_stream_queries = config.max_stream_queries;
         let mut start_key = vec![KeyPrefix::Key as u8];
         start_key.extend(&namespace);
         let prefix_len = namespace.len() + 1;
@@ -557,15 +553,10 @@ impl TestKeyValueStore for ServiceStoreClientInternal {
 pub fn service_config_from_endpoint(
     endpoint: &str,
 ) -> Result<ServiceStoreInternalConfig, ServiceStoreError> {
-    let common_config = CommonStoreInternalConfig {
+    Ok(ServiceStoreInternalConfig {
+        endpoint: endpoint.to_string(),
         max_concurrent_queries: None,
         max_stream_queries: 100,
-        replication_factor: 1,
-    };
-    let endpoint = endpoint.to_string();
-    Ok(ServiceStoreInternalConfig {
-        endpoint,
-        common_config,
     })
 }
 

linera-storage-service/src/common.rs

@@ -4,10 +4,7 @@
 use std::path::PathBuf;
 
 use linera_base::command::resolve_binary;
-use linera_views::{
-    lru_caching::LruCachingConfig,
-    store::{CommonStoreInternalConfig, KeyValueStoreError},
-};
+use linera_views::{lru_caching::LruCachingConfig, store::KeyValueStoreError};
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 use tonic::Status;
@@ -85,8 +82,10 @@ pub fn storage_service_test_endpoint() -> Result<String, ServiceStoreError> {
 pub struct ServiceStoreInternalConfig {
     /// The endpoint used by the shared store
     pub endpoint: String,
-    /// The common configuration code
-    pub common_config: CommonStoreInternalConfig,
+    /// Maximum number of concurrent database queries allowed for this client.
+    pub max_concurrent_queries: Option<usize>,
+    /// Preferred buffer size for async streams.
+    pub max_stream_queries: usize,
 }
 
 /// The config type