GameCult/ModelConnector
Watch
9
0
Fork
You've already forked ModelConnector
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

meter: driven by the connector's own lease state, not the brake store (findings brake-store-broken-reads-as-held, braked-bounce-unobserved, rejected-reading-backs-off-six-hours, meter-failure-kind-read-by-nobody)

Browse source 
The meter reads the week only while a lease is held or was asked for less
than a staleness period ago. A held swarm brake stops the units, so a braked
or quiet swarm spends nothing; the brake path, the duplicate brake type, the
store read, the frozen store fixture and the chrono serde feature are gone.
A request is noted by the QuotaCell for as long as it lives (a 4 h lease
counts) and wakes the meter. A lease asked of a metered cell with no weekly
reading (every restart) waits up to 10 s for the meter's first read before it
is lent, so it is not lent unobserved for its whole life.

A call is a miss only if it left the reading unrefreshed: a rejected turn
that reports its windows and then errors is no miss, so a rejected reading is
kept fresh every period. The failure kind is logged on change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-03 12:45:21 +01:00
commit a98594ed39
7 changed files with 480 additions and 245 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitattributes vendored
View file

@ -1,3 +1,2 @@
schemas/*.json text eol=lf
crates/model-connector/tests/fixtures/wire_v4/* -text
crates/claude-connector/tests/fixtures/*.cc -text

2
crates/claude-connector/Cargo.toml
View file

@ -13,7 +13,7 @@ path = "src/main.rs"
[dependencies]
anyhow = "1"
chrono = { version = "0.4", default-features = false, features = ["serde"] }
chrono = { version = "0.4", default-features = false }
hmac = "0.12"
cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a4eae9056152335662f1115f97bf35fb00ab6d23", package = "cultcache-rs" }
model-connector = { path = "../model-connector", features = ["daemon"] }

57
crates/claude-connector/src/daemon.rs
View file

@ -23,7 +23,7 @@ use crate::lease::{LeaseCaller, LeaseLimits, serve_leases, user_uid};
#[cfg(target_os = "linux")]
use crate::unit::SystemdUnits;
#[cfg(unix)]
use crate::meter::{SWARM_BRAKE_STORE, SwarmBrakeStore};
use crate::meter;
use crate::{
ClaudeCodeBackend, ClaudeCodeConfig, ClaudeCodeLimits, CredentialTokenSource, StderrSink,
};
@ -537,14 +537,16 @@ pub fn check_ready(config: &ClaudeConnectorConfig) -> Result<(), ConfigError> {
/// Loads the document and serves until the listener fails.
#[cfg(unix)]
pub fn serve_config(path: &Path) -> Result<(), ConfigError> {
serve_config_with(path, Path::new(SWARM_BRAKE_STORE))
serve_config_with(path, |_| ())
}
/// [`serve_config`] with the meter reading the swarm brake at `brake_store`:
/// the brake's place is not a setting of the document, so a test reaches it
/// only through here.
/// [`serve_config`] handing the quota cell to `started` once the meter runs:
/// a test stands in for the swarm's lease requests only through here.
#[cfg(unix)]
fn serve_config_with(path: &Path, brake_store: &Path) -> Result<(), ConfigError> {
fn serve_config_with(
path: &Path,
started: impl FnOnce(&Arc<crate::QuotaCell>),
) -> Result<(), ConfigError> {
let config = load_config(path)?;
let admissions = readiness_of(&config)?;
let listen = config.validate()?;
@ -562,16 +564,16 @@ fn serve_config_with(path: &Path, brake_store: &Path) -> Result<(), ConfigError>
)
.map_err(|error| ConfigError::from(ServeError::Service(error)))?,
));
// The meter is running before the first lease can ask, so that lease can
// wait for its first read.
let quota = backend.quota();
meter::start(quota.clone(), Box::new(backend.clone()))
.map_err(|_| ConfigError::Serving("meter"))?;
started(&quota);
#[cfg(target_os = "linux")]
if let Some(socket) = &config.lease_socket {
start_leases(&config, socket, service.clone(), backend.quota())?;
start_leases(&config, socket, service.clone(), quota.clone())?;
}
crate::meter::start(
backend.quota(),
Box::new(backend.clone()),
Box::new(SwarmBrakeStore::new(brake_store)),
)
.map_err(|_| ConfigError::Serving("meter"))?;
serve_service(listener, config.serve_limits(), service, backend)?;
Ok(())
}
@ -598,16 +600,16 @@ mod tests {
}
}
/// `serve` starts the meter beside the listener: with a released brake and
/// no reading of the week, the meter runs the claude binary once and serve
/// stays up. The binary is a script that replays a real captured call.
/// `serve` starts the meter beside the listener, wired to the backend's own
/// cell: it makes no call while nobody asks, and once a lease is asked for
/// it runs the claude binary and serve stays up. The binary is a script
/// that replays a real captured call.
#[cfg(unix)]
#[test]
fn serve_config_starts_the_meter_and_stays_up() {
use std::os::unix::fs::PermissionsExt;
const CALL: &str = include_str!("../tests/fixtures/claude-2.1.312-meter-call.jsonl");
const BRAKE: &[u8] = include_bytes!("../tests/fixtures/swarm-brake-live-2026-11-03.cc");
let dir =
std::env::temp_dir().join(format!("claude-connector-serve-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
@ -625,7 +627,6 @@ mod tests {
std::fs::write(at("claude"), &script).unwrap();
std::fs::set_permissions(at("claude"), std::fs::Permissions::from_mode(0o755)).unwrap();
std::fs::write(at("caller.key"), "k".repeat(48)).unwrap();
std::fs::write(at("swarm.cc"), BRAKE).unwrap();
std::fs::write(at("credentials/claude-token"), "sk-ant-oat01-fake-token").unwrap();
std::fs::write(at("credentials/claude-handle-secret"), "s".repeat(48)).unwrap();
let config = ClaudeConnectorConfig {
@ -666,10 +667,26 @@ mod tests {
unsafe { std::env::set_var("CREDENTIALS_DIRECTORY", at("credentials")) };
let path = at("connector.cc");
let brake = at("swarm.cc");
let served = std::thread::spawn(move || serve_config_with(&path, &brake));
let lease = std::sync::Arc::new(std::sync::Mutex::new(None));
let swarm = lease.clone();
let served = std::thread::spawn(move || {
serve_config_with(&path, |cell| {
*swarm.lock().unwrap() = Some(cell.clone());
})
});
let deadline = std::time::Instant::now() + Duration::from_secs(30);
let cell = loop {
if let Some(cell) = lease.lock().unwrap().clone() {
break cell;
}
assert!(std::time::Instant::now() < deadline, "serve never started");
std::thread::sleep(Duration::from_millis(20));
};
std::thread::sleep(Duration::from_millis(500));
assert!(!at("spawned").exists(), "a quiet swarm spends nothing");
let _lease = cell.asking();
while !at("spawned").exists() {
assert!(
std::time::Instant::now() < deadline,

14
crates/claude-connector/src/lease.rs
View file

@ -468,8 +468,14 @@ impl Lender {
);
}
// The lease is claimed before the token is read, let alone lent. The
// plan the refusal reads is the one this process last observed.
let plan = self.quota.snapshot();
// plan the refusal reads is the one this process last observed, but a
// cell with no weekly reading (a restart empties it) gets the meter's
// first read, bounded, before the lease is lent "unobserved" for its
// whole life. The request is the swarm's sign of life, however it
// ends: the meter reads while one is held or lately asked for.
let _asking = self.quota.asking();
let plan = self.quota.snapshot_with_week(FIRST_READ_WAIT);
let now = unix_ms();
let admission = self
.service
.lock()
@ -603,6 +609,10 @@ impl Lender {
}
}
/// The longest a lease waits for the meter's first read of the week. A call
/// takes a few seconds; the holder's own wait for the grant is far longer.
const FIRST_READ_WAIT: Duration = Duration::from_secs(10);
/// How often a held lease looks again at its holder's unit invocation.
const HOLD_POLL_MS: u64 = 500;

431
crates/claude-connector/src/meter.rs
View file

@ -1,42 +1,33 @@
//! The meter: keeps the plan's weekly reading fresh. The subscription reports
//! its windows only inside the CLI's `rate_limit_event`, on a call the CLI
//! makes, and Personas make theirs under leases that report nothing back. So
//! when the `seven_day` reading in the [`QuotaCell`] is `STALE` old, the
//! meter makes the smallest real call through the backend's own path (one
//! `claude -p` on the cheapest model with a trivial prompt) and the stream
//! fills the cell exactly as it does for any call. The meter holds no
//! credential, speaks no protocol and writes nothing itself.
//! The meter: keeps the plan's weekly reading fresh while the swarm is
//! working. The subscription reports its windows only inside the CLI's
//! `rate_limit_event`, on a call the CLI makes, and Personas make theirs under
//! leases that report nothing back. So when the `seven_day` reading in the
//! [`QuotaCell`] is `STALE` old, the meter makes the smallest real call
//! through the backend's own path (one `claude -p` on the cheapest model with
//! a trivial prompt) and the stream fills the cell exactly as it does for any
//! call. The meter holds no credential, speaks no protocol and writes nothing
//! itself.
//!
//! A held swarm brake spends nothing: the meter reads the swarm's brake store
//! before every call and, like every other organ that acts, treats anything
//! but a decodable released brake as held. A call that fails, or completes
//! without refreshing the reading, is not retried at the staleness period: the
//! wait doubles per consecutive miss, up to `CEILING`.
//! The meter is driven by the connector's own lease state. It reads only
//! while a lease is held or one was asked for less than `STALE` ago: a held
//! swarm brake stops the units, they ask for nothing, and a braked or quiet
//! swarm spends nothing. A call that fails, or completes without refreshing
//! the reading, is not retried at the staleness period: the wait doubles per
//! consecutive miss, up to `CEILING`.
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use chrono::{DateTime, Utc};
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use model_connector::Usage;
use crate::backend::ClaudeCodeBackend;
use crate::quota::{EXPIRED, QuotaCell};
use crate::quota::{EXPIRED, QuotaCell, WEEK};
/// Where the swarm's brake is kept: the operator's root-owned, world-readable
/// store, the same file the Minds' organs read before they act.
pub(crate) const SWARM_BRAKE_STORE: &str = "/etc/gamecult/minds/swarm.cc";
/// The one key the brake document has in its store.
const BRAKE_KEY: &str = "brake";
/// The window the meter keeps fresh.
const WINDOW: &str = "seven_day";
/// A reading this old, or older, is stale; also the least gap between calls.
const STALE: Duration = Duration::from_secs(15 * 60);
/// The longest any backoff may wait.
/// The longest any backoff may wait; also how long an idle meter sleeps
/// before looking again, since a lease request wakes it.
const CEILING: Duration = Duration::from_secs(6 * 60 * 60);
/// How often a held brake is looked at again.
const BRAKE_RECHECK: Duration = Duration::from_secs(60);
/// The smallest real call, as the backend makes it.
pub(crate) trait Probe: Send {
@ -51,69 +42,25 @@ impl Probe for Arc<ClaudeCodeBackend> {
}
}
/// Whether the operator's swarm brake holds.
pub(crate) trait Brake: Send {
fn held(&self) -> bool;
}
/// The brake document of the operator's control stores. Declared here with the
/// type name and field order the stores are written with; the connector reads
/// it and never writes it.
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "eureka.control.brake.v1")]
struct SwarmBrake {
#[cultcache(key = 0)]
released: bool,
#[cultcache(key = 1)]
set_at: DateTime<Utc>,
#[cultcache(key = 2)]
set_by: String,
}
/// The swarm brake read from its store, read-only and without locks. Absent,
/// unreadable, undecodable or unreleased is held.
pub(crate) struct SwarmBrakeStore(PathBuf);
impl SwarmBrakeStore {
pub(crate) fn new(path: impl Into<PathBuf>) -> Self {
Self(path.into())
}
fn released(&self) -> Option<bool> {
let mut cache = CultCache::new();
cache.register_entry_type::<SwarmBrake>().ok()?;
for envelope in SingleFileMessagePackBackingStore::new(&self.0)
.pull_all_read_only_snapshot()
.ok()?
{
cache.put_raw_envelope(envelope).ok()?;
}
Some(cache.get::<SwarmBrake>(BRAKE_KEY).ok()??.released)
}
}
impl Brake for SwarmBrakeStore {
fn held(&self) -> bool {
self.released() != Some(true)
}
}
/// Why the meter is waiting; logged once per change.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, PartialEq, Eq)]
enum State {
Read,
Braked,
Failed,
Idle,
/// The call failed; the kind is the failure's kind, never its message.
Failed(String),
Unread,
}
impl State {
fn line(self) -> &'static str {
fn line(&self) -> String {
match self {
Self::Read => "meter: weekly reading refreshed",
Self::Braked => "meter: swarm brake held, making no call",
Self::Failed => "meter: the call failed, backing off",
Self::Unread => "meter: the call completed without a weekly reading, backing off",
Self::Read => "meter: weekly reading refreshed".to_string(),
Self::Idle => "meter: no lease asked lately, making no call".to_string(),
Self::Failed(kind) => format!("meter: the call failed ({kind}), backing off"),
Self::Unread => {
"meter: the call completed without a weekly reading, backing off".to_string()
}
}
}
}
@ -121,7 +68,6 @@ impl State {
struct Meter {
cell: Arc<QuotaCell>,
probe: Box<dyn Probe>,
brake: Box<dyn Brake>,
log: Box<dyn FnMut(&str) + Send>,
/// No call before this instant, unix ms.
not_before_ms: u64,
@ -131,16 +77,10 @@ struct Meter {
}
impl Meter {
fn new(
cell: Arc<QuotaCell>,
probe: Box<dyn Probe>,
brake: Box<dyn Brake>,
log: Box<dyn FnMut(&str) + Send>,
) -> Self {
fn new(cell: Arc<QuotaCell>, probe: Box<dyn Probe>, log: Box<dyn FnMut(&str) + Send>) -> Self {
Self {
cell,
probe,
brake,
log,
not_before_ms: 0,
misses: 0,
@ -149,28 +89,30 @@ impl Meter {
}
/// Makes the call if one is due at `now_ms`; returns how long to wait
/// before looking again. Nothing is due while the brake is held, nor
/// before the last miss's backoff has passed, nor while the weekly reading
/// is younger than `STALE`.
/// before looking again. Nothing is due while no lease is held or lately
/// asked for, nor before the last miss's backoff has passed, nor while the
/// weekly reading is younger than `STALE`.
fn tick(&mut self, now_ms: u64) -> Duration {
if self.brake.held() {
self.report(State::Braked);
return BRAKE_RECHECK;
if !self.cell.wanted_at(now_ms, STALE) {
self.report(State::Idle);
return CEILING;
}
let due_at = self.due_at_ms();
if now_ms < due_at {
return Duration::from_millis(due_at - now_ms);
}
// A failed call may still have refreshed the reading: a rejected turn
// reports its windows and then errors. Only the cell says whether the
// week was read.
let called = self.probe.call();
let state = if called.is_ok() && self.refreshed_since(now_ms) {
let state = if self.refreshed_since(now_ms) {
self.misses = 0;
State::Read
} else {
self.misses = self.misses.saturating_add(1);
if called.is_ok() {
State::Unread
} else {
State::Failed
match called {
Ok(_) => State::Unread,
Err(kind) => State::Failed(kind),
}
};
self.report(state);
@ -186,7 +128,7 @@ impl Meter {
.snapshot()
.windows
.iter()
.find(|quota| quota.window == WINDOW)
.find(|quota| quota.window == WEEK)
.map_or(0, |quota| {
if quota.status == EXPIRED {
0
@ -202,20 +144,21 @@ impl Meter {
.snapshot()
.windows
.iter()
.any(|quota| quota.window == WINDOW && quota.observed_at_unix_ms >= since_ms)
.any(|quota| quota.window == WEEK && quota.observed_at_unix_ms >= since_ms)
}
fn report(&mut self, state: State) {
if self.last_state != Some(state) {
(self.log)(state.line());
if self.last_state.as_ref() != Some(&state) {
(self.log)(&state.line());
self.last_state = Some(state);
}
}
fn run(mut self) {
loop {
let seen = self.cell.asks_so_far();
let wait = self.tick(unix_ms());
std::thread::sleep(wait);
self.cell.sleep_until_asked(seen, wait);
}
}
}
@ -238,16 +181,16 @@ fn unix_ms() -> u64 {
}
/// Starts the meter thread. It writes only through the calls it makes, and
/// logs one line to stderr when its state changes.
/// logs one line to stderr when its state changes. From here a lease asked of
/// a cell with no weekly reading waits for the meter's first read.
pub(crate) fn start(
cell: Arc<QuotaCell>,
probe: Box<dyn Probe>,
brake: Box<dyn Brake>,
) -> std::io::Result<std::thread::JoinHandle<()>> {
cell.attach_meter();
let meter = Meter::new(
cell,
probe,
brake,
Box::new(|line| eprintln!("claude-connector: {line}")),
);
std::thread::Builder::new()
@ -258,39 +201,51 @@ pub(crate) fn start(
#[cfg(test)]
mod tests {
use super::*;
use crate::quota::Asking;
use model_connector::PlanQuota;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::atomic::{AtomicUsize, Ordering};
const MIN: u64 = 60_000;
const NOW: u64 = 1_800_000_000_000;
const RESET: u64 = 4_102_444_800_000;
fn reading() -> Vec<PlanQuota> {
fn reading(status: &str) -> Vec<PlanQuota> {
vec![PlanQuota {
window: WINDOW.to_string(),
window: WEEK.to_string(),
utilization_permille: Some(60),
resets_at_unix_ms: RESET,
status: "allowed".to_string(),
status: status.to_string(),
observed_at_unix_ms: 0,
}]
}
/// A call that answers as scripted and, when it succeeds with a reading,
/// feeds the cell the way the stream does.
/// What the next call does: the instant its stream stamps a weekly
/// reading (if any) and the kind it then fails with (if any).
#[derive(Clone, Default)]
struct Script {
reading_at: Option<u64>,
status: String,
fails_with: Option<String>,
}
/// A call that answers as scripted and feeds the cell the way the stream
/// does, before it fails, as a rejected turn does.
struct Calls {
cell: Arc<QuotaCell>,
made: Arc<AtomicUsize>,
outcome: Arc<Mutex<Result<Option<u64>, String>>>,
script: Arc<Mutex<Script>>,
}
impl Probe for Calls {
fn call(&self) -> Result<Usage, String> {
self.made.fetch_add(1, Ordering::SeqCst);
match &*self.outcome.lock().unwrap() {
Ok(Some(at)) => self.cell.observe_at(*at, reading()),
Ok(None) => {}
Err(kind) => return Err(kind.clone()),
let script = self.script.lock().unwrap().clone();
if let Some(at) = script.reading_at {
self.cell.observe_at(at, reading(&script.status));
}
if let Some(kind) = script.fails_with {
return Err(kind);
}
Ok(Usage {
input_tokens: 385,
@ -302,28 +257,20 @@ mod tests {
}
}
struct Held(Arc<AtomicBool>);
impl Brake for Held {
fn held(&self) -> bool {
self.0.load(Ordering::SeqCst)
}
}
struct Rig {
meter: Meter,
cell: Arc<QuotaCell>,
made: Arc<AtomicUsize>,
outcome: Arc<Mutex<Result<Option<u64>, String>>>,
braked: Arc<AtomicBool>,
script: Arc<Mutex<Script>>,
log: Arc<Mutex<Vec<String>>>,
/// The lease the swarm holds until a test lets go of it.
lease: Option<Asking>,
}
fn rig() -> Rig {
let cell = Arc::new(QuotaCell::new());
let made = Arc::new(AtomicUsize::new(0));
let outcome = Arc::new(Mutex::new(Ok(None)));
let braked = Arc::new(AtomicBool::new(false));
let script = Arc::new(Mutex::new(Script::default()));
let log = Arc::new(Mutex::new(Vec::new()));
let sink = log.clone();
let meter = Meter::new(
@ -331,34 +278,50 @@ mod tests {
Box::new(Calls {
cell: cell.clone(),
made: made.clone(),
outcome: outcome.clone(),
script: script.clone(),
}),
Box::new(Held(braked.clone())),
Box::new(move |line| sink.lock().unwrap().push(line.to_string())),
);
let lease = Some(cell.asking_at(NOW));
Rig {
meter,
cell,
made,
outcome,
braked,
script,
log,
lease,
}
}
impl Rig {
fn answers(&self, outcome: Result<Option<u64>, &str>) {
*self.outcome.lock().unwrap() = outcome.map_err(str::to_string);
fn reads(&self, at: u64) {
*self.script.lock().unwrap() = Script {
reading_at: Some(at),
status: "allowed".to_string(),
fails_with: None,
};
}
fn reads_nothing(&self) {
*self.script.lock().unwrap() = Script::default();
}
fn fails(&self, kind: &str) {
*self.script.lock().unwrap() = Script {
fails_with: Some(kind.to_string()),
..Script::default()
};
}
fn calls(&self) -> usize {
self.made.load(Ordering::SeqCst)
}
fn lines(&self) -> Vec<String> {
self.log.lock().unwrap().clone()
}
}
#[test]
fn an_unread_week_triggers_one_call_and_the_fresh_reading_ends_the_asking() {
let mut rig = rig();
rig.answers(Ok(Some(NOW)));
rig.reads(NOW);
assert_eq!(rig.meter.tick(NOW), STALE, "fresh for the staleness period");
assert_eq!(rig.calls(), 1);
assert_eq!(
@ -371,8 +334,8 @@ mod tests {
#[test]
fn a_stale_reading_triggers_a_call_at_the_period_and_not_before() {
let mut rig = rig();
rig.cell.observe_at(NOW, reading());
rig.answers(Ok(Some(NOW + 15 * MIN)));
rig.cell.observe_at(NOW, reading("allowed"));
rig.reads(NOW + 15 * MIN);
rig.meter.tick(NOW + 15 * MIN - 1);
assert_eq!(rig.calls(), 0, "one millisecond short of stale");
rig.meter.tick(NOW + 15 * MIN);
@ -383,35 +346,67 @@ mod tests {
fn a_window_past_its_reset_is_read_at_once_not_after_the_period() {
let mut rig = rig();
let now = unix_ms();
let mut over = reading().remove(0);
let mut over = reading("allowed").remove(0);
over.resets_at_unix_ms = 1;
rig.cell.observe_at(now, vec![over]);
rig.answers(Ok(Some(now)));
rig.reads(now);
rig.meter.tick(now);
assert_eq!(rig.calls(), 1);
}
#[test]
fn a_held_brake_makes_no_call_however_stale_the_reading() {
fn a_swarm_that_asks_for_nothing_spends_nothing_however_stale_the_reading() {
let mut rig = rig();
rig.braked.store(true, Ordering::SeqCst);
drop(rig.lease.take());
for step in 0..10 {
assert_eq!(rig.meter.tick(NOW + step * 60 * MIN), BRAKE_RECHECK);
let at = NOW + 15 * MIN + step * 60 * MIN;
assert_eq!(rig.meter.tick(at), CEILING);
}
assert_eq!(rig.calls(), 0);
assert_eq!(*rig.log.lock().unwrap(), vec![State::Braked.line()]);
rig.braked.store(false, Ordering::SeqCst);
rig.answers(Ok(Some(NOW + 600 * MIN)));
rig.meter.tick(NOW + 600 * MIN);
assert_eq!(rig.calls(), 1, "released, the stale week is read at once");
assert_eq!(rig.lines(), vec![State::Idle.line()]);
}
#[test]
fn a_lease_asked_for_wakes_the_meter_and_the_empty_cell_is_read() {
let mut rig = rig();
drop(rig.lease.take());
rig.reads(NOW + 180 * MIN);
rig.meter.tick(NOW + 180 * MIN);
assert_eq!(rig.calls(), 0, "asked three hours ago and let go");
let _later = rig.cell.asking_at(NOW + 180 * MIN);
rig.meter.tick(NOW + 180 * MIN);
assert_eq!(rig.calls(), 1);
}
#[test]
fn a_lease_let_go_is_still_wanted_for_exactly_one_staleness_period() {
let mut rig = rig();
drop(rig.lease.take());
rig.reads(NOW + 15 * MIN - 1);
rig.meter.tick(NOW + 15 * MIN - 1);
assert_eq!(rig.calls(), 1, "one millisecond inside the period");
rig.reads(NOW + 30 * MIN);
rig.meter.tick(NOW + 30 * MIN);
assert_eq!(rig.calls(), 1, "a period after the last ask: unwanted");
}
#[test]
fn a_lease_held_for_four_hours_keeps_the_week_read_throughout() {
let mut rig = rig();
for step in 0..=16 {
let at = NOW + step * 15 * MIN;
rig.reads(at);
rig.meter.tick(at);
}
assert_eq!(rig.calls(), 17, "one read per period, the lease never lapsing");
}
#[test]
fn a_failed_call_leaves_the_cell_and_backs_off_instead_of_hammering() {
let mut rig = rig();
rig.cell.observe_at(NOW, reading());
rig.cell.observe_at(NOW, reading("allowed"));
let before = rig.cell.snapshot();
rig.answers(Err("Provider"));
rig.fails("Provider");
let mut at = NOW + 15 * MIN;
let mut waits = Vec::new();
for _ in 0..4 {
@ -427,7 +422,27 @@ mod tests {
);
rig.meter.tick(at - MIN);
assert_eq!(rig.calls(), 4, "nothing is asked before the backoff ends");
assert_eq!(*rig.log.lock().unwrap(), vec![State::Failed.line()]);
assert_eq!(
rig.lines(),
vec!["meter: the call failed (Provider), backing off"]
);
}
#[test]
fn the_failure_kind_is_logged_again_when_it_changes_and_not_when_it_repeats() {
let mut rig = rig();
rig.fails("Authentication");
rig.meter.tick(NOW);
rig.meter.tick(NOW + 15 * MIN);
rig.fails("InvalidRequest");
rig.meter.tick(NOW + 45 * MIN);
assert_eq!(
rig.lines(),
vec![
"meter: the call failed (Authentication), backing off",
"meter: the call failed (InvalidRequest), backing off",
]
);
}
#[test]
@ -436,11 +451,11 @@ mod tests {
assert_eq!(backoff(1), STALE);
assert_eq!(backoff(u32::MAX), CEILING);
let mut rig = rig();
rig.answers(Err("Provider"));
rig.fails("Provider");
rig.meter.tick(NOW);
rig.meter.tick(NOW + 15 * MIN);
assert_eq!(rig.meter.misses, 2);
rig.answers(Ok(Some(NOW + 45 * MIN)));
rig.reads(NOW + 45 * MIN);
rig.meter.tick(NOW + 45 * MIN);
assert_eq!(rig.meter.misses, 0);
}
@ -448,18 +463,34 @@ mod tests {
#[test]
fn a_call_that_completes_without_a_weekly_reading_is_a_miss_too() {
let mut rig = rig();
rig.answers(Ok(None));
rig.reads_nothing();
assert_eq!(rig.meter.tick(NOW), STALE);
assert_eq!(rig.meter.tick(NOW + 15 * MIN), 2 * STALE);
assert_eq!(rig.calls(), 2);
assert_eq!(*rig.log.lock().unwrap(), vec![State::Unread.line()]);
assert_eq!(rig.lines(), vec![State::Unread.line()]);
}
#[test]
fn a_rejected_turn_that_refreshed_the_reading_and_then_errored_is_no_miss() {
let mut rig = rig();
*rig.script.lock().unwrap() = Script {
reading_at: Some(NOW),
status: "rejected".to_string(),
fails_with: Some("RateLimited".to_string()),
};
assert_eq!(rig.meter.tick(NOW), STALE, "the next read is a period away");
assert_eq!(rig.meter.misses, 0);
assert_eq!(rig.lines(), vec![State::Read.line()]);
rig.reads(NOW + 15 * MIN);
rig.meter.tick(NOW + 15 * MIN);
assert_eq!(rig.calls(), 2, "a rejected reading is kept fresh");
}
#[test]
fn a_reading_from_before_the_call_does_not_count_as_its_answer() {
let mut rig = rig();
// A slow concurrent sensor stamped earlier than this call began.
rig.answers(Ok(Some(NOW - 1)));
rig.reads(NOW - 1);
rig.meter.tick(NOW);
assert_eq!(rig.meter.misses, 1);
}
@ -467,10 +498,14 @@ mod tests {
#[test]
fn the_log_says_what_the_meter_is_doing_in_words() {
assert_eq!(
State::Braked.line(),
"meter: swarm brake held, making no call"
State::Idle.line(),
"meter: no lease asked lately, making no call"
);
assert_eq!(State::Read.line(), "meter: weekly reading refreshed");
assert_eq!(
State::Unread.line(),
"meter: the call completed without a weekly reading, backing off"
);
}
#[test]
@ -481,60 +516,28 @@ mod tests {
assert!((before.as_millis() as u64..=after.as_millis() as u64).contains(&read));
}
// ---- the brake store, read as the operator's tool writes it ----
/// The swarm store as it stood on Yggdrasil on 2026-11-03 (released),
/// copied byte for byte.
const LIVE_STORE: &[u8] = include_bytes!("../tests/fixtures/swarm-brake-live-2026-11-03.cc");
fn store_with(bytes: &[u8]) -> (PathBuf, SwarmBrakeStore) {
static NEXT: AtomicUsize = AtomicUsize::new(0);
let dir = std::env::temp_dir().join(format!(
"claude-meter-brake-{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::SeqCst)
));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("swarm.cc");
std::fs::write(&path, bytes).unwrap();
(dir, SwarmBrakeStore::new(path))
}
/// The live store with `released` flipped to false: the msgpack `true`
/// byte that precedes the timestamp string.
fn held_store() -> Vec<u8> {
let mut bytes = LIVE_STORE.to_vec();
let at = bytes
.windows(5)
.position(|w| w == b"\xc3\xbe202")
.expect("the live store holds released=true before its timestamp");
bytes[at] = 0xc2;
bytes
}
#[test]
fn the_live_released_store_reads_as_released() {
let (dir, store) = store_with(LIVE_STORE);
assert_eq!(store.released(), Some(true));
assert!(!store.held());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn the_same_store_with_the_brake_held_reads_as_held() {
let (dir, store) = store_with(&held_store());
assert_eq!(store.released(), Some(false));
assert!(store.held());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_missing_garbled_or_truncated_store_is_held() {
assert!(SwarmBrakeStore::new("/nonexistent-claude-meter/swarm.cc").held());
for bytes in [&b""[..], b"not a store at all", &LIVE_STORE[..40]] {
let (dir, store) = store_with(bytes);
assert!(store.held(), "{bytes:?}");
let _ = std::fs::remove_dir_all(dir);
fn a_started_meter_sleeps_until_a_lease_asks_and_then_reads() {
let cell = Arc::new(QuotaCell::new());
let made = Arc::new(AtomicUsize::new(0));
let probe = Calls {
cell: cell.clone(),
made: made.clone(),
script: Arc::new(Mutex::new(Script {
reading_at: Some(unix_ms() + 60 * MIN),
status: "allowed".to_string(),
fails_with: None,
})),
};
start(cell.clone(), Box::new(probe)).unwrap();
std::thread::sleep(Duration::from_millis(300));
assert_eq!(made.load(Ordering::SeqCst), 0, "no lease, no call");
let _lease = cell.asking();
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while made.load(Ordering::SeqCst) == 0 {
assert!(std::time::Instant::now() < deadline, "the ask woke nobody");
std::thread::sleep(Duration::from_millis(20));
}
assert_eq!(cell.snapshot_with_week(Duration::ZERO).windows.len(), 1);
}
}

217
crates/claude-connector/src/quota.rs
View file

@ -6,8 +6,9 @@
//! each lease.
use std::collections::BTreeMap;
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use model_connector::{PlanQuota, PlanQuotaSnapshot};
@ -16,18 +17,114 @@ pub(crate) const WEEK: &str = "seven_day";
/// The windows the pacing reads, and so the only ones the cell keeps.
pub(crate) const PACED: [&str; 2] = ["five_hour", WEEK];
/// How often a lease waiting for its first reading looks at the cell again.
const FIRST_READ_POLL: Duration = Duration::from_millis(50);
/// The latest observation of each plan window, shared by the backend that
/// feeds it and whatever reads it out. Each window carries the time of its
/// own observation: an event that names one window says nothing about the
/// others.
///
/// It also keeps who is asking: a lease request is the swarm's only sign of
/// life the connector sees, and the meter reads the plan only while someone
/// asks.
#[derive(Default)]
pub struct QuotaCell(Mutex<BTreeMap<String, PlanQuota>>);
pub struct QuotaCell {
windows: Mutex<BTreeMap<String, PlanQuota>>,
asks: Mutex<Asks>,
asked: Condvar,
metered: AtomicBool,
}
#[derive(Default)]
struct Asks {
/// Lease requests in flight or held.
live: usize,
/// Requests ever made; the meter sleeps until it changes.
count: u64,
/// When the latest request came, unix ms.
last_ms: Option<u64>,
}
/// A lease request while it is in flight or held. Dropping it is the lease
/// ending, by whatever road.
pub(crate) struct Asking(Arc<QuotaCell>);
impl Drop for Asking {
fn drop(&mut self) {
let mut asks = self.0.asks.lock().unwrap_or_else(|e| e.into_inner());
asks.live = asks.live.saturating_sub(1);
}
}
impl QuotaCell {
pub fn new() -> Self {
Self::default()
}
/// Notes a lease request, now, and wakes the meter.
pub(crate) fn asking(self: &Arc<Self>) -> Asking {
self.asking_at(unix_ms())
}
fn asking_at(self: &Arc<Self>, now: u64) -> Asking {
{
let mut asks = self.asks.lock().unwrap_or_else(|e| e.into_inner());
asks.live += 1;
asks.count += 1;
asks.last_ms = Some(now);
}
self.asked.notify_all();
Asking(self.clone())
}
/// Whether a lease is held, or one was asked for less than `within` ago.
pub(crate) fn wanted_at(&self, now: u64, within: Duration) -> bool {
let asks = self.asks.lock().unwrap_or_else(|e| e.into_inner());
asks.live > 0
|| asks.last_ms.is_some_and(|last| {
now < last.saturating_add(u64::try_from(within.as_millis()).unwrap_or(u64::MAX))
})
}
pub(crate) fn asks_so_far(&self) -> u64 {
self.asks.lock().unwrap_or_else(|e| e.into_inner()).count
}
/// Sleeps up to `limit`, or until a request beyond the first `seen` comes.
pub(crate) fn sleep_until_asked(&self, seen: u64, limit: Duration) {
let asks = self.asks.lock().unwrap_or_else(|e| e.into_inner());
drop(
self.asked
.wait_timeout_while(asks, limit, |asks| asks.count == seen)
.unwrap_or_else(|e| e.into_inner()),
);
}
/// Declares that a meter reads this cell, so a first lease may wait for it.
pub(crate) fn attach_meter(&self) {
self.metered.store(true, Ordering::SeqCst);
}
/// The windows as they stand, but with no weekly reading yet (a restart
/// leaves the cell empty) a metered cell is given up to `within` to get
/// one, so a lease is not lent "unobserved" for want of a few seconds.
pub(crate) fn snapshot_with_week(&self, within: Duration) -> PlanQuotaSnapshot {
let deadline = Instant::now() + within;
loop {
let snapshot = self.snapshot();
let read = snapshot
.windows
.iter()
.any(|quota| quota.window == WEEK && quota.status != EXPIRED);
let now = Instant::now();
if read || !self.metered.load(Ordering::SeqCst) || now >= deadline {
return snapshot;
}
std::thread::sleep(FIRST_READ_POLL.min(deadline - now));
}
}
/// Records the windows one event reported, stamping each with now. See
/// [`QuotaCell::observe_at`] for how a reading combines with what is held.
pub fn observe(&self, windows: impl IntoIterator<Item = PlanQuota>) {
@ -39,7 +136,7 @@ impl QuotaCell {
/// saw meanwhile. Windows the readings did not mention stay as they were,
/// stamp included.
pub fn observe_at(&self, at: u64, windows: impl IntoIterator<Item = PlanQuota>) {
let mut held = self.0.lock().unwrap_or_else(|e| e.into_inner());
let mut held = self.windows.lock().unwrap_or_else(|e| e.into_inner());
for mut quota in windows {
quota.observed_at_unix_ms = at;
if held
@ -59,7 +156,7 @@ impl QuotaCell {
/// `expired` with no utilization: the reading described a window that no
/// longer exists.
fn snapshot_at(&self, now: u64) -> PlanQuotaSnapshot {
let held = self.0.lock().unwrap_or_else(|e| e.into_inner());
let held = self.windows.lock().unwrap_or_else(|e| e.into_inner());
PlanQuotaSnapshot {
windows: held
.values()
@ -195,4 +292,114 @@ mod tests {
assert_eq!(windows[0].utilization_permille, Some(80));
assert_eq!(windows[1].status, "expired");
}
const HOUR: u64 = 3_600_000;
#[test]
fn nobody_wants_the_plan_before_anyone_asks() {
let cell = Arc::new(QuotaCell::new());
assert!(!cell.wanted_at(0, Duration::from_secs(900)));
}
#[test]
fn a_lease_in_flight_or_held_is_wanted_for_as_long_as_it_lives() {
let cell = Arc::new(QuotaCell::new());
let lease = cell.asking_at(1_000);
let other = cell.asking_at(2_000);
let day = Duration::from_secs(86_400);
assert!(cell.wanted_at(1_000 + 4 * HOUR, Duration::from_secs(1)));
drop(lease);
assert!(cell.wanted_at(1_000 + 4 * HOUR, Duration::from_secs(1)), "one lease still lives");
drop(other);
assert!(!cell.wanted_at(1_000 + 4 * HOUR, Duration::from_secs(1)));
assert!(cell.wanted_at(2_000 + 4 * HOUR, day), "within the window of the last ask");
}
#[test]
fn a_lease_let_go_is_wanted_until_exactly_the_period_after_its_ask() {
let cell = Arc::new(QuotaCell::new());
drop(cell.asking_at(10_000));
let period = Duration::from_millis(900);
assert!(cell.wanted_at(10_899, period));
assert!(!cell.wanted_at(10_900, period));
}
#[test]
fn each_ask_is_counted_and_wakes_a_sleeper_at_once() {
let cell = Arc::new(QuotaCell::new());
assert_eq!(cell.asks_so_far(), 0);
let sleeper = {
let cell = cell.clone();
std::thread::spawn(move || {
let started = Instant::now();
cell.sleep_until_asked(0, Duration::from_secs(30));
started.elapsed()
})
};
std::thread::sleep(Duration::from_millis(100));
drop(cell.asking());
assert!(sleeper.join().unwrap() < Duration::from_secs(10), "woken, not timed out");
assert_eq!(cell.asks_so_far(), 1);
// An ask that came after `seen` was read ends the sleep before it starts.
let started = Instant::now();
cell.sleep_until_asked(0, Duration::from_secs(30));
assert!(started.elapsed() < Duration::from_secs(10));
// Nothing new: the sleep is the whole limit.
let started = Instant::now();
cell.sleep_until_asked(1, Duration::from_millis(200));
assert!(started.elapsed() >= Duration::from_millis(200));
}
#[test]
fn a_metered_cell_with_no_week_gives_a_first_lease_the_reading_when_it_lands() {
let cell = Arc::new(QuotaCell::new());
cell.attach_meter();
let feeder = {
let cell = cell.clone();
std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
cell.observe([quota(WEEK, 60, 0)]);
})
};
let plan = cell.snapshot_with_week(Duration::from_secs(20));
feeder.join().unwrap();
assert_eq!(plan.windows.len(), 1, "the lease waited for the week");
assert_eq!(plan.windows[0].window, WEEK);
}
#[test]
fn the_wait_for_the_first_reading_is_bounded() {
let cell = Arc::new(QuotaCell::new());
cell.attach_meter();
let started = Instant::now();
let plan = cell.snapshot_with_week(Duration::from_millis(300));
let waited = started.elapsed();
assert!(plan.windows.is_empty(), "no reading came: unobserved, not hung");
assert!(waited >= Duration::from_millis(300), "{waited:?}");
assert!(waited < Duration::from_secs(10), "{waited:?}");
}
#[test]
fn a_cell_nobody_meters_or_that_holds_the_week_is_never_waited_on() {
let bare = QuotaCell::new();
let started = Instant::now();
assert!(bare.snapshot_with_week(Duration::from_secs(20)).windows.is_empty());
let held = QuotaCell::new();
held.attach_meter();
held.observe([quota(WEEK, 60, 0)]);
assert_eq!(held.snapshot_with_week(Duration::from_secs(20)).windows.len(), 1);
assert!(started.elapsed() < Duration::from_secs(5));
}
#[test]
fn a_week_past_its_reset_is_waited_on_like_no_week() {
let cell = QuotaCell::new();
cell.attach_meter();
let mut over = quota(WEEK, 60, 0);
over.resets_at_unix_ms = 1;
cell.observe([over]);
let started = Instant::now();
cell.snapshot_with_week(Duration::from_millis(300));
assert!(started.elapsed() >= Duration::from_millis(300));
}
}

1
crates/claude-connector/tests/fixtures/swarm-brake-live-2026-11-03.cc vendored
View file

@ -1 +0,0 @@
“²cultcache.store.v1‘—·eureka.control.brake.v1·eureka.control.brake.v1ºeureka.control.brake.v1.v1·eureka.control.brake.v1Ùb{"schemaName":"eureka.control.brake.v1","schemaVersion":"eureka.control.brake.v1.v1","members":[]}‘·eureka.control.brake.v1<EFBFBD>‘”¥brake·eureka.control.brake.v1´2026-11-03T08:45:01ZÄ/“þ2026-11-03T08:45:01.468996363Z­gamecultadmin