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