diff --git a/docs/networking/snapshot-store.md b/docs/networking/snapshot-store.md index 8281427a..9669a765 100644 --- a/docs/networking/snapshot-store.md +++ b/docs/networking/snapshot-store.md @@ -108,6 +108,9 @@ the case the store exists to serve. A `btrfs send` stream carries no notion of _which agent_ it belongs to, and the subvolume name inside it's chosen by the sender. The protocol is one `agent ` header line, then the raw stream. +The receiver waits 30 s for the header, and systemd stops a transfer +that runs past 6 h in total. The sender gives connect and header write 15 s +each; it puts no limit on the stream itself. The rule that matters: diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs index 3098c6d3..914a410d 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -19,6 +19,7 @@ //! exists to serve. use std::os::fd::OwnedFd; +use std::time::Duration; use anyhow::{Context as _, Result, bail}; use hive_priv_sock::PrivRequest; @@ -32,6 +33,15 @@ use crate::priv_client; /// `services.hyperhive.swarm.snapshotStore`. const STORE_ENV: &str = "HYPERHIVE_SNAPSHOT_STORE"; +/// Deadline for the TCP connect to the store. An unreachable mesh address +/// otherwise waits out the kernel's own connect timeout while holding the +/// admin-socket handler. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Deadline for writing and flushing the one header line into a connected +/// socket. Shorter than the receiver's own header deadline. +const HEADER_WRITE_TIMEOUT: Duration = Duration::from_secs(15); + /// Send `snapshot` of `agent` to the swarm's snapshot store, optionally /// as an incremental against `parent`. /// @@ -43,21 +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 mut sock = TcpStream::connect(addr) - .await - .with_context(|| format!("connect to the swarm snapshot store at {addr}"))?; - - // The header names which agent this stream belongs to: a btrfs - // stream carries the sender's own subvolume name, not the hive's - // notion of the agent, so the receiver cannot infer it. One line, - // because the receiver reads it with `read` and must not buffer - // past the newline into the byte stream. - sock.write_all(format!("agent {agent}\n").as_bytes()) - .await - .context("send header to the swarm snapshot store")?; - sock.flush() - .await - .context("flush header to the swarm snapshot store")?; + 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 @@ -92,6 +88,50 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st Ok(()) } +/// Connect to the store and write the protocol header. +/// +/// The header names which agent this stream belongs to: a btrfs stream +/// carries the sender's own subvolume name, not the hive's notion of the +/// agent, so the receiver cannot infer it. One line, because the receiver +/// reads it with `read` and must not buffer past the newline into the byte +/// 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, + 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}"))?; + + 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 @@ -137,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() { @@ -158,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/host-modules/swarm-snapshot-store.nix b/nix/host-modules/swarm-snapshot-store.nix index f8906890..cc631300 100644 --- a/nix/host-modules/swarm-snapshot-store.nix +++ b/nix/host-modules/swarm-snapshot-store.nix @@ -33,6 +33,11 @@ let # address must be the bare IP, so strip it. meshAddress = lib.head (lib.splitString "/" wgCfg.address); + # How long a connected peer may stay silent before sending the header + # line. The sender writes it immediately after connecting, so this only + # ever expires for a peer that is not a hive. + headerTimeoutSeconds = 30; + # The receiver. Socket-activated with Accept=yes, so stdin IS the # accepted connection and systemd hands us the peer address in # $REMOTE_ADDR --- which, on this interface, is a cryptographically @@ -57,8 +62,15 @@ let # Read exactly the header line, leaving the byte stream untouched # for btrfs receive. `read` stops at the newline and does not # buffer ahead, which is why the header is a line and not a - # fixed-width record. - if ! read -r keyword agent; then + # fixed-width record. `-t` bounds the wait: a peer that connects + # and stays silent would otherwise hold this root instance forever. + # `read` exits above 128 on timeout and 1 on EOF. + rc=0 + read -r -t ${toString headerTimeoutSeconds} keyword agent || rc=$? + if [ "$rc" -gt 128 ]; then + echo "swarm-snapshot-store: peer ''${REMOTE_ADDR:-?} sent no header within ${toString headerTimeoutSeconds}s" >&2 + exit 1 + elif [ "$rc" -ne 0 ]; then echo "swarm-snapshot-store: peer ''${REMOTE_ADDR:-?} closed before sending a header" >&2 exit 1 fi @@ -236,6 +248,13 @@ in StandardInput = "socket"; StandardError = "journal"; User = "root"; + # Upper bound on one whole transfer, header to the last byte of + # `btrfs receive`. Snapshots are whole agent state roots and run to + # multiple gigabytes (hive-priv `send_agent_snapshot_to_fd`); no + # tighter bound is documented, so this is sized for ~100 GB at + # 5 MB/s (about 5.6 h), not measured. A stalled receive is killed + # here instead of holding a root instance indefinitely. + RuntimeMaxSec = "6h"; PrivateTmp = true; ProtectHome = true; ProtectSystem = "strict"; 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