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            .with_context(|| format!("could not create {}", 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        .with_context(|| format!("could not open {}", 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            .with_context(|| format!("could not open {}", lock_path.display()))?;
235        lock.lock()
236            .with_context(|| format!("could not lock {}", 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(error).with_context(|| format!("could not create {}", path.display())),
249    }
250}
251
252#[derive(Clone, Copy, Debug, Eq, PartialEq)]
253enum DatabaseStorage {
254    InMemory,
255    File,
256}
257
258fn database_storage(connection_input: &str, options: &SqliteConnectOptions) -> DatabaseStorage {
259    let connection_input = connection_input
260        .trim_start_matches("sqlite://")
261        .trim_start_matches("sqlite:");
262    let mut database_and_params = connection_input.splitn(2, '?');
263    let database = database_and_params.next().unwrap_or_default();
264    let uses_memory_mode = database_and_params.next().is_some_and(|params| {
265        url::form_urlencoded::parse(params.as_bytes())
266            .any(|(key, value)| key == "mode" && value == "memory")
267    });
268    let filename = options.get_filename();
269
270    if database == ":memory:"
271        || database == "file::memory:"
272        || filename == Path::new(":memory:")
273        || filename == Path::new("file::memory:")
274        || uses_memory_mode
275    {
276        DatabaseStorage::InMemory
277    } else {
278        DatabaseStorage::File
279    }
280}
281
282async fn initialize_meta(pool: &SqlitePool) -> Result<()> {
283    let mut conn = pool.acquire().await?;
284    insert_meta_if_missing(&mut conn, "client_id", &new_id()).await?;
285    insert_meta_if_missing(&mut conn, "sync_cursor", "0").await?;
286    insert_meta_if_missing(&mut conn, "local_seq", "0").await?;
287    insert_meta_if_missing(&mut conn, "sync_generation", "0").await?;
288    Ok(())
289}
290
291pub(crate) async fn current_schema_version(conn: &mut SqliteConnection) -> Result<i64> {
292    let version: Option<i64> = sqlx::query_scalar("SELECT MAX(version) FROM _sqlx_migrations")
293        .fetch_one(conn)
294        .await?;
295    Ok(version.unwrap_or(0))
296}
297
298pub(crate) async fn get_meta(conn: &mut SqliteConnection, key: &str) -> Result<Option<String>> {
299    Ok(
300        sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?", key)
301            .fetch_optional(&mut *conn)
302            .await?,
303    )
304}
305
306pub(crate) async fn set_meta(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
307    sqlx::query!(
308        "INSERT INTO meta(key, value) VALUES (?, ?)
309         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
310        key,
311        value,
312    )
313    .execute(&mut *conn)
314    .await?;
315    Ok(())
316}
317
318#[cfg(not(any(test, feature = "test-support")))]
319pub(crate) type WriterTransaction<'a> = Transaction<'a, Sqlite>;
320#[cfg(any(test, feature = "test-support"))]
321pub(crate) type WriterTransaction<'a> = crate::test_support::writer_timing::TimedTransaction<'a>;
322
323pub(crate) async fn begin_immediate(
324    conn: &mut SqliteConnection,
325) -> sqlx::Result<WriterTransaction<'_>> {
326    let transaction = conn.begin_with("BEGIN IMMEDIATE").await?;
327    #[cfg(any(test, feature = "test-support"))]
328    let transaction = WriterTransaction::new(transaction);
329    Ok(transaction)
330}
331
332async fn insert_meta_if_missing(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
333    sqlx::query!(
334        "INSERT OR IGNORE INTO meta(key, value) VALUES (?, ?)",
335        key,
336        value
337    )
338    .execute(&mut *conn)
339    .await?;
340    Ok(())
341}
342
343#[cfg(test)]
344mod tests;
345
346#[cfg(test)]
347mod payload_tests;
348
349/// Whether `error` came from the database engine, such as a lock timeout,
350/// rather than a decision made on stored data.
351pub fn is_storage_error(error: &anyhow::Error) -> bool {
352    error.chain().any(|cause| cause.is::<sqlx::Error>())
353}
aven_core/db.rs

aven_core/
db.rs
