diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs index de53f7ec..914a410d 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -53,7 +53,7 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st let configured = std::env::var(STORE_ENV).ok(); let addr = require_store_addr(configured.as_deref())?; - let sock = open_stream(addr, agent).await?; + let sock = open_stream(addr, agent, CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT).await?; // Hand the connected socket over. `call_with_fd` takes ownership and // closes our copy as soon as the kernel has it, so the receiver sees @@ -97,29 +97,41 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st /// stream. Only the connect and the header are bounded; the stream body that /// follows is `btrfs send` writing straight into the socket and may take as /// long as the snapshot is large. -async fn open_stream(addr: &str, agent: &str) -> Result { - let mut sock = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(addr)) - .await - .with_context(|| { - format!( - "connect to the swarm snapshot store at {addr} timed out after {CONNECT_TIMEOUT:?}" - ) - })? - .with_context(|| format!("connect to the swarm snapshot store at {addr}"))?; +async fn open_stream( + addr: &str, + agent: &str, + connect_limit: Duration, + header_limit: Duration, +) -> Result { + let mut sock = bounded( + connect_limit, + "connect to the swarm snapshot store", + TcpStream::connect(addr), + ) + .await? + .with_context(|| format!("connect to the swarm snapshot store at {addr}"))?; - tokio::time::timeout(HEADER_WRITE_TIMEOUT, async { - sock.write_all(format!("agent {agent}\n").as_bytes()) - .await?; - sock.flush().await - }) - .await - .with_context(|| { - format!("send header to the swarm snapshot store timed out after {HEADER_WRITE_TIMEOUT:?}") - })? + bounded( + header_limit, + "send header to the swarm snapshot store", + async { + sock.write_all(format!("agent {agent}\n").as_bytes()) + .await?; + sock.flush().await + }, + ) + .await? .context("send header to the swarm snapshot store")?; Ok(sock) } +/// Run `fut` for at most `limit`; a timeout is an error naming `what`. +async fn bounded(limit: Duration, what: &str, fut: impl Future) -> Result { + tokio::time::timeout(limit, fut) + .await + .map_err(|_| anyhow::anyhow!("{what} timed out after {limit:?}")) +} + /// Turn a tokio socket into a descriptor safe to hand to `btrfs send`. /// /// tokio sockets are `O_NONBLOCK`, and `into_std` does not clear it. The @@ -165,7 +177,12 @@ mod tests { use tokio::net::{TcpListener, TcpStream}; - use super::{detach_blocking, require_store_addr}; + use std::time::Duration; + + use super::{ + CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT, bounded, detach_blocking, open_stream, + require_store_addr, + }; #[tokio::test] async fn the_handed_off_descriptor_is_blocking() { @@ -186,6 +203,60 @@ mod tests { ); } + #[tokio::test] + async fn a_future_that_never_finishes_errors_at_the_deadline() { + let err = bounded( + Duration::from_millis(50), + "connect", + std::future::pending::<()>(), + ) + .await + .expect_err("pending must time out"); + assert!( + format!("{err:#}").contains("connect timed out after"), + "{err:#}" + ); + } + + #[tokio::test] + async fn a_header_the_peer_never_drains_errors_at_the_deadline() { + // The peer accepts and holds the connection without reading, so the + // header cannot drain once the socket buffers fill; a name this + // large exceeds any loopback buffer. + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr").to_string(); + let peer = tokio::spawn(async move { + let (held, _) = listener.accept().await.expect("accept"); + std::future::pending::<()>().await; + drop(held); + }); + let huge = "a".repeat(128 * 1024 * 1024); + let err = open_stream(&addr, &huge, CONNECT_TIMEOUT, Duration::from_millis(200)) + .await + .expect_err("an undrained header must time out"); + peer.abort(); + let msg = format!("{err:#}"); + assert!( + msg.contains("send header to the swarm snapshot store timed out after"), + "{msg}" + ); + } + + #[tokio::test] + async fn a_reachable_peer_gets_the_header() { + use tokio::io::AsyncReadExt as _; + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr").to_string(); + let sock = open_stream(&addr, "alice", CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT) + .await + .expect("open"); + let (mut peer, _) = listener.accept().await.expect("accept"); + drop(sock); + let mut got = String::new(); + peer.read_to_string(&mut got).await.expect("read"); + assert_eq!(got, "agent alice\n"); + } + #[test] fn a_configured_address_passes_through() { assert_eq!( diff --git a/nix/module-eval/snapshot-store.nix b/nix/module-eval/snapshot-store.nix index 5dc4d6c0..a2dc9af3 100644 --- a/nix/module-eval/snapshot-store.nix +++ b/nix/module-eval/snapshot-store.nix @@ -3,7 +3,8 @@ # executes"). # # The snapshot store's receiver socket: where it listens, that it can bind -# before the mesh interface is up, and that it refuses a wildcard. +# before the mesh interface is up, that it refuses a wildcard, and that a +# silent or stalled transfer is bounded. { pkgs, lib, @@ -46,6 +47,10 @@ let snapshotStore.enable = true; }; + unit = store.systemd.services."swarm-snapshot-store@"; + + moduleSource = builtins.readFile ../host-modules/swarm-snapshot-store.nix; + socket = store.systemd.sockets.swarm-snapshot-store.socketConfig; refusedAsWildcard = @@ -80,6 +85,18 @@ let name = "a wildcard mesh address, IPv4 or IPv6, is refused"; ok = refusedAsWildcard storeOnWildcard && refusedAsWildcard storeOnWildcardV6; } + { + # Without it a stalled `btrfs receive` holds a root instance forever. + name = "the receiver unit has a runtime ceiling"; + ok = unit.serviceConfig.RuntimeMaxSec or null == "6h"; + } + { + # The script is a store path, so its text is not evaluable without + # building it; the module source is the nearest thing to read. A + # silent peer must not hold the root instance past the header read. + name = "the receiver script bounds its header read"; + ok = lib.hasInfix "read -r -t \${toString headerTimeoutSeconds} keyword agent" moduleSource; + } ]; in runGroup "snapshot-store" cases