crates/monetize-site-buy/src/ladders.rs
//! **The plans pages' cache** (LADDER-1/2, owner 2026-10-27): the account's plans
//! (`Signup.Plans`, per market) and the new private instance's ladder (`EstateOrder.Ladder`,
//! no company) are kept per site and refreshed by a jera job, so a page view never waits on the
//! ONE box. The ONE box prices every step of a ladder at the cloud, behind the same
//! one-at-a-time lane as its whole-account listings: on gunnar.rs a page view waited about 5 s,
//! and past the relay's 10 s the "New instance" tab said "The shop did not answer".
//!
//! * A page answers from what is kept, at once, with its age (`age_s`).
//! * Nothing kept yet: the page asks for a refresh and waits at most [`FIRST_WAIT`]; then it
//! says the shop's own refusal, or [`NOT_READ`], never a spinner.
//! * The refresh is a jera ChainQueue job (`site_ladders`): every [`EVERY`], and at once on
//! [`Ladders::refresh_now`] (a site calls it when its feed changes). A refresh that fails
//! keeps what is kept and is said.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use axum::http::StatusCode;
use jera::jobs::chain::{ChainQueue, ChainSpec, Priority};
use jera::jobs::JobSink;
use serde_json::json;
use crate::link::ShopLink;
use crate::relay::Answer;
/// What a page says while nothing is kept and the ONE box does not answer.
pub const NOT_READ: &str = "The plans are not read from the shop yet; they are being asked for now. Look again in a moment.";
/// How often the kept ladders are read again.
pub const EVERY: Duration = Duration::from_secs(300);
/// How long a page with nothing kept waits for the first read.
pub const FIRST_WAIT: Duration = Duration::from_secs(1);
/// The jera job kind of one refresh.
pub const JOB: &str = "site_ladders";
/// The key of the new private instance's ladder.
pub const PRIVATE: &str = "private";
/// The key of the account plans in `market`.
pub fn signup_key(market: i32) -> String {
format!("signup:{market}")
}
/// **One site's kept ladders** (cheap to clone: one shared cache).
#[derive(Clone, Default)]
pub struct Ladders(Arc<Inner>);
#[derive(Default)]
struct Inner {
kept: Mutex<BTreeMap<String, (Answer, Instant)>>,
/// The last refusal a refresh met, said while nothing better is kept.
failed: Mutex<BTreeMap<String, Answer>>,
/// The keys a page has asked for: what a refresh reads.
wanted: Mutex<BTreeSet<String>>,
lane: OnceLock<Arc<ChainQueue>>,
filled: tokio::sync::Notify,
}
impl Ladders {
/// The kept answer for `key`, with its age, if one is kept.
pub fn kept(&self, key: &str) -> Option<Answer> {
let (mut a, at) = self
.0
.kept
.lock()
.unwrap_or_else(|p| p.into_inner())
.get(key)
.cloned()?;
if let Some(o) = a.body.as_object_mut() {
o.insert("age_s".into(), json!(at.elapsed().as_secs()));
}
Some(a)
}
fn put(&self, key: &str, a: Answer) {
self.0
.kept
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(key.to_string(), (a, Instant::now()));
self.0
.failed
.lock()
.unwrap_or_else(|p| p.into_inner())
.remove(key);
self.0.filled.notify_waiters();
}
fn failed(&self, key: &str, a: Answer) {
self.0
.failed
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(key.to_string(), a);
self.0.filled.notify_waiters();
}
/// **Read every wanted ladder again, now** (a site's feed changed): a jera job on this
/// site's lane; a second ask while one is queued is that one.
pub fn refresh_now(&self, shop: &ShopLink) {
if let Some(lane) = self.lane(shop) {
let _ = lane.submit(
ChainSpec::new(JOB, "site-ladders", JOB, 0, "monetize-site-buy", json!({}))
.with_target("read the plans pages' ladders")
.with_priority(Priority::Normal),
);
}
}
/// The site's lane, started on the first ask on the caller's runtime, with its ticker.
fn lane(&self, shop: &ShopLink) -> Option<Arc<ChainQueue>> {
if let Some(q) = self.0.lane.get() {
return Some(Arc::clone(q));
}
let rt = tokio::runtime::Handle::try_current().ok()?;
let queue = ChainQueue::new(JobSink::noop());
let q = Arc::clone(self.0.lane.get_or_init(|| Arc::clone(&queue)));
if Arc::ptr_eq(&q, &queue) {
let (me, link, h) = (self.clone(), shop.clone(), rt.clone());
// A32-14, stays a task: the ladders lane's jera worker.
rt.spawn(
Arc::clone(&queue).run(move |_spec| match h.block_on(me.refresh(&link)) {
0 => Ok(json!({ "read": true })),
n => Err(anyhow::anyhow!(
"{n} ladder(s) not read; what is kept is kept"
)),
}),
);
let (me, link) = (self.clone(), shop.clone());
// A32-14, stays a task: the ladders lane's ticker (it submits the jera refreshes).
rt.spawn(async move {
loop {
tokio::time::sleep(EVERY).await;
me.refresh_now(&link);
}
});
}
Some(q)
}
/// **Warm `keys` now** (a site's doors are built): they are read at once, and kept fresh
/// from then on. Nothing happens off a runtime (the first page view reads them then).
pub fn warm(&self, shop: &ShopLink, keys: &[String]) {
self.0
.wanted
.lock()
.unwrap_or_else(|p| p.into_inner())
.extend(keys.iter().cloned());
self.refresh_now(shop);
}
/// **The page's answer for `key`**: what is kept, at once; else a refresh and at most
/// [`FIRST_WAIT`]; else the shop's own refusal, or [`NOT_READ`].
pub async fn answer(&self, shop: &ShopLink, key: &str) -> Answer {
self.0
.wanted
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(key.to_string());
if let Some(a) = self.kept(key) {
return a;
}
self.refresh_now(shop);
let until = Instant::now() + FIRST_WAIT;
loop {
if let Some(a) = self.kept(key) {
return a;
}
if let Some(a) = self
.0
.failed
.lock()
.unwrap_or_else(|p| p.into_inner())
.get(key)
.cloned()
{
return a;
}
let left = until.saturating_duration_since(Instant::now());
if left.is_zero() {
return Answer {
status: StatusCode::SERVICE_UNAVAILABLE,
body: json!({ "error": NOT_READ, "not_read": true }),
see_other: None,
};
}
let _ = tokio::time::timeout(
left.min(Duration::from_millis(50)),
self.0.filled.notified(),
)
.await;
}
}
/// One refresh of every wanted key. Answers how many could not be read.
async fn refresh(&self, shop: &ShopLink) -> usize {
let wanted: Vec<String> = self
.0
.wanted
.lock()
.unwrap_or_else(|p| p.into_inner())
.iter()
.cloned()
.collect();
let mut failed = 0;
for key in wanted {
let read = if key == PRIVATE {
read_private(shop).await
} else if let Some(m) = key
.strip_prefix("signup:")
.and_then(|m| m.parse::<i32>().ok())
{
read_signup(shop, m).await
} else {
continue;
};
match read {
Read::Kept(a) => {
// ★ A new estate's ladder the shop ANSWERED is its word that it sells private
// instances here: kept with it, so the private doors do not ask again (and
// never read a later refusal of the buyer's ticket as "not sold").
if key == PRIVATE && a.status == StatusCode::OK {
shop.note_sells_estates(true);
}
self.put(&key, a)
}
Read::Failed(a) => {
tracing::warn!(target: crate::relay::AUDIT_TARGET, event = "ladders.unread", key = %key, body = %a.body, "the plans page's ladder was not read; what is kept is kept");
self.failed(&key, a);
failed += 1;
}
}
}
failed
}
}
/// What one read gave.
enum Read {
/// An answer to keep (the ladder, or the shop's final word, such as "not sold").
Kept(Answer),
/// A transient refusal: said while nothing is kept, never kept over a good answer.
Failed(Answer),
}
async fn read_signup(shop: &ShopLink, market: i32) -> Read {
let mut client = shop.signup();
let asked = client.plans(monetize_proto::SignupPlansRequest {
product: shop.product().to_owned(),
market,
});
match tokio::time::timeout(shop.deadlines().call, asked).await {
Ok(Ok(a)) => {
let body = crate::relay::plans_body(a.into_inner());
shop.caches()
.last_offer
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(market, (body.clone(), crate::relay::now_ms()));
Read::Kept(Answer::ok(body))
}
Ok(Err(s)) if s.code() == tonic::Code::Unimplemented => {
tracing::warn!(target: crate::relay::AUDIT_TARGET, event = "signup.plans_failed", code = ?s.code(), message = %s.message(), "monetize answered no account plans");
Read::Kept(Answer::refuse(
StatusCode::NOT_IMPLEMENTED,
crate::relay::ACCOUNT_NOT_SOLD_HERE,
))
}
other => {
if let Ok(Err(s)) = &other {
tracing::warn!(target: crate::relay::AUDIT_TARGET, event = "signup.plans_failed", code = ?s.code(), message = %s.message(), "monetize answered no account plans");
}
let held = shop
.caches()
.last_offer
.lock()
.unwrap_or_else(|p| p.into_inner())
.get(&market)
.cloned();
let mut body = json!({ "error": crate::relay::SHOP_UNAVAILABLE, "unavailable": true });
if let Some((offer, at)) = held {
body["last_known"] = offer;
body["as_of_unix_ms"] = json!(at);
}
Read::Failed(Answer {
status: StatusCode::SERVICE_UNAVAILABLE,
body,
see_other: None,
})
}
}
}
async fn read_private(shop: &ShopLink) -> Read {
let mut client = shop.estate_order();
let asked = client.ladder(monetize_proto::EstateOrderLadderRequest {
company: String::new(),
actor_ticket: Vec::new(),
});
match tokio::time::timeout(shop.deadlines().call, asked).await {
Ok(Ok(l)) => Read::Kept(Answer::ok(crate::relay::ladder_body(l.into_inner()))),
Ok(Err(s)) => {
let a = crate::relay::from_status(s);
if a.status == StatusCode::BAD_GATEWAY
|| a.status == StatusCode::SERVICE_UNAVAILABLE
|| a.status == StatusCode::GATEWAY_TIMEOUT
{
Read::Failed(a)
} else {
Read::Kept(a)
}
}
Err(_) => Read::Failed(Answer {
status: StatusCode::SERVICE_UNAVAILABLE,
body: json!({ "error": crate::relay::SHOP_UNAVAILABLE, "unavailable": true }),
see_other: None,
}),
}
}