diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs index 1cd65905..3098c6d3 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -18,6 +18,8 @@ //! stores would split that chain in two, which is the case the store //! exists to serve. +use std::os::fd::OwnedFd; + use anyhow::{Context as _, Result, bail}; use hive_priv_sock::PrivRequest; use tokio::io::AsyncWriteExt as _; @@ -61,10 +63,7 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st // 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 fd = detach_blocking(sock)?; let resp = priv_client::call_with_fd( &PrivRequest::SendAgentSnapshotToFd { @@ -93,6 +92,22 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st Ok(()) } +/// 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 { + 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 @@ -118,7 +133,30 @@ fn require_store_addr(configured: Option<&str>) -> Result<&str> { #[cfg(test)] mod tests { - use super::require_store_addr; + use std::os::fd::AsRawFd as _; + + use tokio::net::{TcpListener, TcpStream}; + + use super::{detach_blocking, 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" + ); + } #[test] fn a_configured_address_passes_through() {