funman300/OpenFUT-Core
1
0
Fork 0
Code Issues 6 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(economy): BEGIN IMMEDIATE for write transactions (concurrency-safe)

Browse Source 
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.
This commit is contained in:
OpenFUT Agent
2026-08-13 20:20:14 +00:00
parent 75b183077f
commit fbb54eac95
2 changed files with 137 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/economy.rs
+63 -25
View File
@@ -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<T>(conn: &mut SqliteConnection, result: AppResult<T>) -> AppResult<T> {
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,15 +242,19 @@ pub async fn purchase_entitlement(
cost: i64,
definition_id: &str,
) -> AppResult<PurchaseReceipt> {
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?;
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
/// cannot afford `cost`, nothing is debited and no item is added. This is the
@@ -245,12 +268,16 @@ pub async fn purchase_item(
item_id: &str,
card_id: &str,
) -> AppResult<i64> {
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?;
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
/// club cannot afford `cost`, nothing is debited and no items are added; if any
@@ -263,14 +290,18 @@ pub async fn purchase_items(
cost: i64,
items: &[GrantedItem],
) -> AppResult<i64> {
let mut tx = pool.begin().await?;
let balance = debit(&mut tx, club_id, cost).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 tx, club_id, &item.item_id, &item.card_id).await?;
add_item(&mut conn, club_id, &item.item_id, &item.card_id).await?;
}
tx.commit().await?;
Ok(balance)
}
.await;
finish(&mut conn, result).await
}
/// Consume an entitlement once and add its granted items, atomically. If any
/// item insert fails (e.g. a colliding instance id) the whole redemption rolls
@@ -281,31 +312,38 @@ pub async fn redeem_entitlement(
entitlement_id: &str,
items: &[GrantedItem],
) -> AppResult<String> {
let mut tx = pool.begin().await?;
let definition_id = consume_entitlement(&mut tx, club_id, entitlement_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 tx, club_id, &item.item_id, &item.card_id).await?;
add_item(&mut conn, club_id, &item.item_id, &item.card_id).await?;
}
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<i64> {
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<i64> {
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)]
tests/concurrency_repro.rs
+65
View File
@@ -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
);
}