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

runnerd_core/
store.rs
