diff --git a/openfut-utas-host/Cargo.toml b/openfut-utas-host/Cargo.toml index 64d5bd4..83563a5 100644 --- a/openfut-utas-host/Cargo.toml +++ b/openfut-utas-host/Cargo.toml @@ -27,6 +27,9 @@ rand = "0.8" # (sqlx runtime-tokio). Created once in `Server::from_config`, shared via `Arc`; # the `AsyncBridge` bridges the synchronous thread-per-connection dispatch to it. tokio = { version = "1", features = ["rt-multi-thread"] } +# Fast, non-poisoning mutex guarding the in-memory hot paths of the durable +# market/pile stores (used in `src/market_store.rs` / `src/pile_store.rs`). +parking_lot = "0.12" [dev-dependencies] parking_lot = "0.12" diff --git a/openfut-utas-host/src/market_store.rs b/openfut-utas-host/src/market_store.rs index 59d049e..70f3903 100644 --- a/openfut-utas-host/src/market_store.rs +++ b/openfut-utas-host/src/market_store.rs @@ -29,11 +29,62 @@ //! `CHECK` constraint even though it is a transient intermediate — omitting it //! would make [`MarketStore::reserve_listing`] fail the constraint. +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use std::time::Duration; +use parking_lot::Mutex; + use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions}; use sqlx::{ConnectOptions, Connection, Row, SqlitePool}; +/// Test-only durable-store fault injector, shared (cheap `Arc` clone) between a +/// store and the failure-injection tests. It is **inert in production**: nothing +/// arms it, so each guarded op reads one relaxed atomic and behaves exactly as +/// before. The economy failure tests use it to force a durable-store write to +/// fail at a chosen point (market complete-sale / reserve, pile write) — the +/// only way to exercise those recovery paths, since the stores are concrete +/// types wired straight into the handlers (no trait seam to substitute). +#[derive(Clone, Default)] +pub struct StoreFault { + inner: Arc, +} + +#[derive(Default)] +struct StoreFaultInner { + any: AtomicBool, + armed: Mutex>, +} + +impl StoreFault { + /// Arm `op` to fail its next `times` invocations, then heal automatically. + pub fn arm(&self, op: &'static str, times: u32) { + self.inner.armed.lock().insert(op, times); + self.inner.any.store(true, Ordering::SeqCst); + } + + /// Consume one armed unit for `op`, returning whether it should fail now. + /// Fast path (unarmed): a single relaxed atomic load, no lock taken. + pub fn tripped(&self, op: &'static str) -> bool { + if !self.inner.any.load(Ordering::Relaxed) { + return false; + } + let mut armed = self.inner.armed.lock(); + let fire = match armed.get_mut(op) { + Some(n) if *n > 0 => { + *n -= 1; + true + } + _ => false, + }; + if armed.values().all(|&n| n == 0) { + self.inner.any.store(false, Ordering::SeqCst); + } + fire + } +} + /// Typed failure of a listing operation. `Db` wraps an infrastructure error /// (transport/encoding); everything else is a modelled lifecycle outcome. #[derive(Debug)] @@ -139,6 +190,7 @@ fn row_to_listing(row: &sqlx::sqlite::SqliteRow) -> Listing { #[derive(Clone)] pub struct MarketStore { pool: SqlitePool, + fault: StoreFault, } impl MarketStore { @@ -171,7 +223,16 @@ impl MarketStore { .execute(&pool) .await .map_err(db)?; - Ok(MarketStore { pool }) + Ok(MarketStore { + pool, + fault: StoreFault::default(), + }) + } + + /// A shared handle to this store's test-only fault switch (inert unless a + /// test arms it). Production never calls it. + pub fn fault(&self) -> StoreFault { + self.fault.clone() } /// Insert a new `active` listing. `listing_id` is the numeric-string trade id @@ -315,12 +376,18 @@ impl MarketStore { /// Reserve an `active` listing (`active -> reserved`). Returns whether this /// caller won the reservation. Exactly one of two concurrent callers wins. pub async fn reserve_listing(&self, listing_id: &str) -> Result { + if self.fault.tripped("reserve") { + return Err(MarketError::Db("injected reserve fault".into())); + } self.cas(listing_id, "active", "reserved").await } /// Finalise a won reservation (`reserved -> sold`). A listing not in /// `reserved` is a [`MarketError::Conflict`]. pub async fn complete_sale(&self, listing_id: &str) -> Result<(), MarketError> { + if self.fault.tripped("complete_sale") { + return Err(MarketError::Db("injected complete_sale fault".into())); + } if self.cas(listing_id, "reserved", "sold").await? { Ok(()) } else { diff --git a/openfut-utas-host/src/pile_store.rs b/openfut-utas-host/src/pile_store.rs index caa049a..95b8d05 100644 --- a/openfut-utas-host/src/pile_store.rs +++ b/openfut-utas-host/src/pile_store.rs @@ -55,6 +55,7 @@ fn now_millis() -> String { #[derive(Clone)] pub struct PileStore { pool: SqlitePool, + fault: crate::market_store::StoreFault, } impl PileStore { @@ -84,7 +85,16 @@ impl PileStore { .execute(&pool) .await .map_err(db)?; - Ok(PileStore { pool }) + Ok(PileStore { + pool, + fault: crate::market_store::StoreFault::default(), + }) + } + + /// A shared handle to this store's test-only fault switch (inert unless a + /// test arms it). Production never calls it. + pub fn fault(&self) -> crate::market_store::StoreFault { + self.fault.clone() } /// The current pile of a Core-owned item, or `None` if none is recorded. @@ -100,6 +110,9 @@ impl PileStore { /// Set (upsert) the pile of a Core-owned item. Durable and race-safe /// (`BEGIN IMMEDIATE` + upsert). pub async fn set(&self, core_item_id: &str, pile: &str) -> Result<(), PileError> { + if self.fault.tripped("set") { + return Err(PileError::Db("injected pile set fault".into())); + } let updated_at = now_millis(); let mut conn = self.pool.acquire().await.map_err(db)?; sqlx::query("BEGIN IMMEDIATE") diff --git a/openfut-utas-host/tests/economy_concurrency.rs b/openfut-utas-host/tests/economy_concurrency.rs new file mode 100644 index 0000000..b40f838 --- /dev/null +++ b/openfut-utas-host/tests/economy_concurrency.rs @@ -0,0 +1,692 @@ +//! Real host↔Core economy CONCURRENCY proofs. +//! +//! Every race here drives the REAL host dispatch (`Server::try_handle_economy`) +//! from several plain `std::thread`s against ONE live in-process Core (axum, +//! ephemeral loopback port, disposable temp SQLite) seeded with the FIFA17 dev +//! inventory. No fakes: the Core economy transport is the real blocking +//! `HttpCoreClient`, the durable listing/pile stores are the real SQLite stores, +//! and the identity store is the real `JsonIdentityStore`. +//! +//! Execution contract (mirrors `economy_integration.rs`): the blocking Core +//! client MUST NOT run while a Tokio runtime is entered on the thread, so all +//! economy work runs on plain `std::thread`s joined under a single +//! `spawn_blocking`, and the async bridge always takes its direct `block_on` +//! path off-runtime. Each racer is its own plain OS thread holding a cheap +//! `Server` clone (all shared state is `Arc`), exactly like the production +//! thread-per-connection server. +//! +//! Safety: temp dir + `127.0.0.1:0` only. Never touches production/`.105`. + +use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog; +use openfut_adapter_fifa17::fut::entities::Fifa17Entities; +use openfut_identity::{ExternalIdentityStore, JsonIdentityStore}; +use openfut_utas_host::async_bridge::AsyncBridge; +use openfut_utas_host::market_store::MarketStore; +use openfut_utas_host::pile_store::PileStore; +use openfut_utas_host::{ + build_content_pool, CoreAccess, CoreEconomy, EconomyServices, Fifa17IdentityResolver, + HttpCoreClient, PassClient, Server, WireResponse, +}; +use serde_json::Value; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +/// Race iterations per case (the brief asks for 50–100 fresh-state repeats). +const ITERS: usize = 50; + +const GAME: &str = "fifa17"; +const OWNED_KIND: &str = "owned-item"; +const OWNED_FLOOR: i64 = 100_000_001; + +// ───────────────────────────── Core boot (seeded) ─────────────────────────── + +/// Boot a Core with the FIFA17 dev CONTENT loaded and, on first boot, the dev +/// inventory SEEDED (fifa17 profile + club, 100k coins, one owned instance per +/// definition). Identical to the reference harness. +async fn start_core_seeded(db_url: &str, seed: bool) -> (tokio::task::JoinHandle<()>, String) { + use openfut_core::config::Config; + use openfut_core::seed::{seed_fifa17_dev, FIFA17_GAME}; + use openfut_core::services::card_db::CardDb; + + let data_dir = "../openfut-core/data"; + let pool = openfut_core::db::init_pool(db_url, 5) + .await + .expect("core pool"); + openfut_core::db::run_migrations(&pool) + .await + .expect("core migrations"); + if seed { + let mut card_db = CardDb::load(data_dir).expect("card_db load"); + card_db + .load_game_dev(data_dir, FIFA17_GAME) + .expect("load fifa17 dev content"); + seed_fifa17_dev(&pool, &card_db) + .await + .expect("seed fifa17 dev inventory"); + } + let cfg = Config { + listen_addr: "127.0.0.1:0".into(), + database_url: "sqlite::memory:".into(), + data_dir: data_dir.into(), + max_connections: 5, + dev_content_games: vec![FIFA17_GAME.to_string()], + content_packs: Vec::new(), + }; + let app = openfut_core::app::build(pool, cfg) + .await + .expect("core app::build"); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral"); + let addr = listener.local_addr().unwrap(); + let handle = tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + (handle, format!("http://{addr}")) +} + +fn wait_ready(base: &str) { + let http = reqwest::blocking::Client::new(); + for _ in 0..1500 { + if let Ok(r) = http.get(format!("{base}/health")).send() { + if r.status().is_success() { + return; + } + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } + panic!("core did not become ready at {base}"); +} + +// ───────────────────────────── Harness ────────────────────────────────────── + +/// Everything a race needs: the real `Server`, a direct client for Core-state +/// assertions, the identity resolver + its concrete store (watermark checks), +/// and a valid wire `resourceId` that reverse-maps to a real Core `card_id`. +struct Harness { + server: Server, + client: HttpCoreClient, + resolver: Arc, + ident: Arc, + sample_resource: i64, +} + +/// Build the real economy `Server` from the seeded Core content (same wiring as +/// `economy_integration::build_econ_server`, plus a retained concrete identity +/// store handle for watermark assertions). +fn build_harness(base: &str, dir: &std::path::Path) -> Harness { + let probe = HttpCoreClient::new(base, GAME); + let owned = probe.all_owned().expect("core collection"); + assert!(!owned.is_empty(), "seed must grant a starter collection"); + let mut entries = String::new(); + let mut seen = HashSet::new(); + let mut asset = 20000u32; + for it in &owned { + if !seen.insert(it.card_id.clone()) { + continue; + } + if !entries.is_empty() { + entries.push(','); + } + entries.push_str(&format!("\"{}\":{{\"asset_id\":{asset}}}", it.card_id)); + asset += 1; + } + let doc = format!("{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{entries}}}}}"); + let catalog = Fifa17CardCatalog::from_json_str(&doc).expect("catalog"); + let ident = Arc::new( + JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).expect("identity"), + ); + let dyn_store: Arc = ident.clone(); + let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, dyn_store)); + let entities = Arc::new(Fifa17Entities::from_maps( + HashMap::new(), + HashMap::new(), + HashMap::new(), + )); + let core: Arc = Arc::new(HttpCoreClient::new(base, GAME)); + let bridge = Arc::new(AsyncBridge::new().unwrap()); + let market_path = dir.join("market.db").to_string_lossy().into_owned(); + let market = Arc::new( + bridge + .block_on(async move { MarketStore::open(&market_path).await }) + .expect("market store"), + ); + let pile_path = dir.join("pile.db").to_string_lossy().into_owned(); + let piles = Arc::new( + bridge + .block_on(async move { PileStore::open(&pile_path).await }) + .expect("pile store"), + ); + let econ: Arc = Arc::new(HttpCoreClient::new(base, GAME)); + let pool = Arc::new(build_content_pool(core.as_ref(), resolver.as_ref())); + assert!( + !pool.is_empty(), + "content pool derived from real Core content" + ); + let services = Arc::new(EconomyServices { + econ, + market, + piles, + bridge, + pool, + }); + let server = Server::new( + core, + entities, + resolver.clone(), + Arc::new(PassClient::new("http://127.0.0.1:9")), + 33068179, + ) + .with_economy(services); + Harness { + server, + client: HttpCoreClient::new(base, GAME), + resolver, + ident, + sample_resource: 20000, + } +} + +// ───────────────────────────── Race helpers ───────────────────────────────── + +type Req = (&'static str, String, Vec); + +/// Dispatch each request on its own plain OS thread (no ambient runtime, so the +/// bridge takes its direct `block_on` path), returning the responses in order. +fn fire(server: &Server, reqs: Vec) -> Vec { + let handles: Vec<_> = reqs + .into_iter() + .map(|(m, p, b)| { + let s = server.clone(); + std::thread::spawn(move || { + s.try_handle_economy(m, &p, &[], &b, None) + .expect("economy route matched") + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() +} + +fn bj(r: &WireResponse) -> Value { + serde_json::from_slice(&r.body).unwrap_or(Value::Null) +} + +/// Drive the club balance to exactly `target` via the real Core economy API. +fn set_balance(client: &HttpCoreClient, target: i64) { + let cur = client.balance().unwrap(); + if cur > target { + client + .purchase_entitlement(cur - target, "econ-test-drain") + .expect("drain"); + } else if cur < target { + client.grant_reward(target - cur).expect("top up"); + } + assert_eq!(client.balance().unwrap(), target, "balance set"); +} + +/// The on-wire quick-sell `discardValue` tiers (host `economy_store::quick_sell_value`), +/// replicated to assert the EXACT credit — the client-visible contract. +fn qs_value(rating: u8) -> i64 { + match rating { + r if r >= 85 => 1500, + r if r >= 80 => 900, + r if r >= 75 => 600, + r if r >= 65 => 300, + _ => 150, + } +} + +fn owns(client: &HttpCoreClient, core_id: &str) -> bool { + client + .all_owned() + .unwrap() + .iter() + .any(|it| it.owned_card_id == core_id) +} + +fn rating_of(client: &HttpCoreClient, core_id: &str) -> u8 { + client + .all_owned() + .unwrap() + .into_iter() + .find(|it| it.owned_card_id == core_id) + .expect("owned") + .rating +} + +/// Mint one owned card via a real Store BUY (pack 1). Returns (wire_id, core_id). +/// Balance is set high enough that the buy always succeeds. +fn mint_one(h: &Harness) -> (i64, String) { + set_balance(&h.client, 50_000); + let resp = h + .server + .try_handle_economy( + "PUT", + "/ut/game/fifa17/store/transaction", + &[], + br#"{"packId":1}"#, + None, + ) + .expect("store buy routed"); + assert_eq!(resp.status, 200, "mint buy 200"); + let items = bj(&resp)["createPackResponse"]["itemList"] + .as_array() + .expect("itemList") + .clone(); + let wire = items[0]["id"].as_i64().expect("wire id"); + let core = h.resolver.owned_id_for_wire(wire).expect("reverse"); + (wire, core) +} + +// ───────────────────────────── Cases ──────────────────────────────────────── + +/// A: two Store BUYs when only ONE is affordable → exactly one 200 + one 461, +/// final balance 0, exactly one debit. +fn case_a_two_buys(h: &Harness) -> String { + let (mut ok, mut refused) = (0u32, 0u32); + for _ in 0..ITERS { + set_balance(&h.client, 400); // exactly one Bronze pack (price 400) + let rs = fire( + &h.server, + vec![ + ( + "PUT", + "/ut/game/fifa17/store/transaction".into(), + b"{\"packId\":1}".to_vec(), + ), + ( + "PUT", + "/ut/game/fifa17/store/transaction".into(), + b"{\"packId\":1}".to_vec(), + ), + ], + ); + let statuses: Vec = rs.iter().map(|r| r.status).collect(); + let wins = statuses.iter().filter(|&&s| s == 200).count(); + let refs = statuses.iter().filter(|&&s| s == 461).count(); + assert_eq!(wins, 1, "exactly one BUY wins (statuses {statuses:?})"); + assert_eq!( + refs, 1, + "exactly one BUY refused 461 (statuses {statuses:?})" + ); + // The winner minted 5 cards; the loser minted nothing. + for r in &rs { + if r.status == 200 { + assert_eq!( + bj(r)["createPackResponse"]["itemList"] + .as_array() + .unwrap() + .len(), + 5 + ); + } + } + assert_eq!(h.client.balance().unwrap(), 0, "exactly one 400 debit → 0"); + ok += wins as u32; + refused += refs as u32; + } + format!("A two-buys: {ITERS} iters, {ok} wins / {refused} refused, always 1+1, final=0") +} + +/// B: duplicate OPEN of the same seeded "70" entitlement → exactly one +/// redemption, inventory granted once (11 cards), entitlement consumed once, +/// the loser fails safe. +fn case_b_dup_open(h: &Harness) -> String { + let mut winners = 0u32; + for _ in 0..ITERS { + // Seed exactly one unopened "70" entitlement (cost 0). + h.client + .purchase_entitlement(0, "70") + .expect("seed pack 70"); + let owned_before = h.client.all_owned().unwrap().len(); + let rs = fire( + &h.server, + vec![ + ( + "POST", + "/ut/game/fifa17/purchased".into(), + b"{\"packId\":70}".to_vec(), + ), + ( + "POST", + "/ut/game/fifa17/purchased".into(), + b"{\"packId\":70}".to_vec(), + ), + ], + ); + // A real open echoes packId==70; the loser is 503 or an empty reveal. + let real_opens = rs + .iter() + .filter(|r| r.status == 200 && bj(r)["packId"].as_u64() == Some(70)) + .count(); + assert_eq!(real_opens, 1, "exactly one redemption"); + let owned_after = h.client.all_owned().unwrap().len(); + assert_eq!( + owned_after - owned_before, + 11, + "inventory granted once (11 cards)" + ); + // Entitlement consumed exactly once → no unopened "70" remains. + let unopened_70 = h + .client + .entitlements() + .unwrap() + .iter() + .filter(|e| e.definition_id == "70") + .count(); + assert_eq!(unopened_70, 0, "entitlement consumed once"); + winners += real_opens as u32; + } + format!("B dup-open: {ITERS} iters, {winners} single redemptions, +11 once, ent consumed once") +} + +/// C: duplicate QUICK-SELL of the same wire id → exactly one sell + one credit + +/// one removal. +fn case_c_dup_quicksell(h: &Harness) -> String { + let mut sells = 0u32; + for _ in 0..ITERS { + let (wire, core) = mint_one(h); + let value = qs_value(rating_of(&h.client, &core)); + let before = h.client.balance().unwrap(); + let path = format!("/ut/game/fifa17/item/{wire}"); + let rs = fire( + &h.server, + vec![ + ("DELETE", path.clone(), Vec::new()), + ("DELETE", path.clone(), Vec::new()), + ], + ); + // Exactly one response actually accounted the sale (items lists the wire). + let sold = rs + .iter() + .filter(|r| { + r.status == 200 + && bj(r)["items"] + .as_array() + .map(|a| a.iter().any(|x| x["id"].as_i64() == Some(wire))) + .unwrap_or(false) + }) + .count(); + assert_eq!(sold, 1, "exactly one quick-sell accounted the item"); + assert!(!owns(&h.client, &core), "item removed exactly once"); + assert_eq!( + h.client.balance().unwrap(), + before + value, + "credited exactly one discardValue (no double credit)" + ); + sells += sold as u32; + } + format!("C dup-quicksell: {ITERS} iters, {sells} single sells, one credit + one removal each") +} + +/// D: two buyers of the SAME listing → one success, one closed/empty; sold once; +/// exactly one debit; exactly one mint. +fn case_d_two_market_buyers(h: &Harness) -> String { + let mut wins = 0u32; + for i in 0..ITERS { + set_balance(&h.client, 50_000); + let item_id = 500_000 + i as i64; // unique listing per iteration + let list = h + .server + .try_handle_economy( + "POST", + "/ut/game/fifa17/auctionhouse", + &[], + format!( + r#"{{"itemData":{{"id":{item_id},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#, + h.sample_resource + ) + .as_bytes(), + None, + ) + .expect("list routed"); + let trade_id = bj(&list)["id"].as_i64().expect("trade id"); + let path = format!("/ut/game/fifa17/trade/{trade_id}"); + let before = h.client.balance().unwrap(); + let rs = fire( + &h.server, + vec![ + ("POST", path.clone(), b"{}".to_vec()), + ("POST", path.clone(), b"{}".to_vec()), + ], + ); + // Winner returns a non-empty auctionInfo; loser an empty one. + let non_empty = rs + .iter() + .filter(|r| { + bj(r)["auctionInfo"] + .as_array() + .map(|a| !a.is_empty()) + .unwrap_or(false) + }) + .count(); + assert_eq!(non_empty, 1, "exactly one buyer wins the listing"); + assert_eq!( + h.client.balance().unwrap(), + before - 1000, + "exactly one 1000 debit" + ); + // Exactly one mint of the deterministic won-card id. + let mint_id = format!("market-buy:{trade_id}"); + let mints = h + .client + .all_owned() + .unwrap() + .iter() + .filter(|it| it.owned_card_id == mint_id) + .count(); + assert_eq!(mints, 1, "exactly one card minted for the sold listing"); + wins += non_empty as u32; + } + format!( + "D two-market-buyers: {ITERS} iters, {wins} single winners, sold once, 1 debit + 1 mint" + ) +} + +/// E: match REWARD + Store BUY concurrently → final balance is one legal +/// serialization (no lost update). Reward (+400) and buy (−400) commute, so the +/// final balance must equal the start exactly. +fn case_e_reward_and_buy(h: &Harness) -> String { + let start = 8_000i64; + for _ in 0..ITERS { + set_balance(&h.client, start); + let rs = fire( + &h.server, + vec![ + ( + "POST", + "/ut/delete/game/fifa17/match".into(), + b"{\"endReason\":\"WIN\"}".to_vec(), + ), + ( + "PUT", + "/ut/game/fifa17/store/transaction".into(), + b"{\"packId\":1}".to_vec(), + ), + ], + ); + assert!(rs.iter().all(|r| r.status == 200), "both ops succeed"); + assert_eq!( + h.client.balance().unwrap(), + start, + "reward(+400) and buy(-400) both applied: no lost update" + ); + } + format!("E reward+buy: {ITERS} iters, final==start ({start}) every time (no lost update)") +} + +/// F: MOVE + QUICK-SELL of the same item → one coherent final state (item sold +/// once, one credit, and no SOLD item rendered in any pile view). +fn case_f_move_and_quicksell(h: &Harness) -> String { + for _ in 0..ITERS { + let (wire, core) = mint_one(h); + let value = qs_value(rating_of(&h.client, &core)); + let before = h.client.balance().unwrap(); + let del_path = format!("/ut/game/fifa17/item/{wire}"); + let rs = fire( + &h.server, + vec![ + ( + "PUT", + "/ut/game/fifa17/item".into(), + format!(r#"{{"itemData":[{{"id":{wire},"pile":"trade"}}]}}"#).into_bytes(), + ), + ("DELETE", del_path, Vec::new()), + ], + ); + assert!(rs.iter().all(|r| r.status == 200), "both ops respond 200"); + // Exactly one real disposal: Core no longer owns the item, credited once. + assert!( + !owns(&h.client, &core), + "item sold (Core no longer owns it)" + ); + assert_eq!( + h.client.balance().unwrap(), + before + value, + "exactly one credit (move never credits)" + ); + // Coherence: the sold item is not rendered in the purchased-pile reveal. + let reveal = h + .server + .try_handle_economy("GET", "/ut/game/fifa17/purchased", &[], b"", None) + .expect("reveal routed"); + let shown = bj(&reveal)["itemData"] + .as_array() + .map(|a| a.iter().any(|x| x["id"].as_i64() == Some(wire))) + .unwrap_or(false); + assert!(!shown, "no sold item left in a visible pile"); + } + format!("F move+quicksell: {ITERS} iters, sold once + one credit + no sold item in a pile") +} + +/// G: LIST + QUICK-SELL of the same owned item → only one legal transition of +/// the real item's Core ownership (the sell). The listing is a synthetic-seller +/// row that does not dispose of the owned instance, so it never becomes a second +/// credit or a double removal. +fn case_g_list_and_quicksell(h: &Harness) -> String { + for i in 0..ITERS { + let (wire, core) = mint_one(h); + let value = qs_value(rating_of(&h.client, &core)); + let before = h.client.balance().unwrap(); + let list_body = format!( + r#"{{"itemData":{{"id":{wire},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#, + h.sample_resource + ); + let del_path = format!("/ut/game/fifa17/item/{wire}"); + let rs = fire( + &h.server, + vec![ + ( + "POST", + "/ut/game/fifa17/auctionhouse".into(), + list_body.into_bytes(), + ), + ("DELETE", del_path, Vec::new()), + ], + ); + assert!(rs.iter().all(|r| r.status == 200), "both ops respond 200"); + // The one legal ownership transition: the item is sold exactly once. + assert!(!owns(&h.client, &core), "item removed exactly once"); + assert_eq!( + h.client.balance().unwrap(), + before + value, + "credited exactly once (listing does not credit)" + ); + let _ = i; + } + format!( + "G list+quicksell: {ITERS} iters, exactly one ownership transition (sell); listing is synthetic" + ) +} + +/// H: concurrent MINTS (Store BUYs) → no duplicate FIFA wire ids across all +/// mints, every wire id reverse-resolves, and the identity watermark stays +/// monotonic (the next id to issue is strictly above every issued id). +fn case_h_concurrent_mints(h: &Harness) -> String { + let mut all: HashSet = HashSet::new(); + let mut total = 0usize; + for _ in 0..ITERS { + set_balance(&h.client, 50_000); + let reqs: Vec = (0..4) + .map(|_| { + ( + "PUT", + "/ut/game/fifa17/store/transaction".to_string(), + b"{\"packId\":1}".to_vec(), + ) + }) + .collect(); + let rs = fire(&h.server, reqs); + for r in &rs { + assert_eq!(r.status, 200, "concurrent buy 200"); + for it in bj(r)["createPackResponse"]["itemList"].as_array().unwrap() { + let wire = it["id"].as_i64().expect("wire id"); + assert!( + all.insert(wire), + "wire id {wire} is globally unique (no duplicate)" + ); + assert!( + h.resolver.owned_id_for_wire(wire).is_some(), + "wire id {wire} reverse-resolves to a Core instance" + ); + total += 1; + } + } + } + // Monotonic watermark: the next id the store would issue is strictly above + // every id it has ever issued. + let next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR); + let max_seen = all.iter().copied().max().unwrap_or(OWNED_FLOOR - 1); + assert!(next > max_seen, "watermark {next} > max issued {max_seen}"); + assert!( + all.iter().all(|&w| w >= OWNED_FLOOR), + "every wire id ≥ floor" + ); + format!( + "H concurrent-mints: {ITERS}×4 buys, {total} mints, all unique + reversible, watermark {next} > {max_seen}" + ) +} + +fn run_all_races(base: &str, dir: &std::path::Path) -> String { + wait_ready(base); + let h = build_harness(base, dir); + [ + case_a_two_buys(&h), + case_b_dup_open(&h), + case_c_dup_quicksell(&h), + case_d_two_market_buyers(&h), + case_e_reward_and_buy(&h), + case_f_move_and_quicksell(&h), + case_g_list_and_quicksell(&h), + case_h_concurrent_mints(&h), + ] + .join("\n") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn economy_concurrency_races() { + let dir = std::env::temp_dir().join(format!( + "openfut-econ-conc-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + let db_url = format!("sqlite://{}/econ.db", dir.display()); + + let (h1, base1) = start_core_seeded(&db_url, true).await; + let (b1, d1) = (base1.clone(), dir.clone()); + let summary = tokio::task::spawn_blocking(move || { + std::thread::spawn(move || run_all_races(&b1, &d1)) + .join() + .expect("race thread") + }) + .await + .expect("races"); + h1.abort(); + std::fs::remove_dir_all(&dir).ok(); + + eprintln!("economy_concurrency summary:\n{summary}"); +} diff --git a/openfut-utas-host/tests/economy_failure.rs b/openfut-utas-host/tests/economy_failure.rs new file mode 100644 index 0000000..37be11f --- /dev/null +++ b/openfut-utas-host/tests/economy_failure.rs @@ -0,0 +1,882 @@ +//! Real host↔Core economy FAILURE-INJECTION proofs. +//! +//! Each case drives the REAL host dispatch (`Server::try_handle_economy`) +//! against a live in-process Core, but with a chosen dependency armed to fail at +//! a chosen point, proving the economy cluster is fail-closed and leaves NO +//! silent Core-vs-host state divergence: +//! +//! * A wrapping [`CoreEconomy`] double (`FaultEconomy`) forwards to the real +//! blocking `HttpCoreClient` but can be armed to fail one op with a +//! `CoreError` — the transport-failure seam. +//! * A wrapping [`ExternalIdentityStore`] double (`FaultIdentity`) forwards to +//! the real `JsonIdentityStore` but can be armed to fail wire-id allocation. +//! * The durable listing/pile stores are the REAL SQLite stores; their narrow, +//! inert-by-default `StoreFault` switch (host `market_store`/`pile_store`) is +//! armed to fail a market reserve / complete-sale / pile write. That seam is +//! the only way to fault those concrete stores, which are wired straight into +//! the handlers with no trait to substitute. +//! +//! Execution contract mirrors `economy_integration.rs`: all economy work runs on +//! plain `std::thread`s off any Tokio runtime, so the blocking Core client and +//! the async bridge behave exactly as in the thread-per-connection server. +//! +//! Safety: temp dir + `127.0.0.1:0` only. Never touches production/`.105`. + +use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog; +use openfut_adapter_fifa17::fut::club_response::ItemIdentityResolver; +use openfut_adapter_fifa17::fut::entities::Fifa17Entities; +use openfut_identity::{ExternalIdentityStore, IdError, JsonIdentityStore}; +use openfut_utas_host::async_bridge::AsyncBridge; +use openfut_utas_host::market_store::MarketStore; +use openfut_utas_host::pile_store::PileStore; +use openfut_utas_host::{ + build_content_pool, CoreAccess, CoreEconomy, CoreError, EconomyEntitlement, EconomyGrantItem, + EconomyPurchase, EconomyServices, Fifa17IdentityResolver, HttpCoreClient, PassClient, Server, + WireResponse, +}; +use parking_lot::Mutex; +use serde_json::Value; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +const GAME: &str = "fifa17"; +const OWNED_KIND: &str = "owned-item"; +const OWNED_FLOOR: i64 = 100_000_001; + +// ───────────────────────────── Fault doubles ──────────────────────────────── + +/// A `CoreEconomy` that forwards every op to the real `HttpCoreClient` but can +/// be armed to fail a named op's next N invocations with a `CoreError`. +struct FaultEconomy { + inner: HttpCoreClient, + fail: Mutex>, +} + +impl FaultEconomy { + fn new(base: &str) -> Self { + FaultEconomy { + inner: HttpCoreClient::new(base, GAME), + fail: Mutex::new(HashMap::new()), + } + } + fn arm(&self, op: &'static str, times: u32) { + self.fail.lock().insert(op, times); + } + fn trip(&self, op: &'static str) -> bool { + let mut g = self.fail.lock(); + match g.get_mut(op) { + Some(n) if *n > 0 => { + *n -= 1; + true + } + _ => false, + } + } + fn injected() -> CoreError { + CoreError::Http("injected core fault".into()) + } +} + +impl CoreEconomy for FaultEconomy { + fn balance(&self) -> Result { + if self.trip("balance") { + return Err(Self::injected()); + } + self.inner.balance() + } + fn entitlements(&self) -> Result, CoreError> { + if self.trip("entitlements") { + return Err(Self::injected()); + } + self.inner.entitlements() + } + fn purchase_entitlement( + &self, + cost: i64, + definition_id: &str, + ) -> Result { + if self.trip("purchase_entitlement") { + return Err(Self::injected()); + } + self.inner.purchase_entitlement(cost, definition_id) + } + fn redeem_entitlement( + &self, + entitlement_id: &str, + items: &[EconomyGrantItem], + ) -> Result { + if self.trip("redeem_entitlement") { + return Err(Self::injected()); + } + self.inner.redeem_entitlement(entitlement_id, items) + } + fn sell_item(&self, item_id: &str, price: i64) -> Result { + if self.trip("sell_item") { + return Err(Self::injected()); + } + self.inner.sell_item(item_id, price) + } + fn grant_reward(&self, amount: i64) -> Result { + if self.trip("grant_reward") { + return Err(Self::injected()); + } + self.inner.grant_reward(amount) + } + fn purchase_item(&self, cost: i64, item_id: &str, card_id: &str) -> Result { + if self.trip("purchase_item") { + return Err(Self::injected()); + } + self.inner.purchase_item(cost, item_id, card_id) + } + fn purchase_items(&self, cost: i64, items: &[EconomyGrantItem]) -> Result { + if self.trip("purchase_items") { + return Err(Self::injected()); + } + self.inner.purchase_items(cost, items) + } +} + +/// An `ExternalIdentityStore` that forwards to a real `JsonIdentityStore` but can +/// be armed to fail wire-id allocation (`resolve_or_allocate`). +struct FaultIdentity { + inner: Arc, + fail_alloc: Mutex, +} + +impl FaultIdentity { + fn new(inner: Arc) -> Self { + FaultIdentity { + inner, + fail_alloc: Mutex::new(0), + } + } + fn arm_alloc(&self, times: u32) { + *self.fail_alloc.lock() = times; + } +} + +impl ExternalIdentityStore for FaultIdentity { + fn resolve_or_allocate( + &self, + game: &str, + kind: &str, + core_id: &str, + base_floor: i64, + ) -> Result { + { + let mut n = self.fail_alloc.lock(); + if *n > 0 { + *n -= 1; + return Err(IdError::Corrupt("injected identity alloc fault".into())); + } + } + self.inner + .resolve_or_allocate(game, kind, core_id, base_floor) + } + fn external_for(&self, game: &str, kind: &str, core_id: &str) -> Result, IdError> { + self.inner.external_for(game, kind, core_id) + } + fn core_for( + &self, + game: &str, + kind: &str, + external_id: i64, + ) -> Result, IdError> { + self.inner.core_for(game, kind, external_id) + } +} + +// ───────────────────────────── Core boot (seeded) ─────────────────────────── + +async fn start_core_seeded(db_url: &str, seed: bool) -> (tokio::task::JoinHandle<()>, String) { + use openfut_core::config::Config; + use openfut_core::seed::{seed_fifa17_dev, FIFA17_GAME}; + use openfut_core::services::card_db::CardDb; + + let data_dir = "../openfut-core/data"; + let pool = openfut_core::db::init_pool(db_url, 5) + .await + .expect("core pool"); + openfut_core::db::run_migrations(&pool) + .await + .expect("core migrations"); + if seed { + let mut card_db = CardDb::load(data_dir).expect("card_db load"); + card_db + .load_game_dev(data_dir, FIFA17_GAME) + .expect("load fifa17 dev content"); + seed_fifa17_dev(&pool, &card_db) + .await + .expect("seed fifa17 dev inventory"); + } + let cfg = Config { + listen_addr: "127.0.0.1:0".into(), + database_url: "sqlite::memory:".into(), + data_dir: data_dir.into(), + max_connections: 5, + dev_content_games: vec![FIFA17_GAME.to_string()], + content_packs: Vec::new(), + }; + let app = openfut_core::app::build(pool, cfg) + .await + .expect("core app::build"); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral"); + let addr = listener.local_addr().unwrap(); + let handle = tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + (handle, format!("http://{addr}")) +} + +fn wait_ready(base: &str) { + let http = reqwest::blocking::Client::new(); + for _ in 0..1500 { + if let Ok(r) = http.get(format!("{base}/health")).send() { + if r.status().is_success() { + return; + } + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } + panic!("core did not become ready at {base}"); +} + +// ───────────────────────────── Harness ────────────────────────────────────── + +struct FailHarness { + server: Server, + client: HttpCoreClient, + resolver: Arc, + ident: Arc, + fault_ident: Arc, + econ: Arc, + market: Arc, + piles: Arc, + bridge: Arc, + core: Arc, + entities: Arc, + sample_resource: i64, +} + +fn catalog_from_core(core: &dyn CoreAccess) -> Fifa17CardCatalog { + let owned = core.all_owned().expect("core collection"); + assert!(!owned.is_empty(), "seed must grant a starter collection"); + let mut entries = String::new(); + let mut seen = HashSet::new(); + let mut asset = 20000u32; + for it in &owned { + if !seen.insert(it.card_id.clone()) { + continue; + } + if !entries.is_empty() { + entries.push(','); + } + entries.push_str(&format!("\"{}\":{{\"asset_id\":{asset}}}", it.card_id)); + asset += 1; + } + let doc = format!("{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{entries}}}}}"); + Fifa17CardCatalog::from_json_str(&doc).expect("catalog") +} + +fn build_fail_harness(base: &str, dir: &std::path::Path) -> FailHarness { + let core: Arc = Arc::new(HttpCoreClient::new(base, GAME)); + let catalog = catalog_from_core(core.as_ref()); + let ident = Arc::new( + JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).expect("identity"), + ); + let fault_ident = Arc::new(FaultIdentity::new(ident.clone())); + let dyn_store: Arc = fault_ident.clone(); + let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, dyn_store)); + let entities = Arc::new(Fifa17Entities::from_maps( + HashMap::new(), + HashMap::new(), + HashMap::new(), + )); + let bridge = Arc::new(AsyncBridge::new().unwrap()); + let market_path = dir.join("market.db").to_string_lossy().into_owned(); + let market = Arc::new( + bridge + .block_on(async move { MarketStore::open(&market_path).await }) + .expect("market store"), + ); + let pile_path = dir.join("pile.db").to_string_lossy().into_owned(); + let piles = Arc::new( + bridge + .block_on(async move { PileStore::open(&pile_path).await }) + .expect("pile store"), + ); + let econ = Arc::new(FaultEconomy::new(base)); + let econ_dyn: Arc = econ.clone(); + let pool = Arc::new(build_content_pool(core.as_ref(), resolver.as_ref())); + assert!( + !pool.is_empty(), + "content pool derived from real Core content" + ); + let services = Arc::new(EconomyServices { + econ: econ_dyn, + market: market.clone(), + piles: piles.clone(), + bridge: bridge.clone(), + pool, + }); + let server = Server::new( + core.clone(), + entities.clone(), + resolver.clone(), + Arc::new(PassClient::new("http://127.0.0.1:9")), + 33068179, + ) + .with_economy(services); + FailHarness { + server, + client: HttpCoreClient::new(base, GAME), + resolver, + ident, + fault_ident, + econ, + market, + piles, + bridge, + core, + entities, + sample_resource: 20000, + } +} + +impl FailHarness { + /// A sibling `Server` whose content pool is EMPTY (fail-closed generator), + /// sharing the same Core/identity/stores. Used to prove the empty-pool + /// generator path never consumes an entitlement. + fn empty_pool_server(&self) -> Server { + let services = Arc::new(EconomyServices { + econ: { + let e: Arc = self.econ.clone(); + e + }, + market: self.market.clone(), + piles: self.piles.clone(), + bridge: self.bridge.clone(), + pool: Arc::new(Vec::new()), + }); + Server::new( + self.core.clone(), + self.entities.clone(), + self.resolver.clone(), + Arc::new(PassClient::new("http://127.0.0.1:9")), + 33068179, + ) + .with_economy(services) + } + + fn dispatch(&self, method: &str, path: &str, body: &[u8]) -> WireResponse { + self.server + .try_handle_economy(method, path, &[], body, None) + .expect("economy route matched") + } + + fn listing_state(&self, listing_id: &str) -> Option { + let m = self.market.clone(); + let id = listing_id.to_string(); + self.bridge + .block_on(async move { m.get_listing(&id).await }) + .ok() + .map(|l| l.state) + } + + fn owns(&self, core_id: &str) -> bool { + self.client + .all_owned() + .unwrap() + .iter() + .any(|it| it.owned_card_id == core_id) + } + + fn owned_ids(&self) -> HashSet { + self.client + .all_owned() + .unwrap() + .into_iter() + .map(|it| it.owned_card_id) + .collect() + } + + fn unopened_70(&self) -> usize { + self.client + .entitlements() + .unwrap() + .iter() + .filter(|e| e.definition_id == "70") + .count() + } + + /// Mint one owned card via a real Store BUY (pack 1). Returns (wire, core). + fn mint_one(&self) -> (i64, String) { + set_balance(&self.client, 50_000); + let resp = self.dispatch( + "PUT", + "/ut/game/fifa17/store/transaction", + br#"{"packId":1}"#, + ); + assert_eq!(resp.status, 200, "mint buy 200"); + let items = bj(&resp)["createPackResponse"]["itemList"] + .as_array() + .expect("itemList") + .clone(); + let wire = items[0]["id"].as_i64().expect("wire id"); + let core = self.resolver.owned_id_for_wire(wire).expect("reverse"); + (wire, core) + } +} + +fn bj(r: &WireResponse) -> Value { + serde_json::from_slice(&r.body).unwrap_or(Value::Null) +} + +fn set_balance(client: &HttpCoreClient, target: i64) { + let cur = client.balance().unwrap(); + if cur > target { + client + .purchase_entitlement(cur - target, "econ-test-drain") + .expect("drain"); + } else if cur < target { + client.grant_reward(target - cur).expect("top up"); + } + assert_eq!(client.balance().unwrap(), target, "balance set"); +} + +// ───────────────────────────── Cases ──────────────────────────────────────── + +/// BUY Core failure → no debit, no grant. +fn case_buy_core_failure(h: &FailHarness) -> String { + set_balance(&h.client, 5_000); + let owned_before = h.owned_ids(); + h.econ.arm("purchase_items", 1); + let resp = h.dispatch( + "PUT", + "/ut/game/fifa17/store/transaction", + br#"{"packId":1}"#, + ); + assert_eq!(resp.status, 503, "BUY Core failure is fail-closed 503"); + assert_eq!( + h.client.balance().unwrap(), + 5_000, + "no debit on BUY failure" + ); + assert_eq!(h.owned_ids(), owned_before, "no grant on BUY failure"); + "BUY core-fail: 503, no debit, no grant".into() +} + +/// OWNED-PACK OPEN Core-redeem failure → entitlement + inventory coherent +/// (all-or-nothing): entitlement NOT consumed, no items granted. +fn case_open_redeem_failure(h: &FailHarness) -> String { + h.client.purchase_entitlement(0, "70").expect("seed 70"); + let owned_before = h.owned_ids(); + assert_eq!(h.unopened_70(), 1, "one unopened 70 before"); + h.econ.arm("redeem_entitlement", 1); + let resp = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#); + assert_eq!(resp.status, 503, "redeem failure is fail-closed 503"); + assert_eq!(h.unopened_70(), 1, "entitlement survives a failed redeem"); + assert_eq!( + h.owned_ids(), + owned_before, + "no items granted on failed redeem" + ); + // Clean up the surviving entitlement so later cases start fresh. + let ok = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#); + assert_eq!( + ok.status, 200, + "the same entitlement opens once Core recovers" + ); + "OPEN redeem-fail: 503, ent survives, no grant; recovers on retry".into() +} + +/// OWNED-PACK OPEN generator failure (empty content pool) → nothing minted, +/// entitlement NOT consumed. +fn case_open_generator_failure(h: &FailHarness) -> String { + h.client.purchase_entitlement(0, "70").expect("seed 70"); + let owned_before = h.owned_ids(); + let empty = h.empty_pool_server(); + let resp = empty + .try_handle_economy( + "POST", + "/ut/game/fifa17/purchased", + &[], + br#"{"packId":70}"#, + None, + ) + .expect("routed"); + assert_eq!(resp.status, 503, "empty-pool generator is fail-closed 503"); + assert_eq!( + h.unopened_70(), + 1, + "entitlement not consumed when generator draws nothing" + ); + assert_eq!( + h.owned_ids(), + owned_before, + "no items minted by empty generator" + ); + // Consume it via the real (populated) pool to reset. + assert_eq!( + h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#) + .status, + 200 + ); + "OPEN generator-fail: 503, ent not consumed, no mint".into() +} + +/// OWNED-PACK OPEN pile-persist failure → Core is coherent (entitlement consumed +/// once, +11 owned); the reveal simply omits the un-piled items (presentation +/// only). Wire-id allocation stays monotonic. +fn case_open_pile_persist_failure(h: &FailHarness) -> String { + h.client.purchase_entitlement(0, "70").expect("seed 70"); + let owned_before = h.owned_ids(); + h.piles.fault().arm("set", 32); // fail every purchased-pile write for this open + let resp = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#); + h.piles.fault().arm("set", 0); // clear any residual armed count for later cases + assert_eq!( + resp.status, 200, + "pile-write failure is non-fatal to the open" + ); + assert_eq!(h.unopened_70(), 0, "entitlement consumed exactly once"); + let new: Vec = h.owned_ids().difference(&owned_before).cloned().collect(); + assert_eq!( + new.len(), + 11, + "inventory granted once (11 cards) despite pile failure" + ); + // Reveal is presentation-only: the un-piled new items are simply not shown. + let reveal = h.dispatch("GET", "/ut/game/fifa17/purchased", b""); + let shown_new = bj(&reveal)["itemData"] + .as_array() + .map(|a| { + a.iter() + .filter(|x| { + x["id"] + .as_i64() + .and_then(|w| h.resolver.owned_id_for_wire(w)) + .map(|c| new.contains(&c)) + .unwrap_or(false) + }) + .count() + }) + .unwrap_or(0); + assert_eq!( + shown_new, 0, + "un-piled items are omitted from the reveal (coherent)" + ); + let next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR); + assert!(next > OWNED_FLOOR, "watermark advanced monotonically"); + "OPEN pile-fail: 200, ent consumed once, +11 owned, reveal omits un-piled (coherent)".into() +} + +/// OWNED-PACK OPEN identity-alloc failure → Core coherent (consumed + 11), the +/// reveal omits the not-yet-resolvable items; on recovery every item resolves to +/// a UNIQUE wire id above the floor (a burned id is never reused — monotonic). +fn case_open_identity_failure(h: &FailHarness) -> String { + h.client.purchase_entitlement(0, "70").expect("seed 70"); + let owned_before = h.owned_ids(); + let before_next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR); + h.fault_ident.arm_alloc(64); // fail every wire-id allocation for this open + let resp = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#); + assert_eq!( + resp.status, 200, + "identity failure is non-fatal to the redeem" + ); + assert_eq!(h.unopened_70(), 0, "entitlement consumed exactly once"); + let new: Vec = h.owned_ids().difference(&owned_before).cloned().collect(); + assert_eq!(new.len(), 11, "inventory granted once (11 cards)"); + // Recovery: disarm identity, then resolve each new item DIRECTLY (no reliance + // on the pile/reveal path). A previously-faulted allocation left nothing + // persisted, so each now receives a fresh monotonic wire id. + h.fault_ident.arm_alloc(0); + let owned_now = h.client.all_owned().unwrap(); + let mut wires = HashSet::new(); + for item in owned_now + .iter() + .filter(|it| new.contains(&it.owned_card_id)) + { + let id = h.resolver.resolve(item).expect("resolves after recovery"); + let w = id.item_id as i64; + assert!(w >= OWNED_FLOOR, "wire id ≥ floor"); + assert!(wires.insert(w), "each recovered wire id is unique"); + } + let after_next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR); + assert!( + after_next > before_next, + "watermark strictly advanced: no id reused ({before_next} -> {after_next})" + ); + assert!( + wires.iter().all(|&w| w < after_next), + "every issued id is below the next watermark (monotonic)" + ); + "OPEN identity-fail: 200, ent consumed once, +11 owned; recovered ids unique + monotonic".into() +} + +/// QUICK-SELL Core failure → item remains, coins unchanged. +fn case_quicksell_core_failure(h: &FailHarness) -> String { + let (wire, core) = h.mint_one(); + let before = h.client.balance().unwrap(); + h.econ.arm("sell_item", 1); + let resp = h.dispatch("DELETE", &format!("/ut/game/fifa17/item/{wire}"), b""); + assert_eq!( + resp.status, 503, + "quick-sell Core failure is fail-closed 503" + ); + assert!(h.owns(&core), "item still owned after failed quick-sell"); + assert_eq!( + h.client.balance().unwrap(), + before, + "coins unchanged after failed quick-sell" + ); + "QUICK-SELL core-fail: 503, item remains, coins unchanged".into() +} + +/// MOVE pile-store write failure → Core ownership and the pile do not silently +/// disagree: Core still owns the item (move never touches Core), and the pile is +/// simply not updated (verdict success=false), so reveal/pile stay coherent. +fn case_move_pile_failure(h: &FailHarness) -> String { + let (wire, core) = h.mint_one(); + h.piles.fault().arm("set", 1); + let resp = h.dispatch( + "PUT", + "/ut/game/fifa17/item", + format!(r#"{{"itemData":[{{"id":{wire},"pile":"trade"}}]}}"#).as_bytes(), + ); + assert_eq!( + resp.status, 200, + "move responds 200 with a per-item verdict" + ); + assert_eq!( + bj(&resp)["itemData"][0]["success"], + false, + "pile write failure surfaces as success=false (never a fabricated move)" + ); + assert!( + h.owns(&core), + "Core still owns the item (move is not an ownership op)" + ); + let pile = h + .bridge + .block_on({ + let p = h.piles.clone(); + let c = core.clone(); + async move { p.get(&c).await } + }) + .unwrap(); + assert_ne!( + pile.as_deref(), + Some("trade"), + "pile was NOT updated (no silent divergence)" + ); + "MOVE pile-fail: success=false, Core owns, pile not updated (coherent)".into() +} + +/// MARKET RESERVE failure → no debit, no grant, listing stays legal (active). +fn case_market_reserve_failure(h: &FailHarness) -> String { + set_balance(&h.client, 50_000); + let item_id = 700_001i64; + let list = h.dispatch( + "POST", + "/ut/game/fifa17/auctionhouse", + format!( + r#"{{"itemData":{{"id":{item_id},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#, + h.sample_resource + ) + .as_bytes(), + ); + let trade_id = bj(&list)["id"].as_i64().expect("trade id"); + let before = h.client.balance().unwrap(); + h.market.fault().arm("reserve", 1); + let resp = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}"); + assert_eq!(resp.status, 503, "reserve infra failure is fail-closed 503"); + assert_eq!( + h.client.balance().unwrap(), + before, + "no debit when reserve fails" + ); + assert!( + !h.owns(&format!("market-buy:{trade_id}")), + "no card minted when reserve fails" + ); + assert_eq!( + h.listing_state(&trade_id.to_string()).as_deref(), + Some("active"), + "listing stays active (legal) after a reserve failure" + ); + "MARKET reserve-fail: 503, no debit, no mint, listing active".into() +} + +/// MARKET Core purchase_item failure AFTER reserve → reservation rolls back to +/// active, no debit, no mint. +fn case_market_purchase_failure(h: &FailHarness) -> String { + set_balance(&h.client, 50_000); + let item_id = 700_002i64; + let list = h.dispatch( + "POST", + "/ut/game/fifa17/auctionhouse", + format!( + r#"{{"itemData":{{"id":{item_id},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#, + h.sample_resource + ) + .as_bytes(), + ); + let trade_id = bj(&list)["id"].as_i64().expect("trade id"); + let before = h.client.balance().unwrap(); + h.econ.arm("purchase_item", 1); + let resp = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}"); + assert_eq!( + resp.status, 503, + "purchase failure after reserve is fail-closed 503" + ); + assert_eq!( + h.client.balance().unwrap(), + before, + "no debit when Core purchase fails" + ); + assert!( + !h.owns(&format!("market-buy:{trade_id}")), + "no card minted when Core purchase fails" + ); + assert_eq!( + h.listing_state(&trade_id.to_string()).as_deref(), + Some("active"), + "reservation rolled back to active (buyable again)" + ); + "MARKET purchase-fail: 503, reservation rolled back to active, no debit, no mint".into() +} + +/// THE critical case. MARKET COMPLETE-SALE failure AFTER a SUCCESSFUL Core +/// purchase must NOT leave the listing buyable with the buyer already debited + +/// minted. Expected (current design): the listing is stuck in `reserved` (NOT +/// active), so no further `active -> reserved` CAS can succeed → not buyable, +/// with exactly one debit + one mint. Returns ("SAFE"|"E3", detail). +fn case_market_complete_sale_failure(h: &FailHarness) -> (String, String) { + set_balance(&h.client, 50_000); + let item_id = 700_003i64; + let list = h.dispatch( + "POST", + "/ut/game/fifa17/auctionhouse", + format!( + r#"{{"itemData":{{"id":{item_id},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#, + h.sample_resource + ) + .as_bytes(), + ); + let trade_id = bj(&list)["id"].as_i64().expect("trade id"); + let mint_id = format!("market-buy:{trade_id}"); + let before = h.client.balance().unwrap(); + + // Fail the complete_sale that runs AFTER Core has debited + minted. + h.market.fault().arm("complete_sale", 1); + let resp = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}"); + assert_eq!( + resp.status, 200, + "purchase committed in Core (complete_sale is post-commit)" + ); + + // Core-side commit really happened: exactly one debit + one mint. + assert_eq!( + h.client.balance().unwrap(), + before - 1000, + "exactly one debit" + ); + let mints = h + .client + .all_owned() + .unwrap() + .iter() + .filter(|it| it.owned_card_id == mint_id) + .count(); + assert_eq!(mints, 1, "exactly one mint"); + + let state = h.listing_state(&trade_id.to_string()); + // Second buy attempt AFTER the committed purchase. + let after = h.client.balance().unwrap(); + let retry = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}"); + let buyable_again = bj(&retry)["auctionInfo"] + .as_array() + .map(|a| !a.is_empty()) + .unwrap_or(false); + let debited_again = h.client.balance().unwrap() != after; + let minted_again = h + .client + .all_owned() + .unwrap() + .iter() + .filter(|it| it.owned_card_id == mint_id) + .count() + > 1; + + if buyable_again || debited_again || minted_again { + return ( + "E3".into(), + format!( + "ATOMICITY DEFECT: after a committed purchase, listing state={state:?}, \ + buyable_again={buyable_again} debited_again={debited_again} minted_again={minted_again}" + ), + ); + } + assert_eq!( + state.as_deref(), + Some("reserved"), + "committed-but-uncompleted listing is left in `reserved`, not `active`" + ); + ( + "SAFE".into(), + "complete_sale failure leaves listing=reserved (not buyable); exactly one \ + debit + one mint; retry returned empty, no further debit/mint" + .to_string(), + ) +} + +fn run_all_failures(base: &str, dir: &std::path::Path) -> (String, String, bool) { + wait_ready(base); + let h = build_fail_harness(base, dir); + let mut lines = vec![ + case_buy_core_failure(&h), + case_open_redeem_failure(&h), + case_open_generator_failure(&h), + case_open_pile_persist_failure(&h), + case_open_identity_failure(&h), + case_quicksell_core_failure(&h), + case_move_pile_failure(&h), + case_market_reserve_failure(&h), + case_market_purchase_failure(&h), + ]; + let (verdict, detail) = case_market_complete_sale_failure(&h); + lines.push(format!("MARKET complete-sale-fail [{verdict}]: {detail}")); + let e3 = verdict == "E3"; + (lines.join("\n"), verdict, e3) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn economy_failure_injection() { + let dir = std::env::temp_dir().join(format!( + "openfut-econ-fail-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + let db_url = format!("sqlite://{}/econ.db", dir.display()); + + let (h1, base1) = start_core_seeded(&db_url, true).await; + let (b1, d1) = (base1.clone(), dir.clone()); + let (summary, verdict, e3) = tokio::task::spawn_blocking(move || { + std::thread::spawn(move || run_all_failures(&b1, &d1)) + .join() + .expect("failure thread") + }) + .await + .expect("failures"); + h1.abort(); + std::fs::remove_dir_all(&dir).ok(); + + eprintln!("economy_failure summary (complete-sale verdict={verdict}, e3={e3}):\n{summary}"); + assert!(!e3, "unrecoverable partial-state (E3) detected: {summary}"); +}