mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

remove the old snapshot store; tombstone its options

Browse source
This commit is contained in:
atlas 2026-11-09 23:13:18 +01:00 • committed by mara
commit 20d286f06a
4 changed files with 97 additions and 1308 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -33,7 +33,6 @@ mod paths;
mod priv_client;
mod secret_file;
mod server;
mod snapshot_push;
mod socket_server;
mod stats;
mod stores;

216
hive-c0re/src/priv_client.rs
View file

@ -12,10 +12,9 @@ use hive_priv_sock::{
NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
};
use hive_types::NixBuilds;
use std::os::fd::{AsRawFd as _, OwnedFd, RawFd};
use std::path::Path;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Interest};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
/// Send a single request to `hive-priv` and return the response.
@ -48,144 +47,6 @@ async fn call_at(sock: &Path, 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> {
call_with_fd_at(Path::new(PRIV_SOCK), req, fd).await
}
async fn call_with_fd_at(path: &Path, req: &PrivRequest, fd: OwnedFd) -> Result<PrivResponse> {
let mut stream = UnixStream::connect(path)
.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`).
@ -727,16 +588,14 @@ fn ok(resp: PrivResponse) -> Result<()> {
mod tests {
use super::{
BindMount, CredentialMount, HiveFacts, NetworkIsolation, NixBuilds, PrivRequest,
PrivResponse, PrivStream, call_at, call_streaming_at, call_with_fd_at, check,
create_container_request, older_hive_priv_refused, parse_qgroup_usage,
resource_limits_request, send_snapshot_request, update_container_request, within,
write_nspawn_flags_request,
PrivResponse, PrivStream, call_at, call_streaming_at, check, create_container_request,
older_hive_priv_refused, parse_qgroup_usage, resource_limits_request,
send_snapshot_request, update_container_request, within, write_nspawn_flags_request,
};
use anyhow::Context as _;
use serde_json::{Value, json};
use std::io::{Read as _, Write as _};
use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd};
use std::os::unix::net::{UnixListener, UnixStream};
use std::os::unix::net::UnixListener;
use std::time::Duration;
fn wire(req: &PrivRequest) -> Value {
@ -1043,69 +902,4 @@ mod tests {
assert!(err.to_string().contains("before sending Done"), "{err}");
server.join().expect("server");
}
/// `recvmsg` the request and the one descriptor that rides with it.
fn recv_with_fd(conn: &UnixStream) -> (Vec<u8>, OwnedFd) {
let mut buf = [0_u8; 256];
let mut iov = libc::iovec {
iov_base: buf.as_mut_ptr().cast(),
iov_len: buf.len(),
};
// u64 elements give the control buffer the 8-byte alignment a
// `cmsghdr` read needs; a byte array would be misaligned.
let mut cmsg = [0_u64; 8];
// SAFETY: an all-zero msghdr is valid; every pointer set below
// refers to a buffer that outlives the recvmsg call.
let (n, fd) = unsafe {
let mut msg: libc::msghdr = std::mem::zeroed();
msg.msg_iov = &raw mut iov;
msg.msg_iovlen = 1;
msg.msg_control = cmsg.as_mut_ptr().cast();
msg.msg_controllen = std::mem::size_of_val(&cmsg);
let n = libc::recvmsg(conn.as_raw_fd(), &raw mut msg, 0);
assert!(n > 0, "recvmsg failed");
let hdr = libc::CMSG_FIRSTHDR(&raw const msg);
assert!(!hdr.is_null(), "no ancillary data arrived");
assert_eq!((*hdr).cmsg_level, libc::SOL_SOCKET);
assert_eq!((*hdr).cmsg_type, libc::SCM_RIGHTS);
let fd = std::ptr::read_unaligned(libc::CMSG_DATA(hdr).cast::<RawFd>());
(n, OwnedFd::from_raw_fd(fd))
};
(buf[..usize::try_from(n).expect("positive")].to_vec(), fd)
}
#[tokio::test]
async fn a_descriptor_arrives_with_the_request_and_is_closed_on_our_side() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("priv.sock");
let listener = UnixListener::bind(&path).expect("bind");
let server = std::thread::spawn(move || {
let (mut conn, _) = listener.accept().expect("accept");
let (request, fd) = recv_with_fd(&conn);
let mut rest = String::new();
conn.read_to_string(&mut rest).expect("read to EOF");
conn.write_all(DONE.as_bytes()).expect("reply");
(request, fd)
});
let (mut read_end, write_end) = std::io::pipe().expect("pipe");
let resp = call_with_fd_at(&path, &PrivRequest::DaemonReload, write_end.into())
.await
.expect("answered");
assert!(resp.ok);
let (request, received) = server.join().expect("server");
assert_eq!(request, b"{\"op\":\"daemon_reload\"}\n");
// The descriptor that arrived is the pipe's write end, and our
// copy is already closed: writing through the received one and
// then dropping it is what ends the read.
let mut write_end = std::fs::File::from(received);
write_end.write_all(b"ping").expect("write via received fd");
drop(write_end);
let mut out = String::new();
read_end
.read_to_string(&mut out)
.expect("EOF after both ends closed");
assert_eq!(out, "ping");
}
}

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

@ -202,11 +202,6 @@ 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,
} => handle_push_snapshot(name.as_str(), label, parent.as_deref()).await?,
})
}
.await;
@ -420,20 +415,6 @@ async fn handle_send_snapshot(
Ok(HostResponse::messages(vec![path]))
}
/// 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>,
) -> Result<HostResponse> {
crate::snapshot_push::push_agent_snapshot(name, label, parent)
.await
.with_context(|| format!("push {name} snapshot (label {label:?}) to the swarm store"))?;
Ok(HostResponse::success())
}
/// Single-agent queue verbs the admin socket exposes. Each submits the
/// matching DAG (persisting the `wanted` intent, serializing on the
/// agent's lease, with the transient/crash-watch suppression the old

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

@ -1,285 +0,0 @@
//! Push an agent snapshot to the swarm's snapshot store.
//!
//! 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
//! 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 std::os::fd::OwnedFd;
use std::time::Duration;
use anyhow::{Context as _, Result, bail};
use hive_priv_sock::PrivRequest;
use tokio::io::AsyncWriteExt as _;
use tokio::net::TcpStream;
use crate::priv_client;
/// `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";
/// Deadline for the TCP connect to the store. An unreachable mesh address
/// otherwise waits out the kernel's own connect timeout while holding the
/// admin-socket handler.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
/// Deadline for writing and flushing the one header line into a connected
/// socket. Shorter than the receiver's own header deadline.
const HEADER_WRITE_TIMEOUT: Duration = Duration::from_secs(15);
/// Send `snapshot` of `agent` to the swarm's snapshot store, optionally
/// as an incremental against `parent`.
///
/// # Errors
///
/// 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 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
// EOF when `btrfs send` finishes rather than hanging on a descriptor
// this process is still holding.
let fd = detach_blocking(sock)?;
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 the swarm snapshot store"))?;
if !resp.ok {
bail!(
"push of {agent}/{snapshot} failed: {}",
resp.error.unwrap_or_else(|| resp.stderr.clone())
);
}
tracing::info!(
agent,
snapshot,
parent,
store = addr,
"pushed agent snapshot to the swarm snapshot store"
);
Ok(())
}
/// Connect to the store and write the protocol header.
///
/// 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. 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,
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}"))?;
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
/// flag lives on the open file description, which `hive-priv` shares with
/// us through `SCM_RIGHTS`, so `btrfs send` would write to a
/// non-blocking socket and get `EAGAIN` whenever the send buffer is full.
fn detach_blocking(sock: TcpStream) -> Result<OwnedFd> {
let std_sock = sock
.into_std()
.context("detach snapshot-store socket for hand-off")?;
std_sock
.set_nonblocking(false)
.context("switch snapshot-store socket to blocking mode for hand-off")?;
Ok(std_sock.into())
}
/// 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 std::os::fd::AsRawFd as _;
use tokio::net::{TcpListener, TcpStream};
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() {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
let sock = TcpStream::connect(listener.local_addr().expect("addr"))
.await
.expect("connect");
let fd = detach_blocking(sock).expect("detach");
// SAFETY: `fd` is a live descriptor owned by this test.
let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) };
assert!(flags >= 0, "F_GETFL failed");
assert_eq!(
flags & libc::O_NONBLOCK,
0,
"btrfs send would get EAGAIN on a full socket buffer"
);
}
#[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!(
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"));
}
}