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

feat(consume): durable per-instance contract state + HTTP apply route
CI / Build, lint & test (push) Successful in 3m16s
Details

Browse Source 
`consume_item` was a complete, tested, atomic apply transaction with zero
production callers and no route -- it could not be reached over HTTP because
its effect is an in-process `ItemMutation` trait object and the host is a
separate process on a synchronous JSON boundary.

Closes that gap with a CLOSED, Core-validated effect vocabulary rather than a
pass-through: `InstanceEffect::AddContractMatches { amount, cap,
default_when_unset }`. A generic "apply this field/value" escape hatch would
hand economic authority back to the caller and break the architecture.

The read-modify-write runs INSIDE the caller's transaction
(`min(cap, COALESCE(contract_matches, default) + amount)`) so two concurrent
applies cannot lose an update, and the reported `granted` stays the requested
amount even when the cap clamps the total.

Migration 0028 adds `owned_cards.contract_matches` NULLABLE: NULL means "Core
tracks no contract here", which keeps the pack-fresh default (a FIFA-specific
7) out of Core and leaves every existing row unchanged in meaning. ADD COLUMN,
not a rebuild -- a rebuild would drop 0026's transfer trigger.

Two ordering fixes forced by putting this on the live path:
* consume_item moves from DEFERRED `pool.begin()` to `BEGIN IMMEDIATE`, the
  discipline economy.rs documents: three reads precede the first write, which
  is exactly the shape that returns SQLITE_BUSY past the busy handler.
* the replay answer now precedes source validation. With DestroyInstance the
  first apply deletes the source, so the old order answered a retry with 404
  instead of the recorded outcome -- replay semantics were unreachable.
This commit is contained in:
funman300
2026-08-22 18:23:05 +00:00
parent 233df1d99d
commit e8be289660
12 changed files with 1019 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/consume.rs
+262 -166
View File
@@ -2,17 +2,24 @@
//!
//! ONE Core transaction that does, in this order and nothing else:
//!
//! 1. validate the SOURCE — it exists, belongs to the club, and is the
//! 1. answer a REPLAY — if `(profile_id, action_identity)` is already recorded,
//! echo that outcome and touch nothing. This precedes every validation on
//! purpose: a completed application has already DESTROYED its source, so
//! checking the source first would answer "not found" to a retried request
//! that in fact succeeded;
//! 2. validate the SOURCE — it exists, belongs to the club, and is the
//! `ContentKind` the caller expected;
//! 2. validate the TARGET — nothing at all (`ConsumeTarget::Club`) or an owned
//! 3. validate the TARGET — nothing at all (`ConsumeTarget::Club`) or an owned
//! instance that exists, belongs to the club, and is the expected kind;
//! 3. write the replay guard — `UNIQUE(profile_id, action_identity)` on
//! 4. write the replay guard — `UNIQUE(profile_id, action_identity)` on
//! `consumable_applications`, so a duplicate is refused BEFORE anything is
//! mutated or consumed (same discipline as `match_completions`);
//! 4. apply the caller's mutation to the target;
//! 5. consume the source EXACTLY ONCE — destroy the instance, or decrement its
//! mutated or consumed (same discipline as `match_completions`). Step 1 is
//! a courtesy; THIS is the guarantee, and it holds against a writer on any
//! other connection or process;
//! 5. apply the caller's mutation to the target;
//! 6. consume the source EXACTLY ONCE — destroy the instance, or decrement its
//! stack and destroy it at zero;
//! 6. commit.
//! 7. commit.
//!
//! Anything failing at any step rolls the whole thing back: the source is never
//! spent without the effect landing, and the effect never lands without the
@@ -21,9 +28,10 @@
//! Core deliberately supplies **no per-category formula**. What a fitness card,
//! a contract, a chemistry style or a position modifier actually DOES to its
//! target is the calling game adapter's reversed behaviour, passed in as
//! [`ItemMutation`]; an unreversed behaviour must not be invented here, and the
//! honest stopping point for one is ownership + projection, i.e. not calling
//! this function at all.
//! [`ItemMutation`] — either in-process, or described over the wire through the
//! closed, Core-validated vocabulary in [`crate::services::instance_effect`]. An
//! unreversed behaviour must not be invented here, and the honest stopping point
//! for one is ownership + projection, i.e. not calling this function at all.
use std::future::Future;
use std::pin::Pin;
@@ -38,6 +46,7 @@ use crate::{
db::Pool,
error::{AppError, AppResult},
models::card::{ContentKind, OwnedCard, OWNED_CARD_SELECT},
services::economy,
};
/// What the transaction does to the source instance once the effect is applied.
@@ -169,6 +178,16 @@ fn require_kind(card: &OwnedCard, expected: ContentKind, role: &str) -> AppResul
Ok(())
}
/// Whether this call did the work or found the identity already recorded.
///
/// The replay branch cannot read the recorded outcome while the transaction is
/// still open on this connection, so it is reported out of the transaction and
/// answered once the connection is free again.
enum Applied {
Fresh(ConsumeOutcome),
Replay,
}
/// Apply one consumable to one target, exactly once. See the module docs.
pub async fn consume_item<M: ItemMutation>(
pool: &Pool,
@@ -197,187 +216,231 @@ pub async fn consume_item<M: ItemMutation>(
}
}
let mut tx = pool.begin().await?;
// ONE connection with an explicit BEGIN IMMEDIATE, never a DEFERRED
// `pool.begin()`: this transaction performs three reads before its first
// write, and a deferred transaction only upgrades to a write at that first
// write — where SQLite answers SQLITE_BUSY *immediately*, bypassing
// `busy_timeout` (the full reasoning is on `economy::finish`). Two clients
// applying consumables at once is ordinary traffic, so take the write lock
// up front and let a rival wait instead of fail.
let mut conn = pool.acquire().await?;
sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?;
// 1. source: owned by this club, and the kind the caller expected.
let source = fetch_owned(&mut tx, req.source_owned_card_id, club_id).await?;
require_kind(&source, req.expected_source_kind, "source item")?;
// A source that is fielded in a squad cannot be consumed: `squad_players`
// holds a FK onto `owned_cards(id)`, so the DELETE below would fail anyway.
// Refuse explicitly instead of surfacing SQLITE_CONSTRAINT, and never
// silently evict a lineup as a side effect of spending an item.
let fielded =
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM squad_players WHERE owned_card_id = ?")
.bind(req.source_owned_card_id)
.fetch_one(&mut *tx)
.await?;
if fielded > 0 {
return Err(AppError::Conflict(format!(
"source item '{}' is fielded in a squad and cannot be consumed",
req.source_owned_card_id
)));
}
// 2. target.
let target = match req.target {
ConsumeTarget::OwnedCard {
owned_card_id,
expected_kind,
} => {
let card = fetch_owned(&mut tx, owned_card_id, club_id).await?;
require_kind(&card, expected_kind, "target item")?;
Some(card)
let result = async {
// 1. a replay is answered before anything is validated: a completed
// application has already destroyed (or drawn down) its source, so
// validating first would answer NotFound to a retry of a request that
// actually succeeded. The write lock is already held, so this read
// cannot race the guard INSERT below — which stays as the real
// guarantee for a writer on any other connection.
if recorded(&mut conn, profile_id, req.action_identity).await? {
return Ok(Applied::Replay);
}
ConsumeTarget::Club => None,
};
// 3. replay guard FIRST — before the mutation and before the consumption, so
// a duplicate cannot apply a second effect or spend a second charge. The
// recorded `effect` is filled in below, once the mutation has produced it.
let application_id = Uuid::new_v4().to_string();
let now = Utc::now().to_rfc3339();
let guard = sqlx::query(
"INSERT INTO consumable_applications \
(id, profile_id, action_identity, source_owned_card_id, source_card_id, \
source_content_kind, source_consumed, source_quantity_after, \
target_owned_card_id, effect, applied_at) \
VALUES (?, ?, ?, ?, ?, ?, 0, NULL, ?, '', ?)",
)
.bind(&application_id)
.bind(profile_id)
.bind(req.action_identity)
.bind(&source.id)
.bind(&source.card_id)
.bind(source.content_kind.as_str())
.bind(target.as_ref().map(|t| t.id.as_str()))
.bind(&now)
.execute(&mut *tx)
.await;
match guard {
Ok(_) => {}
Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
tx.rollback().await?;
return already_applied(pool, profile_id, req.action_identity).await;
// 2. source: owned by this club, and the kind the caller expected.
let source = fetch_owned(&mut conn, req.source_owned_card_id, club_id).await?;
require_kind(&source, req.expected_source_kind, "source item")?;
// A source that is fielded in a squad cannot be consumed: `squad_players`
// holds a FK onto `owned_cards(id)`, so the DELETE below would fail anyway.
// Refuse explicitly instead of surfacing SQLITE_CONSTRAINT, and never
// silently evict a lineup as a side effect of spending an item.
let fielded = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM squad_players WHERE owned_card_id = ?",
)
.bind(req.source_owned_card_id)
.fetch_one(&mut *conn)
.await?;
if fielded > 0 {
return Err(AppError::Conflict(format!(
"source item '{}' is fielded in a squad and cannot be consumed",
req.source_owned_card_id
)));
}
Err(e) => {
tx.rollback().await?;
return Err(e.into());
}
}
// 4. the caller's effect on the target, inside this transaction.
let ctx = ConsumeContext {
profile_id: profile_id.to_string(),
club_id: club_id.to_string(),
source,
target,
};
let effect = mutation.apply(&mut tx, &ctx).await?;
// 5. consume the source exactly once. Both paths assert rows_affected == 1,
// so a concurrent spend of the same instance (which lost the SQLite write
// lock and now sees the row gone / already decremented) fails instead of
// granting a second effect.
let (source_destroyed, source_quantity_after) = match req.consumption {
SourceConsumption::DestroyInstance => {
let deleted = sqlx::query("DELETE FROM owned_cards WHERE id = ? AND club_id = ?")
.bind(&ctx.source.id)
.bind(club_id)
.execute(&mut *tx)
.await?
.rows_affected();
if deleted != 1 {
return Err(AppError::Conflict(format!(
"source item '{}' was already consumed",
ctx.source.id
)));
// 3. target.
let target = match req.target {
ConsumeTarget::OwnedCard {
owned_card_id,
expected_kind,
} => {
let card = fetch_owned(&mut conn, owned_card_id, club_id).await?;
require_kind(&card, expected_kind, "target item")?;
Some(card)
}
(true, None)
ConsumeTarget::Club => None,
};
// 4. replay guard — the durable one. It precedes the mutation and the
// consumption, so a duplicate that got past step 1 on another
// connection still cannot apply a second effect or spend a second
// charge. The recorded `effect` is filled in below, once the mutation
// has produced it.
let application_id = Uuid::new_v4().to_string();
let now = Utc::now().to_rfc3339();
let guard = sqlx::query(
"INSERT INTO consumable_applications \
(id, profile_id, action_identity, source_owned_card_id, source_card_id, \
source_content_kind, source_consumed, source_quantity_after, \
target_owned_card_id, effect, applied_at) \
VALUES (?, ?, ?, ?, ?, ?, 0, NULL, ?, '', ?)",
)
.bind(&application_id)
.bind(profile_id)
.bind(req.action_identity)
.bind(&source.id)
.bind(&source.card_id)
.bind(source.content_kind.as_str())
.bind(target.as_ref().map(|t| t.id.as_str()))
.bind(&now)
.execute(&mut *conn)
.await;
match guard {
Ok(_) => {}
// A collision writes nothing, so this transaction has nothing to undo
// and simply ends; the recorded outcome is read back afterwards.
Err(sqlx::Error::Database(e)) if e.is_unique_violation() => return Ok(Applied::Replay),
Err(e) => return Err(e.into()),
}
SourceConsumption::DecrementStack { amount } => {
let Some(have) = ctx.source.quantity else {
return Err(AppError::BadRequest(format!(
"source item '{}' carries no stack size; it can only be destroyed",
ctx.source.id
)));
};
if have < amount {
return Err(AppError::Conflict(format!(
"source item '{}' holds {have}, cannot consume {amount}",
ctx.source.id
)));
}
let remaining = have - amount;
if remaining == 0 {
let deleted = sqlx::query(
"DELETE FROM owned_cards WHERE id = ? AND club_id = ? AND quantity = ?",
)
.bind(&ctx.source.id)
.bind(club_id)
.bind(have)
.execute(&mut *tx)
.await?
.rows_affected();
// 5. the caller's effect on the target, inside this transaction.
let ctx = ConsumeContext {
profile_id: profile_id.to_string(),
club_id: club_id.to_string(),
source,
target,
};
let effect = mutation.apply(&mut conn, &ctx).await?;
// 6. consume the source exactly once. Both paths assert rows_affected == 1,
// so a concurrent spend of the same instance (which lost the SQLite write
// lock and now sees the row gone / already decremented) fails instead of
// granting a second effect.
let (source_destroyed, source_quantity_after) = match req.consumption {
SourceConsumption::DestroyInstance => {
let deleted = sqlx::query("DELETE FROM owned_cards WHERE id = ? AND club_id = ?")
.bind(&ctx.source.id)
.bind(club_id)
.execute(&mut *conn)
.await?
.rows_affected();
if deleted != 1 {
return Err(AppError::Conflict(format!(
"source item '{}' changed under us",
"source item '{}' was already consumed",
ctx.source.id
)));
}
(true, None)
} else {
let updated = sqlx::query(
"UPDATE owned_cards SET quantity = ? WHERE id = ? AND club_id = ? \
AND quantity = ?",
)
.bind(remaining)
.bind(&ctx.source.id)
.bind(club_id)
.bind(have)
.execute(&mut *tx)
.await?
.rows_affected();
if updated != 1 {
}
SourceConsumption::DecrementStack { amount } => {
let Some(have) = ctx.source.quantity else {
return Err(AppError::BadRequest(format!(
"source item '{}' carries no stack size; it can only be destroyed",
ctx.source.id
)));
};
if have < amount {
return Err(AppError::Conflict(format!(
"source item '{}' changed under us",
"source item '{}' holds {have}, cannot consume {amount}",
ctx.source.id
)));
}
(false, Some(remaining))
let remaining = have - amount;
if remaining == 0 {
let deleted = sqlx::query(
"DELETE FROM owned_cards WHERE id = ? AND club_id = ? AND quantity = ?",
)
.bind(&ctx.source.id)
.bind(club_id)
.bind(have)
.execute(&mut *conn)
.await?
.rows_affected();
if deleted != 1 {
return Err(AppError::Conflict(format!(
"source item '{}' changed under us",
ctx.source.id
)));
}
(true, None)
} else {
let updated = sqlx::query(
"UPDATE owned_cards SET quantity = ? WHERE id = ? AND club_id = ? \
AND quantity = ?",
)
.bind(remaining)
.bind(&ctx.source.id)
.bind(club_id)
.bind(have)
.execute(&mut *conn)
.await?
.rows_affected();
if updated != 1 {
return Err(AppError::Conflict(format!(
"source item '{}' changed under us",
ctx.source.id
)));
}
(false, Some(remaining))
}
}
}
};
};
let effect_text = serde_json::to_string(&effect)?;
sqlx::query(
"UPDATE consumable_applications \
SET effect = ?, source_consumed = ?, source_quantity_after = ? WHERE id = ?",
let effect_text = serde_json::to_string(&effect)?;
sqlx::query(
"UPDATE consumable_applications \
SET effect = ?, source_consumed = ?, source_quantity_after = ? WHERE id = ?",
)
.bind(&effect_text)
.bind(i64::from(source_destroyed))
.bind(source_quantity_after)
.bind(&application_id)
.execute(&mut *conn)
.await?;
Ok(Applied::Fresh(ConsumeOutcome {
applied: true,
action_identity: req.action_identity.to_string(),
source_owned_card_id: ctx.source.id.clone(),
source_destroyed,
source_quantity_after,
target_owned_card_id: ctx.target.as_ref().map(|t| t.id.clone()),
effect,
}))
}
.await;
match economy::finish(&mut conn, result).await? {
Applied::Fresh(outcome) => Ok(outcome),
// Read the recorded outcome on the SAME connection: acquiring a second one
// while still holding this one deadlocks a pool saturated with racing
// appliers, which is exactly the case a replay shows up in.
Applied::Replay => already_applied(&mut conn, profile_id, req.action_identity).await,
}
}
/// Has this identity already been applied? Read inside the transaction, under
/// the write lock, so the answer cannot go stale before the guard INSERT.
async fn recorded(
conn: &mut SqliteConnection,
profile_id: &str,
action_identity: &str,
) -> AppResult<bool> {
let hits = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM consumable_applications \
WHERE profile_id = ? AND action_identity = ?",
)
.bind(&effect_text)
.bind(i64::from(source_destroyed))
.bind(source_quantity_after)
.bind(&application_id)
.execute(&mut *tx)
.bind(profile_id)
.bind(action_identity)
.fetch_one(&mut *conn)
.await?;
tx.commit().await?;
Ok(ConsumeOutcome {
applied: true,
action_identity: req.action_identity.to_string(),
source_owned_card_id: ctx.source.id.clone(),
source_destroyed,
source_quantity_after,
target_owned_card_id: ctx.target.as_ref().map(|t| t.id.clone()),
effect,
})
Ok(hits > 0)
}
/// Echo the recorded outcome of an application that already happened. Mutates
/// nothing and reports `applied = false`.
async fn already_applied(
pool: &Pool,
conn: &mut SqliteConnection,
profile_id: &str,
action_identity: &str,
) -> AppResult<ConsumeOutcome> {
@@ -388,7 +451,7 @@ async fn already_applied(
)
.bind(profile_id)
.bind(action_identity)
.fetch_optional(pool)
.fetch_optional(&mut *conn)
.await?
.ok_or_else(|| {
AppError::Internal(anyhow::anyhow!(
@@ -621,6 +684,39 @@ mod tests {
);
}
/// A retry of a request that ALREADY succeeded must replay, not 404. With
/// `DestroyInstance` the source no longer exists by then, so answering the
/// replay has to precede source validation — otherwise a client that lost the
/// response to a successful apply is told its item was never there.
#[tokio::test]
async fn a_replay_survives_the_source_it_destroyed() {
let (_dir, _url, pool) = fixture().await;
let spend = || {
req(
"act-gone",
"single",
SourceConsumption::DestroyInstance,
"player",
)
};
let first = consume_item(&pool, "prof-a", "club-a", &spend(), &BumpTraining)
.await
.expect("first");
assert!(first.applied);
let replay = consume_item(&pool, "prof-a", "club-a", &spend(), &BumpTraining)
.await
.expect("a retry must replay, not fail on the destroyed source");
assert!(!replay.applied);
assert!(replay.source_destroyed);
assert_eq!(replay.effect, first.effect);
assert_eq!(training(&pool, "player").await, 1, "effect applied once");
assert_eq!(
count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
1
);
}
#[tokio::test]
async fn inline_closure_effect_is_accepted() {
let (_dir, _url, pool) = fixture().await;