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

runnerd_core/
store.rs
