crates/monetize-server/tests/callback_refusals.rs
//! **A refused callback leaves a row, and nothing else.**
//!
//! The failure this file is the guard for was measured on the front
//! 2026-09-14: the box's clock ran two hours fast, so every genuine Stripe
//! delivery carried a signed `t=` outside the 300 s tolerance,
//! `parse_callback` refused all of them, and monetize recorded ABSOLUTELY
//! NOTHING. The money was at the vendor. The only trace was one journald line
//! per delivery and nobody was reading journald.
//!
//! Over the wire, through the real router, with the real webhook route — the
//! refusal path is HTTP and a unit test of the store alone would not have
//! noticed a handler that never called it.
//!
//! Three things are held here, and the second is the one that matters most:
//!
//! 1. a refused callback IS recorded, with our facts in it;
//! 2. a recorded refusal creates NO ledger row, no attempt, no entitlement —
//! the record is a screen for a human and never an input;
//! 3. the bound really bounds: a flood costs a fixed number of rows and a
//! counter, and the counter is what says a flood happened.
use std::net::SocketAddr;
use std::sync::Arc;
use monetize_proto::callback_admin_client::CallbackAdminClient;
use monetize_proto::ledger_client::LedgerClient;
use monetize_proto::query_ledger_request::By;
use monetize_proto::{self as pb};
use monetize_server::refusals::CAPACITY;
use monetize_server::registry::Registry;
use monetize_server::state::AppState;
use monetize_server::testing::{http_post, CallbackPay, TestServer, SIGNATURE_HEADER};
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";
/// A server with the callback vendor linked and NO product: every test here is
/// about the webhook's refusal arms, which are reached before any product is,
/// and a rig that stood up a product would be asserting less with more.
struct Rig {
pay: CallbackPay,
state: Arc<AppState>,
server: TestServer,
channel: Channel,
}
impl Rig {
async fn up() -> Rig {
let pay = CallbackPay::new(SECRET);
let state = Arc::new(AppState::in_memory(Registry::new().sells("gunnar", pay.clone())));
let server = TestServer::spawn(state.clone(), BEARER).await;
let channel = Channel::from_shared(server.endpoint()).unwrap().connect().await.unwrap();
Rig { pay, state, server, channel }
}
fn addr(&self) -> SocketAddr {
self.server.addr
}
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 callback(&self, path: &str, headers: Vec<(String, String)>, body: Vec<u8>) -> (u16, String) {
let (addr, path) = (self.addr(), path.to_owned());
tokio::task::spawn_blocking(move || http_post(addr, &path, &headers, &body).unwrap()).await.unwrap()
}
/// What the CONSOLE would see: the list read over the same rpc, through the
/// same bearer. Asserting on the server's own struct instead would leave the
/// wire untested, and the wire is where the console is.
async fn refusals(&self) -> pb::ListRefusalsResponse {
CallbackAdminClient::new(self.channel.clone())
.list_refusals(self.authed(pb::ListRefusalsRequest { limit: 0 }))
.await
.expect("the refusals list is served")
.into_inner()
}
/// Every ledger row there is, by any reference.
async fn ledger_rows(&self) -> Vec<pb::LedgerRow> {
LedgerClient::new(self.channel.clone())
.query(self.authed(pb::QueryLedgerRequest { by: Some(By::Recent(1000)) }))
.await
.unwrap()
.into_inner()
.rows
}
fn unsigned(&self) -> Vec<(String, String)> {
vec![("content-type".into(), "application/json".into())]
}
fn forged(&self) -> Vec<(String, String)> {
vec![(SIGNATURE_HEADER.into(), "not-the-secret".into()), ("content-type".into(), "application/json".into())]
}
}
/// (r) **The incident, reproduced: a delivery that does not verify is refused
/// 401 AND recorded, with our own facts in the row.**
///
/// RED before the change: the status is the same 401 and `refusals` is empty —
/// which is exactly the morning this was written for. Asserting the FIELDS and
/// not merely that a row exists: a record that stored a row of blanks would
/// satisfy a length check and tell an operator nothing.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_callback_that_does_not_verify_is_refused_and_recorded() {
let rig = Rig::up().await;
let body = CallbackPay::callback_body("gunnar/bob/2026-09-14", "res-1", "captured");
let (code, reply) = rig.callback("/callback/mockpay", rig.forged(), body.clone()).await;
assert_eq!(code, 401, "the status and the reason for it are untouched: {reply}");
assert!(reply.contains("callback did not verify"), "and so is what the caller is told: {reply}");
let out = rig.refusals().await;
assert_eq!(out.refusals.len(), 1, "the refusal survived the request");
let row = &out.refusals[0];
// OUR facts: which of OUR vendors was hit, how big the body was, the nth
// refusal since we started, and our own word for our own arm.
assert_eq!((row.vendor.as_str(), row.kind.as_str()), ("mockpay", "unverified"));
assert_eq!(row.body_bytes, body.len() as u64, "the size of what arrived, not the body itself");
assert_eq!(row.refusal_seq, 1);
assert!(row.at_unix_ms > 0, "our clock, at the moment we received it");
assert!(!row.verified, "the caller proved nothing and the row says so as a VALUE");
// Our sentence, and it is the vendor's parser's reason rather than a
// paraphrase: an operator has to be able to tell a wrong secret from a
// clock that has drifted, and only this string does that.
assert!(row.detail.contains("signature header missing or wrong"), "{}", row.detail);
assert_eq!((out.total_since_start, out.kept, out.capacity), (1, 1, CAPACITY as u64));
assert_eq!((out.first_at_unix_ms, out.last_at_unix_ms), (row.at_unix_ms, row.at_unix_ms));
// NOTHING ELSE was echoed: not the body, not the signature header, not the
// reference or handle the caller asserted. The whole row is searched, so a
// field added later that leaked one of them fails here too.
let whole = format!("{row:?}");
for secret in [SECRET, "not-the-secret", "gunnar/bob/2026-09-14", "res-1"] {
assert!(!whole.contains(secret), "an unauthenticated caller's {secret:?} must not reach an operator's screen: {whole}");
}
// A delivery with no signature at all is the same arm, and the counter moves.
let (code, _) = rig.callback("/callback/mockpay", rig.unsigned(), body).await;
assert_eq!(code, 401);
let out = rig.refusals().await;
assert_eq!((out.refusals.len(), out.total_since_start), (2, 2));
assert_eq!(out.refusals[0].refusal_seq, 2, "newest first");
}
/// (r) **A recorded refusal creates NOTHING.** No ledger row, no attempt, no
/// entitlement, no signup — the row is a screen for a human, and a record that
/// had become an input would be a way for a stranger to move money.
///
/// The ledger is asserted EMPTY rather than unchanged-from-a-baseline, because
/// this rig starts with nothing in it: a callback that had created an attempt
/// out of a stranger's assertions would show up here as the only row there is.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_recorded_refusal_creates_no_ledger_row_and_no_tenant() {
let rig = Rig::up().await;
assert!(rig.ledger_rows().await.is_empty(), "the rig starts with an empty ledger or this proves nothing");
// Every refusal arm a stranger can reach, one after another.
let signed_shape = CallbackPay::callback_body("gunnar/bob/2026-09-14", "res-1", "captured");
rig.callback("/callback/mockpay", rig.forged(), signed_shape).await;
// Verified, and its payload is not one the parser reads. The 400 arm: a
// delivery this server threw away, which the ledger will never hear of either.
let (code, _) = rig.callback("/callback/mockpay", rig.pay.signed_headers(), b"{\"garbled\":\"evt_1\"}".to_vec()).await;
assert_eq!(code, 400);
// A path naming no linked vendor: 404, and deliberately recorded NOWHERE —
// the segment is the caller's to choose, and a store that kept it would let
// anyone write a string of their choosing onto an operator's screen.
let (code, _) = rig.callback("/callback/not-a-vendor", rig.forged(), b"{}".to_vec()).await;
assert_eq!(code, 404);
assert!(rig.ledger_rows().await.is_empty(), "a refused callback must not open an attempt");
assert!(rig.state.changes().is_empty(), "…nor move an entitlement");
assert_eq!(rig.state.signups.list(0).unwrap().len(), 0, "…nor make a signup");
let out = rig.refusals().await;
assert_eq!(out.refusals.len(), 2, "two arms recorded, and the unknown-vendor 404 is not one of them");
assert_eq!(out.refusals.iter().map(|r| r.kind.as_str()).collect::<Vec<_>>(), vec!["not-understood", "unverified"]);
assert!(out.refusals.iter().all(|r| !r.verified), "neither caller proved anything");
assert!(!format!("{:?}", out.refusals).contains("not-a-vendor"), "the caller's own path segment is nowhere in the record");
}
/// (r) **The bound bounds, over the wire, and the counter does not.**
///
/// A flood of forged callbacks is something anyone on the internet can start,
/// so the memory it costs has to be a constant. RED without the ring's
/// `pop_front`: `kept` climbs with the flood and this is a disk-fill bug with a
/// socket in front of it.
///
/// Both halves are asserted. `kept == CAPACITY` alone would also pass for a
/// store that dropped the NEWEST refusals — which would throw away the outage
/// an operator is looking at — so the newest row's sequence is checked too.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_flood_of_forged_callbacks_is_bounded_and_the_counter_says_so() {
let rig = Rig::up().await;
let over = CAPACITY + 17;
for _ in 0..over {
let (code, _) = rig.callback("/callback/mockpay", rig.forged(), b"{}".to_vec()).await;
assert_eq!(code, 401);
}
let out = rig.refusals().await;
assert_eq!(out.total_since_start, over as u64, "every refusal is counted, including the ones the ring dropped");
assert_eq!((out.kept, out.capacity), (CAPACITY as u64, CAPACITY as u64), "and the ring is the whole of the memory it costs");
assert_eq!(out.refusals.len(), CAPACITY, "the list is the ring");
assert_eq!(out.refusals[0].refusal_seq, over as u64, "the NEWEST is kept: the oldest went, not the latest");
assert_eq!(out.refusals.last().unwrap().refusal_seq, 18, "…and the first 17 are gone");
assert!(rig.ledger_rows().await.is_empty(), "a flood writes nothing to the ledger either");
}
/// (r) **A VERIFIED callback naming a handle the ledger never minted is
/// recorded as the VENDOR's word, not a stranger's.**
///
/// The loudest row this panel can show: the signature stood behind it, so a
/// charge may exist against a handle we did not reserve. 409 unchanged — it is
/// the vendor's own delivery and the status reasoning in `callback.rs` is
/// untouched — and `verified` is what lets an operator tell this row from the
/// unauthenticated ones beside it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_verified_callback_naming_another_handle_is_recorded_as_the_vendors_word() {
let rig = Rig::up().await;
// An attempt opened the way the transaction opens one, so the 409 arm is
// reached against a state the ledger can really be in.
let reference = "gunnar/bob/2026-09-14";
rig.state
.ledger
.append(
monetize::testing::NOW,
monetize::ledger::Step::Reserved {
reference: reference.into(),
handle: monetize_pay::ChargeHandle { vendor: "mockpay".into(), id: "res-ours".into() },
purchase: Box::new(monetize::testing::purchase("bob")),
},
)
.expect("the ledger opens the attempt");
let rows_before = rig.ledger_rows().await.len();
let (code, reply) = rig
.callback("/callback/mockpay", rig.pay.signed_headers(), CallbackPay::callback_body(reference, "res-theirs", "captured"))
.await;
assert_eq!(code, 409, "the status is untouched: {reply}");
let out = rig.refusals().await;
assert_eq!(out.refusals.len(), 1);
let row = &out.refusals[0];
assert_eq!(row.kind, "handle-mismatch");
assert!(row.verified, "this one got past parse_callback: the vendor signed it, and the row must not read as a stranger's");
assert!(row.detail.contains("res-ours") && row.detail.contains("res-theirs"), "both handles, or the row cannot be acted on: {}", row.detail);
// And it still moved nothing: the attempt is exactly as it was.
assert_eq!(rig.ledger_rows().await.len(), rows_before, "a 409 appends no row of its own");
assert!(rig.state.changes().is_empty());
}
/// (r) **The refusals the console must NOT be shown**, because each already has
/// a truer home and a panel of noise is a panel nobody reads.
///
/// A verified delivery about a reference we do not hold is a late or duplicate
/// webhook — the poller and the webhook race by design — and it is answered 200
/// precisely because nothing is wrong. Recording it would bury the refusals
/// under the vendor's ordinary traffic, which is how the one row that matters
/// gets scrolled past.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_verified_event_about_nothing_we_hold_is_a_receipt_and_not_a_refusal() {
let rig = Rig::up().await;
let (code, reply) = rig
.callback("/callback/mockpay", rig.pay.signed_headers(), CallbackPay::callback_body("gunnar/nobody/2026-09-14", "res-1", "captured"))
.await;
assert_eq!(code, 200, "{reply}");
assert!(reply.contains("ignored"), "{reply}");
let out = rig.refusals().await;
assert_eq!((out.refusals.len(), out.total_since_start), (0, 0), "nothing happened and nothing was wrong: not a row");
}