//! 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(_)))
}