crates/monetize-ui-server/src/bounded.rs
//! **Every console page answers within [`PAGE_WITHIN`]** (PROD-4, owner 2026-10-26: the
//! Overview and the Guardian page loaded for ever, an estate's map took 20 s). A read a page
//! waits on — the fleet, an estate's map, a product's machines at the cloud — is asked ONCE
//! per key at a time (RA3-1: a single flight; a second page that asks while it is out joins
//! it and starts nothing). It answers the page if it comes within the bound; past it the page
//! is answered from the last answer of the same read, with its age, while the one flight goes
//! on and keeps what it gets: the live data loads in on the page's next ask. With nothing kept
//! yet the page is told so, by the bound — never an endless spinner. The fleet itself is not
//! asked by a page at all: a jera job keeps it ([`crate::estates::FleetKeeper`]) and the page
//! reads what it kept ([`within_kept`]).
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use crate::grpc::Fail;
/// How long a page waits for any one read.
pub const PAGE_WITHIN: Duration = Duration::from_secs(2);
/// One key's last outcome: the last answer and when it came, the last refusal, and the
/// number of the newest outcome of either kind ([`Seen::mark`]).
struct Kept<T> {
answer: Option<(Instant, T)>,
refusal: Option<tonic::Status>,
outcome: u64,
}
impl<T> Default for Kept<T> {
fn default() -> Self {
Kept { answer: None, refusal: None, outcome: 0 }
}
}
struct Inner<T> {
kept: HashMap<String, Kept<T>>,
/// The keys whose one read is out now (the single flight).
flying: HashSet<String>,
/// Counts every outcome, of any key.
n: u64,
}
/// The last answer of each read, by key, and the reads out now.
pub struct Seen<T> {
inner: Arc<Mutex<Inner<T>>>,
/// Wakes the pages that wait: it carries [`Inner::n`].
tick: Arc<tokio::sync::watch::Sender<u64>>,
}
impl<T> Clone for Seen<T> {
fn clone(&self) -> Self {
Seen { inner: self.inner.clone(), tick: self.tick.clone() }
}
}
impl<T> Default for Seen<T> {
fn default() -> Self {
Seen { inner: Arc::new(Mutex::new(Inner { kept: HashMap::new(), flying: HashSet::new(), n: 0 })), tick: Arc::new(tokio::sync::watch::channel(0).0) }
}
}
/// What came for a key after a [`Seen::mark`].
enum Came<T> {
Answer(T),
Refused(tonic::Status),
}
impl<T: Clone> Seen<T> {
fn lock(&self) -> std::sync::MutexGuard<'_, Inner<T>> {
self.inner.lock().unwrap_or_else(|e| e.into_inner())
}
fn outcome(&self, key: &str, f: impl FnOnce(&mut Kept<T>)) {
self.outcome_landing(key, false, f)
}
/// Keep an outcome of `key`; `landed`: its flight ends in the same breath, so an ask that
/// comes after it sees the outcome AND no flight (it starts its own), never one without
/// the other.
fn outcome_landing(&self, key: &str, landed: bool, f: impl FnOnce(&mut Kept<T>)) {
let n = {
let mut g = self.lock();
if landed {
g.flying.remove(key);
}
g.n += 1;
let n = g.n;
let k = g.kept.entry(key.to_string()).or_default();
f(k);
k.outcome = n;
n
};
self.tick.send_replace(n);
}
/// The read under `key` answered `v`.
pub fn put(&self, key: &str, v: T) {
self.outcome(key, |k| {
k.answer = Some((Instant::now(), v));
k.refusal = None;
});
}
/// The read under `key` was refused (`s`); the last answer is kept.
pub fn refuse(&self, key: &str, s: tonic::Status) {
self.outcome(key, |k| k.refusal = Some(s));
}
/// The last answer under `key` and how many seconds old it is.
pub fn get(&self, key: &str) -> Option<(T, u64)> {
self.lock().kept.get(key).and_then(|k| k.answer.as_ref()).map(|(at, v)| (v.clone(), at.elapsed().as_secs()))
}
/// The number of the newest outcome so far: what a waiter waits past.
pub fn mark(&self) -> u64 {
self.lock().n
}
/// Is a read under `key` out now?
pub fn flying(&self, key: &str) -> bool {
self.lock().flying.contains(key)
}
/// What came for `key` after `mark`, if anything has.
fn came_after(&self, key: &str, mark: u64) -> Option<Came<T>> {
let g = self.lock();
let k = g.kept.get(key).filter(|k| k.outcome > mark)?;
match (&k.refusal, &k.answer) {
(Some(s), _) => Some(Came::Refused(s.clone())),
(None, Some((_, v))) => Some(Came::Answer(v.clone())),
(None, None) => None,
}
}
/// **Wait up to `bound` for an outcome of `key` newer than `mark`.**
async fn wait_after(&self, key: &str, mark: u64, bound: Duration) -> Option<Came<T>> {
let mut rx = self.tick.subscribe();
let until = tokio::time::Instant::now() + bound;
loop {
if let Some(c) = self.came_after(key, mark) {
return Some(c);
}
match tokio::time::timeout_at(until, rx.changed()).await {
Ok(Ok(())) => continue,
_ => return self.came_after(key, mark),
}
}
}
}
impl<T: Clone + Send + 'static> Seen<T> {
/// **Start the one read under `key`, unless it is out already** (RA3-1). `true`: this
/// call started it; `false`: it joined the one in flight and `call` is dropped unasked.
/// The read keeps its answer, or its refusal, when it ends.
pub fn fly<F>(&self, key: &str, call: F) -> bool
where
F: Future<Output = Result<T, tonic::Status>> + Send + 'static,
{
if !self.lock().flying.insert(key.to_string()) {
return false;
}
let (keep, k) = (self.clone(), key.to_string());
// A32-14, stays a task: the ONE flight of a page read per key (RA3-1: a second ask joins
// it), ended by the channel's own CALL_WITHIN bound; the fleet's is asked by its jera job.
tokio::spawn(async move {
match call.await {
Ok(v) => keep.outcome_landing(&k, true, |kept| {
kept.answer = Some((Instant::now(), v));
kept.refusal = None;
}),
Err(s) => keep.outcome_landing(&k, true, |kept| kept.refusal = Some(s)),
}
});
true
}
}
/// A refusal a kept answer may stand in for: the read did not come, not "no".
fn transient(s: &tonic::Status) -> bool {
matches!(s.code(), tonic::Code::Unavailable | tonic::Code::DeadlineExceeded | tonic::Code::Cancelled)
}
/// What a bounded read answered: the value and, when it is a kept one, its age in seconds.
pub type Aged<T> = (T, Option<u64>);
/// **`call` within [`PAGE_WITHIN`], else the last answer under `key`** (see the module).
/// `what` names the read in the words a page shows.
pub async fn within<T, F>(seen: &Seen<T>, key: &str, what: &str, call: F) -> Result<Aged<T>, Fail>
where
T: Clone + Send + 'static,
F: Future<Output = Result<T, tonic::Status>> + Send + 'static,
{
within_for(seen, key, what, PAGE_WITHIN, call).await
}
/// [`within`] with its own bound: the one flight of `key` ([`Seen::fly`]), started or joined.
pub async fn within_for<T, F>(seen: &Seen<T>, key: &str, what: &str, bound: Duration, call: F) -> Result<Aged<T>, Fail>
where
T: Clone + Send + 'static,
F: Future<Output = Result<T, tonic::Status>> + Send + 'static,
{
let mark = seen.mark();
seen.fly(key, call);
answer_after(seen, key, what, bound, mark).await
}
/// **What `key` holds within `bound`, asked by someone else** (RA3-1: the fleet, which a jera
/// job keeps). `ask` asks the keeper for a fresh read (a keyed job: a second ask while one is
/// queued or out is the same job); the page waits for its outcome up to `bound` and else
/// answers what was kept, with its age. The page itself starts no read.
pub async fn within_kept<T: Clone>(seen: &Seen<T>, key: &str, what: &str, bound: Duration, ask: impl FnOnce()) -> Result<Aged<T>, Fail> {
let mark = seen.mark();
ask();
answer_after(seen, key, what, bound, mark).await
}
async fn answer_after<T: Clone>(seen: &Seen<T>, key: &str, what: &str, bound: Duration, mark: u64) -> Result<Aged<T>, Fail> {
let why = match seen.wait_after(key, mark, bound).await {
Some(Came::Answer(v)) => return Ok((v, None)),
Some(Came::Refused(s)) if !transient(&s) => return Err(Fail::from(s)),
Some(Came::Refused(s)) => format!("did not answer within {} s ({})", bound.as_secs(), s.message()),
None => format!("did not answer within {} s (still asked: its answer shows on the next look)", bound.as_secs()),
};
match seen.get(key) {
Some((v, age)) => Ok((v, Some(age))),
None => Err(Fail { code: "deadlineexceeded".into(), message: format!("{what} {why}; there is no earlier answer to show yet — look again in a moment") }),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn a_read_within_the_bound_is_fresh_one_past_it_is_the_kept_one_with_its_age() {
let seen: Seen<u32> = Seen::default();
assert_eq!(within_for(&seen, "k", "the fleet", Duration::from_millis(200), async { Ok(7) }).await.unwrap(), (7, None));
let t0 = Instant::now();
let slow = within_for(&seen, "k", "the fleet", Duration::from_millis(200), async {
tokio::time::sleep(Duration::from_millis(400)).await;
Ok(8)
})
.await
.unwrap();
assert!(t0.elapsed() < Duration::from_millis(350));
assert_eq!(slow, (7, Some(0)), "the kept answer, aged");
tokio::time::sleep(Duration::from_millis(400)).await;
assert_eq!(seen.get("k").map(|(v, _)| v), Some(8), "the live answer loaded in after");
let none: Seen<u32> = Seen::default();
let e = within_for(&none, "k", "the map", Duration::from_millis(100), std::future::pending()).await.unwrap_err();
assert!(e.message.contains("the map did not answer within"), "{}", e.message);
assert_eq!(e.code, "deadlineexceeded");
let refused = within_for(&seen, "k", "x", Duration::from_millis(100), async { Err::<u32, _>(tonic::Status::not_found("no such estate")) }).await.unwrap_err();
assert_eq!(refused.code, "notfound", "a refusal is the refusal, never a kept answer");
}
/// ★ RA3-1 RED: **ten pages that look while the read hangs ask it ONCE** — each joins the
/// one flight of its key; another key is its own flight.
#[tokio::test]
async fn ten_looks_while_the_read_hangs_ask_it_once() {
use std::sync::atomic::{AtomicUsize, Ordering};
let seen: Seen<u32> = Seen::default();
let asked = Arc::new(AtomicUsize::new(0));
let looks = (0..10).map(|_| {
let (seen, asked) = (seen.clone(), asked.clone());
async move {
within_for(&seen, "", "the fleet", Duration::from_millis(100), async move {
asked.fetch_add(1, Ordering::SeqCst);
std::future::pending::<Result<u32, tonic::Status>>().await
})
.await
}
});
let all = futures_util::future::join_all(looks).await;
assert!(all.iter().all(|r| r.as_ref().is_err_and(|e| e.code == "deadlineexceeded")), "each look is told by the bound");
assert_eq!(asked.load(Ordering::SeqCst), 1, "one read for ten looks");
assert!(seen.flying(""));
assert!(within_for(&seen, "acme2", "the map", Duration::from_millis(100), async { Ok(3) }).await.is_ok(), "another key flies on its own");
}
/// ★ RA3-1: **a kept read** ([`within_kept`]) starts nothing itself: the keeper's outcome
/// within the bound is fresh, past it the kept answer is aged, a refusal is the refusal.
#[tokio::test]
async fn a_kept_read_answers_what_the_keeper_brings_within_the_bound() {
let seen: Seen<u32> = Seen::default();
let e = within_kept(&seen, "", "the fleet", Duration::from_millis(100), || {}).await.unwrap_err();
assert!(e.message.contains("no earlier answer"), "{}", e.message);
let keeper = seen.clone();
let got = within_kept(&seen, "", "the fleet", Duration::from_millis(500), move || {
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(50)).await;
keeper.put("", 5);
});
})
.await
.unwrap();
assert_eq!(got, (5, None), "the keeper's answer, fresh");
assert_eq!(within_kept(&seen, "", "the fleet", Duration::from_millis(100), || {}).await.unwrap(), (5, Some(0)), "nothing new: the kept one, aged");
let refuser = seen.clone();
let e = within_kept(&seen, "", "the fleet", Duration::from_millis(500), move || refuser.refuse("", tonic::Status::permission_denied("no"))).await.unwrap_err();
assert_eq!(e.code, "permissiondenied");
let shaky = seen.clone();
let got = within_kept(&seen, "", "the fleet", Duration::from_millis(500), move || shaky.refuse("", tonic::Status::unavailable("down"))).await.unwrap();
assert_eq!(got, (5, Some(0)), "a transient refusal: the kept answer stands in");
}
}