pipeline.rs
raw
//! Suspendable, crash-tolerant work pipelines.
//!
//! A pipeline is a set of worker thread pools connected by
//! [`PersistentQueue`]s, plus a controller that watches for completion,
//! failure, a suspend request, or a reason to stop short of the work
//! that is left. The durability rules that make the whole arrangement
//! crash-tolerant:
//!
//! - Workers hold a queue lease while processing an item and only
//! acknowledge it after every downstream effect of the item (queue
//! puts, repository writes) has happened. A crash redelivers the item.
//! - Queue acknowledgements only become durable at an explicit sync, and
//! queues are always synced downstream-first, so no durable state ever
//! claims work is done whose products are not themselves durable.
//! - Every stage is idempotent — redoing an item after a crash stores
//! the same chunks and at worst repeats a manifest entry, which
//! finalization deduplicates.
//!
//! Consequently there is no suspend protocol: suspension just stops the
//! workers and syncs. Killing the process outright is equally safe, it
//! merely redoes a little more work on resume.
pub mod backup;
pub mod restore;
use std::{
cell::Cell,
path::{Path, PathBuf},
sync::{
Mutex,
atomic::{AtomicBool, AtomicU64, Ordering},
},
time::{Duration, Instant},
};
use crossbeam_channel::Sender;
use persistent_queue::PersistentQueue;
use repository::SnapshotId;
/// How long an idle worker dozes before rechecking its queue.
const IDLE_WAIT: Duration = Duration::from_millis(10);
/// How often the controller makes queue state durable while running.
const SYNC_INTERVAL: Duration = Duration::from_secs(5);
/// How often the controller reports totals and queue depths. Far more
/// often than it syncs: this is what makes a display feel live, and it
/// costs only a few atomic loads and a channel send.
const REPORT_INTERVAL: Duration = Duration::from_millis(200);
/// How many entries finalization folds into a manifest between reports,
/// so that a long finalization still shows itself draining.
const FINALIZE_REPORT_EVERY: u64 = 512;
/// The size at which an item's own progress is worth reporting. Below
/// it, a worker is through the item before anyone could read a
/// percentage, and the reports would outnumber every other event.
pub(crate) const WORTH_TRACKING: u64 = 8 << 20;
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error(
"the state directory holds a different suspended operation (on \
{existing:?}), not this one (on {requested:?}); finish or \
discard it first"
)]
StateMismatch {
existing: PathBuf,
requested: PathBuf,
},
#[error("a pipeline worker thread panicked")]
WorkerPanicked,
#[error(
"the partial snapshot this backup published ({snapshot}) has been \
pruned, so what it had already recorded can no longer be found in \
the repository. Remove the state directory ({state_dir:?}) and run \
the backup again."
)]
RecordedWorkPruned {
snapshot: SnapshotId,
state_dir: PathBuf,
},
#[error(transparent)]
Pattern(#[from] crate::patterns::PatternError),
#[error(transparent)]
Repository(#[from] repository::Error),
#[error(transparent)]
Queue(#[from] persistent_queue::Error),
#[error(transparent)]
CiboriumEncode(#[from] ciborium::ser::Error<std::io::Error>),
#[error(transparent)]
CiboriumDecode(#[from] ciborium::de::Error<std::io::Error>),
#[error(transparent)]
IO(#[from] std::io::Error),
}
/// How a pipeline run ended.
#[derive(Debug)]
pub enum Outcome {
/// All work finished; for a backup, this snapshot was stored.
Completed(SnapshotId),
/// Suspension was requested and all state is on disk; running the
/// same pipeline again will pick up where it left off.
///
/// A suspended backup also publishes what it has stored so far as a
/// partial snapshot, named here: the interrupted backup can be
/// restored from, not merely resumed on this machine.
Suspended { restorable: Option<SnapshotId> },
/// The run stopped short of the whole tree deliberately, and
/// published what it had.
///
/// Unlike a suspension this is an ending, and the snapshot is a
/// finished one: the remaining work is dropped and the working
/// state removed, so running the same command again starts a fresh
/// backup. `snapshot` is `None` when the run had recorded nothing
/// at all — a snapshot of nothing would assert that an unvisited
/// tree is empty.
FinishedEarly {
snapshot: Option<SnapshotId>,
reason: EarlyFinish,
},
}
/// Why a run stopped short of the work it had left.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum EarlyFinish {
/// The run was asked, on the command line, to publish what it had
/// and stop.
Requested,
/// The store has less room left than the backup was told to leave
/// free.
StoreNearlyFull {
/// Bytes the store reported free when the floor was crossed.
free: u64,
/// Bytes the run was told to leave free.
floor: u64,
},
/// The repository has grown to the size the backup was told to keep
/// it under.
StoreAtMaximum {
/// Bytes the repository held when it reached the limit.
used: u64,
/// Bytes the run was told to keep it under.
limit: u64,
},
}
/// A limit on what a backup may do to its store. Named on its own so
/// that a run can say which of them it cannot apply.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StoreLimit {
/// Room to leave free ([`backup::BackupOptions::minimum_free_space`]).
MinimumFreeSpace,
/// A size for the repository not to exceed
/// ([`backup::BackupOptions::maximum_store_size`]).
MaximumStoreSize,
}
/// Progress reports emitted by pipeline workers.
///
/// These flow over a channel so a UI (or a test) can watch a run without
/// the pipeline knowing how they are displayed. Skips are reports, not
/// errors: a backup keeps going when individual files are excluded,
/// vanish mid-run, or cannot be read.
///
/// Between them the variants answer three questions a display needs to
/// ask: what has been finished ([`Event::Scanned`] and friends, plus the
/// [`Event::Totals`] samples), what is being worked on right now
/// ([`Event::Began`] / [`Event::Idle`], against the pool sizes from
/// [`Event::Staffing`]), and what is still to come
/// ([`Event::Backlog`]).
#[derive(Debug)]
pub enum Event {
/// A directory's immediate children have been catalogued.
Scanned { path: PathBuf },
/// A file's content is fully in the repository.
Stored { path: PathBuf, bytes: u64 },
/// A filesystem object was deliberately not backed up.
Skipped { path: PathBuf, reason: SkipReason },
/// All content is stored; the manifest and snapshot are being built.
Finalizing,
/// A run is storing what it has backed up so far, so that an
/// interrupted backup can be restored from. Can take a while: it
/// writes a manifest covering every object recorded since the last
/// time it did this.
Checkpointing,
/// A checkpoint is done and the run is back to ordinary work. Sent
/// only by the checkpoints a run takes as it goes; the one it takes
/// on the way out is the last thing it does.
Working,
/// The run is stopping short of its remaining work on purpose, and
/// will publish what it has. Sent once, as the decision is taken.
FinishingEarly(EarlyFinish),
/// A limit the run was given is not being applied, because its
/// store will not say what holding it to that limit would need to
/// know. Sent as the run starts, once per limit.
LimitNotApplied(StoreLimit),
/// A file was written during restore.
Restored { path: PathBuf, bytes: u64 },
/// Emitted once, at the start of a resumed run: what the previous
/// session(s) of this operation accomplished, so progress reporting
/// continues where it left off instead of restarting from zero.
Resumed(Session),
/// The run's totals, sampled at a steady cadence and once more when
/// the work stops. The pipeline maintains these counts anyway, for
/// its own durable state; reporting them instead of leaving each
/// display to re-derive them keeps a single set of numbers.
Totals(Session),
/// How many worker threads a stage runs, emitted once per staffed
/// stage as the run starts. Stages that never report staffing are
/// worked by finalization rather than by a pool of their own.
Staffing { stage: Stage, threads: usize },
/// A worker took an item and is working on it now.
Began {
stage: Stage,
worker: usize,
path: PathBuf,
},
/// How far a worker has got through the item it is holding, for the
/// stages and items where that is knowable: reading a file's
/// content, or writing one back. Reported only for items big enough
/// to be worth watching.
Progressed {
stage: Stage,
worker: usize,
done: u64,
total: u64,
},
/// A worker has nothing in its hands, having run out of work or
/// stopped altogether.
Idle { stage: Stage, worker: usize },
/// How much work is waiting at the listed stages. A sample may
/// mention only some of them; the rest keep the depth they last
/// reported.
Backlog(Vec<StageBacklog>),
/// Bytes just moved between the repository and this machine: sealed
/// chunks written to the backend, or chunk content read back from
/// it. Reported as it flows rather than when whole files land, so a
/// throughput figure means something while a single huge file is
/// being worked on — and stays near zero when a re-backup finds
/// everything already stored.
Transferred { bytes: u64 },
}
/// A place in a pipeline where work waits, and — for all but the
/// finalization stages — is picked up by a pool of workers. Declared in
/// the order work flows through them, which is the order displays show.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Stage {
/// Backup: cataloguing directories.
Scan,
/// Backup: reading files, splitting them, and sealing the pieces.
Chunk,
/// Backup: sending sealed chunks to the repository.
Upload,
/// Backup: manifest entries held until finalization folds them into
/// the snapshot. Has no workers of its own.
Record,
/// Restore: recreating files from the repository.
Write,
/// Restore: directories and symlinks held until finalization applies
/// them. Has no workers of its own.
Apply,
}
impl Stage {
/// A short name for the stage, at most six characters so that
/// columns of them line up.
pub fn label(self) -> &'static str {
match self {
Stage::Scan => "scan",
Stage::Chunk => "chunk",
Stage::Upload => "upload",
Stage::Record => "record",
Stage::Write => "write",
Stage::Apply => "apply",
}
}
}
/// How much work is waiting at one stage: queued items that no worker
/// has picked up yet.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StageBacklog {
pub stage: Stage,
pub waiting: u64,
/// The first few of those items by name, in the order they will be
/// taken up, or `None` where the stage's backlog is not a queue of
/// paths still to be visited — the tree a backup has yet to walk
/// into, or the entries a restore has yet to put back.
pub upcoming: Option<Vec<PathBuf>>,
}
/// How many queued paths a stage names in a report. More than a pane is
/// likely to have room for, and few enough that reading them costs
/// nothing beside the work a stage does between two reports.
const UPCOMING: usize = 32;
/// The paths at the head of a queue, for a display to say what is
/// coming.
///
/// Nothing is delivered or consumed; a concurrent worker may take any of
/// them the moment this returns, which is the nature of naming what is
/// about to happen. A read that fails costs the display a few names,
/// which is not something to fail a run over.
pub(crate) fn peek_paths<T, F>(queue: &PersistentQueue<T>, path: F) -> Vec<PathBuf>
where
T: serde::Serialize + for<'a> serde::Deserialize<'a>,
F: Fn(T) -> PathBuf,
{
queue
.peek(UPCOMING)
.map(|items| items.into_iter().map(path).collect())
.unwrap_or_default()
}
/// Accumulated totals for one logical operation, persisted in the state
/// directory across suspensions.
///
/// After a graceful suspend the resumed totals continue exactly; after
/// a crash they may slightly overcount, since work whose
/// acknowledgement never became durable is redone and recounted — the
/// same bounded replay the pipeline's durability model allows
/// everywhere.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct Progress {
pub scanned: u64,
pub stored_files: u64,
pub stored_bytes: u64,
pub restored_files: u64,
pub restored_bytes: u64,
pub skipped: u64,
}
/// What a run has accomplished so far: its totals, and the time it has
/// spent working. Written to the state directory at every sync, read
/// back when a suspended run resumes, and reported to displays as
/// [`Event::Totals`].
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Session {
/// Deliberately carries no serde default: its presence is what
/// tells this file apart from the bare totals beeping 0.2 wrote,
/// which [`load_progress`] migrates instead of reading as zeros.
pub progress: Progress,
/// Time spent running, summed over every session of this operation.
/// Wall-clock time since the run first started would also count the
/// days it sat suspended, which says nothing about the work.
#[serde(default)]
pub elapsed: Duration,
}
#[derive(Debug)]
pub enum SkipReason {
/// Matched an exclude pattern.
Excluded,
/// A directory on a different filesystem, with mount-point traversal
/// not enabled.
MountPoint,
/// Disappeared between being discovered and being processed.
Vanished,
/// Could not be read; the text describes the error.
Unreadable(String),
/// Sockets, FIFOs, and device nodes are not backed up.
SpecialFile,
/// A manifest entry whose path would escape the restore target
/// (absolute, or containing `..`); never produced by our own backups.
UnsafePath,
/// Beeping's own working data — this run's state directory, other
/// runs' state, or a local repository. Transient or
/// self-referential, and never worth capturing: backing a local
/// repository up into itself would even compound geometrically,
/// since ciphertext never deduplicates.
Internal,
}
/// Shared switchboard between the controller and the workers.
pub(crate) struct Control {
/// Set by the outside world (UI, signal handler) to request
/// suspension.
suspend: AtomicBool,
/// Set when the run is to stop short of its remaining work and
/// publish what it has; the reason is kept beside it.
early: AtomicBool,
early_reason: Mutex<Option<EarlyFinish>>,
/// Set by the controller once it has observed that no work remains.
done: AtomicBool,
/// Set when any worker hits a fatal error; the first error is kept.
failed: AtomicBool,
failure: Mutex<Option<Error>>,
}
impl Control {
pub(crate) fn new() -> Control {
Control {
suspend: AtomicBool::new(false),
early: AtomicBool::new(false),
early_reason: Mutex::new(None),
done: AtomicBool::new(false),
failed: AtomicBool::new(false),
failure: Mutex::new(None),
}
}
pub(crate) fn request_suspend(&self) {
self.suspend.store(true, Ordering::Release);
}
pub(crate) fn suspend_requested(&self) -> bool {
self.suspend.load(Ordering::Acquire)
}
/// Asks the run to stop taking up work and publish what it has.
/// Like a suspension as far as the workers are concerned; what
/// differs is what the run does once they have stopped.
///
/// Returns whether this call is the one that decided the ending, so
/// that the reason is announced exactly once.
pub(crate) fn request_early_finish(&self, reason: EarlyFinish) -> bool {
let mut slot = self.early_reason.lock().unwrap();
let first = slot.is_none();
if first {
*slot = Some(reason);
}
self.early.store(true, Ordering::Release);
first
}
pub(crate) fn early_finish(&self) -> Option<EarlyFinish> {
if !self.early.load(Ordering::Acquire) {
return None;
}
*self.early_reason.lock().unwrap()
}
/// True once the workers are to stop taking up work, for any
/// reason: a suspension, an early finish, or a failure.
pub(crate) fn stopping(&self) -> bool {
self.suspend.load(Ordering::Acquire)
|| self.early.load(Ordering::Acquire)
|| self.failed.load(Ordering::Acquire)
}
pub(crate) fn finish(&self) {
self.done.store(true, Ordering::Release);
}
pub(crate) fn finished(&self) -> bool {
self.done.load(Ordering::Acquire)
}
pub(crate) fn fail(&self, error: Error) {
let mut slot = self.failure.lock().unwrap();
if slot.is_none() {
*slot = Some(error);
}
self.failed.store(true, Ordering::Release);
}
pub(crate) fn take_failure(&self) -> Option<Error> {
if !self.failed.load(Ordering::Acquire) {
return None;
}
self.failure.lock().unwrap().take()
}
}
/// Counts items that exist in, or are on their way into, the pipeline's
/// working queues.
///
/// Workers increment it *before* putting an item into a queue and
/// decrement it *after* acknowledging one, so it reaching zero means no
/// item exists anywhere — queued, leased, or in a worker's hands — and
/// none can appear. That single atomic makes completion detection
/// race-free where checking the queues one after another would not be.
pub(crate) struct WorkTracker(AtomicU64);
impl WorkTracker {
pub(crate) fn new(initial: u64) -> WorkTracker {
WorkTracker(AtomicU64::new(initial))
}
pub(crate) fn add(&self, count: u64) {
self.0.fetch_add(count, Ordering::AcqRel);
}
pub(crate) fn finish_one(&self) {
self.0.fetch_sub(1, Ordering::AcqRel);
}
pub(crate) fn is_idle(&self) -> bool {
self.0.load(Ordering::Acquire) == 0
}
}
/// Thread-safe accumulation of [`Session`], updated as events are
/// emitted, reported to displays at a steady cadence, and snapshotted
/// for persistence at every sync.
pub(crate) struct ProgressCounters {
scanned: AtomicU64,
stored_files: AtomicU64,
stored_bytes: AtomicU64,
restored_files: AtomicU64,
restored_bytes: AtomicU64,
skipped: AtomicU64,
/// Running time accumulated by earlier sessions of this operation,
/// which this session's own running time adds to.
prior_elapsed: Duration,
started: Instant,
}
impl ProgressCounters {
pub(crate) fn new(initial: Session) -> ProgressCounters {
ProgressCounters {
scanned: AtomicU64::new(initial.progress.scanned),
stored_files: AtomicU64::new(initial.progress.stored_files),
stored_bytes: AtomicU64::new(initial.progress.stored_bytes),
restored_files: AtomicU64::new(initial.progress.restored_files),
restored_bytes: AtomicU64::new(initial.progress.restored_bytes),
skipped: AtomicU64::new(initial.progress.skipped),
prior_elapsed: initial.elapsed,
started: Instant::now(),
}
}
pub(crate) fn record(&self, event: &Event) {
match event {
Event::Scanned { .. } => {
self.scanned.fetch_add(1, Ordering::AcqRel);
}
Event::Stored { bytes, .. } => {
self.stored_files.fetch_add(1, Ordering::AcqRel);
self.stored_bytes.fetch_add(*bytes, Ordering::AcqRel);
}
Event::Restored { bytes, .. } => {
self.restored_files.fetch_add(1, Ordering::AcqRel);
self.restored_bytes.fetch_add(*bytes, Ordering::AcqRel);
}
Event::Skipped { .. } => {
self.skipped.fetch_add(1, Ordering::AcqRel);
}
// Reports about the run itself, not about work done
Event::Checkpointing
| Event::Working
| Event::FinishingEarly(_)
| Event::LimitNotApplied(_)
| Event::Finalizing
| Event::Resumed(_)
| Event::Totals(_)
| Event::Staffing { .. }
| Event::Began { .. }
| Event::Progressed { .. }
| Event::Idle { .. }
| Event::Backlog(_)
| Event::Transferred { .. } => {}
}
}
pub(crate) fn snapshot(&self) -> Session {
Session {
progress: Progress {
scanned: self.scanned.load(Ordering::Acquire),
stored_files: self.stored_files.load(Ordering::Acquire),
stored_bytes: self.stored_bytes.load(Ordering::Acquire),
restored_files: self.restored_files.load(Ordering::Acquire),
restored_bytes: self.restored_bytes.load(Ordering::Acquire),
skipped: self.skipped.load(Ordering::Acquire),
},
elapsed: self.prior_elapsed + self.started.elapsed(),
}
}
}
pub(crate) const PROGRESS_FILE: &str = "progress";
/// Reads the progress file, accepting both the current shape and the
/// bare totals written by beeping 0.2, which had no notion of
/// accumulated running time.
///
/// Telling the two apart matters more than it looks: a reader that
/// tolerates unknown fields decodes a 0.2 file into a *zeroed* session
/// perfectly happily, silently throwing away the history of a run that
/// may have been going for weeks. [`Session::progress`] has no serde
/// default so that such a file fails cleanly and lands here instead.
/// Fields that later versions may add are still ignored, so a file from
/// a newer beeping keeps its totals rather than falling into the same
/// trap in the other direction.
pub(crate) fn load_progress(state_dir: &Path) -> Session {
if let Some(session) = load_state_file::<Session>(state_dir, PROGRESS_FILE) {
return session;
}
match load_state_file::<Progress>(state_dir, PROGRESS_FILE) {
Some(progress) => Session {
progress,
elapsed: Duration::ZERO,
},
None => Session::default(),
}
}
/// A path as a persistent-queue item.
///
/// Queue items pass through CBOR, and serde's plain `PathBuf` encoding
/// rejects non-UTF-8 names; this wrapper routes them through
/// [`repository::os_path`] so every real-world file name survives.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct QueuedPath(#[serde(with = "repository::os_path")] pub(crate) PathBuf);
/// One worker's voice in the event stream: it announces the item it is
/// working on, and the moments it is working on nothing.
///
/// "Working on" is narrower than "holding a lease": a chunker waiting
/// its turn at a large file has taken the item but not started it, and
/// says nothing until it may begin. What a stage announces is therefore
/// what a display can truthfully name.
///
/// Idleness is reported on the transition only. A worker with no work
/// rechecks its queue every [`IDLE_WAIT`]; saying so every time would
/// drown out everything else for no added information.
pub(crate) struct Reporter<'a> {
events: &'a Sender<Event>,
stage: Stage,
worker: usize,
/// Whether this worker has announced an item it has not finished.
/// A cell because the stage announces its own work through a shared
/// reference, while the worker loop reports idleness through the
/// same reporter.
busy: Cell<bool>,
}
impl<'a> Reporter<'a> {
pub(crate) fn new(events: &'a Sender<Event>, stage: Stage, worker: usize) -> Reporter<'a> {
Reporter {
events,
stage,
worker,
busy: Cell::new(false),
}
}
pub(crate) fn began(&self, path: &Path) {
self.busy.set(true);
send_event(
self.events,
Event::Began {
stage: self.stage,
worker: self.worker,
path: path.to_owned(),
},
);
}
/// Says how far through the current item this worker has got.
/// Silent for small items, whose percentages would be stale before
/// they could be read.
pub(crate) fn progressed(&self, done: u64, total: u64) {
if total < WORTH_TRACKING {
return;
}
send_event(
self.events,
Event::Progressed {
stage: self.stage,
worker: self.worker,
done,
total,
},
);
}
pub(crate) fn idle(&self) {
if self.busy.replace(false) {
send_event(
self.events,
Event::Idle {
stage: self.stage,
worker: self.worker,
},
);
}
}
}
/// Announces how many workers a stage is about to run, so a display can
/// show "3/8 busy" from its very first frame.
pub(crate) fn report_staffing(events: &Sender<Event>, stage: Stage, threads: usize) {
send_event(events, Event::Staffing { stage, threads });
}
/// The standard worker loop: keep taking leases and processing them
/// until the run begins stopping — suspension, an early finish, or a
/// failure — or the controller declares the pipeline finished.
///
/// `process` receives the lease and must acknowledge it (after all its
/// downstream effects) before returning; returning an error marks the
/// whole pipeline as failed. It also receives the worker's reporter, and
/// is the one to announce the item: a worker holding a lease is not
/// necessarily working on it — it may be waiting for its turn — and
/// only the stage itself knows the difference.
pub(crate) fn worker_loop<T, F>(
queue: &PersistentQueue<T>,
control: &Control,
reporter: Reporter<'_>,
mut process: F,
) where
T: serde::Serialize + for<'a> serde::Deserialize<'a>,
F: FnMut(persistent_queue::Lease<'_, T>, &Reporter<'_>) -> Result<(), Error>,
{
loop {
if control.stopping() {
reporter.idle();
return;
}
match queue.get() {
Ok(Some(lease)) => {
if let Err(error) = process(lease, &reporter) {
control.fail(error);
reporter.idle();
return;
}
// The item is done. Saying so matters as much as
// announcing the start: a worker between items — or
// holding one it has not been allowed to begin yet —
// is working on nothing, and must not go on naming the
// last thing it did.
reporter.idle();
}
Ok(None) => {
reporter.idle();
if control.finished() {
return;
}
std::thread::sleep(IDLE_WAIT);
}
Err(error) => {
control.fail(error.into());
reporter.idle();
return;
}
}
}
}
/// The durable record of an in-progress or just-finished pipeline run,
/// kept in the state directory alongside the queues.
///
/// Its presence is the commit point of initialization: state files
/// without a marker are debris from a run that died while starting up,
/// and are wiped and rebuilt. Carrying the run's identity, it also
/// refuses to resume state that belongs to a different operation.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) enum Marker {
BackupRunning {
#[serde(with = "repository::os_path")]
root: PathBuf,
},
RestoreRunning {
#[serde(with = "repository::os_path")]
target: PathBuf,
snapshot: SnapshotId,
/// The (sorted) cherry-pick patterns this restore was started
/// with; resuming with a different selection is refused.
#[serde(default)]
selection: Vec<String>,
},
Completed {
snapshot: SnapshotId,
/// Why the run stopped short of the whole tree, when it did.
/// Absent from the markers of runs that covered everything, and
/// from those written before a run could finish early.
#[serde(default)]
finished_early: Option<EarlyFinish>,
},
}
impl Marker {
pub(crate) fn load(state_dir: &Path) -> Result<Option<Marker>, Error> {
let contents = match std::fs::read(state_dir.join("run.marker")) {
Ok(contents) => contents,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err.into()),
};
Ok(Some(ciborium::from_reader(contents.as_slice())?))
}
pub(crate) fn store(&self, state_dir: &Path) -> Result<(), Error> {
store_state_file(state_dir, "run.marker", self)
}
}
/// Atomically replaces a small CBOR file in the state directory:
/// written to a temporary file, synced, and renamed into place.
pub(crate) fn store_state_file(
state_dir: &Path,
name: &str,
value: &impl serde::Serialize,
) -> Result<(), Error> {
use std::io::Write as _;
let mut temp = tempfile::NamedTempFile::new_in(state_dir)?;
ciborium::into_writer(value, &mut temp)?;
temp.flush()?;
temp.as_file().sync_all()?;
temp.persist(state_dir.join(name))
.map_err(|err| err.error)?;
std::fs::File::open(state_dir)?.sync_all()?;
Ok(())
}
/// Reads a small CBOR state file the run cannot proceed correctly
/// without, telling "not written yet" apart from "written but
/// unreadable".
///
/// The tolerant [`load_state_file`] is for informational data only.
/// Reading load-bearing state that way would silently continue with a
/// default — which for something like the list of manifest segments
/// already stored means finishing a backup whose snapshot omits
/// everything the earlier sessions did.
pub(crate) fn read_state_file<T: serde::de::DeserializeOwned>(
state_dir: &Path,
name: &str,
) -> Result<Option<T>, Error> {
let contents = match std::fs::read(state_dir.join(name)) {
Ok(contents) => contents,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err.into()),
};
Ok(Some(ciborium::from_reader(contents.as_slice())?))
}
/// Reads a small CBOR state file, treating anything unreadable as
/// absent — for informational data (like progress totals) that must
/// never block a resume.
pub(crate) fn load_state_file<T: serde::de::DeserializeOwned>(
state_dir: &Path,
name: &str,
) -> Option<T> {
let contents = std::fs::read(state_dir.join(name)).ok()?;
ciborium::from_reader(contents.as_slice()).ok()
}
/// Sends an event, ignoring a disconnected receiver: progress reporting
/// must never stop the actual work.
pub(crate) fn send_event(events: &Sender<Event>, event: Event) {
let _ = events.send(event);
}
/// How often the controller polls for completion, failure, or suspension.
const CONTROL_TICK: Duration = Duration::from_millis(50);
/// Watches a running pipeline until it completes (`is_idle` returns
/// true), a worker fails, or it is asked to stop — reporting what it
/// sees, syncing its durable state, and running the pipeline's own
/// periodic work, each at its own cadence along the way. On completion
/// the control is marked finished, which is what releases idle workers.
///
/// `tick` is whatever the pipeline itself has to do as it runs — a
/// backup checkpoints and watches the store's free space — and decides
/// for itself how often each of those is due. A pipeline with nothing
/// to do as it goes passes a closure that does nothing.
pub(crate) fn control_loop(
control: &Control,
suspend: &std::sync::atomic::AtomicBool,
is_idle: impl Fn() -> bool,
report: impl Fn(),
sync: impl Fn() -> Result<(), Error>,
mut tick: impl FnMut() -> Result<(), Error>,
) {
let mut last_sync = Instant::now();
let mut last_report = Instant::now();
report();
loop {
std::thread::sleep(CONTROL_TICK);
if suspend.load(Ordering::Acquire) {
control.request_suspend();
return;
}
if control.stopping() {
// Stopping for a reason of the run's own: a worker failed
// (its error is already recorded), or the run is finishing
// early. Either way the workers are on their way out.
return;
}
if is_idle() {
control.finish();
return;
}
if last_report.elapsed() >= REPORT_INTERVAL {
report();
last_report = Instant::now();
}
if last_sync.elapsed() >= SYNC_INTERVAL {
if let Err(error) = sync() {
control.fail(error);
return;
}
last_sync = Instant::now();
}
// Offered every tick rather than on the sync cadence: how much
// work may sit unpublished is the pipeline's guarantee to make,
// not something to round up to the next sync. It runs on this
// thread, the one that syncs, which is the whole of the mutual
// exclusion a checkpoint needs — its entry acknowledgements
// must not reach the disk before the manifest covering them,
// and nothing else here can sync that queue underneath it.
if let Err(error) = tick() {
control.fail(error);
return;
}
}
}
/// True if joining a restore target with this path stays inside the
/// target: relative, with only normal components. Backup never records
/// anything else, so a violation means foreign or damaged data.
pub(crate) fn path_is_safe(path: &Path) -> bool {
!path.as_os_str().is_empty()
&& path
.components()
.all(|component| matches!(component, std::path::Component::Normal(_)))
}