feat(backup): backup migrate-legacy [backup/restore 10c]

crates/app/src/cmd_backup.rs

@@ -87,9 +87,15 @@
 //! backends purge runs beside live servers; the maintenance claim is the
 //! exclusion.
 //!
+//! `backup migrate-legacy <arn>|--all` copies in-catalog backups into the
+//! configured store and replaces each by its catalog row, under the
+//! maintenance claim on every backend (on SQLite under the serve lock as
+//! well); its module (`cmd_backup_migrate.rs`) states its output and exit.
+//!
 //! Every command opens the catalog through the backend's factory, under its own
 //! name (`extenddb backup list`, `extenddb backup purge`, `extenddb backup
-//! remove-target`, `extenddb backup sync`), and none runs the startup recovery
+//! remove-target`, `extenddb backup sync`, `extenddb backup migrate-legacy`),
+//! and none runs the startup recovery
 //! a server runs (`ServerComponentsOptions::startup_recovery` off): that work
 //! belongs to the servers the command runs beside, and a command that ran it
 //! could exit with it half done. On PostgreSQL, MongoDB and Cassandra the
@@ -220,6 +226,9 @@ use extenddb_storage::error::StorageError;
 use extenddb_storage::hooks::ServingRegistrar;
 use futures::TryStreamExt;
 
+#[path = "cmd_backup_migrate.rs"]
+mod migrate_legacy;
+
 #[derive(Args)]
 #[allow(clippy::doc_markdown)] // Clap help text, not rustdoc
 pub struct BackupArgs {
@@ -312,6 +321,21 @@ enum BackupCommand {
         )]
         forget_backup: Vec<String>,
     },
+    /// Copy in-catalog backups into the configured backup store, each under its own ARN, and
+    /// replace each by its catalog row. Takes the maintenance claim first (every backup and
+    /// restore job answers LimitExceededException until it ends) and waits for running jobs; on
+    /// SQLite it needs the server stopped. Refuses a backup a 0.1.x release took (its id carries
+    /// no instant) and one with an item or a definition the store format cannot hold, keeping its
+    /// in-catalog backup; restore such a backup and take a new one.
+    #[command(group(clap::ArgGroup::new("backups").required(true).args(["arn", "all"])))]
+    MigrateLegacy {
+        /// The backup ARN of the in-catalog backup to migrate
+        #[arg(value_name = "ARN")]
+        arn: Option<String>,
+        /// Migrate every in-catalog backup of every account, going on past a backup it refuses
+        #[arg(long)]
+        all: bool,
+    },
 }
 
 /// How often the waiting purge re-reads the rows.
@@ -325,6 +349,7 @@ const LIST: &str = "extenddb backup list";
 const PURGE: &str = "extenddb backup purge";
 const REMOVE_TARGET: &str = "extenddb backup remove-target";
 const SYNC: &str = "extenddb backup sync";
+const MIGRATE_LEGACY: &str = "extenddb backup migrate-legacy";
 
 pub async fn run(args: BackupArgs, release: &'static str) -> anyhow::Result<()> {
     if !std::path::Path::new(&args.config).exists() {
@@ -391,6 +416,7 @@ pub async fn run(args: BackupArgs, release: &'static str) -> anyhow::Result<()>
         BackupCommand::Purge { confirm: true } => (PURGE, Mode::Write),
         BackupCommand::Sync { .. } if forgetting => (SYNC, Mode::Forget),
         BackupCommand::Sync { .. } => (SYNC, Mode::Sync),
+        BackupCommand::MigrateLegacy { .. } => (MIGRATE_LEGACY, Mode::Write),
         BackupCommand::RemoveTarget { .. } => unreachable!("handled above"),
     };
     let opening = open(
@@ -500,6 +526,21 @@ pub async fn run(args: BackupArgs, release: &'static str) -> anyhow::Result<()>
             done
         }
         (BackupCommand::Sync { .. }, Opening::NoStore(reason)) => anyhow::bail!("{reason}"),
+        (BackupCommand::MigrateLegacy { arn, .. }, Opening::Store(opened)) => {
+            let done = migrate_legacy::migrate_legacy(
+                &opened,
+                &app_config,
+                release,
+                arn.as_deref(),
+                interrupt,
+            )
+            .await;
+            opened.close().await;
+            done
+        }
+        (BackupCommand::MigrateLegacy { .. }, Opening::NoStore(reason)) => {
+            anyhow::bail!("{reason}; backup migrate-legacy copies into the configured store")
+        }
         (_, Opening::CatalogOnly(catalog, reason)) => {
             catalog.close().await;
             anyhow::bail!("{reason}")

crates/app/src/cmd_backup_migrate.rs

@@ -0,0 +1,224 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `extenddb backup migrate-legacy <arn>|--all`: the in-catalog backups
+//! copied into the configured store and replaced there by their catalog
+//! rows, under their own ARNs (`extenddb_engine::migrate_legacy` states
+//! each step and every refusal).
+//!
+//! The command opens the catalog as `serve` does (on SQLite under the serve
+//! lock, so the server is stopped for it) and needs the configured store
+//! healthy, its identity known. It takes the maintenance claim on every
+//! backend and heartbeats it for its whole run, so every backup and restore
+//! job of the catalog answers `LimitExceededException` until it ends (a
+//! second maintenance command fails at once), and it waits, as `purge`
+//! does, until no CREATING or FAILED backup row of the store has a live
+//! owner and no restore job is RUNNING (a legacy restore of a backup about
+//! to be replaced among them). On SQLite it first runs the reconciler's
+//! dead-owner resolutions itself and waits with a zero window, every owner
+//! being dead under the serve lock.
+//!
+//! `<arn>` names one backup, as written or rebuilt in this server's region
+//! (the spellings DescribeBackup resolves); `--all` takes every in-catalog
+//! backup that is not DELETED, of every account, in creation order, and
+//! goes on past a backup it refuses. Each backup gets one line: `migrated`,
+//! `finished` (an earlier run had written its row), `refused` (with the
+//! reason and the way out; its in-catalog backup is kept), `skipped` (not
+//! AVAILABLE, or deleted meanwhile) or `failed` (a store or catalog call;
+//! a later run tries again), then a summary. Each migration is logged under
+//! `extenddb::audit` with the account and the ARN. A first `SIGINT` or
+//! `SIGTERM` stops the run before the next backup or the next data file
+//! (the copy in progress is deleted), and a lost claim, a store the writer
+//! finds unhealthy or fenced, or a catalog call that fails stops it at
+//! once; the claim is then released and the command exits 1. It exits 0
+//! when every backup it reached is in the store, and 1 otherwise, after the
+//! others ran.
+
+use std::time::Duration;
+
+use extenddb_core::validation::RestorableLimits;
+use extenddb_engine::HeldClaim;
+use extenddb_engine::maintenance::Interrupt;
+use extenddb_engine::migrate_legacy::{BackupOutcome, Migration, Outcome, Selection, migrate};
+use extenddb_storage::backup_catalog::JOB_HEARTBEAT_INTERVAL;
+use extenddb_storage::error::StorageError;
+
+use super::{Identity, Opened, config, release, wait_for_jobs};
+
+/// `backup migrate-legacy` (module doc). `arn` is `None` for `--all`;
+/// `version` is this binary's, which the manifests name as their writer.
+pub(super) async fn migrate_legacy(
+    opened: &Opened,
+    app_config: &config::AppConfig,
+    version: &str,
+    arn: Option<&str>,
+    mut interrupt: Interrupt,
+) -> anyhow::Result<()> {
+    let storage = &*opened.catalog.storage;
+    let configured = opened.store();
+    println!("=== extenddb backup migrate-legacy ===");
+    let identity = match &opened.identity {
+        Identity::Known(identity) => {
+            println!(
+                "Store:  {} {} (store id {})",
+                identity.store_kind, identity.store_root, identity.store_id
+            );
+            identity.clone()
+        }
+        Identity::Empty => anyhow::bail!(
+            "the {} backup store {} has no {}: its check did not run",
+            configured.kind,
+            configured.root,
+            extenddb_engine::STORE_IDENTITY_KEY
+        ),
+        Identity::Problem(problem) => anyhow::bail!(
+            "the {} backup store {} is unhealthy: {problem}; nothing is migrated into a store \
+             whose check fails",
+            configured.kind,
+            configured.root
+        ),
+    };
+
+    let backup = &app_config.backup;
+    let stale_after = Duration::from_secs(backup.stale_job_after_secs);
+    let holder = format!(
+        "extenddb backup migrate-legacy on {} (pid {})",
+        extenddb_server::backup_store::hostname(),
+        std::process::id()
+    );
+    let taken = storage
+        .maintenance_claim(&holder, stale_after)
+        .await
+        .map_err(|e| match e {
+            StorageError::AlreadyExists(m) => anyhow::anyhow!(
+                "{m}; one maintenance command runs at a time, and a claim whose holder died \
+                 expires after stale_job_after_secs ({} s)",
+                backup.stale_job_after_secs
+            ),
+            other => anyhow::anyhow!("cannot take the maintenance claim: {other}"),
+        })?;
+    println!(
+        "Took the maintenance claim as {holder:?}: until the command ends every backup and \
+         restore job of this catalog answers LimitExceededException."
+    );
+    let claim = HeldClaim::start(taken, JOB_HEARTBEAT_INTERVAL);
+    if let Err(refusal) = wait_for_jobs(
+        &opened.catalog,
+        std::slice::from_ref(&identity),
+        &claim,
+        &mut interrupt,
+        stale_after,
+        false,
+    )
+    .await
+    {
+        release(claim).await;
+        return Err(refusal);
+    }
+
+    let migration = Migration {
+        configured,
+        identity: &identity,
+        claim: &claim,
+        interrupt: &interrupt,
+        limits: RestorableLimits::from(&app_config.limits),
+        data_file_target_bytes: backup.data_file_target_bytes,
+        region: &app_config.server.region,
+        source_backend: &opened.catalog.backend,
+        source_extenddb_version: version,
+        metrics: None,
+    };
+    let selection = arn.map_or(Selection::All, Selection::Arn);
+    let report = migrate(storage, &migration, selection, print_outcome).await;
+    release(claim).await;
+
+    let migrated = report.count(|o| matches!(o, Outcome::Migrated { .. }));
+    let finished = report.count(|o| matches!(o, Outcome::Finished | Outcome::InStore));
+    let refused = report.count(|o| matches!(o, Outcome::Refused(_)));
+    let skipped = report.count(|o| matches!(o, Outcome::Skipped(_)));
+    let failed = report.count(|o| matches!(o, Outcome::Failed(_)));
+    println!();
+    if report.backups.is_empty() && report.stopped.is_none() {
+        println!("No in-catalog backup to migrate.");
+    }
+    println!(
+        "=== extenddb backup migrate-legacy {}: {migrated} migrated, {finished} finished, \
+         {refused} refused, {skipped} skipped, {failed} failed ===",
+        if report.stopped.is_some() {
+            "stopped"
+        } else {
+            "complete"
+        }
+    );
+    if let Some(reason) = &report.stopped {
+        anyhow::bail!(
+            "stopped before the remaining backups: {reason}. The maintenance claim is released; \
+             run the command again for the rest"
+        );
+    }
+    if !report.complete() {
+        anyhow::bail!(
+            "{} backup(s) were not migrated; see above",
+            refused + skipped + failed
+        );
+    }
+    Ok(())
+}
+
+/// One backup's lines, and the audit line of a migration.
+fn print_outcome(done: &BackupOutcome) {
+    let BackupOutcome {
+        account_id,
+        backup_arn,
+        outcome,
+        notes,
+    } = done;
+    match outcome {
+        Outcome::Migrated {
+            item_count,
+            size_bytes,
+        } => {
+            tracing::info!(
+                target: "extenddb::audit",
+                action = "backup migrate-legacy",
+                account_id = %account_id,
+                backup_arn = %backup_arn,
+                "copied an in-catalog backup into the store and replaced it by its catalog row"
+            );
+            println!(
+                "migrated {backup_arn} (account {account_id}): {item_count} items, {size_bytes} \
+                 bytes"
+            );
+        }
+        Outcome::Finished => {
+            tracing::info!(
+                target: "extenddb::audit",
+                action = "backup migrate-legacy",
+                account_id = %account_id,
+                backup_arn = %backup_arn,
+                "removed what was left of an in-catalog backup whose catalog row was written"
+            );
+            println!(
+                "finished {backup_arn} (account {account_id}): an earlier run had written its \
+                 row; what was left of the in-catalog backup is removed"
+            );
+        }
+        Outcome::InStore => println!(
+            "{backup_arn} (account {account_id}) is already in the store: nothing to migrate"
+        ),
+        Outcome::Refused(reason) => println!(
+            "refused {backup_arn} (account {account_id}): {reason}. Its in-catalog backup is \
+             kept"
+        ),
+        Outcome::Skipped(reason) => {
+            println!("skipped {backup_arn} (account {account_id}): {reason}")
+        }
+        Outcome::Failed(reason) => println!(
+            "failed {backup_arn} (account {account_id}): {reason}. Its in-catalog backup is \
+             kept; a later run tries again"
+        ),
+    }
+    for note in notes {
+        println!("  {note}");
+    }
+}

crates/app/src/lib.rs

@@ -69,8 +69,8 @@ enum Command {
     Init(cmd_init::InitArgs),
     /// Tear down a extenddb deployment (the databases; the backup store is left untouched)
     Destroy(cmd_destroy::DestroyArgs),
-    /// List, purge or sync the backups of the configured backup store, or remove a restore
-    /// target
+    /// List, purge or sync the backups of the configured backup store, migrate in-catalog
+    /// backups into it, or remove a restore target
     Backup(cmd_backup::BackupArgs),
     /// Validate a extenddb deployment
     Verify(cmd_verify::VerifyArgs),

crates/core/src/backup_format/manifest.rs

@@ -130,6 +130,14 @@ impl SnapshotDescriptor {
     /// MongoDB: a session with snapshot read concern; the marker is the
     /// cluster time `<seconds>:<increment>`.
     pub const KIND_MONGODB_SNAPSHOT_READ_CONCERN: &'static str = "mongodb_snapshot_read_concern";
+    /// A copy of an in-catalog backup, written by `extenddb backup
+    /// migrate-legacy`; the marker is empty. The items are those the
+    /// in-catalog backup stored when it was taken, so the snapshot is the
+    /// one its legacy create took. For this kind `source_backend` and
+    /// `source_extenddb_version` name the backend and the release that
+    /// wrote the copy, not the release that took the in-catalog backup,
+    /// whose version was never recorded.
+    pub const KIND_LEGACY_IN_CATALOG: &'static str = "legacy_in_catalog";
 }
 
 /// Order of items within and across a backup's data files.
@@ -1456,8 +1464,10 @@ pub fn check_format_version(found: u32) -> Result<(), FormatError> {
 /// only copy of the alphabet: the engine's `validate_name_member`
 /// (`BackupName` on `CreateBackup`) and `validate_index_name` spell it again,
 /// and the engine counts the bounds in characters where this counts bytes
-/// (the same for this ASCII alphabet).
-fn is_valid_backup_name(name: &str) -> bool {
+/// (the same for this ASCII alphabet). `extenddb backup migrate-legacy`
+/// checks an in-catalog backup's name with it before it writes the copy.
+#[must_use]
+pub fn is_valid_backup_name(name: &str) -> bool {
     (3..=255).contains(&name.len()) && crate::validation::validate_table_name_chars(name).is_ok()
 }
 
@@ -1853,6 +1863,29 @@ mod tests {
         assert_eq!(parsed, manifest);
     }
 
+    #[test]
+    fn a_manifest_of_the_legacy_in_catalog_kind_parses_and_round_trips() {
+        assert_eq!(
+            SnapshotDescriptor::KIND_LEGACY_IN_CATALOG,
+            "legacy_in_catalog"
+        );
+        let mut manifest = sample_manifest();
+        manifest.snapshot = SnapshotDescriptor {
+            kind: SnapshotDescriptor::KIND_LEGACY_IN_CATALOG.to_owned(),
+            marker: String::new(),
+        };
+        let bytes = manifest.to_json_bytes().unwrap();
+        let text = std::str::from_utf8(&bytes).unwrap();
+        assert!(
+            text.contains(r#""snapshot":{"kind":"legacy_in_catalog","marker":""}"#),
+            "the kind is written as named: {text}"
+        );
+        let parsed = ExtenddbManifest::from_json_bytes(&bytes).unwrap();
+        assert_eq!(parsed, manifest);
+        assert_eq!(parsed.snapshot.kind, "legacy_in_catalog");
+        assert_eq!(parsed.snapshot.marker, "");
+    }
+
     #[test]
     fn snapshot_descriptor_serde_round_trip() {
         let snapshot = SnapshotDescriptor {

crates/core/src/backup_format/mod.rs

@@ -94,5 +94,5 @@ pub use layout::{
 pub use manifest::{
     CURRENT_FORMAT_VERSION, DataFileEntry, DataFileReader, ExtenddbManifest, ItemOrder, KeyRange,
     KeyRangeCheck, KeyRangeTracker, SnapshotDescriptor, SourceTableSchema, check_format_version,
-    verify_manifest_checksum,
+    is_valid_backup_name, verify_manifest_checksum,
 };