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    /// Moves `job` to `attempt` once GitHub said which attempt it is of, when it was noted under another of the
208    /// run's: a job a runner took is first noted under its run's latest attempt known, which for a re-run's first
209    /// job is the attempt before. An attempt it alone made is forgotten, and its time goes with the job; a job that
210    /// joined an older attempt counts from `at` (ms). A job not followed stays so, so a run forgotten while its job
211    /// ran is not counted again.
212    pub fn place(
213        &mut self,
214        repository: &str,
215        attempt: Attempt,
216        job: u64,
217        workflow: Option<&str>,
218        at: u64,
219    ) {
220        // When the job was seen: an attempt it alone made began with it, while one it joined began before it.
221        let mut seen: Option<u64> = None;
222        let mut emptied = Vec::new();
223        for (other, run) in self.attempts_mut(attempt.run_id) {
224            if *other != attempt && run.jobs.remove(&job) {
225                if run.jobs.is_empty() {
226                    emptied.push(*other);
227                    seen = Some(run.first.min(seen.unwrap_or(at)));
228                } else {
229                    seen = Some(seen.unwrap_or(at));
230                }
231            }
232        }
233        for other in emptied {
234            self.0.remove(&other);
235        }
236        if let Some(seen) = seen {
237            self.note(repository, attempt, job, workflow, seen);
238        }
239    }
240
241    /// Names the workflow of run `run_id`'s attempts followed whose workflow was not known.
242    pub fn name(&mut self, run_id: u64, workflow: &str) {
243        for (_, run) in self.attempts_mut(run_id) {
244            if run.workflow.is_none() {
245                run.workflow = Some(workflow.to_owned());
246            }
247        }
248    }
249
250    /// The latest attempt of run `run_id` followed, or its first.
251    pub fn latest(&self, run_id: u64) -> Attempt {
252        self.0
253            .range(
254                Attempt {
255                    run_id,
256                    run_attempt: 0,
257                }..=Attempt {
258                    run_id,
259                    run_attempt: u32::MAX,
260                },
261            )
262            .next_back()
263            .map_or(
264                Attempt {
265                    run_id,
266                    run_attempt: 1,
267                },
268                |(attempt, _)| *attempt,
269            )
270    }
271
272    fn attempts_mut(&mut self, run_id: u64) -> impl Iterator<Item = (&Attempt, &mut Run)> {
273        self.0.range_mut(
274            Attempt {
275                run_id,
276                run_attempt: 0,
277            }..=Attempt {
278                run_id,
279                run_attempt: u32::MAX,
280            },
281        )
282    }
283
284    /// Forgets the runs whose first job came `FORGET_AFTER` before `now` (ms). One that stopped counting is kept
285    /// until then, so a job of it seen again, as after a restart, does not count it again as a new run.
286    pub fn expire(&mut self, now: u64) {
287        let after = u64::try_from(FORGET_AFTER.as_millis()).unwrap_or(u64::MAX);
288        self.0
289            .retain(|_, run| now.saturating_sub(run.first) < after);
290    }
291
292    /// The jobs `repository`'s runs are expected to queue still at `now` (ms): per run that still counts, its
293    /// workflow's jobs per run less the jobs of it seen.
294    pub fn remaining(&self, repository: &str, learned: &Learned, now: u64) -> usize {
295        self.0
296            .values()
297            .filter(|run| run.repository == repository && run.counts(learned, now))
298            .filter_map(|run| {
299                let burst = learned.burst(repository, run.workflow.as_deref()?)?;
300                Some(burst.jobs.saturating_sub(run.jobs.len()))
301            })
302            .sum()
303    }
304}
305
306impl Run {
307    /// Whether it still counts at `now` (ms): until its workflow's span plus `IDLE_TIMEOUT` after its first job.
308    fn counts(&self, learned: &Learned, now: u64) -> bool {
309        let span = self
310            .workflow
311            .as_deref()
312            .and_then(|workflow| learned.burst(&self.repository, workflow))
313            .map_or(UNKNOWN_SPAN, |burst| burst.span + IDLE_TIMEOUT);
314        now.saturating_sub(self.first) < u64::try_from(span.as_millis()).unwrap_or(u64::MAX)
315    }
316}
317
318/// How many warm runners a repository with `warm` configured wants: `warm` while it is `active`, plus the jobs its
319/// runs are expected to queue still (`remaining`). None with `warm = 0`.
320pub fn target(warm: u32, active: bool, remaining: usize) -> usize {
321    if warm == 0 {
322        return 0;
323    }
324    let base = if active {
325        usize::try_from(warm).unwrap_or(usize::MAX)
326    } else {
327        0
328    };
329    base.saturating_add(remaining)
330}
331
332#[cfg(test)]
333mod tests;
runnerd_core/scheduler/warm.rs

runnerd_core/scheduler/
warm.rs
