OXYGENHUB
# b66e70b7b6c7 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
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>,
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 
}
31
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
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 
    }
68
69 
    pub fn scope(&self) -> Scope {
70 
        Scope { owner: self.scope_owner.clone(), repo_id: self.scope_repo_id }
71 
    }
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
156 
/// A one-time registration token, valid for a day, for a runner of `scope`.
157 
pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str, scope: &Scope) -> sqlx::Result<()> {
158 
    let now = now_ms();
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)
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();
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) };
214 
    let id: i64 = sqlx::query_scalar(
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",
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 })
223 
    .bind(scope_owner)
224 
    .bind(scope_repo_id)
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 
}