crates/monetize-estate-side/src/pump/duty.rs
//! **The pull channel as a jera job** (house rule: server background work is a jera
//! ChainQueue job, never a raw thread). One job = one round: report, long-poll the
//! orders (≤ 25 s), verify and run each, and report again at once when anything ran so
//! the acks reach monetize. A small ticker submits the round under ONE idempotency key,
//! so a round is never queued twice; the job's ledger row says what it did.
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;
use jera::jobs::{ChainQueue, ChainSpec, JobSink, Priority};
use monetize_embed::fleet::LONG_POLL_MAX_MS;
use super::puller::PumpBox;
use super::puller::{Link, Puller};
/// The ledger kind of one round.
pub const KIND: &str = "fleet.round";
/// The one key every round is submitted under.
pub const KEY: &str = "fleet-round";
/// **The role this box pulls as, asked at every round** (A7 row 1): nordisk-pairing
/// `elect::pull_role` over what the box knows NOW. `None`: not elected yet, the round waits
/// and reports nothing.
pub type RoleNow = Arc<dyn Fn() -> Option<String> + Send + Sync>;
/// Everything one round needs.
pub struct Duty {
pub puller: Mutex<Puller>,
pub link: Mutex<Box<dyn Link + Send>>,
pub boxed: Arc<dyn PumpBox + Send + Sync>,
/// The key orders verify under: replaced live when a later feed names one
/// ([`Duty::set_key`]).
monetize_pubkey: RwLock<Option<[u8; 32]>>,
/// The role at each round; `None` = the puller's own, fixed (tests, an enrolment).
role: Option<RoleNow>,
/// The pause before the next round, as monetize asked (ms).
pub pause_ms: AtomicU64,
/// When this channel started: an `unauthenticated` refusal inside
/// `nordisk_pairing::fleet::NOT_YET_KNOWN_GRACE` of it is "not known yet" ([`after_round`]).
started: std::time::Instant,
/// The `serving` and `data` words of the last report monetize TOOK (`None`: none taken
/// yet): the first one taken, and every change of either word, is said at [`SAID_AT`]
/// ([`accepted_line`]; lane F12: test3-holger-rs's main said nothing once its reports landed).
accepted: Mutex<Option<(String, String)>>,
}
impl Duty {
pub fn new(
puller: Puller,
link: Box<dyn Link + Send>,
boxed: Arc<dyn PumpBox + Send + Sync>,
monetize_pubkey: Option<[u8; 32]>,
) -> Arc<Duty> {
Duty::with_role(puller, link, boxed, monetize_pubkey, None)
}
/// [`Duty::new`] whose role is asked at every round (`role`).
pub fn with_role(
puller: Puller,
link: Box<dyn Link + Send>,
boxed: Arc<dyn PumpBox + Send + Sync>,
monetize_pubkey: Option<[u8; 32]>,
role: Option<RoleNow>,
) -> Arc<Duty> {
Arc::new(Duty {
puller: Mutex::new(puller),
link: Mutex::new(link),
boxed,
monetize_pubkey: RwLock::new(monetize_pubkey),
role,
pause_ms: AtomicU64::new(super::config::DEFAULT_EVERY.as_millis() as u64),
started: std::time::Instant::now(),
accepted: Mutex::new(None),
})
}
/// The key orders verify under now.
pub fn monetize_pubkey(&self) -> Option<[u8; 32]> {
*self
.monetize_pubkey
.read()
.unwrap_or_else(|e| e.into_inner())
}
/// **A key fed after the start**, applied from the next round on (no restart).
pub fn set_key(&self, key: Option<[u8; 32]>) {
*self
.monetize_pubkey
.write()
.unwrap_or_else(|e| e.into_inner()) = key;
}
/// **A door fed after the start** (another `fleet`, `fleet_at`, `fleet_ca` or estate):
/// the next round dials `link` as `instance`.
pub fn relink(&self, link: Box<dyn Link + Send>, instance: &str) {
*self.link.lock().unwrap_or_else(|e| e.into_inner()) = link;
self.puller
.lock()
.unwrap_or_else(|e| e.into_inner())
.me
.instance = instance.to_string();
}
/// **One round**, and what it did (the job row's detail).
pub fn round(&self, wait_ms: u64) -> anyhow::Result<serde_json::Value> {
let mut p = self.puller.lock().unwrap_or_else(|e| e.into_inner());
if let Some(role) = &self.role {
match role() {
None => {
return Ok(
serde_json::json!({ "reported": false, "waiting": "the installer's door has not elected this box's role yet: nothing is reported under a guess" }),
)
}
Some(r) if r != p.me.role => {
log::warn!(
"fleet: this box now reports as `{r}` (it reported as `{}`)",
p.me.role
);
p.me.role = r;
}
Some(_) => {}
}
}
let mut link = self.link.lock().unwrap_or_else(|e| e.into_inner());
let b = self.boxed.as_ref();
let now = crate::now_unix_ms;
let key = self.monetize_pubkey();
let health = b.health();
let words = (
health.get("serving").cloned().unwrap_or_default(),
health.get("data").cloned().unwrap_or_default(),
);
let reported = p.report(
link.as_mut(),
health,
b.product_report(),
b.own_review(),
now(),
);
if reported.is_ok() {
let mut last = self.accepted.lock().unwrap_or_else(|e| e.into_inner());
if let Some(line) = accepted_line(last.as_ref(), &words, &p.me.role) {
log::log!(SAID_AT, "{line}");
}
*last = Some(words);
}
if let Ok(next) = &reported {
if *next > 0 {
self.pause_ms.store(
(*next).clamp(1_000, super::config::DEFAULT_EVERY.as_millis() as u64),
Ordering::Relaxed,
);
}
}
let pulled = p.pull(
link.as_mut(),
b,
key.as_ref(),
wait_ms.min(LONG_POLL_MAX_MS),
now(),
);
let ran = pulled.as_ref().map(|x| !x.acks.is_empty()).unwrap_or(false);
// The acks ride a report at once: the guardian waits on them. One that does not land
// is said (the acks stay pending and ride the next round's report).
let acks_reported = ran.then(|| {
p.report(
link.as_mut(),
b.health(),
b.product_report(),
b.own_review(),
now(),
)
});
let acks_report_error = acks_reported
.as_ref()
.and_then(|r| r.as_ref().err())
.map(|f| f.to_string());
if let Some(e) = &acks_report_error {
log::log!(SAID_AT, "fleet: the report carrying this round's acks FAILED: {e} — they ride the next round's report");
}
if let Ok(x) = &pulled {
for a in &x.acks {
if a.ok {
log::info!("fleet: monetize's order {} ran: {}", a.seq, a.detail);
} else {
log::warn!(
"fleet: monetize's order {} was refused: {}",
a.seq,
a.detail
);
}
}
for f in &x.forged {
log::error!("fleet: an order that is NOT monetize's was offered and ignored: {f}");
}
}
if let Some(line) = report_round_line(&reported) {
// A7 row 31 (Gunnar 59fa3800): "not known yet" is said at INFO and asked again
// within 10 s; past the grace, or any other refusal, at WARN.
let pause = Duration::from_millis(self.pause_ms.load(Ordering::Relaxed));
let (level, pause) = after_round(&reported, self.started.elapsed(), pause);
self.pause_ms
.store(pause.as_millis() as u64, Ordering::Relaxed);
if level == log::Level::Info {
log::info!("{line} (monetize does not know this box yet: the estate's owner opens its window as the install feeds it; asked again in {} s)", pause.as_secs());
} else {
log::log!(level, "{line}");
}
}
let report_err = reported.as_ref().err().map(|f| f.to_string());
let detail = serde_json::json!({
"reported": reported.is_ok(),
"report_error": report_err,
"acks_report_error": acks_report_error,
"acks": pulled.as_ref().map(|x| x.acks.iter().map(|a| serde_json::json!({"seq": a.seq, "ok": a.ok, "detail": a.detail})).collect::<Vec<_>>()).unwrap_or_default(),
"forged": pulled.as_ref().map(|x| x.forged.len()).unwrap_or(0),
"pull_error": pulled.as_ref().err().map(|f| f.to_string()),
});
match (&reported, &pulled) {
(Err(r), Err(o)) if o.code != "no_key" => {
Err(anyhow::anyhow!("report: {r}; orders: {o}"))
}
_ => Ok(detail),
}
}
}
/// **The level the pull channel's own story is said at** (lane R10): an estate box logs at
/// `warn` by default (holger-server-lib `DEFAULT_LOG_FILTER`), so the channel's start, every
/// refused or failed report round and the boot log's `STAGE report` are said at WARN —
/// at `info` none of them was ever seen on a box (test3-holger-rs, 2026-10-20; Holger's lane R10).
pub const SAID_AT: log::Level = log::Level::Warn;
/// Pure: **the level a refused round is said at, and the pause before the next** (A7 row 31,
/// Gunnar 59fa3800): nordisk-pairing `fleet::after_refusal` over the refusal's code and the
/// time since the channel started. `unauthenticated` inside the grace (monetize does not know
/// this box YET: the owner's pin window opens as the install feeds it) is said at INFO and
/// asked again within 10 s; anything else at [`SAID_AT`], after `pause`.
pub fn after_round(
reported: &Result<u64, super::puller::Fault>,
since_start: Duration,
pause: Duration,
) -> (log::Level, Duration) {
match reported {
Err(f) => match nordisk_pairing::fleet::after_refusal(&f.code, since_start, pause) {
(p, true) => (log::Level::Info, p),
(p, false) => (SAID_AT, p),
},
Ok(_) => (SAID_AT, pause),
}
}
/// Pure: **what a report round says when it did not land** — monetize's refusal in its own
/// words (`refused`, `unauthenticated`, `malformed`), or why the call failed; `None` when
/// the report was taken.
pub fn report_round_line(reported: &Result<u64, super::puller::Fault>) -> Option<String> {
let f = reported.as_ref().err()?;
Some(match f.code.as_str() {
"refused" | "unauthenticated" | "malformed" => format!(
"fleet: monetize REFUSED this round's report ({}): {}",
f.code, f.message
),
_ => format!(
"fleet: this round's report FAILED ({}): {} — the next round tries again",
f.code, f.message
),
})
}
/// Pure: **what a report monetize TOOK says** (lane F12): the first one taken, and any
/// later one whose `serving` or `data` word differs from the last one taken; `None` for a
/// report that says what the last one said (a quiet channel is a working one). `words` is
/// (`serving`, `data`); an absent word is said as `nothing`.
pub fn accepted_line(
last: Option<&(String, String)>,
words: &(String, String),
role: &str,
) -> Option<String> {
let shown = |v: &str| {
if v.trim().is_empty() {
"nothing".to_string()
} else {
v.trim().to_string()
}
};
let (serving, data) = (shown(&words.0), shown(&words.1));
match last {
None => Some(format!("fleet: monetize TOOK this box's first report as `{role}`: serving = {serving}, data = {data}")),
Some(l) if l != words => Some(format!("fleet: monetize took this box's report as `{role}`: serving = {serving} (was {}), data = {data} (was {})", shown(&l.0), shown(&l.1))),
Some(_) => None,
}
}
/// The spec of one round, in `product`'s jera workspace.
pub fn spec(generation: i64, product: &str) -> ChainSpec {
ChainSpec::new(
KIND,
KEY,
"fleet",
generation,
product,
serde_json::json!({}),
)
.with_target("the ONE monetize's pull channel")
.with_priority(Priority::Normal)
}
/// **Start the pull channel** on `rt`: a ChainQueue of its own (a held long-poll never
/// waits behind other work, nor other work behind it) and the ticker that submits rounds.
pub fn start(rt: &tokio::runtime::Handle, sink: JobSink, duty: Arc<Duty>) -> Arc<ChainQueue> {
let q = ChainQueue::new(sink);
let d = Arc::clone(&duty);
// A32-14, stays a task: the pull channel's jera worker.
rt.spawn(Arc::clone(&q).run(move |_spec| d.round(LONG_POLL_MAX_MS)));
let tick = Arc::clone(&q);
let product = duty
.puller
.lock()
.unwrap_or_else(|e| e.into_inner())
.me
.product
.clone();
// A32-14, stays a task: the pull channel's ticker (it submits the jera rounds).
rt.spawn(async move {
let mut generation = 0i64;
loop {
generation += 1;
let _ = tick.submit(spec(generation, &product));
let pause = duty.pause_ms.load(Ordering::Relaxed).min(5_000);
tokio::time::sleep(Duration::from_millis(pause)).await;
}
});
q
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pump::key::BoxKey;
use crate::pump::puller::{Addressee, Asker, Ctx, PumpBox};
use crate::pump::puller::{Fault, Signed};
use monetize_embed::fleet::SignedOrder;
/// A link that records the role of every call and answers nothing.
struct Heard(Arc<Mutex<Vec<String>>>);
impl Link for Heard {
fn report(&mut self, s: Signed) -> Result<u64, crate::pump::puller::Fault> {
self.0
.lock()
.unwrap()
.push(format!("report {}/{}", s.call.instance, s.call.role));
Ok(1_000)
}
fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
self.0
.lock()
.unwrap()
.push(format!("orders {}/{}", s.call.instance, s.call.role));
Ok(Vec::new())
}
fn ask(&mut self, _: Signed) -> Result<crate::pump::puller::Asked, Fault> {
Err(Fault::new("none", "no asks here"))
}
}
struct Bare;
impl PumpBox for Bare {
fn run(
&self,
_: &Ctx,
o: &SignedOrder,
_: &mut dyn Asker,
) -> monetize_embed::fleet::OrderAck {
monetize_embed::fleet::OrderAck {
seq: o.seq,
ok: false,
detail: "a bare box runs nothing".into(),
receipt_json: String::new(),
}
}
}
fn duty(role: Option<RoleNow>, heard: &Arc<Mutex<Vec<String>>>) -> Arc<Duty> {
let me = Addressee {
product: "a-product".into(),
instance: "acme".into(),
role: "main".into(),
};
let p = Puller::new(me, BoxKey::from_seed([3; 32]), "test");
Duty::with_role(
p,
Box::new(Heard(Arc::clone(heard))),
Arc::new(Bare),
Some([9; 32]),
role,
)
}
/// ★ A7 row 1: the role is asked at every round. A box the door has not elected
/// reports NOTHING; elected twin after its start, it reports as `twin`, never as `main`.
#[test]
fn a_box_elected_twin_after_its_start_reports_as_the_twin() {
let heard = Arc::new(Mutex::new(Vec::new()));
let elected: Arc<Mutex<Option<Option<&'static str>>>> = Arc::new(Mutex::new(Some(None)));
let e = Arc::clone(&elected);
let role: RoleNow =
Arc::new(move || nordisk_pairing::elect::pull_role("", false, *e.lock().unwrap()));
let d = duty(Some(role), &heard);
d.round(0).unwrap();
assert!(
heard.lock().unwrap().is_empty(),
"not elected yet: nothing reported under a guess, {:?}",
heard.lock().unwrap()
);
*elected.lock().unwrap() = Some(Some("twin"));
d.round(0).unwrap();
assert_eq!(
*heard.lock().unwrap(),
vec!["report acme/twin", "orders acme/twin"]
);
}
/// ★ P4 (lane mon-shared) RED->GREEN: the report that carries a run order's ack back is
/// never dropped. Its failure is said, and the round's row names it; the ack stays
/// pending and rides the next round's report.
#[test]
fn a_failed_acks_report_is_said_and_the_ack_rides_the_next_round() {
struct AcksLost(u32);
impl Link for AcksLost {
fn report(&mut self, _: Signed) -> Result<u64, Fault> {
self.0 += 1;
if self.0 == 2 {
Err(Fault::new("unavailable", "the door went away"))
} else {
Ok(1_000)
}
}
fn orders(&mut self, _: Signed) -> Result<Vec<SignedOrder>, Fault> {
Ok(vec![crate::pump::puller::tests::signed(
4,
"main",
"note",
"{}",
crate::pump::puller::tests::MONETIZE_SEED,
)])
}
fn ask(&mut self, _: Signed) -> Result<crate::pump::puller::Asked, Fault> {
Err(Fault::new("none", "no asks here"))
}
}
let me = Addressee {
product: "a-product".into(),
instance: "acme".into(),
role: "main".into(),
};
let p = Puller::new(me, BoxKey::from_seed([3; 32]), "test");
let key = crate::pump::puller::tests::monetize_key().to_bytes();
let d = Duty::with_role(p, Box::new(AcksLost(0)), Arc::new(Bare), Some(key), None);
let row = d.round(0).unwrap();
assert_eq!(row["acks"].as_array().map(|a| a.len()), Some(1), "{row}");
let said = row["acks_report_error"].as_str().unwrap_or_default();
assert!(
said.contains("the door went away"),
"the failed acks report is in the row: {row}"
);
assert_eq!(
d.puller.lock().unwrap().pulled.pending.len(),
1,
"the ack stays pending for the next report"
);
}
/// A key and a door fed after the start reach the next round (no restart).
#[test]
fn a_late_key_and_a_new_door_reach_the_next_round() {
let heard = Arc::new(Mutex::new(Vec::new()));
let d = duty(None, &heard);
d.set_key(Some([1; 32]));
assert_eq!(d.monetize_pubkey(), Some([1; 32]));
let moved = Arc::new(Mutex::new(Vec::new()));
d.relink(Box::new(Heard(Arc::clone(&moved))), "acme2");
d.round(0).unwrap();
assert!(heard.lock().unwrap().is_empty());
assert_eq!(
*moved.lock().unwrap(),
vec!["report acme2/main", "orders acme2/main"]
);
}
/// ★ R10 RED→GREEN: a box logs at `warn` by default, so the pull channel's story is said
/// at WARN or it is never seen.
#[test]
fn the_pull_channels_story_is_said_at_warn() {
assert!(
SAID_AT <= log::Level::Warn,
"{SAID_AT:?} is dropped by a box's default `warn` filter"
);
}
/// ★ F12 (test3-holger-rs 2026-10-21: after its pin its main said nothing more): the first
/// report monetize TOOK is said, and so is every change of its `serving` or `data` word; a
/// report that says what the last one said is quiet. The duty keeps the words it last had
/// taken, and a refused round keeps none.
#[test]
fn the_first_report_taken_and_every_change_of_its_words_is_said() {
let w = |s: &str, d: &str| (s.to_string(), d.to_string());
let first =
accepted_line(None, &w("yes", "ok"), "main").expect("the first report taken is said");
assert!(
first.contains("TOOK")
&& first.contains("`main`")
&& first.contains("serving = yes")
&& first.contains("data = ok"),
"{first}"
);
assert_eq!(
accepted_line(Some(&w("yes", "ok")), &w("yes", "ok"), "main"),
None,
"the same words again are quiet"
);
let changed = accepted_line(Some(&w("yes", "ok")), &w("no", "fresh"), "main")
.expect("a change is said");
assert!(
changed.contains("serving = no (was yes)") && changed.contains("data = fresh (was ok)"),
"{changed}"
);
assert!(accepted_line(None, &w("", ""), "twin")
.unwrap()
.contains("serving = nothing, data = nothing"));
let heard = Arc::new(Mutex::new(Vec::new()));
let d = duty(None, &heard);
assert!(d.accepted.lock().unwrap().is_none());
d.round(0).unwrap();
assert!(
d.accepted.lock().unwrap().is_some(),
"an accepted round keeps the words it said"
);
}
/// ★ A7 row 31 (Gunnar 59fa3800): `unauthenticated` inside the grace is "not known yet":
/// said at INFO, asked again within 10 s. Past the grace, or any other refusal: WARN and
/// the pause as before. The duty's own round keeps the pause it chose.
#[test]
fn a_box_not_known_yet_is_said_quietly_and_asked_again_soon_inside_the_grace() {
let every = Duration::from_secs(30);
let unauth: Result<u64, Fault> = Err(Fault::new(
"unauthenticated",
"monetize does not know this box key",
));
assert_eq!(
after_round(&unauth, Duration::from_secs(60), every),
(log::Level::Info, Duration::from_secs(10))
);
assert_eq!(
after_round(
&unauth,
nordisk_pairing::fleet::NOT_YET_KNOWN_GRACE + Duration::from_secs(1),
every
),
(SAID_AT, every),
"past the grace: WARN"
);
assert_eq!(
after_round(
&Err(Fault::new("unavailable", "down")),
Duration::from_secs(60),
every
),
(SAID_AT, every)
);
assert_eq!(
after_round(&Ok(30_000), Duration::from_secs(60), every).1,
every,
"a taken report keeps the pause"
);
// A duty refused `unauthenticated` at its first round pauses at most 10 s.
struct Refused;
impl Link for Refused {
fn report(&mut self, _: Signed) -> Result<u64, Fault> {
Err(Fault::new("unauthenticated", "not known"))
}
fn orders(&mut self, _: Signed) -> Result<Vec<SignedOrder>, Fault> {
Err(Fault::new("unauthenticated", "not known"))
}
fn ask(&mut self, _: Signed) -> Result<crate::pump::puller::Asked, Fault> {
Err(Fault::new("none", "no asks here"))
}
}
let me = Addressee {
product: "a-product".into(),
instance: "acme".into(),
role: "main".into(),
};
let d = Duty::new(
Puller::new(me, BoxKey::from_seed([3; 32]), "test"),
Box::new(Refused),
Arc::new(Bare),
Some([9; 32]),
);
let _ = d.round(0);
assert!(
d.pause_ms.load(Ordering::Relaxed) <= 10_000,
"{}",
d.pause_ms.load(Ordering::Relaxed)
);
}
/// ★ R10 RED→GREEN: every refused or failed report round is said, with monetize's words.
#[test]
fn a_refused_or_failed_report_round_is_said_with_monetizes_words() {
let refused = report_round_line(&Err(Fault::new(
"refused",
"no estate acme-7 is sold to this box key",
)))
.unwrap();
assert!(
refused.starts_with("fleet: ")
&& refused.contains("REFUSED")
&& refused.contains("no estate acme-7 is sold to this box key"),
"{refused}"
);
let failed = report_round_line(&Err(Fault::new(
"unavailable",
"tcp connect error: Connection refused",
)))
.unwrap();
assert!(
failed.contains("FAILED")
&& failed.contains("unavailable")
&& failed.contains("tcp connect error"),
"{failed}"
);
assert_eq!(
report_round_line(&Ok(30_000)),
None,
"a taken report says nothing"
);
}
/// ★ P4a: a round is a jera job in the PRODUCT's workspace (it was always "holger"), under
/// the one idempotency key.
#[test]
fn a_round_is_one_job_in_the_products_workspace() {
let s = spec(7, "a-product");
assert_eq!(
(
s.kind.as_str(),
s.key.as_str(),
s.group.as_str(),
s.generation,
s.workspace.as_str()
),
(KIND, KEY, "fleet", 7, "a-product")
);
assert_eq!(s.target, "the ONE monetize's pull channel");
}
}