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

# e9b9619cd857 oxygenhub / crates / hub / src / db / runners.rs

15.2 KiB · Rust · last changed Runners: Connectivity and Status apart, a heartbeat, a filterable queue 21 days 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, Default, 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>,
    /// Labels an administrator gave it, beside those it reports (#98).
    pub admin_labels: String,
    /// A time limit for its jobs in minutes, below the site's, when set.
    pub timeout_minutes: Option<i64>,
    /// It fetches images it lacks from the hub (oxygenflow#39).
    pub fetches_images: bool,
    /// What it measured at its last claim (oxygenflow#40): free disk (MiB),
    /// the 1-minute load ×100, its speed score and class, its kernel.
    pub free_disk_mib: Option<i64>,
    pub load_x100: Option<i64>,
    pub speed: Option<i64>,
    pub speed_class: Option<i64>,
    pub kernel: String,
    /// The tiers it offered at its last claim, comma-separated.
    pub tiers: String,
    /// The images it held then (a JSON array); empty when it did not say.
    pub images: String,
    /// BLAKE3 of what it offers; its declines hold until this changes.
    pub profile: String,
}

/// 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)
    }

    /// Whether the hub has heard from it lately (oxygenhub#243): any call it
    /// makes counts, a busy runner's heartbeat included.
    pub fn connectivity(&self) -> &'static str {
        if self.online() {
            "online"
        } else {
            "offline"
        }
    }

    /// `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",
        }
    }

    /// Every label it carries: what it reports, and what an administrator
    /// gave it.
    pub fn label_list(&self) -> Vec<&str> {
        let mut out: Vec<&str> = self.labels.split(',').filter(|l| !l.is_empty()).collect();
        for l in self.admin_label_list() {
            if !out.contains(&l) {
                out.push(l);
            }
        }
        out
    }

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

    /// The labels it reported itself, as the admin page distinguishes them.
    pub fn reported_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.
/// The administrator's settings for a runner (#98).
pub async fn set_limits(pool: &Pool, id: i64, admin_labels: &str, timeout_minutes: Option<i64>) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_runners SET admin_labels = ?, timeout_minutes = ? WHERE id = ?")
        .bind(admin_labels)
        .bind(timeout_minutes)
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

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(())
}

/// Keep what a runner measured about itself at a claim (oxygenflow#40).
pub async fn set_facts(pool: &Pool, id: i64, f: &flow_core::api::Facts) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_runners SET free_disk_mib = ?, load_x100 = ?, speed = ?, speed_class = ?, kernel = ? WHERE id = ?")
        .bind(f.free_disk_mib.map(|v| v.min(i64::MAX as u64) as i64))
        .bind(f.load_x100.map(i64::from))
        .bind(f.speed.map(i64::from))
        .bind(f.speed_class.map(i64::from))
        .bind(f.kernel.clone().unwrap_or_default())
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

impl Runner {
    /// Its facts as words for the Runners page: "speed 100 (class 3) · load
    /// 1.12 · 41 GiB free"; empty when it has sent none.
    pub fn facts_line(&self) -> String {
        let mut out = Vec::new();
        if let Some(s) = self.speed {
            out.push(format!("speed {s} (class {})", self.speed_class.unwrap_or(0)));
        }
        if let Some(l) = self.load_x100 {
            out.push(format!("load {:.2}", l as f64 / 100.0));
        }
        if let Some(d) = self.free_disk_mib {
            out.push(if d >= 1024 { format!("{} GiB free", d / 1024) } else { format!("{d} MiB free") });
        }
        out.join(" · ")
    }
}

/// What a runner offers at a claim (OxygenFlow spec §11.1): its tiers and
/// images, and the profile hash over those, its architecture, labels and
/// capacity.
pub async fn set_offer(
    pool: &Pool,
    id: i64,
    tiers: &[String],
    images: Option<&[String]>,
    fetches_images: bool,
    profile: &str,
) -> sqlx::Result<()> {
    let images = images.map(|i| serde_json::to_string(i).unwrap_or_default()).unwrap_or_default();
    sqlx::query("UPDATE flow_runners SET tiers = ?, images = ?, fetches_images = ?, profile = ? WHERE id = ?")
        .bind(tiers.join(","))
        .bind(images)
        .bind(fetches_images)
        .bind(profile)
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

impl Runner {
    pub fn tier_list(&self) -> Vec<&str> {
        self.tiers.split(',').filter(|t| !t.is_empty()).collect()
    }

    /// The images it holds, `None` when it does not say.
    pub fn image_list(&self) -> Option<Vec<String>> {
        if self.images.is_empty() {
            None
        } else {
            serde_json::from_str(&self.images).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(())
}

/// The runners that may take a repository's jobs: the site's, its owner's
/// and its own (oxygenhub#41).
pub async fn serving(pool: &Pool, owner: &str, repo_id: i64) -> sqlx::Result<Vec<Runner>> {
    sqlx::query_as(
        "SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) \
         OR (scope_owner = ? AND scope_repo_id IS NULL) OR scope_repo_id = ? ORDER BY name",
    )
    .bind(owner)
    .bind(repo_id)
    .fetch_all(pool)
    .await
}

/// The runners that serve an owner: the site's, and those scoped to the
/// owner or one of its repositories.
pub async fn serving_owner(pool: &Pool, owner: &str) -> sqlx::Result<Vec<Runner>> {
    sqlx::query_as("SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) OR scope_owner = ? ORDER BY name")
        .bind(owner)
        .fetch_all(pool)
        .await
}

/// 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
4d2e44d7 Andy McCall 1 month ago ↶	7	#[derive(Clone, Debug, Default, FromRow)]
b6b7e4cf Andy McCall 1 month ago ↶	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>,
7f5f6c05 Andy McCall 1 month ago ↶	30	/// Labels an administrator gave it, beside those it reports (#98).
	31	pub admin_labels: String,
	32	/// A time limit for its jobs in minutes, below the site's, when set.
	33	pub timeout_minutes: Option<i64>,
92a797b2 Andy McCall 26 days ago ↶	34	/// It fetches images it lacks from the hub (oxygenflow#39).
92a797b2 Andy McCall 26 days ago ↶	35	pub fetches_images: bool,
2fb70d01 Andy McCall 26 days ago ↶	36	/// What it measured at its last claim (oxygenflow#40): free disk (MiB),
	37	/// the 1-minute load ×100, its speed score and class, its kernel.
	38	pub free_disk_mib: Option<i64>,
	39	pub load_x100: Option<i64>,
	40	pub speed: Option<i64>,
	41	pub speed_class: Option<i64>,
	42	pub kernel: String,
4d2e44d7 Andy McCall 1 month ago ↶	43	/// The tiers it offered at its last claim, comma-separated.
	44	pub tiers: String,
	45	/// The images it held then (a JSON array); empty when it did not say.
	46	pub images: String,
	47	/// BLAKE3 of what it offers; its declines hold until this changes.
	48	pub profile: String,
b6b7e4cf Andy McCall 1 month ago ↶	49	}
b6b7e4cf Andy McCall 1 month ago ↶	50
74be9796 Andy McCall 1 month ago ↶	51	/// Which repositories' jobs a runner may take (oxygenhub#41).
	52	#[derive(Clone, Debug, Default, PartialEq, Eq)]
	53	pub struct Scope {
	54	/// Empty: any owner.
	55	pub owner: String,
	56	/// `None`: any of the owner's repositories.
	57	pub repo_id: Option<i64>,
	58	}
	59
	60	impl Scope {
	61	pub fn is_site(&self) -> bool {
	62	self.owner.is_empty() && self.repo_id.is_none()
	63	}
	64	}
	65
b6b7e4cf Andy McCall 1 month ago ↶	66	/// A runner that has not polled for this long is shown offline.
	67	pub const ONLINE_MS: i64 = 90_000;
	68
	69	impl Runner {
	70	pub fn online(&self) -> bool {
	71	self.last_seen_ms.is_some_and(\|t\| now_ms() - t < ONLINE_MS)
	72	}
	73
e9b9619c Andy McCall 21 days ago ↶	74	/// Whether the hub has heard from it lately (oxygenhub#243): any call it
	75	/// makes counts, a busy runner's heartbeat included.
	76	pub fn connectivity(&self) -> &'static str {
	77	if self.online() {
	78	"online"
	79	} else {
	80	"offline"
	81	}
	82	}
	83
b6b7e4cf Andy McCall 1 month ago ↶	84	/// `online`, `offline`, `draining` or `disabled`.
	85	pub fn status(&self) -> &'static str {
	86	match self.state.as_str() {
	87	"disabled" => "disabled",
	88	"draining" => "draining",
	89	_ if self.online() => "online",
	90	_ => "offline",
	91	}
	92	}
	93
7f5f6c05 Andy McCall 1 month ago ↶	94	/// Every label it carries: what it reports, and what an administrator
7f5f6c05 Andy McCall 1 month ago ↶	95	/// gave it.
b6b7e4cf Andy McCall 1 month ago ↶	96	pub fn label_list(&self) -> Vec<&str> {
7f5f6c05 Andy McCall 1 month ago ↶	97	let mut out: Vec<&str> = self.labels.split(',').filter(\|l\| !l.is_empty()).collect();
	98	for l in self.admin_label_list() {
	99	if !out.contains(&l) {
	100	out.push(l);
	101	}
	102	}
	103	out
	104	}
	105
	106	pub fn admin_label_list(&self) -> Vec<&str> {
	107	self.admin_labels.split(',').filter(\|l\| !l.is_empty()).collect()
	108	}
	109
	110	/// The labels it reported itself, as the admin page distinguishes them.
	111	pub fn reported_label_list(&self) -> Vec<&str> {
b6b7e4cf Andy McCall 1 month ago ↶	112	self.labels.split(',').filter(\|l\| !l.is_empty()).collect()
b6b7e4cf Andy McCall 1 month ago ↶	113	}
74be9796 Andy McCall 1 month ago ↶	114
	115	pub fn scope(&self) -> Scope {
	116	Scope { owner: self.scope_owner.clone(), repo_id: self.scope_repo_id }
	117	}
b6b7e4cf Andy McCall 1 month ago ↶	118	}
	119
	120	pub async fn by_token(pool: &Pool, token_hash: &str) -> sqlx::Result<Option<Runner>> {
	121	sqlx::query_as("SELECT * FROM flow_runners WHERE token_hash = ?").bind(token_hash).fetch_optional(pool).await
	122	}
	123
	124	pub async fn get(pool: &Pool, id: i64) -> sqlx::Result<Option<Runner>> {
	125	sqlx::query_as("SELECT * FROM flow_runners WHERE id = ?").bind(id).fetch_optional(pool).await
	126	}
	127
	128	pub async fn list(pool: &Pool) -> sqlx::Result<Vec<Runner>> {
	129	sqlx::query_as("SELECT * FROM flow_runners ORDER BY name, id").fetch_all(pool).await
	130	}
	131
	132	/// The row of a runner configured with a hub.toml token, made on first sight.
	133	pub async fn configured(pool: &Pool, token_hash: &str, name: &str) -> sqlx::Result<Runner> {
	134	sqlx::query("INSERT OR IGNORE INTO flow_runners (name, token_hash, source, created_ms) VALUES (?, ?, 'config', ?)")
	135	.bind(name)
	136	.bind(token_hash)
	137	.bind(now_ms())
	138	.execute(pool)
	139	.await?;
	140	Ok(by_token(pool, token_hash).await?.expect("just made"))
	141	}
	142
	143	/// What a claim says about the runner, and that it was seen now. A
	144	/// configured runner takes the name it claims with.
7f5f6c05 Andy McCall 1 month ago ↶	145	/// The administrator's settings for a runner (#98).
	146	pub async fn set_limits(pool: &Pool, id: i64, admin_labels: &str, timeout_minutes: Option<i64>) -> sqlx::Result<()> {
	147	sqlx::query("UPDATE flow_runners SET admin_labels = ?, timeout_minutes = ? WHERE id = ?")
	148	.bind(admin_labels)
	149	.bind(timeout_minutes)
	150	.bind(id)
	151	.execute(pool)
	152	.await?;
	153	Ok(())
	154	}
	155
b6b7e4cf Andy McCall 1 month ago ↶	156	pub async fn seen(pool: &Pool, id: i64, name: Option<&str>, info: Option<&flow_core::api::RunnerInfo>) -> sqlx::Result<()> {
	157	let now = now_ms();
	158	match info {
	159	Some(i) => {
	160	sqlx::query(
	161	"UPDATE flow_runners SET arch = ?, os = ?, version = ?, labels = ?, max_jobs = ?, memory_mib = ?, cpus = ?, last_seen_ms = ? WHERE id = ?",
	162	)
	163	.bind(&i.arch)
	164	.bind(&i.os)
	165	.bind(&i.version)
	166	.bind(i.labels.join(","))
	167	.bind(i.max_jobs as i64)
	168	.bind(i.memory_mib.map(i64::from))
	169	.bind(i.cpus.map(i64::from))
	170	.bind(now)
	171	.bind(id)
	172	.execute(pool)
	173	.await?;
	174	}
	175	None => {
	176	sqlx::query("UPDATE flow_runners SET last_seen_ms = ? WHERE id = ?").bind(now).bind(id).execute(pool).await?;
	177	}
	178	}
	179	if let Some(n) = name {
	180	sqlx::query("UPDATE flow_runners SET name = ? WHERE id = ? AND source = 'config'").bind(n).bind(id).execute(pool).await?;
	181	}
4d2e44d7 Andy McCall 1 month ago ↶	182	Ok(())
	183	}
	184
2fb70d01 Andy McCall 26 days ago ↶	185	/// Keep what a runner measured about itself at a claim (oxygenflow#40).
	186	pub async fn set_facts(pool: &Pool, id: i64, f: &flow_core::api::Facts) -> sqlx::Result<()> {
	187	sqlx::query("UPDATE flow_runners SET free_disk_mib = ?, load_x100 = ?, speed = ?, speed_class = ?, kernel = ? WHERE id = ?")
	188	.bind(f.free_disk_mib.map(\|v\| v.min(i64::MAX as u64) as i64))
	189	.bind(f.load_x100.map(i64::from))
	190	.bind(f.speed.map(i64::from))
	191	.bind(f.speed_class.map(i64::from))
	192	.bind(f.kernel.clone().unwrap_or_default())
	193	.bind(id)
	194	.execute(pool)
	195	.await?;
	196	Ok(())
	197	}
	198
	199	impl Runner {
	200	/// Its facts as words for the Runners page: "speed 100 (class 3) · load
	201	/// 1.12 · 41 GiB free"; empty when it has sent none.
	202	pub fn facts_line(&self) -> String {
	203	let mut out = Vec::new();
	204	if let Some(s) = self.speed {
	205	out.push(format!("speed {s} (class {})", self.speed_class.unwrap_or(0)));
	206	}
	207	if let Some(l) = self.load_x100 {
	208	out.push(format!("load {:.2}", l as f64 / 100.0));
	209	}
	210	if let Some(d) = self.free_disk_mib {
	211	out.push(if d >= 1024 { format!("{} GiB free", d / 1024) } else { format!("{d} MiB free") });
	212	}
	213	out.join(" · ")
	214	}
	215	}
	216
4d2e44d7 Andy McCall 1 month ago ↶	217	/// What a runner offers at a claim (OxygenFlow spec §11.1): its tiers and
	218	/// images, and the profile hash over those, its architecture, labels and
	219	/// capacity.
92a797b2 Andy McCall 26 days ago ↶	220	pub async fn set_offer(
	221	pool: &Pool,
	222	id: i64,
	223	tiers: &[String],
	224	images: Option<&[String]>,
	225	fetches_images: bool,
	226	profile: &str,
	227	) -> sqlx::Result<()> {
4d2e44d7 Andy McCall 1 month ago ↶	228	let images = images.map(\|i\| serde_json::to_string(i).unwrap_or_default()).unwrap_or_default();
92a797b2 Andy McCall 26 days ago ↶	229	sqlx::query("UPDATE flow_runners SET tiers = ?, images = ?, fetches_images = ?, profile = ? WHERE id = ?")
4d2e44d7 Andy McCall 1 month ago ↶	230	.bind(tiers.join(","))
4d2e44d7 Andy McCall 1 month ago ↶	231	.bind(images)
92a797b2 Andy McCall 26 days ago ↶	232	.bind(fetches_images)
4d2e44d7 Andy McCall 1 month ago ↶	233	.bind(profile)
	234	.bind(id)
	235	.execute(pool)
	236	.await?;
b6b7e4cf Andy McCall 1 month ago ↶	237	Ok(())
4d2e44d7 Andy McCall 1 month ago ↶	238	}
	239
	240	impl Runner {
	241	pub fn tier_list(&self) -> Vec<&str> {
	242	self.tiers.split(',').filter(\|t\| !t.is_empty()).collect()
	243	}
	244
	245	/// The images it holds, `None` when it does not say.
	246	pub fn image_list(&self) -> Option<Vec<String>> {
	247	if self.images.is_empty() {
	248	None
	249	} else {
	250	serde_json::from_str(&self.images).ok()
	251	}
	252	}
b6b7e4cf Andy McCall 1 month ago ↶	253	}
	254
	255	pub async fn set_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {
	256	sqlx::query("UPDATE flow_runners SET state = ? WHERE id = ?").bind(state).bind(id).execute(pool).await?;
	257	Ok(())
	258	}
	259
	260	/// Forget a runner: its credential stops working. Its jobs keep its name.
	261	pub async fn remove(pool: &Pool, id: i64) -> sqlx::Result<()> {
	262	sqlx::query("DELETE FROM flow_runners WHERE id = ?").bind(id).execute(pool).await?;
	263	Ok(())
	264	}
	265
	266	/// The key a runner signs receipts with: the one bound, or `key` (named
	267	/// `name`) when none is yet.
	268	pub async fn bind_key(pool: &Pool, id: i64, key: &str, name: &str) -> sqlx::Result<String> {
	269	sqlx::query("UPDATE flow_runners SET key = ?, key_name = ? WHERE id = ? AND key = ''")
	270	.bind(key)
	271	.bind(name)
	272	.bind(id)
	273	.execute(pool)
	274	.await?;
	275	sqlx::query_scalar("SELECT key FROM flow_runners WHERE id = ?").bind(id).fetch_one(pool).await
	276	}
	277
	278	/// Keys that runners of this hub sign with.
	279	pub async fn keys(pool: &Pool) -> sqlx::Result<Vec<String>> {
	280	sqlx::query_scalar("SELECT key FROM flow_runners WHERE key != ''").fetch_all(pool).await
	281	}
	282
74be9796 Andy McCall 1 month ago ↶	283	/// A one-time registration token, valid for a day, for a runner of `scope`.
74be9796 Andy McCall 1 month ago ↶	284	pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str, scope: &Scope) -> sqlx::Result<()> {
b6b7e4cf Andy McCall 1 month ago ↶	285	let now = now_ms();
74be9796 Andy McCall 1 month ago ↶	286	sqlx::query(
	287	"INSERT INTO flow_runner_registrations (token_hash, created_by, created_ms, expires_ms, scope_owner, scope_repo_id) VALUES (?, ?, ?, ?, ?, ?)",
	288	)
	289	.bind(token_hash)
	290	.bind(by)
	291	.bind(now)
	292	.bind(now + 24 * 3_600_000)
	293	.bind(&scope.owner)
	294	.bind(scope.repo_id)
	295	.execute(pool)
	296	.await?;
	297	Ok(())
	298	}
	299
7974adb0 Andy McCall 1 month ago ↶	300	/// The runners that may take a repository's jobs: the site's, its owner's
	301	/// and its own (oxygenhub#41).
	302	pub async fn serving(pool: &Pool, owner: &str, repo_id: i64) -> sqlx::Result<Vec<Runner>> {
	303	sqlx::query_as(
	304	"SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) \
	305	OR (scope_owner = ? AND scope_repo_id IS NULL) OR scope_repo_id = ? ORDER BY name",
	306	)
	307	.bind(owner)
	308	.bind(repo_id)
	309	.fetch_all(pool)
	310	.await
	311	}
	312
	313	/// The runners that serve an owner: the site's, and those scoped to the
	314	/// owner or one of its repositories.
	315	pub async fn serving_owner(pool: &Pool, owner: &str) -> sqlx::Result<Vec<Runner>> {
	316	sqlx::query_as("SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) OR scope_owner = ? ORDER BY name")
	317	.bind(owner)
	318	.fetch_all(pool)
	319	.await
	320	}
	321
74be9796 Andy McCall 1 month ago ↶	322	/// Change the repositories a runner serves.
	323	pub async fn set_scope(pool: &Pool, id: i64, scope: &Scope) -> sqlx::Result<()> {
	324	sqlx::query("UPDATE flow_runners SET scope_owner = ?, scope_repo_id = ? WHERE id = ?")
	325	.bind(&scope.owner)
	326	.bind(scope.repo_id)
	327	.bind(id)
b6b7e4cf Andy McCall 1 month ago ↶	328	.execute(pool)
	329	.await?;
	330	Ok(())
	331	}
	332
	333	/// Registration tokens not used yet and not expired: (created by, created, expires).
	334	pub async fn pending_registrations(pool: &Pool) -> sqlx::Result<Vec<(String, i64, i64)>> {
	335	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")
	336	.bind(now_ms())
	337	.fetch_all(pool)
	338	.await
	339	}
	340
	341	/// Use registration token `token_hash` to register a runner: `None` when
	342	/// the token is unknown, used or expired.
	343	#[allow(clippy::too_many_arguments)]
	344	pub async fn register(
	345	pool: &Pool,
	346	registration: &str,
	347	name: &str,
	348	token_hash: &str,
	349	key: &str,
	350	key_name: &str,
	351	by_default: &str,
	352	) -> sqlx::Result<Option<Runner>> {
	353	let mut tx = super::begin_write(pool).await?;
	354	let now = now_ms();
74be9796 Andy McCall 1 month ago ↶	355	let found: Option<(String, String, Option<i64>)> = sqlx::query_as(
	356	"SELECT created_by, scope_owner, scope_repo_id FROM flow_runner_registrations WHERE token_hash = ? AND used_ms IS NULL AND expires_ms > ?",
	357	)
	358	.bind(registration)
	359	.bind(now)
	360	.fetch_optional(&mut *tx)
	361	.await?;
	362	let Some((who, scope_owner, scope_repo_id)) = found else { return Ok(None) };
b6b7e4cf Andy McCall 1 month ago ↶	363	let id: i64 = sqlx::query_scalar(
74be9796 Andy McCall 1 month ago ↶	364	"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 ↶	365	)
	366	.bind(name)
	367	.bind(token_hash)
	368	.bind(key)
	369	.bind(key_name)
	370	.bind(now)
	371	.bind(if who.is_empty() { by_default } else { &who })
74be9796 Andy McCall 1 month ago ↶	372	.bind(scope_owner)
74be9796 Andy McCall 1 month ago ↶	373	.bind(scope_repo_id)
b6b7e4cf Andy McCall 1 month ago ↶	374	.fetch_one(&mut *tx)
	375	.await?;
	376	sqlx::query("UPDATE flow_runner_registrations SET used_ms = ?, runner_id = ? WHERE token_hash = ?")
	377	.bind(now)
	378	.bind(id)
	379	.bind(registration)
	380	.execute(&mut *tx)
	381	.await?;
	382	tx.commit().await?;
	383	get(pool, id).await
	384	}
	385
	386	/// A job a runner ran, for its history.
	387	#[derive(Clone, Debug, FromRow)]
	388	pub struct RunnerJob {
	389	pub owner: String,
	390	pub repo: String,
	391	pub run_number: i64,
	392	pub job: String,
	393	pub status: String,
	394	pub started_ms: Option<i64>,
	395	pub finished_ms: Option<i64>,
	396	pub receipt_attest: String,
	397	}
	398
	399	/// Jobs runner `id` ran or is running, newest first.
	400	pub async fn history(pool: &Pool, id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<RunnerJob>> {
	401	sqlx::query_as(
	402	"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, \
	403	j.finished_ms AS finished_ms, j.receipt_attest AS receipt_attest \
	404	FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
	405	WHERE j.runner_id = ? AND (? IS NULL OR j.started_ms < ?) ORDER BY j.started_ms DESC LIMIT ?",
	406	)
	407	.bind(id)
	408	.bind(before)
	409	.bind(before)
	410	.bind(limit)
	411	.fetch_all(pool)
	412	.await
	413	}
	414
	415	/// (jobs run, passed, busy milliseconds) of runner `id`.
	416	pub async fn totals(pool: &Pool, id: i64) -> sqlx::Result<(i64, i64, i64)> {
	417	sqlx::query_as(
	418	"SELECT COUNT(*), COALESCE(SUM(status = 'passed'), 0), COALESCE(SUM(finished_ms - started_ms), 0) \
	419	FROM flow_jobs WHERE runner_id = ? AND finished_ms IS NOT NULL",
	420	)
	421	.bind(id)
	422	.fetch_one(pool)
	423	.await
	424	}
	425
	426	/// Jobs running on each runner now, by runner id.
	427	pub async fn running(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {
	428	sqlx::query_as("SELECT runner_id, COUNT(*) FROM flow_jobs WHERE status = 'running' AND runner_id IS NOT NULL GROUP BY runner_id")
	429	.fetch_all(pool)
	430	.await
	431	}
	432
	433	pub async fn set_job_runner(pool: &Pool, job_id: i64, runner_id: i64) -> sqlx::Result<()> {
	434	sqlx::query("UPDATE flow_jobs SET runner_id = ? WHERE id = ?").bind(runner_id).bind(job_id).execute(pool).await?;
	435	Ok(())
	436	}

Keyboard