fix(sqlite): serialize all writers through the engine write lock

.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:

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
         })

crates/storage-sqlite/src/create_table.rs

@@ -26,6 +26,12 @@ impl SqliteEngine {
         input: CreateTableInput,
     ) -> Result<TableDescription, StorageError> {
         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);

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

@@ -219,6 +219,8 @@ impl SqliteEngine {
         &self,
         max_age_seconds: i64,
     ) -> Result<u64, StorageError> {
+        // 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),
         );

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");
+        }
+    }
 }

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!(