From d5deb6c0106023a6f594ecf7c4508aa4350a9d16 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 00:45:34 +0000 Subject: [PATCH 1/2] test(daemon): pin replay consumer ownership across a reopen Two owners of one project share a hook data root. A late retirement of the older owner must not stop the newer owner's consumer, and a reopen must take over replay from an owner still retiring. Both fail on master (#3982). --- .../src/daemon/hook_v2_replay_consumer.rs | 155 ++++++++++++++++++ 1 file changed, 155 insertions(+) diff --git a/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs b/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs index 57aea13d8f..d22d03560a 100644 --- a/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs +++ b/crates/tracedecay/src/daemon/hook_v2_replay_consumer.rs @@ -609,4 +609,159 @@ 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"); + drop(older); + assert!(fixture.register(&newer), "the reopened owner starts replay"); + + shutdown_hook_v2_replay_consumer(&data_root).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); + + shutdown_hook_v2_replay_consumer(&data_root).await; + drop(older); + + assert!( + newer_registered, + "the reopened owner's registration must be accepted" + ); + assert_newer_owner_survives(&data_root, &newer).await; + } } From cfa3217cb4d34a128666efed3c7cde549c4dbaa5 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 00:50:47 +0000 Subject: [PATCH 2/2] fix(daemon): scope replay consumer shutdown to its owner Retiring an owner stopped whichever consumer held its hook data root, and a reopen that registered while the retiring owner's graph was alive was refused, so the older graph's release left the project without replay. The newest owner now takes the data root over, retirement stops only the consumer its graph owns, and spool detach clears only the wake it attached. Daemon shutdown and project deletion still stop any owner. Fixes #3982. --- .../branch_admin/remote_recovery_lifecycle.rs | 3 +- .../src/daemon/hook_v2_replay_consumer.rs | 76 ++++++++++++++----- .../hook_v2_replay_consumer/spool_watch.rs | 9 ++- .../src/daemon/project_composition.rs | 17 +++-- 4 files changed, 76 insertions(+), 29 deletions(-) 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 d22d03560a..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)] @@ -737,10 +772,11 @@ mod tests { 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"); - shutdown_hook_v2_replay_consumer(&data_root).await; + retire_hook_v2_replay_consumer(&data_root, &older_owner).await; assert_newer_owner_survives(&data_root, &newer).await; } @@ -755,7 +791,7 @@ mod tests { assert!(fixture.register(&older), "the older owner starts replay"); let newer_registered = fixture.register(&newer); - shutdown_hook_v2_replay_consumer(&data_root).await; + retire_hook_v2_replay_consumer(&data_root, &Arc::downgrade(&older)).await; drop(older); assert!( 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()) {