refactor(storage): run handoff grants as awaited writer jobs

crates/tracedecay-contracts/src/handoff.rs

@@ -657,13 +657,14 @@ pub struct HandoffOpenListingV1 {
     pub consumed_at: Option<UtcMicros>,
 }
 
+/// Every method is awaited and never parks the caller's thread.
 pub trait HandoffOpenAuthorityPort: Send + Sync {
     /// Commits a grant or returns the byte-authoritative grant already
     /// committed for the same request identity.
     fn issue(
         &self,
         grant: &HandoffOpenGrantV1,
-    ) -> Result<HandoffOpenGrantV1, HandoffOpenAuthorityError>;
+    ) -> impl Future<Output = Result<HandoffOpenGrantV1, HandoffOpenAuthorityError>> + Send;
 
     /// Enumerates the grants matching `filter`, newest issuance first, capped
     /// at `limit`.
@@ -676,14 +677,14 @@ pub trait HandoffOpenAuthorityPort: Send + Sync {
         &self,
         filter: &HandoffOpenListFilterV1,
         limit: u32,
-    ) -> Result<Vec<HandoffOpenListingV1>, HandoffOpenAuthorityError>;
+    ) -> impl Future<Output = Result<Vec<HandoffOpenListingV1>, HandoffOpenAuthorityError>> + Send;
 
     fn resolve(
         &self,
         token_digest: &ManifestDigest,
         expected: &HandoffOpenExpectationV1,
         observed_at: UtcMicros,
-    ) -> Result<Option<HandoffOpenGrantV1>, HandoffOpenAuthorityError>;
+    ) -> impl Future<Output = Result<Option<HandoffOpenGrantV1>, HandoffOpenAuthorityError>> + Send;
 
     fn consume(
         &self,
@@ -692,7 +693,7 @@ pub trait HandoffOpenAuthorityPort: Send + Sync {
         request_id: &RequestId,
         input_digest: &ManifestDigest,
         consumed_at: UtcMicros,
-    ) -> Result<HandoffOpenConsumeOutcomeV1, HandoffOpenAuthorityError>;
+    ) -> impl Future<Output = Result<HandoffOpenConsumeOutcomeV1, HandoffOpenAuthorityError>> + Send;
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -1087,7 +1088,7 @@ where
             issued_at,
             expires_at,
         )?;
-        self.authority.issue(&grant).map_err(authority_error)
+        self.authority.issue(&grant).await.map_err(authority_error)
     }
 
     /// Enumerates the handoff tokens this caller could redeem.
@@ -1109,6 +1110,7 @@ where
         let mut listings = self
             .authority
             .list(&filter, MAX_HANDOFF_LIST_RESULTS_V1.saturating_add(1))
+            .await
             .map_err(authority_error)?;
         let more_existed = listings.len() > MAX_HANDOFF_LIST_RESULTS_V1 as usize;
         listings.truncate(MAX_HANDOFF_LIST_RESULTS_V1 as usize);
@@ -1201,6 +1203,7 @@ where
         let Some(grant) = self
             .authority
             .resolve(&token_digest, &expected, observed_at)
+            .await
             .map_err(authority_error)?
         else {
             return Err(HandoffOpenError::NotFoundOrNotAuthorized);
@@ -1233,6 +1236,7 @@ where
                 &input_digest,
                 observed_at,
             )
+            .await
             .map_err(authority_error)?
         {
             HandoffOpenConsumeOutcomeV1::Consumed(consumption) => *consumption,

crates/tracedecay-contracts/tests/contracts_suite/handoff_open.rs

@@ -33,7 +33,7 @@ struct MemoryAuthorityState {
 }
 
 impl HandoffOpenAuthorityPort for MemoryAuthority {
-    fn issue(
+    async fn issue(
         &self,
         grant: &HandoffOpenGrantV1,
     ) -> Result<HandoffOpenGrantV1, HandoffOpenAuthorityError> {
@@ -50,7 +50,7 @@ impl HandoffOpenAuthorityPort for MemoryAuthority {
         Ok(grant.clone())
     }
 
-    fn list(
+    async fn list(
         &self,
         filter: &HandoffOpenListFilterV1,
         limit: u32,
@@ -82,7 +82,7 @@ impl HandoffOpenAuthorityPort for MemoryAuthority {
         Ok(listings)
     }
 
-    fn resolve(
+    async fn resolve(
         &self,
         token_digest: &ManifestDigest,
         expected: &HandoffOpenExpectationV1,
@@ -96,7 +96,7 @@ impl HandoffOpenAuthorityPort for MemoryAuthority {
             .cloned())
     }
 
-    fn consume(
+    async fn consume(
         &self,
         token_digest: &ManifestDigest,
         expected: &HandoffOpenExpectationV1,

crates/tracedecay-daemon-service/src/invocation/handoff.rs

@@ -223,7 +223,7 @@ pub(super) async fn execute_handoff_application(
             );
         }
     };
-    let authority = match registered.database.handoff_open_storage() {
+    let authority = match registered.database.handoff_open_storage().await {
         Ok(authority) => authority,
         Err(_) => {
             return DaemonInvocationResponse::problem(

crates/tracedecay-global-db/src/registered.rs

@@ -633,12 +633,12 @@ impl RegisteredGlobalDb {
         self.database.workflow_storage()
     }
 
-    pub fn handoff_open_storage(
+    pub async fn handoff_open_storage(
         &self,
     ) -> tracedecay_domain::errors::Result<
         tracedecay_rusqlite_runtime::handoff::HandoffOpenSqliteAuthority,
     > {
-        self.database.handoff_open_storage()
+        self.database.handoff_open_storage().await
     }
 
     pub fn storage_telemetry_handle(

crates/tracedecay-runtime-core/src/db/connection/tests.rs

@@ -1083,7 +1083,7 @@ async fn handoff_open_storage_validates_mounted_schema_and_retains_one_client()
         .await
         .unwrap();
     assert!(rows.next().await.unwrap().is_some());
-    let storage = database.handoff_open_storage().unwrap();
+    let storage = database.handoff_open_storage().await.unwrap();
     assert_retained_purpose_adapter_blocks_one_client!(owner, control, database, storage);
 }
 
@@ -1316,7 +1316,7 @@ async fn read_only_owner_issuance_preserves_the_published_access_policy() {
     assert!(database.work_storage().is_err());
     assert!(database.workflow_storage().is_err());
     assert!(database.authorized_scope_set_storage().is_err());
-    assert!(database.handoff_open_storage().is_err());
+    assert!(database.handoff_open_storage().await.is_err());
     assert!(
         database
             .remote_storage(purpose_test_remote_keyring())

crates/tracedecay-runtime-core/src/db/purpose.rs

@@ -68,12 +68,13 @@ impl Database {
     }
 
     /// Owner-bound single-use handoff-open authority.
-    pub fn handoff_open_storage(
+    pub async fn handoff_open_storage(
         &self,
     ) -> Result<tracedecay_rusqlite_runtime::handoff::HandoffOpenSqliteAuthority> {
         tracedecay_rusqlite_runtime::handoff::HandoffOpenSqliteAuthority::from_retained_exact_sql(
             self.retained_exact_sql_for_purpose("attach handoff-open storage")?,
         )
+        .await
         .map_err(|error| TraceDecayError::Database {
             message: format!("failed to validate handoff-open storage schema: {error:?}"),
             operation: "attach handoff-open storage".to_owned(),

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

@@ -45,6 +45,105 @@ pub(crate) enum WriterCommand {
         reply: async_channel::Sender<Result<ExactSqlRows, ExactSqlError>>,
         authority: Option<Arc<dyn ExactSqlWriteAuthority>>,
     },
+    /// One whole write transaction, run on this thread without a round trip
+    /// per statement. The job answers whether to commit.
+    Job {
+        job: WriterJob,
+        reply: async_channel::Sender<Result<(), ExactSqlError>>,
+        authority: Option<Arc<dyn ExactSqlWriteAuthority>>,
+    },
+}
+
+pub(crate) type WriterJob = Box<dyn FnOnce(&ExactSqlJobTransaction<'_>) -> bool + Send>;
+
+/// 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>,
+    shutdown_requested: &'transaction Arc<AtomicBool>,
+    authority: Option<&'transaction dyn ExactSqlWriteAuthority>,
+    deadline: Instant,
+}
+
+impl ExactSqlJobTransaction<'_> {
+    pub fn execute(
+        &self,
+        statement: ExactSqlStatement,
+    ) -> Result<super::ExactSqlExecuteResult, ExactSqlError> {
+        match self.dispatch(SqlRequest::Execute(statement))? {
+            SqlResult::Executed(result) => Ok(result),
+            _ => Err(ExactSqlError::TransactionClosed),
+        }
+    }
+
+    pub fn query(&self, statement: ExactSqlStatement) -> Result<ExactSqlRows, ExactSqlError> {
+        match self.dispatch(SqlRequest::Query(statement))? {
+            SqlResult::Queried(rows) => Ok(rows),
+            _ => Err(ExactSqlError::TransactionClosed),
+        }
+    }
+
+    fn dispatch(&self, request: SqlRequest) -> Result<SqlResult, ExactSqlError> {
+        if clock::now() >= self.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,
+        )
+    }
+}
+
+/// Runs one write job: begin IMMEDIATE, the job's statements, then commit when
+/// the job asks for it and rollback otherwise.
+fn run_writer_job(
+    connection: &Connection,
+    job: WriterJob,
+    reply: &async_channel::Sender<Result<(), ExactSqlError>>,
+    authority: Option<&dyn ExactSqlWriteAuthority>,
+    shutdown_requested: &Arc<AtomicBool>,
+) -> Result<(), ExactSqlError> {
+    verify_write_authority(authority, ExactSqlWriteIntent::BeginTransaction)?;
+    let before = connection.total_changes();
+    let transaction = begin_transaction_with_busy_retry(
+        connection,
+        TransactionBehavior::Immediate,
+        shutdown_requested,
+        || reply.is_closed(),
+    )
+    .map_err(|error| sqlite_error("begin exact SQL job", error))?;
+    let _span = tracing::trace_span!("rusqlite.exact_sql.job").entered();
+    let commit = job(&ExactSqlJobTransaction {
+        transaction: &transaction,
+        shutdown_requested,
+        authority,
+        deadline: clock::now() + EXACT_SQL_TRANSACTION_LIMIT,
+    });
+    // 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) {
+        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),
+        };
+    }
+    verify_write_authority(authority, ExactSqlWriteIntent::Commit)?;
+    let _span = tracing::trace_span!("rusqlite.commit").entered();
+    transaction
+        .commit()
+        .map_err(|error| sqlite_error("commit exact SQL job", error))
 }
 
 /// Pause between busy-begin attempts so the acquire deadline is the real bound.
@@ -411,6 +510,20 @@ pub(crate) fn run_writer_command(
             };
             DeferredReply::new(reply, result)
         }
+        WriterCommand::Job {
+            job,
+            reply,
+            authority,
+        } => {
+            let result = run_writer_job(
+                connection,
+                job,
+                &reply,
+                authority.as_deref(),
+                shutdown_requested,
+            );
+            DeferredReply::new(reply, result)
+        }
     }
 }
 
@@ -425,6 +538,9 @@ pub(crate) fn reject_writer_command(command: WriterCommand) {
         WriterCommand::CheckpointWalTruncate { reply, .. } => {
             let _ = reply.try_send(Err(ExactSqlError::WriterUnavailable));
         }
+        WriterCommand::Job { reply, .. } => {
+            let _ = reply.try_send(Err(ExactSqlError::WriterUnavailable));
+        }
     }
 }
 

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

@@ -53,6 +53,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(crate) use command::{WriterCommand, reject_writer_command, run_writer_command};
 
 use command::{TransactionCommand, TransactionLeaseState};
@@ -488,6 +489,61 @@ impl ExactSqlHandle {
         (self.query)(statement, priority, max_wait)
     }
 
+    /// Awaited interactive read. Reader admission and the query run as one
+    /// job on the blocking pool, so the awaiting task parks no runtime thread.
+    /// The storage executor replaces the hop (docs/plans/storage, step 7).
+    pub async fn query_async(
+        &self,
+        statement: ExactSqlStatement,
+        max_wait: Duration,
+    ) -> Result<ExactSqlRows, ExactSqlError> {
+        statement.validate()?;
+        let query = Arc::clone(&self.query);
+        crate::native_runtime::spawn_blocking(move || {
+            query(statement, OperationPriorityV1::Foreground, max_wait)
+        })
+        .await
+        .map_err(ExactSqlError::ReaderUnavailable)?
+    }
+
+    /// Runs `job` as one write transaction on the writer thread: begin
+    /// IMMEDIATE, the job's statements with no round trip between them, then
+    /// commit when the job returns `Ok` and rollback when it returns `Err`.
+    /// The outer error is the transaction's own failure; the inner result is
+    /// the job's verdict. A job is `Fn` over owned inputs so the storage
+    /// executor can run it again after an interrupted read fence.
+    pub async fn transact<R, E, F>(&self, job: F) -> Result<Result<R, E>, ExactSqlError>
+    where
+        F: Fn(&ExactSqlJobTransaction<'_>) -> Result<R, E> + 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 commit = result.is_ok();
+            let _ = outcome.try_send(result);
+            commit
+        });
+        self.writer
+            .as_ref()
+            .ok_or(ExactSqlError::WriterUnavailable)?
+            .try_send(WriterCommand::Job {
+                job,
+                reply,
+                authority: self.write_authority.clone(),
+            })
+            .map_err(map_writer_send_error)?;
+        response
+            .recv()
+            .await
+            .map_err(|_| ExactSqlError::WriterUnavailable)??;
+        verdict
+            .try_recv()
+            .map_err(|_| ExactSqlError::WriterUnavailable)
+    }
+
     /// Checkpoints and truncates the WAL on the serialized writer connection.
     fn enqueue_checkpoint_wal_truncate(
         &self,