Skip to main content

runnerd_core/
github.rs

1//! The GitHub REST API as the scheduler and the history need it: queued jobs, a job's steps, just-in-time runner
2//! configs, repository runners and cancelling a run. Owns the token, conditional requests and rate-limit backoff. It
3//! must never put the token in a log line, a `Debug` output or an error, and it decides nothing about scheduling.
4
5use std::collections::HashMap;
6use std::fmt;
7use std::sync::{Arc, Mutex};
8use std::time::{Duration, SystemTime};
9
10use reqwest::header::{self, HeaderMap, HeaderValue};
11use reqwest::{Method, StatusCode};
12use serde::Deserialize;
13use serde::de::DeserializeOwned;
14
15use crate::api;
16use crate::docker::JitConfig;
17use crate::sync::lock;
18use crate::time;
19
20/// The public API's address.
21pub const API_URL: &str = "https://api.github.com";
22
23/// How long one request may take, connecting included. GitHub answers in well under a second; a request that hangs
24/// holds back the tick that made it.
25const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
26
27/// A GitHub token. Its `Debug` output never shows the value.
28#[derive(Clone)]
29pub struct Token(String);
30
31impl fmt::Debug for Token {
32    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33        f.write_str("Token(<redacted>)")
34    }
35}
36
37/// Where the client gets a token: at start and again when GitHub answers 401.
38pub type TokenSource = Box<dyn Fn() -> Result<Token, Error> + Send + Sync>;
39
40/// The `gh` CLI's token for github.com (`gh auth token`).
41pub fn gh_token() -> Result<Token, Error> {
42    let output = std::process::Command::new("gh")
43        .args(["auth", "token", "--hostname", "github.com"])
44        .output()
45        .map_err(|error| Error::Token(format!("cannot run gh: {error}")))?;
46    if !output.status.success() {
47        return Err(Error::Token(format!(
48            "gh auth token failed: {}",
49            String::from_utf8_lossy(&output.stderr).trim()
50        )));
51    }
52    let token = String::from_utf8(output.stdout)
53        .map_err(|_| Error::Token("gh auth token printed a token that is not UTF-8".to_owned()))?;
54    Ok(Token(token.trim().to_owned()))
55}
56
57/// Makes ring the process's TLS crypto provider, which every `reqwest::Client` needs: reqwest is built without
58/// one. Calling it again does nothing.
59pub fn install_crypto_provider() {
60    // Fails only when a provider is installed already.
61    let _ = rustls::crypto::ring::default_provider().install_default();
62}
63
64/// What a GitHub request can fail with. No variant holds the token.
65#[derive(Debug, thiserror::Error)]
66pub enum Error {
67    /// GitHub's rate limit is used up; no request is sent before `until`.
68    #[error("GitHub's rate limit is used up until {}s after the epoch", secs(*until))]
69    RateLimited {
70        /// When the limit resets.
71        until: SystemTime,
72    },
73    /// GitHub answered with an error status.
74    #[error("GitHub answered {status} to {method} {path}: {message}")]
75    Status {
76        /// The request's method.
77        method: Method,
78        /// The request's path, without the host.
79        path: String,
80        /// What GitHub answered.
81        status: StatusCode,
82        /// GitHub's error message, or the body as text.
83        message: String,
84    },
85    /// The request did not get an answer.
86    #[error("{method} {path}: {source}")]
87    Http {
88        /// The request's method.
89        method: Method,
90        /// The request's path, without the host.
91        path: String,
92        /// What failed, without the URL.
93        source: reqwest::Error,
94    },
95    /// The answer is not the JSON expected.
96    #[error("{path}: unexpected response: {source}")]
97    Decode {
98        /// The request's path, without the host.
99        path: String,
100        /// What did not parse.
101        source: serde_json::Error,
102    },
103    /// No token could be read.
104    #[error("no GitHub token: {0}")]
105    Token(String),
106}
107
108fn secs(time: SystemTime) -> u64 {
109    time.duration_since(SystemTime::UNIX_EPOCH)
110        .unwrap_or_default()
111        .as_secs()
112}
113
114/// A job waiting for a runner.
115#[derive(Debug, Clone, Default, PartialEq, Eq)]
116pub struct QueuedJob {
117    /// The job's ID; IDs grow with time.
118    pub id: u64,
119    /// The workflow run it belongs to.
120    pub run_id: u64,
121    /// The job's name in the workflow.
122    pub name: String,
123    /// The workflow's `runs-on` labels.
124    pub labels: Vec<String>,
125    /// When GitHub queued it, in milliseconds since the epoch; 0 if GitHub's answer did not say.
126    pub created_at: u64,
127    /// 1 for the run's first attempt.
128    pub run_attempt: u32,
129    /// The workflow's name.
130    pub workflow: Option<String>,
131    /// The branch the run is for.
132    pub branch: Option<String>,
133    /// The job's page on GitHub.
134    pub url: Option<String>,
135}
136
137/// A job as `GET /repos/{owner}/{repo}/actions/jobs/{id}` describes it: its progress and steps.
138#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
139pub struct JobDetails {
140    /// The job's ID.
141    pub id: u64,
142    /// The workflow run it belongs to.
143    pub run_id: u64,
144    /// 1 for the run's first attempt.
145    #[serde(default = "first_attempt")]
146    pub run_attempt: u32,
147    /// The job's name in the workflow.
148    pub name: String,
149    /// The workflow's name.
150    pub workflow_name: Option<String>,
151    /// The branch the run is for.
152    pub head_branch: Option<String>,
153    /// The job's page on GitHub.
154    pub html_url: Option<String>,
155    /// The workflow's `runs-on` labels.
156    #[serde(default)]
157    pub labels: Vec<String>,
158    /// When GitHub queued it.
159    pub created_at: Option<String>,
160    /// `queued`, `in_progress`, `completed`, `waiting`, ….
161    pub status: String,
162    /// Once completed: `success`, `failure`, `cancelled`, ….
163    pub conclusion: Option<String>,
164    /// When the runner started it.
165    pub started_at: Option<String>,
166    /// When it completed.
167    pub completed_at: Option<String>,
168    /// The runner that took it, if one did.
169    pub runner_name: Option<String>,
170    /// Its steps so far.
171    #[serde(default)]
172    pub steps: Vec<JobStep>,
173}
174
175/// What started a workflow run, as `GET /repos/{owner}/{repo}/actions/runs/{id}` says.
176#[derive(Debug, Clone, PartialEq, Eq)]
177pub struct RunTrigger {
178    /// `push`, `pull_request`, `schedule`, `workflow_dispatch`, ….
179    pub event: String,
180    /// The branch it ran on: a pull request's head branch, which may be a fork's.
181    pub head_branch: Option<String>,
182    /// `owner/name` of the repository its commit came from: a fork's, for a pull request from one.
183    pub head_repository: Option<String>,
184    /// The commit it ran. A tag's run gives the tag's name as `head_branch`, so only this says whether the
185    /// branch has it.
186    pub head_sha: String,
187    /// Whether people, not bots or apps, started it: GitHub's `actor` and `triggering_actor`, who re-ran it, are
188    /// both of type `User`.
189    pub by_user: bool,
190    /// The pull requests into the repository it runs, by number. GitHub lists only those whose head branch is in
191    /// the repository itself, and may list none once they are closed; those it lists into another repository of a
192    /// fork network are left out.
193    pub pull_requests: Vec<u64>,
194}
195
196/// A pull request, as `GET /repos/{owner}/{repo}/pulls/{number}` says.
197#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct PullRequest {
199    /// Whether a person, not a bot or an app, opened it: its `user` is of type `User`.
200    pub by_user: bool,
201    /// `owner/name` of the repository its head branch is in; `None` once that is deleted.
202    pub head_repository: Option<String>,
203    /// Its head branch.
204    pub head_branch: String,
205}
206
207/// A repository, as `GET /repos/{owner}/{repo}` says.
208#[derive(Debug, Clone, PartialEq, Eq)]
209pub struct RepositoryDetails {
210    /// Its default branch.
211    pub default_branch: String,
212    /// Whether it is public, private or internal.
213    pub visibility: api::Visibility,
214}
215
216fn first_attempt() -> u32 {
217    1
218}
219
220/// A step of `JobDetails`.
221#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
222pub struct JobStep {
223    /// GitHub's step number.
224    pub number: u32,
225    /// The name GitHub shows.
226    pub name: String,
227    /// `pending`, `queued`, `in_progress` or `completed`.
228    pub status: String,
229    /// Once completed: `success`, `failure`, `skipped`, ….
230    pub conclusion: Option<String>,
231    /// When it started.
232    pub started_at: Option<String>,
233    /// When it completed.
234    pub completed_at: Option<String>,
235}
236
237/// A runner registered on a repository.
238#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
239pub struct Runner {
240    /// The runner's ID.
241    pub id: u64,
242    /// Its name; runnerd's runners start with the host's prefix.
243    pub name: String,
244    /// `online` or `offline`.
245    pub status: String,
246    /// Whether it runs a job.
247    pub busy: bool,
248}
249
250/// What deleting a runner did.
251#[derive(Debug, Clone, Copy, PartialEq, Eq)]
252pub enum Deleted {
253    /// GitHub removed it.
254    Deleted,
255    /// It was gone already: an ephemeral runner removes itself after its job.
256    NotFound,
257    /// GitHub refused because the runner runs a job.
258    Busy,
259}
260
261/// The GitHub calls the scheduler makes. `Github` is the real client; the engine's tests use a fake.
262pub trait GithubApi: Send + Sync {
263    /// The repository's jobs waiting for a runner, in queued runs and in runs that started some of their jobs.
264    fn queued_jobs(
265        &self,
266        repository: &str,
267    ) -> impl Future<Output = Result<Vec<QueuedJob>, Error>> + Send;
268    /// The repository's self-hosted runners.
269    fn runners(&self, repository: &str) -> impl Future<Output = Result<Vec<Runner>, Error>> + Send;
270    /// Registers a runner for one job and returns it with the config the runner's listener takes.
271    fn jit_config(
272        &self,
273        repository: &str,
274        name: &str,
275        labels: &[String],
276    ) -> impl Future<Output = Result<(Runner, JitConfig), Error>> + Send;
277    /// Removes a runner from the repository.
278    fn delete_runner(
279        &self,
280        repository: &str,
281        id: u64,
282    ) -> impl Future<Output = Result<Deleted, Error>> + Send;
283    /// One job's progress and steps.
284    fn job(
285        &self,
286        repository: &str,
287        id: u64,
288    ) -> impl Future<Output = Result<JobDetails, Error>> + Send;
289    /// What started a workflow run.
290    fn run_trigger(
291        &self,
292        repository: &str,
293        run: u64,
294    ) -> impl Future<Output = Result<RunTrigger, Error>> + Send;
295    /// One pull request: who opened it and where its head branch is.
296    fn pull_request(
297        &self,
298        repository: &str,
299        number: u64,
300    ) -> impl Future<Output = Result<PullRequest, Error>> + Send;
301    /// The repository's default branch and visibility.
302    fn repository(
303        &self,
304        repository: &str,
305    ) -> impl Future<Output = Result<RepositoryDetails, Error>> + Send;
306    /// Whether `branch` of the repository has the commit `sha`: its head or an ancestor of it. A tag of the same
307    /// name, which git would take for it, does not count.
308    fn on_branch(
309        &self,
310        repository: &str,
311        branch: &str,
312        sha: &str,
313    ) -> impl Future<Output = Result<bool, Error>> + Send;
314    /// Cancels a workflow run, and so every job in it: GitHub cancels runs, not single jobs. A run that already
315    /// completed is left as it is.
316    fn cancel_run(
317        &self,
318        repository: &str,
319        run: u64,
320    ) -> impl Future<Output = Result<(), Error>> + Send;
321}
322
323/// A GitHub API client for one token. Requests on one URL are conditional on its last `ETag`, so an unchanged
324/// answer costs no rate limit.
325pub struct Github {
326    http: reqwest::Client,
327    base: String,
328    /// Run on Tokio's blocking pool: `gh` may wait on a keyring for seconds.
329    token_source: Arc<dyn Fn() -> Result<Token, Error> + Send + Sync>,
330    /// An async lock, held while the source runs: requests that need a token wait without blocking a worker, and
331    /// concurrent 401s read it once.
332    token: tokio::sync::Mutex<Option<Token>>,
333    // Locked with `sync::lock`: a panic while holding one of these locks leaves plain data behind, still usable.
334    etags: Mutex<HashMap<String, (HeaderValue, Vec<u8>)>>,
335    blocked_until: Mutex<Option<SystemTime>>,
336}
337
338impl fmt::Debug for Github {
339    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
340        f.debug_struct("Github")
341            .field("base", &self.base)
342            .finish_non_exhaustive()
343    }
344}
345
346#[derive(Deserialize)]
347struct Runs {
348    workflow_runs: Vec<Run>,
349}
350
351#[derive(Deserialize)]
352struct Run {
353    id: u64,
354}
355
356#[derive(Deserialize)]
357struct Jobs {
358    jobs: Vec<Job>,
359}
360
361#[derive(Deserialize)]
362struct Job {
363    id: u64,
364    run_id: u64,
365    name: String,
366    status: String,
367    labels: Vec<String>,
368    created_at: Option<String>,
369    run_attempt: Option<u32>,
370    workflow_name: Option<String>,
371    head_branch: Option<String>,
372    html_url: Option<String>,
373}
374
375#[derive(Deserialize)]
376struct RunBody {
377    event: String,
378    head_branch: Option<String>,
379    head_repository: Option<RepositoryName>,
380    head_sha: String,
381    actor: Option<Account>,
382    triggering_actor: Option<Account>,
383    #[serde(default)]
384    pull_requests: Vec<PullRequestNumber>,
385}
386
387#[derive(Deserialize)]
388struct Account {
389    #[serde(rename = "type")]
390    kind: String,
391}
392
393impl Account {
394    /// Whether GitHub says `account` is a person's: `Bot` is a GitHub App's or a bot's.
395    fn is_user(account: Option<&Account>) -> bool {
396        account.is_some_and(|account| account.kind == "User")
397    }
398}
399
400#[derive(Deserialize)]
401struct PullRequestNumber {
402    number: u64,
403    base: Option<PullRequestBase>,
404}
405
406#[derive(Deserialize)]
407struct PullRequestBase {
408    repo: Option<RepositoryUrl>,
409}
410
411#[derive(Deserialize)]
412struct RepositoryUrl {
413    url: String,
414}
415
416impl PullRequestNumber {
417    /// Whether it is a pull request into `repository`: in a fork network, a run can also list one into another
418    /// repository whose head is the same branch.
419    fn targets(&self, repository: &str) -> bool {
420        let Some(url) = self
421            .base
422            .as_ref()
423            .and_then(|base| base.repo.as_ref())
424            .map(|repo| repo.url.to_ascii_lowercase())
425        else {
426            return false;
427        };
428        url.ends_with(&format!("/repos/{}", repository.to_ascii_lowercase()))
429    }
430}
431
432#[derive(Deserialize)]
433struct PullRequestBody {
434    user: Option<Account>,
435    head: PullRequestHead,
436}
437
438#[derive(Deserialize)]
439struct PullRequestHead {
440    #[serde(rename = "ref")]
441    branch: String,
442    repo: Option<RepositoryName>,
443}
444
445#[derive(Deserialize)]
446struct RepositoryName {
447    full_name: String,
448}
449
450#[derive(Deserialize)]
451struct CompareBody {
452    status: String,
453}
454
455#[derive(Deserialize)]
456struct BranchBody {
457    commit: CommitBody,
458}
459
460#[derive(Deserialize)]
461struct CommitBody {
462    sha: String,
463}
464
465#[derive(Deserialize)]
466struct RepositoryBody {
467    default_branch: String,
468    visibility: api::Visibility,
469}
470
471#[derive(Deserialize)]
472struct Runners {
473    runners: Vec<Runner>,
474}
475
476#[derive(Deserialize)]
477struct JitResponse {
478    runner: Runner,
479    encoded_jit_config: String,
480}
481
482#[derive(Deserialize)]
483struct ErrorBody {
484    message: String,
485}
486
487impl Github {
488    /// A client for `base` (`API_URL`, or a test server) that takes its token from `token_source`.
489    pub fn new(base: &str, token_source: TokenSource) -> Result<Github, Error> {
490        install_crypto_provider();
491        let mut headers = HeaderMap::new();
492        headers.insert(
493            header::ACCEPT,
494            HeaderValue::from_static("application/vnd.github+json"),
495        );
496        headers.insert(
497            "x-github-api-version",
498            HeaderValue::from_static("2022-11-28"),
499        );
500        let http = reqwest::Client::builder()
501            .user_agent(concat!("runnerd/", env!("CARGO_PKG_VERSION")))
502            .default_headers(headers)
503            .timeout(REQUEST_TIMEOUT)
504            .build()
505            .map_err(|source| Error::Http {
506                method: Method::GET,
507                path: base.to_owned(),
508                source,
509            })?;
510        Ok(Github {
511            http,
512            base: base.trim_end_matches('/').to_owned(),
513            token_source: Arc::from(token_source),
514            token: tokio::sync::Mutex::new(None),
515            etags: Mutex::new(HashMap::new()),
516            blocked_until: Mutex::new(None),
517        })
518    }
519
520    /// When the next request may be sent: `None` now, otherwise when the rate limit resets.
521    pub fn blocked_until(&self) -> Option<SystemTime> {
522        let blocked = *lock(&self.blocked_until);
523        blocked.filter(|until| *until > SystemTime::now())
524    }
525
526    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, Error> {
527        let response = self.send(Method::GET, path, None).await?;
528        let bytes = if response.status == StatusCode::NOT_MODIFIED {
529            let cached = lock(&self.etags).get(path).map(|(_, body)| body.clone());
530            match cached {
531                Some(body) => body,
532                None => {
533                    return Err(Error::Status {
534                        method: Method::GET,
535                        path: path.to_owned(),
536                        status: response.status,
537                        message: "not modified, but nothing is cached".to_owned(),
538                    });
539                }
540            }
541        } else {
542            let etag = response.etag.clone();
543            let response = expect_success(Method::GET, path, response)?;
544            if let Some(etag) = etag {
545                lock(&self.etags).insert(path.to_owned(), (etag, response.bytes.clone()));
546            }
547            response.bytes
548        };
549        decode(path, &bytes)
550    }
551
552    /// Sends a request with the token, once more with a fresh token after a 401. Records the rate limit.
553    async fn send(
554        &self,
555        method: Method,
556        path: &str,
557        body: Option<&serde_json::Value>,
558    ) -> Result<Response, Error> {
559        if let Some(until) = self.blocked_until() {
560            return Err(Error::RateLimited { until });
561        }
562        // The token GitHub refused, once it did.
563        let mut refused: Option<Token> = None;
564        loop {
565            let token = self.token(refused.as_ref()).await?;
566            let mut request = self
567                .http
568                .request(method.clone(), format!("{}{path}", self.base))
569                .bearer_auth(&token.0);
570            if let Some(body) = body {
571                request = request
572                    .header(header::CONTENT_TYPE, "application/json")
573                    .body(body.to_string());
574            }
575            if method == Method::GET
576                && let Some((etag, _)) = lock(&self.etags).get(path)
577            {
578                request = request.header(header::IF_NONE_MATCH, etag.clone());
579            }
580            let response = request.send().await.map_err(|source| Error::Http {
581                method: method.clone(),
582                path: path.to_owned(),
583                source: source.without_url(),
584            })?;
585            let status = response.status();
586            if status == StatusCode::UNAUTHORIZED && refused.is_none() {
587                refused = Some(token);
588                continue;
589            }
590            if let Some(until) = rate_limited_until(status, response.headers(), SystemTime::now()) {
591                *lock(&self.blocked_until) = Some(until);
592                if matches!(
593                    status,
594                    StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
595                ) {
596                    return Err(Error::RateLimited { until });
597                }
598            }
599            let etag = response.headers().get(header::ETAG).cloned();
600            let bytes = response.bytes().await.map_err(|source| Error::Http {
601                method: method.clone(),
602                path: path.to_owned(),
603                source: source.without_url(),
604            })?;
605            return Ok(Response {
606                status,
607                etag,
608                bytes: bytes.to_vec(),
609            });
610        }
611    }
612
613    /// The token, read from the source if there is none yet or it is still the one GitHub `refused`: of several
614    /// requests refused at once, the first reads a new one and the others take it.
615    async fn token(&self, refused: Option<&Token>) -> Result<Token, Error> {
616        let mut token = self.token.lock().await;
617        let stale = match (&*token, refused) {
618            (None, _) => true,
619            (Some(current), Some(refused)) => current.0 == refused.0,
620            (Some(_), None) => false,
621        };
622        if stale {
623            let source = self.token_source.clone();
624            *token = Some(
625                tokio::task::spawn_blocking(move || source())
626                    .await
627                    .map_err(|error| {
628                        Error::Token(format!("the token source stopped: {error}"))
629                    })??,
630            );
631        }
632        token
633            .clone()
634            .ok_or_else(|| Error::Token("the token source returned nothing".to_owned()))
635    }
636}
637
638impl GithubApi for Github {
639    async fn queued_jobs(&self, repository: &str) -> Result<Vec<QueuedJob>, Error> {
640        let mut jobs = Vec::new();
641        let mut listed = Vec::new();
642        for status in ["queued", "in_progress"] {
643            let runs: Runs = self
644                .get(&format!(
645                    "/repos/{repository}/actions/runs?status={status}&per_page=100"
646                ))
647                .await?;
648            for run in runs.workflow_runs {
649                let path = jobs_path(repository, run.id);
650                let run_jobs: Jobs = self.get(&path).await?;
651                listed.push(path);
652                jobs.extend(
653                    run_jobs
654                        .jobs
655                        .into_iter()
656                        .filter(|job| job.status == "queued")
657                        .map(|job| QueuedJob {
658                            id: job.id,
659                            run_id: job.run_id,
660                            name: job.name,
661                            labels: job.labels,
662                            created_at: job
663                                .created_at
664                                .as_deref()
665                                .and_then(time::parse_ms)
666                                .unwrap_or(0),
667                            run_attempt: job.run_attempt.unwrap_or(1),
668                            workflow: job.workflow_name,
669                            branch: job.head_branch,
670                            url: job.html_url,
671                        }),
672                );
673            }
674        }
675        // The cached answers of runs that ended: nothing asks for them again.
676        let runs_prefix = format!("/repos/{repository}/actions/runs/");
677        lock(&self.etags).retain(|path, _| {
678            !(path.starts_with(&runs_prefix) && path.contains("/jobs?")) || listed.contains(path)
679        });
680        jobs.sort_by_key(|job| job.id);
681        jobs.dedup_by_key(|job| job.id);
682        Ok(jobs)
683    }
684
685    async fn runners(&self, repository: &str) -> Result<Vec<Runner>, Error> {
686        let runners: Runners = self
687            .get(&format!("/repos/{repository}/actions/runners?per_page=100"))
688            .await?;
689        Ok(runners.runners)
690    }
691
692    async fn jit_config(
693        &self,
694        repository: &str,
695        name: &str,
696        labels: &[String],
697    ) -> Result<(Runner, JitConfig), Error> {
698        let body = serde_json::json!({
699            "name": name,
700            // The repository's default runner group.
701            "runner_group_id": 1,
702            "labels": labels,
703            "work_folder": "_work",
704        });
705        let path = format!("/repos/{repository}/actions/runners/generate-jitconfig");
706        let response = self.send(Method::POST, &path, Some(&body)).await?;
707        let response = expect_success(Method::POST, &path, response)?;
708        let jit: JitResponse = decode(&path, &response.bytes)?;
709        Ok((jit.runner, JitConfig(jit.encoded_jit_config)))
710    }
711
712    async fn delete_runner(&self, repository: &str, id: u64) -> Result<Deleted, Error> {
713        let path = format!("/repos/{repository}/actions/runners/{id}");
714        let response = self.send(Method::DELETE, &path, None).await?;
715        match response.status {
716            StatusCode::NOT_FOUND => Ok(Deleted::NotFound),
717            StatusCode::UNPROCESSABLE_ENTITY => Ok(Deleted::Busy),
718            _ => expect_success(Method::DELETE, &path, response).map(|_| Deleted::Deleted),
719        }
720    }
721
722    async fn job(&self, repository: &str, id: u64) -> Result<JobDetails, Error> {
723        self.get(&format!("/repos/{repository}/actions/jobs/{id}"))
724            .await
725    }
726
727    async fn run_trigger(&self, repository: &str, run: u64) -> Result<RunTrigger, Error> {
728        let path = format!("/repos/{repository}/actions/runs/{run}");
729        let run: Result<RunBody, Error> = self.get(&path).await;
730        // Asked once per job: its large answer is not worth keeping for a conditional request.
731        lock(&self.etags).remove(&path);
732        let run = run?;
733        Ok(RunTrigger {
734            event: run.event,
735            head_branch: run.head_branch,
736            head_repository: run.head_repository.map(|head| head.full_name),
737            head_sha: run.head_sha,
738            by_user: Account::is_user(run.actor.as_ref())
739                && Account::is_user(run.triggering_actor.as_ref()),
740            pull_requests: run
741                .pull_requests
742                .into_iter()
743                .filter(|pull| pull.targets(repository))
744                .map(|pull| pull.number)
745                .collect(),
746        })
747    }
748
749    async fn pull_request(&self, repository: &str, number: u64) -> Result<PullRequest, Error> {
750        let path = format!("/repos/{repository}/pulls/{number}");
751        let pull: Result<PullRequestBody, Error> = self.get(&path).await;
752        lock(&self.etags).remove(&path);
753        let pull = pull?;
754        Ok(PullRequest {
755            by_user: Account::is_user(pull.user.as_ref()),
756            head_repository: pull.head.repo.map(|repo| repo.full_name),
757            head_branch: pull.head.branch,
758        })
759    }
760
761    async fn repository(&self, repository: &str) -> Result<RepositoryDetails, Error> {
762        let body: RepositoryBody = self.get(&format!("/repos/{repository}")).await?;
763        Ok(RepositoryDetails {
764            default_branch: body.default_branch,
765            visibility: body.visibility,
766        })
767    }
768
769    async fn on_branch(&self, repository: &str, branch: &str, sha: &str) -> Result<bool, Error> {
770        let commit =
771            |sha: &str| !sha.is_empty() && sha.bytes().all(|byte| byte.is_ascii_hexdigit());
772        if !commit(sha) {
773            return Ok(false);
774        }
775        // A branch name may hold `#`, `%` or `?`, which would end or change the path.
776        let base: String = branch
777            .bytes()
778            .map(|byte| {
779                if byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.') {
780                    char::from(byte).to_string()
781                } else {
782                    format!("%{byte:02X}")
783                }
784            })
785            .collect();
786        // Compare would read a bare name as git does, a tag before a branch: a tag `main` would pass for the
787        // branch. This endpoint knows only branches.
788        let path = format!("/repos/{repository}/branches/{base}");
789        let head: Result<BranchBody, Error> = self.get(&path).await;
790        lock(&self.etags).remove(&path);
791        let head = head?.commit.sha;
792        if !commit(&head) {
793            return Ok(false);
794        }
795        // The commits it lists are not needed: one is the fewest it sends.
796        let path = format!("/repos/{repository}/compare/{head}...{sha}?per_page=1");
797        let body: Result<CompareBody, Error> = self.get(&path).await;
798        lock(&self.etags).remove(&path);
799        // `behind`: the branch has moved on past it.
800        Ok(matches!(body?.status.as_str(), "identical" | "behind"))
801    }
802
803    async fn cancel_run(&self, repository: &str, run: u64) -> Result<(), Error> {
804        let path = format!("/repos/{repository}/actions/runs/{run}/cancel");
805        let response = self.send(Method::POST, &path, None).await?;
806        // 409: the run completed already.
807        if response.status == StatusCode::CONFLICT {
808            return Ok(());
809        }
810        expect_success(Method::POST, &path, response).map(|_| ())
811    }
812}
813
814fn jobs_path(repository: &str, run: u64) -> String {
815    format!("/repos/{repository}/actions/runs/{run}/jobs?filter=latest&per_page=100")
816}
817
818struct Response {
819    status: StatusCode,
820    etag: Option<HeaderValue>,
821    bytes: Vec<u8>,
822}
823
824fn expect_success(method: Method, path: &str, response: Response) -> Result<Response, Error> {
825    if response.status.is_success() {
826        return Ok(response);
827    }
828    let message = serde_json::from_slice::<ErrorBody>(&response.bytes).map_or_else(
829        |_| String::from_utf8_lossy(&response.bytes).into_owned(),
830        |body| body.message,
831    );
832    Err(Error::Status {
833        method,
834        path: path.to_owned(),
835        status: response.status,
836        message,
837    })
838}
839
840fn decode<T: DeserializeOwned>(path: &str, bytes: &[u8]) -> Result<T, Error> {
841    serde_json::from_slice(bytes).map_err(|source| Error::Decode {
842        path: path.to_owned(),
843        source,
844    })
845}
846
847/// When requests may resume, if this response says the rate limit is used up: the primary limit's reset time, or
848/// a secondary limit's `retry-after`.
849fn rate_limited_until(
850    status: StatusCode,
851    headers: &HeaderMap,
852    now: SystemTime,
853) -> Option<SystemTime> {
854    let number = |name: &str| {
855        headers
856            .get(name)
857            .and_then(|value| value.to_str().ok())
858            .and_then(|value| value.trim().parse::<u64>().ok())
859    };
860    if number("x-ratelimit-remaining") == Some(0)
861        && let Some(reset) = number("x-ratelimit-reset")
862    {
863        return Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset));
864    }
865    if matches!(
866        status,
867        StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
868    ) && let Some(seconds) = number("retry-after")
869    {
870        return Some(now + Duration::from_secs(seconds));
871    }
872    None
873}
874
875#[cfg(test)]
876mod tests;