crates/hub/src/db/runners.rs at 74be97962ed8 · andymccall/oxygenhub

# 74be97962ed8 oxygenhub / crates / hub / src / db / runners.rs

9.9 KiB · Rust · last changed Runners serve the site, an owner or one repository (refs #41, part 2) 1 month ago source blame history raw

//! OxygenFlow runners (hub spec §13.2): who may claim jobs, what they are,
//! whether they are taking work, and what they have run.

use super::{now_ms, Pool};
use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]
pub struct Runner {
    pub id: i64,
    pub name: String,
    pub token_hash: String,
    pub source: String,
    pub key: String,
    pub key_name: String,
    pub arch: String,
    pub os: String,
    pub version: String,
    pub labels: String,
    pub max_jobs: i64,
    pub memory_mib: Option<i64>,
    pub cpus: Option<i64>,
    pub state: String,
    pub created_ms: i64,
    pub created_by: String,
    pub last_seen_ms: Option<i64>,
    /// The owner whose repositories it serves; empty for the whole site.
    pub scope_owner: String,
    /// The one repository it serves, when it serves one.
    pub scope_repo_id: Option<i64>,
}

/// Which repositories' jobs a runner may take (oxygenhub#41).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Scope {
    /// Empty: any owner.
    pub owner: String,
    /// `None`: any of the owner's repositories.
    pub repo_id: Option<i64>,
}

impl Scope {
    pub fn is_site(&self) -> bool {
        self.owner.is_empty() && self.repo_id.is_none()
    }
}

/// A runner that has not polled for this long is shown offline.
pub const ONLINE_MS: i64 = 90_000;

impl Runner {
    pub fn online(&self) -> bool {
        self.last_seen_ms.is_some_and(|t| now_ms() - t < ONLINE_MS)
    }

    /// `online`, `offline`, `draining` or `disabled`.
    pub fn status(&self) -> &'static str {
        match self.state.as_str() {
            "disabled" => "disabled",
            "draining" => "draining",
            _ if self.online() => "online",
            _ => "offline",
        }
    }

    pub fn label_list(&self) -> Vec<&str> {
        self.labels.split(',').filter(|l| !l.is_empty()).collect()
    }

    pub fn scope(&self) -> Scope {
        Scope { owner: self.scope_owner.clone(), repo_id: self.scope_repo_id }
    }
}

pub async fn by_token(pool: &Pool, token_hash: &str) -> sqlx::Result<Option<Runner>> {
    sqlx::query_as("SELECT * FROM flow_runners WHERE token_hash = ?").bind(token_hash).fetch_optional(pool).await
}

pub async fn get(pool: &Pool, id: i64) -> sqlx::Result<Option<Runner>> {
    sqlx::query_as("SELECT * FROM flow_runners WHERE id = ?").bind(id).fetch_optional(pool).await
}

pub async fn list(pool: &Pool) -> sqlx::Result<Vec<Runner>> {
    sqlx::query_as("SELECT * FROM flow_runners ORDER BY name, id").fetch_all(pool).await
}

/// The row of a runner configured with a hub.toml token, made on first sight.
pub async fn configured(pool: &Pool, token_hash: &str, name: &str) -> sqlx::Result<Runner> {
    sqlx::query("INSERT OR IGNORE INTO flow_runners (name, token_hash, source, created_ms) VALUES (?, ?, 'config', ?)")
        .bind(name)
        .bind(token_hash)
        .bind(now_ms())
        .execute(pool)
        .await?;
    Ok(by_token(pool, token_hash).await?.expect("just made"))
}

/// What a claim says about the runner, and that it was seen now. A
/// configured runner takes the name it claims with.
pub async fn seen(pool: &Pool, id: i64, name: Option<&str>, info: Option<&flow_core::api::RunnerInfo>) -> sqlx::Result<()> {
    let now = now_ms();
    match info {
        Some(i) => {
            sqlx::query(
                "UPDATE flow_runners SET arch = ?, os = ?, version = ?, labels = ?, max_jobs = ?, memory_mib = ?, cpus = ?, last_seen_ms = ? WHERE id = ?",
            )
            .bind(&i.arch)
            .bind(&i.os)
            .bind(&i.version)
            .bind(i.labels.join(","))
            .bind(i.max_jobs as i64)
            .bind(i.memory_mib.map(i64::from))
            .bind(i.cpus.map(i64::from))
            .bind(now)
            .bind(id)
            .execute(pool)
            .await?;
        }
        None => {
            sqlx::query("UPDATE flow_runners SET last_seen_ms = ? WHERE id = ?").bind(now).bind(id).execute(pool).await?;
        }
    }
    if let Some(n) = name {
        sqlx::query("UPDATE flow_runners SET name = ? WHERE id = ? AND source = 'config'").bind(n).bind(id).execute(pool).await?;
    }
    Ok(())
}

pub async fn set_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_runners SET state = ? WHERE id = ?").bind(state).bind(id).execute(pool).await?;
    Ok(())
}

/// Forget a runner: its credential stops working. Its jobs keep its name.
pub async fn remove(pool: &Pool, id: i64) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM flow_runners WHERE id = ?").bind(id).execute(pool).await?;
    Ok(())
}

/// The key a runner signs receipts with: the one bound, or `key` (named
/// `name`) when none is yet.
pub async fn bind_key(pool: &Pool, id: i64, key: &str, name: &str) -> sqlx::Result<String> {
    sqlx::query("UPDATE flow_runners SET key = ?, key_name = ? WHERE id = ? AND key = ''")
        .bind(key)
        .bind(name)
        .bind(id)
        .execute(pool)
        .await?;
    sqlx::query_scalar("SELECT key FROM flow_runners WHERE id = ?").bind(id).fetch_one(pool).await
}

/// Keys that runners of this hub sign with.
pub async fn keys(pool: &Pool) -> sqlx::Result<Vec<String>> {
    sqlx::query_scalar("SELECT key FROM flow_runners WHERE key != ''").fetch_all(pool).await
}

/// A one-time registration token, valid for a day, for a runner of `scope`.
pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str, scope: &Scope) -> sqlx::Result<()> {
    let now = now_ms();
    sqlx::query(
        "INSERT INTO flow_runner_registrations (token_hash, created_by, created_ms, expires_ms, scope_owner, scope_repo_id) VALUES (?, ?, ?, ?, ?, ?)",
    )
    .bind(token_hash)
    .bind(by)
    .bind(now)
    .bind(now + 24 * 3_600_000)
    .bind(&scope.owner)
    .bind(scope.repo_id)
    .execute(pool)
    .await?;
    Ok(())
}

/// Change the repositories a runner serves.
pub async fn set_scope(pool: &Pool, id: i64, scope: &Scope) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_runners SET scope_owner = ?, scope_repo_id = ? WHERE id = ?")
        .bind(&scope.owner)
        .bind(scope.repo_id)
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

/// Registration tokens not used yet and not expired: (created by, created, expires).
pub async fn pending_registrations(pool: &Pool) -> sqlx::Result<Vec<(String, i64, i64)>> {
    sqlx::query_as("SELECT created_by, created_ms, expires_ms FROM flow_runner_registrations WHERE used_ms IS NULL AND expires_ms > ? ORDER BY created_ms DESC")
        .bind(now_ms())
        .fetch_all(pool)
        .await
}

/// Use registration token `token_hash` to register a runner: `None` when
/// the token is unknown, used or expired.
#[allow(clippy::too_many_arguments)]
pub async fn register(
    pool: &Pool,
    registration: &str,
    name: &str,
    token_hash: &str,
    key: &str,
    key_name: &str,
    by_default: &str,
) -> sqlx::Result<Option<Runner>> {
    let mut tx = super::begin_write(pool).await?;
    let now = now_ms();
    let found: Option<(String, String, Option<i64>)> = sqlx::query_as(
        "SELECT created_by, scope_owner, scope_repo_id FROM flow_runner_registrations WHERE token_hash = ? AND used_ms IS NULL AND expires_ms > ?",
    )
    .bind(registration)
    .bind(now)
    .fetch_optional(&mut *tx)
    .await?;
    let Some((who, scope_owner, scope_repo_id)) = found else { return Ok(None) };
    let id: i64 = sqlx::query_scalar(
        "INSERT INTO flow_runners (name, token_hash, source, key, key_name, created_ms, created_by, scope_owner, scope_repo_id) VALUES (?, ?, 'registered', ?, ?, ?, ?, ?, ?) RETURNING id",
    )
    .bind(name)
    .bind(token_hash)
    .bind(key)
    .bind(key_name)
    .bind(now)
    .bind(if who.is_empty() { by_default } else { &who })
    .bind(scope_owner)
    .bind(scope_repo_id)
    .fetch_one(&mut *tx)
    .await?;
    sqlx::query("UPDATE flow_runner_registrations SET used_ms = ?, runner_id = ? WHERE token_hash = ?")
        .bind(now)
        .bind(id)
        .bind(registration)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    get(pool, id).await
}

/// A job a runner ran, for its history.
#[derive(Clone, Debug, FromRow)]
pub struct RunnerJob {
    pub owner: String,
    pub repo: String,
    pub run_number: i64,
    pub job: String,
    pub status: String,
    pub started_ms: Option<i64>,
    pub finished_ms: Option<i64>,
    pub receipt_attest: String,
}

/// Jobs runner `id` ran or is running, newest first.
pub async fn history(pool: &Pool, id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<RunnerJob>> {
    sqlx::query_as(
        "SELECT p.owner AS owner, p.name AS repo, r.number AS run_number, j.name AS job, j.status AS status, j.started_ms AS started_ms, \
         j.finished_ms AS finished_ms, j.receipt_attest AS receipt_attest \
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
         WHERE j.runner_id = ? AND (? IS NULL OR j.started_ms < ?) ORDER BY j.started_ms DESC LIMIT ?",
    )
    .bind(id)
    .bind(before)
    .bind(before)
    .bind(limit)
    .fetch_all(pool)
    .await
}

/// (jobs run, passed, busy milliseconds) of runner `id`.
pub async fn totals(pool: &Pool, id: i64) -> sqlx::Result<(i64, i64, i64)> {
    sqlx::query_as(
        "SELECT COUNT(*), COALESCE(SUM(status = 'passed'), 0), COALESCE(SUM(finished_ms - started_ms), 0) \
         FROM flow_jobs WHERE runner_id = ? AND finished_ms IS NOT NULL",
    )
    .bind(id)
    .fetch_one(pool)
    .await
}

/// Jobs running on each runner now, by runner id.
pub async fn running(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {
    sqlx::query_as("SELECT runner_id, COUNT(*) FROM flow_jobs WHERE status = 'running' AND runner_id IS NOT NULL GROUP BY runner_id")
        .fetch_all(pool)
        .await
}

pub async fn set_job_runner(pool: &Pool, job_id: i64, runner_id: i64) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_jobs SET runner_id = ? WHERE id = ?").bind(runner_id).bind(job_id).execute(pool).await?;
    Ok(())
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

b6b7e4cf Andy McCall 1 month ago ↶	1	//! OxygenFlow runners (hub spec §13.2): who may claim jobs, what they are,
	2	//! whether they are taking work, and what they have run.
	3
	4	use super::{now_ms, Pool};
	5	use sqlx::FromRow;
	6
	7	#[derive(Clone, Debug, FromRow)]
	8	pub struct Runner {
	9	pub id: i64,
	10	pub name: String,
	11	pub token_hash: String,
	12	pub source: String,
	13	pub key: String,
	14	pub key_name: String,
	15	pub arch: String,
	16	pub os: String,
	17	pub version: String,
	18	pub labels: String,
	19	pub max_jobs: i64,
	20	pub memory_mib: Option<i64>,
	21	pub cpus: Option<i64>,
	22	pub state: String,
	23	pub created_ms: i64,
	24	pub created_by: String,
	25	pub last_seen_ms: Option<i64>,
74be9796 Andy McCall 1 month ago ↶	26	/// The owner whose repositories it serves; empty for the whole site.
	27	pub scope_owner: String,
	28	/// The one repository it serves, when it serves one.
	29	pub scope_repo_id: Option<i64>,
b6b7e4cf Andy McCall 1 month ago ↶	30	}
b6b7e4cf Andy McCall 1 month ago ↶	31
74be9796 Andy McCall 1 month ago ↶	32	/// Which repositories' jobs a runner may take (oxygenhub#41).
	33	#[derive(Clone, Debug, Default, PartialEq, Eq)]
	34	pub struct Scope {
	35	/// Empty: any owner.
	36	pub owner: String,
	37	/// `None`: any of the owner's repositories.
	38	pub repo_id: Option<i64>,
	39	}
	40
	41	impl Scope {
	42	pub fn is_site(&self) -> bool {
	43	self.owner.is_empty() && self.repo_id.is_none()
	44	}
	45	}
	46
b6b7e4cf Andy McCall 1 month ago ↶	47	/// A runner that has not polled for this long is shown offline.
	48	pub const ONLINE_MS: i64 = 90_000;
	49
	50	impl Runner {
	51	pub fn online(&self) -> bool {
	52	self.last_seen_ms.is_some_and(\|t\| now_ms() - t < ONLINE_MS)
	53	}
	54
	55	/// `online`, `offline`, `draining` or `disabled`.
	56	pub fn status(&self) -> &'static str {
	57	match self.state.as_str() {
	58	"disabled" => "disabled",
	59	"draining" => "draining",
	60	_ if self.online() => "online",
	61	_ => "offline",
	62	}
	63	}
	64
	65	pub fn label_list(&self) -> Vec<&str> {
	66	self.labels.split(',').filter(\|l\| !l.is_empty()).collect()
	67	}
74be9796 Andy McCall 1 month ago ↶	68
	69	pub fn scope(&self) -> Scope {
	70	Scope { owner: self.scope_owner.clone(), repo_id: self.scope_repo_id }
	71	}
b6b7e4cf Andy McCall 1 month ago ↶	72	}
	73
	74	pub async fn by_token(pool: &Pool, token_hash: &str) -> sqlx::Result<Option<Runner>> {
	75	sqlx::query_as("SELECT * FROM flow_runners WHERE token_hash = ?").bind(token_hash).fetch_optional(pool).await
	76	}
	77
	78	pub async fn get(pool: &Pool, id: i64) -> sqlx::Result<Option<Runner>> {
	79	sqlx::query_as("SELECT * FROM flow_runners WHERE id = ?").bind(id).fetch_optional(pool).await
	80	}
	81
	82	pub async fn list(pool: &Pool) -> sqlx::Result<Vec<Runner>> {
	83	sqlx::query_as("SELECT * FROM flow_runners ORDER BY name, id").fetch_all(pool).await
	84	}
	85
	86	/// The row of a runner configured with a hub.toml token, made on first sight.
	87	pub async fn configured(pool: &Pool, token_hash: &str, name: &str) -> sqlx::Result<Runner> {
	88	sqlx::query("INSERT OR IGNORE INTO flow_runners (name, token_hash, source, created_ms) VALUES (?, ?, 'config', ?)")
	89	.bind(name)
	90	.bind(token_hash)
	91	.bind(now_ms())
	92	.execute(pool)
	93	.await?;
	94	Ok(by_token(pool, token_hash).await?.expect("just made"))
	95	}
	96
	97	/// What a claim says about the runner, and that it was seen now. A
	98	/// configured runner takes the name it claims with.
	99	pub async fn seen(pool: &Pool, id: i64, name: Option<&str>, info: Option<&flow_core::api::RunnerInfo>) -> sqlx::Result<()> {
	100	let now = now_ms();
	101	match info {
	102	Some(i) => {
	103	sqlx::query(
	104	"UPDATE flow_runners SET arch = ?, os = ?, version = ?, labels = ?, max_jobs = ?, memory_mib = ?, cpus = ?, last_seen_ms = ? WHERE id = ?",
	105	)
	106	.bind(&i.arch)
	107	.bind(&i.os)
	108	.bind(&i.version)
	109	.bind(i.labels.join(","))
	110	.bind(i.max_jobs as i64)
	111	.bind(i.memory_mib.map(i64::from))
	112	.bind(i.cpus.map(i64::from))
	113	.bind(now)
	114	.bind(id)
	115	.execute(pool)
	116	.await?;
	117	}
	118	None => {
	119	sqlx::query("UPDATE flow_runners SET last_seen_ms = ? WHERE id = ?").bind(now).bind(id).execute(pool).await?;
	120	}
	121	}
	122	if let Some(n) = name {
	123	sqlx::query("UPDATE flow_runners SET name = ? WHERE id = ? AND source = 'config'").bind(n).bind(id).execute(pool).await?;
	124	}
	125	Ok(())
	126	}
	127
	128	pub async fn set_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {
	129	sqlx::query("UPDATE flow_runners SET state = ? WHERE id = ?").bind(state).bind(id).execute(pool).await?;
	130	Ok(())
	131	}
	132
	133	/// Forget a runner: its credential stops working. Its jobs keep its name.
	134	pub async fn remove(pool: &Pool, id: i64) -> sqlx::Result<()> {
	135	sqlx::query("DELETE FROM flow_runners WHERE id = ?").bind(id).execute(pool).await?;
136	Ok(())
137	}
138
139	/// The key a runner signs receipts with: the one bound, or `key` (named
140	/// `name`) when none is yet.
141	pub async fn bind_key(pool: &Pool, id: i64, key: &str, name: &str) -> sqlx::Result<String> {
142	sqlx::query("UPDATE flow_runners SET key = ?, key_name = ? WHERE id = ? AND key = ''")
143	.bind(key)
144	.bind(name)
145	.bind(id)
146	.execute(pool)
147	.await?;
148	sqlx::query_scalar("SELECT key FROM flow_runners WHERE id = ?").bind(id).fetch_one(pool).await
149	}
150
151	/// Keys that runners of this hub sign with.
152	pub async fn keys(pool: &Pool) -> sqlx::Result<Vec<String>> {
153	sqlx::query_scalar("SELECT key FROM flow_runners WHERE key != ''").fetch_all(pool).await
154	}
155
74be9796 Andy McCall 1 month ago ↶	156	/// A one-time registration token, valid for a day, for a runner of `scope`.
74be9796 Andy McCall 1 month ago ↶	157	pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str, scope: &Scope) -> sqlx::Result<()> {
b6b7e4cf Andy McCall 1 month ago ↶	158	let now = now_ms();
74be9796 Andy McCall 1 month ago ↶	159	sqlx::query(
	160	"INSERT INTO flow_runner_registrations (token_hash, created_by, created_ms, expires_ms, scope_owner, scope_repo_id) VALUES (?, ?, ?, ?, ?, ?)",
	161	)
	162	.bind(token_hash)
	163	.bind(by)
	164	.bind(now)
	165	.bind(now + 24 * 3_600_000)
	166	.bind(&scope.owner)
	167	.bind(scope.repo_id)
	168	.execute(pool)
	169	.await?;
	170	Ok(())
	171	}
	172
	173	/// Change the repositories a runner serves.
	174	pub async fn set_scope(pool: &Pool, id: i64, scope: &Scope) -> sqlx::Result<()> {
	175	sqlx::query("UPDATE flow_runners SET scope_owner = ?, scope_repo_id = ? WHERE id = ?")
	176	.bind(&scope.owner)
	177	.bind(scope.repo_id)
	178	.bind(id)
b6b7e4cf Andy McCall 1 month ago ↶	179	.execute(pool)
	180	.await?;
	181	Ok(())
	182	}
	183
	184	/// Registration tokens not used yet and not expired: (created by, created, expires).
	185	pub async fn pending_registrations(pool: &Pool) -> sqlx::Result<Vec<(String, i64, i64)>> {
	186	sqlx::query_as("SELECT created_by, created_ms, expires_ms FROM flow_runner_registrations WHERE used_ms IS NULL AND expires_ms > ? ORDER BY created_ms DESC")
	187	.bind(now_ms())
	188	.fetch_all(pool)
	189	.await
	190	}
	191
	192	/// Use registration token `token_hash` to register a runner: `None` when
	193	/// the token is unknown, used or expired.
	194	#[allow(clippy::too_many_arguments)]
	195	pub async fn register(
	196	pool: &Pool,
	197	registration: &str,
	198	name: &str,
	199	token_hash: &str,
	200	key: &str,
	201	key_name: &str,
	202	by_default: &str,
	203	) -> sqlx::Result<Option<Runner>> {
	204	let mut tx = super::begin_write(pool).await?;
	205	let now = now_ms();
74be9796 Andy McCall 1 month ago ↶	206	let found: Option<(String, String, Option<i64>)> = sqlx::query_as(
	207	"SELECT created_by, scope_owner, scope_repo_id FROM flow_runner_registrations WHERE token_hash = ? AND used_ms IS NULL AND expires_ms > ?",
	208	)
	209	.bind(registration)
	210	.bind(now)
	211	.fetch_optional(&mut *tx)
	212	.await?;
	213	let Some((who, scope_owner, scope_repo_id)) = found else { return Ok(None) };
b6b7e4cf Andy McCall 1 month ago ↶	214	let id: i64 = sqlx::query_scalar(
74be9796 Andy McCall 1 month ago ↶	215	"INSERT INTO flow_runners (name, token_hash, source, key, key_name, created_ms, created_by, scope_owner, scope_repo_id) VALUES (?, ?, 'registered', ?, ?, ?, ?, ?, ?) RETURNING id",
b6b7e4cf Andy McCall 1 month ago ↶	216	)
	217	.bind(name)
	218	.bind(token_hash)
	219	.bind(key)
	220	.bind(key_name)
	221	.bind(now)
	222	.bind(if who.is_empty() { by_default } else { &who })
74be9796 Andy McCall 1 month ago ↶	223	.bind(scope_owner)
74be9796 Andy McCall 1 month ago ↶	224	.bind(scope_repo_id)
b6b7e4cf Andy McCall 1 month ago ↶	225	.fetch_one(&mut *tx)
	226	.await?;
	227	sqlx::query("UPDATE flow_runner_registrations SET used_ms = ?, runner_id = ? WHERE token_hash = ?")
	228	.bind(now)
	229	.bind(id)
	230	.bind(registration)
	231	.execute(&mut *tx)
	232	.await?;
	233	tx.commit().await?;
	234	get(pool, id).await
	235	}
	236
	237	/// A job a runner ran, for its history.
	238	#[derive(Clone, Debug, FromRow)]
	239	pub struct RunnerJob {
	240	pub owner: String,
	241	pub repo: String,
	242	pub run_number: i64,
	243	pub job: String,
	244	pub status: String,
	245	pub started_ms: Option<i64>,
	246	pub finished_ms: Option<i64>,
	247	pub receipt_attest: String,
	248	}
	249
	250	/// Jobs runner `id` ran or is running, newest first.
	251	pub async fn history(pool: &Pool, id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<RunnerJob>> {
	252	sqlx::query_as(
	253	"SELECT p.owner AS owner, p.name AS repo, r.number AS run_number, j.name AS job, j.status AS status, j.started_ms AS started_ms, \
	254	j.finished_ms AS finished_ms, j.receipt_attest AS receipt_attest \
	255	FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
	256	WHERE j.runner_id = ? AND (? IS NULL OR j.started_ms < ?) ORDER BY j.started_ms DESC LIMIT ?",
	257	)
	258	.bind(id)
	259	.bind(before)
	260	.bind(before)
	261	.bind(limit)
	262	.fetch_all(pool)
	263	.await
	264	}
	265
	266	/// (jobs run, passed, busy milliseconds) of runner `id`.
	267	pub async fn totals(pool: &Pool, id: i64) -> sqlx::Result<(i64, i64, i64)> {
	268	sqlx::query_as(
	269	"SELECT COUNT(*), COALESCE(SUM(status = 'passed'), 0), COALESCE(SUM(finished_ms - started_ms), 0) \
	270	FROM flow_jobs WHERE runner_id = ? AND finished_ms IS NOT NULL",
	271	)
	272	.bind(id)
	273	.fetch_one(pool)
	274	.await
	275	}
	276
	277	/// Jobs running on each runner now, by runner id.
	278	pub async fn running(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {
	279	sqlx::query_as("SELECT runner_id, COUNT(*) FROM flow_jobs WHERE status = 'running' AND runner_id IS NOT NULL GROUP BY runner_id")
	280	.fetch_all(pool)
	281	.await
	282	}
	283
	284	pub async fn set_job_runner(pool: &Pool, job_id: i64, runner_id: i64) -> sqlx::Result<()> {
	285	sqlx::query("UPDATE flow_jobs SET runner_id = ? WHERE id = ?").bind(runner_id).bind(job_id).execute(pool).await?;
	286	Ok(())
	287	}

Keyboard