refactor(data): address review on streaming download (#111)

- TempDownload RAII guard: removes the staging file on every disk-path
  error AND on a panic unwind out of the block_in_place decrypt loop,
  replacing three duplicated cleanup arms (#1). drop(file) before rename
  for Windows.
- New Error::Cancelled variant for a dropped receiver; was misclassified
  as Error::Network (#3). Routed to ApplicationError in classify_error so
  caller-initiated cancellation is not retried as a transport failure.
- Doc the exact channel item type Result<Bytes, Error> on
  file_download_to_sender (#4).
- Drop now-stale #[allow(clippy::unused_async)] on file_download (#7).
- Harden e2e test: assert each streamed chunk is non-empty and >=2
  segments arrive (multi-batch property), rename to
  test_file_download_to_sender_multibatch_round_trip (#6).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: Nic-dorman

ant-core/src/data/client/file.rs

@@ -778,6 +778,55 @@ fn spawn_file_encryption(path: PathBuf) -> Result<EncryptionChannels> {
     Ok((chunk_rx, datamap_rx, handle))
 }
 
+/// RAII guard for the staging temp file used during a disk download.
+///
+/// Removes the file on drop — including a panic unwind out of the
+/// `block_in_place` decrypt loop — unless [`commit`](Self::commit) has
+/// promoted it to its final path. Centralizes the cleanup the explicit error
+/// arms used to repeat.
+struct TempDownload {
+    /// `Some` while the staging file may need cleanup; `None` once committed.
+    path: Option<PathBuf>,
+}
+
+impl TempDownload {
+    fn new(path: PathBuf) -> Self {
+        Self { path: Some(path) }
+    }
+
+    /// Path of the staging file (valid until `commit`).
+    fn path(&self) -> &Path {
+        self.path
+            .as_deref()
+            .expect("TempDownload::path called after commit")
+    }
+
+    /// Rename the staged file to `dest`. On success the guard is defused so
+    /// `Drop` is a no-op; on failure the guard stays armed and `Drop` removes
+    /// the orphaned temp file.
+    fn commit(mut self, dest: &Path) -> std::io::Result<()> {
+        std::fs::rename(self.path(), dest)?; // err → guard armed → Drop cleans up
+        self.path = None; // success → nothing left to clean
+        Ok(())
+    }
+}
+
+impl Drop for TempDownload {
+    fn drop(&mut self) {
+        if let Some(path) = self.path.take() {
+            if let Err(e) = std::fs::remove_file(&path) {
+                // Absent file is fine (never created / already gone).
+                if e.kind() != std::io::ErrorKind::NotFound {
+                    warn!(
+                        "Failed to remove temp download file {}: {e}",
+                        path.display()
+                    );
+                }
+            }
+        }
+    }
+}
+
 impl Client {
     /// Upload a file to the network using streaming self-encryption.
     ///
@@ -2189,7 +2238,6 @@ impl Client {
     ///
     /// Returns an error if any chunk cannot be retrieved, decryption fails,
     /// or the file cannot be written.
-    #[allow(clippy::unused_async)]
     pub async fn file_download(&self, data_map: &DataMap, output: &Path) -> Result<u64> {
         self.file_download_with_progress(data_map, output, None)
             .await
@@ -2574,7 +2622,8 @@ impl Client {
     ///
     /// Same as [`Client::file_download`] but sends [`DownloadEvent`]s for UI
     /// feedback. Streams to a temp file (one decrypt batch resident at a time)
-    /// and renames atomically on success.
+    /// and renames atomically on success. A [`TempDownload`] guard removes the
+    /// staging file on any error path, including a panic.
     pub async fn file_download_with_progress(
         &self,
         data_map: &DataMap,
@@ -2587,47 +2636,28 @@ impl Client {
         let unique: u64 = rand::random();
         let tmp_path = parent.join(format!(".ant_download_{}_{unique}.tmp", std::process::id()));
 
-        let mut file = std::fs::File::create(&tmp_path)?;
-        let write_result = self
+        // Guard removes the staging file on any early return OR a panic unwind
+        // out of the `block_in_place` decrypt loop; defused only by a
+        // successful commit(). Centralizes what used to be three duplicated
+        // cleanup arms.
+        let tmp = TempDownload::new(tmp_path);
+        let mut file = std::fs::File::create(tmp.path())?;
+
+        let bytes_written = self
             .download_decrypted_chunks(data_map, progress, |bytes| {
                 let r = file.write_all(&bytes).map_err(Error::from);
                 std::future::ready(r)
             })
-            .await
-            .and_then(|bytes_written| {
-                file.flush()?;
-                Ok(bytes_written)
-            });
+            .await?;
+        file.flush()?;
+        drop(file); // close the handle before rename (Windows won't rename an open file)
 
-        match write_result {
-            Ok(bytes_written) => match std::fs::rename(&tmp_path, output) {
-                Ok(()) => {
-                    info!(
-                        "File downloaded: {bytes_written} bytes written to {}",
-                        output.display()
-                    );
-                    Ok(bytes_written)
-                }
-                Err(rename_err) => {
-                    if let Err(cleanup_err) = std::fs::remove_file(&tmp_path) {
-                        warn!(
-                            "Failed to remove temp download file {}: {cleanup_err}",
-                            tmp_path.display()
-                        );
-                    }
-                    Err(rename_err.into())
-                }
-            },
-            Err(e) => {
-                if let Err(cleanup_err) = std::fs::remove_file(&tmp_path) {
-                    warn!(
-                        "Failed to remove temp download file {}: {cleanup_err}",
-                        tmp_path.display()
-                    );
-                }
-                Err(e)
-            }
-        }
+        tmp.commit(output)?;
+        info!(
+            "File downloaded: {bytes_written} bytes written to {}",
+            output.display()
+        );
+        Ok(bytes_written)
     }
 
     /// Download and decrypt a file, streaming the plaintext to `sink` instead
@@ -2637,7 +2667,14 @@ impl Client {
     /// receives bytes progressively as each batch decrypts, suitable for
     /// forwarding to an HTTP chunked body or a gRPC response stream. The
     /// bounded `sink` applies backpressure. If the receiver is dropped (e.g.
-    /// the client disconnected) the download stops early and returns an error.
+    /// the client disconnected) the download stops early and returns
+    /// [`Error::Cancelled`].
+    ///
+    /// The channel item type is `Result<Bytes, Error>`, so the caller sets up:
+    ///
+    /// ```ignore
+    /// let (tx, rx) = tokio::sync::mpsc::channel::<Result<Bytes, Error>>(8);
+    /// ```
     ///
     /// Typically the caller `tokio::spawn`s this and converts the matching
     /// `Receiver` into its response stream. Requires a multi-threaded Tokio
@@ -2653,7 +2690,7 @@ impl Client {
             async move {
                 sink.send(Ok(bytes))
                     .await
-                    .map_err(|_| Error::Network("download stream receiver dropped".into()))
+                    .map_err(|_| Error::Cancelled("download stream receiver dropped".into()))
             }
         })
         .await

ant-core/src/data/client/mod.rs

@@ -47,8 +47,10 @@ use tracing::debug;
 ///   chunks could not be stored)
 /// - `AlreadyStored`, `Encryption`, `Crypto`, `Payment`,
 ///   `Serialization`, `InvalidData`, `SignatureVerification`,
-///   `Config`, `InsufficientDiskSpace`, `CostEstimationInconclusive`
-///   -> `ApplicationError` (would happen on a perfectly healthy link)
+///   `Config`, `InsufficientDiskSpace`, `CostEstimationInconclusive`,
+///   `Cancelled` -> `ApplicationError` (would happen on a perfectly
+///   healthy link; `Cancelled` is caller-initiated and must not be retried
+///   as a transport failure)
 pub(crate) fn classify_error(err: &Error) -> Outcome {
     match err {
         Error::Timeout(_) => Outcome::Timeout,
@@ -68,6 +70,7 @@ pub(crate) fn classify_error(err: &Error) -> Outcome {
         | Error::Config(_)
         | Error::InsufficientDiskSpace(_)
         | Error::CostEstimationInconclusive(_)
+        | Error::Cancelled(_)
         | Error::BadQuoteBinding { .. } => Outcome::ApplicationError,
     }
 }
@@ -679,6 +682,7 @@ mod tests {
             | Error::AlreadyStored
             | Error::InsufficientDiskSpace(_)
             | Error::CostEstimationInconclusive(_)
+            | Error::Cancelled(_)
             | Error::PartialUpload { .. }
             | Error::BadQuoteBinding { .. } => (),
         };

ant-core/src/data/error.rs

@@ -60,6 +60,14 @@ pub enum Error {
     #[error("encryption error: {0}")]
     Encryption(String),
 
+    /// The operation was cancelled by the caller rather than failing.
+    ///
+    /// Returned, for example, by streaming downloads when the consumer drops
+    /// its receiver (a client disconnect) — distinct from a transport
+    /// [`Error::Network`] failure, since nothing went wrong on the wire.
+    #[error("operation cancelled: {0}")]
+    Cancelled(String),
+
     /// Data already exists on the network — no payment needed.
     #[error("already stored on network")]
     AlreadyStored,
@@ -207,6 +215,15 @@ mod tests {
         assert_eq!(err.to_string(), "encryption error: decrypt failed");
     }
 
+    #[test]
+    fn test_display_cancelled() {
+        let err = Error::Cancelled("download stream receiver dropped".to_string());
+        assert_eq!(
+            err.to_string(),
+            "operation cancelled: download stream receiver dropped"
+        );
+    }
+
     #[test]
     fn test_display_insufficient_disk_space() {
         let err = Error::InsufficientDiskSpace("need 100 MB but only 10 MB available".to_string());

ant-core/tests/e2e_file.rs

@@ -70,7 +70,7 @@ async fn test_file_upload_download_round_trip() {
 /// batch, then reassembles the stream and asserts equality with the source.
 #[tokio::test(flavor = "multi_thread")]
 #[serial]
-async fn test_file_download_to_sender_streaming() {
+async fn test_file_download_to_sender_multibatch_round_trip() {
     use tokio::sync::mpsc;
 
     let (client, testnet) = setup().await;
@@ -92,8 +92,13 @@ async fn test_file_download_to_sender_streaming() {
     let dl = tokio::spawn(async move { client.file_download_to_sender(&data_map, tx, None).await });
 
     let mut streamed: Vec<u8> = Vec::with_capacity(data.len());
+    let mut chunk_count = 0usize;
     while let Some(item) = rx.recv().await {
         let chunk = item.expect("stream chunk should be Ok");
+        // A buggy "send one empty/sentinel then drop" producer would still
+        // close the channel; assert each delivered chunk carries real bytes.
+        assert!(!chunk.is_empty(), "streamed chunk should be non-empty");
+        chunk_count += 1;
         streamed.extend_from_slice(&chunk);
     }
 
@@ -102,6 +107,12 @@ async fn test_file_download_to_sender_streaming() {
         .expect("download task should join")
         .expect("file_download_to_sender should succeed");
 
+    // The whole point of the streaming path: a multi-batch payload must arrive
+    // as more than one segment, not buffered and emitted in one shot.
+    assert!(
+        chunk_count >= 2,
+        "multi-batch payload should stream as ≥2 segments, got {chunk_count}"
+    );
     assert_eq!(streamed, data, "streamed content should match original");
     assert_eq!(
         bytes_streamed,