runnerd_core/
store.rs

1//! The job history: a SQLite database of jobs, their runners, steps, samples and logs, the events clients resume
2//! from, cache sizes and the paused repositories. Owns the schema and its migrations, the one writer and the event
3//! broadcast. Writes are messages to a writer thread, so the engine never waits on the disk; reads run on Tokio's
4//! blocking pool. It decides nothing about scheduling and stores no secret: no token, no JIT config, no runner
5//! `_diag` log.
6
7use std::collections::BTreeMap;
8use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
9use std::path::Path;
10use std::sync::{Arc, Mutex};
11use std::time::Duration;
12
13use rusqlite::{Connection, OpenFlags};
14use tokio::sync::{OwnedSemaphorePermit, Semaphore, broadcast, mpsc, oneshot};
15use tracing::warn;
16
17use crate::api::{self, Event, EventEnvelope};
18use crate::config::History;
19use crate::github::{JobDetails, QueuedJob};
20use crate::sync::lock;
21
22mod read;
23mod write;
24
25/// The schema's migrations, in order; the database's `user_version` is how many ran.
26const MIGRATIONS: &[&str] = &[
27    include_str!("store/migrations/0001_history.sql"),
28    include_str!("store/migrations/0002_runner_caps.sql"),
29    include_str!("store/migrations/0003_paused.sql"),
30];
31
32/// Bytes of log lines sent and not yet stored, at most: a job's capture waits beyond it, so a noisy job or a slow
33/// disk holds back that job's log, not the daemon's memory.
34pub(crate) const LOG_BACKLOG: u64 = 8 << 20;
35
36/// How many events a slow client may fall behind before its stream ends; it reconnects with `Last-Event-ID`.
37const EVENT_BUFFER: usize = 4096;
38
39/// Why the database cannot be used.
40#[derive(Debug, thiserror::Error)]
41pub enum StoreError {
42    /// SQLite failed.
43    #[error("{0}")]
44    Sqlite(#[from] rusqlite::Error),
45    /// A newer runnerd wrote the database.
46    #[error(
47        "the job history database is version {found}, newer than the {supported} this runnerd reads; \
48         run a newer runnerd or move the database aside"
49    )]
50    Newer {
51        /// The database's version.
52        found: u32,
53        /// The newest version this build knows.
54        supported: u32,
55    },
56    /// The directory or file cannot be created.
57    #[error("{0}")]
58    Io(#[from] std::io::Error),
59    /// A blocking read did not finish.
60    #[error("a database read stopped: {0}")]
61    Join(#[from] tokio::task::JoinError),
62}
63
64/// A runner container as the history records it.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub struct RunnerRecord {
67    /// The runner's name, also its container's.
68    pub name: String,
69    /// `owner/name`.
70    pub repository: String,
71    /// The profile it runs in.
72    pub profile: String,
73    /// 1-based.
74    pub slot: usize,
75    /// The Docker container ID.
76    pub container: String,
77    /// When the container started, in milliseconds since the epoch.
78    pub started_at: u64,
79    /// The container's memory cap in bytes.
80    pub memory_cap: Option<u64>,
81    /// The container's CPU limit in millicores.
82    pub cpu_cap: Option<u64>,
83}
84
85/// One sample of one runner's container.
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct SampleRecord {
88    /// The runner.
89    pub runner: String,
90    /// When, in milliseconds since the epoch.
91    pub at: u64,
92    /// Millicores over the interval before it.
93    pub cpu: u64,
94    /// CPU time so far, in microseconds.
95    pub cpu_usage: u64,
96    /// Bytes without page cache.
97    pub memory: u64,
98    /// Bytes received since the container started.
99    pub network_received: u64,
100    /// Bytes sent since the container started.
101    pub network_sent: u64,
102}
103
104/// The last measurement of the cache volumes.
105#[derive(Debug, Clone, Default, PartialEq, Eq)]
106pub struct Measured {
107    /// When it was taken (seconds); none before the first measurement.
108    pub at: Option<u64>,
109    /// Each volume's size in bytes, by name.
110    pub bytes: BTreeMap<String, u64>,
111}
112
113/// A job the history has not closed, as the engine needs it after a restart.
114#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct OpenJob {
116    /// GitHub's job ID.
117    pub id: u64,
118    /// `owner/name`.
119    pub repository: String,
120    /// `Queued` or `Running`.
121    pub state: api::JobState,
122    /// Its runner, once one took it.
123    pub runner: Option<String>,
124    /// When its runner was removed, in milliseconds since the epoch.
125    pub runner_removed_at: Option<u64>,
126}
127
128/// Where a job's log capture continues: the runner's page file and the byte offset in it.
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub struct LogPosition {
131    /// The page number, from 1.
132    pub page: u64,
133    /// Bytes of that page already stored as lines.
134    pub offset: u64,
135}
136
137/// How far the mirror's log was counted: the time of the last response counted, in nanoseconds since the epoch as
138/// Docker logs it, and how many responses logged at exactly that time were counted, so the next read counts the
139/// ones it was too early to see.
140#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
141pub struct MirrorPosition {
142    /// Nanoseconds since the epoch.
143    pub at: u64,
144    /// Responses logged at `at` already counted.
145    pub seen: u64,
146}
147
148/// What the writer thread is asked to do, in order.
149#[derive(Debug)]
150enum Write {
151    Queued {
152        repository: String,
153        job: QueuedJob,
154    },
155    RunnerStarted(RunnerRecord),
156    RunnerRemoved {
157        runner: String,
158        at: u64,
159    },
160    Claimed {
161        job: u64,
162        run_id: u64,
163        runner: String,
164        at: u64,
165    },
166    Details {
167        details: JobDetails,
168    },
169    Elsewhere {
170        details: JobDetails,
171    },
172    Lost {
173        job: u64,
174        at: u64,
175    },
176    Log {
177        job: u64,
178        lines: Vec<String>,
179        /// Where the capture resumes once these lines are stored; `None` for all but the last part of a batch.
180        position: Option<LogPosition>,
181        max_bytes: u64,
182        /// The lines' share of `LOG_BACKLOG`, given back once they are stored.
183        _backlog: OwnedSemaphorePermit,
184    },
185    Samples(Vec<SampleRecord>),
186    HostSample(api::HostSample),
187    MirrorBytes {
188        bytes: Vec<(String, u64)>,
189        until: MirrorPosition,
190        /// Whether the count committed.
191        done: oneshot::Sender<bool>,
192    },
193    CacheSizes {
194        at: u64,
195        sizes: BTreeMap<String, u64>,
196    },
197    Event(Event),
198    Paused {
199        repositories: Vec<String>,
200        /// Whether it committed.
201        done: oneshot::Sender<bool>,
202    },
203    Prune {
204        now: u64,
205        history: History,
206    },
207    Flush(oneshot::Sender<()>),
208}
209
210/// The job history. Cloning shares it.
211#[derive(Debug, Clone)]
212pub struct Store {
213    writes: mpsc::UnboundedSender<Write>,
214    reader: Arc<Mutex<Connection>>,
215    events: broadcast::Sender<Arc<EventEnvelope>>,
216    backlog: Arc<Semaphore>,
217}
218
219impl Store {
220    /// Opens the database at `path`, creating it and its directory (mode 0700) if missing, and migrates it to the
221    /// latest version. Refuses a database a newer runnerd wrote.
222    pub fn open(path: &Path) -> Result<Store, StoreError> {
223        if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) {
224            std::fs::DirBuilder::new()
225                .recursive(true)
226                .mode(0o700)
227                .create(dir)?;
228        }
229        // Owner-only, as SQLite then makes its `-wal` and `-shm` files: logs may say more than the owner wants shared.
230        std::fs::OpenOptions::new()
231            .create(true)
232            .append(true)
233            .mode(0o600)
234            .open(path)?;
235        let mut db = Connection::open(path)?;
236        configure(&db)?;
237        migrate(&mut db)?;
238        let reader = Connection::open_with_flags(
239            path,
240            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
241        )?;
242        configure(&reader)?;
243        let (events, _) = broadcast::channel(EVENT_BUFFER);
244        let (writes, receiver) = mpsc::unbounded_channel();
245        let thread_events = events.clone();
246        std::thread::Builder::new()
247            .name("runnerd-store".to_owned())
248            .spawn(move || write::run(db, receiver, &thread_events))?;
249        Ok(Store {
250            writes,
251            reader: Arc::new(Mutex::new(reader)),
252            events,
253            backlog: Arc::new(Semaphore::new(
254                usize::try_from(LOG_BACKLOG).unwrap_or(usize::MAX),
255            )),
256        })
257    }
258
259    fn send(&self, write: Write) {
260        // Fails only once the writer thread is gone, after a SQLite error it logged.
261        let _ = self.writes.send(write);
262    }
263
264    /// Records a queued job the repository's runners may run; announces it the first time.
265    pub fn queued(&self, repository: &str, job: &QueuedJob) {
266        self.send(Write::Queued {
267            repository: repository.to_owned(),
268            job: job.clone(),
269        });
270    }
271
272    /// Records a runner container that started, or one an earlier daemon started and this one adopted.
273    pub fn runner_started(&self, runner: RunnerRecord) {
274        self.send(Write::RunnerStarted(runner));
275    }
276
277    /// Records that a runner's container and registration were removed at `at` (ms).
278    pub fn runner_removed(&self, runner: &str, at: u64) {
279        self.send(Write::RunnerRemoved {
280            runner: runner.to_owned(),
281            at,
282        });
283    }
284
285    /// Records that a runner took a job at `at` (ms).
286    pub fn claimed(&self, job: u64, run_id: u64, runner: &str, at: u64) {
287        self.send(Write::Claimed {
288            job,
289            run_id,
290            runner: runner.to_owned(),
291            at,
292        });
293    }
294
295    /// Records a running job's progress and steps as GitHub reports them; announces changed steps.
296    pub fn details(&self, details: JobDetails) {
297        self.send(Write::Details { details });
298    }
299
300    /// Closes a queued job that left the queue without a runner of this daemon.
301    pub fn elsewhere(&self, details: JobDetails) {
302        self.send(Write::Elsewhere { details });
303    }
304
305    /// Closes a job whose container is gone and that GitHub did not report completed in time.
306    pub fn lost(&self, job: u64, at: u64) {
307        self.send(Write::Lost { job, at });
308    }
309
310    /// Appends log lines to a job, up to `max_bytes` per job, and records where the capture continues. Waits while
311    /// `LOG_BACKLOG` bytes of lines are waiting to be stored; a larger batch goes in parts, each waiting its turn.
312    pub async fn log(&self, job: u64, lines: Vec<String>, position: LogPosition, max_bytes: u64) {
313        let size = |line: &String| line.len() as u64 + 1;
314        let mut lines = lines.into_iter().map(|mut line| {
315            // No line takes the whole backlog; the capture already cuts lines far shorter.
316            if size(&line) > LOG_BACKLOG {
317                let mut end = usize::try_from(LOG_BACKLOG / 2).unwrap_or(usize::MAX);
318                while !line.is_char_boundary(end) {
319                    end -= 1;
320                }
321                line.truncate(end);
322            }
323            line
324        });
325        let mut next = lines.next();
326        loop {
327            let (mut part, mut bytes) = (Vec::new(), 0);
328            while let Some(line) = next.take_if(|line| bytes + size(line) <= LOG_BACKLOG) {
329                bytes += size(&line);
330                part.push(line);
331                next = lines.next();
332            }
333            let last = next.is_none();
334            let share = u32::try_from(bytes.max(1)).unwrap_or(u32::MAX);
335            // The semaphore is never closed.
336            let Ok(backlog) = self.backlog.clone().acquire_many_owned(share).await else {
337                return;
338            };
339            self.send(Write::Log {
340                job,
341                lines: part,
342                position: last.then_some(position),
343                max_bytes,
344                _backlog: backlog,
345            });
346            if last {
347                return;
348            }
349        }
350    }
351
352    /// Records samples of runner containers.
353    pub fn samples(&self, samples: Vec<SampleRecord>) {
354        self.send(Write::Samples(samples));
355    }
356
357    /// Records a sample of the slice.
358    pub fn host_sample(&self, sample: api::HostSample) {
359        self.send(Write::HostSample(sample));
360    }
361
362    /// Adds bytes the mirror served to runners, counted from its log up to `until`, and says whether that was
363    /// stored: if not, the caller counts the same lines again.
364    pub async fn mirror_bytes(&self, bytes: Vec<(String, u64)>, until: MirrorPosition) -> bool {
365        let (done, committed) = oneshot::channel();
366        self.send(Write::MirrorBytes { bytes, until, done });
367        committed.await.unwrap_or(false)
368    }
369
370    /// Records volume sizes in bytes, by name, measured at `at` (ms).
371    pub fn cache_sizes(&self, at: u64, sizes: BTreeMap<String, u64>) {
372        self.send(Write::CacheSizes { at, sizes });
373    }
374
375    /// Stores a durable event, or sends a live one, to every subscriber.
376    pub fn event(&self, event: Event) {
377        self.send(Write::Event(event));
378    }
379
380    /// Records the paused repositories, all of them, in place of the ones recorded before. The write is sent at
381    /// once, in order with every other; the future says whether it committed.
382    pub fn paused(&self, repositories: Vec<String>) -> impl Future<Output = bool> + use<> {
383        let (done, committed) = oneshot::channel();
384        self.send(Write::Paused { repositories, done });
385        async move { committed.await.unwrap_or(false) }
386    }
387
388    /// Deletes what is older than `history` keeps.
389    pub fn prune(&self, now: u64, history: &History) {
390        self.send(Write::Prune {
391            now,
392            history: history.clone(),
393        });
394    }
395
396    /// Waits until every write sent before is done.
397    pub async fn flush(&self) {
398        let (done, wait) = oneshot::channel();
399        self.send(Write::Flush(done));
400        let _ = wait.await;
401    }
402
403    /// Every event from now on, durable or live. Subscribe before reading what was stored, then skip what the read
404    /// returned, and nothing is missed or sent twice.
405    pub fn subscribe(&self) -> broadcast::Receiver<Arc<EventEnvelope>> {
406        self.events.subscribe()
407    }
408
409    async fn read<T: Send + 'static>(
410        &self,
411        query: impl FnOnce(&Connection) -> rusqlite::Result<T> + Send + 'static,
412    ) -> Result<T, StoreError> {
413        let reader = self.reader.clone();
414        // A panic during a read leaves the connection usable: reads change nothing.
415        Ok(tokio::task::spawn_blocking(move || query(&lock(&reader))).await??)
416    }
417
418    /// One page of jobs, newest first.
419    pub async fn jobs(&self, query: api::JobQuery) -> Result<api::JobPage, StoreError> {
420        self.read(move |db| read::jobs(db, &query)).await
421    }
422
423    /// A job with its steps.
424    pub async fn job(&self, id: u64) -> Result<Option<api::Job>, StoreError> {
425        self.read(move |db| read::job(db, id)).await
426    }
427
428    /// A job's samples, oldest first; `None` if there is no such job.
429    pub async fn job_samples(
430        &self,
431        id: u64,
432        interval: Duration,
433    ) -> Result<Option<api::SampleSeries>, StoreError> {
434        self.read(move |db| read::job_samples(db, id, interval))
435            .await
436    }
437
438    /// Up to `limit` log lines of a job from line `from`, with their numbers.
439    pub async fn log_lines(
440        &self,
441        job: u64,
442        from: u64,
443        limit: u64,
444    ) -> Result<Vec<(u64, String)>, StoreError> {
445        self.read(move |db| read::log_lines(db, job, from, limit))
446            .await
447    }
448
449    /// Up to `limit` durable events after `after`, oldest first.
450    pub async fn events_after(
451        &self,
452        after: u64,
453        limit: u64,
454    ) -> Result<Vec<EventEnvelope>, StoreError> {
455        self.read(move |db| read::events_after(db, after, limit))
456            .await
457    }
458
459    /// The repositories as the stored events last announced them, so a restarted daemon announces only what
460    /// changed since, including the repositories its config no longer has. Blocks on the database: for startup.
461    pub fn announced_repositories(&self) -> Result<Vec<api::RepositoryStatus>, StoreError> {
462        Ok(read::announced_repositories(&lock(&self.reader))?)
463    }
464
465    /// The repositories recorded paused, for the engine to start with. Blocks on the database: for startup.
466    pub fn paused_repositories(&self) -> Result<Vec<String>, StoreError> {
467        Ok(read::paused(&lock(&self.reader))?)
468    }
469
470    /// The slice's samples since `since` (seconds), oldest first.
471    pub async fn host_samples(&self, since: u64) -> Result<Vec<api::HostSample>, StoreError> {
472        self.read(move |db| read::host_samples(db, since)).await
473    }
474
475    /// Finished jobs per repository and bucket.
476    pub async fn stats(&self, query: api::StatsQuery, now: u64) -> Result<api::Stats, StoreError> {
477        self.read(move |db| read::stats(db, &query, now)).await
478    }
479
480    /// The last measurement of every volume, with its time (seconds), by name.
481    pub async fn cache_sizes_now(&self) -> Result<Measured, StoreError> {
482        self.read(read::latest_cache_sizes).await
483    }
484
485    /// Every volume measurement since `since` (seconds).
486    pub async fn cache_history(&self, since: u64) -> Result<Vec<api::CacheSize>, StoreError> {
487        self.read(move |db| read::cache_history(db, since)).await
488    }
489
490    /// The jobs not yet closed, for the engine after a restart.
491    pub async fn open_jobs(&self) -> Result<Vec<OpenJob>, StoreError> {
492        self.read(read::open_jobs).await
493    }
494
495    /// The job a runner took, if the history knows it.
496    pub async fn runner_job(&self, runner: &str) -> Result<Option<u64>, StoreError> {
497        let runner = runner.to_owned();
498        self.read(move |db| read::runner_job(db, &runner)).await
499    }
500
501    /// Where a job's log capture continues.
502    pub async fn log_position(&self, job: u64) -> Result<LogPosition, StoreError> {
503        self.read(move |db| read::log_position(db, job)).await
504    }
505
506    /// How far the mirror's log was counted.
507    pub async fn mirror_log_until(&self) -> Result<MirrorPosition, StoreError> {
508        self.read(read::mirror_log_until).await
509    }
510}
511
512/// WAL, so the API reads while the writer writes; a busy timeout instead of an immediate error.
513fn configure(db: &Connection) -> rusqlite::Result<()> {
514    db.pragma_update(None, "journal_mode", "wal")?;
515    db.pragma_update(None, "synchronous", "normal")?;
516    db.pragma_update(None, "foreign_keys", "on")?;
517    db.busy_timeout(Duration::from_secs(5))
518}
519
520/// Runs the migrations the database lacks, each in its own transaction with its new version.
521fn migrate(db: &mut Connection) -> Result<(), StoreError> {
522    let found: u32 = db.pragma_query_value(None, "user_version", |row| row.get(0))?;
523    let supported = u32::try_from(MIGRATIONS.len()).unwrap_or(u32::MAX);
524    if found > supported {
525        return Err(StoreError::Newer { found, supported });
526    }
527    for (version, migration) in (1..).zip(MIGRATIONS).skip(found as usize) {
528        let transaction = db.transaction()?;
529        transaction.execute_batch(migration)?;
530        transaction.pragma_update(None, "user_version", version)?;
531        transaction.commit()?;
532    }
533    Ok(())
534}
535
536/// Logs a failed write; the writer goes on with the next one.
537fn logged<T>(what: &str, result: rusqlite::Result<T>) -> Option<T> {
538    result
539        .map_err(|error| warn!("job history: cannot {what}: {error}"))
540        .ok()
541}
542
543#[cfg(test)]
544mod tests;
runnerd_core/store.rs

runnerd_core/
store.rs
