use std::os::fd::{FromRawFd, OwnedFd};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use arcbox_constants::ports::NFS_NFSD_RELAY_PORT;
use arcbox_core::{DEFAULT_MACHINE_NAME, Runtime, VmLifecycleState};
use arcbox_transport::vsock::{VsockShutdown, VsockStream};
use tokio::io::copy_bidirectional;
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};
use crate::context::DaemonContext;
const MOUNT_TIMEOUT: Duration = Duration::from_secs(30);
const MOUNT_RETRY_INTERVAL: Duration = Duration::from_millis(500);
const UNMOUNT_TIMEOUT: Duration = Duration::from_secs(5);
const RETRY_BACKOFF: Duration = Duration::from_secs(30);
static MOUNTED_PATH: OnceLock<PathBuf> = OnceLock::new();
pub fn spawn(ctx: &DaemonContext, runtime: &Arc<Runtime>) {
if !ctx.mount_nfs {
return;
}
let expect_hosts_alias = ctx.layout.data_dir
== arcbox_constants::paths::HostLayout::resolve_for_profile_from_env(ctx.profile, None)
.data_dir;
let runtime = Arc::clone(runtime);
let shutdown = ctx.shutdown.clone();
tokio::spawn(async move {
if let Err(e) = reconcile(runtime, shutdown, expect_hosts_alias).await {
warn!(error = %e, "failed to establish host NFS mount at ~/ArcBox");
}
});
}
pub async fn cleanup(ctx: &DaemonContext) {
if !ctx.mount_nfs {
return;
}
let Some(mount_path) = MOUNTED_PATH.get() else {
return;
};
match current_mount_info(mount_path) {
Some(info) if is_arcbox_nfs_mount(&info) => match unmount(mount_path).await {
Ok(()) => info!(path = %mount_path.display(), "unmounted ~/ArcBox host NFS mount"),
Err(e) => warn!(path = %mount_path.display(), error = %e, "failed to unmount ~/ArcBox"),
},
_ => {}
}
}
async fn reconcile(
runtime: Arc<Runtime>,
shutdown: CancellationToken,
expect_hosts_alias: bool,
) -> Result<()> {
let Some(mount_path) = resolve_mount_path() else {
bail!("could not determine home directory for ~/ArcBox mount");
};
let nfsd_port = spawn_proxy(&runtime, &shutdown, NFS_NFSD_RELAY_PORT).await?;
let mut state = runtime.subscribe_system_vm_state();
let mut first = true;
while wait_for_state(&mut state, &shutdown, VmLifecycleState::is_ready).await {
let occupancy = classify_mount_point(&mount_path)?;
if first && expect_hosts_alias {
wait_for_hosts_alias(&shutdown).await;
}
first = false;
if let Err(e) = establish(&runtime, occupancy, &mount_path, nfsd_port, &shutdown).await {
if shutdown.is_cancelled() {
break;
}
warn!(error = %e, "could not establish the ~/ArcBox export; retrying");
if !sleep_unless_shutdown(RETRY_BACKOFF, &shutdown).await {
break;
}
continue;
}
if !wait_for_state(&mut state, &shutdown, |s| !s.is_ready()).await {
break;
}
info!("system VM stopped; the ~/ArcBox export will be rebuilt when it returns");
}
Ok(())
}
async fn establish(
runtime: &Arc<Runtime>,
occupancy: MountPoint,
mount_path: &Path,
nfsd_port: u16,
shutdown: &CancellationToken,
) -> Result<()> {
if occupancy == MountPoint::Stale {
info!(path = %mount_path.display(), "replacing stale ~/ArcBox NFS mount");
unmount(mount_path).await?;
}
std::fs::create_dir_all(mount_path)?;
ensure_guest_export(runtime, shutdown).await?;
mount_with_retry(mount_path, nfsd_port, shutdown).await
}
async fn wait_for_state(
state: &mut watch::Receiver<VmLifecycleState>,
shutdown: &CancellationToken,
pred: fn(&VmLifecycleState) -> bool,
) -> bool {
loop {
if shutdown.is_cancelled() {
return false;
}
if pred(&state.borrow_and_update()) {
return true;
}
tokio::select! {
biased;
() = shutdown.cancelled() => return false,
changed = state.changed() => {
if changed.is_err() {
return false;
}
}
}
}
}
async fn sleep_unless_shutdown(duration: Duration, shutdown: &CancellationToken) -> bool {
tokio::select! {
biased;
() = shutdown.cancelled() => false,
() = tokio::time::sleep(duration) => true,
}
}
async fn ensure_guest_export(runtime: &Arc<Runtime>, shutdown: &CancellationToken) -> Result<()> {
const ENSURE_TIMEOUT: Duration = Duration::from_secs(120);
const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(15);
const RETRY_INTERVAL: Duration = Duration::from_secs(1);
let deadline = tokio::time::Instant::now() + ENSURE_TIMEOUT;
loop {
let attempt = tokio::select! {
biased;
() = shutdown.cancelled() => {
bail!("daemon shutdown before the guest NFS export was established");
}
result = tokio::time::timeout(ATTEMPT_TIMEOUT, send_ensure_export(runtime)) => result,
};
match attempt {
Ok(Ok(notes)) => {
debug!(notes = ?notes, "guest nfs export ensured");
return Ok(());
}
outcome if tokio::time::Instant::now() >= deadline => {
return match outcome {
Ok(Err(e)) => Err(e).context("guest did not establish the NFS export in time"),
_ => Err(anyhow::anyhow!(
"guest did not establish the NFS export within {ENSURE_TIMEOUT:?}"
)),
};
}
Ok(Err(e)) => debug!(error = %e, "ensure guest nfs export attempt failed, retrying"),
Err(_) => debug!("ensure guest nfs export attempt timed out, retrying"),
}
tokio::select! {
biased;
() = shutdown.cancelled() => {
bail!("daemon shutdown before the guest NFS export was established");
}
() = tokio::time::sleep(RETRY_INTERVAL) => {}
}
}
}
async fn send_ensure_export(runtime: &Arc<Runtime>) -> Result<Vec<String>> {
let machine = DEFAULT_MACHINE_NAME.to_string();
let rt = Arc::clone(runtime);
let mut agent = tokio::task::spawn_blocking(move || rt.get_agent(&machine)).await??;
let resp = if agent.is_blocking() {
tokio::task::spawn_blocking(move || agent.ensure_nfs_export_blocking()).await??
} else {
agent.ensure_nfs_export().await?
};
Ok(resp.notes)
}
async fn wait_for_hosts_alias(shutdown: &CancellationToken) {
const ALIAS_WAIT: Duration = Duration::from_secs(10);
let deadline = tokio::time::Instant::now() + ALIAS_WAIT;
while !hosts_alias_installed() && !shutdown.is_cancelled() {
if tokio::time::Instant::now() >= deadline {
debug!("hosts alias did not appear; mounting by loopback");
return;
}
tokio::time::sleep(Duration::from_millis(250)).await;
}
}
async fn spawn_proxy(
runtime: &Arc<Runtime>,
shutdown: &CancellationToken,
vsock_port: u32,
) -> Result<u16> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let local_port = listener.local_addr()?.port();
debug!(local_port, vsock_port, "NFS proxy listening");
let runtime = Arc::clone(runtime);
let shutdown = shutdown.clone();
tokio::spawn(run_proxy(listener, runtime, shutdown, vsock_port));
Ok(local_port)
}
async fn run_proxy(
listener: TcpListener,
runtime: Arc<Runtime>,
shutdown: CancellationToken,
vsock_port: u32,
) {
loop {
let stream = tokio::select! {
biased;
() = shutdown.cancelled() => return,
result = listener.accept() => match result {
Ok((stream, _)) => stream,
Err(e) => {
warn!(vsock_port, error = %e, "NFS proxy accept failed");
continue;
}
}
};
let runtime = Arc::clone(&runtime);
tokio::spawn(async move {
if let Err(e) = relay_connection(stream, runtime, vsock_port).await {
debug!(vsock_port, error = %e, "NFS proxy relay failed");
}
});
}
}
async fn relay_connection(
mut tcp: tokio::net::TcpStream,
runtime: Arc<Runtime>,
vsock_port: u32,
) -> Result<()> {
let fd = tokio::task::spawn_blocking(move || {
runtime.connect_vsock_port(DEFAULT_MACHINE_NAME, vsock_port)
})
.await??;
let owned = unsafe { OwnedFd::from_raw_fd(fd) };
let mut vsock = VsockStream::from_fd_with_shutdown(owned, VsockShutdown::CloseOnDropOnly)?;
copy_bidirectional(&mut tcp, &mut vsock).await?;
Ok(())
}
async fn mount_with_retry(
mount_path: &Path,
nfsd_port: u16,
shutdown: &CancellationToken,
) -> Result<()> {
let opts = render_mount_opts(nfsd_port);
let deadline = tokio::time::Instant::now() + MOUNT_TIMEOUT;
loop {
if shutdown.is_cancelled() {
bail!("daemon shutdown before ~/ArcBox mount completed");
}
let source = mount_source();
match run_mount(&opts, &source, mount_path).await {
Ok(()) => {
let _ = MOUNTED_PATH.set(mount_path.to_path_buf());
info!(
path = %mount_path.display(),
nfsd_port,
"mounted guest docker data at ~/ArcBox (NFSv4, read-only)"
);
return Ok(());
}
Err(e) if tokio::time::Instant::now() >= deadline => {
bail!("mount_nfs did not succeed within {MOUNT_TIMEOUT:?}: {e}");
}
Err(e) => debug!(error = %e, "mount_nfs attempt failed, retrying"),
}
tokio::time::sleep(MOUNT_RETRY_INTERVAL).await;
}
}
async fn run_mount(opts: &str, source: &str, mount_path: &Path) -> Result<(), String> {
let opts = opts.to_string();
let source = source.to_string();
let mount_path = mount_path.to_path_buf();
tokio::task::spawn_blocking(move || {
let output = Command::new("/sbin/mount_nfs")
.arg("-o")
.arg(&opts)
.arg(&source)
.arg(&mount_path)
.output()
.map_err(|e| format!("failed to execute mount_nfs: {e}"))?;
if output.status.success() {
Ok(())
} else {
Err(format!(
"mount_nfs exited with {}: {}",
output.status.code().unwrap_or(-1),
String::from_utf8_lossy(&output.stderr).trim()
))
}
})
.await
.map_err(|e| format!("mount_nfs task panicked: {e}"))?
}
fn mount_source() -> String {
mount_source_for(hosts_alias_installed())
}
fn hosts_alias_installed() -> bool {
std::fs::read_to_string("/etc/hosts")
.is_ok_and(|content| arcbox_helper::hosts_alias_installed(&content))
}
fn mount_source_for(alias_installed: bool) -> String {
if alias_installed {
format!("{}:/", arcbox_helper::HOSTS_ALIAS_NAME)
} else {
"127.0.0.1:/".to_string()
}
}
fn render_mount_opts(nfsd_port: u16) -> String {
format!("ro,noowners,vers=4,rdirplus,actimeo=10,deadtimeout=60,port={nfsd_port}")
}
fn resolve_mount_path() -> Option<PathBuf> {
if let Some(dir) = std::env::var_os("ARCBOX_HOST_MOUNT_DIR") {
return Some(PathBuf::from(dir));
}
dirs::home_dir().map(|home| resolve_mount_path_from_home(&home))
}
fn resolve_mount_path_from_home(home: &Path) -> PathBuf {
home.join("ArcBox")
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MountPoint {
Free,
Stale,
}
fn classify_mount_point(mount_path: &Path) -> Result<MountPoint> {
match current_mount_info(mount_path) {
None => Ok(MountPoint::Free),
Some(info) if is_arcbox_nfs_mount(&info) => Ok(MountPoint::Stale),
Some(info) => bail!(
"~/ArcBox path {} already occupied by {} ({})",
mount_path.display(),
info.source,
info.fstype
),
}
}
fn is_arcbox_nfs_mount(info: &MountInfo) -> bool {
info.fstype == "nfs"
&& (info.source == mount_source_for(false) || info.source == mount_source_for(true))
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct MountInfo {
source: String,
fstype: String,
}
fn current_mount_info(path: &Path) -> Option<MountInfo> {
let output = Command::new("/sbin/mount").output().ok()?;
if !output.status.success() {
return None;
}
let target = path.to_string_lossy();
String::from_utf8_lossy(&output.stdout)
.lines()
.find_map(|line| match parse_mount_line(line) {
Some((mountpoint, info)) if mountpoint == target => Some(info),
_ => None,
})
}
fn parse_mount_line(line: &str) -> Option<(&str, MountInfo)> {
let (source, rest) = line.split_once(" on ")?;
let (mountpoint, suffix) = rest.split_once(" (")?;
let fstype = suffix
.split([',', ')'])
.next()
.unwrap_or_default()
.trim()
.to_string();
Some((
mountpoint,
MountInfo {
source: source.to_string(),
fstype,
},
))
}
async fn unmount(path: &Path) -> Result<()> {
let mut command = tokio::process::Command::new("/sbin/umount");
command.arg(path).kill_on_drop(true);
let status = tokio::time::timeout(UNMOUNT_TIMEOUT, command.status())
.await
.context("umount timed out")??;
if status.success() {
Ok(())
} else {
bail!("umount exited with {}", status.code().unwrap_or(-1))
}
}
#[cfg(test)]
mod tests {
use super::{
MountInfo, VmLifecycleState, is_arcbox_nfs_mount, mount_source_for, parse_mount_line,
render_mount_opts, resolve_mount_path_from_home, wait_for_state,
};
use std::path::{Path, PathBuf};
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
#[test]
fn only_our_exact_mount_shape_is_reclaimed() {
for source in ["127.0.0.1:/", "ArcBox:/"] {
let ours = MountInfo {
source: source.to_string(),
fstype: "nfs".to_string(),
};
assert!(is_arcbox_nfs_mount(&ours), "{source} should be reclaimed");
}
let foreign_nfs = MountInfo {
source: "fileserver:/export/home".to_string(),
fstype: "nfs".to_string(),
};
assert!(!is_arcbox_nfs_mount(&foreign_nfs));
let smb = MountInfo {
source: "//user@server/share".to_string(),
fstype: "smbfs".to_string(),
};
assert!(!is_arcbox_nfs_mount(&smb));
}
#[test]
fn mount_path_is_arcbox_under_home() {
assert_eq!(
resolve_mount_path_from_home(Path::new("/Users/tester")),
PathBuf::from("/Users/tester/ArcBox")
);
}
#[test]
fn mount_opts_are_readonly_v4_with_nfsd_port_pinned() {
let opts = render_mount_opts(51000);
assert!(opts.contains("vers=4"));
assert!(opts.contains("ro,"));
assert!(opts.contains("port=51000"));
assert!(!opts.contains("rw"));
assert!(opts.contains("noowners"));
}
#[test]
fn mount_source_is_the_v4_pseudo_root_under_both_names() {
assert_eq!(mount_source_for(false), "127.0.0.1:/");
assert_eq!(mount_source_for(true), "ArcBox:/");
}
#[tokio::test]
async fn ready_wait_resumes_on_the_arrival_edge_not_the_departure() {
let (tx, mut rx) = watch::channel(VmLifecycleState::Running);
let shutdown = CancellationToken::new();
assert!(wait_for_state(&mut rx, &shutdown, VmLifecycleState::is_ready).await);
for state in [VmLifecycleState::Stopping, VmLifecycleState::Stopped] {
tx.send_replace(state);
assert!(wait_for_state(&mut rx, &shutdown, |s| !s.is_ready()).await);
}
tx.send_replace(VmLifecycleState::Starting);
let waiter = tokio::spawn({
let shutdown = shutdown.clone();
let mut rx = rx.clone();
async move { wait_for_state(&mut rx, &shutdown, VmLifecycleState::is_ready).await }
});
tokio::task::yield_now().await;
assert!(!waiter.is_finished(), "Starting is not an arrival");
tx.send_replace(VmLifecycleState::Running);
assert!(waiter.await.expect("waiter panicked"));
}
#[tokio::test]
async fn ready_wait_gives_up_on_shutdown_and_on_a_dropped_lifecycle() {
let (tx, mut rx) = watch::channel(VmLifecycleState::Stopped);
let shutdown = CancellationToken::new();
shutdown.cancel();
assert!(!wait_for_state(&mut rx, &shutdown, VmLifecycleState::is_ready).await);
drop(tx);
let shutdown = CancellationToken::new();
assert!(!wait_for_state(&mut rx, &shutdown, VmLifecycleState::is_ready).await);
}
#[test]
fn parse_mount_line_extracts_source_and_fstype() {
let line =
"127.0.0.1:/run/arcbox/nfs-export/docker on /Users/t/ArcBox (nfs, nodev, read-only)";
let (mountpoint, info) = parse_mount_line(line).expect("line should parse");
assert_eq!(mountpoint, "/Users/t/ArcBox");
assert_eq!(
info,
MountInfo {
source: "127.0.0.1:/run/arcbox/nfs-export/docker".to_string(),
fstype: "nfs".to_string(),
}
);
}
}