From 7c262ce14ef54361325fccfa1ed480996348513f Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Sun, 18 Oct 2026 19:01:54 +0200 Subject: [PATCH] fix: files refactoring --- docker-compose.yml | 2 +- justfile | 6 +- scripts/README.md | 1 + src/domain/files/backup.rs | 26 +- src/domain/files/matcher.rs | 39 ++- src/domain/files/mod.rs | 2 + src/domain/files/ping.rs | 1 - src/domain/files/platform.rs | 1 - src/domain/files/restic/backup.rs | 106 +++++++++ .../files}/restic/excludes.rs | 31 +-- src/{services => domain/files}/restic/json.rs | 11 - src/domain/files/restic/mod.rs | 6 + src/domain/files/restic/password.rs | 7 + .../files/restic/repo.rs} | 35 +-- .../files}/restic/restore.rs | 35 +-- src/domain/files/restore.rs | 43 +--- src/domain/files/sync/backup.rs | 45 ++++ .../files}/sync/excludes.rs | 12 +- src/domain/files/sync/mod.rs | 4 + src/{services => domain/files}/sync/rclone.rs | 17 +- src/{services => domain/files}/sync/stats.rs | 6 - src/services/backup/executor.rs | 13 +- src/services/backup/files_engine.rs | 166 +++++++++++++ src/services/backup/mod.rs | 1 + src/services/dashboard_config.rs | 1 - src/services/mod.rs | 2 - src/services/restic/backup.rs | 224 ------------------ src/services/restic/mod.rs | 7 - src/services/restic/password.rs | 42 ---- src/services/restore/executor.rs | 34 ++- src/services/status.rs | 4 - src/services/storage/providers/local.rs | 8 + src/services/sync/backup.rs | 147 ------------ src/services/sync/mod.rs | 6 - src/tests/domain/{files.rs => files/mod.rs} | 3 + .../files/restic.rs} | 45 +--- .../sync_tests.rs => domain/files/sync.rs} | 20 +- src/tests/services/api_models_tests.rs | 2 +- src/tests/services/local_storage_tests.rs | 22 +- src/tests/services/mod.rs | 2 - src/tests/storage/rclone_target.rs | 8 +- src/utils/edge_key.rs | 21 ++ src/utils/file.rs | 13 +- 43 files changed, 503 insertions(+), 724 deletions(-) create mode 100644 src/domain/files/restic/backup.rs rename src/{services => domain/files}/restic/excludes.rs (54%) rename src/{services => domain/files}/restic/json.rs (75%) create mode 100644 src/domain/files/restic/mod.rs create mode 100644 src/domain/files/restic/password.rs rename src/{services/restic/command.rs => domain/files/restic/repo.rs} (80%) rename src/{services => domain/files}/restic/restore.rs (68%) create mode 100644 src/domain/files/sync/backup.rs rename src/{services => domain/files}/sync/excludes.rs (68%) create mode 100644 src/domain/files/sync/mod.rs rename src/{services => domain/files}/sync/rclone.rs (80%) rename src/{services => domain/files}/sync/stats.rs (80%) create mode 100644 src/services/backup/files_engine.rs delete mode 100644 src/services/restic/backup.rs delete mode 100644 src/services/restic/mod.rs delete mode 100644 src/services/restic/password.rs delete mode 100644 src/services/sync/backup.rs delete mode 100644 src/services/sync/mod.rs rename src/tests/domain/{files.rs => files/mod.rs} (99%) rename src/tests/{services/restic_tests.rs => domain/files/restic.rs} (87%) rename src/tests/{services/sync_tests.rs => domain/files/sync.rs} (93%) diff --git a/docker-compose.yml b/docker-compose.yml index ab4b9b7..c8a0fc2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -24,7 +24,7 @@ services: LOG: debug TZ: "Europe/Paris" # TMPDIR: /scratch - EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiODBkZTY1NDktNzZjYS00NGZhLWIyNTItMTY0MWI2YTQzZjJkIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ==" + EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNGUyMzgxYzgtNzI4Mi00MzIxLTk1ZjQtYTU3NTRiZTc2YmFlIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ==" #CHUNK_SIZE_MB: "1" #POOLING: 1 #RETRY_ATTEMPTS: 3 diff --git a/justfile b/justfile index ac701de..390e720 100644 --- a/justfile +++ b/justfile @@ -79,4 +79,8 @@ test: docker compose -f docker-compose.test.yml down --volumes seed-files: - for d in archive snapshot sync; do mkdir -p test-files/$d && echo one > test-files/$d/a.txt && echo two > test-files/$d/b.txt; done \ No newline at end of file + for d in archive snapshot sync; do mkdir -p test-files/$d && echo one > test-files/$d/a.txt && echo two > test-files/$d/b.txt; done + +seed-files-1gb: + echo "Seeding files sources (1GB each)..." + bash ./scripts/files/seed-1gb.sh diff --git a/scripts/README.md b/scripts/README.md index f94a41e..4b4b58a 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -11,6 +11,7 @@ make seed-postgres-1gb make seed-all make seed-sqlite make seed-sqlite SEED=big +just seed-files-1gb ``` ## Verify commands diff --git a/src/domain/files/backup.rs b/src/domain/files/backup.rs index 4169eac..153afbd 100644 --- a/src/domain/files/backup.rs +++ b/src/domain/files/backup.rs @@ -1,4 +1,4 @@ -use super::matcher::ExcludeMatcher; +use super::matcher::{ExcludeMatcher, one_file_system, source_root}; use super::platform::dev; use crate::services::backup::logger::JobLogger; use crate::services::config::DatabaseConfig; @@ -20,20 +20,11 @@ pub async fn run( } fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Result { - // Canonical paths so the output location can be recognised while walking. - let root = std::fs::canonicalize(&cfg.path) - .with_context(|| format!("cannot read source directory {}", cfg.path))?; - if !root.is_dir() { - bail!("source path {} is not a directory", root.display()); - } + let root = source_root(cfg)?; let backup_real = std::fs::canonicalize(backup_dir) .with_context(|| format!("cannot read backup directory {}", backup_dir.display()))?; let matcher = ExcludeMatcher::from_config(cfg)?; - let one_file_system = cfg - .options - .get("one_file_system") - .and_then(|v| v.as_bool()) - .unwrap_or(false); + let one_file_system = one_file_system(cfg); let root_dev = dev(&std::fs::metadata(&root) .with_context(|| format!("cannot read source directory {}", root.display()))?); @@ -46,7 +37,7 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul logger.log("info", format!("Archiving {}", root.display())); let mut files = 0u64; - // Entries that could not be read: a mirror restore of this archive would delete them. + let mut unreadable = 0usize; let mut walker = WalkDir::new(&root) .follow_links(false) @@ -65,7 +56,6 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul let rel = entry.path().strip_prefix(&root)?.to_path_buf(); let ft = entry.file_type(); - // Never archive our own output (backup dir under the source, or the archive itself). let mut skip = matcher.is_excluded(&rel) || entry.path() == backup_real || entry.path() == out_real; if !skip && one_file_system && ft.is_dir() { @@ -86,8 +76,6 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul } if ft.is_file() { - // Only a failure to open is skippable: it happens before anything is - // written. Every later error leaves the tar stream misaligned and must abort. let mut f = match File::open(entry.path()) { Ok(f) => f, Err(e) => { @@ -100,20 +88,18 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul let size = meta.len(); let mut header = tar::Header::new_gnu(); header.set_metadata(&meta); - // The header size is fixed up front: cap the copy if the file grew and - // zero-pad if it shrank, so the entry always matches its header. + let reader = (&mut f).take(size).chain(std::io::repeat(0)).take(size); tar.append_data(&mut header, &rel, reader) .with_context(|| format!("failed to archive {}", rel.display()))?; files += 1; } else if ft.is_dir() || ft.is_symlink() { - // These fail on lstat/readlink before the header is written. + if let Err(e) = tar.append_path_with_name(entry.path(), &rel) { logger.log("warn", format!("Skipping {}: {e}", rel.display())); unreadable += 1; } } else { - // Sockets, fifos and devices hold no data to restore; restore keeps them too. logger.log("warn", format!("Skipping special file {}", rel.display())); } } diff --git a/src/domain/files/matcher.rs b/src/domain/files/matcher.rs index 3c3f5c0..88f96d7 100644 --- a/src/domain/files/matcher.rs +++ b/src/domain/files/matcher.rs @@ -1,10 +1,8 @@ use crate::services::config::DatabaseConfig; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, bail}; use globset::{GlobBuilder, GlobSet, GlobSetBuilder}; use std::path::{Path, PathBuf}; -/// Exclude patterns, matched against paths relative to the source root. -/// No leading `/` matches at any depth; a leading `/` anchors at the root. pub struct ExcludeMatcher { set: GlobSet, literal: Vec, @@ -39,8 +37,6 @@ impl ExcludeMatcher { Self::new(&exclude_patterns(cfg)) } - /// Also exclude exactly this root-relative path (and its subtree), compared - /// literally so glob characters in the name mean nothing. pub fn exclude_path(&mut self, relative: PathBuf) { self.literal.push(relative); } @@ -50,7 +46,6 @@ impl ExcludeMatcher { } } -/// The `options.exclude` patterns of a files source, as configured. pub fn exclude_patterns(cfg: &DatabaseConfig) -> Vec { cfg.options .get("exclude") @@ -64,10 +59,40 @@ pub fn exclude_patterns(cfg: &DatabaseConfig) -> Vec { .unwrap_or_default() } -/// `options.one_file_system` of a files source. pub fn one_file_system(cfg: &DatabaseConfig) -> bool { cfg.options .get("one_file_system") .and_then(|v| v.as_bool()) .unwrap_or(false) } + +pub fn source_root(cfg: &DatabaseConfig) -> Result { + let root = std::fs::canonicalize(&cfg.path) + .with_context(|| format!("cannot read source directory {}", cfg.path))?; + if !root.is_dir() { + bail!("source path {} is not a directory", root.display()); + } + Ok(root) +} + +pub(crate) fn bracket_negation(pattern: &str) -> String { + let mut out = String::with_capacity(pattern.len()); + let mut chars = pattern.chars(); + let mut in_class = false; + while let Some(c) = chars.next() { + out.push(c); + match c { + '\\' => out.extend(chars.next()), + '[' if !in_class => { + in_class = true; + if chars.as_str().starts_with('!') { + chars.next(); + out.push('^'); + } + } + ']' => in_class = false, + _ => {} + } + } + out +} diff --git a/src/domain/files/mod.rs b/src/domain/files/mod.rs index d112947..c1fd77b 100644 --- a/src/domain/files/mod.rs +++ b/src/domain/files/mod.rs @@ -3,4 +3,6 @@ pub mod database; pub mod matcher; pub mod ping; pub mod platform; +pub mod restic; pub mod restore; +pub mod sync; diff --git a/src/domain/files/ping.rs b/src/domain/files/ping.rs index ee61cd3..2af3acf 100644 --- a/src/domain/files/ping.rs +++ b/src/domain/files/ping.rs @@ -4,7 +4,6 @@ use std::path::Path; use tokio::time::{Duration, timeout}; pub async fn run(cfg: DatabaseConfig) -> Result { - // A hung mount (NFS) blocks is_dir/read_dir: keep it off the runtime and bounded. let check = tokio::task::spawn_blocking(move || { let root = Path::new(&cfg.path); root.is_dir() && std::fs::read_dir(root).is_ok() diff --git a/src/domain/files/platform.rs b/src/domain/files/platform.rs index eddb031..60fca91 100644 --- a/src/domain/files/platform.rs +++ b/src/domain/files/platform.rs @@ -14,7 +14,6 @@ pub fn dev(_meta: &std::fs::Metadata) -> u64 { 0 } -/// Whether the agent can restore file ownership. #[cfg(unix)] pub fn is_root() -> bool { unsafe { libc::geteuid() == 0 } diff --git a/src/domain/files/restic/backup.rs b/src/domain/files/restic/backup.rs new file mode 100644 index 0000000..58b10e5 --- /dev/null +++ b/src/domain/files/restic/backup.rs @@ -0,0 +1,106 @@ +use super::excludes::backup_excludes; +use super::json::{BackupSummary, Snapshot, files_removed}; +use super::repo::{ResticRepo, short}; +use crate::domain::files::matcher::{exclude_patterns, one_file_system, source_root}; +use crate::services::backup::logger::JobLogger; +use crate::services::config::DatabaseConfig; +use anyhow::{Context, Result, bail}; + +pub async fn snapshot( + repo: &ResticRepo, + cfg: &DatabaseConfig, + backup_storage_id: &str, + logger: &JobLogger, +) -> Result { + let root = source_root(cfg)?; + repo.ensure_initialized(logger).await?; + + let host = format!("portabase-{}", cfg.generated_id); + let mut args: Vec = vec![ + "backup".into(), + root.to_string_lossy().into_owned(), + "--json".into(), + "--host".into(), + host.clone(), + "--tag".into(), + "portabase".into(), + "--tag".into(), + format!("bs:{backup_storage_id}"), + ]; + for pattern in backup_excludes(&root, &exclude_patterns(cfg))? { + args.push("--exclude".into()); + args.push(pattern); + } + if one_file_system(cfg) { + args.push("--one-file-system".into()); + } + + logger.log("info", format!("Snapshotting {} into {}", root.display(), repo.repository())); + let run = repo.run(&args, logger).await?; + for error in &run.errors { + logger.log("warn", format!("Unreadable: {error}")); + } + let summary: Option = run + .summary + .clone() + .map(serde_json::from_value) + .transpose() + .context("unexpected restic backup summary")?; + + match (run.code, summary) { + (0, Some(mut s)) => { + s.files_removed = match removed(repo, &host, &s, logger).await { + Ok(removed) => removed, + Err(e) => { + logger.log("warn", format!("Could not count removed files: {e:#}")); + None + } + }; + let removed = s.files_removed.map(|n| format!(", {n} removed")).unwrap_or_default(); + logger.log( + "info", + format!( + "Snapshot {}: {} new, {} changed, {} unmodified{removed} file(s); {} byte(s) scanned, {} byte(s) of new data", + short(&s.snapshot_id), s.files_new, s.files_changed, s.files_unmodified, s.total_bytes_processed, s.data_added_packed + ), + ); + Ok(s) + } + (0, None) => bail!("restic backup returned no summary"), + (3, summary) => { + let mut fate = "snapshot discarded"; + if let Some(s) = summary { + if repo.is_append_only() { + logger.log( + "warn", + format!("Incomplete snapshot {} left for the dashboard to forget", short(&s.snapshot_id)), + ); + fate = "snapshot not restorable (the dashboard forgets it)"; + } else { + let forget = repo.run(["forget", s.snapshot_id.as_str()], logger).await?; + if forget.code != 0 { + logger.log( + "error", + format!("Could not forget incomplete snapshot {}: {}", s.snapshot_id, forget.error("forget")), + ); + } + } + } + bail!( + "{} file(s) could not be read; {fate} so a mirror restore cannot delete them (fix permissions or add an exclude)", + run.errors.len() + ) + } + _ => Err(run.error("backup")), + } +} + +async fn removed(repo: &ResticRepo, host: &str, s: &BackupSummary, logger: &JobLogger) -> Result> { + let listed = repo.run(["snapshots", "--json", "--host", host], logger).await?; + if listed.code != 0 { + return Err(listed.error("snapshots")); + } + let snapshots: Vec = + serde_json::from_str(listed.stdout.trim()).context("unexpected `restic snapshots` output")?; + Ok(files_removed(&snapshots, &s.snapshot_id, s.files_changed, s.files_unmodified)) +} diff --git a/src/services/restic/excludes.rs b/src/domain/files/restic/excludes.rs similarity index 54% rename from src/services/restic/excludes.rs rename to src/domain/files/restic/excludes.rs index bc2dbdd..b894b84 100644 --- a/src/services/restic/excludes.rs +++ b/src/domain/files/restic/excludes.rs @@ -1,7 +1,7 @@ +use crate::domain::files::matcher::bracket_negation; use anyhow::{Result, bail}; use std::path::Path; -/// Escapes restic/filepath.Match metacharacters so a literal path matches only itself. pub fn glob_escape(literal: &str) -> String { let mut out = String::with_capacity(literal.len()); for c in literal.chars() { @@ -13,31 +13,6 @@ pub fn glob_escape(literal: &str) -> String { out } -/// globset (the P0 matcher) reads `[!x]` as bracket negation; restic (Go -/// `filepath.Match`) and rclone only know `[^x]` and take `!` literally. -pub(crate) fn bracket_negation(pattern: &str) -> String { - let mut out = String::with_capacity(pattern.len()); - let mut chars = pattern.chars(); - let mut in_class = false; - while let Some(c) = chars.next() { - out.push(c); - match c { - '\\' => out.extend(chars.next()), - '[' if !in_class => { - in_class = true; - if chars.as_str().starts_with('!') { - chars.next(); - out.push('^'); - } - } - ']' => in_class = false, - _ => {} - } - } - out -} - -/// P0 patterns, trimmed like the P0 matcher, as `(anchored, pattern)`; a leading `/` marks anchored ones. fn normalized(patterns: &[String]) -> Result> { let mut out = Vec::new(); for raw in patterns { @@ -57,9 +32,6 @@ fn normalized(patterns: &[String]) -> Result> { Ok(out) } -/// `restic backup` matches patterns against absolute paths, so every pattern is -/// rooted at the source; unanchored ones would otherwise also match directories -/// above it and exclude the whole source (spike S5). pub fn backup_excludes(root: &Path, patterns: &[String]) -> Result> { let root = glob_escape(root.to_string_lossy().trim_end_matches('/')); Ok(normalized(patterns)? @@ -70,7 +42,6 @@ pub fn backup_excludes(root: &Path, patterns: &[String]) -> Result> .collect()) } -/// `restic restore :` matches patterns relative to that subfolder (spike S1). pub fn restore_excludes(patterns: &[String]) -> Result> { Ok(normalized(patterns)? .into_iter() diff --git a/src/services/restic/json.rs b/src/domain/files/restic/json.rs similarity index 75% rename from src/services/restic/json.rs rename to src/domain/files/restic/json.rs index e933830..dcd53c0 100644 --- a/src/services/restic/json.rs +++ b/src/domain/files/restic/json.rs @@ -1,8 +1,6 @@ use serde::Deserialize; use serde_json::{Value, json}; -/// One line of `restic --json` output. stdout carries `status` / `summary`, -/// stderr carries `error` / `exit_error` and plain-text fatal messages. #[derive(Debug, PartialEq)] pub enum Line { Status { percent_done: f64 }, @@ -41,14 +39,11 @@ pub struct BackupSummary { pub files_new: u64, pub files_changed: u64, pub files_unmodified: u64, - /// Files of the parent snapshot this one no longer has: restic does not count them - /// (see `files_removed`). None when unknown. #[serde(skip)] pub files_removed: Option, } impl BackupSummary { - /// Counters the dashboard shows for this snapshot (`size` carries `data_added_packed`). pub fn report(&self) -> Value { json!({ "filesNew": self.files_new, @@ -59,14 +54,12 @@ impl BackupSummary { } } -/// One entry of `restic snapshots --json`. #[derive(Debug, Deserialize)] pub struct Snapshot { pub id: String, pub paths: Vec, #[serde(default)] pub parent: Option, - /// Written by restic >= 0.17. #[serde(default)] pub summary: Option, } @@ -76,10 +69,6 @@ pub struct SnapshotSummary { pub total_files_processed: u64, } -/// Files of the parent missing from `snapshot_id`, from a `restic snapshots --json` listing: -/// restic counts a file as changed or unmodified only when the parent has it, so the parent's -/// file count minus those is what was removed. Some(0) without a parent; None when the parent -/// is not listed or has no summary. pub fn files_removed(snapshots: &[Snapshot], snapshot_id: &str, changed: u64, unmodified: u64) -> Option { let find = |id: &str| snapshots.iter().find(|s| s.id == id); let Some(parent) = find(snapshot_id)?.parent.as_deref() else { diff --git a/src/domain/files/restic/mod.rs b/src/domain/files/restic/mod.rs new file mode 100644 index 0000000..12fbd55 --- /dev/null +++ b/src/domain/files/restic/mod.rs @@ -0,0 +1,6 @@ +pub mod backup; +pub mod excludes; +pub mod json; +pub mod password; +pub mod repo; +pub mod restore; diff --git a/src/domain/files/restic/password.rs b/src/domain/files/restic/password.rs new file mode 100644 index 0000000..a491fb5 --- /dev/null +++ b/src/domain/files/restic/password.rs @@ -0,0 +1,7 @@ +use crate::utils::edge_key::derive_key; +use anyhow::Result; + +pub fn derive_password(master_key_b64: &str, generated_id: &str) -> Result { + let info = format!("portabase/restic/v1/{generated_id}"); + Ok(hex::encode(derive_key(master_key_b64, info.as_bytes())?)) +} diff --git a/src/services/restic/command.rs b/src/domain/files/restic/repo.rs similarity index 80% rename from src/services/restic/command.rs rename to src/domain/files/restic/repo.rs index a99ab02..ff09e2f 100644 --- a/src/services/restic/command.rs +++ b/src/domain/files/restic/repo.rs @@ -1,11 +1,13 @@ use super::json::{self, Line}; -use super::password::{derive_password, local_storage_password, LOCAL_STORAGE_USER}; +use super::password::derive_password; use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::backup::logger::JobLogger; +use crate::services::storage::providers::local::{LOCAL_STORAGE_USER, local_storage_password}; use crate::services::storage::providers::rclone::helpers::{remote_target, write_config}; use crate::services::storage::providers::rclone::target::{RcloneTarget, rclone_target}; use crate::settings::CONFIG; use crate::utils::edge_key::EdgeKey; +use crate::utils::file::folder_or_default; use anyhow::{Context, Result, anyhow}; use serde_json::Value; use std::ffi::{OsStr, OsString}; @@ -15,32 +17,20 @@ use tempfile::NamedTempFile; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::Command; -/// One source's repository on one storage channel: -/// `rclone:://restic/`, or -/// `rest:/storage/restic//` on the dashboard's local storage. pub struct ResticRepo { repository: String, password: String, - /// `None` when the cache directory cannot be created: restic then runs `--no-cache`. cache_dir: Option, - /// rclone config of `rclone:` repositories; `None` on the dashboard's REST server. config: Option, - /// HTTP basic auth of `rest:` repositories (the dashboard's local storage). rest_auth: Option<(String, String)>, - /// Keeps temp files referenced by the config (sftp key) alive. _target: Option, } -/// What a finished restic command left behind. pub struct ResticRun { pub code: i32, - /// stdout lines that are neither `status` nor `summary` (e.g. `snapshots --json`). pub stdout: String, - /// The last `summary` line, if any. pub summary: Option, - /// `error` lines as "item: message" (unreadable files and the like). pub errors: Vec, - /// Last plain-text / `exit_error` stderr lines, for failure messages. pub stderr_tail: String, } @@ -55,8 +45,6 @@ impl ResticRun { } } -/// A full snapshot id. Ids from the dashboard reach restic as positional arguments: -/// anything else could be parsed as a flag (e.g. `--password-command=`). pub fn is_snapshot_id(id: &str) -> bool { id.len() == 64 && id.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) } @@ -65,7 +53,6 @@ pub fn short(id: &str) -> &str { id.get(..8).unwrap_or(id) } -/// restic fails hard (not warns) on an unusable cache dir, so probe it up front. fn usable_cache_dir(cache_dir: PathBuf) -> Option { match std::fs::create_dir_all(&cache_dir) { Ok(()) => Some(cache_dir), @@ -90,17 +77,11 @@ impl ResticRepo { )); } let target = rclone_target(storage)?; - let folder = storage - .folder_name - .as_deref() - .map(|f| f.trim().trim_matches('/')) - .filter(|f| !f.is_empty()) - .unwrap_or("backups"); let mut repo = Self::new( &target.config_text, &target.remote_name, &target.base_path, - folder, + folder_or_default(storage.folder_name.as_deref()), generated_id, &password, cache_dir, @@ -129,8 +110,6 @@ impl ResticRepo { }) } - /// The dashboard's local storage (`rclone serve restic --append-only`): - /// `rest://`, basic auth only in the child env. pub fn rest(base_url: &str, generated_id: &str, password: &str, server_password: &str, cache_dir: PathBuf) -> Self { Self { repository: format!("rest:{}/{generated_id}/", base_url.trim_end_matches('/')), @@ -142,7 +121,6 @@ impl ResticRepo { } } - /// The dashboard's server refuses deletions: the agent cannot forget there. pub fn is_append_only(&self) -> bool { self.rest_auth.is_some() } @@ -151,8 +129,6 @@ impl ResticRepo { &self.repository } - /// `restic --retry-lock 5m` (+ `--no-cache` without a cache dir); secrets only in the child env. `status` - /// lines are logged every 10 %. pub async fn run(&self, args: I, logger: &JobLogger) -> Result where I: IntoIterator, @@ -177,7 +153,6 @@ impl ResticRepo { let mut child = cmd .env("RESTIC_REPOSITORY", &self.repository) .env("RESTIC_PASSWORD", &self.password) - // No TTY: without this restic prints no status lines at all. .env("RESTIC_PROGRESS_FPS", "0.2") .stdin(Stdio::null()) .stdout(Stdio::piped()) @@ -237,8 +212,6 @@ impl ResticRepo { Ok(run) } - /// `restic cat config`; exit 10 (no repository) → `restic init`. Callers hold - /// the source's FileLock, so two inits never race. pub async fn ensure_initialized(&self, logger: &JobLogger) -> Result<()> { let cat = self.run(["cat", "config"], logger).await?; match cat.code { diff --git a/src/services/restic/restore.rs b/src/domain/files/restic/restore.rs similarity index 68% rename from src/services/restic/restore.rs rename to src/domain/files/restic/restore.rs index 12c410f..4be279f 100644 --- a/src/services/restic/restore.rs +++ b/src/domain/files/restic/restore.rs @@ -1,20 +1,15 @@ -use super::command::{ResticRepo, is_snapshot_id, short}; use super::excludes::{glob_escape, restore_excludes}; use super::json::Snapshot; +use super::repo::{ResticRepo, is_snapshot_id, short}; use crate::domain::files::matcher::{exclude_patterns, one_file_system}; use crate::domain::files::platform::dev; use crate::domain::files::restore::ensure_restorable; -use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::backup::logger::JobLogger; -use crate::services::config::{DatabaseConfig, DbType}; -use crate::utils::edge_key::EdgeKey; -use crate::utils::locks::{DbOpLock, FileLock}; +use crate::services::config::DatabaseConfig; use anyhow::{Context, Result, bail}; use std::path::Path; use walkdir::WalkDir; -/// Mirror restore of `snapshot_id` into `cfg.path`: files created since are -/// deleted, paths matching the current exclude patterns are kept (spike S1). pub async fn restore( repo: &ResticRepo, cfg: &DatabaseConfig, @@ -32,7 +27,6 @@ pub async fn restore( } let snapshots: Vec = serde_json::from_str(listed.stdout.trim()) .context("unexpected `restic snapshots` output")?; - // The original path inside the snapshot; the mount point may have moved since. let Some(original) = snapshots.first().and_then(|s| s.paths.first()) else { bail!("snapshot {} not found in {}", short(snapshot_id), repo.repository()); }; @@ -78,8 +72,6 @@ pub async fn restore( Ok(()) } -/// Directories under `root` on another device. With `one_file_system` backup -/// never entered them, so the mirror restore must not delete their content. fn mount_points(root: &Path) -> Vec { let Ok(meta) = std::fs::metadata(root) else { return Vec::new() }; let root_dev = dev(&meta); @@ -99,26 +91,3 @@ fn mount_points(root: &Path) -> Vec { } out } - -/// Snapshot restore job: the source's FileLock is held for the whole restore. -pub async fn run( - edge_key: &EdgeKey, - cfg: &DatabaseConfig, - snapshot_id: &str, - storage: &DatabaseStorage, - logger: &JobLogger, -) -> Result<()> { - if !matches!(cfg.db_type, DbType::Files) { - bail!("snapshot restore needs a files source, got {}", cfg.db_type.as_str()); - } - FileLock::acquire(&cfg.generated_id, DbOpLock::Restore.as_str()) - .await - .context("another backup or restore of this source is running")?; - let result = async { - let repo = ResticRepo::open(storage, &cfg.generated_id, edge_key)?; - restore(&repo, cfg, snapshot_id, logger).await - } - .await; - FileLock::release(&cfg.generated_id).await?; - result -} diff --git a/src/domain/files/restore.rs b/src/domain/files/restore.rs index fb07abc..0a24c48 100644 --- a/src/domain/files/restore.rs +++ b/src/domain/files/restore.rs @@ -1,4 +1,4 @@ -use super::matcher::ExcludeMatcher; +use super::matcher::{ExcludeMatcher, one_file_system}; use super::platform::{dev, is_root}; use crate::services::backup::logger::JobLogger; use crate::services::config::DatabaseConfig; @@ -14,25 +14,16 @@ pub async fn run(cfg: DatabaseConfig, archive: PathBuf, logger: Arc) } fn restore(cfg: &DatabaseConfig, archive: &Path, logger: &JobLogger) -> Result<()> { - // Everything below works on the canonical path, like backup.rs. let root = ensure_restorable(Path::new(&cfg.path))?; let mut matcher = ExcludeMatcher::from_config(cfg)?; - let one_file_system = cfg - .options - .get("one_file_system") - .and_then(|v| v.as_bool()) - .unwrap_or(false); + let one_file_system = one_file_system(cfg); - // Opened once: the wipe may delete the archive's path, never an open handle. let file = File::open(archive) .with_context(|| format!("cannot open archive {}", archive.display()))?; - // The wipe is irreversible: prove the whole archive reads back before touching data. logger.log("info", format!("Validating archive {}", archive.display())); validate_archive(&file).context("archive is corrupt or truncated, nothing was changed")?; - // Keep the archive's own directory (or the archive itself when it sits directly - // in the target), like backup.rs prunes its output. let archive_real = fs::canonicalize(archive)?; let keep = match archive_real.parent() { Some(dir) if dir != root => dir, @@ -46,7 +37,7 @@ fn restore(cfg: &DatabaseConfig, archive: &Path, logger: &JobLogger) -> Result<( "info", format!("Removing non-excluded content of {}", root.display()), ); - // Best effort: whatever could not be removed is kept, and extraction always runs. + let failures = wipe_except_excluded(&root, &matcher, one_file_system, logger)?; logger.log( @@ -65,9 +56,6 @@ fn restore(cfg: &DatabaseConfig, archive: &Path, logger: &JobLogger) -> Result<( Ok(()) } -/// Extract the whole archive into `root`. Filesystems that refuse chown/chmod/utime -/// (NFS root_squash, CIFS, NFSv4 idmap mismatch) fail the first pass, after the wipe, -/// so retry once without metadata: the content matters more than owner and mode. pub fn unpack( file: &File, root: &Path, @@ -79,9 +67,6 @@ pub fn unpack( "warn", format!("extraction with ownership/permissions failed: {e:#}; retrying without metadata preservation"), ); - // ponytail: tar-rs still chmods to `mode & 0o777` with preservation off, so a - // filesystem that rejects chmod on files the agent just created fails this pass - // too; per-entry extraction with deferred directories if that shows up. unpack_once(file, root, false, false)?; } Ok(()) @@ -103,7 +88,6 @@ fn unpack_once( Ok(()) } -/// Read every entry to the end, then drain the gzip stream so its CRC is checked. fn validate_archive(file: &File) -> Result<()> { let mut gz = GzDecoder::new(file); let mut ar = tar::Archive::new(&mut gz); @@ -111,14 +95,10 @@ fn validate_archive(file: &File) -> Result<()> { io::copy(&mut entry?, &mut io::sink())?; } drop(ar); - // tar stops at its end marker, before the gzip trailer. gz.read_to_end(&mut Vec::new())?; Ok(()) } -/// Refuse targets a mirror restore must never touch, and return the canonical path to -/// restore into. The canonical path is checked too, because `/data/..` style escapes -/// resolve to `/`. pub fn ensure_restorable(root: &Path) -> Result { if root.as_os_str().is_empty() || !root.is_absolute() || root == Path::new("/") { bail!( @@ -140,18 +120,6 @@ pub fn ensure_restorable(root: &Path) -> Result { Ok(canonical) } -/// Delete everything under `root` that does not match an exclude pattern. -/// Excluded entries (and their subtrees) are kept; `root` itself is kept. -/// -/// Mount points (a directory on a different device than its parent): with -/// `one_file_system` they are kept untouched like an excluded path, matching what -/// backup never archived. Without it their contents are wiped but the mount point -/// itself is never removed (`remove_dir` on it fails with EBUSY) and counts as kept. -/// Special files (sockets, fifos, devices) are kept: backup never archives them. -/// -/// Best effort: an entry that cannot be read or removed is logged, kept (so its -/// parents are kept too) and counted; the returned count is how many failed. -/// Only an unreadable `root` is an error, and then nothing was deleted. pub fn wipe_except_excluded( root: &Path, matcher: &ExcludeMatcher, @@ -183,8 +151,6 @@ impl Wipe<'_> { self.failures += 1; } - /// Returns true when something was kept inside `dir`. `rel` is `dir` relative to - /// the root and `dir_dev` its device. fn dir(&mut self, dir: &Path, rel: &Path, dir_dev: u64) -> bool { let entries = match fs::read_dir(dir) { Ok(entries) => entries, @@ -209,7 +175,7 @@ impl Wipe<'_> { kept = true; continue; } - // symlink_metadata: a symlinked directory is removed as a link, never followed. + let meta = match fs::symlink_metadata(&path) { Ok(meta) => meta, Err(e) => { @@ -233,7 +199,6 @@ impl Wipe<'_> { } else if ft.is_file() || ft.is_symlink() { fs::remove_file(&path) } else { - // Socket, fifo or device: backup never archives them, so keep them. kept = true; continue; }; diff --git a/src/domain/files/sync/backup.rs b/src/domain/files/sync/backup.rs new file mode 100644 index 0000000..fbca007 --- /dev/null +++ b/src/domain/files/sync/backup.rs @@ -0,0 +1,45 @@ +use super::rclone::{replica_target, sync_dir}; +use super::stats::SyncStats; +use crate::services::api::models::agent::status::DatabaseStorage; +use crate::services::backup::logger::JobLogger; +use crate::services::config::DatabaseConfig; +use crate::services::storage::providers::local::{LOCAL_STORAGE_USER, local_storage_password}; +use crate::services::storage::providers::rclone::helpers::{build_rclone_config, obscure_password}; +use crate::services::storage::providers::rclone::target::rclone_target; +use crate::utils::edge_key::EdgeKey; +use crate::utils::file::folder_or_default; +use anyhow::Result; + +pub fn replica_path(storage: &DatabaseStorage, generated_id: &str) -> String { + format!("{}/sync/{generated_id}/current", folder_or_default(storage.folder_name.as_deref())) +} + +const LOCAL_REMOTE: &str = "pblocal"; + +pub fn local_sync_config(edge_key: &EdgeKey) -> Result { + build_rclone_config( + LOCAL_REMOTE, + &[ + ("type", "webdav".into()), + ("url", format!("{}/storage/sync", edge_key.server_url.trim_end_matches('/'))), + ("vendor", "rclone".into()), + ("user", LOCAL_STORAGE_USER.into()), + ("pass", obscure_password(&local_storage_password(&edge_key.master_key_b64)?)?), + ], + ) +} + +pub async fn replicate(edge_key: &EdgeKey, cfg: &DatabaseConfig, storage: &DatabaseStorage, logger: &JobLogger) -> Result { + if storage.provider == "local" { + let dest = format!("{LOCAL_REMOTE}:{}/current", cfg.generated_id); + return sync_dir(&local_sync_config(edge_key)?, &dest, cfg, logger).await; + } + let target = rclone_target(storage)?; + let dest = replica_target( + &target.remote_name, + &target.base_path, + folder_or_default(storage.folder_name.as_deref()), + &cfg.generated_id, + ); + sync_dir(&target.config_text, &dest, cfg, logger).await +} diff --git a/src/services/sync/excludes.rs b/src/domain/files/sync/excludes.rs similarity index 68% rename from src/services/sync/excludes.rs rename to src/domain/files/sync/excludes.rs index fc24e90..28728e8 100644 --- a/src/services/sync/excludes.rs +++ b/src/domain/files/sync/excludes.rs @@ -1,7 +1,5 @@ -use crate::services::restic::excludes::bracket_negation; +use crate::domain::files::matcher::bracket_negation; -/// P0 exclude patterns → `rclone sync --exclude` values. rclone filters are -/// relative to the sync root (spike 2026-10-03: 19/19 patterns give the P0 file set). pub fn sync_excludes(patterns: &[String]) -> Vec { let mut out = Vec::new(); for raw in patterns { @@ -12,8 +10,6 @@ pub fn sync_excludes(patterns: &[String]) -> Vec { }; let mut rest = rest.trim_end_matches('/'); if !anchored { - // Unanchored rclone patterns already match at any depth; a leading - // `**/` would stop them from matching at the root. while let Some(stripped) = rest.strip_prefix("**/") { rest = stripped; } @@ -22,7 +18,6 @@ pub fn sync_excludes(patterns: &[String]) -> Vec { continue; } let rest = bracket_negation(rest); - // Collapse repeated /**/ into a single /**/ let rest = collapse_double_stars(&rest); let full = if anchored { format!("/{rest}") } else { rest }; for variant in zero_dir_variants(&full) { @@ -33,8 +28,6 @@ pub fn sync_excludes(patterns: &[String]) -> Vec { out } -/// Collapse repeated `/**/` sequences into a single `/**/`. -/// E.g., `a/**/**/b` becomes `a/**/b`. fn collapse_double_stars(pattern: &str) -> String { let mut result = pattern.to_string(); while result.contains("/**/**/") { @@ -43,12 +36,9 @@ fn collapse_double_stars(pattern: &str) -> String { result } -/// rclone's `**` between slashes needs at least one directory; globset's also -/// matches none. Emit every combination of each `/**/` kept or collapsed to `/`. fn zero_dir_variants(pattern: &str) -> Vec { let parts: Vec<&str> = pattern.split("/**/").collect(); let gaps = parts.len() - 1; - // ponytail: 2^gaps variants; past 4 gaps only "all kept" and "all collapsed". if gaps > 4 { return vec![pattern.to_string(), pattern.replace("/**/", "/")]; } diff --git a/src/domain/files/sync/mod.rs b/src/domain/files/sync/mod.rs new file mode 100644 index 0000000..5a26cb6 --- /dev/null +++ b/src/domain/files/sync/mod.rs @@ -0,0 +1,4 @@ +pub mod backup; +pub mod excludes; +pub mod rclone; +pub mod stats; diff --git a/src/services/sync/rclone.rs b/src/domain/files/sync/rclone.rs similarity index 80% rename from src/services/sync/rclone.rs rename to src/domain/files/sync/rclone.rs index 6d361a8..f940c7a 100644 --- a/src/services/sync/rclone.rs +++ b/src/domain/files/sync/rclone.rs @@ -1,6 +1,6 @@ use super::excludes::sync_excludes; use super::stats::{SyncStats, parse_log}; -use crate::domain::files::matcher::{exclude_patterns, one_file_system}; +use crate::domain::files::matcher::{exclude_patterns, one_file_system, source_root}; use crate::services::backup::logger::JobLogger; use crate::services::config::DatabaseConfig; use crate::services::storage::providers::rclone::helpers::{remote_target, write_config}; @@ -9,22 +9,12 @@ use std::path::Path; use std::process::Stdio; use tokio::process::Command; -/// `://sync//current` pub fn replica_target(remote_name: &str, base_path: &str, folder: &str, generated_id: &str) -> String { remote_target(remote_name, base_path, &format!("{folder}/sync/{generated_id}/current")) } -/// Mirrors `cfg.path` onto `dest` (`:`); files matching an exclude are removed -/// from the replica too (`--delete-excluded`). rclone skips its deletions when an error -/// happens before the delete phase ("not deleting files as there were IO errors"); -/// an error during the delete phase can leave the replica partially updated. -/// An empty source is refused so an unmounted volume cannot wipe the replica. pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logger: &JobLogger) -> Result { - let root = std::fs::canonicalize(&cfg.path) - .with_context(|| format!("cannot read source directory {}", cfg.path))?; - if !root.is_dir() { - bail!("source path {} is not a directory", root.display()); - } + let root = source_root(cfg)?; if std::fs::read_dir(&root) .with_context(|| format!("cannot read source directory {}", root.display()))? .next() @@ -45,7 +35,6 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge .arg(dest) .args([ "--links", "--delete-excluded", "--use-json-log", "--stats", "1h", "--stats-log-level", "NOTICE", "--retries", "1", - // An unreachable endpoint fails in ~1–2 min instead of blocking the next runs. "--contimeout", "30s", "--low-level-retries", "3", ]); if one_file_system(cfg) { @@ -80,7 +69,6 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge bail!("rclone sync failed ({}): {reason}; the replica may be partially updated", out.status); } let mut stats = log.stats.unwrap_or_default(); - // Measured on the replica itself, so it is what is stored, not inferred from the source. stats.replica_bytes = match remote_size(config.path(), dest).await { Ok(bytes) => Some(bytes), Err(e) => { @@ -96,7 +84,6 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge Ok(stats) } -/// Bytes stored under `target` (`:`), per `rclone size`. async fn remote_size(config: &Path, target: &str) -> Result { let out = Command::new("rclone") .args(["--contimeout", "30s", "--low-level-retries", "3", "--config"]) diff --git a/src/services/sync/stats.rs b/src/domain/files/sync/stats.rs similarity index 80% rename from src/services/sync/stats.rs rename to src/domain/files/sync/stats.rs index 465c8c8..0772176 100644 --- a/src/services/sync/stats.rs +++ b/src/domain/files/sync/stats.rs @@ -1,7 +1,6 @@ use serde::Deserialize; use serde_json::{Value, json}; -/// Final `stats` object of `rclone --use-json-log --stats 1h --stats-log-level NOTICE`. #[derive(Debug, Default, Deserialize, PartialEq)] pub struct SyncStats { #[serde(default)] @@ -14,13 +13,11 @@ pub struct SyncStats { pub errors: u64, #[serde(default, rename = "lastError")] pub last_error: Option, - /// Replica size after a successful run, from `rclone size` on the destination (not an rclone stat). #[serde(skip)] pub replica_bytes: Option, } impl SyncStats { - /// Counters the dashboard shows for this run (`size` carries `bytes`). pub fn report(&self) -> Value { json!({ "filesTransferred": self.transfers, @@ -30,8 +27,6 @@ impl SyncStats { } } -/// What `rclone sync` printed on stderr: its last stats object and its error -/// messages (JSON `level: error|critical|alert|emergency` lines and any plain-text line). #[derive(Debug, Default)] pub struct SyncLog { pub stats: Option, @@ -61,7 +56,6 @@ pub fn parse_log(stderr: &str) -> SyncLog { } } } - // Keep only the last 20 messages if log.errors.len() > 20 { log.errors = log.errors.into_iter().rev().take(20).collect::>(); log.errors.reverse(); diff --git a/src/services/backup/executor.rs b/src/services/backup/executor.rs index c31884b..269bd3c 100644 --- a/src/services/backup/executor.rs +++ b/src/services/backup/executor.rs @@ -1,10 +1,9 @@ +use super::files_engine::{self, FilesEngine}; use super::logger::JobLogger; use super::models::BackupResult; use super::service::BackupService; use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::config::{DatabaseConfig, DbType}; -use crate::services::restic; -use crate::services::sync; use crate::utils::common::BackupMethod; use crate::utils::locks::FileLock; @@ -34,13 +33,9 @@ impl BackupService { let backup = self.create_backup_record(&generated_id, &method).await?; let backup_id = backup.backup.id; - if engine == "restic" || engine == "sync" { + if let Some(files_engine) = FilesEngine::parse(&engine) { if matches!(db_cfg.db_type, DbType::Files) { - let uploads = if engine == "restic" { - restic::backup::run(&self.ctx, &db_cfg, &storages, &backup_id, &logger).await - } else { - sync::backup::run(&self.ctx, &db_cfg, &storages, &backup_id, &logger).await - }; + let uploads = files_engine::run(&self.ctx, &db_cfg, files_engine, &storages, &backup_id, &logger).await; logger.log("info", "Database backup job finished".to_string()); let result = BackupResult { generated_id: generated_id.clone(), @@ -51,8 +46,6 @@ impl BackupService { }; let duration_ms = start.elapsed().as_millis() as f64; let logs = Arc::try_unwrap(logger).unwrap_or_else(|_| JobLogger::new()).into_entries(); - // fileSize = average logical size across storages (send_result): restic reports - // the size each snapshot processed; sync reports none, so it stays null. self.send_result(result, uploads, &backup_id, logs, duration_ms).await?; return Ok(()); } diff --git a/src/services/backup/files_engine.rs b/src/services/backup/files_engine.rs new file mode 100644 index 0000000..e44389b --- /dev/null +++ b/src/services/backup/files_engine.rs @@ -0,0 +1,166 @@ +use super::logger::JobLogger; +use super::models::UploadResult; +use crate::core::context::Context as CoreContext; +use crate::domain::files::restic::backup::snapshot; +use crate::domain::files::restic::repo::ResticRepo; +use crate::domain::files::sync::backup::{replica_path, replicate}; +use crate::services::api::models::agent::status::DatabaseStorage; +use crate::services::config::DatabaseConfig; +use crate::utils::locks::{DbOpLock, FileLock}; +use anyhow::Result; +use serde_json::Value; + +#[derive(Clone, Copy)] +pub enum FilesEngine { + Restic, + Sync, +} + +impl FilesEngine { + pub fn parse(engine: &str) -> Option { + match engine { + "restic" => Some(Self::Restic), + "sync" => Some(Self::Sync), + _ => None, + } + } + + fn as_str(self) -> &'static str { + match self { + Self::Restic => "restic", + Self::Sync => "sync", + } + } + + fn label(self) -> &'static str { + match self { + Self::Restic => "Snapshot", + Self::Sync => "Sync", + } + } +} + +struct Transfer { + remote_path: String, + size: u64, + report: Value, + total_size: Option, +} + +pub async fn run( + ctx: &CoreContext, + cfg: &DatabaseConfig, + engine: FilesEngine, + storages: &[DatabaseStorage], + backup_id: &str, + logger: &JobLogger, +) -> Vec { + if let Err(e) = FileLock::acquire(&cfg.generated_id, DbOpLock::Backup.as_str()).await { + logger.log("error", format!("{} aborted: {e}", engine.label())); + return Vec::new(); + } + let mut results = Vec::with_capacity(storages.len()); + for storage in storages { + results.push(one_storage(ctx, cfg, engine, storage, backup_id, logger).await); + } + if let Err(e) = FileLock::release(&cfg.generated_id).await { + logger.log("warn", format!("Failed to release the backup lock: {e}")); + } + results +} + +async fn transfer( + ctx: &CoreContext, + cfg: &DatabaseConfig, + engine: FilesEngine, + storage: &DatabaseStorage, + backup_storage_id: &str, + logger: &JobLogger, +) -> Result { + match engine { + FilesEngine::Restic => { + let repo = ResticRepo::open(storage, &cfg.generated_id, &ctx.edge_key)?; + let summary = snapshot(&repo, cfg, backup_storage_id, logger).await?; + Ok(Transfer { + size: summary.data_added_packed, + report: summary.report(), + total_size: Some(summary.total_bytes_processed), + remote_path: summary.snapshot_id, + }) + } + FilesEngine::Sync => { + let stats = replicate(&ctx.edge_key, cfg, storage, logger).await?; + Ok(Transfer { + remote_path: replica_path(storage, &cfg.generated_id), + size: stats.bytes, + report: stats.report(), + total_size: stats.replica_bytes, + }) + } + } +} + +pub(crate) async fn one_storage( + ctx: &CoreContext, + cfg: &DatabaseConfig, + engine: FilesEngine, + storage: &DatabaseStorage, + backup_id: &str, + logger: &JobLogger, +) -> UploadResult { + let agent_id = ctx.edge_key.agent_id.clone(); + let failed = |error: String| UploadResult { + storage_id: storage.id.clone(), + success: false, + error: Some(error), + remote_file_path: None, + total_size: None, + }; + + let backup_storage_id = match ctx + .api + .backup_upload_init(agent_id.clone(), cfg.generated_id.clone(), storage.id.clone(), backup_id, Some(engine.as_str())) + .await + { + Ok(Some(response)) => response.backup_storage.id, + Ok(None) => { + logger.log("error", "Upload init returned empty response"); + return failed("backup_upload_init returned empty response".into()); + } + Err(e) => { + logger.log("error", format!("Upload init failed: {e}")); + return failed("backup_upload_init failed".into()); + } + }; + + match transfer(ctx, cfg, engine, storage, &backup_storage_id, logger).await { + Ok(t) => match ctx + .api + .backup_upload_success(agent_id, cfg.generated_id.clone(), backup_storage_id, t.remote_path.clone(), t.size, t.report, backup_id) + .await + { + Ok(_) => UploadResult { + storage_id: storage.id.clone(), + success: true, + error: None, + remote_file_path: Some(t.remote_path), + total_size: t.total_size, + }, + Err(e) => { + logger.log("error", format!("Upload status update failed for {}: {e}", storage.id)); + failed(e.to_string()) + } + }, + Err(e) => { + logger.log("error", format!("{} to storage {} failed: {e:#}", engine.label(), storage.id)); + if let Err(err) = ctx + .api + .backup_upload_status(agent_id, cfg.generated_id.clone(), backup_storage_id, "failed", String::new(), 0u64, backup_id) + .await + { + logger.log("error", format!("Failed-status update failed for {}: {err}", storage.id)); + } + failed(format!("{e:#}")) + } + } +} diff --git a/src/services/backup/mod.rs b/src/services/backup/mod.rs index b761556..aa65590 100644 --- a/src/services/backup/mod.rs +++ b/src/services/backup/mod.rs @@ -2,6 +2,7 @@ pub mod compressor; pub mod logger; pub mod dispatcher; pub mod executor; +pub mod files_engine; pub mod helpers; pub mod models; pub mod result; diff --git a/src/services/dashboard_config.rs b/src/services/dashboard_config.rs index e5d6b91..babc834 100644 --- a/src/services/dashboard_config.rs +++ b/src/services/dashboard_config.rs @@ -20,7 +20,6 @@ pub fn merge(local: &[DatabaseConfig], dashboard: &[DatabaseConfig]) -> Database DatabasesConfig { databases } } -/// `generated_id`s of local (databases.json) sources that no dashboard config replaces. pub fn local_only_ids(local: &[DatabaseConfig], dashboard: &[DatabaseConfig]) -> HashSet { local .iter() diff --git a/src/services/mod.rs b/src/services/mod.rs index f6516ed..e6c391a 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -3,8 +3,6 @@ pub mod backup; pub mod config; pub mod cron; pub mod dashboard_config; -pub mod restic; pub mod restore; pub mod status; pub mod storage; -pub mod sync; diff --git a/src/services/restic/backup.rs b/src/services/restic/backup.rs deleted file mode 100644 index 401fb83..0000000 --- a/src/services/restic/backup.rs +++ /dev/null @@ -1,224 +0,0 @@ -use super::command::{ResticRepo, short}; -use super::excludes::backup_excludes; -use super::json::{BackupSummary, Snapshot, files_removed}; -use crate::core::context::Context as CoreContext; -use crate::domain::files::matcher::{exclude_patterns, one_file_system}; -use crate::services::api::models::agent::status::DatabaseStorage; -use crate::services::backup::logger::JobLogger; -use crate::services::backup::models::UploadResult; -use crate::services::config::DatabaseConfig; -use crate::utils::locks::{DbOpLock, FileLock}; -use anyhow::{Context, Result, bail}; - -/// One snapshot of `cfg.path`, tagged with its `backup_storage` id. -/// Exit 3 (unreadable files) fails the run and forgets the snapshot: a mirror -/// restore of it would delete the files it is missing (P0 rule R21). On an -/// append-only repository the agent cannot forget it, so the dashboard does. -pub async fn snapshot( - repo: &ResticRepo, - cfg: &DatabaseConfig, - backup_storage_id: &str, - logger: &JobLogger, -) -> Result { - let root = std::fs::canonicalize(&cfg.path) - .with_context(|| format!("cannot read source directory {}", cfg.path))?; - if !root.is_dir() { - bail!("source path {} is not a directory", root.display()); - } - repo.ensure_initialized(logger).await?; - - // Stable host: container hostnames change on recreate and would break - // parent-snapshot detection (full rescan every time). - let host = format!("portabase-{}", cfg.generated_id); - let mut args: Vec = vec![ - "backup".into(), - root.to_string_lossy().into_owned(), - "--json".into(), - "--host".into(), - host.clone(), - "--tag".into(), - "portabase".into(), - "--tag".into(), - format!("bs:{backup_storage_id}"), - ]; - for pattern in backup_excludes(&root, &exclude_patterns(cfg))? { - args.push("--exclude".into()); - args.push(pattern); - } - if one_file_system(cfg) { - args.push("--one-file-system".into()); - } - - logger.log("info", format!("Snapshotting {} into {}", root.display(), repo.repository())); - let run = repo.run(&args, logger).await?; - for error in &run.errors { - logger.log("warn", format!("Unreadable: {error}")); - } - let summary: Option = run - .summary - .clone() - .map(serde_json::from_value) - .transpose() - .context("unexpected restic backup summary")?; - - match (run.code, summary) { - (0, Some(mut s)) => { - s.files_removed = match removed(repo, &host, &s, logger).await { - Ok(removed) => removed, - Err(e) => { - logger.log("warn", format!("Could not count removed files: {e:#}")); - None - } - }; - let removed = s.files_removed.map(|n| format!(", {n} removed")).unwrap_or_default(); - logger.log( - "info", - // data_added_packed: new file and directory blobs, compressed and encrypted. The - // repository grows a little more (pack headers, index and snapshot files). - format!( - "Snapshot {}: {} new, {} changed, {} unmodified{removed} file(s); {} byte(s) scanned, {} byte(s) of new data", - short(&s.snapshot_id), s.files_new, s.files_changed, s.files_unmodified, s.total_bytes_processed, s.data_added_packed - ), - ); - Ok(s) - } - (0, None) => bail!("restic backup returned no summary"), - (3, summary) => { - let mut fate = "snapshot discarded"; - if let Some(s) = summary { - if repo.is_append_only() { - logger.log( - "warn", - format!("Incomplete snapshot {} left for the dashboard to forget", short(&s.snapshot_id)), - ); - fate = "snapshot not restorable (the dashboard forgets it)"; - } else { - let forget = repo.run(["forget", s.snapshot_id.as_str()], logger).await?; - if forget.code != 0 { - logger.log( - "error", - format!("Could not forget incomplete snapshot {}: {}", s.snapshot_id, forget.error("forget")), - ); - } - } - } - bail!( - "{} file(s) could not be read; {fate} so a mirror restore cannot delete them (fix permissions or add an exclude)", - run.errors.len() - ) - } - _ => Err(run.error("backup")), - } -} - -/// Removed files of snapshot `s`, against the parent restic actually used (one listing of -/// the source's snapshots, no index load). -async fn removed(repo: &ResticRepo, host: &str, s: &BackupSummary, logger: &JobLogger) -> Result> { - let listed = repo.run(["snapshots", "--json", "--host", host], logger).await?; - if listed.code != 0 { - return Err(listed.error("snapshots")); - } - let snapshots: Vec = - serde_json::from_str(listed.stdout.trim()).context("unexpected `restic snapshots` output")?; - Ok(files_removed(&snapshots, &s.snapshot_id, s.files_changed, s.files_unmodified)) -} - -/// Snapshots every storage in turn (one scan each, v1). Never errors: each -/// storage reports its own `backup_storage` row, like the archive uploader. -pub async fn run( - ctx: &CoreContext, - cfg: &DatabaseConfig, - storages: &[DatabaseStorage], - backup_id: &str, - logger: &JobLogger, -) -> Vec { - if let Err(e) = FileLock::acquire(&cfg.generated_id, DbOpLock::Backup.as_str()).await { - logger.log("error", format!("Snapshot aborted: {e}")); - return Vec::new(); - } - let mut results = Vec::with_capacity(storages.len()); - for storage in storages { - results.push(one_storage(ctx, cfg, storage, backup_id, logger).await); - } - if let Err(e) = FileLock::release(&cfg.generated_id).await { - logger.log("warn", format!("Failed to release the backup lock: {e}")); - } - results -} - -pub(crate) async fn one_storage( - ctx: &CoreContext, - cfg: &DatabaseConfig, - storage: &DatabaseStorage, - backup_id: &str, - logger: &JobLogger, -) -> UploadResult { - let agent_id = ctx.edge_key.agent_id.clone(); - let failed = |error: String| UploadResult { - storage_id: storage.id.clone(), - success: false, - error: Some(error), - remote_file_path: None, - total_size: None, - }; - - let backup_storage_id = match ctx - .api - .backup_upload_init(agent_id.clone(), cfg.generated_id.clone(), storage.id.clone(), backup_id, Some("restic")) - .await - { - Ok(Some(response)) => response.backup_storage.id, - Ok(None) => { - logger.log("error", "Upload init returned empty response"); - return failed("backup_upload_init returned empty response".into()); - } - Err(e) => { - logger.log("error", format!("Upload init failed: {e}")); - return failed("backup_upload_init failed".into()); - } - }; - - let outcome = match ResticRepo::open(storage, &cfg.generated_id, &ctx.edge_key) { - Ok(repo) => snapshot(&repo, cfg, &backup_storage_id, logger).await, - Err(e) => Err(e), - }; - - match outcome { - Ok(summary) => match ctx - .api - .backup_upload_success( - agent_id, - cfg.generated_id.clone(), - backup_storage_id, - summary.snapshot_id.clone(), - summary.data_added_packed, - summary.report(), - backup_id, - ) - .await - { - Ok(_) => UploadResult { - storage_id: storage.id.clone(), - success: true, - error: None, - remote_file_path: Some(summary.snapshot_id), - total_size: Some(summary.total_bytes_processed), - }, - Err(e) => { - logger.log("error", format!("Upload status update failed for {}: {e}", storage.id)); - failed(e.to_string()) - } - }, - Err(e) => { - logger.log("error", format!("Snapshot to storage {} failed: {e:#}", storage.id)); - if let Err(err) = ctx - .api - .backup_upload_status(agent_id, cfg.generated_id.clone(), backup_storage_id, "failed", String::new(), 0u64, backup_id) - .await - { - logger.log("error", format!("Failed-status update failed for {}: {err}", storage.id)); - } - failed(format!("{e:#}")) - } - } -} diff --git a/src/services/restic/mod.rs b/src/services/restic/mod.rs deleted file mode 100644 index 8d112b5..0000000 --- a/src/services/restic/mod.rs +++ /dev/null @@ -1,7 +0,0 @@ -//! Snapshots mode: restic repositories reached through the P1 rclone mapper. -pub mod backup; -pub mod command; -pub mod excludes; -pub mod json; -pub mod password; -pub mod restore; diff --git a/src/services/restic/password.rs b/src/services/restic/password.rs deleted file mode 100644 index ca08919..0000000 --- a/src/services/restic/password.rs +++ /dev/null @@ -1,42 +0,0 @@ -use anyhow::{Context, Result}; -use base64::{Engine as _, engine::general_purpose}; -use hmac::{Hmac, Mac}; -use sha2::Sha256; - -type HmacSha256 = Hmac; - -/// Repository password: hex(HKDF-SHA256(master key, salt = "", -/// info = "portabase/restic/v1/", 32 bytes)). The dashboard derives -/// the same value (`src/lib/restic/repo.ts`), so it is never stored nor sent. -pub fn derive_password(master_key_b64: &str, generated_id: &str) -> Result { - let master_key = general_purpose::STANDARD - .decode(master_key_b64.trim()) - .context("master key is not valid base64")?; - let info = format!("portabase/restic/v1/{generated_id}"); - Ok(hex::encode(hkdf_sha256_32(&master_key, info.as_bytes()))) -} - -/// User of the dashboard's local-storage servers (`/storage/restic`, `/storage/sync`). -pub const LOCAL_STORAGE_USER: &str = "portabase"; - -/// Their password: hex(HKDF-SHA256(master key, salt = "", info = "portabase/local-storage/v1", -/// 32 bytes)). The dashboard derives the same value (`src/lib/local-storage/credential.ts`). -pub fn local_storage_password(master_key_b64: &str) -> Result { - let master_key = general_purpose::STANDARD - .decode(master_key_b64.trim()) - .context("master key is not valid base64")?; - Ok(hex::encode(hkdf_sha256_32(&master_key, b"portabase/local-storage/v1"))) -} - -/// RFC 5869 with an empty salt and a single expand block (L = HashLen = 32). -pub fn hkdf_sha256_32(ikm: &[u8], info: &[u8]) -> [u8; 32] { - // An empty HMAC key is zero-padded: the RFC's default salt of HashLen zeros. - let mut extract = HmacSha256::new_from_slice(&[]).expect("HMAC takes any key length"); - extract.update(ikm); - let prk = extract.finalize().into_bytes(); - - let mut expand = HmacSha256::new_from_slice(&prk).expect("HMAC takes any key length"); - expand.update(info); - expand.update(&[1]); - expand.finalize().into_bytes().into() -} diff --git a/src/services/restore/executor.rs b/src/services/restore/executor.rs index a38597b..d7575cc 100644 --- a/src/services/restore/executor.rs +++ b/src/services/restore/executor.rs @@ -2,9 +2,11 @@ use super::models::RestoreResult; use super::service::RestoreService; use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::backup::logger::JobLogger; -use crate::services::config::DatabaseConfig; -use crate::services::restic; -use anyhow::Result; +use crate::domain::files::restic::repo::ResticRepo; +use crate::domain::files::restic::restore::restore; +use crate::services::config::{DatabaseConfig, DbType}; +use crate::utils::locks::{DbOpLock, FileLock}; +use anyhow::{Context, Result, bail}; use std::sync::Arc; use std::time::Instant; use tempfile::TempDir; @@ -58,9 +60,7 @@ impl RestoreService { logger.log("info", "Snapshot restore job started".to_string()); let outcome = match (snapshot_id, storage) { - (Some(id), Some(storage)) => { - restic::restore::run(&self.ctx.edge_key, &cfg, &id, &storage, &logger).await - } + (Some(id), Some(storage)) => self.restic_restore(&cfg, &id, &storage, &logger).await, _ => Err(anyhow::anyhow!( "incomplete snapshot restore payload (snapshotId or storage missing)" )), @@ -82,4 +82,26 @@ impl RestoreService { self.send_result(result, logs, duration_ms).await?; Ok(()) } + + async fn restic_restore( + &self, + cfg: &DatabaseConfig, + snapshot_id: &str, + storage: &DatabaseStorage, + logger: &JobLogger, + ) -> Result<()> { + if !matches!(cfg.db_type, DbType::Files) { + bail!("snapshot restore needs a files source, got {}", cfg.db_type.as_str()); + } + FileLock::acquire(&cfg.generated_id, DbOpLock::Restore.as_str()) + .await + .context("another backup or restore of this source is running")?; + let result = async { + let repo = ResticRepo::open(storage, &cfg.generated_id, &self.ctx.edge_key)?; + restore(&repo, cfg, snapshot_id, logger).await + } + .await; + FileLock::release(&cfg.generated_id).await?; + result + } } diff --git a/src/services/status.rs b/src/services/status.rs index 5e9ab9f..ed85daa 100644 --- a/src/services/status.rs +++ b/src/services/status.rs @@ -36,7 +36,6 @@ pub fn resolve_dashboard_config( Ok(()) } -/// A storage channel sent encrypted as a one-element JSON array. fn decrypt_storage(ciphertext: &str, master_key_b64: &str, what: &str) -> Result, String> { let plaintext = decrypt_json_gcm(ciphertext, master_key_b64) .map_err(|e| format!("Failed to decrypt {what} storage: {e}"))?; @@ -45,7 +44,6 @@ fn decrypt_storage(ciphertext: &str, master_key_b64: &str, what: &str) -> Result Ok(storages.into_iter().next()) } -/// Decrypts the storage channel of a snapshot restore (`restore.storageCiphertext`). pub fn resolve_restore_storage(status: &mut DatabaseStatus, master_key_b64: &str) -> Result<(), String> { if let Some(ciphertext) = status.data.restore.storage_ciphertext.clone() { status.data.restore.storage = decrypt_storage(&ciphertext, master_key_b64, "restore")?; @@ -53,8 +51,6 @@ pub fn resolve_restore_storage(status: &mut DatabaseStatus, master_key_b64: &str Ok(()) } -/// `method` reported in the ping for a files source declared in databases.json; -/// `None` for dashboard-managed sources (the dashboard owns their method) and other dbms. pub fn payload_method(db: &DatabaseConfig, local_ids: &HashSet) -> Option<&'static str> { (matches!(db.db_type, DbType::Files) && local_ids.contains(&db.generated_id)).then(|| files_method(db)) } diff --git a/src/services/storage/providers/local.rs b/src/services/storage/providers/local.rs index b93920a..2bd923a 100644 --- a/src/services/storage/providers/local.rs +++ b/src/services/storage/providers/local.rs @@ -3,15 +3,23 @@ use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::backup::models::{BackupResult, UploadResult}; use crate::services::storage::StorageProvider; use crate::utils::common::BackupMethod; +use crate::utils::edge_key::derive_key; use crate::utils::file::{full_file_name, full_file_path}; use crate::utils::stream::build_stream; use crate::utils::tus::upload_to_tus_stream_with_headers; +use anyhow::Result; use async_trait::async_trait; use reqwest::header::{HeaderMap, HeaderValue}; use std::sync::Arc; use tokio::fs; use tracing::error; +pub const LOCAL_STORAGE_USER: &str = "portabase"; + +pub fn local_storage_password(master_key_b64: &str) -> Result { + Ok(hex::encode(derive_key(master_key_b64, b"portabase/local-storage/v1")?)) +} + pub struct LocalProvider; #[async_trait] diff --git a/src/services/sync/backup.rs b/src/services/sync/backup.rs deleted file mode 100644 index 33fbd3b..0000000 --- a/src/services/sync/backup.rs +++ /dev/null @@ -1,147 +0,0 @@ -//! Sync as a backup method: one `rclone sync` per storage channel, each reported -//! as its own `backup_storage` row (`engine = sync`), like restic snapshots. -use super::rclone::{replica_target, sync_dir}; -use super::stats::SyncStats; -use crate::core::context::Context as CoreContext; -use crate::services::api::models::agent::status::DatabaseStorage; -use crate::services::backup::logger::JobLogger; -use crate::services::backup::models::UploadResult; -use crate::services::config::DatabaseConfig; -use crate::services::restic::password::{LOCAL_STORAGE_USER, local_storage_password}; -use crate::services::storage::providers::rclone::helpers::{build_rclone_config, obscure_password}; -use crate::services::storage::providers::rclone::target::rclone_target; -use crate::utils::edge_key::EdgeKey; -use crate::utils::locks::{DbOpLock, FileLock}; -use anyhow::Result; - -/// Syncs every storage in turn, under the source's backup lock. Never errors: -/// each storage reports its own `backup_storage` row. -pub async fn run( - ctx: &CoreContext, - cfg: &DatabaseConfig, - storages: &[DatabaseStorage], - backup_id: &str, - logger: &JobLogger, -) -> Vec { - if let Err(e) = FileLock::acquire(&cfg.generated_id, DbOpLock::Backup.as_str()).await { - logger.log("error", format!("Sync aborted: {e}")); - return Vec::new(); - } - let mut results = Vec::with_capacity(storages.len()); - for storage in storages { - results.push(one_storage(ctx, cfg, storage, backup_id, logger).await); - } - if let Err(e) = FileLock::release(&cfg.generated_id).await { - logger.log("warn", format!("Failed to release the backup lock: {e}")); - } - results -} - -fn folder(storage: &DatabaseStorage) -> &str { - storage - .folder_name - .as_deref() - .map(|f| f.trim().trim_matches('/')) - .filter(|f| !f.is_empty()) - .unwrap_or("backups") -} - -/// `/sync//current`, relative to the channel's base path. -pub fn replica_path(storage: &DatabaseStorage, generated_id: &str) -> String { - format!("{}/sync/{generated_id}/current", folder(storage)) -} - -const LOCAL_REMOTE: &str = "pblocal"; - -/// rclone config of the dashboard's local storage (`rclone serve webdav` at `/storage/sync`). -pub fn local_sync_config(edge_key: &EdgeKey) -> Result { - build_rclone_config( - LOCAL_REMOTE, - &[ - ("type", "webdav".into()), - ("url", format!("{}/storage/sync", edge_key.server_url.trim_end_matches('/'))), - // rclone's own WebDAV extensions keep modification times: unchanged files are skipped. - ("vendor", "rclone".into()), - ("user", LOCAL_STORAGE_USER.into()), - ("pass", obscure_password(&local_storage_password(&edge_key.master_key_b64)?)?), - ], - ) -} - -async fn replicate(ctx: &CoreContext, cfg: &DatabaseConfig, storage: &DatabaseStorage, logger: &JobLogger) -> Result { - if storage.provider == "local" { - let dest = format!("{LOCAL_REMOTE}:{}/current", cfg.generated_id); - return sync_dir(&local_sync_config(&ctx.edge_key)?, &dest, cfg, logger).await; - } - let target = rclone_target(storage)?; - let dest = replica_target(&target.remote_name, &target.base_path, folder(storage), &cfg.generated_id); - sync_dir(&target.config_text, &dest, cfg, logger).await -} - -pub(crate) async fn one_storage( - ctx: &CoreContext, - cfg: &DatabaseConfig, - storage: &DatabaseStorage, - backup_id: &str, - logger: &JobLogger, -) -> UploadResult { - let agent_id = ctx.edge_key.agent_id.clone(); - let failed = |error: String| UploadResult { - storage_id: storage.id.clone(), - success: false, - error: Some(error), - remote_file_path: None, - total_size: None, - }; - - let backup_storage_id = match ctx - .api - .backup_upload_init(agent_id.clone(), cfg.generated_id.clone(), storage.id.clone(), backup_id, Some("sync")) - .await - { - Ok(Some(response)) => response.backup_storage.id, - Ok(None) => { - logger.log("error", "Upload init returned empty response"); - return failed("backup_upload_init returned empty response".into()); - } - Err(e) => { - logger.log("error", format!("Upload init failed: {e}")); - return failed("backup_upload_init failed".into()); - } - }; - - match replicate(ctx, cfg, storage, logger).await { - Ok(stats) => { - let path = replica_path(storage, &cfg.generated_id); - match ctx - .api - .backup_upload_success(agent_id, cfg.generated_id.clone(), backup_storage_id, path.clone(), stats.bytes, stats.report(), backup_id) - .await - { - Ok(_) => UploadResult { - storage_id: storage.id.clone(), - success: true, - error: None, - remote_file_path: Some(path), - // backups.file_size: the replica's size, like an archive's file size. - total_size: stats.replica_bytes, - }, - Err(e) => { - logger.log("error", format!("Upload status update failed for {}: {e}", storage.id)); - failed(e.to_string()) - } - } - } - Err(e) => { - logger.log("error", format!("Sync to storage {} failed: {e:#}", storage.id)); - if let Err(err) = ctx - .api - .backup_upload_status(agent_id, cfg.generated_id.clone(), backup_storage_id, "failed", String::new(), 0u64, backup_id) - .await - { - logger.log("error", format!("Failed-status update failed for {}: {err}", storage.id)); - } - failed(format!("{e:#}")) - } - } -} diff --git a/src/services/sync/mod.rs b/src/services/sync/mod.rs deleted file mode 100644 index b8501ae..0000000 --- a/src/services/sync/mod.rs +++ /dev/null @@ -1,6 +0,0 @@ -//! Sync mode: a destination kept identical to a files source (`rclone sync`). -//! Replication, not backup: deletions propagate. -pub mod backup; -pub mod excludes; -pub mod stats; -pub mod rclone; diff --git a/src/tests/domain/files.rs b/src/tests/domain/files/mod.rs similarity index 99% rename from src/tests/domain/files.rs rename to src/tests/domain/files/mod.rs index d4695ea..4e73066 100644 --- a/src/tests/domain/files.rs +++ b/src/tests/domain/files/mod.rs @@ -1,3 +1,6 @@ +mod restic; +mod sync; + use crate::domain::files::matcher::ExcludeMatcher; use std::path::Path; diff --git a/src/tests/services/restic_tests.rs b/src/tests/domain/files/restic.rs similarity index 87% rename from src/tests/services/restic_tests.rs rename to src/tests/domain/files/restic.rs index d9046a3..accd825 100644 --- a/src/tests/services/restic_tests.rs +++ b/src/tests/domain/files/restic.rs @@ -1,13 +1,11 @@ -//! Snapshots mode (restic). Pure helpers first; the integration tests further -//! down need `restic` and `rclone` on PATH. - use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::backup::logger::JobLogger; -use crate::services::restic::command::is_snapshot_id; -use crate::services::restic::excludes::{backup_excludes, glob_escape, restore_excludes}; -use crate::services::restic::json::{self, BackupSummary, Line}; -use crate::services::restic::password::{derive_password, hkdf_sha256_32}; +use crate::domain::files::restic::repo::is_snapshot_id; +use crate::domain::files::restic::excludes::{backup_excludes, glob_escape, restore_excludes}; +use crate::domain::files::restic::json::{self, BackupSummary, Line}; +use crate::domain::files::restic::password::derive_password; use crate::tests::domain::files::files_config; +use crate::utils::edge_key::hkdf_sha256_32; use serde_json::json; use std::path::Path; @@ -34,7 +32,6 @@ fn hkdf_matches_rfc5869_case_3_first_block() { #[test] fn password_matches_the_dashboard_vector() { - // Same vector as src/lib/restic/repo.ts in the dashboard (spike S4). let password = derive_password( "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=", "test-generated-id", @@ -83,7 +80,6 @@ fn restore_excludes_are_relative_to_the_snapshot_subfolder() { #[test] fn bracket_negation_is_rewritten_for_restic() { - // globset reads `[!a]`; restic (Go filepath.Match) only knows `[^a]`. assert_eq!( backup_excludes(Path::new("/data"), &strings(&["[!a]*.log", "/x/[!0-9]", r"\[!a]", "[a[!]"])).unwrap(), strings(&["/data/**/[^a]*.log", "/data/x/[^0-9]", r"/data/**/\[!a]", "/data/**/[a[!]"]) @@ -106,7 +102,6 @@ fn glob_escape_escapes_every_metacharacter() { #[test] fn json_lines_are_classified() { - // Recorded from restic 0.19.1 (paths shortened). let status = r#"{"message_type":"status","percent_done":0.9999997777778271,"total_files":4,"files_done":2,"total_bytes":9000002,"bytes_done":9000000,"error_count":1}"#; let error = r#"{"message_type":"error","error":{"message":"open /src/u: permission denied"},"during":"archival","item":"/src/u"}"#; let exit = r#"{"message_type":"exit_error","code":3,"message":"Warning: at least one source file could not be read"}"#; @@ -137,7 +132,7 @@ fn backup_summary_deserializes_from_a_recorded_line() { #[test] fn removed_files_come_from_the_parent_summary() { - use crate::services::restic::json::{Snapshot, files_removed}; + use crate::domain::files::restic::json::{Snapshot, files_removed}; let listing: Vec = serde_json::from_value(serde_json::json!([ {"id": "p", "paths": ["/src"], "summary": {"total_files_processed": 3, "files_new": 3}}, {"id": "c", "paths": ["/src"], "parent": "p"}, @@ -168,27 +163,13 @@ fn a_local_channel_opens_the_dashboard_rest_repository() { assert!(repo.is_append_only()); } -#[test] -fn local_storage_password_matches_the_dashboard_vector() { - use crate::services::restic::password::local_storage_password; - // Same vector as src/lib/local-storage/credential.ts in the dashboard. - assert_eq!( - local_storage_password("AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=").unwrap(), - "8abfb788701b8c502658154746af6af768ea08f579381ad73b0f1e7dd43caea2" - ); -} - -// ---------- integration: restic + rclone on PATH ---------- - use crate::domain::files::platform::is_root; -use crate::services::restic::backup::snapshot; -use crate::services::restic::command::ResticRepo; -use crate::services::restic::restore::restore; +use crate::domain::files::restic::backup::snapshot; +use crate::domain::files::restic::repo::ResticRepo; +use crate::domain::files::restic::restore::restore; use std::fs; use tempfile::TempDir; -/// A repository on an rclone `alias` remote pointing at a temp dir (test-only: -/// production channels always go through `rclone_target`). fn local_repo(dir: &Path, generated_id: &str) -> ResticRepo { fs::create_dir_all(dir).unwrap(); let config = format!("[pb]\ntype = alias\nremote = {}\n", dir.display()); @@ -204,7 +185,6 @@ fn tree(root: &Path) { fs::write(root.join("app.log"), "log").unwrap(); } -/// `relative/path=content` for every regular file, sorted. fn files(root: &Path) -> Vec { let mut out: Vec = walkdir::WalkDir::new(root) .min_depth(1) @@ -242,8 +222,6 @@ async fn first_snapshot_initializes_the_repository_and_the_second_reuses_its_par assert_eq!(second.files_removed, Some(0)); let tagged = repo.run(["snapshots", "--json", "--tag", "bs:bs-2"], &logger).await.unwrap(); - // The tag selects only the second snapshot, and its `parent` is the first - // (a plain `contains(first)` check would match that `parent` field). let listed: serde_json::Value = serde_json::from_str(tagged.stdout.trim()).unwrap(); let listed = listed.as_array().unwrap(); assert_eq!(listed.len(), 1, "{}", tagged.stdout); @@ -291,7 +269,7 @@ async fn restore_is_a_mirror_that_keeps_excluded_paths() { #[tokio::test] async fn unreadable_file_fails_the_snapshot_and_forgets_it() { if is_root() { - return; // root reads a 000 file anyway + return; } use std::os::unix::fs::PermissionsExt; let base = TempDir::new().unwrap(); @@ -313,7 +291,6 @@ async fn unreadable_file_fails_the_snapshot_and_forgets_it() { #[tokio::test] async fn unanchored_exclude_never_matches_directories_above_the_source() { - // Regression for spike S5: the source lives under a directory named like the pattern. let base = TempDir::new().unwrap(); let src = base.path().join("tmp/src"); fs::create_dir_all(src.join("tmp")).unwrap(); @@ -367,7 +344,7 @@ async fn an_unusable_cache_dir_runs_restic_without_cache() { let cfg = files_config(src.path(), &[]); let dir = base.path().join("repo"); fs::create_dir_all(&dir).unwrap(); - let blocker = tempfile::NamedTempFile::new().unwrap(); // a regular file: nothing can be created below it + let blocker = tempfile::NamedTempFile::new().unwrap(); let config = format!("[pb]\ntype = alias\nremote = {}\n", dir.display()); let repo = ResticRepo::new(&config, "pb", "", "backups", &cfg.generated_id, "test-password", blocker.path().join("cache")).unwrap(); snapshot(&repo, &cfg, "bs-1", &JobLogger::new()).await.unwrap(); diff --git a/src/tests/services/sync_tests.rs b/src/tests/domain/files/sync.rs similarity index 93% rename from src/tests/services/sync_tests.rs rename to src/tests/domain/files/sync.rs index 6c6a2e6..7197a26 100644 --- a/src/tests/services/sync_tests.rs +++ b/src/tests/domain/files/sync.rs @@ -1,7 +1,5 @@ -//! Sync mode (rclone replica). Pure helpers first; integration tests need `rclone` on PATH. - -use crate::services::sync::excludes::sync_excludes; -use crate::services::sync::stats::{SyncStats, parse_log}; +use crate::domain::files::sync::excludes::sync_excludes; +use crate::domain::files::sync::stats::{SyncStats, parse_log}; fn strings(items: &[&str]) -> Vec { items.iter().map(|s| s.to_string()).collect() @@ -109,11 +107,9 @@ fn sync_excludes_collapses_repeated_double_stars() { ); } -// ---------- integration: rclone on PATH, `alias` remote under a temp dir ---------- - use crate::domain::files::platform::is_root; use crate::services::backup::logger::JobLogger; -use crate::services::sync::rclone::{replica_target, sync_dir}; +use crate::domain::files::sync::rclone::{replica_target, sync_dir}; use crate::tests::domain::files::files_config; use std::fs; use std::path::{Path, PathBuf}; @@ -128,7 +124,6 @@ fn replica(store: &Path, generated_id: &str) -> PathBuf { store.join("backups/sync").join(generated_id).join("current") } -/// `relative/path=content` for every regular file, sorted. fn files(root: &Path) -> Vec { let mut out: Vec = walkdir::WalkDir::new(root) .min_depth(1) @@ -202,7 +197,7 @@ async fn a_later_exclude_removes_files_from_the_replica() { #[tokio::test] async fn failed_sync_keeps_files_the_failed_run_would_delete() { if is_root() { - return; // root reads a 000 file anyway + return; } use std::os::unix::fs::PermissionsExt; let store = TempDir::new().unwrap(); @@ -214,7 +209,6 @@ async fn failed_sync_keeps_files_the_failed_run_would_delete() { fs::remove_file(src.path().join("docs/b.txt")).unwrap(); let locked = src.path().join("docs/a.txt"); - // rclone only opens a file whose size/mtime changed, so the unreadable file must differ from the replica's copy. fs::write(&locked, "changed and now unreadable").unwrap(); fs::set_permissions(&locked, fs::Permissions::from_mode(0o000)).unwrap(); let err = sync_dir(&config, &replica_target("pb", "", "backups", &cfg.generated_id), &cfg, &JobLogger::new()).await.unwrap_err(); @@ -234,10 +228,8 @@ async fn empty_source_is_refused_and_the_replica_is_untouched() { assert!(!store.path().join("backups").exists()); } -// ---------- sync as a backup method ---------- - use crate::services::api::models::agent::status::DatabaseStorage; -use crate::services::sync::backup::replica_path; +use crate::domain::files::sync::backup::replica_path; use serde_json::json; #[test] @@ -252,7 +244,7 @@ fn replica_path_uses_the_channel_folder() { #[test] fn local_sync_config_points_at_the_dashboard_webdav_server() { - use crate::services::sync::backup::local_sync_config; + use crate::domain::files::sync::backup::local_sync_config; use crate::utils::edge_key::EdgeKey; let edge_key = EdgeKey { server_url: "http://dash:8887/".into(), diff --git a/src/tests/services/api_models_tests.rs b/src/tests/services/api_models_tests.rs index f09c24e..5c62f5d 100644 --- a/src/tests/services/api_models_tests.rs +++ b/src/tests/services/api_models_tests.rs @@ -329,7 +329,7 @@ fn ping_payload_sends_method_only_when_set() { #[test] fn upload_status_sends_stats_only_when_set() { use crate::services::api::endpoints::agent::backup::upload::status::StatusUploadRequest; - use crate::services::sync::stats::SyncStats; + use crate::domain::files::sync::stats::SyncStats; let request = |stats: Option| StatusUploadRequest { generated_id: "g".into(), backup_storage_id: "bs".into(), diff --git a/src/tests/services/local_storage_tests.rs b/src/tests/services/local_storage_tests.rs index 1c46d71..e44a4f9 100644 --- a/src/tests/services/local_storage_tests.rs +++ b/src/tests/services/local_storage_tests.rs @@ -4,9 +4,10 @@ use crate::domain::files::platform::is_root; use crate::services::backup::logger::JobLogger; -use crate::services::restic::backup::snapshot; -use crate::services::restic::command::ResticRepo; -use crate::services::restic::restore::restore; +use crate::domain::files::restic::backup::snapshot; +use crate::domain::files::restic::repo::ResticRepo; +use crate::domain::files::restic::restore::restore; +use crate::services::storage::providers::local::local_storage_password; use crate::tests::domain::files::files_config; use std::fs; use std::path::Path; @@ -134,8 +135,7 @@ async fn local_sync_mirrors_through_the_webdav_server() { use crate::core::context::Context; use crate::services::api::ApiClient; use crate::services::api::models::agent::status::DatabaseStorage; - use crate::services::restic::password::local_storage_password; - use crate::services::sync::backup::one_storage; + use crate::services::backup::files_engine::{FilesEngine, one_storage}; use crate::utils::edge_key::EdgeKey; use serde_json::json; use wiremock::matchers::{body_partial_json, method, path}; @@ -177,14 +177,22 @@ async fn local_sync_mirrors_through_the_webdav_server() { let cfg = files_config(&src, &[]); let replica = store.path().join(&cfg.generated_id).join("current"); - let first = one_storage(&ctx, &cfg, &storage, "backup-1", &JobLogger::new()).await; + let first = one_storage(&ctx, &cfg, FilesEngine::Sync, &storage, "backup-1", &JobLogger::new()).await; assert!(first.success, "{:?}", first.error); assert_eq!(first.remote_file_path.as_deref(), Some(format!("backups/sync/{}/current", cfg.generated_id).as_str())); assert_eq!(fs::read_to_string(replica.join("docs/b.txt")).unwrap(), "b"); fs::remove_file(src.join("docs/b.txt")).unwrap(); - let second = one_storage(&ctx, &cfg, &storage, "backup-2", &JobLogger::new()).await; + let second = one_storage(&ctx, &cfg, FilesEngine::Sync, &storage, "backup-2", &JobLogger::new()).await; assert!(second.success, "{:?}", second.error); assert!(!replica.join("docs/b.txt").exists(), "deletion propagated"); assert!(replica.join("a.txt").exists()); } + +#[test] +fn local_storage_password_matches_the_dashboard_vector() { + assert_eq!( + local_storage_password(MASTER_KEY_B64).unwrap(), + "8abfb788701b8c502658154746af6af768ea08f579381ad73b0f1e7dd43caea2" + ); +} diff --git a/src/tests/services/mod.rs b/src/tests/services/mod.rs index e0e7408..158c1a2 100644 --- a/src/tests/services/mod.rs +++ b/src/tests/services/mod.rs @@ -5,6 +5,4 @@ mod config_tests; mod cron_tests; mod dashboard_config_tests; pub(crate) mod local_storage_tests; -mod restic_tests; mod restore_downloader_tests; -mod sync_tests; diff --git a/src/tests/storage/rclone_target.rs b/src/tests/storage/rclone_target.rs index 4812bf1..190c726 100644 --- a/src/tests/storage/rclone_target.rs +++ b/src/tests/storage/rclone_target.rs @@ -194,8 +194,8 @@ async fn gcs_emulator_target_reaches_fake_gcs_through_rclone() { #[tokio::test] async fn s3_target_hosts_a_restic_repository() { use crate::services::backup::logger::JobLogger; - use crate::services::restic::backup::snapshot; - use crate::services::restic::command::ResticRepo; + use crate::domain::files::restic::backup::snapshot; + use crate::domain::files::restic::repo::ResticRepo; use base64::{Engine as _, engine::general_purpose}; let (_container, endpoint) = start_minio().await; // "http://host:port" @@ -236,7 +236,7 @@ async fn s3_target_hosts_a_restic_repository() { #[tokio::test] async fn s3_targets_host_one_sync_replica_per_channel() { use crate::services::backup::logger::JobLogger; - use crate::services::sync::backup::one_storage; + use crate::services::backup::files_engine::{FilesEngine, one_storage}; use crate::tests::services::backup_uploader_tests::ctx_pointing_at; use wiremock::matchers::{body_partial_json, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -284,7 +284,7 @@ async fn s3_targets_host_one_sync_replica_per_channel() { let ctx = ctx_pointing_at(server.uri()); for c in &channels { - let result = one_storage(&ctx, &cfg, c, "backup-1", &JobLogger::new()).await; + let result = one_storage(&ctx, &cfg, FilesEngine::Sync, c, "backup-1", &JobLogger::new()).await; assert!(result.success, "{:?}", result.error); assert_eq!(result.remote_file_path.as_deref(), Some(format!("backups/sync/{}/current", cfg.generated_id).as_str())); assert_eq!(result.total_size, Some(2), "backups.file_size is the replica size"); diff --git a/src/utils/edge_key.rs b/src/utils/edge_key.rs index d18b87d..7a8875d 100644 --- a/src/utils/edge_key.rs +++ b/src/utils/edge_key.rs @@ -1,6 +1,9 @@ +use anyhow::Context; use base64::{Engine as _, engine::general_purpose}; +use hmac::{Hmac, Mac}; use serde::Deserialize; use serde_json::Value; +use sha2::Sha256; use thiserror::Error; #[derive(Debug, Deserialize)] @@ -45,3 +48,21 @@ pub fn decode_edge_key(edge_key: &str) -> Result { Err(EdgeKeyError::InvalidKey) } } + +pub fn derive_key(master_key_b64: &str, info: &[u8]) -> anyhow::Result<[u8; 32]> { + let master_key = general_purpose::STANDARD + .decode(master_key_b64.trim()) + .context("master key is not valid base64")?; + Ok(hkdf_sha256_32(&master_key, info)) +} + +pub fn hkdf_sha256_32(ikm: &[u8], info: &[u8]) -> [u8; 32] { + let mut extract = Hmac::::new_from_slice(&[]).expect("HMAC takes any key length"); + extract.update(ikm); + let prk = extract.finalize().into_bytes(); + + let mut expand = Hmac::::new_from_slice(&prk).expect("HMAC takes any key length"); + expand.update(info); + expand.update(&[1]); + expand.finalize().into_bytes().into() +} diff --git a/src/utils/file.rs b/src/utils/file.rs index 10bac58..1168f45 100644 --- a/src/utils/file.rs +++ b/src/utils/file.rs @@ -46,14 +46,15 @@ pub fn full_file_name(encrypt: bool) -> String { } } -pub fn full_file_path(file_name: &String, folder_name: Option<&str>) -> String { - let folder_name = folder_name - .map(str::trim) - .map(|folder_name| folder_name.trim_matches(char::from(47))) +pub fn folder_or_default(folder_name: Option<&str>) -> &str { + folder_name + .map(|folder_name| folder_name.trim().trim_matches('/')) .filter(|folder_name| !folder_name.is_empty()) - .unwrap_or("backups"); + .unwrap_or("backups") +} - format!("{}/{}/{}", folder_name, chrono::Local::now().format("%Y-%m-%d"), file_name) +pub fn full_file_path(file_name: &String, folder_name: Option<&str>) -> String { + format!("{}/{}/{}", folder_or_default(folder_name), chrono::Local::now().format("%Y-%m-%d"), file_name) } const CHUNK_SIZE: usize = 16 * 1024 * 1024;