From 424300d67219b59f741fd53eb92f99b996925f77 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 1 Oct 2026 03:31:48 +0200 Subject: [PATCH] A publish that loses to a held lock is retried before the fence, not aborted resume_candidate turned any failed pre-Fencing step into a pre-fencing abort, so a topology publish that met a held lock (the nightly backup's shared lock, an operator) ended the deploy or continuity restart on its first attempt. LockContended is now a typed error (drivers::LockContended, raised by mutate and IncidentStore::exchange with the same texts as before) and resume_candidate's one classification keeps such a transaction in its phase with last_error set; the backoff retries it and the phase deadline bounds the wait. Other publish errors still abort. Tests at the resume path: a contended Starting publish stays in Starting across ticks and lands after release with one continuity restart; a non-contention publish error still aborts. Both builds. Co-Authored-By: Claude Sonnet 5.5 --- src/control_plane.rs | 91 +++++++++++++++++++++++++++++++++++++++++++- src/drivers.rs | 18 ++++++++- src/incident.rs | 4 +- 3 files changed, 108 insertions(+), 5 deletions(-) diff --git a/src/control_plane.rs b/src/control_plane.rs index 75af38f..dd2962f 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -40,7 +40,7 @@ use crate::deployment_plan::{ use crate::drivers::{ ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FrozenSourceReceipt, GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver, - ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators, + LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators, RouteObservation, RoutePreflightReceipt, RunnerPort, SourcePort, SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort, WriteLeasePort, @@ -5576,7 +5576,14 @@ impl Engine { // Before the fence an abort that cannot finish a step is // resumable, never a post-fence abort: that path refuses a // pre-fence phase, so choosing it only wedged the tick. - if latest.value.pre_fencing_abort.is_none() { + // + // A publish that lost to a held lock is not a failed step: the + // holder lets go, so the transaction stays in its phase with + // the error recorded and the backoff above retries it. The + // phase deadline, not an abort, bounds the wait. + if latest.value.pre_fencing_abort.is_none() + && error.downcast_ref::().is_none() + { self.begin_pre_fencing_abort(latest, error)?; } else { self.record_resumable_error(latest, &error)?; @@ -15626,6 +15633,86 @@ mod tests { Ok(()) } + /// A Continuity transaction parked in Starting with nothing published yet, + /// plus the stored record the tick will see. + #[cfg(target_os = "linux")] + fn parked_in_starting(world: &EngineFixture) -> Result { + let seeded = transaction_at(world, DeploymentPhase::Warming)?; + let stored = resident(world)?; + let mut starting = seeded.clone(); + starting.phase = DeploymentPhase::Starting; + starting.expected_publication_sha256 = None; + starting.activation = None; + starting.workload = None; + starting.activation_publication_sha256 = None; + starting.warming = None; + replace_transaction(&world.state_store, &stored, &starting)?; + Ok(seeded) + } + + /// A publish that lost to a held lock is transient: before the fence the + /// transaction stays in its phase with the error recorded, no abort is + /// begun, and the next tick after the release lands the publish. A + /// continuity restart is a Continuity transaction on this same path, so + /// its attempt is minted once and a contended publish spends no second one. + #[cfg(target_os = "linux")] + #[test] + fn a_contended_starting_publish_is_retried_on_a_later_tick_not_aborted() -> Result<()> { + use crate::incident::tests::{hold_lock_of, run_while_held}; + + let world = EngineFixture::with_workload(Arc::new(StillWorkload))?; + let seeded = parked_in_starting(&world)?; + let mut meters = TargetSupervision::new("service"); + meters.continuity_restarts = vec![now_millis()?]; + set_meters(&world, &meters)?; + let store = world.engine.options.topology_store.clone(); + + for tick in 0..2 { + run_while_held(hold_lock_of(&store), 1, || world.engine.resume_one_transaction())?; + world.engine.resume_backoff.lock().unwrap().clear(); + let held = resident(&world)?.value; + assert_eq!(held.phase, DeploymentPhase::Starting, "tick {tick}"); + assert!(held.pre_fencing_abort.is_none(), "tick {tick}: a held lock aborted the transaction"); + assert!(!held.is_terminal(), "tick {tick}"); + assert_eq!(held.expected_publication_sha256, None, "tick {tick}"); + let error = held.last_error.context("a contended publish left no last_error")?; + assert!(error.contains("locked by another holder"), "{error}"); + assert!(!store.exists(), "a contended publish wrote the projection"); + } + + // Released: the next tick lands the publish and records its digest. + world.engine.resume_one_transaction()?; + let landed = resident(&world)?.value; + assert_eq!(landed.phase, DeploymentPhase::Starting); + assert!(landed.pre_fencing_abort.is_none()); + assert_eq!( + landed.expected_publication_sha256, + Some(seeded.expected.context("seeded transaction has no Expected")?.canonical_sha256()?) + ); + assert_eq!(meters_of(&world, "service")?.continuity_restarts.len(), 1); + Ok(()) + } + + /// Only contention is transient: a publish that fails for any other reason + /// still ends the transaction before the fence, as it always did. + #[cfg(target_os = "linux")] + #[test] + fn a_non_contention_publish_error_still_aborts_before_the_fence() -> Result<()> { + let world = EngineFixture::with_workload(Arc::new(StillWorkload))?; + parked_in_starting(&world)?; + // A directory where the store should be: the snapshot cannot be read. + std::fs::create_dir_all(&world.engine.options.topology_store)?; + world.engine.resume_one_transaction()?; + let after = resident(&world)?.value; + assert!(after.pre_fencing_abort.is_some(), "a non-contention error did not abort"); + assert!( + !after.last_error.as_deref().unwrap_or_default().contains("locked by another holder"), + "{:?}", + after.last_error + ); + Ok(()) + } + #[test] fn status_prints_an_open_incident_under_its_target() -> Result<()> { let workload = SwitchWorkload::new(); diff --git a/src/drivers.rs b/src/drivers.rs index ada688c..e005ec5 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -4414,7 +4414,7 @@ impl CultCacheTopologyDriver { } TryCompareExchangeSnapshotOutcome::Mismatch => {} TryCompareExchangeSnapshotOutcome::LockContended => { - bail!("the topology projection is locked by another holder") + return Err(LockContended { store: "topology projection" }.into()); } } } @@ -5107,6 +5107,22 @@ pub enum RouteActuation { Survival, } +/// A publish that lost to a held projection-directory lock. Transient by +/// kind: the holder lets go, and the caller's own retry decides what a later +/// attempt does. It names the store kind, never its path. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LockContended { + pub(crate) store: &'static str, +} + +impl std::fmt::Display for LockContended { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "the {} is locked by another holder", self.store) + } +} + +impl std::error::Error for LockContended {} + /// A forward route actuation the ceiling refused. `reopens_at_unix_millis` is /// when the oldest counted actuation leaves the window. #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/src/incident.rs b/src/incident.rs index b1b413a..1d7f095 100644 --- a/src/incident.rs +++ b/src/incident.rs @@ -20,7 +20,7 @@ use cultcache_rs::{ use serde::{Deserialize, Serialize}; use crate::control_plane::{decode_record, require_id, typed_envelope}; -use crate::drivers::{exchange_behind_private_lock, publish_file_mode}; +use crate::drivers::{LockContended, exchange_behind_private_lock, publish_file_mode}; const INCIDENT_SCHEMA: &str = "idunn.operator_incident.v1"; @@ -238,7 +238,7 @@ impl IncidentStore { bail!("the incident store changed while it was being written") } TryCompareExchangeSnapshotOutcome::LockContended => { - bail!("the incident store is locked by another holder") + Err(LockContended { store: "incident store" }.into()) } } }