diff --git a/crates/engram-coordinator/src/api/hosts.rs b/crates/engram-coordinator/src/api/hosts.rs index 45f71c619..b33074bba 100644 --- a/crates/engram-coordinator/src/api/hosts.rs +++ b/crates/engram-coordinator/src/api/hosts.rs @@ -104,10 +104,7 @@ impl HostView { ) -> Self { let mut ready_image_digests = row.ready_images.clone(); ready_image_digests.sort(); - let allocatable_mib = row.utilization.allocatable_mib; - let reserved_mib = reserved.mem_mib.max(0) as u64; - let cpu_budget = engram_core::types::host::host_cpu_budget(row.total_vcpus).max(0) as u64; - let reserved_vcpus = reserved.vcpus.max(0) as u64; + let room = engram_core::types::host::HostHeadroom::of(&row, reserved); let failing_capabilities = row .capabilities .failing_capabilities() @@ -141,13 +138,13 @@ impl HostView { pool_data_total_mib: row.utilization.pool_data_total_mib, pool_metadata_free_mib: row.utilization.pool_metadata_free_mib, pool_metadata_total_mib: row.utilization.pool_metadata_total_mib, - allocatable_mib, - reserved_mib, - free_mib: allocatable_mib.saturating_sub(reserved_mib), + allocatable_mib: room.allocatable_mib.max(0) as u64, + reserved_mib: room.reserved.mem_mib.max(0) as u64, + free_mib: room.free_mib().max(0) as u64, total_vcpus: row.total_vcpus, - cpu_budget_vcpus: cpu_budget, - reserved_vcpus, - free_vcpus: cpu_budget.saturating_sub(reserved_vcpus), + cpu_budget_vcpus: room.cpu_budget.max(0) as u64, + reserved_vcpus: room.reserved.vcpus.max(0) as u64, + free_vcpus: room.free_vcpus().unwrap_or(0).max(0) as u64, last_heartbeat_at: row.last_heartbeat_at, failing_capabilities, fc_snapshot_version, diff --git a/crates/engram-coordinator/src/idle_evictor.rs b/crates/engram-coordinator/src/idle_evictor.rs index 14cab17e5..9e6c2617c 100644 --- a/crates/engram-coordinator/src/idle_evictor.rs +++ b/crates/engram-coordinator/src/idle_evictor.rs @@ -23,10 +23,14 @@ //! observe the resume op before routing. use engram_core::traits::SandboxBackend; +use engram_core::types::host::{HostHeadroom, HostRecord, ReservedBudget}; +use engram_core::types::session::QueuedSession; use engram_core::types::snapshot::SnapshotRecord; use engram_core::types::BindingDisposition; use engram_core::types::SessionState; -use engram_core::{SandboxId, SessionId}; +use std::collections::HashMap; + +use engram_core::{HostId, SandboxId, SessionId}; use crate::session_ops::OpCtx; use crate::state::{SessionEvent, SharedState}; @@ -140,66 +144,157 @@ async fn park_paused_bookkeeping( Ok(()) } -/// ADR 0074 rung 2: does the session's host have memory headroom to -/// keep a VM PAUSED (rung 2) rather than fully evicting it? Reads the -/// heartbeat-persisted `hosts.utilization`. Fails CLOSED (no headroom → -/// full eviction) on a telemetry gap: a paused VM frees no RAM, so -/// parking under UNKNOWN pressure is the dangerous direction (it could -/// leave the host overcommitted), the mirror of the idle detector's -/// fail-open-toward-eviction. +/// What the parking decisions read (ADR 0074 rung 2): the headroom of +/// each host that takes new sessions, and how many queued sessions fit on +/// none of them. /// -/// Headroom threshold reuses the detector's mem floor (default 15% free) -/// plus a margin, so a host that is not "under pressure" for eviction -/// purposes has room to hold a paused VM. -async fn host_has_memory_headroom(state: &SharedState, session_id: SessionId) -> bool { - // Resolve the host through PG, NOT the in-memory `host_registry`: this - // runs on whichever coord replica took the RPC / scanner tick, and a - // replica that never cached the sandbox→host bind would fail closed - // here — silently degrading every park into a full eviction on that - // pod (a coin flip in a 2-replica deployment). - let host_id = match state.services.meta.get_session(session_id).await { - Ok(s) => s.host_id, - Err(_) => return false, - }; - let Some(host_id) = host_id else { - return false; - }; - let hosts = match state.services.meta.list_active_hosts().await { - Ok(h) => h, - Err(_) => return false, - }; - let Some(host) = hosts.into_iter().find(|h| h.id == host_id) else { - return false; - }; - let u = &host.utilization; - if u.mem_total_mib == 0 { - return false; // unmeasured → fail closed (no parking) +/// A parked VM keeps its RAM and its reservation, so a park costs nothing +/// until a session needs that room. A park holds while every queued +/// session fits somewhere; it yields when one does not, or when its host +/// is reserved beyond what it measures. +struct ParkView { + /// Every active host's headroom, and whether placement ranks it. + rooms: HashMap, + /// Queued sessions that fit on no host placement ranks. + unplaced: usize, +} + +impl ParkView { + fn new( + hosts: &[HostRecord], + reserved: &HashMap, + queued: &[QueuedSession], + now: chrono::DateTime, + ) -> Self { + let ttl = crate::placement::placement_ttl(); + let rooms: HashMap = hosts + .iter() + .map(|h| { + let room = HostHeadroom::of(h, reserved.get(&h.id).copied().unwrap_or_default()); + let ranked = crate::placement::host_counts_as_capacity(h, now, ttl); + (h.id, (room, ranked)) + }) + .collect(); + let unplaced = queued + .iter() + .filter(|q| { + !rooms.values().any(|(room, ranked)| { + *ranked && room.fits(q.mem_budget_mib, i64::from(q.cpu_budget_vcpus)) + }) + }) + .count(); + Self { rooms, unplaced } } - // Prefer the RAM ledger's `allocatable_mib` (issue #540): it already - // nets out other parked-but-resident VMs' PSS, so parking decisions - // don't double-count a host that is already holding parked sandboxes - // (which is exactly how a naive `total − used` would over-park a host - // into pressure). Fall back to raw physical free on pre-ledger / - // non-Linux hosts where `allocatable_mib == 0`. - let free_mib = if u.allocatable_mib > 0 { - u.allocatable_mib - } else { - u.mem_total_mib.saturating_sub(u.mem_used_mib) + + async fn read(state: &SharedState) -> Result { + let meta = &state.services.meta; + Ok(Self::new( + &meta.list_active_hosts().await?, + &meta.per_host_reserved().await?, + &meta.list_queued_sessions_fifo().await?, + state.services.clock.now_utc(), + )) + } + + /// Whether a host is reserved beyond the RAM it measures. + fn over_reserved(&self, host: HostId) -> bool { + self.rooms + .get(&host) + .is_some_and(|(room, _)| room.measured() && room.free_mib() < 0) + } + + /// Whether a session that goes idle on `host` parks. It does not when + /// a queued session waits for room, when the host is over-reserved, + /// or when the host is unknown or has not measured its RAM: a paused + /// VM frees no RAM, so a park under unknown pressure is the dangerous + /// direction. + fn may_park(&self, host: HostId) -> bool { + self.unplaced == 0 + && self + .rooms + .get(&host) + .is_some_and(|(room, _)| room.measured() && room.free_mib() >= 0) + } + + /// The parks that yield this pass, with the reason. At most one park + /// yields per host, the oldest, and none on a host where an eviction + /// already runs: that eviction frees room, and the next pass reads + /// the result. An over-reserved host yields a park (`pressure`). Then + /// each unplaced queued session that no running eviction serves takes + /// one park (`demand`) from the hosts placement ranks, the host with + /// the most unreserved RAM first, because it is the nearest to a fit. + fn yields( + &self, + parked: &[engram_core::types::Session], + evicting_on: &[HostId], + ) -> HashMap { + let mut oldest: HashMap = HashMap::new(); + for s in parked { + let Some(host) = s.host_id else { continue }; + if s.park_rung != 2 || s.sandbox_id.is_none() || evicting_on.contains(&host) { + continue; + } + let slot = oldest.entry(host).or_insert(s); + if (s.parked_at, s.id) < (slot.parked_at, slot.id) { + *slot = s; + } + } + let mut out = HashMap::new(); + oldest.retain(|host, s| { + let pressure = self.over_reserved(*host); + if pressure { + out.insert(s.id, "pressure"); + } + !pressure + }); + let served = evicting_on + .iter() + .filter(|h| self.rooms.get(h).is_some_and(|(_, ranked)| *ranked)) + .count() + + out.len(); + let mut hosts: Vec<(i64, HostId)> = oldest + .keys() + .filter_map(|h| { + let (room, ranked) = self.rooms.get(h)?; + ranked.then(|| (room.free_mib(), *h)) + }) + .collect(); + hosts.sort_by(|a, b| b.cmp(a)); + for (_, host) in hosts.into_iter().take(self.unplaced.saturating_sub(served)) { + out.insert(oldest[&host].id, "demand"); + } + out + } +} + +/// Whether the session parks instead of evicting: [`ParkView::may_park`] +/// for its host. The host comes from PG, not the in-memory +/// `host_registry`: a replica that never cached the sandbox's host would +/// turn every park into a full eviction. +async fn session_may_park( + state: &SharedState, + session_id: SessionId, +) -> Result { + let Some(host) = state.services.meta.get_session(session_id).await?.host_id else { + return Ok(false); }; - let free_pct = free_mib.saturating_mul(100) / u.mem_total_mib; - // A comfortable margin above the eviction floor: only park when the - // host has real slack, so parked residency can't tip it into - // pressure. - free_pct as u8 >= park_headroom_floor_pct() + Ok(ParkView::read(state).await?.may_park(host)) } -/// Free-RAM percent above which a host will PARK (rung 2) instead of -/// evict. Above the eviction mem floor by design (env-tunable). -fn park_headroom_floor_pct() -> u8 { - std::env::var("ENGRAM_PARK_HEADROOM_FLOOR_PCT") - .ok() - .and_then(|s| s.parse::().ok()) - .unwrap_or(30) +/// The parks that yield this reaper pass ([`ParkView::yields`]). +async fn park_yields( + state: &SharedState, + parked: &[engram_core::types::Session], +) -> Result, engram_core::error::MetaError> { + let evicting_on: Vec = state + .services + .meta + .list_evicting_sessions() + .await? + .into_iter() + .filter_map(|(s, _)| s.host_id) + .collect(); + Ok(ParkView::read(state).await?.yields(parked, &evicting_on)) } /// ADR 0101 C: how long the eviction scanner trusts a COMPLETED evict @@ -400,12 +495,13 @@ pub(crate) async fn run_evict_pipeline( // long pause, and the hub can still advertise a stale handle // at un-park — a forward then lands in a socket with no reader. // The ADR 0108 A8 attach-signal row recall + heartbeat - // disagreement repair bound that damage. Under real memory - // pressure this branch is skipped and the full eviction below runs (rung 4). Only the + // disagreement repair bound that damage. When the session may + // not park (or that cannot be read) this branch is skipped and the + // full eviction below runs (rung 4). Only the // idle-evict path parks; drain/evac (Evacuating) always captures. // `allow_park == false` is the reaper's DESCENT path (already // parked, now forcing the full capture) — it must never re-park. - if allow_park && host_has_memory_headroom(state, session_id).await { + if allow_park && session_may_park(state, session_id).await.unwrap_or(false) { match state.services.host.pause(sandbox_id, ctx.fence()).await { Ok(()) => match park_paused_bookkeeping(ctx, session_id, entry_status).await { Ok(()) => { @@ -1039,9 +1135,18 @@ pub async fn scanner_run_once( // (the 8h-in-evicting weirdness, and half of the ADR 0077×0090 // livelock surface). let parked = state.services.meta.list_parked_sessions().await?; - for session in parked { - if let Err(e) = park_reaper_advance_one(state, session).await { - tracing::warn!(error = %e, "park reaper per-session advance failed"); + if !parked.is_empty() { + // A failed read yields no park: a descent is a full capture that + // cannot be undone, so the parks hold until a pass reads the fleet. + let yields = park_yields(state, &parked).await.unwrap_or_else(|e| { + tracing::warn!(error = %e, "park reaper: could not read the fleet; parks hold this pass"); + HashMap::new() + }); + for session in parked { + let yield_reason = yields.get(&session.id).copied(); + if let Err(e) = park_reaper_advance_one(state, session, yield_reason).await { + tracing::warn!(error = %e, "park reaper per-session advance failed"); + } } } @@ -1251,10 +1356,11 @@ fn park_dwell_cap() -> Option { /// reaper decides whether to keep holding or DESCEND to a full eviction. /// /// Descent triggers (rung 2, parked-paused): -/// - **dwell**: parked longer than [`park_dwell_cap`] — reclaim the RAM. -/// - **pressure**: the host lost memory headroom — the whole point of -/// parking was to use spare RAM; once it's scarce the parked VM must -/// yield. +/// - **demand** / **pressure**: `yield_reason`, this pass's answer from +/// [`ParkView::yields`] — a queued session has no room, or the host is +/// over-reserved. +/// - **hard_ttl**: parked longer than the idle detector's hard TTL. +/// - **dwell**: parked longer than [`park_dwell_cap`] (off by default). /// /// ADR 0079: descending ENQUEUES an evict op with `allow_park = false` /// (forcing the full snapshot+destroy regardless of headroom); the verb @@ -1268,6 +1374,7 @@ fn park_dwell_cap() -> Option { async fn park_reaper_advance_one( state: &SharedState, session: engram_core::types::Session, + yield_reason: Option<&'static str>, ) -> Result<(), Box> { let session_id = session.id; let Some(_sandbox_id) = session.sandbox_id else { @@ -1317,10 +1424,6 @@ async fn park_reaper_advance_one( return Ok(()); } - // ADR 0074 addendum: pressure is the primary (normally the only) - // trigger. A parked VM on a host with headroom KEEPS ITS RAM — the - // user's return is then an un-pause (ms) instead of a ~27s rebuild. - let has_headroom = host_has_memory_headroom(state, session_id).await; // The absolute ceiling. The idle DETECTOR's hard TTL can't reach a // parked session (it scans `Active` rows; a parked one is `Parked`), // so with the dwell clock retired the reaper owns the ceiling: an @@ -1335,17 +1438,13 @@ async fn park_reaper_advance_one( let hard_exceeded = parked_for.to_std().is_ok_and(|elapsed| elapsed >= hard_cap); // Operator escape hatch (off by default) — see `park_dwell_cap`. let dwell_exceeded = park_dwell_cap().is_some_and(|cap| parked_for >= cap); - let reason = if !has_headroom { - "pressure" - } else if hard_exceeded { - "hard_ttl" - } else if dwell_exceeded { - "dwell" - } else { - // Host has room and the ceiling is far off — HOLD THE PARK. This - // is the ladder's whole point: idle sessions keep their VM (and - // therefore their ms-fast ascent) until the RAM is actually needed. - return Ok(()); + let reason = match yield_reason { + Some(reason) => reason, + None if hard_exceeded => "hard_ttl", + None if dwell_exceeded => "dwell", + // Nothing needs the RAM and the ceiling is far off: the park + // holds, and the user's return is an un-pause. + None => return Ok(()), }; descend_parked_session(state, &session, reason).await?; @@ -3279,8 +3378,8 @@ mod tests { /// ADR 0074 rung 2: push a `HostRecord` into the mock's /// `list_active_hosts` set whose id matches the registry host that /// owns `sandbox_id`, with a memory utilization that either has or - /// lacks parking headroom. `free_mib` feeds `allocatable_mib` (the - /// ledger signal `host_has_memory_headroom` prefers). + /// lacks room. `free_mib` feeds `allocatable_mib`, which the park + /// gate reads net of the host's reservations. fn seed_host_with_free_ram(meta: &Arc, host_id: engram_core::HostId, free_mib: u64) { meta.hosts .lock() @@ -3628,7 +3727,7 @@ mod tests { ); } - /// ADR 0074's primary trigger: the host loses headroom → the parked + /// A host reserved beyond the RAM it measures → the parked /// VM yields its RAM via a full eviction (un-pause → capture → /// destroy → Idle), leaving no trace of the parking. #[tokio::test] @@ -3656,15 +3755,8 @@ mod tests { assert_eq!(op.state, OpState::Done); assert_eq!(meta.session.lock().park_rung, 2, "parked-paused"); - // The host now has none — the RAM the park was borrowing is needed. - // (Mutate the seeded row in place; `seed_host_with_free_ram` PUSHES, - // so re-seeding would leave the roomy record ahead of it.) - // MEASURED-but-tiny, not 0: `host_has_memory_headroom` treats a 0 - // allocatable as "pre-ledger host" and falls back to raw physical - // free, which the fixture reports as plentiful. - for h in meta.hosts.lock().iter_mut().filter(|h| h.id == host_id) { - h.utilization.allocatable_mib = 1; - } + // The host now measures less RAM than is reserved on it. + over_reserve(&meta, host_id, 70_000); wait_for_op_lane_free(&meta, session_id).await; scanner_run_once(&state).await.expect("tick"); @@ -3699,6 +3791,240 @@ mod tests { ); } + /// Parks a session on a roomy host and returns the state. + async fn parked_on_roomy_host( + sandbox_root: &std::path::Path, + ) -> (SharedState, Arc, engram_core::SessionId, HostId) { + let session_id = engram_core::SessionId::new(); + let (state, meta) = build_state_and_meta(evicting_session(session_id), sandbox_root); + let sandbox_id = state.services.host.create(process_spec()).await.unwrap(); + state + .services + .meta + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) + .await + .unwrap(); + let host_id = state + .host_registry + .host_of(sandbox_id) + .expect("sandbox routed"); + meta.session.lock().host_id = Some(host_id); + seed_host_with_free_ram(&meta, host_id, 60_000); + let op = drive_evict(&state, session_id, true, false).await; + assert_eq!(op.state, OpState::Done); + assert_eq!(meta.session.lock().park_rung, 2, "parked with room"); + (state, meta, session_id, host_id) + } + + fn over_reserve(meta: &Arc, host_id: HostId, mem_mib: i64) { + meta.reserved + .lock() + .insert(host_id, ReservedBudget { mem_mib, vcpus: 0 }); + } + + fn queued(mem_budget_mib: i64) -> QueuedSession { + QueuedSession { + session: evicting_session(engram_core::SessionId::new()), + origin: engram_core::types::session::QueueOrigin::Resume, + mem_budget_mib, + cpu_budget_vcpus: 1, + queued_at: chrono::Utc::now(), + } + } + + /// The production case: the parks on a host reserve its RAM and a + /// queued resume has no room. The park yields. + #[tokio::test] + async fn a_park_yields_to_a_queued_session_with_no_room() { + let sandbox_root = TempDir::new().unwrap(); + let (state, meta, session_id, host_id) = parked_on_roomy_host(sandbox_root.path()).await; + // 60 000 MiB allocatable, 53 248 reserved: 6 752 free. + over_reserve(&meta, host_id, 53_248); + meta.queued.lock().push(queued(8_192)); + + wait_for_op_lane_free(&meta, session_id).await; + scanner_run_once(&state).await.expect("tick"); + { + let m = meta.clone(); + wait_for("descended to Idle", move || { + m.session.lock().status == SessionState::Idle + }) + .await; + } + assert_eq!(meta.session.lock().park_rung, 0); + } + + /// A full host holds its parks while nothing waits for the RAM, and + /// while every queued session fits somewhere. + #[tokio::test] + async fn a_full_host_holds_its_parks_while_nothing_waits() { + let sandbox_root = TempDir::new().unwrap(); + let (state, meta, session_id, host_id) = parked_on_roomy_host(sandbox_root.path()).await; + over_reserve(&meta, host_id, 53_248); + meta.queued.lock().push(queued(4_096)); + + wait_for_op_lane_free(&meta, session_id).await; + scanner_run_once(&state).await.expect("tick"); + assert_eq!(meta.session.lock().status, SessionState::Parked); + assert_eq!(meta.session.lock().park_rung, 2, "the park holds"); + } + + /// A session that goes idle while a queued session has no room does + /// not park: its RAM is wanted now. + #[tokio::test] + async fn a_session_does_not_park_while_a_queued_session_has_no_room() { + let session_id = engram_core::SessionId::new(); + let sandbox_root = TempDir::new().unwrap(); + let (state, meta) = build_state_and_meta(evicting_session(session_id), sandbox_root.path()); + let sandbox_id = state.services.host.create(process_spec()).await.unwrap(); + state + .services + .meta + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) + .await + .unwrap(); + let host_id = state + .host_registry + .host_of(sandbox_id) + .expect("sandbox routed"); + meta.session.lock().host_id = Some(host_id); + seed_host_with_free_ram(&meta, host_id, 60_000); + meta.queued.lock().push(queued(64_000)); + + let op = drive_evict(&state, session_id, true, false).await; + assert_eq!(op.state, OpState::Done); + assert_eq!(meta.session.lock().park_rung, 0, "no park"); + assert_eq!(meta.session.lock().status, SessionState::Idle); + } + + fn park_view(rooms: &[(HostId, i64, i64)], unplaced: usize) -> ParkView { + ParkView { + rooms: rooms + .iter() + .map(|&(id, allocatable_mib, mem_mib)| { + let room = HostHeadroom { + allocatable_mib, + cpu_budget: 0, + reserved: ReservedBudget { mem_mib, vcpus: 0 }, + }; + (id, (room, true)) + }) + .collect(), + unplaced, + } + } + + fn parked_at(host: HostId, minutes_ago: i64) -> Session { + let mut s = evicting_session(engram_core::SessionId::new()); + s.status = SessionState::Parked; + s.host_id = Some(host); + s.sandbox_id = Some(engram_core::SandboxId::new()); + s.park_rung = 2; + s.parked_at = Some(chrono::Utc::now() - chrono::Duration::minutes(minutes_ago)); + s + } + + #[test] + fn one_unplaced_session_takes_the_oldest_park_of_the_roomiest_host() { + let (tight, roomy) = (HostId::new(), HostId::new()); + let view = park_view(&[(tight, 60_000, 58_000), (roomy, 60_000, 40_000)], 1); + let parks = [ + parked_at(tight, 90), + parked_at(roomy, 10), + parked_at(roomy, 50), + ]; + let yields = view.yields(&parks, &[]); + assert_eq!(yields, HashMap::from([(parks[2].id, "demand")])); + } + + #[test] + fn a_running_eviction_serves_one_unplaced_session_and_shields_its_host() { + let (a, b) = (HostId::new(), HostId::new()); + let parks = [parked_at(a, 30), parked_at(b, 20)]; + let rooms = [(a, 60_000, 50_000), (b, 60_000, 55_000)]; + assert!(park_view(&rooms, 1).yields(&parks, &[a]).is_empty()); + assert_eq!( + park_view(&rooms, 2).yields(&parks, &[a]), + HashMap::from([(parks[1].id, "demand")]) + ); + } + + #[test] + fn parks_hold_with_no_unplaced_session_and_an_over_reserved_host_yields_one() { + let (full, over) = (HostId::new(), HostId::new()); + let view = park_view(&[(full, 60_000, 60_000), (over, 60_000, 61_000)], 0); + let parks = [parked_at(full, 30), parked_at(over, 20), parked_at(over, 5)]; + assert_eq!( + view.yields(&parks, &[]), + HashMap::from([(parks[1].id, "pressure")]) + ); + assert!(view.may_park(full)); + assert!(!view.may_park(over)); + assert!(!park_view(&[(full, 0, 0)], 0).may_park(full), "unmeasured"); + assert!(!park_view(&[(full, 60_000, 0)], 1).may_park(full), "demand"); + } + + /// A host that has not measured its RAM has no headroom: the idle + /// eviction takes the full path, with no park. + #[tokio::test] + async fn an_unmeasured_host_does_not_park() { + let session_id = engram_core::SessionId::new(); + let sandbox_root = TempDir::new().unwrap(); + let session = evicting_session(session_id); + let (state, meta) = build_state_and_meta(session, sandbox_root.path()); + let sandbox_id = state.services.host.create(process_spec()).await.unwrap(); + state + .services + .meta + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) + .await + .unwrap(); + let host_id = state + .host_registry + .host_of(sandbox_id) + .expect("sandbox routed"); + meta.session.lock().host_id = Some(host_id); + seed_host_with_free_ram(&meta, host_id, 0); + + let op = drive_evict(&state, session_id, true, false).await; + assert_eq!(op.state, OpState::Done); + assert_eq!(meta.session.lock().park_rung, 0, "no park"); + assert_eq!(meta.session.lock().status, SessionState::Idle); + } + + /// A reaper tick that cannot read the reservations holds the park: a + /// failed read is not pressure. + #[tokio::test] + async fn a_failed_headroom_read_holds_the_park() { + let session_id = engram_core::SessionId::new(); + let sandbox_root = TempDir::new().unwrap(); + let session = evicting_session(session_id); + let (state, meta) = build_state_and_meta(session, sandbox_root.path()); + let sandbox_id = state.services.host.create(process_spec()).await.unwrap(); + state + .services + .meta + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) + .await + .unwrap(); + let host_id = state + .host_registry + .host_of(sandbox_id) + .expect("sandbox routed"); + meta.session.lock().host_id = Some(host_id); + seed_host_with_free_ram(&meta, host_id, 60_000); + + let op = drive_evict(&state, session_id, true, false).await; + assert_eq!(op.state, OpState::Done); + assert_eq!(meta.session.lock().park_rung, 2, "parked with room"); + + *meta.reserved_read_fails.lock() = true; + wait_for_op_lane_free(&meta, session_id).await; + scanner_run_once(&state).await.expect("tick"); + assert_eq!(meta.session.lock().park_rung, 2, "the park holds"); + assert_eq!(meta.session.lock().status, SessionState::Parked); + } + /// ADR 0101 C's third `Evicting` shape — the post-capture settle /// window (capture landed, VM DESTROYED, session honestly `Evicting` /// until a later heartbeat's advert settles it Idle) — must NOT be diff --git a/crates/engram-coordinator/src/placement.rs b/crates/engram-coordinator/src/placement.rs index a0ffdfec4..529ae7144 100644 --- a/crates/engram-coordinator/src/placement.rs +++ b/crates/engram-coordinator/src/placement.rs @@ -23,7 +23,7 @@ use std::time::Duration; use chrono::{DateTime, Utc}; use engram_core::traits::{HostClient, MetadataStore, SessionFence}; use engram_core::types::host::{ - CapStatus, HostRecord, HostStatus, ReservedBudget, HOST_POOL_METADATA_FLOOR_MIB, + CapStatus, HostHeadroom, HostRecord, HostStatus, ReservedBudget, HOST_POOL_METADATA_FLOOR_MIB, }; use engram_core::types::snapshot::SnapshotMetadata; use engram_core::{HostId, SandboxError, SandboxId}; @@ -558,25 +558,22 @@ pub fn pick_from( } let need_mib = ctx.memory_mib.unwrap_or(0) as i64; let need_vcpus = ctx.cpu_budget_vcpus.unwrap_or(0) as i64; + let headroom = |id: HostId| -> Option { + let h = hosts.iter().find(|h| h.id == id)?; + Some(HostHeadroom::of( + h, + reserved.get(&id).copied().unwrap_or_default(), + )) + }; // Free RAM for `id`: None ⇒ unmeasured (treated as "fits" softly). let free_mib_of = |id: HostId| -> Option { - let h = hosts.iter().find(|h| h.id == id)?; - let alloc = h.utilization.allocatable_mib as i64; - if alloc <= 0 { - return None; - } - Some(alloc - reserved.get(&id).map(|r| r.mem_mib).unwrap_or(0)) + headroom(id) + .filter(HostHeadroom::measured) + .map(|r| r.free_mib()) }; - // CPU fit for `id`: a host with budget 0 (unreported) doesn't gate. + // CPU fit for `id`: a host with no reported cores doesn't gate. let cpu_fits = |id: HostId| -> bool { - let Some(h) = hosts.iter().find(|h| h.id == id) else { - return false; - }; - let budget = engram_core::types::host::host_cpu_budget(h.total_vcpus); - if budget <= 0 { - return true; - } - budget - reserved.get(&id).map(|r| r.vcpus).unwrap_or(0) >= need_vcpus + headroom(id).is_some_and(|r| r.free_vcpus().is_none_or(|free| free >= need_vcpus)) }; // Tier 0 (ADR 0078): authoritative snapshot-host affinity. If the // host that holds this snapshot's chunks is alive, schedulable, and @@ -734,25 +731,12 @@ pub async fn placement_preview( let (hosts, reserved) = hosts_and_reserved(meta).await?; let ranked = rank_hosts(&hosts, ctx, now, placement_ttl()); Ok(ranked.hosts.iter().any(|id| { - let Some(h) = hosts.iter().find(|h| h.id == *id) else { - return false; - }; - let alloc = h.utilization.allocatable_mib as i64; - if alloc <= 0 { - return true; // unmeasured → soft fallback fits - } - let free_mib = alloc - reserved.get(id).map(|r| r.mem_mib).unwrap_or(0); - if free_mib < mem_mib { - return false; - } - let cpu_budget = engram_core::types::host::host_cpu_budget(h.total_vcpus); - if cpu_budget > 0 { - let free_vcpus = cpu_budget - reserved.get(id).map(|r| r.vcpus).unwrap_or(0); - if free_vcpus < cpu_vcpus { - return false; - } - } - true + hosts.iter().find(|h| h.id == *id).is_some_and(|h| { + let room = HostHeadroom::of(h, reserved.get(id).copied().unwrap_or_default()); + // The placement transaction takes an unmeasured host as its + // last resort without a CPU check. + !room.measured() || room.fits(mem_mib, cpu_vcpus) + }) })) } @@ -1198,12 +1182,11 @@ pub async fn fleet_snapshot( } if host_counts_as_capacity(h, now, ttl) { m.schedulable_hosts += 1; - m.total_mib += h.utilization.allocatable_mib; - let cpu_budget = engram_core::types::host::host_cpu_budget(h.total_vcpus); - m.total_vcpus += cpu_budget as u64; - let r = reserved.get(&h.id).copied().unwrap_or_default(); - m.free_vcpus += (cpu_budget - r.vcpus).max(0) as u64; - m.free_mib += (h.utilization.allocatable_mib as i64 - r.mem_mib).max(0) as u64; + let room = HostHeadroom::of(h, reserved.get(&h.id).copied().unwrap_or_default()); + m.total_mib += room.allocatable_mib.max(0) as u64; + m.total_vcpus += room.cpu_budget.max(0) as u64; + m.free_vcpus += room.free_vcpus().unwrap_or(0).max(0) as u64; + m.free_mib += room.free_mib().max(0) as u64; } } Ok(m) diff --git a/crates/engram-coordinator/src/state.rs b/crates/engram-coordinator/src/state.rs index e009ac098..159e8e74c 100644 --- a/crates/engram-coordinator/src/state.rs +++ b/crates/engram-coordinator/src/state.rs @@ -2375,6 +2375,14 @@ pub(crate) mod tests { /// `send_prompt_core`'s `harness_mode` validation is exercisable /// (the trait default returns `None`, which rejects every mode). pub(crate) harness: PlMutex>, + /// `per_host_reserved`'s answer: the reservations tests stage on + /// the seeded hosts (empty = nothing reserved). + pub(crate) reserved: + PlMutex>, + /// Makes `per_host_reserved` fail, as a database error does. + pub(crate) reserved_read_fails: PlMutex, + /// `list_queued_sessions_fifo`'s answer. + pub(crate) queued: PlMutex>, } impl MiniMeta { @@ -2432,6 +2440,9 @@ pub(crate) mod tests { ops: engram_core::types::session_op::InMemoryOpLog::default(), outbox: PlMutex::new(Vec::new()), harness: PlMutex::new(None), + reserved: PlMutex::new(Default::default()), + reserved_read_fails: PlMutex::new(false), + queued: PlMutex::new(Vec::new()), } } } @@ -2702,6 +2713,22 @@ pub(crate) mod tests { async fn list_active_hosts(&self) -> Result, MetaError> { Ok(self.hosts.lock().clone()) } + async fn list_queued_sessions_fifo( + &self, + ) -> Result, MetaError> { + Ok(self.queued.lock().clone()) + } + async fn per_host_reserved( + &self, + ) -> Result< + std::collections::HashMap, + MetaError, + > { + if *self.reserved_read_fails.lock() { + return Err(MetaError::Db("staged read failure".into())); + } + Ok(self.reserved.lock().clone()) + } async fn set_host_status(&self, _: HostId, _: HostStatus) -> Result<(), MetaError> { Ok(()) } diff --git a/crates/engram-core/src/traits/metadata.rs b/crates/engram-core/src/traits/metadata.rs index 90bea4085..438c5654f 100644 --- a/crates/engram-core/src/traits/metadata.rs +++ b/crates/engram-core/src/traits/metadata.rs @@ -691,8 +691,8 @@ pub trait MetadataStore: Send + Sync { /// ADR 0047/0048: per-host reserved budget (Σ `mem_budget_mib` AND /// Σ `cpu_budget_vcpus` over the memory-reserving session states) — /// the read-side twin of `reserve_and_persist_create`'s aggregate, for the - /// capacity-soft resume/evac picker and the fleet view. Default impl - /// (mocks): empty map (no reservations). + /// capacity-soft resume/evac picker, the fleet view and the parking + /// headroom gate. Default impl (mocks): empty map (no reservations). async fn per_host_reserved( &self, ) -> Result, MetaError> diff --git a/crates/engram-core/src/types/host.rs b/crates/engram-core/src/types/host.rs index c28c8c87b..fc11b5578 100644 --- a/crates/engram-core/src/types/host.rs +++ b/crates/engram-core/src/types/host.rs @@ -559,16 +559,72 @@ pub struct HostHeartbeat { pub lease_renew_until: Option>, } -/// Per-host reserved memory and CPU: Σ over the memory-reserving session -/// states of `mem_budget_mib` and `cpu_budget_vcpus`. The read-side twin -/// of `reserve_placement`'s in-transaction aggregate, for the resume/evac -/// picker and the fleet view. -#[derive(Clone, Copy, Debug, Default)] +/// Per-host reserved memory and CPU: Σ of `mem_budget_mib` and +/// `cpu_budget_vcpus` over the sessions in a memory-reserving state (a +/// parked VM included), the capturing VMs and the reserving teleport legs +/// on the host. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct ReservedBudget { pub mem_mib: i64, pub vcpus: i64, } +/// One host's resource accounting: what the host can hold, what the +/// coordinator has reserved on it, and what is left. Every decision that +/// asks "is there room on this host" reads it: placement (create, resume, +/// queue, capture, teleport, drain guard), parking, and the fleet's +/// autoscaling signal. +/// +/// The host measures `allocatable_mib`: the RAM a reservation may use, +/// with every resident VM's memory added back, so a VM's RAM counts once, +/// in `reserved`. The coordinator owns `reserved`. Free is the +/// difference: a parked VM keeps its reservation until it descends, so +/// it is not free. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct HostHeadroom { + /// Host-measured `allocatable_mib`; `<= 0` means unmeasured (a host + /// that has not sampled its RAM yet). + pub allocatable_mib: i64, + /// `total_vcpus × overcommit` ([`host_cpu_budget`]); `0` means the + /// host has not reported its cores. + pub cpu_budget: i64, + pub reserved: ReservedBudget, +} + +impl HostHeadroom { + pub fn of(host: &HostRecord, reserved: ReservedBudget) -> Self { + Self { + allocatable_mib: host.utilization.allocatable_mib as i64, + cpu_budget: host_cpu_budget(host.total_vcpus), + reserved, + } + } + + /// Whether the host has measured its RAM. An unmeasured host does + /// not gate placement (a brand-new or dev host must still work) and + /// never parks a session. + pub fn measured(&self) -> bool { + self.allocatable_mib > 0 + } + + /// Unreserved RAM. Only meaningful when [`Self::measured`]. + pub fn free_mib(&self) -> i64 { + self.allocatable_mib - self.reserved.mem_mib + } + + /// Unreserved vCPU budget; `None` when the host reports no cores. + pub fn free_vcpus(&self) -> Option { + (self.cpu_budget > 0).then(|| self.cpu_budget - self.reserved.vcpus) + } + + /// Whether a reservation of `mem_mib` and `vcpus` fits now. An + /// unmeasured RAM figure or an unreported core count does not gate. + pub fn fits(&self, mem_mib: i64, vcpus: i64) -> bool { + (!self.measured() || self.free_mib() >= mem_mib) + && self.free_vcpus().is_none_or(|free| free >= vcpus) + } +} + /// ADR 0048: the CPU overcommit factor. The host CPU budget is /// `total_vcpus × factor` — FC guests idle heavily and RAM is the hard /// constraint, so CPU is deliberately oversubscribed to keep packing @@ -950,3 +1006,42 @@ mod retirement_tests { } } } + +#[cfg(test)] +mod headroom_tests { + use super::{HostHeadroom, ReservedBudget}; + + fn room(allocatable_mib: i64, cpu_budget: i64, mem_mib: i64, vcpus: i64) -> HostHeadroom { + HostHeadroom { + allocatable_mib, + cpu_budget, + reserved: ReservedBudget { mem_mib, vcpus }, + } + } + + #[test] + fn free_is_allocatable_less_reserved_in_both_dimensions() { + let r = room(32_768, 32, 28_000, 30); + assert_eq!(r.free_mib(), 4_768); + assert_eq!(r.free_vcpus(), Some(2)); + assert!(r.fits(4_096, 2)); + assert!(!r.fits(8_192, 2), "RAM binds"); + assert!(!r.fits(4_096, 4), "CPU binds"); + } + + #[test] + fn an_unmeasured_host_does_not_gate_placement() { + let r = room(0, 0, 0, 0); + assert!(!r.measured()); + assert!( + r.fits(65_536, 64), + "unmeasured RAM and unreported cores do not gate" + ); + assert_eq!(room(32_768, 0, 0, 0).free_vcpus(), None); + } + + #[test] + fn an_over_reserved_host_has_negative_free_ram() { + assert_eq!(room(60_000, 0, 70_000, 0).free_mib(), -10_000); + } +} diff --git a/crates/engram-dst-cosim/src/scheduler.rs b/crates/engram-dst-cosim/src/scheduler.rs index ce3cd7f8c..0cb527274 100644 --- a/crates/engram-dst-cosim/src/scheduler.rs +++ b/crates/engram-dst-cosim/src/scheduler.rs @@ -222,7 +222,9 @@ impl Cosim { // AND `session_ops::enqueue(Evict)` — but that enqueue is the same // detached spawn that starves the real capture I/O here, so rung 1 // applies the nomination transition directly and drives the Evict - // verb inline below (identical to what the detector enqueues). + // verb inline below. `allow_park` is false, as in a park's descent: + // the roomy simulated host would park, and the capture is what the + // callers exercise. let _ = self .world .meta @@ -236,7 +238,7 @@ impl Cosim { self.enqueue_and_drive( session_id, OpKind::Evict, - serde_json::json!({ "target": "idle", "allow_park": true, "nominated": true }), + serde_json::json!({ "target": "idle", "allow_park": false, "nominated": true }), &key, ) .await; diff --git a/crates/engram-host-agent/src/ram_ledger.rs b/crates/engram-host-agent/src/ram_ledger.rs index cada84f03..873ccd1bd 100644 --- a/crates/engram-host-agent/src/ram_ledger.rs +++ b/crates/engram-host-agent/src/ram_ledger.rs @@ -24,8 +24,12 @@ //! sandbox's PSS is added back into `allocatable_mib` iff its session //! holds a coordinator memory reservation. The host-side proxy for //! "holds a reservation" is the per-sandbox `parked` flag -//! (`engram-sandbox-firecracker`'s `LiveSandbox::parked`, always -//! `false` until epic-parking-ladder lands). Parked residents are +//! (`engram-sandbox-firecracker`'s `LiveSandbox::parked`). No backend +//! sets it: a parked session keeps its reservation (the coordinator +//! counts `Parked` as a reserving state), so its VM is a reserved, +//! running resident here. A backend that sets the flag must come with a +//! coordinator that stops reserving for that session, or the VM's RAM +//! is charged twice. Parked residents are //! charged at their full measured PSS — prod observed rss ≈ pss //! (12.39 GB ≈ 12.39 GB) for the running dev-brain sandbox on //! 2026-07-01, so no sharing discount is ever assumed; density math @@ -65,8 +69,8 @@ pub struct RamLedgerSnapshot { /// The only bucket ever added back into `allocatable_mib`. pub running_vm_pss_mib: u64, /// Σ smaps_rollup PSS over FC sandboxes flagged `parked` — RAM- - /// resident, reservation-free (epic-parking-ladder rungs 2-3). - /// Always 0 until a backend ever parks a sandbox. Never added back + /// resident, reservation-free. Always 0: no backend sets the flag + /// (see the module invariant). Never added back /// into `allocatable_mib` — see the module invariant above. This is /// the seam the ladder/queue-scanner later use to count /// *reclaimable-under-pressure* capacity; deliberately NOT folded diff --git a/crates/engram-host-agent/src/resource.rs b/crates/engram-host-agent/src/resource.rs index b8feffac8..27173b717 100644 --- a/crates/engram-host-agent/src/resource.rs +++ b/crates/engram-host-agent/src/resource.rs @@ -12,7 +12,7 @@ use std::sync::Arc; /// heartbeats report the actual machine size — without this, the /// coord scheduler sees `total_mib=0` on every host and rejects every /// session whose `vm_spec.memory.max_mib` is non-zero (every session -/// in practice), and the idle-evictor's `host_has_memory_headroom` +/// in practice), and the idle-evictor's park gate /// fails closed so rung-2 park never fires (ADR 0096 — macOS/VZ hosts /// reported 0 until this grew a Darwin arm). Returns 0 on failure — /// the caller logs and continues; sessions just won't be scheduled to diff --git a/crates/engram-host-agent/tests/blk_source.rs b/crates/engram-host-agent/tests/blk_source.rs index aa36de687..e7c6650d9 100644 --- a/crates/engram-host-agent/tests/blk_source.rs +++ b/crates/engram-host-agent/tests/blk_source.rs @@ -95,6 +95,16 @@ async fn read_off_runtime(device: &Arc, offset: u64, len: usize) -> Vec Vec { + let mut touched = source.touched(); + touched.retain(|chunk| *chunk != 0); + touched +} + #[tokio::test] #[ignore = "requires root and the ublk_drv module"] async fn a_source_serves_the_manifest_and_records_the_chunks_read_cold() { @@ -116,7 +126,7 @@ async fn a_source_serves_the_manifest_and_records_the_chunks_read_cold() { read_off_runtime(&device, at, 4096).await, image[at as usize..at as usize + 4096] ); - assert_eq!(source.touched(), vec![3]); + assert_eq!(guest_reads(&source), vec![3]); // The copy's stream: the whole device, region by region, ascending. for at in (0..CHUNKS * CHUNK).step_by(REGION as usize) { @@ -126,7 +136,11 @@ async fn a_source_serves_the_manifest_and_records_the_chunks_read_cold() { "region at {at}" ); } - assert_eq!(source.touched(), vec![3], "the stream is not a guest read"); + assert_eq!( + guest_reads(&source), + vec![3], + "the stream is not a guest read" + ); drop(device); source.stop().await.unwrap(); diff --git a/crates/engram-postgres/src/lib.rs b/crates/engram-postgres/src/lib.rs index 56e3d41c4..04279d928 100644 --- a/crates/engram-postgres/src/lib.rs +++ b/crates/engram-postgres/src/lib.rs @@ -8,7 +8,8 @@ //! ship a `.sqlx/` cache. use engram_core::types::host::{ - CordonOwner, HeartbeatAck, RetirementBlocker, RetirementGrant, RetirementStatus, + CordonOwner, HeartbeatAck, HostHeadroom, ReservedBudget, RetirementBlocker, RetirementGrant, + RetirementStatus, }; use engram_core::types::ids::TeleportId; use engram_core::types::teleport::*; @@ -512,15 +513,82 @@ fn catalog_harness_from_row(row: HarnessRow) -> engram_core::types::CatalogHarne const HARNESS_ROW_COLS: &str = "id, owner, name, oci_ref, manifest_digest, descriptor_toml, squashfs_sha256, squashfs_size_bytes, created_at"; -/// ADR 0048: per-host placement inputs. `alloc_mib` is the host-measured -/// RAM headroom (`<= 0` = unmeasured); `cpu_budget` is `total_vcpus × -/// overcommit` (`0` = host hasn't reported its core count → no CPU gate). -#[derive(Clone, Copy, Debug, Default)] -struct HostFit { - alloc_mib: i64, - reserved_mib: i64, - cpu_budget: i64, - reserved_vcpus: i64, +/// The one SQL of the reservations on hosts (ADR 0046/0048/0081/0084): +/// per host, Σ `mem_budget_mib` and Σ `cpu_budget_vcpus` over the sessions +/// in a memory-reserving state (the SQL twin of +/// `SessionState::host_memory_reserving_states()`, a parked VM included), +/// the capture VMs of non-terminal capture jobs, and the reserving +/// teleport legs (destination and source). `hosts` is the host predicate +/// of the session and capture branches, e.g. `= ANY($1)` or `IS NOT NULL`. +/// Columns: `host_id`, `reserved_mib`, `reserved_vcpus`. +/// +/// R3 (#722): a `pending` row reserves unconditionally until it leaves the +/// reserving state. The ADR 0079 backstop reclaims a crash-orphan by a real +/// `pending → failed` transition, so placement never re-sells a slot a +/// revival could reclaim (Σ reserved > allocatable). +fn reserved_by_host_sql(hosts: &str) -> String { + format!( + r#" + SELECT host_id, + COALESCE(SUM(mem), 0)::BIGINT AS reserved_mib, + COALESCE(SUM(cpu), 0)::BIGINT AS reserved_vcpus + FROM ( + SELECT host_id, mem_budget_mib AS mem, cpu_budget_vcpus::BIGINT AS cpu + FROM sessions + WHERE host_id {hosts} + AND status IN ({reserving}) + UNION ALL + SELECT host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT + FROM capture_jobs + WHERE host_id {hosts} + AND stage NOT IN ('done','failed') + UNION ALL + SELECT dest_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT + FROM session_teleports WHERE phase IN ({teleport_reserving}) + UNION ALL + SELECT source_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT + FROM session_teleports WHERE phase IN ({source_reserving}) + ) reserved + GROUP BY host_id + "#, + reserving = reserving_states_sql(), + teleport_reserving = engram_core::types::teleport::TeleportPhase::reserving_phases_sql(), + source_reserving = + engram_core::types::teleport::TeleportPhase::source_reserving_phases_sql(), + ) +} + +/// The fit map: each host row's [`HostHeadroom`] (`id`, `allocatable_mib`, +/// `total_vcpus` columns) with the reservations of [`reserved_by_host_sql`] +/// applied. A reservation on a host not in `host_rows` is dropped. +fn headroom_map( + host_rows: &[sqlx::postgres::PgRow], + reserved_rows: &[sqlx::postgres::PgRow], +) -> Result, MetaError> { + let mut fit = std::collections::HashMap::with_capacity(host_rows.len()); + for r in host_rows { + let hid: uuid::Uuid = sqlx::Row::try_get(r, "id").map_err(db_err)?; + let allocatable_mib: i64 = sqlx::Row::try_get(r, "allocatable_mib").map_err(db_err)?; + let total_vcpus: i32 = sqlx::Row::try_get(r, "total_vcpus").map_err(db_err)?; + fit.insert( + hid, + HostHeadroom { + allocatable_mib, + cpu_budget: engram_core::types::host::host_cpu_budget(total_vcpus.max(0) as u32), + reserved: ReservedBudget::default(), + }, + ); + } + for r in reserved_rows { + let h: uuid::Uuid = sqlx::Row::try_get(r, "host_id").map_err(db_err)?; + if let Some(f) = fit.get_mut(&h) { + f.reserved = ReservedBudget { + mem_mib: sqlx::Row::try_get(r, "reserved_mib").map_err(db_err)?, + vcpus: sqlx::Row::try_get(r, "reserved_vcpus").map_err(db_err)?, + }; + } + } + Ok(fit) } /// ADR 0046/0048: BEST-FIT, 2D placement among `candidates` (ranked, with the @@ -536,7 +604,7 @@ struct HostFit { fn choose_placement_host( candidates: &[uuid::Uuid], affinity_len: usize, - fit: &std::collections::HashMap, + fit: &std::collections::HashMap, budget_mib: i64, budget_vcpus: i64, ) -> Option { @@ -548,7 +616,7 @@ fn choose_placement_host( .or_else(|| { candidates .iter() - .find(|h| fit.get(h).is_some_and(|f| f.alloc_mib <= 0)) + .find(|h| fit.get(h).is_some_and(|f| !f.measured())) .copied() }) } @@ -560,7 +628,7 @@ fn choose_placement_host( /// unit-tested without a database. fn classify_no_fit( candidates: &[uuid::Uuid], - fit: &std::collections::HashMap, + fit: &std::collections::HashMap, budget_mib: i64, budget_vcpus: i64, ) -> Vec { @@ -575,13 +643,9 @@ fn classify_no_fit( free_vcpus: 0, }; }; - let free_vcpus = if f.cpu_budget > 0 { - f.cpu_budget - f.reserved_vcpus - } else { - i64::MAX - }; - let free_mib = f.alloc_mib - f.reserved_mib; - let reason = if f.alloc_mib <= 0 { + let free_vcpus = f.free_vcpus().unwrap_or(i64::MAX); + let free_mib = f.free_mib(); + let reason = if !f.measured() { "unmeasured" } else if free_mib < budget_mib { "ram_full" @@ -605,7 +669,7 @@ fn classify_no_fit( /// Best-fit (smallest free RAM that fits both dims) among MEASURED hosts. fn best_fit_measured( candidates: &[uuid::Uuid], - fit: &std::collections::HashMap, + fit: &std::collections::HashMap, budget_mib: i64, budget_vcpus: i64, ) -> Option { @@ -614,17 +678,13 @@ fn best_fit_measured( let Some(f) = fit.get(h) else { continue; // not ready/draining at lock time }; - if f.alloc_mib <= 0 { + if !f.measured() { continue; // unmeasured — handled by the fallback tier } - let free_mib = f.alloc_mib - f.reserved_mib; - if free_mib < budget_mib { - continue; - } - // CPU dimension: only gate when the host reported a budget. - if f.cpu_budget > 0 && f.cpu_budget - f.reserved_vcpus < budget_vcpus { + if !f.fits(budget_mib, budget_vcpus) { continue; } + let free_mib = f.free_mib(); // SMALLEST free that fits (best-fit); ties toward the earlier // (higher-ranked) candidate. match best { @@ -678,91 +738,14 @@ async fn pick_host_2d( .fetch_all(&mut **tx) .await .map_err(db_err)?; - // ADR 0046/0048: build the 2D fit map. allocatable_mib is the - // host-measured RAM headroom (nets out daemon/OS/chunk-cache/mlock - // baseline; 0 = unmeasured). The CPU budget is total_vcpus × - // overcommit (0 = host hasn't reported its core count → no CPU gate). - let mut fit: std::collections::HashMap = - std::collections::HashMap::with_capacity(host_rows.len()); - for r in &host_rows { - let hid: uuid::Uuid = sqlx::Row::try_get(r, "id").map_err(db_err)?; - let alloc_mib: i64 = sqlx::Row::try_get(r, "allocatable_mib").map_err(db_err)?; - let total_vcpus: i32 = sqlx::Row::try_get(r, "total_vcpus").map_err(db_err)?; - fit.insert( - hid, - HostFit { - alloc_mib, - reserved_mib: 0, - cpu_budget: engram_core::types::host::host_cpu_budget(total_vcpus.max(0) as u32), - reserved_vcpus: 0, - }, - ); - } - // Reserved within the txn — sees the committed reservations of - // placers that locked these hosts before us. The sessions branch is - // the SQL twin of `SessionState::host_memory_reserving_states()` - // (interpolated from the const — see `reserving_states_sql`); - // the enable_jobs branch is ADR 0081's capture-VM reservation. - let res_sql = format!( - r#" - SELECT host_id, - COALESCE(SUM(mem), 0)::BIGINT AS reserved_mib, - COALESCE(SUM(cpu), 0)::BIGINT AS reserved_vcpus - FROM ( - SELECT host_id, mem_budget_mib AS mem, cpu_budget_vcpus::BIGINT AS cpu - FROM sessions - WHERE host_id = ANY($1) - AND status IN ({reserving}) - -- R3 (#722): ONE reservation authority — a `pending` pinned to - -- a host reserves its budget UNCONDITIONALLY, for exactly as - -- long as it is `pending`. No wall-age / live-op exclusion: an - -- aged crash-orphan physically holds its slot until it LEAVES - -- the reserving state, and the ADR 0079 pending-orphan backstop - -- reclaims it by a REAL `pending → failed` transition (the sole - -- reclaimer) — never by a placement-side exclusion. The old - -- 10-minute crash-orphan gate stopped counting an aged pending - -- while the backstop could still revive it, so a revival landed - -- on re-sold capacity (Σ reserved > allocatable). This predicate - -- now equals the unconditional placement-accounting oracle by - -- construction (engram-dst invariants.rs). - UNION ALL - -- ADR 0084 (c): a capture VM reserves like a session. The - -- reservation lives on `capture_jobs` now (moved off - -- `enable_jobs`); a non-terminal row with a bound host_id - -- holds its budget. Release is implicit — a terminal stage - -- drops out of this SUM. - SELECT host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM capture_jobs - WHERE host_id = ANY($1) - AND stage NOT IN ('done','failed') - UNION ALL - SELECT dest_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM session_teleports WHERE phase IN ({teleport_reserving}) - UNION ALL - SELECT source_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM session_teleports WHERE phase IN ({source_reserving}) - ) reserved - GROUP BY host_id - "#, - reserving = reserving_states_sql(), - teleport_reserving = engram_core::types::teleport::TeleportPhase::reserving_phases_sql(), - source_reserving = - engram_core::types::teleport::TeleportPhase::source_reserving_phases_sql(), - ); - let res_rows = sqlx::query(&res_sql) + // Reserved within the txn: sees the committed reservations of placers + // that locked these hosts before us. + let res_rows = sqlx::query(&reserved_by_host_sql("= ANY($1)")) .bind(cand) .fetch_all(&mut **tx) .await .map_err(db_err)?; - for r in &res_rows { - let h: uuid::Uuid = sqlx::Row::try_get(r, "host_id").map_err(db_err)?; - let mem: i64 = sqlx::Row::try_get(r, "reserved_mib").map_err(db_err)?; - let cpu: i64 = sqlx::Row::try_get(r, "reserved_vcpus").map_err(db_err)?; - if let Some(f) = fit.get_mut(&h) { - f.reserved_mib = mem; - f.reserved_vcpus = cpu; - } - } + let fit = headroom_map(&host_rows, &res_rows)?; Ok(choose_placement_host( cand, affinity_len, @@ -777,7 +760,8 @@ async fn pick_host_2d( #[allow(clippy::items_after_test_module)] #[cfg(test)] mod placement_tests { - use super::{choose_placement_host, HostFit}; + use super::choose_placement_host; + use engram_core::types::host::{HostHeadroom, ReservedBudget}; use std::collections::HashMap; use uuid::Uuid; @@ -785,22 +769,28 @@ mod placement_tests { (1..=n as u128).map(Uuid::from_u128).collect() } + fn headroom( + alloc: i64, + reserved_mib: i64, + cpu_budget: i64, + reserved_vcpus: i64, + ) -> HostHeadroom { + HostHeadroom { + allocatable_mib: alloc, + cpu_budget, + reserved: ReservedBudget { + mem_mib: reserved_mib, + vcpus: reserved_vcpus, + }, + } + } + /// Build a fit map from (alloc_mib, reserved_mib) pairs; CPU budget /// left at 0 (= no CPU gate) unless a test overrides it. - fn ram_fit(entries: &[(Uuid, i64, i64)]) -> HashMap { + fn ram_fit(entries: &[(Uuid, i64, i64)]) -> HashMap { entries .iter() - .map(|&(id, alloc, reserved)| { - ( - id, - HostFit { - alloc_mib: alloc, - reserved_mib: reserved, - cpu_budget: 0, - reserved_vcpus: 0, - }, - ) - }) + .map(|&(id, alloc, reserved)| (id, headroom(alloc, reserved, 0, 0))) .collect() } @@ -825,12 +815,7 @@ mod placement_tests { ]); fit.insert( h[2], - HostFit { - alloc_mib: 32768, - reserved_mib: 0, - cpu_budget: 8, - reserved_vcpus: 8, // free 0 < 2 → cpu_full - }, + headroom(32768, 0, 8, 8), // free 0 < 2 → cpu_full ); // h[3] deliberately absent from the map → not_lockable. let details = super::classify_no_fit(&h, &fit, 4096, 2); @@ -873,24 +858,8 @@ mod placement_tests { // land on h1 despite h0 being the tighter RAM fit. let h = ids(2); let fit: HashMap<_, _> = [ - ( - h[0], - HostFit { - alloc_mib: 32768, - reserved_mib: 28000, - cpu_budget: 8, - reserved_vcpus: 8, - }, - ), - ( - h[1], - HostFit { - alloc_mib: 32768, - reserved_mib: 0, - cpu_budget: 32, - reserved_vcpus: 0, - }, - ), + (h[0], headroom(32768, 28000, 8, 8)), + (h[1], headroom(32768, 0, 32, 0)), ] .into(); assert_eq!(choose_placement_host(&h, 0, &fit, 4096, 2), Some(h[1])); @@ -2593,54 +2562,14 @@ impl MetadataStore for PostgresStore { std::collections::HashMap, MetaError, > { - // Same predicate as `pick_host_2d` / `fleet_free_mib` — the - // memory-reserving states (R3 #722: a `pending` reserves - // UNCONDITIONALLY until it leaves that state; no crash-orphan gate), - // UNION the capturing enable jobs (ADR 0081) — summing BOTH budget - // dimensions (ADR 0048). - let sql = format!( - r#" - SELECT host_id, - COALESCE(SUM(mem), 0)::BIGINT, - COALESCE(SUM(cpu), 0)::BIGINT - FROM ( - SELECT host_id, mem_budget_mib AS mem, cpu_budget_vcpus::BIGINT AS cpu - FROM sessions - WHERE host_id IS NOT NULL - AND status IN ({reserving}) - UNION ALL - -- ADR 0084 (c): capturing VMs reserve on `capture_jobs`. - SELECT host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM capture_jobs - WHERE host_id IS NOT NULL - AND stage NOT IN ('done','failed') - UNION ALL - SELECT dest_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM session_teleports WHERE phase IN ({teleport_reserving}) - UNION ALL - SELECT source_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM session_teleports WHERE phase IN ({source_reserving}) - ) reserved - GROUP BY host_id - "#, - reserving = reserving_states_sql(), - teleport_reserving = - engram_core::types::teleport::TeleportPhase::reserving_phases_sql(), - source_reserving = - engram_core::types::teleport::TeleportPhase::source_reserving_phases_sql(), - ); + let sql = reserved_by_host_sql("IS NOT NULL"); let rows: Vec<(uuid::Uuid, i64, i64)> = sqlx::query_as(&sql) .fetch_all(&self.pool) .await .map_err(db_err)?; Ok(rows .into_iter() - .map(|(h, mem_mib, vcpus)| { - ( - HostId(h), - engram_core::types::host::ReservedBudget { mem_mib, vcpus }, - ) - }) + .map(|(h, mem_mib, vcpus)| (HostId(h), ReservedBudget { mem_mib, vcpus })) .collect()) } @@ -2732,75 +2661,12 @@ impl MetadataStore for PostgresStore { .fetch_all(&self.pool) .await .map_err(db_err)?; - let mut fit: std::collections::HashMap = - std::collections::HashMap::with_capacity(host_rows.len()); - for r in &host_rows { - let hid: uuid::Uuid = sqlx::Row::try_get(r, "id").map_err(db_err)?; - let alloc_mib: i64 = sqlx::Row::try_get(r, "allocatable_mib").map_err(db_err)?; - let total_vcpus: i32 = sqlx::Row::try_get(r, "total_vcpus").map_err(db_err)?; - fit.insert( - hid, - HostFit { - alloc_mib, - reserved_mib: 0, - cpu_budget: engram_core::types::host::host_cpu_budget( - total_vcpus.max(0) as u32 - ), - reserved_vcpus: 0, - }, - ); - } - let res_sql = format!( - r#" - SELECT host_id, - COALESCE(SUM(mem), 0)::BIGINT AS reserved_mib, - COALESCE(SUM(cpu), 0)::BIGINT AS reserved_vcpus - FROM ( - SELECT host_id, mem_budget_mib AS mem, cpu_budget_vcpus::BIGINT AS cpu - FROM sessions - WHERE host_id = ANY($1) - AND status IN ({reserving}) - -- R3 (#722): a `pending` reserves UNCONDITIONALLY until it - -- leaves the reserving state — no wall-age / live-op gate. - -- The ADR 0079 backstop reclaims a true crash-orphan by a - -- real `pending → failed` transition (the sole reclaimer), - -- so placement never re-sells a slot a revival could reclaim - -- (Σ reserved > allocatable). Matches the unconditional - -- placement-accounting oracle by construction. - UNION ALL - SELECT host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM capture_jobs - WHERE host_id = ANY($1) - AND stage NOT IN ('done','failed') - UNION ALL - SELECT dest_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM session_teleports WHERE phase IN ({teleport_reserving}) - UNION ALL - SELECT source_host_id, mem_budget_mib, cpu_budget_vcpus::BIGINT - FROM session_teleports WHERE phase IN ({source_reserving}) - ) reserved - GROUP BY host_id - "#, - reserving = reserving_states_sql(), - teleport_reserving = - engram_core::types::teleport::TeleportPhase::reserving_phases_sql(), - source_reserving = - engram_core::types::teleport::TeleportPhase::source_reserving_phases_sql(), - ); - let res_rows = sqlx::query(&res_sql) + let res_rows = sqlx::query(&reserved_by_host_sql("= ANY($1)")) .bind(&cand) .fetch_all(&self.pool) .await .map_err(db_err)?; - for r in &res_rows { - let h: uuid::Uuid = sqlx::Row::try_get(r, "host_id").map_err(db_err)?; - let mem: i64 = sqlx::Row::try_get(r, "reserved_mib").map_err(db_err)?; - let cpu: i64 = sqlx::Row::try_get(r, "reserved_vcpus").map_err(db_err)?; - if let Some(f) = fit.get_mut(&h) { - f.reserved_mib = mem; - f.reserved_vcpus = cpu; - } - } + let fit = headroom_map(&host_rows, &res_rows)?; Ok(classify_no_fit( &cand, &fit, diff --git a/crates/engram-sandbox-firecracker/src/lib.rs b/crates/engram-sandbox-firecracker/src/lib.rs index e9621c1d1..48141645e 100644 --- a/crates/engram-sandbox-firecracker/src/lib.rs +++ b/crates/engram-sandbox-firecracker/src/lib.rs @@ -3309,8 +3309,14 @@ impl FirecrackerBackend { if let Some(status) = handler.try_wait().map_err(|e| { SandboxError::Snapshot(format!("check uffd handler: {e}")) })? { + // The handler's log holds the only reason. + let log = read_tail(&jail_dir.join("uffd-handler.log"), 2048) + .await + .unwrap_or_default(); return Err(SandboxError::Snapshot(format!( - "uffd handler exited before the load ({status})" + "uffd handler exited before the load ({status}); \ + handler log: {}", + log.trim().replace('\n', " | ") ))); } if uffd_uds.exists() && state_path.exists() { diff --git a/crates/engram-sim/src/meta/store_impl.rs b/crates/engram-sim/src/meta/store_impl.rs index edbd945d3..df73be0db 100644 --- a/crates/engram-sim/src/meta/store_impl.rs +++ b/crates/engram-sim/src/meta/store_impl.rs @@ -56,6 +56,16 @@ fn reserves(state: SessionState) -> bool { state.reserves_host_memory() } +/// `h`'s [`engram_core::types::host::HostHeadroom`] with the reservations +/// of [`reserved_budgets`]. +fn headroom_of( + h: &HostRecord, + reserved: &std::collections::BTreeMap, +) -> engram_core::types::host::HostHeadroom { + let (mem_mib, vcpus) = reserved.get(&h.id).copied().unwrap_or((0, 0)); + engram_core::types::host::HostHeadroom::of(h, ReservedBudget { mem_mib, vcpus }) +} + fn reserved_budgets(db: &SimDb) -> std::collections::BTreeMap { let mut out = std::collections::BTreeMap::::new(); let sessions = db @@ -252,18 +262,12 @@ impl SimMetadataStore { return None; } let reserved = reserved_budgets(db); + // Measured hosts that fit, with their free RAM; an unmeasured host + // is the last-resort tier below. let fits = |h: &&HostRecord| -> Option { - let alloc = h.utilization.allocatable_mib as i64; - if alloc <= 0 { - return None; // unmeasured — last-resort tier - } - let (mem_res, cpu_res) = reserved.get(&h.id).copied().unwrap_or((0, 0)); - let free_mem = alloc - mem_res; - // CPU budgets are OVERCOMMITTED (host_cpu_budget = vcpus x - // factor) — conformance caught SimMeta using raw vcpus. - let cpu_budget = engram_core::types::host::host_cpu_budget(h.total_vcpus); - let free_cpu = cpu_budget - cpu_res; - (free_mem >= mem_budget_mib && free_cpu >= cpu_budget_vcpus).then_some(free_mem) + let room = headroom_of(h, &reserved); + (room.measured() && room.fits(mem_budget_mib, cpu_budget_vcpus)) + .then_some(room.free_mib()) }; let best_of = |tier: &[&HostRecord]| -> Option { tier.iter() @@ -281,7 +285,7 @@ impl SimMetadataStore { // Unmeasured hosts as last resort, affinity order. eligible .iter() - .find(|h| h.utilization.allocatable_mib == 0) + .find(|h| !headroom_of(h, &reserved).measured()) .map(|h| h.id) }) } @@ -4842,16 +4846,10 @@ impl MetadataStore for SimMetadataStore { free_vcpus: 0, }; }; - let (res_mib, res_vcpus) = reserved.get(id).copied().unwrap_or((0, 0)); - let alloc = h.utilization.allocatable_mib as i64; - let cpu_budget = engram_core::types::host::host_cpu_budget(h.total_vcpus); - let free_vcpus = if cpu_budget > 0 { - cpu_budget - res_vcpus - } else { - i64::MAX - }; - let free_mib = alloc - res_mib; - let reason = if alloc <= 0 { + let room = headroom_of(h, &reserved); + let free_vcpus = room.free_vcpus().unwrap_or(i64::MAX); + let free_mib = room.free_mib(); + let reason = if !room.measured() { "unmeasured" } else if free_mib < mem_budget_mib { "ram_full" diff --git a/crates/engram-sim/tests/meta_conformance.rs b/crates/engram-sim/tests/meta_conformance.rs index e0cfe3dd8..291034d16 100644 --- a/crates/engram-sim/tests/meta_conformance.rs +++ b/crates/engram-sim/tests/meta_conformance.rs @@ -2974,6 +2974,55 @@ async fn placement_no_fit(ctx: &Ctx) { assert_eq!(details[0].reason, "cpu_full"); } +/// A host that measures its RAM but reports no cores places a session +/// that asks for CPU: an unreported core count is no CPU gate. +async fn a_host_without_reported_cores_has_no_cpu_gate(ctx: &Ctx) { + let meta = &ctx.meta; + let now = ctx.clock.now_utc(); + let host = HostId::new(); + meta.upsert_host(host_record(host, "conf-no-cores", now)) + .await + .unwrap(); + let mut hb = heartbeat_fixture(); + hb.utilization = + serde_json::from_value(serde_json::json!({"allocatable_mib": 8192})).expect("utilization"); + hb.total_vcpus = 0; + meta.touch_host_heartbeat(host, hb).await.unwrap(); + + let details = meta + .placement_no_fit_details(&[host], 4096, 2) + .await + .unwrap(); + assert_eq!(details[0].reason, "fits_now"); + let placed = meta + .reserve_and_persist_create( + engram_core::traits::metadata::SessionCreateWriteSet { + session_id: SessionId::new(), + spec: spec("conf:no-cores"), + mem_budget_mib: 4096, + cpu_budget_vcpus: 2, + sealed_secrets: None, + capabilities: Vec::new(), + integration_policy_json: None, + runtime_spec: engram_core::types::runtime_spec::RuntimeSpec::new( + Vec::new(), + None, + None, + Vec::new(), + ), + oauth_binding: None, + }, + &[host], + 0, + ) + .await + .unwrap(); + assert!(matches!( + placed, + engram_core::traits::metadata::CreateDisposition::Placed(_) + )); +} + /// The register-time rehydrate list and the reconcile assignment list are /// one listing: every VM-resident state with a bound sandbox on the host. /// A rung-parked `Evicting` survivor and a `Created` survivor are as @@ -5543,6 +5592,10 @@ conformance!( super::events_limit_clamps_identically_both_directions ); conformance!(t_placement_no_fit, super::placement_no_fit); +conformance!( + t_a_host_without_reported_cores_has_no_cpu_gate, + super::a_host_without_reported_cores_has_no_cpu_gate +); conformance!( t_enabled_image_config_update, super::enabled_image_config_update diff --git a/docs/adr/0074-parking-ladder.md b/docs/adr/0074-parking-ladder.md index f90392dc2..8e14ae257 100644 --- a/docs/adr/0074-parking-ladder.md +++ b/docs/adr/0074-parking-ladder.md @@ -152,6 +152,21 @@ ladder descends to. this (the `ensure_active` hold loop; the next scanner tick); the unit tests poll `wait_for_lease_free` before the follow-on step. No product code change — the retry paths already existed for rung 1. +- **A park holds until a queued session has no room (2026-11-10).** The + headroom gate read `allocatable_mib` as a share of total RAM. The + ledger does not net out parked VMs (the host adds a parked VM's memory + back, and the park keeps its reservation), so a host whose parks had + reserved its RAM read as roomy, held every park, and a queued resume + never placed on it. A percentage of unreserved RAM is no better: on a + packed host it refuses every park while nothing waits. The gate now + reads demand (`ParkView` in `idle_evictor.rs`): a session parks, and a + park holds, while every queued session fits on a host placement ranks. + When one does not, one park per host yields per reaper pass (the + oldest, on the host nearest to a fit; none where an eviction already + runs). A host reserved beyond its measured RAM yields a park too. + `ENGRAM_PARK_HEADROOM_FLOOR_PCT` and the raw `mem_total − mem_used` + fallback are gone; an unmeasured host (`allocatable_mib == 0`) does + not park. ### Post-ship incident: the un-pause vsock black-hole (2026-07-06)