restore.rs
raw
//! The restore pipeline: a snapshot's manifest is seeded into a durable
//! queue, writer threads recreate the files, and a finalization pass
//! applies directories, symlinks, and directory metadata.
//!
//! ```text
//! seed: manifest entries ──┬── files ────────> restore queue ──> writers
//! └── dirs/symlinks ─> meta queue ─┐
//! v
//! finalization: symlinks, empty dirs, then directory
//! modes and mtimes deepest-first (after the contents)
//! ```
//!
//! The same durability rules as the backup pipeline apply, with one
//! restore-specific addition: a restored file's bytes are synced before
//! its queue acknowledgement can become durable — data first, then the
//! claim that it was written — and the parent directory entries are
//! synced at each queue sync for the same reason. Directories and
//! symlinks are not tracked per-item at all; the whole finalization pass
//! redoes itself if interrupted, which makes their durability moot.
//!
//! Restoring is idempotent from a fresh or partially-restored target: a
//! redelivered file entry is simply rewritten. Restoring over unrelated
//! existing content is not supported.
use std::{
collections::HashSet,
path::{Path, PathBuf},
sync::{Arc, Mutex, atomic::AtomicBool},
};
use crossbeam_channel::Sender;
use persistent_queue::{Lease, PersistentQueue};
use repository::{Entry, EntryKind, Repository, SnapshotId};
use super::{
Control, Error, Event, FINALIZE_REPORT_EVERY, Marker, Outcome, PROGRESS_FILE, ProgressCounters,
Reporter, SkipReason, Stage, StageBacklog, control_loop, load_progress, path_is_safe,
peek_paths, report_staffing, send_event, store_state_file, worker_loop,
};
const RESTORE_QUEUE: &str = "restore.queue";
const META_QUEUE: &str = "meta.queue";
pub struct RestoreOptions {
/// The snapshot to restore.
pub snapshot: SnapshotId,
/// Cherry-pick patterns (see [`crate::patterns`]): only entries
/// matching one of these — or living under a matching directory —
/// are restored, along with the directory entries above them so
/// their metadata comes out right. Empty means the whole snapshot.
pub select: Vec<String>,
/// The directory to restore into; created if absent, expected to be
/// fresh (existing files at manifest paths are overwritten).
pub target: PathBuf,
/// Directory for the pipeline's durable working state; exclusively
/// owned by the pipeline, like the backup state directory.
pub state_dir: PathBuf,
pub writer_threads: usize,
}
impl RestoreOptions {
pub fn new(
snapshot: SnapshotId,
target: impl Into<PathBuf>,
state_dir: impl Into<PathBuf>,
) -> RestoreOptions {
RestoreOptions {
snapshot,
select: Vec::new(),
target: target.into(),
state_dir: state_dir.into(),
writer_threads: std::thread::available_parallelism().map_or(4, usize::from),
}
}
}
struct RestoreContext {
repository: Arc<Repository>,
target: PathBuf,
restore: PersistentQueue<Entry>,
meta: PersistentQueue<Entry>,
control: Control,
events: Sender<Event>,
/// Directories that gained files since the last sync; their entries
/// are fsynced before the queue acks are, so no durable ack can
/// claim a file that could still vanish from its directory.
touched_dirs: Mutex<std::collections::HashSet<PathBuf>>,
/// Running totals, persisted at every sync so a resumed run's
/// progress reporting continues from previous sessions.
progress: ProgressCounters,
}
impl RestoreContext {
fn sync_state(&self) -> Result<(), Error> {
let touched: Vec<PathBuf> = {
let mut guard = self.touched_dirs.lock().unwrap();
guard.drain().collect()
};
for dir in touched {
std::fs::File::open(&dir)?.sync_all()?;
}
self.meta.sync()?;
self.restore.sync()?;
Ok(())
}
/// Reports an event, folding it into the persistent progress totals.
fn emit(&self, event: Event) {
self.progress.record(&event);
send_event(&self.events, event);
}
/// Reports the run's totals and what is queued at each stage — the
/// state a display needs that no single work item announces.
fn report_status(&self) {
send_event(&self.events, Event::Totals(self.progress.snapshot()));
send_event(
&self.events,
Event::Backlog(vec![
StageBacklog {
stage: Stage::Write,
waiting: self.restore.pending() as u64,
upcoming: Some(peek_paths(&self.restore, |entry| entry.path)),
},
StageBacklog {
stage: Stage::Apply,
waiting: self.meta.pending() as u64,
upcoming: Some(peek_paths(&self.meta, |entry| entry.path)),
},
]),
);
}
fn skip(&self, path: &Path, reason: SkipReason) {
self.emit(Event::Skipped {
path: path.to_owned(),
reason,
});
}
}
/// Decides which manifest entries a cherry-picked restore includes.
struct Selection(Option<globset::GlobSet>);
impl Selection {
fn new(patterns: &[String]) -> Result<Selection, Error> {
if patterns.is_empty() {
return Ok(Selection(None));
}
Ok(Selection(Some(crate::patterns::compile_set(patterns)?)))
}
/// True if the entry itself matches, or an ancestor does — which is
/// what makes selecting a directory bring its whole subtree.
fn includes(&self, path: &Path) -> bool {
let Some(set) = &self.0 else {
return true;
};
crate::patterns::matches_with_ancestors(set, path)
}
}
/// Marks every proper ancestor of `path` as a directory the restore
/// needs, so selected entries get their parent directories' metadata.
fn require_ancestors(path: &Path, needed: &mut HashSet<PathBuf>) {
for ancestor in path.ancestors().skip(1) {
if !ancestor.as_os_str().is_empty() {
needed.insert(ancestor.to_owned());
}
}
}
/// Runs a restore to completion or suspension, mirroring
/// [`backup::run`](super::backup::run)'s resume and suspend behavior.
pub fn run(
repository: Arc<Repository>,
options: RestoreOptions,
events: Sender<Event>,
suspend: Arc<AtomicBool>,
) -> Result<Outcome, Error> {
let state_dir = options.state_dir.clone();
// Selection identity is order-insensitive
let mut selection_id = options.select.clone();
selection_id.sort();
std::fs::create_dir_all(&options.target)?;
std::fs::create_dir_all(&state_dir)?;
let target = options.target.canonicalize()?;
let fresh = match Marker::load(&state_dir)? {
Some(Marker::Completed { snapshot, .. }) => {
std::fs::remove_dir_all(&state_dir)?;
return Ok(Outcome::Completed(snapshot));
}
Some(Marker::RestoreRunning {
target: existing,
snapshot,
selection,
}) => {
if existing != target || snapshot != options.snapshot || selection != selection_id {
return Err(Error::StateMismatch {
existing,
requested: target,
});
}
false
}
Some(Marker::BackupRunning { root }) => {
return Err(Error::StateMismatch {
existing: root,
requested: target,
});
}
None => {
std::fs::remove_dir_all(&state_dir)?;
std::fs::create_dir_all(&state_dir)?;
true
}
};
let restore: PersistentQueue<Entry> = PersistentQueue::open(state_dir.join(RESTORE_QUEUE))?;
let meta: PersistentQueue<Entry> = PersistentQueue::open(state_dir.join(META_QUEUE))?;
if fresh {
// Seed both queues from the manifest, make them durable, then
// commit the initialization by writing the marker. Directory
// entries are buffered and decided last: a directory belongs in
// the restore if it is selected itself or if anything selected
// lives beneath it.
let selection = Selection::new(&selection_id)?;
let snapshot = repository.load_snapshot(options.snapshot)?;
let mut directories = Vec::new();
let mut needed: HashSet<PathBuf> = HashSet::new();
for entry in repository.manifest_entries(snapshot.manifest) {
let entry = entry?;
match entry.kind {
EntryKind::Directory => directories.push(entry),
EntryKind::File { .. } => {
if selection.includes(&entry.path) {
require_ancestors(&entry.path, &mut needed);
restore.put(&entry)?;
}
}
EntryKind::Symlink { .. } => {
if selection.includes(&entry.path) {
require_ancestors(&entry.path, &mut needed);
meta.put(&entry)?;
}
}
}
}
for directory in &directories {
if selection.includes(&directory.path) {
require_ancestors(&directory.path, &mut needed);
}
}
for directory in directories {
if needed.contains(&directory.path) || selection.includes(&directory.path) {
meta.put(&directory)?;
}
}
meta.sync()?;
restore.sync()?;
Marker::RestoreRunning {
target: target.clone(),
snapshot: options.snapshot,
selection: selection_id,
}
.store(&state_dir)?;
}
let prior = load_progress(&state_dir);
if !fresh {
send_event(&events, Event::Resumed(prior));
}
let context = Arc::new(RestoreContext {
repository,
target,
restore,
meta,
control: Control::new(),
events,
touched_dirs: Mutex::new(std::collections::HashSet::new()),
progress: ProgressCounters::new(prior),
});
let mut workers = Vec::new();
let writer_threads = options.writer_threads.max(1);
report_staffing(&context.events, Stage::Write, writer_threads);
for index in 0..writer_threads {
let context = context.clone();
workers.push(
std::thread::Builder::new()
.name(format!("writer-{index}"))
.spawn(move || {
worker_loop(
&context.restore,
&context.control,
Reporter::new(&context.events, Stage::Write, index),
|lease, reporter| write_one(&context, lease, reporter),
)
})?,
);
}
let sync_all = || {
store_state_file(&state_dir, PROGRESS_FILE, &context.progress.snapshot())?;
context.sync_state()
};
control_loop(
&context.control,
&suspend,
|| context.restore.is_empty(),
|| context.report_status(),
sync_all,
// A restore publishes nothing as it goes: what it produces is
// the tree itself, and the queues are what make it resumable
|| Ok(()),
);
for worker in workers {
if worker.join().is_err() {
context.control.fail(Error::WorkerPanicked);
}
}
sync_all()?;
// The writers have stopped, so this sample is the exact final word
// on what they accomplished
context.report_status();
if let Some(error) = context.control.take_failure() {
return Err(error);
}
if !context.control.finished() {
// A restore publishes nothing: what it has written is simply
// there, in the target directory
return Ok(Outcome::Suspended { restorable: None });
}
finalize(&context)?;
Marker::Completed {
snapshot: options.snapshot,
// A restore has no tree of its own to fall short of
finished_early: None,
}
.store(&state_dir)?;
context.meta.sync()?;
drop(context);
std::fs::remove_dir_all(&state_dir)?;
Ok(Outcome::Completed(options.snapshot))
}
/// Writer stage: recreate one file from its chunks.
///
/// Failures reading the *repository* are fatal — a restore that cannot
/// produce recorded data must not quietly succeed. Only entries that
/// could never have been produced by a backup are skipped.
fn write_one(
context: &RestoreContext,
lease: Lease<'_, Entry>,
reporter: &Reporter<'_>,
) -> Result<(), Error> {
let entry: &Entry = &lease;
reporter.began(&entry.path);
let EntryKind::File { chunks, len } = &entry.kind else {
// Seeding routes everything else to the meta queue
lease.ack();
return Ok(());
};
if !path_is_safe(&entry.path) {
context.skip(&entry.path, SkipReason::UnsafePath);
lease.ack();
return Ok(());
}
let absolute = context.target.join(&entry.path);
let parent = absolute
.parent()
.expect("a joined relative path always has a parent")
.to_owned();
std::fs::create_dir_all(&parent)?;
let mut file = std::fs::File::create(&absolute)?;
let mut written = 0u64;
{
use std::io::Write as _;
let mut writer = std::io::BufWriter::new(&mut file);
for id in chunks {
let chunk = context.repository.load_chunk(*id)?;
// What came off the backend, give or take each chunk's
// small envelope of nonce and authentication tag
send_event(
&context.events,
Event::Transferred {
bytes: chunk.len() as u64,
},
);
writer.write_all(&chunk)?;
written += chunk.len() as u64;
reporter.progressed(written, *len);
}
writer.flush()?;
}
debug_assert_eq!(written, *len);
// The file's bytes must be on stable storage before the ack that
// claims it was restored can be synced
file.sync_all()?;
apply_metadata(&file, entry)?;
context.touched_dirs.lock().unwrap().insert(parent);
context.emit(Event::Restored {
path: entry.path.clone(),
bytes: written,
});
lease.ack();
Ok(())
}
/// Applies symlinks, creates any still-missing (empty) directories, and
/// then sets directory modes and mtimes deepest-first — after all
/// content, so that read-only directory modes and accurate mtimes stick.
///
/// This pass is redone in full if interrupted: its acknowledgements only
/// become durable after the completion marker is written.
fn finalize(context: &RestoreContext) -> Result<(), Error> {
context.emit(Event::Finalizing);
let mut directories: Vec<Entry> = Vec::new();
// Finalization is the one worker this stage ever has, so it reports
// like any other: the entry it has in hand names the apply row that
// stood empty while the writers were still going.
let reporter = Reporter::new(&context.events, Stage::Apply, 0);
let mut applied = 0u64;
loop {
let lease = match context.meta.get()? {
Some(lease) => lease,
None => break,
};
let entry = lease.ack();
// A snapshot's directories and symlinks can be numerous; name
// the entry in hand and say how much of the holding queue is
// left, starting with the very first one so the row is never
// silent while finalization runs
if applied.is_multiple_of(FINALIZE_REPORT_EVERY) {
reporter.began(&entry.path);
context.report_status();
}
applied += 1;
if !path_is_safe(&entry.path) {
context.skip(&entry.path, SkipReason::UnsafePath);
continue;
}
match &entry.kind {
EntryKind::Directory => {
std::fs::create_dir_all(context.target.join(&entry.path))?;
directories.push(entry);
}
EntryKind::Symlink { target } => {
apply_symlink(context, &entry, target)?;
}
EntryKind::File { .. } => {
// Cannot happen: seeding routes files to the restore
// queue. Tolerate it anyway by leaving the file absent.
context.skip(&entry.path, SkipReason::Vanished);
}
}
}
// Children before parents, so parent mtimes and read-only modes are
// applied after the last write inside them
directories.sort_by_key(|entry| std::cmp::Reverse(entry.path.components().count()));
for (position, entry) in directories.iter().enumerate() {
// Deep trees hold a great many directories; keep naming the one
// being finished rather than going quiet for the whole pass
if position as u64 % FINALIZE_REPORT_EVERY == 0 {
reporter.began(&entry.path);
context.report_status();
}
let dir = std::fs::File::open(context.target.join(&entry.path))?;
apply_metadata(&dir, entry)?;
}
reporter.idle();
Ok(())
}
/// Recreates one symlink, replacing whatever a previous interrupted
/// finalization may have left at its path.
fn apply_symlink(context: &RestoreContext, entry: &Entry, target: &Path) -> Result<(), Error> {
let absolute = context.target.join(&entry.path);
if let Some(parent) = absolute.parent() {
std::fs::create_dir_all(parent)?;
}
match std::fs::symlink_metadata(&absolute) {
Ok(_) => std::fs::remove_file(&absolute)?,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => return Err(err.into()),
}
#[cfg(unix)]
std::os::unix::fs::symlink(target, &absolute)?;
#[cfg(not(unix))]
{
let _ = target;
context.skip(&entry.path, SkipReason::SpecialFile);
}
Ok(())
}
/// Sets the recorded permissions and modification time on an open file
/// or directory. Absent metadata (from another platform, or a pre-epoch
/// mtime) is simply not applied.
fn apply_metadata(file: &std::fs::File, entry: &Entry) -> Result<(), Error> {
#[cfg(unix)]
if let Some(mode) = entry.mode {
use std::os::unix::fs::PermissionsExt as _;
file.set_permissions(std::fs::Permissions::from_mode(mode))?;
}
if let Some(mtime) = entry.mtime
&& mtime >= 0
{
let modified = std::time::UNIX_EPOCH + std::time::Duration::from_nanos(mtime as u64);
file.set_modified(modified)?;
}
Ok(())
}