use std::{
collections::BTreeMap,
path::{Path, PathBuf},
time::{Duration, Instant},
};
use jiff::Timestamp;
use miette::{Result, bail};
use serde::Deserialize;
use tracing::debug;
mod holders;
#[derive(Debug)]
pub struct Prepared {
pub path: PathBuf,
pub taken_at: Option<Timestamp>,
pub extra_tags: BTreeMap<String, String>,
pub ignore: Vec<String>,
pub(super) teardown: Teardown,
}
#[derive(Debug)]
pub(super) enum Teardown {
Simple(super::simple::Cleanup),
Btrfs(super::postgresql::btrfs::Mounts),
Lvm(super::postgresql::lvm::Snapshot),
#[cfg(windows)]
Vss(super::postgresql::vss::Shadow),
BaseBackup(PathBuf),
}
#[derive(Debug, Clone, Deserialize)]
pub struct SimpleConfig {
pub path: PathBuf,
}
#[derive(Debug, Clone, Deserialize)]
pub struct PostgresqlConfig {
pub cluster: String,
#[serde(default)]
pub data_dir: Option<PathBuf>,
#[serde(default)]
pub version: Option<String>,
#[serde(default)]
pub connection_url: Option<String>,
#[serde(default)]
pub port: Option<u16>,
#[serde(default)]
pub socket: Option<PathBuf>,
#[serde(default)]
pub strategy: Option<String>,
#[serde(default)]
pub staging_dir: Option<PathBuf>,
#[serde(default)]
#[cfg_attr(
not(windows),
expect(dead_code, reason = "only read on Windows; Unix derives the systemd unit")
)]
pub service_name: Option<String>,
}
#[derive(Debug, Clone)]
pub enum Method {
Simple(SimpleConfig),
Postgresql(PostgresqlConfig),
}
impl Method {
pub fn name(&self) -> &'static str {
match self {
Method::Simple(_) => "simple",
Method::Postgresql(_) => "postgresql",
}
}
pub async fn prepare(&self, backup_type: &str) -> Result<Prepared> {
match self {
Method::Simple(config) => {
let (path, cleanup) = super::simple::prepare(&config.path, backup_type).await?;
Ok(Prepared {
path,
taken_at: None,
extra_tags: BTreeMap::new(),
ignore: Vec::new(),
teardown: Teardown::Simple(cleanup),
})
}
Method::Postgresql(config) => super::postgresql::prepare(config, backup_type).await,
}
}
pub fn supports_hold(&self) -> bool {
match self {
Method::Simple(_) => false,
Method::Postgresql(_) => true,
}
}
pub(super) async fn hold(
&self,
prepared: Prepared,
id: &str,
) -> Result<(PathBuf, super::hold::HeldCapture)> {
let source = prepared.path;
match prepared.teardown {
Teardown::Simple(cleanup) => {
super::simple::teardown(cleanup).await?;
bail!(
"the simple method captures a live view of {}, not a point in time, \
so it cannot be held as a rollback point",
source.display()
)
}
Teardown::Btrfs(mounts) => super::postgresql::btrfs::hold(mounts, id, &source).await,
Teardown::Lvm(snapshot) => super::postgresql::lvm::hold(snapshot, id, &source).await,
#[cfg(windows)]
Teardown::Vss(shadow) => super::postgresql::vss::hold(shadow, id, &source).await,
Teardown::BaseBackup(root) => super::postgresql::basebackup::hold(root, id, &source).await,
}
}
pub async fn cleanup(&self, prepared: Prepared) -> Result<()> {
match prepared.teardown {
Teardown::Simple(cleanup) => super::simple::teardown(cleanup).await,
Teardown::Btrfs(mounts) => super::postgresql::btrfs::teardown(mounts).await,
Teardown::Lvm(snapshot) => super::postgresql::lvm::teardown(snapshot).await,
#[cfg(windows)]
Teardown::Vss(shadow) => super::postgresql::vss::teardown(shadow).await,
Teardown::BaseBackup(root) => super::postgresql::basebackup::teardown(root).await,
}
}
pub fn staging_dir(&self, target_override: Option<&Path>, pid: u32) -> PathBuf {
let parent = match self {
Method::Simple(config) => target_override
.map(Path::to_path_buf)
.unwrap_or_else(|| config.path.clone())
.parent()
.map(Path::to_path_buf)
.unwrap_or_else(std::env::temp_dir),
Method::Postgresql(config) => super::postgresql::resolve::restore_staging_parent(config),
};
parent.join(format!(".bestool-restore.{pid}"))
}
pub async fn restore(&self, staging: &Path, opts: &RestoreOpts) -> Result<()> {
match self {
Method::Simple(config) => {
let target = opts.target.clone().unwrap_or_else(|| config.path.clone());
ensure_not_clobbering(&target, opts.clobber)?;
replace_dir(staging, &target).await
}
Method::Postgresql(config) => super::postgresql::restore(config, staging, opts).await,
}
}
}
#[derive(Debug, Clone, Default)]
pub struct RestoreOpts {
pub target: Option<PathBuf>,
pub clobber: bool,
}
pub fn ensure_not_clobbering(target: &Path, clobber: bool) -> Result<()> {
if clobber || !dir_has_entries(target) {
return Ok(());
}
bail!(
"{} already contains data; refusing to overwrite without confirmation \
(pass --clobber-existing-data-yes-i-am-sure, or confirm interactively)",
target.display()
);
}
pub fn dir_has_entries(path: &Path) -> bool {
std::fs::read_dir(path)
.map(|mut it| it.next().is_some())
.unwrap_or(false)
}
pub(super) async fn replace_dir(staging: &Path, target: &Path) -> Result<()> {
use miette::{Context as _, IntoDiagnostic as _};
if target.exists() {
let backup = with_extension_suffix(target, "old");
if backup.exists() {
tokio::fs::remove_dir_all(&backup)
.await
.into_diagnostic()
.wrap_err_with(|| format!("removing stale {}", backup.display()))?;
}
rename_when_free(target, &backup).await.wrap_err_with(|| {
format!(
"moving {} aside to {}{}",
target.display(),
backup.display(),
holders::describe_holders(target)
)
})?;
}
if let Some(parent) = target.parent() {
tokio::fs::create_dir_all(parent).await.ok();
}
rename_when_free(staging, target)
.await
.wrap_err_with(|| format!("moving restored data into {}", target.display()))
}
const BUSY_RETRY_FOR: Duration = Duration::from_secs(10);
async fn rename_when_free(from: &Path, to: &Path) -> Result<()> {
use miette::IntoDiagnostic as _;
let deadline = Instant::now() + BUSY_RETRY_FOR;
let mut backoff = Duration::from_millis(100);
loop {
match tokio::fs::rename(from, to).await {
Ok(()) => return Ok(()),
Err(err) if is_busy(&err) && Instant::now() + backoff < deadline => {
debug!("{} is busy ({err}); retrying in {backoff:?}", from.display());
tokio::time::sleep(backoff).await;
backoff = (backoff * 2).min(Duration::from_secs(1));
}
Err(err) if is_cross_device(&err) => {
bail!(
"cannot move {} to {}: they are on different filesystems, so the \
swap can't be an atomic rename. Set `staging_dir` in the backup \
def to a path on the same filesystem as {}, then retry.",
from.display(),
to.display(),
to.display(),
);
}
Err(err) => return Err(err).into_diagnostic(),
}
}
}
fn is_busy(err: &std::io::Error) -> bool {
cfg!(windows)
&& matches!(
err.raw_os_error(),
Some(5 | 32 )
)
}
fn is_cross_device(err: &std::io::Error) -> bool {
#[cfg(windows)]
const CODE: i32 = 17; #[cfg(not(windows))]
const CODE: i32 = 18; err.raw_os_error() == Some(CODE)
}
pub(super) fn with_extension_suffix(path: &Path, suffix: &str) -> PathBuf {
let mut name = path.file_name().unwrap_or_default().to_os_string();
name.push(".");
name.push(suffix);
path.with_file_name(name)
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(not(target_os = "linux"))]
#[tokio::test]
async fn simple_prepare_returns_its_path_and_no_tags() {
let method = Method::Simple(SimpleConfig {
path: PathBuf::from("/data/custom"),
});
let prepared = method.prepare("custom").await.unwrap();
assert_eq!(prepared.path, PathBuf::from("/data/custom"));
assert!(prepared.extra_tags.is_empty());
assert!(prepared.ignore.is_empty());
method.cleanup(prepared).await.unwrap();
}
#[test]
fn clobber_guard_blocks_occupied_dir_unless_forced() {
let tmp = tempfile::tempdir().unwrap();
let occupied = tmp.path().join("data");
std::fs::create_dir_all(&occupied).unwrap();
std::fs::write(occupied.join("PG_VERSION"), "16").unwrap();
assert!(ensure_not_clobbering(&occupied, false).is_err());
assert!(ensure_not_clobbering(&occupied, true).is_ok());
assert!(ensure_not_clobbering(&tmp.path().join("absent"), false).is_ok());
let empty = tmp.path().join("empty");
std::fs::create_dir_all(&empty).unwrap();
assert!(ensure_not_clobbering(&empty, false).is_ok());
}
#[test]
fn extension_suffix_appends() {
assert_eq!(
with_extension_suffix(Path::new("/var/lib/postgresql/16/main"), "old"),
PathBuf::from("/var/lib/postgresql/16/main.old")
);
}
#[test]
fn only_windows_busy_errors_are_worth_retrying() {
let denied = std::io::Error::from_raw_os_error(5);
let missing = std::io::Error::from_raw_os_error(2);
assert_eq!(is_busy(&denied), cfg!(windows));
assert!(!is_busy(&missing));
assert!(!is_busy(&std::io::Error::other("no os code")));
}
#[test]
fn cross_device_is_recognised_and_distinct_from_busy() {
#[cfg(not(windows))]
let (same_dev, other_dev) = (18, 17);
#[cfg(windows)]
let (same_dev, other_dev) = (17, 18);
assert!(is_cross_device(&std::io::Error::from_raw_os_error(same_dev)));
assert!(!is_cross_device(&std::io::Error::from_raw_os_error(other_dev)));
let xdev = std::io::Error::from_raw_os_error(same_dev);
assert!(!(is_busy(&xdev) && is_cross_device(&xdev)));
for busy in [5, 32] {
assert!(!is_cross_device(&std::io::Error::from_raw_os_error(busy)));
}
}
#[tokio::test]
async fn a_hopeless_rename_fails_without_burning_the_retry_budget() {
let tmp = tempfile::tempdir().unwrap();
let started = Instant::now();
rename_when_free(&tmp.path().join("absent"), &tmp.path().join("dest"))
.await
.unwrap_err();
assert!(started.elapsed() < BUSY_RETRY_FOR);
}
#[tokio::test]
async fn replace_dir_keeps_old_and_moves_in() {
let tmp = tempfile::tempdir().unwrap();
let target = tmp.path().join("data");
std::fs::create_dir_all(&target).unwrap();
std::fs::write(target.join("old-marker"), "x").unwrap();
let staging = tmp.path().join("staging");
std::fs::create_dir_all(&staging).unwrap();
std::fs::write(staging.join("new-marker"), "y").unwrap();
replace_dir(&staging, &target).await.unwrap();
assert!(target.join("new-marker").exists());
assert!(!target.join("old-marker").exists());
assert!(tmp.path().join("data.old").join("old-marker").exists());
}
}