fix(storage): read GSI propagation delay live so runtime changes apply

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

@@ -39,8 +39,7 @@ impl PostgresEngine {
         let sys_delay = if indexes.is_empty() {
             0
         } else {
-            self.gsi_default_delay_ms
-                .load(std::sync::atomic::Ordering::Relaxed)
+            self.gsi_default_delay().await
         };
 
         let needs_tx = condition.is_some() || return_old || !indexes.is_empty() || stream.is_some();

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

@@ -59,8 +59,7 @@ impl PostgresEngine {
         let sys_delay = if indexes.is_empty() {
             0
         } else {
-            self.gsi_default_delay_ms
-                .load(std::sync::atomic::Ordering::Relaxed)
+            self.gsi_default_delay().await
         };
 
         // When there's a condition, return_old, indexes, or stream capture, we need a transaction

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

@@ -83,10 +83,9 @@ impl PostgresEngine {
             }
         }
 
-        // D-4: Read system default delay from cache (P119).
-        let sys_delay = self
-            .gsi_default_delay_ms
-            .load(std::sync::atomic::Ordering::Relaxed);
+        // D-4: Read the system default delay live (P119), so a runtime change
+        // applies to this transaction rather than up to 30 s later.
+        let sys_delay = self.gsi_default_delay().await;
 
         let mut tx = self
             .data_pool

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

@@ -50,8 +50,7 @@ impl PostgresEngine {
         let sys_delay = if indexes.is_empty() {
             0
         } else {
-            self.gsi_default_delay_ms
-                .load(std::sync::atomic::Ordering::Relaxed)
+            self.gsi_default_delay().await
         };
 
         // Fetch existing item

crates/storage-postgres/src/gsi_queue.rs

@@ -273,13 +273,22 @@ async fn worker(worker_id: u64, q: Arc<GsiQueue>) {
 /// backstop interval). A row already due maps to a near-zero wait so the loop
 /// re-claims promptly.
 async fn next_ready_wait(pool: &PgPool, worker_id: u64) -> Option<std::time::Duration> {
+    // The ::float8 cast is load-bearing. Since PostgreSQL 14, EXTRACT returns
+    // `numeric`, which sqlx refuses to decode into f64. Without the cast this
+    // query fails on every partition that has a pending row, and the error was
+    // swallowed into `None` below, indistinguishable from "no rows". The worker
+    // then slept its full idle backstop instead of until `ready_at`, so every
+    // asynchronous GSI propagation took ~1 s regardless of the configured
+    // delay. Errors are logged now so a decode or connection failure can never
+    // silently degrade propagation latency again.
     let secs: Option<f64> = sqlx::query_scalar(
-        "SELECT EXTRACT(EPOCH FROM (MIN(ready_at) - NOW())) FROM gsi_pending \
+        "SELECT EXTRACT(EPOCH FROM (MIN(ready_at) - NOW()))::float8 FROM gsi_pending \
          WHERE worker_partition = $1",
     )
     .bind(worker_id as i32)
     .fetch_one(pool)
     .await
+    .map_err(|e| tracing::error!("GSI worker {worker_id}: next_ready_wait failed: {e}"))
     .ok()
     .flatten();
     secs.map(|s| {

crates/storage-postgres/src/lib.rs

@@ -134,6 +134,12 @@ pub struct PostgresConfig {
 /// settings, accounts, IAM) and `data_pool` for per-DynamoDB-table data
 /// (`_ddb_*` tables, GSI tables). This separation allows the catalog and
 /// data to live in different `PostgreSQL` databases (Bug 1, P54).
+/// Default GSI propagation delay (milliseconds) when the
+/// `gsi_propagation_delay_ms` setting is absent. Mirrors the value seeded by
+/// the catalog schema, and is the single definition used by both the live read
+/// on the write path and the background refresh worker.
+pub(crate) const DEFAULT_GSI_PROPAGATION_DELAY_MS: u64 = 10;
+
 pub struct PostgresEngine {
     pub(crate) pool: PgPool,
     /// Connection pool for the data database where `_ddb_*` tables live.
@@ -145,8 +151,10 @@ pub struct PostgresEngine {
     pub(crate) control_plane_notify: Arc<tokio::sync::Notify>,
     /// D-4: Async GSI update queue. `None` until `start_gsi_workers()` is called.
     pub(crate) gsi_queue: Option<Arc<gsi_queue::GsiQueue>>,
-    /// P119: Cached GSI default propagation delay (milliseconds). Updated by
-    /// background poller every 30s. Avoids per-request DB query on write path.
+    /// P119: Cached GSI default propagation delay (milliseconds). Refreshed by
+    /// the background poller every 30s and re-warmed by `gsi_default_delay`.
+    /// This is only a fallback for when the live read fails; the write path
+    /// reads the setting live so a runtime change applies to the next write.
     pub gsi_default_delay_ms: Arc<std::sync::atomic::AtomicU64>,
 }
 
@@ -208,7 +216,7 @@ impl PostgresEngine {
         .ok()
         .flatten()
         .and_then(|(v,)| v.parse::<u64>().ok())
-        .unwrap_or(10);
+        .unwrap_or(DEFAULT_GSI_PROPAGATION_DELAY_MS);
 
         Ok(Self {
             pool,
@@ -226,6 +234,39 @@ impl PostgresEngine {
     /// Must be called after construction, before serving requests.
     /// Returns `&Self` for chaining.
     #[must_use]
+    /// Current GSI propagation delay (ms); `0` means synchronous.
+    ///
+    /// Reads the `gsi_propagation_delay_ms` setting live from the catalog so an
+    /// out-of-process change (`extenddb settings set`) applies to the next write
+    /// rather than up to 30 s later when the poll worker refreshes the cache.
+    /// Callers skip this entirely for tables with no secondary indexes, so a
+    /// table that cannot propagate pays nothing.
+    ///
+    /// On a read error the cached value is used and the error is logged, so a
+    /// degraded catalog serves a stale delay loudly rather than silently. On
+    /// success the cache is re-warmed, keeping the fallback fresh.
+    pub(crate) async fn gsi_default_delay(&self) -> u64 {
+        use std::sync::atomic::Ordering;
+        let live = sqlx::query_as::<_, (String,)>(
+            "SELECT value FROM settings WHERE key = 'gsi_propagation_delay_ms'",
+        )
+        .fetch_optional(&self.pool)
+        .await;
+        match live {
+            Ok(row) => {
+                let ms = row
+                    .and_then(|(v,)| v.parse::<u64>().ok())
+                    .unwrap_or(DEFAULT_GSI_PROPAGATION_DELAY_MS);
+                self.gsi_default_delay_ms.store(ms, Ordering::Relaxed);
+                ms
+            }
+            Err(e) => {
+                tracing::debug!("gsi_default_delay: live read failed, using cache: {e:?}");
+                self.gsi_default_delay_ms.load(Ordering::Relaxed)
+            }
+        }
+    }
+
     pub fn start_gsi_workers(mut self) -> Self {
         self.gsi_queue = Some(gsi_queue::GsiQueue::spawn(self.data_pool.clone()));
         self

crates/storage-postgres/src/workers.rs

@@ -183,7 +183,10 @@ pub(crate) async fn poll_gsi_delay<S: SettingsStore + ?Sized>(
             }
             Ok(None) => {
                 // Setting removed - revert to default
-                gsi_delay.store(10, std::sync::atomic::Ordering::Relaxed);
+                gsi_delay.store(
+                    crate::DEFAULT_GSI_PROPAGATION_DELAY_MS,
+                    std::sync::atomic::Ordering::Relaxed,
+                );
             }
             Err(e) => {
                 tracing::debug!("Failed to query gsi_propagation_delay_ms: {e:?}");

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

@@ -23,11 +23,15 @@ impl SqliteEngine {
         maps: &ExpressionMaps,
         stream: Option<&StreamCapture>,
     ) -> Result<Option<Item>, StorageError> {
+        // Read the propagation delay BEFORE taking the write lock. It is a
+        // runtime setting, not an invariant of this write, so it does not need
+        // to be read under the lock, and the lock serialises every write in the
+        // process: work done inside it is the backend's throughput bottleneck.
+        let system_delay = self.gsi_default_delay().await;
         let _writer = self.write_lock.lock().await;
         // Read the index set after acquiring the write lock so a concurrently
         // added GSI (UpdateTable holds the same lock) is not missed.
         let indexes = fetch_indexes_for_table(&key_info.table_id, &self.pool).await?;
-        let system_delay = self.gsi_default_delay();
 
         let mut tx = self
             .pool

crates/storage-sqlite/src/data/put_item.rs

@@ -33,6 +33,11 @@ impl SqliteEngine {
         // Read the index set only after acquiring the write lock, so a GSI added
         // by a concurrent UpdateTable (which holds the same lock) cannot be missed
         // and left unmaintained by this write.
+        // Read the propagation delay BEFORE taking the write lock. It is a
+        // runtime setting, not an invariant of this write, so it does not need
+        // to be read under the lock, and the lock serialises every write in the
+        // process: work done inside it is the backend's throughput bottleneck.
+        let system_delay = self.gsi_default_delay().await;
         let _writer = self.write_lock.lock().await;
         let indexes = fetch_indexes_for_table(&key_info.table_id, &self.pool).await?;
 
@@ -56,7 +61,6 @@ impl SqliteEngine {
             .map_err(|e| StorageError::Validation(e.to_string()))?;
         }
 
-        let system_delay = self.gsi_default_delay();
         let need_old = condition.is_some() || return_old || !indexes.is_empty() || stream.is_some();
 
         let mut tx = self

crates/storage-sqlite/src/data/transactions.rs

@@ -68,6 +68,11 @@ impl SqliteEngine {
         ops: &[TransactWriteOp<'_>],
         idempotency: Option<IdempotencyKey<'_>>,
     ) -> Result<(), StorageError> {
+        // Read the propagation delay BEFORE taking the write lock. It is a
+        // runtime setting, not an invariant of this write, so it does not need
+        // to be read under the lock, and the lock serialises every write in the
+        // process: work done inside it is the backend's throughput bottleneck.
+        let system_delay = self.gsi_default_delay().await;
         let _writer = self.write_lock.lock().await;
         // Fetch index metadata per distinct table AFTER acquiring the write lock,
         // so a GSI added by a concurrent UpdateTable (same lock) is not missed and
@@ -81,7 +86,6 @@ impl SqliteEngine {
             }
         }
 
-        let system_delay = self.gsi_default_delay();
         let mut tx = self
             .pool
             .begin_with("BEGIN IMMEDIATE")

crates/storage-sqlite/src/data/update_item.rs

@@ -33,11 +33,15 @@ impl SqliteEngine {
         maps: &ExpressionMaps,
         stream: Option<&StreamCapture>,
     ) -> Result<(Option<Item>, Option<Item>), StorageError> {
+        // Read the propagation delay BEFORE taking the write lock. It is a
+        // runtime setting, not an invariant of this write, so it does not need
+        // to be read under the lock, and the lock serialises every write in the
+        // process: work done inside it is the backend's throughput bottleneck.
+        let system_delay = self.gsi_default_delay().await;
         let _writer = self.write_lock.lock().await;
         // Read the index set after acquiring the write lock so a concurrently
         // added GSI (UpdateTable holds the same lock) is not missed.
         let indexes = fetch_indexes_for_table(&key_info.table_id, &self.pool).await?;
-        let system_delay = self.gsi_default_delay();
 
         let mut tx = self
             .pool

crates/storage-sqlite/src/lib.rs

@@ -45,6 +45,12 @@ mod update_table;
 mod worker;
 mod workers;
 
+/// Default GSI propagation delay (milliseconds) when the
+/// `gsi_propagation_delay_ms` setting is absent. Mirrors the value seeded by
+/// the catalog schema, and is the single definition used by both the live read
+/// on the write path and the background refresh worker.
+pub(crate) const DEFAULT_GSI_PROPAGATION_DELAY_MS: u64 = 10;
+
 pub use bootstrapper::SqliteBootstrapper;
 pub use catalog_store::SqliteCatalogStore;
 pub use config::SqliteConfig;

crates/storage-sqlite/src/store.rs

@@ -232,10 +232,35 @@ impl SqliteEngine {
         Ok(if from_env { None } else { Some(password) })
     }
 
-    /// Current cached GSI propagation delay (ms); `0` means synchronous.
-    pub(crate) fn gsi_default_delay(&self) -> u64 {
-        self.gsi_default_delay_ms
-            .load(std::sync::atomic::Ordering::Relaxed)
+    /// Current GSI propagation delay (ms); `0` means synchronous.
+    ///
+    /// Reads the `gsi_propagation_delay_ms` setting live from the catalog so
+    /// out-of-process changes (`extenddb settings set`) take effect on the
+    /// next write, not up to 30 s later when the poll worker refreshes the
+    /// cache. `SQLite` is a local file, so this is an indexed point lookup with
+    /// negligible cost next to the write it precedes. On a read error the
+    /// cached value (still refreshed by the poll worker) is the fallback; on
+    /// success the cache is re-warmed so fallback reads stay fresh.
+    pub(crate) async fn gsi_default_delay(&self) -> u64 {
+        use std::sync::atomic::Ordering;
+        let live: Result<Option<(String,)>, _> =
+            sqlx::query_as("SELECT value FROM settings WHERE key = 'gsi_propagation_delay_ms'")
+                .fetch_optional(&self.pool)
+                .await;
+        match live {
+            Ok(row) => {
+                // Missing row means the default, matching poll_gsi_delay.
+                let ms = row
+                    .and_then(|(v,)| v.parse::<u64>().ok())
+                    .unwrap_or(crate::DEFAULT_GSI_PROPAGATION_DELAY_MS);
+                self.gsi_default_delay_ms.store(ms, Ordering::Relaxed);
+                ms
+            }
+            Err(e) => {
+                tracing::debug!("gsi_default_delay: live read failed, using cache: {e:?}");
+                self.gsi_default_delay_ms.load(Ordering::Relaxed)
+            }
+        }
     }
 
     /// Handle to the GSI propagation notifier, woken after an enqueue.

crates/storage-sqlite/src/workers.rs

@@ -540,7 +540,9 @@ pub(crate) async fn poll_gsi_delay<S: SettingsStore + ?Sized>(
                     gsi_default.store(ms, Ordering::Relaxed);
                 }
             }
-            Ok(None) => gsi_default.store(10, Ordering::Relaxed),
+            Ok(None) => {
+                gsi_default.store(crate::DEFAULT_GSI_PROPAGATION_DELAY_MS, Ordering::Relaxed)
+            }
             Err(e) => tracing::debug!("poll_gsi_delay: {e:?}"),
         }
     }