1
0
forked from wrenn/wrenn
Reviewed-on: wrenn/wrenn#57
Co-authored-by: pptx704 <rafeed@omukk.dev>
Co-committed-by: pptx704 <rafeed@omukk.dev>
This commit is contained in:
2026-07-11 13:22:53 +00:00
committed by Rafeed M. Bhuiyan
parent af69ed3442
commit e824b97894
72 changed files with 2531 additions and 429 deletions

3
.gitignore vendored
View File

@ -56,3 +56,6 @@ internal/dashboard/static/*
# Added by code-review-graph
.code-review-graph/
.mcp.json
# tokensave
.tokensave/

View File

@ -83,7 +83,7 @@ It can optionally implement any of these hook interfaces (the OSS server type-as
**Auth helpers for extensions** (`pkg/auth/session/middleware/`, re-exported via `cpextension`): `RequireSession(sctx)`, `RequireSessionOrAPIKey(sctx)`, `RequireAdmin(sctx)`, `RequireCSRF()`, `IssueSession(w, r, sctx, userID, teamID)`, `ClearSessionCookies(w, r)`. Cookie/header names are exported as `SessionCookieName`, `CSRFCookieName`, `CSRFHeaderName`. OSS handlers (`internal/api/middleware_session.go`) are thin shims over this package — single source of truth.
**pkg/ vs internal/ decision rule:** A package belongs in `pkg/` only if the cloud repo needs to import it directly. Everything else stays in `internal/`. New OSS services (e.g. email, notifications) go in `internal/` — the cloud repo accesses them through `ServerContext`, not by importing the package. Do not put a service in `pkg/` just because the cloud repo uses it.
**pkg/ vs internal/ decision rule:** A package belongs in `pkg/` only if an external module needs to import it directly — the cloud repo, or standalone consumers of the sandbox runtime (e.g. the `wr` CLI, which embeds the host agent's orchestration packages to run sandboxes without a control plane). Everything else stays in `internal/`. New OSS services (e.g. email, notifications) go in `internal/` — the cloud repo accesses them through `ServerContext`, not by importing the package. Do not put a service in `pkg/` just because the cloud repo uses it.
Startup (`cmd/control-plane/main.go`) is a thin wrapper: `cpserver.Run(cpserver.WithVersion(...))`. All 20 initialization steps live in `pkg/cpserver/run.go`: config → pgxpool → `db.Queries` → Redis → mTLS CA → host client pool → scheduler → OAuth → channels → audit logger → `api.New()` → background workers → HTTP server. Everything flows through constructor injection.
@ -95,18 +95,20 @@ Startup (`cmd/control-plane/main.go`) is a thin wrapper: `cpserver.Run(cpserver.
### Host Agent
**Packages:** `internal/hostagent/`, `internal/sandbox/`, `internal/vm/`, `internal/network/`, `internal/devicemapper/`, `internal/envdclient/`, `internal/snapshot/`
**Packages:** `internal/hostagent/` (CP-facing: RPC server, registration, heartbeat, mTLS, proxy), plus the public sandbox runtime importable by external consumers such as the `wr` CLI: `pkg/sandbox/`, `pkg/vm/`, `pkg/network/`, `pkg/devicemapper/`, `pkg/envdclient/`, `pkg/layout/`, `pkg/models/`. `internal/snapshot/` stays internal (only used by `pkg/sandbox`, which may import it intra-module).
**Standalone/library use:** `pkg/sandbox` exposes the full startup ritual (`CheckPrivileges`, `EnsureIPForward`, `Setup(SetupOptions)`) so a CLI can wire a `sandbox.Manager` without the RPC server. A library consumer that re-attaches to sandboxes created by earlier processes must set `SetupOptions.SkipCleanup` — the default stale-cleanup kills every cloud-hypervisor process on the host. Running sandboxes persist re-attach state to `{wrennDir}/sandboxes/{id}/sandbox.json`; `Manager.RestoreRunningSandboxes()` (call before `RestorePausedSandboxes()`) re-attaches live VMs across processes. Network slots are claimed cross-process-safely via O_EXCL files in `{wrennDir}/slots/`.
**Production deployment:** `make setup-host` (→ `scripts/setup-host.sh`) prepares the host: creates the `wrenn` system user, sets Linux capabilities (setcap) on wrenn-agent and all child binaries (iptables, losetup, dmsetup, etc.), installs an apt hook to restore capabilities after package updates, configures udev rules for `/dev/net/tun`, and loads required kernel modules. No sudo grants — all privilege is via capabilities. `make install` then copies the binaries to `/usr/local/bin` and installs the systemd units from `deploy/systemd/`.
Startup (`cmd/host-agent/main.go`) wires: root/capabilities check → enable IP forwarding → clean up stale dm devices → `sandbox.Manager` (containing `vm.Manager` + `network.SlotAllocator` + `devicemapper.LoopRegistry`) → `hostagent.Server` (Connect RPC handler) → HTTP server.
- **RPC Server** (`internal/hostagent/server.go`): implements `hostagentv1connect.HostAgentServiceHandler`. Thin wrapper — every method delegates to `sandbox.Manager`. Maps Connect error codes on return.
- **Sandbox Manager** (`internal/sandbox/manager.go`): the core orchestration layer. Maintains in-memory state in `boxes map[string]*sandboxState` (protected by `sync.RWMutex`). Each `sandboxState` holds a `models.Sandbox`, a `*network.Slot`, and an `*envdclient.Client`. Runs a TTL reaper (every 10s) that auto-destroys timed-out sandboxes.
- **VM Manager** (`internal/vm/manager.go`, `ch.go`, `config.go`): manages Cloud Hypervisor processes. Uses raw HTTP API over Unix socket (`/tmp/ch-{sandboxID}.sock`). Launches Cloud Hypervisor via `unshare -m` + `ip netns exec` with `--api-socket path=...`. Configures and boots VM via `PUT /vm.create` + `PUT /vm.boot`. Snapshot restore uses `--restore source_url=file://...`.
- **Network** (`internal/network/setup.go`, `allocator.go`): per-sandbox network namespace with veth pair + TAP device. See Networking section below.
- **Device Mapper** (`internal/devicemapper/devicemapper.go`): CoW rootfs via device-mapper snapshots. Shared read-only loop devices per base template (refcounted `LoopRegistry`), per-sandbox sparse CoW files, dm-snapshot create/restore/remove/flatten operations.
- **envd Client** (`internal/envdclient/client.go`, `health.go`): dual interface to the guest agent. Connect RPC for streaming process exec (`process.Start()` bidirectional stream). Plain HTTP for file operations (POST/GET `/files?path=...&username=root`). Health check polls `GET /health` every 100ms until ready (30s timeout).
- **Sandbox Manager** (`pkg/sandbox/manager.go`): the core orchestration layer. Maintains in-memory state in `boxes map[string]*sandboxState` (protected by `sync.RWMutex`). Each `sandboxState` holds a `models.Sandbox`, a `*network.Slot`, and an `*envdclient.Client`. Runs a TTL reaper (every 10s) that auto-destroys timed-out sandboxes.
- **VM Manager** (`pkg/vm/manager.go`, `ch.go`, `config.go`): manages Cloud Hypervisor processes. Uses raw HTTP API over Unix socket (`/tmp/ch-{sandboxID}.sock`). Launches Cloud Hypervisor via `unshare -m` + `ip netns exec` with `--api-socket path=...`. Configures and boots VM via `PUT /vm.create` + `PUT /vm.boot`. Snapshot restore uses `--restore source_url=file://...`. `Reattach()` adopts a still-live CH process started by an earlier process (exit detection via kill-0 polling).
- **Network** (`pkg/network/setup.go`, `allocator.go`): per-sandbox network namespace with veth pair + TAP device. Slot indices claimed cross-process via O_EXCL files. See Networking section below.
- **Device Mapper** (`pkg/devicemapper/devicemapper.go`): CoW rootfs via device-mapper snapshots. Shared read-only loop devices per base template (refcounted `LoopRegistry`), per-sandbox sparse CoW files, dm-snapshot create/restore/remove/flatten/reattach operations.
- **envd Client** (`pkg/envdclient/client.go`, `health.go`): dual interface to the guest agent. Connect RPC for streaming process exec (`process.Start()` bidirectional stream). Plain HTTP for file operations (POST/GET `/files?path=...&username=root`). Health check polls `GET /health` every 100ms until ready (30s timeout).
### envd (Guest Agent)
@ -156,7 +158,7 @@ veth-{idx} ←──── veth pair ────→ eth0
- **Outbound NAT**: guest (169.254.0.21) → SNAT to vpeerIP inside namespace → MASQUERADE on host to default interface
- **Inbound NAT**: host traffic to 10.11.0.{idx} → DNAT to 169.254.0.21 inside namespace
- IP forwarding enabled inside each namespace
- All details in `internal/network/setup.go`
- All details in `pkg/network/setup.go`
### Sandbox State Machine
```

View File

@ -115,7 +115,7 @@ vet:
go vet ./...
test:
go test -race -v ./internal/...
go test -race -v ./internal/... ./pkg/...
cd envd-rs && cargo test
test-integration:

View File

@ -1 +1 @@
0.3.0
0.3.1

View File

@ -1 +1 @@
0.3.0
0.3.1

View File

@ -1,32 +1,26 @@
package main
import (
"bufio"
"context"
"crypto/tls"
"flag"
"fmt"
"log/slog"
"net/http"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"sync"
"syscall"
"time"
"github.com/joho/godotenv"
"git.omukk.dev/wrenn/wrenn/internal/devicemapper"
"git.omukk.dev/wrenn/wrenn/internal/hostagent"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/internal/network"
"git.omukk.dev/wrenn/wrenn/internal/sandbox"
"git.omukk.dev/wrenn/wrenn/internal/vm"
"git.omukk.dev/wrenn/wrenn/pkg/auth"
"git.omukk.dev/wrenn/wrenn/pkg/logging"
"git.omukk.dev/wrenn/wrenn/pkg/network"
"git.omukk.dev/wrenn/wrenn/pkg/sandbox"
"git.omukk.dev/wrenn/wrenn/proto/hostagent/gen/hostagentv1connect"
)
@ -48,29 +42,26 @@ func main() {
cleanupLog := logging.Setup(filepath.Join(rootDir, "logs"), "host-agent")
defer cleanupLog()
if err := checkPrivileges(); err != nil {
if err := sandbox.CheckPrivileges(); err != nil {
slog.Error("insufficient privileges", "error", err)
os.Exit(1)
}
// Enable IP forwarding (required for NAT). The write may fail if running
// as non-root without DAC_OVERRIDE on this path — that's OK if the systemd
// unit's ExecStartPre already set it. We verify the value regardless.
if err := os.WriteFile("/proc/sys/net/ipv4/ip_forward", []byte("1"), 0644); err != nil {
slog.Warn("failed to enable ip_forward (may have been set by systemd unit)", "error", err)
}
if b, err := os.ReadFile("/proc/sys/net/ipv4/ip_forward"); err != nil || strings.TrimSpace(string(b)) != "1" {
// unit's ExecStartPre already set it.
if err := sandbox.EnsureIPForward(); err != nil {
slog.Error("ip_forward is not enabled — sandbox networking will be broken", "error", err)
os.Exit(1)
}
// Clean up stale resources from a previous crash. Order matters:
// kill stale CH processes first — they hold dm-snapshot devices open and
// would otherwise cause "Device or resource busy" on dmsetup remove.
vm.CleanupStaleProcesses()
devicemapper.CleanupStaleDevices()
devicemapper.LogLoopState()
network.CleanupStaleNamespaces()
// Install the egress guard: block capsule traffic to private ranges from
// leaving the public NIC (which would leak private-destined packets onto the
// upstream network and read as a port scan) and deny cross-capsule reach.
if err := network.EnsureEgressGuard(); err != nil {
slog.Error("failed to install network egress guard", "error", err)
os.Exit(1)
}
listenAddr := envOrDefault("WRENN_HOST_LISTEN_ADDR", ":50051")
cpURL := os.Getenv("WRENN_CP_URL")
@ -116,37 +107,27 @@ func main() {
slog.Info("using custom rootfs size", "size_mb", defaultRootfsSizeMB)
}
// Expand base images to the configured disk size (sparse, no extra physical
// disk). This ensures dm-snapshot sandboxes see the full size from boot.
if err := sandbox.EnsureImageSizes(rootDir, defaultRootfsSizeMB); err != nil {
slog.Error("failed to expand base images", "error", err)
os.Exit(1)
}
// Resolve latest kernel version.
kernelPath, kernelVersion, err := layout.LatestKernel(rootDir)
// Run the startup ritual: stale-resource cleanup, base image expansion,
// kernel resolution, cloud-hypervisor detection, orphan pause-dir GC.
env, err := sandbox.Setup(sandbox.SetupOptions{
WrennDir: rootDir,
CHBin: envOrDefault("WRENN_CH_BIN", sandbox.DefaultCHBin),
DefaultRootfsSizeMB: defaultRootfsSizeMB,
})
if err != nil {
slog.Error("failed to find kernel", "error", err)
slog.Error("host setup failed", "error", err)
os.Exit(1)
}
slog.Info("resolved kernel", "version", kernelVersion, "path", kernelPath)
// Detect cloud-hypervisor version.
chBin := envOrDefault("WRENN_CH_BIN", "/usr/local/bin/cloud-hypervisor")
chVersion, err := sandbox.DetectCHVersion(chBin)
if err != nil {
slog.Error("failed to detect cloud-hypervisor version", "error", err)
os.Exit(1)
}
slog.Info("resolved cloud-hypervisor", "version", chVersion, "path", chBin)
slog.Info("resolved kernel", "version", env.KernelVersion, "path", env.KernelPath)
slog.Info("resolved cloud-hypervisor", "version", env.CHVersion, "path", env.CHBin)
cfg := sandbox.Config{
WrennDir: rootDir,
DefaultRootfsSizeMB: defaultRootfsSizeMB,
KernelPath: kernelPath,
KernelVersion: kernelVersion,
VMMBin: chBin,
VMMVersion: chVersion,
KernelPath: env.KernelPath,
KernelVersion: env.KernelVersion,
VMMBin: env.CHBin,
VMMVersion: env.CHVersion,
AgentVersion: version,
ProxyDomain: envOrDefault("WRENN_PROXY_DOMAIN", "wrenn.dev"),
@ -157,11 +138,6 @@ func main() {
DiskFloorBps: envUint64("WRENN_DISK_FLOOR_BPS"),
}
// Remove any *.staging-* / *.trash-* directories left behind by a
// previous Pause that crashed before completing the atomic swap. Must
// run before any Resume so we don't race with a sandbox restoration.
sandbox.CleanupOrphanPauseDirs(rootDir)
mgr := sandbox.New(cfg)
// Set up lifecycle event callback sender so autonomous events
@ -169,12 +145,15 @@ func main() {
cb := hostagent.NewCallbackSender(cpURL, credsFile, creds.HostID)
mgr.SetEventSender(hostagent.NewEventSender(cb))
// Restore paused sandboxes from disk so ListSandboxes reports them as
// 'paused' immediately. Without this, the CP's HostMonitor would mark
// every paused-on-disk sandbox 'stopped' via the missing→stopped
// reconcile path on the first ListSandboxes after agent restart.
// Must run before HTTP server starts serving (an early Create would
// race the slot reservation).
// Sweep stale running-state files first (Setup's stale cleanup killed
// every CH process, so nothing can actually be re-attached), then restore
// paused sandboxes from disk so ListSandboxes reports them as 'paused'
// immediately. Without the latter, the CP's HostMonitor would mark every
// paused-on-disk sandbox 'stopped' via the missing→stopped reconcile path
// on the first ListSandboxes after agent restart. Must run before the
// HTTP server starts serving (an early Create would race the slot
// reservation).
mgr.RestoreRunningSandboxes()
mgr.RestorePausedSandboxes()
mgr.StartTTLReaper(ctx)
@ -361,63 +340,3 @@ func envUint64(key string) uint64 {
}
return n
}
// checkPrivileges verifies the process has the required Linux capabilities.
// Always reads CapEff — even for root — because a root process inside a
// restricted container (e.g. docker --cap-drop=all) may not have all caps.
func checkPrivileges() error {
capEff, err := readEffectiveCaps()
if err != nil {
return fmt.Errorf("read capabilities: %w", err)
}
// All capabilities required by the host agent at runtime.
required := []struct {
bit uint
name string
}{
{1, "CAP_DAC_OVERRIDE"}, // /dev/loop*, /dev/mapper/*, /dev/net/tun
{5, "CAP_KILL"}, // SIGTERM/SIGKILL to cloud-hypervisor processes
{12, "CAP_NET_ADMIN"}, // netlink, iptables, routing, TAP/veth
{13, "CAP_NET_RAW"}, // raw sockets (iptables)
{19, "CAP_SYS_PTRACE"}, // reading /proc/self/ns/net (netns.Get)
{21, "CAP_SYS_ADMIN"}, // netns, mount ns, losetup, dmsetup
{27, "CAP_MKNOD"}, // device-mapper node creation
}
var missing []string
for _, cap := range required {
if capEff&(1<<cap.bit) == 0 {
missing = append(missing, cap.name)
}
}
if len(missing) > 0 {
return fmt.Errorf("missing capabilities: %s — run as root or apply setcap to the binary",
strings.Join(missing, ", "))
}
return nil
}
// readEffectiveCaps parses the CapEff bitmask from /proc/self/status.
func readEffectiveCaps() (uint64, error) {
f, err := os.Open("/proc/self/status")
if err != nil {
return 0, err
}
defer f.Close()
scanner := bufio.NewScanner(f)
for scanner.Scan() {
line := scanner.Text()
if hexStr, ok := strings.CutPrefix(line, "CapEff:"); ok {
return strconv.ParseUint(strings.TrimSpace(hexStr), 16, 64)
}
}
if err := scanner.Err(); err != nil {
return 0, fmt.Errorf("read /proc/self/status: %w", err)
}
return 0, fmt.Errorf("CapEff not found in /proc/self/status")
}

View File

@ -0,0 +1,25 @@
-- +goose Up
-- Bump the default sandbox specs from 1 vCPU / 512 MB to 2 vCPUs / 2048 MB.
-- Applies to any row inserted without explicit vcpus/memory_mb across the
-- sandboxes, templates, and template_builds tables.
ALTER TABLE sandboxes ALTER COLUMN vcpus SET DEFAULT 2;
ALTER TABLE sandboxes ALTER COLUMN memory_mb SET DEFAULT 2048;
ALTER TABLE templates ALTER COLUMN vcpus SET DEFAULT 2;
ALTER TABLE templates ALTER COLUMN memory_mb SET DEFAULT 2048;
ALTER TABLE template_builds ALTER COLUMN vcpus SET DEFAULT 2;
ALTER TABLE template_builds ALTER COLUMN memory_mb SET DEFAULT 2048;
-- +goose Down
ALTER TABLE sandboxes ALTER COLUMN vcpus SET DEFAULT 1;
ALTER TABLE sandboxes ALTER COLUMN memory_mb SET DEFAULT 512;
ALTER TABLE templates ALTER COLUMN vcpus SET DEFAULT 1;
ALTER TABLE templates ALTER COLUMN memory_mb SET DEFAULT 512;
ALTER TABLE template_builds ALTER COLUMN vcpus SET DEFAULT 1;
ALTER TABLE template_builds ALTER COLUMN memory_mb SET DEFAULT 512;

View File

@ -28,3 +28,6 @@ DELETE FROM team_api_keys WHERE team_id = $1;
-- name: DeleteAPIKeysByCreator :exec
DELETE FROM team_api_keys WHERE created_by = $1;
-- name: DeleteAPIKeysByTeamAndCreator :exec
DELETE FROM team_api_keys WHERE team_id = $1 AND created_by = $2;

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",

View File

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

View File

@ -71,16 +71,22 @@ pub async fn post_init(
// back in the source memory-ranges file as the host re-restored them
// lazily. Reset the flags so the next POST /memory/preload kicks off
// a new loader instead of returning the stale "already-done".
//
// The generation bump + reset happen under the error mutex — the same
// critical section a loader thread publishes its result in. A loader
// from the PREVIOUS lifecycle can be frozen mid-walk by the pause and
// thaw here; the bump makes it stop and discard its result instead of
// storing a stale done=true that the host would trust for the next
// snapshot. cancel is deliberately cleared (not set): the stale thread
// stops on generation mismatch, and the new run must start uncancelled.
use std::sync::atomic::Ordering;
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);
*state.mem_preload_error.lock().unwrap() = None;
{
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_started.store(false, Ordering::SeqCst);
state.reset_preload_run(&mut err);
}
if let Some(ref port_sub) = state.port_subsystem {
tracing::info!("lifecycle changed, restarting port subsystem");
@ -191,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
@ -204,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
@ -215,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
@ -254,6 +260,14 @@ async fn validate_init_access_token(state: &AppState, request_token: &str) -> Re
}
async fn setup_hyperloop(address: &str, env_vars: &dashmap::DashMap<String, String>) {
// Reject anything that is not a bare IP address before it reaches
// /etc/hosts. Without this, a newline in `address` would inject arbitrary
// additional host entries into the file.
if address.parse::<std::net::IpAddr>().is_err() {
tracing::error!(%address, "hyperloop address is not a valid IP; skipping /etc/hosts entry");
return;
}
// Write to /etc/hosts: events.wrenn.local → address
let entry = format!("{address} events.wrenn.local\n");

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,21 +53,64 @@ 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();
// 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
}
};
if we_start {
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
// materialisation. Lifecycle bump on the next /init clears the flags
// for a fresh run after a restore.
// for a fresh run after a restore; this thread stops (and refuses to
// publish) once the generation it captured is no longer current — a
// pause can freeze it mid-walk and thaw it in the NEXT lifecycle.
std::thread::spawn(move || {
let started = Instant::now();
match preload_blocking(&state_clone) {
let outcome = preload_blocking(&state_clone, generation);
// Publish under the error mutex so a concurrent /init bump can't
// interleave: either this store lands before the bump (and /init
// resets it), or the generation no longer matches and the result
// is discarded.
let mut err = state_clone.mem_preload_error.lock().unwrap();
if state_clone.mem_preload_generation.load(Ordering::SeqCst) != generation {
tracing::info!(
generation,
"memory preload result discarded: stale lifecycle"
);
return;
}
match outcome {
Ok((source, regions, pages, bytes)) => {
let elapsed = started.elapsed().as_secs_f64();
state_clone
@ -76,7 +122,7 @@ pub async fn post_memory_preload(State(state): State<Arc<AppState>>) -> impl Int
.mem_preload_elapsed_us
.store((elapsed * 1_000_000.0) as u64, Ordering::SeqCst);
set_source(&state_clone, source);
*state_clone.mem_preload_error.lock().unwrap() = None;
*err = None;
state_clone.mem_preload_done.store(true, Ordering::SeqCst);
tracing::info!(
regions,
@ -92,7 +138,7 @@ pub async fn post_memory_preload(State(state): State<Arc<AppState>>) -> impl Int
state_clone
.mem_preload_elapsed_us
.store((elapsed * 1_000_000.0) as u64, Ordering::SeqCst);
*state_clone.mem_preload_error.lock().unwrap() = Some(e.clone());
*err = Some(e.clone());
state_clone.mem_preload_done.store(true, Ordering::SeqCst);
tracing::warn!(error = %e, "memory preload failed");
}
@ -159,7 +205,10 @@ fn get_source(state: &AppState) -> &'static str {
}
}
fn preload_blocking(state: &AppState) -> Result<(&'static str, u64, u64, u64), String> {
fn preload_blocking(
state: &AppState,
generation: u64,
) -> Result<(&'static str, u64, u64, u64), String> {
let ranges = parse_system_ram_ranges().map_err(|e| format!("iomem: {e}"))?;
if ranges.is_empty() {
return Err("no System RAM ranges found in /proc/iomem".into());
@ -168,7 +217,7 @@ fn preload_blocking(state: &AppState) -> Result<(&'static str, u64, u64, u64), S
let mut pages: u64 = 0;
let mut bytes: u64 = 0;
match preload_via_devmem(&ranges, state, &mut pages, &mut bytes) {
match preload_via_devmem(&ranges, state, generation, &mut pages, &mut bytes) {
Ok(()) => Ok(("/dev/mem", ranges.len() as u64, pages, bytes)),
Err(devmem_err) => {
tracing::warn!(
@ -177,13 +226,22 @@ fn preload_blocking(state: &AppState) -> Result<(&'static str, u64, u64, u64), S
);
pages = 0;
bytes = 0;
preload_via_kcore(state, &mut pages, &mut bytes)
preload_via_kcore(state, generation, &mut pages, &mut bytes)
.map_err(|e| format!("/dev/mem: {devmem_err}; /proc/kcore: {e}"))?;
Ok(("/proc/kcore", ranges.len() as u64, pages, bytes))
}
}
}
// should_stop reports whether the walk must abort: an explicit cancel, or the
// lifecycle moved on (a pause froze this thread and a resume /init bumped the
// generation — continuing would fault pages nobody waits for and eventually
// publish a stale result).
fn should_stop(state: &AppState, generation: u64) -> bool {
state.mem_preload_cancel.load(Ordering::SeqCst)
|| state.mem_preload_generation.load(Ordering::SeqCst) != generation
}
fn parse_system_ram_ranges() -> std::io::Result<Vec<(u64, u64)>> {
let data = fs::read_to_string("/proc/iomem")?;
let mut out = Vec::new();
@ -212,6 +270,7 @@ fn parse_system_ram_ranges() -> std::io::Result<Vec<(u64, u64)>> {
fn preload_via_devmem(
ranges: &[(u64, u64)],
state: &AppState,
generation: u64,
pages: &mut u64,
bytes: &mut u64,
) -> std::io::Result<()> {
@ -220,7 +279,7 @@ fn preload_via_devmem(
for (start, end) in ranges {
let mut off = *start;
while off < *end {
if state.mem_preload_cancel.load(Ordering::SeqCst) {
if should_stop(state, generation) {
return Ok(());
}
f.read_at(&mut buf, off)?;
@ -238,13 +297,23 @@ fn preload_via_devmem(
Ok(())
}
// Read /proc/kcore's direct-map segment to materialise physical RAM. The
// direct map's PT_LOAD covers the kernel's *maximum* possible direct-map
// region (64TB on x86_64), not just the present physical RAM — iterating the
// whole segment would loop for billions of pages. Bound the walk to the sum
// of System RAM ranges from /proc/iomem; sequential reads through the
// segment touch consecutive physical pages 1:1, which is what we need.
fn preload_via_kcore(state: &AppState, pages: &mut u64, bytes: &mut u64) -> std::io::Result<()> {
// Read physical RAM through /proc/kcore's per-range direct-map segments.
//
// The kernel emits one KCORE_RAM PT_LOAD *per contiguous System RAM range*
// (walk_system_ram_range), each with vaddr = direct_map_base + phys_start and
// file_size = range size — there is NO single segment covering all of RAM.
// Segments must therefore be matched range-by-range against /proc/iomem;
// picking "a big kernel-space segment" instead lands on the vmalloc or
// vmemmap region, which the kernel zero-fills for unmapped addresses without
// ever touching a physical page — the reads "succeed" instantly and nothing
// is materialised. Matching failure is a hard error for the same reason:
// reading the wrong segment is indistinguishable from success.
fn preload_via_kcore(
state: &AppState,
generation: u64,
pages: &mut u64,
bytes: &mut u64,
) -> std::io::Result<()> {
let ram_ranges = parse_system_ram_ranges()?;
if ram_ranges.is_empty() {
return Err(std::io::Error::new(
@ -252,7 +321,6 @@ fn preload_via_kcore(state: &AppState, pages: &mut u64, bytes: &mut u64) -> std:
"no System RAM ranges to bound kcore walk",
));
}
let total_ram_bytes: u64 = ram_ranges.iter().map(|(s, e)| e - s).sum();
let mut f = fs::File::open("/proc/kcore")?;
let segments = parse_kcore_pt_load(&mut f)?;
@ -263,55 +331,253 @@ fn preload_via_kcore(state: &AppState, pages: &mut u64, bytes: &mut u64) -> std:
));
}
// Pick the direct map: highest-vaddr kernel-space segment large enough
// to plausibly cover RAM. KASLR randomises the base, so don't hardcode it.
// Kernel virtual addresses start at 0xffff800000000000 on x86_64; vmalloc
// / modules sit above the direct map and are usually smaller.
const KERNEL_SPACE_MIN: u64 = 0xffff_8000_0000_0000;
let direct_map = segments
.iter()
.filter(|s| s.vaddr >= KERNEL_SPACE_MIN && s.file_size >= total_ram_bytes)
.min_by_key(|s| s.vaddr)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"no PT_LOAD segment large enough for {} bytes of RAM in /proc/kcore",
total_ram_bytes
),
)
})?;
let ram_segments = match_ram_segments(&ram_ranges, &segments)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
let mut buf = [0u8; 1];
let read_bytes = total_ram_bytes.min(direct_map.file_size);
let start = direct_map.file_offset;
let end = start.saturating_add(read_bytes);
let mut off = start;
while off < end {
if state.mem_preload_cancel.load(Ordering::SeqCst) {
return Ok(());
}
// Reads into MMIO holes within the direct map can fail; ignore so the
// loop keeps making progress over the present RAM ranges either side.
if f.read_at(&mut buf, off).is_ok() {
*pages += 1;
*bytes += PAGE_SIZE;
if *pages % 256 == 0 {
state.mem_preload_pages.store(*pages, Ordering::SeqCst);
state.mem_preload_bytes.store(*bytes, Ordering::SeqCst);
let mut read_errors: u64 = 0;
for seg in &ram_segments {
let end = seg.file_offset.saturating_add(seg.len);
let mut off = seg.file_offset;
while off < end {
if should_stop(state, generation) {
return Ok(());
}
match f.read_at(&mut buf, off) {
Ok(_) => {
*pages += 1;
*bytes += PAGE_SIZE;
if *pages % 256 == 0 {
state.mem_preload_pages.store(*pages, Ordering::SeqCst);
state.mem_preload_bytes.store(*bytes, Ordering::SeqCst);
}
}
Err(_) => read_errors += 1,
}
off = off.saturating_add(PAGE_SIZE);
}
off = off.saturating_add(PAGE_SIZE);
}
// A few failed reads are tolerable (hwpoison, offline pages); a walk where
// nothing was read means the segments were wrong — report it rather than
// letting the host trust an empty "done".
if *pages == 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("kcore walk read no pages ({read_errors} read errors)"),
));
}
if read_errors > 0 {
tracing::warn!(
read_errors,
pages = *pages,
"kcore preload skipped unreadable pages"
);
}
Ok(())
}
// A resolved (file_offset, length) window in /proc/kcore corresponding to one
// System RAM range.
struct RamSegment {
file_offset: u64,
len: u64,
}
// 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 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 &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 {
if let Some(out) = try_base(*base) {
return Ok(out);
}
}
Err(format!(
"no consistent direct-map base for {} RAM ranges across {} PT_LOAD segments \
({} candidate bases tried)",
clamped.len(),
segments.len(),
candidates.len(),
))
}
struct KcoreSegment {
file_offset: u64,
file_size: u64,
vaddr: u64,
}
#[cfg(test)]
mod tests {
use super::*;
const BASE: u64 = 0xffff_8880_0000_0000; // typical direct-map base, no KASLR
const GIB: u64 = 1 << 30;
const TIB: u64 = 1 << 40;
fn seg(vaddr: u64, file_size: u64, file_offset: u64) -> KcoreSegment {
KcoreSegment {
file_offset,
file_size,
vaddr,
}
}
// Typical CH guest: low RAM hole below 1MB, main range, plus the huge
// vmalloc/vmemmap segments that the old heuristic used to pick.
fn typical_segments() -> Vec<KcoreSegment> {
vec![
seg(0xffff_c900_0000_0000, 32 * TIB, 0x1000), // vmalloc
seg(0xffff_ea00_0000_0000, TIB, 0x2000), // vmemmap
seg(BASE + 0x1000, 0x9e000, 0x10000), // RAM 0x1000-0x9efff
seg(BASE + 0x100000, 2 * GIB - 0x100000, 0x20000), // RAM 1MB-2GB
]
}
// 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, 0x9fc00), (0x100000, 2 * GIB)]
}
#[test]
fn matches_all_ranges_and_skips_vmalloc() {
let out = match_ram_segments(&typical_ranges(), &typical_segments()).unwrap();
assert_eq!(out.len(), 2);
assert_eq!(out[0].file_offset, 0x10000);
assert_eq!(out[0].len, 0x9e000);
assert_eq!(out[1].file_offset, 0x20000);
assert_eq!(out[1].len, 2 * GIB - 0x100000);
}
#[test]
fn kaslr_base_is_derived_not_assumed() {
let kaslr = BASE + 37 * GIB; // PUD-granular randomisation
let segments = vec![
seg(0xffff_c900_0000_0000, 32 * TIB, 0x1000),
seg(kaslr + 0x1000, 0x9e000, 0x10000),
seg(kaslr + 0x100000, 2 * GIB - 0x100000, 0x20000),
];
let out = match_ram_segments(&typical_ranges(), &segments).unwrap();
assert_eq!(out[1].file_offset, 0x20000);
}
#[test]
fn errors_when_a_range_has_no_segment() {
// Second RAM range's segment missing → no base covers all ranges.
let segments = vec![
seg(0xffff_c900_0000_0000, 32 * TIB, 0x1000),
seg(BASE + 0x1000, 0x9e000, 0x10000),
];
assert!(match_ram_segments(&typical_ranges(), &segments).is_err());
}
#[test]
fn errors_instead_of_falling_back_to_vmalloc() {
// Only huge kernel segments present (the exact shape that fooled the
// old `file_size >= total_ram` heuristic).
let segments = vec![
seg(0xffff_c900_0000_0000, 32 * TIB, 0x1000),
seg(0xffff_ea00_0000_0000, TIB, 0x2000),
];
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
// must be rejected; only the true base matches both ranges.
let ranges = vec![(0x0, GIB), (2 * GIB, 3 * GIB)];
let segments = vec![seg(BASE, GIB, 0x10000), seg(BASE + 2 * GIB, GIB, 0x20000)];
let out = match_ram_segments(&ranges, &segments).unwrap();
assert_eq!(out[0].file_offset, 0x10000);
assert_eq!(out[1].file_offset, 0x20000);
}
}
fn parse_kcore_pt_load(f: &mut fs::File) -> std::io::Result<Vec<KcoreSegment>> {
let mut hdr = [0u8; 64];
f.seek(SeekFrom::Start(0))?;

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;
}

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}"));

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 {

View File

@ -1,6 +1,7 @@
use std::collections::VecDeque;
use std::io::Read;
use std::os::unix::io::AsRawFd;
use std::os::fd::{AsFd, OwnedFd};
use std::os::unix::io::{AsRawFd, RawFd};
use std::os::unix::process::CommandExt;
use std::process::Stdio;
use std::sync::{Arc, Mutex};
@ -10,6 +11,8 @@ use connectrpc::{ConnectError, ErrorCode};
use nix::pty::{Winsize, openpty};
use nix::sys::signal::{self, Signal};
use nix::unistd::Pid;
use tokio::io::Interest;
use tokio::io::unix::AsyncFd;
use tokio::sync::broadcast;
use crate::rpc::pb::process::*;
@ -32,6 +35,14 @@ const COALESCE_CAP: usize = 64 * 1024;
// buffer can never grow without bound for a chatty long-running process.
const OUTPUT_LOG_CAPACITY: usize = 256 * 1024;
// Bound on how long one input write may wait for the target process to drain
// its buffer. The stdin pipe / pty master is O_NONBLOCK, so a full buffer
// parks the writing task (never an OS thread); within this window a
// briefly-full buffer recovers transparently, past it the write fails with
// ResourceExhausted instead of hanging forever on a process that stopped
// reading its input.
const INPUT_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Clone)]
pub enum DataEvent {
Stdout(Vec<u8>),
@ -136,6 +147,24 @@ where
}
publish(std::mem::take(&mut acc));
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
// The pty master is O_NONBLOCK for the input write path and
// shares its file description with this reader. Wait for
// readability and retry; on POLLHUP/POLLERR the retried read
// returns 0/EIO and ends the loop.
let mut pfd = libc::pollfd {
fd,
events: libc::POLLIN,
revents: 0,
};
if unsafe { libc::poll(&mut pfd, 1, -1) } < 0 {
let errno = std::io::Error::last_os_error().raw_os_error();
if errno != Some(libc::EINTR) {
break;
}
}
}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(_) => break,
}
}
@ -156,6 +185,10 @@ pub struct ProcessHandle {
stdin: Mutex<Option<std::process::ChildStdin>>,
pty_master: Mutex<Option<std::fs::File>>,
// Serializes input writes so concurrent chunks land in order. A tokio
// mutex: a writer parked on a full buffer holds it across an await, and
// the next writer waits as a suspended task — no OS thread is pinned.
input_gate: tokio::sync::Mutex<()>,
}
impl ProcessHandle {
@ -201,32 +234,41 @@ impl ProcessHandle {
})
}
pub fn write_stdin(&self, data: &[u8]) -> Result<(), ConnectError> {
use std::io::Write;
let mut guard = self.stdin.lock().unwrap();
match guard.as_mut() {
Some(stdin) => stdin.write_all(data).map_err(|e| {
ConnectError::new(ErrorCode::Internal, format!("error writing to stdin: {e}"))
}),
None => Err(ConnectError::new(
ErrorCode::FailedPrecondition,
"stdin not enabled or closed",
)),
}
pub async fn write_stdin(&self, data: &[u8]) -> Result<(), ConnectError> {
let _ordered = self.input_gate.lock().await;
// Dup the fd under the std mutex, then release it before awaiting:
// close_stdin stays responsive while a write is parked, and the fd
// stays valid for this write even if stdin is closed concurrently.
let fd = {
let guard = self.stdin.lock().unwrap();
match guard.as_ref() {
Some(stdin) => dup_writer_fd(stdin, "stdin")?,
None => {
return Err(ConnectError::new(
ErrorCode::FailedPrecondition,
"stdin not enabled or closed",
));
}
}
};
write_nonblocking(fd, data, "stdin").await
}
pub fn write_pty(&self, data: &[u8]) -> Result<(), ConnectError> {
use std::io::Write;
let mut guard = self.pty_master.lock().unwrap();
match guard.as_mut() {
Some(master) => master.write_all(data).map_err(|e| {
ConnectError::new(ErrorCode::Internal, format!("error writing to pty: {e}"))
}),
None => Err(ConnectError::new(
ErrorCode::FailedPrecondition,
"pty not assigned to process",
)),
}
pub async fn write_pty(&self, data: &[u8]) -> Result<(), ConnectError> {
let _ordered = self.input_gate.lock().await;
let fd = {
let guard = self.pty_master.lock().unwrap();
match guard.as_ref() {
Some(master) => dup_writer_fd(master, "pty")?,
None => {
return Err(ConnectError::new(
ErrorCode::FailedPrecondition,
"pty not assigned to process",
));
}
}
};
write_nonblocking(fd, data, "pty").await
}
pub fn close_stdin(&self) -> Result<(), ConnectError> {
@ -272,12 +314,109 @@ impl ProcessHandle {
}
}
fn dup_writer_fd(f: &impl AsFd, what: &str) -> Result<OwnedFd, ConnectError> {
f.as_fd()
.try_clone_to_owned()
.map_err(|e| ConnectError::new(ErrorCode::Internal, format!("dup {what} fd: {e}")))
}
/// Write `data` to a non-blocking fd from async context. Small interactive
/// writes complete inline; a full buffer surfaces as WouldBlock and we await
/// writability with a deadline, so a process that stopped reading its input
/// costs a parked task — never a pinned OS thread.
async fn write_nonblocking(fd: OwnedFd, data: &[u8], what: &str) -> Result<(), ConnectError> {
let afd = AsyncFd::with_interest(fd, Interest::WRITABLE)
.map_err(|e| ConnectError::new(ErrorCode::Internal, format!("register {what} fd: {e}")))?;
let deadline = tokio::time::Instant::now() + INPUT_WRITE_TIMEOUT;
let mut written = 0usize;
while written < data.len() {
let rest = &data[written..];
let n = unsafe {
libc::write(
afd.get_ref().as_raw_fd(),
rest.as_ptr() as *const libc::c_void,
rest.len(),
)
};
if n >= 0 {
written += n as usize;
continue;
}
let err = std::io::Error::last_os_error();
match err.kind() {
std::io::ErrorKind::WouldBlock => {
match tokio::time::timeout_at(deadline, afd.writable()).await {
Ok(Ok(mut ready)) => ready.clear_ready(),
Ok(Err(e)) => {
return Err(ConnectError::new(
ErrorCode::Internal,
format!("error waiting for {what} to become writable: {e}"),
));
}
Err(_) => {
return Err(ConnectError::new(
ErrorCode::ResourceExhausted,
format!(
"{what} input buffer full ({written} of {} bytes written): process is not reading its input",
data.len()
),
));
}
}
}
std::io::ErrorKind::Interrupted => {}
_ => {
return Err(ConnectError::new(
ErrorCode::Internal,
format!("error writing to {what}: {err}"),
));
}
}
}
Ok(())
}
fn set_nonblocking(fd: RawFd) -> std::io::Result<()> {
let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
if flags < 0 {
return Err(std::io::Error::last_os_error());
}
if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 {
return Err(std::io::Error::last_os_error());
}
Ok(())
}
pub struct SpawnedProcess {
pub handle: Arc<ProcessHandle>,
pub data_rx: broadcast::Receiver<DataEvent>,
pub end_rx: broadcast::Receiver<EndEvent>,
}
/// Switch the child to the target gid/uid, dropping supplementary groups first
/// and checking every step. Order is load-bearing: supplementary groups must be
/// cleared while still privileged (setgroups after setuid fails), otherwise a
/// non-root child inherits PID 1's root supplementary groups. Checking the
/// return values turns a failed drop into a failed spawn instead of silently
/// executing the command as root.
///
/// Runs inside `pre_exec`, so it must stay async-signal-safe: raw libc calls
/// only, no allocation.
unsafe fn switch_user(uid: libc::uid_t, gid: libc::gid_t) -> std::io::Result<()> {
unsafe {
if libc::setgroups(0, std::ptr::null()) != 0 {
return Err(std::io::Error::last_os_error());
}
if libc::setgid(gid) != 0 {
return Err(std::io::Error::last_os_error());
}
if libc::setuid(uid) != 0 {
return Err(std::io::Error::last_os_error());
}
}
Ok(())
}
pub fn spawn_process(
cmd_str: &str,
args: &[String],
@ -415,8 +554,7 @@ exec {nice_prefix}{target}"#
if slave_raw > 2 {
libc::close(slave_raw);
}
libc::setgid(gid);
libc::setuid(uid);
switch_user(uid, gid)?;
Ok(())
});
}
@ -436,6 +574,19 @@ exec {nice_prefix}{target}"#
let pid = child.id();
let master_file: std::fs::File = master_fd.into();
// O_NONBLOCK is per file description, so it covers the write path and
// the reader clone alike; coalesce_read_loop handles the resulting
// WouldBlock by polling. Without it a write would block an OS thread,
// so treat failure as a failed spawn.
if let Err(e) = set_nonblocking(master_file.as_raw_fd()) {
let mut child = child;
let _ = signal::kill(Pid::from_raw(-(pid as i32)), Signal::SIGKILL);
let _ = child.wait();
return Err(ConnectError::new(
ErrorCode::Internal,
format!("set pty master non-blocking: {e}"),
));
}
let master_clone = master_file.try_clone().unwrap();
let handle = Arc::new(ProcessHandle {
@ -448,6 +599,7 @@ exec {nice_prefix}{target}"#
output_log: Mutex::new(OutputLog::default()),
stdin: Mutex::new(None),
pty_master: Mutex::new(Some(master_file)),
input_gate: tokio::sync::Mutex::new(()),
});
let data_rx = handle.subscribe_data();
@ -515,8 +667,7 @@ exec {nice_prefix}{target}"#
// for free via setsid().
nix::unistd::setpgid(Pid::from_raw(0), Pid::from_raw(0))
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
libc::setgid(gid);
libc::setuid(uid);
switch_user(uid, gid)?;
Ok(())
});
}
@ -530,6 +681,21 @@ exec {nice_prefix}{target}"#
let stdout = child.stdout.take();
let stderr = child.stderr.take();
// Non-blocking stdin is what lets write_stdin fail fast instead of
// pinning an OS thread when the process stops draining its input.
// Only the pipe's write end is affected — the child's read end is a
// separate file description.
if let Some(s) = stdin.as_ref() {
if let Err(e) = set_nonblocking(s.as_raw_fd()) {
let _ = signal::kill(Pid::from_raw(-(pid as i32)), Signal::SIGKILL);
let _ = child.wait();
return Err(ConnectError::new(
ErrorCode::Internal,
format!("set stdin non-blocking: {e}"),
));
}
}
let handle = Arc::new(ProcessHandle {
config,
tag,
@ -540,6 +706,7 @@ exec {nice_prefix}{target}"#
output_log: Mutex::new(OutputLog::default()),
stdin: Mutex::new(stdin),
pty_master: Mutex::new(None),
input_gate: tokio::sync::Mutex::new(()),
});
let data_rx = handle.subscribe_data();
@ -613,3 +780,112 @@ fn current_nice() -> i32 {
prio
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
fn spawn(cmd: &str, enable_stdin: bool, pty: Option<(u16, u16)>) -> SpawnedProcess {
let user = nix::unistd::User::from_uid(nix::unistd::getuid())
.unwrap()
.unwrap();
spawn_process(
cmd,
&[],
&HashMap::new(),
"/tmp",
pty,
enable_stdin,
None,
&user,
&dashmap::DashMap::new(),
)
.unwrap()
}
/// Wait until the accumulated bytes from `recv` contain `needle`.
async fn recv_until_contains(
rx: &mut broadcast::Receiver<DataEvent>,
needle: &[u8],
what: &str,
) -> Vec<u8> {
let mut acc: Vec<u8> = Vec::new();
tokio::time::timeout(Duration::from_secs(10), async {
loop {
match rx.recv().await {
Ok(DataEvent::Stdout(d)) | Ok(DataEvent::Pty(d)) => {
acc.extend_from_slice(&d);
if acc.windows(needle.len()).any(|w| w == needle) {
break;
}
}
Ok(DataEvent::Stderr(_)) => {}
Err(broadcast::error::RecvError::Lagged(_)) => {}
Err(e) => panic!("{what}: data stream closed early: {e}"),
}
}
})
.await
.unwrap_or_else(|_| panic!("{what}: output never contained expected bytes"));
acc
}
#[tokio::test(flavor = "multi_thread")]
async fn write_to_stuck_stdin_errors_instead_of_hanging() {
// sleep never reads stdin: the 64K pipe buffer fills and the write
// must fail with ResourceExhausted after the bounded wait — the old
// write_all here blocked a pool thread forever.
let spawned = spawn("sleep 60", true, None);
let handle = Arc::clone(&spawned.handle);
let big = vec![b'x'; 1024 * 1024];
let writer = tokio::spawn(async move { handle.write_stdin(&big).await });
// While that write is parked, the agent must stay fully responsive:
// a fresh process spawns, runs, and reports its exit.
let mut quick = spawn("true", false, None);
let end = tokio::time::timeout(Duration::from_secs(10), quick.end_rx.recv())
.await
.expect("agent unresponsive while stdin write pending")
.expect("end event");
assert!(end.exited);
let err = tokio::time::timeout(INPUT_WRITE_TIMEOUT + Duration::from_secs(10), writer)
.await
.expect("stdin write hung past its deadline")
.expect("writer task panicked")
.expect_err("write into a full pipe nobody reads must error");
assert!(
matches!(err.code, ErrorCode::ResourceExhausted),
"expected ResourceExhausted, got {:?}",
err.code
);
let _ = spawned.handle.send_signal(Signal::SIGKILL);
}
#[tokio::test(flavor = "multi_thread")]
async fn stdin_write_and_eof_still_work() {
let mut spawned = spawn("cat", true, None);
spawned.handle.write_stdin(b"hello\n").await.unwrap();
recv_until_contains(&mut spawned.data_rx, b"hello\n", "cat stdout").await;
// close_stdin must still deliver EOF with the non-blocking pipe.
spawned.handle.close_stdin().unwrap();
let end = tokio::time::timeout(Duration::from_secs(10), spawned.end_rx.recv())
.await
.expect("cat did not exit after stdin EOF")
.expect("end event");
assert_eq!(end.exit_code, 0);
}
#[tokio::test(flavor = "multi_thread")]
async fn pty_write_and_echo_still_work() {
// O_NONBLOCK on the pty master is shared with the output reader;
// this covers both directions surviving it.
let mut spawned = spawn("cat", false, Some((80, 24)));
spawned.handle.write_pty(b"hello\r").await.unwrap();
recv_until_contains(&mut spawned.data_rx, b"hello", "pty echo").await;
let _ = spawned.handle.send_signal(Signal::SIGKILL);
}
}

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,10 +392,18 @@ impl Process for ProcessServiceImpl {
}
}
fn write_input(handle: &ProcessHandle, input: &ProcessInputView) -> Result<(), ConnectError> {
// Input writes go straight to the fd, which is O_NONBLOCK (set at spawn):
// small interactive writes complete inline on the async worker, and a full
// buffer parks the task awaiting writability with a bounded deadline inside
// write_stdin / write_pty — no blocking-pool thread is ever pinned by a
// process that stopped reading its input.
async fn write_input(
handle: &Arc<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),
Some(process_input::InputView::Pty(d)) => handle.write_pty(d).await,
Some(process_input::InputView::Stdin(d)) => handle.write_stdin(d).await,
None => Ok(()),
}
}

View File

@ -31,6 +31,17 @@ 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
/// frozen mid-walk) cannot store a stale `done=true` for the NEXT
/// lifecycle's preload. Publication happens under `mem_preload_error`'s
/// mutex, which /init also holds while bumping, closing the
/// freeze-between-check-and-store window.
pub mem_preload_generation: AtomicU64,
pub mem_preload_regions: AtomicU64,
pub mem_preload_pages: AtomicU64,
pub mem_preload_bytes: AtomicU64,
@ -66,6 +77,7 @@ impl AppState {
mem_preload_started: AtomicBool::new(false),
mem_preload_done: AtomicBool::new(false),
mem_preload_cancel: AtomicBool::new(false),
mem_preload_generation: AtomicU64::new(0),
mem_preload_regions: AtomicU64::new(0),
mem_preload_pages: AtomicU64::new(0),
mem_preload_bytes: AtomicU64::new(0),
@ -101,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 {

View File

@ -11,7 +11,7 @@
};
let { open, onclose, oncreated, templateSource = 'team' }: Props = $props();
let createForm = $state<CreateCapsuleParams>({ template: 'minimal-ubuntu', vcpus: 1, memory_mb: 512, timeout_sec: 0 });
let createForm = $state<CreateCapsuleParams>({ template: 'minimal-ubuntu', vcpus: 2, memory_mb: 2048, timeout_sec: 0 });
let creating = $state(false);
let createError = $state<string | null>(null);
@ -120,7 +120,7 @@
const creator = templateSource === 'platform' ? createAdminCapsule : createCapsule;
const result = await creator(createForm);
if (result.ok) {
createForm = { template: 'minimal-ubuntu', vcpus: 1, memory_mb: 512, timeout_sec: 0 };
createForm = { template: 'minimal-ubuntu', vcpus: 2, memory_mb: 2048, timeout_sec: 0 };
templateQuery = 'minimal-ubuntu';
onclose();
oncreated?.(result.data);

View File

@ -45,8 +45,8 @@
let createForm = $state({
name: '',
base_template: 'minimal-ubuntu',
vcpus: 1,
memory_mb: 512,
vcpus: 2,
memory_mb: 2048,
recipe: '',
healthcheck: '',
skip_pre_post: false,
@ -152,7 +152,7 @@
if (result.ok) {
showCreate = false;
createForm = { name: '', base_template: 'minimal-ubuntu', vcpus: 1, memory_mb: 512, recipe: '', healthcheck: '', skip_pre_post: false, run_as_root: false, archive: null };
createForm = { name: '', base_template: 'minimal-ubuntu', vcpus: 2, memory_mb: 2048, recipe: '', healthcheck: '', skip_pre_post: false, run_as_root: false, archive: null };
toast.success('Build queued');
goto(`/admin/templates/builds/${result.data.id}`);
} else {
@ -246,7 +246,7 @@
</p>
</div>
<button
onclick={() => { showCreate = true; createError = null; createForm = { name: '', base_template: 'minimal-ubuntu', vcpus: 1, memory_mb: 512, recipe: '', healthcheck: '', skip_pre_post: false, run_as_root: false, archive: null }; }}
onclick={() => { showCreate = true; createError = null; createForm = { name: '', base_template: 'minimal-ubuntu', vcpus: 2, memory_mb: 2048, recipe: '', healthcheck: '', skip_pre_post: false, run_as_root: false, archive: null }; }}
class="group flex items-center gap-2.5 rounded-[var(--radius-button)] bg-[var(--color-accent)] px-5 py-2.5 text-ui font-semibold text-white shadow-sm transition-all duration-200 hover:shadow-[0_0_20px_var(--color-accent-glow-mid)] hover:brightness-115 hover:-translate-y-px active:translate-y-0"
>
<svg width="13" height="13" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round" class="transition-transform duration-200 group-hover:rotate-90"><line x1="12" y1="5" x2="12" y2="19"/><line x1="5" y1="12" x2="19" y2="12"/></svg>
@ -397,7 +397,7 @@
</p>
{#if type === 'templates'}
<button
onclick={() => { showCreate = true; createError = null; createForm = { name: '', base_template: 'minimal-ubuntu', vcpus: 1, memory_mb: 512, recipe: '', healthcheck: '', skip_pre_post: false, run_as_root: false, archive: null }; }}
onclick={() => { showCreate = true; createError = null; createForm = { name: '', base_template: 'minimal-ubuntu', vcpus: 2, memory_mb: 2048, recipe: '', healthcheck: '', skip_pre_post: false, run_as_root: false, archive: null }; }}
class="mt-6 flex items-center gap-2 rounded-[var(--radius-button)] border border-[var(--color-accent)]/30 bg-[var(--color-accent)]/10 px-4 py-2 text-ui font-medium text-[var(--color-accent-bright)] transition-all duration-200 hover:bg-[var(--color-accent)]/20 hover:border-[var(--color-accent)]/50"
>
<svg width="13" height="13" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><line x1="12" y1="5" x2="12" y2="19"/><line x1="5" y1="12" x2="19" y2="12"/></svg>

View File

@ -35,8 +35,8 @@
// Launch state
let launchTarget = $state<Snapshot | null>(null);
let launchVcpus = $state(1);
let launchMemoryMb = $state(512);
let launchVcpus = $state(2);
let launchMemoryMb = $state(2048);
let launchTimeoutSec = $state(0);
let launching = $state(false);
let launchError = $state<string | null>(null);
@ -81,8 +81,8 @@
function openLaunch(snapshot: Snapshot) {
launchTarget = snapshot;
launchVcpus = snapshot.vcpus ?? 1;
launchMemoryMb = snapshot.memory_mb ?? 512;
launchVcpus = snapshot.vcpus ?? 2;
launchMemoryMb = snapshot.memory_mb ?? 2048;
launchTimeoutSec = 0;
launchError = null;
}
@ -637,7 +637,7 @@
</label>
{#if launchTarget.type === 'snapshot'}
<div class="rounded-[var(--radius-input)] border border-[var(--color-border)] bg-[var(--color-bg-0)] px-3 py-2 font-mono text-ui text-[var(--color-text-muted)]">
{launchTarget.memory_mb ?? 512}
{launchTarget.memory_mb ?? 2048}
</div>
{:else}
<input

View File

@ -11,11 +11,11 @@ import (
"github.com/go-chi/chi/v5"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/pkg/audit"
"git.omukk.dev/wrenn/wrenn/pkg/auth"
"git.omukk.dev/wrenn/wrenn/pkg/db"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/lifecycle"
"git.omukk.dev/wrenn/wrenn/pkg/service"
"git.omukk.dev/wrenn/wrenn/pkg/validate"

View File

@ -12,11 +12,11 @@ import (
"github.com/jackc/pgx/v5/pgtype"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/pkg/audit"
"git.omukk.dev/wrenn/wrenn/pkg/auth"
"git.omukk.dev/wrenn/wrenn/pkg/db"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/lifecycle"
"git.omukk.dev/wrenn/wrenn/pkg/service"
"git.omukk.dev/wrenn/wrenn/pkg/validate"

View File

@ -3852,10 +3852,10 @@ components:
default: minimal-ubuntu
vcpus:
type: integer
default: 1
default: 2
memory_mb:
type: integer
default: 512
default: 2048
timeout_sec:
type: integer
minimum: 0

View File

@ -95,7 +95,7 @@ func New(
apiKeySvc := &service.APIKeyService{DB: queries}
templateSvc := &service.TemplateService{DB: queries}
hostSvc := &service.HostService{DB: queries, Redis: rdb, JWT: jwtSecret, Pool: pool, CA: ca}
teamSvc := &service.TeamService{DB: queries, Pool: pgPool, HostPool: pool}
teamSvc := &service.TeamService{DB: queries, Pool: pgPool, HostPool: pool, Sessions: sessionSvc}
userSvc := &service.UserService{DB: queries, SandboxSvc: sandboxSvc}
auditSvc := &service.AuditService{DB: queries}
statsSvc := &service.StatsService{DB: queries, Pool: pgPool}

View File

@ -3,7 +3,7 @@ package hostagent
import (
"context"
"git.omukk.dev/wrenn/wrenn/internal/sandbox"
"git.omukk.dev/wrenn/wrenn/pkg/sandbox"
)
// callbackAdapter adapts CallbackSender to satisfy sandbox.EventSender.

View File

@ -13,7 +13,7 @@ import (
"sync"
"time"
"git.omukk.dev/wrenn/wrenn/internal/sandbox"
"git.omukk.dev/wrenn/wrenn/pkg/sandbox"
)
const (

View File

@ -19,8 +19,8 @@ import (
pb "git.omukk.dev/wrenn/wrenn/proto/hostagent/gen"
"git.omukk.dev/wrenn/wrenn/proto/hostagent/gen/hostagentv1connect"
"git.omukk.dev/wrenn/wrenn/internal/envdclient"
"git.omukk.dev/wrenn/wrenn/internal/sandbox"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
"git.omukk.dev/wrenn/wrenn/pkg/sandbox"
)
// Server implements the HostAgentService Connect RPC handler.
@ -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) {

View File

@ -1,57 +0,0 @@
package network
import (
"fmt"
"sync"
)
// SlotAllocator manages network slot indices for sandboxes.
// Each sandbox needs a unique slot index for its network addressing.
type SlotAllocator struct {
mu sync.Mutex
inUse map[int]bool
}
// NewSlotAllocator creates a new slot allocator.
func NewSlotAllocator() *SlotAllocator {
return &SlotAllocator{
inUse: make(map[int]bool),
}
}
// Allocate returns the next available slot index (1-based).
func (a *SlotAllocator) Allocate() (int, error) {
a.mu.Lock()
defer a.mu.Unlock()
for i := 1; i <= 32767; i++ {
if !a.inUse[i] {
a.inUse[i] = true
return i, nil
}
}
return 0, fmt.Errorf("no free network slots")
}
// Release frees a slot index for reuse.
func (a *SlotAllocator) Release(index int) {
a.mu.Lock()
defer a.mu.Unlock()
delete(a.inUse, index)
}
// Reserve marks a specific slot index as in use. Returns an error if the
// index is out of range or already taken. Used on resume to re-acquire the
// slot a sandbox previously held so its host-reachable IP stays stable.
func (a *SlotAllocator) Reserve(index int) error {
if index < 1 || index > 32767 {
return fmt.Errorf("slot index out of range: %d", index)
}
a.mu.Lock()
defer a.mu.Unlock()
if a.inUse[index] {
return fmt.Errorf("slot %d already in use", index)
}
a.inUse[index] = true
return nil
}

View File

@ -116,7 +116,13 @@ func hydrateFromDB(queries *db.Queries) func(context.Context, *session.Session)
role := ""
if err == nil {
role = membership.Role
} else if !errors.Is(err, pgx.ErrNoRows) {
} else if errors.Is(err, pgx.ErrNoRows) {
// The session's active team no longer lists this user as a member
// (e.g. they were removed from the team). Strip the team binding so
// the stale session can no longer authorize against team-scoped
// resources such as GetSandboxByTeam.
sess.TeamID = pgtype.UUID{}
} else {
return err
}
sess.Email = user.Email

View File

@ -43,6 +43,20 @@ func (q *Queries) DeleteAPIKeysByTeam(ctx context.Context, teamID pgtype.UUID) e
return err
}
const deleteAPIKeysByTeamAndCreator = `-- name: DeleteAPIKeysByTeamAndCreator :exec
DELETE FROM team_api_keys WHERE team_id = $1 AND created_by = $2
`
type DeleteAPIKeysByTeamAndCreatorParams struct {
TeamID pgtype.UUID `json:"team_id"`
CreatedBy pgtype.UUID `json:"created_by"`
}
func (q *Queries) DeleteAPIKeysByTeamAndCreator(ctx context.Context, arg DeleteAPIKeysByTeamAndCreatorParams) error {
_, err := q.db.Exec(ctx, deleteAPIKeysByTeamAndCreator, arg.TeamID, arg.CreatedBy)
return err
}
const getAPIKeyByHash = `-- name: GetAPIKeyByHash :one
SELECT id, team_id, name, key_hash, key_prefix, created_by, created_at, last_used FROM team_api_keys WHERE key_hash = $1
`

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)
}
@ -205,6 +216,27 @@ func RemoveSnapshot(ctx context.Context, dev *SnapshotDevice) error {
return nil
}
// ReattachSnapshot reconstructs the SnapshotDevice handle for a dm-snapshot
// that was created by an earlier process and is still live in the kernel. No
// device-mapper state is modified — the existing dm device is verified and
// the CoW loop device is rediscovered from its backing file, so a later
// RemoveSnapshot can detach it as usual.
func ReattachSnapshot(name, cowPath string) (*SnapshotDevice, error) {
if !dmDeviceExists(name) {
return nil, fmt.Errorf("dm device %s does not exist", name)
}
cowLoopDev, err := losetupFindByFile(cowPath)
if err != nil {
return nil, fmt.Errorf("find cow loop for %s: %w", cowPath, err)
}
return &SnapshotDevice{
Name: name,
DevicePath: "/dev/mapper/" + name,
CowPath: cowPath,
CowLoopDev: cowLoopDev,
}, nil
}
// FlattenSnapshot reads the full contents of a dm-snapshot device and writes
// it to a new file. This merges the base image + CoW changes into a standalone
// rootfs image suitable for use as a new template.
@ -284,10 +316,55 @@ func LogLoopState() {
}
}
// DetachLoopsByFile best-effort detaches every loop device backed by path.
//
// Safety net for teardown paths that are about to delete a backing file whose
// loop may still be attached (e.g. RemoveSnapshot failed on a busy dm device,
// or startup cleanup removed the dm device before the CoW loop was found).
// Deleting the file first orphans the loop permanently — losetup can no
// longer find it by file, so no later sweep can reclaim it. Detaching first
// always works: if device-mapper still references the loop, the kernel defers
// the release via autoclear until that reference drops.
func DetachLoopsByFile(path string) {
out, err := exec.Command("losetup", "-j", path, "--output", "NAME", "--noheadings").Output()
if err != nil {
slog.Debug("losetup -j failed", "path", path, "error", err)
return
}
for _, dev := range strings.Fields(strings.TrimSpace(string(out))) {
if err := losetupDetachRetry(dev); err != nil {
slog.Warn("loop detach by file failed", "device", dev, "path", path, "error", err)
} else {
slog.Info("detached leftover loop device", "device", dev, "path", path)
}
}
}
// --- low-level helpers ---
// losetupCreate attaches a file as a read-only loop device.
// losetupCreate attaches a file as a read-only loop device, reusing an existing
// attachment for the same backing file when one is already present.
//
// `losetup --find` always allocates a NEW device even when the file is already
// mapped. Within a single long-lived process the LoopRegistry hides this by
// caching the device per image path, but a daemonless caller runs many
// short-lived processes, each with its own registry: when a later process
// re-attaches a sandbox created by an earlier one, a naive --find would attach a
// second loop to the same origin image while the dm-snapshot still references
// the first. That later process then balances its refcount by detaching only
// its own (second) loop, orphaning the original — a loop leak that accumulates
// one device per template per re-attach/destroy cycle. Reusing the existing
// attachment keeps the kernel's loop set 1:1 with backing files, so the device a
// caller releases is the same one the dm origin depends on. Mirrors how
// ReattachSnapshot already rediscovers the CoW loop via losetupFindByFile.
func losetupCreate(imagePath string) (string, error) {
// Reuse when exactly one loop already backs the image. losetupFindByFile
// errors on both "none" and "more than one"; either way fall through to
// attach a fresh device (the multi-attachment case only arises from loops
// leaked before this reuse logic existed, and is reclaimed separately).
if dev, err := losetupFindByFile(imagePath); err == nil {
return dev, nil
}
out, err := exec.Command("losetup", "--read-only", "--find", "--show", imagePath).Output()
if err != nil {
return "", fmt.Errorf("losetup --read-only: %w", err)
@ -304,6 +381,23 @@ func losetupCreateRW(path string) (string, error) {
return strings.TrimSpace(string(out)), nil
}
// losetupFindByFile returns the loop device backed by the given file.
// Errors if none or more than one is attached.
func losetupFindByFile(path string) (string, error) {
out, err := exec.Command("losetup", "-j", path, "--output", "NAME", "--noheadings").Output()
if err != nil {
return "", fmt.Errorf("losetup -j: %w", err)
}
devs := strings.Fields(strings.TrimSpace(string(out)))
if len(devs) == 0 {
return "", fmt.Errorf("no loop device attached to %s", path)
}
if len(devs) > 1 {
return "", fmt.Errorf("multiple loop devices attached to %s: %v", path, devs)
}
return devs[0], nil
}
// losetupDetach detaches a loop device.
func losetupDetach(dev string) error {
return exec.Command("losetup", "-d", dev).Run()

View File

@ -34,7 +34,23 @@ type Client struct {
// New creates a new envd client that connects to the given host IP.
func New(hostIP string) *Client {
base := baseURL(hostIP)
return newWithBase(hostIP, baseURL(hostIP))
}
// NewWithBaseURL creates an envd client against an explicit base URL instead
// of the standard http://{hostIP}:{envdPort}. Intended for tests (httptest
// servers) and nonstandard listeners.
func NewWithBaseURL(base string) *Client {
host := base
// 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)
}
func newWithBase(hostIP, base string) *Client {
httpClient := newHTTPClient()
streamingClient := newStreamingHTTPClient()

View File

@ -169,6 +169,13 @@ func compareSemver(a, b string) int {
return 0
}
// SlotsDir returns the directory holding network slot claim files. Each
// allocated slot index is an empty file named after the index, created with
// O_EXCL so concurrent processes can never claim the same slot.
func SlotsDir(wrennDir string) string {
return filepath.Join(wrennDir, "slots")
}
// ImagesRoot returns the root images directory.
func ImagesRoot(wrennDir string) string {
return filepath.Join(wrennDir, "images")

143
pkg/network/allocator.go Normal file
View File

@ -0,0 +1,143 @@
package network
import (
"fmt"
"os"
"path/filepath"
"strconv"
"sync"
)
// SlotAllocator manages network slot indices for sandboxes.
// Each sandbox needs a unique slot index for its network addressing.
//
// When slotsDir is set, every claim is additionally recorded as a file
// {slotsDir}/{index} created with O_CREAT|O_EXCL, which is atomic on local
// filesystems. That makes allocation safe across processes: two concurrent
// short-lived CLI invocations (each with its own in-memory allocator) can
// never claim the same slot, which would otherwise silently give two VMs the
// same host-reachable IP.
type SlotAllocator struct {
mu sync.Mutex
inUse map[int]bool
// seeded marks slots that have a claim file on disk but no owner in this
// process yet (populated by SeedFromDir). Allocate never hands them out;
// Reserve may adopt one — the caller restoring a paused or running
// sandbox owns the on-disk state that justifies the claim.
seeded map[int]bool
slotsDir string // empty → in-memory only (single-process mode)
}
// NewSlotAllocator creates a new slot allocator. slotsDir enables
// cross-process claim files; pass "" for in-memory-only allocation.
func NewSlotAllocator(slotsDir string) *SlotAllocator {
return &SlotAllocator{
inUse: make(map[int]bool),
seeded: make(map[int]bool),
slotsDir: slotsDir,
}
}
// Allocate returns the next available slot index (1-based).
func (a *SlotAllocator) Allocate() (int, error) {
a.mu.Lock()
defer a.mu.Unlock()
for i := 1; i <= 32767; i++ {
if a.inUse[i] {
continue
}
if err := a.claimFile(i); err != nil {
if os.IsExist(err) {
// Another process holds this slot; remember and move on.
a.inUse[i] = true
continue
}
return 0, fmt.Errorf("claim slot %d: %w", i, err)
}
a.inUse[i] = true
return i, nil
}
return 0, fmt.Errorf("no free network slots")
}
// Release frees a slot index for reuse.
func (a *SlotAllocator) Release(index int) {
a.mu.Lock()
defer a.mu.Unlock()
delete(a.inUse, index)
delete(a.seeded, index)
if a.slotsDir != "" {
_ = os.Remove(a.slotFile(index))
}
}
// Reserve marks a specific slot index as in use. Returns an error if the
// index is out of range or already taken in this process. Used on resume and
// re-attach to re-acquire the slot a sandbox previously held so its
// host-reachable IP stays stable. The claim file may legitimately already
// exist (written by the process that created the sandbox), so EEXIST is not
// an error here — on-disk sandbox state is what proves ownership.
func (a *SlotAllocator) Reserve(index int) error {
if index < 1 || index > 32767 {
return fmt.Errorf("slot index out of range: %d", index)
}
a.mu.Lock()
defer a.mu.Unlock()
if a.inUse[index] && !a.seeded[index] {
return fmt.Errorf("slot %d already in use", index)
}
if err := a.claimFile(index); err != nil && !os.IsExist(err) {
return fmt.Errorf("claim slot %d: %w", index, err)
}
a.inUse[index] = true
delete(a.seeded, index)
return nil
}
// SeedFromDir marks every slot that has a claim file as in-use in this
// process. Called once at startup (before any Allocate) so slots held by
// concurrently running processes — including creates that have not yet
// persisted their sandbox state — are never handed out.
func (a *SlotAllocator) SeedFromDir() error {
if a.slotsDir == "" {
return nil
}
entries, err := os.ReadDir(a.slotsDir)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return fmt.Errorf("read slots dir: %w", err)
}
a.mu.Lock()
defer a.mu.Unlock()
for _, e := range entries {
if idx, err := strconv.Atoi(e.Name()); err == nil && idx >= 1 && idx <= 32767 {
a.inUse[idx] = true
a.seeded[idx] = true
}
}
return nil
}
// claimFile atomically creates the claim file for a slot. Returns an error
// satisfying os.IsExist when another process already holds the slot. No-op
// (nil) when slotsDir is unset.
func (a *SlotAllocator) claimFile(index int) error {
if a.slotsDir == "" {
return nil
}
if err := os.MkdirAll(a.slotsDir, 0o755); err != nil {
return err
}
f, err := os.OpenFile(a.slotFile(index), os.O_CREATE|os.O_EXCL, 0o644)
if err != nil {
return err
}
return f.Close()
}
func (a *SlotAllocator) slotFile(index int) string {
return filepath.Join(a.slotsDir, strconv.Itoa(index))
}

View File

@ -0,0 +1,125 @@
package network
import (
"os"
"path/filepath"
"strconv"
"testing"
)
func TestSlotAllocator_InMemory(t *testing.T) {
a := NewSlotAllocator("")
idx, err := a.Allocate()
if err != nil {
t.Fatalf("Allocate: %v", err)
}
if idx != 1 {
t.Fatalf("first slot = %d, want 1", idx)
}
if err := a.Reserve(idx); err == nil {
t.Fatal("Reserve of allocated slot should fail")
}
a.Release(idx)
if err := a.Reserve(idx); err != nil {
t.Fatalf("Reserve after Release: %v", err)
}
}
func TestSlotAllocator_ClaimFiles(t *testing.T) {
dir := t.TempDir()
a := NewSlotAllocator(dir)
idx, err := a.Allocate()
if err != nil {
t.Fatalf("Allocate: %v", err)
}
claim := filepath.Join(dir, strconv.Itoa(idx))
if _, err := os.Stat(claim); err != nil {
t.Fatalf("claim file missing after Allocate: %v", err)
}
a.Release(idx)
if _, err := os.Stat(claim); !os.IsNotExist(err) {
t.Fatalf("claim file still present after Release (err=%v)", err)
}
}
func TestSlotAllocator_AllocateSkipsExternalClaims(t *testing.T) {
dir := t.TempDir()
// Simulate another process holding slots 1 and 2.
for _, n := range []string{"1", "2"} {
if err := os.WriteFile(filepath.Join(dir, n), nil, 0o644); err != nil {
t.Fatal(err)
}
}
a := NewSlotAllocator(dir)
idx, err := a.Allocate()
if err != nil {
t.Fatalf("Allocate: %v", err)
}
if idx != 3 {
t.Fatalf("Allocate = %d, want 3 (slots 1-2 externally claimed)", idx)
}
}
func TestSlotAllocator_SeedAndReserveAdopt(t *testing.T) {
dir := t.TempDir()
if err := os.WriteFile(filepath.Join(dir, "5"), nil, 0o644); err != nil {
t.Fatal(err)
}
a := NewSlotAllocator(dir)
if err := a.SeedFromDir(); err != nil {
t.Fatalf("SeedFromDir: %v", err)
}
// Seeded slots are never handed out by Allocate.
idx, err := a.Allocate()
if err != nil {
t.Fatalf("Allocate: %v", err)
}
if idx == 5 {
t.Fatal("Allocate handed out a seeded slot")
}
// A seeded slot can be adopted exactly once via Reserve (restore path)...
if err := a.Reserve(5); err != nil {
t.Fatalf("Reserve of seeded slot: %v", err)
}
// ...after which it is owned and a second Reserve fails.
if err := a.Reserve(5); err == nil {
t.Fatal("second Reserve of adopted slot should fail")
}
}
func TestSlotAllocator_ReserveCreatesClaimFile(t *testing.T) {
dir := t.TempDir()
a := NewSlotAllocator(dir)
if err := a.Reserve(7); err != nil {
t.Fatalf("Reserve: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, "7")); err != nil {
t.Fatalf("claim file missing after Reserve: %v", err)
}
}
func TestSlotAllocator_ReserveOutOfRange(t *testing.T) {
a := NewSlotAllocator("")
for _, idx := range []int{0, -1, 32768} {
if err := a.Reserve(idx); err == nil {
t.Fatalf("Reserve(%d) should fail", idx)
}
}
}
func TestSlotAllocator_SeedFromMissingDir(t *testing.T) {
a := NewSlotAllocator(filepath.Join(t.TempDir(), "does-not-exist"))
if err := a.SeedFromDir(); err != nil {
t.Fatalf("SeedFromDir on missing dir: %v", err)
}
}

View File

@ -545,6 +545,107 @@ func RemoveNetwork(slot *Slot) error {
return errors.Join(errs...)
}
var privateEgressGuardNets = []string{
"10.0.0.0/8",
"172.16.0.0/12",
"192.168.0.0/16",
"169.254.0.0/16",
"100.64.0.0/10",
}
var sandboxSuperNets = []string{"10.11.0.0/16", "10.12.0.0/16"}
const vethMatch = "wrenn-veth+"
// Installs host-level protections that stop capsule from (a) leaking traffic
// destined to private ranges onto the upstream network via the public NIC
// and (b) reaching another capsule's host-reachable IP.
func EnsureEgressGuard() error {
defaultIface, err := getDefaultInterface()
if err != nil {
return fmt.Errorf("resolve default interface: %w", err)
}
// Drop forwarded capsule traffic destined to private ranges out the public
// NIC. Insert at the head of FORWARD so it precedes the per-sandbox ACCEPT
// rules appended when a sandbox is created.
for _, n := range privateEgressGuardNets {
if err := ensureHostRule(
[]string{"-C", "FORWARD", "-o", defaultIface, "-d", n, "-j", "DROP"},
[]string{"-I", "FORWARD", "1", "-o", defaultIface, "-d", n, "-j", "DROP"},
); err != nil {
return fmt.Errorf("install egress guard for %s: %w", n, err)
}
}
// Deny capsule-to-capsule forwarding outright: a veth may reach the uplink,
// never another sandbox's veth. This closes cross-tenant reach to the
// unauthenticated guest agent even for allocated slots.
if err := ensureHostRule(
[]string{"-C", "FORWARD", "-i", vethMatch, "-o", vethMatch, "-j", "DROP"},
[]string{"-I", "FORWARD", "1", "-i", vethMatch, "-o", vethMatch, "-j", "DROP"},
); err != nil {
return fmt.Errorf("install inter-sandbox guard: %w", err)
}
// Deny capsule-initiated connections to the host's own services. The
// FORWARD guard above only covers transit traffic; packets addressed to
// one of the host's own IPs are delivered locally via INPUT and would
// otherwise reach anything the host binds on all interfaces (control
// plane, host agent, SSH). Replies to host-initiated connections
// (e.g. envd health checks / exec, where the host is the client) arrive
// on the same veth and MUST still be accepted, so allow ESTABLISHED first
// and drop only NEW capsule-sourced flows. Guest DNS points at public
// resolvers (routed out the uplink, not INPUT), so this does not affect
// name resolution.
// Insert the DROP first, then prepend the ESTABLISHED accept ahead of it
// (each "-I INPUT 1" prepends), so the final order is always accept-then-
// drop without hardcoding a chain position that other host rules could
// shift.
if err := ensureHostRule(
[]string{"-C", "INPUT", "-i", vethMatch, "-j", "DROP"},
[]string{"-I", "INPUT", "1", "-i", vethMatch, "-j", "DROP"},
); err != nil {
return fmt.Errorf("install host input guard: %w", err)
}
if err := ensureHostRule(
[]string{"-C", "INPUT", "-i", vethMatch, "-m", "conntrack", "--ctstate", "ESTABLISHED,RELATED", "-j", "ACCEPT"},
[]string{"-I", "INPUT", "1", "-i", vethMatch, "-m", "conntrack", "--ctstate", "ESTABLISHED,RELATED", "-j", "ACCEPT"},
); err != nil {
return fmt.Errorf("install host input allow-established: %w", err)
}
// Blackhole the sandbox supernets so packets to unallocated slot IPs are
// dropped rather than routed to the default gateway.
for _, cidr := range sandboxSuperNets {
if err := ensureBlackholeRoute(cidr); err != nil {
return fmt.Errorf("blackhole %s: %w", cidr, err)
}
}
slog.Info("egress guard installed", "default_iface", defaultIface)
return nil
}
// ensureHostRule inserts a host iptables rule only when an identical rule is
// not already present, keeping EnsureEgressGuard idempotent across restarts.
func ensureHostRule(checkArgs, insertArgs []string) error {
if exec.Command("iptables", checkArgs...).Run() == nil {
return nil // already present
}
return iptablesHost(insertArgs...)
}
// ensureBlackholeRoute idempotently installs a blackhole route via `ip route
// replace` (a no-op if the route already matches).
func ensureBlackholeRoute(cidr string) error {
cmd := exec.Command("ip", "route", "replace", "blackhole", cidr)
if out, err := cmd.CombinedOutput(); err != nil {
return fmt.Errorf("ip route replace blackhole %s: %s: %w", cidr, string(out), err)
}
return nil
}
// nsExec runs a command inside a network namespace.
func nsExec(nsName string, command string, args ...string) error {
cmdArgs := append([]string{"netns", "exec", nsName, command}, args...)

View File

@ -3,7 +3,7 @@ package sandbox
import (
"testing"
"git.omukk.dev/wrenn/wrenn/internal/envdclient"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
)
func TestIsBusy(t *testing.T) {

View File

@ -11,8 +11,8 @@ import (
"github.com/jackc/pgx/v5/pgtype"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
)
// DefaultDiskSizeMB is the standard disk size for base images. Images smaller

View File

@ -17,11 +17,11 @@ import (
"github.com/jackc/pgx/v5/pgtype"
"git.omukk.dev/wrenn/wrenn/internal/devicemapper"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/internal/models"
"git.omukk.dev/wrenn/wrenn/internal/network"
"git.omukk.dev/wrenn/wrenn/pkg/devicemapper"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/models"
"git.omukk.dev/wrenn/wrenn/pkg/network"
)
// createFromSnapshotTemplate launches a new sandbox from a snapshot-template
@ -147,6 +147,7 @@ func (m *Manager) createFromSnapshotTemplate(
dmDevice: dmDev,
baseImagePath: baseRootfs,
sandboxDirOverride: meta.SandboxDir,
lazyRestore: true,
}
sb.client.Store(client)
@ -162,6 +163,7 @@ func (m *Manager) createFromSnapshotTemplate(
m.startSampler(sb)
m.startCrashWatcher(sb)
m.writeRunningState(sb)
slog.Info("sandbox launched from snapshot template",
"id", sandboxID,

View File

@ -15,13 +15,13 @@ import (
"github.com/jackc/pgx/v5/pgtype"
"git.omukk.dev/wrenn/wrenn/internal/devicemapper"
"git.omukk.dev/wrenn/wrenn/internal/envdclient"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/internal/models"
"git.omukk.dev/wrenn/wrenn/internal/network"
"git.omukk.dev/wrenn/wrenn/internal/vm"
"git.omukk.dev/wrenn/wrenn/pkg/devicemapper"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/models"
"git.omukk.dev/wrenn/wrenn/pkg/network"
"git.omukk.dev/wrenn/wrenn/pkg/vm"
envdpb "git.omukk.dev/wrenn/wrenn/proto/envd/gen"
)
@ -211,6 +211,15 @@ type sandboxState struct {
// find rootfs.ext4 in the new mount namespace.
sandboxDirOverride string
// lazyRestore records that this sandbox's CH process was launched with
// --restore ...,memory_restore_mode=ondemand (Resume or snapshot-template
// launch): guest pages fault in lazily from the source memory-ranges, so
// any vm.snapshot taken before the guest-side preload completes would emit
// holes for never-faulted pages — silent corruption on the next restore.
// Pause/CreateSnapshot consult this via ensureMemoryMaterialized. Persisted
// in the running-state file so re-attached sandboxes keep the guarantee.
lazyRestore bool
// Background memory loading state (set during Resume for UFFD sandboxes).
// nil for freshly-created sandboxes. For resumed sandboxes, memLoadDone
// is closed when the background loader finishes (success or failure).
@ -264,10 +273,16 @@ func New(cfg Config) *Manager {
if cfg.EnvdTimeout == 0 {
cfg.EnvdTimeout = 30 * time.Second
}
slots := network.NewSlotAllocator(layout.SlotsDir(cfg.WrennDir))
// Honor slots claimed by other live processes (e.g. a concurrent CLI
// invocation mid-Create) before handing any out ourselves.
if err := slots.SeedFromDir(); err != nil {
slog.Warn("failed to seed slot allocator from disk", "error", err)
}
return &Manager{
cfg: cfg,
vm: vm.NewManager(),
slots: network.NewSlotAllocator(),
slots: slots,
loops: devicemapper.NewLoopRegistry(),
boxes: make(map[string]*sandboxState),
creates: make(map[string]*createHandle),
@ -314,10 +329,10 @@ func (m *Manager) Create(
}
if vcpus <= 0 {
vcpus = 1
vcpus = 2
}
if memoryMB <= 0 {
memoryMB = 512
memoryMB = 2048
}
if diskSizeMB <= 0 {
diskSizeMB = m.cfg.DefaultRootfsSizeMB
@ -505,6 +520,7 @@ func (m *Manager) Create(
m.startSampler(sb)
m.startCrashWatcher(sb)
m.writeRunningState(sb)
slog.Info("sandbox created",
"id", sandboxID,
@ -606,6 +622,8 @@ 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)
@ -613,8 +631,10 @@ func (m *Manager) cleanup(ctx context.Context, sb *sandboxState) {
os.Remove(sb.dmDevice.CowPath)
}
// Paused branch: dm-snapshot and loop were already released by
// releaseRuntime; the CoW file inside the sandbox dir is removed by
// Destroy's os.RemoveAll(SandboxDir) below.
// releaseRuntime, which also cleared dmDevice and baseImagePath — both
// guards below are nil/empty for a paused sandbox, so nothing double-
// releases. The CoW file inside the sandbox dir is removed by Destroy's
// os.RemoveAll(SandboxDir) below.
if sb.baseImagePath != "" {
m.loops.Release(sb.baseImagePath)
}
@ -1262,6 +1282,8 @@ 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)

View File

@ -41,13 +41,13 @@ import (
"github.com/jackc/pgx/v5/pgtype"
"git.omukk.dev/wrenn/wrenn/internal/devicemapper"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/internal/models"
"git.omukk.dev/wrenn/wrenn/internal/network"
"git.omukk.dev/wrenn/wrenn/internal/snapshot"
"git.omukk.dev/wrenn/wrenn/internal/vm"
"git.omukk.dev/wrenn/wrenn/pkg/devicemapper"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/models"
"git.omukk.dev/wrenn/wrenn/pkg/network"
"git.omukk.dev/wrenn/wrenn/pkg/vm"
)
const (
@ -178,11 +178,12 @@ func (m *Manager) Pause(ctx context.Context, sandboxID string) error {
lastPhase = now
}
// Wait for the post-resume memory loader to finish before snapshotting.
// Without this, ch.snapshot's SEEK_DATA/SEEK_HOLE writer would emit holes
// for any page not yet faulted in, which read back as zero on the next
// restore — silent corruption across pause/resume chains.
if err := m.waitForMemoryLoader(ctx, sb); err != nil {
// Ensure every guest page is resident before snapshotting. Without this,
// ch.snapshot's SEEK_DATA/SEEK_HOLE writer would emit holes for any page
// not yet faulted in, which read back as zero on the next restore —
// silent corruption across pause/resume chains. Verified against envd's
// preload status, not just the in-process loader goroutine.
if err := m.ensureMemoryMaterialized(ctx, sb); err != nil {
return fmt.Errorf("pause %s: %w", sandboxID, err)
}
phase("memory_loader_wait")
@ -204,6 +205,26 @@ func (m *Manager) Pause(ctx context.Context, sandboxID string) error {
sb.Status = models.StatusError
m.mu.Unlock()
sb.connTracker.Reset()
// The VM is dead (typically it crashed mid-quiesce — the crash
// watcher saw StatusPausing and deliberately stood down). Nothing
// will ever resume this sandbox, so release its runtime now:
// otherwise the network slot, dm-snapshot, loop refcount, and
// sampler goroutine stay held until an explicit Destroy or agent
// shutdown. Zero SlotIndex so a later Destroy's cleanup() cannot
// release the (possibly re-allocated) slot a second time.
m.releaseRuntime(sb, dropCow)
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)
}
@ -220,7 +241,7 @@ func (m *Manager) Pause(ctx context.Context, sandboxID string) error {
phase("quiesce_and_pause")
// Memory materialisation is handled out-of-band by the background loader
// kicked off by Resume after /init. We blocked on it above (waitForMemoryLoader)
// kicked off by Resume after /init. We blocked on it above (ensureMemoryMaterialized)
// so by the time we reach ch.snapshot every guest page is resident in CH's
// memfile and SEEK_DATA/SEEK_HOLE produces a self-contained snapshot.
@ -499,6 +520,8 @@ 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)
@ -511,10 +534,16 @@ func (m *Manager) releaseRuntime(sb *sandboxState, cow cowDisposition) {
m.loops.Release(sb.baseImagePath)
}
// Clear runtime references; they're rebuilt on resume.
// Clear runtime references; they're rebuilt on resume. baseImagePath MUST
// be cleared along with the rest: it pairs 1:1 with the loop refcount just
// released, and a later Destroy-of-paused runs cleanup(), which releases
// the loop again whenever baseImagePath is set — underflowing the shared
// registry and detaching the base image loop out from under every other
// sandbox running on the same template.
sb.slot = nil
sb.client.Store(nil)
sb.dmDevice = nil
sb.baseImagePath = ""
}
// Resume re-launches a paused sandbox from its on-disk snapshot. The same
@ -667,6 +696,7 @@ func (m *Manager) resumeFromMeta(ctx context.Context, sb *sandboxState, meta *sn
// baseImagePath is populated — the restored entry intentionally leaves
// it empty so a Destroy-before-Resume cannot underflow the registry.
sb.baseImagePath = meta.BaseTemplate
sb.lazyRestore = true
sb.connTracker.Reset()
sb.HostIP = slot.HostIP
sb.RootfsPath = dmDev.DevicePath
@ -676,6 +706,7 @@ func (m *Manager) resumeFromMeta(ctx context.Context, sb *sandboxState, meta *sn
m.startSampler(sb)
m.startCrashWatcher(sb)
m.writeRunningState(sb)
// Background memory loader is started by the outer Resume AFTER /init
// completes — see comment there for the race rationale.
@ -752,23 +783,65 @@ func (m *Manager) startMemoryLoader(sb *sandboxState) {
}()
}
// waitForMemoryLoader blocks until the background memory loader finishes, or
// until ctx is cancelled. Returns nil if the loader is already done or not
// running. A pause must wait on this before ch.snapshot so the resulting
// memory-ranges is self-contained.
func (m *Manager) waitForMemoryLoader(ctx context.Context, sb *sandboxState) error {
// ensureMemoryMaterialized guarantees that every guest page of a
// lazily-restored sandbox is resident in CH's memfile before a vm.snapshot.
// For cold-booted sandboxes (lazyRestore unset) it is a no-op.
//
// envd's /memory/preload status is the source of truth — not the in-process
// loader goroutine, which does not survive a host-agent restart and whose
// completion only means "polling ended", not "preload succeeded". The
// in-process channel is still awaited first so a running loader isn't raced;
// the verdict then comes from the guest:
//
// done → safe to snapshot
// idle / running → (re)start and wait: idle happens when the agent died
// between restore and the loader's POST, or a previous
// start attempt failed on the wire
// failed / cancelled → hard error; snapshotting now would silently zero all
// never-faulted guest memory on the next restore
func (m *Manager) ensureMemoryMaterialized(ctx context.Context, sb *sandboxState) error {
m.mu.RLock()
lazy := sb.lazyRestore
done := sb.memLoadDone
m.mu.RUnlock()
if done == nil {
if !lazy {
return nil
}
select {
case <-done:
return nil
case <-ctx.Done():
return fmt.Errorf("wait for memory loader: %w", ctx.Err())
if done != nil {
select {
case <-done:
case <-ctx.Done():
return fmt.Errorf("wait for memory loader: %w", ctx.Err())
}
}
client := sb.client.Load()
if client == nil {
return fmt.Errorf("memory preload verify: envd client unavailable")
}
// 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 start: %w", err)
}
if status.State != "done" {
status, err = client.WaitMemoryPreload(ctx)
if err != nil {
return fmt.Errorf("memory preload wait: %w", err)
}
}
if status.State != "done" {
return fmt.Errorf("memory preload %s: %s (source %q, %d pages, %d bytes)",
status.State, status.Error, status.Source, status.Pages, status.Bytes)
}
return nil
}
// CreateSnapshot writes a self-contained template snapshot to
@ -812,10 +885,10 @@ func (m *Manager) CreateSnapshot(ctx context.Context, sandboxID string, teamID,
func (m *Manager) snapshotRunningToTemplate(ctx context.Context, sb *sandboxState, teamID, templateID pgtype.UUID, name string) (int64, error) {
sandboxID := sb.ID
// Same rationale as Pause: wait for the background memory loader so the
// resulting memory-ranges is self-contained when this sandbox itself was
// previously restored from an ondemand snapshot.
if err := m.waitForMemoryLoader(ctx, sb); err != nil {
// Same rationale as Pause: ensure guest memory is fully materialised so
// the resulting memory-ranges is self-contained when this sandbox itself
// was previously restored from an ondemand snapshot.
if err := m.ensureMemoryMaterialized(ctx, sb); err != nil {
return 0, fmt.Errorf("create snapshot %s: %w", sandboxID, err)
}

152
pkg/sandbox/pause_test.go Normal file
View File

@ -0,0 +1,152 @@
package sandbox
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
)
// preloadServer fakes envd's /memory/preload endpoint. GET returns the states
// in sequence (last one repeats); POST marks the loader started and returns
// the current state.
type preloadServer struct {
states []string
idx atomic.Int64
posts atomic.Int64
gets atomic.Int64
}
func (s *preloadServer) handler(w http.ResponseWriter, r *http.Request) {
if !strings.HasSuffix(r.URL.Path, "/memory/preload") {
http.NotFound(w, r)
return
}
switch r.Method {
case http.MethodPost:
s.posts.Add(1)
case http.MethodGet:
s.gets.Add(1)
}
i := s.idx.Load()
if int(i) >= len(s.states) {
i = int64(len(s.states) - 1)
} else {
s.idx.Add(1)
}
state := s.states[i]
resp := map[string]any{"state": state}
if state == "failed" {
resp["error"] = "kcore walk read no pages"
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(resp)
}
func TestEnsureMemoryMaterialized(t *testing.T) {
tests := []struct {
name string
lazy bool
nilClient bool
states []string // envd responses in order
wantErr string // substring; "" = expect nil
wantPosts int64 // exact POST count; -1 = don't check
}{
{
name: "cold boot is a no-op",
lazy: false,
states: []string{"idle"},
wantErr: "",
wantPosts: 0,
},
{
name: "lazy and already done",
lazy: true,
states: []string{"done"},
wantErr: "",
wantPosts: 1, // single POST reports done, no polling
},
{
name: "lazy idle self-heals via start",
lazy: true,
states: []string{"idle", "running", "done", "done"},
wantErr: "",
wantPosts: 1,
},
{
name: "lazy failed after wait is a hard error",
lazy: true,
states: []string{"idle", "failed"},
wantErr: "memory preload failed",
wantPosts: 1,
},
{
name: "lazy cancelled is a hard error",
lazy: true,
states: []string{"running", "cancelled"},
wantErr: "memory preload cancelled",
wantPosts: 1,
},
{
name: "lazy without client fails",
lazy: true,
nilClient: true,
wantErr: "envd client unavailable",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
srv := &preloadServer{states: tt.states}
ts := httptest.NewServer(http.HandlerFunc(srv.handler))
defer ts.Close()
m := &Manager{}
sb := &sandboxState{lazyRestore: tt.lazy}
if !tt.nilClient {
sb.client.Store(envdclient.NewWithBaseURL(ts.URL))
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err := m.ensureMemoryMaterialized(ctx, sb)
if tt.wantErr == "" {
if err != nil {
t.Fatalf("ensureMemoryMaterialized: %v", err)
}
} else {
if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
t.Fatalf("want error containing %q, got %v", tt.wantErr, err)
}
}
if tt.wantPosts >= 0 && srv.posts.Load() != tt.wantPosts {
t.Fatalf("POST count = %d, want %d", srv.posts.Load(), tt.wantPosts)
}
if !tt.lazy && srv.gets.Load() != 0 {
t.Fatalf("cold boot must not query envd, got %d GETs", srv.gets.Load())
}
})
}
}
// ensureMemoryMaterialized must first drain an in-process loader goroutine
// before consulting envd, and must honor ctx cancellation while blocked on it.
func TestEnsureMemoryMaterializedWaitsForLoader(t *testing.T) {
m := &Manager{}
sb := &sandboxState{lazyRestore: true}
sb.memLoadDone = make(chan struct{}) // never closed
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
err := m.ensureMemoryMaterialized(ctx, sb)
if err == nil || !strings.Contains(err.Error(), "wait for memory loader") {
t.Fatalf("want loader-wait timeout error, got %v", err)
}
}

View File

@ -10,7 +10,7 @@ import (
"strconv"
"strings"
"git.omukk.dev/wrenn/wrenn/internal/envdclient"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
)
// cpuStat holds raw CPU jiffies read from /proc/{pid}/stat.

View File

@ -17,9 +17,9 @@ import (
"log/slog"
"path/filepath"
"git.omukk.dev/wrenn/wrenn/internal/envdclient"
"git.omukk.dev/wrenn/wrenn/internal/network"
"git.omukk.dev/wrenn/wrenn/internal/vm"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
"git.omukk.dev/wrenn/wrenn/pkg/network"
"git.omukk.dev/wrenn/wrenn/pkg/vm"
)
// restoreInputs is the common set of fields needed to build a restore VMConfig.

View File

@ -11,8 +11,8 @@ import (
"github.com/google/uuid"
"git.omukk.dev/wrenn/wrenn/internal/layout"
"git.omukk.dev/wrenn/wrenn/internal/models"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/models"
)
// RestorePausedSandboxes scans WRENN_DIR/sandboxes/ for paused-sandbox
@ -68,6 +68,29 @@ func (m *Manager) RestorePausedSandboxes() {
continue
}
// Skip sandboxes already registered — a resumed-then-still-running
// sandbox re-attached by RestoreRunningSandboxes keeps its previous
// pause generation on disk, but the live VM wins.
m.mu.RLock()
_, known := m.boxes[name]
m.mu.RUnlock()
if known {
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 {

View File

@ -0,0 +1,269 @@
package sandbox
import (
"context"
"fmt"
"log/slog"
"os"
"strings"
"syscall"
"time"
"git.omukk.dev/wrenn/wrenn/pkg/devicemapper"
"git.omukk.dev/wrenn/wrenn/pkg/envdclient"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/models"
"git.omukk.dev/wrenn/wrenn/pkg/network"
"git.omukk.dev/wrenn/wrenn/pkg/vm"
)
// reattachProbeTimeout bounds the CH API probe issued while re-attaching to
// a live VM. Local unix-socket call — kept tight so a dead socket fails fast.
const reattachProbeTimeout = 3 * time.Second
// RestoreRunningSandboxes scans WRENN_DIR/sandboxes/ for sandboxes whose
// creating process has exited but whose VMs are still running, and re-attaches
// them to this manager. This is what lets a short-lived CLI process Exec into
// or Destroy a sandbox created by an earlier invocation.
//
// A sandbox qualifies when its dir contains a sandbox.json running-state file
// (written on create/resume, removed by every teardown path) AND its CH
// process is verifiably alive: the recorded PID exists, its cmdline still
// references this sandbox (PID-recycling guard), and the CH API socket
// answers. Anything else is a stale record from a crash — its leftover
// resources are swept, but a pause snapshot in the same dir is preserved so
// RestorePausedSandboxes can still recover it.
//
// Must be called once at startup, BEFORE RestorePausedSandboxes: a resumed
// sandbox's dir contains both the running-state file and the previous
// generation's pause snapshot, and the running state must win while the VM
// is alive.
//
// Never returns an error — a single unrecoverable sandbox must not block
// startup. Mirrors RestorePausedSandboxes.
func (m *Manager) RestoreRunningSandboxes() {
sandboxesDir := layout.SandboxesDir(m.cfg.WrennDir)
entries, err := os.ReadDir(sandboxesDir)
if err != nil {
// Directory does not exist yet — fresh install, nothing to restore.
return
}
restored, swept := 0, 0
for _, e := range entries {
if !e.IsDir() {
continue
}
name := e.Name()
if strings.Contains(name, ".staging-") || strings.Contains(name, ".trash-") {
continue
}
st, err := readRunningState(m.cfg.WrennDir, name)
if err != nil {
if !os.IsNotExist(err) {
slog.Warn("restore running: unreadable state file, removing",
"id", name, "error", err)
deleteRunningState(m.cfg.WrennDir, name)
}
continue
}
m.mu.RLock()
_, known := m.boxes[name]
m.mu.RUnlock()
if known {
continue
}
if !chProcessAlive(st) {
m.sweepDeadRunning(st)
swept++
continue
}
if err := m.reattachRunning(st); err != nil {
slog.Warn("restore running: re-attach failed", "id", name, "error", err)
continue
}
restored++
}
if restored > 0 || swept > 0 {
slog.Info("running sandbox restore complete", "restored", restored, "swept", swept)
}
}
// chProcessAlive reports whether the recorded CH process is still the CH
// process for this sandbox: PID alive, cmdline still references the sandbox
// (guards against PID recycling), API socket present.
func chProcessAlive(st *runningState) bool {
if st.CHPID <= 0 {
return false
}
if err := syscall.Kill(st.CHPID, 0); err != nil {
return false
}
// The PID is the unshare wrapper whose bash script embeds the sandbox's
// socket path and dm device path — both contain the sandbox ID.
cmdline, err := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", st.CHPID))
if err != nil || !strings.Contains(string(cmdline), st.ID) {
return false
}
if _, err := os.Stat(st.CHSocket); err != nil {
return false
}
return true
}
// reattachRunning rebuilds the in-memory sandboxState for a verified-live
// sandbox and registers it in m.boxes.
func (m *Manager) reattachRunning(st *runningState) error {
teamBytes, err := parsePlainUUID(st.TeamID)
if err != nil {
return fmt.Errorf("bad team_id: %w", err)
}
templateBytes, err := parsePlainUUID(st.TemplateID)
if err != nil {
return fmt.Errorf("bad template_id: %w", err)
}
if st.SlotIndex == 0 {
return fmt.Errorf("running state has no slot_index")
}
// Adopt the slot (its claim file already exists — written by the creating
// process) so a concurrent Create cannot be handed this sandbox's IP.
if err := m.slots.Reserve(st.SlotIndex); err != nil {
return fmt.Errorf("reserve slot %d: %w", st.SlotIndex, err)
}
slot := network.NewSlot(st.SlotIndex)
// NO rollback past this point. chProcessAlive just verified a live CH
// process that depends on this slot's netns/IP, the dm device, and the
// loop attachments. Releasing any of it on a failed adoption would pull
// resources out from under a running VM: freeing the slot lets a new
// Create collide with the live netns, and dropping the loop refcount can
// detach the origin loop the dm table references (LoopRegistry adopts the
// previous process's loop, so refcount 1 here IS that live attachment).
// On failure, leave everything held and unmanaged — the next agent
// restart retries the adoption, or sweeps properly once CH is dead.
// Rediscover the live dm-snapshot; no device-mapper state is modified.
dmDev, err := devicemapper.ReattachSnapshot(st.DMName, st.CowPath)
if err != nil {
return fmt.Errorf("reattach dm-snapshot (slot %d left reserved for live CH): %w",
st.SlotIndex, err)
}
// Re-acquire the base image in this process's loop registry so the
// refcounting stays balanced with the Release in cleanup()/releaseRuntime.
// (The dm device keeps referencing the creating process's origin loop by
// device number; this Acquire only matters for bookkeeping and for other
// sandboxes sharing the base in this process.)
if _, err := m.loops.Acquire(st.BaseImagePath); err != nil {
return fmt.Errorf("acquire base image loop (slot %d left reserved for live CH): %w",
st.SlotIndex, err)
}
// Register the live CH process; probes the API socket before committing.
probeCtx, cancel := context.WithTimeout(context.Background(), reattachProbeTimeout)
defer cancel()
_, err = m.vm.Reattach(probeCtx, vm.VMConfig{
SandboxID: st.ID,
SocketPath: st.CHSocket,
SandboxDir: st.SandboxDir,
VCPUs: st.VCPUs,
MemoryMB: st.MemoryMB,
}, st.CHPID)
if err != nil {
return fmt.Errorf("reattach VM (slot %d and loop refcount left held for live CH): %w",
st.SlotIndex, err)
}
sb := &sandboxState{
Sandbox: models.Sandbox{
ID: st.ID,
Status: models.StatusRunning,
TemplateTeamID: teamBytes,
TemplateID: templateBytes,
VCPUs: st.VCPUs,
MemoryMB: st.MemoryMB,
TimeoutSec: st.TimeoutSec,
SlotIndex: st.SlotIndex,
HostIP: slot.HostIP,
RootfsPath: dmDev.DevicePath,
CreatedAt: st.CreatedAt,
LastActiveAt: time.Now(),
Metadata: st.Metadata,
},
slot: slot,
connTracker: &ConnTracker{},
dmDevice: dmDev,
baseImagePath: st.BaseImagePath,
sandboxDirOverride: st.SandboxDirOverride,
lazyRestore: st.LazyRestore,
}
sb.client.Store(envdclient.New(slot.HostIP.String()))
m.mu.Lock()
m.boxes[st.ID] = sb
m.mu.Unlock()
// A lazily-restored VM may still be materialising guest memory when the
// creating process died. Re-arm the loader: envd's POST is CAS-idempotent
// (a completed preload answers "done" immediately), and having memLoadDone
// set again restores the TTL-reaper/activity-sampler guards. Pause remains
// safe either way — ensureMemoryMaterialized verifies against envd itself.
if sb.lazyRestore {
m.startMemoryLoader(sb)
}
m.startSampler(sb)
m.startCrashWatcher(sb)
slog.Info("re-attached running sandbox",
"id", st.ID, "slot", st.SlotIndex, "pid", st.CHPID, "host_ip", slot.HostIP.String())
return nil
}
// sweepDeadRunning cleans up after a sandbox whose running-state file exists
// but whose VM is gone (creating process crashed, host rebooted, VM killed).
// Best-effort: every resource may or may not still exist.
//
// A pause snapshot in the same dir is preserved — the sandbox may have been
// resumed and then died, in which case the previous pause generation is still
// the best available recovery point and RestorePausedSandboxes (called after
// this) will register it. Without a snapshot the dir is removed entirely.
func (m *Manager) sweepDeadRunning(st *runningState) {
slog.Info("restore running: sweeping dead sandbox", "id", st.ID)
if dmDev, err := devicemapper.ReattachSnapshot(st.DMName, st.CowPath); err == nil {
if err := devicemapper.RemoveSnapshot(context.Background(), dmDev); err != nil {
slog.Warn("sweep: dm-snapshot remove failed", "id", st.ID, "error", err)
}
}
// ReattachSnapshot usually fails here: Setup's CleanupStaleDevices already
// removed the dead sandbox's dm device (it was not held open by any live
// CH), so RemoveSnapshot above never runs and the CoW loop from the
// previous agent stays attached. Detach it by backing file before the
// os.RemoveAll below deletes that file and orphans the loop for good.
devicemapper.DetachLoopsByFile(st.CowPath)
if err := network.RemoveNetwork(network.NewSlot(st.SlotIndex)); err != nil {
slog.Warn("sweep: network remove failed", "id", st.ID, "error", err)
}
_ = os.Remove(st.CHSocket)
deleteRunningState(m.cfg.WrennDir, st.ID)
snapDir := layout.PauseSnapshotDir(m.cfg.WrennDir, st.ID)
if _, err := readSnapshotMeta(snapDir); err == nil {
// Recoverable pause snapshot present: keep the dir, the CoW file, and
// the slot claim so RestorePausedSandboxes can adopt them.
return
}
if err := os.RemoveAll(layout.SandboxDir(m.cfg.WrennDir, st.ID)); err != nil {
slog.Warn("sweep: sandbox dir remove failed", "id", st.ID, "error", err)
}
if st.SlotIndex > 0 {
m.slots.Release(st.SlotIndex)
}
}

182
pkg/sandbox/setup.go Normal file
View File

@ -0,0 +1,182 @@
package sandbox
import (
"bufio"
"fmt"
"os"
"strconv"
"strings"
"git.omukk.dev/wrenn/wrenn/pkg/devicemapper"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/network"
"git.omukk.dev/wrenn/wrenn/pkg/vm"
)
// DefaultCHBin is the conventional install path for the Cloud Hypervisor
// binary, used when SetupOptions.CHBin is empty.
const DefaultCHBin = "/usr/local/bin/cloud-hypervisor"
// CheckPrivileges verifies the process has all Linux capabilities required to
// run sandboxes (DAC_OVERRIDE, KILL, NET_ADMIN, NET_RAW, SYS_PTRACE,
// SYS_ADMIN, MKNOD). It always reads CapEff — even for root — because a root
// process inside a restricted container (e.g. docker --cap-drop=all) may not
// have all caps.
func CheckPrivileges() error {
capEff, err := readEffectiveCaps()
if err != nil {
return fmt.Errorf("read capabilities: %w", err)
}
required := []struct {
bit uint
name string
}{
{1, "CAP_DAC_OVERRIDE"}, // /dev/loop*, /dev/mapper/*, /dev/net/tun
{5, "CAP_KILL"}, // SIGTERM/SIGKILL to cloud-hypervisor processes
{12, "CAP_NET_ADMIN"}, // netlink, iptables, routing, TAP/veth
{13, "CAP_NET_RAW"}, // raw sockets (iptables)
{19, "CAP_SYS_PTRACE"}, // reading /proc/self/ns/net (netns.Get)
{21, "CAP_SYS_ADMIN"}, // netns, mount ns, losetup, dmsetup
{27, "CAP_MKNOD"}, // device-mapper node creation
}
var missing []string
for _, c := range required {
if capEff&(1<<c.bit) == 0 {
missing = append(missing, c.name)
}
}
if len(missing) > 0 {
return fmt.Errorf("missing capabilities: %s — run as root or apply setcap to the binary",
strings.Join(missing, ", "))
}
return nil
}
// readEffectiveCaps parses the CapEff bitmask from /proc/self/status.
func readEffectiveCaps() (uint64, error) {
f, err := os.Open("/proc/self/status")
if err != nil {
return 0, err
}
defer f.Close()
scanner := bufio.NewScanner(f)
for scanner.Scan() {
line := scanner.Text()
if hexStr, ok := strings.CutPrefix(line, "CapEff:"); ok {
return strconv.ParseUint(strings.TrimSpace(hexStr), 16, 64)
}
}
if err := scanner.Err(); err != nil {
return 0, fmt.Errorf("read /proc/self/status: %w", err)
}
return 0, fmt.Errorf("CapEff not found in /proc/self/status")
}
// EnsureIPForward enables IPv4 forwarding (required for sandbox NAT) and
// verifies the resulting value. The write may fail when the caller lacks
// DAC_OVERRIDE on /proc/sys — that is tolerated as long as forwarding was
// already enabled externally (e.g. by a systemd unit's ExecStartPre).
func EnsureIPForward() error {
_ = os.WriteFile("/proc/sys/net/ipv4/ip_forward", []byte("1"), 0644)
b, err := os.ReadFile("/proc/sys/net/ipv4/ip_forward")
if err != nil {
return fmt.Errorf("read ip_forward: %w", err)
}
if strings.TrimSpace(string(b)) != "1" {
return fmt.Errorf("ip_forward is not enabled — sandbox networking will be broken")
}
return nil
}
// SetupOptions controls which phases of the host startup ritual run.
type SetupOptions struct {
// WrennDir is the root data directory (e.g. /var/lib/wrenn).
WrennDir string
// CHBin is the cloud-hypervisor binary path; empty → DefaultCHBin.
CHBin string
// DefaultRootfsSizeMB is forwarded to EnsureImageSizes; 0 → DefaultDiskSizeMB.
DefaultRootfsSizeMB int
// SkipCleanup skips the stale-resource sweep (kill orphaned CH processes,
// remove stale dm devices and network namespaces).
//
// The sweep assumes this process is the only sandbox owner on the host —
// it kills EVERY cloud-hypervisor process not tracked by this process.
// A short-lived CLI re-attaching to sandboxes created by earlier
// invocations MUST set this, or it will destroy its own running VMs.
SkipCleanup bool
// SkipEnsureImages skips expanding base images to the configured disk size.
SkipEnsureImages bool
}
// SetupResult holds environment facts resolved by Setup, ready to be copied
// into Config.
type SetupResult struct {
KernelPath string
KernelVersion string
CHBin string
CHVersion string
}
// Setup performs the host startup ritual shared by the host agent and
// standalone library consumers: stale-resource cleanup, base-image expansion,
// kernel resolution, and cloud-hypervisor version detection.
//
// Call CheckPrivileges and EnsureIPForward first; they are separate because
// callers surface their failures differently.
func Setup(opts SetupOptions) (*SetupResult, error) {
if opts.CHBin == "" {
opts.CHBin = DefaultCHBin
}
if !opts.SkipCleanup {
// Order matters: kill stale CH processes first — they hold dm-snapshot
// devices open and would otherwise cause "Device or resource busy" on
// dmsetup remove.
vm.CleanupStaleProcesses()
devicemapper.CleanupStaleDevices()
devicemapper.LogLoopState()
network.CleanupStaleNamespaces()
// The sweep above killed every sandbox process on the host, so all
// slot claim files are stale. Remove them; live claims are
// re-established when paused/running sandboxes are restored.
if err := os.RemoveAll(layout.SlotsDir(opts.WrennDir)); err != nil {
return nil, fmt.Errorf("reset slot claims: %w", err)
}
}
if !opts.SkipEnsureImages {
// Expand base images to the configured disk size (sparse, no extra
// physical disk) so dm-snapshot sandboxes see the full size from boot.
if err := EnsureImageSizes(opts.WrennDir, opts.DefaultRootfsSizeMB); err != nil {
return nil, fmt.Errorf("expand base images: %w", err)
}
}
kernelPath, kernelVersion, err := layout.LatestKernel(opts.WrennDir)
if err != nil {
return nil, fmt.Errorf("find kernel: %w", err)
}
chVersion, err := DetectCHVersion(opts.CHBin)
if err != nil {
return nil, fmt.Errorf("detect cloud-hypervisor version: %w", err)
}
// Remove any *.staging-* / *.trash-* directories left behind by a
// previous Pause that crashed before completing the atomic swap. Must
// run before any Resume so we don't race a sandbox restoration.
CleanupOrphanPauseDirs(opts.WrennDir)
return &SetupResult{
KernelPath: kernelPath,
KernelVersion: kernelVersion,
CHBin: opts.CHBin,
CHVersion: chVersion,
}, nil
}

152
pkg/sandbox/state_file.go Normal file
View File

@ -0,0 +1,152 @@
package sandbox
import (
"encoding/json"
"fmt"
"log/slog"
"os"
"path/filepath"
"time"
"github.com/jackc/pgx/v5/pgtype"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
"git.omukk.dev/wrenn/wrenn/pkg/vm"
)
// uuidString16 renders a raw 16-byte UUID as the standard hyphenated string.
func uuidString16(b [16]byte) string {
return id.UUIDString(pgtype.UUID{Bytes: b, Valid: true})
}
// runningStateFile is the per-sandbox state file that records everything a
// fresh process needs to re-attach to a RUNNING sandbox created by an earlier
// process (see RestoreRunningSandboxes). It lives in the sandbox dir alongside
// rootfs.cow, so every existing teardown path already removes it:
//
// - Pause's swapDir replaces the dir with a staging dir that lacks it
// - Destroy's os.RemoveAll(SandboxDir) wipes it with the dir
// - cleanupAfterCrash's os.RemoveAll(SandboxDir) likewise
//
// It is therefore only present while the sandbox is (or last was) running.
const runningStateFile = "sandbox.json"
// runningStateVersion is bumped on breaking schema changes; readers refuse
// unknown versions rather than mis-restoring.
const runningStateVersion = 1
// runningState is the on-disk schema of runningStateFile.
type runningState struct {
Version int `json:"version"`
ID string `json:"id"`
TeamID string `json:"team_id"`
TemplateID string `json:"template_id"`
VCPUs int `json:"vcpus"`
MemoryMB int `json:"memory_mb"`
TimeoutSec int `json:"timeout_sec"`
SlotIndex int `json:"slot_index"`
// BaseImagePath is the base template rootfs backing the dm-snapshot.
BaseImagePath string `json:"base_image_path"`
CowPath string `json:"cow_path"`
DMName string `json:"dm_name"`
// SandboxDir is the tmpfs path baked into CH's config (effectiveSandboxDir) —
// differs from vm.SandboxTmpDir(ID) for sandboxes launched from snapshot
// templates.
SandboxDir string `json:"sandbox_dir"`
// CHPID is the PID of the unshare wrapper; the exec chain keeps the same
// PID, so it is also the cloud-hypervisor PID.
CHPID int `json:"ch_pid"`
CHSocket string `json:"ch_socket"`
// SandboxDirOverride is non-empty when the sandbox was launched from a
// snapshot template (see sandboxState.sandboxDirOverride).
SandboxDirOverride string `json:"sandbox_dir_override,omitempty"`
// LazyRestore mirrors sandboxState.lazyRestore: the VM was restored with
// memory_restore_mode=ondemand, so a re-attaching process must verify the
// guest memory preload before any pause/snapshot.
LazyRestore bool `json:"lazy_restore,omitempty"`
CreatedAt time.Time `json:"created_at"`
Metadata map[string]string `json:"metadata,omitempty"`
}
// writeRunningState persists sb's re-attach state next to its CoW file.
// Best-effort by design: the sandbox is already running and functional, so a
// failed write only means a later process cannot re-attach it. Must be called
// after the VM is registered in m.vm (the PID is read from there).
func (m *Manager) writeRunningState(sb *sandboxState) {
v, ok := m.vm.Get(sb.ID)
if !ok {
slog.Warn("running state: VM not found, skipping persist", "id", sb.ID)
return
}
st := runningState{
Version: runningStateVersion,
ID: sb.ID,
TeamID: uuidString16(sb.TemplateTeamID),
TemplateID: uuidString16(sb.TemplateID),
VCPUs: sb.VCPUs,
MemoryMB: sb.MemoryMB,
TimeoutSec: sb.TimeoutSec,
SlotIndex: sb.SlotIndex,
CowPath: sb.dmDevice.CowPath,
DMName: sb.dmDevice.Name,
BaseImagePath: sb.baseImagePath,
SandboxDir: effectiveSandboxDir(sb),
CHPID: v.PID(),
CHSocket: vm.SandboxSocketPath(sb.ID),
SandboxDirOverride: sb.sandboxDirOverride,
LazyRestore: sb.lazyRestore,
CreatedAt: sb.CreatedAt,
Metadata: sb.Metadata,
}
data, err := json.MarshalIndent(st, "", " ")
if err != nil {
slog.Warn("running state: marshal failed", "id", sb.ID, "error", err)
return
}
// Atomic write (tmp + rename) so a concurrent reader never sees a torn file.
path := runningStatePath(m.cfg.WrennDir, sb.ID)
tmp := path + ".tmp"
if err := os.WriteFile(tmp, data, 0o644); err != nil {
slog.Warn("running state: write failed", "id", sb.ID, "error", err)
return
}
if err := os.Rename(tmp, path); err != nil {
_ = os.Remove(tmp)
slog.Warn("running state: rename failed", "id", sb.ID, "error", err)
}
}
// readRunningState loads and validates the running-state file for a sandbox.
func readRunningState(wrennDir, sandboxID string) (*runningState, error) {
data, err := os.ReadFile(runningStatePath(wrennDir, sandboxID))
if err != nil {
return nil, err
}
var st runningState
if err := json.Unmarshal(data, &st); err != nil {
return nil, fmt.Errorf("unmarshal running state: %w", err)
}
if st.Version != runningStateVersion {
return nil, fmt.Errorf("unsupported running state version %d (want %d)", st.Version, runningStateVersion)
}
if st.ID != sandboxID {
return nil, fmt.Errorf("running state id mismatch: file says %q, dir is %q", st.ID, sandboxID)
}
return &st, nil
}
// deleteRunningState removes the running-state file. Most teardown paths do
// not need this (they remove or swap the whole sandbox dir); it exists for
// the dead-sandbox path in RestoreRunningSandboxes, which must invalidate the
// file while preserving a possible pause snapshot in the same dir.
func deleteRunningState(wrennDir, sandboxID string) {
_ = os.Remove(runningStatePath(wrennDir, sandboxID))
}
func runningStatePath(wrennDir, sandboxID string) string {
return filepath.Join(layout.SandboxDir(wrennDir, sandboxID), runningStateFile)
}

View File

@ -0,0 +1,111 @@
package sandbox
import (
"encoding/json"
"os"
"path/filepath"
"testing"
"time"
"git.omukk.dev/wrenn/wrenn/pkg/layout"
)
func writeTestRunningState(t *testing.T, wrennDir string, st *runningState) {
t.Helper()
dir := layout.SandboxDir(wrennDir, st.ID)
if err := os.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
data, err := json.Marshal(st)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dir, runningStateFile), data, 0o644); err != nil {
t.Fatal(err)
}
}
func TestRunningStateRoundtrip(t *testing.T) {
wrennDir := t.TempDir()
want := &runningState{
Version: runningStateVersion,
ID: "sb-cafe0123",
TeamID: "00000000-0000-0000-0000-000000000000",
TemplateID: "00000000-0000-0000-0000-000000000001",
VCPUs: 2,
MemoryMB: 1024,
TimeoutSec: 300,
SlotIndex: 7,
BaseImagePath: "/var/lib/wrenn/images/teams/x/y/rootfs.ext4",
CowPath: "/var/lib/wrenn/sandboxes/sb-cafe0123/rootfs.cow",
DMName: "wrenn-sb-cafe0123",
SandboxDir: "/tmp/ch-vm-sb-cafe0123",
CHPID: 4242,
CHSocket: "/tmp/ch-sb-cafe0123.sock",
SandboxDirOverride: "/tmp/ch-vm-sb-original",
LazyRestore: true,
CreatedAt: time.Now().Truncate(0),
Metadata: map[string]string{"kernel_version": "6.1.102"},
}
writeTestRunningState(t, wrennDir, want)
got, err := readRunningState(wrennDir, want.ID)
if err != nil {
t.Fatalf("readRunningState: %v", err)
}
if got.ID != want.ID || got.SlotIndex != want.SlotIndex ||
got.CHPID != want.CHPID || got.DMName != want.DMName ||
got.SandboxDirOverride != want.SandboxDirOverride ||
got.LazyRestore != want.LazyRestore ||
!got.CreatedAt.Equal(want.CreatedAt) ||
got.Metadata["kernel_version"] != want.Metadata["kernel_version"] {
t.Fatalf("roundtrip mismatch:\ngot %+v\nwant %+v", got, want)
}
}
func TestRunningStateVersionMismatch(t *testing.T) {
wrennDir := t.TempDir()
st := &runningState{Version: runningStateVersion + 1, ID: "sb-deadbeef", SlotIndex: 1}
writeTestRunningState(t, wrennDir, st)
if _, err := readRunningState(wrennDir, st.ID); err == nil {
t.Fatal("readRunningState should reject unknown version")
}
}
func TestRunningStateIDMismatch(t *testing.T) {
wrennDir := t.TempDir()
st := &runningState{Version: runningStateVersion, ID: "sb-other", SlotIndex: 1}
// Write the file under a directory whose name does not match st.ID.
dir := layout.SandboxDir(wrennDir, "sb-dirname")
if err := os.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
data, _ := json.Marshal(st)
if err := os.WriteFile(filepath.Join(dir, runningStateFile), data, 0o644); err != nil {
t.Fatal(err)
}
if _, err := readRunningState(wrennDir, "sb-dirname"); err == nil {
t.Fatal("readRunningState should reject id mismatch")
}
}
func TestRunningStateMissingFile(t *testing.T) {
if _, err := readRunningState(t.TempDir(), "sb-none"); !os.IsNotExist(err) {
t.Fatalf("want os.IsNotExist error, got %v", err)
}
}
func TestDeleteRunningState(t *testing.T) {
wrennDir := t.TempDir()
st := &runningState{Version: runningStateVersion, ID: "sb-gone", SlotIndex: 1}
writeTestRunningState(t, wrennDir, st)
deleteRunningState(wrennDir, st.ID)
if _, err := readRunningState(wrennDir, st.ID); !os.IsNotExist(err) {
t.Fatalf("state file should be gone, got %v", err)
}
// Idempotent.
deleteRunningState(wrennDir, st.ID)
}

View File

@ -109,10 +109,10 @@ func (s *BuildService) Create(ctx context.Context, p BuildCreateParams) (db.Temp
p.BaseTemplate = "minimal-ubuntu"
}
if p.VCPUs <= 0 {
p.VCPUs = 1
p.VCPUs = 2
}
if p.MemoryMB <= 0 {
p.MemoryMB = 512
p.MemoryMB = 2048
}
// Assemble the recipe. Unless run_as_root is set, the non-root build user

View File

@ -54,14 +54,14 @@ type SandboxCreateParams struct {
Metadata map[string]string
}
// MinTimeoutSec mirrors internal/sandbox.MinTimeoutSec. Sub-minute TTLs race
// MinTimeoutSec mirrors pkg/sandbox.MinTimeoutSec. Sub-minute TTLs race
// the post-create startup window (DB insert → /init → memory loader); the
// agent silently clamps anyway, but the CP must clamp too so the DB record
// agrees with what the agent runs. 0 is preserved (no TTL).
const MinTimeoutSec int32 = 60
// clampTimeout normalises a caller-supplied TTL the same way the host agent
// does. Keep in sync with internal/sandbox.clampTimeout.
// does. Keep in sync with pkg/sandbox.clampTimeout.
func clampTimeout(timeoutSec int32) int32 {
if timeoutSec <= 0 {
return 0
@ -132,10 +132,10 @@ func (s *SandboxService) Create(ctx context.Context, p SandboxCreateParams) (db.
return db.Sandbox{}, fmt.Errorf("invalid metadata: %w", err)
}
if p.VCPUs <= 0 {
p.VCPUs = 1
p.VCPUs = 2
}
if p.MemoryMB <= 0 {
p.MemoryMB = 512
p.MemoryMB = 2048
}
p.TimeoutSec = clampTimeout(p.TimeoutSec)

View File

@ -12,6 +12,7 @@ import (
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
"git.omukk.dev/wrenn/wrenn/pkg/auth/session"
"git.omukk.dev/wrenn/wrenn/pkg/db"
"git.omukk.dev/wrenn/wrenn/pkg/id"
"git.omukk.dev/wrenn/wrenn/pkg/lifecycle"
@ -25,6 +26,10 @@ type TeamService struct {
DB *db.Queries
Pool *pgxpool.Pool
HostPool *lifecycle.HostClientPool
// Sessions drops cached sessions when a member is removed so their access
// is stripped at the next request. Optional: nil in contexts without a
// session store (revocation is then skipped).
Sessions *session.Service
}
// TeamWithRole pairs a team with the calling user's role in it.
@ -378,6 +383,26 @@ func (s *TeamService) RemoveMember(ctx context.Context, teamID, callerUserID, ta
}); err != nil {
return fmt.Errorf("delete member: %w", err)
}
// Revoke the removed member's standing access to this team immediately.
// Deleting the membership row alone leaves two credentials live: the
// team-scoped API keys they created, and any cached session still holding
// this team_id. Purge the keys, and drop the session cache so the next
// request rehydrates from Postgres (where the membership is now gone) —
// hydrateFromDB then clears the stale team binding.
if err := s.DB.DeleteAPIKeysByTeamAndCreator(ctx, db.DeleteAPIKeysByTeamAndCreatorParams{
TeamID: teamID,
CreatedBy: targetUserID,
}); err != nil {
slog.Warn("failed to delete API keys for removed member",
"team_id", teamID, "user_id", targetUserID, "error", err)
}
if s.Sessions != nil {
if err := s.Sessions.InvalidateCacheForUser(ctx, targetUserID); err != nil {
slog.Warn("failed to invalidate session cache for removed member",
"user_id", targetUserID, "error", err)
}
}
return nil
}

View File

@ -22,8 +22,9 @@ const (
// not be set by users — they carry the immutable system facts about the VM.
// Keep in sync with (*sandbox.Manager).buildMetadata: if a key is added there
// but not here, a user could set it at create-time and watch it silently get
// overwritten by the agent on boot. (validate cannot import internal/sandbox,
// hence the duplication — same pattern as service.MinTimeoutSec.)
// overwritten by the agent on boot. (validate deliberately does not import
// pkg/sandbox — it would drag the whole host runtime into the control plane's
// validation layer — hence the duplication; same pattern as service.MinTimeoutSec.)
var reservedMetadataKeys = map[string]struct{}{
"kernel_version": {},
"vmm_version": {},

View File

@ -7,6 +7,7 @@ import (
"os"
"strings"
"sync"
"syscall"
"time"
)
@ -250,7 +251,7 @@ func (m *Manager) CreateFromSnapshot(ctx context.Context, cfg VMConfig) (*VM, er
// PID returns the process ID of the unshare wrapper process.
func (v *VM) PID() int {
return v.process.cmd.Process.Pid
return v.process.pid()
}
// Exited returns a channel that is closed when the VM process exits.
@ -258,6 +259,56 @@ func (v *VM) Exited() <-chan struct{} {
return v.process.exited()
}
// reattachPollInterval is how often a re-attached (non-child) CH process is
// probed for liveness. A non-child cannot be Wait()ed, so exit detection is
// kill(pid, 0) polling.
const reattachPollInterval = 2 * time.Second
// Reattach registers a Cloud Hypervisor process that was started by an
// earlier process (e.g. a previous CLI invocation) so this Manager can
// operate on it — Destroy, Pause, Snapshot, Info all work as usual through
// the still-live API socket. The caller must have verified the PID belongs
// to the expected CH process; Reattach only confirms the API socket answers.
func (m *Manager) Reattach(ctx context.Context, cfg VMConfig, pid int) (*VM, error) {
cfg.applyDefaults()
proc := &process{
attachedPID: pid,
exitCh: make(chan struct{}),
}
client := newCHClient(cfg.SocketPath)
// Probe the CH API before registering — a dead socket means the state
// file is stale and the caller should clean up instead.
if _, err := client.vmInfo(ctx); err != nil {
return nil, fmt.Errorf("vm.info probe: %w", err)
}
go func() {
ticker := time.NewTicker(reattachPollInterval)
defer ticker.Stop()
for range ticker.C {
if err := syscall.Kill(pid, 0); err != nil {
close(proc.exitCh)
return
}
}
}()
vm := &VM{
Config: cfg,
process: proc,
client: client,
}
m.mu.Lock()
m.vms[cfg.SandboxID] = vm
m.mu.Unlock()
slog.Info("re-attached to running VM", "sandbox", cfg.SandboxID, "pid", pid)
return vm, nil
}
// Get returns a running VM by sandbox ID.
func (m *Manager) Get(sandboxID string) (*VM, bool) {
m.mu.RLock()

View File

@ -20,6 +20,19 @@ type process struct {
exitCh chan struct{}
exitErr error
logFile *os.File
// attachedPID is set instead of cmd when the CH process was started by an
// earlier process and re-attached via Manager.Reattach. A non-child cannot
// be cmd.Wait()ed; exit detection falls back to kill(pid, 0) polling.
attachedPID int
}
// pid returns the process ID regardless of how the process was obtained.
func (p *process) pid() int {
if p.cmd != nil && p.cmd.Process != nil {
return p.cmd.Process.Pid
}
return p.attachedPID
}
// startProcess launches the Cloud Hypervisor binary inside an isolated mount
@ -145,13 +158,15 @@ exec %[4]s
}
// stop sends SIGTERM and waits for the process to exit. If it doesn't exit
// within 10 seconds, SIGKILL is sent.
// within 10 seconds, SIGKILL is sent. Signals target the process group
// (Setsid at launch), which works for both child and re-attached processes.
func (p *process) stop() error {
if p.cmd.Process == nil {
pid := p.pid()
if pid == 0 {
return nil
}
if err := syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM); err != nil {
if err := syscall.Kill(-pid, syscall.SIGTERM); err != nil {
slog.Debug("sigterm failed, process may have exited", "error", err)
}
@ -160,7 +175,7 @@ func (p *process) stop() error {
return nil
case <-time.After(10 * time.Second):
slog.Warn("cloud-hypervisor did not exit after SIGTERM, sending SIGKILL")
if err := syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL); err != nil {
if err := syscall.Kill(-pid, syscall.SIGKILL); err != nil {
slog.Debug("sigkill failed", "error", err)
}
<-p.exitCh