diff --git a/crates/storage-sqlite/src/admin_store.rs b/crates/storage-sqlite/src/admin_store.rs index cbd2c5607..6780f317b 100644 --- a/crates/storage-sqlite/src/admin_store.rs +++ b/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) diff --git a/crates/storage-sqlite/src/catalog_store.rs b/crates/storage-sqlite/src/catalog_store.rs index 0bdfc2fc7..410aa46d6 100644 --- a/crates/storage-sqlite/src/catalog_store.rs +++ b/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>, + /// 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>, } 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>, + ) -> 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> { 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) diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 87427b883..b68a7c62f 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/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 = 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 = + Arc::new(SqliteCatalogStore::with_encryption_key( + catalog_pool.clone(), + enc_key.clone(), + Arc::clone(&engine.write_lock), + )); let credential_store: Arc = Arc::new(SqliteCredentialStore::new(catalog_pool, enc_key)); diff --git a/crates/storage-sqlite/src/management_store/access_keys.rs b/crates/storage-sqlite/src/management_store/access_keys.rs index 3429cfba5..3405be0d8 100644 --- a/crates/storage-sqlite/src/management_store/access_keys.rs +++ b/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) \ diff --git a/crates/storage-sqlite/src/management_store/accounts.rs b/crates/storage-sqlite/src/management_store/accounts.rs index 4a2a9938b..4013f21d4 100644 --- a/crates/storage-sqlite/src/management_store/accounts.rs +++ b/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()) diff --git a/crates/storage-sqlite/src/management_store/groups.rs b/crates/storage-sqlite/src/management_store/groups.rs index 2b27c06e3..e653c81a8 100644 --- a/crates/storage-sqlite/src/management_store/groups.rs +++ b/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 = ?", diff --git a/crates/storage-sqlite/src/management_store/mod.rs b/crates/storage-sqlite/src/management_store/mod.rs index 41f6fcd8b..8f858413e 100644 --- a/crates/storage-sqlite/src/management_store/mod.rs +++ b/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, \ diff --git a/crates/storage-sqlite/src/management_store/policies.rs b/crates/storage-sqlite/src/management_store/policies.rs index 0d92c3e90..da9e82e03 100644 --- a/crates/storage-sqlite/src/management_store/policies.rs +++ b/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 = ?", diff --git a/crates/storage-sqlite/src/management_store/roles.rs b/crates/storage-sqlite/src/management_store/roles.rs index 3ea40de2a..b3d81582b 100644 --- a/crates/storage-sqlite/src/management_store/roles.rs +++ b/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") diff --git a/crates/storage-sqlite/src/management_store/users.rs b/crates/storage-sqlite/src/management_store/users.rs index dc04018c4..13f1e97fa 100644 --- a/crates/storage-sqlite/src/management_store/users.rs +++ b/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") diff --git a/crates/storage-sqlite/src/store.rs b/crates/storage-sqlite/src/store.rs index 995ed7397..32eee9a51 100644 --- a/crates/storage-sqlite/src/store.rs +++ b/crates/storage-sqlite/src/store.rs @@ -29,12 +29,14 @@ //! 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. +//! The management stores (users, access keys, policies, groups, roles, +//! accounts, admin, settings, metrics) write through the separate catalog +//! pool in `lib.rs`, which for a file-backed database opens the same file; +//! the server wiring shares this same lock instance into +//! `SqliteCatalogStore`, so those writers queue with the engine's. The +//! credential store only reads. The one deliberate exclusion from the lock, +//! so the invariant stays auditable, is init-time bootstrap (in this file and +//! in `bootstrapper.rs`), which runs before the server serves traffic. use std::sync::Arc; use std::sync::atomic::AtomicU64; @@ -628,4 +630,155 @@ mod d1_write_lock_tests { ) .await; } + + /// Build a catalog store over the engine's pool and, critically, the + /// engine's OWN write lock, mirroring the server wiring in `lib.rs`. The + /// cached key lets access-key creation encrypt without a settings lookup. + fn catalog_store(engine: &SqliteEngine) -> crate::SqliteCatalogStore { + use base64::Engine as _; + let key = base64::engine::general_purpose::STANDARD.encode([0u8; 32]); + crate::SqliteCatalogStore::with_encryption_key( + engine.pool.clone(), + key, + std::sync::Arc::clone(&engine.write_lock), + ) + } + + /// The management-store writer observed losing its file lock in the + /// `run-integration-sqlite` flake (`put_policy: ... database is locked`, + /// surfacing as HTTP 500 where 204 was expected). + #[tokio::test] + async fn management_put_policy_waits_for_the_write_lock() { + use extenddb_storage::management_store::ManagementStore; + let engine = engine().await; + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + ManagementStore::put_policy( + &store, + "000000000000", + "user", + "d1-user", + "full", + &json!({"Version": "2012-10-17", "Statement": []}), + ) + .await + .expect("put policy"); + }, + "put_policy", + ) + .await; + } + + #[tokio::test] + async fn management_create_user_waits_for_the_write_lock() { + use extenddb_storage::management_store::ManagementStore; + let engine = engine().await; + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + ManagementStore::create_user(&store, "000000000000", "d1-user", None) + .await + .expect("create user"); + }, + "create_user", + ) + .await; + } + + #[tokio::test] + async fn management_create_access_key_waits_for_the_write_lock() { + use extenddb_storage::management_store::ManagementStore; + let engine = engine().await; + // The owning user must exist before the lock window (foreign key); + // key generation and encryption also run pre-lock inside the writer. + sqlx::query( + "INSERT INTO iam_users (account_id, user_name, user_arn) \ + VALUES ('000000000000', 'd1-key-user', 'arn:aws:iam::000000000000:user/d1-key-user')", + ) + .execute(&engine.pool) + .await + .expect("user"); + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + ManagementStore::create_access_key(&store, "000000000000", "d1-key-user") + .await + .expect("create access key"); + }, + "create_access_key", + ) + .await; + } + + #[tokio::test] + async fn management_create_account_waits_for_the_write_lock() { + use extenddb_storage::management_store::ManagementStore; + let engine = engine().await; + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + ManagementStore::create_account(&store, "111111111111", "d1-acct") + .await + .expect("create account"); + }, + "create_account", + ) + .await; + } + + #[tokio::test] + async fn management_create_group_waits_for_the_write_lock() { + use extenddb_storage::management_store::ManagementStore; + let engine = engine().await; + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + ManagementStore::create_group(&store, "000000000000", "d1-group") + .await + .expect("create group"); + }, + "create_group", + ) + .await; + } + + #[tokio::test] + async fn admin_create_waits_for_the_write_lock() { + use extenddb_storage::management_store::AdminStore; + let engine = engine().await; + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + AdminStore::create_admin(&store, "d1-admin", "hash") + .await + .expect("create admin"); + }, + "create_admin", + ) + .await; + } + + #[tokio::test] + async fn set_setting_waits_for_the_write_lock() { + use extenddb_storage::management_store::SettingsStore; + let engine = engine().await; + let store = catalog_store(&engine); + assert_serialized( + &engine, + async move { + SettingsStore::set_setting(&store, "d1-setting", "v") + .await + .expect("set setting"); + }, + "set_setting", + ) + .await; + } }