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}
187
188fn first_attempt() -> u32 {
189    1
190}
191
192/// A step of `JobDetails`.
193#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
194pub struct JobStep {
195    /// GitHub's step number.
196    pub number: u32,
197    /// The name GitHub shows.
198    pub name: String,
199    /// `pending`, `queued`, `in_progress` or `completed`.
200    pub status: String,
201    /// Once completed: `success`, `failure`, `skipped`, ….
202    pub conclusion: Option<String>,
203    /// When it started.
204    pub started_at: Option<String>,
205    /// When it completed.
206    pub completed_at: Option<String>,
207}
208
209/// A runner registered on a repository.
210#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
211pub struct Runner {
212    /// The runner's ID.
213    pub id: u64,
214    /// Its name; runnerd's runners start with the host's prefix.
215    pub name: String,
216    /// `online` or `offline`.
217    pub status: String,
218    /// Whether it runs a job.
219    pub busy: bool,
220}
221
222/// What deleting a runner did.
223#[derive(Debug, Clone, Copy, PartialEq, Eq)]
224pub enum Deleted {
225    /// GitHub removed it.
226    Deleted,
227    /// It was gone already: an ephemeral runner removes itself after its job.
228    NotFound,
229    /// GitHub refused because the runner runs a job.
230    Busy,
231}
232
233/// The GitHub calls the scheduler makes. `Github` is the real client; the engine's tests use a fake.
234pub trait GithubApi: Send + Sync {
235    /// The repository's jobs waiting for a runner, in queued runs and in runs that started some of their jobs.
236    fn queued_jobs(
237        &self,
238        repository: &str,
239    ) -> impl Future<Output = Result<Vec<QueuedJob>, Error>> + Send;
240    /// The repository's self-hosted runners.
241    fn runners(&self, repository: &str) -> impl Future<Output = Result<Vec<Runner>, Error>> + Send;
242    /// Registers a runner for one job and returns it with the config `run.sh --jitconfig` takes.
243    fn jit_config(
244        &self,
245        repository: &str,
246        name: &str,
247        labels: &[String],
248    ) -> impl Future<Output = Result<(Runner, JitConfig), Error>> + Send;
249    /// Removes a runner from the repository.
250    fn delete_runner(
251        &self,
252        repository: &str,
253        id: u64,
254    ) -> impl Future<Output = Result<Deleted, Error>> + Send;
255    /// One job's progress and steps.
256    fn job(
257        &self,
258        repository: &str,
259        id: u64,
260    ) -> impl Future<Output = Result<JobDetails, Error>> + Send;
261    /// What started a workflow run.
262    fn run_trigger(
263        &self,
264        repository: &str,
265        run: u64,
266    ) -> impl Future<Output = Result<RunTrigger, Error>> + Send;
267    /// The repository's default branch.
268    fn default_branch(
269        &self,
270        repository: &str,
271    ) -> impl Future<Output = Result<String, Error>> + Send;
272    /// Whether `branch` of the repository has the commit `sha`: its head or an ancestor of it. A tag of the same
273    /// name, which git would take for it, does not count.
274    fn on_branch(
275        &self,
276        repository: &str,
277        branch: &str,
278        sha: &str,
279    ) -> impl Future<Output = Result<bool, Error>> + Send;
280    /// Cancels a workflow run, and so every job in it: GitHub cancels runs, not single jobs. A run that already
281    /// completed is left as it is.
282    fn cancel_run(
283        &self,
284        repository: &str,
285        run: u64,
286    ) -> impl Future<Output = Result<(), Error>> + Send;
287}
288
289/// A GitHub API client for one token. Requests on one URL are conditional on its last `ETag`, so an unchanged
290/// answer costs no rate limit.
291pub struct Github {
292    http: reqwest::Client,
293    base: String,
294    /// Run on Tokio's blocking pool: `gh` may wait on a keyring for seconds.
295    token_source: Arc<dyn Fn() -> Result<Token, Error> + Send + Sync>,
296    /// An async lock, held while the source runs: requests that need a token wait without blocking a worker, and
297    /// concurrent 401s read it once.
298    token: tokio::sync::Mutex<Option<Token>>,
299    // Locked with `sync::lock`: a panic while holding one of these locks leaves plain data behind, still usable.
300    etags: Mutex<HashMap<String, (HeaderValue, Vec<u8>)>>,
301    blocked_until: Mutex<Option<SystemTime>>,
302}
303
304impl fmt::Debug for Github {
305    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
306        f.debug_struct("Github")
307            .field("base", &self.base)
308            .finish_non_exhaustive()
309    }
310}
311
312#[derive(Deserialize)]
313struct Runs {
314    workflow_runs: Vec<Run>,
315}
316
317#[derive(Deserialize)]
318struct Run {
319    id: u64,
320}
321
322#[derive(Deserialize)]
323struct Jobs {
324    jobs: Vec<Job>,
325}
326
327#[derive(Deserialize)]
328struct Job {
329    id: u64,
330    run_id: u64,
331    name: String,
332    status: String,
333    labels: Vec<String>,
334    created_at: Option<String>,
335    run_attempt: Option<u32>,
336    workflow_name: Option<String>,
337    head_branch: Option<String>,
338    html_url: Option<String>,
339}
340
341#[derive(Deserialize)]
342struct RunBody {
343    event: String,
344    head_branch: Option<String>,
345    head_repository: Option<RepositoryName>,
346    head_sha: String,
347}
348
349#[derive(Deserialize)]
350struct RepositoryName {
351    full_name: String,
352}
353
354#[derive(Deserialize)]
355struct CompareBody {
356    status: String,
357}
358
359#[derive(Deserialize)]
360struct BranchBody {
361    commit: CommitBody,
362}
363
364#[derive(Deserialize)]
365struct CommitBody {
366    sha: String,
367}
368
369#[derive(Deserialize)]
370struct RepositoryBody {
371    default_branch: String,
372}
373
374#[derive(Deserialize)]
375struct Runners {
376    runners: Vec<Runner>,
377}
378
379#[derive(Deserialize)]
380struct JitResponse {
381    runner: Runner,
382    encoded_jit_config: String,
383}
384
385#[derive(Deserialize)]
386struct ErrorBody {
387    message: String,
388}
389
390impl Github {
391    /// A client for `base` (`API_URL`, or a test server) that takes its token from `token_source`.
392    pub fn new(base: &str, token_source: TokenSource) -> Result<Github, Error> {
393        install_crypto_provider();
394        let mut headers = HeaderMap::new();
395        headers.insert(
396            header::ACCEPT,
397            HeaderValue::from_static("application/vnd.github+json"),
398        );
399        headers.insert(
400            "x-github-api-version",
401            HeaderValue::from_static("2022-11-28"),
402        );
403        let http = reqwest::Client::builder()
404            .user_agent(concat!("runnerd/", env!("CARGO_PKG_VERSION")))
405            .default_headers(headers)
406            .timeout(REQUEST_TIMEOUT)
407            .build()
408            .map_err(|source| Error::Http {
409                method: Method::GET,
410                path: base.to_owned(),
411                source,
412            })?;
413        Ok(Github {
414            http,
415            base: base.trim_end_matches('/').to_owned(),
416            token_source: Arc::from(token_source),
417            token: tokio::sync::Mutex::new(None),
418            etags: Mutex::new(HashMap::new()),
419            blocked_until: Mutex::new(None),
420        })
421    }
422
423    /// When the next request may be sent: `None` now, otherwise when the rate limit resets.
424    pub fn blocked_until(&self) -> Option<SystemTime> {
425        let blocked = *lock(&self.blocked_until);
426        blocked.filter(|until| *until > SystemTime::now())
427    }
428
429    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, Error> {
430        let response = self.send(Method::GET, path, None).await?;
431        let bytes = if response.status == StatusCode::NOT_MODIFIED {
432            let cached = lock(&self.etags).get(path).map(|(_, body)| body.clone());
433            match cached {
434                Some(body) => body,
435                None => {
436                    return Err(Error::Status {
437                        method: Method::GET,
438                        path: path.to_owned(),
439                        status: response.status,
440                        message: "not modified, but nothing is cached".to_owned(),
441                    });
442                }
443            }
444        } else {
445            let etag = response.etag.clone();
446            let response = expect_success(Method::GET, path, response)?;
447            if let Some(etag) = etag {
448                lock(&self.etags).insert(path.to_owned(), (etag, response.bytes.clone()));
449            }
450            response.bytes
451        };
452        decode(path, &bytes)
453    }
454
455    /// Sends a request with the token, once more with a fresh token after a 401. Records the rate limit.
456    async fn send(
457        &self,
458        method: Method,
459        path: &str,
460        body: Option<&serde_json::Value>,
461    ) -> Result<Response, Error> {
462        if let Some(until) = self.blocked_until() {
463            return Err(Error::RateLimited { until });
464        }
465        // The token GitHub refused, once it did.
466        let mut refused: Option<Token> = None;
467        loop {
468            let token = self.token(refused.as_ref()).await?;
469            let mut request = self
470                .http
471                .request(method.clone(), format!("{}{path}", self.base))
472                .bearer_auth(&token.0);
473            if let Some(body) = body {
474                request = request
475                    .header(header::CONTENT_TYPE, "application/json")
476                    .body(body.to_string());
477            }
478            if method == Method::GET
479                && let Some((etag, _)) = lock(&self.etags).get(path)
480            {
481                request = request.header(header::IF_NONE_MATCH, etag.clone());
482            }
483            let response = request.send().await.map_err(|source| Error::Http {
484                method: method.clone(),
485                path: path.to_owned(),
486                source: source.without_url(),
487            })?;
488            let status = response.status();
489            if status == StatusCode::UNAUTHORIZED && refused.is_none() {
490                refused = Some(token);
491                continue;
492            }
493            if let Some(until) = rate_limited_until(status, response.headers(), SystemTime::now()) {
494                *lock(&self.blocked_until) = Some(until);
495                if matches!(
496                    status,
497                    StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
498                ) {
499                    return Err(Error::RateLimited { until });
500                }
501            }
502            let etag = response.headers().get(header::ETAG).cloned();
503            let bytes = response.bytes().await.map_err(|source| Error::Http {
504                method: method.clone(),
505                path: path.to_owned(),
506                source: source.without_url(),
507            })?;
508            return Ok(Response {
509                status,
510                etag,
511                bytes: bytes.to_vec(),
512            });
513        }
514    }
515
516    /// The token, read from the source if there is none yet or it is still the one GitHub `refused`: of several
517    /// requests refused at once, the first reads a new one and the others take it.
518    async fn token(&self, refused: Option<&Token>) -> Result<Token, Error> {
519        let mut token = self.token.lock().await;
520        let stale = match (&*token, refused) {
521            (None, _) => true,
522            (Some(current), Some(refused)) => current.0 == refused.0,
523            (Some(_), None) => false,
524        };
525        if stale {
526            let source = self.token_source.clone();
527            *token = Some(
528                tokio::task::spawn_blocking(move || source())
529                    .await
530                    .map_err(|error| {
531                        Error::Token(format!("the token source stopped: {error}"))
532                    })??,
533            );
534        }
535        token
536            .clone()
537            .ok_or_else(|| Error::Token("the token source returned nothing".to_owned()))
538    }
539}
540
541impl GithubApi for Github {
542    async fn queued_jobs(&self, repository: &str) -> Result<Vec<QueuedJob>, Error> {
543        let mut jobs = Vec::new();
544        let mut listed = Vec::new();
545        for status in ["queued", "in_progress"] {
546            let runs: Runs = self
547                .get(&format!(
548                    "/repos/{repository}/actions/runs?status={status}&per_page=100"
549                ))
550                .await?;
551            for run in runs.workflow_runs {
552                let path = jobs_path(repository, run.id);
553                let run_jobs: Jobs = self.get(&path).await?;
554                listed.push(path);
555                jobs.extend(
556                    run_jobs
557                        .jobs
558                        .into_iter()
559                        .filter(|job| job.status == "queued")
560                        .map(|job| QueuedJob {
561                            id: job.id,
562                            run_id: job.run_id,
563                            name: job.name,
564                            labels: job.labels,
565                            created_at: job
566                                .created_at
567                                .as_deref()
568                                .and_then(time::parse_ms)
569                                .unwrap_or(0),
570                            run_attempt: job.run_attempt.unwrap_or(1),
571                            workflow: job.workflow_name,
572                            branch: job.head_branch,
573                            url: job.html_url,
574                        }),
575                );
576            }
577        }
578        // The cached answers of runs that ended: nothing asks for them again.
579        let runs_prefix = format!("/repos/{repository}/actions/runs/");
580        lock(&self.etags).retain(|path, _| {
581            !(path.starts_with(&runs_prefix) && path.contains("/jobs?")) || listed.contains(path)
582        });
583        jobs.sort_by_key(|job| job.id);
584        jobs.dedup_by_key(|job| job.id);
585        Ok(jobs)
586    }
587
588    async fn runners(&self, repository: &str) -> Result<Vec<Runner>, Error> {
589        let runners: Runners = self
590            .get(&format!("/repos/{repository}/actions/runners?per_page=100"))
591            .await?;
592        Ok(runners.runners)
593    }
594
595    async fn jit_config(
596        &self,
597        repository: &str,
598        name: &str,
599        labels: &[String],
600    ) -> Result<(Runner, JitConfig), Error> {
601        let body = serde_json::json!({
602            "name": name,
603            // The repository's default runner group.
604            "runner_group_id": 1,
605            "labels": labels,
606            "work_folder": "_work",
607        });
608        let path = format!("/repos/{repository}/actions/runners/generate-jitconfig");
609        let response = self.send(Method::POST, &path, Some(&body)).await?;
610        let response = expect_success(Method::POST, &path, response)?;
611        let jit: JitResponse = decode(&path, &response.bytes)?;
612        Ok((jit.runner, JitConfig(jit.encoded_jit_config)))
613    }
614
615    async fn delete_runner(&self, repository: &str, id: u64) -> Result<Deleted, Error> {
616        let path = format!("/repos/{repository}/actions/runners/{id}");
617        let response = self.send(Method::DELETE, &path, None).await?;
618        match response.status {
619            StatusCode::NOT_FOUND => Ok(Deleted::NotFound),
620            StatusCode::UNPROCESSABLE_ENTITY => Ok(Deleted::Busy),
621            _ => expect_success(Method::DELETE, &path, response).map(|_| Deleted::Deleted),
622        }
623    }
624
625    async fn job(&self, repository: &str, id: u64) -> Result<JobDetails, Error> {
626        self.get(&format!("/repos/{repository}/actions/jobs/{id}"))
627            .await
628    }
629
630    async fn run_trigger(&self, repository: &str, run: u64) -> Result<RunTrigger, Error> {
631        let path = format!("/repos/{repository}/actions/runs/{run}");
632        let run: Result<RunBody, Error> = self.get(&path).await;
633        // Asked once per job: its large answer is not worth keeping for a conditional request.
634        lock(&self.etags).remove(&path);
635        let run = run?;
636        Ok(RunTrigger {
637            event: run.event,
638            head_branch: run.head_branch,
639            head_repository: run.head_repository.map(|head| head.full_name),
640            head_sha: run.head_sha,
641        })
642    }
643
644    async fn default_branch(&self, repository: &str) -> Result<String, Error> {
645        let body: RepositoryBody = self.get(&format!("/repos/{repository}")).await?;
646        Ok(body.default_branch)
647    }
648
649    async fn on_branch(&self, repository: &str, branch: &str, sha: &str) -> Result<bool, Error> {
650        let commit =
651            |sha: &str| !sha.is_empty() && sha.bytes().all(|byte| byte.is_ascii_hexdigit());
652        if !commit(sha) {
653            return Ok(false);
654        }
655        // A branch name may hold `#`, `%` or `?`, which would end or change the path.
656        let base: String = branch
657            .bytes()
658            .map(|byte| {
659                if byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.') {
660                    char::from(byte).to_string()
661                } else {
662                    format!("%{byte:02X}")
663                }
664            })
665            .collect();
666        // Compare would read a bare name as git does, a tag before a branch: a tag `main` would pass for the
667        // branch. This endpoint knows only branches.
668        let path = format!("/repos/{repository}/branches/{base}");
669        let head: Result<BranchBody, Error> = self.get(&path).await;
670        lock(&self.etags).remove(&path);
671        let head = head?.commit.sha;
672        if !commit(&head) {
673            return Ok(false);
674        }
675        // The commits it lists are not needed: one is the fewest it sends.
676        let path = format!("/repos/{repository}/compare/{head}...{sha}?per_page=1");
677        let body: Result<CompareBody, Error> = self.get(&path).await;
678        lock(&self.etags).remove(&path);
679        // `behind`: the branch has moved on past it.
680        Ok(matches!(body?.status.as_str(), "identical" | "behind"))
681    }
682
683    async fn cancel_run(&self, repository: &str, run: u64) -> Result<(), Error> {
684        let path = format!("/repos/{repository}/actions/runs/{run}/cancel");
685        let response = self.send(Method::POST, &path, None).await?;
686        // 409: the run completed already.
687        if response.status == StatusCode::CONFLICT {
688            return Ok(());
689        }
690        expect_success(Method::POST, &path, response).map(|_| ())
691    }
692}
693
694fn jobs_path(repository: &str, run: u64) -> String {
695    format!("/repos/{repository}/actions/runs/{run}/jobs?filter=latest&per_page=100")
696}
697
698struct Response {
699    status: StatusCode,
700    etag: Option<HeaderValue>,
701    bytes: Vec<u8>,
702}
703
704fn expect_success(method: Method, path: &str, response: Response) -> Result<Response, Error> {
705    if response.status.is_success() {
706        return Ok(response);
707    }
708    let message = serde_json::from_slice::<ErrorBody>(&response.bytes).map_or_else(
709        |_| String::from_utf8_lossy(&response.bytes).into_owned(),
710        |body| body.message,
711    );
712    Err(Error::Status {
713        method,
714        path: path.to_owned(),
715        status: response.status,
716        message,
717    })
718}
719
720fn decode<T: DeserializeOwned>(path: &str, bytes: &[u8]) -> Result<T, Error> {
721    serde_json::from_slice(bytes).map_err(|source| Error::Decode {
722        path: path.to_owned(),
723        source,
724    })
725}
726
727/// When requests may resume, if this response says the rate limit is used up: the primary limit's reset time, or
728/// a secondary limit's `retry-after`.
729fn rate_limited_until(
730    status: StatusCode,
731    headers: &HeaderMap,
732    now: SystemTime,
733) -> Option<SystemTime> {
734    let number = |name: &str| {
735        headers
736            .get(name)
737            .and_then(|value| value.to_str().ok())
738            .and_then(|value| value.trim().parse::<u64>().ok())
739    };
740    if number("x-ratelimit-remaining") == Some(0)
741        && let Some(reset) = number("x-ratelimit-reset")
742    {
743        return Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset));
744    }
745    if matches!(
746        status,
747        StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
748    ) && let Some(seconds) = number("retry-after")
749    {
750        return Some(now + Duration::from_secs(seconds));
751    }
752    None
753}
754
755#[cfg(test)]
756mod tests;
runnerd_core/github.rs

runnerd_core/
github.rs
