diff --git a/docs/networking/snapshot-store.md b/docs/networking/snapshot-store.md index 9669a765..8281427a 100644 --- a/docs/networking/snapshot-store.md +++ b/docs/networking/snapshot-store.md @@ -108,9 +108,6 @@ 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 914a410d..3098c6d3 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -19,7 +19,6 @@ //! exists to serve. use std::os::fd::OwnedFd; -use std::time::Duration; use anyhow::{Context as _, Result, bail}; use hive_priv_sock::PrivRequest; @@ -33,15 +32,6 @@ 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`. /// @@ -53,7 +43,21 @@ 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, CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT).await?; + 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")?; // 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 @@ -88,50 +92,6 @@ 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 @@ -177,12 +137,7 @@ mod tests { use tokio::net::{TcpListener, TcpStream}; - use std::time::Duration; - - use super::{ - CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT, bounded, detach_blocking, open_stream, - require_store_addr, - }; + use super::{detach_blocking, require_store_addr}; #[tokio::test] async fn the_handed_off_descriptor_is_blocking() { @@ -203,60 +158,6 @@ 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 cc631300..f8906890 100644 --- a/nix/host-modules/swarm-snapshot-store.nix +++ b/nix/host-modules/swarm-snapshot-store.nix @@ -33,11 +33,6 @@ 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 @@ -62,15 +57,8 @@ 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. `-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 + # fixed-width record. + if ! read -r keyword agent; then echo "swarm-snapshot-store: peer ''${REMOTE_ADDR:-?} closed before sending a header" >&2 exit 1 fi @@ -248,13 +236,6 @@ 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 a2dc9af3..5dc4d6c0 100644 --- a/nix/module-eval/snapshot-store.nix +++ b/nix/module-eval/snapshot-store.nix @@ -3,8 +3,7 @@ # executes"). # # The snapshot store's receiver socket: where it listens, that it can bind -# before the mesh interface is up, that it refuses a wildcard, and that a -# silent or stalled transfer is bounded. +# before the mesh interface is up, and that it refuses a wildcard. { pkgs, lib, @@ -47,10 +46,6 @@ 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 = @@ -85,18 +80,6 @@ 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