runners.rs 96.4%

318 of 330 lines run, 74 of 132 functions · run #497 on main · run missed

//! 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,
    /// What it last said about its update (oxygenhub#322), and when.
    pub update_note: String,
    pub update_ms: 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)
    }

    /// 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 = $1").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 = $1").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 INTO flow_runners (name, token_hash, source, created_ms) VALUES ($1, $2, 'config', $3) ON CONFLICT DO NOTHING")
        .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 = $1, timeout_minutes = $2 WHERE id = $3")
        .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 = $1, os = $2, version = $3, labels = $4, max_jobs = $5, memory_mib = $6, cpus = $7, last_seen_ms = $8 WHERE id = $9",
            )
            .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 = $1 WHERE id = $2").bind(now).bind(id).execute(pool).await?;
        }
    }
    if let Some(n) = name {
        sqlx::query("UPDATE flow_runners SET name = $1 WHERE id = $2 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 = $1, load_x100 = $2, speed = $3, speed_class = $4, kernel = $5 WHERE id = $6")
        .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 = $1, images = $2, fetches_images = $3, profile = $4 WHERE id = $5")
        .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 = $1, state_ms = $2 WHERE id = $3")
        .bind(state)
        .bind(super::now_ms())
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

/// Whether runner `id`'s last `n` jobs since its state last changed all
/// ended in an infrastructure error (oxygenflow#30), with the last one's.
pub async fn erroring(pool: &Pool, id: i64, n: u32) -> sqlx::Result<Option<String>> {
    if n == 0 {
        return Ok(None);
    }
    let last: Vec<(String, String)> = sqlx::query_as(
        "SELECT j.status, j.error FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id
         WHERE j.runner_id = $1 AND j.finished_ms IS NOT NULL AND j.finished_ms >= r.state_ms
         ORDER BY j.finished_ms DESC, j.id DESC LIMIT $2",
    )
    .bind(id)
    .bind(n as i64)
    .fetch_all(pool)
    .await?;
    Ok((last.len() == n as usize && last.iter().all(|(s, _)| s == "error")).then(|| last[0].1.clone()))
}

/// 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 runner_group_members WHERE runner_id = $1").bind(id).execute(pool).await?;
    sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'runner' AND key = $1").bind(id.to_string()).execute(pool).await?;
    sqlx::query("DELETE FROM flow_runners WHERE id = $1").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 = $1, key_name = $2 WHERE id = $3 AND key = ''")
        .bind(key)
        .bind(name)
        .bind(id)
        .execute(pool)
        .await?;
    sqlx::query_scalar("SELECT key FROM flow_runners WHERE id = $1").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 ($1, $2, $3, $4, $5, $6)",
    )
    .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 = $1 AND scope_repo_id IS NULL) OR scope_repo_id = $2 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 = $1 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 = $1, scope_repo_id = $2 WHERE id = $3")
        .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 > $1 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 = $1 AND used_ms IS NULL AND expires_ms > $2",
    )
    .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 ($1, $2, 'registered', $3, $4, $5, $6, $7, $8) 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 = $1, runner_id = $2 WHERE token_hash = $3")
        .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 = $1 AND ($2 IS NULL OR j.started_ms < $3) ORDER BY j.started_ms DESC LIMIT $4",
    )
    .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(CASE WHEN status = 'passed' THEN 1 ELSE 0 END), 0), CAST(COALESCE(SUM(finished_ms - started_ms), 0) AS BIGINT) \
         FROM flow_jobs WHERE runner_id = $1 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 = $1 WHERE id = $2").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

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

Coverage / crates/hub/src/db/runners.rs 96.4%

Keyboard