usage-meter-poll: the usage poll, the QuotaCell's second sensor
usage_poll asks the subscription's usage endpoint when the seven_day reading is 15 min old or older, maps five_hour and seven_day onto PlanQuota with status polled, and honours Retry-After; failures wait, log once per change of state, and never touch the cell. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
c1802d85be
commit
79792eea85
5 changed files with 1116 additions and 4 deletions
378
Cargo.lock
generated
378
Cargo.lock
generated
|
|
@ -139,6 +139,7 @@ name = "claude-connector"
|
|||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
"cultcache-rs",
|
||||
"hmac",
|
||||
"libc",
|
||||
|
|
@ -149,6 +150,7 @@ dependencies = [
|
|||
"serde_json",
|
||||
"sha2",
|
||||
"thiserror",
|
||||
"ureq",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
|
|
@ -296,6 +298,17 @@ dependencies = [
|
|||
"subtle",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "displaydoc"
|
||||
version = "0.2.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dyn-clone"
|
||||
version = "1.0.20"
|
||||
|
|
@ -338,6 +351,15 @@ version = "0.1.11"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890"
|
||||
|
||||
[[package]]
|
||||
name = "form_urlencoded"
|
||||
version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf"
|
||||
dependencies = [
|
||||
"percent-encoding",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fs2"
|
||||
version = "0.4.3"
|
||||
|
|
@ -459,6 +481,110 @@ dependencies = [
|
|||
"cc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "icu_collections"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fe090a42a0d3140fed8c63263ed1fecc02422cb2f4bfe55cd8de2de1b8e105dd"
|
||||
dependencies = [
|
||||
"displaydoc",
|
||||
"potential_utf",
|
||||
"utf8_iter",
|
||||
"yoke",
|
||||
"zerofrom",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "icu_locale_core"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "25d29b57c17c6528d0dc4c2141cfa2d08200465acb537a3909c8b9b8a4a984c0"
|
||||
dependencies = [
|
||||
"displaydoc",
|
||||
"litemap",
|
||||
"tinystr",
|
||||
"writeable",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "icu_normalizer"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "404a3b8c486f205c87fe7f23ab6894d84a4ad76d6e4671ae1981ba7def87c599"
|
||||
dependencies = [
|
||||
"icu_collections",
|
||||
"icu_normalizer_data",
|
||||
"icu_properties",
|
||||
"icu_provider",
|
||||
"smallvec",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "icu_normalizer_data"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "20f0088bfb7441f7a37fa0c9e20cb1e9cba1cac6b9ce8cd2a6dc44294f8e8cbc"
|
||||
|
||||
[[package]]
|
||||
name = "icu_properties"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2ea83d991c56158f7cf898bf86436c1dfe7c07ee82fe88951f2695c9f37a379f"
|
||||
dependencies = [
|
||||
"displaydoc",
|
||||
"icu_collections",
|
||||
"icu_locale_core",
|
||||
"icu_properties_data",
|
||||
"icu_provider",
|
||||
"zerotrie",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "icu_properties_data"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b98f29af82d02b163fe50161388350e743bb3107cc8e8c2f7314cb505faddfcf"
|
||||
|
||||
[[package]]
|
||||
name = "icu_provider"
|
||||
version = "2.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "413099315636c588ee8f011190b329b971a09a8fae9d157e01dcae8379468514"
|
||||
dependencies = [
|
||||
"displaydoc",
|
||||
"icu_locale_core",
|
||||
"writeable",
|
||||
"yoke",
|
||||
"zerofrom",
|
||||
"zerotrie",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "idna"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de"
|
||||
dependencies = [
|
||||
"idna_adapter",
|
||||
"smallvec",
|
||||
"utf8_iter",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "idna_adapter"
|
||||
version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714"
|
||||
dependencies = [
|
||||
"icu_normalizer",
|
||||
"icu_properties",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "inout"
|
||||
version = "0.1.4"
|
||||
|
|
@ -491,6 +617,12 @@ version = "0.2.189"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
|
||||
|
||||
[[package]]
|
||||
name = "litemap"
|
||||
version = "0.8.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae"
|
||||
|
||||
[[package]]
|
||||
name = "log"
|
||||
version = "0.4.34"
|
||||
|
|
@ -554,6 +686,12 @@ version = "0.3.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
|
||||
|
||||
[[package]]
|
||||
name = "percent-encoding"
|
||||
version = "2.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-lite"
|
||||
version = "0.2.17"
|
||||
|
|
@ -582,6 +720,15 @@ dependencies = [
|
|||
"universal-hash",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "potential_utf"
|
||||
version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661"
|
||||
dependencies = [
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ppv-lite86"
|
||||
version = "0.2.21"
|
||||
|
|
@ -698,6 +845,20 @@ dependencies = [
|
|||
"syn 3.0.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ring"
|
||||
version = "0.17.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"cfg-if",
|
||||
"getrandom 0.2.17",
|
||||
"libc",
|
||||
"untrusted",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rmp"
|
||||
version = "0.8.15"
|
||||
|
|
@ -737,6 +898,41 @@ dependencies = [
|
|||
"semver",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustls"
|
||||
version = "0.23.46"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf"
|
||||
dependencies = [
|
||||
"log",
|
||||
"once_cell",
|
||||
"ring",
|
||||
"rustls-pki-types",
|
||||
"rustls-webpki",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustls-pki-types"
|
||||
version = "1.15.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
|
||||
dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustls-webpki"
|
||||
version = "0.103.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281"
|
||||
dependencies = [
|
||||
"ring",
|
||||
"rustls-pki-types",
|
||||
"untrusted",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustversion"
|
||||
version = "1.0.23"
|
||||
|
|
@ -881,6 +1077,12 @@ version = "0.4.12"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
|
||||
|
||||
[[package]]
|
||||
name = "smallvec"
|
||||
version = "1.16.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.5.10"
|
||||
|
|
@ -901,6 +1103,12 @@ dependencies = [
|
|||
"der",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "stable_deref_trait"
|
||||
version = "1.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
|
||||
|
||||
[[package]]
|
||||
name = "stacker"
|
||||
version = "0.1.25"
|
||||
|
|
@ -942,6 +1150,17 @@ dependencies = [
|
|||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "synstructure"
|
||||
version = "0.14.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror"
|
||||
version = "2.0.20"
|
||||
|
|
@ -962,6 +1181,16 @@ dependencies = [
|
|||
"syn 3.0.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tinystr"
|
||||
version = "0.8.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "abc840c238154d58f9e86b2792261edfde565a17696e1cda1a449de27979801a"
|
||||
dependencies = [
|
||||
"displaydoc",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "typenum"
|
||||
version = "1.20.1"
|
||||
|
|
@ -984,6 +1213,54 @@ dependencies = [
|
|||
"subtle",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "untrusted"
|
||||
version = "0.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
|
||||
|
||||
[[package]]
|
||||
name = "ureq"
|
||||
version = "2.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "02d1a66277ed75f640d608235660df48c8e3c19f3b4edb6a263315626cc3c01d"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"log",
|
||||
"once_cell",
|
||||
"rustls",
|
||||
"rustls-pki-types",
|
||||
"url",
|
||||
"webpki-roots 0.26.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "url"
|
||||
version = "2.5.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed"
|
||||
dependencies = [
|
||||
"form_urlencoded",
|
||||
"idna",
|
||||
"percent-encoding",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "utf8_iter"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "af3bd9d3e571c26fcce9d033b87ea5ab061b8d2ffe9b7a5675d810d68f03edf0"
|
||||
dependencies = [
|
||||
"utf_types",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "utf_types"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3a3b425286980a367278453f3b6d28cdcb134b574e5a35824a6ff5eebad50bb6"
|
||||
|
||||
[[package]]
|
||||
name = "uuid"
|
||||
version = "1.25.0"
|
||||
|
|
@ -1061,6 +1338,24 @@ dependencies = [
|
|||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "webpki-roots"
|
||||
version = "0.26.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
|
||||
dependencies = [
|
||||
"webpki-roots 1.0.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "webpki-roots"
|
||||
version = "1.0.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
|
||||
dependencies = [
|
||||
"rustls-pki-types",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "winapi"
|
||||
version = "0.3.9"
|
||||
|
|
@ -1239,6 +1534,35 @@ version = "0.57.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
|
||||
|
||||
[[package]]
|
||||
name = "writeable"
|
||||
version = "0.6.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc"
|
||||
|
||||
[[package]]
|
||||
name = "yoke"
|
||||
version = "0.8.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "417ef3fcf6c300b7dbd25c85531a746931c02ad536f3cf4786ed1e88223340b3"
|
||||
dependencies = [
|
||||
"stable_deref_trait",
|
||||
"yoke-derive",
|
||||
"zerofrom",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "yoke-derive"
|
||||
version = "0.8.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
"synstructure",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerocopy"
|
||||
version = "0.8.63"
|
||||
|
|
@ -1259,12 +1583,66 @@ dependencies = [
|
|||
"syn 2.0.119",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerofrom"
|
||||
version = "0.1.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272"
|
||||
dependencies = [
|
||||
"zerofrom-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerofrom-derive"
|
||||
version = "0.1.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
"synstructure",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zeroize"
|
||||
version = "1.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
|
||||
|
||||
[[package]]
|
||||
name = "zerotrie"
|
||||
version = "0.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f"
|
||||
dependencies = [
|
||||
"displaydoc",
|
||||
"yoke",
|
||||
"zerofrom",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerovec"
|
||||
version = "0.11.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f58a2799cfe9a1670c7ea8edbd52ef8edb93cb37a7f83bf31ba9e35d9dae6a40"
|
||||
dependencies = [
|
||||
"yoke",
|
||||
"zerofrom",
|
||||
"zerovec-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerovec-derive"
|
||||
version = "0.11.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c3743ddbf47738d929cd1ae2e199d40d0162979dfe31aef8ae646d2bf00bf97a"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zmij"
|
||||
version = "1.0.23"
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ path = "src/main.rs"
|
|||
|
||||
[dependencies]
|
||||
anyhow = "1"
|
||||
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"] }
|
||||
|
|
@ -21,6 +22,7 @@ serde = { version = "1", features = ["derive"] }
|
|||
serde_json = "1"
|
||||
sha2 = "0.10"
|
||||
thiserror = "2"
|
||||
ureq = { version = "2", default-features = false, features = ["tls"] }
|
||||
zeroize = "1"
|
||||
|
||||
[target.'cfg(unix)'.dependencies]
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ mod quota;
|
|||
mod stream;
|
||||
#[cfg(target_os = "linux")]
|
||||
mod unit;
|
||||
mod usage_poll;
|
||||
|
||||
pub use backend::{ClaudeCodeBackend, CredentialTokenSource, StderrSink, TokenSource};
|
||||
pub use config::{ClaudeCodeConfig, ClaudeCodeLimits};
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
//! The plan's own usage windows, as the CLI reports them. The subscription
|
||||
//! decides how much a plan may burn; the CLI's `rate_limit_event` is the only
|
||||
//! place that number appears, so the latest observation of each window is kept
|
||||
//! here and published as the usage readout and lent with each lease.
|
||||
//! The plan's own usage windows, as the subscription reports them. The
|
||||
//! subscription decides how much a plan may burn, and two sensors observe it:
|
||||
//! the CLI's `rate_limit_event` on the connector's own calls, and the usage
|
||||
//! poll of `usage_poll` while that stream is quiet. The latest observation of
|
||||
//! each window, from either, is kept here and published as the usage readout
|
||||
//! and lent with each lease.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Mutex;
|
||||
|
|
|
|||
729
crates/claude-connector/src/usage_poll.rs
Normal file
729
crates/claude-connector/src/usage_poll.rs
Normal file
|
|
@ -0,0 +1,729 @@
|
|||
//! The second sensor of the plan's usage windows. The CLI's `rate_limit_event`
|
||||
//! only arrives when the connector's own calls run, so while the swarm is
|
||||
//! quiet the weekly reading goes stale; this thread asks the subscription's
|
||||
//! usage endpoint when the `seven_day` reading is `STALE` or older, and
|
||||
//! records the answer in the same [`QuotaCell`] the stream feeds. Neither
|
||||
//! sensor owns the reading: the newest observation of a window wins. A failed
|
||||
//! poll never touches the cell; only the cell's own reset rule retires a
|
||||
//! window.
|
||||
//!
|
||||
//! The endpoint is undocumented (the shape is the one Claude Code 2.1.312
|
||||
//! parses), so every answer it can give is handled and none is trusted to
|
||||
//! change the schedule beyond the rules below.
|
||||
|
||||
use std::io::Read;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use model_connector::PlanQuota;
|
||||
use serde_json::Value;
|
||||
use zeroize::Zeroizing;
|
||||
|
||||
use crate::backend::TokenSource;
|
||||
use crate::quota::QuotaCell;
|
||||
|
||||
/// The usage endpoint.
|
||||
const ENDPOINT: &str = "https://api.anthropic.com/api/oauth/usage";
|
||||
/// A reading this old, or older, is stale; also the least gap between polls.
|
||||
const STALE: Duration = Duration::from_secs(15 * 60);
|
||||
/// The longest any backoff may wait.
|
||||
const CEILING: Duration = Duration::from_secs(6 * 60 * 60);
|
||||
const TIMEOUT: Duration = Duration::from_secs(10);
|
||||
/// A usage body is a few hundred bytes; this is a ceiling, not an expectation.
|
||||
const MAX_BODY: u64 = 1 << 20;
|
||||
/// The status every polled window carries.
|
||||
const POLLED: &str = "polled";
|
||||
|
||||
/// What the endpoint said, reduced to what the schedule needs. It never holds
|
||||
/// the token.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
enum Answer {
|
||||
Body(String),
|
||||
Status(u16, Option<Duration>),
|
||||
Unreachable,
|
||||
}
|
||||
|
||||
/// Why the poller is waiting; logged once per change.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum State {
|
||||
Polled,
|
||||
NoToken,
|
||||
RateLimited,
|
||||
Refused,
|
||||
NotFound,
|
||||
Unreadable,
|
||||
Unexpected,
|
||||
Unreachable,
|
||||
}
|
||||
|
||||
impl State {
|
||||
fn line(self) -> &'static str {
|
||||
match self {
|
||||
Self::Polled => "usage poll: reading recorded",
|
||||
Self::NoToken => "usage poll: no token to ask with",
|
||||
Self::RateLimited => "usage poll: rate limited, waiting",
|
||||
Self::Refused => {
|
||||
"usage poll: refused (401/403), the token likely lacks the user:profile scope; polling stopped until restart"
|
||||
}
|
||||
Self::NotFound => "usage poll: endpoint not found, backing off",
|
||||
Self::Unreadable => "usage poll: answer not readable, backing off",
|
||||
Self::Unexpected => "usage poll: unexpected status, retrying later",
|
||||
Self::Unreachable => "usage poll: endpoint unreachable, retrying later",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
enum Next {
|
||||
Wait(Duration),
|
||||
Stop,
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
struct Verdict {
|
||||
windows: Vec<PlanQuota>,
|
||||
next: Next,
|
||||
state: State,
|
||||
}
|
||||
|
||||
/// What one answer means: the windows to record (only a readable body has
|
||||
/// any), how long to wait, and the state to report.
|
||||
fn judge(answer: &Answer) -> Verdict {
|
||||
let (windows, next, state) = match answer {
|
||||
Answer::Body(body) => match parse_usage(body) {
|
||||
Some(windows) => (windows, Next::Wait(STALE), State::Polled),
|
||||
None => (Vec::new(), Next::Wait(CEILING), State::Unreadable),
|
||||
},
|
||||
Answer::Status(429, retry_after) => (
|
||||
Vec::new(),
|
||||
Next::Wait(retry_after.unwrap_or_default().clamp(STALE, CEILING)),
|
||||
State::RateLimited,
|
||||
),
|
||||
Answer::Status(401 | 403, _) => (Vec::new(), Next::Stop, State::Refused),
|
||||
Answer::Status(404, _) => (Vec::new(), Next::Wait(CEILING), State::NotFound),
|
||||
Answer::Status(..) => (Vec::new(), Next::Wait(STALE), State::Unexpected),
|
||||
Answer::Unreachable => (Vec::new(), Next::Wait(STALE), State::Unreachable),
|
||||
};
|
||||
Verdict {
|
||||
windows,
|
||||
next,
|
||||
state,
|
||||
}
|
||||
}
|
||||
|
||||
/// The plan windows a usage body reports: `five_hour` and `seven_day`, each
|
||||
/// `{utilization: percent 0-100, resets_at: ISO 8601}`. A window missing
|
||||
/// either field, or reporting nonsense, is skipped, and an unknown reading is
|
||||
/// never zero. Other keys are ignored. `None` when the body is not a JSON
|
||||
/// object at all.
|
||||
fn parse_usage(body: &str) -> Option<Vec<PlanQuota>> {
|
||||
let usage: Value = serde_json::from_str(body).ok()?;
|
||||
let usage = usage.as_object()?;
|
||||
Some(
|
||||
["five_hour", "seven_day"]
|
||||
.into_iter()
|
||||
.filter_map(|window| {
|
||||
let reading = usage.get(window)?;
|
||||
let percent = reading.get("utilization")?.as_f64()?;
|
||||
let resets_at = reading.get("resets_at")?.as_str()?;
|
||||
if !percent.is_finite() || percent < 0.0 {
|
||||
return None;
|
||||
}
|
||||
let resets_at = chrono::DateTime::parse_from_rfc3339(resets_at).ok()?;
|
||||
Some(PlanQuota {
|
||||
window: window.to_string(),
|
||||
// `as` saturates a float at the integer's bounds.
|
||||
utilization_permille: Some((percent * 10.0).round() as u32),
|
||||
resets_at_unix_ms: u64::try_from(resets_at.timestamp_millis()).ok()?,
|
||||
status: POLLED.to_string(),
|
||||
// Stamped by the cell that keeps it.
|
||||
observed_at_unix_ms: 0,
|
||||
})
|
||||
})
|
||||
.collect(),
|
||||
)
|
||||
}
|
||||
|
||||
/// The blocking HTTPS client: 10 s end to end, no redirects (the token must
|
||||
/// not follow one anywhere).
|
||||
struct UsageClient {
|
||||
agent: ureq::Agent,
|
||||
url: String,
|
||||
}
|
||||
|
||||
impl UsageClient {
|
||||
fn new(url: impl Into<String>) -> Self {
|
||||
Self {
|
||||
agent: ureq::AgentBuilder::new()
|
||||
.timeout(TIMEOUT)
|
||||
.redirects(0)
|
||||
.build(),
|
||||
url: url.into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn ask(&self, token: &Zeroizing<String>) -> Answer {
|
||||
let bearer = Zeroizing::new(format!("Bearer {}", token.as_str()));
|
||||
let sent = self
|
||||
.agent
|
||||
.get(&self.url)
|
||||
.set("Authorization", bearer.as_str())
|
||||
.set("anthropic-beta", "oauth-2025-04-20")
|
||||
.call();
|
||||
match sent {
|
||||
Ok(response) if response.status() == 200 => {
|
||||
let mut body = String::new();
|
||||
match response
|
||||
.into_reader()
|
||||
.take(MAX_BODY)
|
||||
.read_to_string(&mut body)
|
||||
{
|
||||
Ok(_) => Answer::Body(body),
|
||||
Err(_) => Answer::Unreachable,
|
||||
}
|
||||
}
|
||||
Ok(response) => Answer::Status(response.status(), None),
|
||||
Err(ureq::Error::Status(code, response)) => {
|
||||
// Whole seconds only; the HTTP-date form counts as absent.
|
||||
let retry_after = response
|
||||
.header("retry-after")
|
||||
.and_then(|seconds| seconds.trim().parse::<u64>().ok())
|
||||
.map(Duration::from_secs);
|
||||
Answer::Status(code, retry_after)
|
||||
}
|
||||
Err(ureq::Error::Transport(_)) => Answer::Unreachable,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The schedule and the thing it drives. Time comes in as an argument so the
|
||||
/// schedule is testable without waiting.
|
||||
struct Poller {
|
||||
cell: Arc<QuotaCell>,
|
||||
tokens: Box<dyn TokenSource>,
|
||||
client: UsageClient,
|
||||
log: Box<dyn FnMut(&str) + Send>,
|
||||
/// No poll before this instant, unix ms: the last answer's own wait.
|
||||
not_before_ms: u64,
|
||||
started: bool,
|
||||
stopped: bool,
|
||||
last_state: Option<State>,
|
||||
}
|
||||
|
||||
impl Poller {
|
||||
fn new(
|
||||
cell: Arc<QuotaCell>,
|
||||
tokens: Box<dyn TokenSource>,
|
||||
client: UsageClient,
|
||||
log: Box<dyn FnMut(&str) + Send>,
|
||||
) -> Self {
|
||||
Self {
|
||||
cell,
|
||||
tokens,
|
||||
client,
|
||||
log,
|
||||
not_before_ms: 0,
|
||||
started: false,
|
||||
stopped: false,
|
||||
last_state: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Polls if a poll is due at `now_ms`; returns how long to wait before
|
||||
/// asking again, or `None` once polling has stopped for good.
|
||||
///
|
||||
/// The first call polls. After that a poll is due only when the cell's
|
||||
/// `seven_day` observation is `STALE` old or older (or absent), and never
|
||||
/// before the last answer's own wait has passed.
|
||||
fn tick(&mut self, now_ms: u64) -> Option<Duration> {
|
||||
if self.stopped {
|
||||
return None;
|
||||
}
|
||||
let due_at = self.due_at_ms();
|
||||
if now_ms < due_at {
|
||||
return Some(Duration::from_millis(due_at - now_ms));
|
||||
}
|
||||
self.started = true;
|
||||
let verdict = match self.tokens.read() {
|
||||
Some(token) => judge(&self.client.ask(&token)),
|
||||
None => Verdict {
|
||||
windows: Vec::new(),
|
||||
next: Next::Wait(STALE),
|
||||
state: State::NoToken,
|
||||
},
|
||||
};
|
||||
self.cell.observe(verdict.windows);
|
||||
if self.last_state != Some(verdict.state) {
|
||||
(self.log)(verdict.state.line());
|
||||
self.last_state = Some(verdict.state);
|
||||
}
|
||||
match verdict.next {
|
||||
Next::Stop => {
|
||||
self.stopped = true;
|
||||
None
|
||||
}
|
||||
Next::Wait(wait) => {
|
||||
self.not_before_ms = now_ms.saturating_add(duration_ms(wait));
|
||||
Some(Duration::from_millis(
|
||||
self.due_at_ms().saturating_sub(now_ms),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The earliest instant the next poll may happen.
|
||||
fn due_at_ms(&self) -> u64 {
|
||||
if !self.started {
|
||||
return 0;
|
||||
}
|
||||
let stale_at = self
|
||||
.cell
|
||||
.snapshot()
|
||||
.windows
|
||||
.iter()
|
||||
.find(|quota| quota.window == "seven_day")
|
||||
.map_or(0, |quota| {
|
||||
quota.observed_at_unix_ms.saturating_add(duration_ms(STALE))
|
||||
});
|
||||
stale_at.max(self.not_before_ms)
|
||||
}
|
||||
|
||||
fn run(mut self) {
|
||||
while let Some(wait) = self.tick(unix_ms()) {
|
||||
std::thread::sleep(wait);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn duration_ms(duration: Duration) -> u64 {
|
||||
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
|
||||
}
|
||||
|
||||
fn unix_ms() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map_or(0, duration_ms)
|
||||
}
|
||||
|
||||
/// Starts the poll thread. It reads the token afresh at each poll, writes
|
||||
/// only to `cell`, and logs one line to stderr when its state changes.
|
||||
pub fn start(
|
||||
cell: Arc<QuotaCell>,
|
||||
tokens: Box<dyn TokenSource>,
|
||||
) -> std::io::Result<std::thread::JoinHandle<()>> {
|
||||
let poller = Poller::new(
|
||||
cell,
|
||||
tokens,
|
||||
UsageClient::new(ENDPOINT),
|
||||
Box::new(|line| eprintln!("claude-connector: {line}")),
|
||||
);
|
||||
std::thread::Builder::new()
|
||||
.name("claude-usage-poll".to_string())
|
||||
.spawn(move || poller.run())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::Write;
|
||||
use std::net::TcpListener;
|
||||
use std::sync::Mutex;
|
||||
|
||||
const MIN: u64 = 60_000;
|
||||
const CANARY: &str = "sk-ant-oat01-CANARY-must-never-surface";
|
||||
const FAR: &str = "2100-01-01T00:00:00+00:00";
|
||||
const FAR_MS: u64 = 4_102_444_800_000;
|
||||
|
||||
fn body(five: &str, seven: &str) -> String {
|
||||
format!(
|
||||
r#"{{"five_hour":{five},"seven_day":{seven},"seven_day_opus":null,"extra_usage":{{"x":1}}}}"#
|
||||
)
|
||||
}
|
||||
|
||||
fn reading(percent: f64) -> String {
|
||||
format!(r#"{{"utilization":{percent},"resets_at":"{FAR}"}}"#)
|
||||
}
|
||||
|
||||
fn secs(s: u64) -> Duration {
|
||||
Duration::from_secs(s)
|
||||
}
|
||||
|
||||
// ---- mapping ----
|
||||
|
||||
#[test]
|
||||
fn usage_body_maps_percent_to_permille() {
|
||||
let windows = parse_usage(&body(&reading(6.0), &reading(35.0))).unwrap();
|
||||
let expect = |window: &str, permille| PlanQuota {
|
||||
window: window.to_string(),
|
||||
utilization_permille: Some(permille),
|
||||
resets_at_unix_ms: FAR_MS,
|
||||
status: "polled".to_string(),
|
||||
observed_at_unix_ms: 0,
|
||||
};
|
||||
assert_eq!(
|
||||
windows,
|
||||
vec![expect("five_hour", 60), expect("seven_day", 350)]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn usage_percent_rounds_to_the_nearest_permille() {
|
||||
let windows = parse_usage(&body(&reading(12.34), &reading(0.06))).unwrap();
|
||||
assert_eq!(windows[0].utilization_permille, Some(123));
|
||||
assert_eq!(windows[1].utilization_permille, Some(1));
|
||||
let zero = parse_usage(&body(&reading(0.0), "null")).unwrap();
|
||||
assert_eq!(zero[0].utilization_permille, Some(0), "a real zero stays");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resets_at_is_parsed_with_its_offset_to_unix_ms() {
|
||||
let seven = r#"{"utilization":1.0,"resets_at":"2026-11-07T12:30:15.250+02:00"}"#;
|
||||
let windows = parse_usage(&body("null", seven)).unwrap();
|
||||
// 2026-11-07T10:30:15.250Z
|
||||
assert_eq!(windows[0].resets_at_unix_ms, 1_794_047_415_250);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_window_missing_a_field_or_reading_nonsense_is_skipped_not_zero() {
|
||||
let no_percent = format!(r#"{{"utilization":null,"resets_at":"{FAR}"}}"#);
|
||||
let no_reset = r#"{"utilization":50.0,"resets_at":null}"#;
|
||||
assert_eq!(parse_usage(&body(&no_percent, no_reset)).unwrap(), vec![]);
|
||||
let junk = r#"{"utilization":5.0,"resets_at":"soon"}"#;
|
||||
assert_eq!(parse_usage(&body(&reading(-1.0), junk)).unwrap(), vec![]);
|
||||
let before_epoch = r#"{"utilization":5.0,"resets_at":"1969-01-01T00:00:00Z"}"#;
|
||||
assert_eq!(parse_usage(&body(before_epoch, "null")).unwrap(), vec![]);
|
||||
// One good window survives its neighbour's null.
|
||||
let one = parse_usage(&body(&no_percent, &reading(35.0))).unwrap();
|
||||
assert_eq!(one.len(), 1);
|
||||
assert_eq!(one[0].window, "seven_day");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_body_that_is_not_an_object_is_unreadable() {
|
||||
for bad in ["", "not json", "[1,2]", "42", "null"] {
|
||||
assert_eq!(parse_usage(bad), None, "{bad:?}");
|
||||
}
|
||||
assert_eq!(parse_usage("{}"), Some(vec![]));
|
||||
}
|
||||
|
||||
// ---- judgement ----
|
||||
|
||||
fn wait_of(answer: &Answer) -> Next {
|
||||
judge(answer).next
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rate_limit_waits_the_longer_of_retry_after_and_the_floor_capped() {
|
||||
let limited = |after| wait_of(&Answer::Status(429, after));
|
||||
assert_eq!(limited(Some(secs(3600))), Next::Wait(secs(3600)));
|
||||
assert_eq!(limited(Some(secs(20 * 60))), Next::Wait(secs(20 * 60)));
|
||||
assert_eq!(limited(Some(secs(10))), Next::Wait(STALE));
|
||||
assert_eq!(limited(None), Next::Wait(STALE));
|
||||
assert_eq!(limited(Some(secs(100_000))), Next::Wait(CEILING));
|
||||
assert_eq!(limited(Some(secs(6 * 3600))), Next::Wait(CEILING));
|
||||
assert_eq!(judge(&Answer::Status(429, None)).state, State::RateLimited);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn refusal_stops_polling_and_the_rest_back_off() {
|
||||
assert_eq!(wait_of(&Answer::Status(401, None)), Next::Stop);
|
||||
assert_eq!(wait_of(&Answer::Status(403, Some(secs(5)))), Next::Stop);
|
||||
assert_eq!(judge(&Answer::Status(403, None)).state, State::Refused);
|
||||
assert_eq!(wait_of(&Answer::Status(404, None)), Next::Wait(CEILING));
|
||||
assert_eq!(
|
||||
wait_of(&Answer::Body("<html>".to_string())),
|
||||
Next::Wait(CEILING)
|
||||
);
|
||||
assert_eq!(wait_of(&Answer::Unreachable), Next::Wait(STALE));
|
||||
assert_eq!(wait_of(&Answer::Status(503, None)), Next::Wait(STALE));
|
||||
assert_eq!(
|
||||
wait_of(&Answer::Body(body(&reading(1.0), "null"))),
|
||||
Next::Wait(STALE)
|
||||
);
|
||||
}
|
||||
|
||||
// ---- a fake server, a poller, and the cell ----
|
||||
|
||||
/// Serves each canned reply to one connection in turn and remembers the
|
||||
/// requests.
|
||||
fn serve(replies: Vec<String>) -> (String, Arc<Mutex<Vec<String>>>) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let url = format!("http://{}/api/oauth/usage", listener.local_addr().unwrap());
|
||||
let seen = Arc::new(Mutex::new(Vec::new()));
|
||||
let record = seen.clone();
|
||||
std::thread::spawn(move || {
|
||||
for reply in replies {
|
||||
let Ok((mut stream, _)) = listener.accept() else {
|
||||
return;
|
||||
};
|
||||
let mut request = vec![0u8; 4096];
|
||||
let n = stream.read(&mut request).unwrap_or(0);
|
||||
record
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(String::from_utf8_lossy(&request[..n]).into_owned());
|
||||
let _ = stream.write_all(reply.as_bytes());
|
||||
}
|
||||
});
|
||||
(url, seen)
|
||||
}
|
||||
|
||||
fn http(status: &str, headers: &str, body: &str) -> String {
|
||||
format!(
|
||||
"HTTP/1.1 {status}\r\n{headers}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
|
||||
body.len()
|
||||
)
|
||||
}
|
||||
|
||||
struct Fixed(Option<&'static str>);
|
||||
impl TokenSource for Fixed {
|
||||
fn read(&self) -> Option<Zeroizing<String>> {
|
||||
self.0.map(|t| Zeroizing::new(t.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
type Lines = Arc<Mutex<Vec<String>>>;
|
||||
|
||||
fn poller(cell: &Arc<QuotaCell>, url: &str, token: Option<&'static str>) -> (Poller, Lines) {
|
||||
let lines = Arc::new(Mutex::new(Vec::new()));
|
||||
let sink = lines.clone();
|
||||
let poller = Poller::new(
|
||||
cell.clone(),
|
||||
Box::new(Fixed(token)),
|
||||
UsageClient::new(url),
|
||||
Box::new(move |line| sink.lock().unwrap().push(line.to_string())),
|
||||
);
|
||||
(poller, lines)
|
||||
}
|
||||
|
||||
fn window(cell: &QuotaCell, name: &str) -> Option<PlanQuota> {
|
||||
cell.snapshot()
|
||||
.windows
|
||||
.into_iter()
|
||||
.find(|q| q.window == name)
|
||||
}
|
||||
|
||||
fn stream_reading(window: &str, permille: u32) -> PlanQuota {
|
||||
PlanQuota {
|
||||
window: window.to_string(),
|
||||
utilization_permille: Some(permille),
|
||||
resets_at_unix_ms: FAR_MS,
|
||||
status: "allowed".to_string(),
|
||||
observed_at_unix_ms: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn requests(seen: &Arc<Mutex<Vec<String>>>) -> usize {
|
||||
seen.lock().unwrap().len()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_first_tick_polls_and_records_what_the_endpoint_said() {
|
||||
let (url, seen) = serve(vec![http(
|
||||
"200 OK",
|
||||
"Content-Type: application/json\r\n",
|
||||
&body(&reading(6.0), &reading(35.0)),
|
||||
)]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, lines) = poller(&cell, &url, Some(CANARY));
|
||||
let wait = poller.tick(unix_ms()).unwrap();
|
||||
let windows = cell.snapshot().windows;
|
||||
assert_eq!(windows.len(), 2);
|
||||
assert_eq!(windows[0].status, "polled");
|
||||
assert_eq!(windows[1].utilization_permille, Some(350));
|
||||
assert!(
|
||||
(STALE..STALE + secs(5)).contains(&wait),
|
||||
"a fresh reading waits until it is stale: {wait:?}"
|
||||
);
|
||||
let request = seen.lock().unwrap()[0].to_lowercase();
|
||||
assert!(request.contains(&format!("authorization: bearer {}", CANARY.to_lowercase())));
|
||||
assert!(request.contains("anthropic-beta: oauth-2025-04-20"));
|
||||
assert_eq!(lines.lock().unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_only_polls() {
|
||||
let (url, seen) = serve(vec![http("200 OK", "", &body("null", &reading(35.0))); 3]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, _) = poller(&cell, &url, Some(CANARY));
|
||||
poller.tick(unix_ms());
|
||||
assert_eq!(requests(&seen), 1, "the startup poll");
|
||||
let observed = window(&cell, "seven_day").unwrap().observed_at_unix_ms;
|
||||
|
||||
// 5 min old: no request, and the wait is the time left.
|
||||
let wait = poller.tick(observed + 5 * MIN).unwrap();
|
||||
assert_eq!(requests(&seen), 1);
|
||||
assert_eq!(wait, Duration::from_millis(10 * MIN));
|
||||
// One millisecond short of 15 min: still none.
|
||||
poller.tick(observed + 15 * MIN - 1);
|
||||
assert_eq!(requests(&seen), 1);
|
||||
// 16 min old: one request.
|
||||
poller.tick(observed + 16 * MIN);
|
||||
assert_eq!(requests(&seen), 2);
|
||||
// That poll's answer holds its own 15 min, whatever the stamp says.
|
||||
poller.tick(observed + 16 * MIN + 1);
|
||||
assert_eq!(requests(&seen), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_reading_is_stale_at_exactly_fifteen_minutes() {
|
||||
let (url, seen) = serve(vec![http("200 OK", "", &body("null", &reading(35.0))); 2]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, _) = poller(&cell, &url, Some(CANARY));
|
||||
poller.tick(unix_ms());
|
||||
let observed = window(&cell, "seven_day").unwrap().observed_at_unix_ms;
|
||||
poller.tick(observed + 15 * MIN);
|
||||
assert_eq!(requests(&seen), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_stream_observation_keeps_the_poll_quiet() {
|
||||
let (url, seen) = serve(vec![http("200 OK", "", &body("null", &reading(35.0)))]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, _) = poller(&cell, &url, Some(CANARY));
|
||||
poller.tick(unix_ms());
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
// The stream reports seven_day afresh, after the poll.
|
||||
cell.observe([stream_reading("seven_day", 400)]);
|
||||
let observed = window(&cell, "seven_day").unwrap().observed_at_unix_ms;
|
||||
let wait = poller.tick(observed + 14 * MIN).unwrap();
|
||||
assert_eq!(requests(&seen), 1);
|
||||
assert_eq!(wait, Duration::from_millis(MIN));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failures_never_touch_the_cell() {
|
||||
let limited = |headers: &str| http("429 Too Many Requests", headers, "{}");
|
||||
let cases = [
|
||||
(limited("Retry-After: 3600\r\n"), Some(60)),
|
||||
(limited(""), Some(15)),
|
||||
(
|
||||
limited("Retry-After: Wed, 21 Oct 2026 07:28:00 GMT\r\n"),
|
||||
Some(15),
|
||||
),
|
||||
(http("403 Forbidden", "", "{}"), None),
|
||||
(http("401 Unauthorized", "", "{}"), None),
|
||||
(http("404 Not Found", "", "{}"), Some(360)),
|
||||
(http("200 OK", "", "<html>not json</html>"), Some(360)),
|
||||
(
|
||||
http("302 Found", "Location: http://127.0.0.1:9/\r\n", ""),
|
||||
Some(15),
|
||||
),
|
||||
];
|
||||
for (reply, minutes) in cases {
|
||||
let (url, _) = serve(vec![reply.clone()]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
cell.observe([
|
||||
stream_reading("five_hour", 120),
|
||||
stream_reading("seven_day", 300),
|
||||
]);
|
||||
let before = cell.snapshot();
|
||||
let (mut poller, lines) = poller(&cell, &url, Some(CANARY));
|
||||
let wait = poller.tick(unix_ms());
|
||||
assert_eq!(cell.snapshot(), before, "{reply}");
|
||||
assert_eq!(wait.map(|w| w.as_secs() / 60), minutes, "{reply}");
|
||||
assert_eq!(lines.lock().unwrap().len(), 1, "{reply}");
|
||||
assert_eq!(poller.tick(unix_ms()).is_none(), minutes.is_none());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unreachable_endpoint_waits_the_floor_and_a_missing_token_never_asks() {
|
||||
// Nothing listens here: bind, learn the port, drop.
|
||||
let port = TcpListener::bind("127.0.0.1:0")
|
||||
.unwrap()
|
||||
.local_addr()
|
||||
.unwrap()
|
||||
.port();
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, lines) =
|
||||
poller(&cell, &format!("http://127.0.0.1:{port}/u"), Some(CANARY));
|
||||
assert_eq!(poller.tick(unix_ms()), Some(STALE));
|
||||
assert!(cell.snapshot().windows.is_empty());
|
||||
assert_eq!(lines.lock().unwrap().len(), 1);
|
||||
|
||||
let (url, seen) = serve(vec![http("200 OK", "", "{}")]);
|
||||
let (mut poller, lines) = poller(&cell, &url, None);
|
||||
assert_eq!(poller.tick(unix_ms()), Some(STALE));
|
||||
assert_eq!(requests(&seen), 0);
|
||||
assert_eq!(lines.lock().unwrap()[0], State::NoToken.line());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_state_is_logged_once_until_it_changes() {
|
||||
let limited = http("429 Too Many Requests", "", "{}");
|
||||
let (url, _) = serve(vec![
|
||||
limited.clone(),
|
||||
limited,
|
||||
http("200 OK", "", &body("null", &reading(35.0))),
|
||||
]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, lines) = poller(&cell, &url, Some(CANARY));
|
||||
let t = unix_ms();
|
||||
poller.tick(t);
|
||||
poller.tick(t + 16 * MIN);
|
||||
assert_eq!(lines.lock().unwrap().len(), 1, "same state, one line");
|
||||
poller.tick(t + 32 * MIN);
|
||||
let lines = lines.lock().unwrap();
|
||||
assert_eq!(lines.len(), 2);
|
||||
assert_eq!(lines[1], State::Polled.line());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nothing_logged_or_stored_holds_the_token() {
|
||||
let echo = format!(r#"{{"echo":"{CANARY}"}}"#);
|
||||
let (url, _) = serve(vec![
|
||||
http("429 Too Many Requests", "Retry-After: 60\r\n", &echo),
|
||||
http(
|
||||
"200 OK",
|
||||
"",
|
||||
&format!(r#"{{"five_hour":{},"note":"{CANARY}"}}"#, reading(1.0)),
|
||||
),
|
||||
http("403 Forbidden", "", CANARY),
|
||||
]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, lines) = poller(&cell, &url, Some(CANARY));
|
||||
let t = unix_ms();
|
||||
poller.tick(t);
|
||||
poller.tick(t + 16 * MIN);
|
||||
assert!(poller.tick(t + 32 * MIN).is_none());
|
||||
let lines = lines.lock().unwrap();
|
||||
assert_eq!(lines.len(), 3, "one line per change of state");
|
||||
let seen = format!("{lines:?} {:?}", cell.snapshot());
|
||||
assert!(!seen.contains("CANARY"), "{seen}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_and_poll_share_one_cell() {
|
||||
let (url, _) = serve(vec![http(
|
||||
"200 OK",
|
||||
"",
|
||||
&body(&reading(6.0), &reading(35.0)),
|
||||
)]);
|
||||
let cell = Arc::new(QuotaCell::new());
|
||||
let (mut poller, _) = poller(&cell, &url, Some(CANARY));
|
||||
poller.tick(unix_ms());
|
||||
assert_eq!(window(&cell, "five_hour").unwrap().status, "polled");
|
||||
// A later stream reading of one window replaces the poll's; the other
|
||||
// window keeps the poll's.
|
||||
cell.observe([stream_reading("five_hour", 90)]);
|
||||
let five = window(&cell, "five_hour").unwrap();
|
||||
assert_eq!(
|
||||
(five.status.as_str(), five.utilization_permille),
|
||||
("allowed", Some(90))
|
||||
);
|
||||
assert_eq!(window(&cell, "seven_day").unwrap().status, "polled");
|
||||
// And a later poll replaces the stream's.
|
||||
let (url, _) = serve(vec![http("200 OK", "", &body(&reading(7.0), "null"))]);
|
||||
let (mut again, _) = poller_at(&cell, &url);
|
||||
again.tick(unix_ms());
|
||||
assert_eq!(
|
||||
window(&cell, "five_hour").unwrap().utilization_permille,
|
||||
Some(70)
|
||||
);
|
||||
}
|
||||
|
||||
fn poller_at(cell: &Arc<QuotaCell>, url: &str) -> (Poller, Lines) {
|
||||
poller(cell, url, Some(CANARY))
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue