diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index eb308245d..33772d04b 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -916,6 +916,35 @@ Graph cleanup uses that same history capacity, counting active and archived records together, so repeated restores cannot exceed a smaller cleanup-only ceiling after allocation succeeds. Matching ownership, container bindings and unambiguous records are still required. +Retired graph history is compacted under the engine mutation lease where no retained +effect identity still binds the run's environment inventory: right after an ordinary +(non-relay) `graph cleanup` commits, and in every graph restore (foreground +`owner-restore` redelivery and fresh-owner `serve-restore`) after all prior slots are +re-retired and the acknowledged stopped receipt is superseded on disk, before the new +generation allocates. Enrolled relay cleanup (`graph cleanup` of a relay-owned graph, +or owner shutdown) does not compact, because its acknowledged effect is re-verified +against the inventory; its last generation stays active until the next restore or +archive. Each compacted record must match a receipt container that is absent and a +guest slot that is absent, or nothing is written. Exact intent digests move into one +value-free per-run tombstone in `run/environment-lease-tombstones`, bound to the VM +incarnation, and only then are the active records unlinked. Prior boots fold into one +sealed summary each; the current boot keeps per-slot reservations (at most 16384, about +1170 generations of a 14-service graph in one boot). A retired slot or any slot of a +sealed boot is never recorded, delivered, retired as live or counted again. Restore and +cleanup therefore scan only the active generation, and repeated restores no longer +consume the 4096 active-record budget; tombstone reads still grow with the current +boot's per-slot reservations until the next boot seals them. +Pending, live, changed, misbound or foreign-incarnation records refuse compaction, +and other runs' records are untouched. A refusal never fails or reverts the cleanup +or restore that triggered it: the records stay active as exact evidence and the next +cleanup, restore or archive retries; a run that keeps refusing eventually meets the +active-record budget's `environment_recovery` admission refusal. An interruption is +resumed by the next compaction or by archive; exact residue reads as retired, changed +residue refuses. Tombstones are retained permanently as slot-ID reservations, like +standalone retention exports: archive and prune do not remove them, so the directory +holds one file per environment-attached run: about 160 bytes per current-boot +retired slot plus one summary per sealed boot, capped at 4 MiB and refused, never +truncated, beyond that. No retained evidence is automatically deleted to admit new work. This environment budget does not increase the runtime's separate container, CPU or memory limits and does not establish 32-branch qualification. diff --git a/packages/runtime-core/src/provider/environment_recovery.rs b/packages/runtime-core/src/provider/environment_recovery.rs index 6ec1ea917..9d6d82295 100644 --- a/packages/runtime-core/src/provider/environment_recovery.rs +++ b/packages/runtime-core/src/provider/environment_recovery.rs @@ -1,7 +1,9 @@ //! Immutable, value-free allocation intent. Recovery grants cleanup, never renewed delivery. +mod compaction; mod retention; use super::{environment::EnvironmentLease, lifecycle::OwnedGuest, state}; use crate::{Candidate, CandidateError}; +pub(super) use compaction::{Settled, compact_graph, resume_graph}; pub use retention::{RetiredExport, export_retired}; use serde::{Deserialize, Serialize}; use std::{fs, path::PathBuf}; @@ -148,29 +150,39 @@ pub(super) fn record( if lease.native.is_some() { return Err(error()); } - if retention::reserved(candidate, &lease.slot)? { + record_intent( + candidate, + &Intent { + version: 1, + graph: lease.graph.clone(), + service: lease.service.clone(), + uid: lease.uid, + gid: lease.gid, + slot: lease.slot.clone(), + incarnation: lease.incarnation.clone(), + boot: lease.boot.clone(), + }, + ) +} +/// A slot reserved by standalone export or by its graph's compaction tombstone is never reused. +fn record_intent(candidate: &Candidate, intent: &Intent) -> Result<(), CandidateError> { + if retention::reserved(candidate, &intent.slot)? + || intent.graph.as_ref().map_or(Ok(false), |binding| { + compaction::reserved(candidate, &binding.run, &intent.incarnation, &intent.slot) + })? + { return Err(error()); } let directory = root(candidate); state::private_directory(&directory)?; preflight_records(candidate, 1)?; - let intent = Intent { - version: 1, - graph: lease.graph.clone(), - service: lease.service.clone(), - uid: lease.uid, - gid: lease.gid, - slot: lease.slot.clone(), - incarnation: lease.incarnation.clone(), - boot: lease.boot.clone(), - }; - validate(&intent, &lease.slot)?; - let path = directory.join(format!("{}.json", lease.slot)); + validate(intent, &intent.slot)?; + let path = directory.join(format!("{}.json", intent.slot)); match fs::symlink_metadata(&path) { Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} _ => return Err(error()), } - state::write(&path, &intent) + state::write(&path, intent) } fn read( candidate: &Candidate, @@ -287,11 +299,20 @@ pub(super) fn graph_slots( guest: &OwnedGuest<'_>, run: &str, ) -> Result, CandidateError> { + graph_slots_for(candidate, guest.incarnation(), run) +} +/// Exact residue of a durable compaction is retired history, not an allocation to retire again. +fn graph_slots_for( + candidate: &Candidate, + incarnation: &str, + run: &str, +) -> Result, CandidateError> { + let tombstone = compaction::load(candidate, run, incarnation)?; let mut bindings = Vec::new(); for slot in recorded_slots(candidate)? { - let intent = read_mode(candidate, &slot, guest.incarnation(), None, false)?; + let intent = read_mode(candidate, &slot, incarnation, None, false)?; if let Some(binding) = intent.graph { - if binding.run == run { + if binding.run == run && !compaction::residue(candidate, tombstone.as_ref(), &slot)? { bindings.push((slot, intent.service, binding)); } } @@ -347,6 +368,7 @@ fn active_exec_slot_for( || binding.run != run || binding.container != container || retention::reserved(candidate, slot)? + || compaction::reserved(candidate, run, incarnation, slot)? { return Err(error()); } @@ -386,6 +408,7 @@ pub(super) fn graph_inventory( if !hex(run, 32) || !hex(incarnation, 32) { return Err(error()); } + let tombstone = compaction::load(candidate, run, incarnation)?; let mut selected = std::collections::BTreeMap::new(); let mut insert = |intent: Intent| -> Result<(), CandidateError> { validate(&intent, &intent.slot)?; @@ -394,6 +417,7 @@ pub(super) fn graph_inventory( g.run == run && containers.get(&intent.service) == Some(&g.container) }) || retention::reserved(candidate, &intent.slot)? + || compaction::reserves(tombstone.as_ref(), &intent.slot) || selected.len() >= MAX_GRAPH_INTENT_ENTRIES || selected.insert(intent.slot.clone(), intent).is_some() { @@ -403,7 +427,9 @@ pub(super) fn graph_inventory( }; for slot in recorded_slots(candidate)? { let intent = read_mode(candidate, &slot, incarnation, None, false)?; - if intent.graph.as_ref().is_some_and(|g| g.run == run) { + if intent.graph.as_ref().is_some_and(|g| g.run == run) + && !compaction::residue(candidate, tombstone.as_ref(), &slot)? + { insert(intent)?; } } @@ -568,9 +594,24 @@ pub(super) fn retire( if retention::reserved(candidate, slot)? { return retention::verify_retired(candidate, guest, slot, lease); } - let intent = read(candidate, slot, guest.incarnation(), lease)?; + let intent = live_intent(candidate, guest.incarnation(), slot, lease)?; retire_intent(guest, &intent) } +/// The recorded intent a retirement may act on. A slot reserved by its graph's compaction +/// tombstone is retained history (exact residue or a replay), never a live allocation. +fn live_intent( + candidate: &Candidate, + incarnation: &str, + slot: &str, + lease: Option<&EnvironmentLease>, +) -> Result { + if let Some(binding) = read_mode(candidate, slot, incarnation, lease, false)?.graph { + if compaction::reserved(candidate, &binding.run, incarnation, slot)? { + return Err(error()); + } + } + read(candidate, slot, incarnation, lease) +} fn retire_intent(guest: &OwnedGuest<'_>, intent: &Intent) -> Result<(), CandidateError> { let slot = &intent.slot; if let Some(binding) = &intent.graph { diff --git a/packages/runtime-core/src/provider/environment_recovery/compaction.rs b/packages/runtime-core/src/provider/environment_recovery/compaction.rs new file mode 100644 index 000000000..ba7310599 --- /dev/null +++ b/packages/runtime-core/src/provider/environment_recovery/compaction.rs @@ -0,0 +1,484 @@ +//! Compaction of retired graph-bound allocation intents. +//! +//! Ordinary graph cleanup retires every bound slot under the provider mutation lease, but each +//! immutable intent used to stay in `run/environment-leases` indefinitely. Admission, cleanup and +//! restore then scanned every historical generation of a repeatedly started graph. After the +//! caller confirms each recorded container and guest slot absent, compaction retains the exact +//! intent digest in one per-run tombstone and unlinks the active record. +//! +//! Commit and recovery points, in order: +//! 1. The tombstone gains per-slot `retired` digests (atomic rename plus directory sync). No +//! active record is unlinked before this is durable, so a leftover tombstone temporary file +//! is an uncommitted write with no authority and is discarded. +//! 2. Active records whose exact bytes match a retired digest are unlinked; the directory is +//! synced. Until then such a record is residue: read-only inventory treats it as retired and +//! the next compaction completes the unlink. Changed bytes are refused, never removed. +//! 3. Retired slots of prior guest boots are folded into one sealed summary per boot. A prior +//! boot cannot allocate again, so the whole boot stays reserved without per-slot entries. +//! +//! Trigger points. Compaction rewrites the run's environment inventory, so it runs only where no +//! retained effect identity still binds that inventory: +//! - ordinary `graph cleanup` (no relay enrollment), right after its cleaned receipt commits; +//! - graph restore (foreground redelivery and fresh-owner serve-restore), after every prior slot +//! is re-retired and the acknowledged stopped receipt is superseded on disk, before the new +//! generation allocates. Enrolled relay cleanup itself never compacts: its acknowledged +//! effect, which includes the inventory, is re-verified until a restore supersedes it; +//! - archive, which only resumes an interrupted compaction. +//! +//! A refusal after a committed cleanup is [`Settled::Retained`], never a failure of that cleanup. +//! +//! Reservation: a retired slot, or any slot of a sealed boot, can never be recorded again, +//! selected for delivery, retired as live, or counted as a live allocation. Records are never +//! deleted to make room: an uncertain (pending), live, changed, misbound or foreign-owned record +//! refuses the whole compaction before its first write. Records of other runs, standalone +//! records and other incarnations' state are never modified. +use super::*; +use sha2::{Digest, Sha256}; +use std::collections::{BTreeMap, BTreeSet}; +use std::os::unix::fs::MetadataExt; +use std::path::Path; + +const VERSION: u32 = 1; +/// Per-slot reservations retained for boots that may still allocate (normally the current one). +const MAX_RETIRED_SLOTS: usize = 16 * 1024; +const MAX_SEALED_BOOTS: usize = 4096; +/// Covers both maxima in pretty-printed JSON with margin; larger state is refused, not truncated. +const TOMBSTONE_LIMIT: u64 = 4 * 1024 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Sealed { + slots: u32, + sha256: String, +} + +/// Value-free retained evidence for one graph run. Bound to the owning VM incarnation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Tombstone { + version: u32, + run: String, + incarnation: String, + sealed: BTreeMap, + retired: BTreeMap, +} + +#[derive(Debug, PartialEq, Eq)] +enum Reservation<'a> { + Free, + Retired(&'a str), + Sealed, +} + +/// Value-free outcome counts for one compaction. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)] +pub(crate) struct Compaction { + /// Newly confirmed records moved into the tombstone. + pub compacted: usize, + /// Residue of an interrupted compaction whose unlink was completed. + pub completed: usize, + /// Prior boots folded into sealed reservations. + pub sealed_boots: usize, +} + +/// Outcome of compaction attempted after the caller's cleanup already committed. A refusal +/// (uncertain or changed record, retired-slot cap, sealed-boot cap, malformed tombstone, I/O) +/// reverts nothing and is not reported as a failure of that cleanup: every record it did not +/// durably retire stays active as exact evidence, and the next confirmed cleanup, restore or +/// archive retries. A run that keeps refusing eventually meets the active-record budget, whose +/// admission refusal is the closed `environment_recovery` error. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Settled { + Compacted(Compaction), + Retained, +} + +impl From> for Settled { + fn from(result: Result) -> Self { + result.map_or(Self::Retained, Self::Compacted) + } +} + +/// `Resume` only completes an interrupted compaction: it selects no new record and seals nothing, +/// so it is safe where live allocations of the run may remain. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Mode { + Compact, + Resume, +} + +/// Owner identity and receipt bindings for one compaction. +#[derive(Clone, Copy)] +pub(super) struct Request<'a> { + pub mode: Mode, + /// VM incarnation that owns every record; others refuse. + pub incarnation: &'a str, + /// Current guest boot; it may still allocate, so it is never sealed. + pub boot: &'a str, + pub run: &'a str, + /// Receipt service-to-container bindings each selected record must match. + pub containers: &'a BTreeMap, +} + +/// Fault boundaries observed by in-process interruption tests. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Step { + Recorded, + Unlinked, + Synced, + Sealed, +} + +fn directory(candidate: &Candidate) -> PathBuf { + candidate + .state_root + .join("run/environment-lease-tombstones") +} + +/// Boot UUID of a `hack-env-lease--` slot. Callers validate the slot first; an +/// invalid identity yields an empty boot that matches no reservation. +fn boot_of(slot: &str) -> &str { + let start = "hack-env-lease-".len(); + slot.get(start..start + 36).unwrap_or("") +} + +fn digest(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + +fn present(path: &Path) -> Result { + match fs::symlink_metadata(path) { + Ok(_) => Ok(true), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(e) => Err(state::io(e)), + } +} + +fn sync(directory: &Path) -> Result<(), CandidateError> { + fs::File::open(directory) + .and_then(|f| f.sync_all()) + .map_err(state::io) +} + +impl Tombstone { + fn new(run: &str, incarnation: &str) -> Self { + Self { + version: VERSION, + run: run.into(), + incarnation: incarnation.into(), + sealed: BTreeMap::new(), + retired: BTreeMap::new(), + } + } + + fn validate(&self, run: &str, incarnation: &str) -> Result<(), CandidateError> { + if self.version != VERSION + || self.run != run + || self.incarnation != incarnation + || !hex(run, 32) + || !hex(incarnation, 32) + || self.retired.len() > MAX_RETIRED_SLOTS + || self.sealed.len() > MAX_SEALED_BOOTS + || self + .sealed + .iter() + .any(|(boot, sealed)| !uuid(boot) || sealed.slots == 0 || !hex(&sealed.sha256, 64)) + || self.retired.iter().any(|(slot, sha)| { + !valid_slot(slot) || !hex(sha, 64) || self.sealed.contains_key(boot_of(slot)) + }) + { + return Err(error()); + } + Ok(()) + } + + fn reservation(&self, slot: &str) -> Reservation<'_> { + if !valid_slot(slot) || self.sealed.contains_key(boot_of(slot)) { + return Reservation::Sealed; + } + match self.retired.get(slot) { + Some(sha) => Reservation::Retired(sha), + None => Reservation::Free, + } + } +} + +fn load_at(path: &Path, run: &str, incarnation: &str) -> Result, CandidateError> { + if !present(path)? { + return Ok(None); + } + let tombstone: Tombstone = state::read_bounded(path, TOMBSTONE_LIMIT)?; + tombstone.validate(run, incarnation)?; + Ok(Some(tombstone)) +} + +/// Read-only. A run that cannot name a valid binding has no tombstone to consult. +pub(super) fn load( + candidate: &Candidate, + run: &str, + incarnation: &str, +) -> Result, CandidateError> { + if !hex(run, 32) { + return Ok(None); + } + let parent = directory(candidate); + crate::reject_aliased_state(&parent)?; + if !parent.try_exists().map_err(state::io)? { + return Ok(None); + } + state::check_private_directory(&parent)?; + load_at(&parent.join(format!("{run}.json")), run, incarnation) +} + +/// Whether an already loaded tombstone reserves the slot. +pub(super) fn reserves(tombstone: Option<&Tombstone>, slot: &str) -> bool { + tombstone.is_some_and(|tombstone| tombstone.reservation(slot) != Reservation::Free) +} + +/// Whether the slot is reserved by retained evidence, whether or not residue remains active. +pub(super) fn reserved( + candidate: &Candidate, + run: &str, + incarnation: &str, + slot: &str, +) -> Result { + Ok(reserves(load(candidate, run, incarnation)?.as_ref(), slot)) +} + +/// Read-only classification for inventory. `true` means the committed active record is exact +/// residue of a durable compaction: retired, never live, and removed by the next compaction. +/// A sealed-boot replay, changed bytes or an uncertain pending record refuses. +pub(super) fn residue( + candidate: &Candidate, + tombstone: Option<&Tombstone>, + slot: &str, +) -> Result { + let Some(tombstone) = tombstone else { + return Ok(false); + }; + match tombstone.reservation(slot) { + Reservation::Free => Ok(false), + Reservation::Sealed => Err(error()), + Reservation::Retired(sha) => { + let committed = root(candidate).join(format!("{slot}.json")); + if present(&committed.with_extension("pending"))? + || digest(&retention::bytes(&committed)?) != sha + { + return Err(error()); + } + Ok(true) + } + } +} + +/// A leftover temporary file means the rename never committed (see module commit point 1). +fn discard_uncommitted(path: &Path) -> Result<(), CandidateError> { + let metadata = match fs::symlink_metadata(path) { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(e) => return Err(state::io(e)), + Ok(metadata) => metadata, + }; + // SAFETY: geteuid has no preconditions. + if !metadata.file_type().is_file() + || metadata.nlink() != 1 + || metadata.uid() != unsafe { libc::geteuid() } + || metadata.len() > TOMBSTONE_LIMIT + { + return Err(error()); + } + fs::remove_file(path).map_err(state::io)?; + sync(path.parent().ok_or_else(error)?) +} + +/// Commit point: atomic rename plus parent sync. Only the uncommitted temporary is replaced. +fn commit(parent: &Path, path: &Path, tombstone: &Tombstone) -> Result<(), CandidateError> { + state::private_directory(parent)?; + discard_uncommitted(&path.with_extension("pending"))?; + state::write(path, tombstone) +} + +/// Compact every confirmed-retired record bound to `run`. The caller holds the provider mutation +/// lease, has completed graph cleanup and passes the receipt's service-to-container bindings. +pub(crate) fn compact_graph( + candidate: &Candidate, + guest: &OwnedGuest<'_>, + run: &str, + containers: &BTreeMap, +) -> Result { + compact_with( + candidate, + Request { + mode: Mode::Compact, + incarnation: guest.incarnation(), + boot: guest.boot_id(), + run, + containers, + }, + |intent| { + if let Some(binding) = &intent.graph { + super::super::engine::require_container_absent(guest, &binding.container)?; + } + retention::absent(guest, &intent.slot) + }, + |_| Ok(()), + ) +} + +/// Complete only an interrupted compaction of `run`, under the caller's mutation lease. Exact +/// residue is unlinked; every other record is left untouched. No tombstone means no effect. +pub(crate) fn resume_graph( + candidate: &Candidate, + guest: &OwnedGuest<'_>, + run: &str, +) -> Result { + if load(candidate, run, guest.incarnation())?.is_none() { + return Ok(Compaction::default()); + } + compact_with( + candidate, + Request { + mode: Mode::Resume, + incarnation: guest.incarnation(), + boot: guest.boot_id(), + run, + containers: &BTreeMap::new(), + }, + |_| Err(error()), + |_| Ok(()), + ) +} + +/// Filesystem state machine behind [`compact_graph`]. `confirm` proves one selected allocation +/// absent; `fault` lets tests stop at each commit boundary as an interruption would. +pub(super) fn compact_with( + candidate: &Candidate, + request: Request<'_>, + confirm: impl Fn(&Intent) -> Result<(), CandidateError>, + mut fault: impl FnMut(Step) -> Result<(), CandidateError>, +) -> Result { + let Request { + mode, + incarnation, + boot, + run, + containers, + } = request; + if !hex(run, 32) || !hex(incarnation, 32) || !uuid(boot) { + return Err(error()); + } + let parent = directory(candidate); + let path = parent.join(format!("{run}.json")); + let mut tombstone = + load(candidate, run, incarnation)?.unwrap_or_else(|| Tombstone::new(run, incarnation)); + if tombstone.sealed.contains_key(boot) { + return Err(error()); + } + + // Inventory and confirm every record bound to this run before the first write. + let active = root(candidate); + let mut residue = Vec::new(); + let mut selected = BTreeMap::new(); + for slot in recorded_slots(candidate)? { + let intent = read_mode(candidate, &slot, incarnation, None, false)?; + let Some(binding) = intent.graph.as_ref().filter(|binding| binding.run == run) else { + continue; + }; + let reservation = tombstone.reservation(&slot); + if mode == Mode::Resume && reservation == Reservation::Free { + continue; + } + let committed = active.join(format!("{slot}.json")); + if present(&committed.with_extension("pending"))? { + return Err(error()); + } + let sha = digest(&retention::bytes(&committed)?); + match reservation { + Reservation::Sealed => return Err(error()), + Reservation::Retired(expected) if expected == sha => residue.push(slot), + Reservation::Retired(_) => return Err(error()), + Reservation::Free => { + if containers.get(&intent.service) != Some(&binding.container) + || retention::reserved(candidate, &slot)? + { + return Err(error()); + } + confirm(&intent)?; + selected.insert(slot, sha); + } + } + } + let outcome = Compaction { + compacted: selected.len(), + completed: residue.len(), + sealed_boots: 0, + }; + + if !selected.is_empty() { + if tombstone.retired.len() + selected.len() > MAX_RETIRED_SLOTS { + return Err(error()); + } + tombstone.retired.extend(selected.clone()); + commit(&parent, &path, &tombstone)?; + fault(Step::Recorded)?; + } + for slot in residue.iter().chain(selected.keys()) { + let committed = active.join(format!("{slot}.json")); + if tombstone.retired.get(slot) != Some(&digest(&retention::bytes(&committed)?)) { + return Err(error()); + } + fs::remove_file(&committed).map_err(state::io)?; + fault(Step::Unlinked)?; + } + if outcome.compacted + outcome.completed > 0 { + sync(&active)?; + fault(Step::Synced)?; + } + + // Resume may leave unselected records of this run active, so it never seals a boot. + if mode == Mode::Resume { + return Ok(outcome); + } + // Every record of this run was unlinked above or refused the compaction, so no active + // allocation of a prior boot remains. Its whole boot is then reserved by one summary. + let prior: BTreeSet = tombstone + .retired + .keys() + .map(|slot| boot_of(slot).to_owned()) + .filter(|retired| retired != boot) + .collect(); + if prior.is_empty() { + return Ok(outcome); + } + if tombstone.sealed.len() + prior.len() > MAX_SEALED_BOOTS { + return Err(error()); + } + for retired in &prior { + let slots: Vec<(String, String)> = tombstone + .retired + .iter() + .filter(|(slot, _)| boot_of(slot) == retired) + .map(|(slot, sha)| (slot.clone(), sha.clone())) + .collect(); + let mut summary = Sha256::new(); + for (slot, sha) in &slots { + summary.update(format!("{slot} {sha}\n")); + tombstone.retired.remove(slot); + } + tombstone.sealed.insert( + retired.clone(), + Sealed { + slots: u32::try_from(slots.len()).map_err(|_| error())?, + sha256: format!("{:x}", summary.finalize()), + }, + ); + } + commit(&parent, &path, &tombstone)?; + fault(Step::Sealed)?; + Ok(Compaction { + sealed_boots: prior.len(), + ..outcome + }) +} + +#[cfg(test)] +mod tests; diff --git a/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs b/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs new file mode 100644 index 000000000..fdc039aa1 --- /dev/null +++ b/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs @@ -0,0 +1,881 @@ +//! Filesystem-only tests. Guest absence is injected through `confirm`; no VM, engine, +//! real HOME or process interruption is involved. Faults stop the state machine in-process. +use super::super::{ + active_exec_slot_for, graph_slots_for, live_intent, preflight_records, record_intent, +}; +use super::*; +use std::{ + cell::Cell, + io::Read, + os::unix::fs::{MetadataExt, OpenOptionsExt}, +}; + +const RUN: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +const OTHER_RUN: &str = "cccccccccccccccccccccccccccccccc"; +const INCARNATION: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; +const FOREIGN_INCARNATION: &str = "dddddddddddddddddddddddddddddddd"; +const PRIOR: &str = "11111111-1111-4111-8111-111111111111"; +const CURRENT: &str = "22222222-2222-4222-8222-222222222222"; +const SERVICES: usize = 14; + +/// Bytes, inode and modification time of each directory entry. +type Snapshot = BTreeMap, u64, i64, i64)>; + +struct Fixture { + candidate: Candidate, + next: Cell, +} + +impl Fixture { + fn new() -> Self { + let mut bytes = [0; 16]; + fs::File::open("/dev/urandom") + .unwrap() + .read_exact(&mut bytes) + .unwrap(); + let token: String = bytes.iter().map(|b| format!("{b:02x}")).collect(); + let path = std::env::temp_dir() + .canonicalize() + .unwrap() + .join(format!("hack-env-compaction-{token}")); + state::private_directory(&path).unwrap(); + let candidate = Candidate::discover(&path).unwrap(); + state::private_directory(&root(&candidate)).unwrap(); + Self { + candidate, + next: Cell::new(0), + } + } + + fn intent(&self, run: Option<&str>, boot: &str, service: usize) -> Intent { + let token = self.next.get(); + self.next.set(token + 1); + Intent { + version: 1, + graph: run.map(|run| GraphBinding { + run: run.into(), + container: format!("hkg-{run}-container-{service}"), + }), + service: format!("svc{service}"), + uid: 0, + gid: 0, + slot: format!("hack-env-lease-{boot}-{token:032x}"), + incarnation: INCARNATION.into(), + boot: boot.into(), + } + } + + /// One start of every service: a generation of committed records. + fn generation(&self, run: &str, boot: &str) -> Vec { + (0..SERVICES) + .map(|service| { + let intent = self.intent(Some(run), boot, service); + self.commit(&intent); + intent + }) + .collect() + } + + fn commit(&self, intent: &Intent) { + state::write(&self.active(&intent.slot), intent).unwrap(); + } + + fn active(&self, slot: &str) -> PathBuf { + root(&self.candidate).join(format!("{slot}.json")) + } + + fn tombstone_path(&self, run: &str) -> PathBuf { + directory(&self.candidate).join(format!("{run}.json")) + } + + fn compact( + &self, + mode: Mode, + incarnation: &str, + boot: &str, + run: &str, + fault: impl FnMut(Step) -> Result<(), CandidateError>, + ) -> Result { + compact_with( + &self.candidate, + Request { + mode, + incarnation, + boot, + run, + containers: &containers(run), + }, + |_| Ok(()), + fault, + ) + } + + fn compact_now(&self, run: &str) -> Result { + self.compact(Mode::Compact, INCARNATION, CURRENT, run, |_| Ok(())) + } + + fn slots(&self, run: &str) -> Result, CandidateError> { + Ok(graph_slots_for(&self.candidate, INCARNATION, run)? + .into_iter() + .map(|(slot, _, _)| slot) + .collect()) + } + + /// Bytes plus inode and modification time of every entry, to prove "untouched". + fn snapshot(&self, directory: &Path) -> Snapshot { + let Ok(entries) = fs::read_dir(directory) else { + return BTreeMap::new(); + }; + entries + .map(|entry| { + let entry = entry.unwrap(); + let metadata = fs::symlink_metadata(entry.path()).unwrap(); + let bytes = if metadata.file_type().is_symlink() { + fs::read_link(entry.path()) + .unwrap() + .into_os_string() + .into_encoded_bytes() + } else { + fs::read(entry.path()).unwrap() + }; + ( + entry.file_name().into_string().unwrap(), + ( + bytes, + metadata.ino(), + metadata.mtime(), + metadata.mtime_nsec(), + ), + ) + }) + .collect() + } + + fn state(&self) -> (Snapshot, Snapshot) { + ( + self.snapshot(&root(&self.candidate)), + self.snapshot(&directory(&self.candidate)), + ) + } + + /// Records of another run, a standalone record and a pending record of another run. + fn foreign(&self) -> Vec { + let other = self.intent(Some(OTHER_RUN), CURRENT, 0); + self.commit(&other); + let standalone = self.intent(None, CURRENT, 1); + self.commit(&standalone); + let uncertain = self.intent(Some(OTHER_RUN), CURRENT, 2); + let pending = self.active(&uncertain.slot).with_extension("pending"); + write_raw(&pending, &serde_json::to_vec(&uncertain).unwrap()); + vec![ + self.active(&other.slot), + self.active(&standalone.slot), + pending, + ] + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + fs::remove_dir_all(&self.candidate.checkout).unwrap(); + } +} + +fn containers(run: &str) -> BTreeMap { + (0..SERVICES) + .map(|service| { + ( + format!("svc{service}"), + format!("hkg-{run}-container-{service}"), + ) + }) + .collect() +} + +fn write_raw(path: &Path, bytes: &[u8]) { + std::io::Write::write_all( + &mut fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(path) + .unwrap(), + bytes, + ) + .unwrap(); +} + +fn stop_at(target: usize) -> impl FnMut(Step) -> Result<(), CandidateError> { + let mut seen = 0; + move |_| { + seen += 1; + if seen == target + 1 { + return Err(CandidateError::new("test_interrupted", "Interrupted.")); + } + Ok(()) + } +} + +#[test] +fn compaction_retains_digests_seals_prior_boots_and_frees_active_budget() { + let fixture = Fixture::new(); + let mut retired = fixture.generation(RUN, PRIOR); + retired.extend(fixture.generation(RUN, PRIOR)); + let current = fixture.generation(RUN, CURRENT); + retired.extend(current.clone()); + let originals: BTreeMap> = retired + .iter() + .map(|intent| { + ( + intent.slot.clone(), + fs::read(fixture.active(&intent.slot)).unwrap(), + ) + }) + .collect(); + let foreign = fixture.foreign(); + let untouched = fixture.snapshot(&root(&fixture.candidate)); + let budget = MAX_INTENT_ENTRIES - foreign.len(); + assert!(preflight_records(&fixture.candidate, budget).is_err()); + + let confirmed = Cell::new(0); + let outcome = compact_with( + &fixture.candidate, + Request { + mode: Mode::Compact, + incarnation: INCARNATION, + boot: CURRENT, + run: RUN, + containers: &containers(RUN), + }, + |intent| { + assert_eq!(intent.graph.as_ref().unwrap().run, RUN); + confirmed.set(confirmed.get() + 1); + Ok(()) + }, + |_| Ok(()), + ) + .unwrap(); + assert_eq!( + outcome, + Compaction { + compacted: 3 * SERVICES, + completed: 0, + sealed_boots: 1, + } + ); + assert_eq!(confirmed.get(), 3 * SERVICES, "every record proven absent"); + + let remaining = fixture.snapshot(&root(&fixture.candidate)); + assert_eq!(remaining.len(), foreign.len()); + for (name, entry) in &remaining { + assert_eq!(untouched.get(name), Some(entry), "foreign record changed"); + } + assert!(preflight_records(&fixture.candidate, budget).is_ok()); + + let tombstone = load(&fixture.candidate, RUN, INCARNATION).unwrap().unwrap(); + assert_eq!(tombstone.sealed.len(), 1); + assert_eq!(tombstone.sealed[PRIOR].slots, 2 * SERVICES as u32); + assert_eq!(tombstone.retired.len(), SERVICES); + for intent in ¤t { + assert_eq!( + tombstone.retired[&intent.slot], + digest(&originals[&intent.slot]) + ); + } + for slot in originals.keys() { + assert!(reserved(&fixture.candidate, RUN, INCARNATION, slot).unwrap()); + } + assert!(fixture.slots(RUN).unwrap().is_empty()); + assert_eq!( + fixture.slots(OTHER_RUN).unwrap().len(), + 2, + "committed and pending" + ); + + // A fresh allocation in the current boot stays admissible. + let fresh = fixture.intent(Some(RUN), CURRENT, 0); + assert!(!reserved(&fixture.candidate, RUN, INCARNATION, &fresh.slot).unwrap()); + record_intent(&fixture.candidate, &fresh).unwrap(); + assert_eq!(fixture.slots(RUN).unwrap(), vec![fresh.slot.clone()]); + fs::remove_file(fixture.active(&fresh.slot)).unwrap(); + + let bytes = fs::read(fixture.tombstone_path(RUN)).unwrap(); + assert_eq!(fixture.compact_now(RUN).unwrap(), Compaction::default()); + assert_eq!( + fs::read(fixture.tombstone_path(RUN)).unwrap(), + bytes, + "re-run is idempotent" + ); +} + +#[test] +fn interruption_at_each_compaction_step_recovers_without_loss() { + let build = || { + let fixture = Fixture::new(); + let mut intents = fixture.generation(RUN, PRIOR); + intents.extend(fixture.generation(RUN, CURRENT)); + let originals: BTreeMap> = intents + .iter() + .map(|intent| { + ( + intent.slot.clone(), + fs::read(fixture.active(&intent.slot)).unwrap(), + ) + }) + .collect(); + let foreign = fixture.foreign(); + let untouched = fixture.snapshot(&root(&fixture.candidate)); + (fixture, originals, foreign, untouched) + }; + let (reference, _, _, _) = build(); + let steps = Cell::new(0); + reference + .compact(Mode::Compact, INCARNATION, CURRENT, RUN, |_| { + steps.set(steps.get() + 1); + Ok(()) + }) + .unwrap(); + // Recorded, one unlink per record, directory sync, prior-boot seal. + assert_eq!(steps.get(), 2 * SERVICES + 3); + let expected = load(&reference.candidate, RUN, INCARNATION) + .unwrap() + .unwrap(); + + for target in 0..steps.get() { + let (fixture, originals, foreign, untouched) = build(); + assert!( + fixture + .compact(Mode::Compact, INCARNATION, CURRENT, RUN, stop_at(target)) + .is_err() + ); + // No loss: every record is still active with its exact bytes, or durably retained. + let tombstone = load(&fixture.candidate, RUN, INCARNATION).unwrap().unwrap(); + for (slot, bytes) in &originals { + let active = fixture.active(slot); + if active.exists() { + assert_eq!(&fs::read(&active).unwrap(), bytes); + assert_eq!(tombstone.retired.get(slot), Some(&digest(bytes))); + } else { + assert!( + reserves(Some(&tombstone), slot), + "step {target} lost {slot}" + ); + } + } + // Read-only inventory treats durable residue as retired, never as live. + assert!(fixture.slots(RUN).unwrap().is_empty(), "step {target}"); + #[cfg(target_os = "macos")] + assert!( + graph_inventory( + &fixture.candidate, + INCARNATION, + RUN, + &containers(RUN), + &fixture + .candidate + .state_root + .join("graph-environment-archive"), + ) + .unwrap() + .is_empty() + ); + let now = fixture.snapshot(&root(&fixture.candidate)); + for path in &foreign { + let name = path.file_name().unwrap().to_str().unwrap(); + assert_eq!(untouched.get(name), now.get(name), "step {target} foreign"); + } + // Alternate recovery entry points: archive's resume-only pass, or the next cleanup. + if target % 2 == 0 { + let resumed = fixture + .compact(Mode::Resume, INCARNATION, CURRENT, RUN, |_| Ok(())) + .unwrap(); + assert_eq!(resumed.compacted, 0); + assert_eq!(resumed.sealed_boots, 0); + } + fixture.compact_now(RUN).unwrap(); + assert_eq!( + load(&fixture.candidate, RUN, INCARNATION).unwrap().unwrap(), + expected, + "step {target} converges" + ); + for slot in originals.keys() { + assert!(!fixture.active(slot).exists()); + } + assert_eq!(fixture.compact_now(RUN).unwrap(), Compaction::default()); + } +} + +#[test] +fn uncommitted_tombstone_write_has_no_authority_and_is_replaced() { + let fixture = Fixture::new(); + let intents = fixture.generation(RUN, CURRENT); + let parent = directory(&fixture.candidate); + state::private_directory(&parent).unwrap(); + let torn = fixture.tombstone_path(RUN).with_extension("pending"); + write_raw(&torn, b"{\"version\":1,\"run\":"); + // Before commit point 1, every record is still live inventory. + assert_eq!(fixture.slots(RUN).unwrap().len(), SERVICES); + assert!(!reserved(&fixture.candidate, RUN, INCARNATION, &intents[0].slot).unwrap()); + assert_eq!(fixture.compact_now(RUN).unwrap().compacted, SERVICES); + assert!(!torn.exists()); + assert!(fixture.slots(RUN).unwrap().is_empty()); + // A temporary that is not a private regular file refuses rather than being removed. + std::os::unix::fs::symlink("elsewhere", &torn).unwrap(); + fixture.generation(RUN, CURRENT); + let before = fixture.state(); + assert!(fixture.compact_now(RUN).is_err()); + assert_eq!(fixture.state(), before); +} + +#[test] +fn wrong_owner_stale_boot_live_misbound_and_uncertain_records_refuse_without_effects() { + let fixture = Fixture::new(); + let intents = fixture.generation(RUN, CURRENT); + fixture.foreign(); + let before = fixture.state(); + let unchanged = |label: &str| assert_eq!(fixture.state(), before, "{label}"); + + assert!( + fixture + .compact(Mode::Compact, FOREIGN_INCARNATION, CURRENT, RUN, |_| Ok(())) + .is_err() + ); + unchanged("wrong owner"); + assert!(!directory(&fixture.candidate).exists()); + + let live = intents[3].slot.clone(); + assert!( + compact_with( + &fixture.candidate, + Request { + mode: Mode::Compact, + incarnation: INCARNATION, + boot: CURRENT, + run: RUN, + containers: &containers(RUN), + }, + |intent| if intent.slot == live { + Err(error()) + } else { + Ok(()) + }, + |_| Ok(()), + ) + .is_err() + ); + unchanged("live slot"); + + let mut misbound = containers(RUN); + misbound.insert("svc5".into(), format!("hkg-{RUN}-container-6")); + assert!( + compact_with( + &fixture.candidate, + Request { + mode: Mode::Compact, + incarnation: INCARNATION, + boot: CURRENT, + run: RUN, + containers: &misbound, + }, + |_| Ok(()), + |_| Ok(()), + ) + .is_err() + ); + unchanged("misbound container"); + + let uncertain = fixture.intent(Some(RUN), CURRENT, 7); + let pending = fixture.active(&uncertain.slot).with_extension("pending"); + write_raw(&pending, &serde_json::to_vec(&uncertain).unwrap()); + let with_pending = fixture.state(); + assert!(fixture.compact_now(RUN).is_err()); + assert_eq!(fixture.state(), with_pending, "uncertain record"); + // Resume never selects, so it also leaves the uncertain record alone. + assert_eq!( + fixture + .compact(Mode::Resume, INCARNATION, CURRENT, RUN, |_| Ok(())) + .unwrap(), + Compaction::default() + ); + assert_eq!(fixture.state(), with_pending); + fs::remove_file(&pending).unwrap(); + + // Seal CURRENT by compacting from a later boot; the stale boot may not compact again. + let later = "33333333-3333-4333-8333-333333333333"; + assert_eq!( + fixture + .compact(Mode::Compact, INCARNATION, later, RUN, |_| Ok(())) + .unwrap() + .sealed_boots, + 1 + ); + let sealed = fixture.state(); + assert!( + fixture + .compact(Mode::Compact, INCARNATION, CURRENT, RUN, |_| Ok(())) + .is_err() + ); + assert_eq!(fixture.state(), sealed, "stale boot"); +} + +#[test] +fn replayed_changed_and_copied_state_never_resurrects_a_retired_lease() { + let fixture = Fixture::new(); + let prior = fixture.generation(RUN, PRIOR); + let current = fixture.generation(RUN, CURRENT); + let replay = |intent: &Intent| { + let path = fixture.active(&intent.slot); + let bytes = serde_json::to_vec_pretty(intent).unwrap(); + write_raw(&path, &bytes); + (path, bytes) + }; + let prior_bytes = fs::read(fixture.active(&prior[0].slot)).unwrap(); + let current_bytes = fs::read(fixture.active(¤t[0].slot)).unwrap(); + fixture.compact_now(RUN).unwrap(); + + // Exact current-boot replay: retired history, never delivery or a new allocation. + let (path, bytes) = replay(¤t[0]); + assert_eq!(bytes, current_bytes, "exact copy of the retired record"); + assert!(fixture.slots(RUN).unwrap().is_empty()); + assert!(record_intent(&fixture.candidate, ¤t[0]).is_err()); + let exec = |slot: &str| { + active_exec_slot_for( + &fixture.candidate, + INCARNATION, + CURRENT, + RUN, + "svc0", + &format!("hkg-{RUN}-container-0"), + Some(slot), + ) + }; + assert!(exec(¤t[0].slot).is_err()); + assert_eq!(fixture.compact_now(RUN).unwrap().completed, 1); + assert!(!path.exists()); + // With no residue left, the retired identity still cannot be recorded again. + assert!(record_intent(&fixture.candidate, ¤t[2]).is_err()); + assert!(!fixture.active(¤t[2].slot).exists()); + + // Stale-boot replay of a sealed boot refuses everywhere and is retained untouched. + let (path, _) = replay(&prior[0]); + assert_eq!(fs::read(&path).unwrap(), prior_bytes); + let before = fixture.state(); + assert!(fixture.slots(RUN).is_err()); + assert!(fixture.compact_now(RUN).is_err()); + assert!(record_intent(&fixture.candidate, &prior[0]).is_err()); + #[cfg(target_os = "macos")] + assert!( + graph_inventory( + &fixture.candidate, + INCARNATION, + RUN, + &containers(RUN), + &fixture + .candidate + .state_root + .join("graph-environment-archive"), + ) + .is_err() + ); + assert_eq!(fixture.state(), before); + fs::remove_file(&path).unwrap(); + + // A changed record under a retired slot identity is refused, never removed, and + // blocks compaction of an otherwise eligible record before any tombstone write. + let mut changed = current[1].clone(); + changed.uid = 1000; + changed.gid = 1000; + let (path, bytes) = replay(&changed); + let eligible = fixture.intent(Some(RUN), CURRENT, 4); + fixture.commit(&eligible); + let before = fixture.state(); + assert!(fixture.slots(RUN).is_err()); + assert!(fixture.compact_now(RUN).is_err()); + assert!(exec(&changed.slot).is_err()); + assert_eq!(fixture.state(), before); + assert_eq!(fs::read(&path).unwrap(), bytes); + fs::remove_file(&path).unwrap(); + assert_eq!(fixture.compact_now(RUN).unwrap().compacted, 1); + + // Copied stores: another run's name, another incarnation, or a linked copy refuse. + let tombstone = fixture.tombstone_path(RUN); + let copied = fixture.tombstone_path(OTHER_RUN); + fs::copy(&tombstone, &copied).unwrap(); + assert!(load(&fixture.candidate, OTHER_RUN, INCARNATION).is_err()); + assert!(fixture.slots(OTHER_RUN).is_err()); + fs::remove_file(&copied).unwrap(); + assert!(load(&fixture.candidate, RUN, FOREIGN_INCARNATION).is_err()); + assert!(graph_slots_for(&fixture.candidate, FOREIGN_INCARNATION, RUN).is_err()); + fs::hard_link(&tombstone, &copied).unwrap(); + assert!(load(&fixture.candidate, RUN, INCARNATION).is_err()); + assert!(fixture.compact_now(RUN).is_err()); + fs::remove_file(&copied).unwrap(); + assert!( + load(&fixture.candidate, RUN, INCARNATION) + .unwrap() + .is_some() + ); +} + +#[test] +fn malformed_tombstones_refuse() { + let fixture = Fixture::new(); + fixture.generation(RUN, PRIOR); + fixture.compact_now(RUN).unwrap(); + let path = fixture.tombstone_path(RUN); + let original: serde_json::Value = state::read(&path).unwrap(); + let mut cases = Vec::new(); + let mut overlap = original.clone(); + overlap["retired"][format!("hack-env-lease-{PRIOR}-{}", "f".repeat(32))] = + serde_json::Value::String("0".repeat(64)); + cases.push(overlap); + let mut empty_seal = original.clone(); + empty_seal["sealed"][PRIOR]["slots"] = 0.into(); + cases.push(empty_seal); + let mut unknown = original.clone(); + unknown["values"] = serde_json::json!({"TOKEN": "x"}); + cases.push(unknown); + let mut version = original; + version["version"] = 2.into(); + cases.push(version); + for case in cases { + fs::remove_file(&path).unwrap(); + state::write(&path, &case).unwrap(); + assert!(load(&fixture.candidate, RUN, INCARNATION).is_err()); + assert!(fixture.compact_now(RUN).is_err()); + } +} + +#[test] +fn record_changed_after_tombstone_commit_is_retained_not_unlinked() { + let fixture = Fixture::new(); + let intents = fixture.generation(RUN, CURRENT); + let target = fixture.active(&intents[0].slot); + let mut changed = intents[0].clone(); + changed.uid = 1000; + let replacement = serde_json::to_vec_pretty(&changed).unwrap(); + // An interruption after commit point 1 during which the record is replaced. + let outcome = fixture.compact(Mode::Compact, INCARNATION, CURRENT, RUN, |step| { + if step == Step::Recorded { + fs::remove_file(&target).unwrap(); + write_raw(&target, &replacement); + } + Ok(()) + }); + assert!(outcome.is_err()); + assert_eq!(fs::read(&target).unwrap(), replacement); + assert!(fixture.slots(RUN).is_err()); + assert!(fixture.compact_now(RUN).is_err()); + assert_eq!(fs::read(&target).unwrap(), replacement); +} + +#[test] +fn compacted_residue_and_replay_are_never_retired_as_live() { + let fixture = Fixture::new(); + let intents = fixture.generation(RUN, CURRENT); + let slot = intents[0].slot.clone(); + // Before compaction the record is a live allocation that retirement may act on. + assert!( + live_intent(&fixture.candidate, INCARNATION, &slot, None).unwrap() == intents[0], + "live record is retirable" + ); + // Commit point 1 only: every record is exact residue of a durable tombstone. + assert!( + fixture + .compact(Mode::Compact, INCARNATION, CURRENT, RUN, stop_at(0)) + .is_err() + ); + assert!(fixture.active(&slot).exists()); + let before = fixture.state(); + assert!(live_intent(&fixture.candidate, INCARNATION, &slot, None).is_err()); + assert_eq!(fixture.state(), before, "refusal is read-only"); + // After completion, an exact replay of the retired record is refused the same way. + fixture.compact_now(RUN).unwrap(); + write_raw( + &fixture.active(&slot), + &serde_json::to_vec_pretty(&intents[0]).unwrap(), + ); + assert!(live_intent(&fixture.candidate, INCARNATION, &slot, None).is_err()); +} + +#[cfg(target_os = "macos")] +#[test] +fn archived_copy_of_a_compacted_record_is_refused_by_inventory() { + let fixture = Fixture::new(); + let intents = fixture.generation(RUN, CURRENT); + fixture.compact_now(RUN).unwrap(); + let archive = fixture + .candidate + .state_root + .join("graph-environment-archive"); + let inventory = || { + graph_inventory( + &fixture.candidate, + INCARNATION, + RUN, + &containers(RUN), + &archive, + ) + }; + state::private_directory(&archive).unwrap(); + assert!(inventory().unwrap().is_empty()); + // The archived directory has no residue rule: a compacted identity there is a copy. + state::write( + &archive.join(format!("{}.json", intents[0].slot)), + &intents[0], + ) + .unwrap(); + assert!(inventory().is_err()); + fs::remove_file(archive.join(format!("{}.json", intents[0].slot))).unwrap(); + // A fresh, unreserved archived record of the run is still ordinary inventory. + let fresh = fixture.intent(Some(RUN), CURRENT, 0); + state::write(&archive.join(format!("{}.json", fresh.slot)), &fresh).unwrap(); + assert_eq!(inventory().unwrap().intents.len(), 1); +} + +#[test] +fn residue_with_an_uncertain_pending_record_refuses() { + let fixture = Fixture::new(); + let intents = fixture.generation(RUN, CURRENT); + assert!( + fixture + .compact(Mode::Compact, INCARNATION, CURRENT, RUN, stop_at(0)) + .is_err() + ); + let tombstone = load(&fixture.candidate, RUN, INCARNATION).unwrap(); + let slot = &intents[0].slot; + assert!(residue(&fixture.candidate, tombstone.as_ref(), slot).unwrap()); + let pending = fixture.active(slot).with_extension("pending"); + write_raw(&pending, &serde_json::to_vec(&intents[0]).unwrap()); + assert!(residue(&fixture.candidate, tombstone.as_ref(), slot).is_err()); + fs::remove_file(&pending).unwrap(); + assert!(residue(&fixture.candidate, tombstone.as_ref(), slot).unwrap()); +} + +#[test] +fn retired_slot_cap_refuses_rather_than_discards() { + let filled = |entries: usize| { + let fixture = Fixture::new(); + let mut tombstone = Tombstone::new(RUN, INCARNATION); + for index in 0..entries { + tombstone.retired.insert( + format!( + "hack-env-lease-{CURRENT}-{:032x}", + u32::MAX as usize - index + ), + "e".repeat(64), + ); + } + commit( + &directory(&fixture.candidate), + &fixture.tombstone_path(RUN), + &tombstone, + ) + .unwrap(); + fixture.generation(RUN, CURRENT); + fixture + }; + // Exactly at the cap the compaction is admitted. + let fixture = filled(MAX_RETIRED_SLOTS - SERVICES); + assert_eq!(fixture.compact_now(RUN).unwrap().compacted, SERVICES); + let tombstone = load(&fixture.candidate, RUN, INCARNATION).unwrap().unwrap(); + assert_eq!(tombstone.retired.len(), MAX_RETIRED_SLOTS); + // One past the cap refuses before any write: nothing is discarded or unlinked. + let fixture = filled(MAX_RETIRED_SLOTS - SERVICES + 1); + let before = fixture.state(); + let refused = fixture.compact_now(RUN); + assert!(refused.is_err()); + assert_eq!(fixture.state(), before); + assert_eq!( + fixture.slots(RUN).unwrap().len(), + SERVICES, + "still live inventory" + ); + // After a committed cleanup that refusal is retained evidence, not a cleanup failure. + assert_eq!(Settled::from(refused), Settled::Retained); +} + +/// HACK-1193 state-machine proof: more than 256 start/cleanup/restore generations of a +/// 14-service graph across several guest boots. Every generation is admitted through the +/// ordinary record path; active records and per-generation scans stay at one generation; +/// retained bytes stay bounded. Guest and container absence is injected; this does not +/// measure native cleanup time or VM disk allocation. +#[test] +fn repeated_generations_beyond_256_stay_bounded_by_active_allocations() { + const GENERATIONS: usize = 260; + const PER_BOOT: usize = 32; + let fixture = Fixture::new(); + let foreign = fixture.foreign(); + let untouched = fixture.snapshot(&root(&fixture.candidate)); + let boot = + |generation: usize| format!("{:08x}-2222-4222-8222-222222222222", generation / PER_BOOT); + let mut peak_tombstone = 0; + for generation in 0..GENERATIONS { + let boot = boot(generation); + let intents: Vec = (0..SERVICES) + .map(|service| fixture.intent(Some(RUN), &boot, service)) + .collect(); + // The whole batch is admitted; one record goes through the full durable record path + // (budget plus standalone and tombstone reservations). The others are written without + // per-file fsync only to keep this long test fast; their bytes are identical. + preflight_records(&fixture.candidate, SERVICES).unwrap(); + for (index, intent) in intents.iter().enumerate() { + if index == 0 { + record_intent(&fixture.candidate, intent).unwrap(); + } else { + assert!(!reserved(&fixture.candidate, RUN, INCARNATION, &intent.slot).unwrap()); + write_raw( + &fixture.active(&intent.slot), + &serde_json::to_vec_pretty(intent).unwrap(), + ); + } + } + // Cleanup and the next restore see only the active generation, never history. + assert_eq!( + fixture.slots(RUN).unwrap().len(), + SERVICES, + "generation {generation}" + ); + let outcome = fixture + .compact(Mode::Compact, INCARNATION, &boot, RUN, |_| Ok(())) + .unwrap(); + assert_eq!(outcome.compacted, SERVICES); + assert_eq!( + outcome.sealed_boots, + usize::from(generation % PER_BOOT == 0 && generation > 0) + ); + assert!(fixture.slots(RUN).unwrap().is_empty()); + assert_eq!( + fs::read_dir(root(&fixture.candidate)).unwrap().count(), + foreign.len(), + "generation {generation}" + ); + let metadata = fs::metadata(fixture.tombstone_path(RUN)).unwrap(); + peak_tombstone = peak_tombstone.max(metadata.blocks() * 512); + } + let tombstone = load(&fixture.candidate, RUN, INCARNATION).unwrap().unwrap(); + let boots = GENERATIONS.div_ceil(PER_BOOT); + assert_eq!(tombstone.sealed.len(), boots - 1); + assert!( + tombstone + .sealed + .values() + .all(|sealed| sealed.slots as usize == PER_BOOT * SERVICES) + ); + assert_eq!( + tombstone.retired.len(), + (GENERATIONS - (boots - 1) * PER_BOOT) * SERVICES + ); + // Retained bytes follow the current boot's generations, not total history: at most one + // boot of per-slot entries plus one summary per sealed boot. + assert!( + peak_tombstone <= 256 * 1024, + "allocated {peak_tombstone} bytes" + ); + assert!(preflight_records(&fixture.candidate, MAX_INTENT_ENTRIES - foreign.len()).is_ok()); + let now = fixture.snapshot(&root(&fixture.candidate)); + assert_eq!(now, untouched, "foreign records unchanged"); + eprintln!( + "generations={GENERATIONS} services={SERVICES} boots={boots} peak_tombstone_allocated_bytes={peak_tombstone}" + ); +} diff --git a/packages/runtime-core/src/provider/environment_recovery/retention.rs b/packages/runtime-core/src/provider/environment_recovery/retention.rs index 6812d07f1..5d3b2272c 100644 --- a/packages/runtime-core/src/provider/environment_recovery/retention.rs +++ b/packages/runtime-core/src/provider/environment_recovery/retention.rs @@ -39,7 +39,7 @@ fn present(path: &Path) -> Result { Err(e) => Err(state::io(e)), } } -fn bytes(path: &Path) -> Result, CandidateError> { +pub(super) fn bytes(path: &Path) -> Result, CandidateError> { let mut file = fs::OpenOptions::new() .read(true) .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK) diff --git a/packages/runtime-core/src/provider/graph/environment.rs b/packages/runtime-core/src/provider/graph/environment.rs index d965f8b91..e8c52d904 100644 --- a/packages/runtime-core/src/provider/graph/environment.rs +++ b/packages/runtime-core/src/provider/graph/environment.rs @@ -134,6 +134,33 @@ pub(super) fn verify_cleanup( environment_recovery::verify_graph_retired(engine.guest(), expected) } +/// Service-to-container bindings an environment-attached receipt's records must match; `None` +/// when the graph never recorded an environment allocation. +pub(super) fn bound_containers(receipt: &Receipt) -> Option> { + receipt.environment_attached.then(|| { + receipt + .resources + .values() + .filter(|r| r.kind == Kind::Container) + .map(|r| (r.key.clone(), r.name.clone())) + .collect() + }) +} + +/// Compact `run`'s confirmed-retired allocation records. The caller holds the engine's mutation +/// lease, has already committed a cleanup or retirement, and no retained effect identity still +/// binds this run's environment inventory (see `environment_recovery::compaction`). Every +/// record must match `containers` with that container and its guest slot absent. A refusal is +/// retained evidence, never a failure of the caller's committed cleanup. +pub(super) fn compact_history( + candidate: &Candidate, + engine: &Engine<'_>, + run: &str, + containers: &BTreeMap, +) -> environment_recovery::Settled { + environment_recovery::compact_graph(candidate, engine.guest(), run, containers).into() +} + pub(super) fn require_replay_supported(receipt: &Receipt) -> Result<(), CandidateError> { if receipt.environment_attached { return Err(error( @@ -156,6 +183,9 @@ pub(super) fn archive_slots( if !receipt.environment_attached { return Ok(()); } + // Finish an interrupted compaction so exact residue is not stranded in the active + // directory after the graph leaves cleanup. No new record is compacted here. + environment_recovery::resume_graph(candidate, engine.guest(), &receipt.run)?; let containers = receipt .resources .values() diff --git a/packages/runtime-core/src/provider/graph/environment/tests.rs b/packages/runtime-core/src/provider/graph/environment/tests.rs index 616afbbcb..19ce3086b 100644 --- a/packages/runtime-core/src/provider/graph/environment/tests.rs +++ b/packages/runtime-core/src/provider/graph/environment/tests.rs @@ -27,6 +27,101 @@ fn bound_receipt_never_silently_loses_its_redelivery_gate() { ); } #[test] +fn history_bindings_cover_only_attached_graph_containers() { + let mut receipt: Receipt = serde_json::from_value(json!({"version":1,"run":"a".repeat(32),"owner":"b".repeat(32),"namespace":"c".repeat(64),"plan_id":"d".repeat(64),"phase":"stopped-data-retained","readiness":{},"resources":{ + "container:app":{"key":"app","kind":"container","name":"hkg-app","phase":"absent"}, + "container:db":{"key":"db","kind":"container","name":"hkg-db","phase":"absent"}, + "volume:data":{"key":"data","kind":"volume","name":"hkg-data","phase":"created"} + }})).unwrap(); + assert_eq!( + bound_containers(&receipt), + None, + "no environment, no history" + ); + receipt.environment_attached = true; + assert_eq!( + bound_containers(&receipt).unwrap(), + BTreeMap::from([ + ("app".to_owned(), "hkg-app".to_owned()), + ("db".to_owned(), "hkg-db".to_owned()), + ]) + ); +} + +/// Wiring guard. The trigger points need an owned VM and engine, so their effects are proven +/// natively, not here; this only fails if a trigger is removed or moved past the point where +/// the new generation allocates. The state machine itself is covered by compaction's tests. +#[test] +fn compaction_triggers_stay_wired_at_their_lease_points() { + fn body<'a>(source: &'a str, start: &str, end: &str) -> &'a str { + let from = source.find(start).expect("function start"); + let to = from + source[from..].find(end).expect("function end"); + &source[from..to] + } + let cleanup = body( + include_str!("../mod.rs"), + "pub fn cleanup(", + "fn cleanup_owned(", + ); + let committed = cleanup.find("cleanup_owned(").expect("cleanup effect"); + let compacted = cleanup + .find("environment::compact_history(") + .expect("ordinary cleanup compacts"); + assert!( + committed < compacted, + "compaction follows the committed cleanup" + ); + + let restore = body( + include_str!("../restore.rs"), + "fn restore_inputs(", + "fn retain_restore_failure(", + ); + let retired = restore + .find("environment_recovery::retire(") + .expect("prior slots re-retired"); + let captured = restore + .find("environment::bound_containers(&receipt)") + .expect("history bindings captured"); + let replaced = restore + .find("*resource = prepared.resources[key].clone()") + .expect("fresh resources"); + let prepared = restore + .find("driver.prepare(") + .expect("fresh owner prepare"); + let compacted = restore + .find("environment::compact_history(") + .expect("restore compacts"); + let allocated = restore + .find("session.create_resources(true)") + .expect("new generation allocates"); + assert!( + retired < captured && captured < replaced, + "bindings of the retired generation" + ); + assert!( + prepared < compacted && compacted < allocated, + "superseded, then allocate" + ); + + let relay = body( + include_str!("../host_relay.rs"), + "pub(super) fn cleanup_with_relay_expected(", + "pub fn resume_relay_cleanup(", + ); + assert!( + !relay.contains("compact_history("), + "acknowledged relay cleanup re-verifies its inventory; it must not compact" + ); + + let archive = body( + include_str!("../environment.rs"), + "pub(super) fn archive_slots(", + "environment_recovery::archive_graph(", + ); + assert!(archive.contains("environment_recovery::resume_graph(")); +} +#[test] #[ignore = "Manual owned development VM, pinned Bun image, explicit candidate and external watchdog"] fn container_attachment_records_ownership_before_create_and_omits_values_from_metadata() { let candidate = candidate(); diff --git a/packages/runtime-core/src/provider/graph/mod.rs b/packages/runtime-core/src/provider/graph/mod.rs index e277f8981..0ea71c9ce 100644 --- a/packages/runtime-core/src/provider/graph/mod.rs +++ b/packages/runtime-core/src/provider/graph/mod.rs @@ -1624,7 +1624,15 @@ pub fn cleanup( let engine = Engine::connect_cleanup(candidate)?; let (receipt, root) = load(candidate, &engine, run)?; cleanup_enrollment::ordinary_mutation(&root, &receipt)?; - cleanup_owned(candidate, &engine, receipt, &root, remove_data) + let receipt = cleanup_owned(candidate, &engine, receipt, &root, remove_data)?; + // The cleaned receipt is committed and ordinary cleanup binds no relay effect identity to + // its environment inventory, so history is compacted under the same lease. A refusal keeps + // the records for the next cleanup, restore or archive; it never reports this cleanup failed. + // Relay-enrolled graphs compact in restore instead (see `environment_recovery::compaction`). + if let Some(containers) = environment::bound_containers(&receipt) { + environment::compact_history(candidate, &engine, &receipt.run, &containers); + } + Ok(receipt) } // The caller retains the VM mutation lease through this entire effect and any diff --git a/packages/runtime-core/src/provider/graph/restore.rs b/packages/runtime-core/src/provider/graph/restore.rs index f60b9dfab..fcf4a5dad 100644 --- a/packages/runtime-core/src/provider/graph/restore.rs +++ b/packages/runtime-core/src/provider/graph/restore.rs @@ -478,6 +478,8 @@ fn restore_inputs( for slot in environment::cleanup_slots(candidate, &engine, &receipt)? { super::super::environment_recovery::retire(candidate, engine.guest(), &slot, None)?; } + // Bindings of the retired generations, captured before fresh resources replace them. + let retired_history = environment::bound_containers(&receipt); let launcher = if environments.is_empty() { None } else { @@ -528,6 +530,10 @@ fn restore_inputs( if let Some(selected) = &source_rebind { selected.reverify(&session.engine)?; } + // A redelivery restore committed its restoring receipt (relay marker dormant) above; a + // fresh owner's prepare commits its own enrollment. Only then is the acknowledged + // stopped receipt, whose cleanup effect binds the environment inventory, superseded. + let mut superseded = fresh.is_none(); if fresh.is_some() && let Some(driver) = session.startup.as_mut() { @@ -537,6 +543,18 @@ fn restore_inputs( &session.root, &mut session.configs, )?; + superseded = true; + } + // Every prior slot was re-retired under this lease and each prior container is absent: + // compact that history before the new generation allocates, so restore and the next + // cleanup scan only active allocations. A refusal retains the records. + if superseded && let Some(containers) = &retired_history { + environment::compact_history( + candidate, + &session.engine, + &session.receipt.run, + containers, + ); } session.create_resources(true)?; execution::run(&prepared.graph, &mut session, options.timeout)