feat(admin): add configurable S3 backups

apps/aether-gateway/src/backup/config.rs

@@ -0,0 +1,398 @@
+use std::fmt;
+
+use serde_json::{Map, Value};
+
+use super::schedule::{BackupSchedule, BackupScheduleUnit};
+use super::scopes::BackupScope;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct S3BackupConfig {
+    pub(crate) enabled: bool,
+    pub(crate) scope: BackupScope,
+    pub(crate) endpoint: String,
+    pub(crate) region: String,
+    pub(crate) bucket: String,
+    pub(crate) prefix: String,
+    pub(crate) access_key_id: String,
+    pub(crate) secret_access_key: String,
+    pub(crate) path_style: bool,
+    pub(crate) compression: String,
+    pub(crate) schedule: BackupSchedule,
+    pub(crate) retention_count: u32,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct BackupConfigError {
+    message: String,
+}
+
+impl BackupConfigError {
+    fn new(message: impl Into<String>) -> Self {
+        Self {
+            message: message.into(),
+        }
+    }
+}
+
+impl fmt::Display for BackupConfigError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.message)
+    }
+}
+
+impl std::error::Error for BackupConfigError {}
+
+impl S3BackupConfig {
+    pub(crate) fn from_json_map(entries: &Map<String, Value>) -> Result<Self, BackupConfigError> {
+        let enabled = optional_bool(entries, "backup_s3_enabled")?.unwrap_or(false);
+        let mut schedule = BackupSchedule::default();
+        schedule.unit = optional_string(entries, "backup_s3_schedule_unit")?
+            .map(|value| {
+                BackupScheduleUnit::from_config_value(&value)
+                    .ok_or_else(|| BackupConfigError::new("Schedule Unit（计划单位）配置值无效"))
+            })
+            .transpose()?
+            .unwrap_or(schedule.unit);
+        schedule.interval =
+            optional_u32(entries, "backup_s3_schedule_interval")?.unwrap_or(schedule.interval);
+        schedule.minute =
+            optional_u32(entries, "backup_s3_schedule_minute")?.unwrap_or(schedule.minute);
+        schedule.hour = optional_u32(entries, "backup_s3_schedule_hour")?.unwrap_or(schedule.hour);
+        schedule.weekday =
+            optional_u32(entries, "backup_s3_schedule_weekday")?.unwrap_or(schedule.weekday);
+        schedule.month_day =
+            optional_u32(entries, "backup_s3_schedule_month_day")?.unwrap_or(schedule.month_day);
+        validate_range("Interval（计划间隔）", schedule.interval, 1, u32::MAX)?;
+        validate_range("Minute（计划分钟）", schedule.minute, 0, 59)?;
+        validate_range("Hour（计划小时）", schedule.hour, 0, 23)?;
+        validate_range("Weekday（计划星期）", schedule.weekday, 1, 7)?;
+        validate_range("Month Day（计划月日）", schedule.month_day, 1, 31)?;
+
+        let scope = optional_string(entries, "backup_s3_scope")?
+            .map(|value| {
+                BackupScope::from_config_value(&value)
+                    .ok_or_else(|| BackupConfigError::new("Scope（备份范围）配置值无效"))
+            })
+            .transpose()?
+            .unwrap_or(BackupScope::Data);
+        let retention_count = optional_u32(entries, "backup_s3_retention_count")?.unwrap_or(7);
+        validate_range("Retention（保留份数）", retention_count, 1, u32::MAX)?;
+        let endpoint = required_or_disabled_string(
+            entries,
+            "backup_s3_endpoint",
+            "Endpoint（S3 地址）",
+            enabled,
+        )?;
+        let bucket =
+            required_or_disabled_string(entries, "backup_s3_bucket", "Bucket（存储桶）", enabled)?;
+        let access_key_id = required_or_disabled_string(
+            entries,
+            "backup_s3_access_key_id",
+            "Access Key ID（访问密钥 ID）",
+            enabled,
+        )?;
+        let secret_access_key = required_or_disabled_string(
+            entries,
+            "backup_s3_secret_access_key",
+            "Secret Access Key（访问密钥）",
+            enabled,
+        )?;
+
+        Ok(Self {
+            enabled,
+            scope,
+            endpoint,
+            region: optional_string(entries, "backup_s3_region")?
+                .unwrap_or_else(|| "auto".to_string()),
+            bucket,
+            prefix: optional_string(entries, "backup_s3_prefix")?
+                .unwrap_or_else(|| "aether/backups/".to_string()),
+            access_key_id,
+            secret_access_key,
+            path_style: optional_bool(entries, "backup_s3_path_style")?.unwrap_or(true),
+            compression: optional_string(entries, "backup_s3_compression")?
+                .unwrap_or_else(|| "zstd".to_string()),
+            schedule,
+            retention_count,
+        })
+    }
+}
+
+fn validate_range(label: &str, value: u32, min: u32, max: u32) -> Result<(), BackupConfigError> {
+    if (min..=max).contains(&value) {
+        Ok(())
+    } else {
+        Err(BackupConfigError::new(format!(
+            "{label}配置值无效，应在 {min}..={max} 范围内"
+        )))
+    }
+}
+
+fn required_or_disabled_string(
+    entries: &Map<String, Value>,
+    key: &str,
+    label: &str,
+    enabled: bool,
+) -> Result<String, BackupConfigError> {
+    if enabled {
+        required_string(entries, key, label)
+    } else {
+        Ok(optional_string(entries, key)?.unwrap_or_default())
+    }
+}
+
+fn required_string(
+    entries: &Map<String, Value>,
+    key: &str,
+    label: &str,
+) -> Result<String, BackupConfigError> {
+    optional_string(entries, key)?
+        .filter(|value| !value.is_empty())
+        .ok_or_else(|| BackupConfigError::new(format!("{label}为必填配置")))
+}
+
+fn optional_string(
+    entries: &Map<String, Value>,
+    key: &str,
+) -> Result<Option<String>, BackupConfigError> {
+    let Some(value) = entries.get(key) else {
+        return Ok(None);
+    };
+    match value {
+        Value::String(value) => {
+            let trimmed = value.trim();
+            if trimmed.is_empty() {
+                Ok(None)
+            } else {
+                Ok(Some(trimmed.to_string()))
+            }
+        }
+        Value::Null => Ok(None),
+        _ => Err(BackupConfigError::new(format!(
+            "{} 字符串值无效",
+            config_label(key)
+        ))),
+    }
+}
+
+fn config_label(key: &str) -> &str {
+    match key {
+        "backup_s3_endpoint" => "Endpoint（S3 地址）",
+        "backup_s3_region" => "Region（S3 区域）",
+        "backup_s3_bucket" => "Bucket（存储桶）",
+        "backup_s3_prefix" => "Prefix（备份前缀）",
+        "backup_s3_access_key_id" => "Access Key ID（访问密钥 ID）",
+        "backup_s3_secret_access_key" => "Secret Access Key（访问密钥）",
+        "backup_s3_compression" => "Compression（压缩格式）",
+        "backup_s3_scope" => "Scope（备份范围）",
+        "backup_s3_schedule_unit" => "Schedule Unit（计划单位）",
+        _ => key,
+    }
+}
+
+fn optional_bool(
+    entries: &Map<String, Value>,
+    key: &str,
+) -> Result<Option<bool>, BackupConfigError> {
+    let Some(value) = entries.get(key) else {
+        return Ok(None);
+    };
+    match value {
+        Value::Bool(value) => Ok(Some(*value)),
+        Value::String(value) => match value.trim() {
+            "true" => Ok(Some(true)),
+            "false" => Ok(Some(false)),
+            "" => Ok(None),
+            _ => Err(BackupConfigError::new(format!("{key} 布尔值无效"))),
+        },
+        Value::Null => Ok(None),
+        _ => Err(BackupConfigError::new(format!("{key} 布尔值无效"))),
+    }
+}
+
+fn optional_u32(entries: &Map<String, Value>, key: &str) -> Result<Option<u32>, BackupConfigError> {
+    let Some(value) = entries.get(key) else {
+        return Ok(None);
+    };
+    match value {
+        Value::Number(value) => value
+            .as_u64()
+            .and_then(|value| u32::try_from(value).ok())
+            .map(Some)
+            .ok_or_else(|| BackupConfigError::new(format!("{key} 数值无效"))),
+        Value::String(value) => {
+            let trimmed = value.trim();
+            if trimmed.is_empty() {
+                Ok(None)
+            } else {
+                trimmed
+                    .parse::<u32>()
+                    .map(Some)
+                    .map_err(|_| BackupConfigError::new(format!("{key} 数值无效")))
+            }
+        }
+        Value::Null => Ok(None),
+        _ => Err(BackupConfigError::new(format!("{key} 数值无效"))),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::schedule::BackupScheduleUnit;
+    use super::super::scopes::BackupScope;
+    use super::S3BackupConfig;
+
+    #[test]
+    fn parses_minimal_valid_s3_backup_config() {
+        let entries = serde_json::json!({
+            "backup_s3_enabled": true,
+            "backup_s3_scope": "data",
+            "backup_s3_endpoint": "https://s3.example.com",
+            "backup_s3_region": "auto",
+            "backup_s3_bucket": "aether-backups",
+            "backup_s3_prefix": "prod/",
+            "backup_s3_access_key_id": "access",
+            "backup_s3_secret_access_key": "secret",
+            "backup_s3_path_style": true,
+            "backup_s3_compression": "zstd",
+            "backup_s3_schedule_unit": "days",
+            "backup_s3_schedule_interval": 1,
+            "backup_s3_schedule_hour": 3,
+            "backup_s3_schedule_minute": 15,
+            "backup_s3_retention_count": 7
+        });
+
+        let config = S3BackupConfig::from_json_map(entries.as_object().unwrap())
+            .expect("config should parse");
+
+        assert_eq!(config.scope, BackupScope::Data);
+        assert_eq!(config.bucket, "aether-backups");
+        assert_eq!(config.prefix, "prod/");
+        assert_eq!(config.schedule.unit, BackupScheduleUnit::Days);
+        assert_eq!(config.retention_count, 7);
+    }
+
+    #[test]
+    fn rejects_missing_bucket_for_backup() {
+        let entries = serde_json::json!({
+            "backup_s3_enabled": true,
+            "backup_s3_endpoint": "https://s3.example.com",
+            "backup_s3_access_key_id": "access",
+            "backup_s3_secret_access_key": "secret"
+        });
+
+        let err = S3BackupConfig::from_json_map(entries.as_object().unwrap())
+            .expect_err("bucket is required");
+
+        assert!(err.to_string().contains("Bucket"));
+    }
+
+    #[test]
+    fn parses_disabled_default_s3_backup_config_with_null_credentials() {
+        let entries = serde_json::json!({
+            "backup_s3_enabled": false,
+            "backup_s3_scope": "data",
+            "backup_s3_endpoint": null,
+            "backup_s3_region": "auto",
+            "backup_s3_bucket": null,
+            "backup_s3_prefix": "aether/backups/",
+            "backup_s3_access_key_id": null,
+            "backup_s3_secret_access_key": null,
+            "backup_s3_path_style": true,
+            "backup_s3_compression": "zstd",
+            "backup_s3_schedule_unit": "days",
+            "backup_s3_schedule_interval": 1,
+            "backup_s3_schedule_hour": 3,
+            "backup_s3_schedule_minute": 0,
+            "backup_s3_schedule_weekday": 1,
+            "backup_s3_schedule_month_day": 1,
+            "backup_s3_retention_count": 7
+        });
+
+        let config = S3BackupConfig::from_json_map(entries.as_object().unwrap())
+            .expect("disabled default config should parse");
+
+        assert_eq!(config.enabled, false);
+        assert_eq!(config.endpoint, "");
+        assert_eq!(config.bucket, "");
+        assert_eq!(config.access_key_id, "");
+        assert_eq!(config.secret_access_key, "");
+        assert_eq!(config.schedule.unit, BackupScheduleUnit::Days);
+    }
+
+    #[test]
+    fn rejects_invalid_schedule_numbers() {
+        let cases = [
+            ("backup_s3_schedule_interval", 0, "Interval"),
+            ("backup_s3_schedule_minute", 60, "Minute"),
+            ("backup_s3_schedule_hour", 24, "Hour"),
+            ("backup_s3_schedule_weekday", 0, "Weekday"),
+            ("backup_s3_schedule_month_day", 32, "Month Day"),
+            ("backup_s3_retention_count", 0, "Retention"),
+        ];
+
+        for (key, value, label) in cases {
+            let mut entries = serde_json::json!({
+                "backup_s3_enabled": true,
+                "backup_s3_endpoint": "https://s3.example.com",
+                "backup_s3_bucket": "aether-backups",
+                "backup_s3_access_key_id": "access",
+                "backup_s3_secret_access_key": "secret"
+            });
+            entries.as_object_mut().unwrap().insert(
+                key.to_string(),
+                serde_json::Value::Number(serde_json::Number::from(value)),
+            );
+
+            let err = S3BackupConfig::from_json_map(entries.as_object().unwrap())
+                .expect_err("invalid numeric config should fail");
+
+            assert!(
+                err.to_string().contains(label),
+                "{key} should mention {label}, got {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn rejects_non_string_endpoint_config() {
+        let entries = serde_json::json!({
+            "backup_s3_enabled": true,
+            "backup_s3_endpoint": {"url": "https://s3.example.com"},
+            "backup_s3_bucket": "aether-backups",
+            "backup_s3_access_key_id": "access",
+            "backup_s3_secret_access_key": "secret"
+        });
+
+        let err = S3BackupConfig::from_json_map(entries.as_object().unwrap())
+            .expect_err("endpoint object should fail");
+
+        assert!(err.to_string().contains("Endpoint"));
+    }
+
+    #[test]
+    fn applies_default_values_from_system_config_contract() {
+        let entries = serde_json::json!({
+            "backup_s3_endpoint": "https://s3.example.com",
+            "backup_s3_bucket": "aether-backups",
+            "backup_s3_access_key_id": "access",
+            "backup_s3_secret_access_key": "secret"
+        });
+
+        let config = S3BackupConfig::from_json_map(entries.as_object().unwrap())
+            .expect("config should parse with defaults");
+
+        assert_eq!(config.scope, BackupScope::Data);
+        assert_eq!(config.region, "auto");
+        assert_eq!(config.prefix, "aether/backups/");
+        assert_eq!(config.path_style, true);
+        assert_eq!(config.compression, "zstd");
+        assert_eq!(config.schedule.unit, BackupScheduleUnit::Days);
+        assert_eq!(config.schedule.interval, 1);
+        assert_eq!(config.schedule.hour, 3);
+        assert_eq!(config.schedule.minute, 0);
+        assert_eq!(config.schedule.weekday, 1);
+        assert_eq!(config.schedule.month_day, 1);
+        assert_eq!(config.retention_count, 7);
+    }
+}

apps/aether-gateway/src/backup/executor.rs

@@ -0,0 +1,224 @@
+use bytes::Bytes;
+use chrono::{DateTime, SecondsFormat, Utc};
+use serde_json::Value;
+use sha2::{Digest, Sha256};
+
+use super::config::S3BackupConfig;
+use super::scopes::BackupScope;
+use super::store::{BackupObjectStore, BackupStoreError};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct BackupRunResult {
+    pub(crate) scope: BackupScope,
+    pub(crate) bucket: String,
+    pub(crate) object_key: String,
+    pub(crate) bytes: usize,
+    pub(crate) sha256: String,
+    pub(crate) export_version: String,
+    pub(crate) exported_at: String,
+    pub(crate) compression: String,
+    pub(crate) deleted_old_objects: usize,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum BackupExecutionError {
+    #[error("S3 backup JSON serialization failed: {0}")]
+    Json(#[from] serde_json::Error),
+
+    #[error("S3 backup compression failed: {0}")]
+    Compression(#[from] std::io::Error),
+
+    #[error("{0}")]
+    Store(#[from] BackupStoreError),
+
+    #[error("S3 backup compression `{0}` is not supported; expected `zstd`")]
+    UnknownCompression(String),
+}
+
+pub(crate) async fn run_backup_with_store<S>(
+    config: &S3BackupConfig,
+    store: &S,
+    payload: Value,
+    now_utc: DateTime<Utc>,
+) -> Result<BackupRunResult, BackupExecutionError>
+where
+    S: BackupObjectStore + ?Sized,
+{
+    let export_version = payload_string_field(&payload, "version").unwrap_or_default();
+    let exported_at = payload_string_field(&payload, "exported_at")
+        .unwrap_or_else(|| now_utc.to_rfc3339_opts(SecondsFormat::Secs, true));
+    let json_bytes = serde_json::to_vec(&payload)?;
+    let compression = config.compression.trim().to_string();
+    let upload_bytes = match compression.as_str() {
+        "zstd" => zstd::stream::encode_all(json_bytes.as_slice(), 0)?,
+        other => return Err(BackupExecutionError::UnknownCompression(other.to_string())),
+    };
+    let bytes = upload_bytes.len();
+    let sha256 = format!("{:x}", Sha256::digest(&upload_bytes));
+    let timestamp = now_utc.format("%Y%m%d-%H%M%S").to_string();
+    let object_key = config.scope.object_key(&config.prefix, &timestamp);
+
+    store
+        .put_object(&object_key, Bytes::from(upload_bytes))
+        .await?;
+    let deleted_old_objects = prune_old_backups(config, store, &object_key).await?;
+
+    Ok(BackupRunResult {
+        scope: config.scope,
+        bucket: config.bucket.clone(),
+        object_key,
+        bytes,
+        sha256,
+        export_version,
+        exported_at,
+        compression,
+        deleted_old_objects,
+    })
+}
+
+async fn prune_old_backups<S>(
+    config: &S3BackupConfig,
+    store: &S,
+    current_object_key: &str,
+) -> Result<usize, BackupExecutionError>
+where
+    S: BackupObjectStore + ?Sized,
+{
+    let keys = store.list_keys(&config.prefix).await?;
+    let mut matching_keys = config.scope.matching_backup_keys(&config.prefix, keys);
+    matching_keys.sort_by(|left, right| right.cmp(left));
+
+    let mut deleted = 0;
+    let mut retained = usize::from(
+        config.retention_count > 0 && matching_keys.iter().any(|key| key == current_object_key),
+    );
+    for key in matching_keys {
+        if config.retention_count > 0 && key == current_object_key {
+            continue;
+        }
+        if retained < config.retention_count as usize {
+            retained += 1;
+            continue;
+        }
+
+        store.delete_object(&key).await?;
+        deleted += 1;
+    }
+
+    Ok(deleted)
+}
+
+fn payload_string_field(payload: &Value, field: &str) -> Option<String> {
+    payload
+        .get(field)
+        .and_then(Value::as_str)
+        .map(ToString::to_string)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::config::S3BackupConfig;
+    use super::super::schedule::BackupSchedule;
+    use super::super::scopes::BackupScope;
+    use super::super::store::{BackupObjectStore, FakeBackupObjectStore};
+    use super::run_backup_with_store;
+    use bytes::Bytes;
+    use chrono::{DateTime, Utc};
+    use serde_json::json;
+
+    fn sample_backup_config(scope: BackupScope, retention_count: u32) -> S3BackupConfig {
+        S3BackupConfig {
+            enabled: true,
+            scope,
+            endpoint: "https://example.com".to_string(),
+            region: "auto".to_string(),
+            bucket: "aether-backups".to_string(),
+            prefix: "prod/".to_string(),
+            access_key_id: "test-access-key".to_string(),
+            secret_access_key: "test-secret-key".to_string(),
+            path_style: true,
+            compression: "zstd".to_string(),
+            schedule: BackupSchedule::default(),
+            retention_count,
+        }
+    }
+
+    #[tokio::test]
+    async fn backup_executor_uploads_payload_and_prunes_same_scope_only() {
+        let store = FakeBackupObjectStore::default();
+        store
+            .put_object(
+                "prod/aether-data-backup-20260524-010000.json.zst",
+                Bytes::from_static(b"old"),
+            )
+            .await
+            .unwrap();
+        store
+            .put_object(
+                "prod/aether-config-backup-20260524-010000.json.zst",
+                Bytes::from_static(b"keep-config"),
+            )
+            .await
+            .unwrap();
+
+        let config = sample_backup_config(BackupScope::Data, 1);
+        let payload = json!({
+            "version": "1.0",
+            "exported_at": "2026-05-24T03:15:00Z",
+            "config_data": {},
+            "user_data": {}
+        });
+        let now_utc = DateTime::parse_from_rfc3339("2026-05-24T03:15:00+08:00")
+            .unwrap()
+            .with_timezone(&Utc);
+
+        let result = run_backup_with_store(&config, &store, payload, now_utc)
+            .await
+            .expect("backup should succeed");
+
+        assert_eq!(result.scope, BackupScope::Data);
+        assert_eq!(result.bucket, "aether-backups");
+        assert_eq!(
+            result.object_key,
+            "prod/aether-data-backup-20260523-191500.json.zst"
+        );
+        assert!(result.bytes > 0);
+        assert_eq!(result.sha256.len(), 64);
+        assert_eq!(result.export_version, "1.0");
+        assert_eq!(result.exported_at, "2026-05-24T03:15:00Z");
+        assert_eq!(result.compression, "zstd");
+        assert_eq!(result.deleted_old_objects, 1);
+
+        let keys = store.list_keys("prod/").await.unwrap();
+        assert!(keys
+            .iter()
+            .any(|key| key == "prod/aether-config-backup-20260524-010000.json.zst"));
+        assert!(keys
+            .iter()
+            .any(|key| key == "prod/aether-data-backup-20260523-191500.json.zst"));
+        assert!(!keys
+            .iter()
+            .any(|key| key == "prod/aether-data-backup-20260524-010000.json.zst"));
+    }
+
+    #[tokio::test]
+    async fn backup_executor_rejects_unknown_compression() {
+        let store = FakeBackupObjectStore::default();
+        let mut config = sample_backup_config(BackupScope::Data, 1);
+        config.compression = "brotli".to_string();
+
+        let payload = json!({
+            "version": "1.0",
+            "exported_at": "2026-05-24T03:15:00Z"
+        });
+        let now_utc = DateTime::parse_from_rfc3339("2026-05-24T03:15:00+08:00")
+            .unwrap()
+            .with_timezone(&Utc);
+
+        let error = run_backup_with_store(&config, &store, payload, now_utc)
+            .await
+            .expect_err("unknown compression should fail");
+
+        assert!(error.to_string().contains("brotli"));
+    }
+}

apps/aether-gateway/src/backup/mod.rs

@@ -0,0 +1,9 @@
+pub(crate) mod config;
+pub(crate) mod executor;
+pub(crate) mod schedule;
+pub(crate) mod scopes;
+pub(crate) mod store;
+pub(crate) mod task;
+pub(crate) mod worker;
+
+pub(crate) const S3_BACKUP_LAST_SLOT_KEY: &str = "backup_s3_last_slot";

apps/aether-gateway/src/backup/schedule.rs

@@ -0,0 +1,366 @@
+use chrono::{DateTime, Datelike, NaiveDate, TimeZone, Timelike, Utc};
+use chrono_tz::Tz;
+
+const BACKUP_SCHEDULE_DEFAULT_TIMEZONE: &str = "Asia/Shanghai";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum BackupScheduleUnit {
+    Hours,
+    Days,
+    Weeks,
+    Months,
+}
+
+impl BackupScheduleUnit {
+    pub(crate) fn from_config_value(value: &str) -> Option<Self> {
+        match value.trim() {
+            "hours" => Some(Self::Hours),
+            "days" => Some(Self::Days),
+            "weeks" => Some(Self::Weeks),
+            "months" => Some(Self::Months),
+            _ => None,
+        }
+    }
+
+    fn slot_prefix(self) -> &'static str {
+        match self {
+            Self::Hours => "hours",
+            Self::Days => "days",
+            Self::Weeks => "weeks",
+            Self::Months => "months",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct BackupSchedule {
+    pub(crate) unit: BackupScheduleUnit,
+    pub(crate) interval: u32,
+    pub(crate) minute: u32,
+    pub(crate) hour: u32,
+    pub(crate) weekday: u32,
+    pub(crate) month_day: u32,
+}
+
+impl Default for BackupSchedule {
+    fn default() -> Self {
+        Self {
+            unit: BackupScheduleUnit::Days,
+            interval: 1,
+            minute: 0,
+            hour: 3,
+            weekday: 1,
+            month_day: 1,
+        }
+    }
+}
+
+impl BackupSchedule {
+    pub(crate) fn due_slot(&self, now_utc: DateTime<Utc>) -> Option<String> {
+        let timezone = backup_schedule_timezone();
+        let local_now = now_utc.with_timezone(&timezone);
+        let interval = self.interval.max(1);
+        if local_now.minute() != self.minute {
+            return None;
+        }
+
+        let due = match self.unit {
+            BackupScheduleUnit::Hours => {
+                (local_epoch_hour(local_now.date_naive(), local_now.hour()) - i64::from(self.hour))
+                    .rem_euclid(i64::from(interval))
+                    == 0
+            }
+            BackupScheduleUnit::Days => {
+                local_now.hour() == self.hour
+                    && local_epoch_day(local_now.date_naive()) % i64::from(interval) == 0
+            }
+            BackupScheduleUnit::Weeks => {
+                local_now.hour() == self.hour
+                    && local_now.weekday().number_from_monday() == self.weekday
+                    && local_epoch_week(local_now.date_naive()) % i64::from(interval) == 0
+            }
+            BackupScheduleUnit::Months => {
+                local_now.hour() == self.hour
+                    && local_now.day() == self.month_day
+                    && month_ordinal(local_now.year(), local_now.month0()) % i64::from(interval)
+                        == 0
+            }
+        };
+        if !due {
+            return None;
+        }
+
+        let slot = Utc.from_utc_datetime(&now_utc.date_naive().and_hms_opt(
+            now_utc.hour(),
+            now_utc.minute(),
+            0,
+        )?);
+        Some(format!(
+            "{}:{}",
+            self.unit.slot_prefix(),
+            slot.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
+        ))
+    }
+}
+
+pub(crate) fn backup_schedule_timezone() -> Tz {
+    std::env::var("APP_TIMEZONE")
+        .ok()
+        .map(|value| value.trim().to_string())
+        .filter(|value| !value.is_empty())
+        .as_deref()
+        .unwrap_or(BACKUP_SCHEDULE_DEFAULT_TIMEZONE)
+        .parse()
+        .unwrap_or(chrono_tz::Asia::Shanghai)
+}
+
+fn local_epoch_day(date: NaiveDate) -> i64 {
+    let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("unix epoch date should be valid");
+    date.signed_duration_since(epoch).num_days()
+}
+
+fn local_epoch_week(date: NaiveDate) -> i64 {
+    local_epoch_day(date).div_euclid(7)
+}
+
+fn local_epoch_hour(date: NaiveDate, hour: u32) -> i64 {
+    local_epoch_day(date) * 24 + i64::from(hour)
+}
+
+fn month_ordinal(year: i32, month0: u32) -> i64 {
+    i64::from(year) * 12 + i64::from(month0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{BackupSchedule, BackupScheduleUnit};
+    use std::ffi::OsString;
+    use std::sync::{Mutex, OnceLock};
+
+    struct TimezoneEnvGuard {
+        previous: Option<OsString>,
+    }
+
+    impl TimezoneEnvGuard {
+        fn set(value: Option<&str>) -> Self {
+            let previous = std::env::var_os("APP_TIMEZONE");
+            match value {
+                Some(value) => std::env::set_var("APP_TIMEZONE", value),
+                None => std::env::remove_var("APP_TIMEZONE"),
+            }
+            Self { previous }
+        }
+    }
+
+    impl Drop for TimezoneEnvGuard {
+        fn drop(&mut self) {
+            match &self.previous {
+                Some(value) => std::env::set_var("APP_TIMEZONE", value),
+                None => std::env::remove_var("APP_TIMEZONE"),
+            }
+        }
+    }
+
+    fn timezone_env_lock() -> std::sync::MutexGuard<'static, ()> {
+        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+        LOCK.get_or_init(|| Mutex::new(()))
+            .lock()
+            .unwrap_or_else(|err| err.into_inner())
+    }
+
+    #[test]
+    fn hourly_schedule_returns_stable_slot_once_per_due_hour() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(None);
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Hours,
+            interval: 6,
+            minute: 10,
+            hour: 0,
+            weekday: 1,
+            month_day: 1,
+        };
+        let now = chrono::DateTime::parse_from_rfc3339("2026-05-24T12:10:30+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(
+            schedule.due_slot(now).as_deref(),
+            Some("hours:2026-05-24T04:10:00Z")
+        );
+    }
+
+    #[test]
+    fn hourly_schedule_interval_does_not_reset_at_midnight() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(None);
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Hours,
+            interval: 5,
+            minute: 10,
+            hour: 0,
+            weekday: 1,
+            month_day: 1,
+        };
+        let midnight = chrono::DateTime::parse_from_rfc3339("2026-05-25T00:10:30+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+        let due_after_midnight = chrono::DateTime::parse_from_rfc3339("2026-05-25T03:10:30+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(schedule.due_slot(midnight), None);
+        assert_eq!(
+            schedule.due_slot(due_after_midnight).as_deref(),
+            Some("hours:2026-05-24T19:10:00Z")
+        );
+    }
+
+    #[test]
+    fn hourly_schedule_supports_intervals_longer_than_one_day() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(None);
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Hours,
+            interval: 25,
+            minute: 10,
+            hour: 0,
+            weekday: 1,
+            month_day: 1,
+        };
+        let daily_midnight = chrono::DateTime::parse_from_rfc3339("2026-05-25T00:10:30+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+        let due_after_25_hours = chrono::DateTime::parse_from_rfc3339("2026-05-25T23:10:30+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(schedule.due_slot(daily_midnight), None);
+        assert_eq!(
+            schedule.due_slot(due_after_25_hours).as_deref(),
+            Some("hours:2026-05-25T15:10:00Z")
+        );
+    }
+
+    #[test]
+    fn daily_schedule_uses_maintenance_timezone_for_hour_and_interval() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(None);
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Days,
+            interval: 2,
+            minute: 15,
+            hour: 3,
+            weekday: 1,
+            month_day: 1,
+        };
+        let due = chrono::DateTime::parse_from_rfc3339("2026-05-23T03:15:45+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+        let not_due = chrono::DateTime::parse_from_rfc3339("2026-05-24T03:15:45+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(
+            schedule.due_slot(due).as_deref(),
+            Some("days:2026-05-22T19:15:00Z")
+        );
+        assert_eq!(schedule.due_slot(not_due), None);
+    }
+
+    #[test]
+    fn weekly_schedule_uses_maintenance_timezone_for_weekday_and_interval() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(None);
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Weeks,
+            interval: 2,
+            minute: 30,
+            hour: 5,
+            weekday: 1,
+            month_day: 1,
+        };
+        let due = chrono::DateTime::parse_from_rfc3339("2026-05-25T05:30:59+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+        let wrong_week = chrono::DateTime::parse_from_rfc3339("2026-05-18T05:30:59+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(
+            schedule.due_slot(due).as_deref(),
+            Some("weeks:2026-05-24T21:30:00Z")
+        );
+        assert_eq!(schedule.due_slot(wrong_week), None);
+    }
+
+    #[test]
+    fn monthly_schedule_uses_maintenance_timezone_for_month_day_and_interval() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(None);
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Months,
+            interval: 3,
+            minute: 45,
+            hour: 2,
+            weekday: 1,
+            month_day: 1,
+        };
+        let due = chrono::DateTime::parse_from_rfc3339("2026-04-01T02:45:01+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+        let wrong_month = chrono::DateTime::parse_from_rfc3339("2026-05-01T02:45:01+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(
+            schedule.due_slot(due).as_deref(),
+            Some("months:2026-03-31T18:45:00Z")
+        );
+        assert_eq!(schedule.due_slot(wrong_month), None);
+    }
+
+    #[test]
+    fn hourly_schedule_uses_utc_when_app_timezone_is_utc() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(Some("UTC"));
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Hours,
+            interval: 6,
+            minute: 10,
+            hour: 4,
+            weekday: 1,
+            month_day: 1,
+        };
+        let now = chrono::DateTime::parse_from_rfc3339("2026-05-24T04:10:30Z")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(
+            schedule.due_slot(now).as_deref(),
+            Some("hours:2026-05-24T04:10:00Z")
+        );
+    }
+
+    #[test]
+    fn daily_schedule_slot_uses_actual_instant_for_non_whole_hour_timezone() {
+        let _guard = timezone_env_lock();
+        let _env = TimezoneEnvGuard::set(Some("Asia/Kolkata"));
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Days,
+            interval: 1,
+            minute: 30,
+            hour: 3,
+            weekday: 1,
+            month_day: 1,
+        };
+        let now = chrono::DateTime::parse_from_rfc3339("2026-05-24T03:30:45+05:30")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        assert_eq!(
+            schedule.due_slot(now).as_deref(),
+            Some("days:2026-05-23T22:00:00Z")
+        );
+    }
+}

apps/aether-gateway/src/backup/scopes.rs

@@ -0,0 +1,211 @@
+use std::fmt;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum BackupScope {
+    Config,
+    Users,
+    Data,
+}
+
+impl BackupScope {
+    pub(crate) fn from_config_value(value: &str) -> Option<Self> {
+        match value.trim() {
+            "config" => Some(Self::Config),
+            "users" => Some(Self::Users),
+            "data" => Some(Self::Data),
+            _ => None,
+        }
+    }
+
+    pub(crate) fn as_config_value(self) -> &'static str {
+        match self {
+            Self::Config => "config",
+            Self::Users => "users",
+            Self::Data => "data",
+        }
+    }
+
+    pub(crate) fn route_kind(self) -> &'static str {
+        match self {
+            Self::Config => "config_export",
+            Self::Users => "users_export",
+            Self::Data => "data_export",
+        }
+    }
+
+    pub(crate) fn file_stem(self) -> &'static str {
+        match self {
+            Self::Config => "aether-config-backup",
+            Self::Users => "aether-users-backup",
+            Self::Data => "aether-data-backup",
+        }
+    }
+
+    pub(crate) fn object_key(self, prefix: &str, timestamp: &str) -> String {
+        let file_name = self.file_name(timestamp);
+        let prefix = normalized_prefix(prefix);
+
+        if prefix.is_empty() {
+            file_name
+        } else {
+            format!("{prefix}/{file_name}")
+        }
+    }
+
+    pub(crate) fn matching_backup_keys(
+        self,
+        prefix: &str,
+        keys: impl IntoIterator<Item = String>,
+    ) -> Vec<String> {
+        let normalized_prefix = normalized_prefix(prefix);
+        let expected_prefix = if normalized_prefix.is_empty() {
+            String::new()
+        } else {
+            format!("{normalized_prefix}/")
+        };
+        let file_prefix = format!("{}-", self.file_stem());
+        let file_suffix = ".json.zst";
+
+        keys.into_iter()
+            .filter(|key| {
+                let Some(file_name) = key.strip_prefix(&expected_prefix) else {
+                    return false;
+                };
+                if file_name.contains('/') {
+                    return false;
+                }
+                let Some(timestamp) = file_name
+                    .strip_prefix(&file_prefix)
+                    .and_then(|rest| rest.strip_suffix(file_suffix))
+                else {
+                    return false;
+                };
+
+                is_aether_backup_timestamp(timestamp)
+            })
+            .collect()
+    }
+
+    fn file_name(self, timestamp: &str) -> String {
+        format!("{}-{timestamp}.json.zst", self.file_stem())
+    }
+}
+
+impl fmt::Display for BackupScope {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(self.as_config_value())
+    }
+}
+
+fn normalized_prefix(prefix: &str) -> &str {
+    prefix.trim_end_matches('/')
+}
+
+fn is_aether_backup_timestamp(timestamp: &str) -> bool {
+    let bytes = timestamp.as_bytes();
+
+    bytes.len() == 15
+        && bytes[8] == b'-'
+        && bytes[..8].iter().all(|byte| byte.is_ascii_digit())
+        && bytes[9..].iter().all(|byte| byte.is_ascii_digit())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::BackupScope;
+
+    #[test]
+    fn backup_scope_matches_export_routes_and_object_prefixes() {
+        assert_eq!(BackupScope::Config.as_config_value(), "config");
+        assert_eq!(BackupScope::Users.as_config_value(), "users");
+        assert_eq!(BackupScope::Data.as_config_value(), "data");
+
+        assert_eq!(BackupScope::Config.route_kind(), "config_export");
+        assert_eq!(BackupScope::Users.route_kind(), "users_export");
+        assert_eq!(BackupScope::Data.route_kind(), "data_export");
+
+        assert_eq!(BackupScope::Config.file_stem(), "aether-config-backup");
+        assert_eq!(BackupScope::Users.file_stem(), "aether-users-backup");
+        assert_eq!(BackupScope::Data.file_stem(), "aether-data-backup");
+
+        assert_eq!(
+            BackupScope::Config.object_key("prod/", "20260524-031500"),
+            "prod/aether-config-backup-20260524-031500.json.zst"
+        );
+        assert_eq!(
+            BackupScope::Users.object_key("prod/", "20260524-031500"),
+            "prod/aether-users-backup-20260524-031500.json.zst"
+        );
+        assert_eq!(
+            BackupScope::Data.object_key("prod/", "20260524-031500"),
+            "prod/aether-data-backup-20260524-031500.json.zst"
+        );
+    }
+
+    #[test]
+    fn retention_filter_only_matches_same_scope() {
+        let keys = vec![
+            "prod/aether-config-backup-20260524-010000.json.zst".to_string(),
+            "prod/aether-users-backup-20260524-010000.json.zst".to_string(),
+            "prod/aether-data-backup-20260524-010000.json.zst".to_string(),
+            "prod/random.json.zst".to_string(),
+        ];
+
+        let matched = BackupScope::Users.matching_backup_keys("prod/", keys);
+
+        assert_eq!(
+            matched,
+            vec!["prod/aether-users-backup-20260524-010000.json.zst"]
+        );
+    }
+
+    #[test]
+    fn retention_filter_requires_aether_timestamp_format() {
+        let keys = vec![
+            "prod/aether-users-backup-20260524-010000.json.zst".to_string(),
+            "prod/aether-users-backup-foo.json.zst".to_string(),
+            "prod/aether-users-backup-2026052-010000.json.zst".to_string(),
+            "prod/aether-users-backup-202605240-010000.json.zst".to_string(),
+            "prod/aether-users-backup-20260524-01000.json.zst".to_string(),
+            "prod/aether-users-backup-20260524-0100000.json.zst".to_string(),
+            "prod/aether-users-backup-20260524010000.json.zst".to_string(),
+            "prod/aether-users-backup-2026052a-010000.json.zst".to_string(),
+            "prod/aether-users-backup-20260524-01000x.json.zst".to_string(),
+        ];
+
+        let matched = BackupScope::Users.matching_backup_keys("prod/", keys);
+
+        assert_eq!(
+            matched,
+            vec!["prod/aether-users-backup-20260524-010000.json.zst"]
+        );
+    }
+
+    #[test]
+    fn backup_key_prefix_boundaries_are_exact() {
+        assert_eq!(
+            BackupScope::Config.object_key("", "20260524-031500"),
+            "aether-config-backup-20260524-031500.json.zst"
+        );
+        assert_eq!(
+            BackupScope::Config.object_key("prod", "20260524-031500"),
+            "prod/aether-config-backup-20260524-031500.json.zst"
+        );
+
+        let keys = vec![
+            "prod/aether-config-backup-20260524-010000.json.zst".to_string(),
+            "prod//aether-config-backup-20260524-010000.json.zst".to_string(),
+            "prod-backups/aether-config-backup-20260524-010000.json.zst".to_string(),
+            "prod/aether-config-backup-20260524-010000.json".to_string(),
+            "prod/aether-config-backup-.json.zst".to_string(),
+            "aether-config-backup-20260524-010000.json.zst".to_string(),
+        ];
+
+        let matched = BackupScope::Config.matching_backup_keys("prod", keys);
+
+        assert_eq!(
+            matched,
+            vec!["prod/aether-config-backup-20260524-010000.json.zst"]
+        );
+    }
+}

apps/aether-gateway/src/backup/store.rs

@@ -0,0 +1,227 @@
+use std::collections::BTreeMap;
+use std::fmt;
+use std::sync::Arc;
+
+use bytes::Bytes;
+use futures_util::TryStreamExt;
+use object_store::aws::AmazonS3Builder;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::sync::RwLock;
+
+use super::config::S3BackupConfig;
+
+#[async_trait::async_trait]
+pub(crate) trait BackupObjectStore: Send + Sync {
+    async fn put_object(&self, key: &str, bytes: Bytes) -> Result<(), BackupStoreError>;
+
+    async fn list_keys(&self, prefix: &str) -> Result<Vec<String>, BackupStoreError>;
+
+    async fn delete_object(&self, key: &str) -> Result<(), BackupStoreError>;
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct BackupStoreError {
+    message: String,
+}
+
+impl BackupStoreError {
+    fn new(message: impl Into<String>) -> Self {
+        Self {
+            message: message.into(),
+        }
+    }
+
+    fn object_store(operation: &str, key: &str, error: impl fmt::Display) -> Self {
+        Self::new(format!(
+            "S3 backup object store {operation} failed for `{key}`: {error}"
+        ))
+    }
+}
+
+impl fmt::Display for BackupStoreError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.message)
+    }
+}
+
+impl std::error::Error for BackupStoreError {}
+
+#[derive(Debug, Default, Clone)]
+pub(crate) struct FakeBackupObjectStore {
+    objects: Arc<RwLock<BTreeMap<String, Bytes>>>,
+}
+
+#[async_trait::async_trait]
+impl BackupObjectStore for FakeBackupObjectStore {
+    async fn put_object(&self, key: &str, bytes: Bytes) -> Result<(), BackupStoreError> {
+        self.objects.write().await.insert(key.to_string(), bytes);
+        Ok(())
+    }
+
+    async fn list_keys(&self, prefix: &str) -> Result<Vec<String>, BackupStoreError> {
+        let prefix = directory_list_prefix(prefix);
+        Ok(self
+            .objects
+            .read()
+            .await
+            .keys()
+            .filter(|key| key.starts_with(&prefix))
+            .cloned()
+            .collect())
+    }
+
+    async fn delete_object(&self, key: &str) -> Result<(), BackupStoreError> {
+        self.objects.write().await.remove(key);
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct ObjectStoreS3BackupStore {
+    store: object_store::aws::AmazonS3,
+}
+
+impl ObjectStoreS3BackupStore {
+    pub(crate) fn from_config(config: &S3BackupConfig) -> Result<Self, BackupStoreError> {
+        let store = AmazonS3Builder::new()
+            .with_endpoint(config.endpoint.clone())
+            .with_region(config.region.clone())
+            .with_bucket_name(config.bucket.clone())
+            .with_access_key_id(config.access_key_id.clone())
+            .with_secret_access_key(config.secret_access_key.clone())
+            .with_virtual_hosted_style_request(!config.path_style)
+            .build()
+            .map_err(|error| {
+                BackupStoreError::new(format!(
+                    "S3 backup object store configuration failed: {error}"
+                ))
+            })?;
+
+        Ok(Self { store })
+    }
+}
+
+#[async_trait::async_trait]
+impl BackupObjectStore for ObjectStoreS3BackupStore {
+    async fn put_object(&self, key: &str, bytes: Bytes) -> Result<(), BackupStoreError> {
+        self.store
+            .put(&Path::from(key), bytes.into())
+            .await
+            .map(|_| ())
+            .map_err(|error| BackupStoreError::object_store("put", key, error))
+    }
+
+    async fn list_keys(&self, prefix: &str) -> Result<Vec<String>, BackupStoreError> {
+        let prefix_path = list_prefix_path(prefix);
+        let mut keys = self
+            .store
+            .list(prefix_path.as_ref())
+            .map_ok(|meta| meta.location.to_string())
+            .try_collect::<Vec<_>>()
+            .await
+            .map_err(|error| BackupStoreError::object_store("list", prefix, error))?;
+        keys.sort();
+        Ok(keys)
+    }
+
+    async fn delete_object(&self, key: &str) -> Result<(), BackupStoreError> {
+        self.store
+            .delete(&Path::from(key))
+            .await
+            .map_err(|error| BackupStoreError::object_store("delete", key, error))
+    }
+}
+
+fn directory_list_prefix(prefix: &str) -> String {
+    let prefix = prefix.trim_end_matches('/');
+    if prefix.is_empty() {
+        String::new()
+    } else {
+        format!("{prefix}/")
+    }
+}
+
+fn list_prefix_path(prefix: &str) -> Option<Path> {
+    let prefix = prefix.trim_end_matches('/');
+    if prefix.is_empty() {
+        None
+    } else {
+        Some(Path::from(prefix))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{list_prefix_path, BackupObjectStore, FakeBackupObjectStore};
+
+    #[tokio::test]
+    async fn fake_backup_object_store_puts_lists_and_deletes() {
+        let store = FakeBackupObjectStore::default();
+        store
+            .put_object(
+                "prod/aether-data-backup-20260524-010000.json.zst",
+                bytes::Bytes::from_static(b"one"),
+            )
+            .await
+            .unwrap();
+        store
+            .put_object(
+                "prod/aether-data-backup-20260524-020000.json.zst",
+                bytes::Bytes::from_static(b"two"),
+            )
+            .await
+            .unwrap();
+
+        let keys = store.list_keys("prod/").await.unwrap();
+        assert_eq!(keys.len(), 2);
+
+        store
+            .delete_object("prod/aether-data-backup-20260524-010000.json.zst")
+            .await
+            .unwrap();
+        let keys = store.list_keys("prod/").await.unwrap();
+        assert_eq!(
+            keys,
+            vec!["prod/aether-data-backup-20260524-020000.json.zst"]
+        );
+    }
+
+    #[tokio::test]
+    async fn fake_backup_object_store_lists_normalized_directory_prefixes() {
+        let store = FakeBackupObjectStore::default();
+        store
+            .put_object(
+                "prod/aether-data-backup-20260524-010000.json.zst",
+                bytes::Bytes::from_static(b"one"),
+            )
+            .await
+            .unwrap();
+        store
+            .put_object(
+                "prod-backups/aether-data-backup-20260524-010000.json.zst",
+                bytes::Bytes::from_static(b"two"),
+            )
+            .await
+            .unwrap();
+
+        let keys = store.list_keys("prod").await.unwrap();
+
+        assert_eq!(
+            keys,
+            vec!["prod/aether-data-backup-20260524-010000.json.zst"]
+        );
+    }
+
+    #[test]
+    fn s3_list_prefix_path_lets_object_store_add_directory_delimiter() {
+        assert_eq!(
+            list_prefix_path("prod/")
+                .as_ref()
+                .map(std::string::ToString::to_string)
+                .as_deref(),
+            Some("prod")
+        );
+        assert!(list_prefix_path("").is_none());
+    }
+}

apps/aether-gateway/src/backup/task.rs

@@ -0,0 +1,821 @@
+use std::fmt;
+use std::time::Duration;
+
+use aether_admin::system::admin_system_config_default_value;
+use aether_data_contracts::repository::background_tasks::{
+    BackgroundTaskKind, BackgroundTaskListQuery, BackgroundTaskStatus, StoredBackgroundTaskRun,
+    UpsertBackgroundTaskRun,
+};
+use aether_runtime_state::RuntimeLockLease;
+use axum::http::StatusCode;
+use chrono::Utc;
+use futures_util::FutureExt;
+use serde::Serialize;
+use serde_json::{json, Map, Value};
+use tracing::warn;
+
+use super::config::S3BackupConfig;
+use super::executor::{run_backup_with_store, BackupRunResult};
+use super::scopes::BackupScope;
+use super::store::ObjectStoreS3BackupStore;
+use crate::admin_api::AdminAppState;
+use crate::handlers::shared::decrypt_catalog_secret_with_fallbacks;
+use crate::task_runtime::{
+    append_event_with_logging, build_task_run_id, now_unix_secs, spawn_fire_and_forget,
+    task_definition, update_run_status, upsert_run_with_logging, TASK_KEY_SYSTEM_S3_BACKUP,
+};
+use crate::{AppState, GatewayError};
+
+const S3_BACKUP_CONFIG_KEYS: &[&str] = &[
+    "backup_s3_enabled",
+    "backup_s3_scope",
+    "backup_s3_endpoint",
+    "backup_s3_region",
+    "backup_s3_bucket",
+    "backup_s3_prefix",
+    "backup_s3_access_key_id",
+    "backup_s3_secret_access_key",
+    "backup_s3_path_style",
+    "backup_s3_compression",
+    "backup_s3_schedule_unit",
+    "backup_s3_schedule_interval",
+    "backup_s3_schedule_minute",
+    "backup_s3_schedule_hour",
+    "backup_s3_schedule_weekday",
+    "backup_s3_schedule_month_day",
+    "backup_s3_retention_count",
+];
+
+const S3_BACKUP_QUEUED_MESSAGE: &str = "S3 备份任务已提交";
+const S3_BACKUP_TASK_LOCK_KEY: &str = "task_runtime:lock:system.s3.backup";
+const S3_BACKUP_TASK_LOCK_TTL: Duration = Duration::from_secs(60 * 60 * 6);
+const S3_BACKUP_TASK_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(60 * 5);
+const S3_BACKUP_ACTIVE_TASK_STALE_AFTER_SECS: u64 = 60 * 60 * 6;
+
+#[derive(Debug, Clone, Serialize)]
+pub(crate) struct S3BackupTaskStart {
+    pub(crate) id: String,
+    pub(crate) task_key: &'static str,
+    pub(crate) status: &'static str,
+    pub(crate) progress_message: &'static str,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct S3BackupTaskError {
+    status: StatusCode,
+    detail: String,
+}
+
+impl S3BackupTaskError {
+    fn bad_request(detail: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::BAD_REQUEST,
+            detail: detail.into(),
+        }
+    }
+
+    fn internal(detail: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::INTERNAL_SERVER_ERROR,
+            detail: detail.into(),
+        }
+    }
+
+    fn service_unavailable(detail: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::SERVICE_UNAVAILABLE,
+            detail: detail.into(),
+        }
+    }
+
+    fn conflict(detail: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::CONFLICT,
+            detail: detail.into(),
+        }
+    }
+
+    pub(crate) fn status(&self) -> StatusCode {
+        self.status
+    }
+
+    pub(crate) fn detail(&self) -> &str {
+        &self.detail
+    }
+}
+
+impl fmt::Display for S3BackupTaskError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(&self.detail)
+    }
+}
+
+impl std::error::Error for S3BackupTaskError {}
+
+impl From<GatewayError> for S3BackupTaskError {
+    fn from(error: GatewayError) -> Self {
+        Self::internal(format!("{error:?}"))
+    }
+}
+
+pub(crate) async fn start_s3_backup_task(
+    app: AppState,
+    trigger: &str,
+    created_by: Option<&str>,
+) -> Result<S3BackupTaskStart, S3BackupTaskError> {
+    start_s3_backup_task_with_slot(app, trigger, created_by, None).await
+}
+
+pub(crate) async fn start_s3_backup_task_for_schedule(
+    app: AppState,
+    scheduled_slot: String,
+) -> Result<S3BackupTaskStart, S3BackupTaskError> {
+    start_s3_backup_task_with_slot(app, "scheduled", None, Some(scheduled_slot)).await
+}
+
+async fn start_s3_backup_task_with_slot(
+    app: AppState,
+    trigger: &str,
+    created_by: Option<&str>,
+    scheduled_slot: Option<String>,
+) -> Result<S3BackupTaskStart, S3BackupTaskError> {
+    let config = load_s3_backup_config_for_run(&app).await?;
+    ensure_background_task_storage(&app)?;
+    let lock = acquire_s3_backup_task_lock(&app).await?;
+    let active_run_exists = match has_active_s3_backup_task(&app).await {
+        Ok(value) => value,
+        Err(error) => {
+            release_s3_backup_task_lock(&app, lock).await;
+            return Err(error);
+        }
+    };
+    if active_run_exists {
+        release_s3_backup_task_lock(&app, lock).await;
+        return Err(S3BackupTaskError::conflict(
+            "已有 S3 备份任务正在执行，请等待当前任务完成后再试",
+        ));
+    }
+
+    let run_id = build_task_run_id();
+    let created_at = now_unix_secs();
+    let max_attempts = task_definition(TASK_KEY_SYSTEM_S3_BACKUP)
+        .map(|item| item.retry_policy.max_attempts)
+        .unwrap_or(1);
+
+    let run = UpsertBackgroundTaskRun {
+        id: run_id.clone(),
+        task_key: TASK_KEY_SYSTEM_S3_BACKUP.to_string(),
+        kind: BackgroundTaskKind::Scheduled,
+        trigger: trigger.to_string(),
+        status: BackgroundTaskStatus::Queued,
+        attempt: 1,
+        max_attempts,
+        owner_instance: Some(app.tunnel.local_instance_id().to_string()),
+        progress_percent: 0,
+        progress_message: Some(S3_BACKUP_QUEUED_MESSAGE.to_string()),
+        payload_json: Some(s3_backup_task_payload_json(
+            &config,
+            trigger,
+            scheduled_slot.as_deref(),
+        )),
+        result_json: None,
+        error_message: None,
+        cancel_requested: false,
+        created_by: Some(created_by.unwrap_or("admin").to_string()),
+        created_at_unix_secs: created_at,
+        started_at_unix_secs: None,
+        finished_at_unix_secs: None,
+        updated_at_unix_secs: created_at,
+    };
+    if upsert_run_with_logging(&app, run).await.is_none() {
+        release_s3_backup_task_lock(&app, lock).await;
+        return Err(S3BackupTaskError::service_unavailable(
+            "无法创建 S3 备份后台任务记录，请检查后台任务存储是否可用",
+        ));
+    }
+    append_event_with_logging(
+        &app,
+        &run_id,
+        "queued",
+        "S3 backup task queued",
+        Some(json!({ "trigger": trigger })),
+    )
+    .await;
+
+    spawn_s3_backup_worker(app, run_id.clone(), config, lock, scheduled_slot);
+
+    Ok(S3BackupTaskStart {
+        id: run_id,
+        task_key: TASK_KEY_SYSTEM_S3_BACKUP,
+        status: BackgroundTaskStatus::Queued.as_database(),
+        progress_message: S3_BACKUP_QUEUED_MESSAGE,
+    })
+}
+
+fn s3_backup_task_payload_json(
+    config: &S3BackupConfig,
+    trigger: &str,
+    scheduled_slot: Option<&str>,
+) -> Value {
+    let mut payload = json!({
+        "scope": config.scope.as_config_value(),
+        "bucket": config.bucket.clone(),
+        "prefix": config.prefix.clone(),
+        "compression": config.compression.clone(),
+        "trigger": trigger,
+    });
+    if let Some(scheduled_slot) = scheduled_slot {
+        payload["scheduled_slot"] = Value::String(scheduled_slot.to_string());
+    }
+    payload
+}
+
+fn spawn_s3_backup_worker(
+    app: AppState,
+    run_id: String,
+    config: S3BackupConfig,
+    lock: RuntimeLockLease,
+    scheduled_slot: Option<String>,
+) {
+    spawn_fire_and_forget("task-runtime-system-s3-backup", async move {
+        let app_for_worker = app.clone();
+        let run_id_for_worker = run_id.clone();
+        let result = std::panic::AssertUnwindSafe(run_s3_backup_worker_inner(
+            app_for_worker,
+            run_id_for_worker,
+            config,
+            lock.clone(),
+            scheduled_slot,
+        ))
+        .catch_unwind()
+        .await;
+        if result.is_err() {
+            warn!(run_id = %run_id, "S3 backup task panicked");
+            let _ = update_run_status(
+                &app,
+                &run_id,
+                BackgroundTaskStatus::Failed,
+                Some(100),
+                Some("S3 备份任务异常退出".to_string()),
+                None,
+                Some("S3 backup task panicked".to_string()),
+                None,
+                Some(now_unix_secs()),
+            )
+            .await;
+            append_event_with_logging(&app, &run_id, "failed", "S3 backup task panicked", None)
+                .await;
+        }
+
+        release_s3_backup_task_lock(&app, lock).await;
+    });
+}
+
+async fn run_s3_backup_worker_inner(
+    app: AppState,
+    run_id: String,
+    config: S3BackupConfig,
+    lock: RuntimeLockLease,
+    scheduled_slot: Option<String>,
+) {
+    let started_at = now_unix_secs();
+    let _ = update_run_status(
+        &app,
+        &run_id,
+        BackgroundTaskStatus::Running,
+        Some(5),
+        Some("S3 备份任务开始执行".to_string()),
+        None,
+        None,
+        Some(started_at),
+        None,
+    )
+    .await;
+    append_event_with_logging(&app, &run_id, "running", "S3 backup task started", None).await;
+
+    let heartbeat = spawn_s3_backup_task_heartbeat(app.clone(), run_id.clone(), lock);
+    let result = run_s3_backup_once(&app, &config).await;
+    heartbeat.abort();
+    let _ = heartbeat.await;
+
+    match result {
+        Ok(result) => {
+            if let Some(slot) = scheduled_backup_slot_to_record(scheduled_slot.as_deref(), true) {
+                if let Err(error) = record_scheduled_backup_slot(&app, &slot).await {
+                    warn!(error = ?error, run_id = %run_id, "S3 backup slot record failed");
+                    let _ = update_run_status(
+                        &app,
+                        &run_id,
+                        BackgroundTaskStatus::Failed,
+                        Some(100),
+                        Some("S3 备份任务完成，但记录调度时间失败".to_string()),
+                        None,
+                        Some(format!("S3 backup slot record failed: {error:?}")),
+                        None,
+                        Some(now_unix_secs()),
+                    )
+                    .await;
+                    append_event_with_logging(
+                        &app,
+                        &run_id,
+                        "failed",
+                        "S3 backup slot record failed",
+                        Some(json!({ "error": format!("{error:?}") })),
+                    )
+                    .await;
+                    return;
+                }
+            }
+            let result_json = backup_run_result_json(&result);
+            let _ = update_run_status(
+                &app,
+                &run_id,
+                BackgroundTaskStatus::Succeeded,
+                Some(100),
+                Some("S3 备份任务完成".to_string()),
+                Some(result_json.clone()),
+                None,
+                None,
+                Some(now_unix_secs()),
+            )
+            .await;
+            append_event_with_logging(
+                &app,
+                &run_id,
+                "succeeded",
+                "S3 backup task completed",
+                Some(result_json),
+            )
+            .await;
+        }
+        Err(error) => {
+            warn!(error = %error, run_id = %run_id, "S3 backup task failed");
+            let _ = update_run_status(
+                &app,
+                &run_id,
+                BackgroundTaskStatus::Failed,
+                Some(100),
+                Some("S3 备份任务失败".to_string()),
+                None,
+                Some(error.to_string()),
+                None,
+                Some(now_unix_secs()),
+            )
+            .await;
+            append_event_with_logging(
+                &app,
+                &run_id,
+                "failed",
+                "S3 backup task failed",
+                Some(json!({ "error": error.to_string() })),
+            )
+            .await;
+        }
+    }
+}
+
+fn spawn_s3_backup_task_heartbeat(
+    app: AppState,
+    run_id: String,
+    lock: RuntimeLockLease,
+) -> tokio::task::JoinHandle<()> {
+    spawn_fire_and_forget("task-runtime-system-s3-backup-heartbeat", async move {
+        let mut interval = tokio::time::interval(S3_BACKUP_TASK_HEARTBEAT_INTERVAL);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        interval.tick().await;
+        loop {
+            interval.tick().await;
+            let _ = app
+                .runtime_state
+                .lock_renew(&lock, S3_BACKUP_TASK_LOCK_TTL)
+                .await;
+            let _ = update_run_status(
+                &app,
+                &run_id,
+                BackgroundTaskStatus::Running,
+                Some(50),
+                Some("S3 备份任务执行中".to_string()),
+                None,
+                None,
+                None,
+                None,
+            )
+            .await;
+        }
+    })
+}
+
+fn scheduled_backup_slot_to_record(
+    scheduled_slot: Option<&str>,
+    task_succeeded: bool,
+) -> Option<String> {
+    if !task_succeeded {
+        return None;
+    }
+    scheduled_slot
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(ToOwned::to_owned)
+}
+
+async fn record_scheduled_backup_slot(
+    app: &AppState,
+    scheduled_slot: &str,
+) -> Result<(), GatewayError> {
+    app.upsert_system_config_json_value(
+        super::S3_BACKUP_LAST_SLOT_KEY,
+        &Value::String(scheduled_slot.to_string()),
+        None,
+    )
+    .await?;
+    Ok(())
+}
+
+fn ensure_background_task_storage(app: &AppState) -> Result<(), S3BackupTaskError> {
+    if app.has_background_task_data_reader() && app.has_background_task_data_writer() {
+        return Ok(());
+    }
+
+    Err(S3BackupTaskError::service_unavailable(
+        "当前节点未启用后台任务存储，无法提交 S3 备份任务",
+    ))
+}
+
+async fn acquire_s3_backup_task_lock(
+    app: &AppState,
+) -> Result<RuntimeLockLease, S3BackupTaskError> {
+    match app
+        .runtime_state
+        .lock_try_acquire(
+            S3_BACKUP_TASK_LOCK_KEY,
+            app.tunnel.local_instance_id(),
+            S3_BACKUP_TASK_LOCK_TTL,
+        )
+        .await
+    {
+        Ok(Some(lock)) => Ok(lock),
+        Ok(None) => Err(S3BackupTaskError::conflict(
+            "已有 S3 备份任务正在执行，请等待当前任务完成后再试",
+        )),
+        Err(error) => Err(S3BackupTaskError::service_unavailable(format!(
+            "无法获取 S3 备份任务锁：{error}"
+        ))),
+    }
+}
+
+async fn release_s3_backup_task_lock(app: &AppState, lock: RuntimeLockLease) {
+    let _ = app.runtime_state.lock_release(&lock).await;
+}
+
+async fn has_active_s3_backup_task(app: &AppState) -> Result<bool, S3BackupTaskError> {
+    let now = now_unix_secs();
+    for status in [BackgroundTaskStatus::Queued, BackgroundTaskStatus::Running] {
+        let page = app
+            .list_background_task_runs(&BackgroundTaskListQuery {
+                task_key_substring: Some(TASK_KEY_SYSTEM_S3_BACKUP.to_string()),
+                kind: Some(BackgroundTaskKind::Scheduled),
+                status: Some(status),
+                trigger: None,
+                offset: 0,
+                limit: 100,
+            })
+            .await?;
+        if page
+            .items
+            .iter()
+            .any(|run| is_blocking_active_s3_backup_run(run, status, now))
+        {
+            return Ok(true);
+        }
+    }
+    Ok(false)
+}
+
+fn is_blocking_active_s3_backup_run(
+    run: &StoredBackgroundTaskRun,
+    status: BackgroundTaskStatus,
+    now_unix_secs: u64,
+) -> bool {
+    run.task_key == TASK_KEY_SYSTEM_S3_BACKUP
+        && run.kind == BackgroundTaskKind::Scheduled
+        && run.status == status
+        && !run.cancel_requested
+        && now_unix_secs.saturating_sub(run.updated_at_unix_secs)
+            < S3_BACKUP_ACTIVE_TASK_STALE_AFTER_SECS
+}
+
+async fn run_s3_backup_once(
+    app: &AppState,
+    config: &S3BackupConfig,
+) -> Result<BackupRunResult, S3BackupTaskError> {
+    let admin_state = AdminAppState::new(app);
+    let payload = match config.scope {
+        BackupScope::Config => {
+            admin_state
+                .build_admin_system_config_export_payload()
+                .await?
+        }
+        BackupScope::Users => {
+            admin_state
+                .build_admin_system_users_export_payload()
+                .await?
+        }
+        BackupScope::Data => admin_state.build_admin_system_data_export_payload().await?,
+    };
+    let store = ObjectStoreS3BackupStore::from_config(config)
+        .map_err(|error| S3BackupTaskError::internal(error.to_string()))?;
+    run_backup_with_store(config, &store, payload, Utc::now())
+        .await
+        .map_err(|error| S3BackupTaskError::internal(error.to_string()))
+}
+
+async fn load_s3_backup_config_for_run(
+    app: &AppState,
+) -> Result<S3BackupConfig, S3BackupTaskError> {
+    let mut values = load_s3_backup_config_values(app).await?;
+    values.insert("backup_s3_enabled".to_string(), Value::Bool(true));
+    S3BackupConfig::from_json_map(&values)
+        .map_err(|error| S3BackupTaskError::bad_request(format!("S3 备份配置无效：{error}")))
+}
+
+pub(crate) async fn load_s3_backup_config_values(
+    app: &AppState,
+) -> Result<Map<String, Value>, S3BackupTaskError> {
+    let mut values = Map::new();
+    for key in S3_BACKUP_CONFIG_KEYS {
+        let value = app
+            .read_system_config_json_value(key)
+            .await
+            .map_err(S3BackupTaskError::from)?
+            .or_else(|| admin_system_config_default_value(key));
+        if let Some(value) = value {
+            let value = if *key == "backup_s3_secret_access_key" {
+                decrypt_s3_secret_access_key(app, value)?
+            } else {
+                value
+            };
+            values.insert((*key).to_string(), value);
+        }
+    }
+    Ok(values)
+}
+
+fn decrypt_s3_secret_access_key(app: &AppState, value: Value) -> Result<Value, S3BackupTaskError> {
+    let Some(ciphertext) = value
+        .as_str()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+    else {
+        return Ok(value);
+    };
+    let Some(plaintext) = decrypt_catalog_secret_with_fallbacks(app.encryption_key(), ciphertext)
+    else {
+        return Err(S3BackupTaskError::bad_request(
+            "S3 备份配置无效：Secret Access Key（访问密钥）无法解密，请重新填写",
+        ));
+    };
+    Ok(Value::String(plaintext))
+}
+
+fn backup_run_result_json(result: &BackupRunResult) -> Value {
+    json!({
+        "scope": result.scope.as_config_value(),
+        "bucket": result.bucket,
+        "object_key": result.object_key,
+        "bytes": result.bytes,
+        "sha256": result.sha256,
+        "export_version": result.export_version,
+        "exported_at": result.exported_at,
+        "compression": result.compression,
+        "deleted_old_objects": result.deleted_old_objects,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use aether_crypto::{encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY};
+    use aether_data::repository::background_tasks::InMemoryBackgroundTaskRepository;
+    use aether_data_contracts::repository::background_tasks::{
+        BackgroundTaskKind, BackgroundTaskStatus, StoredBackgroundTaskRun,
+    };
+
+    use crate::data::GatewayDataState;
+    use crate::state::AppState;
+    use crate::task_runtime::{now_unix_secs, TASK_KEY_SYSTEM_S3_BACKUP};
+
+    fn valid_s3_backup_config_values() -> Vec<(String, serde_json::Value)> {
+        vec![
+            (
+                "backup_s3_endpoint".to_string(),
+                serde_json::json!("https://s3.example.com"),
+            ),
+            (
+                "backup_s3_bucket".to_string(),
+                serde_json::json!("aether-backups"),
+            ),
+            ("backup_s3_prefix".to_string(), serde_json::json!("prod/")),
+            (
+                "backup_s3_access_key_id".to_string(),
+                serde_json::json!("access-key-id"),
+            ),
+            (
+                "backup_s3_secret_access_key".to_string(),
+                serde_json::json!(encrypt_python_fernet_plaintext(
+                    DEVELOPMENT_ENCRYPTION_KEY,
+                    "secret"
+                )
+                .expect("test secret should encrypt")),
+            ),
+        ]
+    }
+
+    fn stored_s3_backup_run(status: BackgroundTaskStatus) -> StoredBackgroundTaskRun {
+        let now = now_unix_secs();
+        StoredBackgroundTaskRun {
+            id: "existing-s3-backup-run".to_string(),
+            task_key: TASK_KEY_SYSTEM_S3_BACKUP.to_string(),
+            kind: BackgroundTaskKind::Scheduled,
+            trigger: "manual".to_string(),
+            status,
+            attempt: 1,
+            max_attempts: 1,
+            owner_instance: Some("test-instance".to_string()),
+            progress_percent: 5,
+            progress_message: Some("running".to_string()),
+            payload_json: None,
+            result_json: None,
+            error_message: None,
+            cancel_requested: false,
+            created_by: Some("admin".to_string()),
+            created_at_unix_secs: now,
+            started_at_unix_secs: Some(now),
+            finished_at_unix_secs: None,
+            updated_at_unix_secs: now,
+        }
+    }
+
+    #[tokio::test]
+    async fn start_s3_backup_task_rejects_missing_bucket_for_manual_run() {
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled().with_system_config_values_for_tests(vec![(
+                    "backup_s3_endpoint".to_string(),
+                    serde_json::json!("https://s3.example.com"),
+                )]),
+            );
+
+        let err = super::start_s3_backup_task(app, "manual", Some("admin-user-123"))
+            .await
+            .expect_err("missing bucket should reject the backup run");
+
+        assert_eq!(err.status(), axum::http::StatusCode::BAD_REQUEST);
+        assert!(err.to_string().contains("Bucket"));
+    }
+
+    #[tokio::test]
+    async fn start_s3_backup_task_requires_background_task_storage() {
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled()
+                    .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY)
+                    .with_system_config_values_for_tests(valid_s3_backup_config_values()),
+            );
+
+        let err = super::start_s3_backup_task(app, "manual", Some("admin-user-123"))
+            .await
+            .expect_err("manual backup should require observable background task storage");
+
+        assert_eq!(err.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
+        assert!(err.to_string().contains("后台任务存储"));
+    }
+
+    #[tokio::test]
+    async fn start_s3_backup_task_rejects_when_same_task_is_active() {
+        let repository = Arc::new(InMemoryBackgroundTaskRepository::seed_runs([
+            stored_s3_backup_run(BackgroundTaskStatus::Running),
+        ]));
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled()
+                    .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY)
+                    .with_system_config_values_for_tests(valid_s3_backup_config_values())
+                    .with_background_task_repository_for_tests(repository),
+            );
+
+        let err = super::start_s3_backup_task(app, "manual", Some("admin-user-123"))
+            .await
+            .expect_err("manual backup should reject duplicate active runs");
+
+        assert_eq!(err.status(), axum::http::StatusCode::CONFLICT);
+        assert!(err.to_string().contains("已有 S3 备份任务正在执行"));
+    }
+
+    #[tokio::test]
+    async fn active_s3_backup_detection_blocks_queued_runs() {
+        let repository = Arc::new(InMemoryBackgroundTaskRepository::seed_runs([
+            stored_s3_backup_run(BackgroundTaskStatus::Queued),
+        ]));
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled().with_background_task_repository_for_tests(repository),
+            );
+
+        assert!(super::has_active_s3_backup_task(&app)
+            .await
+            .expect("active task lookup should succeed"));
+    }
+
+    #[tokio::test]
+    async fn active_s3_backup_detection_ignores_cancelled_and_stale_runs() {
+        let now = now_unix_secs();
+        let mut stale = stored_s3_backup_run(BackgroundTaskStatus::Running);
+        stale.id = "stale-s3-backup-run".to_string();
+        stale.updated_at_unix_secs = now
+            .saturating_sub(super::S3_BACKUP_ACTIVE_TASK_STALE_AFTER_SECS)
+            .saturating_sub(1);
+        let mut cancelled = stored_s3_backup_run(BackgroundTaskStatus::Queued);
+        cancelled.id = "cancelled-s3-backup-run".to_string();
+        cancelled.cancel_requested = true;
+        let repository = Arc::new(InMemoryBackgroundTaskRepository::seed_runs([
+            stale, cancelled,
+        ]));
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled().with_background_task_repository_for_tests(repository),
+            );
+
+        assert!(!super::has_active_s3_backup_task(&app)
+            .await
+            .expect("active task lookup should succeed"));
+    }
+
+    #[tokio::test]
+    async fn queued_s3_backup_task_payload_does_not_include_secret() {
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled()
+                    .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY)
+                    .with_system_config_values_for_tests(valid_s3_backup_config_values()),
+            );
+        let values = super::load_s3_backup_config_values(&app)
+            .await
+            .expect("config should load");
+        let config = super::S3BackupConfig::from_json_map(&values)
+            .expect("config should parse for payload test");
+
+        let payload = super::s3_backup_task_payload_json(&config, "manual", None);
+
+        assert!(payload["bucket"].is_string());
+        assert_eq!(payload["trigger"], serde_json::json!("manual"));
+        assert!(!payload.to_string().contains("secret"));
+    }
+
+    #[test]
+    fn scheduled_backup_slot_records_only_successful_scheduled_runs() {
+        assert_eq!(
+            super::scheduled_backup_slot_to_record(Some("days:2026-05-24T19:00:00Z"), true),
+            Some("days:2026-05-24T19:00:00Z".to_string())
+        );
+        assert_eq!(
+            super::scheduled_backup_slot_to_record(Some("days:2026-05-24T19:00:00Z"), false),
+            None
+        );
+        assert_eq!(super::scheduled_backup_slot_to_record(None, true), None);
+        assert_eq!(
+            super::scheduled_backup_slot_to_record(Some("  "), true),
+            None
+        );
+    }
+
+    #[tokio::test]
+    async fn record_scheduled_backup_slot_updates_system_config() {
+        let app = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                GatewayDataState::disabled().with_system_config_values_for_tests(Vec::<(
+                    String,
+                    serde_json::Value,
+                )>::new(
+                )),
+            );
+
+        super::record_scheduled_backup_slot(&app, "days:2026-05-24T19:00:00Z")
+            .await
+            .expect("slot record should write system config");
+
+        assert_eq!(
+            app.read_system_config_json_value(super::super::S3_BACKUP_LAST_SLOT_KEY)
+                .await
+                .expect("slot config should be readable"),
+            Some(serde_json::json!("days:2026-05-24T19:00:00Z"))
+        );
+    }
+}

apps/aether-gateway/src/backup/worker.rs

@@ -0,0 +1,114 @@
+use std::time::Duration;
+
+use chrono::{DateTime, Utc};
+use tokio::task::JoinHandle;
+use tracing::warn;
+
+use super::config::S3BackupConfig;
+use crate::{AppState, GatewayError};
+
+pub(crate) const S3_BACKUP_WORKER_TASK_KEY: &str =
+    crate::task_runtime::TASK_KEY_SYSTEM_S3_BACKUP_WORKER;
+const S3_BACKUP_WORKER_INTERVAL: Duration = Duration::from_secs(60);
+
+pub(crate) fn should_start_scheduled_backup(last_slot: Option<&str>, current_slot: &str) -> bool {
+    last_slot != Some(current_slot)
+}
+
+pub(crate) fn spawn_s3_backup_worker(app: AppState) -> Option<JoinHandle<()>> {
+    if !app.data.has_system_config_store() {
+        return None;
+    }
+
+    Some(tokio::spawn(async move {
+        let mut interval = tokio::time::interval(S3_BACKUP_WORKER_INTERVAL);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        interval.tick().await;
+        loop {
+            interval.tick().await;
+            if let Err(error) = run_s3_backup_schedule_tick(&app, Utc::now()).await {
+                warn!(error = ?error, "S3 backup schedule tick failed");
+            }
+        }
+    }))
+}
+
+async fn run_s3_backup_schedule_tick(
+    app: &AppState,
+    now: DateTime<Utc>,
+) -> Result<(), GatewayError> {
+    let values = match super::task::load_s3_backup_config_values(app).await {
+        Ok(values) => values,
+        Err(error) => {
+            warn!(error = %error, "S3 backup schedule config load failed");
+            return Ok(());
+        }
+    };
+    let config = match S3BackupConfig::from_json_map(&values) {
+        Ok(config) => config,
+        Err(error) => {
+            warn!(error = %error, "S3 backup schedule config is invalid");
+            return Ok(());
+        }
+    };
+    if !config.enabled {
+        return Ok(());
+    }
+    let Some(slot) = config.schedule.due_slot(now) else {
+        return Ok(());
+    };
+    let last_slot = read_last_backup_slot(app).await?;
+    if !should_start_scheduled_backup(last_slot.as_deref(), &slot) {
+        return Ok(());
+    }
+
+    match super::task::start_s3_backup_task_for_schedule(app.clone(), slot).await {
+        Ok(_) => {}
+        Err(error) => {
+            warn!(error = %error, "S3 backup scheduled task submission failed");
+        }
+    }
+    Ok(())
+}
+
+async fn read_last_backup_slot(app: &AppState) -> Result<Option<String>, GatewayError> {
+    Ok(app
+        .read_system_config_json_value(super::S3_BACKUP_LAST_SLOT_KEY)
+        .await?
+        .and_then(|value| value.as_str().map(str::trim).map(str::to_string))
+        .filter(|value| !value.is_empty()))
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::backup::schedule::{BackupSchedule, BackupScheduleUnit};
+    use crate::task_runtime::{task_definition, TASK_KEY_SYSTEM_S3_BACKUP};
+
+    #[test]
+    fn backup_worker_skips_already_recorded_slot() {
+        let schedule = BackupSchedule {
+            unit: BackupScheduleUnit::Days,
+            interval: 1,
+            minute: 0,
+            hour: 3,
+            weekday: 1,
+            month_day: 1,
+        };
+        let now = chrono::DateTime::parse_from_rfc3339("2026-05-24T03:00:30+08:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+        let slot = schedule.due_slot(now).expect("slot should be due");
+
+        assert!(super::should_start_scheduled_backup(
+            Some("days:2026-05-22T19:00:00Z"),
+            &slot
+        ));
+        assert!(!super::should_start_scheduled_backup(Some(&slot), &slot));
+    }
+
+    #[test]
+    fn backup_worker_has_distinct_supervisor_task_key() {
+        assert_ne!(super::S3_BACKUP_WORKER_TASK_KEY, TASK_KEY_SYSTEM_S3_BACKUP);
+        assert!(task_definition(super::S3_BACKUP_WORKER_TASK_KEY).is_some());
+    }
+}

apps/aether-gateway/src/control/route/admin/system_families.rs

@@ -130,6 +130,15 @@ pub(super) fn classify_admin_system_family_route(
             "admin:system",
             false,
         ))
+    } else if method == http::Method::POST && normalized_path == "/api/admin/system/backups/s3/run"
+    {
+        Some(classified(
+            "admin_proxy",
+            "system_manage",
+            "s3_backup_run",
+            "admin:system",
+            false,
+        ))
     } else if method == http::Method::POST && normalized_path == "/api/admin/system/config/import" {
         Some(classified(
             "admin_proxy",

apps/aether-gateway/src/control/tests/admin_core.rs

@@ -175,6 +175,25 @@ fn classifies_admin_system_data_export_as_admin_proxy_route() {
     assert!(!decision.is_execution_runtime_candidate());
 }
 
+#[test]
+fn classifies_admin_system_s3_backup_start_as_admin_proxy_route() {
+    let headers = headers(&[]);
+    let uri: Uri = "/api/admin/system/backups/s3/run"
+        .parse()
+        .expect("uri should parse");
+    let decision =
+        classify_control_route(&http::Method::POST, &uri, &headers).expect("route should classify");
+
+    assert_eq!(decision.route_class.as_deref(), Some("admin_proxy"));
+    assert_eq!(decision.route_family.as_deref(), Some("system_manage"));
+    assert_eq!(decision.route_kind.as_deref(), Some("s3_backup_run"));
+    assert_eq!(
+        decision.auth_endpoint_signature.as_deref(),
+        Some("admin:system")
+    );
+    assert!(!decision.is_execution_runtime_candidate());
+}
+
 #[test]
 fn classifies_admin_system_maintenance_write_routes_as_admin_proxy_route() {
     let headers = headers(&[]);

apps/aether-gateway/src/data/state/testing/mod.rs

@@ -9,7 +9,8 @@ use aether_data_contracts::repository::usage::UsageRepository;
 use super::{
     AnnouncementReadRepository, AnnouncementWriteRepository, AuthApiKeyReadRepository,
     AuthApiKeyWriteRepository, AuthModuleReadRepository, AuthModuleWriteRepository,
-    BillingReadRepository, GatewayDataConfig, GatewayDataState, GeminiFileMappingReadRepository,
+    BackgroundTaskReadRepository, BackgroundTaskWriteRepository, BillingReadRepository,
+    GatewayDataConfig, GatewayDataState, GeminiFileMappingReadRepository,
     GeminiFileMappingWriteRepository, GlobalModelReadRepository, GlobalModelWriteRepository,
     ManagementTokenReadRepository, ManagementTokenWriteRepository,
     MinimalCandidateSelectionReadRepository, OAuthProviderReadRepository,
@@ -261,6 +262,18 @@ impl GatewayDataState {
         self
     }
 
+    #[cfg(test)]
+    pub(crate) fn with_background_task_repository_for_tests<T>(mut self, repository: Arc<T>) -> Self
+    where
+        T: BackgroundTaskReadRepository + BackgroundTaskWriteRepository + 'static,
+    {
+        let background_task_reader: Arc<dyn BackgroundTaskReadRepository> = repository.clone();
+        let background_task_writer: Arc<dyn BackgroundTaskWriteRepository> = repository;
+        self.background_task_reader = Some(background_task_reader);
+        self.background_task_writer = Some(background_task_writer);
+        self
+    }
+
     #[cfg(test)]
     pub(crate) fn with_global_model_repository_for_tests<T>(mut self, repository: Arc<T>) -> Self
     where

apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs

@@ -367,6 +367,39 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
         ));
     }
 
+    if decision.route_kind.as_deref() == Some("s3_backup_run")
+        && request_method == http::Method::POST
+        && request_path == "/api/admin/system/backups/s3/run"
+    {
+        return Ok(Some(
+            match crate::backup::task::start_s3_backup_task(
+                state.cloned_app(),
+                "manual",
+                decision
+                    .admin_principal
+                    .as_ref()
+                    .map(|principal| principal.user_id.as_str()),
+            )
+            .await
+            {
+                Ok(task) => attach_admin_audit_response(
+                    Json(json!({
+                        "message": "S3 备份任务已提交",
+                        "task": task,
+                    }))
+                    .into_response(),
+                    "admin_system_s3_backup_task_started",
+                    "run_s3_backup",
+                    "s3_backup",
+                    "global",
+                ),
+                Err(error) => {
+                    (error.status(), Json(json!({ "detail": error.detail() }))).into_response()
+                }
+            },
+        ));
+    }
+
     if decision.route_kind.as_deref() == Some("smtp_test")
         && request_method == http::Method::POST
         && request_path == "/api/admin/system/smtp/test"

apps/aether-gateway/src/lib.rs

@@ -30,6 +30,7 @@ mod api;
 mod async_task;
 mod audit;
 mod auth;
+mod backup;
 mod bark_push;
 mod cache;
 mod client_session_affinity;

apps/aether-gateway/src/state/core.rs

@@ -1268,6 +1268,10 @@ impl AppState {
             crate::task_runtime::TASK_KEY_VIDEO_TASK_POLLER,
             spawn_video_task_poller(self.clone()),
         );
+        supervise_worker(
+            crate::backup::worker::S3_BACKUP_WORKER_TASK_KEY,
+            crate::backup::worker::spawn_s3_backup_worker(self.clone()),
+        );
 
         supervisor
     }

apps/aether-gateway/src/task_runtime/mod.rs

@@ -17,6 +17,8 @@ use crate::{AppState, GatewayError};
 
 pub(crate) const TASK_KEY_PROVIDER_DELETE: &str = "admin.provider.delete";
 pub(crate) const TASK_KEY_PROVIDER_OAUTH_BATCH_IMPORT: &str = "admin.provider.oauth.batch_import";
+pub(crate) const TASK_KEY_SYSTEM_S3_BACKUP: &str = "system.s3.backup";
+pub(crate) const TASK_KEY_SYSTEM_S3_BACKUP_WORKER: &str = "system.s3.backup.worker";
 pub(crate) const TASK_KEY_USAGE_QUEUE_WORKER: &str = "usage.queue.worker";
 pub(crate) const TASK_KEY_USAGE_COUNTER_FLUSH: &str = "usage.counter.flush.worker";
 pub(crate) const TASK_KEY_VIDEO_TASK_POLLER: &str = "video.task.poller";
@@ -64,6 +66,22 @@ const TASK_DEFINITIONS: &[TaskDefinition] = &[
         true,
         RETRY_ONCE,
     ),
+    TaskDefinition::new(
+        TASK_KEY_SYSTEM_S3_BACKUP,
+        TaskKind::Scheduled,
+        "manual",
+        false,
+        true,
+        RETRY_ONCE,
+    ),
+    TaskDefinition::new(
+        TASK_KEY_SYSTEM_S3_BACKUP_WORKER,
+        TaskKind::Daemon,
+        "daemon",
+        true,
+        true,
+        RETRY_ONCE,
+    ),
     TaskDefinition::new(
         TASK_KEY_USAGE_QUEUE_WORKER,
         TaskKind::Daemon,