diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/journal.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/journal.rs index f9d8dc039f..67b96dddc4 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/journal.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/journal.rs @@ -30,15 +30,15 @@ where Self { store } } - pub fn begin_or_replay( + pub async fn begin_or_replay( &self, request: GitIndexTransactionBeginRequestV1, ) -> Result { request.validate()?; - Ok(self.store.begin_or_replay(request)?) + Ok(self.store.begin_or_replay(request).await?) } - pub fn advance( + pub async fn advance( &self, idempotency_key: &GitIndexIdempotencyKey, current: &GitIndexTransactionJournalV1, @@ -48,14 +48,13 @@ where let mut replacement = current.clone(); let expected_phase_epoch = replacement.phase_epoch; replacement.advance(next, updated_at)?; - Ok(self.store.compare_and_swap_journal( - idempotency_key, - expected_phase_epoch, - replacement, - )?) + Ok(self + .store + .compare_and_swap_journal(idempotency_key, expected_phase_epoch, replacement) + .await?) } - pub fn write_terminal( + pub async fn write_terminal( &self, idempotency_key: &GitIndexIdempotencyKey, current: &GitIndexTransactionJournalV1, @@ -76,6 +75,6 @@ where receipt, }; write.validate()?; - Ok(self.store.write_terminal(write)?) + Ok(self.store.write_terminal(write).await?) } } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/mod.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/mod.rs index 526b8ba727..13de3f9bad 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/mod.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/mod.rs @@ -55,7 +55,7 @@ pub use service::{ CurrentGitIndexPolicyStateV1, DaemonGitIndexTransactionPort, GitIndexNativeExecutor, GitIndexPolicyRecheckPort, NativeGitIndexApplyResult, }; -pub use store::{DaemonGitIndexTransactionStore, SharedDaemonGitIndexTransactionStore}; +pub use store::DaemonGitIndexTransactionStore; /// The only constructor that makes a daemon Git transaction port available to /// callers. It creates one queue-owning service, completes durable startup @@ -73,29 +73,29 @@ where S: GitIndexTransactionStore, N: GitIndexRecoveryExecutor, { - pub fn save_preview_input( + pub async fn save_preview_input( &self, input: GitIndexPreviewInputV1, ) -> Result<(), GitIndexTransactionPortError> { - self.port.save_preview_input(input) + self.port.save_preview_input(input).await } - pub fn read_preview_input( + pub async fn read_preview_input( &self, preview_id: &GitIndexPreviewId, observed_at: UtcMicros, ) -> Result { - self.port.read_preview_input(preview_id, observed_at) + self.port.read_preview_input(preview_id, observed_at).await } - pub fn read_preview( + pub async fn read_preview( &self, preview_id: &GitIndexPreviewId, ) -> Result { - self.port.read_preview(preview_id) + self.port.read_preview(preview_id).await } - pub fn start( + pub async fn start( store: S, native: N, classifier: C, @@ -105,7 +105,7 @@ where ) -> Result { let port = DaemonGitIndexTransactionPort::new(store, native, classifier, authorization, queue); - port.recover_startup(observed_at)?; + port.recover_startup(observed_at).await?; Ok(Self { port }) } } @@ -114,15 +114,17 @@ impl DaemonGitIndexTransactionService where S: GitIndexTransactionStore, N: GitIndexNativeExecutor + GitIndexRecoveryExecutor, - C: GitEffectClassifier, + C: GitEffectClassifier + Sync, A: GitIndexPolicyRecheckPort, { - pub fn apply_cancellable( + pub async fn apply_cancellable( &self, request: &GitIndexApplyRequestV1, - cancellation_requested: impl Fn() -> Option, + cancellation_requested: impl Fn() -> Option + Sync, ) -> Result { - self.port.apply_cancellable(request, cancellation_requested) + self.port + .apply_cancellable(request, cancellation_requested) + .await } } @@ -130,28 +132,28 @@ impl GitIndexTransactionPort for DaemonGitIndexTransactionService Result { - self.port.preview(request) + self.port.preview(request).await } - fn apply( + async fn apply( &self, request: &GitIndexApplyRequestV1, ) -> Result { - self.port.apply(request) + self.port.apply(request).await } - fn recover( + async fn recover( &self, request: &GitIndexRecoveryRequestV1, ) -> Result { - self.port.recover(request) + self.port.recover(request).await } } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/native.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/native.rs index 29989dbc91..a4f3919f47 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/native.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/native.rs @@ -1090,7 +1090,7 @@ impl FixedDaemonGitIndexExecutor { impl GitIndexNativeExecutor for FixedDaemonGitIndexExecutor where - A: GitIndexPreviewAssembler, + A: GitIndexPreviewAssembler + Sync, { #[tracing::instrument(name = "daemon.git.tx.native_preview", level = "trace", skip_all)] fn preview( @@ -1220,7 +1220,7 @@ where impl GitIndexRecoveryExecutor for FixedDaemonGitIndexExecutor where - A: GitIndexPreviewAssembler, + A: GitIndexPreviewAssembler + Sync, { fn reconcile( &self, diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/owner.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/owner.rs index 354bfab4f1..ce2454c21b 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/owner.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/owner.rs @@ -22,10 +22,9 @@ use tracedecay_policy::{GitConflictRiskV1, GitEffectAuthorizationV1, GitEffectCl use tracedecay_tool_catalog::{CapabilityId, CatalogSnapshotV1}; use super::{ - CurrentGitIndexPolicyStateV1, DaemonGitIndexTransactionService, + CurrentGitIndexPolicyStateV1, DaemonGitIndexTransactionService, DaemonGitIndexTransactionStore, DaemonProjectGitIndexPreviewAssembler, FixedDaemonGitIndexExecutor, GitIndexPolicyRecheckPort, - GitIndexTransactionStoreRegistry, RepositoryMutationQueue, - SharedDaemonGitIndexTransactionStore, canonicalize_repository_root, + GitIndexTransactionStoreRegistry, RepositoryMutationQueue, canonicalize_repository_root, }; use tracedecay_application::ProjectSourceAccessSnapshot; use tracedecay_application::project_open_authorization::mounted_grant_expires_at; @@ -83,7 +82,7 @@ impl DaemonGitAuthoritySource for ProductionDaemonGitAuthoritySource { capability_id: &CapabilityId, ) -> Result { let evaluated_at = now_micros(); - // The Git transaction port is synchronous while the configuration + // The authority source is synchronous while the configuration // authority is asynchronous. Callers run on blocking threads, where // the native runtime may park the caller until the read settles. let current = native_runtime::block_on(self.configuration.current()) @@ -358,7 +357,7 @@ fn scope_matches_snapshot( } pub type DaemonProjectGitIndexTransactionService = DaemonGitIndexTransactionService< - SharedDaemonGitIndexTransactionStore, + DaemonGitIndexTransactionStore, FixedDaemonGitIndexExecutor, GitEffectClassifierV1, DaemonGitIndexPolicyRecheck, @@ -425,9 +424,9 @@ impl ServiceKey { } } -/// Owns the store actor, native executor, classifier, policy recheck, and +/// Owns the store, native executor, classifier, policy recheck, and /// repository queue for each exact project/worktree identity. Linked -/// worktrees share one session store actor without sharing native executors or +/// worktrees share one session store without sharing native executors or /// mutation authority. pub struct DaemonGitIndexTransactionServiceRegistry { catalog: ApplicationCatalogComposer, @@ -500,25 +499,24 @@ impl DaemonGitIndexTransactionServiceRegistry { project_id, source, observed_at, - || self.stores.ensure(database), + self.stores.ensure(database), ) .await } - pub(super) async fn mount_with( + pub(super) async fn mount_with( &self, database_path: PathBuf, repository_root: PathBuf, project_id: ProjectId, source: Arc, observed_at: UtcMicros, - open_store: F, - ) -> Result - where - F: FnOnce() -> tracedecay_store::GitIndexTransactionStoreResult< - super::SharedDaemonGitIndexTransactionStore, + open_store: impl Future< + Output = tracedecay_store::GitIndexTransactionStoreResult< + DaemonGitIndexTransactionStore, + >, >, - { + ) -> Result { if self.shutdown_fenced.load(Ordering::SeqCst) { return Err(GitIndexTransactionPortError::DaemonUnavailable); } @@ -545,14 +543,16 @@ impl DaemonGitIndexTransactionServiceRegistry { } } - // Open/retain the store actor under the creation gate before native - // recovery runs on a blocking thread. Later mounts reuse this actor. - let store = open_store().map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)?; + // Open/retain the store under the creation gate before native recovery + // runs on a blocking thread. Later mounts reuse this store. + let store = open_store + .await + .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)?; let native_root = repository_root.clone(); let authority = Arc::new(DaemonGitAuthoritySlot::new(source)); let service_authority = Arc::clone(&authority); let mutation_queue = Arc::clone(&self.mutation_queue); - let (project_id, service) = native_runtime::spawn_blocking(move || { + let (project_id, service) = native_runtime::on_blocking_pool(async move { let native = FixedDaemonGitIndexExecutor::new( DaemonProjectGitIndexPreviewAssembler::new(native_root, project_id.clone()), ); @@ -564,6 +564,7 @@ impl DaemonGitIndexTransactionServiceRegistry { mutation_queue, observed_at, ) + .await .map(|service| (project_id, service)) }) .await @@ -593,9 +594,8 @@ impl DaemonGitIndexTransactionServiceRegistry { .lock() .await .retain(|key, _| key.project_id != *project_id || key.database_path != database_path); - self.stores - .remove(database_path) - .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable) + self.stores.remove(database_path).await; + Ok(()) } /// Resolve only an owner already mounted by project-open admission. @@ -640,11 +640,7 @@ impl DaemonGitIndexTransactionServiceRegistry { }; let services_closed = services.len(); drop(services); - let store_actors_joined = self - .stores - .shutdown_all() - .await - .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)?; + let store_actors_joined = self.stores.shutdown_all().await; let receipt = DaemonGitIndexShutdownReceiptV1 { services_closed, store_actors_joined, @@ -665,17 +661,10 @@ impl DaemonGitIndexTransactionServiceRegistry { .for_repository_root(repository_root) .await? .ok_or(GitIndexTransactionPortError::DaemonUnavailable)?; - // The store actor waits on the project writer lane, which a task on - // the caller's runtime may hold, so the synchronous port cannot run - // on that runtime's thread. - let preview = preview.clone(); - native_runtime::spawn_blocking(move || { - owner - .service - .port - .quarantine_preview_for_test(&preview, observed_at) - }) - .await - .unwrap_or(Err(GitIndexTransactionPortError::DaemonUnavailable)) + owner + .service + .port + .quarantine_preview_for_test(preview, observed_at) + .await } } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/queue.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/queue.rs index f879f8bb9d..39187c4d05 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/queue.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/queue.rs @@ -5,11 +5,17 @@ use std::collections::BTreeMap; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Condvar, Mutex}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use thiserror::Error; +use tokio::sync::Semaphore; use tracedecay_domain::{RepositoryId, UtcMicros}; +use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; + +/// How often a queued mutation re-checks cancellation while it waits for the +/// repository permit. +const CANCELLATION_POLL: Duration = Duration::from_millis(50); #[derive(Debug, Error)] pub enum RepositoryMutationQueueError { @@ -19,35 +25,8 @@ pub enum RepositoryMutationQueueError { Saturated, } -struct RepositoryGate { - occupied: Mutex, - available: Condvar, -} - -impl RepositoryGate { - fn new() -> Self { - Self { - occupied: Mutex::new(false), - available: Condvar::new(), - } - } -} - -struct OccupiedGate<'a> { - gate: &'a RepositoryGate, -} - -impl Drop for OccupiedGate<'_> { - fn drop(&mut self) { - if let Ok(mut occupied) = self.gate.occupied.lock() { - *occupied = false; - self.gate.available.notify_one(); - } - } -} - pub struct RepositoryMutationQueue { - gates: std::sync::Mutex>>, + gates: Mutex>>, pending: AtomicUsize, capacity: usize, } @@ -64,13 +43,24 @@ impl Default for RepositoryMutationQueue { } } +/// Holds one pending-capacity slot and one repository gate reference, and +/// returns both when the mutation ends, however it ends. struct MutationAdmission<'a> { - pending: &'a AtomicUsize, + queue: &'a RepositoryMutationQueue, + repository_id: &'a RepositoryId, + gate: Arc, } impl Drop for MutationAdmission<'_> { fn drop(&mut self) { - self.pending.fetch_sub(1, Ordering::Release); + if let Ok(mut gates) = self.queue.gates.lock() + && gates.get(self.repository_id).is_some_and(|current| { + Arc::ptr_eq(current, &self.gate) && Arc::strong_count(current) == 2 + }) + { + gates.remove(self.repository_id); + } + self.queue.pending.fetch_sub(1, Ordering::Release); } } @@ -84,12 +74,16 @@ impl RepositoryMutationQueue { } } - pub fn with_repository( + pub async fn with_repository( &self, repository_id: &RepositoryId, - operation: impl FnOnce() -> T, - ) -> Result { + operation: impl FnOnce() -> F, + ) -> Result + where + F: Future, + { self.with_repository_cancellable(repository_id, || None, |_| operation()) + .await } /// Waits for the exact repository permit while polling cancellation. @@ -99,86 +93,57 @@ impl RepositoryMutationQueue { /// callers durably record cancellation without waiting behind unrelated /// native work. #[tracing::instrument(name = "daemon.git.tx.queue", level = "trace", skip_all)] - pub fn with_repository_cancellable( + pub async fn with_repository_cancellable( &self, repository_id: &RepositoryId, cancellation_requested: impl Fn() -> Option, - operation: impl FnOnce(Option) -> T, - ) -> Result { + operation: impl FnOnce(Option) -> F, + ) -> Result + where + F: Future, + { self.pending .try_update(Ordering::AcqRel, Ordering::Acquire, |pending| { (pending < self.capacity).then_some(pending + 1) }) .map_err(|_| RepositoryMutationQueueError::Saturated)?; - let _admission = MutationAdmission { - pending: &self.pending, - }; - let gate = { - let mut gates = self - .gates - .lock() - .map_err(|_| RepositoryMutationQueueError::Unavailable)?; - Arc::clone( + let gate = match self.gates.lock() { + Ok(mut gates) => Arc::clone( gates .entry(repository_id.clone()) - .or_insert_with(|| Arc::new(RepositoryGate::new())), - ) - }; - if let Some(cancelled_at) = cancellation_requested() { - self.release_idle_gate(repository_id, &gate)?; - return Ok(operation(Some(cancelled_at))); - } - let mut occupied = gate - .occupied - .lock() - .map_err(|_| RepositoryMutationQueueError::Unavailable)?; - loop { - if !*occupied { - *occupied = true; - break; + .or_insert_with(|| Arc::new(Semaphore::new(1))), + ), + Err(_) => { + self.pending.fetch_sub(1, Ordering::Release); + return Err(RepositoryMutationQueueError::Unavailable); } + }; + let admission = MutationAdmission { + queue: self, + repository_id, + gate, + }; + let permit = loop { if let Some(cancelled_at) = cancellation_requested() { - drop(occupied); - self.release_idle_gate(repository_id, &gate)?; - return Ok(operation(Some(cancelled_at))); + return Ok(operation(Some(cancelled_at)).await); } - let (guard, _) = gate - .available - .wait_timeout(occupied, Duration::from_millis(50)) - .map_err(|_| RepositoryMutationQueueError::Unavailable)?; - occupied = guard; - } - drop(occupied); - let _held = OccupiedGate { gate: &gate }; + match native_runtime::timeout(CANCELLATION_POLL, admission.gate.acquire()).await { + Ok(Ok(permit)) => break permit, + Err(TimeoutError::Elapsed) => {} + Ok(Err(_)) | Err(TimeoutError::TimerUnavailable(_)) => { + return Err(RepositoryMutationQueueError::Unavailable); + } + } + }; if let Some(cancelled_at) = cancellation_requested() { - drop(_held); - self.release_idle_gate(repository_id, &gate)?; - return Ok(operation(Some(cancelled_at))); + drop(permit); + return Ok(operation(Some(cancelled_at)).await); } - let result = operation(None); - drop(_held); - self.release_idle_gate(repository_id, &gate)?; + let result = operation(None).await; + drop(permit); Ok(result) } - fn release_idle_gate( - &self, - repository_id: &RepositoryId, - gate: &Arc, - ) -> Result<(), RepositoryMutationQueueError> { - let mut gates = self - .gates - .lock() - .map_err(|_| RepositoryMutationQueueError::Unavailable)?; - if gates - .get(repository_id) - .is_some_and(|current| Arc::ptr_eq(current, gate) && Arc::strong_count(current) == 2) - { - gates.remove(repository_id); - } - Ok(()) - } - #[cfg(test)] pub fn retained_gate_count_for_test(&self) -> Result { self.gates diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/recovery.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/recovery.rs index af1b14169f..3b6e46d6d5 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/recovery.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/recovery.rs @@ -27,7 +27,7 @@ pub enum GitIndexRecoveryError { /// Native reconciliation sees the real repository state but cannot initiate a /// fresh mutation. A `NeedsInspection` receipt is the required outcome when /// state drift prevents proof. -pub trait GitIndexRecoveryExecutor { +pub trait GitIndexRecoveryExecutor: Sync { fn reconcile( &self, record: &GitIndexTransactionRecordV1, @@ -48,15 +48,15 @@ where Self { store, native } } - pub fn recover_repository( + pub async fn recover_repository( &self, repository_id: &RepositoryId, observed_at: UtcMicros, ) -> Result, GitIndexRecoveryError> { - let records = self.store.recovery_candidates(repository_id)?; + let records = self.store.recovery_candidates(repository_id).await?; let mut receipts = Vec::with_capacity(records.len()); for record in records { - receipts.push(self.recover_record(&record, observed_at)?); + receipts.push(self.recover_record(&record, observed_at).await?); } Ok(receipts) } @@ -65,7 +65,7 @@ where /// admitted transaction whose native boundary became ambiguous; neither /// caller is allowed to invoke native apply a second time. #[tracing::instrument(name = "daemon.git.tx.recover_record", level = "trace", skip_all)] - pub fn recover_record( + pub async fn recover_record( &self, record: &GitIndexTransactionRecordV1, observed_at: UtcMicros, @@ -76,7 +76,7 @@ where if original_receipt.outcome != GitIndexReceiptOutcomeV1::NeedsInspection { return Ok(original_receipt.clone()); } - let proof = self.reconcile_or_quarantine(record, observed_at)?; + let proof = self.reconcile_or_quarantine(record, observed_at).await?; // A terminal inspection receipt has no post-boundary phase that // can distinguish a coincidental candidate from our mutation. // Exact restoration to the old snapshot is the only automatic @@ -85,41 +85,44 @@ where if proof.outcome != GitIndexReceiptOutcomeV1::AbortedNoChange || !receipt_binds_record(&proof, record) { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; return Err(GitIndexRecoveryError::Indeterminate); } let proof = recovery_proof_at(&proof, record, observed_at)?; - self.store.clear_repository_quarantine( - &record.journal.repository_id, - &record.journal.transaction_id, - proof.clone(), - )?; + self.store + .clear_repository_quarantine( + &record.journal.repository_id, + &record.journal.transaction_id, + proof.clone(), + ) + .await?; return Ok(proof); } - let receipt = self.reconcile_or_quarantine(record, observed_at)?; + let receipt = self.reconcile_or_quarantine(record, observed_at).await?; if !receipt_binds_record(&receipt, record) { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; return Err(GitIndexRecoveryError::Indeterminate); } let receipt = recovery_proof_at(&receipt, record, observed_at)?; - let Ok(preterminal) = advance_to_terminal(self.store, record, receipt.outcome, observed_at) + let Ok(preterminal) = + advance_to_terminal(self.store, record, receipt.outcome, observed_at).await else { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; return Err(GitIndexRecoveryError::Indeterminate); }; if receipt.outcome == GitIndexReceiptOutcomeV1::NeedsInspection { // Persist the blocking truth first. If the subsequent atomic // receipt write fails, the repository still remains fenced. - quarantine(self.store, record)?; + quarantine(self.store, record).await?; } let mut terminal_journal = preterminal; if terminal_journal .advance(terminal_phase(receipt.outcome), observed_at) .is_err() { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; return Err(GitIndexRecoveryError::Indeterminate); } let write = GitIndexTransactionTerminalWriteV1 { @@ -129,18 +132,18 @@ where receipt: receipt.clone(), }; if write.validate().is_err() { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; return Err(GitIndexRecoveryError::Indeterminate); } - if let Ok(stored) = self.store.write_terminal(write) { + if let Ok(stored) = self.store.write_terminal(write).await { Ok(stored) } else { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; Err(GitIndexRecoveryError::Indeterminate) } } - fn reconcile_or_quarantine( + async fn reconcile_or_quarantine( &self, record: &GitIndexTransactionRecordV1, observed_at: UtcMicros, @@ -154,7 +157,7 @@ where } { Ok(receipt) } else { - quarantine(self.store, record)?; + quarantine(self.store, record).await?; unobserved_needs_inspection(record, observed_at) } } @@ -201,7 +204,7 @@ fn unobserved_needs_inspection( .map_err(GitIndexRecoveryError::Domain) } -fn advance_to_terminal( +async fn advance_to_terminal( store: &S, record: &GitIndexTransactionRecordV1, outcome: GitIndexReceiptOutcomeV1, @@ -228,11 +231,9 @@ where for phase in phases { let expected_phase_epoch = journal.phase_epoch; journal.advance(*phase, observed_at)?; - journal = store.compare_and_swap_journal( - &record.idempotency_key, - expected_phase_epoch, - journal, - )?; + journal = store + .compare_and_swap_journal(&record.idempotency_key, expected_phase_epoch, journal) + .await?; } Ok(journal) } @@ -244,17 +245,19 @@ fn receipt_binds_record( record.receipt_binds_preview(receipt) } -fn quarantine( +async fn quarantine( store: &S, record: &GitIndexTransactionRecordV1, ) -> Result<(), GitIndexRecoveryError> where S: GitIndexTransactionStore, { - store.quarantine_repository( - &record.journal.repository_id, - &record.journal.transaction_id, - )?; + store + .quarantine_repository( + &record.journal.repository_id, + &record.journal.transaction_id, + ) + .await?; Ok(()) } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/registry.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/registry.rs index dc825c1abc..13a73839c4 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/registry.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/registry.rs @@ -1,12 +1,11 @@ //! Process-local singleton ownership for Git index transaction stores. //! -//! One bounded store actor is retained per registered project-session path. -//! Startup recovery and later mutation services must share that actor so a -//! second queue/journal authority cannot appear for the same database. +//! One store is retained per registered project-session path. Startup +//! recovery and later mutation services must share it so a second +//! queue/journal authority cannot appear for the same database. use std::collections::HashMap; use std::path::PathBuf; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use tracedecay_store::{GitIndexTransactionStoreError, GitIndexTransactionStoreResult}; @@ -14,32 +13,31 @@ use tracedecay_store::{GitIndexTransactionStoreError, GitIndexTransactionStoreRe use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use super::DaemonGitIndexTransactionStore; -use super::SharedDaemonGitIndexTransactionStore; -/// Retains the one `DaemonGitIndexTransactionStore` actor for each daemon-owned -/// project database. Dropping the registry closes every actor when the daemon -/// store administration shuts down. +/// Retains the one `DaemonGitIndexTransactionStore` for each daemon-owned +/// project database. Shutting the registry down closes every store when the +/// daemon store administration shuts down. pub struct GitIndexTransactionStoreRegistry { - stores: std::sync::Mutex>, + stores: tokio::sync::Mutex>, closed: AtomicBool, } impl Default for GitIndexTransactionStoreRegistry { fn default() -> Self { Self { - stores: std::sync::Mutex::new(HashMap::new()), + stores: tokio::sync::Mutex::new(HashMap::new()), closed: AtomicBool::new(false), } } } impl GitIndexTransactionStoreRegistry { - /// Returns the existing actor for `database`, or opens exactly one. + /// Returns the existing store for `database`, or opens exactly one. #[tracing::instrument(name = "daemon.git.tx.store_ensure", level = "trace", skip_all)] - pub fn ensure( + pub async fn ensure( &self, database: RegisteredGlobalDbLeaseV1, - ) -> GitIndexTransactionStoreResult { + ) -> GitIndexTransactionStoreResult { if self.closed.load(Ordering::SeqCst) { return Err(GitIndexTransactionStoreError::unavailable( "git index transaction store registry is shut down", @@ -49,10 +47,7 @@ impl GitIndexTransactionStoreRegistry { // database identity. Avoid a second filesystem lookup because a fresh // SQLite shard may not have materialized its path yet. let path = database.db_path().to_path_buf(); - let mut stores = self - .stores - .lock() - .map_err(GitIndexTransactionStoreError::unavailable)?; + let mut stores = self.stores.lock().await; if self.closed.load(Ordering::SeqCst) { return Err(GitIndexTransactionStoreError::unavailable( "git index transaction store registry is shut down", @@ -61,42 +56,25 @@ impl GitIndexTransactionStoreRegistry { if let Some(existing) = stores.get(&path) { return Ok(existing.clone()); } - let store = SharedDaemonGitIndexTransactionStore::from_arc(Arc::new( - DaemonGitIndexTransactionStore::open(database)?, - )); + let store = DaemonGitIndexTransactionStore::open(database).await?; stores.insert(path, store.clone()); Ok(store) } - /// Drops the exact project-session actor before its backing shard is + /// Drops the exact project-session store before its backing shard is /// destructively removed by daemon-owned lifecycle administration. - pub fn remove(&self, path: &std::path::Path) -> GitIndexTransactionStoreResult<()> { - let mut stores = self - .stores - .lock() - .map_err(GitIndexTransactionStoreError::unavailable)?; - stores.remove(path); - Ok(()) + pub async fn remove(&self, path: &std::path::Path) { + self.stores.lock().await.remove(path); } #[tracing::instrument(name = "daemon.git.tx.store_shutdown", level = "trace", skip_all)] - pub async fn shutdown_all(&self) -> GitIndexTransactionStoreResult { + pub async fn shutdown_all(&self) -> usize { self.closed.store(true, Ordering::SeqCst); - let stores = { - let mut retained = self - .stores - .lock() - .map_err(GitIndexTransactionStoreError::unavailable)?; - retained.drain().map(|(_, store)| store).collect::>() - }; - tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - let mut joined = 0usize; - for store in stores { - joined = joined.saturating_add(usize::from(store.shutdown()?)); - } - Ok(joined) - }) - .await - .map_err(GitIndexTransactionStoreError::unavailable)? + self.stores + .lock() + .await + .drain() + .map(|(_, store)| usize::from(store.shutdown())) + .sum() } } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/service.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/service.rs index d126f8050d..890b3ff595 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/service.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/service.rs @@ -25,6 +25,8 @@ use tracedecay_store::{ GitIndexTransactionBeginResultV1, GitIndexTransactionStore, GitIndexTransactionStoreError, }; +use tracing::Instrument; + use super::{ DurableGitIndexJournal, GitIndexJournalError, GitIndexRecoveryCoordinator, GitIndexRecoveryExecutor, RepositoryMutationQueue, RepositoryMutationQueueError, @@ -42,7 +44,7 @@ pub struct CurrentGitIndexPolicyStateV1 { /// Current grant, scope, policy, and configuration authority resolved /// immediately before the durable native effect boundary. -pub trait GitIndexPolicyRecheckPort { +pub trait GitIndexPolicyRecheckPort: Sync { fn recheck( &self, request: &GitIndexApplyRequestV1, @@ -76,7 +78,7 @@ pub enum NativeGitIndexApplyOutcomeV1 { /// adapters. Every post-admission result is classified so safe failures can /// receive an abort receipt while unknown commit-boundary state is reconciled /// exactly once rather than replayed. -pub trait GitIndexNativeExecutor { +pub trait GitIndexNativeExecutor: Sync { fn preview( &self, request: &GitIndexPreviewRequestV1, @@ -128,16 +130,17 @@ impl DaemonGitIndexTransactionPort where S: GitIndexTransactionStore, { - pub fn save_preview_input( + pub async fn save_preview_input( &self, input: GitIndexPreviewInputV1, ) -> Result<(), GitIndexTransactionPortError> { self.store .save_preview_input(input) + .await .map_err(map_store_error) } - pub fn read_preview_input( + pub async fn read_preview_input( &self, preview_id: &GitIndexPreviewId, observed_at: UtcMicros, @@ -145,6 +148,7 @@ where match self .store .read_preview_input(preview_id, observed_at) + .await .map_err(map_store_error)? { GitIndexPreviewInputReadV1::Available(input) => Ok(*input), @@ -155,12 +159,13 @@ where } } - pub fn read_preview( + pub async fn read_preview( &self, preview_id: &GitIndexPreviewId, ) -> Result { self.store .read_preview(preview_id) + .await .map_err(map_store_error)? .ok_or(GitIndexTransactionPortError::StalePreview) } @@ -172,7 +177,7 @@ where S: GitIndexTransactionStore, { #[cfg_attr(not(unix), allow(dead_code))] // exercised only by unix-only daemon tests - pub fn quarantine_preview_for_test( + pub async fn quarantine_preview_for_test( &self, preview: &GitIndexPreviewV1, observed_at: tracedecay_domain::UtcMicros, @@ -202,9 +207,11 @@ where ) .map_err(|_| GitIndexTransactionPortError::StalePreview)?, }) + .await .map_err(map_store_error)?; self.store .quarantine_repository(&preview.repository_snapshot.repository_id, &transaction_id) + .await .map_err(map_store_error) } } @@ -213,11 +220,11 @@ impl GitIndexTransactionPort for DaemonGitIndexTransactionPort Result { @@ -228,21 +235,21 @@ where if result.validate_for(request).is_err() { return Err(GitIndexTransactionPortError::StalePreview); } - if let Err(error) = self.store.save_preview(result.preview.clone()) { + if let Err(error) = self.store.save_preview(result.preview.clone()).await { return Err(map_store_error(error)); } Ok(result) } - fn apply( + async fn apply( &self, request: &GitIndexApplyRequestV1, ) -> Result { - self.apply_cancellable(request, || None) + self.apply_cancellable(request, || None).await } #[tracing::instrument(name = "daemon.git.tx.recover", level = "trace", skip_all)] - fn recover( + async fn recover( &self, request: &GitIndexRecoveryRequestV1, ) -> Result { @@ -255,16 +262,18 @@ where .validate() .map_err(|_| GitIndexTransactionPortError::StalePreview)?; self.queue - .with_repository(&request.repository_id, || { + .with_repository(&request.repository_id, || async { let coordinator = GitIndexRecoveryCoordinator::new(&self.store, &self.native); let receipts = coordinator .recover_repository(&request.repository_id, request.observed_at) + .await .map_err(|_| GitIndexTransactionPortError::NeedsInspection)?; receipts .into_iter() .find(|receipt| receipt.transaction_id == request.transaction_id) .ok_or(GitIndexTransactionPortError::StalePreview) }) + .await .map_err(map_queue_error)? } } @@ -273,19 +282,19 @@ impl DaemonGitIndexTransactionPort where S: GitIndexTransactionStore, N: GitIndexNativeExecutor + GitIndexRecoveryExecutor, - C: GitEffectClassifier, + C: GitEffectClassifier + Sync, A: GitIndexPolicyRecheckPort, { #[tracing::instrument(name = "daemon.git.tx.apply", level = "trace", skip_all)] - pub fn apply_cancellable( + pub async fn apply_cancellable( &self, request: &GitIndexApplyRequestV1, - cancellation_requested: impl Fn() -> Option, + cancellation_requested: impl Fn() -> Option + Sync, ) -> Result { request .validate() .map_err(|_| GitIndexTransactionPortError::StalePreview)?; - let preview = match self.store.read_preview(&request.preview_id) { + let preview = match self.store.read_preview(&request.preview_id).await { Ok(Some(preview)) => preview, Ok(None) => { return Err(GitIndexTransactionPortError::StalePreview); @@ -303,6 +312,7 @@ where .with_repository_cancellable(&repository_id, &cancellation_requested, |cancelled_at| { self.apply_serialized(request, &preview, cancelled_at, &cancellation_requested) }) + .await .map_err(map_queue_error); result? } @@ -312,15 +322,15 @@ impl DaemonGitIndexTransactionPort where S: GitIndexTransactionStore, N: GitIndexNativeExecutor + GitIndexRecoveryExecutor, - C: GitEffectClassifier, + C: GitEffectClassifier + Sync, A: GitIndexPolicyRecheckPort, { - fn apply_serialized( + async fn apply_serialized( &self, request: &GitIndexApplyRequestV1, preview: &GitIndexPreviewV1, cancellation_observed_while_queued: Option, - cancellation_requested: &impl Fn() -> Option, + cancellation_requested: &(impl Fn() -> Option + Sync), ) -> Result { let idempotency_key = request .native_idempotency_key() @@ -331,6 +341,7 @@ where if let Some(existing) = self .store .read_record(&idempotency_key) + .await .map_err(map_store_error)? { if existing.input_digest != input_digest || existing.preview != *preview { @@ -341,7 +352,9 @@ where None => Err(GitIndexTransactionPortError::RecoveryRequired), }; } - let input = self.read_preview_input(&request.preview_id, request.observed_at)?; + let input = self + .read_preview_input(&request.preview_id, request.observed_at) + .await?; if input.operation != preview.operation || input.repository_snapshot != preview.repository_snapshot || input.repository_snapshot_digest != preview.repository_snapshot_digest @@ -362,10 +375,10 @@ where journal, }; let durable = DurableGitIndexJournal::new(&self.store); - let match_result = { - let _span = tracing::trace_span!("daemon.git.tx.journal_begin").entered(); - durable.begin_or_replay(begin) - }; + let match_result = durable + .begin_or_replay(begin) + .instrument(tracing::trace_span!("daemon.git.tx.journal_begin")) + .await; let begin = match match_result { Ok(begin) => begin, Err(error) => return Err(map_journal_error(error)), @@ -392,7 +405,8 @@ where request, EffectTermination::Cancelled, Some(cancelled_at), - ); + ) + .await; } // Terminal replay has already returned above. All remaining paths // have admitted a new durable record and therefore must end in an @@ -410,7 +424,8 @@ where request, EffectTermination::Failed, None, - ); + ) + .await; } if let Some(cancelled_at) = cancellation_requested() { return finish_aborted_no_change( @@ -423,18 +438,22 @@ where request, EffectTermination::Cancelled, Some(cancelled_at), - ); + ) + .await; } - let Ok(started) = durable.advance( - &idempotency_key, - &record.journal, - GitIndexJournalPhaseV1::NativeApplyStarted, - request.observed_at, - ) else { + let Ok(started) = durable + .advance( + &idempotency_key, + &record.journal, + GitIndexJournalPhaseV1::NativeApplyStarted, + request.observed_at, + ) + .await + else { // No native boundary was entered. Prefer a durable no-change // terminal receipt; quarantine only if that safe terminal // write cannot be proven. - return finish_aborted_no_change( + let finished = finish_aborted_no_change( &self.store, &durable, &idempotency_key, @@ -445,9 +464,11 @@ where EffectTermination::Failed, None, ) - .inspect_err(|_| { - let _ = quarantine_after_admission(&self.store, preview, &transaction_id); - }); + .await; + if finished.is_err() { + let _ = quarantine_after_admission(&self.store, preview, &transaction_id).await; + } + return finished; }; if let Some(cancelled_at) = cancellation_requested() { return finish_aborted_no_change( @@ -460,20 +481,24 @@ where request, EffectTermination::Cancelled, Some(cancelled_at), - ); + ) + .await; } match self.native.apply(&transaction_id, preview, &input, request) { - Ok(NativeGitIndexApplyOutcomeV1::ProvenNoMutation) => finish_aborted_no_change( - &self.store, - &durable, - &idempotency_key, - &started, - &transaction_id, - preview, - request, - EffectTermination::Failed, - None, - ), + Ok(NativeGitIndexApplyOutcomeV1::ProvenNoMutation) => { + finish_aborted_no_change( + &self.store, + &durable, + &idempotency_key, + &started, + &transaction_id, + preview, + request, + EffectTermination::Failed, + None, + ) + .await + } Ok(NativeGitIndexApplyOutcomeV1::Completed(native)) if native_result_binds_record(&native, &record, &transaction_id) => { @@ -486,7 +511,8 @@ where &transaction_id, native.receipt, request.observed_at, - )?; + ) + .await?; result_from_receipt( request, deterministic_effect_id(&transaction_id)?, @@ -495,7 +521,7 @@ where ) } Ok(NativeGitIndexApplyOutcomeV1::Completed(_)) => { - quarantine_after_admission(&self.store, preview, &transaction_id)?; + quarantine_after_admission(&self.store, preview, &transaction_id).await?; Err(GitIndexTransactionPortError::NeedsInspection) } Ok(NativeGitIndexApplyOutcomeV1::CommitBoundaryUnknown) | Err(_) => { @@ -503,6 +529,7 @@ where recovery_record.journal = started; let receipt = GitIndexRecoveryCoordinator::new(&self.store, &self.native) .recover_record(&recovery_record, request.observed_at) + .await .map_err(|_| GitIndexTransactionPortError::NeedsInspection)?; let termination = match receipt.outcome { GitIndexReceiptOutcomeV1::Committed => EffectTermination::Completed, @@ -567,7 +594,7 @@ where /// Reconcile every unresolved durable record and active quarantine before /// the daemon admits any new transaction for an affected repository. #[tracing::instrument(name = "daemon.git.tx.recover_startup", level = "trace", skip_all)] - pub fn recover_startup( + pub async fn recover_startup( &self, observed_at: tracedecay_domain::UtcMicros, ) -> Result, GitIndexTransactionPortError> { @@ -575,15 +602,18 @@ where for repository_id in self .store .recovery_repositories() + .await .map_err(map_store_error)? { let recovered = self .queue - .with_repository(&repository_id, || { + .with_repository(&repository_id, || async { GitIndexRecoveryCoordinator::new(&self.store, &self.native) .recover_repository(&repository_id, observed_at) + .await .map_err(|_| GitIndexTransactionPortError::NeedsInspection) }) + .await .map_err(map_queue_error)??; receipts.extend(recovered); } @@ -592,7 +622,7 @@ where } #[allow(clippy::too_many_arguments)] -fn finish_aborted_no_change( +async fn finish_aborted_no_change( store: &S, durable: &DurableGitIndexJournal<'_, S>, idempotency_key: &tracedecay_domain::GitIndexIdempotencyKey, @@ -626,7 +656,8 @@ where transaction_id, receipt, request.observed_at, - )?; + ) + .await?; result_from_receipt( request, deterministic_effect_id(transaction_id)?, @@ -637,7 +668,7 @@ where #[allow(clippy::too_many_arguments)] #[tracing::instrument(name = "daemon.git.tx.journal_terminal", level = "trace", skip_all)] -fn terminalize_admitted_receipt( +async fn terminalize_admitted_receipt( store: &S, durable: &DurableGitIndexJournal<'_, S>, idempotency_key: &tracedecay_domain::GitIndexIdempotencyKey, @@ -651,7 +682,7 @@ where S: GitIndexTransactionStore, { if receipt.outcome == GitIndexReceiptOutcomeV1::NeedsInspection { - quarantine_after_admission(store, preview, transaction_id)?; + quarantine_after_admission(store, preview, transaction_id).await?; } let Ok(final_journal) = advance_for_native_outcome( durable, @@ -659,16 +690,19 @@ where journal, receipt.outcome, observed_at, - ) else { - quarantine_after_admission(store, preview, transaction_id)?; + ) + .await + else { + quarantine_after_admission(store, preview, transaction_id).await?; return Err(GitIndexTransactionPortError::NeedsInspection); }; - if let Ok(receipt) = - durable.write_terminal(idempotency_key, &final_journal, receipt, observed_at) + if let Ok(receipt) = durable + .write_terminal(idempotency_key, &final_journal, receipt, observed_at) + .await { Ok(receipt) } else { - quarantine_after_admission(store, preview, transaction_id)?; + quarantine_after_admission(store, preview, transaction_id).await?; Err(GitIndexTransactionPortError::NeedsInspection) } } @@ -690,7 +724,7 @@ fn native_result_binds_record( && record.receipt_binds_preview(receipt) } -fn quarantine_after_admission( +async fn quarantine_after_admission( store: &S, preview: &GitIndexPreviewV1, transaction_id: &GitIndexTransactionId, @@ -700,6 +734,7 @@ where { store .quarantine_repository(&preview.repository_snapshot.repository_id, transaction_id) + .await .map_err(|_| GitIndexTransactionPortError::NeedsInspection) } @@ -737,7 +772,7 @@ fn receipt_id( .map_err(|_| GitIndexTransactionPortError::NeedsInspection) } -fn advance_for_native_outcome( +async fn advance_for_native_outcome( durable: &DurableGitIndexJournal<'_, S>, idempotency_key: &tracedecay_domain::GitIndexIdempotencyKey, journal: &GitIndexTransactionJournalV1, @@ -758,7 +793,9 @@ where ], }; for phase in phases { - journal = durable.advance(idempotency_key, &journal, *phase, observed_at)?; + journal = durable + .advance(idempotency_key, &journal, *phase, observed_at) + .await?; } Ok(journal) } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/store.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/store.rs index d91fb675df..a4a0e292b2 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/store.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/store.rs @@ -1,21 +1,22 @@ -//! Bounded synchronous bridge to the registered session-store runtime. +//! Git index transaction store over the registered project session database. //! -//! The application Git port is deliberately synchronous because its native -//! executor is synchronous. Calling an async database through `block_on` on -//! a Tokio worker would pin that worker while an `IMMEDIATE` writer waits. This -//! adapter instead owns one bounded actor thread; the actor owns the async -//! rusqlite-runtime calls and every synchronous port call receives exactly one reply. -//! It has no filesystem path and cannot create a JSON side-file authority. - -use std::sync::Arc; -use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender, TrySendError, sync_channel}; +//! Every call awaits the async global-db store directly. Each mutation there is +//! one `IMMEDIATE` SQLite transaction (journal compare-and-swap included), so +//! concurrent callers need no actor to serialize them, and per-transaction +//! ordering comes from the caller awaiting each step in turn. It has no +//! filesystem path and cannot create a JSON side-file authority. + +use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::Notify; use tracedecay_domain::{ GitIndexIdempotencyKey, GitIndexPreviewId, GitIndexPreviewInputV1, GitIndexPreviewV1, GitIndexTransactionId, GitIndexTransactionJournalV1, GitIndexTransactionReceiptV1, RepositoryId, UtcMicros, }; +use tracedecay_global_db::{GitIndexReadExecutor, RegisteredGlobalDbLeaseV1}; +use tracedecay_runtime_core::native_runtime::{self, NativeTask, TimeoutError}; use tracedecay_store::{ CodeReadOperationV1, CodeReadResultV1, CodeRecoveryCandidatesQueryV1, CodeRecoveryRepositoriesQueryV1, GitIndexPreviewInputReadV1, GitIndexTransactionBeginRequestV1, @@ -24,99 +25,109 @@ use tracedecay_store::{ GitIndexTransactionTerminalWriteV1, MAX_GIT_INDEX_PREVIEW_INPUT_GC_BATCH, }; -use tracedecay_global_db::{ - GitIndexReadExecutor, GlobalDbGitIndexTransactionStore, RegisteredGlobalDbLeaseV1, -}; -use tracedecay_runtime_core::native_runtime; - -/// The actor queue is intentionally finite: saturation fails closed instead of -/// accumulating unbounded mutation work while a durable writer is stalled. -const GIT_INDEX_TRANSACTION_STORE_ACTOR_CAPACITY: usize = 64; -// Keep the sync port bounded by the same five-second writer wait used by -// `RegisteredGlobalDb`; callers can reconcile durable state after an unavailable result -// instead of pinning a daemon worker forever. -const GIT_INDEX_TRANSACTION_STORE_ACTOR_TIMEOUT: Duration = Duration::from_secs(5); - -type Reply = SyncSender>; - -enum StoreCommand { - SavePreviewInput(GitIndexPreviewInputV1, Reply<()>), - ReadPreviewInput( - GitIndexPreviewId, - UtcMicros, - Reply, - ), - PurgeExpiredPreviewInputs(UtcMicros, usize, Reply), - SavePreview(GitIndexPreviewV1, Reply<()>), - ReadCode(CodeReadOperationV1, Reply), - BeginOrReplay( - Box, - Reply, - ), - CompareAndSwapJournal( - GitIndexIdempotencyKey, - u64, - GitIndexTransactionJournalV1, - Reply, - ), - WriteTerminal( - GitIndexTransactionTerminalWriteV1, - Reply, - ), - QuarantineRepository(RepositoryId, GitIndexTransactionId, Reply<()>), - ClearRepositoryQuarantine( - RepositoryId, - GitIndexTransactionId, - GitIndexTransactionReceiptV1, - Reply<()>, - ), -} +/// A failed preview-input purge retries after this delay, the writer wait +/// budget of `RegisteredGlobalDb`, instead of spinning on a stalled writer. +const PREVIEW_INPUT_GC_RETRY: Duration = Duration::from_secs(5); -/// Synchronous `tracedecay-store` contract adapter over one already-open, -/// canonical registered project session database. +/// Shared handle to the one daemon-owned store for a project database. /// -/// Dropping the last adapter closes the command channel and lets the dedicated -/// actor exit. It intentionally has no `Clone` implementation: one daemon -/// service owns one bounded queue and actor for its transaction authority. +/// Clones share the lease and the preview-input GC task. `shutdown`, or +/// dropping the last clone, releases the lease and stops the GC task; calls +/// after shutdown fail closed as unavailable. +#[derive(Clone)] pub struct DaemonGitIndexTransactionStore { - commands: std::sync::Mutex>>, - worker: std::sync::Mutex>>, + inner: Arc, } -enum ActorDatabase { - Registered { - database: RegisteredGlobalDbLeaseV1, - #[cfg(test)] - gc_observer: Option>, - }, +struct StoreInner { + shared: Arc, + gc_task: NativeTask<()>, +} + +impl Drop for StoreInner { + fn drop(&mut self) { + self.close(); + } +} + +impl StoreInner { + fn close(&self) -> bool { + self.gc_task.abort(); + self.shared + .database + .lock() + .unwrap_or_else(PoisonError::into_inner) + .take() + .is_some() + } +} + +/// State the GC task shares with the store handles. The task holds only this, +/// so the handles alone decide when the lease is released. +struct StoreShared { + database: Mutex>, + next_expiry: Mutex>, + rearm: Notify, + retry_delay: Duration, + #[cfg(test)] + gc_observer: Option>, } -impl ActorDatabase { - fn git_index_transaction_store(&self) -> GlobalDbGitIndexTransactionStore<'_> { - match self { - Self::Registered { database, .. } => database.git_index_transaction_store(), +impl StoreShared { + fn database(&self) -> GitIndexTransactionStoreResult { + self.database + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + .ok_or_else(|| { + GitIndexTransactionStoreError::unavailable( + "git index transaction store is shut down", + ) + }) + } + + fn next_expiry(&self) -> Option { + *self + .next_expiry + .lock() + .unwrap_or_else(PoisonError::into_inner) + } + + /// Records an expiry deadline, waking the GC task when it moves earlier. + fn note_expiry(&self, expires_at: UtcMicros) { + if self.merge_expiry(Some(expires_at)) { + self.rearm.notify_one(); } } - async fn next_live_preview_input_expiry( - &self, - ) -> GitIndexTransactionStoreResult> { - self.git_index_transaction_store() - .next_live_preview_input_expiry() - .await + fn merge_expiry(&self, expires_at: Option) -> bool { + let Some(expires_at) = expires_at else { + return false; + }; + let mut next = self + .next_expiry + .lock() + .unwrap_or_else(PoisonError::into_inner); + let earlier = next.is_none_or(|current| expires_at < current); + if earlier { + *next = Some(expires_at); + } + earlier } - async fn purge_expired_preview_inputs( - &self, - observed_at: UtcMicros, - limit: usize, - ) -> GitIndexTransactionStoreResult { + /// Takes the due deadline before a purge, so a deadline a concurrent save + /// records while the purge runs is merged with the purge's result instead + /// of being overwritten by it. + fn claim_expiry(&self) -> Option { + self.next_expiry + .lock() + .unwrap_or_else(PoisonError::into_inner) + .take() + } + + async fn purge_due(&self) -> GitIndexTransactionStoreResult> { #[cfg(test)] - if let Self::Registered { - gc_observer: Some(observer), - .. - } = self - { + if let Some(observer) = &self.gc_observer { observer .purge_attempts .fetch_add(1, std::sync::atomic::Ordering::SeqCst); @@ -129,9 +140,15 @@ impl ActorDatabase { )); } } - self.git_index_transaction_store() - .purge_expired_preview_inputs_and_next(observed_at, limit) + let database = self.database()?; + database + .git_index_transaction_store() + .purge_expired_preview_inputs_and_next( + current_utc_micros()?, + MAX_GIT_INDEX_PREVIEW_INPUT_GC_BATCH, + ) .await + .map(|result| result.next_expiry) } } @@ -142,461 +159,264 @@ struct PreviewGcTestObserver { fail_purge: std::sync::atomic::AtomicBool, } +/// Purges expired preview inputs in bounded batches at each input's expiry +/// deadline. Saves move the deadline earlier through `rearm`; a full batch +/// leaves a due deadline, so the next batch runs at once. +async fn run_preview_input_gc(shared: Arc) { + loop { + let rearmed = shared.rearm.notified(); + let Some(expires_at) = shared.next_expiry() else { + rearmed.await; + continue; + }; + let wait = current_utc_micros().map_or(shared.retry_delay, |observed_at| { + duration_until(expires_at, observed_at) + }); + if !wait.is_zero() { + match native_runtime::timeout(wait, rearmed).await { + Ok(()) => continue, + Err(TimeoutError::Elapsed) => {} + Err(TimeoutError::TimerUnavailable(error)) => { + tracing::warn!(%error, "git index preview input GC timer unavailable"); + shared.rearm.notified().await; + continue; + } + } + } + let claimed = shared.claim_expiry(); + match shared.purge_due().await { + Ok(next_expiry) => { + shared.merge_expiry(next_expiry); + } + Err(error) => { + tracing::warn!(%error, "git index preview input GC failed"); + shared.merge_expiry(claimed); + let _ = native_runtime::timeout(shared.retry_delay, shared.rearm.notified()).await; + } + } + } +} + impl DaemonGitIndexTransactionStore { #[tracing::instrument(name = "daemon.git.tx.store_open", level = "trace", skip_all)] - pub fn open(database: RegisteredGlobalDbLeaseV1) -> GitIndexTransactionStoreResult { - Self::open_actor(ActorDatabase::Registered { + pub async fn open(database: RegisteredGlobalDbLeaseV1) -> GitIndexTransactionStoreResult { + Self::open_with( database, + PREVIEW_INPUT_GC_RETRY, #[cfg(test)] - gc_observer: None, - }) + None, + ) + .await } - #[cfg(test)] - fn open_with_gc_observer( + async fn open_with( database: RegisteredGlobalDbLeaseV1, - observer: Arc, + retry_delay: Duration, + #[cfg(test)] gc_observer: Option>, ) -> GitIndexTransactionStoreResult { - Self::open_actor(ActorDatabase::Registered { - database, - gc_observer: Some(observer), - }) - } - - fn open_actor(database: ActorDatabase) -> GitIndexTransactionStoreResult { - let (commands, receiver) = sync_channel(GIT_INDEX_TRANSACTION_STORE_ACTOR_CAPACITY); - let (ready, started) = sync_channel::>(1); - let worker = std::thread::Builder::new() - .name("tracedecay-git-index-store".to_owned()) - .spawn(move || { - let next_expiry = settle(database.next_live_preview_input_expiry()); - let Ok(next_expiry) = next_expiry else { - let _ = ready.send(Err(GitIndexTransactionStoreError::unavailable( - "failed to read next live preview input expiry", - ))); - return; - }; - if ready.send(Ok(())).is_err() { - return; - } - run_store_actor(&database, &receiver, next_expiry); - }) + let next_expiry = database + .git_index_transaction_store() + .next_live_preview_input_expiry() + .await?; + let shared = Arc::new(StoreShared { + database: Mutex::new(Some(database)), + next_expiry: Mutex::new(next_expiry), + rearm: Notify::new(), + retry_delay, + #[cfg(test)] + gc_observer, + }); + let gc_task = NativeTask::spawn(run_preview_input_gc(Arc::clone(&shared))) .map_err(GitIndexTransactionStoreError::unavailable)?; - let startup = started - .recv_timeout(GIT_INDEX_TRANSACTION_STORE_ACTOR_TIMEOUT) - .map_err(GitIndexTransactionStoreError::unavailable) - .and_then(|result| result); - if let Err(error) = startup { - drop(commands); - let _ = worker.join(); - return Err(error); - } Ok(Self { - commands: std::sync::Mutex::new(Some(commands)), - worker: std::sync::Mutex::new(Some(worker)), + inner: Arc::new(StoreInner { shared, gc_task }), }) } - fn submit(&self, command: StoreCommand) -> GitIndexTransactionStoreResult<()> { - self.commands - .lock() - .map_err(GitIndexTransactionStoreError::unavailable)? - .as_ref() - .ok_or_else(|| { - GitIndexTransactionStoreError::unavailable("git index store actor is shut down") - })? - .try_send(command) - .map_err(|error| match error { - TrySendError::Full(_) | TrySendError::Disconnected(_) => { - GitIndexTransactionStoreError::unavailable(error) - } - }) + /// Releases the lease and stops the GC task. Returns whether this call + /// closed the store. + pub fn shutdown(&self) -> bool { + self.inner.close() } - fn await_reply( - receiver: &Receiver>, - ) -> GitIndexTransactionStoreResult { - receiver - .recv_timeout(GIT_INDEX_TRANSACTION_STORE_ACTOR_TIMEOUT) - .map_err(|error| match error { - RecvTimeoutError::Timeout | RecvTimeoutError::Disconnected => { - GitIndexTransactionStoreError::unavailable(error) - } - })? + fn database(&self) -> GitIndexTransactionStoreResult { + self.inner.shared.database() } - fn execute_code_read( + async fn execute_code_read( &self, operation: CodeReadOperationV1, ) -> GitIndexTransactionStoreResult { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::ReadCode(operation, reply))?; - Self::await_reply(&receiver) - } - - pub fn shutdown(&self) -> GitIndexTransactionStoreResult { - self.commands - .lock() - .map_err(GitIndexTransactionStoreError::unavailable)? - .take(); - let worker = self - .worker - .lock() - .map_err(GitIndexTransactionStoreError::unavailable)? - .take(); - let Some(worker) = worker else { - return Ok(false); - }; - worker.join().map_err(|_| { - GitIndexTransactionStoreError::unavailable("git index store worker panicked") - })?; - Ok(true) + let database = self.database()?; + let store = database.git_index_transaction_store(); + GitIndexReadExecutor::new(&store) + .execute_read(&operation) + .await } } -impl Drop for DaemonGitIndexTransactionStore { - fn drop(&mut self) { - let _ = self.shutdown(); - } +fn unexpected_read_result() -> GitIndexTransactionStoreError { + GitIndexTransactionStoreError::unavailable("git index store returned an unexpected read result") } impl GitIndexTransactionStore for DaemonGitIndexTransactionStore { - fn save_preview_input( + async fn save_preview_input( &self, input: GitIndexPreviewInputV1, ) -> GitIndexTransactionStoreResult<()> { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::SavePreviewInput(input, reply))?; - Self::await_reply(&receiver) + let expires_at = input.expires_at; + self.database()? + .git_index_transaction_store() + .save_preview_input(input) + .await?; + self.inner.shared.note_expiry(expires_at); + Ok(()) } - fn read_preview_input( + async fn read_preview_input( &self, preview_id: &GitIndexPreviewId, observed_at: UtcMicros, ) -> GitIndexTransactionStoreResult { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::ReadPreviewInput( - preview_id.clone(), - observed_at, - reply, - ))?; - Self::await_reply(&receiver) + self.database()? + .git_index_transaction_store() + .read_preview_input(preview_id, observed_at) + .await } - fn purge_expired_preview_inputs( + async fn purge_expired_preview_inputs( &self, observed_at: UtcMicros, limit: usize, ) -> GitIndexTransactionStoreResult { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::PurgeExpiredPreviewInputs( - observed_at, - limit, - reply, - ))?; - Self::await_reply(&receiver) + self.database()? + .git_index_transaction_store() + .purge_expired_preview_inputs(observed_at, limit) + .await } - fn save_preview(&self, preview: GitIndexPreviewV1) -> GitIndexTransactionStoreResult<()> { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::SavePreview(preview, reply))?; - Self::await_reply(&receiver) + async fn save_preview(&self, preview: GitIndexPreviewV1) -> GitIndexTransactionStoreResult<()> { + self.database()? + .git_index_transaction_store() + .save_preview(preview) + .await } - fn read_preview( + async fn read_preview( &self, preview_id: &GitIndexPreviewId, ) -> GitIndexTransactionStoreResult> { - match self.execute_code_read(CodeReadOperationV1::Preview(preview_id.clone()))? { + match self + .execute_code_read(CodeReadOperationV1::Preview(preview_id.clone())) + .await? + { CodeReadResultV1::Preview(preview) => Ok(*preview), - _ => Err(GitIndexTransactionStoreError::unavailable( - "git index store returned an unexpected read result", - )), + _ => Err(unexpected_read_result()), } } - fn read_record( + async fn read_record( &self, idempotency_key: &GitIndexIdempotencyKey, ) -> GitIndexTransactionStoreResult> { - match self.execute_code_read(CodeReadOperationV1::TransactionRecord( - idempotency_key.clone(), - ))? { + match self + .execute_code_read(CodeReadOperationV1::TransactionRecord( + idempotency_key.clone(), + )) + .await? + { CodeReadResultV1::TransactionRecord(record) => Ok(*record), - _ => Err(GitIndexTransactionStoreError::unavailable( - "git index store returned an unexpected read result", - )), + _ => Err(unexpected_read_result()), } } - fn begin_or_replay( + async fn begin_or_replay( &self, request: GitIndexTransactionBeginRequestV1, ) -> GitIndexTransactionStoreResult { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::BeginOrReplay(Box::new(request), reply))?; - Self::await_reply(&receiver) + self.database()? + .git_index_transaction_store() + .begin_or_replay(request) + .await } - fn compare_and_swap_journal( + async fn compare_and_swap_journal( &self, idempotency_key: &GitIndexIdempotencyKey, expected_phase_epoch: u64, replacement: GitIndexTransactionJournalV1, ) -> GitIndexTransactionStoreResult { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::CompareAndSwapJournal( - idempotency_key.clone(), - expected_phase_epoch, - replacement, - reply, - ))?; - Self::await_reply(&receiver) + self.database()? + .git_index_transaction_store() + .compare_and_swap_journal(idempotency_key, expected_phase_epoch, replacement) + .await } - fn write_terminal( + async fn write_terminal( &self, write: GitIndexTransactionTerminalWriteV1, ) -> GitIndexTransactionStoreResult { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::WriteTerminal(write, reply))?; - Self::await_reply(&receiver) + self.database()? + .git_index_transaction_store() + .write_terminal(write) + .await } - fn recovery_candidates( + async fn recovery_candidates( &self, repository_id: &RepositoryId, ) -> GitIndexTransactionStoreResult> { - match self.execute_code_read(CodeReadOperationV1::RecoveryCandidates( - CodeRecoveryCandidatesQueryV1 { - repository_id: repository_id.clone(), - after: None, - limit: u32::MAX, - }, - ))? { + match self + .execute_code_read(CodeReadOperationV1::RecoveryCandidates( + CodeRecoveryCandidatesQueryV1 { + repository_id: repository_id.clone(), + after: None, + limit: u32::MAX, + }, + )) + .await? + { CodeReadResultV1::RecoveryCandidates(page) => Ok(page.records), - _ => Err(GitIndexTransactionStoreError::unavailable( - "git index store returned an unexpected read result", - )), + _ => Err(unexpected_read_result()), } } - fn recovery_repositories(&self) -> GitIndexTransactionStoreResult> { - match self.execute_code_read(CodeReadOperationV1::RecoveryRepositories( - CodeRecoveryRepositoriesQueryV1 { - after: None, - limit: u32::MAX, - }, - ))? { + async fn recovery_repositories(&self) -> GitIndexTransactionStoreResult> { + match self + .execute_code_read(CodeReadOperationV1::RecoveryRepositories( + CodeRecoveryRepositoriesQueryV1 { + after: None, + limit: u32::MAX, + }, + )) + .await? + { CodeReadResultV1::RecoveryRepositories(page) => Ok(page.repositories), - _ => Err(GitIndexTransactionStoreError::unavailable( - "git index store returned an unexpected read result", - )), + _ => Err(unexpected_read_result()), } } - fn quarantine_repository( + async fn quarantine_repository( &self, repository_id: &RepositoryId, transaction_id: &GitIndexTransactionId, ) -> GitIndexTransactionStoreResult<()> { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::QuarantineRepository( - repository_id.clone(), - transaction_id.clone(), - reply, - ))?; - Self::await_reply(&receiver) + self.database()? + .git_index_transaction_store() + .quarantine_repository(repository_id, transaction_id) + .await } - fn clear_repository_quarantine( + async fn clear_repository_quarantine( &self, repository_id: &RepositoryId, transaction_id: &GitIndexTransactionId, recovery_receipt: GitIndexTransactionReceiptV1, ) -> GitIndexTransactionStoreResult<()> { - let (reply, receiver) = sync_channel(1); - self.submit(StoreCommand::ClearRepositoryQuarantine( - repository_id.clone(), - transaction_id.clone(), - recovery_receipt, - reply, - ))?; - Self::await_reply(&receiver) - } -} - -fn settle( - operation: impl Future>, -) -> GitIndexTransactionStoreResult { - native_runtime::block_on(operation) - .unwrap_or_else(|error| Err(GitIndexTransactionStoreError::unavailable(error))) -} - -fn run_store_actor( - database: &ActorDatabase, - receiver: &Receiver, - mut next_expiry: Option, -) { - let mut gc_due = false; - loop { - let timeout = if gc_due { - None - } else if let Some(expires_at) = next_expiry { - let observed_at = match current_utc_micros() { - Ok(observed_at) => observed_at, - Err(clock_error) => match receiver.recv() { - Ok(command) => { - reject_store_command(command, clock_error); - continue; - } - Err(_) => break, - }, - }; - Some(duration_until(expires_at, observed_at)) - } else { - None - }; - let command = match timeout { - Some(timeout) => match receiver.recv_timeout(timeout) { - Ok(command) => Some(command), - Err(RecvTimeoutError::Disconnected) => break, - Err(RecvTimeoutError::Timeout) => None, - }, - None => match receiver.recv() { - Ok(command) => Some(command), - Err(_) => break, - }, - }; - - let observed_at = match current_utc_micros() { - Ok(observed_at) => observed_at, - Err(clock_error) => { - if let Some(command) = command { - reject_store_command(command, clock_error); - } - continue; - } - }; - gc_due |= next_expiry.is_some_and(|expires_at| expires_at <= observed_at); - if gc_due { - match settle( - database.purge_expired_preview_inputs( - observed_at, - MAX_GIT_INDEX_PREVIEW_INPUT_GC_BATCH, - ), - ) { - Ok(result) => { - next_expiry = result.next_expiry; - gc_due = false; - } - Err(error) => { - tracing::warn!(%error, "git index preview input GC failed"); - if let Some(command) = command { - let cause = match &error { - GitIndexTransactionStoreError::Unavailable(reason) => reason.clone(), - other => other.to_string(), - }; - reject_store_command( - command, - GitIndexTransactionStoreError::unavailable(format!( - "expired preview-input purge failed before this command: {cause}" - )), - ); - } - continue; - } - } - } - let Some(command) = command else { - continue; - }; - match command { - StoreCommand::SavePreviewInput(input, reply) => { - let expires_at = input.expires_at; - let result = settle( - database - .git_index_transaction_store() - .save_preview_input(input), - ); - if result.is_ok() && next_expiry.is_none_or(|current| expires_at < current) { - next_expiry = Some(expires_at); - } - let _ = reply.send(result); - } - StoreCommand::ReadPreviewInput(preview_id, observed_at, reply) => { - let result = settle( - database - .git_index_transaction_store() - .read_preview_input(&preview_id, observed_at), - ); - let _ = reply.send(result); - } - StoreCommand::PurgeExpiredPreviewInputs(observed_at, limit, reply) => { - let result = settle(database.purge_expired_preview_inputs(observed_at, limit)); - match result { - Ok(result) => { - next_expiry = result.next_expiry; - let _ = reply.send(Ok(result.purged)); - } - Err(error) => { - let _ = reply.send(Err(error)); - } - } - } - StoreCommand::SavePreview(preview, reply) => { - let result = settle(database.git_index_transaction_store().save_preview(preview)); - let _ = reply.send(result); - } - StoreCommand::ReadCode(operation, reply) => { - let store = database.git_index_transaction_store(); - let executor = GitIndexReadExecutor::new(&store); - let result = settle(executor.execute_read(&operation)); - let _ = reply.send(result); - } - StoreCommand::BeginOrReplay(request, reply) => { - let result = settle( - database - .git_index_transaction_store() - .begin_or_replay(*request), - ); - let _ = reply.send(result); - } - StoreCommand::CompareAndSwapJournal(key, epoch, replacement, reply) => { - let result = settle( - database - .git_index_transaction_store() - .compare_and_swap_journal(&key, epoch, replacement), - ); - let _ = reply.send(result); - } - StoreCommand::WriteTerminal(write, reply) => { - let result = settle(database.git_index_transaction_store().write_terminal(write)); - let _ = reply.send(result); - } - StoreCommand::QuarantineRepository(repository_id, transaction_id, reply) => { - let result = settle( - database - .git_index_transaction_store() - .quarantine_repository(&repository_id, &transaction_id), - ); - let _ = reply.send(result); - } - StoreCommand::ClearRepositoryQuarantine( - repository_id, - transaction_id, - receipt, - reply, - ) => { - let result = settle( - database - .git_index_transaction_store() - .clear_repository_quarantine(&repository_id, &transaction_id, receipt), - ); - let _ = reply.send(result); - } - } + self.database()? + .git_index_transaction_store() + .clear_repository_quarantine(repository_id, transaction_id, recovery_receipt) + .await } } @@ -607,44 +427,6 @@ fn duration_until(expires_at: UtcMicros, observed_at: UtcMicros) -> Duration { Duration::from_micros(expires_at.0.saturating_sub(observed_at.0).unsigned_abs()) } -/// Reject a queued command with the typed cause that prevented executing it, -/// so callers observe the real failure (for example a failed due purge) -/// instead of a fabricated generic rejection. -fn reject_store_command(command: StoreCommand, error: GitIndexTransactionStoreError) { - match command { - StoreCommand::SavePreviewInput(_, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::ReadPreviewInput(_, _, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::PurgeExpiredPreviewInputs(_, _, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::SavePreview(_, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::ReadCode(_, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::BeginOrReplay(_, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::CompareAndSwapJournal(_, _, _, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::WriteTerminal(_, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::QuarantineRepository(_, _, reply) => { - let _ = reply.send(Err(error)); - } - StoreCommand::ClearRepositoryQuarantine(_, _, _, reply) => { - let _ = reply.send(Err(error)); - } - } -} - fn current_utc_micros() -> GitIndexTransactionStoreResult { let elapsed = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -654,123 +436,6 @@ fn current_utc_micros() -> GitIndexTransactionStoreResult { Ok(UtcMicros(micros)) } -/// Shared handle to the one daemon-owned store actor for a project database. -/// -/// This local newtype exists so the foreign `GitIndexTransactionStore` trait -/// can be implemented for a shareable handle without violating orphan rules -/// around `Arc`. -#[derive(Clone)] -pub struct SharedDaemonGitIndexTransactionStore { - inner: Arc, -} - -impl SharedDaemonGitIndexTransactionStore { - pub fn from_arc(inner: Arc) -> Self { - Self { inner } - } - - pub fn shutdown(&self) -> GitIndexTransactionStoreResult { - self.inner.shutdown() - } -} - -impl GitIndexTransactionStore for SharedDaemonGitIndexTransactionStore { - fn save_preview_input( - &self, - input: GitIndexPreviewInputV1, - ) -> GitIndexTransactionStoreResult<()> { - self.inner.save_preview_input(input) - } - - fn read_preview_input( - &self, - preview_id: &GitIndexPreviewId, - observed_at: UtcMicros, - ) -> GitIndexTransactionStoreResult { - self.inner.read_preview_input(preview_id, observed_at) - } - - fn purge_expired_preview_inputs( - &self, - observed_at: UtcMicros, - limit: usize, - ) -> GitIndexTransactionStoreResult { - self.inner.purge_expired_preview_inputs(observed_at, limit) - } - - fn save_preview(&self, preview: GitIndexPreviewV1) -> GitIndexTransactionStoreResult<()> { - self.inner.save_preview(preview) - } - - fn read_preview( - &self, - preview_id: &GitIndexPreviewId, - ) -> GitIndexTransactionStoreResult> { - self.inner.read_preview(preview_id) - } - - fn read_record( - &self, - idempotency_key: &GitIndexIdempotencyKey, - ) -> GitIndexTransactionStoreResult> { - self.inner.read_record(idempotency_key) - } - - fn begin_or_replay( - &self, - request: GitIndexTransactionBeginRequestV1, - ) -> GitIndexTransactionStoreResult { - self.inner.begin_or_replay(request) - } - - fn compare_and_swap_journal( - &self, - idempotency_key: &GitIndexIdempotencyKey, - expected_phase_epoch: u64, - replacement: GitIndexTransactionJournalV1, - ) -> GitIndexTransactionStoreResult { - self.inner - .compare_and_swap_journal(idempotency_key, expected_phase_epoch, replacement) - } - - fn write_terminal( - &self, - write: GitIndexTransactionTerminalWriteV1, - ) -> GitIndexTransactionStoreResult { - self.inner.write_terminal(write) - } - - fn recovery_candidates( - &self, - repository_id: &RepositoryId, - ) -> GitIndexTransactionStoreResult> { - self.inner.recovery_candidates(repository_id) - } - - fn recovery_repositories(&self) -> GitIndexTransactionStoreResult> { - self.inner.recovery_repositories() - } - - fn quarantine_repository( - &self, - repository_id: &RepositoryId, - transaction_id: &GitIndexTransactionId, - ) -> GitIndexTransactionStoreResult<()> { - self.inner - .quarantine_repository(repository_id, transaction_id) - } - - fn clear_repository_quarantine( - &self, - repository_id: &RepositoryId, - transaction_id: &GitIndexTransactionId, - recovery_receipt: GitIndexTransactionReceiptV1, - ) -> GitIndexTransactionStoreResult<()> { - self.inner - .clear_repository_quarantine(repository_id, transaction_id, recovery_receipt) - } -} - #[cfg(test)] mod gc_tests { use std::sync::atomic::Ordering; @@ -778,25 +443,21 @@ mod gc_tests { use super::*; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; + use tracedecay_runtime_core::native_runtime::NativeTaskOutcome; - fn registered_database(label: &str) -> RegisteredGlobalDbHarness { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("registered fixture runtime"); - runtime.block_on(RegisteredGlobalDbHarness::open(label)) - } - - fn store_with_gc_observer( + async fn store_with_gc_observer( label: &str, observer: Arc, + retry_delay: Duration, ) -> (RegisteredGlobalDbHarness, DaemonGitIndexTransactionStore) { - let database = registered_database(label); - let store = DaemonGitIndexTransactionStore::open_with_gc_observer( + let database = RegisteredGlobalDbHarness::open(label).await; + let store = DaemonGitIndexTransactionStore::open_with( database.registered.clone(), - observer, + retry_delay, + Some(observer), ) - .expect("registered store actor"); + .await + .expect("registered store"); (database, store) } @@ -841,11 +502,41 @@ mod gc_tests { ); } - #[test] - fn one_expiry_causes_one_bounded_purge_and_then_sleeps() { + /// A purge attempt is counted before its write commits, so a reader + /// waits for the tombstone itself. + async fn wait_for_tombstone( + store: &DaemonGitIndexTransactionStore, + preview_id: &GitIndexPreviewId, + ) -> GitIndexPreviewInputReadV1 { + let deadline = Instant::now() + Duration::from_secs(3); + loop { + let read = store + .read_preview_input(preview_id, current_utc_micros().expect("clock")) + .await + .expect("read preview input"); + let purged = matches!( + read, + GitIndexPreviewInputReadV1::Expired { + purged_at: Some(_), + .. + } + ); + if purged || Instant::now() >= deadline { + return read; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + #[tokio::test] + async fn one_expiry_causes_one_bounded_purge_and_then_sleeps() { let observer = Arc::new(PreviewGcTestObserver::default()); - let (_database, store) = - store_with_gc_observer("git-index-gc-one-expiry", Arc::clone(&observer)); + let (_database, store) = store_with_gc_observer( + "git-index-gc-one-expiry", + Arc::clone(&observer), + PREVIEW_INPUT_GC_RETRY, + ) + .await; let created_at = current_utc_micros().expect("clock"); let input = preview_input( "one", @@ -853,16 +544,17 @@ mod gc_tests { UtcMicros(created_at.0.saturating_add(100_000)), ); let preview_id = input.preview_id.clone(); - store.save_preview_input(input).expect("save preview input"); + store + .save_preview_input(input) + .await + .expect("save preview input"); wait_for_attempts(&observer, 1); std::thread::sleep(Duration::from_millis(150)); assert_eq!(observer.purge_attempts.load(Ordering::SeqCst), 1); assert!(matches!( - store - .read_preview_input(&preview_id, current_utc_micros().expect("clock")) - .expect("read tombstone"), + wait_for_tombstone(&store, &preview_id).await, GitIndexPreviewInputReadV1::Expired { purged_at: Some(_), .. @@ -870,14 +562,14 @@ mod gc_tests { )); } - #[test] - fn restart_reconstructs_expired_cleanup_deadline() { - let database = registered_database("git-index-gc-restart"); + #[tokio::test] + async fn restart_reconstructs_expired_cleanup_deadline() { + let database = RegisteredGlobalDbHarness::open("git-index-gc-restart").await; let now = current_utc_micros().expect("clock"); - // The input is created durable and already expired, and the daemon - // actor that wrote it is gone: only the restarted actor can - // reconstruct the cleanup deadline. Seeding through the registered - // store keeps the pre-restart daemon GC from racing the fixture. + // The input is durable and already expired, and the store that wrote + // it is gone: only the reopened store can reconstruct the cleanup + // deadline. Seeding through the registered store keeps a pre-restart + // GC task from racing the fixture. let created_at = UtcMicros(now.0.saturating_sub(1_000_000)); let input = preview_input( "restart", @@ -885,31 +577,25 @@ mod gc_tests { UtcMicros(created_at.0.saturating_add(100_000)), ); let preview_id = input.preview_id.clone(); - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("seed runtime"); - runtime - .block_on( - database - .registered - .git_index_transaction_store() - .save_preview_input(input), - ) + database + .registered + .git_index_transaction_store() + .save_preview_input(input) + .await .expect("save preview input"); let observer = Arc::new(PreviewGcTestObserver::default()); - let store = DaemonGitIndexTransactionStore::open_with_gc_observer( + let store = DaemonGitIndexTransactionStore::open_with( database.registered.clone(), - Arc::clone(&observer), + PREVIEW_INPUT_GC_RETRY, + Some(Arc::clone(&observer)), ) - .expect("restarted registered store actor"); + .await + .expect("reopened registered store"); wait_for_attempts(&observer, 1); assert!(matches!( - store - .read_preview_input(&preview_id, current_utc_micros().expect("clock")) - .expect("read restart tombstone"), + wait_for_tombstone(&store, &preview_id).await, GitIndexPreviewInputReadV1::Expired { purged_at: Some(_), .. @@ -917,12 +603,16 @@ mod gc_tests { )); } - #[test] - fn failed_due_gc_waits_for_command_and_never_fabricates_success() { + #[tokio::test] + async fn failed_gc_retries_after_its_delay_without_failing_commands() { let observer = Arc::new(PreviewGcTestObserver::default()); observer.fail_purge.store(true, Ordering::SeqCst); - let (_database, store) = - store_with_gc_observer("git-index-gc-failure", Arc::clone(&observer)); + let (_database, store) = store_with_gc_observer( + "git-index-gc-failure", + Arc::clone(&observer), + Duration::from_secs(1), + ) + .await; let created_at = current_utc_micros().expect("clock"); let input = preview_input( "failure", @@ -930,28 +620,80 @@ mod gc_tests { UtcMicros(created_at.0.saturating_add(100_000)), ); let preview_id = input.preview_id.clone(); - store.save_preview_input(input).expect("save preview input"); + store + .save_preview_input(input) + .await + .expect("save preview input"); wait_for_attempts(&observer, 1); std::thread::sleep(Duration::from_millis(150)); - assert_eq!(observer.purge_attempts.load(Ordering::SeqCst), 1); + assert_eq!( + observer.purge_attempts.load(Ordering::SeqCst), + 1, + "a failed purge waits out its retry delay" + ); assert!(matches!( - store.read_preview_input(&preview_id, current_utc_micros().expect("clock")), - Err(GitIndexTransactionStoreError::Unavailable(reason)) - if reason.contains("injected preview-input purge failure") + store + .read_preview_input(&preview_id, current_utc_micros().expect("clock")) + .await + .expect("a failed purge does not fail unrelated commands"), + GitIndexPreviewInputReadV1::Expired { + purged_at: None, + .. + } )); - assert_eq!(observer.purge_attempts.load(Ordering::SeqCst), 2); observer.fail_purge.store(false, Ordering::SeqCst); + wait_for_attempts(&observer, 2); assert!(matches!( - store - .read_preview_input(&preview_id, current_utc_micros().expect("clock")) - .expect("retry purge before command"), + wait_for_tombstone(&store, &preview_id).await, GitIndexPreviewInputReadV1::Expired { purged_at: Some(_), .. } )); - assert_eq!(observer.purge_attempts.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn store_awaited_on_a_native_worker_never_parks_it() { + let database = RegisteredGlobalDbHarness::open("git-index-native-worker").await; + let registered = database.registered.clone(); + let writer_waits = native_runtime::native_thread_writer_waits(); + let reader_waits = native_runtime::native_thread_reader_waits(); + + let task = NativeTask::spawn(async move { + assert!( + native_runtime::Runtime::current_handle().is_some(), + "the store must be awaited on a native worker" + ); + let store = DaemonGitIndexTransactionStore::open(registered).await?; + let input = + super::super::test_support::preview_input(&super::super::test_support::preview()); + let preview_id = input.preview_id.clone(); + let observed_at = input.created_at; + store.save_preview_input(input).await?; + let read = store.read_preview_input(&preview_id, observed_at).await; + store.shutdown(); + read + }) + .expect("native task admitted"); + + let NativeTaskOutcome::Completed(read) = task.wait().await else { + panic!("store task did not complete"); + }; + assert!(matches!( + read.expect("store calls settle on a native worker"), + GitIndexPreviewInputReadV1::Available(_) + )); + assert_eq!( + native_runtime::native_thread_writer_waits(), + writer_waits, + "no writer wait parked a native worker" + ); + assert_eq!( + native_runtime::native_thread_reader_waits(), + reader_waits, + "no reader wait parked a native worker" + ); } } diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/test_support.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/test_support.rs index 8ce91325be..479c12408a 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/test_support.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/test_support.rs @@ -330,7 +330,7 @@ fn preview_for_operation( } /// Fixture timeline anchored at a fixed far-future epoch (2100-01-01 UTC) so -/// the store actor's autonomous wall-clock preview-input GC never observes +/// the store's autonomous wall-clock preview-input GC never observes /// fixture expiries as already elapsed. Ordinals are spaced 200ms apart: wide /// enough to preserve every strict ordering the tests assert, narrow enough /// that fixture preview lifetimes stay inside the domain's 30s maximum. @@ -505,28 +505,26 @@ pub fn execution_for( } } -pub fn test_store() -> (RegisteredGlobalDbHarness, DaemonGitIndexTransactionStore) { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("registered fixture runtime"); - let database = runtime.block_on(RegisteredGlobalDbHarness::open("git-index-transaction")); +pub async fn test_store() -> (RegisteredGlobalDbHarness, DaemonGitIndexTransactionStore) { + let database = RegisteredGlobalDbHarness::open("git-index-transaction").await; let store = DaemonGitIndexTransactionStore::open(database.registered.clone()) - .expect("canonical registered store actor"); + .await + .expect("canonical registered store"); (database, store) } -pub fn test_port_from_preview( +pub async fn test_port_from_preview( operation: GitIndexTransactionOperationV1, native_modes: impl IntoIterator, recovery_modes: impl IntoIterator, ) -> TestHarness { - let (database, store) = test_store(); + let (database, store) = test_store().await; let preview_template = preview_for_operation(operation, fixture_time(100)); let authority_request = apply_request(&preview_template, "idempotency.transport"); let preview_request = preview_request(&preview_template, &authority_request); store .save_preview_input(preview_input(&preview_template)) + .await .expect("save daemon preview input"); let native = FakeNative::with_preview(Some(preview_template.clone()), native_modes, recovery_modes); @@ -546,6 +544,7 @@ pub fn test_port_from_preview( ); let preview = port .preview(&preview_request) + .await .expect("preview reaches the native owner") .preview; let request = apply_request(&preview, "idempotency.transport"); diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/tests.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/tests.rs index 4720e51072..fd98457850 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/tests.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/tests.rs @@ -1,6 +1,5 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::thread; use std::time::Duration; use std::collections::BTreeSet; @@ -42,7 +41,6 @@ use super::test_support::{ use super::{ DaemonGitAuthorityStateV1, DaemonGitIndexTransactionPort, DaemonGitIndexTransactionService, DaemonGitIndexTransactionServiceRegistry, DaemonGitInvocationOwner, - SharedDaemonGitIndexTransactionStore, }; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; @@ -53,8 +51,8 @@ fn test_catalog_snapshot() -> Result Ok(CatalogSnapshotBuilderV1::new().build()?) } -#[test] -fn same_repository_mutations_never_enter_the_native_section_together() { +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn same_repository_mutations_never_enter_the_native_section_together() { let queue = Arc::new(RepositoryMutationQueue::default()); let repository = RepositoryId::new("repository.fixture").expect("repository id"); let active = Arc::new(AtomicUsize::new(0)); @@ -66,31 +64,35 @@ fn same_repository_mutations_never_enter_the_native_section_together() { let repository = repository.clone(); let active = Arc::clone(&active); let peak = Arc::clone(&peak); - thread::spawn(move || { + tokio::spawn(async move { queue - .with_repository(&repository, || { + .with_repository(&repository, || async { let current = active.fetch_add(1, Ordering::SeqCst) + 1; peak.fetch_max(current, Ordering::SeqCst); - thread::sleep(Duration::from_millis(5)); + tokio::time::sleep(Duration::from_millis(5)).await; active.fetch_sub(1, Ordering::SeqCst); }) + .await .expect("queue is available"); }) }) .collect(); for worker in workers { - worker.join().expect("worker joins"); + worker.await.expect("worker joins"); } assert_eq!(peak.load(Ordering::SeqCst), 1); } -#[test] -fn repository_mutation_capacity_fails_closed_before_waiting() { +#[tokio::test] +async fn repository_mutation_capacity_fails_closed_before_waiting() { let queue = RepositoryMutationQueue::with_capacity_for_test(1); let repository = RepositoryId::new("repository.capacity").expect("repository id"); let nested = queue - .with_repository(&repository, || queue.with_repository(&repository, || ())) + .with_repository(&repository, || { + queue.with_repository(&repository, || std::future::ready(())) + }) + .await .expect("outer mutation admitted"); assert!(matches!( @@ -98,19 +100,23 @@ fn repository_mutation_capacity_fails_closed_before_waiting() { Err(RepositoryMutationQueueError::Saturated) )); assert!( - queue.with_repository(&repository, || ()).is_ok(), + queue + .with_repository(&repository, || std::future::ready(())) + .await + .is_ok(), "capacity is released when an operation exits" ); } -#[test] -fn completed_repository_mutation_gates_are_reclaimed() { +#[tokio::test] +async fn completed_repository_mutation_gates_are_reclaimed() { let queue = RepositoryMutationQueue::default(); for index in 0..256 { let repository = RepositoryId::new(format!("repository.reclaimed.{index}")).expect("repository id"); queue - .with_repository(&repository, || ()) + .with_repository(&repository, || std::future::ready(())) + .await .expect("repository mutation"); } @@ -123,32 +129,32 @@ fn completed_repository_mutation_gates_are_reclaimed() { ); } -#[test] -fn queued_repository_mutation_observes_live_cancellation() { +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn queued_repository_mutation_observes_live_cancellation() { let queue = Arc::new(RepositoryMutationQueue::default()); let repository = RepositoryId::new("repository.cancellation").expect("repository id"); - let (entered_tx, entered_rx) = std::sync::mpsc::channel(); - let (release_tx, release_rx) = std::sync::mpsc::channel(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>(); let holder_queue = Arc::clone(&queue); let holder_repository = repository.clone(); - let holder = thread::spawn(move || { + let holder = tokio::spawn(async move { holder_queue - .with_repository(&holder_repository, || { + .with_repository(&holder_repository, || async { entered_tx.send(()).expect("signal queue holder"); - release_rx.recv().expect("release queue holder"); + release_rx.await.expect("release queue holder"); }) + .await .expect("holder queue"); }); - entered_rx.recv().expect("queue holder entered"); + entered_rx.await.expect("queue holder entered"); let cancellation = Arc::new(AtomicBool::new(false)); let checks = Arc::new(AtomicUsize::new(0)); let waiter_queue = Arc::clone(&queue); let waiter_cancellation = Arc::clone(&cancellation); let waiter_checks = Arc::clone(&checks); - let (waiter_result_tx, waiter_result_rx) = std::sync::mpsc::channel(); - let waiter = thread::spawn(move || { - let result = waiter_queue + let waiter = tokio::spawn(async move { + waiter_queue .with_repository_cancellable( &repository, || { @@ -157,10 +163,10 @@ fn queued_repository_mutation_observes_live_cancellation() { .load(Ordering::SeqCst) .then_some(fixture_time(25)) }, - |observed| observed.is_some(), + |observed| std::future::ready(observed.is_some()), ) - .expect("waiter queue"); - waiter_result_tx.send(result).expect("waiter result"); + .await + .expect("waiter queue") }); let polling = std::time::Instant::now() + Duration::from_secs(5); @@ -169,22 +175,22 @@ fn queued_repository_mutation_observes_live_cancellation() { std::time::Instant::now() < polling, "queued waiter never polled its cancellation" ); - thread::yield_now(); + tokio::time::sleep(Duration::from_millis(1)).await; } // The waiter returns only from a check that observed the cancellation, so // its result is the signal to wait for; a check count read after the store // can already include that final check. cancellation.store(true, Ordering::SeqCst); assert!( - waiter_result_rx - .recv_timeout(Duration::from_secs(5)) + tokio::time::timeout(Duration::from_secs(5), waiter) + .await .expect("cancelled waiter exits before repository holder") + .expect("waiter joins") ); assert!(checks.load(Ordering::SeqCst) > 2); release_tx.send(()).expect("release holder"); - holder.join().expect("holder joins"); - waiter.join().expect("waiter joins"); + holder.await.expect("holder joins"); } #[test] @@ -259,55 +265,57 @@ fn daemon_policy_recheck_rejects_a_capability_revoked_after_preview() { struct StartupUnavailableStore(DaemonGitIndexTransactionStore); impl GitIndexTransactionStore for StartupUnavailableStore { - fn save_preview_input( + async fn save_preview_input( &self, input: GitIndexPreviewInputV1, ) -> GitIndexTransactionStoreResult<()> { - self.0.save_preview_input(input) + self.0.save_preview_input(input).await } - fn read_preview_input( + async fn read_preview_input( &self, preview_id: &GitIndexPreviewId, observed_at: UtcMicros, ) -> GitIndexTransactionStoreResult { - self.0.read_preview_input(preview_id, observed_at) + self.0.read_preview_input(preview_id, observed_at).await } - fn purge_expired_preview_inputs( + async fn purge_expired_preview_inputs( &self, observed_at: UtcMicros, limit: usize, ) -> GitIndexTransactionStoreResult { - self.0.purge_expired_preview_inputs(observed_at, limit) + self.0 + .purge_expired_preview_inputs(observed_at, limit) + .await } - fn save_preview(&self, preview: GitIndexPreviewV1) -> GitIndexTransactionStoreResult<()> { - self.0.save_preview(preview) + async fn save_preview(&self, preview: GitIndexPreviewV1) -> GitIndexTransactionStoreResult<()> { + self.0.save_preview(preview).await } - fn read_preview( + async fn read_preview( &self, preview_id: &GitIndexPreviewId, ) -> GitIndexTransactionStoreResult> { - self.0.read_preview(preview_id) + self.0.read_preview(preview_id).await } - fn read_record( + async fn read_record( &self, idempotency_key: &GitIndexIdempotencyKey, ) -> GitIndexTransactionStoreResult> { - self.0.read_record(idempotency_key) + self.0.read_record(idempotency_key).await } - fn begin_or_replay( + async fn begin_or_replay( &self, request: GitIndexTransactionBeginRequestV1, ) -> GitIndexTransactionStoreResult { - self.0.begin_or_replay(request) + self.0.begin_or_replay(request).await } - fn compare_and_swap_journal( + async fn compare_and_swap_journal( &self, idempotency_key: &GitIndexIdempotencyKey, expected_phase_epoch: u64, @@ -315,37 +323,42 @@ impl GitIndexTransactionStore for StartupUnavailableStore { ) -> GitIndexTransactionStoreResult { self.0 .compare_and_swap_journal(idempotency_key, expected_phase_epoch, replacement) + .await } - fn write_terminal( + async fn write_terminal( &self, write: GitIndexTransactionTerminalWriteV1, ) -> GitIndexTransactionStoreResult { - self.0.write_terminal(write) + self.0.write_terminal(write).await } - fn recovery_candidates( + async fn recovery_candidates( &self, repository_id: &RepositoryId, ) -> GitIndexTransactionStoreResult> { - self.0.recovery_candidates(repository_id) + self.0.recovery_candidates(repository_id).await } - fn recovery_repositories(&self) -> GitIndexTransactionStoreResult> { - Err(GitIndexTransactionStoreError::Unavailable( + fn recovery_repositories( + &self, + ) -> impl Future>> + Send { + std::future::ready(Err(GitIndexTransactionStoreError::Unavailable( "recovery repository enumeration is unavailable in this fixture".to_string(), - )) + ))) } - fn quarantine_repository( + async fn quarantine_repository( &self, repository_id: &RepositoryId, transaction_id: &GitIndexTransactionId, ) -> GitIndexTransactionStoreResult<()> { - self.0.quarantine_repository(repository_id, transaction_id) + self.0 + .quarantine_repository(repository_id, transaction_id) + .await } - fn clear_repository_quarantine( + async fn clear_repository_quarantine( &self, repository_id: &RepositoryId, transaction_id: &GitIndexTransactionId, @@ -353,19 +366,24 @@ impl GitIndexTransactionStore for StartupUnavailableStore { ) -> GitIndexTransactionStoreResult<()> { self.0 .clear_repository_quarantine(repository_id, transaction_id, recovery_receipt) + .await } } -fn test_port( +async fn test_port( native_modes: impl IntoIterator, recovery_modes: impl IntoIterator, ) -> TestHarness { - let (database, store) = test_store(); + let (database, store) = test_store().await; let preview = preview(); store .save_preview_input(preview_input(&preview)) + .await .expect("save preview input"); - store.save_preview(preview.clone()).expect("save preview"); + store + .save_preview(preview.clone()) + .await + .expect("save preview"); let request = apply_request(&preview, "idempotency.fixture"); let native = FakeNative::new(native_modes, recovery_modes); let apply_calls = Arc::clone(&native.apply_calls); @@ -395,7 +413,7 @@ fn test_port( } } -fn seed_prepared_transaction( +async fn seed_prepared_transaction( store: &DaemonGitIndexTransactionStore, preview: &GitIndexPreviewV1, request: &GitIndexApplyRequestV1, @@ -414,17 +432,17 @@ fn seed_prepared_transaction( .expect("prepared journal"), }; assert!(matches!( - store.begin_or_replay(begin), + store.begin_or_replay(begin).await, Ok(GitIndexTransactionBeginResultV1::Started(_)) )); } -fn seed_terminal_abort( +async fn seed_terminal_abort( store: &DaemonGitIndexTransactionStore, preview: &GitIndexPreviewV1, request: &GitIndexApplyRequestV1, ) { - seed_prepared_transaction(store, preview, request); + seed_prepared_transaction(store, preview, request).await; let transaction_id = transaction_id_for(request, preview); let mut journal = GitIndexTransactionJournalV1::prepared( transaction_id.clone(), @@ -449,16 +467,18 @@ fn seed_terminal_abort( GitIndexReceiptOutcomeV1::AbortedNoChange, ), }) + .await .expect("terminal receipt"); } -#[test] -fn safe_native_failure_receives_terminal_abort_and_replays_without_native_work() { - let harness = test_port([NativeMode::ProvenNoMutation], []); +#[tokio::test] +async fn safe_native_failure_receives_terminal_abort_and_replays_without_native_work() { + let harness = test_port([NativeMode::ProvenNoMutation], []).await; let first = harness .port .apply(&harness.request) + .await .expect("safe failure is receipted"); assert_eq!( first.receipt.outcome, @@ -468,6 +488,7 @@ fn safe_native_failure_receives_terminal_abort_and_replays_without_native_work() let replay = harness .port .apply(&harness.request) + .await .expect("terminal replay"); assert_eq!(replay.receipt, first.receipt); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); @@ -475,21 +496,24 @@ fn safe_native_failure_receives_terminal_abort_and_replays_without_native_work() assert_eq!(harness.policy_calls.load(Ordering::SeqCst), 1); } -#[test] -fn terminal_replay_bypasses_revalidated_policy_but_conflicting_effect_input_rejects() { +#[tokio::test] +async fn terminal_replay_bypasses_revalidated_policy_but_conflicting_effect_input_rejects() { let harness = test_port( [NativeMode::Completed(GitIndexReceiptOutcomeV1::Committed)], [], - ); + ) + .await; let first = harness .port .apply(&harness.request) + .await .expect("complete transaction"); harness.allow.store(false, Ordering::SeqCst); let replay = harness .port .apply(&harness.request) + .await .expect("replay ignores later policy denial"); assert_eq!(replay.receipt, first.receipt); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); @@ -500,24 +524,25 @@ fn terminal_replay_bypasses_revalidated_policy_but_conflicting_effect_input_reje let replay = harness .port .apply(&revalidated) + .await .expect("authorization evidence does not change semantic effect identity"); assert_eq!(replay.receipt, first.receipt); let mut conflicting = harness.request.clone(); conflicting.preview_digest = digest('f'); assert_eq!( - harness.port.apply(&conflicting), + harness.port.apply(&conflicting).await, Err(GitIndexTransactionPortError::IdempotencyConflict) ); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); } -#[test] -fn rejected_admitted_apply_never_reaches_native_git() { - let harness = test_port([], []); +#[tokio::test] +async fn rejected_admitted_apply_never_reaches_native_git() { + let harness = test_port([], []).await; harness.allow.store(false, Ordering::SeqCst); - let result = harness.port.apply(&harness.request); + let result = harness.port.apply(&harness.request).await; assert!( result.is_ok(), @@ -526,9 +551,9 @@ fn rejected_admitted_apply_never_reaches_native_git() { assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 0); } -#[test] -fn deadline_elapsed_during_policy_recheck_never_reaches_native_git() { - let harness = test_port([], []); +#[tokio::test] +async fn deadline_elapsed_during_policy_recheck_never_reaches_native_git() { + let harness = test_port([], []).await; *harness .policy_evaluated_at .lock() @@ -537,6 +562,7 @@ fn deadline_elapsed_during_policy_recheck_never_reaches_native_git() { let result = harness .port .apply(&harness.request) + .await .expect("elapsed deadline is a durable no-change result"); assert_eq!( @@ -546,9 +572,9 @@ fn deadline_elapsed_during_policy_recheck_never_reaches_native_git() { assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 0); } -#[test] -fn cancellation_at_the_last_pre_native_boundary_returns_a_cancelled_receipt() { - let harness = test_port([], []); +#[tokio::test] +async fn cancellation_at_the_last_pre_native_boundary_returns_a_cancelled_receipt() { + let harness = test_port([], []).await; let cancellation_checks = AtomicUsize::new(0); let result = harness @@ -557,6 +583,7 @@ fn cancellation_at_the_last_pre_native_boundary_returns_a_cancelled_receipt() { (cancellation_checks.fetch_add(1, Ordering::SeqCst) + 1 >= 5) .then_some(fixture_time(25)) }) + .await .expect("pre-native cancellation is a durable no-change result"); assert_eq!( @@ -580,9 +607,9 @@ fn cancellation_at_the_last_pre_native_boundary_returns_a_cancelled_receipt() { .expect("cancelled apply result contract"); } -#[test] -fn cancellation_after_native_entry_does_not_rewrite_the_terminal_outcome() { - let harness = test_port([NativeMode::ProvenNoMutation], []); +#[tokio::test] +async fn cancellation_after_native_entry_does_not_rewrite_the_terminal_outcome() { + let harness = test_port([NativeMode::ProvenNoMutation], []).await; let result = harness .port @@ -592,6 +619,7 @@ fn cancellation_after_native_entry_does_not_rewrite_the_terminal_outcome() { .load(Ordering::SeqCst) .then_some(fixture_time(25)) }) + .await .expect("native outcome remains authoritative"); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); @@ -603,12 +631,12 @@ fn cancellation_after_native_entry_does_not_rewrite_the_terminal_outcome() { assert!(result.execution.cancellation.is_none()); } -#[test] -fn terminal_replay_bypasses_preview_expiry_before_policy_or_native_execution() { - let (_database, store) = test_store(); +#[tokio::test] +async fn terminal_replay_bypasses_preview_expiry_before_policy_or_native_execution() { + let (_database, store) = test_store().await; let preview = preview_with_expiry(fixture_time(14)); let request = apply_request(&preview, "idempotency.expired-replay"); - seed_terminal_abort(&store, &preview, &request); + seed_terminal_abort(&store, &preview, &request).await; let native = FakeNative::new([], []); let apply_calls = Arc::clone(&native.apply_calls); let policy = TestPolicy::allowing(); @@ -622,7 +650,7 @@ fn terminal_replay_bypasses_preview_expiry_before_policy_or_native_execution() { Arc::new(RepositoryMutationQueue::default()), ); - let replay = port.apply(&request).expect("expired terminal replay"); + let replay = port.apply(&request).await.expect("expired terminal replay"); assert_eq!( replay.receipt.outcome, GitIndexReceiptOutcomeV1::AbortedNoChange @@ -631,15 +659,17 @@ fn terminal_replay_bypasses_preview_expiry_before_policy_or_native_execution() { assert_eq!(policy_calls.load(Ordering::SeqCst), 0); } -#[test] -fn unstarted_apply_reports_expired_private_input_without_entering_native_git() { - let (_database, store) = test_store(); +#[tokio::test] +async fn unstarted_apply_reports_expired_private_input_without_entering_native_git() { + let (_database, store) = test_store().await; let preview = preview_with_expiry(fixture_time(14)); store .save_preview_input(preview_input(&preview)) + .await .expect("save preview input"); store .save_preview(preview.clone()) + .await .expect("save public preview"); let mut request = apply_request(&preview, "idempotency.expired-unstarted"); request.observed_at = fixture_time(14); @@ -654,22 +684,24 @@ fn unstarted_apply_reports_expired_private_input_without_entering_native_git() { ); assert_eq!( - port.apply(&request), + port.apply(&request).await, Err(GitIndexTransactionPortError::ExpiredPreview) ); assert_eq!(apply_calls.load(Ordering::SeqCst), 0); } -#[test] -fn ambiguous_boundary_reconciles_once_without_replaying_native_apply() { +#[tokio::test] +async fn ambiguous_boundary_reconciles_once_without_replaying_native_apply() { let harness = test_port( [NativeMode::CommitBoundaryUnknown], [RecoveryMode::AbortedNoChange], - ); + ) + .await; let result = harness .port .apply(&harness.request) + .await .expect("reconciliation result"); assert_eq!( result.receipt.outcome, @@ -679,16 +711,18 @@ fn ambiguous_boundary_reconciles_once_without_replaying_native_apply() { assert_eq!(harness.recovery_calls.load(Ordering::SeqCst), 1); } -#[test] -fn needs_inspection_quarantines_new_keys_but_allows_terminal_replay() { +#[tokio::test] +async fn needs_inspection_quarantines_new_keys_but_allows_terminal_replay() { let harness = test_port( [NativeMode::CommitBoundaryUnknown], [RecoveryMode::NeedsInspection], - ); + ) + .await; let first = harness .port .apply(&harness.request) + .await .expect("inspection receipt"); assert_eq!( first.receipt.outcome, @@ -697,24 +731,26 @@ fn needs_inspection_quarantines_new_keys_but_allows_terminal_replay() { let replay = harness .port .apply(&harness.request) + .await .expect("terminal replay remains available"); assert_eq!(replay.receipt, first.receipt); let new_key = apply_request(&harness.preview, "idempotency.new-key"); assert_eq!( - harness.port.apply(&new_key), + harness.port.apply(&new_key).await, Err(GitIndexTransactionPortError::RecoveryRequired) ); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); assert_eq!(harness.recovery_calls.load(Ordering::SeqCst), 1); } -#[test] -fn indeterminate_recovery_keeps_durable_quarantine_without_native_replay() { - let harness = test_port([NativeMode::CommitBoundaryUnknown], [RecoveryMode::Error]); +#[tokio::test] +async fn indeterminate_recovery_keeps_durable_quarantine_without_native_replay() { + let harness = test_port([NativeMode::CommitBoundaryUnknown], [RecoveryMode::Error]).await; let first = harness .port .apply(&harness.request) + .await .expect("unobservable recovery terminalizes inspection"); assert_eq!( first.receipt.outcome, @@ -724,23 +760,24 @@ fn indeterminate_recovery_keeps_durable_quarantine_without_native_replay() { let replay = harness .port .apply(&harness.request) + .await .expect("inspection receipt replays"); assert_eq!(replay.receipt, first.receipt); let new_key = apply_request(&harness.preview, "idempotency.indeterminate-new-key"); assert_eq!( - harness.port.apply(&new_key), + harness.port.apply(&new_key).await, Err(GitIndexTransactionPortError::RecoveryRequired) ); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); assert_eq!(harness.recovery_calls.load(Ordering::SeqCst), 1); } -#[test] -fn nonterminal_key_is_recovery_only_and_startup_recovery_is_idempotent() { - let (_database, store) = test_store(); +#[tokio::test] +async fn nonterminal_key_is_recovery_only_and_startup_recovery_is_idempotent() { + let (_database, store) = test_store().await; let preview = preview(); let request = apply_request(&preview, "idempotency.fixture"); - seed_prepared_transaction(&store, &preview, &request); + seed_prepared_transaction(&store, &preview, &request).await; let native = FakeNative::new([], [RecoveryMode::AbortedNoChange]); let apply_calls = Arc::clone(&native.apply_calls); let recovery_calls = Arc::clone(&native.recovery_calls); @@ -754,12 +791,13 @@ fn nonterminal_key_is_recovery_only_and_startup_recovery_is_idempotent() { ); assert_eq!( - port.apply(&request), + port.apply(&request).await, Err(GitIndexTransactionPortError::RecoveryRequired) ); assert_eq!(apply_calls.load(Ordering::SeqCst), 0); let recovered = port .recover_startup(fixture_time(20)) + .await .expect("startup recovery"); assert_eq!(recovered.len(), 1); assert_eq!( @@ -768,6 +806,7 @@ fn nonterminal_key_is_recovery_only_and_startup_recovery_is_idempotent() { ); assert_eq!( port.recover_startup(fixture_time(21)) + .await .expect("idempotent startup recovery") .len(), 0 @@ -775,12 +814,12 @@ fn nonterminal_key_is_recovery_only_and_startup_recovery_is_idempotent() { assert_eq!(recovery_calls.load(Ordering::SeqCst), 1); } -#[test] -fn startup_service_recovers_before_exposing_the_mutation_port() { - let (_database, store) = test_store(); +#[tokio::test] +async fn startup_service_recovers_before_exposing_the_mutation_port() { + let (_database, store) = test_store().await; let preview = preview(); let request = apply_request(&preview, "idempotency.startup-service"); - seed_prepared_transaction(&store, &preview, &request); + seed_prepared_transaction(&store, &preview, &request).await; let native = FakeNative::new([], [RecoveryMode::AbortedNoChange]); let apply_calls = Arc::clone(&native.apply_calls); let recovery_calls = Arc::clone(&native.recovery_calls); @@ -793,11 +832,13 @@ fn startup_service_recovers_before_exposing_the_mutation_port() { Arc::new(RepositoryMutationQueue::default()), fixture_time(20), ) + .await .expect("startup recovery completes before service publication"); assert_eq!(recovery_calls.load(Ordering::SeqCst), 1); let replay = service .apply(&request) + .await .expect("recovered terminal receipt is immediately replayable"); assert_eq!( replay.receipt.outcome, @@ -806,9 +847,9 @@ fn startup_service_recovers_before_exposing_the_mutation_port() { assert_eq!(apply_calls.load(Ordering::SeqCst), 0); } -#[test] -fn startup_recovery_failure_returns_no_mutation_service() { - let (_database, store) = test_store(); +#[tokio::test] +async fn startup_recovery_failure_returns_no_mutation_service() { + let (_database, store) = test_store().await; let store = StartupUnavailableStore(store); let native = FakeNative::new([], []); let apply_calls = Arc::clone(&native.apply_calls); @@ -820,7 +861,8 @@ fn startup_recovery_failure_returns_no_mutation_service() { TestPolicy::allowing(), Arc::new(RepositoryMutationQueue::default()), fixture_time(20), - ); + ) + .await; assert!(matches!( result, @@ -829,28 +871,31 @@ fn startup_recovery_failure_returns_no_mutation_service() { assert_eq!(apply_calls.load(Ordering::SeqCst), 0); } -#[test] -fn quarantine_clears_only_after_a_proven_recovery_receipt() { +#[tokio::test] +async fn quarantine_clears_only_after_a_proven_recovery_receipt() { let harness = test_port( [NativeMode::CommitBoundaryUnknown], [RecoveryMode::NeedsInspection, RecoveryMode::AbortedNoChange], - ); + ) + .await; let receipt = harness .port .apply(&harness.request) + .await .expect("inspection receipt") .receipt; assert_eq!(receipt.outcome, GitIndexReceiptOutcomeV1::NeedsInspection); let new_key = apply_request(&harness.preview, "idempotency.blocked"); assert_eq!( - harness.port.apply(&new_key), + harness.port.apply(&new_key).await, Err(GitIndexTransactionPortError::RecoveryRequired) ); let cleared = harness .port .recover_startup(fixture_time(20)) + .await .expect("proven clear"); assert_eq!(cleared.len(), 1); assert_eq!( @@ -860,6 +905,7 @@ fn quarantine_clears_only_after_a_proven_recovery_receipt() { let admitted = harness .port .apply(&new_key) + .await .expect("new key after proven clear"); assert_eq!( admitted.receipt.outcome, @@ -903,7 +949,7 @@ async fn daemon_owner_reuses_one_service_for_the_same_project_database() { } #[tokio::test] -async fn daemon_owner_shutdown_fences_admission_clears_services_and_joins_store_actor() { +async fn daemon_owner_shutdown_fences_admission_clears_services_and_closes_store() { let directory = tempfile::tempdir().expect("project directory"); let database = RegisteredGlobalDbHarness::open("git-index-owner-shutdown").await; let registry = DaemonGitIndexTransactionServiceRegistry::new(test_catalog_snapshot); @@ -950,7 +996,8 @@ async fn daemon_owner_shutdown_fences_admission_clears_services_and_joins_store_ assert_eq!( owner .service - .read_preview(&GitIndexPreviewId::new("preview.after-shutdown").unwrap()), + .read_preview(&GitIndexPreviewId::new("preview.after-shutdown").unwrap()) + .await, Err(GitIndexTransactionPortError::DaemonUnavailable) ); } @@ -977,7 +1024,7 @@ async fn daemon_owner_racing_a_cold_mount_is_absent_until_published_with_authori privacy_digest: digest('4'), evaluated_at: fixture_time(20), }; - let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>(); let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); let mount = tokio::spawn({ let registry = Arc::clone(®istry); @@ -996,12 +1043,10 @@ async fn daemon_owner_racing_a_cold_mount_is_absent_until_published_with_authori project_id, source, fixture_time(20), - move || { + async move { entered_tx.send(()).expect("mount reached the store open"); - release_rx.recv().expect("test releases the store open"); - DaemonGitIndexTransactionStore::open(store_database).map(|store| { - SharedDaemonGitIndexTransactionStore::from_arc(Arc::new(store)) - }) + release_rx.await.expect("test releases the store open"); + DaemonGitIndexTransactionStore::open(store_database).await }, ) .await diff --git a/crates/tracedecay-code-index-runtime/src/git_transactions/transport_tests.rs b/crates/tracedecay-code-index-runtime/src/git_transactions/transport_tests.rs index 16c71dc7c1..96358c0301 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/transport_tests.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/transport_tests.rs @@ -12,18 +12,20 @@ const EFFECTS: [GitIndexTransactionOperationV1; 2] = [ GitIndexTransactionOperationV1::UnstageHunks, ]; -#[test] -fn preview_apply_owner_mounts_all_git_effects_and_replays() { +#[tokio::test] +async fn preview_apply_owner_mounts_all_git_effects_and_replays() { for operation in EFFECTS { let harness = test_port_from_preview( operation, [NativeMode::Completed(GitIndexReceiptOutcomeV1::Committed)], [], - ); + ) + .await; assert_eq!(harness.preview.operation, operation); let first = harness .port .apply(&harness.request) + .await .expect("effect reaches the preview owner"); assert_eq!(first.receipt.operation, operation); assert_eq!(first.receipt.outcome, GitIndexReceiptOutcomeV1::Committed); @@ -31,44 +33,48 @@ fn preview_apply_owner_mounts_all_git_effects_and_replays() { let replay = harness .port .apply(&harness.request) + .await .expect("same input replays the durable receipt"); assert_eq!(replay.receipt, first.receipt); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); } } -#[test] -fn preview_apply_owner_rejects_a_conflicting_preview_cas() { +#[tokio::test] +async fn preview_apply_owner_rejects_a_conflicting_preview_cas() { for operation in EFFECTS { let harness = test_port_from_preview( operation, [NativeMode::Completed(GitIndexReceiptOutcomeV1::Committed)], [], - ); + ) + .await; harness .port .apply(&harness.request) + .await .expect("seed durable receipt for the CAS check"); let mut conflicting = harness.request.clone(); conflicting.preview_digest = digest('f'); assert_eq!( - harness.port.apply(&conflicting), + harness.port.apply(&conflicting).await, Err(GitIndexTransactionPortError::IdempotencyConflict) ); assert_eq!(harness.apply_calls.load(Ordering::SeqCst), 1); } } -#[test] -fn preview_apply_owner_honors_cancellation_before_native_entry() { +#[tokio::test] +async fn preview_apply_owner_honors_cancellation_before_native_entry() { for operation in EFFECTS { - let harness = test_port_from_preview(operation, [], []); + let harness = test_port_from_preview(operation, [], []).await; let checks = AtomicUsize::new(0); let result = harness .port .apply_cancellable(&harness.request, || { (checks.fetch_add(1, Ordering::SeqCst) + 1 >= 5).then_some(UtcMicros(25)) }) + .await .expect("cancellation is a durable no-change result"); assert_eq!( result.receipt.outcome, @@ -82,17 +88,19 @@ fn preview_apply_owner_honors_cancellation_before_native_entry() { } } -#[test] -fn preview_apply_owner_recovers_an_ambiguous_boundary_without_replay() { +#[tokio::test] +async fn preview_apply_owner_recovers_an_ambiguous_boundary_without_replay() { for operation in EFFECTS { let harness = test_port_from_preview( operation, [NativeMode::CommitBoundaryUnknown], [RecoveryMode::AbortedNoChange], - ); + ) + .await; let result = harness .port .apply(&harness.request) + .await .expect("recovery result"); assert_eq!( result.receipt.outcome, diff --git a/crates/tracedecay-contracts/src/git/transactions.rs b/crates/tracedecay-contracts/src/git/transactions.rs index a8ad9db625..8327bbae7a 100644 --- a/crates/tracedecay-contracts/src/git/transactions.rs +++ b/crates/tracedecay-contracts/src/git/transactions.rs @@ -373,17 +373,17 @@ pub trait GitIndexTransactionPort { fn preview( &self, request: &GitIndexPreviewRequestV1, - ) -> Result; + ) -> impl Future> + Send; fn apply( &self, request: &GitIndexApplyRequestV1, - ) -> Result; + ) -> impl Future> + Send; fn recover( &self, request: &GitIndexRecoveryRequestV1, - ) -> Result; + ) -> impl Future> + Send; } /// Stable, transport-neutral failure taxonomy for the daemon/native port. @@ -424,12 +424,12 @@ where } #[tracing::instrument(name = "application.git.index.preview", level = "trace", skip_all)] - pub fn preview( + pub async fn preview( &self, request: GitIndexPreviewRequestV1, ) -> Result, GitIndexTransactionApplicationError> { request.validate()?; - let result = self.port.preview(&request)?; + let result = self.port.preview(&request).await?; result.validate_for(&request)?; let preview = result.preview; Ok(PreviewResult::new( @@ -444,13 +444,13 @@ where } #[tracing::instrument(name = "application.git.index.apply", level = "trace", skip_all)] - pub fn apply( + pub async fn apply( &self, request: GitIndexApplyRequestV1, ) -> Result, GitIndexTransactionApplicationError> { request.validate()?; - let result = self.port.apply(&request)?; + let result = self.port.apply(&request).await?; result.validate_for(&request)?; let input_digest = request.input_digest()?; let expected_state = result.receipt.old_snapshot_digest.clone(); @@ -492,7 +492,7 @@ where } #[tracing::instrument(name = "application.git.index.recover", level = "trace", skip_all)] - pub fn recover( + pub async fn recover( &self, request: GitIndexRecoveryRequestV1, ) -> Result { @@ -504,7 +504,7 @@ where .transaction_id .validate() .map_err(ApplicationContractError::from)?; - let receipt = self.port.recover(&request)?; + let receipt = self.port.recover(&request).await?; receipt.validate().map_err(ApplicationContractError::from)?; Ok(receipt) } diff --git a/crates/tracedecay-daemon-service/src/invocation/git.rs b/crates/tracedecay-daemon-service/src/invocation/git.rs index 89522797d5..ee08c7b6f0 100644 --- a/crates/tracedecay-daemon-service/src/invocation/git.rs +++ b/crates/tracedecay-daemon-service/src/invocation/git.rs @@ -412,22 +412,8 @@ pub(super) async fn execute_git_read( ); } }; - let service = Arc::clone(&owner.service); - match tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - service.save_preview_input(input) - }) - .await - { - Ok(Ok(())) => {} - Ok(Err(error)) => { - return application_problem(wire_request_id, map_git_port_problem(error)); - } - Err(_) => { - return DaemonInvocationResponse::problem( - wire_request_id, - DaemonInvocationProblem::Unavailable, - ); - } + if let Err(error) = Box::pin(owner.service.save_preview_input(input)).await { + return application_problem(wire_request_id, map_git_port_problem(error)); } } let packet = match git_read_evidence_packet( @@ -518,9 +504,9 @@ pub(super) async fn execute_git_preview( // this async state machine: constructing that future on the socket poll // stack overflows. Keep every payload on a blocking thread or the heap. let join_id = wire_request_id.clone(); - let prepared = match tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - prepare_git_preview(wire_request_id, owner, request, deadline, cancellation) - }) + let prepared = match tracedecay_runtime_core::native_runtime::on_blocking_pool( + prepare_git_preview(wire_request_id, owner, request, deadline, cancellation), + ) .await { Ok(Ok(prepared)) => prepared, @@ -550,9 +536,9 @@ pub(super) async fn execute_git_apply( return missing_registered_owner_problem(publication, wire_request_id); }; let join_id = wire_request_id.clone(); - let prepared = match tracedecay_runtime_core::native_runtime::spawn_blocking(move || { - prepare_git_apply(wire_request_id, owner, request, deadline, cancellation) - }) + let prepared = match tracedecay_runtime_core::native_runtime::on_blocking_pool( + prepare_git_apply(wire_request_id, owner, request, deadline, cancellation), + ) .await { Ok(Ok(prepared)) => prepared, @@ -579,7 +565,7 @@ struct PreparedGitApply { request: GitIndexApplyRequestV1, } -fn prepare_git_preview( +async fn prepare_git_preview( wire_request_id: String, owner: DaemonGitInvocationOwner, request: GitPreviewSurfaceRequest, @@ -606,7 +592,9 @@ fn prepare_git_preview( invalid_git_request(), ))); }; - let preview_input = match service.read_preview_input(&preview_input_id, authority.evaluated_at) + let preview_input = match service + .read_preview_input(&preview_input_id, authority.evaluated_at) + .await { Ok(input) if input.operation == operation => input, Ok(_) => return Err(Box::new(concealed_application_problem(wire_request_id))), @@ -636,7 +624,7 @@ fn prepare_git_preview( })) } -fn prepare_git_apply( +async fn prepare_git_apply( wire_request_id: String, owner: DaemonGitInvocationOwner, request: GitApplySurfaceRequest, @@ -644,12 +632,15 @@ fn prepare_git_apply( cancellation: CancellationContext, ) -> Result, Box> { let service = Arc::clone(&owner.service); - let preview = service.read_preview(&request.preview_id).map_err(|error| { - Box::new(application_problem( - wire_request_id.clone(), - map_git_port_problem(error), - )) - })?; + let preview = service + .read_preview(&request.preview_id) + .await + .map_err(|error| { + Box::new(application_problem( + wire_request_id.clone(), + map_git_port_problem(error), + )) + })?; if preview.preview_digest != request.preview_digest || preview.repository_snapshot.project_id != owner.project_id { @@ -702,12 +693,13 @@ async fn settle_prepared_git_preview( let started_at = prepared.request.observed_at; let effective_deadline = prepared.request.context.deadline().clone(); let wire_request_id = prepared.wire_request_id.clone(); - let result = tracedecay_runtime_core::native_runtime::spawn_blocking(move || { + let result = tracedecay_runtime_core::native_runtime::on_blocking_pool(async move { GitIndexTransactionService::new(SharedGitTransactionPort { service: prepared.service, cancellation: None, }) .preview(prepared.request) + .await }) .await; let response = match result { @@ -754,12 +746,13 @@ async fn settle_prepared_git_apply( let effective_deadline = prepared.request.context.deadline().clone(); let wire_request_id = prepared.wire_request_id.clone(); let cancellation = emitter.clone(); - let result = tracedecay_runtime_core::native_runtime::spawn_blocking(move || { + let result = tracedecay_runtime_core::native_runtime::on_blocking_pool(async move { GitIndexTransactionService::new(SharedGitTransactionPort { service: prepared.service, cancellation: Some(cancellation), }) .apply(prepared.request) + .await }) .await; let response = match result { diff --git a/crates/tracedecay-daemon-service/src/invocation/types.rs b/crates/tracedecay-daemon-service/src/invocation/types.rs index 21a07a4b6f..92784f507b 100644 --- a/crates/tracedecay-daemon-service/src/invocation/types.rs +++ b/crates/tracedecay-daemon-service/src/invocation/types.rs @@ -1440,31 +1440,32 @@ pub(super) struct SharedGitTransactionPort { } impl GitIndexTransactionPort for SharedGitTransactionPort { - fn preview( + async fn preview( &self, request: &GitIndexPreviewRequestV1, ) -> Result { - self.service.preview(request) + self.service.preview(request).await } - fn apply( + async fn apply( &self, request: &GitIndexApplyRequestV1, ) -> Result { - self.cancellation.as_ref().map_or_else( - || self.service.apply(request), - |emitter| { + match &self.cancellation { + None => self.service.apply(request).await, + Some(emitter) => { self.service .apply_cancellable(request, || emitter.cancellation_requested_at()) - }, - ) + .await + } + } } - fn recover( + async fn recover( &self, request: &GitIndexRecoveryRequestV1, ) -> Result { - self.service.recover(request) + self.service.recover(request).await } } diff --git a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs index 6062c3b582..ddc80e3d2d 100644 --- a/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs +++ b/crates/tracedecay-rusqlite-runtime/src/native_runtime.rs @@ -19,7 +19,7 @@ use std::time::{Duration, Instant}; use asupersync::runtime::{JoinError, RuntimeHandle}; pub use asupersync::runtime::{Runtime, RuntimeBuilder}; -pub use crate::exact_sql::native_thread_writer_waits; +pub use crate::exact_sql::{native_thread_reader_waits, native_thread_writer_waits}; use asupersync::types::PanicPayload; use asupersync::{Budget, Cx}; use futures_util::future::{AbortHandle, Aborted, BoxFuture, FutureExt, Shared, abortable}; @@ -123,6 +123,20 @@ where spawn_blocking(move || poll_on_current_thread(future)) } +/// Polls `future` on one blocking-pool thread wherever the caller runs. +/// +/// For futures that interleave awaits with synchronous work (native Git, a +/// synchronous port) that must not hold any async worker: unlike +/// [`off_native_thread`], a caller on a Tokio or other foreign executor does +/// not poll in place. Dropping the returned future abandons only the wait. +pub fn on_blocking_pool(future: F) -> BoxFuture<'static, Result> +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + spawn_blocking(move || poll_on_current_thread(future)) +} + fn poll_on_current_thread(future: F) -> F::Output { struct Unpark(std::thread::Thread); impl std::task::Wake for Unpark { @@ -963,6 +977,36 @@ mod tests { assert_eq!(tokio_runtime_missing_total(), before_total + 1); } + #[tokio::test] + async fn on_blocking_pool_never_polls_on_a_tokio_caller_thread() { + let caller = std::thread::current().id(); + let in_place = off_native_thread(async { std::thread::current().id() }) + .await + .expect("in-place poll"); + assert_eq!( + in_place, caller, + "off_native_thread polls a Tokio caller in place" + ); + + let polled_on = on_blocking_pool(async { + sleep_until(Instant::now() + Duration::from_millis(1)) + .await + .expect("native timer"); + (std::thread::current().id(), ON_BLOCKING_POOL.get()) + }) + .await + .expect("blocking-pool poll"); + + assert_ne!( + polled_on.0, caller, + "the future must leave the Tokio worker" + ); + assert!( + polled_on.1, + "the future must be polled on the native blocking pool" + ); + } + #[tokio::test(flavor = "multi_thread")] async fn timeout_returns_a_value_that_lands_before_the_deadline() { let output = timeout(HANG_GUARD, async { diff --git a/crates/tracedecay-store/src/git_index_transactions.rs b/crates/tracedecay-store/src/git_index_transactions.rs index 99bca03b37..6aa9c3b7d5 100644 --- a/crates/tracedecay-store/src/git_index_transactions.rs +++ b/crates/tracedecay-store/src/git_index_transactions.rs @@ -227,65 +227,70 @@ impl GitIndexTransactionTerminalWriteV1 { } /// Append-only preview/receipt store with a mutable, compare-and-swap journal. -pub trait GitIndexTransactionStore { +pub trait GitIndexTransactionStore: Sync { fn save_preview_input( &self, input: GitIndexPreviewInputV1, - ) -> GitIndexTransactionStoreResult<()>; + ) -> impl Future> + Send; fn read_preview_input( &self, preview_id: &GitIndexPreviewId, observed_at: UtcMicros, - ) -> GitIndexTransactionStoreResult; + ) -> impl Future> + Send; fn purge_expired_preview_inputs( &self, observed_at: UtcMicros, limit: usize, - ) -> GitIndexTransactionStoreResult; + ) -> impl Future> + Send; - fn save_preview(&self, preview: GitIndexPreviewV1) -> GitIndexTransactionStoreResult<()>; + fn save_preview( + &self, + preview: GitIndexPreviewV1, + ) -> impl Future> + Send; fn read_preview( &self, preview_id: &tracedecay_domain::GitIndexPreviewId, - ) -> GitIndexTransactionStoreResult>; + ) -> impl Future>> + Send; fn read_record( &self, idempotency_key: &GitIndexIdempotencyKey, - ) -> GitIndexTransactionStoreResult>; + ) -> impl Future>> + Send; fn begin_or_replay( &self, request: GitIndexTransactionBeginRequestV1, - ) -> GitIndexTransactionStoreResult; + ) -> impl Future> + Send; fn compare_and_swap_journal( &self, idempotency_key: &GitIndexIdempotencyKey, expected_phase_epoch: u64, replacement: GitIndexTransactionJournalV1, - ) -> GitIndexTransactionStoreResult; + ) -> impl Future> + Send; fn write_terminal( &self, write: GitIndexTransactionTerminalWriteV1, - ) -> GitIndexTransactionStoreResult; + ) -> impl Future> + Send; fn recovery_candidates( &self, repository_id: &RepositoryId, - ) -> GitIndexTransactionStoreResult>; + ) -> impl Future>> + Send; - fn recovery_repositories(&self) -> GitIndexTransactionStoreResult>; + fn recovery_repositories( + &self, + ) -> impl Future>> + Send; fn quarantine_repository( &self, repository_id: &RepositoryId, transaction_id: &tracedecay_domain::GitIndexTransactionId, - ) -> GitIndexTransactionStoreResult<()>; + ) -> impl Future> + Send; /// Clear one active repository quarantine only with a fresh, native /// recovery proof. Implementations retain the proof durably so a later @@ -295,5 +300,5 @@ pub trait GitIndexTransactionStore { repository_id: &RepositoryId, transaction_id: &tracedecay_domain::GitIndexTransactionId, recovery_receipt: GitIndexTransactionReceiptV1, - ) -> GitIndexTransactionStoreResult<()>; + ) -> impl Future> + Send; } diff --git a/crates/tracedecay/src/daemon/project_open_owners/git_catalog_tests.rs b/crates/tracedecay/src/daemon/project_open_owners/git_catalog_tests.rs index 4467e7d571..98ec0e443b 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/git_catalog_tests.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/git_catalog_tests.rs @@ -153,8 +153,8 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() let before = git(&project_root, &["rev-parse", "HEAD"]); std::fs::write(project_root.join("file.txt"), "accepted\n").unwrap(); let index_tree_before = git(&project_root, &["write-tree"]); - let accepted = transaction_preview(&owner, "accepted", operation); - let accepted_result = owner.service.apply(&accepted).unwrap(); + let accepted = transaction_preview(&owner, "accepted", operation).await; + let accepted_result = owner.service.apply(&accepted).await.unwrap(); assert_eq!( accepted_result.receipt.outcome, GitIndexReceiptOutcomeV1::Committed @@ -164,13 +164,13 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() assert_eq!(git(&project_root, &["rev-parse", "HEAD"]), before); std::fs::write(project_root.join("file.txt"), "previewed\n").unwrap(); - let stale = transaction_preview(&owner, "stale", operation); + let stale = transaction_preview(&owner, "stale", operation).await; std::fs::write(project_root.join("file.txt"), "changed\n").unwrap(); let head_before_stale = git(&project_root, &["rev-parse", "HEAD"]); let index_before_stale = std::fs::read(project_root.join(".git/index")).unwrap(); // Repository revalidation happens after durable admission, so refusal // must return a terminal no-change receipt. - let stale_result = owner.service.apply(&stale).unwrap(); + let stale_result = owner.service.apply(&stale).await.unwrap(); stale_result.validate_for(&stale).unwrap(); assert_eq!( stale_result.receipt.outcome, @@ -192,7 +192,7 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() std::fs::read(project_root.join("file.txt")).unwrap(), b"changed\n" ); - let denied = transaction_preview(&owner, "revoked", operation); + let denied = transaction_preview(&owner, "revoked", operation).await; let before_denial = git(&project_root, &["rev-parse", "HEAD"]); let index_before_denial = std::fs::read(project_root.join(".git/index")).unwrap(); let mut revoked = access.clone(); @@ -210,7 +210,7 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() owner.current_authority(operation), Err(GitIndexTransactionPortError::PolicyDenied) )); - let denied_result = owner.service.apply(&denied).unwrap(); + let denied_result = owner.service.apply(&denied).await.unwrap(); denied_result.validate_for(&denied).unwrap(); assert_eq!( denied_result.receipt.outcome, @@ -246,8 +246,8 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() && current.grant_expires_at.0 <= current.evaluated_at.0 + horizon, "an authorized request keeps a finite grant expiry: {current:?}" ); - let day_old_preview = transaction_preview(&owner, "day-old", operation); - let day_old_result = owner.service.apply(&day_old_preview).unwrap(); + let day_old_preview = transaction_preview(&owner, "day-old", operation).await; + let day_old_result = owner.service.apply(&day_old_preview).await.unwrap(); assert_eq!( day_old_result.receipt.outcome, GitIndexReceiptOutcomeV1::Committed @@ -266,7 +266,7 @@ fn git(root: &std::path::Path, arguments: &[&str]) -> Vec { output.stdout } -fn transaction_preview( +async fn transaction_preview( owner: &tracedecay_code_index_runtime::git_transactions::DaemonGitInvocationOwner, suffix: &str, operation: GitIndexTransactionOperationV1, @@ -315,7 +315,7 @@ fn transaction_preview( expires_at, ) .unwrap(); - owner.service.save_preview_input(input).unwrap(); + owner.service.save_preview_input(input).await.unwrap(); let grant = CapabilityGrantSnapshot::new( CapabilityGrantId::new(format!("grant.catalog.{suffix}")).unwrap(), current.policy_revision, @@ -361,6 +361,7 @@ fn transaction_preview( selected_hunks: hunks, observed_at: current.evaluated_at, }) + .await .unwrap() .preview; assert_eq!( diff --git a/crates/tracedecay/src/daemon/tests/bootstrap.rs b/crates/tracedecay/src/daemon/tests/bootstrap.rs index feccb0912b..7e1be80cce 100644 --- a/crates/tracedecay/src/daemon/tests/bootstrap.rs +++ b/crates/tracedecay/src/daemon/tests/bootstrap.rs @@ -1941,8 +1941,8 @@ async fn shutdown_fences_git_index_transactions_and_joins_store_actors() { )); // The idempotent receipt proves engine shutdown already closed the one - // mounted service and joined its store actor thread, so the registered - // session database Arc retained by the actor dropped with + // mounted service and closed its store, so the registered + // session database lease retained by the store dropped with // `shutdown_servers` rather than at process exit. let receipt = registry .shutdown()