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

.github/workflows/integration.yml

@@ -296,13 +296,17 @@ jobs:
         run: devtools/run-tests --extenddb --rust-integration --release
 
       # 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. They build their own
-      # throwaway databases; the connection string is the server, not a database.
+      # this backend declares no vector search capability, and the backup restore
+      # cases here need catalog rows the current binary would not write, so these
+      # tests drive the storage layer directly against this job's PostgreSQL. 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
+        run: >-
+          cargo test --release -p extenddb-storage-postgres
+          --test vector_control_plane --test backup_restore
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.

crates/core/src/error/mod.rs

@@ -24,6 +24,8 @@ pub enum DynamoDbError {
     #[error("{0}")]
     BackupNotFoundException(String),
     #[error("{0}")]
+    BackupInUseException(String),
+    #[error("{0}")]
     ResourceInUseException(String),
     /// A per-table or per-account limit was exceeded.
     ///
@@ -114,6 +116,7 @@ impl DynamoDbError {
             Self::ValidationException(_)
             | Self::ResourceNotFoundException(_)
             | Self::BackupNotFoundException(_)
+            | Self::BackupInUseException(_)
             | Self::ResourceInUseException(_)
             | Self::LimitExceededException(_)
             | Self::ConditionalCheckFailedException(..)
@@ -155,6 +158,7 @@ impl DynamoDbError {
             Self::ValidationException(_) => "ValidationException",
             Self::ResourceNotFoundException(_) => "ResourceNotFoundException",
             Self::BackupNotFoundException(_) => "BackupNotFoundException",
+            Self::BackupInUseException(_) => "BackupInUseException",
             Self::ResourceInUseException(_) => "ResourceInUseException",
             Self::LimitExceededException(_) => "LimitExceededException",
             Self::ConditionalCheckFailedException(..) => "ConditionalCheckFailedException",
@@ -221,6 +225,7 @@ impl DynamoDbError {
             Self::ValidationException(m)
             | Self::ResourceNotFoundException(m)
             | Self::BackupNotFoundException(m)
+            | Self::BackupInUseException(m)
             | Self::ResourceInUseException(m)
             | Self::LimitExceededException(m)
             | Self::ConditionalCheckFailedException(m, _)
@@ -317,6 +322,7 @@ mod tests {
             (DynamoDbError::ResourceInUseException(String::new()), 400),
             (DynamoDbError::ResourceNotFoundException(String::new()), 400),
             (DynamoDbError::BackupNotFoundException(String::new()), 400),
+            (DynamoDbError::BackupInUseException(String::new()), 400),
             (DynamoDbError::SerializationException(String::new()), 400),
             (DynamoDbError::ServiceUnavailable(String::new()), 503),
             (DynamoDbError::ThrottlingException(String::new()), 400),

crates/engine/src/backup.rs

@@ -145,6 +145,7 @@ pub(crate) async fn handle_restore_table_from_backup(
                     .to_owned(),
             )
         })?;
+    extenddb_core::validation::validate_table_name(target_table_name, &ctx.limits)?;
     let backup_arn = backup_arn_field(&body, &ctx.account_id)?;
 
     let mut desc = ctx
@@ -156,16 +157,21 @@ pub(crate) async fn handle_restore_table_from_backup(
     // The restore response's TableDescription reports where the data came
     // from and that the restore is under way: SourceBackupArn and
     // RestoreInProgress: true, pinned by the ground-truth runs of 2026-08-24
-    // (us-east-1 and eu-west-2). Set here rather than in each backend because
-    // the summary is response metadata about this call, not table state the
-    // backends persist.
+    // (us-east-1 and eu-west-2). The service returns the table CREATING with
+    // the restore in progress; the backends report CREATING here too, whatever
+    // the copy has reached. The time is the backend's own record where it
+    // keeps one, so the response and later DescribeTable calls agree.
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map(|d| d.as_secs_f64())
         .unwrap_or_default();
+    let restore_date_time = desc
+        .restore_summary
+        .as_ref()
+        .map_or(now, |r| r.restore_date_time);
     desc.restore_summary = Some(extenddb_core::types::RestoreSummary {
         source_backup_arn: Some(backup_arn.clone()),
-        restore_date_time: now,
+        restore_date_time,
         restore_in_progress: true,
     });
 
@@ -292,6 +298,12 @@ fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> DynamoDbEr
         extenddb_storage::error::StorageError::Unsupported(msg) => {
             DynamoDbError::ValidationException(msg)
         }
+        extenddb_storage::error::StorageError::LimitExceeded(msg) => {
+            DynamoDbError::LimitExceededException(msg)
+        }
+        extenddb_storage::error::StorageError::BackupInUse(msg) => {
+            DynamoDbError::BackupInUseException(msg)
+        }
         other => {
             tracing::error!(internal_error = %other, "backup storage error");
             DynamoDbError::InternalServerError("Internal server error".to_owned())

crates/engine/src/create_table.rs

@@ -134,6 +134,7 @@ pub(crate) fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) ->
         // state: the documented delete-table sentence and the measured
         // phase-dependent vector refusal both arrive through this one arm.
         StorageError::IndexesInUse(msg) => DynamoDbError::ResourceInUseException(msg),
+        StorageError::BackupInUse(msg) => DynamoDbError::BackupInUseException(msg),
         StorageError::LimitExceeded(msg) => DynamoDbError::LimitExceededException(msg), // Retryable by definition, so it maps like Connection: a 503 the SDKs
         // retry, rather than a 500 they surface.
         StorageError::Transient(msg) => {

crates/storage-mongodb/src/backup_engine.rs

@@ -502,6 +502,30 @@ impl BackupEngine for MongoEngine {
                     StorageError::Validation(format!("Backup not found: {backup_arn}"))
                 })?;
 
+            // Refuse while a restore from this backup is still running, as the
+            // service does.
+            let restoring = self
+                .catalog_db
+                .collection::<Document>("tables")
+                .find_one(doc! {
+                    "_id.account_id": &account_id,
+                    "restore_source_backup_arn": &backup_arn,
+                    "table_status": "CREATING",
+                })
+                .await
+                .map_err(|e| StorageError::Internal(e.to_string()))?;
+            if let Some(t) = restoring {
+                let name = t
+                    .get_document("_id")
+                    .ok()
+                    .and_then(|id| id.get_str("table_name").ok())
+                    .unwrap_or("?")
+                    .to_owned();
+                return Err(StorageError::BackupInUse(format!(
+                    "Backup is being used to restore table {name}: {backup_arn}"
+                )));
+            }
+
             // Drop the backup collection. If backup_id is absent (e.g., a
             // pre-`$out` backup on an old catalog) we skip — nothing to drop
             // at the collection level in that case.
@@ -641,10 +665,34 @@ impl BackupEngine for MongoEngine {
             // Create the table with the ACTIVE transition deferred: it enters
             // CREATING with no scheduled flip, so the table cannot become
             // ACTIVE until we schedule it below, after the data copy completes.
-            let desc = self
+            let mut desc = self
                 .create_table_impl(&account_id, create_input, true)
                 .await?;
 
+            // Provenance, recorded before the copy: DescribeTable reports it as
+            // RestoreSummary, and DeleteBackup refuses while a table still
+            // CREATING names this backup.
+            let restore_at = bson::DateTime::now();
+            self.catalog_db
+                .collection::<Document>("tables")
+                .update_one(
+                    doc! { "_id": { "account_id": &account_id, "table_name": &target_table_name } },
+                    doc! { "$set": {
+                        "restore_source_backup_arn": &backup_arn,
+                        "restore_date_time": restore_at,
+                    } },
+                )
+                .await
+                .map_err(|e| StorageError::Internal(e.to_string()))?;
+            #[allow(clippy::cast_precision_loss)]
+            {
+                desc.restore_summary = Some(extenddb_core::types::RestoreSummary {
+                    source_backup_arn: Some(backup_arn.clone()),
+                    restore_date_time: restore_at.timestamp_millis() as f64 / 1000.0,
+                    restore_in_progress: true,
+                });
+            }
+
             // Restore items from the backup collection using server-side `$out`.
             // The backup collection was written by `create_backup` in the same
             // document shape as the source data collection, so this is a

crates/storage-mongodb/src/table_engine.rs

@@ -1456,6 +1456,20 @@ impl MongoEngine {
         let on_demand_throughput: Option<OnDemandThroughput> = doc
             .get("on_demand_throughput")
             .and_then(|b| bson::from_bson(b.clone()).ok());
+        // Set on a table created by RestoreTableFromBackup; in progress until
+        // the table is ACTIVE (the restore's index backfill runs while it is
+        // CREATING).
+        #[allow(clippy::cast_precision_loss)]
+        let restore_summary = doc.get_str("restore_source_backup_arn").ok().map(|arn| {
+            extenddb_core::types::RestoreSummary {
+                source_backup_arn: Some(arn.to_owned()),
+                restore_date_time: doc
+                    .get_datetime("restore_date_time")
+                    .map(|d| d.timestamp_millis() as f64 / 1000.0)
+                    .unwrap_or(0.0),
+                restore_in_progress: table_status == TableStatus::Creating,
+            }
+        });
 
         Ok(TableDescription {
             table_name,
@@ -1479,6 +1493,7 @@ impl MongoEngine {
             sse_description,
             table_class_summary,
             on_demand_throughput,
+            restore_summary,
             // Fields for features this backend does not implement, vector
             // indexes today, take their defaults. Adding one to
             // TableDescription then does not break this build.

crates/storage-postgres/migrations/003_backup_definitions.sql

@@ -0,0 +1,41 @@
+-- Copyright 2026 ExtendDB contributors
+-- SPDX-License-Identifier: Apache-2.0
+-- Migration 003: record a backup's table definition, and which backup a
+-- restored table came from (catalog version 0.0.4).
+--
+-- A backup kept the source table's key schema, attribute definitions, and
+-- billing mode, and nothing else, so a restored table came back without its
+-- global and local secondary indexes, with 5/5 provisioned throughput, and
+-- without its table class or encryption settings. One row per backup holds
+-- those, in the wire's own shape behind a version marker (see
+-- `extenddb_storage::backup_definition`). A backup taken before this
+-- migration has no row and restores as before.
+--
+-- Written to tolerate a replay, like 002: the runner applies a migration and
+-- records it in `schema_history` as two separate commits, so a crash in
+-- between leaves this file applied but unrecorded, and the next migrate runs
+-- it again. CREATE TABLE IF NOT EXISTS and the version UPDATE are idempotent.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS backup_definitions (
+    backup_arn TEXT PRIMARY KEY REFERENCES backups(backup_arn) ON DELETE CASCADE,
+    definition JSONB NOT NULL
+);
+
+-- One row per table created by RestoreTableFromBackup: the backup it came
+-- from and when, reported by DescribeTable as RestoreSummary, and used to
+-- refuse DeleteBackup while the restore is still running. Not a foreign key
+-- to backups: the backup may be deleted after the restore, and the summary
+-- keeps naming it, as on the service.
+CREATE TABLE IF NOT EXISTS table_restores (
+    table_id TEXT PRIMARY KEY REFERENCES tables(table_id) ON DELETE CASCADE,
+    source_backup_arn TEXT NOT NULL,
+    restore_date_time TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_table_restores_backup ON table_restores (source_backup_arn);
+
+UPDATE settings SET value = '0.0.4' WHERE key = 'catalog_version';
+
+COMMIT;