diff --git a/src/control_plane.rs b/src/control_plane.rs index b52cbc6..819195a 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -40,10 +40,10 @@ use crate::deployment_plan::{ }; use crate::drivers::{ ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FilesystemStaticRelease, - FrozenSourceReceipt, GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver, + FlipRefused, FrozenSourceReceipt, GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver, LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators, RouteObservation, RoutePreflightReceipt, RunnerPort, RunnerStepFailure, SourcePort, - StaticReleasePort, SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort, + StaticFlip, StaticReleasePort, SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort, WriteLeasePort, require_sha256_id, }; use crate::incident::{CloseReason, IncidentCondition, IncidentStore, render_incidents}; @@ -2136,6 +2136,14 @@ impl DeploymentTransaction { .is_some_and(|(_, binding)| matches!(binding.workload, WorkloadBinding::StaticRelease(_))) } + /// Whether `error` can no longer end this transaction before fencing: a + /// static flip is recorded as intended and only `FlipRefused` (the link did + /// not move) says it did not happen. Any other error after the intent may + /// follow the rename, so the step is retried, and a retry observes the link. + fn flip_may_have_moved_current(&self, error: &anyhow::Error) -> bool { + self.flip.is_some() && error.downcast_ref::().is_none() + } + /// The runtime and the sealed release a deployment brake release must name. /// One reading for both kinds of target: the binding's runtime identity and /// the sealed release, which a service's Expected projection copies. @@ -2644,6 +2652,11 @@ impl DeploymentTransaction { } if let Some(abort) = &self.pre_fencing_abort { require_detail(&abort.error, "pre-fencing abort error")?; + ensure!( + !self.flip.as_ref().is_some_and(FlipEvidence::landed), + "a transaction whose flip landed cannot fail before fencing" + ); + ensure!( self.post_commit_cleanup.is_none(), "aborted transaction also carries post-commit cleanup" @@ -6738,6 +6751,7 @@ impl Engine { // phase deadline, not an abort, bounds the wait. if latest.value.pre_fencing_abort.is_none() && error.downcast_ref::().is_none() + && !latest.value.flip_may_have_moved_current(&error) { if let Some(line) = self.keep_runner_failure(&latest.value, &error) { eprintln!("{line}"); @@ -7917,9 +7931,21 @@ impl Engine { }); }; if !flip.landed() { - let landed = self - .static_release - .carry_and_flip(workload, &flip.release_id)?; + // An earlier attempt may have moved the link and failed to record + // it: then nothing is flipped again, and the carry already inside + // the release is what is recorded. + let landed = if self.static_release.observe_current(workload)?.as_deref() + == Some(flip.release_id.as_str()) + { + StaticFlip { + carried_sha256: self + .static_release + .carried_sha256(workload, &flip.release_id)?, + } + } else { + self.static_release + .carry_and_flip(workload, &flip.release_id)? + }; return self.persist_same_phase(current, |next| { let flip = next.flip.as_mut().expect("flip intent read above"); flip.carried_sha256 = Some(landed.carried_sha256); @@ -26130,5 +26156,164 @@ GAMECULT_RUNTIME_PRESENCE_IDENTITY = "/etc/gamecult/service/runtime-presence-ide Ok(()) } + + /// The filesystem port, with faults injected where the real one could only + /// fail after the link had moved: `flip_fault_after_rename` fails the flip + /// call after the rename landed, `observe_fault_at` fails the nth observe. + struct FaultyPort { + inner: FilesystemStaticRelease, + flip_fault_after_rename: std::sync::atomic::AtomicBool, + observe_calls: std::sync::atomic::AtomicUsize, + observe_fault_at: usize, + } + + impl FaultyPort { + fn new(flip_fault_after_rename: bool, observe_fault_at: usize) -> Arc { + Arc::new(Self { + inner: FilesystemStaticRelease, + flip_fault_after_rename: flip_fault_after_rename.into(), + observe_calls: 0.into(), + observe_fault_at, + }) + } + } + + impl crate::drivers::StaticReleasePort for FaultyPort { + fn observe_current(&self, binding: &crate::deployment::StaticReleaseBinding) -> Result> { + use std::sync::atomic::Ordering; + let call = self.observe_calls.fetch_add(1, Ordering::SeqCst) + 1; + ensure!(call != self.observe_fault_at, "injected observe fault (call {call})"); + self.inner.observe_current(binding) + } + + fn carry_and_flip( + &self, + binding: &crate::deployment::StaticReleaseBinding, + release_id: &str, + ) -> Result { + let flip = self.inner.carry_and_flip(binding, release_id)?; + if self.flip_fault_after_rename.swap(false, std::sync::atomic::Ordering::SeqCst) { + bail!("injected fault after the rename"); + } + Ok(flip) + } + + fn carried_sha256( + &self, + binding: &crate::deployment::StaticReleaseBinding, + release_id: &str, + ) -> Result { + self.inner.carried_sha256(binding, release_id) + } + } + + /// Drive the first Deploy through the brake to its end. + fn drive_static_to_end(w: &StaticWorld) -> Result { + let id = w.transaction_of("up-service")?; + let waiting = w.run_to_brake(&id)?; + w.release_brake(&waiting)?; + w.tick_until(&id, "the end", |t| t.is_terminal())?; + latest_id(&w.world, &id) + } + + #[cfg(unix)] + #[test] + fn static_release_error_after_the_rename_never_ends_before_fencing() -> Result<()> { + for (flip_fault, observe_fault_at) in [(true, 0), (false, 4)] { + let mut w = static_world()?; + w.world.engine.static_release = FaultyPort::new(flip_fault, observe_fault_at); + let done = drive_static_to_end(&w)?; + assert!( + matches!(done.completion, Some(TransactionCompletion::Admitted { .. })), + "a post-rename fault ended the transaction: {:?} {:?}", + done.completion, + done.last_error + ); + assert!(done.pre_fencing_abort.is_none()); + let flip = done.flip.as_ref().context("no flip")?; + assert!(flip.landed()); + assert_eq!(w.current_release().as_deref(), Some(flip.release_id.as_str())); + assert_eq!( + flip.carried_sha256.as_deref(), + Some(sha256_of(SITE_CONFIG_ONE).as_str()) + ); + } + Ok(()) + } + + #[cfg(unix)] + #[test] + fn static_release_validation_refuses_a_landed_flip_that_failed_before_fencing() -> Result<()> { + let w = static_world()?; + let mut done = drive_static_to_end(&w)?; + assert!(done.flip.as_ref().is_some_and(FlipEvidence::landed)); + done.completion = Some(TransactionCompletion::FailedBeforeFencing { error: "x".into() }); + done.pre_fencing_abort = Some(PreFencingAbort { + error: "x".into(), + candidate_cleanup: CleanupEvidence::Skipped, + topology_reconciliation: CleanupEvidence::Skipped, + source_cleanup: CleanupEvidence::Complete, + }); + done.post_commit_cleanup = None; + done.phase_deadline = None; + let error = done.validate().unwrap_err(); + assert!(format!("{error:#}").contains("flip landed"), "{error:#}"); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_populated_v4_transaction_lifts_to_the_exact_v5_field_by_field() -> Result<()> { + let w = static_world()?; + let done = drive_static_to_end(&w)?; + // A populated record a v4 Idunn could have written: a deploy that + // consumed its brake and ended before fencing. + let mut expected = done.clone(); + expected.flip = None; + expected.completion = Some(TransactionCompletion::FailedBeforeFencing { error: "refused".into() }); + expected.pre_fencing_abort = Some(PreFencingAbort { + error: "refused".into(), + candidate_cleanup: CleanupEvidence::Skipped, + topology_reconciliation: CleanupEvidence::Skipped, + source_cleanup: CleanupEvidence::Complete, + }); + expected.post_commit_cleanup = None; + expected.phase_deadline = None; + expected.last_error = Some("refused".into()); + expected.validate()?; + assert_ne!(expected.created_at_unix_millis, expected.updated_at_unix_millis); + assert!(expected.completion.is_some() && expected.deployment_authorization.is_some()); + // The v4 bytes of this record: the v5 encoding with the old schema name, + // one element shorter (key 37 is the last, a nil here). + let mut as_v4 = expected.clone(); + as_v4.schema_version = DEPLOYMENT_TRANSACTION_SCHEMA_V4.into(); + let mut payload = rmp_serde::to_vec(&as_v4)?; + assert_eq!(&payload[..3], &[0xdc, 0x00, 0x26]); + payload[2] = 0x25; + assert_eq!(payload.pop(), Some(0xc0)); + let envelope = CultCacheEnvelope { + key: expected.transaction_id.clone(), + r#type: DeploymentTransaction::TYPE.into(), + payload, + stored_at: rfc3339_millis(100)?, + schema_id: Some(DEPLOYMENT_TRANSACTION_SCHEMA_V4.into()), + }; + let lifted = read_transaction_record(&envelope)?; + assert_eq!(lifted.schema_version, DEPLOYMENT_TRANSACTION_SCHEMA); + assert_eq!(lifted.transaction_id, expected.transaction_id); + assert_eq!(lifted.created_at_unix_millis, expected.created_at_unix_millis); + assert_eq!(lifted.updated_at_unix_millis, expected.updated_at_unix_millis); + assert_eq!(lifted.completion, expected.completion); + assert_eq!(lifted.plan, expected.plan); + assert_eq!(lifted.sealed_release, expected.sealed_release); + assert_eq!(lifted.installed_release, expected.installed_release); + assert_eq!(lifted.frozen_source, expected.frozen_source); + assert_eq!(lifted.deployment_authorization, expected.deployment_authorization); + assert_eq!(lifted.post_commit_cleanup, expected.post_commit_cleanup); + assert_eq!(lifted.phase_deadline, expected.phase_deadline); + assert_eq!(lifted.flip, None); + assert_eq!(lifted, expected, "some field changed in the lift"); + Ok(()) + } // END static-release tests } diff --git a/src/drivers.rs b/src/drivers.rs index 67068f2..605ccc8 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -634,6 +634,20 @@ pub struct StaticFlip { pub carried_sha256: String, } +/// A flip that was refused before the link moved. It is the only error that may +/// end a static transaction before fencing: once the rename has happened, +/// nothing may say the flip did not. +#[derive(Debug)] +pub(crate) struct FlipRefused(pub(crate) String); + +impl std::fmt::Display for FlipRefused { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.0) + } +} + +impl std::error::Error for FlipRefused {} + /// The one external change a static release makes: copy the operator's site /// file into an installed release and repoint the current link at it. The port /// decides nothing; the engine decides when, and records what it reports. @@ -646,8 +660,15 @@ pub trait StaticReleasePort { /// Copy the carry into `release_id` and repoint the link at it. Nothing /// moves unless the carry source was read, so a missing site file leaves /// the link where it was. Repeating it is harmless. + /// + /// An error is `FlipRefused` exactly when the link did not move; the rename + /// is the last fallible step, and what follows it cannot fail the flip. fn carry_and_flip(&self, binding: &StaticReleaseBinding, release_id: &str) -> Result; + + /// The digest of the carry already inside an installed release: what a + /// resumed flip records when an earlier attempt landed. + fn carried_sha256(&self, binding: &StaticReleaseBinding, release_id: &str) -> Result; } #[derive(Clone, Copy, Debug, Default)] @@ -694,6 +715,22 @@ impl StaticReleasePort for FilesystemStaticRelease { Ok(Some(id.to_owned())) } + fn carried_sha256(&self, binding: &StaticReleaseBinding, release_id: &str) -> Result { + require_sha256_id(release_id, "release carrying the site file")?; + let carried = binding + .release_root + .join(release_id) + .join(&binding.carry.destination); + ensure!( + fs::symlink_metadata(&carried)?.is_file(), + "carried {} is not a regular file", + carried.display() + ); + let mut hasher = Sha256::new(); + hash_file_exact(&carried, fs::metadata(&carried)?.len(), &mut hasher)?; + Ok(format!("sha256-{:x}", hasher.finalize())) + } + fn carry_and_flip( &self, binding: &StaticReleaseBinding, @@ -701,94 +738,107 @@ impl StaticReleasePort for FilesystemStaticRelease { ) -> Result { use std::os::unix::fs::{OpenOptionsExt, symlink}; - require_sha256_id(release_id, "release to flip to")?; - let release = binding.release_root.join(release_id); - let release_metadata = fs::symlink_metadata(&release) - .with_context(|| format!("installed release {}", release.display()))?; - ensure!( - release_metadata.is_dir(), - "installed release {} is not a directory", - release.display() - ); - let link_parent = binding - .current_link - .parent() - .context("current link has no parent")?; - match fs::symlink_metadata(&binding.current_link) { - Ok(metadata) => ensure!( - metadata.file_type().is_symlink(), - "current {} is not a symlink and will not be replaced", - binding.current_link.display() - ), - Err(error) if error.kind() == ErrorKind::NotFound => {} - Err(error) => return Err(error).context("reading current link"), - } - - // Everything that can fail on the operator's side fails here, before - // the link moves. - let source = &binding.carry.source; - let source_metadata = fs::symlink_metadata(source) - .with_context(|| format!("reading carry source {}", source.display()))?; - ensure!( - source_metadata.is_file(), - "carry source {} is not a regular file", - source.display() - ); - ensure!( - source_metadata.len() <= STATIC_CARRY_LIMIT_BYTES, - "carry source {} is larger than {STATIC_CARRY_LIMIT_BYTES} bytes", - source.display() - ); - let mut bytes = Vec::new(); - fs::File::open(source)? - .take(STATIC_CARRY_LIMIT_BYTES + 1) - .read_to_end(&mut bytes)?; - ensure!( - bytes.len() as u64 <= STATIC_CARRY_LIMIT_BYTES, - "carry source {} grew past {STATIC_CARRY_LIMIT_BYTES} bytes", - source.display() - ); - let carried_sha256 = format!("sha256-{:x}", Sha256::digest(&bytes)); - - let carried = release.join(&binding.carry.destination); - let temporary_carry = release.join(format!(".carry-{}", Uuid::new_v4())); - let written = (|| -> Result<()> { - let mut file = OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o444) - .open(&temporary_carry)?; - file.write_all(&bytes)?; - file.sync_all()?; - fs::rename(&temporary_carry, &carried)?; - fs::File::open(&release)?.sync_all()?; - Ok(()) - })(); - if let Err(error) = written { - let _ = fs::remove_file(&temporary_carry); - return Err(error) - .with_context(|| format!("carrying the site file into {}", release.display())); - } - - // A link is renamed over, never edited: a reader sees the old release - // or the new one. A temporary link a crashed flip left is replaced. - let temporary_link = link_parent.join(format!( - ".{}.flip", - binding + // Everything up to and including the rename can refuse, and only that + // refuses: the link has not moved when `FlipRefused` is returned. + let refuse = |error: anyhow::Error| anyhow::Error::new(FlipRefused(format!("{error:#}"))); + let staged = (|| -> Result<(PathBuf, PathBuf, String)> { + require_sha256_id(release_id, "release to flip to")?; + let release = binding.release_root.join(release_id); + let release_metadata = fs::symlink_metadata(&release) + .with_context(|| format!("installed release {}", release.display()))?; + ensure!( + release_metadata.is_dir(), + "installed release {} is not a directory", + release.display() + ); + let link_parent = binding .current_link - .file_name() - .and_then(OsStr::to_str) - .context("current link has no name")? - )); - if fs::symlink_metadata(&temporary_link).is_ok() { - fs::remove_file(&temporary_link).context("removing a stale flip link")?; - } - symlink(&release, &temporary_link).context("creating the flip link")?; + .parent() + .context("current link has no parent")?; + match fs::symlink_metadata(&binding.current_link) { + Ok(metadata) => ensure!( + metadata.file_type().is_symlink(), + "current {} is not a symlink and will not be replaced", + binding.current_link.display() + ), + Err(error) if error.kind() == ErrorKind::NotFound => {} + Err(error) => return Err(error).context("reading current link"), + } + + let source = &binding.carry.source; + let source_metadata = fs::symlink_metadata(source) + .with_context(|| format!("reading carry source {}", source.display()))?; + ensure!( + source_metadata.is_file(), + "carry source {} is not a regular file", + source.display() + ); + ensure!( + source_metadata.len() <= STATIC_CARRY_LIMIT_BYTES, + "carry source {} is larger than {STATIC_CARRY_LIMIT_BYTES} bytes", + source.display() + ); + let mut bytes = Vec::new(); + fs::File::open(source)? + .take(STATIC_CARRY_LIMIT_BYTES + 1) + .read_to_end(&mut bytes)?; + ensure!( + bytes.len() as u64 <= STATIC_CARRY_LIMIT_BYTES, + "carry source {} grew past {STATIC_CARRY_LIMIT_BYTES} bytes", + source.display() + ); + let carried_sha256 = format!("sha256-{:x}", Sha256::digest(&bytes)); + + let carried = release.join(&binding.carry.destination); + let temporary_carry = release.join(format!(".carry-{}", Uuid::new_v4())); + let written = (|| -> Result<()> { + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o444) + .open(&temporary_carry)?; + file.write_all(&bytes)?; + file.sync_all()?; + fs::rename(&temporary_carry, &carried)?; + fs::File::open(&release)?.sync_all()?; + Ok(()) + })(); + if let Err(error) = written { + let _ = fs::remove_file(&temporary_carry); + return Err(error) + .with_context(|| format!("carrying the site file into {}", release.display())); + } + + // A link is renamed over, never edited: a reader sees the old + // release or the new one. A temporary link a crashed flip left is + // replaced. + let temporary_link = link_parent.join(format!( + ".{}.flip", + binding + .current_link + .file_name() + .and_then(OsStr::to_str) + .context("current link has no name")? + )); + if fs::symlink_metadata(&temporary_link).is_ok() { + fs::remove_file(&temporary_link).context("removing a stale flip link")?; + } + symlink(&release, &temporary_link).context("creating the flip link")?; + Ok((temporary_link, release, carried_sha256)) + })() + .map_err(refuse)?; + let (temporary_link, _, carried_sha256) = staged; if let Err(error) = fs::rename(&temporary_link, &binding.current_link) { let _ = fs::remove_file(&temporary_link); - return Err(error).context("repointing the current link"); + return Err(refuse( + anyhow::Error::new(error).context("repointing the current link"), + )); + } + // The link has moved and nothing below can say it did not. The parent + // directory sync is durability only, so it is best-effort. + if let Some(parent) = binding.current_link.parent() { + let _ = fs::File::open(parent).and_then(|directory| directory.sync_all()); } - fs::File::open(link_parent)?.sync_all()?; Ok(StaticFlip { carried_sha256 }) } } @@ -802,6 +852,10 @@ impl StaticReleasePort for FilesystemStaticRelease { fn carry_and_flip(&self, _: &StaticReleaseBinding, _: &str) -> Result { bail!("a static release flip requires Unix symlinks") } + + fn carried_sha256(&self, _: &StaticReleaseBinding, _: &str) -> Result { + bail!("a static release flip requires Unix symlinks") + } } pub trait TopologyPort { @@ -16970,6 +17024,7 @@ processes 9 fs::remove_file(&site.binding.carry.source)?; let error = port.carry_and_flip(&site.binding, RELEASE_B).unwrap_err(); assert!(format!("{error:#}").contains("carry source"), "{error:#}"); + assert!(error.downcast_ref::().is_some(), "a pre-rename refusal is typed"); assert_eq!(port.observe_current(&site.binding)?.as_deref(), Some(RELEASE_A)); assert!( !site.binding.release_root.join(RELEASE_B).join("config.json").exists(), @@ -17018,8 +17073,13 @@ processes 9 let port = FilesystemStaticRelease; // A real directory where the link belongs is left alone. fs::create_dir(&site.binding.current_link)?; - assert!(port.carry_and_flip(&site.binding, RELEASE_A).is_err()); + let error = port.carry_and_flip(&site.binding, RELEASE_A).unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); assert!(site.binding.current_link.is_dir()); + assert!( + !site.binding.release_root.join(RELEASE_A).join("config.json").exists(), + "a refused flip wrote its carry" + ); assert!(port.observe_current(&site.binding).is_err()); fs::remove_dir(&site.binding.current_link)?; // A link that names something outside the release root has no release