diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs index ea947ab172..55f6bb092a 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs @@ -6,7 +6,7 @@ use std::{ sync::{ - Arc, Mutex, + Arc, Mutex, PoisonError, atomic::{AtomicBool, Ordering}, }, time::{Duration, Instant}, @@ -18,10 +18,10 @@ use super::clock; use super::guard::{ExactSqlAuthority, with_exact_sql_guard}; use super::{ EXACT_SQL_TRANSACTION_IDLE_LIMIT, EXACT_SQL_TRANSACTION_LIMIT, ExactSqlAttachment, - ExactSqlCommitReceipt, ExactSqlError, ExactSqlRollbackReceipt, ExactSqlRows, ExactSqlStatement, - ExactSqlWriteAuthority, ExactSqlWriteIntent, ExecutionPolicy, MAX_EXACT_SQL_ATTACHMENTS, - SqlRequest, SqlResult, TransactionPolicy, attach_database, detach_database, - execute_query_unchecked, execute_request, read_across_wal_recovery, + ExactSqlBatchResult, ExactSqlCommitReceipt, ExactSqlError, ExactSqlRollbackReceipt, + ExactSqlRows, ExactSqlStatement, ExactSqlWriteAuthority, ExactSqlWriteIntent, ExecutionPolicy, + MAX_EXACT_SQL_ATTACHMENTS, SqlRequest, SqlResult, TransactionPolicy, attach_database, + detach_database, execute_query_unchecked, execute_request, read_across_wal_recovery, request_across_wal_recovery, sqlite_error, verify_write_authority, }; use crate::connection::ConnectionMode; @@ -51,21 +51,72 @@ pub(crate) enum WriterCommand { /// per statement. The job answers whether to commit. Job { job: WriterJob, + policy: JobPolicy, reply: async_channel::Sender>, authority: Option>, }, } -pub(crate) type WriterJob = Box) -> bool + Send>; +/// Answers whether to commit; an error rolls back and becomes the job's reply. +pub(crate) type WriterJob = + Box) -> Result + Send>; + +/// How long a write job may hold the writer. `Bounded` is one fixed lease. +/// `AuthorizedLongLease` renews the lease after every successful statement, +/// as the interactive long lease does, and needs attached write authority. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum JobPolicy { + Bounded, + AuthorizedLongLease, +} /// The statements a write job may run inside its transaction, on the writer /// thread. Each statement gets the same guard, authority check and deadline as /// a statement of an interactive transaction. pub struct ExactSqlJobTransaction<'transaction> { - transaction: &'transaction Transaction<'transaction>, + transaction: &'transaction (dyn JobConnection + 'transaction), shutdown_requested: &'transaction Arc, authority: Option<&'transaction dyn ExactSqlWriteAuthority>, - deadline: Instant, + policy: JobPolicy, + deadline: Mutex, +} + +const _: () = { + const fn shareable() {} + shareable::>(); + shareable::<&ExactSqlJobTransaction<'static>>(); +}; + +/// The job's transaction behind a trait object. A `Mutex` field would name the +/// connection's lifetime and make [`ExactSqlJobTransaction`] invariant, and +/// callers such as `WorkSqliteJobStorage` shorten its lifetime. +trait JobConnection: Sync { + fn execute( + &self, + request: SqlRequest, + shutdown_requested: &Arc, + deadline: Instant, + ) -> Result; +} + +impl JobConnection for Mutex> { + fn execute( + &self, + request: SqlRequest, + shutdown_requested: &Arc, + deadline: Instant, + ) -> Result { + let transaction = self.lock().unwrap_or_else(PoisonError::into_inner); + execute_request( + &transaction, + request, + true, + Some(Arc::clone(shutdown_requested)), + Some(deadline), + true, + None, + ) + } } impl ExactSqlJobTransaction<'_> { @@ -86,20 +137,27 @@ impl ExactSqlJobTransaction<'_> { } } + pub fn execute_batch(&self, sql: String) -> Result { + match self.dispatch(SqlRequest::ExecuteBatch(sql))? { + SqlResult::BatchExecuted(result) => Ok(result), + _ => Err(ExactSqlError::TransactionClosed), + } + } + fn dispatch(&self, request: SqlRequest) -> Result { - if clock::now() >= self.deadline { + let deadline = *self.deadline.lock().unwrap_or_else(PoisonError::into_inner); + if clock::now() >= deadline { return Err(ExactSqlError::TransactionExpired); } verify_write_authority(self.authority, request.intent())?; - execute_request( - self.transaction, - request, - true, - Some(Arc::clone(self.shutdown_requested)), - Some(self.deadline), - true, - None, - ) + let result = self + .transaction + .execute(request, self.shutdown_requested, deadline); + if result.is_ok() && self.policy == JobPolicy::AuthorizedLongLease { + *self.deadline.lock().unwrap_or_else(PoisonError::into_inner) = + clock::now() + EXACT_SQL_TRANSACTION_LIMIT; + } + result } } @@ -108,10 +166,16 @@ impl ExactSqlJobTransaction<'_> { async fn run_writer_job( connection: &mut Connection, job: WriterJob, + policy: JobPolicy, reply: &async_channel::Sender>, authority: Option<&dyn ExactSqlWriteAuthority>, shutdown_requested: &Arc, ) -> Result<(), ExactSqlError> { + if policy == JobPolicy::AuthorizedLongLease && authority.is_none() { + return Err(ExactSqlError::AuthorityDenied( + "long-lease job requires attached write authority".to_owned(), + )); + } verify_write_authority(authority, ExactSqlWriteIntent::BeginTransaction)?; let before = connection.total_changes(); let transaction = begin_transaction_with_busy_retry( @@ -122,24 +186,37 @@ async fn run_writer_job( ) .await .map_err(|error| sqlite_error("begin exact SQL job", error))?; + // SAFETY: `TaskTransaction::new`'s contract holds. The transaction + // reborrows `run_writer_job`'s `&mut Connection`, so nothing else reaches + // the connection. The job may use it from any thread, but only through + // this mutex. + let transaction = Mutex::new(unsafe { TaskTransaction::new(transaction) }); let _span = tracing::trace_span!("rusqlite.exact_sql.job").entered(); - let commit = job(&ExactSqlJobTransaction { + let verdict = job(&ExactSqlJobTransaction { transaction: &transaction, shutdown_requested, authority, - deadline: clock::now() + EXACT_SQL_TRANSACTION_LIMIT, + policy, + deadline: Mutex::new(clock::now() + EXACT_SQL_TRANSACTION_LIMIT), }); + let transaction = transaction + .into_inner() + .unwrap_or_else(PoisonError::into_inner) + .into_inner(); // An interrupted write statement already rolled the transaction back // (see `discard_transaction`), so there is nothing left to commit. - if !commit || transaction.is_autocommit() || shutdown_requested.load(Ordering::Acquire) { + if !matches!(verdict, Ok(true)) + || transaction.is_autocommit() + || shutdown_requested.load(Ordering::Acquire) + { let rollback = discard_transaction(transaction, before); if rollback.is_err() { shutdown_requested.store(true, Ordering::Release); } - return match (commit, rollback) { - (_, Err(error)) => Err(error), - (false, Ok(_)) => Ok(()), - (true, Ok(_)) => Err(ExactSqlError::TransactionClosed), + return match (verdict, rollback) { + (_, Err(error)) | (Err(error), Ok(_)) => Err(error), + (Ok(false), Ok(_)) => Ok(()), + (Ok(true), Ok(_)) => Err(ExactSqlError::TransactionClosed), }; } verify_write_authority(authority, ExactSqlWriteIntent::Commit)?; @@ -506,12 +583,14 @@ pub(crate) async fn run_writer_command( } WriterCommand::Job { job, + policy, reply, authority, } => { let result = run_writer_job( connection, job, + policy, &reply, authority.as_deref(), shutdown_requested, diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs index de3be689ac..c048525b3f 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs @@ -55,7 +55,7 @@ pub(crate) use column::{ pub(crate) use json::{decode as decode_json, encode as encode_json}; pub use types::*; -pub use command::ExactSqlJobTransaction; +pub use command::{ExactSqlJobTransaction, JobPolicy}; pub(crate) use command::{WriterCommand, reject_writer_command, run_writer_command}; use command::{TransactionCommand, TransactionLeaseState}; @@ -534,20 +534,74 @@ impl ExactSqlHandle { F: Fn(&ExactSqlJobTransaction<'_>) -> Result + Send + Sync + 'static, R: Send + 'static, E: Send + 'static, + { + self.transact_with_policy(JobPolicy::Bounded, job).await + } + + /// [`Self::transact`] under `policy`. + pub async fn transact_with_policy( + &self, + policy: JobPolicy, + job: F, + ) -> Result, ExactSqlError> + where + F: Fn(&ExactSqlJobTransaction<'_>) -> Result + Send + Sync + 'static, + R: Send + 'static, + E: Send + 'static, + { + self.run_job(policy, move |transaction| Ok(job(transaction))) + .await + } + + /// [`Self::transact_with_policy`] for an async body, so an async helper + /// over the job's statements runs unchanged. The body is polled once + /// inside the job: its statements are synchronous, so it completes on that + /// poll unless it awaited something else, which would hold the writer + /// across the wait. Such a job rolls back with + /// [`ExactSqlError::JobAwaitedForeignFuture`]. + pub async fn transact_async( + &self, + policy: JobPolicy, + body: F, + ) -> Result, ExactSqlError> + where + F: AsyncFn(&ExactSqlJobTransaction<'_>) -> Result + Send + Sync + 'static, + R: Send + 'static, + E: Send + 'static, + { + self.run_job(policy, move |transaction| { + complete_now(body(transaction)).ok_or(ExactSqlError::JobAwaitedForeignFuture) + }) + .await + } + + async fn run_job( + &self, + policy: JobPolicy, + job: F, + ) -> Result, ExactSqlError> + where + F: Fn(&ExactSqlJobTransaction<'_>) -> Result, ExactSqlError> + + Send + + Sync + + 'static, + R: Send + 'static, + E: Send + 'static, { let (outcome, verdict) = async_channel::bounded(1); let (reply, response) = async_channel::bounded(1); let job: command::WriterJob = Box::new(move |transaction| { - let result = job(transaction); + let result = job(transaction)?; let commit = result.is_ok(); let _ = outcome.try_send(result); - commit + Ok(commit) }); self.writer .as_ref() .ok_or(ExactSqlError::WriterUnavailable)? .try_send(WriterCommand::Job { job, + policy, reply, authority: self.write_authority.clone(), }) @@ -1296,6 +1350,15 @@ fn verify_write_authority( } } +/// One poll; `None` means the future is pending. +fn complete_now(future: F) -> Option { + match std::pin::pin!(future).poll(&mut std::task::Context::from_waker(std::task::Waker::noop())) + { + std::task::Poll::Ready(output) => Some(output), + std::task::Poll::Pending => None, + } +} + fn validate_request(request: &SqlRequest) -> Result<(), ExactSqlError> { match request { SqlRequest::Validate(statement) diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs index 7874ec61ec..2decbe4769 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs @@ -760,6 +760,95 @@ fn a_write_job_commits_on_ok_and_rolls_back_on_err() { assert_eq!(job_count(&handle), 1, "an accepted job commits its row"); } +fn insert_job_row(value: i64) -> ExactSqlStatement { + statement( + "INSERT INTO job_rows (value) VALUES (?1)", + vec![ExactSqlValue::Integer(value)], + ) +} + +#[test] +fn a_job_body_that_awaits_a_foreign_future_is_refused_and_rolled_back() { + let (_fixture, handle) = job_fixture(); + + let refused = crate::native_runtime::block_on(handle.transact_async( + JobPolicy::Bounded, + async |transaction| { + transaction + .execute(insert_job_row(1)) + .map_err(|error| error.to_string())?; + crate::storage_pool::sleep(Duration::from_millis(1)).await; + Ok::<_, String>(()) + }, + )) + .unwrap(); + + assert_eq!(refused, Err(ExactSqlError::JobAwaitedForeignFuture)); + assert_eq!(job_count(&handle), 0, "the refused job's row rolled back"); +} + +#[test] +fn an_async_job_body_over_several_statements_commits_and_returns_its_value() { + let (_fixture, handle) = job_fixture(); + async fn insert( + transaction: &ExactSqlJobTransaction<'_>, + value: i64, + ) -> Result { + transaction + .execute(insert_job_row(value)) + .map(|result| result.changed_rows) + } + + let total = crate::native_runtime::block_on(handle.transact_async( + JobPolicy::Bounded, + async |transaction| { + insert(transaction, 1).await?; + insert(transaction, 2).await?; + transaction.execute_batch("INSERT INTO job_rows (value) VALUES (3);".to_owned())?; + let sum = transaction.query(statement("SELECT SUM(value) FROM job_rows", vec![]))?; + Ok::<_, ExactSqlError>(sum.rows[0].values[0].clone()) + }, + )) + .unwrap() + .unwrap(); + + assert_eq!(total, Ok(ExactSqlValue::Integer(6))); + assert_eq!( + job_count(&handle), + 3, + "every statement of the body committed" + ); +} + +#[test] +fn a_long_lease_job_renews_its_lease_after_each_statement() { + let fixture = fixture('a', 'a'); + let clock = FakeClock::frozen(); + let handle = ExactSqlHandle::attach(&fixture.writer, &fixture.readers) + .unwrap() + .with_write_authority(Arc::new(FakeClockAuthority { + clock: Arc::clone(&clock), + advance: Some(( + ExactSqlWriteIntent::Query, + EXACT_SQL_TRANSACTION_LIMIT * 2 / 3, + )), + })) + .unwrap(); + let three_reads = |policy| { + crate::native_runtime::block_on(handle.transact_with_policy(policy, |transaction| { + (0..3).try_for_each(|_| transaction.query(statement("SELECT 1", vec![])).map(|_| ())) + })) + .unwrap() + .unwrap() + }; + + assert_eq!(three_reads(JobPolicy::AuthorizedLongLease), Ok(())); + assert!( + three_reads(JobPolicy::Bounded).is_err(), + "a bounded job outlives its one lease" + ); +} + #[test] fn abandoned_commit_retains_completion_until_writer_settlement() { use std::sync::atomic::{AtomicUsize, Ordering}; diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/types.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/types.rs index 287fbb5229..873d9a14fd 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/types.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/types.rs @@ -357,6 +357,9 @@ pub enum ExactSqlError { ReaderUnavailable(String), TransactionClosed, TransactionExpired, + /// A job body awaited something other than its own statements, so it was + /// rolled back rather than hold the writer across the wait. + JobAwaitedForeignFuture, Sqlite { operation: &'static str, code: Option, @@ -403,6 +406,9 @@ impl fmt::Display for ExactSqlError { formatter.write_str("exact SQL transaction is already closed") } Self::TransactionExpired => formatter.write_str("exact SQL transaction lease expired"), + Self::JobAwaitedForeignFuture => { + formatter.write_str("exact SQL job awaited a future other than its own statements") + } Self::Sqlite { operation, message, .. } => { diff --git a/crates/tracedecay-rusqlite-runtime/src/storage_pool.rs b/crates/tracedecay-rusqlite-runtime/src/storage_pool.rs index c90ef46fb7..21953c40ec 100644 --- a/crates/tracedecay-rusqlite-runtime/src/storage_pool.rs +++ b/crates/tracedecay-rusqlite-runtime/src/storage_pool.rs @@ -214,21 +214,22 @@ pub(crate) fn wait_until(future: F, deadline: Option) -> Opt } /// A `rusqlite::Transaction` is `!Send` only because it holds a -/// `&Connection`, and `Connection` is `!Sync`. A task is polled by one thread -/// at a time, so a transaction on a connection only that task can reach is -/// never used from two threads at once wherever the task moves. +/// `&Connection`, and `Connection` is `!Sync`. A transaction begun through a +/// `&mut Connection` is the only path to that connection while it lives, so +/// whichever thread holds it uses the connection alone. A write job shares +/// one behind a `Mutex`, which keeps that true across threads. pub(crate) struct TaskTransaction<'connection>(rusqlite::Transaction<'connection>); -// SAFETY: [`TaskTransaction::new`]'s contract keeps every other path to the -// borrowed connection inside the task that holds the transaction. +// SAFETY: [`TaskTransaction::new`]'s contract makes the transaction the only +// path to its connection, and `Connection` is `Send`. unsafe impl Send for TaskTransaction<'_> {} impl<'connection> TaskTransaction<'connection> { /// # Safety /// - /// `transaction` must borrow a connection that the calling task owns or - /// holds by `&mut`, and no other `&Connection` to it may outlive this - /// value outside that task. + /// `transaction` must borrow a connection that the caller owns or holds + /// by `&mut`, and nothing else may reach that connection while this value + /// lives. pub(crate) unsafe fn new(transaction: rusqlite::Transaction<'connection>) -> Self { Self(transaction) } diff --git a/docs/plans/storage/README.md b/docs/plans/storage/README.md index 6abb030701..8256d97570 100644 --- a/docs/plans/storage/README.md +++ b/docs/plans/storage/README.md @@ -269,7 +269,7 @@ in. validates, builds a job over owned inputs, `try_send`s it to the file's lane (full means `Busy`), and awaits a oneshot. No storage API is sync, so nothing parks a native worker, a Tokio worker or a blocking-pool thread on SQLite. -- Jobs are re-runnable: `Fn(&mut WriteTx) -> Result` and `Fn(&ReadTx) -> +- Jobs are re-runnable: `Fn(&WriteTx) -> Result` and `Fn(&ReadTx) -> Result` over owned or shared inputs, with no effect outside their transaction. That is what lets the executor run a job again after a fence interrupt or a reconnect. Batched writes take one savepoint per job, so a job's ordinary error @@ -731,6 +731,26 @@ changes a file that holds memory. each domain keeps its own idempotency. - Reads use `query_async` and preview/apply use `on_blocking_pool`. Both are blocking-pool hops that step 7 replaces. +- Step 5, job transactions (2026-10-18, #4244). + The interactive transactions that remain become writer jobs without rewriting + the async helpers that run inside them. `ExactSqlHandle::transact_async` takes an + async body and polls it once, with a no-op waker, inside the job. The body's + statements are synchronous, so a body that awaits only them completes on that + poll. A body that awaits anything else returns `Pending`, and the job rolls back + with `ExactSqlError::JobAwaitedForeignFuture` instead of holding the writer + across the wait. That is section 4's "a write is one job, which cannot `.await`" + as a test. The job transaction is now `Sync`, so `&ExactSqlJobTransaction` crosses + the `Send` boundaries the helpers already require. It holds the writer task's + transaction in a `Mutex` behind a trait object, which keeps the type covariant + and adds no `unsafe impl`. `JobPolicy::AuthorizedLongLease` with + `transact_with_policy` renews the lease after each successful statement, as the + interactive long lease does, and `execute_batch` runs a batch inside the job. + Existing `transact` callers are unchanged. Next, in order: runtime-core `WriteTx` + over the executor traits with `Database::transact` and `transact_bulk`, then the + call sites one crate per PR (runtime-core, global-db in five slices, sessions, + session-memory, schema stages and invariants, `remote/*`, graph publication), + then deletion of the interactive APIs in runtime-core and global-db, and last in + rusqlite-runtime. - #4138 (2026-10-16). Shutdown store close waits up to 3 s for leases to drain before deciding a store's WAL cannot be truncated. Under load, owners joined before store close could still be unwinding the tasks that held their leases,