chore: release 1.16.1

feat: docker-volume-backup-restore

.github/workflows/codecov.yml

@@ -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
             "
 

CITATION.cff

@@ -27,5 +27,5 @@ keywords:
   - self-hosted
   - portabase
 license: Apache-2.0
-version: 1.13.0
+version: 1.16.1
 date-released: '2026-02-24'

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "portabase-agent"
-version = "1.13.0"
+version = "1.16.1"
 edition = "2024"
 
 [dependencies]
@@ -35,12 +35,14 @@ 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"] }
 azure_core = "1.0.0"
 azure_storage_blob = "1.0.0"
+google-cloud-storage = "1.15"
+google-cloud-auth = "1.13"
 async-compression = { version = "0.4.37", features = ["tokio", "gzip"] }
 tokio-tar = "0.3.1"
 oauth2 = "5.0.0"
@@ -55,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"] }

databases.json

@@ -10,6 +10,19 @@
       "host": "db-postgres",
       "generated_id": "16678159-ff7e-4c97-8c83-0adeff214681"
     },
+    {
+      "name": "Test database 1 - PostgreSQL - BIS",
+      "database": "devdb2",
+      "type": "postgresql",
+      "username": "devuser2",
+      "password": "changeme2",
+      "port": 5432,
+      "host": "db-postgres-2",
+      "generated_id": "16678159-ff7e-5697-8c83-0adeff214681",
+      "options": {
+        "keep_ownership": true
+      }
+    },
     {
       "name": "Test database 2 - MariaDB",
       "database": "mariadb",
@@ -111,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"
     }
   ]
 }

docker-compose.databases.yml

@@ -15,6 +15,20 @@ services:
     networks:
       - portabase
 
+  db-postgres-2:
+    container_name: db-postgres-2
+    image: postgres:17-alpine
+    ports:
+      - "5438:5432"
+    volumes:
+      - postgres-data-2:/var/lib/postgresql/data
+    environment:
+      - POSTGRES_DB=devdb2
+      - POSTGRES_USER=devuser2
+      - POSTGRES_PASSWORD=changeme2
+    networks:
+      - portabase
+
   db-mariadb:
     container_name: db-mariadb
     image: mariadb:latest
@@ -179,6 +193,7 @@ services:
 
 volumes:
   postgres-data:
+  postgres-data-2:
   mariadb-data:
   mysql-data:
   mongodb-data:

docker-compose.yml

@@ -11,7 +11,7 @@ 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
@@ -19,7 +19,7 @@ services:
       APP_ENV: development
       LOG: debug
       TZ: "Europe/Paris"
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNWRkZTE1NTctZWQ1ZC00MjUxLThiZDMtMDE0MjkxOTg2OGZjIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiMDJlYjQ5MjEtY2VhZC00NjlhLTg2NzItM2IyY2YwZDEzNWE1IiwibWFzdGVyS2V5QjY0Ijoidkd4UU5vbGJxcU1XYWExQWU1Y1JDZ1BTVFBrMHo5cUJPNW5xRHN3bGh2OD0ifQ=="
       #CHUNK_SIZE_MB: "1"
       #POOLING: 1
       #DATABASES_CONFIG_FILE: "config.toml"
@@ -27,12 +27,9 @@ services:
       - "localhost:host-gateway"
     networks:
       - portabase
-
     cpus: "1.50"
-
     mem_limit: 4g
     memswap_limit: 4g
-
     pids_limit: 512
 
 

src/domain/docker_volume/backup.rs

@@ -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?
+}

src/domain/docker_volume/database.rs

@@ -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
+    }
+}

src/domain/docker_volume/docker.rs

@@ -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)
+}

src/domain/docker_volume/mod.rs

@@ -0,0 +1,5 @@
+pub mod backup;
+pub mod database;
+pub mod docker;
+pub mod ping;
+pub mod restore;

src/domain/docker_volume/ping.rs

@@ -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()),
+    }
+}

src/domain/docker_volume/restore.rs

@@ -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?
+}

src/domain/factory.rs

@@ -1,5 +1,7 @@
+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;
 use crate::domain::postgres::database::PostgresDatabase;
 use crate::domain::postgres::{detect_format_from_file, detect_format_from_size};
 use crate::domain::redis::database::RedisDatabase;
@@ -31,6 +33,7 @@ impl DatabaseFactory {
                 let format = detect_format_from_size(&cfg).await;
                 Arc::new(PostgresDatabase::new(cfg, format))
             }
+            DbType::PostgresqlCluster => Arc::new(PostgresClusterDatabase::new(cfg)),
             DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
             DbType::Mariadb => Arc::new(MariaDBDatabase::new(cfg)),
             DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
@@ -39,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)),
         }
     }
 
@@ -48,6 +52,7 @@ impl DatabaseFactory {
                 let format = detect_format_from_file(restore_file);
                 Arc::new(PostgresDatabase::new(cfg, format))
             }
+            DbType::PostgresqlCluster => Arc::new(PostgresClusterDatabase::new(cfg)),
             DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
             DbType::Mariadb => Arc::new(MariaDBDatabase::new(cfg)),
             DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
@@ -56,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)),
         }
     }
 }

src/domain/mod.rs

@@ -1,3 +1,4 @@
+pub mod docker_volume;
 pub mod factory;
 mod mongodb;
 pub mod mysql;

src/domain/postgres/cluster/backup.rs

@@ -0,0 +1,81 @@
+use anyhow::Result;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::process::Command;
+use std::sync::Arc;
+use std::time::Instant;
+
+use super::super::connection::{
+    is_superuser, pg_dumpall_binary_name, select_pg_path, server_version,
+};
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+
+pub async fn run(
+    cfg: DatabaseConfig,
+    backup_dir: PathBuf,
+    env: HashMap<String, String>,
+    logger: Arc<JobLogger>,
+) -> Result<PathBuf> {
+    tokio::task::spawn_blocking(move || -> Result<PathBuf> {
+        logger.log("info", format!("Starting cluster backup for {}", cfg.name));
+
+        let version = match futures::executor::block_on(server_version(&cfg)) {
+            Ok(v) => v,
+            Err(e) => {
+                logger.log("error", format!("Failed to get server version for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        };
+
+        match futures::executor::block_on(is_superuser(&cfg)) {
+            Ok(true) => {}
+            Ok(false) => {
+                logger.log("error", format!("postgresql-cluster backup requires a superuser role for {}", cfg.name));
+                anyhow::bail!("postgresql-cluster backup requires a superuser role for {}", cfg.name);
+            }
+            Err(e) => {
+                logger.log("error", format!("Failed to check superuser status for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        }
+
+        let pg_dumpall = select_pg_path(&version).join(pg_dumpall_binary_name());
+        let file_path = backup_dir.join(format!("{}.sql", cfg.generated_id));
+
+        logger.log("info", format!("Running pg_dumpall for cluster {} via {:?}", cfg.name, pg_dumpall));
+
+        let start = Instant::now();
+        let output = Command::new(&pg_dumpall)
+            .arg("--host").arg(&cfg.host)
+            .arg("--port").arg(cfg.port.to_string())
+            .arg("--username").arg(&cfg.username)
+            .arg("--clean")
+            .arg("--if-exists")
+            .arg("-v")
+            .arg("-f").arg(&file_path)
+            .envs(env)
+            .output();
+        let duration_ms = start.elapsed().as_millis() as f64;
+
+        match output {
+            Ok(o) => {
+                let stderr = String::from_utf8_lossy(&o.stderr).to_string();
+                let exit_code = o.status.code().unwrap_or(-1);
+                if o.status.success() {
+                    logger.log_command("pg_dumpall", if stderr.is_empty() { None } else { Some(stderr) }, Some(0), Some(duration_ms));
+                    logger.log("info", format!("Cluster backup completed for {} at {:?}", cfg.name, file_path));
+                    Ok(file_path)
+                } else {
+                    logger.log_command("pg_dumpall", Some(stderr), Some(exit_code), Some(duration_ms));
+                    anyhow::bail!("Cluster backup (pg_dumpall) failed for {}", cfg.name);
+                }
+            }
+            Err(e) => {
+                logger.log_command("pg_dumpall", Some(e.to_string()), Some(-1), Some(duration_ms));
+                Err(e.into())
+            }
+        }
+    })
+    .await?
+}

src/domain/postgres/cluster/database.rs

@@ -0,0 +1,53 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use super::super::ping;
+use super::{backup, 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 PostgresClusterDatabase {
+    pub cfg: DatabaseConfig,
+}
+
+impl PostgresClusterDatabase {
+    pub fn new(cfg: DatabaseConfig) -> Self {
+        Self { cfg }
+    }
+
+    fn build_env(&self) -> HashMap<String, String> {
+        let mut envs = std::env::vars().collect::<HashMap<_, _>>();
+        envs.insert("PGPASSWORD".to_string(), self.cfg.password.to_string());
+        envs
+    }
+}
+
+#[async_trait]
+impl Database for PostgresClusterDatabase {
+    fn file_extension(&self) -> &'static str {
+        ".sql"
+    }
+
+    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(), self.build_env(), 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(), self.build_env(), logger).await;
+        FileLock::release(&self.cfg.generated_id).await?;
+        res
+    }
+}

src/domain/postgres/cluster/mod.rs

@@ -0,0 +1,3 @@
+pub mod backup;
+pub mod database;
+pub mod restore;

src/domain/postgres/cluster/restore.rs

@@ -0,0 +1,84 @@
+use anyhow::Result;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::process::Command;
+use std::sync::Arc;
+use std::time::Instant;
+
+use super::super::connection::{is_superuser, psql_binary_name, select_pg_path, server_version, terminate_all_connections};
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+
+pub async fn run(
+    cfg: DatabaseConfig,
+    restore_file: PathBuf,
+    env: HashMap<String, String>,
+    logger: Arc<JobLogger>,
+) -> Result<()> {
+    tokio::task::spawn_blocking(move || -> Result<()> {
+        logger.log("info", format!("Starting cluster restore for {}", cfg.name));
+
+        let version = match futures::executor::block_on(server_version(&cfg)) {
+            Ok(v) => v,
+            Err(e) => {
+                logger.log("error", format!("Failed to get server version for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        };
+
+        match futures::executor::block_on(is_superuser(&cfg)) {
+            Ok(true) => {}
+            Ok(false) => {
+                logger.log("error", format!("postgresql-cluster restore requires a superuser role for {}", cfg.name));
+                anyhow::bail!("postgresql-cluster restore requires a superuser role for {}", cfg.name);
+            }
+            Err(e) => {
+                logger.log("error", format!("Failed to check superuser status for {}: {:?}", cfg.name, e));
+                return Err(e.into());
+            }
+        }
+
+        let psql = select_pg_path(&version).join(psql_binary_name());
+
+        if let Err(e) = futures::executor::block_on(terminate_all_connections(&cfg)) {
+            logger.log("error", format!("Failed to terminate connections for cluster {}: {:?}", cfg.name, e));
+            return Err(e.into());
+        }
+        logger.log("info", format!("All user database connections terminated for cluster {}", cfg.name));
+
+        logger.log("info", format!("Replaying cluster dump for {} via {:?}", cfg.name, psql));
+
+        let start = Instant::now();
+        let output = Command::new(&psql)
+            .arg("--host").arg(&cfg.host)
+            .arg("--port").arg(cfg.port.to_string())
+            .arg("--username").arg(&cfg.username)
+            .arg("--dbname").arg("postgres")
+            .arg("-f").arg(&restore_file)
+            .envs(env)
+            .output();
+        let duration_ms = start.elapsed().as_millis() as f64;
+
+        match output {
+            Ok(o) => {
+                let stderr = String::from_utf8_lossy(&o.stderr).to_string();
+                let stdout = String::from_utf8_lossy(&o.stdout).to_string();
+                let combined = format!("{}{}", stdout, stderr);
+                let exit_code = o.status.code().unwrap_or(-1);
+                if o.status.success() {
+                    logger.log_command("psql", if combined.is_empty() { None } else { Some(combined) }, Some(0), Some(duration_ms));
+                    logger.log("info", format!("Cluster restore completed for {}", cfg.name));
+                    Ok(())
+                } else {
+                    logger.log_command("psql", if combined.is_empty() { None } else { Some(combined) }, Some(exit_code), Some(duration_ms));
+                    anyhow::bail!("Cluster restore (psql) failed for {}", cfg.name);
+                }
+            }
+            Err(e) => {
+                logger.log_command("psql", Some(e.to_string()), Some(-1), Some(duration_ms));
+                Err(e.into())
+            }
+        }
+    })
+    .await?
+}

src/domain/postgres/connection.rs

@@ -33,32 +33,21 @@ pub async fn server_version(cfg: &DatabaseConfig) -> Result<String> {
     Ok(version)
 }
 
-/// Resolves the `bin` directory of a PostgreSQL installation for the given
-/// major version, in a cross-platform way.
-///
-/// Resolution order:
-/// 1. The `PG_BIN_DIR` environment variable, if set, is used as-is. This
-///    allows users/CI to override detection for non-standard installs
-///    (e.g. portable PostgreSQL distributions, custom install locations).
-/// 2. Platform-specific default install locations (Debian/Ubuntu packages,
-///    the official Windows installer, Homebrew/Postgres.app on macOS, and
-///    common RPM-based layouts on other Linux distros).
-/// 3. A `PATH` lookup for `pg_dump` (`pg_dump.exe` on Windows), returning
-///    its parent directory.
-/// 4. The historical Debian/Ubuntu path as a last-resort fallback, so the
-///    function keeps returning a `PathBuf` (never panics) even when nothing
-///    was found, preserving the previous behavior for callers.
-///
-/// The override is sourced from `CONFIG.pg_bin_dir` (the `PG_BIN_DIR`
-/// environment variable). An empty value means "unset" and falls through to
-/// detection.
+pub async fn is_superuser(cfg: &DatabaseConfig) -> Result<bool> {
+    let client = connect(cfg).await?;
+    let is_super: bool = client
+        .query_one("SELECT current_setting('is_superuser') = 'on';", &[])
+        .await?
+        .get(0);
+
+    Ok(is_super)
+}
+
+
 pub fn select_pg_path(version: &str) -> std::path::PathBuf {
     select_pg_path_with(version, &CONFIG.pg_bin_dir)
 }
 
-/// Inner resolver behind [`select_pg_path`], parameterized over the
-/// `PG_BIN_DIR` override. Kept pure (no env / no `CONFIG` access) so it is
-/// unit-testable without mutating process-global state.
 pub(crate) fn select_pg_path_with(version: &str, pg_bin_dir: &str) -> std::path::PathBuf {
     let major = version.split('.').next().unwrap_or("17");
 
@@ -109,6 +98,22 @@ pub(crate) fn pg_dump_binary_name() -> &'static str {
     }
 }
 
+pub(crate) fn pg_dumpall_binary_name() -> &'static str {
+    if cfg!(target_os = "windows") {
+        "pg_dumpall.exe"
+    } else {
+        "pg_dumpall"
+    }
+}
+
+pub(crate) fn psql_binary_name() -> &'static str {
+    if cfg!(target_os = "windows") {
+        "psql.exe"
+    } else {
+        "psql"
+    }
+}
+
 pub(crate) fn pg_dump_exists_in(dir: &std::path::Path) -> bool {
     dir.join(pg_dump_binary_name()).is_file()
 }
@@ -139,6 +144,27 @@ pub async fn terminate_connections(cfg: &DatabaseConfig) -> Result<()> {
     Ok(())
 }
 
+pub async fn terminate_all_connections(cfg: &DatabaseConfig) -> Result<()> {
+    let mut admin = cfg.clone();
+    admin.database = "postgres".to_string().into();
+
+    let client = connect(&admin).await?;
+
+    client
+        .execute(
+            r#"
+            SELECT pg_terminate_backend(pid)
+            FROM pg_stat_activity
+            WHERE datname NOT IN ('postgres', 'template0', 'template1')
+              AND pid <> pg_backend_pid();
+            "#,
+            &[],
+        )
+        .await?;
+
+    Ok(())
+}
+
 pub fn detect_format_from_file(restore_file: &Path) -> PostgresDumpFormat {
     match restore_file.extension().and_then(|e| e.to_str()) {
         Some("dump") => PostgresDumpFormat::Fc,

src/domain/postgres/mod.rs

@@ -1,4 +1,5 @@
 pub mod backup;
+pub(crate) mod cluster;
 pub(crate) mod connection;
 pub mod database;
 mod format;

src/domain/postgres/restore.rs

@@ -41,13 +41,26 @@ pub async fn run(
         }
         logger.log("info", format!("Connections terminated for database {}", cfg.name));
 
+        let keep_ownership = cfg.options
+            .get("keep_ownership")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false);
+
+        if keep_ownership {
+            logger.log("info", format!("Restoring ownership and privileges for {}", cfg.name));
+        } else {
+            logger.log("info", format!("Stripping ownership and privileges for {} (--no-owner --no-privileges)", cfg.name));
+        }
+
         match format {
             PostgresDumpFormat::Fc => {
                 logger.log("info", format!("Running FC restore for {}", cfg.name));
                 let start = Instant::now();
-                let output = Command::new(&pg_restore)
-                    .arg("--no-owner")
-                    .arg("--no-privileges")
+                let mut cmd = Command::new(&pg_restore);
+                if !keep_ownership {
+                    cmd.arg("--no-owner").arg("--no-privileges");
+                }
+                let output = cmd
                     .arg("--clean")
                     .arg("--if-exists")
                     // .arg("--create")
@@ -147,9 +160,11 @@ pub async fn run(
                 };
 
                 let start = Instant::now();
-                let output = Command::new(&pg_restore)
-                    .arg("--no-owner")
-                    .arg("--no-privileges")
+                let mut cmd = Command::new(&pg_restore);
+                if !keep_ownership {
+                    cmd.arg("--no-owner").arg("--no-privileges");
+                }
+                let output = cmd
                     .arg("--clean")
                     .arg("--if-exists")
                     // .arg("--create")

src/main.rs

@@ -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;

src/services/api/models/agent/status.rs

@@ -1,6 +1,6 @@
 #![allow(dead_code)]
 
-use crate::utils::deserializer::deserialize_snake_case;
+use crate::utils::deserializer::{deserialize_snake_case, string_or_number_to_string};
 use serde::{Deserialize, Serialize};
 use toml::Value;
 
@@ -54,4 +54,6 @@ pub struct RestoreInfo {
     pub file: Option<String>,
     #[serde(rename = "metaFile")]
     pub meta_file: Option<String>,
+    #[serde(default, deserialize_with = "string_or_number_to_string")]
+    pub size: Option<String>,
 }

src/services/backup/uploader.rs

@@ -104,6 +104,7 @@ impl BackupService {
                         method,
                         &storage,
                         Some(encrypt),
+                        &backup_storage_id,
                     )
                     .await;
 

src/services/config.rs

@@ -3,6 +3,7 @@
 use crate::core::context::Context;
 use serde::Deserialize;
 use serde_json;
+use std::collections::HashMap;
 use std::fs::File;
 use std::io::Read;
 use std::path::Path;
@@ -17,12 +18,16 @@ pub enum DbType {
     Mysql,
     Mariadb,
     Postgresql,
+    #[serde(rename = "postgresql-cluster")]
+    PostgresqlCluster,
     MongoDB,
     Sqlite,
     Redis,
     Valkey,
     Firebird,
     Mssql,
+    #[serde(rename = "docker-volume")]
+    DockerVolume,
 }
 
 impl DbType {
@@ -31,12 +36,14 @@ impl DbType {
             DbType::Mysql => "mysql",
             DbType::Mariadb => "mariadb",
             DbType::Postgresql => "postgresql",
+            DbType::PostgresqlCluster => "postgresql-cluster",
             DbType::MongoDB => "mongodb",
             DbType::Sqlite => "sqlite",
             DbType::Redis => "redis",
             DbType::Valkey => "valkey",
             DbType::Firebird => "firebird",
             DbType::Mssql => "mssql",
+            DbType::DockerVolume => "docker-volume",
         }
     }
 }
@@ -55,6 +62,9 @@ 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>,
 }
 
 #[allow(dead_code)]
@@ -77,6 +87,9 @@ 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>>,
 }
 
 #[allow(dead_code)]
@@ -169,21 +182,26 @@ impl ConfigService {
             }
 
             let username = match db.db_type {
-                DbType::Postgresql | DbType::Mysql | DbType::Mariadb | DbType::Mssql => {
-                    required(&db.username, &db.name, "username")?
-                }
+                DbType::Postgresql
+                | DbType::PostgresqlCluster
+                | DbType::Mysql
+                | DbType::Mariadb
+                | DbType::Mssql => required(&db.username, &db.name, "username")?,
                 _ => optional(&db.username),
             };
 
             let password = match db.db_type {
-                DbType::Postgresql | DbType::Mysql | DbType::Mariadb | DbType::Mssql => {
-                    required(&db.password, &db.name, "password")?
-                }
+                DbType::Postgresql
+                | DbType::PostgresqlCluster
+                | DbType::Mysql
+                | DbType::Mariadb
+                | DbType::Mssql => required(&db.password, &db.name, "password")?,
                 _ => optional(&db.password),
             };
 
             let host = match db.db_type {
                 DbType::Postgresql
+                | DbType::PostgresqlCluster
                 | DbType::Mysql
                 | DbType::Mariadb
                 | DbType::MongoDB
@@ -191,11 +209,12 @@ 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 {
                 DbType::Postgresql
+                | DbType::PostgresqlCluster
                 | DbType::Mysql
                 | DbType::Mariadb
                 | DbType::MongoDB
@@ -203,11 +222,17 @@ 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()
+                    .unwrap_or_else(|| "postgres".to_string()),
                 _ => required(&db.database, &db.name, "database")?,
             };
 
@@ -223,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,
@@ -234,6 +265,9 @@ impl ConfigService {
                 generated_id: db.generated_id,
                 path: path_val,
                 max_packet_size,
+                volume_name,
+                container_name,
+                options: db.options.unwrap_or_default(),
             });
         }
 

src/services/restore/archive.rs

@@ -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)
     }
 }

src/services/restore/dispatcher.rs

@@ -20,6 +20,8 @@ impl RestoreService {
             return;
         };
 
+        let expected_size = db.data.restore.size.clone();
+
         let service = Self {
             ctx: self.ctx.clone(),
         };
@@ -27,7 +29,10 @@ impl RestoreService {
         let db_cfg = cfg.clone();
 
         tokio::spawn(async move {
-            if let Err(e) = service.execute_restore(db_cfg, file_to_restore).await {
+            if let Err(e) = service
+                .execute_restore(db_cfg, file_to_restore, expected_size)
+                .await
+            {
                 error!("Restore failed: {}", e);
             }
         });

src/services/restore/downloader.rs

@@ -1,20 +1,40 @@
 use super::service::RestoreService;
 
 use anyhow::Result;
+use futures::StreamExt;
 use reqwest::{Client, Url};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::time::Instant;
+use tokio::io::AsyncWriteExt;
 use crate::services::backup::logger::JobLogger;
 
+fn human_size(bytes: u64) -> String {
+    if bytes >= 1024 * 1024 {
+        format!("{} MB", bytes / 1024 / 1024)
+    } else if bytes >= 1024 {
+        format!("{} KB", bytes / 1024)
+    } else {
+        format!("{bytes} B")
+    }
+}
+
 impl RestoreService {
-    pub async fn download_backup(&self, file_url: &str, tmp_path: &Path, logger: Arc<JobLogger>) -> Result<PathBuf> {
+    pub async fn download_backup(
+        &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());
 
         let client = Client::new();
 
         let response = client.get(file_url).send().await?;
+        let status = response.status();
 
-        if !response.status().is_success() {
+        if !status.is_success() {
             logger.log("error", "Failed to download".to_string());
             anyhow::bail!("download failed");
         }
@@ -39,11 +59,66 @@ impl RestoreService {
 
         let path = tmp_path.join(&filename);
 
-        let bytes = response.bytes().await?;
+        let total = expected_size
+            .as_deref()
+            .and_then(|s| s.trim().parse::<u64>().ok())
+            .filter(|&n| n > 0);
 
-        tokio::fs::write(&path, &bytes).await?;
+        logger.log(
+            "info",
+            format!(
+                "Downloading backup '{}' ({})",
+                filename,
+                total.map(human_size).unwrap_or_else(|| "unknown size".to_string())
+            ),
+        );
+
+        let start = Instant::now();
+        let mut file = tokio::fs::File::create(&path).await?;
+        let mut stream = response.bytes_stream();
+        let mut downloaded: u64 = 0;
+        let mut next_pct: u64 = 10;
+
+        while let Some(chunk) = stream.next().await {
+            let chunk = chunk?;
+            file.write_all(&chunk).await?;
+            downloaded += chunk.len() as u64;
+
+            if let Some(total) = total {
+                let pct = (downloaded.saturating_mul(100) / total).min(100);
+                let milestone = pct / 10 * 10;
+                if milestone >= next_pct {
+                    logger.log(
+                        "info",
+                        format!(
+                            "Download progress: {}% ({} / {} bytes)",
+                            milestone, downloaded, total
+                        ),
+                    );
+                    next_pct = milestone + 10;
+                }
+            }
+        }
+
+        file.flush().await?;
+
+        if downloaded == 0 {
+            logger.log(
+                "warn",
+                format!("Downloaded 0 bytes (status {status}); backup body was empty"),
+            );
+        }
+
+        logger.log(
+            "info",
+            format!(
+                "Backup downloaded to {} ( {} bytes in {:.1}s)",
+                path.display(),
+                downloaded,
+                start.elapsed().as_secs_f64()
+            ),
+        );
 
-        logger.log("info", format!("Backup downloaded to {}", path.display()));
         Ok(path)
     }
 }

src/services/restore/executor.rs

@@ -7,7 +7,12 @@ use std::time::Instant;
 use tempfile::TempDir;
 
 impl RestoreService {
-    pub async fn execute_restore(&self, cfg: DatabaseConfig, file_url: String) -> Result<()> {
+    pub async fn execute_restore(
+        &self,
+        cfg: DatabaseConfig,
+        file_url: String,
+        expected_size: Option<String>,
+    ) -> Result<()> {
         let logger = Arc::new(JobLogger::new());
         let start = Instant::now();
 
@@ -18,9 +23,13 @@ impl RestoreService {
 
         logger.log("info", format!("Created temp directory {}", tmp_path.display()));
 
-        let downloaded = self.download_backup(&file_url, tmp_path, Arc::clone(&logger)).await?;
+        let downloaded = self
+            .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?;
 

src/services/storage/mod.rs

@@ -6,6 +6,7 @@ use crate::services::backup::models::{BackupResult, UploadResult};
 use crate::utils::common::BackupMethod;
 use async_trait::async_trait;
 use providers::azure_blob;
+use providers::google_cloud_storage;
 use providers::google_drive;
 use providers::local;
 use providers::s3;
@@ -21,6 +22,7 @@ pub trait StorageProvider: Send + Sync {
         method: BackupMethod,
         config: &DatabaseStorage,
         encrypt: Option<bool>,
+        backup_storage_id: &str,
     ) -> UploadResult;
 }
 
@@ -34,6 +36,9 @@ pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider
         "s3" => Some(Box::new(s3::S3Provider {})),
         "blob" => Some(Box::new(azure_blob::AzureBlobProvider {})),
         "google-drive" => Some(Box::new(google_drive::GoogleDriveProvider {})),
+        "google-cloud-storage" => Some(Box::new(
+            google_cloud_storage::GoogleCloudStorageProvider {},
+        )),
         _ => {
             error!("Unknown storage provider: {}", storage.provider);
             None

src/services/storage/providers/azure_blob/helpers.rs

@@ -44,8 +44,6 @@ pub(crate) fn hmac_sha256_b64(key: &[u8], data: &str) -> Result<String> {
     Ok(STANDARD.encode(sig))
 }
 
-/// Build Service SAS query pairs (raw, un-encoded) for `canonical_resource`
-/// e.g. `/blob/{account}/{container}/{blob}`.
 pub fn build_service_sas(
     resolved: &ResolvedAzure,
     canonical_resource: &str,
@@ -81,7 +79,6 @@ pub fn build_service_sas(
     ])
 }
 
-/// Build a SAS-scoped URL for a blob (or container when `blob` is empty).
 pub fn build_sas_url(
     resolved: &ResolvedAzure,
     container: &str,
@@ -110,12 +107,10 @@ pub fn build_sas_url(
     Ok(url)
 }
 
-/// Default block size for the provider path (mirrors the S3 provider's PART_SIZE).
 pub const BLOCK_SIZE: usize = 100 * 1024 * 1024;
 
 type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>;
 
-/// Stage one block under a zero-padded sequential id; records the RAW id bytes.
 async fn stage_block(
     bbc: &BlockBlobClient,
     index: u32,
@@ -132,13 +127,6 @@ async fn stage_block(
     Ok(())
 }
 
-/// Stream `body` to `{container}/{blob}` using Azure block upload. Never buffers the
-/// full payload: at most one `block_size` block plus one inbound chunk is resident
-/// (mirrors the S3 provider's per-part guarantee).
-///
-/// Assumes the container already exists — S3-faithful, no container creation. Uncommitted
-/// blocks are garbage-collected by Azure if `commit_block_list` is never reached, so no
-/// explicit abort is needed on the error path (unlike S3 multipart).
 pub async fn upload_stream_to_azure(
     resolved: &ResolvedAzure,
     container: &str,

src/services/storage/providers/azure_blob/mod.rs

@@ -1,5 +1,5 @@
 pub mod helpers;
-mod models;
+pub(crate) mod models;
 
 use crate::core::context::Context;
 use crate::services::api::models::agent::status::DatabaseStorage;
@@ -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 {

src/services/storage/providers/azure_blob/models.rs

@@ -1,13 +1,18 @@
 use crate::services::storage::providers::azure_blob::helpers::ResolvedAzure;
 use anyhow::{Result, anyhow};
 use serde::{Deserialize, Serialize};
+use url::Url;
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct AzureBlobProviderConfig {
+    #[serde(default)]
     pub account_name: String,
+    #[serde(default)]
     pub account_key: String,
     pub container_name: String,
     #[serde(default)]
+    pub auth_mode: Option<String>,
+    #[serde(default)]
     pub connection_string: String,
     #[serde(default)]
     pub endpoint_url: Option<String>,
@@ -25,10 +30,36 @@ fn parse_connection_string(cs: &str) -> std::collections::HashMap<String, String
         .collect()
 }
 
+
+pub(crate) fn ensure_account_in_endpoint(endpoint: &str, account: &str) -> String {
+    let trimmed = endpoint.trim_end_matches('/');
+    if account.is_empty() {
+        return trimmed.to_string();
+    }
+    if let Ok(url) = Url::parse(trimmed) {
+        let host = url.host_str().unwrap_or("");
+        if host.contains(account) {
+            return trimmed.to_string();
+        }
+        let path = url.path().trim_matches('/');
+        if path == account || path.starts_with(&format!("{account}/")) {
+            return trimmed.to_string();
+        }
+    }
+    format!("{trimmed}/{account}")
+}
+
 impl AzureBlobProviderConfig {
-    /// Resolve effective connection params, preferring the connection string when non-empty.
     pub fn resolve(&self) -> Result<ResolvedAzure> {
-        if !self.connection_string.trim().is_empty() {
+        let mode = self.auth_mode.as_deref().unwrap_or("").trim();
+        let has_connection_string = !self.connection_string.trim().is_empty();
+
+        if mode == "connectionString" || (mode.is_empty() && has_connection_string) {
+            if !has_connection_string {
+                return Err(anyhow!(
+                    "authMode is connectionString but connectionString is empty"
+                ));
+            }
             let map = parse_connection_string(&self.connection_string);
             let account_name = map
                 .get("AccountName")
@@ -49,11 +80,22 @@ impl AzureBlobProviderConfig {
             });
         }
 
-        let blob_endpoint = self
+        if self.account_name.trim().is_empty() {
+            return Err(anyhow!("accountName required for accountKey auth"));
+        }
+        if self.account_key.trim().is_empty() {
+            return Err(anyhow!("accountKey required for accountKey auth"));
+        }
+
+        let blob_endpoint = match self
             .endpoint_url
-            .clone()
-            .filter(|s| !s.trim().is_empty())
-            .ok_or_else(|| anyhow!("endpointUrl required when connectionString is empty"))?;
+            .as_deref()
+            .map(str::trim)
+            .filter(|s| !s.is_empty())
+        {
+            Some(endpoint) => ensure_account_in_endpoint(endpoint, &self.account_name),
+            None => format!("https://{}.blob.core.windows.net", self.account_name),
+        };
 
         Ok(ResolvedAzure {
             account_name: self.account_name.clone(),

src/services/storage/providers/google_cloud_storage/helpers.rs

@@ -0,0 +1,122 @@
+use crate::services::storage::providers::google_cloud_storage::models::GoogleCloudStorageProviderConfig;
+use anyhow::{Context, Result};
+use bytes::Bytes;
+use futures::Stream;
+use futures::StreamExt;
+use google_cloud_auth::credentials::Credentials;
+use google_cloud_storage::client::Storage;
+use google_cloud_storage::streaming_source::{SizeHint, StreamingSource};
+use std::pin::Pin;
+
+pub fn build_credentials(cfg: &GoogleCloudStorageProviderConfig) -> Result<Credentials> {
+    // Service-account JSON stores the PEM with `\n` escape sequences. When the key is
+    // carried through config as a JSON string those can arrive as literal two-char `\n`
+    // sequences rather than real newlines, so the PEM parser finds no `-----BEGIN-----`
+    // line ("no items found"). Normalize them back to real newlines. A PEM that already
+    // has real newlines contains no literal `\n` pairs, so this is a no-op for it.
+    let private_key = cfg.private_key.replace("\\n", "\n");
+
+    let key = serde_json::json!({
+        "type": "service_account",
+        "project_id": cfg.project_id,
+        "client_email": cfg.client_email,
+        "private_key": private_key,
+        "private_key_id": "",
+        "token_uri": "https://oauth2.googleapis.com/token",
+        "universe_domain": "googleapis.com",
+    });
+
+    google_cloud_auth::credentials::service_account::Builder::new(key)
+        .build()
+        .context("failed to build GCS service account credentials")
+}
+
+pub async fn build_client(cfg: &GoogleCloudStorageProviderConfig) -> Result<Storage> {
+    let endpoint = cfg.api_endpoint.as_deref().filter(|s| !s.trim().is_empty());
+
+    // A custom endpoint means a local emulator (fake-gcs-server), which does not verify
+    // credentials. Use anonymous creds so a dummy/empty `private_key` in the emulator
+    // config doesn't trip the service-account PEM parser. Real GCS still uses the
+    // service-account key built from config.
+    let builder = if let Some(ep) = endpoint {
+        let creds = google_cloud_auth::credentials::anonymous::Builder::new().build();
+        Storage::builder()
+            .with_credentials(creds)
+            .with_endpoint(ep.to_string())
+    } else {
+        Storage::builder().with_credentials(build_credentials(cfg)?)
+    };
+
+    builder.build().await.context("failed to build GCS client")
+}
+
+/// Bridges `build_stream`'s `Send`-only byte stream into the SDK's `StreamingSource`
+/// (which `send_buffered` requires to be `Send + Sync + 'static`) via a bounded mpsc
+/// channel. Also reports an exact `size_hint`: the SDK picks single-shot vs resumable
+/// upload from `size_hint().upper()` — an unknown bound forces resumable unconditionally
+/// (see `upload_with_client`).
+pub struct StreamSource {
+    rx: tokio::sync::mpsc::Receiver<Result<Bytes, std::io::Error>>,
+    total_size: u64,
+}
+
+impl StreamSource {
+    pub fn from_stream(
+        mut stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
+        total_size: u64,
+    ) -> Self {
+        let (tx, rx) = tokio::sync::mpsc::channel(8);
+        tokio::spawn(async move {
+            while let Some(item) = stream.next().await {
+                if tx.send(item).await.is_err() {
+                    break;
+                }
+            }
+        });
+        StreamSource { rx, total_size }
+    }
+}
+
+impl StreamingSource for StreamSource {
+    type Error = std::io::Error;
+    async fn next(&mut self) -> Option<Result<Bytes, Self::Error>> {
+        self.rx.recv().await
+    }
+
+    // Report the exact size so the SDK can choose single-shot uploads. The default
+    // impl returns an unknown bound, which forces the resumable path unconditionally.
+    async fn size_hint(&self) -> Result<SizeHint, Self::Error> {
+        Ok(SizeHint::with_exact(self.total_size))
+    }
+}
+
+pub async fn upload_with_client(
+    client: &Storage,
+    bucket: &str,
+    object: &str,
+    source: StreamSource,
+    force_single_shot: bool,
+) -> Result<()> {
+    // `write_object` uses gRPC-style resource names: the bucket must be passed as
+    // `projects/_/buckets/<name>`, not the bare bucket id.
+    let bucket_resource = format!("projects/_/buckets/{bucket}");
+
+    let mut builder = client.write_object(bucket_resource, object, source);
+
+    // Resumable uploads follow a server-generated `Location` URL. When pointed at a
+    // custom `apiEndpoint` on a non-443 port, the SDK's transport drops the port from
+    // the `Host` header (google-cloud-gax-internal `host.rs`), so emulators that build
+    // the `Location` from `Host` hand back a portless URL the SDK then hangs on. A
+    // single-shot upload issues one request to the configured endpoint (no `Location`
+    // to follow), sidestepping the bug. We force it only for custom endpoints; against
+    // real GCS we keep resumable (bounded memory + resume on large backups).
+    if force_single_shot {
+        builder = builder.with_resumable_upload_threshold(usize::MAX);
+    }
+
+    builder
+        .send_buffered()
+        .await
+        .context("GCS write_object failed")?;
+    Ok(())
+}

src/services/storage/providers/google_cloud_storage/mod.rs

@@ -0,0 +1,149 @@
+pub mod helpers;
+mod models;
+
+use crate::core::context::Context;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::models::{BackupResult, UploadResult};
+use crate::services::storage::StorageProvider;
+use crate::services::storage::providers::google_cloud_storage::helpers::{
+    StreamSource, build_client, upload_with_client,
+};
+use crate::services::storage::providers::google_cloud_storage::models::GoogleCloudStorageProviderConfig;
+use crate::utils::common::BackupMethod;
+use crate::utils::file::{full_file_name, full_file_path};
+use crate::utils::stream::build_stream;
+use async_trait::async_trait;
+use std::sync::Arc;
+use tokio::fs;
+use tracing::{error, info};
+
+pub struct GoogleCloudStorageProvider {}
+
+#[async_trait]
+impl StorageProvider for GoogleCloudStorageProvider {
+    async fn upload(
+        &self,
+        ctx: Arc<Context>,
+        result: BackupResult,
+        _method: BackupMethod,
+        storage: &DatabaseStorage,
+        encrypt: Option<bool>,
+        _backup_storage_id: &str,
+    ) -> UploadResult {
+        let Some(file_path) = result.backup_file else {
+            return UploadResult {
+                storage_id: storage.id.clone(),
+                success: false,
+                error: Some("Missing backup file path".to_string()),
+                remote_file_path: None,
+                total_size: None,
+            };
+        };
+
+        let total_size = match fs::metadata(&file_path).await {
+            Ok(meta) => meta.len(),
+            Err(e) => {
+                error!("Failed to get file size: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let encrypt = encrypt.unwrap_or(false);
+
+        let upload = match build_stream(&file_path, encrypt, &ctx.edge_key.master_key_b64).await {
+            Ok(u) => u,
+            Err(e) => {
+                error!("Stream build failed: {}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let config: GoogleCloudStorageProviderConfig = match storage.clone().config.try_into() {
+            Ok(c) => c,
+            Err(e) => {
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let file_name = full_file_name(encrypt);
+        info!("Uploading file {}", file_name);
+        let remote_file_path = full_file_path(&file_name);
+
+        let client = match build_client(&config).await {
+            Ok(c) => c,
+            Err(e) => {
+                error!("GCS client build failed: {:?}", e);
+                return UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: None,
+                };
+            }
+        };
+
+        let source = StreamSource::from_stream(upload.stream, total_size);
+
+        // A custom apiEndpoint (self-hosted / emulator) on a non-443 port trips an
+        // upstream SDK bug in the resumable-upload path; force single-shot for it.
+        let force_single_shot = config
+            .api_endpoint
+            .as_deref()
+            .is_some_and(|s| !s.trim().is_empty());
+
+        info!(
+            "Starting GCS upload to {}/{} (single_shot={})",
+            config.bucket_name, remote_file_path, force_single_shot
+        );
+
+        match upload_with_client(
+            &client,
+            &config.bucket_name,
+            &remote_file_path,
+            source,
+            force_single_shot,
+        )
+        .await
+        {
+            Ok(_) => {
+                info!("GCS upload successful: {}", remote_file_path);
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: true,
+                    error: None,
+                    remote_file_path: Some(remote_file_path),
+                    total_size: Some(total_size),
+                }
+            }
+            Err(e) => {
+                error!("GCS upload failed: {:?}", e);
+                UploadResult {
+                    storage_id: storage.id.clone(),
+                    success: false,
+                    error: Some(e.to_string()),
+                    remote_file_path: None,
+                    total_size: Some(total_size),
+                }
+            }
+        }
+    }
+}

src/services/storage/providers/google_cloud_storage/models.rs

@@ -0,0 +1,11 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct GoogleCloudStorageProviderConfig {
+    pub project_id: String,
+    pub bucket_name: String,
+    pub client_email: String,
+    pub private_key: String,
+    #[serde(default)]
+    pub api_endpoint: Option<String>,
+}

src/services/storage/providers/google_drive/mod.rs

@@ -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 {

src/services/storage/providers/local.rs

@@ -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 {
@@ -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",

src/services/storage/providers/mod.rs

@@ -1,4 +1,5 @@
 pub mod azure_blob;
+pub mod google_cloud_storage;
 pub mod google_drive;
 pub mod local;
 pub mod s3;

src/services/storage/providers/s3/mod.rs

@@ -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 {

src/tests/domain/cluster/backup.rs

@@ -0,0 +1,53 @@
+use super::{env_for, start_cluster};
+use crate::domain::postgres::{cluster, connection};
+use crate::services::backup::logger::JobLogger;
+use crate::tests::init_tracing_for_test;
+use std::sync::Arc;
+use tempfile::TempDir;
+
+#[tokio::test]
+async fn produces_sql_with_roles_and_databases() {
+    init_tracing_for_test();
+    let (_c, cfg) = start_cluster("testuser").await;
+
+    let dir = TempDir::new().unwrap();
+    let logger = Arc::new(JobLogger::new());
+    let sql = cluster::backup::run(cfg.clone(), dir.path().to_path_buf(), env_for(&cfg), logger)
+        .await
+        .unwrap();
+
+    assert!(sql.is_file());
+    let contents = std::fs::read_to_string(&sql).unwrap();
+    assert!(contents.contains("CREATE ROLE"), "expected CREATE ROLE in dump");
+    assert!(
+        contents.contains("CREATE DATABASE") || contents.contains("\\connect"),
+        "expected database statements in dump"
+    );
+}
+
+#[tokio::test]
+async fn requires_superuser() {
+    init_tracing_for_test();
+    let (_c, super_cfg) = start_cluster("testuser").await;
+
+    // Create a NON-superuser login role on the cluster.
+    let client = connection::connect(&super_cfg).await.unwrap();
+    client
+        .batch_execute("CREATE ROLE appuser LOGIN PASSWORD 'changeme' NOSUPERUSER;")
+        .await
+        .unwrap();
+
+    let mut weak = super_cfg.clone();
+    weak.username = "appuser".to_string();
+
+    let dir = TempDir::new().unwrap();
+    let logger = Arc::new(JobLogger::new());
+    let err = cluster::backup::run(weak.clone(), dir.path().to_path_buf(), env_for(&weak), logger)
+        .await
+        .unwrap_err();
+
+    assert!(
+        err.to_string().contains("superuser"),
+        "expected a superuser error, got: {err}"
+    );
+}

src/tests/domain/cluster/database.rs

@@ -0,0 +1,33 @@
+use crate::domain::factory::DatabaseFactory;
+use crate::services::config::{DatabaseConfig, DbType};
+use std::path::Path;
+
+fn cluster_config() -> DatabaseConfig {
+    DatabaseConfig {
+        name: "cluster".to_string(),
+        database: "postgres".to_string(),
+        db_type: DbType::PostgresqlCluster,
+        username: "postgres".to_string(),
+        password: "changeme".to_string(),
+        port: 5432,
+        host: "localhost".to_string(),
+        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(),
+    }
+}
+
+#[tokio::test]
+async fn factory_routes_cluster_for_backup_with_sql_extension() {
+    let db = DatabaseFactory::create_for_backup(cluster_config()).await;
+    assert_eq!(db.file_extension(), ".sql");
+}
+
+#[tokio::test]
+async fn factory_routes_cluster_for_restore_with_sql_extension() {
+    let db = DatabaseFactory::create_for_restore(cluster_config(), Path::new("dump.sql")).await;
+    assert_eq!(db.file_extension(), ".sql");
+}

src/tests/domain/cluster/mod.rs

@@ -0,0 +1,50 @@
+mod backup;
+mod database;
+mod restore;
+
+use crate::services::config::{DatabaseConfig, DbType};
+use std::collections::HashMap;
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{ContainerAsync, ImageExt};
+use testcontainers_modules::postgres::Postgres;
+use url::Host;
+
+async fn start_cluster(user: &str) -> (ContainerAsync<Postgres>, DatabaseConfig) {
+    let container = Postgres::default()
+        .with_env_var("POSTGRES_DB", "postgres")
+        .with_env_var("POSTGRES_USER", user)
+        .with_env_var("POSTGRES_PASSWORD", "changeme")
+        .with_tag("17")
+        .start()
+        .await
+        .unwrap();
+
+    let host = container
+        .get_host()
+        .await
+        .unwrap_or(Host::parse("127.0.0.1").unwrap());
+    let port = container.get_host_port_ipv4(5432).await.unwrap_or(5432);
+
+    let config = DatabaseConfig {
+        name: format!("cluster-{}", user),
+        database: "postgres".to_string(),
+        db_type: DbType::PostgresqlCluster,
+        username: user.to_string(),
+        password: "changeme".to_string(),
+        port,
+        host: host.to_string(),
+        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)
+}
+
+fn env_for(cfg: &DatabaseConfig) -> HashMap<String, String> {
+    let mut env = std::env::vars().collect::<HashMap<_, _>>();
+    env.insert("PGPASSWORD".to_string(), cfg.password.clone());
+    env
+}

src/tests/domain/cluster/restore.rs

@@ -0,0 +1,83 @@
+use super::{env_for, start_cluster};
+use crate::domain::postgres::{cluster, connection};
+use crate::services::backup::logger::JobLogger;
+use crate::tests::init_tracing_for_test;
+use std::sync::Arc;
+use tempfile::TempDir;
+
+#[tokio::test]
+async fn backup_restore_round_trip_preserves_ownership() {
+    init_tracing_for_test();
+
+    // Source cluster A: seed a role + a table owned by that role.
+    let (_a, src) = start_cluster("testuser").await;
+    let client = connection::connect(&src).await.unwrap();
+    client
+        .batch_execute(
+            "CREATE ROLE appowner LOGIN PASSWORD 'changeme' NOSUPERUSER;\n\
+             CREATE TABLE owned_tbl (id int);\n\
+             ALTER TABLE owned_tbl OWNER TO appowner;",
+        )
+        .await
+        .unwrap();
+
+    let dir = TempDir::new().unwrap();
+    let sql = cluster::backup::run(src.clone(), dir.path().to_path_buf(), env_for(&src), Arc::new(JobLogger::new()))
+        .await
+        .unwrap();
+
+    // Target cluster B: fresh, same bootstrap user.
+    let (_b, mut dst) = start_cluster("testuser").await;
+
+    cluster::restore::run(dst.clone(), sql.clone(), env_for(&dst), Arc::new(JobLogger::new()))
+        .await
+        .unwrap();
+
+    // Verify the seeded role exists and the table's owner was preserved on B.
+    dst.database = "postgres".to_string();
+    let bclient = connection::connect(&dst).await.unwrap();
+    let role_exists: bool = bclient
+        .query_one("SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'appowner');", &[])
+        .await
+        .unwrap()
+        .get(0);
+    assert!(role_exists, "appowner role must be recreated on the target");
+
+    let owner: String = bclient
+        .query_one(
+            "SELECT tableowner FROM pg_tables WHERE tablename = 'owned_tbl';",
+            &[],
+        )
+        .await
+        .unwrap()
+        .get(0);
+    assert_eq!(owner, "appowner", "table ownership must be preserved");
+}
+
+#[tokio::test]
+async fn requires_superuser() {
+    init_tracing_for_test();
+    let (_c, super_cfg) = start_cluster("testuser").await;
+
+    // A non-superuser login role must be rejected before psql runs.
+    let client = connection::connect(&super_cfg).await.unwrap();
+    client
+        .batch_execute("CREATE ROLE appuser LOGIN PASSWORD 'changeme' NOSUPERUSER;")
+        .await
+        .unwrap();
+
+    let mut weak = super_cfg.clone();
+    weak.username = "appuser".to_string();
+
+    // The superuser pre-check happens before the dump file is read, so a
+    // non-existent restore path is fine — it must never be touched.
+    let missing = std::path::PathBuf::from("/nonexistent/cluster.sql");
+    let err = cluster::restore::run(weak.clone(), missing, env_for(&weak), Arc::new(JobLogger::new()))
+        .await
+        .unwrap_err();
+
+    assert!(
+        err.to_string().contains("superuser"),
+        "expected a superuser error, got: {err}"
+    );
+}

src/tests/domain/docker_volume.rs

@@ -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();
+}

src/tests/domain/firebird.rs

@@ -40,6 +40,9 @@ 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(),
     };
 
     (container, config)

src/tests/domain/mariadb.rs

@@ -32,6 +32,9 @@ 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(),
     };
 
     (container, config)

src/tests/domain/mod.rs

@@ -2,7 +2,9 @@ mod mariadb;
 mod mongodb;
 mod mysql;
 mod postgres;
+mod cluster;
 mod redis;
 mod valkey;
 mod firebird;
 mod mssql;
+mod docker_volume;

src/tests/domain/mongodb.rs

@@ -30,6 +30,9 @@ 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(),
     };
 
     (container, config)

src/tests/domain/mssql.rs

@@ -55,6 +55,9 @@ 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(),
     }
 }
 

src/tests/domain/mysql.rs

@@ -32,6 +32,9 @@ 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(),
     };
 
     (container, config)

src/tests/domain/postgres.rs

@@ -39,6 +39,9 @@ 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(),
     };
 
     (container, config)
@@ -56,6 +59,20 @@ async fn postgres_ping_test() {
     assert_eq!(reachable, true);
 }
 
+#[tokio::test]
+async fn is_superuser_detects_superuser_role() {
+    init_tracing_for_test();
+
+    // The testcontainer's POSTGRES_USER ("testuser") is the bootstrap superuser.
+    let (_container, config) = create_config().await;
+
+    let is_super = crate::domain::postgres::connection::is_superuser(&config)
+        .await
+        .unwrap();
+
+    assert!(is_super);
+}
+
 #[tokio::test]
 async fn postgres_backup_restore_test() {
     init_tracing_for_test();
@@ -142,6 +159,9 @@ 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(),
     };
 
     let db = DatabaseFactory::create_for_backup(config.clone()).await;
@@ -152,7 +172,8 @@ async fn postgres_password_with_slash_test() {
 
 mod select_pg_path_tests {
     use crate::domain::postgres::connection::{
-        pg_dump_binary_name, pg_dump_exists_in, select_pg_path_with,
+        pg_dump_binary_name, pg_dump_exists_in, pg_dumpall_binary_name, psql_binary_name,
+        select_pg_path_with,
     };
 
     // `select_pg_path_with` takes the `PG_BIN_DIR` override as a plain
@@ -209,4 +230,24 @@ mod select_pg_path_tests {
         let dir = std::path::Path::new("this/path/almost-certainly/does-not-exist-12345");
         assert!(!pg_dump_exists_in(dir));
     }
+
+    #[test]
+    fn pg_dumpall_binary_name_is_platform_specific() {
+        let name = pg_dumpall_binary_name();
+        if cfg!(target_os = "windows") {
+            assert_eq!(name, "pg_dumpall.exe");
+        } else {
+            assert_eq!(name, "pg_dumpall");
+        }
+    }
+
+    #[test]
+    fn psql_binary_name_is_platform_specific() {
+        let name = psql_binary_name();
+        if cfg!(target_os = "windows") {
+            assert_eq!(name, "psql.exe");
+        } else {
+            assert_eq!(name, "psql");
+        }
+    }
 }

src/tests/domain/redis.rs

@@ -29,6 +29,9 @@ 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(),
     };
 
     (container, config)

src/tests/domain/valkey.rs

@@ -28,6 +28,9 @@ 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(),
     };
 
     (container, config)

src/tests/services/config_tests.rs

@@ -0,0 +1,266 @@
+use crate::core::context::Context;
+use crate::services::api::ApiClient;
+use crate::services::config::ConfigService;
+use crate::utils::edge_key::EdgeKey;
+use std::io::Write;
+use std::sync::Arc;
+use tempfile::NamedTempFile;
+
+// `ConfigService::load` never touches `self.ctx` on the `Some(file_path)` path,
+// so the values here don't matter — but `Context::new()` panics without an
+// `EDGE_KEY` env var, so build the struct directly (mirrors
+// backup_uploader_tests.rs's `ctx_pointing_at`).
+fn test_context() -> Arc<Context> {
+    Arc::new(Context {
+        edge_key: EdgeKey {
+            server_url: String::new(),
+            agent_id: "agent-1".to_string(),
+            master_key_b64: String::new(),
+        },
+        api: ApiClient::new(String::new()),
+    })
+}
+
+fn write_json(contents: &str) -> NamedTempFile {
+    let mut file = NamedTempFile::with_suffix(".json").unwrap();
+    file.write_all(contents.as_bytes()).unwrap();
+    file
+}
+
+#[test]
+fn parses_postgresql_cluster_type() {
+    let file = write_json(
+        r#"{
+            "databases": [
+                {
+                    "name": "cluster1",
+                    "type": "postgresql-cluster",
+                    "username": "postgres",
+                    "password": "p",
+                    "port": 5432,
+                    "host": "localhost",
+                    "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].db_type.as_str(), "postgresql-cluster");
+    // `database` is optional for cluster entries and defaults to "postgres".
+    assert_eq!(cfg.databases[0].database, "postgres");
+}
+
+#[test]
+fn postgresql_cluster_respects_explicit_database() {
+    let file = write_json(
+        r#"{
+            "databases": [
+                {
+                    "name": "cluster1",
+                    "type": "postgresql-cluster",
+                    "database": "maintenance",
+                    "username": "postgres",
+                    "password": "p",
+                    "port": 5432,
+                    "host": "localhost",
+                    "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].database, "maintenance");
+}
+
+#[test]
+fn postgresql_options_keep_ownership_parses() {
+    let file = write_json(
+        r#"{
+            "databases": [
+                {
+                    "name": "db1",
+                    "type": "postgresql",
+                    "username": "u",
+                    "password": "p",
+                    "port": 5432,
+                    "host": "localhost",
+                    "database": "mydb",
+                    "generated_id": "16678159-ff7e-4c97-8c83-0adeff214681",
+                    "options": {
+                        "keep_ownership": true
+                    }
+                }
+            ]
+        }"#,
+    );
+
+    let service = ConfigService::new(test_context());
+    let cfg = service.load(Some(file.path().to_str().unwrap())).unwrap();
+
+    let keep = cfg.databases[0]
+        .options
+        .get("keep_ownership")
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false);
+
+    assert!(keep);
+}
+
+#[test]
+fn postgresql_options_absent_defaults_to_empty() {
+    let file = write_json(
+        r#"{
+            "databases": [
+                {
+                    "name": "db1",
+                    "type": "postgresql",
+                    "username": "u",
+                    "password": "p",
+                    "port": 5432,
+                    "host": "localhost",
+                    "database": "mydb",
+                    "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!(cfg.databases[0].options.is_empty());
+}
+
+#[test]
+fn postgresql_options_non_bool_keep_ownership_falls_back_to_false() {
+    let file = write_json(
+        r#"{
+            "databases": [
+                {
+                    "name": "db1",
+                    "type": "postgresql",
+                    "username": "u",
+                    "password": "p",
+                    "port": 5432,
+                    "host": "localhost",
+                    "database": "mydb",
+                    "generated_id": "16678159-ff7e-4c97-8c83-0adeff214681",
+                    "options": {
+                        "keep_ownership": "yes"
+                    }
+                }
+            ]
+        }"#,
+    );
+
+    let service = ConfigService::new(test_context());
+    let cfg = service.load(Some(file.path().to_str().unwrap())).unwrap();
+
+    let keep = cfg.databases[0]
+        .options
+        .get("keep_ownership")
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false);
+
+    assert!(!keep);
+}
+
+#[test]
+fn keep_ownership_extraction_logic() {
+    use serde_json::Value;
+    use std::collections::HashMap;
+
+    // true → keep ownership
+    let mut opts: HashMap<String, Value> = HashMap::new();
+    opts.insert("keep_ownership".to_string(), Value::Bool(true));
+    let keep = opts.get("keep_ownership").and_then(|v| v.as_bool()).unwrap_or(false);
+    assert!(keep, "should keep ownership when flag is true");
+
+    // false → strip
+    let mut opts2: HashMap<String, Value> = HashMap::new();
+    opts2.insert("keep_ownership".to_string(), Value::Bool(false));
+    let keep2 = opts2.get("keep_ownership").and_then(|v| v.as_bool()).unwrap_or(false);
+    assert!(!keep2, "should strip when flag is false");
+
+    // missing → strip
+    let opts3: HashMap<String, Value> = HashMap::new();
+    let keep3 = opts3.get("keep_ownership").and_then(|v| v.as_bool()).unwrap_or(false);
+    assert!(!keep3, "should strip when key absent");
+
+    // wrong type → strip
+    let mut opts4: HashMap<String, Value> = HashMap::new();
+    opts4.insert("keep_ownership".to_string(), Value::String("yes".to_string()));
+    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}");
+}

src/tests/services/mod.rs

@@ -1,2 +1,3 @@
 mod api_models_tests;
 mod backup_uploader_tests;
+mod config_tests;

src/tests/storage/azure_blob.rs

@@ -16,9 +16,7 @@ use testcontainers::runners::AsyncRunner;
 use testcontainers::{GenericImage, ImageExt};
 use url::Url;
 
-/// Build an Account SAS query set (test-only). Azurite cannot authorize container-create with
-/// a container-scoped Service SAS, so tests create the target container with an Account SAS.
-/// Reuses the production HMAC primitive (`hmac_sha256_b64`) to avoid duplicating signing logic.
+
 fn build_account_sas(
     resolved: &ResolvedAzure,
     services: &str,
@@ -33,12 +31,9 @@ fn build_account_sas(
     let signed_expiry = (Utc::now() + Duration::hours(1))
         .format("%Y-%m-%dT%H:%M:%SZ")
         .to_string();
-    let signed_protocol = "https,http"; // Azurite is http
+    let signed_protocol = "https,http";
     let signed_ip = String::new();
     let encryption_scope = String::new();
-
-    // Account SAS string-to-sign for sv >= 2020-12-06:
-    // account \n sp \n ss \n srt \n st \n se \n sip \n spr \n sv \n ses \n  (trailing newline)
     let string_to_sign = format!(
         "{acc}\n{sp}\n{ss}\n{srt}\n{st}\n{se}\n{sip}\n{spr}\n{sv}\n{ses}\n",
         acc = resolved.account_name, sp = permissions, ss = services, srt = resource_types,
@@ -59,7 +54,6 @@ fn build_account_sas(
     ])
 }
 
-/// Build an Account-SAS-scoped URL for a container (test-only container creation).
 fn build_account_sas_container_url(
     resolved: &ResolvedAzure,
     container: &str,
@@ -86,21 +80,12 @@ const AZURITE_KEY: &str =
 async fn start_azurite() -> (testcontainers::ContainerAsync<GenericImage>, ResolvedAzure) {
     let container = GenericImage::new("mcr.microsoft.com/azure-storage/azurite", "latest")
         .with_exposed_port(10000.tcp())
-        // The current `latest` image logs (on stdout):
-        //   "Azurite Blob service successfully listens on http://0.0.0.0:10000"
-        // Older builds phrased it "...is successfully listening"; this substring matches
-        // the wording the pulled image actually emits.
         .with_wait_for(WaitFor::message_on_stdout(
             "Azurite Blob service successfully listens on",
         ))
-        // The GA SDK sends a very recent `x-ms-version`; Azurite 3.35 rejects unknown
-        // versions unless we tell it to skip that check.
         .with_cmd(["azurite-blob", "--blobHost", "0.0.0.0", "--skipApiVersionCheck"])
         .start().await.unwrap();
-    // Use the testcontainers-resolved host (not a hardcoded 127.0.0.1): under
-    // docker-out-of-docker / remote daemons the published port is not on the test
-    // process's loopback. All other container tests (mssql, valkey, postgres, ...)
-    // already do this; azure_blob was the only one hardcoding the host.
+
     let host = container.get_host().await.unwrap().to_string();
     let port = container.get_host_port_ipv4(10000).await.unwrap();
     let resolved = ResolvedAzure {
@@ -118,8 +103,7 @@ async fn spike_sas_block_roundtrip_against_azurite() {
     let container = "portabase";
     let blob = "spike/hello.txt";
 
-    // Container creation must use an Account SAS (service=blob, resource-type=container,
-    // perms=create+write). Azurite cannot authorize container-create with a Service SAS.
+
     let container_url =
         build_account_sas_container_url(&resolved, container, "b", "c", "cw").unwrap();
     let container_client =
@@ -135,9 +119,6 @@ async fn spike_sas_block_roundtrip_against_azurite() {
     bbc.stage_block(&raw_id, payload.len() as u64, RequestContent::from(payload.to_vec()), None)
         .await.unwrap();
 
-    // `BlockLookupList.latest` is `Option<Vec<Vec<u8>>>` and base64-encodes each entry
-    // internally during XML serialization, exactly as `stage_block` base64-encodes the
-    // `blockid` query. So `latest` must hold the SAME RAW id bytes passed to `stage_block`.
     let block_list = BlockLookupList { latest: Some(vec![raw_id.clone()]), ..Default::default() };
     bbc.commit_block_list(block_list.try_into().unwrap(), None).await.unwrap();
 
@@ -146,6 +127,85 @@ async fn spike_sas_block_roundtrip_against_azurite() {
     assert!(read_client.exists().await.unwrap());
 }
 
+mod resolve {
+    use crate::services::storage::providers::azure_blob::models::{
+        AzureBlobProviderConfig, ensure_account_in_endpoint,
+    };
+
+    const AZURITE_KEY: &str =
+        "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+    const CONNECTION_STRING: &str = "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://localhost:10000/devstoreaccount1;QueueEndpoint=http://localhost:10001/devstoreaccount1;TableEndpoint=http://localhost:10002/devstoreaccount1;";
+
+
+    #[test]
+    fn resolve_connection_string_mode() {
+        let cfg = AzureBlobProviderConfig {
+            account_name: String::new(),
+            account_key: String::new(),
+            container_name: "portabase".into(),
+            auth_mode: Some("connectionString".into()),
+            connection_string: CONNECTION_STRING.into(),
+            endpoint_url: None,
+        };
+        let r = cfg.resolve().unwrap();
+        assert_eq!(r.account_name, "devstoreaccount1");
+        assert_eq!(r.account_key, AZURITE_KEY);
+        assert_eq!(r.blob_endpoint, "http://localhost:10000/devstoreaccount1");
+    }
+
+
+    #[test]
+    fn resolve_account_key_mode_injects_account_path() {
+        let cfg = AzureBlobProviderConfig {
+            account_name: "devstoreaccount1".into(),
+            account_key: AZURITE_KEY.into(),
+            container_name: "portabase".into(),
+            auth_mode: Some("accountKey".into()),
+            connection_string: CONNECTION_STRING.into(),
+            endpoint_url: Some("http://localhost:10000".into()),
+        };
+        let r = cfg.resolve().unwrap();
+        assert_eq!(r.account_name, "devstoreaccount1");
+        assert_eq!(r.account_key, AZURITE_KEY);
+        assert_eq!(r.blob_endpoint, "http://localhost:10000/devstoreaccount1");
+    }
+
+    #[test]
+    fn resolve_implicit_connection_string() {
+        let cfg = AzureBlobProviderConfig {
+            account_name: String::new(),
+            account_key: String::new(),
+            container_name: "portabase".into(),
+            auth_mode: None,
+            connection_string: CONNECTION_STRING.into(),
+            endpoint_url: None,
+        };
+        let r = cfg.resolve().unwrap();
+        assert_eq!(r.blob_endpoint, "http://localhost:10000/devstoreaccount1");
+    }
+
+    #[test]
+    fn resolve_account_key_default_endpoint() {
+        let cfg = AzureBlobProviderConfig {
+            account_name: "myaccount".into(),
+            account_key: AZURITE_KEY.into(),
+            container_name: "portabase".into(),
+            auth_mode: Some("accountKey".into()),
+            connection_string: String::new(),
+            endpoint_url: None,
+        };
+        let r = cfg.resolve().unwrap();
+        assert_eq!(r.blob_endpoint, "https://myaccount.blob.core.windows.net");
+    }
+
+    #[test]
+    fn ensure_account_keeps_host_style_endpoint() {
+        let got =
+            ensure_account_in_endpoint("https://myaccount.blob.core.windows.net", "myaccount");
+        assert_eq!(got, "https://myaccount.blob.core.windows.net");
+    }
+}
+
 #[tokio::test]
 async fn upload_stream_multi_block_roundtrip() {
     init_tracing_for_test();
@@ -156,7 +216,6 @@ async fn upload_stream_multi_block_roundtrip() {
     let container = "portabase";
     let blob = "backups/multi.bin";
 
-    // Container setup (provider itself never creates it): Account SAS create.
     let container_url =
         build_account_sas_container_url(&resolved, container, "b", "c", "cw").unwrap();
     azure_storage_blob::clients::BlobContainerClient::new(container_url, None, None)
@@ -165,7 +224,6 @@ async fn upload_stream_multi_block_roundtrip() {
         .await
         .unwrap();
 
-    // 10 KiB fed as 1 KiB chunks, forced into 4 KiB blocks => 3 blocks (multi-block path).
     let data = vec![7u8; 10 * 1024];
     let chunks: Vec<Result<Bytes, std::io::Error>> = data
         .chunks(1024)
@@ -177,7 +235,6 @@ async fn upload_stream_multi_block_roundtrip() {
         .await
         .unwrap();
 
-    // Verify the committed blob reassembles to the exact source bytes via a read-SAS GET.
     let read_url = build_sas_url(&resolved, container, blob, SasResource::Blob, "r").unwrap();
     let got = reqwest::get(read_url).await.unwrap().bytes().await.unwrap();
     assert_eq!(got.as_ref(), data.as_slice());

src/tests/storage/google_cloud_storage.rs

@@ -0,0 +1,99 @@
+use crate::services::storage::providers::google_cloud_storage::helpers::{
+    StreamSource, upload_with_client,
+};
+use crate::tests::init_tracing_for_test;
+
+use bytes::Bytes;
+use futures::stream;
+use google_cloud_storage::client::Storage;
+use testcontainers::core::{IntoContainerPort, WaitFor};
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{GenericImage, ImageExt};
+
+const BUCKET: &str = "portabase";
+
+async fn start_fake_gcs() -> (testcontainers::ContainerAsync<GenericImage>, String) {
+    // Natural random host port (no port-80 pin). The provider forces a single-shot
+    // upload for custom endpoints, which issues one request to this endpoint and never
+    // follows a server-built `Location` — so it works on any port, unlike the resumable
+    // path that the SDK's Host-header port-drop bug breaks on non-443 ports.
+    let container = GenericImage::new("fsouza/fake-gcs-server", "latest")
+        .with_exposed_port(4443.tcp())
+        .with_wait_for(WaitFor::message_on_stderr("server started at"))
+        .with_cmd(["-scheme", "http", "-backend", "memory", "-port", "4443"])
+        .start()
+        .await
+        .unwrap();
+
+    let host = container.get_host().await.unwrap().to_string();
+    let port = container.get_host_port_ipv4(4443).await.unwrap();
+    let endpoint = format!("http://{host}:{port}");
+    (container, endpoint)
+}
+
+async fn anon_client(endpoint: &str) -> Storage {
+    let creds = google_cloud_auth::credentials::anonymous::Builder::new().build();
+    Storage::builder()
+        .with_credentials(creds)
+        .with_endpoint(endpoint.to_string())
+        .build()
+        .await
+        .unwrap()
+}
+
+async fn create_bucket(endpoint: &str) {
+    let url = format!("{endpoint}/storage/v1/b?project=test-project");
+    let res = reqwest::Client::new()
+        .post(&url)
+        .json(&serde_json::json!({ "name": BUCKET }))
+        .send()
+        .await
+        .unwrap();
+    assert!(
+        res.status().is_success(),
+        "bucket create failed: {}",
+        res.status()
+    );
+}
+
+#[tokio::test]
+async fn upload_stream_roundtrip_against_fake_gcs() {
+    init_tracing_for_test();
+    let (_container, endpoint) = start_fake_gcs().await;
+    create_bucket(&endpoint).await;
+
+    let object = "backups/multi.bin";
+
+    // 10 KiB fed as 1 KiB chunks -> multi-chunk streaming path.
+    let data = vec![7u8; 10 * 1024];
+    let chunks: Vec<Result<Bytes, std::io::Error>> = data
+        .chunks(1024)
+        .map(|c| Ok(Bytes::copy_from_slice(c)))
+        .collect();
+    let source = StreamSource::from_stream(Box::pin(stream::iter(chunks)), data.len() as u64);
+
+    let client = anon_client(&endpoint).await;
+
+    // force_single_shot = true (custom endpoint). Guard with a timeout so a regression
+    // into the resumable path (which would hang forever on this non-443 port) fails the
+    // test instead of stalling it.
+    tokio::time::timeout(
+        std::time::Duration::from_secs(60),
+        upload_with_client(&client, BUCKET, object, source, true),
+    )
+    .await
+    .expect("upload hung (regressed to resumable path on a non-443 endpoint?)")
+    .unwrap();
+
+    let read_url = format!(
+        "{endpoint}/storage/v1/b/{BUCKET}/o/{}?alt=media",
+        object.replace('/', "%2F")
+    );
+    let got = reqwest::get(&read_url)
+        .await
+        .unwrap()
+        .bytes()
+        .await
+        .unwrap();
+    assert_eq!(got.as_ref(), data.as_slice());
+}

src/tests/storage/mod.rs

@@ -1 +1,2 @@
 mod azure_blob;
+mod google_cloud_storage;

src/tests/utils/common_tests.rs

@@ -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"));
+}

src/tests/utils/compress_tests.rs

@@ -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(())
+}

src/utils/common.rs

@@ -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 {

src/utils/compress.rs

@@ -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(())
+}
+