chore: Release zfsbackup version 0.5.0

Cargo.lock

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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zfsbackup"
-version = "0.4.0"
+version = "0.5.0"
 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 {
+        let mut reader: Box<dyn Read> = Box::new(send_stdout);
-        reader = filter.filter(reader);
+        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())?;
+    // The send process will typically finish first. If it failed, terminate the
-
+    // receiver.
-    let receive_status = receive_process.wait().map_err(|e| e.to_string())?;
     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/main.rs

@@ -5,7 +5,7 @@ use std::io::IsTerminal;
 use std::sync::mpsc::channel;
 use std::thread;
 use zfsbackup::job::JobBuilder;
-use zfsbackup::progress::{Progressor, log, terminal};
+use zfsbackup::progress::{Progressor, curt, log, terminal};
 
 #[derive(Parser)]
 #[command(version, about, long_about = None)]
@@ -29,6 +29,9 @@ struct Args {
     #[arg(short, long)]
     retain: Option<usize>,
 
+    #[arg(short, long)]
+    progressor: Option<String>,
+
     datasets: Vec<String>,
 }
 
@@ -49,10 +52,23 @@ fn main() -> Result<(), Box<dyn Error>> {
     }
 
     let (tx, rx) = channel();
-    let mut pr: Box<dyn Progressor> = if std::io::stdout().is_terminal() {
+    let mut pr: Box<dyn Progressor> = match args.progressor {
-        Box::new(terminal::Progressor::new(rx))
+        Some(name) => match name.as_str() {
-    } else {
+            "curt" => Box::new(curt::Progressor::new(rx)),
-        Box::new(log::Progressor::new(rx))
+            "terminal" => Box::new(terminal::Progressor::new(rx)),
+            "log" => Box::new(log::Progressor::new(rx)),
+            _ => {
+                eprintln!("progressor {} not found, defaulting to 'log'", name);
+                Box::new(log::Progressor::new(rx))
+            }
+        },
+        None => {
+            if std::io::stdout().is_terminal() {
+                Box::new(terminal::Progressor::new(rx))
+            } else {
+                Box::new(log::Progressor::new(rx))
+            }
+        }
     };
     let handle = thread::spawn(move || pr.run());
     builder = builder.sender(tx);

src/progress/curt.rs

@@ -0,0 +1,84 @@
+use super::{BackupEvent, human_bytes};
+use std::io::Write;
+use std::sync::mpsc::Receiver;
+
+pub struct Progressor {
+    receiver: Receiver<BackupEvent>,
+    source: String,
+}
+
+impl super::Progressor for Progressor {
+    fn run(&mut self) {
+        while let Ok(event) = self.receiver.recv() {
+            match event {
+                BackupEvent::Estimate(_) => {}
+                BackupEvent::FullBackupStart {
+                    source,
+                    dest: _,
+                    index: _,
+                    total: _,
+                } => {
+                    self.source = source;
+                    print!("{}\r", self.source);
+                }
+                BackupEvent::IncrementalBackupStart {
+                    source,
+                    dest: _,
+                    index: _,
+                    total: _,
+                } => {
+                    self.source = source;
+                    print!("{}\r", self.source);
+                }
+                BackupEvent::SnapshotCreate(_) => {}
+                BackupEvent::SnapshotDelete(_) => {}
+                BackupEvent::BytesTransferred {
+                    bytes,
+                    estimated_total,
+                } => {
+                    match estimated_total {
+                        Some(total) => {
+                            let percent: f64 = if total > 0 {
+                                bytes as f64 / total as f64
+                            } else {
+                                0.0
+                            };
+
+                            print!("\x1b[2K{}: {:>3.0}%\r", self.source, percent * 100.0,);
+                        }
+                        None => {
+                            print!(
+                                "\x1b[2K{}: {} transferred\r",
+                                self.source,
+                                human_bytes(bytes as f64)
+                            );
+                        }
+                    }
+                    std::io::stdout().flush().ok();
+                }
+                BackupEvent::SendComplete(bytes) => {
+                    print!(
+                        "\x1b[2K{}: send complete ({})\r",
+                        self.source,
+                        human_bytes(bytes as f64)
+                    );
+                }
+                BackupEvent::DatasetComplete(_) => {
+                    println!("\x1b[2K{}: complete", self.source);
+                }
+                BackupEvent::DryrunComplete(_) => {
+                    println!("\x1b[2K{}: complete", self.source);
+                }
+            }
+        }
+    }
+}
+
+impl Progressor {
+    pub fn new(receiver: Receiver<BackupEvent>) -> Self {
+        Self {
+            receiver,
+            source: String::new(),
+        }
+    }
+}

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

@@ -1,3 +1,4 @@
+pub mod curt;
 pub mod filter;
 pub mod log;
 mod rate;
@@ -16,7 +17,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 +30,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),
+    SnapshotCreate(String),
-    SnapshotDeleted(String),
+    SnapshotDelete(String),
     BytesTransferred {
         bytes: u64,
         estimated_total: Option<u64>,
     },
-    DatasetCompleted(String),
+    SendComplete(u64),
-    DryrunCompleted(String),
+    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) => {
+                BackupEvent::SendComplete(bytes) => {
-                    println!("Completed backup of dataset: {}", name);
+                    println!(
+                        "\x1b[2KSend completed at {}, awaiting receive completion",
+                        human_bytes(bytes as f64)
+                    );
                 }
-                BackupEvent::DryrunCompleted(name) => {
+                BackupEvent::DatasetComplete(name) => {
-                    println!("Completed dry run backup of dataset: {}", name);
+                    println!("\x1b[2KCompleted backup of dataset: {}", name);
+                }
+                BackupEvent::DryrunComplete(name) => {
+                    println!("\x1b[2KCompleted dry run backup of dataset: {}", name);
                 }
             }
         }