fix(sqlite): serialize management store writes through the engine write lock

crates/storage-sqlite/src/admin_store.rs

@@ -14,6 +14,8 @@ impl AdminStore for SqliteCatalogStore {
         let admin_name = admin_name.to_owned();
         let password_hash = password_hash.to_owned();
         Box::pin(async move {
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             sqlx::query("INSERT INTO admin_users (admin_name, password_hash) VALUES (?, ?)")
                 .bind(&admin_name)
                 .bind(&password_hash)
@@ -57,6 +59,8 @@ impl AdminStore for SqliteCatalogStore {
     fn delete_admin(&self, admin_name: &str) -> BoxFuture<'_, OpResult<()>> {
         let admin_name = admin_name.to_owned();
         Box::pin(async move {
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             let result = sqlx::query("DELETE FROM admin_users WHERE admin_name = ?")
                 .bind(&admin_name)
                 .execute(self.pool())
@@ -80,6 +84,8 @@ impl AdminStore for SqliteCatalogStore {
         let admin_name = admin_name.to_owned();
         let password_hash = password_hash.to_owned();
         Box::pin(async move {
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             let result =
                 sqlx::query("UPDATE admin_users SET password_hash = ? WHERE admin_name = ?")
                     .bind(&password_hash)

crates/storage-sqlite/src/catalog_store.rs

@@ -27,22 +27,42 @@ use crate::sqlite_util::{format_timestamp, parse_timestamp};
 pub struct SqliteCatalogStore {
     pool: SqlitePool,
     encryption_key: Option<Arc<str>>,
+    /// Serializes this store's writers with the engine's (design decision D1,
+    /// see `store.rs`). For a file-backed database the catalog pool opens the
+    /// same file as the engine pool, so an uncoordinated catalog write can
+    /// hold the SQLite file write lock past a locked writer's `busy_timeout`
+    /// (and vice versa), failing an unrelated request with `database is
+    /// locked`. The server wiring in `lib.rs` shares the engine's lock here so
+    /// both sets of writers queue on one lock.
+    write_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 impl SqliteCatalogStore {
     /// Construct without a cached encryption key (settings/diagnostics-only use).
+    ///
+    /// Used by the out-of-process factories (the `settings`/`doctor` CLI
+    /// paths), where no engine exists in the process; the fresh lock still
+    /// serializes this store's own writers with each other.
     pub fn new(pool: SqlitePool) -> Self {
         Self {
             pool,
             encryption_key: None,
+            write_lock: Arc::new(tokio::sync::Mutex::new(())),
         }
     }
 
-    /// Construct with the cached AES-256-GCM encryption key (base64).
-    pub fn with_encryption_key(pool: SqlitePool, encryption_key: String) -> Self {
+    /// Construct with the cached AES-256-GCM encryption key (base64) and the
+    /// engine's write lock, so catalog writers queue with engine writers
+    /// (design decision D1).
+    pub fn with_encryption_key(
+        pool: SqlitePool,
+        encryption_key: String,
+        write_lock: Arc<tokio::sync::Mutex<()>>,
+    ) -> Self {
         Self {
             pool,
             encryption_key: Some(Arc::from(encryption_key.as_str())),
+            write_lock,
         }
     }
 
@@ -53,6 +73,16 @@ impl SqliteCatalogStore {
     pub(crate) fn encryption_key(&self) -> Option<&Arc<str>> {
         self.encryption_key.as_ref()
     }
+
+    /// Acquire the write lock (design decision D1: every writer holds it for
+    /// the duration of its statement or transaction, so no two writers ever
+    /// contend at the SQLite level). Reads never take it. Callers must acquire
+    /// it before opening a transaction or running a write statement, never
+    /// while already holding a pool connection another lock holder might wait
+    /// on.
+    pub(crate) async fn lock_writes(&self) -> tokio::sync::MutexGuard<'_, ()> {
+        self.write_lock.lock().await
+    }
 }
 
 // ── SettingsStore ──────────────────────────────────────────────────────
@@ -74,6 +104,8 @@ impl SettingsStore for SqliteCatalogStore {
         let key = key.to_owned();
         let value = value.to_owned();
         Box::pin(async move {
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             sqlx::query(
                 "INSERT INTO settings (key, value) VALUES (?, ?) \
                  ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -120,6 +152,9 @@ impl MetricsStore for SqliteCatalogStore {
     fn insert_metrics(&self, rows: &[MetricsRow]) -> BoxFuture<'_, OpResult<()>> {
         let rows = rows.to_vec();
         Box::pin(async move {
+            // D1: every writer holds the write lock. One acquisition for the
+            // whole batch; each row is its own autocommit statement.
+            let _writer = self.lock_writes().await;
             for row in &rows {
                 let bucket = format_timestamp(row.bucket);
                 let result = sqlx::query(
@@ -215,6 +250,8 @@ impl MetricsStore for SqliteCatalogStore {
             let cutoff = OffsetDateTime::now_utc()
                 - time::Duration::seconds(i64::try_from(retention.as_secs()).unwrap_or(i64::MAX));
             let cutoff_str = format_timestamp(cutoff);
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             sqlx::query("DELETE FROM metrics WHERE bucket < ?")
                 .bind(&cutoff_str)
                 .execute(&self.pool)
@@ -279,6 +316,8 @@ impl RateLimitStore for SqliteCatalogStore {
         let source_ip = source_ip.map(str::to_owned);
         Box::pin(async move {
             let now = format_timestamp(OffsetDateTime::now_utc());
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             let result = sqlx::query(
                 "INSERT INTO login_attempts (principal, attempted_at, success, source_ip) \
                  VALUES (?, ?, 0, ?)",
@@ -299,6 +338,8 @@ impl RateLimitStore for SqliteCatalogStore {
             let cutoff = format_timestamp(
                 OffsetDateTime::now_utc() - time::Duration::seconds(max_age_seconds),
             );
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             if let Err(e) = sqlx::query("DELETE FROM login_attempts WHERE attempted_at < ?")
                 .bind(&cutoff)
                 .execute(&self.pool)

crates/storage-sqlite/src/lib.rs

@@ -296,9 +296,19 @@ fn sqlite_server_components_factory(
                 })?;
         let enc_key = enc_key.ok_or(BackendError::MissingEncryptionKey)?;
 
-        let catalog_store: Arc<dyn extenddb_storage::CatalogStore> = Arc::new(
-            SqliteCatalogStore::with_encryption_key(catalog_pool.clone(), enc_key.clone()),
-        );
+        // The catalog store writes through the second pool over the same
+        // file, so its writers must hold the SAME lock instance as the
+        // engine's (design decision D1); an unshared lock would leave the two
+        // pools contending at the SQLite level, where a slow commit exhausts
+        // a concurrent writer's busy_timeout ("database is locked" -> 500).
+        // For an in-memory database the pools are one and the sharing is
+        // harmless. The credential store only reads and needs no lock.
+        let catalog_store: Arc<dyn extenddb_storage::CatalogStore> =
+            Arc::new(SqliteCatalogStore::with_encryption_key(
+                catalog_pool.clone(),
+                enc_key.clone(),
+                Arc::clone(&engine.write_lock),
+            ));
         let credential_store: Arc<dyn CredentialStore> =
             Arc::new(SqliteCredentialStore::new(catalog_pool, enc_key));
 

crates/storage-sqlite/src/management_store/access_keys.rs

@@ -46,6 +46,9 @@ impl SqliteCatalogStore {
         let encrypted = encrypt_secret(&secret_access_key, &enc_key, &access_key_id)
             .map_err(|e| OpError::Internal(format!("encrypt secret: {e}")))?;
 
+        // D1: every writer holds the write lock. Key generation, encryption,
+        // and the encryption-key read above all run before it.
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "INSERT INTO access_keys \
              (access_key_id, secret_key_encrypted, account_id, user_name) \
@@ -78,6 +81,8 @@ impl SqliteCatalogStore {
         user_name: &str,
         key_id: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query(
             "DELETE FROM access_keys \
              WHERE access_key_id = ? AND account_id = ? AND user_name = ?",
@@ -137,6 +142,9 @@ impl SqliteCatalogStore {
         let encrypted = encrypt_secret(secret_access_key, &enc_key, access_key_id)
             .map_err(|e| OpError::Internal(format!("encrypt secret: {e}")))?;
 
+        // D1: every writer holds the write lock. Encryption and the
+        // encryption-key read above run before it.
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "INSERT INTO access_keys \
              (access_key_id, secret_key_encrypted, account_id, user_name) \

crates/storage-sqlite/src/management_store/accounts.rs

@@ -15,6 +15,8 @@ impl SqliteCatalogStore {
         account_id: &str,
         account_name: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, ?)")
             .bind(account_id)
             .bind(account_name)
@@ -62,6 +64,9 @@ impl SqliteCatalogStore {
             ));
         }
 
+        // D1: every writer holds the write lock. The existence and dependency
+        // checks above are reads and run before it.
+        let _writer = self.lock_writes().await;
         sqlx::query("DELETE FROM accounts WHERE account_id = ?")
             .bind(account_id)
             .execute(self.pool())

crates/storage-sqlite/src/management_store/groups.rs

@@ -15,6 +15,8 @@ impl SqliteCatalogStore {
         group_name: &str,
     ) -> OpResult<()> {
         let group_arn = format!("arn:aws:iam::{account_id}:group/{group_name}");
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         sqlx::query("INSERT INTO iam_groups (account_id, group_name, group_arn) VALUES (?, ?, ?)")
             .bind(account_id)
             .bind(group_name)
@@ -39,6 +41,8 @@ impl SqliteCatalogStore {
         account_id: &str,
         group_name: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query("DELETE FROM iam_groups WHERE account_id = ? AND group_name = ?")
             .bind(account_id)
             .bind(group_name)
@@ -140,6 +144,8 @@ impl SqliteCatalogStore {
         group_name: &str,
         user_name: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "INSERT INTO iam_group_members (account_id, group_name, user_name) VALUES (?, ?, ?)",
         )
@@ -167,6 +173,8 @@ impl SqliteCatalogStore {
         group_name: &str,
         user_name: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query(
             "DELETE FROM iam_group_members \
              WHERE account_id = ? AND group_name = ? AND user_name = ?",

crates/storage-sqlite/src/management_store/mod.rs

@@ -554,6 +554,8 @@ impl ManagementStore for SqliteCatalogStore {
         let session_policy = session_policy.as_ref().map(ToString::to_string);
         let expires = format_timestamp(expires_at);
         Box::pin(async move {
+            // D1: every writer holds the write lock.
+            let _writer = self.lock_writes().await;
             sqlx::query(
                 "INSERT INTO iam_sessions \
                  (session_token, access_key_id, secret_key_encrypted, account_id, role_name, \

crates/storage-sqlite/src/management_store/policies.rs

@@ -20,6 +20,10 @@ impl SqliteCatalogStore {
     ) -> OpResult<()> {
         let doc = serde_json::to_string(document)
             .map_err(|e| OpError::Internal(format!("serialize policy: {e}")))?;
+        // D1: every writer holds the write lock. This is the writer observed
+        // losing its file lock in the run-integration-sqlite flake
+        // (`put_policy: ... database is locked`).
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "INSERT INTO iam_policies \
              (account_id, principal_type, principal_name, policy_name, policy_document) \
@@ -48,6 +52,8 @@ impl SqliteCatalogStore {
         principal_name: &str,
         policy_name: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query(
             "DELETE FROM iam_policies \
              WHERE account_id = ? AND principal_type = ? AND principal_name = ? AND policy_name = ?",
@@ -112,6 +118,8 @@ impl SqliteCatalogStore {
     ) -> OpResult<()> {
         let doc = serde_json::to_string(document)
             .map_err(|e| OpError::Internal(format!("serialize boundary: {e}")))?;
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "INSERT INTO iam_permissions_boundaries \
              (account_id, principal_type, principal_name, policy_document) \
@@ -163,6 +171,8 @@ impl SqliteCatalogStore {
         principal_type: &str,
         principal_name: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "DELETE FROM iam_permissions_boundaries \
              WHERE account_id = ? AND principal_type = ? AND principal_name = ?",

crates/storage-sqlite/src/management_store/roles.rs

@@ -26,6 +26,8 @@ impl SqliteCatalogStore {
         let role_arn = format!("arn:aws:iam::{account_id}:role/{role_name}");
         let trust = serde_json::to_string(trust_policy)
             .map_err(|e| OpError::Internal(format!("serialize trust policy: {e}")))?;
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         sqlx::query(
             "INSERT INTO iam_roles (account_id, role_name, role_arn, trust_policy) \
              VALUES (?, ?, ?, ?)",
@@ -50,6 +52,8 @@ impl SqliteCatalogStore {
     }
 
     pub(crate) async fn delete_role_impl(&self, account_id: &str, role_name: &str) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query("DELETE FROM iam_roles WHERE account_id = ? AND role_name = ?")
             .bind(account_id)
             .bind(role_name)
@@ -160,6 +164,9 @@ impl SqliteCatalogStore {
         role_name: &str,
         tags: &[(String, String)],
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock, acquired before the
+        // transaction begins.
+        let _writer = self.lock_writes().await;
         let mut tx = self
             .pool()
             .begin_with("BEGIN IMMEDIATE")
@@ -198,6 +205,9 @@ impl SqliteCatalogStore {
         role_name: &str,
         tag_keys: &[String],
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock, acquired before the
+        // transaction begins.
+        let _writer = self.lock_writes().await;
         let mut tx = self
             .pool()
             .begin_with("BEGIN IMMEDIATE")

crates/storage-sqlite/src/management_store/users.rs

@@ -18,6 +18,9 @@ impl SqliteCatalogStore {
     ) -> OpResult<()> {
         let user_arn = format!("arn:aws:iam::{account_id}:user/{user_name}");
 
+        // D1: every writer holds the write lock, acquired before the
+        // transaction begins so the lock never waits on a pool connection.
+        let _writer = self.lock_writes().await;
         let mut tx = self
             .pool()
             .begin_with("BEGIN IMMEDIATE")
@@ -89,6 +92,8 @@ impl SqliteCatalogStore {
     }
 
     pub(crate) async fn delete_user_impl(&self, account_id: &str, user_name: &str) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query("DELETE FROM iam_users WHERE account_id = ? AND user_name = ?")
             .bind(account_id)
             .bind(user_name)
@@ -232,6 +237,8 @@ impl SqliteCatalogStore {
         user_name: &str,
         password_hash: &str,
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock.
+        let _writer = self.lock_writes().await;
         let result = sqlx::query(
             "UPDATE iam_users SET password_hash = ? WHERE account_id = ? AND user_name = ?",
         )
@@ -256,6 +263,9 @@ impl SqliteCatalogStore {
         user_name: &str,
         tags: &[(String, String)],
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock, acquired before the
+        // transaction begins.
+        let _writer = self.lock_writes().await;
         let mut tx = self
             .pool()
             .begin_with("BEGIN IMMEDIATE")
@@ -294,6 +304,9 @@ impl SqliteCatalogStore {
         user_name: &str,
         tag_keys: &[String],
     ) -> OpResult<()> {
+        // D1: every writer holds the write lock, acquired before the
+        // transaction begins.
+        let _writer = self.lock_writes().await;
         let mut tx = self
             .pool()
             .begin_with("BEGIN IMMEDIATE")