feat(fifa17): wire async economy handlers into the host via a runtime bridge (unrouted)
Bridges the synchronous thread-per-connection host to the async
transfer-market/pile handlers WITHOUT flipping the classifier. classify()
is untouched; production still proxies every economy route to Python. The
new dispatch is exercised only by the integration harness via
Server::try_handle_economy — handler wiring, not authority cutover.
async_bridge.rs: AsyncBridge owns ONE process-lifetime multi-threaded Tokio
runtime, shared by every connection via Arc. block_on() runs a future from
the sync dispatch thread; if invoked from within an ambient runtime it
offloads onto its own runtime + a std channel instead of panicking
("cannot start a runtime from within a runtime"). 4 unit tests incl. the
nested-runtime-safety case and concurrent multi-thread drivers.
lib.rs: EconomyRoute + classify_economy (mirrors the Python route table:
credits, purchasegroup, store/transaction, purchased, item DELETE/PUT,
ut/delete match/item/trade, auctionhouse/transfermarket, tradePile, trade).
EconomyServices (Core econ transport + durable MarketStore/PileStore + the
bridge + the pack-content pool), attached via Server::with_economy (kept out
of `new`/`from_config` so existing tests build a DB-less Server; production
from_config attachment is the barrier step). Server::try_handle_economy
dispatches: sync handlers (credits/purchasegroup/store-buy/pack-open/
quick-sell/match) inline; async handlers (market list/query/buy/cancel,
move) on the bridge via owned `async move` blocks. build_content_pool
derives the resolvable FIFA∩Core candidate pool from Core content.
market.rs: FIX the load-bearing hazard the FakeEconomy tests missed — the
async market handlers call the BLOCKING reqwest Core client, which panics
(reqwest::blocking::wait::enter) when run while a Tokio runtime is entered.
off_runtime() hops each Core call to a fresh OS thread with no runtime
entered, so blocking is legal. handle_move_items resolver gains `+ Sync`
(future must be Send for the bridge).
tests/economy_integration.rs: economy_full_sequence_through_dispatch drives
the WHOLE cluster through the REAL Server dispatch + bridge against a live
in-process Core (seeded with fifa17 dev content: 100k coins + owned cards),
on a plain OS thread (direct bridge path), over the real blocking
HttpCoreClient — no fakes: Store BUY (pool draw + shape + debit 400 + mint 5),
credits, quick-sell (reverse-resolve + credit), match WIN (+400), market
list->query->buy->query(sold)->second-buy-fails(no double debit), cancel
(cancelled not buyable), move-items, then reopen the durable market/pile
stores from disk (sold + pile persist). Deterministic. start_core_seeded
loads fifa17 dev content so /collection renders real definitions.
Tests: host 68 lib (+4 bridge) + 2 economy_integration + 24 host_test, all
green; adapter unchanged-green; clippy -D warnings + fmt clean.
This commit is contained in:
OpenFUT Agent
@@ -23,6 +23,10 @@ sqlx = { version = "0.7", features = ["sqlite", "runtime-tokio"] }
|
||||
# writer handlers (`economy_store`). Production seeds it from entropy (so minted
|
||||
# Core instance ids never collide); tests seed a fixed value for reproducibility.
|
||||
rand = "0.8"
|
||||
# One process-lifetime runtime drives the async transfer-market / pile handlers
|
||||
# (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"] }
|
||||
|
||||
[dev-dependencies]
|
||||
parking_lot = "0.12"
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
//! Sync → async execution bridge for the economy handlers.
|
||||
//!
|
||||
//! The UTAS host is a **synchronous, thread-per-connection** server
|
||||
//! ([`crate::Server::serve_listener`] spawns one `std::thread` per socket and
|
||||
//! [`crate::Server::handle_with_ip`] runs to completion on that plain OS
|
||||
//! thread). Most handlers — `/club`, squad, credits, match reward, the Store
|
||||
//! writers — are fully synchronous (Core is reached over a **blocking** reqwest
|
||||
//! client). Only the durable transfer-market + pile handlers are `async`,
|
||||
//! because their store is `sqlx` with the `runtime-tokio` feature.
|
||||
//!
|
||||
//! Rather than sprinkle per-request runtimes, the host owns ONE long-lived
|
||||
//! multi-threaded Tokio runtime for the whole process lifetime and drives the
|
||||
//! few async handlers on it with [`AsyncBridge::block_on`].
|
||||
//!
|
||||
//! ## Execution contract
|
||||
//!
|
||||
//! * Exactly one runtime exists for the host lifetime; it is created once
|
||||
//! (`Server::from_config`) and shared by every connection thread via `Arc`.
|
||||
//! * `block_on` is called from the synchronous dispatch thread, which is NOT a
|
||||
//! Tokio worker, so the common path is a direct [`tokio::runtime::Runtime::block_on`].
|
||||
//! * **Nested-runtime safety:** should dispatch ever be invoked from inside a
|
||||
//! Tokio context (it is not in production, but a future async caller or a test
|
||||
//! might), calling `block_on`/`Handle::block_on` on the current thread would
|
||||
//! panic (*"Cannot start a runtime from within a runtime"*). We detect an
|
||||
//! ambient runtime with [`tokio::runtime::Handle::try_current`] and, in that
|
||||
//! case, offload the future onto our own runtime and block the caller on a
|
||||
//! std channel — no runtime is entered on the current thread, so it cannot
|
||||
//! panic.
|
||||
//! * The blocking Core client used *inside* the async handlers is safe here: it
|
||||
//! runs its own private runtime on a dedicated background thread, so calling
|
||||
//! it from a runtime worker parks the worker on a channel rather than entering
|
||||
//! a runtime. The host client is process-lifetime (`Arc`) and never dropped in
|
||||
//! an async context.
|
||||
|
||||
use std::future::Future;
|
||||
|
||||
use tokio::runtime::{Builder, Handle, Runtime};
|
||||
|
||||
/// A process-lifetime Tokio runtime plus a nested-safe blocking bridge.
|
||||
pub struct AsyncBridge {
|
||||
rt: Runtime,
|
||||
}
|
||||
|
||||
impl AsyncBridge {
|
||||
/// Build the shared multi-threaded runtime. Multi-threaded (not
|
||||
/// current-thread) so that a handler which parks a worker on the blocking
|
||||
/// Core client still leaves other workers to drive `sqlx` IO to completion.
|
||||
pub fn new() -> std::io::Result<Self> {
|
||||
let rt = Builder::new_multi_thread()
|
||||
.worker_threads(4)
|
||||
.enable_all()
|
||||
.thread_name("openfut-econ")
|
||||
.build()?;
|
||||
Ok(Self { rt })
|
||||
}
|
||||
|
||||
/// A cloneable handle to the shared runtime (used by callers that want to
|
||||
/// `spawn` onto the same executor rather than block).
|
||||
pub fn handle(&self) -> Handle {
|
||||
self.rt.handle().clone()
|
||||
}
|
||||
|
||||
/// Drive `fut` to completion from a synchronous caller, returning its output.
|
||||
///
|
||||
/// The `Send + 'static` bound lets the nested-safety fallback move the future
|
||||
/// onto our runtime; dispatch satisfies it by building `async move` blocks
|
||||
/// that own `Arc` clones of the shared services (never per-request state).
|
||||
pub fn block_on<F>(&self, fut: F) -> F::Output
|
||||
where
|
||||
F: Future + Send + 'static,
|
||||
F::Output: Send + 'static,
|
||||
{
|
||||
match Handle::try_current() {
|
||||
// Common path: a plain OS dispatch thread, no ambient runtime.
|
||||
Err(_) => self.rt.block_on(fut),
|
||||
// Ambient runtime present: driving on the current thread would panic.
|
||||
// Offload to our runtime and block on a std channel — the current
|
||||
// thread never enters a runtime, so it is panic-free.
|
||||
Ok(_) => {
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel(1);
|
||||
self.rt.spawn(async move {
|
||||
let _ = tx.send(fut.await);
|
||||
});
|
||||
rx.recv().expect("AsyncBridge offloaded task panicked")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn block_on_from_plain_thread_runs_future() {
|
||||
let bridge = AsyncBridge::new().unwrap();
|
||||
let out = bridge.block_on(async { 2 + 3 });
|
||||
assert_eq!(out, 5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_from_plain_thread_drives_async_io() {
|
||||
// A future with an actual await point (timer) completes on our runtime.
|
||||
let bridge = AsyncBridge::new().unwrap();
|
||||
let out = bridge.block_on(async {
|
||||
tokio::task::yield_now().await;
|
||||
41 + 1
|
||||
});
|
||||
assert_eq!(out, 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_is_nested_runtime_safe() {
|
||||
// Simulate the hazardous case: dispatch invoked from WITHIN a Tokio
|
||||
// context. A naive `Runtime::block_on` here panics; the bridge must not.
|
||||
let bridge = AsyncBridge::new().unwrap();
|
||||
let ambient = Builder::new_multi_thread()
|
||||
.worker_threads(2)
|
||||
.enable_all()
|
||||
.build()
|
||||
.unwrap();
|
||||
let out = ambient.block_on(async {
|
||||
// We are now inside `ambient`. Calling the bridge must offload, not
|
||||
// panic with "Cannot start a runtime from within a runtime".
|
||||
bridge.block_on(async {
|
||||
tokio::task::yield_now().await;
|
||||
7 * 6
|
||||
})
|
||||
});
|
||||
assert_eq!(out, 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn block_on_concurrently_from_many_threads() {
|
||||
// The shared runtime is driven by many dispatch threads at once, exactly
|
||||
// as thread-per-connection dispatch does.
|
||||
let bridge = std::sync::Arc::new(AsyncBridge::new().unwrap());
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..16i64 {
|
||||
let b = bridge.clone();
|
||||
handles.push(std::thread::spawn(move || {
|
||||
b.block_on(async move {
|
||||
tokio::task::yield_now().await;
|
||||
i * 2
|
||||
})
|
||||
}));
|
||||
}
|
||||
let sum: i64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
|
||||
assert_eq!(sum, (0..16i64).map(|i| i * 2).sum::<i64>());
|
||||
}
|
||||
}
|
||||
@@ -34,6 +34,7 @@
|
||||
//! faked (see `club_response`). With today's empty mapping, `/club` returns
|
||||
//! `{"itemData":[]}` — the honest state until Core inventory is asset-backed.
|
||||
|
||||
pub mod async_bridge;
|
||||
pub mod config;
|
||||
pub mod economy_store;
|
||||
pub mod market;
|
||||
@@ -56,6 +57,7 @@ use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
|
||||
use openfut_adapter_fifa17::fut::owned_query::{
|
||||
is_special_rareflag, map_to_core, parse_club_query, MapError,
|
||||
};
|
||||
use openfut_adapter_fifa17::fut::pack_content::GeneratedCandidate;
|
||||
use openfut_adapter_fifa17::fut::squad::{parse_squad_put, save_ack, SquadWireResolver};
|
||||
use openfut_adapter_fifa17::fut::squad_ext::{
|
||||
build_squad_write, Fifa17SquadExtensionV1, SquadBuildError, EXT_NAMESPACE, EXT_SCHEMA_VERSION,
|
||||
@@ -69,6 +71,7 @@ use openfut_adapter_fifa17::fut::store_session::{
|
||||
validate_capability, SessionStore, StoreMode, SENTINEL_PACK_ID,
|
||||
};
|
||||
use openfut_identity::ExternalIdentityStore;
|
||||
use rand::SeedableRng;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use config::HostConfig;
|
||||
@@ -162,6 +165,87 @@ fn is_numeric_squad_tail(tail: &str) -> bool {
|
||||
}
|
||||
}
|
||||
|
||||
/// A FIFA17 economy route, classified separately from [`classify`]. This is the
|
||||
/// *target* ownership map for the economy cutover. It is deliberately NOT wired
|
||||
/// into [`Server::handle_with_ip`] yet: handler wiring and authority cutover are
|
||||
/// distinct steps. Until the single barrier commit flips the whole cluster,
|
||||
/// production classification ([`classify`]) still sends every one of these to
|
||||
/// Python; only integration tests drive them through [`Server::try_handle_economy`].
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum EconomyRoute {
|
||||
/// `GET …/user/credits` — coins + unopened-pack count.
|
||||
Credits,
|
||||
/// `GET …/store/purchasegroup` — catalogue + owned packs, full-generated.
|
||||
PurchaseGroup,
|
||||
/// `PUT …/store/transaction` — Store BUY (open-on-buy).
|
||||
StoreBuy,
|
||||
/// `POST …/purchased` — open a pack / redeem an owned entitlement.
|
||||
PackOpen,
|
||||
/// `DELETE …/item/<id>` — single-card quick-sell.
|
||||
QuickSellPath,
|
||||
/// `POST /ut/delete/game/<sku>/item` — bulk quick-sell.
|
||||
QuickSellBody,
|
||||
/// `PUT …/item` — FutMoveCard pile move.
|
||||
MoveItems,
|
||||
/// `POST /ut/delete/game/<sku>/match` — match END (the coin-crediting call).
|
||||
MatchEnd,
|
||||
/// `…/auctionhouse` | `…/transfermarket` — list-for-sale / browse.
|
||||
MarketList,
|
||||
/// `GET …/tradePile` — the user's own active listings.
|
||||
MarketQuery,
|
||||
/// `…/trade/<id>` — view / buy-now.
|
||||
MarketBuy,
|
||||
/// `DELETE /ut/delete/game/<sku>/trade/<id>` — cancel a listing.
|
||||
MarketCancel,
|
||||
}
|
||||
|
||||
/// `item/<digits>` — the single-card quick-sell tail (DELETE).
|
||||
fn is_item_id_tail(tail: &str) -> bool {
|
||||
match tail.strip_prefix("item/") {
|
||||
Some(id) => !id.is_empty() && id.bytes().all(|b| b.is_ascii_digit()),
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Classify a FIFA17 economy route from method + path, mirroring the Python
|
||||
/// oracle's route table (`utas_server.py` §1418-1553). Returns `None` for any
|
||||
/// non-economy path. Path is already query-stripped by the caller.
|
||||
pub fn classify_economy(method: &str, path: &str) -> Option<EconomyRoute> {
|
||||
let get = method.eq_ignore_ascii_case("GET");
|
||||
let put = method.eq_ignore_ascii_case("PUT");
|
||||
let post = method.eq_ignore_ascii_case("POST");
|
||||
let delete = method.eq_ignore_ascii_case("DELETE");
|
||||
|
||||
// The `/ut/delete/game/<sku>/…` family is NOT `/ut/game/…`-prefixed.
|
||||
if let Some(rest) = path.strip_prefix("/ut/delete/game/") {
|
||||
if let Some((_sku, tail)) = rest.split_once('/') {
|
||||
if tail == "item" && post {
|
||||
return Some(EconomyRoute::QuickSellBody);
|
||||
}
|
||||
if tail.starts_with("trade") && delete {
|
||||
return Some(EconomyRoute::MarketCancel);
|
||||
}
|
||||
if tail == "match" && post {
|
||||
return Some(EconomyRoute::MatchEnd);
|
||||
}
|
||||
}
|
||||
return None;
|
||||
}
|
||||
|
||||
match ut_tail(path) {
|
||||
Some("user/credits") if get => Some(EconomyRoute::Credits),
|
||||
Some(t) if get && t.starts_with("store/purchasegroup") => Some(EconomyRoute::PurchaseGroup),
|
||||
Some("store/transaction") if put => Some(EconomyRoute::StoreBuy),
|
||||
Some("purchased") if post => Some(EconomyRoute::PackOpen),
|
||||
Some(t) if delete && is_item_id_tail(t) => Some(EconomyRoute::QuickSellPath),
|
||||
Some("item") if put => Some(EconomyRoute::MoveItems),
|
||||
Some(t) if (t == "auctionhouse" || t == "transfermarket") => Some(EconomyRoute::MarketList),
|
||||
Some("tradePile") if get => Some(EconomyRoute::MarketQuery),
|
||||
Some(t) if t.starts_with("trade") => Some(EconomyRoute::MarketBuy),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────────── Core access boundary ─────────────────────────
|
||||
|
||||
/// Failure reaching or reading OpenFUT Core.
|
||||
@@ -1594,6 +1678,61 @@ impl PassClient {
|
||||
|
||||
// ───────────────────────────── Server ───────────────────────────────────────
|
||||
|
||||
/// The durable services the FIFA17 economy handlers need, wired into [`Server`]
|
||||
/// ONCE at construction (never per request). All are process-lifetime `Arc`s:
|
||||
/// the Core economy transport, the two host-owned durable SQLite stores (listing
|
||||
/// and pile), the shared Tokio runtime bridge, and the pack-content candidate
|
||||
/// pool (the resolvable FIFA∩Core card universe).
|
||||
#[derive(Clone)]
|
||||
pub struct EconomyServices {
|
||||
pub econ: Arc<dyn CoreEconomy>,
|
||||
pub market: Arc<crate::market_store::MarketStore>,
|
||||
pub piles: Arc<crate::pile_store::PileStore>,
|
||||
pub bridge: Arc<crate::async_bridge::AsyncBridge>,
|
||||
/// The resolvable FIFA∩Core card universe a pack can award (empty → the Store
|
||||
/// fail-closes: it draws nothing and debits nothing).
|
||||
pub pool: Arc<Vec<GeneratedCandidate>>,
|
||||
}
|
||||
|
||||
/// Build the pack-content candidate pool from Core's current content, evidenced
|
||||
/// by the owned inventory: every distinct owned card definition that resolves to
|
||||
/// a real FIFA asset id is a candidate (`gold` = rating ≥ 75; `special` from the
|
||||
/// catalog `rareflag > 1`). This is the resolvable FIFA∩Core card universe — the
|
||||
/// cards a pack can award and the shared shaper can render. An empty pool (no
|
||||
/// content, or Core unreachable) is fail-closed by construction: the generator
|
||||
/// returns no cards, so the Store neither mints nor debits.
|
||||
pub fn build_content_pool(
|
||||
core: &dyn CoreAccess,
|
||||
resolver: &Fifa17IdentityResolver,
|
||||
) -> Vec<GeneratedCandidate> {
|
||||
let owned = match core.all_owned() {
|
||||
Ok(v) => v,
|
||||
Err(_) => return Vec::new(),
|
||||
};
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
let mut pool = Vec::new();
|
||||
for item in &owned {
|
||||
if !seen.insert(item.card_id.clone()) {
|
||||
continue;
|
||||
}
|
||||
let Some(id) = resolver.resolve(item) else {
|
||||
continue;
|
||||
};
|
||||
pool.push(GeneratedCandidate {
|
||||
card_id: item.card_id.clone(),
|
||||
rating: item.rating,
|
||||
position: item.position.clone(),
|
||||
nation: item.nation.clone(),
|
||||
league: item.league.clone(),
|
||||
club: item.club.clone(),
|
||||
attributes: item.attributes,
|
||||
gold: item.rating >= 75,
|
||||
special: id.rareflag > 1,
|
||||
});
|
||||
}
|
||||
pool
|
||||
}
|
||||
|
||||
/// The migration host. Cheap to clone (all shared state is `Arc`).
|
||||
#[derive(Clone)]
|
||||
pub struct Server {
|
||||
@@ -1613,6 +1752,11 @@ pub struct Server {
|
||||
sessions: Arc<Mutex<SessionStore>>,
|
||||
/// Monotonic clock origin for the session/pending TTLs.
|
||||
start: Instant,
|
||||
/// FIFA17 economy authority services (Core transport + durable listing/pile
|
||||
/// stores + runtime bridge + content pool). `None` until wired via
|
||||
/// [`Server::with_economy`]; the economy dispatch is inert without it, and
|
||||
/// `handle_with_ip` does not consult it until the classifier barrier.
|
||||
economy: Option<Arc<EconomyServices>>,
|
||||
}
|
||||
|
||||
impl Server {
|
||||
@@ -1632,6 +1776,7 @@ impl Server {
|
||||
persona_id,
|
||||
sessions: Arc::new(Mutex::new(SessionStore::new())),
|
||||
start: Instant::now(),
|
||||
economy: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1658,6 +1803,7 @@ impl Server {
|
||||
persona_id: cfg.persona_id,
|
||||
sessions: Arc::new(Mutex::new(SessionStore::new())),
|
||||
start: Instant::now(),
|
||||
economy: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1671,6 +1817,139 @@ impl Server {
|
||||
}
|
||||
}
|
||||
|
||||
/// Attach the FIFA17 economy authority services. Kept separate from
|
||||
/// construction so the (many) squad/club tests build a `Server` without a
|
||||
/// database, while the economy integration path wires real durable stores +
|
||||
/// the runtime bridge once.
|
||||
pub fn with_economy(mut self, economy: Arc<EconomyServices>) -> Self {
|
||||
self.economy = Some(economy);
|
||||
self
|
||||
}
|
||||
|
||||
/// Dispatch a FIFA17 economy route to its Rust handler, or `None` if the path
|
||||
/// is not an economy route (or no economy services are wired). This is the
|
||||
/// handler-wiring entry point exercised by the integration harness; it is
|
||||
/// deliberately NOT called by `handle_with_ip` yet — handler wiring and the
|
||||
/// authority cutover are distinct steps, and the classifier barrier is one
|
||||
/// later coherent flip. Sync handlers run inline; the async transfer-market /
|
||||
/// move handlers run on the shared runtime via the bridge.
|
||||
pub fn try_handle_economy(
|
||||
&self,
|
||||
method: &str,
|
||||
target: &str,
|
||||
headers: &[(String, String)],
|
||||
body: &[u8],
|
||||
client_ip: Option<&str>,
|
||||
) -> Option<WireResponse> {
|
||||
let svc = self.economy.as_ref()?;
|
||||
let path = target.split('?').next().unwrap_or(target);
|
||||
let route = classify_economy(method, path)?;
|
||||
use crate::economy_store::{
|
||||
handle_pack_open, handle_quick_sell_body, handle_quick_sell_path, handle_store_buy,
|
||||
CoreItemLookup, QuickSellDeps, StoreDeps,
|
||||
};
|
||||
let resp = match route {
|
||||
EconomyRoute::Credits => handle_credits(svc.econ.as_ref()),
|
||||
EconomyRoute::PurchaseGroup => {
|
||||
let sid = header(headers, "x-ut-sid").unwrap_or("");
|
||||
let mode =
|
||||
self.sessions
|
||||
.lock()
|
||||
.unwrap()
|
||||
.empty_mypacks_mode(sid, client_ip, self.now());
|
||||
handle_purchasegroup(svc.econ.as_ref(), mode)
|
||||
}
|
||||
EconomyRoute::StoreBuy => {
|
||||
let mut rng = rand::rngs::StdRng::from_entropy();
|
||||
let deps = StoreDeps {
|
||||
econ: svc.econ.as_ref(),
|
||||
assets: self.resolver.as_ref(),
|
||||
entities: self.entities.as_ref(),
|
||||
pool: svc.pool.as_ref(),
|
||||
};
|
||||
handle_store_buy(body, &deps, &mut rng)
|
||||
}
|
||||
EconomyRoute::PackOpen => {
|
||||
let mut rng = rand::rngs::StdRng::from_entropy();
|
||||
let deps = StoreDeps {
|
||||
econ: svc.econ.as_ref(),
|
||||
assets: self.resolver.as_ref(),
|
||||
entities: self.entities.as_ref(),
|
||||
pool: svc.pool.as_ref(),
|
||||
};
|
||||
handle_pack_open(body, &deps, &mut rng)
|
||||
}
|
||||
EconomyRoute::QuickSellPath => {
|
||||
let id = ut_tail(path)
|
||||
.and_then(|t| t.strip_prefix("item/"))
|
||||
.and_then(|d| d.parse::<i64>().ok())?;
|
||||
let lookup = CoreItemLookup {
|
||||
core: self.core.as_ref(),
|
||||
};
|
||||
let deps = QuickSellDeps {
|
||||
econ: svc.econ.as_ref(),
|
||||
reverse: self.resolver.as_ref(),
|
||||
items: &lookup,
|
||||
};
|
||||
handle_quick_sell_path(id, &deps)
|
||||
}
|
||||
EconomyRoute::QuickSellBody => {
|
||||
let lookup = CoreItemLookup {
|
||||
core: self.core.as_ref(),
|
||||
};
|
||||
let deps = QuickSellDeps {
|
||||
econ: svc.econ.as_ref(),
|
||||
reverse: self.resolver.as_ref(),
|
||||
items: &lookup,
|
||||
};
|
||||
handle_quick_sell_body(body, &deps)
|
||||
}
|
||||
EconomyRoute::MatchEnd => handle_match_end(svc.econ.as_ref(), body),
|
||||
EconomyRoute::MoveItems => {
|
||||
let (bridge, piles, resolver) =
|
||||
(svc.bridge.clone(), svc.piles.clone(), self.resolver.clone());
|
||||
let body = body.to_vec();
|
||||
bridge.block_on(async move {
|
||||
crate::market::handle_move_items(&body, resolver.as_ref(), piles.as_ref()).await
|
||||
})
|
||||
}
|
||||
EconomyRoute::MarketList => {
|
||||
let (bridge, market, econ) =
|
||||
(svc.bridge.clone(), svc.market.clone(), svc.econ.clone());
|
||||
let (m, body) = (method.to_string(), body.to_vec());
|
||||
bridge.block_on(async move {
|
||||
crate::market::handle_market_list(&m, &body, econ.as_ref(), market.as_ref())
|
||||
.await
|
||||
})
|
||||
}
|
||||
EconomyRoute::MarketQuery => {
|
||||
let (bridge, market, econ) =
|
||||
(svc.bridge.clone(), svc.market.clone(), svc.econ.clone());
|
||||
bridge.block_on(async move {
|
||||
crate::market::handle_market_query("active", econ.as_ref(), market.as_ref())
|
||||
.await
|
||||
})
|
||||
}
|
||||
EconomyRoute::MarketBuy => {
|
||||
let (bridge, market, econ) =
|
||||
(svc.bridge.clone(), svc.market.clone(), svc.econ.clone());
|
||||
let (m, p, body) = (method.to_string(), path.to_string(), body.to_vec());
|
||||
bridge.block_on(async move {
|
||||
crate::market::handle_market_buy(&m, &p, &body, econ.as_ref(), market.as_ref())
|
||||
.await
|
||||
})
|
||||
}
|
||||
EconomyRoute::MarketCancel => {
|
||||
let (bridge, market) = (svc.bridge.clone(), svc.market.clone());
|
||||
let (p, owner) = (path.to_string(), client_ip.map(|s| s.to_string()));
|
||||
bridge.block_on(async move {
|
||||
crate::market::handle_market_cancel(&p, owner.as_deref(), market.as_ref()).await
|
||||
})
|
||||
}
|
||||
};
|
||||
Some(resp)
|
||||
}
|
||||
|
||||
/// 4-arg entrypoint (tests + callers without a peer address). Session-bound
|
||||
/// routes fall back to a `None` client IP.
|
||||
pub fn handle(
|
||||
|
||||
@@ -102,11 +102,26 @@ fn auction_record(l: &Listing) -> Value {
|
||||
})
|
||||
}
|
||||
|
||||
/// Run a BLOCKING closure — the blocking Core client — on a fresh OS thread that
|
||||
/// has NO Tokio runtime entered, then block until it finishes. The market
|
||||
/// handlers are `async` and driven on the shared runtime by the host bridge, but
|
||||
/// `CoreEconomy` is a `reqwest::blocking` client: calling it while a runtime is
|
||||
/// entered panics (`reqwest::blocking::wait::enter`). Hopping to a plain thread
|
||||
/// keeps the Core call off the runtime, so blocking is legal. Cheap relative to
|
||||
/// the network round-trip it guards.
|
||||
fn off_runtime<T, F>(f: F) -> T
|
||||
where
|
||||
F: FnOnce() -> T + Send,
|
||||
T: Send,
|
||||
{
|
||||
std::thread::scope(|s| s.spawn(f).join().expect("off-runtime Core call panicked"))
|
||||
}
|
||||
|
||||
/// Current Core balance as the `credits` field, or 0 if Core is unreachable
|
||||
/// (used only to decorate an already-decided response; the transactional path
|
||||
/// never trusts a fabricated balance).
|
||||
fn credits_or_zero(econ: &dyn CoreEconomy) -> i64 {
|
||||
econ.balance().unwrap_or(0)
|
||||
off_runtime(|| econ.balance()).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// `/auctionhouse` — search (GET), list-for-sale (POST FutISStart), relist (PUT).
|
||||
@@ -280,7 +295,7 @@ pub async fn handle_market_buy(
|
||||
|
||||
// Pre-check funds so an affordability failure is a clean 461, distinct from a
|
||||
// Core transport failure (503). Core's purchase is still the atomic authority.
|
||||
let balance = match econ.balance() {
|
||||
let balance = match off_runtime(|| econ.balance()) {
|
||||
Ok(b) => b,
|
||||
Err(_) => {
|
||||
let _ = store.rollback_reservation(&id).await;
|
||||
@@ -298,7 +313,7 @@ pub async fn handle_market_buy(
|
||||
// Mint the won card into the buyer's club. A listing sells at most once (the
|
||||
// CAS guarantees it), so a deterministic minted id is safe and idempotent.
|
||||
let minted_item_id = format!("market-buy:{id}");
|
||||
match econ.purchase_item(price, &minted_item_id, &listing.card_id) {
|
||||
match off_runtime(|| econ.purchase_item(price, &minted_item_id, &listing.card_id)) {
|
||||
Ok(new_balance) => {
|
||||
// Core has taken the coins and minted the item; finalise the listing.
|
||||
// If completing fails, Core is still authoritative — report success.
|
||||
@@ -336,7 +351,7 @@ pub async fn handle_market_buy(
|
||||
/// never a fabricated move.
|
||||
pub async fn handle_move_items(
|
||||
body: &[u8],
|
||||
resolver: &dyn SquadWireResolver,
|
||||
resolver: &(dyn SquadWireResolver + Sync),
|
||||
pile_store: &PileStore,
|
||||
) -> WireResponse {
|
||||
let b = parse_body(body);
|
||||
|
||||
@@ -11,12 +11,21 @@
|
||||
//! Safety: uses only a temp directory + `127.0.0.1:0` ephemeral ports. Never
|
||||
//! touches the production Core DB, production ports/containers, or `.105`.
|
||||
|
||||
use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
|
||||
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
|
||||
use openfut_adapter_fifa17::fut::store_session::StoreMode;
|
||||
use openfut_identity::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::{
|
||||
handle_credits, handle_match_end, handle_purchasegroup, overlay_massinfo_economy, CoreEconomy,
|
||||
HttpCoreClient,
|
||||
build_content_pool, handle_credits, handle_match_end, handle_purchasegroup,
|
||||
overlay_massinfo_economy, CoreAccess, CoreEconomy, EconomyServices, Fifa17IdentityResolver,
|
||||
HttpCoreClient, PassClient, Server,
|
||||
};
|
||||
use serde_json::{json, Value};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Boot a Core instance against `db_url`, serving on an ephemeral port. Returns
|
||||
/// the serve task handle and its base URL.
|
||||
@@ -45,6 +54,53 @@ async fn start_core(db_url: &str) -> (tokio::task::JoinHandle<()>, String) {
|
||||
(handle, format!("http://{addr}"))
|
||||
}
|
||||
|
||||
/// Boot a Core with the FIFA17 dev CONTENT loaded (so `/collection` renders real
|
||||
/// card definitions) and, on first boot, the dev inventory SEEDED (a fifa17
|
||||
/// profile + club with 100k coins + one owned instance per definition). On
|
||||
/// restart pass `seed=false`: content is reloaded but the durable DB is left as
|
||||
/// is, proving persistence rather than re-seeding.
|
||||
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();
|
||||
// Generous ceiling (~30s): returns on first success, so it only ever waits
|
||||
@@ -194,3 +250,390 @@ async fn economy_end_to_end_and_restart_persistence() {
|
||||
|
||||
std::fs::remove_dir_all(&dir).ok();
|
||||
}
|
||||
|
||||
// ─────────────── Full economy sequence through the real host dispatch ────────
|
||||
//
|
||||
// Everything below drives the ACTUAL `Server::try_handle_economy` path (the same
|
||||
// dispatch + async bridge the barrier will route production through), against a
|
||||
// live in-process Core over the real blocking `HttpCoreClient`. No fakes. The
|
||||
// sequence runs on a plain OS thread (no ambient Tokio runtime), exactly like the
|
||||
// thread-per-connection server, so the bridge takes its DIRECT `block_on` path.
|
||||
|
||||
/// Facts captured from the write sequence, re-checked after reopening the stores.
|
||||
struct SeqResult {
|
||||
sold_listing: String,
|
||||
moved_core_id: String,
|
||||
}
|
||||
|
||||
/// Build a real `Server` with economy authority wired: real Core transport, a
|
||||
/// catalog derived from the seeded content (so the pack pool + shaper resolve),
|
||||
/// a persistent identity store, and the two durable SQLite stores opened via the
|
||||
/// bridge. Returns the server, a direct Core client for balance assertions, and
|
||||
/// the identity resolver (to reverse a wire id for the restart pile check).
|
||||
fn build_econ_server(
|
||||
base: &str,
|
||||
dir: &std::path::Path,
|
||||
) -> (Server, HttpCoreClient, Arc<Fifa17IdentityResolver>) {
|
||||
let probe = HttpCoreClient::new(base, "fifa17");
|
||||
let owned = probe.all_owned().expect("core collection");
|
||||
assert!(!owned.is_empty(), "seed must grant a starter collection");
|
||||
// A catalog mapping every seeded definition to a real-looking asset id (in
|
||||
// production this is the shipped FIFA catalog; here it is derived so the E2E
|
||||
// exercises real shaping without a static fixture).
|
||||
let mut entries = String::new();
|
||||
let mut seen = std::collections::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 store = JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).unwrap();
|
||||
let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
|
||||
let entities = Arc::new(Fifa17Entities::from_maps(
|
||||
HashMap::new(),
|
||||
HashMap::new(),
|
||||
HashMap::new(),
|
||||
));
|
||||
let core: Arc<dyn CoreAccess> = Arc::new(HttpCoreClient::new(base, "fifa17"));
|
||||
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<dyn CoreEconomy> = Arc::new(HttpCoreClient::new(base, "fifa17"));
|
||||
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);
|
||||
(server, probe, resolver)
|
||||
}
|
||||
|
||||
/// Drive the whole writer+reader cluster through the real dispatch. Panics on any
|
||||
/// mismatch. Runs on a plain OS thread (no ambient runtime).
|
||||
fn economy_sequence(base: &str, dir: &std::path::Path) -> SeqResult {
|
||||
wait_ready(base);
|
||||
// Core is seeded (start_core_seeded): a fifa17 profile with 100k coins + one
|
||||
// owned instance per definition. No /auth/local — the profile already exists.
|
||||
let (server, client, resolver) = build_econ_server(base, dir);
|
||||
let start = client.balance().unwrap();
|
||||
assert!(start >= 5000, "seeded dev balance present ({start})");
|
||||
|
||||
// 1) Store BUY (pack 1 = Bronze, price 400, 5 cards): debit + mint + reveal.
|
||||
let buy = server
|
||||
.try_handle_economy(
|
||||
"PUT",
|
||||
"/ut/game/fifa17/store/transaction",
|
||||
&[],
|
||||
br#"{"packId":1}"#,
|
||||
None,
|
||||
)
|
||||
.expect("store BUY routed");
|
||||
assert_eq!(buy.status, 200, "BUY 200");
|
||||
let bv: Value = serde_json::from_slice(&buy.body).unwrap();
|
||||
let items = bv["createPackResponse"]["itemList"]
|
||||
.as_array()
|
||||
.expect("itemList")
|
||||
.clone();
|
||||
assert_eq!(items.len(), 5, "pack 1 awards 5 cards");
|
||||
assert_eq!(bv["createPackResponse"]["numberItems"], 5);
|
||||
assert!(
|
||||
items[0]["id"].as_i64().unwrap() >= 100_000_000,
|
||||
"minted wire id above the FIFA floor"
|
||||
);
|
||||
assert_eq!(
|
||||
client.balance().unwrap(),
|
||||
start - 400,
|
||||
"BUY debited exactly 400"
|
||||
);
|
||||
|
||||
// 2) credits reads the SAME Core authority.
|
||||
let cr = server
|
||||
.try_handle_economy("GET", "/ut/game/fifa17/user/credits", &[], b"", None)
|
||||
.unwrap();
|
||||
let crv: Value = serde_json::from_slice(&cr.body).unwrap();
|
||||
assert_eq!(
|
||||
crv["currencies"][0]["funds"],
|
||||
start - 400,
|
||||
"credits == Core balance"
|
||||
);
|
||||
|
||||
// 3) Quick-sell one minted card: reverse-resolve wire -> Core id, credit.
|
||||
let sell_wire = items[0]["id"].as_i64().unwrap();
|
||||
let before = client.balance().unwrap();
|
||||
let qs = server
|
||||
.try_handle_economy(
|
||||
"DELETE",
|
||||
&format!("/ut/game/fifa17/item/{sell_wire}"),
|
||||
&[],
|
||||
b"",
|
||||
None,
|
||||
)
|
||||
.expect("quick-sell routed");
|
||||
assert_eq!(qs.status, 200);
|
||||
let qv: Value = serde_json::from_slice(&qs.body).unwrap();
|
||||
assert_eq!(qv["items"].as_array().unwrap().len(), 1, "sold exactly one");
|
||||
let after_sell = client.balance().unwrap();
|
||||
assert!(
|
||||
after_sell > before,
|
||||
"quick-sell credited ({before}->{after_sell})"
|
||||
);
|
||||
assert_eq!(
|
||||
qv["totalCredits"].as_i64().unwrap(),
|
||||
after_sell,
|
||||
"totalCredits == absolute Core balance"
|
||||
);
|
||||
|
||||
// 4) Match END reward through dispatch (WIN = +400).
|
||||
let before_match = client.balance().unwrap();
|
||||
let mm = server
|
||||
.try_handle_economy(
|
||||
"POST",
|
||||
"/ut/delete/game/fifa17/match",
|
||||
&[],
|
||||
br#"{"endReason":"WIN"}"#,
|
||||
None,
|
||||
)
|
||||
.expect("match routed");
|
||||
assert_eq!(mm.status, 200);
|
||||
assert_eq!(
|
||||
client.balance().unwrap(),
|
||||
before_match + 400,
|
||||
"WIN credited +400 via Core"
|
||||
);
|
||||
|
||||
// 5) MARKET buy-now (async handlers via the bridge): list -> query -> buy ->
|
||||
// query -> second buy fails, exactly one debit + one sale.
|
||||
let list = server
|
||||
.try_handle_economy(
|
||||
"POST",
|
||||
"/ut/game/fifa17/auctionhouse",
|
||||
&[],
|
||||
br#"{"itemData":{"id":777,"resourceId":777},"buyNowPrice":1000,"startingBid":500}"#,
|
||||
None,
|
||||
)
|
||||
.expect("market list routed");
|
||||
let lv: Value = serde_json::from_slice(&list.body).unwrap();
|
||||
let trade_id = lv["id"].as_i64().expect("trade id");
|
||||
let trade_path = format!("/ut/game/fifa17/trade/{trade_id}");
|
||||
|
||||
let q1 = server
|
||||
.try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None)
|
||||
.unwrap();
|
||||
let q1v: Value = serde_json::from_slice(&q1.body).unwrap();
|
||||
assert_eq!(
|
||||
q1v["auctionInfo"].as_array().unwrap().len(),
|
||||
1,
|
||||
"listing active before buy"
|
||||
);
|
||||
|
||||
let before_buy = client.balance().unwrap();
|
||||
let buy1 = server
|
||||
.try_handle_economy("POST", &trade_path, &[], b"{}", None)
|
||||
.expect("market buy routed");
|
||||
assert_eq!(buy1.status, 200);
|
||||
assert_eq!(
|
||||
client.balance().unwrap(),
|
||||
before_buy - 1000,
|
||||
"buy debited exactly the buy-now price"
|
||||
);
|
||||
|
||||
let q2 = server
|
||||
.try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None)
|
||||
.unwrap();
|
||||
let q2v: Value = serde_json::from_slice(&q2.body).unwrap();
|
||||
assert_eq!(
|
||||
q2v["auctionInfo"].as_array().unwrap().len(),
|
||||
0,
|
||||
"listing sold: no longer active"
|
||||
);
|
||||
|
||||
let after_buy = client.balance().unwrap();
|
||||
let buy2 = server
|
||||
.try_handle_economy("POST", &trade_path, &[], b"{}", None)
|
||||
.unwrap();
|
||||
let buy2v: Value = serde_json::from_slice(&buy2.body).unwrap();
|
||||
assert_eq!(
|
||||
buy2v["auctionInfo"].as_array().unwrap().len(),
|
||||
0,
|
||||
"second buy sees a closed auction"
|
||||
);
|
||||
assert_eq!(
|
||||
client.balance().unwrap(),
|
||||
after_buy,
|
||||
"second buy does NOT debit again"
|
||||
);
|
||||
|
||||
// 6) MARKET cancel: a cancelled listing cannot be bought.
|
||||
let clist = server
|
||||
.try_handle_economy(
|
||||
"POST",
|
||||
"/ut/game/fifa17/auctionhouse",
|
||||
&[],
|
||||
br#"{"itemData":{"id":888,"resourceId":888},"buyNowPrice":1000,"startingBid":500}"#,
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
let cancel_id = serde_json::from_slice::<Value>(&clist.body).unwrap()["id"]
|
||||
.as_i64()
|
||||
.unwrap();
|
||||
server
|
||||
.try_handle_economy(
|
||||
"DELETE",
|
||||
&format!("/ut/delete/game/fifa17/trade/{cancel_id}"),
|
||||
&[],
|
||||
b"",
|
||||
None,
|
||||
)
|
||||
.expect("market cancel routed");
|
||||
let bal_before_cancel_buy = client.balance().unwrap();
|
||||
let cancel_buy = server
|
||||
.try_handle_economy(
|
||||
"POST",
|
||||
&format!("/ut/game/fifa17/trade/{cancel_id}"),
|
||||
&[],
|
||||
b"{}",
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<Value>(&cancel_buy.body).unwrap()["auctionInfo"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.len(),
|
||||
0,
|
||||
"cancelled listing is not buyable"
|
||||
);
|
||||
assert_eq!(
|
||||
client.balance().unwrap(),
|
||||
bal_before_cancel_buy,
|
||||
"buying a cancelled listing does not debit"
|
||||
);
|
||||
|
||||
// 7) Move a still-owned minted card to the trade pile (durable pile metadata).
|
||||
let move_wire = items[1]["id"].as_i64().unwrap();
|
||||
let mv = server
|
||||
.try_handle_economy(
|
||||
"PUT",
|
||||
"/ut/game/fifa17/item",
|
||||
&[],
|
||||
format!(r#"{{"itemData":[{{"id":{move_wire},"pile":"trade"}}]}}"#).as_bytes(),
|
||||
None,
|
||||
)
|
||||
.expect("move routed");
|
||||
let mvv: Value = serde_json::from_slice(&mv.body).unwrap();
|
||||
assert_eq!(mvv["itemData"][0]["success"], true, "move recorded");
|
||||
let moved_core_id = resolver
|
||||
.owned_id_for_wire(move_wire)
|
||||
.expect("moved item reverses to a Core id");
|
||||
|
||||
SeqResult {
|
||||
sold_listing: trade_id.to_string(),
|
||||
moved_core_id,
|
||||
}
|
||||
}
|
||||
|
||||
/// Reopen the durable market/pile SQLite stores from the SAME files (a fresh
|
||||
/// process would do exactly this) and prove the sold listing and the pile move
|
||||
/// persisted. Core-side coin/inventory persistence across a full Core restart is
|
||||
/// proven by `economy_end_to_end_and_restart_persistence`; here the focus is the
|
||||
/// host-owned durable stores.
|
||||
fn verify_store_durability(dir: &std::path::Path, seq: &SeqResult) {
|
||||
let bridge = AsyncBridge::new().unwrap();
|
||||
let market_path = dir.join("market.db").to_string_lossy().into_owned();
|
||||
let market = bridge
|
||||
.block_on(async move { MarketStore::open(&market_path).await })
|
||||
.unwrap();
|
||||
let sold = seq.sold_listing.clone();
|
||||
let m2 = market.clone();
|
||||
let listing = bridge
|
||||
.block_on(async move { m2.get_listing(&sold).await })
|
||||
.expect("sold listing survives reopen");
|
||||
assert_eq!(listing.state, "sold", "sold stays sold after reopen");
|
||||
let pile_path = dir.join("pile.db").to_string_lossy().into_owned();
|
||||
let piles = bridge
|
||||
.block_on(async move { PileStore::open(&pile_path).await })
|
||||
.unwrap();
|
||||
let moved = seq.moved_core_id.clone();
|
||||
let p2 = piles.clone();
|
||||
let pile = bridge
|
||||
.block_on(async move { p2.get(&moved).await })
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
pile.as_deref(),
|
||||
Some("trade"),
|
||||
"pile move persisted after reopen"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
||||
async fn economy_full_sequence_through_dispatch() {
|
||||
let dir = std::env::temp_dir().join(format!(
|
||||
"openfut-econ-dispatch-{}-{}",
|
||||
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());
|
||||
|
||||
// Core #1: run the full write sequence on a plain OS thread (direct bridge
|
||||
// path). The join runs in spawn_blocking so Core's runtime keeps serving.
|
||||
let (h1, base1) = start_core_seeded(&db_url, true).await;
|
||||
let (b1, d1) = (base1.clone(), dir.clone());
|
||||
let seq = tokio::task::spawn_blocking(move || {
|
||||
let t = std::thread::spawn(move || economy_sequence(&b1, &d1));
|
||||
t.join().expect("sequence thread")
|
||||
})
|
||||
.await
|
||||
.expect("write sequence");
|
||||
h1.abort();
|
||||
|
||||
// Durable host stores: reopen the market/pile files from disk (as a fresh
|
||||
// process would) and prove sold/pile state persisted. No Core rebuild — the
|
||||
// synthetic market mint uses the wire resourceId as a placeholder card id,
|
||||
// which Core's content preflight (correctly) rejects on reboot; Core-side
|
||||
// coin/inventory restart persistence is covered by the sibling test.
|
||||
let d2 = dir.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let t = std::thread::spawn(move || verify_store_durability(&d2, &seq));
|
||||
t.join().expect("durability thread")
|
||||
})
|
||||
.await
|
||||
.expect("durability phase");
|
||||
|
||||
std::fs::remove_dir_all(&dir).ok();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user