From dbc1a896d16736c26e60b7b7ea12e6c152af0e7a Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Thu, 27 Aug 2026 15:52:20 +0000 Subject: [PATCH 1/4] fix(sqlite): serialize every writer through the engine write lock Design decision D1 says the engine serializes all writers through write_lock, so SQLITE_BUSY cannot arise from competing writers. Five writers ran outside the lock: CreateTable's two transactions, the TTL metadata and index DDL, tagging, the table-size refresh, and the stream-record and idempotency-token cleanup sweeps. Any of them can hold the SQLite write lock past a locked writer's 5s busy_timeout when a commit is slow, and the locked writer then fails an unrelated request with 'database is locked', which the engine maps to a 500. Measured on 2026-08-27 against a local server: an uncoordinated writer holding the file lock fails a plain PutItem with the exact InternalServerError seen in the run-integration-sqlite flake on main (commit e41f370), including the server-side 'database is locked' line. Also add a syslog dump step to the integration jobs: devtools/run-tests restarts the server daemonized, so its logs go to syslog and a failing job otherwise records no server-side evidence. --- .github/workflows/integration.yml | 32 +++++++++++++++++++ crates/storage-sqlite/src/create_table.rs | 6 ++++ .../storage-sqlite/src/data/transactions.rs | 2 ++ crates/storage-sqlite/src/metadata.rs | 15 +++++++++ crates/storage-sqlite/src/store.rs | 11 +++++++ crates/storage-sqlite/src/stream.rs | 2 ++ 6 files changed, 68 insertions(+) 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/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/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..2c91afa2 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/crates/storage-sqlite/src/store.rs @@ -17,6 +17,17 @@ //! `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. use std::sync::Arc; use std::sync::atomic::AtomicU64; 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), ); From 9e61e574ae63695185af98af5333f32464bb6a43 Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Thu, 27 Aug 2026 15:52:58 +0000 Subject: [PATCH 2/4] test(sqlite): pin the D1 invariant for the writers brought under the lock Each test holds the engine write lock, spawns one of the previously unserialized writers, asserts it makes no progress while the lock is held, and asserts it completes after release. Failing-first verified: with the lock removed from create_table_impl and the idempotency cleanup, the matching tests fail with 'completed while the engine write lock was held'. --- crates/storage-sqlite/src/store.rs | 133 +++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) diff --git a/crates/storage-sqlite/src/store.rs b/crates/storage-sqlite/src/store.rs index 2c91afa2..36ffa65e 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/crates/storage-sqlite/src/store.rs @@ -422,3 +422,136 @@ 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 { + // 2 connections: one for the spawned writer under test, one spare, so a + // writer that (incorrectly) runs while the lock is held is stopped by + // the missing lock alone, never by pool exhaustion. + 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"); + 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 { + let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({ + "TableName": "d1-lock-t", + "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], + "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], + "BillingMode": "PAY_PER_REQUEST" + })) + .expect("input"); + e.create_table_impl("000000000000", input) + .await + .expect("create table"); + }, + "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; + } +} From 5299dbd2277062e718f810a13f8b62362fc58aa5 Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Thu, 27 Aug 2026 16:45:03 +0000 Subject: [PATCH 3/4] fix(sqlite): bring the backup writers under the write lock Review round 1 findings on the D1 serialization fix: - delete_backup ran two unlocked writes on the engine pool, and the backup_items delete is a bulk statement (one row per backed-up item), which is exactly the slow-commit shape the parent commit eliminates. Wire-reachable through DeleteBackup and automatically after every point-in-time restore. - update_continuous_backups ran an unlocked upsert. Same class, one small row; the lock is scoped so the trailing read-only describe runs after release. - The store.rs module doc now names the deliberate exclusions (init-time bootstrap, the management stores on the separate catalog pool) and the residual same-file contention risk that pool carries. - The test helper comment claimed a 2-connection pool; in-memory engines are pinned to a single connection, so the comment now describes the actual mechanism. Two new D1 tests cover the backup writers. Failing-first verified: with the locks removed, both fail with 'completed while the engine write lock was held'. --- crates/storage-sqlite/src/backup.rs | 33 ++++++---- crates/storage-sqlite/src/store.rs | 95 +++++++++++++++++++++++++---- 2 files changed, 104 insertions(+), 24 deletions(-) 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/store.rs b/crates/storage-sqlite/src/store.rs index 36ffa65e..d5e912f8 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/crates/storage-sqlite/src/store.rs @@ -28,6 +28,13 @@ //! 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; @@ -430,9 +437,11 @@ mod d1_write_lock_tests { use std::time::Duration; async fn engine() -> SqliteEngine { - // 2 connections: one for the spawned writer under test, one spare, so a - // writer that (incorrectly) runs while the lock is held is stopped by - // the missing lock alone, never by pool exhaustion. + // 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"); @@ -443,6 +452,14 @@ mod d1_write_lock_tests { .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 } @@ -482,16 +499,7 @@ mod d1_write_lock_tests { assert_serialized( &engine, async move { - let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({ - "TableName": "d1-lock-t", - "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}], - "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}], - "BillingMode": "PAY_PER_REQUEST" - })) - .expect("input"); - e.create_table_impl("000000000000", input) - .await - .expect("create table"); + create_table(&e, "d1-lock-t").await; }, "create_table_impl", ) @@ -554,4 +562,65 @@ mod d1_write_lock_tests { ) .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; + } } From e7846e2bf05ce4c89d29db29b2e5b6f2b784ee85 Mon Sep 17 00:00:00 2001 From: Anandh Somasundaram Date: Fri, 28 Aug 2026 16:44:46 +0000 Subject: [PATCH 4/4] fix(sqlite): bring the vector build's catalog flips under the write lock PR review finding: set_backfilling and mark_active ran UPDATE vector_indexes on the engine pool with no write lock, while their siblings backfill_batch and reset_data_table on the same struct take it. A slow locked commit could exhaust their busy_timeout and fail a build step. mark_active takes the lock for the whole method, so its empty-flip branch no longer re-acquires (that would deadlock). One new test pins both flips, failing-first verified for each independently. --- .../storage-sqlite/src/data/vector_index.rs | 55 ++++++++++++++++++- 1 file changed, 54 insertions(+), 1 deletion(-) 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"); + } + } }