use std::path::{Path, PathBuf};
use tracing::{info, warn};
use super::Chronix;
use super::{BackupManifest, catalog_dir_of, chrono_timestamp_ms, segments_dir_of, wal_dir_of};
use crate::error::{DbError, Result};
const MANIFEST_NAME: &str = "backup_manifest.json";
impl Chronix {
#[must_use = "backup errors indicate incomplete backup"]
pub fn backup(&self, target_dir: &Path) -> Result<BackupManifest> {
self.check_open()?;
let target_abs = std::path::absolute(target_dir).unwrap_or_else(|_| target_dir.to_owned());
let data_abs = std::path::absolute(&self.config.data_dir)
.unwrap_or_else(|_| self.config.data_dir.clone());
let source_segments_probe = segments_dir_of(&data_abs);
if target_abs == data_abs
|| target_abs.starts_with(&source_segments_probe)
|| source_segments_probe.starts_with(&target_abs)
{
return Err(DbError::InvalidRequest(format!(
"refusing to check point {} into its own segment directory",
self.config.data_dir.display()
)));
}
let started = std::time::Instant::now();
let checkpoint =
|| -> Result<BackupManifest> { self.checkpoint_inner(target_dir, started) };
let result = checkpoint();
if result.is_err() {
metrics::counter!("chronix_backup_failures_total").increment(1);
}
result
}
fn checkpoint_inner(
&self,
target_dir: &Path,
started: std::time::Instant,
) -> Result<BackupManifest> {
self.flush()?;
info!(target = %target_dir.display(), "Starting database checkpoint");
let target_segments = segments_dir_of(target_dir);
let target_catalog = catalog_dir_of(target_dir);
std::fs::create_dir_all(&target_segments)?;
std::fs::create_dir_all(wal_dir_of(target_dir))?;
let _ = std::fs::remove_file(target_dir.join(MANIFEST_NAME));
let (files, _lease) = {
let catalog = self.catalog.read();
let files: Vec<chronix_core::SegmentFile> = catalog
.all_segments()
.iter()
.map(|e| e.file.clone())
.collect();
let lease = self
.segment_leases
.acquire(catalog.all_segments().iter().map(|e| e.segment_id));
catalog
.write_snapshot_to(&target_catalog)
.map_err(|e| DbError::Internal(format!("catalog snapshot failed: {e}")))?;
(files, lease)
};
let source_segments = self.segments_dir();
let mut file_count = 0usize;
let mut total_bytes = 0u64;
let mut wanted: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();
for file in &files {
let relative = file.as_relative();
let src = source_segments.join(relative);
let dst = target_segments.join(relative);
wanted.insert(dst.clone());
total_bytes += Self::place(&src, &dst)?;
file_count += 1;
let src_side = chronix_engine::index::series_index::sidecar_path(&src);
if src_side.exists() {
let dst_side = chronix_engine::index::series_index::sidecar_path(&dst);
wanted.insert(dst_side.clone());
total_bytes += Self::place(&src_side, &dst_side)?;
file_count += 1;
}
}
Self::prune_unlisted(&target_segments, &wanted);
let manifest = BackupManifest {
version: 2,
created_at: chrono_timestamp_ms(),
wal_sequence: self.wal.current_sequence(),
segments: files.len() as u64,
file_count,
total_bytes,
};
let manifest_json = serde_json::to_string_pretty(&manifest)
.map_err(|e| DbError::Internal(format!("manifest serialization: {e}")))?;
std::fs::write(target_dir.join(MANIFEST_NAME), manifest_json)?;
metrics::histogram!("chronix_backup_duration_seconds")
.record(started.elapsed().as_secs_f64());
metrics::counter!("chronix_backups_total").increment(1);
metrics::counter!("chronix_backup_bytes_total").increment(total_bytes);
info!(
target = %target_dir.display(),
segments = manifest.segments,
files = file_count,
bytes = total_bytes,
wal_seq = manifest.wal_sequence,
duration_ms = started.elapsed().as_millis(),
"Checkpoint complete"
);
Ok(manifest)
}
fn place(src: &Path, dst: &Path) -> Result<u64> {
if let Some(parent) = dst.parent() {
std::fs::create_dir_all(parent)?;
}
if dst.exists() {
std::fs::remove_file(dst)?;
}
match std::fs::hard_link(src, dst) {
Ok(()) => {}
Err(_) => {
std::fs::copy(src, dst)?;
}
}
Ok(std::fs::metadata(dst)?.len())
}
fn prune_unlisted(dir: &Path, keep: &std::collections::HashSet<PathBuf>) {
let Ok(shards) = std::fs::read_dir(dir) else {
return;
};
for shard in shards.flatten() {
let shard_path = shard.path();
if !shard_path.is_dir() {
continue;
}
let Ok(entries) = std::fs::read_dir(&shard_path) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
if keep.contains(&path) {
continue;
}
if let Err(e) = std::fs::remove_file(&path) {
warn!(path = %path.display(), error = %e, "backup: stale file left behind");
}
}
}
}
#[must_use = "restore errors indicate incomplete restore"]
pub fn restore(backup_dir: &Path, target_dir: &Path) -> Result<BackupManifest> {
let manifest = Self::read_manifest(backup_dir)?;
if target_dir.exists() {
return Err(DbError::InvalidRequest(format!(
"target directory {} already exists — aborting restore to prevent data loss",
target_dir.display()
)));
}
info!(
source = %backup_dir.display(),
target = %target_dir.display(),
wal_seq = manifest.wal_sequence,
"Restoring database from checkpoint"
);
Self::verify(backup_dir, &manifest)?;
std::fs::create_dir_all(target_dir)?;
let copied = (|| -> Result<()> {
let segments_src = segments_dir_of(backup_dir);
let segments_dst = segments_dir_of(target_dir);
std::fs::create_dir_all(&segments_dst)?;
if segments_src.exists() {
Self::link_dir_recursive(&segments_src, &segments_dst)?;
}
for dir_name in ["wal", "catalog"] {
let src = backup_dir.join(dir_name);
let dst = target_dir.join(dir_name);
std::fs::create_dir_all(&dst)?;
if src.exists() {
Self::copy_dir_recursive(&src, &dst)?;
}
}
Ok(())
})();
if let Err(e) = copied {
let _ = std::fs::remove_dir_all(target_dir);
return Err(e);
}
info!(target = %target_dir.display(), "Restore complete");
Ok(manifest)
}
pub fn verify_backup(backup_dir: &Path) -> Result<BackupManifest> {
let manifest = Self::read_manifest(backup_dir)?;
Self::verify(backup_dir, &manifest)?;
Ok(manifest)
}
fn read_manifest(backup_dir: &Path) -> Result<BackupManifest> {
let manifest_path = backup_dir.join(MANIFEST_NAME);
if !manifest_path.exists() {
return Err(DbError::InvalidRequest(format!(
"{MANIFEST_NAME} not found — {} is not a backup directory",
backup_dir.display()
)));
}
let data = std::fs::read_to_string(&manifest_path)?;
serde_json::from_str(&data)
.map_err(|e| DbError::InvalidRequest(format!("corrupt backup manifest: {e}")))
}
fn verify(backup_dir: &Path, manifest: &BackupManifest) -> Result<()> {
let catalog = chronix_engine::index::SegmentCatalog::open(catalog_dir_of(backup_dir))
.map_err(|e| DbError::InvalidRequest(format!("backup catalog is unreadable: {e}")))?;
let segments_dir = segments_dir_of(backup_dir);
let mut missing: Vec<String> = Vec::new();
let mut short: Vec<String> = Vec::new();
let entries = catalog.all_segments();
for entry in &entries {
let path = entry.file.resolve(&segments_dir);
match std::fs::metadata(&path) {
Err(_) => missing.push(entry.file.to_string()),
Ok(m) if m.len() != entry.byte_size => short.push(format!(
"{} ({} bytes, catalog says {})",
entry.file,
m.len(),
entry.byte_size
)),
Ok(_) => {}
}
}
if !missing.is_empty() || !short.is_empty() {
return Err(DbError::InvalidRequest(format!(
"incomplete backup in {}: {} segment(s) missing{}, {} truncated{}",
backup_dir.display(),
missing.len(),
if missing.is_empty() {
String::new()
} else {
format!(" ({})", missing.join(", "))
},
short.len(),
if short.is_empty() {
String::new()
} else {
format!(" ({})", short.join(", "))
},
)));
}
if entries.len() as u64 != manifest.segments {
return Err(DbError::InvalidRequest(format!(
"backup in {} holds {} segments, its manifest claims {}",
backup_dir.display(),
entries.len(),
manifest.segments
)));
}
Ok(())
}
fn link_dir_recursive(src: &Path, dst: &Path) -> Result<()> {
std::fs::create_dir_all(dst)?;
for entry in std::fs::read_dir(src)? {
let entry = entry?;
let src_path = entry.path();
let dst_path = dst.join(entry.file_name());
if src_path.is_dir() {
Self::link_dir_recursive(&src_path, &dst_path)?;
} else {
Self::place(&src_path, &dst_path)?;
}
}
Ok(())
}
fn copy_dir_recursive(src: &Path, dst: &Path) -> Result<(usize, u64)> {
std::fs::create_dir_all(dst)?;
let mut files = 0usize;
let mut bytes = 0u64;
for entry in std::fs::read_dir(src)? {
let entry = entry?;
let src_path = entry.path();
let dst_path = dst.join(entry.file_name());
if src_path.is_dir() {
let (f, b) = Self::copy_dir_recursive(&src_path, &dst_path)?;
files += f;
bytes += b;
} else {
let b = std::fs::copy(&src_path, &dst_path)?;
files += 1;
bytes += b;
}
}
Ok((files, bytes))
}
}
impl Chronix {
pub(crate) fn scheduled_checkpoint(&self) -> Result<Option<PathBuf>> {
let settings = &self.config.checkpoints;
let (Some(_), Some(root)) = (settings.interval(), settings.directory.as_ref()) else {
return Ok(None);
};
std::fs::create_dir_all(root)?;
let stamp = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
let target = root.join(&stamp);
if target.exists() {
return Ok(None);
}
self.backup(&target)?;
Self::prune_checkpoints(root, settings.keep);
Ok(Some(target))
}
fn prune_checkpoints(root: &Path, keep: usize) {
let Ok(entries) = std::fs::read_dir(root) else {
return;
};
let mut complete: Vec<PathBuf> = Vec::new();
for entry in entries.flatten() {
let path = entry.path();
if !path.is_dir() {
continue;
}
if path.join(MANIFEST_NAME).exists() {
complete.push(path);
} else if let Err(e) = std::fs::remove_dir_all(&path) {
warn!(path = %path.display(), error = %e, "checkpoint: could not remove an unfinished run");
}
}
complete.sort();
let keep = keep.max(1);
if complete.len() <= keep {
return;
}
for path in &complete[..complete.len() - keep] {
match std::fs::remove_dir_all(path) {
Ok(()) => {
tracing::debug!(path = %path.display(), "checkpoint: pruned");
}
Err(e) => {
warn!(path = %path.display(), error = %e, "checkpoint: could not prune");
}
}
}
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use chronix_core::ChronixConfig;
use super::*;
#[test]
fn a_retirement_honours_the_lease_a_checkpoint_takes() {
use chronix_core::{FieldValue, Point, SeriesKey};
let dir = tempfile::tempdir().unwrap();
let db = Chronix::open(
ChronixConfig::builder()
.data_dir(dir.path())
.maintenance_interval(Duration::from_secs(86_400 * 365))
.build()
.unwrap(),
)
.unwrap();
let now = i64::try_from(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos(),
)
.unwrap();
const HOUR: i64 = 3_600_000_000_000;
let key = SeriesKey::new(
"cpu",
std::collections::BTreeMap::from([("host".to_owned(), "a".to_owned())]),
)
.unwrap();
let old: Vec<Point> = (0..200i64)
.map(|i| {
let fields =
std::collections::BTreeMap::from([("v".to_owned(), FieldValue::F64(i as f64))]);
Point::new(key.clone(), fields, now - 120 * HOUR + i * 1_000_000).unwrap()
})
.collect();
db.backfill(&old).unwrap().into_complete().unwrap();
db.flush().unwrap();
let fresh = SeriesKey::new(
"cpu",
std::collections::BTreeMap::from([("host".to_owned(), "b".to_owned())]),
)
.unwrap();
db.insert(
&Point::new(
fresh,
std::collections::BTreeMap::from([("v".to_owned(), FieldValue::F64(1.0))]),
now,
)
.unwrap(),
)
.unwrap();
db.flush().unwrap();
let (paths, lease) = {
let catalog = db.catalog.read();
let paths: Vec<std::path::PathBuf> = catalog
.all_segments()
.iter()
.map(|e| e.file.resolve(&db.segments_dir()))
.collect();
let lease = db
.segment_leases
.acquire(catalog.all_segments().iter().map(|e| e.segment_id));
(paths, lease)
};
assert!(
!paths.is_empty(),
"the flush has to have produced a segment"
);
let result = db
.enforce_retention(Duration::from_secs(2 * 86_400))
.unwrap();
assert!(
result.segments_deleted > 0,
"the pass has to remove something for this test to mean anything"
);
assert!(
result.segments_awaiting_readers > 0,
"the checkpoint holds those segments, so their files cannot have gone"
);
for path in &paths {
assert!(
path.exists(),
"a checkpoint copying {} must still find it",
path.display()
);
}
drop(lease);
db.gc().unwrap();
assert!(
paths.iter().any(|p| !p.exists()),
"the last reader is gone, so the expired segments must be reclaimed"
);
db.close().unwrap();
}
}