snapshot store: test the sender deadlines and the receiver bounds

open_stream takes its deadlines as parameters so the tests can use short
ones. The header-write test connects to a peer that holds the connection
without reading and sends a header larger than any loopback buffer; the
connect deadline is tested through the shared bounded helper with a
future that never completes, since a connect that hangs is not
reproducible without the network. The module-eval case asserts the
unit's RuntimeMaxSec and that the receiver script reads its header with
a timeout.

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<TcpStream> {
-    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<TcpStream> {
+    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<T>(limit: Duration, what: &str, fut: impl Future<Output = T>) -> Result<T> {
+    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!(

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