server/lib/src/meter_services.rs
//! **The three host services a meter guest is lent: usage, audit, clock.**
//!
//! [`holger_plugin_host::MeterHostServices`] is the seam; until this file existed
//! the only implementation that shipped was `NoServices`, which refuses all
//! three by name. The meter plane was complete and inert: modules loaded, the
//! census reported them, and nothing they could ask for had an answer.
//!
//! # The grant is still the authority, and this file adds no way around it
//!
//! Nothing here is reachable by a guest that was not granted the matching
//! capability. There are two enforcements above this file and neither is
//! weakened by anything in it:
//!
//! * `GrantedHost::check` gates every method on the native path; and
//! * on the wasm path the LINKER is built from the grant, so an ungranted
//! import is not defined and the module does not instantiate.
//!
//! So `usage_read` here is only ever called for a module granted `usage.read`,
//! `audit_write` only for `audit.write`, `clock` only for `clock.read`. This
//! implementation deliberately performs **no capability check of its own** —
//! not because it is trusted, but because a second, independent copy of the rule
//! is a second place for the rule to be wrong. There is one authority: the
//! grant table.
//!
//! # ⚠ The store this lends is not a write surface
//!
//! It never was, and it must not become one to make wiring easier. A meter that
//! can write into the store it measures can move its own number; the open
//! guest's selftest is a read-only cross-check for exactly that reason. The
//! store side is `crate::…DirStore` handed in at load time and bounded by
//! `store.read` / `store.write` / `store.list` in the grant — this file does not
//! touch it.
//!
//! # ⚠ Why the audit stamp is a thread-local and not a field
//!
//! [`WireAuditRecord`] deliberately carries no time, no identity and no source
//! address: the host stamps all three "from the call it is already inside", so a
//! guest can never file a record naming somebody else. But `MeterHostServices`
//! is built ONCE at load time and shared by every concurrent request, so there
//! is no field on it that could hold "the call it is already inside" without
//! two requests overwriting each other's identity.
//!
//! Every call into a meter is synchronous and runs on exactly one thread — the
//! `spawn_blocking` worker for the `ReadUsage` door, the promotion thread for
//! the gate. So the stamp is a thread-local installed by [`in_call`] for the
//! duration of that one call, and restored after. A record written with no call
//! in scope is stamped [`UNATTRIBUTED`] rather than being given somebody's
//! identity or being silently dropped.
use std::cell::RefCell;
use std::sync::Arc;
use holger_plugin_host::abi::v2::{
meters, ClockSource, Enforcement, MeterScope as WireScope, ReadingValue, WireAuditRecord,
WireInstant, WireMeterReading,
};
use holger_plugin_host::{MeterHostServices, MeterRuntime};
use crate::audit::{AuditAction, AuditEvent, AuditLog};
use crate::exposed::fast_routes::FastRoutes;
use crate::grpc::holger_proto::{MeterReading, MeterScope as ProtoScope};
use crate::metering::Counted;
/// The identity an audit record is stamped with when a meter wrote one outside
/// any request. Named rather than blank: "nobody was on the other end of this"
/// and "the identity was not recorded" are different facts.
pub const UNATTRIBUTED: &str = "meter(unattributed)";
/// Who the host is serving right now, for the duration of one synchronous call
/// into a meter guest.
#[derive(Debug, Clone, Default)]
pub struct CallStamp {
/// `AuditEvent::ident` — ONE spelling of identity across this server.
pub ident: String,
/// Client source address (`ip:port`) as the transport saw it; empty when it
/// exposed none.
pub source_ip: String,
}
thread_local! {
static CURRENT_CALL: RefCell<Option<CallStamp>> = const { RefCell::new(None) };
}
/// Run `f` with `stamp` installed as the call a meter guest is inside.
///
/// Restores whatever was there before, so a nested call cannot leave a stale
/// identity behind on the worker thread for the next request that lands on it.
pub fn in_call<R>(stamp: CallStamp, f: impl FnOnce() -> R) -> R {
let previous = CURRENT_CALL.with(|c| c.borrow_mut().replace(stamp));
let out = f();
CURRENT_CALL.with(|c| *c.borrow_mut() = previous);
out
}
/// The call in scope on this thread, if any.
pub fn current_call() -> Option<CallStamp> {
CURRENT_CALL.with(|c| c.borrow().clone())
}
/// **holger, standing behind the three services.**
///
/// Holds the route table (the usage source counts through it) and the server's
/// one append-only audit sink. Both are the same objects the doors use; there is
/// no second counter and no second log.
pub struct HolgerMeterServices {
routes: FastRoutes,
audit: Arc<dyn AuditLog>,
/// The filesystem the product-scope storage meters are measured over — the
/// SAME data root the door uses, so a guest and the door cannot disagree
/// about how much storage there is.
///
/// `None` ⇒ the three storage meters are refused by name
/// ([`crate::metering::NO_FILESYSTEM_SOURCE`]) rather than answered zero.
data_root: Option<std::path::PathBuf>,
/// Where each repository's bytes are (`disk_bytes`); empty ⇒ that meter is
/// refused by name.
storage_places: crate::metering::StoragePlaces,
}
impl HolgerMeterServices {
pub fn new(routes: FastRoutes, audit: Arc<dyn AuditLog>) -> Self {
Self {
routes,
audit,
data_root: None,
storage_places: Default::default(),
}
}
/// **Install where each repository's bytes are** (builder style).
pub fn with_storage_places(mut self, places: crate::metering::StoragePlaces) -> Self {
self.storage_places = places;
self
}
/// **Install the data root the storage meters measure** (builder style).
///
/// A builder rather than a fourth argument to [`new`] because the serve path
/// is the only caller that HAS one: a test standing this up over a fixture
/// route table has no filesystem it means to measure, and the honest answer
/// there is the named refusal, not the test runner's own disk.
///
/// [`new`]: HolgerMeterServices::new
pub fn with_data_root(mut self, root: impl Into<std::path::PathBuf>) -> Self {
self.data_root = Some(root.into());
self
}
}
/// holger's proto scope → the plugin ABI's. `Repository` is one tenant;
/// `Server` is the whole product. The ABI has no third word and neither has the
/// proto, so this is total in both directions.
fn scope_to_wire(s: ProtoScope) -> WireScope {
match s {
ProtoScope::Repository => WireScope::Tenant,
// `Unspecified` cannot be produced by `count`, and mapping it to
// `Product` rather than inventing a variant keeps this total.
ProtoScope::Server | ProtoScope::Unspecified => WireScope::Product,
}
}
fn scope_from_wire(s: WireScope) -> ProtoScope {
match s {
WireScope::Tenant => ProtoScope::Repository,
WireScope::Product => ProtoScope::Server,
}
}
impl From<Counted> for WireMeterReading {
fn from(c: Counted) -> Self {
WireMeterReading {
meter: c.name.to_string(),
scope: scope_to_wire(c.scope),
// Everything holger counts is MEASURED. That is not modesty: a bound
// may be attached only to an `Enforced` meter, so saying `Measured`
// here is what makes a cap against these meters refusable up front.
enforcement: Enforcement::Measured,
value: match c.value {
Ok(v) => ReadingValue::Counted(v),
// The refusal keeps its reason across the ABI. This is the whole
// purpose of `ReadingValue`, and the reason a guest can pass a
// number it could not get through without turning it into a zero.
Err(reason) => ReadingValue::Refused { reason },
},
}
}
}
impl MeterHostServices for HolgerMeterServices {
/// **The usage source.** The same [`crate::metering::count`] the `ReadUsage`
/// door reads, so a guest reports numbers this server would have reported
/// itself.
fn usage_read(&self, repository: &str) -> Result<Vec<WireMeterReading>, String> {
Ok(crate::metering::count(
&self.routes,
repository,
self.data_root.as_deref(),
Some(&self.storage_places),
)?
.into_iter()
.map(WireMeterReading::from)
.collect())
}
/// **The audit sink.** The server's one append-only log, with the time, the
/// identity and the source address stamped by the host from the call it is
/// already inside — never by the guest.
fn audit_write(&self, record: &WireAuditRecord) -> Result<(), String> {
let stamp = current_call().unwrap_or_else(|| CallStamp {
ident: UNATTRIBUTED.to_string(),
source_ip: String::new(),
});
// `AuditEvent::new` stamps `ts_nanos` from the host clock. An unknown
// action degrades to `Other` rather than failing, so a newer guest
// writing a newer verb into an older holger still gets a record.
let event = AuditEvent::new(
stamp.ident,
AuditAction::from_wire(&record.action),
record.repository.clone(),
record.artifact.clone(),
stamp.source_ip,
record.status,
record.bytes,
)
.with_detail(record.detail.clone());
self.audit
.record(event)
.map_err(|e| format!("the audit sink refused the record: {e}"))
}
/// **The clock.** holger's wall clock, said to be exactly that.
///
/// `trusted_until_unix_nanos: 0` is "no horizon, do not extrapolate", which
/// is the honest answer: this server has no signed time source, so a guest
/// must not treat this reading as good for any interval at all.
fn clock(&self) -> Result<WireInstant, String> {
let unix_nanos = crate::audit::now_nanos();
Ok(WireInstant {
unix_nanos,
source: ClockSource::HostWall,
trusted_until_unix_nanos: 0,
})
}
}
/// **Ask the loaded guests for one repository's usage**, and put what comes back
/// on the proto wire.
///
/// The readings cross the ABI twice: out through `usage.read` and back out of
/// the guest's `read` export. `source` is `meter:<module>` on every row, so an
/// operator reading a `ReadUsage` response can tell a number the guest narrated
/// from one this server counted for itself.
///
/// More than one meter may be loaded. Each is asked, in catalog order, and the
/// FIRST row for a given meter name wins — a later module cannot overwrite an
/// earlier one's answer. Nothing is silently dropped: a guest that errors fails
/// the whole call rather than leaving a gap that would read as a zero.
pub fn read_through_guests(
guests: &[Arc<dyn MeterRuntime>],
repository: &str,
) -> Result<Vec<MeterReading>, String> {
let mut out: Vec<MeterReading> = Vec::new();
for g in guests {
let module = g.manifest().name.clone();
let source = format!("meter:{module}");
for row in g
.read(repository)
.map_err(|e| format!("meter `{module}`: {e}"))?
{
if out.iter().any(|r| r.name == row.meter) {
continue;
}
let scope = scope_from_wire(row.scope);
out.push(match row.value {
ReadingValue::Counted(v) => MeterReading {
name: row.meter,
value: v,
scope: scope as i32,
source: source.clone(),
counted: true,
refusal_reason: String::new(),
},
// A refusal that survived the round trip is still a refusal.
// `value: 0` because the field must hold something, and
// `counted: false` is what says the zero is not a measurement.
ReadingValue::Refused { reason } => MeterReading {
name: row.meter,
value: 0,
scope: scope as i32,
source: source.clone(),
counted: false,
refusal_reason: reason,
},
});
}
}
Ok(out)
}
/// The product-scope capacity figure, straight off the first loaded guest.
///
/// Exposed so the meter plane can be exercised end to end without a socket:
/// `capacity` is the guest export that reads `usage.read` for
/// [`meters::PRODUCT_SCOPE`], which is the empty repository name.
pub fn product_scope() -> &'static str {
meters::PRODUCT_SCOPE
}
#[cfg(test)]
mod tests {
use super::*;
use crate::audit::AuditError;
use std::sync::Mutex;
#[derive(Default)]
struct Recording(Mutex<Vec<AuditEvent>>);
impl AuditLog for Recording {
fn record(&self, event: AuditEvent) -> Result<(), AuditError> {
self.0.lock().unwrap().push(event);
Ok(())
}
}
fn services(audit: Arc<dyn AuditLog>) -> HolgerMeterServices {
HolgerMeterServices::new(FastRoutes::new(Vec::new()), audit)
}
#[test]
fn the_usage_source_is_the_same_counter_the_door_reads() {
let routes = FastRoutes::new(Vec::new());
let svc = HolgerMeterServices::new(routes.clone(), Arc::new(Recording::default()));
let direct = crate::metering::count(&routes, "", None, None).unwrap();
let through = svc.usage_read("").unwrap();
assert_eq!(direct.len(), through.len());
for (d, t) in direct.iter().zip(through.iter()) {
assert_eq!(d.name, t.meter);
}
}
#[test]
fn a_repository_nothing_serves_is_a_named_refusal_not_six_zeroes() {
let svc = services(Arc::new(Recording::default()));
let err = svc.usage_read("nothing-here").unwrap_err();
assert!(err.contains("nothing-here"), "{err}");
assert!(err.contains("no repository named"), "{err}");
}
#[test]
fn the_product_scope_refuses_the_storage_meters_by_name_rather_than_answering_zero() {
let svc = services(Arc::new(Recording::default()));
let rows = svc.usage_read(product_scope()).unwrap();
assert_eq!(rows.len(), 3, "{rows:?}");
for r in &rows {
assert_eq!(r.scope, WireScope::Product);
match &r.value {
ReadingValue::Refused { reason } => {
assert!(reason.contains("no data root"), "{reason}");
}
other => panic!("{} answered {other:?} with no data root wired", r.meter),
}
// A refusal yields no number at all — not a zero.
assert_eq!(r.value.counted(), None);
}
}
/// The other half of the same rule: with a data root wired, the three are
/// MEASURED, and the guest seam carries the numbers rather than the refusal.
#[test]
fn a_wired_data_root_makes_the_three_storage_meters_measured_through_the_guest_seam() {
let svc =
HolgerMeterServices::new(FastRoutes::new(Vec::new()), Arc::new(Recording::default()))
// The repository's own root: a directory that certainly exists on
// every box this test runs on, and whose filesystem is a real one.
.with_data_root(env!("CARGO_MANIFEST_DIR"));
let rows = svc.usage_read(product_scope()).unwrap();
assert_eq!(rows.len(), 3, "{rows:?}");
for r in &rows {
assert_eq!(r.scope, WireScope::Product);
assert!(
r.value.counted().is_some(),
"{} was refused over a data root that exists: {:?}",
r.meter,
r.value
);
}
let by = |name: &str| {
rows.iter()
.find(|r| r.meter == name)
.unwrap()
.value
.counted()
.unwrap()
};
let total = by(crate::metering::METER_STORAGE_TOTAL_BYTES);
let servable = by(crate::metering::METER_STORAGE_SERVABLE_BYTES);
let used = by(crate::metering::METER_STORAGE_USED_BYTES);
assert!(total > 0, "a mounted filesystem with no size");
// The reserve lives in the gap; see `metering::statvfs`.
assert!(
used + servable <= total,
"used {used} + servable {servable} > total {total}"
);
}
#[test]
fn the_audit_stamp_is_the_call_the_host_is_inside_never_the_guests_word() {
let log = Arc::new(Recording::default());
let svc = services(log.clone());
let record = WireAuditRecord {
action: "Put".into(),
repository: "releases".into(),
artifact: "widget@1.0.0".into(),
status: 200,
bytes: 11,
detail: "metered".into(),
};
in_call(
CallStamp {
ident: "sub-123".into(),
source_ip: "10.0.0.1:443".into(),
},
|| svc.audit_write(&record).unwrap(),
);
let events = log.0.lock().unwrap();
assert_eq!(events.len(), 1);
let e = &events[0];
assert_eq!(e.ident, "sub-123");
assert_eq!(e.source_ip, "10.0.0.1:443");
assert_eq!(e.repo, "releases");
assert_eq!(e.artifact, "widget@1.0.0");
assert_eq!(e.bytes, 11);
// The host stamped the time; the record carried none.
assert!(e.ts_nanos > 0);
}
#[test]
fn a_record_written_outside_any_call_is_named_unattributed_not_given_an_identity() {
let log = Arc::new(Recording::default());
let svc = services(log.clone());
svc.audit_write(&WireAuditRecord::default()).unwrap();
assert_eq!(log.0.lock().unwrap()[0].ident, UNATTRIBUTED);
}
#[test]
fn the_stamp_does_not_outlive_the_call_on_a_reused_worker_thread() {
let log = Arc::new(Recording::default());
let svc = services(log.clone());
in_call(
CallStamp {
ident: "sub-9".into(),
source_ip: String::new(),
},
|| {},
);
// Same thread, next request: the previous identity must be gone.
svc.audit_write(&WireAuditRecord::default()).unwrap();
assert_eq!(log.0.lock().unwrap()[0].ident, UNATTRIBUTED);
}
#[test]
fn the_clock_says_it_is_a_wall_clock_and_vouches_for_no_interval() {
let svc = services(Arc::new(Recording::default()));
let t = svc.clock().unwrap();
assert_eq!(t.source, ClockSource::HostWall);
assert_eq!(
t.trusted_until_unix_nanos, 0,
"a server with no signed time source must not vouch for a horizon"
);
assert!(t.unix_nanos > 0);
}
#[test]
fn the_scope_word_survives_the_round_trip_in_both_directions() {
for s in [ProtoScope::Repository, ProtoScope::Server] {
assert_eq!(scope_from_wire(scope_to_wire(s)), s);
}
for s in [WireScope::Tenant, WireScope::Product] {
assert_eq!(scope_to_wire(scope_from_wire(s)), s);
}
}
}