diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index bc3fd011..6f562175 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -35,7 +35,6 @@ mod snapshot_push; mod socket_server; mod stats; mod stores; -mod swarm_peers; mod webhook_secret; mod workers; diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 5e19962e..bec0cd75 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -268,8 +268,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { name, label, parent, - peer, - } => handle_push_snapshot(name.as_str(), label, parent.as_deref(), peer).await?, + } => handle_push_snapshot(name.as_str(), label, parent.as_deref()).await?, }) } .await; @@ -667,18 +666,17 @@ async fn handle_send_snapshot( Ok(HostResponse::messages(vec![path])) } -/// Push a snapshot to a peer hive's store. The network sibling of +/// Push a snapshot to the swarm'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) + crate::snapshot_push::push_agent_snapshot(name, label, parent) .await - .with_context(|| format!("push {name} snapshot (label {label:?}) to peer {peer:?}"))?; + .with_context(|| format!("push {name} snapshot (label {label:?}) to the swarm store"))?; Ok(HostResponse::success()) } diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs index 75a3ef0a..1cd65905 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -1,15 +1,22 @@ -//! Push an agent snapshot to a peer hive's snapshot store. +//! Push an agent snapshot to the swarm'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. +//! This is the c0re half of the transport: it resolves the store, +//! 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 +//! store 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. +//! +//! There is exactly **one** store per swarm, not one per peer hive. The +//! receiver keys destinations by *agent*, so an agent that migrates +//! between hives keeps a single unbroken incremental chain; per-hive +//! stores would split that chain in two, which is the case the store +//! exists to serve. use anyhow::{Context as _, Result, bail}; use hive_priv_sock::PrivRequest; @@ -17,28 +24,26 @@ 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`. +/// `host:port` of the swarm's snapshot store, rendered by +/// `nix/host-modules/hive-c0re/environment.nix` from +/// `services.hyperhive.swarm.snapshotStore`. +const STORE_ENV: &str = "HYPERHIVE_SNAPSHOT_STORE"; + +/// Send `snapshot` of `agent` to the swarm'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()?; +/// Fails when no store is configured, it is unreachable, or hive-priv +/// reports the `btrfs send` failed. +pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&str>) -> Result<()> { + let configured = std::env::var(STORE_ENV).ok(); + let addr = require_store_addr(configured.as_deref())?; - let mut sock = TcpStream::connect(&addr) + let mut sock = TcpStream::connect(addr) .await - .with_context(|| format!("connect to {peer_domain} snapshot store at {addr}"))?; + .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 @@ -47,10 +52,10 @@ pub async fn push_agent_snapshot( // 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"))?; + .context("send header to the swarm snapshot store")?; sock.flush() .await - .with_context(|| format!("flush header to {peer_domain} snapshot store"))?; + .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 @@ -70,11 +75,11 @@ pub async fn push_agent_snapshot( fd, ) .await - .with_context(|| format!("stream {agent}/{snapshot} to {peer_domain}"))?; + .with_context(|| format!("stream {agent}/{snapshot} to the swarm snapshot store"))?; if !resp.ok { bail!( - "push of {agent}/{snapshot} to {peer_domain} failed: {}", + "push of {agent}/{snapshot} failed: {}", resp.error.unwrap_or_else(|| resp.stderr.clone()) ); } @@ -82,8 +87,62 @@ pub async fn push_agent_snapshot( agent, snapshot, parent, - peer = peer_domain, - "pushed agent snapshot to peer snapshot store" + store = addr, + "pushed agent snapshot to the swarm snapshot store" ); Ok(()) } + +/// Require a configured store address. +/// +/// Split out of [`push_agent_snapshot`] so the "not configured" path is +/// testable without setting a process-wide env var, which would race +/// other tests. +/// +/// # Errors +/// +/// Fails when unset, naming the option to set. There is deliberately no +/// fallback: an address is a deployment fact this host cannot derive, +/// and guessing one means pushing an agent's state at whatever happens +/// to answer. +fn require_store_addr(configured: Option<&str>) -> Result<&str> { + match configured { + Some(addr) if !addr.is_empty() => Ok(addr), + _ => bail!( + "no snapshot store configured for this swarm — set \ + services.hyperhive.swarm.snapshotStore.address (and .port if it \ + differs from the default) on this host" + ), + } +} + +#[cfg(test)] +mod tests { + use super::require_store_addr; + + #[test] + fn a_configured_address_passes_through() { + assert_eq!( + require_store_addr(Some("10.100.0.1:51821")).expect("configured"), + "10.100.0.1:51821" + ); + } + + #[test] + fn an_unconfigured_store_names_the_option_and_never_guesses() { + let err = require_store_addr(None).expect_err("no store means no push"); + let msg = format!("{err:#}"); + assert!(msg.contains("swarm.snapshotStore.address"), "{msg}"); + // A guessed address or port is the failure this is guarding + // against: it would push an agent's state at whatever answers. + assert!(!msg.contains("51821"), "must not suggest a value: {msg}"); + } + + #[test] + fn an_empty_value_is_treated_as_unset() { + // An empty env var is a misconfiguration, not an address; it + // must fail the same way rather than reaching connect(). + let err = require_store_addr(Some("")).expect_err("empty is not an address"); + assert!(format!("{err:#}").contains("no snapshot store configured")); + } +} diff --git a/hive-c0re/src/swarm_peers.rs b/hive-c0re/src/swarm_peers.rs deleted file mode 100644 index aa2c8cdc..00000000 --- a/hive-c0re/src/swarm_peers.rs +++ /dev/null @@ -1,188 +0,0 @@ -//! 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 f74bee85..5a28902a 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -375,22 +375,22 @@ 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. + /// Push a snapshot to the swarm'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 the store'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. + /// There is no destination field: a swarm has exactly one store, and + /// the daemon reads its address from + /// `services.hyperhive.swarm.snapshotStore`, failing if none is + /// configured. Bare success — nothing is written on this host to + /// report a path for. PushSnapshot { name: Ident, label: String, #[serde(default)] parent: Option, - peer: String, }, } diff --git a/hivectl/src/cli.rs b/hivectl/src/cli.rs index 34f45582..6a619b7d 100644 --- a/hivectl/src/cli.rs +++ b/hivectl/src/cli.rs @@ -638,14 +638,17 @@ pub enum SnapshotCmd { #[arg(long)] dest: String, }, - /// Stream a snapshot to a peer hive's snapshot store over the + /// Stream a snapshot to the swarm'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 + /// binds the sender's address to its key), so there is no credential /// to pass here. + /// + /// There is no destination argument: a swarm has one store, read + /// from `services.hyperhive.swarm.snapshotStore`. Push { /// Snapshot label passed to `subvol snapshot create --label`. label: String, @@ -655,10 +658,5 @@ pub enum SnapshotCmd { /// 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 2f680176..b1427732 100644 --- a/hivectl/src/subvol.rs +++ b/hivectl/src/subvol.rs @@ -44,11 +44,9 @@ 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, + SnapshotCmd::Push { label, parent } => { + subvol_snapshot_push(socket, name, &label, parent.as_deref()).await + } }, } } @@ -223,18 +221,18 @@ async fn subvol_snapshot_send( .await } -/// `subvol snapshot push