Make ScyllaDB replication factor configurable

ndr-ds · ndr-ds · commit e6786ca8ec92 · 2025-05-02T16:46:28.000-03:00
diff --git a/CLI.md b/CLI.md
@@ -165,6 +165,9 @@ A Byzantine-fault tolerant sidechain with low-latency finality and high throughp
 * `--wasm-runtime <WASM_RUNTIME>` — The WebAssembly runtime to use
 * `--tokio-threads <TOKIO_THREADS>` — The number of Tokio worker threads to use
 * `--tokio-blocking-threads <TOKIO_BLOCKING_THREADS>` — The number of Tokio blocking threads to use
+* `--storage-replication-factor <STORAGE_REPLICATION_FACTOR>` — The replication factor for the keyspace
+
+  Default value: `1`
 
 
 
diff --git a/docker/Dockerfile b/docker/Dockerfile
@@ -116,6 +116,7 @@ COPY --from=binaries \
 COPY --chmod=755 \
     docker/server-entrypoint.sh \
     docker/server-init.sh \
+    docker/proxy-entrypoint.sh \
     docker/proxy-init.sh \
     docker/compose-server-entrypoint.sh \
     docker/compose-proxy-entrypoint.sh \
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -42,7 +42,7 @@ services:
   shard-init:
     image: "${LINERA_IMAGE:-linera}"
     container_name: shard-init
-    command: [ "./compose-server-init.sh", "scylladb:tcp:scylla:9042" ]
+    command: [ "./compose-server-init.sh", "scylladb:tcp:scylla:9042", "1" ]
     volumes:
       - .:/config
     depends_on:
diff --git a/docker/proxy-entrypoint.sh b/docker/proxy-entrypoint.sh
@@ -0,0 +1,9 @@
+#!/bin/sh
+
+storage_replication_factor=$1
+
+exec ./linera-proxy \
+    --storage scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042 \
+    --genesis /config/genesis.json \
+    --storage-replication-factor $storage_replication_factor \
+    /config/server.json
diff --git a/docker/server-entrypoint.sh b/docker/server-entrypoint.sh
@@ -1,5 +1,7 @@
 #!/bin/sh
 
+storage_replication_factor=$1
+
 # Extract the ordinal number from the pod hostname
 ORDINAL="${HOSTNAME##*-}"
 storage=$1
@@ -8,4 +10,5 @@ exec ./linera-server run \
   --storage $storage \
   --server /config/server.json \
   --shard $ORDINAL \
-  --genesis /config/genesis.json
+  --genesis /config/genesis.json \
+  --storage-replication-factor $storage_replication_factor
diff --git a/docker/server-init.sh b/docker/server-init.sh
@@ -1,6 +1,7 @@
 #!/bin/sh
 
 storage=$1
+storage_replication_factor=$2
 
 while true; do
   ./linera storage check-existence --storage $storage
@@ -13,7 +14,8 @@ while true; do
     echo "Database does not exist, attempting to initialize..."
     if ./linera-server initialize \
       --storage $storage \
-      --genesis /config/genesis.json; then
+      --genesis /config/genesis.json \
+      --storage-replication-factor $storage_replication_factor; then
       echo "Initialization successful."
       exit 0
     else
diff --git a/kubernetes/linera-validator/helmfile.yaml b/kubernetes/linera-validator/helmfile.yaml
@@ -66,7 +66,7 @@ releases:
       - scylla-manager/scylla-manager
       - scylla-operator/scylla-operator
     values:
-      - {{ env "LINERA_HELMFILE_VALUES_SCYLLA" | default "scylla.values.yaml" }}
+      - {{ env "LINERA_HELMFILE_VALUES_SCYLLA" | default "scylla.values.yaml.gotmpl" }}
   - name: scylla-manager
     version: v1.13.0
     namespace: scylla-manager
diff --git a/kubernetes/linera-validator/scylla.values.yaml.gotmpl b/kubernetes/linera-validator/scylla.values.yaml.gotmpl
@@ -4,7 +4,7 @@ sysctls:
 datacenter: validator
 racks:
   - name: rack
-    members: 1
+    members: {{ env "LINERA_HELMFILE_SET_SCYLLA_REPLICATION_FACTOR" | default 1 }}
     scyllaConfig: "scylla-config"
     storage:
       capacity: 2Gi
diff --git a/kubernetes/linera-validator/templates/proxy.yaml b/kubernetes/linera-validator/templates/proxy.yaml
@@ -69,15 +69,7 @@ spec:
               name: linera-port
             - containerPort: 20100
               name: linera-port-int
-          command:
-            [
-              "./linera-proxy",
-              "--storage",
-              "scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042",
-              "--genesis",
-              "/config/genesis.json",
-            ]
-          args: ["/config/server.json"]
+          command: ["./proxy-entrypoint.sh", {{ .Values.storageReplicationFactor | quote }}]
           env:
             - name: RUST_LOG
               value: {{ .Values.logLevel }}
diff --git a/kubernetes/linera-validator/templates/shards.yaml b/kubernetes/linera-validator/templates/shards.yaml
@@ -35,7 +35,7 @@ spec:
         - name: linera-server-initializer
           image: {{ .Values.lineraImage }}
           imagePullPolicy: {{ .Values.lineraImagePullPolicy }}
-          command: ["./server-init.sh", {{ .Values.storage | quote }}]
+          command: ["./server-init.sh", {{ .Values.storage | quote }}, {{ .Values.storageReplicationFactor | quote }}]
           env:
             - name: RUST_LOG
               value: {{ .Values.logLevel }}
@@ -53,7 +53,7 @@ spec:
         - name: linera-server
           image: {{ .Values.lineraImage }}
           imagePullPolicy: {{ .Values.lineraImagePullPolicy }}
-          command: ["./server-entrypoint.sh", {{ .Values.storage | quote }}]
+          command: ["./server-entrypoint.sh", {{ .Values.storage | quote }}, {{ .Values.storageReplicationFactor | quote }}]
           env:
             - name: RUST_LOG
               value: {{ .Values.logLevel }}
diff --git a/kubernetes/linera-validator/values-local.yaml.gotmpl b/kubernetes/linera-validator/values-local.yaml.gotmpl
@@ -11,6 +11,7 @@ metricsPort: 21100
 numShards: {{ env "LINERA_HELMFILE_SET_NUM_SHARDS" | default 10 }}
 storage: {{ env "LINERA_HELMFILE_SET_STORAGE" | default "scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042" }}
 runLocalCsiSetup: {{ env "LINERA_RUN_LOCAL_CSI_SETUP" | default "false" }}
+storageReplicationFactor: {{ env "LINERA_HELMFILE_SET_STORAGE_REPLICATION_FACTOR" | default 1 }}
 
 # Loki
 loki-stack:
diff --git a/linera-indexer/lib/src/rocks_db.rs b/linera-indexer/lib/src/rocks_db.rs
@@ -60,6 +60,7 @@ impl RocksDbRunner {
             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);
diff --git a/linera-indexer/lib/src/scylla_db.rs b/linera-indexer/lib/src/scylla_db.rs
@@ -41,6 +41,10 @@ pub struct ScyllaDbConfig {
     /// The maximal number of entries in the storage cache.
     #[arg(long, default_value = "1000")]
     pub max_cache_entries: usize,
+
+    /// The replication factor for the keyspace
+    #[arg(long, default_value = "1")]
+    pub replication_factor: u32,
 }
 
 pub type ScyllaDbRunner = Runner<ScyllaDbStore, ScyllaDbConfig>;
@@ -57,6 +61,7 @@ impl ScyllaDbRunner {
             max_concurrent_queries: config.client.max_concurrent_queries,
             max_stream_queries: config.client.max_stream_queries,
             storage_cache_config,
+            replication_factor: config.client.replication_factor,
         };
         let namespace = config.client.table.clone();
         let store_config = ScyllaDbStoreConfig::new(config.client.uri.clone(), common_config);
diff --git a/linera-service/src/linera-exporter/main.rs b/linera-service/src/linera-exporter/main.rs
@@ -96,6 +96,7 @@ impl ExporterOptions {
             max_concurrent_queries: self.max_concurrent_queries,
             max_stream_queries: self.max_stream_queries,
             storage_cache_config,
+            replication_factor: 1,
         };
 
         let mut runtime_builder = match self.tokio_threads {
diff --git a/linera-service/src/linera/main.rs b/linera-service/src/linera/main.rs
@@ -1503,6 +1503,10 @@ struct ClientOptions {
     /// Subcommand.
     #[command(subcommand)]
     command: ClientCommand,
+
+    /// The replication factor for the keyspace
+    #[arg(long, default_value = "1")]
+    storage_replication_factor: u32,
 }
 
 impl ClientOptions {
@@ -1520,6 +1524,7 @@ impl ClientOptions {
             max_concurrent_queries: self.max_concurrent_queries,
             max_stream_queries: self.max_stream_queries,
             storage_cache_config,
+            replication_factor: self.storage_replication_factor,
         }
     }
 
diff --git a/linera-service/src/proxy/main.rs b/linera-service/src/proxy/main.rs
@@ -95,6 +95,10 @@ pub struct ProxyOptions {
     /// Path to the file describing the initial user chains (aka genesis state)
     #[arg(long = "genesis")]
     genesis_config_path: PathBuf,
+
+    /// The replication factor for the keyspace
+    #[arg(long, default_value = "1")]
+    storage_replication_factor: u32,
 }
 
 /// A Linera Proxy, either gRPC or over 'Simple Transport', meaning TCP or UDP.
@@ -409,6 +413,7 @@ impl ProxyOptions {
             max_concurrent_queries: self.max_concurrent_queries,
             max_stream_queries: self.max_stream_queries,
             storage_cache_config,
+            replication_factor: self.storage_replication_factor,
         };
         let genesis_config: GenesisConfig = util::read_json(&self.genesis_config_path)?;
         let store_config = self.storage_config.add_common_config(common_config).await?;
diff --git a/linera-service/src/server.rs b/linera-service/src/server.rs
@@ -387,6 +387,10 @@ enum ServerCommand {
         /// The maximal number of entries in the storage cache.
         #[arg(long, default_value = "1000")]
         max_cache_entries: usize,
+
+        /// The replication factor for the storage.
+        #[arg(long, default_value = "1")]
+        storage_replication_factor: u32,
     },
 
     /// Act as a trusted third-party and generate all server configurations
@@ -436,6 +440,10 @@ enum ServerCommand {
         /// The maximal number of entries in the storage cache.
         #[arg(long, default_value = "1000")]
         max_cache_entries: usize,
+
+        /// The replication factor for the storage.
+        #[arg(long, default_value = "1")]
+        storage_replication_factor: u32,
     },
 
     /// Replaces the configurations of the shards by following the given template.
@@ -538,6 +546,7 @@ async fn run(options: ServerOptions) {
             max_cache_size,
             max_entry_size,
             max_cache_entries,
+            storage_replication_factor,
         } => {
             linera_version::VERSION_INFO.log();
 
@@ -564,6 +573,7 @@ async fn run(options: ServerOptions) {
                 max_concurrent_queries,
                 max_stream_queries,
                 storage_cache_config,
+                replication_factor: storage_replication_factor,
             };
             let store_config = storage_config
                 .add_common_config(common_config)
@@ -626,6 +636,7 @@ async fn run(options: ServerOptions) {
             max_cache_size,
             max_entry_size,
             max_cache_entries,
+            storage_replication_factor,
         } => {
             let genesis_config: GenesisConfig =
                 util::read_json(&genesis_config_path).expect("Failed to read initial chain config");
@@ -638,6 +649,7 @@ async fn run(options: ServerOptions) {
                 max_concurrent_queries,
                 max_stream_queries,
                 storage_cache_config,
+                replication_factor: storage_replication_factor,
             };
             let store_config = storage_config
                 .add_common_config(common_config)
diff --git a/linera-storage-service/src/client.rs b/linera-storage-service/src/client.rs
@@ -547,6 +547,7 @@ pub fn service_config_from_endpoint(
     let common_config = CommonStoreInternalConfig {
         max_concurrent_queries: None,
         max_stream_queries: 100,
+        replication_factor: 1,
     };
     let endpoint = endpoint.to_string();
     Ok(ServiceStoreInternalConfig {
diff --git a/linera-views/src/backends/dynamo_db.rs b/linera-views/src/backends/dynamo_db.rs
@@ -1159,6 +1159,7 @@ impl TestKeyValueStore for JournalingKeyValueStore<DynamoDbStoreInternal> {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: Some(TEST_DYNAMO_DB_MAX_CONCURRENT_QUERIES),
             max_stream_queries: TEST_DYNAMO_DB_MAX_STREAM_QUERIES,
+            replication_factor: 1,
         };
         Ok(DynamoDbStoreInternalConfig {
             use_dynamodb_local: true,
diff --git a/linera-views/src/backends/memory.rs b/linera-views/src/backends/memory.rs
@@ -32,11 +32,12 @@ pub struct MemoryStoreConfig {
 }
 
 impl MemoryStoreConfig {
-    /// Creates a `MemoryStoreConfig`. `max_concurrent_queries` and `cache_size` are not used.
+    /// Creates a `MemoryStoreConfig`. `max_concurrent_queries`, `cache_size` and `replication_factor` are not used.
     pub fn new(max_stream_queries: usize) -> Self {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: None,
             max_stream_queries,
+            replication_factor: 1,
         };
         Self { common_config }
     }
@@ -281,6 +282,7 @@ impl MemoryStore {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: None,
             max_stream_queries,
+            replication_factor: 1,
         };
         let config = MemoryStoreConfig { common_config };
         let kill_on_drop = false;
@@ -296,6 +298,7 @@ impl MemoryStore {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: None,
             max_stream_queries,
+            replication_factor: 1,
         };
         let config = MemoryStoreConfig { common_config };
         let kill_on_drop = true;
@@ -323,6 +326,7 @@ impl AdminKeyValueStore for MemoryStore {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: None,
             max_stream_queries,
+            replication_factor: 1,
         };
         let config = MemoryStoreConfig { common_config };
         let mut memory_stores = MEMORY_STORES
@@ -384,6 +388,7 @@ impl TestKeyValueStore for MemoryStore {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: None,
             max_stream_queries,
+            replication_factor: 1,
         };
         Ok(MemoryStoreConfig { common_config })
     }
diff --git a/linera-views/src/backends/rocks_db.rs b/linera-views/src/backends/rocks_db.rs
@@ -579,6 +579,7 @@ impl TestKeyValueStore for RocksDbStoreInternal {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: None,
             max_stream_queries: TEST_ROCKS_DB_MAX_STREAM_QUERIES,
+            replication_factor: 1,
         };
         let spawn_mode = RocksDbSpawnMode::get_spawn_mode_from_runtime();
         Ok(RocksDbStoreInternalConfig {
diff --git a/linera-views/src/backends/scylla_db.rs b/linera-views/src/backends/scylla_db.rs
@@ -811,10 +811,13 @@ impl AdminKeyValueStore for ScyllaDbStoreInternal {
             .boxed()
             .await?;
         // Create a keyspace if it doesn't exist
-        let query = "CREATE KEYSPACE IF NOT EXISTS kv WITH REPLICATION = { \
-            'class' : 'SimpleStrategy', \
-            'replication_factor' : 1 \
-        }";
+        let query = format!(
+            "CREATE KEYSPACE IF NOT EXISTS kv WITH REPLICATION = {{ \
+                'class' : 'SimpleStrategy', \
+                'replication_factor' : {} \
+            }}",
+            config.common_config.replication_factor
+        );
 
         // Execute the query
         let prepared = session.prepare(query).await?;
@@ -889,6 +892,7 @@ impl TestKeyValueStore for JournalingKeyValueStore<ScyllaDbStoreInternal> {
         let common_config = CommonStoreInternalConfig {
             max_concurrent_queries: Some(TEST_SCYLLA_DB_MAX_CONCURRENT_QUERIES),
             max_stream_queries: TEST_SCYLLA_DB_MAX_STREAM_QUERIES,
+            replication_factor: 1,
         };
         Ok(ScyllaDbStoreInternalConfig { uri, common_config })
     }
diff --git a/linera-views/src/store.rs b/linera-views/src/store.rs
@@ -23,6 +23,8 @@ pub struct CommonStoreInternalConfig {
     pub max_concurrent_queries: Option<usize>,
     /// The number of streams used for the async streams.
     pub max_stream_queries: usize,
+    /// The replication factor for the keyspace
+    pub replication_factor: u32,
 }
 
 /// The common initialization parameters for the `KeyValueStore`
@@ -34,6 +36,8 @@ pub struct CommonStoreConfig {
     pub max_stream_queries: usize,
     /// The cache size being used.
     pub storage_cache_config: StorageCacheConfig,
+    /// The replication factor for the keyspace
+    pub replication_factor: u32,
 }
 
 impl CommonStoreConfig {
@@ -42,6 +46,7 @@ impl CommonStoreConfig {
         CommonStoreInternalConfig {
             max_concurrent_queries: self.max_concurrent_queries,
             max_stream_queries: self.max_stream_queries,
+            replication_factor: self.replication_factor,
         }
     }
 }
@@ -52,6 +57,7 @@ impl Default for CommonStoreConfig {
             max_concurrent_queries: None,
             max_stream_queries: 10,
             storage_cache_config: DEFAULT_STORAGE_CACHE_CONFIG,
+            replication_factor: 1,
         }
     }
 }
