diff --git a/crates/storage-postgres/src/data/delete_item.rs b/crates/storage-postgres/src/data/delete_item.rs index ab477e7f..6765bae1 100755 --- a/crates/storage-postgres/src/data/delete_item.rs +++ b/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(); diff --git a/crates/storage-postgres/src/data/put_item.rs b/crates/storage-postgres/src/data/put_item.rs index e68dbe97..b8b0920a 100755 --- a/crates/storage-postgres/src/data/put_item.rs +++ b/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 diff --git a/crates/storage-postgres/src/data/transactions.rs b/crates/storage-postgres/src/data/transactions.rs index 23482b95..5e098046 100644 --- a/crates/storage-postgres/src/data/transactions.rs +++ b/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 diff --git a/crates/storage-postgres/src/data/update_item.rs b/crates/storage-postgres/src/data/update_item.rs index fceb7463..a2c668af 100755 --- a/crates/storage-postgres/src/data/update_item.rs +++ b/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 diff --git a/crates/storage-postgres/src/gsi_queue.rs b/crates/storage-postgres/src/gsi_queue.rs index 30ca096a..3a701f4b 100644 --- a/crates/storage-postgres/src/gsi_queue.rs +++ b/crates/storage-postgres/src/gsi_queue.rs @@ -273,13 +273,22 @@ async fn worker(worker_id: u64, q: Arc) { /// 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 { + // 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 = 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| { diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index ac77eb58..dcdb4507 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/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, /// D-4: Async GSI update queue. `None` until `start_gsi_workers()` is called. pub(crate) gsi_queue: Option>, - /// 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, } @@ -208,7 +216,7 @@ impl PostgresEngine { .ok() .flatten() .and_then(|(v,)| v.parse::().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::().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 diff --git a/crates/storage-postgres/src/workers.rs b/crates/storage-postgres/src/workers.rs index e7388f88..7e1d144a 100644 --- a/crates/storage-postgres/src/workers.rs +++ b/crates/storage-postgres/src/workers.rs @@ -183,7 +183,10 @@ pub(crate) async fn poll_gsi_delay( } 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:?}"); diff --git a/crates/storage-sqlite/src/data/delete_item.rs b/crates/storage-sqlite/src/data/delete_item.rs index 81e5ac4e..a807045e 100644 --- a/crates/storage-sqlite/src/data/delete_item.rs +++ b/crates/storage-sqlite/src/data/delete_item.rs @@ -23,11 +23,15 @@ impl SqliteEngine { maps: &ExpressionMaps, stream: Option<&StreamCapture>, ) -> 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; // 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 diff --git a/crates/storage-sqlite/src/data/put_item.rs b/crates/storage-sqlite/src/data/put_item.rs index 0eac5f1c..de43d830 100644 --- a/crates/storage-sqlite/src/data/put_item.rs +++ b/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 diff --git a/crates/storage-sqlite/src/data/transactions.rs b/crates/storage-sqlite/src/data/transactions.rs index c1f0525b..f3ae8dec 100644 --- a/crates/storage-sqlite/src/data/transactions.rs +++ b/crates/storage-sqlite/src/data/transactions.rs @@ -68,6 +68,11 @@ impl SqliteEngine { ops: &[TransactWriteOp<'_>], idempotency: Option>, ) -> 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") diff --git a/crates/storage-sqlite/src/data/update_item.rs b/crates/storage-sqlite/src/data/update_item.rs index 4d53add2..92ac84a5 100644 --- a/crates/storage-sqlite/src/data/update_item.rs +++ b/crates/storage-sqlite/src/data/update_item.rs @@ -33,11 +33,15 @@ impl SqliteEngine { maps: &ExpressionMaps, stream: Option<&StreamCapture>, ) -> Result<(Option, Option), 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 diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 87696472..9f7b3c68 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/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; diff --git a/crates/storage-sqlite/src/store.rs b/crates/storage-sqlite/src/store.rs index 4a0df770..39f29370 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/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, _> = + 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::().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. diff --git a/crates/storage-sqlite/src/workers.rs b/crates/storage-sqlite/src/workers.rs index 6438a4a2..82776713 100644 --- a/crates/storage-sqlite/src/workers.rs +++ b/crates/storage-sqlite/src/workers.rs @@ -540,7 +540,9 @@ pub(crate) async fn poll_gsi_delay( 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:?}"), } }