diff --git a/envd-rs/Cargo.lock b/envd-rs/Cargo.lock index f4c12f2..08109df 100644 --- a/envd-rs/Cargo.lock +++ b/envd-rs/Cargo.lock @@ -529,7 +529,7 @@ dependencies = [ [[package]] name = "envd" -version = "0.5.0" +version = "0.6.0" dependencies = [ "async-stream", "axum", diff --git a/envd-rs/Cargo.toml b/envd-rs/Cargo.toml index 95ded34..a9e41bf 100644 --- a/envd-rs/Cargo.toml +++ b/envd-rs/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "envd" -version = "0.5.0" +version = "0.6.0" edition = "2024" rust-version = "1.95" diff --git a/envd-rs/src/http/init.rs b/envd-rs/src/http/init.rs index da18994..86bdce4 100644 --- a/envd-rs/src/http/init.rs +++ b/envd-rs/src/http/init.rs @@ -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 diff --git a/envd-rs/src/http/memory.rs b/envd-rs/src/http/memory.rs index 02e65eb..36fcd81 100644 --- a/envd-rs/src/http/memory.rs +++ b/envd-rs/src/http/memory.rs @@ -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>) -> 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, 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 = 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> { + 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 diff --git a/envd-rs/src/http/snapshot.rs b/envd-rs/src/http/snapshot.rs index cbb2f4d..965449c 100644 --- a/envd-rs/src/http/snapshot.rs +++ b/envd-rs/src/http/snapshot.rs @@ -20,15 +20,11 @@ pub async fn post_snapshot_prepare(State(state): State>) -> 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>) -> 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>) -> 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; +} diff --git a/envd-rs/src/permissions/path.rs b/envd-rs/src/permissions/path.rs index 68791fa..b363682 100644 --- a/envd-rs/src/permissions/path.rs +++ b/envd-rs/src/permissions/path.rs @@ -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(¤t) - .map_err(|e| format!("failed to create directory {current_str}: {e}"))?; - chown(¤t, 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(¤t) { + // 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(¤t).map(|m| m.is_dir()).unwrap_or(false); + if !now_dir { + return Err(format!("failed to create directory {current_str}: {ce}")); + } + } else { + chown(¤t, 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}")); diff --git a/envd-rs/src/rpc/filesystem_service.rs b/envd-rs/src/rpc/filesystem_service.rs index 8c17d31..efc6d08 100644 --- a/envd-rs/src/rpc/filesystem_service.rs +++ b/envd-rs/src/rpc/filesystem_service.rs @@ -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 { diff --git a/envd-rs/src/rpc/process_service.rs b/envd-rs/src/rpc/process_service.rs index 556abf5..70be370 100644 --- a/envd-rs/src/rpc/process_service.rs +++ b/envd-rs/src/rpc/process_service.rs @@ -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, + 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, diff --git a/envd-rs/src/state.rs b/envd-rs/src/state.rs index 180e824..933956e 100644 --- a/envd-rs/src/state.rs +++ b/envd-rs/src/state.rs @@ -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) { + *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 { diff --git a/internal/hostagent/server.go b/internal/hostagent/server.go index f9cdaf4..fa4ca68 100644 --- a/internal/hostagent/server.go +++ b/internal/hostagent/server.go @@ -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) { diff --git a/pkg/devicemapper/devicemapper.go b/pkg/devicemapper/devicemapper.go index 52325a2..2a7ce1c 100644 --- a/pkg/devicemapper/devicemapper.go +++ b/pkg/devicemapper/devicemapper.go @@ -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) } diff --git a/pkg/envdclient/client.go b/pkg/envdclient/client.go index 369daa5..3480de8 100644 --- a/pkg/envdclient/client.go +++ b/pkg/envdclient/client.go @@ -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) diff --git a/pkg/sandbox/manager.go b/pkg/sandbox/manager.go index 55919a6..0dda4cb 100644 --- a/pkg/sandbox/manager.go +++ b/pkg/sandbox/manager.go @@ -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 != "" { diff --git a/pkg/sandbox/pause.go b/pkg/sandbox/pause.go index 9448e0f..cf2199f 100644 --- a/pkg/sandbox/pause.go +++ b/pkg/sandbox/pause.go @@ -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) diff --git a/pkg/sandbox/pause_test.go b/pkg/sandbox/pause_test.go index 98f149c..9f6eedd 100644 --- a/pkg/sandbox/pause_test.go +++ b/pkg/sandbox/pause_test.go @@ -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", diff --git a/pkg/sandbox/restore_paused.go b/pkg/sandbox/restore_paused.go index c1d8756..772c1d5 100644 --- a/pkg/sandbox/restore_paused.go +++ b/pkg/sandbox/restore_paused.go @@ -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 {