pub mod basebackup;
pub mod btrfs;
pub mod lvm;
pub mod resolve;
mod service;
pub mod strategy;
mod sys;
pub mod vss;
use std::{
collections::BTreeMap,
path::{Path, PathBuf},
};
use miette::{Context as _, IntoDiagnostic as _, Result, bail};
use tracing::{info, warn};
use self::strategy::Strategy;
use super::method::{PostgresqlConfig, Prepared, Teardown};
pub(super) fn stable_source_dir(backup_type: &str) -> PathBuf {
#[cfg(unix)]
{
PathBuf::from("/var/lib/bestool/backup-source").join(backup_type)
}
#[cfg(not(unix))]
{
let base = std::env::var_os("ProgramData")
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(r"C:\ProgramData"));
base.join("bestool").join("backup-source").join(backup_type)
}
}
fn ignore_globs() -> Vec<String> {
["postmaster.pid", "*.log", "pg_stat_tmp/*", "lost+found"]
.into_iter()
.map(String::from)
.collect()
}
fn metadata_tags(resolved: &resolve::ResolvedCluster, strategy: Strategy) -> BTreeMap<String, String> {
BTreeMap::from([
("pg-version".to_owned(), resolved.version.clone()),
("pg-cluster".to_owned(), resolved.cluster.clone()),
("pg-strategy".to_owned(), format!("{strategy:?}").to_lowercase()),
])
}
pub async fn prepare(config: &PostgresqlConfig, backup_type: &str) -> Result<Prepared> {
let resolved = resolve::resolve(config)?;
let strategy = strategy::detect(config.strategy.as_deref(), &resolved.data_dir)?;
info!(
cluster = %resolved.cluster,
version = %resolved.version,
?strategy,
data_dir = %resolved.data_dir.display(),
"preparing postgresql backup",
);
checkpoint(config, &resolved.data_dir).await;
match strategy {
Strategy::BaseBackup => basebackup_prepared(&resolved, backup_type, config).await,
snapshot => match snapshot_prepared(snapshot, &resolved, backup_type).await {
Ok(prepared) => Ok(prepared),
Err(err) => {
warn!(
strategy = ?snapshot,
"snapshot backend unavailable ({err}); falling back to pg_basebackup"
);
basebackup_prepared(&resolved, backup_type, config).await
}
},
}
}
async fn snapshot_prepared(
strategy: Strategy,
resolved: &resolve::ResolvedCluster,
backup_type: &str,
) -> Result<Prepared> {
let (path, teardown) = match strategy {
Strategy::Btrfs => {
let (path, mounts) = btrfs::prepare(resolved, backup_type).await?;
(path, Teardown::Btrfs(mounts))
}
Strategy::ThinLvm => {
let (path, snapshot) = lvm::prepare(resolved, backup_type).await?;
(path, Teardown::Lvm(snapshot))
}
Strategy::Vss => {
let (path, shadow) = vss::prepare(resolved, backup_type).await?;
(path, Teardown::Vss(shadow))
}
Strategy::BaseBackup => unreachable!("basebackup is handled by the caller"),
};
Ok(Prepared {
path,
extra_tags: metadata_tags(resolved, strategy),
ignore: ignore_globs(),
teardown,
})
}
async fn basebackup_prepared(
resolved: &resolve::ResolvedCluster,
backup_type: &str,
config: &PostgresqlConfig,
) -> Result<Prepared> {
let (path, root) = basebackup::prepare(resolved, backup_type, config).await?;
Ok(Prepared {
path,
extra_tags: metadata_tags(resolved, Strategy::BaseBackup),
ignore: ignore_globs(),
teardown: Teardown::BaseBackup(root),
})
}
pub async fn restore(
config: &PostgresqlConfig,
staging: &Path,
opts: &super::method::RestoreOpts,
) -> Result<()> {
let mut target = resolve::resolve_target(config)?;
let plan = resolve::plan_restore(staging, &target)?;
target.version = plan.data_major.clone();
info!(
cluster = %target.cluster,
version = %target.version,
dest = %plan.dest.display(),
whole_install = plan.whole_install,
"restoring postgres cluster",
);
super::method::ensure_not_clobbering(&plan.dest, opts.clobber)?;
if !plan.whole_install {
let major = plan.data_major.clone();
crate::interactive::retry("checking the installed postgres version", async || {
resolve::ensure_server_version_available(&major)
})
.await?;
}
crate::interactive::retry("stopping the postgres cluster", async || {
service::stop(&target, config).await
})
.await?;
crate::interactive::retry("moving the restored data into place", async || {
super::method::replace_dir(&plan.source, &plan.dest).await
})
.await?;
crate::interactive::retry("fixing data directory ownership", async || {
fix_ownership(&target.data_dir).await
})
.await?;
crate::interactive::retry_or_recover(
"starting the postgres cluster",
"reset the write-ahead log",
"force-reset the WAL so the cluster can start without replaying it — \
destructive: can discard recent transactions or corrupt an \
otherwise-healthy cluster; only sound for a backup that won't start any \
other way",
async || service::start(&target, config).await,
async || pg_resetwal(&target.data_dir, &target.version).await,
)
.await?;
verify(config, &target.data_dir, &target.version).await;
info!("restore complete; run migrations / config sync as needed");
Ok(())
}
#[cfg(unix)]
async fn fix_ownership(data_dir: &Path) -> Result<()> {
run_status("chown", &["-R", "postgres:postgres", path(data_dir)]).await?;
run_status("chmod", &["0750", path(data_dir)]).await
}
#[cfg(not(unix))]
async fn fix_ownership(_data_dir: &Path) -> Result<()> {
Ok(())
}
async fn pg_resetwal(data_dir: &Path, major: &str) -> Result<()> {
let mut cmd = pg_command(&postgres_bin_versioned("pg_resetwal", major, data_dir));
cmd.arg("-f").arg(data_dir);
run_checked(cmd, "pg_resetwal").await
}
async fn verify(config: &PostgresqlConfig, data_dir: &Path, major: &str) {
let mut cmd = pg_command(&postgres_bin_versioned("psql", major, data_dir));
cmd.args(["-X", "-q", "-w", "-tAc", "SELECT 1"]);
apply_connection(&mut cmd, config);
cmd.stdin(std::process::Stdio::null());
match cmd.status().await {
Ok(s) if s.success() => info!("restored cluster accepts connections"),
Ok(s) => warn!(%s, "post-restore verification query failed"),
Err(err) => warn!("could not run verification query: {err}"),
}
}
#[cfg(unix)]
fn path(p: &Path) -> &str {
p.to_str().unwrap_or_default()
}
#[cfg(unix)]
pub(super) async fn run_status(program: &str, args: &[&str]) -> Result<()> {
let status = tokio::process::Command::new(program)
.args(args)
.stdin(std::process::Stdio::null())
.status()
.await
.into_diagnostic()
.wrap_err_with(|| format!("spawning {program}"))?;
if !status.success() {
bail!("{program} {} failed ({status})", args.join(" "));
}
Ok(())
}
pub(super) fn postgres_bin(name: &str, data_dir: &Path) -> String {
#[cfg(windows)]
if let Some(candidate) = bin_beside_data_dir(name, data_dir).filter(|p| p.is_file()) {
return candidate.to_string_lossy().into_owned();
}
#[cfg(not(windows))]
let _ = data_dir;
crate::find_postgres::find_postgres_bin(name)
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_else(|_| name.to_owned())
}
fn postgres_bin_versioned(name: &str, major: &str, data_dir: &Path) -> String {
#[cfg(unix)]
{
let candidate = PathBuf::from("/usr/lib/postgresql")
.join(major)
.join("bin")
.join(name);
if candidate.is_file() {
return candidate.to_string_lossy().into_owned();
}
}
#[cfg(not(unix))]
let _ = major;
postgres_bin(name, data_dir)
}
#[cfg(any(windows, test))]
fn bin_beside_data_dir(name: &str, data_dir: &Path) -> Option<PathBuf> {
let exe = if cfg!(windows) {
format!("{name}.exe")
} else {
name.to_owned()
};
data_dir.parent().map(|p| p.join("bin").join(exe))
}
pub(super) fn pg_command(bin: &str) -> tokio::process::Command {
#[cfg(unix)]
{
let mut cmd = tokio::process::Command::new("sudo");
cmd.args(["-u", "postgres", bin]);
cmd
}
#[cfg(not(unix))]
{
tokio::process::Command::new(bin)
}
}
pub(super) fn apply_connection(cmd: &mut tokio::process::Command, config: &PostgresqlConfig) {
if let Some(url) = &config.connection_url {
cmd.arg("-d").arg(url);
return;
}
if let Some(socket) = &config.socket {
cmd.arg("-h").arg(socket);
}
if let Some(port) = config.port {
cmd.arg("-p").arg(port.to_string());
}
}
async fn run_checked(mut cmd: tokio::process::Command, what: &str) -> Result<()> {
let status = cmd
.stdin(std::process::Stdio::null())
.status()
.await
.into_diagnostic()
.wrap_err_with(|| format!("spawning {what}"))?;
if !status.success() {
bail!("{what} failed ({status})");
}
Ok(())
}
async fn checkpoint(config: &PostgresqlConfig, data_dir: &Path) {
let mut cmd = pg_command(&postgres_bin("psql", data_dir));
cmd.args(["-X", "-q", "-w"]);
apply_connection(&mut cmd, config);
cmd.args(["-c", "CHECKPOINT;"]);
cmd.stdin(std::process::Stdio::null());
match cmd.status().await {
Ok(status) if status.success() => info!("issued CHECKPOINT before snapshot"),
Ok(status) => warn!(
%status,
"CHECKPOINT failed; snapshot is still crash-consistent, recovery may just replay more WAL"
),
Err(err) => warn!("could not run CHECKPOINT (continuing): {err}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn bin_beside_data_dir_is_sibling_of_data() {
let candidate = bin_beside_data_dir("pg_basebackup", Path::new("/opt/pg/16/data")).unwrap();
let name = if cfg!(windows) {
"pg_basebackup.exe"
} else {
"pg_basebackup"
};
assert_eq!(candidate, Path::new("/opt/pg/16/bin").join(name));
}
#[test]
fn ignore_globs_never_include_required_dirs() {
let globs = ignore_globs();
assert!(globs.contains(&"postmaster.pid".to_owned()));
for required in ["pg_wal", "pg_xact", "pg_control", "global"] {
assert!(
!globs.iter().any(|g| g.contains(required)),
"{required} must never be ignored"
);
}
}
fn pg_config(connection_url: Option<&str>, socket: Option<&str>, port: Option<u16>) -> PostgresqlConfig {
PostgresqlConfig {
cluster: "main".into(),
data_dir: None,
version: None,
connection_url: connection_url.map(str::to_owned),
port,
socket: socket.map(PathBuf::from),
strategy: None,
service_name: None,
}
}
#[test]
fn apply_connection_prefers_the_url() {
let mut cmd = tokio::process::Command::new("psql");
apply_connection(&mut cmd, &pg_config(Some("postgresql://u:p@h/db"), Some("/run/pg"), Some(5433)));
let args: Vec<_> = cmd
.as_std()
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["-d", "postgresql://u:p@h/db"]);
}
#[test]
fn apply_connection_falls_back_to_socket_and_port() {
let mut cmd = tokio::process::Command::new("psql");
apply_connection(&mut cmd, &pg_config(None, Some("/run/pg"), Some(5433)));
let args: Vec<_> = cmd
.as_std()
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["-h", "/run/pg", "-p", "5433"]);
}
#[test]
fn metadata_tags_carry_version_cluster_strategy() {
let resolved = resolve::ResolvedCluster {
data_dir: "/var/lib/postgresql/16/main".into(),
version: "16".into(),
cluster: "main".into(),
};
let tags = metadata_tags(&resolved, Strategy::Btrfs);
assert_eq!(tags.get("pg-version").map(String::as_str), Some("16"));
assert_eq!(tags.get("pg-cluster").map(String::as_str), Some("main"));
assert_eq!(tags.get("pg-strategy").map(String::as_str), Some("btrfs"));
}
}