refactor(git-index): make transaction store async, drop actor

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<GitIndexTransactionBeginResultV1, GitIndexJournalError> {
         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?)
     }
 }

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<GitIndexPreviewInputV1, GitIndexTransactionPortError> {
-        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<GitIndexPreviewV1, GitIndexTransactionPortError> {
-        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<Self, GitIndexTransactionPortError> {
         let port =
             DaemonGitIndexTransactionPort::new(store, native, classifier, authorization, queue);
-        port.recover_startup(observed_at)?;
+        port.recover_startup(observed_at).await?;
         Ok(Self { port })
     }
 }
@@ -114,44 +114,46 @@ impl<S, N, C, A> DaemonGitIndexTransactionService<S, N, C, A>
 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<UtcMicros>,
+        cancellation_requested: impl Fn() -> Option<UtcMicros> + Sync,
     ) -> Result<GitIndexApplyPortResultV1, GitIndexTransactionPortError> {
-        self.port.apply_cancellable(request, cancellation_requested)
+        self.port
+            .apply_cancellable(request, cancellation_requested)
+            .await
     }
 }
 
 impl<S, N, C, A> GitIndexTransactionPort for DaemonGitIndexTransactionService<S, N, C, A>
 where
     S: GitIndexTransactionStore,
     N: GitIndexNativeExecutor + GitIndexRecoveryExecutor,
-    C: GitEffectClassifier,
+    C: GitEffectClassifier + Sync,
     A: GitIndexPolicyRecheckPort,
 {
-    fn preview(
+    async fn preview(
         &self,
         request: &GitIndexPreviewRequestV1,
     ) -> Result<GitIndexPreviewPortResultV1, GitIndexTransactionPortError> {
-        self.port.preview(request)
+        self.port.preview(request).await
     }
 
-    fn apply(
+    async fn apply(
         &self,
         request: &GitIndexApplyRequestV1,
     ) -> Result<GitIndexApplyPortResultV1, GitIndexTransactionPortError> {
-        self.port.apply(request)
+        self.port.apply(request).await
     }
 
-    fn recover(
+    async fn recover(
         &self,
         request: &GitIndexRecoveryRequestV1,
     ) -> Result<GitIndexTransactionReceiptV1, GitIndexTransactionPortError> {
-        self.port.recover(request)
+        self.port.recover(request).await
     }
 }
 

crates/tracedecay-code-index-runtime/src/git_transactions/native.rs

@@ -1090,7 +1090,7 @@ impl<A> FixedDaemonGitIndexExecutor<A> {
 
 impl<A> GitIndexNativeExecutor for FixedDaemonGitIndexExecutor<A>
 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<A> GitIndexRecoveryExecutor for FixedDaemonGitIndexExecutor<A>
 where
-    A: GitIndexPreviewAssembler,
+    A: GitIndexPreviewAssembler + Sync,
 {
     fn reconcile(
         &self,

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<DaemonGitAuthorityStateV1, GitIndexTransactionPortError> {
         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<DaemonProjectGitIndexPreviewAssembler>,
     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<F>(
+    pub(super) async fn mount_with(
         &self,
         database_path: PathBuf,
         repository_root: PathBuf,
         project_id: ProjectId,
         source: Arc<dyn DaemonGitAuthoritySource>,
         observed_at: UtcMicros,
-        open_store: F,
-    ) -> Result<DaemonGitInvocationOwner, GitIndexTransactionPortError>
-    where
-        F: FnOnce() -> tracedecay_store::GitIndexTransactionStoreResult<
-            super::SharedDaemonGitIndexTransactionStore,
+        open_store: impl Future<
+            Output = tracedecay_store::GitIndexTransactionStoreResult<
+                DaemonGitIndexTransactionStore,
+            >,
         >,
-    {
+    ) -> Result<DaemonGitInvocationOwner, GitIndexTransactionPortError> {
         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
     }
 }