linera_storage_runtime/
lib.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Storage configuration and runtime infrastructure for the Linera protocol.
5
6#![deny(missing_docs)]
7
8use std::{fmt, path::PathBuf, str::FromStr};
9
10use anyhow::{anyhow, bail};
11use async_trait::async_trait;
12use linera_core::GenesisConfig;
13use linera_execution::WasmRuntime;
14pub use linera_storage::StorageCacheConfig;
15/// Backward-compatible alias.
16pub type StorageCacheSizes = StorageCacheConfig;
17use linera_storage::{DbStorage, Storage, WallClock, DEFAULT_NAMESPACE};
18#[cfg(feature = "storage-service")]
19use linera_storage_service::{
20    client::StorageServiceDatabase,
21    common::{StorageServiceStoreConfig, StorageServiceStoreInternalConfig},
22};
23#[cfg(feature = "rocksdb")]
24use linera_views::rocks_db::{
25    PathWithGuard, RocksDbDatabase, RocksDbSpawnMode, RocksDbStatisticsLevel, RocksDbStoreConfig,
26    RocksDbStoreInternalConfig,
27};
28use linera_views::{
29    lru_prefix_cache::StorageCacheConfig as ViewsStorageCacheConfig,
30    memory::{MemoryDatabase, MemoryStoreConfig},
31    store::{KeyValueDatabase, KeyValueStore},
32};
33use serde::{Deserialize, Serialize};
34use tracing::error;
35#[cfg(all(feature = "rocksdb", feature = "scylladb"))]
36use {
37    linera_storage::ChainStatesFirstAssignment,
38    linera_views::backends::dual::{DualDatabase, DualStoreConfig},
39    std::path::Path,
40};
41#[cfg(feature = "scylladb")]
42use {
43    linera_views::scylla_db::{ScyllaDbDatabase, ScyllaDbStoreConfig, ScyllaDbStoreInternalConfig},
44    std::num::NonZeroU16,
45    tracing::debug,
46};
47
48/// Command-line options shared by all storage backends.
49#[derive(Clone, Debug, clap::Parser)]
50pub struct CommonStorageOptions {
51    /// The maximal number of simultaneous queries to the database
52    #[arg(long, global = true)]
53    pub storage_max_concurrent_queries: Option<usize>,
54
55    /// The maximal number of simultaneous stream queries to the database
56    #[arg(long, default_value = "10", global = true)]
57    pub storage_max_stream_queries: usize,
58
59    /// The maximal memory used in the storage cache.
60    #[arg(long, default_value = "10000000", global = true)]
61    pub storage_max_cache_size: usize,
62
63    /// The maximal size of a value entry in the storage cache.
64    #[arg(long, default_value = "1000000", global = true)]
65    pub storage_max_value_entry_size: usize,
66
67    /// The maximal size of a find-keys entry in the storage cache.
68    #[arg(long, default_value = "1000000", global = true)]
69    pub storage_max_find_keys_entry_size: usize,
70
71    /// The maximal size of a find-key-values entry in the storage cache.
72    #[arg(long, default_value = "1000000", global = true)]
73    pub storage_max_find_key_values_entry_size: usize,
74
75    /// The maximal number of entries in the storage cache.
76    #[arg(long, default_value = "1000", global = true)]
77    pub storage_max_cache_entries: usize,
78
79    /// The maximal memory used in the value cache.
80    #[arg(long, default_value = "10000000", global = true)]
81    pub storage_max_cache_value_size: usize,
82
83    /// The maximal memory used in the find_keys_by_prefix cache.
84    #[arg(long, default_value = "10000000", global = true)]
85    pub storage_max_cache_find_keys_size: usize,
86
87    /// The maximal memory used in the find_key_values_by_prefix cache.
88    #[arg(long, default_value = "10000000", global = true)]
89    pub storage_max_cache_find_key_values_size: usize,
90
91    /// The maximal number of entries in the blob cache.
92    #[arg(long, default_value = "1000", global = true)]
93    pub blob_cache_size: usize,
94
95    /// The maximal number of entries in the confirmed block cache.
96    #[arg(long, default_value = "1000", global = true)]
97    pub confirmed_block_cache_size: usize,
98
99    /// The maximal number of entries in the confirmed block certificate cache.
100    #[arg(long, default_value = "1000", global = true)]
101    pub certificate_cache_size: usize,
102
103    /// The maximal number of entries in the raw certificate cache.
104    #[arg(long, default_value = "1000", global = true)]
105    pub certificate_raw_cache_size: usize,
106
107    /// The maximal number of entries in the event cache.
108    #[arg(long, default_value = "1000", global = true)]
109    pub event_cache_size: usize,
110
111    /// The maximal number of entries in the block-hash-by-height cache.
112    #[arg(long, default_value = "1000", global = true)]
113    pub block_hash_by_height_cache_size: usize,
114
115    /// The number of entries in the block cache.
116    #[arg(long, default_value = "5000", global = true)]
117    pub block_cache_size: usize,
118
119    /// The number of entries in the execution state cache.
120    #[arg(long, default_value = "10000", global = true)]
121    pub execution_state_cache_size: usize,
122
123    /// The replication factor for the keyspace
124    #[arg(long, default_value = "1", global = true)]
125    pub storage_replication_factor: u32,
126
127    /// Enable RocksDB's internal statistics collection and export them as Prometheus
128    /// metrics. Off by default; enable it on nodes whose metrics are scraped.
129    #[cfg(feature = "rocksdb")]
130    #[arg(long, global = true)]
131    pub rocksdb_enable_statistics: bool,
132
133    /// The level of detail collected when `--rocksdb-enable-statistics` is set. Higher
134    /// levels collect more, and more expensive, data. One of: `disable-all`,
135    /// `except-histogram-or-timers`, `except-timers`, `except-detailed-timers`,
136    /// `except-time-for-mutex`, `all`.
137    #[cfg(feature = "rocksdb")]
138    #[arg(
139        long,
140        default_value = "except-histogram-or-timers",
141        value_parser = RocksDbStatisticsLevel::from_str,
142        global = true
143    )]
144    pub rocksdb_statistics_level: RocksDbStatisticsLevel,
145}
146
147impl CommonStorageOptions {
148    /// Builds the storage-level cache configuration from these options.
149    pub fn storage_cache_sizes(&self) -> StorageCacheSizes {
150        StorageCacheSizes {
151            blob_cache_size: self.blob_cache_size,
152            confirmed_block_cache_size: self.confirmed_block_cache_size,
153            certificate_cache_size: self.certificate_cache_size,
154            certificate_raw_cache_size: self.certificate_raw_cache_size,
155            event_cache_size: self.event_cache_size,
156            block_hash_by_height_cache_size: self.block_hash_by_height_cache_size,
157            cache_cleanup_interval_secs: linera_storage::DEFAULT_CLEANUP_INTERVAL_SECS,
158        }
159    }
160
161    /// Builds the views-level cache configuration from these options.
162    pub fn storage_cache_config(&self) -> ViewsStorageCacheConfig {
163        ViewsStorageCacheConfig {
164            max_cache_size: self.storage_max_cache_size,
165            max_value_entry_size: self.storage_max_value_entry_size,
166            max_find_keys_entry_size: self.storage_max_find_keys_entry_size,
167            max_find_key_values_entry_size: self.storage_max_find_key_values_entry_size,
168            max_cache_entries: self.storage_max_cache_entries,
169            max_cache_value_size: self.storage_max_cache_value_size,
170            max_cache_find_keys_size: self.storage_max_cache_find_keys_size,
171            max_cache_find_key_values_size: self.storage_max_cache_find_key_values_size,
172        }
173    }
174
175    /// Returns options matching the clap-defined defaults.
176    pub fn with_defaults() -> Self {
177        use clap::Parser as _;
178        Self::parse_from(std::iter::empty::<String>())
179    }
180}
181
182/// The configuration of the key value store in use.
183#[derive(Clone, Debug, Deserialize, Serialize)]
184pub enum StoreConfig {
185    /// The memory key value store
186    Memory {
187        /// The store configuration.
188        config: MemoryStoreConfig,
189        /// The namespace used.
190        namespace: String,
191        /// The path to the genesis configuration.
192        genesis_path: PathBuf,
193    },
194    /// The storage service key-value store
195    #[cfg(feature = "storage-service")]
196    StorageService {
197        /// The store configuration.
198        config: StorageServiceStoreConfig,
199        /// The namespace used.
200        namespace: String,
201    },
202    /// The RocksDB key value store
203    #[cfg(feature = "rocksdb")]
204    RocksDb {
205        /// The store configuration.
206        config: RocksDbStoreConfig,
207        /// The namespace used.
208        namespace: String,
209    },
210    /// The ScyllaDB key value store
211    #[cfg(feature = "scylladb")]
212    ScyllaDb {
213        /// The store configuration.
214        config: ScyllaDbStoreConfig,
215        /// The namespace used.
216        namespace: String,
217    },
218    /// The dual RocksDB and ScyllaDB key value store
219    #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
220    DualRocksDbScyllaDb {
221        /// The store configuration.
222        config: DualStoreConfig<RocksDbStoreConfig, ScyllaDbStoreConfig>,
223        /// The namespace used.
224        namespace: String,
225    },
226}
227
228/// The description of a storage implementation.
229#[derive(Clone, Debug)]
230#[cfg_attr(any(test), derive(Eq, PartialEq))]
231pub enum InnerStorageConfig {
232    /// The memory description.
233    Memory {
234        /// The path to the genesis configuration. This is needed because we reinitialize
235        /// memory databases from the genesis config everytime.
236        genesis_path: PathBuf,
237    },
238    /// The storage service description.
239    #[cfg(feature = "storage-service")]
240    Service {
241        /// The endpoint used.
242        endpoint: String,
243    },
244    /// The RocksDB description.
245    #[cfg(feature = "rocksdb")]
246    RocksDb {
247        /// The path used.
248        path: PathBuf,
249        /// Whether to use `block_in_place` or `spawn_blocking`.
250        spawn_mode: RocksDbSpawnMode,
251    },
252    /// The ScyllaDB description.
253    #[cfg(feature = "scylladb")]
254    ScyllaDb {
255        /// The URI for accessing the database.
256        uri: String,
257    },
258    /// The dual RocksDB and ScyllaDB description.
259    #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
260    DualRocksDbScyllaDb {
261        /// The path used.
262        path_with_guard: PathWithGuard,
263        /// Whether to use `block_in_place` or `spawn_blocking`.
264        spawn_mode: RocksDbSpawnMode,
265        /// The URI for accessing the database.
266        uri: String,
267    },
268}
269
270/// The description of a storage implementation.
271#[derive(Clone, Debug)]
272#[cfg_attr(any(test), derive(Eq, PartialEq))]
273pub struct StorageConfig {
274    /// The inner storage config.
275    pub inner_storage_config: InnerStorageConfig,
276    /// The namespace used
277    pub namespace: String,
278}
279
280const MEMORY: &str = "memory:";
281#[cfg(feature = "storage-service")]
282const STORAGE_SERVICE: &str = "service:";
283#[cfg(feature = "rocksdb")]
284const ROCKS_DB: &str = "rocksdb:";
285#[cfg(feature = "scylladb")]
286const SCYLLA_DB: &str = "scylladb:";
287#[cfg(all(feature = "rocksdb", feature = "scylladb"))]
288const DUAL_ROCKS_DB_SCYLLA_DB: &str = "dualrocksdbscylladb:";
289
290impl FromStr for StorageConfig {
291    type Err = anyhow::Error;
292
293    fn from_str(input: &str) -> Result<Self, Self::Err> {
294        if let Some(s) = input.strip_prefix(MEMORY) {
295            let parts = s.split(':').collect::<Vec<_>>();
296            if parts.len() == 1 {
297                let genesis_path = parts[0].to_string().into();
298                let namespace = DEFAULT_NAMESPACE.to_string();
299                let inner_storage_config = InnerStorageConfig::Memory { genesis_path };
300                return Ok(StorageConfig {
301                    inner_storage_config,
302                    namespace,
303                });
304            }
305            if parts.len() != 2 {
306                bail!("We should have one genesis config path and one optional namespace");
307            }
308            let genesis_path = parts[0].to_string().into();
309            let namespace = parts[1].to_string();
310            let inner_storage_config = InnerStorageConfig::Memory { genesis_path };
311            return Ok(StorageConfig {
312                inner_storage_config,
313                namespace,
314            });
315        }
316        #[cfg(feature = "storage-service")]
317        if let Some(s) = input.strip_prefix(STORAGE_SERVICE) {
318            if s.is_empty() {
319                bail!(
320                    "For Storage service, the formatting has to be service:endpoint:namespace,\
321example service:tcp:127.0.0.1:7878:table_do_my_test"
322                );
323            }
324            let parts = s.split(':').collect::<Vec<_>>();
325            if parts.len() != 4 {
326                bail!("We should have one endpoint and one namespace");
327            }
328            let protocol = parts[0];
329            if protocol != "tcp" {
330                bail!("Only allowed protocol is tcp");
331            }
332            let endpoint = parts[1];
333            let port = parts[2];
334            let mut endpoint = endpoint.to_string();
335            endpoint.push(':');
336            endpoint.push_str(port);
337            let endpoint = endpoint.to_string();
338            let namespace = parts[3].to_string();
339            let inner_storage_config = InnerStorageConfig::Service { endpoint };
340            return Ok(StorageConfig {
341                inner_storage_config,
342                namespace,
343            });
344        }
345        #[cfg(feature = "rocksdb")]
346        if let Some(s) = input.strip_prefix(ROCKS_DB) {
347            if s.is_empty() {
348                bail!(
349                    "For RocksDB, the formatting has to be rocksdb:directory or rocksdb:directory:spawn_mode:namespace");
350            }
351            let parts = s.split(':').collect::<Vec<_>>();
352            if parts.len() == 1 {
353                let path = parts[0].to_string().into();
354                let namespace = DEFAULT_NAMESPACE.to_string();
355                let spawn_mode = RocksDbSpawnMode::SpawnBlocking;
356                let inner_storage_config = InnerStorageConfig::RocksDb { path, spawn_mode };
357                return Ok(StorageConfig {
358                    inner_storage_config,
359                    namespace,
360                });
361            }
362            if parts.len() == 2 || parts.len() == 3 {
363                let path = parts[0].to_string().into();
364                let spawn_mode_str = parts.get(1).expect("length already checked");
365                let spawn_mode = match *spawn_mode_str {
366                    "spawn_blocking" => Ok(RocksDbSpawnMode::SpawnBlocking),
367                    "block_in_place" => Ok(RocksDbSpawnMode::BlockInPlace),
368                    "runtime" => Ok(RocksDbSpawnMode::get_spawn_mode_from_runtime()),
369                    _ => Err(anyhow!("Failed to parse {spawn_mode_str} as a spawn_mode")),
370                }?;
371                let namespace = if parts.len() == 2 {
372                    DEFAULT_NAMESPACE.to_string()
373                } else {
374                    (*parts.get(2).expect("length already checked")).to_string()
375                };
376                let inner_storage_config = InnerStorageConfig::RocksDb { path, spawn_mode };
377                return Ok(StorageConfig {
378                    inner_storage_config,
379                    namespace,
380                });
381            }
382            bail!("We should have one, two or three parts");
383        }
384        #[cfg(feature = "scylladb")]
385        if let Some(s) = input.strip_prefix(SCYLLA_DB) {
386            let mut uri: Option<String> = None;
387            let mut namespace: Option<String> = None;
388            let parse_error: &'static str = "Correct format is tcp:db_hostname:port.";
389            if !s.is_empty() {
390                let mut parts = s.split(':');
391                while let Some(part) = parts.next() {
392                    match part {
393                        "tcp" => {
394                            let address = parts.next().ok_or_else(|| {
395                                anyhow!("Failed to find address for {s}. {parse_error}")
396                            })?;
397                            let port_str = parts.next().ok_or_else(|| {
398                                anyhow!("Failed to find port for {s}. {parse_error}")
399                            })?;
400                            let port = NonZeroU16::from_str(port_str).map_err(|_| {
401                                anyhow!(
402                                    "Failed to find parse port {port_str} for {s}. {parse_error}",
403                                )
404                            })?;
405                            if uri.is_some() {
406                                bail!("The uri has already been assigned");
407                            }
408                            uri = Some(format!("{address}:{port}"));
409                        }
410                        _ if part.starts_with("table") => {
411                            if namespace.is_some() {
412                                bail!("The namespace has already been assigned");
413                            }
414                            namespace = Some(part.to_string());
415                        }
416                        _ => {
417                            bail!("the entry \"{part}\" is not matching");
418                        }
419                    }
420                }
421            }
422            let uri = uri.unwrap_or_else(|| "localhost:9042".to_string());
423            let namespace = namespace.unwrap_or_else(|| DEFAULT_NAMESPACE.to_string());
424            let inner_storage_config = InnerStorageConfig::ScyllaDb { uri };
425            debug!("ScyllaDB connection info: {:?}", inner_storage_config);
426            return Ok(StorageConfig {
427                inner_storage_config,
428                namespace,
429            });
430        }
431        #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
432        if let Some(s) = input.strip_prefix(DUAL_ROCKS_DB_SCYLLA_DB) {
433            let parts = s.split(':').collect::<Vec<_>>();
434            if parts.len() != 5 && parts.len() != 6 {
435                bail!(
436                    "For DualRocksDbScyllaDb, the formatting has to be dualrocksdbscylladb:directory:mode:tcp:hostname:port:namespace"
437                );
438            }
439            let path = Path::new(parts[0]);
440            let path = path.to_path_buf();
441            let path_with_guard = PathWithGuard::new(path);
442            let spawn_mode_str = parts.get(1).expect("length already checked");
443            let spawn_mode = match *spawn_mode_str {
444                "spawn_blocking" => Ok(RocksDbSpawnMode::SpawnBlocking),
445                "block_in_place" => Ok(RocksDbSpawnMode::BlockInPlace),
446                "runtime" => Ok(RocksDbSpawnMode::get_spawn_mode_from_runtime()),
447                _ => Err(anyhow!("Failed to parse {spawn_mode_str} as a spawn_mode",)),
448            }?;
449            let protocol = parts[2];
450            if protocol != "tcp" {
451                bail!("The only allowed protocol is tcp");
452            }
453            let address = parts[3];
454            let port_str = parts[4];
455            let port = NonZeroU16::from_str(port_str)
456                .map_err(|_| anyhow!("Failed to find parse port {port_str} for {s}"))?;
457            let uri = format!("{address}:{port}");
458            let inner_storage_config = InnerStorageConfig::DualRocksDbScyllaDb {
459                path_with_guard,
460                spawn_mode,
461                uri,
462            };
463            let namespace = if parts.len() == 5 {
464                DEFAULT_NAMESPACE.to_string()
465            } else {
466                parts[5].to_string()
467            };
468            return Ok(StorageConfig {
469                inner_storage_config,
470                namespace,
471            });
472        }
473        error!("available storage: memory");
474        #[cfg(feature = "storage-service")]
475        error!("Also available is linera-storage-service");
476        #[cfg(feature = "rocksdb")]
477        error!("Also available is RocksDB");
478        #[cfg(feature = "scylladb")]
479        error!("Also available is ScyllaDB");
480        #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
481        error!("Also available is DualRocksDbScyllaDb");
482        Err(anyhow!("The input has not matched: {input}"))
483    }
484}
485
486impl StorageConfig {
487    /// Appends a per-shard subdirectory to the path, for backends that store data on disk.
488    #[allow(unused_variables)]
489    pub fn maybe_append_shard_path(&mut self, shard: usize) -> std::io::Result<()> {
490        match &mut self.inner_storage_config {
491            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
492            InnerStorageConfig::DualRocksDbScyllaDb {
493                path_with_guard,
494                spawn_mode: _,
495                uri: _,
496            } => {
497                let shard_str = format!("shard_{shard}");
498                path_with_guard.path_buf.push(shard_str);
499                std::fs::create_dir_all(&path_with_guard.path_buf)
500            }
501            _ => Ok(()),
502        }
503    }
504
505    /// The addition of the common config to get a full configuration
506    pub fn add_common_storage_options(
507        &self,
508        options: &CommonStorageOptions,
509    ) -> Result<StoreConfig, anyhow::Error> {
510        let namespace = self.namespace.clone();
511        match &self.inner_storage_config {
512            InnerStorageConfig::Memory { genesis_path } => {
513                let config = MemoryStoreConfig {
514                    max_stream_queries: options.storage_max_stream_queries,
515                    kill_on_drop: false,
516                };
517                let genesis_path = genesis_path.clone();
518                Ok(StoreConfig::Memory {
519                    config,
520                    namespace,
521                    genesis_path,
522                })
523            }
524            #[cfg(feature = "storage-service")]
525            InnerStorageConfig::Service { endpoint } => {
526                let inner_config = StorageServiceStoreInternalConfig {
527                    endpoint: endpoint.clone(),
528                    max_concurrent_queries: options.storage_max_concurrent_queries,
529                    max_stream_queries: options.storage_max_stream_queries,
530                };
531                let config = StorageServiceStoreConfig {
532                    inner_config,
533                    storage_cache_config: options.storage_cache_config(),
534                };
535                Ok(StoreConfig::StorageService { config, namespace })
536            }
537            #[cfg(feature = "rocksdb")]
538            InnerStorageConfig::RocksDb { path, spawn_mode } => {
539                let path_with_guard = PathWithGuard::new(path.to_path_buf());
540                let inner_config = RocksDbStoreInternalConfig {
541                    spawn_mode: *spawn_mode,
542                    path_with_guard,
543                    max_stream_queries: options.storage_max_stream_queries,
544                    enable_statistics: options.rocksdb_enable_statistics,
545                    statistics_level: options.rocksdb_statistics_level,
546                };
547                let config = RocksDbStoreConfig {
548                    inner_config,
549                    storage_cache_config: options.storage_cache_config(),
550                };
551                Ok(StoreConfig::RocksDb { config, namespace })
552            }
553            #[cfg(feature = "scylladb")]
554            InnerStorageConfig::ScyllaDb { uri } => {
555                let inner_config = ScyllaDbStoreInternalConfig {
556                    uri: uri.clone(),
557                    max_stream_queries: options.storage_max_stream_queries,
558                    max_concurrent_queries: options.storage_max_concurrent_queries,
559                    replication_factor: options.storage_replication_factor,
560                };
561                let config = ScyllaDbStoreConfig {
562                    inner_config,
563                    storage_cache_config: options.storage_cache_config(),
564                };
565                Ok(StoreConfig::ScyllaDb { config, namespace })
566            }
567            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
568            InnerStorageConfig::DualRocksDbScyllaDb {
569                path_with_guard,
570                spawn_mode,
571                uri,
572            } => {
573                let inner_config = RocksDbStoreInternalConfig {
574                    spawn_mode: *spawn_mode,
575                    path_with_guard: path_with_guard.clone(),
576                    max_stream_queries: options.storage_max_stream_queries,
577                    enable_statistics: options.rocksdb_enable_statistics,
578                    statistics_level: options.rocksdb_statistics_level,
579                };
580                let first_config = RocksDbStoreConfig {
581                    inner_config,
582                    storage_cache_config: options.storage_cache_config(),
583                };
584
585                let inner_config = ScyllaDbStoreInternalConfig {
586                    uri: uri.clone(),
587                    max_stream_queries: options.storage_max_stream_queries,
588                    max_concurrent_queries: options.storage_max_concurrent_queries,
589                    replication_factor: options.storage_replication_factor,
590                };
591                let second_config = ScyllaDbStoreConfig {
592                    inner_config,
593                    storage_cache_config: options.storage_cache_config(),
594                };
595
596                let config = DualStoreConfig {
597                    first_config,
598                    second_config,
599                };
600                Ok(StoreConfig::DualRocksDbScyllaDb { config, namespace })
601            }
602        }
603    }
604}
605
606impl fmt::Display for StorageConfig {
607    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
608        let namespace = &self.namespace;
609        match &self.inner_storage_config {
610            #[cfg(feature = "storage-service")]
611            InnerStorageConfig::Service { endpoint } => {
612                write!(f, "service:tcp:{endpoint}:{namespace}")
613            }
614            InnerStorageConfig::Memory { genesis_path } => {
615                write!(f, "memory:{}:{namespace}", genesis_path.display())
616            }
617            #[cfg(feature = "rocksdb")]
618            InnerStorageConfig::RocksDb { path, spawn_mode } => {
619                let spawn_mode = spawn_mode.to_string();
620                write!(f, "rocksdb:{}:{spawn_mode}:{namespace}", path.display())
621            }
622            #[cfg(feature = "scylladb")]
623            InnerStorageConfig::ScyllaDb { uri } => {
624                write!(f, "scylladb:tcp:{uri}:{namespace}")
625            }
626            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
627            InnerStorageConfig::DualRocksDbScyllaDb {
628                path_with_guard,
629                spawn_mode,
630                uri,
631            } => {
632                write!(
633                    f,
634                    "dualrocksdbscylladb:{}:{}:tcp:{}:{}",
635                    path_with_guard.path_buf.display(),
636                    spawn_mode,
637                    uri,
638                    namespace
639                )
640            }
641        }
642    }
643}
644
645/// A job that runs against a connected, high-level [`Storage`].
646#[async_trait]
647pub trait Runnable {
648    /// The type produced by running the job.
649    type Output;
650
651    /// Runs the job against the given storage.
652    async fn run<S>(self, storage: S) -> Self::Output
653    where
654        S: Storage + Clone + Send + Sync + 'static;
655}
656
657/// A job that runs directly against a key-value store, without a full [`Storage`].
658#[async_trait]
659pub trait RunnableWithStore {
660    /// The type produced by running the job.
661    type Output;
662
663    /// Runs the job against the store described by the given config and namespace.
664    async fn run<D>(
665        self,
666        config: D::Config,
667        namespace: String,
668        cache_sizes: StorageCacheSizes,
669    ) -> Result<Self::Output, anyhow::Error>
670    where
671        D: KeyValueDatabase + Clone + Send + Sync + 'static,
672        D::Store: KeyValueStore + Clone + Send + Sync + 'static,
673        D::Error: Send + Sync;
674}
675
676/// Reads a JSON value from a file at the given path.
677fn read_json<T: serde::de::DeserializeOwned>(path: impl Into<PathBuf>) -> anyhow::Result<T> {
678    Ok(serde_json::from_reader(fs_err::File::open(path.into())?)?)
679}
680
681impl StoreConfig {
682    /// Connects to the storage backend and runs the given job against it.
683    pub async fn run_with_storage<Job>(
684        self,
685        wasm_runtime: Option<WasmRuntime>,
686        allow_application_logs: bool,
687        cache_sizes: StorageCacheSizes,
688        job: Job,
689    ) -> Result<Job::Output, anyhow::Error>
690    where
691        Job: Runnable,
692    {
693        match self {
694            StoreConfig::Memory {
695                config,
696                namespace,
697                genesis_path,
698            } => {
699                let mut storage = DbStorage::<MemoryDatabase, _>::maybe_create_and_connect(
700                    &config,
701                    &namespace,
702                    wasm_runtime,
703                    cache_sizes,
704                )
705                .await?
706                .with_allow_application_logs(allow_application_logs);
707                let genesis_config = read_json::<GenesisConfig>(genesis_path)?;
708                // Memory storage must be initialized every time.
709                genesis_config.initialize_storage(&mut storage).await?;
710                Ok(job.run(storage).await)
711            }
712            #[cfg(feature = "storage-service")]
713            StoreConfig::StorageService { config, namespace } => {
714                let storage = DbStorage::<StorageServiceDatabase, _>::connect(
715                    &config,
716                    &namespace,
717                    wasm_runtime,
718                    cache_sizes,
719                )
720                .await?
721                .with_allow_application_logs(allow_application_logs);
722                Ok(job.run(storage).await)
723            }
724            #[cfg(feature = "rocksdb")]
725            StoreConfig::RocksDb { config, namespace } => {
726                let storage = DbStorage::<RocksDbDatabase, _>::connect(
727                    &config,
728                    &namespace,
729                    wasm_runtime,
730                    cache_sizes,
731                )
732                .await?
733                .with_allow_application_logs(allow_application_logs);
734                Ok(job.run(storage).await)
735            }
736            #[cfg(feature = "scylladb")]
737            StoreConfig::ScyllaDb { config, namespace } => {
738                let storage = DbStorage::<ScyllaDbDatabase, _>::connect(
739                    &config,
740                    &namespace,
741                    wasm_runtime,
742                    cache_sizes,
743                )
744                .await?
745                .with_allow_application_logs(allow_application_logs);
746                Ok(job.run(storage).await)
747            }
748            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
749            StoreConfig::DualRocksDbScyllaDb { config, namespace } => {
750                let storage =
751                    DbStorage::<
752                        DualDatabase<RocksDbDatabase, ScyllaDbDatabase, ChainStatesFirstAssignment>,
753                        _,
754                    >::connect(&config, &namespace, wasm_runtime, cache_sizes)
755                    .await?
756                    .with_allow_application_logs(allow_application_logs);
757                Ok(job.run(storage).await)
758            }
759        }
760    }
761
762    /// Connects to the key-value store and runs the given store-level job against it.
763    #[allow(unused_variables)]
764    pub async fn run_with_store<Job>(
765        self,
766        cache_sizes: StorageCacheSizes,
767        job: Job,
768    ) -> Result<Job::Output, anyhow::Error>
769    where
770        Job: RunnableWithStore,
771    {
772        match self {
773            StoreConfig::Memory { .. } => {
774                Err(anyhow!("Cannot run admin operations on the memory store"))
775            }
776            #[cfg(feature = "storage-service")]
777            StoreConfig::StorageService { config, namespace } => Ok(job
778                .run::<StorageServiceDatabase>(config, namespace, cache_sizes)
779                .await?),
780            #[cfg(feature = "rocksdb")]
781            StoreConfig::RocksDb { config, namespace } => Ok(job
782                .run::<RocksDbDatabase>(config, namespace, cache_sizes)
783                .await?),
784            #[cfg(feature = "scylladb")]
785            StoreConfig::ScyllaDb { config, namespace } => Ok(job
786                .run::<ScyllaDbDatabase>(config, namespace, cache_sizes)
787                .await?),
788            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
789            StoreConfig::DualRocksDbScyllaDb { config, namespace } => Ok(job
790                .run::<DualDatabase<RocksDbDatabase, ScyllaDbDatabase, ChainStatesFirstAssignment>>(
791                    config,
792                    namespace,
793                    cache_sizes,
794                )
795                .await?),
796        }
797    }
798}
799
800/// A store-level job that migrates the storage to the latest version if needed.
801pub struct StorageMigration;
802
803#[async_trait]
804impl RunnableWithStore for StorageMigration {
805    type Output = ();
806
807    async fn run<D>(
808        self,
809        config: D::Config,
810        namespace: String,
811        cache_sizes: StorageCacheSizes,
812    ) -> Result<Self::Output, anyhow::Error>
813    where
814        D: KeyValueDatabase + Clone + Send + Sync + 'static,
815        D::Store: KeyValueStore + Clone + Send + Sync + 'static,
816        D::Error: Send + Sync,
817    {
818        if D::exists(&config, &namespace).await? {
819            let wasm_runtime = None;
820            let storage =
821                DbStorage::<D, WallClock>::connect(&config, &namespace, wasm_runtime, cache_sizes)
822                    .await?;
823            storage.migrate_if_needed().await?;
824        }
825        Ok(())
826    }
827}
828
829/// A store-level job that asserts the storage is already at version 1.
830pub struct AssertStorageV1;
831
832#[async_trait]
833impl RunnableWithStore for AssertStorageV1 {
834    type Output = ();
835
836    async fn run<D>(
837        self,
838        config: D::Config,
839        namespace: String,
840        cache_sizes: StorageCacheSizes,
841    ) -> Result<Self::Output, anyhow::Error>
842    where
843        D: KeyValueDatabase + Clone + Send + Sync + 'static,
844        D::Store: KeyValueStore + Clone + Send + Sync + 'static,
845        D::Error: Send + Sync,
846    {
847        if D::exists(&config, &namespace).await? {
848            let wasm_runtime = None;
849            let storage =
850                DbStorage::<D, WallClock>::connect(&config, &namespace, wasm_runtime, cache_sizes)
851                    .await?;
852            storage.assert_is_migrated_storage().await?;
853        }
854        Ok(())
855    }
856}
857
858#[test]
859fn test_memory_storage_config_from_str() {
860    assert_eq!(
861        StorageConfig::from_str("memory:path/to/genesis.json").unwrap(),
862        StorageConfig {
863            inner_storage_config: InnerStorageConfig::Memory {
864                genesis_path: PathBuf::from("path/to/genesis.json")
865            },
866            namespace: DEFAULT_NAMESPACE.into()
867        }
868    );
869    assert_eq!(
870        StorageConfig::from_str("memory:path/to/genesis.json:namespace").unwrap(),
871        StorageConfig {
872            inner_storage_config: InnerStorageConfig::Memory {
873                genesis_path: PathBuf::from("path/to/genesis.json")
874            },
875            namespace: "namespace".into()
876        }
877    );
878    assert!(StorageConfig::from_str("memory").is_err(),);
879}
880
881#[cfg(feature = "storage-service")]
882#[test]
883fn test_shared_store_config_from_str() {
884    assert_eq!(
885        StorageConfig::from_str("service:tcp:127.0.0.1:8942:linera").unwrap(),
886        StorageConfig {
887            inner_storage_config: InnerStorageConfig::Service {
888                endpoint: "127.0.0.1:8942".to_string()
889            },
890            namespace: "linera".into()
891        }
892    );
893    assert!(StorageConfig::from_str("service:tcp:127.0.0.1:8942").is_err());
894    assert!(StorageConfig::from_str("service:tcp:127.0.0.1:linera").is_err());
895}
896
897#[cfg(feature = "rocksdb")]
898#[test]
899fn test_rocks_db_storage_config_from_str() {
900    assert!(StorageConfig::from_str("rocksdb_foo.db").is_err());
901    assert_eq!(
902        StorageConfig::from_str("rocksdb:foo.db").unwrap(),
903        StorageConfig {
904            inner_storage_config: InnerStorageConfig::RocksDb {
905                path: "foo.db".into(),
906                spawn_mode: RocksDbSpawnMode::SpawnBlocking,
907            },
908            namespace: DEFAULT_NAMESPACE.to_string()
909        }
910    );
911    assert_eq!(
912        StorageConfig::from_str("rocksdb:foo.db:block_in_place").unwrap(),
913        StorageConfig {
914            inner_storage_config: InnerStorageConfig::RocksDb {
915                path: "foo.db".into(),
916                spawn_mode: RocksDbSpawnMode::BlockInPlace,
917            },
918            namespace: DEFAULT_NAMESPACE.to_string()
919        }
920    );
921    assert_eq!(
922        StorageConfig::from_str("rocksdb:foo.db:block_in_place:chosen_namespace").unwrap(),
923        StorageConfig {
924            inner_storage_config: InnerStorageConfig::RocksDb {
925                path: "foo.db".into(),
926                spawn_mode: RocksDbSpawnMode::BlockInPlace,
927            },
928            namespace: "chosen_namespace".into()
929        }
930    );
931}
932
933#[cfg(feature = "scylladb")]
934#[test]
935fn test_scylla_db_storage_config_from_str() {
936    assert_eq!(
937        StorageConfig::from_str("scylladb:").unwrap(),
938        StorageConfig {
939            inner_storage_config: InnerStorageConfig::ScyllaDb {
940                uri: "localhost:9042".to_string()
941            },
942            namespace: DEFAULT_NAMESPACE.to_string()
943        }
944    );
945    assert_eq!(
946        StorageConfig::from_str("scylladb:tcp:db_hostname:230:table_other_storage").unwrap(),
947        StorageConfig {
948            inner_storage_config: InnerStorageConfig::ScyllaDb {
949                uri: "db_hostname:230".to_string()
950            },
951            namespace: "table_other_storage".to_string()
952        }
953    );
954    assert_eq!(
955        StorageConfig::from_str("scylladb:tcp:db_hostname:230").unwrap(),
956        StorageConfig {
957            inner_storage_config: InnerStorageConfig::ScyllaDb {
958                uri: "db_hostname:230".to_string()
959            },
960            namespace: DEFAULT_NAMESPACE.to_string()
961        }
962    );
963    assert!(StorageConfig::from_str("scylladb:-10").is_err());
964    assert!(StorageConfig::from_str("scylladb:70000").is_err());
965    assert!(StorageConfig::from_str("scylladb:230:234").is_err());
966    assert!(StorageConfig::from_str("scylladb:tcp:address1").is_err());
967    assert!(StorageConfig::from_str("scylladb:tcp:address1:tcp:/address2").is_err());
968    assert!(StorageConfig::from_str("scylladb:wrong").is_err());
969}
linera_storage_runtime/lib.rs

linera_storage_runtime/
lib.rs
