diff --git a/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs b/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs index e855a17c48..c7434ad199 100644 --- a/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs +++ b/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs @@ -263,8 +263,9 @@ pub(super) async fn retire_runtime_work( // store retirement inspects live clients. for server in &servers { let graph = server.cg().await; - crate::daemon::hook_v2_replay_consumer::shutdown_hook_v2_replay_consumer( + crate::daemon::hook_v2_replay_consumer::retire_hook_v2_replay_consumer( &graph.hook_store_layout().data_root, + &Arc::downgrade(&graph), ) .await; } diff --git a/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs b/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs index 57aea13d8f..c9c560d50e 100644 --- a/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs +++ b/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs @@ -402,6 +402,8 @@ struct RegisteredReplayConsumer { delivery_settlements: Weak, task: Option>, + /// This registration's spool wake; its identity is the registration's. + wake: Arc, /// The failures of the last finished sweep; `None` before the first. last_sweep: Option>, } @@ -451,8 +453,11 @@ pub(crate) fn hook_v2_replay_consumer_registered(data_root: &Path) -> bool { }) } -/// Start the per-project replay consumer exactly once per hook data root. -/// Returns `false` when one is already running for this root. +/// Start `graph`'s replay consumer for its hook data root. The newest owner +/// of a data root takes it over: a reopen can register while the retiring +/// owner's server still holds the older graph, and that graph's release must +/// not leave the reopened project without a consumer. Returns `false` when +/// `graph`'s own consumer is already running. pub(crate) fn register_hook_v2_replay_consumer( graph: Arc, delivery_settlements: Arc, @@ -463,10 +468,12 @@ pub(crate) fn register_hook_v2_replay_consumer( let project_root = graph.project_root().to_path_buf(); let graph = Arc::downgrade(&graph); let delivery_settlements = Arc::downgrade(&delivery_settlements); - match registered_replay_roots().lock() { + let wake = Arc::new(tokio::sync::Notify::new()); + let displaced = match registered_replay_roots().lock() { Ok(mut roots) => { if roots.get(&data_root).is_some_and(|consumer| { - consumer.graph.upgrade().is_some() + Weak::ptr_eq(&consumer.graph, &graph) + && consumer.graph.upgrade().is_some() && consumer.delivery_settlements.upgrade().is_some() }) { return false; @@ -477,19 +484,23 @@ pub(crate) fn register_hook_v2_replay_consumer( graph: graph.clone(), delivery_settlements: delivery_settlements.clone(), task: None, + wake: Arc::clone(&wake), last_sweep: None, }, - ); + ) } Err(_) => return false, + }; + if let Some(task) = displaced.and_then(|consumer| consumer.task) { + task.abort(); } let task_data_root = data_root.clone(); let task_graph = graph.clone(); let task_delivery_settlements = delivery_settlements.clone(); let task_project_sessions = project_sessions; let task_background_cpu = background_cpu; - let wake = Arc::new(tokio::sync::Notify::new()); spool_watch::attach_consumer(&data_root, &project_root, Arc::clone(&wake)); + let task_wake = Arc::clone(&wake); let task = tokio::spawn(async move { loop { let (Some(graph_owner), Some(delivery_settlements)) = @@ -525,7 +536,7 @@ pub(crate) fn register_hook_v2_replay_consumer( slept = tracedecay_runtime_core::native_runtime::sleep_until( std::time::Instant::now() + REPLAY_INTERVAL, ) => slept, - () = wake.notified() => Ok(()), + () = task_wake.notified() => Ok(()), } }, tracing::trace_span!("daemon.hook_replay.interval_wait"), @@ -537,18 +548,18 @@ pub(crate) fn register_hook_v2_replay_consumer( break; } } - spool_watch::detach_consumer(&task_data_root); + spool_watch::detach_consumer(&task_data_root, &task_wake); if let Ok(mut roots) = registered_replay_roots().lock() && roots .get(&task_data_root) - .is_some_and(|registered| Weak::ptr_eq(®istered.graph, &task_graph)) + .is_some_and(|registered| Arc::ptr_eq(®istered.wake, &task_wake)) { roots.remove(&task_data_root); } }); match registered_replay_roots().lock() { Ok(mut roots) => match roots.get_mut(&data_root) { - Some(registered) if Weak::ptr_eq(®istered.graph, &graph) => { + Some(registered) if Arc::ptr_eq(®istered.wake, &wake) => { registered.task = Some(task); } _ => task.abort(), @@ -558,18 +569,42 @@ pub(crate) fn register_hook_v2_replay_consumer( true } -/// Stop and join the exact project replay consumer before releasing its graph. +/// Stop and join whichever replay consumer owns `data_root`. For daemon +/// shutdown and project deletion, which end every owner of the data root. pub(crate) async fn shutdown_hook_v2_replay_consumer(data_root: &Path) { - let task = registered_replay_roots() - .lock() - .ok() - .and_then(|mut roots| roots.remove(data_root)) - .and_then(|consumer| consumer.task); - if let Some(task) = task { + stop_replay_consumer(data_root, None).await; +} + +/// Stop and join `owner`'s replay consumer before releasing its graph. A +/// newer owner's consumer for the same data root keeps running. +pub(crate) async fn retire_hook_v2_replay_consumer( + data_root: &Path, + owner: &Weak, +) { + stop_replay_consumer(data_root, Some(owner)).await; +} + +async fn stop_replay_consumer( + data_root: &Path, + owner: Option<&Weak>, +) { + let consumer = { + let mut roots = registered_replay_roots() + .lock() + .unwrap_or_else(PoisonError::into_inner); + let owned = roots.get(data_root).is_some_and(|registered| { + owner.is_none_or(|owner| Weak::ptr_eq(®istered.graph, owner)) + }); + owned.then(|| roots.remove(data_root)).flatten() + }; + let Some(consumer) = consumer else { + return; + }; + spool_watch::detach_consumer(data_root, &consumer.wake); + if let Some(task) = consumer.task { task.abort(); let _ = task.await; } - spool_watch::detach_consumer(data_root); } #[cfg(test)] @@ -609,4 +644,160 @@ mod tests { assert_eq!(spool_root_absent(&file), Ok(false)); assert!(spool_root_absent(&file.join("child")).is_err()); } + + struct ReplayOwnerFixture { + _temp: tempfile::TempDir, + _settlement_runtime: tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime, + project_sessions: tracedecay_global_db::RegisteredGlobalDbLeaseV1, + settlements: Arc, + background_cpu: Arc, + } + + type ProjectGraph = Arc; + + /// Two owners of one project, so both resolve the same hook data root. + async fn replay_owner_fixture() -> (ReplayOwnerFixture, ProjectGraph, ProjectGraph) { + let temp = tempfile::tempdir().expect("replay owner fixture"); + let project = temp.path().join("project"); + let profile = temp.path().join("profile"); + std::fs::create_dir_all(project.join("src")).expect("project source directory"); + std::fs::write(project.join("src/lib.rs"), "pub fn fixture() {}\n").expect("source"); + let options = tracedecay_project::project::TraceDecayOpenOptions { + profile_root: Some(profile.clone()), + global_db_path: Some(profile.join("global.db")), + }; + let older = Arc::new( + tracedecay_project::project::TraceDecay::init_with_options_for_test( + &project, + options.clone(), + ) + .await + .expect("older project owner"), + ); + let newer = Arc::new( + tracedecay_project::project::TraceDecay::open_with_options_for_test(&project, options) + .await + .expect("newer project owner"), + ); + let project_id = tracedecay_domain::ProjectId::new("project.replay.owner").expect("id"); + let settlement_runtime = + tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime::project( + &temp.path().join("settlement-profile"), + &project, + project_id.clone(), + ) + .await + .expect("settlement runtime"); + let project_sessions = settlement_runtime + .project_database_arc() + .expect("project sessions database"); + let identity = tracedecay_application::observability::ObservabilityProducerIdentityV1 { + authorized_scope_ref: project_id.as_str().to_owned(), + process_boot_id: "boot:replay-owner".to_owned(), + producer_revision: "replay-owner-producer.v1".to_owned(), + configuration_revision: "replay-owner-config.v1".to_owned(), + policy_revision: "replay-owner-policy.v1".to_owned(), + }; + let producer = Arc::new( + tracedecay_application::observability::BoundedObservabilityProducerV1::start( + project_sessions.clone(), + identity.clone(), + 8, + ) + .expect("producer"), + ); + let settlements = Arc::new( + tracedecay_application::observability::DeliverySettlementAuthorityV1::new( + project_sessions.clone(), + producer, + identity, + ) + .expect("settlement authority"), + ); + let fixture = ReplayOwnerFixture { + _temp: temp, + _settlement_runtime: settlement_runtime, + project_sessions, + settlements, + background_cpu: Arc::new( + tracedecay_runtime_core::background_cpu::ProcessBackgroundCpuV1::new( + std::num::NonZeroUsize::MIN, + ), + ), + }; + (fixture, older, newer) + } + + impl ReplayOwnerFixture { + fn register(&self, graph: &ProjectGraph) -> bool { + register_hook_v2_replay_consumer( + Arc::clone(graph), + Arc::clone(&self.settlements), + self.project_sessions.clone(), + Arc::clone(&self.background_cpu), + ) + } + } + + async fn assert_newer_owner_survives(data_root: &Path, newer: &ProjectGraph) { + let newer_task_running = registered_replay_roots() + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(data_root) + .is_some_and(|registered| { + Weak::ptr_eq(®istered.graph, &Arc::downgrade(newer)) + && registered + .task + .as_ref() + .is_some_and(|task| !task.is_finished()) + }); + assert!( + newer_task_running, + "the newer owner must own the data root's running replay consumer" + ); + assert!(hook_v2_replay_consumer_registered(data_root)); + assert!( + spool_watch::consumer_attached(data_root), + "the newer owner's consumer must keep the spool wake" + ); + shutdown_hook_v2_replay_consumer(data_root).await; + assert!(!hook_v2_replay_consumer_registered(data_root)); + } + + /// The older owner's graph is gone by the time the project reopens, so + /// the newer owner registers; the older owner's retirement then arrives. + #[tokio::test] + async fn a_late_retirement_leaves_the_newer_owners_replay_consumer_running() { + let (fixture, older, newer) = replay_owner_fixture().await; + let data_root = older.hook_store_layout().data_root.clone(); + assert_eq!(data_root, newer.hook_store_layout().data_root); + assert!(fixture.register(&older), "the older owner starts replay"); + let older_owner = Arc::downgrade(&older); + drop(older); + assert!(fixture.register(&newer), "the reopened owner starts replay"); + + retire_hook_v2_replay_consumer(&data_root, &older_owner).await; + + assert_newer_owner_survives(&data_root, &newer).await; + } + + /// The project reopens while its retiring owner's server still holds the + /// older graph. The reopened owner must take over the data root, or the + /// older graph's release leaves the project with no consumer at all. + #[tokio::test] + async fn a_reopen_takes_over_replay_from_an_owner_still_retiring() { + let (fixture, older, newer) = replay_owner_fixture().await; + let data_root = older.hook_store_layout().data_root.clone(); + assert!(fixture.register(&older), "the older owner starts replay"); + let newer_registered = fixture.register(&newer); + + retire_hook_v2_replay_consumer(&data_root, &Arc::downgrade(&older)).await; + drop(older); + + assert!( + newer_registered, + "the reopened owner's registration must be accepted" + ); + assert_newer_owner_survives(&data_root, &newer).await; + } } diff --git a/crates/tracedecay/src/daemon/hook_v2_replay_consumer/spool_watch.rs b/crates/tracedecay/src/daemon/hook_v2_replay_consumer/spool_watch.rs index 899607f273..0d5c5cb735 100644 --- a/crates/tracedecay/src/daemon/hook_v2_replay_consumer/spool_watch.rs +++ b/crates/tracedecay/src/daemon/hook_v2_replay_consumer/spool_watch.rs @@ -191,8 +191,9 @@ pub(super) fn attach_consumer(data_root: &Path, project_root: &Path, consumer: A ); } -/// Once its consumer stops, a project's next append opens it again. -pub(super) fn detach_consumer(data_root: &Path) { +/// Once its consumer stops, a project's next append opens it again. A newer +/// consumer that already took over the project's wakes keeps them. +pub(super) fn detach_consumer(data_root: &Path, consumer: &Arc) { let Ok(watch) = spool_watch().lock() else { return; }; @@ -201,6 +202,10 @@ pub(super) fn detach_consumer(data_root: &Path) { }; if let Ok(mut targets) = watch.targets.lock() && let Some(watched) = targets.spools.get_mut(data_root) + && watched + .consumer + .as_ref() + .is_some_and(|attached| Arc::ptr_eq(attached, consumer)) { watched.consumer = None; } diff --git a/crates/tracedecay/src/daemon/project_composition.rs b/crates/tracedecay/src/daemon/project_composition.rs index 138540c3d1..295b90e13e 100644 --- a/crates/tracedecay/src/daemon/project_composition.rs +++ b/crates/tracedecay/src/daemon/project_composition.rs @@ -117,10 +117,13 @@ async fn release_one_idle_project_server_before_open( tracedecay_runtime_core::path_safety::canonical_root_identity(&key.project_root) }) .collect::>(); - let mut hook_data_roots = std::collections::BTreeSet::new(); + let mut hook_replay_owners = Vec::with_capacity(retired_servers.len()); for (_, server) in &retired_servers { let graph = server.cg().await; - hook_data_roots.insert(graph.hook_store_layout().data_root.clone()); + hook_replay_owners.push(( + graph.hook_store_layout().data_root.clone(), + Arc::downgrade(&graph), + )); } let retired_servers = retired_servers .into_iter() @@ -151,8 +154,9 @@ async fn release_one_idle_project_server_before_open( ) .await; super::project_server_lifecycle::retire_project_servers(retired_servers, None).await; - for data_root in hook_data_roots { - super::hook_v2_replay_consumer::shutdown_hook_v2_replay_consumer(&data_root).await; + for (data_root, owner) in hook_replay_owners { + super::hook_v2_replay_consumer::retire_hook_v2_replay_consumer(&data_root, &owner) + .await; } // A prior same-owner retirement can be a bind eviction cleanup, // which waits on this capacity gate before its owner check. @@ -318,7 +322,7 @@ pub(crate) fn daemon_transcript_source_profile( /// registered. async fn retire_unserved_process_owners( store_administration: &StoreAdministration, - cg: &tracedecay_project::project::TraceDecay, + cg: &Arc, ) { let graph_db_path = cg.db().canonical_database_path(); let served = store_administration @@ -335,8 +339,9 @@ async fn retire_unserved_process_owners( { let _ = crate::test_support::hold_after_unserved_owner_check_for_test().await; } - super::hook_v2_replay_consumer::shutdown_hook_v2_replay_consumer( + super::hook_v2_replay_consumer::retire_hook_v2_replay_consumer( &cg.hook_store_layout().data_root, + &Arc::downgrade(cg), ) .await; if let Some(project_id) = hook_project_id_for_layout(cg.hook_store_layout()) {