feat(#2862): push a snapshot to a peer hive's store over the mesh

Adds the caller the fd-passing machinery existed for: hivectl agent
<name> subvol snapshot push --peer <hive> resolves the peer, connects
to its snapshot store, writes the agent header, and hands the connected
socket to hive-priv, which runs btrfs send straight into it.

The split keeps the root helper ignorant. Everything that involves
knowing where a peer is, what the wire protocol looks like, and which
hive to trust happens in the unprivileged daemon; hive-priv only ever
receives an already-open descriptor. Once btrfs send starts, neither
process is in the data path, so a multi-gigabyte transfer costs no
per-byte work and survives a hive-c0re restart.

call_with_fd takes the descriptor by value and closes it as soon as the
kernel has it. A socket stays open until every copy closes, so holding
one back would leave the receiver waiting for an EOF that never comes:
btrfs receive blocks and this side reports success for a transfer the
peer never committed. Ownership makes that unrepresentable.

The peer's store port is a new swarm.peers.<domain>.snapshotStorePort
option rather than a constant matching the module default. A pushing
hive cannot read the receiver's configuration, so assuming 51821 would
push at a port nobody promised to listen on; absent, the push fails
naming the option. swarm_peers parses the mesh address the host module
has always rendered into HYPERHIVE_PEERS but nothing read.

hive-c0re/src/main.rs

@@ -31,9 +31,11 @@ mod paths;
 mod priv_client;
 mod questions;
 mod server;
+mod snapshot_push;
 mod socket_server;
 mod stats;
 mod stores;
+mod swarm_peers;
 mod webhook_secret;
 mod workers;
 

hive-c0re/src/priv_client.rs

@@ -11,7 +11,9 @@ use hive_priv_sock::{
     BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery, NetworkIsolation,
     PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
 };
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use std::os::fd::{AsRawFd as _, OwnedFd, RawFd};
+
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Interest};
 use tokio::net::UnixStream;
 
 /// Send a single request to `hive-priv` and return the response.
@@ -40,6 +42,140 @@ pub async fn call(req: &PrivRequest) -> Result<PrivResponse> {
     }
 }
 
+/// Ancillary-data buffer sized and aligned for one `SCM_RIGHTS`
+/// message. `CMSG_SPACE` is not a `const fn`, so the size is a literal
+/// with room to spare; the union member supplies the `cmsghdr`
+/// alignment `CMSG_FIRSTHDR` requires.
+#[repr(C)]
+union CmsgSpace {
+    _align: libc::cmsghdr,
+    bytes: [u8; 32],
+}
+
+/// `sendmsg` `bytes` with `fd` attached as `SCM_RIGHTS`, returning how
+/// many bytes were accepted.
+///
+/// The descriptor rides on this one call — ancillary data cannot be
+/// sent separately from payload — so the caller must not have written
+/// any of `bytes` beforehand.
+fn send_with_fd(sock: RawFd, bytes: &[u8], fd: RawFd) -> std::io::Result<usize> {
+    const FD_SIZE: usize = std::mem::size_of::<RawFd>();
+
+    let mut iov = libc::iovec {
+        iov_base: bytes.as_ptr().cast::<libc::c_void>().cast_mut(),
+        iov_len: bytes.len(),
+    };
+    let mut cmsg = CmsgSpace { bytes: [0; 32] };
+    // SAFETY: msghdr is a plain C struct with no invalid bit patterns;
+    // every field we rely on is set immediately below.
+    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
+    msg.msg_iov = &raw mut iov;
+    msg.msg_iovlen = 1;
+    msg.msg_control = std::ptr::addr_of_mut!(cmsg.bytes).cast();
+    // SAFETY: CMSG_SPACE is a pure size computation.
+    let space = unsafe { libc::CMSG_SPACE(u32::try_from(FD_SIZE).unwrap_or(4)) };
+    msg.msg_controllen = space as _;
+
+    // SAFETY: the control buffer is live, aligned, and long enough for
+    // the header CMSG_SPACE just sized.
+    let hdr = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) };
+    if hdr.is_null() {
+        return Err(std::io::Error::other(
+            "control buffer too small for SCM_RIGHTS",
+        ));
+    }
+    // SAFETY: CMSG_LEN is a pure size computation; `hdr` points into
+    // our own buffer, and write_unaligned tolerates its alignment.
+    unsafe {
+        let len = libc::CMSG_LEN(u32::try_from(FD_SIZE).unwrap_or(4));
+        std::ptr::write_unaligned(
+            hdr,
+            libc::cmsghdr {
+                cmsg_len: len as _,
+                cmsg_level: libc::SOL_SOCKET,
+                cmsg_type: libc::SCM_RIGHTS,
+            },
+        );
+        // Copied in byte-wise: the control buffer is only cmsghdr-
+        // aligned, so casting CMSG_DATA to a *mut RawFd would be
+        // unsound even where it happens to work.
+        std::ptr::copy_nonoverlapping(
+            std::ptr::from_ref(&fd).cast::<u8>(),
+            libc::CMSG_DATA(hdr),
+            FD_SIZE,
+        );
+    }
+
+    // SAFETY: `msg` points at a live iovec over `bytes` and the control
+    // buffer we just filled in.
+    let n = unsafe { libc::sendmsg(sock, &raw const msg, 0) };
+    if n < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+    Ok(usize::try_from(n).unwrap_or_default())
+}
+
+/// Send a request to `hive-priv` with an open file descriptor attached,
+/// and return the response.
+///
+/// The helper receives the descriptor itself — not a path or an address
+/// — so it can act on something it was handed without being told what
+/// that thing is or how to reach it. Used for
+/// [`PrivRequest::SendAgentSnapshotToFd`], where the descriptor is a
+/// socket already connected to a peer hive's snapshot store.
+///
+/// ⚠️ Takes the descriptor by value and closes it as soon as the kernel
+/// has it, *before* awaiting the response. That is not tidiness: a
+/// socket stays open until every copy of it is closed, so a caller
+/// holding one back would leave the receiving end waiting for an EOF
+/// that never comes — `btrfs receive` blocks, and this side reports
+/// success for a transfer the peer has not committed. Passing ownership
+/// makes that mistake unrepresentable.
+///
+/// # Errors
+///
+/// Fails if the socket is unreachable, the descriptor cannot be
+/// attached, or hive-priv answers with something other than a terminal
+/// event.
+pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result<PrivResponse> {
+    let mut stream = UnixStream::connect(PRIV_SOCK)
+        .await
+        .context("connect to hive-priv socket (fd-passing)")?;
+    let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
+    let bytes = line.as_bytes();
+
+    let sock = stream.as_raw_fd();
+    let raw_fd = fd.as_raw_fd();
+    let sent = stream
+        .async_io(Interest::WRITABLE, || send_with_fd(sock, bytes, raw_fd))
+        .await
+        .context("send request + descriptor to hive-priv")?;
+    // The kernel has duplicated the descriptor into hive-priv's queue,
+    // so our copy has done its job. Close it now, before waiting on the
+    // response: see the EOF note on this function.
+    drop(fd);
+
+    // A short sendmsg is legal; the descriptor went with the first
+    // call, so the tail is an ordinary write.
+    if sent < bytes.len() {
+        stream
+            .write_all(&bytes[sent..])
+            .await
+            .context("send remainder of request to hive-priv")?;
+    }
+    stream.shutdown().await.context("shutdown write half")?;
+
+    let mut resp_line = String::new();
+    BufReader::new(stream)
+        .read_line(&mut resp_line)
+        .await
+        .context("read response from hive-priv")?;
+    match serde_json::from_str::<PrivEvent>(&resp_line).context("parse PrivResponse")? {
+        PrivEvent::Done(resp) => Ok(resp),
+        PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"),
+    }
+}
+
 /// Send a streaming request to `hive-priv`, calling `on_line` for each
 /// `PrivEvent::Line` as it arrives, then returning the terminal
 /// `PrivResponse`. Used for long-running ops (`create` / `update`).

hive-c0re/src/server.rs

@@ -264,6 +264,12 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
                 parent,
                 dest,
             } => handle_send_snapshot(name.as_str(), label, parent.as_deref(), dest).await?,
+            HostRequest::PushSnapshot {
+                name,
+                label,
+                parent,
+                peer,
+            } => handle_push_snapshot(name.as_str(), label, parent.as_deref(), peer).await?,
         })
     }
     .await;
@@ -661,6 +667,21 @@ async fn handle_send_snapshot(
     Ok(HostResponse::messages(vec![path]))
 }
 
+/// Push a snapshot to a peer hive's store. The network sibling of
+/// [`handle_send_snapshot`]: nothing lands on this host, so success is
+/// bare rather than a path.
+async fn handle_push_snapshot(
+    name: &str,
+    label: &str,
+    parent: Option<&str>,
+    peer: &str,
+) -> Result<HostResponse> {
+    crate::snapshot_push::push_agent_snapshot(name, label, parent, peer)
+        .await
+        .with_context(|| format!("push {name} snapshot (label {label:?}) to peer {peer:?}"))?;
+    Ok(HostResponse::success())
+}
+
 async fn handle_matrix_sync_admin() -> Result<HostResponse> {
     require_matrix_present().await?;
     let register_token =

hive-c0re/src/snapshot_push.rs

@@ -0,0 +1,89 @@
+//! Push an agent snapshot to a peer hive's snapshot store.
+//!
+//! This is the c0re half of the transport: it resolves the peer, opens
+//! the connection, writes the protocol header, and hands the connected
+//! socket to `hive-priv`, which runs `btrfs send` straight into it.
+//!
+//! The split is the point. Everything that requires knowing *where* the
+//! peer is and *what* the wire protocol looks like happens here, in the
+//! unprivileged daemon. hive-priv only ever receives an already-open
+//! descriptor, so the root helper never learns an address, never parses
+//! a peer list, and never decides who to trust — and once `btrfs send`
+//! starts, neither process is in the data path.
+
+use anyhow::{Context as _, Result, bail};
+use hive_priv_sock::PrivRequest;
+use tokio::io::AsyncWriteExt as _;
+use tokio::net::TcpStream;
+
+use crate::priv_client;
+use crate::swarm_peers;
+
+/// Send `snapshot` of `agent` to `peer_domain`'s snapshot store,
+/// optionally as an incremental against `parent`.
+///
+/// # Errors
+///
+/// Fails when the peer is unknown, declares no mesh address or store
+/// port, is unreachable, or when hive-priv reports the `btrfs send`
+/// failed.
+pub async fn push_agent_snapshot(
+    agent: &str,
+    snapshot: &str,
+    parent: Option<&str>,
+    peer_domain: &str,
+) -> Result<()> {
+    let peer = swarm_peers::peer(peer_domain)?;
+    let addr = peer.snapshot_store_addr()?;
+
+    let mut sock = TcpStream::connect(&addr)
+        .await
+        .with_context(|| format!("connect to {peer_domain} 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
+        .with_context(|| format!("send header to {peer_domain} snapshot store"))?;
+    sock.flush()
+        .await
+        .with_context(|| format!("flush header to {peer_domain} 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
+    // EOF when `btrfs send` finishes rather than hanging on a descriptor
+    // this process is still holding.
+    let fd = sock
+        .into_std()
+        .context("detach snapshot-store socket for hand-off")?
+        .into();
+
+    let resp = priv_client::call_with_fd(
+        &PrivRequest::SendAgentSnapshotToFd {
+            agent_name: agent.to_owned(),
+            snapshot_name: snapshot.to_owned(),
+            parent_snapshot_name: parent.map(str::to_owned),
+        },
+        fd,
+    )
+    .await
+    .with_context(|| format!("stream {agent}/{snapshot} to {peer_domain}"))?;
+
+    if !resp.ok {
+        bail!(
+            "push of {agent}/{snapshot} to {peer_domain} failed: {}",
+            resp.error.unwrap_or_else(|| resp.stderr.clone())
+        );
+    }
+    tracing::info!(
+        agent,
+        snapshot,
+        parent,
+        peer = peer_domain,
+        "pushed agent snapshot to peer snapshot store"
+    );
+    Ok(())
+}

hive-c0re/src/swarm_peers.rs

@@ -0,0 +1,188 @@
+//! Peer-hive lookup for intra-swarm connections.
+//!
+//! Reads the `HYPERHIVE_PEERS` env var the host module renders from
+//! `services.hyperhive.swarm.peers` and resolves a peer domain to the
+//! address of a service running on that peer.
+//!
+//! Deliberately separate from the dashboard's `PeerHiveView`, which
+//! parses the same variable: that type is shaped for rendering the
+//! P33RS tab and drops the fields a connection needs. Nothing here is
+//! defaulted — a peer that hasn't declared an address or a port is an
+//! error naming what's missing, never a guess at a well-known value.
+//! Ports and addresses are deployment facts, so they come from the nix
+//! side or not at all.
+
+use anyhow::{Context as _, Result, bail};
+use serde::Deserialize;
+
+/// Env var carrying the serialised peer list. Rendered by
+/// `nix/host-modules/hive-c0re/environment.nix`.
+const PEERS_ENV: &str = "HYPERHIVE_PEERS";
+
+/// One peer hive, as serialised into [`PEERS_ENV`].
+///
+/// Only the fields a connection needs are modelled; `cert_fingerprint`
+/// and any later additions are ignored by serde rather than duplicated
+/// from the dashboard's view type.
+#[derive(Debug, Clone, Deserialize)]
+pub struct SwarmPeer {
+    /// The peer's DNS domain — the attrset key on the nix side, and the
+    /// name an operator refers to the peer by.
+    pub domain: String,
+    /// The peer's address on the WireGuard mesh, with prefix length
+    /// (`10.100.0.2/32`). Absent when the peer isn't in the mesh.
+    #[serde(default)]
+    pub wireguard_address: Option<String>,
+    /// TCP port of the peer's snapshot store. Absent when it runs none.
+    #[serde(default)]
+    pub snapshot_store_port: Option<u16>,
+}
+
+impl SwarmPeer {
+    /// `host:port` for this peer's snapshot store.
+    ///
+    /// # Errors
+    ///
+    /// Fails when the peer is not in the mesh, or runs no snapshot
+    /// store. Both are configuration facts the pushing hive cannot
+    /// discover on its own, so they're reported rather than guessed.
+    pub fn snapshot_store_addr(&self) -> Result<String> {
+        let Some(addr) = self.wireguard_address.as_deref() else {
+            bail!(
+                "peer {} has no wireguardAddress — it is not in the mesh, \
+                 and the mesh is the only route to a snapshot store",
+                self.domain
+            );
+        };
+        let Some(port) = self.snapshot_store_port else {
+            bail!(
+                "peer {} declares no snapshotStorePort — it hosts no snapshot store \
+                 (set services.hyperhive.swarm.peers.\"{}\".snapshotStorePort on this \
+                 host to match the receiver's services.hyperhive.snapshotStore.port)",
+                self.domain,
+                self.domain
+            );
+        };
+        // `wireguardAddress` is CIDR because WireGuard's `allowedIPs`
+        // wants it that way; a connect() wants the bare address.
+        let host = addr.split('/').next().unwrap_or(addr);
+        Ok(format!("{host}:{port}"))
+    }
+}
+
+/// Parse the peer list out of `HYPERHIVE_PEERS`.
+///
+/// # Errors
+///
+/// Fails when the variable is unset (this host declares no peers, so it
+/// is not in a swarm) or does not parse.
+pub fn peers() -> Result<Vec<SwarmPeer>> {
+    let raw = std::env::var(PEERS_ENV).with_context(|| {
+        format!("{PEERS_ENV} is unset — this host declares no swarm.peers, so it has no peers")
+    })?;
+    parse_peers(&raw)
+}
+
+/// Find one peer by domain.
+///
+/// # Errors
+///
+/// Fails when no peer matches, naming the peers that do exist — an
+/// operator typo is the likeliest cause and the list is short.
+pub fn peer(domain: &str) -> Result<SwarmPeer> {
+    let all = peers()?;
+    if let Some(found) = all.iter().find(|p| p.domain == domain) {
+        return Ok(found.clone());
+    }
+    let known: Vec<&str> = all.iter().map(|p| p.domain.as_str()).collect();
+    bail!("no swarm peer named {domain}; declared peers: {}", {
+        if known.is_empty() {
+            "(none)".to_owned()
+        } else {
+            known.join(", ")
+        }
+    })
+}
+
+/// Split out of [`peers`] so the parsing is testable without touching
+/// process-wide environment, which would race other tests.
+fn parse_peers(raw: &str) -> Result<Vec<SwarmPeer>> {
+    serde_json::from_str(raw).with_context(|| format!("{PEERS_ENV} is not valid peer JSON"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{SwarmPeer, parse_peers};
+
+    fn peer_with(addr: Option<&str>, port: Option<u16>) -> SwarmPeer {
+        SwarmPeer {
+            domain: "lab.example.com".to_owned(),
+            wireguard_address: addr.map(str::to_owned),
+            snapshot_store_port: port,
+        }
+    }
+
+    #[test]
+    fn parses_the_rendered_shape_and_ignores_unknown_fields() {
+        // cert_fingerprint is in the real payload and irrelevant here;
+        // it must not break parsing.
+        let raw = r#"[{"domain":"lab.example.com","cert_fingerprint":null,
+                       "wireguard_address":"10.100.0.2/32","snapshot_store_port":51821}]"#;
+        let peers = parse_peers(raw).expect("parse");
+        assert_eq!(peers.len(), 1);
+        assert_eq!(peers[0].snapshot_store_port, Some(51821));
+    }
+
+    #[test]
+    fn absent_optional_fields_are_none_not_an_error() {
+        // A peer that is only a federation/dashboard link declares
+        // neither field; that must parse, and fail later with a
+        // specific message, rather than fail to parse at all.
+        let raw = r#"[{"domain":"edge.corp","cert_fingerprint":null}]"#;
+        let peers = parse_peers(raw).expect("parse");
+        assert!(peers[0].wireguard_address.is_none());
+        assert!(peers[0].snapshot_store_port.is_none());
+    }
+
+    #[test]
+    fn store_addr_strips_the_cidr_prefix() {
+        // allowedIPs wants `10.100.0.2/32`; connect() does not.
+        let addr = peer_with(Some("10.100.0.2/32"), Some(51821))
+            .snapshot_store_addr()
+            .expect("addr");
+        assert_eq!(addr, "10.100.0.2:51821");
+    }
+
+    #[test]
+    fn a_bare_address_without_a_prefix_still_works() {
+        let addr = peer_with(Some("10.100.0.2"), Some(51821))
+            .snapshot_store_addr()
+            .expect("addr");
+        assert_eq!(addr, "10.100.0.2:51821");
+    }
+
+    #[test]
+    fn no_mesh_address_names_the_mesh_as_the_problem() {
+        let err = peer_with(None, Some(51821))
+            .snapshot_store_addr()
+            .expect_err("a peer off the mesh has no route");
+        let msg = format!("{err:#}");
+        assert!(msg.contains("wireguardAddress"), "{msg}");
+    }
+
+    #[test]
+    fn no_port_is_an_error_rather_than_the_module_default() {
+        // The receiving module defaults to 51821, but this host cannot
+        // read the receiver's config: assuming the default here would
+        // silently push at a port nobody promised to listen on.
+        let err = peer_with(Some("10.100.0.2/32"), None)
+            .snapshot_store_addr()
+            .expect_err("no port must fail");
+        let msg = format!("{err:#}");
+        assert!(msg.contains("snapshotStorePort"), "{msg}");
+        assert!(
+            !msg.contains("51821"),
+            "must not suggest a guessed port: {msg}"
+        );
+    }
+}