wrenn/wrenn
2
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden envd concurrency and fix CoW loop device leaks

Browse Source 
envd (0.6.0):
- Run blocking syscalls (sync/drop_caches, directory walks and
  removals, stdin/pty writes) on the blocking pool so health checks
  and exec streams don't stall during snapshot quiesce or large-tree
  operations
- Stop writing sandbox metadata into envd's own environment at init;
  it was undefined behavior with the async runtime running
- Retry a memory preload that previously failed or was cancelled
  instead of blocking every future pause until the next init
- Match RAM ranges against kernel core segments on page boundaries so
  preload no longer rejects kernels with unaligned low-memory ranges
- Tolerate a concurrent request creating a directory first

host agent:
- Detach the copy-on-write loop device when snapshot removal fails so
  deleting its backing file can no longer orphan it permanently
- Clean the sandbox directory on pause rollback, and skip paused
  restore for directories still owned by a live VM that failed to
  re-attach
This commit is contained in:
Rafeed M. Bhuiyan
2026-07-07 20:48:39 +06:00
parent 7e1ce86413
commit e34c9e1abe
16 changed files with 297 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
envd-rs/Cargo.lock generated
View File

@ -529,7 +529,7 @@ dependencies = [
[[package]]
name = "envd"
version = "0.5.0"
version = "0.6.0"
dependencies = [
"async-stream",
"axum",

2
envd-rs/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "envd"
version = "0.5.0"
version = "0.6.0"
edition = "2024"
rust-version = "1.95"

22
envd-rs/src/http/init.rs
View File

@ -84,14 +84,8 @@ pub async fn post_init(
let mut err = state.mem_preload_error.lock().unwrap();
state.mem_preload_generation.fetch_add(1, Ordering::SeqCst);
state.mem_preload_cancel.store(false, Ordering::SeqCst);
state.mem_preload_done.store(false, Ordering::SeqCst);
state.mem_preload_started.store(false, Ordering::SeqCst);
state.mem_preload_regions.store(0, Ordering::SeqCst);
state.mem_preload_pages.store(0, Ordering::SeqCst);
state.mem_preload_bytes.store(0, Ordering::SeqCst);
state.mem_preload_elapsed_us.store(0, Ordering::SeqCst);
state.mem_preload_source.store(0, Ordering::SeqCst);
*err = None;
state.reset_preload_run(&mut err);
}
if let Some(ref port_sub) = state.port_subsystem {
@ -203,11 +197,15 @@ pub async fn post_init(
futures::future::join_all(futs).await;
}
// Set sandbox/template metadata from request body.
// Set sandbox/template metadata from request body. Deliberately NOT
// written into envd's own process environment: std::env::set_var is
// undefined behavior with the multi-threaded runtime live (concurrent
// getenv from any thread races the environ rewrite), and nothing needs
// it there — spawned processes get these via defaults.env_vars, and
// out-of-band consumers (e.g. the `envd ports` subcommand's
// read_identity) fall back to the run files.
if let Some(ref id) = init_req.sandbox_id {
tracing::debug!(sandbox_id = %id, "setting sandbox ID from init request");
// SAFETY: envd is single-threaded at init time; no concurrent env reads.
unsafe { std::env::set_var("WRENN_SANDBOX_ID", id) };
write_run_file(".WRENN_SANDBOX_ID", id);
state
.defaults
@ -216,8 +214,6 @@ pub async fn post_init(
}
if let Some(ref id) = init_req.template_id {
tracing::debug!(template_id = %id, "setting template ID from init request");
// SAFETY: envd is single-threaded at init time; no concurrent env reads.
unsafe { std::env::set_var("WRENN_TEMPLATE_ID", id) };
write_run_file(".WRENN_TEMPLATE_ID", id);
state
.defaults
@ -227,8 +223,6 @@ pub async fn post_init(
if let Some(ref domain) = init_req.proxy_domain {
if !domain.is_empty() {
tracing::debug!(proxy_domain = %domain, "setting proxy domain from init request");
// SAFETY: envd is single-threaded at init time; no concurrent env reads.
unsafe { std::env::set_var("WRENN_PROXY_DOMAIN", domain) };
write_run_file(".WRENN_PROXY_DOMAIN", domain);
state
.defaults

157
envd-rs/src/http/memory.rs
View File

@ -10,8 +10,11 @@
// handler fills the page from the source memory-ranges file.
//
// Wire protocol:
// POST /memory/preload — starts the loader (idempotent) and returns
// the current status JSON immediately
// POST /memory/preload — starts the loader (idempotent while a run
// is in flight or completed successfully;
// restarts a run that ended failed/cancelled)
// and returns the current status JSON
// immediately
// GET /memory/preload — returns the current status JSON
// POST /memory/preload/cancel — signals the loader to stop early
//
@ -50,29 +53,40 @@ pub struct PreloadStatus {
pub async fn post_memory_preload(State(state): State<Arc<AppState>>) -> impl IntoResponse {
// First caller wins the CAS and spawns the loader; subsequent callers
// just report the existing status.
let we_start = state
.mem_preload_started
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_ok();
if we_start {
// Fresh run: clear leftovers from a previous lifecycle's loader so a
// stale done=true (or stale counters) can't masquerade as this run's
// result. Done under the error mutex, the same critical section /init
// and result publication use.
{
let mut err = state.mem_preload_error.lock().unwrap();
*err = None;
state.mem_preload_done.store(false, Ordering::SeqCst);
state.mem_preload_regions.store(0, Ordering::SeqCst);
state.mem_preload_pages.store(0, Ordering::SeqCst);
state.mem_preload_bytes.store(0, Ordering::SeqCst);
state.mem_preload_elapsed_us.store(0, Ordering::SeqCst);
state.mem_preload_source.store(0, Ordering::SeqCst);
// just report the existing status — EXCEPT when the previous run finished
// in failed/cancelled state, which may be retried. Without the retry, one
// transient failure would block the host's pause/snapshot gate until the
// next /init (which never comes while the sandbox stays running).
//
// Both the fresh-run reset and the retry check run under the error mutex
// — the same critical section /init's reset and the loader's result
// publication use — so concurrent POSTs cannot both claim a retry and a
// frozen /init cannot interleave.
// The generation MUST be captured inside the same mutex-held block as the
// CAS/reset: loading it after the lock drops lets an interleaving /init
// (bump + started=false) tag this loader with the NEW generation, and a
// follow-up POST would then spawn a second concurrent full-RAM walk.
let start_generation = {
let mut err = state.mem_preload_error.lock().unwrap();
let fresh = state
.mem_preload_started
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_ok();
let retry = !fresh
&& state.mem_preload_done.load(Ordering::SeqCst)
&& (err.is_some() || state.mem_preload_cancel.load(Ordering::SeqCst));
if fresh || retry {
// Clear leftovers (previous lifecycle's or failed run's result) so
// a stale done=true can't masquerade as this run's completion.
state.reset_preload_run(&mut err);
state.mem_preload_cancel.store(false, Ordering::SeqCst);
Some(state.mem_preload_generation.load(Ordering::SeqCst))
} else {
None
}
let generation = state.mem_preload_generation.load(Ordering::SeqCst);
};
if let Some(generation) = start_generation {
let state_clone = Arc::clone(&state);
// Detached blocking thread — no axum task lifetime ties it to the
// request, so the connection can close immediately without aborting
@ -373,45 +387,63 @@ struct RamSegment {
// Match every System RAM range from /proc/iomem to its KCORE_RAM PT_LOAD
// segment. RAM segments share a single direct-map base (vaddr = base +
// phys_start; base is KASLR-randomised, so it must be derived, not assumed).
// Candidate bases come from exact-size (segment, range) pairs; a base is
// accepted only if EVERY RAM range has a segment at base + start with the
// exact range size. This can never select vmalloc/vmemmap segments: their
// Candidate bases come from size-matched (segment, range) pairs; a base is
// accepted only if EVERY RAM range has a segment at its expected vaddr with
// the expected size. This can never select vmalloc/vmemmap segments: their
// sizes (tens of TB) match no iomem range.
//
// Sizes are compared after clamping the iomem range to page boundaries:
// kcore's segments are PFN-granular (walk_system_ram_range uses
// PFN_UP/PFN_DOWN) while iomem ranges are byte-granular — the classic
// 0x1000-0x9fbff low-RAM range is 0x9ec00 bytes in iomem but 0x9e000 in
// kcore. Exact byte matching would reject every kernel that has such a range.
fn match_ram_segments(
ranges: &[(u64, u64)],
segments: &[KcoreSegment],
) -> Result<Vec<RamSegment>, String> {
const KERNEL_SPACE_MIN: u64 = 0xffff_8000_0000_0000;
// Page-clamp: [PFN_UP(start), PFN_DOWN(end)). Ranges smaller than one
// page after clamping have no kcore segment and nothing to preload.
let clamped: Vec<(u64, u64)> = ranges
.iter()
.map(|(start, end)| (start.next_multiple_of(PAGE_SIZE), end & !(PAGE_SIZE - 1)))
.filter(|(start, end)| end > start)
.collect();
if clamped.is_empty() {
return Err("no page-sized System RAM ranges after clamping".into());
}
let mut candidates: Vec<u64> = Vec::new();
for s in segments.iter().filter(|s| s.vaddr >= KERNEL_SPACE_MIN) {
for (start, end) in ranges {
let len = end - start;
if s.file_size == len && s.vaddr.checked_sub(*start).is_some() {
candidates.push(s.vaddr - start);
for (start, end) in &clamped {
if s.file_size == end - start {
if let Some(base) = s.vaddr.checked_sub(*start) {
candidates.push(base);
}
}
}
}
candidates.sort_unstable();
candidates.dedup();
let try_base = |base: u64| -> Option<Vec<RamSegment>> {
clamped
.iter()
.map(|(start, end)| {
segments
.iter()
.find(|s| s.vaddr == base + start && s.file_size == end - start)
.map(|s| RamSegment {
file_offset: s.file_offset,
len: end - start,
})
})
.collect()
};
for base in &candidates {
let mut out = Vec::with_capacity(ranges.len());
for (start, end) in ranges {
let len = end - start;
let Some(seg) = segments
.iter()
.find(|s| s.vaddr == base.wrapping_add(*start) && s.file_size == len)
else {
out.clear();
break;
};
out.push(RamSegment {
file_offset: seg.file_offset,
len,
});
}
if out.len() == ranges.len() {
if let Some(out) = try_base(*base) {
return Ok(out);
}
}
@ -419,7 +451,7 @@ fn match_ram_segments(
Err(format!(
"no consistent direct-map base for {} RAM ranges across {} PT_LOAD segments \
({} candidate bases tried)",
ranges.len(),
clamped.len(),
segments.len(),
candidates.len(),
))
@ -458,8 +490,11 @@ mod tests {
]
}
// Byte-granular, as /proc/iomem reports them: the low range ends at
// 0x9fbff (parse adds 1 → 0x9fc00), NOT page-aligned. kcore's segment for
// it is PFN-clamped to 0x9e000 bytes — matching must tolerate that.
fn typical_ranges() -> Vec<(u64, u64)> {
vec![(0x1000, 0x9f000), (0x100000, 2 * GIB)]
vec![(0x1000, 0x9fc00), (0x100000, 2 * GIB)]
}
#[test]
@ -505,6 +540,32 @@ mod tests {
assert!(match_ram_segments(&typical_ranges(), &segments).is_err());
}
#[test]
fn unaligned_range_is_page_clamped() {
// Range with unaligned start AND end: kcore clamps to
// [PFN_UP(start), PFN_DOWN(end)).
let ranges = vec![(0x2400, 0x9fbff + 1)];
let segments = vec![
seg(0xffff_c900_0000_0000, 32 * TIB, 0x1000),
seg(BASE + 0x3000, 0x9c000, 0x10000), // 0x3000..0x9f000
];
let out = match_ram_segments(&ranges, &segments).unwrap();
assert_eq!(out.len(), 1);
assert_eq!(out[0].file_offset, 0x10000);
assert_eq!(out[0].len, 0x9c000);
}
#[test]
fn sub_page_range_is_skipped() {
// A range smaller than one page after clamping (e.g. 0x9fc00-0x9ffff)
// has no kcore segment; it must be dropped, not fail the match.
let ranges = vec![(0x9fc00, 0xa0000), (0x100000, 2 * GIB)];
let segments = vec![seg(BASE + 0x100000, 2 * GIB - 0x100000, 0x20000)];
let out = match_ram_segments(&ranges, &segments).unwrap();
assert_eq!(out.len(), 1);
assert_eq!(out[0].file_offset, 0x20000);
}
#[test]
fn ambiguous_same_size_ranges_still_resolve() {
// Two RAM ranges of identical size: candidate bases from cross pairs

40
envd-rs/src/http/snapshot.rs
View File

@ -20,15 +20,11 @@ pub async fn post_snapshot_prepare(State(state): State<Arc<AppState>>) -> impl I
port_sub.stop();
}
// sync(2) flushes the in-memory FS state. Done before drop_caches so the
// pages we drop are clean.
sync();
// Drop the VFS page cache + dentries/inodes. Reduces snapshot size by
// ensuring CH only persists memory pages that the guest actually needs.
if let Err(e) = std::fs::write("/proc/sys/vm/drop_caches", "3") {
tracing::warn!(error = %e, "drop_caches (first pass) failed (continuing)");
}
// Flush in-memory FS state, then drop the VFS page cache +
// dentries/inodes. sync first so the pages we drop are clean; dropping
// reduces snapshot size by ensuring CH only persists memory pages the
// guest actually needs.
flush_and_drop_caches("first pass").await;
// Best-effort fstrim on the rootfs so unused blocks are returned to the
// dm-snapshot, keeping CoW size minimal.
@ -37,14 +33,10 @@ pub async fn post_snapshot_prepare(State(state): State<Arc<AppState>>) -> impl I
.output()
.await;
// Second drop_caches pass after fstrim: fstrim re-reads superblock /
// group descriptor pages that we just evicted, putting them back in the
// page cache. A second pass drops those and any other late readers (e.g.
// sync flushers).
sync();
if let Err(e) = std::fs::write("/proc/sys/vm/drop_caches", "3") {
tracing::warn!(error = %e, "drop_caches (second pass) failed (continuing)");
}
// Second pass after fstrim: fstrim re-reads superblock / group descriptor
// pages that we just evicted, putting them back in the page cache. This
// drops those and any other late readers (e.g. sync flushers).
flush_and_drop_caches("second pass").await;
// No balloon settle window here: free-page reporting drains asynchronously,
// so any pages not yet hole-punched by the host at snapshot time are written
@ -59,3 +51,17 @@ pub async fn post_snapshot_prepare(State(state): State<Arc<AppState>>) -> impl I
[(header::CACHE_CONTROL, "no-store")],
)
}
// sync(2) can block for seconds flushing dirty pages, and the drop_caches
// write blocks while the kernel evicts — both run on the blocking pool or
// they starve every async task sharing the worker thread (health probes,
// exec streams) for the whole quiesce window.
async fn flush_and_drop_caches(pass: &'static str) {
let _ = tokio::task::spawn_blocking(move || {
sync();
if let Err(e) = std::fs::write("/proc/sys/vm/drop_caches", "3") {
tracing::warn!(error = %e, pass, "drop_caches failed (continuing)");
}
})
.await;
}

16
envd-rs/src/permissions/path.rs
View File

@ -57,10 +57,18 @@ pub fn ensure_dirs(path: &str, uid: Uid, gid: Gid) -> Result<(), String> {
}
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
fs::create_dir(&current)
.map_err(|e| format!("failed to create directory {current_str}: {e}"))?;
chown(&current, Some(uid.as_raw()), Some(gid.as_raw()))
.map_err(|e| format!("failed to chown directory {current_str}: {e}"))?;
if let Err(ce) = fs::create_dir(&current) {
// Concurrent request may have created it between the stat
// and here — fine as long as it's a directory now. The
// winner did its own chown; don't re-chown their dir.
let now_dir = fs::metadata(&current).map(|m| m.is_dir()).unwrap_or(false);
if !now_dir {
return Err(format!("failed to create directory {current_str}: {ce}"));
}
} else {
chown(&current, Some(uid.as_raw()), Some(gid.as_raw()))
.map_err(|e| format!("failed to chown directory {current_str}: {e}"))?;
}
}
Err(e) => {
return Err(format!("failed to stat directory {current_str}: {e}"));

64
envd-rs/src/rpc/filesystem_service.rs
View File

@ -159,26 +159,33 @@ impl Filesystem for FilesystemServiceImpl {
let path = self.resolve_path(request.path, &ctx)?;
let resolved = std::fs::canonicalize(&path).map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
ConnectError::new(ErrorCode::NotFound, format!("path not found: {e}"))
} else {
ConnectError::new(ErrorCode::Internal, format!("error resolving path: {e}"))
// The recursive walk stats every entry (plus uid/gid lookups) — on a
// large tree that is seconds of blocking syscalls, so it runs on the
// blocking pool instead of a runtime worker thread.
let entries = tokio::task::spawn_blocking(move || {
let resolved = std::fs::canonicalize(&path).map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
ConnectError::new(ErrorCode::NotFound, format!("path not found: {e}"))
} else {
ConnectError::new(ErrorCode::Internal, format!("error resolving path: {e}"))
}
})?;
let resolved_str = resolved.to_string_lossy().to_string();
let meta = std::fs::metadata(&resolved).map_err(|e| {
ConnectError::new(ErrorCode::Internal, format!("error getting file info: {e}"))
})?;
if !meta.is_dir() {
return Err(ConnectError::new(
ErrorCode::InvalidArgument,
format!("path is not a directory: {path}"),
));
}
})?;
let resolved_str = resolved.to_string_lossy().to_string();
let meta = std::fs::metadata(&resolved).map_err(|e| {
ConnectError::new(ErrorCode::Internal, format!("error getting file info: {e}"))
})?;
if !meta.is_dir() {
return Err(ConnectError::new(
ErrorCode::InvalidArgument,
format!("path is not a directory: {path}"),
));
}
let entries = walk_dir(&path, &resolved_str, depth)?;
walk_dir(&path, &resolved_str, depth)
})
.await
.map_err(|e| ConnectError::new(ErrorCode::Internal, format!("list_dir task: {e}")))??;
Ok((
ListDirResponse {
entries,
@ -195,14 +202,21 @@ impl Filesystem for FilesystemServiceImpl {
) -> Result<(RemoveResponse, Context), ConnectError> {
let path = self.resolve_path(request.path, &ctx)?;
if let Err(e1) = std::fs::remove_dir_all(&path) {
if let Err(e2) = std::fs::remove_file(&path) {
return Err(ConnectError::new(
ErrorCode::Internal,
format!("error removing: {e1}; also tried as file: {e2}"),
));
// remove_dir_all recurses through the whole tree — blocking pool, not
// a runtime worker thread.
tokio::task::spawn_blocking(move || {
if let Err(e1) = std::fs::remove_dir_all(&path) {
if let Err(e2) = std::fs::remove_file(&path) {
return Err(ConnectError::new(
ErrorCode::Internal,
format!("error removing: {e1}; also tried as file: {e2}"),
));
}
}
}
Ok(())
})
.await
.map_err(|e| ConnectError::new(ErrorCode::Internal, format!("remove task: {e}")))??;
Ok((
RemoveResponse {

33
envd-rs/src/rpc/process_service.rs
View File

@ -305,7 +305,7 @@ impl Process for ProcessServiceImpl {
ConnectError::new(ErrorCode::FailedPrecondition, "no start event received")
})?;
if let Some(input) = data.input.as_option() {
write_input(h, input)?;
write_input(h, input).await?;
}
}
Some(stream_input_request::EventView::Keepalive(_)) => {}
@ -332,7 +332,7 @@ impl Process for ProcessServiceImpl {
let handle = self.get_process_by_selector(selector)?;
if let Some(input) = request.input.as_option() {
write_input(&handle, input)?;
write_input(&handle, input).await?;
}
Ok((
@ -392,12 +392,29 @@ impl Process for ProcessServiceImpl {
}
}
fn write_input(handle: &ProcessHandle, input: &ProcessInputView) -> Result<(), ConnectError> {
match &input.input {
Some(process_input::InputView::Pty(d)) => handle.write_pty(d),
Some(process_input::InputView::Stdin(d)) => handle.write_stdin(d),
None => Ok(()),
}
// write_input copies the payload and performs the pipe/pty write on the
// blocking pool: write_all blocks the calling thread whenever the target
// process stops draining its input (full 64K pipe buffer, stopped job), and
// on a runtime worker thread that would stall every other in-flight request.
async fn write_input(
handle: &Arc<ProcessHandle>,
input: &ProcessInputView<'_>,
) -> Result<(), ConnectError> {
let (data, is_pty) = match &input.input {
Some(process_input::InputView::Pty(d)) => (d.to_vec(), true),
Some(process_input::InputView::Stdin(d)) => (d.to_vec(), false),
None => return Ok(()),
};
let h = Arc::clone(handle);
tokio::task::spawn_blocking(move || {
if is_pty {
h.write_pty(&data)
} else {
h.write_stdin(&data)
}
})
.await
.map_err(|e| ConnectError::new(ErrorCode::Internal, format!("input writer task: {e}")))?
}
/// Shared event pump for `Start` and `Connect`. Yields a leading start event,

19
envd-rs/src/state.rs
View File

@ -31,6 +31,9 @@ pub struct AppState {
pub mem_preload_started: AtomicBool,
pub mem_preload_done: AtomicBool,
pub mem_preload_cancel: AtomicBool,
/// See `reset_preload_run` for the per-run reset shared by /init and the
/// POST /memory/preload start/retry paths.
///
/// Bumped by every /init lifecycle change. A loader thread captures the
/// value at spawn and refuses to run — or to publish results — once it no
/// longer matches, so a thread that survived a pause/resume (the VM can be
@ -110,6 +113,22 @@ impl AppState {
/// Records a new lifecycle ID, returning true if it changed (i.e. this
/// is the first /init since a resume). First-ever call returns false:
/// Clears the per-run memory-preload result fields (done flag, counters,
/// source, error). Shared by /init's lifecycle reset and POST
/// /memory/preload's start/retry paths so the two field lists cannot
/// drift. The caller MUST hold the `mem_preload_error` mutex and pass its
/// guard's contents in — that mutex is the critical section that also
/// serializes the loader thread's result publication.
pub fn reset_preload_run(&self, err: &mut Option<String>) {
*err = None;
self.mem_preload_done.store(false, Ordering::SeqCst);
self.mem_preload_regions.store(0, Ordering::SeqCst);
self.mem_preload_pages.store(0, Ordering::SeqCst);
self.mem_preload_bytes.store(0, Ordering::SeqCst);
self.mem_preload_elapsed_us.store(0, Ordering::SeqCst);
self.mem_preload_source.store(0, Ordering::SeqCst);
}
/// boot-time /init doesn't need port-subsystem restart since the
/// subsystem hasn't been started yet by anything else.
pub fn bump_lifecycle(&self, new_id: &str) -> bool {

5
internal/hostagent/server.go
View File

@ -76,8 +76,11 @@ func (s *Server) CreateSandbox(
return nil, err
}
// disk_size_mb in the request is deprecated and never set by the control
// plane; passing 0 lets the manager pick the size (DefaultRootfsSizeMB,
// floored at the origin image size).
sb, diskSizeBytes, err := s.mgr.Create(ctx, msg.SandboxId, teamID, templateID,
int(msg.Vcpus), int(msg.MemoryMb), int(msg.TimeoutSec), int(msg.DiskSizeMb),
int(msg.Vcpus), int(msg.MemoryMb), int(msg.TimeoutSec), 0,
msg.DefaultUser, msg.DefaultEnv)
if err != nil {
if errors.Is(err, sandbox.ErrDraining) {

11
pkg/devicemapper/devicemapper.go
View File

@ -192,8 +192,19 @@ func RestoreSnapshot(ctx context.Context, name, originLoopDev, cowPath string, o
// RemoveSnapshot tears down a dm-snapshot device and its CoW loop device.
// The CoW file is NOT deleted — the caller decides whether to keep or remove it.
//
// The CoW loop is detached even when the dm removal fails (busy device):
// callers treat a failed removal as fatal teardown and typically delete the
// CoW backing file next, which would orphan a still-attached loop forever
// (losetup can no longer find it by file). Detaching first is always safe —
// while dm still references the loop the kernel just defers the release via
// autoclear until that reference drops.
func RemoveSnapshot(ctx context.Context, dev *SnapshotDevice) error {
if err := dmsetupRemove(ctx, dev.Name); err != nil {
if derr := losetupDetachRetry(dev.CowLoopDev); derr != nil {
slog.Warn("cow loop detach after failed dm remove",
"device", dev.CowLoopDev, "name", dev.Name, "error", derr)
}
return fmt.Errorf("dmsetup remove %s: %w", dev.Name, err)
}

5
pkg/envdclient/client.go
View File

@ -41,9 +41,10 @@ func New(hostIP string) *Client {
// of the standard http://{hostIP}:{envdPort}. Intended for tests (httptest
// servers) and nonstandard listeners.
func NewWithBaseURL(base string) *Client {
u, err := url.Parse(base)
host := base
if err == nil {
// Hostname() is empty for scheme-less inputs ("127.0.0.1:8080" parses
// with "127.0.0.1" as the scheme) — keep the raw base then.
if u, err := url.Parse(base); err == nil && u.Hostname() != "" {
host = u.Hostname()
}
return newWithBase(host, base)

12
pkg/sandbox/manager.go
View File

@ -622,14 +622,11 @@ func (m *Manager) cleanup(ctx context.Context, sb *sandboxState) {
m.slots.Release(sb.SlotIndex)
// Tear down dm-snapshot and release the base image loop device.
// RemoveSnapshot detaches the CoW loop even on failure, so deleting the
// CoW file below cannot orphan it.
if sb.dmDevice != nil {
if err := devicemapper.RemoveSnapshot(context.Background(), sb.dmDevice); err != nil {
slog.Warn("dm-snapshot remove error", "id", sb.ID, "error", err)
// RemoveSnapshot bails before detaching the CoW loop when the dm
// device is busy. The CoW file is deleted just below — detach the
// loop first (autoclear if dm still holds it) or it becomes
// permanently unreclaimable once its backing file is gone.
devicemapper.DetachLoopsByFile(sb.dmDevice.CowPath)
}
os.Remove(sb.dmDevice.CowPath)
}
@ -1285,12 +1282,11 @@ func (m *Manager) cleanupAfterCrash(sb *sandboxState) {
}
m.slots.Release(sb.SlotIndex)
// RemoveSnapshot detaches the CoW loop even on failure, so the RemoveAll
// below (which deletes the CoW backing file) cannot orphan it.
if sb.dmDevice != nil {
if err := devicemapper.RemoveSnapshot(context.Background(), sb.dmDevice); err != nil {
slog.Warn("crash cleanup: dm-snapshot error", "id", sb.ID, "error", err)
// os.RemoveAll below deletes the CoW backing file; detach its
// loop first or it can never be reclaimed (see DetachLoopsByFile).
devicemapper.DetachLoopsByFile(sb.dmDevice.CowPath)
}
}
if sb.baseImagePath != "" {

31
pkg/sandbox/pause.go
View File

@ -216,6 +216,15 @@ func (m *Manager) Pause(ctx context.Context, sandboxID string) error {
m.mu.Lock()
sb.SlotIndex = 0
m.mu.Unlock()
// Remove the sandbox dir wholesale, mirroring cleanupAfterCrash.
// This drops the running-state file — its recorded slot is freed
// and may be re-claimed, so a later restart's sweep must not tear
// down the new owner's network — and any previous pause
// generation, which is unusable anyway: the CoW has advanced past
// the memory image it was captured with.
if err := os.RemoveAll(layout.SandboxDir(m.cfg.WrennDir, sandboxID)); err != nil {
slog.Warn("pause rollback: sandbox dir remove failed", "id", sandboxID, "error", err)
}
return fmt.Errorf("pause %s: %s: %w (and resume failed: %v)",
sandboxID, stage, cause, rerr)
}
@ -511,14 +520,11 @@ func (m *Manager) releaseRuntime(sb *sandboxState, cow cowDisposition) {
m.slots.Release(sb.SlotIndex)
}
// RemoveSnapshot detaches the CoW loop even on failure, so dropCow's file
// deletion cannot orphan it and keepCow's Resume cannot double-attach.
if sb.dmDevice != nil {
if err := devicemapper.RemoveSnapshot(context.Background(), sb.dmDevice); err != nil {
slog.Warn("dm-snapshot remove on pause", "id", sb.ID, "error", err)
// The CoW loop was not detached (RemoveSnapshot stops on a busy
// dm device). Detach it by backing file — dropCow deletes the file
// below, and even for keepCow a lingering attachment would make
// Resume's losetupCreateRW attach a second loop to the same file.
devicemapper.DetachLoopsByFile(sb.dmDevice.CowPath)
}
if cow == dropCow {
os.Remove(sb.dmDevice.CowPath)
@ -815,15 +821,16 @@ func (m *Manager) ensureMemoryMaterialized(ctx context.Context, sb *sandboxState
return fmt.Errorf("memory preload verify: envd client unavailable")
}
status, err := client.GetMemoryPreloadStatus(ctx)
// One POST covers every state: "done" reports immediately, "running"
// reports the in-flight run, "idle" starts one (agent died between
// restore and the loader's POST, or an earlier start failed on the wire),
// and a previous "failed"/"cancelled" run is re-armed — so a transient
// guest-side failure blocks only this attempt, not every future pause.
status, err := client.StartMemoryPreload(ctx)
if err != nil {
return fmt.Errorf("memory preload verify: %w", err)
return fmt.Errorf("memory preload start: %w", err)
}
if status.State == "idle" || status.State == "running" {
if _, err := client.StartMemoryPreload(ctx); err != nil {
return fmt.Errorf("memory preload start: %w", err)
}
if status.State != "done" {
status, err = client.WaitMemoryPreload(ctx)
if err != nil {
return fmt.Errorf("memory preload wait: %w", err)

19
pkg/sandbox/pause_test.go
View File

@ -70,7 +70,7 @@ func TestEnsureMemoryMaterialized(t *testing.T) {
lazy: true,
states: []string{"done"},
wantErr: "",
wantPosts: 0,
wantPosts: 1, // single POST reports done, no polling
},
{
name: "lazy idle self-heals via start",
@ -80,25 +80,18 @@ func TestEnsureMemoryMaterialized(t *testing.T) {
wantPosts: 1,
},
{
name: "lazy failed is a hard error",
name: "lazy failed after wait is a hard error",
lazy: true,
states: []string{"failed"},
states: []string{"idle", "failed"},
wantErr: "memory preload failed",
wantPosts: 0,
},
{
name: "lazy start ends in failed",
lazy: true,
states: []string{"idle", "idle", "failed"},
wantErr: "memory preload failed",
wantPosts: -1,
wantPosts: 1,
},
{
name: "lazy cancelled is a hard error",
lazy: true,
states: []string{"cancelled"},
states: []string{"running", "cancelled"},
wantErr: "memory preload cancelled",
wantPosts: 0,
wantPosts: 1,
},
{
name: "lazy without client fails",

13
pkg/sandbox/restore_paused.go
View File

@ -78,6 +78,19 @@ func (m *Manager) RestorePausedSandboxes() {
continue
}
// A readable running-state file that survived RestoreRunningSandboxes
// means the adoption of a verified-LIVE CH process failed (probe
// timeout etc.) and its slot/loop were deliberately left held. The
// stale pause generation in the same dir must not be registered — and
// above all not trashed: trashing renames the whole sandbox dir,
// carrying away the live VM's CoW file and the state file the next
// restart needs to retry the adoption.
if _, err := readRunningState(m.cfg.WrennDir, name); err == nil {
slog.Warn("restore: skipping paused candidate — dir owned by live un-adopted VM",
"id", name)
continue
}
snapDir := layout.PauseSnapshotDir(m.cfg.WrennDir, name)
meta, err := readSnapshotMeta(snapDir)
if err != nil {