runnerd-core 0.0.18

//! The writer thread: applies writes in the order they were sent, closes job records once both GitHub and the
//! engine are done with them, and announces what changed. A change and the durable events it announces commit in
//! one transaction and are broadcast only after, so a subscriber that reads the stored events after subscribing
//! misses none, and a crash never keeps a change without its event.

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};

use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
use tokio::sync::{broadcast, mpsc, oneshot};

use super::read::{self, secs};
use super::{
    HOST_SAMPLE_BATCH, LogPosition, MirrorPosition, Resolution, RunnerRecord, SampleRecord, Write,
    logged,
};
use crate::api::{self, Event, EventEnvelope};
use crate::github::{JobDetails, JobStep, QueuedJob};
use crate::sync::lock;
use crate::time;

/// Makes an event its repository's latest in `announced`.
const ANNOUNCE_REPOSITORY: &str =
    "INSERT OR REPLACE INTO announced (repository, event) VALUES (?1, ?2)";
/// Makes an event its profile's latest in `announced_profiles`.
const ANNOUNCE_PROFILE: &str =
    "INSERT OR REPLACE INTO announced_profiles (profile, event) VALUES (?1, ?2)";

pub(super) fn run(
    db: Connection,
    mut queue: mpsc::UnboundedReceiver<Write>,
    events: &broadcast::Sender<Arc<EventEnvelope>>,
    held: &Mutex<Vec<api::HostSample>>,
) {
    let writer = Writer { db, events, held };
    while let Some(write) = queue.blocking_recv() {
        writer.apply(write);
    }
    writer.store_host_samples();
}

struct Writer<'store> {
    db: Connection,
    events: &'store broadcast::Sender<Arc<EventEnvelope>>,
    /// `Store::held`.
    held: &'store Mutex<Vec<api::HostSample>>,
}

/// The events a change announces once it commits.
type Announced = Vec<Event>;

fn timestamp(text: Option<&str>) -> Option<u64> {
    text.and_then(time::parse_ms)
}

/// How `records.removed_by` stores a removal: its name in the API.
fn removal_name(by: api::RecordRemoval) -> String {
    serde_json::to_value(by)
        .ok()
        .and_then(|value| value.as_str().map(str::to_owned))
        .unwrap_or_default()
}

impl Writer<'_> {
    #[allow(
        clippy::too_many_lines,
        reason = "one arm per kind of write, each a call; split, it would only move the match"
    )]
    fn apply(&self, write: Write) {
        match write {
            Write::Queued { repository, job } => {
                self.change("record a queued job", |announced| {
                    self.queued(&repository, &job, announced)
                });
            }
            Write::RunnerStarted(runner) => self.runner_started(&runner),
            Write::RunnerAdmission { runner, admission } => self.admission(&runner, admission),
            Write::RunnerOomKilled(runner) => self.runner_oom_killed(&runner),
            Write::HostWaited { jobs, ms } => self.host_waited(&jobs, ms),
            Write::BuildCache { runner, counts } => self.build_cache(&runner, &counts),
            Write::RunnerRemoved { runner, at } => {
                self.change("record a removed runner", |announced| {
                    self.runner_removed(&runner, at, announced)
                });
            }
            Write::RunnersGone { running, at } => {
                self.change("record removed runners", |announced| {
                    self.runners_gone(&running, at, announced)
                });
            }
            Write::Claimed {
                job,
                run_id,
                runner,
                at,
            } => {
                self.change("record a claimed job", |announced| {
                    self.claimed(job, run_id, &runner, at, announced)
                });
            }
            Write::Details { details } => {
                self.change("record a job's progress", |announced| {
                    self.details(&details, announced)
                });
            }
            Write::Elsewhere { details } => {
                self.change("close a job that ran elsewhere", |announced| {
                    self.elsewhere(&details, announced)
                });
            }
            Write::Lost { job, at } => {
                self.change("close a lost job", |announced| {
                    self.close(job, "lost", Some(at), announced)
                });
            }
            Write::Log {
                job,
                lines,
                position,
                max_bytes,
                _backlog,
            } => {
                self.change("store log lines", |announced| {
                    self.log(job, lines, position, max_bytes, announced)
                });
            }
            Write::Samples(samples) => {
                self.change("record samples", |announced| {
                    self.samples(&samples, announced)
                });
            }
            Write::HostSample(sample) => {
                // Live: no ID to wait for.
                let _ = self.events.send(Arc::new(EventEnvelope {
                    id: None,
                    at: secs(time::now_ms()),
                    event: Event::HostSample { sample },
                }));
                if lock(self.held).len() >= HOST_SAMPLE_BATCH {
                    self.store_host_samples();
                }
            }
            Write::MirrorBytes { bytes, until, done } => {
                let _ = done.send(self.mirror_bytes(&bytes, until));
            }
            Write::CacheSizes { at, sizes } => self.cache_sizes(at, sizes),
            Write::Event(event) => {
                self.change("store an event", |announced| {
                    announced.push(*event);
                    Ok(())
                });
            }
            Write::RecordMade(record) => self.record_made(record),
            Write::RecordSkipped { job, why } => self.record_skipped(job, &why),
            Write::RecordRemoved { job, at, by } => self.record_removed(job, at, by),
            Write::Paused { repositories, done } => {
                let _ = done.send(self.paused(&repositories));
            }
            Write::SizingIntent { change, done } => {
                let _ = done.send(self.sizing_intent(&change));
            }
            Write::SizingResolved {
                id,
                resolution,
                done,
            } => self.sizing_resolved(id, resolution, done),
            Write::Notify { change, done } => self.notify(change, done),
            Write::Flush(done) => {
                self.store_host_samples();
                let _ = done.send(());
            }
        }
    }

    fn record_made(&self, record: api::Record) {
        self.change("record a job's record", |announced| {
            self.db.execute(
                "INSERT OR REPLACE INTO records (job, created_at, bytes) VALUES (?1, ?2, ?3)",
                params![record.job, record.created_at * 1000, record.bytes],
            )?;
            announced.push(Event::RecordCreated { record });
            Ok(())
        });
    }

    fn record_skipped(&self, job: u64, why: &str) {
        logged(
            "record a skipped record",
            self.db.execute(
                "INSERT OR REPLACE INTO records (job, skipped) VALUES (?1, ?2)",
                params![job, why],
            ),
        );
    }

    fn record_removed(&self, job: u64, at: u64, by: api::RecordRemoval) {
        self.change("record a removed record", |announced| {
            self.db.execute(
                "INSERT INTO records (job, removed_at, removed_by) VALUES (?1, ?2, ?3) \
                 ON CONFLICT (job) DO UPDATE SET removed_at = ?2, removed_by = ?3",
                params![job, at, removal_name(by)],
            )?;
            announced.push(Event::RecordDeleted {
                job,
                removed_by: by,
            });
            Ok(())
        });
    }

    /// Stores the held host samples in one transaction, then lets go of them. If that fails they are held for the
    /// next try, ten batches at most: older ones are dropped.
    fn store_host_samples(&self) {
        let samples = lock(self.held).clone();
        if samples.is_empty() {
            return;
        }
        let stored = self.change("record host samples", |_| {
            for sample in &samples {
                self.db.execute(
                    "INSERT OR REPLACE INTO host_samples (at, cpu, memory) VALUES (?1, ?2, ?3)",
                    params![sample.at * 1000, sample.cpu, sample.memory],
                )?;
            }
            Ok(())
        });
        let mut held = lock(self.held);
        let done = if stored {
            samples.len()
        } else {
            held.len().saturating_sub(HOST_SAMPLE_BATCH * 10)
        };
        let len = held.len();
        held.drain(..done.min(len));
    }

    fn runner_started(&self, runner: &RunnerRecord) {
        logged(
            "record a runner",
            self.db.execute(
                "INSERT OR IGNORE INTO runners (name, repository, profile, slot, container, started_at, \
                 memory_cap, cpu_cap, admission, memory_measure) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 'without_page_cache')",
                params![
                    runner.name,
                    runner.repository,
                    runner.profile,
                    runner.slot,
                    runner.container,
                    runner.started_at,
                    runner.memory_cap,
                    runner.cpu_cap,
                    runner
                        .admission
                        .and_then(|admission| serde_json::to_string(&admission).ok())
                ],
            ),
        );
    }

    fn cache_sizes(&self, at: u64, sizes: BTreeMap<String, u64>) {
        for (name, bytes) in sizes {
            logged(
                "record a cache size",
                self.db.execute(
                    "INSERT OR REPLACE INTO cache_sizes (at, name, bytes) VALUES (?1, ?2, ?3)",
                    params![at, name, bytes],
                ),
            );
        }
    }

    fn admission(&self, runner: &str, admission: api::Admission) {
        logged(
            "record what a runner reserves",
            self.db.execute(
                "UPDATE runners SET admission = ?2 WHERE name = ?1",
                params![runner, serde_json::to_string(&admission).ok()],
            ),
        );
    }

    fn runner_oom_killed(&self, runner: &str) {
        logged(
            "record a kill for memory",
            self.db.execute(
                "UPDATE runners SET oom_kills = max(coalesce(oom_kills, 0), 1) WHERE name = ?1",
                [runner],
            ),
        );
    }

    fn host_waited(&self, jobs: &[u64], ms: u64) {
        for job in jobs {
            logged(
                "record a wait for the host's memory",
                self.db.execute(
                    "UPDATE jobs SET host_waited = host_waited + ?2 WHERE id = ?1",
                    params![job, ms],
                ),
            );
        }
    }

    fn build_cache(&self, runner: &str, counts: &api::BuildCacheCounts) {
        logged(
            "record the build cache's counts",
            self.db.execute(
                "UPDATE runners SET build_cache_hits = build_cache_hits + ?2, \
                 build_cache_misses = build_cache_misses + ?3, \
                 build_cache_read_bytes = build_cache_read_bytes + ?4, \
                 build_cache_written_bytes = build_cache_written_bytes + ?5, \
                 build_cache_refused = build_cache_refused + ?6, \
                 build_cache_writer = max(coalesce(build_cache_writer, 0), ?7) WHERE name = ?1",
                params![
                    runner,
                    counts.hits,
                    counts.misses,
                    counts.read_bytes,
                    counts.written_bytes,
                    counts.refused,
                    counts.writer
                ],
            ),
        );
    }

    /// Makes a change and stores the durable events it announces in one transaction, then sends them, with their
    /// IDs, to every subscriber. On any error nothing is kept or sent. Returns whether it committed.
    fn change(
        &self,
        what: &str,
        change: impl FnOnce(&mut Announced) -> rusqlite::Result<()>,
    ) -> bool {
        let result = (|| {
            // Immediate, so a busy database is waited for: a deferred transaction that read first fails at once.
            let transaction = Transaction::new_unchecked(&self.db, TransactionBehavior::Immediate)?;
            let mut announced = Vec::new();
            change(&mut announced)?;
            let envelopes = announced
                .into_iter()
                .map(|event| self.store_event(event))
                .collect::<rusqlite::Result<Vec<_>>>()?;
            transaction.commit()?;
            Ok(envelopes)
        })();
        let Some(envelopes) = logged(what, result) else {
            return false;
        };
        for envelope in envelopes {
            // No subscriber is not an error.
            let _ = self.events.send(Arc::new(envelope));
        }
        true
    }

    /// Stores an event if it is durable, giving it its ID. A repository's or profile's event becomes
    /// its latest in `announced` or `announced_profiles`.
    fn store_event(&self, event: Event) -> rusqlite::Result<EventEnvelope> {
        let at = time::now_ms();
        let id = if event.is_durable() {
            let body = serde_json::to_string(&event)
                .map_err(|error| rusqlite::Error::ToSqlConversionFailure(Box::new(error)))?;
            self.db.execute(
                "INSERT INTO events (at, body) VALUES (?1, ?2)",
                params![at, body],
            )?;
            let id = self.db.last_insert_rowid();
            let latest = match &event {
                Event::RepositoryChanged { repository } => {
                    Some((ANNOUNCE_REPOSITORY, &repository.name))
                }
                Event::RepositoryRemoved { repository } => Some((ANNOUNCE_REPOSITORY, repository)),
                Event::ProfileChanged { profile } => Some((ANNOUNCE_PROFILE, &profile.name)),
                Event::ProfileRemoved { profile } => Some((ANNOUNCE_PROFILE, profile)),
                _ => None,
            };
            if let Some((upsert, name)) = latest {
                self.db.execute(upsert, params![name, id])?;
            }
            u64::try_from(id).ok()
        } else {
            None
        };
        Ok(EventEnvelope {
            id,
            at: secs(at),
            event,
        })
    }

    fn queued(
        &self,
        repository: &str,
        job: &QueuedJob,
        announced: &mut Announced,
    ) -> rusqlite::Result<()> {
        let labels = serde_json::to_string(&job.labels).unwrap_or_else(|_| "[]".to_owned());
        let queued_at = if job.created_at == 0 {
            time::now_ms()
        } else {
            job.created_at
        };
        let inserted = self.db.execute(
            "INSERT INTO jobs (id, repository, run_id, run_attempt, workflow, name, branch, url, labels, state, \
             queued_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 'queued', ?10) ON CONFLICT (id) DO NOTHING",
            params![
                job.id,
                repository,
                job.run_id,
                job.run_attempt,
                job.workflow,
                job.name,
                job.branch,
                job.url,
                labels,
                queued_at
            ],
        )?;
        if inserted == 1
            && let Some(job) = read::summary(&self.db, job.id)?
        {
            announced.push(Event::JobQueued { job });
        }
        Ok(())
    }

    fn claimed(
        &self,
        job: u64,
        run_id: u64,
        runner: &str,
        at: u64,
        announced: &mut Announced,
    ) -> rusqlite::Result<()> {
        // A job the daemon never saw queued: GitHub's listing lagged, or it was queued before the daemon started.
        let inserted = self.db.execute(
            "INSERT INTO jobs (id, repository, run_id, run_attempt, name, labels, state, runner, queued_at, \
             started_at) SELECT ?1, repository, ?2, 1, '', '[]', 'running', name, ?3, ?3 FROM runners \
             WHERE name = ?4 ON CONFLICT (id) DO NOTHING",
            params![job, run_id, at, runner],
        )?;
        let updated = self.db.execute(
            "UPDATE jobs SET state = 'running', runner = ?2, started_at = coalesce(started_at, ?3) \
             WHERE id = ?1 AND state = 'queued'",
            params![job, runner, at],
        )?;
        if (inserted == 1 || updated == 1)
            && let Some(job) = read::summary(&self.db, job)?
        {
            announced.push(Event::JobStarted { job });
        }
        Ok(())
    }

    fn details(&self, details: &JobDetails, announced: &mut Announced) -> rusqlite::Result<()> {
        let completed = details.status == "completed";
        let labels = serde_json::to_string(&details.labels).unwrap_or_else(|_| "[]".to_owned());
        let updated = self.db.execute(
            "UPDATE jobs SET run_id = ?2, run_attempt = ?3, name = ?4, workflow = ?5, branch = ?6, url = ?7, \
             labels = ?8, queued_at = coalesce(?9, queued_at), started_at = coalesce(?10, started_at), \
             finished_at = CASE WHEN ?11 THEN coalesce(?12, finished_at) ELSE finished_at END, \
             conclusion = ?13, github_completed = ?11 WHERE id = ?1 AND state = 'running'",
            params![
                details.id,
                details.run_id,
                details.run_attempt,
                details.name,
                details.workflow_name,
                details.head_branch,
                details.html_url,
                labels,
                timestamp(details.created_at.as_deref()),
                timestamp(details.started_at.as_deref()),
                completed,
                timestamp(details.completed_at.as_deref()),
                details.conclusion,
            ],
        )?;
        if updated != 1 {
            return Ok(());
        }
        for step in &details.steps {
            if let Some(step) = self.step(details.id, step)? {
                announced.push(Event::JobStep {
                    job: details.id,
                    step,
                });
            }
        }
        self.close_if_done(details.id, announced)
    }

    /// Stores a step; returns it if it is new or changed.
    fn step(&self, job: u64, step: &JobStep) -> rusqlite::Result<Option<api::Step>> {
        let stored = api::Step {
            number: step.number,
            name: step.name.clone(),
            status: step.status.clone(),
            conclusion: step.conclusion.clone(),
            started_at: timestamp(step.started_at.as_deref()).map(secs),
            completed_at: timestamp(step.completed_at.as_deref()).map(secs),
        };
        let changed = self.db.execute(
            "INSERT INTO steps (job, number, name, status, conclusion, started_at, completed_at) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) ON CONFLICT (job, number) DO UPDATE SET \
             name = excluded.name, status = excluded.status, conclusion = excluded.conclusion, \
             started_at = excluded.started_at, completed_at = excluded.completed_at \
             WHERE (name, status, conclusion, started_at, completed_at) IS NOT \
             (excluded.name, excluded.status, excluded.conclusion, excluded.started_at, excluded.completed_at)",
            params![
                job,
                step.number,
                step.name,
                step.status,
                step.conclusion,
                timestamp(step.started_at.as_deref()),
                timestamp(step.completed_at.as_deref()),
            ],
        )?;
        Ok((changed == 1).then_some(stored))
    }

    fn runner_removed(
        &self,
        runner: &str,
        at: u64,
        announced: &mut Announced,
    ) -> rusqlite::Result<()> {
        self.db.execute(
            "UPDATE runners SET removed_at = ?2 WHERE name = ?1 AND removed_at IS NULL",
            params![runner, at],
        )?;
        let job: Option<u64> = self
            .db
            .query_row(
                "SELECT id FROM jobs WHERE runner = ?1 AND state = 'running'",
                [runner],
                |row| row.get(0),
            )
            .optional()?;
        match job {
            Some(job) => self.close_if_done(job, announced),
            None => Ok(()),
        }
    }

    fn runners_gone(
        &self,
        running: &[String],
        at: u64,
        announced: &mut Announced,
    ) -> rusqlite::Result<()> {
        let open: Vec<String> = self
            .db
            .prepare_cached("SELECT name FROM runners WHERE removed_at IS NULL")?
            .query_map([], |row| row.get(0))?
            .collect::<rusqlite::Result<_>>()?;
        for runner in open.iter().filter(|runner| !running.contains(runner)) {
            self.runner_removed(runner, at, announced)?;
        }
        Ok(())
    }

    /// Closes a running job once GitHub reports it completed and its runner is removed.
    fn close_if_done(&self, job: u64, announced: &mut Announced) -> rusqlite::Result<()> {
        let done = self
            .db
            .query_row(
                "SELECT j.github_completed AND r.removed_at IS NOT NULL FROM jobs j \
                 JOIN runners r ON r.name = j.runner WHERE j.id = ?1 AND j.state = 'running'",
                [job],
                |row| row.get::<_, bool>(0),
            )
            .optional()?;
        if done == Some(true) {
            self.close(job, "completed", None, announced)?;
        }
        Ok(())
    }

    /// Keeps a job's metrics and makes its record final.
    fn close(
        &self,
        job: u64,
        state: &str,
        finished_at: Option<u64>,
        announced: &mut Announced,
    ) -> rusqlite::Result<()> {
        let metrics = read::metrics(&self.db, job)?;
        let closed = self.db.execute(
            "UPDATE jobs SET state = ?2, finished_at = coalesce(finished_at, ?3), cpu_time = ?4, \
             cpu_peak = ?5, cpu_average = ?6, memory_peak = ?7, memory_average = ?8, \
             network_received = ?9, network_sent = ?10 WHERE id = ?1 AND state = 'running'",
            params![
                job,
                state,
                finished_at,
                metrics.cpu_time,
                metrics.cpu_peak,
                metrics.cpu_average,
                metrics.memory_peak,
                metrics.memory_average,
                metrics.network_received,
                metrics.network_sent,
            ],
        )?;
        if closed == 1
            && let Some(job) = read::summary(&self.db, job)?
        {
            announced.push(Event::JobFinished { job });
        }
        Ok(())
    }

    fn elsewhere(&self, details: &JobDetails, announced: &mut Announced) -> rusqlite::Result<()> {
        // GitHub gives a job no runner took, as one a newer push cancelled while queued, a start time: its creation.
        // Its runner's name is null, or empty for a job GitHub skipped.
        let started_at = details
            .runner_name
            .as_ref()
            .filter(|name| !name.is_empty())
            .and_then(|_| timestamp(details.started_at.as_deref()));
        let closed = self.db.execute(
            "UPDATE jobs SET state = 'elsewhere', conclusion = ?2, started_at = ?3, finished_at = ?4 \
             WHERE id = ?1 AND state = 'queued'",
            params![
                details.id,
                details.conclusion,
                started_at,
                timestamp(details.completed_at.as_deref()),
            ],
        )?;
        if closed == 1
            && let Some(job) = read::summary(&self.db, details.id)?
        {
            announced.push(Event::JobFinished { job });
        }
        Ok(())
    }

    fn log(
        &self,
        job: u64,
        lines: Vec<String>,
        position: Option<LogPosition>,
        max_bytes: u64,
        announced: &mut Announced,
    ) -> rusqlite::Result<()> {
        let (mut count, mut bytes, mut truncated): (u64, u64, bool) = self.db.query_row(
            "SELECT log_lines, log_bytes, log_truncated FROM jobs WHERE id = ?1",
            [job],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        )?;
        for line in lines {
            if truncated {
                break;
            }
            let size = line.len() as u64 + 1;
            if bytes + size > max_bytes {
                truncated = true;
                break;
            }
            self.db.execute(
                "INSERT OR IGNORE INTO log_lines (job, line, text) VALUES (?1, ?2, ?3)",
                params![job, count, line],
            )?;
            announced.push(Event::LogLine {
                job,
                line: count,
                text: line,
            });
            count += 1;
            bytes += size;
        }
        self.db.execute(
            "UPDATE jobs SET log_lines = ?2, log_bytes = ?3, log_truncated = ?4, \
             log_page = coalesce(?5, log_page), log_offset = coalesce(?6, log_offset) WHERE id = ?1",
            params![
                job,
                count,
                bytes,
                truncated,
                position.map(|position| position.page),
                position.map(|position| position.offset)
            ],
        )?;
        Ok(())
    }

    fn samples(&self, samples: &[SampleRecord], announced: &mut Announced) -> rusqlite::Result<()> {
        for sample in samples {
            self.db.execute(
                "INSERT OR REPLACE INTO samples (runner, at, cpu, cpu_usage, memory, network_received, \
                 network_sent) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
                params![
                    sample.runner,
                    sample.at,
                    sample.cpu,
                    sample.cpu_usage,
                    sample.memory,
                    sample.network_received,
                    sample.network_sent
                ],
            )?;
            if let Some(ooms) = sample.events.oom {
                self.db.execute(
                    "UPDATE runners SET ooms = max(coalesce(ooms, 0), ?2) WHERE name = ?1",
                    params![sample.runner, ooms],
                )?;
            }
            if let Some(ooms) = sample.events.local_oom {
                self.db.execute(
                    "UPDATE runners SET local_ooms = max(coalesce(local_ooms, 0), ?2) WHERE name = ?1",
                    params![sample.runner, ooms],
                )?;
            }
            if let Some(throttled) = sample.events.sock_throttled {
                self.db.execute(
                    "UPDATE runners SET sock_throttled = max(coalesce(sock_throttled, 0), ?2) WHERE name = ?1",
                    params![sample.runner, throttled],
                )?;
            }
            if let Some(kills) = sample.events.oom_kill.filter(|kills| *kills > 0) {
                self.db.execute(
                    "UPDATE runners SET oom_kills = max(coalesce(oom_kills, 0), ?2) WHERE name = ?1",
                    params![sample.runner, kills],
                )?;
            }
            let job: Option<u64> = self
                .db
                .query_row(
                    "SELECT id FROM jobs WHERE runner = ?1 AND state = 'running'",
                    [&sample.runner],
                    |row| row.get(0),
                )
                .optional()?;
            if let Some(job) = job {
                announced.push(Event::JobSample {
                    job,
                    sample: api::Sample {
                        at: secs(sample.at),
                        cpu: sample.cpu,
                        memory: sample.memory,
                        network_received: sample.network_received,
                        network_sent: sample.network_sent,
                    },
                });
            }
        }
        Ok(())
    }

    /// Replaces the paused repositories.
    fn paused(&self, repositories: &[String]) -> bool {
        self.change("record the paused repositories", |_| {
            self.db.execute("DELETE FROM paused", [])?;
            for repository in repositories {
                self.db
                    .execute("INSERT INTO paused (repository) VALUES (?1)", [repository])?;
            }
            Ok(())
        })
    }

    /// Records what the notifier did, and sends whether it committed.
    fn notify(&self, change: super::NotifyChange, done: oneshot::Sender<bool>) {
        let committed = match change {
            super::NotifyChange::Record {
                target,
                set,
                remove,
            } => self.notify_record(&target, &set, &remove),
            super::NotifyChange::Keep(targets) => self.notify_keep(&targets),
        };
        let _ = done.send(committed);
    }

    fn notify_record(&self, target: &str, set: &[(String, u64)], remove: &[String]) -> bool {
        self.change("record what was notified", |_| {
            for (key, value) in set {
                self.db.execute(
                    "INSERT OR REPLACE INTO notify (target, key, value) VALUES (?1, ?2, ?3)",
                    params![target, key, value],
                )?;
            }
            for key in remove {
                self.db.execute(
                    "DELETE FROM notify WHERE target = ?1 AND key = ?2",
                    params![target, key],
                )?;
            }
            Ok(())
        })
    }

    fn notify_keep(&self, targets: &[String]) -> bool {
        self.change("forget removed notification targets", |_| {
            let kept: Vec<String> = self
                .db
                .prepare_cached("SELECT DISTINCT target FROM notify")?
                .query_map([], |row| row.get(0))?
                .collect::<rusqlite::Result<_>>()?;
            for gone in kept.iter().filter(|target| !targets.contains(target)) {
                self.db
                    .execute("DELETE FROM notify WHERE target = ?1", [gone])?;
            }
            Ok(())
        })
    }

    /// Runs `write` with its commits synced to disk (`synchronous = FULL`), for what must outlast a power failure:
    /// an auto sizing change recorded before the config file is written and synced, and what became of it. `None`
    /// when the database will not sync them.
    fn durably<T>(&self, write: impl FnOnce() -> T) -> Option<T> {
        logged(
            "sync commits to disk",
            self.db.pragma_update(None, "synchronous", "full"),
        )?;
        let written = write();
        logged(
            "stop syncing every commit",
            self.db.pragma_update(None, "synchronous", "normal"),
        );
        Some(written)
    }

    fn sizing_intent(&self, change: &api::SizingChange) -> Option<i64> {
        self.durably(|| self.record_intent(change)).flatten()
    }

    fn record_intent(&self, change: &api::SizingChange) -> Option<i64> {
        let mut id = None;
        self.change("record an auto sizing change", |_| {
            self.db.execute(
                "INSERT INTO auto_sizing (profile, at, from_memory, from_slots, to_memory, to_slots, reason, state, \
                 from_cpus, to_cpus) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'pending', ?8, ?9)",
                params![
                    change.profile,
                    change.at.saturating_mul(1000),
                    change.from.memory,
                    change.from.slots,
                    change.to.memory,
                    change.to.slots,
                    change.reason,
                    change.from.cpus,
                    change.to.cpus
                ],
            )?;
            id = Some(self.db.last_insert_rowid());
            Ok(())
        })
        .then_some(id)
        .flatten()
    }

    fn sizing_resolved(&self, id: i64, resolution: Resolution, done: oneshot::Sender<bool>) {
        let resolved = self
            .durably(|| self.resolve(id, resolution))
            .unwrap_or(false);
        let _ = done.send(resolved);
    }

    fn resolve(&self, id: i64, resolution: Resolution) -> bool {
        self.change("resolve an auto sizing change", |announced| {
            let state = match resolution {
                Resolution::Done => "done",
                Resolution::Failed => "failed",
                Resolution::Uncertain => "uncertain",
            };
            self.db.execute(
                "UPDATE auto_sizing SET state = ?2 WHERE id = ?1 AND state = 'pending'",
                params![id, state],
            )?;
            if resolution == Resolution::Done {
                let change = self.db.query_row(
                    "SELECT * FROM auto_sizing WHERE id = ?1",
                    [id],
                    read::sizing_change,
                )?;
                announced.push(Event::SizingChanged { change });
            }
            Ok(())
        })
    }

    /// Counts a batch of the mirror's log and where it ends, together: counting part of it, or counting it without
    /// moving past it, would count it again after a restart.
    fn mirror_bytes(&self, bytes: &[(String, u64)], until: MirrorPosition) -> bool {
        self.change("count mirror bytes", |_| {
            for (runner, bytes) in bytes {
                self.db.execute(
                    "UPDATE runners SET mirror_bytes = mirror_bytes + ?2 WHERE name = ?1",
                    params![runner, bytes],
                )?;
            }
            self.db.execute(
                "INSERT OR REPLACE INTO meta (key, value) VALUES ('mirror_log_until', ?1), ('mirror_log_seen', ?2)",
                [until.at, until.seen],
            )?;
            Ok(())
        })
    }
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    reason = "tests: a database that cannot open fails the test"
)]
mod tests {
    use std::sync::Mutex;

    use rusqlite::Connection;
    use tokio::sync::broadcast;

    use super::Writer;
    use crate::testdir::TestDir;

    fn synchronous(db: &Connection) -> u8 {
        db.pragma_query_value(None, "synchronous", |row| row.get(0))
            .unwrap()
    }

    #[test]
    fn a_durable_write_syncs_its_commits_and_the_writer_goes_back_to_normal_after() {
        let dir = TestDir::new("store-write");
        let db = Connection::open(dir.0.join("runnerd.db")).unwrap();
        super::super::configure(&db).unwrap();
        let (events, _) = broadcast::channel(1);
        let held = Mutex::new(Vec::new());
        let writer = Writer {
            db,
            events: &events,
            held: &held,
        };
        // 2 is FULL, 1 NORMAL.
        assert_eq!(writer.durably(|| synchronous(&writer.db)), Some(2));
        assert_eq!(synchronous(&writer.db), 1);
    }
}