GameCult/Idunn
Watch
9
0
Fork
You've already forked Idunn
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Idunn: a static flip that moved current can no longer end before fencing

Browse source 
carry_and_flip returns FlipRefused only before the rename (the last fallible
step); the parent-directory sync after it is best-effort. Once a flip intent is
recorded, any error that is not FlipRefused leaves the transaction resumable
instead of taking the pre-fencing abort, and a resumed step that finds current
already naming the release records the carry already in it (new port method
carried_sha256) instead of flipping again. validate refuses a pre-fencing abort
beside a landed flip.

Tests: post-rename port fault and post-landing observe fault both end Admitted
with current on the new release; validation refuses FailedBeforeFencing with a
landed flip; a refused flip over a real directory writes no carry; a populated
v4 record lifts to the exact v5 field by field.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-11-13 16:44:04 +00:00
commit dc86c6326c
2 changed files with 334 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

195
src/control_plane.rs
View file

@ -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::<FlipRefused>().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::<LockContended>().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<Self> {
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<Option<String>> {
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<StaticFlip> {
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<String> {
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<DeploymentTransaction> {
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
}

230
src/drivers.rs
View file

@ -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<StaticFlip>;
/// 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<String>;
}
#[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<String> {
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<StaticFlip> {
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<StaticFlip> {
bail!("a static release flip requires Unix symlinks")
}
fn carried_sha256(&self, _: &StaticReleaseBinding, _: &str) -> Result<String> {
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::<FlipRefused>().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::<FlipRefused>().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