diff --git a/crosslink/src/daemon.rs b/crosslink/src/daemon.rs index 515630ab3..b47a81e9c 100644 --- a/crosslink/src/daemon.rs +++ b/crosslink/src/daemon.rs @@ -25,6 +25,14 @@ const START_LOCK_DEADLINE_SECS: u64 = 5; const START_POLL_MILLIS: u64 = 25; const READINESS_POLL_MILLIS: u64 = 50; const MAX_RETRY_SECS: u64 = 15; +/// Housekeeping re-hydrates and retries a readiness refresh this many times +/// within one tick when the hub moves under the write, before handing the +/// tick back to reconciliation. +const MAX_RECORD_REFRESH_ATTEMPTS: u32 = 3; +/// Consecutive housekeeping-triggered reconciliations, without a completed +/// tick in between, after which the daemon parks as `blocked_corrupt` with the +/// cause instead of flapping between ready and reconciling. +const MAX_CONSECUTIVE_RECONCILES: u32 = 5; const LEASE_LIVENESS_SWEEP_POLLS: u8 = 40; pub const WAITING_EXIT_CODE: i32 = 20; @@ -235,11 +243,33 @@ pub fn run_daemon(crosslink_dir: &Path, requested_epoch: Option<&str>) -> Result reason: None, }, )?; - while reconcile_until_ready(crosslink_dir, &identity, &should_exit)? { + let mut budget = ReconcileBudget::default(); + let mut reconcile_reason: Option = None; + while reconcile_until_ready( + crosslink_dir, + &identity, + &should_exit, + reconcile_reason.as_deref(), + )? { match run_normal_loop(crosslink_dir, &should_exit)? { NormalLoopExit::Shutdown => break, - NormalLoopExit::RecordExpired => { - println!("readiness record expired; reconciling again."); + NormalLoopExit::Reconcile { + reason, + completed_ticks, + } => { + if budget.note_reconcile(completed_ticks) { + let parked = format!( + "housekeeping sent the daemon back to reconciliation {} times in a row without completing a tick; last cause: {reason}", + budget.consecutive + ); + tracing::error!("{parked}"); + println!("{parked}; parking as blocked_corrupt."); + park_blocked(crosslink_dir, &identity, &should_exit, &parked)?; + break; + } + tracing::warn!("{reason}; reconciling again"); + println!("{reason}; reconciling again."); + reconcile_reason = Some(reason); } } } @@ -247,6 +277,73 @@ pub fn run_daemon(crosslink_dir: &Path, requested_epoch: Option<&str>) -> Result Ok(()) } +/// Consecutive housekeeping-triggered reconciliations. A completed +/// housekeeping tick in between means the previous reconciliation worked and +/// resets the count; reaching the limit without one means the fault is +/// persistent and the daemon should park with the cause. +#[derive(Default)] +struct ReconcileBudget { + consecutive: u32, +} + +impl ReconcileBudget { + /// Record one reconciliation request. Returns true when the budget is spent. + const fn note_reconcile(&mut self, completed_ticks_since_last: u64) -> bool { + if completed_ticks_since_last > 0 { + self.consecutive = 0; + } + self.consecutive += 1; + self.consecutive >= MAX_CONSECUTIVE_RECONCILES + } +} + +/// Record `blocked_corrupt` with the cause and hold the daemon there until it +/// is told to exit, so `daemon status` shows why and `daemon stop` plus +/// `ensure` recovers, like the reconciliation path's own blocked state. +fn park_blocked( + crosslink_dir: &Path, + identity: &DaemonIdentity, + should_exit: &AtomicBool, + reason: &str, +) -> Result<()> { + let attempt_id = Uuid::new_v4().to_string(); + let transition = readiness::acquire_transition_permit_observed( + crosslink_dir, + Some(should_exit), + None, + || { + readiness::write_record( + crosslink_dir, + ReadinessDraft { + daemon_epoch: &identity.daemon_epoch, + daemon_pid: identity.pid, + attempt_id: &attempt_id, + state: ReadinessState::Reconciling, + generation_id: None, + reason: Some("waiting for active repository mutations to finish"), + }, + ) + .map(|_| ()) + }, + )?; + for state in [ReadinessState::Reconciling, ReadinessState::BlockedCorrupt] { + readiness::write_record( + crosslink_dir, + ReadinessDraft { + daemon_epoch: &identity.daemon_epoch, + daemon_pid: identity.pid, + attempt_id: &attempt_id, + state, + generation_id: None, + reason: Some(reason), + }, + )?; + } + drop(transition); + while !wait_interruptible(should_exit, Duration::from_secs(1)) {} + Ok(()) +} + fn ensure_process(crosslink_dir: &Path) -> Result { if let Some(identity) = active_identity(crosslink_dir)? { return Ok(identity); @@ -531,16 +628,29 @@ fn reconcile_until_ready( crosslink_dir: &Path, identity: &DaemonIdentity, should_exit: &AtomicBool, + reason: Option<&str>, ) -> Result { - reconcile_until_ready_loop(crosslink_dir, identity, should_exit) + reconcile_until_ready_with(crosslink_dir, identity, should_exit, reason, || Ok(())) } -fn reconcile_until_ready_loop( +/// Reconcile until the repository is ready, waiting or blocked. `reason` is +/// recorded on the reconciling record so an operator can see why the daemon +/// left the ready state. `before_record` runs between activation and the +/// ready-record write; production passes a no-op and tests use it to move the +/// hub at exactly that point. +fn reconcile_until_ready_with( crosslink_dir: &Path, identity: &DaemonIdentity, should_exit: &AtomicBool, -) -> Result { + reason: Option<&str>, + mut before_record: F, +) -> Result +where + F: FnMut() -> Result<()>, +{ let mut retry_seconds = 1; + let mut reconcile_reason = reason.map(str::to_string); + let mut stale_retries = 0_u32; loop { if should_exit.load(Ordering::SeqCst) { return Ok(false); @@ -573,7 +683,7 @@ fn reconcile_until_ready_loop( attempt_id: &attempt_id, state: ReadinessState::Reconciling, generation_id: None, - reason: None, + reason: reconcile_reason.as_deref(), }, )?; let activation = match activate_repository(crosslink_dir) { @@ -582,39 +692,15 @@ fn reconcile_until_ready_loop( reason: format!("{error:#}"), }, }; - match activation { + let (state, generation_id) = match activation { RepositoryActivation::ReadyCurrent { generation_id } => { - record_ready( - crosslink_dir, - identity, - &attempt_id, - ReadinessState::ReadyCurrent, - &generation_id, - )?; - drop(transition); - return Ok(true); + (ReadinessState::ReadyCurrent, generation_id) } RepositoryActivation::ReadyMigrated { generation_id } => { - record_ready( - crosslink_dir, - identity, - &attempt_id, - ReadinessState::ReadyMigrated, - &generation_id, - )?; - drop(transition); - return Ok(true); + (ReadinessState::ReadyMigrated, generation_id) } RepositoryActivation::ReadyAdopted { generation_id } => { - record_ready( - crosslink_dir, - identity, - &attempt_id, - ReadinessState::ReadyAdopted, - &generation_id, - )?; - drop(transition); - return Ok(true); + (ReadinessState::ReadyAdopted, generation_id) } RepositoryActivation::WaitingForRemote { reason } => { readiness::write_record( @@ -633,6 +719,7 @@ fn reconcile_until_ready_loop( return Ok(false); } retry_seconds = (retry_seconds * 2).min(MAX_RETRY_SECS); + continue; } RepositoryActivation::BlockedCorrupt { reason } => { readiness::write_record( @@ -650,6 +737,38 @@ fn reconcile_until_ready_loop( while !wait_interruptible(should_exit, Duration::from_secs(1)) {} return Ok(false); } + }; + before_record()?; + match record_ready(crosslink_dir, identity, &attempt_id, state, &generation_id) { + Ok(()) => { + drop(transition); + return Ok(true); + } + // Another checkout published between activation and the record + // write. The activation is still sound; only its currency is gone, + // so derive it again instead of ending the daemon. + Err(error) if error.downcast_ref::().is_some() => { + drop(transition); + stale_retries += 1; + let moved = format!("{error:#}"); + if stale_retries >= MAX_CONSECUTIVE_RECONCILES { + let parked = format!( + "the hub moved under {stale_retries} consecutive reconciliations; last cause: {moved}" + ); + tracing::error!("{parked}"); + println!("{parked}; parking as blocked_corrupt."); + park_blocked(crosslink_dir, identity, should_exit, &parked)?; + return Ok(false); + } + tracing::warn!("{moved}; reconciling again"); + println!("{moved}; reconciling again."); + reconcile_reason = Some(moved); + if wait_interruptible(should_exit, Duration::from_secs(retry_seconds)) { + return Ok(false); + } + retry_seconds = (retry_seconds * 2).min(MAX_RETRY_SECS); + } + Err(error) => return Err(error), } } } @@ -672,53 +791,44 @@ fn record_ready( enum NormalLoopExit { Shutdown, - RecordExpired, + Reconcile { + reason: String, + completed_ticks: u64, + }, +} + +enum Housekeeping { + Done { active_issue_id: Option }, + Reconcile(String), } fn run_normal_loop(crosslink_dir: &Path, should_exit: &AtomicBool) -> Result { let db_path = crosslink_dir.join("issues.db"); let session_file = crosslink_dir.join("session.json"); let mut heartbeat_counter = 0_u64; + let mut completed_ticks = 0_u64; while !wait_interruptible(should_exit, Duration::from_secs(FLUSH_INTERVAL_SECS)) { let identity = readiness::read_daemon_identity(crosslink_dir)? .ok_or_else(|| anyhow::anyhow!("daemon identity disappeared"))?; if defer_reconciliation_for_active_mutations(crosslink_dir, &identity)? { continue; } - let Some(mutation_operation) = - acquire_housekeeping_unless_expired(crosslink_dir, &identity)? - else { - return Ok(NormalLoopExit::RecordExpired); - }; - let mut active_issue_id = None; - let db = Database::open(&db_path).context("opening projection for daemon housekeeping")?; - crate::hydration::maybe_auto_hydrate_under_operation(crosslink_dir, &db) - .context("hydrating current authority during daemon housekeeping")?; - let service = crate::application::RepositoryService::projection(&db); - let agent_id = crate::identity::AgentConfig::load(crosslink_dir) - .ok() - .flatten() - .map(|agent| agent.agent_id); - if let Ok(Some(session)) = - crate::application::LocalStateService::get_current_session_for_agent( - &service, - agent_id.as_deref(), - ) - { - active_issue_id = session.active_issue_id; - let data = serde_json::json!({ - "session_id": session.id, - "started_at": session.started_at.to_rfc3339(), - "active_issue_id": session.active_issue_id, - }); - if let Ok(bytes) = serde_json::to_vec_pretty(&data) { - if let Err(error) = fs::write(&session_file, bytes) { - tracing::warn!("failed to write session file: {error}"); - } + let active_issue_id = match housekeeping_tick( + crosslink_dir, + &identity, + &db_path, + &session_file, + || Ok(()), + )? { + Housekeeping::Done { active_issue_id } => active_issue_id, + Housekeeping::Reconcile(reason) => { + return Ok(NormalLoopExit::Reconcile { + reason, + completed_ticks, + }) } - } - refresh_ready_record_with(crosslink_dir, &identity, || Ok(()))?; - drop(mutation_operation); + }; + completed_ticks += 1; heartbeat_counter += 1; if heartbeat_counter.is_multiple_of(5) { if let Err(error) = run_sync_tick(crosslink_dir, &db_path, active_issue_id) { @@ -729,6 +839,86 @@ fn run_normal_loop(crosslink_dir: &Path, should_exit: &AtomicBool) -> Result( + crosslink_dir: &Path, + identity: &DaemonIdentity, + db_path: &Path, + session_file: &Path, + mut before_refresh_write: F, +) -> Result +where + F: FnMut() -> Result<()>, +{ + let Some(mutation_operation) = acquire_housekeeping_unless_expired(crosslink_dir, identity)? + else { + return Ok(Housekeeping::Reconcile( + "readiness record expired".to_string(), + )); + }; + let db = Database::open(db_path).context("opening projection for daemon housekeeping")?; + let mut attempts = 0_u32; + loop { + attempts += 1; + if let Err(error) = crate::hydration::maybe_auto_hydrate_under_operation(crosslink_dir, &db) + { + drop(mutation_operation); + return Ok(Housekeeping::Reconcile(format!( + "hydrating current authority during daemon housekeeping failed: {error:#}" + ))); + } + match refresh_ready_record_with(crosslink_dir, identity, &mut before_refresh_write) { + Ok(_) => break, + Err(error) + if attempts < MAX_RECORD_REFRESH_ATTEMPTS + && error.downcast_ref::().is_some() => + { + tracing::debug!( + "the hub moved during the readiness refresh (attempt {attempts}); hydrating again: {error:#}" + ); + } + Err(error) => { + drop(mutation_operation); + return Ok(Housekeeping::Reconcile(format!( + "refreshing the readiness record failed after {attempts} attempt(s): {error:#}" + ))); + } + } + } + let service = crate::application::RepositoryService::projection(&db); + let agent_id = crate::identity::AgentConfig::load(crosslink_dir) + .ok() + .flatten() + .map(|agent| agent.agent_id); + let mut active_issue_id = None; + if let Ok(Some(session)) = crate::application::LocalStateService::get_current_session_for_agent( + &service, + agent_id.as_deref(), + ) { + active_issue_id = session.active_issue_id; + let data = serde_json::json!({ + "session_id": session.id, + "started_at": session.started_at.to_rfc3339(), + "active_issue_id": session.active_issue_id, + }); + if let Ok(bytes) = serde_json::to_vec_pretty(&data) { + if let Err(error) = fs::write(session_file, bytes) { + tracing::warn!("failed to write session file: {error}"); + } + } + } + drop(mutation_operation); + Ok(Housekeeping::Done { active_issue_id }) +} + fn acquire_housekeeping_unless_expired( crosslink_dir: &Path, identity: &DaemonIdentity, @@ -767,7 +957,7 @@ fn defer_reconciliation_for_active_mutations_with( before_refresh_write: F, ) -> Result where - F: FnOnce() -> Result<()>, + F: FnMut() -> Result<()>, { if !readiness::has_active_mutation_permits(crosslink_dir)? { return Ok(false); @@ -781,10 +971,10 @@ where fn refresh_ready_record_with( crosslink_dir: &Path, identity: &DaemonIdentity, - before_write: F, + mut before_write: F, ) -> Result where - F: FnOnce() -> Result<()>, + F: FnMut() -> Result<()>, { let Some(record) = readiness::read_record(crosslink_dir)? else { return Ok(false); @@ -1080,7 +1270,7 @@ mod tests { let worker_identity = identity; let worker_exit = Arc::clone(&should_exit); let worker = thread::spawn(move || { - reconcile_until_ready_loop(&worker_dir, &worker_identity, &worker_exit) + reconcile_until_ready(&worker_dir, &worker_identity, &worker_exit, None) }); let deadline = Instant::now() + Duration::from_secs(5); let first = loop { @@ -1527,13 +1717,260 @@ mod tests { assert!(acquire_housekeeping_unless_expired(&crosslink, &identity) .unwrap() .is_none()); - assert!(reconcile_until_ready(&crosslink, &identity, &AtomicBool::new(false)).unwrap()); + assert!( + reconcile_until_ready(&crosslink, &identity, &AtomicBool::new(false), None).unwrap() + ); let record = readiness::read_record(&crosslink).unwrap().unwrap(); assert_eq!(record.daemon_epoch, identity.daemon_epoch); readiness::validate_record(&crosslink, &record).unwrap(); assert!(readiness::require_mutation_ready(&crosslink).is_ok()); } + /// Publish one event to the hub as another checkout would, with its own + /// sequence so repeated races stay valid. + fn publish_as_racing_agent(cache: &Path, agent_seq: u64) -> Result<()> { + crate::hub_v3::append_event_to_ref( + cache, + "racing-agent", + &crate::events::EventEnvelope { + agent_id: "racing-agent".to_string(), + agent_seq, + timestamp: chrono::Utc::now(), + event: crate::events::Event::LockClaimed { + issue_display_id: 1, + branch: None, + }, + signed_by: None, + signature: None, + }, + )?; + Ok(()) + } + + #[test] + fn authority_moving_during_every_record_write_reconciles_instead_of_failing() { + let (_work, _remote, crosslink, identity) = ready_connected(); + let sync = SyncManager::new(&crosslink).unwrap(); + let cache = sync.cache_path().to_path_buf(); + let db_path = crosslink.join("issues.db"); + let session_file = crosslink.join("session.json"); + // Another checkout publishes between the daemon's validation and its + // record write, every time the daemon tries: the race a busy hub of + // kickoff worktrees produces. + let mut races = 0; + let outcome = housekeeping_tick(&crosslink, &identity, &db_path, &session_file, || { + races += 1; + publish_as_racing_agent(&cache, races) + }) + .unwrap(); + assert_eq!(races, u64::from(MAX_RECORD_REFRESH_ATTEMPTS)); + let Housekeeping::Reconcile(reason) = outcome else { + panic!("the moved authority must send housekeeping back to reconciliation"); + }; + assert!( + reason.contains("refusing to publish ready state for a stale projection"), + "{reason}" + ); + assert!(reason.contains("after 3 attempt"), "{reason}"); + assert!(reconcile_until_ready( + &crosslink, + &identity, + &AtomicBool::new(false), + Some(&reason) + ) + .unwrap()); + let record = readiness::read_record(&crosslink).unwrap().unwrap(); + assert_eq!(record.daemon_epoch, identity.daemon_epoch); + readiness::validate_record(&crosslink, &record).unwrap(); + assert!(readiness::projection_is_current(&crosslink).unwrap()); + } + + #[test] + fn a_single_race_during_the_record_write_is_absorbed_within_the_tick() { + let (_work, _remote, crosslink, identity) = ready_connected(); + let sync = SyncManager::new(&crosslink).unwrap(); + let cache = sync.cache_path().to_path_buf(); + let before = readiness::read_record(&crosslink).unwrap().unwrap(); + let mut calls = 0; + let outcome = housekeeping_tick( + &crosslink, + &identity, + &crosslink.join("issues.db"), + &crosslink.join("session.json"), + || { + calls += 1; + if calls == 1 { + publish_as_racing_agent(&cache, 1) + } else { + Ok(()) + } + }, + ) + .unwrap(); + assert_eq!(calls, 2, "the tick hydrates again and refreshes once more"); + assert!(matches!(outcome, Housekeeping::Done { .. })); + let after = readiness::read_record(&crosslink).unwrap().unwrap(); + assert!(after.sequence > before.sequence); + assert_eq!(after.attempt_id, before.attempt_id); + assert!(readiness::projection_is_current(&crosslink).unwrap()); + } + + #[test] + fn authority_moving_during_the_reconciliation_record_write_reconciles_again() { + let (_work, _remote, crosslink, identity) = ready_connected(); + let sync = SyncManager::new(&crosslink).unwrap(); + let cache = sync.cache_path().to_path_buf(); + // Another checkout publishes between activation and the ready-record + // write of the reconciliation path itself. + let mut calls = 0; + let ready = reconcile_until_ready_with( + &crosslink, + &identity, + &AtomicBool::new(false), + None, + || { + calls += 1; + if calls == 1 { + publish_as_racing_agent(&cache, 1) + } else { + Ok(()) + } + }, + ) + .unwrap(); + assert!(ready); + assert_eq!( + calls, 2, + "the first write is refused and activation runs again" + ); + let record = readiness::read_record(&crosslink).unwrap().unwrap(); + assert!(record.state.grants_mutations()); + assert_eq!(record.daemon_epoch, identity.daemon_epoch); + readiness::validate_record(&crosslink, &record).unwrap(); + assert!(readiness::projection_is_current(&crosslink).unwrap()); + } + + #[test] + fn a_hub_moving_under_every_reconciliation_write_parks_with_the_cause() { + let (_work, _remote, crosslink, identity) = ready_connected(); + let sync = SyncManager::new(&crosslink).unwrap(); + let cache = sync.cache_path().to_path_buf(); + let should_exit = Arc::new(AtomicBool::new(false)); + let worker_dir = crosslink.clone(); + let worker_exit = Arc::clone(&should_exit); + let worker = thread::spawn(move || { + let mut races = 0; + reconcile_until_ready_with(&worker_dir, &identity, &worker_exit, None, || { + races += 1; + publish_as_racing_agent(&cache, races) + }) + .map(|ready| (ready, races)) + }); + // 1 + 2 + 4 + 8 s of backoff precede the fifth attempt; allow for the + // activations around them on a slow machine. + let deadline = Instant::now() + Duration::from_secs(120); + let parked = loop { + if let Some(record) = readiness::read_record(&crosslink).unwrap() { + if record.state == ReadinessState::BlockedCorrupt { + break record; + } + } + assert!(Instant::now() < deadline, "the daemon did not park"); + thread::sleep(Duration::from_millis(100)); + }; + should_exit.store(true, Ordering::SeqCst); + let (ready, races) = worker.join().unwrap().unwrap(); + assert!(!ready); + assert_eq!(races, MAX_CONSECUTIVE_RECONCILES as u64); + let reason = parked.reason.unwrap_or_default(); + assert!(reason.contains("consecutive reconciliations"), "{reason}"); + assert!( + reason.contains("refusing to publish ready state for a stale projection"), + "{reason}" + ); + } + + /// Unix only: the failure is injected through directory permissions. + #[cfg(unix)] + #[test] + fn hydration_failure_during_housekeeping_reconciles_instead_of_failing() { + use std::os::unix::fs::PermissionsExt; + if Command::new("id") + .arg("-u") + .output() + .is_ok_and(|output| String::from_utf8_lossy(&output.stdout).trim() == "0") + { + eprintln!( + "skipped: root ignores directory permissions, so the failure cannot be injected" + ); + return; + } + let (_work, _remote, crosslink, identity) = ready_connected(); + let sync = SyncManager::new(&crosslink).unwrap(); + // The hub moves, so the tick must hydrate, and the hydration cannot + // record the new frontier because its directory is not writable. + publish_as_racing_agent(sync.cache_path(), 1).unwrap(); + let frontier_dir = crosslink.join(crate::hydration::HYDRATED_FRONTIER_DIR); + fs::set_permissions(&frontier_dir, fs::Permissions::from_mode(0o555)).unwrap(); + let outcome = housekeeping_tick( + &crosslink, + &identity, + &crosslink.join("issues.db"), + &crosslink.join("session.json"), + || Ok(()), + ); + fs::set_permissions(&frontier_dir, fs::Permissions::from_mode(0o755)).unwrap(); + let Housekeeping::Reconcile(reason) = outcome.unwrap() else { + panic!( + "a hydration that cannot complete must send housekeeping back to reconciliation" + ); + }; + assert!( + reason.contains("hydrating current authority during daemon housekeeping failed"), + "{reason}" + ); + } + + #[test] + fn the_reconcile_budget_parks_persistent_faults_and_resets_on_a_completed_tick() { + let mut budget = ReconcileBudget::default(); + for _ in 1..MAX_CONSECUTIVE_RECONCILES { + assert!(!budget.note_reconcile(0)); + } + assert!( + budget.note_reconcile(0), + "the limit is reached without a completed tick" + ); + + let mut budget = ReconcileBudget::default(); + for _ in 1..MAX_CONSECUTIVE_RECONCILES { + assert!(!budget.note_reconcile(0)); + } + assert!( + !budget.note_reconcile(1), + "a completed tick resets the count" + ); + assert_eq!(budget.consecutive, 1); + } + + #[test] + fn a_quiet_housekeeping_tick_completes_and_keeps_the_record_fresh() { + let (_work, _remote, crosslink, identity) = ready_connected(); + let before = readiness::read_record(&crosslink).unwrap().unwrap(); + let outcome = housekeeping_tick( + &crosslink, + &identity, + &crosslink.join("issues.db"), + &crosslink.join("session.json"), + || Ok(()), + ) + .unwrap(); + assert!(matches!(outcome, Housekeeping::Done { .. })); + let after = readiness::read_record(&crosslink).unwrap().unwrap(); + assert!(after.sequence > before.sequence); + assert_eq!(after.attempt_id, before.attempt_id); + } + #[test] fn expired_record_of_another_daemon_is_still_an_error() { let (_work, _remote, crosslink, identity) = ready_connected(); @@ -1694,7 +2131,7 @@ mod tests { let worker_exit = Arc::clone(&should_exit); let (done_tx, done_rx) = std::sync::mpsc::sync_channel(1); let worker = thread::spawn(move || { - let result = reconcile_until_ready(&worker_dir, &worker_identity, &worker_exit); + let result = reconcile_until_ready(&worker_dir, &worker_identity, &worker_exit, None); let _ = done_tx.send(result); }); let deadline = Instant::now() + Duration::from_secs(5); diff --git a/crosslink/src/hydration.rs b/crosslink/src/hydration.rs index 152607522..e37567c3b 100644 --- a/crosslink/src/hydration.rs +++ b/crosslink/src/hydration.rs @@ -1338,7 +1338,7 @@ fn hydrate_relations( } const LAST_HYDRATED_REF_FILE: &str = ".last-hydrated-ref"; -const HYDRATED_FRONTIER_DIR: &str = "hydrated-frontiers"; +pub(crate) const HYDRATED_FRONTIER_DIR: &str = "hydrated-frontiers"; const MAX_HYDRATED_FRONTIERS: usize = 64; pub fn hydrate_current_authority_under_operation( diff --git a/crosslink/src/reconcile/migration.rs b/crosslink/src/reconcile/migration.rs index de9bd55c1..0f720285f 100644 --- a/crosslink/src/reconcile/migration.rs +++ b/crosslink/src/reconcile/migration.rs @@ -97,24 +97,29 @@ pub(crate) fn write_ready_activation( }, )?; anyhow::ensure!( - readiness::projection_frontier(crosslink_dir)?.is_some() - && readiness::projection_is_current(crosslink_dir)?, + readiness::projection_frontier(crosslink_dir)?.is_some(), "reconciliation completed without a projection frontier" ); + if !readiness::projection_is_current(crosslink_dir)? { + return Err(anyhow::Error::new(readiness::StaleProjection) + .context("the hub moved while reconciliation completed")); + } anyhow::ensure!( readiness::projection_schema_version(crosslink_dir)? == Some(crate::db::SCHEMA_VERSION), "reconciliation completed without a current projection schema" ); let sync = SyncManager::new(crosslink_dir)?; - anyhow::ensure!( - crate::reconcile::publication::generation_id_at_ref( - sync.cache_path(), - crate::reconcile::publication::GENERATION_REF, - )? - .as_deref() - == Some(generation_id), - "reconciliation generation identifier does not match the verified descriptor" - ); + if crate::reconcile::publication::generation_id_at_ref( + sync.cache_path(), + crate::reconcile::publication::GENERATION_REF, + )? + .as_deref() + != Some(generation_id) + { + return Err(anyhow::Error::new(readiness::StaleProjection).context( + "the hub generation moved while reconciliation completed, so the verified descriptor is no longer current", + )); + } readiness::write_record( crosslink_dir, ReadinessDraft { diff --git a/crosslink/src/reconcile/readiness.rs b/crosslink/src/reconcile/readiness.rs index 755b43e00..4f0686f71 100644 --- a/crosslink/src/reconcile/readiness.rs +++ b/crosslink/src/reconcile/readiness.rs @@ -17,6 +17,20 @@ use uuid::Uuid; use super::publication::GENERATION_REF; +/// The projection moved under a ready-state write: another checkout published +/// to the hub between validation and the write. Callers that can re-derive +/// readiness treat this as "try again", not as a fault. +#[derive(Debug, Clone, Copy)] +pub struct StaleProjection; + +impl std::fmt::Display for StaleProjection { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("refusing to publish ready state for a stale projection") + } +} + +impl std::error::Error for StaleProjection {} + pub const READINESS_SCHEMA_VERSION: u32 = 1; pub const READINESS_PROTOCOL_VERSION: u32 = 1; const READINESS_DIR: &str = "readiness"; @@ -639,10 +653,9 @@ pub fn write_record(crosslink_dir: &Path, draft: ReadinessDraft<'_>) -> Result return Err(error), }; if draft.state.grants_mutations() { - anyhow::ensure!( - projection_frontier.is_some() && projection_is_current(crosslink_dir)?, - "refusing to publish ready state for a stale projection" - ); + if !(projection_frontier.is_some() && projection_is_current(crosslink_dir)?) { + return Err(StaleProjection.into()); + } anyhow::ensure!( projection_schema_version == Some(crate::db::SCHEMA_VERSION), "refusing to publish ready state for an incompatible projection schema"