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

Compare commits

base: vinzenz:7b50ae57e4a70bbaafde87e03b6f303f58941fe7
Branches Tags
vinzenz:main
..
compare: vinzenz:d67c86677bc49670f4e1b192ca46d870f69855a3
Branches Tags
vinzenz:main
4 changed files with 19 additions and 157 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docs/networking/snapshot-store.md
View file

@ -108,9 +108,6 @@ the case the store exists to serve.
A `btrfs send` stream carries no notion of _which agent_ it belongs to,
and the subvolume name inside it's chosen by the sender. The
protocol is one `agent <name>` header line, then the raw stream.
The receiver waits 30 s for the header, and systemd stops a transfer
that runs past 6 h in total. The sender gives connect and header write 15 s
each; it puts no limit on the stream itself.
The rule that matters:

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

@ -19,7 +19,6 @@
//! exists to serve.
use std::os::fd::OwnedFd;
use std::time::Duration;
use anyhow::{Context as _, Result, bail};
use hive_priv_sock::PrivRequest;
@ -33,15 +32,6 @@ use crate::priv_client;
/// `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`.
///
@ -53,7 +43,21 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st
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?;
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
@ -88,50 +92,6 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st
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
@ -177,12 +137,7 @@ mod tests {
use tokio::net::{TcpListener, TcpStream};
use std::time::Duration;
use super::{
CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT, bounded, detach_blocking, open_stream,
require_store_addr,
};
use super::{detach_blocking, require_store_addr};
#[tokio::test]
async fn the_handed_off_descriptor_is_blocking() {
@ -203,60 +158,6 @@ mod tests {
);
}
#[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!(

23
nix/host-modules/swarm-snapshot-store.nix
View file

@ -33,11 +33,6 @@ let
# address must be the bare IP, so strip it.
meshAddress = lib.head (lib.splitString "/" wgCfg.address);
# How long a connected peer may stay silent before sending the header
# line. The sender writes it immediately after connecting, so this only
# ever expires for a peer that is not a hive.
headerTimeoutSeconds = 30;
# The receiver. Socket-activated with Accept=yes, so stdin IS the
# accepted connection and systemd hands us the peer address in
# $REMOTE_ADDR --- which, on this interface, is a cryptographically
@ -62,15 +57,8 @@ let
# Read exactly the header line, leaving the byte stream untouched
# for btrfs receive. `read` stops at the newline and does not
# buffer ahead, which is why the header is a line and not a
# fixed-width record. `-t` bounds the wait: a peer that connects
# and stays silent would otherwise hold this root instance forever.
# `read` exits above 128 on timeout and 1 on EOF.
rc=0
read -r -t ${toString headerTimeoutSeconds} keyword agent || rc=$?
if [ "$rc" -gt 128 ]; then
echo "swarm-snapshot-store: peer ''${REMOTE_ADDR:-?} sent no header within ${toString headerTimeoutSeconds}s" >&2
exit 1
elif [ "$rc" -ne 0 ]; then
# fixed-width record.
if ! read -r keyword agent; then
echo "swarm-snapshot-store: peer ''${REMOTE_ADDR:-?} closed before sending a header" >&2
exit 1
fi
@ -248,13 +236,6 @@ in
StandardInput = "socket";
StandardError = "journal";
User = "root";
# Upper bound on one whole transfer, header to the last byte of
# `btrfs receive`. Snapshots are whole agent state roots and run to
# multiple gigabytes (hive-priv `send_agent_snapshot_to_fd`); no
# tighter bound is documented, so this is sized for ~100 GB at
# 5 MB/s (about 5.6 h), not measured. A stalled receive is killed
# here instead of holding a root instance indefinitely.
RuntimeMaxSec = "6h";
PrivateTmp = true;
ProtectHome = true;
ProtectSystem = "strict";

19
nix/module-eval/snapshot-store.nix
View file

@ -3,8 +3,7 @@
# executes").
#
# The snapshot store's receiver socket: where it listens, that it can bind
# before the mesh interface is up, that it refuses a wildcard, and that a
# silent or stalled transfer is bounded.
# before the mesh interface is up, and that it refuses a wildcard.
{
pkgs,
lib,
@ -47,10 +46,6 @@ let
snapshotStore.enable = true;
};
unit = store.systemd.services."swarm-snapshot-store@";
moduleSource = builtins.readFile ../host-modules/swarm-snapshot-store.nix;
socket = store.systemd.sockets.swarm-snapshot-store.socketConfig;
refusedAsWildcard =
@ -85,18 +80,6 @@ let
name = "a wildcard mesh address, IPv4 or IPv6, is refused";
ok = refusedAsWildcard storeOnWildcard && refusedAsWildcard storeOnWildcardV6;
}
{
# Without it a stalled `btrfs receive` holds a root instance forever.
name = "the receiver unit has a runtime ceiling";
ok = unit.serviceConfig.RuntimeMaxSec or null == "6h";
}
{
# The script is a store path, so its text is not evaluable without
# building it; the module source is the nearest thing to read. A
# silent peer must not hold the root instance past the header read.
name = "the receiver script bounds its header read";
ok = lib.hasInfix "read -r -t \${toString headerTimeoutSeconds} keyword agent" moduleSource;
}
];
in
runGroup "snapshot-store" cases