From 63117a4e71bf70764c99558426da1ec96bbb1bdb Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 11:54:04 +0000 Subject: [PATCH] refactor(storage): run handoff grants as awaited writer jobs --- crates/tracedecay-contracts/src/handoff.rs | 14 +- .../tests/contracts_suite/handoff_open.rs | 8 +- .../src/invocation/handoff.rs | 2 +- crates/tracedecay-global-db/src/registered.rs | 4 +- .../src/db/connection/tests.rs | 4 +- .../tracedecay-runtime-core/src/db/purpose.rs | 3 +- .../src/exact_sql/command.rs | 116 +++++++ .../src/exact_sql/mod.rs | 56 ++++ .../src/exact_sql/tests/async_dispatch.rs | 108 +++++++ .../src/handoff.rs | 297 ++++++++++-------- .../rusqlite_suite/handoff_open_storage.rs | 103 +++++- 11 files changed, 563 insertions(+), 152 deletions(-) diff --git a/crates/tracedecay-contracts/src/handoff.rs b/crates/tracedecay-contracts/src/handoff.rs index 6031e013a9..46c6713f01 100644 --- a/crates/tracedecay-contracts/src/handoff.rs +++ b/crates/tracedecay-contracts/src/handoff.rs @@ -657,13 +657,14 @@ pub struct HandoffOpenListingV1 { pub consumed_at: Option, } +/// 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; + ) -> impl Future> + 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, HandoffOpenAuthorityError>; + ) -> impl Future, HandoffOpenAuthorityError>> + Send; fn resolve( &self, token_digest: &ManifestDigest, expected: &HandoffOpenExpectationV1, observed_at: UtcMicros, - ) -> Result, HandoffOpenAuthorityError>; + ) -> impl Future, HandoffOpenAuthorityError>> + Send; fn consume( &self, @@ -692,7 +693,7 @@ pub trait HandoffOpenAuthorityPort: Send + Sync { request_id: &RequestId, input_digest: &ManifestDigest, consumed_at: UtcMicros, - ) -> Result; + ) -> impl Future> + 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, diff --git a/crates/tracedecay-contracts/tests/contracts_suite/handoff_open.rs b/crates/tracedecay-contracts/tests/contracts_suite/handoff_open.rs index 0bbd51e3ff..6d89920c5f 100644 --- a/crates/tracedecay-contracts/tests/contracts_suite/handoff_open.rs +++ b/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 { @@ -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, diff --git a/crates/tracedecay-daemon-service/src/invocation/handoff.rs b/crates/tracedecay-daemon-service/src/invocation/handoff.rs index b6b27d2cfc..caa859be21 100644 --- a/crates/tracedecay-daemon-service/src/invocation/handoff.rs +++ b/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( diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index 4ee8c6166c..ac0a4bb50f 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/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( diff --git a/crates/tracedecay-runtime-core/src/db/connection/tests.rs b/crates/tracedecay-runtime-core/src/db/connection/tests.rs index 34390cb102..ffc7f106ca 100644 --- a/crates/tracedecay-runtime-core/src/db/connection/tests.rs +++ b/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()) diff --git a/crates/tracedecay-runtime-core/src/db/purpose.rs b/crates/tracedecay-runtime-core/src/db/purpose.rs index 815b3c6c9e..a61443e616 100644 --- a/crates/tracedecay-runtime-core/src/db/purpose.rs +++ b/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::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(), diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs index e05089c5a0..51b1e308fd 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/command.rs @@ -45,6 +45,105 @@ pub(crate) enum WriterCommand { reply: async_channel::Sender>, authority: Option>, }, + /// 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>, + authority: Option>, + }, +} + +pub(crate) type WriterJob = Box) -> 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, + authority: Option<&'transaction dyn ExactSqlWriteAuthority>, + deadline: Instant, +} + +impl ExactSqlJobTransaction<'_> { + pub fn execute( + &self, + statement: ExactSqlStatement, + ) -> Result { + match self.dispatch(SqlRequest::Execute(statement))? { + SqlResult::Executed(result) => Ok(result), + _ => Err(ExactSqlError::TransactionClosed), + } + } + + pub fn query(&self, statement: ExactSqlStatement) -> Result { + match self.dispatch(SqlRequest::Query(statement))? { + SqlResult::Queried(rows) => Ok(rows), + _ => Err(ExactSqlError::TransactionClosed), + } + } + + fn dispatch(&self, request: SqlRequest) -> Result { + 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>, + authority: Option<&dyn ExactSqlWriteAuthority>, + shutdown_requested: &Arc, +) -> 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)); + } } } diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs index 7441cc76dc..2ed7d6d2d0 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs +++ b/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 { + 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(&self, job: F) -> Result, ExactSqlError> + where + F: Fn(&ExactSqlJobTransaction<'_>) -> Result + 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, 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 ba065846dc..b58190b5c7 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 @@ -651,3 +651,111 @@ fn repeated_reader_waits_from_one_site_are_counted_but_logged_once() { "the first hit logs at WARN and later hits log nothing" ); } + +fn job_count(handle: &ExactSqlHandle) -> i64 { + match crate::native_runtime::block_on(handle.query_async( + statement("SELECT COUNT(*) FROM job_rows", vec![]), + Duration::from_secs(5), + )) + .unwrap() + .unwrap() + .rows[0] + .values[0] + { + ExactSqlValue::Integer(count) => count, + ref other => panic!("count is an integer, got {other:?}"), + } +} + +fn job_fixture() -> (Fixture, ExactSqlHandle) { + let fixture = fixture('a', 'a'); + let handle = ExactSqlHandle::attach(&fixture.writer, &fixture.readers).unwrap(); + crate::native_runtime::block_on( + handle.execute_batch_async("CREATE TABLE job_rows (value INTEGER NOT NULL)".to_owned()), + ) + .unwrap() + .unwrap(); + (fixture, handle) +} + +#[test] +fn a_write_job_and_an_awaited_read_on_a_native_worker_park_no_native_thread() { + let _serial = NATIVE_WAIT_COUNTER + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let (_fixture, handle) = job_fixture(); + let writer_waits = native_thread_writer_waits(); + let reader_waits = native_thread_reader_waits(); + let worker_handle = handle.clone(); + let task = crate::native_runtime::NativeTask::spawn(async move { + let committed = worker_handle + .transact(|transaction| { + transaction.execute(statement( + "INSERT INTO job_rows (value) VALUES (?1)", + vec![ExactSqlValue::Integer(7)], + ))?; + transaction + .query(statement("SELECT COUNT(*) FROM job_rows", vec![])) + .map(|rows| rows.rows.len()) + }) + .await + .map_err(|error| error.to_string())? + .map_err(|error: ExactSqlError| error.to_string())?; + let read = worker_handle + .query_async( + statement("SELECT value FROM job_rows", vec![]), + Duration::from_secs(5), + ) + .await + .map_err(|error| error.to_string())?; + Ok::<_, String>((committed, read.rows.len())) + }) + .unwrap(); + let outcome = crate::native_runtime::block_on(task.wait()).unwrap(); + assert_eq!( + outcome, + crate::native_runtime::NativeTaskOutcome::Completed(Ok((1, 1))) + ); + assert_eq!( + native_thread_writer_waits() - writer_waits, + 0, + "the job ran on the writer thread without parking the worker" + ); + assert_eq!( + native_thread_reader_waits() - reader_waits, + 0, + "the awaited read ran on the blocking pool" + ); +} + +#[test] +fn a_write_job_commits_on_ok_and_rolls_back_on_err() { + let (_fixture, handle) = job_fixture(); + let insert = |value| { + statement( + "INSERT INTO job_rows (value) VALUES (?1)", + vec![ExactSqlValue::Integer(value)], + ) + }; + + let refused = crate::native_runtime::block_on(handle.transact(move |transaction| { + transaction + .execute(insert(1)) + .map_err(|error| error.to_string())?; + Err::<(), _>("refused after its write".to_owned()) + })) + .unwrap() + .unwrap(); + assert_eq!(refused, Err("refused after its write".to_owned())); + assert_eq!(job_count(&handle), 0, "a refused job leaves no row"); + + let committed = crate::native_runtime::block_on(handle.transact(move |transaction| { + transaction + .execute(insert(2)) + .map(|result| result.changed_rows) + })) + .unwrap() + .unwrap(); + assert_eq!(committed, Ok(1)); + assert_eq!(job_count(&handle), 1, "an accepted job commits its row"); +} diff --git a/crates/tracedecay-rusqlite-runtime/src/handoff.rs b/crates/tracedecay-rusqlite-runtime/src/handoff.rs index cd59f1ec38..8900d896e7 100644 --- a/crates/tracedecay-rusqlite-runtime/src/handoff.rs +++ b/crates/tracedecay-rusqlite-runtime/src/handoff.rs @@ -14,11 +14,11 @@ use tracedecay_contracts::{ use tracedecay_domain::{ManifestDigest, UtcMicros}; use crate::exact_sql::{ - ExactSqlError, ExactSqlHandle, ExactSqlRows, ExactSqlValue, decode_json, encode_json, - optional_text_at, text_column, + ExactSqlError, ExactSqlHandle, ExactSqlJobTransaction, ExactSqlRows, ExactSqlValue, + decode_json, encode_json, optional_text_at, text_column, }; use crate::repository::RetainedExactSqlCapability; -use crate::workflow::{execute_tx, query_tx, statement}; +use crate::workflow::statement; pub const HANDOFF_OPEN_SCHEMA_V1: &str = " CREATE TABLE IF NOT EXISTS handoff_open_grants_v1 ( @@ -48,11 +48,11 @@ pub struct HandoffOpenSqliteAuthority { } impl HandoffOpenSqliteAuthority { - pub fn from_retained_exact_sql( + pub async fn from_retained_exact_sql( retained: RetainedExactSqlCapability, ) -> Result { let authority = Self { retained }; - require_handoff_open_schema(authority.handle())?; + require_handoff_open_schema(authority.handle()).await?; Ok(authority) } @@ -74,15 +74,17 @@ fn codec_unavailable() -> HandoffOpenAuthorityError { HandoffOpenAuthorityError::Unavailable } -fn query_handle( +async fn query_handle( handle: &ExactSqlHandle, sql: &str, params: Vec, ) -> Result { - handle.query(statement(sql, params)?, Duration::from_secs(5)) + handle + .query_async(statement(sql, params)?, Duration::from_secs(5)) + .await } -fn require_handoff_open_schema( +async fn require_handoff_open_schema( handle: &ExactSqlHandle, ) -> Result<(), HandoffOpenSqliteAuthorityBuildError> { let rows = query_handle( @@ -90,6 +92,7 @@ fn require_handoff_open_schema( "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1", vec![ExactSqlValue::Text("handoff_open_grants_v1".to_owned())], ) + .await .map_err(|_| HandoffOpenSqliteAuthorityBuildError::Unavailable)?; if rows.rows.len() == 1 { Ok(()) @@ -99,54 +102,19 @@ fn require_handoff_open_schema( } impl HandoffOpenAuthorityPort for HandoffOpenSqliteAuthority { - fn issue( + async fn issue( &self, grant: &HandoffOpenGrantV1, ) -> Result { let payload = encode_json(grant, |_| codec_unavailable())?; - let transaction = self.handle().begin_immediate().map_err(unavailable)?; - let existing = query_tx( - &transaction, - "SELECT grant_payload FROM handoff_open_grants_v1 - WHERE token_digest = ?1 OR issued_request_id = ?2", - vec![ - ExactSqlValue::Text(grant.token_digest().as_str().to_owned()), - ExactSqlValue::Text(grant.issued_request_id().as_str().to_owned()), - ], - ) - .map_err(unavailable)?; - if let Some(row) = existing.rows.first() { - let persisted_payload = text_column(&row.values, 0).ok_or_else(codec_unavailable)?; - let persisted: HandoffOpenGrantV1 = - decode_json(persisted_payload, |_| codec_unavailable())?; - let _ = transaction.rollback(); - if persisted.same_issue_identity(grant) { - return Ok(persisted); - } - return Err(HandoffOpenAuthorityError::Conflict); - } - execute_tx( - &transaction, - "INSERT INTO handoff_open_grants_v1 ( - token_digest, issued_request_id, grant_payload, issued_at, expires_at, - consumed_request_id, consumed_input_digest, consumption_payload - ) VALUES (?1, ?2, ?3, ?4, ?5, NULL, NULL, NULL)", - vec![ - ExactSqlValue::Text(grant.token_digest().as_str().to_owned()), - ExactSqlValue::Text(grant.issued_request_id().as_str().to_owned()), - ExactSqlValue::Text(payload), - ExactSqlValue::Integer(grant.issued_at().0), - ExactSqlValue::Integer(grant.expires_at().0), - ], - ) - .map_err(unavailable)?; - transaction - .commit() - .map(|_| grant.clone()) - .map_err(unavailable) + let grant = grant.clone(); + self.handle() + .transact(move |transaction| issue_grant(transaction, &grant, &payload)) + .await + .map_err(unavailable)? } - fn list( + async fn list( &self, filter: &HandoffOpenListFilterV1, limit: u32, @@ -170,6 +138,7 @@ impl HandoffOpenAuthorityPort for HandoffOpenSqliteAuthority { LIMIT ?1", vec![ExactSqlValue::Integer(ceiling)], ) + .await .map_err(unavailable)?; let mut listings = Vec::new(); @@ -196,7 +165,7 @@ impl HandoffOpenAuthorityPort for HandoffOpenSqliteAuthority { Ok(listings) } - fn resolve( + async fn resolve( &self, token_digest: &ManifestDigest, expected: &HandoffOpenExpectationV1, @@ -207,6 +176,7 @@ impl HandoffOpenAuthorityPort for HandoffOpenSqliteAuthority { "SELECT grant_payload FROM handoff_open_grants_v1 WHERE token_digest = ?1", vec![ExactSqlValue::Text(token_digest.as_str().to_owned())], ) + .await .map_err(unavailable)?; let Some(row) = rows.rows.first() else { return Ok(None); @@ -219,7 +189,7 @@ impl HandoffOpenAuthorityPort for HandoffOpenSqliteAuthority { Ok(Some(grant)) } - fn consume( + async fn consume( &self, token_digest: &ManifestDigest, expected: &HandoffOpenExpectationV1, @@ -227,82 +197,163 @@ impl HandoffOpenAuthorityPort for HandoffOpenSqliteAuthority { input_digest: &ManifestDigest, consumed_at: UtcMicros, ) -> Result { - let transaction = self.handle().begin_immediate().map_err(unavailable)?; - let rows = query_tx( - &transaction, - "SELECT grant_payload, consumed_request_id, consumed_input_digest, - consumption_payload - FROM handoff_open_grants_v1 - WHERE token_digest = ?1", - vec![ExactSqlValue::Text(token_digest.as_str().to_owned())], - ) - .map_err(unavailable)?; - let Some(row) = rows.rows.first() else { - let _ = transaction.rollback(); - return Ok(HandoffOpenConsumeOutcomeV1::Concealed); + let consumption = ConsumeRequest { + token_digest: token_digest.clone(), + expected: expected.clone(), + request_id: request_id.clone(), + input_digest: input_digest.clone(), + consumed_at, }; - let grant_payload = text_column(&row.values, 0).ok_or_else(codec_unavailable)?; - let grant: HandoffOpenGrantV1 = decode_json(grant_payload, |_| codec_unavailable())?; - if !expected.matches(grant.context()) || consumed_at >= *grant.expires_at() { - let _ = transaction.rollback(); - return Ok(HandoffOpenConsumeOutcomeV1::Concealed); + self.handle() + .transact(move |transaction| consume_grant(transaction, &consumption)) + .await + .map_err(unavailable)? + } +} + +fn job_query( + transaction: &ExactSqlJobTransaction<'_>, + sql: &str, + params: Vec, +) -> Result { + transaction + .query(statement(sql, params).map_err(unavailable)?) + .map_err(unavailable) +} + +fn job_execute( + transaction: &ExactSqlJobTransaction<'_>, + sql: &str, + params: Vec, +) -> Result<(), HandoffOpenAuthorityError> { + transaction + .execute(statement(sql, params).map_err(unavailable)?) + .map(|_| ()) + .map_err(unavailable) +} + +fn issue_grant( + transaction: &ExactSqlJobTransaction<'_>, + grant: &HandoffOpenGrantV1, + payload: &str, +) -> Result { + let existing = job_query( + transaction, + "SELECT grant_payload FROM handoff_open_grants_v1 + WHERE token_digest = ?1 OR issued_request_id = ?2", + vec![ + ExactSqlValue::Text(grant.token_digest().as_str().to_owned()), + ExactSqlValue::Text(grant.issued_request_id().as_str().to_owned()), + ], + )?; + if let Some(row) = existing.rows.first() { + let persisted_payload = text_column(&row.values, 0).ok_or_else(codec_unavailable)?; + let persisted: HandoffOpenGrantV1 = + decode_json(persisted_payload, |_| codec_unavailable())?; + if persisted.same_issue_identity(grant) { + return Ok(persisted); } + return Err(HandoffOpenAuthorityError::Conflict); + } + job_execute( + transaction, + "INSERT INTO handoff_open_grants_v1 ( + token_digest, issued_request_id, grant_payload, issued_at, expires_at, + consumed_request_id, consumed_input_digest, consumption_payload + ) VALUES (?1, ?2, ?3, ?4, ?5, NULL, NULL, NULL)", + vec![ + ExactSqlValue::Text(grant.token_digest().as_str().to_owned()), + ExactSqlValue::Text(grant.issued_request_id().as_str().to_owned()), + ExactSqlValue::Text(payload.to_owned()), + ExactSqlValue::Integer(grant.issued_at().0), + ExactSqlValue::Integer(grant.expires_at().0), + ], + )?; + Ok(grant.clone()) +} + +struct ConsumeRequest { + token_digest: ManifestDigest, + expected: HandoffOpenExpectationV1, + request_id: RequestId, + input_digest: ManifestDigest, + consumed_at: UtcMicros, +} + +fn consume_grant( + transaction: &ExactSqlJobTransaction<'_>, + request: &ConsumeRequest, +) -> Result { + let rows = job_query( + transaction, + "SELECT grant_payload, consumed_request_id, consumed_input_digest, + consumption_payload + FROM handoff_open_grants_v1 + WHERE token_digest = ?1", + vec![ExactSqlValue::Text( + request.token_digest.as_str().to_owned(), + )], + )?; + let Some(row) = rows.rows.first() else { + return Ok(HandoffOpenConsumeOutcomeV1::Concealed); + }; + let grant_payload = text_column(&row.values, 0).ok_or_else(codec_unavailable)?; + let grant: HandoffOpenGrantV1 = decode_json(grant_payload, |_| codec_unavailable())?; + if !request.expected.matches(grant.context()) || request.consumed_at >= *grant.expires_at() { + return Ok(HandoffOpenConsumeOutcomeV1::Concealed); + } - let consumed_request_id = - optional_text_at(&row.values, 1).map_err(|_| codec_unavailable())?; - let consumed_input_digest = - optional_text_at(&row.values, 2).map_err(|_| codec_unavailable())?; - let consumption_payload = - optional_text_at(&row.values, 3).map_err(|_| codec_unavailable())?; - match ( - consumed_request_id, - consumed_input_digest, - consumption_payload, - ) { - (Some(stored_request_id), Some(stored_input_digest), Some(payload)) => { - let consumption: HandoffOpenConsumptionV1 = - decode_json(payload, |_| codec_unavailable())?; - let _ = transaction.rollback(); - if stored_request_id != request_id.as_str() { - return Ok(HandoffOpenConsumeOutcomeV1::Concealed); - } - if stored_input_digest != input_digest.as_str() { - return Err(HandoffOpenAuthorityError::IdempotencyConflict); - } - if consumption.request_id() != request_id - || consumption.input_digest() != input_digest - { - return Err(codec_unavailable()); - } - return Ok(HandoffOpenConsumeOutcomeV1::Consumed(Box::new(consumption))); + let consumed_request_id = optional_text_at(&row.values, 1).map_err(|_| codec_unavailable())?; + let consumed_input_digest = + optional_text_at(&row.values, 2).map_err(|_| codec_unavailable())?; + let consumption_payload = optional_text_at(&row.values, 3).map_err(|_| codec_unavailable())?; + match ( + consumed_request_id, + consumed_input_digest, + consumption_payload, + ) { + (Some(stored_request_id), Some(stored_input_digest), Some(payload)) => { + let consumption: HandoffOpenConsumptionV1 = + decode_json(payload, |_| codec_unavailable())?; + if stored_request_id != request.request_id.as_str() { + return Ok(HandoffOpenConsumeOutcomeV1::Concealed); + } + if stored_input_digest != request.input_digest.as_str() { + return Err(HandoffOpenAuthorityError::IdempotencyConflict); } - (None, None, None) => {} - _ => return Err(codec_unavailable()), + if consumption.request_id() != &request.request_id + || consumption.input_digest() != &request.input_digest + { + return Err(codec_unavailable()); + } + return Ok(HandoffOpenConsumeOutcomeV1::Consumed(Box::new(consumption))); } + (None, None, None) => {} + _ => return Err(codec_unavailable()), + } - let consumption = grant - .consume(request_id.clone(), input_digest.clone(), consumed_at) - .map_err(|_| codec_unavailable())?; - let payload = encode_json(&consumption, |_| codec_unavailable())?; - execute_tx( - &transaction, - "UPDATE handoff_open_grants_v1 - SET consumed_request_id = ?2, - consumed_input_digest = ?3, - consumption_payload = ?4 - WHERE token_digest = ?1 - AND consumption_payload IS NULL", - vec![ - ExactSqlValue::Text(token_digest.as_str().to_owned()), - ExactSqlValue::Text(request_id.as_str().to_owned()), - ExactSqlValue::Text(input_digest.as_str().to_owned()), - ExactSqlValue::Text(payload), - ], + let consumption = grant + .consume( + request.request_id.clone(), + request.input_digest.clone(), + request.consumed_at, ) - .map_err(unavailable)?; - transaction - .commit() - .map(|_| HandoffOpenConsumeOutcomeV1::Consumed(Box::new(consumption))) - .map_err(unavailable) - } + .map_err(|_| codec_unavailable())?; + let payload = encode_json(&consumption, |_| codec_unavailable())?; + job_execute( + transaction, + "UPDATE handoff_open_grants_v1 + SET consumed_request_id = ?2, + consumed_input_digest = ?3, + consumption_payload = ?4 + WHERE token_digest = ?1 + AND consumption_payload IS NULL", + vec![ + ExactSqlValue::Text(request.token_digest.as_str().to_owned()), + ExactSqlValue::Text(request.request_id.as_str().to_owned()), + ExactSqlValue::Text(request.input_digest.as_str().to_owned()), + ExactSqlValue::Text(payload), + ], + )?; + Ok(HandoffOpenConsumeOutcomeV1::Consumed(Box::new(consumption))) } diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/handoff_open_storage.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/handoff_open_storage.rs index 6492b245d3..b7e42f0347 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/handoff_open_storage.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/handoff_open_storage.rs @@ -111,8 +111,10 @@ fn run(future: impl Future) -> T { #[test] fn consume_is_atomic_secret_free_and_idempotent_across_restart() { let store = RegisteredWorkflowStore::start("handoff-open"); - let sqlite = - HandoffOpenSqliteAuthority::from_retained_exact_sql(store.retained_exact_sql()).unwrap(); + let sqlite = run(HandoffOpenSqliteAuthority::from_retained_exact_sql( + store.retained_exact_sql(), + )) + .unwrap(); let issue_context = context("request.handoff.issue"); let service = HandoffOpenService::new(sqlite, CurrentTarget); let token = HandoffOpenToken::new(TOKEN_SECRET.to_owned()).unwrap(); @@ -148,8 +150,10 @@ fn consume_is_atomic_secret_free_and_idempotent_across_restart() { drop(service); let store = store.restart("handoff-open-restart"); - let sqlite = - HandoffOpenSqliteAuthority::from_retained_exact_sql(store.retained_exact_sql()).unwrap(); + let sqlite = run(HandoffOpenSqliteAuthority::from_retained_exact_sql( + store.retained_exact_sql(), + )) + .unwrap(); let open_context = context("request.handoff.open"); let service = HandoffOpenService::new(sqlite, CurrentTarget); let request = OpenTaskHandoffRequestV1 { @@ -199,8 +203,10 @@ fn consume_is_atomic_secret_free_and_idempotent_across_restart() { #[test] fn changed_input_for_the_same_request_is_an_idempotency_conflict() { let store = RegisteredWorkflowStore::start("handoff-open-idempotency-conflict"); - let sqlite = - HandoffOpenSqliteAuthority::from_retained_exact_sql(store.retained_exact_sql()).unwrap(); + let sqlite = run(HandoffOpenSqliteAuthority::from_retained_exact_sql( + store.retained_exact_sql(), + )) + .unwrap(); let issue_context = context("request.handoff.issue-idempotency-conflict"); let service = HandoffOpenService::new(sqlite.clone(), CurrentTarget); let token = HandoffOpenToken::new(TOKEN_SECRET.to_owned()).unwrap(); @@ -235,7 +241,7 @@ fn changed_input_for_the_same_request_is_an_idempotency_conflict() { assert_eq!(replay.receipt, first.receipt); assert_eq!( - sqlite.consume( + run(sqlite.consume( grant.token_digest(), &HandoffOpenExpectationV1::from_request( &open_context, @@ -246,7 +252,7 @@ fn changed_input_for_the_same_request_is_an_idempotency_conflict() { open_context.request_id(), &digest('d'), UtcMicros(2_200_000), - ), + )), Err(HandoffOpenAuthorityError::IdempotencyConflict) ); @@ -268,8 +274,10 @@ fn changed_input_for_the_same_request_is_an_idempotency_conflict() { #[test] fn wrong_session_and_expired_grants_are_concealed_without_consuming() { let store = RegisteredWorkflowStore::start("handoff-open-conceal"); - let sqlite = - HandoffOpenSqliteAuthority::from_retained_exact_sql(store.retained_exact_sql()).unwrap(); + let sqlite = run(HandoffOpenSqliteAuthority::from_retained_exact_sql( + store.retained_exact_sql(), + )) + .unwrap(); let issue_context = context("request.handoff.issue-conceal"); let service = HandoffOpenService::new(sqlite, CurrentTarget); let token = HandoffOpenToken::new(TOKEN_SECRET.to_owned()).unwrap(); @@ -328,8 +336,10 @@ fn wrong_session_and_expired_grants_are_concealed_without_consuming() { #[test] fn enumeration_reads_the_durable_frontier_secret_free_across_a_restart() { let store = RegisteredWorkflowStore::start("handoff-open-list"); - let sqlite = - HandoffOpenSqliteAuthority::from_retained_exact_sql(store.retained_exact_sql()).unwrap(); + let sqlite = run(HandoffOpenSqliteAuthority::from_retained_exact_sql( + store.retained_exact_sql(), + )) + .unwrap(); let issue_context = context("request.handoff.issue-list"); let service = HandoffOpenService::new(sqlite, CurrentTarget); let token = HandoffOpenToken::new(TOKEN_SECRET.to_owned()).unwrap(); @@ -351,8 +361,10 @@ fn enumeration_reads_the_durable_frontier_secret_free_across_a_restart() { // not from anything the issuing process held in memory. drop(service); let store = store.restart("handoff-open-list-restart"); - let sqlite = - HandoffOpenSqliteAuthority::from_retained_exact_sql(store.retained_exact_sql()).unwrap(); + let sqlite = run(HandoffOpenSqliteAuthority::from_retained_exact_sql( + store.retained_exact_sql(), + )) + .unwrap(); let service = HandoffOpenService::new(sqlite, CurrentTarget); let live = run(service.list_task( @@ -411,3 +423,66 @@ fn enumeration_reads_the_durable_frontier_secret_free_across_a_restart() { .unwrap(); assert!(other.handoffs.is_empty()); } + +/// Issuing and opening a handoff on a native worker parks no runtime thread: +/// the authority's writes are writer jobs and its reads are awaited. +#[test] +fn handoff_calls_awaited_on_a_native_worker_park_no_native_thread() { + use tracedecay_rusqlite_runtime::exact_sql::native_thread_reader_waits; + use tracedecay_rusqlite_runtime::native_runtime::{ + NativeTask, NativeTaskOutcome, block_on, native_thread_writer_waits, + }; + + let store = RegisteredWorkflowStore::start("handoff-open-native-worker"); + let retained = store.retained_exact_sql(); + let writer_waits = native_thread_writer_waits(); + let reader_waits = native_thread_reader_waits(); + let task = NativeTask::spawn(async move { + let sqlite = HandoffOpenSqliteAuthority::from_retained_exact_sql(retained) + .await + .map_err(|error| format!("{error:?}"))?; + let service = HandoffOpenService::new(sqlite, CurrentTarget); + let issue_context = context("request.handoff.native-issue"); + let token = HandoffOpenToken::new(TOKEN_SECRET.to_owned()).unwrap(); + service + .issue( + &issue_context, + binding(&issue_context), + &token, + UtcMicros(1_000_000), + UtcMicros(61_000_000), + ) + .await + .map_err(|error| format!("{error:?}"))?; + service + .open_task( + &context("request.handoff.native-open"), + OpenTaskHandoffRequestV1 { + token: TOKEN_SECRET.to_owned(), + session_id: id::("lsp-session.handoff.runtime-store"), + }, + authority_snapshot(), + UtcMicros(2_000_000), + ) + .await + .map_err(|error| format!("{error:?}"))?; + Ok::<_, String>(()) + }) + .unwrap(); + + assert_eq!( + block_on(task.wait()).unwrap(), + NativeTaskOutcome::Completed(Ok(())) + ); + assert_eq!(store.count("handoff_open_grants_v1"), 1); + assert_eq!( + native_thread_writer_waits() - writer_waits, + 0, + "no handoff write parked the native worker on the writer" + ); + assert_eq!( + native_thread_reader_waits() - reader_waits, + 0, + "no handoff read parked the native worker on a reader" + ); +}