diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 443c7d6f..da5d274a 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -79,6 +79,14 @@ jobs: EXTENDDB_TEST_PG_ADMIN_CONNECTION_STRING: postgresql://postgres:devpass@127.0.0.1:5432 run: devtools/run-tests --extenddb --pytest --comprehensive --parallel --filter "not import_export" + # run-tests restarts the server daemonized to apply the import/export + # config, and a daemonized server logs to syslog, not the job log. Without + # this dump, the storage error behind a client-visible 500 leaves no + # evidence anywhere in CI. + - name: Dump server syslog on failure + if: failure() + run: sudo journalctl -t extenddb --no-pager -n 500 || true + run-integration-sqlite: runs-on: ubuntu-latest steps: @@ -130,6 +138,12 @@ jobs: EXTENDDB_ADMIN_PASSWORD: ${{ steps.init.outputs.admin_password }} run: devtools/run-tests --extenddb --pytest --comprehensive --parallel --filter "not import_export" + # Same rationale as the postgres job: the daemonized server logs to + # syslog, so this dump is the only server-side evidence on failure. + - name: Dump server syslog on failure + if: failure() + run: sudo journalctl -t extenddb --no-pager -n 500 || true + run-integration-dev-mode: # dev-mode was shipped with no CI coverage at all, which is how the batch and # transaction authorization regression reached main: the build compiled, so @@ -269,6 +283,12 @@ jobs: 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 + # The daemonized server logs to syslog; dump it so server-side failures + # are diagnosable from the job log. + - name: Dump server syslog on failure + if: failure() + run: sudo journalctl -t extenddb --no-pager -n 500 || true + # The runtime-detection proof: the same binary, against a PostgreSQL with no # pgvector, must refuse vector indexes over the wire. Today the job above runs # on a plain image too, so this looks like a duplicate of it; it is not, because @@ -337,6 +357,12 @@ jobs: devtools/run-tests --extenddb --rust-integration --release --filter vector_index_unsupported + # The daemonized server logs to syslog; dump it so server-side failures + # are diagnosable from the job log. + - name: Dump server syslog on failure + if: failure() + run: sudo journalctl -t extenddb --no-pager -n 500 || true + # No storage-level step here on purpose. Those tests build their own # extension-free scratch databases, so they behave identically on either # image and the job above already runs them; repeating them here would buy @@ -401,6 +427,12 @@ jobs: EXTENDDB_EXPECT_VECTORS: "1" run: devtools/run-tests --extenddb --rust-integration --release + # The daemonized server logs to syslog; dump it so server-side failures + # are diagnosable from the job log. + - name: Dump server syslog on failure + if: failure() + run: sudo journalctl -t extenddb --no-pager -n 500 || true + integration: runs-on: ubuntu-latest needs: diff --git a/crates/storage-sqlite/src/backup.rs b/crates/storage-sqlite/src/backup.rs index 48b1dc4f..6e076f76 100644 --- a/crates/storage-sqlite/src/backup.rs +++ b/crates/storage-sqlite/src/backup.rs @@ -270,6 +270,12 @@ impl BackupEngine for SqliteEngine { // reported missing here and the writes below never run. let desc = self.describe_backup(&account_id, &backup_arn).await?; + // D1: every writer holds the engine write lock. The item delete + // below is a bulk statement (one row per backed-up item), which is + // exactly the slow-commit shape that can hold the SQLite file lock + // past a concurrent locked writer's busy_timeout. + let _writer = self.write_lock.lock().await; + // The account predicate is repeated on both writes rather than // relying on the lookup above, so the statements are correct on // their own terms. @@ -488,17 +494,22 @@ impl BackupEngine for SqliteEngine { if !exists { return Err(StorageError::TableNotFound(table_name)); } - sqlx::query( - "INSERT INTO continuous_backups (account_id, table_name, pitr_enabled) \ - VALUES (?, ?, ?) \ - ON CONFLICT (account_id, table_name) DO UPDATE SET pitr_enabled = excluded.pitr_enabled", - ) - .bind(&account_id) - .bind(&table_name) - .bind(pitr_enabled) - .execute(&self.pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + { + // D1: every writer holds the engine write lock. Scoped so the + // read-only describe below runs after release. + let _writer = self.write_lock.lock().await; + sqlx::query( + "INSERT INTO continuous_backups (account_id, table_name, pitr_enabled) \ + VALUES (?, ?, ?) \ + ON CONFLICT (account_id, table_name) DO UPDATE SET pitr_enabled = excluded.pitr_enabled", + ) + .bind(&account_id) + .bind(&table_name) + .bind(pitr_enabled) + .execute(&self.pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + } self.describe_continuous_backups(&account_id, &table_name) .await }) diff --git a/crates/storage-sqlite/src/create_table.rs b/crates/storage-sqlite/src/create_table.rs index 754cac0c..88935853 100644 --- a/crates/storage-sqlite/src/create_table.rs +++ b/crates/storage-sqlite/src/create_table.rs @@ -26,6 +26,12 @@ impl SqliteEngine { input: CreateTableInput, ) -> Result { Self::validate_account_id(account_id)?; + // D1: every writer holds the engine write lock. This method runs two + // write transactions (catalog rows, then data-table DDL); without the + // lock they contend with data-plane writers at the SQLite level, and a + // slow DDL commit can exhaust a concurrent writer's busy_timeout, + // surfacing as a 500 on an unrelated request. + let _writer = self.write_lock.lock().await; let table_id = uuid::Uuid::new_v4().to_string(); let table_arn = table_arn(&self.region, account_id, &input.table_name); let billing_mode = input.billing_mode.unwrap_or(BillingMode::Provisioned); diff --git a/crates/storage-sqlite/src/data/transactions.rs b/crates/storage-sqlite/src/data/transactions.rs index a60dba92..93e278f1 100644 --- a/crates/storage-sqlite/src/data/transactions.rs +++ b/crates/storage-sqlite/src/data/transactions.rs @@ -219,6 +219,8 @@ impl SqliteEngine { &self, max_age_seconds: i64, ) -> Result { + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; let cutoff = format_timestamp( time::OffsetDateTime::now_utc() - time::Duration::seconds(max_age_seconds), ); diff --git a/crates/storage-sqlite/src/data/vector_index.rs b/crates/storage-sqlite/src/data/vector_index.rs index 54c45d18..1f578edb 100644 --- a/crates/storage-sqlite/src/data/vector_index.rs +++ b/crates/storage-sqlite/src/data/vector_index.rs @@ -591,6 +591,9 @@ impl VectorIndexBuild for SqliteVectorBuild { } async fn set_backfilling(&mut self) -> Result<(), StorageError> { + // D1: every writer holds the engine write lock, the build's catalog + // flips included. + let _writer = self.write_lock.lock().await; sqlx::query( "UPDATE vector_indexes SET backfilling = 1 WHERE table_id = ? AND index_id = ?", ) @@ -603,6 +606,9 @@ impl VectorIndexBuild for SqliteVectorBuild { } async fn mark_active(&mut self, skipped: usize) -> Result<(), StorageError> { + // D1: every writer holds the engine write lock. Taken here for the + // whole method, so the empty-flip branch below must not re-acquire. + let _writer = self.write_lock.lock().await; // `backfilling` is cleared to NULL rather than set to 0, because the // service removes the member once ACTIVE and the catalog CHECK // constraint enforces that pairing. @@ -624,7 +630,6 @@ impl VectorIndexBuild for SqliteVectorBuild { // definition back. A delete landing in between drops the table it // could see, and this build then recreates one nothing references. // The absent catalog row is the proof the index is gone. - let _writer = self.write_lock.lock().await; crate::SqliteEngine::drop_vector_data_table_by_id( &self.pool, &self.table_id, @@ -1042,4 +1047,52 @@ mod tests { "one base key must yield one index row, not a duplicate" ); } + /// The build's catalog flips are writers like any other: while the engine + /// write lock is held, neither may complete; after release, both must. + /// Mirrors the D1 tests in `store.rs` for the `SqliteEngine` writers. + /// + /// No catalog row is seeded on purpose: `set_backfilling` updates zero rows + /// and `mark_active` takes its empty-flip branch (which also proves that + /// branch cannot deadlock now that the lock spans the whole method). + #[tokio::test] + async fn build_catalog_flips_wait_for_the_write_lock() { + let engine = crate::SqliteEngine::new(":memory:", 1, "us-east-1", 409_600) + .await + .expect("engine"); + crate::schema::apply(&engine.pool).await.expect("schema"); + + let build = || SqliteVectorBuild { + pool: engine.pool.clone(), + write_lock: std::sync::Arc::clone(&engine.write_lock), + gsi_notify: engine.gsi_notify(), + table_id: "t-d1".to_owned(), + index_id: "vidx-d1".to_owned(), + base_key_schema: Vec::new(), + attribute_definitions: Vec::new(), + meta: None, + }; + + for (name, mut ops, which) in [ + ("set_backfilling", build(), 0_u8), + ("mark_active", build(), 1_u8), + ] { + let guard = engine.write_lock.lock().await; + let task = tokio::spawn(async move { + match which { + 0 => ops.set_backfilling().await.expect("set_backfilling"), + _ => ops.mark_active(0).await.expect("mark_active"), + } + }); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + !task.is_finished(), + "{name} completed while the engine write lock was held (D1 violation)" + ); + drop(guard); + tokio::time::timeout(std::time::Duration::from_secs(10), task) + .await + .unwrap_or_else(|_| panic!("{name} did not complete after lock release")) + .expect("writer task panicked"); + } + } } diff --git a/crates/storage-sqlite/src/metadata.rs b/crates/storage-sqlite/src/metadata.rs index 44837105..a692dd52 100644 --- a/crates/storage-sqlite/src/metadata.rs +++ b/crates/storage-sqlite/src/metadata.rs @@ -90,6 +90,8 @@ impl MetadataEngine for SqliteEngine { let table_name = table_name.to_owned(); let attribute_name = attribute_name.to_owned(); Box::pin(async move { + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; let ttl_val: Option<&str> = if enabled { Some(&attribute_name) } else { None }; let result = sqlx::query( "UPDATE tables SET ttl_attribute = ?, ttl_index_ready = 0 \ @@ -123,6 +125,8 @@ impl MetadataEngine for SqliteEngine { let arn = arn.to_owned(); let tags = tags.to_vec(); Box::pin(async move { + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; for tag in &tags { sqlx::query( "INSERT INTO tags (resource_arn, tag_key, tag_value) VALUES (?, ?, ?) \ @@ -147,6 +151,8 @@ impl MetadataEngine for SqliteEngine { let arn = arn.to_owned(); let tag_keys = tag_keys.to_vec(); Box::pin(async move { + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; for key in &tag_keys { sqlx::query("DELETE FROM tags WHERE resource_arn = ? AND tag_key = ?") .bind(&arn) @@ -232,6 +238,11 @@ impl MetadataEngine for SqliteEngine { let ttl_attribute = ttl_attribute.to_owned(); Box::pin(async move { Self::validate_account_id(&account_id)?; + // D1: every writer holds the engine write lock. CREATE INDEX over a + // populated data table is the slowest single statement this backend + // issues; unserialized it can hold the SQLite write lock past a + // concurrent writer's busy_timeout. + let _writer = self.write_lock.lock().await; let table_id = self.table_id_for(&account_id, &table_name).await?; let data_table = data::data_table_name(&table_id); @@ -270,6 +281,8 @@ impl MetadataEngine for SqliteEngine { let table_name = table_name.to_owned(); Box::pin(async move { Self::validate_account_id(&account_id)?; + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; let table_id = self.table_id_for(&account_id, &table_name).await?; sqlx::query( "UPDATE tables SET ttl_index_ready = 0 WHERE account_id = ? AND table_name = ?", @@ -345,6 +358,8 @@ impl MetadataEngine for SqliteEngine { let table_name = table_name.to_owned(); Box::pin(async move { Self::validate_account_id(&account_id)?; + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; let table_id = self.table_id_for(&account_id, &table_name).await?; let data_table = data::data_table_name(&table_id); let (item_count, table_size): (i64, i64) = sqlx::query_as(&format!( diff --git a/crates/storage-sqlite/src/store.rs b/crates/storage-sqlite/src/store.rs index e1a7dbbb..d5e912f8 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/crates/storage-sqlite/src/store.rs @@ -17,6 +17,24 @@ //! `SQLITE_BUSY_SNAPSHOT` (a deferred read-then-write whose snapshot is //! invalidated by another pool committing) rather than surfacing it as a 500. //! Reads run concurrently from the pool against WAL snapshots and take no lock. +//! +//! "All writers" includes the control-plane paths (CreateTable's DDL, TTL +//! metadata, tagging) and the periodic maintenance workers (table-size +//! refresh, TTL index creation, stream-record and idempotency-token cleanup), +//! not just the item write paths. A writer outside the lock contends at the +//! SQLite level instead, and when its commit is slow (a large `CREATE INDEX`, +//! a stalled fsync on a loaded CI host) a concurrent locked writer exhausts +//! `busy_timeout` and fails an unrelated request with `database is locked`, +//! which the engine maps to a 500. Measured 2026-08-27: an uncoordinated +//! writer holding the file lock fails a plain `PutItem` with exactly the +//! `InternalServerError` seen in the `run-integration-sqlite` CI flake. +//! +//! Deliberate exclusions from the lock, so the invariant stays auditable: +//! init-time bootstrap in this file (runs before the server serves traffic), +//! and the management/credential stores, which write through the separate +//! catalog pool in `lib.rs`. For a file-backed database that second pool +//! opens the same file, so its small single-row autocommit writes carry a +//! residual, much smaller, version of the same contention risk. use std::sync::Arc; use std::sync::atomic::AtomicU64; @@ -411,3 +429,198 @@ impl SqliteEngine { .map_or_else(|| "(not configured)".to_owned(), |(v,)| v) } } + +#[cfg(test)] +mod d1_write_lock_tests { + use super::SqliteEngine; + use serde_json::json; + use std::time::Duration; + + async fn engine() -> SqliteEngine { + // The pool size is nominal: `SqliteEngine::new` pins in-memory + // databases to a single connection regardless. The tests below never + // hold a pool connection on the asserting side, so a writer that + // (incorrectly) ignores the lock is stopped by nothing at all, which + // is what the 200ms grace window detects. + let engine = SqliteEngine::new(":memory:", 2, "us-east-1", 409_600) + .await + .expect("engine"); + crate::schema::apply(&engine.pool).await.expect("schema"); + sqlx::query( + "INSERT INTO accounts (account_id, account_name) VALUES ('000000000000', 'default')", + ) + .execute(&engine.pool) + .await + .expect("account"); + // Zero control-plane delay: tables become ACTIVE at create time, since + // no transition poller runs inside a unit test. + sqlx::query( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0')", + ) + .execute(&engine.pool) + .await + .expect("settings"); + engine + } + + /// Assert the D1 invariant for one writer: while the engine write lock is + /// held, the writer must not complete; after release, it must. + /// + /// This is the discriminating shape for the 2026-08-27 `run-integration-sqlite` + /// flake (`PutItem` returning `InternalServerError`, server-side `database is + /// locked`): a writer outside the lock contends at the SQLite level, where a + /// slow commit exhausts a concurrent writer's 5s `busy_timeout`. Before the + /// fix, each writer below completed while the lock was held; with it, they + /// queue behind the lock and cannot collide. + async fn assert_serialized(engine: &SqliteEngine, writer: F, name: &str) + where + F: std::future::Future + Send + 'static, + { + let guard = engine.write_lock.lock().await; + let task = tokio::spawn(writer); + // Generous grace period: a writer that ignores the lock finishes these + // single-statement transactions in well under 200ms. + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + !task.is_finished(), + "{name} completed while the engine write lock was held (D1 violation)" + ); + drop(guard); + tokio::time::timeout(Duration::from_secs(10), task) + .await + .unwrap_or_else(|_| panic!("{name} did not complete after lock release")) + .expect("writer task panicked"); + } + + #[tokio::test] + async fn create_table_waits_for_the_write_lock() { + let engine = engine().await; + let e = engine.clone(); + assert_serialized( + &engine, + async move { + create_table(&e, "d1-lock-t").await; + }, + "create_table_impl", + ) + .await; + } + + #[tokio::test] + async fn idempotency_token_cleanup_waits_for_the_write_lock() { + let engine = engine().await; + let e = engine.clone(); + assert_serialized( + &engine, + async move { + e.cleanup_expired_idempotency_tokens_impl(0) + .await + .expect("cleanup"); + }, + "cleanup_expired_idempotency_tokens", + ) + .await; + } + + #[tokio::test] + async fn tag_resource_waits_for_the_write_lock() { + use extenddb_storage::MetadataEngine; + let engine = engine().await; + let e = engine.clone(); + assert_serialized( + &engine, + async move { + MetadataEngine::tag_resource( + &e, + "arn:aws:dynamodb:us-east-1:000000000000:table/d1", + &[extenddb_core::types::Tag { + key: "k".to_owned(), + value: "v".to_owned(), + }], + ) + .await + .expect("tag"); + }, + "tag_resource", + ) + .await; + } + + #[tokio::test] + async fn stream_record_cleanup_waits_for_the_write_lock() { + use extenddb_storage::StreamEngine; + let engine = engine().await; + let e = engine.clone(); + assert_serialized( + &engine, + async move { + StreamEngine::cleanup_expired_stream_records(&e, 0) + .await + .expect("cleanup"); + }, + "cleanup_expired_stream_records", + ) + .await; + } + + /// Create a plain table outside the lock window, for the writers whose + /// pre-lock reads refuse to proceed without one. + async fn create_table(engine: &SqliteEngine, name: &str) { + let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({ + "TableName": name, + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input"); + engine + .create_table_impl("000000000000", input) + .await + .expect("create table"); + } + + #[tokio::test] + async fn delete_backup_waits_for_the_write_lock() { + use extenddb_storage::BackupEngine; + let engine = engine().await; + create_table(&engine, "d1-bkp-t").await; + // The backup must exist before the lock window: delete_backup resolves + // it with a read first and returns early when it is missing, which + // would complete without ever reaching the writes under test. + let details = BackupEngine::create_backup(&engine, "000000000000", "d1-bkp-t", "b") + .await + .expect("backup"); + let e = engine.clone(); + assert_serialized( + &engine, + async move { + BackupEngine::delete_backup(&e, "000000000000", &details.backup_arn) + .await + .expect("delete backup"); + }, + "delete_backup", + ) + .await; + } + + #[tokio::test] + async fn update_continuous_backups_waits_for_the_write_lock() { + use extenddb_storage::BackupEngine; + let engine = engine().await; + // The table must exist before the lock window: the pre-lock existence + // check returns TableNotFound otherwise, completing without reaching + // the write under test. + create_table(&engine, "d1-pitr-t").await; + let e = engine.clone(); + assert_serialized( + &engine, + async move { + BackupEngine::update_continuous_backups(&e, "000000000000", "d1-pitr-t", true) + .await + .expect("update continuous backups"); + }, + "update_continuous_backups", + ) + .await; + } +} diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs index 52ba76c8..3d8850e8 100644 --- a/crates/storage-sqlite/src/stream.rs +++ b/crates/storage-sqlite/src/stream.rs @@ -361,6 +361,8 @@ impl StreamEngine for SqliteEngine { retention_hours: i64, ) -> BoxFuture<'_, Result> { Box::pin(async move { + // D1: every writer holds the engine write lock. + let _writer = self.write_lock.lock().await; let cutoff = format_timestamp( time::OffsetDateTime::now_utc() - time::Duration::hours(retention_hours), );