vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#2862): one snapshot store per swarm, not one per peer

Browse source 
The push side modelled a store per peer hive: a --peer argument, a
swarm.peers.<domain>.snapshotStorePort option, and a swarm_peers module
whose entire job was answering "which peer". A swarm has exactly one
store, so none of that had anything to select between.

The receiver already proved it. It keys destination directories by
agent, not by sending hive, precisely so an agent that migrates keeps
one unbroken incremental chain -- which only makes sense if every hive
pushes to the same place. Per-hive stores would split the chain in two,
the case that keying exists to prevent.

So the destination moves to services.hyperhive.swarm.snapshotStore,
rendered into HYPERHIVE_SNAPSHOT_STORE, and swarm_peers is deleted
rather than adapted. address has no default because it is a
deployment fact this host cannot derive; port defaults because it is a
convention both ends read from the same option docs. An unset or empty
address fails naming the option instead of connecting somewhere
arbitrary, and a test asserts the message suggests no value.
This commit is contained in:
atlas 2026-07-31 22:00:59 +02:00 • committed by mara
commit ba71e45486
9 changed files with 171 additions and 279 deletions
Download patch file Download diff file Expand all files Collapse all files

1
hive-c0re/src/main.rs
View file

@ -35,7 +35,6 @@ mod snapshot_push;
mod socket_server;
mod stats;
mod stores;
mod swarm_peers;
mod webhook_secret;
mod workers;

10
hive-c0re/src/server.rs
View file

@ -268,8 +268,7 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> 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<HostResponse> {
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())
}

113
hive-c0re/src/snapshot_push.rs
View file

@ -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"));
}
}

188
hive-c0re/src/swarm_peers.rs
View file

@ -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<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}"
);
}
}

20
hive-host-sock/src/lib.rs
View file

@ -375,22 +375,22 @@ pub enum HostRequest {
parent: Option<String>,
dest: String,
},
/// Push a snapshot to a peer hive's snapshot store over the
/// WireGuard mesh (`hivectl agent <name> 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 <name> 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<String>,
peer: String,
},
}

12
hivectl/src/cli.rs
View file

@ -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<String>,
/// 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,
},
}

25
hivectl/src/subvol.rs
View file

@ -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 <agent> <label> [--parent <label>] --peer <hive>`
/// — stream a snapshot to a peer hive's snapshot store over the mesh.
/// `subvol snapshot push <agent> <label> [--parent <label>]` — stream a
/// snapshot to the swarm's snapshot store over the mesh.
///
/// The network sibling of `send`. Nothing is staged on this host, so
/// there is no path to print: success is silent apart from the
/// confirmation below.
/// confirmation below. No destination argument — a swarm has one store,
/// and the daemon reads its address from the host config.
async fn subvol_snapshot_push(
socket: &Path,
name: &str,
label: &str,
parent: Option<&str>,
peer: &str,
) -> Result<()> {
daemon_request(
socket,
@ -242,14 +240,17 @@ async fn subvol_snapshot_push(
name: crate::util::parse_ident(name)?,
label: label.to_owned(),
parent: parent.map(str::to_owned),
peer: peer.to_owned(),
},
"snapshot push",
)
.await?;
match parent {
Some(p) => println!("pushed {name} snapshot {label:?} (incremental from {p:?}) to {peer}"),
None => println!("pushed {name} snapshot {label:?} (full) to {peer}"),
Some(p) => {
println!(
"pushed {name} snapshot {label:?} (incremental from {p:?}) to the swarm store"
);
}
None => println!("pushed {name} snapshot {label:?} (full) to the swarm store"),
}
Ok(())
}

26
nix/host-modules/hive-c0re/environment.nix
View file

@ -149,15 +149,24 @@ in
# or no gatewayHost is set (no browser-reachable matrix vhost).
HIVE_MATRIX_PUBLIC_URL = "https://${config.services.hyperhive.matrix.gatewayHost}/";
}
// lib.optionalAttrs (config.services.hyperhive.swarm.snapshotStore.address != null) {
# `host:port` of the swarm's single snapshot store, for pushing agent
# snapshots (hive-c0re::snapshot_push). One per swarm, not one per
# peer — the receiver keys destinations by agent so a migrating agent
# keeps one incremental chain. Absent when no store is configured, and
# a push then fails naming the option rather than guessing.
HYPERHIVE_SNAPSHOT_STORE =
let
s = config.services.hyperhive.swarm.snapshotStore;
in
"${s.address}:${toString s.port}";
}
// lib.optionalAttrs (config.services.hyperhive.swarm.peers != { }) {
# Peer hives serialised as a JSON array of {domain, cert_fingerprint,
# wireguard_address?, snapshot_store_port?} objects. Consumed by
# hive-agent::identity::peers(), the dashboard's peer_hives
# StateSnapshot field (P33RS tab), and snapshot pushes
# (hive-c0re::swarm_peers). Domain is the attrset key;
# cert_fingerprint is null for CA-trusted peers; wireguard_address is
# omitted when not part of the mesh; snapshot_store_port is omitted
# when the peer runs no snapshot store.
# wireguard_address?} objects. Consumed by hive-agent::identity::peers()
# + the dashboard's peer_hives StateSnapshot field (P33RS tab). Domain
# is the attrset key; cert_fingerprint is null for CA-trusted peers;
# wireguard_address is omitted when not part of the mesh.
HYPERHIVE_PEERS = builtins.toJSON (
lib.mapAttrsToList (
domain: p:
@ -168,9 +177,6 @@ in
// lib.optionalAttrs (p.wireguardAddress != null) {
wireguard_address = p.wireguardAddress;
}
// lib.optionalAttrs (p.snapshotStorePort != null) {
snapshot_store_port = p.snapshotStorePort;
}
) config.services.hyperhive.swarm.peers
);
}

55
nix/host-modules/swarm.nix
View file

@ -106,24 +106,6 @@
'';
};
snapshotStorePort = lib.mkOption {
type = lib.types.nullOr lib.types.port;
default = null;
example = 51821;
description = ''
TCP port this peer's snapshot store listens on, when it
runs one (`services.hyperhive.snapshotStore`). Injected
into `HYPERHIVE_PEERS` so a pushing hive can reach the
receiver at `wireguardAddress:snapshotStorePort`.
Null means this peer hosts no snapshot store, and pushing
to it fails with that message rather than guessing a port.
The port lives here — on the peer, alongside the mesh
address it pairs with — because it describes *that host's*
deployment, and a pushing hive cannot read the receiver's
own configuration.
'';
};
};
}
);
@ -144,4 +126,41 @@
'';
};
options.services.hyperhive.swarm.snapshotStore = {
address = lib.mkOption {
type = lib.types.nullOr lib.types.str;
default = null;
example = "10.100.0.1";
description = ''
Mesh address of the swarm's snapshot store — the single
`btrfs receive` endpoint every hive in this swarm pushes agent
snapshots to. Bare IP, no prefix.
There is exactly **one** store per swarm, not one per peer: 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.
Null means this swarm has no store configured, and pushing fails
saying so rather than guessing an address. Set it on every hive
that pushes; the receiving host separately sets
`services.hyperhive.snapshotStore.enable`.
'';
};
port = lib.mkOption {
type = lib.types.port;
default = 51821;
description = ''
TCP port the swarm's snapshot store listens on. Must match the
receiving host's `services.hyperhive.snapshotStore.port`.
Defaulted (unlike `address`) because it is a shared convention
both sides read from the same option docs — whereas an address
is deployment-specific and cannot be guessed.
'';
};
};
}