From 38ea93b6548516e0f34d8d43bff599ecf11e9647 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Mon, 19 Oct 2026 11:59:42 -0400 Subject: [PATCH 1/2] feat(runtime-core): compact retired environment leases after confirmed cleanup Ordinary graph cleanup retired every bound environment slot but left each immutable intent in run/environment-leases, so admission, cleanup and restore scanned every historical generation of a repeatedly started graph. After `graph cleanup` commits, the same engine mutation lease now compacts the run's records. Each 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 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. A retired slot, or any slot of a sealed boot, is never recorded, delivered, retired as live or counted again. Pending, live, changed, misbound or foreign-incarnation records refuse the whole compaction before its first write; other runs' and standalone records are untouched. Exact residue of an interrupted compaction reads as retired and is completed by the next cleanup or by archive's resume-only pass; changed residue refuses. Recovery flows that carry an exact environment inventory keep using the fenced path without compaction. Refs HACK-1193 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/README.md | 11 + .../src/provider/environment_recovery.rs | 66 +- .../environment_recovery/compaction.rs | 455 ++++++++++++ .../environment_recovery/compaction/tests.rs | 668 ++++++++++++++++++ .../environment_recovery/retention.rs | 2 +- .../src/provider/graph/environment.rs | 24 + .../runtime-core/src/provider/graph/mod.rs | 6 +- 7 files changed, 1213 insertions(+), 19 deletions(-) create mode 100644 packages/runtime-core/src/provider/environment_recovery/compaction.rs create mode 100644 packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index eb308245d..3342e1d79 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -916,6 +916,17 @@ 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. +After ordinary `graph cleanup` commits, the same engine lease compacts that run's +records: each must match a receipt container that is absent and a guest slot that is +absent, or nothing is written. Their 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 (bounded at 16384). A +retired slot or any slot of a sealed boot is never recorded, delivered, retired as +live or counted again, so history no longer consumes the 4096 active-record budget. +Pending, live, changed, misbound or foreign-incarnation records refuse compaction, +and other runs' records are untouched. An interruption is resumed by the next +cleanup or archive; exact residue reads as retired, changed residue refuses. 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..a1e4cb831 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::{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,6 +594,12 @@ pub(super) fn retire( if retention::reserved(candidate, slot)? { return retention::verify_retired(candidate, guest, slot, lease); } + // A compacted graph slot is retained history; never treat it as a live allocation. + if let Some(binding) = read_mode(candidate, slot, guest.incarnation(), lease, false)?.graph { + if compaction::reserved(candidate, &binding.run, guest.incarnation(), slot)? { + return Err(error()); + } + } let intent = read(candidate, slot, guest.incarnation(), lease)?; retire_intent(guest, &intent) } 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..d3ac05792 --- /dev/null +++ b/packages/runtime-core/src/provider/environment_recovery/compaction.rs @@ -0,0 +1,455 @@ +//! 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. +//! +//! 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, +} + +/// `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..f4a9de77a --- /dev/null +++ b/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs @@ -0,0 +1,668 @@ +//! 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, 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); +} 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..bd470fa8f 100644 --- a/packages/runtime-core/src/provider/graph/environment.rs +++ b/packages/runtime-core/src/provider/graph/environment.rs @@ -134,6 +134,27 @@ pub(super) fn verify_cleanup( environment_recovery::verify_graph_retired(engine.guest(), expected) } +/// Compact this graph's confirmed-retired allocation records after ordinary cleanup. The caller +/// holds the engine's mutation lease and has committed the cleaned receipt; every record must +/// match a receipt container that is absent, and its guest slot must be absent. +pub(super) fn compact_retired( + candidate: &Candidate, + engine: &Engine<'_>, + receipt: &Receipt, +) -> Result<(), CandidateError> { + if !receipt.environment_attached { + return Ok(()); + } + let containers = receipt + .resources + .values() + .filter(|r| r.kind == Kind::Container) + .map(|r| (r.key.clone(), r.name.clone())) + .collect(); + environment_recovery::compact_graph(candidate, engine.guest(), &receipt.run, &containers)?; + Ok(()) +} + pub(super) fn require_replay_supported(receipt: &Receipt) -> Result<(), CandidateError> { if receipt.environment_attached { return Err(error( @@ -156,6 +177,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/mod.rs b/packages/runtime-core/src/provider/graph/mod.rs index e277f8981..f94177fab 100644 --- a/packages/runtime-core/src/provider/graph/mod.rs +++ b/packages/runtime-core/src/provider/graph/mod.rs @@ -1624,7 +1624,11 @@ 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)?; + // Only ordinary cleanup compacts. Recovery flows retain an exact environment inventory + // across their confirmation and use the fenced path without this effect. + environment::compact_retired(candidate, &engine, &receipt)?; + Ok(receipt) } // The caller retains the VM mutation lease through this entire effect and any From 907726ee4ce6c21b4e8ef386497b0b9eaecbb8f2 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Mon, 19 Oct 2026 13:32:02 -0400 Subject: [PATCH 2/2] fix(runtime-core): compact relay-owned graph lease history at restore Every environment-bound graph the CLI creates is relay-owned, so ordinary graph cleanup never ran compaction for it. Foreground redelivery and fresh-owner restore now compact the run's retired history under the engine lease. Compaction runs after every prior slot is re-retired and after the acknowledged stopped receipt is superseded on disk, before the new generation allocates. Enrolled relay cleanup still does not compact, because the acknowledged-publisher check re-verifies its inventory. A compaction refusal after a committed cleanup is now Settled::Retained instead of an error. The records stay active and the next cleanup, restore or archive retries. graph::cleanup no longer reports a cleanup that succeeded on disk as failed. New tests cover the retire guard (live_intent), archived-copy refusal, residue with a pending record, the 16384 retired-slot cap, and 260 generations of a 14-service graph across 9 boots. A source-order wiring guard covers the trigger points. The README now describes the trigger points, refusal semantics, current-boot growth and tombstone retention. Refs: HACK-1193 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/README.md | 38 +++- .../src/provider/environment_recovery.rs | 21 +- .../environment_recovery/compaction.rs | 29 +++ .../environment_recovery/compaction/tests.rs | 215 +++++++++++++++++- .../src/provider/graph/environment.rs | 40 ++-- .../src/provider/graph/environment/tests.rs | 95 ++++++++ .../runtime-core/src/provider/graph/mod.rs | 10 +- .../src/provider/graph/restore.rs | 18 ++ 8 files changed, 429 insertions(+), 37 deletions(-) diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index 3342e1d79..33772d04b 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -916,17 +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. -After ordinary `graph cleanup` commits, the same engine lease compacts that run's -records: each must match a receipt container that is absent and a guest slot that is -absent, or nothing is written. Their 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 (bounded at 16384). A -retired slot or any slot of a sealed boot is never recorded, delivered, retired as -live or counted again, so history no longer consumes the 4096 active-record budget. +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. An interruption is resumed by the next -cleanup or archive; exact residue reads as retired, changed residue refuses. +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 a1e4cb831..9d6d82295 100644 --- a/packages/runtime-core/src/provider/environment_recovery.rs +++ b/packages/runtime-core/src/provider/environment_recovery.rs @@ -3,7 +3,7 @@ mod compaction; mod retention; use super::{environment::EnvironmentLease, lifecycle::OwnedGuest, state}; use crate::{Candidate, CandidateError}; -pub(super) use compaction::{compact_graph, resume_graph}; +pub(super) use compaction::{Settled, compact_graph, resume_graph}; pub use retention::{RetiredExport, export_retired}; use serde::{Deserialize, Serialize}; use std::{fs, path::PathBuf}; @@ -594,14 +594,23 @@ pub(super) fn retire( if retention::reserved(candidate, slot)? { return retention::verify_retired(candidate, guest, slot, lease); } - // A compacted graph slot is retained history; never treat it as a live allocation. - if let Some(binding) = read_mode(candidate, slot, guest.incarnation(), lease, false)?.graph { - if compaction::reserved(candidate, &binding.run, guest.incarnation(), slot)? { + 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()); } } - let intent = read(candidate, slot, guest.incarnation(), lease)?; - retire_intent(guest, &intent) + read(candidate, slot, incarnation, lease) } fn retire_intent(guest: &OwnedGuest<'_>, intent: &Intent) -> Result<(), CandidateError> { let slot = &intent.slot; diff --git a/packages/runtime-core/src/provider/environment_recovery/compaction.rs b/packages/runtime-core/src/provider/environment_recovery/compaction.rs index d3ac05792..ba7310599 100644 --- a/packages/runtime-core/src/provider/environment_recovery/compaction.rs +++ b/packages/runtime-core/src/provider/environment_recovery/compaction.rs @@ -16,6 +16,17 @@ //! 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 @@ -70,6 +81,24 @@ pub(crate) struct Compaction { 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)] diff --git a/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs b/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs index f4a9de77a..fdc039aa1 100644 --- a/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs +++ b/packages/runtime-core/src/provider/environment_recovery/compaction/tests.rs @@ -1,6 +1,8 @@ //! 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, preflight_records, record_intent}; +use super::super::{ + active_exec_slot_for, graph_slots_for, live_intent, preflight_records, record_intent, +}; use super::*; use std::{ cell::Cell, @@ -666,3 +668,214 @@ fn record_changed_after_tombstone_commit_is_retained_not_unlinked() { 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/graph/environment.rs b/packages/runtime-core/src/provider/graph/environment.rs index bd470fa8f..e8c52d904 100644 --- a/packages/runtime-core/src/provider/graph/environment.rs +++ b/packages/runtime-core/src/provider/graph/environment.rs @@ -134,25 +134,31 @@ pub(super) fn verify_cleanup( environment_recovery::verify_graph_retired(engine.guest(), expected) } -/// Compact this graph's confirmed-retired allocation records after ordinary cleanup. The caller -/// holds the engine's mutation lease and has committed the cleaned receipt; every record must -/// match a receipt container that is absent, and its guest slot must be absent. -pub(super) fn compact_retired( +/// 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<'_>, - receipt: &Receipt, -) -> Result<(), CandidateError> { - if !receipt.environment_attached { - return Ok(()); - } - let containers = receipt - .resources - .values() - .filter(|r| r.kind == Kind::Container) - .map(|r| (r.key.clone(), r.name.clone())) - .collect(); - environment_recovery::compact_graph(candidate, engine.guest(), &receipt.run, &containers)?; - Ok(()) + 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> { 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 f94177fab..0ea71c9ce 100644 --- a/packages/runtime-core/src/provider/graph/mod.rs +++ b/packages/runtime-core/src/provider/graph/mod.rs @@ -1625,9 +1625,13 @@ pub fn cleanup( let (receipt, root) = load(candidate, &engine, run)?; cleanup_enrollment::ordinary_mutation(&root, &receipt)?; let receipt = cleanup_owned(candidate, &engine, receipt, &root, remove_data)?; - // Only ordinary cleanup compacts. Recovery flows retain an exact environment inventory - // across their confirmation and use the fenced path without this effect. - environment::compact_retired(candidate, &engine, &receipt)?; + // 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) } 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)