autumn-cli 0.10.0

//! Online-safe `SQLite` snapshot and restore for `autumn db backup` /
//! `autumn db restore` (issue #1909).
//!
//! `pg_dump`/`pg_restore` are Postgres-only. The `SQLite` tier (#1614) ships one
//! binary and no external tools, so a `SQLite` target uses `VACUUM INTO`. One
//! transactional statement copies the database. In WAL mode the copy does not
//! block the writer, so a backup of a live app neither corrupts nor stalls it.
//!
//! `verify` calls `autumn_web::replication::sqlite::integrity_check`, the same
//! check the in-process replication verifier uses.
//!
//! diesel's `SQLite` backend is in the CLI's dependency graph unconditionally
//! (the workspace `diesel` pin enables it), so none of this needs the
//! `--features sqlite` backend flip, exactly like `autumn db replica`.

use std::path::{Path, PathBuf};

use autumn_web::replication;
use diesel::connection::SimpleConnection as _;
use diesel::prelude::*;
use diesel::sql_types::Text;

/// How long a snapshot/restore connection waits for a lock before giving up.
///
/// Longer than the replicator's one-second tick. A backup is one-shot and an
/// operator waits for it. On a rollback-journal database a writer blocks readers,
/// so wait it out rather than fail.
const BUSY_TIMEOUT_MS: u32 = 30_000;

/// Failure modes of a `SQLite` snapshot or restore. `Display` is credential-safe:
/// a `SQLite` target carries no password, and the one variant that can be reached
/// with a foreign URL redacts it.
#[derive(Debug)]
pub enum SnapshotError {
    /// The target names no database file (an in-memory database, which no
    /// snapshot can outlive).
    NotAFile {
        /// The configured target. Redacted before display: a mis-dispatched
        /// restore can reach here with a Postgres URL, which carries a password.
        target: String,
    },
    /// The database file does not exist.
    Missing {
        /// The resolved path.
        path: String,
    },
    /// The database (or artifact) could not be opened.
    Open {
        /// The resolved path.
        path: String,
        /// diesel/`SQLite` detail.
        detail: String,
    },
    /// A statement failed.
    Query {
        /// What was being run.
        op: &'static str,
        /// diesel/`SQLite` detail.
        detail: String,
    },
    /// The artifact failed `PRAGMA integrity_check`.
    Integrity {
        /// `SQLite`'s first complaint.
        detail: String,
    },
    /// A filesystem operation failed.
    Io {
        /// What was being attempted.
        context: String,
        /// The underlying error.
        detail: String,
    },
}

impl std::fmt::Display for SnapshotError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotAFile { target } => write!(
                f,
                "the target {:?} names no SQLite database file (an in-memory database cannot \
                 be backed up, and a Postgres URL is not a SQLite target).\n  Point \
                 `database.url` at a file, e.g. sqlite:///var/lib/myapp/app.db.",
                replication::redact_credentials(target)
            ),
            Self::Missing { path } => write!(
                f,
                "the SQLite database file {path} does not exist.\n  Run `autumn migrate` (or \
                 start the app) to create it before backing it up."
            ),
            Self::Open { path, detail } => {
                write!(f, "could not open the SQLite database at {path}: {detail}")
            }
            Self::Query { op, detail } => write!(f, "{op} failed: {detail}"),
            Self::Integrity { detail } => write!(f, "SQLite integrity check failed: {detail}"),
            Self::Io { context, detail } => write!(f, "{context}: {detail}"),
        }
    }
}

impl SnapshotError {
    fn io(context: impl Into<String>) -> impl FnOnce(std::io::Error) -> Self {
        let context = context.into();
        move |e| Self::Io {
            context,
            detail: e.to_string(),
        }
    }
}

/// Resolve the database FILE a `SQLite` target names, refusing every in-memory
/// spelling.
///
/// The spelling rules come from `autumn_web::replication::database_file`, the
/// runtime's own resolver, so the CLI and the app agree on which file a URL
/// means.
///
/// # Errors
///
/// [`SnapshotError::NotAFile`] when the target is in-memory (or not a `SQLite`
/// URL at all).
pub fn database_path(target: &str) -> Result<PathBuf, SnapshotError> {
    database_target(target).map(replication::DatabaseTarget::into_path)
}

/// Resolve a `SQLite` target to its file and its open options.
///
/// # Errors
///
/// [`SnapshotError::NotAFile`] when the target is in-memory (or not a `SQLite`
/// URL at all).
pub fn database_target(target: &str) -> Result<replication::DatabaseTarget, SnapshotError> {
    replication::database_target(target).ok_or_else(|| SnapshotError::NotAFile {
        target: target.to_owned(),
    })
}

/// Open a private connection to a `SQLite` database file, such as an artifact.
///
/// Never creates the file. See [`open_target`].
///
/// # Errors
///
/// [`SnapshotError::Missing`] or [`SnapshotError::Open`].
pub fn open(path: &Path) -> Result<SqliteConnection, SnapshotError> {
    require_file(path)?;
    connect(
        &replication::no_create_uri(path).ok_or_else(|| not_unicode(path))?,
        path,
    )
}

/// Open the database a target names, with the target's own URI options.
///
/// Never creates the file. The existence check gives a clear error. The
/// no-create `mode=rw` closes the race after it: a file removed in between
/// fails the open, and `SQLite` does not create an empty one (#2589 item 7).
/// The source `vfs` stays, so the locks match the app's (item 12).
///
/// # Errors
///
/// [`SnapshotError::Missing`] or [`SnapshotError::Open`].
pub fn open_target(
    target: &replication::DatabaseTarget,
) -> Result<SqliteConnection, SnapshotError> {
    let path = target.path();
    require_file(path)?;
    connect(
        &target.no_create_uri().ok_or_else(|| not_unicode(path))?,
        path,
    )
}

fn require_file(path: &Path) -> Result<(), SnapshotError> {
    if path.exists() {
        return Ok(());
    }
    Err(SnapshotError::Missing {
        path: path.display().to_string(),
    })
}

fn not_unicode(path: &Path) -> SnapshotError {
    SnapshotError::Open {
        path: path.display().to_string(),
        detail: "the path is not valid Unicode, so no SQLite URI can name it".to_owned(),
    }
}

/// Open `uri` (a no-create URI for `path`) and set the busy timeout.
fn connect(uri: &str, path: &Path) -> Result<SqliteConnection, SnapshotError> {
    let mut conn = SqliteConnection::establish(uri).map_err(|e| SnapshotError::Open {
        path: path.display().to_string(),
        detail: e.to_string(),
    })?;
    conn.batch_execute(&format!("PRAGMA busy_timeout = {BUSY_TIMEOUT_MS};"))
        .map_err(|e| SnapshotError::Query {
            op: "busy_timeout",
            detail: e.to_string(),
        })?;
    Ok(conn)
}

/// Write an online-safe snapshot of the `SQLite` database `target` names into
/// `out_path`.
///
/// `VACUUM INTO` runs inside a read transaction, so the snapshot is a single
/// consistent point in time even while the app writes. `out_path` must not name
/// an existing database; the caller's fresh backup run directory guarantees that.
///
/// The artifact is flushed before it is reported, so retention cannot prune a
/// good older run in favour of one a power loss would have left empty.
///
/// # Errors
///
/// See [`SnapshotError`].
pub fn snapshot(target: &str, out_path: &Path) -> Result<(), SnapshotError> {
    let mut conn = open_target(&database_target(target)?)?;
    // `VACUUM INTO` reads a URI, so the output is one too. A lossy string
    // changes a non-UTF-8 name (#2589 item 15), and a relative name that
    // starts with `file:` becomes a URI (item 18).
    let out = replication::file_uri(out_path).ok_or_else(|| not_unicode(out_path))?;
    diesel::sql_query("VACUUM INTO ?")
        .bind::<Text, _>(&out)
        .execute(&mut conn)
        .map_err(|e| SnapshotError::Query {
            op: "VACUUM INTO",
            detail: e.to_string(),
        })?;
    flush(out_path)
}

/// Grade a `SQLite` artifact: it must be a non-empty file that opens and passes
/// `PRAGMA integrity_check`.
///
/// # Errors
///
/// See [`SnapshotError`].
pub fn verify(path: &Path) -> Result<(), SnapshotError> {
    let len = std::fs::metadata(path)
        .map_err(SnapshotError::io(format!("stat {}", path.display())))?
        .len();
    if len == 0 {
        return Err(SnapshotError::Integrity {
            detail: format!("{} is empty (0 bytes)", path.display()),
        });
    }
    let mut conn = open(path)?;
    replication::sqlite::integrity_check(&mut conn).map_err(|e| SnapshotError::Integrity {
        detail: e.to_string(),
    })
}

/// Every sidecar `SQLite` can leave beside a database file.
///
/// Both journal modes matter. WAL mode leaves `-wal` and `-shm`; the default
/// rollback-journal mode leaves `-journal`. A restore must clear all three: each
/// describes pages of the file it replaces, and `SQLite` replays a hot `-journal`
/// (or `-wal`) onto whatever now sits at that path.
///
/// `VACUUM INTO` writes its output in rollback-journal mode whatever the source
/// used, so a restored database really can acquire a `-journal`.
fn sidecars(db: &Path) -> [PathBuf; 3] {
    let mut journal = db.as_os_str().to_owned();
    journal.push("-journal");
    [
        replication::wal::wal_path(db),
        replication::wal::shm_path(db),
        PathBuf::from(journal),
    ]
}

/// A staging path beside `db`, unique to this process.
///
/// Unique because a fixed name lets two concurrent restores stage over each
/// other and publish a half-copied file.
fn staging_path(db: &Path) -> PathBuf {
    let mut path = db.as_os_str().to_owned();
    path.push(format!(".autumn-restore-{}.tmp", std::process::id()));
    PathBuf::from(path)
}

/// Copy `artifact` to `staged`, give it the target's mode and owner, verify it,
/// and flush it to disk.
///
/// The staged copy is verified again, not just the source: a short write or a
/// full disk must not survive as far as the rename.
fn stage(artifact: &Path, staged: &Path, db: &Path) -> Result<(), SnapshotError> {
    // `create_new` is `O_CREAT | O_EXCL`, which FAILS if anything already sits at
    // this path and never follows a symlink there. `std::fs::copy` does follow
    // one, and the staging name is predictable, so between the caller's
    // `remove_file` and this create an account that can write the database
    // directory could plant a link and have a privileged restore truncate an
    // arbitrary file it points at — before `verify` ever runs. Failing closed on
    // an occupied path closes that window; the copy then goes through this
    // handle, so it cannot be redirected afterwards either.
    let mut out = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(staged)
        .map_err(SnapshotError::io(format!(
            "creating the staging file {}",
            staged.display()
        )))?;
    let mut source = std::fs::File::open(artifact)
        .map_err(SnapshotError::io(format!("opening {}", artifact.display())))?;
    std::io::copy(&mut source, &mut out).map_err(SnapshotError::io(format!(
        "copying {} to {}",
        artifact.display(),
        staged.display()
    )))?;
    // Everything that can be done to the FILE is done through this handle, never
    // by reopening the name. Creating it with `O_CREAT | O_EXCL` only proves
    // nothing was there when we made it; the name stays writable by whoever can
    // write the directory, so a `set_permissions(path)`/`chown(path)`/reopen
    // after the handle is dropped can be redirected at a symlink planted in
    // between — and `chown` through one hands an attacker the mode and owner of
    // whatever it points at.
    inherit_target_permissions_via(db, &out, staged)?;
    out.sync_all()
        .map_err(SnapshotError::io(format!("flushing {}", staged.display())))?;

    // `verify` and the publishing rename must address the file by NAME —
    // SQLite opens a path, and `rename(2)` takes one. So confirm the name still
    // resolves to the file this handle owns, and refuse if it does not.
    same_file_as_handle(&out, staged)?;
    verify(staged)?;
    same_file_as_handle(&out, staged)
}

/// Refuse unless `path` still names the exact file `handle` owns.
///
/// `lstat` (`symlink_metadata`), so a symlink planted at the name is compared as
/// the symlink it is and cannot match. Narrows, and does not eliminate, the race
/// on the two steps that must use a name: an account able to write the database's
/// own directory could still swap the entry between this check and `SQLite`'s open
/// or the rename. Not granting write on that directory is the actual control;
/// this makes the remaining window a tight race rather than an open door.
#[cfg(unix)]
fn same_file_as_handle(handle: &std::fs::File, path: &Path) -> Result<(), SnapshotError> {
    use std::os::unix::fs::MetadataExt as _;

    let owned = handle
        .metadata()
        .map_err(SnapshotError::io("inspecting the staging file".to_owned()))?;
    let named = std::fs::symlink_metadata(path)
        .map_err(SnapshotError::io(format!("inspecting {}", path.display())))?;
    if owned.dev() == named.dev() && owned.ino() == named.ino() {
        return Ok(());
    }
    Err(SnapshotError::Io {
        context: format!("publishing the staging file {}", path.display()),
        detail: "it was replaced while the restore was preparing it, so it is no longer \
                 the verified copy — refusing to publish it"
            .to_owned(),
    })
}

/// Non-unix: there is no portable inode identity to compare, and the hazard this
/// guards — a symlink swapped in by an account that can write the database's
/// directory — is a POSIX deployment concern, as the ownership handling above
/// already is.
#[cfg(not(unix))]
#[allow(clippy::unnecessary_wraps)]
fn same_file_as_handle(_handle: &std::fs::File, _path: &Path) -> Result<(), SnapshotError> {
    Ok(())
}

/// Flush a file to disk.
///
/// Opened for WRITING, not with `File::open`: Windows refuses `FlushFileBuffers`
/// on a read-only handle with "Access is denied", where POSIX allows `fsync` on
/// a read-only descriptor.
fn flush(path: &Path) -> Result<(), SnapshotError> {
    let file = std::fs::OpenOptions::new()
        .write(true)
        .open(path)
        .map_err(SnapshotError::io(format!("opening {}", path.display())))?;
    file.sync_all()
        .map_err(SnapshotError::io(format!("flushing {}", path.display())))
}

/// Give `staged` the mode and owner of the database it replaces.
///
/// `std::fs::copy` copies the ARTIFACT's mode, and the staged file belongs to
/// whoever ran the command. Without this a `0600` database can come back
/// world-readable, and a restore run under `sudo` can leave a root-owned file the
/// service account cannot write.
///
/// Only a privileged user may change ownership, so `chown` can fail. That is
/// safe only when the old owner can still read and write the file. If not, the
/// restore stops before the rename (#2589 item 4): the service could not open
/// its own database. A missing target (a first restore onto a fresh path)
/// leaves both alone.
#[cfg(unix)]
fn inherit_target_permissions_via(
    db: &Path,
    staged: &std::fs::File,
    staged_path: &Path,
) -> Result<(), SnapshotError> {
    use std::os::unix::fs::MetadataExt as _;

    let Ok(existing) = std::fs::metadata(db) else {
        return Ok(());
    };
    // `File::set_permissions` is `fchmod` and `fchown` takes the descriptor, so
    // neither can be redirected at a symlink planted at `staged_path`. The path
    // is carried only to name the file in an error.
    staged
        .set_permissions(existing.permissions())
        .map_err(SnapshotError::io(format!(
            "setting the mode of {}",
            staged_path.display()
        )))?;
    // One call changes both or neither. If the uid cannot change, still try to
    // keep the group: a member of it may give the file that group.
    if std::os::unix::fs::fchown(staged, Some(existing.uid()), Some(existing.gid())).is_err() {
        let _ = std::os::unix::fs::fchown(staged, None, Some(existing.gid()));
    }
    let staged_uid = staged
        .metadata()
        .map_err(SnapshotError::io(format!(
            "inspecting {}",
            staged_path.display()
        )))?
        .uid();
    if owner_keeps_access(existing.uid(), staged_uid, existing.mode()) {
        return Ok(());
    }
    Err(SnapshotError::Io {
        context: format!("keeping the owner of {}", db.display()),
        detail: format!(
            "this account (uid {staged_uid}) cannot give the restored file to its owner \
             (uid {owner}), and mode {mode:o} does not let that owner read and write it, so \
             the service could not open its database. Run the restore as uid {owner}, or as \
             an account that can change file ownership.",
            owner = existing.uid(),
            mode = existing.mode() & 0o7777,
        ),
    })
}

/// Whether uid `owner` can read and write a file that uid `file` owns, with
/// `mode`.
///
/// The owner may or may not be in the file's group, and POSIX then uses the
/// group bits or the other bits. So both must give read and write.
#[cfg(unix)]
const fn owner_keeps_access(owner: u32, file: u32, mode: u32) -> bool {
    owner == 0 || owner == file || mode & 0o066 == 0o066
}

/// Non-unix mode inheritance: `set_permissions` carries the read-only flag, and
/// there is no ownership to copy.
#[cfg(not(unix))]
fn inherit_target_permissions_via(
    db: &Path,
    staged: &std::fs::File,
    staged_path: &Path,
) -> Result<(), SnapshotError> {
    let Ok(existing) = std::fs::metadata(db) else {
        return Ok(());
    };
    staged
        .set_permissions(existing.permissions())
        .map_err(SnapshotError::io(format!(
            "setting the mode of {}",
            staged_path.display()
        )))
}

/// Resolve a symlink at the configured database path.
///
/// A deployed `SQLite` app reaches its database through a link in the release
/// dir (`autumn deploy` keeps the real file in `shared/data`). Renaming over the
/// LINK would detach the app from its own database and let the next deploy undo
/// the restore, and the sidecars would be cleared in the wrong directory.
///
/// Falls back to the given path when it names nothing yet, and resolves a
/// dangling link by hand — `canonicalize` refuses one.
fn follow_link(db: PathBuf) -> PathBuf {
    if let Ok(real) = std::fs::canonicalize(&db) {
        return real;
    }
    match std::fs::read_link(&db) {
        Ok(target) if target.is_absolute() => target,
        Ok(target) => db
            .parent()
            .map_or_else(|| target.clone(), |dir| dir.join(&target)),
        Err(_) => db,
    }
}

/// Flush a directory entry so a rename survives a power loss. Best-effort:
/// opening a directory is not portable, and a failure here costs durability, not
/// correctness.
fn sync_dir(dir: &Path) {
    if let Ok(handle) = std::fs::File::open(dir) {
        let _ = handle.sync_all();
    }
}

/// Restore `artifact` over the `SQLite` database `target` names.
///
/// **Stop the app first.** A running process keeps its own open file handles, so
/// after the rename it serves the old database and loses every write it makes.
///
/// A symlink at the configured path is followed first, so a deployed app whose
/// release dir links to `shared/data` gets its real database replaced, not the
/// link.
///
/// The artifact is verified, copied beside the target, verified again, flushed,
/// and renamed onto it, so a failure part-way through never publishes a
/// half-written database.
///
/// The sidecars go BEFORE the rename. A `-wal` or `-journal` left from the old
/// database describes pages of a file that no longer exists, and `SQLite` would
/// replay it onto the restored one. The reverse order has a worse failure: a
/// removal that fails after the rename leaves a stale journal over the NEW
/// database. Losing the old WAL costs nothing on the success path — the restore
/// discards that database anyway — but a rename that fails afterwards has already
/// discarded it, so its error says so.
///
/// # Errors
///
/// See [`SnapshotError`].
pub fn restore(artifact: &Path, target: &str) -> Result<(), SnapshotError> {
    let db = follow_link(database_path(target)?);
    verify(artifact)?;

    if let Some(parent) = db.parent().filter(|p| !p.as_os_str().is_empty()) {
        std::fs::create_dir_all(parent)
            .map_err(SnapshotError::io(format!("creating {}", parent.display())))?;
    }

    // Stage beside the target so the final `rename` is same-filesystem (atomic).
    let staged = staging_path(&db);
    let _ = std::fs::remove_file(&staged);
    if let Err(e) = stage(artifact, &staged, &db) {
        let _ = std::fs::remove_file(&staged);
        return Err(e);
    }

    for sidecar in sidecars(&db) {
        if let Err(e) = std::fs::remove_file(&sidecar)
            && e.kind() != std::io::ErrorKind::NotFound
        {
            let _ = std::fs::remove_file(&staged);
            return Err(SnapshotError::io(format!("removing {}", sidecar.display()))(e));
        }
    }

    if let Err(e) = std::fs::rename(&staged, &db) {
        let _ = std::fs::remove_file(&staged);
        return Err(SnapshotError::Io {
            context: format!(
                "replacing {} with the restored database. Its write-ahead log was already \
                 removed, so uncommitted-to-disk writes from before the restore are gone",
                db.display()
            ),
            detail: e.to_string(),
        });
    }
    if let Some(parent) = db.parent().filter(|p| !p.as_os_str().is_empty()) {
        sync_dir(parent);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Seed a WAL-mode database with one row and leave the connection open, so
    /// the file has a live `-wal` beside it (the realistic live-app shape).
    fn seeded(path: &Path) -> SqliteConnection {
        seeded_in(path, "WAL")
    }

    /// Seed with a given journal mode. Through a URI, so an awkward name
    /// (non-UTF-8, `file:`) is created exactly.
    fn seeded_in(path: &Path, journal_mode: &str) -> SqliteConnection {
        let uri = replication::file_uri(path).expect("a URI") + "?mode=rwc";
        let mut conn = SqliteConnection::establish(&uri).expect("establish sqlite");
        conn.batch_execute(&format!(
            "PRAGMA journal_mode = {journal_mode}; \
             CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL); \
             INSERT INTO t (v) VALUES ('a'), ('b');"
        ))
        .expect("seed");
        conn
    }

    #[derive(QueryableByName)]
    struct Count {
        #[diesel(sql_type = diesel::sql_types::BigInt)]
        n: i64,
    }

    fn row_count(path: &Path) -> i64 {
        let mut conn = open(path).expect("open sqlite");
        let rows: Vec<Count> = diesel::sql_query("SELECT COUNT(*) AS n FROM t")
            .load(&mut conn)
            .expect("count");
        rows.first().map_or(-1, |r| r.n)
    }

    #[test]
    fn database_path_resolves_every_accepted_spelling_and_refuses_in_memory() {
        assert_eq!(
            database_path("sqlite:///var/lib/app.db").expect("abs"),
            PathBuf::from("/var/lib/app.db")
        );
        assert_eq!(
            database_path("sqlite:app.db").expect("rel"),
            PathBuf::from("app.db")
        );
        for memory in ["sqlite::memory:", "sqlite://:memory:", "sqlite://"] {
            let err = database_path(memory).expect_err("in-memory must be refused");
            assert!(
                matches!(err, SnapshotError::NotAFile { .. }),
                "{memory}: {err}"
            );
            assert!(
                err.to_string().contains("in-memory"),
                "the refusal must name the reason: {err}"
            );
        }
        // A mis-dispatched restore can reach here with a Postgres URL, which
        // carries a password. It must be refused, and the password must not be
        // printed.
        let err = database_path("postgres://app:hunter2@db.example.com/app")
            .expect_err("a Postgres URL is not a SQLite target");
        let message = err.to_string();
        assert!(!message.contains("hunter2"), "credential leaked: {message}");
        assert!(message.contains("<redacted>@"), "{message}");
    }

    /// The resolver and the database engine must name the SAME file. diesel opens
    /// with `SQLITE_OPEN_URI`, so a `file:` target is percent-decoded — this opens
    /// one through diesel and asserts the file that appears on disk is the one
    /// `database_path` reports.
    #[test]
    fn database_path_names_the_file_diesel_actually_opens_for_a_file_uri() {
        let dir = tempfile::tempdir().expect("tempdir");
        let url = format!("file:{}/app%20data.db?mode=rwc", dir.path().display());

        let mut conn = SqliteConnection::establish(&url).expect("diesel opens the URI");
        conn.batch_execute("CREATE TABLE t (id INTEGER PRIMARY KEY);")
            .expect("write");
        drop(conn);

        let resolved = database_path(&url).expect("a file: URI names a file");
        assert!(
            resolved.is_file(),
            "the resolver named {resolved:?}, which does not exist"
        );
        assert_eq!(resolved, dir.path().join("app data.db"));
    }

    #[test]
    fn snapshot_captures_a_live_wal_database_and_the_copy_verifies() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join("app.db");
        // The seeding connection stays open for the whole test: the snapshot must
        // work against a database another connection is holding, WAL and all.
        let mut live = seeded(&db);
        assert!(
            dir.path().join("app.db-wal").exists(),
            "the seed must leave a -wal beside the database"
        );

        let out = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", db.display()), &out).expect("snapshot");
        verify(&out).expect("the snapshot must verify");
        assert_eq!(row_count(&out), 2, "the snapshot must carry the WAL's rows");

        // …and the source is still writable afterwards: the backup neither locked
        // nor corrupted the live database.
        live.batch_execute("INSERT INTO t (v) VALUES ('c');")
            .expect("the live app must still write after a backup");
        assert_eq!(row_count(&db), 3);
        assert_eq!(row_count(&out), 2, "the snapshot is a point in time");
    }

    #[test]
    fn snapshot_refuses_a_missing_database_instead_of_creating_one() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join("absent.db");
        let out = dir.path().join("control.sqlite");
        let err = snapshot(&format!("sqlite://{}", db.display()), &out)
            .expect_err("a missing database must fail");
        assert!(matches!(err, SnapshotError::Missing { .. }), "{err}");
        assert!(
            !db.exists(),
            "the failed backup must not create the database"
        );
        assert!(!out.exists(), "no artifact may be left behind");
    }

    #[test]
    fn verify_rejects_an_empty_and_a_truncated_artifact() {
        let dir = tempfile::tempdir().expect("tempdir");
        let empty = dir.path().join("empty.sqlite");
        std::fs::write(&empty, b"").expect("write");
        assert!(matches!(
            verify(&empty).expect_err("empty must fail"),
            SnapshotError::Integrity { .. }
        ));

        let db = dir.path().join("app.db");
        drop(seeded(&db));
        let good = dir.path().join("good.sqlite");
        snapshot(&format!("sqlite://{}", db.display()), &good).expect("snapshot");
        let bytes = std::fs::read(&good).expect("read");
        let torn = dir.path().join("torn.sqlite");
        std::fs::write(&torn, &bytes[..bytes.len() / 2]).expect("write");
        verify(&torn).expect_err("a truncated artifact must fail");
    }

    #[test]
    fn restore_replaces_the_target_and_clears_the_stale_wal() {
        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");

        // The restore target is a DIFFERENT database, left with the -wal/-shm a
        // crashed app leaves behind (a clean close checkpoints and removes them).
        // Those describe pages of a file the restore is about to replace, so
        // SQLite would replay them onto the restored database and corrupt it.
        let target = dir.path().join("target.db");
        let mut live = seeded(&target);
        live.batch_execute("INSERT INTO t (v) VALUES ('c');")
            .expect("third row");
        drop(live);
        let wal = dir.path().join("target.db-wal");
        let shm = dir.path().join("target.db-shm");
        std::fs::write(&wal, b"stale wal frames").expect("write");
        std::fs::write(&shm, b"stale shm").expect("write");

        restore(&artifact, &format!("sqlite://{}", target.display())).expect("restore");
        assert!(!wal.exists(), "the stale -wal must be removed");
        assert!(!shm.exists(), "the stale -shm must be removed");
        assert_eq!(row_count(&target), 2, "the target must hold the artifact");
    }

    /// The rollback-journal sidecar is the one a WAL-only guard misses.
    /// `VACUUM INTO` writes its output in rollback-journal mode whatever the
    /// source used, so a restored database really can acquire a `-journal`, and a
    /// hot one left from the OLD database is replayed onto the new file and
    /// destroys it.
    #[test]
    fn restore_clears_the_rollback_journal_sidecar_too() {
        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");

        let target = dir.path().join("target.db");
        drop(seeded(&target));
        let journal = dir.path().join("target.db-journal");
        std::fs::write(&journal, b"stale rollback journal").expect("write");

        restore(&artifact, &format!("sqlite://{}", target.display())).expect("restore");
        assert!(!journal.exists(), "the stale -journal must be removed");
        assert_eq!(row_count(&target), 2);
    }

    /// A `0600` database must not come back world-readable because the artifact
    /// was `0644` (an offsite download is created with the default umask).
    #[cfg(unix)]
    #[test]
    fn restore_keeps_the_targets_mode() {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");
        std::fs::set_permissions(&artifact, std::fs::Permissions::from_mode(0o644))
            .expect("chmod artifact");

        let target = dir.path().join("target.db");
        drop(seeded(&target));
        std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o600))
            .expect("chmod target");

        restore(&artifact, &format!("sqlite://{}", target.display())).expect("restore");
        let mode = std::fs::metadata(&target)
            .expect("stat")
            .permissions()
            .mode()
            & 0o777;
        assert_eq!(
            mode, 0o600,
            "the restored database must keep the target's mode"
        );
    }

    /// `autumn deploy` links a release at `shared/data/app.db`, so the configured
    /// path is a symlink. A restore must replace the FILE the link points at.
    /// Replacing the link would detach the app from its database and let the next
    /// deploy re-link over the restored data.
    #[cfg(unix)]
    #[test]
    fn restore_follows_a_symlink_to_the_real_database() {
        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");

        let shared = dir.path().join("shared");
        std::fs::create_dir_all(&shared).expect("mkdir");
        let real = shared.join("app.db");
        let mut live = seeded(&real);
        live.batch_execute("INSERT INTO t (v) VALUES ('c');")
            .expect("third row");
        drop(live);
        let stale_wal = shared.join("app.db-wal");
        std::fs::write(&stale_wal, b"stale wal frames").expect("write");

        let release = dir.path().join("release");
        std::fs::create_dir_all(&release).expect("mkdir");
        let link = release.join("app.db");
        std::os::unix::fs::symlink(&real, &link).expect("symlink");

        restore(&artifact, &format!("sqlite://{}", link.display())).expect("restore");

        assert!(
            std::fs::symlink_metadata(&link)
                .expect("stat")
                .file_type()
                .is_symlink(),
            "the release path must still be a link, not a real file"
        );
        assert_eq!(
            row_count(&real),
            2,
            "the shared file must hold the artifact"
        );
        assert!(
            !stale_wal.exists(),
            "the sidecars must be cleared beside the REAL file"
        );
    }

    /// A symlink planted at the staging path must NOT be followed.
    ///
    /// The staging name is predictable and `restore` removes it before copying,
    /// so an account that can write the database directory has a window to plant
    /// a link there. `std::fs::copy` follows one, which under a privileged
    /// restore truncates whatever it points at — before verification runs. The
    /// exclusive create must fail instead.
    #[cfg(unix)]
    #[test]
    fn staging_never_follows_a_symlink_planted_at_its_path() {
        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");

        let target = dir.path().join("target.db");
        drop(seeded(&target));
        let victim = dir.path().join("victim");
        std::fs::write(&victim, b"MUST NOT BE TRUNCATED").expect("write victim");

        // Plant the link exactly where the stage will be created.
        let staged = staging_path(&target);
        std::os::unix::fs::symlink(&victim, &staged).expect("plant symlink");

        stage(&artifact, &staged, &target).expect_err("staging must refuse an occupied path");
        assert_eq!(
            std::fs::read(&victim).expect("victim readable"),
            b"MUST NOT BE TRUNCATED",
            "the symlink target must be untouched"
        );
    }

    /// The staging file is owned through its HANDLE, so a swap after creation
    /// cannot redirect the mode, the owner or the publication (#2589 round 18).
    ///
    /// `O_CREAT | O_EXCL` proves only that nothing was there when the file was
    /// made. The name stays writable by whoever can write the directory, so the
    /// earlier code — which dropped the handle and then reopened the NAME for
    /// `set_permissions`, `chown`, `verify` and the flush — could be pointed at a
    /// symlink planted in between, handing an attacker the mode and owner of
    /// whatever it pointed at and publishing their link as the database.
    #[cfg(unix)]
    #[test]
    fn a_staging_file_swapped_after_creation_is_never_published() {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");

        let target = dir.path().join("target.db");
        drop(seeded(&target));
        let victim = dir.path().join("victim");
        std::fs::write(&victim, b"MUST NOT BE TOUCHED").expect("write victim");
        let victim_before = std::fs::metadata(&victim).expect("victim metadata");

        // Simulate the swap the guard exists to catch: the staging file is
        // created, then replaced by a symlink at the same name before the steps
        // that must address it by name run.
        let staged = staging_path(&target);
        let handle = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&staged)
            .expect("create the staging file");
        std::fs::remove_file(&staged).expect("unlink it");
        std::os::unix::fs::symlink(&victim, &staged).expect("plant symlink");

        same_file_as_handle(&handle, &staged).expect_err("a swapped staging name must be refused");

        // And the victim is untouched — mode, owner and bytes.
        let victim_after = std::fs::metadata(&victim).expect("victim metadata");
        assert_eq!(
            std::fs::read(&victim).expect("victim readable"),
            b"MUST NOT BE TOUCHED"
        );
        assert_eq!(
            victim_before.permissions().mode(),
            victim_after.permissions().mode(),
            "the victim's mode must not have been inherited onto it"
        );

        std::fs::remove_file(&staged).ok();
    }

    /// The same-file check passes for the file the handle actually owns, so the
    /// guard above refuses a swap rather than refusing everything.
    #[cfg(unix)]
    #[test]
    fn the_staging_handle_matches_its_own_unswapped_name() {
        let dir = tempfile::tempdir().expect("tempdir");
        let staged = dir.path().join("staged.db");
        let handle = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&staged)
            .expect("create");
        same_file_as_handle(&handle, &staged).expect("an unswapped name must pass");
    }

    /// Item 7: the source opens with no create. A file removed after the
    /// existence check fails the open and is not created again.
    #[test]
    fn connect_never_creates_a_database_that_vanished() {
        let dir = tempfile::tempdir().expect("tempdir");
        let gone = dir.path().join("gone.db");
        let uri = replication::no_create_uri(&gone).expect("a URI");
        let Err(err) = connect(&uri, &gone) else {
            panic!("a missing file must not open");
        };
        assert!(matches!(err, SnapshotError::Open { .. }), "{err}");
        assert!(!gone.exists(), "the open must not create {gone:?}");
    }

    /// Item 3: a non-UTF-8 `file:` source is backed up.
    ///
    /// Not on macOS: APFS refuses a name that is not UTF-8 (`EILSEQ`), so
    /// there such a file cannot exist.
    #[cfg(all(unix, not(target_os = "macos")))]
    #[test]
    fn snapshot_reads_a_non_utf8_file_uri_source() {
        use std::os::unix::ffi::OsStrExt as _;

        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join(std::ffi::OsStr::from_bytes(b"app\xFF.db"));
        drop(seeded(&db));
        let url = format!("file:{}/app%FF.db", dir.path().display());
        let out = dir.path().join("control.sqlite");
        snapshot(&url, &out).expect("snapshot");
        assert_eq!(row_count(&out), 2);
    }

    /// Item 12: a source opened through a locking VFS is backed up through the
    /// same VFS.
    #[cfg(unix)]
    #[test]
    fn snapshot_keeps_the_source_vfs() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join("app.db");
        // Rollback journal: `unix-dotfile` has no shared memory, so WAL cannot
        // open through it.
        drop(seeded_in(&db, "DELETE"));
        let url = format!("file:{}?vfs=unix-dotfile", db.display());
        let target = database_target(&url).expect("a file");
        assert!(
            target
                .no_create_uri()
                .is_some_and(|uri| uri.contains("vfs=unix-dotfile")),
            "the opener must keep the VFS"
        );
        let out = dir.path().join("control.sqlite");
        snapshot(&url, &out).expect("snapshot");
        assert_eq!(row_count(&out), 2);

        // An unknown VFS fails, so the opener really uses the source VFS.
        let unknown = format!("file:{}?vfs=autumn-no-such-vfs", db.display());
        let err = snapshot(&unknown, &dir.path().join("other.sqlite"))
            .expect_err("an unknown VFS must fail the open");
        assert!(matches!(err, SnapshotError::Open { .. }), "{err}");
    }

    /// Item 15: a non-UTF-8 output directory is used exactly.
    ///
    /// Not on macOS: APFS refuses a name that is not UTF-8 (`EILSEQ`), so
    /// there such a file cannot exist.
    #[cfg(all(unix, not(target_os = "macos")))]
    #[test]
    fn snapshot_writes_into_a_non_utf8_directory() {
        use std::os::unix::ffi::OsStrExt as _;

        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join("app.db");
        drop(seeded(&db));
        let out_dir = dir.path().join(std::ffi::OsStr::from_bytes(b"run\xFF"));
        std::fs::create_dir(&out_dir).expect("mkdir");
        let out = out_dir.join("control.sqlite");
        snapshot(&format!("sqlite://{}", db.display()), &out).expect("snapshot");
        assert_eq!(row_count(&out), 2, "the artifact must be at {out:?}");
    }

    /// Item 18: a relative output path that starts with `file:` is a path, not
    /// a URI.
    ///
    /// It writes a directory in the working directory, because the hazard is
    /// only for a relative path. The name is unique to the process, and the
    /// guard removes it. Unix only: Windows does not allow `:` in a name.
    #[cfg(unix)]
    #[test]
    fn snapshot_writes_to_a_relative_path_that_starts_with_file() {
        struct Cleanup(PathBuf);
        impl Drop for Cleanup {
            fn drop(&mut self) {
                let _ = std::fs::remove_dir_all(&self.0);
            }
        }

        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join("app.db");
        drop(seeded(&db));
        let out_dir = PathBuf::from(format!("file:autumn-2589-{}", std::process::id()));
        let _cleanup = Cleanup(out_dir.clone());
        std::fs::create_dir_all(&out_dir).expect("mkdir");
        let out = out_dir.join("control.sqlite");
        snapshot(&format!("sqlite://{}", db.display()), &out).expect("snapshot");
        assert_eq!(row_count(&out), 2, "the artifact must be at {out:?}");
    }

    /// Item 14: a restore replaces a read-only (`0444`) database and keeps its
    /// mode.
    #[cfg(unix)]
    #[test]
    fn restore_replaces_a_read_only_target() {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("source.db");
        drop(seeded(&source));
        let artifact = dir.path().join("control.sqlite");
        snapshot(&format!("sqlite://{}", source.display()), &artifact).expect("snapshot");

        let target = dir.path().join("target.db");
        let mut live = seeded(&target);
        live.batch_execute("INSERT INTO t (v) VALUES ('c');")
            .expect("third row");
        drop(live);
        std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o444))
            .expect("chmod target");

        restore(&artifact, &format!("sqlite://{}", target.display())).expect("restore");
        assert_eq!(row_count(&target), 2);
        let mode = std::fs::metadata(&target)
            .expect("stat")
            .permissions()
            .mode()
            & 0o777;
        assert_eq!(mode, 0o444);
    }

    /// Item 4: when the uid cannot be kept, the old owner must still read and
    /// write the file.
    #[cfg(unix)]
    #[test]
    fn owner_keeps_access_needs_the_same_uid_or_shared_read_write() {
        // (owner uid, file uid, mode, owner keeps access)
        for (owner, file, mode, keeps) in [
            (1000, 1000, 0o600, true),
            (1000, 1001, 0o600, false),
            (1000, 1001, 0o660, false),
            (1000, 1001, 0o666, true),
            (1000, 1001, 0o646, false),
            (1000, 1001, 0o644, false),
            (0, 1001, 0o600, true),
        ] {
            assert_eq!(
                owner_keeps_access(owner, file, mode),
                keeps,
                "owner={owner} file={file} mode={mode:o}"
            );
        }
    }

    #[test]
    fn restore_refuses_a_corrupt_artifact_without_touching_the_target() {
        let dir = tempfile::tempdir().expect("tempdir");
        let target = dir.path().join("target.db");
        drop(seeded(&target));
        let before = std::fs::read(&target).expect("read");

        let artifact = dir.path().join("junk.sqlite");
        std::fs::write(&artifact, b"not a sqlite database at all").expect("write");
        restore(&artifact, &format!("sqlite://{}", target.display()))
            .expect_err("a corrupt artifact must be refused");

        assert_eq!(
            std::fs::read(&target).expect("read"),
            before,
            "a refused restore must leave the target byte-identical"
        );
        assert!(
            !staging_path(&target).exists(),
            "no staging file may be left behind"
        );
    }
}