refactor(storage): run store writers and readers on one fixed pool

crates/tracedecay-rusqlite-runtime/src/checkpoint/types.rs

@@ -116,10 +116,10 @@ pub enum CheckpointBlocker {
 pub(crate) trait CheckpointBlockerSource: Send + Sync {
     fn checkpoint_blockers(&self) -> CheckpointBlockers;
 
-    /// Blocks until snapshots whose owners had already released them when
+    /// Completes once snapshots whose owners had already released them when
     /// this call began have rolled back, so a requested checkpoint never
     /// reports a reader that no caller still holds as pinning the WAL.
-    fn await_released_snapshots(&self);
+    fn await_released_snapshots(&self) -> futures_util::future::BoxFuture<'static, ()>;
 
     /// Ends every snapshot that has been open at least `age`: its worker
     /// interrupts any running statement and rolls the read back, and the

crates/tracedecay-rusqlite-runtime/src/exact_sql/clock.rs

@@ -85,4 +85,13 @@ pub(super) mod fake {
             read
         }
     }
+
+    #[derive(Default)]
+    pub(crate) struct TaskClock(Option<Arc<FakeClock>>);
+
+    impl TaskClock {
+        pub(crate) fn swap(&mut self) {
+            INSTALLED.with(|installed| std::mem::swap(&mut *installed.borrow_mut(), &mut self.0));
+        }
+    }
 }

crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs

@@ -5,7 +5,7 @@
 
 use std::{
     collections::BTreeMap,
-    sync::{Arc, Mutex, atomic::AtomicBool, mpsc},
+    sync::{Arc, Mutex, atomic::AtomicBool},
     time::{Duration, Instant},
 };
 
@@ -40,6 +40,8 @@ const ROW_ALLOCATION_OVERHEAD: usize =
 const CELL_ALLOCATION_OVERHEAD: usize = std::mem::size_of::<ExactSqlValue>();
 
 mod clock;
+#[cfg(test)]
+pub(crate) use clock::fake::TaskClock;
 mod column;
 mod command;
 mod guard;
@@ -119,7 +121,7 @@ pub struct ExactSqlHandle {
 fn recv_writer_reply<T>(
     response: async_channel::Receiver<T>,
 ) -> Result<T, async_channel::RecvError> {
-    if asupersync::runtime::Runtime::current_handle().is_some() {
+    if crate::native_runtime::on_native_thread() {
         NATIVE_THREAD_WRITER_WAITS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         if record_native_thread_wait("writer", std::panic::Location::caller()) {
             tracing::warn!(
@@ -132,6 +134,13 @@ fn recv_writer_reply<T>(
     response.recv_blocking()
 }
 
+fn refuse_on_pool() -> Result<(), ExactSqlError> {
+    if crate::storage_pool::on_pool() {
+        return Err(ExactSqlError::WriterUnavailable);
+    }
+    Ok(())
+}
+
 static NATIVE_THREAD_WRITER_WAITS: std::sync::atomic::AtomicU64 =
     std::sync::atomic::AtomicU64::new(0);
 
@@ -192,7 +201,7 @@ pub fn native_thread_writer_waits() -> u64 {
 /// `recv_writer_reply` does for writer waits.
 #[track_caller]
 pub(crate) fn note_native_thread_reader_wait() {
-    if asupersync::runtime::Runtime::current_handle().is_some() {
+    if crate::native_runtime::on_native_thread() {
         NATIVE_THREAD_READER_WAITS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         if record_native_thread_wait("reader", std::panic::Location::caller()) {
             tracing::warn!(
@@ -570,6 +579,7 @@ impl ExactSqlHandle {
 
     #[track_caller]
     pub fn checkpoint_wal_truncate(&self) -> Result<ExactSqlRows, ExactSqlError> {
+        refuse_on_pool()?;
         recv_writer_reply(self.enqueue_checkpoint_wal_truncate()?)
             .map_err(|_| ExactSqlError::WriterUnavailable)?
     }
@@ -786,7 +796,7 @@ impl ExactSqlHandle {
         ),
         ExactSqlError,
     > {
-        let (commands, receiver) = mpsc::sync_channel(1);
+        let (commands, receiver) = async_channel::bounded(1);
         let (reply, response) = async_channel::bounded(1);
         let lease = Arc::new(TransactionLeaseState::default());
         self.writer
@@ -816,6 +826,7 @@ impl ExactSqlHandle {
         behavior: TransactionBehavior,
         policy: TransactionPolicy,
     ) -> Result<ExactSqlTransaction, ExactSqlError> {
+        refuse_on_pool()?;
         let (transaction, response) = self.enqueue_transaction(behavior, policy)?;
         recv_writer_reply(response).map_err(|_| ExactSqlError::WriterUnavailable)??;
         Ok(transaction)
@@ -860,6 +871,7 @@ impl ExactSqlHandle {
 
     #[track_caller]
     fn dispatch_writer(&self, request: SqlRequest) -> Result<SqlResult, ExactSqlError> {
+        refuse_on_pool()?;
         {
             let _span = tracing::trace_span!("rusqlite.exact_sql.dispatch").entered();
             {
@@ -914,7 +926,7 @@ impl ExactSqlReadSnapshot {
 }
 
 pub struct ExactSqlTransaction {
-    commands: Option<mpsc::SyncSender<TransactionCommand>>,
+    commands: Option<async_channel::Sender<TransactionCommand>>,
     lease: Arc<TransactionLeaseState>,
     policy: TransactionPolicy,
 }
@@ -930,6 +942,7 @@ enum RollbackDispatch {
 impl ExactSqlTransaction {
     #[track_caller]
     pub fn attach_database(&self, attachment: ExactSqlAttachment) -> Result<(), ExactSqlError> {
+        refuse_on_pool()?;
         let sender = self
             .commands
             .as_ref()
@@ -1096,6 +1109,7 @@ impl ExactSqlTransaction {
     }
     #[track_caller]
     pub fn commit(self) -> Result<ExactSqlCommitReceipt, ExactSqlError> {
+        refuse_on_pool()?;
         let lease = Arc::clone(&self.lease);
         recv_writer_reply(self.enqueue_commit()?).map_err(|_| transaction_terminal_error(&lease))?
     }
@@ -1139,6 +1153,7 @@ impl ExactSqlTransaction {
 
     #[track_caller]
     pub fn rollback(self) -> Result<ExactSqlRollbackReceipt, ExactSqlError> {
+        refuse_on_pool()?;
         match self.begin_rollback() {
             RollbackDispatch::Awaiting { lease, response } => recv_writer_reply(response)
                 .unwrap_or_else(|_| settled_rollback_or_terminal_error(&lease)),
@@ -1195,6 +1210,7 @@ impl ExactSqlTransaction {
         request: SqlRequest,
         execution_policy: ExecutionPolicy,
     ) -> Result<SqlResult, ExactSqlError> {
+        refuse_on_pool()?;
         recv_writer_reply(self.enqueue_transaction_request(request, execution_policy)?)
             .map_err(|_| transaction_terminal_error(&self.lease))?
     }
@@ -1423,37 +1439,50 @@ pub(crate) fn execute_query(
     )
 }
 
-/// Prepares a read statement, riding out the short window in which a
-/// freshly opened WAL database is still being recovered by its writer.
-///
-/// SQLite answers a reader that arrives during WAL recovery with
-/// `SQLITE_BUSY_RECOVERY`, which bypasses the connection's busy handler, so
-/// a reader started right after its writer could fail its very first
-/// statement with "database is locked" under load. The window is bounded by
-/// recovery itself; retry a few times before surfacing the error.
+const PREPARE_QUERY: &str = "prepare query";
+
 fn prepare_read_statement<'c>(
     connection: &'c Connection,
     sql: &str,
 ) -> Result<rusqlite::CachedStatement<'c>, ExactSqlError> {
-    const ATTEMPTS: u32 = 8;
-    const BACKOFF: std::time::Duration = std::time::Duration::from_millis(25);
-    let mut attempt = 0;
-    loop {
-        match guard::prepare_cached(connection, sql) {
-            Ok(statement) => return Ok(statement),
-            Err(rusqlite::Error::SqliteFailure(failure, _))
-                if attempt < ATTEMPTS
-                    && matches!(
-                        failure.code,
-                        rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
-                    ) =>
-            {
-                attempt += 1;
-                std::thread::sleep(BACKOFF);
-            }
-            Err(error) => return Err(sqlite_error("prepare query", error)),
+    guard::prepare_cached(connection, sql).map_err(|error| sqlite_error(PREPARE_QUERY, error))
+}
+
+/// SQLite answers a connection that starts a read while another connection
+/// recovers the WAL with `SQLITE_BUSY_RECOVERY`, which bypasses the busy
+/// handler, so a reader started right after its writer, or a writer whose
+/// file another process just opened, could fail a statement with "database
+/// is locked" under load. The window is bounded by recovery itself; retry a
+/// few times before surfacing the error.
+pub(crate) async fn read_across_wal_recovery<T>(
+    mut read: impl FnMut() -> Result<T, ExactSqlError>,
+) -> Result<T, ExactSqlError> {
+    const RETRIES: u32 = 8;
+    const BACKOFF: Duration = Duration::from_millis(25);
+    for _ in 0..RETRIES {
+        match read() {
+            Err(ExactSqlError::Sqlite {
+                operation: PREPARE_QUERY,
+                code: Some(rusqlite::ffi::SQLITE_BUSY | rusqlite::ffi::SQLITE_LOCKED),
+                ..
+            }) => crate::storage_pool::sleep(BACKOFF).await,
+            done => return done,
         }
     }
+    read()
+}
+
+async fn request_across_wal_recovery<T>(
+    request: SqlRequest,
+    mut run: impl FnMut(SqlRequest) -> Result<T, ExactSqlError>,
+) -> Result<T, ExactSqlError> {
+    if !matches!(
+        request,
+        SqlRequest::Query(_) | SqlRequest::ExecuteReturning(_)
+    ) {
+        return run(request);
+    }
+    read_across_wal_recovery(|| run(request.clone())).await
 }
 
 fn execute_query_unchecked(
@@ -1588,12 +1617,12 @@ fn settled_rollback_or_terminal_error(
 }
 
 fn map_transaction_send_error(
-    error: mpsc::TrySendError<TransactionCommand>,
+    error: async_channel::TrySendError<TransactionCommand>,
     lease: &TransactionLeaseState,
 ) -> ExactSqlError {
     match error {
-        mpsc::TrySendError::Full(_) => ExactSqlError::Busy,
-        mpsc::TrySendError::Disconnected(_) => transaction_terminal_error(lease),
+        async_channel::TrySendError::Full(_) => ExactSqlError::Busy,
+        async_channel::TrySendError::Closed(_) => transaction_terminal_error(lease),
     }
 }
 

crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/prepare_cache.rs

@@ -172,3 +172,58 @@ fn execute_query_unchecked_reuses_the_cached_statement() {
     }
     assert_eq!(schema_version(&connection), version);
 }
+
+#[test]
+fn a_writer_query_rides_out_a_file_another_connection_briefly_locks() {
+    let directory = TempDir::new().unwrap();
+    let path = directory.path().join("locked.sqlite3");
+    Connection::open(&path)
+        .unwrap()
+        .execute_batch("PRAGMA journal_mode=WAL; CREATE TABLE t(x); INSERT INTO t VALUES (1);")
+        .unwrap();
+    let holder = Connection::open(&path).unwrap();
+    holder
+        .execute_batch("PRAGMA locking_mode=EXCLUSIVE")
+        .unwrap();
+    holder
+        .query_row("SELECT count(*) FROM t", [], |row| row.get::<_, i64>(0))
+        .unwrap();
+    let mut writer = Connection::open(&path).unwrap();
+    writer.busy_timeout(Duration::ZERO).unwrap();
+    assert!(
+        writer.prepare("SELECT x FROM t").is_err(),
+        "the holder locks the file against the writer's prepare"
+    );
+    let (reply, answer) = async_channel::bounded(1);
+    let (report, events) = std::sync::mpsc::channel();
+    crate::storage_pool::spawn(async move {
+        let _ = report.send("started");
+        let shutdown_requested = Arc::new(AtomicBool::new(false));
+        command::run_writer_command(
+            &mut writer,
+            command::WriterCommand::Dispatch {
+                request: SqlRequest::Query(statement("SELECT x FROM t", vec![])),
+                reply,
+                authority: None,
+            },
+            &shutdown_requested,
+        )
+        .await
+        .send();
+        let _ = report.send("answered");
+    })
+    .unwrap_or_else(|_| panic!("the storage pool must start"));
+    assert_eq!(events.recv_timeout(Duration::from_secs(10)), Ok("started"));
+    std::thread::sleep(Duration::from_millis(60));
+    drop(holder);
+    assert_eq!(events.recv_timeout(Duration::from_secs(10)), Ok("answered"));
+    match answer.try_recv().unwrap() {
+        Ok(SqlResult::Queried(rows)) => assert_eq!(
+            rows.rows,
+            vec![ExactSqlRow {
+                values: vec![ExactSqlValue::Integer(1)]
+            }]
+        ),
+        other => panic!("the query must outlast the lock, got {other:?}"),
+    }
+}