crates/monetize-server/tests/callback_answers_at_once.rs
//! **The vendor's webhook is answered at once; the grant runs behind it**
//! (stinkies money-webhook-504-in-growth).
//!
//! MEASURED 2026-09-30 on the d-upgrade mock (money-1, M3 DK): the completed
//! event for a paid +10 GiB order reached monetize while the data set was
//! growing. The handler resumed the purchase INSIDE the request — the cloud's
//! ensure and the entitlement push to gunnar both queue behind a growth — and
//! vm1's proxy cut the request at 30 s: Stripe saw `504` and retried, while
//! monetize granted the order 29 s later anyway. A vendor that sees 5xx
//! retries for days and then disables the endpoint.
//!
//! What is held here, over the real route through the real router:
//!
//! * a grant that outlasts any proxy is ANSWERED within [`ANSWER_WITHIN`] and
//! GRANTED afterwards, without a redelivery and without a poller tick;
//! * the event is kept DURABLY before the answer, idempotent by its id;
//! * two deliveries of the same event racing each other — the vendor's retry
//! landing while the first grant is still in flight — are ONE capture and
//! ONE push: a retried event cannot double-grant;
//! * a push the product refuses FOR NOW is retried by monetize, not by the
//! vendor, and said on every attempt;
//! * a paid SIGNUP whose sink is slow (it creates the principal and pushes the
//! entitlement on the same box) is answered at once too.
use std::sync::Arc;
use std::time::{Duration, Instant};
use monetize::testing::{Answer, MockCloud, MockProduct};
use monetize_pay::{Currency, Money, PaidSignup};
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::config::SignupPlan;
use monetize_server::registry::Registry;
use monetize_server::state::AppState;
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;
/// What a webhook may take, whatever the grant behind it takes. vm1's proxy
/// cuts at 30 s and Stripe's own delivery gives up sooner; the answer must be
/// far inside both.
const ANSWER_WITHIN: Duration = Duration::from_secs(4);
/// How long a push to the product takes in these tests: longer than
/// [`ANSWER_WITHIN`], so an answer that waited for it is caught.
const SLOW_PUSH: Duration = Duration::from_secs(7);
struct Rig {
pay: CallbackPay,
product: Arc<MockProduct>,
state: Arc<AppState>,
server: TestServer,
channel: Channel,
}
impl Rig {
async fn up() -> Rig {
let pay = CallbackPay::new(SECRET);
let product = Arc::new(MockProduct::gunnar());
product.can_serve(300 * GIB, 400 * GIB);
let cloud = MockCloud::new();
cloud.answer_all(Answer::OnHand);
let product_dyn: Arc<dyn monetize_product::Product> = product.clone();
let mut state = AppState::in_memory(Registry::new().shared_product(product_dyn).sells("gunnar", pay.clone()).cloud(cloud));
state.checkout_urls.get_mut().unwrap().insert("mockpay".into(), "https://pay.example/{handle}".into());
let state = Arc::new(state);
let server = TestServer::spawn(state.clone(), BEARER).await;
let channel = Channel::from_shared(server.endpoint()).unwrap().connect().await.unwrap();
Rig { 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
}
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()
}
async fn status(&self, reference: &str) -> i32 {
PurchaseClient::new(self.channel.clone())
.status(self.authed(pb::PurchaseStatusRequest { reference: reference.into(), ..Default::default() }))
.await
.unwrap()
.into_inner()
.state
}
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()
}
/// One delivery, as the vendor POSTs it, timed. `(status, body, took)`.
async fn deliver(&self, body: Vec<u8>) -> (u16, String, Duration) {
let (addr, headers) = (self.server.addr, self.pay.signed_headers());
tokio::task::spawn_blocking(move || {
let t = Instant::now();
let (code, reply) = http_post(addr, "/callback/mockpay", &headers, &body).expect("the webhook answers");
(code, reply, t.elapsed())
})
.await
.unwrap()
}
/// The product's next push takes [`SLOW_PUSH`] — a gunnar whose
/// `Entitlement.Set` queues behind a data-set growth.
fn next_push_is_slow(&self) {
*self.product.during_push.lock().unwrap() = Some(Box::new(|| std::thread::sleep(SLOW_PUSH)));
}
async fn paid_within(&self, reference: &str, within: Duration) -> bool {
let t = Instant::now();
while t.elapsed() < within {
if self.status(reference).await == PurchaseState::Paid as i32 {
return true;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
false
}
}
/// ★ **A grant that outlasts the proxy is answered at once and granted after.**
///
/// RED before the fix: the answer took the whole push (7 s here; 30 s and a
/// 504 at vm1 during a growth).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_grant_that_outlasts_the_proxy_is_answered_at_once_and_granted_after() {
let rig = Rig::up().await;
let started = rig.start("nordisk").await;
rig.pay.authorize(&started.handle_id);
rig.next_push_is_slow();
let (code, reply, took) = rig.deliver(CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized")).await;
assert_eq!(code, 200, "{reply}");
assert!(took < ANSWER_WITHIN, "the webhook waited {} ms for the grant: {reply}", took.as_millis());
let v: serde_json::Value = serde_json::from_str(&reply).unwrap();
assert_eq!(v["reference"].as_str(), Some(started.reference.as_str()), "{reply}");
assert_eq!(v["state"].as_str(), Some("PENDING"), "answered before the grant, so not yet PAID: {reply}");
assert_eq!(v["accepted"].as_bool(), Some(true), "the answer says the event is kept and is being finished: {reply}");
// Kept before the answer: the event's row is in the inbox already.
let rows = rig.state.callbacks.rows().unwrap();
assert_eq!(rows.len(), 1, "one delivery, one kept event: {rows:?}");
assert_eq!(rows[0].reference, started.reference);
// Granted afterwards, with nothing more from the vendor and no poller tick.
assert!(rig.paid_within(&started.reference, SLOW_PUSH + Duration::from_secs(8)).await, "the grant never landed after the answer");
assert_eq!(rig.product.pushed().len(), 1, "one order, one push");
// The row is closed a moment after the ledger's Paid row: waited for.
let t = Instant::now();
let rows = loop {
let rows = rig.state.callbacks.rows().unwrap();
if rows[0].is_done() || t.elapsed() > Duration::from_secs(2) {
break rows;
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
assert!(rows[0].is_done(), "the kept event is closed with the grant's verdict: {rows:?}");
}
/// ★ **The vendor's retry, landing while the first grant is in flight, does
/// not grant twice.** The incident's shape exactly: Stripe saw a 504 and
/// delivered the same event again while monetize was still pushing.
///
/// RED before the fix: both deliveries resumed the same open attempt at once
/// and both pushed the entitlement.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_retried_event_landing_mid_grant_is_one_capture_and_one_push() {
let rig = Rig::up().await;
let started = rig.start("nordisk").await;
// `Start` already asked the vendor to draw once (not authorized then).
let captures_before = rig.pay.captures();
rig.pay.authorize(&started.handle_id);
rig.next_push_is_slow();
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let first = rig.deliver(body.clone());
let again = async {
tokio::time::sleep(Duration::from_millis(800)).await;
rig.deliver(body.clone()).await
};
let ((c1, r1, t1), (c2, r2, t2)) = tokio::join!(first, again);
assert_eq!((c1, c2), (200, 200), "{r1} / {r2}");
assert!(t1 < ANSWER_WITHIN && t2 < ANSWER_WITHIN, "{} ms / {} ms: {r1} / {r2}", t1.as_millis(), t2.as_millis());
assert!(rig.paid_within(&started.reference, SLOW_PUSH + Duration::from_secs(8)).await, "the grant never landed");
// Let anything a second delivery might have started finish.
tokio::time::sleep(Duration::from_secs(2)).await;
assert_eq!(rig.product.push_attempts().len(), 1, "a retried event pushed the entitlement again");
assert_eq!(rig.pay.captures() - captures_before, 1, "a retried event captured again");
let kinds = rig.ledger_kinds(&started.reference).await;
assert_eq!(kinds.iter().filter(|k| k.contains("EntitlementPushed") || k.contains("entitlement_pushed")).count(), 1, "{kinds:?}");
// And once it is granted, the same event once more is answered from the
// kept row at once and moves nothing.
let (c3, r3, t3) = rig.deliver(body).await;
assert_eq!(c3, 200, "{r3}");
assert!(t3 < ANSWER_WITHIN, "{r3}");
assert_eq!(rig.product.push_attempts().len(), 1, "{r3}");
let rows = rig.state.callbacks.rows().unwrap();
assert_eq!(rows.len(), 1, "one event id, one row, however often it is delivered: {rows:?}");
// The two deliveries that met an OPEN attempt are counted on the row; the
// one after the close is answered by the ledger's closed attempt before the
// inbox is asked (`ignored: no open attempt`), as every late delivery is.
assert_eq!(rows[0].deliveries, 2, "{rows:?}");
assert!(r3.contains("no open attempt"), "{r3}");
}
/// ★ **A push the product refuses for now is retried by monetize, not by the
/// vendor.** The webhook has answered 200 — the vendor will not deliver again
/// — so the retry is ours: without a redelivery and without a poller tick.
///
/// RED before the fix: the attempt stayed open until the next poller tick
/// (minutes on an estate), and no line said a retry was due.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_push_refused_for_now_is_retried_without_a_redelivery() {
let rig = Rig::up().await;
let started = rig.start("nordisk").await;
rig.pay.authorize(&started.handle_id);
*rig.product.fail_push.lock().unwrap() = true;
let (code, reply, took) = rig.deliver(CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized")).await;
assert_eq!(code, 200, "{reply}");
assert!(took < ANSWER_WITHIN, "{reply}");
assert_eq!(rig.status(&started.reference).await, PurchaseState::Pending as i32);
// The product comes back; nobody delivers anything and no tick runs.
tokio::time::sleep(Duration::from_millis(500)).await;
*rig.product.fail_push.lock().unwrap() = false;
assert!(rig.paid_within(&started.reference, Duration::from_secs(12)).await, "a failed push was never retried without a redelivery");
assert_eq!(rig.product.pushed().len(), 1);
assert!(rig.product.push_attempts().len() >= 2, "the retry is a second attempt: {}", rig.product.push_attempts().len());
tokio::time::sleep(Duration::from_millis(300)).await;
let rows = rig.state.callbacks.rows().unwrap();
assert!(rows[0].attempts >= 2 && rows[0].is_done(), "the kept event counts its attempts: {rows:?}");
}
/// A sink that takes its time — gunnar's creates the principal and then pushes
/// the entitlement, on the same box a growth is holding.
struct SlowSink;
impl monetize_pay::SignupSink for SlowSink {
fn paid(&self, s: &PaidSignup) -> Result<monetize_pay::SignupOutcome, monetize_pay::SignupError> {
std::thread::sleep(SLOW_PUSH);
Ok(monetize_pay::SignupOutcome::Created { namespace: s.username.clone(), username: s.username.clone() })
}
fn name_verdict(&self, _username: &str) -> Result<monetize_pay::NameVerdict, monetize_pay::SignupError> {
Ok(monetize_pay::NameVerdict::Free)
}
}
/// ★ **A paid signup whose sink is slow is answered at once too** — the same
/// pattern on the other money arm of the same route.
///
/// RED before the fix: the answer waited for the sink.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_paid_signup_whose_sink_is_slow_is_answered_at_once_and_created_after() {
let pay = CallbackPay::new(SECRET);
let mut state = AppState::in_memory(Registry::new().sells("gunnar", pay.clone()));
state.signup_plans = [(
"gunnar/10gb-1m".to_string(),
SignupPlan { amount_minor: 10_000, currency: Currency::Sek, cap_bytes: 10_737_418_240, period_days: 30 },
)]
.into();
state.signup_sink = Arc::new(SlowSink);
let state = Arc::new(state);
let server = TestServer::spawn(state.clone(), BEARER).await;
let s = PaidSignup {
vendor: "mockpay".into(),
session: "cs_slow".into(),
reference: "pi_1".into(),
email: "bob@example.com".into(),
username: "bob".into(),
product: "gunnar".into(),
plan: "gunnar/10gb-1m".into(),
amount: Money { minor: 10_000, currency: Currency::Sek },
tax: Money { minor: 0, currency: Currency::Sek },
paid_at_unix_ms: 1_788_436_800_000,
customer_country: None,
};
state.signups.record_minted(&s.vendor, &s.session, None).expect("minted at Start (ST-7)");
let (addr, headers, body) = (server.addr, pay.signed_headers(), CallbackPay::signup_body(&s));
let (code, reply, took) = tokio::task::spawn_blocking(move || {
let t = Instant::now();
let (c, r) = http_post(addr, "/callback/mockpay", &headers, &body).unwrap();
(c, r, t.elapsed())
})
.await
.unwrap();
assert_eq!(code, 200, "{reply}");
assert!(took < ANSWER_WITHIN, "the signup webhook waited {} ms for the sink: {reply}", took.as_millis());
assert!(reply.contains("PENDING"), "{reply}");
// Kept before the answer.
assert!(state.signups.get("mockpay", "cs_slow").unwrap().is_some(), "the signup was not recorded before the answer");
let t = Instant::now();
loop {
let rec = state.signups.get("mockpay", "cs_slow").unwrap().unwrap();
if matches!(rec.state, monetize_server::signup::SignupState::Created { .. }) {
break;
}
assert!(t.elapsed() < SLOW_PUSH + Duration::from_secs(8), "the signup was never created after the answer: {:?}", rec.state);
tokio::time::sleep(Duration::from_millis(200)).await;
}
}