//! Tier 1a — full snapshot sink.
//!
//! `VACUUM INTO` a temp file → upload the whole file to an object store
//! (S3 / R2 / MinIO) → restore = download + open with SQLite. The DB-header
//! `change_counter` gates skip-if-unchanged. Uses turso's public API only.
//!
//! @yah:relay(R003, "Tier 1a — full snapshot sink (VACUUM INTO → S3)")
//! @yah:at(2026-05-26T22:28:30Z)
//! @yah:status(open)
//! @yah:phase(P1)
//! @yah:parent(Q002)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//!
//! @yah:ticket(R003-F2, "Full-snapshot sink: VACUUM INTO temp + object_store upload (S3/R2/MinIO, path-style); change_counter skip-if-unchanged")
//! @yah:assignee(bundle-anthropic-ashguard)
//! @yah:at(2026-05-27T00:51:38Z)
//! @yah:status(review)
//! @yah:phase(P1)
//! @yah:parent(R003)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R003-T1)
//! @yah:handoff("Implemented snapshot_and_upload with a TWO-GATE skip guard: gate 1 = source-file hash (main + -wal) skips vacuum+upload on the quiescent path; gate 2 = canonical VACUUM INTO output hash (byte-deterministic, verified) dedups the upload when a checkpoint reshuffled bytes without changing content. VACUUM INTO unique temp -> object_store.put + two sidecars (latest.source-fingerprint, latest.snapshot-fingerprint). Returns SnapshotOutcome::{Uploaded,Unchanged,Deduplicated}. Backend-agnostic (Arc<dyn ObjectStore>) -> path-style S3/R2/MinIO is the caller's builder config.")
//! @yah:handoff("DEVIATION from ticket title: change_counter is DEAD in turso — stays 1 across rollback/WAL/WAL+checkpoint (verified; VACUUM INTO doesn't copy it). Hashes replace it. Two-gate design follows user's steer (determinism worth the lift when not expensive); gate 1 keeps the idle path off the vacuum. Still in review for final sign-off. Working doc + Cargo.toml + rustdoc updated.")
//! @yah:verify("cargo test -p turso-backup: 2 tests green on InMemory store — (1) upload -> SQLite-magic + turso 100-row readback -> gate-1 Unchanged -> mutate -> re-upload; (2) gate-2 Deduplicated path + fingerprint refresh. cargo build/clippy clean.")
//! @yah:next("R003-F3: restore_latest (list snapshots/ by last-modified, get newest, write file). Layout: {prefix}/snapshots/snapshot-{unix_nanos:020}.db + {prefix}/latest.fingerprint.")
//! @yah:next("R003-T4: MinIO/path-style e2e in harness; thread experimental_multiprocess_wal through vacuum_into if the writer's WAL format requires it.")
//!
//! @yah:ticket(R003-F3, "Restore path: download latest snapshot, open with sqlite3, verify")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:04Z)
//! @yah:status(review)
//! @yah:phase(P1)
//! @yah:parent(R003)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R003-F2)
//! @yah:verify("cargo test -p turso-backup: 3 green on InMemory store (new restore_latest_picks_newest_and_round_trips). cargo clippy --all-targets clean.")
//! @yah:handoff("Implemented restore_latest(target, dest_path) -> Result<String>: list_with_delimiter under {prefix}/snapshots/ (excludes the latest.* sidecars by construction), pick the lexically-greatest key (zero-padded nanos = newest; clock-skew-immune vs last_modified), get -> std::fs::write to dest. Returns the restored key. Errors if no snapshots. Moved the fn above mod tests (clippy items_after_test_module).")
//! @yah:next("R003-T4: wire writer -> snapshot_and_upload -> MinIO -> restore_latest -> verifier (5000 rows, sqlite3 integrity_check). The literal 'open with sqlite3' verify lives in the harness; F3 covers it at the lib level (SQLite-magic + turso readback).")
//!
//! @yah:ticket(R003-T4, "Green end-to-end in harness: writer -> snapshot sink -> MinIO -> restore -> verifier checks (5000 rows)")
//! @yah:assignee(bundle-anthropic-ashguard)
//! @yah:at(2026-05-27T08:29:26Z)
//! @yah:status(review)
//! @yah:phase(P1)
//! @yah:parent(R003)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R003-F3)
//! @yah:verify("bash ci/run.sh (staged: minio -> minio-init -> writer -> verifier); verifier exit 0 = green")
//! @yah:handoff("GREEN e2e at 5000 rows. Rewired the harness off Litestream onto turso-backup: new harness/ crate (writer + restore bins + shared backup_target lib, path-dep on turso-backup, object_store aws feature for MinIO). One multi-stage Dockerfile (shared builder, writer/verifier targets). Deleted dead Litestream writer/ + verifier/ dirs. Writer uploaded 446KB snapshot; verifier restored + passed sqlite3 integrity/rowcount/boundary/hash (matched byte-for-byte), exit 0.")
use anyhow::{Context, Result};
use object_store::path::Path as ObjPath;
use object_store::{ObjectStore, ObjectStoreExt};
use sha2::{Digest, Sha256};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use crate::stream::validated_against_source;
/// Configuration for an object-store backup target (S3 / R2 / MinIO, path-style).
///
/// Wraps a generic `object_store::ObjectStore` so the caller picks the backend
/// (e.g., `AmazonS3Builder::new().with_endpoint("http://minio:9000").build()`),
/// then hands this crate a `BackupTarget` to use for snapshot operations.
pub struct BackupTarget {
pub store: Arc<dyn ObjectStore>,
pub prefix: String,
}
impl BackupTarget {
/// Object key for a snapshot taken at `unix_nanos` (zero-padded so lexical
/// order matches chronological order — restore can pick the newest by name
/// or by last-modified).
fn snapshot_key(&self, unix_nanos: u128) -> ObjPath {
join_key(&self.prefix, &format!("snapshots/snapshot-{unix_nanos:020}.db"))
}
/// Sidecar recording the hash of the **source files** at the last run —
/// the cheap gate-1 short-circuit (skip vacuum+upload when bytes match).
fn source_fingerprint_key(&self) -> ObjPath {
join_key(&self.prefix, "latest.source-fingerprint")
}
/// Sidecar recording the hash of the last uploaded **snapshot** (the
/// canonical `VACUUM INTO` output) — the deterministic gate-2 dedup.
fn snapshot_fingerprint_key(&self) -> ObjPath {
join_key(&self.prefix, "latest.snapshot-fingerprint")
}
}
/// Result of a [`snapshot_and_upload`] call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SnapshotOutcome {
/// A new snapshot was written to `key` (`bytes` long). `snapshot_hash` is
/// the canonical hash of the uploaded file, now recorded in the sidecar.
Uploaded {
key: String,
bytes: usize,
snapshot_hash: String,
},
/// Gate 1: the source files were byte-identical to the last run, so nothing
/// was vacuumed or uploaded. `source_hash` is the unchanged source hash.
Unchanged { source_hash: String },
/// Gate 2: the source bytes moved (e.g. an auto-checkpoint reshuffled
/// pages) but the canonical `VACUUM INTO` output matched the last snapshot,
/// so the upload was skipped. The cheap source fingerprint is refreshed so
/// the next run short-circuits at gate 1.
Deduplicated { snapshot_hash: String },
}
/// Take a full `VACUUM INTO` snapshot of the turso database at `db_path` and
/// upload the resulting file to the object store, behind a two-gate
/// skip-if-unchanged guard.
///
/// 1. **Gate 1 (cheap):** hash the source files (`db_path` + `-wal`) and compare
/// to the last run's sidecar. Identical bytes → return [`Unchanged`] without
/// touching the DB. This is the common quiescent path.
/// 2. Otherwise `VACUUM INTO` a unique temp file (point-in-time-consistent,
/// fsync'd; the output is valid vanilla SQLite).
/// 3. **Gate 2 (canonical):** `VACUUM INTO` output is byte-deterministic for
/// identical content, so if its hash matches the last uploaded snapshot the
/// source changed without the content changing → return [`Deduplicated`]
/// (upload skipped, source fingerprint refreshed).
/// 4. Otherwise upload the snapshot and record both fingerprints.
///
/// ## Why hashes, not the SQLite `change_counter`
///
/// The original plan used the DB-header `change_counter` (offset 24) as the
/// "did anything change?" signal. The turso rewrite **does not maintain it** —
/// it is left at the default `1` across every write in rollback, WAL, and
/// WAL+checkpoint mode (verified empirically; `VACUUM INTO` doesn't even copy it
/// through). Using it would skip every snapshot after the first. The two-gate
/// hash design is conservative: it only ever skips on a proven match, so the
/// failure direction is a redundant upload, never a missed backup.
///
/// [`Unchanged`]: SnapshotOutcome::Unchanged
/// [`Deduplicated`]: SnapshotOutcome::Deduplicated
pub async fn snapshot_and_upload(db_path: &str, target: &BackupTarget) -> Result<SnapshotOutcome> {
// Gate 1 (cheap): source files unchanged since last run -> nothing to do.
let source_hash = source_fingerprint(db_path)?;
let prev_source = read_text(&target.store, &target.source_fingerprint_key()).await?;
if prev_source.as_deref() == Some(source_hash.as_str()) {
return Ok(SnapshotOutcome::Unchanged { source_hash });
}
// Source moved -> take the canonical snapshot. SQLite refuses an existing
// VACUUM INTO target, so the nanosecond + pid name is unique and removed first.
let nanos = unix_nanos();
let temp_path = std::env::temp_dir().join(format!(
"turso-snapshot-{}-{nanos}.db",
std::process::id()
));
let _ = std::fs::remove_file(&temp_path);
vacuum_into(db_path, &temp_path).await?;
let bytes = std::fs::read(&temp_path)
.with_context(|| format!("reading vacuum output {}", temp_path.display()))?;
let _ = std::fs::remove_file(&temp_path);
// Gate 2 (canonical): the vacuum output is byte-deterministic for identical
// content, so a match means the bytes moved without the content changing
// (e.g. an auto-checkpoint reshuffle). Skip the upload, but refresh the
// cheap source fingerprint so the next run short-circuits at gate 1.
let snapshot_hash = sha256_hex(&bytes);
let prev_snapshot = read_text(&target.store, &target.snapshot_fingerprint_key()).await?;
if prev_snapshot.as_deref() == Some(snapshot_hash.as_str()) {
put_text(&target.store, &target.source_fingerprint_key(), &source_hash).await?;
return Ok(SnapshotOutcome::Deduplicated { snapshot_hash });
}
// Genuinely new content -> upload, then record both fingerprints.
let key = target.snapshot_key(nanos);
let len = bytes.len();
target
.store
.put(&key, bytes.into())
.await
.with_context(|| format!("uploading snapshot to {key}"))?;
put_text(&target.store, &target.snapshot_fingerprint_key(), &snapshot_hash).await?;
put_text(&target.store, &target.source_fingerprint_key(), &source_hash).await?;
Ok(SnapshotOutcome::Uploaded {
key: key.to_string(),
bytes: len,
snapshot_hash,
})
}
/// Publish an already-materialized database image as a tier-1a snapshot,
/// without opening the source database at all.
///
/// [`snapshot_and_upload`] is the right entry point when this process may open
/// `db_path` freely. Two kinds of caller cannot:
///
/// - One that already holds the only permitted handle on the file. roadcase's
/// residency ladder is the example: `turso_core::Database::open_file_with_flags`
/// consults a **process-global registry keyed by file id** before it reads the
/// open flags, so a second opener gets the first one's handle back with its
/// own flags discarded. Such a caller has a live `turso_core::Connection` and
/// must produce the image through it.
/// - One that needs the snapshot to be *page-compatible* with the WAL frames
/// [`crate::stream::tail_frames`] will upload next. `VACUUM INTO` repacks the
/// database, so its output is a valid SQLite file but not necessarily the
/// same page layout the engine will number subsequent frames against. The
/// image a tier-2 base wants is the main database file itself, read after a
/// `PRAGMA wal_checkpoint(TRUNCATE)` has emptied the WAL into it.
///
/// Writes one object under the same `snapshots/snapshot-{unix_nanos:020}.db`
/// layout [`snapshot_and_upload`] uses, so [`restore_latest`] finds it and
/// [`crate::stream::StreamConfig::base_snapshot_key`] can name it. Returns that
/// key.
///
/// Deliberately writes **neither fingerprint sidecar.** Both gate
/// `snapshot_and_upload`, and this path has hashed no source files — recording
/// a `latest.source-fingerprint` here would assert a correspondence between an
/// object and a set of file bytes that was never checked, which is the one
/// thing that could make the two-gate skip drop a real backup.
pub async fn upload_base_snapshot(target: &BackupTarget, image: &[u8]) -> Result<String> {
anyhow::ensure!(
image.starts_with(b"SQLite format 3\0"),
"refusing to publish a base snapshot that is not a SQLite database \
({} bytes, prefix {:?})",
image.len(),
&image[..image.len().min(16)],
);
let key = target.snapshot_key(unix_nanos());
target
.store
.put(&key, image.to_vec().into())
.await
.with_context(|| format!("uploading base snapshot to {key}"))?;
Ok(key.to_string())
}
/// [`upload_base_snapshot`] behind the canonical (gate-2) skip, for a caller
/// that publishes an image *repeatedly*.
///
/// R850-F1. `upload_base_snapshot` is a one-shot: every call writes an object,
/// which is right for publishing a tier-2 base once and wrong for a tail that
/// re-snapshots a mostly-idle database every interval forever. This applies the
/// content hash gate `snapshot_and_upload` calls gate 2 — identical image bytes
/// to the last publish means skip the upload — and returns
/// [`SnapshotOutcome::Deduplicated`] when it fires.
///
/// It records **only** `latest.snapshot-fingerprint`, never
/// `latest.source-fingerprint`, for the reason `upload_base_snapshot` gives at
/// length: this path hashes no source *files*, so claiming a correspondence
/// between an object and a set of file bytes it never read is the one thing that
/// could make `snapshot_and_upload`'s cheap gate 1 skip a real backup later.
/// [`SnapshotOutcome::Unchanged`] is therefore never returned from here.
pub async fn upload_snapshot_image(
target: &BackupTarget,
image: &[u8],
) -> Result<SnapshotOutcome> {
let snapshot_hash = sha256_hex(image);
let prev = read_text(&target.store, &target.snapshot_fingerprint_key()).await?;
if prev.as_deref() == Some(snapshot_hash.as_str()) {
return Ok(SnapshotOutcome::Deduplicated { snapshot_hash });
}
let key = upload_base_snapshot(target, image).await?;
put_text(
&target.store,
&target.snapshot_fingerprint_key(),
&snapshot_hash,
)
.await?;
Ok(SnapshotOutcome::Uploaded {
key,
bytes: image.len(),
snapshot_hash,
})
}
/// SHA-256 of the source database files: the main `.db` plus its `-wal`
/// sidecar if present (WAL-mode commits live there until a checkpoint). A
/// quiescent database hashes identically on re-read; any committed write
/// changes the result. Returned as a lowercase hex string.
fn source_fingerprint(db_path: &str) -> Result<String> {
let mut hasher = Sha256::new();
let main = std::fs::read(db_path).with_context(|| format!("reading source db {db_path}"))?;
hasher.update(&main);
if let Ok(wal) = std::fs::read(format!("{db_path}-wal")) {
hasher.update(b"\0-wal\0");
hasher.update(&wal);
}
Ok(hex::encode(hasher.finalize()))
}
/// Lowercase-hex SHA-256 of an in-memory buffer (used for the canonical
/// snapshot fingerprint).
fn sha256_hex(bytes: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(bytes);
hex::encode(hasher.finalize())
}
/// Read a small text sidecar (a fingerprint), or `None` if it doesn't exist yet.
async fn read_text(store: &Arc<dyn ObjectStore>, key: &ObjPath) -> Result<Option<String>> {
match store.get(key).await {
Ok(res) => {
let bytes = res.bytes().await.context("reading fingerprint object")?;
Ok(Some(String::from_utf8_lossy(&bytes).trim().to_string()))
}
Err(object_store::Error::NotFound { .. }) => Ok(None),
Err(e) => Err(e).context("fetching fingerprint object"),
}
}
/// Write a small text sidecar (a fingerprint).
async fn put_text(store: &Arc<dyn ObjectStore>, key: &ObjPath, text: &str) -> Result<()> {
store
.put(key, text.to_owned().into_bytes().into())
.await
.with_context(|| format!("recording fingerprint {key}"))?;
Ok(())
}
/// `VACUUM INTO '<temp_path>'` against the database at `db_path`, opened
/// **read-only** and validated against a concurrent foreign writer.
///
/// R858-B18 changed two things here, for one reason: a backup is a reader, and
/// this was the second of the crate's two source opens that did not say so.
///
/// 1. **`turso_core` with `OpenFlags::ReadOnly`, not `turso::Builder`.** The
/// friendly wrapper exposes no `OpenFlags`, so it opened the source writable
/// and took turso's whole-file exclusive `fcntl` lock — which is refused
/// outright when the application that owns the database is running
/// (measured: `examples/foreign_checkpoint_probe.rs` probe S1, `Locking
/// error: File is locked by another process`). `ReadOnly` skips that lock per
/// handle, and `VACUUM INTO` still runs on such a connection (probe S2: 50/50
/// rows, `integrity_check ok`) — note `VACUUM INTO` is not gated by
/// `DatabaseOpts::enable_vacuum`, only bare `VACUUM` is
/// (`turso_core/translate/vacuum.rs:39`).
/// 2. **Wrapped in [`validated_against_source`].** Opening without a lock is not
/// coordination: a foreign checkpoint can still land mid-vacuum, and VACUUM
/// INTO rebuilds the b-tree, so a torn read of the source yields an output
/// that passes `integrity_check` while holding the wrong rows. Same protocol
/// as tier 2's copy — accept only if the source provably held still, else
/// retry, else refuse.
///
/// The destination is removed at the start of every attempt, because SQLite
/// refuses to `VACUUM INTO` a file that already exists and a retried attempt
/// would otherwise fail on its predecessor's output.
async fn vacuum_into(db_path: &str, temp_path: &std::path::Path) -> Result<()> {
// Single-quote the path SQLite-style (double any embedded quote).
let escaped = temp_path.to_string_lossy().replace('\'', "''");
validated_against_source(db_path, "VACUUM INTO snapshot", || async {
let _ = std::fs::remove_file(temp_path);
let io: Arc<dyn turso_core::IO> =
Arc::new(turso_core::PlatformIO::new().context("creating turso_core PlatformIO")?);
let db = turso_core::Database::open_file_with_flags(
io,
db_path,
turso_core::OpenFlags::ReadOnly,
turso_core::DatabaseOpts::new(),
None,
)
.with_context(|| format!("opening turso db {db_path} read-only"))?;
let conn = db.connect().context("connecting to turso db")?;
conn.execute(format!("VACUUM INTO '{escaped}'"))
.context("VACUUM INTO failed")?;
Ok(())
})
.await
}
/// Join an object-store prefix and a leaf into a normalized [`ObjPath`],
/// tolerating empty/slash-padded prefixes.
fn join_key(prefix: &str, leaf: &str) -> ObjPath {
let prefix = prefix.trim_matches('/');
if prefix.is_empty() {
ObjPath::from(leaf)
} else {
ObjPath::from(format!("{prefix}/{leaf}"))
}
}
fn unix_nanos() -> u128 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
}
/// Download the most recent snapshot and write it to `dest_path`, returning the
/// object key it was restored from.
///
/// Lists the snapshot objects under `{prefix}/snapshots/` and picks the newest.
/// Snapshot keys are zero-padded nanosecond timestamps (see
/// [`BackupTarget::snapshot_key`]), so the lexically-greatest key is the most
/// recent. We order by the key's baked-in capture time rather than the store's
/// `last_modified` so the choice is deterministic and immune to clock skew
/// between the snapshotting host and the object store.
///
/// The `snapshots/` prefix naturally excludes the `latest.*-fingerprint`
/// sidecars (they live one level up at `{prefix}/`). The downloaded bytes are a
/// valid vanilla-SQLite file written verbatim to `dest_path`, ready to open with
/// `sqlite3` or re-open through turso.
///
/// Errors if no snapshots exist under the prefix yet.
pub async fn restore_latest(target: &BackupTarget, dest_path: &str) -> Result<String> {
let snapshots_prefix = join_key(&target.prefix, "snapshots");
let newest = latest_snapshot_key(target)
.await?
.with_context(|| format!("no snapshots found under {snapshots_prefix}"))?;
let location = ObjPath::from(newest.clone());
let bytes = target
.store
.get(&location)
.await
.with_context(|| format!("downloading snapshot {location}"))?
.bytes()
.await
.with_context(|| format!("reading snapshot body {location}"))?;
std::fs::write(dest_path, &bytes)
.with_context(|| format!("writing restored snapshot to {dest_path}"))?;
Ok(newest)
}
/// The key [`restore_latest`] would download, without downloading it — or `None`
/// when the prefix holds no snapshot at all.
///
/// R850-F1 split this out for the tail side, which needs the *name* of an
/// existing tier-2 base rather than its bytes: a restarted tail that cannot see
/// that a base already exists re-copies the whole database on every restart, and
/// a `None` here is what tells it there is genuinely nothing to anchor to.
///
/// Absence is a `None` rather than the `Err` [`restore_latest`] returns because
/// the two callers want opposite things from it: a restore with no snapshot has
/// failed, while a first-ever tail with no snapshot is simply first.
pub async fn latest_snapshot_key(target: &BackupTarget) -> Result<Option<String>> {
let snapshots_prefix = join_key(&target.prefix, "snapshots");
let listing = target
.store
.list_with_delimiter(Some(&snapshots_prefix))
.await
.with_context(|| format!("listing snapshots under {snapshots_prefix}"))?;
Ok(listing
.objects
.into_iter()
.max_by(|a, b| a.location.cmp(&b.location))
.map(|o| o.location.to_string()))
}
#[cfg(test)]
mod tests {
use super::*;
use object_store::memory::InMemory;
// The library itself no longer opens a source database through the friendly
// wrapper (R858-B18 moved that to turso_core + OpenFlags::ReadOnly); these
// tests still use it to *write* their fixtures, which is the right surface
// for a writer.
use turso::Builder;
/// A throwaway db path under the OS temp dir, cleaned up on drop.
struct TempDb(std::path::PathBuf);
impl TempDb {
fn new(tag: &str) -> Self {
let p = std::env::temp_dir().join(format!(
"turso-backup-test-{}-{}-{}.db",
tag,
std::process::id(),
unix_nanos()
));
TempDb(p)
}
fn path(&self) -> &str {
self.0.to_str().unwrap()
}
}
impl Drop for TempDb {
fn drop(&mut self) {
for suffix in ["", "-wal", "-shm"] {
let _ = std::fs::remove_file(format!("{}{suffix}", self.0.display()));
}
}
}
async fn seed_db(path: &str, rows: i64) {
let db = Builder::new_local(path).build().await.unwrap();
let conn = db.connect().unwrap();
conn.execute("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v TEXT)", ())
.await
.unwrap();
conn.execute("BEGIN", ()).await.unwrap();
for i in 0..rows {
conn.execute("INSERT INTO t (id, v) VALUES (?, ?)", (i, format!("v{i}")))
.await
.unwrap();
}
conn.execute("COMMIT", ()).await.unwrap();
}
async fn count_rows(path: &str) -> i64 {
let db = Builder::new_local(path).build().await.unwrap();
let conn = db.connect().unwrap();
let mut r = conn.query("SELECT COUNT(*) FROM t", ()).await.unwrap();
let row = r.next().await.unwrap().unwrap();
row.get::<i64>(0).unwrap()
}
#[tokio::test]
async fn uploads_then_skips_unchanged_then_reuploads_on_change() {
let src = TempDb::new("src");
seed_db(src.path(), 100).await;
let target = BackupTarget {
store: Arc::new(InMemory::new()),
prefix: "backups".into(),
};
// First snapshot: uploaded.
let out = snapshot_and_upload(src.path(), &target).await.unwrap();
let (key, snap1) = match out {
SnapshotOutcome::Uploaded { key, bytes, snapshot_hash } => {
assert!(bytes > 0, "snapshot should be non-empty");
assert!(key.starts_with("backups/snapshots/snapshot-"), "key was {key}");
(key, snapshot_hash)
}
other => panic!("expected Uploaded, got {other:?}"),
};
// The uploaded object is a real vanilla-SQLite file.
let bytes = target
.store
.get(&ObjPath::from(key.clone()))
.await
.unwrap()
.bytes()
.await
.unwrap();
assert!(
bytes.starts_with(b"SQLite format 3\0"),
"uploaded object is not a SQLite database"
);
// ...and round-trips through turso with the right row count.
let restored = TempDb::new("restored");
std::fs::write(restored.path(), &bytes).unwrap();
assert_eq!(count_rows(restored.path()).await, 100);
// Gate 1: no source change -> skipped without vacuuming.
match snapshot_and_upload(src.path(), &target).await.unwrap() {
SnapshotOutcome::Unchanged { .. } => {}
other => panic!("expected Unchanged, got {other:?}"),
}
// Mutate the source, then snapshot again: uploaded, new snapshot hash.
{
let db = Builder::new_local(src.path()).build().await.unwrap();
let conn = db.connect().unwrap();
conn.execute("INSERT INTO t (id, v) VALUES (?, ?)", (10_000, "new"))
.await
.unwrap();
}
match snapshot_and_upload(src.path(), &target).await.unwrap() {
SnapshotOutcome::Uploaded { snapshot_hash, .. } => assert_ne!(snapshot_hash, snap1),
other => panic!("expected Uploaded after change, got {other:?}"),
}
}
/// Gate 2: when the source bytes look changed but the canonical VACUUM
/// output matches the last snapshot, the upload is deduplicated. We simulate
/// the "source moved without content change" case by deleting the cheap
/// source-fingerprint sidecar so gate 1 misses and the vacuum runs.
#[tokio::test]
async fn deduplicates_when_vacuum_output_matches() {
let src = TempDb::new("dedup-src");
seed_db(src.path(), 50).await;
let target = BackupTarget {
store: Arc::new(InMemory::new()),
prefix: String::new(),
};
let snap1 = match snapshot_and_upload(src.path(), &target).await.unwrap() {
SnapshotOutcome::Uploaded { snapshot_hash, .. } => snapshot_hash,
other => panic!("expected Uploaded, got {other:?}"),
};
// Force gate 1 to miss without changing content.
target
.store
.delete(&target.source_fingerprint_key())
.await
.unwrap();
match snapshot_and_upload(src.path(), &target).await.unwrap() {
SnapshotOutcome::Deduplicated { snapshot_hash } => assert_eq!(snapshot_hash, snap1),
other => panic!("expected Deduplicated, got {other:?}"),
}
// And the refreshed source fingerprint makes the next run short-circuit.
match snapshot_and_upload(src.path(), &target).await.unwrap() {
SnapshotOutcome::Unchanged { .. } => {}
other => panic!("expected Unchanged after dedup refresh, got {other:?}"),
}
}
/// restore_latest errors when empty, then downloads the newest of several
/// snapshots and round-trips it back through turso with the latest row count.
#[tokio::test]
async fn restore_latest_picks_newest_and_round_trips() {
let src = TempDb::new("restore-src");
seed_db(src.path(), 100).await;
let target = BackupTarget {
store: Arc::new(InMemory::new()),
prefix: "backups".into(),
};
// No snapshots yet -> error (and no file written to dest).
let dest = TempDb::new("restore-dest");
assert!(
restore_latest(&target, dest.path()).await.is_err(),
"restore should fail when no snapshots exist"
);
// First snapshot (100 rows).
snapshot_and_upload(src.path(), &target).await.unwrap();
// Mutate, then a second snapshot (101 rows) -> a lexically-newer key.
{
let db = Builder::new_local(src.path()).build().await.unwrap();
let conn = db.connect().unwrap();
conn.execute("INSERT INTO t (id, v) VALUES (?, ?)", (10_000, "new"))
.await
.unwrap();
}
let newest_key = match snapshot_and_upload(src.path(), &target).await.unwrap() {
SnapshotOutcome::Uploaded { key, .. } => key,
other => panic!("expected Uploaded, got {other:?}"),
};
// restore_latest reports it pulled the newest snapshot...
let restored_from = restore_latest(&target, dest.path()).await.unwrap();
assert_eq!(restored_from, newest_key, "should restore the newest snapshot");
// ...the on-disk file is a valid vanilla-SQLite database...
let on_disk = std::fs::read(dest.path()).unwrap();
assert!(
on_disk.starts_with(b"SQLite format 3\0"),
"restored file is not a SQLite database"
);
// ...and re-opens through turso with the post-mutation row count.
assert_eq!(count_rows(dest.path()).await, 101);
}
}