crates/monetize-guardian/src/store.rs
//! **The guardian's store** (redb): the estates it knows and each box's fleet key, the
//! newest report per box, the order queue, the cases and the per-estate auto-fix opt-in.
//!
//! The queue is G1's `orders(instance, seq)`: an order stays until the box acks it in a
//! report, so a restart of monetize loses nothing — an unacked order is simply served again.
use std::path::Path;
use monetize_embed::fleet::{OrderAck, Report, SignedOrder};
use monetize_embed::signing::{base64_decode, base64_encode};
use monetize_product::guardian::AckView;
use redb::{Database, ReadableTable, TableDefinition};
use crate::case::Case;
const ESTATES: TableDefinition<&str, &str> = TableDefinition::new("estates");
const KEYS: TableDefinition<&str, &str> = TableDefinition::new("keys");
const BOXES: TableDefinition<&str, &str> = TableDefinition::new("boxes");
const ORDERS: TableDefinition<&str, &str> = TableDefinition::new("orders");
const COUNTERS: TableDefinition<&str, u64> = TableDefinition::new("counters");
const AUTO: TableDefinition<&str, u64> = TableDefinition::new("auto_fix");
const CASES: TableDefinition<&str, &str> = TableDefinition::new("cases");
/// ★ SEC-S1: the boxes an operator stopped ON PURPOSE (`<instance>/<role>` → [`Hold`] JSON).
const HOLDS: TableDefinition<&str, &str> = TableDefinition::new("holds");
/// **A box an operator stopped on purpose** (SEC-S1): who, why and when. The start pass never
/// starts it; only the same strong session's release does (then the start pass may).
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct Hold {
pub by: String,
pub why: String,
pub at_unix_ms: u64,
}
/// **A pin window with no end of its own** ([`Store::open_pin`]): open until the box's first
/// call pins its key, or until the box is forgotten ([`Store::forget_box`], an estate's
/// terminate). F6 (2026-10-20): a box whose install outlived the 10-minute window — a main
/// restarted half an hour after its estate was adopted — was refused for ever.
pub const UNTIL_PINNED: u64 = u64::MAX;
/// The acks kept per box, newest last.
pub const ACKS_KEPT: usize = 64;
/// ★ RA3-7: what a store error IS — its words, and whether it is a refusal of what was asked
/// (asking again changes nothing) or a failure of the store itself (the disk, redb: asking again
/// may pass).
#[derive(Debug, thiserror::Error)]
#[error("the guardian store: {0}")]
pub struct StoreError(pub String, pub StoreErrorKind);
/// [`StoreError`]'s kind.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StoreErrorKind {
/// What was asked is refused (a name that is no estate, a key that is no key).
Refused,
/// The store could not do it now (its disk, redb).
Failed,
}
impl StoreError {
/// A refusal of what was asked.
pub fn refused(why: String) -> StoreError {
StoreError(why, StoreErrorKind::Refused)
}
/// Whether it is a refusal (else a failure of the store, worth asking again).
pub fn is_refusal(&self) -> bool {
self.1 == StoreErrorKind::Refused
}
}
fn e<E: std::fmt::Display>(err: E) -> StoreError {
StoreError(err.to_string(), StoreErrorKind::Failed)
}
/// What monetize last heard from one box.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct BoxRecord {
/// When its newest report arrived (monetize's clock); 0: not heard since its key was
/// registered.
pub heard_unix_ms: u64,
pub report: Report,
pub acks: Vec<AckView>,
}
fn box_key(instance: &str, role: &str) -> String {
format!("{instance}/{role}")
}
fn order_key(instance: &str, seq: u64) -> String {
format!("{instance}/{seq:020}")
}
pub struct Store {
db: Database,
}
impl Store {
pub fn open(path: &Path) -> Result<Self, StoreError> {
Self::init(Database::create(path).map_err(e)?)
}
pub fn in_memory() -> Result<Self, StoreError> {
Self::with_backend(redb::backends::InMemoryBackend::new())
}
/// The store over any redb backend: a file, memory, or a test's disk that fails on demand.
pub fn with_backend(backend: impl redb::StorageBackend) -> Result<Self, StoreError> {
Self::init(Database::builder().create_with_backend(backend).map_err(e)?)
}
fn init(db: Database) -> Result<Self, StoreError> {
let w = db.begin_write().map_err(e)?;
{
w.open_table(ESTATES).map_err(e)?;
w.open_table(KEYS).map_err(e)?;
w.open_table(BOXES).map_err(e)?;
w.open_table(ORDERS).map_err(e)?;
w.open_table(COUNTERS).map_err(e)?;
w.open_table(AUTO).map_err(e)?;
w.open_table(CASES).map_err(e)?;
w.open_table(HOLDS).map_err(e)?;
}
w.commit().map_err(e)?;
Ok(Self { db })
}
/// **An estate the guardian watches**, of `product`.
pub fn declare(&self, instance: &str, product: &str) -> Result<(), StoreError> {
if instance.is_empty() || instance.contains('/') || product.is_empty() {
return Err(StoreError::refused(format!("{instance:?} of {product:?} is not an estate name")));
}
let w = self.db.begin_write().map_err(e)?;
w.open_table(ESTATES).map_err(e)?.insert(instance, product).map_err(e)?;
w.commit().map_err(e)
}
/// **A box's fleet key** (base64 of the 32-byte Ed25519 public key): what its calls are
/// verified under. A NEW key for a role is a new box: what was heard from the old one is
/// forgotten, so "heard" means heard from this box.
pub fn register_key(&self, instance: &str, role: &str, key_b64: &str) -> Result<(), StoreError> {
let raw = base64_decode(key_b64.trim()).filter(|k| k.len() == 32).ok_or_else(|| StoreError::refused(format!("the {role} key of {instance} is not a base64 Ed25519 public key")))?;
let key = base64_encode(&raw);
let w = self.db.begin_write().map_err(e)?;
{
let mut keys = w.open_table(KEYS).map_err(e)?;
let was = keys.get(box_key(instance, role).as_str()).map_err(e)?.map(|v| v.value().to_string());
if was.as_deref() != Some(key.as_str()) {
keys.insert(box_key(instance, role).as_str(), key.as_str()).map_err(e)?;
w.open_table(BOXES).map_err(e)?.remove(box_key(instance, role).as_str()).map_err(e)?;
}
}
w.commit().map_err(e)
}
/// **Open a pin window** for `(instance, role)` until `until_unix_ms`: the box's first call
/// may pin its key (TOFU). Closed by the pin.
pub fn open_pin(&self, instance: &str, role: &str, until_unix_ms: u64) -> Result<(), StoreError> {
let w = self.db.begin_write().map_err(e)?;
w.open_table(COUNTERS).map_err(e)?.insert(format!("pin/{instance}/{role}").as_str(), until_unix_ms).map_err(e)?;
w.commit().map_err(e)
}
/// Until when a pin window for `(instance, role)` is open; 0: none.
pub fn pin_window(&self, instance: &str, role: &str) -> u64 {
let Ok(r) = self.db.begin_read() else { return 0 };
let Ok(t) = r.open_table(COUNTERS) else { return 0 };
t.get(format!("pin/{instance}/{role}").as_str()).ok().flatten().map(|v| v.value()).unwrap_or(0)
}
/// **The newest owner's install notice taken for `(instance, role)`** (its
/// `installed_unix_ms`; 0: none) — so one notice opens one window, never one per tick.
pub fn notice_taken(&self, instance: &str, role: &str) -> u64 {
let Ok(r) = self.db.begin_read() else { return 0 };
let Ok(t) = r.open_table(COUNTERS) else { return 0 };
t.get(format!("notice/{instance}/{role}").as_str()).ok().flatten().map(|v| v.value()).unwrap_or(0)
}
/// Record that the owner's notice of `at_unix_ms` was taken for `(instance, role)`.
pub fn take_notice(&self, instance: &str, role: &str, at_unix_ms: u64) -> Result<(), StoreError> {
let w = self.db.begin_write().map_err(e)?;
w.open_table(COUNTERS).map_err(e)?.insert(format!("notice/{instance}/{role}").as_str(), at_unix_ms).map_err(e)?;
w.commit().map_err(e)
}
/// Forget the box in `role` (its key, what was heard and its pin window): a box that is
/// gone.
pub fn forget_box(&self, instance: &str, role: &str) -> Result<(), StoreError> {
let w = self.db.begin_write().map_err(e)?;
w.open_table(KEYS).map_err(e)?.remove(box_key(instance, role).as_str()).map_err(e)?;
w.open_table(BOXES).map_err(e)?.remove(box_key(instance, role).as_str()).map_err(e)?;
w.open_table(COUNTERS).map_err(e)?.remove(format!("pin/{instance}/{role}").as_str()).map_err(e)?;
w.commit().map_err(e)
}
/// **Forget an estate that is gone** (OVW-1, a finished terminate): its fleet row, every
/// box's key, record, pin window and hold, its unacked orders and its auto-fix opt-in. Its
/// counters (the order seqs, so a later estate of the same name never reuses a seq) and
/// its cases (what happened) stay.
pub fn forget_estate(&self, instance: &str) -> Result<(), StoreError> {
let (from, to) = (format!("{instance}/"), format!("{instance}0"));
let w = self.db.begin_write().map_err(e)?;
{
w.open_table(ESTATES).map_err(e)?.remove(instance).map_err(e)?;
w.open_table(AUTO).map_err(e)?.remove(instance).map_err(e)?;
for table in [KEYS, BOXES, ORDERS, HOLDS] {
let mut t = w.open_table(table).map_err(e)?;
let keys: Vec<String> = t.range(from.as_str()..to.as_str()).map_err(e)?.filter_map(|row| row.ok().map(|(k, _)| k.value().to_string())).collect();
for k in keys {
t.remove(k.as_str()).map_err(e)?;
}
}
let mut counters = w.open_table(COUNTERS).map_err(e)?;
let (pin_from, pin_to) = (format!("pin/{instance}/"), format!("pin/{instance}0"));
let pins: Vec<String> = counters.range(pin_from.as_str()..pin_to.as_str()).map_err(e)?.filter_map(|row| row.ok().map(|(k, _)| k.value().to_string())).collect();
for k in pins {
counters.remove(k.as_str()).map_err(e)?;
}
}
w.commit().map_err(e)
}
pub fn product_of(&self, instance: &str) -> Option<String> {
let r = self.db.begin_read().ok()?;
let t = r.open_table(ESTATES).ok()?;
t.get(instance).ok()?.map(|v| v.value().to_string())
}
pub fn key_of(&self, instance: &str, role: &str) -> Option<ed25519_dalek::VerifyingKey> {
let r = self.db.begin_read().ok()?;
let t = r.open_table(KEYS).ok()?;
let b64 = t.get(box_key(instance, role).as_str()).ok()??.value().to_string();
let raw: [u8; 32] = base64_decode(&b64)?.try_into().ok()?;
ed25519_dalek::VerifyingKey::from_bytes(&raw).ok()
}
/// Every estate: `(instance, product)`.
pub fn estates(&self) -> Vec<(String, String)> {
let Ok(r) = self.db.begin_read() else { return Vec::new() };
let Ok(t) = r.open_table(ESTATES) else { return Vec::new() };
t.iter().map(|it| it.filter_map(|row| row.ok().map(|(k, v)| (k.value().to_string(), v.value().to_string()))).collect()).unwrap_or_default()
}
/// The roles with a registered key.
pub fn roles_of(&self, instance: &str) -> Vec<String> {
let Ok(r) = self.db.begin_read() else { return Vec::new() };
let Ok(t) = r.open_table(KEYS) else { return Vec::new() };
let from = format!("{instance}/");
let to = format!("{instance}0");
t.range(from.as_str()..to.as_str())
.map(|it| it.filter_map(|row| row.ok().map(|(k, _)| k.value()[from.len()..].to_string())).collect())
.unwrap_or_default()
}
pub fn box_record(&self, instance: &str, role: &str) -> Option<BoxRecord> {
let r = self.db.begin_read().ok()?;
let t = r.open_table(BOXES).ok()?;
let v = t.get(box_key(instance, role).as_str()).ok()??;
serde_json::from_str(v.value()).ok()
}
/// **A box's report**, taken in at `now`: its acks remove their orders from the queue (in
/// the same transaction) and are kept with the order's kind. Answers the seqs acked.
pub fn record_report(&self, instance: &str, role: &str, now_unix_ms: u64, report: &Report) -> Result<Vec<u64>, StoreError> {
let w = self.db.begin_write().map_err(e)?;
let mut acked = Vec::new();
{
let mut orders = w.open_table(ORDERS).map_err(e)?;
let mut boxes = w.open_table(BOXES).map_err(e)?;
let mut rec: BoxRecord = boxes.get(box_key(instance, role).as_str()).map_err(e)?.and_then(|v| serde_json::from_str(v.value()).ok()).unwrap_or_default();
for OrderAck { seq, ok, detail, .. } in &report.acked {
let Some(o) = orders.get(order_key(instance, *seq).as_str()).map_err(e)?.and_then(|v| serde_json::from_str::<SignedOrder>(v.value()).ok()) else { continue };
if o.role != role {
continue;
}
orders.remove(order_key(instance, *seq).as_str()).map_err(e)?;
rec.acks.push(AckView { seq: *seq, kind: o.kind.clone(), ok: *ok, detail: detail.clone() });
acked.push(*seq);
}
let keep = rec.acks.len().saturating_sub(ACKS_KEPT);
rec.acks.drain(..keep);
rec.heard_unix_ms = now_unix_ms;
rec.report = report.clone();
boxes.insert(box_key(instance, role).as_str(), serde_json::to_string(&rec).map_err(e)?.as_str()).map_err(e)?;
}
w.commit().map_err(e)?;
Ok(acked)
}
/// **Queue an order**: its seq is the instance's next, taken in the same transaction it
/// is stored in; `sign` signs it with that seq.
pub fn queue(&self, mut order: SignedOrder, sign: impl FnOnce(SignedOrder) -> SignedOrder) -> Result<SignedOrder, StoreError> {
let w = self.db.begin_write().map_err(e)?;
let order = {
let mut counters = w.open_table(COUNTERS).map_err(e)?;
let k = format!("seq/{}", order.instance);
let seq = counters.get(k.as_str()).map_err(e)?.map(|v| v.value()).unwrap_or(0) + 1;
counters.insert(k.as_str(), seq).map_err(e)?;
order.seq = seq;
let order = sign(order);
w.open_table(ORDERS).map_err(e)?.insert(order_key(&order.instance, seq).as_str(), serde_json::to_string(&order).map_err(e)?.as_str()).map_err(e)?;
order
};
w.commit().map_err(e)?;
Ok(order)
}
/// **Queue an order that supersedes**: every unacked order of the same instance that
/// `stale` names is dropped in the same transaction the new one is queued in — newest
/// wins (an entitlement per tenant). A box that already pulled a dropped one may still
/// run it; the newer seq follows it.
pub fn queue_superseding(&self, order: SignedOrder, stale: impl Fn(&SignedOrder) -> bool, sign: impl FnOnce(SignedOrder) -> SignedOrder) -> Result<SignedOrder, StoreError> {
self.queue_with(order, stale, None, sign)
}
/// **Queue an order and remember it under `mark`** (`principal/<instance>/<signup>`), in
/// the same transaction: whoever asks again finds the seq ([`Store::mark`]) and reads its
/// ack instead of queueing a second one.
pub fn queue_marked(&self, order: SignedOrder, mark: &str, sign: impl FnOnce(SignedOrder) -> SignedOrder) -> Result<SignedOrder, StoreError> {
self.queue_with(order, |_| false, Some(mark), sign)
}
/// The seq last queued under `mark`.
pub fn mark(&self, mark: &str) -> Option<u64> {
let r = self.db.begin_read().ok()?;
let t = r.open_table(COUNTERS).ok()?;
let v = t.get(format!("mark/{mark}").as_str()).ok()??;
Some(v.value())
}
fn queue_with(&self, mut order: SignedOrder, stale: impl Fn(&SignedOrder) -> bool, mark: Option<&str>, sign: impl FnOnce(SignedOrder) -> SignedOrder) -> Result<SignedOrder, StoreError> {
let w = self.db.begin_write().map_err(e)?;
let order = {
let mut orders = w.open_table(ORDERS).map_err(e)?;
let from = format!("{}/", order.instance);
let to = format!("{}0", order.instance);
let drop: Vec<String> = orders
.range(from.as_str()..to.as_str())
.map_err(e)?
.filter_map(|row| row.ok())
.filter(|(_, v)| serde_json::from_str::<SignedOrder>(v.value()).is_ok_and(|o| stale(&o)))
.map(|(k, _)| k.value().to_string())
.collect();
for k in drop {
orders.remove(k.as_str()).map_err(e)?;
}
let mut counters = w.open_table(COUNTERS).map_err(e)?;
let k = format!("seq/{}", order.instance);
let seq = counters.get(k.as_str()).map_err(e)?.map(|v| v.value()).unwrap_or(0) + 1;
counters.insert(k.as_str(), seq).map_err(e)?;
if let Some(mark) = mark {
counters.insert(format!("mark/{mark}").as_str(), seq).map_err(e)?;
}
order.seq = seq;
let order = sign(order);
orders.insert(order_key(&order.instance, seq).as_str(), serde_json::to_string(&order).map_err(e)?.as_str()).map_err(e)?;
order
};
w.commit().map_err(e)?;
Ok(order)
}
/// The orders queued for one box after `after`, oldest first.
pub fn orders_for(&self, instance: &str, role: &str, after: u64) -> Vec<SignedOrder> {
self.queued(instance).into_iter().filter(|o| o.role == role && o.seq > after).collect()
}
/// Every unacked order of an estate, oldest first.
pub fn queued(&self, instance: &str) -> Vec<SignedOrder> {
let Ok(r) = self.db.begin_read() else { return Vec::new() };
let Ok(t) = r.open_table(ORDERS) else { return Vec::new() };
let from = format!("{instance}/");
let to = format!("{instance}0");
t.range(from.as_str()..to.as_str()).map(|it| it.filter_map(|row| row.ok().and_then(|(_, v)| serde_json::from_str(v.value()).ok())).collect()).unwrap_or_default()
}
/// The next value of a named counter (`epoch/<instance>`, `case/<instance>`).
pub fn next(&self, counter: &str) -> Result<u64, StoreError> {
let w = self.db.begin_write().map_err(e)?;
let n = {
let mut t = w.open_table(COUNTERS).map_err(e)?;
let n = t.get(counter).map_err(e)?.map(|v| v.value()).unwrap_or(0) + 1;
t.insert(counter, n).map_err(e)?;
n
};
w.commit().map_err(e)?;
Ok(n)
}
/// **The estate's auto-fix opt-in.** Off unless turned on.
pub fn auto_fix(&self, instance: &str) -> bool {
let Ok(r) = self.db.begin_read() else { return false };
let Ok(t) = r.open_table(AUTO) else { return false };
t.get(instance).ok().flatten().is_some_and(|v| v.value() == 1)
}
pub fn set_auto_fix(&self, instance: &str, on: bool) -> Result<(), StoreError> {
let w = self.db.begin_write().map_err(e)?;
w.open_table(AUTO).map_err(e)?.insert(instance, u64::from(on)).map_err(e)?;
w.commit().map_err(e)
}
/// ★ SEC-S1: **the operator's hold on the box in `role`**, if one is recorded.
pub fn hold_of(&self, instance: &str, role: &str) -> Option<Hold> {
let r = self.db.begin_read().ok()?;
let t = r.open_table(HOLDS).ok()?;
let v = t.get(box_key(instance, role).as_str()).ok()??;
serde_json::from_str(v.value()).ok()
}
/// Record (`Some`) or clear (`None`) the operator's hold on the box in `role`.
pub fn set_hold(&self, instance: &str, role: &str, hold: Option<&Hold>) -> Result<(), StoreError> {
let w = self.db.begin_write().map_err(e)?;
{
let mut t = w.open_table(HOLDS).map_err(e)?;
match hold {
Some(h) => {
t.insert(box_key(instance, role).as_str(), serde_json::to_string(h).map_err(e)?.as_str()).map_err(e)?;
}
None => {
t.remove(box_key(instance, role).as_str()).map_err(e)?;
}
}
}
w.commit().map_err(e)
}
pub fn put_case(&self, case: &Case) -> Result<(), StoreError> {
let w = self.db.begin_write().map_err(e)?;
w.open_table(CASES).map_err(e)?.insert(case.id.as_str(), serde_json::to_string(case).map_err(e)?.as_str()).map_err(e)?;
w.commit().map_err(e)
}
pub fn case(&self, id: &str) -> Option<Case> {
let r = self.db.begin_read().ok()?;
let t = r.open_table(CASES).ok()?;
let v = t.get(id).ok()??;
serde_json::from_str(v.value()).ok()
}
/// The estate's cases, newest first.
pub fn cases_of(&self, instance: &str) -> Vec<Case> {
let Ok(r) = self.db.begin_read() else { return Vec::new() };
let Ok(t) = r.open_table(CASES) else { return Vec::new() };
let mut out: Vec<Case> = t
.iter()
.map(|it| it.filter_map(|row| row.ok().and_then(|(_, v)| serde_json::from_str::<Case>(v.value()).ok())).filter(|c| c.instance == instance).collect())
.unwrap_or_default();
out.sort_by(|a, b| b.opened_unix_ms.cmp(&a.opened_unix_ms).then(b.id.cmp(&a.id)));
out
}
/// The estate's one open case, if any.
pub fn open_case(&self, instance: &str) -> Option<Case> {
self.cases_of(instance).into_iter().find(|c| c.state.open())
}
}