diff --git a/.gitattributes b/.gitattributes index 33940bb..56862bb 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1,3 +1,2 @@ schemas/*.json text eol=lf crates/model-connector/tests/fixtures/wire_v4/* -text -crates/claude-connector/tests/fixtures/*.cc -text diff --git a/crates/claude-connector/Cargo.toml b/crates/claude-connector/Cargo.toml index 13eaad7..6cb9e45 100644 --- a/crates/claude-connector/Cargo.toml +++ b/crates/claude-connector/Cargo.toml @@ -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"] } diff --git a/crates/claude-connector/src/daemon.rs b/crates/claude-connector/src/daemon.rs index 7102d74..e64e473 100644 --- a/crates/claude-connector/src/daemon.rs +++ b/crates/claude-connector/src/daemon.rs @@ -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), +) -> 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("a); #[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, diff --git a/crates/claude-connector/src/lease.rs b/crates/claude-connector/src/lease.rs index 0d4a582..7e503aa 100644 --- a/crates/claude-connector/src/lease.rs +++ b/crates/claude-connector/src/lease.rs @@ -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; diff --git a/crates/claude-connector/src/meter.rs b/crates/claude-connector/src/meter.rs index 0cdc3a4..2f572a2 100644 --- a/crates/claude-connector/src/meter.rs +++ b/crates/claude-connector/src/meter.rs @@ -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 { } } -/// 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, - #[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) -> Self { - Self(path.into()) - } - - fn released(&self) -> Option { - let mut cache = CultCache::new(); - cache.register_entry_type::().ok()?; - for envelope in SingleFileMessagePackBackingStore::new(&self.0) - .pull_all_read_only_snapshot() - .ok()? - { - cache.put_raw_envelope(envelope).ok()?; - } - Some(cache.get::(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, probe: Box, - brake: Box, log: Box, /// No call before this instant, unix ms. not_before_ms: u64, @@ -131,16 +77,10 @@ struct Meter { } impl Meter { - fn new( - cell: Arc, - probe: Box, - brake: Box, - log: Box, - ) -> Self { + fn new(cell: Arc, probe: Box, log: Box) -> 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, probe: Box, - brake: Box, ) -> std::io::Result> { + 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 { + fn reading(status: &str) -> Vec { 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, + status: String, + fails_with: Option, + } + + /// 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, made: Arc, - outcome: Arc, String>>>, + script: Arc>, } impl Probe for Calls { fn call(&self) -> Result { 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); - - impl Brake for Held { - fn held(&self) -> bool { - self.0.load(Ordering::SeqCst) - } - } - struct Rig { meter: Meter, cell: Arc, made: Arc, - outcome: Arc, String>>>, - braked: Arc, + script: Arc>, log: Arc>>, + /// The lease the swarm holds until a test lets go of it. + lease: Option, } 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, &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 { + 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 { - 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); } } diff --git a/crates/claude-connector/src/quota.rs b/crates/claude-connector/src/quota.rs index 8d80a1c..3d57aea 100644 --- a/crates/claude-connector/src/quota.rs +++ b/crates/claude-connector/src/quota.rs @@ -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>); +pub struct QuotaCell { + windows: Mutex>, + asks: Mutex, + 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, +} + +/// A lease request while it is in flight or held. Dropping it is the lease +/// ending, by whatever road. +pub(crate) struct Asking(Arc); + +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) -> Asking { + self.asking_at(unix_ms()) + } + + fn asking_at(self: &Arc, 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) { @@ -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) { - 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)); + } } diff --git a/crates/claude-connector/tests/fixtures/swarm-brake-live-2026-11-03.cc b/crates/claude-connector/tests/fixtures/swarm-brake-live-2026-11-03.cc deleted file mode 100644 index 837a882..0000000 --- a/crates/claude-connector/tests/fixtures/swarm-brake-live-2026-11-03.cc +++ /dev/null @@ -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‘”¥brake·eureka.control.brake.v1´2026-11-03T08:45:01ZÄ/“þ2026-11-03T08:45:01.468996363Z­gamecultadmin \ No newline at end of file