Tidy command execution

Also clean up some match arms.

.github/workflows/goreleaser.yaml

@@ -5,6 +5,9 @@ on:
     tags:
       - "*"
 
+permissions:
+  contents: write
+
 jobs:
   release:
     runs-on: ubuntu-latest

Cargo.lock

@@ -154,12 +154,24 @@ version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
 
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
 [[package]]
 name = "find-msvc-tools"
 version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
 
+[[package]]
+name = "hashbrown"
+version = "0.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
+
 [[package]]
 name = "heck"
 version = "0.5.0"
@@ -190,6 +202,16 @@ dependencies = [
  "cc",
 ]
 
+[[package]]
+name = "indexmap"
+version = "2.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
+dependencies = [
+ "equivalent",
+ "hashbrown",
+]
+
 [[package]]
 name = "is_terminal_polyfill"
 version = "1.70.2"
@@ -413,8 +435,9 @@ dependencies = [
 
 [[package]]
 name = "zfsbackup"
-version = "0.2.2"
+version = "0.3.0"
 dependencies = [
  "chrono",
  "clap",
+ "indexmap",
 ]

Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "zfsbackup"
-version = "0.2.2"
+version = "0.3.0"
 edition = "2024"
 publish = false
 
 [dependencies]
 chrono = "0.4.44"
 clap = { version = "4.6.1", features = ["derive"] }
+indexmap = "2.14.0"

src/command.rs

@@ -6,7 +6,12 @@ pub trait Filter {
     fn filter(&self, reader: Box<dyn Read>) -> Box<dyn Read>;
 }
 
-pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
+// TODO: Make this support quoting for args with spaces (and escape nested quotes)
+fn display_command(command: &Vec<&str>) -> String {
+    command.join(" ")
+}
+
+pub fn exec(command: &Vec<&str>) -> Result<String, String> {
     if command.is_empty() {
         return Err("Command is empty".to_string());
     }
@@ -14,11 +19,13 @@ pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
     if command.len() > 1 {
         cmd.args(&command[1..]);
     }
-    let output = cmd.output().map_err(|e| e.to_string())?;
+    let output = cmd
+        .output()
+        .map_err(|e| format!("Failed to run command {}: {}", command[0], e))?;
     if !output.status.success() {
         return Err(format!(
-            "Command {:?} failed with status {}: {}",
+            "Command '{}' failed with status {}: {}",
-            command,
+            display_command(command),
             output.status,
             String::from_utf8_lossy(&output.stderr)
         ));
@@ -27,7 +34,10 @@ pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
     Ok(output_str)
 }
 
-pub fn exec_piped_commands(
+/// Executes a pipeline of commands, where the output of the source command is passed through a
+/// series of filters before being piped into the destination command. The filters are applied in
+/// the order they are provided.
+pub fn exec_pipe(
     source: &Vec<&str>,
     dest: &Vec<&str>,
     filters: Vec<Box<dyn Filter>>,

src/job.rs

@@ -2,34 +2,331 @@ use crate::command;
 use crate::progress::BackupEvent;
 use crate::progress::filter;
 use chrono::{Local, NaiveDateTime};
+use indexmap::set::IndexSet;
 use std::collections::HashSet;
+use std::fmt;
 use std::sync::mpsc::Sender;
 
+pub struct Job {
+    datasets: IndexSet<String>,
+    target: String,
+    source_zfs_command: Vec<String>,
+    target_zfs_command: Vec<String>,
+    retain: usize,
+    sender: Option<Sender<BackupEvent>>,
+}
+
+#[derive(Copy, Clone)]
+enum JobSide {
+    Source,
+    Target,
+}
+
+struct Snapshot {
+    dataset: String,
+    snapshot_id: String,
+    side: JobSide,
+}
+
+impl fmt::Display for Snapshot {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}@{}", self.dataset, self.snapshot_id)
+    }
+}
+
+impl Job {
+    fn send_event(&self, event: BackupEvent) {
+        if let Some(sender) = &self.sender {
+            sender.send(event).ok();
+        }
+    }
+
+    fn get_side_command(&self, side: JobSide) -> Vec<&str> {
+        match side {
+            JobSide::Source => self.source_zfs_command.iter().map(|s| s.as_str()).collect(),
+            JobSide::Target => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
+        }
+    }
+
+    fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<Snapshot, String> {
+        let snapshot = Snapshot {
+            dataset: dataset.to_string(),
+            snapshot_id: Local::now().format("%Y-%m-%dT%H:%M:%S").to_string(),
+            side,
+        };
+        let name = snapshot.to_string();
+        let mut cmd = self.get_side_command(side);
+        cmd.extend(["snapshot", &name]);
+        command::exec(&cmd)?;
+        self.send_event(BackupEvent::SnapshotCreated(name.clone()));
+        Ok(snapshot)
+    }
+
+    fn estimate(&self, source: &str, inc_snapshot: Option<&str>) -> Result<u64, String> {
+        let mut cmd = self.get_side_command(JobSide::Source);
+        cmd.push("send");
+        if let Some(inc) = inc_snapshot {
+            cmd.extend(["-i", inc]);
+        }
+        cmd.extend(["-n", "-P"]);
+        cmd.push(source);
+        let output = command::exec(&cmd)?;
+        let size = output
+            .lines()
+            .last()
+            .ok_or("estimate parse error")?
+            .split_whitespace()
+            .last()
+            .ok_or("estimate parse error")?
+            .parse::<u64>()
+            .map_err(|e| format!("estimate parse error: {}", e))?;
+        self.send_event(BackupEvent::Estimate(size));
+        Ok(size)
+    }
+
+    fn send_receive(
+        &self,
+        source: &str,
+        dest: &str,
+        inc_snapshot: Option<&str>,
+        total: Option<u64>,
+    ) -> Result<(), String> {
+        let mut send_cmd = self.get_side_command(JobSide::Source);
+        send_cmd.push("send");
+        if let Some(inc) = inc_snapshot {
+            send_cmd.extend(["-i", inc]);
+        }
+        send_cmd.push(source);
+
+        let mut receive_cmd = self.get_side_command(JobSide::Target);
+        receive_cmd.extend(["receive", "-F", dest]);
+        let filters: Vec<Box<dyn command::Filter>> = match self.sender.as_ref() {
+            Some(sender) => vec![Box::new(filter::ByteCountFilter::new(
+                sender.clone(),
+                total,
+            ))],
+            None => vec![],
+        };
+        command::exec_pipe(&send_cmd, &receive_cmd, filters)?;
+        self.send_event(BackupEvent::DatasetCompleted(source.to_string()));
+        Ok(())
+    }
+
+    fn list_snapshot_ids(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
+        let mut cmd = self.get_side_command(side);
+        cmd.extend(["list", "-H", "-o", "name", "-t", "snapshot", source]);
+        let output = command::exec(&cmd)?;
+        let snapshots: Vec<&str> = output
+            .split_whitespace()
+            .map(|s| {
+                let parts: Vec<&str> = s.split("@").collect();
+                if parts.len() == 2 {
+                    Ok(parts[1])
+                } else {
+                    Err(format!("invalid snapshot name: {}", s))
+                }
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        let result: Vec<String> = snapshots
+            .iter()
+            .filter(|s| NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").is_ok())
+            .map(|s| s.to_string())
+            .collect();
+        Ok(result)
+    }
+
+    fn find_latest_matching_snapshot_id(&self, source: &str, dest: &str) -> Result<String, String> {
+        self.find_matching_snapshot_ids(source, dest)?
+            .into_iter()
+            .next()
+            .ok_or(String::from("no matching snapshots"))
+    }
+
+    fn find_matching_snapshot_ids(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
+        let source_snapshots = self.list_snapshot_ids(source, JobSide::Source)?;
+        let mut dest_snapshots = self.list_snapshot_ids(dest, JobSide::Target)?;
+        dest_snapshots.sort_by(|a, b| b.cmp(a));
+
+        let source_set: HashSet<String> = source_snapshots.into_iter().collect();
+
+        let matching_snapshots: Vec<String> = dest_snapshots
+            .into_iter()
+            .filter(|s| source_set.contains(s))
+            .collect();
+
+        if matching_snapshots.is_empty() {
+            Err(String::from("no matching snapshots"))
+        } else {
+            Ok(matching_snapshots)
+        }
+    }
+
+    fn find_old_snapshots(&self, source: &str, dest: &str) -> Result<Vec<Snapshot>, String> {
+        let source_snapshots = self.list_snapshot_ids(source, JobSide::Source)?;
+        let dest_snapshots = self.list_snapshot_ids(dest, JobSide::Target)?;
+        let matching_snapshots = self.find_matching_snapshot_ids(source, dest)?;
+        if matching_snapshots.is_empty() {
+            return Err(String::from("no matching snapshots found"));
+        }
+        let retain: HashSet<String> = matching_snapshots.into_iter().take(self.retain).collect();
+        let result = source_snapshots
+            .into_iter()
+            .filter(|s| !retain.contains(s))
+            .map(|s| Snapshot {
+                dataset: source.to_string(),
+                snapshot_id: s,
+                side: JobSide::Source,
+            })
+            .chain(
+                dest_snapshots
+                    .into_iter()
+                    .filter(|s| !retain.contains(s))
+                    .map(|s| Snapshot {
+                        dataset: dest.to_string(),
+                        snapshot_id: s,
+                        side: JobSide::Target,
+                    }),
+            )
+            .collect();
+        Ok(result)
+    }
+
+    fn delete_snapshot(&self, snapshot: Snapshot) -> Result<(), String> {
+        let mut cmd = self.get_side_command(snapshot.side);
+        let name = snapshot.to_string();
+        cmd.extend(["destroy", &name]);
+        command::exec(&cmd)?;
+        self.send_event(BackupEvent::SnapshotDeleted(name.clone()));
+        Ok(())
+    }
+
+    pub fn run(&self, execute: bool) -> Result<(), String> {
+        for (index, source) in self.datasets.iter().enumerate() {
+            // Check the source exists
+            let mut cmd = self.get_side_command(JobSide::Source);
+            cmd.extend(["list", "-H", "-o", "name", source]);
+            let _ = command::exec(&cmd)?;
+
+            // Check whether the destination exists
+            // TODO: This will assume the destination doesn't exist if the
+            // command fails for any reason.
+            let dest = format!("{}/{}", self.target, source);
+            cmd = self.get_side_command(JobSide::Target);
+            cmd.extend(["list", "-H", "-o", "name", &dest]);
+            let dest_exists = command::exec(&cmd).is_ok();
+
+            // Run backup
+            if dest_exists {
+                self.send_event(BackupEvent::StartingIncrementalBackup {
+                    source: source.clone(),
+                    dest: dest.clone(),
+                    index: index + 1,
+                    total: self.datasets.len(),
+                });
+                let inc_snapshot = format!(
+                    "{}@{}",
+                    source,
+                    self.find_latest_matching_snapshot_id(source, &dest)?
+                );
+
+                let snapshot = self.create_snapshot(source, JobSide::Source)?;
+                let total = self
+                    .estimate(&snapshot.to_string(), Some(&inc_snapshot))
+                    .ok();
+                if !execute {
+                    self.send_event(BackupEvent::DryrunCompleted(source.clone()));
+                    self.delete_snapshot(snapshot)?;
+                    continue;
+                }
+                self.send_receive(&snapshot.to_string(), &dest, Some(&inc_snapshot), total)?;
+            } else {
+                self.send_event(BackupEvent::StartingFullBackup {
+                    source: source.clone(),
+                    dest: dest.clone(),
+                    index: index + 1,
+                    total: self.datasets.len(),
+                });
+                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.delete_snapshot(snapshot)?;
+                    continue;
+                }
+                self.send_receive(&snapshot.to_string(), &dest, None, total)?;
+            }
+
+            // Clean up snapshots
+            self.find_old_snapshots(source, &dest)?
+                .into_iter()
+                .map(|s| self.delete_snapshot(s))
+                .collect::<Result<Vec<_>, _>>()?;
+        }
+        Ok(())
+    }
+}
+
 pub struct JobBuilder {
     sources: Vec<String>,
     target: String,
     source_zfs_command: Vec<String>,
     target_zfs_command: Vec<String>,
-    dryrun: bool,
     retain: usize,
     sender: Option<Sender<BackupEvent>>,
 }
 
+/// Parse a command string into a vector of strings, splitting on whitespace.
+/// Quoting is supported using ' or ".
 fn parse_command(commandstr: &str) -> Vec<String> {
-    commandstr
+    let mut arg = String::new();
-        .split_whitespace()
+    let mut command: Vec<String> = Vec::new();
-        .map(|s| s.to_string())
+    let mut in_single_quote = false;
-        .collect()
+    let mut in_double_quote = false;
+
+    for c in commandstr.chars() {
+        match c {
+            '\'' if !in_double_quote => {
+                in_single_quote = !in_single_quote;
+            }
+            '\'' => {
+                arg.push(c);
+            }
+            '"' if !in_single_quote => {
+                in_double_quote = !in_double_quote;
+            }
+            '"' => {
+                arg.push(c);
+            }
+            ' ' if in_single_quote || in_double_quote => {
+                arg.push(c);
+            }
+            ' ' => {
+                if !arg.is_empty() {
+                    command.push(arg.clone());
+                    arg.clear();
+                }
+            }
+            _ => {
+                arg.push(c);
+            }
+        }
+    }
+    if !arg.is_empty() {
+        command.push(arg);
+    }
+    command
 }
 
 impl JobBuilder {
+    /// Create a new JobBuilder with the given sources and target. The sources should be ZFS
+    /// datasets, optionally ending with "/..." to include all child datasets. The target should be
+    /// a ZFS dataset.
     pub fn new(sources: Vec<String>, target: String) -> Self {
         JobBuilder {
             sources,
             target,
             source_zfs_command: vec!["zfs".to_string()],
             target_zfs_command: vec!["zfs".to_string()],
-            dryrun: false,
             retain: 2,
             sender: None,
         }
@@ -54,11 +351,6 @@ impl JobBuilder {
         self
     }
 
-    pub fn dryrun(mut self) -> Self {
-        self.dryrun = true;
-        self
-    }
-
     pub fn retain(mut self, retain: usize) -> Self {
         self.retain = retain;
         self
@@ -70,16 +362,7 @@ impl JobBuilder {
     }
 
     pub fn build(self) -> Result<Job, String> {
-        let mut job = Job {
+        let mut datasets: IndexSet<String> = IndexSet::new();
-            datasets: vec![],
-            target: self.target,
-            source_zfs_command: self.source_zfs_command,
-            target_zfs_command: self.target_zfs_command,
-            dryrun: self.dryrun,
-            retain: self.retain,
-            sender: self.sender,
-        };
-        let mut datasets: Vec<String> = vec![];
         for source in &self.sources {
             let recurse = source.ends_with("/...");
             let source = source.trim_end_matches("/...");
@@ -90,271 +373,68 @@ impl JobBuilder {
             }
             args.push(source);
 
-            let mut cmd = job.get_side_command(JobSide::Source);
+            let mut cmd: Vec<&str> = self.source_zfs_command.iter().map(|s| s.as_str()).collect();
             cmd.extend(args);
-            let output = command::exec_command(&cmd)?;
+            let output = command::exec(&cmd)?;
             datasets.extend(output.lines().map(str::to_string));
         }
         if datasets.is_empty() {
             return Err(String::from("no matching source datasets found"));
         }
-        job.datasets = datasets;
+        Ok(Job {
-        Ok(job)
+            datasets,
+            target: self.target,
+            source_zfs_command: self.source_zfs_command,
+            target_zfs_command: self.target_zfs_command,
+            retain: self.retain,
+            sender: self.sender,
+        })
     }
 }
 
-pub struct Job {
+#[cfg(test)]
-    datasets: Vec<String>,
+mod tests {
-    target: String,
+    use super::*;
-    source_zfs_command: Vec<String>,
-    target_zfs_command: Vec<String>,
-    dryrun: bool,
-    retain: usize,
-    sender: Option<Sender<BackupEvent>>,
-}
 
-#[derive(Debug)]
+    #[test]
-enum JobSide {
+    fn test_parse_command_simple() {
-    Source,
+        assert_eq!(parse_command("zfs list"), vec!["zfs", "list"]);
-    Destination,
-}
-
-#[derive(Debug)]
-struct Snapshot {
-    snapshot: String,
-    side: JobSide,
-}
-
-impl Job {
-    pub fn dump(&self) {
-        println!("Datasets: {:?}", self.datasets);
-        println!("Target: {}", self.target);
-        println!("Source ZFS Command: {:?}", self.source_zfs_command);
-        println!("Target ZFS Command: {:?}", self.target_zfs_command);
-        println!("Dryrun: {}", self.dryrun);
     }
 
-    fn send_event(&self, event: BackupEvent) {
+    #[test]
-        if let Some(sender) = &self.sender {
+    fn test_parse_command_quoted() {
-            sender.send(event).ok();
+        assert_eq!(
-        }
+            parse_command("ssh host sudo zfs"),
+            vec!["ssh", "host", "sudo", "zfs"]
+        );
     }
 
-    fn get_side_command(&self, side: JobSide) -> Vec<&str> {
+    #[test]
-        match side {
+    fn test_parse_command_with_quotes() {
-            JobSide::Source => self.source_zfs_command.iter().map(|s| s.as_str()).collect(),
+        assert_eq!(
-            JobSide::Destination => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
+            parse_command(r#"ssh "my host" sudo zfs"#),
-        }
+            vec!["ssh", "my host", "sudo", "zfs"]
+        );
     }
 
-    fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<String, String> {
+    #[test]
-        let snapshot = format!("{}@{}", dataset, Local::now().format("%Y-%m-%dT%H:%M:%S"));
+    fn test_parse_command_with_nested_quotes() {
-        let mut cmd = self.get_side_command(side);
+        assert_eq!(
-        cmd.extend(["snapshot", &snapshot]);
+            parse_command(r#"ssh 'my " host' "su'do" zfs"#),
-        command::exec_command(&cmd)?;
+            vec!["ssh", "my \" host", "su'do", "zfs"]
-        self.send_event(BackupEvent::SnapshotCreated(snapshot.clone()));
+        );
-        Ok(snapshot)
     }
 
-    fn estimate(&self, source: &str, inc_snapshot: Option<&str>) -> Result<u64, String> {
+    #[test]
-        let mut cmd = self.get_side_command(JobSide::Source);
+    fn test_parse_command_extra_spaces() {
-        cmd.push("send");
+        assert_eq!(
-        if let Some(inc) = inc_snapshot {
+            parse_command("  ssh   \"  ho st  \"   sudo   zfs  "),
-            cmd.extend(["-i", inc]);
+            vec!["ssh", "  ho st  ", "sudo", "zfs"]
-        }
+        );
-        cmd.extend(["-n", "-P"]);
-        cmd.push(source);
-        let output = command::exec_command(&cmd)?;
-        let size = output
-            .lines()
-            .last()
-            .ok_or("estimate parse error")?
-            .split_whitespace()
-            .last()
-            .ok_or("estimate parse error")?
-            .parse::<u64>()
-            .map_err(|e| format!("estimate parse error: {}", e))?;
-        self.send_event(BackupEvent::Estimate(size));
-        Ok(size)
-    }
-    fn send_receive(
-        &self,
-        source: &str,
-        dest: &str,
-        inc_snapshot: Option<&str>,
-        total: Option<u64>,
-    ) -> Result<(), String> {
-        let mut send_cmd = self.get_side_command(JobSide::Source);
-        send_cmd.push("send");
-        if let Some(inc) = inc_snapshot {
-            send_cmd.extend(["-i", inc]);
-        }
-        send_cmd.push(source);
-
-        let mut receive_cmd = self.get_side_command(JobSide::Destination);
-        receive_cmd.extend(["receive", "-F", dest]);
-        let filters = match self.sender.as_ref() {
-            Some(sender) => vec![filter::CountingReaderBuilder::build(sender.clone(), total)],
-            None => vec![],
-        };
-        command::exec_piped_commands(&send_cmd, &receive_cmd, filters)?;
-        self.send_event(BackupEvent::DatasetCompleted(source.to_string()));
-        Ok(())
     }
 
-    fn list_snapshots(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
+    #[test]
-        let mut cmd = self.get_side_command(side);
+    fn test_parse_command_empty() {
-        cmd.extend(["list", "-H", "-o", "name", "-t", "snapshot", source]);
+        assert_eq!(parse_command(""), Vec::<String>::new());
-        let output = command::exec_command(&cmd)?;
-        let snapshots: Vec<&str> = output
-            .split_whitespace()
-            .map(|s| {
-                let parts: Vec<&str> = s.split("@").collect();
-                if parts.len() == 2 {
-                    Ok(parts[1])
-                } else {
-                    Err(format!("invalid snapshot name: {}", s))
-                }
-            })
-            .collect::<Result<Vec<_>, _>>()?;
-        let result: Vec<String> = snapshots
-            .iter()
-            .filter(|s| NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").is_ok())
-            .map(|s| s.to_string())
-            .collect();
-        Ok(result)
-    }
-
-    fn find_latest_matching_snapshot(&self, source: &str, dest: &str) -> Result<String, String> {
-        self.find_matching_snapshots(source, dest)?
-            .into_iter()
-            .next()
-            .ok_or(String::from("no matching snapshots"))
-    }
-
-    fn find_matching_snapshots(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
-        let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
-        let mut dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
-        dest_snapshots.sort_by(|a, b| b.cmp(a));
-
-        let source_set: HashSet<String> = source_snapshots.into_iter().collect();
-
-        let matching_snapshots: Vec<String> = dest_snapshots
-            .into_iter()
-            .filter(|s| source_set.contains(s))
-            .collect();
-
-        if matching_snapshots.is_empty() {
-            Err(String::from("no matching snapshots"))
-        } else {
-            Ok(matching_snapshots)
-        }
-    }
-
-    fn find_old_snapshots(&self, source: &str, dest: &str) -> Result<Vec<Snapshot>, String> {
-        let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
-        let dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
-        let matching_snapshots = self.find_matching_snapshots(source, dest)?;
-        if matching_snapshots.is_empty() {
-            return Err(String::from("no matching snapshots found"));
-        }
-        let retain: HashSet<String> = matching_snapshots.into_iter().take(self.retain).collect();
-        let result = source_snapshots
-            .into_iter()
-            .filter(|s| !retain.contains(s))
-            .map(|s| Snapshot {
-                snapshot: format!("{}@{}", source, s),
-                side: JobSide::Source,
-            })
-            .chain(
-                dest_snapshots
-                    .into_iter()
-                    .filter(|s| !retain.contains(s))
-                    .map(|s| Snapshot {
-                        snapshot: format!("{}@{}", dest, s),
-                        side: JobSide::Destination,
-                    }),
-            )
-            .collect();
-        Ok(result)
-    }
-
-    fn delete_snapshot(&self, snapshot: Snapshot) -> Result<String, String> {
-        let mut cmd = self.get_side_command(snapshot.side);
-        cmd.extend(["destroy", &snapshot.snapshot]);
-        let res = command::exec_command(&cmd);
-        if res.is_ok() {
-            self.send_event(BackupEvent::SnapshotDeleted(snapshot.snapshot.clone()));
-        }
-        res
-    }
-
-    pub fn run(&self) -> Result<(), String> {
-        for (index, source) in self.datasets.iter().enumerate() {
-            // Check the source exists
-            let mut cmd = self.get_side_command(JobSide::Source);
-            cmd.extend(["list", "-H", "-o", "name", source]);
-            let _ = command::exec_command(&cmd)?;
-
-            // Check whether the destination exists
-            // TODO: This will assume the destination doesn't exist if the
-            // command fails for any reason.
-            let dest = format!("{}/{}", self.target, source);
-            cmd = self.get_side_command(JobSide::Destination);
-            cmd.extend(["list", "-H", "-o", "name", &dest]);
-            let dest_exists = command::exec_command(&cmd).is_ok();
-
-            // Run backup
-            if dest_exists {
-                self.send_event(BackupEvent::StartingIncrementalBackup {
-                    source: source.clone(),
-                    dest: dest.clone(),
-                    index: index + 1,
-                    total: self.datasets.len(),
-                });
-                let inc_snapshot = format!(
-                    "{}@{}",
-                    source,
-                    self.find_latest_matching_snapshot(source, &dest)?
-                );
-
-                let snapshot = self.create_snapshot(source, JobSide::Source)?;
-                let total = self.estimate(&snapshot, Some(&inc_snapshot)).ok();
-                if self.dryrun {
-                    self.send_event(BackupEvent::DryrunCompleted(source.clone()));
-                    self.delete_snapshot(Snapshot {
-                        snapshot: snapshot.clone(),
-                        side: JobSide::Source,
-                    })?;
-                    return Ok(());
-                }
-                self.send_receive(&snapshot, &dest, Some(&inc_snapshot), total)?;
-            } else {
-                self.send_event(BackupEvent::StartingFullBackup {
-                    source: source.clone(),
-                    dest: dest.clone(),
-                    index: index + 1,
-                    total: self.datasets.len(),
-                });
-                let snapshot = self.create_snapshot(source, JobSide::Source)?;
-                let total = self.estimate(&snapshot, None).ok();
-                if self.dryrun {
-                    self.send_event(BackupEvent::DryrunCompleted(source.clone()));
-                    self.delete_snapshot(Snapshot {
-                        snapshot: snapshot.clone(),
-                        side: JobSide::Source,
-                    })?;
-                    return Ok(());
-                }
-                self.send_receive(&snapshot, &dest, None, total)?;
-            }
-
-            // Clean up snapshots
-            self.find_old_snapshots(source, &dest)?
-                .into_iter()
-                .map(|s| self.delete_snapshot(s))
-                .collect::<Result<Vec<_>, _>>()?;
-        }
-        Ok(())
     }
 }

src/main.rs

@@ -35,9 +35,6 @@ struct Args {
 fn main() -> Result<(), Box<dyn Error>> {
     let args = Args::parse();
     let mut builder = JobBuilder::new(args.datasets, args.target);
-    if args.dry_run {
-        builder = builder.dryrun();
-    }
     if let Some(cmd) = args.zfs_command {
         builder = builder.zfs_command(&cmd);
     }
@@ -57,11 +54,15 @@ fn main() -> Result<(), Box<dyn Error>> {
     } else {
         Box::new(log::Progressor::new(rx))
     };
-    thread::spawn(move || pr.run());
+    let handle = thread::spawn(move || pr.run());
-
     builder = builder.sender(tx);
 
-    let job = builder.build()?;
+    // Create the job in a block so the sender is dropped before
-    job.run()?;
+    // joining the progress thread, allowing it to exit cleanly.
+    {
+        let job = builder.build()?;
+        job.run(!args.dry_run)?;
+    }
+    handle.join().unwrap();
     Ok(())
 }

src/progress/filter.rs

@@ -4,7 +4,7 @@ use std::io::{self, Read};
 use std::sync::mpsc::Sender;
 use std::time::{Duration, Instant};
 
-struct CountingReader<R: Read> {
+struct ByteCountReader<R: Read> {
     inner: R,
     sender: Sender<BackupEvent>,
     bytes: u64,
@@ -12,7 +12,7 @@ struct CountingReader<R: Read> {
     total: Option<u64>,
 }
 
-impl<R: Read> Read for CountingReader<R> {
+impl<R: Read> Read for ByteCountReader<R> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let n = self.inner.read(buf)?;
         self.bytes += n as u64;
@@ -29,31 +29,31 @@ impl<R: Read> Read for CountingReader<R> {
     }
 }
 
-impl<R: Read> CountingReader<R> {
+impl<R: Read> ByteCountReader<R> {
     fn new(inner: R, sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
         Self {
             inner,
             sender,
-            total,
             bytes: 0,
             last_send: Instant::now() - Duration::from_secs(1),
+            total,
         }
     }
 }
 
-pub struct CountingReaderBuilder {
+pub(crate) struct ByteCountFilter {
     sender: Sender<BackupEvent>,
     total: Option<u64>,
 }
 
-impl Filter for CountingReaderBuilder {
+impl ByteCountFilter {
-    fn filter(&self, reader: Box<dyn Read>) -> Box<dyn Read> {
+    pub(crate) fn new(sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
-        Box::new(CountingReader::new(reader, self.sender.clone(), self.total))
+        Self { sender, total }
     }
 }
 
-impl CountingReaderBuilder {
+impl Filter for ByteCountFilter {
-    pub fn build(sender: Sender<BackupEvent>, total: Option<u64>) -> Box<dyn Filter> {
+    fn filter(&self, inner: Box<dyn Read>) -> Box<dyn Read> {
-        Box::new(Self { sender, total })
+        Box::new(ByteCountReader::new(inner, self.sender.clone(), self.total))
     }
 }