use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::Arc;
use autumn_web::alerts::{Alert, AlertChannel, AlertCondition};
use crate::db::s3::{self, S3Client, S3Config, S3Credentials};
use crate::migrate;
const PG_BIN_DIR_ENV: &str = "AUTUMN_PG_BIN_DIR";
const MANAGED_PG_DATA_DIR_ENV: &str = "AUTUMN_MANAGED_PG_DATA_DIR";
const DEFAULT_BACKUP_DIR: &str = "backups";
const MAX_RUN_DIR_ATTEMPTS: usize = 1000;
const PLAIN_COMPLETE_MARKER: &str = "PostgreSQL database dump complete";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum BackupFormat {
#[default]
Custom,
Plain,
}
impl BackupFormat {
const fn extension(self) -> &'static str {
match self {
Self::Custom => "dump",
Self::Plain => "sql",
}
}
const fn pg_dump_format_flag(self) -> &'static str {
match self {
Self::Custom => "custom",
Self::Plain => "plain",
}
}
pub fn parse(s: &str) -> Result<Self, String> {
match s.trim().to_ascii_lowercase().as_str() {
"custom" | "c" => Ok(Self::Custom),
"plain" | "sql" | "p" => Ok(Self::Plain),
other => Err(format!(
"unknown backup format {other:?} (expected `custom` or `plain`)"
)),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TargetSelector {
All,
ControlOnly,
Shard(String),
}
#[derive(Debug, Clone)]
pub struct BackupArgs {
pub profile: Option<String>,
pub dir: Option<PathBuf>,
pub format: BackupFormat,
pub keep: Option<usize>,
pub target: TargetSelector,
pub upload: bool,
}
#[derive(Debug, Clone)]
pub struct RestoreArgs {
pub artifact: PathBuf,
pub profile: Option<String>,
pub force: bool,
pub shard: Option<String>,
pub offsite: bool,
}
#[derive(Debug)]
pub enum BackupError {
NoUrl,
UnknownShard { name: String, known: Vec<String> },
ToolMissing { tool: String },
ToolFailed { tool: String, context: String },
Io {
context: String,
source: std::io::Error,
},
IntegrityFailed { detail: String },
BadArtifact { detail: String },
ProductionRefused { profile: String },
OffsiteNotConfigured,
OffsiteConfig { detail: String },
SharedBucketRefused { bucket: String },
MissingCredentialEnv { var: String },
OffsiteUploadFailed { local_path: String, detail: String },
Offsite { op: &'static str, detail: String },
InvalidOffsiteRef { reference: String },
}
impl std::fmt::Display for BackupError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::NoUrl => write!(
f,
"No database URL found.\n Set database.primary_url (or database.url) in autumn.toml, \
or set AUTUMN_DATABASE__PRIMARY_URL / AUTUMN_DATABASE__URL / DATABASE_URL."
),
Self::UnknownShard { name, known } => {
let detail = if known.is_empty() {
"No [[database.shards]] entries found in autumn.toml or environment.".to_owned()
} else {
format!("Known shards: {}", known.join(", "))
};
write!(f, "Unknown shard {name:?}.\n {detail}")
}
Self::ToolMissing { tool } => write!(
f,
"`{tool}` was not found.\n Install the PostgreSQL client tools, set \
{PG_BIN_DIR_ENV} to their directory, or run against a managed-Postgres app \
(which bundles them)."
),
Self::ToolFailed { tool, context } => {
write!(f, "`{tool}` failed: {context}")
}
Self::Io { context, source } => write!(f, "{context}: {source}"),
Self::IntegrityFailed { detail } => write!(
f,
"Backup integrity check failed: {detail}\n The artifact was NOT reported as \
successful; any partial files were removed."
),
Self::BadArtifact { detail } => write!(f, "Cannot restore: {detail}"),
Self::ProductionRefused { profile } => write!(
f,
"Refusing to restore over the {profile:?} profile database.\n \
Re-run with --force if you really mean it (this overwrites data)."
),
Self::OffsiteNotConfigured => write!(
f,
"No offsite destination is configured.\n Add a [backup.offsite] section \
(with [backup.offsite.s3] bucket/region/endpoint and *_env credential \
indirection) to autumn.toml, or set AUTUMN_BACKUP__OFFSITE__* env vars."
),
Self::OffsiteConfig { detail } => {
write!(f, "Invalid [backup.offsite] configuration: {detail}")
}
Self::SharedBucketRefused { bucket } => write!(
f,
"The offsite destination bucket {bucket:?} is the same as the app's \
[storage.s3] bucket at the same endpoint.\n A shared bucket is a deliberate \
choice: set backup.offsite.allow_shared_bucket = true to opt in, or point the \
offsite backup at a distinct bucket."
),
Self::MissingCredentialEnv { var } => write!(
f,
"Offsite credential environment variable {var:?} is not set (or the config \
names no *_env variable).\n Set it to the S3 access key / secret, or fix \
backup.offsite.s3.access_key_id_env / secret_access_key_env."
),
Self::OffsiteUploadFailed { local_path, detail } => write!(
f,
"Local backup OK at {local_path}\n OFFSITE UPLOAD FAILED: {detail}\n \
The local artifact is intact — re-run `autumn db backup --upload` after \
fixing the offsite destination."
),
Self::Offsite { op, detail } => write!(f, "Offsite {op} failed: {detail}"),
Self::InvalidOffsiteRef { reference } => write!(
f,
"Invalid offsite reference {reference:?}.\n Expected \
offsite:<profile>/<timestamp|latest> (for example offsite:prod/latest)."
),
}
}
}
impl BackupError {
fn io(context: impl Into<String>) -> impl FnOnce(std::io::Error) -> Self {
let context = context.into();
move |source| Self::Io { context, source }
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ResolvedTarget {
label: String,
url: String,
}
pub fn run_backup(args: &BackupArgs) {
eprintln!("\u{1F342} autumn db backup\n");
if let Err(e) = backup(args) {
eprintln!("\u{2717} {e}");
std::process::exit(1);
}
}
pub fn run_restore(args: &RestoreArgs) {
eprintln!("\u{1F342} autumn db restore\n");
if let Err(e) = restore(args) {
eprintln!("\u{2717} {e}");
std::process::exit(1);
}
}
fn backup(args: &BackupArgs) -> Result<(), BackupError> {
let targets = resolve_targets(args.profile.as_deref(), &args.target)?;
let tools = PgTools::locate_with_extra(managed_pg_data_dir());
let pg_dump = tools.require("pg_dump")?;
let profile = migrate::effective_profile(args.profile.as_deref());
let should_upload = args.upload || auto_upload_probe(&profile);
let upload_ctx = if should_upload {
let offsite = load_offsite(&profile)?
.ok_or(BackupError::OffsiteNotConfigured)?
.resolve()?;
let client = offsite.build_client()?;
Some((offsite, client))
} else {
None
};
let root = backup_root(args.dir.as_deref(), &profile);
let run_dir = create_unique_run_dir(&root, &run_dir_name(&now_utc()))?;
let result = backup_into(&run_dir, &targets, args.format, &pg_dump, &tools, &profile);
if let Err(e) = result {
let _ = std::fs::remove_dir_all(&run_dir);
return Err(e);
}
eprintln!(
"\n\u{2713} Backup complete: {} ({} target(s), {} format).",
run_dir.display(),
targets.len(),
args.format.pg_dump_format_flag(),
);
if let Some(keep) = args.keep {
prune(&root, keep)?;
}
if let Some((offsite, client)) = upload_ctx {
let local_run_id = run_dir
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
let remote_id = remote_run_id(&local_run_id, &remote_run_token());
let remote_profile = migrate::canonical_profile(&profile);
if let Err(detail) = upload_run(&offsite, &client, &run_dir, &remote_profile, &remote_id) {
let local_path = run_dir.display().to_string();
emit_offsite_upload_failure_alert(&local_path, &detail);
return Err(BackupError::OffsiteUploadFailed { local_path, detail });
}
}
Ok(())
}
fn emit_offsite_upload_failure_alert(local_path: &str, detail: &str) {
let config = match autumn_web::config::AutumnConfig::load() {
Ok(c) => c,
Err(e) => {
eprintln!(" \u{26A0} offsite-upload-failed alert skipped: could not load config: {e}");
return;
}
};
let client = autumn_web::http::Client::from_config(&config.http.client);
let channels = crate::alert::configured_http_channels(&config.alerts, &client);
deliver_offsite_upload_alert(&channels, local_path, detail);
}
fn build_offsite_upload_alert(local_path: &str, detail: &str) -> Alert {
Alert::trigger(
AlertCondition::ScheduledTaskFailure,
"scheduled_task_failure:db-backup-offsite-upload",
)
.title("Offsite backup upload failed")
.summary(detail)
.detail("local_path", local_path)
.detail("error", detail)
.build()
}
fn deliver_offsite_upload_alert(
channels: &[Arc<dyn AlertChannel>],
local_path: &str,
detail: &str,
) {
if channels.is_empty() {
return;
}
let alert = build_offsite_upload_alert(local_path, detail);
let runtime = match tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => {
eprintln!(
" \u{26A0} offsite-upload-failed alert not sent: could not start runtime: {e}"
);
return;
}
};
runtime.block_on(async {
for channel in channels {
if let Err(error) = channel.deliver(&alert).await {
eprintln!(
" \u{26A0} offsite-upload-failed alert not delivered via {}: {error}",
channel.name()
);
}
}
});
}
fn backup_into(
run_dir: &Path,
targets: &[ResolvedTarget],
format: BackupFormat,
pg_dump: &Path,
tools: &PgTools,
profile: &str,
) -> Result<(), BackupError> {
let mut manifest_targets = Vec::with_capacity(targets.len());
for target in targets {
let file_name = artifact_file_name(&target.label, format);
let out_path = run_dir.join(&file_name);
eprintln!(
"\u{2500}\u{2500} backing up {} \u{2500}\u{2500}",
target.label
);
let db = parsed_db_name(&target.url);
run_pg_dump(pg_dump, &target.url, &out_path, format, &db)?;
verify_artifact(&out_path, format, &db, tools)?;
eprintln!(" \u{2713} {file_name} verified.");
manifest_targets.push(ManifestTarget {
label: target.label.clone(),
file: file_name,
database: db,
});
}
let manifest = Manifest {
autumn_version: env!("CARGO_PKG_VERSION").to_owned(),
created_at: now_utc().to_rfc3339(),
profile: profile.to_owned(),
format: format.pg_dump_format_flag().to_owned(),
targets: manifest_targets,
};
write_manifest(run_dir, &manifest)
}
fn pg_command(program: &Path, url: &str) -> (Command, String) {
let (safe_url, password) = split_password(url);
let mut cmd = Command::new(program);
if let Some(pw) = password {
cmd.env("PGPASSWORD", pw);
}
(cmd, safe_url)
}
fn split_password(url: &str) -> (String, Option<String>) {
if let Ok(mut parsed) = url::Url::parse(url) {
let password = parsed.password().map(|pw| {
percent_encoding::percent_decode_str(pw)
.decode_utf8_lossy()
.into_owned()
});
if password.is_some() {
let _ = parsed.set_password(None);
}
return (parsed.to_string(), password);
}
split_password_kv(url)
}
fn split_password_kv(conn: &str) -> (String, Option<String>) {
let Some(pairs) = parse_libpq_kv(conn) else {
return (conn.to_owned(), None);
};
let mut password = None;
let mut kept = Vec::with_capacity(pairs.len());
for (key, value) in pairs {
if key == "password" {
password = Some(value);
} else {
kept.push((key, value));
}
}
if password.is_none() {
return (conn.to_owned(), None);
}
let rebuilt = kept
.into_iter()
.map(|(k, v)| format!("{k}={}", quote_libpq_value(&v)))
.collect::<Vec<_>>()
.join(" ");
(rebuilt, password)
}
fn parse_libpq_kv(conn: &str) -> Option<Vec<(String, String)>> {
let chars: Vec<char> = conn.chars().collect();
let len = chars.len();
let mut i = 0;
let mut pairs = Vec::new();
loop {
while i < len && chars[i].is_whitespace() {
i += 1;
}
if i == len {
break;
}
let start = i;
while i < len && !chars[i].is_whitespace() && chars[i] != '=' {
i += 1;
}
if i == start {
return None;
}
let keyword: String = chars[start..i].iter().collect();
while i < len && chars[i].is_whitespace() {
i += 1;
}
if i == len || chars[i] != '=' {
return None;
}
i += 1; while i < len && chars[i].is_whitespace() {
i += 1;
}
let mut value = String::new();
if i < len && chars[i] == '\'' {
i += 1;
loop {
if i == len {
return None; }
match chars[i] {
'\\' => {
i += 1;
if i < len {
value.push(chars[i]);
i += 1;
}
}
'\'' => {
i += 1;
break;
}
c => {
value.push(c);
i += 1;
}
}
}
} else {
while i < len && !chars[i].is_whitespace() {
if chars[i] == '\\' {
i += 1;
if i < len {
value.push(chars[i]);
i += 1;
}
} else {
value.push(chars[i]);
i += 1;
}
}
}
pairs.push((keyword, value));
}
(!pairs.is_empty()).then_some(pairs)
}
fn quote_libpq_value(value: &str) -> String {
let needs_quoting = value.is_empty()
|| value
.chars()
.any(|c| c.is_whitespace() || c == '\'' || c == '\\');
if !needs_quoting {
return value.to_owned();
}
let mut out = String::with_capacity(value.len() + 2);
out.push('\'');
for c in value.chars() {
if c == '\'' || c == '\\' {
out.push('\\');
}
out.push(c);
}
out.push('\'');
out
}
fn pg_dump_args(format: BackupFormat, out_path: &Path, dbname: &str) -> Vec<std::ffi::OsString> {
use std::ffi::OsString;
let mut args: Vec<OsString> = vec![
"--format".into(),
format.pg_dump_format_flag().into(),
"--no-owner".into(),
"--no-privileges".into(),
];
if matches!(format, BackupFormat::Plain) {
args.push("--clean".into());
args.push("--if-exists".into());
}
args.push("--file".into());
args.push(out_path.as_os_str().to_owned());
args.push("--dbname".into());
args.push(dbname.into());
args
}
fn run_pg_dump(
pg_dump: &Path,
url: &str,
out_path: &Path,
format: BackupFormat,
db: &str,
) -> Result<(), BackupError> {
let (mut cmd, safe_url) = pg_command(pg_dump, url);
let status = cmd
.args(pg_dump_args(format, out_path, &safe_url))
.status()
.map_err(|e| match e.kind() {
std::io::ErrorKind::NotFound => BackupError::ToolMissing {
tool: "pg_dump".to_owned(),
},
_ => BackupError::ToolFailed {
tool: "pg_dump".to_owned(),
context: format!("could not spawn for database {db:?}: {e}"),
},
})?;
if !status.success() {
return Err(BackupError::ToolFailed {
tool: "pg_dump".to_owned(),
context: format!(
"dump of database {db:?} exited {}",
exit_desc(status.code())
),
});
}
Ok(())
}
fn verify_artifact(
path: &Path,
format: BackupFormat,
db: &str,
tools: &PgTools,
) -> Result<(), BackupError> {
let len = std::fs::metadata(path)
.map_err(BackupError::io(format!("stat {}", path.display())))?
.len();
if len == 0 {
return Err(BackupError::IntegrityFailed {
detail: format!("dump of {db:?} is empty (0 bytes)"),
});
}
match format {
BackupFormat::Custom => {
let pg_restore = tools.require("pg_restore")?;
verify_custom_archive(&pg_restore, path, db)
}
BackupFormat::Plain => verify_plain_dump(path, db),
}
}
fn verify_custom_archive(pg_restore: &Path, path: &Path, db: &str) -> Result<(), BackupError> {
let output = Command::new(pg_restore)
.arg("--list")
.arg(path)
.output()
.map_err(|e| match e.kind() {
std::io::ErrorKind::NotFound => BackupError::ToolMissing {
tool: "pg_restore".to_owned(),
},
_ => BackupError::ToolFailed {
tool: "pg_restore".to_owned(),
context: format!("could not spawn to verify {db:?}: {e}"),
},
})?;
if !output.status.success() {
return Err(BackupError::IntegrityFailed {
detail: format!(
"`pg_restore --list` rejected the {db:?} archive ({})",
exit_desc(output.status.code())
),
});
}
let listing = String::from_utf8_lossy(&output.stdout);
if toc_entry_count(&listing) == 0 {
return Err(BackupError::IntegrityFailed {
detail: format!("the {db:?} archive contains no objects"),
});
}
Ok(())
}
fn verify_plain_dump(path: &Path, db: &str) -> Result<(), BackupError> {
use std::io::{Read, Seek, SeekFrom};
let mut file =
std::fs::File::open(path).map_err(BackupError::io(format!("open {}", path.display())))?;
let len = file
.metadata()
.map_err(BackupError::io(format!("stat {}", path.display())))?
.len();
if len == 0 {
return Err(BackupError::IntegrityFailed {
detail: format!("the {db:?} SQL dump is empty (likely truncated)"),
});
}
let read_len = len.min(1024);
let offset = -i64::try_from(read_len).expect("read_len is at most 1024");
file.seek(SeekFrom::End(offset))
.map_err(BackupError::io(format!("seek {}", path.display())))?;
let mut tail = Vec::with_capacity(read_len as usize);
file.take(read_len)
.read_to_end(&mut tail)
.map_err(BackupError::io(format!("read {}", path.display())))?;
let tail = String::from_utf8_lossy(&tail);
if plain_dump_is_complete(&tail) {
Ok(())
} else {
Err(BackupError::IntegrityFailed {
detail: format!(
"the {db:?} SQL dump is missing pg_dump's completion marker (likely truncated)"
),
})
}
}
fn restore(args: &RestoreArgs) -> Result<(), BackupError> {
let profile = migrate::effective_profile(args.profile.as_deref());
super::guard_destructive(&profile, args.force).map_err(|_| BackupError::ProductionRefused {
profile: profile.clone(),
})?;
let artifact_str = args.artifact.to_string_lossy();
if let Some(oref) = resolve_offsite_ref(args.offsite, &artifact_str) {
return restore_from_offsite(args, &oref?);
}
let plan = RestorePlan::discover(&args.artifact, args.shard.as_deref())?;
apply_restore_plan(&plan, args)
}
fn apply_restore_plan(plan: &RestorePlan, args: &RestoreArgs) -> Result<(), BackupError> {
let format = plan.format;
let entries = plan.select(args.shard.as_deref())?;
let targets = resolve_targets_for_restore(args.profile.as_deref(), &entries)?;
let tools = PgTools::locate_with_extra(managed_pg_data_dir());
for (entry, _url) in &targets {
let db = format!("(artifact) {}", entry.label);
verify_artifact(&plan.dir.join(&entry.file), format, &db, &tools)?;
eprintln!(" \u{2713} {} integrity verified.", entry.file);
}
for (entry, url) in &targets {
eprintln!(
"\u{2500}\u{2500} restoring {} \u{2500}\u{2500}",
entry.label
);
let artifact = plan.dir.join(&entry.file);
let db = parsed_db_name(url);
run_restore_one(&tools, url, &artifact, format, &db)?;
eprintln!(" \u{2713} restored {}.", entry.label);
}
eprintln!("\n\u{2713} Restore complete ({} target(s)).", targets.len());
Ok(())
}
fn run_restore_one(
tools: &PgTools,
url: &str,
artifact: &Path,
format: BackupFormat,
db: &str,
) -> Result<(), BackupError> {
match format {
BackupFormat::Custom => {
let pg_restore = tools.require("pg_restore")?;
let (mut cmd, safe_url) = pg_command(&pg_restore, url);
let status = cmd
.arg("--clean")
.arg("--if-exists")
.arg("--no-owner")
.arg("--dbname")
.arg(&safe_url)
.arg(artifact)
.status()
.map_err(|e| spawn_err("pg_restore", db, &e))?;
if !status.success() {
return Err(BackupError::ToolFailed {
tool: "pg_restore".to_owned(),
context: format!(
"restore into database {db:?} exited {}",
exit_desc(status.code())
),
});
}
Ok(())
}
BackupFormat::Plain => {
let psql = tools.require("psql")?;
let (mut cmd, safe_url) = pg_command(&psql, url);
let status = cmd
.arg("--set")
.arg("ON_ERROR_STOP=1")
.arg("--dbname")
.arg(&safe_url)
.arg("--file")
.arg(artifact)
.status()
.map_err(|e| spawn_err("psql", db, &e))?;
if !status.success() {
return Err(BackupError::ToolFailed {
tool: "psql".to_owned(),
context: format!(
"restore into database {db:?} exited {}",
exit_desc(status.code())
),
});
}
Ok(())
}
}
}
fn spawn_err(tool: &str, db: &str, e: &std::io::Error) -> BackupError {
match e.kind() {
std::io::ErrorKind::NotFound => BackupError::ToolMissing {
tool: tool.to_owned(),
},
_ => BackupError::ToolFailed {
tool: tool.to_owned(),
context: format!("could not spawn for database {db:?}: {e}"),
},
}
}
struct RestorePlan {
dir: PathBuf,
format: BackupFormat,
entries: Vec<ManifestTarget>,
}
impl RestorePlan {
fn discover(path: &Path, shard: Option<&str>) -> Result<Self, BackupError> {
if path.is_dir() {
let manifest = read_manifest(path)?;
let format = BackupFormat::parse(&manifest.format).map_err(|detail| {
BackupError::BadArtifact {
detail: format!("manifest has an invalid format: {detail}"),
}
})?;
return Ok(Self {
dir: path.to_path_buf(),
format,
entries: manifest.targets,
});
}
if path.is_file() {
let format = infer_format_from_path(path).ok_or_else(|| BackupError::BadArtifact {
detail: format!(
"{} is neither a run directory nor a .dump/.sql artifact",
path.display()
),
})?;
let dir = path
.parent()
.map_or_else(|| PathBuf::from("."), Path::to_path_buf);
let file = path
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
let label = single_file_target_label(&file, shard)?;
return Ok(Self {
dir,
format,
entries: vec![ManifestTarget {
label,
file,
database: String::new(),
}],
});
}
Err(BackupError::BadArtifact {
detail: format!("{} does not exist", path.display()),
})
}
fn select(&self, shard: Option<&str>) -> Result<Vec<ManifestTarget>, BackupError> {
let Some(shard) = shard else {
return Ok(self.entries.clone());
};
let wanted = format!("shard:{shard}");
let selected: Vec<ManifestTarget> = self
.entries
.iter()
.filter(|t| t.label == wanted)
.cloned()
.collect();
if selected.is_empty() {
let known: Vec<String> = self.entries.iter().map(|t| t.label.clone()).collect();
return Err(BackupError::UnknownShard {
name: shard.to_owned(),
known,
});
}
Ok(selected)
}
}
fn dotenv_env() -> autumn_web::dotenv::DotenvOsEnv {
match autumn_web::dotenv::os_env_with_dotenv() {
Ok(env) => env,
Err(e) => {
eprintln!(" \u{274C} .env: {e}");
std::process::exit(1);
}
}
}
fn dotenv_env_for_profile(profile: &str) -> autumn_web::dotenv::DotenvOsEnv {
let profile = migrate::canonical_profile(profile);
match autumn_web::dotenv::os_env_with_dotenv_for_profile(&profile) {
Ok(env) => env,
Err(e) => {
eprintln!(" \u{274C} .env: {e}");
std::process::exit(1);
}
}
}
fn managed_pg_fallback_url() -> Option<String> {
crate::serve::managed_pg_env(None).and_then(|env| env.attach_url)
}
fn managed_pg_data_dir() -> Option<PathBuf> {
crate::serve::managed_pg_env(None).map(|env| env.data_dir)
}
fn resolve_control_url<Ff>(resolved: Option<String>, fallback: Ff) -> Option<String>
where
Ff: FnOnce() -> Option<String>,
{
resolved.or_else(fallback)
}
fn resolve_targets(
profile: Option<&str>,
selector: &TargetSelector,
) -> Result<Vec<ResolvedTarget>, BackupError> {
use autumn_web::config::Env as _;
let table =
migrate::read_autumn_toml_table_with_profile(Some(&migrate::effective_profile(profile)));
let env = dotenv_env();
let control = resolve_control_url(
migrate::resolve_primary_database_url_from_sources(|k| env.var(k), table.as_ref()),
managed_pg_fallback_url,
);
let shards = migrate::resolve_shard_database_urls_from_sources(|k| env.var(k), table.as_ref());
build_targets(control, shards, selector)
}
fn build_targets(
control: Option<String>,
shards: Vec<(String, String)>,
selector: &TargetSelector,
) -> Result<Vec<ResolvedTarget>, BackupError> {
match selector {
TargetSelector::ControlOnly => control
.map(|url| {
vec![ResolvedTarget {
label: "control".to_owned(),
url,
}]
})
.ok_or(BackupError::NoUrl),
TargetSelector::Shard(name) => {
let Some((_, url)) = shards.iter().find(|(shard, _)| shard == name) else {
return Err(BackupError::UnknownShard {
name: name.clone(),
known: shards.into_iter().map(|(n, _)| n).collect(),
});
};
Ok(vec![ResolvedTarget {
label: format!("shard:{name}"),
url: url.clone(),
}])
}
TargetSelector::All => {
let mut targets = Vec::new();
if let Some(control_url) = control {
targets.push(ResolvedTarget {
label: "control".to_owned(),
url: control_url,
});
} else if shards.is_empty() {
return Err(BackupError::NoUrl);
}
for (name, url) in shards {
targets.push(ResolvedTarget {
label: format!("shard:{name}"),
url,
});
}
Ok(targets)
}
}
}
fn resolve_targets_for_restore(
profile: Option<&str>,
entries: &[ManifestTarget],
) -> Result<Vec<(ManifestTarget, String)>, BackupError> {
use autumn_web::config::Env as _;
let table =
migrate::read_autumn_toml_table_with_profile(Some(&migrate::effective_profile(profile)));
let env = dotenv_env();
let control = resolve_control_url(
migrate::resolve_primary_database_url_from_sources(|k| env.var(k), table.as_ref()),
managed_pg_fallback_url,
);
let shards = migrate::resolve_shard_database_urls_from_sources(|k| env.var(k), table.as_ref());
let mut out = Vec::with_capacity(entries.len());
for entry in entries {
let url = if entry.label == "control" {
control.clone().ok_or(BackupError::NoUrl)?
} else if let Some(shard_name) = entry.label.strip_prefix("shard:") {
shards
.iter()
.find(|(n, _)| n == shard_name)
.map(|(_, u)| u.clone())
.ok_or_else(|| BackupError::UnknownShard {
name: shard_name.to_owned(),
known: shards.iter().map(|(n, _)| n.clone()).collect(),
})?
} else {
return Err(BackupError::BadArtifact {
detail: format!(
"manifest target has an unrecognized label {:?}",
entry.label
),
});
};
out.push((entry.clone(), url));
}
Ok(out)
}
fn prune(root: &Path, keep: usize) -> Result<(), BackupError> {
let mut runs = list_run_dirs(root)?;
runs.sort(); let to_remove = plan_pruning(&runs, keep);
for name in &to_remove {
let path = root.join(name);
std::fs::remove_dir_all(&path)
.map_err(BackupError::io(format!("pruning {}", path.display())))?;
eprintln!(" \u{1F5D1} pruned old backup {name}");
}
if !to_remove.is_empty() {
eprintln!(
" \u{2139} retention: kept newest {keep}, pruned {}.",
to_remove.len()
);
}
Ok(())
}
fn list_run_dirs(root: &Path) -> Result<Vec<String>, BackupError> {
if !root.exists() {
return Ok(Vec::new());
}
let mut names = Vec::new();
for entry in
std::fs::read_dir(root).map_err(BackupError::io(format!("read {}", root.display())))?
{
let entry = entry.map_err(BackupError::io(format!("read entry in {}", root.display())))?;
let path = entry.path();
if path.is_dir() && path.join(MANIFEST_FILE).is_file() {
names.push(entry.file_name().to_string_lossy().into_owned());
}
}
Ok(names)
}
fn plan_pruning(sorted_runs: &[String], keep: usize) -> Vec<String> {
let keep = keep.max(1);
if sorted_runs.len() <= keep {
return Vec::new();
}
let remove_count = sorted_runs.len() - keep;
sorted_runs[..remove_count].to_vec()
}
pub struct PgTools {
dirs: Vec<PathBuf>,
}
impl PgTools {
pub fn locate() -> Self {
Self::locate_with_extra(None)
}
pub fn locate_with_extra(managed_data_dir: Option<PathBuf>) -> Self {
let mut dirs = candidate_bin_dirs(
std::env::var_os(PG_BIN_DIR_ENV).map(PathBuf::from),
std::env::var_os(MANAGED_PG_DATA_DIR_ENV).map(PathBuf::from),
);
dirs.extend(candidate_bin_dirs(None, managed_data_dir));
Self { dirs }
}
#[cfg(test)]
pub const fn with_dirs(dirs: Vec<PathBuf>) -> Self {
Self { dirs }
}
fn resolve(&self, tool: &str) -> PathBuf {
resolve_tool_in(&self.dirs, tool)
}
pub fn require(&self, tool: &str) -> Result<PathBuf, BackupError> {
let resolved = self.resolve(tool);
if resolved.is_absolute() {
return Ok(resolved);
}
if tool_on_path(tool) {
Ok(resolved)
} else {
Err(BackupError::ToolMissing {
tool: tool.to_owned(),
})
}
}
}
fn candidate_bin_dirs(
pg_bin_dir: Option<PathBuf>,
managed_data_dir: Option<PathBuf>,
) -> Vec<PathBuf> {
let mut dirs = Vec::new();
if let Some(dir) = pg_bin_dir {
dirs.push(dir);
}
if let Some(data_dir) = managed_data_dir {
let install = data_dir.parent().unwrap_or(&data_dir).join("postgresql");
dirs.push(install.join("bin"));
dirs.extend(nested_bin_dirs(&install));
}
dirs
}
fn nested_bin_dirs(install: &Path) -> Vec<PathBuf> {
let mut out = Vec::new();
if let Ok(entries) = std::fs::read_dir(install) {
for entry in entries.flatten() {
if entry.path().is_dir() {
out.push(entry.path().join("bin"));
}
}
}
out
}
fn resolve_tool_in(dirs: &[PathBuf], tool: &str) -> PathBuf {
let exe = exe_name(tool);
for dir in dirs {
let candidate = dir.join(&exe);
if candidate.is_file() {
return candidate;
}
}
PathBuf::from(tool)
}
fn exe_name(tool: &str) -> String {
if cfg!(windows) {
format!("{tool}.exe")
} else {
tool.to_owned()
}
}
fn tool_on_path(tool: &str) -> bool {
Command::new(tool)
.arg("--version")
.output()
.is_ok_and(|o| o.status.success())
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
struct ManifestTarget {
label: String,
file: String,
#[serde(default)]
database: String,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
struct Manifest {
autumn_version: String,
created_at: String,
profile: String,
format: String,
targets: Vec<ManifestTarget>,
}
const MANIFEST_FILE: &str = "manifest.json";
fn write_manifest(run_dir: &Path, manifest: &Manifest) -> Result<(), BackupError> {
let json = serde_json::to_string_pretty(manifest).map_err(|e| BackupError::Io {
context: "serializing manifest".to_owned(),
source: std::io::Error::other(e),
})?;
let path = run_dir.join(MANIFEST_FILE);
std::fs::write(&path, json).map_err(BackupError::io(format!("writing {}", path.display())))
}
fn read_manifest(run_dir: &Path) -> Result<Manifest, BackupError> {
let path = run_dir.join(MANIFEST_FILE);
let json = std::fs::read_to_string(&path).map_err(|e| BackupError::BadArtifact {
detail: format!("{} has no readable {MANIFEST_FILE}: {e}", run_dir.display()),
})?;
serde_json::from_str(&json).map_err(|e| BackupError::BadArtifact {
detail: format!("{} is not a valid manifest: {e}", path.display()),
})
}
fn backup_root(dir: Option<&Path>, profile: &str) -> PathBuf {
let base = dir.map_or_else(|| PathBuf::from(DEFAULT_BACKUP_DIR), Path::to_path_buf);
base.join(sanitize_component(profile))
}
fn run_dir_name(now: &chrono::DateTime<chrono::Utc>) -> String {
now.format("%Y%m%dT%H%M%SZ").to_string()
}
fn create_unique_run_dir(root: &Path, base: &str) -> Result<PathBuf, BackupError> {
std::fs::create_dir_all(root)
.map_err(BackupError::io(format!("creating {}", root.display())))?;
for attempt in 1..=MAX_RUN_DIR_ATTEMPTS {
let name = if attempt == 1 {
base.to_owned()
} else {
format!("{base}-{attempt}")
};
let candidate = root.join(&name);
match std::fs::create_dir(&candidate) {
Ok(()) => return Ok(candidate),
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(e) => {
return Err(BackupError::io(format!("creating {}", candidate.display()))(e));
}
}
}
Err(BackupError::Io {
context: format!("creating a unique run directory under {}", root.display()),
source: std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
format!("exhausted {MAX_RUN_DIR_ATTEMPTS} same-timestamp attempts for {base:?}"),
),
})
}
fn artifact_file_name(label: &str, format: BackupFormat) -> String {
let ext = format.extension();
if label == "control" {
format!("control.{ext}")
} else if let Some(name) = label.strip_prefix("shard:") {
format!("shard-{}.{ext}", sanitize_component(name))
} else {
format!("{}.{ext}", sanitize_component(label))
}
}
fn sanitize_component(s: &str) -> String {
let cleaned: String = s
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.') {
c
} else {
'_'
}
})
.collect();
if cleaned.is_empty() || cleaned == "." || cleaned == ".." {
"unnamed".to_owned()
} else {
cleaned
}
}
fn single_file_target_label(file: &str, shard: Option<&str>) -> Result<String, BackupError> {
if let Some(name) = shard {
return Ok(format!("shard:{name}"));
}
let stem = Path::new(file)
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or(file);
if stem == "control" {
return Ok("control".to_owned());
}
if let Some(shard_name) = stem.strip_prefix("shard-") {
if shard_name.is_empty() {
return Err(BackupError::BadArtifact {
detail: format!(
"{file:?} looks like a shard artifact but its shard name is missing.\n \
Pass --shard <name>, or restore the run directory (with its manifest.json)."
),
});
}
return Ok(format!("shard:{shard_name}"));
}
Ok("control".to_owned())
}
fn infer_format_from_path(path: &Path) -> Option<BackupFormat> {
match path.extension().and_then(|e| e.to_str()) {
Some("dump") => Some(BackupFormat::Custom),
Some("sql") => Some(BackupFormat::Plain),
_ => None,
}
}
fn parsed_db_name(url: &str) -> String {
url::Url::parse(url)
.ok()
.and_then(|u| {
u.path_segments()
.and_then(|mut s| s.next().map(str::to_owned))
})
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "(database)".to_owned())
}
fn toc_entry_count(listing: &str) -> usize {
listing
.lines()
.filter(|l| {
let t = l.trim();
!t.is_empty() && !t.starts_with(';')
})
.count()
}
fn plain_dump_is_complete(contents: &str) -> bool {
contents.contains(PLAIN_COMPLETE_MARKER)
}
fn exit_desc(code: Option<i32>) -> String {
code.map_or_else(|| "with a signal".to_owned(), |c| format!("with code {c}"))
}
fn now_utc() -> chrono::DateTime<chrono::Utc> {
chrono::Utc::now()
}
struct ResolvedOffsite {
s3: S3Config,
prefix: String,
keep: Option<usize>,
allow_shared_bucket: bool,
access_key_id_env: Option<String>,
secret_access_key_env: Option<String>,
app_storage_bucket: Option<String>,
app_storage_endpoint: Option<String>,
profile: String,
}
impl ResolvedOffsite {
fn build_client(&self) -> Result<S3Client, BackupError> {
if !self.allow_shared_bucket
&& destinations_conflict(
&self.s3.bucket,
self.s3.endpoint.as_deref(),
self.app_storage_bucket.as_deref(),
self.app_storage_endpoint.as_deref(),
)
{
return Err(BackupError::SharedBucketRefused {
bucket: self.s3.bucket.clone(),
});
}
let creds = self.resolve_credentials()?;
let client = S3Client::new(self.s3.clone(), creds).map_err(|e| BackupError::Offsite {
op: "connect",
detail: e.to_string(),
})?;
Ok(match self.multipart_part_size_override() {
Some(bytes) => client.with_part_size(bytes),
None => client,
})
}
fn multipart_part_size_override(&self) -> Option<u64> {
use autumn_web::config::Env as _;
let env = dotenv_env_for_profile(&self.profile);
env.var("AUTUMN_OFFSITE_MULTIPART_PART_SIZE_BYTES")
.ok()
.and_then(|v: String| v.trim().parse::<u64>().ok())
.filter(|&n| n > 0)
}
fn resolve_credentials(&self) -> Result<S3Credentials, BackupError> {
let env = dotenv_env_for_profile(&self.profile);
let access_key_id = read_credential_env(&env, self.access_key_id_env.as_deref())?;
let secret_access_key = read_credential_env(&env, self.secret_access_key_env.as_deref())?;
Ok(S3Credentials {
access_key_id,
secret_access_key,
})
}
}
fn read_credential_env(
env: &dyn autumn_web::config::Env,
var: Option<&str>,
) -> Result<String, BackupError> {
let var = var
.map(str::trim)
.filter(|v| !v.is_empty())
.ok_or_else(|| BackupError::MissingCredentialEnv {
var: "(none configured)".to_owned(),
})?;
env.var(var)
.ok()
.filter(|v| !v.is_empty())
.ok_or_else(|| BackupError::MissingCredentialEnv {
var: var.to_owned(),
})
}
struct LoadedOffsite {
offsite: autumn_web::config::OffsiteBackupConfig,
app_storage_bucket: Option<String>,
app_storage_endpoint: Option<String>,
profile: String,
}
impl LoadedOffsite {
fn resolve(self) -> Result<ResolvedOffsite, BackupError> {
let Self {
offsite,
app_storage_bucket,
app_storage_endpoint,
profile,
} = self;
let bucket = offsite
.s3
.bucket
.as_deref()
.map(str::trim)
.filter(|b| !b.is_empty())
.map(str::to_owned)
.ok_or_else(|| BackupError::OffsiteConfig {
detail: "backup.offsite.s3.bucket is required".to_owned(),
})?;
let region = offsite
.s3
.region
.as_deref()
.map(str::trim)
.filter(|r| !r.is_empty())
.unwrap_or("us-east-1")
.to_owned();
let s3 = S3Config {
bucket,
region,
endpoint: offsite
.s3
.endpoint
.as_deref()
.map(str::trim)
.filter(|e| !e.is_empty())
.map(str::to_owned),
force_path_style: offsite.s3.force_path_style,
};
Ok(ResolvedOffsite {
s3,
prefix: normalize_offsite_prefix(offsite.prefix.as_deref()),
keep: offsite.keep,
allow_shared_bucket: offsite.allow_shared_bucket,
access_key_id_env: offsite.s3.access_key_id_env,
secret_access_key_env: offsite.s3.secret_access_key_env,
app_storage_bucket,
app_storage_endpoint,
profile,
})
}
}
fn auto_upload_probe(profile: &str) -> bool {
use autumn_web::config::Env as _;
let profile = migrate::canonical_profile(profile);
let env = dotenv_env_for_profile(&profile);
let table = migrate::read_autumn_toml_table_with_profile_from_config_dir(Some(&profile));
auto_upload_from_sources(|k| env.var(k).ok(), table.as_ref())
}
fn auto_upload_from_sources<F>(env_var: F, table: Option<&toml::Table>) -> bool
where
F: Fn(&str) -> Option<String>,
{
if let Some(raw) = env_var("AUTUMN_BACKUP__OFFSITE__AUTO_UPLOAD")
&& let Some(b) = parse_bool_flag(&raw)
{
return b;
}
table
.and_then(|t| t.get("backup"))
.and_then(|v| v.get("offsite"))
.and_then(|v| v.get("auto_upload"))
.and_then(toml::Value::as_bool)
.unwrap_or(false)
}
fn parse_bool_flag(raw: &str) -> Option<bool> {
match raw.trim().to_ascii_lowercase().as_str() {
"1" | "true" => Some(true),
"0" | "false" => Some(false),
_ => None,
}
}
fn load_offsite(profile: &str) -> Result<Option<LoadedOffsite>, BackupError> {
let profile = migrate::canonical_profile(profile);
let env = dotenv_env_for_profile(&profile);
let table = migrate::read_autumn_toml_table_with_profile_from_config_dir(Some(&profile));
resolve_loaded_offsite(table.as_ref(), &env, &profile)
}
fn resolve_loaded_offsite(
table: Option<&toml::Table>,
env: &dyn autumn_web::config::Env,
profile: &str,
) -> Result<Option<LoadedOffsite>, BackupError> {
let mut cfg = match table {
Some(t) => {
let toml_str = toml::to_string(t).map_err(|e| BackupError::Offsite {
op: "config",
detail: e.to_string(),
})?;
toml::from_str::<autumn_web::config::AutumnConfig>(&toml_str).map_err(|e| {
BackupError::Offsite {
op: "config",
detail: e.to_string(),
}
})?
}
None => autumn_web::config::AutumnConfig::default(),
};
cfg.apply_env_overrides_with_env(env);
let Some(offsite) = cfg.backup.offsite else {
return Ok(None);
};
let storage_is_s3 = cfg.storage.backend == autumn_web::storage::StorageBackend::S3;
let (app_storage_bucket, app_storage_endpoint) = if storage_is_s3 {
(
cfg.storage
.s3
.bucket
.as_deref()
.map(str::trim)
.filter(|b| !b.is_empty())
.map(str::to_owned),
cfg.storage
.s3
.endpoint
.as_deref()
.map(str::trim)
.filter(|e| !e.is_empty())
.map(str::to_owned),
)
} else {
(None, None)
};
Ok(Some(LoadedOffsite {
offsite: *offsite,
app_storage_bucket,
app_storage_endpoint,
profile: profile.to_owned(),
}))
}
pub fn destinations_conflict(
offsite_bucket: &str,
offsite_endpoint: Option<&str>,
app_bucket: Option<&str>,
app_endpoint: Option<&str>,
) -> bool {
let Some(app_bucket) = app_bucket else {
return false;
};
app_bucket == offsite_bucket
&& canonical_authority(offsite_endpoint, offsite_bucket)
== canonical_authority(app_endpoint, app_bucket)
}
fn canonical_authority(endpoint: Option<&str>, bucket: &str) -> String {
let raw = endpoint.unwrap_or("").trim();
if raw.is_empty() {
return "aws".to_owned();
}
let with_scheme = if raw.contains("://") {
raw.to_owned()
} else {
format!("https://{raw}")
};
let Ok(url) = url::Url::parse(&with_scheme) else {
return raw.to_ascii_lowercase();
};
let host = url.host_str().unwrap_or("").to_ascii_lowercase();
if host == "amazonaws.com" || host.ends_with(".amazonaws.com") {
return "aws".to_owned();
}
let bucket_prefix = format!("{}.", bucket.to_ascii_lowercase());
let host = host.strip_prefix(&bucket_prefix).unwrap_or(&host);
let scheme = url.scheme().to_ascii_lowercase();
let default_port = match scheme.as_str() {
"https" => Some(443),
"http" => Some(80),
_ => None,
};
match url.port() {
Some(port) if Some(port) != default_port => format!("{scheme}://{host}:{port}"),
_ => format!("{scheme}://{host}"),
}
}
fn normalize_offsite_prefix(prefix: Option<&str>) -> String {
prefix.unwrap_or("").trim().trim_matches('/').to_owned()
}
fn join_key(parts: &[&str]) -> String {
parts
.iter()
.filter(|p| !p.is_empty())
.copied()
.collect::<Vec<_>>()
.join("/")
}
fn offsite_object_key(prefix: &str, profile: &str, run_id: &str, file: &str) -> String {
join_key(&[prefix, profile, run_id, file])
}
fn remote_run_token() -> String {
use sha2::{Digest, Sha256};
let host = std::env::var("HOSTNAME")
.or_else(|_| std::env::var("COMPUTERNAME"))
.unwrap_or_default();
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_nanos());
let mut hasher = Sha256::new();
hasher.update(host.as_bytes());
hasher.update([0]); hasher.update(std::process::id().to_le_bytes());
hasher.update(nanos.to_le_bytes());
hex::encode(&hasher.finalize()[..4])
}
fn remote_run_id(local_run_id: &str, token: &str) -> String {
format!("{local_run_id}-{token}")
}
fn offsite_profile_prefix(prefix: &str, profile: &str) -> String {
format!("{}/", join_key(&[prefix, profile]))
}
fn offsite_run_prefix(prefix: &str, profile: &str, run_id: &str) -> String {
format!("{}/", join_key(&[prefix, profile, run_id]))
}
fn upload_run(
offsite: &ResolvedOffsite,
client: &S3Client,
run_dir: &Path,
profile: &str,
run_id: &str,
) -> Result<(), String> {
eprintln!("\u{2500}\u{2500} offsite upload \u{2500}\u{2500}");
let mut files = list_run_files(run_dir).map_err(|e| e.to_string())?;
files.sort();
files.sort_by_key(|f| f.as_str() == MANIFEST_FILE);
if files.is_empty() {
return Err("the completed run directory contains no files to upload".to_owned());
}
let mut written_keys: Vec<String> = Vec::new();
for file in &files {
let path = run_dir.join(file);
let key = offsite_object_key(&offsite.prefix, profile, run_id, file);
written_keys.push(key.clone());
if let Err(e) = client.put_file_and_verify(&key, &path) {
let detail = format!("{file}: {e}");
cleanup_failed_upload(client, &written_keys);
return Err(detail);
}
eprintln!(" \u{2713} uploaded + verified {file}");
}
eprintln!(
"\u{2713} Offsite upload verified: {} object(s) under {}",
files.len(),
offsite_run_prefix(&offsite.prefix, profile, run_id),
);
if let Some(keep) = offsite.keep
&& let Err(e) = prune_remote(offsite, client, profile, run_id, keep)
{
eprintln!(" \u{26A0} offsite retention skipped: {e}");
}
Ok(())
}
fn cleanup_failed_upload(client: &S3Client, written_keys: &[String]) {
for key in order_cleanup_keys(written_keys) {
let _ = client.delete_object(&key);
}
}
fn order_cleanup_keys(written_keys: &[String]) -> Vec<String> {
let mut keys = written_keys.to_vec();
keys.sort_by_key(|k| !key_is_manifest(k));
keys
}
fn key_is_manifest(key: &str) -> bool {
key.rsplit('/').next() == Some(MANIFEST_FILE)
}
fn list_run_files(run_dir: &Path) -> Result<Vec<String>, BackupError> {
let mut files = Vec::new();
for entry in std::fs::read_dir(run_dir)
.map_err(BackupError::io(format!("read {}", run_dir.display())))?
{
let entry = entry.map_err(BackupError::io(format!(
"read entry in {}",
run_dir.display()
)))?;
if entry.path().is_file() {
files.push(entry.file_name().to_string_lossy().into_owned());
}
}
Ok(files)
}
fn prune_remote(
offsite: &ResolvedOffsite,
client: &S3Client,
profile: &str,
keep_run_id: &str,
keep: usize,
) -> Result<(), String> {
let list_prefix = offsite_profile_prefix(&offsite.prefix, profile);
let objects = client
.list_objects_v2(&list_prefix)
.map_err(|e| e.to_string())?;
let run_ids = complete_remote_run_ids(&objects, &list_prefix);
let to_remove = plan_remote_pruning(&run_ids, keep, keep_run_id);
for run_id in &to_remove {
let run_prefix = offsite_run_prefix(&offsite.prefix, profile, run_id);
let run_objects = client
.list_objects_v2(&run_prefix)
.map_err(|e| e.to_string())?;
for obj in &run_objects {
client.delete_object(&obj.key).map_err(|e| e.to_string())?;
}
eprintln!(" \u{1F5D1} pruned remote backup {run_id}");
}
Ok(())
}
#[cfg(test)]
fn remote_run_ids(objects: &[s3::S3Object], list_prefix: &str) -> Vec<String> {
let mut ids: Vec<String> = objects
.iter()
.filter_map(|o| o.key.strip_prefix(list_prefix))
.filter_map(|rest| rest.split('/').next())
.filter(|s| !s.is_empty())
.map(str::to_owned)
.collect();
ids.sort();
ids.dedup();
ids
}
fn complete_remote_run_ids(objects: &[s3::S3Object], list_prefix: &str) -> Vec<String> {
let mut ids: Vec<String> = objects
.iter()
.filter_map(|o| o.key.strip_prefix(list_prefix))
.filter_map(|rest| {
let mut parts = rest.splitn(2, '/');
match (parts.next(), parts.next()) {
(Some(run_id), Some(file)) if !run_id.is_empty() && file == MANIFEST_FILE => {
Some(run_id.to_owned())
}
_ => None,
}
})
.collect();
ids.sort();
ids.dedup();
ids
}
fn plan_remote_pruning(sorted_run_ids: &[String], keep: usize, keep_run_id: &str) -> Vec<String> {
let keep = keep.max(1);
if sorted_run_ids.len() <= keep {
return Vec::new();
}
let remove_count = sorted_run_ids.len() - keep;
sorted_run_ids[..remove_count]
.iter()
.filter(|id| id.as_str() != keep_run_id)
.cloned()
.collect()
}
pub fn run_offsite_list(profile: Option<&str>) {
eprintln!("\u{1F342} autumn db offsite list\n");
if let Err(e) = offsite_list(profile) {
eprintln!("\u{2717} {e}");
std::process::exit(1);
}
}
fn offsite_list(profile: Option<&str>) -> Result<(), BackupError> {
let profile = migrate::canonical_profile(&migrate::effective_profile(profile));
let offsite = load_offsite(&profile)?
.ok_or(BackupError::OffsiteNotConfigured)?
.resolve()?;
let client = offsite.build_client()?;
let list_prefix = offsite_profile_prefix(&offsite.prefix, &profile);
let objects = client
.list_objects_v2(&list_prefix)
.map_err(|e| BackupError::Offsite {
op: "list",
detail: e.to_string(),
})?;
let runs = group_offsite_objects(&objects, &list_prefix);
if runs.is_empty() {
eprintln!(" \u{2139} No offsite backups found for profile {profile:?}.");
return Ok(());
}
print!("{}", format_offsite_listing(&runs));
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct OffsiteRun {
run_id: String,
files: Vec<(String, u64)>,
total: u64,
complete: bool,
}
fn group_offsite_objects(objects: &[s3::S3Object], list_prefix: &str) -> Vec<OffsiteRun> {
let mut map: std::collections::BTreeMap<String, OffsiteRun> = std::collections::BTreeMap::new();
for obj in objects {
let Some(rest) = obj.key.strip_prefix(list_prefix) else {
continue;
};
let mut parts = rest.splitn(2, '/');
let (Some(run_id), Some(file)) = (parts.next(), parts.next()) else {
continue;
};
if run_id.is_empty() || file.is_empty() {
continue;
}
let run = map.entry(run_id.to_owned()).or_insert_with(|| OffsiteRun {
run_id: run_id.to_owned(),
files: Vec::new(),
total: 0,
complete: false,
});
run.files.push((file.to_owned(), obj.size));
run.total += obj.size;
if file == MANIFEST_FILE {
run.complete = true;
}
}
map.into_values().collect()
}
fn format_offsite_listing(runs: &[OffsiteRun]) -> String {
use std::fmt::Write as _;
let mut out = String::new();
let _ = writeln!(out, "{:<28} {:>10} FILES", "RUN ID", "SIZE");
for run in runs {
let mut labels: Vec<&str> = run.files.iter().map(|(f, _)| f.as_str()).collect();
labels.sort_unstable();
let mut files = labels.join(", ");
if !run.complete {
files.push_str(" (INCOMPLETE — no manifest, not restorable)");
}
let _ = writeln!(
out,
"{:<28} {:>10} {}",
run.run_id,
human_size(run.total),
files,
);
}
out
}
fn human_size(bytes: u64) -> String {
const UNITS: [&str; 4] = ["B", "KiB", "MiB", "GiB"];
#[allow(clippy::cast_precision_loss)]
let mut size = bytes as f64;
let mut unit = 0;
while size >= 1024.0 && unit < UNITS.len() - 1 {
size /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{bytes} B")
} else {
format!("{size:.1} {}", UNITS[unit])
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct OffsiteRef {
profile: String,
selector: OffsiteSelector,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum OffsiteSelector {
Latest,
Exact(String),
}
fn resolve_offsite_ref(flag: bool, artifact: &str) -> Option<Result<OffsiteRef, BackupError>> {
let has_prefix = artifact.starts_with("offsite:");
if !flag && !has_prefix {
return None;
}
Some(
parse_offsite_ref(artifact).ok_or_else(|| BackupError::InvalidOffsiteRef {
reference: artifact.to_owned(),
}),
)
}
fn parse_offsite_ref(s: &str) -> Option<OffsiteRef> {
let s = s.strip_prefix("offsite:").unwrap_or(s).trim();
let (profile, selector) = s.split_once('/')?;
let profile = profile.trim();
let selector = selector.trim();
if profile.is_empty() || selector.is_empty() {
return None;
}
let selector = if selector.eq_ignore_ascii_case("latest") {
OffsiteSelector::Latest
} else {
OffsiteSelector::Exact(selector.to_owned())
};
Some(OffsiteRef {
profile: profile.to_owned(),
selector,
})
}
fn is_safe_leaf_name(name: &str) -> bool {
use std::path::{Component, Path};
if name.is_empty() || name.contains('/') || name.contains('\\') {
return false;
}
let p = Path::new(name);
p.components().count() == 1
&& matches!(p.components().next(), Some(Component::Normal(c)) if c == name)
}
fn restore_from_offsite(args: &RestoreArgs, oref: &OffsiteRef) -> Result<(), BackupError> {
let profile = migrate::canonical_profile(&oref.profile);
let offsite = load_offsite(&profile)?
.ok_or(BackupError::OffsiteNotConfigured)?
.resolve()?;
let client = offsite.build_client()?;
let run_id = match &oref.selector {
OffsiteSelector::Exact(sel) => resolve_exact_run(&offsite, &client, &profile, sel)?,
OffsiteSelector::Latest => resolve_latest_run(&offsite, &client, &profile)?,
};
eprintln!(" \u{2139} restoring from offsite {profile}/{run_id}");
let run_prefix = offsite_run_prefix(&offsite.prefix, &profile, &run_id);
let objects = client
.list_objects_v2(&run_prefix)
.map_err(|e| BackupError::Offsite {
op: "list",
detail: e.to_string(),
})?;
if objects.is_empty() {
return Err(BackupError::BadArtifact {
detail: format!(
"no offsite backup found at {profile}/{run_id} (nothing under {run_prefix})"
),
});
}
let temp =
tempfile::tempdir().map_err(BackupError::io("creating a temp dir for offsite restore"))?;
for obj in &objects {
let file = obj
.key
.strip_prefix(run_prefix.as_str())
.filter(|f| is_safe_leaf_name(f))
.ok_or_else(|| BackupError::Offsite {
op: "download",
detail: format!(
"refusing offsite object with an unsafe key layout: {} (each object name \
must be a single plain filename — no path separators, traversal, or drive \
prefix)",
obj.key
),
})?;
let dest = temp.path().join(file);
let mut out = std::fs::File::create(&dest)
.map_err(BackupError::io(format!("creating downloaded {file}")))?;
client
.download_object(&obj.key, &mut out)
.map_err(|e| BackupError::Offsite {
op: "download",
detail: format!("{file}: {e}"),
})?;
eprintln!(" \u{2913} downloaded {file}");
}
let plan = RestorePlan::discover(temp.path(), args.shard.as_deref())?;
apply_restore_plan(&plan, args)
}
fn resolve_exact_run(
offsite: &ResolvedOffsite,
client: &S3Client,
profile: &str,
selector: &str,
) -> Result<String, BackupError> {
let list_prefix = offsite_profile_prefix(&offsite.prefix, profile);
let objects = client
.list_objects_v2(&list_prefix)
.map_err(|e| BackupError::Offsite {
op: "list",
detail: e.to_string(),
})?;
let run_ids = complete_remote_run_ids(&objects, &list_prefix);
match match_exact_remote_run(&run_ids, selector) {
Ok(id) => Ok(id),
Err(candidates) if candidates.is_empty() => Err(BackupError::BadArtifact {
detail: format!(
"no complete offsite backup matching {selector:?} for profile {profile:?}"
),
}),
Err(candidates) => Err(BackupError::BadArtifact {
detail: format!(
"offsite reference {selector:?} is ambiguous for profile {profile:?}: it matches \
{} runs ({}).\n Re-run restore with the full run id (see `autumn db offsite list`).",
candidates.len(),
candidates.join(", "),
),
}),
}
}
fn match_exact_remote_run(run_ids: &[String], selector: &str) -> Result<String, Vec<String>> {
let token_prefix = format!("{selector}-");
let matches: Vec<String> = run_ids
.iter()
.filter(|id| id.as_str() == selector || id.starts_with(&token_prefix))
.cloned()
.collect();
if let [only] = matches.as_slice() {
Ok(only.clone())
} else {
Err(matches)
}
}
fn resolve_latest_run(
offsite: &ResolvedOffsite,
client: &S3Client,
profile: &str,
) -> Result<String, BackupError> {
let list_prefix = offsite_profile_prefix(&offsite.prefix, profile);
let objects = client
.list_objects_v2(&list_prefix)
.map_err(|e| BackupError::Offsite {
op: "list",
detail: e.to_string(),
})?;
complete_remote_run_ids(&objects, &list_prefix)
.into_iter()
.next_back() .ok_or_else(|| BackupError::BadArtifact {
detail: format!("no complete offsite backups found for profile {profile:?}"),
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn format_parse_accepts_documented_spellings() {
assert_eq!(BackupFormat::parse("custom").unwrap(), BackupFormat::Custom);
assert_eq!(BackupFormat::parse("C").unwrap(), BackupFormat::Custom);
assert_eq!(BackupFormat::parse("plain").unwrap(), BackupFormat::Plain);
assert_eq!(BackupFormat::parse("sql").unwrap(), BackupFormat::Plain);
assert!(BackupFormat::parse("gzip").is_err());
}
#[test]
fn default_format_is_custom_compressed() {
assert_eq!(BackupFormat::default(), BackupFormat::Custom);
assert_eq!(BackupFormat::Custom.extension(), "dump");
assert_eq!(BackupFormat::Plain.extension(), "sql");
}
#[test]
fn backup_root_nests_under_profile() {
assert_eq!(
backup_root(None, "dev"),
PathBuf::from("backups").join("dev")
);
assert_eq!(
backup_root(Some(Path::new("/var/backups")), "prod"),
PathBuf::from("/var/backups").join("prod")
);
}
#[test]
fn run_dir_name_is_sortable_utc() {
use chrono::TimeZone as _;
let ts = chrono::Utc.with_ymd_and_hms(2026, 7, 10, 4, 5, 6).unwrap();
assert_eq!(run_dir_name(&ts), "20260710T040506Z");
}
#[test]
fn artifact_file_name_distinguishes_control_and_shards() {
assert_eq!(
artifact_file_name("control", BackupFormat::Custom),
"control.dump"
);
assert_eq!(
artifact_file_name("shard:us_east", BackupFormat::Custom),
"shard-us_east.dump"
);
assert_eq!(
artifact_file_name("shard:us east", BackupFormat::Plain),
"shard-us_east.sql"
);
}
#[test]
fn sanitize_component_replaces_unsafe_chars() {
assert_eq!(sanitize_component("us-east_1.a"), "us-east_1.a");
assert_eq!(sanitize_component("a/b:c"), "a_b_c");
assert_eq!(sanitize_component(""), "unnamed");
assert_eq!(sanitize_component("../etc"), ".._etc");
}
#[test]
fn sanitize_component_rejects_traversal_components() {
assert_eq!(sanitize_component("shard1"), "shard1");
assert_eq!(sanitize_component("."), "unnamed");
assert_eq!(sanitize_component(".."), "unnamed");
assert_eq!(sanitize_component(""), "unnamed");
}
#[test]
fn build_targets_all_puts_control_first_then_shards() {
let targets = build_targets(
Some("postgres://localhost/app".to_owned()),
vec![
("east".to_owned(), "postgres://localhost/east".to_owned()),
("west".to_owned(), "postgres://localhost/west".to_owned()),
],
&TargetSelector::All,
)
.unwrap();
let labels: Vec<&str> = targets.iter().map(|t| t.label.as_str()).collect();
assert_eq!(labels, ["control", "shard:east", "shard:west"]);
}
#[test]
fn build_targets_shard_only_shape_is_allowed() {
let targets = build_targets(
None,
vec![("east".to_owned(), "postgres://localhost/east".to_owned())],
&TargetSelector::All,
)
.unwrap();
assert_eq!(targets.len(), 1);
assert_eq!(targets[0].label, "shard:east");
}
#[test]
fn build_targets_all_with_no_url_errors() {
assert!(matches!(
build_targets(None, vec![], &TargetSelector::All),
Err(BackupError::NoUrl)
));
}
#[test]
fn build_targets_control_only_ignores_shards() {
let targets = build_targets(
Some("postgres://localhost/app".to_owned()),
vec![("east".to_owned(), "postgres://localhost/east".to_owned())],
&TargetSelector::ControlOnly,
)
.unwrap();
assert_eq!(targets.len(), 1);
assert_eq!(targets[0].label, "control");
}
#[test]
fn build_targets_named_shard_selects_one() {
let targets = build_targets(
Some("postgres://localhost/app".to_owned()),
vec![
("east".to_owned(), "postgres://localhost/east".to_owned()),
("west".to_owned(), "postgres://localhost/west".to_owned()),
],
&TargetSelector::Shard("west".to_owned()),
)
.unwrap();
assert_eq!(targets.len(), 1);
assert_eq!(targets[0].label, "shard:west");
assert_eq!(targets[0].url, "postgres://localhost/west");
}
#[test]
fn build_targets_unknown_shard_errors_with_known_list() {
let err = build_targets(
Some("postgres://localhost/app".to_owned()),
vec![("east".to_owned(), "postgres://localhost/east".to_owned())],
&TargetSelector::Shard("nope".to_owned()),
)
.unwrap_err();
match err {
BackupError::UnknownShard { name, known } => {
assert_eq!(name, "nope");
assert_eq!(known, vec!["east".to_owned()]);
}
other => panic!("expected UnknownShard, got {other:?}"),
}
}
#[test]
fn plan_pruning_keeps_newest_n() {
let runs: Vec<String> = ["a", "b", "c", "d", "e"]
.iter()
.map(|s| (*s).to_owned())
.collect();
assert_eq!(
plan_pruning(&runs, 2),
vec!["a".to_owned(), "b".to_owned(), "c".to_owned()]
);
assert!(plan_pruning(&runs, 5).is_empty());
assert!(plan_pruning(&runs, 99).is_empty());
}
#[test]
fn list_run_dirs_only_counts_dirs_with_a_manifest() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
let good = root.join("20260710T040506Z");
std::fs::create_dir_all(&good).unwrap();
std::fs::write(good.join(MANIFEST_FILE), b"{}").unwrap();
std::fs::create_dir_all(root.join("notes")).unwrap();
std::fs::write(root.join("README.txt"), b"hi").unwrap();
let dirs = list_run_dirs(root).unwrap();
assert_eq!(dirs, vec!["20260710T040506Z".to_owned()]);
}
#[test]
fn create_unique_run_dir_never_overwrites_same_timestamp() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path().join("dev");
let base = "20260710T040506Z";
let first = create_unique_run_dir(&root, base).unwrap();
std::fs::write(first.join(MANIFEST_FILE), b"{}").unwrap();
let second = create_unique_run_dir(&root, base).unwrap();
assert_ne!(first, second);
assert_eq!(first, root.join(base));
assert_eq!(second, root.join(format!("{base}-2")));
assert!(first.join(MANIFEST_FILE).is_file());
assert!(base < format!("{base}-2").as_str());
let third = create_unique_run_dir(&root, base).unwrap();
assert_eq!(third, root.join(format!("{base}-3")));
}
#[test]
fn plan_pruning_never_removes_everything_on_keep_zero() {
let runs: Vec<String> = ["a", "b"].iter().map(|s| (*s).to_owned()).collect();
assert_eq!(plan_pruning(&runs, 0), vec!["a".to_owned()]);
}
#[test]
fn candidate_bin_dirs_orders_override_then_managed() {
let dirs = candidate_bin_dirs(
Some(PathBuf::from("/opt/pg/bin")),
Some(PathBuf::from("/data/app/pg")),
);
assert_eq!(dirs[0], PathBuf::from("/opt/pg/bin"));
assert!(dirs.contains(&PathBuf::from("/data/app/postgresql/bin")));
}
#[test]
fn candidate_bin_dirs_empty_without_env() {
assert!(candidate_bin_dirs(None, None).is_empty());
}
#[test]
fn resolve_tool_in_falls_back_to_bare_name() {
assert_eq!(
resolve_tool_in(&[PathBuf::from("/nonexistent")], "pg_dump"),
PathBuf::from("pg_dump")
);
}
#[test]
fn resolve_tool_in_prefers_existing_candidate() {
let tmp = tempfile::tempdir().unwrap();
let bin = tmp.path().join("bin");
std::fs::create_dir_all(&bin).unwrap();
let exe = bin.join(exe_name("pg_dump"));
std::fs::write(&exe, b"#!/bin/sh\n").unwrap();
assert_eq!(resolve_tool_in(std::slice::from_ref(&bin), "pg_dump"), exe);
}
#[test]
fn managed_data_dir_candidate_locates_bundled_tools() {
let tmp = tempfile::tempdir().unwrap();
let data_dir = tmp.path().join("pgdata");
std::fs::create_dir_all(&data_dir).unwrap();
let bundle_bin = tmp.path().join("postgresql").join("bin");
std::fs::create_dir_all(&bundle_bin).unwrap();
let pg_dump = bundle_bin.join(exe_name("pg_dump"));
let pg_restore = bundle_bin.join(exe_name("pg_restore"));
std::fs::write(&pg_dump, b"#!/bin/sh\n").unwrap();
std::fs::write(&pg_restore, b"#!/bin/sh\n").unwrap();
let dirs = candidate_bin_dirs(None, Some(data_dir));
assert!(dirs.contains(&bundle_bin));
assert_eq!(resolve_tool_in(&dirs, "pg_dump"), pg_dump);
assert_eq!(resolve_tool_in(&dirs, "pg_restore"), pg_restore);
}
#[cfg(unix)]
#[test]
fn verify_artifact_resolves_pg_restore_from_managed_tools() {
use std::os::unix::fs::PermissionsExt;
let tmp = tempfile::tempdir().unwrap();
let bundle_bin = tmp.path().join("postgresql").join("bin");
std::fs::create_dir_all(&bundle_bin).unwrap();
let pg_restore = bundle_bin.join("pg_restore");
std::fs::write(
&pg_restore,
"#!/bin/sh\necho ';'\necho '215; 1259 16385 TABLE public posts app'\n",
)
.unwrap();
std::fs::set_permissions(&pg_restore, std::fs::Permissions::from_mode(0o755)).unwrap();
let data_dir = tmp.path().join("pgdata");
std::fs::create_dir_all(&data_dir).unwrap();
let tools = PgTools::with_dirs(candidate_bin_dirs(None, Some(data_dir)));
let archive = tmp.path().join("control.dump");
std::fs::write(&archive, b"PGDMP synthetic archive").unwrap();
verify_artifact(&archive, BackupFormat::Custom, "app", &tools)
.expect("verify resolves the bundled pg_restore and accepts the TOC");
}
#[test]
fn toc_entry_count_ignores_comment_lines() {
let listing = "\
;
; Archive created at 2026-07-10
;
215; 1259 16385 TABLE public posts app
216; 1259 16390 TABLE public comments app
";
assert_eq!(toc_entry_count(listing), 2);
}
#[test]
fn toc_entry_count_zero_for_header_only() {
let listing = ";\n; only comments\n;\n";
assert_eq!(toc_entry_count(listing), 0);
}
#[test]
fn plain_dump_completion_marker_detected() {
assert!(plain_dump_is_complete(
"CREATE TABLE x();\n--\n-- PostgreSQL database dump complete\n--\n"
));
assert!(!plain_dump_is_complete("CREATE TABLE x();\n-- truncated"));
}
#[test]
fn verify_plain_dump_reads_only_tail() {
let tmp = tempfile::tempdir().unwrap();
let complete = tmp.path().join("complete.sql");
let mut body = "CREATE TABLE x();\n".repeat(4096);
body.push_str("--\n-- PostgreSQL database dump complete\n--\n");
std::fs::write(&complete, &body).unwrap();
assert!(verify_plain_dump(&complete, "db").is_ok());
let truncated = tmp.path().join("truncated.sql");
std::fs::write(&truncated, "CREATE TABLE x();\n".repeat(4096)).unwrap();
assert!(matches!(
verify_plain_dump(&truncated, "db"),
Err(BackupError::IntegrityFailed { .. })
));
let tiny = tmp.path().join("tiny.sql");
std::fs::write(&tiny, "-- PostgreSQL database dump complete\n").unwrap();
assert!(verify_plain_dump(&tiny, "db").is_ok());
let empty = tmp.path().join("empty.sql");
std::fs::write(&empty, b"").unwrap();
assert!(matches!(
verify_plain_dump(&empty, "db"),
Err(BackupError::IntegrityFailed { .. })
));
}
#[test]
fn pg_dump_args_bake_clean_into_plain_only() {
let to_str = |args: &[std::ffi::OsString]| -> Vec<String> {
args.iter()
.map(|s| s.to_string_lossy().into_owned())
.collect()
};
let plain = pg_dump_args(
BackupFormat::Plain,
Path::new("/backups/dev/run/control.sql"),
"postgres://user@db.example.com/my_app",
);
let plain = to_str(&plain);
assert!(plain.contains(&"--clean".to_owned()));
assert!(plain.contains(&"--if-exists".to_owned()));
assert!(
plain
.windows(2)
.any(|w| w[0] == "--format" && w[1] == "plain")
);
assert!(
plain
.windows(2)
.any(|w| w[0] == "--dbname" && w[1] == "postgres://user@db.example.com/my_app")
);
let custom = pg_dump_args(
BackupFormat::Custom,
Path::new("/backups/dev/run/control.dump"),
"postgres://user@db.example.com/my_app",
);
let custom = to_str(&custom);
assert!(!custom.contains(&"--clean".to_owned()));
assert!(!custom.contains(&"--if-exists".to_owned()));
assert!(
custom
.windows(2)
.any(|w| w[0] == "--format" && w[1] == "custom")
);
}
#[test]
fn resolve_control_url_prefers_config_over_managed_pg_fallback() {
let mut fallback_called = false;
let got = resolve_control_url(Some("postgres://cfg/db".to_owned()), || {
fallback_called = true;
Some("postgres://managed/db".to_owned())
});
assert_eq!(got.as_deref(), Some("postgres://cfg/db"));
assert!(
!fallback_called,
"managed-pg fallback must not run when config/env resolves a URL"
);
let got = resolve_control_url(None, || Some("postgres://managed/db".to_owned()));
assert_eq!(got.as_deref(), Some("postgres://managed/db"));
let got = resolve_control_url(None, || None);
assert_eq!(got, None);
}
#[test]
fn infer_format_from_extension() {
assert_eq!(
infer_format_from_path(Path::new("control.dump")),
Some(BackupFormat::Custom)
);
assert_eq!(
infer_format_from_path(Path::new("control.sql")),
Some(BackupFormat::Plain)
);
assert_eq!(infer_format_from_path(Path::new("control.txt")), None);
}
#[test]
fn parsed_db_name_is_credential_safe() {
let name = parsed_db_name("postgres://user:hunter2@db.example.com:6543/my_app");
assert_eq!(name, "my_app");
assert!(!name.contains("hunter2"));
assert_eq!(parsed_db_name("not a url"), "(database)");
}
#[test]
fn split_password_moves_secret_out_of_url() {
let (safe, pw) = split_password("postgres://user:hunter2@db.example.com:6543/my_app");
assert_eq!(pw.as_deref(), Some("hunter2"));
assert!(!safe.contains("hunter2"));
assert!(safe.contains("user@db.example.com"));
assert!(safe.contains("/my_app"));
}
#[test]
fn split_password_percent_decodes_for_pgpassword() {
let raw = "p@ss:w/ord x";
let encoded = "p%40ss%3Aw%2Ford%20x";
let url = format!("postgres://user:{encoded}@db.example.com:6543/my_app");
let (safe, pw) = split_password(&url);
assert_eq!(pw.as_deref(), Some(raw));
assert!(!safe.contains(encoded));
assert!(!safe.contains(raw));
assert!(safe.contains("user@db.example.com"));
assert!(safe.contains("/my_app"));
assert!(url::Url::parse(&safe).is_ok());
}
#[test]
fn split_password_passthrough_when_absent_or_unparseable() {
let (safe, pw) = split_password("postgres://user@localhost/app");
assert_eq!(pw, None);
assert_eq!(safe, "postgres://user@localhost/app");
let (safe, pw) = split_password("not a url");
assert_eq!(pw, None);
assert_eq!(safe, "not a url");
}
#[test]
fn split_password_strips_keyword_value_password() {
let (safe, pw) = split_password("host=db user=app password=secret dbname=app");
assert_eq!(pw.as_deref(), Some("secret"));
assert!(!safe.contains("password"));
assert!(!safe.contains("secret"));
assert!(safe.contains("host=db"));
assert!(safe.contains("user=app"));
assert!(safe.contains("dbname=app"));
let reparsed = parse_libpq_kv(&safe).unwrap();
assert!(reparsed.iter().all(|(k, _)| k != "password"));
}
#[test]
fn split_password_handles_single_quoted_keyword_value() {
let (safe, pw) = split_password("host=db password='p a$s' dbname=app");
assert_eq!(pw.as_deref(), Some("p a$s"));
assert!(!safe.contains("p a$s"));
assert!(!safe.contains("password"));
assert!(safe.contains("host=db"));
assert!(safe.contains("dbname=app"));
}
#[test]
fn split_password_handles_escaped_quote_in_keyword_value() {
let (_safe, pw) = split_password(r"host=db password='a\'b\\c'");
assert_eq!(pw.as_deref(), Some(r"a'b\c"));
}
#[test]
fn split_password_keyword_value_without_password_passes_through() {
let input = "host=db user=app dbname=app";
let (safe, pw) = split_password(input);
assert_eq!(pw, None);
assert_eq!(safe, input);
}
#[test]
fn restore_plan_select_filters_to_named_shard() {
let plan = RestorePlan {
dir: PathBuf::from("/x"),
format: BackupFormat::Custom,
entries: vec![
ManifestTarget {
label: "control".to_owned(),
file: "control.dump".to_owned(),
database: "app".to_owned(),
},
ManifestTarget {
label: "shard:east".to_owned(),
file: "shard-east.dump".to_owned(),
database: "east".to_owned(),
},
],
};
let all = plan.select(None).unwrap();
assert_eq!(all.len(), 2);
let one = plan.select(Some("east")).unwrap();
assert_eq!(one.len(), 1);
assert_eq!(one[0].label, "shard:east");
assert!(matches!(
plan.select(Some("nope")),
Err(BackupError::UnknownShard { .. })
));
}
#[test]
fn single_file_target_label_infers_from_filename_convention() {
assert_eq!(
single_file_target_label("control.dump", None).unwrap(),
"control"
);
assert_eq!(
single_file_target_label("control.sql", None).unwrap(),
"control"
);
assert_eq!(
single_file_target_label("shard-east.dump", None).unwrap(),
"shard:east"
);
assert_eq!(
single_file_target_label("control.dump", Some("west")).unwrap(),
"shard:west"
);
assert!(matches!(
single_file_target_label("shard-.dump", None),
Err(BackupError::BadArtifact { .. })
));
assert_eq!(
single_file_target_label("mybackup.dump", None).unwrap(),
"control"
);
}
#[test]
fn discover_bare_shard_file_resolves_to_shard_not_control() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("shard-east.dump");
std::fs::write(&path, b"x").unwrap();
let plan = RestorePlan::discover(&path, None).unwrap();
assert_eq!(plan.entries.len(), 1);
assert_eq!(plan.entries[0].label, "shard:east");
assert_eq!(plan.select(Some("east")).unwrap()[0].label, "shard:east");
}
#[test]
fn discover_bare_control_file_resolves_to_control() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("control.dump");
std::fs::write(&path, b"x").unwrap();
let plan = RestorePlan::discover(&path, None).unwrap();
assert_eq!(plan.entries[0].label, "control");
}
#[test]
fn discover_bare_file_honors_explicit_shard_flag() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("control.dump");
std::fs::write(&path, b"x").unwrap();
let plan = RestorePlan::discover(&path, Some("east")).unwrap();
assert_eq!(plan.entries[0].label, "shard:east");
}
#[test]
fn manifest_round_trips_through_disk() {
let tmp = tempfile::tempdir().unwrap();
let manifest = Manifest {
autumn_version: "0.6.0".to_owned(),
created_at: "2026-07-10T04:05:06+00:00".to_owned(),
profile: "dev".to_owned(),
format: "custom".to_owned(),
targets: vec![ManifestTarget {
label: "control".to_owned(),
file: "control.dump".to_owned(),
database: "app".to_owned(),
}],
};
write_manifest(tmp.path(), &manifest).unwrap();
let read = read_manifest(tmp.path()).unwrap();
assert_eq!(read, manifest);
}
#[test]
fn errors_are_credential_safe() {
let e = BackupError::ProductionRefused {
profile: "prod".to_owned(),
};
let s = e.to_string();
assert!(s.contains("prod"));
assert!(!s.contains("postgres://"));
}
#[test]
fn offsite_object_key_joins_and_skips_empty_prefix() {
assert_eq!(
offsite_object_key("db", "prod", "20260710T040506Z", "control.dump"),
"db/prod/20260710T040506Z/control.dump"
);
assert_eq!(
offsite_object_key("", "prod", "20260710T040506Z", "manifest.json"),
"prod/20260710T040506Z/manifest.json"
);
}
#[test]
fn normalize_offsite_prefix_trims_slashes() {
assert_eq!(normalize_offsite_prefix(Some("/db/backups/")), "db/backups");
assert_eq!(normalize_offsite_prefix(Some(" db ")), "db");
assert_eq!(normalize_offsite_prefix(None), "");
assert_eq!(normalize_offsite_prefix(Some("")), "");
}
#[test]
fn offsite_profile_and_run_prefixes_are_scoped() {
assert_eq!(offsite_profile_prefix("db", "prod"), "db/prod/");
assert_eq!(offsite_profile_prefix("", "prod"), "prod/");
assert_eq!(
offsite_run_prefix("db", "prod", "20260710T040506Z"),
"db/prod/20260710T040506Z/"
);
}
#[test]
fn offsite_key_prefix_is_canonical_across_profile_spellings() {
for spelling in ["prod", "production", "PROD", "Production", " prod "] {
let p = migrate::canonical_profile(spelling);
assert_eq!(offsite_profile_prefix("db", &p), "db/prod/", "{spelling:?}");
assert_eq!(
offsite_run_prefix("db", &p, "r"),
"db/prod/r/",
"{spelling:?}"
);
assert_eq!(
offsite_object_key("db", &p, "r", "manifest.json"),
"db/prod/r/manifest.json",
"{spelling:?}"
);
}
assert_eq!(
offsite_profile_prefix("db", &migrate::canonical_profile("prod")),
offsite_profile_prefix("db", &migrate::canonical_profile("production")),
);
assert_eq!(
offsite_profile_prefix("db", &migrate::canonical_profile("development")),
"db/dev/"
);
}
#[test]
fn destinations_conflict_requires_same_bucket_and_endpoint() {
assert!(destinations_conflict(
"shared",
Some("https://s3.example.test"),
Some("shared"),
Some("https://s3.example.test/"),
));
assert!(!destinations_conflict(
"shared",
Some("https://offsite.test"),
Some("shared"),
Some("https://app.test"),
));
assert!(!destinations_conflict(
"offsite",
Some("https://s3.example.test"),
Some("app"),
Some("https://s3.example.test"),
));
assert!(!destinations_conflict("offsite", None, None, None));
assert!(destinations_conflict("shared", None, Some("shared"), None));
}
#[test]
fn destinations_conflict_normalizes_ports_and_aws_defaults() {
assert!(destinations_conflict(
"shared",
Some("https://minio.example:443"),
Some("shared"),
Some("https://minio.example"),
));
assert!(destinations_conflict(
"shared",
Some("http://gw:80"),
Some("shared"),
Some("http://gw"),
));
assert!(destinations_conflict(
"shared",
None,
Some("shared"),
Some("https://s3.us-east-1.amazonaws.com"),
));
assert!(destinations_conflict(
"shared",
None,
Some("shared"),
Some("https://shared.s3.us-east-1.amazonaws.com"),
));
assert!(!destinations_conflict(
"shared",
Some("https://minio.example:9000"),
Some("shared"),
Some("https://minio.example"),
));
assert!(destinations_conflict(
"shared",
Some("https://shared.minio.example"),
Some("shared"),
Some("https://minio.example"),
));
}
#[test]
fn shared_bucket_refusal_message_is_credential_safe() {
let e = BackupError::SharedBucketRefused {
bucket: "shared".to_owned(),
};
let s = e.to_string();
assert!(s.contains("shared"));
assert!(s.contains("allow_shared_bucket"));
assert!(!s.contains("secret"));
}
#[test]
fn missing_credential_env_names_only_the_variable() {
let e = BackupError::MissingCredentialEnv {
var: "OFFSITE_SECRET".to_owned(),
};
let s = e.to_string();
assert!(s.contains("OFFSITE_SECRET"));
assert!(!s.contains("hunter2"));
}
#[test]
fn upload_failed_reports_split_outcome() {
let e = BackupError::OffsiteUploadFailed {
local_path: "/backups/prod/20260710T040506Z".to_owned(),
detail: "S3 put returned HTTP 403 (AccessDenied)".to_owned(),
};
let s = e.to_string();
assert!(s.contains("Local backup OK at /backups/prod/20260710T040506Z"));
assert!(s.contains("OFFSITE UPLOAD FAILED"));
assert!(s.contains("intact"));
}
#[derive(Default)]
struct CapturingChannel {
received: Arc<std::sync::Mutex<Vec<Alert>>>,
}
impl AlertChannel for CapturingChannel {
fn name(&self) -> &'static str {
"capturing"
}
fn deliver<'a>(&'a self, alert: &'a Alert) -> autumn_web::alerts::AlertDeliveryFuture<'a> {
let received = Arc::clone(&self.received);
let cloned = alert.clone();
Box::pin(async move {
received.lock().expect("lock").push(cloned);
Ok(())
})
}
}
#[test]
fn build_offsite_upload_alert_is_scheduled_task_failure() {
let alert = build_offsite_upload_alert(
"/backups/prod/20260710T040506Z",
"S3 put returned HTTP 403 (AccessDenied)",
);
assert_eq!(alert.condition, AlertCondition::ScheduledTaskFailure);
assert_eq!(alert.title, "Offsite backup upload failed");
assert_eq!(alert.summary, "S3 put returned HTTP 403 (AccessDenied)");
assert_eq!(
alert.dedup_key,
"scheduled_task_failure:db-backup-offsite-upload"
);
}
#[test]
fn deliver_offsite_upload_alert_reaches_configured_channel() {
let capture = Arc::new(std::sync::Mutex::new(Vec::new()));
let channel = Arc::new(CapturingChannel {
received: Arc::clone(&capture),
});
let channels: Vec<Arc<dyn AlertChannel>> = vec![channel];
deliver_offsite_upload_alert(
&channels,
"/backups/prod/20260710T040506Z",
"S3 put returned HTTP 403 (AccessDenied)",
);
let delivered = {
let received = capture.lock().expect("lock");
received.clone()
};
assert_eq!(delivered.len(), 1, "exactly one alert must be delivered");
let alert = &delivered[0];
assert_eq!(alert.condition, AlertCondition::ScheduledTaskFailure);
assert_eq!(alert.title, "Offsite backup upload failed");
assert_eq!(alert.summary, "S3 put returned HTTP 403 (AccessDenied)");
}
#[test]
fn deliver_offsite_upload_alert_is_noop_without_channels() {
deliver_offsite_upload_alert(&[], "/backups/prod/run", "boom");
}
#[test]
fn remote_run_ids_extracts_unique_sorted_ids() {
let list_prefix = "db/prod/";
let objects = vec![
s3::S3Object {
key: "db/prod/20260710T010101Z/control.dump".to_owned(),
size: 1,
},
s3::S3Object {
key: "db/prod/20260710T010101Z/manifest.json".to_owned(),
size: 1,
},
s3::S3Object {
key: "db/prod/20260709T010101Z/control.dump".to_owned(),
size: 1,
},
];
assert_eq!(
remote_run_ids(&objects, list_prefix),
vec!["20260709T010101Z".to_owned(), "20260710T010101Z".to_owned()]
);
}
#[test]
fn local_backup_does_not_validate_incomplete_offsite() {
let offsite = autumn_web::config::OffsiteBackupConfig::default();
assert!(offsite.s3.bucket.is_none(), "precondition: no bucket set");
let loaded = LoadedOffsite {
offsite,
app_storage_bucket: None,
app_storage_endpoint: None,
profile: "test".to_owned(),
};
assert!(matches!(
loaded.resolve(),
Err(BackupError::OffsiteConfig { .. })
));
}
#[test]
fn auto_upload_probe_precedence_and_repro() {
let no_env = |_: &str| None;
assert!(!auto_upload_from_sources(no_env, None));
let empty: toml::Table = toml::Table::new();
assert!(!auto_upload_from_sources(no_env, Some(&empty)));
let toml_true: toml::Table =
toml::from_str("[backup.offsite]\nauto_upload = true\n").unwrap();
assert!(auto_upload_from_sources(no_env, Some(&toml_true)));
let env_true =
|k: &str| (k == "AUTUMN_BACKUP__OFFSITE__AUTO_UPLOAD").then(|| "true".to_owned());
assert!(auto_upload_from_sources(env_true, None));
let env_false =
|k: &str| (k == "AUTUMN_BACKUP__OFFSITE__AUTO_UPLOAD").then(|| "false".to_owned());
assert!(!auto_upload_from_sources(env_false, Some(&toml_true)));
let env_junk =
|k: &str| (k == "AUTUMN_BACKUP__OFFSITE__AUTO_UPLOAD").then(|| "yesplease".to_owned());
assert!(auto_upload_from_sources(env_junk, Some(&toml_true)));
}
#[test]
fn offsite_resolves_from_config_dir_table_not_cwd() {
let tmp = tempfile::tempdir().unwrap();
std::fs::write(
tmp.path().join("autumn.toml"),
"[backup.offsite]\n\
auto_upload = true\n\
prefix = \"db\"\n\
[backup.offsite.s3]\n\
bucket = \"offsite-bucket\"\n\
region = \"us-east-1\"\n\
endpoint = \"https://minio.example:9000\"\n\
force_path_style = true\n\
access_key_id_env = \"OFFSITE_KEY\"\n\
secret_access_key_env = \"OFFSITE_SECRET\"\n",
)
.unwrap();
let table = migrate::read_autumn_toml_table_with_profile_in(tmp.path(), Some("dev"));
assert!(auto_upload_from_sources(|_| None, table.as_ref()));
let env = autumn_web::config::MockEnv::new();
let loaded = resolve_loaded_offsite(table.as_ref(), &env, "dev").unwrap();
assert!(
loaded.is_some(),
"offsite must resolve from the config-dir autumn.toml"
);
}
#[test]
fn resolved_offsite_carries_target_profile_for_credentials() {
let mut offsite = autumn_web::config::OffsiteBackupConfig::default();
offsite.s3.bucket = Some("offsite-bucket".to_owned());
offsite.s3.access_key_id_env = Some("OFFSITE_KEY".to_owned());
offsite.s3.secret_access_key_env = Some("OFFSITE_SECRET".to_owned());
let loaded = LoadedOffsite {
offsite,
app_storage_bucket: None,
app_storage_endpoint: None,
profile: "test".to_owned(),
};
let resolved = loaded.resolve().expect("bucket set, resolves");
assert_eq!(resolved.profile, "test");
}
#[test]
fn offsite_resolves_without_app_credential_store() {
let table: toml::Table = toml::from_str(
"[backup.offsite]\nprefix = \"db\"\n\
[backup.offsite.s3]\nbucket = \"dr-bucket\"\nregion = \"us-east-1\"\n\
endpoint = \"https://minio.example:9000\"\nforce_path_style = true\n\
access_key_id_env = \"OFFSITE_KEY\"\nsecret_access_key_env = \"OFFSITE_SECRET\"\n",
)
.unwrap();
let env = autumn_web::config::MockEnv::new()
.with("OFFSITE_KEY", "AKIA_DR")
.with("OFFSITE_SECRET", "shh");
let loaded = resolve_loaded_offsite(Some(&table), &env, "prod")
.expect("resolves without a credential store")
.expect("offsite section present");
let resolved = loaded.resolve().expect("bucket set");
assert_eq!(resolved.s3.bucket, "dr-bucket");
assert_eq!(resolved.profile, "prod");
let env2 = autumn_web::config::MockEnv::new()
.with("AUTUMN_BACKUP__OFFSITE__S3__BUCKET", "env-bucket");
let loaded2 = resolve_loaded_offsite(None, &env2, "prod")
.unwrap()
.expect("materialized from env");
assert_eq!(loaded2.offsite.s3.bucket.as_deref(), Some("env-bucket"));
}
#[test]
fn shared_bucket_guard_only_applies_when_storage_backend_is_s3() {
let env = autumn_web::config::MockEnv::new();
let make = |backend: &str| -> LoadedOffsite {
let table: toml::Table = toml::from_str(&format!(
"[storage]\nbackend = \"{backend}\"\n\
[storage.s3]\nbucket = \"shared\"\nendpoint = \"https://minio.example:9000\"\n\
[backup.offsite]\nprefix = \"db\"\n\
[backup.offsite.s3]\nbucket = \"shared\"\nregion = \"us-east-1\"\n\
endpoint = \"https://minio.example:9000\"\nforce_path_style = true\n"
))
.unwrap();
resolve_loaded_offsite(Some(&table), &env, "prod")
.expect("resolves")
.expect("offsite section present")
};
let local = make("local");
assert_eq!(local.app_storage_bucket, None);
assert!(!destinations_conflict(
"shared",
Some("https://minio.example:9000"),
local.app_storage_bucket.as_deref(),
local.app_storage_endpoint.as_deref(),
));
assert_eq!(make("disabled").app_storage_bucket, None);
let s3 = make("s3");
assert_eq!(s3.app_storage_bucket.as_deref(), Some("shared"));
assert!(destinations_conflict(
"shared",
Some("https://minio.example:9000"),
s3.app_storage_bucket.as_deref(),
s3.app_storage_endpoint.as_deref(),
));
}
#[test]
fn key_is_manifest_matches_only_the_run_manifest() {
assert!(key_is_manifest(
"db/prod/20260710T040506Z-abcd1234/manifest.json"
));
assert!(!key_is_manifest(
"db/prod/20260710T040506Z-abcd1234/control.dump"
));
assert!(!key_is_manifest("db/prod/r/not-manifest.json.dump"));
}
#[test]
fn order_cleanup_keys_deletes_manifest_first() {
let written = vec![
"db/prod/r/control.dump".to_owned(),
"db/prod/r/shard-a.dump".to_owned(),
"db/prod/r/manifest.json".to_owned(),
];
let ordered = order_cleanup_keys(&written);
assert_eq!(ordered[0], "db/prod/r/manifest.json");
assert_eq!(
&ordered[1..],
&[
"db/prod/r/control.dump".to_owned(),
"db/prod/r/shard-a.dump".to_owned()
]
);
let list_prefix = "db/prod/";
let after_cleanup = vec![s3::S3Object {
key: "db/prod/r/control.dump".to_owned(), size: 10,
}];
assert!(complete_remote_run_ids(&after_cleanup, list_prefix).is_empty());
}
#[test]
fn complete_remote_run_ids_only_counts_runs_with_a_manifest() {
let list_prefix = "db/prod/";
let objects = vec![
s3::S3Object {
key: "db/prod/20260709T010101Z/control.dump".to_owned(),
size: 10,
},
s3::S3Object {
key: "db/prod/20260709T010101Z/manifest.json".to_owned(),
size: 1,
},
s3::S3Object {
key: "db/prod/20260710T010101Z/control.dump".to_owned(),
size: 10,
},
];
assert_eq!(
remote_run_ids(&objects, list_prefix),
vec!["20260709T010101Z".to_owned(), "20260710T010101Z".to_owned()]
);
let complete = complete_remote_run_ids(&objects, list_prefix);
assert_eq!(complete, vec!["20260709T010101Z".to_owned()]);
assert_eq!(
complete.into_iter().next_back().as_deref(),
Some("20260709T010101Z")
);
}
#[test]
fn remote_retention_counts_only_complete_runs() {
let list_prefix = "db/prod/";
let a = "20260701T000000Z"; let b = "20260702T000000Z"; let c = "20260703T000000Z"; let d = "20260704T000000Z"; let objects = vec![
s3::S3Object {
key: format!("{list_prefix}{a}/control.dump"),
size: 1,
},
s3::S3Object {
key: format!("{list_prefix}{a}/manifest.json"),
size: 1,
},
s3::S3Object {
key: format!("{list_prefix}{b}/control.dump"),
size: 1,
},
s3::S3Object {
key: format!("{list_prefix}{b}/manifest.json"),
size: 1,
},
s3::S3Object {
key: format!("{list_prefix}{c}/control.dump"),
size: 1,
},
s3::S3Object {
key: format!("{list_prefix}{d}/control.dump"),
size: 1,
},
s3::S3Object {
key: format!("{list_prefix}{d}/manifest.json"),
size: 1,
},
];
let complete = complete_remote_run_ids(&objects, list_prefix);
assert_eq!(
complete,
vec![a.to_owned(), b.to_owned(), d.to_owned()],
"partial run C must not count as a retention slot"
);
let to_remove = plan_remote_pruning(&complete, 2, d);
assert_eq!(to_remove, vec![a.to_owned()]);
assert!(!to_remove.contains(&c.to_owned()));
assert!(!to_remove.contains(&d.to_owned()));
assert!(!to_remove.contains(&b.to_owned()));
}
#[test]
fn upload_orders_manifest_last() {
let mut files = vec![
"manifest.json".to_owned(),
"shard-east.dump".to_owned(),
"control.dump".to_owned(),
];
files.sort();
files.sort_by_key(|f| f.as_str() == MANIFEST_FILE);
assert_eq!(
files,
vec![
"control.dump".to_owned(),
"shard-east.dump".to_owned(),
"manifest.json".to_owned(),
]
);
}
#[test]
fn offsite_listing_flags_incomplete_runs() {
let list_prefix = "db/prod/";
let objects = vec![s3::S3Object {
key: "db/prod/20260710T010101Z/control.dump".to_owned(),
size: 10,
}];
let runs = group_offsite_objects(&objects, list_prefix);
assert_eq!(runs.len(), 1);
assert!(!runs[0].complete);
let table = format_offsite_listing(&runs);
assert!(table.contains("INCOMPLETE"), "table was: {table}");
}
#[test]
fn is_safe_leaf_name_rejects_separators_and_traversal() {
for good in ["control.dump", "manifest.json", "shard-orders.dump"] {
assert!(is_safe_leaf_name(good), "{good:?} should be accepted");
}
for bad in [
"",
"a/b",
"a\\b",
"..\\..\\evil",
"../evil",
"C:\\evil",
"..",
".",
"/abs",
"sub/manifest.json",
] {
assert!(!is_safe_leaf_name(bad), "{bad:?} should be rejected");
}
}
#[test]
fn remote_run_token_is_eight_lowercase_hex() {
let token = remote_run_token();
assert_eq!(token.len(), 8, "token was {token:?}");
assert!(
token
.bytes()
.all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()),
"token must be lowercase hex: {token:?}"
);
}
#[test]
fn remote_run_id_appends_token_to_local_id() {
let id = remote_run_id("20260710T040506Z", "deadbeef");
assert_eq!(id, "20260710T040506Z-deadbeef");
assert!(id.starts_with("20260710T040506Z"));
}
#[test]
fn match_exact_remote_run_resolves_by_full_id_and_timestamp() {
let ids: Vec<String> = ["20260709T010101Z-aaaa1111", "20260710T040506Z-bbbb2222"]
.iter()
.map(|s| (*s).to_owned())
.collect();
assert_eq!(
match_exact_remote_run(&ids, "20260710T040506Z-bbbb2222").unwrap(),
"20260710T040506Z-bbbb2222"
);
assert_eq!(
match_exact_remote_run(&ids, "20260710T040506Z").unwrap(),
"20260710T040506Z-bbbb2222"
);
assert_eq!(
match_exact_remote_run(&ids, "20990101T000000Z").unwrap_err(),
Vec::<String>::new()
);
}
#[test]
fn match_exact_remote_run_errors_on_ambiguous_same_second() {
let ids: Vec<String> = ["20260710T040506Z-aaaa1111", "20260710T040506Z-bbbb2222"]
.iter()
.map(|s| (*s).to_owned())
.collect();
let err = match_exact_remote_run(&ids, "20260710T040506Z").unwrap_err();
assert_eq!(err.len(), 2);
assert!(err.contains(&"20260710T040506Z-aaaa1111".to_owned()));
assert!(err.contains(&"20260710T040506Z-bbbb2222".to_owned()));
assert_eq!(
match_exact_remote_run(&ids, "20260710T040506Z-aaaa1111").unwrap(),
"20260710T040506Z-aaaa1111"
);
}
#[test]
fn complete_remote_run_ids_sort_newest_last_with_tokens() {
let list_prefix = "db/prod/";
let objects = vec![
s3::S3Object {
key: "db/prod/20260710T040506Z-zzzz9999/manifest.json".to_owned(),
size: 1,
},
s3::S3Object {
key: "db/prod/20260711T040506Z-aaaa0000/manifest.json".to_owned(),
size: 1,
},
];
let ids = complete_remote_run_ids(&objects, list_prefix);
assert_eq!(ids.last().unwrap(), "20260711T040506Z-aaaa0000");
}
#[test]
fn plan_remote_pruning_keeps_newest_and_never_the_just_uploaded() {
let runs: Vec<String> = ["r1", "r2", "r3", "r4", "r5"]
.iter()
.map(|s| (*s).to_owned())
.collect();
assert_eq!(
plan_remote_pruning(&runs, 2, "r5"),
vec!["r1".to_owned(), "r2".to_owned(), "r3".to_owned()]
);
assert!(plan_remote_pruning(&runs, 5, "r5").is_empty());
assert_eq!(plan_remote_pruning(&runs, 0, "r5").len(), 4);
}
#[test]
fn plan_remote_pruning_excludes_just_uploaded_even_if_oldest() {
let runs: Vec<String> = ["a", "b", "c"].iter().map(|s| (*s).to_owned()).collect();
assert_eq!(plan_remote_pruning(&runs, 1, "a"), vec!["b".to_owned()]);
}
#[test]
fn parse_offsite_ref_handles_latest_and_explicit() {
let latest = parse_offsite_ref("offsite:prod/latest").unwrap();
assert_eq!(latest.profile, "prod");
assert_eq!(latest.selector, OffsiteSelector::Latest);
assert_eq!(
parse_offsite_ref("prod/LATEST").unwrap().selector,
OffsiteSelector::Latest
);
let exact = parse_offsite_ref("offsite:prod/20260710T040506Z").unwrap();
assert_eq!(exact.profile, "prod");
assert_eq!(
exact.selector,
OffsiteSelector::Exact("20260710T040506Z".to_owned())
);
assert!(parse_offsite_ref("offsite:prod").is_none());
assert!(parse_offsite_ref("offsite:/latest").is_none());
assert!(parse_offsite_ref("prod/").is_none());
}
#[test]
fn resolve_offsite_ref_routes_local_only_when_not_indicated() {
assert!(resolve_offsite_ref(false, "backups/prod/20260710T040506Z").is_none());
assert!(resolve_offsite_ref(false, "latest").is_none());
let ok = resolve_offsite_ref(false, "offsite:prod/latest").unwrap();
assert_eq!(
ok.unwrap(),
OffsiteRef {
profile: "prod".to_owned(),
selector: OffsiteSelector::Latest,
}
);
let ok = resolve_offsite_ref(true, "prod/20260710T040506Z").unwrap();
assert_eq!(
ok.unwrap(),
OffsiteRef {
profile: "prod".to_owned(),
selector: OffsiteSelector::Exact("20260710T040506Z".to_owned()),
}
);
}
#[test]
fn resolve_offsite_ref_errors_on_malformed_when_indicated() {
let err = resolve_offsite_ref(true, "latest").unwrap().unwrap_err();
assert!(matches!(err, BackupError::InvalidOffsiteRef { .. }));
let msg = err.to_string();
assert!(msg.contains("latest"));
assert!(msg.contains("offsite:<profile>/<timestamp|latest>"));
assert!(matches!(
resolve_offsite_ref(false, "offsite:prod").unwrap(),
Err(BackupError::InvalidOffsiteRef { .. })
));
assert!(matches!(
resolve_offsite_ref(false, "offsite:/latest").unwrap(),
Err(BackupError::InvalidOffsiteRef { .. })
));
assert!(matches!(
resolve_offsite_ref(true, "").unwrap(),
Err(BackupError::InvalidOffsiteRef { .. })
));
}
#[test]
fn group_and_format_offsite_listing() {
let list_prefix = "db/prod/";
let objects = vec![
s3::S3Object {
key: "db/prod/20260710T040506Z/manifest.json".to_owned(),
size: 128,
},
s3::S3Object {
key: "db/prod/20260710T040506Z/control.dump".to_owned(),
size: 4096,
},
s3::S3Object {
key: "db/prod/20260709T040506Z/control.dump".to_owned(),
size: 2048,
},
];
let runs = group_offsite_objects(&objects, list_prefix);
assert_eq!(runs.len(), 2);
assert_eq!(runs[0].run_id, "20260709T040506Z");
assert_eq!(runs[1].run_id, "20260710T040506Z");
assert_eq!(runs[1].total, 128 + 4096);
let table = format_offsite_listing(&runs);
assert!(table.contains("RUN ID"));
assert!(table.contains("20260710T040506Z"));
assert!(table.contains("control.dump"));
assert!(table.contains("manifest.json"));
}
#[test]
fn human_size_scales_units() {
assert_eq!(human_size(512), "512 B");
assert_eq!(human_size(2048), "2.0 KiB");
assert_eq!(human_size(5 * 1024 * 1024), "5.0 MiB");
}
#[test]
#[ignore = "requires a live Postgres and pg_dump/pg_restore on PATH"]
fn backup_restore_round_trip() {
use diesel::connection::SimpleConnection as _;
use diesel::{Connection as _, PgConnection, RunQueryDsl as _, sql_query};
#[derive(diesel::QueryableByName)]
struct Count {
#[diesel(sql_type = diesel::sql_types::BigInt)]
n: i64,
}
#[derive(diesel::QueryableByName)]
struct Rt {
#[diesel(sql_type = diesel::sql_types::Integer)]
id: i32,
#[diesel(sql_type = diesel::sql_types::Text)]
name: String,
}
let Ok(url) = std::env::var("DATABASE_URL") else {
eprintln!("DATABASE_URL not set; skipping round-trip");
return;
};
let mut conn = PgConnection::establish(&url).expect("connect");
conn.batch_execute(
"DROP TABLE IF EXISTS backup_rt; \
CREATE TABLE backup_rt (id INT PRIMARY KEY, name TEXT NOT NULL); \
INSERT INTO backup_rt VALUES (1,'alpha'),(2,'beta'),(3,'gamma');",
)
.expect("seed");
let tmp = tempfile::tempdir().unwrap();
let tools = PgTools::locate();
let pg_dump = tools.require("pg_dump").expect("pg_dump present");
let run_dir = tmp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let targets = vec![ResolvedTarget {
label: "control".to_owned(),
url: url.clone(),
}];
backup_into(
&run_dir,
&targets,
BackupFormat::Custom,
&pg_dump,
&tools,
"dev",
)
.expect("backup succeeds and verifies");
conn.batch_execute("DELETE FROM backup_rt;")
.expect("delete");
let after_delete: Count = sql_query("SELECT COUNT(*) AS n FROM backup_rt")
.get_result(&mut conn)
.unwrap();
assert_eq!(after_delete.n, 0);
let artifact = run_dir.join("control.dump");
run_restore_one(&tools, &url, &artifact, BackupFormat::Custom, "dev")
.expect("restore succeeds");
let rows: Vec<Rt> = sql_query("SELECT id, name FROM backup_rt ORDER BY id")
.load(&mut conn)
.unwrap();
let got: Vec<(i32, &str)> = rows.iter().map(|r| (r.id, r.name.as_str())).collect();
assert_eq!(got, vec![(1, "alpha"), (2, "beta"), (3, "gamma")]);
conn.batch_execute("DROP TABLE backup_rt;").ok();
}
}