hyperhive/hive-c0re/src/snapshot_push.rs

//! 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 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";

/// 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 mut sock = TcpStream::connect(addr)
        .await
        .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
    // 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
        .context("send header to the swarm snapshot store")?;
    sock.flush()
        .await
        .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
    // 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 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(())
}

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