From 3c5e8e540214c3960be0954fd6bd25e3d1f44698 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 10 Jul 2026 16:35:12 +0000 Subject: [PATCH] fix(storage): retire applied consolidation inputs --- src/doctor/heal.rs | 318 ++++++++++++++++- src/lifecycle_lease.rs | 37 +- src/main.rs | 4 +- src/migrate/consolidate/mod.rs | 589 ++++++++++++++++++++++++++++++- src/migrate/consolidate/tests.rs | 151 ++++++++ src/update_cmd.rs | 39 +- 6 files changed, 1095 insertions(+), 43 deletions(-) diff --git a/src/doctor/heal.rs b/src/doctor/heal.rs index 102cc04c..cda3c80b 100644 --- a/src/doctor/heal.rs +++ b/src/doctor/heal.rs @@ -17,6 +17,9 @@ //! system temp directory are purged (the automated equivalent of //! `tracedecay migrate registry-gc --prefix --apply`), and only when //! BOTH the canonical and display roots are gone. +//! - input store manifests from completed schema-2 consolidations are renamed +//! out of the canonical discovery path after the applied ledger and both +//! repository markers prove the destination identity. //! //! Those auto-applied remedies are safe precisely because quarantine renames //! instead of deleting and the GC removes only temp-rooted registry metadata @@ -41,6 +44,10 @@ pub struct BranchMetaQuarantine { /// Outcome of one post-update health pass. #[derive(Debug, Default)] pub struct HealthPassReport { + /// Input manifests retired from already-applied consolidations. + pub retired_consolidation_manifests: Vec, + /// Superseded source/target registry projects removed after validation. + pub retired_consolidation_registry_projects: usize, pub quarantined_branch_meta: Vec, /// `None` when the global DB could not be opened, so the GC never ran. pub purged_temp_registry_rows: Option, @@ -50,32 +57,72 @@ pub struct HealthPassReport { pub warnings: Vec, } -/// Runs the full post-update health pass and prints a doctor-style summary. -/// -/// Never fails: every error is collected as a warning so a broken store can -/// never abort `tracedecay update`. -pub async fn run_post_update_health_pass() -> HealthPassReport { - eprintln!("\n\x1b[1mPost-update health pass\x1b[0m (skip with --no-heal)"); - +#[doc(hidden)] +pub async fn run_post_update_health_pass_under_lease( + lifecycle_lease: &crate::lifecycle_lease::LifecycleLease, +) -> HealthPassReport { let Some(profile_root) = crate::config::user_data_dir() else { + return render_missing_profile_report(); + }; + if let Some(warning) = health_pass_lease_error( + lifecycle_lease, + &profile_root, + crate::daemon::daemon_reachable(), + ) { let report = HealthPassReport { - warnings: vec!["could not determine the profile data directory".to_string()], + warnings: vec![warning], ..HealthPassReport::default() }; render_warnings(&report.warnings); return report; - }; + } + run_post_update_health_pass_for_profile(&profile_root).await +} - let report = compute_health_pass_report(&profile_root).await; +fn health_pass_lease_error( + lifecycle_lease: &crate::lifecycle_lease::LifecycleLease, + profile_root: &Path, + daemon_reachable: bool, +) -> Option { + if !lifecycle_lease.is_exclusive() { + return Some("post-update health pass requires an exclusive lifecycle lease".to_string()); + } + if !lifecycle_lease.guards_profile(profile_root) { + return Some(format!( + "post-update health pass lifecycle lease does not guard profile '{}'", + profile_root.display() + )); + } + daemon_reachable.then(|| { + "post-update health pass requires the TraceDecay daemon to be unreachable".to_string() + }) +} + +async fn run_post_update_health_pass_for_profile(profile_root: &Path) -> HealthPassReport { + eprintln!("\n\x1b[1mPost-update health pass\x1b[0m (skip with --no-heal)"); + let report = compute_health_pass_report(profile_root).await; render_health_pass_report(&report); report } +fn render_missing_profile_report() -> HealthPassReport { + let report = HealthPassReport { + warnings: vec!["could not determine the profile data directory".to_string()], + ..HealthPassReport::default() + }; + render_warnings(&report.warnings); + report +} + /// Applies the safe remedies and gathers everything the pass has to say into /// a [`HealthPassReport`], without printing anything. async fn compute_health_pass_report(profile_root: &Path) -> HealthPassReport { let mut report = HealthPassReport::default(); + // The post-update command holds the exclusive lifecycle lease around this + // entire pass, so manifest retirement cannot race daemon or hook opens. + retire_completed_consolidation_manifests(profile_root, &mut report).await; + let (quarantined, warnings) = quarantine_corrupt_branch_meta(profile_root); report.quarantined_branch_meta = quarantined; report.warnings.extend(warnings); @@ -117,8 +164,37 @@ async fn compute_health_pass_report(profile_root: &Path) -> HealthPassReport { report } +async fn retire_completed_consolidation_manifests( + profile_root: &Path, + report: &mut HealthPassReport, +) { + let retirement = + crate::migrate::consolidate::retire_applied_input_manifests(profile_root).await; + report.retired_consolidation_manifests = retirement.retired; + report.retired_consolidation_registry_projects = retirement.retired_registry_projects; + report.warnings.extend(retirement.warnings); +} + /// Prints the doctor-style summary for a computed report. fn render_health_pass_report(report: &HealthPassReport) { + if report.retired_consolidation_manifests.is_empty() { + eprintln!(" \x1b[32m✔\x1b[0m No completed consolidation manifests to retire"); + } else { + eprintln!( + " \x1b[32m✔\x1b[0m Retired {} completed consolidation input manifest(s):", + report.retired_consolidation_manifests.len() + ); + for path in &report.retired_consolidation_manifests { + eprintln!(" • {}", path.display()); + } + } + if report.retired_consolidation_registry_projects > 0 { + eprintln!( + " \x1b[32m✔\x1b[0m Retired {} superseded consolidation registry project(s)", + report.retired_consolidation_registry_projects + ); + } + if report.quarantined_branch_meta.is_empty() { eprintln!(" \x1b[32m✔\x1b[0m No corrupt branch metadata files"); } else { @@ -396,4 +472,226 @@ mod tests { assert!(quarantines.is_empty()); assert!(warnings.is_empty()); } + + #[test] + fn under_lease_health_pass_rejects_shared_wrong_profile_and_reachable_daemon() { + let dir = tempfile::TempDir::new().unwrap(); + let shared = match crate::lifecycle_lease::try_acquire_shared_for_profile( + dir.path(), + "shared healer test", + ) + .unwrap() + { + crate::lifecycle_lease::SharedLeaseAttempt::Acquired(lease) => lease, + crate::lifecycle_lease::SharedLeaseAttempt::Busy => panic!("unexpected busy lease"), + }; + assert!( + health_pass_lease_error(&shared, dir.path(), false) + .unwrap() + .contains("requires an exclusive lifecycle lease") + ); + drop(shared); + + let guarded_profile = dir.path().join("guarded"); + let other_profile = dir.path().join("other"); + let exclusive = crate::lifecycle_lease::acquire_exclusive_for_profile( + &guarded_profile, + "exclusive healer test", + ) + .unwrap(); + assert!( + health_pass_lease_error(&exclusive, &other_profile, false) + .unwrap() + .contains("does not guard profile") + ); + assert!( + health_pass_lease_error(&exclusive, &guarded_profile, true) + .unwrap() + .contains("daemon to be unreachable") + ); + assert!(health_pass_lease_error(&exclusive, &guarded_profile, false).is_none()); + } + + #[tokio::test] + async fn post_update_retires_applied_consolidation_manifests_idempotently() { + let dir = tempfile::TempDir::new().unwrap(); + let project = dir.path().join("repo"); + let profile = dir.path().join("profile"); + std::fs::create_dir_all(&project).unwrap(); + let status = std::process::Command::new("git") + .args(["init", "--quiet"]) + .current_dir(&project) + .status() + .unwrap(); + assert!(status.success()); + let project = project.canonicalize().unwrap(); + let git_common_dir = crate::worktree::git_common_dir(&project).unwrap(); + let source_id = "proj_heal_source"; + let target_id = "proj_heal_target"; + let destination_id = crate::migrate::consolidate::destination_project_id( + &git_common_dir, + source_id, + target_id, + ); + for project_id in [source_id, target_id, destination_id.as_str()] { + let layout = crate::storage::profile_sharded_layout( + &project, + &profile, + &crate::storage::EnrollmentMarker { + project_id: project_id.to_string(), + storage_mode: crate::storage::StorageMode::ProfileSharded, + }, + ) + .unwrap(); + std::fs::create_dir_all(&layout.data_root).unwrap(); + crate::storage::write_store_manifest(&layout).unwrap(); + } + crate::storage::write_repository_identity_marker(&project, &destination_id).unwrap(); + crate::storage::write_enrollment_marker( + &project, + &crate::storage::EnrollmentMarker { + project_id: destination_id.clone(), + storage_mode: crate::storage::StorageMode::ProfileSharded, + }, + ) + .unwrap(); + + let migration_id = format!("consolidate_{}", &destination_id[5..]); + let ledger_root = profile.join("migration-inventory"); + std::fs::create_dir_all(&ledger_root).unwrap(); + std::fs::write( + ledger_root.join(format!("{migration_id}.json")), + serde_json::to_vec_pretty(&serde_json::json!({ + "schema_version": 2, + "migration_id": migration_id, + "confirmation_token": "confirm-healer", + "input_fingerprint": "healer-fixture", + "source_project_id": source_id, + "target_project_id": target_id, + "destination_project_id": destination_id.clone(), + "project_root": project, + "git_common_dir": git_common_dir, + "state": "applied", + "graph_offsets": [], + "session_offsets": null, + "preserved_collisions": [] + })) + .unwrap(), + ) + .unwrap(); + + let global = crate::global_db::GlobalDb::open_at(&profile.join("global.db")) + .await + .unwrap(); + for project_id in [source_id, target_id, destination_id.as_str()] { + global + .upsert_code_project( + project_id, + &project, + Some(&git_common_dir), + None, + Some("main"), + ) + .await + .unwrap(); + global + .upsert_store_instance(crate::global_db::StoreInstanceUpsert { + store_id: format!("store:{project_id}:profile_sharded"), + project_id: project_id.to_string(), + store_kind: "code_project".to_string(), + storage_mode: "profile_sharded".to_string(), + store_relpath: format!("projects/{project_id}"), + manifest_relpath: Some(format!( + "projects/{project_id}/{}", + crate::storage::STORE_MANIFEST_FILENAME + )), + last_verified_at: Some(1_800_000_000), + last_write_at: Some(1_800_000_000), + }) + .await + .unwrap(); + } + global + .upsert_project_alias(&project, &destination_id) + .await + .unwrap(); + global.checkpoint().await; + global.close(); + + let _lease = crate::lifecycle_lease::acquire_exclusive_for_profile( + &profile, + "post-update healer test", + ) + .unwrap(); + std::fs::write( + profile + .join("migration-inventory") + .join(".fail-registry-retirement-once"), + b"fail once", + ) + .unwrap(); + let mut interrupted = HealthPassReport::default(); + retire_completed_consolidation_manifests(&profile, &mut interrupted).await; + assert_eq!(interrupted.warnings.len(), 1); + assert!(interrupted.warnings[0].contains("synthetic registry retirement failure")); + for project_id in [source_id, target_id] { + let root = profile.join("projects").join(project_id); + assert!( + !root.join(crate::storage::STORE_MANIFEST_FILENAME).exists() + && root + .join(format!( + "store_manifest.consolidated-into-{destination_id}.json" + )) + .is_file() + ); + } + let global = crate::global_db::GlobalDb::open_at(&profile.join("global.db")) + .await + .unwrap(); + assert_eq!(global.list_code_projects(usize::MAX).await.len(), 3); + global + .conn() + .execute( + "UPDATE code_projects SET canonical_root=?1 WHERE project_id=?2", + libsql::params!["/moved/elsewhere", source_id], + ) + .await + .unwrap(); + global.close(); + let mut moved = HealthPassReport::default(); + retire_completed_consolidation_manifests(&profile, &mut moved).await; + assert!(moved.warnings.is_empty(), "{:?}", moved.warnings); + assert!(moved.retired_consolidation_manifests.is_empty()); + assert_eq!(moved.retired_consolidation_registry_projects, 1); + for project_id in [source_id, target_id] { + let root = profile.join("projects").join(project_id); + assert!(!root.join(crate::storage::STORE_MANIFEST_FILENAME).exists()); + assert!( + root.join(format!( + "store_manifest.consolidated-into-{destination_id}.json" + )) + .is_file() + ); + } + + let mut retried = HealthPassReport::default(); + retire_completed_consolidation_manifests(&profile, &mut retried).await; + assert!(retried.warnings.is_empty(), "{:?}", retried.warnings); + assert!(retried.retired_consolidation_manifests.is_empty()); + assert_eq!(retried.retired_consolidation_registry_projects, 0); + + let global = crate::global_db::GlobalDb::open_at(&profile.join("global.db")) + .await + .unwrap(); + let owners = global.list_code_projects(usize::MAX).await; + assert_eq!(owners.len(), 2); + assert!(owners.iter().any(|project| { + project.project_id == source_id && project.canonical_root == "/moved/elsewhere" + })); + assert!( + owners + .iter() + .any(|project| project.project_id == destination_id) + ); + } } diff --git a/src/lifecycle_lease.rs b/src/lifecycle_lease.rs index 3a0f74ae..f69c2008 100644 --- a/src/lifecycle_lease.rs +++ b/src/lifecycle_lease.rs @@ -24,6 +24,8 @@ enum LeaseHold { pub struct LifecycleLease { hold: LeaseHold, token: Option, + lock_path: PathBuf, + exclusive: bool, } #[derive(Debug)] @@ -36,6 +38,19 @@ impl LifecycleLease { pub fn token(&self) -> Option<&str> { self.token.as_deref() } + + pub fn is_exclusive(&self) -> bool { + self.exclusive + } + + pub fn guards_profile(&self, profile_root: &Path) -> bool { + let expected = profile_root.join(LIFECYCLE_LOCK_FILENAME); + canonical_or_original(&self.lock_path) == canonical_or_original(&expected) + } +} + +fn canonical_or_original(path: &Path) -> PathBuf { + path.canonicalize().unwrap_or_else(|_| path.to_path_buf()) } impl Drop for LifecycleLease { @@ -109,6 +124,8 @@ fn acquire_shared_or_inherited_at(path: &Path, operation: &str) -> Result Ok(LifecycleLease { hold: LeaseHold::File(file), token: None, + lock_path: path.to_path_buf(), + exclusive: false, }), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { let owner = read_owner(&mut file); @@ -117,6 +134,8 @@ fn acquire_shared_or_inherited_at(path: &Path, operation: &str) -> Result Result Ok(SharedLeaseAttempt::Acquired(LifecycleLease { hold: LeaseHold::File(file), token: None, + lock_path: path.to_path_buf(), + exclusive: false, })), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { let owner = read_owner(&mut file); @@ -140,6 +161,8 @@ fn try_acquire_shared_or_inherited_at(path: &Path, operation: &str) -> Result Result { let mut file = open_lock_file(path)?; match fs2::FileExt::try_lock_exclusive(&file) { - Ok(()) => own_exclusive(file, operation), + Ok(()) => own_exclusive(file, path, operation), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { let owner = read_owner(&mut file); if let Some(token) = inherited.filter(|token| { @@ -179,6 +202,8 @@ fn acquire_exclusive_or_inherited_at( Ok(LifecycleLease { hold: LeaseHold::Inherited, token: Some(token), + lock_path: path.to_path_buf(), + exclusive: true, }) } else { Err(busy_error(operation, owner.as_deref())) @@ -215,7 +240,7 @@ fn lifecycle_lock_path_for_profile(profile_root: &Path) -> Result { fn acquire_exclusive_at(path: &Path, operation: &str) -> Result { let file = open_lock_file(path)?; match fs2::FileExt::try_lock_exclusive(&file) { - Ok(()) => own_exclusive(file, operation), + Ok(()) => own_exclusive(file, path, operation), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { let mut file = file; let owner = read_owner(&mut file); @@ -231,6 +256,8 @@ fn acquire_shared_at(path: &Path, operation: &str) -> Result { Ok(()) => Ok(LifecycleLease { hold: LeaseHold::File(file), token: None, + lock_path: path.to_path_buf(), + exclusive: false, }), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { let owner = read_owner(&mut file); @@ -246,6 +273,8 @@ fn try_acquire_shared_at(path: &Path, operation: &str) -> Result Ok(SharedLeaseAttempt::Acquired(LifecycleLease { hold: LeaseHold::File(file), token: None, + lock_path: path.to_path_buf(), + exclusive: false, })), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { Ok(SharedLeaseAttempt::Busy) @@ -254,7 +283,7 @@ fn try_acquire_shared_at(path: &Path, operation: &str) -> Result Result { +fn own_exclusive(mut file: File, path: &Path, operation: &str) -> Result { let token = lease_token(); file.set_len(0).map_err(|error| owner_write_error(&error))?; file.seek(SeekFrom::Start(0)) @@ -266,6 +295,8 @@ fn own_exclusive(mut file: File, operation: &str) -> Result { Ok(LifecycleLease { hold: LeaseHold::File(file), token: Some(token), + lock_path: path.to_path_buf(), + exclusive: true, }) } diff --git a/src/main.rs b/src/main.rs index fcad3fb7..5ec8efd5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -595,11 +595,11 @@ async fn dispatch_command(command: Commands) -> tracedecay::errors::Result<()> { no_reinstall, lifecycle_lease_token, } => { - let _lifecycle_lease = tracedecay::lifecycle_lease::acquire_exclusive_or_inherited( + let lifecycle_lease = tracedecay::lifecycle_lease::acquire_exclusive_or_inherited( "post-update", lifecycle_lease_token.as_deref(), )?; - update_cmd::run_post_update_tasks(no_heal, no_reinstall).await?; + update_cmd::run_post_update_tasks(no_heal, no_reinstall, &lifecycle_lease).await?; } Commands::Channel { channel } => match channel { Some(target) => { diff --git a/src/migrate/consolidate/mod.rs b/src/migrate/consolidate/mod.rs index 51ab2d11..47e798e8 100644 --- a/src/migrate/consolidate/mod.rs +++ b/src/migrate/consolidate/mod.rs @@ -19,6 +19,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use libsql::params; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -137,6 +138,27 @@ struct ConsolidationLedger { preserved_collisions: Vec, } +#[derive(Debug, Default)] +pub(crate) struct ManifestRetirementReport { + pub retired: Vec, + pub retired_registry_projects: usize, + pub warnings: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ManifestRetirementAction { + RenameCanonical, + RemoveDuplicateCanonical, + AlreadyRetired, +} + +struct ManifestRetirementPlan { + canonical_path: PathBuf, + retired_path: PathBuf, + manifest: StoreManifest, + action: ManifestRetirementAction, +} + struct ResolvedPlan { report: ConsolidationReport, input_fingerprint: String, @@ -242,6 +264,7 @@ async fn apply_with_faults( let mut ledger = load_or_create_ledger(&resolved, &ledger_path)?; validate_ledger(&ledger, &resolved)?; if ledger.state == ConsolidationState::Applied { + finalize_applied_consolidation(&options.profile_root, &ledger).await?; let mut report = resolved.report; report.state = ConsolidationState::Applied; report.dry_run = false; @@ -301,6 +324,10 @@ async fn apply_with_faults( save_ledger(&ledger_path, &ledger)?; } + if ledger.state == ConsolidationState::Applied { + finalize_applied_consolidation(&options.profile_root, &ledger).await?; + } + let mut report = resolved.report; report.state = ledger.state; report.dry_run = false; @@ -344,15 +371,35 @@ async fn resolve_plan_inner( .map_err(|error| config_error(format!("could not resolve profile root: {error}")))?; let git_common_dir = crate::worktree::git_common_dir(&project_root) .ok_or_else(|| config_error("project must be an attached git checkout"))?; - let source_layout = layout_for_id(&project_root, &profile_root, &options.source_project_id)?; - let target_layout = layout_for_id(&project_root, &profile_root, &options.target_project_id)?; - let source_manifest = validate_manifest(&source_layout, &options.source_project_id)?; - let target_manifest = validate_manifest(&target_layout, &options.target_project_id)?; let destination_project_id = destination_project_id( &git_common_dir, &options.source_project_id, &options.target_project_id, ); + let migration_id = format!("consolidate_{}", &destination_project_id[5..]); + let ledger_path = profile_root + .join(LEDGER_DIR) + .join(format!("{migration_id}.json")); + let applied_ledger = load_ledger(&ledger_path)?.filter(|ledger| { + ledger.schema_version == LEDGER_SCHEMA_VERSION + && ledger.state == ConsolidationState::Applied + }); + let allow_retired_manifests = allow_destination_marker && applied_ledger.is_some(); + let source_layout = layout_for_id(&project_root, &profile_root, &options.source_project_id)?; + let target_layout = layout_for_id(&project_root, &profile_root, &options.target_project_id)?; + let source_manifest = validate_input_manifest( + &source_layout, + &options.source_project_id, + &destination_project_id, + allow_retired_manifests, + )?; + let target_manifest = validate_input_manifest( + &target_layout, + &options.target_project_id, + &destination_project_id, + allow_retired_manifests, + ); + let target_manifest = target_manifest?; let repository_marker = storage::read_repository_identity_marker(&project_root)? .ok_or_else(|| config_error("repository identity marker is required"))?; let marker_ok = repository_marker.project_id == options.target_project_id @@ -378,14 +425,16 @@ async fn resolve_plan_inner( "source and target manifests do not prove one exact git-common-dir identity", )); } - reject_ambiguous_shards( - options, - &profile_root, - &project_root, - &git_common_dir, - &target_manifest, - &destination_project_id, - )?; + if applied_ledger.is_none() { + reject_ambiguous_shards( + options, + &profile_root, + &project_root, + &git_common_dir, + &target_manifest, + &destination_project_id, + )?; + } let mut source_meta = load_required_branch_meta(&source_layout)?; let mut target_meta = load_required_branch_meta(&target_layout)?; @@ -434,15 +483,18 @@ async fn resolve_plan_inner( ) .await?; let collisions = collision_summary(&evidence, &source_layout, &target_layout).await?; - let input_fingerprint = fingerprint_inputs(&evidence, &source_layout, &target_layout)?; - let migration_id = format!("consolidate_{}", &destination_project_id[5..]); + let input_fingerprint = fingerprint_inputs( + &evidence, + &source_layout, + &target_layout, + applied_ledger + .as_ref() + .map(|_| destination_project_id.as_str()), + )?; let confirmation_token = confirmation_token(&input_fingerprint, &migration_id); let destination_data_root = storage::profile_sharded_data_root(&profile_root, &destination_project_id); let backup_root = profile_root.join(BACKUP_DIR).join(&migration_id); - let ledger_path = profile_root - .join(LEDGER_DIR) - .join(format!("{migration_id}.json")); let state = load_ledger(&ledger_path)? .map(|ledger| ledger.state) .unwrap_or(ConsolidationState::Planned); @@ -492,6 +544,26 @@ fn validate_manifest(layout: &StoreLayout, project_id: &str) -> Result Result { + if !allow_retired { + return validate_manifest(layout, project_id); + } + Ok(inspect_manifest_retirement(layout, project_id, destination_project_id)?.manifest) +} + +fn validate_manifest_path( + path: &Path, + layout: &StoreLayout, + project_id: &str, +) -> Result { let manifest = storage::read_store_manifest(path)?; if manifest.project_id.as_deref() != Some(project_id) || manifest.schema_version != storage::STORE_MANIFEST_SCHEMA_VERSION @@ -726,7 +798,7 @@ async fn collision_summary( }) } -fn destination_project_id(git_common_dir: &Path, source: &str, target: &str) -> String { +pub(crate) fn destination_project_id(git_common_dir: &Path, source: &str, target: &str) -> String { let mut ids = [source, target]; ids.sort_unstable(); let mut hash = Sha256::new(); @@ -756,6 +828,7 @@ fn fingerprint_inputs( evidence: &InputReadEvidence, source: &StoreLayout, target: &StoreLayout, + retired_destination_project_id: Option<&str>, ) -> Result { let mut hash = Sha256::new(); for (label, root, graph) in [ @@ -763,7 +836,28 @@ fn fingerprint_inputs( ("target", &target.data_root, &evidence.target_graph), ] { hash.update(label.as_bytes()); + let mut files = BTreeMap::new(); for (relative, path) in relative_file_map(root)? { + let retired_manifest_name = retired_destination_project_id + .map(|destination| format!("store_manifest.consolidated-into-{destination}.json")); + let relative = retired_destination_project_id + .filter(|_| { + retired_manifest_name + .as_deref() + .is_some_and(|name| relative == Path::new(name)) + }) + .map(|_| PathBuf::from(storage::STORE_MANIFEST_FILENAME)) + .unwrap_or(relative); + if let Some(existing) = files.insert(relative.clone(), path.clone()) + && file_digest(&existing)? != file_digest(&path)? + { + return Err(config_error(format!( + "duplicate normalized consolidation input '{}' diverges", + relative.display() + ))); + } + } + for (relative, path) in files { if is_runtime_lock(&relative) || is_sqlite_sidecar(&relative) { continue; } @@ -883,6 +977,465 @@ fn save_ledger(path: &Path, ledger: &ConsolidationLedger) -> Result<()> { PrivateStoreIo::write_file_atomically(path, &temp, &bytes).map_err(io_error) } +pub(crate) async fn retire_applied_input_manifests( + profile_root: &Path, +) -> ManifestRetirementReport { + let mut report = ManifestRetirementReport::default(); + let ledger_root = profile_root.join(LEDGER_DIR); + let entries = match fs::read_dir(&ledger_root) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return report, + Err(error) => { + report.warnings.push(format!( + "could not read consolidation ledger directory '{}': {error}", + ledger_root.display() + )); + return report; + } + }; + let mut ledger_paths = entries + .flatten() + .map(|entry| entry.path()) + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with("consolidate_")) + && path + .extension() + .is_some_and(|extension| extension.eq_ignore_ascii_case("json")) + }) + .collect::>(); + ledger_paths.sort(); + + for ledger_path in ledger_paths { + let ledger = match load_ledger(&ledger_path) { + Ok(Some(ledger)) => ledger, + Ok(None) => continue, + Err(error) => { + report.warnings.push(error.to_string()); + continue; + } + }; + if ledger.state != ConsolidationState::Applied { + continue; + } + if ledger.schema_version != LEDGER_SCHEMA_VERSION { + report.warnings.push(format!( + "applied consolidation ledger '{}' uses schema version {}; expected {}", + ledger_path.display(), + ledger.schema_version, + LEDGER_SCHEMA_VERSION + )); + continue; + } + match finalize_applied_consolidation(profile_root, &ledger).await { + Ok((retired, registry_projects)) => { + report.retired.extend(retired); + report.retired_registry_projects = report + .retired_registry_projects + .saturating_add(registry_projects); + } + Err(error) => report.warnings.push(error.to_string()), + } + } + report +} + +async fn finalize_applied_consolidation( + profile_root: &Path, + ledger: &ConsolidationLedger, +) -> Result<(Vec, usize)> { + validate_applied_retirement_authority(profile_root, ledger)?; + let source_layout = layout_for_id( + &ledger.project_root, + profile_root, + &ledger.source_project_id, + )?; + let target_layout = layout_for_id( + &ledger.project_root, + profile_root, + &ledger.target_project_id, + )?; + let plans = [ + inspect_manifest_retirement( + &source_layout, + &ledger.source_project_id, + &ledger.destination_project_id, + )?, + inspect_manifest_retirement( + &target_layout, + &ledger.target_project_id, + &ledger.destination_project_id, + )?, + ]; + let mut retired = Vec::new(); + for plan in plans { + let parent = plan + .canonical_path + .parent() + .ok_or_else(|| config_error("input store manifest has no parent directory"))?; + match plan.action { + ManifestRetirementAction::RenameCanonical => { + fs::rename(&plan.canonical_path, &plan.retired_path).map_err(io_error)?; + files::sync_parent_directory(parent)?; + retired.push(plan.retired_path); + } + ManifestRetirementAction::RemoveDuplicateCanonical => { + fs::remove_file(&plan.canonical_path).map_err(io_error)?; + files::sync_parent_directory(parent)?; + retired.push(plan.retired_path); + } + ManifestRetirementAction::AlreadyRetired => {} + } + } + let retired_registry_projects = retire_legacy_registry_owners(profile_root, ledger).await?; + Ok((retired, retired_registry_projects)) +} + +fn validate_applied_retirement_authority( + profile_root: &Path, + ledger: &ConsolidationLedger, +) -> Result<()> { + if ledger.schema_version != LEDGER_SCHEMA_VERSION || ledger.state != ConsolidationState::Applied + { + return Err(config_error( + "source manifest retirement requires an applied schema-2 consolidation ledger", + )); + } + for project_id in [ + &ledger.source_project_id, + &ledger.target_project_id, + &ledger.destination_project_id, + ] { + storage::validate_project_id(project_id).map_err(config_error)?; + } + let expected_destination = destination_project_id( + &ledger.git_common_dir, + &ledger.source_project_id, + &ledger.target_project_id, + ); + let expected_migration = format!("consolidate_{}", &expected_destination[5..]); + if ledger.destination_project_id != expected_destination + || ledger.migration_id != expected_migration + { + return Err(config_error( + "applied consolidation ledger identity does not match its deterministic destination", + )); + } + + let repository = storage::read_repository_identity_marker(&ledger.project_root)? + .ok_or_else(|| config_error("repository identity marker is missing after consolidation"))?; + let enrollment = storage::read_enrollment_marker(&ledger.project_root)? + .ok_or_else(|| config_error("enrollment marker is missing after consolidation"))?; + if repository.project_id != ledger.destination_project_id + || !same_path( + Path::new(&repository.git_common_dir), + &ledger.git_common_dir, + ) + || enrollment.project_id != ledger.destination_project_id + || enrollment.storage_mode != StorageMode::ProfileSharded + { + return Err(config_error(format!( + "consolidation marker mismatch for destination project '{}'", + ledger.destination_project_id + ))); + } + + let destination_layout = layout_for_id( + &ledger.project_root, + profile_root, + &ledger.destination_project_id, + )?; + validate_manifest(&destination_layout, &ledger.destination_project_id)?; + Ok(()) +} + +async fn retire_legacy_registry_owners( + profile_root: &Path, + ledger: &ConsolidationLedger, +) -> Result { + let global_path = profile_root.join("global.db"); + if !global_path.is_file() { + return Err(config_error(format!( + "global registry '{}' is missing for applied consolidation", + global_path.display() + ))); + } + let db = GlobalDb::open_at(&global_path) + .await + .ok_or_else(|| config_error("could not open global registry for consolidation cleanup"))?; + let conn = db.conn(); + conn.execute("BEGIN IMMEDIATE", ()) + .await + .map_err(|error| config_error(format!("could not begin registry cleanup: {error}")))?; + + #[cfg(test)] + { + let injected_failure = profile_root + .join(LEDGER_DIR) + .join(".fail-registry-retirement-once"); + if injected_failure.is_file() { + let _ = fs::remove_file(injected_failure); + let _ = conn.execute("ROLLBACK", ()).await; + return Err(config_error( + "synthetic registry retirement failure after manifest retirement", + )); + } + } + + let result = async { + let canonical_root = GlobalDb::canonical_project_key(&ledger.project_root); + let mut rows = conn + .query( + "SELECT canonical_root, COALESCE(git_common_dir, '') + FROM code_projects WHERE project_id=?1", + params![ledger.destination_project_id.as_str()], + ) + .await + .map_err(|error| { + config_error(format!("could not validate destination project: {error}")) + })?; + let destination = rows + .next() + .await + .map_err(|error| config_error(format!("could not read destination project: {error}")))? + .ok_or_else(|| config_error("destination registry project is missing"))?; + let registered_root = destination.get::(0).map_err(|error| { + config_error(format!("invalid destination canonical root: {error}")) + })?; + let registered_common = destination.get::(1).map_err(|error| { + config_error(format!("invalid destination git common dir: {error}")) + })?; + if registered_root != canonical_root + || registered_common.is_empty() + || !same_path(Path::new(®istered_common), &ledger.git_common_dir) + { + return Err(config_error( + "destination registry project does not match the applied consolidation ledger", + )); + } + + let mut rows = conn + .query( + "SELECT project_id FROM project_aliases WHERE alias_path=?1", + params![canonical_root.as_str()], + ) + .await + .map_err(|error| { + config_error(format!("could not validate destination alias: {error}")) + })?; + let alias_owner = rows + .next() + .await + .map_err(|error| config_error(format!("could not read destination alias: {error}")))? + .and_then(|row| row.get::(0).ok()); + if alias_owner.as_deref() != Some(ledger.destination_project_id.as_str()) { + return Err(config_error( + "destination registry alias does not match the applied consolidation ledger", + )); + } + + let store_id = format!("store:{}:profile_sharded", ledger.destination_project_id); + let store_relpath = format!("projects/{}", ledger.destination_project_id); + let manifest_relpath = format!("{store_relpath}/{}", storage::STORE_MANIFEST_FILENAME); + let mut rows = conn + .query( + "SELECT project_id, store_kind, storage_mode, store_relpath, + COALESCE(manifest_relpath, '') + FROM store_instances WHERE store_id=?1", + params![store_id.as_str()], + ) + .await + .map_err(|error| { + config_error(format!("could not validate destination store: {error}")) + })?; + let store = rows + .next() + .await + .map_err(|error| config_error(format!("could not read destination store: {error}")))? + .ok_or_else(|| config_error("destination registry store is missing"))?; + let store_values = (0..5) + .map(|index| { + store.get::(index).map_err(|error| { + config_error(format!("invalid destination store registry row: {error}")) + }) + }) + .collect::>>()?; + if store_values + != vec![ + ledger.destination_project_id.clone(), + "code_project".to_string(), + "profile_sharded".to_string(), + store_relpath, + manifest_relpath, + ] + { + return Err(config_error( + "destination registry store does not match the applied consolidation ledger", + )); + } + + let canonical_common = GlobalDb::canonical_project_key(&ledger.git_common_dir); + let mut rows = conn + .query( + "SELECT project_id FROM code_projects WHERE canonical_root=?1 ORDER BY project_id", + params![canonical_root.as_str()], + ) + .await + .map_err(|error| { + config_error(format!("could not validate canonical owners: {error}")) + })?; + let mut owners = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| config_error(format!("could not read canonical owner: {error}")))? + { + owners.push(row.get::(0).map_err(|error| { + config_error(format!("invalid canonical owner registry row: {error}")) + })?); + } + let allowed = BTreeSet::from([ + ledger.source_project_id.clone(), + ledger.target_project_id.clone(), + ledger.destination_project_id.clone(), + ]); + if owners.iter().any(|owner| !allowed.contains(owner)) + || !owners.contains(&ledger.destination_project_id) + { + return Err(config_error(format!( + "canonical root has unexpected registry owners: {owners:?}" + ))); + } + + // Old project IDs can have been rebound to another repository. Match + // the full consolidation identity so those moved rows survive. + let deleted = conn + .execute( + "DELETE FROM code_projects + WHERE project_id IN (?1, ?2) + AND canonical_root=?3 + AND git_common_dir=?4", + params![ + ledger.source_project_id.as_str(), + ledger.target_project_id.as_str(), + canonical_root.as_str(), + canonical_common.as_str() + ], + ) + .await + .map_err(|error| { + config_error(format!("could not retire legacy registry owners: {error}")) + })?; + + let mut rows = conn + .query( + "SELECT project_id FROM code_projects WHERE canonical_root=?1 ORDER BY project_id", + params![canonical_root.as_str()], + ) + .await + .map_err(|error| { + config_error(format!("could not verify canonical owner cleanup: {error}")) + })?; + let remaining = rows + .next() + .await + .map_err(|error| { + config_error(format!("could not read remaining canonical owner: {error}")) + })? + .and_then(|row| row.get::(0).ok()); + let extra = rows.next().await.map_err(|error| { + config_error(format!("could not read extra canonical owner: {error}")) + })?; + if remaining.as_deref() != Some(ledger.destination_project_id.as_str()) || extra.is_some() { + return Err(config_error( + "registry cleanup did not leave exactly one destination canonical owner", + )); + } + usize::try_from(deleted) + .map_err(|_| config_error("legacy registry cleanup count overflowed usize")) + } + .await; + + match result { + Ok(deleted) => match conn.execute("COMMIT", ()).await { + Ok(_) => Ok(deleted), + Err(error) => { + let _ = conn.execute("ROLLBACK", ()).await; + Err(config_error(format!( + "could not commit legacy registry cleanup: {error}" + ))) + } + }, + Err(error) => { + let _ = conn.execute("ROLLBACK", ()).await; + Err(error) + } + } +} + +fn inspect_manifest_retirement( + layout: &StoreLayout, + project_id: &str, + destination_project_id: &str, +) -> Result { + let canonical_path = layout + .manifest_path + .clone() + .ok_or_else(|| config_error("profile shard has no store manifest path"))?; + let retired_path = canonical_path.with_file_name(format!( + "store_manifest.consolidated-into-{destination_project_id}.json" + )); + let canonical = read_optional_regular_file(&canonical_path)?; + let retired = read_optional_regular_file(&retired_path)?; + let (path, action) = match (&canonical, &retired) { + (Some(_), None) => ( + canonical_path.as_path(), + ManifestRetirementAction::RenameCanonical, + ), + (None, Some(_)) => ( + retired_path.as_path(), + ManifestRetirementAction::AlreadyRetired, + ), + (Some(canonical), Some(retired)) if canonical == retired => ( + retired_path.as_path(), + ManifestRetirementAction::RemoveDuplicateCanonical, + ), + (Some(_), Some(_)) => { + return Err(config_error(format!( + "canonical and retired store manifests diverge for project '{project_id}'" + ))); + } + (None, None) => { + return Err(config_error(format!( + "neither canonical nor retired store manifest exists for project '{project_id}'" + ))); + } + }; + let manifest = validate_manifest_path(path, layout, project_id)?; + Ok(ManifestRetirementPlan { + canonical_path, + retired_path, + manifest, + action, + }) +} + +fn read_optional_regular_file(path: &Path) -> Result>> { + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(io_error(error)), + }; + if !metadata.is_file() || metadata.file_type().is_symlink() { + return Err(config_error(format!( + "store manifest '{}' is not a regular file", + path.display() + ))); + } + fs::read(path).map(Some).map_err(io_error) +} + fn backup_store(layout: &StoreLayout, backup_root: &Path) -> Result<()> { let project_id = layout.identity.project_id.as_deref().unwrap_or("unknown"); copy_tree_exact(&layout.data_root, &backup_root.join(project_id)) diff --git a/src/migrate/consolidate/tests.rs b/src/migrate/consolidate/tests.rs index a6e79d4c..f6b26baf 100644 --- a/src/migrate/consolidate/tests.rs +++ b/src/migrate/consolidate/tests.rs @@ -164,6 +164,20 @@ impl Fixture { } } +fn input_manifest_paths( + fixture: &Fixture, + project_id: &str, + destination_project_id: &str, +) -> (PathBuf, PathBuf) { + let root = fixture.profile.join("projects").join(project_id); + ( + root.join(storage::STORE_MANIFEST_FILENAME), + root.join(format!( + "store_manifest.consolidated-into-{destination_project_id}.json" + )), + ) +} + #[tokio::test] async fn dry_run_reports_live_split_shape_without_mutation() { let fixture = fixture().await; @@ -276,6 +290,13 @@ async fn interrupted_apply_retries_without_duplicates_and_cuts_over_last() { b"target forensic database", ) .unwrap(); + let input_database_digests = [ + source_root.join(crate::config::DB_FILENAME), + source_root.join(storage::SESSIONS_DB_FILENAME), + target_root.join(crate::config::DB_FILENAME), + target_root.join(storage::SESSIONS_DB_FILENAME), + ] + .map(|path| (path.clone(), file_digest(&path).unwrap())); let report = plan(&options).await.unwrap(); let error = apply_with_stop( @@ -398,6 +419,32 @@ async fn interrupted_apply_retries_without_duplicates_and_cuts_over_last() { .join(&fixture.source_id) .is_dir() ); + for project_id in [&fixture.source_id, &fixture.target_id] { + let (canonical, retired) = + input_manifest_paths(&fixture, project_id, &applied.destination_project_id); + assert!(!canonical.exists()); + assert!(retired.is_file()); + } + for (path, digest) in input_database_digests { + assert_eq!( + file_digest(&path).unwrap(), + digest, + "{} changed", + path.display() + ); + } + let global = GlobalDb::open_at(&fixture.profile.join("global.db")) + .await + .unwrap(); + let owners = global + .list_code_projects(usize::MAX) + .await + .into_iter() + .filter(|project| same_path(Path::new(&project.canonical_root), &fixture.project)) + .map(|project| project.project_id) + .collect::>(); + assert_eq!(owners, vec![applied.destination_project_id.clone()]); + global.close(); assert!( fixture .profile @@ -484,6 +531,74 @@ async fn mixed_page_destination_survives_overlapping_watcher_opens() { } } +#[tokio::test] +async fn applied_manifest_retirement_handles_retry_states_and_fails_closed() { + let fixture = fixture().await; + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + let (source_canonical, source_retired) = input_manifest_paths( + &fixture, + &fixture.source_id, + &applied.destination_project_id, + ); + let (_, target_retired) = input_manifest_paths( + &fixture, + &fixture.target_id, + &applied.destination_project_id, + ); + + fs::copy(&source_retired, &source_canonical).unwrap(); + let both_identical = apply(&options, &report.confirmation_token).await.unwrap(); + assert_eq!(both_identical.state, ConsolidationState::Applied); + assert!(!source_canonical.exists()); + assert!(source_retired.is_file()); + + fs::write(&source_canonical, b"divergent manifest").unwrap(); + let divergent = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!(divergent.to_string().contains("manifests diverge")); + assert_eq!(fs::read(&source_canonical).unwrap(), b"divergent manifest"); + assert!(source_retired.is_file()); + assert!(target_retired.is_file()); + + fs::remove_file(&source_canonical).unwrap(); + storage::write_enrollment_marker( + &fixture.project, + &EnrollmentMarker { + project_id: fixture.target_id.clone(), + storage_mode: StorageMode::ProfileSharded, + }, + ) + .unwrap(); + let marker_mismatch = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!(marker_mismatch.to_string().contains("marker mismatch")); + assert!(source_retired.is_file()); + assert!(target_retired.is_file()); + + storage::write_enrollment_marker( + &fixture.project, + &EnrollmentMarker { + project_id: applied.destination_project_id.clone(), + storage_mode: StorageMode::ProfileSharded, + }, + ) + .unwrap(); + fs::remove_file(&source_retired).unwrap(); + let missing = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!( + missing + .to_string() + .contains("neither canonical nor retired") + ); + assert!(target_retired.is_file()); +} + #[tokio::test] async fn destination_preparation_restarts_after_every_publish_boundary() { for stop in [ @@ -2232,6 +2347,42 @@ async fn fixture() -> Fixture { false, ) .await; + let global = GlobalDb::open_at(&profile.join("global.db")).await.unwrap(); + let git_common_dir = crate::worktree::git_common_dir(&project).unwrap(); + for project_id in [&source_id, &target_id] { + global + .upsert_code_project( + project_id, + &project, + Some(&git_common_dir), + None, + Some("main"), + ) + .await + .unwrap(); + global + .upsert_store_instance(StoreInstanceUpsert { + store_id: format!("store:{project_id}:profile_sharded"), + project_id: project_id.clone(), + store_kind: "code_project".to_string(), + storage_mode: "profile_sharded".to_string(), + store_relpath: format!("projects/{project_id}"), + manifest_relpath: Some(format!( + "projects/{project_id}/{}", + storage::STORE_MANIFEST_FILENAME + )), + last_verified_at: Some(1_800_000_000), + last_write_at: Some(1_800_000_000), + }) + .await + .unwrap(); + } + global + .upsert_project_alias(&project, &target_id) + .await + .unwrap(); + global.checkpoint().await; + global.close(); storage::write_repository_identity_marker(&project, &target_id).unwrap(); Fixture { _temp: temp, diff --git a/src/update_cmd.rs b/src/update_cmd.rs index 86db3bb4..c5eb3895 100644 --- a/src/update_cmd.rs +++ b/src/update_cmd.rs @@ -273,14 +273,18 @@ pub(crate) fn run_upgrade_command( } /// The binary to re-exec for `post-update`: the freshly installed one when -/// the upgrade reported where it landed, otherwise the usual resolution. -/// Never `which_tracedecay()` alone — its current-exe-first order can point -/// at the OLD binary (e.g. a stale Homebrew keg) right after an upgrade. +/// the upgrade reported where it landed, otherwise the currently running +/// binary. This keeps source-built dogfood on the source-built binary. fn post_update_binary(installed: Option<&Path>) -> tracedecay::errors::Result { - match installed.filter(|path| path.exists()) { - Some(path) => Ok(path.to_string_lossy().into_owned()), - None => tracedecay_bin_on_path(), - } + let current = std::env::current_exe().ok(); + post_update_binary_from(installed, current.as_deref()).map_or_else(tracedecay_bin_on_path, Ok) +} + +fn post_update_binary_from(installed: Option<&Path>, current: Option<&Path>) -> Option { + installed + .filter(|path| path.exists()) + .map(normalize_bin_path) + .or_else(|| current_tracedecay_exe_from(current)) } fn run_post_update_subcommand( @@ -373,9 +377,10 @@ pub(crate) async fn reinstall_tracked_agents(user_config: &UserConfig) -> Reinst pub(crate) async fn run_post_update_tasks( no_heal: bool, no_reinstall: bool, + lifecycle_lease: &tracedecay::lifecycle_lease::LifecycleLease, ) -> tracedecay::errors::Result<()> { let previous_daemon_state = tracedecay::daemon::quiesce_installed_service_under_lease()?; - let mutation_result = run_post_update_mutations(no_heal, no_reinstall).await; + let mutation_result = run_post_update_mutations(no_heal, no_reinstall, lifecycle_lease).await; let restart_result = refresh_daemon_service_after_update(previous_daemon_state); mutation_result?; restart_result @@ -384,12 +389,13 @@ pub(crate) async fn run_post_update_tasks( async fn run_post_update_mutations( no_heal: bool, no_reinstall: bool, + lifecycle_lease: &tracedecay::lifecycle_lease::LifecycleLease, ) -> tracedecay::errors::Result<()> { refresh_generated_plugins().await?; if no_heal { eprintln!("Skipping post-update health pass (--no-heal)."); } else { - tracedecay::doctor::heal::run_post_update_health_pass().await; + tracedecay::doctor::heal::run_post_update_health_pass_under_lease(lifecycle_lease).await; } if no_reinstall { @@ -492,7 +498,7 @@ mod tests { use super::{ RefreshPolicy, ReinstallOutcome, current_tracedecay_exe_from, partition_reinstall_results, - post_update_binary, run_install_then_refresh, + post_update_binary, post_update_binary_from, run_install_then_refresh, }; use tempfile::TempDir; use tracedecay::upgrade::UpgradeOutcome; @@ -809,6 +815,19 @@ mod tests { assert_eq!(resolved, installed.to_string_lossy()); } + #[test] + fn post_update_binary_keeps_source_built_current_executable() { + let temp = tempfile::tempdir().expect("tempdir should exist"); + let current = temp.path().join("tracedecay"); + std::fs::write(¤t, b"source-built").expect("binary should be writable"); + let expected = current.to_string_lossy().replace('\\', "/"); + + assert_eq!( + post_update_binary_from(None, Some(¤t)).as_deref(), + Some(expected.as_str()) + ); + } + #[test] fn post_update_binary_ignores_a_missing_installed_path() { let temp = tempfile::tempdir().expect("tempdir should exist");