matrix_sdk_sqlite/

connection.rs

// Copyright 2025 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! An implementation of `deadpool` for `rusqlite`.
//!
//! Initially, we were using `deadpool-sqlite`, that is also using `rusqlite` as
//! the SQLite interface. However, in the implementation of
//! [`deadpool::managed::Manager`], when recycling an object (i.e. an SQLite
//! connection), [a SQL query is run to detect whether the connection is still
//! alive][connection-test]. It creates performance issues:
//!
//! 1. It runs a prepared SQL query, which has a non-negligle cost. Imagine each
//!    connection is used to run on average one query; when recycled, a second
//!    query was constantly run. Even if it's a simple query, it requires to
//!    prepare a statement, to run and to query it.
//! 2. The SQL query was run in a blocking task. Indeed,
//!    `deadpool_runtime::spawn_blocking` is used (via
//!    `deadpool_sync::SyncWrapper::interact`), which includes [blocking the
//!    thread, acquiring a lock][spawn_blocking] etc. All this has more
//!    performance cost.
//!
//! Measures have shown it is a performance bottleneck for us, especially on
//! Android. Why specifically on Android and not other systems? This is still
//! unclear at the time of writing (2025-11-11), despites having spent several
//! days digging and trying to find an answer to this question.
//!
//! We have tried to use another approach to test the aliveness of the
//! connections without running queries. It has involved patching `rusqlite` to
//! add more bindings to SQLite, and patching `deadpool` itself, but without any
//! successful results.
//!
//! Finally, we have started questioning the reason of this test: why testing
//! whether the connection was still alive? After all, there is no reason a
//! connection should die in our case:
//!
//! - all connections are local,
//! - all interactions are behind [WAL], which is local only,
//! - even if for an unknown reason the connection died, using it next time
//!   would create an error… exactly what would happen when recycling the
//!   connection.
//!
//! Consequently, we have created a new implementation of `deadpool` for
//! `rusqlite` that doesn't test the aliveness of the connections when recycled.
//! We assume they are all alive.
//!
//! This implementation is, at the time of writing (2025-11-11):
//!
//! - 3.5 times faster on Android than `deadpool-sqlite`, removing the lock and
//!   thread contention entirely,
//! - 2 times faster on iOS.
//!
//! [connection-test]: https://github.com/deadpool-rs/deadpool/blob/d6f7d58756f0cc7bdd1f3d54d820c1332d67e4d5/crates/deadpool-sqlite/src/lib.rs#L80-L100
//! [spawn_blocking]: https://github.com/deadpool-rs/deadpool/blob/d6f7d58756f0cc7bdd1f3d54d820c1332d67e4d5/crates/deadpool-sync/src/lib.rs#L113-L131
//! [WAL]: https://www.sqlite.org/wal.html

use std::{convert::Infallible, path::PathBuf, sync::Arc, time::Duration};

pub use deadpool::managed::reexports::*;
use deadpool::managed::{self, Metrics, PoolConfig, RecycleError};
use deadpool_sync::SyncWrapper;
use tokio::sync::Mutex;
use tracing::{info, warn};

/// The default runtime used by `matrix-sdk-sqlite` for `deadpool`.
pub const RUNTIME: Runtime = Runtime::Tokio1;

deadpool::managed_reexports!(
    "matrix-sdk-sqlite",
    Manager,
    managed::Object<Manager>,
    rusqlite::Error,
    Infallible
);

/// Type representing a connection to SQLite from the [`Pool`].
pub type Connection = Object;

/// [`Manager`][managed::Manager] for creating and recycling SQLite
/// [`Connection`]s.
#[derive(Debug)]
pub struct Manager {
    pub(crate) database_path: PathBuf,
}

impl Manager {
    /// Creates a new [`Manager`] for a database.
    #[must_use]
    pub fn new(database_path: PathBuf) -> Self {
        Self { database_path }
    }
}

impl managed::Manager for Manager {
    type Type = SyncWrapper<rusqlite::Connection>;
    type Error = rusqlite::Error;

    async fn create(&self) -> Result<Self::Type, Self::Error> {
        let path = self.database_path.clone();
        SyncWrapper::new(RUNTIME, move || rusqlite::Connection::open(path)).await
    }

    async fn recycle(
        &self,
        conn: &mut Self::Type,
        _: &Metrics,
    ) -> managed::RecycleResult<Self::Error> {
        if conn.is_mutex_poisoned() {
            return Err(RecycleError::Message(
                "Mutex is poisoned. Connection is considered unusable.".into(),
            ));
        }

        Ok(())
    }
}

/// Gracefully close the store-owned write connection.
///
/// Callers are expected to remove the enclosing [`SqliteConnections`] from the
/// store before calling this helper so no new write acquisitions can happen
/// through the store API.
///
/// 1. Waits for any in-flight write to complete by acquiring the lock.
/// 2. Runs a WAL checkpoint (TRUNCATE) to flush pending data to the main
///    database file and release WAL locks.
/// 3. Drops the store-owned `Arc` on a blocking thread.
///
/// This is still best effort: if another cloned `Arc` or an
/// `OwnedMutexGuard<Connection>` is still alive elsewhere, the underlying
/// SQLite connection may remain alive until that handle is dropped.
pub async fn close_connection(write_connection: Arc<Mutex<Connection>>) {
    // Acquire the lock to wait for any in-flight write to complete.
    let guard = write_connection.lock().await;

    // Flush WAL and release locks while we still own the connection.
    let _ = guard
        .interact(|raw| {
            raw.execute_batch("PRAGMA locking_mode = NORMAL; PRAGMA wal_checkpoint(TRUNCATE);")
                .ok();
        })
        .await;

    drop(guard);

    // Drop the store-owned Arc on a blocking thread.
    let _ = tokio::task::spawn_blocking(move || drop(write_connection)).await;
}

/// Live database connections held by each SQLite store.
///
/// Wrapped in `Option<SqliteConnections>` guarded by a `Mutex` inside each
/// store; `Some` means the store is active, `None` means it is closed.
pub(crate) struct SqliteConnections {
    /// The pool of read connections.
    pub pool: Pool,
    /// The dedicated write connection.
    ///
    /// This lives behind `Arc<Mutex<_>>` so stores can clone the `Arc` and
    /// obtain an `OwnedMutexGuard<Connection>` without holding the outer
    /// `connections` mutex across await points.
    pub write_connection: Arc<Mutex<Connection>>,
}

/// Close a store by taking its connections out.
///
/// After this returns, any new call to `read()` or `write()` through the
/// store will fail with [`crate::error::Error::StoreClosed`] until
/// [`reopen_connections`] is called.
///
/// Idempotent: if the store is already closed this is a no-op.
pub(crate) async fn close_connections(connections: &Mutex<Option<SqliteConnections>>, label: &str) {
    let mut guard = connections.lock().await;
    let Some(conns) = guard.take() else {
        // Already closed — idempotent.
        return;
    };

    let SqliteConnections { pool, write_connection } = conns;

    // Close the pool. Idle read connections are dropped immediately;
    // in-flight reads complete and their connections are discarded (not
    // recycled) on release. New pool.get() calls return PoolError::Closed.
    pool.close();

    let status = pool.status();
    info!(
        size = status.size,
        max_size = status.max_size,
        available = status.available,
        "{label} pause: pool closed"
    );

    // Close the write connection: wait for any in-flight write to finish,
    // run a WAL checkpoint, then drop on a blocking thread.
    close_connection(write_connection).await;

    let status = pool.status();
    info!(
        size = status.size,
        max_size = status.max_size,
        available = status.available,
        "{label} pause: write connection released"
    );

    // Wait for any in-flight read connections to drain.
    // The write connection has already been released above, so
    // pool.status().size == 0 now correctly means every connection is gone
    // and no SQLite file locks are held.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while pool.status().size > 0 {
        if tokio::time::Instant::now() >= deadline {
            let status = pool.status();
            warn!(
                size = status.size,
                max_size = status.max_size,
                available = status.available,
                "Timed out waiting for SQLite pool connections to drain"
            );
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// Resume a store by rebuilding its connections.
///
/// Idempotent: if the store is already active this is a no-op.
pub(crate) async fn reopen_connections(
    connections: &Mutex<Option<SqliteConnections>>,
    db_path: PathBuf,
    pool_config: PoolConfig,
    runtime_config: crate::RuntimeConfig,
) -> crate::error::Result<()> {
    use crate::utils::SqliteAsyncConnExt as _;

    let mut guard = connections.lock().await;
    if guard.is_some() {
        // Not closed — idempotent.
        return Ok(());
    }

    // Rebuild the pool (connections are created lazily on first get()).
    let pool =
        Pool::builder(Manager::new(db_path)).config(pool_config).runtime(RUNTIME).build().map_err(
            |e| crate::error::Error::InvalidData {
                details: format!("Failed to rebuild connection pool: {e}"),
            },
        )?;

    let write_conn = pool.get().await?;
    // Re-apply runtime config (WAL mode, busy timeout, etc.)
    write_conn.apply_runtime_config(runtime_config).await?;

    *guard = Some(SqliteConnections { pool, write_connection: Arc::new(Mutex::new(write_conn)) });

    Ok(())
}