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