use std::{
collections::{BTreeMap, BTreeSet},
fmt,
future::Future,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::Duration,
};
use runner_manager_domain::model::{
Arch, Clock, Org, Os, OwnerRepo, RefreshInterval, ScaleTarget, TargetScope, Timestamp,
};
use serde::Deserialize;
use serde::de::DeserializeOwned;
use tokio::sync::watch;
use crate::{ApiRequest, ApiResponse, AuthenticatedClient, GithubError, MAX_PAGES};
pub const PER_PAGE: u32 = 100;
pub const HOURLY_REQUEST_CEILING: u32 = 5_000;
pub const BUDGET_SHARE_DIVISOR: u32 = 2;
pub const SECONDS_PER_HOUR: u32 = 3_600;
pub const RUNNER_INVENTORY_REQUESTS_PER_REFRESH: u32 = 1;
pub const ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH: u32 = 1;
pub const MAX_ACTIVITY_FALLBACK_PAGES: usize = 4;
const _: () = assert!(
MAX_ACTIVITY_FALLBACK_PAGES < MAX_PAGES,
"the activity page budget must stay below the runaway `Link`-cycle ceiling"
);
const MAX_BENIGN_TOTAL_COUNT_SKEW: u64 = 16;
const _: () = assert!(
MAX_BENIGN_TOTAL_COUNT_SKEW > 0,
"a zero skew re-creates the assert that trips on a run finishing mid-serialisation"
);
pub const DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH: u32 = 2;
pub const DEFAULT_RATE_LIMIT_BACKOFF: Duration = Duration::from_secs(60);
pub const MAX_RATE_LIMIT_BACKOFF: Duration = Duration::from_secs(15 * 60);
#[derive(Debug, Clone, Default)]
pub struct CancelToken {
inner: Arc<CancelInner>,
}
#[derive(Debug)]
struct CancelInner {
tx: watch::Sender<bool>,
}
impl Default for CancelInner {
fn default() -> Self {
Self {
tx: watch::Sender::new(false),
}
}
}
impl CancelToken {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub fn cancel(&self) {
self.inner.tx.send_replace(true);
}
#[must_use]
pub fn is_cancelled(&self) -> bool {
*self.inner.tx.borrow()
}
pub fn check(&self) -> Result<(), InventoryError> {
if self.is_cancelled() {
return Err(InventoryError::Cancelled);
}
Ok(())
}
pub async fn cancelled(&self) {
let mut rx = self.inner.tx.subscribe();
if rx.wait_for(|cancelled| *cancelled).await.is_err() {
std::future::pending::<()>().await;
}
}
pub async fn run<T>(
&self,
work: impl Future<Output = Result<T, InventoryError>>,
) -> Result<T, InventoryError> {
tokio::select! {
biased;
() = self.cancelled() => Err(InventoryError::Cancelled),
result = work => result,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum RateLimitKind {
Primary,
Secondary,
}
impl fmt::Display for RateLimitKind {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::Primary => "primary",
Self::Secondary => "secondary",
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RateLimited {
pub kind: RateLimitKind,
pub retry_after: Option<Duration>,
pub remaining: Option<u64>,
pub reset_unix_secs: Option<u64>,
}
impl RateLimited {
#[must_use]
pub fn detect(error: &GithubError) -> Option<Self> {
let (status, message) = match error {
GithubError::Status {
status, message, ..
} => (*status, message.as_deref()),
GithubError::Forbidden { message, .. } => (403, message.as_deref()),
_ => return None,
};
if !matches!(status, 403 | 429) {
return None;
}
let evidence = error.rate_limit();
let remaining = evidence.and_then(|e| e.remaining);
let says_rate_limit =
message.is_some_and(|m| m.to_ascii_lowercase().contains("rate limit"));
if status != 429 && !says_rate_limit {
return None;
}
let kind = if remaining == Some(0) {
RateLimitKind::Primary
} else {
RateLimitKind::Secondary
};
Some(Self {
kind,
retry_after: error.retry_after(),
remaining,
reset_unix_secs: evidence.and_then(|e| e.reset_unix_secs),
})
}
#[must_use]
pub fn delay_from(&self, now: Timestamp) -> Duration {
let requested = self.retry_after.or_else(|| {
let reset = self.reset_unix_secs?;
let seconds = i64::try_from(reset).ok()? - now.timestamp();
u64::try_from(seconds).ok().map(Duration::from_secs)
});
let requested = match requested {
Some(d) if d > Duration::ZERO => d,
_ => DEFAULT_RATE_LIMIT_BACKOFF,
};
requested.min(MAX_RATE_LIMIT_BACKOFF)
}
}
impl fmt::Display for RateLimited {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "GitHub's {} rate limit is exhausted", self.kind)?;
if let Some(retry_after) = self.retry_after {
write!(
f,
"; it asked to be left alone for {}s",
retry_after.as_secs()
)?;
}
if let Some(remaining) = self.remaining {
write!(f, "; {remaining} requests remain in the hourly quota")?;
}
f.write_str(". Refreshes are delayed, not lost")
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct RateLimitHeadroom {
pub limit: Option<u64>,
pub remaining: Option<u64>,
pub reset_unix_secs: Option<u64>,
}
impl RateLimitHeadroom {
fn from_response(response: &ApiResponse) -> Option<Self> {
let read = |name: &str| {
response
.header(name)
.and_then(|v| v.trim().parse::<u64>().ok())
};
let headroom = Self {
limit: read("x-ratelimit-limit"),
remaining: read("x-ratelimit-remaining"),
reset_unix_secs: read("x-ratelimit-reset"),
};
if headroom == Self::default() {
return None;
}
Some(headroom)
}
}
#[derive(Debug, thiserror::Error)]
pub enum InventoryError {
#[error("{0}")]
RateLimited(RateLimited),
#[error("the refresh was cancelled before it completed")]
Cancelled,
#[error(transparent)]
Github(#[from] GithubError),
}
impl InventoryError {
#[must_use]
pub fn is_rate_limited(&self) -> bool {
matches!(self, Self::RateLimited(_))
}
#[must_use]
pub fn is_cancelled(&self) -> bool {
matches!(self, Self::Cancelled)
}
#[must_use]
pub fn rate_limited(&self) -> Option<&RateLimited> {
match self {
Self::RateLimited(limit) => Some(limit),
_ => None,
}
}
#[must_use]
pub fn is_offline(&self) -> bool {
matches!(self, Self::Github(GithubError::Transport(_)))
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RefreshState {
Ready(Box<InventorySnapshot>),
RateLimited(RateLimited),
Unauthorized,
LockedOut { retry_after: Duration },
Forbidden { message: Option<String> },
Offline,
Failed {
status: Option<u16>,
message: String,
},
Cancelled,
}
impl RefreshState {
#[must_use]
pub fn from_result(result: Result<InventorySnapshot, InventoryError>) -> Self {
match result {
Ok(snapshot) => Self::Ready(Box::new(snapshot)),
Err(error) => Self::from_error(&error),
}
}
#[must_use]
pub fn from_error(error: &InventoryError) -> Self {
match error {
InventoryError::RateLimited(limit) => Self::RateLimited(*limit),
InventoryError::Cancelled => Self::Cancelled,
InventoryError::Github(github) => match github {
GithubError::AuthenticationFailed => Self::Unauthorized,
GithubError::AuthenticationLockout { retry_after } => Self::LockedOut {
retry_after: *retry_after,
},
GithubError::Forbidden { message, .. } => Self::Forbidden {
message: message.clone(),
},
GithubError::Transport(_) => Self::Offline,
GithubError::Status { status, .. } => Self::Failed {
status: Some(*status),
message: github.to_string(),
},
other => Self::Failed {
status: None,
message: other.to_string(),
},
},
}
}
#[must_use]
pub fn is_ready(&self) -> bool {
matches!(self, Self::Ready(_))
}
#[must_use]
pub fn snapshot(&self) -> Option<&InventorySnapshot> {
match self {
Self::Ready(snapshot) => Some(&**snapshot),
_ => None,
}
}
#[must_use]
pub fn retry_delay(&self, now: Timestamp) -> Option<Duration> {
match self {
Self::RateLimited(limit) => Some(limit.delay_from(now)),
Self::LockedOut { retry_after } => Some(*retry_after),
_ => None,
}
}
}
impl fmt::Display for RefreshState {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Ready(snapshot) => write!(
f,
"{} runners, {} in progress",
snapshot.runners.len(),
snapshot.activity.total()
),
Self::RateLimited(limit) => write!(f, "{limit}"),
Self::Unauthorized => f.write_str(
"GitHub rejected the stored credential; run `runner-manager auth login`",
),
Self::LockedOut { retry_after } => write!(
f,
"GitHub has temporarily locked out authentication; retrying in {}s. \
The credential itself is not the problem",
retry_after.as_secs()
),
Self::Forbidden { message } => match message {
Some(message) => write!(f, "GitHub denied the request: {message}"),
None => f.write_str("GitHub denied the request"),
},
Self::Offline => f.write_str("GitHub is unreachable"),
Self::Failed { message, .. } => f.write_str(message),
Self::Cancelled => f.write_str("the refresh was cancelled"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum RunnerStatus {
Online,
Offline,
Other(String),
}
impl RunnerStatus {
#[must_use]
pub fn from_wire(raw: &str) -> Self {
match raw.trim().to_ascii_lowercase().as_str() {
"online" => Self::Online,
"offline" => Self::Offline,
_ => Self::Other(raw.trim().to_string()),
}
}
#[must_use]
pub fn is_online(&self) -> bool {
matches!(self, Self::Online)
}
}
impl fmt::Display for RunnerStatus {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Online => f.write_str("online"),
Self::Offline => f.write_str("offline"),
Self::Other(raw) => f.write_str(raw),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Runner {
pub id: u64,
pub name: String,
pub os: String,
pub status: RunnerStatus,
pub busy: bool,
pub ephemeral: Option<bool>,
pub labels: Vec<String>,
}
impl Runner {
#[must_use]
pub fn has_label(&self, label: &str) -> bool {
self.labels
.iter()
.any(|held| held.eq_ignore_ascii_case(label.trim()))
}
#[must_use]
pub fn parsed_os(&self) -> Option<Os> {
self.os.parse().ok()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunnerInventory {
target: ScaleTarget,
runners: Vec<Runner>,
reported_total: Option<u64>,
pages: usize,
truncated: bool,
}
impl RunnerInventory {
#[must_use]
pub fn new(target: ScaleTarget, runners: Vec<Runner>) -> Self {
let reported_total = Some(u64::try_from(runners.len()).unwrap_or(u64::MAX));
Self {
target,
runners,
reported_total,
pages: 1,
truncated: false,
}
}
#[must_use]
pub fn paged(
target: ScaleTarget,
runners: Vec<Runner>,
reported_total: Option<u64>,
pages: usize,
truncated: bool,
) -> Self {
Self {
target,
runners,
reported_total,
pages,
truncated,
}
}
#[must_use]
pub fn target(&self) -> &ScaleTarget {
&self.target
}
#[must_use]
pub fn runners(&self) -> &[Runner] {
&self.runners
}
#[must_use]
pub fn len(&self) -> usize {
self.runners.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.runners.is_empty()
}
#[must_use]
pub fn busy_count(&self) -> usize {
self.runners.iter().filter(|runner| runner.busy).count()
}
#[must_use]
pub fn online_count(&self) -> usize {
self.runners
.iter()
.filter(|runner| runner.status.is_online())
.count()
}
#[must_use]
pub fn reported_total(&self) -> Option<u64> {
self.reported_total
}
#[must_use]
pub fn pages(&self) -> usize {
self.pages
}
#[must_use]
pub fn truncated(&self) -> bool {
self.truncated
}
#[must_use]
pub fn missing(&self) -> Option<u64> {
let total = self.reported_total?;
let collected = u64::try_from(self.runners.len()).unwrap_or(u64::MAX);
total.checked_sub(collected).filter(|missing| *missing > 0)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ActivityScope {
target: ScaleTarget,
repositories: Vec<OwnerRepo>,
}
impl ActivityScope {
#[must_use]
pub fn repository(repo: OwnerRepo) -> Self {
Self {
target: ScaleTarget::Repository(repo.clone()),
repositories: vec![repo],
}
}
#[must_use]
pub fn organization(org: Org, repositories: impl IntoIterator<Item = OwnerRepo>) -> Self {
Self {
target: ScaleTarget::Organization(org),
repositories: repositories.into_iter().collect(),
}
}
#[must_use]
pub fn target(&self) -> &ScaleTarget {
&self.target
}
#[must_use]
pub fn repositories(&self) -> &[OwnerRepo] {
&self.repositories
}
#[must_use]
pub fn requests_per_refresh(&self) -> u32 {
u32::try_from(self.repositories.len()).unwrap_or(u32::MAX)
* ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ActivityCount {
per_repository: BTreeMap<OwnerRepo, u32>,
unavailable: Vec<UnavailableRepository>,
truncated: BTreeSet<OwnerRepo>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnavailableRepository {
pub repository: OwnerRepo,
pub reason: String,
}
impl ActivityCount {
#[must_use]
pub fn new(per_repository: BTreeMap<OwnerRepo, u32>) -> Self {
Self {
per_repository,
unavailable: Vec::new(),
truncated: BTreeSet::new(),
}
}
#[must_use]
pub fn of(repository: OwnerRepo, count: u32) -> Self {
Self::new(BTreeMap::from([(repository, count)]))
}
#[must_use]
pub fn with_truncated(mut self, repository: OwnerRepo) -> Self {
self.truncated.insert(repository);
self
}
#[must_use]
pub fn with_unavailable(mut self, repository: OwnerRepo, reason: impl Into<String>) -> Self {
self.unavailable.push(UnavailableRepository {
repository,
reason: reason.into(),
});
self
}
#[must_use]
pub fn total(&self) -> u32 {
self.per_repository.values().copied().sum()
}
#[must_use]
pub fn per_repository(&self) -> &BTreeMap<OwnerRepo, u32> {
&self.per_repository
}
#[must_use]
pub fn for_repository(&self, repository: &OwnerRepo) -> Option<u32> {
self.per_repository.get(repository).copied()
}
#[must_use]
pub fn unavailable(&self) -> &[UnavailableRepository] {
&self.unavailable
}
#[must_use]
pub fn truncated(&self) -> &BTreeSet<OwnerRepo> {
&self.truncated
}
#[must_use]
pub fn is_truncated(&self, repository: &OwnerRepo) -> bool {
self.truncated.contains(repository)
}
#[must_use]
pub fn is_complete(&self) -> bool {
self.unavailable.is_empty() && self.truncated.is_empty()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunnerDownload {
pub os: String,
pub architecture: String,
pub download_url: String,
pub filename: String,
pub sha256_checksum: Option<String>,
}
impl RunnerDownload {
#[must_use]
pub fn matches(&self, os: Os, arch: Arch) -> bool {
self.os.parse::<Os>().is_ok_and(|found| found == os)
&& self
.architecture
.parse::<Arch>()
.is_ok_and(|found| found == arch)
}
#[must_use]
pub fn sha256_checksum(&self) -> Option<&str> {
self.sha256_checksum.as_deref()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RunnerDownloads {
entries: Vec<RunnerDownload>,
}
impl RunnerDownloads {
#[must_use]
pub fn new(entries: Vec<RunnerDownload>) -> Self {
Self { entries }
}
#[must_use]
pub fn entries(&self) -> &[RunnerDownload] {
&self.entries
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
#[must_use]
pub fn select(&self, os: Os, arch: Arch) -> Option<&RunnerDownload> {
self.entries.iter().find(|entry| entry.matches(os, arch))
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InventorySnapshot {
pub target: ScaleTarget,
pub runners: RunnerInventory,
pub activity: ActivityCount,
pub observed_at: Timestamp,
pub headroom: Option<RateLimitHeadroom>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TargetCost {
scope: TargetScope,
installed_repositories: u32,
demand_requests_per_repository: u32,
}
impl TargetCost {
#[must_use]
pub const fn repository() -> Self {
Self {
scope: TargetScope::Repository,
installed_repositories: 1,
demand_requests_per_repository: DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH,
}
}
#[must_use]
pub const fn organization(installed_repositories: u32) -> Self {
Self {
scope: TargetScope::Organization,
installed_repositories,
demand_requests_per_repository: DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH,
}
}
#[must_use]
pub fn with_demand_requests_per_repository(mut self, requests: u32) -> Self {
self.demand_requests_per_repository = requests;
self
}
#[must_use]
pub fn from_activity_scope(scope: &ActivityScope) -> Self {
match scope.target().scope() {
TargetScope::Repository => Self::repository(),
TargetScope::Organization => {
Self::organization(u32::try_from(scope.repositories().len()).unwrap_or(u32::MAX))
}
}
}
#[must_use]
pub const fn scope(&self) -> TargetScope {
self.scope
}
#[must_use]
pub const fn installed_repositories(&self) -> u32 {
self.installed_repositories
}
#[must_use]
pub const fn requests_per_refresh(&self) -> u32 {
let repositories = match self.scope {
TargetScope::Repository => 1,
TargetScope::Organization => self.installed_repositories,
};
RUNNER_INVENTORY_REQUESTS_PER_REFRESH
+ repositories.saturating_mul(
ACTIVITY_REQUESTS_PER_REPOSITORY_PER_REFRESH + self.demand_requests_per_repository,
)
}
#[must_use]
pub fn requests_per_hour(&self, interval: RefreshInterval) -> u32 {
self.requests_per_refresh()
.saturating_mul(refreshes_per_hour(interval))
}
}
#[must_use]
pub fn refreshes_per_hour(interval: RefreshInterval) -> u32 {
SECONDS_PER_HOUR / u32::from(interval.as_secs())
}
#[must_use]
pub const fn budget_allowance() -> u32 {
HOURLY_REQUEST_CEILING / BUDGET_SHARE_DIVISOR
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BudgetProjection {
interval: RefreshInterval,
targets: Vec<TargetCost>,
}
impl BudgetProjection {
#[must_use]
pub fn new(interval: RefreshInterval, targets: impl IntoIterator<Item = TargetCost>) -> Self {
Self {
interval,
targets: targets.into_iter().collect(),
}
}
#[must_use]
pub fn interval(&self) -> RefreshInterval {
self.interval
}
#[must_use]
pub fn targets(&self) -> &[TargetCost] {
&self.targets
}
#[must_use]
pub fn refreshes_per_hour(&self) -> u32 {
refreshes_per_hour(self.interval)
}
#[must_use]
pub fn requests_per_hour(&self) -> u32 {
self.targets
.iter()
.map(|target| target.requests_per_hour(self.interval))
.fold(0, u32::saturating_add)
}
#[must_use]
pub fn ceiling(&self) -> u32 {
HOURLY_REQUEST_CEILING
}
#[must_use]
pub fn allowance(&self) -> u32 {
budget_allowance()
}
#[must_use]
pub fn headroom(&self) -> u32 {
self.allowance().saturating_sub(self.requests_per_hour())
}
#[must_use]
pub fn exceeds_allowance(&self) -> bool {
self.requests_per_hour() > self.allowance()
}
#[must_use]
pub fn max_repository_targets(interval: RefreshInterval) -> u32 {
let per_target = TargetCost::repository().requests_per_hour(interval);
if per_target == 0 {
return 0;
}
budget_allowance() / per_target
}
#[must_use]
pub fn admit(&self, candidate: TargetCost) -> Admission {
let candidate_per_hour = candidate.requests_per_hour(self.interval);
let projected = self.requests_per_hour().saturating_add(candidate_per_hour);
let allowance = self.allowance();
if projected > allowance {
return Admission::Refused {
candidate,
candidate_requests_per_hour: candidate_per_hour,
projected_requests_per_hour: projected,
allowance,
ceiling: self.ceiling(),
interval: self.interval,
max_repository_targets: Self::max_repository_targets(self.interval),
};
}
Admission::Admitted {
projected_requests_per_hour: projected,
headroom_after: allowance - projected,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Admission {
Admitted {
projected_requests_per_hour: u32,
headroom_after: u32,
},
Refused {
candidate: TargetCost,
candidate_requests_per_hour: u32,
projected_requests_per_hour: u32,
allowance: u32,
ceiling: u32,
interval: RefreshInterval,
max_repository_targets: u32,
},
}
impl Admission {
#[must_use]
pub fn is_admitted(&self) -> bool {
matches!(self, Self::Admitted { .. })
}
}
impl fmt::Display for Admission {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Admitted {
projected_requests_per_hour,
headroom_after,
} => write!(
f,
"projected {projected_requests_per_hour} requests/hour, \
{headroom_after} remaining in this host's share of the budget"
),
Self::Refused {
candidate,
candidate_requests_per_hour,
projected_requests_per_hour,
allowance,
ceiling,
interval,
max_repository_targets,
} => {
write!(
f,
"refused: this target would take the host to \
{projected_requests_per_hour} requests/hour, over the {allowance} it may \
plan to spend (half of GitHub's {ceiling}/hour ceiling) at a \
{}-second refresh interval. This host can serve about \
{max_repository_targets} repository targets at that interval",
interval.as_secs()
)?;
if candidate.scope() == TargetScope::Organization {
write!(
f,
". This organization alone costs {candidate_requests_per_hour} \
requests/hour because the App is installed on {} of its repositories, \
and workflow runs are a per-repository resource",
candidate.installed_repositories()
)?;
}
Ok(())
}
}
}
}
#[derive(Debug)]
pub struct RefreshCoalescer<T> {
generation: AtomicU64,
gate: tokio::sync::Mutex<()>,
last: std::sync::Mutex<Option<T>>,
performed: AtomicU64,
joined: AtomicU64,
}
impl<T: Clone> Default for RefreshCoalescer<T> {
fn default() -> Self {
Self::new()
}
}
impl<T: Clone> RefreshCoalescer<T> {
#[must_use]
pub fn new() -> Self {
Self {
generation: AtomicU64::new(0),
gate: tokio::sync::Mutex::new(()),
last: std::sync::Mutex::new(None),
performed: AtomicU64::new(0),
joined: AtomicU64::new(0),
}
}
pub async fn refresh<F, Fut>(&self, work: F) -> T
where
F: FnOnce() -> Fut,
Fut: Future<Output = T>,
{
let sampled = self.generation.load(Ordering::SeqCst);
let _guard = self.gate.lock().await;
if self.generation.load(Ordering::SeqCst) != sampled
&& let Some(shared) = self.last.lock().expect("refresh lock poisoned").clone()
{
self.joined.fetch_add(1, Ordering::SeqCst);
tracing::debug!("joined an in-flight refresh instead of issuing a second request");
return shared;
}
let outcome = work().await;
*self.last.lock().expect("refresh lock poisoned") = Some(outcome.clone());
self.performed.fetch_add(1, Ordering::SeqCst);
self.generation.fetch_add(1, Ordering::SeqCst);
outcome
}
#[must_use]
pub fn performed(&self) -> u64 {
self.performed.load(Ordering::SeqCst)
}
#[must_use]
pub fn joined(&self) -> u64 {
self.joined.load(Ordering::SeqCst)
}
#[must_use]
pub fn last(&self) -> Option<T> {
self.last.lock().expect("refresh lock poisoned").clone()
}
}
#[async_trait::async_trait]
pub trait InventoryGateway: fmt::Debug + Send + Sync {
async fn list_runners(
&self,
target: &ScaleTarget,
cancel: &CancelToken,
) -> Result<RunnerInventory, InventoryError>;
async fn remove_runner(
&self,
target: &ScaleTarget,
runner_id: u64,
cancel: &CancelToken,
) -> Result<(), InventoryError>;
async fn in_progress_activity(
&self,
scope: &ActivityScope,
cancel: &CancelToken,
) -> Result<ActivityCount, InventoryError>;
async fn runner_downloads(
&self,
target: &ScaleTarget,
cancel: &CancelToken,
) -> Result<RunnerDownloads, InventoryError>;
fn headroom(&self) -> Option<RateLimitHeadroom>;
fn now(&self) -> Timestamp;
async fn snapshot(
&self,
scope: &ActivityScope,
cancel: &CancelToken,
) -> Result<InventorySnapshot, InventoryError> {
let runners = self.list_runners(scope.target(), cancel).await?;
let activity = self.in_progress_activity(scope, cancel).await?;
Ok(InventorySnapshot {
target: scope.target().clone(),
runners,
activity,
observed_at: self.now(),
headroom: self.headroom(),
})
}
}
#[derive(Debug)]
struct RateLimitState {
until: Option<Timestamp>,
last: Option<RateLimited>,
}
pub struct RestInventory {
client: Arc<AuthenticatedClient>,
clock: Arc<dyn Clock>,
rate_limit: std::sync::Mutex<RateLimitState>,
headroom: std::sync::Mutex<Option<RateLimitHeadroom>>,
requests_issued: AtomicU64,
}
impl fmt::Debug for RestInventory {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let backing_off = match self.rate_limit.try_lock() {
Ok(state) => state
.until
.is_some_and(|until| self.clock.now() < until)
.to_string(),
Err(_) => "unknown (the rate-limit state is being updated)".to_string(),
};
f.debug_struct("RestInventory")
.field(
"requests_issued",
&self.requests_issued.load(Ordering::Relaxed),
)
.field("rate_limited", &backing_off)
.finish_non_exhaustive()
}
}
impl RestInventory {
#[must_use]
pub fn new(client: Arc<AuthenticatedClient>, clock: Arc<dyn Clock>) -> Self {
Self {
client,
clock,
rate_limit: std::sync::Mutex::new(RateLimitState {
until: None,
last: None,
}),
headroom: std::sync::Mutex::new(None),
requests_issued: AtomicU64::new(0),
}
}
#[must_use]
pub fn requests_issued(&self) -> u64 {
self.requests_issued.load(Ordering::SeqCst)
}
#[must_use]
pub fn rate_limit_backoff(&self) -> Option<Duration> {
let state = self.rate_limit.lock().expect("rate-limit lock poisoned");
let until = state.until?;
let now = self.clock.now();
if now >= until {
return None;
}
(until - now).to_std().ok()
}
#[must_use]
pub fn rate_limit_state(&self) -> Option<RateLimited> {
let remaining = self.rate_limit_backoff()?;
let state = self.rate_limit.lock().expect("rate-limit lock poisoned");
let mut limit = state.last?;
limit.retry_after = Some(remaining);
Some(limit)
}
pub fn clear_rate_limit(&self) {
self.rate_limit
.lock()
.expect("rate-limit lock poisoned")
.until = None;
}
async fn issue(
&self,
request: &ApiRequest,
cancel: &CancelToken,
) -> Result<ApiResponse, InventoryError> {
cancel.check()?;
if let Some(limit) = self.rate_limit_state() {
tracing::debug!(
method = request.method().as_str(),
path = %request.path(),
remaining_secs = limit.retry_after.unwrap_or_default().as_secs(),
"suppressed a request: GitHub's rate limit is still backing off"
);
return Err(InventoryError::RateLimited(limit));
}
let result = cancel
.run(async {
self.requests_issued.fetch_add(1, Ordering::SeqCst);
self.client
.send(request)
.await
.map_err(InventoryError::from)
})
.await;
match result {
Ok(response) => {
if let Some(headroom) = RateLimitHeadroom::from_response(&response) {
*self.headroom.lock().expect("headroom lock poisoned") = Some(headroom);
}
Ok(response)
}
Err(InventoryError::Github(error)) => Err(self.classify(error)),
Err(other) => Err(other),
}
}
fn classify(&self, error: GithubError) -> InventoryError {
let Some(limit) = RateLimited::detect(&error) else {
return InventoryError::Github(error);
};
let now = self.clock.now();
let delay = limit.delay_from(now);
if let Ok(delta) = chrono::TimeDelta::from_std(delay) {
let mut state = self.rate_limit.lock().expect("rate-limit lock poisoned");
state.until = Some(now + delta);
state.last = Some(limit);
}
tracing::warn!(
kind = %limit.kind,
delay_secs = delay.as_secs(),
remaining = limit.remaining,
"GitHub is rate limiting this credential; delaying refreshes and reporting it"
);
InventoryError::RateLimited(limit)
}
async fn collect_pages<P: WirePage>(
&self,
first: ApiRequest,
cancel: &CancelToken,
) -> Result<Collected<P::Item>, InventoryError> {
let mut items = Vec::new();
let mut reported_total = None;
let mut pages = 0_usize;
let mut truncated = false;
let mut next = Some(first);
while let Some(request) = next.take() {
let response = self.issue(&request, cancel).await?;
let page: P = response.json()?;
reported_total = page.reported_total().or(reported_total);
items.extend(page.into_items());
pages += 1;
if pages >= MAX_PAGES {
truncated = true;
tracing::warn!(
what = P::WHAT,
pages,
collected = items.len(),
"stopped following pages at the ceiling; a `Link: rel=next` that never \
ends would otherwise loop forever"
);
break;
}
next = response
.next_page()
.map(|url| ApiRequest::get(url.as_str()));
}
Ok(Collected {
items,
reported_total,
pages,
truncated,
})
}
async fn repository_in_progress(
&self,
repository: &OwnerRepo,
cancel: &CancelToken,
) -> Result<RepositoryActivity, InventoryError> {
let request = ApiRequest::get(format!(
"/repos/{}/{}/actions/runs",
repository.owner(),
repository.repo()
))
.query("status", "in_progress")
.query("per_page", PER_PAGE);
let response = self.issue(&request, cancel).await?;
let page: RunsPage = response.json()?;
if let Some(total) = page.total_count {
let listed = page.workflow_runs.len() as u64;
if response.next_page().is_none() && total != listed {
tracing::warn!(
repository = %repository,
total_count = total,
listed,
"GitHub's `total_count` disagrees with the single page it sent for a \
filtered query; this layer reads `total_count` as the count of the \
filtered set, and that reading looks wrong"
);
debug_assert!(
total <= listed.saturating_add(MAX_BENIGN_TOTAL_COUNT_SKEW),
"`total_count` ({total}) exceeds the {listed} run(s) on the only page \
of a filtered query by more than {MAX_BENIGN_TOTAL_COUNT_SKEW}, which \
is far past the run-finishing-mid-serialisation race; `total_count` \
is not the filtered count, and every in-progress figure — and `c4`'s \
demand — is being read off the wrong field"
);
}
return Ok(RepositoryActivity::from_reported_total(total, repository));
}
let mut counted = page.workflow_runs.len();
let mut pages = 1_usize;
let mut next = response
.next_page()
.map(|url| ApiRequest::get(url.as_str()));
while let Some(request) = next.take() {
if pages >= MAX_ACTIVITY_FALLBACK_PAGES {
tracing::warn!(
repository = %repository,
pages,
counted,
"stopped counting in-progress runs at the activity page budget; the \
count reported for this repository is a floor, not a total"
);
return Ok(RepositoryActivity::floor(counted));
}
let response = self.issue(&request, cancel).await?;
let page: RunsPage = response.json()?;
counted += page.workflow_runs.len();
pages += 1;
next = response
.next_page()
.map(|url| ApiRequest::get(url.as_str()));
}
Ok(RepositoryActivity::exact(counted))
}
fn runners_path(target: &ScaleTarget) -> String {
match target {
ScaleTarget::Repository(repo) => {
format!("/repos/{}/{}/actions/runners", repo.owner(), repo.repo())
}
ScaleTarget::Organization(org) => format!("/orgs/{}/actions/runners", org.as_str()),
}
}
}
fn is_repository_local_failure(error: &InventoryError) -> bool {
match error {
InventoryError::Github(GithubError::Forbidden { .. }) => true,
InventoryError::Github(GithubError::Status { status, .. }) => *status == 404,
_ => false,
}
}
#[async_trait::async_trait]
impl InventoryGateway for RestInventory {
async fn list_runners(
&self,
target: &ScaleTarget,
cancel: &CancelToken,
) -> Result<RunnerInventory, InventoryError> {
let request = ApiRequest::get(Self::runners_path(target)).query("per_page", PER_PAGE);
let collected = self.collect_pages::<RunnersPage>(request, cancel).await?;
let runners: Vec<Runner> = collected.items.into_iter().map(Runner::from).collect();
let inventory = RunnerInventory::paged(
target.clone(),
runners,
collected.reported_total,
collected.pages,
collected.truncated,
);
if let Some(missing) = inventory.missing() {
tracing::warn!(
target = %target,
missing,
collected = inventory.len(),
"GitHub reported more runners than pagination collected; this inventory is \
incomplete"
);
}
Ok(inventory)
}
async fn remove_runner(
&self,
target: &ScaleTarget,
runner_id: u64,
cancel: &CancelToken,
) -> Result<(), InventoryError> {
let path = format!("{}/{runner_id}", Self::runners_path(target));
match self.issue(&ApiRequest::delete(path), cancel).await {
Ok(_) => Ok(()),
Err(InventoryError::Github(GithubError::Status { status: 404, .. })) => Ok(()),
Err(error) => Err(error),
}
}
async fn in_progress_activity(
&self,
scope: &ActivityScope,
cancel: &CancelToken,
) -> Result<ActivityCount, InventoryError> {
let mut per_repository = BTreeMap::new();
let mut unavailable = Vec::new();
let mut truncated = BTreeSet::new();
let aggregating = scope.target().scope() == TargetScope::Organization;
for repository in scope.repositories() {
match self.repository_in_progress(repository, cancel).await {
Ok(activity) => {
per_repository.insert(repository.clone(), activity.count);
if !activity.exact {
truncated.insert(repository.clone());
}
}
Err(error) if aggregating && is_repository_local_failure(&error) => {
tracing::warn!(
repository = %repository,
error = %error,
"a repository in this organization could not be counted; the aggregate \
reports it as unavailable rather than as zero"
);
unavailable.push(UnavailableRepository {
repository: repository.clone(),
reason: error.to_string(),
});
}
Err(error) => return Err(error),
}
}
Ok(ActivityCount {
per_repository,
unavailable,
truncated,
})
}
async fn runner_downloads(
&self,
target: &ScaleTarget,
cancel: &CancelToken,
) -> Result<RunnerDownloads, InventoryError> {
let path = match target {
ScaleTarget::Repository(repo) => format!(
"/repos/{}/{}/actions/runners/downloads",
repo.owner(),
repo.repo()
),
ScaleTarget::Organization(org) => {
format!("/orgs/{}/actions/runners/downloads", org.as_str())
}
};
let response = self.issue(&ApiRequest::get(path), cancel).await?;
let raw: Vec<RawDownload> = response.json()?;
Ok(RunnerDownloads::new(
raw.into_iter().map(RunnerDownload::from).collect(),
))
}
fn headroom(&self) -> Option<RateLimitHeadroom> {
*self.headroom.lock().expect("headroom lock poisoned")
}
fn now(&self) -> Timestamp {
self.clock.now()
}
}
struct Collected<T> {
items: Vec<T>,
reported_total: Option<u64>,
pages: usize,
truncated: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct RepositoryActivity {
count: u32,
exact: bool,
}
impl RepositoryActivity {
fn exact(count: usize) -> Self {
Self {
count: u32::try_from(count).unwrap_or(u32::MAX),
exact: true,
}
}
fn floor(count: usize) -> Self {
Self {
count: u32::try_from(count).unwrap_or(u32::MAX),
exact: false,
}
}
fn from_reported_total(total: u64, repository: &OwnerRepo) -> Self {
match u32::try_from(total) {
Ok(count) => Self { count, exact: true },
Err(_) => {
tracing::warn!(
repository = %repository,
total_count = total,
"GitHub reported an in-progress total wider than this product renders; \
it is clamped and reported as a floor rather than as a count"
);
Self {
count: u32::MAX,
exact: false,
}
}
}
}
}
trait WirePage: DeserializeOwned {
type Item;
const WHAT: &'static str;
fn reported_total(&self) -> Option<u64>;
fn into_items(self) -> Vec<Self::Item>;
}
#[derive(Debug, Deserialize)]
struct RunnersPage {
total_count: Option<u64>,
#[serde(default)]
runners: Vec<RawRunner>,
}
impl WirePage for RunnersPage {
type Item = RawRunner;
const WHAT: &'static str = "runners";
fn reported_total(&self) -> Option<u64> {
self.total_count
}
fn into_items(self) -> Vec<Self::Item> {
self.runners
}
}
#[derive(Debug, Deserialize)]
struct RawRunner {
id: u64,
#[serde(default)]
name: String,
#[serde(default)]
os: String,
status: String,
busy: bool,
ephemeral: Option<bool>,
#[serde(default)]
labels: Vec<RawLabel>,
}
#[derive(Debug, Deserialize)]
struct RawLabel {
name: String,
}
impl From<RawRunner> for Runner {
fn from(raw: RawRunner) -> Self {
Self {
id: raw.id,
name: raw.name,
os: raw.os,
status: RunnerStatus::from_wire(&raw.status),
busy: raw.busy,
ephemeral: raw.ephemeral,
labels: raw.labels.into_iter().map(|label| label.name).collect(),
}
}
}
#[derive(Debug, Deserialize)]
struct RunsPage {
total_count: Option<u64>,
#[serde(default)]
workflow_runs: Vec<serde::de::IgnoredAny>,
}
#[derive(Debug, Deserialize)]
struct RawDownload {
#[serde(default)]
os: String,
#[serde(default)]
architecture: String,
#[serde(default)]
download_url: String,
#[serde(default)]
filename: String,
sha256_checksum: Option<String>,
}
impl From<RawDownload> for RunnerDownload {
fn from(raw: RawDownload) -> Self {
Self {
os: raw.os,
architecture: raw.architecture,
download_url: raw.download_url,
filename: raw.filename,
sha256_checksum: raw.sha256_checksum,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::testing::{FIXTURE_TOKEN, Script, TestClock};
use crate::{Endpoints, UserAccessToken};
use secrecy::SecretString;
use serde_json::{Value, json};
use wiremock::{
Mock, MockServer, ResponseTemplate,
matchers::{method, path, query_param},
};
fn repo() -> OwnerRepo {
OwnerRepo::parse("octo/dashboard").expect("a valid owner/repo")
}
fn other_repo() -> OwnerRepo {
OwnerRepo::parse("octo/api").expect("a valid owner/repo")
}
fn third_repo() -> OwnerRepo {
OwnerRepo::parse("octo/docs").expect("a valid owner/repo")
}
fn repo_target() -> ScaleTarget {
ScaleTarget::Repository(repo())
}
fn org_target() -> ScaleTarget {
ScaleTarget::organization("octo-org").expect("a valid organization login")
}
const REPO_RUNNERS: &str = "/repos/octo/dashboard/actions/runners";
const ORG_RUNNERS: &str = "/orgs/octo-org/actions/runners";
const REPO_RUNS: &str = "/repos/octo/dashboard/actions/runs";
fn runners_path(target: &ScaleTarget) -> &'static str {
match target {
ScaleTarget::Repository(_) => REPO_RUNNERS,
ScaleTarget::Organization(_) => ORG_RUNNERS,
}
}
fn gateway(server: &MockServer, clock: Arc<TestClock>) -> RestInventory {
let client = AuthenticatedClient::new(
Endpoints::for_test_server(&server.uri()).expect("a valid test base"),
UserAccessToken::new(SecretString::from(FIXTURE_TOKEN)),
clock.clone(),
)
.expect("the HTTP client builds");
RestInventory::new(Arc::new(client), clock)
}
fn runner_page(ids: std::ops::Range<u64>, total: u64) -> Value {
let runners: Vec<Value> = ids
.map(|id| {
json!({
"id": id,
"name": format!("runner-{id:04}"),
"os": "win",
"status": "online",
"busy": false,
"ephemeral": true,
"labels": [{ "id": 1, "name": "rm-home-win-x64", "type": "read-only" }]
})
})
.collect();
json!({ "total_count": total, "runners": runners })
}
fn link_next(url: &str) -> String {
format!("<{url}>; rel=\"next\"")
}
async fn requests_seen(server: &MockServer) -> usize {
server
.received_requests()
.await
.expect("the mock server records requests")
.len()
}
#[tokio::test]
async fn a_multi_page_runner_inventory_returns_every_runner_at_both_scopes() {
for target in [repo_target(), org_target()] {
let server = MockServer::start().await;
let first = runners_path(&target);
let page_two = format!("{}/page/2", server.uri());
let page_three = format!("{}/page/3", server.uri());
Mock::given(method("GET"))
.and(path(first))
.respond_with(
ResponseTemplate::new(200)
.insert_header("link", link_next(&page_two).as_str())
.set_body_json(runner_page(1..101, 250)),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/page/2"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("link", link_next(&page_three).as_str())
.set_body_json(runner_page(101..201, 250)),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/page/3"))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(201..251, 250)))
.expect(1)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let inventory = gateway
.list_runners(&target, &CancelToken::new())
.await
.expect("three pages are readable");
assert_eq!(
inventory.len(),
250,
"{target}: a first page is not a complete inventory"
);
assert_eq!(inventory.pages(), 3, "{target}");
assert_eq!(inventory.reported_total(), Some(250), "{target}");
assert_eq!(
inventory.missing(),
None,
"{target}: pagination collected everything GitHub said existed"
);
assert!(!inventory.truncated(), "{target}");
assert_eq!(inventory.runners()[0].id, 1, "{target}");
assert_eq!(inventory.runners()[249].id, 250, "{target}");
assert_eq!(gateway.requests_issued(), 3, "{target}");
}
}
#[tokio::test]
async fn a_next_page_url_containing_a_comma_does_not_truncate_the_inventory() {
let server = MockServer::start().await;
let page_two = format!("{}/page/2?labels=self-hosted,windows", server.uri());
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.insert_header("link", link_next(&page_two).as_str())
.set_body_json(runner_page(1..101, 150)),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/page/2"))
.and(query_param("labels", "self-hosted,windows"))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..151, 150)))
.expect(1)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let inventory = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect("both pages are readable");
assert_eq!(inventory.len(), 150, "the comma ended pagination at page 1");
assert_eq!(inventory.pages(), 2);
}
#[tokio::test]
async fn removing_a_runner_deletes_that_id_and_treats_an_absent_one_as_done() {
let server = MockServer::start().await;
Mock::given(method("DELETE"))
.and(path(format!("{REPO_RUNNERS}/73")))
.respond_with(ResponseTemplate::new(204))
.expect(1)
.mount(&server)
.await;
Mock::given(method("DELETE"))
.and(path(format!("{REPO_RUNNERS}/99")))
.respond_with(ResponseTemplate::new(404).set_body_json(serde_json::json!({
"message": "Not Found"
})))
.expect(1)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
gateway
.remove_runner(&repo_target(), 73, &CancelToken::new())
.await
.expect("a registration this agent owns is deletable");
gateway
.remove_runner(&repo_target(), 99, &CancelToken::new())
.await
.expect(
"already gone is the postcondition asked for; failing here would strand every \
attempt GitHub retired on its own",
);
}
#[tokio::test]
async fn removing_an_organization_runner_uses_the_organization_path() {
let server = MockServer::start().await;
Mock::given(method("DELETE"))
.and(path(format!("{ORG_RUNNERS}/12")))
.respond_with(ResponseTemplate::new(204))
.expect(1)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
gateway
.remove_runner(&org_target(), 12, &CancelToken::new())
.await
.expect("the organization scope deletes under its own path");
}
#[tokio::test]
async fn an_inventory_shorter_than_the_reported_total_says_how_short() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(1..11, 40)))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let inventory = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect("one page is readable");
assert_eq!(inventory.len(), 10);
assert_eq!(
inventory.missing(),
Some(30),
"GitHub said 40 and pagination found 10; a caller has to be able to see that"
);
}
#[tokio::test]
async fn a_self_referential_next_link_stops_at_the_page_ceiling() {
let server = MockServer::start().await;
let itself = format!("{}{}", server.uri(), REPO_RUNNERS);
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.insert_header("link", link_next(&itself).as_str())
.set_body_json(runner_page(1..2, 1)),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let inventory = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect("the walk terminates");
assert_eq!(inventory.pages(), MAX_PAGES);
assert!(
inventory.truncated(),
"a truncated walk must say so, or it reads as a complete inventory"
);
assert_eq!(gateway.requests_issued() as usize, MAX_PAGES);
}
fn runs_body(total: Option<u64>, listed: usize) -> Value {
let runs: Vec<Value> = (0..listed)
.map(|i| json!({ "id": i + 1, "status": "in_progress" }))
.collect();
match total {
Some(total) => json!({ "total_count": total, "workflow_runs": runs }),
None => json!({ "workflow_runs": runs }),
}
}
fn mount_runs(repository: &OwnerRepo, total: u64) -> Mock {
assert!(
total <= u64::from(PER_PAGE),
"a single-page fixture cannot hold {total} runs; a larger one needs a \
`Link: rel=next` and a second page, or it is claiming a total the page \
does not support"
);
let listed = usize::try_from(total).expect("a fixture total fits a usize");
Mock::given(method("GET"))
.and(path(format!(
"/repos/{}/{}/actions/runs",
repository.owner(),
repository.repo()
)))
.and(query_param("status", "in_progress"))
.respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(total), listed)))
}
#[tokio::test]
async fn a_repository_activity_count_is_one_request_and_reads_the_reported_total() {
let server = MockServer::start().await;
mount_runs(&repo(), 7).expect(1).mount(&server).await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::repository(repo());
let activity = gateway
.in_progress_activity(&scope, &CancelToken::new())
.await
.expect("the count is readable");
assert_eq!(activity.total(), 7);
assert_eq!(activity.for_repository(&repo()), Some(7));
assert!(activity.is_complete());
assert_eq!(
gateway.requests_issued(),
1,
"reading `total_count` is what keeps this at the one request the budget \
table projects"
);
}
#[tokio::test]
async fn an_organization_activity_count_aggregates_across_installed_repositories() {
let server = MockServer::start().await;
mount_runs(&repo(), 4).mount(&server).await;
mount_runs(&other_repo(), 9).mount(&server).await;
mount_runs(&third_repo(), 0).mount(&server).await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::organization(
Org::new("octo-org").expect("a valid organization login"),
[repo(), other_repo(), third_repo()],
);
let activity = gateway
.in_progress_activity(&scope, &CancelToken::new())
.await
.expect("every repository answers");
assert_eq!(activity.total(), 13);
assert_eq!(activity.for_repository(&repo()), Some(4));
assert_eq!(activity.for_repository(&other_repo()), Some(9));
assert_eq!(
activity.for_repository(&third_repo()),
Some(0),
"a repository with no in-progress runs is a zero, not an absence"
);
assert_eq!(
gateway.requests_issued(),
3,
"one request per installed repository: this is the cost the budget model \
projects and the reason an organization is not a flat per-target constant"
);
}
#[tokio::test]
async fn the_in_progress_count_and_the_busy_runner_count_are_distinct() {
let server = MockServer::start().await;
let runners = json!({
"total_count": 5,
"runners": (1..=5).map(|id| json!({
"id": id,
"name": format!("runner-{id}"),
"os": "win",
"status": "online",
"busy": id <= 3,
"ephemeral": true,
"labels": []
})).collect::<Vec<_>>()
});
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(ResponseTemplate::new(200).set_body_json(runners))
.mount(&server)
.await;
mount_runs(&repo(), 7).mount(&server).await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::repository(repo());
let snapshot = gateway
.snapshot(&scope, &CancelToken::new())
.await
.expect("both read models are readable");
assert_eq!(snapshot.runners.len(), 5);
assert_eq!(snapshot.runners.busy_count(), 3);
assert_eq!(snapshot.runners.online_count(), 5);
assert_eq!(snapshot.activity.total(), 7);
assert_ne!(
u32::try_from(snapshot.runners.busy_count()).unwrap(),
snapshot.activity.total(),
"a workflow run is not a busy runner; `g2` renders them as separate \
aggregates and cannot do that if this layer conflates them"
);
assert_eq!(snapshot.target, repo_target());
assert_eq!(snapshot.observed_at, TestClock::default().now());
}
#[tokio::test]
async fn an_activity_count_without_a_reported_total_counts_the_runs_instead() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNS))
.respond_with(ResponseTemplate::new(200).set_body_json(runs_body(None, 4)))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let activity = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect("the runs are countable");
assert_eq!(
activity.total(),
4,
"a missing `total_count` must not read as an idle repository"
);
assert!(
activity.is_complete(),
"a fallback that reached the end of the pages counted everything"
);
assert!(activity.truncated().is_empty());
}
async fn mount_endless_runs_pages(server: &MockServer, first_path: &str, loop_path: &str) {
let body = || runs_body(None, usize::try_from(PER_PAGE).expect("PER_PAGE fits"));
let onward = link_next(&format!("{}{loop_path}", server.uri()));
for at in [first_path, loop_path] {
Mock::given(method("GET"))
.and(path(at.to_owned()))
.respond_with(
ResponseTemplate::new(200)
.insert_header("link", onward.as_str())
.set_body_json(body()),
)
.mount(server)
.await;
}
}
#[tokio::test]
async fn the_activity_fallback_stops_at_the_budget_not_at_the_runaway_ceiling() {
let server = MockServer::start().await;
mount_endless_runs_pages(&server, REPO_RUNS, "/runs/onward").await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let activity = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect("the walk ends at the budget rather than erroring");
assert_eq!(
gateway.requests_issued() as usize,
MAX_ACTIVITY_FALLBACK_PAGES,
"the walk spends its page budget and not one request more; `MAX_PAGES` \
here would be {MAX_PAGES} requests for one repository's count, per refresh"
);
assert_eq!(
activity.total(),
PER_PAGE * u32::try_from(MAX_ACTIVITY_FALLBACK_PAGES).unwrap(),
"what it did count, it counted"
);
}
#[tokio::test]
async fn a_count_the_page_budget_cut_short_is_reported_as_a_floor_not_as_a_total() {
let server = MockServer::start().await;
mount_endless_runs_pages(&server, REPO_RUNS, "/runs/onward").await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let activity = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect("a truncated count is an answer, not a failure");
assert!(
activity.is_truncated(&repo()),
"the repository whose walk was cut short has to be named"
);
assert_eq!(activity.truncated().len(), 1);
assert!(
!activity.is_complete(),
"a partial answer is usable only when it says it is partial -- and a caller \
asking the one obvious question must hear about truncation, not only about \
repositories that failed outright"
);
assert!(
activity.unavailable().is_empty(),
"truncated is not unavailable: this repository answered, the answer is a floor"
);
}
#[tokio::test]
async fn one_truncated_repository_makes_the_whole_aggregate_a_floor() {
let server = MockServer::start().await;
mount_runs(&repo(), 4).mount(&server).await;
mount_endless_runs_pages(&server, "/repos/octo/api/actions/runs", "/api-runs/onward").await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::organization(
Org::new("octo-org").expect("a valid organization login"),
[repo(), other_repo()],
);
let activity = gateway
.in_progress_activity(&scope, &CancelToken::new())
.await
.expect("the aggregate completes");
assert!(activity.is_truncated(&other_repo()));
assert!(
!activity.is_truncated(&repo()),
"the repository that answered exactly is not tarred with it"
);
assert!(
!activity.is_complete(),
"one floor in the sum makes the sum a floor"
);
assert_eq!(
activity.total(),
4 + PER_PAGE * u32::try_from(MAX_ACTIVITY_FALLBACK_PAGES).unwrap()
);
}
#[cfg(debug_assertions)]
#[tokio::test]
#[should_panic(expected = "is not the filtered count")]
async fn a_total_count_that_disagrees_with_its_only_page_is_caught() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNS))
.respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(5_000), 3)))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let _ = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await;
}
#[cfg(not(debug_assertions))]
#[tokio::test]
async fn a_total_count_that_disagrees_with_its_only_page_still_answers_in_release() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNS))
.respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(5_000), 3)))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let activity = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect("a suspect total is still an answer in release");
assert_eq!(activity.total(), 5_000);
assert_eq!(
gateway.requests_issued(),
1,
"noticing the disagreement must stay free"
);
}
#[cfg(debug_assertions)]
#[tokio::test]
async fn a_total_count_one_over_its_only_page_is_the_documented_race_not_a_panic() {
let listed = 3_usize;
let total = 4_u64;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNS))
.respond_with(ResponseTemplate::new(200).set_body_json(runs_body(Some(total), listed)))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let activity = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect("the documented race is an answer, not a panic");
assert_eq!(
activity.total(),
u32::try_from(total).expect("the fixture total fits"),
"the reported total is still what the layer reads"
);
assert_eq!(
gateway.requests_issued(),
1,
"and noticing the skew must stay free"
);
}
#[tokio::test]
async fn a_total_count_larger_than_a_page_is_not_a_contradiction() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNS))
.respond_with(
ResponseTemplate::new(200)
.insert_header(
"link",
link_next(&format!("{}/page/2", server.uri())).as_str(),
)
.set_body_json(runs_body(
Some(250),
usize::try_from(PER_PAGE).expect("PER_PAGE fits"),
)),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let activity = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect("a paginated total is exactly what `total_count` is for");
assert_eq!(activity.total(), 250);
assert!(activity.is_complete());
assert_eq!(
gateway.requests_issued(),
1,
"reading the reported total is what keeps a 250-run repository at one \
request; the check must not have provoked a second"
);
}
#[tokio::test]
async fn a_repository_that_cannot_be_counted_is_reported_as_unavailable_not_as_zero() {
let server = MockServer::start().await;
mount_runs(&repo(), 6).mount(&server).await;
Mock::given(method("GET"))
.and(path("/repos/octo/api/actions/runs"))
.respond_with(
ResponseTemplate::new(404).set_body_json(json!({ "message": "Not Found" })),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::organization(
Org::new("octo-org").expect("a valid organization login"),
[repo(), other_repo()],
);
let activity = gateway
.in_progress_activity(&scope, &CancelToken::new())
.await
.expect("one unreadable repository is not fatal to the aggregate");
assert_eq!(activity.total(), 6);
assert_eq!(activity.for_repository(&other_repo()), None);
assert!(
!activity.is_complete(),
"a partial total is usable only when it says it is partial"
);
assert_eq!(activity.unavailable().len(), 1);
assert_eq!(activity.unavailable()[0].repository, other_repo());
}
#[tokio::test]
async fn a_repository_targets_activity_failure_propagates_rather_than_becoming_zero() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNS))
.respond_with(
ResponseTemplate::new(403)
.set_body_json(json!({ "message": "Resource not accessible by integration" })),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let error = gateway
.in_progress_activity(&ActivityScope::repository(repo()), &CancelToken::new())
.await
.expect_err("a scope of one has no partial answer to give");
assert!(matches!(
RefreshState::from_error(&error),
RefreshState::Forbidden { .. }
));
}
#[tokio::test]
async fn a_rate_limit_during_an_aggregate_aborts_it_rather_than_under_reporting() {
let server = MockServer::start().await;
mount_runs(&repo(), 6).mount(&server).await;
Mock::given(method("GET"))
.and(path("/repos/octo/api/actions/runs"))
.respond_with(
ResponseTemplate::new(429)
.insert_header("retry-after", "30")
.set_body_json(
json!({ "message": "You have exceeded a secondary rate limit" }),
),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::organization(
Org::new("octo-org").expect("a valid organization login"),
[repo(), other_repo(), third_repo()],
);
let error = gateway
.in_progress_activity(&scope, &CancelToken::new())
.await
.expect_err("a rate limit is systemic, not a fact about one repository");
assert!(error.is_rate_limited(), "{error}");
}
#[tokio::test]
async fn retry_after_is_obeyed_by_issuing_no_request_until_it_elapses() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(Script::new(vec![
ResponseTemplate::new(429)
.insert_header("retry-after", "120")
.set_body_json(
json!({ "message": "You have exceeded a secondary rate limit" }),
),
ResponseTemplate::new(200).set_body_json(runner_page(1..2, 1)),
]))
.mount(&server)
.await;
let clock = Arc::new(TestClock::default());
let gateway = gateway(&server, clock.clone());
let cancel = CancelToken::new();
let first = gateway
.list_runners(&repo_target(), &cancel)
.await
.expect_err("GitHub is rate limiting");
let limit = first.rate_limited().expect("a distinct rate-limited state");
assert_eq!(limit.kind, RateLimitKind::Secondary);
assert_eq!(limit.retry_after, Some(Duration::from_secs(120)));
assert_eq!(requests_seen(&server).await, 1);
let second = gateway
.list_runners(&repo_target(), &cancel)
.await
.expect_err("the back-off is still running");
assert!(second.is_rate_limited(), "{second}");
assert_eq!(
requests_seen(&server).await,
1,
"obeying `retry-after` means sending nothing, not sending and waiting"
);
assert_eq!(
gateway.rate_limit_backoff(),
Some(Duration::from_secs(120)),
"the reported wait is what is left of it"
);
clock.advance_secs(90);
assert!(
gateway
.list_runners(&repo_target(), &cancel)
.await
.is_err_and(|error| error.is_rate_limited())
);
assert_eq!(gateway.rate_limit_backoff(), Some(Duration::from_secs(30)));
assert_eq!(requests_seen(&server).await, 1);
clock.advance_secs(30);
assert_eq!(gateway.rate_limit_backoff(), None);
let inventory = gateway
.list_runners(&repo_target(), &cancel)
.await
.expect("the back-off elapsed");
assert_eq!(inventory.len(), 1);
assert_eq!(requests_seen(&server).await, 2);
}
#[tokio::test]
async fn a_cancelled_call_inside_a_latched_window_is_cancelled_not_rate_limited() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(429)
.insert_header("retry-after", "120")
.set_body_json(
json!({ "message": "You have exceeded a secondary rate limit" }),
),
)
.mount(&server)
.await;
let clock = Arc::new(TestClock::default());
let gateway = gateway(&server, clock.clone());
let live = CancelToken::new();
let first = gateway
.list_runners(&repo_target(), &live)
.await
.expect_err("GitHub is rate limiting");
assert!(first.is_rate_limited(), "{first}");
assert_eq!(
gateway.rate_limit_backoff(),
Some(Duration::from_secs(120)),
"the window has to actually be open, or this test proves nothing"
);
let cancelled = CancelToken::new();
cancelled.cancel();
let error = gateway
.list_runners(&repo_target(), &cancelled)
.await
.expect_err("a cancelled call is still an error");
assert!(
error.is_cancelled(),
"the answer a withdrawn caller gets is `Cancelled`: {error}"
);
assert!(
!error.is_rate_limited(),
"answering `RateLimited` tells a caller that navigated away to wait out a \
back-off it will never return for; the suppression branch must not \
outrank the cancellation: {error}"
);
assert_eq!(
requests_seen(&server).await,
1,
"and neither answer reached the wire: the window suppressed nothing extra \
and the cancellation opened no socket"
);
assert_eq!(
gateway.requests_issued(),
1,
"the budget accounting agrees: only the call that latched the window spent \
anything"
);
}
#[tokio::test]
async fn an_exhausted_hourly_quota_is_a_distinct_displayable_state() {
let server = MockServer::start().await;
let now = TestClock::default().now().timestamp();
let reset = now + 300;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(403)
.insert_header("x-ratelimit-remaining", "0")
.insert_header("x-ratelimit-limit", "5000")
.insert_header("x-ratelimit-reset", reset.to_string().as_str())
.set_body_json(json!({ "message": "API rate limit exceeded" })),
)
.mount(&server)
.await;
let clock = Arc::new(TestClock::default());
let gateway = gateway(&server, clock);
let error = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect_err("the quota is gone");
let limit = *error.rate_limited().expect("a rate-limited state");
assert_eq!(limit.kind, RateLimitKind::Primary);
assert_eq!(limit.remaining, Some(0));
assert_eq!(
limit.reset_unix_secs,
Some(u64::try_from(reset).unwrap()),
"the reset instant is what tells an operator how long this lasts"
);
assert_eq!(gateway.rate_limit_backoff(), Some(Duration::from_secs(300)));
let state = RefreshState::from_error(&error);
assert_eq!(state, RefreshState::RateLimited(limit));
assert!(!state.is_ready());
let rendered = state.to_string();
assert!(rendered.contains("primary"), "{rendered}");
assert!(rendered.contains("Refreshes are delayed"), "{rendered}");
assert_eq!(
state.retry_delay(TestClock::default().now()),
Some(Duration::from_secs(300))
);
}
#[tokio::test]
async fn a_404_carrying_an_exhausted_remaining_header_is_not_a_rate_limit() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(404)
.insert_header("x-ratelimit-remaining", "0")
.set_body_json(json!({ "message": "Not Found" })),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let error = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect_err("the repository is not there");
assert!(!error.is_rate_limited(), "{error}");
assert!(
gateway.rate_limit_backoff().is_none(),
"a 404 must not silence this gateway"
);
assert!(matches!(
RefreshState::from_error(&error),
RefreshState::Failed {
status: Some(404),
..
}
));
}
#[tokio::test]
async fn a_permissions_403_is_forbidden_and_not_a_rate_limit() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(403)
.set_body_json(json!({ "message": "Resource not accessible by integration" })),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let error = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect_err("the installation does not grant it");
assert!(!error.is_rate_limited(), "{error}");
assert!(gateway.rate_limit_backoff().is_none());
let state = RefreshState::from_error(&error);
assert!(
matches!(&state, RefreshState::Forbidden { message } if message.as_deref()
== Some("Resource not accessible by integration")),
"{state:?}: waiting does not fix a missing grant, so it must not be \
rendered as something to wait for"
);
assert_eq!(
state.retry_delay(TestClock::default().now()),
None,
"there is nothing to wait for"
);
}
#[tokio::test]
async fn a_permissions_403_that_lands_on_an_exhausted_quota_is_still_forbidden() {
let server = MockServer::start().await;
let reset = TestClock::default().now().timestamp() + 900;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(403)
.insert_header("x-ratelimit-remaining", "0")
.insert_header("x-ratelimit-limit", "5000")
.insert_header("x-ratelimit-reset", reset.to_string().as_str())
.set_body_json(json!({ "message": "Resource not accessible by integration" })),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let error = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect_err("the installation does not grant it");
assert!(
!error.is_rate_limited(),
"a missing grant does not become a rate limit because the quota also ran \
out on the same request: {error}"
);
assert!(
gateway.rate_limit_backoff().is_none(),
"and it must not latch a window that silences every other target too"
);
let state = RefreshState::from_error(&error);
assert!(
matches!(&state, RefreshState::Forbidden { message } if message.as_deref()
== Some("Resource not accessible by integration")),
"{state:?}"
);
assert_eq!(
state.retry_delay(TestClock::default().now()),
None,
"waiting for the quota to reset will not grant the permission"
);
}
#[test]
fn a_rate_limit_with_no_usable_delay_still_backs_off() {
let now = TestClock::default().now();
let bare = RateLimited {
kind: RateLimitKind::Secondary,
retry_after: None,
remaining: None,
reset_unix_secs: None,
};
assert_eq!(bare.delay_from(now), DEFAULT_RATE_LIMIT_BACKOFF);
let stale_reset = RateLimited {
reset_unix_secs: Some(u64::try_from(now.timestamp() - 60).unwrap()),
..bare
};
assert_eq!(
stale_reset.delay_from(now),
DEFAULT_RATE_LIMIT_BACKOFF,
"a reset already in the past means the clocks disagree, not that the \
limit has lifted"
);
let absurd = RateLimited {
retry_after: Some(Duration::from_secs(86_400)),
..bare
};
assert_eq!(
absurd.delay_from(now),
MAX_RATE_LIMIT_BACKOFF,
"a remote header does not get to decide how long this product stays dark"
);
}
#[tokio::test]
async fn the_hourly_quota_is_read_from_successful_responses_too() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.insert_header("x-ratelimit-limit", "5000")
.insert_header("x-ratelimit-remaining", "4873")
.insert_header("x-ratelimit-reset", "1787274000")
.set_body_json(runner_page(1..3, 2)),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
assert_eq!(gateway.headroom(), None, "nothing observed yet");
gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect("readable");
assert_eq!(
gateway.headroom(),
Some(RateLimitHeadroom {
limit: Some(5_000),
remaining: Some(4_873),
reset_unix_secs: Some(1_787_274_000),
}),
"a quota display that only appears once the quota is gone is not a display"
);
}
struct CancelWhileServingPage {
token: CancelToken,
next_page: String,
body: Value,
}
impl wiremock::Respond for CancelWhileServingPage {
fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate {
self.token.cancel();
ResponseTemplate::new(200)
.insert_header("x-ratelimit-limit", "5000")
.insert_header("x-ratelimit-remaining", "4999")
.insert_header("x-ratelimit-reset", "1787274000")
.insert_header("link", link_next(&self.next_page).as_str())
.set_body_json(self.body.clone())
}
}
#[tokio::test]
async fn a_cancellation_landing_mid_request_drops_the_response_unparsed() {
let server = MockServer::start().await;
let cancel = CancelToken::new();
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(CancelWhileServingPage {
token: cancel.clone(),
next_page: format!("{}/page/2", server.uri()),
body: runner_page(1..101, 200),
})
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/page/2"))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..201, 200)))
.expect(0)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
assert!(
!cancel.is_cancelled(),
"the walk has to start live, or this is a test of page zero"
);
let error = gateway
.list_runners(&repo_target(), &cancel)
.await
.expect_err("the token was flipped between page one and page two");
assert!(error.is_cancelled(), "{error}");
assert!(
cancel.is_cancelled(),
"page one was served, which is what flipped the token"
);
assert_eq!(
requests_seen(&server).await,
1,
"page one, and nothing after it: the `Link` header offered page two and the \
walk declined to spend the request"
);
assert_eq!(
gateway.requests_issued(),
1,
"and the budget accounting agrees with the wire"
);
assert!(
gateway.headroom().is_none(),
"the response carried `x-ratelimit-*` and they were never read, which is what \
`abandoned in flight` means: `issue` returned `Cancelled` from `run` without \
reaching its `Ok` arm. `Some` here would mean page one was actually parsed \
and this test had silently become the between-pages case"
);
}
static CANCEL_WHILE_PARSING: std::sync::Mutex<Option<CancelToken>> =
std::sync::Mutex::new(None);
#[derive(Debug, Deserialize)]
struct CancelOnParsePage {
total_count: Option<u64>,
#[serde(default)]
runners: Vec<RawRunner>,
}
impl WirePage for CancelOnParsePage {
type Item = RawRunner;
const WHAT: &'static str = "runners";
fn reported_total(&self) -> Option<u64> {
self.total_count
}
fn into_items(self) -> Vec<Self::Item> {
if let Some(token) = CANCEL_WHILE_PARSING
.lock()
.expect("the parse-time cancel seam is not poisoned")
.take()
{
token.cancel();
}
self.runners
}
}
#[tokio::test]
async fn cancelling_between_pages_stops_the_walk() {
let server = MockServer::start().await;
let page_two = format!("{}/page/2", server.uri());
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.insert_header("x-ratelimit-limit", "5000")
.insert_header("x-ratelimit-remaining", "4999")
.insert_header("x-ratelimit-reset", "1787274000")
.insert_header("link", link_next(&page_two).as_str())
.set_body_json(runner_page(1..101, 200)),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/page/2"))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..201, 200)))
.expect(0)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let cancel = CancelToken::new();
*CANCEL_WHILE_PARSING
.lock()
.expect("the parse-time cancel seam is not poisoned") = Some(cancel.clone());
assert!(
!cancel.is_cancelled(),
"the walk has to start live, or this is a test of page zero"
);
let first = ApiRequest::get(REPO_RUNNERS).query("per_page", PER_PAGE);
let error = match gateway
.collect_pages::<CancelOnParsePage>(first, &cancel)
.await
{
Err(error) => error,
Ok(_) => panic!("the token was flipped between page one and page two"),
};
assert!(error.is_cancelled(), "{error}");
assert!(
cancel.is_cancelled(),
"page one was parsed, which is what flipped the token"
);
assert!(
gateway.headroom().is_some(),
"page one's response must have been parsed for this to be the between-pages \
case at all; `headroom` is written only in `issue`'s `Ok` arm, so `None` here \
would mean page one was cancelled in flight and the walk never saw the \
`Link` header it is supposed to decline to follow"
);
assert_eq!(
requests_seen(&server).await,
1,
"page one, and nothing after it: the `Link` header offered page two and the \
walk declined to spend the request"
);
assert_eq!(
gateway.requests_issued(),
1,
"and the budget accounting agrees with the wire"
);
}
#[tokio::test]
async fn an_already_cancelled_token_stops_a_walk_before_its_first_page() {
let server = MockServer::start().await;
let page_two = format!("{}/page/2", server.uri());
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.insert_header("link", link_next(&page_two).as_str())
.set_body_json(runner_page(1..101, 200)),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/page/2"))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(101..201, 200)))
.expect(0)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let cancel = CancelToken::new();
let first = ApiRequest::get(REPO_RUNNERS).query("per_page", PER_PAGE);
let response = gateway
.issue(&first, &cancel)
.await
.expect("page one is readable");
assert!(response.next_page().is_some(), "page two is on offer");
cancel.cancel();
let error = gateway
.list_runners(&repo_target(), &cancel)
.await
.expect_err("the token is cancelled");
assert!(error.is_cancelled(), "{error}");
assert_eq!(
requests_seen(&server).await,
1,
"the manual request only; the walk spent nothing at all"
);
assert_eq!(
gateway.requests_issued(),
1,
"and the budget accounting agrees with the wire: a request that was \
never polled is not a request that was issued"
);
}
#[tokio::test]
async fn cancelling_an_in_flight_request_abandons_it() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_secs(20))
.set_body_json(runner_page(1..2, 1)),
)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let cancel = CancelToken::new();
let token = cancel.clone();
let canceller = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(50)).await;
token.cancel();
});
let started = std::time::Instant::now();
let error = gateway
.list_runners(&repo_target(), &cancel)
.await
.expect_err("the caller withdrew");
let elapsed = started.elapsed();
canceller.await.expect("the canceller completes");
assert!(error.is_cancelled(), "{error}");
assert!(
elapsed < Duration::from_secs(10),
"the request was awaited to completion rather than abandoned: {elapsed:?}"
);
assert!(cancel.is_cancelled());
assert!(
CancelToken::new().check().is_ok(),
"a fresh token is not cancelled"
);
}
#[tokio::test]
async fn an_absent_sha256_checksum_is_absent_and_an_empty_one_is_empty() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/repos/octo/dashboard/actions/runners/downloads"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!([
{
"os": "win",
"architecture": "x64",
"download_url": "https://example.invalid/win-x64.zip",
"filename": "actions-runner-win-x64.zip",
"sha256_checksum": "abc123"
},
{
"os": "osx",
"architecture": "arm64",
"download_url": "https://example.invalid/osx-arm64.tar.gz",
"filename": "actions-runner-osx-arm64.tar.gz"
},
{
"os": "linux",
"architecture": "x64",
"download_url": "https://example.invalid/linux-x64.tar.gz",
"filename": "actions-runner-linux-x64.tar.gz",
"sha256_checksum": null
},
{
"os": "linux",
"architecture": "arm64",
"download_url": "https://example.invalid/linux-arm64.tar.gz",
"filename": "actions-runner-linux-arm64.tar.gz",
"sha256_checksum": ""
}
])))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let downloads = gateway
.runner_downloads(&repo_target(), &CancelToken::new())
.await
.expect("the metadata is readable");
let windows = downloads
.select(Os::Windows, Arch::X64)
.expect("selected by OS and architecture");
assert_eq!(windows.sha256_checksum(), Some("abc123"));
let missing = downloads.select(Os::MacOs, Arch::Arm64).expect("selected");
assert_eq!(
missing.sha256_checksum(),
None,
"`e2` fails closed on an absent digest, and can only do that if this \
layer does not paper the absence over"
);
let null = downloads.select(Os::Linux, Arch::X64).expect("selected");
assert_eq!(
null.sha256_checksum(),
None,
"an explicit null is absent too"
);
let empty = downloads.select(Os::Linux, Arch::Arm64).expect("selected");
assert_eq!(
empty.sha256_checksum(),
Some(""),
"an empty digest is a different fact from a missing one, and \
collapsing them would leave `e2` unable to report which it saw"
);
assert_ne!(
empty.sha256_checksum(),
missing.sha256_checksum(),
"absent and empty must be distinguishable"
);
assert_eq!(
downloads.select(Os::Windows, Arch::Arm32),
None,
"an unpublished pair is refused rather than substituted"
);
assert_eq!(gateway.requests_issued(), 1, "downloads are not paginated");
}
#[tokio::test]
async fn runner_downloads_are_read_at_organization_scope_too() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/octo-org/actions/runners/downloads"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!([{
"os": "linux",
"architecture": "arm",
"download_url": "https://example.invalid/linux-arm.tar.gz",
"filename": "actions-runner-linux-arm.tar.gz",
"sha256_checksum": "deadbeef"
}])))
.expect(1)
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let downloads = gateway
.runner_downloads(&org_target(), &CancelToken::new())
.await
.expect("readable");
assert!(downloads.select(Os::Linux, Arch::Arm32).is_some());
}
#[tokio::test]
async fn labels_are_read_as_github_stores_them_and_matched_case_insensitively() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"total_count": 2,
"runners": [
{
"id": 73,
"name": "rm-d18-spike-ivanpc-1753",
"os": "win",
"status": "offline",
"busy": false,
"ephemeral": true,
"labels": [
{ "id": 1, "name": "rm-home-win-x64", "type": "read-only" },
{ "id": 2, "name": "windows", "type": "read-only" }
]
},
{
"id": 74,
"name": "legacy-persistent",
"os": "Linux",
"status": "provisioning",
"busy": true,
"labels": []
}
]
})))
.mount(&server)
.await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let inventory = gateway
.list_runners(&repo_target(), &CancelToken::new())
.await
.expect("readable");
let spike = &inventory.runners()[0];
assert_eq!(spike.labels, ["rm-home-win-x64", "windows"]);
assert!(
spike.has_label("Windows"),
"GitHub lower-cases what it stores"
);
assert!(spike.has_label(" windows "));
assert!(
!spike.has_label("self-hosted"),
"no label is added implicitly (D18, point 1)"
);
assert_eq!(spike.status, RunnerStatus::Offline);
assert_eq!(spike.ephemeral, Some(true));
assert_eq!(spike.parsed_os(), Some(Os::Windows));
let legacy = &inventory.runners()[1];
assert_eq!(
legacy.status,
RunnerStatus::Other("provisioning".to_string()),
"an unrecognised status is something to display, not something to guess at"
);
assert_eq!(
legacy.ephemeral, None,
"absent is not `false`: a runner whose ephemerality is unknown is \
exactly the one an operator wants flagged"
);
assert_eq!(legacy.parsed_os(), Some(Os::Linux));
assert!(legacy.busy);
assert_eq!(inventory.busy_count(), 1);
assert_eq!(inventory.online_count(), 0);
}
#[tokio::test]
async fn a_manual_refresh_during_an_in_flight_one_coalesces_into_a_single_request() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_millis(150))
.set_body_json(runner_page(1..4, 3)),
)
.mount(&server)
.await;
mount_runs(&repo(), 2).mount(&server).await;
let gateway = Arc::new(gateway(&server, Arc::new(TestClock::default())));
let coalescer: Arc<RefreshCoalescer<RefreshState>> = Arc::new(RefreshCoalescer::new());
let scope = ActivityScope::repository(repo());
let refresh = || {
let gateway = gateway.clone();
let coalescer = coalescer.clone();
let scope = scope.clone();
async move {
coalescer
.refresh(|| async {
RefreshState::from_result(
gateway.snapshot(&scope, &CancelToken::new()).await,
)
})
.await
}
};
let (scheduled, manual) = tokio::join!(refresh(), refresh());
assert_eq!(coalescer.performed(), 1, "one refresh actually ran");
assert_eq!(coalescer.joined(), 1, "the other joined it");
assert_eq!(scheduled, manual, "and both callers got the same answer");
assert!(scheduled.is_ready(), "{scheduled}");
assert_eq!(
scheduled.snapshot().expect("ready").runners.len(),
3,
"joining must return the answer, not an empty placeholder"
);
assert_eq!(
requests_seen(&server).await,
2,
"one refresh is one runners request plus one runs request; a second \
refresh would have made it four"
);
assert_eq!(gateway.requests_issued(), 2);
assert_eq!(coalescer.last(), Some(scheduled));
}
#[tokio::test]
async fn a_refresh_after_the_previous_one_completed_is_not_coalesced() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(REPO_RUNNERS))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(1..2, 1)))
.mount(&server)
.await;
mount_runs(&repo(), 0).mount(&server).await;
let gateway = gateway(&server, Arc::new(TestClock::default()));
let coalescer: RefreshCoalescer<RefreshState> = RefreshCoalescer::new();
let scope = ActivityScope::repository(repo());
for _ in 0..3 {
let state = coalescer
.refresh(|| async {
RefreshState::from_result(gateway.snapshot(&scope, &CancelToken::new()).await)
})
.await;
assert!(state.is_ready(), "{state}");
}
assert_eq!(coalescer.performed(), 3);
assert_eq!(
coalescer.joined(),
0,
"coalescing an in-flight refresh must not become caching a finished one"
);
assert_eq!(gateway.requests_issued(), 6);
}
fn interval(secs: u16) -> RefreshInterval {
RefreshInterval::from_secs(secs).expect("at or above the documented floor")
}
#[test]
fn a_repository_target_costs_the_documented_number_of_requests() {
let default = interval(RefreshInterval::DEFAULT_SECS);
let floor = interval(RefreshInterval::MIN_SECS);
assert_eq!(refreshes_per_hour(default), 60);
assert_eq!(refreshes_per_hour(floor), 120);
let target = TargetCost::repository();
assert_eq!(target.requests_per_refresh(), 4);
assert_eq!(
target.requests_per_hour(default),
240,
"the documented per-target total at the 60-second default"
);
assert_eq!(
target.requests_per_hour(floor),
480,
"and at the 30-second floor"
);
}
#[test]
fn the_projection_reproduces_the_documented_target_ceilings() {
assert_eq!(HOURLY_REQUEST_CEILING, 5_000);
assert_eq!(budget_allowance(), 2_500, "half the ceiling");
assert_eq!(
BudgetProjection::max_repository_targets(interval(RefreshInterval::DEFAULT_SECS)),
10
);
assert_eq!(
BudgetProjection::max_repository_targets(interval(RefreshInterval::MIN_SECS)),
5
);
let default = interval(RefreshInterval::DEFAULT_SECS);
let ten = BudgetProjection::new(default, vec![TargetCost::repository(); 10]);
assert_eq!(ten.requests_per_hour(), 2_400);
assert!(!ten.exceeds_allowance());
assert_eq!(ten.headroom(), 100);
let eleven = BudgetProjection::new(default, vec![TargetCost::repository(); 11]);
assert_eq!(eleven.requests_per_hour(), 2_640);
assert!(
eleven.exceeds_allowance(),
"the eleventh repository is the one an operator needs told about"
);
assert_eq!(eleven.headroom(), 0);
}
#[test]
fn an_organization_target_costs_materially_more_than_a_repository_target() {
let default = interval(RefreshInterval::DEFAULT_SECS);
let repository = TargetCost::repository().requests_per_hour(default);
assert_eq!(
TargetCost::organization(1).requests_per_hour(default),
repository,
"at one installed repository the two models agree exactly, which is \
what makes this a refinement of the documented table rather than a \
contradiction of it"
);
let ten = TargetCost::organization(10);
assert_eq!(ten.requests_per_refresh(), 31);
assert_eq!(ten.requests_per_hour(default), 1_860);
assert!(
ten.requests_per_hour(default) > repository * 7,
"an organization on ten repositories costs nearly eight times a \
repository target; projecting it flat understates the real spend by \
exactly that factor"
);
let empty = BudgetProjection::new(default, Vec::new());
assert!(empty.admit(TargetCost::repository()).is_admitted());
assert!(empty.admit(TargetCost::organization(13)).is_admitted());
let refusal = empty.admit(TargetCost::organization(14));
assert!(
!refusal.is_admitted(),
"a single organization on fourteen repositories already exceeds a \
host's whole share of the budget"
);
}
#[test]
fn a_refused_configuration_states_the_numbers_and_the_maximum_target_count() {
let default = interval(RefreshInterval::DEFAULT_SECS);
let full = BudgetProjection::new(default, vec![TargetCost::repository(); 10]);
let Admission::Refused {
projected_requests_per_hour,
allowance,
max_repository_targets,
..
} = full.admit(TargetCost::repository())
else {
panic!("the eleventh repository must be refused");
};
assert_eq!(projected_requests_per_hour, 2_640);
assert_eq!(allowance, 2_500);
assert_eq!(max_repository_targets, 10);
let message = full.admit(TargetCost::repository()).to_string();
for expected in ["2640", "2500", "5000", "60-second", "about 10 repository"] {
assert!(
message.contains(expected),
"{expected:?} missing from: {message}"
);
}
assert!(
!message.contains("because the App is installed on"),
"the organization clause belongs only on an organization refusal: {message}"
);
let org_message = full.admit(TargetCost::organization(4)).to_string();
assert!(
org_message.contains("installed on 4 of its repositories"),
"{org_message}"
);
let admitted = BudgetProjection::new(default, vec![TargetCost::repository(); 2])
.admit(TargetCost::repository())
.to_string();
assert!(admitted.contains("720"), "{admitted}");
assert!(admitted.contains("1780"), "{admitted}");
}
#[tokio::test]
async fn the_budget_model_matches_the_requests_the_gateway_really_issues() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(ORG_RUNNERS))
.respond_with(ResponseTemplate::new(200).set_body_json(runner_page(1..4, 3)))
.mount(&server)
.await;
for repository in [repo(), other_repo(), third_repo()] {
mount_runs(&repository, 1).mount(&server).await;
}
let gateway = gateway(&server, Arc::new(TestClock::default()));
let scope = ActivityScope::organization(
Org::new("octo-org").expect("a valid organization login"),
[repo(), other_repo(), third_repo()],
);
gateway
.snapshot(&scope, &CancelToken::new())
.await
.expect("readable");
let cost = TargetCost::from_activity_scope(&scope);
assert_eq!(cost.installed_repositories(), 3);
assert_eq!(cost.scope(), TargetScope::Organization);
let modelled_without_demand = cost.requests_per_refresh()
- DEMAND_REQUESTS_PER_REPOSITORY_PER_REFRESH * cost.installed_repositories();
assert_eq!(
gateway.requests_issued(),
u64::from(modelled_without_demand),
"the model projects {modelled_without_demand} inventory-and-activity \
requests per refresh for this scope, and the gateway issued {}",
gateway.requests_issued()
);
assert_eq!(
modelled_without_demand,
RUNNER_INVENTORY_REQUESTS_PER_REFRESH + scope.requests_per_refresh()
);
}
#[test]
fn an_organization_with_no_installed_repositories_is_projected_as_such() {
let scope = ActivityScope::organization(
Org::new("octo-org").expect("a valid organization login"),
[],
);
assert_eq!(scope.requests_per_refresh(), 0);
let cost = TargetCost::from_activity_scope(&scope);
assert_eq!(cost.installed_repositories(), 0);
assert_eq!(
cost.requests_per_refresh(),
RUNNER_INVENTORY_REQUESTS_PER_REFRESH,
"the runners endpoint is still polled; nothing else is"
);
}
#[test]
fn the_demand_cost_can_be_reported_by_the_task_that_measures_it() {
let default = interval(RefreshInterval::DEFAULT_SECS);
assert_eq!(
TargetCost::repository().requests_per_hour(default),
240,
"the documented estimate is the default"
);
let measured = TargetCost::repository().with_demand_requests_per_repository(3);
assert_eq!(measured.requests_per_refresh(), 5);
assert_eq!(measured.requests_per_hour(default), 300);
let org = TargetCost::organization(4).with_demand_requests_per_repository(3);
assert_eq!(org.requests_per_refresh(), 1 + 4 * (1 + 3));
assert_eq!(
org.requests_per_hour(default),
1_020,
"a worse demand cost lands hardest on an organization, which is \
exactly the effect a flat per-target model would hide"
);
}
#[test]
fn a_repository_activity_scope_covers_exactly_one_repository() {
let scope = ActivityScope::repository(repo());
assert_eq!(scope.repositories(), [repo()]);
assert_eq!(scope.requests_per_refresh(), 1);
assert_eq!(scope.target(), &repo_target());
assert_eq!(
TargetCost::from_activity_scope(&scope),
TargetCost::repository()
);
}
#[test]
fn the_authentication_taxonomy_survives_the_summary() {
assert_eq!(
RefreshState::from_error(&InventoryError::Github(GithubError::AuthenticationFailed)),
RefreshState::Unauthorized
);
assert_eq!(
RefreshState::from_error(&InventoryError::Github(
GithubError::AuthenticationLockout {
retry_after: Duration::from_secs(60)
}
)),
RefreshState::LockedOut {
retry_after: Duration::from_secs(60)
}
);
assert_eq!(
RefreshState::from_error(&InventoryError::Cancelled),
RefreshState::Cancelled
);
let now = TestClock::default().now();
assert_eq!(
RefreshState::LockedOut {
retry_after: Duration::from_secs(60)
}
.retry_delay(now),
Some(Duration::from_secs(60))
);
assert_eq!(RefreshState::Unauthorized.retry_delay(now), None);
assert_eq!(RefreshState::Offline.retry_delay(now), None);
}
#[test]
fn an_empty_snapshot_is_ready_rather_than_a_failure() {
let snapshot = InventorySnapshot {
target: repo_target(),
runners: RunnerInventory::new(repo_target(), Vec::new()),
activity: ActivityCount::of(repo(), 0),
observed_at: TestClock::default().now(),
headroom: None,
};
let state = RefreshState::from_result(Ok(snapshot));
assert!(state.is_ready());
assert!(state.snapshot().expect("ready").runners.is_empty());
assert_eq!(state.to_string(), "0 runners, 0 in progress");
}
}