espresso_node/persistence/sql/
sqlite.rs

1//! Persistence on an embedded SQLite file.
2
3use std::path::Path;
4
5use anyhow::Context as _;
6use hotshot_query_service::{data_source::storage::sql::sqlite as qs, include_migrations};
7use light_client::storage::LightClientDbOptions;
8use sqlx::{Decode, Row, Type};
9
10#[expect(
11    clippy::duplicate_mod,
12    reason = "each backend compiles the shared tree against its own `Db`"
13)]
14#[path = "../sql.rs"]
15mod imp;
16
17pub use imp::*;
18use qs::{Config, SqlStorage};
19
20use super::Options as SqlOptions;
21#[cfg(any(test, feature = "testing"))]
22use super::SqliteOptions;
23use crate::{
24    api::options::Query,
25    persistence::storage_probe::{self, SqlitePragmas, StorageProbe},
26};
27
28impl BackendOptions for qs::Db {
29    fn apply(mut cfg: Config, opt: &SqlOptions) -> anyhow::Result<Config> {
30        let path = opt
31            .sqlite_options
32            .path
33            .clone()
34            .context("ESPRESSO_NODE_STORAGE_PATH must be set for an embedded database")?;
35        if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) {
36            std::fs::create_dir_all(dir)
37                .with_context(|| format!("failed to create directory: {dir:?}"))?;
38        }
39
40        cfg = cfg.migrations(include_migrations!(
41            "$CARGO_MANIFEST_DIR/api/migrations/sqlite"
42        ));
43
44        Ok(cfg.db_path(path))
45    }
46
47    fn light_client_db(_cfg: &Config, query: &Query) -> LightClientDbOptions {
48        LightClientDbOptions::Sqlite(query.sqlite_light_client_db())
49    }
50
51    async fn reset_light_client(_cfg: &Config) -> anyhow::Result<()> {
52        Ok(())
53    }
54
55    async fn storage_probe(
56        db: &SqlStorage,
57        opt: &SqlOptions,
58    ) -> anyhow::Result<Option<StorageProbe>> {
59        let pragmas = read_pragmas(&db.pool()).await;
60        let path = opt
61            .sqlite_options
62            .path
63            .clone()
64            .context("ESPRESSO_NODE_STORAGE_PATH must be set for an embedded database")?;
65        Ok(Some(
66            storage_probe::probe(sqlite_probe_dir(&path), pragmas).await?,
67        ))
68    }
69}
70
71/// The directory `storage_probe::probe` should classify for a SQLite database at `path`. Falls
72/// back to `.` when `path` has no parent, e.g. a bare relative filename, or the empty `PathBuf`
73/// from `SqliteOptions::default()`.
74fn sqlite_probe_dir(path: &Path) -> &Path {
75    match path.parent() {
76        Some(parent) if !parent.as_os_str().is_empty() => parent,
77        _ => Path::new("."),
78    }
79}
80
81async fn read_pragma<T>(pool: &sqlx::Pool<qs::Db>, pragma: &str) -> Option<T>
82where
83    T: for<'a> Decode<'a, qs::Db> + Type<qs::Db>,
84{
85    match sqlx::query(pragma).fetch_one(pool).await {
86        Ok(row) => row.try_get(0).ok(),
87        Err(err) => {
88            tracing::debug!(pragma, ?err, "storage probe: pragma query failed");
89            None
90        },
91    }
92}
93
94/// The first failure short-circuits the rest: a pragma query only fails if the pool is unusable.
95async fn read_pragmas(pool: &sqlx::Pool<qs::Db>) -> Option<SqlitePragmas> {
96    let journal_mode = read_pragma(pool, "PRAGMA journal_mode").await?;
97    // SQLite reports these two back as their numeric settings, not the names used to set them.
98    let synchronous = match read_pragma::<i64>(pool, "PRAGMA synchronous").await? {
99        0 => "off",
100        1 => "normal",
101        2 => "full",
102        3 => "extra",
103        _ => "unknown",
104    };
105    let auto_vacuum = match read_pragma::<i64>(pool, "PRAGMA auto_vacuum").await? {
106        0 => "none",
107        1 => "full",
108        2 => "incremental",
109        _ => "unknown",
110    };
111    let page_size: i64 = read_pragma(pool, "PRAGMA page_size").await?;
112
113    Some(SqlitePragmas {
114        journal_mode,
115        synchronous,
116        auto_vacuum,
117        page_size: page_size as u64,
118    })
119}
120
121#[cfg(any(test, feature = "testing"))]
122pub(crate) fn tmp_options(db: &qs::testing::TmpDb) -> SqlOptions {
123    SqliteOptions {
124        path: Some(db.path()),
125    }
126    .into()
127}
128
129#[cfg(test)]
130mod test {
131    use std::path::{Path, PathBuf};
132
133    use espresso_types::traits::{PersistenceOptions as _, SequencerPersistence as _};
134    use hotshot_query_service::metrics::PrometheusMetrics;
135    use hotshot_types::traits::metrics::Metrics;
136
137    use super::{
138        Config, SqlOptions, SqliteOptions, imp::Persistence, read_pragmas, sqlite_probe_dir,
139    };
140    use crate::persistence::tests::TestablePersistence as _;
141
142    #[test]
143    fn embedded_db_without_a_storage_path_is_an_error() {
144        let opt = SqlOptions {
145            embedded_db: true,
146            sqlite_options: SqliteOptions { path: None },
147            ..SqlOptions::default()
148        };
149
150        let Err(err) = Config::try_from(&opt) else {
151            panic!("an embedded database with no storage path was accepted");
152        };
153        assert!(
154            format!("{err:#}").contains("ESPRESSO_NODE_STORAGE_PATH"),
155            "{err:#}"
156        );
157    }
158
159    #[tokio::test]
160    async fn read_pragmas_falls_back_to_none_on_query_error() {
161        let pool = sqlx::sqlite::SqlitePoolOptions::new()
162            .connect(":memory:")
163            .await
164            .unwrap();
165        pool.close().await;
166
167        assert!(read_pragmas(&pool).await.is_none());
168    }
169
170    #[tokio::test]
171    async fn read_pragmas_reads_live_values() {
172        let pool = sqlx::sqlite::SqlitePoolOptions::new()
173            .connect(":memory:")
174            .await
175            .unwrap();
176
177        let pragmas = read_pragmas(&pool).await.expect("pragmas readable");
178
179        assert_eq!(pragmas.journal_mode, "memory");
180        assert_ne!(pragmas.synchronous, "unknown");
181        assert!(pragmas.page_size > 0);
182    }
183
184    #[test]
185    fn sqlite_probe_dir_falls_back_to_cwd_without_a_parent() {
186        // `SqliteOptions::default()`'s `path` is empty, whose `parent()` is `Some("")`, not `None`.
187        assert_eq!(sqlite_probe_dir(&PathBuf::new()), Path::new("."));
188        assert_eq!(sqlite_probe_dir(Path::new("database")), Path::new("."));
189        assert_eq!(
190            sqlite_probe_dir(Path::new("/var/lib/espresso/sqlite/database")),
191            Path::new("/var/lib/espresso/sqlite")
192        );
193    }
194
195    /// The probe is taken in `create()` and only reaches the exported registry through
196    /// `enable_metrics`, which `init_node` calls.
197    #[test_log::test(tokio::test(flavor = "multi_thread"))]
198    async fn test_storage_probe_reaches_exported_registry() {
199        let tmp = Persistence::tmp_storage().await;
200        let mut persistence = Persistence::options(&tmp).create().await.unwrap();
201
202        // Dropping WAL from `sqlite_options()` should fail this assertion.
203        assert_eq!(
204            persistence
205                .probe
206                .as_ref()
207                .and_then(|probe| probe.pragmas.as_ref())
208                .map(|pragmas| pragmas.journal_mode.as_str()),
209            Some("wal")
210        );
211
212        let metrics = PrometheusMetrics::default();
213        persistence.enable_metrics(&*Metrics::subgroup(&metrics, "consensus".to_string()));
214
215        let exported = metrics.export().unwrap();
216        assert!(exported.contains("consensus_disk_info"), "{exported}");
217        assert!(exported.contains("backend=\"sqlite\""), "{exported}");
218        assert!(
219            exported.contains("consensus_disk_fsync_micros"),
220            "{exported}"
221        );
222    }
223}
espresso_node/persistence/sql/sqlite.rs

espresso_node/persistence/sql/
sqlite.rs
