runner_manager_github/
traffic.rs

1// owner: c2-authenticated-client (traffic accounting)
2
3//! What this host actually spends against GitHub's rate limits, measured.
4//!
5//! The budget model in [`crate::rest`] *projects* a cost from a table of
6//! request counts. This module *measures* one: every response the shared
7//! [`crate::AuthenticatedClient`] receives is counted here by class, and the
8//! rate-limit headers GitHub attaches to it are kept.
9//!
10//! # Why the classes are full, not-modified, and failed
11//!
12//! GitHub's own words, from "Best practices for using the REST API"
13//! (<https://docs.github.com/en/rest/using-the-rest-api/best-practices-for-using-the-rest-api#use-conditional-requests-if-appropriate>):
14//!
15//! > Making a conditional request does not count against your primary rate
16//! > limit if a `304` response is returned and the request was made while
17//! > correctly authorized with an `Authorization` header.
18//!
19//! So a `304` is free against the hourly quota and a `200` is not. That is the
20//! whole reason [`crate::demand`] sends `If-None-Match`, and the reason this
21//! module counts the two apart: the share of `304`s is the number that says
22//! whether conditional polling is paying off. Errors are counted separately
23//! because they do count against the quota and are not data.
24//!
25//! # The headers describe the ACCOUNT, not this machine
26//!
27//! "Primary rate limits for GitHub App user access tokens … are dictated by the
28//! primary rate limits for the authenticated user. This rate limit is combined
29//! with any requests that another GitHub App or OAuth app makes on that user's
30//! behalf and any requests that the user makes with a personal access token"
31//! (<https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api#primary-rate-limit-for-github-app-installations>).
32//!
33//! Every host signed in as the same person therefore draws from **one** bucket,
34//! and so does that person's `gh` CLI. `x-ratelimit-remaining` on any response
35//! is the bucket's level, whoever drained it — which is exactly why it is the
36//! right input for slowing down: this machine cannot count the other machines'
37//! requests, but it can see what they left.
38//!
39//! # The secondary limit has no header until it is exceeded
40//!
41//! "No more than 900 points per minute are allowed for REST API endpoints", and
42//! a `GET` costs one point
43//! (<https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api#about-secondary-rate-limits>).
44//! GitHub does not say a `304` is exempt from that limit, so this module
45//! assumes it is not: every request is a point. The secondary headroom shown
46//! to an operator is therefore this host's busiest minute against
47//! [`SECONDARY_POINTS_PER_MINUTE`], stated as one host's share of a limit the
48//! other hosts on the account share too.
49
50use std::sync::Mutex;
51use std::time::Duration;
52
53use reqwest::StatusCode;
54use runner_manager_domain::model::Timestamp;
55use serde::{Deserialize, Serialize};
56
57use crate::HeaderMap;
58
59/// GitHub's documented REST secondary limit, in points per minute.
60///
61/// A `GET` is one point, so for this product — which only polls with `GET` —
62/// it is also a request count.
63pub const SECONDARY_POINTS_PER_MINUTE: u32 = 900;
64
65/// Below this share of the hourly quota left, polling stretches.
66///
67/// One fifth: 1,000 of 5,000. Chosen so that a host noticing it still has
68/// enough budget to keep polling at its *active* interval for the rest of the
69/// hour, while giving the fleet's interactive users (`gh`, other apps on the
70/// same account) room.
71pub const LOW_PRIMARY_REMAINING_PERCENT: u64 = 20;
72
73/// Below this share, polling stretches further.
74///
75/// One twentieth: 250 of 5,000. At that level a single busy repository's job
76/// listings can drain the rest of the bucket, and an exhausted bucket stops
77/// every host on the account, not just this one.
78pub const CRITICAL_PRIMARY_REMAINING_PERCENT: u64 = 5;
79
80const _: () = assert!(
81    CRITICAL_PRIMARY_REMAINING_PERCENT < LOW_PRIMARY_REMAINING_PERCENT,
82    "the critical threshold must sit below the low one"
83);
84
85/// How many minutes of history are kept.
86const WINDOW_MINUTES: usize = 60;
87
88/// What GitHub said about the account's hourly quota on one response.
89#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
90pub struct RateLimitSnapshot {
91    /// `x-ratelimit-limit`.
92    pub limit: u64,
93    /// `x-ratelimit-remaining`.
94    pub remaining: u64,
95    /// `x-ratelimit-used`, when sent.
96    pub used: Option<u64>,
97    /// `x-ratelimit-reset`, a Unix timestamp in seconds.
98    pub reset_unix_secs: u64,
99    /// When this host read it.
100    pub observed_at: Timestamp,
101}
102
103impl RateLimitSnapshot {
104    /// The `core` quota's state from a response's headers, or `None` when the
105    /// response did not carry all three of limit, remaining and reset, or
106    /// described a different resource (`search`, `graphql`, …), whose bucket
107    /// is not the one demand polling spends.
108    #[must_use]
109    pub fn from_headers(headers: &HeaderMap, now: Timestamp) -> Option<Self> {
110        let text = |name: &str| headers.get(name).and_then(|v| v.to_str().ok());
111        if text("x-ratelimit-resource").is_some_and(|r| !r.trim().eq_ignore_ascii_case("core")) {
112            return None;
113        }
114        let number = |name: &str| text(name).and_then(|v| v.trim().parse::<u64>().ok());
115        Some(Self {
116            limit: number("x-ratelimit-limit")?,
117            remaining: number("x-ratelimit-remaining")?,
118            used: number("x-ratelimit-used"),
119            reset_unix_secs: number("x-ratelimit-reset")?,
120            observed_at: now,
121        })
122    }
123
124    /// How long until the quota resets, zero once it has.
125    #[must_use]
126    pub fn reset_in(&self, now: Timestamp) -> Duration {
127        let reset = i64::try_from(self.reset_unix_secs).unwrap_or(i64::MAX);
128        u64::try_from(reset.saturating_sub(now.timestamp()))
129            .map_or(Duration::ZERO, Duration::from_secs)
130    }
131
132    /// How hard this reading says to slow down.
133    ///
134    /// A reading whose reset has passed says nothing about the current window,
135    /// so it is [`PrimaryPressure::Normal`] however low it was.
136    #[must_use]
137    pub fn pressure(&self, now: Timestamp) -> PrimaryPressure {
138        if self.limit == 0 || self.reset_in(now).is_zero() {
139            return PrimaryPressure::Normal;
140        }
141        let percent = self.remaining.saturating_mul(100) / self.limit;
142        if percent < CRITICAL_PRIMARY_REMAINING_PERCENT {
143            PrimaryPressure::Critical
144        } else if percent < LOW_PRIMARY_REMAINING_PERCENT {
145            PrimaryPressure::Low
146        } else {
147            PrimaryPressure::Normal
148        }
149    }
150}
151
152/// How much of the shared hourly quota is left, as a decision input.
153#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
154#[serde(rename_all = "snake_case")]
155pub enum PrimaryPressure {
156    /// At least [`LOW_PRIMARY_REMAINING_PERCENT`] left.
157    Normal,
158    /// Under [`LOW_PRIMARY_REMAINING_PERCENT`].
159    Low,
160    /// Under [`CRITICAL_PRIMARY_REMAINING_PERCENT`].
161    Critical,
162}
163
164impl PrimaryPressure {
165    /// How many times the active interval a poll should wait at this level.
166    #[must_use]
167    pub const fn stretch_factor(self) -> u32 {
168        match self {
169            Self::Normal => 1,
170            Self::Low => 2,
171            Self::Critical => 4,
172        }
173    }
174}
175
176#[derive(Debug, Clone, Copy, Default)]
177struct Minute {
178    /// Minutes since the Unix epoch this bucket counts, so a bucket left over
179    /// from an hour ago is recognised as stale rather than added in.
180    epoch_minute: i64,
181    full: u32,
182    not_modified: u32,
183    failed: u32,
184}
185
186impl Minute {
187    const fn total(&self) -> u32 {
188        self.full
189            .saturating_add(self.not_modified)
190            .saturating_add(self.failed)
191    }
192}
193
194#[derive(Debug)]
195struct Inner {
196    minutes: [Minute; WINDOW_MINUTES],
197    first_seen: Option<Timestamp>,
198    latest: Option<RateLimitSnapshot>,
199}
200
201impl Default for Inner {
202    fn default() -> Self {
203        Self {
204            minutes: [Minute::default(); WINDOW_MINUTES],
205            first_seen: None,
206            latest: None,
207        }
208    }
209}
210
211/// Every response one client received, by class and by minute, for the last
212/// hour, plus the newest rate-limit reading.
213#[derive(Debug, Default)]
214pub struct ApiTraffic {
215    inner: Mutex<Inner>,
216}
217
218impl ApiTraffic {
219    #[must_use]
220    pub fn new() -> Self {
221        Self::default()
222    }
223
224    /// Count one response.
225    pub fn record(&self, status: StatusCode, headers: &HeaderMap, now: Timestamp) {
226        // The quota is read only off answers made with the credential: a
227        // `401` carries the unauthenticated 60-an-hour bucket, which is not
228        // the one this host spends.
229        let snapshot = (status.is_success() || status == StatusCode::NOT_MODIFIED)
230            .then(|| RateLimitSnapshot::from_headers(headers, now))
231            .flatten();
232        let mut inner = self
233            .inner
234            .lock()
235            .unwrap_or_else(std::sync::PoisonError::into_inner);
236        inner.first_seen.get_or_insert(now);
237        let epoch_minute = now.timestamp().div_euclid(60);
238        let slot = usize::try_from(epoch_minute.rem_euclid(WINDOW_MINUTES as i64)).unwrap_or(0);
239        let bucket = &mut inner.minutes[slot];
240        if bucket.epoch_minute != epoch_minute {
241            *bucket = Minute {
242                epoch_minute,
243                ..Minute::default()
244            };
245        }
246        if status == StatusCode::NOT_MODIFIED {
247            bucket.not_modified = bucket.not_modified.saturating_add(1);
248        } else if status.is_success() {
249            bucket.full = bucket.full.saturating_add(1);
250        } else {
251            bucket.failed = bucket.failed.saturating_add(1);
252        }
253        if let Some(snapshot) = snapshot {
254            inner.latest = Some(snapshot);
255        }
256    }
257
258    /// The newest `core` rate-limit reading, from any response.
259    #[must_use]
260    pub fn latest_rate_limit(&self) -> Option<RateLimitSnapshot> {
261        self.inner
262            .lock()
263            .unwrap_or_else(std::sync::PoisonError::into_inner)
264            .latest
265    }
266
267    /// The last hour, summed.
268    #[must_use]
269    pub fn summary(&self, now: Timestamp) -> TrafficSummary {
270        let inner = self
271            .inner
272            .lock()
273            .unwrap_or_else(std::sync::PoisonError::into_inner);
274        let current = now.timestamp().div_euclid(60);
275        let mut summary = TrafficSummary {
276            observed_minutes: 0,
277            full: 0,
278            not_modified: 0,
279            failed: 0,
280            peak_requests_per_minute: 0,
281            rate_limit: inner.latest,
282        };
283        for minute in &inner.minutes {
284            // Strictly inside the last hour, the current minute included.
285            if minute.epoch_minute > current - WINDOW_MINUTES as i64
286                && minute.epoch_minute <= current
287            {
288                summary.full = summary.full.saturating_add(minute.full);
289                summary.not_modified = summary.not_modified.saturating_add(minute.not_modified);
290                summary.failed = summary.failed.saturating_add(minute.failed);
291                summary.peak_requests_per_minute =
292                    summary.peak_requests_per_minute.max(minute.total());
293            }
294        }
295        summary.observed_minutes = inner.first_seen.map_or(0, |first| {
296            let minutes = current - first.timestamp().div_euclid(60) + 1;
297            u32::try_from(minutes.clamp(0, WINDOW_MINUTES as i64)).unwrap_or(0)
298        });
299        summary
300    }
301}
302
303/// One host's measured traffic over (up to) the last hour.
304#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
305pub struct TrafficSummary {
306    /// How many minutes the counts below cover: the time since the first
307    /// response, capped at sixty. Under sixty, the counts are not yet an
308    /// hourly rate.
309    pub observed_minutes: u32,
310    /// `2xx` responses: data, and charged against the primary limit.
311    pub full: u32,
312    /// `304` responses: free against the primary limit.
313    pub not_modified: u32,
314    /// Everything else. Charged, and not data.
315    pub failed: u32,
316    /// The most requests of any one minute, every class counted: the number
317    /// to hold against [`SECONDARY_POINTS_PER_MINUTE`].
318    pub peak_requests_per_minute: u32,
319    /// The newest reading of the account's hourly quota.
320    pub rate_limit: Option<RateLimitSnapshot>,
321}
322
323impl TrafficSummary {
324    /// Every response counted.
325    #[must_use]
326    pub const fn total(&self) -> u32 {
327        self.full
328            .saturating_add(self.not_modified)
329            .saturating_add(self.failed)
330    }
331
332    /// The share of responses that were `304`, in whole percent, or `None`
333    /// before any response.
334    #[must_use]
335    pub fn not_modified_percent(&self) -> Option<u32> {
336        let total = self.total();
337        (total > 0).then(|| {
338            u32::try_from(u64::from(self.not_modified) * 100 / u64::from(total)).unwrap_or(100)
339        })
340    }
341
342    /// `count` over the observed minutes, scaled to an hour.
343    #[must_use]
344    pub fn per_hour(&self, count: u32) -> u32 {
345        if self.observed_minutes == 0 {
346            return 0;
347        }
348        u32::try_from(u64::from(count) * 60 / u64::from(self.observed_minutes)).unwrap_or(u32::MAX)
349    }
350}
351
352/// `requests_per_minute` as a share of [`SECONDARY_POINTS_PER_MINUTE`], in
353/// whole percent: a `GET` is one point.
354#[must_use]
355pub const fn secondary_share_percent(requests_per_minute: u32) -> u32 {
356    requests_per_minute.saturating_mul(100) / SECONDARY_POINTS_PER_MINUTE
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362    use chrono::TimeZone;
363    use reqwest::header::HeaderValue;
364
365    fn at(secs: i64) -> Timestamp {
366        chrono::Utc
367            .timestamp_opt(secs, 0)
368            .single()
369            .expect("a valid instant")
370    }
371
372    fn headers(pairs: &[(&'static str, &str)]) -> HeaderMap {
373        let mut map = HeaderMap::new();
374        for (name, value) in pairs {
375            map.insert(*name, HeaderValue::from_str(value).expect("a header value"));
376        }
377        map
378    }
379
380    fn quota(remaining: u64, reset: u64) -> HeaderMap {
381        headers(&[
382            ("x-ratelimit-limit", "5000"),
383            ("x-ratelimit-remaining", &remaining.to_string()),
384            ("x-ratelimit-used", &(5000 - remaining).to_string()),
385            ("x-ratelimit-reset", &reset.to_string()),
386            ("x-ratelimit-resource", "core"),
387        ])
388    }
389
390    #[test]
391    fn a_304_is_counted_apart_from_a_200_and_from_a_failure() {
392        let traffic = ApiTraffic::new();
393        let now = at(1_000_000);
394        traffic.record(StatusCode::OK, &HeaderMap::new(), now);
395        traffic.record(StatusCode::NOT_MODIFIED, &HeaderMap::new(), now);
396        traffic.record(StatusCode::NOT_MODIFIED, &HeaderMap::new(), now);
397        traffic.record(StatusCode::NOT_FOUND, &HeaderMap::new(), now);
398        let summary = traffic.summary(now);
399        assert_eq!(
400            (summary.full, summary.not_modified, summary.failed),
401            (1, 2, 1)
402        );
403        assert_eq!(
404            summary.full + summary.failed,
405            2,
406            "a 304 is not charged; a 404 is"
407        );
408        assert_eq!(summary.not_modified_percent(), Some(50));
409        assert_eq!(summary.peak_requests_per_minute, 4);
410    }
411
412    #[test]
413    fn counts_older_than_an_hour_fall_out_of_the_window() {
414        let traffic = ApiTraffic::new();
415        traffic.record(StatusCode::OK, &HeaderMap::new(), at(0));
416        // Same ring slot, one hour later: the old count must not be added in.
417        traffic.record(StatusCode::NOT_MODIFIED, &HeaderMap::new(), at(3_600));
418        let summary = traffic.summary(at(3_600));
419        assert_eq!((summary.full, summary.not_modified), (0, 1));
420        assert_eq!(summary.observed_minutes, 60);
421        assert_eq!(traffic.summary(at(3_600 + 3_600)).total(), 0);
422    }
423
424    #[test]
425    fn per_hour_scales_a_short_observation_to_an_hourly_rate() {
426        let traffic = ApiTraffic::new();
427        for second in 0..10 {
428            traffic.record(StatusCode::OK, &HeaderMap::new(), at(60 * second));
429        }
430        let summary = traffic.summary(at(60 * 9));
431        assert_eq!(summary.observed_minutes, 10);
432        assert_eq!(summary.per_hour(summary.full), 60);
433    }
434
435    #[test]
436    fn the_newest_core_reading_is_kept_and_another_resource_is_ignored() {
437        let traffic = ApiTraffic::new();
438        traffic.record(StatusCode::OK, &quota(4_000, 5_000), at(10));
439        let mut search = quota(1, 5_000);
440        search.insert("x-ratelimit-resource", HeaderValue::from_static("search"));
441        traffic.record(StatusCode::OK, &search, at(11));
442        let latest = traffic.latest_rate_limit().expect("a reading");
443        assert_eq!(latest.remaining, 4_000);
444        assert_eq!(latest.used, Some(1_000));
445    }
446
447    #[test]
448    fn the_quota_is_read_only_off_answers_made_with_the_credential() {
449        let traffic = ApiTraffic::new();
450        traffic.record(StatusCode::OK, &quota(4_000, 5_000), at(10));
451        let mut anonymous = quota(59, 5_000);
452        anonymous.insert("x-ratelimit-limit", HeaderValue::from_static("60"));
453        traffic.record(StatusCode::UNAUTHORIZED, &anonymous, at(11));
454        assert_eq!(
455            traffic.latest_rate_limit().map(|quota| quota.remaining),
456            Some(4_000),
457            "a 401 carries the unauthenticated bucket, not this host's"
458        );
459        traffic.record(StatusCode::NOT_MODIFIED, &quota(3_999, 5_000), at(12));
460        assert_eq!(
461            traffic.latest_rate_limit().map(|quota| quota.remaining),
462            Some(3_999)
463        );
464    }
465
466    #[test]
467    fn pressure_follows_the_documented_thresholds_and_expires_at_reset() {
468        let now = at(1_000);
469        let reading = |remaining| {
470            RateLimitSnapshot::from_headers(&quota(remaining, 4_600), now).expect("a reading")
471        };
472        assert_eq!(reading(1_000).pressure(now), PrimaryPressure::Normal);
473        assert_eq!(reading(999).pressure(now), PrimaryPressure::Low);
474        assert_eq!(reading(250).pressure(now), PrimaryPressure::Low);
475        assert_eq!(reading(249).pressure(now), PrimaryPressure::Critical);
476        assert_eq!(reading(0).reset_in(now), Duration::from_secs(3_600));
477        assert_eq!(
478            reading(0).pressure(at(4_600)),
479            PrimaryPressure::Normal,
480            "a reading from a window that has reset says nothing about the new one"
481        );
482    }
483
484    #[test]
485    fn the_secondary_share_is_the_busiest_minute_against_nine_hundred() {
486        let traffic = ApiTraffic::new();
487        for _ in 0..90 {
488            traffic.record(StatusCode::NOT_MODIFIED, &HeaderMap::new(), at(120));
489        }
490        traffic.record(StatusCode::OK, &HeaderMap::new(), at(180));
491        assert_eq!(
492            secondary_share_percent(traffic.summary(at(180)).peak_requests_per_minute),
493            10
494        );
495    }
496}
runner_manager_github/traffic.rs

runner_manager_github/
traffic.rs
