From fbb54eac954118ee49c1bf3c0eb32060f03ed30c Mon Sep 17 00:00:00 2001 From: OpenFUT Agent Date: Thu, 13 Aug 2026 20:20:14 +0000 Subject: [PATCH] fix(economy): BEGIN IMMEDIATE for write transactions (concurrency-safe) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the fresh-DB multi-connection write failure: economy writes ran in a DEFERRED transaction (pool.begin() = BEGIN) that read then wrote; SQLite returns SQLITE_BUSY (code 5, 'database is locked') *immediately* when a deferred tx upgrades to a write while another holds the write lock, bypassing busy_timeout to avoid deadlock. Fix: open each write op with BEGIN IMMEDIATE on a dedicated pooled connection (finish() commits/rolls back), taking the write lock up front so busy_timeout serializes writers. Reproduction test: 100 fresh DBs x 8 concurrent grant_reward — was 644/800 failures, now 800/800 succeed with correct final balance (no lost update). Reads unchanged. --- src/services/economy.rs | 106 +++++++++++++++++++++++++------------ tests/concurrency_repro.rs | 65 +++++++++++++++++++++++ 2 files changed, 137 insertions(+), 34 deletions(-) create mode 100644 tests/concurrency_repro.rs diff --git a/src/services/economy.rs b/src/services/economy.rs index e0504b5..8041ddc 100644 --- a/src/services/economy.rs +++ b/src/services/economy.rs @@ -215,6 +215,25 @@ pub async fn list_unopened_entitlements(pool: &Pool, club_id: &str) -> AppResult .collect()) } +/// Commit on `Ok`, roll back on `Err`. Paired with a `BEGIN IMMEDIATE` opened on +/// the same connection, so the write lock is held for the whole op and a +/// concurrent writer waits (honoring `busy_timeout`) instead of failing: a +/// DEFERRED `pool.begin()` upgrades to a write only at the first write, where +/// SQLite returns SQLITE_BUSY *immediately* (bypassing the busy handler to avoid +/// deadlock) — the fresh-DB multi-connection write failure. +async fn finish(conn: &mut SqliteConnection, result: AppResult) -> AppResult { + match result { + Ok(v) => { + sqlx::query("COMMIT").execute(&mut *conn).await?; + Ok(v) + } + Err(e) => { + let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await; + Err(e) + } + } +} + /// Debit `cost` and grant one entitlement, atomically. Fail-closed: if the club /// cannot afford `cost`, nothing is debited and no entitlement is created. pub async fn purchase_entitlement( @@ -223,14 +242,18 @@ pub async fn purchase_entitlement( cost: i64, definition_id: &str, ) -> AppResult { - let mut tx = pool.begin().await?; - let balance = debit(&mut tx, club_id, cost).await?; - let entitlement_id = grant_entitlement(&mut tx, club_id, definition_id).await?; - tx.commit().await?; - Ok(PurchaseReceipt { - balance, - entitlement_id, - }) + let mut conn = pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?; + let result = async { + let balance = debit(&mut conn, club_id, cost).await?; + let entitlement_id = grant_entitlement(&mut conn, club_id, definition_id).await?; + Ok(PurchaseReceipt { + balance, + entitlement_id, + }) + } + .await; + finish(&mut conn, result).await } /// Debit `cost` and mint one owned item, atomically. Fail-closed: if the club @@ -245,11 +268,15 @@ pub async fn purchase_item( item_id: &str, card_id: &str, ) -> AppResult { - let mut tx = pool.begin().await?; - let balance = debit(&mut tx, club_id, cost).await?; - add_item(&mut tx, club_id, item_id, card_id).await?; - tx.commit().await?; - Ok(balance) + let mut conn = pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?; + let result = async { + let balance = debit(&mut conn, club_id, cost).await?; + add_item(&mut conn, club_id, item_id, card_id).await?; + Ok(balance) + } + .await; + finish(&mut conn, result).await } /// Debit `cost` and mint several owned items, atomically. Fail-closed: if the @@ -263,13 +290,17 @@ pub async fn purchase_items( cost: i64, items: &[GrantedItem], ) -> AppResult { - let mut tx = pool.begin().await?; - let balance = debit(&mut tx, club_id, cost).await?; - for item in items { - add_item(&mut tx, club_id, &item.item_id, &item.card_id).await?; + let mut conn = pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?; + let result = async { + let balance = debit(&mut conn, club_id, cost).await?; + for item in items { + add_item(&mut conn, club_id, &item.item_id, &item.card_id).await?; + } + Ok(balance) } - tx.commit().await?; - Ok(balance) + .await; + finish(&mut conn, result).await } /// Consume an entitlement once and add its granted items, atomically. If any @@ -281,31 +312,38 @@ pub async fn redeem_entitlement( entitlement_id: &str, items: &[GrantedItem], ) -> AppResult { - let mut tx = pool.begin().await?; - let definition_id = consume_entitlement(&mut tx, club_id, entitlement_id).await?; - for item in items { - add_item(&mut tx, club_id, &item.item_id, &item.card_id).await?; + let mut conn = pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?; + let result = async { + let definition_id = consume_entitlement(&mut conn, club_id, entitlement_id).await?; + for item in items { + add_item(&mut conn, club_id, &item.item_id, &item.card_id).await?; + } + Ok(definition_id) } - tx.commit().await?; - Ok(definition_id) + .await; + finish(&mut conn, result).await } /// Remove an owned item and credit `price`, atomically. Fail-closed: if the item /// is not owned by the club nothing is credited. pub async fn sell_item(pool: &Pool, club_id: &str, item_id: &str, price: i64) -> AppResult { - let mut tx = pool.begin().await?; - remove_item(&mut tx, club_id, item_id).await?; - let balance = credit(&mut tx, club_id, price).await?; - tx.commit().await?; - Ok(balance) + let mut conn = pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?; + let result = async { + remove_item(&mut conn, club_id, item_id).await?; + credit(&mut conn, club_id, price).await + } + .await; + finish(&mut conn, result).await } /// Credit a reward to a club's balance atomically. pub async fn grant_reward(pool: &Pool, club_id: &str, amount: i64) -> AppResult { - let mut tx = pool.begin().await?; - let balance = credit(&mut tx, club_id, amount).await?; - tx.commit().await?; - Ok(balance) + let mut conn = pool.acquire().await?; + sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?; + let result = async { credit(&mut conn, club_id, amount).await }.await; + finish(&mut conn, result).await } #[cfg(test)] diff --git a/tests/concurrency_repro.rs b/tests/concurrency_repro.rs new file mode 100644 index 0000000..35939e9 --- /dev/null +++ b/tests/concurrency_repro.rs @@ -0,0 +1,65 @@ +//! Reproduction for the fresh-DB multi-connection warm-up write failure. +//! Forces several pooled connections to open concurrently on a brand-new DB and +//! captures the ACTUAL sqlx/SQLite error (not the service's generic string). + +use openfut_core::db::{init_pool, run_migrations}; +use openfut_core::services::economy; + +async fn seed_club(pool: &sqlx::SqlitePool) { + sqlx::query( + "INSERT INTO profiles (id, username, created_at, updated_at) VALUES ('p','p','t','t')", + ) + .execute(pool) + .await + .unwrap(); + sqlx::query("INSERT INTO clubs (id, profile_id, name, coins, created_at, updated_at) VALUES ('c','p','c',100000,'t','t')") + .execute(pool) + .await + .unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn fresh_db_multiconn_concurrent_writes() { + let base = std::env::temp_dir().join(format!("ofut-cc-{}", std::process::id())); + std::fs::create_dir_all(&base).unwrap(); + let iters = 100usize; + let mut failures = 0usize; + let mut first_err = String::new(); + for i in 0..iters { + let url = format!("sqlite://{}/db{i}.db", base.display()); + let pool = init_pool(&url, 5).await.expect("init_pool"); + run_migrations(&pool).await.expect("migrations"); + seed_club(&pool).await; + // Fire concurrent credits to force several connections to warm up at once + // on the brand-new DB, then a write — the harness's failing shape. + let mut handles = Vec::new(); + for _ in 0..8 { + let p = pool.clone(); + handles.push(tokio::spawn(async move { + economy::grant_reward(&p, "c", 1).await + })); + } + for h in handles { + match h.await.unwrap() { + Ok(_) => {} + Err(e) => { + failures += 1; + if first_err.is_empty() { + first_err = format!("{e:?}"); + } + } + } + } + // Serialization correctness: 8 concurrent +1 credits, no lost update. + let bal = economy::balance(&pool, "c").await.unwrap(); + assert_eq!(bal, 100_008, "iter {i}: lost update under concurrency"); + pool.close().await; + } + std::fs::remove_dir_all(&base).ok(); + assert_eq!( + failures, + 0, + "{failures}/{} iterations had a write failure; first error: {first_err}", + iters * 8 + ); +}