Compare commits

...

27 Commits

Author SHA1 Message Date
charles-gauthereau eef182ad5f fix 2026-07-21 23:57:15 +02:00
github-actions[bot] f050fa35e0 chore: release 1.17.1 2026-07-21 21:29:30 +00:00
Charles GTE 90d55f7630 fix: backup folder name (#86)
* fix: support configurable backup file prefix (#83)

* fix: backup folder name

* fix: ghcr publish agent image

---------

Co-authored-by: Antonin Jousson <18756890+Antoninj@users.noreply.github.com>
Co-authored-by: charles-gauthereau <charles.gauthereau@soluce-technologies.com>
2026-07-21 23:27:10 +02:00
github-actions[bot] 82ffb71496 chore: release 1.17.0 2026-07-11 10:08:21 +00:00
Charles GTE 20b3b58024 Merge pull request #80 from Portabase/fix/storage-encryption
fix/storage-encryption
2026-07-11 12:05:59 +02:00
charles-gauthereau 6cddc73565 Merge branch 'main' into fix/storage-encryption
# Conflicts:
#	docker-compose.yml
2026-07-10 18:49:01 +02:00
github-actions[bot] fd3b0076c0 chore: release 1.16.3 2026-07-09 09:11:23 +00:00
Charles GTE 334ab51125 fix: cron (#79)
* fix: cron log

---------

Co-authored-by: charles-gauthereau <charles.gauthereau@soluce-technologies.com>
2026-07-09 11:03:55 +02:00
github-actions[bot] 4df3fe2d0c chore: release 1.16.2 2026-07-09 07:38:49 +00:00
Charles GTE ec7715ca26 fix: cron (#78)
* fix: docker-compose.yml

* fix: cron

---------

Co-authored-by: charles-gauthereau <charles.gauthereau@soluce-technologies.com>
2026-07-09 09:36:53 +02:00
charles-gauthereau 32d4e22196 fix: status.rs 2026-07-08 22:24:50 +02:00
charles-gauthereau f277c5485f feat: decrypt encrypted storages in status ping 2026-07-08 20:52:17 +02:00
charles-gauthereau 585e0bea72 feat: accept encrypted storages markers in DatabaseStatus 2026-07-08 20:47:20 +02:00
charles-gauthereau ee97c6056a feat: add decrypt_json_gcm for encrypted status storages 2026-07-08 20:43:01 +02:00
charles-gauthereau b37a4f2180 fix: docker-compose.yml 2026-07-07 18:43:30 +02:00
github-actions[bot] 1d15f40662 chore: release 1.16.1 2026-07-04 10:26:50 +00:00
Charles GTE 324a2ea3f2 fix: security (#75) 2026-07-04 12:24:37 +02:00
github-actions[bot] 9ec92af6c7 chore: release 1.16.0 2026-07-03 06:16:42 +00:00
Charles GTE 1dafdca2a0 Merge pull request #74 from Portabase/feat/docker-volume-backup-restore
feat: docker-volume-backup-restore
2026-07-03 08:14:30 +02:00
charles-gauthereau d12817a960 fix: codecov.yml 2026-07-02 22:42:02 +02:00
charles-gauthereau 608b82a254 fix: test docker_volume.rs 2026-07-02 22:08:55 +02:00
charles-gauthereau bbecca92f9 fix: databases.json 2026-07-02 21:51:17 +02:00
charles-gauthereau b2b2733c4d feat: docker-volume provider
refactor: move build_tar to utils::compress

refactor: move choose_restore_path to utils::common

docs: remove docker-volume README section

fix: keep archive path for multi-file non-docker-volume restores

docs: trim docker-volume README section to essentials

refactor: run docker-volume backup/restore inside spawn_blocking like other providers

docs: enable docker socket, add volume example and security notes

feat: sweep orphaned ephemeral helper containers on startup

feat: docker-volume clean-replace restore via upload_to_container

fix: serialize env-var access in docker-volume tests to avoid setenv/getenv UB

feat: docker-volume backup via download_from_container

feat: docker-volume ping via inspect_volume

fix: return extraction dir for multi-file restore archives

feat: gzip already-tar inputs directly instead of double-wrapping

feat: docker helper (client, self-image, container lifecycle, sweep)

feat: docker-volume provider
2026-07-02 21:45:11 +02:00
github-actions[bot] 2c257c5a58 chore: release 1.15.0 2026-06-29 17:14:03 +00:00
Charles GTE c24c0d7058 Merge pull request #73 from Portabase/fix/postgres-ownership
fix: postgres-ownership
2026-06-29 19:12:01 +02:00
github-actions[bot] 39a77b18a4 chore: release 1.14.1 2026-06-29 13:22:10 +00:00
Charles GTE e62167182e Merge pull request #72 from Portabase/fix/postgres-cluster
fix: postgres cluster dump command
2026-06-29 15:19:54 +02:00
54 changed files with 1360 additions and 70 deletions
+1 -1
View File
@@ -36,7 +36,7 @@ jobs:
agent-test bash -c "
mkdir -p /app/coverage &&
rm -rf /app/target/* /app/coverage/* &&
cargo test --verbose &&
cargo test --verbose -- --test-threads=2 &&
sync
"
+72
View File
@@ -0,0 +1,72 @@
name: GHCR Publish
on:
workflow_call:
inputs:
version:
required: true
type: string
ref:
required: true
type: string
add_latest:
required: false
type: boolean
default: false
dockerfile:
required: false
type: string
default: "./docker/Dockerfile"
target:
required: false
type: string
default: "prod"
permissions:
contents: read
packages: write
jobs:
publish:
name: Build and push to GHCR
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
ref: ${{ inputs.ref }}
fetch-depth: 0
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to GHCR
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract Docker metadata
id: meta
uses: docker/metadata-action@v5
with:
images: ghcr.io/${{ github.repository_owner }}/agent
tags: |
type=semver,pattern={{version}},value=${{ inputs.version }}
type=semver,pattern={{major}}.{{minor}},value=${{ inputs.version }}
type=semver,pattern={{major}},value=${{ inputs.version }}
type=raw,value=latest,enable=${{ inputs.add_latest }}
- name: Build and push image
uses: docker/build-push-action@v6
with:
context: .
file: ${{ inputs.dockerfile }}
platforms: linux/amd64,linux/arm64
push: true
tags: ${{ steps.meta.outputs.tags }}
target: ${{ inputs.target }}
provenance: false
cache-from: type=gha,scope=ghcr-build
cache-to: type=gha,mode=max,scope=ghcr-build,ignore-error=true
+9
View File
@@ -100,6 +100,15 @@ jobs:
DOCKER_USERNAME: ${{ secrets.DOCKER_USERNAME }}
DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
publish-docker-ghcr:
needs: create-release
if: ${{ needs.create-release.result == 'success' }}
uses: ./.github/workflows/ghcr.yml
with:
version: ${{ needs.create-release.outputs.version }}
ref: ${{ needs.create-release.outputs.version }}
add_latest: true
publish-helm:
needs: create-release
if: ${{ needs.create-release.result == 'success' }}
+1 -1
View File
@@ -27,5 +27,5 @@ keywords:
- self-hosted
- portabase
license: Apache-2.0
version: 1.14.0
version: 1.17.1
date-released: '2026-02-24'
Generated
+2 -1
View File
@@ -3503,7 +3503,7 @@ dependencies = [
[[package]]
name = "portabase-agent"
version = "1.14.0"
version = "1.17.1"
dependencies = [
"aes",
"aes-gcm",
@@ -3516,6 +3516,7 @@ dependencies = [
"azure_core",
"azure_storage_blob",
"base64 0.22.1",
"bollard",
"bytes",
"chrono",
"cron",
+3 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "portabase-agent"
version = "1.14.0"
version = "1.17.1"
edition = "2024"
[dependencies]
@@ -35,7 +35,7 @@ rand = "0.9.2"
bytes = "1.11.0"
async-stream = "0.3.6"
uuid = { version = "1.20.0", features = ["v4"] }
tokio-util = { version = "0.7.18", features = ["compat"] }
tokio-util = { version = "0.7.18", features = ["compat", "io"] }
tiberius = { version = "0.12", default-features = false, features = ["rustls", "chrono"] }
aws-config = "1.8.13"
aws-sdk-s3 = { version = "1.122.0", features = ["behavior-version-latest"] }
@@ -57,6 +57,7 @@ testcontainers = "0.27.1"
testcontainers-modules = { version = "0.15.0", features = ["postgres", "redis", "valkey", "mysql", "mariadb", "mongo"] }
postgres = "0.19.12"
url = "2.5.8"
bollard = "0.20.0"
[dev-dependencies]
tokio = { version = "1", features = ["full"] }
+7
View File
@@ -124,6 +124,13 @@
"port": 1433,
"host": "db-mssql",
"generated_id": "16706125-ff7e-4c97-8c83-0adeff214682"
},
{
"name": "Test database 14 - Docker Volume",
"type": "docker-volume",
"volume_name": "databases_sqlite-data",
"generated_id": "16706126-ff7e-4c97-8c83-0adeff214690",
"container_name": "db-sqlite"
}
]
}
+4 -2
View File
@@ -11,15 +11,17 @@ services:
- cargo-git:/usr/local/cargo/git
- ./databases.json:/config/config.json
#- ./databases.toml:/config/config.toml
#- /var/run/docker.sock:/var/run/docker.sock
- /var/run/docker.sock:/var/run/docker.sock
# - cargo-target:/app/target
- databases_sqlite-data:/sqlite-data/workspace/data
- ./scripts/sqlite/test-db:/sqlite-data-2/workspace/data
# - /bigdisk:/scratch
environment:
APP_ENV: development
LOG: debug
TZ: "Europe/Paris"
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNWE2YjcxMDgtMGJhYS00Yjg1LTgwMmMtNTNjNjJiMDAzZDgzIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ=="
# TMPDIR: /scratch
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNTljYzRjYTUtOTAyNy00ZThiLTk1NDktMjAzOTI3ZDVjNmUyIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ=="
#CHUNK_SIZE_MB: "1"
#POOLING: 1
#DATABASES_CONFIG_FILE: "config.toml"
+68
View File
@@ -0,0 +1,68 @@
use crate::domain::docker_volume::docker::{
client, create_helper, remove_helper, resolve_helper_image, start_container, stop_container,
};
use crate::services::backup::logger::JobLogger;
use crate::services::config::DatabaseConfig;
use anyhow::{Context, Result};
use bollard::query_parameters::DownloadFromContainerOptions;
use futures_util::StreamExt;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;
use tokio::fs::File;
use tokio::io::AsyncWriteExt;
pub async fn run(cfg: DatabaseConfig, backup_dir: PathBuf, logger: Arc<JobLogger>) -> Result<PathBuf> {
tokio::task::spawn_blocking(move || -> Result<PathBuf> {
futures::executor::block_on(async move {
logger.log("info", format!("Starting docker-volume backup for {}", cfg.name));
let docker = client()?;
let image = resolve_helper_image(&docker).await?;
logger.log("debug", format!("Helper image: {image}"));
if let Some(name) = &cfg.container_name {
logger.log("info", format!("Stopping container {name} for consistent backup"));
stop_container(&docker, name).await?;
}
let result = async {
let helper = create_helper(&docker, &image, &cfg.volume_name, &cfg.generated_id, true, None).await?;
let file_path = backup_dir.join(format!("{}.tar", cfg.generated_id));
let start = Instant::now();
let dl_opts = DownloadFromContainerOptions { path: "/vol".to_string() };
let mut stream = docker.download_from_container(&helper.id, Some(dl_opts));
let mut out = File::create(&file_path)
.await
.with_context(|| format!("Failed to create backup file {}", file_path.display()))?;
let mut bytes_written: u64 = 0;
while let Some(chunk) = stream.next().await {
let chunk = chunk.context("Error streaming volume archive from Docker")?;
bytes_written += chunk.len() as u64;
out.write_all(&chunk).await?;
}
out.flush().await?;
let duration_ms = start.elapsed().as_millis() as f64;
logger.log_command("docker download_from_container", None, Some(0), Some(duration_ms));
logger.log("info", format!("Volume backup wrote {bytes_written} bytes to {}", file_path.display()));
remove_helper(&docker, &helper.id).await;
anyhow::Ok(file_path)
}
.await;
if let Some(name) = &cfg.container_name {
if let Err(e) = start_container(&docker, name).await {
logger.log("error", format!("Failed to restart container {name}: {e}"));
}
}
result
})
})
.await?
}
+45
View File
@@ -0,0 +1,45 @@
use anyhow::Result;
use async_trait::async_trait;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use super::{backup, ping, restore};
use crate::domain::factory::Database;
use crate::services::backup::logger::JobLogger;
use crate::services::config::DatabaseConfig;
use crate::utils::locks::{DbOpLock, FileLock};
pub struct DockerVolumeDatabase {
cfg: DatabaseConfig,
}
impl DockerVolumeDatabase {
pub fn new(cfg: DatabaseConfig) -> Self {
Self { cfg }
}
}
#[async_trait]
impl Database for DockerVolumeDatabase {
fn file_extension(&self) -> &'static str {
".tar"
}
async fn ping(&self) -> Result<bool> {
ping::run(self.cfg.clone()).await
}
async fn backup(&self, dir: &Path, logger: Arc<JobLogger>) -> Result<PathBuf> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
let res = backup::run(self.cfg.clone(), dir.to_path_buf(), logger).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
async fn restore(&self, file: &Path, logger: Arc<JobLogger>) -> Result<()> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
let res = restore::run(self.cfg.clone(), file.to_path_buf(), logger).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
}
+169
View File
@@ -0,0 +1,169 @@
#![allow(dead_code)]
use anyhow::{Context, Result};
use bollard::Docker;
use bollard::models::{ContainerCreateBody, HostConfig};
use bollard::query_parameters::{
CreateContainerOptions, InspectContainerOptions, ListContainersOptions,
RemoveContainerOptions, StartContainerOptions, StopContainerOptions,
};
use std::collections::HashMap;
use tracing::{info, warn};
use uuid::Uuid;
pub const EPHEMERAL_LABEL: &str = "io.portabase.ephemeral";
const HELPER_MOUNT: &str = "/vol";
pub fn client() -> Result<Docker> {
Docker::connect_with_unix_defaults().context("Failed to connect to Docker daemon socket")
}
pub fn parse_container_id(mountinfo: &str, cgroup: &str) -> Option<String> {
for src in [mountinfo, cgroup] {
for line in src.lines() {
for marker in ["/containers/", "/docker/"] {
if let Some(idx) = line.find(marker) {
let rest = &line[idx + marker.len()..];
let id: String = rest.chars().take_while(|c| c.is_ascii_hexdigit()).collect();
if id.len() >= 64 {
return Some(id[..64].to_string());
}
}
}
}
}
None
}
pub async fn resolve_helper_image(docker: &Docker) -> Result<String> {
if let Ok(img) = std::env::var("PORTABASE_HELPER_IMAGE") {
if !img.trim().is_empty() {
return Ok(img);
}
}
let mountinfo = std::fs::read_to_string("/proc/self/mountinfo").unwrap_or_default();
let cgroup = std::fs::read_to_string("/proc/self/cgroup").unwrap_or_default();
let id = parse_container_id(&mountinfo, &cgroup).context(
"Could not determine own container id; set PORTABASE_HELPER_IMAGE to a locally-present image",
)?;
let info = docker
.inspect_container(&id, None::<InspectContainerOptions>)
.await
.with_context(|| format!("Failed to inspect self container {id}"))?;
info.image
.context("Self container inspection returned no image reference")
}
pub struct Helper {
pub id: String,
}
pub async fn create_helper(
docker: &Docker,
image: &str,
volume_name: &str,
generated_id: &str,
read_only: bool,
cmd: Option<Vec<String>>,
) -> Result<Helper> {
let bind = format!(
"{volume_name}:{HELPER_MOUNT}{}",
if read_only { ":ro" } else { "" }
);
let mut labels = HashMap::new();
labels.insert(EPHEMERAL_LABEL.to_string(), "true".to_string());
labels.insert("com.docker.compose.project".to_string(), String::new());
labels.insert("com.docker.compose.service".to_string(), String::new());
labels.insert("com.docker.compose.oneoff".to_string(), String::new());
let name = format!(
"portabase-vol-{generated_id}-{}",
&Uuid::new_v4().to_string()[..8]
);
let body = ContainerCreateBody {
image: Some(image.to_string()),
cmd,
labels: Some(labels),
host_config: Some(HostConfig {
binds: Some(vec![bind]),
auto_remove: Some(false),
..Default::default()
}),
..Default::default()
};
let opts = CreateContainerOptions {
name: Some(name),
..Default::default()
};
let res = docker
.create_container(Some(opts), body)
.await
.with_context(|| format!("Failed to create helper container for volume {volume_name}"))?;
Ok(Helper { id: res.id })
}
pub async fn remove_helper(docker: &Docker, id: &str) {
let stop_opts = StopContainerOptions {
t: Some(2),
..Default::default()
};
let _ = docker.stop_container(id, Some(stop_opts)).await;
if let Ok(info) = docker
.inspect_container(id, None::<InspectContainerOptions>)
.await
{
let name = info.name.unwrap_or_default();
let name = name.trim_start_matches('/');
let code = info.state.and_then(|s| s.exit_code).unwrap_or_default();
info!("Helper container {name} exited with code {code}");
}
let opts = RemoveContainerOptions {
force: true,
..Default::default()
};
if let Err(e) = docker.remove_container(id, Some(opts)).await {
warn!("Failed to remove helper container {id}: {e}");
}
}
pub async fn stop_container(docker: &Docker, name: &str) -> Result<()> {
docker
.stop_container(name, None::<StopContainerOptions>)
.await
.with_context(|| format!("Failed to stop container {name}"))
}
pub async fn start_container(docker: &Docker, name: &str) -> Result<()> {
docker
.start_container(name, None::<StartContainerOptions>)
.await
.with_context(|| format!("Failed to start container {name}"))
}
pub async fn sweep_ephemeral(docker: &Docker) -> Result<usize> {
let mut filters = HashMap::new();
filters.insert("label".to_string(), vec![format!("{EPHEMERAL_LABEL}=true")]);
let opts = ListContainersOptions {
all: true,
filters: Some(filters),
..Default::default()
};
let list = docker.list_containers(Some(opts)).await?;
let mut removed = 0;
for c in list {
if let Some(id) = c.id {
remove_helper(docker, &id).await;
removed += 1;
}
}
Ok(removed)
}
+5
View File
@@ -0,0 +1,5 @@
pub mod backup;
pub mod database;
pub mod docker;
pub mod ping;
pub mod restore;
+12
View File
@@ -0,0 +1,12 @@
use crate::domain::docker_volume::docker::client;
use crate::services::config::DatabaseConfig;
use anyhow::Result;
pub async fn run(cfg: DatabaseConfig) -> Result<bool> {
let docker = client()?;
match docker.inspect_volume(&cfg.volume_name).await {
Ok(_) => Ok(true),
Err(bollard::errors::Error::DockerResponseServerError { status_code: 404, .. }) => Ok(false),
Err(e) => Err(e.into()),
}
}
+102
View File
@@ -0,0 +1,102 @@
use crate::domain::docker_volume::docker::{
client, create_helper, remove_helper, resolve_helper_image, start_container, stop_container,
};
use crate::services::backup::logger::JobLogger;
use crate::services::config::DatabaseConfig;
use anyhow::{Context, Result};
use bollard::exec::StartExecResults;
use bollard::models::ExecConfig;
use bollard::query_parameters::UploadToContainerOptions;
use futures_util::StreamExt;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;
pub async fn run(cfg: DatabaseConfig, archive: PathBuf, logger: Arc<JobLogger>) -> Result<()> {
tokio::task::spawn_blocking(move || -> Result<()> {
futures::executor::block_on(async move {
logger.log("info", format!("Starting docker-volume restore for {}", cfg.name));
let docker = client()?;
let image = resolve_helper_image(&docker).await?;
logger.log("debug", format!("Restore archive: {}", archive.display()));
if let Some(name) = &cfg.container_name {
logger.log("info", format!("Stopping container {name} for restore"));
stop_container(&docker, name).await?;
}
let result = async {
let helper = create_helper(
&docker,
&image,
&cfg.volume_name,
&cfg.generated_id,
false,
Some(vec![
"sh".into(),
"-c".into(),
"trap 'exit 0' TERM; sleep 2147483647 & wait".into(),
]),
)
.await?;
start_container(&docker, &helper.id).await?;
let exec = docker
.create_exec(
&helper.id,
ExecConfig {
cmd: Some(vec![
"sh".to_string(),
"-c".to_string(),
"rm -rf /vol/* /vol/.[!.]* 2>/dev/null || true".to_string(),
]),
attach_stdout: Some(true),
attach_stderr: Some(true),
..Default::default()
},
)
.await
.context("Failed to create wipe exec")?;
if let StartExecResults::Attached { mut output, .. } =
docker.start_exec(&exec.id, None).await.context("Failed to run wipe exec")?
{
while output.next().await.is_some() {}
}
let start = Instant::now();
let file = tokio::fs::File::open(&archive)
.await
.with_context(|| format!("Failed to open {}", archive.display()))?;
let stream = tokio_util::io::ReaderStream::new(file);
let up_opts = UploadToContainerOptions { path: "/".to_string(), ..Default::default() };
docker
.upload_to_container(&helper.id, Some(up_opts), bollard::body_try_stream(stream))
.await
.context("Failed to upload volume archive")?;
let duration_ms = start.elapsed().as_millis() as f64;
logger.log_command("docker upload_to_container", None, Some(0), Some(duration_ms));
remove_helper(&docker, &helper.id).await;
logger.log("info", format!("Volume restore completed for {}", cfg.name));
anyhow::Ok(())
}
.await;
if let Some(name) = &cfg.container_name {
if let Err(e) = start_container(&docker, name).await {
logger.log("error", format!("Failed to restart container {name}: {e}"));
}
}
result
})
})
.await?
}
+3
View File
@@ -1,3 +1,4 @@
use crate::domain::docker_volume::database::DockerVolumeDatabase;
use crate::domain::mongodb::database::MongoDatabase;
use crate::domain::mysql::database::MySQLDatabase;
use crate::domain::postgres::cluster::database::PostgresClusterDatabase;
@@ -41,6 +42,7 @@ impl DatabaseFactory {
DbType::Valkey => Arc::new(ValkeyDatabase::new(cfg)),
DbType::Firebird => Arc::new(FirebirdDatabase::new(cfg)),
DbType::Mssql => Arc::new(MssqlDatabase::new(cfg)),
DbType::DockerVolume => Arc::new(DockerVolumeDatabase::new(cfg)),
}
}
@@ -59,6 +61,7 @@ impl DatabaseFactory {
DbType::Valkey => Arc::new(ValkeyDatabase::new(cfg)),
DbType::Firebird => Arc::new(FirebirdDatabase::new(cfg)),
DbType::Mssql => Arc::new(MssqlDatabase::new(cfg)),
DbType::DockerVolume => Arc::new(DockerVolumeDatabase::new(cfg)),
}
}
}
+1
View File
@@ -1,3 +1,4 @@
pub mod docker_volume;
pub mod factory;
mod mongodb;
pub mod mysql;
+10
View File
@@ -22,6 +22,16 @@ async fn main() {
eprintln!("Failed to clean locks on startup: {:?}", e);
}
// Best-effort cleanup of ephemeral helper containers orphaned by a crash.
match crate::domain::docker_volume::docker::client() {
Ok(docker) => match crate::domain::docker_volume::docker::sweep_ephemeral(&docker).await {
Ok(n) if n > 0 => tracing::info!("Removed {n} orphaned ephemeral helper container(s)"),
Ok(_) => {}
Err(e) => tracing::warn!("Ephemeral helper sweep failed: {e}"),
},
Err(e) => tracing::debug!("Docker socket unavailable, skipping helper sweep: {e}"),
}
tokio::join!(ping_server(), async {
let conn = redis_client::redis_connection().await;
scheduler::scheduler_loop(conn).await;
+7
View File
@@ -24,6 +24,8 @@ pub struct DatabaseStorage {
#[serde(deserialize_with = "deserialize_snake_case")]
pub config: Value,
pub provider: String,
#[serde(default, rename = "folderName")]
pub folder_name: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -31,7 +33,12 @@ pub struct DatabaseStatus {
pub dbms: String,
#[serde(rename = "generatedId")]
pub generated_id: String,
#[serde(default)]
pub storages: Vec<DatabaseStorage>,
#[serde(default)]
pub storages_encrypted: Option<bool>,
#[serde(default)]
pub storages_ciphertext: Option<String>,
pub encrypt: bool,
pub data: DatabaseData,
}
+1
View File
@@ -104,6 +104,7 @@ impl BackupService {
method,
&storage,
Some(encrypt),
&backup_storage_id,
)
.await;
+20 -3
View File
@@ -26,6 +26,8 @@ pub enum DbType {
Valkey,
Firebird,
Mssql,
#[serde(rename = "docker-volume")]
DockerVolume,
}
impl DbType {
@@ -41,6 +43,7 @@ impl DbType {
DbType::Valkey => "valkey",
DbType::Firebird => "firebird",
DbType::Mssql => "mssql",
DbType::DockerVolume => "docker-volume",
}
}
}
@@ -59,6 +62,8 @@ pub struct DatabaseConfig {
pub generated_id: String,
pub path: String,
pub max_packet_size: String,
pub volume_name: String,
pub container_name: Option<String>,
pub options: HashMap<String, serde_json::Value>,
}
@@ -82,6 +87,8 @@ pub struct InputDatabaseConfig {
pub generated_id: String,
pub path: Option<String>,
pub max_packet_size: Option<String>,
pub volume_name: Option<String>,
pub container_name: Option<String>,
pub options: Option<HashMap<String, serde_json::Value>>,
}
@@ -202,7 +209,7 @@ impl ConfigService {
| DbType::Firebird
| DbType::Valkey
| DbType::Mssql => required(&db.host, &db.name, "host")?,
DbType::Sqlite => optional(&db.host),
DbType::Sqlite | DbType::DockerVolume => optional(&db.host),
};
let port = match db.db_type {
@@ -215,11 +222,13 @@ impl ConfigService {
| DbType::Firebird
| DbType::Valkey
| DbType::Mssql => required(&db.port, &db.name, "port")?,
DbType::Sqlite => db.port.unwrap_or(0),
DbType::Sqlite | DbType::DockerVolume => db.port.unwrap_or(0),
};
let database_name = match db.db_type {
DbType::Sqlite | DbType::Redis | DbType::Valkey => optional(&db.database),
DbType::Sqlite | DbType::Redis | DbType::Valkey | DbType::DockerVolume => {
optional(&db.database)
}
DbType::PostgresqlCluster => db
.database
.clone()
@@ -239,6 +248,12 @@ impl ConfigService {
_ => String::new(),
};
let volume_name = match db.db_type {
DbType::DockerVolume => required(&db.volume_name, &db.name, "volume_name")?,
_ => optional(&db.volume_name),
};
let container_name = db.container_name.clone();
databases.push(DatabaseConfig {
name: db.name,
database: database_name,
@@ -250,6 +265,8 @@ impl ConfigService {
generated_id: db.generated_id,
path: path_val,
max_packet_size,
volume_name,
container_name,
options: db.options.unwrap_or_default(),
});
}
+13 -9
View File
@@ -1,18 +1,19 @@
use super::service::RestoreService;
use crate::utils::compress::decompress_large_tar_gz;
use crate::utils::file::decrypt_file_stream_gcm;
use anyhow::Result;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::services::backup::logger::JobLogger;
use crate::services::config::DbType;
use crate::utils::common::choose_restore_path;
impl RestoreService {
pub async fn prepare_archive(
&self,
downloaded_file: PathBuf,
tmp_path: &Path,
db_type: &DbType,
logger: Arc<JobLogger>
) -> Result<PathBuf> {
logger.log("info", "Start preparing backup archive".to_string());
@@ -59,6 +60,13 @@ impl RestoreService {
archive = decrypted;
}
if matches!(db_type, DbType::DockerVolume) {
let raw_tar = tmp_path.join("volume.tar");
crate::utils::compress::gunzip_to_file(archive.as_path(), &raw_tar).await?;
logger.log("info", format!("Docker volume archive gunzipped to {}", raw_tar.display()));
return Ok(raw_tar);
}
logger.log("info", format!("Decompressing archive {}", archive.display()));
let files = match decompress_large_tar_gz(archive.as_path(), tmp_path).await {
@@ -76,12 +84,8 @@ impl RestoreService {
logger.log("info", format!("Archive prepared, {} file(s) extracted", files.len()));
if files.len() == 1 {
logger.log("debug", format!("Using single extracted file: {}", files[0].display()));
Ok(files[0].clone())
} else {
logger.log("debug", format!("Multiple files extracted, using archive root: {}", archive.display()));
Ok(archive)
}
let chosen = choose_restore_path(&files, tmp_path, &archive);
logger.log("debug", format!("Restore source resolved to: {}", chosen.display()));
Ok(chosen)
}
}
+3 -1
View File
@@ -27,7 +27,9 @@ impl RestoreService {
.download_backup(&file_url, tmp_path, Arc::clone(&logger), expected_size)
.await?;
let backup_file = self.prepare_archive(downloaded, tmp_path, Arc::clone(&logger)).await?;
let backup_file = self
.prepare_archive(downloaded, tmp_path, &cfg.db_type, Arc::clone(&logger))
.await?;
let result = self.run_restore(cfg, backup_file, Arc::clone(&logger)).await?;
+25 -8
View File
@@ -1,16 +1,18 @@
#![allow(dead_code)]
use crate::core::context::Context;
use crate::domain::factory::DatabaseFactory;
use crate::services::api::endpoints::status::DatabasePayload;
use crate::services::api::models::agent::status::DatabaseStorage;
use crate::services::api::models::agent::status::PingResult;
use crate::services::config::DatabaseConfig;
use crate::settings::CONFIG;
use crate::utils::file::decrypt_json_gcm;
use futures_util::future::try_join_all;
use reqwest::Client;
use std::error::Error;
use std::sync::Arc;
use futures_util::future::try_join_all;
use tracing::info;
use crate::domain::factory::DatabaseFactory;
pub struct StatusService {
ctx: Arc<Context>,
@@ -28,12 +30,12 @@ impl StatusService {
pub async fn ping(&self, databases: &[DatabaseConfig]) -> Result<PingResult, Box<dyn Error>> {
let edge_key = &self.ctx.edge_key;
let databases_payload: Vec<DatabasePayload> = try_join_all(
databases.into_iter().map(|db| async move {
let databases_payload: Vec<DatabasePayload> =
try_join_all(databases.into_iter().map(|db| async move {
let db_engine = DatabaseFactory::create_for_backup(db.clone()).await;
let reachable = db_engine.ping().await?;
info!("Ping {} => {:?}",db.name, reachable);
info!("Ping {} => {:?}", db.name, reachable);
Ok::<DatabasePayload, anyhow::Error>(DatabasePayload {
name: &db.name,
@@ -41,16 +43,31 @@ impl StatusService {
generated_id: &db.generated_id,
ping_status: reachable,
})
})
).await?;
}))
.await?;
let version_str = CONFIG.app_version.as_str();
let result = self
let mut result = self
.ctx
.api
.agent_status(&edge_key.agent_id, &version_str, databases_payload)
.await?
.unwrap();
for db in result.databases.iter_mut() {
if db.storages_encrypted == Some(true) {
let ciphertext = db
.storages_ciphertext
.as_deref()
.ok_or("storages_encrypted set but storages_ciphertext missing")?;
let plaintext = decrypt_json_gcm(ciphertext, &edge_key.master_key_b64)
.map_err(|e| format!("Failed to decrypt storages: {e}"))?;
db.storages = serde_json::from_slice::<Vec<DatabaseStorage>>(&plaintext)
.map_err(|e| format!("Failed to parse decrypted storages: {e}"))?;
}
}
Ok(result)
}
}
+1
View File
@@ -22,6 +22,7 @@ pub trait StorageProvider: Send + Sync {
method: BackupMethod,
config: &DatabaseStorage,
encrypt: Option<bool>,
backup_storage_id: &str,
) -> UploadResult;
}
@@ -26,6 +26,7 @@ impl StorageProvider for AzureBlobProvider {
_method: BackupMethod,
storage: &DatabaseStorage,
encrypt: Option<bool>,
_backup_storage_id: &str,
) -> UploadResult {
let Some(file_path) = result.backup_file else {
return UploadResult {
@@ -94,7 +95,7 @@ impl StorageProvider for AzureBlobProvider {
};
let file_name = full_file_name(encrypt);
let remote_file_path = full_file_path(&file_name);
let remote_file_path = full_file_path(&file_name, storage.folder_name.as_deref());
info!(
"Starting block upload to azure blob {}/{}",
config.container_name, remote_file_path
@@ -28,6 +28,7 @@ impl StorageProvider for GoogleCloudStorageProvider {
_method: BackupMethod,
storage: &DatabaseStorage,
encrypt: Option<bool>,
_backup_storage_id: &str,
) -> UploadResult {
let Some(file_path) = result.backup_file else {
return UploadResult {
@@ -84,7 +85,7 @@ impl StorageProvider for GoogleCloudStorageProvider {
let file_name = full_file_name(encrypt);
info!("Uploading file {}", file_name);
let remote_file_path = full_file_path(&file_name);
let remote_file_path = full_file_path(&file_name, storage.folder_name.as_deref());
let client = match build_client(&config).await {
Ok(c) => c,
@@ -26,6 +26,7 @@ impl StorageProvider for GoogleDriveProvider {
_method: BackupMethod,
storage: &DatabaseStorage,
encrypt: Option<bool>,
_backup_storage_id: &str,
) -> UploadResult {
let Some(file_path) = result.backup_file else {
return UploadResult {
@@ -84,7 +85,7 @@ impl StorageProvider for GoogleDriveProvider {
info!("Uploading file {}", file_name);
let remote_file_path = full_file_path(&file_name);
let remote_file_path = full_file_path(&file_name, storage.folder_name.as_deref());
match upload_stream_to_google_drive(
&config,
+6 -1
View File
@@ -23,6 +23,7 @@ impl StorageProvider for LocalProvider {
method: BackupMethod,
storage: &DatabaseStorage,
encrypt: Option<bool>,
backup_storage_id: &str,
) -> UploadResult {
let Some(file_path) = result.backup_file else {
return UploadResult {
@@ -37,7 +38,7 @@ impl StorageProvider for LocalProvider {
let encrypt = encrypt.unwrap_or(false);
let file_name = full_file_name(encrypt);
let remote_file_path = full_file_path(&file_name);
let remote_file_path = full_file_path(&file_name, storage.folder_name.as_deref());
let total_size = match fs::metadata(&file_path).await {
Ok(meta) => meta.len(),
@@ -82,6 +83,10 @@ impl StorageProvider for LocalProvider {
"X-Generated-Id",
HeaderValue::from_str(&result.generated_id).unwrap(),
);
extra_headers.insert(
"X-Backup-Storage-Id",
HeaderValue::from_str(backup_storage_id).unwrap(),
);
extra_headers.insert("X-Status", HeaderValue::from_str(&result.status).unwrap());
extra_headers.insert(
"X-Method",
+2 -1
View File
@@ -34,6 +34,7 @@ impl StorageProvider for S3Provider {
_method: BackupMethod,
storage: &DatabaseStorage,
encrypt: Option<bool>,
_backup_storage_id: &str,
) -> UploadResult {
let Some(file_path) = result.backup_file else {
return UploadResult {
@@ -147,7 +148,7 @@ impl StorageProvider for S3Provider {
info!("Uploading file {}", file_name);
let bucket = &config.bucket_name;
let remote_file_path = full_file_path(&file_name);
let remote_file_path = full_file_path(&file_name, storage.folder_name.as_deref());
info!("S3 key {:}", remote_file_path);
info!(
"Starting multipart upload to s3://{}/{}",
+2
View File
@@ -14,6 +14,8 @@ fn cluster_config() -> DatabaseConfig {
generated_id: "40875631-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
path: String::new(),
max_packet_size: String::new(),
volume_name: String::new(),
container_name: None,
options: std::collections::HashMap::new(),
}
}
+2
View File
@@ -36,6 +36,8 @@ async fn start_cluster(user: &str) -> (ContainerAsync<Postgres>, DatabaseConfig)
generated_id: "40875631-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
(container, config)
+279
View File
@@ -0,0 +1,279 @@
use crate::domain::docker_volume::docker::parse_container_id;
static ENV_GUARD: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
#[test]
fn parse_container_id_from_mountinfo_line() {
let id = "a".repeat(64);
let mountinfo = format!(
"1234 1000 0:50 / /etc/hostname rw shared:1 - ext4 /var/lib/docker/containers/{id}/hostname rw"
);
assert_eq!(parse_container_id(&mountinfo, ""), Some(id));
}
#[test]
fn parse_container_id_from_cgroup_v1() {
let id = "b".repeat(64);
let cgroup = format!("12:memory:/docker/{id}\n11:cpu:/docker/{id}\n");
assert_eq!(parse_container_id("", &cgroup), Some(id));
}
#[test]
fn parse_container_id_none_on_cgroup_v2() {
assert_eq!(parse_container_id("", "0::/\n"), None);
}
#[tokio::test]
async fn docker_volume_ping_true_for_existing_volume() {
use crate::domain::docker_volume::docker::client;
use bollard::models::VolumeCreateRequest;
use bollard::query_parameters::RemoveVolumeOptions;
let docker = client().expect("docker daemon required for this test");
let vol = format!("portabase-test-{}", uuid::Uuid::new_v4());
docker
.create_volume(VolumeCreateRequest { name: Some(vol.clone()), ..Default::default() })
.await
.unwrap();
let cfg = volume_config(&vol);
let reachable = crate::domain::docker_volume::ping::run(cfg).await.unwrap();
assert!(reachable);
let missing = volume_config("portabase-does-not-exist-xyz");
assert!(!crate::domain::docker_volume::ping::run(missing).await.unwrap());
docker.remove_volume(&vol, None::<RemoveVolumeOptions>).await.ok();
}
fn volume_config(volume_name: &str) -> crate::services::config::DatabaseConfig {
use crate::services::config::{DatabaseConfig, DbType};
DatabaseConfig {
name: "vol-test".to_string(),
database: "".to_string(),
db_type: DbType::DockerVolume,
username: "".to_string(),
password: "".to_string(),
port: 0,
host: "".to_string(),
generated_id: uuid::Uuid::new_v4().to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: volume_name.to_string(),
container_name: None,
options: std::collections::HashMap::new(),
}
}
async fn ensure_image(docker: &bollard::Docker, image: &str) {
use bollard::query_parameters::CreateImageOptionsBuilder;
use futures_util::StreamExt;
let (name, tag) = image.split_once(':').unwrap_or((image, "latest"));
let opts = CreateImageOptionsBuilder::default()
.from_image(name)
.tag(tag)
.build();
let mut stream = docker.create_image(Some(opts), None, None);
while let Some(item) = stream.next().await {
item.unwrap();
}
}
async fn seed_volume(docker: &bollard::Docker, volume: &str, filename: &str, content: &str) {
use bollard::models::{ContainerCreateBody, HostConfig};
use bollard::query_parameters::{
CreateContainerOptions, RemoveContainerOptions, StartContainerOptions,
WaitContainerOptions,
};
use futures_util::StreamExt;
ensure_image(docker, "busybox").await;
let body = ContainerCreateBody {
image: Some("busybox".to_string()),
cmd: Some(vec![
"sh".into(),
"-c".into(),
format!("printf '%s' '{content}' > /vol/{filename}"),
]),
host_config: Some(HostConfig {
binds: Some(vec![format!("{volume}:/vol")]),
..Default::default()
}),
..Default::default()
};
let created = docker
.create_container(None::<CreateContainerOptions>, body)
.await
.unwrap();
docker.start_container(&created.id, None::<StartContainerOptions>).await.unwrap();
let mut wait = docker.wait_container(&created.id, None::<WaitContainerOptions>);
while wait.next().await.is_some() {}
docker
.remove_container(&created.id, Some(RemoveContainerOptions { force: true, ..Default::default() }))
.await
.ok();
}
#[tokio::test]
async fn docker_volume_backup_captures_files() {
use crate::domain::docker_volume::docker::client;
use bollard::models::VolumeCreateRequest;
use bollard::query_parameters::RemoveVolumeOptions;
let _env_guard = ENV_GUARD.lock().await;
unsafe { std::env::set_var("PORTABASE_HELPER_IMAGE", "busybox"); }
let docker = client().expect("docker daemon required");
let vol = format!("portabase-test-{}", uuid::Uuid::new_v4());
docker
.create_volume(VolumeCreateRequest { name: Some(vol.clone()), ..Default::default() })
.await
.unwrap();
seed_volume(&docker, &vol, "hello.txt", "backup-me").await;
let tmp = tempfile::TempDir::new().unwrap();
let cfg = volume_config(&vol);
let logger = std::sync::Arc::new(crate::services::backup::logger::JobLogger::new());
let tar = crate::domain::docker_volume::backup::run(cfg, tmp.path().to_path_buf(), logger)
.await
.unwrap();
assert!(tar.is_file());
let names = tar_entry_names(&tar).await;
assert!(names.iter().any(|n| n.ends_with("hello.txt")), "entries: {names:?}");
docker.remove_volume(&vol, None::<RemoveVolumeOptions>).await.ok();
}
async fn tar_entry_names(tar_path: &std::path::Path) -> Vec<String> {
use tokio_stream::StreamExt;
let f = tokio::fs::File::open(tar_path).await.unwrap();
let mut archive = tokio_tar::Archive::new(f);
let mut names = Vec::new();
let mut entries = archive.entries().unwrap();
while let Some(e) = entries.next().await {
let e = e.unwrap();
names.push(e.path().unwrap().to_string_lossy().to_string());
}
names
}
#[tokio::test]
async fn docker_volume_restore_is_clean_replace() {
use crate::domain::docker_volume::docker::client;
use bollard::models::VolumeCreateRequest;
use bollard::query_parameters::RemoveVolumeOptions;
let _env_guard = ENV_GUARD.lock().await;
unsafe { std::env::set_var("PORTABASE_HELPER_IMAGE", "busybox"); }
let docker = client().expect("docker daemon required");
let vol = format!("portabase-test-{}", uuid::Uuid::new_v4());
docker
.create_volume(VolumeCreateRequest { name: Some(vol.clone()), ..Default::default() })
.await
.unwrap();
seed_volume(&docker, &vol, "keeper.txt", "original").await;
let tmp = tempfile::TempDir::new().unwrap();
let logger = std::sync::Arc::new(crate::services::backup::logger::JobLogger::new());
let tar = crate::domain::docker_volume::backup::run(
volume_config(&vol),
tmp.path().to_path_buf(),
logger.clone(),
)
.await
.unwrap();
seed_volume(&docker, &vol, "drift.txt", "added-later").await;
// Restore uploads the raw Docker tar directly.
crate::domain::docker_volume::restore::run(volume_config(&vol), tar.clone(), logger)
.await
.unwrap();
let listing = list_volume(&docker, &vol).await;
assert!(listing.contains("keeper.txt"), "listing: {listing}");
assert!(!listing.contains("drift.txt"), "clean-replace failed, listing: {listing}");
docker.remove_volume(&vol, None::<RemoveVolumeOptions>).await.ok();
}
async fn list_volume(docker: &bollard::Docker, volume: &str) -> String {
use bollard::models::{ContainerCreateBody, HostConfig};
use bollard::query_parameters::{
CreateContainerOptions, LogsOptions, RemoveContainerOptions, StartContainerOptions,
WaitContainerOptions,
};
use tokio_stream::StreamExt;
ensure_image(docker, "busybox").await;
let body = ContainerCreateBody {
image: Some("busybox".to_string()),
cmd: Some(vec!["sh".into(), "-c".into(), "ls -A /vol".into()]),
host_config: Some(HostConfig {
binds: Some(vec![format!("{volume}:/vol")]),
..Default::default()
}),
..Default::default()
};
let created = docker.create_container(None::<CreateContainerOptions>, body).await.unwrap();
docker.start_container(&created.id, None::<StartContainerOptions>).await.unwrap();
let mut wait = docker.wait_container(&created.id, None::<WaitContainerOptions>);
while wait.next().await.is_some() {}
let mut logs = docker.logs(
&created.id,
Some(LogsOptions { stdout: true, stderr: false, ..Default::default() }),
);
let mut out = String::new();
while let Some(chunk) = logs.next().await {
if let Ok(l) = chunk {
out.push_str(&l.to_string());
}
}
docker
.remove_container(&created.id, Some(RemoveContainerOptions { force: true, ..Default::default() }))
.await
.ok();
out
}
#[tokio::test]
async fn sweep_removes_labeled_helpers() {
use crate::domain::docker_volume::docker::{client, create_helper, sweep_ephemeral, EPHEMERAL_LABEL};
use bollard::models::VolumeCreateRequest;
use bollard::query_parameters::{ListContainersOptions, RemoveVolumeOptions};
use std::collections::HashMap;
let _env_guard = ENV_GUARD.lock().await;
unsafe { std::env::set_var("PORTABASE_HELPER_IMAGE", "busybox"); }
let docker = client().expect("docker daemon required");
let vol = format!("portabase-test-{}", uuid::Uuid::new_v4());
docker
.create_volume(VolumeCreateRequest { name: Some(vol.clone()), ..Default::default() })
.await
.unwrap();
ensure_image(&docker, "busybox").await;
let helper = create_helper(&docker, "busybox", &vol, "sweep-test", true, None).await.unwrap();
let removed = sweep_ephemeral(&docker).await.unwrap();
assert!(removed >= 1);
let mut filters = HashMap::new();
filters.insert("label".to_string(), vec![format!("{EPHEMERAL_LABEL}=true")]);
let remaining = docker
.list_containers(Some(ListContainersOptions { all: true, filters: Some(filters), ..Default::default() }))
.await
.unwrap();
assert!(remaining.iter().all(|c| c.id.as_deref() != Some(helper.id.as_str())));
docker.remove_volume(&vol, None::<RemoveVolumeOptions>).await.ok();
}
+2
View File
@@ -40,6 +40,8 @@ async fn create_config() -> (ContainerAsync<GenericImage>, DatabaseConfig) {
generated_id: "3c445eb4-c2c6-4bde-a423-ee1385dcf6d2".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+2
View File
@@ -32,6 +32,8 @@ async fn create_config() -> (ContainerAsync<Mariadb>, DatabaseConfig) {
generated_id: "3c4b4eb4-c2c6-4bde-a423-ee1385dcf6d2".to_string(),
path: "".to_string(),
max_packet_size: "512M".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+1
View File
@@ -7,3 +7,4 @@ mod redis;
mod valkey;
mod firebird;
mod mssql;
mod docker_volume;
+2
View File
@@ -30,6 +30,8 @@ async fn create_config() -> (ContainerAsync<Mongo>, DatabaseConfig) {
generated_id: "96d30a9f-ff4b-47c9-aaab-f3147bb34f16".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+2
View File
@@ -55,6 +55,8 @@ fn make_config(host: String, port: u16, database: &str, generated_id: &str) -> D
generated_id: generated_id.to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
}
}
+2
View File
@@ -32,6 +32,8 @@ async fn create_config() -> (ContainerAsync<Mysql>, DatabaseConfig) {
generated_id: "0f1bb8f2-35a0-4c91-8098-e36873d3ce31".to_string(),
path: "".to_string(),
max_packet_size: "512M".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+4
View File
@@ -39,6 +39,8 @@ async fn create_config() -> (ContainerAsync<Postgres>, DatabaseConfig) {
generated_id: "40875631-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
@@ -157,6 +159,8 @@ async fn postgres_password_with_slash_test() {
generated_id: "5a1f0e3c-9b8a-4a8e-9b1b-0a1c2d3e4f5a".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+2
View File
@@ -29,6 +29,8 @@ async fn create_config() -> (ContainerAsync<Redis>, DatabaseConfig) {
generated_id: "40875631-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+2
View File
@@ -28,6 +28,8 @@ async fn create_config() -> (ContainerAsync<Valkey>, DatabaseConfig) {
generated_id: "40875485-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
path: "".to_string(),
max_packet_size: "".to_string(),
volume_name: "".to_string(),
container_name: None,
options: std::collections::HashMap::new(),
};
+39 -1
View File
@@ -2,7 +2,7 @@ use serde_json::json;
use crate::services::api::models::agent::backup::{BackupResponse, BackupUploadResponse};
use crate::services::api::models::agent::restore::ResultRestoreResponse;
use crate::services::api::models::agent::status::PingResult;
use crate::services::api::models::agent::status::{DatabaseStatus, PingResult};
#[test]
fn backup_response_deserializes_nested_backup_id() {
@@ -110,3 +110,41 @@ fn ping_result_deserializes_and_normalizes_storage_config_keys() {
assert!(result.databases[0].data.restore.file.is_none());
assert!(result.databases[0].data.restore.meta_file.is_none());
}
#[test]
fn database_status_legacy_plaintext_storages() {
let status: DatabaseStatus = serde_json::from_value(json!({
"dbms": "postgres",
"generatedId": "gen-1",
"storages": [ { "id": "s1", "config": { "bucket": "b" }, "provider": "s3" } ],
"encrypt": true,
"data": {
"backup": { "action": false, "cron": null },
"restore": { "action": false, "file": null, "metaFile": null, "size": null }
}
})).unwrap();
assert_eq!(status.storages.len(), 1);
assert_eq!(status.storages_encrypted, None);
assert!(status.storages_ciphertext.is_none());
}
#[test]
fn database_status_encrypted_envelope() {
let status: DatabaseStatus = serde_json::from_value(json!({
"dbms": "postgres",
"generatedId": "gen-1",
"storages": [],
"storages_encrypted": true,
"storages_ciphertext": "AQIDBA==",
"encrypt": true,
"data": {
"backup": { "action": true, "cron": null },
"restore": { "action": false, "file": null, "metaFile": null, "size": null }
}
})).unwrap();
assert!(status.storages.is_empty());
assert_eq!(status.storages_encrypted, Some(true));
assert_eq!(status.storages_ciphertext.as_deref(), Some("AQIDBA=="));
}
+65
View File
@@ -199,3 +199,68 @@ fn keep_ownership_extraction_logic() {
let keep4 = opts4.get("keep_ownership").and_then(|v| v.as_bool()).unwrap_or(false);
assert!(!keep4, "should strip when value is not bool");
}
#[test]
fn parses_docker_volume_type() {
let file = write_json(
r#"{
"databases": [
{
"name": "uploads",
"type": "docker-volume",
"volume_name": "myapp_uploads",
"generated_id": "16678159-ff7e-4c97-8c83-0adeff214681",
"container_name": "myapp"
}
]
}"#,
);
let service = ConfigService::new(test_context());
let cfg = service.load(Some(file.path().to_str().unwrap())).unwrap();
assert_eq!(cfg.databases[0].db_type.as_str(), "docker-volume");
assert_eq!(cfg.databases[0].volume_name, "myapp_uploads");
assert_eq!(cfg.databases[0].container_name.as_deref(), Some("myapp"));
}
#[test]
fn docker_volume_container_name_optional() {
let file = write_json(
r#"{
"databases": [
{
"name": "uploads",
"type": "docker-volume",
"volume_name": "myapp_uploads",
"generated_id": "16678159-ff7e-4c97-8c83-0adeff214681"
}
]
}"#,
);
let service = ConfigService::new(test_context());
let cfg = service.load(Some(file.path().to_str().unwrap())).unwrap();
assert_eq!(cfg.databases[0].volume_name, "myapp_uploads");
assert!(cfg.databases[0].container_name.is_none());
}
#[test]
fn docker_volume_requires_volume_name() {
let file = write_json(
r#"{
"databases": [
{
"name": "uploads",
"type": "docker-volume",
"generated_id": "16678159-ff7e-4c97-8c83-0adeff214681"
}
]
}"#,
);
let service = ConfigService::new(test_context());
let err = service.load(Some(file.path().to_str().unwrap())).unwrap_err();
assert!(err.contains("volume_name"), "error was: {err}");
}
+27 -1
View File
@@ -1,5 +1,6 @@
use crate::utils::common::{BackupMethod, vec_to_option_json};
use crate::utils::common::{BackupMethod, choose_restore_path, vec_to_option_json};
use serde_json::json;
use std::path::{Path, PathBuf};
#[test]
fn backup_method_to_string_automatic() {
@@ -47,3 +48,28 @@ fn vec_to_option_json_serializes_struct_vector() {
]))
);
}
#[test]
fn choose_restore_path_single_file_returns_that_file() {
let dir = Path::new("/tmp/extract");
let archive = Path::new("/tmp/backup.tar.gz");
let files = vec![PathBuf::from("/tmp/extract/dump.sql")];
// Single extracted file: restore from that file directly.
assert_eq!(
choose_restore_path(&files, dir, archive),
PathBuf::from("/tmp/extract/dump.sql")
);
}
#[test]
fn choose_restore_path_multi_file_non_docker_volume_returns_archive_path() {
let dir = Path::new("/tmp/extract");
let archive = Path::new("/tmp/backup.tar.gz");
let files = vec![
PathBuf::from("/tmp/extract/toc.dat"),
PathBuf::from("/tmp/extract/3141.dat.gz"),
];
let chosen = choose_restore_path(&files, dir, archive);
assert_eq!(chosen, PathBuf::from("/tmp/backup.tar.gz"));
assert_eq!(chosen.extension().and_then(|e| e.to_str()), Some("gz"));
}
+54
View File
@@ -71,3 +71,57 @@ async fn decompress_multiple_files() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn compress_tar_is_not_double_wrapped() -> Result<()> {
use tokio_tar::Builder as TarBuilder;
let tmp = tempdir()?;
// Build a real tar containing a single entry "payload.txt".
let payload = tmp.path().join("payload.txt");
write(&payload, b"volume-bytes").await?;
let tar_path = tmp.path().join("volume.tar");
{
let f = tokio::fs::File::create(&tar_path).await?;
let mut b = TarBuilder::new(f);
b.append_path_with_name(&payload, "payload.txt").await?;
b.finish().await?;
}
let result = compress_to_tar_gz_large(
&tar_path,
std::sync::Arc::new(crate::services::backup::logger::JobLogger::new()),
)
.await?;
assert_eq!(result.compressed_path, tmp.path().join("volume.tar.gz"));
// Decompress and confirm the FIRST tar entry is "payload.txt" — i.e. our tar
// was gzipped directly, not wrapped inside another tar named "volume.tar".
let out = tmp.path().join("out");
tokio::fs::create_dir_all(&out).await?;
let files = decompress_large_tar_gz(&result.compressed_path, &out).await?;
assert_eq!(files.len(), 1);
assert_eq!(files[0].file_name().unwrap(), "payload.txt");
assert_eq!(read(&files[0]).await?, b"volume-bytes");
Ok(())
}
#[tokio::test]
async fn gunzip_to_file_restores_tar_byte_for_byte() -> Result<()> {
use crate::utils::compress::gunzip_to_file;
let tmp = tempdir()?;
let tar_path = tmp.path().join("input.tar");
let original: Vec<u8> = (0u32..50_000).map(|n| (n % 256) as u8).collect();
write(&tar_path, &original).await?;
let gz = compress_to_tar_gz_large(&tar_path, std::sync::Arc::new(crate::services::backup::logger::JobLogger::new())).await?;
let out_tar = tmp.path().join("out.tar");
gunzip_to_file(&gz.compressed_path, &out_tar).await?;
assert_eq!(read(&out_tar).await?, original);
Ok(())
}
+32 -4
View File
@@ -30,12 +30,16 @@ fn full_file_name_matches_expected_suffix() {
}
#[test]
fn full_file_path_prefixes_backups_directory_and_date() {
fn full_file_path_uses_default_or_configured_folder_name() {
let file_name = "backup.tar.gz".to_string();
let full_path = full_file_path(&file_name);
assert!(full_path.starts_with("backups/"));
assert!(full_path.ends_with("/backup.tar.gz"));
let default_path = full_file_path(&file_name, None);
assert!(default_path.starts_with("backups/"));
assert!(default_path.ends_with("/backup.tar.gz"));
let configured_path = full_file_path(&file_name, Some("/portabase/"));
assert!(configured_path.starts_with("portabase/"));
assert!(configured_path.ends_with("/backup.tar.gz"));
}
#[tokio::test]
@@ -86,3 +90,27 @@ async fn encrypt_stream_starts_with_json_header_line() -> Result<()> {
Ok(())
}
use crate::utils::file::decrypt_json_gcm;
const VECTOR_KEY_B64: &str = "BwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwc=";
const VECTOR_ENVELOPE_B64: &str = "AQIDBAUGBwgJCgsM4AfK9w7I2A7UDzMvpJaScfnUYAGDZgPWT5Chrp1pdzMPPQVpNjb6ZEiFea9YdWVFv1UEo9RGmmf+zYUv4I3gE4SU/SBrMwkCHEpJGJOzJtK3tSpJmzLVX3+7EeUNwp4qjZheL8p0pe1x6dRUtx3JmLjz1W/RhWd6zuReDItv6+0jg4CaPOHvFXBreaGNCTRslxbImD+lFBoEOvw8lsbH";
const VECTOR_PLAINTEXT: &str = "[{\"id\":\"11111111-1111-1111-1111-111111111111\",\"config\":{\"bucket\":\"my-bucket\",\"accessKeyId\":\"AKIA\",\"secretAccessKey\":\"s3cr3t\"},\"provider\":\"s3\"}]";
#[test]
fn decrypt_json_gcm_decrypts_node_vector() {
let plaintext = decrypt_json_gcm(VECTOR_ENVELOPE_B64, VECTOR_KEY_B64).unwrap();
assert_eq!(String::from_utf8(plaintext).unwrap(), VECTOR_PLAINTEXT);
}
#[test]
fn decrypt_json_gcm_rejects_wrong_key() {
let wrong_key = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
assert!(decrypt_json_gcm(VECTOR_ENVELOPE_B64, wrong_key).is_err());
}
#[test]
fn decrypt_json_gcm_rejects_short_input() {
// 8 bytes base64 -> shorter than nonce(12)+tag(16)
assert!(decrypt_json_gcm("AAAAAAAAAAA=", VECTOR_KEY_B64).is_err());
}
+49 -1
View File
@@ -31,12 +31,60 @@ fn normalized_expression_is_valid_for_cron_schedule() {
#[test]
fn next_run_timestamp_returns_future_timestamp() {
let expr = normalize_cron("*/1 * * * *");
let ts = next_run_timestamp(&expr);
let ts = next_run_timestamp(&expr).unwrap();
let now = chrono::Local::now().timestamp();
assert!(ts > now);
}
#[test]
fn normalize_converts_unix_sunday_zero_to_crate_dow() {
let input = "00 06 * * 0";
let normalized = normalize_cron(input);
assert_eq!(normalized, "0 00 06 * * 1");
let schedule = Schedule::from_str(&normalized);
assert!(schedule.is_ok());
}
#[test]
fn normalize_converts_unix_saturday_to_crate_dow() {
assert_eq!(normalize_cron("00 06 * * 6"), "0 00 06 * * 7");
}
#[test]
fn normalized_sunday_actually_fires_on_sunday() {
use chrono::{Datelike, Timelike, Weekday};
let normalized = normalize_cron("00 03 * * 0");
assert_eq!(normalized, "0 00 03 * * 1");
let schedule = Schedule::from_str(&normalized).unwrap();
let next = schedule.upcoming(chrono::Utc).next().unwrap();
assert_eq!(next.weekday(), Weekday::Sun);
assert_eq!(next.hour(), 3);
assert_eq!(next.minute(), 0);
}
#[test]
fn normalize_converts_unix_dow_range() {
assert_eq!(normalize_cron("00 06 * * 0-4"), "0 00 06 * * 1-5");
}
#[test]
fn next_run_timestamp_returns_none_for_invalid_cron() {
assert!(next_run_timestamp("not a cron").is_none());
}
#[test]
fn normalize_leaves_out_of_range_dow_untouched() {
let normalized = normalize_cron("* * * * 100");
assert_eq!(normalized, "0 * * * * 100");
assert!(Schedule::from_str(&normalized).is_err());
}
#[test]
fn normalization_does_not_break_schedule_parsing() {
let input = "0 */10 * * * *";
+12
View File
@@ -1,5 +1,17 @@
use serde::Serialize;
use serde_json::Value;
use std::path::{Path, PathBuf};
pub(crate) fn choose_restore_path(
extracted: &[PathBuf],
_extraction_dir: &Path,
archive: &Path,
) -> PathBuf {
match extracted.len() {
1 => extracted[0].clone(),
_ => archive.to_path_buf(),
}
}
#[derive(Clone, Copy)]
pub enum BackupMethod {
+54 -1
View File
@@ -1,4 +1,4 @@
use anyhow::Result;
use anyhow::{Context, Result};
use async_compression::tokio::bufread::GzipDecoder;
use async_compression::tokio::write::GzipEncoder as AsyncGzipEncoder;
use futures::StreamExt;
@@ -32,6 +32,39 @@ pub async fn compress_to_tar_gz_large(file: &PathBuf, logger: Arc<JobLogger>) ->
});
}
if file
.file_name()
.and_then(|n| n.to_str())
.map(|n| n.ends_with(".tar"))
.unwrap_or(false)
{
let gz_path = PathBuf::from(format!("{}.gz", file.display()));
logger.log("info", format!("Input {:?} is a raw tar, gzipping directly", file));
let input = File::open(file)
.await
.map_err(|e| anyhow::anyhow!("Failed to open tar {:?}: {}", file, e))?;
let mut reader = BufReader::with_capacity(8 * 1024 * 1024, input);
let output_file = File::create(&gz_path)
.await
.map_err(|e| anyhow::anyhow!("Failed to create {:?}: {}", gz_path, e))?;
let mut encoder = AsyncGzipEncoder::new(output_file);
tokio::io::copy(&mut reader, &mut encoder)
.await
.map_err(|e| anyhow::anyhow!("Gzip copy failed: {}", e))?;
encoder
.shutdown()
.await
.map_err(|e| anyhow::anyhow!("Gzip shutdown failed: {}", e))?;
logger.log("info", format!("Compressed {:?} to {:?}", file, gz_path));
return Ok(CompressionResult {
compressed_path: gz_path,
});
}
let tar_gz_path = file.with_extension("").with_extension("tar.gz");
let output_file = File::create(&tar_gz_path)
@@ -119,3 +152,23 @@ pub async fn decompress_large_tar_gz(
Ok(extracted_files)
}
pub async fn gunzip_to_file(gz_path: &Path, out_path: &Path) -> Result<()> {
let file = File::open(gz_path)
.await
.with_context(|| format!("Failed to open {}", gz_path.display()))?;
let buf_reader = BufReader::with_capacity(8 * 1024 * 1024, file);
let mut decoder = GzipDecoder::new(buf_reader);
let out = File::create(out_path)
.await
.with_context(|| format!("Failed to create {}", out_path.display()))?;
let mut writer = tokio::io::BufWriter::new(out);
tokio::io::copy(&mut decoder, &mut writer).await?;
writer.shutdown().await?;
info!("Gunzipped {:?} into {:?}", gz_path, out_path);
Ok(())
}
+37 -2
View File
@@ -47,8 +47,14 @@ pub fn full_file_name(encrypt: bool) -> String {
}
}
pub fn full_file_path(file_name: &String) -> String {
format!("backups/{}/{}", Utc::now().format("%Y-%m-%d"), file_name)
pub fn full_file_path(file_name: &String, folder_name: Option<&str>) -> String {
let folder_name = folder_name
.map(str::trim)
.map(|folder_name| folder_name.trim_matches(char::from(47)))
.filter(|folder_name| !folder_name.is_empty())
.unwrap_or("backups");
format!("{}/{}/{}", folder_name, Utc::now().format("%Y-%m-%d"), file_name)
}
const CHUNK_SIZE: usize = 16 * 1024 * 1024;
@@ -175,3 +181,32 @@ pub async fn decrypt_file_stream_gcm(
writer.flush().await?;
Ok(())
}
/// Decrypt a base64 `nonce(12) ‖ ciphertext ‖ tag(16)` AES-256-GCM envelope
/// using the raw master key (STANDARD base64). Returns the plaintext bytes.
pub fn decrypt_json_gcm(ciphertext_b64: &str, master_key_b64: &str) -> Result<Vec<u8>> {
let master_key_bytes = general_purpose::STANDARD
.decode(master_key_b64)
.map_err(|_| anyhow::anyhow!("Invalid base64 master key"))?;
let data = general_purpose::STANDARD
.decode(ciphertext_b64)
.map_err(|_| anyhow::anyhow!("Invalid base64 ciphertext"))?;
if data.len() < 12 + 16 {
return Err(anyhow::anyhow!("Ciphertext too short"));
}
let key = Key::<Aes256Gcm>::try_from(master_key_bytes.as_slice())
.map_err(|_| anyhow::anyhow!("Invalid AES-256 key length"))?;
let cipher = Aes256Gcm::new(&key);
let nonce = Nonce::try_from(&data[..12])
.map_err(|_| anyhow::anyhow!("Invalid nonce length"))?;
let plaintext = cipher
.decrypt(&nonce, &data[12..])
.map_err(|e| anyhow::anyhow!("AES-GCM decryption failed: {:?}", e))?;
Ok(plaintext)
}
+19 -21
View File
@@ -10,9 +10,9 @@ use std::str::FromStr;
use tracing::debug;
use tracing::info;
pub fn next_run_timestamp(expr: &str) -> i64 {
let schedule = Schedule::from_str(expr).unwrap();
schedule.upcoming(Local).next().unwrap().timestamp()
pub fn next_run_timestamp(expr: &str) -> Option<i64> {
let schedule = Schedule::from_str(expr).ok()?;
Some(schedule.upcoming(Local).next()?.timestamp())
}
pub async fn check_and_update_cron(
@@ -38,8 +38,9 @@ pub async fn check_and_update_cron(
}
Some(cron) => {
let cron = normalize_cron(&cron);
debug!("Task cron (normalized): {:?}", cron);
let raw_cron = cron;
let cron = normalize_cron(&raw_cron);
debug!("Task cron normalized: unix \"{}\" -> crate \"{}\"", raw_cron, cron);
if exists {
let raw: String = conn.hget(&redis_key, "data").await.unwrap();
@@ -50,24 +51,21 @@ pub async fn check_and_update_cron(
let metadata_changed = stored.metadata != metadata;
if cron_changed || args_changed || metadata_changed {
upsert_task(conn, &task_name, task, &cron, args.clone(), metadata)
.await
.unwrap_or_else(|e| {
tracing::error!("Failed to update task {}: {:?}", task_name, e);
});
info!(
"Task {} updated (cron: {}, args: {}, metadata: {})",
task_name, cron_changed, args_changed, metadata_changed
);
match upsert_task(conn, &task_name, task, &cron, args.clone(), metadata).await {
Ok(()) => info!(
"Task {} updated (cron: {}, args: {}, metadata: {})",
task_name, cron_changed, args_changed, metadata_changed
),
Err(e) => {
tracing::error!("Failed to update task {}: {:?}", task_name, e)
}
}
}
} else {
upsert_task(conn, &task_name, task, &cron, args, metadata)
.await
.unwrap_or_else(|e| {
tracing::error!("Failed to create task {}: {:?}", task_name, e);
});
info!("Task {} created", task_name);
match upsert_task(conn, &task_name, task, &cron, args, metadata).await {
Ok(()) => info!("Task {} created", task_name),
Err(e) => tracing::error!("Failed to create task {}: {:?}", task_name, e),
}
}
}
}
+11 -2
View File
@@ -48,8 +48,17 @@ pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
task_clone.task, e
);
}
let next_ts = next_run_timestamp(&task_clone.cron);
let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap();
match next_run_timestamp(&task_clone.cron) {
Some(next_ts) => {
let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap();
}
None => {
error!(
"Invalid cron expression for task={}: {}",
task_clone.task, task_clone.cron
);
}
}
});
}
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+10 -1
View File
@@ -16,7 +16,16 @@ pub async fn upsert_task(
metadata: Option<Value>,
) -> redis::RedisResult<()> {
let key = format!("redbeat:{}", name);
let next_ts = next_run_timestamp(cron);
let next_ts = match next_run_timestamp(cron) {
Some(ts) => ts,
None => {
return Err(redis::RedisError::from((
redis::ErrorKind::Client,
"invalid cron expression",
cron.to_string(),
)));
}
};
let entry = PeriodicTask {
task: task.to_string(),
+51 -3
View File
@@ -1,7 +1,55 @@
pub fn normalize_cron(expr: &str) -> String {
if expr.split_whitespace().count() == 5 {
format!("0 {}", expr)
let fields: Vec<&str> = expr.split_whitespace().collect();
let (sec, mut rest): (String, Vec<String>) = match fields.len() {
5 => ("0".to_string(), fields.iter().map(|s| s.to_string()).collect()),
6 => (
fields[0].to_string(),
fields[1..].iter().map(|s| s.to_string()).collect(),
),
_ => return expr.to_string(),
};
if let Some(last) = rest.last_mut() {
*last = convert_dow(last);
}
format!("{} {}", sec, rest.join(" "))
}
fn convert_dow(field: &str) -> String {
field
.split(',')
.map(convert_dow_part)
.collect::<Vec<_>>()
.join(",")
}
fn convert_dow_part(part: &str) -> String {
let (base, step) = match part.split_once('/') {
Some((b, s)) => (b, Some(s)),
None => (part, None),
};
let converted = if let Some((start, end)) = base.split_once('-') {
match (start.parse::<u8>(), end.parse::<u8>()) {
(Ok(a), Ok(b)) if a <= 7 && b <= 7 => {
format!("{}-{}", (a % 7) + 1, (b % 7) + 1)
}
_ => base.to_string(),
}
} else if let Ok(n) = base.parse::<u8>() {
if n <= 7 {
((n % 7) + 1).to_string()
} else {
base.to_string()
}
} else {
expr.to_string()
base.to_string()
};
match step {
Some(s) => format!("{}/{}", converted, s),
None => converted,
}
}