test(fifa17): prove host economy concurrency and failure rollback

Two real host-dispatch test files (no fakes) driving Server::try_handle_economy
against a live in-process Core over the real blocking client + durable
MarketStore/PileStore + JsonIdentityStore, each racer its own OS thread
(off-runtime pattern).

economy_concurrency.rs — 8 races x 50 iterations:
  A two BUYs (coins for one) -> exactly one 200 + one 461, final 0, one debit.
  B duplicate owned-pack open -> one redemption, +11 once, entitlement once.
  C duplicate quick-sell -> one sell + one credit + one removal.
  D two market buyers -> one win, one debit, one mint, sold once.
  E reward+BUY -> no lost update (Core relative UPDATE under BEGIN IMMEDIATE).
  F move+quick-sell / G list+quick-sell -> one coherent transition.
  H 1000 concurrent mints -> unique + reversible wire ids, monotonic watermark.

economy_failure.rs — 10 fault-injection sub-cases, all fail-closed:
  BUY/open-redeem/generator/pile/identity, quick-sell, move, market
  reserve/purchase/complete. CRITICAL complete-sale-after-commit = SAFE: the
  listing is left `reserved` (not active), so the active->reserved reserve CAS
  can never win again -> not buyable, exactly one debit + one mint. No E3.

Fault injection uses test-file CoreEconomy/ExternalIdentityStore doubles plus a
NARROW, inert-by-default `StoreFault` seam in market_store.rs + pile_store.rs
(the concrete stores have no trait boundary; 3 `tripped()` checks + a field,
zero behaviour unless a test arms it). `parking_lot` promoted to a normal dep
(the seam's Mutex is used at lib scope). Classifier/ROUTE_AUTHORITY/Python
untouched. host lib 71/71; both new tests pass.

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<StoreFaultInner>,
+}
+
+#[derive(Default)]
+struct StoreFaultInner {
+    any: AtomicBool,
+    armed: Mutex<HashMap<&'static str, u32>>,
+}
+
+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<bool, MarketError> {
+        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 {

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")