hive-c0re: clear O_NONBLOCK on the snapshot-store socket before btrfs send gets it

The tokio socket stays non-blocking after into_std, and the flag is
shared with hive-priv through SCM_RIGHTS, so btrfs send wrote into a
non-blocking socket.

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<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
@@ -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() {