aven_core/
db.rs

1use crate::ids::WorkspaceId;
2use std::fs;
3use std::ops::{Deref, DerefMut};
4use std::path::{Path, PathBuf};
5use std::str::FromStr;
6use std::sync::Arc;
7use std::time::Duration;
8
9use anyhow::{Context, Result};
10#[cfg(not(any(test, feature = "test-support")))]
11use sqlx::Transaction;
12use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
13use sqlx::{Connection as _, Sqlite, SqliteConnection, SqlitePool};
14use tokio::sync::{Mutex, OwnedMutexGuard};
15
16use crate::ids::new_id;
17use crate::workspaces::ensure_default_workspace;
18
19mod backup;
20mod changes;
21mod field_versions;
22mod inspection;
23pub mod installation;
24mod rows;
25
26pub use backup::{
27    backup_database, default_backup_path, default_sqlite_backup_path, restore_database_file,
28    shm_path, wal_path,
29};
30pub(crate) use backup::{
31    backup_database_with_connection, create_restore_safety_backup, detach_backup_snapshot,
32    ensure_file_has_no_active_local_shared_capture,
33};
34pub(crate) use changes::{IdentifiedChange, insert_change, insert_change_with_identity};
35pub(crate) use field_versions::{
36    conflict_exists, entity_conflict_exists, entity_field_version, field_version,
37    set_entity_field_version, set_field_version,
38};
39pub use inspection::{DatabaseInspection, InspectedDatabase};
40pub(crate) use rows::{
41    recurrence_occurrence_from_row, recurrence_pause_interval_from_row, recurrence_series_from_row,
42    recurrence_series_label_from_row, task_from_row,
43};
44
45static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
46const FILE_DATABASE_CONNECTIONS: u32 = 5;
47// Bounds retained WAL space on reuse after checkpointing, not transaction size.
48const WAL_JOURNAL_SIZE_LIMIT: u64 = 4 * 1024 * 1024;
49
50#[derive(Clone)]
51pub struct Database {
52    pool: SqlitePool,
53    writer: Arc<Mutex<()>>,
54    path: PathBuf,
55    file_identity: Option<PathBuf>,
56    _inspection_dir: Option<Arc<tempfile::TempDir>>,
57    pub(crate) membership_cache: crate::sync::seed_claim::membership::persistence::Cache,
58}
59
60pub(crate) struct WriterConnection {
61    connection: sqlx::pool::PoolConnection<Sqlite>,
62    _guard: OwnedMutexGuard<()>,
63    #[cfg(any(test, feature = "test-support"))]
64    acquired: std::time::Instant,
65}
66
67impl Deref for WriterConnection {
68    type Target = SqliteConnection;
69
70    fn deref(&self) -> &Self::Target {
71        &self.connection
72    }
73}
74
75impl DerefMut for WriterConnection {
76    fn deref_mut(&mut self) -> &mut Self::Target {
77        &mut self.connection
78    }
79}
80
81#[cfg(any(test, feature = "test-support"))]
82impl Drop for WriterConnection {
83    fn drop(&mut self) {
84        crate::test_support::writer_timing::record_gate_hold(self.acquired.elapsed());
85    }
86}
87
88impl Database {
89    pub async fn open(path: &Path) -> Result<Self> {
90        let connection_input = path.to_string_lossy();
91        let options = SqliteConnectOptions::from_str(&connection_input)?;
92        let storage = database_storage(&connection_input, &options);
93        let pool = open_db(path).await?;
94        let file_identity = match storage {
95            DatabaseStorage::InMemory => None,
96            DatabaseStorage::File => {
97                Some(fs::canonicalize(options.get_filename()).with_context(|| {
98                    format!(
99                        "could not resolve database path {}",
100                        options.get_filename().display()
101                    )
102                })?)
103            }
104        };
105        Ok(Self {
106            pool,
107            writer: Arc::new(Mutex::new(())),
108            path: path.to_path_buf(),
109            file_identity,
110            _inspection_dir: None,
111            membership_cache: Default::default(),
112        })
113    }
114
115    pub fn path(&self) -> &Path {
116        &self.path
117    }
118
119    /// Closes every connection, including those held by clones, and waits
120    /// until SQLite has released the files. Dropping a database closes its
121    /// connections in the background instead, so delete or replace the files
122    /// only after this returns.
123    pub async fn close(&self) {
124        self.pool.close().await;
125    }
126
127    #[doc(hidden)]
128    pub fn file_identity(&self) -> Option<&Path> {
129        self.file_identity.as_deref()
130    }
131
132    pub async fn meta(&self, key: &str) -> Result<Option<String>> {
133        let mut conn = self.acquire_reader().await?;
134        get_meta(&mut conn, key).await
135    }
136
137    pub async fn conflict_exists(
138        &self,
139        workspace_id: &WorkspaceId,
140        task_id: &crate::ids::TaskId,
141        field: &str,
142    ) -> Result<bool> {
143        let mut conn = self.acquire_reader().await?;
144        conflict_exists(&mut conn, workspace_id, task_id, field).await
145    }
146
147    #[cfg(any(test, feature = "test-support"))]
148    pub(crate) fn pool(&self) -> &SqlitePool {
149        &self.pool
150    }
151
152    pub(crate) async fn acquire_reader(&self) -> Result<sqlx::pool::PoolConnection<Sqlite>> {
153        Ok(self.pool.acquire().await?)
154    }
155
156    pub(crate) async fn acquire_writer(&self) -> Result<WriterConnection> {
157        #[cfg(any(test, feature = "test-support"))]
158        let waiting = std::time::Instant::now();
159        let guard = self.writer.clone().lock_owned().await;
160        let connection = self.pool.acquire().await?;
161        #[cfg(any(test, feature = "test-support"))]
162        crate::test_support::writer_timing::record_gate_wait(waiting.elapsed());
163        Ok(WriterConnection {
164            connection,
165            _guard: guard,
166            #[cfg(any(test, feature = "test-support"))]
167            acquired: std::time::Instant::now(),
168        })
169    }
170}
171
172pub(crate) async fn open_db(path: &Path) -> Result<SqlitePool> {
173    let connection_input = path.to_string_lossy();
174    let mut options = SqliteConnectOptions::from_str(&connection_input)?;
175    let storage = database_storage(&connection_input, &options);
176    let existed_before_open = storage == DatabaseStorage::File && path.exists();
177    if storage == DatabaseStorage::File
178        && let Some(parent) = path.parent()
179    {
180        fs::create_dir_all(parent)
181            .map_err(|error| anyhow::anyhow!("could not create {}: {error}", parent.display()))?;
182    }
183    let _setup = match storage {
184        DatabaseStorage::File => Some(lock_database_setup(options.get_filename()).await?),
185        DatabaseStorage::InMemory => None,
186    };
187    if storage == DatabaseStorage::File {
188        create_private_database_file(options.get_filename())?;
189    }
190    options = options
191        .create_if_missing(true)
192        .foreign_keys(true)
193        .busy_timeout(Duration::from_secs(5));
194    let pool_options = match storage {
195        DatabaseStorage::InMemory => SqlitePoolOptions::new()
196            .min_connections(1)
197            .max_connections(1)
198            .idle_timeout(None)
199            .max_lifetime(None),
200        DatabaseStorage::File => {
201            options = options
202                .journal_mode(SqliteJournalMode::Wal)
203                .pragma("journal_size_limit", WAL_JOURNAL_SIZE_LIMIT.to_string());
204            SqlitePoolOptions::new().max_connections(FILE_DATABASE_CONNECTIONS)
205        }
206    };
207    let pool = pool_options
208        .connect_with(options)
209        .await
210        .map_err(|error| anyhow::anyhow!("could not open {}: {error}", path.display()))?;
211    backup::backup_before_pending_migrations(path, existed_before_open, &pool).await?;
212    MIGRATOR.run(&pool).await?;
213    initialize_meta(&pool).await?;
214    let mut conn = pool.acquire().await?;
215    crate::sync::protocol::replica_protocol(&mut conn).await?;
216    ensure_default_workspace(&mut conn).await?;
217    let mut tx = begin_immediate(&mut conn).await?;
218    crate::epic_membership::recover(&mut tx, false).await?;
219    tx.commit().await?;
220    Ok(pool)
221}
222
223/// Serializes opening one database file across connections and processes.
224/// Enabling WAL and running migrations are not safe to run concurrently.
225async fn lock_database_setup(path: &Path) -> Result<fs::File> {
226    let mut name = path
227        .file_name()
228        .context("database path has no file name")?
229        .to_os_string();
230    name.push(".aven-open.lock");
231    let lock_path = path.with_file_name(name);
232    tokio::task::spawn_blocking(move || {
233        let lock = crate::private_fs::open_lock_file(&lock_path)
234            .map_err(|error| anyhow::anyhow!("could not open {}: {error}", lock_path.display()))?;
235        lock.lock()
236            .map_err(|error| anyhow::anyhow!("could not lock {}: {error}", lock_path.display()))?;
237        Ok(lock)
238    })
239    .await?
240}
241
242/// New databases are owner-only; SQLite gives WAL, SHM, and journal files the
243/// main database's mode. An existing database keeps its user-managed mode.
244fn create_private_database_file(path: &Path) -> Result<()> {
245    match crate::private_fs::create_new_file(path) {
246        Ok(_) => Ok(()),
247        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(()),
248        Err(error) => Err(anyhow::anyhow!(
249            "could not create {}: {error}",
250            path.display()
251        )),
252    }
253}
254
255#[derive(Clone, Copy, Debug, Eq, PartialEq)]
256enum DatabaseStorage {
257    InMemory,
258    File,
259}
260
261fn database_storage(connection_input: &str, options: &SqliteConnectOptions) -> DatabaseStorage {
262    let connection_input = connection_input
263        .trim_start_matches("sqlite://")
264        .trim_start_matches("sqlite:");
265    let mut database_and_params = connection_input.splitn(2, '?');
266    let database = database_and_params.next().unwrap_or_default();
267    let uses_memory_mode = database_and_params.next().is_some_and(|params| {
268        url::form_urlencoded::parse(params.as_bytes())
269            .any(|(key, value)| key == "mode" && value == "memory")
270    });
271    let filename = options.get_filename();
272
273    if database == ":memory:"
274        || database == "file::memory:"
275        || filename == Path::new(":memory:")
276        || filename == Path::new("file::memory:")
277        || uses_memory_mode
278    {
279        DatabaseStorage::InMemory
280    } else {
281        DatabaseStorage::File
282    }
283}
284
285async fn initialize_meta(pool: &SqlitePool) -> Result<()> {
286    let mut conn = pool.acquire().await?;
287    insert_meta_if_missing(&mut conn, "client_id", &new_id()).await?;
288    insert_meta_if_missing(&mut conn, "sync_cursor", "0").await?;
289    insert_meta_if_missing(&mut conn, "local_seq", "0").await?;
290    insert_meta_if_missing(&mut conn, "sync_generation", "0").await?;
291    Ok(())
292}
293
294pub(crate) async fn current_schema_version(conn: &mut SqliteConnection) -> Result<i64> {
295    let version: Option<i64> = sqlx::query_scalar("SELECT MAX(version) FROM _sqlx_migrations")
296        .fetch_one(conn)
297        .await?;
298    Ok(version.unwrap_or(0))
299}
300
301pub(crate) async fn get_meta(conn: &mut SqliteConnection, key: &str) -> Result<Option<String>> {
302    Ok(
303        sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?", key)
304            .fetch_optional(&mut *conn)
305            .await?,
306    )
307}
308
309pub(crate) async fn set_meta(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
310    sqlx::query!(
311        "INSERT INTO meta(key, value) VALUES (?, ?)
312         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
313        key,
314        value,
315    )
316    .execute(&mut *conn)
317    .await?;
318    Ok(())
319}
320
321#[cfg(not(any(test, feature = "test-support")))]
322pub(crate) type WriterTransaction<'a> = Transaction<'a, Sqlite>;
323#[cfg(any(test, feature = "test-support"))]
324pub(crate) type WriterTransaction<'a> = crate::test_support::writer_timing::TimedTransaction<'a>;
325
326pub(crate) async fn begin_immediate(
327    conn: &mut SqliteConnection,
328) -> sqlx::Result<WriterTransaction<'_>> {
329    let transaction = conn.begin_with("BEGIN IMMEDIATE").await?;
330    #[cfg(any(test, feature = "test-support"))]
331    let transaction = WriterTransaction::new(transaction);
332    Ok(transaction)
333}
334
335async fn insert_meta_if_missing(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
336    sqlx::query!(
337        "INSERT OR IGNORE INTO meta(key, value) VALUES (?, ?)",
338        key,
339        value
340    )
341    .execute(&mut *conn)
342    .await?;
343    Ok(())
344}
345
346#[cfg(test)]
347mod tests;
348
349#[cfg(test)]
350mod payload_tests;
351
352/// Whether `error` came from the database engine, such as a lock timeout,
353/// rather than a decision made on stored data.
354pub fn is_storage_error(error: &anyhow::Error) -> bool {
355    error.chain().any(|cause| cause.is::<sqlx::Error>())
356}
aven_core/db.rs

aven_core/
db.rs
