chore: Release zfsbackup version 0.4.2

Cargo.lock

@@ -435,7 +435,7 @@ dependencies = [
 
 [[package]]
 name = "zfsbackup"
-version = "0.4.0"
+version = "0.4.2"
 dependencies = [
  "chrono",
  "clap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zfsbackup"
-version = "0.4.0"
+version = "0.4.2"
 edition = "2024"
 publish = false
 

src/command.rs

@@ -67,15 +67,26 @@ pub fn exec_pipe(
     let send_stdout = send_process.stdout.take().unwrap();
     let mut receive_stdin = receive_process.stdin.take().unwrap();
 
-    let mut reader: Box<dyn Read> = Box::new(send_stdout);
-    for filter in filters {
-        reader = filter.filter(reader);
+    {
+        let mut reader: Box<dyn Read> = Box::new(send_stdout);
+        for filter in filters {
+            reader = filter.filter(reader);
+        }
+
+        std::io::copy(&mut reader, &mut receive_stdin).map_err(|e| e.to_string())?;
     }
 
-    std::io::copy(&mut reader, &mut receive_stdin).map_err(|e| e.to_string())?;
-
-    let receive_status = receive_process.wait().map_err(|e| e.to_string())?;
+    // The send process will typically finish first. If it failed, terminate the
+    // receiver.
     let send_status = send_process.wait().map_err(|e| e.to_string())?;
+    if !send_status.success() {
+        receive_process.kill().ok();
+        return Err(format!(
+            "Send command {:?} failed with status {}",
+            source, send_status
+        ));
+    }
+    let receive_status = receive_process.wait().map_err(|e| e.to_string())?;
     if !receive_status.success() {
         send_process.kill().ok();
         return Err(format!(
@@ -83,11 +94,5 @@ pub fn exec_pipe(
             dest, receive_status
         ));
     }
-    if !send_status.success() {
-        return Err(format!(
-            "Send command {:?} failed with status {}",
-            source, send_status
-        ));
-    }
     Ok(())
 }

src/job.rs

@@ -58,7 +58,7 @@ impl Job {
         let mut cmd = self.get_side_command(side);
         cmd.extend(["snapshot", &name]);
         command::exec(&cmd)?;
-        self.send_event(BackupEvent::SnapshotCreated(name.clone()));
+        self.send_event(BackupEvent::SnapshotCreate(name.clone()));
         Ok(snapshot)
     }
 
@@ -108,7 +108,7 @@ impl Job {
             None => vec![],
         };
         command::exec_pipe(&send_cmd, &receive_cmd, filters)?;
-        self.send_event(BackupEvent::DatasetCompleted(source.to_string()));
+        self.send_event(BackupEvent::DatasetComplete(source.to_string()));
         Ok(())
     }
 
@@ -196,7 +196,7 @@ impl Job {
         let name = snapshot.to_string();
         cmd.extend(["destroy", &name]);
         command::exec(&cmd)?;
-        self.send_event(BackupEvent::SnapshotDeleted(name.clone()));
+        self.send_event(BackupEvent::SnapshotDelete(name.clone()));
         Ok(())
     }
 
@@ -217,7 +217,7 @@ impl Job {
 
             // Run backup
             if dest_exists {
-                self.send_event(BackupEvent::StartingIncrementalBackup {
+                self.send_event(BackupEvent::IncrementalBackupStart {
                     source: source.clone(),
                     dest: dest.clone(),
                     index: index + 1,
@@ -234,13 +234,13 @@ impl Job {
                     .estimate(&snapshot.to_string(), Some(&inc_snapshot))
                     .ok();
                 if !execute {
-                    self.send_event(BackupEvent::DryrunCompleted(source.clone()));
+                    self.send_event(BackupEvent::DryrunComplete(source.clone()));
                     self.delete_snapshot(snapshot)?;
                     continue;
                 }
                 self.send_receive(&snapshot.to_string(), &dest, Some(&inc_snapshot), total)?;
             } else {
-                self.send_event(BackupEvent::StartingFullBackup {
+                self.send_event(BackupEvent::FullBackupStart {
                     source: source.clone(),
                     dest: dest.clone(),
                     index: index + 1,
@@ -249,7 +249,7 @@ impl Job {
                 let snapshot = self.create_snapshot(source, JobSide::Source)?;
                 let total = self.estimate(&snapshot.to_string(), None).ok();
                 if !execute {
-                    self.send_event(BackupEvent::DryrunCompleted(source.clone()));
+                    self.send_event(BackupEvent::DryrunComplete(source.clone()));
                     self.delete_snapshot(snapshot)?;
                     continue;
                 }

src/progress/filter.rs

@@ -29,6 +29,12 @@ impl<R: Read> Read for ByteCountReader<R> {
     }
 }
 
+impl<R: Read> Drop for ByteCountReader<R> {
+    fn drop(&mut self) {
+        self.sender.send(BackupEvent::SendComplete(self.bytes)).ok();
+    }
+}
+
 impl<R: Read> ByteCountReader<R> {
     fn new(inner: R, sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
         Self {

src/progress/log.rs

@@ -14,7 +14,7 @@ impl super::Progressor for Progressor {
                     println!("Estimated total backup size: {} bytes", size);
                     self.estimated_size = size;
                 }
-                BackupEvent::StartingFullBackup {
+                BackupEvent::FullBackupStart {
                     source,
                     dest,
                     index,
@@ -25,7 +25,7 @@ impl super::Progressor for Progressor {
                         source, dest, index, total
                     );
                 }
-                BackupEvent::StartingIncrementalBackup {
+                BackupEvent::IncrementalBackupStart {
                     source,
                     dest,
                     index,
@@ -36,18 +36,21 @@ impl super::Progressor for Progressor {
                         source, dest, index, total
                     );
                 }
-                BackupEvent::SnapshotCreated(name) => {
+                BackupEvent::SnapshotCreate(name) => {
                     println!("Created snapshot: {}", name);
                 }
-                BackupEvent::SnapshotDeleted(name) => {
+                BackupEvent::SnapshotDelete(name) => {
                     println!("Deleted snapshot: {}", name);
                 }
                 BackupEvent::BytesTransferred { .. } => {}
-                BackupEvent::DatasetCompleted(name) => {
+                BackupEvent::SendComplete(bytes) => {
+                    println!("Send completed: {} bytes", bytes);
+                }
+                BackupEvent::DatasetComplete(name) => {
                     println!("Completed backup of dataset: {}", name);
                     self.estimated_size = 0;
                 }
-                BackupEvent::DryrunCompleted(name) => {
+                BackupEvent::DryrunComplete(name) => {
                     println!("Completed dry run backup of dataset: {}", name);
                     self.estimated_size = 0;
                 }

src/progress/mod.rs

@@ -16,7 +16,7 @@ fn human_bytes(bytes: f64) -> String {
         a /= 1024.0;
         i += 1;
     }
-    format!("{:.2} {:2}", a, units[i])
+    format!("{:.2} {}", a, units[i])
 }
 
 fn format_eta(seconds: f64) -> String {
@@ -29,24 +29,25 @@ fn format_eta(seconds: f64) -> String {
 
 pub enum BackupEvent {
     Estimate(u64),
-    StartingFullBackup {
+    FullBackupStart {
         source: String,
         dest: String,
         index: usize,
         total: usize,
     },
-    StartingIncrementalBackup {
+    IncrementalBackupStart {
         source: String,
         dest: String,
         index: usize,
         total: usize,
     },
-    SnapshotCreated(String),
-    SnapshotDeleted(String),
+    SnapshotCreate(String),
+    SnapshotDelete(String),
     BytesTransferred {
         bytes: u64,
         estimated_total: Option<u64>,
     },
-    DatasetCompleted(String),
-    DryrunCompleted(String),
+    SendComplete(u64),
+    DatasetComplete(String),
+    DryrunComplete(String),
 }

src/progress/terminal.rs

@@ -18,7 +18,7 @@ impl super::Progressor for Progressor {
                     println!("Estimated total backup size: {}", human_bytes(size as f64));
                     self.estimated_size = size;
                 }
-                BackupEvent::StartingFullBackup {
+                BackupEvent::FullBackupStart {
                     source,
                     dest,
                     index,
@@ -29,7 +29,7 @@ impl super::Progressor for Progressor {
                         source, dest, index, total
                     );
                 }
-                BackupEvent::StartingIncrementalBackup {
+                BackupEvent::IncrementalBackupStart {
                     source,
                     dest,
                     index,
@@ -40,10 +40,10 @@ impl super::Progressor for Progressor {
                         source, dest, index, total
                     );
                 }
-                BackupEvent::SnapshotCreated(name) => {
+                BackupEvent::SnapshotCreate(name) => {
                     println!("Created snapshot: {}", name);
                 }
-                BackupEvent::SnapshotDeleted(name) => {
+                BackupEvent::SnapshotDelete(name) => {
                     println!("Deleted snapshot: {}", name);
                 }
                 BackupEvent::BytesTransferred {
@@ -89,11 +89,17 @@ impl super::Progressor for Progressor {
                     }
                     std::io::stdout().flush().ok();
                 }
-                BackupEvent::DatasetCompleted(name) => {
-                    println!("Completed backup of dataset: {}", name);
+                BackupEvent::SendComplete(bytes) => {
+                    println!(
+                        "\x1b[2KSend completed at {}, awaiting receive completion",
+                        human_bytes(bytes as f64)
+                    );
                 }
-                BackupEvent::DryrunCompleted(name) => {
-                    println!("Completed dry run backup of dataset: {}", name);
+                BackupEvent::DatasetComplete(name) => {
+                    println!("\x1b[2KCompleted backup of dataset: {}", name);
+                }
+                BackupEvent::DryrunComplete(name) => {
+                    println!("\x1b[2KCompleted dry run backup of dataset: {}", name);
                 }
             }
         }