crates/monetize-server/src/refusals.rs
//! **The refused callbacks — the trace a failed purchase used to leave nowhere.**
//!
//! MEASURED 2026-09-14 on the front: the box's clock ran two hours fast, so
//! every genuine Stripe delivery carried a signed `t=` outside the 300 s
//! tolerance and `parse_callback` refused all of them. The money was at the
//! vendor and monetize held nothing. The only trace was one journald line per
//! delivery, and nobody was reading journald. The console could not have shown
//! it because nothing durable was written — the refusal arm counted a
//! `functional_status` and returned 401.
//!
//! This is the record. [`crate::callback`] writes it; `CallbackAdmin.ListRefusals`
//! reads it; the console draws it beside the paid signups, where an operator
//! already looks for "money arrived and nothing happened".
//!
//! # It is UNAUTHENTICATED INPUT, and the shape says so
//!
//! `POST /callback/<vendor>` is on the public edge and carries no bearer — the
//! vendor's signature is its credential, and a refused callback is precisely
//! one that did not present it. **Anyone on the internet can put a row here.**
//! So:
//!
//! * **Nothing here is ever read back as authority.** A row creates no ledger
//! attempt, no tenant, no entitlement and no signup; it is not an input to
//! any decision this server makes. It is a screen for a human.
//! * **What is stored is mostly OURS**: the time by our clock, which of our own
//! linked vendors the endpoint belongs to, which refusal arm our code took,
//! how many bytes the body was, and the nth-since-boot counter. A path naming
//! no linked vendor is answered 404 and recorded nowhere, so even the vendor
//! name cannot be chosen by the caller.
//! * **The one tainted field is [`Refused::detail`]** — our sentence, which for
//! the Stripe arm quotes the caller's own `t=` back. It is bounded to
//! [`DETAIL_CEILING`] characters and stripped of control characters here, at
//! the one place a row is made, and [`Refused::verified`] tells a reader
//! whether the vendor's signature stood behind it at all.
//! * **Nothing else off the wire is kept**: not the body, not one header, not
//! the signature, not an event id, not an amount, not an address. A store
//! that echoed an attacker's strings back at an operator would be the second
//! bug wearing the first one's clothes.
//!
//! # The bound: the last [`CAPACITY`] refusals, and a counter that keeps counting
//!
//! A ring, in memory, capacity 128, oldest dropped — the same decision
//! [`crate::bus`] makes about a slow subscriber, and for the same reason: a
//! flood a stranger can start must not be able to grow this process. What a
//! flood costs is O(1) memory and a LARGE NUMBER beside a short list, because
//! `total_since_start` keeps counting past the ring. One forged POST a
//! millisecond for a day moves that counter to eighty million and moves the
//! memory by nothing.
//!
//! 128 is chosen, not arbitrary: the incident it is for is "every delivery for
//! some hours", and Stripe retries a failed delivery a handful of times per
//! event. A morning of refusals is tens of rows, so the ring holds the whole
//! incident; a flood overruns it, which is what the counter is for, and reading
//! 128 lines is something a person can actually do.
//!
//! # In memory, deliberately — and a restart clears it
//!
//! Nothing here is persisted, for the same reason [`crate::health`] persists
//! nothing and a stronger one. The ledger is the durable record of MONEY and it
//! is written only by authenticated paths; a refusal moved no money, and its
//! durable counterpart is at the vendor, in the delivery table this row exists
//! to send an operator to. Giving unauthenticated input a redb table on an
//! appliance that also holds a bankgiro ledger would put a stranger's POST rate
//! in charge of our disk and our write amplification — a bound we would then
//! have to defend with a sweeper instead of by construction.
//!
//! The cost is honest and small: a restart empties the ring, and the console
//! says "since this monetize started" rather than implying a history it does
//! not have. The incident it was built for lasted hours with the service up.
use std::collections::VecDeque;
use std::sync::Mutex;
use monetize_proto as pb;
/// **How many refused callbacks are kept.** See the module doc for why 128 and
/// why the oldest goes rather than the newest.
pub const CAPACITY: usize = 128;
/// **How long a refusal's `detail` may be**, in characters. Our sentences are
/// well under it; the ceiling is there because part of one can be the caller's
/// (a `t=` quoted back), and an unbounded string from a stranger is a row that
/// can be made any size. Cut text ends in `…` so a reader knows it was cut.
pub const DETAIL_CEILING: usize = 200;
/// The most rows one [`RefusalLog::recent`] will return when the caller names
/// no limit. Equal to [`CAPACITY`]: there is nothing behind the ring to page
/// into, and a ceiling smaller than the store would hide rows with no way to
/// ask for them.
pub const LIST_CEILING: usize = CAPACITY;
/// **Which refusal it was** — our classification of our own arm, never a word
/// off the wire. One variant per [`crate::callback`] arm that ends with money
/// possibly at the vendor and nothing here.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Refusal {
/// `parse_callback` said `Unverified`: the signature was missing, wrong, or
/// outside the vendor's replay window. **The caller proved nothing**, and
/// this is the arm the 2026-09-14 clock incident took for every genuine
/// delivery. Answered 401.
Unverified,
/// `parse_callback` refused for any other reason — a body that is not the
/// shape this vendor's parser reads. Answered 400. The caller proved
/// nothing here either: a parse can fail before or after a signature check
/// and the arm cannot tell which, so it claims the weaker of the two.
NotUnderstood,
/// **VERIFIED, and it names a handle the ledger did not mint for that
/// reference.** The vendor's signature stood behind this one, so it is the
/// vendor's word and not a stranger's — and it is the loudest of the three:
/// somebody may have been charged against a handle we never reserved.
/// Answered 409.
HandleMismatch,
}
impl Refusal {
/// The wire and JSON spelling. One writer, so the proto row, the console's
/// JSON and a log line cannot disagree.
pub fn name(self) -> &'static str {
match self {
Refusal::Unverified => "unverified",
Refusal::NotUnderstood => "not-understood",
Refusal::HandleMismatch => "handle-mismatch",
}
}
/// **Did the vendor's own signature prove the delivery?** The one bit that
/// separates "a stranger posted this" from "our vendor said this", and the
/// reason it is a field rather than a comment: the console renders it, and
/// an operator reading a `detail` has to know whose words are inside it.
pub fn verified(self) -> bool {
matches!(self, Refusal::HandleMismatch)
}
}
/// One refused callback, as it is kept and as it is shown.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct Refused {
/// **Our clock**, when WE received it. Not any time from the body: the
/// whole incident was a disagreement about what time it was, and a record
/// that took the caller's word for that would have been unreadable.
pub at_unix_ms: u64,
/// The LINKED vendor whose endpoint was hit. Ours, not the caller's — the
/// path segment matched the registry or the request was a 404 that is
/// recorded nowhere.
pub vendor: String,
pub refusal: Refusal,
/// **Our refusal, in our words — and it may quote the caller.** Bounded and
/// stripped by [`bounded_detail`] on the way in. Read it next to
/// [`Refused::verified`].
pub detail: String,
/// How big the body was. A fact about the connection, and the cheapest way
/// to tell a genuine vendor delivery apart from an empty probe.
pub body_bytes: u64,
/// The nth refusal since this server started, from 1. A row whose `seq` is
/// far above the number of rows shown is a ring that overran: refusals are
/// missing between them, and the count says how many.
pub seq: u64,
}
impl Refused {
/// Whether the vendor's signature stood behind this delivery. Derived from
/// the arm, so it cannot be set to something the arm disagrees with.
pub fn verified(&self) -> bool {
self.refusal.verified()
}
/// The wire row. The SAME message the bus event carries (one shape), so a
/// console that learned a refusal live and one that read the list later are
/// rendering identical fields.
pub fn to_pb(&self) -> pb::CallbackRefused {
pb::CallbackRefused {
at_unix_ms: self.at_unix_ms,
kind: self.refusal.name().to_owned(),
vendor: self.vendor.clone(),
verified: self.verified(),
detail: self.detail.clone(),
body_bytes: self.body_bytes,
refusal_seq: self.seq,
}
}
}
/// What the ring holds, for the console's header and for a test.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub struct RefusalStats {
/// **Every refusal since this server started**, including the ones the ring
/// has dropped. The number a flood shows up in.
pub total: u64,
/// How many are still kept.
pub kept: u64,
pub capacity: u64,
/// The oldest and newest KEPT refusal's instant; 0 when nothing is kept.
pub first_at_unix_ms: u64,
pub last_at_unix_ms: u64,
}
/// **Bound a refusal's words and take the control characters out.**
///
/// Called at the one place a row is made, so there is no path by which an
/// unbounded or newline-carrying string reaches the store. Control characters
/// become spaces rather than being dropped: a caller must not be able to make
/// one row look like two, and deleting them silently would let `a\nb` and `ab`
/// be the same stored string.
///
/// The console renders this through Leptos, which escapes text nodes, so markup
/// is not the exposure here — the shape of a log line is.
pub fn bounded_detail(detail: &str) -> String {
let mut out = String::with_capacity(detail.len().min(DETAIL_CEILING * 4));
for c in detail.chars().take(DETAIL_CEILING) {
out.push(if c.is_control() { ' ' } else { c });
}
// The ellipsis only where something was actually left behind, so a sentence
// that happens to be exactly the ceiling does not read as truncated.
if detail.chars().nth(DETAIL_CEILING).is_some() {
out.push('…');
}
out
}
/// **The ring.** Bounded by construction, in memory, and never read back as
/// authority — see the module doc.
#[derive(Debug)]
pub struct RefusalLog {
inner: Mutex<Inner>,
capacity: usize,
}
#[derive(Debug, Default)]
struct Inner {
recent: VecDeque<Refused>,
total: u64,
}
impl Default for RefusalLog {
fn default() -> Self {
RefusalLog::new(CAPACITY)
}
}
impl RefusalLog {
/// A log holding at most `capacity` rows. `0` is raised to 1: a store that
/// silently kept nothing would make every test of this file pass.
pub fn new(capacity: usize) -> Self {
RefusalLog {
inner: Mutex::new(Inner::default()),
capacity: capacity.max(1),
}
}
/// **Record one refusal and hand back the row that was kept.**
///
/// `at_unix_ms` is the caller's clock reading — the store has no opinion
/// about the time separate from the data, exactly as
/// [`crate::state::UsageStore`] has none. Never fails: a refusal that could
/// not be recorded because a lock was poisoned would be the original bug
/// again, so the lock is taken through its poison.
pub fn record(
&self,
at_unix_ms: u64,
vendor: &str,
refusal: Refusal,
detail: &str,
body_bytes: u64,
) -> Refused {
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.total += 1;
let row = Refused {
at_unix_ms,
vendor: vendor.to_owned(),
refusal,
detail: bounded_detail(detail),
body_bytes,
seq: inner.total,
};
if inner.recent.len() == self.capacity {
inner.recent.pop_front();
}
inner.recent.push_back(row.clone());
row
}
/// **The kept refusals, NEWEST FIRST**, at most `limit` (0 = the server's
/// [`LIST_CEILING`]). Newest first because the question is always "what just
/// happened", and a list an operator has to scroll to the bottom of to see
/// the current outage is a list nobody reads.
pub fn recent(&self, limit: usize) -> Vec<Refused> {
let limit = if limit == 0 {
LIST_CEILING
} else {
limit.min(LIST_CEILING)
};
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.recent.iter().rev().take(limit).cloned().collect()
}
pub fn stats(&self) -> RefusalStats {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
RefusalStats {
total: inner.total,
kept: inner.recent.len() as u64,
capacity: self.capacity as u64,
first_at_unix_ms: inner.recent.front().map(|r| r.at_unix_ms).unwrap_or(0),
last_at_unix_ms: inner.recent.back().map(|r| r.at_unix_ms).unwrap_or(0),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn log() -> RefusalLog {
RefusalLog::new(4)
}
/// (r) **The bound actually bounds, and the counter does not.**
///
/// A thousand forged callbacks cost four rows of memory and one number.
/// RED without the `pop_front`: `kept` is 1000 and this is a disk-fill bug
/// with a socket in front of it. Asserting BOTH halves, because a ring that
/// dropped the NEWEST would also keep four rows — and would have thrown
/// away the outage the operator is looking at.
#[test]
fn a_flood_costs_four_rows_and_a_large_number() {
let log = log();
for i in 1..=1000u64 {
log.record(i, "stripe", Refusal::Unverified, "no", 10);
}
let stats = log.stats();
assert_eq!(
(stats.total, stats.kept, stats.capacity),
(1000, 4, 4),
"the ring is the bound; the counter keeps counting"
);
let rows = log.recent(0);
assert_eq!(rows.len(), 4);
// Newest first, and the four kept are the LAST four — the oldest went.
assert_eq!(
rows.iter().map(|r| r.seq).collect::<Vec<_>>(),
vec![1000, 999, 998, 997]
);
assert_eq!((stats.first_at_unix_ms, stats.last_at_unix_ms), (997, 1000));
// `seq` far above the rows shown is how a reader sees the hole.
assert_eq!(
rows[0].seq - rows.len() as u64,
996,
"996 refusals are not in this list and the numbers say so"
);
}
/// (r) **A caller's words are bounded and cannot forge a second line.**
///
/// The Stripe refusal quotes the caller's own `t=` back, so `detail` is the
/// one tainted field. RED without [`bounded_detail`]: the newline survives
/// and one row prints as two, and the length is whatever the caller sent.
#[test]
fn a_detail_is_cut_to_the_ceiling_and_holds_no_control_characters() {
let log = log();
let nasty = format!(
"stripe: webhook timestamp {} outside tolerance",
"9".repeat(1000)
);
let row = log.record(1, "stripe", Refusal::Unverified, &nasty, 512);
assert_eq!(
row.detail.chars().count(),
DETAIL_CEILING + 1,
"cut, with the ellipsis that says so"
);
assert!(row.detail.ends_with('…'));
let row = log.record(
2,
"stripe",
Refusal::Unverified,
"first\nsecond\r\tthird\u{0}",
0,
);
assert_eq!(
row.detail, "first second third ",
"control characters become spaces, never vanish"
);
assert!(!row.detail.contains('\n'), "one refusal is one line");
// A short, clean sentence is carried verbatim: the bound must not be a
// paraphrase of the words an operator is meant to read.
let whole =
"stripe: webhook timestamp 1789384242 outside tolerance (300s from now 1789391442)";
assert_eq!(
log.record(3, "stripe", Refusal::Unverified, whole, 1)
.detail,
whole
);
}
/// (r) **Whose word each row is, as a VALUE.** The 401 and 400 arms are a
/// stranger's; the 409 arm carries the vendor's signature. A console that
/// had to read that off the prose would get it wrong on the first refusal
/// nobody anticipated.
#[test]
fn verified_is_the_arm_and_not_a_claim_the_caller_can_make() {
assert!(!Refusal::Unverified.verified());
assert!(!Refusal::NotUnderstood.verified());
assert!(
Refusal::HandleMismatch.verified(),
"it got past parse_callback: the vendor signed it"
);
assert_eq!(
[
Refusal::Unverified.name(),
Refusal::NotUnderstood.name(),
Refusal::HandleMismatch.name()
],
["unverified", "not-understood", "handle-mismatch"]
);
let log = log();
let row = log.record(
7,
"stripe",
Refusal::HandleMismatch,
"names another handle",
3,
);
assert!(row.verified() && row.to_pb().verified);
assert_eq!(row.to_pb().kind, "handle-mismatch");
assert_eq!(
(
row.to_pb().at_unix_ms,
row.to_pb().vendor.as_str(),
row.to_pb().body_bytes
),
(7, "stripe", 3)
);
}
/// (r) An empty log is empty, not a row of zeroes — and `limit` cuts from
/// the newest end, which is the end that matters.
#[test]
fn an_empty_log_says_nothing_and_a_limit_keeps_the_newest() {
let log = log();
assert!(log.recent(0).is_empty());
assert_eq!(
log.stats(),
RefusalStats {
total: 0,
kept: 0,
capacity: 4,
first_at_unix_ms: 0,
last_at_unix_ms: 0
}
);
for i in 1..=3u64 {
log.record(
i * 10,
"mockpay",
Refusal::NotUnderstood,
"body is not JSON",
i,
);
}
assert_eq!(
log.recent(2)
.iter()
.map(|r| r.at_unix_ms)
.collect::<Vec<_>>(),
vec![30, 20]
);
assert_eq!(log.recent(99).len(), 3, "a limit past the ring is the ring");
}
}