fix(backup): restore sort keys, indexes, and throughput; bound memory; recover abandoned restores

.github/workflows/integration.yml

@@ -298,14 +298,18 @@ jobs:
       # The control plane for vector indexes is not reachable over the wire while
       # this backend declares no vector search capability, so its tests drive the
       # storage layer directly against this job's PostgreSQL. The PutItem create
-      # race tests hold a create open in an outside transaction, and the
-      # write-transaction conflict tests force a deadlock; both also need direct
-      # access. They build their own throwaway databases; the connection string is
-      # the server, not a database.
+      # race tests hold a create open in an outside transaction, the
+      # write-transaction conflict tests force a deadlock, and the backup restore
+      # cases need catalog rows the current binary would not write; all need
+      # direct access. They build their own throwaway databases; the connection
+      # string is the server, not a database.
       - name: Run PostgreSQL storage-level tests
         env:
           EXTENDDB_TEST_PG_CONNECTION_STRING: postgresql://postgres:devpass@127.0.0.1:5432
-        run: cargo test --release -p extenddb-storage-postgres --test vector_control_plane --test put_create_race --test twi_conflict --test stream_shard_retention
+        run: >-
+          cargo test --release -p extenddb-storage-postgres
+          --test vector_control_plane --test put_create_race --test twi_conflict
+          --test stream_shard_retention --test backup_restore
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.

crates/app/Cargo.toml

@@ -40,3 +40,6 @@ rustls = { workspace = true }
 [target.'cfg(unix)'.dependencies]
 daemonize = { workspace = true }
 libc = { workspace = true }
+
+[dev-dependencies]
+async-trait = { workspace = true }

crates/app/src/cmd_destroy.rs

@@ -112,12 +112,17 @@ pub async fn run(args: DestroyArgs) -> anyhow::Result<()> {
     drop(bootstrap);
     let bootstrap = extenddb_storage::bootstrapper::create_bootstrapper(&args.config, &cli_args)
         .await
-        .map_err(|e| anyhow::anyhow!("Cannot connect as admin: {e:?}"))?;
+        .map_err(|e| {
+            anyhow::anyhow!(
+                "Cannot connect as admin: {}",
+                crate::cmd_migrate::storage_err(e)
+            )
+        })?;
 
     bootstrap
         .drop_databases(&data_db)
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     println!();
     println!("=== extenddb destroy complete ===");

crates/app/src/cmd_init.rs

@@ -60,7 +60,8 @@ pub struct InitArgs {
     #[arg(long)]
     cassandra_user: Option<String>,
 
-    /// Cassandra admin password
+    /// Cassandra admin password.
+    /// Prefer `EXTENDDB_CASSANDRA_PASSWORD` to keep the value out of process arguments.
     #[arg(long)]
     cassandra_pass: Option<String>,
 
@@ -199,6 +200,12 @@ pub async fn run(args: InitArgs) -> anyhow::Result<u8> {
         std::env::var_os("EXTENDDB_APP_PASSWORD"),
         "EXTENDDB_APP_PASSWORD",
     )?;
+    crate::util::append_secret_arg(
+        &mut cli_args,
+        "--cassandra-pass",
+        std::env::var_os("EXTENDDB_CASSANDRA_PASSWORD"),
+        "EXTENDDB_CASSANDRA_PASSWORD",
+    )?;
 
     // Extract bind_addr from CLI args
     let bind_addr =
@@ -213,32 +220,32 @@ pub async fn run(args: InitArgs) -> anyhow::Result<u8> {
     // Create bootstrapper via registry (no hardcoded match!)
     let bootstrapper = extenddb_storage::bootstrapper::create_bootstrapper(&args.config, &cli_args)
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::storage_err)?;
 
     // Ensure application user exists.
     bootstrapper
         .ensure_app_user()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     // Grant the application role to the admin user so CREATE DATABASE ... OWNER
     // succeeds on RDS/Aurora where the admin is not a true superuser.
     bootstrapper
         .grant_app_role_to_admin()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     // Create catalog database — abort if it already exists.
     bootstrapper
         .create_catalog_db()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     // Create data database — abort if it already exists.
     bootstrapper
         .create_data_db()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     // Take the same lock `migrate` uses around the schema work below, so an
     // init cannot race a migrate running on another replica and fail on a
@@ -254,10 +261,13 @@ pub async fn run(args: InitArgs) -> anyhow::Result<u8> {
     bootstrapper
         .acquire_migration_lock()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
     let bootstrap_result = run_init_bootstrap(bootstrapper.as_ref()).await;
     if let Err(e) = bootstrapper.release_migration_lock().await {
-        tracing::warn!("Failed to release migration lock: {e:?}");
+        tracing::warn!(
+            "Failed to release migration lock: {}",
+            crate::cmd_migrate::op_err(e)
+        );
     }
     let admin_result = bootstrap_result?;
 
@@ -326,20 +336,20 @@ async fn run_init_bootstrap(
     bootstrapper
         .bootstrap_encryption_key()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?; // REQ-AUTH-010
+        .map_err(crate::cmd_migrate::op_err)?; // REQ-AUTH-010
 
     bootstrapper
         .bootstrap_default_account()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     // REQ-AUTH-003
     let env_user = std::env::var("EXTENDDB_ADMIN_USER").ok();
     let env_pass = std::env::var("EXTENDDB_ADMIN_PASSWORD").ok();
     bootstrapper
         .bootstrap_admin_user(env_user.as_deref(), env_pass.as_deref())
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))
+        .map_err(crate::cmd_migrate::op_err)
 }
 
 async fn run_init_migrations(
@@ -349,28 +359,28 @@ async fn run_init_migrations(
     let initialized = bootstrapper
         .is_catalog_initialized()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     if initialized {
         println!("--- Catalog already initialized. Use 'extenddb migrate' for pending migrations.");
     } else {
         bootstrapper
             .run_catalog_migrations()
             .await
-            .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+            .map_err(crate::cmd_migrate::op_err)?;
     }
 
     // Record data database connection in catalog.
     bootstrapper
         .record_data_connection()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     // Initialize data database schema.
     bootstrapper
         .run_data_migrations()
         .await
-        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
+        .map_err(crate::cmd_migrate::op_err)?;
 
     Ok(())
 }