Skip to main content

runner_manager_github/
rest.rs

1// owner: c3-rest-inventory-gateway
2
3//! This gateway is deliberately client-secret-free, as D3 requires.
4//!
5//! Every read model the dashboard and the CLI display, over `api.github.com`:
6//! the runner inventory, the in-progress workflow count, and the runner-package
7//! download metadata — plus the two behaviours that make those numbers
8//! trustworthy rather than merely present.
9//!
10//! Everything here is built on [`crate::AuthenticatedClient`]. There is no
11//! second authentication path in this module and none may be added; the one
12//! credential is obtained by [`crate::device_flow`] and applied by that client.
13//!
14//! # The three read models
15//!
16//! | Operation | Endpoint | Type |
17//! |---|---|---|
18//! | [`InventoryGateway::list_runners`] | `/repos/{o}/{r}/actions/runners`, `/orgs/{org}/actions/runners` | [`RunnerInventory`] |
19//! | [`InventoryGateway::in_progress_activity`] | `/repos/{o}/{r}/actions/runs?status=in_progress` | [`ActivityCount`] |
20//! | [`InventoryGateway::runner_downloads`] | `…/actions/runners/downloads` | [`RunnerDownloads`] |
21//!
22//! **The in-progress workflow count and the busy-runner count are different
23//! numbers with different meanings**, and this module keeps them in different
24//! types on purpose. A workflow run is work GitHub has accepted; a busy runner
25//! is a machine this product can see executing something. `g2` renders them as
26//! separate aggregates, and collapsing them here would make that impossible to
27//! do correctly downstream.
28//!
29//! # Pagination is mandatory
30//!
31//! `04-subsystem-contracts.md`: "Pagination is mandatory; the dashboard must not
32//! treat a first page as a complete inventory." A target with more runners than
33//! one page is the ordinary case for an organization, and a silently truncated
34//! list reads as "no runners" rather than as an error — the failure is invisible
35//! at exactly the moment it matters.
36//!
37//! Every collection here therefore follows `Link: rel="next"` through
38//! [`crate::ApiResponse::next_page`], which is `c2`'s single reader of that
39//! header rather than a second one written here. Following the same reader is
40//! the point: it already handles a `rel="next"` that is not first, quoted and
41//! unquoted parameter forms, and — the case that silently stopped pagination at
42//! page one until a review caught it — a next-page URL that itself contains a
43//! comma, which a runner query carries routinely as `labels=self-hosted,windows`.
44//!
45//! Two facts travel with a collection so that a caller can tell a complete
46//! answer from an incomplete one: [`RunnerInventory::reported_total`], which is
47//! GitHub's own `total_count`, and [`RunnerInventory::truncated`], which is set
48//! when the [`crate::MAX_PAGES`] ceiling stopped the walk.
49//!
50//! # Rate limiting is a policy, and it lives here
51//!
52//! `c2` deliberately implemented none of it — it stopped *discarding* the
53//! evidence and handed it across the seam through [`GithubError::headers`],
54//! [`GithubError::retry_after`] and [`GithubError::rate_limit`]. This module is
55//! where the evidence becomes a decision, and the decision has three parts:
56//!
57//! 1. **`retry-after` is obeyed by not sending anything.** A detected limit
58//!    latches a window ([`RestInventory::rate_limit_backoff`]) during which this
59//!    gateway opens no socket at all and answers
60//!    [`InventoryError::RateLimited`] immediately. Obeying a back-off by
61//!    *sleeping inside a request* would be the same wait, spent invisibly, with
62//!    the caller's cancellation and refresh scheduling both bypassed.
63//! 2. **It is surfaced, never hidden** (`04-subsystem-contracts.md`, "Rate
64//!    limiting increases the refresh delay and is displayed, never hidden").
65//!    [`RateLimited`] is a displayable state carrying what GitHub said, and
66//!    [`RefreshState::retry_delay`] is the **absolute floor** on when `e1` may
67//!    try again — `next_attempt_at = now + retry_delay`, not the ordinary
68//!    interval plus that. Adding it would only wait longer than necessary: this
69//!    gateway already enforces the window itself, at no request cost.
70//! 3. **A rate limit is never confused with a permissions answer.** See
71//!    [`RateLimited::detect`]: GitHub sends `x-ratelimit-*` on *every* response,
72//!    so "remaining is zero" alone would turn an ordinary `404` into a rate
73//!    limit.
74//!
75//! # The shared request budget (the D4 consequence)
76//!
77//! Under scale sets, demand arrived over a long poll carried by the Actions
78//! service, which did not touch the `api.github.com` budget. After D4 it does,
79//! and that makes one number a product constraint rather than an implementation
80//! detail: demand, runner inventory and in-progress counts all draw on **one**
81//! ceiling of [`HOURLY_REQUEST_CEILING`] requests per hour.
82//!
83//! The projection lives here because this is the layer that sees every request.
84//! See [`TargetCost`] and [`BudgetProjection`] — and in particular
85//! [`TargetCost::organization`], because an organization target's cost scales
86//! with the number of repositories the App is installed on there. Projecting an
87//! organization as a flat per-target constant understates its real cost by
88//! exactly that factor, which is the one error this model exists to prevent.
89
90use std::{
91    collections::{BTreeMap, BTreeSet},
92    fmt,
93    future::Future,
94    sync::{
95        Arc,
96        atomic::{AtomicU64, Ordering},
97    },
98    time::Duration,
99};
100
101use runner_manager_domain::model::{
102    Arch, Clock, Org, Os, OwnerRepo, RefreshInterval, ScaleTarget, TargetScope, Timestamp,
103};
104use serde::Deserialize;
105use serde::de::DeserializeOwned;
106use tokio::sync::watch;
107
108use crate::{ApiRequest, ApiResponse, AuthenticatedClient, GithubError, MAX_PAGES};
109
110// ---------------------------------------------------------------------------
111// Constants
112// ---------------------------------------------------------------------------
113
114/// Items per page asked for on every paginated call.
115///
116/// GitHub's maximum. Asking for fewer multiplies the request count against a
117/// budget this module also has to project, which is the one place in this
118/// product where a lazy default is directly a product constraint.
119pub const PER_PAGE: u32 = 100;
120
121/// The documented hourly REST ceiling for a user-to-server token, measured
122/// 2026-08-21 (`04-subsystem-contracts.md`).
123pub const HOURLY_REQUEST_CEILING: u32 = 5_000;
124
125/// The fraction of [`HOURLY_REQUEST_CEILING`] a host may plan to spend.
126///
127/// `04-subsystem-contracts.md`: `add` "refuses a configuration that would exceed
128/// **half** of it". Half rather than all, because the projection covers only the
129/// agent's steady-state polling: an interactive `auth status`, a `repo add`
130/// validation, a JIT registration and a runner deletion all draw on the same
131/// ceiling and none of them is periodic enough to model.
132pub const BUDGET_SHARE_DIVISOR: u32 = 2;
133
134/// Seconds in the hour the ceiling is measured over.
135pub const SECONDS_PER_HOUR: u32 = 3_600;
136
137/// Requests one runner-inventory refresh costs, per target.
138///
139/// One, at either scope: a repository and an organization each have a single
140/// runners endpoint. A target whose inventory spans pages costs more than this
141/// in practice, and that is stated rather than modelled — see
142/// [`TargetCost::requests_per_refresh`].
143pub const RUNNER_INVENTORY_REQUESTS_PER_REFRESH: u32 = 1;
144
145/// Requests one in-progress workflow count costs, **per repository**.
146///
147/// Workflow runs are a per-repository resource. There is no organization-wide
148/// workflow-runs endpoint, so an organization pays this once per repository the
149/// App is installed on.
150///
151/// # This is the best case, not the worst one
152///
153/// One request is what a repository costs **when GitHub sends `total_count`**,
154/// which is the ordinary answer from the workflow-runs endpoint and the reason
155/// the figure is `1`. When it is absent the count falls back to walking pages,
156/// and that walk may spend up to [`MAX_ACTIVITY_FALLBACK_PAGES`] — so the true
157/// worst case per repository per refresh is **four**, not one.
158///
159/// The gap is stated rather than modelled, deliberately, and the same way
160/// [`RUNNER_INVENTORY_REQUESTS_PER_REFRESH`] states that a paginated inventory
161/// costs more than the one request it claims. But it is worth naming here
162/// because things are built on top of it: `f1`'s `host show` headroom and `f2`'s
163/// `add` refusals both read *this* constant, so both are projecting the
164/// best case. A target sitting at the edge of what `f2` will allow could
165/// overrun by up to 4x on repositories whose counts take the fallback.
166///
167/// [`BUDGET_SHARE_DIVISOR`] is what absorbs this: the projection is compared
168/// against half the ceiling precisely so that the half this model does not
169/// attempt to count has somewhere to go. The fallback is also bounded and
170/// **says when it was reached** — a repository that walked to the ceiling lands
171/// in [`ActivityCount::truncated`] — so an overrun is visible rather than
172/// silent. That visibility, not the number `1`, is what makes the projection
173/// honest.
174pub const ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH: u32 = 1;
175
176/// The most pages one repository's in-progress count may walk when GitHub sends
177/// no `total_count`.
178///
179/// [`crate::MAX_PAGES`] is the wrong ceiling for this walk, and the distinction
180/// is a budget one rather than a stylistic one. `MAX_PAGES` exists to stop a
181/// `Link: rel="next"` cycle looping *forever*; it is not a number anything
182/// budgeted for. This walk is charged against
183/// [`ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH`], which is **one**, and that
184/// constant is what [`TargetCost`] projects and what `f2` computes its `add`
185/// refusals from.
186///
187/// The arithmetic is why the two cannot share a ceiling. At the 60-second
188/// default a target refreshes 60 times an hour, so the projection budgets 60
189/// requests for one repository's activity count. A fallback allowed to reach
190/// `MAX_PAGES` could spend 6,000 — more than the whole
191/// [`HOURLY_REQUEST_CEILING`], for a single repository's count — which would
192/// make every refusal `f2` computes from the projection a fiction.
193///
194/// Four pages counts 400 in-progress runs exactly, at a worst case of 240
195/// requests/hour against the ~2,500 [`BUDGET_SHARE_DIVISOR`] leaves as slack.
196/// Past that the answer stops being exact and **says so**: the repository lands
197/// in [`ActivityCount::truncated`]. That is what makes a bounded walk honest
198/// rather than merely cheap — an unbounded walk and a silently-clipped one are
199/// both wrong, in opposite directions.
200pub const MAX_ACTIVITY_FALLBACK_PAGES: usize = 4;
201
202// Enforced at compile time rather than by a test, because the two ceilings
203// collapsing back into one is the defect, not a symptom of it: a budget that is
204// not tighter than the runaway ceiling is not a budget.
205const _: () = assert!(
206    MAX_ACTIVITY_FALLBACK_PAGES < MAX_PAGES,
207    "the activity page budget must stay below the runaway `Link`-cycle ceiling"
208);
209
210/// How far GitHub's `total_count` may exceed the single page it arrived with
211/// before the disagreement stops being a race and starts being evidence.
212///
213/// The tripwire in `RestInventory::repository_in_progress` has two very
214/// different customers, and they are separated by *size* rather than by
215/// existence:
216///
217/// * The **benign race** — a run finishing between GitHub computing
218///   `total_count` and serialising the page — makes `total` exceed `listed` by a
219///   handful. It is documented, it is real, and a debug build pointed at live
220///   GitHub during `c4`'s development is the build most likely to meet it.
221/// * The **defect being hunted** — `total_count` carrying the *unfiltered*
222///   lifetime total instead of the filtered one — is gross: thousands over a
223///   page of three, which is exactly the shape
224///   `a_total_count_that_disagrees_with_its_only_page_is_caught` pins at 5,000
225///   over 3.
226///
227/// So the always-on `warn!` fires on any disagreement at all, and the
228/// `debug_assert!` fires only past this threshold. An assert that fired on both
229/// would panic a development build over a race its own documentation calls
230/// legitimate — and a tripwire that cries wolf is a tripwire the next reader
231/// deletes, which costs the real check.
232///
233/// Only the *upward* gap is gated. `total` coming in **below** `listed` means a
234/// run started after the total was computed, which is the same race in the other
235/// direction and never the unfiltered-total defect, so it stays a `warn!` alone.
236const MAX_BENIGN_TOTAL_COUNT_SKEW: u64 = 16;
237
238// A zero skew is `total == listed` again, which is the check that panicked on
239// the documented race. Enforced at compile time rather than by a test because a
240// test cannot do it: any test that derives its fixture from this constant moves
241// with it and stays green at zero, which is exactly the false comfort this
242// assertion exists to refuse.
243const _: () = assert!(
244    MAX_BENIGN_TOTAL_COUNT_SKEW > 0,
245    "a zero skew re-creates the assert that trips on a run finishing mid-serialisation"
246);
247
248/// Requests one demand poll costs, **per repository**: the queued runs, then
249/// their jobs.
250///
251/// `c4` owns demand and reports its real per-poll count; this constant is the
252/// steady-state figure `04-subsystem-contracts.md` tabulates (~120 requests per
253/// hour at the 60-second default, which is two per refresh).
254pub const DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH: u32 = 2;
255
256/// How long a detected rate limit backs off for when GitHub gives no usable
257/// `retry-after` and no `x-ratelimit-reset`.
258///
259/// Sixty seconds is GitHub's own documented floor for its secondary rate limits,
260/// and the same value [`crate::DEFAULT_LOCKOUT_BACKOFF`] uses for the
261/// authentication lockout.
262pub const DEFAULT_RATE_LIMIT_BACKOFF: Duration = Duration::from_secs(60);
263
264/// The longest a rate limit may silence this gateway, whatever GitHub asked for.
265///
266/// The reasoning is [`crate::MAX_LOCKOUT_BACKOFF`]'s, and so is the consequence:
267/// because a still-limited response simply re-latches, this ceiling is a
268/// *polling interval* and not a deadline. A primary limit resets at most an hour
269/// out, so a fifteen-minute clamp costs at most three extra probe requests
270/// across that hour — against the alternative of letting a single header take
271/// the dashboard down for the rest of the hour.
272pub const MAX_RATE_LIMIT_BACKOFF: Duration = Duration::from_secs(15 * 60);
273
274// ---------------------------------------------------------------------------
275// Cancellation
276// ---------------------------------------------------------------------------
277
278/// A latch a caller flips to stop in-flight gateway work.
279///
280/// `04-subsystem-contracts.md` requires the gateway to support cancellation, and
281/// the requirement has teeth precisely because of pagination: an organization
282/// inventory is a *sequence* of requests, and a refresh the operator has already
283/// navigated away from should not keep spending the shared budget on pages
284/// nobody will read.
285///
286/// So cancellation is checked in two places, and both matter. Before each
287/// request — which is what stops a multi-page walk between pages — and
288/// concurrently with the request in flight, which is what stops a walk that is
289/// blocked on a socket.
290///
291/// Cloning shares the latch. Cancelling is one-way: a token that has been
292/// cancelled stays cancelled, because "cancel, then reuse" is how a caller ends
293/// up with a token whose state depends on a race.
294#[derive(Debug, Clone, Default)]
295pub struct CancelToken {
296    inner: Arc<CancelInner>,
297}
298
299#[derive(Debug)]
300struct CancelInner {
301    tx: watch::Sender<bool>,
302}
303
304impl Default for CancelInner {
305    fn default() -> Self {
306        Self {
307            tx: watch::Sender::new(false),
308        }
309    }
310}
311
312impl CancelToken {
313    /// A token nothing has cancelled yet.
314    #[must_use]
315    pub fn new() -> Self {
316        Self::default()
317    }
318
319    /// Cancel every operation holding this token, now and in the future.
320    pub fn cancel(&self) {
321        // `send_replace` rather than `send`: `send` reports an error when there
322        // are no receivers, and "nobody is waiting yet" is not a failure to
323        // cancel. The state is what callers read, and it is set either way.
324        self.inner.tx.send_replace(true);
325    }
326
327    #[must_use]
328    pub fn is_cancelled(&self) -> bool {
329        *self.inner.tx.borrow()
330    }
331
332    /// `Err(`[`InventoryError::Cancelled`]`)` once cancelled, so a call site can
333    /// bail with `?`.
334    ///
335    /// # Errors
336    /// [`InventoryError::Cancelled`].
337    pub fn check(&self) -> Result<(), InventoryError> {
338        if self.is_cancelled() {
339            return Err(InventoryError::Cancelled);
340        }
341        Ok(())
342    }
343
344    /// Resolves when this token is cancelled, and never otherwise.
345    pub async fn cancelled(&self) {
346        let mut rx = self.inner.tx.subscribe();
347        // `subscribe` snapshots the current version, so a `cancel` racing this
348        // line is still observed by `wait_for` — that is the property `Notify`
349        // does not have, and the reason this is a `watch` channel.
350        //
351        // The error arm is unreachable: the sender lives in the same `Arc` as
352        // this receiver, so it cannot be dropped while `self` is alive. It is
353        // written as a `pending` rather than a `return` because returning would
354        // report a cancellation that never happened.
355        if rx.wait_for(|cancelled| *cancelled).await.is_err() {
356            std::future::pending::<()>().await;
357        }
358    }
359
360    /// Run `work`, abandoning it if this token is cancelled first.
361    ///
362    /// # Errors
363    /// [`InventoryError::Cancelled`], or whatever `work` fails with.
364    pub async fn run<T>(
365        &self,
366        work: impl Future<Output = Result<T, InventoryError>>,
367    ) -> Result<T, InventoryError> {
368        tokio::select! {
369            // Biased so that an already-cancelled token loses no time to a
370            // request that was going to be abandoned anyway.
371            biased;
372            () = self.cancelled() => Err(InventoryError::Cancelled),
373            result = work => result,
374        }
375    }
376}
377
378// ---------------------------------------------------------------------------
379// Rate limiting
380// ---------------------------------------------------------------------------
381
382/// Which of GitHub's two rate limits a response was attributed to.
383#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
384pub enum RateLimitKind {
385    /// The hourly quota: `x-ratelimit-remaining: 0`. Resets at
386    /// `x-ratelimit-reset`.
387    Primary,
388    /// A short-term abuse limit: `429`, or a `403` whose message says so. Sends
389    /// `retry-after`.
390    Secondary,
391}
392
393impl fmt::Display for RateLimitKind {
394    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
395        f.write_str(match self {
396            Self::Primary => "primary",
397            Self::Secondary => "secondary",
398        })
399    }
400}
401
402/// An exhausted rate limit, as a state something can display.
403///
404/// The Definition of Done asks for "a distinct, displayable state rather than an
405/// opaque error", and the distinction is the point: a rate limit is the one
406/// failure in this gateway that is neither the operator's fault nor a reason to
407/// change anything. It resolves by waiting, and the operator's only legitimate
408/// question is "how long", which is what every field here answers.
409#[derive(Debug, Clone, Copy, PartialEq, Eq)]
410pub struct RateLimited {
411    pub kind: RateLimitKind,
412    /// `retry-after`, when GitHub sent one in the integer-seconds form.
413    pub retry_after: Option<Duration>,
414    /// `x-ratelimit-remaining`.
415    pub remaining: Option<u64>,
416    /// `x-ratelimit-reset`, a Unix timestamp in seconds.
417    pub reset_unix_secs: Option<u64>,
418}
419
420impl RateLimited {
421    /// Whether this failure is GitHub declining to serve any more requests for
422    /// now — and if so, which limit.
423    ///
424    /// # Why this is narrower than "remaining is zero"
425    ///
426    /// GitHub attaches `x-ratelimit-*` to **every** response, successful ones
427    /// included. A `404` that happens to arrive on the request that exhausted
428    /// the hourly quota therefore carries `x-ratelimit-remaining: 0` while
429    /// having nothing to do with rate limiting — and reporting it as a rate
430    /// limit would tell the operator to wait for a repository name that will
431    /// never resolve.
432    ///
433    /// So the status has to be one GitHub actually rate-limits with — `403` or
434    /// `429` — before the headers are read at all. And a `403` is not enough on
435    /// its own, because `403` is *also* how GitHub refuses a missing permission.
436    /// The two questions are therefore answered in order:
437    ///
438    /// **Is this a rate limit at all?** Only a `429`, or a `403` whose message
439    /// says "rate limit", qualifies. The message is the same evidence
440    /// [`crate::AuthenticatedClient`] already uses to keep a rate limit from
441    /// being misreported as an authentication lockout, and reading it the same
442    /// way here is what keeps the two layers agreeing. Everything else is a
443    /// permissions answer, **whatever the headers say** — see
444    /// `a_permissions_403_that_lands_on_an_exhausted_quota_is_still_forbidden`.
445    ///
446    /// **Which limit is it?** Now the headers matter.
447    /// `x-ratelimit-remaining: 0` is the primary limit and takes precedence,
448    /// because a `429` sent while the hourly quota is exhausted resets on the
449    /// hourly schedule rather than on a short back-off. Anything else is the
450    /// secondary limit.
451    #[must_use]
452    pub fn detect(error: &GithubError) -> Option<Self> {
453        let (status, message) = match error {
454            GithubError::Status {
455                status, message, ..
456            } => (*status, message.as_deref()),
457            GithubError::Forbidden { message, .. } => (403, message.as_deref()),
458            _ => return None,
459        };
460        if !matches!(status, 403 | 429) {
461            return None;
462        }
463
464        let evidence = error.rate_limit();
465        let remaining = evidence.and_then(|e| e.remaining);
466        let says_rate_limit =
467            message.is_some_and(|m| m.to_ascii_lowercase().contains("rate limit"));
468
469        // *Whether* this is a rate limit is decided before *which* one, and the
470        // headers get no say in the first question. A `403` is GitHub's answer
471        // to a missing grant as well as to an abuse limit, so within `403` the
472        // message is the only evidence that separates them — and
473        // `x-ratelimit-remaining: 0` rides on the permissions refusal too, when
474        // the denial happens to land on the request that exhausted the quota.
475        //
476        // Reading `remaining == 0` first classified that denial as a primary
477        // limit: an operator told to wait out a grant that will never arrive,
478        // and a latched window silencing every other target meanwhile. That is
479        // the same harm the `404` case above avoids, one status code over.
480        //
481        // This is not the inverse of `AuthenticatedClient::is_rate_limited`,
482        // which checks `remaining` first as well. It uses the answer only to
483        // *avoid* misreporting a limit as an authentication lockout — safe in
484        // that direction, because a false "not a lockout" costs nothing. Here
485        // the same test would positively assert a rate limit, which it cannot
486        // support.
487        //
488        // # The residual this trade leaves, recorded rather than fixed
489        //
490        // Making the message the *sole* discriminator within `403` has a cost,
491        // and it is one-directional: a genuine exhausted-quota `403` whose body
492        // did not parse into a `message` — an empty body, an intermediary's own
493        // error page, a shape GitHub has not sent before — is reported as
494        // `Forbidden`. No window is latched, and the client goes on spending
495        // against a quota that is already dead until the hour rolls over.
496        //
497        // There is no header-only discriminator to fall back on:
498        // `x-ratelimit-remaining: 0` rides on the permissions refusal too, which
499        // is exactly what
500        // `a_permissions_403_that_lands_on_an_exhausted_quota_is_still_forbidden`
501        // demonstrates. So the trade is forced — the only choice is which way to
502        // be wrong when the message is missing. Being wrong toward "permissions"
503        // costs this one target its requests for the rest of the hour. Being
504        // wrong toward "rate limit" latches a window that silences *every*
505        // target, waiting out a grant that will never arrive. The narrower harm
506        // is chosen deliberately.
507        if status != 429 && !says_rate_limit {
508            return None;
509        }
510        let kind = if remaining == Some(0) {
511            // Takes precedence over a `429`: a secondary limit hit while the
512            // hourly quota is also gone resets on the hourly schedule.
513            RateLimitKind::Primary
514        } else {
515            RateLimitKind::Secondary
516        };
517
518        Some(Self {
519            kind,
520            retry_after: error.retry_after(),
521            remaining,
522            reset_unix_secs: evidence.and_then(|e| e.reset_unix_secs),
523        })
524    }
525
526    /// How long to wait before asking again, given the current instant.
527    ///
528    /// `retry-after` first, because it is GitHub's explicit instruction;
529    /// `x-ratelimit-reset` second, because a primary limit says when rather than
530    /// how long; [`DEFAULT_RATE_LIMIT_BACKOFF`] when neither is usable, because
531    /// "GitHub said stop and named no time" must still stop.
532    ///
533    /// Clamped to [`MAX_RATE_LIMIT_BACKOFF`]. A remote header is not allowed to
534    /// decide how long this product stays dark.
535    #[must_use]
536    pub fn delay_from(&self, now: Timestamp) -> Duration {
537        let requested = self.retry_after.or_else(|| {
538            let reset = self.reset_unix_secs?;
539            let seconds = i64::try_from(reset).ok()? - now.timestamp();
540            u64::try_from(seconds).ok().map(Duration::from_secs)
541        });
542        // A zero or absent delay still has to be a wait: `reset` already in the
543        // past means the clock disagrees with GitHub, and answering "wait zero
544        // seconds" would turn a rate limit into a busy loop against the very
545        // endpoint that asked for quiet.
546        let requested = match requested {
547            Some(d) if d > Duration::ZERO => d,
548            _ => DEFAULT_RATE_LIMIT_BACKOFF,
549        };
550        requested.min(MAX_RATE_LIMIT_BACKOFF)
551    }
552}
553
554impl fmt::Display for RateLimited {
555    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
556        write!(f, "GitHub's {} rate limit is exhausted", self.kind)?;
557        if let Some(retry_after) = self.retry_after {
558            write!(
559                f,
560                "; it asked to be left alone for {}s",
561                retry_after.as_secs()
562            )?;
563        }
564        if let Some(remaining) = self.remaining {
565            write!(f, "; {remaining} requests remain in the hourly quota")?;
566        }
567        f.write_str(". Refreshes are delayed, not lost")
568    }
569}
570
571/// What GitHub last said about this credential's hourly quota, read from a
572/// response that **succeeded**.
573///
574/// Rate limiting must be "displayed, never hidden", and a state that only
575/// appears once the quota is already gone is not a display of it. These are the
576/// numbers `f1`'s `host show` and `g3`'s settings screen render alongside the
577/// projected budget, so an operator can compare what the projection expected
578/// against what the account is actually spending.
579#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
580pub struct RateLimitHeadroom {
581    /// `x-ratelimit-limit`.
582    pub limit: Option<u64>,
583    /// `x-ratelimit-remaining`.
584    pub remaining: Option<u64>,
585    /// `x-ratelimit-reset`, a Unix timestamp in seconds.
586    pub reset_unix_secs: Option<u64>,
587}
588
589impl RateLimitHeadroom {
590    fn from_response(response: &ApiResponse) -> Option<Self> {
591        let read = |name: &str| {
592            response
593                .header(name)
594                .and_then(|v| v.trim().parse::<u64>().ok())
595        };
596        let headroom = Self {
597            limit: read("x-ratelimit-limit"),
598            remaining: read("x-ratelimit-remaining"),
599            reset_unix_secs: read("x-ratelimit-reset"),
600        };
601        if headroom == Self::default() {
602            return None;
603        }
604        Some(headroom)
605    }
606}
607
608// ---------------------------------------------------------------------------
609// Errors and the displayable refresh state
610// ---------------------------------------------------------------------------
611
612/// Everything an inventory read can fail with.
613///
614/// [`GithubError`] is carried through rather than flattened, because `c2`'s
615/// taxonomy already separates the three outcomes `f1` branches on — a rejected
616/// credential, an authentication lockout, and a permissions refusal — and
617/// re-deciding that here would give the product two answers to the same
618/// question. The two variants added in front of it are the ones `c2` explicitly
619/// left to this layer.
620#[derive(Debug, thiserror::Error)]
621pub enum InventoryError {
622    /// GitHub is refusing further requests for now. Resolves by waiting.
623    #[error("{0}")]
624    RateLimited(RateLimited),
625
626    /// The caller withdrew the request. Nothing is known about the target.
627    #[error("the refresh was cancelled before it completed")]
628    Cancelled,
629
630    #[error(transparent)]
631    Github(#[from] GithubError),
632}
633
634impl InventoryError {
635    #[must_use]
636    pub fn is_rate_limited(&self) -> bool {
637        matches!(self, Self::RateLimited(_))
638    }
639
640    #[must_use]
641    pub fn is_cancelled(&self) -> bool {
642        matches!(self, Self::Cancelled)
643    }
644
645    /// The rate limit behind this failure, when there is one.
646    #[must_use]
647    pub fn rate_limited(&self) -> Option<&RateLimited> {
648        match self {
649            Self::RateLimited(limit) => Some(limit),
650            _ => None,
651        }
652    }
653
654    /// `true` when GitHub could not be reached at all, as opposed to answering
655    /// something unwelcome. `e1`'s offline handling turns on this distinction:
656    /// an outage retains running runners, while a rejection does not.
657    #[must_use]
658    pub fn is_offline(&self) -> bool {
659        matches!(self, Self::Github(GithubError::Transport(_)))
660    }
661}
662
663/// One refresh's outcome, as a value that can be stored, compared and rendered.
664///
665/// [`InventoryError`] cannot be any of those things — it owns a
666/// `reqwest::Error` and a `serde_json::Error`, neither of which is `Clone` —
667/// and the TUI needs a state it can hold in a snapshot and diff against the
668/// previous frame. So the error is *summarised* into this enum exactly once, at
669/// the gateway boundary, rather than each screen inventing its own summary.
670///
671/// `g2`'s Definition of Done names "loading, empty, unauthorized, rate-limited,
672/// and offline states"; four of those are variants here, and "loading" and
673/// "empty" are the caller's (no state yet, and a `Ready` snapshot with nothing
674/// in it).
675#[derive(Debug, Clone, PartialEq, Eq)]
676pub enum RefreshState {
677    /// The refresh completed. Note that an *empty* snapshot is still `Ready`:
678    /// "this target has no runners" is an answer, and rendering it as a failure
679    /// is how an idle host looks broken.
680    Ready(Box<InventorySnapshot>),
681    /// GitHub is rate limiting this credential.
682    RateLimited(RateLimited),
683    /// The stored credential was rejected. Terminal until `auth login`.
684    Unauthorized,
685    /// GitHub's temporary authentication lockout. The credential is fine.
686    LockedOut { retry_after: Duration },
687    /// A permissions answer. Re-authenticating will not change it.
688    Forbidden { message: Option<String> },
689    /// GitHub could not be reached.
690    Offline,
691    /// Anything else GitHub answered.
692    Failed {
693        status: Option<u16>,
694        message: String,
695    },
696    /// The caller withdrew the refresh.
697    Cancelled,
698}
699
700impl RefreshState {
701    /// Summarise a completed refresh.
702    #[must_use]
703    pub fn from_result(result: Result<InventorySnapshot, InventoryError>) -> Self {
704        match result {
705            Ok(snapshot) => Self::Ready(Box::new(snapshot)),
706            Err(error) => Self::from_error(&error),
707        }
708    }
709
710    /// Summarise a failure without consuming it.
711    #[must_use]
712    pub fn from_error(error: &InventoryError) -> Self {
713        match error {
714            InventoryError::RateLimited(limit) => Self::RateLimited(*limit),
715            InventoryError::Cancelled => Self::Cancelled,
716            InventoryError::Github(github) => match github {
717                GithubError::AuthenticationFailed => Self::Unauthorized,
718                GithubError::AuthenticationLockout { retry_after } => Self::LockedOut {
719                    retry_after: *retry_after,
720                },
721                GithubError::Forbidden { message, .. } => Self::Forbidden {
722                    message: message.clone(),
723                },
724                GithubError::Transport(_) => Self::Offline,
725                GithubError::Status { status, .. } => Self::Failed {
726                    status: Some(*status),
727                    message: github.to_string(),
728                },
729                other => Self::Failed {
730                    status: None,
731                    message: other.to_string(),
732                },
733            },
734        }
735    }
736
737    #[must_use]
738    pub fn is_ready(&self) -> bool {
739        matches!(self, Self::Ready(_))
740    }
741
742    #[must_use]
743    pub fn snapshot(&self) -> Option<&InventorySnapshot> {
744        match self {
745            Self::Ready(snapshot) => Some(&**snapshot),
746            _ => None,
747        }
748    }
749
750    /// How long to wait before trying again, or `None` when waiting is not what
751    /// this state needs.
752    ///
753    /// `04-subsystem-contracts.md`: "Rate limiting increases the refresh delay
754    /// and is displayed, never hidden." This is the increase. It is deliberately
755    /// `None` for [`RefreshState::Unauthorized`] and
756    /// [`RefreshState::Forbidden`], which no amount of waiting fixes.
757    ///
758    /// # An absolute floor, not an addend
759    ///
760    /// The scheduling rule is `next_attempt_at = now + retry_delay`. It is *not*
761    /// the ordinary refresh interval **plus** this — adding the two would double
762    /// the wait for no benefit, because the gateway enforces the same window
763    /// itself: a request issued inside it is answered
764    /// [`InventoryError::RateLimited`] without a socket being opened
765    /// ([`RestInventory::rate_limit_backoff`]).
766    ///
767    /// Trying too early is therefore cheap and self-correcting rather than
768    /// harmful. A suppressed request returns before [`RateLimited::detect`] is
769    /// ever reached, so an early attempt cannot ratchet the window outward by
770    /// this gateway's own silence, and repeated attempts converge on the instant
771    /// GitHub named. What an addend buys is one wasted interval per limit; what
772    /// it costs is a dashboard that stays dark longer than GitHub asked for.
773    ///
774    /// Note also that a [`RefreshState::RateLimited`] read back from
775    /// [`RestInventory::rate_limit_state`] carries the time *remaining*, not the
776    /// delay GitHub originally asked for — so this value shrinks as the window
777    /// elapses, which is another reason it reads as a deadline rather than as an
778    /// increment.
779    #[must_use]
780    pub fn retry_delay(&self, now: Timestamp) -> Option<Duration> {
781        match self {
782            Self::RateLimited(limit) => Some(limit.delay_from(now)),
783            Self::LockedOut { retry_after } => Some(*retry_after),
784            _ => None,
785        }
786    }
787}
788
789impl fmt::Display for RefreshState {
790    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
791        match self {
792            Self::Ready(snapshot) => write!(
793                f,
794                "{} runners, {} in progress",
795                snapshot.runners.len(),
796                snapshot.activity.total()
797            ),
798            Self::RateLimited(limit) => write!(f, "{limit}"),
799            Self::Unauthorized => f.write_str(
800                "GitHub rejected the stored credential; run `runner-manager auth login`",
801            ),
802            Self::LockedOut { retry_after } => write!(
803                f,
804                "GitHub has temporarily locked out authentication; retrying in {}s. \
805                 The credential itself is not the problem",
806                retry_after.as_secs()
807            ),
808            Self::Forbidden { message } => match message {
809                Some(message) => write!(f, "GitHub denied the request: {message}"),
810                None => f.write_str("GitHub denied the request"),
811            },
812            Self::Offline => f.write_str("GitHub is unreachable"),
813            Self::Failed { message, .. } => f.write_str(message),
814            Self::Cancelled => f.write_str("the refresh was cancelled"),
815        }
816    }
817}
818
819// ---------------------------------------------------------------------------
820// Runners
821// ---------------------------------------------------------------------------
822
823/// A runner's connection state, as GitHub reports it.
824#[derive(Debug, Clone, PartialEq, Eq, Hash)]
825pub enum RunnerStatus {
826    Online,
827    Offline,
828    /// Anything else GitHub sends. Kept verbatim rather than mapped onto one of
829    /// the two known values: a status this product does not recognise is
830    /// something to display, not something to guess at.
831    Other(String),
832}
833
834impl RunnerStatus {
835    #[must_use]
836    pub fn from_wire(raw: &str) -> Self {
837        match raw.trim().to_ascii_lowercase().as_str() {
838            "online" => Self::Online,
839            "offline" => Self::Offline,
840            _ => Self::Other(raw.trim().to_string()),
841        }
842    }
843
844    #[must_use]
845    pub fn is_online(&self) -> bool {
846        matches!(self, Self::Online)
847    }
848}
849
850impl fmt::Display for RunnerStatus {
851    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
852        match self {
853            Self::Online => f.write_str("online"),
854            Self::Offline => f.write_str("offline"),
855            Self::Other(raw) => f.write_str(raw),
856        }
857    }
858}
859
860/// One self-hosted runner GitHub knows about, local or not.
861///
862/// `07-security.md` and `g2` both require that runners this product did *not*
863/// create still appear — a legacy persistent runner is part of the operator's
864/// real inventory, and hiding it would make the dashboard a worse answer than
865/// GitHub's own page. Nothing here filters by ownership; deciding what is
866/// locally owned is `e1`'s, from the routing label.
867///
868/// # Labels arrive lower-cased
869///
870/// The D18 spike registered `Windows` and `X64` and read back `windows` and
871/// `x64` (`docs/spikes/d18-org-jit-verification.md`, point 3). It also
872/// established that **no label is added implicitly** — a runner carries exactly
873/// what was requested, with no `self-hosted`, no OS and no architecture unless
874/// they were asked for. So [`Runner::labels`] is what GitHub stores, verbatim,
875/// and [`Runner::has_label`] compares case-insensitively rather than pretending
876/// the case survived.
877#[derive(Debug, Clone, PartialEq, Eq)]
878pub struct Runner {
879    pub id: u64,
880    pub name: String,
881    /// GitHub's own OS string, unparsed. [`Runner::parsed_os`] is the lenient
882    /// reading; this is the fact.
883    pub os: String,
884    pub status: RunnerStatus,
885    pub busy: bool,
886    /// `None` when GitHub did not send the field.
887    ///
888    /// Absent is not `false`. A runner whose ephemerality is unknown is exactly
889    /// the runner an operator most wants flagged, and defaulting it to "not
890    /// ephemeral" would render that as a settled fact.
891    pub ephemeral: Option<bool>,
892    pub labels: Vec<String>,
893}
894
895impl Runner {
896    /// Whether this runner carries `label`, compared case-insensitively because
897    /// GitHub lower-cases what it stores.
898    #[must_use]
899    pub fn has_label(&self, label: &str) -> bool {
900        self.labels
901            .iter()
902            .any(|held| held.eq_ignore_ascii_case(label.trim()))
903    }
904
905    /// This runner's OS as a domain value, when it is one of the three the
906    /// product supports.
907    ///
908    /// `None` rather than an error: an unrecognised OS is a runner to display,
909    /// not a refresh to fail.
910    #[must_use]
911    pub fn parsed_os(&self) -> Option<Os> {
912        self.os.parse().ok()
913    }
914}
915
916/// Every runner GitHub reports for one target, across every page.
917#[derive(Debug, Clone, PartialEq, Eq)]
918pub struct RunnerInventory {
919    target: ScaleTarget,
920    runners: Vec<Runner>,
921    reported_total: Option<u64>,
922    pages: usize,
923    truncated: bool,
924}
925
926impl RunnerInventory {
927    /// A complete inventory read in one page. The constructor test doubles and
928    /// callers use; the gateway builds them through [`RunnerInventory::paged`].
929    #[must_use]
930    pub fn new(target: ScaleTarget, runners: Vec<Runner>) -> Self {
931        let reported_total = Some(u64::try_from(runners.len()).unwrap_or(u64::MAX));
932        Self {
933            target,
934            runners,
935            reported_total,
936            pages: 1,
937            truncated: false,
938        }
939    }
940
941    /// An inventory that took `pages` requests to read.
942    #[must_use]
943    pub fn paged(
944        target: ScaleTarget,
945        runners: Vec<Runner>,
946        reported_total: Option<u64>,
947        pages: usize,
948        truncated: bool,
949    ) -> Self {
950        Self {
951            target,
952            runners,
953            reported_total,
954            pages,
955            truncated,
956        }
957    }
958
959    #[must_use]
960    pub fn target(&self) -> &ScaleTarget {
961        &self.target
962    }
963
964    #[must_use]
965    pub fn runners(&self) -> &[Runner] {
966        &self.runners
967    }
968
969    #[must_use]
970    pub fn len(&self) -> usize {
971        self.runners.len()
972    }
973
974    #[must_use]
975    pub fn is_empty(&self) -> bool {
976        self.runners.is_empty()
977    }
978
979    /// Runners GitHub reports as executing a job.
980    ///
981    /// **This is not the in-progress workflow count.** See
982    /// [`ActivityCount::total`]; the two are different aggregates and `g2`
983    /// renders them separately.
984    #[must_use]
985    pub fn busy_count(&self) -> usize {
986        self.runners.iter().filter(|runner| runner.busy).count()
987    }
988
989    #[must_use]
990    pub fn online_count(&self) -> usize {
991        self.runners
992            .iter()
993            .filter(|runner| runner.status.is_online())
994            .count()
995    }
996
997    /// GitHub's own `total_count`, when it sent one.
998    #[must_use]
999    pub fn reported_total(&self) -> Option<u64> {
1000        self.reported_total
1001    }
1002
1003    /// How many requests reading this inventory took.
1004    #[must_use]
1005    pub fn pages(&self) -> usize {
1006        self.pages
1007    }
1008
1009    /// `true` when the [`MAX_PAGES`] ceiling stopped the walk, so this is a
1010    /// prefix of the inventory rather than the inventory.
1011    #[must_use]
1012    pub fn truncated(&self) -> bool {
1013        self.truncated
1014    }
1015
1016    /// How many runners GitHub said exist that this walk did not collect.
1017    ///
1018    /// The whole reason pagination is mandatory, made checkable: a caller that
1019    /// wants to refuse to render an incomplete inventory can, and one that
1020    /// renders it anyway can say so.
1021    /// `checked_sub` rather than a `>` test and a subtraction, because
1022    /// collecting *more* than GitHub reported is reachable: a `rel="next"` that
1023    /// points back at the page it arrived on is answered by the [`MAX_PAGES`]
1024    /// ceiling, and by then the same page has been collected a hundred times
1025    /// against a `total_count` of one. The eager `then_some` this replaced
1026    /// panicked with a subtraction overflow on exactly that path — in a debug
1027    /// build, from inside the agent's reconciliation loop.
1028    #[must_use]
1029    pub fn missing(&self) -> Option<u64> {
1030        let total = self.reported_total?;
1031        let collected = u64::try_from(self.runners.len()).unwrap_or(u64::MAX);
1032        total.checked_sub(collected).filter(|missing| *missing > 0)
1033    }
1034}
1035
1036// ---------------------------------------------------------------------------
1037// In-progress workflow activity
1038// ---------------------------------------------------------------------------
1039
1040/// Which repositories one activity count covers.
1041///
1042/// An in-progress workflow count is a **per-repository** number, because
1043/// workflow runs are a per-repository resource and GitHub publishes no
1044/// organization-wide runs endpoint. A repository target is therefore one
1045/// request; an organization target is one request per repository the App is
1046/// installed on there.
1047///
1048/// That asymmetry is why this type exists rather than a bare [`ScaleTarget`].
1049/// The repository list has to come from the caller — `f1` and `e1` already hold
1050/// it, from [`crate::AuthenticatedClient::discover_installations`] — and
1051/// re-discovering it on every refresh would cost more requests than the count
1052/// itself. Carrying it explicitly also means [`TargetCost::from_activity_scope`]
1053/// can project the real cost instead of a flat per-target constant.
1054#[derive(Debug, Clone, PartialEq, Eq)]
1055pub struct ActivityScope {
1056    target: ScaleTarget,
1057    repositories: Vec<OwnerRepo>,
1058}
1059
1060impl ActivityScope {
1061    /// A repository target: it counts its own runs and nothing else.
1062    #[must_use]
1063    pub fn repository(repo: OwnerRepo) -> Self {
1064        Self {
1065            target: ScaleTarget::Repository(repo.clone()),
1066            repositories: vec![repo],
1067        }
1068    }
1069
1070    /// An organization target, aggregating across the repositories the App is
1071    /// installed on.
1072    ///
1073    /// An empty list is legal and means exactly what it says: the App reaches no
1074    /// repository in this organization, so the aggregate is zero and costs
1075    /// nothing. It is not silently treated as "one".
1076    #[must_use]
1077    pub fn organization(org: Org, repositories: impl IntoIterator<Item = OwnerRepo>) -> Self {
1078        Self {
1079            target: ScaleTarget::Organization(org),
1080            repositories: repositories.into_iter().collect(),
1081        }
1082    }
1083
1084    #[must_use]
1085    pub fn target(&self) -> &ScaleTarget {
1086        &self.target
1087    }
1088
1089    #[must_use]
1090    pub fn repositories(&self) -> &[OwnerRepo] {
1091        &self.repositories
1092    }
1093
1094    /// Requests one in-progress count over this scope costs.
1095    #[must_use]
1096    pub fn requests_per_refresh(&self) -> u32 {
1097        u32::try_from(self.repositories.len()).unwrap_or(u32::MAX)
1098            * ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH
1099    }
1100}
1101
1102/// In-progress workflow runs, per repository and in total.
1103///
1104/// **Not the busy-runner count.** A workflow run is work GitHub has accepted and
1105/// started; a busy runner is a machine executing a job. One run can occupy
1106/// several runners, a run can be in progress with none of its jobs assigned yet,
1107/// and a busy runner may be executing a job for a workflow this product does not
1108/// poll at all. `04-subsystem-contracts.md` and `g2` both require them rendered
1109/// as distinct aggregates, and they are distinct types here so that they cannot
1110/// be added together by accident.
1111/// # A count can be short in two different ways, and both have to say so
1112///
1113/// A repository can fail to answer at all ([`ActivityCount::unavailable`]), and
1114/// a repository can answer with a number that is only a **floor**
1115/// ([`ActivityCount::truncated`]) — the fallback walk stopped at
1116/// [`MAX_ACTIVITY_FALLBACK_PAGES`], or GitHub's own total was wider than the
1117/// `u32` this product renders. [`ActivityCount::is_complete`] is `false` for
1118/// either, because `04-subsystem-contracts.md` forbids exactly this shape of
1119/// mistake on the other read model — "must never treat a first page as a
1120/// complete inventory" — and a count truncated at page four is the same defect
1121/// wearing a different endpoint.
1122#[derive(Debug, Clone, PartialEq, Eq, Default)]
1123pub struct ActivityCount {
1124    per_repository: BTreeMap<OwnerRepo, u32>,
1125    unavailable: Vec<UnavailableRepository>,
1126    /// Repositories whose count is a floor rather than a total.
1127    truncated: BTreeSet<OwnerRepo>,
1128}
1129
1130/// A repository the aggregate could not read, and why.
1131///
1132/// Carried out of the count rather than folded into it. An organization whose
1133/// App installation includes an archived or since-deleted repository would
1134/// otherwise fail its whole activity refresh forever, or — worse — quietly
1135/// return a total that is short by an unknown amount. `c2`'s installation
1136/// discovery makes the same choice for a nameless installation, and for the same
1137/// reason: a partial answer is usable only when it says it is partial.
1138#[derive(Debug, Clone, PartialEq, Eq)]
1139pub struct UnavailableRepository {
1140    pub repository: OwnerRepo,
1141    pub reason: String,
1142}
1143
1144impl ActivityCount {
1145    #[must_use]
1146    pub fn new(per_repository: BTreeMap<OwnerRepo, u32>) -> Self {
1147        Self {
1148            per_repository,
1149            unavailable: Vec::new(),
1150            truncated: BTreeSet::new(),
1151        }
1152    }
1153
1154    /// One repository's count, for the common single-repository case.
1155    #[must_use]
1156    pub fn of(repository: OwnerRepo, count: u32) -> Self {
1157        Self::new(BTreeMap::from([(repository, count)]))
1158    }
1159
1160    /// Mark `repository`'s count a **floor** rather than a total.
1161    ///
1162    /// The programmable counterpart to what the fallback walk does when it stops
1163    /// at [`MAX_ACTIVITY_FALLBACK_PAGES`], and it exists so that the incomplete
1164    /// case is reachable from outside this module at all. [`Self::new`] and
1165    /// [`Self::of`] were the only public constructors; both yield an empty
1166    /// `truncated` **and** an empty `unavailable`, and the fields are private —
1167    /// so [`Self::is_complete`] could only ever be `true` for a caller holding a
1168    /// hand-built count, and every downstream consumer that renders the `false`
1169    /// path had no way to write a test for it.
1170    #[must_use]
1171    pub fn with_truncated(mut self, repository: OwnerRepo) -> Self {
1172        self.truncated.insert(repository);
1173        self
1174    }
1175
1176    /// Record a repository the count could not read, and why.
1177    ///
1178    /// The counterpart to [`Self::with_truncated`] for the *other* cause of an
1179    /// incomplete count — see [`Self::is_complete`] for why the two are not
1180    /// interchangeable. Deliberately does **not** insert a zero into
1181    /// [`Self::per_repository`]: a repository that could not be counted is
1182    /// unknown, not idle, and flattening it to zero is the exact defect
1183    /// [`UnavailableRepository`] exists to prevent.
1184    #[must_use]
1185    pub fn with_unavailable(mut self, repository: OwnerRepo, reason: impl Into<String>) -> Self {
1186        self.unavailable.push(UnavailableRepository {
1187            repository,
1188            reason: reason.into(),
1189        });
1190        self
1191    }
1192
1193    /// In-progress workflow runs across every repository in scope.
1194    ///
1195    /// A **floor** rather than a total when [`Self::truncated`] is non-empty,
1196    /// and short by an unknown amount when [`Self::unavailable`] is. Both make
1197    /// [`Self::is_complete`] `false`, which is the one question a caller
1198    /// rendering this number has to ask.
1199    #[must_use]
1200    pub fn total(&self) -> u32 {
1201        self.per_repository.values().copied().sum()
1202    }
1203
1204    #[must_use]
1205    pub fn per_repository(&self) -> &BTreeMap<OwnerRepo, u32> {
1206        &self.per_repository
1207    }
1208
1209    /// This repository's count, or `None` when it was not in scope.
1210    #[must_use]
1211    pub fn for_repository(&self, repository: &OwnerRepo) -> Option<u32> {
1212        self.per_repository.get(repository).copied()
1213    }
1214
1215    #[must_use]
1216    pub fn unavailable(&self) -> &[UnavailableRepository] {
1217        &self.unavailable
1218    }
1219
1220    /// Repositories whose count is a **floor**, not a total.
1221    ///
1222    /// The counterpart to [`RunnerInventory::truncated`], and here for the same
1223    /// reason: a number clipped by a page ceiling that does not say it was
1224    /// clipped is indistinguishable from a real one, and `g2` renders this
1225    /// number with no other way to find out.
1226    #[must_use]
1227    pub fn truncated(&self) -> &BTreeSet<OwnerRepo> {
1228        &self.truncated
1229    }
1230
1231    /// Whether this repository's count is a floor rather than a total.
1232    #[must_use]
1233    pub fn is_truncated(&self, repository: &OwnerRepo) -> bool {
1234        self.truncated.contains(repository)
1235    }
1236
1237    /// `true` when every repository in scope answered **and** every answer was
1238    /// exact.
1239    ///
1240    /// Deliberately one question rather than two. A caller that has to remember
1241    /// to ask about truncation separately is a caller that will forget, which is
1242    /// the same argument `is_repository_local_failure` makes about stepping over
1243    /// the only repository in scope.
1244    ///
1245    /// # `false` has two causes, and they have opposite remedies
1246    ///
1247    /// One question is right for *rendering* the number. It is not enough for
1248    /// *acting* on it, because the two ways a count can be incomplete point in
1249    /// opposite directions:
1250    ///
1251    /// * [`Self::truncated`] — the count is a **lower bound**. The repository
1252    ///   answered and there is at least this much work in progress, so scaling
1253    ///   **up** from it is sound; the real figure is only ever larger.
1254    /// * [`Self::unavailable`] — the count is **unknown**. Nothing was learned
1255    ///   about that repository, and a missing count is not a zero. Scaling on it
1256    ///   is guessing.
1257    ///
1258    /// So a caller that reads `false` as a uniform "do nothing" stalls scale-up
1259    /// on a repository that is demonstrably busy — the truncated case is
1260    /// *evidence of load*, not absence of it. Ask this question to decide
1261    /// whether to caveat the number; ask [`Self::truncated`] versus
1262    /// [`Self::unavailable`] to decide what to do about it.
1263    #[must_use]
1264    pub fn is_complete(&self) -> bool {
1265        self.unavailable.is_empty() && self.truncated.is_empty()
1266    }
1267}
1268
1269// ---------------------------------------------------------------------------
1270// Runner package downloads
1271// ---------------------------------------------------------------------------
1272
1273/// One runner-package download GitHub publishes.
1274///
1275/// # `sha256_checksum` is optional, and stays optional
1276///
1277/// It is optional in GitHub's response schema, and this layer passes that
1278/// through faithfully — as [`Option`], never as an empty string and never as a
1279/// default. `e2` **fails closed** on its absence, requiring an operator-pinned
1280/// digest rather than installing an unverified 150-300 MB package
1281/// (`05-infrastructure.md`), and it can only do that if this layer does not
1282/// paper the absence over.
1283///
1284/// Absent and empty are also kept apart. A missing field and a `null` both read
1285/// as `None`; a field GitHub sent as `""` reads as `Some("")`. Both are unusable
1286/// as a digest, but they are different facts about GitHub's response, and
1287/// collapsing them would leave `e2` unable to report which one it saw.
1288#[derive(Debug, Clone, PartialEq, Eq)]
1289pub struct RunnerDownload {
1290    /// GitHub's OS token: `win`, `osx`, `linux`.
1291    pub os: String,
1292    /// GitHub's architecture token: `x64`, `arm64`, `arm`.
1293    pub architecture: String,
1294    pub download_url: String,
1295    pub filename: String,
1296    pub sha256_checksum: Option<String>,
1297}
1298
1299impl RunnerDownload {
1300    /// Whether this entry is the package for `os`/`arch`.
1301    ///
1302    /// Both sides are parsed through the domain's own [`Os`] and [`Arch`], whose
1303    /// `FromStr` already accepts GitHub's package tokens — `win`/`osx`/`linux`
1304    /// and `x64`/`arm64`/`arm` — because [`Os::label_token`] was chosen to be
1305    /// those very tokens. Comparing parsed values rather than strings is what
1306    /// keeps a `windows`/`win` spelling difference from silently matching
1307    /// nothing.
1308    #[must_use]
1309    pub fn matches(&self, os: Os, arch: Arch) -> bool {
1310        self.os.parse::<Os>().is_ok_and(|found| found == os)
1311            && self
1312                .architecture
1313                .parse::<Arch>()
1314                .is_ok_and(|found| found == arch)
1315    }
1316
1317    /// The published digest, if GitHub published one.
1318    #[must_use]
1319    pub fn sha256_checksum(&self) -> Option<&str> {
1320        self.sha256_checksum.as_deref()
1321    }
1322}
1323
1324/// Every runner package GitHub publishes for a target.
1325#[derive(Debug, Clone, PartialEq, Eq, Default)]
1326pub struct RunnerDownloads {
1327    entries: Vec<RunnerDownload>,
1328}
1329
1330impl RunnerDownloads {
1331    #[must_use]
1332    pub fn new(entries: Vec<RunnerDownload>) -> Self {
1333        Self { entries }
1334    }
1335
1336    #[must_use]
1337    pub fn entries(&self) -> &[RunnerDownload] {
1338        &self.entries
1339    }
1340
1341    #[must_use]
1342    pub fn is_empty(&self) -> bool {
1343        self.entries.is_empty()
1344    }
1345
1346    /// The package for one OS and architecture, or `None` when GitHub publishes
1347    /// none — which `e2` must refuse before downloading anything, rather than
1348    /// falling back to a hardcoded URL.
1349    #[must_use]
1350    pub fn select(&self, os: Os, arch: Arch) -> Option<&RunnerDownload> {
1351        self.entries.iter().find(|entry| entry.matches(os, arch))
1352    }
1353}
1354
1355// ---------------------------------------------------------------------------
1356// The composed snapshot
1357// ---------------------------------------------------------------------------
1358
1359/// One target's read models, as of one instant.
1360///
1361/// This is what the TUI holds and what `e1` recomputes each refresh. The two
1362/// counts are deliberately reachable only through their own types — there is no
1363/// `total` on this struct — so that a screen has to say which aggregate it is
1364/// rendering.
1365#[derive(Debug, Clone, PartialEq, Eq)]
1366pub struct InventorySnapshot {
1367    pub target: ScaleTarget,
1368    pub runners: RunnerInventory,
1369    pub activity: ActivityCount,
1370    pub observed_at: Timestamp,
1371    /// What GitHub last said about the hourly quota on the responses that built
1372    /// this snapshot.
1373    pub headroom: Option<RateLimitHeadroom>,
1374}
1375
1376// ---------------------------------------------------------------------------
1377// The shared request budget
1378// ---------------------------------------------------------------------------
1379
1380/// What one target costs, per refresh, in requests against the shared ceiling.
1381///
1382/// # Why an organization is not a constant
1383///
1384/// `04-subsystem-contracts.md` tabulates a flat "per target, per hour" cost —
1385/// ~240 at the 60-second default, ~480 at the 30-second floor — and that table
1386/// is right for a **repository** target and wrong for an organization one. Two
1387/// of the three request classes are per-repository resources:
1388///
1389/// | Class | Repository target | Organization target with `n` installed repositories |
1390/// |---|---|---|
1391/// | runner inventory | 1 | 1 (there *is* an org runners endpoint) |
1392/// | in-progress workflow count | 1 | `n` |
1393/// | demand: queued runs plus jobs | 2 | `2n` |
1394/// | **per refresh** | **4** | **1 + 3n** |
1395///
1396/// At `n = 1` the two agree exactly, at 4 requests per refresh and 240 per hour
1397/// at the default interval, which is what makes this a refinement of the
1398/// documented table rather than a contradiction of it. At `n = 10` an
1399/// organization costs 31 requests per refresh — nearly eight times a repository
1400/// — and projecting it as one flat target understates the real spend by exactly
1401/// that factor. `f2`'s `org add` refusal therefore arrives much earlier than a
1402/// repository's would, which is a thing it has to be able to explain.
1403///
1404/// # What this model does not claim
1405///
1406/// It is a projection of *steady-state polling*, in whole requests per refresh.
1407/// It does not model a target whose runner inventory spans pages (a second page
1408/// is a second request), an interactive `auth status`, a JIT registration, or a
1409/// runner deletion. That is what [`BUDGET_SHARE_DIVISOR`] is for: the
1410/// projection is compared against half the ceiling, and the other half absorbs
1411/// everything this model deliberately does not attempt to count.
1412///
1413/// It also prices each repository's activity count at its **best case** of one
1414/// request. A count that has to take the no-`total_count` fallback costs up to
1415/// [`MAX_ACTIVITY_FALLBACK_PAGES`] — see
1416/// [`ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH`], which `f1` and `f2` read
1417/// directly.
1418#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1419pub struct TargetCost {
1420    scope: TargetScope,
1421    installed_repositories: u32,
1422    demand_requests_per_repository: u32,
1423}
1424
1425impl TargetCost {
1426    /// A repository target: one repository, by construction.
1427    #[must_use]
1428    pub const fn repository() -> Self {
1429        Self {
1430            scope: TargetScope::Repository,
1431            installed_repositories: 1,
1432            demand_requests_per_repository: DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH,
1433        }
1434    }
1435
1436    /// An organization target reaching `installed_repositories` repositories.
1437    #[must_use]
1438    pub const fn organization(installed_repositories: u32) -> Self {
1439        Self {
1440            scope: TargetScope::Organization,
1441            installed_repositories,
1442            demand_requests_per_repository: DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH,
1443        }
1444    }
1445
1446    /// Replace the demand cost with the one `c4` measured.
1447    ///
1448    /// `c4`'s specification says to "report the per-poll request count to `c3`'s
1449    /// budget model rather than estimating it there", and this is where it
1450    /// reports it. Without a seam, honouring that sentence would mean editing
1451    /// [`DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH`] — in this file, which `c4`
1452    /// does not own — so the sentence would have been unfollowable and the
1453    /// estimate would have quietly stayed the truth.
1454    ///
1455    /// The default is [`DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH`], which is
1456    /// what `04-subsystem-contracts.md` tabulates. This overrides that number
1457    /// and nothing else: the inventory and activity costs are this task's own
1458    /// and are measured against the requests it really issues.
1459    #[must_use]
1460    pub fn with_demand_requests_per_repository(mut self, requests: u32) -> Self {
1461        self.demand_requests_per_repository = requests;
1462        self
1463    }
1464
1465    /// The cost of the scope an activity refresh will actually walk.
1466    ///
1467    /// Preferred over [`TargetCost::organization`] wherever the repository set
1468    /// is already in hand, because it takes the count from the same list the
1469    /// requests will be issued against rather than from a number somebody
1470    /// passed in.
1471    #[must_use]
1472    pub fn from_activity_scope(scope: &ActivityScope) -> Self {
1473        match scope.target().scope() {
1474            TargetScope::Repository => Self::repository(),
1475            TargetScope::Organization => {
1476                Self::organization(u32::try_from(scope.repositories().len()).unwrap_or(u32::MAX))
1477            }
1478        }
1479    }
1480
1481    #[must_use]
1482    pub const fn scope(&self) -> TargetScope {
1483        self.scope
1484    }
1485
1486    #[must_use]
1487    pub const fn installed_repositories(&self) -> u32 {
1488        self.installed_repositories
1489    }
1490
1491    /// Requests one refresh of this target costs.
1492    #[must_use]
1493    pub const fn requests_per_refresh(&self) -> u32 {
1494        let repositories = match self.scope {
1495            TargetScope::Repository => 1,
1496            TargetScope::Organization => self.installed_repositories,
1497        };
1498        RUNNER_INVENTORY_REQUESTS_PER_REFRESH
1499            + repositories.saturating_mul(
1500                ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH + self.demand_requests_per_repository,
1501            )
1502    }
1503
1504    /// Requests one hour of refreshing this target at `interval` costs.
1505    #[must_use]
1506    pub fn requests_per_hour(&self, interval: RefreshInterval) -> u32 {
1507        self.requests_per_refresh()
1508            .saturating_mul(refreshes_per_hour(interval))
1509    }
1510}
1511
1512/// Refreshes one hour holds at `interval`.
1513#[must_use]
1514pub fn refreshes_per_hour(interval: RefreshInterval) -> u32 {
1515    SECONDS_PER_HOUR / u32::from(interval.as_secs())
1516}
1517
1518/// The requests per hour a host may plan to spend: half the documented ceiling.
1519#[must_use]
1520pub const fn budget_allowance() -> u32 {
1521    HOURLY_REQUEST_CEILING / BUDGET_SHARE_DIVISOR
1522}
1523
1524/// What a host's configured target set will cost per hour, and whether that
1525/// fits.
1526///
1527/// `f1`'s `host show` renders [`BudgetProjection::requests_per_hour`],
1528/// [`BudgetProjection::headroom`] and
1529/// [`BudgetProjection::max_repository_targets`]; `f2`'s `repo add` and `org add`
1530/// call [`BudgetProjection::admit`] and refuse on
1531/// [`Admission::Refused`]. `g3` shows the same numbers in the TUI. All four read
1532/// one model, which is the only way the CLI and the TUI can agree about why an
1533/// eleventh repository was refused.
1534#[derive(Debug, Clone, PartialEq, Eq)]
1535pub struct BudgetProjection {
1536    interval: RefreshInterval,
1537    targets: Vec<TargetCost>,
1538}
1539
1540impl BudgetProjection {
1541    #[must_use]
1542    pub fn new(interval: RefreshInterval, targets: impl IntoIterator<Item = TargetCost>) -> Self {
1543        Self {
1544            interval,
1545            targets: targets.into_iter().collect(),
1546        }
1547    }
1548
1549    #[must_use]
1550    pub fn interval(&self) -> RefreshInterval {
1551        self.interval
1552    }
1553
1554    #[must_use]
1555    pub fn targets(&self) -> &[TargetCost] {
1556        &self.targets
1557    }
1558
1559    #[must_use]
1560    pub fn refreshes_per_hour(&self) -> u32 {
1561        refreshes_per_hour(self.interval)
1562    }
1563
1564    /// The projected hourly request count for the whole target set.
1565    #[must_use]
1566    pub fn requests_per_hour(&self) -> u32 {
1567        self.targets
1568            .iter()
1569            .map(|target| target.requests_per_hour(self.interval))
1570            .fold(0, u32::saturating_add)
1571    }
1572
1573    #[must_use]
1574    pub fn ceiling(&self) -> u32 {
1575        HOURLY_REQUEST_CEILING
1576    }
1577
1578    #[must_use]
1579    pub fn allowance(&self) -> u32 {
1580        budget_allowance()
1581    }
1582
1583    /// Requests per hour still available inside the allowance.
1584    #[must_use]
1585    pub fn headroom(&self) -> u32 {
1586        self.allowance().saturating_sub(self.requests_per_hour())
1587    }
1588
1589    #[must_use]
1590    pub fn exceeds_allowance(&self) -> bool {
1591        self.requests_per_hour() > self.allowance()
1592    }
1593
1594    /// How many **repository** targets one host can serve at `interval`.
1595    ///
1596    /// `04-subsystem-contracts.md` states the answer as "roughly 10 targets per
1597    /// host at the 60-second default and 5 at the 30-second floor", and this
1598    /// reproduces both. It is stated in repository targets because that is the
1599    /// only target whose cost is a constant; an organization's depends on its
1600    /// installed repository count, so "how many organizations fit" has no single
1601    /// answer and this deliberately does not invent one.
1602    #[must_use]
1603    pub fn max_repository_targets(interval: RefreshInterval) -> u32 {
1604        let per_target = TargetCost::repository().requests_per_hour(interval);
1605        if per_target == 0 {
1606            return 0;
1607        }
1608        budget_allowance() / per_target
1609    }
1610
1611    /// Whether one more target fits.
1612    #[must_use]
1613    pub fn admit(&self, candidate: TargetCost) -> Admission {
1614        let candidate_per_hour = candidate.requests_per_hour(self.interval);
1615        let projected = self.requests_per_hour().saturating_add(candidate_per_hour);
1616        let allowance = self.allowance();
1617        if projected > allowance {
1618            return Admission::Refused {
1619                candidate,
1620                candidate_requests_per_hour: candidate_per_hour,
1621                projected_requests_per_hour: projected,
1622                allowance,
1623                ceiling: self.ceiling(),
1624                interval: self.interval,
1625                max_repository_targets: Self::max_repository_targets(self.interval),
1626            };
1627        }
1628        Admission::Admitted {
1629            projected_requests_per_hour: projected,
1630            headroom_after: allowance - projected,
1631        }
1632    }
1633}
1634
1635/// The answer `f2`'s `repo add` and `org add` act on.
1636#[derive(Debug, Clone, PartialEq, Eq)]
1637pub enum Admission {
1638    Admitted {
1639        projected_requests_per_hour: u32,
1640        headroom_after: u32,
1641    },
1642    Refused {
1643        candidate: TargetCost,
1644        candidate_requests_per_hour: u32,
1645        projected_requests_per_hour: u32,
1646        allowance: u32,
1647        ceiling: u32,
1648        interval: RefreshInterval,
1649        max_repository_targets: u32,
1650    },
1651}
1652
1653impl Admission {
1654    #[must_use]
1655    pub fn is_admitted(&self) -> bool {
1656        matches!(self, Self::Admitted { .. })
1657    }
1658}
1659
1660impl fmt::Display for Admission {
1661    /// The refusal has to explain itself: "an operator who adds an eleventh
1662    /// repository needs to know why it was refused"
1663    /// (`04-subsystem-contracts.md`). So the message carries the computed
1664    /// numbers rather than the rule, and — for an organization — says which
1665    /// repository count drove them, because that is the part a flat per-target
1666    /// reading of the design would not have predicted.
1667    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1668        match self {
1669            Self::Admitted {
1670                projected_requests_per_hour,
1671                headroom_after,
1672            } => write!(
1673                f,
1674                "projected {projected_requests_per_hour} requests/hour, \
1675                 {headroom_after} remaining in this host's share of the budget"
1676            ),
1677            Self::Refused {
1678                candidate,
1679                candidate_requests_per_hour,
1680                projected_requests_per_hour,
1681                allowance,
1682                ceiling,
1683                interval,
1684                max_repository_targets,
1685            } => {
1686                write!(
1687                    f,
1688                    "refused: this target would take the host to \
1689                     {projected_requests_per_hour} requests/hour, over the {allowance} it may \
1690                     plan to spend (half of GitHub's {ceiling}/hour ceiling) at a \
1691                     {}-second refresh interval. This host can serve about \
1692                     {max_repository_targets} repository targets at that interval",
1693                    interval.as_secs()
1694                )?;
1695                if candidate.scope() == TargetScope::Organization {
1696                    write!(
1697                        f,
1698                        ". This organization alone costs {candidate_requests_per_hour} \
1699                         requests/hour because the App is installed on {} of its repositories, \
1700                         and workflow runs are a per-repository resource",
1701                        candidate.installed_repositories()
1702                    )?;
1703                }
1704                Ok(())
1705            }
1706        }
1707    }
1708}
1709
1710// ---------------------------------------------------------------------------
1711// Coalescing a manual refresh with an in-flight one
1712// ---------------------------------------------------------------------------
1713
1714/// Runs one refresh at a time; a refresh asked for while another is in flight
1715/// **joins** it instead of issuing a second.
1716///
1717/// `04-subsystem-contracts.md`: "Manual refresh coalesces with an in-flight
1718/// request." The requirement is a budget one before it is a latency one — `F5`
1719/// held down on the dashboard would otherwise be an operator-driven denial of
1720/// service against a 5,000/hour ceiling shared with the polling that keeps
1721/// runners starting.
1722///
1723/// The mechanism is the generation-and-gate pattern
1724/// [`crate::AuthenticatedClient::revalidate`] already uses for single-flight
1725/// re-validation, and it is here rather than there because the two coalesce
1726/// different things. A caller samples the generation *before* queuing on the
1727/// gate; if it moved while the caller waited, some other refresh covered it and
1728/// this one returns that result without calling `work` at all. `work` being
1729/// `FnOnce` is what makes "no second request" structural rather than
1730/// remembered: the joining path never has a future to poll.
1731///
1732/// # One instance per target. This is a requirement, not a convention
1733///
1734/// `last` is a single slot and `generation` is a single counter, so an instance
1735/// can only ever be a cache of *one* thing. Sharing one coalescer across two
1736/// targets does not merely lose cache hits — it hands target A's caller target
1737/// B's snapshot, silently and with no error, because joining a generation that
1738/// moved is precisely how this type reports "somebody else already refreshed
1739/// what you asked for". Nothing here can detect that the somebody else was
1740/// refreshing something different.
1741///
1742/// `e1` and `g2` therefore hold one instance per [`ScaleTarget`] — keyed by
1743/// target in whatever map they already keep — and never one per host. The type
1744/// cannot enforce it, which is exactly why it is written down.
1745#[derive(Debug)]
1746pub struct RefreshCoalescer<T> {
1747    generation: AtomicU64,
1748    gate: tokio::sync::Mutex<()>,
1749    last: std::sync::Mutex<Option<T>>,
1750    performed: AtomicU64,
1751    joined: AtomicU64,
1752}
1753
1754impl<T: Clone> Default for RefreshCoalescer<T> {
1755    fn default() -> Self {
1756        Self::new()
1757    }
1758}
1759
1760impl<T: Clone> RefreshCoalescer<T> {
1761    #[must_use]
1762    pub fn new() -> Self {
1763        Self {
1764            generation: AtomicU64::new(0),
1765            gate: tokio::sync::Mutex::new(()),
1766            last: std::sync::Mutex::new(None),
1767            performed: AtomicU64::new(0),
1768            joined: AtomicU64::new(0),
1769        }
1770    }
1771
1772    /// Refresh, or join the refresh already running.
1773    ///
1774    /// # Panics
1775    /// If a previous holder panicked while the result lock was held.
1776    pub async fn refresh<F, Fut>(&self, work: F) -> T
1777    where
1778        F: FnOnce() -> Fut,
1779        Fut: Future<Output = T>,
1780    {
1781        let sampled = self.generation.load(Ordering::SeqCst);
1782        let _guard = self.gate.lock().await;
1783
1784        if self.generation.load(Ordering::SeqCst) != sampled
1785            && let Some(shared) = self.last.lock().expect("refresh lock poisoned").clone()
1786        {
1787            self.joined.fetch_add(1, Ordering::SeqCst);
1788            tracing::debug!("joined an in-flight refresh instead of issuing a second request");
1789            return shared;
1790        }
1791
1792        let outcome = work().await;
1793        *self.last.lock().expect("refresh lock poisoned") = Some(outcome.clone());
1794        self.performed.fetch_add(1, Ordering::SeqCst);
1795        // Bumped last and under the gate: a caller that sampled before this
1796        // point and is still queued will see the change and join.
1797        self.generation.fetch_add(1, Ordering::SeqCst);
1798        outcome
1799    }
1800
1801    /// How many refreshes actually ran.
1802    #[must_use]
1803    pub fn performed(&self) -> u64 {
1804        self.performed.load(Ordering::SeqCst)
1805    }
1806
1807    /// How many refreshes were served by joining one already in flight.
1808    #[must_use]
1809    pub fn joined(&self) -> u64 {
1810        self.joined.load(Ordering::SeqCst)
1811    }
1812
1813    /// The most recent outcome, if there has been one.
1814    ///
1815    /// # Panics
1816    /// If a previous holder panicked while the result lock was held.
1817    #[must_use]
1818    pub fn last(&self) -> Option<T> {
1819        self.last.lock().expect("refresh lock poisoned").clone()
1820    }
1821}
1822
1823// ---------------------------------------------------------------------------
1824// The gateway seam
1825// ---------------------------------------------------------------------------
1826
1827/// Every read model the dashboard and the CLI display.
1828///
1829/// A trait rather than a concrete type, so that `e1`, `f1`, `g2` and `g3` can be
1830/// tested against `runner_manager_testkit::github::FakeGithub` with no network
1831/// and no `wiremock` in their dependency graphs. [`RestInventory`] is the one
1832/// implementation that talks to GitHub.
1833#[async_trait::async_trait]
1834pub trait InventoryGateway: fmt::Debug + Send + Sync {
1835    /// Every runner GitHub reports for `target`, across every page.
1836    ///
1837    /// # Errors
1838    /// Every variant of [`InventoryError`].
1839    async fn list_runners(
1840        &self,
1841        target: &ScaleTarget,
1842        cancel: &CancelToken,
1843    ) -> Result<RunnerInventory, InventoryError>;
1844
1845    /// Delete one runner registration from `target`.
1846    ///
1847    /// The agent owns the registrations it created, and this is how it gives
1848    /// them back. GitHub retires an ephemeral runner promptly once that runner
1849    /// *completes a job*. Every other ending is on its own schedule: a
1850    /// registration whose runner never got work, or whose process died still
1851    /// holding it, lingers in the target's runner settings — one was observed
1852    /// listed for **33 hours** after the attempt behind it had concluded. GitHub
1853    /// does clear such a registration eventually, so this is not the difference
1854    /// between forever and not; it is the difference between an operator seeing
1855    /// a runner row that matches reality and one that does not.
1856    ///
1857    /// A `404` is success: the registration is gone, which is the postcondition
1858    /// asked for, and treating GitHub having already removed it as a failure
1859    /// would strand every attempt that concluded the ordinary way.
1860    ///
1861    /// # Errors
1862    /// Every variant of [`InventoryError`].
1863    async fn remove_runner(
1864        &self,
1865        target: &ScaleTarget,
1866        runner_id: u64,
1867        cancel: &CancelToken,
1868    ) -> Result<(), InventoryError>;
1869
1870    /// In-progress workflow runs across `scope`.
1871    ///
1872    /// # Errors
1873    /// Every variant of [`InventoryError`].
1874    async fn in_progress_activity(
1875        &self,
1876        scope: &ActivityScope,
1877        cancel: &CancelToken,
1878    ) -> Result<ActivityCount, InventoryError>;
1879
1880    /// The runner packages GitHub publishes for `target`.
1881    ///
1882    /// # Errors
1883    /// Every variant of [`InventoryError`].
1884    async fn runner_downloads(
1885        &self,
1886        target: &ScaleTarget,
1887        cancel: &CancelToken,
1888    ) -> Result<RunnerDownloads, InventoryError>;
1889
1890    /// What GitHub last said about the hourly quota, if anything.
1891    fn headroom(&self) -> Option<RateLimitHeadroom>;
1892
1893    /// The instant every snapshot is stamped with.
1894    fn now(&self) -> Timestamp;
1895
1896    /// Both read models for one target, in one refresh.
1897    ///
1898    /// A provided method rather than a required one: it is the composition every
1899    /// caller wants and it must not be possible for an implementation to compose
1900    /// the two counts differently from another.
1901    ///
1902    /// # Errors
1903    /// Every variant of [`InventoryError`].
1904    async fn snapshot(
1905        &self,
1906        scope: &ActivityScope,
1907        cancel: &CancelToken,
1908    ) -> Result<InventorySnapshot, InventoryError> {
1909        let runners = self.list_runners(scope.target(), cancel).await?;
1910        let activity = self.in_progress_activity(scope, cancel).await?;
1911        Ok(InventorySnapshot {
1912            target: scope.target().clone(),
1913            runners,
1914            activity,
1915            observed_at: self.now(),
1916            headroom: self.headroom(),
1917        })
1918    }
1919}
1920
1921// ---------------------------------------------------------------------------
1922// The GitHub implementation
1923// ---------------------------------------------------------------------------
1924
1925#[derive(Debug)]
1926struct RateLimitState {
1927    until: Option<Timestamp>,
1928    last: Option<RateLimited>,
1929}
1930
1931/// [`InventoryGateway`] over `api.github.com`.
1932///
1933/// Holds no credential of its own: authentication is entirely
1934/// [`AuthenticatedClient`]'s, and this type only ever hands it an
1935/// [`ApiRequest`].
1936pub struct RestInventory {
1937    client: Arc<AuthenticatedClient>,
1938    clock: Arc<dyn Clock>,
1939    rate_limit: std::sync::Mutex<RateLimitState>,
1940    headroom: std::sync::Mutex<Option<RateLimitHeadroom>>,
1941    requests_issued: AtomicU64,
1942}
1943
1944impl fmt::Debug for RestInventory {
1945    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1946        // `try_lock`, for the reason `AuthenticatedClient`'s own `Debug` records:
1947        // a `Debug` impl must never be able to block, and these locks are held
1948        // across code that could plausibly grow a `tracing` call.
1949        let backing_off = match self.rate_limit.try_lock() {
1950            Ok(state) => state
1951                .until
1952                .is_some_and(|until| self.clock.now() < until)
1953                .to_string(),
1954            Err(_) => "unknown (the rate-limit state is being updated)".to_string(),
1955        };
1956        f.debug_struct("RestInventory")
1957            .field(
1958                "requests_issued",
1959                &self.requests_issued.load(Ordering::Relaxed),
1960            )
1961            .field("rate_limited", &backing_off)
1962            .finish_non_exhaustive()
1963    }
1964}
1965
1966impl RestInventory {
1967    #[must_use]
1968    pub fn new(client: Arc<AuthenticatedClient>, clock: Arc<dyn Clock>) -> Self {
1969        Self {
1970            client,
1971            clock,
1972            rate_limit: std::sync::Mutex::new(RateLimitState {
1973                until: None,
1974                last: None,
1975            }),
1976            headroom: std::sync::Mutex::new(None),
1977            requests_issued: AtomicU64::new(0),
1978        }
1979    }
1980
1981    /// How many HTTP requests this gateway has issued.
1982    ///
1983    /// The budget model above projects a per-refresh cost in whole requests, and
1984    /// a projection nothing measures is a table in a document. This is what the
1985    /// tests measure it against.
1986    #[must_use]
1987    pub fn requests_issued(&self) -> u64 {
1988        self.requests_issued.load(Ordering::SeqCst)
1989    }
1990
1991    /// How much of a rate-limit back-off is left, or `None` when not backing
1992    /// off.
1993    ///
1994    /// # Panics
1995    /// If a previous holder panicked while the rate-limit lock was held.
1996    #[must_use]
1997    pub fn rate_limit_backoff(&self) -> Option<Duration> {
1998        let state = self.rate_limit.lock().expect("rate-limit lock poisoned");
1999        let until = state.until?;
2000        let now = self.clock.now();
2001        if now >= until {
2002            return None;
2003        }
2004        (until - now).to_std().ok()
2005    }
2006
2007    /// The rate limit currently being backed off from, for display.
2008    ///
2009    /// # Panics
2010    /// If a previous holder panicked while the rate-limit lock was held.
2011    #[must_use]
2012    pub fn rate_limit_state(&self) -> Option<RateLimited> {
2013        let remaining = self.rate_limit_backoff()?;
2014        let state = self.rate_limit.lock().expect("rate-limit lock poisoned");
2015        let mut limit = state.last?;
2016        // Report what is left of the wait, not what GitHub asked for when the
2017        // window opened. A countdown that never moves reads as a hung refresh.
2018        limit.retry_after = Some(remaining);
2019        Some(limit)
2020    }
2021
2022    /// Forget a rate-limit back-off. Nothing in the product needs this — the
2023    /// window expires against the clock — but a test that wants to prove the
2024    /// window is what suppressed a request does.
2025    ///
2026    /// # Panics
2027    /// If a previous holder panicked while the rate-limit lock was held.
2028    pub fn clear_rate_limit(&self) {
2029        self.rate_limit
2030            .lock()
2031            .expect("rate-limit lock poisoned")
2032            .until = None;
2033    }
2034
2035    /// One request, with cancellation and the rate-limit gate applied.
2036    ///
2037    /// # Cancellation is consulted twice, and the two are not redundant
2038    ///
2039    /// [`CancelToken::check`] decides *before* the rate-limit gate is read, and
2040    /// [`CancelToken::run`] covers a token flipped while the socket is already
2041    /// open. Removing either one leaves a real hole: without `check`, a caller
2042    /// that cancelled a refresh which was also rate-limited is answered
2043    /// [`InventoryError::RateLimited`] — told to wait for something it has
2044    /// already withdrawn — and without `run`, a cancellation arriving mid-flight
2045    /// is not noticed until the response does.
2046    ///
2047    /// They do overlap for the between-pages case, and deliberately: it is the
2048    /// one the shared budget cares about, and a walk that keeps paging after the
2049    /// operator navigated away spends real requests. A mutation test that
2050    /// disables `check` alone leaves that case still guarded by `run`, which is
2051    /// what defence in depth is supposed to look like.
2052    async fn issue(
2053        &self,
2054        request: &ApiRequest,
2055        cancel: &CancelToken,
2056    ) -> Result<ApiResponse, InventoryError> {
2057        cancel.check()?;
2058        if let Some(limit) = self.rate_limit_state() {
2059            // Obeying `retry-after` by issuing nothing. No socket is opened, so
2060            // the wait costs the shared budget nothing at all.
2061            tracing::debug!(
2062                method = request.method().as_str(),
2063                path = %request.path(),
2064                remaining_secs = limit.retry_after.unwrap_or_default().as_secs(),
2065                "suppressed a request: GitHub's rate limit is still backing off"
2066            );
2067            return Err(InventoryError::RateLimited(limit));
2068        }
2069
2070        let result = cancel
2071            .run(async {
2072                // Counted *inside* the future, so the count is of requests
2073                // actually attempted. Counting before `run` over-reported by one
2074                // whenever a token was flipped between the check above and the
2075                // first poll: `run`'s biased `select!` then answers
2076                // `Cancelled` without ever polling this block, so no socket is
2077                // opened — and a budget model measured against an over-count is
2078                // a budget model that drifts every time an operator cancels.
2079                self.requests_issued.fetch_add(1, Ordering::SeqCst);
2080                self.client
2081                    .send(request)
2082                    .await
2083                    .map_err(InventoryError::from)
2084            })
2085            .await;
2086
2087        match result {
2088            Ok(response) => {
2089                if let Some(headroom) = RateLimitHeadroom::from_response(&response) {
2090                    *self.headroom.lock().expect("headroom lock poisoned") = Some(headroom);
2091                }
2092                Ok(response)
2093            }
2094            Err(InventoryError::Github(error)) => Err(self.classify(error)),
2095            Err(other) => Err(other),
2096        }
2097    }
2098
2099    /// Turn a failure into a rate limit when GitHub's own evidence says it is
2100    /// one, and latch the back-off window if so.
2101    fn classify(&self, error: GithubError) -> InventoryError {
2102        let Some(limit) = RateLimited::detect(&error) else {
2103            return InventoryError::Github(error);
2104        };
2105        let now = self.clock.now();
2106        let delay = limit.delay_from(now);
2107        if let Ok(delta) = chrono::TimeDelta::from_std(delay) {
2108            let mut state = self.rate_limit.lock().expect("rate-limit lock poisoned");
2109            state.until = Some(now + delta);
2110            state.last = Some(limit);
2111        }
2112        tracing::warn!(
2113            kind = %limit.kind,
2114            delay_secs = delay.as_secs(),
2115            remaining = limit.remaining,
2116            "GitHub is rate limiting this credential; delaying refreshes and reporting it"
2117        );
2118        InventoryError::RateLimited(limit)
2119    }
2120
2121    /// Follow `Link: rel="next"` to the end of a collection.
2122    async fn collect_pages<P: WirePage>(
2123        &self,
2124        first: ApiRequest,
2125        cancel: &CancelToken,
2126    ) -> Result<Collected<P::Item>, InventoryError> {
2127        let mut items = Vec::new();
2128        let mut reported_total = None;
2129        let mut pages = 0_usize;
2130        let mut truncated = false;
2131        let mut next = Some(first);
2132
2133        while let Some(request) = next.take() {
2134            // Cancellation is checked at the top of `issue`, which is what makes a
2135            // token flipped after page one stop the walk before page two.
2136            let response = self.issue(&request, cancel).await?;
2137            let page: P = response.json()?;
2138            reported_total = page.reported_total().or(reported_total);
2139            items.extend(page.into_items());
2140            pages += 1;
2141
2142            if pages >= MAX_PAGES {
2143                truncated = true;
2144                tracing::warn!(
2145                    what = P::WHAT,
2146                    pages,
2147                    collected = items.len(),
2148                    "stopped following pages at the ceiling; a `Link: rel=next` that never \
2149                     ends would otherwise loop forever"
2150                );
2151                break;
2152            }
2153            next = response
2154                .next_page()
2155                .map(|url| ApiRequest::get(url.as_str()));
2156        }
2157
2158        Ok(Collected {
2159            items,
2160            reported_total,
2161            pages,
2162            truncated,
2163        })
2164    }
2165
2166    /// In-progress workflow runs for one repository.
2167    ///
2168    /// One request in the ordinary case. GitHub answers the filtered query with
2169    /// its own `total_count`, which is the count this product wants, so a
2170    /// repository with 400 in-progress runs still costs one request rather than
2171    /// four — and the budget table's "one request per refresh" stays true.
2172    ///
2173    /// The fallback matters anyway: a response with no `total_count` is counted
2174    /// by walking the pages, because guessing zero from a missing field would
2175    /// render a busy repository as idle. That walk is bounded by
2176    /// [`MAX_ACTIVITY_FALLBACK_PAGES`] rather than [`crate::MAX_PAGES`], and
2177    /// stopping at the bound makes the answer inexact rather than merely
2178    /// smaller.
2179    ///
2180    /// # The `total_count` assumption is checked, because checking it is free
2181    ///
2182    /// Reading the reported total for a *filtered* query assumes that total is
2183    /// the count of the filtered set rather than of every run the repository has
2184    /// ever had. GitHub documents it that way and this product depends on it —
2185    /// but the same envelope reaches `c4`'s `clamp()` on `status=queued`, so the
2186    /// assumption is worth more than a dashboard number.
2187    ///
2188    /// It is checkable with no extra request. When there is no `rel="next"`, the
2189    /// whole filtered set is on this page, so `total_count` **must** equal
2190    /// `workflow_runs.len()`. A repository with 3 in-progress runs out of 5,000
2191    /// lifetime runs would answer `len() == 3`, no `Link`, and
2192    /// `total_count == 5000` — a contradiction visible on the first response.
2193    /// Discarding that disagreement is what would leave the assumption
2194    /// falsifiable only by an operator noticing a wrong number.
2195    async fn repository_in_progress(
2196        &self,
2197        repository: &OwnerRepo,
2198        cancel: &CancelToken,
2199    ) -> Result<RepositoryActivity, InventoryError> {
2200        let request = ApiRequest::get(format!(
2201            "/repos/{}/{}/actions/runs",
2202            repository.owner(),
2203            repository.repo()
2204        ))
2205        .query("status", "in_progress")
2206        .query("per_page", PER_PAGE);
2207
2208        let response = self.issue(&request, cancel).await?;
2209        let page: RunsPage = response.json()?;
2210        if let Some(total) = page.total_count {
2211            let listed = page.workflow_runs.len() as u64;
2212            if response.next_page().is_none() && total != listed {
2213                tracing::warn!(
2214                    repository = %repository,
2215                    total_count = total,
2216                    listed,
2217                    "GitHub's `total_count` disagrees with the single page it sent for a \
2218                     filtered query; this layer reads `total_count` as the count of the \
2219                     filtered set, and that reading looks wrong"
2220                );
2221                // The `warn!` above fires on any disagreement; this does not, and
2222                // the asymmetry is the whole point — see
2223                // `MAX_BENIGN_TOTAL_COUNT_SKEW`. A handful over is the race this
2224                // check's own doc calls legitimate; thousands over is the
2225                // unfiltered total it was written to catch.
2226                debug_assert!(
2227                    total <= listed.saturating_add(MAX_BENIGN_TOTAL_COUNT_SKEW),
2228                    "`total_count` ({total}) exceeds the {listed} run(s) on the only page \
2229                     of a filtered query by more than {MAX_BENIGN_TOTAL_COUNT_SKEW}, which \
2230                     is far past the run-finishing-mid-serialisation race; `total_count` \
2231                     is not the filtered count, and every in-progress figure — and `c4`'s \
2232                     demand — is being read off the wrong field"
2233                );
2234            }
2235            return Ok(RepositoryActivity::from_reported_total(total, repository));
2236        }
2237
2238        // No `total_count`: count what is there, following pages.
2239        let mut counted = page.workflow_runs.len();
2240        let mut pages = 1_usize;
2241        let mut next = response
2242            .next_page()
2243            .map(|url| ApiRequest::get(url.as_str()));
2244        while let Some(request) = next.take() {
2245            if pages >= MAX_ACTIVITY_FALLBACK_PAGES {
2246                tracing::warn!(
2247                    repository = %repository,
2248                    pages,
2249                    counted,
2250                    "stopped counting in-progress runs at the activity page budget; the \
2251                     count reported for this repository is a floor, not a total"
2252                );
2253                // A floor, and it says so. Returning it as an exact count is the
2254                // defect `04-subsystem-contracts.md` forbids for inventory, on
2255                // the other read model.
2256                return Ok(RepositoryActivity::floor(counted));
2257            }
2258            let response = self.issue(&request, cancel).await?;
2259            let page: RunsPage = response.json()?;
2260            counted += page.workflow_runs.len();
2261            pages += 1;
2262            next = response
2263                .next_page()
2264                .map(|url| ApiRequest::get(url.as_str()));
2265        }
2266        Ok(RepositoryActivity::exact(counted))
2267    }
2268
2269    /// The runners path for either scope.
2270    fn runners_path(target: &ScaleTarget) -> String {
2271        match target {
2272            ScaleTarget::Repository(repo) => {
2273                format!("/repos/{}/{}/actions/runners", repo.owner(), repo.repo())
2274            }
2275            ScaleTarget::Organization(org) => format!("/orgs/{}/actions/runners", org.as_str()),
2276        }
2277    }
2278}
2279
2280/// Whether a per-repository failure should be recorded and stepped over, or
2281/// should abort the whole aggregate.
2282///
2283/// The line is between a fact about *that repository* and a fact about the
2284/// credential or the connection. A `404` (deleted, renamed, or never reachable)
2285/// and a plain `403` (Actions disabled on that repository) are the first;
2286/// everything else — a rate limit, a rejected credential, an authentication
2287/// lockout, an unreachable host, an undecodable body — is the second, because
2288/// stepping over those would report a total that is short by an unknown amount
2289/// while looking complete.
2290///
2291/// # It applies to an aggregate only, never to a repository target
2292///
2293/// Stepping over the *only* repository in scope turns a permissions failure into
2294/// `ActivityCount { total: 0 }`, and a dashboard that reads `total()` — which is
2295/// the obvious thing to read — then renders "0 in progress" for a target it
2296/// cannot see at all. [`ActivityCount::is_complete`] says otherwise, but a
2297/// safety property that depends on every caller remembering to ask a second
2298/// question is not a safety property.
2299///
2300/// So the caller checks [`TargetScope`] first: an organization aggregate steps
2301/// over a bad repository, and a repository target propagates. The step-over
2302/// exists because one archived repository must not take down an organization's
2303/// whole activity refresh — a scope of one has no such problem to solve.
2304fn is_repository_local_failure(error: &InventoryError) -> bool {
2305    match error {
2306        InventoryError::Github(GithubError::Forbidden { .. }) => true,
2307        InventoryError::Github(GithubError::Status { status, .. }) => *status == 404,
2308        _ => false,
2309    }
2310}
2311
2312#[async_trait::async_trait]
2313impl InventoryGateway for RestInventory {
2314    async fn list_runners(
2315        &self,
2316        target: &ScaleTarget,
2317        cancel: &CancelToken,
2318    ) -> Result<RunnerInventory, InventoryError> {
2319        let request = ApiRequest::get(Self::runners_path(target)).query("per_page", PER_PAGE);
2320        let collected = self.collect_pages::<RunnersPage>(request, cancel).await?;
2321
2322        let runners: Vec<Runner> = collected.items.into_iter().map(Runner::from).collect();
2323        let inventory = RunnerInventory::paged(
2324            target.clone(),
2325            runners,
2326            collected.reported_total,
2327            collected.pages,
2328            collected.truncated,
2329        );
2330        if let Some(missing) = inventory.missing() {
2331            // Not an error, and deliberately not silence either: `g2` can render
2332            // "showing 200 of 250" but only if it is told.
2333            tracing::warn!(
2334                target = %target,
2335                missing,
2336                collected = inventory.len(),
2337                "GitHub reported more runners than pagination collected; this inventory is \
2338                 incomplete"
2339            );
2340        }
2341        Ok(inventory)
2342    }
2343
2344    async fn remove_runner(
2345        &self,
2346        target: &ScaleTarget,
2347        runner_id: u64,
2348        cancel: &CancelToken,
2349    ) -> Result<(), InventoryError> {
2350        let path = format!("{}/{runner_id}", Self::runners_path(target));
2351        match self.issue(&ApiRequest::delete(path), cancel).await {
2352            Ok(_) => Ok(()),
2353            // Already gone is the state this asks for. See the trait method.
2354            Err(InventoryError::Github(GithubError::Status { status: 404, .. })) => Ok(()),
2355            Err(error) => Err(error),
2356        }
2357    }
2358
2359    async fn in_progress_activity(
2360        &self,
2361        scope: &ActivityScope,
2362        cancel: &CancelToken,
2363    ) -> Result<ActivityCount, InventoryError> {
2364        let mut per_repository = BTreeMap::new();
2365        let mut unavailable = Vec::new();
2366        let mut truncated = BTreeSet::new();
2367        // Only an aggregate steps over a bad repository; see
2368        // `is_repository_local_failure`.
2369        let aggregating = scope.target().scope() == TargetScope::Organization;
2370
2371        for repository in scope.repositories() {
2372            match self.repository_in_progress(repository, cancel).await {
2373                Ok(activity) => {
2374                    per_repository.insert(repository.clone(), activity.count);
2375                    if !activity.exact {
2376                        // A floor travels with the aggregate rather than being
2377                        // flattened into it: one truncated repository makes the
2378                        // *total* a floor too, and `g2` has no other way to know.
2379                        truncated.insert(repository.clone());
2380                    }
2381                }
2382                Err(error) if aggregating && is_repository_local_failure(&error) => {
2383                    tracing::warn!(
2384                        repository = %repository,
2385                        error = %error,
2386                        "a repository in this organization could not be counted; the aggregate \
2387                         reports it as unavailable rather than as zero"
2388                    );
2389                    unavailable.push(UnavailableRepository {
2390                        repository: repository.clone(),
2391                        reason: error.to_string(),
2392                    });
2393                }
2394                Err(error) => return Err(error),
2395            }
2396        }
2397
2398        Ok(ActivityCount {
2399            per_repository,
2400            unavailable,
2401            truncated,
2402        })
2403    }
2404
2405    async fn runner_downloads(
2406        &self,
2407        target: &ScaleTarget,
2408        cancel: &CancelToken,
2409    ) -> Result<RunnerDownloads, InventoryError> {
2410        let path = match target {
2411            ScaleTarget::Repository(repo) => format!(
2412                "/repos/{}/{}/actions/runners/downloads",
2413                repo.owner(),
2414                repo.repo()
2415            ),
2416            ScaleTarget::Organization(org) => {
2417                format!("/orgs/{}/actions/runners/downloads", org.as_str())
2418            }
2419        };
2420        // Not paginated: GitHub answers this one with a bare JSON array of the
2421        // packages it publishes, which is a fixed handful.
2422        let response = self.issue(&ApiRequest::get(path), cancel).await?;
2423        let raw: Vec<RawDownload> = response.json()?;
2424        Ok(RunnerDownloads::new(
2425            raw.into_iter().map(RunnerDownload::from).collect(),
2426        ))
2427    }
2428
2429    fn headroom(&self) -> Option<RateLimitHeadroom> {
2430        *self.headroom.lock().expect("headroom lock poisoned")
2431    }
2432
2433    fn now(&self) -> Timestamp {
2434        self.clock.now()
2435    }
2436}
2437
2438// ---------------------------------------------------------------------------
2439// Wire shapes
2440// ---------------------------------------------------------------------------
2441
2442struct Collected<T> {
2443    items: Vec<T>,
2444    reported_total: Option<u64>,
2445    pages: usize,
2446    truncated: bool,
2447}
2448
2449/// One repository's in-progress count, and whether that number is the whole
2450/// truth.
2451///
2452/// [`Collected::truncated`]'s counterpart for the activity read model. It exists
2453/// so that "the walk stopped early" cannot be dropped on the floor between
2454/// [`RestInventory::repository_in_progress`] and the aggregate that renders it:
2455/// a `u32` alone has nowhere to carry the fact.
2456#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2457struct RepositoryActivity {
2458    count: u32,
2459    /// `false` when `count` is a **floor**: the page budget stopped the walk, or
2460    /// GitHub's own total was wider than the `u32` this product renders.
2461    exact: bool,
2462}
2463
2464impl RepositoryActivity {
2465    fn exact(count: usize) -> Self {
2466        Self {
2467            // A count this layer assembled itself, one page at a time, cannot
2468            // exceed `MAX_ACTIVITY_FALLBACK_PAGES * PER_PAGE`. The saturation is
2469            // unreachable rather than lossy.
2470            count: u32::try_from(count).unwrap_or(u32::MAX),
2471            exact: true,
2472        }
2473    }
2474
2475    fn floor(count: usize) -> Self {
2476        Self {
2477            count: u32::try_from(count).unwrap_or(u32::MAX),
2478            exact: false,
2479        }
2480    }
2481
2482    /// GitHub's own `total_count`, narrowed to the width this product renders.
2483    ///
2484    /// A total that does not fit a `u32` is not a number to saturate silently:
2485    /// `unwrap_or(u32::MAX)` alone would put `4294967295` on a dashboard as
2486    /// though it were a measurement. It is still a *floor* — the real count is
2487    /// larger, not smaller — so it is reported as one, through the same signal
2488    /// the page budget uses.
2489    fn from_reported_total(total: u64, repository: &OwnerRepo) -> Self {
2490        match u32::try_from(total) {
2491            Ok(count) => Self { count, exact: true },
2492            Err(_) => {
2493                tracing::warn!(
2494                    repository = %repository,
2495                    total_count = total,
2496                    "GitHub reported an in-progress total wider than this product renders; \
2497                     it is clamped and reported as a floor rather than as a count"
2498                );
2499                Self {
2500                    count: u32::MAX,
2501                    exact: false,
2502                }
2503            }
2504        }
2505    }
2506}
2507
2508/// One page of a paginated GitHub collection.
2509///
2510/// A trait rather than two near-identical loops, because the loop is where the
2511/// mandatory-pagination requirement actually lives: one implementation of
2512/// "follow `rel=next` until it stops, and stop at the ceiling" cannot disagree
2513/// with itself.
2514trait WirePage: DeserializeOwned {
2515    type Item;
2516    /// Named in the ceiling warning, so the log says which collection wedged.
2517    const WHAT: &'static str;
2518    fn reported_total(&self) -> Option<u64>;
2519    fn into_items(self) -> Vec<Self::Item>;
2520}
2521
2522#[derive(Debug, Deserialize)]
2523struct RunnersPage {
2524    total_count: Option<u64>,
2525    #[serde(default)]
2526    runners: Vec<RawRunner>,
2527}
2528
2529impl WirePage for RunnersPage {
2530    type Item = RawRunner;
2531    const WHAT: &'static str = "runners";
2532
2533    fn reported_total(&self) -> Option<u64> {
2534        self.total_count
2535    }
2536
2537    fn into_items(self) -> Vec<Self::Item> {
2538        self.runners
2539    }
2540}
2541
2542#[derive(Debug, Deserialize)]
2543struct RawRunner {
2544    id: u64,
2545    #[serde(default)]
2546    name: String,
2547    #[serde(default)]
2548    os: String,
2549    status: String,
2550    busy: bool,
2551    /// Optional in the wire schema and kept optional here. See
2552    /// [`Runner::ephemeral`].
2553    ephemeral: Option<bool>,
2554    #[serde(default)]
2555    labels: Vec<RawLabel>,
2556}
2557
2558#[derive(Debug, Deserialize)]
2559struct RawLabel {
2560    name: String,
2561}
2562
2563impl From<RawRunner> for Runner {
2564    fn from(raw: RawRunner) -> Self {
2565        Self {
2566            id: raw.id,
2567            name: raw.name,
2568            os: raw.os,
2569            status: RunnerStatus::from_wire(&raw.status),
2570            busy: raw.busy,
2571            ephemeral: raw.ephemeral,
2572            labels: raw.labels.into_iter().map(|label| label.name).collect(),
2573        }
2574    }
2575}
2576
2577#[derive(Debug, Deserialize)]
2578struct RunsPage {
2579    total_count: Option<u64>,
2580    #[serde(default)]
2581    workflow_runs: Vec<serde::de::IgnoredAny>,
2582}
2583
2584#[derive(Debug, Deserialize)]
2585struct RawDownload {
2586    #[serde(default)]
2587    os: String,
2588    #[serde(default)]
2589    architecture: String,
2590    #[serde(default)]
2591    download_url: String,
2592    #[serde(default)]
2593    filename: String,
2594    /// **No `#[serde(default)]`, on purpose.** `Option` already makes an absent
2595    /// field `None`; adding a default here would be harmless today and is
2596    /// exactly the edit that would later be "simplified" into
2597    /// `#[serde(default)] sha256_checksum: String`, turning an absent digest
2598    /// into an empty one and silently disarming `e2`'s fail-closed rule.
2599    sha256_checksum: Option<String>,
2600}
2601
2602impl From<RawDownload> for RunnerDownload {
2603    fn from(raw: RawDownload) -> Self {
2604        Self {
2605            os: raw.os,
2606            architecture: raw.architecture,
2607            download_url: raw.download_url,
2608            filename: raw.filename,
2609            sha256_checksum: raw.sha256_checksum,
2610        }
2611    }
2612}
2613
2614// The unit tests below are inline rather than in a `src/rest/tests.rs`, and
2615// that is a constraint rather than a preference. `lib.rs`'s
2616// `the_confidential_credential_scan_covers_every_source_file` walks `src/`
2617// recursively and requires every `.rs` file under it to appear in
2618// `CRATE_SOURCES` — a list that lives in `lib.rs`, which `c2` owns. A second
2619// file in this directory would fail that pin, and the only way to fix it would
2620// be to edit another task's file.
2621//
2622// They are also unit tests rather than an integration test under `tests/`,
2623// because they use `crate::testing`, which is `pub(crate)`. An integration test
2624// cannot reach it, and it cannot use `runner-manager-testkit` in its place:
2625// `testkit` depends on this crate, so a unit test that linked it would compile
2626// a second instance of this library whose types would not unify with these.
2627#[cfg(test)]
2628mod tests {
2629    use super::*;
2630    use crate::testing::{FIXTURE_TOKEN, Script, TestClock};
2631    use crate::{Endpoints, UserAccessToken};
2632    use secrecy::SecretString;
2633    use serde_json::{Value, json};
2634    use wiremock::{
2635        Mock, MockServer, ResponseTemplate,
2636        matchers::{method, path, query_param},
2637    };
2638
2639    // -- fixtures -----------------------------------------------------------
2640
2641    fn repo() -> OwnerRepo {
2642        OwnerRepo::parse("octo/dashboard").expect("a valid owner/repo")
2643    }
2644
2645    fn other_repo() -> OwnerRepo {
2646        OwnerRepo::parse("octo/api").expect("a valid owner/repo")
2647    }
2648
2649    fn third_repo() -> OwnerRepo {
2650        OwnerRepo::parse("octo/docs").expect("a valid owner/repo")
2651    }
2652
2653    fn repo_target() -> ScaleTarget {
2654        ScaleTarget::Repository(repo())
2655    }
2656
2657    fn org_target() -> ScaleTarget {
2658        ScaleTarget::organization("octo-org").expect("a valid organization login")
2659    }
2660
2661    const REPO_RUNNERS: &str = "/repos/octo/dashboard/actions/runners";
2662    const ORG_RUNNERS: &str = "/orgs/octo-org/actions/runners";
2663    const REPO_RUNS: &str = "/repos/octo/dashboard/actions/runs";
2664
2665    fn runners_path(target: &ScaleTarget) -> &'static str {
2666        match target {
2667            ScaleTarget::Repository(_) => REPO_RUNNERS,
2668            ScaleTarget::Organization(_) => ORG_RUNNERS,
2669        }
2670    }
2671
2672    fn gateway(server: &MockServer, clock: Arc<TestClock>) -> RestInventory {
2673        let client = AuthenticatedClient::new(
2674            Endpoints::for_test_server(&server.uri()).expect("a valid test base"),
2675            UserAccessToken::new(SecretString::from(FIXTURE_TOKEN)),
2676            clock.clone(),
2677        )
2678        .expect("the HTTP client builds");
2679        RestInventory::new(Arc::new(client), clock)
2680    }
2681
2682    /// A page of runners with ids in `ids`, all online and idle.
2683    fn runner_page(ids: std::ops::Range<u64>, total: u64) -> Value {
2684        let runners: Vec<Value> = ids
2685            .map(|id| {
2686                json!({
2687                    "id": id,
2688                    "name": format!("runner-{id:04}"),
2689                    "os": "win",
2690                    "status": "online",
2691                    "busy": false,
2692                    "ephemeral": true,
2693                    "labels": [{ "id": 1, "name": "rm-home-win-x64", "type": "read-only" }]
2694                })
2695            })
2696            .collect();
2697        json!({ "total_count": total, "runners": runners })
2698    }
2699
2700    fn link_next(url: &str) -> String {
2701        format!("<{url}>; rel=\"next\"")
2702    }
2703
2704    async fn requests_seen(server: &MockServer) -> usize {
2705        server
2706            .received_requests()
2707            .await
2708            .expect("the mock server records requests")
2709            .len()
2710    }
2711
2712    // -- pagination ---------------------------------------------------------
2713
2714    /// The Definition of Done's first item, at both scopes under one body.
2715    ///
2716    /// `04-subsystem-contracts.md` forbids treating a first page as a complete
2717    /// inventory, and the reason it forbids it rather than merely discouraging
2718    /// it is that the failure is silent: 250 runners reported as 100 renders as
2719    /// a smaller fleet, not as an error. So the assertion is on the *whole*
2720    /// collection, and on the page count that proves three requests were spent
2721    /// getting it.
2722    ///
2723    /// One body over both targets, the way the domain's own
2724    /// `repository_and_organization_targets_are_equivalent` runs one body over
2725    /// both variants: the scopes differ in the endpoint and in nothing else, and
2726    /// a second copy of this test is where that stops being true.
2727    #[tokio::test]
2728    async fn a_multi_page_runner_inventory_returns_every_runner_at_both_scopes() {
2729        for target in [repo_target(), org_target()] {
2730            let server = MockServer::start().await;
2731            let first = runners_path(&target);
2732            let page_two = format!("{}/page/2", server.uri());
2733            let page_three = format!("{}/page/3", server.uri());
2734
2735            Mock::given(method("GET"))
2736                .and(path(first))
2737                .respond_with(
2738                    ResponseTemplate::new(200)
2739                        .insert_header("link", link_next(&page_two).as_str())
2740                        .set_body_json(runner_page(1..101, 250)),
2741                )
2742                .expect(1)
2743                .mount(&server)
2744                .await;
2745            Mock::given(method("GET"))
2746                .and(path("/page/2"))
2747                .respond_with(
2748                    ResponseTemplate::new(200)
2749                        .insert_header("link", link_next(&page_three).as_str())
2750                        .set_body_json(runner_page(101..201, 250)),
2751                )
2752                .expect(1)
2753                .mount(&server)
2754                .await;
2755            Mock::given(method("GET"))
2756                .and(path("/page/3"))
2757                .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(201..251, 250)))
2758                .expect(1)
2759                .mount(&server)
2760                .await;
2761
2762            let gateway = gateway(&server, Arc::new(TestClock::default()));
2763            let inventory = gateway
2764                .list_runners(&target, &CancelToken::new())
2765                .await
2766                .expect("three pages are readable");
2767
2768            assert_eq!(
2769                inventory.len(),
2770                250,
2771                "{target}: a first page is not a complete inventory"
2772            );
2773            assert_eq!(inventory.pages(), 3, "{target}");
2774            assert_eq!(inventory.reported_total(), Some(250), "{target}");
2775            assert_eq!(
2776                inventory.missing(),
2777                None,
2778                "{target}: pagination collected everything GitHub said existed"
2779            );
2780            assert!(!inventory.truncated(), "{target}");
2781            assert_eq!(inventory.runners()[0].id, 1, "{target}");
2782            assert_eq!(inventory.runners()[249].id, 250, "{target}");
2783            assert_eq!(gateway.requests_issued(), 3, "{target}");
2784        }
2785    }
2786
2787    /// The `Link` header case that silently stopped pagination at page one until
2788    /// a review caught it, exercised through *this* module's loop rather than
2789    /// only through `c2`'s parser.
2790    ///
2791    /// A runner query carries `labels=self-hosted,windows` routinely, so the
2792    /// next-page URL contains a comma — and a parser that splits the header on
2793    /// `,` first tears that URL in half and loses the relation. That this
2794    /// module reuses [`crate::ApiResponse::next_page`] rather than writing a
2795    /// second reader is what makes it immune; this test is what says so, because
2796    /// "we reuse it" is a claim about code that a later edit can quietly falsify.
2797    #[tokio::test]
2798    async fn a_next_page_url_containing_a_comma_does_not_truncate_the_inventory() {
2799        let server = MockServer::start().await;
2800        let page_two = format!("{}/page/2?labels=self-hosted,windows", server.uri());
2801
2802        Mock::given(method("GET"))
2803            .and(path(REPO_RUNNERS))
2804            .respond_with(
2805                ResponseTemplate::new(200)
2806                    .insert_header("link", link_next(&page_two).as_str())
2807                    .set_body_json(runner_page(1..101, 150)),
2808            )
2809            .mount(&server)
2810            .await;
2811        Mock::given(method("GET"))
2812            .and(path("/page/2"))
2813            .and(query_param("labels", "self-hosted,windows"))
2814            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..151, 150)))
2815            .expect(1)
2816            .mount(&server)
2817            .await;
2818
2819        let gateway = gateway(&server, Arc::new(TestClock::default()));
2820        let inventory = gateway
2821            .list_runners(&repo_target(), &CancelToken::new())
2822            .await
2823            .expect("both pages are readable");
2824
2825        assert_eq!(inventory.len(), 150, "the comma ended pagination at page 1");
2826        assert_eq!(inventory.pages(), 2);
2827    }
2828
2829    /// The deletion goes to the one runner asked for, under the scope's own
2830    /// path, and a registration GitHub has already dropped is a success.
2831    #[tokio::test]
2832    async fn removing_a_runner_deletes_that_id_and_treats_an_absent_one_as_done() {
2833        let server = MockServer::start().await;
2834        Mock::given(method("DELETE"))
2835            .and(path(format!("{REPO_RUNNERS}/73")))
2836            .respond_with(ResponseTemplate::new(204))
2837            .expect(1)
2838            .mount(&server)
2839            .await;
2840        Mock::given(method("DELETE"))
2841            .and(path(format!("{REPO_RUNNERS}/99")))
2842            .respond_with(ResponseTemplate::new(404).set_body_json(serde_json::json!({
2843                "message": "Not Found"
2844            })))
2845            .expect(1)
2846            .mount(&server)
2847            .await;
2848
2849        let gateway = gateway(&server, Arc::new(TestClock::default()));
2850        gateway
2851            .remove_runner(&repo_target(), 73, &CancelToken::new())
2852            .await
2853            .expect("a registration this agent owns is deletable");
2854        gateway
2855            .remove_runner(&repo_target(), 99, &CancelToken::new())
2856            .await
2857            .expect(
2858                "already gone is the postcondition asked for; failing here would strand every \
2859                 attempt GitHub retired on its own",
2860            );
2861    }
2862
2863    /// An organization target deletes under `/orgs`, not `/repos`.
2864    #[tokio::test]
2865    async fn removing_an_organization_runner_uses_the_organization_path() {
2866        let server = MockServer::start().await;
2867        Mock::given(method("DELETE"))
2868            .and(path(format!("{ORG_RUNNERS}/12")))
2869            .respond_with(ResponseTemplate::new(204))
2870            .expect(1)
2871            .mount(&server)
2872            .await;
2873
2874        let gateway = gateway(&server, Arc::new(TestClock::default()));
2875        gateway
2876            .remove_runner(&org_target(), 12, &CancelToken::new())
2877            .await
2878            .expect("the organization scope deletes under its own path");
2879    }
2880
2881    /// A collection shorter than GitHub's own `total_count` is reported as
2882    /// short, rather than as the inventory.
2883    #[tokio::test]
2884    async fn an_inventory_shorter_than_the_reported_total_says_how_short() {
2885        let server = MockServer::start().await;
2886        Mock::given(method("GET"))
2887            .and(path(REPO_RUNNERS))
2888            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(1..11, 40)))
2889            .mount(&server)
2890            .await;
2891
2892        let gateway = gateway(&server, Arc::new(TestClock::default()));
2893        let inventory = gateway
2894            .list_runners(&repo_target(), &CancelToken::new())
2895            .await
2896            .expect("one page is readable");
2897
2898        assert_eq!(inventory.len(), 10);
2899        assert_eq!(
2900            inventory.missing(),
2901            Some(30),
2902            "GitHub said 40 and pagination found 10; a caller has to be able to see that"
2903        );
2904    }
2905
2906    /// A `rel="next"` that never ends is stopped at the ceiling instead of
2907    /// wedging the agent's reconciliation loop.
2908    #[tokio::test]
2909    async fn a_self_referential_next_link_stops_at_the_page_ceiling() {
2910        let server = MockServer::start().await;
2911        let itself = format!("{}{}", server.uri(), REPO_RUNNERS);
2912        Mock::given(method("GET"))
2913            .and(path(REPO_RUNNERS))
2914            .respond_with(
2915                ResponseTemplate::new(200)
2916                    .insert_header("link", link_next(&itself).as_str())
2917                    .set_body_json(runner_page(1..2, 1)),
2918            )
2919            .mount(&server)
2920            .await;
2921
2922        let gateway = gateway(&server, Arc::new(TestClock::default()));
2923        let inventory = gateway
2924            .list_runners(&repo_target(), &CancelToken::new())
2925            .await
2926            .expect("the walk terminates");
2927
2928        assert_eq!(inventory.pages(), MAX_PAGES);
2929        assert!(
2930            inventory.truncated(),
2931            "a truncated walk must say so, or it reads as a complete inventory"
2932        );
2933        assert_eq!(gateway.requests_issued() as usize, MAX_PAGES);
2934    }
2935
2936    // -- in-progress workflow counts ----------------------------------------
2937
2938    fn runs_body(total: Option<u64>, listed: usize) -> Value {
2939        let runs: Vec<Value> = (0..listed)
2940            .map(|i| json!({ "id": i + 1, "status": "in_progress" }))
2941            .collect();
2942        match total {
2943            Some(total) => json!({ "total_count": total, "workflow_runs": runs }),
2944            None => json!({ "workflow_runs": runs }),
2945        }
2946    }
2947
2948    /// One repository's whole in-progress set, on a single page.
2949    ///
2950    /// `total_count` and the listed runs **agree**, because that is what GitHub
2951    /// sends when there is no `rel="next"` — and it is now the invariant
2952    /// `repository_in_progress` checks. These fixtures previously declared a
2953    /// `total_count` over an empty `workflow_runs`, which was not a smaller
2954    /// fixture but a fixture of a response GitHub does not send; every one of
2955    /// them tripped the new check the moment it existed, which is the check
2956    /// earning its place before it ever reaches the live API.
2957    fn mount_runs(repository: &OwnerRepo, total: u64) -> Mock {
2958        assert!(
2959            total <= u64::from(PER_PAGE),
2960            "a single-page fixture cannot hold {total} runs; a larger one needs a \
2961             `Link: rel=next` and a second page, or it is claiming a total the page \
2962             does not support"
2963        );
2964        let listed = usize::try_from(total).expect("a fixture total fits a usize");
2965        Mock::given(method("GET"))
2966            .and(path(format!(
2967                "/repos/{}/{}/actions/runs",
2968                repository.owner(),
2969                repository.repo()
2970            )))
2971            .and(query_param("status", "in_progress"))
2972            .respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(total), listed)))
2973    }
2974
2975    /// A repository target counts its own runs, in one request, from GitHub's
2976    /// own `total_count`.
2977    #[tokio::test]
2978    async fn a_repository_activity_count_is_one_request_and_reads_the_reported_total() {
2979        let server = MockServer::start().await;
2980        mount_runs(&repo(), 7).expect(1).mount(&server).await;
2981
2982        let gateway = gateway(&server, Arc::new(TestClock::default()));
2983        let scope = ActivityScope::repository(repo());
2984        let activity = gateway
2985            .in_progress_activity(&scope, &CancelToken::new())
2986            .await
2987            .expect("the count is readable");
2988
2989        assert_eq!(activity.total(), 7);
2990        assert_eq!(activity.for_repository(&repo()), Some(7));
2991        assert!(activity.is_complete());
2992        assert_eq!(
2993            gateway.requests_issued(),
2994            1,
2995            "reading `total_count` is what keeps this at the one request the budget \
2996             table projects"
2997        );
2998    }
2999
3000    /// An organization target aggregates across the repositories the App is
3001    /// installed on — because workflow runs are a per-repository resource and
3002    /// GitHub publishes no organization-wide runs endpoint.
3003    #[tokio::test]
3004    async fn an_organization_activity_count_aggregates_across_installed_repositories() {
3005        let server = MockServer::start().await;
3006        mount_runs(&repo(), 4).mount(&server).await;
3007        mount_runs(&other_repo(), 9).mount(&server).await;
3008        mount_runs(&third_repo(), 0).mount(&server).await;
3009
3010        let gateway = gateway(&server, Arc::new(TestClock::default()));
3011        let scope = ActivityScope::organization(
3012            Org::new("octo-org").expect("a valid organization login"),
3013            [repo(), other_repo(), third_repo()],
3014        );
3015        let activity = gateway
3016            .in_progress_activity(&scope, &CancelToken::new())
3017            .await
3018            .expect("every repository answers");
3019
3020        assert_eq!(activity.total(), 13);
3021        assert_eq!(activity.for_repository(&repo()), Some(4));
3022        assert_eq!(activity.for_repository(&other_repo()), Some(9));
3023        assert_eq!(
3024            activity.for_repository(&third_repo()),
3025            Some(0),
3026            "a repository with no in-progress runs is a zero, not an absence"
3027        );
3028        assert_eq!(
3029            gateway.requests_issued(),
3030            3,
3031            "one request per installed repository: this is the cost the budget model \
3032             projects and the reason an organization is not a flat per-target constant"
3033        );
3034    }
3035
3036    /// The Definition of Done's "they are different numbers with different
3037    /// meanings", asserted on one snapshot where they genuinely differ.
3038    #[tokio::test]
3039    async fn the_in_progress_count_and_the_busy_runner_count_are_distinct() {
3040        let server = MockServer::start().await;
3041        let runners = json!({
3042            "total_count": 5,
3043            "runners": (1..=5).map(|id| json!({
3044                "id": id,
3045                "name": format!("runner-{id}"),
3046                "os": "win",
3047                "status": "online",
3048                // Three of five are executing something.
3049                "busy": id <= 3,
3050                "ephemeral": true,
3051                "labels": []
3052            })).collect::<Vec<_>>()
3053        });
3054        Mock::given(method("GET"))
3055            .and(path(REPO_RUNNERS))
3056            .respond_with(ResponseTemplate::new(200).set_body_json(runners))
3057            .mount(&server)
3058            .await;
3059        mount_runs(&repo(), 7).mount(&server).await;
3060
3061        let gateway = gateway(&server, Arc::new(TestClock::default()));
3062        let scope = ActivityScope::repository(repo());
3063        let snapshot = gateway
3064            .snapshot(&scope, &CancelToken::new())
3065            .await
3066            .expect("both read models are readable");
3067
3068        assert_eq!(snapshot.runners.len(), 5);
3069        assert_eq!(snapshot.runners.busy_count(), 3);
3070        assert_eq!(snapshot.runners.online_count(), 5);
3071        assert_eq!(snapshot.activity.total(), 7);
3072        assert_ne!(
3073            u32::try_from(snapshot.runners.busy_count()).unwrap(),
3074            snapshot.activity.total(),
3075            "a workflow run is not a busy runner; `g2` renders them as separate \
3076             aggregates and cannot do that if this layer conflates them"
3077        );
3078        assert_eq!(snapshot.target, repo_target());
3079        assert_eq!(snapshot.observed_at, TestClock::default().now());
3080    }
3081
3082    /// A response with no `total_count` is counted rather than guessed at.
3083    #[tokio::test]
3084    async fn an_activity_count_without_a_reported_total_counts_the_runs_instead() {
3085        let server = MockServer::start().await;
3086        Mock::given(method("GET"))
3087            .and(path(REPO_RUNS))
3088            .respond_with(ResponseTemplate::new(200).set_body_json(runs_body(None, 4)))
3089            .mount(&server)
3090            .await;
3091
3092        let gateway = gateway(&server, Arc::new(TestClock::default()));
3093        let activity = gateway
3094            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3095            .await
3096            .expect("the runs are countable");
3097
3098        assert_eq!(
3099            activity.total(),
3100            4,
3101            "a missing `total_count` must not read as an idle repository"
3102        );
3103        assert!(
3104            activity.is_complete(),
3105            "a fallback that reached the end of the pages counted everything"
3106        );
3107        assert!(activity.truncated().is_empty());
3108    }
3109
3110    /// An **endless** no-`total_count` page sequence at `first_path`: every page
3111    /// is full and every page offers another, forever.
3112    ///
3113    /// Deliberately endless rather than a chain of `n` pages. A finite fixture
3114    /// makes an unbounded walk fail by running off the end into a `404`, which
3115    /// is a fixture artefact — the test would then be red for the wrong reason
3116    /// and would stay red if the bound were changed to any other finite number.
3117    /// Against this one, the number of requests the walk spends *is* the
3118    /// measurement, and an unbounded walk answers with `MAX_PAGES` instead of
3119    /// the budget. It is also the shape `MAX_PAGES` exists for: a `rel="next"`
3120    /// that never ends.
3121    async fn mount_endless_runs_pages(server: &MockServer, first_path: &str, loop_path: &str) {
3122        let body = || runs_body(None, usize::try_from(PER_PAGE).expect("PER_PAGE fits"));
3123        let onward = link_next(&format!("{}{loop_path}", server.uri()));
3124        for at in [first_path, loop_path] {
3125            Mock::given(method("GET"))
3126                .and(path(at.to_owned()))
3127                .respond_with(
3128                    ResponseTemplate::new(200)
3129                        .insert_header("link", onward.as_str())
3130                        .set_body_json(body()),
3131                )
3132                .mount(server)
3133                .await;
3134        }
3135    }
3136
3137    /// The fallback walk is bounded by the **budget**, not by the runaway
3138    /// ceiling.
3139    ///
3140    /// `MAX_PAGES` is the number that keeps a `Link` cycle from looping forever;
3141    /// it is not a number anything budgeted for. Charging 100 requests to a line
3142    /// item the projection prices at one — 6,000/hour against a 5,000 ceiling,
3143    /// for a single repository's count — is what would make `f2`'s `add`
3144    /// refusals a fiction, since it computes them from that projection.
3145    #[tokio::test]
3146    async fn the_activity_fallback_stops_at_the_budget_not_at_the_runaway_ceiling() {
3147        let server = MockServer::start().await;
3148        mount_endless_runs_pages(&server, REPO_RUNS, "/runs/onward").await;
3149
3150        let gateway = gateway(&server, Arc::new(TestClock::default()));
3151        let activity = gateway
3152            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3153            .await
3154            .expect("the walk ends at the budget rather than erroring");
3155
3156        assert_eq!(
3157            gateway.requests_issued() as usize,
3158            MAX_ACTIVITY_FALLBACK_PAGES,
3159            "the walk spends its page budget and not one request more; `MAX_PAGES` \
3160             here would be {MAX_PAGES} requests for one repository's count, per refresh"
3161        );
3162        assert_eq!(
3163            activity.total(),
3164            PER_PAGE * u32::try_from(MAX_ACTIVITY_FALLBACK_PAGES).unwrap(),
3165            "what it did count, it counted"
3166        );
3167    }
3168
3169    /// A count the page budget cut short is a **floor**, and says so.
3170    ///
3171    /// `04-subsystem-contracts.md` forbids treating a first page as a complete
3172    /// inventory; a count clipped at page four is the same defect on the other
3173    /// read model. `RunnerInventory` already honours this with `truncated()` and
3174    /// `missing()` — returning the partial activity count as though it were the
3175    /// answer left `g2` rendering a number with no way to know.
3176    #[tokio::test]
3177    async fn a_count_the_page_budget_cut_short_is_reported_as_a_floor_not_as_a_total() {
3178        let server = MockServer::start().await;
3179        mount_endless_runs_pages(&server, REPO_RUNS, "/runs/onward").await;
3180
3181        let gateway = gateway(&server, Arc::new(TestClock::default()));
3182        let activity = gateway
3183            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3184            .await
3185            .expect("a truncated count is an answer, not a failure");
3186
3187        assert!(
3188            activity.is_truncated(&repo()),
3189            "the repository whose walk was cut short has to be named"
3190        );
3191        assert_eq!(activity.truncated().len(), 1);
3192        assert!(
3193            !activity.is_complete(),
3194            "a partial answer is usable only when it says it is partial -- and a caller \
3195             asking the one obvious question must hear about truncation, not only about \
3196             repositories that failed outright"
3197        );
3198        assert!(
3199            activity.unavailable().is_empty(),
3200            "truncated is not unavailable: this repository answered, the answer is a floor"
3201        );
3202    }
3203
3204    /// Truncation of one repository makes an **organization's** total a floor
3205    /// too, and the aggregate carries which repository did it.
3206    #[tokio::test]
3207    async fn one_truncated_repository_makes_the_whole_aggregate_a_floor() {
3208        let server = MockServer::start().await;
3209        // `octo/dashboard` answers exactly, in one request. `octo/api` sends no
3210        // `total_count` and never stops offering pages.
3211        mount_runs(&repo(), 4).mount(&server).await;
3212        mount_endless_runs_pages(&server, "/repos/octo/api/actions/runs", "/api-runs/onward").await;
3213
3214        let gateway = gateway(&server, Arc::new(TestClock::default()));
3215        let scope = ActivityScope::organization(
3216            Org::new("octo-org").expect("a valid organization login"),
3217            [repo(), other_repo()],
3218        );
3219        let activity = gateway
3220            .in_progress_activity(&scope, &CancelToken::new())
3221            .await
3222            .expect("the aggregate completes");
3223
3224        assert!(activity.is_truncated(&other_repo()));
3225        assert!(
3226            !activity.is_truncated(&repo()),
3227            "the repository that answered exactly is not tarred with it"
3228        );
3229        assert!(
3230            !activity.is_complete(),
3231            "one floor in the sum makes the sum a floor"
3232        );
3233        assert_eq!(
3234            activity.total(),
3235            4 + PER_PAGE * u32::try_from(MAX_ACTIVITY_FALLBACK_PAGES).unwrap()
3236        );
3237    }
3238
3239    /// The assumption this layer's one-request activity count rests on, checked
3240    /// against the wire at no extra request cost.
3241    ///
3242    /// Reading `total_count` off a *filtered* query assumes it counts the
3243    /// filtered set. When GitHub sends no `rel="next"`, the whole filtered set
3244    /// is on the page in hand, so `total_count` must equal
3245    /// `workflow_runs.len()`. An unfiltered total would show up here as exactly
3246    /// the contradiction below — 5,000 lifetime runs reported over the 3 that
3247    /// are in progress — and it is caught on the first response rather than by
3248    /// an operator noticing a wrong dashboard number weeks later.
3249    ///
3250    /// The same envelope reaches `c4`'s `clamp()` on `status=queued`, which is
3251    /// why this is worth a check rather than a comment.
3252    ///
3253    /// # Debug-only, because the tripwire is
3254    ///
3255    /// A contradiction here means a wire contract is not what this layer read it
3256    /// to be, which is a thing to find in development and not a reason to panic
3257    /// a shipped dashboard — and there is a legitimate way to see it in the
3258    /// wild: a run finishing between GitHub computing `total_count` and
3259    /// serialising the page. So the loud half is a `debug_assert!` and the
3260    /// always-on half is the `warn!`, which
3261    /// `a_total_count_that_disagrees_with_its_only_page_still_answers_in_release`
3262    /// covers.
3263    #[cfg(debug_assertions)]
3264    #[tokio::test]
3265    #[should_panic(expected = "is not the filtered count")]
3266    async fn a_total_count_that_disagrees_with_its_only_page_is_caught() {
3267        let server = MockServer::start().await;
3268        Mock::given(method("GET"))
3269            .and(path(REPO_RUNS))
3270            // No `Link`, so this page is the whole filtered set -- and yet the
3271            // total claims 5,000. One of the two is not what it says it is.
3272            .respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(5_000), 3)))
3273            .mount(&server)
3274            .await;
3275
3276        let gateway = gateway(&server, Arc::new(TestClock::default()));
3277        let _ = gateway
3278            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3279            .await;
3280    }
3281
3282    /// The release half of the same case: the `debug_assert!` is compiled out,
3283    /// so the contradiction is a `warn!` and the refresh keeps working.
3284    ///
3285    /// Asserted so that "it panics in debug" is never quietly also "it panics in
3286    /// production", and so that the check cannot start costing a second request
3287    /// to resolve the disagreement it noticed.
3288    #[cfg(not(debug_assertions))]
3289    #[tokio::test]
3290    async fn a_total_count_that_disagrees_with_its_only_page_still_answers_in_release() {
3291        let server = MockServer::start().await;
3292        Mock::given(method("GET"))
3293            .and(path(REPO_RUNS))
3294            .respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(5_000), 3)))
3295            .mount(&server)
3296            .await;
3297
3298        let gateway = gateway(&server, Arc::new(TestClock::default()));
3299        let activity = gateway
3300            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3301            .await
3302            .expect("a suspect total is still an answer in release");
3303
3304        assert_eq!(activity.total(), 5_000);
3305        assert_eq!(
3306            gateway.requests_issued(),
3307            1,
3308            "noticing the disagreement must stay free"
3309        );
3310    }
3311
3312    /// The narrowing: the race this check's **own documentation** calls
3313    /// legitimate must not panic the build most likely to meet it.
3314    ///
3315    /// A run finishing between GitHub computing `total_count` and serialising
3316    /// the page leaves `total` a handful over `listed`. That is benign, it is
3317    /// real, and a debug build pointed at live GitHub during `c4`'s development
3318    /// is precisely where it shows up. While the assert read `total == listed`
3319    /// it fired on this too — so the tripwire's *only* observable behaviour in
3320    /// development would have been a false positive, which is how a real check
3321    /// gets deleted by the next reader.
3322    ///
3323    /// The gap is a **literal**, deliberately. A fixture derived from
3324    /// [`MAX_BENIGN_TOTAL_COUNT_SKEW`] moves with the constant and stays green
3325    /// even at zero — where the check is `total == listed` again and the race
3326    /// panics — so it would assert nothing about the threshold at all. That end
3327    /// is held by a compile-time assertion next to the constant; this test holds
3328    /// the case an operator actually meets: one run finished between the total
3329    /// being computed and the page being serialised.
3330    #[cfg(debug_assertions)]
3331    #[tokio::test]
3332    async fn a_total_count_one_over_its_only_page_is_the_documented_race_not_a_panic() {
3333        let listed = 3_usize;
3334        let total = 4_u64;
3335
3336        let server = MockServer::start().await;
3337        Mock::given(method("GET"))
3338            .and(path(REPO_RUNS))
3339            .respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(total), listed)))
3340            .mount(&server)
3341            .await;
3342
3343        let gateway = gateway(&server, Arc::new(TestClock::default()));
3344        let activity = gateway
3345            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3346            .await
3347            .expect("the documented race is an answer, not a panic");
3348
3349        assert_eq!(
3350            activity.total(),
3351            u32::try_from(total).expect("the fixture total fits"),
3352            "the reported total is still what the layer reads"
3353        );
3354        assert_eq!(
3355            gateway.requests_issued(),
3356            1,
3357            "and noticing the skew must stay free"
3358        );
3359    }
3360
3361    /// The check is scoped to the page that *is* the whole set. A total larger
3362    /// than one page is the ordinary case and must stay silent.
3363    #[tokio::test]
3364    async fn a_total_count_larger_than_a_page_is_not_a_contradiction() {
3365        let server = MockServer::start().await;
3366        Mock::given(method("GET"))
3367            .and(path(REPO_RUNS))
3368            .respond_with(
3369                ResponseTemplate::new(200)
3370                    .insert_header(
3371                        "link",
3372                        link_next(&format!("{}/page/2", server.uri())).as_str(),
3373                    )
3374                    .set_body_json(runs_body(
3375                        Some(250),
3376                        usize::try_from(PER_PAGE).expect("PER_PAGE fits"),
3377                    )),
3378            )
3379            .mount(&server)
3380            .await;
3381
3382        let gateway = gateway(&server, Arc::new(TestClock::default()));
3383        let activity = gateway
3384            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3385            .await
3386            .expect("a paginated total is exactly what `total_count` is for");
3387
3388        assert_eq!(activity.total(), 250);
3389        assert!(activity.is_complete());
3390        assert_eq!(
3391            gateway.requests_issued(),
3392            1,
3393            "reading the reported total is what keeps a 250-run repository at one \
3394             request; the check must not have provoked a second"
3395        );
3396    }
3397
3398    /// One unreadable repository does not take down an organization's whole
3399    /// aggregate, and does not silently vanish from it either.
3400    #[tokio::test]
3401    async fn a_repository_that_cannot_be_counted_is_reported_as_unavailable_not_as_zero() {
3402        let server = MockServer::start().await;
3403        mount_runs(&repo(), 6).mount(&server).await;
3404        Mock::given(method("GET"))
3405            .and(path("/repos/octo/api/actions/runs"))
3406            .respond_with(
3407                ResponseTemplate::new(404).set_body_json(json!({ "message": "Not Found" })),
3408            )
3409            .mount(&server)
3410            .await;
3411
3412        let gateway = gateway(&server, Arc::new(TestClock::default()));
3413        let scope = ActivityScope::organization(
3414            Org::new("octo-org").expect("a valid organization login"),
3415            [repo(), other_repo()],
3416        );
3417        let activity = gateway
3418            .in_progress_activity(&scope, &CancelToken::new())
3419            .await
3420            .expect("one unreadable repository is not fatal to the aggregate");
3421
3422        assert_eq!(activity.total(), 6);
3423        assert_eq!(activity.for_repository(&other_repo()), None);
3424        assert!(
3425            !activity.is_complete(),
3426            "a partial total is usable only when it says it is partial"
3427        );
3428        assert_eq!(activity.unavailable().len(), 1);
3429        assert_eq!(activity.unavailable()[0].repository, other_repo());
3430    }
3431
3432    /// The step-over is for an aggregate. A **repository** target's failure
3433    /// propagates, because turning the only repository in scope into a zero
3434    /// renders a permissions failure as "0 in progress" for anything that reads
3435    /// `total()` — which is the obvious thing to read.
3436    #[tokio::test]
3437    async fn a_repository_targets_activity_failure_propagates_rather_than_becoming_zero() {
3438        let server = MockServer::start().await;
3439        Mock::given(method("GET"))
3440            .and(path(REPO_RUNS))
3441            .respond_with(
3442                ResponseTemplate::new(403)
3443                    .set_body_json(json!({ "message": "Resource not accessible by integration" })),
3444            )
3445            .mount(&server)
3446            .await;
3447
3448        let gateway = gateway(&server, Arc::new(TestClock::default()));
3449        let error = gateway
3450            .in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
3451            .await
3452            .expect_err("a scope of one has no partial answer to give");
3453
3454        assert!(matches!(
3455            RefreshState::from_error(&error),
3456            RefreshState::Forbidden { .. }
3457        ));
3458    }
3459
3460    /// A rate limit hit part-way through an aggregate aborts it, because
3461    /// stepping over it would report a total that is short by an unknown amount
3462    /// while looking complete.
3463    #[tokio::test]
3464    async fn a_rate_limit_during_an_aggregate_aborts_it_rather_than_under_reporting() {
3465        let server = MockServer::start().await;
3466        mount_runs(&repo(), 6).mount(&server).await;
3467        Mock::given(method("GET"))
3468            .and(path("/repos/octo/api/actions/runs"))
3469            .respond_with(
3470                ResponseTemplate::new(429)
3471                    .insert_header("retry-after", "30")
3472                    .set_body_json(
3473                        json!({ "message": "You have exceeded a secondary rate limit" }),
3474                    ),
3475            )
3476            .mount(&server)
3477            .await;
3478
3479        let gateway = gateway(&server, Arc::new(TestClock::default()));
3480        let scope = ActivityScope::organization(
3481            Org::new("octo-org").expect("a valid organization login"),
3482            [repo(), other_repo(), third_repo()],
3483        );
3484        let error = gateway
3485            .in_progress_activity(&scope, &CancelToken::new())
3486            .await
3487            .expect_err("a rate limit is systemic, not a fact about one repository");
3488
3489        assert!(error.is_rate_limited(), "{error}");
3490    }
3491
3492    // -- rate limiting ------------------------------------------------------
3493
3494    /// The Definition of Done's `retry-after`, obeyed in the only way that costs
3495    /// the shared budget nothing: by issuing no request at all.
3496    #[tokio::test]
3497    async fn retry_after_is_obeyed_by_issuing_no_request_until_it_elapses() {
3498        let server = MockServer::start().await;
3499        Mock::given(method("GET"))
3500            .and(path(REPO_RUNNERS))
3501            .respond_with(Script::new(vec![
3502                ResponseTemplate::new(429)
3503                    .insert_header("retry-after", "120")
3504                    .set_body_json(
3505                        json!({ "message": "You have exceeded a secondary rate limit" }),
3506                    ),
3507                ResponseTemplate::new(200).set_body_json(runner_page(1..2, 1)),
3508            ]))
3509            .mount(&server)
3510            .await;
3511
3512        let clock = Arc::new(TestClock::default());
3513        let gateway = gateway(&server, clock.clone());
3514        let cancel = CancelToken::new();
3515
3516        let first = gateway
3517            .list_runners(&repo_target(), &cancel)
3518            .await
3519            .expect_err("GitHub is rate limiting");
3520        let limit = first.rate_limited().expect("a distinct rate-limited state");
3521        assert_eq!(limit.kind, RateLimitKind::Secondary);
3522        assert_eq!(limit.retry_after, Some(Duration::from_secs(120)));
3523        assert_eq!(requests_seen(&server).await, 1);
3524
3525        // The window is open. A second call must not reach the wire.
3526        let second = gateway
3527            .list_runners(&repo_target(), &cancel)
3528            .await
3529            .expect_err("the back-off is still running");
3530        assert!(second.is_rate_limited(), "{second}");
3531        assert_eq!(
3532            requests_seen(&server).await,
3533            1,
3534            "obeying `retry-after` means sending nothing, not sending and waiting"
3535        );
3536        assert_eq!(
3537            gateway.rate_limit_backoff(),
3538            Some(Duration::from_secs(120)),
3539            "the reported wait is what is left of it"
3540        );
3541
3542        // Part-way through, still suppressed, and the countdown has moved.
3543        clock.advance_secs(90);
3544        assert!(
3545            gateway
3546                .list_runners(&repo_target(), &cancel)
3547                .await
3548                .is_err_and(|error| error.is_rate_limited())
3549        );
3550        assert_eq!(gateway.rate_limit_backoff(), Some(Duration::from_secs(30)));
3551        assert_eq!(requests_seen(&server).await, 1);
3552
3553        // Elapsed. Traffic resumes.
3554        clock.advance_secs(30);
3555        assert_eq!(gateway.rate_limit_backoff(), None);
3556        let inventory = gateway
3557            .list_runners(&repo_target(), &cancel)
3558            .await
3559            .expect("the back-off elapsed");
3560        assert_eq!(inventory.len(), 1);
3561        assert_eq!(requests_seen(&server).await, 2);
3562    }
3563
3564    /// `issue`'s `cancel.check()` runs **before** the rate-limit gate, and that
3565    /// ordering is the one effect the check does not share with `run`.
3566    ///
3567    /// Everywhere else the two overlap: a token flipped before or during a
3568    /// request is caught by `run`'s biased `select!` whether or not `check` ran
3569    /// first. Inside a **latched back-off window** it cannot be, because `issue`
3570    /// returns at the suppression branch without ever reaching `run`. Delete the
3571    /// `check` and this call is answered [`InventoryError::RateLimited`] — a
3572    /// caller that has already navigated away is told to wait out a back-off it
3573    /// is never coming back for, and `f1`'s countdown would render for a refresh
3574    /// nobody asked for any more.
3575    ///
3576    /// That is the real content of "removing `check` alone reds nothing": before
3577    /// this test, the guard's only non-redundant behaviour was the one behaviour
3578    /// nothing exercised.
3579    #[tokio::test]
3580    async fn a_cancelled_call_inside_a_latched_window_is_cancelled_not_rate_limited() {
3581        let server = MockServer::start().await;
3582        Mock::given(method("GET"))
3583            .and(path(REPO_RUNNERS))
3584            .respond_with(
3585                ResponseTemplate::new(429)
3586                    .insert_header("retry-after", "120")
3587                    .set_body_json(
3588                        json!({ "message": "You have exceeded a secondary rate limit" }),
3589                    ),
3590            )
3591            .mount(&server)
3592            .await;
3593
3594        let clock = Arc::new(TestClock::default());
3595        let gateway = gateway(&server, clock.clone());
3596
3597        // Latch the window with a live token, exactly as an ordinary refresh
3598        // would.
3599        let live = CancelToken::new();
3600        let first = gateway
3601            .list_runners(&repo_target(), &live)
3602            .await
3603            .expect_err("GitHub is rate limiting");
3604        assert!(first.is_rate_limited(), "{first}");
3605        assert_eq!(
3606            gateway.rate_limit_backoff(),
3607            Some(Duration::from_secs(120)),
3608            "the window has to actually be open, or this test proves nothing"
3609        );
3610
3611        // Same gateway, same open window — but this caller has withdrawn.
3612        let cancelled = CancelToken::new();
3613        cancelled.cancel();
3614        let error = gateway
3615            .list_runners(&repo_target(), &cancelled)
3616            .await
3617            .expect_err("a cancelled call is still an error");
3618
3619        assert!(
3620            error.is_cancelled(),
3621            "the answer a withdrawn caller gets is `Cancelled`: {error}"
3622        );
3623        assert!(
3624            !error.is_rate_limited(),
3625            "answering `RateLimited` tells a caller that navigated away to wait out a \
3626             back-off it will never return for; the suppression branch must not \
3627             outrank the cancellation: {error}"
3628        );
3629        assert_eq!(
3630            requests_seen(&server).await,
3631            1,
3632            "and neither answer reached the wire: the window suppressed nothing extra \
3633             and the cancellation opened no socket"
3634        );
3635        assert_eq!(
3636            gateway.requests_issued(),
3637            1,
3638            "the budget accounting agrees: only the call that latched the window spent \
3639             anything"
3640        );
3641    }
3642
3643    /// The primary limit: a `403` carrying `x-ratelimit-remaining: 0`. The wait
3644    /// comes from `x-ratelimit-reset`, because a primary limit says *when*
3645    /// rather than *how long*.
3646    #[tokio::test]
3647    async fn an_exhausted_hourly_quota_is_a_distinct_displayable_state() {
3648        let server = MockServer::start().await;
3649        let now = TestClock::default().now().timestamp();
3650        let reset = now + 300;
3651        Mock::given(method("GET"))
3652            .and(path(REPO_RUNNERS))
3653            .respond_with(
3654                ResponseTemplate::new(403)
3655                    .insert_header("x-ratelimit-remaining", "0")
3656                    .insert_header("x-ratelimit-limit", "5000")
3657                    .insert_header("x-ratelimit-reset", reset.to_string().as_str())
3658                    .set_body_json(json!({ "message": "API rate limit exceeded" })),
3659            )
3660            .mount(&server)
3661            .await;
3662
3663        let clock = Arc::new(TestClock::default());
3664        let gateway = gateway(&server, clock);
3665        let error = gateway
3666            .list_runners(&repo_target(), &CancelToken::new())
3667            .await
3668            .expect_err("the quota is gone");
3669
3670        let limit = *error.rate_limited().expect("a rate-limited state");
3671        assert_eq!(limit.kind, RateLimitKind::Primary);
3672        assert_eq!(limit.remaining, Some(0));
3673        assert_eq!(
3674            limit.reset_unix_secs,
3675            Some(u64::try_from(reset).unwrap()),
3676            "the reset instant is what tells an operator how long this lasts"
3677        );
3678        assert_eq!(gateway.rate_limit_backoff(), Some(Duration::from_secs(300)));
3679
3680        // Displayable rather than opaque: a state, a sentence, and a delay `e1`
3681        // can add to its refresh interval.
3682        let state = RefreshState::from_error(&error);
3683        assert_eq!(state, RefreshState::RateLimited(limit));
3684        assert!(!state.is_ready());
3685        let rendered = state.to_string();
3686        assert!(rendered.contains("primary"), "{rendered}");
3687        assert!(rendered.contains("Refreshes are delayed"), "{rendered}");
3688        assert_eq!(
3689            state.retry_delay(TestClock::default().now()),
3690            Some(Duration::from_secs(300))
3691        );
3692    }
3693
3694    /// The false positive this detection is narrowed to avoid.
3695    ///
3696    /// GitHub attaches `x-ratelimit-*` to **every** response. A `404` that
3697    /// happens to arrive on the request that exhausted the quota therefore
3698    /// carries `remaining: 0` while having nothing to do with rate limiting —
3699    /// and reporting it as one would tell an operator to wait for a repository
3700    /// name that will never resolve, while silently latching a back-off that
3701    /// suppresses every other target's refresh too.
3702    #[tokio::test]
3703    async fn a_404_carrying_an_exhausted_remaining_header_is_not_a_rate_limit() {
3704        let server = MockServer::start().await;
3705        Mock::given(method("GET"))
3706            .and(path(REPO_RUNNERS))
3707            .respond_with(
3708                ResponseTemplate::new(404)
3709                    .insert_header("x-ratelimit-remaining", "0")
3710                    .set_body_json(json!({ "message": "Not Found" })),
3711            )
3712            .mount(&server)
3713            .await;
3714
3715        let gateway = gateway(&server, Arc::new(TestClock::default()));
3716        let error = gateway
3717            .list_runners(&repo_target(), &CancelToken::new())
3718            .await
3719            .expect_err("the repository is not there");
3720
3721        assert!(!error.is_rate_limited(), "{error}");
3722        assert!(
3723            gateway.rate_limit_backoff().is_none(),
3724            "a 404 must not silence this gateway"
3725        );
3726        assert!(matches!(
3727            RefreshState::from_error(&error),
3728            RefreshState::Failed {
3729                status: Some(404),
3730                ..
3731            }
3732        ));
3733    }
3734
3735    /// The other false positive: an ordinary permissions refusal.
3736    #[tokio::test]
3737    async fn a_permissions_403_is_forbidden_and_not_a_rate_limit() {
3738        let server = MockServer::start().await;
3739        Mock::given(method("GET"))
3740            .and(path(REPO_RUNNERS))
3741            .respond_with(
3742                ResponseTemplate::new(403)
3743                    .set_body_json(json!({ "message": "Resource not accessible by integration" })),
3744            )
3745            .mount(&server)
3746            .await;
3747
3748        let gateway = gateway(&server, Arc::new(TestClock::default()));
3749        let error = gateway
3750            .list_runners(&repo_target(), &CancelToken::new())
3751            .await
3752            .expect_err("the installation does not grant it");
3753
3754        assert!(!error.is_rate_limited(), "{error}");
3755        assert!(gateway.rate_limit_backoff().is_none());
3756        let state = RefreshState::from_error(&error);
3757        assert!(
3758            matches!(&state, RefreshState::Forbidden { message } if message.as_deref()
3759                == Some("Resource not accessible by integration")),
3760            "{state:?}: waiting does not fix a missing grant, so it must not be \
3761             rendered as something to wait for"
3762        );
3763        assert_eq!(
3764            state.retry_delay(TestClock::default().now()),
3765            None,
3766            "there is nothing to wait for"
3767        );
3768    }
3769
3770    /// The `403` twin of the `404` case above, and the one the status gate alone
3771    /// did not close.
3772    ///
3773    /// GitHub attaches `x-ratelimit-*` to every response, so a permissions
3774    /// refusal that happens to land on the request which exhausted the hourly
3775    /// quota arrives as a `403` carrying `remaining: 0`. Classifying it on the
3776    /// header alone made it a `Primary` limit — an operator told to wait out a
3777    /// grant that will never arrive, and a latched window suppressing every
3778    /// other target's refresh meanwhile, which is exactly the harm the `404`
3779    /// test names.
3780    ///
3781    /// The message is what separates them, and a genuine primary limit always
3782    /// carries "API rate limit exceeded"
3783    /// (`an_exhausted_hourly_quota_is_a_distinct_displayable_state` sends it).
3784    #[tokio::test]
3785    async fn a_permissions_403_that_lands_on_an_exhausted_quota_is_still_forbidden() {
3786        let server = MockServer::start().await;
3787        let reset = TestClock::default().now().timestamp() + 900;
3788        Mock::given(method("GET"))
3789            .and(path(REPO_RUNNERS))
3790            .respond_with(
3791                ResponseTemplate::new(403)
3792                    // The headers of a genuine exhausted quota...
3793                    .insert_header("x-ratelimit-remaining", "0")
3794                    .insert_header("x-ratelimit-limit", "5000")
3795                    .insert_header("x-ratelimit-reset", reset.to_string().as_str())
3796                    // ...on a body that is plainly a permissions refusal.
3797                    .set_body_json(json!({ "message": "Resource not accessible by integration" })),
3798            )
3799            .mount(&server)
3800            .await;
3801
3802        let gateway = gateway(&server, Arc::new(TestClock::default()));
3803        let error = gateway
3804            .list_runners(&repo_target(), &CancelToken::new())
3805            .await
3806            .expect_err("the installation does not grant it");
3807
3808        assert!(
3809            !error.is_rate_limited(),
3810            "a missing grant does not become a rate limit because the quota also ran \
3811             out on the same request: {error}"
3812        );
3813        assert!(
3814            gateway.rate_limit_backoff().is_none(),
3815            "and it must not latch a window that silences every other target too"
3816        );
3817        let state = RefreshState::from_error(&error);
3818        assert!(
3819            matches!(&state, RefreshState::Forbidden { message } if message.as_deref()
3820                == Some("Resource not accessible by integration")),
3821            "{state:?}"
3822        );
3823        assert_eq!(
3824            state.retry_delay(TestClock::default().now()),
3825            None,
3826            "waiting for the quota to reset will not grant the permission"
3827        );
3828    }
3829
3830    /// A rate limit that names no wait still waits: answering "retry in zero
3831    /// seconds" would turn a rate limit into a busy loop against the endpoint
3832    /// that asked for quiet.
3833    #[test]
3834    fn a_rate_limit_with_no_usable_delay_still_backs_off() {
3835        let now = TestClock::default().now();
3836        let bare = RateLimited {
3837            kind: RateLimitKind::Secondary,
3838            retry_after: None,
3839            remaining: None,
3840            reset_unix_secs: None,
3841        };
3842        assert_eq!(bare.delay_from(now), DEFAULT_RATE_LIMIT_BACKOFF);
3843
3844        let stale_reset = RateLimited {
3845            reset_unix_secs: Some(u64::try_from(now.timestamp() - 60).unwrap()),
3846            ..bare
3847        };
3848        assert_eq!(
3849            stale_reset.delay_from(now),
3850            DEFAULT_RATE_LIMIT_BACKOFF,
3851            "a reset already in the past means the clocks disagree, not that the \
3852             limit has lifted"
3853        );
3854
3855        let absurd = RateLimited {
3856            retry_after: Some(Duration::from_secs(86_400)),
3857            ..bare
3858        };
3859        assert_eq!(
3860            absurd.delay_from(now),
3861            MAX_RATE_LIMIT_BACKOFF,
3862            "a remote header does not get to decide how long this product stays dark"
3863        );
3864    }
3865
3866    /// Rate limiting is "displayed, never hidden" — including before it bites.
3867    #[tokio::test]
3868    async fn the_hourly_quota_is_read_from_successful_responses_too() {
3869        let server = MockServer::start().await;
3870        Mock::given(method("GET"))
3871            .and(path(REPO_RUNNERS))
3872            .respond_with(
3873                ResponseTemplate::new(200)
3874                    .insert_header("x-ratelimit-limit", "5000")
3875                    .insert_header("x-ratelimit-remaining", "4873")
3876                    .insert_header("x-ratelimit-reset", "1787274000")
3877                    .set_body_json(runner_page(1..3, 2)),
3878            )
3879            .mount(&server)
3880            .await;
3881
3882        let gateway = gateway(&server, Arc::new(TestClock::default()));
3883        assert_eq!(gateway.headroom(), None, "nothing observed yet");
3884        gateway
3885            .list_runners(&repo_target(), &CancelToken::new())
3886            .await
3887            .expect("readable");
3888
3889        assert_eq!(
3890            gateway.headroom(),
3891            Some(RateLimitHeadroom {
3892                limit: Some(5_000),
3893                remaining: Some(4_873),
3894                reset_unix_secs: Some(1_787_274_000),
3895            }),
3896            "a quota display that only appears once the quota is gone is not a display"
3897        );
3898    }
3899
3900    // -- cancellation -------------------------------------------------------
3901
3902    /// Cancels a token **while a request is in flight**, deterministically.
3903    ///
3904    /// wiremock calls `respond` to *build* the template, which is strictly
3905    /// before any response byte is written. So the flip lands while the request
3906    /// that provoked it is still open, and [`CancelToken::run`]'s biased
3907    /// `select!` — whose watch channel wakes the task to force exactly that poll
3908    /// — answers `Cancelled` for **that request itself**. Its response is
3909    /// dropped unparsed.
3910    ///
3911    /// # This is the in-flight case, not the between-pages one
3912    ///
3913    /// It reads like the between-pages case and it is not, which is worth
3914    /// stating because this fixture spent a round claiming to be it. A walk cut
3915    /// short here never obtains the `Link` header, so it is not *declining* to
3916    /// follow page two — it never saw page two offered. All the outward
3917    /// assertions (`is_cancelled`, one request seen, one request issued) hold
3918    /// identically under both mechanisms, which is precisely why they cannot
3919    /// tell them apart.
3920    ///
3921    /// [`RestInventory::headroom`] is what separates them: it is written only in
3922    /// `issue`'s `Ok(response)` arm, so it stays `None` here and is `Some` in
3923    /// `cancelling_between_pages_stops_the_walk`. The two tests assert opposite
3924    /// sides of that one observable, and neither can pass as the other.
3925    struct CancelWhileServingPage {
3926        token: CancelToken,
3927        next_page: String,
3928        body: Value,
3929    }
3930
3931    impl wiremock::Respond for CancelWhileServingPage {
3932        fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate {
3933            self.token.cancel();
3934            ResponseTemplate::new(200)
3935                // Carried so that the discriminator is *available* to be
3936                // observed and is still absent. `headroom` staying `None` when
3937                // the response plainly offered it is what proves this response
3938                // was never parsed at all.
3939                .insert_header("x-ratelimit-limit", "5000")
3940                .insert_header("x-ratelimit-remaining", "4999")
3941                .insert_header("x-ratelimit-reset", "1787274000")
3942                .insert_header("link", link_next(&self.next_page).as_str())
3943                .set_body_json(self.body.clone())
3944        }
3945    }
3946
3947    /// A token flipped **while a page is in flight** abandons that page, and
3948    /// spends nothing after it.
3949    ///
3950    /// The walk begins un-cancelled and fetches page one; the token flips inside
3951    /// the mock server, so `CancelToken::run` abandons page one's own request
3952    /// and its response is dropped unparsed. Page two is offered by a `Link`
3953    /// header that the walk consequently never reads, and must never be asked
3954    /// for either way.
3955    ///
3956    /// This is [`CancelToken::run`]'s half of the guard — "a cancellation
3957    /// arriving mid-flight is not noticed until the response does", as
3958    /// `RestInventory::get`'s doc puts it. The between-pages half is
3959    /// `cancelling_between_pages_stops_the_walk`, which needs a seam this
3960    /// fixture cannot provide; the `headroom` assertion below is what keeps the
3961    /// two from being mistaken for each other.
3962    ///
3963    /// Distinct from `cancelling_an_in_flight_request_abandons_it`, which
3964    /// asserts *promptness* — that the walk does not sit out a 20-second
3965    /// response — and pays a spawned canceller and a wall-clock bound to do it.
3966    /// This one asserts the *observable consequence* of abandoning, that the
3967    /// response is never parsed, and is deterministic.
3968    #[tokio::test]
3969    async fn a_cancellation_landing_mid_request_drops_the_response_unparsed() {
3970        let server = MockServer::start().await;
3971        let cancel = CancelToken::new();
3972        Mock::given(method("GET"))
3973            .and(path(REPO_RUNNERS))
3974            .respond_with(CancelWhileServingPage {
3975                token: cancel.clone(),
3976                next_page: format!("{}/page/2", server.uri()),
3977                body: runner_page(1..101, 200),
3978            })
3979            .expect(1)
3980            .mount(&server)
3981            .await;
3982        Mock::given(method("GET"))
3983            .and(path("/page/2"))
3984            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..201, 200)))
3985            .expect(0)
3986            .mount(&server)
3987            .await;
3988
3989        let gateway = gateway(&server, Arc::new(TestClock::default()));
3990        assert!(
3991            !cancel.is_cancelled(),
3992            "the walk has to start live, or this is a test of page zero"
3993        );
3994
3995        let error = gateway
3996            .list_runners(&repo_target(), &cancel)
3997            .await
3998            .expect_err("the token was flipped between page one and page two");
3999        assert!(error.is_cancelled(), "{error}");
4000        assert!(
4001            cancel.is_cancelled(),
4002            "page one was served, which is what flipped the token"
4003        );
4004        assert_eq!(
4005            requests_seen(&server).await,
4006            1,
4007            "page one, and nothing after it: the `Link` header offered page two and the \
4008             walk declined to spend the request"
4009        );
4010        assert_eq!(
4011            gateway.requests_issued(),
4012            1,
4013            "and the budget accounting agrees with the wire"
4014        );
4015        assert!(
4016            gateway.headroom().is_none(),
4017            "the response carried `x-ratelimit-*` and they were never read, which is what \
4018             `abandoned in flight` means: `issue` returned `Cancelled` from `run` without \
4019             reaching its `Ok` arm. `Some` here would mean page one was actually parsed \
4020             and this test had silently become the between-pages case"
4021        );
4022    }
4023
4024    /// The seam the between-pages property needs: a token flipped **while no
4025    /// request is in flight**.
4026    ///
4027    /// Cancelling from the mock server cannot express it. wiremock builds the
4028    /// template before writing a byte, so that flip always lands mid-request and
4029    /// `run` abandons the very page that triggered it — see
4030    /// [`CancelWhileServingPage`]. A `set_delay` plus a spawned canceller would
4031    /// hit the right window on an idle machine and the wrong one on a loaded
4032    /// one, which is a coin-flip dressed as a test.
4033    ///
4034    /// Parsing is the seam. `collect_pages` calls `into_items` after `issue` has
4035    /// returned `Ok` for the page in hand — `headroom` already recorded — and
4036    /// before it reads the `Link` header or issues anything further. There is no
4037    /// socket open at that instant, so a token flipped here is flipped *exactly*
4038    /// between pages, by construction rather than by timing.
4039    ///
4040    /// The token travels in a static because `serde` builds this type and cannot
4041    /// be handed one. Only `cancelling_between_pages_stops_the_walk` arms it, and
4042    /// it is `take`n on first use so a second page could not re-trigger it.
4043    static CANCEL_WHILE_PARSING: std::sync::Mutex<Option<CancelToken>> =
4044        std::sync::Mutex::new(None);
4045
4046    /// A [`WirePage`] that is byte-identical to [`RunnersPage`] on the wire and
4047    /// flips [`CANCEL_WHILE_PARSING`] as `collect_pages` unwraps it.
4048    #[derive(Debug, Deserialize)]
4049    struct CancelOnParsePage {
4050        total_count: Option<u64>,
4051        #[serde(default)]
4052        runners: Vec<RawRunner>,
4053    }
4054
4055    impl WirePage for CancelOnParsePage {
4056        type Item = RawRunner;
4057        const WHAT: &'static str = "runners";
4058
4059        fn reported_total(&self) -> Option<u64> {
4060            self.total_count
4061        }
4062
4063        fn into_items(self) -> Vec<Self::Item> {
4064            if let Some(token) = CANCEL_WHILE_PARSING
4065                .lock()
4066                .expect("the parse-time cancel seam is not poisoned")
4067                .take()
4068            {
4069                token.cancel();
4070            }
4071            self.runners
4072        }
4073    }
4074
4075    /// A token flipped **between pages** stops the walk there, rather than
4076    /// spending the shared budget on pages nobody will read.
4077    ///
4078    /// The walk begins un-cancelled and fetches page one. Page one is served
4079    /// completely, parsed, and its `Link: rel="next"` really is in hand — the
4080    /// `headroom` assertion below is the proof, since `issue` writes it only on
4081    /// the `Ok` path. *Then* the token flips, with nothing in flight. Page two
4082    /// is therefore a request the walk is in a position to make and **declines**
4083    /// to, which is the actual property: this is the case `RestInventory::get`'s
4084    /// doc calls "the one the shared budget cares about", and the case a long
4085    /// organization walk hits.
4086    ///
4087    /// Driven through `collect_pages` rather than `list_runners` because the
4088    /// seam is the page type, and `list_runners` fixes that to [`RunnersPage`].
4089    /// The walk under test is the same one either way — `list_runners` is a thin
4090    /// wrapper over this call — and the between-pages decision lives entirely in
4091    /// `collect_pages` and `issue`.
4092    #[tokio::test]
4093    async fn cancelling_between_pages_stops_the_walk() {
4094        let server = MockServer::start().await;
4095        let page_two = format!("{}/page/2", server.uri());
4096        Mock::given(method("GET"))
4097            .and(path(REPO_RUNNERS))
4098            .respond_with(
4099                ResponseTemplate::new(200)
4100                    // Read on the `Ok` path, and the discriminator that
4101                    // separates this test from the in-flight one.
4102                    .insert_header("x-ratelimit-limit", "5000")
4103                    .insert_header("x-ratelimit-remaining", "4999")
4104                    .insert_header("x-ratelimit-reset", "1787274000")
4105                    .insert_header("link", link_next(&page_two).as_str())
4106                    .set_body_json(runner_page(1..101, 200)),
4107            )
4108            .expect(1)
4109            .mount(&server)
4110            .await;
4111        Mock::given(method("GET"))
4112            .and(path("/page/2"))
4113            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..201, 200)))
4114            .expect(0)
4115            .mount(&server)
4116            .await;
4117
4118        let gateway = gateway(&server, Arc::new(TestClock::default()));
4119        let cancel = CancelToken::new();
4120        *CANCEL_WHILE_PARSING
4121            .lock()
4122            .expect("the parse-time cancel seam is not poisoned") = Some(cancel.clone());
4123
4124        assert!(
4125            !cancel.is_cancelled(),
4126            "the walk has to start live, or this is a test of page zero"
4127        );
4128
4129        let first = ApiRequest::get(REPO_RUNNERS).query("per_page", PER_PAGE);
4130        // Matched rather than `expect_err`: `Collected` is a private walk result
4131        // that does not derive `Debug`, and a test is not a reason to widen it.
4132        let error = match gateway
4133            .collect_pages::<CancelOnParsePage>(first, &cancel)
4134            .await
4135        {
4136            Err(error) => error,
4137            Ok(_) => panic!("the token was flipped between page one and page two"),
4138        };
4139
4140        assert!(error.is_cancelled(), "{error}");
4141        assert!(
4142            cancel.is_cancelled(),
4143            "page one was parsed, which is what flipped the token"
4144        );
4145        assert!(
4146            gateway.headroom().is_some(),
4147            "page one's response must have been parsed for this to be the between-pages \
4148             case at all; `headroom` is written only in `issue`'s `Ok` arm, so `None` here \
4149             would mean page one was cancelled in flight and the walk never saw the \
4150             `Link` header it is supposed to decline to follow"
4151        );
4152        assert_eq!(
4153            requests_seen(&server).await,
4154            1,
4155            "page one, and nothing after it: the `Link` header offered page two and the \
4156             walk declined to spend the request"
4157        );
4158        assert_eq!(
4159            gateway.requests_issued(),
4160            1,
4161            "and the budget accounting agrees with the wire"
4162        );
4163    }
4164
4165    /// The neighbouring case, and the one the between-pages test used to be.
4166    ///
4167    /// A walk handed a token that is *already* cancelled stops before page one
4168    /// rather than between pages. Worth keeping — it is what a caller that
4169    /// navigated away before the refresh started actually does — but it is a
4170    /// different property, and naming it after the between-pages case left that
4171    /// case uncovered.
4172    #[tokio::test]
4173    async fn an_already_cancelled_token_stops_a_walk_before_its_first_page() {
4174        let server = MockServer::start().await;
4175        let page_two = format!("{}/page/2", server.uri());
4176        Mock::given(method("GET"))
4177            .and(path(REPO_RUNNERS))
4178            .respond_with(
4179                ResponseTemplate::new(200)
4180                    .insert_header("link", link_next(&page_two).as_str())
4181                    .set_body_json(runner_page(1..101, 200)),
4182            )
4183            .mount(&server)
4184            .await;
4185        Mock::given(method("GET"))
4186            .and(path("/page/2"))
4187            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..201, 200)))
4188            .expect(0)
4189            .mount(&server)
4190            .await;
4191
4192        let gateway = gateway(&server, Arc::new(TestClock::default()));
4193        let cancel = CancelToken::new();
4194
4195        // One request, issued by hand and outside any walk, purely to establish
4196        // that the server would serve a page and offer a second.
4197        let first = ApiRequest::get(REPO_RUNNERS).query("per_page", PER_PAGE);
4198        let response = gateway
4199            .issue(&first, &cancel)
4200            .await
4201            .expect("page one is readable");
4202        assert!(response.next_page().is_some(), "page two is on offer");
4203        cancel.cancel();
4204
4205        let error = gateway
4206            .list_runners(&repo_target(), &cancel)
4207            .await
4208            .expect_err("the token is cancelled");
4209        assert!(error.is_cancelled(), "{error}");
4210        assert_eq!(
4211            requests_seen(&server).await,
4212            1,
4213            "the manual request only; the walk spent nothing at all"
4214        );
4215        assert_eq!(
4216            gateway.requests_issued(),
4217            1,
4218            "and the budget accounting agrees with the wire: a request that was \
4219             never polled is not a request that was issued"
4220        );
4221    }
4222
4223    /// Cancellation of a request already on the wire, which is the case a
4224    /// between-pages check alone does not cover.
4225    #[tokio::test]
4226    async fn cancelling_an_in_flight_request_abandons_it() {
4227        let server = MockServer::start().await;
4228        Mock::given(method("GET"))
4229            .and(path(REPO_RUNNERS))
4230            .respond_with(
4231                ResponseTemplate::new(200)
4232                    .set_delay(Duration::from_secs(20))
4233                    .set_body_json(runner_page(1..2, 1)),
4234            )
4235            .mount(&server)
4236            .await;
4237
4238        let gateway = gateway(&server, Arc::new(TestClock::default()));
4239        let cancel = CancelToken::new();
4240        let token = cancel.clone();
4241        let canceller = tokio::spawn(async move {
4242            tokio::time::sleep(Duration::from_millis(50)).await;
4243            token.cancel();
4244        });
4245
4246        let started = std::time::Instant::now();
4247        let error = gateway
4248            .list_runners(&repo_target(), &cancel)
4249            .await
4250            .expect_err("the caller withdrew");
4251        let elapsed = started.elapsed();
4252        canceller.await.expect("the canceller completes");
4253
4254        assert!(error.is_cancelled(), "{error}");
4255        assert!(
4256            elapsed < Duration::from_secs(10),
4257            "the request was awaited to completion rather than abandoned: {elapsed:?}"
4258        );
4259        assert!(cancel.is_cancelled());
4260        assert!(
4261            CancelToken::new().check().is_ok(),
4262            "a fresh token is not cancelled"
4263        );
4264    }
4265
4266    // -- runner package downloads -------------------------------------------
4267
4268    /// The Definition of Done's optional checksum: absent stays absent, and is
4269    /// distinguishable from empty.
4270    #[tokio::test]
4271    async fn an_absent_sha256_checksum_is_absent_and_an_empty_one_is_empty() {
4272        let server = MockServer::start().await;
4273        Mock::given(method("GET"))
4274            .and(path("/repos/octo/dashboard/actions/runners/downloads"))
4275            .respond_with(ResponseTemplate::new(200).set_body_json(json!([
4276                {
4277                    "os": "win",
4278                    "architecture": "x64",
4279                    "download_url": "https://example.invalid/win-x64.zip",
4280                    "filename": "actions-runner-win-x64.zip",
4281                    "sha256_checksum": "abc123"
4282                },
4283                {
4284                    // The field is simply not there.
4285                    "os": "osx",
4286                    "architecture": "arm64",
4287                    "download_url": "https://example.invalid/osx-arm64.tar.gz",
4288                    "filename": "actions-runner-osx-arm64.tar.gz"
4289                },
4290                {
4291                    // The field is there and null.
4292                    "os": "linux",
4293                    "architecture": "x64",
4294                    "download_url": "https://example.invalid/linux-x64.tar.gz",
4295                    "filename": "actions-runner-linux-x64.tar.gz",
4296                    "sha256_checksum": null
4297                },
4298                {
4299                    // The field is there and empty, which is a different fact.
4300                    "os": "linux",
4301                    "architecture": "arm64",
4302                    "download_url": "https://example.invalid/linux-arm64.tar.gz",
4303                    "filename": "actions-runner-linux-arm64.tar.gz",
4304                    "sha256_checksum": ""
4305                }
4306            ])))
4307            .mount(&server)
4308            .await;
4309
4310        let gateway = gateway(&server, Arc::new(TestClock::default()));
4311        let downloads = gateway
4312            .runner_downloads(&repo_target(), &CancelToken::new())
4313            .await
4314            .expect("the metadata is readable");
4315
4316        let windows = downloads
4317            .select(Os::Windows, Arch::X64)
4318            .expect("selected by OS and architecture");
4319        assert_eq!(windows.sha256_checksum(), Some("abc123"));
4320
4321        let missing = downloads.select(Os::MacOs, Arch::Arm64).expect("selected");
4322        assert_eq!(
4323            missing.sha256_checksum(),
4324            None,
4325            "`e2` fails closed on an absent digest, and can only do that if this \
4326             layer does not paper the absence over"
4327        );
4328
4329        let null = downloads.select(Os::Linux, Arch::X64).expect("selected");
4330        assert_eq!(
4331            null.sha256_checksum(),
4332            None,
4333            "an explicit null is absent too"
4334        );
4335
4336        let empty = downloads.select(Os::Linux, Arch::Arm64).expect("selected");
4337        assert_eq!(
4338            empty.sha256_checksum(),
4339            Some(""),
4340            "an empty digest is a different fact from a missing one, and \
4341             collapsing them would leave `e2` unable to report which it saw"
4342        );
4343        assert_ne!(
4344            empty.sha256_checksum(),
4345            missing.sha256_checksum(),
4346            "absent and empty must be distinguishable"
4347        );
4348
4349        assert_eq!(
4350            downloads.select(Os::Windows, Arch::Arm32),
4351            None,
4352            "an unpublished pair is refused rather than substituted"
4353        );
4354        assert_eq!(gateway.requests_issued(), 1, "downloads are not paginated");
4355    }
4356
4357    /// The organization form of the same endpoint.
4358    #[tokio::test]
4359    async fn runner_downloads_are_read_at_organization_scope_too() {
4360        let server = MockServer::start().await;
4361        Mock::given(method("GET"))
4362            .and(path("/orgs/octo-org/actions/runners/downloads"))
4363            .respond_with(ResponseTemplate::new(200).set_body_json(json!([{
4364                "os": "linux",
4365                "architecture": "arm",
4366                "download_url": "https://example.invalid/linux-arm.tar.gz",
4367                "filename": "actions-runner-linux-arm.tar.gz",
4368                "sha256_checksum": "deadbeef"
4369            }])))
4370            .expect(1)
4371            .mount(&server)
4372            .await;
4373
4374        let gateway = gateway(&server, Arc::new(TestClock::default()));
4375        let downloads = gateway
4376            .runner_downloads(&org_target(), &CancelToken::new())
4377            .await
4378            .expect("readable");
4379        assert!(downloads.select(Os::Linux, Arch::Arm32).is_some());
4380    }
4381
4382    // -- runner shape -------------------------------------------------------
4383
4384    /// The D18 spike's label facts, kept true in the type.
4385    #[tokio::test]
4386    async fn labels_are_read_as_github_stores_them_and_matched_case_insensitively() {
4387        let server = MockServer::start().await;
4388        Mock::given(method("GET"))
4389            .and(path(REPO_RUNNERS))
4390            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
4391                "total_count": 2,
4392                "runners": [
4393                    {
4394                        "id": 73,
4395                        "name": "rm-d18-spike-ivanpc-1753",
4396                        "os": "win",
4397                        "status": "offline",
4398                        "busy": false,
4399                        "ephemeral": true,
4400                        // Lower-cased by GitHub, and carrying exactly what was
4401                        // requested — no `self-hosted`, no OS, no architecture.
4402                        "labels": [
4403                            { "id": 1, "name": "rm-home-win-x64", "type": "read-only" },
4404                            { "id": 2, "name": "windows", "type": "read-only" }
4405                        ]
4406                    },
4407                    {
4408                        "id": 74,
4409                        "name": "legacy-persistent",
4410                        "os": "Linux",
4411                        "status": "provisioning",
4412                        "busy": true,
4413                        "labels": []
4414                    }
4415                ]
4416            })))
4417            .mount(&server)
4418            .await;
4419
4420        let gateway = gateway(&server, Arc::new(TestClock::default()));
4421        let inventory = gateway
4422            .list_runners(&repo_target(), &CancelToken::new())
4423            .await
4424            .expect("readable");
4425
4426        let spike = &inventory.runners()[0];
4427        assert_eq!(spike.labels, ["rm-home-win-x64", "windows"]);
4428        assert!(
4429            spike.has_label("Windows"),
4430            "GitHub lower-cases what it stores"
4431        );
4432        assert!(spike.has_label("  windows  "));
4433        assert!(
4434            !spike.has_label("self-hosted"),
4435            "no label is added implicitly (D18, point 1)"
4436        );
4437        assert_eq!(spike.status, RunnerStatus::Offline);
4438        assert_eq!(spike.ephemeral, Some(true));
4439        assert_eq!(spike.parsed_os(), Some(Os::Windows));
4440
4441        let legacy = &inventory.runners()[1];
4442        assert_eq!(
4443            legacy.status,
4444            RunnerStatus::Other("provisioning".to_string()),
4445            "an unrecognised status is something to display, not something to guess at"
4446        );
4447        assert_eq!(
4448            legacy.ephemeral, None,
4449            "absent is not `false`: a runner whose ephemerality is unknown is \
4450             exactly the one an operator wants flagged"
4451        );
4452        assert_eq!(legacy.parsed_os(), Some(Os::Linux));
4453        assert!(legacy.busy);
4454        assert_eq!(inventory.busy_count(), 1);
4455        assert_eq!(inventory.online_count(), 0);
4456    }
4457
4458    // -- coalescing ---------------------------------------------------------
4459
4460    /// The Definition of Done's coalescing item, measured where it matters: at
4461    /// the mock server's request log.
4462    ///
4463    /// The requirement is a budget one before it is a latency one. `F5` held
4464    /// down on the dashboard would otherwise be an operator-driven denial of
4465    /// service against a 5,000/hour ceiling shared with the polling that keeps
4466    /// runners starting.
4467    #[tokio::test]
4468    async fn a_manual_refresh_during_an_in_flight_one_coalesces_into_a_single_request() {
4469        let server = MockServer::start().await;
4470        Mock::given(method("GET"))
4471            .and(path(REPO_RUNNERS))
4472            .respond_with(
4473                ResponseTemplate::new(200)
4474                    .set_delay(Duration::from_millis(150))
4475                    .set_body_json(runner_page(1..4, 3)),
4476            )
4477            .mount(&server)
4478            .await;
4479        mount_runs(&repo(), 2).mount(&server).await;
4480
4481        let gateway = Arc::new(gateway(&server, Arc::new(TestClock::default())));
4482        let coalescer: Arc<RefreshCoalescer<RefreshState>> = Arc::new(RefreshCoalescer::new());
4483        let scope = ActivityScope::repository(repo());
4484
4485        let refresh = || {
4486            let gateway = gateway.clone();
4487            let coalescer = coalescer.clone();
4488            let scope = scope.clone();
4489            async move {
4490                coalescer
4491                    .refresh(|| async {
4492                        RefreshState::from_result(
4493                            gateway.snapshot(&scope, &CancelToken::new()).await,
4494                        )
4495                    })
4496                    .await
4497            }
4498        };
4499
4500        // The scheduled poll and an operator's manual refresh, together.
4501        let (scheduled, manual) = tokio::join!(refresh(), refresh());
4502
4503        assert_eq!(coalescer.performed(), 1, "one refresh actually ran");
4504        assert_eq!(coalescer.joined(), 1, "the other joined it");
4505        assert_eq!(scheduled, manual, "and both callers got the same answer");
4506        assert!(scheduled.is_ready(), "{scheduled}");
4507        assert_eq!(
4508            scheduled.snapshot().expect("ready").runners.len(),
4509            3,
4510            "joining must return the answer, not an empty placeholder"
4511        );
4512        assert_eq!(
4513            requests_seen(&server).await,
4514            2,
4515            "one refresh is one runners request plus one runs request; a second \
4516             refresh would have made it four"
4517        );
4518        assert_eq!(gateway.requests_issued(), 2);
4519        assert_eq!(coalescer.last(), Some(scheduled));
4520    }
4521
4522    /// A refresh that arrives *after* the previous one finished is not a
4523    /// coalescing candidate — it is a new refresh, and must issue its own
4524    /// requests.
4525    #[tokio::test]
4526    async fn a_refresh_after_the_previous_one_completed_is_not_coalesced() {
4527        let server = MockServer::start().await;
4528        Mock::given(method("GET"))
4529            .and(path(REPO_RUNNERS))
4530            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(1..2, 1)))
4531            .mount(&server)
4532            .await;
4533        mount_runs(&repo(), 0).mount(&server).await;
4534
4535        let gateway = gateway(&server, Arc::new(TestClock::default()));
4536        let coalescer: RefreshCoalescer<RefreshState> = RefreshCoalescer::new();
4537        let scope = ActivityScope::repository(repo());
4538
4539        for _ in 0..3 {
4540            let state = coalescer
4541                .refresh(|| async {
4542                    RefreshState::from_result(gateway.snapshot(&scope, &CancelToken::new()).await)
4543                })
4544                .await;
4545            assert!(state.is_ready(), "{state}");
4546        }
4547
4548        assert_eq!(coalescer.performed(), 3);
4549        assert_eq!(
4550            coalescer.joined(),
4551            0,
4552            "coalescing an in-flight refresh must not become caching a finished one"
4553        );
4554        assert_eq!(gateway.requests_issued(), 6);
4555    }
4556
4557    // -- the shared request budget ------------------------------------------
4558
4559    fn interval(secs: u16) -> RefreshInterval {
4560        RefreshInterval::from_secs(secs).expect("at or above the documented floor")
4561    }
4562
4563    /// `04-subsystem-contracts.md`'s per-target table, reproduced exactly.
4564    ///
4565    /// | Per target, per hour | 60 s default | 30 s floor |
4566    /// |---|---|---|
4567    /// | demand | ~120 | ~240 |
4568    /// | runner inventory | ~60 | ~120 |
4569    /// | in-progress workflow count | ~60 | ~120 |
4570    /// | **total** | **~240** | **~480** |
4571    #[test]
4572    fn a_repository_target_costs_the_documented_number_of_requests() {
4573        let default = interval(RefreshInterval::DEFAULT_SECS);
4574        let floor = interval(RefreshInterval::MIN_SECS);
4575
4576        assert_eq!(refreshes_per_hour(default), 60);
4577        assert_eq!(refreshes_per_hour(floor), 120);
4578
4579        let target = TargetCost::repository();
4580        assert_eq!(target.requests_per_refresh(), 4);
4581        assert_eq!(
4582            target.requests_per_hour(default),
4583            240,
4584            "the documented per-target total at the 60-second default"
4585        );
4586        assert_eq!(
4587            target.requests_per_hour(floor),
4588            480,
4589            "and at the 30-second floor"
4590        );
4591    }
4592
4593    /// The Definition of Done's "roughly 10 targets per host at the 60-second
4594    /// default and 5 at the 30-second floor".
4595    #[test]
4596    fn the_projection_reproduces_the_documented_target_ceilings() {
4597        assert_eq!(HOURLY_REQUEST_CEILING, 5_000);
4598        assert_eq!(budget_allowance(), 2_500, "half the ceiling");
4599
4600        assert_eq!(
4601            BudgetProjection::max_repository_targets(interval(RefreshInterval::DEFAULT_SECS)),
4602            10
4603        );
4604        assert_eq!(
4605            BudgetProjection::max_repository_targets(interval(RefreshInterval::MIN_SECS)),
4606            5
4607        );
4608
4609        // And the boundary is where the documented ceilings say it is.
4610        let default = interval(RefreshInterval::DEFAULT_SECS);
4611        let ten = BudgetProjection::new(default, vec![TargetCost::repository(); 10]);
4612        assert_eq!(ten.requests_per_hour(), 2_400);
4613        assert!(!ten.exceeds_allowance());
4614        assert_eq!(ten.headroom(), 100);
4615
4616        let eleven = BudgetProjection::new(default, vec![TargetCost::repository(); 11]);
4617        assert_eq!(eleven.requests_per_hour(), 2_640);
4618        assert!(
4619            eleven.exceeds_allowance(),
4620            "the eleventh repository is the one an operator needs told about"
4621        );
4622        assert_eq!(eleven.headroom(), 0);
4623    }
4624
4625    /// The correction this task owns: an organization is not a flat per-target
4626    /// constant.
4627    #[test]
4628    fn an_organization_target_costs_materially_more_than_a_repository_target() {
4629        let default = interval(RefreshInterval::DEFAULT_SECS);
4630        let repository = TargetCost::repository().requests_per_hour(default);
4631
4632        assert_eq!(
4633            TargetCost::organization(1).requests_per_hour(default),
4634            repository,
4635            "at one installed repository the two models agree exactly, which is \
4636             what makes this a refinement of the documented table rather than a \
4637             contradiction of it"
4638        );
4639
4640        let ten = TargetCost::organization(10);
4641        assert_eq!(ten.requests_per_refresh(), 31);
4642        assert_eq!(ten.requests_per_hour(default), 1_860);
4643        assert!(
4644            ten.requests_per_hour(default) > repository * 7,
4645            "an organization on ten repositories costs nearly eight times a \
4646             repository target; projecting it flat understates the real spend by \
4647             exactly that factor"
4648        );
4649
4650        // Which is why the refusal arrives far earlier for an organization.
4651        let empty = BudgetProjection::new(default, Vec::new());
4652        assert!(empty.admit(TargetCost::repository()).is_admitted());
4653        assert!(empty.admit(TargetCost::organization(13)).is_admitted());
4654        let refusal = empty.admit(TargetCost::organization(14));
4655        assert!(
4656            !refusal.is_admitted(),
4657            "a single organization on fourteen repositories already exceeds a \
4658             host's whole share of the budget"
4659        );
4660    }
4661
4662    /// `f2`'s refusal has to explain itself with the computed numbers, not with
4663    /// the rule.
4664    #[test]
4665    fn a_refused_configuration_states_the_numbers_and_the_maximum_target_count() {
4666        let default = interval(RefreshInterval::DEFAULT_SECS);
4667        let full = BudgetProjection::new(default, vec![TargetCost::repository(); 10]);
4668
4669        let Admission::Refused {
4670            projected_requests_per_hour,
4671            allowance,
4672            max_repository_targets,
4673            ..
4674        } = full.admit(TargetCost::repository())
4675        else {
4676            panic!("the eleventh repository must be refused");
4677        };
4678        assert_eq!(projected_requests_per_hour, 2_640);
4679        assert_eq!(allowance, 2_500);
4680        assert_eq!(max_repository_targets, 10);
4681
4682        let message = full.admit(TargetCost::repository()).to_string();
4683        for expected in ["2640", "2500", "5000", "60-second", "about 10 repository"] {
4684            assert!(
4685                message.contains(expected),
4686                "{expected:?} missing from: {message}"
4687            );
4688        }
4689        assert!(
4690            !message.contains("because the App is installed on"),
4691            "the organization clause belongs only on an organization refusal: {message}"
4692        );
4693
4694        // An organization refusal says which repository count drove it, because
4695        // that is the part a flat per-target reading would not have predicted.
4696        let org_message = full.admit(TargetCost::organization(4)).to_string();
4697        assert!(
4698            org_message.contains("installed on 4 of its repositories"),
4699            "{org_message}"
4700        );
4701
4702        let admitted = BudgetProjection::new(default, vec![TargetCost::repository(); 2])
4703            .admit(TargetCost::repository())
4704            .to_string();
4705        assert!(admitted.contains("720"), "{admitted}");
4706        assert!(admitted.contains("1780"), "{admitted}");
4707    }
4708
4709    /// The projection's per-refresh constants, pinned against the requests the
4710    /// gateway actually issues.
4711    ///
4712    /// Without this the budget model is a table in a document that happens to be
4713    /// written in Rust. Demand is `c4`'s and is not issued here, so the two
4714    /// classes this task owns are compared on their own: an organization
4715    /// refresh is one runners request plus one runs request per installed
4716    /// repository, and the model has to say the same.
4717    #[tokio::test]
4718    async fn the_budget_model_matches_the_requests_the_gateway_really_issues() {
4719        let server = MockServer::start().await;
4720        Mock::given(method("GET"))
4721            .and(path(ORG_RUNNERS))
4722            .respond_with(ResponseTemplate::new(200).set_body_json(runner_page(1..4, 3)))
4723            .mount(&server)
4724            .await;
4725        for repository in [repo(), other_repo(), third_repo()] {
4726            mount_runs(&repository, 1).mount(&server).await;
4727        }
4728
4729        let gateway = gateway(&server, Arc::new(TestClock::default()));
4730        let scope = ActivityScope::organization(
4731            Org::new("octo-org").expect("a valid organization login"),
4732            [repo(), other_repo(), third_repo()],
4733        );
4734        gateway
4735            .snapshot(&scope, &CancelToken::new())
4736            .await
4737            .expect("readable");
4738
4739        let cost = TargetCost::from_activity_scope(&scope);
4740        assert_eq!(cost.installed_repositories(), 3);
4741        assert_eq!(cost.scope(), TargetScope::Organization);
4742
4743        let modelled_without_demand = cost.requests_per_refresh()
4744            - DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH * cost.installed_repositories();
4745        assert_eq!(
4746            gateway.requests_issued(),
4747            u64::from(modelled_without_demand),
4748            "the model projects {modelled_without_demand} inventory-and-activity \
4749             requests per refresh for this scope, and the gateway issued {}",
4750            gateway.requests_issued()
4751        );
4752        assert_eq!(
4753            modelled_without_demand,
4754            RUNNER_INVENTORY_REQUESTS_PER_REFRESH + scope.requests_per_refresh()
4755        );
4756    }
4757
4758    /// An organization the App reaches no repository in is projected as zero
4759    /// repositories, not silently as one.
4760    #[test]
4761    fn an_organization_with_no_installed_repositories_is_projected_as_such() {
4762        let scope = ActivityScope::organization(
4763            Org::new("octo-org").expect("a valid organization login"),
4764            [],
4765        );
4766        assert_eq!(scope.requests_per_refresh(), 0);
4767        let cost = TargetCost::from_activity_scope(&scope);
4768        assert_eq!(cost.installed_repositories(), 0);
4769        assert_eq!(
4770            cost.requests_per_refresh(),
4771            RUNNER_INVENTORY_REQUESTS_PER_REFRESH,
4772            "the runners endpoint is still polled; nothing else is"
4773        );
4774    }
4775
4776    /// `c4` reports its measured demand cost rather than this file estimating
4777    /// it, which is what its specification asks for and what it could not do if
4778    /// the constant were the only way in.
4779    #[test]
4780    fn the_demand_cost_can_be_reported_by_the_task_that_measures_it() {
4781        let default = interval(RefreshInterval::DEFAULT_SECS);
4782
4783        assert_eq!(
4784            TargetCost::repository().requests_per_hour(default),
4785            240,
4786            "the documented estimate is the default"
4787        );
4788
4789        // A demand poll that turned out to need three requests per repository,
4790        // not two.
4791        let measured = TargetCost::repository().with_demand_requests_per_repository(3);
4792        assert_eq!(measured.requests_per_refresh(), 5);
4793        assert_eq!(measured.requests_per_hour(default), 300);
4794
4795        // And it scales with an organization's repository count like everything
4796        // else per-repository does.
4797        let org = TargetCost::organization(4).with_demand_requests_per_repository(3);
4798        assert_eq!(org.requests_per_refresh(), 1 + 4 * (1 + 3));
4799        assert_eq!(
4800            org.requests_per_hour(default),
4801            1_020,
4802            "a worse demand cost lands hardest on an organization, which is \
4803             exactly the effect a flat per-target model would hide"
4804        );
4805    }
4806
4807    /// A repository target's activity scope is its own repository, whatever a
4808    /// caller passes.
4809    #[test]
4810    fn a_repository_activity_scope_covers_exactly_one_repository() {
4811        let scope = ActivityScope::repository(repo());
4812        assert_eq!(scope.repositories(), [repo()]);
4813        assert_eq!(scope.requests_per_refresh(), 1);
4814        assert_eq!(scope.target(), &repo_target());
4815        assert_eq!(
4816            TargetCost::from_activity_scope(&scope),
4817            TargetCost::repository()
4818        );
4819    }
4820
4821    // -- error summarising --------------------------------------------------
4822
4823    /// Every authentication outcome `c2` separates stays separate here. `f1`
4824    /// reports four states and must not collapse them.
4825    #[test]
4826    fn the_authentication_taxonomy_survives_the_summary() {
4827        assert_eq!(
4828            RefreshState::from_error(&InventoryError::Github(GithubError::AuthenticationFailed)),
4829            RefreshState::Unauthorized
4830        );
4831        assert_eq!(
4832            RefreshState::from_error(&InventoryError::Github(
4833                GithubError::AuthenticationLockout {
4834                    retry_after: Duration::from_secs(60)
4835                }
4836            )),
4837            RefreshState::LockedOut {
4838                retry_after: Duration::from_secs(60)
4839            }
4840        );
4841        assert_eq!(
4842            RefreshState::from_error(&InventoryError::Cancelled),
4843            RefreshState::Cancelled
4844        );
4845
4846        // A lockout is waited out; a rejected credential is not.
4847        let now = TestClock::default().now();
4848        assert_eq!(
4849            RefreshState::LockedOut {
4850                retry_after: Duration::from_secs(60)
4851            }
4852            .retry_delay(now),
4853            Some(Duration::from_secs(60))
4854        );
4855        assert_eq!(RefreshState::Unauthorized.retry_delay(now), None);
4856        assert_eq!(RefreshState::Offline.retry_delay(now), None);
4857    }
4858
4859    /// An empty inventory is an answer, not a failure. An idle host that
4860    /// rendered as broken would be a support ticket a week.
4861    #[test]
4862    fn an_empty_snapshot_is_ready_rather_than_a_failure() {
4863        let snapshot = InventorySnapshot {
4864            target: repo_target(),
4865            runners: RunnerInventory::new(repo_target(), Vec::new()),
4866            activity: ActivityCount::of(repo(), 0),
4867            observed_at: TestClock::default().now(),
4868            headroom: None,
4869        };
4870        let state = RefreshState::from_result(Ok(snapshot));
4871        assert!(state.is_ready());
4872        assert!(state.snapshot().expect("ready").runners.is_empty());
4873        assert_eq!(state.to_string(), "0 runners, 0 in progress");
4874    }
4875}