feat(backup): support cross-platform background database backups
This commit is contained in:
CSXFanMeng authored and GitHub committed
1 parent
250e0c0a10
commit
2f960f104c
56 files changed
+4195
-962
No files matched your search
@@ -36,6 +36,8 @@ log = "0.4"
|
||||
tokio = { version = "1", features = ["full"] }
|
||||
tokio-util = { version = "0.7", features = ["compat"] }
|
||||
chrono = { version = "0.4", features = ["serde"] }
|
||||
chrono-tz = "0.10"
|
||||
fs2 = "0.4"
|
||||
uuid = { version = "1", features = ["v4", "serde"] }
|
||||
tokio-postgres = { version = "0.7", features = ["with-chrono-0_4", "with-uuid-1", "with-serde_json-1"] }
|
||||
deadpool-postgres = { version = "0.14", features = ["rt_tokio_1"] }
|
||||
|
||||
@@ -2350,6 +2350,29 @@ async fn save_export_destination_identity(
|
||||
state.storage.save_state(&export_destination_state_key(dir), &value, "application/octet-stream").await
|
||||
}
|
||||
|
||||
/// Read/delete operations must not trust a replaced backup mount or create a missing directory.
|
||||
pub async fn verify_export_destination_identity(
|
||||
state: &crate::connection::AppState,
|
||||
dir: &std::path::Path,
|
||||
) -> Result<(), String> {
|
||||
if !dir.is_dir() {
|
||||
return Err(format!("Backup directory is unavailable: {}", dir.display()));
|
||||
}
|
||||
let recorded = state
|
||||
.storage
|
||||
.load_state(&export_destination_state_key(dir))
|
||||
.await?
|
||||
.map(|(bytes, _)| ExportDestinationIdentity::decode(&bytes))
|
||||
.transpose()?
|
||||
.flatten();
|
||||
if recorded.as_ref().is_some_and(|identity| {
|
||||
recorded_export_destination_identity_mismatch(identity, export_destination_identity_for_path(dir).as_ref())
|
||||
}) {
|
||||
return Err(format!("Backup directory now resolves to a different filesystem: {}", dir.display()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Returns whether this macOS destination still has the transient, untagged
|
||||
/// `st_dev` identity written by DBX versions before persistent volume UUIDs
|
||||
/// were introduced. The caller must require an explicit directory selection
|
||||
@@ -2692,6 +2715,19 @@ fn export_source_file_name(file_path: &str) -> String {
|
||||
format!("{stem}.sql")
|
||||
}
|
||||
|
||||
tokio::task_local! {
|
||||
static TRACKED_BACKUP_DESTINATION: (std::path::PathBuf, Arc<std::sync::atomic::AtomicBool>);
|
||||
}
|
||||
|
||||
/// Tracks successful create_new separately from preparation errors and path collisions.
|
||||
pub(crate) async fn track_backup_destination<F: std::future::Future>(
|
||||
path: std::path::PathBuf,
|
||||
created: Arc<std::sync::atomic::AtomicBool>,
|
||||
operation: F,
|
||||
) -> F::Output {
|
||||
TRACKED_BACKUP_DESTINATION.scope((path, created), operation).await
|
||||
}
|
||||
|
||||
async fn create_database_export_writer(
|
||||
state: &Arc<crate::connection::AppState>,
|
||||
request: &DatabaseExportRequest,
|
||||
@@ -2744,6 +2780,13 @@ async fn create_database_export_writer(
|
||||
// at the same path in between. Re-check the identity of the handle we
|
||||
// actually opened, not just the path, and refuse to keep a backup that
|
||||
// landed on the wrong filesystem. See #6327.
|
||||
if request.prevent_overwrite {
|
||||
let _ = TRACKED_BACKUP_DESTINATION.try_with(|(path, created)| {
|
||||
if path == std::path::Path::new(&request.file_path) {
|
||||
created.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
});
|
||||
}
|
||||
let opened_destination_identity = export_destination_identity_for_file(&file);
|
||||
if export_destination_identity_mismatch(
|
||||
expected_destination_identity.as_ref(),
|
||||
@@ -4253,6 +4296,37 @@ mod tests {
|
||||
assert_eq!(output, "SELECT 1;\n");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scheduled_backup_destination_tracking_grants_ownership_only_after_create_new() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let path = directory.path().join("tracked.sql");
|
||||
let state = Arc::new(test_app_state(directory.path()).await);
|
||||
let mut request = export_request(true, true, true, Vec::new());
|
||||
request.file_path = path.to_string_lossy().into_owned();
|
||||
request.prevent_overwrite = true;
|
||||
let created = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
let writer = super::track_backup_destination(
|
||||
path.clone(),
|
||||
created.clone(),
|
||||
create_database_export_writer(&state, &request),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(created.load(std::sync::atomic::Ordering::Relaxed));
|
||||
drop(writer);
|
||||
std::fs::write(&path, b"keep existing").unwrap();
|
||||
let collision = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
assert!(super::track_backup_destination(
|
||||
path.clone(),
|
||||
collision.clone(),
|
||||
create_database_export_writer(&state, &request)
|
||||
)
|
||||
.await
|
||||
.is_err());
|
||||
assert!(!collision.load(std::sync::atomic::Ordering::Relaxed));
|
||||
assert_eq!(std::fs::read(path).unwrap(), b"keep existing");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn backup_writer_does_not_overwrite_an_existing_file() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
|
||||
@@ -81,6 +81,7 @@ pub use persistence::saved_sql;
|
||||
pub use query::query_cancel;
|
||||
pub use query::redis_ops;
|
||||
pub use safety::risk_metrics;
|
||||
pub mod scheduled_backup;
|
||||
pub mod schema;
|
||||
pub use connection::session_credentials;
|
||||
pub use connection::task_supervisor;
|
||||
|
||||
@@ -0,0 +1,437 @@
|
||||
use chrono::Utc;
|
||||
use std::{fs::OpenOptions, path::Path, sync::Arc, time::Duration};
|
||||
use tokio::sync::watch;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use super::{models::*, BackupService};
|
||||
use crate::{
|
||||
connection::AppState,
|
||||
database_export::{self, DatabaseExportOutputCompression, DatabaseExportRequest, ExportStatus},
|
||||
models::connection::DatabaseType,
|
||||
query, schema,
|
||||
};
|
||||
|
||||
impl BackupService {
|
||||
pub(crate) async fn serve(&self, stop: CancellationToken, drain: CancellationToken) {
|
||||
let mut leader = None;
|
||||
let mut migration_ready = false;
|
||||
while !stop.is_cancelled() && !drain.is_cancelled() {
|
||||
if !migration_ready {
|
||||
migration_ready = match self.state.storage.inspect_data_migration().await {
|
||||
Ok(status) => status.is_ready(),
|
||||
Err(error) => {
|
||||
log::warn!("[database-backup] waiting for data migration: {error}");
|
||||
false
|
||||
}
|
||||
};
|
||||
}
|
||||
if migration_ready && leader.is_none() {
|
||||
let acquired = std::fs::create_dir_all(&self.store.directory)
|
||||
.and_then(|_| {
|
||||
OpenOptions::new()
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.read(true)
|
||||
.write(true)
|
||||
.open(self.store.directory.join("worker.lock"))
|
||||
})
|
||||
.ok()
|
||||
.filter(|file| fs2::FileExt::try_lock_exclusive(file).is_ok());
|
||||
if let Some(file) = acquired {
|
||||
match self.store.recover().await {
|
||||
Ok(()) => leader = Some(file),
|
||||
Err(error) => log::error!("[database-backup] recovery failed: {error}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
if migration_ready && leader.is_some() && !drain.is_cancelled() {
|
||||
if let Err(error) = self.tick(&stop).await {
|
||||
log::error!("[database-backup] scheduler failed: {error}");
|
||||
}
|
||||
}
|
||||
tokio::select! { _ = stop.cancelled() => break, _ = drain.cancelled() => break, _ = tokio::time::sleep(Duration::from_secs(2)) => {} }
|
||||
}
|
||||
// Dropping the locked handle transfers leadership, including after an unclean process exit.
|
||||
drop(leader);
|
||||
}
|
||||
|
||||
async fn tick(&self, stop: &CancellationToken) -> Result<(), String> {
|
||||
self.store.heartbeat().await?;
|
||||
self.store.enqueue_due().await?;
|
||||
let Some(job) = self.store.claim().await? else {
|
||||
return Ok(());
|
||||
};
|
||||
let state = Arc::new(AppState::new_with_plugin_dir(
|
||||
self.state.storage.clone(),
|
||||
self.state.storage.data_dir().join("plugins"),
|
||||
));
|
||||
let cancellation = stop.child_token();
|
||||
if self.store.cancelled(job.run.id.clone()).await? {
|
||||
cancellation.cancel();
|
||||
}
|
||||
let (progress, receiver) = watch::channel(job.run.clone());
|
||||
let service = self.clone();
|
||||
let worker_state = state.clone();
|
||||
let worker_cancel = cancellation.clone();
|
||||
let run_id = job.run.id.clone();
|
||||
let fallback = job.run.clone();
|
||||
let runtime = tokio::runtime::Handle::current();
|
||||
// SQL formatting and compression perform synchronous writes; keep them off async worker threads.
|
||||
let mut worker = tokio::task::spawn_blocking(move || {
|
||||
runtime.block_on(service.execute(job, worker_state, worker_cancel, progress))
|
||||
});
|
||||
let mut run = loop {
|
||||
tokio::select! {
|
||||
result = &mut worker => break match result {
|
||||
Ok(run) => run,
|
||||
Err(error) => { let mut run = receiver.borrow().clone(); run.status = "failed".into(); run.error = Some(format!("Backup worker failed: {error}")); break run; }
|
||||
},
|
||||
_ = tokio::time::sleep(Duration::from_millis(500)) => {
|
||||
if stop.is_cancelled() || self.store.cancelled(run_id.clone()).await.unwrap_or(true) {
|
||||
cancellation.cancel();
|
||||
database_export::set_export_cancelled(&run_id).await;
|
||||
}
|
||||
if let Err(error) = self.store.heartbeat().await { log::error!("[database-backup] heartbeat failed: {error}"); cancellation.cancel(); }
|
||||
let latest = receiver.borrow().clone();
|
||||
if let Err(error) = self.store.progress(latest).await { log::error!("[database-backup] progress persistence failed: {error}"); cancellation.cancel(); }
|
||||
}
|
||||
}
|
||||
};
|
||||
database_export::clear_export_cancelled(&run_id).await;
|
||||
state.shutdown(Duration::from_secs(3)).await;
|
||||
if run.status != "success" {
|
||||
if let Err(error) = self.cleanup(&mut run).await {
|
||||
run.error = Some(format!(
|
||||
"{}; partial files retained: {error}",
|
||||
run.error.as_deref().unwrap_or("Backup cancelled")
|
||||
));
|
||||
}
|
||||
}
|
||||
run.completed_at = Some(Utc::now().to_rfc3339());
|
||||
let success = run.status == "success";
|
||||
self.store.finish(run).await?;
|
||||
if success {
|
||||
let snapshot = self.store.snapshot().await?;
|
||||
if let Some(schedule) = snapshot.schedules.iter().find(|s| Some(&s.id) == fallback.schedule_id.as_ref()) {
|
||||
let ids = snapshot
|
||||
.runs
|
||||
.iter()
|
||||
.filter(|r| r.schedule_id.as_ref() == Some(&schedule.id) && r.status == "success")
|
||||
.skip(schedule.retention_count)
|
||||
.map(|r| r.id.clone())
|
||||
.collect();
|
||||
self.delete_runs(ids).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn execute(
|
||||
&self,
|
||||
mut job: Job,
|
||||
state: Arc<AppState>,
|
||||
stop: CancellationToken,
|
||||
progress: watch::Sender<BackupRun>,
|
||||
) -> BackupRun {
|
||||
let result = self.export_job(&mut job, &state, &stop, &progress).await;
|
||||
job.run.status = if stop.is_cancelled() {
|
||||
"cancelled"
|
||||
} else if result.is_ok() {
|
||||
"success"
|
||||
} else {
|
||||
"failed"
|
||||
}
|
||||
.into();
|
||||
job.run.error = result.err();
|
||||
if job.run.status == "success" {
|
||||
job.run.progress_percent = 100.0;
|
||||
}
|
||||
job.run
|
||||
}
|
||||
|
||||
async fn export_job(
|
||||
&self,
|
||||
job: &mut Job,
|
||||
state: &Arc<AppState>,
|
||||
stop: &CancellationToken,
|
||||
progress: &watch::Sender<BackupRun>,
|
||||
) -> Result<(), String> {
|
||||
check_cancel(stop)?;
|
||||
job.config.validate()?;
|
||||
self.validate_destination(&job.config.destination_directory)?;
|
||||
let connection = state
|
||||
.storage
|
||||
.load_connections()
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|c| c.id == job.config.connection_id)
|
||||
.ok_or("Saved backup connection is unavailable")?;
|
||||
if !matches!(connection.db_type, DatabaseType::Mysql | DatabaseType::Postgres) {
|
||||
return Err("This connection does not support consistent scheduled backups".into());
|
||||
}
|
||||
if !connection.save_password || connection.one_time {
|
||||
return Err("Unattended backups require a saved connection and saved credentials".into());
|
||||
}
|
||||
let postgres = connection.db_type == DatabaseType::Postgres;
|
||||
job.run.connection_name = connection.name.clone();
|
||||
state.configs.write().await.insert(connection.id.clone(), connection);
|
||||
check_cancel(stop)?;
|
||||
let available: Vec<String> =
|
||||
schema::list_databases_core(state, &job.config.connection_id).await?.into_iter().map(|d| d.name).collect();
|
||||
let databases = if job.config.databases.is_empty() {
|
||||
available.iter().filter(|name| !system_database(name, postgres)).cloned().collect::<Vec<_>>()
|
||||
} else {
|
||||
for name in &job.config.databases {
|
||||
if !available.contains(name) {
|
||||
return Err(format!("Configured backup database is unavailable: {name}"));
|
||||
}
|
||||
}
|
||||
job.config.databases.clone()
|
||||
};
|
||||
if databases.is_empty() {
|
||||
return Err("No databases are available for this backup".into());
|
||||
}
|
||||
let mut sensitive = true;
|
||||
if !postgres && job.config.table_filter_mode != "all" {
|
||||
if let Ok(result) = query::execute_sql_statement(
|
||||
state,
|
||||
&job.config.connection_id,
|
||||
"",
|
||||
"SHOW VARIABLES LIKE 'lower_case_table_names'",
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
sensitive = !result
|
||||
.rows
|
||||
.first()
|
||||
.and_then(|r| r.get(1))
|
||||
.is_some_and(|v| matches!(v.as_str(), Some("1" | "2")) || matches!(v.as_i64(), Some(1 | 2)));
|
||||
}
|
||||
}
|
||||
let directory = Path::new(&job.config.destination_directory).to_path_buf();
|
||||
let output = if let Some(pattern) = &job.directory_pattern {
|
||||
directory.join(render_template(pattern, &job.run, "", &job.time_zone, true)?)
|
||||
} else {
|
||||
directory
|
||||
};
|
||||
let mut included_count = 0;
|
||||
for (db_index, database) in databases.iter().enumerate() {
|
||||
check_cancel(stop)?;
|
||||
let schemas = if postgres {
|
||||
schema::list_schemas_core(state, &job.config.connection_id, database)
|
||||
.await?
|
||||
.into_iter()
|
||||
.filter(|s| s != "information_schema" && !s.starts_with("pg_"))
|
||||
.collect::<Vec<_>>()
|
||||
} else {
|
||||
vec![database.clone()]
|
||||
};
|
||||
if schemas.is_empty() {
|
||||
return Err(format!("Database {database} has no exportable schemas"));
|
||||
}
|
||||
// Metadata discovery may be slow; finish it before starting the snapshot idle timeout.
|
||||
let mut targets = Vec::new();
|
||||
for schema_name in &schemas {
|
||||
check_cancel(stop)?;
|
||||
let mut selected_tables = Vec::new();
|
||||
let mut excluded_tables = Vec::new();
|
||||
if job.config.table_filter_mode != "all" {
|
||||
let tables = schema::list_tables_core(
|
||||
state,
|
||||
&job.config.connection_id,
|
||||
database,
|
||||
schema_name,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
for table in &tables {
|
||||
let matched = job
|
||||
.config
|
||||
.table_patterns
|
||||
.iter()
|
||||
.any(|p| matches_pattern(p, &table.name, database, schema_name, sensitive));
|
||||
if matched {
|
||||
if job.config.table_filter_mode == "include" {
|
||||
selected_tables.push(table.name.clone());
|
||||
} else {
|
||||
excluded_tables.push(table.name.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
let included = if job.config.table_filter_mode == "include" {
|
||||
selected_tables.len()
|
||||
} else {
|
||||
tables.len() - excluded_tables.len()
|
||||
};
|
||||
included_count += included;
|
||||
if included == 0 {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
targets.push((schema_name.clone(), selected_tables, excluded_tables));
|
||||
}
|
||||
if targets.is_empty() {
|
||||
continue;
|
||||
}
|
||||
check_cancel(stop)?;
|
||||
let snapshot = database_export::begin_database_backup_snapshot_core_for_export(
|
||||
state,
|
||||
&job.config.connection_id,
|
||||
database,
|
||||
Some(&job.run.id),
|
||||
)
|
||||
.await?;
|
||||
let result = async {
|
||||
for (schema_index, (schema_name, selected_tables, excluded_tables)) in targets.iter().enumerate() {
|
||||
check_cancel(stop)?;
|
||||
let stem = if schemas.len() > 1 { format!("{database}.{schema_name}") } else { database.clone() };
|
||||
let name = render_template(
|
||||
job.config.file_name_pattern.as_deref().unwrap_or(DEFAULT_FILE),
|
||||
&job.run,
|
||||
&stem,
|
||||
&job.time_zone,
|
||||
false,
|
||||
)?;
|
||||
let extension = if job.config.output_compression == "gzip" { "sql.gz" } else { "sql" };
|
||||
let path = output.join(format!("{name}.{extension}"));
|
||||
check_output_ancestors(&path, Path::new(&job.config.destination_directory))?;
|
||||
if path.try_exists().map_err(|e| e.to_string())? {
|
||||
return Err(format!("Backup file already exists: {}", path.display()));
|
||||
}
|
||||
if job.run.files.iter().any(|f| Path::new(&f.file_path) == path) {
|
||||
return Err("Backup filename template produced duplicate paths".into());
|
||||
}
|
||||
let file = BackupFile {
|
||||
database: database.clone(),
|
||||
schema: schema_name.clone(),
|
||||
display_name: stem,
|
||||
file_path: path.to_string_lossy().into_owned(),
|
||||
owned: false,
|
||||
};
|
||||
job.run.files.push(file.clone());
|
||||
self.store.progress(job.run.clone()).await?;
|
||||
progress.send_replace(job.run.clone());
|
||||
let request = DatabaseExportRequest {
|
||||
export_id: job.run.id.clone(),
|
||||
connection_id: job.config.connection_id.clone(),
|
||||
database: database.clone(),
|
||||
schema: schema_name.clone(),
|
||||
file_path: file.file_path,
|
||||
selected_tables: selected_tables.clone(),
|
||||
excluded_tables: excluded_tables.clone(),
|
||||
include_structure: job.config.include_structure,
|
||||
include_data: job.config.include_data,
|
||||
include_objects: job.config.include_objects,
|
||||
include_create_database: false,
|
||||
drop_table_if_exists: job.config.drop_table_if_exists,
|
||||
omit_auto_increment: false,
|
||||
fail_on_error: true,
|
||||
prevent_overwrite: true,
|
||||
output_compression: if job.config.output_compression == "gzip" {
|
||||
DatabaseExportOutputCompression::Gzip
|
||||
} else {
|
||||
DatabaseExportOutputCompression::None
|
||||
},
|
||||
snapshot_session_id: Some(snapshot.session_id.clone()),
|
||||
batch_size: 1000,
|
||||
split_max_mb: None,
|
||||
};
|
||||
let terminal_cancelled = std::sync::atomic::AtomicBool::new(false);
|
||||
let created = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
let result = database_export::track_backup_destination(
|
||||
path,
|
||||
created.clone(),
|
||||
database_export::export_database_sql_core(state, &request, |event| {
|
||||
if matches!(event.status, ExportStatus::Cancelled) {
|
||||
terminal_cancelled.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
let mut run = job.run.clone();
|
||||
if let Some(file) = run.files.last_mut() {
|
||||
file.owned = created.load(std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
let fraction = if matches!(event.status, ExportStatus::Done) {
|
||||
1.0
|
||||
} else if event.total_objects > 0 {
|
||||
event.object_index as f64 / event.total_objects as f64
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
run.progress_percent = ((db_index as f64
|
||||
+ (schema_index as f64 + fraction) / targets.len() as f64)
|
||||
/ databases.len() as f64
|
||||
* 100.0)
|
||||
.min(99.0);
|
||||
progress.send_replace(run);
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
if created.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
if let Some(file) = job.run.files.last_mut() {
|
||||
file.owned = true;
|
||||
}
|
||||
} else {
|
||||
// Preparation failures and external writers grant no file deletion authority.
|
||||
job.run.files.pop();
|
||||
}
|
||||
result?;
|
||||
if terminal_cancelled.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
stop.cancel();
|
||||
}
|
||||
check_cancel(stop)?;
|
||||
job.run.progress_percent = progress.borrow().progress_percent;
|
||||
}
|
||||
Ok::<_, String>(())
|
||||
}
|
||||
.await;
|
||||
let rollback = query::rollback_manual_transaction(state, &snapshot.session_id).await;
|
||||
result?;
|
||||
rollback?;
|
||||
}
|
||||
if job.config.table_filter_mode != "all" && included_count == 0 {
|
||||
return Err("No tables matched the backup table scope".into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn check_cancel(stop: &CancellationToken) -> Result<(), String> {
|
||||
if stop.is_cancelled() {
|
||||
Err("Backup cancelled".into())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn system_database(name: &str, postgres: bool) -> bool {
|
||||
let name = name.to_ascii_lowercase();
|
||||
if postgres {
|
||||
matches!(name.as_str(), "template0" | "template1")
|
||||
} else {
|
||||
matches!(name.as_str(), "information_schema" | "mysql" | "performance_schema" | "sys")
|
||||
}
|
||||
}
|
||||
|
||||
fn check_output_ancestors(path: &Path, root: &Path) -> Result<(), String> {
|
||||
let relative = path.strip_prefix(root).map_err(|_| "Backup output escaped destination")?;
|
||||
let mut current = root.to_path_buf();
|
||||
for component in relative.components() {
|
||||
if !matches!(component, std::path::Component::Normal(_)) {
|
||||
return Err("Invalid backup output path".into());
|
||||
}
|
||||
current.push(component);
|
||||
match std::fs::symlink_metadata(¤t) {
|
||||
Ok(metadata) if metadata.file_type().is_symlink() => {
|
||||
return Err("Backup output contains a symbolic link".into())
|
||||
}
|
||||
Err(error) if error.kind() != std::io::ErrorKind::NotFound => return Err(error.to_string()),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
//! Durable, UI-independent database backup scheduling shared by desktop and Web.
|
||||
mod engine;
|
||||
mod models;
|
||||
mod service;
|
||||
mod store;
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
pub use models::{BackupConfig, BackupFile, BackupRun, BackupSchedule, Migration, RunRequest};
|
||||
pub use service::{BackupCommand, BackupService};
|
||||
pub use store::{BackupSnapshot, BackupStore};
|
||||
|
||||
/// Metadata and export command chains nest very large async futures - a single
|
||||
/// frame can reach 60-150 KiB - which does not fit into tokio's default 2 MiB
|
||||
/// worker stack. Every runtime that drives them (desktop, background worker and
|
||||
/// Web server) has to reserve the same roomy stack, otherwise the process dies
|
||||
/// with `fatal runtime error: stack overflow` instead of reporting an error.
|
||||
pub const WORKER_STACK_SIZE: usize = 16 * 1024 * 1024;
|
||||
|
||||
/// Builds the multi-threaded runtime used by processes that run metadata and
|
||||
/// backup work, on a stack that can hold those nested futures.
|
||||
pub fn worker_runtime() -> std::io::Result<tokio::runtime::Runtime> {
|
||||
tokio::runtime::Builder::new_multi_thread().enable_all().thread_stack_size(WORKER_STACK_SIZE).build()
|
||||
}
|
||||
@@ -0,0 +1,297 @@
|
||||
use chrono::{DateTime, Datelike, Duration, LocalResult, NaiveTime, TimeZone, Utc};
|
||||
use chrono_tz::Tz;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
pub const DEFAULT_DIRECTORY: &str = "dbx-backup__{schedule}__{timestamp}__{runId}";
|
||||
pub const DEFAULT_FILE: &str = "dbx-backup__{schedule}__{timestamp}__{database}__{runId}";
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackupConfig {
|
||||
pub connection_id: String,
|
||||
#[serde(default)]
|
||||
pub databases: Vec<String>,
|
||||
#[serde(default = "all")]
|
||||
pub table_filter_mode: String,
|
||||
#[serde(default)]
|
||||
pub table_patterns: Vec<String>,
|
||||
pub destination_directory: String,
|
||||
pub include_structure: bool,
|
||||
pub include_data: bool,
|
||||
pub include_objects: bool,
|
||||
#[serde(default)]
|
||||
pub drop_table_if_exists: bool,
|
||||
#[serde(default = "none")]
|
||||
pub output_compression: String,
|
||||
#[serde(default)]
|
||||
pub file_name_pattern: Option<String>,
|
||||
}
|
||||
|
||||
fn all() -> String {
|
||||
"all".into()
|
||||
}
|
||||
fn none() -> String {
|
||||
"none".into()
|
||||
}
|
||||
fn utc() -> String {
|
||||
"UTC".into()
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackupSchedule {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub enabled: bool,
|
||||
#[serde(flatten)]
|
||||
pub config: BackupConfig,
|
||||
pub frequency: String,
|
||||
pub interval_hours: u32,
|
||||
pub time_of_day: String,
|
||||
pub weekday: u32,
|
||||
pub retention_count: usize,
|
||||
#[serde(default)]
|
||||
pub run_directory_pattern: Option<String>,
|
||||
#[serde(default = "utc")]
|
||||
pub time_zone: String,
|
||||
pub created_at: String,
|
||||
pub updated_at: String,
|
||||
#[serde(default)]
|
||||
pub next_run_at: String,
|
||||
#[serde(default)]
|
||||
pub last_run_at: Option<String>,
|
||||
#[serde(default)]
|
||||
pub last_run_status: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackupFile {
|
||||
pub database: String,
|
||||
pub schema: String,
|
||||
pub display_name: String,
|
||||
pub file_path: String,
|
||||
#[serde(default = "owned_by_legacy_run")]
|
||||
pub owned: bool,
|
||||
}
|
||||
|
||||
fn owned_by_legacy_run() -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackupRun {
|
||||
pub id: String,
|
||||
#[serde(default)]
|
||||
pub schedule_id: Option<String>,
|
||||
pub schedule_name: String,
|
||||
#[serde(default)]
|
||||
pub display_name: Option<String>,
|
||||
pub connection_id: String,
|
||||
pub connection_name: String,
|
||||
#[serde(default)]
|
||||
pub destination_directory: String,
|
||||
pub trigger: String,
|
||||
pub source: String,
|
||||
pub status: String,
|
||||
pub started_at: String,
|
||||
#[serde(default)]
|
||||
pub completed_at: Option<String>,
|
||||
#[serde(default)]
|
||||
pub files: Vec<BackupFile>,
|
||||
#[serde(default)]
|
||||
pub progress_percent: f64,
|
||||
#[serde(default)]
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RunRequest {
|
||||
pub schedule_id: Option<String>,
|
||||
pub config: Option<BackupConfig>,
|
||||
pub display_name: Option<String>,
|
||||
pub time_zone: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub(crate) struct Job {
|
||||
pub run: BackupRun,
|
||||
pub config: BackupConfig,
|
||||
pub directory_pattern: Option<String>,
|
||||
pub time_zone: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct Migration {
|
||||
#[serde(default)]
|
||||
pub schedules: Vec<BackupSchedule>,
|
||||
#[serde(default)]
|
||||
pub runs: Vec<BackupRun>,
|
||||
}
|
||||
|
||||
impl BackupConfig {
|
||||
pub fn validate(&self) -> Result<(), String> {
|
||||
bounded_text(&self.connection_id, 256, "Connection")?;
|
||||
bounded_text(&self.destination_directory, 4096, "Backup directory")?;
|
||||
if !std::path::Path::new(&self.destination_directory).is_absolute() {
|
||||
return Err("Backup directory must be absolute".into());
|
||||
}
|
||||
if !self.include_structure && !self.include_data && !self.include_objects {
|
||||
return Err("Select backup contents".into());
|
||||
}
|
||||
if !matches!(self.output_compression.as_str(), "none" | "gzip") {
|
||||
return Err("Unsupported backup compression".into());
|
||||
}
|
||||
if !matches!(self.table_filter_mode.as_str(), "all" | "include" | "exclude") {
|
||||
return Err("Unsupported backup table filter".into());
|
||||
}
|
||||
if self.databases.len() > 10_000 || self.table_patterns.len() > 1000 {
|
||||
return Err("Too many backup targets or patterns".into());
|
||||
}
|
||||
for value in self.databases.iter().chain(self.table_patterns.iter()) {
|
||||
bounded_text(value, 1024, "Backup target")?;
|
||||
}
|
||||
if self.table_filter_mode != "all" && self.table_patterns.is_empty() {
|
||||
return Err("Select backup table patterns".into());
|
||||
}
|
||||
validate_template(self.file_name_pattern.as_deref().unwrap_or(DEFAULT_FILE), false)
|
||||
}
|
||||
}
|
||||
|
||||
impl BackupSchedule {
|
||||
pub fn validate(&self) -> Result<(), String> {
|
||||
self.config.validate()?;
|
||||
bounded_text(&self.id, 256, "Schedule ID")?;
|
||||
bounded_text(&self.name, 256, "Schedule name")?;
|
||||
if !matches!(self.frequency.as_str(), "hourly" | "daily" | "weekly")
|
||||
|| !(1..=168).contains(&self.interval_hours)
|
||||
|| self.weekday > 6
|
||||
|| !(1..=100).contains(&self.retention_count)
|
||||
{
|
||||
return Err("Invalid backup schedule".into());
|
||||
}
|
||||
NaiveTime::parse_from_str(&self.time_of_day, "%H:%M").map_err(|_| "Invalid backup time")?;
|
||||
self.time_zone.parse::<Tz>().map_err(|_| "Invalid backup time zone")?;
|
||||
validate_template(self.run_directory_pattern.as_deref().unwrap_or(DEFAULT_DIRECTORY), true)
|
||||
}
|
||||
|
||||
pub fn next_after(&self, after: DateTime<Utc>) -> Result<DateTime<Utc>, String> {
|
||||
if self.frequency == "hourly" {
|
||||
return Ok(after + Duration::hours(self.interval_hours.clamp(1, 168) as i64));
|
||||
}
|
||||
let tz: Tz = self.time_zone.parse().map_err(|_| "Invalid backup time zone")?;
|
||||
let time = NaiveTime::parse_from_str(&self.time_of_day, "%H:%M").map_err(|_| "Invalid backup time")?;
|
||||
let today = after.with_timezone(&tz).date_naive();
|
||||
for offset in 0..15 {
|
||||
let date = today + Duration::days(offset);
|
||||
if self.frequency == "weekly" && date.weekday().num_days_from_sunday() != self.weekday {
|
||||
continue;
|
||||
}
|
||||
// A repeated wall-clock time runs only at its first occurrence.
|
||||
// A nonexistent spring-forward time runs at the first valid minute after the gap.
|
||||
for minute in 0..180 {
|
||||
let local = date.and_time(time) + Duration::minutes(minute);
|
||||
let candidate = match tz.from_local_datetime(&local) {
|
||||
LocalResult::Single(value) => Some(value),
|
||||
LocalResult::Ambiguous(first, _) => Some(first),
|
||||
LocalResult::None => None,
|
||||
};
|
||||
if let Some(value) = candidate {
|
||||
if value.with_timezone(&Utc) > after {
|
||||
return Ok(value.with_timezone(&Utc));
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err("Unable to calculate next backup time".into())
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn bounded_text(value: &str, max: usize, label: &str) -> Result<(), String> {
|
||||
if value.trim().is_empty() || value.len() > max || value.chars().any(char::is_control) {
|
||||
Err(format!("Invalid {label}"))
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn validate_template(value: &str, directory: bool) -> Result<(), String> {
|
||||
bounded_text(value, 1024, "backup name template")?;
|
||||
if directory {
|
||||
let drive_prefix = value.as_bytes().get(1) == Some(&b':') && value.as_bytes()[0].is_ascii_alphabetic();
|
||||
if value.starts_with(['/', '\\'])
|
||||
|| value.ends_with(['/', '\\'])
|
||||
|| drive_prefix
|
||||
|| value.split(['/', '\\']).any(|part| part == "." || part == "..")
|
||||
{
|
||||
return Err("Invalid backup directory template".into());
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
if value.trim() != value
|
||||
|| value.starts_with(['/', '\\'])
|
||||
|| value.ends_with(['/', '\\', '.', ' '])
|
||||
|| value.contains([':', '*', '?', '"', '<', '>', '|'])
|
||||
|| (!directory && value.contains(['/', '\\']))
|
||||
|| value.split(['/', '\\']).any(|part| part.is_empty() || part == "." || part == "..")
|
||||
{
|
||||
return Err("Invalid backup name template".into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn segment(value: &str) -> String {
|
||||
let cleaned: String =
|
||||
value.chars().map(|c| if c.is_control() || "\\/:*?\"<>|".contains(c) { '_' } else { c }).collect();
|
||||
let cleaned = cleaned.trim().trim_end_matches(['.', ' ']);
|
||||
let name = if cleaned.is_empty() { "database" } else { cleaned };
|
||||
let first = name.split('.').next().unwrap_or(name).to_ascii_uppercase();
|
||||
if matches!(first.as_str(), "CON" | "PRN" | "AUX" | "NUL")
|
||||
|| (first.len() == 4
|
||||
&& (first.starts_with("COM") || first.starts_with("LPT"))
|
||||
&& matches!(first.as_bytes()[3], b'1'..=b'9'))
|
||||
{
|
||||
format!("{name}_")
|
||||
} else {
|
||||
name.into()
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn render_template(
|
||||
template: &str,
|
||||
run: &BackupRun,
|
||||
database: &str,
|
||||
time_zone: &str,
|
||||
directory: bool,
|
||||
) -> Result<String, String> {
|
||||
validate_template(template, directory)?;
|
||||
let tz: Tz = time_zone.parse().map_err(|_| "Invalid backup time zone")?;
|
||||
let started =
|
||||
DateTime::parse_from_rfc3339(&run.started_at).map_err(|_| "Invalid backup start time")?.with_timezone(&tz);
|
||||
let timestamp = started.format(if directory { "%Y%m%d%H%M%S" } else { "%Y%m%d-%H%M%S" }).to_string();
|
||||
let rendered = template
|
||||
.replace("{schedule}", &segment(&run.schedule_name))
|
||||
.replace("{date}", &started.format("%Y%m%d").to_string())
|
||||
.replace("{timestamp}", ×tamp)
|
||||
.replace("{runId}", &run.id.chars().take(8).collect::<String>());
|
||||
if directory {
|
||||
Ok(rendered.split(['/', '\\']).filter(|part| !part.is_empty()).map(segment).collect::<Vec<_>>().join("/"))
|
||||
} else if template.contains("{database}") {
|
||||
Ok(segment(&rendered.replace("{database}", &segment(database))))
|
||||
} else {
|
||||
Ok(format!("{}__{}", segment(&rendered), segment(database)))
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn matches_pattern(pattern: &str, table: &str, database: &str, schema: &str, sensitive: bool) -> bool {
|
||||
let expression = format!("^{}$", regex::escape(pattern).replace("\\*", ".*").replace("\\?", "."));
|
||||
regex::RegexBuilder::new(&expression).case_insensitive(!sensitive).build().is_ok_and(|re| {
|
||||
re.is_match(table)
|
||||
|| (!schema.is_empty() && re.is_match(&format!("{schema}.{table}")))
|
||||
|| (!database.is_empty() && !schema.is_empty() && re.is_match(&format!("{database}.{schema}.{table}")))
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,212 @@
|
||||
use serde::Deserialize;
|
||||
use serde_json::Value;
|
||||
use std::{
|
||||
path::{Component, Path, PathBuf},
|
||||
sync::Arc,
|
||||
};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use super::{models::*, BackupStore};
|
||||
use crate::{connection::AppState, database_export::record_export_destination_identity};
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(tag = "action", rename_all = "camelCase")]
|
||||
pub enum BackupCommand {
|
||||
Snapshot,
|
||||
Preview { schedule: BackupSchedule },
|
||||
Save { schedule: BackupSchedule },
|
||||
DeleteSchedule { id: String },
|
||||
Migrate { migration: Migration },
|
||||
Run { request: RunRequest },
|
||||
Cancel { id: String },
|
||||
Rename { id: String, name: String },
|
||||
DeleteRuns { ids: Vec<String> },
|
||||
File { id: String, index: usize },
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct BackupService {
|
||||
pub(crate) state: Arc<AppState>,
|
||||
pub store: BackupStore,
|
||||
pub(crate) root: Option<PathBuf>,
|
||||
}
|
||||
|
||||
impl BackupService {
|
||||
/// Web callers must supply a server-owned root, never one from the request.
|
||||
pub fn new(state: Arc<AppState>, data_dir: &Path, root: Option<PathBuf>) -> Self {
|
||||
Self { state, store: BackupStore::new(data_dir), root }
|
||||
}
|
||||
|
||||
pub fn start(&self, stop: CancellationToken) -> tokio::task::JoinHandle<()> {
|
||||
self.start_with_drain(stop, CancellationToken::new())
|
||||
}
|
||||
|
||||
/// Draining finishes the active job before releasing leadership; stopping cancels it.
|
||||
pub fn start_with_drain(&self, stop: CancellationToken, drain: CancellationToken) -> tokio::task::JoinHandle<()> {
|
||||
let service = self.clone();
|
||||
tokio::spawn(async move { service.serve(stop, drain).await })
|
||||
}
|
||||
|
||||
pub async fn command(&self, command: BackupCommand) -> Result<Value, String> {
|
||||
match command {
|
||||
BackupCommand::File { id, index } => {
|
||||
Ok(Value::String(self.file(&id, index).await?.to_string_lossy().into_owned()))
|
||||
}
|
||||
BackupCommand::Preview { schedule } => {
|
||||
Ok(Value::String(schedule.next_after(chrono::Utc::now())?.to_rfc3339()))
|
||||
}
|
||||
BackupCommand::Snapshot => {
|
||||
let mut value = serde_json::to_value(self.store.snapshot().await?).map_err(|e| e.to_string())?;
|
||||
value["destinationRoot"] = serde_json::to_value(&self.root).map_err(|e| e.to_string())?;
|
||||
Ok(value)
|
||||
}
|
||||
BackupCommand::Save { schedule } => {
|
||||
schedule.validate()?;
|
||||
self.validate_destination(&schedule.config.destination_directory)?;
|
||||
let old = self.store.snapshot().await?.schedules.into_iter().find(|s| s.id == schedule.id);
|
||||
if old.is_none_or(|s| s.config.destination_directory != schedule.config.destination_directory) {
|
||||
record_export_destination_identity(&self.state, Path::new(&schedule.config.destination_directory))
|
||||
.await?;
|
||||
}
|
||||
serde_json::to_value(self.store.save_schedule(schedule).await?).map_err(|e| e.to_string())
|
||||
}
|
||||
BackupCommand::DeleteSchedule { id } => {
|
||||
self.store.delete_schedule(id).await?;
|
||||
Ok(Value::Null)
|
||||
}
|
||||
BackupCommand::Migrate { migration } => {
|
||||
if self.store.snapshot().await?.migrated {
|
||||
return Ok(Value::Null);
|
||||
}
|
||||
// Browser-local history must not grant authority over arbitrary server files.
|
||||
if self.root.is_some() && (!migration.schedules.is_empty() || !migration.runs.is_empty()) {
|
||||
return Err("Desktop backups cannot be imported into a Web server automatically".into());
|
||||
}
|
||||
self.store.migrate(migration).await?;
|
||||
Ok(Value::Null)
|
||||
}
|
||||
BackupCommand::Run { request } => {
|
||||
if let Some(config) = &request.config {
|
||||
self.validate_destination(&config.destination_directory)?;
|
||||
}
|
||||
serde_json::to_value(self.store.enqueue(request).await?).map_err(|e| e.to_string())
|
||||
}
|
||||
BackupCommand::Cancel { id } => Ok(Value::Bool(self.store.cancel(id).await?)),
|
||||
BackupCommand::Rename { id, name } => {
|
||||
self.store.rename(id, name).await?;
|
||||
Ok(Value::Null)
|
||||
}
|
||||
BackupCommand::DeleteRuns { ids } => {
|
||||
self.delete_runs(ids).await?;
|
||||
Ok(Value::Null)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn validate_destination(&self, path: &str) -> Result<PathBuf, String> {
|
||||
let path = Path::new(path);
|
||||
if !path.is_absolute() || path.components().any(|c| matches!(c, Component::ParentDir)) {
|
||||
return Err("Backup directory must be an absolute path without traversal".into());
|
||||
}
|
||||
let resolved = path.canonicalize().map_err(|e| format!("Backup directory is unavailable: {e}"))?;
|
||||
if !resolved.is_dir() {
|
||||
return Err("Backup destination is not a directory".into());
|
||||
}
|
||||
if let Some(root) = &self.root {
|
||||
let root = root.canonicalize().map_err(|e| format!("Server backup root is unavailable: {e}"))?;
|
||||
if !resolved.starts_with(root) {
|
||||
return Err("Backup directory is outside the server backup root".into());
|
||||
}
|
||||
}
|
||||
Ok(resolved)
|
||||
}
|
||||
|
||||
pub async fn file(&self, id: &str, index: usize) -> Result<PathBuf, String> {
|
||||
let snapshot = self.store.snapshot().await?;
|
||||
let run =
|
||||
snapshot.runs.iter().find(|r| r.id == id && r.status == "success").ok_or("Completed backup not found")?;
|
||||
let file = run.files.get(index).ok_or("Backup file not found")?;
|
||||
self.checked_file(run, Path::new(&file.file_path)).await
|
||||
}
|
||||
|
||||
async fn checked_file(&self, run: &BackupRun, path: &Path) -> Result<PathBuf, String> {
|
||||
let root = self.validate_destination(&run.destination_directory)?;
|
||||
crate::database_export::verify_export_destination_identity(&self.state, Path::new(&run.destination_directory))
|
||||
.await?;
|
||||
let metadata = std::fs::symlink_metadata(path).map_err(|e| e.to_string())?;
|
||||
if !metadata.is_file() || metadata.file_type().is_symlink() {
|
||||
return Err("Backup file is not a regular file".into());
|
||||
}
|
||||
let resolved = path.canonicalize().map_err(|e| e.to_string())?;
|
||||
if resolved == root || !resolved.starts_with(root) {
|
||||
return Err("Backup file escaped its destination".into());
|
||||
}
|
||||
let name = resolved.file_name().unwrap_or_default().to_string_lossy().to_ascii_lowercase();
|
||||
if !name.ends_with(".sql") && !name.ends_with(".sql.gz") {
|
||||
return Err("Not a SQL backup file".into());
|
||||
}
|
||||
Ok(resolved)
|
||||
}
|
||||
|
||||
pub(crate) async fn cleanup(&self, run: &mut BackupRun) -> Result<(), String> {
|
||||
let others = self.store.snapshot().await?.runs;
|
||||
let mut retained = Vec::new();
|
||||
let mut errors = Vec::new();
|
||||
for file in &run.files {
|
||||
if others.iter().any(|r| r.id != run.id && r.files.iter().any(|f| f.file_path == file.file_path)) {
|
||||
continue;
|
||||
}
|
||||
let path = Path::new(&file.file_path);
|
||||
match std::fs::symlink_metadata(path) {
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
|
||||
_ => {}
|
||||
}
|
||||
if !file.owned {
|
||||
retained.push(file.clone());
|
||||
errors.push("File ownership could not be confirmed after interruption; inspect the file before removing it manually".into());
|
||||
continue;
|
||||
}
|
||||
match self.checked_file(run, path).await {
|
||||
Ok(path) => {
|
||||
if let Err(e) = tokio::fs::remove_file(path).await {
|
||||
if e.kind() != std::io::ErrorKind::NotFound {
|
||||
retained.push(file.clone());
|
||||
errors.push(e.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
retained.push(file.clone());
|
||||
errors.push(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
run.files = retained;
|
||||
if errors.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(errors.join("; "))
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn delete_runs(&self, ids: Vec<String>) -> Result<(), String> {
|
||||
if ids.len() > 1000 {
|
||||
return Err("Too many backups selected".into());
|
||||
}
|
||||
let mut errors = Vec::new();
|
||||
for mut run in self.store.mark_deleting(ids).await? {
|
||||
if let Err(error) = self.cleanup(&mut run).await {
|
||||
run.error = Some(format!("Backup cleanup failed: {error}"));
|
||||
self.store.deletion_failed(run).await?;
|
||||
errors.push(error);
|
||||
continue;
|
||||
}
|
||||
self.store.remove_run(run.id).await?;
|
||||
}
|
||||
if errors.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(errors.join("; "))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,484 @@
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use chrono::Utc;
|
||||
use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior};
|
||||
use serde::{de::DeserializeOwned, Deserialize, Serialize};
|
||||
|
||||
use super::models::*;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct BackupStore {
|
||||
pub(crate) directory: PathBuf,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BackupSnapshot {
|
||||
pub schedules: Vec<BackupSchedule>,
|
||||
pub runs: Vec<BackupRun>,
|
||||
pub migrated: bool,
|
||||
pub heartbeat: Option<String>,
|
||||
}
|
||||
|
||||
fn encode(value: &impl Serialize) -> Result<String, String> {
|
||||
serde_json::to_string(value).map_err(|e| e.to_string())
|
||||
}
|
||||
fn decode<T: DeserializeOwned>(value: String) -> Result<T, String> {
|
||||
serde_json::from_str(&value).map_err(|e| e.to_string())
|
||||
}
|
||||
fn sql_error(error: rusqlite::Error) -> String {
|
||||
error.to_string()
|
||||
}
|
||||
|
||||
fn schedules(conn: &Connection) -> Result<Vec<BackupSchedule>, String> {
|
||||
let mut statement = conn.prepare("SELECT payload FROM schedules ORDER BY id").map_err(sql_error)?;
|
||||
let rows = statement.query_map([], |row| row.get::<_, String>(0)).map_err(sql_error)?;
|
||||
rows.map(|row| decode(row.map_err(sql_error)?)).collect()
|
||||
}
|
||||
|
||||
fn run_rows(conn: &Connection) -> Result<Vec<BackupRun>, String> {
|
||||
let mut statement =
|
||||
conn.prepare("SELECT payload FROM runs ORDER BY created_at DESC, id DESC").map_err(sql_error)?;
|
||||
let rows = statement.query_map([], |row| row.get::<_, String>(0)).map_err(sql_error)?;
|
||||
rows.map(|row| decode(row.map_err(sql_error)?)).collect()
|
||||
}
|
||||
|
||||
fn active(conn: &Connection, schedule_id: &str) -> Result<bool, String> {
|
||||
conn.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM runs WHERE schedule_id=? AND state IN ('queued','running'))",
|
||||
[schedule_id],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sql_error)
|
||||
}
|
||||
|
||||
fn insert_job(
|
||||
conn: &Connection,
|
||||
config: BackupConfig,
|
||||
schedule: Option<&BackupSchedule>,
|
||||
name: String,
|
||||
trigger: &str,
|
||||
time_zone: Option<String>,
|
||||
) -> Result<BackupRun, String> {
|
||||
let count: usize = conn
|
||||
.query_row("SELECT COUNT(*) FROM runs WHERE state IN ('queued','running')", [], |r| r.get(0))
|
||||
.map_err(sql_error)?;
|
||||
if count >= 100 {
|
||||
return Err("Too many pending backups".into());
|
||||
}
|
||||
let run = BackupRun {
|
||||
id: uuid::Uuid::new_v4().simple().to_string(),
|
||||
schedule_id: schedule.map(|s| s.id.clone()),
|
||||
schedule_name: name,
|
||||
display_name: None,
|
||||
connection_id: config.connection_id.clone(),
|
||||
connection_name: String::new(),
|
||||
destination_directory: config.destination_directory.clone(),
|
||||
trigger: trigger.into(),
|
||||
source: if schedule.is_some() { "scheduled" } else { "one-shot" }.into(),
|
||||
status: "running".into(),
|
||||
started_at: Utc::now().to_rfc3339(),
|
||||
completed_at: None,
|
||||
files: Vec::new(),
|
||||
progress_percent: 0.0,
|
||||
error: None,
|
||||
};
|
||||
let time_zone = schedule.map(|s| s.time_zone.clone()).or(time_zone).unwrap_or_else(|| "UTC".into());
|
||||
time_zone.parse::<chrono_tz::Tz>().map_err(|_| "Invalid backup time zone")?;
|
||||
let job = Job {
|
||||
run: run.clone(),
|
||||
config,
|
||||
directory_pattern: schedule
|
||||
.map(|s| s.run_directory_pattern.clone().unwrap_or_else(|| DEFAULT_DIRECTORY.into())),
|
||||
time_zone,
|
||||
};
|
||||
conn.execute(
|
||||
"INSERT INTO runs(id,schedule_id,state,payload,job,created_at) VALUES(?,?,'queued',?,?,?)",
|
||||
params![run.id, run.schedule_id, encode(&run)?, encode(&job)?, run.started_at],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
Ok(run)
|
||||
}
|
||||
|
||||
impl BackupStore {
|
||||
pub fn new(data_dir: &Path) -> Self {
|
||||
Self { directory: data_dir.join("database-backups") }
|
||||
}
|
||||
|
||||
async fn access<T: Send + 'static>(
|
||||
&self,
|
||||
action: impl FnOnce(&mut Connection) -> Result<T, String> + Send + 'static,
|
||||
) -> Result<T, String> {
|
||||
let dir = self.directory.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
|
||||
let mut conn = Connection::open(dir.join("state.db")).map_err(sql_error)?;
|
||||
conn.busy_timeout(std::time::Duration::from_secs(5)).map_err(sql_error)?;
|
||||
conn.execute_batch("PRAGMA journal_mode=WAL;
|
||||
CREATE TABLE IF NOT EXISTS schedules(id TEXT PRIMARY KEY,payload TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS runs(id TEXT PRIMARY KEY,schedule_id TEXT,state TEXT NOT NULL,payload TEXT NOT NULL,job TEXT,created_at TEXT NOT NULL,cancel INTEGER NOT NULL DEFAULT 0);
|
||||
CREATE INDEX IF NOT EXISTS runs_pending ON runs(state,created_at);
|
||||
CREATE TABLE IF NOT EXISTS metadata(key TEXT PRIMARY KEY,value TEXT NOT NULL);").map_err(sql_error)?;
|
||||
action(&mut conn)
|
||||
}).await.map_err(|e| e.to_string())?
|
||||
}
|
||||
|
||||
pub async fn snapshot(&self) -> Result<BackupSnapshot, String> {
|
||||
self.access(|conn| {
|
||||
let tx = conn.transaction().map_err(sql_error)?;
|
||||
Ok(BackupSnapshot {
|
||||
schedules: schedules(&tx)?,
|
||||
runs: run_rows(&tx)?,
|
||||
migrated: tx
|
||||
.query_row("SELECT EXISTS(SELECT 1 FROM metadata WHERE key='migrated')", [], |r| r.get(0))
|
||||
.map_err(sql_error)?,
|
||||
heartbeat: tx
|
||||
.query_row("SELECT value FROM metadata WHERE key='heartbeat'", [], |r| r.get(0))
|
||||
.optional()
|
||||
.map_err(sql_error)?,
|
||||
})
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn save_schedule(&self, mut value: BackupSchedule) -> Result<BackupSchedule, String> {
|
||||
value.validate()?;
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
if active(&tx, &value.id)? {
|
||||
return Err("Cannot edit a running backup schedule".into());
|
||||
}
|
||||
let old: Option<String> = tx
|
||||
.query_row("SELECT payload FROM schedules WHERE id=?", [&value.id], |r| r.get(0))
|
||||
.optional()
|
||||
.map_err(sql_error)?;
|
||||
let now = Utc::now();
|
||||
let mut timing_changed = true;
|
||||
if let Some(old) = old {
|
||||
let old: BackupSchedule = decode(old)?;
|
||||
if old.updated_at != value.updated_at {
|
||||
return Err("Backup schedule changed; reload before saving".into());
|
||||
}
|
||||
timing_changed = old.frequency != value.frequency
|
||||
|| old.interval_hours != value.interval_hours
|
||||
|| old.time_of_day != value.time_of_day
|
||||
|| old.weekday != value.weekday
|
||||
|| old.time_zone != value.time_zone
|
||||
|| (!old.enabled && value.enabled);
|
||||
value.created_at = old.created_at;
|
||||
value.last_run_at = old.last_run_at;
|
||||
value.last_run_status = old.last_run_status;
|
||||
value.next_run_at = old.next_run_at;
|
||||
} else {
|
||||
value.created_at = now.to_rfc3339();
|
||||
value.last_run_at = None;
|
||||
value.last_run_status = None;
|
||||
}
|
||||
if timing_changed {
|
||||
value.next_run_at = value.next_after(now)?.to_rfc3339();
|
||||
}
|
||||
value.updated_at = now.to_rfc3339();
|
||||
tx.execute(
|
||||
"INSERT INTO schedules(id,payload) VALUES(?,?) ON CONFLICT(id) DO UPDATE SET payload=excluded.payload",
|
||||
params![value.id, encode(&value)?],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
tx.commit().map_err(sql_error)?;
|
||||
Ok(value)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn delete_schedule(&self, id: String) -> Result<(), String> {
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
if active(&tx, &id)? {
|
||||
return Err("Cannot delete a running backup schedule".into());
|
||||
}
|
||||
tx.execute("DELETE FROM schedules WHERE id=?", [id]).map_err(sql_error)?;
|
||||
tx.commit().map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn migrate(&self, migration: Migration) -> Result<(), String> {
|
||||
if migration.schedules.len() > 1000 || migration.runs.len() > 10_000 {
|
||||
return Err("Backup migration is too large".into());
|
||||
}
|
||||
for schedule in &migration.schedules {
|
||||
schedule.validate()?;
|
||||
}
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let done: bool = tx
|
||||
.query_row("SELECT EXISTS(SELECT 1 FROM metadata WHERE key='migrated')", [], |r| r.get(0))
|
||||
.map_err(sql_error)?;
|
||||
if done {
|
||||
return Ok(());
|
||||
}
|
||||
for mut schedule in migration.schedules {
|
||||
if chrono::DateTime::parse_from_rfc3339(&schedule.next_run_at).is_err() {
|
||||
schedule.next_run_at = schedule.next_after(Utc::now())?.to_rfc3339();
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT OR IGNORE INTO schedules(id,payload) VALUES(?,?)",
|
||||
params![schedule.id, encode(&schedule)?],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
}
|
||||
for mut run in migration.runs {
|
||||
if run.status == "running" {
|
||||
run.status = "failed".into();
|
||||
run.error = Some("Backup interrupted before migration".into());
|
||||
run.completed_at = Some(Utc::now().to_rfc3339());
|
||||
}
|
||||
if !matches!(run.status.as_str(), "success" | "failed" | "cancelled") {
|
||||
return Err("Invalid migrated backup status".into());
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT OR IGNORE INTO runs(id,schedule_id,state,payload,created_at) VALUES(?,?,?,?,?)",
|
||||
params![run.id, run.schedule_id, run.status, encode(&run)?, run.started_at],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
}
|
||||
tx.execute("INSERT INTO metadata(key,value) VALUES('migrated','1')", []).map_err(sql_error)?;
|
||||
tx.commit().map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn enqueue(&self, request: RunRequest) -> Result<BackupRun, String> {
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let run = if let Some(id) = request.schedule_id {
|
||||
if active(&tx, &id)? {
|
||||
return Err("Backup schedule is already running".into());
|
||||
}
|
||||
let value: String =
|
||||
tx.query_row("SELECT payload FROM schedules WHERE id=?", [&id], |r| r.get(0)).map_err(sql_error)?;
|
||||
let schedule: BackupSchedule = decode(value)?;
|
||||
insert_job(&tx, schedule.config.clone(), Some(&schedule), schedule.name.clone(), "manual", None)?
|
||||
} else {
|
||||
let config = request.config.ok_or("Backup configuration is required")?;
|
||||
config.validate()?;
|
||||
let name = request.display_name.unwrap_or_else(|| "Database backup".into());
|
||||
bounded_text(&name, 256, "Backup name")?;
|
||||
insert_job(&tx, config, None, name, "manual", request.time_zone)?
|
||||
};
|
||||
tx.commit().map_err(sql_error)?;
|
||||
Ok(run)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn enqueue_due(&self) -> Result<(), String> {
|
||||
self.access(|conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let now = Utc::now();
|
||||
for mut schedule in schedules(&tx)? {
|
||||
if !schedule.enabled {
|
||||
continue;
|
||||
}
|
||||
let pending: usize = tx
|
||||
.query_row("SELECT COUNT(*) FROM runs WHERE state IN ('queued','running')", [], |r| r.get(0))
|
||||
.map_err(sql_error)?;
|
||||
if pending >= 100 {
|
||||
break;
|
||||
}
|
||||
let due = chrono::DateTime::parse_from_rfc3339(&schedule.next_run_at)
|
||||
.map_err(|_| "Invalid persisted backup time")?
|
||||
.with_timezone(&Utc)
|
||||
<= now;
|
||||
if !schedule.enabled || !due || active(&tx, &schedule.id)? {
|
||||
continue;
|
||||
}
|
||||
insert_job(&tx, schedule.config.clone(), Some(&schedule), schedule.name.clone(), "scheduled", None)?;
|
||||
// Persist advancement together with the queue entry; a crash cannot replay every missed interval.
|
||||
schedule.next_run_at = schedule.next_after(now)?.to_rfc3339();
|
||||
tx.execute("UPDATE schedules SET payload=? WHERE id=?", params![encode(&schedule)?, schedule.id])
|
||||
.map_err(sql_error)?;
|
||||
}
|
||||
tx.commit().map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn claim(&self) -> Result<Option<Job>, String> {
|
||||
self.access(|conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let payload: Option<String> = tx
|
||||
.query_row("SELECT job FROM runs WHERE state='queued' ORDER BY created_at,id LIMIT 1", [], |r| r.get(0))
|
||||
.optional()
|
||||
.map_err(sql_error)?;
|
||||
let job: Option<Job> = payload.map(decode).transpose()?;
|
||||
if let Some(job) = &job {
|
||||
tx.execute("UPDATE runs SET state='running' WHERE id=?", [&job.run.id]).map_err(sql_error)?;
|
||||
}
|
||||
tx.commit().map_err(sql_error)?;
|
||||
Ok(job)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn cancel(&self, id: String) -> Result<bool, String> {
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let queued: Option<String> = tx
|
||||
.query_row("SELECT payload FROM runs WHERE id=? AND state='queued'", [&id], |r| r.get(0))
|
||||
.optional()
|
||||
.map_err(sql_error)?;
|
||||
let accepted = if let Some(payload) = queued {
|
||||
let mut run: BackupRun = decode(payload)?;
|
||||
run.status = "cancelled".into();
|
||||
run.completed_at = Some(Utc::now().to_rfc3339());
|
||||
tx.execute(
|
||||
"UPDATE runs SET state='cancelled',cancel=1,job=NULL,payload=? WHERE id=?",
|
||||
params![encode(&run)?, id],
|
||||
)
|
||||
.map_err(sql_error)?
|
||||
> 0
|
||||
} else {
|
||||
tx.execute("UPDATE runs SET cancel=1 WHERE id=? AND state='running'", [&id]).map_err(sql_error)? > 0
|
||||
};
|
||||
tx.commit().map_err(sql_error)?;
|
||||
Ok(accepted)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn cancelled(&self, id: String) -> Result<bool, String> {
|
||||
self.access(move |conn| {
|
||||
conn.query_row("SELECT cancel FROM runs WHERE id=?", [id], |r| r.get(0)).map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn progress(&self, run: BackupRun) -> Result<(), String> {
|
||||
self.access(move |conn| {
|
||||
conn.execute("UPDATE runs SET payload=? WHERE id=? AND state='running'", params![encode(&run)?, run.id])
|
||||
.map_err(sql_error)?;
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn finish(&self, run: BackupRun) -> Result<(), String> {
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
tx.execute(
|
||||
"UPDATE runs SET state=?,payload=?,job=NULL WHERE id=?",
|
||||
params![run.status, encode(&run)?, run.id],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
if let Some(id) = &run.schedule_id {
|
||||
let payload: Option<String> = tx
|
||||
.query_row("SELECT payload FROM schedules WHERE id=?", [id], |r| r.get(0))
|
||||
.optional()
|
||||
.map_err(sql_error)?;
|
||||
if let Some(payload) = payload {
|
||||
let mut schedule: BackupSchedule = decode(payload)?;
|
||||
schedule.last_run_at = run.completed_at.clone();
|
||||
schedule.last_run_status = Some(run.status.clone());
|
||||
let now = Utc::now();
|
||||
if chrono::DateTime::parse_from_rfc3339(&schedule.next_run_at)
|
||||
.is_ok_and(|t| t.with_timezone(&Utc) <= now)
|
||||
{
|
||||
schedule.next_run_at = schedule.next_after(now)?.to_rfc3339();
|
||||
}
|
||||
tx.execute("UPDATE schedules SET payload=? WHERE id=?", params![encode(&schedule)?, id])
|
||||
.map_err(sql_error)?;
|
||||
}
|
||||
}
|
||||
tx.commit().map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn rename(&self, id: String, name: String) -> Result<(), String> {
|
||||
bounded_text(&name, 256, "Backup name")?;
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let payload: String = tx
|
||||
.query_row(
|
||||
"SELECT payload FROM runs WHERE id=? AND state IN ('success','failed','cancelled')",
|
||||
[&id],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
let mut run: BackupRun = decode(payload)?;
|
||||
run.display_name = Some(name);
|
||||
tx.execute("UPDATE runs SET payload=? WHERE id=?", params![encode(&run)?, id]).map_err(sql_error)?;
|
||||
tx.commit().map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn mark_deleting(&self, ids: Vec<String>) -> Result<Vec<BackupRun>, String> {
|
||||
self.access(move |conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
let mut runs = Vec::new();
|
||||
for id in ids {
|
||||
let payload: String = tx
|
||||
.query_row(
|
||||
"SELECT payload FROM runs WHERE id=? AND state IN ('success','failed','cancelled','deleting')",
|
||||
[&id],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
runs.push(decode(payload)?);
|
||||
tx.execute("UPDATE runs SET state='deleting' WHERE id=?", [id]).map_err(sql_error)?;
|
||||
}
|
||||
tx.commit().map_err(sql_error)?;
|
||||
Ok(runs)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn remove_run(&self, id: String) -> Result<(), String> {
|
||||
self.access(move |conn| {
|
||||
conn.execute("DELETE FROM runs WHERE id=? AND state='deleting'", [id]).map_err(sql_error)?;
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn deletion_failed(&self, run: BackupRun) -> Result<(), String> {
|
||||
self.access(move |conn| {
|
||||
conn.execute(
|
||||
"UPDATE runs SET state=?,payload=? WHERE id=? AND state='deleting'",
|
||||
params![run.status, encode(&run)?, run.id],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) async fn heartbeat(&self) -> Result<(), String> {
|
||||
self.access(|conn| { conn.execute("INSERT INTO metadata(key,value) VALUES('heartbeat',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", [Utc::now().to_rfc3339()]).map_err(sql_error)?; Ok(()) }).await
|
||||
}
|
||||
|
||||
pub(crate) async fn recover(&self) -> Result<(), String> {
|
||||
self.access(|conn| {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(sql_error)?;
|
||||
for mut run in run_rows(&tx)? {
|
||||
let state: String =
|
||||
tx.query_row("SELECT state FROM runs WHERE id=?", [&run.id], |r| r.get(0)).map_err(sql_error)?;
|
||||
if state != "running" && state != "deleting" {
|
||||
continue;
|
||||
}
|
||||
run.status = "failed".into();
|
||||
run.error =
|
||||
Some("Backup interrupted: the previous worker stopped. Retained files may be incomplete.".into());
|
||||
run.completed_at = Some(Utc::now().to_rfc3339());
|
||||
tx.execute(
|
||||
"UPDATE runs SET state='failed',payload=?,job=NULL WHERE id=?",
|
||||
params![encode(&run)?, run.id],
|
||||
)
|
||||
.map_err(sql_error)?;
|
||||
}
|
||||
tx.commit().map_err(sql_error)
|
||||
})
|
||||
.await
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,642 @@
|
||||
use super::*;
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde_json::json;
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
fn schedule(dir: &std::path::Path) -> BackupSchedule {
|
||||
serde_json::from_value(json!({
|
||||
"id":"daily", "name":"Daily backup", "enabled":true, "connectionId":"mysql", "databases":[],
|
||||
"destinationDirectory":dir.to_string_lossy(), "includeStructure":true, "includeData":true, "includeObjects":true,
|
||||
"frequency":"daily", "intervalHours":1, "timeOfDay":"02:30", "weekday":0, "retentionCount":2,
|
||||
"timeZone":"America/New_York", "createdAt":"2026-01-01T00:00:00Z", "updatedAt":"2026-01-01T00:00:00Z",
|
||||
"nextRunAt":"2026-01-01T00:00:00Z"
|
||||
})).unwrap()
|
||||
}
|
||||
fn utc(value: &str) -> DateTime<Utc> {
|
||||
value.parse().unwrap()
|
||||
}
|
||||
fn request(config: BackupConfig) -> RunRequest {
|
||||
RunRequest { schedule_id: None, config: Some(config), display_name: None, time_zone: None }
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wall_clock_schedule_handles_dst_and_does_not_repeat_ambiguous_time() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut schedule = schedule(dir.path());
|
||||
assert_eq!(schedule.next_after(utc("2026-03-08T06:00:00Z")).unwrap(), utc("2026-03-08T07:00:00Z"));
|
||||
schedule.time_of_day = "01:30".into();
|
||||
assert_eq!(schedule.next_after(utc("2026-11-01T04:00:00Z")).unwrap(), utc("2026-11-01T05:30:00Z"));
|
||||
assert_eq!(schedule.next_after(utc("2026-11-01T05:31:00Z")).unwrap(), utc("2026-11-02T06:30:00Z"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hourly_uses_elapsed_time_and_weekly_uses_saved_zone() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut schedule = schedule(dir.path());
|
||||
schedule.frequency = "hourly".into();
|
||||
assert_eq!(schedule.next_after(utc("2026-03-08T06:30:00Z")).unwrap(), utc("2026-03-08T07:30:00Z"));
|
||||
schedule.frequency = "weekly".into();
|
||||
schedule.time_zone = "Asia/Shanghai".into();
|
||||
assert_eq!(schedule.next_after(utc("2026-09-12T10:00:00Z")).unwrap(), utc("2026-09-12T18:30:00Z"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_schedule_and_path_templates_are_rejected() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut schedule = schedule(dir.path());
|
||||
schedule.time_zone = "invalid/timezone".into();
|
||||
assert!(schedule.validate().is_err());
|
||||
for path in ["../escape", "/absolute", "C:\\escape", "safe/../escape", "a/", "a\\..\\b"] {
|
||||
assert!(super::models::validate_template(path, true).is_err(), "{path}");
|
||||
}
|
||||
assert!(super::models::validate_template("{date}/{schedule}/{runId}", true).is_ok());
|
||||
assert!(super::models::matches_pattern("app.public.user?", "users", "app", "public", true));
|
||||
assert!(!super::models::matches_pattern("PUBLIC.*", "users", "app", "public", true));
|
||||
assert!(super::models::matches_pattern("PUBLIC.*", "users", "app", "public", false));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn schedule_survives_reopen_and_rejects_stale_edit() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let store = BackupStore::new(dir.path());
|
||||
let first = store.save_schedule(schedule(dir.path())).await.unwrap();
|
||||
let mut next = first.clone();
|
||||
next.name = "Renamed".into();
|
||||
store.save_schedule(next).await.unwrap();
|
||||
assert!(store.save_schedule(first).await.unwrap_err().contains("changed"));
|
||||
let reopened = BackupStore::new(dir.path()).snapshot().await.unwrap();
|
||||
assert_eq!(reopened.schedules[0].name, "Renamed");
|
||||
assert!(DateTime::parse_from_rfc3339(&reopened.schedules[0].next_run_at).unwrap().with_timezone(&Utc) > Utc::now());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn migration_is_atomic_idempotent_and_keeps_only_one_catchup_job() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let store = BackupStore::new(dir.path());
|
||||
let mut invalid = schedule(dir.path());
|
||||
invalid.time_zone = "bad".into();
|
||||
assert!(store.migrate(Migration { schedules: vec![schedule(dir.path()), invalid], runs: vec![] }).await.is_err());
|
||||
assert!(!store.snapshot().await.unwrap().migrated);
|
||||
store.migrate(Migration { schedules: vec![schedule(dir.path())], runs: vec![] }).await.unwrap();
|
||||
store.migrate(Migration::default()).await.unwrap();
|
||||
let (a, b) = tokio::join!(store.enqueue_due(), store.enqueue_due());
|
||||
a.unwrap();
|
||||
b.unwrap();
|
||||
let snapshot = store.snapshot().await.unwrap();
|
||||
assert!(snapshot.migrated);
|
||||
assert_eq!(snapshot.schedules.len(), 1);
|
||||
assert_eq!(snapshot.runs.len(), 1);
|
||||
assert_eq!(snapshot.runs[0].trigger, "scheduled");
|
||||
assert!(utc(&snapshot.schedules[0].next_run_at) > Utc::now());
|
||||
assert!(store.delete_schedule("daily".into()).await.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_manual_requests_do_not_duplicate_a_schedule() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let store = BackupStore::new(dir.path());
|
||||
store.save_schedule(schedule(dir.path())).await.unwrap();
|
||||
let req = RunRequest { schedule_id: Some("daily".into()), config: None, display_name: None, time_zone: None };
|
||||
let (left, right) = tokio::join!(store.enqueue(req.clone()), store.enqueue(req));
|
||||
assert_ne!(left.is_ok(), right.is_ok());
|
||||
assert_eq!(store.snapshot().await.unwrap().runs.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovery_preserves_queued_jobs_and_marks_orphaned_running_jobs_failed() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let store = BackupStore::new(dir.path());
|
||||
let run = store.enqueue(request(schedule(dir.path()).config)).await.unwrap();
|
||||
store.recover().await.unwrap();
|
||||
assert_eq!(store.claim().await.unwrap().unwrap().run.id, run.id);
|
||||
store.recover().await.unwrap();
|
||||
assert!(store.claim().await.unwrap().is_none());
|
||||
let recovered = &store.snapshot().await.unwrap().runs[0];
|
||||
assert_eq!(recovered.status, "failed");
|
||||
assert!(recovered.error.as_ref().unwrap().contains("interrupted"));
|
||||
}
|
||||
|
||||
async fn service(dir: &std::path::Path, root: Option<std::path::PathBuf>) -> BackupService {
|
||||
let storage = crate::storage::Storage::open(&dir.join("dbx.db")).await.unwrap();
|
||||
BackupService::new(Arc::new(crate::connection::AppState::new(storage)), dir, root)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn web_rejects_external_directories_and_untrusted_history_import() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let root = dir.path().join("backups");
|
||||
std::fs::create_dir(&root).unwrap();
|
||||
let service = service(dir.path(), Some(root.clone())).await;
|
||||
assert!(service.command(BackupCommand::Save { schedule: schedule(dir.path()) }).await.is_err());
|
||||
assert!(service
|
||||
.command(BackupCommand::Migrate { migration: Migration { schedules: vec![schedule(&root)], runs: vec![] } })
|
||||
.await
|
||||
.is_err());
|
||||
service.command(BackupCommand::Save { schedule: schedule(&root) }).await.unwrap();
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
assert_eq!(snapshot.schedules.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn deletion_only_removes_recorded_files_within_the_backup_root() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let root = dir.path().join("backups");
|
||||
std::fs::create_dir(&root).unwrap();
|
||||
let service = service(dir.path(), Some(root.clone())).await;
|
||||
let inside = root.join("test.sql");
|
||||
std::fs::write(&inside, "SELECT 1;").unwrap();
|
||||
let outside = dir.path().join("outside.sql");
|
||||
std::fs::write(&outside, "keep").unwrap();
|
||||
let mut run = service.store.enqueue(request(schedule(&root).config)).await.unwrap();
|
||||
service.store.claim().await.unwrap();
|
||||
run.status = "success".into();
|
||||
run.files = vec![BackupFile {
|
||||
database: "app".into(),
|
||||
schema: "app".into(),
|
||||
display_name: "test".into(),
|
||||
file_path: outside.to_string_lossy().into_owned(),
|
||||
owned: true,
|
||||
}];
|
||||
service.store.finish(run.clone()).await.unwrap();
|
||||
assert!(service.file(&run.id, 0).await.is_err());
|
||||
assert!(service.delete_runs(vec![run.id.clone()]).await.is_err());
|
||||
assert!(outside.exists());
|
||||
assert!(inside.exists());
|
||||
run.files[0].file_path = inside.to_string_lossy().into_owned();
|
||||
run.files[0].owned = false;
|
||||
service.store.finish(run.clone()).await.unwrap();
|
||||
assert!(service.delete_runs(vec![run.id.clone()]).await.is_err());
|
||||
assert!(inside.exists());
|
||||
run.files[0].owned = true;
|
||||
service.store.finish(run.clone()).await.unwrap();
|
||||
assert_eq!(service.file(&run.id, 0).await.unwrap(), inside.canonicalize().unwrap());
|
||||
assert!(service.file(&run.id, 1).await.is_err());
|
||||
assert!(service.file("unknown", 0).await.is_err());
|
||||
service.delete_runs(vec![run.id]).await.unwrap();
|
||||
assert!(!inside.exists());
|
||||
assert!(outside.exists());
|
||||
assert!(service.store.snapshot().await.unwrap().runs.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn draining_transfers_leadership_without_cancelling_and_cancel_survives_reopen() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let service = service(dir.path(), None).await;
|
||||
let run = service.store.enqueue(request(schedule(dir.path()).config)).await.unwrap();
|
||||
assert!(service.store.cancel(run.id.clone()).await.unwrap());
|
||||
let first_stop = CancellationToken::new();
|
||||
let first_drain = CancellationToken::new();
|
||||
let second_stop = CancellationToken::new();
|
||||
let first = service.start_with_drain(first_stop.clone(), first_drain.clone());
|
||||
tokio::time::timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
if snapshot.heartbeat.is_some() {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let probe =
|
||||
std::fs::OpenOptions::new().read(true).write(true).open(service.store.directory.join("worker.lock")).unwrap();
|
||||
assert!(fs2::FileExt::try_lock_exclusive(&probe).is_err());
|
||||
let second = service.start(second_stop.clone());
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
first_drain.cancel();
|
||||
tokio::time::timeout(Duration::from_secs(5), first).await.unwrap().unwrap();
|
||||
assert!(!first_stop.is_cancelled());
|
||||
let previous = service.store.snapshot().await.unwrap().heartbeat;
|
||||
tokio::time::timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
if service.store.snapshot().await.unwrap().heartbeat != previous {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(fs2::FileExt::try_lock_exclusive(&probe).is_err());
|
||||
second_stop.cancel();
|
||||
second.await.unwrap();
|
||||
fs2::FileExt::try_lock_exclusive(&probe).unwrap();
|
||||
fs2::FileExt::unlock(&probe).unwrap();
|
||||
let snapshot = BackupStore::new(dir.path()).snapshot().await.unwrap();
|
||||
assert_eq!(snapshot.runs.len(), 1);
|
||||
assert_eq!(snapshot.runs[0].status, "cancelled");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn worker_waits_for_security_migration_and_resumes_without_restarting() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let legacy_path = directory.path().join("connections.json");
|
||||
std::fs::write(&legacy_path, "[]").unwrap();
|
||||
let storage = crate::storage::Storage::open_unmigrated(&directory.path().join("dbx.db")).await.unwrap();
|
||||
let service = BackupService::new(Arc::new(crate::connection::AppState::new(storage)), directory.path(), None);
|
||||
let queued = service.store.enqueue(request(schedule(directory.path()).config)).await.unwrap();
|
||||
assert!(!service.state.storage.inspect_data_migration().await.unwrap().is_ready());
|
||||
let stop = CancellationToken::new();
|
||||
let worker = service.start(stop.clone());
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(2200)).await;
|
||||
let waiting = service.store.snapshot().await.unwrap();
|
||||
assert!(waiting.heartbeat.is_none());
|
||||
assert_eq!(waiting.runs[0].id, queued.id);
|
||||
assert_eq!(waiting.runs[0].status, queued.status);
|
||||
assert_eq!(std::fs::read_to_string(&legacy_path).unwrap(), "[]");
|
||||
assert!(!directory.path().join("connections.json.bak").exists());
|
||||
|
||||
service.state.storage.start_data_migration().await.unwrap();
|
||||
let resumed = tokio::time::timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
if snapshot.heartbeat.is_some() && snapshot.runs[0].status == "failed" {
|
||||
break snapshot;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
})
|
||||
.await;
|
||||
stop.cancel();
|
||||
tokio::time::timeout(Duration::from_secs(5), worker).await.unwrap().unwrap();
|
||||
let resumed = resumed.unwrap();
|
||||
assert_eq!(resumed.runs[0].id, queued.id);
|
||||
assert!(resumed.runs[0].error.is_some());
|
||||
assert!(!legacy_path.exists());
|
||||
assert!(directory.path().join("connections.json.bak").exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stopped_or_drained_startup_does_not_consume_existing_jobs() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let service = service(dir.path(), None).await;
|
||||
let queued = service.store.enqueue(request(schedule(dir.path()).config)).await.unwrap();
|
||||
for draining in [false, true] {
|
||||
let stop = CancellationToken::new();
|
||||
let drain = CancellationToken::new();
|
||||
if draining {
|
||||
drain.cancel();
|
||||
} else {
|
||||
stop.cancel();
|
||||
}
|
||||
tokio::time::timeout(Duration::from_secs(5), service.start_with_drain(stop, drain)).await.unwrap().unwrap();
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
assert!(snapshot.heartbeat.is_none());
|
||||
assert_eq!(snapshot.runs.len(), 1);
|
||||
assert_eq!(snapshot.runs[0].id, queued.id);
|
||||
assert_eq!(snapshot.runs[0].status, queued.status);
|
||||
}
|
||||
assert_eq!(service.store.claim().await.unwrap().unwrap().run.id, queued.id);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upgrade_reopen_recovers_interrupted_work_without_losing_queued_jobs_or_disabled_schedules() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let store = BackupStore::new(dir.path());
|
||||
let mut disabled = schedule(dir.path());
|
||||
disabled.enabled = false;
|
||||
store.save_schedule(disabled).await.unwrap();
|
||||
let interrupted = store.enqueue(request(schedule(dir.path()).config)).await.unwrap();
|
||||
assert_eq!(store.claim().await.unwrap().unwrap().run.id, interrupted.id);
|
||||
let queued = store.enqueue(request(schedule(dir.path()).config)).await.unwrap();
|
||||
let cancelled = store.enqueue(request(schedule(dir.path()).config)).await.unwrap();
|
||||
store.cancel(cancelled.id.clone()).await.unwrap();
|
||||
drop(store);
|
||||
|
||||
let reopened = BackupStore::new(dir.path());
|
||||
reopened.recover().await.unwrap();
|
||||
reopened.recover().await.unwrap();
|
||||
let snapshot = reopened.snapshot().await.unwrap();
|
||||
assert!(!snapshot.schedules[0].enabled);
|
||||
assert_eq!(snapshot.runs.len(), 3);
|
||||
assert_eq!(snapshot.runs.iter().find(|run| run.id == interrupted.id).unwrap().status, "failed");
|
||||
assert_eq!(snapshot.runs.iter().find(|run| run.id == cancelled.id).unwrap().status, "cancelled");
|
||||
assert_eq!(reopened.claim().await.unwrap().unwrap().run.id, queued.id);
|
||||
assert!(reopened.claim().await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore = "requires a disposable MySQL endpoint configured by DBX_LIVE_SQL_FILE_MYSQL_* variables"]
|
||||
fn live_mysql_worker_exports_saved_connection_and_applies_retention() {
|
||||
live_backup_runtime().block_on(live_mysql_worker_scenario());
|
||||
}
|
||||
|
||||
async fn live_mysql_worker_scenario() {
|
||||
use crate::{models::connection::ConnectionConfig, query::execute_sql_statement};
|
||||
use futures::FutureExt;
|
||||
use std::io::Read;
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let service = service(dir.path(), None).await;
|
||||
let config: ConnectionConfig = serde_json::from_value(json!({
|
||||
"id":"mysql", "name":"Worker test", "db_type":"mysql", "save_password":true,
|
||||
"host":std::env::var("DBX_LIVE_SQL_FILE_MYSQL_HOST").unwrap(),
|
||||
"port":std::env::var("DBX_LIVE_SQL_FILE_MYSQL_PORT").ok().and_then(|p| p.parse::<u16>().ok()).unwrap_or(3306),
|
||||
"username":std::env::var("DBX_LIVE_SQL_FILE_MYSQL_USER").unwrap(),
|
||||
"password":std::env::var("DBX_LIVE_SQL_FILE_MYSQL_PASSWORD").unwrap(),
|
||||
"database":null, "connect_timeout_secs":10, "query_timeout_secs":30
|
||||
}))
|
||||
.unwrap();
|
||||
service.state.storage.save_connections(std::slice::from_ref(&config)).await.unwrap();
|
||||
service.state.configs.write().await.insert(config.id.clone(), config);
|
||||
let database = format!("dbx_worker_{}", uuid::Uuid::new_v4().simple());
|
||||
execute_sql_statement(&service.state, "mysql", "", &format!("CREATE DATABASE `{database}`"), None, None)
|
||||
.await
|
||||
.unwrap();
|
||||
let stop = CancellationToken::new();
|
||||
let worker = service.start(stop.clone());
|
||||
let outcome = std::panic::AssertUnwindSafe(async {
|
||||
execute_sql_statement(
|
||||
&service.state,
|
||||
"mysql",
|
||||
&database,
|
||||
"CREATE TABLE chosen (id INT PRIMARY KEY, value TEXT)",
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
execute_sql_statement(
|
||||
&service.state,
|
||||
"mysql",
|
||||
&database,
|
||||
"INSERT INTO chosen VALUES (101, 'background-worker-sentinel')",
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
execute_sql_statement(&service.state, "mysql", &database, "CREATE TABLE skipped (id INT)", None, None).await?;
|
||||
let mut plan = schedule(dir.path());
|
||||
plan.enabled = false;
|
||||
plan.retention_count = 1;
|
||||
plan.config.databases = vec![database.clone()];
|
||||
plan.config.output_compression = "gzip".into();
|
||||
plan.config.table_filter_mode = "include".into();
|
||||
plan.config.table_patterns = vec!["chosen".into()];
|
||||
service.command(BackupCommand::Save { schedule: plan }).await?;
|
||||
let mut previous = None;
|
||||
for _ in 0..2 {
|
||||
let run = service
|
||||
.store
|
||||
.enqueue(RunRequest {
|
||||
schedule_id: Some("daily".into()),
|
||||
config: None,
|
||||
display_name: None,
|
||||
time_zone: None,
|
||||
})
|
||||
.await?;
|
||||
let completed = tokio::time::timeout(Duration::from_secs(60), async {
|
||||
loop {
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
let current = snapshot.runs.iter().find(|r| r.id == run.id).unwrap();
|
||||
if current.status != "running" {
|
||||
break current.clone();
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|_| "Backup timed out")?;
|
||||
assert_eq!(completed.status, "success", "{:?}", completed.error);
|
||||
assert_eq!(completed.files.len(), 1);
|
||||
let path = std::path::PathBuf::from(&completed.files[0].file_path);
|
||||
let mut sql = String::new();
|
||||
flate2::read::GzDecoder::new(std::fs::File::open(&path).unwrap()).read_to_string(&mut sql).unwrap();
|
||||
assert!(sql.contains("background-worker-sentinel"));
|
||||
assert!(!sql.contains("CREATE TABLE `skipped`"));
|
||||
if let Some(previous) = previous {
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
while std::path::Path::new(&previous).exists() {
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|_| "Retention did not remove the old file")?;
|
||||
}
|
||||
previous = Some(path);
|
||||
}
|
||||
assert_eq!(service.store.snapshot().await?.runs.len(), 1);
|
||||
let mut empty_scope = schedule(dir.path()).config;
|
||||
empty_scope.databases = vec![database.clone()];
|
||||
empty_scope.table_filter_mode = "include".into();
|
||||
empty_scope.table_patterns = vec!["missing_table".into()];
|
||||
let run = service.store.enqueue(request(empty_scope)).await?;
|
||||
let completed = tokio::time::timeout(Duration::from_secs(60), async {
|
||||
loop {
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
let current = snapshot.runs.iter().find(|r| r.id == run.id).unwrap();
|
||||
if current.status != "running" {
|
||||
break current.clone();
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|_| "Empty-scope backup timed out")?;
|
||||
assert_eq!(completed.status, "failed");
|
||||
assert!(completed.error.as_deref().unwrap().contains("No tables matched"));
|
||||
assert!(completed.files.is_empty());
|
||||
assert!(previous.unwrap().exists());
|
||||
Ok::<_, String>(())
|
||||
})
|
||||
.catch_unwind()
|
||||
.await;
|
||||
stop.cancel();
|
||||
worker.await.unwrap();
|
||||
let cleanup =
|
||||
execute_sql_statement(&service.state, "mysql", "", &format!("DROP DATABASE `{database}`"), None, None).await;
|
||||
service.state.shutdown(Duration::from_secs(3)).await;
|
||||
cleanup.unwrap();
|
||||
outcome.unwrap().unwrap();
|
||||
}
|
||||
|
||||
/// Runs a live scenario on a runtime configured like the processes that drive
|
||||
/// backups. The export path nests very large async futures, so it needs the same
|
||||
/// roomy worker stack the desktop runtime sets; on tokio's default stack the
|
||||
/// process aborts with `fatal runtime error: stack overflow` instead of
|
||||
/// reporting a failed assertion.
|
||||
fn live_backup_runtime() -> tokio::runtime::Runtime {
|
||||
super::worker_runtime().unwrap()
|
||||
}
|
||||
|
||||
fn live_postgres_config(id: &str, database: &str) -> crate::models::connection::ConnectionConfig {
|
||||
serde_json::from_value(json!({
|
||||
"id": id, "name": id, "db_type": "postgres", "save_password": true,
|
||||
"host": std::env::var("DBX_LIVE_SQL_FILE_POSTGRES_HOST").unwrap(),
|
||||
"port": std::env::var("DBX_LIVE_SQL_FILE_POSTGRES_PORT").ok().and_then(|p| p.parse::<u16>().ok()).unwrap_or(5432),
|
||||
"username": std::env::var("DBX_LIVE_SQL_FILE_POSTGRES_USER").unwrap(),
|
||||
"password": std::env::var("DBX_LIVE_SQL_FILE_POSTGRES_PASSWORD").unwrap(),
|
||||
"database": database, "connect_timeout_secs": 10, "query_timeout_secs": 30
|
||||
}))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn await_run(service: &BackupService, run_id: &str) -> BackupRun {
|
||||
tokio::time::timeout(Duration::from_secs(60), async {
|
||||
loop {
|
||||
let snapshot = service.store.snapshot().await.unwrap();
|
||||
let current = snapshot.runs.iter().find(|r| r.id == run_id).unwrap().clone();
|
||||
if current.status != "running" && current.status != "queued" {
|
||||
break current;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("backup run did not finish within 60s")
|
||||
}
|
||||
|
||||
async fn enqueue_and_run(service: &BackupService, schedule_id: &str) -> BackupRun {
|
||||
let run = service
|
||||
.store
|
||||
.enqueue(RunRequest {
|
||||
schedule_id: Some(schedule_id.into()),
|
||||
config: None,
|
||||
display_name: None,
|
||||
time_zone: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
await_run(service, &run.id).await
|
||||
}
|
||||
|
||||
fn plan_with_id(dir: &std::path::Path, id: &str) -> BackupSchedule {
|
||||
let mut plan = schedule(dir);
|
||||
plan.id = id.to_string();
|
||||
plan.enabled = false;
|
||||
plan
|
||||
}
|
||||
|
||||
fn read_backup(path: &str, gzip: bool) -> String {
|
||||
let file = std::fs::File::open(path).unwrap();
|
||||
if !gzip {
|
||||
return std::fs::read_to_string(path).unwrap();
|
||||
}
|
||||
use std::io::Read;
|
||||
let mut sql = String::new();
|
||||
flate2::read::GzDecoder::new(file).read_to_string(&mut sql).unwrap();
|
||||
sql
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore = "requires a disposable PostgreSQL endpoint configured by DBX_LIVE_SQL_FILE_POSTGRES_* variables"]
|
||||
fn live_postgres_worker_exports_selected_tables_across_schemas() {
|
||||
use crate::query::execute_sql_statement;
|
||||
live_backup_runtime().block_on(async {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let service = service(dir.path(), None).await;
|
||||
let stop = CancellationToken::new();
|
||||
let worker = service.start(stop.clone());
|
||||
let admin_id = "postgres-admin";
|
||||
let admin = live_postgres_config(admin_id, "postgres");
|
||||
service.state.storage.save_connections(std::slice::from_ref(&admin)).await.unwrap();
|
||||
service.state.configs.write().await.insert(admin.id.clone(), admin);
|
||||
let database = format!("dbx_pg_backup_{}", uuid::Uuid::new_v4().simple());
|
||||
execute_sql_statement(
|
||||
&service.state,
|
||||
admin_id,
|
||||
"postgres",
|
||||
&format!("CREATE DATABASE \"{database}\""),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let config = live_postgres_config("postgres", &database);
|
||||
service.state.storage.save_connections(std::slice::from_ref(&config)).await.unwrap();
|
||||
service.state.configs.write().await.insert(config.id.clone(), config);
|
||||
|
||||
for statement in [
|
||||
"CREATE SCHEMA app",
|
||||
"CREATE SCHEMA audit",
|
||||
"CREATE TABLE public.accounts (id integer PRIMARY KEY, name text NOT NULL)",
|
||||
"CREATE TABLE app.orders (id integer PRIMARY KEY, total numeric(12,2) NOT NULL)",
|
||||
"CREATE TABLE app.\"MixedCase\" (id integer PRIMARY KEY)",
|
||||
"CREATE VIEW app.big_orders AS SELECT id, total FROM app.orders WHERE total > 100",
|
||||
"CREATE TABLE audit.events (id integer PRIMARY KEY, note text)",
|
||||
"INSERT INTO public.accounts VALUES (1, 'Alice')",
|
||||
"INSERT INTO app.orders VALUES (7, 42.50), (8, 420.00)",
|
||||
"INSERT INTO app.\"MixedCase\" VALUES (1)",
|
||||
"INSERT INTO audit.events VALUES (9, 'sentinel-event')",
|
||||
] {
|
||||
execute_sql_statement(&service.state, "postgres", &database, statement, None, None).await.unwrap();
|
||||
}
|
||||
|
||||
// Table scope: only `app.orders` (schema-qualified pattern) and the
|
||||
// case-sensitive `MixedCase` name, which pg resolves case-sensitively.
|
||||
let mut plan = plan_with_id(dir.path(), "scope");
|
||||
plan.retention_count = 2;
|
||||
plan.config.connection_id = "postgres".into();
|
||||
plan.config.databases = vec![database.clone()];
|
||||
plan.config.output_compression = "gzip".into();
|
||||
plan.config.table_filter_mode = "include".into();
|
||||
plan.config.table_patterns = vec!["app.orders".into(), "MixedCase".into()];
|
||||
service.command(BackupCommand::Save { schedule: plan }).await.unwrap();
|
||||
|
||||
let completed = enqueue_and_run(&service, "scope").await;
|
||||
assert_eq!(completed.status, "success", "{:?}", completed.error);
|
||||
assert_eq!(
|
||||
completed.files.len(),
|
||||
1,
|
||||
"{:?}",
|
||||
completed.files.iter().map(|f| &f.display_name).collect::<Vec<_>>()
|
||||
);
|
||||
let sql = read_backup(&completed.files[0].file_path, true);
|
||||
assert!(sql.contains("CREATE TABLE \"app\".\"orders\""), "{sql}");
|
||||
assert!(sql.contains("\"MixedCase\""), "{sql}");
|
||||
assert!(sql.contains("420.00"), "{sql}");
|
||||
assert!(!sql.contains("accounts"), "{sql}");
|
||||
assert!(!sql.contains("events"), "{sql}");
|
||||
// A view inside the same schema is not part of a table-name scope.
|
||||
assert!(!sql.contains("big_orders"), "{sql}");
|
||||
|
||||
// Patterns are case-sensitive on PostgreSQL, and a scope that matches
|
||||
// nothing must fail without leaving a partial file behind.
|
||||
let mut empty = plan_with_id(dir.path(), "empty");
|
||||
empty.config.connection_id = "postgres".into();
|
||||
empty.config.databases = vec![database.clone()];
|
||||
empty.config.output_compression = "gzip".into();
|
||||
empty.config.table_filter_mode = "include".into();
|
||||
empty.config.table_patterns = vec!["mixedcase".into()];
|
||||
service.command(BackupCommand::Save { schedule: empty }).await.unwrap();
|
||||
let cancelled = enqueue_and_run(&service, "empty").await;
|
||||
assert_eq!(cancelled.status, "failed");
|
||||
assert!(cancelled.error.as_deref().unwrap().contains("No tables matched"), "{:?}", cancelled.error);
|
||||
assert!(cancelled.files.is_empty());
|
||||
|
||||
// Every schema of the database is exported when no table scope is set.
|
||||
let mut full = plan_with_id(dir.path(), "full");
|
||||
full.config.connection_id = "postgres".into();
|
||||
full.config.databases = vec![database.clone()];
|
||||
full.config.output_compression = "gzip".into();
|
||||
service.command(BackupCommand::Save { schedule: full }).await.unwrap();
|
||||
let full_run = enqueue_and_run(&service, "full").await;
|
||||
assert_eq!(full_run.status, "success", "{:?}", full_run.error);
|
||||
assert_eq!(full_run.files.len(), 3, "{:?}", full_run.files.iter().map(|f| &f.display_name).collect::<Vec<_>>());
|
||||
let bodies =
|
||||
full_run.files.iter().map(|f| (f.schema.clone(), read_backup(&f.file_path, true))).collect::<Vec<_>>();
|
||||
let app = bodies.iter().find(|(schema, _)| schema == "app").unwrap();
|
||||
assert!(app.1.contains("CREATE TABLE \"app\".\"MixedCase\""));
|
||||
assert!(app.1.contains("big_orders"), "views belong to the schema-wide export");
|
||||
let audit = bodies.iter().find(|(schema, _)| schema == "audit").unwrap();
|
||||
assert!(audit.1.contains("sentinel-event"));
|
||||
assert!(!audit.1.contains("orders"));
|
||||
let public = bodies.iter().find(|(schema, _)| schema == "public").unwrap();
|
||||
assert!(public.1.contains("accounts"));
|
||||
|
||||
stop.cancel();
|
||||
worker.await.unwrap();
|
||||
let cleanup = execute_sql_statement(
|
||||
&service.state,
|
||||
admin_id,
|
||||
"postgres",
|
||||
&format!("DROP DATABASE \"{database}\" WITH (FORCE)"),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
service.state.shutdown(Duration::from_secs(3)).await;
|
||||
cleanup.unwrap();
|
||||
});
|
||||
}
|
||||
@@ -312,8 +312,12 @@ fn add_mq_routes(router: Router<Arc<WebState>>) -> Router<Arc<WebState>> {
|
||||
router
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
fn main() {
|
||||
let runtime = dbx_core::scheduled_backup::worker_runtime().expect("Failed to build tokio runtime");
|
||||
runtime.block_on(serve());
|
||||
}
|
||||
|
||||
async fn serve() {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
@@ -394,6 +398,7 @@ async fn main() {
|
||||
transfer_progress_channels: RwLock::new(HashMap::new()),
|
||||
table_import_channels: RwLock::new(HashMap::new()),
|
||||
sql_file_executions: RwLock::new(HashMap::new()),
|
||||
managed_sql_previews: Default::default(),
|
||||
nacos_imports: RwLock::new(HashMap::new()),
|
||||
login_rate_limit: tokio::sync::Mutex::new(state::LoginRateLimit { fail_count: 0, locked_until: None }),
|
||||
export_files: RwLock::new(HashMap::new()),
|
||||
@@ -402,6 +407,15 @@ async fn main() {
|
||||
});
|
||||
|
||||
ssh_prompt::install_web_ssh_prompt_bridge(web_state.ssh_prompts.clone());
|
||||
routes::sql_file::start_sql_file_cleanup(&web_state);
|
||||
|
||||
let backup_root = std::env::var_os("DBX_BACKUP_ROOT")
|
||||
.map(std::path::PathBuf::from)
|
||||
.unwrap_or_else(|| web_state.data_dir.join("backups"));
|
||||
std::fs::create_dir_all(&backup_root).expect("Failed to create server backup root");
|
||||
let backup_service = routes::scheduled_backup::service(&web_state).expect("Failed to resolve server backup root");
|
||||
let backup_stop = tokio_util::sync::CancellationToken::new();
|
||||
let backup_worker = backup_service.start(backup_stop.clone());
|
||||
|
||||
// API routes
|
||||
let api = Router::new()
|
||||
@@ -409,6 +423,9 @@ async fn main() {
|
||||
.route("/migration/start", post(routes::migration::start))
|
||||
.route("/migration/retry", post(routes::migration::retry))
|
||||
.route("/migration/cleanup-backups", post(routes::migration::cleanup_backups))
|
||||
.route("/database-backups", post(routes::scheduled_backup::command))
|
||||
.route("/database-backups/{id}/files/{index}", get(routes::scheduled_backup::download))
|
||||
.route("/database-backups/{id}/files/{index}/restore", post(routes::scheduled_backup::prepare_restore))
|
||||
// Auth
|
||||
.route("/auth/login", post(auth::login))
|
||||
.route("/auth/check", get(auth::check))
|
||||
@@ -1137,6 +1154,7 @@ async fn main() {
|
||||
.layer(DefaultBodyLimit::max(routes::sql_file::sql_file_upload_hard_cap_bytes())),
|
||||
)
|
||||
.route("/sql-file/execute", post(routes::sql_file::execute_sql_file))
|
||||
.route("/sql-file/preview/release", post(routes::sql_file::release_sql_file_preview))
|
||||
.route("/sql-file/tables", post(routes::sql_file::inspect_sql_file_tables))
|
||||
.route("/sql-file/progress/{executionId}", get(routes::sql_file::sql_file_progress))
|
||||
.route("/sql-file/cancel", post(routes::sql_file::cancel_sql_file))
|
||||
@@ -1263,13 +1281,20 @@ async fn main() {
|
||||
let listener = tokio::net::TcpListener::bind(addr).await.expect("Failed to bind address");
|
||||
let shutdown_state = web_state.app.clone();
|
||||
axum::serve(listener, app)
|
||||
.with_graceful_shutdown(async {
|
||||
if let Err(error) = tokio::signal::ctrl_c().await {
|
||||
tracing::warn!("Failed to listen for shutdown signal: {error}");
|
||||
.with_graceful_shutdown(async move {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
|
||||
.expect("Failed to listen for SIGTERM");
|
||||
tokio::select! { _ = tokio::signal::ctrl_c() => {}, _ = terminate.recv() => {} }
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
let _ = tokio::signal::ctrl_c().await;
|
||||
backup_stop.cancel();
|
||||
})
|
||||
.await
|
||||
.expect("Server error");
|
||||
let _ = tokio::time::timeout(std::time::Duration::from_secs(60), backup_worker).await;
|
||||
shutdown_state.shutdown(std::time::Duration::from_secs(3)).await;
|
||||
}
|
||||
|
||||
|
||||
@@ -30,6 +30,7 @@ pub mod query_result_export;
|
||||
pub mod redis;
|
||||
pub mod redis_pubsub_ws;
|
||||
pub mod saved_sql;
|
||||
pub mod scheduled_backup;
|
||||
pub mod schema;
|
||||
pub mod schema_cache;
|
||||
pub mod schema_diff;
|
||||
|
||||
@@ -1567,6 +1567,7 @@ mod tests {
|
||||
transfer_progress_channels: RwLock::new(HashMap::new()),
|
||||
table_import_channels: RwLock::new(HashMap::new()),
|
||||
sql_file_executions: RwLock::new(HashMap::new()),
|
||||
managed_sql_previews: Default::default(),
|
||||
nacos_imports: RwLock::new(HashMap::new()),
|
||||
login_rate_limit: Mutex::new(LoginRateLimit { fail_count: 0, locked_until: None }),
|
||||
export_files: RwLock::new(HashMap::new()),
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
use crate::{error::AppError, state::WebState};
|
||||
use axum::{
|
||||
body::Body,
|
||||
extract::{Path, State},
|
||||
http::header,
|
||||
response::Response,
|
||||
Json,
|
||||
};
|
||||
use dbx_core::scheduled_backup::{BackupCommand, BackupService};
|
||||
use std::{path::PathBuf, sync::Arc};
|
||||
use tokio_util::io::ReaderStream;
|
||||
|
||||
pub fn service(state: &WebState) -> Result<BackupService, String> {
|
||||
let root = std::env::var_os("DBX_BACKUP_ROOT").map(PathBuf::from).unwrap_or_else(|| state.data_dir.join("backups"));
|
||||
let root = root.canonicalize().map_err(|e| format!("Cannot resolve the server backup root: {e}"))?;
|
||||
Ok(BackupService::new(state.app.clone(), &state.data_dir, Some(root)))
|
||||
}
|
||||
|
||||
pub async fn command(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(command): Json<BackupCommand>,
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
Ok(Json(service(&state)?.command(command).await?))
|
||||
}
|
||||
|
||||
pub async fn download(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Path((id, index)): Path<(String, usize)>,
|
||||
) -> Result<Response, AppError> {
|
||||
let path = service(&state)?.file(&id, index).await?;
|
||||
let name = path.file_name().unwrap_or_default().to_string_lossy();
|
||||
let disposition = super::export_download::attachment_content_disposition(&name);
|
||||
let file = tokio::fs::File::open(&path).await.map_err(|e| AppError::from(e.to_string()))?;
|
||||
Response::builder()
|
||||
.header(header::CONTENT_TYPE, "application/octet-stream")
|
||||
.header(header::CONTENT_DISPOSITION, disposition)
|
||||
.header(header::CACHE_CONTROL, "no-store")
|
||||
.body(Body::from_stream(ReaderStream::new(file)))
|
||||
.map_err(|e| AppError::from(e.to_string()))
|
||||
}
|
||||
|
||||
pub async fn prepare_restore(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Path((id, index)): Path<(String, usize)>,
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
let source = service(&state)?.file(&id, index).await?;
|
||||
Ok(Json(super::sql_file::prepare_backup_preview(&state, source).await?))
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use std::sync::{Arc, Mutex, Weak};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use axum::extract::{Multipart, Path as AxumPath, State};
|
||||
use axum::response::sse::{Event, Sse};
|
||||
@@ -25,6 +26,119 @@ pub const SQL_FILE_UPLOAD_MAX_BYTES: usize = 200 * 1024 * 1024;
|
||||
const SQL_FILE_UPLOAD_MIN_BYTES: usize = 1024 * 1024;
|
||||
const SQL_FILE_UPLOAD_MAX_BYTES_LIMIT: usize = 4096 * 1024 * 1024;
|
||||
|
||||
struct ManagedPreview {
|
||||
created: Instant,
|
||||
directory: Weak<tempfile::TempDir>,
|
||||
pending: Option<Arc<tempfile::TempDir>>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct ManagedSqlPreviews(Mutex<HashMap<String, ManagedPreview>>);
|
||||
|
||||
impl ManagedSqlPreviews {
|
||||
fn insert(&self, directory: tempfile::TempDir) -> String {
|
||||
let directory = Arc::new(directory);
|
||||
let token = directory.path().file_name().unwrap().to_string_lossy().into_owned();
|
||||
self.0.lock().unwrap().insert(
|
||||
token.clone(),
|
||||
ManagedPreview { created: Instant::now(), directory: Arc::downgrade(&directory), pending: Some(directory) },
|
||||
);
|
||||
token
|
||||
}
|
||||
|
||||
fn release(&self, token: &str) {
|
||||
let mut entries = self.0.lock().unwrap();
|
||||
if entries.get(token).is_some_and(|entry| entry.pending.is_some()) {
|
||||
entries.remove(token);
|
||||
}
|
||||
}
|
||||
|
||||
fn claim(&self, paths: &[PathBuf]) -> Result<Vec<Arc<tempfile::TempDir>>, AppError> {
|
||||
let mut entries = self.0.lock().unwrap();
|
||||
let mut tokens = HashSet::new();
|
||||
for path in paths {
|
||||
let Some(parent) = path.parent() else { continue };
|
||||
let Some(token) = parent.file_name().and_then(|name| name.to_str()) else { continue };
|
||||
if !token.starts_with("restore-") {
|
||||
continue;
|
||||
}
|
||||
let valid = entries
|
||||
.get(token)
|
||||
.and_then(|entry| entry.pending.as_ref())
|
||||
.is_some_and(|directory| directory.path().canonicalize().is_ok_and(|directory| directory == parent));
|
||||
if !valid {
|
||||
return Err(AppError::from("Backup preview expired or already consumed".to_string()));
|
||||
}
|
||||
tokens.insert(token.to_owned());
|
||||
}
|
||||
Ok(tokens.into_iter().map(|token| entries.get_mut(&token).unwrap().pending.take().unwrap()).collect())
|
||||
}
|
||||
|
||||
fn cleanup(&self, tmp_dir: &Path, max_age: Duration, managed_only: bool) {
|
||||
let mut entries = self.0.lock().unwrap();
|
||||
for entry in entries.values_mut() {
|
||||
if entry.created.elapsed() >= max_age {
|
||||
entry.pending = None;
|
||||
}
|
||||
}
|
||||
entries.retain(|_, entry| entry.directory.strong_count() > 0);
|
||||
cleanup_sql_file_uploads_except(tmp_dir, max_age, &entries.keys().cloned().collect(), managed_only);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn start_sql_file_cleanup(state: &Arc<WebState>) {
|
||||
let state = Arc::downgrade(state);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Some(state) = state.upgrade() else { break };
|
||||
state.managed_sql_previews.cleanup(&state.data_dir.join("tmp/sql_file"), SQL_FILE_UPLOAD_MAX_AGE, true);
|
||||
drop(state);
|
||||
tokio::time::sleep(Duration::from_secs(60)).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ReleaseSqlFilePreviewRequest {
|
||||
pub cleanup_token: String,
|
||||
}
|
||||
|
||||
pub async fn release_sql_file_preview(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(body): Json<ReleaseSqlFilePreviewRequest>,
|
||||
) -> Json<()> {
|
||||
state.managed_sql_previews.release(&body.cleanup_token);
|
||||
Json(())
|
||||
}
|
||||
|
||||
pub(super) async fn prepare_backup_preview(state: &WebState, source: PathBuf) -> Result<serde_json::Value, AppError> {
|
||||
let tmp_dir = state.data_dir.join("tmp/sql_file");
|
||||
std::fs::create_dir_all(&tmp_dir).map_err(|error| AppError::from(error.to_string()))?;
|
||||
state.managed_sql_previews.cleanup(&tmp_dir, SQL_FILE_UPLOAD_MAX_AGE, true);
|
||||
let limit = sql_file_upload_limit(state).await as u64;
|
||||
let (directory, file_path, size) = tokio::task::spawn_blocking(move || {
|
||||
let directory = tempfile::Builder::new().prefix(&format!("restore-{}-", Uuid::new_v4())).tempdir_in(tmp_dir)?;
|
||||
let file_path =
|
||||
directory.path().join(source.file_name().ok_or_else(|| std::io::Error::other("Invalid backup name"))?);
|
||||
if std::fs::metadata(&source)?.len() > limit {
|
||||
return Err(std::io::Error::other("Backup exceeds the SQL file upload limit"));
|
||||
}
|
||||
let size = std::fs::copy(source, &file_path)?;
|
||||
if size > limit {
|
||||
return Err(std::io::Error::other("Backup exceeds the SQL file upload limit"));
|
||||
}
|
||||
Ok((directory, file_path, size))
|
||||
})
|
||||
.await
|
||||
.map_err(|error| AppError::from(error.to_string()))?
|
||||
.map_err(|error: std::io::Error| AppError::from(error.to_string()))?;
|
||||
let file_name = file_path.file_name().unwrap().to_string_lossy();
|
||||
let mut preview = preview_uploaded_sql_file(&file_path, &file_name, size).await?;
|
||||
preview["cleanupToken"] = state.managed_sql_previews.insert(directory).into();
|
||||
Ok(preview)
|
||||
}
|
||||
|
||||
pub fn sql_file_upload_max_bytes_from_value(value: Option<&str>) -> usize {
|
||||
let mb = value
|
||||
.and_then(|value| value.trim().parse::<usize>().ok())
|
||||
@@ -82,7 +196,7 @@ pub async fn preview_sql_file(
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
let tmp_dir = state.data_dir.join("tmp").join("sql_file");
|
||||
std::fs::create_dir_all(&tmp_dir).map_err(|e| AppError::from(e.to_string()))?;
|
||||
cleanup_expired_sql_file_uploads(&tmp_dir);
|
||||
state.managed_sql_previews.cleanup(&tmp_dir, SQL_FILE_UPLOAD_MAX_AGE, false);
|
||||
|
||||
if let Some(field) = multipart.next_field().await.map_err(|e| AppError::from(e.to_string()))? {
|
||||
let file_name = field.file_name().unwrap_or("upload.sql").to_string();
|
||||
@@ -95,48 +209,56 @@ pub async fn preview_sql_file(
|
||||
let file_path = safe_uploaded_sql_path(&tmp_dir, &file_name)?;
|
||||
std::fs::write(&file_path, &data).map_err(|e| AppError::from(e.to_string()))?;
|
||||
|
||||
if file_name.to_ascii_lowercase().ends_with(".zip") {
|
||||
let extraction_dir = tmp_dir.join(format!("package-{}", Uuid::new_v4()));
|
||||
let package = dbx_core::sql_file_zip_package::extract_sql_file_zip_package(&file_path, &extraction_dir)
|
||||
.map_err(AppError::from)?;
|
||||
let paths = dbx_core::sql_file_zip_package::extracted_sql_zip_paths(&extraction_dir, &package)
|
||||
.into_iter()
|
||||
.map(|path| path.to_string_lossy().to_string())
|
||||
.collect::<Vec<_>>();
|
||||
let content =
|
||||
sql::decode_sql_file_bytes(&std::fs::read(&paths[0]).map_err(|e| AppError::from(e.to_string()))?)
|
||||
.map_err(AppError::from)?;
|
||||
let preview: String = content.chars().take(20_000).collect();
|
||||
let bootstrap_analysis = dbx_core::sql_file_import::mysql_like_sql_file_bootstrap_analysis(&content);
|
||||
return Ok(Json(serde_json::json!({
|
||||
"fileName": file_name,
|
||||
"filePath": file_path.to_string_lossy(),
|
||||
"sizeBytes": data.len(),
|
||||
"preview": preview,
|
||||
"canExecuteWithoutSelectedDatabase": bootstrap_analysis.can_execute_without_selected_database,
|
||||
"establishesDatabaseContext": bootstrap_analysis.establishes_database_context,
|
||||
"packageFilePaths": paths,
|
||||
"packagePartCount": package.part_names.len(),
|
||||
})));
|
||||
}
|
||||
return preview_uploaded_sql_file(&file_path, &file_name, data.len() as u64).await.map(Json);
|
||||
}
|
||||
|
||||
let size_bytes = data.len() as u64;
|
||||
let content =
|
||||
dbx_core::sql_file_import::read_sql_file_preview(&file_path, 1_000_000).await.map_err(AppError::from)?;
|
||||
Err(AppError::from("No file uploaded".to_string()))
|
||||
}
|
||||
|
||||
async fn preview_uploaded_sql_file(
|
||||
file_path: &Path,
|
||||
file_name: &str,
|
||||
size_bytes: u64,
|
||||
) -> Result<serde_json::Value, AppError> {
|
||||
let tmp_dir = file_path.parent().ok_or_else(|| AppError::from("Invalid SQL file path".to_string()))?;
|
||||
|
||||
if file_name.to_ascii_lowercase().ends_with(".zip") {
|
||||
let extraction_dir = tmp_dir.join(format!("package-{}", Uuid::new_v4()));
|
||||
let package = dbx_core::sql_file_zip_package::extract_sql_file_zip_package(file_path, &extraction_dir)
|
||||
.map_err(AppError::from)?;
|
||||
let paths = dbx_core::sql_file_zip_package::extracted_sql_zip_paths(&extraction_dir, &package)
|
||||
.into_iter()
|
||||
.map(|path| path.to_string_lossy().to_string())
|
||||
.collect::<Vec<_>>();
|
||||
let content = sql::decode_sql_file_bytes(&std::fs::read(&paths[0]).map_err(|e| AppError::from(e.to_string()))?)
|
||||
.map_err(AppError::from)?;
|
||||
let preview: String = content.chars().take(20_000).collect();
|
||||
let bootstrap_analysis = dbx_core::sql_file_import::mysql_like_sql_file_bootstrap_analysis(&content);
|
||||
|
||||
return Ok(Json(serde_json::json!({
|
||||
return Ok(serde_json::json!({
|
||||
"fileName": file_name,
|
||||
"filePath": file_path.to_string_lossy(),
|
||||
"sizeBytes": size_bytes,
|
||||
"preview": preview,
|
||||
"canExecuteWithoutSelectedDatabase": bootstrap_analysis.can_execute_without_selected_database,
|
||||
"establishesDatabaseContext": bootstrap_analysis.establishes_database_context,
|
||||
})));
|
||||
"packageFilePaths": paths,
|
||||
"packagePartCount": package.part_names.len(),
|
||||
}));
|
||||
}
|
||||
|
||||
Err(AppError::from("No file uploaded".to_string()))
|
||||
let content =
|
||||
dbx_core::sql_file_import::read_sql_file_preview(file_path, 1_000_000).await.map_err(AppError::from)?;
|
||||
let preview: String = content.chars().take(20_000).collect();
|
||||
let bootstrap_analysis = dbx_core::sql_file_import::mysql_like_sql_file_bootstrap_analysis(&content);
|
||||
|
||||
Ok(serde_json::json!({
|
||||
"fileName": file_name,
|
||||
"filePath": file_path.to_string_lossy(),
|
||||
"sizeBytes": size_bytes,
|
||||
"preview": preview,
|
||||
"canExecuteWithoutSelectedDatabase": bootstrap_analysis.can_execute_without_selected_database,
|
||||
"establishesDatabaseContext": bootstrap_analysis.establishes_database_context,
|
||||
}))
|
||||
}
|
||||
|
||||
pub async fn execute_sql_file(
|
||||
@@ -145,6 +267,13 @@ pub async fn execute_sql_file(
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
let req = body.request;
|
||||
|
||||
let requested_paths = if body.file_paths.is_empty() { vec![req.file_path.clone()] } else { body.file_paths };
|
||||
let file_paths = requested_paths
|
||||
.iter()
|
||||
.map(|file_path| validated_uploaded_sql_path(&state.data_dir, file_path))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let managed_previews = state.managed_sql_previews.claim(&file_paths)?;
|
||||
|
||||
// Fast-fail: reject early if the connection is read-only (individual statements are also checked in do_execute)
|
||||
if let Some(name) = dbx_core::query::connection_readonly_name(&state.app, &req.connection_id).await {
|
||||
return Err(AppError::from(format!(
|
||||
@@ -154,14 +283,9 @@ pub async fn execute_sql_file(
|
||||
}
|
||||
|
||||
let execution_id = req.execution_id.clone();
|
||||
let requested_paths = if body.file_paths.is_empty() { vec![req.file_path.clone()] } else { body.file_paths };
|
||||
if requested_paths.is_empty() {
|
||||
return Err(AppError::from("No SQL files selected".to_string()));
|
||||
}
|
||||
let file_paths = requested_paths
|
||||
.iter()
|
||||
.map(|file_path| validated_uploaded_sql_path(&state.data_dir, file_path))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let token = CancellationToken::new();
|
||||
|
||||
{
|
||||
@@ -180,6 +304,7 @@ pub async fn execute_sql_file(
|
||||
let state_clone = state.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _managed_previews = managed_previews;
|
||||
let started_at = std::time::Instant::now();
|
||||
let mut progress_emitter = SqlFileProgressEmitter::new(|progress| {
|
||||
send_sql_file_progress(&tx, progress);
|
||||
@@ -276,15 +401,22 @@ fn safe_uploaded_sql_path(tmp_dir: &Path, file_name: &str) -> Result<PathBuf, Ap
|
||||
Ok(tmp_dir.join(unique_name))
|
||||
}
|
||||
|
||||
fn cleanup_expired_sql_file_uploads(tmp_dir: &Path) {
|
||||
cleanup_sql_file_uploads_older_than(tmp_dir, SQL_FILE_UPLOAD_MAX_AGE);
|
||||
#[cfg(test)]
|
||||
fn cleanup_sql_file_uploads_older_than(tmp_dir: &Path, max_age: Duration) {
|
||||
cleanup_sql_file_uploads_except(tmp_dir, max_age, &HashSet::new(), false);
|
||||
}
|
||||
|
||||
fn cleanup_sql_file_uploads_older_than(tmp_dir: &Path, max_age: Duration) {
|
||||
fn cleanup_sql_file_uploads_except(tmp_dir: &Path, max_age: Duration, active: &HashSet<String>, managed_only: bool) {
|
||||
let Ok(entries) = std::fs::read_dir(tmp_dir) else {
|
||||
return;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
if managed_only && !entry.file_name().to_string_lossy().starts_with("restore-") {
|
||||
continue;
|
||||
}
|
||||
if active.contains(&entry.file_name().to_string_lossy().into_owned()) {
|
||||
continue;
|
||||
}
|
||||
let path = entry.path();
|
||||
let Ok(metadata) = entry.metadata() else {
|
||||
continue;
|
||||
@@ -364,10 +496,142 @@ pub async fn cancel_sql_file(
|
||||
mod tests {
|
||||
use std::time::Duration;
|
||||
|
||||
use super::{
|
||||
cleanup_sql_file_uploads_older_than, safe_uploaded_sql_path, sql_file_upload_max_bytes_from_value,
|
||||
validated_uploaded_sql_path, SQL_FILE_UPLOAD_MAX_BYTES,
|
||||
};
|
||||
use super::*;
|
||||
|
||||
async fn restore_state(directory: &Path) -> WebState {
|
||||
let storage = dbx_core::storage::Storage::open(&directory.join("dbx.db")).await.unwrap();
|
||||
WebState::for_tests(Arc::new(dbx_core::connection::AppState::new(storage)), directory.to_path_buf())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn restored_preview_uses_a_managed_copy_and_release_cannot_remove_the_backup() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let state = restore_state(directory.path()).await;
|
||||
let source = directory.path().join("backup.sql");
|
||||
std::fs::write(&source, "SELECT 1;").unwrap();
|
||||
let preview = prepare_backup_preview(&state, source.clone()).await.unwrap();
|
||||
assert_eq!(preview["fileName"], "backup.sql");
|
||||
assert_eq!(preview["preview"], "SELECT 1;");
|
||||
let copy = PathBuf::from(preview["filePath"].as_str().unwrap());
|
||||
assert!(copy.starts_with(directory.path().join("tmp/sql_file")));
|
||||
assert_ne!(copy, source);
|
||||
assert!(validated_uploaded_sql_path(directory.path(), copy.to_str().unwrap()).is_ok());
|
||||
state.managed_sql_previews.release(source.to_str().unwrap());
|
||||
assert!(source.exists());
|
||||
assert!(copy.exists());
|
||||
state.managed_sql_previews.release(preview["cleanupToken"].as_str().unwrap());
|
||||
state.managed_sql_previews.release(preview["cleanupToken"].as_str().unwrap());
|
||||
assert!(!copy.exists());
|
||||
assert_eq!(std::fs::read_to_string(source).unwrap(), "SELECT 1;");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failed_backup_copy_and_gzip_preview_leave_no_managed_files() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let state = restore_state(directory.path()).await;
|
||||
assert!(prepare_backup_preview(&state, directory.path().join("missing.sql")).await.is_err());
|
||||
let source = directory.path().join("invalid.sql.gz");
|
||||
std::fs::write(&source, "not gzip").unwrap();
|
||||
assert!(prepare_backup_preview(&state, source.clone()).await.is_err());
|
||||
let oversized = directory.path().join("oversized.sql");
|
||||
std::fs::File::create(&oversized).unwrap().set_len(sql_file_upload_limit(&state).await as u64 + 1).unwrap();
|
||||
assert!(prepare_backup_preview(&state, oversized.clone()).await.is_err());
|
||||
assert!(oversized.exists());
|
||||
assert_eq!(std::fs::read_dir(directory.path().join("tmp/sql_file")).unwrap().count(), 0);
|
||||
assert!(source.exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn consumed_preview_survives_release_and_expiry_until_execution_finishes() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let state = restore_state(directory.path()).await;
|
||||
let source = directory.path().join("backup.sql");
|
||||
std::fs::write(&source, "SELECT 1;").unwrap();
|
||||
let preview = prepare_backup_preview(&state, source.clone()).await.unwrap();
|
||||
let copy = PathBuf::from(preview["filePath"].as_str().unwrap());
|
||||
let paths = vec![copy.canonicalize().unwrap()];
|
||||
let execution = state.managed_sql_previews.claim(&paths).unwrap();
|
||||
assert!(state.managed_sql_previews.claim(&paths).is_err());
|
||||
state.managed_sql_previews.release(preview["cleanupToken"].as_str().unwrap());
|
||||
state.managed_sql_previews.cleanup(&directory.path().join("tmp/sql_file"), Duration::ZERO, true);
|
||||
assert!(copy.exists());
|
||||
drop(execution);
|
||||
assert!(!copy.exists());
|
||||
assert!(source.exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn expired_and_abandoned_previews_are_removed_without_touching_ordinary_uploads() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let state = restore_state(directory.path()).await;
|
||||
let source = directory.path().join("backup.sql");
|
||||
std::fs::write(&source, "SELECT 1;").unwrap();
|
||||
let preview = prepare_backup_preview(&state, source.clone()).await.unwrap();
|
||||
let tmp = directory.path().join("tmp/sql_file");
|
||||
let abandoned = tmp.join("restore-abandoned");
|
||||
std::fs::create_dir(&abandoned).unwrap();
|
||||
std::fs::write(abandoned.join("backup.sql"), "SELECT 2;").unwrap();
|
||||
let upload = tmp.join("upload.sql");
|
||||
std::fs::write(&upload, "SELECT 3;").unwrap();
|
||||
state.managed_sql_previews.cleanup(&tmp, Duration::ZERO, true);
|
||||
assert!(!Path::new(preview["filePath"].as_str().unwrap()).exists());
|
||||
assert!(!abandoned.exists());
|
||||
assert!(upload.exists());
|
||||
assert!(source.exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execution_failure_releases_the_managed_copy() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let state = Arc::new(restore_state(directory.path()).await);
|
||||
let source = directory.path().join("backup.sql");
|
||||
std::fs::write(&source, "SELECT 1;").unwrap();
|
||||
let preview = prepare_backup_preview(&state, source.clone()).await.unwrap();
|
||||
let copy = PathBuf::from(preview["filePath"].as_str().unwrap());
|
||||
let request = serde_json::from_value(serde_json::json!({
|
||||
"executionId": "restore-error", "connectionId": "missing", "database": "app",
|
||||
"filePath": copy, "continueOnError": false
|
||||
}))
|
||||
.unwrap();
|
||||
let response =
|
||||
execute_sql_file(State(state.clone()), Json(SqlFileExecuteWrapper { request, file_paths: vec![] }))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.0["executionId"], "restore-error");
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
while copy.exists() {
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(source.exists());
|
||||
assert!(!state.sql_file_executions.read().await.contains_key("restore-error"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn duplicate_execution_releases_the_new_preview_without_cancelling_the_existing_job() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let state = Arc::new(restore_state(directory.path()).await);
|
||||
let source = directory.path().join("backup.sql");
|
||||
std::fs::write(&source, "SELECT 1;").unwrap();
|
||||
let preview = prepare_backup_preview(&state, source.clone()).await.unwrap();
|
||||
let copy = PathBuf::from(preview["filePath"].as_str().unwrap());
|
||||
let token = CancellationToken::new();
|
||||
state.sql_file_executions.write().await.insert("existing-job".into(), token.clone());
|
||||
let request = serde_json::from_value(serde_json::json!({
|
||||
"executionId": "existing-job", "connectionId": "missing", "database": "app",
|
||||
"filePath": copy, "continueOnError": false
|
||||
}))
|
||||
.unwrap();
|
||||
assert!(execute_sql_file(State(state.clone()), Json(SqlFileExecuteWrapper { request, file_paths: vec![] }))
|
||||
.await
|
||||
.is_err());
|
||||
assert!(!copy.exists());
|
||||
assert!(source.exists());
|
||||
assert!(!token.is_cancelled());
|
||||
assert!(state.sql_file_executions.read().await.contains_key("existing-job"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sql_file_upload_limit_defaults_and_clamps() {
|
||||
|
||||
@@ -39,6 +39,7 @@ pub struct WebState {
|
||||
pub transfer_progress_channels: RwLock<HashMap<String, Arc<TransferProgressChannel>>>,
|
||||
pub table_import_channels: RwLock<HashMap<String, watch::Sender<String>>>,
|
||||
pub sql_file_executions: RwLock<HashMap<String, CancellationToken>>,
|
||||
pub managed_sql_previews: crate::routes::sql_file::ManagedSqlPreviews,
|
||||
pub nacos_imports: RwLock<HashMap<String, NacosImportContext>>,
|
||||
pub login_rate_limit: Mutex<LoginRateLimit>,
|
||||
/// Completed Web export temp files waiting for the browser download.
|
||||
@@ -66,6 +67,7 @@ impl WebState {
|
||||
transfer_progress_channels: RwLock::new(HashMap::new()),
|
||||
table_import_channels: RwLock::new(HashMap::new()),
|
||||
sql_file_executions: RwLock::new(HashMap::new()),
|
||||
managed_sql_previews: Default::default(),
|
||||
nacos_imports: RwLock::new(HashMap::new()),
|
||||
login_rate_limit: Mutex::new(LoginRateLimit { fail_count: 0, locked_until: None }),
|
||||
export_files: RwLock::new(HashMap::new()),
|
||||
|
||||
Reference in new issue
Block a user