linera_storage_runtime/
storage_config.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{fmt, path::PathBuf, str::FromStr};
5
6use anyhow::{anyhow, bail};
7use linera_storage::DEFAULT_NAMESPACE;
8#[cfg(feature = "rocksdb")]
9use linera_views::rocks_db::{PathWithGuard, RocksDbSpawnMode};
10use tracing::error;
11#[cfg(all(feature = "rocksdb", feature = "scylladb"))]
12use {linera_views::backends::dual::DualStoreConfig, std::path::Path};
13#[cfg(feature = "scylladb")]
14use {std::num::NonZeroU16, tracing::debug};
15
16use crate::{CommonStorageOptions, StoreConfig};
17
18/// The description of a storage implementation.
19#[derive(Clone, Debug)]
20#[cfg_attr(any(test), derive(Eq, PartialEq))]
21pub enum InnerStorageConfig {
22    /// The memory description.
23    Memory {
24        /// The path to the genesis configuration. This is needed because we reinitialize
25        /// memory databases from the genesis config everytime.
26        genesis_path: PathBuf,
27    },
28    /// The storage service description.
29    #[cfg(feature = "storage-service")]
30    Service {
31        /// The endpoint used.
32        endpoint: String,
33    },
34    /// The RocksDB description.
35    #[cfg(feature = "rocksdb")]
36    RocksDb {
37        /// The path used.
38        path: PathBuf,
39        /// Whether to use `block_in_place` or `spawn_blocking`.
40        spawn_mode: RocksDbSpawnMode,
41    },
42    /// The ScyllaDB description.
43    #[cfg(feature = "scylladb")]
44    ScyllaDb {
45        /// The URI for accessing the database.
46        uri: String,
47    },
48    /// The dual RocksDB / ScyllaDB description.
49    #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
50    DualRocksDbScyllaDb {
51        /// The path used.
52        path_with_guard: PathWithGuard,
53        /// Whether to use `block_in_place` or `spawn_blocking`.
54        spawn_mode: RocksDbSpawnMode,
55        /// The URI for accessing the database.
56        uri: String,
57    },
58}
59
60/// The description of a storage implementation.
61#[derive(Clone, Debug)]
62#[cfg_attr(any(test), derive(Eq, PartialEq))]
63pub struct StorageConfig {
64    /// The inner storage config.
65    pub inner_storage_config: InnerStorageConfig,
66    /// The namespace used
67    pub namespace: String,
68}
69
70const MEMORY: &str = "memory:";
71#[cfg(feature = "storage-service")]
72const STORAGE_SERVICE: &str = "service:";
73#[cfg(feature = "rocksdb")]
74const ROCKS_DB: &str = "rocksdb:";
75#[cfg(feature = "scylladb")]
76const SCYLLA_DB: &str = "scylladb:";
77#[cfg(all(feature = "rocksdb", feature = "scylladb"))]
78const DUAL_ROCKS_DB_SCYLLA_DB: &str = "dualrocksdbscylladb:";
79
80impl FromStr for StorageConfig {
81    type Err = anyhow::Error;
82
83    fn from_str(input: &str) -> Result<Self, Self::Err> {
84        if let Some(s) = input.strip_prefix(MEMORY) {
85            let parts = s.split(':').collect::<Vec<_>>();
86            if parts.len() == 1 {
87                let genesis_path = parts[0].to_string().into();
88                let namespace = DEFAULT_NAMESPACE.to_string();
89                let inner_storage_config = InnerStorageConfig::Memory { genesis_path };
90                return Ok(StorageConfig {
91                    inner_storage_config,
92                    namespace,
93                });
94            }
95            if parts.len() != 2 {
96                bail!("We should have one genesis config path and one optional namespace");
97            }
98            let genesis_path = parts[0].to_string().into();
99            let namespace = parts[1].to_string();
100            let inner_storage_config = InnerStorageConfig::Memory { genesis_path };
101            return Ok(StorageConfig {
102                inner_storage_config,
103                namespace,
104            });
105        }
106        #[cfg(feature = "storage-service")]
107        if let Some(s) = input.strip_prefix(STORAGE_SERVICE) {
108            if s.is_empty() {
109                bail!(
110                    "For Storage service, the formatting has to be service:endpoint:namespace,\
111example service:tcp:127.0.0.1:7878:table_do_my_test"
112                );
113            }
114            let parts = s.split(':').collect::<Vec<_>>();
115            if parts.len() != 4 {
116                bail!("We should have one endpoint and one namespace");
117            }
118            let protocol = parts[0];
119            if protocol != "tcp" {
120                bail!("Only allowed protocol is tcp");
121            }
122            let endpoint = parts[1];
123            let port = parts[2];
124            let mut endpoint = endpoint.to_string();
125            endpoint.push(':');
126            endpoint.push_str(port);
127            let endpoint = endpoint.to_string();
128            let namespace = parts[3].to_string();
129            let inner_storage_config = InnerStorageConfig::Service { endpoint };
130            return Ok(StorageConfig {
131                inner_storage_config,
132                namespace,
133            });
134        }
135        #[cfg(feature = "rocksdb")]
136        if let Some(s) = input.strip_prefix(ROCKS_DB) {
137            if s.is_empty() {
138                bail!(
139                    "For RocksDB, the formatting has to be rocksdb:directory or rocksdb:directory:spawn_mode:namespace");
140            }
141            let parts = s.split(':').collect::<Vec<_>>();
142            if parts.len() == 1 {
143                let path = parts[0].to_string().into();
144                let namespace = DEFAULT_NAMESPACE.to_string();
145                let spawn_mode = RocksDbSpawnMode::SpawnBlocking;
146                let inner_storage_config = InnerStorageConfig::RocksDb { path, spawn_mode };
147                return Ok(StorageConfig {
148                    inner_storage_config,
149                    namespace,
150                });
151            }
152            if parts.len() == 2 || parts.len() == 3 {
153                let path = parts[0].to_string().into();
154                let spawn_mode_name = parts
155                    .get(1)
156                    .copied()
157                    .expect("validated by the parts length check above");
158                let spawn_mode = match spawn_mode_name {
159                    "spawn_blocking" => Ok(RocksDbSpawnMode::SpawnBlocking),
160                    "block_in_place" => Ok(RocksDbSpawnMode::BlockInPlace),
161                    "runtime" => Ok(RocksDbSpawnMode::get_spawn_mode_from_runtime()),
162                    _ => Err(anyhow!("Failed to parse {spawn_mode_name} as a spawn_mode")),
163                }?;
164                let namespace = if parts.len() == 2 {
165                    DEFAULT_NAMESPACE.to_string()
166                } else {
167                    parts[2].to_string()
168                };
169                let inner_storage_config = InnerStorageConfig::RocksDb { path, spawn_mode };
170                return Ok(StorageConfig {
171                    inner_storage_config,
172                    namespace,
173                });
174            }
175            bail!("We should have one, two or three parts");
176        }
177        #[cfg(feature = "scylladb")]
178        if let Some(s) = input.strip_prefix(SCYLLA_DB) {
179            let mut uri: Option<String> = None;
180            let mut namespace: Option<String> = None;
181            let parse_error: &'static str = "Correct format is tcp:db_hostname:port.";
182            if !s.is_empty() {
183                let mut parts = s.split(':');
184                while let Some(part) = parts.next() {
185                    match part {
186                        "tcp" => {
187                            let address = parts.next().ok_or_else(|| {
188                                anyhow!("Failed to find address for {s}. {parse_error}")
189                            })?;
190                            let port_str = parts.next().ok_or_else(|| {
191                                anyhow!("Failed to find port for {s}. {parse_error}")
192                            })?;
193                            let port = NonZeroU16::from_str(port_str).map_err(|_| {
194                                anyhow!(
195                                    "Failed to find parse port {port_str} for {s}. {parse_error}",
196                                )
197                            })?;
198                            if uri.is_some() {
199                                bail!("The uri has already been assigned");
200                            }
201                            uri = Some(format!("{address}:{port}"));
202                        }
203                        _ if part.starts_with("table") => {
204                            if namespace.is_some() {
205                                bail!("The namespace has already been assigned");
206                            }
207                            namespace = Some(part.to_string());
208                        }
209                        _ => {
210                            bail!("the entry \"{part}\" is not matching");
211                        }
212                    }
213                }
214            }
215            let uri = uri.unwrap_or_else(|| "localhost:9042".to_string());
216            let namespace = namespace.unwrap_or_else(|| DEFAULT_NAMESPACE.to_string());
217            let inner_storage_config = InnerStorageConfig::ScyllaDb { uri };
218            debug!("ScyllaDB connection info: {:?}", inner_storage_config);
219            return Ok(StorageConfig {
220                inner_storage_config,
221                namespace,
222            });
223        }
224        #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
225        if let Some(s) = input.strip_prefix(DUAL_ROCKS_DB_SCYLLA_DB) {
226            let parts = s.split(':').collect::<Vec<_>>();
227            if parts.len() != 5 && parts.len() != 6 {
228                bail!(
229                    "For DualRocksDbScyllaDb, the formatting has to be dualrocksdbscylladb:directory:mode:tcp:hostname:port:namespace"
230                );
231            }
232            let path = Path::new(parts[0]);
233            let path = path.to_path_buf();
234            let path_with_guard = PathWithGuard::new(path);
235            let spawn_mode_name = parts
236                .get(1)
237                .copied()
238                .expect("validated by the parts length check above");
239            let spawn_mode = match spawn_mode_name {
240                "spawn_blocking" => Ok(RocksDbSpawnMode::SpawnBlocking),
241                "block_in_place" => Ok(RocksDbSpawnMode::BlockInPlace),
242                "runtime" => Ok(RocksDbSpawnMode::get_spawn_mode_from_runtime()),
243                _ => Err(anyhow!("Failed to parse {spawn_mode_name} as a spawn_mode",)),
244            }?;
245            let protocol = parts[2];
246            if protocol != "tcp" {
247                bail!("The only allowed protocol is tcp");
248            }
249            let address = parts[3];
250            let port_str = parts[4];
251            let port = NonZeroU16::from_str(port_str)
252                .map_err(|_| anyhow!("Failed to find parse port {port_str} for {s}"))?;
253            let uri = format!("{address}:{port}");
254            let inner_storage_config = InnerStorageConfig::DualRocksDbScyllaDb {
255                path_with_guard,
256                spawn_mode,
257                uri,
258            };
259            let namespace = if parts.len() == 5 {
260                DEFAULT_NAMESPACE.to_string()
261            } else {
262                parts[5].to_string()
263            };
264            return Ok(StorageConfig {
265                inner_storage_config,
266                namespace,
267            });
268        }
269        error!("available storage: memory");
270        #[cfg(feature = "storage-service")]
271        error!("Also available is linera-storage-service");
272        #[cfg(feature = "rocksdb")]
273        error!("Also available is RocksDB");
274        #[cfg(feature = "scylladb")]
275        error!("Also available is ScyllaDB");
276        #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
277        error!("Also available is DualRocksDbScyllaDb");
278        Err(anyhow!("The input has not matched: {input}"))
279    }
280}
281
282impl StorageConfig {
283    /// Appends a shard-specific subdirectory to the storage path, if applicable.
284    #[allow(unused_variables)]
285    pub fn maybe_append_shard_path(&mut self, shard: usize) -> std::io::Result<()> {
286        match &mut self.inner_storage_config {
287            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
288            InnerStorageConfig::DualRocksDbScyllaDb {
289                path_with_guard,
290                spawn_mode: _,
291                uri: _,
292            } => {
293                let shard_str = format!("shard_{shard}");
294                path_with_guard.path_buf.push(shard_str);
295                std::fs::create_dir_all(&path_with_guard.path_buf)
296            }
297            _ => Ok(()),
298        }
299    }
300
301    /// The addition of the common config to get a full configuration
302    pub fn add_common_storage_options(
303        &self,
304        #[cfg_attr(
305            not(any(feature = "storage-service", feature = "rocksdb", feature = "scylladb")),
306            allow(unused_variables)
307        )]
308        options: &CommonStorageOptions,
309    ) -> Result<StoreConfig, anyhow::Error> {
310        let namespace = self.namespace.clone();
311        match &self.inner_storage_config {
312            InnerStorageConfig::Memory { genesis_path } => {
313                let config = linera_views::memory::MemoryStoreConfig {
314                    kill_on_drop: false,
315                };
316                let genesis_path = genesis_path.clone();
317                Ok(StoreConfig::Memory {
318                    config,
319                    namespace,
320                    genesis_path,
321                })
322            }
323            #[cfg(feature = "storage-service")]
324            InnerStorageConfig::Service { endpoint } => {
325                let inner_config =
326                    linera_storage_service::common::StorageServiceStoreInternalConfig {
327                        endpoint: endpoint.clone(),
328                        max_concurrent_queries: options.storage_max_concurrent_queries,
329                    };
330                let config = linera_storage_service::common::StorageServiceStoreConfig {
331                    inner_config,
332                    storage_cache_config: options.views_storage_cache_config(),
333                };
334                Ok(StoreConfig::StorageService { config, namespace })
335            }
336            #[cfg(feature = "rocksdb")]
337            InnerStorageConfig::RocksDb { path, spawn_mode } => {
338                let path_with_guard = PathWithGuard::new(path.to_path_buf());
339                let inner_config = linera_views::rocks_db::RocksDbStoreInternalConfig {
340                    spawn_mode: *spawn_mode,
341                    path_with_guard,
342                    enable_statistics: options.rocksdb_enable_statistics,
343                    statistics_level: options.rocksdb_statistics_level,
344                };
345                let config = linera_views::rocks_db::RocksDbStoreConfig {
346                    inner_config,
347                    storage_cache_config: options.views_storage_cache_config(),
348                };
349                Ok(StoreConfig::RocksDb { config, namespace })
350            }
351            #[cfg(feature = "scylladb")]
352            InnerStorageConfig::ScyllaDb { uri } => {
353                let inner_config = linera_views::scylla_db::ScyllaDbStoreInternalConfig {
354                    uri: uri.clone(),
355                    max_concurrent_queries: options.storage_max_concurrent_queries,
356                    replication_factor: options.storage_replication_factor,
357                };
358                let config = linera_views::scylla_db::ScyllaDbStoreConfig {
359                    inner_config,
360                    storage_cache_config: options.views_storage_cache_config(),
361                };
362                Ok(StoreConfig::ScyllaDb { config, namespace })
363            }
364            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
365            InnerStorageConfig::DualRocksDbScyllaDb {
366                path_with_guard,
367                spawn_mode,
368                uri,
369            } => {
370                let inner_config = linera_views::rocks_db::RocksDbStoreInternalConfig {
371                    spawn_mode: *spawn_mode,
372                    path_with_guard: path_with_guard.clone(),
373                    enable_statistics: options.rocksdb_enable_statistics,
374                    statistics_level: options.rocksdb_statistics_level,
375                };
376                let first_config = linera_views::rocks_db::RocksDbStoreConfig {
377                    inner_config,
378                    storage_cache_config: options.views_storage_cache_config(),
379                };
380
381                let inner_config = linera_views::scylla_db::ScyllaDbStoreInternalConfig {
382                    uri: uri.clone(),
383                    max_concurrent_queries: options.storage_max_concurrent_queries,
384                    replication_factor: options.storage_replication_factor,
385                };
386                let second_config = linera_views::scylla_db::ScyllaDbStoreConfig {
387                    inner_config,
388                    storage_cache_config: options.views_storage_cache_config(),
389                };
390
391                let config = DualStoreConfig {
392                    first_config,
393                    second_config,
394                };
395                Ok(StoreConfig::DualRocksDbScyllaDb { config, namespace })
396            }
397        }
398    }
399}
400
401impl fmt::Display for StorageConfig {
402    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
403        let namespace = &self.namespace;
404        match &self.inner_storage_config {
405            #[cfg(feature = "storage-service")]
406            InnerStorageConfig::Service { endpoint } => {
407                write!(f, "service:tcp:{endpoint}:{namespace}")
408            }
409            InnerStorageConfig::Memory { genesis_path } => {
410                write!(f, "memory:{}:{}", genesis_path.display(), namespace)
411            }
412            #[cfg(feature = "rocksdb")]
413            InnerStorageConfig::RocksDb { path, spawn_mode } => {
414                let spawn_mode = spawn_mode.to_string();
415                write!(f, "rocksdb:{}:{}:{}", path.display(), spawn_mode, namespace)
416            }
417            #[cfg(feature = "scylladb")]
418            InnerStorageConfig::ScyllaDb { uri } => {
419                write!(f, "scylladb:tcp:{uri}:{namespace}")
420            }
421            #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
422            InnerStorageConfig::DualRocksDbScyllaDb {
423                path_with_guard,
424                spawn_mode,
425                uri,
426            } => {
427                write!(
428                    f,
429                    "dualrocksdbscylladb:{}:{}:tcp:{}:{}",
430                    path_with_guard.path_buf.display(),
431                    spawn_mode,
432                    uri,
433                    namespace
434                )
435            }
436        }
437    }
438}
439
440#[test]
441fn test_memory_storage_config_from_str() {
442    assert_eq!(
443        StorageConfig::from_str("memory:path/to/genesis.json").unwrap(),
444        StorageConfig {
445            inner_storage_config: InnerStorageConfig::Memory {
446                genesis_path: PathBuf::from("path/to/genesis.json")
447            },
448            namespace: DEFAULT_NAMESPACE.into()
449        }
450    );
451    assert_eq!(
452        StorageConfig::from_str("memory:path/to/genesis.json:namespace").unwrap(),
453        StorageConfig {
454            inner_storage_config: InnerStorageConfig::Memory {
455                genesis_path: PathBuf::from("path/to/genesis.json")
456            },
457            namespace: "namespace".into()
458        }
459    );
460    assert!(StorageConfig::from_str("memory").is_err(),);
461}
462
463#[cfg(feature = "storage-service")]
464#[test]
465fn test_shared_store_config_from_str() {
466    assert_eq!(
467        StorageConfig::from_str("service:tcp:127.0.0.1:8942:linera").unwrap(),
468        StorageConfig {
469            inner_storage_config: InnerStorageConfig::Service {
470                endpoint: "127.0.0.1:8942".to_string()
471            },
472            namespace: "linera".into()
473        }
474    );
475    assert!(StorageConfig::from_str("service:tcp:127.0.0.1:8942").is_err());
476    assert!(StorageConfig::from_str("service:tcp:127.0.0.1:linera").is_err());
477}
478
479#[cfg(feature = "rocksdb")]
480#[test]
481fn test_rocks_db_storage_config_from_str() {
482    assert!(StorageConfig::from_str("rocksdb_foo.db").is_err());
483    assert_eq!(
484        StorageConfig::from_str("rocksdb:foo.db").unwrap(),
485        StorageConfig {
486            inner_storage_config: InnerStorageConfig::RocksDb {
487                path: "foo.db".into(),
488                spawn_mode: RocksDbSpawnMode::SpawnBlocking,
489            },
490            namespace: DEFAULT_NAMESPACE.to_string()
491        }
492    );
493    assert_eq!(
494        StorageConfig::from_str("rocksdb:foo.db:block_in_place").unwrap(),
495        StorageConfig {
496            inner_storage_config: InnerStorageConfig::RocksDb {
497                path: "foo.db".into(),
498                spawn_mode: RocksDbSpawnMode::BlockInPlace,
499            },
500            namespace: DEFAULT_NAMESPACE.to_string()
501        }
502    );
503    assert_eq!(
504        StorageConfig::from_str("rocksdb:foo.db:block_in_place:chosen_namespace").unwrap(),
505        StorageConfig {
506            inner_storage_config: InnerStorageConfig::RocksDb {
507                path: "foo.db".into(),
508                spawn_mode: RocksDbSpawnMode::BlockInPlace,
509            },
510            namespace: "chosen_namespace".into()
511        }
512    );
513}
514
515#[cfg(feature = "scylladb")]
516#[test]
517fn test_scylla_db_storage_config_from_str() {
518    assert_eq!(
519        StorageConfig::from_str("scylladb:").unwrap(),
520        StorageConfig {
521            inner_storage_config: InnerStorageConfig::ScyllaDb {
522                uri: "localhost:9042".to_string()
523            },
524            namespace: DEFAULT_NAMESPACE.to_string()
525        }
526    );
527    assert_eq!(
528        StorageConfig::from_str("scylladb:tcp:db_hostname:230:table_other_storage").unwrap(),
529        StorageConfig {
530            inner_storage_config: InnerStorageConfig::ScyllaDb {
531                uri: "db_hostname:230".to_string()
532            },
533            namespace: "table_other_storage".to_string()
534        }
535    );
536    assert_eq!(
537        StorageConfig::from_str("scylladb:tcp:db_hostname:230").unwrap(),
538        StorageConfig {
539            inner_storage_config: InnerStorageConfig::ScyllaDb {
540                uri: "db_hostname:230".to_string()
541            },
542            namespace: DEFAULT_NAMESPACE.to_string()
543        }
544    );
545    assert!(StorageConfig::from_str("scylladb:-10").is_err());
546    assert!(StorageConfig::from_str("scylladb:70000").is_err());
547    assert!(StorageConfig::from_str("scylladb:230:234").is_err());
548    assert!(StorageConfig::from_str("scylladb:tcp:address1").is_err());
549    assert!(StorageConfig::from_str("scylladb:tcp:address1:tcp:/address2").is_err());
550    assert!(StorageConfig::from_str("scylladb:wrong").is_err());
551}
linera_storage_runtime/storage_config.rs

linera_storage_runtime/
storage_config.rs
