crates/monetize-server/src/inbox.rs
//! **The vendor's deliveries, kept before they are acted on** — and the one
//! grant per reference that acts on them (stinkies money-webhook-504-in-growth).
//!
//! # Why the webhook no longer waits for the grant
//!
//! MEASURED 2026-09-30 on the d-upgrade mock (money-1, M3 DK): the completed
//! event for a paid +10 GiB order arrived 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; monetize granted the order
//! 29 s later anyway. Every non-2xx is a failed delivery the vendor counts
//! against the endpoint, and enough of them disable it.
//!
//! So a VERIFIED delivery is now:
//!
//! 1. kept here, durably, keyed `(vendor, event id)` — one row however often
//! it is delivered ([`CallbackInbox::receive`]);
//! 2. handed to ONE grant per reference ([`Flights`]) on its own thread, which
//! retries what clears on its own — the product down, a push refused for
//! now, a vendor that did not answer — with a backoff, and says every attempt;
//! 3. answered within [`ANSWER_WITHIN`]: the grant's own verdict when it was
//! that quick (an ordinary sale reads exactly as it always did), else `200
//! {reference, state: PENDING, accepted: true}`.
//!
//! # Why a retried event cannot double-grant
//!
//! Three locks, from the outside in:
//!
//! * **the row**: a delivery of an event whose row is DONE is answered from the
//! row and moves nothing;
//! * **the flight**: one resume per reference at a time in this process — the
//! webhook's grant, a redelivery's, the poller's tick and a `Purchase.Status`
//! all go through `services::purchase::resume_reference`, which takes it. The
//! loser answers PENDING and touches nothing. Before it, a redelivery that
//! landed while the first grant was still pushing resumed the same open
//! attempt and pushed the entitlement a second time;
//! * **the ledger**: a closed attempt is never resumed (`reconcile_reference`
//! reads `is_open` first), and a captured one is never captured again.
//!
//! # What is not here
//!
//! The money's own record stays the LEDGER (the open attempt) and, for a paid
//! signup, `crate::signup::SignupStore`; this file is the delivery's record,
//! and losing it costs nothing the poller does not repair — every tick resumes
//! every open attempt. Its own redb file (`callbacks.redb`), for the reason the
//! other stores are separate: one writer per file.
use std::collections::HashSet;
use std::path::Path;
use std::sync::Mutex;
use std::time::Duration;
use redb::{Database, ReadableTable, TableDefinition};
use crate::store::StoreError;
/// How long a webhook waits for the grant behind it before answering
/// `accepted`. Far inside vm1's proxy (30 s) and the vendor's own delivery
/// timeout; long enough that an ordinary sale answers its verdict.
pub const ANSWER_WITHIN: Duration = Duration::from_secs(2);
/// How often the grant behind a webhook tries again on its own, and the waits
/// between: 2, 4, 8, 16, 30 s. After the last the poller's tick carries on.
pub const RETRY_WAITS: [Duration; 5] =
[Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];
const EVENTS: TableDefinition<(&str, &str), &[u8]> = TableDefinition::new("callback_events");
/// What became of a kept delivery.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum InboxState {
/// Kept, and a grant is on it (or will be, on the next delivery or tick).
Accepted,
/// The last attempt did not finish; `last_error` says why.
Retrying,
/// Finished: the purchase's verdict word (`PAID`, `RELEASED`, …), or
/// `CLOSED` when somebody else had already finished it.
Done { state: String, reason: String },
}
/// One kept delivery.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct InboxRow {
pub vendor: String,
pub event_id: String,
pub reference: String,
pub received_at_unix_ms: i64,
/// How many times the vendor delivered it.
pub deliveries: u32,
/// How many times a grant ran for it.
pub attempts: u32,
pub last_error: String,
pub state: InboxState,
pub updated_at_unix_ms: i64,
}
impl InboxRow {
pub fn is_done(&self) -> bool {
matches!(self.state, InboxState::Done { .. })
}
}
/// [`CallbackInbox::receive`]'s answer.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Received {
/// The first delivery of this event.
Fresh(InboxRow),
/// A redelivery; the row as it stands (its `deliveries` counted).
Again(InboxRow),
}
impl Received {
pub fn row(&self) -> &InboxRow {
match self {
Received::Fresh(r) | Received::Again(r) => r,
}
}
}
/// **The event's id**: the vendor's own (`"id"` at the top of a JSON body —
/// Stripe's `evt_…`, which a retry repeats), else a digest of the verified
/// body (a retry repeats the body byte for byte; only the signature header's
/// timestamp moves). Only ever computed on a body that VERIFIED.
pub fn event_id(body: &[u8]) -> String {
let named = serde_json::from_slice::<serde_json::Value>(body)
.ok()
.and_then(|v| v.get("id").and_then(|i| i.as_str()).map(str::to_owned))
.filter(|id| !id.is_empty() && id.len() <= 128 && id.chars().all(|c| c.is_ascii_graphic()));
named.unwrap_or_else(|| {
// FNV-1a 64: not a MAC — the body is already verified — only a stable name.
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for b in body {
h ^= u64::from(*b);
h = h.wrapping_mul(0x0100_0000_01b3);
}
format!("body:{h:016x}:{}", body.len())
})
}
/// **The kept deliveries.** See the module doc.
pub struct CallbackInbox {
db: Database,
}
impl CallbackInbox {
pub fn open(path: &Path) -> Result<Self, StoreError> {
Self::init(Database::create(path)?)
}
pub fn in_memory() -> Result<Self, StoreError> {
Self::init(Database::builder().create_with_backend(redb::backends::InMemoryBackend::new())?)
}
fn init(db: Database) -> Result<Self, StoreError> {
let txn = db.begin_write()?;
txn.open_table(EVENTS)?;
txn.commit()?;
Ok(Self { db })
}
/// **Keep `(vendor, event_id)`, or count a redelivery of it** — the read
/// and the write are ONE redb transaction, so two deliveries racing on two
/// threads make one row and the loser sees [`Received::Again`].
pub fn receive(&self, vendor: &str, event_id: &str, reference: &str, now_unix_ms: i64) -> Result<Received, StoreError> {
let txn = self.db.begin_write()?;
let out = {
let mut t = txn.open_table(EVENTS)?;
let held: Option<Vec<u8>> = t.get((vendor, event_id))?.map(|v| v.value().to_vec());
let out = match held {
Some(bytes) => {
let mut row: InboxRow = serde_json::from_slice(&bytes)
.map_err(|e| StoreError::corrupted(format!("callback row {vendor}:{event_id} will not parse: {e}")))?;
row.deliveries = row.deliveries.saturating_add(1);
Received::Again(row)
}
None => Received::Fresh(InboxRow {
vendor: vendor.into(),
event_id: event_id.into(),
reference: reference.into(),
received_at_unix_ms: now_unix_ms,
deliveries: 1,
attempts: 0,
last_error: String::new(),
state: InboxState::Accepted,
updated_at_unix_ms: now_unix_ms,
}),
};
let bytes = StoreError::encode("a callback row", out.row())?;
t.insert((vendor, event_id), bytes.as_slice())?;
out
};
txn.commit()?;
Ok(out)
}
/// Change a kept row in place. `Ok(false)` when there is none.
pub fn update(&self, vendor: &str, event_id: &str, f: impl FnOnce(&mut InboxRow)) -> Result<bool, StoreError> {
let txn = self.db.begin_write()?;
let wrote = {
let mut t = txn.open_table(EVENTS)?;
let held: Option<Vec<u8>> = t.get((vendor, event_id))?.map(|v| v.value().to_vec());
match held.and_then(|b| serde_json::from_slice::<InboxRow>(&b).ok()) {
Some(mut row) => {
f(&mut row);
let bytes = StoreError::encode("a callback row", &row)?;
t.insert((vendor, event_id), bytes.as_slice())?;
true
}
None => false,
}
};
txn.commit()?;
Ok(wrote)
}
pub fn get(&self, vendor: &str, event_id: &str) -> Result<Option<InboxRow>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(EVENTS)?;
let row = t.get((vendor, event_id))?.and_then(|v| serde_json::from_slice(v.value()).ok());
Ok(row)
}
/// Every kept row, oldest delivery first.
pub fn rows(&self) -> Result<Vec<InboxRow>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(EVENTS)?;
let mut out: Vec<InboxRow> = Vec::new();
for entry in t.iter()? {
let (_, v) = entry?;
if let Ok(row) = serde_json::from_slice(v.value()) {
out.push(row);
}
}
out.sort_by_key(|r| r.received_at_unix_ms);
Ok(out)
}
}
/// **One resume per reference at a time, in this process.** See the module doc.
#[derive(Default)]
pub struct Flights {
held: Mutex<HashSet<String>>,
}
/// One reference's resume, held by [`Flights::fly`]; dropping it frees it.
pub struct Flight<'a> {
flights: &'a Flights,
key: String,
}
impl Drop for Flight<'_> {
fn drop(&mut self) {
self.flights.held.lock().unwrap_or_else(|p| p.into_inner()).remove(&self.key);
}
}
impl Flights {
/// Take `key`, or `None` when another caller holds it.
pub fn fly(&self, key: &str) -> Option<Flight<'_>> {
let fresh = self.held.lock().unwrap_or_else(|p| p.into_inner()).insert(key.to_owned());
fresh.then(|| Flight { flights: self, key: key.to_owned() })
}
pub fn holds(&self, key: &str) -> bool {
self.held.lock().unwrap_or_else(|p| p.into_inner()).contains(key)
}
}
/// What the first attempt of a grant came to — what a webhook still waiting
/// inside [`ANSWER_WITHIN`] answers.
pub enum Granted {
/// The purchase's verdict (PAID, RELEASED, … or PENDING with its reason).
Verdict(Box<crate::services::purchase::Verdict>),
/// No open attempt: finished before, by somebody else.
Closed,
/// The resume failed; the code and the words.
Failed(tonic::Code, String),
}
/// Is a PENDING verdict worth trying again on our own? Everything that clears
/// by itself — the product down, a push refused for now, the vendor or the
/// cloud not answering — is. A customer who has not paid yet is not: the next
/// event, the return page or the poller's tick moves that one.
fn worth_retrying(reason: &str) -> bool {
reason != crate::services::purchase::IN_FLIGHT && !reason.starts_with("the customer has not paid yet")
}
/// **The grant behind a kept delivery.** Runs on its own thread; the first
/// attempt's answer goes to `first` (the webhook waiting for it, if it still
/// is), then it retries on [`RETRY_WAITS`] whatever clears by itself, and says
/// every attempt. After the last wait it stops: the attempt is still OPEN in
/// the ledger, and the poller's tick resumes every open attempt.
pub fn grant(state: &crate::state::AppState, vendor: &str, event_id: &str, reference: &str, first: Option<std::sync::mpsc::Sender<Granted>>) {
use crate::services::purchase::resume_reference;
let mut first = first;
let waits = std::iter::once(Duration::ZERO).chain(RETRY_WAITS);
let tries = RETRY_WAITS.len() + 1;
for (n, wait) in waits.enumerate() {
if !wait.is_zero() {
std::thread::sleep(wait);
}
let attempt = n + 1;
let now = state.now() as i64;
let got = resume_reference(state, reference);
// What to keep, whether to go on, and what to say.
let (kept, again, said, answer) = match got {
Ok(None) => (InboxState::Done { state: "CLOSED".into(), reason: "no open attempt: finished before".into() }, false, "no open attempt — finished before".to_string(), Granted::Closed),
Ok(Some(v)) if v.state == monetize_proto::PurchaseState::Pending => {
if v.reason == crate::services::purchase::IN_FLIGHT {
// Somebody else holds the reference: they finish it.
if let Some(tx) = first.take() {
let _ = tx.send(Granted::Verdict(Box::new(v)));
}
eprintln!("monetize-server: callback grant {vendor}:{event_id} {reference}: another caller is finishing it — this one stops");
return;
}
let again = worth_retrying(&v.reason) && attempt < tries;
let said = format!("not yet: {}", v.reason);
(if again { InboxState::Retrying } else { InboxState::Accepted }, again, said, Granted::Verdict(Box::new(v)))
}
Ok(Some(v)) => {
let word = format!("{:?}", v.state).to_uppercase();
(InboxState::Done { state: word.clone(), reason: v.reason.clone() }, false, word, Granted::Verdict(Box::new(v)))
}
Err(status) => {
let again = status.code() == tonic::Code::Unavailable && attempt < tries;
let said = format!("failed ({:?}): {}", status.code(), status.message());
(if again { InboxState::Retrying } else { InboxState::Accepted }, again, said, Granted::Failed(status.code(), status.message().to_owned()))
}
};
let done = matches!(kept, InboxState::Done { .. });
let last_error = if done { String::new() } else { said.clone() };
if let Err(e) = state.callbacks.update(vendor, event_id, |r| {
r.attempts = r.attempts.saturating_add(1);
r.state = kept;
r.last_error = last_error;
r.updated_at_unix_ms = now;
}) {
eprintln!("monetize-server: callback grant {vendor}:{event_id} {reference}: the inbox row could not be updated: {e}");
}
if let Some(tx) = first.take() {
let _ = tx.send(answer);
}
if done {
skidbladnir::functional_status("monetize-server/callback", vendor, true, &format!("grant {reference} attempt {attempt}: {said}"));
eprintln!("monetize-server: callback grant {vendor}:{event_id} {reference} attempt {attempt}/{tries}: {said}");
return;
}
if again {
let next = RETRY_WAITS[n];
eprintln!(
"monetize-server: callback grant {vendor}:{event_id} {reference} attempt {attempt}/{tries}: {said} — trying again in {} s",
next.as_secs()
);
continue;
}
let tail = if attempt >= tries { "out of retries here; the attempt stays open and the poller's tick carries on" } else { "not retried here; the attempt stays open for the next event, the return page or the poller's tick" };
skidbladnir::functional_status("monetize-server/callback", vendor, false, &format!("grant {reference} attempt {attempt}: {said} — {tail}"));
eprintln!("monetize-server: callback grant {vendor}:{event_id} {reference} attempt {attempt}/{tries}: {said} — {tail}");
return;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_vendors_own_event_id_names_the_row_and_a_body_without_one_is_named_by_its_bytes() {
assert_eq!(event_id(br#"{"id":"evt_1Q","type":"checkout.session.completed"}"#), "evt_1Q");
let a = event_id(br#"{"reference":"r","handle":"h","event":"authorized"}"#);
assert!(a.starts_with("body:"), "{a}");
assert_eq!(a, event_id(br#"{"reference":"r","handle":"h","event":"authorized"}"#), "a retry repeats the body, so the id");
assert_ne!(a, event_id(br#"{"reference":"r","handle":"h","event":"captured"}"#));
assert!(event_id(b"not json").starts_with("body:"));
}
#[test]
fn one_event_is_one_row_however_often_it_is_delivered() {
let inbox = CallbackInbox::in_memory().unwrap();
assert!(matches!(inbox.receive("stripe", "evt_1", "gunnar/nordisk/x", 10).unwrap(), Received::Fresh(_)));
match inbox.receive("stripe", "evt_1", "gunnar/nordisk/x", 20).unwrap() {
Received::Again(r) => assert_eq!((r.deliveries, r.received_at_unix_ms), (2, 10)),
other => panic!("{other:?}"),
}
assert!(inbox.update("stripe", "evt_1", |r| r.state = InboxState::Done { state: "PAID".into(), reason: String::new() }).unwrap());
assert!(inbox.get("stripe", "evt_1").unwrap().unwrap().is_done());
assert_eq!(inbox.rows().unwrap().len(), 1);
}
#[test]
fn a_kept_event_survives_a_restart() {
let dir = std::env::temp_dir().join(format!("monetize-inbox-{}", std::process::id()));
let _ = std::fs::create_dir_all(&dir);
let path = dir.join("callbacks.redb");
let _ = std::fs::remove_file(&path);
CallbackInbox::open(&path).unwrap().receive("stripe", "evt_9", "gunnar/nordisk/y", 5).unwrap();
let again = CallbackInbox::open(&path).unwrap();
assert_eq!(again.get("stripe", "evt_9").unwrap().unwrap().reference, "gunnar/nordisk/y");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn a_reference_is_flown_once_at_a_time() {
let f = Flights::default();
let one = f.fly("r").expect("free");
assert!(f.fly("r").is_none(), "a second resume of the same reference waits its turn");
assert!(f.fly("other").is_some());
drop(one);
assert!(f.fly("r").is_some());
}
}