runnerd_core/scheduler/
warm.rs

1//! How many warm runners each repository wants (ADR 0050): its `warm` while it is active, by its jobs of the last hour
2//! and the hours of the day its jobs usually come, plus the jobs its workflow runs seen so far are expected to queue
3//! still, from the history. Pure: the engine reads the history and the clock and passes them in.
4
5use std::collections::{BTreeMap, BTreeSet};
6use std::time::Duration;
7
8use super::IDLE_TIMEOUT;
9
10/// How long after a repository's last job it stays active.
11pub const ACTIVE_AFTER: Duration = Duration::from_secs(3600);
12
13/// How far back the history is read, in milliseconds: 14 days.
14pub const LOOKBACK: u64 = 14 * DAY;
15
16/// An hour of the day is a repository's usual one when its jobs were queued in it on this many days of `LOOKBACK`.
17const USUAL_DAYS: usize = 3;
18
19/// A workflow's runs before its jobs per run are expected of the next.
20const MIN_RUNS: usize = 3;
21
22/// How long a run is remembered after its first job, whether or not it still counts.
23const FORGET_AFTER: Duration = Duration::from_hours(24);
24
25/// How long a run whose workflow is not known yet counts.
26const UNKNOWN_SPAN: Duration = Duration::from_secs(3600);
27
28const HOUR: u64 = 3600 * 1000;
29
30/// A run as the history tells runs apart: repository, workflow, run ID and attempt.
31type RunKey<'seen> = (&'seen str, &'seen str, u64, u32);
32const DAY: u64 = 24 * HOUR;
33
34/// A job the history saw queued.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct Seen {
37    /// `owner/name`.
38    pub repository: String,
39    /// The workflow's name, when known.
40    pub workflow: Option<String>,
41    /// The workflow run it belongs to.
42    pub run_id: u64,
43    /// 1 for the run's first attempt.
44    pub run_attempt: u32,
45    /// When GitHub queued it, in milliseconds since the epoch.
46    pub queued_at: u64,
47}
48
49/// What a workflow's runs are expected to queue.
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub struct Burst {
52    /// The median number of jobs per run.
53    pub jobs: usize,
54    /// The 90th percentile of the time from a run's first job queued to its last.
55    pub span: Duration,
56}
57
58/// What warm runners learn from the history: when each repository's jobs come, and how many jobs its workflows'
59/// runs queue.
60#[derive(Debug, Clone, Default, PartialEq, Eq)]
61pub struct Learned {
62    /// The hours of the day (UTC) each repository's jobs usually come in.
63    usual: BTreeMap<String, BTreeSet<u64>>,
64    /// When each repository's last job was queued, in milliseconds.
65    last: BTreeMap<String, u64>,
66    /// By repository and workflow.
67    bursts: BTreeMap<(String, String), Burst>,
68}
69
70impl Learned {
71    /// Learns from the jobs queued in the `LOOKBACK` before `now` (ms).
72    pub fn new(seen: &[Seen], now: u64) -> Learned {
73        let since = now.saturating_sub(LOOKBACK);
74        let mut learned = Learned::default();
75        let mut days: BTreeMap<(&str, u64), BTreeSet<u64>> = BTreeMap::new();
76        // Each run's jobs, and when its first and last were queued.
77        let mut runs: BTreeMap<RunKey<'_>, (usize, u64, u64)> = BTreeMap::new();
78        for job in seen.iter().filter(|job| job.queued_at >= since) {
79            learned.note(&job.repository, job.queued_at);
80            days.entry((&job.repository, job.queued_at % DAY / HOUR))
81                .or_default()
82                .insert(job.queued_at / DAY);
83            if let Some(workflow) = &job.workflow {
84                let run = runs
85                    .entry((&job.repository, workflow, job.run_id, job.run_attempt))
86                    .or_insert((0, job.queued_at, job.queued_at));
87                run.0 += 1;
88                run.1 = run.1.min(job.queued_at);
89                run.2 = run.2.max(job.queued_at);
90            }
91        }
92        for ((repository, hour), days) in days {
93            if days.len() >= USUAL_DAYS {
94                learned
95                    .usual
96                    .entry(repository.to_owned())
97                    .or_default()
98                    .insert(hour);
99            }
100        }
101        let mut workflows: BTreeMap<(&str, &str), Vec<(usize, u64)>> = BTreeMap::new();
102        for ((repository, workflow, ..), (jobs, first, last)) in runs {
103            workflows
104                .entry((repository, workflow))
105                .or_default()
106                .push((jobs, last - first));
107        }
108        for ((repository, workflow), runs) in workflows {
109            if runs.len() < MIN_RUNS {
110                continue;
111            }
112            let mut jobs: Vec<usize> = runs.iter().map(|(jobs, _)| *jobs).collect();
113            let mut spans: Vec<u64> = runs.iter().map(|(_, span)| *span).collect();
114            jobs.sort_unstable();
115            spans.sort_unstable();
116            learned.bursts.insert(
117                (repository.to_owned(), workflow.to_owned()),
118                Burst {
119                    jobs: jobs[(jobs.len() - 1) / 2],
120                    span: Duration::from_millis(spans[(spans.len() * 90).div_ceil(100) - 1]),
121                },
122            );
123        }
124        learned
125    }
126
127    /// Keeps the last jobs `older` noted that this one has not read yet.
128    pub fn absorb(&mut self, older: &Learned) {
129        for (repository, at) in &older.last {
130            self.note(repository, *at);
131        }
132    }
133
134    /// Notes a job of `repository` queued or taken at `at` (ms).
135    pub fn note(&mut self, repository: &str, at: u64) {
136        let last = self.last.entry(repository.to_owned()).or_default();
137        *last = (*last).max(at);
138    }
139
140    /// Whether `repository` is active at `now` (ms): its last job came within `ACTIVE_AFTER`, or `now` is in an hour
141    /// of the day its jobs usually come in.
142    pub fn active(&self, repository: &str, now: u64) -> bool {
143        let after = u64::try_from(ACTIVE_AFTER.as_millis()).unwrap_or(u64::MAX);
144        self.last
145            .get(repository)
146            .is_some_and(|last| now.saturating_sub(*last) < after)
147            || self
148                .usual
149                .get(repository)
150                .is_some_and(|hours| hours.contains(&(now % DAY / HOUR)))
151    }
152
153    /// What runs of `workflow` in `repository` are expected to queue, once it has `MIN_RUNS` runs.
154    pub fn burst(&self, repository: &str, workflow: &str) -> Option<Burst> {
155        self.bursts
156            .get(&(repository.to_owned(), workflow.to_owned()))
157            .copied()
158    }
159}
160
161/// A workflow run seen lately, with the jobs of it seen so far.
162#[derive(Debug, Clone, PartialEq, Eq)]
163struct Run {
164    repository: String,
165    workflow: Option<String>,
166    jobs: BTreeSet<u64>,
167    /// When its first job was queued or seen, in milliseconds.
168    first: u64,
169}
170
171/// An attempt of a workflow run: a re-run keeps the run's ID, with the next attempt, and queues its jobs anew.
172#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
173pub struct Attempt {
174    /// The workflow run's ID.
175    pub run_id: u64,
176    /// 1 for the run's first attempt.
177    pub run_attempt: u32,
178}
179
180/// The workflow runs seen lately, by attempt: their jobs not queued yet get warm runners.
181#[derive(Debug, Clone, Default, PartialEq, Eq)]
182pub struct Runs(BTreeMap<Attempt, Run>);
183
184impl Runs {
185    /// Notes `job` of `attempt` in `repository`, queued or taken at `at` (ms), and its workflow when known.
186    pub fn note(
187        &mut self,
188        repository: &str,
189        attempt: Attempt,
190        job: u64,
191        workflow: Option<&str>,
192        at: u64,
193    ) {
194        let run = self.0.entry(attempt).or_insert_with(|| Run {
195            repository: repository.to_owned(),
196            workflow: None,
197            jobs: BTreeSet::new(),
198            first: at,
199        });
200        run.jobs.insert(job);
201        run.first = run.first.min(at);
202        if run.workflow.is_none() {
203            run.workflow = workflow.map(str::to_owned);
204        }
205    }
206
207    /// Names the workflow of run `run_id`'s attempts followed whose workflow was not known.
208    pub fn name(&mut self, run_id: u64, workflow: &str) {
209        for (_, run) in self.attempts_mut(run_id) {
210            if run.workflow.is_none() {
211                run.workflow = Some(workflow.to_owned());
212            }
213        }
214    }
215
216    /// The latest attempt of run `run_id` followed, or its first.
217    pub fn latest(&self, run_id: u64) -> Attempt {
218        self.0
219            .range(
220                Attempt {
221                    run_id,
222                    run_attempt: 0,
223                }..=Attempt {
224                    run_id,
225                    run_attempt: u32::MAX,
226                },
227            )
228            .next_back()
229            .map_or(
230                Attempt {
231                    run_id,
232                    run_attempt: 1,
233                },
234                |(attempt, _)| *attempt,
235            )
236    }
237
238    fn attempts_mut(&mut self, run_id: u64) -> impl Iterator<Item = (&Attempt, &mut Run)> {
239        self.0.range_mut(
240            Attempt {
241                run_id,
242                run_attempt: 0,
243            }..=Attempt {
244                run_id,
245                run_attempt: u32::MAX,
246            },
247        )
248    }
249
250    /// Forgets the runs whose first job came `FORGET_AFTER` before `now` (ms). One that stopped counting is kept
251    /// until then, so a job of it seen again, as after a restart, does not count it again as a new run.
252    pub fn expire(&mut self, now: u64) {
253        let after = u64::try_from(FORGET_AFTER.as_millis()).unwrap_or(u64::MAX);
254        self.0
255            .retain(|_, run| now.saturating_sub(run.first) < after);
256    }
257
258    /// The jobs `repository`'s runs are expected to queue still at `now` (ms): per run that still counts, its
259    /// workflow's jobs per run less the jobs of it seen.
260    pub fn remaining(&self, repository: &str, learned: &Learned, now: u64) -> usize {
261        self.0
262            .values()
263            .filter(|run| run.repository == repository && run.counts(learned, now))
264            .filter_map(|run| {
265                let burst = learned.burst(repository, run.workflow.as_deref()?)?;
266                Some(burst.jobs.saturating_sub(run.jobs.len()))
267            })
268            .sum()
269    }
270}
271
272impl Run {
273    /// Whether it still counts at `now` (ms): until its workflow's span plus `IDLE_TIMEOUT` after its first job.
274    fn counts(&self, learned: &Learned, now: u64) -> bool {
275        let span = self
276            .workflow
277            .as_deref()
278            .and_then(|workflow| learned.burst(&self.repository, workflow))
279            .map_or(UNKNOWN_SPAN, |burst| burst.span + IDLE_TIMEOUT);
280        now.saturating_sub(self.first) < u64::try_from(span.as_millis()).unwrap_or(u64::MAX)
281    }
282}
283
284/// How many warm runners a repository with `warm` configured wants: `warm` while it is `active`, plus the jobs its
285/// runs are expected to queue still (`remaining`). None with `warm = 0`.
286pub fn target(warm: u32, active: bool, remaining: usize) -> usize {
287    if warm == 0 {
288        return 0;
289    }
290    let base = if active {
291        usize::try_from(warm).unwrap_or(usize::MAX)
292    } else {
293        0
294    };
295    base.saturating_add(remaining)
296}
297
298#[cfg(test)]
299mod tests;
runnerd_core/scheduler/warm.rs

runnerd_core/scheduler/
warm.rs
