fix(postgres): build a GSI added by UpdateTable correctly while writes continue

.github/workflows/integration.yml

@@ -300,12 +300,13 @@ jobs:
       # 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.
+      # 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
+        run: cargo test --release -p extenddb-storage-postgres --test vector_control_plane --test put_create_race --test twi_conflict --test stream_shard_retention --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.

crates/core/src/types/mod.rs

@@ -55,7 +55,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,

crates/core/src/types/table.rs

@@ -589,6 +589,14 @@ 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 adds or deletes a GSI while another
+/// 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

crates/core/src/validation/mod.rs

@@ -492,9 +492,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(())

crates/storage-postgres/Cargo.toml

@@ -35,4 +35,5 @@ aes-gcm = { workspace = true }
 async-trait = { workspace = true }
 zeroize = { workspace = true }
 urlencoding = { workspace = true }
+libc = { workspace = true }
 

crates/storage-postgres/src/data/ddl.rs

@@ -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,71 @@ 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.
+    pub(crate) async fn drop_table_storage(
+        data: &sqlx::PgPool,
+        table_id: &str,
+        index_ids: &[String],
+        vector_index_ids: &[String],
+    ) -> 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()))?;
+        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(())
+    }
+}

crates/storage-postgres/src/data/delete_item.rs

@@ -9,6 +9,7 @@ use extenddb_storage::StreamCapture;
 use extenddb_storage::error::StorageError;
 use extenddb_storage::util::{SortKeyValue, parse_sk, pk_to_text, sk_column, sk_info};
 
+use super::gsi_build::fix_up_stale_write;
 use super::index::{enqueue_async_indexes, fetch_write_path_indexes, sync_indexes};
 use super::query::check_condition;
 use super::tx_helpers::write_stream_record_in_tx;
@@ -47,12 +48,12 @@ impl PostgresEngine {
         // cached membership gate, and that is gone: an index takes effect the moment
         // its catalog row commits. What remains is a window between this read and
         // the data transaction's commit, in which an index created concurrently is
-        // missed. That window cannot be closed here, because the catalog and the
-        // data tables are different databases and no transaction spans them. It is
-        // also exactly the window the secondary indexes have, for the same reason
-        // and with the same read: parity with a GSI is the bar, and the backfill
-        // that publishes a new index is what covers writes older than it.
-        let (indexes, vector_metas) =
+        // missed. No transaction can close it, because the catalog and the data
+        // tables are different databases. For a GSI it is closed by time instead:
+        // the read's stamp lets a write that was slow to commit queue a repair
+        // for any GSI it did not see (`gsi_build::fix_up_stale_write`), and a fast
+        // write is one the backfill sees. Vector indexes still have the window.
+        let (indexes, vector_metas, stamp) =
             fetch_write_path_indexes(&key_info.table_id, &self.pool).await?;
         // Read whenever anything can propagate, secondary or vector. Gating this on
         // the secondary set alone made a vector-only table ignore the configured
@@ -65,11 +66,15 @@ impl PostgresEngine {
             self.index_propagation_delay().await
         };
 
+        // A write that is already slow takes the transaction path even with no
+        // index to maintain, so that a repair for a GSI created meanwhile commits
+        // with the write (`gsi_build::fix_up_stale_write`) instead of after it.
         let needs_tx = condition.is_some()
             || return_old
             || !indexes.is_empty()
             || !vector_metas.is_empty()
-            || stream.is_some();
+            || stream.is_some()
+            || stamp.elapsed() > super::gsi_build::GRACE_WRITE;
 
         if let Some((sk_name, sk_type)) =
             sk_info(&key_info.key_schema, &key_info.attribute_definitions)
@@ -155,7 +160,6 @@ impl PostgresEngine {
                     sync_indexes(
                         &mut tx,
                         &key_info.key_schema,
-                        &key_info.attribute_definitions,
                         &indexes,
                         oi.as_ref(),
                         None,
@@ -213,6 +217,9 @@ impl PostgresEngine {
                 )
                 .await?;
 
+                async_enqueued +=
+                    fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, key).await?;
+
                 tx.commit()
                     .await
                     .map_err(|e| StorageError::Internal(e.to_string()))?;
@@ -254,6 +261,7 @@ impl PostgresEngine {
                     }
                 }
                 .map_err(|e| StorageError::Internal(e.to_string()))?;
+                self.fix_up_autocommit(stamp, key_info, key).await;
                 Ok(None)
             }
         } else {
@@ -313,7 +321,6 @@ impl PostgresEngine {
                     sync_indexes(
                         &mut tx,
                         &key_info.key_schema,
-                        &key_info.attribute_definitions,
                         &indexes,
                         oi.as_ref(),
                         None,
@@ -371,6 +378,9 @@ impl PostgresEngine {
                 )
                 .await?;
 
+                async_enqueued +=
+                    fix_up_stale_write(&mut tx, &self.pool, stamp, key_info, &indexes, key).await?;
+
                 tx.commit()
                     .await
                     .map_err(|e| StorageError::Internal(e.to_string()))?;
@@ -393,6 +403,7 @@ impl PostgresEngine {
                     .execute(&self.data_pool)
                     .await
                     .map_err(|e| StorageError::Internal(e.to_string()))?;
+                self.fix_up_autocommit(stamp, key_info, key).await;
                 Ok(None)
             }
         }