From 08b5c0edd626ca66b412615e95ee03f2303af4a2 Mon Sep 17 00:00:00 2001 From: Mathieu Baudet <1105398+ma2bd@users.noreply.github.com> Date: Wed, 10 Jun 2026 02:54:10 +0200 Subject: [PATCH] Add CLI and environment passthrough for RocksDB and ScyllaDB tuning options --- CLI.md | 1 + linera-bridge/src/relay/mod.rs | 1 + linera-indexer/lib/src/rocks_db.rs | 1 + linera-indexer/lib/src/scylla_db.rs | 1 + linera-storage-runtime/src/common_options.rs | 32 ++ linera-storage-runtime/src/storage_config.rs | 16 + linera-storage-runtime/src/store_config.rs | 4 + linera-storage-service/src/server.rs | 1 + linera-views/src/backends/rocks_db.rs | 397 ++++++++++++++++++- linera-views/src/backends/scylla_db.rs | 339 +++++++++++++++- 10 files changed, 757 insertions(+), 36 deletions(-) diff --git a/CLI.md b/CLI.md index 0c361ad2a022..1ce6a40585f6 100644 --- a/CLI.md +++ b/CLI.md @@ -302,6 +302,7 @@ Client implementation and command-line tool for the Linera blockchain * `--storage-replication-factor ` — The replication factor for the keyspace Default value: `1` +* `--rocksdb-opt ` — Extra RocksDB tuning options, passed as repeated `KEY=VALUE` pairs, e.g. `--rocksdb-opt write_buffer_size=268435456 --rocksdb-opt max_open_files=512`. Can also be set via the `LINERA_ROCKSDB_OPTS` environment variable as a comma-separated list. These only affect runtime behavior, never the on-disk storage format; format-defining keys are rejected. Run with an unknown key to see the list of supported options * `--wasm-runtime ` — The WebAssembly runtime to use * `--with-application-logs` — Output log messages from contract execution * `--tokio-threads ` — The number of Tokio worker threads to use diff --git a/linera-bridge/src/relay/mod.rs b/linera-bridge/src/relay/mod.rs index 71fd56f37a45..0245be84877e 100644 --- a/linera-bridge/src/relay/mod.rs +++ b/linera-bridge/src/relay/mod.rs @@ -222,6 +222,7 @@ async fn create_rocksdb_storage( path_with_guard: PathWithGuard::new(path.to_path_buf()), spawn_mode: RocksDbSpawnMode::get_spawn_mode_from_runtime(), max_stream_queries: 10, + tuning_options: Default::default(), }, storage_cache_config: StorageCacheConfig { max_cache_size: 10_000_000, diff --git a/linera-indexer/lib/src/rocks_db.rs b/linera-indexer/lib/src/rocks_db.rs index d5f0298c39e6..f37a7b1f1e69 100644 --- a/linera-indexer/lib/src/rocks_db.rs +++ b/linera-indexer/lib/src/rocks_db.rs @@ -113,6 +113,7 @@ impl RocksDbRunner { spawn_mode, path_with_guard, max_stream_queries: config.client.max_stream_queries, + tuning_options: Default::default(), }; let store_config = RocksDbStoreConfig { inner_config, diff --git a/linera-indexer/lib/src/scylla_db.rs b/linera-indexer/lib/src/scylla_db.rs index 724ec4413b10..608644182996 100644 --- a/linera-indexer/lib/src/scylla_db.rs +++ b/linera-indexer/lib/src/scylla_db.rs @@ -107,6 +107,7 @@ impl ScyllaDbRunner { max_stream_queries: config.client.max_stream_queries, max_concurrent_queries: config.client.max_concurrent_queries, replication_factor: config.client.replication_factor, + tuning_options: Default::default(), }; let store_config = ScyllaDbStoreConfig { inner_config, diff --git a/linera-storage-runtime/src/common_options.rs b/linera-storage-runtime/src/common_options.rs index a6d27eb66516..1ffbc30d9dc5 100644 --- a/linera-storage-runtime/src/common_options.rs +++ b/linera-storage-runtime/src/common_options.rs @@ -73,6 +73,38 @@ pub struct CommonStorageOptions { /// The replication factor for the keyspace #[arg(long, default_value = "1", global = true)] pub storage_replication_factor: u32, + + /// Extra RocksDB tuning options, passed as repeated `KEY=VALUE` pairs, e.g. + /// `--rocksdb-opt write_buffer_size=268435456 --rocksdb-opt max_open_files=512`. + /// Can also be set via the `LINERA_ROCKSDB_OPTS` environment variable as a + /// comma-separated list. These only affect runtime behavior, never the + /// on-disk storage format; format-defining keys are rejected. Run with an + /// unknown key to see the list of supported options. + #[cfg(feature = "rocksdb")] + #[arg( + long = "rocksdb-opt", + value_name = "KEY=VALUE", + env = "LINERA_ROCKSDB_OPTS", + value_delimiter = ',', + global = true + )] + pub rocksdb_opts: Vec, + + /// Extra ScyllaDB driver options, passed as repeated `KEY=VALUE` pairs, e.g. + /// `--scylladb-opt consistency=local_quorum --scylladb-opt request_timeout_ms=5000`. + /// Can also be set via the `LINERA_SCYLLADB_OPTS` environment variable as a + /// comma-separated list. These only affect the driver/session behavior, never + /// the on-disk storage format or table schema. Run with an unknown key to see + /// the list of supported options. + #[cfg(feature = "scylladb")] + #[arg( + long = "scylladb-opt", + value_name = "KEY=VALUE", + env = "LINERA_SCYLLADB_OPTS", + value_delimiter = ',', + global = true + )] + pub scylladb_opts: Vec, } impl CommonStorageOptions { diff --git a/linera-storage-runtime/src/storage_config.rs b/linera-storage-runtime/src/storage_config.rs index 5f63e7103805..0f8742cf5654 100644 --- a/linera-storage-runtime/src/storage_config.rs +++ b/linera-storage-runtime/src/storage_config.rs @@ -332,10 +332,14 @@ impl StorageConfig { #[cfg(feature = "rocksdb")] InnerStorageConfig::RocksDb { path, spawn_mode } => { let path_with_guard = PathWithGuard::new(path.to_path_buf()); + let tuning_options = linera_views::rocks_db::RocksDbTuningOptions::from_kv_pairs( + &options.rocksdb_opts, + )?; let inner_config = linera_views::rocks_db::RocksDbStoreInternalConfig { spawn_mode: *spawn_mode, path_with_guard, max_stream_queries: options.storage_max_stream_queries, + tuning_options, }; let config = linera_views::rocks_db::RocksDbStoreConfig { inner_config, @@ -345,11 +349,15 @@ impl StorageConfig { } #[cfg(feature = "scylladb")] InnerStorageConfig::ScyllaDb { uri } => { + let tuning_options = linera_views::scylla_db::ScyllaDbTuningOptions::from_kv_pairs( + &options.scylladb_opts, + )?; let inner_config = linera_views::scylla_db::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, + tuning_options, }; let config = linera_views::scylla_db::ScyllaDbStoreConfig { inner_config, @@ -363,21 +371,29 @@ impl StorageConfig { spawn_mode, uri, } => { + let tuning_options = linera_views::rocks_db::RocksDbTuningOptions::from_kv_pairs( + &options.rocksdb_opts, + )?; let inner_config = linera_views::rocks_db::RocksDbStoreInternalConfig { spawn_mode: *spawn_mode, path_with_guard: path_with_guard.clone(), max_stream_queries: options.storage_max_stream_queries, + tuning_options, }; let first_config = linera_views::rocks_db::RocksDbStoreConfig { inner_config, storage_cache_config: options.views_storage_cache_config(), }; + let tuning_options = linera_views::scylla_db::ScyllaDbTuningOptions::from_kv_pairs( + &options.scylladb_opts, + )?; let inner_config = linera_views::scylla_db::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, + tuning_options, }; let second_config = linera_views::scylla_db::ScyllaDbStoreConfig { inner_config, diff --git a/linera-storage-runtime/src/store_config.rs b/linera-storage-runtime/src/store_config.rs index 8ab2c2251b43..5baf630aa3ae 100644 --- a/linera-storage-runtime/src/store_config.rs +++ b/linera-storage-runtime/src/store_config.rs @@ -23,6 +23,10 @@ use serde::{Deserialize, Serialize}; use {linera_storage::ChainStatesFirstAssignment, linera_views::backends::dual::DualDatabase}; /// The configuration of the key value store in use. +// The RocksDB / dual variants carry the full backend tuning configuration, which +// is legitimately larger than the other variants. This config is built once at +// startup, so the size disparity is not a concern. +#[allow(clippy::large_enum_variant)] #[derive(Clone, Debug, Deserialize, Serialize)] pub enum StoreConfig { /// The memory key value store diff --git a/linera-storage-service/src/server.rs b/linera-storage-service/src/server.rs index 93e53378c2bc..14d57a9e29e6 100644 --- a/linera-storage-service/src/server.rs +++ b/linera-storage-service/src/server.rs @@ -681,6 +681,7 @@ async fn main() { spawn_mode, path_with_guard, max_stream_queries, + tuning_options: Default::default(), }; let storage_cache_config = StorageCacheConfig { max_cache_size, diff --git a/linera-views/src/backends/rocks_db.rs b/linera-views/src/backends/rocks_db.rs index cde9000b967e..f69a6a957637 100644 --- a/linera-views/src/backends/rocks_db.rs +++ b/linera-views/src/backends/rocks_db.rs @@ -315,6 +315,239 @@ pub struct RocksDbStoreInternalConfig { pub spawn_mode: RocksDbSpawnMode, /// Preferred buffer size for async streams. pub max_stream_queries: usize, + /// Runtime-tunable RocksDB options that override the built-in defaults. + #[serde(default)] + pub tuning_options: RocksDbTuningOptions, +} + +/// Runtime-tunable RocksDB options. +/// +/// These options change only the runtime behavior of RocksDB — memory budgets, +/// parallelism, compaction triggers, and how *new* data is written. They never +/// change the on-disk storage format in a way that would make existing data +/// unreadable, so they are safe to change between restarts of an existing +/// database (options that affect newly written SSTables, such as compression or +/// block size, simply take full effect as old files are rewritten by +/// compaction). Every field defaults to `None`, meaning the built-in default is +/// used. +/// +/// Format-defining options (the prefix extractor, block format version, +/// compaction style, whole-key filtering, …) are deliberately *not* exposed +/// here, and are rejected by [`RocksDbTuningOptions::from_kv_pairs`]. +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +pub struct RocksDbTuningOptions { + /// Size of a single memtable, in bytes (`write_buffer_size`). + pub write_buffer_size: Option, + /// Maximum number of memtables held in memory (`max_write_buffer_number`). + pub max_write_buffer_number: Option, + /// Number of L0 files that triggers a write slowdown. + pub level_zero_slowdown_writes_trigger: Option, + /// Number of L0 files that triggers a write stop. + pub level_zero_stop_writes_trigger: Option, + /// Number of L0 files that triggers compaction. + pub level_zero_file_num_compaction_trigger: Option, + /// Background thread parallelism (`increase_parallelism`). + pub parallelism: Option, + /// Maximum number of concurrent background jobs (flush + compaction). + pub max_background_jobs: Option, + /// Maximum number of threads used by a single compaction job. + pub max_subcompactions: Option, + /// Target SST file size at the base level, in bytes. + pub target_file_size_base: Option, + /// Block cache size, in bytes. + pub block_cache_size: Option, + /// Total memtable memory budget across column families, in bytes. + pub write_buffer_manager_size: Option, + /// Maximum number of open files (`-1` for unlimited). + pub max_open_files: Option, + /// Table block size, in bytes (affects newly written SSTables). + pub block_size: Option, + /// Bloom filter bits per key (affects newly written SSTables). + pub bloom_filter_bits_per_key: Option, + /// Compression algorithm for newly written blocks. + pub compression_type: Option, +} + +/// The block compression algorithm used by RocksDB for newly written blocks. +/// +/// Changing this is safe: RocksDB records the algorithm per block, so existing +/// blocks remain readable regardless of the current setting. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize, Serialize)] +pub enum RocksDbCompressionType { + /// No compression. + None, + /// Snappy. + Snappy, + /// LZ4 (the default). + Lz4, + /// LZ4 high-compression. + Lz4hc, + /// Zstandard. + Zstd, + /// Zlib. + Zlib, + /// Bzip2. + Bz2, +} + +impl RocksDbCompressionType { + fn to_rocksdb(self) -> rocksdb::DBCompressionType { + match self { + RocksDbCompressionType::None => rocksdb::DBCompressionType::None, + RocksDbCompressionType::Snappy => rocksdb::DBCompressionType::Snappy, + RocksDbCompressionType::Lz4 => rocksdb::DBCompressionType::Lz4, + RocksDbCompressionType::Lz4hc => rocksdb::DBCompressionType::Lz4hc, + RocksDbCompressionType::Zstd => rocksdb::DBCompressionType::Zstd, + RocksDbCompressionType::Zlib => rocksdb::DBCompressionType::Zlib, + RocksDbCompressionType::Bz2 => rocksdb::DBCompressionType::Bz2, + } + } + + fn from_token(token: &str) -> Option { + Some(match token.to_ascii_lowercase().as_str() { + "none" => RocksDbCompressionType::None, + "snappy" => RocksDbCompressionType::Snappy, + "lz4" => RocksDbCompressionType::Lz4, + "lz4hc" => RocksDbCompressionType::Lz4hc, + "zstd" => RocksDbCompressionType::Zstd, + "zlib" => RocksDbCompressionType::Zlib, + "bz2" => RocksDbCompressionType::Bz2, + _ => return None, + }) + } +} + +impl RocksDbTuningOptions { + /// Option keys that affect the on-disk storage format. Passing any of these + /// to [`Self::from_kv_pairs`] is an error: they cannot be changed safely on + /// an existing database. + const FORMAT_SENSITIVE_KEYS: &'static [&'static str] = &[ + "prefix_extractor", + "format_version", + "compaction_style", + "whole_key_filtering", + "memtable_prefix_bloom_ratio", + "comparator", + "merge_operator", + "max_key_size", + "max_value_size", + ]; + + /// The list of supported option keys, for use in error messages and docs. + pub fn supported_keys() -> &'static [&'static str] { + &[ + "write_buffer_size", + "max_write_buffer_number", + "level_zero_slowdown_writes_trigger", + "level_zero_stop_writes_trigger", + "level_zero_file_num_compaction_trigger", + "parallelism", + "max_background_jobs", + "max_subcompactions", + "target_file_size_base", + "block_cache_size", + "write_buffer_manager_size", + "max_open_files", + "block_size", + "bloom_filter_bits_per_key", + "compression_type", + ] + } + + /// Parses a list of `KEY=VALUE` strings into typed tuning options. + /// + /// Unknown keys, malformed entries, values that fail to parse, and + /// format-sensitive keys all produce a descriptive error. + pub fn from_kv_pairs(pairs: I) -> Result + where + I: IntoIterator, + S: AsRef, + { + let mut options = Self::default(); + for pair in pairs { + let pair = pair.as_ref(); + let (key, value) = pair.split_once('=').ok_or_else(|| { + RocksDbStoreInternalError::InvalidTuningOption(format!( + "expected `KEY=VALUE`, got `{pair}`" + )) + })?; + let key = key.trim(); + let value = value.trim(); + if Self::FORMAT_SENSITIVE_KEYS.contains(&key) { + return Err(RocksDbStoreInternalError::InvalidTuningOption(format!( + "`{key}` affects the on-disk storage format and cannot be changed at runtime" + ))); + } + match key { + "write_buffer_size" => { + options.write_buffer_size = Some(parse_tuning_value(key, value)?) + } + "max_write_buffer_number" => { + options.max_write_buffer_number = Some(parse_tuning_value(key, value)?) + } + "level_zero_slowdown_writes_trigger" => { + options.level_zero_slowdown_writes_trigger = + Some(parse_tuning_value(key, value)?) + } + "level_zero_stop_writes_trigger" => { + options.level_zero_stop_writes_trigger = Some(parse_tuning_value(key, value)?) + } + "level_zero_file_num_compaction_trigger" => { + options.level_zero_file_num_compaction_trigger = + Some(parse_tuning_value(key, value)?) + } + "parallelism" => options.parallelism = Some(parse_tuning_value(key, value)?), + "max_background_jobs" => { + options.max_background_jobs = Some(parse_tuning_value(key, value)?) + } + "max_subcompactions" => { + options.max_subcompactions = Some(parse_tuning_value(key, value)?) + } + "target_file_size_base" => { + options.target_file_size_base = Some(parse_tuning_value(key, value)?) + } + "block_cache_size" => { + options.block_cache_size = Some(parse_tuning_value(key, value)?) + } + "write_buffer_manager_size" => { + options.write_buffer_manager_size = Some(parse_tuning_value(key, value)?) + } + "max_open_files" => options.max_open_files = Some(parse_tuning_value(key, value)?), + "block_size" => options.block_size = Some(parse_tuning_value(key, value)?), + "bloom_filter_bits_per_key" => { + options.bloom_filter_bits_per_key = Some(parse_tuning_value(key, value)?) + } + "compression_type" => { + options.compression_type = + Some(RocksDbCompressionType::from_token(value).ok_or_else(|| { + RocksDbStoreInternalError::InvalidTuningOption(format!( + "unknown compression type `{value}`; \ + expected one of none, snappy, lz4, lz4hc, zstd, zlib, bz2" + )) + })?) + } + _ => { + return Err(RocksDbStoreInternalError::InvalidTuningOption(format!( + "unknown option `{key}`; supported options are: {}", + Self::supported_keys().join(", ") + ))) + } + } + } + Ok(options) + } +} + +fn parse_tuning_value(key: &str, value: &str) -> Result +where + T: std::str::FromStr, + T::Err: Display, +{ + value.parse::().map_err(|err| { + RocksDbStoreInternalError::InvalidTuningOption(format!( + "invalid value `{value}` for option `{key}`: {err}" + )) + }) } impl RocksDbDatabaseInternal { @@ -365,28 +598,53 @@ impl RocksDbStoreInternal { ); let num_cpus = get_available_cpus(); let total_ram = get_available_memory(&sys); + // Runtime overrides for the defaults below. `None` keeps the default. + let tuning = &config.tuning_options; let mut options = rocksdb::Options::default(); options.create_if_missing(true); options.create_missing_column_families(true); // Flush in-memory buffer to disk more often - options.set_write_buffer_size(WRITE_BUFFER_SIZE); - options.set_max_write_buffer_number(MAX_WRITE_BUFFER_NUMBER); - options.set_compression_type(rocksdb::DBCompressionType::Lz4); - options.set_level_zero_slowdown_writes_trigger(8); - options.set_level_zero_stop_writes_trigger(12); - options.set_level_zero_file_num_compaction_trigger(2); + let write_buffer_size = tuning.write_buffer_size.unwrap_or(WRITE_BUFFER_SIZE); + options.set_write_buffer_size(write_buffer_size); + options.set_max_write_buffer_number( + tuning + .max_write_buffer_number + .unwrap_or(MAX_WRITE_BUFFER_NUMBER), + ); + options.set_compression_type( + tuning + .compression_type + .map_or(rocksdb::DBCompressionType::Lz4, |c| c.to_rocksdb()), + ); + options.set_level_zero_slowdown_writes_trigger( + tuning.level_zero_slowdown_writes_trigger.unwrap_or(8), + ); + options.set_level_zero_stop_writes_trigger( + tuning.level_zero_stop_writes_trigger.unwrap_or(12), + ); + options.set_level_zero_file_num_compaction_trigger( + tuning.level_zero_file_num_compaction_trigger.unwrap_or(2), + ); // We deliberately give RocksDB one background thread *per* CPU so that // flush + (N-1) compactions can hammer the NVMe at full bandwidth while // still leaving enough CPU time for the foreground application threads. - options.increase_parallelism(num_cpus); - options.set_max_background_jobs(num_cpus); - options.set_max_subcompactions(num_cpus as u32); + options.increase_parallelism(tuning.parallelism.unwrap_or(num_cpus)); + options.set_max_background_jobs(tuning.max_background_jobs.unwrap_or(num_cpus)); + options.set_max_subcompactions(tuning.max_subcompactions.unwrap_or(num_cpus as u32)); options.set_level_compaction_dynamic_level_bytes(true); options.set_compaction_style(DBCompactionStyle::Level); - options.set_target_file_size_base(2 * WRITE_BUFFER_SIZE as u64); + options.set_target_file_size_base( + tuning + .target_file_size_base + .unwrap_or(2 * write_buffer_size as u64), + ); + // By default RocksDB keeps every file open (`-1`); only override if asked. + if let Some(max_open_files) = tuning.max_open_files { + options.set_max_open_files(max_open_files); + } let mut block_options = BlockBasedOptions::default(); block_options.set_pin_l0_filter_and_index_blocks_in_cache(true); @@ -396,23 +654,25 @@ impl RocksDbStoreInternal { // - Small enough to leave memory for other system components // - Follows common practice for database caching in server environments // - Prevents excessive memory pressure that could lead to swapping or OOM conditions + let block_cache_size = tuning.block_cache_size.unwrap_or(total_ram / 4); block_options.set_block_cache(&Cache::new_hyper_clock_cache( - total_ram / 4, + block_cache_size, HYPER_CLOCK_CACHE_BLOCK_SIZE, )); // Cap total memtable memory to prevent unbounded growth when multiple column // families are used or many memtables accumulate before flushing. + let write_buffer_manager_size = tuning.write_buffer_manager_size.unwrap_or(total_ram / 4); let write_buffer_manager = - WriteBufferManager::new_write_buffer_manager(total_ram / 4, true); + WriteBufferManager::new_write_buffer_manager(write_buffer_manager_size, true); options.set_write_buffer_manager(&write_buffer_manager); // Configure bloom filters for prefix iteration optimization - block_options.set_bloom_filter(10.0, false); + block_options.set_bloom_filter(tuning.bloom_filter_bits_per_key.unwrap_or(10.0), false); block_options.set_whole_key_filtering(false); // 32KB blocks instead of default 4KB - reduces iterator seeks - block_options.set_block_size(32 * 1024); + block_options.set_block_size(tuning.block_size.unwrap_or(32 * 1024)); // Use latest format for better compression and performance block_options.set_format_version(5); @@ -688,6 +948,7 @@ impl TestKeyValueDatabase for RocksDbDatabaseInternal { path_with_guard, spawn_mode, max_stream_queries, + tuning_options: RocksDbTuningOptions::default(), }) } } @@ -723,6 +984,10 @@ pub enum RocksDbStoreInternalError { #[error("Namespace contains forbidden characters")] InvalidNamespace, + /// A RocksDB tuning option could not be parsed. + #[error("invalid RocksDB tuning option: {0}")] + InvalidTuningOption(String), + /// Filesystem error #[error("Filesystem error: {0}")] FsError(#[from] std::io::Error), @@ -806,3 +1071,107 @@ impl crate::backends::DatabaseBackup for RocksDbDatabaseInternal { Ok(()) } } + +#[cfg(test)] +mod tests { + use tempfile::TempDir; + + use super::{ + PathWithGuard, RocksDbCompressionType, RocksDbSpawnMode, RocksDbStoreInternal, + RocksDbStoreInternalConfig, RocksDbTuningOptions, ROOT_KEY_DOMAIN, + }; + + #[test] + fn parses_typed_values() { + let options = RocksDbTuningOptions::from_kv_pairs([ + "write_buffer_size=268435456", + "max_open_files=512", + "bloom_filter_bits_per_key=12.5", + "compression_type=zstd", + ]) + .unwrap(); + assert_eq!(options.write_buffer_size, Some(268435456)); + assert_eq!(options.max_open_files, Some(512)); + assert_eq!(options.bloom_filter_bits_per_key, Some(12.5)); + assert_eq!(options.compression_type, Some(RocksDbCompressionType::Zstd)); + // Untouched options stay at their defaults. + assert_eq!(options.max_background_jobs, None); + } + + #[test] + fn empty_input_is_all_defaults() { + let options = RocksDbTuningOptions::from_kv_pairs(Vec::::new()).unwrap(); + assert!(options.write_buffer_size.is_none()); + assert!(options.compression_type.is_none()); + } + + #[test] + fn trims_whitespace_around_key_and_value() { + let options = RocksDbTuningOptions::from_kv_pairs([" write_buffer_size = 1024 "]).unwrap(); + assert_eq!(options.write_buffer_size, Some(1024)); + } + + #[test] + fn rejects_unknown_key() { + let error = RocksDbTuningOptions::from_kv_pairs(["not_a_real_option=1"]).unwrap_err(); + assert!(error.to_string().contains("unknown option")); + } + + #[test] + fn rejects_format_sensitive_key() { + for key in [ + "prefix_extractor=8", + "format_version=6", + "compaction_style=universal", + ] { + let error = RocksDbTuningOptions::from_kv_pairs([key]).unwrap_err(); + assert!( + error.to_string().contains("storage format"), + "unexpected error for {key}: {error}" + ); + } + } + + #[test] + fn rejects_malformed_pair() { + let error = RocksDbTuningOptions::from_kv_pairs(["write_buffer_size"]).unwrap_err(); + assert!(error.to_string().contains("KEY=VALUE")); + } + + #[test] + fn rejects_unparseable_value() { + let error = + RocksDbTuningOptions::from_kv_pairs(["write_buffer_size=not_a_number"]).unwrap_err(); + assert!(error.to_string().contains("invalid value")); + } + + #[test] + fn rejects_unknown_compression() { + let error = RocksDbTuningOptions::from_kv_pairs(["compression_type=gzip"]).unwrap_err(); + assert!(error.to_string().contains("unknown compression type")); + } + + /// Opens a real RocksDB store with overridden options to confirm they are + /// accepted by RocksDB (in particular the conditionally-set `max_open_files` + /// and the cache-size paths). + #[test] + fn build_applies_overrides() { + let dir = TempDir::new().unwrap(); + let config = RocksDbStoreInternalConfig { + path_with_guard: PathWithGuard::new(dir.path().to_path_buf()), + spawn_mode: RocksDbSpawnMode::SpawnBlocking, + max_stream_queries: 10, + tuning_options: RocksDbTuningOptions::from_kv_pairs([ + "write_buffer_size=1048576", + "max_open_files=128", + "compression_type=none", + "block_cache_size=8388608", + "write_buffer_manager_size=8388608", + "max_background_jobs=2", + ]) + .unwrap(), + }; + let store = RocksDbStoreInternal::build(&config, "test_ns", ROOT_KEY_DOMAIN.to_vec()); + assert!(store.is_ok(), "build failed: {:?}", store.err()); + } +} diff --git a/linera-views/src/backends/scylla_db.rs b/linera-views/src/backends/scylla_db.rs index 7fbd21a2bb35..d88b5ce739fe 100644 --- a/linera-views/src/backends/scylla_db.rs +++ b/linera-views/src/backends/scylla_db.rs @@ -9,12 +9,13 @@ use std::{ collections::{BTreeSet, HashMap}, + num::NonZeroUsize, ops::Deref, sync::{ atomic::{AtomicI64, Ordering}, Arc, }, - time::{SystemTime, UNIX_EPOCH}, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use async_lock::{Semaphore, SemaphoreGuard}; @@ -25,6 +26,7 @@ use scylla::{ execution_profile::{ExecutionProfile, ExecutionProfileHandle}, session::Session, session_builder::SessionBuilder, + Compression, PoolSize, }, deserialize::{DeserializationError, TypeCheckError}, errors::{ @@ -251,33 +253,57 @@ impl ScyllaDbClient { }) } - fn build_default_policy() -> Arc { - DefaultPolicy::builder().token_aware(true).build() + fn build_default_policy(tuning: &ScyllaDbTuningOptions) -> Arc { + DefaultPolicy::builder() + .token_aware(tuning.token_aware.unwrap_or(true)) + .build() } fn build_default_execution_profile_handle( policy: Arc, + tuning: &ScyllaDbTuningOptions, ) -> ExecutionProfileHandle { - let default_profile = ExecutionProfile::builder() + let mut builder = ExecutionProfile::builder() .load_balancing_policy(policy) .retry_policy(Arc::new(DefaultRetryPolicy::new())) - .consistency(Consistency::LocalQuorum) - .build(); - default_profile.into_handle() + .consistency( + tuning + .consistency + .map_or(Consistency::LocalQuorum, |c| c.to_scylla()), + ); + if let Some(ms) = tuning.request_timeout_ms { + // A timeout of zero disables the per-request timeout entirely. + let timeout = (ms != 0).then(|| Duration::from_millis(ms)); + builder = builder.request_timeout(timeout); + } + builder.build().into_handle() } - async fn build_default_session(uri: &str) -> Result { + async fn build_default_session( + config: &ScyllaDbStoreInternalConfig, + ) -> Result { // This explicitly sets a lot of default parameters for clarity and for making future changes // easier. - SessionBuilder::new() - .known_node(uri) + let tuning = &config.tuning_options; + let mut builder = SessionBuilder::new() + .known_node(&config.uri) .default_execution_profile_handle(Self::build_default_execution_profile_handle( - Self::build_default_policy(), - )) - .build() - .boxed_sync() - .await - .map_err(Into::into) + Self::build_default_policy(tuning), + tuning, + )); + if let Some(compression) = tuning.compression { + builder = builder.compression(compression.to_scylla()); + } + if let Some(pool_size) = tuning.connection_pool_size_per_host { + // The parser guarantees this is non-zero, but guard defensively. + let pool_size = NonZeroUsize::new(pool_size).ok_or_else(|| { + ScyllaDbStoreInternalError::InvalidTuningOption( + "`connection_pool_size_per_host` must be at least 1".to_string(), + ) + })?; + builder = builder.pool_size(PoolSize::PerHost(pool_size)); + } + builder.build().boxed_sync().await.map_err(Into::into) } async fn get_multi_key_values_statement( @@ -798,6 +824,10 @@ pub enum ScyllaDbStoreInternalError { #[error("Namespace contains forbidden characters")] InvalidNamespace, + /// A ScyllaDB tuning option could not be parsed. + #[error("invalid ScyllaDB tuning option: {0}")] + InvalidTuningOption(String), + /// The key must have at most `MAX_KEY_SIZE` bytes #[error("The key must have at most MAX_KEY_SIZE")] KeyTooLong, @@ -1037,6 +1067,203 @@ pub struct ScyllaDbStoreInternalConfig { pub max_stream_queries: usize, /// The replication factor. pub replication_factor: u32, + /// Runtime-tunable ScyllaDB driver options that override the built-in defaults. + #[serde(default)] + pub tuning_options: ScyllaDbTuningOptions, +} + +/// Runtime-tunable ScyllaDB driver and session options. +/// +/// These options only affect how the client connects to and talks to ScyllaDB +/// (consistency, timeouts, connection pooling, transport compression, load +/// balancing). They never change the on-disk storage format or the table +/// schema, so they are safe to change between restarts. Every field defaults to +/// `None`, meaning the built-in default is used. +/// +/// Note that table-creation properties (compaction strategy, table compression, +/// `gc_grace_seconds`, …) are deliberately *not* exposed here: they are part of +/// the `CREATE TABLE` statement and only take effect at namespace creation time. +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +pub struct ScyllaDbTuningOptions { + /// The consistency level used for queries. + pub consistency: Option, + /// Per-request timeout in milliseconds. `0` disables the timeout. + pub request_timeout_ms: Option, + /// Number of connections to open per node. + pub connection_pool_size_per_host: Option, + /// Transport-level compression for the CQL protocol. + pub compression: Option, + /// Whether the load-balancing policy is token-aware. + pub token_aware: Option, +} + +/// The consistency level used by the ScyllaDB client. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize, Serialize)] +#[allow(missing_docs)] +pub enum ScyllaDbConsistency { + Any, + One, + Two, + Three, + Quorum, + All, + LocalQuorum, + EachQuorum, + LocalOne, + Serial, + LocalSerial, +} + +impl ScyllaDbConsistency { + fn to_scylla(self) -> Consistency { + match self { + ScyllaDbConsistency::Any => Consistency::Any, + ScyllaDbConsistency::One => Consistency::One, + ScyllaDbConsistency::Two => Consistency::Two, + ScyllaDbConsistency::Three => Consistency::Three, + ScyllaDbConsistency::Quorum => Consistency::Quorum, + ScyllaDbConsistency::All => Consistency::All, + ScyllaDbConsistency::LocalQuorum => Consistency::LocalQuorum, + ScyllaDbConsistency::EachQuorum => Consistency::EachQuorum, + ScyllaDbConsistency::LocalOne => Consistency::LocalOne, + ScyllaDbConsistency::Serial => Consistency::Serial, + ScyllaDbConsistency::LocalSerial => Consistency::LocalSerial, + } + } + + fn from_token(token: &str) -> Option { + Some(match token.to_ascii_lowercase().as_str() { + "any" => ScyllaDbConsistency::Any, + "one" => ScyllaDbConsistency::One, + "two" => ScyllaDbConsistency::Two, + "three" => ScyllaDbConsistency::Three, + "quorum" => ScyllaDbConsistency::Quorum, + "all" => ScyllaDbConsistency::All, + "local_quorum" => ScyllaDbConsistency::LocalQuorum, + "each_quorum" => ScyllaDbConsistency::EachQuorum, + "local_one" => ScyllaDbConsistency::LocalOne, + "serial" => ScyllaDbConsistency::Serial, + "local_serial" => ScyllaDbConsistency::LocalSerial, + _ => return None, + }) + } +} + +/// The transport-level compression used by the ScyllaDB client. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize, Serialize)] +pub enum ScyllaDbCompression { + /// No compression. + None, + /// LZ4. + Lz4, + /// Snappy. + Snappy, +} + +impl ScyllaDbCompression { + fn to_scylla(self) -> Option { + match self { + ScyllaDbCompression::None => None, + ScyllaDbCompression::Lz4 => Some(Compression::Lz4), + ScyllaDbCompression::Snappy => Some(Compression::Snappy), + } + } + + fn from_token(token: &str) -> Option { + Some(match token.to_ascii_lowercase().as_str() { + "none" => ScyllaDbCompression::None, + "lz4" => ScyllaDbCompression::Lz4, + "snappy" => ScyllaDbCompression::Snappy, + _ => return None, + }) + } +} + +impl ScyllaDbTuningOptions { + /// The list of supported option keys, for use in error messages and docs. + pub fn supported_keys() -> &'static [&'static str] { + &[ + "consistency", + "request_timeout_ms", + "connection_pool_size_per_host", + "compression", + "token_aware", + ] + } + + /// Parses a list of `KEY=VALUE` strings into typed tuning options. + /// + /// Unknown keys, malformed entries, and values that fail to parse all + /// produce a descriptive error. + pub fn from_kv_pairs(pairs: I) -> Result + where + I: IntoIterator, + S: AsRef, + { + let mut options = Self::default(); + for pair in pairs { + let pair = pair.as_ref(); + let (key, value) = pair.split_once('=').ok_or_else(|| { + ScyllaDbStoreInternalError::InvalidTuningOption(format!( + "expected `KEY=VALUE`, got `{pair}`" + )) + })?; + let key = key.trim(); + let value = value.trim(); + match key { + "consistency" => { + options.consistency = + Some(ScyllaDbConsistency::from_token(value).ok_or_else(|| { + ScyllaDbStoreInternalError::InvalidTuningOption(format!( + "unknown consistency `{value}`; expected one of any, one, two, \ + three, quorum, all, local_quorum, each_quorum, local_one, \ + serial, local_serial" + )) + })?) + } + "request_timeout_ms" => { + options.request_timeout_ms = Some(parse_tuning_value(key, value)?) + } + "connection_pool_size_per_host" => { + let n = parse_tuning_value(key, value)?; + if n == 0 { + return Err(ScyllaDbStoreInternalError::InvalidTuningOption( + "`connection_pool_size_per_host` must be at least 1".to_string(), + )); + } + options.connection_pool_size_per_host = Some(n) + } + "compression" => { + options.compression = + Some(ScyllaDbCompression::from_token(value).ok_or_else(|| { + ScyllaDbStoreInternalError::InvalidTuningOption(format!( + "unknown compression `{value}`; expected one of none, lz4, snappy" + )) + })?) + } + "token_aware" => options.token_aware = Some(parse_tuning_value(key, value)?), + _ => { + return Err(ScyllaDbStoreInternalError::InvalidTuningOption(format!( + "unknown option `{key}`; supported options are: {}", + Self::supported_keys().join(", ") + ))) + } + } + } + Ok(options) + } +} + +fn parse_tuning_value(key: &str, value: &str) -> Result +where + T: std::str::FromStr, + T::Err: std::fmt::Display, +{ + value.parse::().map_err(|err| { + ScyllaDbStoreInternalError::InvalidTuningOption(format!( + "invalid value `{value}` for option `{key}`: {err}" + )) + }) } impl KeyValueDatabase for ScyllaDbDatabaseInternal { @@ -1052,7 +1279,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal { namespace: &str, ) -> Result { Self::check_namespace(namespace)?; - let session = ScyllaDbClient::build_default_session(&config.uri).await?; + let session = ScyllaDbClient::build_default_session(config).await?; let store = ScyllaDbClient::new(session, namespace).await?; let store = Arc::new(store); let semaphore = config @@ -1097,7 +1324,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal { } async fn list_all(config: &Self::Config) -> Result, ScyllaDbStoreInternalError> { - let session = ScyllaDbClient::build_default_session(&config.uri).await?; + let session = ScyllaDbClient::build_default_session(config).await?; let statement = session .prepare(format!("DESCRIBE KEYSPACE {KEYSPACE}")) .await?; @@ -1153,7 +1380,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal { } async fn delete_all(store_config: &Self::Config) -> Result<(), ScyllaDbStoreInternalError> { - let session = ScyllaDbClient::build_default_session(&store_config.uri).await?; + let session = ScyllaDbClient::build_default_session(store_config).await?; let statement = session .prepare(format!("DROP KEYSPACE IF EXISTS {KEYSPACE}")) .await?; @@ -1170,7 +1397,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal { namespace: &str, ) -> Result { Self::check_namespace(namespace)?; - let session = ScyllaDbClient::build_default_session(&config.uri).await?; + let session = ScyllaDbClient::build_default_session(config).await?; // We check the way the test can fail. It can fail in different ways. let result = session @@ -1214,7 +1441,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal { namespace: &str, ) -> Result<(), ScyllaDbStoreInternalError> { Self::check_namespace(namespace)?; - let session = ScyllaDbClient::build_default_session(&config.uri).await?; + let session = ScyllaDbClient::build_default_session(config).await?; // Create a keyspace if it doesn't exist let statement = session @@ -1268,7 +1495,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal { namespace: &str, ) -> Result<(), ScyllaDbStoreInternalError> { Self::check_namespace(namespace)?; - let session = ScyllaDbClient::build_default_session(&config.uri).await?; + let session = ScyllaDbClient::build_default_session(config).await?; let statement = session .prepare(format!("DROP TABLE IF EXISTS {KEYSPACE}.\"{namespace}\";")) .await?; @@ -1315,6 +1542,7 @@ impl TestKeyValueDatabase for JournalingKeyValueDatabase; /// The combined error type for the `ScyllaDbDatabase`. pub type ScyllaDbStoreError = ValueSplittingError>; + +#[cfg(test)] +mod tests { + use super::{ScyllaDbCompression, ScyllaDbConsistency, ScyllaDbTuningOptions}; + + #[test] + fn parses_typed_values() { + let options = ScyllaDbTuningOptions::from_kv_pairs([ + "consistency=local_quorum", + "request_timeout_ms=5000", + "connection_pool_size_per_host=4", + "compression=lz4", + "token_aware=false", + ]) + .unwrap(); + assert_eq!(options.consistency, Some(ScyllaDbConsistency::LocalQuorum)); + assert_eq!(options.request_timeout_ms, Some(5000)); + assert_eq!(options.connection_pool_size_per_host, Some(4)); + assert_eq!(options.compression, Some(ScyllaDbCompression::Lz4)); + assert_eq!(options.token_aware, Some(false)); + } + + #[test] + fn empty_input_is_all_defaults() { + let options = ScyllaDbTuningOptions::from_kv_pairs(Vec::::new()).unwrap(); + assert!(options.consistency.is_none()); + assert!(options.compression.is_none()); + assert!(options.token_aware.is_none()); + } + + #[test] + fn rejects_unknown_key() { + let error = ScyllaDbTuningOptions::from_kv_pairs(["not_a_real_option=1"]).unwrap_err(); + assert!(error.to_string().contains("unknown option")); + } + + #[test] + fn rejects_unknown_consistency() { + let error = ScyllaDbTuningOptions::from_kv_pairs(["consistency=banana"]).unwrap_err(); + assert!(error.to_string().contains("unknown consistency")); + } + + #[test] + fn rejects_unknown_compression() { + let error = ScyllaDbTuningOptions::from_kv_pairs(["compression=gzip"]).unwrap_err(); + assert!(error.to_string().contains("unknown compression")); + } + + #[test] + fn rejects_zero_pool_size() { + let error = + ScyllaDbTuningOptions::from_kv_pairs(["connection_pool_size_per_host=0"]).unwrap_err(); + assert!(error.to_string().contains("at least 1")); + } + + #[test] + fn rejects_malformed_pair() { + let error = ScyllaDbTuningOptions::from_kv_pairs(["consistency"]).unwrap_err(); + assert!(error.to_string().contains("KEY=VALUE")); + } + + #[test] + fn rejects_unparseable_value() { + let error = ScyllaDbTuningOptions::from_kv_pairs(["request_timeout_ms=soon"]).unwrap_err(); + assert!(error.to_string().contains("invalid value")); + } +}