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