hive-priv: don't leave a partial snapshot export at dest on failure

send_agent_snapshot_to_file created `dest` with create_new(true) before
checking whether the parent snapshot exists, and left dest behind on any
spawn()/wait_with_output() failure too. Cleanup only ran on the one
remaining path: btrfs send spawning fine and exiting non-zero.

Every other failure listed in the wire doc left a zero-byte export, and
because dest already existed, a retry against the same --dest always hit
the no-overwrite guard — the only way out was an operator manually
removing the file.

Fixes #4687.

- check_send_snapshot_preconditions() validates the snapshot and its
  optional parent before dest is touched at all.
- open_export_dest() combines that check with the create_new open, so the
  ordering can't drift apart again, and is unit-testable without btrfs.
- PartialExportGuard removes dest on drop unless disarmed, covering every
  failure path after dest is created (including the spawn/wait `?`s that
  previously leaked it), not just the btrfs-exit-nonzero case.

hive-priv/src/main.rs

@@ -2049,6 +2049,94 @@ async fn delete_agent_snapshot(agent_name: &str, snapshot_name: &str) -> Result<
     Ok((String::new(), String::new()))
 }
 
+/// Both paths a `SendAgentSnapshotToFile` request reads from — the
+/// snapshot itself, and its optional incremental parent — must exist
+/// before we touch `dest`. Pulled out on its own so that invariant is
+/// checkable without a `dest` path, a `MIGRATE_STAGING_ROOT`, or `btrfs`
+/// at all: a caller cannot create the output file by accident while
+/// still holding a reference to this check.
+fn check_send_snapshot_preconditions(snap: &Path, parent_path: Option<&Path>) -> Result<()> {
+    if !snap.exists() {
+        bail!(
+            "snapshot {} does not exist — create it with `subvol snapshot create` first",
+            snap.display()
+        );
+    }
+    if let Some(parent_path) = parent_path
+        && !parent_path.exists()
+    {
+        bail!(
+            "parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
+            parent_path.display()
+        );
+    }
+    Ok(())
+}
+
+/// Validate preconditions, then atomically create `dest` for a snapshot
+/// export. The two are combined in one function on purpose: keeping
+/// `check_send_snapshot_preconditions` ahead of the `create_new` call is
+/// the entire fix, and folding them together here means that ordering
+/// can't drift back apart in a later edit to `send_agent_snapshot_to_file`.
+/// No `btrfs` involved, so this is unit-testable against plain files.
+fn open_export_dest(snap: &Path, parent_path: Option<&Path>, dest: &Path) -> Result<std::fs::File> {
+    check_send_snapshot_preconditions(snap, parent_path)?;
+    // `create_new` (O_CREAT|O_EXCL) makes the no-overwrite guarantee atomic
+    // instead of a check-then-create race against a concurrent request.
+    match std::fs::File::options()
+        .write(true)
+        .create_new(true)
+        .open(dest)
+    {
+        Ok(f) => Ok(f),
+        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => bail!(
+            "{} already exists — pick a different destination or remove it first \
+             (send never overwrites an existing export)",
+            dest.display()
+        ),
+        Err(e) => Err(e).with_context(|| format!("create {}", dest.display())),
+    }
+}
+
+/// Deletes `path` when dropped, unless [`disarm`](Self::disarm)ed first.
+/// Guards `dest` between the moment `send_agent_snapshot_to_file` creates
+/// it and the moment the export fully succeeds, so any `?`/`bail!` on
+/// that path — including ones a future edit adds — removes the partial
+/// file instead of leaving a zero-byte export that then makes every
+/// retry against the same `--dest` fail with "already exists".
+struct PartialExportGuard<'a> {
+    path: &'a Path,
+    armed: bool,
+}
+
+impl<'a> PartialExportGuard<'a> {
+    fn new(path: &'a Path) -> Self {
+        Self { path, armed: true }
+    }
+
+    fn disarm(mut self) {
+        self.armed = false;
+    }
+}
+
+impl Drop for PartialExportGuard<'_> {
+    fn drop(&mut self) {
+        if !self.armed {
+            return;
+        }
+        // Best-effort: warn (don't fail the whole call over it) if removal
+        // itself fails, so a stuck partial file that later masquerades as a
+        // completed export is at least visible in the log.
+        if let Err(rm_err) = std::fs::remove_file(self.path) {
+            tracing::warn!(
+                dest = %self.path.display(), error = %rm_err,
+                "failed to remove partial export after failure — \
+                 next attempt at this dest will hit the already-exists guard"
+            );
+        }
+    }
+}
+
 /// `SendAgentSnapshotToFile` — stream a read-only snapshot (optionally
 /// incremental against `parent_name`) to a file under
 /// `MIGRATE_STAGING_ROOT` via `btrfs send`. Local-file half of the
@@ -2061,45 +2149,21 @@ async fn send_agent_snapshot_to_file(
     dest_file_name: &str,
 ) -> Result<(String, String)> {
     let snap = snapshot_path(agent_name, snapshot_name);
-    if !snap.exists() {
-        bail!(
-            "snapshot {} does not exist — create it with `subvol snapshot create` first",
-            snap.display()
-        );
-    }
+    let parent_path = parent_name.map(|parent| snapshot_path(agent_name, parent));
 
     std::fs::create_dir_all(MIGRATE_STAGING_ROOT)
         .with_context(|| format!("create {MIGRATE_STAGING_ROOT}"))?;
     let dest = Path::new(MIGRATE_STAGING_ROOT).join(dest_file_name);
-    // `create_new` (O_CREAT|O_EXCL) makes the no-overwrite guarantee atomic
-    // instead of a check-then-create race against a concurrent request.
-    let dest_file = match std::fs::File::options()
-        .write(true)
-        .create_new(true)
-        .open(&dest)
-    {
-        Ok(f) => f,
-        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => bail!(
-            "{} already exists — pick a different destination or remove it first \
-             (send never overwrites an existing export)",
-            dest.display()
-        ),
-        Err(e) => {
-            return Err(e).with_context(|| format!("create {}", dest.display()));
-        }
-    };
+    // Preconditions are checked *inside* this call, before `dest` is
+    // created — see `open_export_dest`'s doc comment for why that ordering
+    // matters and is kept in one place.
+    let dest_file = open_export_dest(&snap, parent_path.as_deref(), &dest)?;
+    let cleanup = PartialExportGuard::new(&dest);
 
     let mut cmd = Command::new("btrfs");
     cmd.arg("send");
-    if let Some(parent) = parent_name {
-        let parent_path = snapshot_path(agent_name, parent);
-        if !parent_path.exists() {
-            bail!(
-                "parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
-                parent_path.display()
-            );
-        }
-        cmd.arg("-p").arg(&parent_path);
+    if let Some(parent_path) = &parent_path {
+        cmd.arg("-p").arg(parent_path);
     }
     cmd.arg(&snap);
     cmd.stdout(std::process::Stdio::from(dest_file));
@@ -2112,24 +2176,13 @@ async fn send_agent_snapshot_to_file(
         .await
         .with_context(|| format!("wait on btrfs send {}", snap.display()))?;
     if !out.status.success() {
-        // Clean up a partial/failed export so a retry doesn't trip the
-        // "already exists" guard on garbage. Best-effort: warn (don't fail
-        // the whole call over it) if removal itself fails, so a stuck
-        // partial file that later masquerades as a completed export is at
-        // least visible in the log.
-        if let Err(rm_err) = std::fs::remove_file(&dest) {
-            tracing::warn!(
-                dest = %dest.display(), error = %rm_err,
-                "failed to remove partial export after btrfs send failure — \
-                 next attempt at this dest will hit the already-exists guard"
-            );
-        }
         bail!(
             "btrfs send {} failed: {}",
             snap.display(),
             String::from_utf8_lossy(&out.stderr).trim()
         );
     }
+    cleanup.disarm();
     tracing::info!(
         agent = %agent_name, snapshot = %snap.display(), dest = %dest.display(),
         parent = ?parent_name, "exported agent snapshot to file"
@@ -3191,10 +3244,10 @@ mod tests {
         AgentTmpfilesEntry, BindMount, OwnedFd, PAUSED_MARKER_FILE, PrivRequest,
         agent_tmpfiles_content, check_fd_agreement, clear_runner_credentials,
         contains_secret_shaped_run, ensure_plain_filename, git_overlay_flags, limits_dropin_body,
-        matrix_token_filename, partial_name, redact_secret_line, remove_marker_in,
-        single_output_path, toplevel_attr, validate_account_name, validate_credential_name,
-        validate_snapshot_name, write_agent_dir_file, write_bridge_dns_marker_in,
-        write_state_file_nofollow,
+        matrix_token_filename, open_export_dest, partial_name, redact_secret_line,
+        remove_marker_in, single_output_path, toplevel_attr, validate_account_name,
+        validate_credential_name, validate_snapshot_name, write_agent_dir_file,
+        write_bridge_dns_marker_in, write_state_file_nofollow,
     };
     use std::path::PathBuf;
     use std::sync::atomic::{AtomicU32, Ordering};
@@ -3933,4 +3986,91 @@ mod tests {
         assert!(ensure_plain_filename("test", "sub/dir").is_err());
         assert!(ensure_plain_filename("test", "matrix-token").is_ok());
     }
+
+    /// Unique scratch dir per test, no external tempfile dep. Same pattern
+    /// as `scratch()` above, kept separate so a rename of one doesn't
+    /// collide with the other's directory names under parallel test runs.
+    fn snapshot_export_scratch() -> PathBuf {
+        static CTR: AtomicU32 = AtomicU32::new(0);
+        let n = CTR.fetch_add(1, Ordering::Relaxed);
+        let dir = std::env::temp_dir().join(format!(
+            "hive-priv-snapshot-export-test-{}-{n}",
+            std::process::id()
+        ));
+        std::fs::create_dir_all(&dir).unwrap();
+        dir
+    }
+
+    /// The fix: a missing parent must be caught before `dest` is created,
+    /// not after. Fails against the pre-fix ordering (`dest` created via
+    /// `create_new` first, parent checked second) — see the issue for the
+    /// inverted run.
+    #[test]
+    fn missing_parent_leaves_no_file_at_dest() {
+        let dir = snapshot_export_scratch();
+        let snap = dir.join("snap");
+        std::fs::write(&snap, b"").unwrap();
+        let parent = dir.join("parent-that-does-not-exist");
+        let dest = dir.join("dest");
+
+        let err = open_export_dest(&snap, Some(&parent), &dest).unwrap_err();
+        assert!(
+            err.to_string().contains("does not exist"),
+            "wrong error: {err}"
+        );
+        assert!(
+            !dest.exists(),
+            "a failed export must not leave a file at dest"
+        );
+    }
+
+    /// A retry against the same `dest` after a precondition failure must
+    /// not trip the no-overwrite guard — there is nothing there to
+    /// overwrite, since the failed attempt never created `dest` (the test
+    /// above). This is the behavior a "stale parent falls back to a full
+    /// send, retried without -p" caller needs against a fixed `--dest`.
+    #[test]
+    fn retry_after_precondition_failure_succeeds() {
+        let dir = snapshot_export_scratch();
+        let snap = dir.join("snap");
+        std::fs::write(&snap, b"").unwrap();
+        let parent = dir.join("parent-that-does-not-exist");
+        let dest = dir.join("dest");
+
+        assert!(open_export_dest(&snap, Some(&parent), &dest).is_err());
+        assert!(!dest.exists());
+
+        // Retry without the missing parent, same dest.
+        let file = open_export_dest(&snap, None, &dest);
+        assert!(
+            file.is_ok(),
+            "retry against a dest no previous attempt touched must succeed: {:?}",
+            file.err()
+        );
+        assert!(dest.exists());
+    }
+
+    /// Control for the two tests above: an export that actually landed at
+    /// `dest` must still refuse a second one. Without this, a bug that
+    /// simply stopped creating `dest` at all (rather than fixing the
+    /// ordering) would also make the retry test above pass.
+    #[test]
+    fn existing_dest_still_refuses_overwrite() {
+        let dir = snapshot_export_scratch();
+        let snap = dir.join("snap");
+        std::fs::write(&snap, b"").unwrap();
+        let dest = dir.join("dest");
+        std::fs::write(&dest, b"already here").unwrap();
+
+        let err = open_export_dest(&snap, None, &dest).unwrap_err();
+        assert!(
+            err.to_string().contains("already exists"),
+            "wrong error: {err}"
+        );
+        assert_eq!(
+            std::fs::read(&dest).unwrap(),
+            b"already here",
+            "must not have touched the existing file"
+        );
+    }
 }