crates/monetize-server/tests/callback_ordering.rs
//! **The vendor's deliveries arrive in the wrong order, twice, or about money
//! going back** — over the real webhook route, through the real router.
//!
//! The vendor's own outages and the refused-delivery ring are held elsewhere
//! (`callback_refusals.rs`, `transaction::tests::reconcile_*`). What is here is
//! ORDER, and one gap named.
//!
//! The property that makes order safe is easy to miss when reading
//! `callback.rs`: the handler takes the event's `reference` and `handle` and
//! **throws the status away**. It then calls `resume_reference`, which asks the
//! vendor what it holds NOW. So a delivery is a doorbell, never a fact — and a
//! stale one cannot un-capture a purchase, because nothing downstream of the
//! parse ever reads what the stale one said.
//!
//! That is the right design and it had no test. A future handler that switched
//! on `event.status` — which is sitting right there on the struct — would pass
//! every existing test in this repo and break this one.
use std::sync::Arc;
use monetize::testing::{Answer, MockCloud};
use monetize_proto::entitlement_client::EntitlementClient;
use monetize_proto::ledger_client::LedgerClient;
use monetize_proto::purchase_client::PurchaseClient;
use monetize_proto::query_ledger_request::By;
use monetize_proto::{self as pb, PurchaseState};
use monetize_server::registry::Registry;
use monetize_server::state::AppState;
use monetize_server::testing::estate::FakeEstate;
use monetize_server::testing::products::TestProduct;
use monetize_server::testing::{http_post, CallbackPay, TestServer};
use tonic::metadata::MetadataValue;
use tonic::transport::Channel;
use tonic::Request;
const BEARER: &str = "test-bearer-token-of-sufficient-length";
const SECRET: &str = "mockpay-shared-secret";
const GIB: u64 = 1 << 30;
struct Rig {
operator: String,
pay: CallbackPay,
/// The product's estate (its main pulls what it is granted).
product: FakeEstate,
state: Arc<AppState>,
server: TestServer,
channel: Channel,
}
impl Rig {
async fn up() -> Rig {
Self::up_with_operator("").await
}
async fn up_with_operator(operator: &str) -> Rig {
Self::up_with(operator, None).await
}
/// A rig whose clock stands still at `now`, when given.
async fn up_with(operator: &str, now: Option<u64>) -> Rig {
let pay = CallbackPay::new(SECRET);
let cloud = MockCloud::new();
cloud.answer_all(Answer::OnHand);
let mut state = AppState::in_memory(Registry::new().pulled_product(TestProduct::gunnar(), "gunnar.rs").sells("gunnar", pay.clone()).cloud(cloud));
let product = FakeEstate::declare(&state, "gunnar", "gunnar.rs");
product.report(|r| {
r.servable = Some(monetize_product::Servable::Measured { servable_bytes: 300 * GIB, total_bytes: 400 * GIB, growth: monetize_product::Growth::Unknown("a test estate".into()), ceilings: Default::default() });
});
state.checkout_urls.get_mut().unwrap().insert("mockpay".into(), "https://pay.example/{handle}".into());
if let Some(now) = now {
state.clock = Box::new(move || now);
}
let state = Arc::new(state);
let server = TestServer::spawn_with_operator(state.clone(), BEARER, operator).await;
let channel = Channel::from_shared(server.endpoint()).unwrap().connect().await.unwrap();
Rig { operator: operator.to_owned(), pay, product, state, server, channel }
}
fn authed<T>(&self, req: T) -> Request<T> {
let mut r = Request::new(req);
r.metadata_mut().insert("authorization", MetadataValue::try_from(format!("Bearer {BEARER}")).unwrap());
r
}
/// An order for 10 GiB. The customer has NOT authorized at the vendor, so
/// it comes back PENDING with a redirect — the shape every test here needs.
async fn start(&self, tenant: &str) -> pb::StartPurchaseResponse {
PurchaseClient::new(self.channel.clone())
.start(self.authed(pb::StartPurchaseRequest {
product: "gunnar".into(),
tenant: tenant.into(),
caps: [("pack_bytes".to_string(), 10 * GIB)].into(),
vendor: "mockpay".into(),
payer: String::new(),
starts_unix_ms: monetize::testing::NOW,
market: pb::Market::Se as i32,
renew: false,
actor_ticket: Vec::new(),
bundle: String::new(),
users: None,
}))
.await
.expect("the order goes through")
.into_inner()
}
fn as_operator<T>(&self, req: T) -> Request<T> {
let mut r = Request::new(req);
r.metadata_mut().insert("authorization", MetadataValue::try_from(format!("Bearer {}", self.operator)).unwrap());
r
}
/// A reversal delivery (`refunded` | `dispute_opened` | `dispute_won` |
/// `dispute_lost`) of `amount` minor units.
fn deliver_reversal(&self, reference: &str, handle: &str, event: &str, amount: u64) -> (u16, String) {
let body = CallbackPay::reversal_body(reference, handle, event, amount);
http_post(self.server.addr, "/callback/mockpay", &self.pay.signed_headers(), &body).expect("the webhook answers")
}
/// One delivery, as the vendor would POST it. `(status, body)`.
fn deliver(&self, reference: &str, handle: &str, event: &str) -> (u16, String) {
let body = CallbackPay::callback_body(reference, handle, event);
http_post(self.server.addr, "/callback/mockpay", &self.pay.signed_headers(), &body).expect("the webhook answers")
}
async fn ledger_kinds(&self, reference: &str) -> Vec<String> {
LedgerClient::new(self.channel.clone())
.query(self.authed(pb::QueryLedgerRequest { by: Some(By::Reference(reference.into())) }))
.await
.unwrap()
.into_inner()
.rows
.into_iter()
.map(|r| r.kind)
.collect()
}
async fn caps_of(&self, tenant: &str) -> Option<std::collections::BTreeMap<String, u64>> {
let out = EntitlementClient::new(self.channel.clone())
.get(self.authed(pb::GetEntitlementRequest { product: "gunnar".into(), tenant: tenant.into() }))
.await
.unwrap()
.into_inner();
out.fact.map(|f| f.caps)
}
}
/// ★ **TWO DELIVERIES IN THE WRONG ORDER ARE ONE CAPTURE AND ONE GRANT.**
///
/// The vendor emits `authorized` and then `captured`; the network delivers
/// them the other way round, which is ordinary and which no vendor promises
/// not to do. Both are verified, both name the same reference and handle.
///
/// The handler must reach the same place either way, because it decides from
/// the vendor's CURRENT state and not from the delivery. The applied outputs
/// are the vendor's own capture count and the product's pushed facts — never a
/// round-trip through something this server wrote.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_deliveries_in_the_wrong_order_capture_once_and_grant_once() {
let rig = Rig::up().await;
let started = rig.start("alice").await;
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured"]);
assert!(rig.product.entitlements().is_empty(), "nothing granted before the money moved");
// `start` already asked the vendor to capture once and was told `Reserved`
// — the customer had not authorized — so the interesting number is how
// many MORE captures the two deliveries cause.
let captures_before = rig.pay.captures();
// The customer authorizes at the vendor's page. Both events now exist
// there; the LATER one is delivered first.
rig.pay.authorize(&started.handle_id);
let (code, body) = rig.deliver(&started.reference, &started.handle_id, "captured");
assert_eq!(code, 200, "{body}");
assert!(body.contains("\"PAID\""), "{body}");
// And now the EARLIER one arrives, saying the charge is merely reserved.
// If anything downstream believed the delivery, this is where a captured
// purchase would be walked backwards.
let (code, body) = rig.deliver(&started.reference, &started.handle_id, "authorized");
assert_eq!(code, 200, "a duplicate delivery must never be a 5xx: {body}");
assert!(body.contains("ignored"), "the attempt is closed, so it says it did nothing: {body}");
// ONE OF EVERYTHING.
assert_eq!(rig.pay.captures() - captures_before, 1, "the vendor was asked to draw the money twice");
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured", "Captured", "EntitlementPushed"]);
assert_eq!(rig.product.entitlements().len(), 1, "the customer was granted their room twice");
assert_eq!(rig.caps_of("alice").await.expect("alice holds a fact")["pack_bytes"], 10 * GIB);
assert!(rig.state.ledger.unfinished().unwrap().is_empty());
rig.server.stop().await;
}
/// ★ **A STALE `released` AFTER THE CAPTURE DOES NOT UNDO A PAID PURCHASE.**
///
/// The worst ordering: the customer's abandoned-checkout event is emitted
/// before they come back and pay, and is delivered after. Taken at its word it
/// says *this charge was released* about a charge that has been drawn and an
/// entitlement that has been granted.
///
/// It is answered 200 — it is the vendor working correctly — and it changes
/// nothing, because the attempt is closed and the handler asks the vendor
/// rather than the envelope.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_stale_release_delivered_after_the_capture_changes_nothing() {
let rig = Rig::up().await;
let started = rig.start("bob").await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
let rows = rig.ledger_kinds(&started.reference).await;
assert_eq!(rows, ["Reserved", "Ensured", "Captured", "EntitlementPushed"]);
let (code, body) = rig.deliver(&started.reference, &started.handle_id, "released");
assert_eq!(code, 200, "{body}");
assert!(body.contains("ignored"), "{body}");
assert_eq!(rig.ledger_kinds(&started.reference).await, rows, "the ledger moved after a stale event");
assert_eq!(rig.product.entitlements().len(), 1);
assert!(matches!(rig.pay.vendor_state(&started.handle_id), Some(monetize_pay::ChargeStatus::Captured { .. })), "the money is still drawn");
assert_eq!(rig.caps_of("bob").await.expect("bob still holds a fact")["pack_bytes"], 10 * GIB);
rig.server.stop().await;
}
/// ★ **A REFUND AFTER THE GRANT IS WRITTEN DOWN, ONCE** (g-disputes, owner
/// 2026-09-28: "a dispute webhook → handled"). Until then it was answered 200
/// and changed nothing — not even a row (the old `known_gap_…` test).
///
/// The delivery is a verified `Reversal`: the ledger gets ONE `Reversed` row
/// however many times the vendor delivers it and the sale stops being
/// declared. The paid time is taken back (the ladder moves the tenant on, see
/// `a_full_refund_or_a_lost_dispute_takes_the_paid_time_back_…`); the caps
/// themselves stay until the ladder says otherwise.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_refund_after_the_grant_is_one_ledger_row_however_often_it_is_delivered() {
let rig = Rig::up().await;
let started = rig.start("carol").await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
let rows = rig.ledger_kinds(&started.reference).await;
assert_eq!(rows, ["Reserved", "Ensured", "Captured", "EntitlementPushed"]);
for _ in 0..2 {
let (code, body) = rig.deliver_reversal(&started.reference, &started.handle_id, "refunded", 12_500);
assert_eq!(code, 200, "a refund notification must not be a 5xx: {body}");
assert!(body.contains("\"reversal\":\"refunded\""), "{body}");
}
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured", "Captured", "EntitlementPushed", "Reversed"], "one row, however many deliveries");
assert_eq!(rig.caps_of("carol").await.expect("carol still holds a fact")["pack_bytes"], 10 * GIB, "a refund does not cut the caps by itself");
let sale = rig.state.ledger.sales().unwrap().into_iter().find(|s| s.reference == started.reference).unwrap();
assert_eq!(sale.back_minor(), 12_500);
assert_eq!(rig.state.refusals.recent(10).len(), 0, "a verified refund is not a refused delivery");
rig.server.stop().await;
}
/// ★ **A DISPUTE IS HANDLED**: opened, then lost — two rows, the sale's money
/// counted as gone back, and a signup's dispute lands on its own reference.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_dispute_opened_then_lost_is_two_rows_and_its_money_is_counted_as_gone() {
let rig = Rig::up().await;
let started = rig.start("dave").await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
let (code, body) = rig.deliver_reversal(&started.reference, &started.handle_id, "dispute_opened", 12_500);
assert_eq!(code, 200, "{body}");
let (code, body) = rig.deliver_reversal(&started.reference, &started.handle_id, "dispute_lost", 12_500);
assert_eq!(code, 200, "{body}");
let kinds = rig.ledger_kinds(&started.reference).await;
assert_eq!(kinds.iter().filter(|k| *k == "Reversed").count(), 2, "{kinds:?}");
let sale = rig.state.ledger.sales().unwrap().into_iter().find(|s| s.reference == started.reference).unwrap();
assert_eq!((sale.dispute.as_deref(), sale.back_minor()), (Some("dispute-lost"), 12_500));
rig.server.stop().await;
}
/// ★ **THE OPERATOR REFUNDS AN ORDER** (g-refund, owner 2026-09-28): the
/// same `SignupAdmin.Refund` verb, operator bearer only, finds the ORDER by
/// its vendor handle, writes `RefundRequested` before the vendor is called
/// and `Reversed` after, and a second call answers the first refund without
/// calling the vendor again.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_operator_refunds_an_order_once_and_the_ledger_says_so() {
use monetize_proto::signup_admin_client::SignupAdminClient;
let rig = Rig::up_with_operator("operator-bearer-of-sufficient-length").await;
let started = rig.start("erin").await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
let ask = || pb::RefundSignupRequest { vendor: "mockpay".into(), session: started.handle_id.clone(), amount_minor: 0, reason: "goodwill".into(), force: false };
// The service bearer is refused.
let denied = SignupAdminClient::new(rig.channel.clone()).refund(rig.authed(ask())).await.unwrap_err();
assert_eq!(denied.code(), tonic::Code::PermissionDenied, "{denied:?}");
let first = SignupAdminClient::new(rig.channel.clone()).refund(rig.as_operator(ask())).await.expect("the order is refunded").into_inner();
assert_eq!(first.state(), pb::RefundState::Done);
let second = SignupAdminClient::new(rig.channel.clone()).refund(rig.as_operator(ask())).await.expect("asked again").into_inner();
assert_eq!(second.refund_id, first.refund_id, "the second call answers the first refund");
assert_eq!(rig.pay.refund_calls().len(), 1, "the vendor was asked once: {:?}", rig.pay.refund_calls());
let kinds = rig.ledger_kinds(&started.reference).await;
assert_eq!(&kinds[kinds.len() - 2..], ["RefundRequested", "Reversed"], "{kinds:?}");
rig.server.stop().await;
}
/// What the order under `reference` came to, gross (tax included) — what a
/// refund can at most give back.
fn gross_of(state: &AppState, reference: &str) -> u64 {
let sale = state.ledger.sales().unwrap().into_iter().find(|s| s.reference == reference).unwrap();
sale.price.minor + sale.tax().minor
}
/// ★ **A REFUND AFTER A LOST DISPUTE IS CAPPED** at what the dispute left: the
/// vendor already took the disputed money back, so the operator's refund can
/// only give back the rest — never the same money twice.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_refund_after_a_lost_dispute_gives_back_only_what_the_dispute_left() {
let rig = Rig::up().await;
let started = rig.start("fran").await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
let gross = gross_of(&rig.state, &started.reference);
assert!(gross > 1_000, "{gross}");
assert_eq!(rig.deliver_reversal(&started.reference, &started.handle_id, "dispute_lost", gross - 1_000).0, 200);
let state = rig.state.clone();
let handle = started.handle_id.clone();
let (whole, rest) = tokio::task::spawn_blocking(move || {
let whole = monetize_server::reversal::refund_order(&state, "mockpay", &handle, gross, "goodwill");
let rest = monetize_server::reversal::refund_order(&state, "mockpay", &handle, 0, "goodwill");
(whole, rest)
})
.await
.unwrap();
assert!(matches!(whole, Err(monetize_server::signup::RefundRefusal::Invalid(_))), "the whole sale again is refused: {whole:?}");
let Ok(monetize_server::signup::SignupRefund::Done { amount, .. }) = rest else { panic!("the rest is refunded: {rest:?}") };
assert_eq!(amount.minor, 1_000, "only what the lost dispute left");
let sale = rig.state.ledger.sales().unwrap().into_iter().find(|s| s.reference == started.reference).unwrap();
assert_eq!(sale.back_minor(), gross, "all of it is back, once");
rig.server.stop().await;
}
/// ★ **TWO REFUNDS AT ONCE CANNOT GIVE BACK MORE THAN THE SALE.** Each asks
/// for more than half; whichever is first gives its amount back and every
/// other is refused — the vendor is never asked for a second, different key.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_refunds_of_one_order_never_exceed_the_sale() {
let rig = Rig::up().await;
for (round, tenant) in ["gus", "hedda", "ivar"].into_iter().enumerate() {
let started = rig.start(tenant).await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
let gross = gross_of(&rig.state, &started.reference);
let n = 16u64;
let barrier = std::sync::Barrier::new(n as usize);
let state = rig.state.clone();
let handle = started.handle_id.clone();
let given: Vec<u64> = std::thread::scope(|scope| {
let workers: Vec<_> = (0..n)
.map(|i| {
let (state, handle, barrier) = (&state, &handle, &barrier);
scope.spawn(move || {
barrier.wait();
match monetize_server::reversal::refund_order(state, "mockpay", handle, gross * 6 / 10 + i, "race") {
Ok(monetize_server::signup::SignupRefund::Done { amount, .. }) => amount.minor,
_ => 0,
}
})
})
.collect();
workers.into_iter().map(|w| w.join().unwrap()).collect()
});
let prefix = format!("{}:refund:", started.handle_id);
let keys: std::collections::BTreeSet<String> = rig.pay.refund_calls().into_iter().filter(|k| k.starts_with(&prefix)).collect();
assert_eq!(keys.len(), 1, "round {round}: the vendor was asked under more than one amount: {keys:?}");
let sale = rig.state.ledger.sales().unwrap().into_iter().find(|s| s.reference == started.reference).unwrap();
assert!(sale.back_minor() <= gross, "round {round}: {} given back of {gross}", sale.back_minor());
let distinct: std::collections::BTreeSet<u64> = given.into_iter().filter(|a| *a > 0).collect();
assert_eq!(distinct.len(), 1, "round {round}: one refund answered: {distinct:?}");
}
rig.server.stop().await;
}
/// ★ **A SALE THAT IS ALL GONE BACK TAKES ITS PAID TIME WITH IT.** A refund
/// of everything (the vendor's, or the operator's) and a lost dispute leave
/// the tenant's fact paid only up to now, so the ladder moves it — Grace,
/// then Suspended — instead of it staying Paid on money that went back. A
/// partial refund is goodwill and leaves the grant as it is; a repeated
/// delivery takes nothing more.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_full_refund_or_a_lost_dispute_takes_the_paid_time_back_and_a_partial_refund_does_not() {
let rig = Rig::up_with("", Some(monetize::testing::NOW + 86_400_000)).await;
let fact = |tenant: &str| rig.state.ledger.entitlement("gunnar", tenant).unwrap().expect("a fact");
let mut paid = Vec::new();
for tenant in ["jon", "kim", "lea", "mo"] {
let started = rig.start(tenant).await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.deliver(&started.reference, &started.handle_id, "authorized").0, 200);
assert_eq!(fact(tenant).state, monetize_product::State::Paid);
paid.push(started);
}
let gross = gross_of(&rig.state, &paid[0].reference);
assert_eq!(rig.deliver_reversal(&paid[0].reference, &paid[0].handle_id, "refunded", gross).0, 200);
assert_eq!(rig.deliver_reversal(&paid[1].reference, &paid[1].handle_id, "dispute_lost", gross - 1_000).0, 200);
assert_eq!(rig.deliver_reversal(&paid[2].reference, &paid[2].handle_id, "refunded", gross / 2).0, 200);
let state = rig.state.clone();
let handle = paid[3].handle_id.clone();
let refunded = tokio::task::spawn_blocking(move || monetize_server::reversal::refund_order(&state, "mockpay", &handle, 0, "the operator's full refund")).await.unwrap();
assert!(refunded.is_ok(), "{refunded:?}");
let now = rig.state.now();
for tenant in ["jon", "kim", "mo"] {
let f = fact(tenant);
assert_ne!(f.state, monetize_product::State::Paid, "{tenant} keeps no paid time for money that went back: {f:?}");
assert!(f.paid_until_unix_ms.is_some_and(|p| p <= now), "{tenant}: {f:?}");
assert_eq!(f.caps["pack_bytes"], 10 * GIB, "{tenant}: the room is the ladder's to move, not taken at once");
}
let lea = fact("lea");
assert_eq!(lea.state, monetize_product::State::Paid, "a partial refund is goodwill: {lea:?}");
assert!(lea.paid_until_unix_ms.is_some_and(|p| p > now));
let before = fact("jon");
assert_eq!(rig.deliver_reversal(&paid[0].reference, &paid[0].handle_id, "refunded", gross).0, 200);
assert_eq!(fact("jon").paid_until_unix_ms, before.paid_until_unix_ms, "a repeated delivery takes nothing more");
rig.server.stop().await;
}