diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 58bcb0b4..bfb7e92c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -129,6 +129,12 @@ jobs: failed="" for t in $targets; do extra="" + if [ "$t" = devnet_lifecycle ] && [ "$RUNNER_OS" = Windows ]; then + # It stops ant-devnet with SIGTERM and SIGKILL, so it is Unix + # only, and compiles to nothing on Windows. + echo "skipping $t on $RUNNER_OS (Unix only)" + continue + fi if [ "$t" = storage_scale ]; then # Its gates compare the store scan against timed directory walks, and # were measured on Linux; it has only ever run there. @@ -165,6 +171,18 @@ jobs: - name: Unit tests without default features (no logging, no webrtc-direct) if: runner.os == 'Linux' run: cargo test --lib --no-default-features + # The run-directory tests that mount a tmpfs and bind mounts inside run + # directories, make directories another user owns, or act as another + # user. They need root, which the runner gives through sudo, so + # `cargo test` alone skips them. + - name: ant-devnet run directories against real mounts and other users + if: runner.os == 'Linux' + shell: bash + run: | + set -euo pipefail + cargo test --lib --features test-utils with_sudo_ -- --ignored --test-threads=1 2>&1 | tee /tmp/run_dir_sudo.log + grep -qE 'test result: ok\. [1-9]' /tmp/run_dir_sudo.log \ + || { echo 'the run-directory sudo tests ran nothing'; exit 1; } filesystems: name: Storage on ${{ matrix.fs }} diff --git a/Cargo.lock b/Cargo.lock index 56385dab..9bff3cf6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -888,6 +888,7 @@ dependencies = [ "tracing-appender", "tracing-subscriber", "webrtc", + "winapi-util", "xor_name", "zip", ] diff --git a/Cargo.toml b/Cargo.toml index 355364f3..bac2be3d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -122,12 +122,18 @@ self_encryption = { version = "0.36", optional = true } libc = "0.2" # Safe wrappers for `openat`/`unlinkat`, so the storage-migration cleanup can delete relative # to a directory handle it opened `O_NOFOLLOW` rather than by re-resolving a path it has -# already checked. Adds nothing to the build: `tempfile`, a direct dependency, already pulls -# this exact crate and version in with the `fs` feature on. -rustix = { version = "1", features = ["fs"] } +# already checked. Adds no crate to the build: `tempfile`, a direct dependency, already pulls +# this exact crate and version in with the `fs` feature on. `process` is for `geteuid`, which +# the devnet uses to tell a run directory it created from one another user put in its place. +rustix = { version = "1", features = ["fs", "process"] } [target.'cfg(windows)'.dependencies] self-replace = "1" +# The volume serial number and file index of a directory handle, which tells +# a devnet run directory from a copy put in its place. Already in Cargo.lock +# (through walkdir), and what it needs, windows-sys 0.48, already ships in the +# Windows build through `directories`. +winapi-util = "0.1" [target.'cfg(target_os = "macos")'.dependencies] objc2 = "0.6" diff --git a/src/bin/ant-devnet/cli.rs b/src/bin/ant-devnet/cli.rs index 0c9c7c17..11834b03 100644 --- a/src/bin/ant-devnet/cli.rs +++ b/src/bin/ant-devnet/cli.rs @@ -25,11 +25,17 @@ pub struct Cli { #[arg(long)] pub base_port: Option, - /// Data directory for node state. + /// Directory to create this run's own data directory in (default: + /// `devnets/` in the ant data root). The run directory is removed on + /// shutdown unless `--no-cleanup` is set; this directory, and anything else + /// in it, never is. Run directories left here by runs that were killed + /// are removed when the next run starts. On Unix only you and root may be + /// able to change it, or any directory above it. #[arg(long)] pub data_dir: Option, - /// Keep node data directories on shutdown instead of removing them. + /// Keep this run's data directory on shutdown. Later runs leave it alone + /// too. #[arg(long)] pub no_cleanup: bool, diff --git a/src/bin/ant-devnet/main.rs b/src/bin/ant-devnet/main.rs index 4e9484dd..e25ac19e 100644 --- a/src/bin/ant-devnet/main.rs +++ b/src/bin/ant-devnet/main.rs @@ -42,6 +42,13 @@ use cli::Cli; async fn main() -> color_eyre::Result<()> { color_eyre::install()?; + // Listen for stop requests before anything starts. Until a handler is + // installed, SIGTERM kills the process outright: the devnet is never shut + // down, Anvil is orphaned and the data directory is left behind + // (ant-node#261). + let stop = stop_requested()?; + tokio::pin!(stop); + let cli = Cli::parse(); #[cfg(feature = "logging")] @@ -118,19 +125,54 @@ async fn main() -> color_eyre::Result<()> { if let Some(base_port) = cli.webrtc_direct_base_port { config.webrtc_direct_base_port = base_port; } + // A stop request during EVM setup drops the half-finished setup, and with + // it any Anvil it already started. let ResolvedEvm { manifest: evm_info, - local_testnet: _local_evm_testnet, - } = resolve_evm_info( - cli.evm_network.as_deref(), - cli.enable_evm, - cli.host, - &mut config, - ) - .await?; + local_testnet, + } = tokio::select! { + evm = resolve_evm_info( + cli.evm_network.as_deref(), + cli.enable_evm, + cli.host, + &mut config, + ) => evm?, + signal = &mut stop => { + ant_node::logging::info!("{signal} during EVM setup; exiting"); + return Ok(()); + } + }; let mut devnet = Devnet::new(config).await?; - devnet.start().await?; + // A stop request during startup cancels the start rather than dropping + // it, so the start returns with every node it started registered and + // the shutdown below stops them all before removing the data directory. + let cancel = devnet.cancellation_token(); + let started = { + let start = devnet.start(); + tokio::pin!(start); + tokio::select! { + started = &mut start => started, + signal = &mut stop => { + ant_node::logging::info!("{signal} during startup; stopping the devnet"); + cancel.cancel(); + start.await + } + } + }; + // Stopped or failed part-way: shut down the nodes that did start and + // remove the data directory. Anvil stops when `local_testnet` drops on + // return. + if cancel.is_cancelled() { + devnet.shutdown().await?; + return Ok(()); + } + if let Err(e) = started { + if let Err(cleanup) = devnet.shutdown().await { + ant_node::logging::warn!("Cleanup after a failed start also failed: {cleanup}"); + } + return Err(e.into()); + } let created_at = chrono::Utc::now().to_rfc3339(); @@ -209,13 +251,52 @@ async fn main() -> color_eyre::Result<()> { )?; } - ant_node::logging::info!("Devnet running. Press Ctrl+C to stop."); - tokio::signal::ctrl_c().await?; + ant_node::logging::info!("Devnet running. Press Ctrl+C or send SIGTERM to stop."); + let signal = stop.await; + ant_node::logging::info!("{signal} received; stopping the devnet"); devnet.shutdown().await?; + // Stop Anvil after the nodes, which verify payments against it. + drop(local_testnet); Ok(()) } +/// A future that resolves when the process is asked to stop, naming the +/// request: SIGINT (Ctrl+C), SIGTERM (`kill`, service managers, `ant dev +/// stop`) or SIGHUP (the terminal closed) on Unix; Ctrl+C or Ctrl+Break on +/// Windows. +/// +/// The handlers are installed by this call, not when the future is first +/// polled, so call it before starting anything. +fn stop_requested() -> std::io::Result> { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + let mut interrupt = signal(SignalKind::interrupt())?; + let mut terminate = signal(SignalKind::terminate())?; + let mut hangup = signal(SignalKind::hangup())?; + Ok(async move { + tokio::select! { + _ = interrupt.recv() => "SIGINT", + _ = terminate.recv() => "SIGTERM", + _ = hangup.recv() => "SIGHUP", + } + }) + } + #[cfg(windows)] + { + use tokio::signal::windows::{ctrl_break, ctrl_c}; + let mut interrupt = ctrl_c()?; + let mut brk = ctrl_break()?; + Ok(async move { + tokio::select! { + _ = interrupt.recv() => "Ctrl+C", + _ = brk.recv() => "Ctrl+Break", + } + }) + } +} + #[cfg(all(feature = "webrtc-direct", feature = "test-utils"))] async fn load_public_file( path: Option<&std::path::Path>, @@ -451,7 +532,7 @@ fn serve_manifest_api( /// payloads are tiny and a devnet serves a handful of LAN devices, so a single /// thread bounds resource use (no per-connection thread to exhaust). Each /// connection gets a read timeout so a slow/idle client can't stall the loop. -/// The thread is detached and dies when the process exits on Ctrl+C. +/// The thread is detached and dies when the process exits. fn spawn_manifest_server( listener: std::net::TcpListener, manifest_json: String, diff --git a/src/devnet.rs b/src/devnet.rs index 4c836614..aa68711a 100644 --- a/src/devnet.rs +++ b/src/devnet.rs @@ -27,6 +27,7 @@ use tokio::sync::RwLock; use tokio::task::JoinHandle; use tokio::time::Instant; use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; #[cfg(all(feature = "webrtc-direct", feature = "test-utils"))] use crate::ant_protocol::{ChunkMessage, ChunkMessageBody, ChunkPutRequest, ChunkPutResponse}; @@ -41,6 +42,11 @@ use bytes::Bytes; #[cfg(all(feature = "webrtc-direct", feature = "test-utils"))] use std::collections::HashMap; +mod run_dir; + +use run_dir::RunDir; +pub use run_dir::{LOCK_FILE, MARKER_FILE}; + // ============================================================================= // Devnet Constants // ============================================================================= @@ -55,6 +61,10 @@ pub const DEVNET_PORT_RANGE_MAX: u16 = 60_000; // Default Timing Constants // ============================================================================= +/// Directory under the ant data root that devnet runs create their own data +/// directories in by default (see [`DevnetConfig::data_dir`]). +pub const DEVNETS_SUBDIR: &str = "devnets"; + /// Default delay between spawning nodes (milliseconds). const DEFAULT_SPAWN_DELAY_MS: u64 = 200; @@ -82,6 +92,11 @@ const STABILIZATION_MIN_CONNECTIONS_CAP: usize = 3; /// Health monitor check interval (seconds). const HEALTH_CHECK_INTERVAL_SECS: u64 = 5; +/// How long shutdown gives a node's request handlers, and then the storage +/// writes they started, to finish (seconds), as a node does. A node still +/// writing after that keeps the run directory. +const DRAIN_GRACE_SECS: u64 = 5; + // ============================================================================= // AntProtocol Devnet Configuration // ============================================================================= @@ -201,7 +216,21 @@ pub struct DevnetConfig { /// Number of bootstrap nodes (first N nodes, default: 3). pub bootstrap_count: usize, - /// Root directory for devnet data. + /// Directory each run creates its own data directory in (default: + /// `{ant data root}/devnets`). + /// + /// [`Devnet::new`] creates a fresh `{pid}-{random}` directory inside it, + /// and from then on [`Devnet::config`]'s `data_dir` is that run + /// directory. A devnet deletes only a directory it created and marked as + /// its own ([`MARKER_FILE`]), never this one or anything else in it: the + /// ant data root also holds real nodes' data and the client's payment + /// caches and upload history (ant-node#261). Run directories that earlier + /// runs left here, killed before they could remove them, are removed when + /// the next run starts, unless they were kept. + /// + /// On Unix it must be a directory only the user and root can change, as + /// must every directory above it, or [`Devnet::new`] fails: another user + /// who could would be able to redirect what the nodes write. pub data_dir: PathBuf, /// Delay between node spawns (default: 200ms). @@ -216,7 +245,8 @@ pub struct DevnetConfig { /// Enable verbose logging for devnet nodes. pub enable_node_logging: bool, - /// Whether to remove the data directory on shutdown. + /// Whether to remove the run's data directory on shutdown. When `false`, + /// its marker also tells later runs to leave it alone. pub cleanup_data_dir: bool, /// Optional EVM network for payment verification. @@ -247,7 +277,7 @@ impl Default for DevnetConfig { node_count: DEFAULT_NODE_COUNT, base_port, bootstrap_count: DEFAULT_BOOTSTRAP_COUNT, - data_dir: default_root_dir(), + data_dir: default_root_dir().join(DEVNETS_SUBDIR), spawn_delay: Duration::from_millis(DEFAULT_SPAWN_DELAY_MS), stabilization_timeout: Duration::from_secs(DEFAULT_STABILIZATION_TIMEOUT_SECS), node_startup_timeout: Duration::from_secs(DEFAULT_NODE_STARTUP_TIMEOUT_SECS), @@ -340,6 +370,9 @@ pub struct DevnetNode { state: Arc>, bootstrap_addrs: Vec, protocol_task: Option>, + /// The request handlers `protocol_task` has started, which shutdown + /// waits for. + handlers: TaskTracker, #[cfg(feature = "webrtc-direct")] webrtc_direct_task: Option>, #[cfg(feature = "webrtc-direct")] @@ -360,12 +393,23 @@ impl DevnetNode { } /// A local devnet composed of multiple nodes. +/// +/// Call [`shutdown`](Self::shutdown) before dropping it. Dropped without one, +/// the nodes it started are not stopped, and its run directory is left, +/// locked until the process exits so that no other run deletes it while they +/// may still be writing there. pub struct Devnet { config: DevnetConfig, nodes: Vec, shutdown: CancellationToken, state: Arc>, health_monitor: Option>, + /// This run's data directory, until [`shutdown`](Self::shutdown) removes + /// or releases it. Dropped without a shutdown, the devnet keeps it locked + /// until the process exits. + run_dir: Option, + /// [`DRAIN_GRACE_SECS`], shorter in tests. + drain_grace: Duration, #[cfg(feature = "webrtc-direct")] browser_endpoint_catalog: Arc, } @@ -373,6 +417,10 @@ pub struct Devnet { impl Devnet { /// Create a new devnet with the given configuration. /// + /// Creates this run's own data directory inside `config.data_dir`, after + /// removing any that earlier runs left there and that are over and not + /// kept (see [`DevnetConfig::data_dir`]). + /// /// # Errors /// /// Returns `DevnetError::Config` if the configuration is invalid (e.g. bootstrap @@ -475,7 +523,13 @@ impl Devnet { } } - tokio::fs::create_dir_all(&config.data_dir).await?; + let parent = config.data_dir.clone(); + let keep = !config.cleanup_data_dir; + let run_dir = tokio::task::spawn_blocking(move || RunDir::create(&parent, keep)) + .await + .map_err(std::io::Error::other)??; + config.data_dir = run_dir.path().to_path_buf(); + info!("Devnet data directory: {}", config.data_dir.display()); Ok(Self { config, @@ -483,6 +537,8 @@ impl Devnet { shutdown: CancellationToken::new(), state: Arc::new(RwLock::new(NetworkState::Uninitialized)), health_monitor: None, + run_dir: Some(run_dir), + drain_grace: Duration::from_secs(DRAIN_GRACE_SECS), #[cfg(feature = "webrtc-direct")] browser_endpoint_catalog: Arc::new(crate::web_rtc::BrowserEndpointCatalog::default()), }) @@ -518,6 +574,12 @@ impl Devnet { /// Shutdown the devnet. /// + /// Stops every node, then removes the run's data directory, but only if + /// they have all stopped writing to it. A node that has not, because a + /// request is still being handled after a few seconds or its P2P node + /// failed to shut down, leaves the directory, locked until the process + /// exits; the next run removes it then. + /// /// # Errors /// /// Returns `DevnetError::Io` if the data directory cleanup fails. @@ -531,41 +593,95 @@ impl Devnet { handle.abort(); } - let mut shutdown_futures = Vec::with_capacity(self.nodes.len()); + let grace = self.drain_grace; + let mut stops = Vec::with_capacity(self.nodes.len()); for node in self.nodes.iter_mut().rev() { debug!("Stopping node {}", node.index); - if let Some(handle) = node.protocol_task.take() { - handle.abort(); - } - #[cfg(feature = "webrtc-direct")] - if let Some(handle) = node.webrtc_direct_task.take() { - if let Err(error) = handle.await { - warn!( - "Error stopping node {} WebRtcDirect listener: {error}", - node.index - ); - } - } - let node_index = node.index; let node_state = Arc::clone(&node.state); + let protocol_task = node.protocol_task.take(); + #[cfg(feature = "webrtc-direct")] + let webrtc_direct_task = node.webrtc_direct_task.take(); + let handlers = node.handlers.clone(); + let storage = node + .ant_protocol + .as_ref() + .map(|protocol| protocol.storage()); let p2p_node = node.p2p_node.take(); - shutdown_futures.push(async move { + stops.push(async move { + let mut stopped = true; + // The token cancelled above closes the browser listener, + // which drains its own requests before it returns. + #[cfg(feature = "webrtc-direct")] + if let Some(handle) = webrtc_direct_task { + if let Err(error) = handle.await { + warn!("Error stopping node {node_index} WebRtcDirect listener: {error}"); + stopped = false; + } + } + // As a node does: the routing loop first, aborted and awaited + // so that it starts no more handlers, then the handlers it + // started, given a moment to finish. + if let Some(handle) = protocol_task { + handle.abort(); + let _ = handle.await; + } + handlers.close(); + if tokio::time::timeout(grace, handlers.wait()).await.is_err() { + warn!( + "Node {node_index}: {} request handler(s) still running after {grace:?}", + handlers.len() + ); + stopped = false; + } + // A storage write outlives the future that started it. + if let Some(storage) = storage { + if tokio::time::timeout(grace, storage.wait_idle()) + .await + .is_err() + { + warn!("Node {node_index}: storage writes still running after {grace:?}"); + stopped = false; + } + } if let Some(p2p) = p2p_node { if let Err(e) = p2p.shutdown().await { warn!("Error shutting down node {node_index}: {e}"); + stopped = false; } } *node_state.write().await = NodeState::Stopped; + stopped }); } - futures::future::join_all(shutdown_futures).await; - - if self.config.cleanup_data_dir { - if let Err(e) = tokio::fs::remove_dir_all(&self.config.data_dir).await { - warn!("Failed to cleanup devnet data directory: {e}"); + let all_stopped = futures::future::join_all(stops) + .await + .into_iter() + .all(|stopped| stopped); + + match self.run_dir.take() { + Some(run_dir) if !all_stopped => { + warn!( + "Leaving devnet data directory {}: not every node has stopped writing to it. \ + It stays locked until this process exits, and the next run removes it.", + self.config.data_dir.display() + ); + run_dir.keep_locked_until_exit(); + } + Some(run_dir) if self.config.cleanup_data_dir => { + match tokio::task::spawn_blocking(move || run_dir.remove()).await { + Ok(Ok(())) => {} + Ok(Err(e)) => warn!( + "Failed to remove devnet data directory {}: {e}", + self.config.data_dir.display() + ), + Err(e) => warn!("Failed to remove devnet data directory: {e}"), + } } + // Without cleanup, dropping the run directory releases its lock + // and leaves it, marked to be kept. + Some(_) | None => {} } *self.state.write().await = NetworkState::Stopped; @@ -573,6 +689,29 @@ impl Devnet { Ok(()) } + /// A token that stops this devnet's [`start`](Self::start) when cancelled. + /// + /// A start in progress returns an error at its next safe point: between + /// node spawns, or while it waits for nodes to come up or the network to + /// stabilize. Every node it started is registered by then, so a following + /// [`shutdown`](Self::shutdown) stops them all before it removes the data + /// directory. Dropping a `start` future part-way instead can leave a node + /// running unregistered, still writing into that directory. It is the same + /// token `shutdown` cancels. + #[must_use] + pub fn cancellation_token(&self) -> CancellationToken { + self.shutdown.clone() + } + + /// Wait `delay`, or fail at once if the devnet has been cancelled. + async fn pause(&self, delay: Duration) -> Result<()> { + tokio::select! { + biased; + () = self.shutdown.cancelled() => Err(start_cancelled()), + () = tokio::time::sleep(delay) => Ok(()), + } + } + /// Get devnet configuration. #[must_use] pub fn config(&self) -> &DevnetConfig { @@ -837,7 +976,7 @@ impl Devnet { for i in 0..self.config.bootstrap_count { let node = self.create_node(i, true, vec![]).await?; self.start_node(node).await?; - tokio::time::sleep(self.config.spawn_delay).await; + self.pause(self.config.spawn_delay).await?; } self.wait_for_nodes_ready(0..self.config.bootstrap_count) @@ -873,7 +1012,7 @@ impl Devnet { for i in self.config.bootstrap_count..self.config.node_count { let node = self.create_node(i, false, bootstrap_addrs.clone()).await?; self.start_node(node).await?; - tokio::time::sleep(self.config.spawn_delay).await; + self.pause(self.config.spawn_delay).await?; } info!("All regular nodes started"); @@ -922,6 +1061,7 @@ impl Devnet { state: Arc::new(RwLock::new(NodeState::Pending)), bootstrap_addrs, protocol_task: None, + handlers: TaskTracker::new(), #[cfg(feature = "webrtc-direct")] webrtc_direct_task: None, #[cfg(feature = "webrtc-direct")] @@ -1027,6 +1167,13 @@ impl Devnet { node.p2p_node = Some(Arc::new(p2p_node)); *node.state.write().await = NodeState::Running; + // Registered as soon as it runs, so that should the rest of its + // startup fail, shutdown still stops it before removing its data. + self.nodes.push(node); + let node = self + .nodes + .last_mut() + .ok_or_else(|| DevnetError::Startup(format!("Node {index} was not registered")))?; #[cfg(feature = "webrtc-direct")] if self.config.webrtc_direct { @@ -1092,6 +1239,7 @@ impl Devnet { let mut events = p2p.subscribe_events(); let p2p_clone = Arc::clone(p2p); let protocol_clone = Arc::clone(protocol); + let handlers = node.handlers.clone(); let node_index = node.index; node.protocol_task = Some(tokio::spawn(async move { while let Ok(event) = events.recv().await { @@ -1109,7 +1257,7 @@ impl Devnet { let protocol = Arc::clone(&protocol_clone); let p2p = Arc::clone(&p2p_clone); let received_at = Instant::now(); - tokio::spawn(handle_chunk_message( + handlers.spawn(handle_chunk_message( protocol, p2p, source, @@ -1124,7 +1272,6 @@ impl Devnet { } debug!("Node {} started successfully", node.index); - self.nodes.push(node); Ok(()) } @@ -1146,8 +1293,8 @@ impl Devnet { return Err(DevnetError::Startup(format!("Node {i} failed: {e}"))); } _ => { - tokio::time::sleep(Duration::from_millis(NODE_READY_POLL_INTERVAL_MS)) - .await; + self.pause(Duration::from_millis(NODE_READY_POLL_INTERVAL_MS)) + .await?; } } } @@ -1189,7 +1336,8 @@ impl Devnet { "Waiting for stabilization: {} total connections", total_connections ); - tokio::time::sleep(Duration::from_secs(STABILIZATION_POLL_INTERVAL_SECS)).await; + self.pause(Duration::from_secs(STABILIZATION_POLL_INTERVAL_SECS)) + .await?; } Err(DevnetError::Stabilization( @@ -1224,12 +1372,24 @@ impl Devnet { } } +/// The error a cancelled [`Devnet::start`] returns. +fn start_cancelled() -> DevnetError { + DevnetError::Startup("devnet start cancelled".to_string()) +} + impl Drop for Devnet { fn drop(&mut self) { self.shutdown.cancel(); if let Some(handle) = self.health_monitor.take() { handle.abort(); } + // Not shut down, so nodes it started may still be running: their + // tasks hold them, and outlive this. Keep the run directory locked + // until the process exits, so that no sweep deletes it under them. A + // later run removes it then, as it would after a SIGKILL. + if let Some(run_dir) = self.run_dir.take() { + run_dir.keep_locked_until_exit(); + } } } @@ -1245,4 +1405,222 @@ mod tests { fn default_config_has_no_advertise_ip() { assert!(DevnetConfig::default().advertise_ip.is_none()); } + + #[test] + fn runs_go_in_the_devnets_dir_of_the_data_root_by_default() { + let devnets = default_root_dir().join(DEVNETS_SUBDIR); + for config in [ + DevnetConfig::default(), + DevnetConfig::minimal(), + DevnetConfig::small(), + ] { + assert_eq!(config.data_dir, devnets); + assert!(config.cleanup_data_dir); + } + } + + #[tokio::test] + async fn each_devnet_gets_and_removes_only_its_own_run_dir( + ) -> std::result::Result<(), Box> { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("devnets"); + let mut config = DevnetConfig::minimal(); + config.data_dir.clone_from(&parent); + let mut first = Devnet::new(config.clone()).await?; + let mut second = Devnet::new(config).await?; + let first_dir = first.config().data_dir.clone(); + let second_dir = second.config().data_dir.clone(); + + assert_ne!(first_dir, second_dir); + for dir in [&first_dir, &second_dir] { + assert_eq!( + dir.parent().map(std::fs::canonicalize).transpose()?, + Some(std::fs::canonicalize(&parent)?), + "{}", + dir.display() + ); + assert!(dir.join(MARKER_FILE).is_file(), "{}", dir.display()); + } + + first.shutdown().await?; + assert!(!first_dir.exists()); + assert!(second_dir.join(MARKER_FILE).is_file()); + second.shutdown().await?; + assert!(!second_dir.exists()); + assert!(parent.is_dir(), "the parent is not the devnet's to delete"); + Ok(()) + } + + #[tokio::test] + async fn without_cleanup_the_run_dir_is_kept( + ) -> std::result::Result<(), Box> { + let temp = tempfile::tempdir()?; + let mut config = DevnetConfig::minimal(); + config.data_dir = temp.path().to_path_buf(); + config.cleanup_data_dir = false; + let mut devnet = Devnet::new(config.clone()).await?; + let kept = devnet.config().data_dir.clone(); + devnet.shutdown().await?; + drop(devnet); + + // A later run's sweep leaves it too. + config.cleanup_data_dir = true; + let mut later = Devnet::new(config).await?; + assert!(kept.join(MARKER_FILE).is_file()); + later.shutdown().await?; + assert!(kept.join(MARKER_FILE).is_file()); + Ok(()) + } + + /// A devnet of two nodes, quick to start, running in `parent`. + fn two_nodes(parent: &std::path::Path) -> DevnetConfig { + let mut config = DevnetConfig::minimal(); + config.node_count = 2; + config.bootstrap_count = 1; + config.base_port = 0; + config.spawn_delay = Duration::from_millis(20); + config.data_dir = parent.join("devnets"); + config + } + + /// Hermes round 3 on #263: a started devnet dropped without a shutdown + /// leaves nodes running, still writing into its run directory, so the + /// next devnet's sweep must leave it. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_started_devnet_dropped_without_shutdown_is_not_swept( + ) -> std::result::Result<(), Box> { + let temp = tempfile::tempdir()?; + let config = two_nodes(temp.path()); + let mut dropped = Devnet::new(config.clone()).await?; + dropped.start().await?; + let dropped_dir = dropped.config().data_dir.clone(); + drop(dropped); + + let mut next = Devnet::new(config).await?; + + assert!( + dropped_dir.join(MARKER_FILE).is_file(), + "the next devnet swept {}", + dropped_dir.display() + ); + next.shutdown().await?; + Ok(()) + } + + /// Hermes round 4 on #263: a PUT still being written when the devnet + /// shuts down keeps the run directory, locked, rather than having it + /// removed under the write. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[allow(clippy::await_holding_lock)] + async fn a_put_still_being_written_at_shutdown_keeps_the_run_dir( + ) -> std::result::Result<(), Box> { + use crate::ant_protocol::{ChunkMessage, ChunkMessageBody, ChunkPutRequest}; + + let temp = tempfile::tempdir()?; + let mut devnet = Devnet::new(two_nodes(temp.path())).await?; + devnet.start().await?; + devnet.drain_grace = Duration::from_millis(500); + let run_dir = devnet.config().data_dir.clone(); + let sender = devnet + .nodes + .first() + .and_then(|node| node.p2p_node.clone()) + .ok_or("no sender")?; + let receiver = devnet.nodes.get(1).ok_or("no receiver")?; + let peer = receiver.peer_id; + let protocol = receiver.ant_protocol.clone().ok_or("no protocol")?; + let storage = protocol.storage(); + let content = b"a chunk still being written at shutdown"; + let address = ChunkStore::compute_address(content); + protocol.payment_verifier().cache_insert(address); + let request = ChunkMessage { + request_id: 1, + body: ChunkMessageBody::PutRequest(ChunkPutRequest::new( + address, + bytes::Bytes::from_static(content), + )), + } + .encode()?; + + // Held, the write waits on its way into the store. + let gate = storage.test_put_gate(); + let held = gate.write(); + sender + .send_message(&peer, CHUNK_PROTOCOL_ID, request, &[]) + .await?; + let deadline = Instant::now() + Duration::from_secs(30); + while storage.tasks_in_flight() == 0 { + assert!(Instant::now() < deadline, "the PUT never reached the store"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + devnet.shutdown().await?; + + assert!( + run_dir.join(MARKER_FILE).is_file(), + "{} was removed under a write", + run_dir.display() + ); + drop(held); + storage.wait_idle().await; + assert!(run_dir.join(MARKER_FILE).is_file()); + // Kept locked, so the next run leaves it too. + let mut next = Devnet::new(two_nodes(temp.path())).await?; + assert!(run_dir.join(MARKER_FILE).is_file()); + next.shutdown().await?; + Ok(()) + } + + /// Hermes round 4 on #263: a node whose startup fails once its P2P node + /// is running, here because its WebRTC Direct port is taken, is + /// registered all the same, so shutdown stops it. + #[cfg(feature = "webrtc-direct")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_node_whose_startup_fails_part_way_is_stopped_by_shutdown( + ) -> std::result::Result<(), Box> { + let temp = tempfile::tempdir()?; + let mut config = two_nodes(temp.path()); + config.webrtc_direct = true; + let mut devnet = Devnet::new(config).await?; + let taken = std::net::UdpSocket::bind(( + Ipv4Addr::LOCALHOST, + devnet.config().webrtc_direct_base_port, + ))?; + + let started = devnet.start().await; + + assert!( + matches!(started, Err(DevnetError::Startup(_))), + "{started:?}" + ); + let p2p = devnet + .nodes + .first() + .and_then(|node| node.p2p_node.clone()) + .ok_or("the node that failed part-way was not registered")?; + assert!(p2p.is_running()); + devnet.shutdown().await?; + assert!(!p2p.is_running(), "shutdown left the node running"); + assert!(!devnet.config().data_dir.exists()); + drop(taken); + Ok(()) + } + + #[tokio::test] + async fn a_cancelled_devnet_stops_pausing_at_once( + ) -> std::result::Result<(), Box> { + let temp = tempfile::tempdir()?; + let mut config = DevnetConfig::minimal(); + config.data_dir = temp.path().join("devnet"); + config.cleanup_data_dir = false; + let devnet = Devnet::new(config).await?; + + assert!(devnet.pause(Duration::ZERO).await.is_ok()); + devnet.cancellation_token().cancel(); + let started = Instant::now(); + let paused = devnet.pause(Duration::from_mins(1)).await; + assert!(matches!(paused, Err(DevnetError::Startup(_))), "{paused:?}"); + assert!(started.elapsed() < Duration::from_secs(5)); + Ok(()) + } } diff --git a/src/devnet/run_dir.rs b/src/devnet/run_dir.rs new file mode 100644 index 00000000..7f55bf45 --- /dev/null +++ b/src/devnet/run_dir.rs @@ -0,0 +1,2072 @@ +//! A devnet run's own data directory, and the one rule for deleting one: a +//! devnet deletes only a directory it can show it created (ant-node#261). +//! +//! Each run creates a fresh `{pid}-{random}` directory inside the configured +//! parent, exclusively, and claims it only if, once opened, it is still empty +//! and, on Unix, owned by the run's user, so it never adopts something that +//! was already there or that another user put in its place. (A process of +//! the run's own user could put an empty directory there, but it could as +//! well delete whatever the run keeps: no check could hold that line. On +//! Windows, which has no owner check without unsafe code, the parent's +//! permissions decide who could.) Before anything else goes in, it takes an exclusive lock on +//! [`LOCK_FILE`], held for as long as the run lasts, and then writes +//! [`MARKER_FILE`] naming the run, the directory itself and the lock file, +//! by their file numbers (inode, or file index on Windows): a copy of the +//! marker in any other directory does not match, nor does a lock file put +//! in place of the one the run holds. The parent itself is never deleted. +//! +//! - On shutdown a run deletes its directory through the handle it opened +//! when it created it, and only if the name still refers to that directory +//! and the marker there still names the run. +//! - Anything else, a later run included, may delete a run directory only if +//! it belongs to the user (on Unix), is on the parent's filesystem and +//! mount, its marker names that very directory and does not ask to keep it +//! (`--no-cleanup`), and the lock file the +//! marker names is there and free. The OS releases the lock however the +//! process ends, SIGKILL included, and only a run's own process ever takes +//! it, so a free lock means the run is over. A directory without such a +//! marker is never touched. (The marker keeps no device number: on some +//! filesystems, btrfs and NFS among them, that changes when they are +//! mounted again.) +//! - Deletion first checks that the name still refers to the directory it +//! has open, then removes the data, and the marker and lock last, so one +//! that stops part-way leaves a directory the next run still recognises as +//! a finished run, and finishes. It does not cross into another mount: on +//! Linux mounts are told apart by their mount ids, which tell a bind mount +//! from the filesystem it shows, and where the kernel cannot give one (it +//! is older than 5.8, or a sandbox refuses `statx`) nothing is deleted. +//! macOS has no bind mounts, so there the device tells them apart. +//! +//! On Unix every step after the parent is opened goes through a handle on the +//! parent or the run directory, and deletion walks the tree through handles +//! opened without following links. The handles do not keep the paths still, +//! though. Whoever can write to a directory can rename what is in it: swap a +//! directory above the run directory for a link to another tree, which +//! writes through [`RunDir::path`] would then follow, or move the run +//! directory away while it is being deleted. So the parent must be one no +//! other user can change: it, and every directory above it, must belong to +//! the user or root, and be one no one else can write to, or sticky, as +//! `/tmp` is, which lets no one rename what they do not own. (On Linux, one +//! its group can write to passes if the group is the user's own, as with +//! Ubuntu's user private groups, and no ACL lets anyone else in. On macOS, +//! where every user shares a group, it does not.) A run refuses to start in +//! any other parent. Windows has no such calls in the standard library. +//! There, links in the parent's path are resolved once, and the +//! parent and the run directory are held open without delete sharing: while +//! they are, neither they nor any directory above them can be renamed or +//! removed, so the paths built on them keep naming the same directories, and +//! [`std::fs::remove_dir_all`] does not follow links inside the tree. Either +//! way, retargeting a link above the run directory mid-run redirects nothing: +//! not the deletion, and not what is written through [`RunDir::path`], which +//! has the links in the parent's path resolved. + +use crate::logging::{debug, info, warn}; +use rand::Rng; +use serde::{Deserialize, Serialize}; +use std::fs::{File, TryLockError}; +use std::io::{self, Read, Write}; +use std::path::{Path, PathBuf}; +use std::time::{SystemTime, UNIX_EPOCH}; + +/// File in a run directory naming the run it belongs to (JSON). +pub const MARKER_FILE: &str = ".ant-devnet-run.json"; + +/// File in a run directory whose exclusive lock the run holds while it lasts. +pub const LOCK_FILE: &str = ".ant-devnet-run.lock"; + +/// The marker is written here first and renamed into place, so a reader sees +/// either no marker or a whole one. +const MARKER_TEMP_FILE: &str = ".ant-devnet-run.json.tmp"; + +/// A marker is a few hundred bytes. Anything larger is not one. +const MAX_MARKER_LEN: usize = 4096; + +/// Names tried before giving up when a random name is already taken. +const CREATE_ATTEMPTS: usize = 8; + +/// What [`MARKER_FILE`] holds. +#[derive(Debug, Serialize, Deserialize)] +struct Marker { + /// Random and unique to the run. + run_id: String, + /// The run directory's file number on its filesystem ([`FileId::file`]). + file_id: u64, + /// The lock file's, so that one put in its place does not pass for it. + lock_id: u64, + /// The process that ran the devnet. + pid: u32, + /// When the run started, in seconds since the Unix epoch. + started_at: u64, + /// The ant-node version that wrote the marker. + version: String, + /// Set by `--no-cleanup`: nothing but the user deletes this directory. + keep: bool, +} + +/// Which directory or file a handle or a name refers to. +#[derive(Debug, PartialEq, Eq)] +struct FileId { + /// Its device (Unix) or volume serial number (Windows). Compared only + /// between handles open at the same time, never kept. + device: String, + /// Its inode (Unix) or file index (Windows). + file: u64, +} + +/// This run's directory: created, locked and marked by [`RunDir::create`]. +/// +/// Dropping it releases the lock and leaves the directory, which a later run +/// then removes unless it is kept. +pub struct RunDir { + path: PathBuf, + name: String, + run_id: String, + parent: Parent, + /// The directory as created, held open until it is deleted. + run: Run, + lock: File, +} + +impl RunDir { + /// Create a fresh run directory inside `parent`, creating `parent` first + /// if it does not exist. Run directories that earlier runs left in + /// `parent`, and that are over and not kept, are removed first. + /// + /// `keep` marks the directory as one that nothing but the user deletes. + /// + /// # Errors + /// + /// Fails if `parent` cannot be created or opened, or, on Unix, another + /// user could change it or a directory above it (see the module docs); or + /// if the run directory cannot be created, locked or marked. + pub fn create(parent: &Path, keep: bool) -> io::Result { + let parent = Parent::open(parent)?; + sweep(&parent); + + let pid = std::process::id(); + let mut rng = rand::thread_rng(); + let mut attempt = 0; + let name = loop { + let name = format!("{pid}-{:08x}", rng.gen::()); + match parent.create_run(&name) { + Ok(()) => break name, + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { + attempt += 1; + if attempt == CREATE_ATTEMPTS { + return Err(e); + } + } + Err(e) => return Err(e), + } + }; + + #[cfg(test)] + after_create(&parent.path.join(&name))?; + let run = parent.open_run(&name)?; + // Between its creation and its opening, something else could have put + // a directory of its own in place of this one. If what was opened has + // anything in it, or another user owns it, it is not this run's: + // leave it as it is. + #[cfg(unix)] + let replaced = !run.is_empty()? || !run.is_owned_by_this_user()?; + #[cfg(windows)] + let replaced = !run.is_empty()?; + if replaced { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!( + "{} was replaced before this run could open it", + parent.path.join(&name).display() + ), + )); + } + let claimed = run.id().and_then(|id| { + let mut marker = Marker { + run_id: hex::encode(rng.gen::<[u8; 16]>()), + file_id: id.file, + // Set by `claim`, once the lock file exists. + lock_id: 0, + pid, + started_at: SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |since| since.as_secs()), + version: env!("CARGO_PKG_VERSION").to_string(), + keep, + }; + let lock = claim(&run, &mut marker)?; + Ok((marker.run_id, lock)) + }); + let (run_id, lock) = match claimed { + Ok(claimed) => claimed, + Err(e) => { + // Unmarked, it would be left alone for good, so remove it, but + // only if it is empty: claiming has removed what it created, + // and anything else in it is not known to be this run's. + if let Err(cleanup) = + same_dir(&parent, &name, &run).and_then(|()| parent.remove_empty(&name, run)) + { + warn!("Could not remove half-created devnet run directory {name}: {cleanup}"); + } + return Err(e); + } + }; + + Ok(Self { + path: parent.path.join(&name), + name, + run_id, + parent, + run, + lock, + }) + } + + /// The run directory, with any link in the parent's path resolved, so + /// that retargeting one does not redirect what is written through it. + pub fn path(&self) -> &Path { + &self.path + } + + /// Delete the run directory, if its name still refers to the directory + /// this run created and the marker there still names this run. + /// + /// # Errors + /// + /// Fails, deleting nothing, if the directory is no longer this run's: + /// moved, replaced, or marked by another run. Otherwise fails if the + /// deletion does, leaving the marker and lock for the next run's sweep. + pub fn remove(self) -> io::Result<()> { + let Self { + path, + name, + run_id, + parent, + run, + lock, + } = self; + let marker = read_marker(&run)?; + if marker.run_id != run_id || marker.file_id != run.id()?.file { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "{} is no longer this run's directory: its marker names another run", + path.display() + ), + )); + } + delete(&parent, &name, run, lock) + } + + /// Leave the directory, and keep it locked until the process exits, when + /// the OS releases the lock. For a run that may still have nodes writing + /// into it: no sweep deletes it while it is locked. + pub fn keep_locked_until_exit(self) { + let Self { lock, .. } = self; + std::mem::forget(lock); + } +} + +/// Lock `run`, then mark it, recording the lock file in `marker`. Marking +/// second is what lets a free lock mean the run is over: a marked run has +/// always taken its lock already. Should it fail, it removes what it created, +/// and nothing else. +fn claim(run: &Run, marker: &mut Marker) -> io::Result { + let lock = run.create_file(LOCK_FILE)?; + let outcome = lock.try_lock().map_err(io::Error::from).and_then(|()| { + marker.lock_id = file_id(&lock)?.file; + write_marker(run, marker) + }); + match outcome { + Ok(()) => Ok(lock), + Err(e) => { + drop(lock); + if let Err(cleanup) = run.remove_file(LOCK_FILE) { + warn!("Could not remove the lock file of a devnet run that failed to start: {cleanup}"); + } + Err(e) + } + } +} + +/// Write `marker` under a temporary name, then rename it into place. +fn write_marker(run: &Run, marker: &Marker) -> io::Result<()> { + let json = serde_json::to_vec_pretty(marker).map_err(io::Error::other)?; + let mut temp = run.create_file(MARKER_TEMP_FILE)?; + let written = temp.write_all(&json).and_then(|()| temp.sync_all()); + drop(temp); + let renamed = written.and_then(|()| run.rename(MARKER_TEMP_FILE, MARKER_FILE)); + if renamed.is_err() { + if let Err(cleanup) = run.remove_file(MARKER_TEMP_FILE) { + warn!("Could not remove the marker of a devnet run that failed to start: {cleanup}"); + } + } + renamed +} + +fn read_marker(run: &Run) -> io::Result { + let mut json = Vec::new(); + run.open_file(MARKER_FILE)? + .take(MAX_MARKER_LEN as u64 + 1) + .read_to_end(&mut json)?; + if json.len() > MAX_MARKER_LEN { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "the marker is too large to be one", + )); + } + serde_json::from_slice(&json).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) +} + +/// Fail unless `name` in `parent` is still the directory `run` has open. +fn same_dir(parent: &Parent, name: &str, run: &Run) -> io::Result<()> { + if parent.id_of(name)? == run.id()? { + Ok(()) + } else { + Err(io::Error::new( + io::ErrorKind::InvalidData, + "it has been moved or replaced since it was opened", + )) + } +} + +/// Delete the run directory `name`, which `run` has open and whose `lock` is +/// held, if `name` still refers to it: the data first, then the marker, then +/// the lock, then the directory, empty by then. Should it stop before the +/// marker goes, the marker and the free lock make the rest a finished run +/// that the next sweep removes. After that, only an empty directory and an +/// empty lock file can be left. +fn delete(parent: &Parent, name: &str, run: Run, lock: File) -> io::Result<()> { + // Moved since it was opened, it would be emptied through `run` wherever + // it went. Once this has passed, only the user or root can still move it: + // no one else can change the parent (see the module docs), and on Windows + // the handle keeps it where it is. + same_dir(parent, name, &run)?; + run.remove_all_but(&[MARKER_FILE, LOCK_FILE])?; + run.remove_file(MARKER_FILE)?; + // Windows removes no directory with a file open in it. + drop(lock); + run.remove_file(LOCK_FILE)?; + // Only an empty directory is removed, so should `name` be swapped between + // this check and the removal, nothing with data in it goes. + same_dir(parent, name, &run)?; + parent.remove_empty(name, run) +} + +/// Whether `name` has the shape of a run directory's: `{pid}-{8 hex digits}`. +fn is_run_name(name: &str) -> bool { + name.split_once('-').is_some_and(|(pid, random)| { + !pid.is_empty() + && pid.bytes().all(|b| b.is_ascii_digit()) + && random.len() == 8 + && random + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + }) +} + +/// Remove the run directories in `parent` that are over and not kept. Never +/// fails: whatever it cannot read or remove, it leaves. +fn sweep(parent: &Parent) { + let names = match parent.names() { + Ok(names) => names, + Err(e) => { + warn!( + "Could not list {} for finished devnet runs: {e}", + parent.path.display() + ); + return; + } + }; + let mut removed = 0_usize; + for name in names.iter().filter(|name| is_run_name(name)) { + match finished_run(parent, name) { + Ok(Some((run, lock))) => { + #[cfg(test)] + after_found(&parent.path.join(name)); + match delete(parent, name, run, lock) { + Ok(()) => removed += 1, + Err(e) => warn!( + "Could not remove the finished devnet run directory {}: {e}", + parent.path.join(name).display() + ), + } + } + Ok(None) => {} + Err(e) => debug!("Leaving {}: {e}", parent.path.join(name).display()), + } + } + if removed > 0 { + info!( + "Removed {removed} directories left by earlier devnet runs from {}", + parent.path.display() + ); + } +} + +/// `Some`, with its lock taken, if `name` is a run that is over and not kept, +/// so may be deleted. +fn finished_run(parent: &Parent, name: &str) -> io::Result> { + let run = parent.open_run(name)?; + // Another user's is theirs, whatever is in it, and in a parent anyone can + // make directories in, a sticky one, they could still move it. + #[cfg(unix)] + if !run.is_owned_by_this_user()? { + return Ok(None); + } + let marker = read_marker(&run)?; + let pid_in_name = name.split_once('-').map(|(pid, _)| pid); + let id = run.id()?; + if marker.keep + || pid_in_name != Some(marker.pid.to_string().as_str()) + || marker.file_id != id.file + || id.device != parent.id()?.device + { + return Ok(None); + } + // Not a mount wearing a run's name, a bind mount of a finished run's + // directory kept elsewhere for one: deleting through it would empty what + // it shows. (On Windows a mounted folder is a link, which `open_run` + // refuses.) + #[cfg(unix)] + if run.mount()? != parent.mount()? { + return Ok(None); + } + let lock = run.open_file(LOCK_FILE)?; + // The run's lock stays held under whatever name its file is moved to, so + // a free lock means something only if it is that file. + let lock_id = file_id(&lock)?; + let the_runs_lock = lock_id.file == marker.lock_id; + if !the_runs_lock || lock_id.device != id.device { + return Ok(None); + } + match lock.try_lock() { + // Held while the directory is deleted. No run takes this lock again: + // a run only ever locks a directory it has just created. + Ok(()) => Ok(Some((run, lock))), + Err(TryLockError::WouldBlock) => Ok(None), + Err(TryLockError::Error(e)) => Err(e), + } +} + +#[cfg(test)] +thread_local! { + /// Tests set this to make a deletion fail after removing this many + /// entries, as one interrupted part-way would. + static FAIL_AFTER: std::cell::Cell> = const { std::cell::Cell::new(None) }; + /// Tests set this to act on a new run directory, by its path, between + /// its creation and its opening, as something else could. + static AFTER_CREATE: std::cell::RefCell> = + const { std::cell::RefCell::new(None) }; + /// Tests set this to act on a finished run's directory, by its path, + /// once the sweep has found it finished and before it deletes it. + static AFTER_FOUND: std::cell::RefCell> = + const { std::cell::RefCell::new(None) }; + /// Tests set this to have a directory of this name inside a run + /// directory count as a mount point. + #[cfg(unix)] + static MOUNT_POINT: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; +} + +#[cfg(test)] +type AfterCreate = Box io::Result<()>>; + +#[cfg(test)] +type AfterFound = Box; + +#[cfg(test)] +fn after_create(path: &Path) -> io::Result<()> { + AFTER_CREATE + .with(|act| act.borrow_mut().take()) + .map_or(Ok(()), |act| act(path)) +} + +#[cfg(test)] +fn after_found(path: &Path) { + if let Some(act) = AFTER_FOUND.with(|act| act.borrow_mut().take()) { + act(path); + } +} + +#[cfg(all(test, unix))] +fn is_test_mount_point(name: &std::ffi::CStr) -> bool { + MOUNT_POINT.with(|mount| mount.borrow().as_deref().map(str::as_bytes) == Some(name.to_bytes())) +} + +/// Called before each entry a deletion removes, to fail where a test asks. +#[cfg(test)] +fn injected_failure() -> io::Result<()> { + FAIL_AFTER.with(|left| match left.get() { + Some(0) => Err(io::Error::other("injected failure")), + Some(n) => { + left.set(Some(n - 1)); + Ok(()) + } + None => Ok(()), + }) +} + +#[cfg(unix)] +use unix::{file_id, Parent, Run}; + +#[cfg(windows)] +use windows::{file_id, Parent, Run}; + +/// Every operation relative to a directory handle, never following a link. +#[cfg(unix)] +mod unix { + use super::FileId; + use rustix::fs::{AtFlags, Dev, FileType, Mode, OFlags, Stat}; + use rustix::io::Errno; + use std::ffi::{CStr, CString}; + use std::fs::File; + use std::io; + use std::os::fd::OwnedFd; + use std::path::{Path, PathBuf}; + + const DIR_FLAGS: OFlags = OFlags::RDONLY + .union(OFlags::DIRECTORY) + .union(OFlags::NOFOLLOW) + .union(OFlags::CLOEXEC); + + /// The directory run directories are created in, opened once. + pub struct Parent { + /// The path it was opened by, with every link in it resolved. + pub path: PathBuf, + fd: OwnedFd, + } + + /// A run directory, opened without following a link. + pub struct Run { + fd: OwnedFd, + } + + impl Parent { + pub fn open(path: &Path) -> io::Result { + std::fs::create_dir_all(path)?; + let real = std::fs::canonicalize(path)?; + no_one_else_can_change(&real)?; + let fd = rustix::fs::open( + &real, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, + Mode::empty(), + )?; + // The run directory's path is built on `real`, so it must still + // be free of links, and name the directory just opened. + if std::fs::canonicalize(&real)? != real + || id(&rustix::fs::stat(&real)?) != id(&rustix::fs::fstat(&fd)?) + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{} changed while it was being opened", path.display()), + )); + } + Ok(Self { path: real, fd }) + } + + pub fn mount(&self) -> io::Result { + mount_of(&self.fd) + } + + pub fn create_run(&self, name: &str) -> io::Result<()> { + Ok(rustix::fs::mkdirat( + &self.fd, + name, + Mode::from_raw_mode(0o755), + )?) + } + + pub fn open_run(&self, name: &str) -> io::Result { + let fd = rustix::fs::openat(&self.fd, name, DIR_FLAGS, Mode::empty())?; + Ok(Run { fd }) + } + + pub fn names(&self) -> io::Result> { + Ok(read_names(&self.fd)? + .into_iter() + .filter_map(|name| name.into_string().ok()) + .collect()) + } + + pub fn id(&self) -> io::Result { + Ok(id(&rustix::fs::fstat(&self.fd)?)) + } + + /// The directory `name`, not following a link. + pub fn id_of(&self, name: &str) -> io::Result { + let stat = rustix::fs::statat(&self.fd, name, AtFlags::SYMLINK_NOFOLLOW)?; + if FileType::from_raw_mode(stat.st_mode) != FileType::Directory { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{name} is not a directory"), + )); + } + Ok(id(&stat)) + } + + /// Close `run`, then remove the entry `name` if it is an empty + /// directory. + pub fn remove_empty(&self, name: &str, run: Run) -> io::Result<()> { + drop(run); + gone_is_fine(rustix::fs::unlinkat(&self.fd, name, AtFlags::REMOVEDIR)) + } + } + + impl Run { + pub fn id(&self) -> io::Result { + Ok(id(&rustix::fs::fstat(&self.fd)?)) + } + + pub fn is_empty(&self) -> io::Result { + Ok(read_names(&self.fd)?.is_empty()) + } + + pub fn is_owned_by_this_user(&self) -> io::Result { + Ok(rustix::fs::fstat(&self.fd)?.st_uid == rustix::process::geteuid().as_raw()) + } + + pub fn mount(&self) -> io::Result { + mount_of(&self.fd) + } + + pub fn create_file(&self, name: &str) -> io::Result { + let fd = rustix::fs::openat( + &self.fd, + name, + OFlags::RDWR | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::CLOEXEC, + Mode::from_raw_mode(0o644), + )?; + Ok(File::from(fd)) + } + + /// Open an existing regular file for reading and locking. `NONBLOCK` + /// so that a FIFO wearing the name cannot stall the open. + pub fn open_file(&self, name: &str) -> io::Result { + let fd = rustix::fs::openat( + &self.fd, + name, + OFlags::RDWR | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC, + Mode::empty(), + )?; + if FileType::from_raw_mode(rustix::fs::fstat(&fd)?.st_mode) != FileType::RegularFile { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{name} is not a regular file"), + )); + } + Ok(File::from(fd)) + } + + pub fn rename(&self, from: &str, to: &str) -> io::Result<()> { + Ok(rustix::fs::renameat(&self.fd, from, &self.fd, to)?) + } + + pub fn remove_file(&self, name: &str) -> io::Result<()> { + gone_is_fine(rustix::fs::unlinkat(&self.fd, name, AtFlags::empty())) + } + + /// Delete everything in the directory but the entries named in `keep`. + pub fn remove_all_but(&self, keep: &[&str]) -> io::Result<()> { + let mount = mount_of(&self.fd)?; + for name in read_names(&self.fd)? { + if keep.iter().any(|keep| keep.as_bytes() == name.as_bytes()) { + continue; + } + #[cfg(test)] + super::injected_failure()?; + remove_entry(&self.fd, &name, &mount)?; + } + Ok(()) + } + } + + pub fn file_id(file: &File) -> io::Result { + Ok(id(&rustix::fs::fstat(file)?)) + } + + fn id(stat: &Stat) -> FileId { + FileId { + device: stat.st_dev.to_string(), + file: stat.st_ino, + } + } + + /// Fail unless only the user and root can change `real`: neither it nor + /// any directory above it may belong to another user, or be one another + /// user can write to, unless it is sticky, which lets no one rename what + /// they do not own. Whoever could write to one could move what is in it, + /// a run directory or a directory above one. + fn no_one_else_can_change(real: &Path) -> io::Result<()> { + let uid = rustix::process::geteuid().as_raw(); + for dir in real.ancestors() { + let stat = rustix::fs::lstat(dir)?; + if FileType::from_raw_mode(stat.st_mode) != FileType::Directory { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{} changed while it was being opened", real.display()), + )); + } + let mode = Mode::from_raw_mode(stat.st_mode); + let who = if stat.st_uid != 0 && stat.st_uid != uid { + Some(format!("belongs to user {}", stat.st_uid)) + } else if mode.contains(Mode::SVTX) { + None + } else if mode.contains(Mode::WOTH) { + Some("can be written to by every user".to_string()) + } else if mode.contains(Mode::WGRP) { + group_writers(dir, &stat)? + } else { + None + }; + if let Some(who) = who { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + format!( + "{} {who}, so another user could move or redirect a devnet's \ + data under it: use a data directory that only you and root \ + can change, with only such directories above it", + dir.display() + ), + )); + } + } + Ok(()) + } + + /// Who but the user can write to `dir`, which its group can write to. + fn group_writers(dir: &Path, stat: &Stat) -> io::Result> { + // On Linux, the user's own group, which with user private groups, + // Ubuntu's default, is the user alone. macOS has none: every user's + // group there is `staff`. + #[cfg(target_os = "linux")] + if stat.st_gid == rustix::process::getegid().as_raw() { + // An ACL shows in the group bits only as its mask, the most it + // grants anyone it names. + return match rustix::fs::lgetxattr(dir, "system.posix_acl_access", &mut [0_u8; 0]) { + Ok(_) => Ok(Some( + "has an ACL, which can let other users write to it".to_string(), + )), + Err(Errno::NODATA | Errno::OPNOTSUPP) => Ok(None), + Err(e) => Err(e.into()), + }; + } + #[cfg(not(target_os = "linux"))] + let _ = dir; + Ok(Some(format!("can be written to by group {}", stat.st_gid))) + } + + /// Where a directory is mounted: its device and, on Linux, the mount + /// itself, which also tells a bind mount from the filesystem it shows. + #[derive(Debug, PartialEq, Eq)] + pub struct Mount { + device: Dev, + #[cfg(target_os = "linux")] + id: u64, + } + + /// Fails where mounts cannot be told apart, so that nothing is deleted + /// there. + pub(super) fn mount_of(dir: &OwnedFd) -> io::Result { + let device = rustix::fs::fstat(dir)?.st_dev; + #[cfg(target_os = "linux")] + { + use rustix::fs::StatxFlags; + let unsupported = || { + io::Error::new( + io::ErrorKind::Unsupported, + "cannot tell a bind mount from the directory it is in: that needs \ + mount ids, from Linux 5.8, and `statx` allowed", + ) + }; + let statx = match rustix::fs::statx(dir, c"", AtFlags::EMPTY_PATH, StatxFlags::MNT_ID) { + Ok(statx) => statx, + Err(Errno::NOSYS | Errno::PERM) => return Err(unsupported()), + Err(e) => return Err(e.into()), + }; + if statx.stx_mask & StatxFlags::MNT_ID.bits() == 0 { + return Err(unsupported()); + } + Ok(Mount { + device, + id: statx.stx_mnt_id, + }) + } + // No bind mounts on macOS: another mount is another device. + #[cfg(target_os = "macos")] + { + Ok(Mount { device }) + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + { + let _ = device; + Err(io::Error::new( + io::ErrorKind::Unsupported, + "cannot tell mounts apart on this system", + )) + } + } + + /// Delete the entry `name` in `dir`, descending only through handles + /// opened without following a link, and only on `mount`: a link is + /// removed, never entered, and a mount point is left, with all in it. + fn remove_entry(dir: &OwnedFd, name: &CStr, mount: &Mount) -> io::Result<()> { + let stat = match rustix::fs::statat(dir, name, AtFlags::SYMLINK_NOFOLLOW) { + Ok(stat) => stat, + Err(Errno::NOENT) => return Ok(()), + Err(e) => return Err(e.into()), + }; + if FileType::from_raw_mode(stat.st_mode) != FileType::Directory { + return gone_is_fine(rustix::fs::unlinkat(dir, name, AtFlags::empty())); + } + let child = match rustix::fs::openat(dir, name, DIR_FLAGS, Mode::empty()) { + Ok(child) => child, + Err(Errno::NOENT) => return Ok(()), + Err(e) => return Err(e.into()), + }; + #[cfg(not(test))] + let mount_point = mount_of(&child)? != *mount; + #[cfg(test)] + let mount_point = mount_of(&child)? != *mount || super::is_test_mount_point(name); + if mount_point { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "{} is a mount point, so it and what is in it are left", + name.to_string_lossy() + ), + )); + } + for entry in read_names(&child)? { + remove_entry(&child, &entry, mount)?; + } + gone_is_fine(rustix::fs::unlinkat(dir, name, AtFlags::REMOVEDIR)) + } + + /// Every name in `dir` but `.` and `..`. Collected before anything is + /// deleted: unlinking while a directory stream is open is not something + /// every filesystem keeps coherent. + fn read_names(dir: &OwnedFd) -> io::Result> { + let mut names = Vec::new(); + for entry in rustix::fs::Dir::read_from(dir)? { + let name = entry?.file_name().to_owned(); + if name.as_bytes() != b"." && name.as_bytes() != b".." { + names.push(name); + } + } + Ok(names) + } + + /// Something else, another run's sweep for one, may delete first. + fn gone_is_fine(result: rustix::io::Result<()>) -> io::Result<()> { + match result { + Ok(()) | Err(Errno::NOENT) => Ok(()), + Err(e) => Err(e.into()), + } + } +} + +/// Paths built on directories held open without delete sharing, which keeps +/// them, and every directory above them, from being renamed or removed. +#[cfg(windows)] +mod windows { + use super::FileId; + use std::fs::{File, OpenOptions}; + use std::io; + use std::os::windows::fs::{MetadataExt, OpenOptionsExt}; + use std::path::{Path, PathBuf}; + + /// Lets `CreateFileW` open a directory. + const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000; + /// Opens a link itself rather than what it points at. + const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000; + const FILE_ATTRIBUTE_DIRECTORY: u32 = 0x10; + const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x400; + /// `FILE_SHARE_READ | FILE_SHARE_WRITE`, without `FILE_SHARE_DELETE`. + const SHARE_ALL_BUT_DELETE: u32 = 0x1 | 0x2; + + /// The directory run directories are created in, held open. + pub struct Parent { + /// `real`, without the `\\?\` prefix where it means the same. + pub path: PathBuf, + /// The path it was opened by, with every link in it resolved. + real: PathBuf, + held: File, + } + + /// A run directory, held open. + pub struct Run { + path: PathBuf, + held: File, + } + + impl Parent { + pub fn open(path: &Path) -> io::Result { + std::fs::create_dir_all(path)?; + let real = std::fs::canonicalize(path)?; + let held = hold_dir(&real)?; + // Held, none of its directories can be renamed or removed any + // more, so if `real` has no link in it now, it never will. + if std::fs::canonicalize(&real)? != real { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{} changed while it was being opened", path.display()), + )); + } + Ok(Self { + path: plain(&real), + real, + held, + }) + } + + pub fn create_run(&self, name: &str) -> io::Result<()> { + std::fs::create_dir(self.real.join(name)) + } + + pub fn open_run(&self, name: &str) -> io::Result { + let path = self.real.join(name); + let held = hold_dir(&path)?; + Ok(Run { path, held }) + } + + pub fn names(&self) -> io::Result> { + let mut names = Vec::new(); + for entry in std::fs::read_dir(&self.real)? { + if let Ok(name) = entry?.file_name().into_string() { + names.push(name); + } + } + Ok(names) + } + + pub fn id(&self) -> io::Result { + file_id(&self.held) + } + + /// The directory `name`, not following a link. + pub fn id_of(&self, name: &str) -> io::Result { + file_id(&hold_dir(&self.real.join(name))?) + } + + /// Close `run`, then remove the entry `name` if it is an empty + /// directory or a link. + pub fn remove_empty(&self, name: &str, run: Run) -> io::Result<()> { + drop(run); + gone_is_fine(std::fs::remove_dir(self.real.join(name))) + } + } + + impl Run { + pub fn id(&self) -> io::Result { + file_id(&self.held) + } + + pub fn is_empty(&self) -> io::Result { + Ok(std::fs::read_dir(&self.path)?.next().is_none()) + } + + pub fn create_file(&self, name: &str) -> io::Result { + OpenOptions::new() + .read(true) + .write(true) + .create_new(true) + .open(self.path.join(name)) + } + + /// Open an existing regular file for reading and locking, not + /// following a link. + pub fn open_file(&self, name: &str) -> io::Result { + let file = OpenOptions::new() + .read(true) + .write(true) + .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT) + .open(self.path.join(name))?; + let meta = file.metadata()?; + if !meta.is_file() || meta.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{name} is not a regular file"), + )); + } + Ok(file) + } + + pub fn rename(&self, from: &str, to: &str) -> io::Result<()> { + std::fs::rename(self.path.join(from), self.path.join(to)) + } + + pub fn remove_file(&self, name: &str) -> io::Result<()> { + gone_is_fine(std::fs::remove_file(self.path.join(name))) + } + + /// Delete everything in the directory but the entries named in `keep`. + pub fn remove_all_but(&self, keep: &[&str]) -> io::Result<()> { + for entry in std::fs::read_dir(&self.path)? { + let entry = entry?; + if keep.iter().any(|keep| entry.file_name() == *keep) { + continue; + } + #[cfg(test)] + super::injected_failure()?; + // An entry's metadata describes a link, not what it points + // at. A link to a directory counts as a directory, and + // `remove_dir_all` removes the link itself. + let path = entry.path(); + gone_is_fine( + if entry.metadata()?.file_attributes() & FILE_ATTRIBUTE_DIRECTORY == 0 { + std::fs::remove_file(path) + } else { + std::fs::remove_dir_all(path) + }, + )?; + } + Ok(()) + } + } + + /// `real` without its `\\?\` prefix, if it names the same without it, + /// as a path on a drive does: [`super::RunDir::path`] is shown in logs and + /// the manifest, and read by other programs. + fn plain(real: &Path) -> PathBuf { + real.to_str() + .and_then(|text| text.strip_prefix(r"\\?\")) + .map(PathBuf::from) + .filter(|plain| std::fs::canonicalize(plain).is_ok_and(|again| again == real)) + .unwrap_or_else(|| real.to_path_buf()) + } + + /// Open the directory `path` to hold it, refusing a link. + fn hold_dir(path: &Path) -> io::Result { + let dir = OpenOptions::new() + .read(true) + .share_mode(SHARE_ALL_BUT_DELETE) + .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT) + .open(path)?; + let meta = dir.metadata()?; + if !meta.is_dir() || meta.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("{} is not a directory", path.display()), + )); + } + Ok(dir) + } + + pub fn file_id(file: &File) -> io::Result { + let info = winapi_util::file::information(file)?; + Ok(FileId { + device: info.volume_serial_number().to_string(), + file: info.file_index(), + }) + } + + /// Something else, another run's sweep for one, may delete first. + fn gone_is_fine(result: io::Result<()>) -> io::Result<()> { + match result { + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), + result => result, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + type TestResult = std::result::Result<(), Box>; + + fn marker_in(dir: &Path) -> std::result::Result> { + Ok(serde_json::from_slice(&std::fs::read( + dir.join(MARKER_FILE), + )?)?) + } + + /// Whether another handle can take `dir`'s lock: false while its run lasts. + fn lock_is_free(dir: &Path) -> std::result::Result> { + let lock = File::options() + .read(true) + .write(true) + .open(dir.join(LOCK_FILE))?; + match lock.try_lock() { + Ok(()) => Ok(true), + Err(TryLockError::WouldBlock) => Ok(false), + Err(TryLockError::Error(e)) => Err(e.into()), + } + } + + fn file_id_of(path: &Path) -> std::result::Result> { + Ok(file_id(&File::open(path)?)?.file) + } + + /// A finished run's directory, as a run killed by `SIGKILL` leaves it. + fn finished( + parent: &Path, + keep: bool, + ) -> std::result::Result> { + let run = RunDir::create(parent, keep)?; + let path = run.path().to_path_buf(); + std::fs::create_dir_all(path.join("nodes").join("peer"))?; + std::fs::write(path.join("nodes").join("peer").join("chunk"), b"data")?; + drop(run); + Ok(path) + } + + /// A directory `name` in `parent` carrying a free lock and a marker that + /// names it, for this process, as a finished run would, then `edit`ed; + /// and a sentinel file. + fn forged( + parent: &Path, + name: &str, + edit: impl FnOnce(&mut Marker), + ) -> std::result::Result> { + let dir = parent.join(name); + std::fs::create_dir_all(&dir)?; + let file_id = Parent::open(parent)?.open_run(name)?.id()?.file; + std::fs::write(dir.join(LOCK_FILE), b"")?; + let lock_id = file_id_of(&dir.join(LOCK_FILE))?; + let mut marker = Marker { + run_id: "forged".to_string(), + file_id, + lock_id, + pid: std::process::id(), + started_at: 0, + version: env!("CARGO_PKG_VERSION").to_string(), + keep: false, + }; + edit(&mut marker); + std::fs::write(dir.join(MARKER_FILE), serde_json::to_vec(&marker)?)?; + std::fs::write(dir.join("sentinel"), b"keep")?; + Ok(dir) + } + + #[test] + fn create_makes_a_locked_marked_directory_inside_the_parent() -> TestResult { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("not").join("there").join("yet"); + + let run = RunDir::create(&parent, false)?; + + assert_eq!( + run.path().parent().map(std::fs::canonicalize).transpose()?, + Some(std::fs::canonicalize(&parent)?) + ); + #[cfg(windows)] + assert!( + !run.path().to_string_lossy().starts_with(r"\\?\"), + "{}", + run.path().display() + ); + let name = run + .path() + .file_name() + .and_then(|name| name.to_str()) + .ok_or("no name")?; + assert!(is_run_name(name), "{name}"); + let marker = marker_in(run.path())?; + assert_eq!(marker.pid, std::process::id()); + assert_eq!(marker.run_id, run.run_id); + assert_eq!(marker.file_id, run.run.id()?.file); + assert_eq!(marker.lock_id, file_id(&run.lock)?.file); + assert!(!marker.keep); + assert!(!run.path().join(MARKER_TEMP_FILE).exists()); + assert!( + !lock_is_free(run.path())?, + "the lock is free while the run lasts" + ); + Ok(()) + } + + #[test] + fn remove_deletes_the_run_directory_and_nothing_beside_it() -> TestResult { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("devnets"); + std::fs::create_dir_all(parent.join("mine"))?; + std::fs::write(parent.join("mine").join("sentinel"), b"keep")?; + std::fs::write(temp.path().join("sentinel"), b"keep")?; + + let run = RunDir::create(&parent, false)?; + let path = run.path().to_path_buf(); + std::fs::create_dir_all(path.join("nodes").join("peer").join("deeper"))?; + std::fs::write( + path.join("nodes").join("peer").join("deeper").join("f"), + b"x", + )?; + run.remove()?; + + assert!(!path.exists()); + assert!(parent.is_dir(), "the parent is not the run's to delete"); + assert!(parent.join("mine").join("sentinel").is_file()); + assert!(temp.path().join("sentinel").is_file()); + Ok(()) + } + + #[test] + fn remove_refuses_a_directory_another_run_has_marked() -> TestResult { + let temp = tempfile::tempdir()?; + let run = RunDir::create(temp.path(), false)?; + let path = run.path().to_path_buf(); + let mut marker = marker_in(&path)?; + marker.run_id = "someone-else".to_string(); + std::fs::write(path.join(MARKER_FILE), serde_json::to_vec(&marker)?)?; + + assert!(run.remove().is_err()); + assert!(path.join(MARKER_FILE).is_file()); + Ok(()) + } + + /// Hermes round 3 on #263: once the run directory has been moved away, a + /// directory put in its place, with copies of its marker and lock, is not + /// the run's, and nothing is deleted. + #[cfg(unix)] + #[test] + fn remove_refuses_a_replacement_carrying_a_copy_of_the_marker() -> TestResult { + let temp = tempfile::tempdir()?; + let run = RunDir::create(temp.path(), false)?; + let path = run.path().to_path_buf(); + std::fs::write(path.join("data"), b"the run's")?; + let moved = temp.path().join("moved"); + std::fs::rename(&path, &moved)?; + std::fs::create_dir(&path)?; + std::fs::copy(moved.join(MARKER_FILE), path.join(MARKER_FILE))?; + std::fs::copy(moved.join(LOCK_FILE), path.join(LOCK_FILE))?; + std::fs::write(path.join("sentinel"), b"not the run's")?; + + assert!(run.remove().is_err()); + assert!(path.join("sentinel").is_file()); + assert!(path.join(MARKER_FILE).is_file()); + assert!(moved.join("data").is_file()); + Ok(()) + } + + /// On Windows the same replacement cannot even be set up: while a run + /// lasts, neither its directory nor the parent can be moved. + /// Hermes round 6 on #263: nor can any directory above the parent, so + /// `RunDir::path` keeps naming the run directory. + #[cfg(windows)] + #[test] + fn a_run_directory_and_every_directory_above_it_cannot_be_moved_while_the_run_lasts( + ) -> TestResult { + let temp = tempfile::tempdir()?; + let above = temp.path().join("above"); + let parent = above.join("devnets"); + let run = RunDir::create(&parent, false)?; + let path = run.path().to_path_buf(); + + assert!(std::fs::rename(&path, parent.join("moved")).is_err()); + assert!(std::fs::rename(&parent, above.join("moved")).is_err()); + assert!(std::fs::rename(&above, temp.path().join("moved")).is_err()); + let temp_moved = temp.path().with_extension("moved"); + assert!(std::fs::rename(temp.path(), &temp_moved).is_err()); + + run.remove()?; + assert!(!path.exists()); + Ok(()) + } + + #[test] + fn the_next_run_sweeps_a_finished_run() -> TestResult { + let temp = tempfile::tempdir()?; + let old = finished(temp.path(), false)?; + assert!(old.is_dir()); + + let run = RunDir::create(temp.path(), false)?; + + assert!(!old.exists(), "{} was not swept", old.display()); + assert!(run.path().is_dir()); + Ok(()) + } + + /// Hermes round 3 on #263: a deletion that stops part-way leaves the + /// marker and the lock, so the next run still knows the rest is a + /// finished run's, and removes it. + #[test] + fn a_removal_that_stops_part_way_is_finished_by_the_next_run() -> TestResult { + let temp = tempfile::tempdir()?; + let run = RunDir::create(temp.path(), false)?; + let path = run.path().to_path_buf(); + for i in 0..4 { + std::fs::create_dir_all(path.join(format!("node-{i}")))?; + std::fs::write(path.join(format!("node-{i}")).join("chunk"), b"data")?; + } + + FAIL_AFTER.with(|left| left.set(Some(2))); + let stopped = run.remove(); + FAIL_AFTER.with(|left| left.set(None)); + + assert!(stopped.is_err()); + assert!(path.join(MARKER_FILE).is_file()); + assert!(path.join(LOCK_FILE).is_file()); + let _next = RunDir::create(temp.path(), false)?; + assert!(!path.exists(), "{} was not swept", path.display()); + Ok(()) + } + + #[test] + fn a_live_run_is_never_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let first = RunDir::create(temp.path(), false)?; + let second = RunDir::create(temp.path(), false)?; + + assert!(first.path().join(MARKER_FILE).is_file()); + assert!(second.path().join(MARKER_FILE).is_file()); + assert_ne!(first.path(), second.path()); + Ok(()) + } + + #[test] + fn a_run_kept_locked_until_exit_is_never_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let run = RunDir::create(temp.path(), false)?; + let path = run.path().to_path_buf(); + run.keep_locked_until_exit(); + + let _next = RunDir::create(temp.path(), false)?; + + assert!(path.join(MARKER_FILE).is_file()); + assert!(!lock_is_free(&path)?); + Ok(()) + } + + /// Hermes round 4 on #263: a live run's lock file moved aside and + /// another put in its place. That one is free, but it is not the lock the + /// run holds. + #[test] + fn a_live_run_whose_lock_file_is_replaced_is_never_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let run = RunDir::create(temp.path(), false)?; + let path = run.path().to_path_buf(); + std::fs::write(path.join("data"), b"the run's")?; + std::fs::rename(path.join(LOCK_FILE), path.join("moved.lock"))?; + std::fs::write(path.join(LOCK_FILE), b"")?; + assert!(lock_is_free(&path)?); + + let _next = RunDir::create(temp.path(), false)?; + + assert!(path.join("data").is_file(), "{} was swept", path.display()); + drop(run); + Ok(()) + } + + /// Hermes round 4 on #263: a directory put in place of a new run + /// directory before the run opens it is not the run's, so the run neither + /// claims it nor deletes anything in it, whether or not claiming it would + /// have failed. + #[test] + fn a_directory_put_in_place_of_a_new_run_directory_is_left_alone() -> TestResult { + for with_lock in [false, true] { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("devnets"); + let moved = temp.path().join("moved"); + let replaced = std::rc::Rc::new(std::cell::RefCell::new(None)); + let record = std::rc::Rc::clone(&replaced); + AFTER_CREATE.with(|act| { + *act.borrow_mut() = Some(Box::new(move |path: &Path| { + std::fs::rename(path, &moved)?; + std::fs::create_dir(path)?; + std::fs::write(path.join("sentinel"), b"not the run's")?; + if with_lock { + std::fs::write(path.join(LOCK_FILE), b"not the run's")?; + } + *record.borrow_mut() = Some(path.to_path_buf()); + Ok(()) + })); + }); + + let created = RunDir::create(&parent, false); + AFTER_CREATE.with(|act| act.borrow_mut().take()); + + assert!(created.is_err(), "with_lock: {with_lock}"); + let path = replaced.borrow_mut().take().ok_or("never replaced")?; + assert!(path.join("sentinel").is_file(), "with_lock: {with_lock}"); + assert!(!path.join(MARKER_FILE).exists(), "with_lock: {with_lock}"); + assert_eq!(path.join(LOCK_FILE).is_file(), with_lock); + } + Ok(()) + } + + #[test] + fn a_kept_run_is_never_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let kept = finished(temp.path(), true)?; + assert!(marker_in(&kept)?.keep); + + let _run = RunDir::create(temp.path(), false)?; + + assert!(kept.join("nodes").join("peer").join("chunk").is_file()); + Ok(()) + } + + /// Hermes round 3 on #263: a finished run's directory moved away, and + /// another put in its place with a copy of its marker and its very lock + /// file. The copy names the moved directory, not this one. + #[test] + fn a_copy_of_a_finished_runs_marker_is_never_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let old = finished(temp.path(), false)?; + let moved = temp.path().join("moved"); + std::fs::rename(&old, &moved)?; + std::fs::create_dir(&old)?; + std::fs::copy(moved.join(MARKER_FILE), old.join(MARKER_FILE))?; + std::fs::rename(moved.join(LOCK_FILE), old.join(LOCK_FILE))?; + std::fs::write(old.join("sentinel"), b"keep")?; + + let _run = RunDir::create(temp.path(), false)?; + + assert!(old.join("sentinel").is_file()); + assert!(moved.join("nodes").join("peer").join("chunk").is_file()); + Ok(()) + } + + /// Hermes round 6 on #263: a finished run's directory moved out of its + /// name once the sweep has found it finished is not emptied where it went. + /// On Windows the sweep holds it, and it cannot be moved. + #[test] + fn a_finished_run_moved_once_it_is_found_is_left() -> TestResult { + let temp = tempfile::tempdir()?; + let old = finished(temp.path(), false)?; + let moved = temp.path().join("moved"); + let to = moved.clone(); + let outcome = std::rc::Rc::new(std::cell::RefCell::new(None)); + let record = std::rc::Rc::clone(&outcome); + AFTER_FOUND.with(|act| { + *act.borrow_mut() = Some(Box::new(move |path: &Path| { + *record.borrow_mut() = Some(std::fs::rename(path, &to).is_ok()); + })); + }); + + let created = RunDir::create(temp.path(), false); + AFTER_FOUND.with(|act| act.borrow_mut().take()); + let _run = created?; + + let was_moved = outcome.borrow_mut().take().ok_or("never found")?; + if cfg!(windows) { + assert!(!was_moved, "moved while the sweep held it"); + assert!(!old.exists(), "{} was not swept", old.display()); + } else { + assert!(was_moved); + assert!( + moved.join("nodes").join("peer").join("chunk").is_file(), + "emptied where it was moved to" + ); + assert!(moved.join(MARKER_FILE).is_file()); + } + Ok(()) + } + + #[test] + fn directories_without_a_matching_marker_are_never_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let parent = temp.path(); + let pid = std::process::id(); + let survivors = [ + // The right shape but no marker. + { + let dir = parent.join("1-deadbeef"); + std::fs::create_dir_all(&dir)?; + std::fs::write(dir.join("sentinel"), b"keep")?; + dir + }, + // A marker, but not a run directory's name. + forged(parent, "other", |_| {})?, + forged(parent, &format!("{pid}-DEADBEEF"), |_| {})?, + forged(parent, &format!("{pid}-deadbeef0"), |_| {})?, + // A marker naming another process than the name does. + forged(parent, &format!("{}-deadbeef", pid.wrapping_add(1)), |_| {})?, + // A marker naming another directory. + forged(parent, &format!("{pid}-d1ffe4e7"), |marker| { + marker.file_id = marker.file_id.wrapping_add(1); + })?, + // A marker naming another lock file than the one there. + forged(parent, &format!("{pid}-10c4ed00"), |marker| { + marker.lock_id = marker.lock_id.wrapping_add(1); + })?, + // A marker asking to keep it. + forged(parent, &format!("{pid}-cee9cee9"), |marker| { + marker.keep = true; + })?, + // A marker that is not one. + { + let dir = forged(parent, &format!("{pid}-0badf00d"), |_| {})?; + std::fs::write(dir.join(MARKER_FILE), b"{ not json")?; + dir + }, + { + let dir = forged(parent, &format!("{pid}-0ddba11a"), |_| {})?; + let mut big = std::fs::read(dir.join(MARKER_FILE))?; + big.resize(MAX_MARKER_LEN + 1, b' '); + std::fs::write(dir.join(MARKER_FILE), big)?; + dir + }, + // A marker but no lock: a run never marks before it has locked. + { + let dir = forged(parent, &format!("{pid}-ca11ab1e"), |_| {})?; + std::fs::remove_file(dir.join(LOCK_FILE))?; + dir + }, + ]; + // Unedited, the same forgery is swept: each survivor survives for + // its one difference alone. + let control = forged(parent, &format!("{pid}-c0ffee00"), |_| {})?; + + let _run = RunDir::create(parent, false)?; + + for dir in &survivors { + assert!( + dir.join("sentinel").is_file(), + "{} was swept", + dir.display() + ); + } + assert!(!control.exists(), "{} was not swept", control.display()); + Ok(()) + } + + #[test] + fn run_names() { + assert!(is_run_name("1-deadbeef")); + assert!(is_run_name("123456-0000000a")); + for name in [ + "", + "-deadbeef", + "1-", + "1-deadbee", + "1-deadbeef0", + "1-DEADBEEF", + "a-deadbeef", + "1_deadbeef", + "1-deadbeeg", + "1-dead-bee", + ] { + assert!(!is_run_name(name), "{name}"); + } + } + + /// Hermes round 4 on #263: a deletion does not cross into another mount + /// inside the run directory. It stops, leaving the mount point, what is + /// in it, and the marker. + #[cfg(unix)] + #[test] + fn removal_does_not_cross_into_another_mount() -> TestResult { + let temp = tempfile::tempdir()?; + let run = RunDir::create(temp.path(), false)?; + let path = run.path().to_path_buf(); + let mounted = path.join("nodes").join("mounted"); + std::fs::create_dir_all(&mounted)?; + std::fs::write(mounted.join("data"), b"not the run's")?; + + MOUNT_POINT.with(|mount| *mount.borrow_mut() = Some("mounted".to_string())); + let removed = run.remove(); + MOUNT_POINT.with(|mount| mount.borrow_mut().take()); + + assert!(removed.is_err()); + assert!(mounted.join("data").is_file()); + assert!(path.join(MARKER_FILE).is_file()); + Ok(()) + } + + /// Run `sudo -n`, with `args` and then `path`. + #[cfg(target_os = "linux")] + fn sudo(args: &[&std::ffi::OsStr], path: &Path) -> io::Result<()> { + let status = std::process::Command::new("sudo") + .arg("-n") + .args(args) + .arg(path) + .status()?; + if status.success() { + Ok(()) + } else { + Err(io::Error::other(format!( + "sudo {args:?} {} failed", + path.display() + ))) + } + } + + /// A mount made through `sudo`, undone when dropped. + #[cfg(target_os = "linux")] + struct Mounted(PathBuf); + + #[cfg(target_os = "linux")] + impl Mounted { + fn new(at: &Path, how: &[&std::ffi::OsStr]) -> io::Result { + sudo(&[&[std::ffi::OsStr::new("mount")], how].concat(), at)?; + Ok(Self(at.to_path_buf())) + } + } + + #[cfg(target_os = "linux")] + impl Drop for Mounted { + fn drop(&mut self) { + if let Err(e) = sudo(&[std::ffi::OsStr::new("umount")], &self.0) { + eprintln!("{e}"); + } + } + } + + /// Whether `mv from to`, run as `nobody` through `sudo -n`, succeeds. + #[cfg(target_os = "linux")] + fn nobody_moves(from: &Path, to: &Path) -> io::Result { + Ok(std::process::Command::new("sudo") + .args(["-n", "-u", "nobody", "mv"]) + .arg(from) + .arg(to) + .stderr(std::process::Stdio::null()) + .status()? + .success()) + } + + /// Hermes round 5 on #263, with real mounts: neither a tmpfs nor a bind + /// mount of a directory on the same filesystem, which only the mount id + /// tells apart, is crossed. The `with_sudo_` tests need root, through + /// `sudo -n`; CI runs them in a step of their own. + #[cfg(target_os = "linux")] + #[test] + #[ignore = "mounts through sudo; CI runs it"] + fn with_sudo_mounts_inside_a_run_directory_are_not_crossed() -> TestResult { + use std::ffi::OsStr; + for bind in [false, true] { + let temp = tempfile::tempdir()?; + let source = temp.path().join("source"); + std::fs::create_dir(&source)?; + std::fs::write(source.join("data"), b"not the run's")?; + let run = RunDir::create(&temp.path().join("devnets"), false)?; + let path = run.path().to_path_buf(); + let at = path.join("nodes").join("mounted"); + std::fs::create_dir_all(&at)?; + let _mounted = if bind { + Mounted::new(&at, &[OsStr::new("--bind"), source.as_os_str()])? + } else { + let tmpfs = ["-t", "tmpfs", "-o", "mode=1777", "none"].map(OsStr::new); + let mounted = Mounted::new(&at, &tmpfs)?; + std::fs::write(at.join("data"), b"not the run's")?; + mounted + }; + + assert!(run.remove().is_err(), "bind: {bind}"); + assert!(at.join("data").is_file(), "bind: {bind}"); + assert!(source.join("data").is_file(), "bind: {bind}"); + assert!(path.join(MARKER_FILE).is_file(), "bind: {bind}"); + } + Ok(()) + } + + /// Hermes round 5 on #263: a bind mount of a finished run's directory, + /// kept elsewhere on the same filesystem, put in the parent under the + /// run's name. Everything about it matches but its mount, and the sweep + /// would delete through it what it shows. + #[cfg(target_os = "linux")] + #[test] + #[ignore = "mounts through sudo; CI runs it"] + fn with_sudo_a_mount_wearing_a_finished_runs_name_is_not_swept() -> TestResult { + let temp = tempfile::tempdir()?; + let outside = finished(&temp.path().join("outside"), false)?; + let parent = temp.path().join("devnets"); + let at = parent.join(outside.file_name().ok_or("no name")?); + std::fs::create_dir_all(&at)?; + let _mounted = Mounted::new(&at, &[std::ffi::OsStr::new("--bind"), outside.as_os_str()])?; + assert!(at.join(MARKER_FILE).is_file()); + + let _run = RunDir::create(&parent, false)?; + + assert!( + outside.join("nodes").join("peer").join("chunk").is_file(), + "swept through the mount" + ); + Ok(()) + } + + /// Hermes round 5 on #263: an empty directory another user owns, put in + /// place of a new run directory before the run opens it, is not claimed, + /// though the run could write in it. + #[cfg(target_os = "linux")] + #[test] + #[ignore = "makes a directory another user owns through sudo; CI runs it"] + fn with_sudo_an_empty_directory_of_another_user_is_not_claimed() -> TestResult { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("devnets"); + let moved = temp.path().join("moved"); + let replaced = std::rc::Rc::new(std::cell::RefCell::new(None)); + let record = std::rc::Rc::clone(&replaced); + AFTER_CREATE.with(|act| { + *act.borrow_mut() = Some(Box::new(move |path: &Path| { + std::fs::rename(path, &moved)?; + // root's, and open to anyone. + sudo(&["mkdir", "-m", "0777"].map(std::ffi::OsStr::new), path)?; + *record.borrow_mut() = Some(path.to_path_buf()); + Ok(()) + })); + }); + + let created = RunDir::create(&parent, false); + AFTER_CREATE.with(|act| act.borrow_mut().take()); + + assert!(created.is_err()); + let path = replaced.borrow_mut().take().ok_or("never replaced")?; + assert!(!path.join(MARKER_FILE).exists()); + assert!(!path.join(LOCK_FILE).exists()); + Ok(()) + } + + /// Hermes round 6 on #263, as another user: `nobody` can move what the + /// user makes in a directory anyone can write to, and so could swap a + /// directory above a run directory for a link; a run refuses to start + /// below one. Below a sticky one, which a run does start in, `nobody` + /// cannot, and writes through `RunDir::path` land in the run directory. + #[cfg(target_os = "linux")] + #[test] + #[ignore = "acts as another user through sudo; CI runs it"] + fn with_sudo_another_user_cannot_redirect_what_a_run_writes() -> TestResult { + use std::ffi::OsStr; + use std::os::unix::fs::PermissionsExt; + let temp = tempfile::tempdir()?; + // Within `nobody`'s reach. + std::fs::set_permissions(temp.path(), std::fs::Permissions::from_mode(0o755))?; + + // root's, open to anyone, not sticky. + let open = temp.path().join("open"); + sudo(&["mkdir", "-m", "0777"].map(OsStr::new), &open)?; + std::fs::create_dir(open.join("mine"))?; + assert!(nobody_moves(&open.join("mine"), &open.join("moved"))?); + let refused = RunDir::create(&open.join("moved").join("devnets"), false); + assert_eq!( + refused.err().map(|e| e.kind()), + Some(io::ErrorKind::PermissionDenied) + ); + + // root's, open to anyone, sticky. + let sticky = temp.path().join("sticky"); + sudo(&["mkdir", "-m", "1777"].map(OsStr::new), &sticky)?; + let mine = sticky.join("mine"); + let run = RunDir::create(&mine.join("devnets"), false)?; + assert!(!nobody_moves(&mine, &sticky.join("moved"))?); + std::fs::write(run.path().join("written"), b"the run's")?; + let name = run.path().file_name().ok_or("no name")?; + assert!(mine.join("devnets").join(name).join("written").is_file()); + run.remove()?; + Ok(()) + } + + /// Hermes round 6 on #263: nor below a directory another user owns, or + /// one a group other than the user's own can write to. + #[cfg(target_os = "linux")] + #[test] + #[ignore = "makes directories another user owns through sudo; CI runs it"] + fn with_sudo_a_parent_below_another_users_or_groups_directory_is_refused() -> TestResult { + use std::ffi::OsStr; + let temp = tempfile::tempdir()?; + let me = rustix::process::geteuid().as_raw().to_string(); + for (name, how) in [ + ("theirs", ["-o", "65534", "-m", "0755"]), + ("their-group", ["-g", "65534", "-m", "0775"]), + ] { + let above = temp.path().join(name); + let parent = above.join("devnets"); + let make: Vec<&OsStr> = ["install", "-d"] + .into_iter() + .chain(how) + .map(OsStr::new) + .collect(); + sudo(&make, &above)?; + sudo( + &["install", "-d", "-o", me.as_str()].map(OsStr::new), + &parent, + )?; + + let refused = RunDir::create(&parent, false); + + assert_eq!( + refused.err().map(|e| e.kind()), + Some(io::ErrorKind::PermissionDenied), + "{name}" + ); + assert!(std::fs::read_dir(&parent)?.next().is_none(), "{name}"); + } + Ok(()) + } + + /// Hermes round 6 on #263: a finished run's directory another user owns + /// is theirs, whatever is in it, and in a sticky parent they could still + /// move it once the sweep has found it. It is not swept, though the user + /// could delete what is in it. + #[cfg(target_os = "linux")] + #[test] + #[ignore = "hands a directory to another user through sudo; CI runs it"] + fn with_sudo_another_users_finished_run_is_not_swept() -> TestResult { + use std::ffi::OsStr; + let temp = tempfile::tempdir()?; + let old = finished(temp.path(), false)?; + sudo(&["chown", "-R", "65534"].map(OsStr::new), &old)?; + sudo(&["chmod", "-R", "a+rwX"].map(OsStr::new), &old)?; + + let _run = RunDir::create(temp.path(), false)?; + + assert!(old.join("nodes").join("peer").join("chunk").is_file()); + Ok(()) + } + + /// What that rests on: a directory on another mount is told apart from + /// one on the run's. + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[test] + fn a_directory_on_another_mount_is_told_apart() -> TestResult { + use rustix::fs::{Mode, OFlags}; + let open = |path: &Path| { + rustix::fs::open( + path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, + Mode::empty(), + ) + }; + let temp = tempfile::tempdir()?; + std::fs::create_dir(temp.path().join("sub"))?; + // procfs on Linux, devfs on macOS: always mounts of their own. + let other = if cfg!(target_os = "linux") { + "/proc" + } else { + "/dev" + }; + + let here = unix::mount_of(&open(temp.path())?)?; + assert_eq!(unix::mount_of(&open(&temp.path().join("sub"))?)?, here); + assert_ne!(unix::mount_of(&open(Path::new(other))?)?, here); + Ok(()) + } + + /// Hermes round 2 on #263: a parent reached through a dangling symlink + /// fails to start and deletes nothing. + #[cfg(unix)] + #[test] + fn a_parent_behind_a_dangling_symlink_fails_closed() -> TestResult { + let temp = tempfile::tempdir()?; + let victim = temp.path().join("victim"); + std::fs::create_dir_all(&victim)?; + std::fs::write(victim.join("sentinel"), b"keep")?; + std::os::unix::fs::symlink(victim.join("not-yet"), temp.path().join("alias"))?; + + assert!(RunDir::create(&temp.path().join("alias").join("devnets"), false).is_err()); + + assert!(victim.join("sentinel").is_file()); + assert!(!victim.join("not-yet").exists()); + Ok(()) + } + + /// `parent` inside `above`, inside a fresh temporary directory, with + /// `mode` set on whichever of the two `which` names. + #[cfg(unix)] + fn parent_with( + which: &str, + mode: u32, + ) -> std::result::Result<(tempfile::TempDir, PathBuf), Box> { + use std::os::unix::fs::PermissionsExt; + let temp = tempfile::tempdir()?; + let above = temp.path().join("above"); + let parent = above.join("devnets"); + std::fs::create_dir_all(&parent)?; + let dir = if which == "above" { &above } else { &parent }; + std::fs::set_permissions(dir, std::fs::Permissions::from_mode(mode))?; + Ok((temp, parent)) + } + + /// Hermes round 6 on #263: another user who can write to the parent, or + /// to a directory above it, can move what is in it. A run refuses to + /// start there, and makes nothing there. + #[cfg(unix)] + #[test] + fn a_parent_another_user_could_change_is_refused() -> TestResult { + let mut cases = vec![("above", 0o777), ("parent", 0o777)]; + // No user private groups on macOS: a group there is shared. + if cfg!(not(target_os = "linux")) { + cases.push(("above", 0o775)); + } + for (which, mode) in cases { + let (_temp, parent) = parent_with(which, mode)?; + + let refused = RunDir::create(&parent, false); + + assert_eq!( + refused.err().map(|e| e.kind()), + Some(io::ErrorKind::PermissionDenied), + "{which} {mode:o}" + ); + assert!( + std::fs::read_dir(&parent)?.next().is_none(), + "{which} {mode:o}" + ); + } + Ok(()) + } + + /// What a run does start in: a parent in, or below, a sticky directory + /// anyone can write to, as `/tmp` is, which lets no one rename what they + /// do not own; and on Linux, below one the user's own group can write to. + #[cfg(unix)] + #[test] + fn a_parent_only_this_user_can_change_is_accepted() -> TestResult { + let mut cases = vec![("above", 0o1777), ("parent", 0o1777)]; + if cfg!(target_os = "linux") { + cases.push(("above", 0o775)); + } + for (which, mode) in cases { + let (_temp, parent) = parent_with(which, mode)?; + + let run = + RunDir::create(&parent, false).map_err(|e| format!("{which} {mode:o}: {e}"))?; + + assert!(run.path().join(MARKER_FILE).is_file(), "{which} {mode:o}"); + } + Ok(()) + } + + /// With an ACL, a directory's group bits are only its mask, the most it + /// grants anyone it names, so one the user's own group can write to does + /// not pass if it has one. + #[cfg(target_os = "linux")] + #[test] + fn a_parent_below_a_directory_with_an_acl_is_refused() -> TestResult { + let (_temp, parent) = parent_with("above", 0o755)?; + let above = parent.parent().ok_or("no parent")?; + // user::rwx, user:65534:rwx, group::r-x, mask::rwx, other::r-x, in the + // kernel's `system.posix_acl_access` format. + let mut acl = 2_u32.to_le_bytes().to_vec(); + for (tag, perm, id) in [ + (0x01_u16, 7_u16, u32::MAX), + (0x02, 7, 65534), + (0x04, 5, u32::MAX), + (0x10, 7, u32::MAX), + (0x20, 5, u32::MAX), + ] { + acl.extend_from_slice(&tag.to_le_bytes()); + acl.extend_from_slice(&perm.to_le_bytes()); + acl.extend_from_slice(&id.to_le_bytes()); + } + rustix::fs::lsetxattr( + above, + "system.posix_acl_access", + &acl, + rustix::fs::XattrFlags::empty(), + )?; + let mode = rustix::fs::Mode::from_raw_mode(rustix::fs::lstat(above)?.st_mode); + assert!(mode.contains(rustix::fs::Mode::WGRP), "{mode:?}"); + + let refused = RunDir::create(&parent, false); + + assert_eq!( + refused.err().map(|e| e.kind()), + Some(io::ErrorKind::PermissionDenied) + ); + Ok(()) + } + + /// A link wearing a finished run's name is removed, never followed. + #[cfg(unix)] + #[test] + fn a_link_named_like_a_run_is_never_followed() -> TestResult { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("devnets"); + let outside = finished(&temp.path().join("outside"), false)?; + std::fs::write(outside.join("sentinel"), b"keep")?; + std::fs::create_dir_all(&parent)?; + let link = parent.join(outside.file_name().ok_or("no name")?); + std::os::unix::fs::symlink(&outside, &link)?; + + let _run = RunDir::create(&parent, false)?; + + assert!(outside.join("sentinel").is_file()); + assert!(std::fs::symlink_metadata(&link)?.file_type().is_symlink()); + Ok(()) + } + + /// Hermes round 2 on #263: retargeting a symlink above the run directory + /// while the run lasts does not redirect its removal, even onto a copy of + /// the run directory carrying the run's own marker. + #[cfg(unix)] + #[test] + fn retargeting_a_symlink_above_the_run_does_not_redirect_removal() -> TestResult { + let temp = tempfile::tempdir()?; + let first = temp.path().join("first"); + let second = temp.path().join("second"); + std::fs::create_dir_all(&first)?; + std::fs::create_dir_all(&second)?; + let link = temp.path().join("link"); + std::os::unix::fs::symlink(&first, &link)?; + + let run = RunDir::create(&link.join("devnets"), false)?; + let name = run.path().file_name().ok_or("no name")?.to_owned(); + let decoy = second.join("devnets").join(&name); + std::fs::create_dir_all(&decoy)?; + std::fs::copy(run.path().join(MARKER_FILE), decoy.join(MARKER_FILE))?; + std::fs::write(decoy.join("sentinel"), b"keep")?; + std::fs::remove_file(&link)?; + std::os::unix::fs::symlink(&second, &link)?; + + // Hermes round 5: nor are writes through the path it hands out. + std::fs::write(run.path().join("written"), b"the run's")?; + assert!(first.join("devnets").join(&name).join("written").is_file()); + assert!(!decoy.join("written").exists(), "path() followed the link"); + run.remove()?; + + assert!(!first.join("devnets").join(&name).exists()); + assert!( + decoy.join("sentinel").is_file(), + "the removal followed the retargeted link" + ); + Ok(()) + } + + /// `mklink /J`: a directory junction, which needs no privilege to make. + /// The paths reach `cmd` through the environment and are expanded inside + /// quotes, once, where none of its metacharacters act, and a Windows path + /// cannot hold a quote: whatever they contain, they stay two paths. + #[cfg(windows)] + fn junction(link: &Path, target: &Path) -> TestResult { + use std::os::windows::process::CommandExt; + let status = std::process::Command::new("cmd") + .raw_arg(r#"/D /V:OFF /S /C "mklink /J "%ANT_TEST_LINK%" "%ANT_TEST_TARGET%"""#) + .env("ANT_TEST_LINK", link) + .env("ANT_TEST_TARGET", target) + .stdout(std::process::Stdio::null()) + .status()?; + if status.success() { + Ok(()) + } else { + Err(format!("mklink /J {} {} failed", link.display(), target.display()).into()) + } + } + + /// The same on Windows, with a junction. + #[cfg(windows)] + #[test] + fn retargeting_a_junction_above_the_run_does_not_redirect_removal() -> TestResult { + let temp = tempfile::tempdir()?; + // Names that mean something to `cmd`, which `junction` goes through. + let first = temp.path().join("first & %USERNAME% ^(1)!"); + let second = temp.path().join("second ; %PATH% = ^&"); + std::fs::create_dir_all(&first)?; + std::fs::create_dir_all(&second)?; + let link = temp.path().join("link & %TEMP%"); + junction(&link, &first)?; + + let run = RunDir::create(&link.join("devnets"), false)?; + let name = run.path().file_name().ok_or("no name")?.to_owned(); + let decoy = second.join("devnets").join(&name); + std::fs::create_dir_all(&decoy)?; + std::fs::copy(run.path().join(MARKER_FILE), decoy.join(MARKER_FILE))?; + std::fs::write(decoy.join("sentinel"), b"keep")?; + std::fs::remove_dir(&link)?; + junction(&link, &second)?; + + std::fs::write(run.path().join("written"), b"the run's")?; + assert!(first.join("devnets").join(&name).join("written").is_file()); + assert!( + !decoy.join("written").exists(), + "path() followed the junction" + ); + run.remove()?; + + assert!(!first.join("devnets").join(&name).exists()); + assert!( + decoy.join("sentinel").is_file(), + "the removal followed the retargeted junction" + ); + Ok(()) + } +} diff --git a/tests/devnet_lifecycle.rs b/tests/devnet_lifecycle.rs new file mode 100644 index 00000000..64503ccc --- /dev/null +++ b/tests/devnet_lifecycle.rs @@ -0,0 +1,325 @@ +//! ant-devnet under stop signals (ant-node#261). +//! +//! SIGTERM, from `kill`, a service manager or `ant dev stop`, used to kill +//! ant-devnet outright: the devnet was never shut down, Anvil was orphaned +//! still holding its port, and the data directory was left behind. These +//! tests run the real binary and check that a SIGTERM, while running or +//! part-way through startup, now shuts it down cleanly, removing only the +//! run's own directory, and that the directory a run killed by `SIGKILL` +//! has to leave behind is removed by the next run. +//! +//! The binary runs with `HOME` and `XDG_DATA_HOME` pointed at a temporary +//! directory, so its default data root, and anything it removes, is inside +//! the test's own directory. `anvil` must be on `PATH` (CI installs Foundry). + +#![cfg(unix)] +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use ant_node::devnet::MARKER_FILE; +use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; +use std::os::unix::process::CommandExt; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, ExitStatus, Stdio}; +use std::thread::sleep; +use std::time::{Duration, Instant}; + +/// Long enough for a 5-node devnet with Anvil on a loaded CI runner. +const STARTUP_TIMEOUT: Duration = Duration::from_mins(4); +/// How long a stopped devnet may take to shut down and exit. +const EXIT_TIMEOUT: Duration = Duration::from_mins(1); + +/// A running ant-devnet, in a process group that Anvil, its child, joins. +/// Whatever is left of the group is killed when a test ends, however it ends. +struct Devnet { + process: Child, + /// The group's leader, which does nothing. It is reaped only once the + /// group has been killed, so until then the group's id cannot be reused, + /// and the kill cannot reach some other group: ant-devnet itself is + /// reaped as soon as a test sees it exit. + leader: Child, +} + +impl Drop for Devnet { + /// SIGTERM first, if it is still running: ant-devnet then stops Anvil + /// itself. Then SIGKILL to the whole group, for an Anvil left running by + /// a devnet that was killed, or that exited without stopping it. + fn drop(&mut self) { + if matches!(self.process.try_wait(), Ok(None)) { + let _ = Command::new("kill") + .args(["-TERM", &self.process.id().to_string()]) + .status(); + let deadline = Instant::now() + EXIT_TIMEOUT; + while Instant::now() < deadline && matches!(self.process.try_wait(), Ok(None)) { + sleep(Duration::from_millis(100)); + } + } + let _ = Command::new("kill") + .args(["-s", "KILL", "--", &format!("-{}", self.leader.id())]) + .stderr(Stdio::null()) + .status(); + let _ = self.leader.kill(); + let _ = self.leader.wait(); + let _ = self.process.kill(); + let _ = self.process.wait(); + } +} + +impl Devnet { + fn spawn(home: &Path, args: &[&str]) -> Self { + let leader = Command::new("sleep") + .arg("86400") + .process_group(0) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn the process group's leader"); + let group = i32::try_from(leader.id()).expect("a pid fits in an i32"); + let process = Command::new(env!("CARGO_BIN_EXE_ant-devnet")) + .args(args) + .process_group(group) + .env("HOME", home) + .env("XDG_DATA_HOME", home.join("data")) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn ant-devnet"); + Self { process, leader } + } + + /// Send `signal` (a name `kill` accepts, e.g. "TERM") to the devnet, + /// through the system `kill` so the test needs no unsafe code. + fn signal(&self, signal: &str) { + let status = Command::new("kill") + .arg(format!("-{signal}")) + .arg(self.process.id().to_string()) + .status() + .expect("run kill"); + assert!( + status.success(), + "kill -{signal} {} failed", + self.process.id() + ); + } + + fn wait_for_exit(&mut self) -> ExitStatus { + let deadline = Instant::now() + EXIT_TIMEOUT; + loop { + if let Some(status) = self.process.try_wait().expect("try_wait") { + return status; + } + assert!( + Instant::now() < deadline, + "ant-devnet still running {}s after the stop signal", + EXIT_TIMEOUT.as_secs() + ); + sleep(Duration::from_millis(100)); + } + } + + /// Fail if the devnet exited on its own, e.g. a startup error. + fn assert_running(&mut self) { + if let Some(status) = self.process.try_wait().expect("try_wait") { + panic!("ant-devnet exited early with {status}"); + } + } +} + +/// Poll until `ready` returns `Some`, or fail after `STARTUP_TIMEOUT`. +fn wait_until(devnet: &mut Devnet, what: &str, mut ready: impl FnMut() -> Option) -> T { + let deadline = Instant::now() + STARTUP_TIMEOUT; + loop { + if let Some(value) = ready() { + return value; + } + devnet.assert_running(); + assert!( + Instant::now() < deadline, + "timed out after {}s waiting for {what}", + STARTUP_TIMEOUT.as_secs() + ); + sleep(Duration::from_millis(200)); + } +} + +/// The run directories in `parent`: those with a run marker. +fn run_dirs(parent: &Path) -> Vec { + std::fs::read_dir(parent) + .map(|entries| { + entries + .filter_map(|entry| Some(entry.ok()?.path())) + .filter(|dir| dir.join(MARKER_FILE).is_file()) + .collect() + }) + .unwrap_or_default() +} + +fn read_manifest(path: &Path) -> Option { + let text = std::fs::read_to_string(path).ok()?; + serde_json::from_str(&text).ok() +} + +/// `http://host:port/` → every address the host resolves to. `localhost` +/// can resolve to `::1` first while Anvil listens on `127.0.0.1`. +fn rpc_addrs(rpc_url: &str) -> Vec { + let host_port = rpc_url + .trim_start_matches("http://") + .trim_start_matches("https://") + .trim_end_matches('/'); + let addrs: Vec = host_port + .to_socket_addrs() + .expect("rpc_url resolves") + .collect(); + assert!(!addrs.is_empty(), "{rpc_url} resolves to no address"); + addrs +} + +fn accepting(addrs: &[SocketAddr]) -> bool { + addrs + .iter() + .any(|addr| TcpStream::connect_timeout(addr, Duration::from_millis(500)).is_ok()) +} + +#[test] +fn sigterm_while_running_removes_the_data_dir_and_stops_anvil() { + let home = tempfile::tempdir().unwrap(); + let manifest_path = home.path().join("manifest.json"); + let mut devnet = Devnet::spawn( + home.path(), + &[ + "--preset", + "minimal", + "--enable-evm", + "--manifest", + manifest_path.to_str().unwrap(), + ], + ); + + let manifest = wait_until(&mut devnet, "the manifest", || { + read_manifest(&manifest_path) + }); + let data_dir = PathBuf::from(manifest["data_dir"].as_str().expect("data_dir")); + let anvil = rpc_addrs(manifest["evm"]["rpc_url"].as_str().expect("evm.rpc_url")); + + // The default data directory is a fresh one for this run, inside the + // (redirected) ant data root's devnets directory. + let home_dir = std::fs::canonicalize(home.path()).unwrap(); + assert!(data_dir.starts_with(&home_dir), "{}", data_dir.display()); + assert_eq!( + data_dir.parent().and_then(Path::file_name), + Some(std::ffi::OsStr::new("devnets")), + "{}", + data_dir.display() + ); + assert!( + data_dir.join(MARKER_FILE).is_file(), + "{}", + data_dir.display() + ); + assert!(accepting(&anvil), "Anvil is not listening on {anvil:?}"); + + devnet.signal("TERM"); + let status = devnet.wait_for_exit(); + + assert!(status.success(), "ant-devnet exited with {status}"); + assert!(!data_dir.exists(), "{} was left behind", data_dir.display()); + assert!(data_dir.parent().is_some_and(Path::is_dir)); + // Anvil is stopped, so nothing listens on its port any more. + let deadline = Instant::now() + Duration::from_secs(10); + while accepting(&anvil) { + assert!( + Instant::now() < deadline, + "Anvil still listening on {anvil:?}" + ); + sleep(Duration::from_millis(200)); + } +} + +#[test] +fn sigterm_during_startup_removes_the_run_dir_only() { + let home = tempfile::tempdir().unwrap(); + // `--data-dir` names the directory the run creates its own in. + let data_dir = home.path().join("devnet"); + std::fs::create_dir_all(&data_dir).unwrap(); + std::fs::write(data_dir.join("sentinel"), b"not the devnet's").unwrap(); + let manifest_path = home.path().join("manifest.json"); + // Ten nodes a second apart: startup takes at least ten seconds, so the + // signal lands while nodes are still being spawned. + let mut devnet = Devnet::spawn( + home.path(), + &[ + "--preset", + "small", + "--spawn-delay-ms", + "1000", + "--data-dir", + data_dir.to_str().unwrap(), + "--manifest", + manifest_path.to_str().unwrap(), + ], + ); + + let run = wait_until(&mut devnet, "the first node directory", || { + run_dirs(&data_dir).into_iter().find(|run| { + std::fs::read_dir(run.join("nodes")) + .ok() + .and_then(|mut entries| entries.next()) + .is_some() + }) + }); + assert!( + !manifest_path.exists(), + "startup finished before the signal; the test needs a slower startup" + ); + + devnet.signal("TERM"); + let status = devnet.wait_for_exit(); + + assert!(status.success(), "ant-devnet exited with {status}"); + assert!(!run.exists(), "{} was left behind", run.display()); + assert!(data_dir.join("sentinel").is_file()); + assert!( + !manifest_path.exists(), + "a devnet stopped during startup must not write a manifest" + ); +} + +#[test] +fn the_next_run_removes_what_a_sigkilled_run_left() { + let home = tempfile::tempdir().unwrap(); + let data_dir = home.path().join("devnets"); + let other = data_dir.join("not-a-run"); + std::fs::create_dir_all(&other).unwrap(); + std::fs::write(other.join("sentinel"), b"not the devnet's").unwrap(); + let args = [ + "--preset", + "minimal", + "--data-dir", + data_dir.to_str().unwrap(), + ]; + + let mut killed = Devnet::spawn(home.path(), &args); + let left = wait_until(&mut killed, "the run directory", || { + run_dirs(&data_dir).pop() + }); + killed.signal("KILL"); + killed.wait_for_exit(); + // SIGKILL can't be caught, so the run could not remove its directory. + assert!(left.join(MARKER_FILE).is_file(), "{}", left.display()); + + let mut next = Devnet::spawn(home.path(), &args); + // It removes it before it creates its own. + let runs = wait_until(&mut next, "the next run to replace it", || { + let runs = run_dirs(&data_dir); + (!left.exists() && !runs.is_empty()).then_some(runs) + }); + assert_eq!(runs.len(), 1, "{runs:?}"); + assert!(other.join("sentinel").is_file()); + + next.signal("TERM"); + let status = next.wait_for_exit(); + assert!(status.success(), "ant-devnet exited with {status}"); + assert_eq!(run_dirs(&data_dir), Vec::::new()); + assert!(other.join("sentinel").is_file()); +}