main.rs
raw
mod cli;
mod config;
mod ui;
use std::{
io::IsTerminal as _,
sync::{Arc, atomic::AtomicBool},
};
use anyhow::{Context as _, Result};
use clap::Parser as _;
use repository::{Repository, SnapshotId};
use beeping::pipeline::{self, EarlyFinish, Event, Outcome};
use crate::{
cli::{BackupArgs, Cli, Command, ExciseArgs, ListArgs, PruneArgs, RepoArgs, RestoreArgs},
config::ConfigFile,
};
fn main() -> Result<()> {
let cli = Cli::parse();
let config = ConfigFile::load()?;
match cli.command {
Command::Init(args) => init(args, &config),
Command::Backup(args) => backup(args, &config),
Command::Restore(args) => restore(args, &config),
Command::Snapshots(args) => snapshots(args, &config),
Command::List(args) => list(args, &config),
Command::Prune(args) => prune(args, &config),
Command::Excise(args) => excise(args, &config),
}
}
fn init(args: RepoArgs, config: &ConfigFile) -> Result<()> {
let location = config::repository_location(args.repository, config)?;
let backend = config::open_backend(&location)?;
let password = config::password(true)?;
Repository::create(backend, password)?;
println!("repository created at {location}");
println!(
"note: the encryption keys are derived from the passphrase and are \
not stored anywhere.\nThe passphrase cannot be changed, and if it \
is lost the backups are unrecoverable."
);
Ok(())
}
fn open_repository(args: RepoArgs, config: &ConfigFile) -> Result<(Repository, String)> {
let location = config::repository_location(args.repository, config)?;
let backend = config::open_backend(&location)?;
let password = config::password(false)?;
let repository = Repository::open(backend, password).map_err(|err| {
anyhow::Error::from(err).context(format!("opening repository {location}"))
})?;
Ok((repository, location))
}
/// Attaches this repository's local chunk cache, which spares a round
/// trip for every chunk the repository is already known to hold — the
/// difference between a resumed run re-asking about every chunk of a
/// huge file and simply reading past it.
///
/// Failing to open one is not worth stopping a backup over: it only
/// means asking the backend as before.
fn attach_chunk_cache(repository: &mut Repository) {
let Ok(base) = config::cache_base() else {
return;
};
// The identity is what the cache checks itself against: a file
// belonging to another repository, or written before someone pruned
// this one, is started again rather than believed
let identity = match repository.cache_identity() {
Ok(identity) => identity,
Err(err) => {
eprintln!("warning: cannot identify the repository ({err}); no chunk cache");
return;
}
};
let path = base.join(hex::encode(repository.fingerprint()));
match repository::ChunkCache::open(path, identity) {
Ok(cache) => repository.attach_chunk_cache(cache),
Err(err) => eprintln!("warning: no chunk cache ({err}); every chunk will be asked about"),
}
}
fn backup(args: BackupArgs, config: &ConfigFile) -> Result<()> {
let (mut repository, location) = open_repository(args.repo, config)?;
if !args.no_chunk_cache {
attach_chunk_cache(&mut repository);
}
let repository = Arc::new(repository);
let root = args
.root
.canonicalize()
.with_context(|| format!("backup root {:?}", args.root))?;
let state_dir = match args.state_dir {
Some(dir) => dir,
None => config::default_state_dir(&[
b"backup",
location.as_bytes(),
root.as_os_str().as_encoded_bytes(),
])?,
};
let mut options = pipeline::backup::BackupOptions::new(&root, state_dir);
options.exclude = config::exclude_globs(config, &args.exclude)?;
options.cross_mount_points = args.cross_mount_points || config.cross_mount_points;
options.concurrent_large_files = args.concurrent_large_files;
options.early_finish = args.early_finish;
// Limits a run cannot honour are reported by the run itself, which
// is the one that knows what its store will say about itself
if !args.ignore_store_limits {
options.minimum_free_space = args.minimum_free_space.or(config.minimum_free_space);
options.maximum_store_size = args.maximum_store_size.or(config.maximum_store_size);
}
for (given, option) in [
(args.scanner_threads, &mut options.scanner_threads),
(args.chunker_threads, &mut options.chunker_threads),
(args.uploader_threads, &mut options.uploader_threads),
] {
if let Some(threads) = given {
*option = threads;
}
}
if let Some(mib) = args.large_file_threshold {
options.large_file_threshold = mib.saturating_mul(1 << 20);
}
// Never capture other suspended runs' state, or the chunk caches,
// either (the pipeline already protects this run's own state and a
// local repository)
for base in [config::state_base(), config::cache_base()] {
if let Ok(base) = base {
options.protect.push(base);
}
}
let title = format!("beeping backup: {} -> {location}", root.display());
run_pipeline(args.plain, title, move |events, suspend| {
pipeline::backup::run(repository, options, events, suspend)
})
}
fn restore(args: RestoreArgs, config: &ConfigFile) -> Result<()> {
let (repository, location) = open_repository(args.repo, config)?;
let repository = Arc::new(repository);
let snapshot = SnapshotId::from_hex(&args.snapshot)
.map_err(|_| anyhow::anyhow!("{:?} is not a snapshot id", args.snapshot))?;
let state_dir = match args.state_dir {
Some(dir) => dir,
None => {
// Different selections are different resumable operations
let snapshot_hex = snapshot.to_hex();
let mut parts: Vec<&[u8]> = vec![
b"restore",
location.as_bytes(),
snapshot_hex.as_bytes(),
args.target.as_os_str().as_encoded_bytes(),
];
parts.extend(args.select.iter().map(|pattern| pattern.as_bytes()));
config::default_state_dir(&parts)?
}
};
// Restoring what a backup has managed to store so far is a
// reasonable thing to want; quietly producing a partial tree is not
if repository.load_snapshot(snapshot)?.coverage == repository::Coverage::Partial {
eprintln!(
"warning: {snapshot} is a partial snapshot, published by a \
backup that had not finished. It covers only what that backup \
had stored when it was written."
);
}
let mut options = pipeline::restore::RestoreOptions::new(snapshot, &args.target, state_dir);
options.select = args.select.clone();
let title = format!("beeping restore: {snapshot} -> {}", args.target.display());
run_pipeline(args.plain, title, move |events, suspend| {
pipeline::restore::run(repository, options, events, suspend)
})
}
/// Runs a pipeline on a worker thread with a reporter on this one, wiring
/// up suspension via UI keys and via SIGINT/SIGTERM.
fn run_pipeline(
plain: bool,
title: String,
launch: impl FnOnce(
crossbeam_channel::Sender<Event>,
Arc<AtomicBool>,
) -> Result<Outcome, pipeline::Error>
+ Send
+ 'static,
) -> Result<()> {
let (events, receiver) = crossbeam_channel::unbounded();
let suspend = Arc::new(AtomicBool::new(false));
for signal in [signal_hook::consts::SIGINT, signal_hook::consts::SIGTERM] {
signal_hook::flag::register(signal, suspend.clone())?;
}
let pipeline = {
let suspend = suspend.clone();
std::thread::Builder::new()
.name("pipeline".to_string())
.spawn(move || launch(events, suspend))?
};
let summary = if plain || !std::io::stdout().is_terminal() {
ui::plain(receiver)
} else {
ui::dashboard(&title, receiver, suspend)?
};
let outcome = pipeline
.join()
.map_err(|_| anyhow::anyhow!("the pipeline thread panicked"))??;
println!("{summary}");
match outcome {
Outcome::Completed(snapshot) => println!("done: snapshot {snapshot}"),
Outcome::Suspended { restorable } => {
println!("suspended; run the same command again to resume");
if let Some(snapshot) = restorable {
println!(
"partial snapshot {snapshot} covers what has been backed \
up so far, and can be restored from"
);
}
}
Outcome::FinishedEarly { snapshot, reason } => {
println!("finished early: {}", ui::describe_early_finish(reason));
let stopped_by_the_store = matches!(
reason,
EarlyFinish::StoreNearlyFull { .. } | EarlyFinish::StoreAtMaximum { .. }
);
match snapshot {
Some(snapshot) => {
println!(
"snapshot {snapshot} covers what was backed up; the \
rest of the tree was never visited. This run is \
over — the same command starts a fresh backup."
);
if stopped_by_the_store {
print!("{}", FULL_STORE_ADVICE);
}
}
None => {
println!("nothing had been backed up, so there was no snapshot to publish");
if stopped_by_the_store {
print!("{}", NOTHING_FITS_ADVICE);
}
}
}
}
}
Ok(())
}
/// What to do about a store that cannot hold the whole tree. The catch
/// is in step 2: the limit that stopped this run is still reached, so
/// the run that publishes the tree you settle for has to be told to
/// ignore it.
const FULL_STORE_ADVICE: &str = "\
The store is still at its limit, so backing this tree up again would
stop the same way, immediately. To turn what fits into a backup worth
keeping:
1. see what was covered with `beeping list`, and exclude (or delete)
what was not
2. run the backup again with --ignore-store-limits. Its chunks are
already stored, so this run writes little more than a manifest
3. prune the snapshot above, which frees whatever only it was holding
";
/// What to do about a store that was already at its limit before the
/// run started, which had therefore nothing it was allowed to do.
const NOTHING_FITS_ADVICE: &str = "\
There was no room to back anything up. Prune what the repository no
longer needs, or give it a limit with some room under it. If this is
the run meant to publish what an earlier one managed to store, it is
--ignore-store-limits that lets it past the limit that stopped that
one.
";
fn prune(args: PruneArgs, config: &ConfigFile) -> Result<()> {
let (mut repository, _) = open_repository(args.repo, config)?;
// The cache is attached so that the prune can take the swept chunks
// out of it, rather than leaving it to be discarded wholesale
attach_chunk_cache(&mut repository);
let mut removing = Vec::new();
for text in &args.snapshots {
removing.push(
SnapshotId::from_hex(text)
.map_err(|_| anyhow::anyhow!("{text:?} is not a snapshot id"))?,
);
}
let outcome = beeping::prune::prune(
&repository,
&removing,
args.dry_run,
args.break_lock,
|note| println!("{note}"),
)?;
if args.dry_run {
println!(
"would remove {} snapshot(s) and sweep {} chunk(s), keeping {} \
snapshot(s) and {} chunk(s)",
outcome.removed.len(),
outcome.swept,
outcome.kept,
outcome.retained
);
} else {
println!(
"removed {} snapshot(s) and swept {} chunk(s); {} snapshot(s) and \
{} chunk(s) remain",
outcome.removed.len(),
outcome.swept,
outcome.kept,
outcome.retained
);
}
Ok(())
}
fn excise(args: ExciseArgs, config: &ConfigFile) -> Result<()> {
let (mut repository, location) = open_repository(args.repo, config)?;
// Attached so the sweep can take what it removes out of the cache,
// rather than leaving it to be discarded wholesale
attach_chunk_cache(&mut repository);
// The recorded root is the one the backup canonicalized; matching it
// from a machine where the tree is not there falls back to the path
// as written
let root = args.root.canonicalize().unwrap_or(args.root.clone());
let state_dir = match args.state_dir {
Some(dir) => dir,
None => config::default_state_dir(&[
b"backup",
location.as_bytes(),
root.as_os_str().as_encoded_bytes(),
])?,
};
let select = beeping::patterns::compile_set(&args.select)?;
let outcome = beeping::excise::excise(
&repository,
&root,
&select,
&state_dir,
args.dry_run,
args.break_lock,
|note| println!("{note}"),
)?;
for snapshot in &outcome.rewritten {
match snapshot.to {
Some(to) => println!(
"{} -> {to}: removed {} entries ({})",
snapshot.from,
ui::count(snapshot.entries),
ui::human_bytes(snapshot.content_bytes)
),
None => println!(
"{}: would remove {} entries ({})",
snapshot.from,
ui::count(snapshot.entries),
ui::human_bytes(snapshot.content_bytes)
),
}
}
let verb = match args.dry_run {
true => "would rewrite",
false => "rewrote",
};
println!(
"{verb} {} snapshot(s), leaving {} untouched",
outcome.rewritten.len(),
outcome.untouched
);
if outcome.unqueued > 0 {
println!(
"{} {} queued path(s) from the suspended backup of this tree; \
exclude them from it as well, or the next run will find them \
again",
match args.dry_run {
true => "would drop",
false => "dropped",
},
ui::count(outcome.unqueued)
);
}
match args.dry_run {
true => println!("nothing was changed; run again without --dry-run"),
false => println!(
"swept {} chunk(s); {} remain",
ui::count(outcome.swept as u64),
ui::count(outcome.retained as u64)
),
}
Ok(())
}
fn list(args: ListArgs, config: &ConfigFile) -> Result<()> {
use std::io::Write as _;
let (repository, _) = open_repository(args.repo, config)?;
let snapshot = SnapshotId::from_hex(&args.snapshot)
.map_err(|_| anyhow::anyhow!("{:?} is not a snapshot id", args.snapshot))?;
let snapshot = repository.load_snapshot(snapshot)?;
let stdout = std::io::stdout();
let mut out = stdout.lock();
for entry in repository.manifest_entries(snapshot.manifest) {
let entry = entry?;
let written = match &entry.kind {
repository::EntryKind::File { len, .. } => {
writeln!(out, "{:>12} {}", len, entry.path.display())
}
repository::EntryKind::Directory => {
writeln!(out, "{:>12} {}/", "dir", entry.path.display())
}
repository::EntryKind::Symlink { target } => {
writeln!(
out,
"{:>12} {} -> {}",
"link",
entry.path.display(),
target.display()
)
}
};
match written {
Ok(()) => {}
// The reader (head, grep -m, ...) has seen enough
Err(err) if err.kind() == std::io::ErrorKind::BrokenPipe => return Ok(()),
Err(err) => return Err(err.into()),
}
}
Ok(())
}
fn snapshots(args: RepoArgs, config: &ConfigFile) -> Result<()> {
let (repository, _) = open_repository(args, config)?;
let mut snapshots = Vec::new();
for id in repository.snapshots()? {
snapshots.push(repository.load_snapshot(id)?);
}
snapshots.sort_by_key(|snapshot| snapshot.created);
if snapshots.is_empty() {
println!("no snapshots");
return Ok(());
}
for snapshot in snapshots {
let created = std::time::UNIX_EPOCH + std::time::Duration::from_secs(snapshot.created);
let coverage = match snapshot.coverage {
repository::Coverage::Complete => "",
repository::Coverage::Partial => " (partial: a backup that had not finished)",
};
println!(
"{} {} {} {} entries, {}{coverage}",
snapshot.id,
humantime::format_rfc3339_seconds(created),
snapshot.root.display(),
snapshot.entries,
ui::human_bytes(snapshot.content_bytes),
);
}
Ok(())
}