From 282bbc3709f2c489cfad8c81b0926d3ea47208d9 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 31 Jul 2026 21:40:34 +0200 Subject: [PATCH] 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 subvol snapshot push --peer 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..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 | 2 + hive-c0re/src/priv_client.rs | 138 ++++++++++++++- hive-c0re/src/server.rs | 21 +++ hive-c0re/src/snapshot_push.rs | 89 ++++++++++ hive-c0re/src/swarm_peers.rs | 188 +++++++++++++++++++++ hive-host-sock/src/lib.rs | 17 ++ hivectl/src/cli.rs | 31 +++- hivectl/src/subvol.rs | 36 ++++ nix/host-modules/hive-c0re/environment.nix | 14 +- nix/host-modules/swarm.nix | 19 +++ 10 files changed, 546 insertions(+), 9 deletions(-) create mode 100644 hive-c0re/src/snapshot_push.rs create mode 100644 hive-c0re/src/swarm_peers.rs diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index c6264986..bc3fd011 100644 --- a/hive-c0re/src/main.rs +++ b/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; diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index 2ce0020a..4663db86 100644 --- a/hive-c0re/src/priv_client.rs +++ b/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 { } } +/// 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 { + const FD_SIZE: usize = std::mem::size_of::(); + + let mut iov = libc::iovec { + iov_base: bytes.as_ptr().cast::().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::(), + 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 { + 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::(&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`). diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 93dba9f3..5e19962e 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -264,6 +264,12 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> 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 { + 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 { require_matrix_present().await?; let register_token = diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs new file mode 100644 index 00000000..75a3ef0a --- /dev/null +++ b/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(()) +} diff --git a/hive-c0re/src/swarm_peers.rs b/hive-c0re/src/swarm_peers.rs new file mode 100644 index 00000000..aa2c8cdc --- /dev/null +++ b/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, + /// TCP port of the peer's snapshot store. Absent when it runs none. + #[serde(default)] + pub snapshot_store_port: Option, +} + +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 { + 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> { + 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 { + 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> { + 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) -> 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}" + ); + } +} diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index b702f10c..f74bee85 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -375,6 +375,23 @@ pub enum HostRequest { parent: Option, dest: String, }, + /// Push a snapshot to a peer hive's snapshot store over the + /// WireGuard mesh (`hivectl agent subvol snapshot push`). + /// The network sibling of [`HostRequest::SendSnapshot`]: same + /// snapshot and optional incremental `parent`, but the stream goes + /// to `peer`'s receiver instead of a local file. + /// + /// `peer` is a domain from `services.hyperhive.swarm.peers`; the + /// daemon resolves its mesh address and store port from there and + /// fails if the peer declares neither. Bare success — nothing is + /// written on this host to report a path for. + PushSnapshot { + name: Ident, + label: String, + #[serde(default)] + parent: Option, + peer: String, + }, } /// One agent's btrfs qgroup usage row — the [`HostRequest::QuotaShow`] diff --git a/hivectl/src/cli.rs b/hivectl/src/cli.rs index d3d5f8b1..34f45582 100644 --- a/hivectl/src/cli.rs +++ b/hivectl/src/cli.rs @@ -620,10 +620,10 @@ pub enum SnapshotCmd { /// Snapshot label passed to `subvol snapshot create --label`. label: String, }, - /// Export a snapshot to a local file via `btrfs send` (the local-file - /// half of inter-hive migration transport; the cross-hive `ssh ... - /// btrfs receive` leg isn't wired up yet). Also useful standalone as a - /// point-in-time backup: a full send with no `--parent` produces a + /// Export a snapshot to a local file via `btrfs send` — the + /// local-file half of the inter-hive migration transport (`push` + /// is the network half). Also useful standalone as a point-in-time + /// backup: a full send with no `--parent` produces a /// self-contained archive of the snapshot. Send { /// Snapshot label passed to `subvol snapshot create --label`. @@ -638,4 +638,27 @@ pub enum SnapshotCmd { #[arg(long)] dest: String, }, + /// Stream a snapshot to a peer hive's snapshot store over the + /// WireGuard mesh — the network half of the migration transport. + /// + /// Nothing is staged locally: `btrfs send` writes straight into the + /// connection, so a multi-gigabyte agent needs no scratch space on + /// this host. The mesh is the authentication (cryptokey routing + /// binds the peer's address to its key), so there is no credential + /// to pass here. + Push { + /// Snapshot label passed to `subvol snapshot create --label`. + label: String, + /// Optional parent snapshot label for an incremental send + /// (`btrfs send -p`) — must be an existing, older snapshot of the + /// same agent, and must already be present on the receiver. + /// Omit for a full send. + #[arg(long)] + parent: Option, + /// Peer hive domain, as declared in + /// `services.hyperhive.swarm.peers`. Its mesh address and + /// snapshot-store port are read from there. + #[arg(long)] + peer: String, + }, } diff --git a/hivectl/src/subvol.rs b/hivectl/src/subvol.rs index 811fe800..2f680176 100644 --- a/hivectl/src/subvol.rs +++ b/hivectl/src/subvol.rs @@ -44,6 +44,11 @@ pub(crate) async fn dispatch_subvol(socket: &Path, name: &str, cmd: SubvolCmd) - parent, dest, } => subvol_snapshot_send(socket, name, &label, parent.as_deref(), &dest).await, + SnapshotCmd::Push { + label, + parent, + peer, + } => subvol_snapshot_push(socket, name, &label, parent.as_deref(), &peer).await, }, } } @@ -217,3 +222,34 @@ async fn subvol_snapshot_send( ) .await } + +/// `subvol snapshot push