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..de53f7ec 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).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,38 @@ 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) -> 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}"))?; + + 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:?}") + })? + .context("send header to the swarm snapshot store")?; + Ok(sock) +} + /// 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 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";