Skip to main content

boatramp_node/
compute.rs

1//! Compute-backend assembly (moved from the binary — node-library N2).
2//!
3//! Builds this node's compute [`BackendRegistry`](boatramp_core::compute::BackendRegistry)
4//! and scheduler [`Node`](boatramp_core::compute::Node) inventory from the optional
5//! `[compute]` config, capability-detecting Docker, native-container, and
6//! embedded-VMM backends. Lives here (not in the backend-agnostic
7//! `boatramp-server`) because it depends on the concrete backend crates; the
8//! binary and future `assemble()` call [`build_compute`].
9
10use std::sync::Arc;
11
12/// The posture-scaled kernel-trust gate wired into the compute backends: it runs
13/// [`boatramp_core::kernel_trust::verify_kernel`] on the staged kernel right
14/// before boot. The always-on check is the content hash; under the strict
15/// (multi-tenant) posture it additionally requires the pinned hash to be on the
16/// static allow-list and to carry a signature — sourced from the **live fleet
17/// default kernel** — verifying against a static signing key. No daemon, or a hash
18/// that isn't the current signed default, has no signature source and so **fails
19/// closed** under strict: the kernel does not boot.
20#[cfg(target_os = "linux")]
21struct PostureKernelVerifier {
22    strict: bool,
23    signing_keys: Vec<String>,
24    allowed_hashes: Vec<String>,
25    daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
26}
27
28// `KernelVerifier` requires `Debug`, but `DaemonRuntime` isn't `Debug` (it holds a
29// lock + a `Notify`); summarise instead of recursing into it.
30#[cfg(target_os = "linux")]
31impl std::fmt::Debug for PostureKernelVerifier {
32    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33        f.debug_struct("PostureKernelVerifier")
34            .field("strict", &self.strict)
35            .field("signing_keys", &self.signing_keys.len())
36            .field("allowed_hashes", &self.allowed_hashes.len())
37            .field("has_daemon", &self.daemon.is_some())
38            .finish()
39    }
40}
41
42#[cfg(target_os = "linux")]
43impl boatramp_firecracker::KernelVerifier for PostureKernelVerifier {
44    // Fully-qualified: this module aliases `Result<T>` to its own error type.
45    fn verify(&self, bytes: &[u8], expected_hash: &str) -> std::result::Result<(), String> {
46        // The only signature we trust for this hash is the one on the current
47        // fleet default kernel (the operator-vetted kernel); any other hash has no
48        // signature source and fails the strict bar.
49        let sig = self
50            .daemon
51            .as_ref()
52            .and_then(|d| d.effective().default_kernel.clone())
53            .filter(|dk| dk.sha256 == expected_hash)
54            .and_then(|dk| dk.sig);
55        let kref = boatramp_core::daemon_config::KernelRef {
56            source: expected_hash.to_string(),
57            sha256: expected_hash.to_string(),
58            sig,
59        };
60        boatramp_core::kernel_trust::verify_kernel(
61            bytes,
62            &kref,
63            self.strict,
64            &self.signing_keys,
65            &self.allowed_hashes,
66        )
67        .map_err(|e| e.to_string())
68    }
69}
70
71/// The macOS-VMM twin of [`PostureKernelVerifier`], implementing
72/// [`boatramp_vz::KernelVerifier`] with the identical posture-scaled trust logic
73/// so the Virtualization.framework backend enforces the same verify-before-boot
74/// bar as the KVM backend (the kernel is ring-0 code on either substrate).
75#[cfg(target_os = "macos")]
76struct VzPostureKernelVerifier {
77    strict: bool,
78    signing_keys: Vec<String>,
79    allowed_hashes: Vec<String>,
80    daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
81}
82
83#[cfg(target_os = "macos")]
84impl std::fmt::Debug for VzPostureKernelVerifier {
85    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86        f.debug_struct("VzPostureKernelVerifier")
87            .field("strict", &self.strict)
88            .field("signing_keys", &self.signing_keys.len())
89            .field("allowed_hashes", &self.allowed_hashes.len())
90            .field("has_daemon", &self.daemon.is_some())
91            .finish()
92    }
93}
94
95#[cfg(target_os = "macos")]
96impl boatramp_vz::KernelVerifier for VzPostureKernelVerifier {
97    fn verify(&self, bytes: &[u8], expected_hash: &str) -> std::result::Result<(), String> {
98        let sig = self
99            .daemon
100            .as_ref()
101            .and_then(|d| d.effective().default_kernel.clone())
102            .filter(|dk| dk.sha256 == expected_hash)
103            .and_then(|dk| dk.sig);
104        let kref = boatramp_core::daemon_config::KernelRef {
105            source: expected_hash.to_string(),
106            sha256: expected_hash.to_string(),
107            sig,
108        };
109        boatramp_core::kernel_trust::verify_kernel(
110            bytes,
111            &kref,
112            self.strict,
113            &self.signing_keys,
114            &self.allowed_hashes,
115        )
116        .map_err(|e| e.to_string())
117    }
118}
119
120/// Whether this host can run the macOS VMM backend: **Apple silicon** (arm64) on
121/// **macOS 15+** (the Virtualization.framework Linux-container floor). Detected via
122/// `sysctl` — `hw.optional.arm64 == 1` and `kern.osproductversion >= 15`. macOS 26
123/// is recommended (macOS 15 lacks container-to-container networking over vmnet),
124/// but single-node serve works on 15, so 15 is the floor; the operator's macOS
125/// version determines multi-replica cross-VM reachability, not the user surface.
126#[cfg(target_os = "macos")]
127fn macos_supports_vz() -> bool {
128    // Apple silicon: the Virtualization.framework Linux path is arm64-only.
129    if cfg!(not(target_arch = "aarch64")) {
130        return false;
131    }
132    let major = sysctl_string("kern.osproductversion")
133        .and_then(|v| v.split('.').next().and_then(|m| m.parse::<u32>().ok()));
134    matches!(major, Some(m) if m >= 15)
135}
136
137/// Read a string `sysctl` by name (e.g. `kern.osproductversion`). `None` on any
138/// failure — the caller treats an unreadable sysctl as "unsupported" (fail-closed).
139#[cfg(target_os = "macos")]
140fn sysctl_string(name: &str) -> Option<String> {
141    let out = std::process::Command::new("sysctl")
142        .args(["-n", name])
143        .output()
144        .ok()?;
145    if !out.status.success() {
146        return None;
147    }
148    Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
149}
150
151/// Build this node's compute [`BackendRegistry`] + scheduler [`Node`] inventory
152/// from the optional `[compute]` config. Backends
153/// are **capability-detected**: a reachable Docker daemon ⇒ `docker`; Linux ⇒ the
154/// native `container` backend; Linux + `/dev/kvm` ⇒ the in-process
155/// `vmm-embedded` microVM backend (strongest isolation). Absent config ⇒ an empty
156/// registry + a node advertising nothing, so the reconcile loop stays a no-op.
157pub async fn build_compute(
158    cfg: Option<&crate::config::ComputeConfig>,
159    storage: std::sync::Arc<dyn boatramp_core::Storage>,
160    data_dir: &std::path::Path,
161    node_id: u64,
162    strict: bool,
163    daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
164    // The binary the re-exec'd container/microVM workers run as; `None` ⇒ this
165    // process's own executable (`current_exe`). An embedding harness points it at a
166    // built `boatramp` binary so the workers find the `__sandbox`/`__vmm-run`/
167    // `__vz-run` subcommands. See [`crate::node::NodeInput::worker_exe`].
168    worker_exe: Option<&std::path::Path>,
169) -> (
170    boatramp_core::compute::BackendRegistry,
171    boatramp_core::compute::Node,
172) {
173    use boatramp_core::compute::{BackendKind, BackendRegistry, Node};
174    let mut backends: BackendRegistry = std::collections::BTreeMap::new();
175    let empty_node = |id| Node {
176        id,
177        region: None,
178        labels: std::collections::BTreeMap::new(),
179        free_vcpus: 0,
180        free_mem_mib: 0,
181        backends: Vec::new(),
182    };
183    let Some(cfg) = cfg else {
184        return (backends, empty_node(node_id));
185    };
186
187    // Remote docker: register only if a daemon actually answers.
188    match boatramp_docker::DockerBackend::connect() {
189        Ok(docker) => {
190            // `writable_root` and `cap_add` are honored only under the single-tenant
191            // posture (`!strict`); the multi-tenant guard keeps the hardened read-only
192            // root and every capability dropped.
193            let docker = docker
194                .with_endpoint(cfg.docker_endpoint)
195                .with_volume_mode(cfg.docker_volume_mode)
196                .with_data_dir(data_dir)
197                .with_writable_root_allowed(!strict)
198                .with_cap_add_allowed(!strict);
199            if docker.reachable().await {
200                backends.insert("docker".to_string(), std::sync::Arc::new(docker));
201            } else {
202                tracing::debug!("no reachable docker daemon; skipping docker backend");
203            }
204        }
205        Err(e) => tracing::debug!(%e, "docker backend unavailable"),
206    }
207
208    // Ensure the shared compute bridge exists before the backends that enslave a
209    // veth/tap to it. boatramp creates it itself over netlink (needs `CAP_NET_ADMIN`)
210    // rather than requiring the operator to pre-create it — so a stock image on a fresh
211    // host is turnkey. If it can't be created, the container + embedded-VMM backends are
212    // skipped rather than advertised and then failing at launch on the missing bridge.
213    #[cfg(target_os = "linux")]
214    let bridge_ready = match boatramp_core::ipam::IpPool::new(&cfg.subnet) {
215        Ok(pool) => {
216            match boatramp_container::ensure_bridge(&cfg.bridge, pool.gateway(), pool.prefix_len())
217                .await
218            {
219                Ok(()) => true,
220                Err(e) => {
221                    tracing::warn!(%e, bridge = %cfg.bridge, "could not create the compute bridge (need CAP_NET_ADMIN); container + embedded-VMM backends disabled");
222                    false
223                }
224            }
225        }
226        Err(e) => {
227            tracing::warn!(%e, subnet = %cfg.subnet, "bad compute subnet; container + embedded-VMM backends disabled");
228            false
229        }
230    };
231
232    // Native container backend (Linux only).
233    #[cfg(target_os = "linux")]
234    if bridge_ready {
235        match worker_exe.map_or_else(std::env::current_exe, |p| Ok(p.to_path_buf())) {
236            Ok(self_exe) => match boatramp_container::ContainerBackend::new(
237                storage.clone(),
238                data_dir.to_path_buf(),
239                cfg.bridge.clone(),
240                &cfg.subnet,
241                self_exe,
242            ) {
243                Ok(c) => {
244                    // Single-tenant posture (`!strict`) may honor `cap_add`; multi-tenant
245                    // keeps every capability dropped.
246                    let c = c.with_cap_add_allowed(!strict);
247                    backends.insert("container".to_string(), std::sync::Arc::new(c));
248                }
249                Err(e) => tracing::warn!(%e, "container backend unavailable"),
250            },
251            Err(e) => tracing::warn!(%e, "current_exe for container backend"),
252        }
253    }
254    // Embedded VMM backend (Linux + x86_64 + `/dev/kvm`): in-process microVMs, no
255    // external `firecracker` binary — the strongest isolation when KVM is available.
256    // Like the container backend it enslaves each tap to `cfg.bridge` (ensured above,
257    // hence the `bridge_ready` gate). The embedded VMM is KVM-x86-specific, so this is
258    // x86_64-only; boatramp
259    // still serves on linux/aarch64 (with the container backend, no embedded VMM).
260    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
261    if bridge_ready && std::path::Path::new("/dev/kvm").exists() {
262        match (
263            worker_exe.map_or_else(std::env::current_exe, |p| Ok(p.to_path_buf())),
264            boatramp_core::ipam::IpPool::new(&cfg.subnet),
265        ) {
266            (Ok(self_exe), Ok(pool)) => {
267                let gateway = pool.gateway().to_string();
268                // Verify-before-boot gate for every kernel this backend stages.
269                let verifier: Arc<dyn boatramp_firecracker::KernelVerifier> =
270                    Arc::new(PostureKernelVerifier {
271                        strict,
272                        signing_keys: cfg.kernel_signing_pubkeys.clone(),
273                        allowed_hashes: cfg.kernel_allowed_hashes.clone(),
274                        daemon: daemon.clone(),
275                    });
276                match boatramp_firecracker::EmbeddedVmmBackend::new(
277                    storage.clone(),
278                    self_exe, // re-exec'd as `__vmm-run` per VM (jailed subprocess)
279                    data_dir.to_path_buf(),
280                    cfg.bridge.clone(),
281                    gateway,
282                    &cfg.subnet,
283                    verifier,
284                ) {
285                    Ok(vmm) => {
286                        backends.insert("vmm-embedded".to_string(), std::sync::Arc::new(vmm));
287                    }
288                    Err(e) => tracing::warn!(%e, "embedded VMM backend unavailable"),
289                }
290            }
291            (Err(e), _) => tracing::warn!(%e, "current_exe for VMM backend"),
292            (_, Err(e)) => tracing::warn!(%e, "bad compute subnet for VMM backend"),
293        }
294    } else {
295        tracing::debug!("no /dev/kvm; skipping embedded VMM backend");
296    }
297
298    // macOS-native VMM backend (Apple silicon + macOS): each replica is a Linux
299    // microVM under Virtualization.framework, run by a re-exec'd `__vz-run`
300    // worker. Strong isolation (VmKvm), matching the KVM backend's user surface.
301    // Capability-detected + log-skipped on Intel / older macOS, exactly like the
302    // `/dev/kvm` check gates the Linux VMM.
303    #[cfg(target_os = "macos")]
304    if macos_supports_vz() {
305        match (
306            worker_exe.map_or_else(std::env::current_exe, |p| Ok(p.to_path_buf())),
307            boatramp_core::ipam::IpPool::new(&cfg.subnet),
308        ) {
309            (Ok(self_exe), Ok(_pool)) => {
310                let verifier: Arc<dyn boatramp_vz::KernelVerifier> =
311                    Arc::new(VzPostureKernelVerifier {
312                        strict,
313                        signing_keys: cfg.kernel_signing_pubkeys.clone(),
314                        allowed_hashes: cfg.kernel_allowed_hashes.clone(),
315                        daemon: daemon.clone(),
316                    });
317                match boatramp_vz::VzBackend::new(
318                    storage.clone(),
319                    self_exe, // re-exec'd as `__vz-run` per VM
320                    data_dir.to_path_buf(),
321                    &cfg.subnet, // the vmnet range (e.g. 192.168.64.0/24); `.1` = gateway
322                    verifier,
323                ) {
324                    // `writable_root` honored only under the single-tenant posture.
325                    Ok(vz) => {
326                        let vz = vz.with_writable_root_allowed(!strict);
327                        backends.insert("vmm-vz".to_string(), std::sync::Arc::new(vz));
328                    }
329                    Err(e) => tracing::warn!(%e, "macOS VMM backend unavailable"),
330                }
331            }
332            (Err(e), _) => tracing::warn!(%e, "current_exe for macOS VMM backend"),
333            (_, Err(e)) => tracing::warn!(%e, "bad compute subnet for macOS VMM backend"),
334        }
335    } else {
336        tracing::debug!("not Apple silicon + macOS 15+; skipping macOS VMM backend");
337    }
338
339    let _ = (&storage, data_dir); // used only on Linux/macOS (container / VMM backends)
340                                  // The kernel-trust verifier is wired for the embedded VMM (x86_64 Linux) and
341                                  // the macOS VMM; silence `strict`/`daemon` on the platforms that wire neither
342                                  // (linux/aarch64, and any non-Linux non-macOS host).
343    #[cfg(not(any(all(target_os = "linux", target_arch = "x86_64"), target_os = "macos")))]
344    let _ = (strict, &daemon);
345
346    let free_vcpus = if cfg.vcpus > 0 {
347        cfg.vcpus
348    } else {
349        std::thread::available_parallelism()
350            .map(|n| n.get() as u32)
351            .unwrap_or(1)
352    };
353    let free_mem_mib = if cfg.mem_mib > 0 { cfg.mem_mib } else { 1024 };
354    let advertised: Vec<BackendKind> = backends
355        .iter()
356        .map(|(id, b)| {
357            let caps = b.capabilities();
358            BackendKind {
359                id: id.clone(),
360                isolation: caps.isolation,
361                persistent_volumes: caps.persistent_volumes,
362                scale_to_zero: caps.scale_to_zero,
363            }
364        })
365        .collect();
366    tracing::info!(backends = ?advertised, free_vcpus, free_mem_mib, "compute node inventory");
367    let node = Node {
368        id: node_id,
369        region: cfg.region.clone(),
370        labels: std::collections::BTreeMap::new(),
371        free_vcpus,
372        free_mem_mib,
373        backends: advertised,
374    };
375    (backends, node)
376}
377
378/// The node's [`ComputeExec`](boatramp_core::compute::ComputeExec): resolve a
379/// workload's running replica from the control-plane state, pick its backend, and
380/// run the command inside it. Backs `POST /api/compute/{name}/exec`; the API gates
381/// it behind the `allow_compute_exec` posture. Only the shared-kernel backends
382/// (native `container`, remote `docker`) actually implement
383/// [`ComputeBackend::exec`](boatramp_core::compute::ComputeBackend::exec); the rest
384/// surface as [`ExecError::Unsupported`](boatramp_core::compute::ExecError).
385pub struct NodeComputeExec {
386    backends: boatramp_core::compute::BackendRegistry,
387    deploy: boatramp_core::deploy::DeployStore,
388}
389
390impl NodeComputeExec {
391    /// Build over this node's compute backends + the control-plane store. The
392    /// registry is a cheap `BTreeMap` of `Arc` backends (clone it before the
393    /// reconcile loop consumes the original).
394    pub fn new(
395        backends: boatramp_core::compute::BackendRegistry,
396        deploy: boatramp_core::deploy::DeployStore,
397    ) -> Self {
398        Self { backends, deploy }
399    }
400}
401
402#[async_trait::async_trait]
403impl boatramp_core::compute::ComputeExec for NodeComputeExec {
404    async fn exec(
405        &self,
406        project: &str,
407        workload: &str,
408        argv: &[String],
409        stdin: Option<&[u8]>,
410    ) -> Result<boatramp_core::compute::ExecOutput, boatramp_core::compute::ExecError> {
411        use boatramp_core::compute::{BackendError, ExecError, ReplicaPhase};
412        use boatramp_core::project::ProjectRef;
413        let states = self
414            .deploy
415            .list_replica_states(ProjectRef::new(project), workload)
416            .await
417            .map_err(|e| ExecError::Other(e.to_string()))?;
418        // A running replica — prefer a healthy one, else any running (a just-launched
419        // DB may not be health-marked yet but can still accept an exec).
420        let target = states
421            .iter()
422            .find(|s| s.phase == ReplicaPhase::Running && s.healthy)
423            .or_else(|| states.iter().find(|s| s.phase == ReplicaPhase::Running))
424            .ok_or_else(|| ExecError::NoReplica(workload.to_string()))?;
425        let backend = self
426            .backends
427            .get(&target.backend)
428            .ok_or_else(|| ExecError::Unsupported(target.backend.clone()))?;
429        match backend.exec(&target.handle, argv, stdin).await {
430            Ok(out) => Ok(out),
431            Err(BackendError::Unsupported) => Err(ExecError::Unsupported(target.backend.clone())),
432            Err(e) => Err(ExecError::Other(e.to_string())),
433        }
434    }
435}
436
437/// The node's [`ComputeVolumes`](boatramp_core::compute::ComputeVolumes): list +
438/// reclaim persistent volumes. Backs `GET /api/compute/volumes` +
439/// `DELETE /api/compute/volumes/{name}` (admin-scoped). Lists every
440/// volume-capable backend's on-node volumes, flags which are still referenced by a
441/// registered workload's active spec (in use vs orphaned), and refuses to remove
442/// an in-use volume unless forced — so `compute rm <workload>` (which unregisters
443/// it, then the reconcile loop stops the replica) is the safe precondition for
444/// reclaiming its volume.
445pub struct NodeComputeVolumes {
446    backends: boatramp_core::compute::BackendRegistry,
447    deploy: boatramp_core::deploy::DeployStore,
448}
449
450impl NodeComputeVolumes {
451    /// Build over this node's compute backends + the control-plane store.
452    pub fn new(
453        backends: boatramp_core::compute::BackendRegistry,
454        deploy: boatramp_core::deploy::DeployStore,
455    ) -> Self {
456        Self { backends, deploy }
457    }
458
459    /// The set of volume names still referenced by **any** registered workload's
460    /// active spec, across every project (the `_all` fan-out). A name in this set
461    /// is "in use": a running or relaunching replica mounts it, so removing its
462    /// backing would corrupt live data. Resolves each workload's content-addressed
463    /// spec to read its `volumes[].name`; a workload whose spec can't be resolved
464    /// is skipped (it can't be actively mounting a volume the backend still backs).
465    async fn referenced_volume_names(
466        &self,
467    ) -> Result<std::collections::BTreeSet<String>, boatramp_core::compute::VolumeError> {
468        use boatramp_core::compute::VolumeError;
469        let mut names = std::collections::BTreeSet::new();
470        let workloads = self
471            .deploy
472            .list_compute_workloads_all()
473            .await
474            .map_err(|e| VolumeError::Other(e.to_string()))?;
475        for (_project, workload) in workloads {
476            let spec = self
477                .deploy
478                .get_compute_spec(&workload.active)
479                .await
480                .map_err(|e| VolumeError::Other(e.to_string()))?;
481            if let Some(spec) = spec {
482                for vol in spec.volumes {
483                    names.insert(vol.name);
484                }
485            }
486        }
487        Ok(names)
488    }
489}
490
491#[async_trait::async_trait]
492impl boatramp_core::compute::ComputeVolumes for NodeComputeVolumes {
493    async fn list(
494        &self,
495    ) -> Result<Vec<boatramp_core::compute::VolumeStatus>, boatramp_core::compute::VolumeError>
496    {
497        use boatramp_core::compute::{VolumeError, VolumeStatus};
498        let referenced = self.referenced_volume_names().await?;
499        // Union the volumes every backend reports (dedup by name — a name is unique
500        // per node's volumes dir). A backend that doesn't back volumes returns the
501        // empty default, so this naturally reduces to the volume-capable backend(s).
502        let mut by_name: std::collections::BTreeMap<String, u64> =
503            std::collections::BTreeMap::new();
504        for backend in self.backends.values() {
505            let vols = backend
506                .list_volumes()
507                .await
508                .map_err(|e| VolumeError::Other(e.to_string()))?;
509            for v in vols {
510                // Keep the largest reported size if two backends somehow name-collide.
511                let slot = by_name.entry(v.name).or_insert(0);
512                *slot = (*slot).max(v.size_bytes);
513            }
514        }
515        Ok(by_name
516            .into_iter()
517            .map(|(name, size_bytes)| VolumeStatus {
518                in_use: referenced.contains(&name),
519                info: boatramp_core::compute::VolumeInfo { name, size_bytes },
520            })
521            .collect())
522    }
523
524    async fn remove(
525        &self,
526        name: &str,
527        force: bool,
528    ) -> Result<bool, boatramp_core::compute::VolumeError> {
529        use boatramp_core::compute::{BackendError, VolumeError};
530        // Safety guard: refuse to pull a volume out from under a registered
531        // workload unless the operator forces it. `compute rm <workload>` first is
532        // the safe flow; `--force` is the disposable-data override.
533        if !force && self.referenced_volume_names().await?.contains(name) {
534            return Err(VolumeError::InUse(name.to_string()));
535        }
536        // Remove from whichever backend owns it. `true` from any backend ⇒ existed.
537        // Every backend reports `Unsupported` ⇒ no volume-capable backend here.
538        let mut existed = false;
539        let mut any_supported = false;
540        for backend in self.backends.values() {
541            match backend.remove_volume(name).await {
542                Ok(removed) => {
543                    any_supported = true;
544                    existed |= removed;
545                }
546                Err(BackendError::Unsupported) => {}
547                Err(e) => return Err(VolumeError::Other(e.to_string())),
548            }
549        }
550        if !any_supported {
551            return Err(VolumeError::Unsupported);
552        }
553        Ok(existed)
554    }
555}
556
557#[cfg(test)]
558mod tests {
559    use super::*;
560    use async_trait::async_trait;
561    use boatramp_core::compute::{
562        Artifact, BackendError, Capabilities, ComputeBackend, ComputeSpec, ComputeVolumes,
563        ComputeWorkload, Health, Instance, InstanceHandle, IsolationClass, IsolationRequirement,
564        LaunchRequest, RestartPolicy, RootSource, VolumeError, VolumeInfo, VolumeRef,
565    };
566    use boatramp_core::deploy::DeployStore;
567    use boatramp_core::project::ProjectRef;
568    use boatramp_core::{ByteStream, GetObject, ObjectMeta, PutMeta, Storage, StorageError};
569    use std::collections::BTreeMap;
570    use std::sync::{Arc, Mutex};
571
572    /// A `Storage` the `DeployStore` never actually reads on the volume paths (the
573    /// spec/workload records live in the KV) — every method is a stub.
574    struct NullStorage;
575    #[async_trait]
576    impl Storage for NullStorage {
577        async fn get(&self, _: &str) -> Result<GetObject, StorageError> {
578            Err(StorageError::NotFound(String::new()))
579        }
580        async fn get_range(
581            &self,
582            _: &str,
583            _: u64,
584            _: Option<u64>,
585        ) -> Result<GetObject, StorageError> {
586            Err(StorageError::unsupported("range"))
587        }
588        async fn put(
589            &self,
590            _: &str,
591            _: ByteStream,
592            _: PutMeta,
593        ) -> Result<ObjectMeta, StorageError> {
594            Err(StorageError::unsupported("put"))
595        }
596        async fn head(&self, _: &str) -> Result<ObjectMeta, StorageError> {
597            Err(StorageError::NotFound(String::new()))
598        }
599        async fn delete(&self, _: &str) -> Result<(), StorageError> {
600            Ok(())
601        }
602        async fn list(&self, _: &str) -> Result<Vec<ObjectMeta>, StorageError> {
603            Ok(Vec::new())
604        }
605    }
606
607    /// A fake volume-capable backend over an in-memory set of `(name, size)`
608    /// volumes — enough to drive `NodeComputeVolumes` without a real container node.
609    struct FakeVolumeBackend {
610        vols: Mutex<BTreeMap<String, u64>>,
611    }
612    impl FakeVolumeBackend {
613        fn with(names: &[(&str, u64)]) -> Self {
614            Self {
615                vols: Mutex::new(names.iter().map(|(n, s)| (n.to_string(), *s)).collect()),
616            }
617        }
618    }
619    #[async_trait]
620    impl ComputeBackend for FakeVolumeBackend {
621        fn id(&self) -> &'static str {
622            "container"
623        }
624        fn capabilities(&self) -> Capabilities {
625            Capabilities {
626                isolation: IsolationClass::Namespace,
627                scale_to_zero: false,
628                persistent_volumes: true,
629                max_vcpus: None,
630                max_mem_mib: None,
631            }
632        }
633        async fn materialize(&self, _: &ComputeSpec) -> Result<Artifact, BackendError> {
634            Err(BackendError::Unsupported)
635        }
636        async fn launch(&self, _: &LaunchRequest) -> Result<Instance, BackendError> {
637            Err(BackendError::Unsupported)
638        }
639        async fn stop(&self, _: &InstanceHandle) -> Result<(), BackendError> {
640            Ok(())
641        }
642        async fn health(&self, _: &InstanceHandle) -> Result<Health, BackendError> {
643            Ok(Health::Unknown)
644        }
645        async fn list_volumes(&self) -> Result<Vec<VolumeInfo>, BackendError> {
646            Ok(self
647                .vols
648                .lock()
649                .unwrap()
650                .iter()
651                .map(|(name, size)| VolumeInfo {
652                    name: name.clone(),
653                    size_bytes: *size,
654                })
655                .collect())
656        }
657        async fn remove_volume(&self, name: &str) -> Result<bool, BackendError> {
658            Ok(self.vols.lock().unwrap().remove(name).is_some())
659        }
660    }
661
662    fn spec_with_volume(vol: Option<&str>) -> ComputeSpec {
663        ComputeSpec {
664            version: 1,
665            root: RootSource::Image("img".into()),
666            kernel: String::new(),
667            kernel_cmdline: None,
668            vcpus: 1,
669            mem_mib: 64,
670            entrypoint: vec![],
671            env: BTreeMap::new(),
672            port: 8080,
673            restart: RestartPolicy::Always,
674            scale_to_zero: false,
675            volumes: vol
676                .map(|n| {
677                    vec![VolumeRef {
678                        mount: "/data".into(),
679                        name: n.into(),
680                        size_mib: 128,
681                    }]
682                })
683                .unwrap_or_default(),
684            writable_root: false,
685            cap_add: vec![],
686            user: None,
687            isolation: IsolationRequirement::Trusted,
688            prefer_backend: None,
689            bindings: vec![],
690        }
691    }
692
693    /// Build a store with one workload named `wl` whose spec references volume
694    /// `referenced` (or none), plus a `NodeComputeVolumes` over a fake backend that
695    /// backs `backend_vols`.
696    async fn setup(referenced: Option<&str>, backend_vols: &[(&str, u64)]) -> NodeComputeVolumes {
697        let store = DeployStore::new(
698            Arc::new(NullStorage),
699            Arc::new(boatramp_core::kv::MemoryKv::new()),
700        );
701        let spec = spec_with_volume(referenced);
702        let hash = store.put_compute_spec(&spec).await.expect("put spec");
703        let workload = ComputeWorkload {
704            version: 1,
705            name: "wl".into(),
706            active: hash,
707            replicas: 1,
708            placement: Default::default(),
709        };
710        store
711            .set_compute_workload(ProjectRef::DEFAULT, &workload)
712            .await
713            .expect("set workload");
714        let mut backends: boatramp_core::compute::BackendRegistry = BTreeMap::new();
715        backends.insert(
716            "container".into(),
717            Arc::new(FakeVolumeBackend::with(backend_vols)) as Arc<dyn ComputeBackend>,
718        );
719        NodeComputeVolumes::new(backends, store)
720    }
721
722    #[tokio::test]
723    async fn list_flags_referenced_volume_in_use_and_orphan_free() {
724        // "data" is referenced by the workload spec; "old" is an orphan.
725        let vols = setup(Some("data"), &[("data", 100), ("old", 50)]).await;
726        let listed = vols.list().await.expect("list");
727        assert_eq!(listed.len(), 2);
728        let data = listed.iter().find(|v| v.info.name == "data").unwrap();
729        let old = listed.iter().find(|v| v.info.name == "old").unwrap();
730        assert!(data.in_use, "spec-referenced volume is in use");
731        assert_eq!(data.info.size_bytes, 100);
732        assert!(!old.in_use, "unreferenced volume is orphaned");
733        assert_eq!(old.info.size_bytes, 50);
734    }
735
736    #[tokio::test]
737    async fn remove_refuses_in_use_without_force_and_allows_with_force() {
738        let vols = setup(Some("data"), &[("data", 100)]).await;
739        // Without force: refused (in use by the registered workload).
740        assert!(matches!(
741            vols.remove("data", false).await,
742            Err(VolumeError::InUse(n)) if n == "data"
743        ));
744        // The volume is still there (refusal didn't remove it).
745        assert!(vols
746            .list()
747            .await
748            .unwrap()
749            .iter()
750            .any(|v| v.info.name == "data"));
751        // With force: removed.
752        assert!(vols.remove("data", true).await.expect("forced remove"));
753        assert!(vols.list().await.unwrap().is_empty());
754    }
755
756    #[tokio::test]
757    async fn remove_orphan_succeeds_and_absent_reports_false() {
758        // No workload references "old"; it removes without force.
759        let vols = setup(None, &[("old", 50)]).await;
760        assert!(vols.remove("old", false).await.expect("remove orphan"));
761        // Removing an absent volume reports "did not exist".
762        assert!(!vols.remove("gone", false).await.expect("remove absent"));
763    }
764}