Watch
0
0
Fork
You've already forked hyperhive
0

snapshot store: bound the transfer on both ends

The receiver blocked in read until a peer sent a header, and a stalled
btrfs receive was never reaped, so a silent or stuck connection held a
root service instance indefinitely. The receiver now gives the header 30 s
and the unit has RuntimeMaxSec=6h.

The sender had no deadline on the TCP connect or the header write, so an
unreachable mesh address held the admin-socket handler until the kernel
gave up. Both now time out after 15 s and report it like any other
failure. The stream body stays unbounded on the sender: the receiver's
RuntimeMaxSec already caps it, and btrfs send writes it outside this
process.
This commit is contained in:
atlas 2026-10-16 10:08:23 +02:00
commit b03549fde6
3 changed files with 67 additions and 17 deletions

View file

@ -19,6 +19,7 @@
//! exists to serve.
use std::os::fd::OwnedFd;
use std::time::Duration;
use anyhow::{Context as _, Result, bail};
use hive_priv_sock::PrivRequest;
@ -32,6 +33,15 @@ 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`.
///
@ -43,21 +53,7 @@ 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 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")?;
let sock = open_stream(addr, agent).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
@ -92,6 +88,38 @@ 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) -> Result<TcpStream> {
let mut sock = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(addr))
.await
.with_context(|| {
format!(
"connect to the swarm snapshot store at {addr} timed out after {CONNECT_TIMEOUT:?}"
)
})?
.with_context(|| format!("connect to the swarm snapshot store at {addr}"))?;
tokio::time::timeout(HEADER_WRITE_TIMEOUT, async {
sock.write_all(format!("agent {agent}\n").as_bytes())
.await?;
sock.flush().await
})
.await
.with_context(|| {
format!("send header to the swarm snapshot store timed out after {HEADER_WRITE_TIMEOUT:?}")
})?
.context("send header to the swarm snapshot store")?;
Ok(sock)
}
/// 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