robinnsc · Oct 23, 2026
@@ -301,15 +301,17 @@ jobs:
       # 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.
+      # direct access. The online GSI build tests reorder queue rows by hand and
+      # pause writers with a test hook. 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 --test backup_restore
+          --test gsi_online_build --test gsi_stale_writer
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.

@@ -56,7 +56,7 @@ pub use table::{
     DescribeTimeToLiveOutput, DistanceFunction, GlobalSecondaryIndexUpdate, GsiDescription,
     GsiInput, IndexStatus, ListTablesInput, ListTablesOutput, ListTagsOfResourceInput,
     ListTagsOfResourceOutput, LsiDescription, LsiInput, MAX_VECTOR_INDEXES_PER_TABLE,
-    OnDemandThroughput, Projection, ProjectionType, ProvisionedThroughput,
+    ONLINE_INDEX_LIMIT, OnDemandThroughput, Projection, ProjectionType, ProvisionedThroughput,
     ProvisionedThroughputDescription, RestoreSummary, SearchSchemaElement, SearchSchemaElementType,
     SseDescription, SseType, StreamSpecification, StreamViewType, TableDescription, TableStatus,
     Tag, TagResourceInput, TimeToLiveDescription, TimeToLiveSpecification,

@@ -615,6 +615,15 @@ pub const VECTOR_TABLE_REQUIRES_PAY_PER_REQUEST_MODE: &str = "One or more parame
 pub const VECTOR_INDEX_COUNT_LIMIT_CREATE: &str =
     "One or more parameter values were invalid: VectorIndex count exceeds the per-table limit of 5";
 
+/// A second online GSI change while one is still being built.
+///
+/// The service answers an UpdateTable that creates a GSI, or deletes a GSI other
+/// than the one building, while a GSI on the table is `CREATING` with this
+/// LimitExceededException; the same sentence it uses for two index actions in
+/// one request.
+pub const ONLINE_INDEX_LIMIT: &str = "Subscriber limit exceeded: Only 1 online index can be \
+     created or deleted simultaneously per table";
+
 /// Per-table vector index limit exceeded on `UpdateTable`.
 ///
 /// Measured 2026-08-13: reported as a `LimitExceededException`, not a

@@ -579,9 +579,7 @@ pub fn validate_vector_index_updates(
         .sum::<usize>();
     if actions > 1 {
         return Err(DynamoDbError::LimitExceededException(
-            "Subscriber limit exceeded: Only 1 online index can be created or deleted \
-             simultaneously per table"
-                .to_owned(),
+            crate::types::ONLINE_INDEX_LIMIT.to_owned(),
         ));
     }
     Ok(())

@@ -35,4 +35,5 @@ aes-gcm = { workspace = true }
 async-trait = { workspace = true }
 zeroize = { workspace = true }
 urlencoding = { workspace = true }
+libc = { workspace = true }
 
@@ -54,6 +54,7 @@ impl DataEngine for PostgresEngine {
                 stream.as_ref(),
             )
             .await
+            .map_err(|e| table_gone(e, &key_info.table_name))
         })
     }
 
@@ -64,7 +65,11 @@ impl DataEngine for PostgresEngine {
     ) -> BoxFuture<'_, Result<Option<Item>, StorageError>> {
         let key_info = key_info.clone();
         let key = key.clone();
-        Box::pin(async move { self.get_item_impl(&key_info, &key).await })
+        Box::pin(async move {
+            self.get_item_impl(&key_info, &key)
+                .await
+                .map_err(|e| table_gone(e, &key_info.table_name))
+        })
     }
 
     fn delete_item(
@@ -91,6 +96,7 @@ impl DataEngine for PostgresEngine {
                 stream.as_ref(),
             )
             .await
+            .map_err(|e| table_gone(e, &key_info.table_name))
         })
     }
 
@@ -123,6 +129,7 @@ impl DataEngine for PostgresEngine {
                 stream.as_ref(),
             )
             .await
+            .map_err(|e| table_gone(e, &key_info.table_name))
         })
     }
 
@@ -152,6 +159,7 @@ impl DataEngine for PostgresEngine {
                 index_name.as_deref(),
             )
             .await
+            .map_err(|e| read_target_gone(e, &key_info.table_name, index_name.as_deref()))
         })
     }
 
@@ -177,6 +185,7 @@ impl DataEngine for PostgresEngine {
                 index_name.as_deref(),
             )
             .await
+            .map_err(|e| read_target_gone(e, &key_info.table_name, index_name.as_deref()))
         })
     }
 
@@ -227,7 +236,13 @@ impl DataEngine for PostgresEngine {
                 .iter()
                 .map(|(key_info, key)| TransactGetOp { key_info, key })
                 .collect();
-            self.transact_get_items_impl(&borrowed_ops).await
+            let tables: Vec<(&str, &str)> = owned_ops
+                .iter()
+                .map(|(k, _)| (k.table_id.as_str(), k.table_name.as_str()))
+                .collect();
+            self.transact_get_items_impl(&borrowed_ops)
+                .await
+                .map_err(|e| table_gone_among(e, &tables))
         })
     }
 
@@ -376,6 +391,17 @@ impl DataEngine for PostgresEngine {
                     },
                 )
                 .collect();
+            let tables: Vec<(&str, &str)> = borrowed_ops
+                .iter()
+                .map(|op| match op {
+                    TransactWriteOp::Put { key_info, .. }
+                    | TransactWriteOp::Delete { key_info, .. }
+                    | TransactWriteOp::Update { key_info, .. }
+                    | TransactWriteOp::ConditionCheck { key_info, .. } => {
+                        (key_info.table_id.as_str(), key_info.table_name.as_str())
+                    }
+                })
+                .collect();
             self.transact_write_items_impl(
                 &borrowed_ops,
                 idempotency
@@ -387,6 +413,7 @@ impl DataEngine for PostgresEngine {
                     }),
             )
             .await
+            .map_err(|e| table_gone_among(e, &tables))
         })
     }
 
@@ -400,3 +427,116 @@ impl DataEngine for PostgresEngine {
         })
     }
 }
+
+/// A data table that is gone mid-request is a table that is gone.
+///
+/// A request that resolved the table just before a DeleteTable committed reaches
+/// the data database after the drop, and PostgreSQL answers 42P01. The service
+/// answers ResourceNotFoundException for a table being deleted, which is what
+/// `TableNotFound` becomes on the wire; passing the 42P01 through was a 500.
+fn table_gone(e: StorageError, table_name: &str) -> StorageError {
+    if data_table_missing(&e) {
+        StorageError::TableNotFound(table_name.to_owned())
+    } else {
+        e
+    }
+}
+
+/// [`table_gone`] for a transaction over several tables: the table named is the
+/// one whose data table the error names, or the first one if the error names an
+/// index table (dropped with its table) or is not specific.
+fn table_gone_among(e: StorageError, tables: &[(&str, &str)]) -> StorageError {
+    if !data_table_missing(&e) {
+        return e;
+    }
+    let StorageError::Internal(msg) = &e else {
+        return e;
+    };
+    let name = tables
+        .iter()
+        .find(|(id, _)| msg.contains(&format!("_ddb_{id}\"")))
+        .or(tables.first())
+        .map(|(_, name)| (*name).to_owned());
+    match name {
+        Some(name) => StorageError::TableNotFound(name),
+        None => e,
+    }
+}
+
+/// [`table_gone`] for a Query or Scan, which may name an index: a 42P01 there can
+/// be the index's table, dropped by a GSI delete that committed after the index
+/// was resolved, and the service answers that as an index it does not have.
+fn read_target_gone(e: StorageError, table_name: &str, index_name: Option<&str>) -> StorageError {
+    match index_name {
+        Some(index) if data_table_missing(&e) => StorageError::IndexNotFound(index.to_owned()),
+        _ => table_gone(e, table_name),
+    }
+}
+
+/// Whether an error is PostgreSQL's undefined_table for one of this backend's
+/// data tables. Many write paths render the error with `to_string()`, which keeps
+/// the message and drops the SQLSTATE, so the message is matched as well: it
+/// names the relation, and every data table's name starts `_ddb_`.
+fn data_table_missing(e: &StorageError) -> bool {
+    match e {
+        StorageError::Internal(msg) => {
+            crate::gsi_queue::is_undefined_table(e)
+                || (msg.contains("relation \"_ddb_") && msg.contains("does not exist"))
+        }
+        _ => false,
+    }
+}
+
+#[cfg(test)]
+mod table_gone_tests {
+    use super::*;
+
+    #[test]
+    fn a_missing_data_table_is_a_missing_table_with_or_without_the_sqlstate() {
+        for msg in [
+            "SQLSTATE 42P01: error returned from database: relation \"_ddb_abc\" does not exist",
+            "error returned from database: relation \"_ddb_abc\" does not exist",
+        ] {
+            let e = table_gone(StorageError::Internal(msg.to_owned()), "t");
+            assert!(
+                matches!(e, StorageError::TableNotFound(ref n) if n == "t"),
+                "{msg}"
+            );
+        }
+    }
+
+    #[test]
+    fn other_errors_pass_through() {
+        let e = table_gone(StorageError::Internal("deadlock detected".to_owned()), "t");
+        assert!(matches!(e, StorageError::Internal(_)));
+        let e = table_gone(
+            StorageError::Internal("relation \"settings\" does not exist".to_owned()),
+            "t",
+        );
+        assert!(matches!(e, StorageError::Internal(_)), "not a data table");
+    }
+
+    #[test]
+    fn a_transaction_names_the_table_whose_data_table_is_gone() {
+        let tables = [("id-a", "a"), ("id-b", "b")];
+        let e = table_gone_among(
+            StorageError::Internal("relation \"_ddb_id-b\" does not exist".to_owned()),
+            &tables,
+        );
+        assert!(matches!(e, StorageError::TableNotFound(ref n) if n == "b"));
+        let e = table_gone_among(
+            StorageError::Internal("relation \"_ddb_some-index\" does not exist".to_owned()),
+            &tables,
+        );
+        assert!(matches!(e, StorageError::TableNotFound(ref n) if n == "a"));
+        let e = table_gone_among(StorageError::Internal("deadlock".to_owned()), &tables);
+        assert!(matches!(e, StorageError::Internal(_)));
+    }
+
+    #[test]
+    fn a_missing_index_table_on_a_read_of_an_index_is_a_missing_index() {
+        let msg = "relation \"_ddb_idx\" does not exist".to_owned();
+        let e = read_target_gone(StorageError::Internal(msg), "t", Some("gsi1"));
+        assert!(matches!(e, StorageError::IndexNotFound(ref n) if n == "gsi1"));
+    }
+}
@@ -599,14 +599,20 @@ impl PostgresEngine {
         table_id: &str,
         index_name: &str,
     ) -> Result<IndexInfo, StorageError> {
-        let idx_row: Option<(String, String, serde_json::Value, serde_json::Value)> = sqlx::query_as(
-            "SELECT index_type, index_id, key_schema, projection FROM indexes WHERE table_id = $1 AND index_name = $2",
-        )
-        .bind(table_id)
-        .bind(index_name)
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(|e| StorageError::Internal(e.to_string()))?;
+        // A GSI still being built is not readable: its table is partial, or not
+        // there yet. The service answers a Query or Scan naming one with "The
+        // table does not have the specified index", the same as for no index at
+        // all, which is what `IndexNotFound` becomes on the wire.
+        let idx_row: Option<(String, String, serde_json::Value, serde_json::Value)> =
+            sqlx::query_as(
+                "SELECT index_type, index_id, key_schema, projection FROM indexes \
+             WHERE table_id = $1 AND index_name = $2 AND index_status <> 'CREATING'",
+            )
+            .bind(table_id)
+            .bind(index_name)
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
 
         let (idx_type_str, idx_id, ks_json, proj_json) =
             idx_row.ok_or_else(|| StorageError::IndexNotFound(index_name.to_owned()))?;
@@ -635,3 +641,86 @@ impl PostgresEngine {
         })
     }
 }
+
+impl PostgresEngine {
+    /// Drop a deleted table's data tables and its queue rows, in an order that
+    /// cannot deadlock with the propagation queue.
+    ///
+    /// The worker claims a `gsi_pending` row, then takes `FOR SHARE` on the base
+    /// row (the state-based apply). A drop that takes the base table first and
+    /// then deletes the pending rows waits on the worker's claimed row while the
+    /// worker waits on the table: a cycle, and when this side is the victim the
+    /// catalog rows are already gone and the data tables stay behind.
+    ///
+    /// So the queue is stopped first. A `vector_index_holds` row for the table,
+    /// committed on its own so the claim query can see it, keeps the worker from
+    /// taking any more of the table's rows (the claim already honours holds, for
+    /// vector builds). Then the pending rows are deleted, which waits for the
+    /// claims already in flight to commit, each of which holds the base row only
+    /// until then. Only after that are the tables dropped, with nothing left that
+    /// could be waiting on them. The hold is removed in the same transaction as
+    /// the drops, so a table id that no longer exists is never left held; a
+    /// crash in between leaves a hold the recovery sweep clears by age (ten
+    /// minutes for a hold of this kind, which a drop waiting behind a long
+    /// client transaction can legitimately keep for a while).
+    ///
+    /// What this does not cover is a deadlock that does not go through the
+    /// queue: the drop waits on a client transaction that waits, through another
+    /// table, on a client transaction queued behind the drop. That is the same
+    /// exposure any DDL has here and was there before.
+    ///
+    /// With `lock_timeout`, every wait in the dropping transaction (the pending
+    /// rows' claims, a CreateBackup reading the table, a client transaction) is
+    /// bounded; a timeout fails the call with nothing dropped, and the hold
+    /// stays until the next attempt or the recovery sweep. The control plane
+    /// worker uses it so one table cannot stall a whole pass; DeleteTable itself
+    /// waits.
+    pub(crate) async fn drop_table_storage(
+        data: &sqlx::PgPool,
+        table_id: &str,
+        index_ids: &[String],
+        vector_index_ids: &[String],
+        lock_timeout: Option<&str>,
+    ) -> Result<(), StorageError> {
+        sqlx::query(
+            "INSERT INTO vector_index_holds (table_id, index_id) VALUES ($1, $1) \
+             ON CONFLICT (table_id, index_id) DO NOTHING",
+        )
+        .bind(table_id)
+        .execute(data)
+        .await
+        .map_err(|e| StorageError::Internal(e.to_string()))?;
+        let mut tx = data
+            .begin()
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        if let Some(timeout) = lock_timeout {
+            sqlx::query("SELECT set_config('lock_timeout', $1, true)")
+                .bind(timeout)
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| StorageError::Internal(e.to_string()))?;
+        }
+        sqlx::query("DELETE FROM gsi_pending WHERE table_id = $1")
+            .bind(table_id)
+            .execute(&mut *tx)
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        Self::drop_data_table(&mut tx, table_id).await?;
+        for idx_id in index_ids {
+            Self::drop_index_data_table(&mut tx, idx_id).await?;
+        }
+        for idx_id in vector_index_ids {
+            Self::drop_vector_data_table(&mut tx, idx_id).await?;
+        }
+        sqlx::query("DELETE FROM vector_index_holds WHERE table_id = $1")
+            .bind(table_id)
+            .execute(&mut *tx)
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        tx.commit()
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        Ok(())
+    }
+}