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 67c3b7c0d7..bb2dadee5d 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/owner.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/owner.rs @@ -204,46 +204,34 @@ impl DaemonGitAuthoritySource for ProductionDaemonGitAuthoritySource { } } +enum DaemonGitAuthority { + Installed(Arc), + /// Daemon shutdown revoked the authority of an owner a request may + /// already hold. + Revoked, +} + +/// A slot is only ever constructed around an installed source, so a mounted +/// owner always carries authority until shutdown revokes it. struct DaemonGitAuthoritySlot { - source: ProfiledStdRwLock>>, + source: ProfiledStdRwLock, } -impl Default for DaemonGitAuthoritySlot { - fn default() -> Self { +impl DaemonGitAuthoritySlot { + fn new(source: Arc) -> Self { Self { source: hotpath::rw_lock!( - std::sync::RwLock::new(None), + std::sync::RwLock::new(DaemonGitAuthority::Installed(source)), label = "daemon.git.tx.authority_source" ), } } -} -impl DaemonGitAuthoritySlot { - fn install( - &self, - source: Arc, - ) -> Result<(), GitIndexTransactionPortError> { + fn set(&self, authority: DaemonGitAuthority) -> Result<(), GitIndexTransactionPortError> { *self .source .write() - .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)? = Some(source); - Ok(()) - } - - fn installed(&self) -> Result { - Ok(self - .source - .read() - .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)? - .is_some()) - } - - fn clear(&self) -> Result<(), GitIndexTransactionPortError> { - self.source - .write() - .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)? - .take(); + .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)? = authority; Ok(()) } } @@ -253,13 +241,16 @@ impl DaemonGitAuthoritySource for DaemonGitAuthoritySlot { &self, capability_id: &CapabilityId, ) -> Result { - let source = self + let source = match &*self .source .read() .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)? - .as_ref() - .ok_or(GitIndexTransactionPortError::PolicyDenied)? - .clone(); + { + DaemonGitAuthority::Installed(source) => Arc::clone(source), + DaemonGitAuthority::Revoked => { + return Err(GitIndexTransactionPortError::DaemonUnavailable); + } + }; source.current_capability(capability_id) } } @@ -415,6 +406,17 @@ struct ServiceEntry { authority: Arc, } +impl ServiceEntry { + fn owner(&self) -> DaemonGitInvocationOwner { + DaemonGitInvocationOwner { + project_id: self.project_id.clone(), + repository_root: self.repository_root.clone(), + service: Arc::clone(&self.service), + authority: Arc::clone(&self.authority), + } + } +} + #[derive(Clone, Debug, PartialEq, Eq, Hash)] struct ServiceKey { database_path: PathBuf, @@ -481,40 +483,57 @@ pub struct DaemonGitIndexShutdownReceiptV1 { } impl DaemonGitIndexTransactionServiceRegistry { - #[hotpath::measure(label = "daemon.git.tx.ensure", future = true)] - pub async fn ensure( + /// Mounts the owner for one exact project/worktree with its authority in + /// a single publication: the service starts around an already installed + /// authority and becomes resolvable only once both exist. Remounting the + /// same identity reuses the started service and replaces its authority. + #[hotpath::measure(label = "daemon.git.tx.mount", future = true)] + pub async fn mount( &self, database: RegisteredGlobalDbLeaseV1, repository_root: PathBuf, - project_id: ProjectId, + access: ProjectSourceAccessSnapshot, observed_at: UtcMicros, - ) -> Result, GitIndexTransactionPortError> { - if self.shutdown_fenced.load(Ordering::SeqCst) { - return Err(GitIndexTransactionPortError::DaemonUnavailable); + runtime: tokio::runtime::Handle, + ) -> Result { + let project_id = access.scope.project_id.clone(); + if access.binding.authority != AuthorityRef::Project(project_id.clone()) { + return Err(GitIndexTransactionPortError::PolicyDenied); } // `RegisteredGlobalDb` already carries the canonical path admitted by // its retained runtime authority. Do not rediscover identity through // the filesystem: a newly opened SQLite shard may not have // materialized its directory entry yet. let database_path = database.db_path().to_path_buf(); - self.ensure_with( + let source = Arc::new(ProductionDaemonGitAuthoritySource { + access, + configuration: + OwnedGlobalDbConfigurationControlStore::from_registered_project_runtime_db( + database.clone(), + ), + catalog: self.catalog.clone(), + runtime, + }); + self.mount_with( database_path, repository_root, project_id, + source, observed_at, || self.stores.ensure(database), ) .await } - async fn ensure_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, GitIndexTransactionPortError> + ) -> Result where F: FnOnce() -> tracedecay_store::GitIndexTransactionStoreResult< super::SharedDaemonGitIndexTransactionStore, @@ -525,29 +544,32 @@ impl DaemonGitIndexTransactionServiceRegistry { } let repository_root = canonicalize_repository_root(&repository_root) .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)?; - if let Some(service) = self - .existing(&database_path, &repository_root, &project_id) - .await? - { - return Ok(service); - } + let service_key = ServiceKey::new(&database_path, &project_id, &repository_root); let _creation = self.creation_gate.lock().await; if self.shutdown_fenced.load(Ordering::SeqCst) { return Err(GitIndexTransactionPortError::DaemonUnavailable); } - if let Some(service) = self - .existing(&database_path, &repository_root, &project_id) - .await? { - return Ok(service); + let services = self.services.lock().await; + // Another identity mounted at this root makes the root ambiguous. + if services + .iter() + .any(|(key, entry)| *key != service_key && entry.repository_root == repository_root) + { + return Err(GitIndexTransactionPortError::PolicyDenied); + } + if let Some(entry) = services.get(&service_key) { + entry.authority.set(DaemonGitAuthority::Installed(source))?; + return Ok(entry.owner()); + } } // Open/retain the store actor under the creation gate before native - // recovery runs on a blocking thread. Later ensures reuse this actor. + // recovery runs on a blocking thread. Later mounts reuse this actor. let store = open_store().map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)?; let native_root = repository_root.clone(); - let authority = Arc::new(DaemonGitAuthoritySlot::default()); + 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) = tokio::task::spawn_blocking(move || { @@ -566,71 +588,15 @@ impl DaemonGitIndexTransactionServiceRegistry { }) .await .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)??; - let service = Arc::new(service); - let service_key = ServiceKey::new(&database_path, &project_id, &repository_root); - self.services.lock().await.insert( - service_key, - ServiceEntry { - project_id, - repository_root, - service: Arc::clone(&service), - authority, - }, - ); - Ok(service) - } - - async fn existing( - &self, - database_path: &Path, - repository_root: &Path, - project_id: &ProjectId, - ) -> Result>, GitIndexTransactionPortError> - { - let services = self.services.lock().await; - Ok(services - .get(&ServiceKey::new(database_path, project_id, repository_root)) - .map(|entry| Arc::clone(&entry.service))) - } - - #[hotpath::measure(label = "daemon.git.tx.install_authority", future = true)] - pub async fn install_authority( - &self, - repository_root: &std::path::Path, - access: ProjectSourceAccessSnapshot, - configuration_database: RegisteredGlobalDbLeaseV1, - runtime: tokio::runtime::Handle, - ) -> Result<(), GitIndexTransactionPortError> { - if self.shutdown_fenced.load(Ordering::SeqCst) { - return Err(GitIndexTransactionPortError::DaemonUnavailable); - } - let repository_root = canonicalize_repository_root(repository_root) - .map_err(|_| GitIndexTransactionPortError::DaemonUnavailable)?; - let services = self.services.lock().await; - let mut matches = services - .values() - .filter(|entry| entry.repository_root == repository_root); - let Some(entry) = matches.next() else { - return Err(GitIndexTransactionPortError::PolicyDenied); + let entry = ServiceEntry { + project_id, + repository_root, + service: Arc::new(service), + authority, }; - if matches.next().is_some() - || access.scope.project_id != entry.project_id - || access.binding.authority != AuthorityRef::Project(entry.project_id.clone()) - { - return Err(GitIndexTransactionPortError::PolicyDenied); - } - entry - .authority - .install(Arc::new(ProductionDaemonGitAuthoritySource { - access, - configuration: - OwnedGlobalDbConfigurationControlStore::from_registered_project_runtime_db( - configuration_database, - ), - catalog: self.catalog.clone(), - runtime, - }))?; - Ok(()) + let owner = entry.owner(); + self.services.lock().await.insert(service_key, entry); + Ok(owner) } /// Retires every invocation owner attached to one exact project-session @@ -654,9 +620,6 @@ impl DaemonGitIndexTransactionServiceRegistry { /// Resolve only an owner already mounted by project-open admission. /// Missing and ambiguous roots deliberately share the same outcome. - /// Project open publishes the service before it installs the service's - /// authority; until then the owner is still mounting, and resolving it - /// would answer every read as a policy denial. pub async fn for_repository_root( &self, repository_root: &std::path::Path, @@ -673,15 +636,10 @@ impl DaemonGitIndexTransactionServiceRegistry { let Some(entry) = matches.next() else { return Ok(None); }; - if matches.next().is_some() || !entry.authority.installed()? { + if matches.next().is_some() { return Ok(None); } - Ok(Some(DaemonGitInvocationOwner { - project_id: entry.project_id.clone(), - repository_root: entry.repository_root.clone(), - service: Arc::clone(&entry.service), - authority: Arc::clone(&entry.authority), - })) + Ok(Some(entry.owner())) } #[hotpath::measure(label = "daemon.git.tx.shutdown", future = true)] @@ -696,7 +654,7 @@ impl DaemonGitIndexTransactionServiceRegistry { let services = { let mut retained = self.services.lock().await; for entry in retained.values() { - entry.authority.clear()?; + entry.authority.set(DaemonGitAuthority::Revoked)?; } retained.drain().map(|(_, entry)| entry).collect::>() }; 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 e9afebfc00..37623fc250 100644 --- a/crates/tracedecay-code-index-runtime/src/git_transactions/tests.rs +++ b/crates/tracedecay-code-index-runtime/src/git_transactions/tests.rs @@ -18,8 +18,9 @@ use tracedecay_domain::configuration::{ use tracedecay_domain::{ ActorId, GitIndexIdempotencyKey, GitIndexJournalPhaseV1, GitIndexPreviewId, GitIndexPreviewInputV1, GitIndexPreviewV1, GitIndexReceiptOutcomeV1, GitIndexTransactionId, - GitIndexTransactionJournalV1, GitIndexTransactionReceiptV1, GitOperationStateV1, LocatorDigest, - ManifestDigest, ProjectId, RepositoryId, RepositoryIndexStateV1, UtcMicros, WorktreeId, + GitIndexTransactionJournalV1, GitIndexTransactionOperationV1, GitIndexTransactionReceiptV1, + GitOperationStateV1, LocatorDigest, ManifestDigest, ProjectId, RepositoryId, + RepositoryIndexStateV1, UtcMicros, WorktreeId, }; use tracedecay_policy::{GitConflictRiskV1, GitEffectClassifierV1}; use tracedecay_store::{ @@ -40,7 +41,8 @@ use super::test_support::{ }; use super::{ DaemonGitAuthorityStateV1, DaemonGitIndexTransactionPort, DaemonGitIndexTransactionService, - DaemonGitIndexTransactionServiceRegistry, + DaemonGitIndexTransactionServiceRegistry, DaemonGitInvocationOwner, + SharedDaemonGitIndexTransactionStore, }; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; @@ -865,6 +867,24 @@ fn quarantine_clears_only_after_a_proven_recovery_receipt() { assert_eq!(harness.recovery_calls.load(Ordering::SeqCst), 2); } +async fn mount_owner( + registry: &DaemonGitIndexTransactionServiceRegistry, + database: &RegisteredGlobalDbHarness, + repository_root: &std::path::Path, + project_id: &ProjectId, + ordinal: i64, +) -> Result { + registry + .mount( + database.registered.clone(), + repository_root.to_path_buf(), + source_access(project_id), + fixture_time(ordinal), + tokio::runtime::Handle::current(), + ) + .await +} + #[tokio::test] async fn daemon_owner_reuses_one_service_for_the_same_project_database() { let directory = tempfile::tempdir().expect("project directory"); @@ -872,26 +892,14 @@ async fn daemon_owner_reuses_one_service_for_the_same_project_database() { let registry = DaemonGitIndexTransactionServiceRegistry::new(test_catalog_snapshot); let project_id = id::("project.singleton.fixture"); - let first = registry - .ensure( - database.registered.clone(), - directory.path().to_path_buf(), - project_id.clone(), - fixture_time(20), - ) + let first = mount_owner(®istry, &database, directory.path(), &project_id, 20) .await .expect("first project service"); - let second = registry - .ensure( - database.registered.clone(), - directory.path().to_path_buf(), - project_id, - fixture_time(21), - ) + let second = mount_owner(®istry, &database, directory.path(), &project_id, 21) .await .expect("reused project service"); - assert!(Arc::ptr_eq(&first, &second)); + assert!(Arc::ptr_eq(&first.service, &second.service)); } #[tokio::test] @@ -899,19 +907,27 @@ async fn daemon_owner_shutdown_fences_admission_clears_services_and_joins_store_ let directory = tempfile::tempdir().expect("project directory"); let database = RegisteredGlobalDbHarness::open("git-index-owner-shutdown").await; let registry = DaemonGitIndexTransactionServiceRegistry::new(test_catalog_snapshot); - let project_id = id::("project.shutdown.fixture"); - let service = registry - .ensure( - database.registered.clone(), - directory.path().to_path_buf(), - project_id.clone(), - fixture_time(20), - ) + let project_id = id::("project.singleton.fixture"); + mount_owner(®istry, &database, directory.path(), &project_id, 20) + .await + .expect("project open mounts the owner"); + let owner = registry + .for_repository_root(directory.path()) .await - .expect("project service"); + .expect("owner lookup") + .expect("mounted owner"); let receipt = registry.shutdown().await.expect("Git owner shutdown"); + assert_eq!( + owner + .current_authority(GitIndexTransactionOperationV1::StageHunks) + .err(), + Some(GitIndexTransactionPortError::DaemonUnavailable), + "a request already routed to the owner must see the daemon shutting down, \ + not an authority-missing policy denial" + ); + assert_eq!( receipt, super::owner::DaemonGitIndexShutdownReceiptV1 { @@ -928,22 +944,99 @@ async fn daemon_owner_shutdown_fences_admission_clears_services_and_joins_store_ Err(GitIndexTransactionPortError::DaemonUnavailable) )); assert!(matches!( - registry - .ensure( - database.registered.clone(), - directory.path().to_path_buf(), - project_id, - fixture_time(21), - ) - .await, + mount_owner(®istry, &database, directory.path(), &project_id, 21).await, Err(GitIndexTransactionPortError::DaemonUnavailable) )); assert_eq!( - service.read_preview(&GitIndexPreviewId::new("preview.after-shutdown").unwrap()), + owner + .service + .read_preview(&GitIndexPreviewId::new("preview.after-shutdown").unwrap()), Err(GitIndexTransactionPortError::DaemonUnavailable) ); } +/// A request racing a cold mount resolves either nothing (the caller answers +/// `mounting`) or the owner with its authority, never an owner without one. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn daemon_owner_racing_a_cold_mount_is_absent_until_published_with_authority() { + let directory = tempfile::tempdir().expect("project directory"); + let database = RegisteredGlobalDbHarness::open("git-index-owner-cold-mount").await; + let registry = Arc::new(DaemonGitIndexTransactionServiceRegistry::new( + test_catalog_snapshot, + )); + let project_id = id::("project.singleton.fixture"); + let mounted_authority = DaemonGitAuthorityStateV1 { + scope: source_access(&project_id).scope, + requester: id::("actor.singleton.fixture"), + effective_capabilities: BTreeSet::new(), + grant_expires_at: fixture_time(100), + policy_revision: 7, + policy_digest: digest('1'), + configuration_digest: digest('2'), + catalog_digest: digest('3'), + privacy_digest: digest('4'), + evaluated_at: fixture_time(20), + }; + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let mount = tokio::spawn({ + let registry = Arc::clone(®istry); + let database_path = database.registered.db_path().to_path_buf(); + let store_database = database.registered.clone(); + let repository_root = directory.path().to_path_buf(); + let project_id = project_id.clone(); + let source = Arc::new(MutableDaemonGitAuthority { + current: Mutex::new(mounted_authority.clone()), + }); + async move { + registry + .mount_with( + database_path, + repository_root, + project_id, + source, + fixture_time(20), + 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)) + }) + }, + ) + .await + } + }); + entered_rx.await.expect("mount reached the store open"); + + assert!( + registry + .for_repository_root(directory.path()) + .await + .expect("lookup while mounting") + .is_none(), + "an owner still mounting must not be resolvable" + ); + + release_tx.send(()).expect("release the store open"); + let mounted = mount + .await + .expect("mount task") + .expect("mount publishes the owner"); + let resolved = registry + .for_repository_root(directory.path()) + .await + .expect("lookup after mount") + .expect("published owner"); + assert!(Arc::ptr_eq(&resolved.service, &mounted.service)); + assert_eq!( + resolved + .current_authority(GitIndexTransactionOperationV1::StageHunks) + .map(|authority| (authority.policy_revision, authority.policy_digest)), + Ok((7, digest('1'))) + ); +} + #[tokio::test] async fn daemon_owner_isolates_worktrees_sharing_a_project_database() { let directory = tempfile::tempdir().expect("project directory"); @@ -951,55 +1044,24 @@ async fn daemon_owner_isolates_worktrees_sharing_a_project_database() { let database = RegisteredGlobalDbHarness::open("git-index-owner-worktrees").await; let registry = DaemonGitIndexTransactionServiceRegistry::new(test_catalog_snapshot); let project_id = id::("project.singleton.fixture"); - let primary = registry - .ensure( - database.registered.clone(), - directory.path().to_path_buf(), - project_id.clone(), - fixture_time(20), - ) + let primary = mount_owner(®istry, &database, directory.path(), &project_id, 20) .await .expect("first project service"); - let linked = registry - .ensure( - database.registered.clone(), - alternate.path().to_path_buf(), - project_id.clone(), - fixture_time(21), - ) + let linked = mount_owner(®istry, &database, alternate.path(), &project_id, 21) .await .expect("linked worktree service"); assert!( - !Arc::ptr_eq(&primary, &linked), + !Arc::ptr_eq(&primary.service, &linked.service), "worktrees sharing one store need independent native executors" ); assert!( Arc::ptr_eq( - primary.mutation_queue_for_test(), - linked.mutation_queue_for_test() + primary.service.mutation_queue_for_test(), + linked.service.mutation_queue_for_test() ), "all Git mutation services must serialize through the daemon registry queue" ); - assert!( - registry - .for_repository_root(directory.path()) - .await - .expect("mounting owner lookup") - .is_none(), - "an owner whose authority project open has not installed yet is still mounting" - ); - for root in [directory.path(), alternate.path()] { - registry - .install_authority( - root, - source_access(&project_id), - database.registered.clone(), - tokio::runtime::Handle::current(), - ) - .await - .expect("project open installs each worktree's authority"); - } let primary_owner = registry .for_repository_root(directory.path()) .await @@ -1010,8 +1072,8 @@ async fn daemon_owner_isolates_worktrees_sharing_a_project_database() { .await .expect("linked owner lookup") .expect("linked owner"); - assert!(Arc::ptr_eq(&primary_owner.service, &primary)); - assert!(Arc::ptr_eq(&linked_owner.service, &linked)); + assert!(Arc::ptr_eq(&primary_owner.service, &primary.service)); + assert!(Arc::ptr_eq(&linked_owner.service, &linked.service)); } fn source_access(project_id: &ProjectId) -> ProjectSourceAccessSnapshot { @@ -1056,26 +1118,14 @@ async fn daemon_owner_resolves_symlink_alias_to_the_canonical_mounted_root() { let database = RegisteredGlobalDbHarness::open("git-index-owner-alias").await; let registry = DaemonGitIndexTransactionServiceRegistry::new(test_catalog_snapshot); let project_id = id::("project.alias.fixture"); - let first = registry - .ensure( - database.registered.clone(), - directory.path().to_path_buf(), - project_id.clone(), - fixture_time(30), - ) + let first = mount_owner(®istry, &database, directory.path(), &project_id, 30) .await .expect("mount through real root"); - let second = registry - .ensure( - database.registered.clone(), - alias, - project_id, - fixture_time(31), - ) + let second = mount_owner(®istry, &database, &alias, &project_id, 31) .await .expect("reuse through symlink alias"); assert!( - Arc::ptr_eq(&first, &second), + Arc::ptr_eq(&first.service, &second.service), "symlink alias must resolve to the same mounted owner as the canonical root" ); } diff --git a/crates/tracedecay-daemon-service/src/invocation/git.rs b/crates/tracedecay-daemon-service/src/invocation/git.rs index cba73c6f7f..63ed072206 100644 --- a/crates/tracedecay-daemon-service/src/invocation/git.rs +++ b/crates/tracedecay-daemon-service/src/invocation/git.rs @@ -239,7 +239,7 @@ pub(super) async fn execute_git_read( } let initial = match owner.current_read_authority(&request.request) { Ok(authority) => authority, - Err(_) => return concealed_application_problem(wire_request_id), + Err(error) => return application_problem(wire_request_id, map_git_port_problem(error)), }; let remaining_micros = deadline.expires_at.0.saturating_sub(now_micros().0).max(0) as u64; let bounds = tracedecay_application::git_query::GitQueryBounds { @@ -357,7 +357,7 @@ pub(super) async fn execute_git_read( ); let terminal = match owner.current_read_authority(&request.request) { Ok(authority) => authority, - Err(_) => return concealed_application_problem(wire_request_id), + Err(error) => return application_problem(wire_request_id, map_git_port_problem(error)), }; if initial.scope != terminal.scope || initial.requester != terminal.requester diff --git a/crates/tracedecay/src/daemon.rs b/crates/tracedecay/src/daemon.rs index 8af445b197..32d0492dcf 100644 --- a/crates/tracedecay/src/daemon.rs +++ b/crates/tracedecay/src/daemon.rs @@ -282,10 +282,7 @@ mod core_admission; mod engine; #[cfg(unix)] use engine::DaemonEngine; -use engine::{ - ensure_context_scout_owner_before_advertising, - ensure_git_index_transactions_for_mutation_owners, -}; +use engine::ensure_context_scout_owner_before_advertising; mod core_client; mod core_doctor; mod core_handshake; diff --git a/crates/tracedecay/src/daemon/engine.rs b/crates/tracedecay/src/daemon/engine.rs index 99b47383c1..381e1ad593 100644 --- a/crates/tracedecay/src/daemon/engine.rs +++ b/crates/tracedecay/src/daemon/engine.rs @@ -104,73 +104,6 @@ pub(super) struct DaemonEngine { Arc>>, } -/// Retain one daemon-owned Git index transaction service for the project store -/// and reconcile any durable records before mutation owners become available. -/// Read-only core tools and edit previews do not depend on this service. The -/// service owns the store actor; constructing a second service for the same -/// database is rejected by the registry. -#[hotpath::measure(label = "daemon.engine.git_index_transactions", future = true)] -pub(super) async fn ensure_git_index_transactions_for_mutation_owners( - store_administration: &StoreAdministration, - session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, - project_root: &Path, - project_id: Option<&str>, -) -> Result<()> { - ensure_git_index_transactions_for_mutation_owners_inner( - store_administration, - session_db, - project_root, - project_id, - ) - .await -} - -fn ensure_git_index_transactions_for_mutation_owners_inner<'a>( - store_administration: &'a StoreAdministration, - session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, - project_root: &'a Path, - project_id: Option<&'a str>, -) -> std::pin::Pin> + Send + 'a>> { - // Erase the deeply nested future before it reaches the measured wrapper - // so every profiling feature can compute its layout. - Box::pin(async move { - let Some(project_id) = project_id else { - // Linked/anonymous project opens without a durable project id cannot - // own index-mutation authority; skip rather than invent an identity. - return Ok(()); - }; - let project_id = - tracedecay_domain::ProjectId::new(project_id.to_owned()).map_err(|error| { - TraceDecayError::Config { - message: format!("git index transaction project identity is invalid: {error}"), - } - })?; - let Some(repository_root) = - tracedecay_runtime_core::worktree::git_worktree_root(project_root) - else { - // Non-Git projects remain valid TraceDecay projects. They advertise no - // Git mutation authority and must not fail project-open admission. - return Ok(()); - }; - let observed_at = tracedecay_domain::UtcMicros( - i64::try_from( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |duration| duration.as_micros()), - ) - .unwrap_or(i64::MAX), - ); - store_administration - .git_index_transaction_services() - .ensure(session_db, repository_root, project_id, observed_at) - .await - .map(|_| ()) - .map_err(|error| TraceDecayError::Config { - message: format!("git index transaction startup did not complete: {error}"), - }) - }) -} - #[hotpath::measure(label = "daemon.engine.context_scout.ensure_owner")] pub(super) fn ensure_context_scout_owner_before_advertising( project: &tracedecay_project::project::TraceDecay, diff --git a/crates/tracedecay/src/daemon/project_composition.rs b/crates/tracedecay/src/daemon/project_composition.rs index 7f2f649b0f..7d3ef0014c 100644 --- a/crates/tracedecay/src/daemon/project_composition.rs +++ b/crates/tracedecay/src/daemon/project_composition.rs @@ -613,10 +613,8 @@ struct PublishedFullServer { } /// What the published full server still has to mount before it serves: the -/// project session database its dependent owners open, and the Doctor report -/// reader published once they have. +/// Doctor report reader published once its dependent owners have. struct PendingFullServerOwners { - session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, doctor_report_reader: tracedecay_dashboard_api::DoctorReportReader, } @@ -1448,7 +1446,6 @@ impl ProjectOpenInputs<'_> { Ok(PublishedFullServer { server: full_candidate, pending: PendingFullServerOwners { - session_db, doctor_report_reader, }, }) @@ -1468,7 +1465,6 @@ impl ProjectOpenInputs<'_> { opened: &OpenedProjectGraph, core: &ComposedCoreServer, full_server: &crate::mcp::McpServer, - session_db: tracedecay_global_db::RegisteredGlobalDbLeaseV1, core_source_edit_mutation: Option< Arc, >, @@ -1496,14 +1492,6 @@ impl ProjectOpenInputs<'_> { ) }; self.log_phase("source_edit_preview_ready", None, full_setup_started); - ensure_git_index_transactions_for_mutation_owners( - self.store_administration, - session_db, - self.canonical_project_path, - opened.key.owner.project_id.as_deref(), - ) - .await?; - self.log_phase("git_transactions_ready", None, full_setup_started); let dependent_owners = if opened.project_database_is_read_only { None } else { @@ -1561,7 +1549,6 @@ impl ProjectOpenInputs<'_> { pending: PendingFullServerOwners, ) -> Result<()> { let PendingFullServerOwners { - session_db, doctor_report_reader, } = pending; self.log_phase("session_capabilities_published", None, self.started); @@ -1569,7 +1556,6 @@ impl ProjectOpenInputs<'_> { opened, core, full_server.as_ref(), - session_db, activation.core_source_edit_mutation.clone(), )) .await?; diff --git a/crates/tracedecay/src/daemon/project_open_owners.rs b/crates/tracedecay/src/daemon/project_open_owners.rs index e07e9ab26b..8808b8aae7 100644 --- a/crates/tracedecay/src/daemon/project_open_owners.rs +++ b/crates/tracedecay/src/daemon/project_open_owners.rs @@ -426,26 +426,29 @@ pub(super) async fn register_project_open_production_owners( ); owner_phase_started = Instant::now(); - // One worktree discovery serves both the Git transaction authority here - // and the native-integration mount below. + // One worktree discovery serves both the Git transaction owner here and + // the native-integration mount below. let repository_root = tracedecay_runtime_core::worktree::git_worktree_root(project_root); if let Some(repository_root) = repository_root.as_deref() { + // Mounting reconciles durable Git transaction records before the + // mutation lane below opens. hotpath::future!( - git_transactions.install_authority( - repository_root, - access.clone(), + git_transactions.mount( session_db.clone(), + repository_root.to_path_buf(), + access.clone(), + now_micros(), tokio::runtime::Handle::current(), ), - label = "daemon.project.open.owners.git_authority" + label = "daemon.project.open.owners.git_transactions" ) .await .map_err(|error| TraceDecayError::Config { - message: format!("project-open Git authority registration failed: {error}"), + message: format!("project-open Git transaction owner did not mount: {error}"), })?; } // Preview executors were published with the read-only core. Open their - // mutation lane only after the exact Git transaction authority exists. + // mutation lane only after the exact Git transaction owner exists. // A later failure in this function retires the whole server, and project // open marks the lane failed as it does so, so the lane never stays warming. source_edit_mutation.mark_ready(); 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 d64afd91ea..1f6ae4130c 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 @@ -85,27 +85,11 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() let registry = DaemonGitIndexTransactionServiceRegistry::new(build_application_catalog_snapshot); registry - .ensure( + .mount( database.clone(), project_root.clone(), - project_id.clone(), - tracedecay_contracts::now_micros(), - ) - .await - .unwrap(); - assert!( - registry - .for_repository_root(&project_root) - .await - .unwrap() - .is_none(), - "an owner whose authority project open has not installed yet is still mounting" - ); - registry - .install_authority( - &project_root, access.clone(), - database.clone(), + tracedecay_contracts::now_micros(), tokio::runtime::Handle::current(), ) .await @@ -130,19 +114,11 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() // canonical dependency already retained by the first owner. let independent = DaemonGitIndexTransactionServiceRegistry::new(unavailable_catalog); independent - .ensure( + .mount( database.clone(), project_root.clone(), - project_id, - tracedecay_contracts::now_micros(), - ) - .await - .unwrap(); - independent - .install_authority( - &project_root, access.clone(), - database.clone(), + tracedecay_contracts::now_micros(), tokio::runtime::Handle::current(), ) .await @@ -228,10 +204,11 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() let mut revoked = access.clone(); revoked.effective_capabilities.clear(); registry - .install_authority( - &project_root, - revoked, + .mount( database.clone(), + project_root.clone(), + revoked, + tracedecay_contracts::now_micros(), tokio::runtime::Handle::current(), ) .await @@ -264,10 +241,11 @@ async fn git_owner_uses_explicit_canonical_catalog_and_rechecks_authorization() assert!(issued_at < expired.grant_expires_at); assert!(expired.grant_expires_at < observed_at); registry - .install_authority( - &project_root, - expired, + .mount( database, + project_root.clone(), + expired, + tracedecay_contracts::now_micros(), tokio::runtime::Handle::current(), ) .await