mirror of
https://github.com/Portabase/agent.git
synced 2026-09-10 18:16:55 +00:00
Compare commits
20 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 5f92297108 | |||
| d1821f7045 | |||
| 0fe4d50cbd | |||
| 1f29466a28 | |||
| b74aaa0bbc | |||
| 7b5e5b2c78 | |||
| 99f1ef2081 | |||
| 42c3c5945a | |||
| db0f87c2e9 | |||
| f95f0aa73a | |||
| 2177dfd44b | |||
| 0a53eec184 | |||
| 87f33af772 | |||
| b6a120fcbf | |||
| 5298d82576 | |||
| b0da2e40a3 | |||
| 55e20d48e7 | |||
| f7f5f7e141 | |||
| 2170f96a72 | |||
| 298d46ba81 |
+1
-1
@@ -27,5 +27,5 @@ keywords:
|
||||
- self-hosted
|
||||
- portabase
|
||||
license: Apache-2.0
|
||||
version: 1.19.0
|
||||
version: 1.20.1
|
||||
date-released: '2026-02-24'
|
||||
|
||||
Generated
+1
-1
@@ -3503,7 +3503,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "portabase-agent"
|
||||
version = "1.19.0"
|
||||
version = "1.20.1"
|
||||
dependencies = [
|
||||
"aes",
|
||||
"aes-gcm",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "portabase-agent"
|
||||
version = "1.19.0"
|
||||
version = "1.20.1"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
|
||||
+3
-1
@@ -21,9 +21,11 @@ services:
|
||||
LOG: debug
|
||||
TZ: "Europe/Paris"
|
||||
# TMPDIR: /scratch
|
||||
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiZDY4MzU2MTQtNzE2NC00OTQ4LWJlZjMtMTlkZDc5NGQzYmRhIiwibWFzdGVyS2V5QjY0IjoiV2NiM0pQQkVTaFBjRjg5UXZwRVJuamU4NGZmak1kNm4vS2dJOUpjMCtmVT0ifQ=="
|
||||
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiOWMxMzM5NjItMGE5OC00MmRkLTk1NjUtOTA5ZTkyYTI5N2VkIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ=="
|
||||
#CHUNK_SIZE_MB: "1"
|
||||
#POOLING: 1
|
||||
#RETRY_ATTEMPTS: 3
|
||||
#RETRY_BACKOFF_MS: 1000
|
||||
#DATABASES_CONFIG_FILE: "config.toml"
|
||||
extra_hosts:
|
||||
- "localhost:host-gateway"
|
||||
|
||||
@@ -142,6 +142,8 @@ RUN curl -sSL https://dot.net/v1/dotnet-install.sh -o /tmp/dotnet-install.sh \
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
RUN mkdir -p /config
|
||||
|
||||
COPY --from=builder /app/target/release/app /usr/local/bin/app
|
||||
COPY --from=builder /app/version.env /app/version.env
|
||||
COPY entrypoint.sh /entrypoint.sh
|
||||
|
||||
@@ -7,4 +7,6 @@ data:
|
||||
TZ: {{ .Values.env.TZ | quote }}
|
||||
POLLING: {{ .Values.env.POLLING | quote }}
|
||||
APP_ENV: {{ .Values.env.APP_ENV | quote }}
|
||||
LOG: {{ .Values.env.LOG | quote }}
|
||||
LOG: {{ .Values.env.LOG | quote }}
|
||||
RETRY_ATTEMPTS: {{ .Values.env.RETRY_ATTEMPTS | quote }}
|
||||
RETRY_BACKOFF_MS: {{ .Values.env.RETRY_BACKOFF_MS | quote }}
|
||||
|
||||
@@ -11,6 +11,8 @@ env:
|
||||
POLLING: "5"
|
||||
APP_ENV: "production"
|
||||
LOG: "info"
|
||||
RETRY_ATTEMPTS: "3"
|
||||
RETRY_BACKOFF_MS: "1000"
|
||||
|
||||
resources:
|
||||
limits:
|
||||
|
||||
@@ -20,6 +20,7 @@ pub async fn run(cfg: DatabaseConfig) -> anyhow::Result<bool> {
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.kill_on_drop(true)
|
||||
.spawn()?;
|
||||
|
||||
let query = b"SELECT 1 FROM RDB$DATABASE;\nQUIT;\n";
|
||||
|
||||
@@ -12,7 +12,8 @@ pub async fn run(cfg: DatabaseConfig, env: HashMap<String, String>) -> anyhow::R
|
||||
.arg("--user")
|
||||
.arg(cfg.username)
|
||||
.arg("ping")
|
||||
.envs(env);
|
||||
.envs(env)
|
||||
.kill_on_drop(true);
|
||||
|
||||
let result = timeout(Duration::from_secs(10), cmd.output()).await;
|
||||
|
||||
|
||||
@@ -12,7 +12,8 @@ pub async fn run(cfg: DatabaseConfig, env: HashMap<String, String>) -> anyhow::R
|
||||
.arg("--user")
|
||||
.arg(cfg.username)
|
||||
.arg("ping")
|
||||
.envs(env);
|
||||
.envs(env)
|
||||
.kill_on_drop(true);
|
||||
|
||||
let result = timeout(Duration::from_secs(10), cmd.output()).await;
|
||||
|
||||
|
||||
@@ -21,6 +21,8 @@ pub async fn run(cfg: DatabaseConfig) -> Result<bool> {
|
||||
|
||||
cmd.arg("PING");
|
||||
|
||||
cmd.kill_on_drop(true);
|
||||
|
||||
debug!("Command Ping Redis: {:?}", cmd);
|
||||
|
||||
let result = timeout(Duration::from_secs(10), cmd.output()).await;
|
||||
|
||||
@@ -20,6 +20,7 @@ pub async fn run(cfg: DatabaseConfig) -> Result<bool> {
|
||||
}
|
||||
|
||||
cmd.arg("PING");
|
||||
cmd.kill_on_drop(true);
|
||||
|
||||
debug!("Command Ping Valkey: {:?}", cmd);
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
#![allow(dead_code)]
|
||||
|
||||
use crate::services::config::DbType;
|
||||
use std::fmt::{self, Display, Formatter};
|
||||
use std::path::PathBuf;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -20,3 +21,9 @@ pub struct UploadResult {
|
||||
pub remote_file_path: Option<String>,
|
||||
pub total_size: Option<u64>,
|
||||
}
|
||||
|
||||
impl Display for UploadResult {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}", self.error.as_deref().unwrap_or("unknown error"))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ use super::service::BackupService;
|
||||
|
||||
use crate::domain::factory::DatabaseFactory;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
use crate::utils::retry::{RetryPolicy, retry};
|
||||
|
||||
use anyhow::Result;
|
||||
use std::path::Path;
|
||||
@@ -39,7 +40,31 @@ impl BackupService {
|
||||
});
|
||||
}
|
||||
|
||||
match db.backup(tmp_path, Arc::clone(&logger)).await {
|
||||
let policy = RetryPolicy::default();
|
||||
|
||||
let db_ref = &db;
|
||||
let logger_ref = &logger;
|
||||
|
||||
let outcome = retry("Database backup", &logger, &policy, move |attempt| {
|
||||
let dir = tmp_path.join(format!("attempt-{attempt}"));
|
||||
|
||||
async move {
|
||||
if let Err(e) = tokio::fs::create_dir_all(&dir).await {
|
||||
return Err(anyhow::Error::from(e));
|
||||
}
|
||||
|
||||
match db_ref.backup(&dir, Arc::clone(logger_ref)).await {
|
||||
Ok(f) => Ok(f),
|
||||
Err(e) => {
|
||||
let _ = tokio::fs::remove_dir_all(&dir).await;
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
match outcome {
|
||||
Ok(file) => Ok(BackupResult {
|
||||
generated_id,
|
||||
db_type,
|
||||
|
||||
@@ -4,6 +4,7 @@ use super::service::BackupService;
|
||||
use crate::services::api::models::agent::status::DatabaseStorage;
|
||||
use crate::services::storage;
|
||||
use crate::utils::common::BackupMethod;
|
||||
use crate::utils::retry::{RetryPolicy, retry};
|
||||
use anyhow::{Result, bail};
|
||||
use futures::future::join_all;
|
||||
use std::sync::Arc;
|
||||
@@ -97,16 +98,44 @@ impl BackupService {
|
||||
/*
|
||||
STORAGE UPLOAD
|
||||
*/
|
||||
let upload_result = provider
|
||||
.upload(
|
||||
ctx_clone.clone(),
|
||||
result_clone,
|
||||
method,
|
||||
&storage,
|
||||
Some(encrypt),
|
||||
&backup_storage_id,
|
||||
let policy = RetryPolicy::default();
|
||||
|
||||
let attempt_result = if result_clone.backup_file.is_none() {
|
||||
logger_clone.log("error", format!("Missing backup file for storage {}", storage_id));
|
||||
|
||||
Err(UploadResult {
|
||||
storage_id: storage_id.clone(),
|
||||
success: false,
|
||||
error: Some("Missing backup file path".into()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
})
|
||||
} else {
|
||||
retry(
|
||||
&format!("Upload to storage {storage_id}"),
|
||||
&logger_clone,
|
||||
&policy,
|
||||
|_| async {
|
||||
let r = provider
|
||||
.upload(
|
||||
ctx_clone.clone(),
|
||||
result_clone.clone(),
|
||||
method,
|
||||
&storage,
|
||||
Some(encrypt),
|
||||
&backup_storage_id,
|
||||
)
|
||||
.await;
|
||||
|
||||
if r.success { Ok(r) } else { Err(r) }
|
||||
},
|
||||
)
|
||||
.await;
|
||||
.await
|
||||
};
|
||||
|
||||
let upload_result = match attempt_result {
|
||||
Ok(r) | Err(r) => r,
|
||||
};
|
||||
|
||||
let status = if upload_result.success { "success" } else { "failed" };
|
||||
|
||||
@@ -117,8 +146,6 @@ impl BackupService {
|
||||
upload_result.error.as_deref().unwrap_or("unknown error")
|
||||
));
|
||||
|
||||
// `backup_upload_init` opened a per-storage record; close it as "failed"
|
||||
// so the server is notified of the failure (no path/size on this path).
|
||||
if let Err(err) = ctx_clone
|
||||
.api
|
||||
.backup_upload_status(
|
||||
|
||||
+22
-7
@@ -207,16 +207,19 @@ impl ConfigService {
|
||||
ConfigService { ctx }
|
||||
}
|
||||
|
||||
pub fn load(&self, file_path: Option<&str>) -> Result<DatabasesConfig, String> {
|
||||
let path: String = if let Some(fp) = file_path {
|
||||
fp.to_string()
|
||||
} else {
|
||||
format!(
|
||||
fn resolve_path(file_path: Option<&str>) -> String {
|
||||
match file_path {
|
||||
Some(fp) => fp.to_string(),
|
||||
None => format!(
|
||||
"{}/{}",
|
||||
crate::settings::CONFIG.data_path,
|
||||
crate::settings::CONFIG.databases_config_file
|
||||
)
|
||||
};
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn load(&self, file_path: Option<&str>) -> Result<DatabasesConfig, String> {
|
||||
let path = Self::resolve_path(file_path);
|
||||
|
||||
info!("Loading databases config from: {}", path);
|
||||
|
||||
@@ -260,6 +263,18 @@ impl ConfigService {
|
||||
}
|
||||
|
||||
pub fn load_optional(&self, file_path: Option<&str>) -> DatabasesConfig {
|
||||
let path = Self::resolve_path(file_path);
|
||||
|
||||
if !Path::new(&path).exists() {
|
||||
info!(
|
||||
"No local databases config at {}; using dashboard-defined databases only",
|
||||
path
|
||||
);
|
||||
return DatabasesConfig {
|
||||
databases: Vec::new(),
|
||||
};
|
||||
}
|
||||
|
||||
self.load(file_path).unwrap_or_else(|e| {
|
||||
tracing::warn!(
|
||||
"Local databases config unavailable ({e}); continuing with dashboard-defined databases only"
|
||||
|
||||
@@ -46,6 +46,9 @@ pub fn persist_cache(path: &Path, databases: &[DatabaseConfig]) -> std::io::Resu
|
||||
};
|
||||
let json = serde_json::to_string_pretty(&wrapper)
|
||||
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent)?;
|
||||
}
|
||||
let tmp = path.with_extension("json.tmp");
|
||||
std::fs::write(&tmp, json)?;
|
||||
std::fs::rename(&tmp, path)?;
|
||||
|
||||
@@ -8,6 +8,7 @@ use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::utils::retry::{RetryPolicy, retry};
|
||||
|
||||
fn human_size(bytes: u64) -> String {
|
||||
if bytes >= 1024 * 1024 {
|
||||
@@ -26,6 +27,34 @@ impl RestoreService {
|
||||
tmp_path: &Path,
|
||||
logger: Arc<JobLogger>,
|
||||
expected_size: Option<String>,
|
||||
) -> Result<PathBuf> {
|
||||
let policy = RetryPolicy::default();
|
||||
|
||||
let logger_ref = &logger;
|
||||
|
||||
let outcome = retry("Backup download", &logger, &policy, move |_| {
|
||||
let expected = expected_size.clone();
|
||||
|
||||
async move {
|
||||
self.download_once(file_url, tmp_path, Arc::clone(logger_ref), expected)
|
||||
.await
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
if let Err(e) = &outcome {
|
||||
logger.log("error", format!("Download failed: {e}"));
|
||||
}
|
||||
|
||||
outcome
|
||||
}
|
||||
|
||||
pub async fn download_once(
|
||||
&self,
|
||||
file_url: &str,
|
||||
tmp_path: &Path,
|
||||
logger: Arc<JobLogger>,
|
||||
expected_size: Option<String>,
|
||||
) -> Result<PathBuf> {
|
||||
logger.log("info", "Start downloading backup archive".to_string());
|
||||
|
||||
|
||||
+23
-1
@@ -16,6 +16,8 @@ pub struct Settings {
|
||||
pub timezone: String,
|
||||
pub log: String,
|
||||
pub chunk_size: usize, // bytes
|
||||
pub retry_attempts: u32,
|
||||
pub retry_backoff_ms: u64,
|
||||
}
|
||||
|
||||
impl Settings {
|
||||
@@ -49,6 +51,24 @@ impl Settings {
|
||||
|
||||
let chunk_size = chunk_size_mb * 1024 * 1024;
|
||||
|
||||
let retry_attempts = env::var("RETRY_ATTEMPTS")
|
||||
.unwrap_or_else(|_| "3".to_string())
|
||||
.parse::<u32>()
|
||||
.expect("RETRY_ATTEMPTS must be a valid positive integer");
|
||||
|
||||
if retry_attempts < 3 || retry_attempts > 5 {
|
||||
panic!("RETRY_ATTEMPTS must be between 3 and 5");
|
||||
}
|
||||
|
||||
let retry_backoff_ms = env::var("RETRY_BACKOFF_MS")
|
||||
.unwrap_or_else(|_| "1000".to_string())
|
||||
.parse::<u64>()
|
||||
.expect("RETRY_BACKOFF_MS must be a valid positive integer");
|
||||
|
||||
if retry_backoff_ms < 100 || retry_backoff_ms > 30_000 {
|
||||
panic!("RETRY_BACKOFF_MS must be between 100 and 30000 milliseconds");
|
||||
}
|
||||
|
||||
let tz = env::var("TZ").unwrap_or_else(|_| "UTC".to_string());
|
||||
|
||||
Self {
|
||||
@@ -64,7 +84,9 @@ impl Settings {
|
||||
pooling: pooling_seconds,
|
||||
timezone: tz,
|
||||
log: env::var("LOG").unwrap_or_else(|_| "info".into()),
|
||||
chunk_size
|
||||
chunk_size,
|
||||
retry_attempts,
|
||||
retry_backoff_ms,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
use crate::services::backup::BackupService;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::config::{DatabaseConfig, DbType};
|
||||
use crate::tests::init_tracing_for_test;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn sqlite_config(path: &str) -> DatabaseConfig {
|
||||
DatabaseConfig {
|
||||
name: "retry-test".to_string(),
|
||||
database: String::new(),
|
||||
db_type: DbType::Sqlite,
|
||||
username: String::new(),
|
||||
password: String::new(),
|
||||
port: 0,
|
||||
host: String::new(),
|
||||
generated_id: "retry-test-gen".to_string(),
|
||||
path: path.to_string(),
|
||||
max_packet_size: String::new(),
|
||||
volume_name: String::new(),
|
||||
container_name: None,
|
||||
options: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failing_backup_is_retried_and_leaves_no_attempt_directory() {
|
||||
init_tracing_for_test();
|
||||
|
||||
let temp_dir = TempDir::new().unwrap();
|
||||
let tmp_path = temp_dir.path();
|
||||
let logger = Arc::new(JobLogger::new());
|
||||
|
||||
let cfg = sqlite_config("/nonexistent/definitely-not-here.sqlite");
|
||||
|
||||
let result = BackupService::run(cfg, tmp_path, Arc::clone(&logger))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.status, "failed");
|
||||
assert!(result.backup_file.is_none());
|
||||
|
||||
let entries = Arc::try_unwrap(logger).unwrap().into_entries();
|
||||
assert_eq!(
|
||||
entries.iter().filter(|e| e.level == "warn").count(),
|
||||
2,
|
||||
"expected one warn per non-final failed attempt"
|
||||
);
|
||||
assert!(
|
||||
entries
|
||||
.iter()
|
||||
.any(|e| e.level == "error" && e.message.starts_with("Backup failed:")),
|
||||
"expected a single terminal error from the runner"
|
||||
);
|
||||
|
||||
let leftovers: Vec<_> = std::fs::read_dir(tmp_path)
|
||||
.unwrap()
|
||||
.filter_map(|e| e.ok())
|
||||
.filter(|e| e.file_name().to_string_lossy().starts_with("attempt-"))
|
||||
.collect();
|
||||
assert!(
|
||||
leftovers.is_empty(),
|
||||
"failed attempt directories must be cleaned up, found {:?}",
|
||||
leftovers.iter().map(|e| e.file_name()).collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
@@ -14,7 +14,9 @@ use crate::utils::common::BackupMethod;
|
||||
use crate::utils::edge_key::EdgeKey;
|
||||
|
||||
use serde_json::json;
|
||||
use std::io::Write;
|
||||
use std::sync::Arc;
|
||||
use tempfile::NamedTempFile;
|
||||
use wiremock::matchers::{body_partial_json, method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
@@ -93,3 +95,111 @@ async fn failed_upload_reports_failed_status_to_server() {
|
||||
|
||||
// MockServer drop verifies both `.expect(1)` mounts were hit — including the "failed" PATCH.
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failing_upload_is_retried_until_it_succeeds() {
|
||||
init_tracing_for_test();
|
||||
let server = MockServer::start().await;
|
||||
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/agent/agent-1/backup/upload/init"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"message": "ok",
|
||||
"backupStorage": { "id": "bs-1" }
|
||||
})))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/tus/files"))
|
||||
.respond_with(ResponseTemplate::new(500))
|
||||
.up_to_n_times(2)
|
||||
.with_priority(1)
|
||||
.expect(2)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/tus/files"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(201)
|
||||
.insert_header("Location", format!("{}/tus/files/upload-1", server.uri()).as_str()),
|
||||
)
|
||||
.with_priority(2)
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
Mock::given(method("PATCH"))
|
||||
.and(path("/tus/files/upload-1"))
|
||||
.respond_with(ResponseTemplate::new(204))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
Mock::given(method("PATCH"))
|
||||
.and(path("/agent/agent-1/backup/upload/status"))
|
||||
.and(body_partial_json(json!({ "status": "success" })))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"message": "ok",
|
||||
"backupStorage": { "id": "bs-1" }
|
||||
})))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let mut backup_file = NamedTempFile::new().unwrap();
|
||||
backup_file.write_all(b"portabase-retry-test-payload").unwrap();
|
||||
backup_file.flush().unwrap();
|
||||
|
||||
let ctx = Context {
|
||||
edge_key: EdgeKey {
|
||||
server_url: server.uri(),
|
||||
agent_id: "agent-1".to_string(),
|
||||
master_key_b64: String::new(),
|
||||
},
|
||||
api: ApiClient::new(server.uri()),
|
||||
};
|
||||
|
||||
let service = BackupService::new(Arc::new(ctx));
|
||||
|
||||
let result = BackupResult {
|
||||
generated_id: "gen-1".to_string(),
|
||||
db_type: DbType::Postgresql,
|
||||
status: "success".to_string(),
|
||||
backup_file: Some(backup_file.path().to_path_buf()),
|
||||
code: None,
|
||||
};
|
||||
|
||||
let storage: DatabaseStorage = serde_json::from_value(json!({
|
||||
"id": "storage-1",
|
||||
"provider": "local",
|
||||
"config": {}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let backup_id = "backup-1".to_string();
|
||||
let logger = Arc::new(JobLogger::new());
|
||||
|
||||
let results = service
|
||||
.upload(
|
||||
result,
|
||||
BackupMethod::Manual,
|
||||
vec![storage],
|
||||
false,
|
||||
&backup_id,
|
||||
Arc::clone(&logger),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(results.len(), 1);
|
||||
assert!(results[0].success);
|
||||
|
||||
let entries = Arc::try_unwrap(logger).unwrap().into_entries();
|
||||
assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2);
|
||||
assert!(
|
||||
entries.iter().any(|e| e.message
|
||||
== "Upload to storage storage-1 succeeded on attempt 3/3")
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
mod api_models_tests;
|
||||
mod backup_runner_tests;
|
||||
mod backup_uploader_tests;
|
||||
mod config_tests;
|
||||
mod dashboard_config_tests;
|
||||
mod restore_downloader_tests;
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
use crate::core::context::Context;
|
||||
use crate::services::api::ApiClient;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::restore::RestoreService;
|
||||
use crate::tests::init_tracing_for_test;
|
||||
use crate::utils::edge_key::EdgeKey;
|
||||
|
||||
use std::sync::Arc;
|
||||
use tempfile::TempDir;
|
||||
use wiremock::matchers::{method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failing_download_is_retried_until_it_succeeds() {
|
||||
init_tracing_for_test();
|
||||
let server = MockServer::start().await;
|
||||
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/backups/archive.tar.gz"))
|
||||
.respond_with(ResponseTemplate::new(503))
|
||||
.up_to_n_times(2)
|
||||
.with_priority(1)
|
||||
.expect(2)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/backups/archive.tar.gz"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_bytes(b"portabase-archive".to_vec()))
|
||||
.with_priority(2)
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let ctx = Context {
|
||||
edge_key: EdgeKey {
|
||||
server_url: server.uri(),
|
||||
agent_id: "agent-1".to_string(),
|
||||
master_key_b64: String::new(),
|
||||
},
|
||||
api: ApiClient::new(server.uri()),
|
||||
};
|
||||
|
||||
let service = RestoreService::new(Arc::new(ctx));
|
||||
|
||||
let temp_dir = TempDir::new().unwrap();
|
||||
let logger = Arc::new(JobLogger::new());
|
||||
let url = format!("{}/backups/archive.tar.gz", server.uri());
|
||||
|
||||
let downloaded = service
|
||||
.download_backup(&url, temp_dir.path(), Arc::clone(&logger), None)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(std::fs::read(&downloaded).unwrap(), b"portabase-archive");
|
||||
|
||||
let entries = Arc::try_unwrap(logger).unwrap().into_entries();
|
||||
assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2);
|
||||
assert!(
|
||||
entries
|
||||
.iter()
|
||||
.any(|e| e.message == "Backup download succeeded on attempt 3/3")
|
||||
);
|
||||
}
|
||||
@@ -4,4 +4,5 @@ mod deserializer;
|
||||
mod edge_key_tests;
|
||||
mod file_tests;
|
||||
mod normalize_cron_tests;
|
||||
mod retry_tests;
|
||||
mod stream_tests;
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::tests::init_tracing_for_test;
|
||||
use crate::utils::retry::{RetryPolicy, retry};
|
||||
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
fn fast_policy(attempts: u32) -> RetryPolicy {
|
||||
RetryPolicy {
|
||||
attempts,
|
||||
base_backoff: Duration::from_millis(1),
|
||||
max_backoff: Duration::from_millis(4),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn delay_grows_with_the_attempt_number() {
|
||||
let policy = RetryPolicy {
|
||||
attempts: 5,
|
||||
base_backoff: Duration::from_millis(100),
|
||||
max_backoff: Duration::from_secs(30),
|
||||
};
|
||||
|
||||
assert!(policy.delay(1) >= Duration::from_millis(50));
|
||||
assert!(policy.delay(1) <= Duration::from_millis(100));
|
||||
assert!(policy.delay(2) >= Duration::from_millis(100));
|
||||
assert!(policy.delay(2) <= Duration::from_millis(200));
|
||||
assert!(policy.delay(3) >= Duration::from_millis(200));
|
||||
assert!(policy.delay(3) <= Duration::from_millis(400));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn delay_never_exceeds_max_backoff() {
|
||||
let policy = RetryPolicy {
|
||||
attempts: 5,
|
||||
base_backoff: Duration::from_millis(1000),
|
||||
max_backoff: Duration::from_millis(2000),
|
||||
};
|
||||
|
||||
for attempt in 1..=5 {
|
||||
assert!(policy.delay(attempt) <= Duration::from_millis(2000));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn first_attempt_success_logs_nothing() {
|
||||
init_tracing_for_test();
|
||||
let logger = JobLogger::new();
|
||||
|
||||
let result: Result<u32, anyhow::Error> =
|
||||
retry("Test op", &logger, &fast_policy(3), |_| async { Ok(7) }).await;
|
||||
|
||||
assert_eq!(result.unwrap(), 7);
|
||||
assert!(logger.into_entries().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retries_until_success_and_logs_each_attempt() {
|
||||
init_tracing_for_test();
|
||||
let logger = JobLogger::new();
|
||||
let calls = AtomicU32::new(0);
|
||||
|
||||
let result: Result<u32, anyhow::Error> =
|
||||
retry("Test op", &logger, &fast_policy(3), |_| async {
|
||||
let n = calls.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
if n < 3 {
|
||||
Err(anyhow::anyhow!("boom {n}"))
|
||||
} else {
|
||||
Ok(n)
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
assert_eq!(result.unwrap(), 3);
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 3);
|
||||
|
||||
let entries = logger.into_entries();
|
||||
|
||||
let warns: Vec<_> = entries.iter().filter(|e| e.level == "warn").collect();
|
||||
assert_eq!(warns.len(), 2);
|
||||
assert!(warns[0].message.starts_with("Test op attempt 1/3 failed: boom 1"));
|
||||
assert!(warns[1].message.starts_with("Test op attempt 2/3 failed: boom 2"));
|
||||
|
||||
let infos: Vec<_> = entries.iter().filter(|e| e.level == "info").collect();
|
||||
assert_eq!(infos.len(), 1);
|
||||
assert_eq!(infos[0].message, "Test op succeeded on attempt 3/3");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn exhausts_attempts_and_logs_no_terminal_error() {
|
||||
init_tracing_for_test();
|
||||
let logger = JobLogger::new();
|
||||
let calls = AtomicU32::new(0);
|
||||
|
||||
let result: Result<(), anyhow::Error> =
|
||||
retry("Test op", &logger, &fast_policy(3), |_| async {
|
||||
calls.fetch_add(1, Ordering::SeqCst);
|
||||
Err(anyhow::anyhow!("always"))
|
||||
})
|
||||
.await;
|
||||
|
||||
assert!(result.is_err());
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 3);
|
||||
|
||||
let entries = logger.into_entries();
|
||||
assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2);
|
||||
assert_eq!(entries.iter().filter(|e| e.level == "error").count(), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn closure_receives_the_attempt_number() {
|
||||
init_tracing_for_test();
|
||||
let logger = JobLogger::new();
|
||||
let seen = Mutex::new(Vec::new());
|
||||
let seen_ref = &seen;
|
||||
|
||||
let result: Result<(), anyhow::Error> =
|
||||
retry("Test op", &logger, &fast_policy(3), move |attempt| async move {
|
||||
seen_ref.lock().unwrap().push(attempt);
|
||||
Err(anyhow::anyhow!("nope"))
|
||||
})
|
||||
.await;
|
||||
|
||||
assert!(result.is_err());
|
||||
assert_eq!(*seen.lock().unwrap(), vec![1, 2, 3]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_defaults_are_within_the_documented_range() {
|
||||
let policy = RetryPolicy::default();
|
||||
assert!(policy.attempts >= 3 && policy.attempts <= 5);
|
||||
assert!(policy.base_backoff >= Duration::from_millis(100));
|
||||
assert!(policy.base_backoff <= Duration::from_millis(30_000));
|
||||
}
|
||||
@@ -6,6 +6,7 @@ pub mod file;
|
||||
pub mod locks;
|
||||
pub mod logging;
|
||||
pub mod redis_client;
|
||||
pub mod retry;
|
||||
pub mod stream;
|
||||
pub mod task_manager;
|
||||
pub mod text;
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::settings::CONFIG;
|
||||
use rand::Rng;
|
||||
use std::fmt::Display;
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
pub struct RetryPolicy {
|
||||
pub attempts: u32,
|
||||
pub base_backoff: Duration,
|
||||
pub max_backoff: Duration,
|
||||
}
|
||||
|
||||
impl Default for RetryPolicy {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
attempts: CONFIG.retry_attempts,
|
||||
base_backoff: Duration::from_millis(CONFIG.retry_backoff_ms),
|
||||
max_backoff: Duration::from_secs(30),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl RetryPolicy {
|
||||
pub(crate) fn delay(&self, attempt: u32) -> Duration {
|
||||
let exp = self.base_backoff.saturating_mul(1u32 << (attempt - 1).min(16));
|
||||
let capped = exp.min(self.max_backoff);
|
||||
let half = capped / 2;
|
||||
let jitter = rand::rng().random_range(0..=half.as_millis() as u64);
|
||||
|
||||
half + Duration::from_millis(jitter)
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn retry<T, E, F, Fut>(
|
||||
op: &str,
|
||||
logger: &JobLogger,
|
||||
policy: &RetryPolicy,
|
||||
mut f: F,
|
||||
) -> Result<T, E>
|
||||
where
|
||||
F: FnMut(u32) -> Fut,
|
||||
Fut: Future<Output = Result<T, E>> + Send,
|
||||
T: Send,
|
||||
E: Display + Send,
|
||||
{
|
||||
let total = policy.attempts;
|
||||
let mut attempt = 1;
|
||||
|
||||
loop {
|
||||
match f(attempt).await {
|
||||
Ok(v) => {
|
||||
if attempt > 1 {
|
||||
logger.log("info", format!("{op} succeeded on attempt {attempt}/{total}"));
|
||||
}
|
||||
return Ok(v);
|
||||
}
|
||||
Err(e) if attempt < total => {
|
||||
let delay = policy.delay(attempt);
|
||||
logger.log(
|
||||
"warn",
|
||||
format!(
|
||||
"{op} attempt {attempt}/{total} failed: {e} - retrying in {}ms",
|
||||
delay.as_millis()
|
||||
),
|
||||
);
|
||||
tokio::time::sleep(delay).await;
|
||||
attempt += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -79,7 +79,12 @@ pub async fn execute_task(
|
||||
let ctx = Arc::new(Context::new());
|
||||
let config_service = ConfigService::new(ctx.clone());
|
||||
let backup_service = BackupService::new(ctx.clone());
|
||||
let config = config_service.load(None).unwrap();
|
||||
|
||||
let local = config_service.load_optional(None);
|
||||
let cache_path = std::path::PathBuf::from(&crate::settings::CONFIG.data_path)
|
||||
.join("dashboard_databases.json");
|
||||
let dashboard = crate::services::dashboard_config::load_cache(&cache_path);
|
||||
let config = crate::services::dashboard_config::merge(&local.databases, &dashboard);
|
||||
|
||||
let metadata_obj = metadata
|
||||
.into_iter()
|
||||
|
||||
Reference in New Issue
Block a user