| 1 | //! Self-hosted runners: registering them, handing them work, hearing back, |
| 2 | //! and the people's side (listing, removing, groups, settings). See |
| 3 | //! `g1t_contracts::runners` for the model and the wire. |
| 4 | //! |
| 5 | //! Work reaches a runner only when it asks for it (`runner_poll`), so a |
| 6 | //! runner never needs an open port. A workflow job is taken by flipping its |
| 7 | //! row from `queued` to `in_progress` in one statement, exactly as g1t's |
| 8 | //! own sandboxes take one, and the harness on the machine then fetches it |
| 9 | //! and reports with the job's own token through the same API. Agent work |
| 10 | //! from the runner service is kept as a task with its environment sealed, |
| 11 | //! and handed over once. |
| 12 | |
| 13 | use g1t_contracts::access::Capability; |
| 14 | use g1t_contracts::audit::{AuditActor, AuditOutcome, AuditTarget, NewAuditEntry, RecordAuditArgs, Surface}; |
| 15 | use g1t_contracts::billing::{ComputeKind, RecordSandboxArgs}; |
| 16 | use g1t_contracts::repos::{Repo, RepoPath}; |
| 17 | use g1t_contracts::runners::{ |
| 18 | self as model, ACTIVITY_LIMIT, CloudJob, HandedOverTask, RunnerActivity, RunnerActivityArgs, StuckJob, StuckJobsArgs, Assignment, CancelTaskArgs, CreateRegistrationTokenArgs, DeleteRunnerGroupArgs, EnqueueTaskArgs, FinishedArgs, |
| 19 | ListRunnersArgs, Poll, PollArgs, RegisterArgs, Registered, RegistrationToken, RemoveRunnerArgs, RemoveSelfArgs, RouteArgs, Runner, |
| 20 | RunnerAuth, RunnerGroup, RunnerGroupsArgs, RunnerSettings, RunnerSettingsArgs, RunnerWork, RunnersOwner, SetRunnerGroupArgs, |
| 21 | SetRunnerSettingsArgs, Wanted, |
| 22 | }; |
| 23 | use g1t_contracts::time::rfc3339; |
| 24 | use g1t_contracts::{FailureCode, Outcome, PrincipalKind, Role, User, new_id}; |
| 25 | use g1t_kit::now_ms; |
| 26 | use g1t_secrets::{random_hex, sha256_hex}; |
| 27 | use serde::Deserialize; |
| 28 | use serde_json::{Map, Value, json}; |
| 29 | use worker::Result; |
| 30 | use worker::wasm_bindgen::JsValue; |
| 31 | |
| 32 | use crate::plan::JobRow; |
| 33 | use crate::{Actions, Count, SILENT_MS, check, fail, optional}; |
| 34 | |
| 35 | /// What one poll has learned while it looks for work, so its looks again |
| 36 | /// do not ask again: each repository's visibility, and the jobs it already |
| 37 | /// said wait for a sandboxed runner. |
| 38 | #[derive(Default)] |
| 39 | struct Looked { |
| 40 | visibilities: std::collections::HashMap<String, Option<String>>, |
| 41 | noted: std::collections::HashSet<String>, |
| 42 | } |
| 43 | |
| 44 | /// How often a poll looks again for work while it waits. |
| 45 | const POLL_EVERY_MS: u64 = 2_500; |
| 46 | /// A runner that has been offline this long is removed, as on GitHub. |
| 47 | const FORGET_OFFLINE_MS: u64 = 14 * 24 * 60 * 60 * 1000; |
| 48 | /// How often a poll writes down that a runner, or the work it holds, is |
| 49 | /// still there. Polls come every twenty seconds or so; a write each time |
| 50 | /// would be most of the service's writes for nothing. Well inside |
| 51 | /// [`model::ONLINE_WITHIN_MS`] and [`crate::SILENT_MS`]. |
| 52 | const SEEN_RESOLUTION_MS: u64 = 60 * 1000; |
| 53 | |
| 54 | /// Whether `seen` is old enough, at `at`, to be written down again. |
| 55 | fn seen_is_stale(seen: Option<&str>, at: u64) -> bool { |
| 56 | seen.is_none_or(|seen| seen < rfc3339(at.saturating_sub(SEEN_RESOLUTION_MS)).as_str()) |
| 57 | } |
| 58 | |
| 59 | #[derive(Clone, Debug, Deserialize)] |
| 60 | pub struct RunnerRow { |
| 61 | pub id: String, |
| 62 | pub workspace: String, |
| 63 | pub repo_id: Option<String>, |
| 64 | pub repo: Option<String>, |
| 65 | pub group_id: Option<String>, |
| 66 | pub name: String, |
| 67 | pub labels: String, |
| 68 | pub os: String, |
| 69 | pub arch: String, |
| 70 | pub version: String, |
| 71 | pub ephemeral: u32, |
| 72 | pub credential_hash: String, |
| 73 | pub previous_hash: Option<String>, |
| 74 | pub rotated_at: String, |
| 75 | pub work_id: Option<String>, |
| 76 | pub work_kind: Option<String>, |
| 77 | pub spent: u32, |
| 78 | pub last_seen_at: Option<String>, |
| 79 | pub created_at: String, |
| 80 | pub created_by: Option<String>, |
| 81 | /// `nested-container`, `container` or `host` (migration 0015); none |
| 82 | /// for a runner that never said. |
| 83 | #[serde(default)] |
| 84 | pub isolation: Option<String>, |
| 85 | } |
| 86 | |
| 87 | impl RunnerRow { |
| 88 | fn labels(&self) -> Vec<String> { |
| 89 | serde_json::from_str(&self.labels).unwrap_or_default() |
| 90 | } |
| 91 | |
| 92 | fn online(&self, now: u64) -> bool { |
| 93 | self.last_seen_at.as_deref().is_some_and(|seen| seen >= rfc3339(now.saturating_sub(model::ONLINE_WITHIN_MS)).as_str()) |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | #[derive(Clone, Debug, Deserialize)] |
| 98 | struct GroupRow { |
| 99 | id: String, |
| 100 | name: String, |
| 101 | is_default: u32, |
| 102 | repositories: String, |
| 103 | updated_at: String, |
| 104 | } |
| 105 | |
| 106 | impl GroupRow { |
| 107 | fn repositories(&self) -> Vec<String> { |
| 108 | serde_json::from_str(&self.repositories).unwrap_or_default() |
| 109 | } |
| 110 | |
| 111 | /// Whether `repo` (`owner/name`, or a bare name) may use the group. |
| 112 | fn allows(&self, repo: &str) -> bool { |
| 113 | let name = repo.rsplit('/').next().unwrap_or(repo); |
| 114 | let listed = self.repositories(); |
| 115 | listed.is_empty() || listed.iter().any(|r| r.eq_ignore_ascii_case(name)) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | #[derive(Clone, Debug, Deserialize)] |
| 120 | struct RegistrationRow { |
| 121 | workspace: String, |
| 122 | repo_id: Option<String>, |
| 123 | repo: Option<String>, |
| 124 | group_id: Option<String>, |
| 125 | expires_at: String, |
| 126 | } |
| 127 | |
| 128 | #[derive(Clone, Debug, Deserialize)] |
| 129 | struct SettingsRow { |
| 130 | agents: u32, |
| 131 | agent_labels: String, |
| 132 | fork_pulls: u32, |
| 133 | } |
| 134 | |
| 135 | #[derive(Clone, Debug, Deserialize)] |
| 136 | pub struct TaskRow { |
| 137 | pub id: String, |
| 138 | pub sandbox: String, |
| 139 | pub repo_id: Option<String>, |
| 140 | pub repo: String, |
| 141 | pub kind: String, |
| 142 | pub title: String, |
| 143 | pub labels: String, |
| 144 | pub env: Option<String>, |
| 145 | pub timeout_minutes: u32, |
| 146 | pub status: String, |
| 147 | pub runner_id: Option<String>, |
| 148 | pub runner_name: Option<String>, |
| 149 | pub started_at: Option<String>, |
| 150 | #[serde(default)] |
| 151 | pub seen_at: Option<String>, |
| 152 | /// The agent run it belongs to (migration 0012); none for older tasks. |
| 153 | #[serde(default)] |
| 154 | pub run_id: Option<String>, |
| 155 | } |
| 156 | |
| 157 | /// Who may do what with a set of runners: the workspace, and the |
| 158 | /// repository when they are a repository's own. |
| 159 | struct Place { |
| 160 | workspace: String, |
| 161 | repo: Option<Repo>, |
| 162 | } |
| 163 | |
| 164 | fn now() -> String { |
| 165 | rfc3339(now_ms()) |
| 166 | } |
| 167 | |
| 168 | fn valid_name(name: &str) -> std::result::Result<String, String> { |
| 169 | let name = name.trim(); |
| 170 | let ok = !name.is_empty() |
| 171 | && name.len() <= 64 |
| 172 | && name.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')); |
| 173 | if ok { |
| 174 | Ok(name.to_owned()) |
| 175 | } else { |
| 176 | Err("A runner's name is 1 to 64 letters, digits, `-`, `_` and `.`.".to_owned()) |
| 177 | } |
| 178 | } |
| 179 | |
| 180 | fn valid_group_name(name: &str) -> std::result::Result<String, String> { |
| 181 | let name = name.trim(); |
| 182 | if name.is_empty() || name.len() > 64 || name.chars().any(|c| c.is_control()) { |
| 183 | return Err("A group's name is 1 to 64 characters.".to_owned()); |
| 184 | } |
| 185 | Ok(name.to_owned()) |
| 186 | } |
| 187 | |
| 188 | /// What a runner's credential is checked against, and what a registration |
| 189 | /// token is found by. |
| 190 | fn hash(secret: &str) -> String { |
| 191 | sha256_hex(secret) |
| 192 | } |
| 193 | |
| 194 | /// The actor of a runner's own entries in the audit log. |
| 195 | fn runner_actor(runner: &RunnerRow) -> AuditActor { |
| 196 | AuditActor { |
| 197 | actor_kind: Some(g1t_contracts::audit::ActorKind::Runner), |
| 198 | actor: runner.name.clone(), |
| 199 | actor_id: runner.id.clone(), |
| 200 | ..AuditActor::default() |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | fn entry(actor: AuditActor, action: &str, workspace: &str, repo: Option<String>, path: Option<String>, message: String) -> NewAuditEntry { |
| 205 | NewAuditEntry { |
| 206 | actor, |
| 207 | action: action.to_owned(), |
| 208 | surface: Surface::Rest, |
| 209 | target: AuditTarget { |
| 210 | workspace: workspace.to_owned(), |
| 211 | repo, |
| 212 | path, |
| 213 | ..AuditTarget::default() |
| 214 | }, |
| 215 | outcome: AuditOutcome::Allowed, |
| 216 | rule: "runner".to_owned(), |
| 217 | result: Some("ok".to_owned()), |
| 218 | message: Some(message), |
| 219 | request_id: new_id("req", now_ms()), |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | impl Actions { |
| 224 | // --- Shared ------------------------------------------------------------ |
| 225 | |
| 226 | pub(crate) async fn audit(&self, entries: Vec<NewAuditEntry>) { |
| 227 | let recorded: Result<u32> = g1t_kit::call(&self.events, "audit_record", &RecordAuditArgs { entries }).await; |
| 228 | if let Err(error) = recorded { |
| 229 | worker::console_error!("actions: audit entries not recorded: {error}"); |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | /// Where `owner` points, if `actor` may see it (`manage` false) or |
| 234 | /// change it (`manage` true): a workspace's runners are seen by its |
| 235 | /// members and changed by its owners; a repository's own, by its |
| 236 | /// admins. Agents and workspace tokens never change them, so a |
| 237 | /// workflow's `G1T_TOKEN` cannot add a machine to run its own jobs. |
| 238 | async fn runner_place(&self, actor: &User, owner: &RunnersOwner, manage: bool) -> Result<Outcome<Place>> { |
| 239 | if actor.kind == PrincipalKind::Agent { |
| 240 | return Ok(fail(FailureCode::Forbidden, "Agents cannot see or change self-hosted runners.")); |
| 241 | } |
| 242 | if manage && actor.kind == PrincipalKind::Workspace { |
| 243 | return Ok(fail( |
| 244 | FailureCode::Forbidden, |
| 245 | "A workspace's tokens, G1T_TOKEN included, cannot change self-hosted runners. Use a person's token with the runners:admin scope.", |
| 246 | )); |
| 247 | } |
| 248 | match (&owner.repo, &owner.workspace) { |
| 249 | (Some(path), _) => { |
| 250 | let repo = check!(self.may(actor, path, Capability::ManageIntegrations).await?); |
| 251 | Ok(Outcome::Ok(Place { workspace: repo.namespace.to_lowercase(), repo: Some(repo) })) |
| 252 | } |
| 253 | (None, Some(slug)) => { |
| 254 | let slug = slug.to_lowercase(); |
| 255 | let role = actor.workspaces.iter().find(|m| m.slug.eq_ignore_ascii_case(&slug)).map(|m| m.role); |
| 256 | match role { |
| 257 | None => Ok(fail(FailureCode::NotFound, "There is no such workspace, or you are not a member of it.")), |
| 258 | // A workspace's machines and their tokens are its owners' alone, to see as well as change. |
| 259 | Some(Role::Member) => Ok(fail( |
| 260 | FailureCode::Forbidden, |
| 261 | format!("Only owners of {slug} can {} its self-hosted runners.", if manage { "change" } else { "see" }), |
| 262 | )), |
| 263 | Some(_) => Ok(Outcome::Ok(Place { workspace: slug, repo: None })), |
| 264 | } |
| 265 | } |
| 266 | (None, None) => Ok(fail(FailureCode::Invalid, "Give `repo` or `workspace`.")), |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | async fn group_rows(&self, workspace: &str) -> Result<Vec<GroupRow>> { |
| 271 | self.db |
| 272 | .prepare("SELECT * FROM runner_groups WHERE workspace = ? ORDER BY is_default DESC, name") |
| 273 | .bind(&[workspace.into()])? |
| 274 | .all() |
| 275 | .await? |
| 276 | .results::<GroupRow>() |
| 277 | } |
| 278 | |
| 279 | /// The workspace's default group, made the first time it is needed. |
| 280 | async fn default_group(&self, workspace: &str) -> Result<GroupRow> { |
| 281 | let at = now(); |
| 282 | self.db |
| 283 | .prepare( |
| 284 | "INSERT OR IGNORE INTO runner_groups (id, workspace, name, is_default, repositories, created_at, updated_at) |
| 285 | VALUES (?, ?, 'Default', 1, '[]', ?, ?)", |
| 286 | ) |
| 287 | .bind(&[new_id("rng", now_ms()).into(), workspace.into(), at.as_str().into(), at.as_str().into()])? |
| 288 | .run() |
| 289 | .await?; |
| 290 | let row = self |
| 291 | .db |
| 292 | .prepare("SELECT * FROM runner_groups WHERE workspace = ? AND is_default = 1 LIMIT 1") |
| 293 | .bind(&[workspace.into()])? |
| 294 | .first::<GroupRow>(None) |
| 295 | .await?; |
| 296 | row.ok_or_else(|| worker::Error::RustError("the default runner group could not be made".into())) |
| 297 | } |
| 298 | |
| 299 | /// A group of the workspace by id or name. |
| 300 | async fn find_group(&self, workspace: &str, wanted: &str) -> Result<Option<GroupRow>> { |
| 301 | self.default_group(workspace).await?; |
| 302 | Ok(self |
| 303 | .group_rows(workspace) |
| 304 | .await? |
| 305 | .into_iter() |
| 306 | .find(|g| g.id == wanted || g.name.eq_ignore_ascii_case(wanted.trim()))) |
| 307 | } |
| 308 | |
| 309 | async fn runner_rows(&self, workspace: &str) -> Result<Vec<RunnerRow>> { |
| 310 | self.db |
| 311 | .prepare("SELECT * FROM runners WHERE workspace = ? ORDER BY name") |
| 312 | .bind(&[workspace.into()])? |
| 313 | .all() |
| 314 | .await? |
| 315 | .results::<RunnerRow>() |
| 316 | } |
| 317 | |
| 318 | async fn runner_by_id(&self, id: &str) -> Result<Option<RunnerRow>> { |
| 319 | self.db.prepare("SELECT * FROM runners WHERE id = ?").bind(&[id.into()])?.first::<RunnerRow>(None).await |
| 320 | } |
| 321 | |
| 322 | /// What a runner is doing, for people. |
| 323 | async fn work_of(&self, row: &RunnerRow) -> Result<Option<RunnerWork>> { |
| 324 | let Some(id) = &row.work_id else { return Ok(None) }; |
| 325 | if row.work_kind.as_deref() == Some("agent") { |
| 326 | let task = self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[id.as_str().into()])?.first::<TaskRow>(None).await?; |
| 327 | return Ok(task.filter(|t| t.status == "in_progress").map(|t| RunnerWork { |
| 328 | kind: "agent".into(), |
| 329 | id: t.id, |
| 330 | name: t.title, |
| 331 | repo: Some(t.repo), |
| 332 | run_id: t.run_id, |
| 333 | started_at: t.started_at, |
| 334 | })); |
| 335 | } |
| 336 | #[derive(Deserialize)] |
| 337 | struct Found { |
| 338 | id: String, |
| 339 | name: String, |
| 340 | run_id: String, |
| 341 | status: String, |
| 342 | started_at: Option<String>, |
| 343 | repo: String, |
| 344 | } |
| 345 | let job = self |
| 346 | .db |
| 347 | .prepare("SELECT jobs.id, jobs.name, jobs.run_id, jobs.status, jobs.started_at, runs.repo FROM jobs JOIN runs ON runs.id = jobs.run_id WHERE jobs.id = ?") |
| 348 | .bind(&[id.as_str().into()])? |
| 349 | .first::<Found>(None) |
| 350 | .await?; |
| 351 | Ok(job.filter(|j| j.status == "in_progress").map(|j| RunnerWork { |
| 352 | kind: "workflow".into(), |
| 353 | id: j.id, |
| 354 | name: j.name, |
| 355 | repo: Some(j.repo), |
| 356 | run_id: Some(j.run_id), |
| 357 | started_at: j.started_at, |
| 358 | })) |
| 359 | } |
| 360 | |
| 361 | async fn runner_view(&self, row: &RunnerRow, groups: &[GroupRow], at: u64) -> Result<Runner> { |
| 362 | let online = row.online(at); |
| 363 | let work = if online { self.work_of(row).await? } else { None }; |
| 364 | Ok(Runner { |
| 365 | id: row.id.clone(), |
| 366 | name: row.name.clone(), |
| 367 | workspace: row.workspace.clone(), |
| 368 | repo: row.repo.clone(), |
| 369 | group: row |
| 370 | .group_id |
| 371 | .as_ref() |
| 372 | .and_then(|id| groups.iter().find(|g| &g.id == id)) |
| 373 | .map(|g| g.name.clone()), |
| 374 | labels: row.labels(), |
| 375 | os: row.os.clone(), |
| 376 | arch: row.arch.clone(), |
| 377 | version: row.version.clone(), |
| 378 | ephemeral: row.ephemeral != 0, |
| 379 | status: if !online { |
| 380 | "offline" |
| 381 | } else if work.is_some() { |
| 382 | "busy" |
| 383 | } else { |
| 384 | "online" |
| 385 | } |
| 386 | .to_owned(), |
| 387 | work, |
| 388 | last_seen_at: row.last_seen_at.clone(), |
| 389 | created_at: row.created_at.clone(), |
| 390 | created_by: row.created_by.clone(), |
| 391 | }) |
| 392 | } |
| 393 | |
| 394 | // --- People's side ----------------------------------------------------- |
| 395 | |
| 396 | /// `runners`: a workspace's runners, or a repository's: its own and |
| 397 | /// the workspace's that its group lets it use. |
| 398 | pub async fn runners(&self, a: ListRunnersArgs) -> Result<Outcome<Vec<Runner>>> { |
| 399 | let place = check!(self.runner_place(&a.actor, &a.owner, false).await?); |
| 400 | self.default_group(&place.workspace).await?; |
| 401 | let groups = self.group_rows(&place.workspace).await?; |
| 402 | let at = now_ms(); |
| 403 | let mut out = Vec::new(); |
| 404 | for row in self.runner_rows(&place.workspace).await? { |
| 405 | let shown = match &place.repo { |
| 406 | None => row.repo_id.is_none(), |
| 407 | Some(repo) => match &row.repo_id { |
| 408 | Some(id) => id == &repo.id, |
| 409 | None => groups.iter().find(|g| Some(&g.id) == row.group_id.as_ref()).is_none_or(|g| g.allows(&repo.name)), |
| 410 | }, |
| 411 | }; |
| 412 | if shown { |
| 413 | out.push(self.runner_view(&row, &groups, at).await?); |
| 414 | } |
| 415 | } |
| 416 | Ok(Outcome::Ok(out)) |
| 417 | } |
| 418 | |
| 419 | /// `create_registration_token`. |
| 420 | pub async fn create_registration_token(&self, a: CreateRegistrationTokenArgs) -> Result<Outcome<RegistrationToken>> { |
| 421 | let place = check!(self.runner_place(&a.actor, &a.owner, true).await?); |
| 422 | let made = self |
| 423 | .db |
| 424 | .prepare("SELECT COUNT(*) AS n FROM runner_registrations WHERE workspace = ? AND created_at > ?") |
| 425 | .bind(&[place.workspace.as_str().into(), rfc3339(now_ms().saturating_sub(60 * 60 * 1000)).into()])? |
| 426 | .first::<Count>(None) |
| 427 | .await? |
| 428 | .map_or(0, |c| c.n); |
| 429 | if made >= model::MAX_TOKENS_PER_HOUR { |
| 430 | return Ok(fail(FailureCode::Conflict, "Too many registration tokens in the last hour. Each lasts an hour and registers any number of runners.")); |
| 431 | } |
| 432 | let group = match (&place.repo, a.group.as_deref().map(str::trim).filter(|g| !g.is_empty())) { |
| 433 | (Some(_), Some(_)) => return Ok(fail(FailureCode::Invalid, "A repository's own runners are in no group.")), |
| 434 | (Some(_), None) => None, |
| 435 | (None, Some(wanted)) => match self.find_group(&place.workspace, wanted).await? { |
| 436 | Some(group) => Some(group), |
| 437 | None => return Ok(fail(FailureCode::NotFound, format!("There is no runner group called {wanted}."))), |
| 438 | }, |
| 439 | (None, None) => Some(self.default_group(&place.workspace).await?), |
| 440 | }; |
| 441 | let token = format!("{}{}", model::REGISTRATION_PREFIX, random_hex(24)); |
| 442 | let at = now_ms(); |
| 443 | let expires_at = rfc3339(at + model::REGISTRATION_TTL_SECONDS * 1000); |
| 444 | let repo_name = place.repo.as_ref().map(|r| format!("{}/{}", r.namespace, r.name)); |
| 445 | self.db |
| 446 | .prepare( |
| 447 | "INSERT INTO runner_registrations (token_hash, workspace, repo_id, repo, group_id, created_by, created_at, expires_at) |
| 448 | VALUES (?, ?, ?, ?, ?, ?, ?, ?)", |
| 449 | ) |
| 450 | .bind(&[ |
| 451 | hash(&token).into(), |
| 452 | place.workspace.as_str().into(), |
| 453 | optional(place.repo.as_ref().map(|r| r.id.as_str())), |
| 454 | optional(repo_name.as_deref()), |
| 455 | optional(group.as_ref().map(|g| g.id.as_str())), |
| 456 | a.actor.username.as_str().into(), |
| 457 | rfc3339(at).into(), |
| 458 | expires_at.as_str().into(), |
| 459 | ])? |
| 460 | .run() |
| 461 | .await?; |
| 462 | self.audit(vec![entry( |
| 463 | AuditActor::of(&a.actor), |
| 464 | "create_runner_registration_token", |
| 465 | &place.workspace, |
| 466 | repo_name.clone(), |
| 467 | None, |
| 468 | format!("Made a runner registration token{}.", group.as_ref().map(|g| format!(" for group {}", g.name)).unwrap_or_default()), |
| 469 | )]) |
| 470 | .await; |
| 471 | Ok(Outcome::Ok(RegistrationToken { |
| 472 | token, |
| 473 | expires_at, |
| 474 | workspace: place.workspace, |
| 475 | repo: repo_name, |
| 476 | group: group.map(|g| g.name), |
| 477 | url: crate::SITE.to_owned(), |
| 478 | })) |
| 479 | } |
| 480 | |
| 481 | /// `remove_runner`: a person removing one. |
| 482 | pub async fn remove_runner(&self, a: RemoveRunnerArgs) -> Result<Outcome<bool>> { |
| 483 | let place = check!(self.runner_place(&a.actor, &a.owner, true).await?); |
| 484 | let Some(row) = self.runner_by_id(&a.id).await? else { |
| 485 | return Ok(fail(FailureCode::NotFound, "There is no such runner.")); |
| 486 | }; |
| 487 | let ours = row.workspace == place.workspace |
| 488 | && match &place.repo { |
| 489 | Some(repo) => row.repo_id.as_deref() == Some(repo.id.as_str()), |
| 490 | None => row.repo_id.is_none(), |
| 491 | }; |
| 492 | if !ours { |
| 493 | return Ok(fail(FailureCode::NotFound, "There is no such runner.")); |
| 494 | } |
| 495 | self.forget_runner(&row, &format!("{} removed the runner {}.", a.actor.username, row.name)).await?; |
| 496 | self.audit(vec![entry( |
| 497 | AuditActor::of(&a.actor), |
| 498 | "remove_runner", |
| 499 | &row.workspace, |
| 500 | row.repo.clone(), |
| 501 | Some(row.name.clone()), |
| 502 | format!("Removed the self-hosted runner {}.", row.name), |
| 503 | )]) |
| 504 | .await; |
| 505 | Ok(Outcome::Ok(true)) |
| 506 | } |
| 507 | |
| 508 | /// Removes a runner: what it was running fails, and its credential |
| 509 | /// stops working at once. |
| 510 | async fn forget_runner(&self, row: &RunnerRow, why: &str) -> Result<()> { |
| 511 | self.db.prepare("DELETE FROM runners WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?; |
| 512 | self.abandon_work(&row.id, why).await |
| 513 | } |
| 514 | |
| 515 | /// Fails whatever a runner was running. |
| 516 | async fn abandon_work(&self, runner_id: &str, why: &str) -> Result<()> { |
| 517 | let jobs = self |
| 518 | .db |
| 519 | .prepare("SELECT id FROM jobs WHERE runner_id = ? AND status = 'in_progress'") |
| 520 | .bind(&[runner_id.into()])? |
| 521 | .all() |
| 522 | .await? |
| 523 | .results::<IdRow>()?; |
| 524 | for job in jobs { |
| 525 | self.finish_job(&job.id, "failure", Some(why), None).await?; |
| 526 | } |
| 527 | let tasks = self |
| 528 | .db |
| 529 | .prepare("SELECT * FROM runner_tasks WHERE runner_id = ? AND status = 'in_progress'") |
| 530 | .bind(&[runner_id.into()])? |
| 531 | .all() |
| 532 | .await? |
| 533 | .results::<TaskRow>()?; |
| 534 | for task in tasks { |
| 535 | self.end_task(&task, 1, why).await?; |
| 536 | } |
| 537 | Ok(()) |
| 538 | } |
| 539 | |
| 540 | /// `runner_groups`. |
| 541 | pub async fn runner_groups(&self, a: RunnerGroupsArgs) -> Result<Outcome<Vec<RunnerGroup>>> { |
| 542 | let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) }; |
| 543 | let place = check!(self.runner_place(&a.actor, &owner, false).await?); |
| 544 | self.default_group(&place.workspace).await?; |
| 545 | let groups = self.group_rows(&place.workspace).await?; |
| 546 | let runners = self.runner_rows(&place.workspace).await?; |
| 547 | Ok(Outcome::Ok(groups.iter().map(|g| group_view(g, &runners)).collect())) |
| 548 | } |
| 549 | |
| 550 | /// `set_runner_group`. |
| 551 | pub async fn set_runner_group(&self, a: SetRunnerGroupArgs) -> Result<Outcome<RunnerGroup>> { |
| 552 | let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) }; |
| 553 | let place = check!(self.runner_place(&a.actor, &owner, true).await?); |
| 554 | self.default_group(&place.workspace).await?; |
| 555 | let groups = self.group_rows(&place.workspace).await?; |
| 556 | let name = match a.name.as_deref().map(valid_group_name).transpose() { |
| 557 | Ok(name) => name, |
| 558 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 559 | }; |
| 560 | let repositories: Option<Vec<String>> = a.repositories.map(|list| { |
| 561 | let mut out: Vec<String> = Vec::new(); |
| 562 | for repo in list { |
| 563 | let name = repo.trim().rsplit('/').next().unwrap_or_default().to_owned(); |
| 564 | if !name.is_empty() && !out.iter().any(|r: &String| r.eq_ignore_ascii_case(&name)) { |
| 565 | out.push(name); |
| 566 | } |
| 567 | } |
| 568 | out |
| 569 | }); |
| 570 | if let Some(name) = &name |
| 571 | && groups.iter().any(|g| g.name.eq_ignore_ascii_case(name) && Some(&g.id) != a.id.as_ref()) |
| 572 | { |
| 573 | return Ok(fail(FailureCode::Conflict, format!("There is already a group called {name}."))); |
| 574 | } |
| 575 | let at = now(); |
| 576 | let id = match &a.id { |
| 577 | Some(id) => { |
| 578 | let Some(group) = groups.iter().find(|g| &g.id == id) else { |
| 579 | return Ok(fail(FailureCode::NotFound, "There is no such runner group.")); |
| 580 | }; |
| 581 | self.db |
| 582 | .prepare("UPDATE runner_groups SET name = ?, repositories = ?, updated_at = ? WHERE id = ?") |
| 583 | .bind(&[ |
| 584 | name.clone().unwrap_or(group.name.clone()).into(), |
| 585 | serde_json::to_string(&repositories.clone().unwrap_or(group.repositories()))?.into(), |
| 586 | at.as_str().into(), |
| 587 | id.as_str().into(), |
| 588 | ])? |
| 589 | .run() |
| 590 | .await?; |
| 591 | id.clone() |
| 592 | } |
| 593 | None => { |
| 594 | let Some(name) = name.clone() else { |
| 595 | return Ok(fail(FailureCode::Invalid, "A new group needs a name.")); |
| 596 | }; |
| 597 | if groups.len() >= 50 { |
| 598 | return Ok(fail(FailureCode::Conflict, "A workspace has at most 50 runner groups.")); |
| 599 | } |
| 600 | let id = new_id("rng", now_ms()); |
| 601 | self.db |
| 602 | .prepare( |
| 603 | "INSERT INTO runner_groups (id, workspace, name, is_default, repositories, created_at, updated_at) VALUES (?, ?, ?, 0, ?, ?, ?)", |
| 604 | ) |
| 605 | .bind(&[ |
| 606 | id.as_str().into(), |
| 607 | place.workspace.as_str().into(), |
| 608 | name.into(), |
| 609 | serde_json::to_string(&repositories.clone().unwrap_or_default())?.into(), |
| 610 | at.as_str().into(), |
| 611 | at.as_str().into(), |
| 612 | ])? |
| 613 | .run() |
| 614 | .await?; |
| 615 | id |
| 616 | } |
| 617 | }; |
| 618 | let groups = self.group_rows(&place.workspace).await?; |
| 619 | let runners = self.runner_rows(&place.workspace).await?; |
| 620 | let Some(group) = groups.iter().find(|g| g.id == id) else { |
| 621 | return Ok(fail(FailureCode::NotFound, "There is no such runner group.")); |
| 622 | }; |
| 623 | self.audit(vec![entry( |
| 624 | AuditActor::of(&a.actor), |
| 625 | if a.id.is_some() { "update_runner_group" } else { "create_runner_group" }, |
| 626 | &place.workspace, |
| 627 | None, |
| 628 | Some(group.name.clone()), |
| 629 | format!( |
| 630 | "Runner group {}: {}.", |
| 631 | group.name, |
| 632 | if group.repositories().is_empty() { "every repository".to_owned() } else { group.repositories().join(", ") } |
| 633 | ), |
| 634 | )]) |
| 635 | .await; |
| 636 | Ok(Outcome::Ok(group_view(group, &runners))) |
| 637 | } |
| 638 | |
| 639 | /// `delete_runner_group`. |
| 640 | pub async fn delete_runner_group(&self, a: DeleteRunnerGroupArgs) -> Result<Outcome<bool>> { |
| 641 | let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) }; |
| 642 | let place = check!(self.runner_place(&a.actor, &owner, true).await?); |
| 643 | let default = self.default_group(&place.workspace).await?; |
| 644 | let Some(group) = self.group_rows(&place.workspace).await?.into_iter().find(|g| g.id == a.id) else { |
| 645 | return Ok(fail(FailureCode::NotFound, "There is no such runner group.")); |
| 646 | }; |
| 647 | if group.is_default != 0 { |
| 648 | return Ok(fail(FailureCode::Conflict, "The default group cannot be deleted.")); |
| 649 | } |
| 650 | self.db |
| 651 | .batch(vec![ |
| 652 | self.db.prepare("UPDATE runners SET group_id = ? WHERE group_id = ?").bind(&[default.id.as_str().into(), group.id.as_str().into()])?, |
| 653 | self.db.prepare("DELETE FROM runner_groups WHERE id = ?").bind(&[group.id.as_str().into()])?, |
| 654 | ]) |
| 655 | .await?; |
| 656 | self.audit(vec![entry( |
| 657 | AuditActor::of(&a.actor), |
| 658 | "delete_runner_group", |
| 659 | &place.workspace, |
| 660 | None, |
| 661 | Some(group.name.clone()), |
| 662 | format!("Deleted the runner group {}; its runners joined {}.", group.name, default.name), |
| 663 | )]) |
| 664 | .await; |
| 665 | Ok(Outcome::Ok(true)) |
| 666 | } |
| 667 | |
| 668 | async fn settings_row(&self, owner: &str) -> Result<Option<SettingsRow>> { |
| 669 | self.db.prepare("SELECT * FROM runner_settings WHERE owner = ?").bind(&[owner.into()])?.first::<SettingsRow>(None).await |
| 670 | } |
| 671 | |
| 672 | fn settings_of(row: Option<SettingsRow>, inherited: bool) -> RunnerSettings { |
| 673 | match row { |
| 674 | Some(row) => RunnerSettings { |
| 675 | agents_on_self_hosted: row.agents != 0, |
| 676 | agent_labels: serde_json::from_str(&row.agent_labels).unwrap_or_else(|_| vec![model::SELF_HOSTED.to_owned()]), |
| 677 | fork_pull_requests: row.fork_pulls != 0, |
| 678 | inherited, |
| 679 | }, |
| 680 | None => RunnerSettings { |
| 681 | agent_labels: vec![model::SELF_HOSTED.to_owned()], |
| 682 | inherited, |
| 683 | ..RunnerSettings::default() |
| 684 | }, |
| 685 | } |
| 686 | } |
| 687 | |
| 688 | /// The settings that apply in a workspace, or in one of its |
| 689 | /// repositories: the repository's own if it has them. |
| 690 | pub async fn effective_runner_settings(&self, workspace: &str, repo_id: Option<&str>) -> Result<RunnerSettings> { |
| 691 | if let Some(repo_id) = repo_id |
| 692 | && let Some(own) = self.settings_row(repo_id).await? |
| 693 | { |
| 694 | return Ok(Self::settings_of(Some(own), false)); |
| 695 | } |
| 696 | let inherited = repo_id.is_some(); |
| 697 | Ok(Self::settings_of(self.settings_row(&workspace.to_lowercase()).await?, inherited)) |
| 698 | } |
| 699 | |
| 700 | /// `runner_settings`. |
| 701 | pub async fn runner_settings(&self, a: RunnerSettingsArgs) -> Result<Outcome<RunnerSettings>> { |
| 702 | let place = check!(self.runner_place(&a.actor, &a.owner, false).await?); |
| 703 | Ok(Outcome::Ok(self.effective_runner_settings(&place.workspace, place.repo.as_ref().map(|r| r.id.as_str())).await?)) |
| 704 | } |
| 705 | |
| 706 | /// `set_runner_settings`. |
| 707 | pub async fn set_runner_settings(&self, a: SetRunnerSettingsArgs) -> Result<Outcome<RunnerSettings>> { |
| 708 | let place = check!(self.runner_place(&a.actor, &a.owner, true).await?); |
| 709 | let (owner, scope) = match &place.repo { |
| 710 | Some(repo) => (repo.id.clone(), "repository"), |
| 711 | None => (place.workspace.clone(), "workspace"), |
| 712 | }; |
| 713 | if a.inherit && place.repo.is_some() { |
| 714 | self.db.prepare("DELETE FROM runner_settings WHERE owner = ?").bind(&[owner.as_str().into()])?.run().await?; |
| 715 | } else { |
| 716 | let current = self.effective_runner_settings(&place.workspace, place.repo.as_ref().map(|r| r.id.as_str())).await?; |
| 717 | let labels = match &a.agent_labels { |
| 718 | Some(given) => match model::runner_labels(given, model::SELF_HOSTED, model::SELF_HOSTED) { |
| 719 | Ok(labels) => labels, |
| 720 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 721 | }, |
| 722 | None => current.agent_labels.clone(), |
| 723 | }; |
| 724 | let agents = a.agents_on_self_hosted.unwrap_or(current.agents_on_self_hosted); |
| 725 | let forks = a.fork_pull_requests.unwrap_or(current.fork_pull_requests); |
| 726 | self.db |
| 727 | .prepare( |
| 728 | "INSERT INTO runner_settings (owner, scope, agents, agent_labels, fork_pulls, updated_at, updated_by) |
| 729 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) |
| 730 | ON CONFLICT (owner) DO UPDATE SET agents = ?3, agent_labels = ?4, fork_pulls = ?5, updated_at = ?6, updated_by = ?7", |
| 731 | ) |
| 732 | .bind(&[ |
| 733 | owner.as_str().into(), |
| 734 | scope.into(), |
| 735 | u32::from(agents).into(), |
| 736 | serde_json::to_string(&labels)?.into(), |
| 737 | u32::from(forks).into(), |
| 738 | now().into(), |
| 739 | a.actor.username.as_str().into(), |
| 740 | ])? |
| 741 | .run() |
| 742 | .await?; |
| 743 | } |
| 744 | let settings = self.effective_runner_settings(&place.workspace, place.repo.as_ref().map(|r| r.id.as_str())).await?; |
| 745 | self.audit(vec![entry( |
| 746 | AuditActor::of(&a.actor), |
| 747 | "update_runner_settings", |
| 748 | &place.workspace, |
| 749 | place.repo.as_ref().map(|r| format!("{}/{}", r.namespace, r.name)), |
| 750 | None, |
| 751 | format!( |
| 752 | "Agents on self-hosted runners: {}; labels {}; pull requests from forks: {}.", |
| 753 | if settings.agents_on_self_hosted { "on" } else { "off" }, |
| 754 | settings.agent_labels.join(", "), |
| 755 | if settings.fork_pull_requests { "allowed" } else { "not allowed" } |
| 756 | ), |
| 757 | )]) |
| 758 | .await; |
| 759 | Ok(Outcome::Ok(settings)) |
| 760 | } |
| 761 | |
| 762 | // --- The runner's side --------------------------------------------------- |
| 763 | |
| 764 | /// `runner_register`. |
| 765 | pub async fn runner_register(&self, a: RegisterArgs) -> Result<Outcome<Registered>> { |
| 766 | let refused = || fail(FailureCode::Unauthenticated, "That registration token is not valid, or it expired. Make a new one under Settings, Runners."); |
| 767 | if !a.token.starts_with(model::REGISTRATION_PREFIX) { |
| 768 | return Ok(refused()); |
| 769 | } |
| 770 | let registration = self |
| 771 | .db |
| 772 | .prepare("SELECT * FROM runner_registrations WHERE token_hash = ?") |
| 773 | .bind(&[hash(&a.token).into()])? |
| 774 | .first::<RegistrationRow>(None) |
| 775 | .await?; |
| 776 | let at = now_ms(); |
| 777 | let Some(registration) = registration.filter(|r| r.expires_at.as_str() > rfc3339(at).as_str()) else { |
| 778 | return Ok(refused()); |
| 779 | }; |
| 780 | let name = match valid_name(&a.name) { |
| 781 | Ok(name) => name, |
| 782 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 783 | }; |
| 784 | let (Some(os), Some(arch)) = (model::os_label(&a.os), model::arch_label(&a.arch)) else { |
| 785 | return Ok(fail(FailureCode::Invalid, "A runner's os is linux, macos or windows, and its arch x64 or arm64.")); |
| 786 | }; |
| 787 | let labels = match model::runner_labels(&a.labels, os, arch) { |
| 788 | Ok(labels) => labels, |
| 789 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 790 | }; |
| 791 | let workspace = registration.workspace.clone(); |
| 792 | // Abuse: so many machines, so fast, are not a team's. |
| 793 | let recent = self |
| 794 | .db |
| 795 | .prepare("SELECT COUNT(*) AS n FROM runners WHERE workspace = ? AND created_at > ?") |
| 796 | .bind(&[workspace.as_str().into(), rfc3339(at.saturating_sub(60 * 1000)).into()])? |
| 797 | .first::<Count>(None) |
| 798 | .await? |
| 799 | .map_or(0, |c| c.n); |
| 800 | if recent >= model::MAX_REGISTRATIONS_PER_MINUTE { |
| 801 | return Ok(fail(FailureCode::Conflict, "Too many runners registered in the last minute. Wait a moment and try again.")); |
| 802 | } |
| 803 | let total = self |
| 804 | .db |
| 805 | .prepare("SELECT COUNT(*) AS n FROM runners WHERE workspace = ?") |
| 806 | .bind(&[workspace.as_str().into()])? |
| 807 | .first::<Count>(None) |
| 808 | .await? |
| 809 | .map_or(0, |c| c.n); |
| 810 | // A group: the one asked for, the token's, or the default. |
| 811 | let group = if registration.repo_id.is_some() { |
| 812 | None |
| 813 | } else { |
| 814 | match a.group.as_deref().map(str::trim).filter(|g| !g.is_empty()) { |
| 815 | Some(wanted) => match self.find_group(&workspace, wanted).await? { |
| 816 | Some(group) => Some(group.id), |
| 817 | None => return Ok(fail(FailureCode::NotFound, format!("There is no runner group called {wanted}."))), |
| 818 | }, |
| 819 | None => match registration.group_id.clone() { |
| 820 | Some(id) => Some(id), |
| 821 | None => Some(self.default_group(&workspace).await?.id), |
| 822 | }, |
| 823 | } |
| 824 | }; |
| 825 | let existing = self |
| 826 | .db |
| 827 | .prepare("SELECT * FROM runners WHERE workspace = ? AND COALESCE(repo_id, '') = ? AND lower(name) = lower(?)") |
| 828 | .bind(&[workspace.as_str().into(), registration.repo_id.clone().unwrap_or_default().into(), name.as_str().into()])? |
| 829 | .first::<RunnerRow>(None) |
| 830 | .await?; |
| 831 | if let Some(existing) = &existing { |
| 832 | if !a.replace { |
| 833 | return Ok(fail( |
| 834 | FailureCode::Conflict, |
| 835 | format!("A runner called {name} is already registered here. Register with --replace to take its place, or choose another --name."), |
| 836 | )); |
| 837 | } |
| 838 | self.forget_runner(existing, &format!("The runner {name} was replaced by a new registration.")).await?; |
| 839 | } else if total >= model::MAX_RUNNERS { |
| 840 | return Ok(fail(FailureCode::Conflict, format!("A workspace has at most {} self-hosted runners.", model::MAX_RUNNERS))); |
| 841 | } |
| 842 | // How it keeps work from its machine, if it said (older runners do not). |
| 843 | let isolation = a.isolation.as_deref().filter(|level| model::ISOLATIONS.contains(level)); |
| 844 | let id = new_id("rnr", at); |
| 845 | let credential = format!("{}{}", model::CREDENTIAL_PREFIX, random_hex(32)); |
| 846 | let created = rfc3339(at); |
| 847 | self.db |
| 848 | .prepare( |
| 849 | "INSERT INTO runners (id, workspace, repo_id, repo, group_id, name, labels, os, arch, version, ephemeral, credential_hash, |
| 850 | rotated_at, last_seen_at, created_at, created_by, isolation) |
| 851 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 852 | ) |
| 853 | .bind(&[ |
| 854 | id.as_str().into(), |
| 855 | workspace.as_str().into(), |
| 856 | optional(registration.repo_id.as_deref()), |
| 857 | optional(registration.repo.as_deref()), |
| 858 | optional(group.as_deref()), |
| 859 | name.as_str().into(), |
| 860 | serde_json::to_string(&labels)?.into(), |
| 861 | os.into(), |
| 862 | arch.into(), |
| 863 | a.version.chars().take(40).collect::<String>().into(), |
| 864 | u32::from(a.ephemeral).into(), |
| 865 | hash(&credential).into(), |
| 866 | created.as_str().into(), |
| 867 | created.as_str().into(), |
| 868 | created.as_str().into(), |
| 869 | JsValue::NULL, |
| 870 | optional(isolation), |
| 871 | ])? |
| 872 | .run() |
| 873 | .await?; |
| 874 | self.db |
| 875 | .prepare("UPDATE runner_registrations SET used = used + 1 WHERE token_hash = ?") |
| 876 | .bind(&[hash(&a.token).into()])? |
| 877 | .run() |
| 878 | .await?; |
| 879 | let row = self.runner_by_id(&id).await?.ok_or_else(|| worker::Error::RustError("the runner was not kept".into()))?; |
| 880 | self.audit(vec![entry( |
| 881 | runner_actor(&row), |
| 882 | "register_runner", |
| 883 | &workspace, |
| 884 | row.repo.clone(), |
| 885 | Some(name.clone()), |
| 886 | format!("Registered the self-hosted runner {name} ({os}, {arch}; labels {}).", labels.join(", ")), |
| 887 | )]) |
| 888 | .await; |
| 889 | let groups = self.group_rows(&workspace).await?; |
| 890 | Ok(Outcome::Ok(Registered { |
| 891 | runner: self.runner_view(&row, &groups, at).await?, |
| 892 | credential, |
| 893 | })) |
| 894 | } |
| 895 | |
| 896 | /// The runner a call is from, if its credential is that runner's (or |
| 897 | /// the one before a rotation, until the new one is used). |
| 898 | async fn authenticated(&self, auth: &RunnerAuth) -> Result<Option<(RunnerRow, bool)>> { |
| 899 | if !auth.credential.starts_with(model::CREDENTIAL_PREFIX) { |
| 900 | return Ok(None); |
| 901 | } |
| 902 | let Some(row) = self.runner_by_id(&auth.runner).await? else { return Ok(None) }; |
| 903 | let given = hash(&auth.credential); |
| 904 | if g1t_secrets::same(&row.credential_hash, &given) { |
| 905 | if row.previous_hash.is_some() { |
| 906 | self.db.prepare("UPDATE runners SET previous_hash = NULL WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?; |
| 907 | } |
| 908 | return Ok(Some((row, true))); |
| 909 | } |
| 910 | if row.previous_hash.as_deref().is_some_and(|previous| g1t_secrets::same(previous, &given)) { |
| 911 | return Ok(Some((row, false))); |
| 912 | } |
| 913 | Ok(None) |
| 914 | } |
| 915 | |
| 916 | fn unknown_runner<T>() -> Outcome<T> { |
| 917 | fail(FailureCode::Unauthenticated, "This runner is not registered, or its credential is not valid. Register it again.") |
| 918 | } |
| 919 | |
| 920 | /// `runner_poll`. |
| 921 | pub async fn runner_poll(&self, a: PollArgs) -> Result<Outcome<Poll>> { |
| 922 | let Some((row, current)) = self.authenticated(&a.auth).await? else { |
| 923 | return Ok(Self::unknown_runner()); |
| 924 | }; |
| 925 | let at = now_ms(); |
| 926 | let seen = rfc3339(at); |
| 927 | // A new credential every day, handed over on a poll that used the |
| 928 | // current one. |
| 929 | let mut credential = None; |
| 930 | let rotate_after = rfc3339(at.saturating_sub(model::ROTATE_AFTER_MS)); |
| 931 | if current && row.rotated_at.as_str() < rotate_after.as_str() { |
| 932 | let fresh = format!("{}{}", model::CREDENTIAL_PREFIX, random_hex(32)); |
| 933 | self.db |
| 934 | .prepare("UPDATE runners SET previous_hash = credential_hash, credential_hash = ?, rotated_at = ? WHERE id = ?") |
| 935 | .bind(&[hash(&fresh).into(), seen.as_str().into(), row.id.as_str().into()])? |
| 936 | .run() |
| 937 | .await?; |
| 938 | credential = Some(fresh); |
| 939 | } |
| 940 | // What it holds that it should stop, and the heartbeat for the rest. |
| 941 | let mut cancel = Vec::new(); |
| 942 | let mut active: Option<(String, &'static str)> = None; |
| 943 | for id in a.running.iter().take(10) { |
| 944 | if let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[id.as_str().into()])?.first::<JobRow>(None).await? { |
| 945 | if job.status == "in_progress" && job.runner_id.as_deref() == Some(row.id.as_str()) { |
| 946 | if seen_is_stale(job.seen_at.as_deref(), at) { |
| 947 | self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?; |
| 948 | } |
| 949 | active = Some((id.clone(), "workflow")); |
| 950 | } else { |
| 951 | cancel.push(id.clone()); |
| 952 | } |
| 953 | continue; |
| 954 | } |
| 955 | match self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[id.as_str().into()])?.first::<TaskRow>(None).await? { |
| 956 | Some(task) if task.status == "in_progress" && task.runner_id.as_deref() == Some(row.id.as_str()) => { |
| 957 | if seen_is_stale(task.seen_at.as_deref(), at) { |
| 958 | self.db.prepare("UPDATE runner_tasks SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?; |
| 959 | } |
| 960 | active = Some((id.clone(), "agent")); |
| 961 | } |
| 962 | _ => cancel.push(id.clone()), |
| 963 | } |
| 964 | } |
| 965 | // Written when something changed, or once a minute to keep it online. |
| 966 | let version: String = if a.version.is_empty() { row.version.clone() } else { a.version.chars().take(40).collect() }; |
| 967 | let work_id = active.as_ref().map(|(id, _)| id.as_str()); |
| 968 | let work_kind = active.as_ref().map(|(_, kind)| *kind); |
| 969 | // How it is isolated, as it says now: a runner re-registered or |
| 970 | // restarted with or without Docker says so on its next poll. |
| 971 | let isolation = a |
| 972 | .isolation |
| 973 | .as_deref() |
| 974 | .filter(|level| model::ISOLATIONS.contains(level)) |
| 975 | .map(str::to_owned) |
| 976 | .or_else(|| row.isolation.clone()); |
| 977 | let changed = version != row.version |
| 978 | || work_id != row.work_id.as_deref() |
| 979 | || work_kind != row.work_kind.as_deref() |
| 980 | || isolation != row.isolation; |
| 981 | if changed || seen_is_stale(row.last_seen_at.as_deref(), at) { |
| 982 | self.db |
| 983 | .prepare("UPDATE runners SET last_seen_at = ?, version = ?, work_id = ?, work_kind = ?, isolation = ? WHERE id = ?") |
| 984 | .bind(&[ |
| 985 | seen.as_str().into(), |
| 986 | version.into(), |
| 987 | optional(work_id), |
| 988 | optional(work_kind), |
| 989 | optional(isolation.as_deref()), |
| 990 | row.id.as_str().into(), |
| 991 | ])? |
| 992 | .run() |
| 993 | .await?; |
| 994 | } |
| 995 | let row = RunnerRow { isolation, ..row }; |
| 996 | let mut poll = Poll { assignment: None, cancel, credential, removed: false }; |
| 997 | // Busy, or an ephemeral runner that has had its one job. |
| 998 | if active.is_some() || !a.running.is_empty() { |
| 999 | return Ok(Outcome::Ok(poll)); |
| 1000 | } |
| 1001 | if row.ephemeral != 0 && row.spent != 0 { |
| 1002 | self.forget_runner(&row, "The ephemeral runner finished its job.").await?; |
| 1003 | poll.removed = true; |
| 1004 | return Ok(Outcome::Ok(poll)); |
| 1005 | } |
| 1006 | let groups = self.group_rows(&row.workspace).await?; |
| 1007 | let group = row.group_id.as_ref().and_then(|id| groups.iter().find(|g| &g.id == id)).cloned(); |
| 1008 | let deadline = at + a.wait_ms.min(model::MAX_POLL_WAIT_MS); |
| 1009 | let mut looked = Looked::default(); |
| 1010 | loop { |
| 1011 | if let Some(assignment) = self.claim(&row, group.as_ref(), &mut looked).await? { |
| 1012 | poll.assignment = Some(assignment); |
| 1013 | return Ok(Outcome::Ok(poll)); |
| 1014 | } |
| 1015 | if now_ms() + POLL_EVERY_MS > deadline { |
| 1016 | return Ok(Outcome::Ok(poll)); |
| 1017 | } |
| 1018 | worker::Delay::from(std::time::Duration::from_millis(POLL_EVERY_MS)).await; |
| 1019 | } |
| 1020 | } |
| 1021 | |
| 1022 | /// Takes the oldest work this runner can do, if there is any. |
| 1023 | async fn claim(&self, runner: &RunnerRow, group: Option<&GroupRow>, looked: &mut Looked) -> Result<Option<Assignment>> { |
| 1024 | let labels = runner.labels(); |
| 1025 | let group_name = group.map(|g| g.name.clone()); |
| 1026 | // Workflow jobs. |
| 1027 | let mut sql = "SELECT jobs.*, runs.repo AS run_repo FROM jobs JOIN runs ON runs.id = jobs.run_id |
| 1028 | WHERE jobs.status = 'queued' AND jobs.labels IS NOT NULL AND lower(jobs.namespace) = ?" |
| 1029 | .to_owned(); |
| 1030 | let mut binds: Vec<JsValue> = vec![runner.workspace.as_str().into()]; |
| 1031 | if let Some(repo_id) = &runner.repo_id { |
| 1032 | sql.push_str(" AND jobs.repo_id = ?"); |
| 1033 | binds.push(repo_id.as_str().into()); |
| 1034 | } |
| 1035 | // One job of a concurrency group runs at a time. |
| 1036 | sql.push_str( |
| 1037 | " AND (jobs.concurrency_group IS NULL OR NOT EXISTS (SELECT 1 FROM jobs o WHERE o.repo_id = jobs.repo_id |
| 1038 | AND o.concurrency_group = jobs.concurrency_group AND o.status IN ('in_progress', 'calling')))", |
| 1039 | ); |
| 1040 | sql.push_str(" ORDER BY jobs.rowid LIMIT 50"); |
| 1041 | #[derive(Deserialize)] |
| 1042 | struct Queued { |
| 1043 | #[serde(flatten)] |
| 1044 | job: JobRow, |
| 1045 | run_repo: String, |
| 1046 | } |
| 1047 | let queued = self.db.prepare(sql).bind(&binds)?.all().await?.results::<Queued>()?; |
| 1048 | for Queued { job, run_repo } in queued { |
| 1049 | let stored: Vec<String> = job.labels.as_deref().and_then(|l| serde_json::from_str(l).ok()).unwrap_or_default(); |
| 1050 | let wanted = Wanted::from_stored(&stored); |
| 1051 | if !wanted.matches(&labels, group_name.as_deref()) { |
| 1052 | continue; |
| 1053 | } |
| 1054 | if runner.repo_id.is_none() && !group.is_none_or(|g| g.allows(&run_repo)) { |
| 1055 | continue; |
| 1056 | } |
| 1057 | // A host-mode runner takes no public repository's job: it waits |
| 1058 | // for a sandboxed runner, and says so when none could take it. |
| 1059 | if runner.isolation.as_deref() == Some("host") { |
| 1060 | let visibility = self.visibility_once(looked, &job.repo_id).await; |
| 1061 | if !model::isolation_allows(runner.isolation.as_deref(), visibility.as_deref()) { |
| 1062 | if looked.noted.insert(job.id.clone()) { |
| 1063 | self.wait_for_sandbox(&job, &run_repo, &wanted).await?; |
| 1064 | } |
| 1065 | continue; |
| 1066 | } |
| 1067 | } |
| 1068 | if let Some(max) = job.max_parallel { |
| 1069 | let siblings = self |
| 1070 | .db |
| 1071 | .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'") |
| 1072 | .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])? |
| 1073 | .first::<Count>(None) |
| 1074 | .await? |
| 1075 | .map_or(0, |c| c.n); |
| 1076 | if siblings >= max { |
| 1077 | continue; |
| 1078 | } |
| 1079 | } |
| 1080 | let token = random_hex(24); |
| 1081 | let at = now(); |
| 1082 | let claimed = self |
| 1083 | .db |
| 1084 | .prepare( |
| 1085 | "UPDATE jobs SET status = 'in_progress', token_hash = ?, runner_id = ?, runner_name = ?, reason = NULL, started_at = ?, seen_at = ? |
| 1086 | WHERE id = ? AND status = 'queued' RETURNING id", |
| 1087 | ) |
| 1088 | .bind(&[ |
| 1089 | sha256_hex(&token).into(), |
| 1090 | runner.id.as_str().into(), |
| 1091 | runner.name.as_str().into(), |
| 1092 | at.as_str().into(), |
| 1093 | at.as_str().into(), |
| 1094 | job.id.as_str().into(), |
| 1095 | ])? |
| 1096 | .first::<Value>(None) |
| 1097 | .await?; |
| 1098 | if claimed.is_none() { |
| 1099 | continue; |
| 1100 | } |
| 1101 | self.db |
| 1102 | .batch(vec![ |
| 1103 | self.db |
| 1104 | .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'") |
| 1105 | .bind(&[at.as_str().into(), job.run_id.as_str().into()])?, |
| 1106 | self.db |
| 1107 | .prepare("UPDATE runners SET work_id = ?, work_kind = 'workflow', spent = ephemeral WHERE id = ?") |
| 1108 | .bind(&[job.id.as_str().into(), runner.id.as_str().into()])?, |
| 1109 | ]) |
| 1110 | .await?; |
| 1111 | self.audit(vec![entry( |
| 1112 | runner_actor(runner), |
| 1113 | "runner.take_job", |
| 1114 | &runner.workspace, |
| 1115 | Some(run_repo.clone()), |
| 1116 | Some(job.name.clone()), |
| 1117 | format!("The self-hosted runner {} took the job {}.", runner.name, job.name), |
| 1118 | )]) |
| 1119 | .await; |
| 1120 | // A job that deploys: its run's deployment is under way. |
| 1121 | self.job_started(&job.id).await?; |
| 1122 | let image = self.job_image(&job).await?; |
| 1123 | let visibility = self.visibility(&job.repo_id).await; |
| 1124 | return Ok(Some(Assignment { |
| 1125 | kind: "workflow".into(), |
| 1126 | id: job.id.clone(), |
| 1127 | name: job.name.clone(), |
| 1128 | repo: run_repo, |
| 1129 | timeout_minutes: job.timeout_minutes, |
| 1130 | image, |
| 1131 | token: Some(token), |
| 1132 | env: None, |
| 1133 | visibility, |
| 1134 | })); |
| 1135 | } |
| 1136 | |
| 1137 | // Agent work. |
| 1138 | let tasks = self |
| 1139 | .db |
| 1140 | .prepare("SELECT * FROM runner_tasks WHERE status = 'queued' AND workspace = ? ORDER BY created_at LIMIT 20") |
| 1141 | .bind(&[runner.workspace.as_str().into()])? |
| 1142 | .all() |
| 1143 | .await? |
| 1144 | .results::<TaskRow>()?; |
| 1145 | for task in tasks { |
| 1146 | let wanted = Wanted::of(&serde_json::from_str::<Vec<String>>(&task.labels).unwrap_or_default(), None); |
| 1147 | if !wanted.matches(&labels, None) { |
| 1148 | continue; |
| 1149 | } |
| 1150 | // Nor a public repository's agent work. |
| 1151 | if runner.isolation.as_deref() == Some("host") |
| 1152 | && let Some(repo_id) = &task.repo_id |
| 1153 | { |
| 1154 | let visibility = self.visibility_once(looked, repo_id).await; |
| 1155 | if !model::isolation_allows(runner.isolation.as_deref(), visibility.as_deref()) { |
| 1156 | continue; |
| 1157 | } |
| 1158 | } |
| 1159 | if let Some(repo_id) = &runner.repo_id |
| 1160 | && task.repo_id.as_deref() != Some(repo_id.as_str()) |
| 1161 | { |
| 1162 | continue; |
| 1163 | } |
| 1164 | if runner.repo_id.is_none() && !group.is_none_or(|g| g.allows(&task.repo)) { |
| 1165 | continue; |
| 1166 | } |
| 1167 | let Some(sealed) = task.env.clone() else { continue }; |
| 1168 | let at = now(); |
| 1169 | let claimed = self |
| 1170 | .db |
| 1171 | .prepare( |
| 1172 | "UPDATE runner_tasks SET status = 'in_progress', runner_id = ?, runner_name = ?, env = NULL, started_at = ?, seen_at = ? |
| 1173 | WHERE id = ? AND status = 'queued' RETURNING id", |
| 1174 | ) |
| 1175 | .bind(&[runner.id.as_str().into(), runner.name.as_str().into(), at.as_str().into(), at.as_str().into(), task.id.as_str().into()])? |
| 1176 | .first::<Value>(None) |
| 1177 | .await?; |
| 1178 | if claimed.is_none() { |
| 1179 | continue; |
| 1180 | } |
| 1181 | self.db |
| 1182 | .prepare("UPDATE runners SET work_id = ?, work_kind = 'agent', spent = ephemeral WHERE id = ?") |
| 1183 | .bind(&[task.id.as_str().into(), runner.id.as_str().into()])? |
| 1184 | .run() |
| 1185 | .await?; |
| 1186 | let env: Map<String, Value> = self |
| 1187 | .sealer |
| 1188 | .as_ref() |
| 1189 | .and_then(|sealer| sealer.open(&sealed, &task.id)) |
| 1190 | .and_then(|text| serde_json::from_str(&text).ok()) |
| 1191 | .unwrap_or_default(); |
| 1192 | if env.is_empty() { |
| 1193 | self.end_task(&TaskRow { status: "in_progress".into(), ..task.clone() }, 1, "The work's environment could not be opened.").await?; |
| 1194 | continue; |
| 1195 | } |
| 1196 | self.audit(vec![entry( |
| 1197 | runner_actor(runner), |
| 1198 | "runner.take_task", |
| 1199 | &runner.workspace, |
| 1200 | Some(task.repo.clone()), |
| 1201 | Some(task.kind.clone()), |
| 1202 | format!("The self-hosted runner {} took agent work: {}.", runner.name, task.title), |
| 1203 | )]) |
| 1204 | .await; |
| 1205 | let visibility = match &task.repo_id { |
| 1206 | Some(repo_id) => self.visibility(repo_id).await, |
| 1207 | None => None, |
| 1208 | }; |
| 1209 | return Ok(Some(Assignment { |
| 1210 | kind: "agent".into(), |
| 1211 | id: task.id.clone(), |
| 1212 | name: task.title.clone(), |
| 1213 | repo: task.repo.clone(), |
| 1214 | timeout_minutes: task.timeout_minutes, |
| 1215 | image: None, |
| 1216 | token: None, |
| 1217 | env: Some(env), |
| 1218 | visibility, |
| 1219 | })); |
| 1220 | } |
| 1221 | Ok(None) |
| 1222 | } |
| 1223 | |
| 1224 | /// [`Self::visibility`], asked once per repository in a poll. |
| 1225 | async fn visibility_once(&self, looked: &mut Looked, repo_id: &str) -> Option<String> { |
| 1226 | if let Some(known) = looked.visibilities.get(repo_id) { |
| 1227 | return known.clone(); |
| 1228 | } |
| 1229 | let found = self.visibility(repo_id).await; |
| 1230 | looked.visibilities.insert(repo_id.to_owned(), found.clone()); |
| 1231 | found |
| 1232 | } |
| 1233 | |
| 1234 | /// A public repository's job that a host-mode runner passed over: when |
| 1235 | /// no online runner that may take it could (its labels, group, |
| 1236 | /// repository, and isolation), its page says it waits for a sandboxed |
| 1237 | /// one. Written only when that changes what it says. |
| 1238 | async fn wait_for_sandbox(&self, job: &JobRow, run_repo: &str, wanted: &Wanted) -> Result<()> { |
| 1239 | let at = now_ms(); |
| 1240 | let groups = self.group_rows(&job.namespace.to_lowercase()).await?; |
| 1241 | let runners = self.runner_rows(&job.namespace.to_lowercase()).await?; |
| 1242 | let sandboxed = runners.iter().any(|other| takes_public_job(other, &groups, &job.repo_id, run_repo, wanted, at)); |
| 1243 | if sandboxed { |
| 1244 | return Ok(()); |
| 1245 | } |
| 1246 | self.db |
| 1247 | .prepare("UPDATE jobs SET reason = ? WHERE id = ? AND status = 'queued' AND reason IS NOT ?") |
| 1248 | .bind(&[model::WAITING_FOR_SANDBOX.into(), job.id.as_str().into(), model::WAITING_FOR_SANDBOX.into()])? |
| 1249 | .run() |
| 1250 | .await?; |
| 1251 | Ok(()) |
| 1252 | } |
| 1253 | |
| 1254 | /// A repository's visibility for a runner, `public` or `private`; none |
| 1255 | /// when it cannot be read, and the runner decides without it. |
| 1256 | async fn visibility(&self, repo_id: &str) -> Option<String> { |
| 1257 | match self.repo_by_id(repo_id).await { |
| 1258 | Ok(Some((repo, _))) => Some(if repo.is_private { "private" } else { "public" }.to_owned()), |
| 1259 | _ => None, |
| 1260 | } |
| 1261 | } |
| 1262 | |
| 1263 | /// The image a job's `container:` names, when it names one plainly. |
| 1264 | async fn job_image(&self, job: &JobRow) -> Result<Option<String>> { |
| 1265 | let Some(run) = self.run_row(&job.run_id).await? else { return Ok(None) }; |
| 1266 | let raw = match job.callee() { |
| 1267 | Some((_, spec, _)) => spec.raw, |
| 1268 | None => match g1t_actions::workflow::parse(&run.source) { |
| 1269 | Ok(workflow) => match workflow.jobs.into_iter().find(|j| j.id == job.key) { |
| 1270 | Some(spec) => spec.raw, |
| 1271 | None => return Ok(None), |
| 1272 | }, |
| 1273 | Err(_) => return Ok(None), |
| 1274 | }, |
| 1275 | }; |
| 1276 | let image = match raw.get("container") { |
| 1277 | Some(Value::String(image)) => Some(image.clone()), |
| 1278 | Some(Value::Object(container)) => container.get("image").and_then(Value::as_str).map(str::to_owned), |
| 1279 | _ => None, |
| 1280 | }; |
| 1281 | Ok(image.filter(|image| !image.contains("${{") && !image.trim().is_empty())) |
| 1282 | } |
| 1283 | |
| 1284 | /// `runner_finished`. |
| 1285 | pub async fn runner_finished(&self, a: FinishedArgs) -> Result<Outcome<bool>> { |
| 1286 | let Some((row, _)) = self.authenticated(&a.auth).await? else { |
| 1287 | return Ok(Self::unknown_runner()); |
| 1288 | }; |
| 1289 | let reason = a.reason.clone().filter(|r| !r.trim().is_empty()).map(|r| r.chars().take(2000).collect::<String>()); |
| 1290 | let mut changed = false; |
| 1291 | if let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.id.as_str().into()])?.first::<JobRow>(None).await? { |
| 1292 | if job.runner_id.as_deref() == Some(row.id.as_str()) && job.status == "in_progress" { |
| 1293 | let why = reason.unwrap_or_else(|| format!("The job's process on {} exited with code {} before it reported how the job went.", row.name, a.exit_code)); |
| 1294 | self.finish_job(&job.id, "failure", Some(&why), None).await?; |
| 1295 | changed = true; |
| 1296 | } |
| 1297 | } else if let Some(task) = self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[a.id.as_str().into()])?.first::<TaskRow>(None).await? |
| 1298 | && task.runner_id.as_deref() == Some(row.id.as_str()) |
| 1299 | && task.status == "in_progress" |
| 1300 | { |
| 1301 | let why = reason.unwrap_or_else(|| if a.exit_code == 0 { String::new() } else { format!("The work exited with code {} on {}.", a.exit_code, row.name) }); |
| 1302 | self.end_task(&task, a.exit_code, &why).await?; |
| 1303 | changed = true; |
| 1304 | } |
| 1305 | self.db |
| 1306 | .prepare("UPDATE runners SET work_id = NULL, work_kind = NULL, last_seen_at = ? WHERE id = ? AND work_id = ?") |
| 1307 | .bind(&[now().into(), row.id.as_str().into(), a.id.as_str().into()])? |
| 1308 | .run() |
| 1309 | .await?; |
| 1310 | Ok(Outcome::Ok(changed)) |
| 1311 | } |
| 1312 | |
| 1313 | /// `runner_remove_self`. |
| 1314 | pub async fn runner_remove_self(&self, a: RemoveSelfArgs) -> Result<Outcome<bool>> { |
| 1315 | let Some((row, _)) = self.authenticated(&a.auth).await? else { |
| 1316 | return Ok(Self::unknown_runner()); |
| 1317 | }; |
| 1318 | self.forget_runner(&row, &format!("The runner {} was removed from its machine.", row.name)).await?; |
| 1319 | self.audit(vec![entry( |
| 1320 | runner_actor(&row), |
| 1321 | "remove_runner", |
| 1322 | &row.workspace, |
| 1323 | row.repo.clone(), |
| 1324 | Some(row.name.clone()), |
| 1325 | format!("The self-hosted runner {} removed itself.", row.name), |
| 1326 | )]) |
| 1327 | .await; |
| 1328 | Ok(Outcome::Ok(true)) |
| 1329 | } |
| 1330 | |
| 1331 | /// A self-hosted job finished, however it did: its runner is free, and |
| 1332 | /// its time goes on the workspace's usage as self-hosted, at $0. |
| 1333 | pub async fn released(&self, job: &JobRow) -> Result<()> { |
| 1334 | let Some(runner_id) = &job.runner_id else { return Ok(()) }; |
| 1335 | self.db |
| 1336 | .prepare("UPDATE runners SET work_id = NULL, work_kind = NULL WHERE id = ? AND work_id = ?") |
| 1337 | .bind(&[runner_id.as_str().into(), job.id.as_str().into()])? |
| 1338 | .run() |
| 1339 | .await?; |
| 1340 | let (Some(started), Some(finished)) = (job.started_at.as_deref(), job.finished_at.as_deref().map(str::to_owned).or_else(|| Some(now()))) else { |
| 1341 | return Ok(()); |
| 1342 | }; |
| 1343 | let seconds = match (g1t_contracts::time::parse_rfc3339(started), g1t_contracts::time::parse_rfc3339(&finished)) { |
| 1344 | (Some(a), Some(b)) if b > a => ((b - a) / 1000).max(1), |
| 1345 | _ => 1, |
| 1346 | }; |
| 1347 | let run = self.run_row(&job.run_id).await?; |
| 1348 | let repo = run.as_ref().map(|r| r.repo.clone()); |
| 1349 | let recorded: Result<Outcome<bool>> = g1t_kit::call( |
| 1350 | &self.billing, |
| 1351 | "record_sandbox", |
| 1352 | &RecordSandboxArgs { |
| 1353 | workspace: job.namespace.to_lowercase(), |
| 1354 | seconds: u32::try_from(seconds).unwrap_or(u32::MAX), |
| 1355 | description: format!( |
| 1356 | "{} in {} on the self-hosted runner {}", |
| 1357 | job.name, |
| 1358 | repo.clone().unwrap_or_default(), |
| 1359 | job.runner_name.clone().unwrap_or_default() |
| 1360 | ), |
| 1361 | repo, |
| 1362 | reference: format!("selfhosted/{}/{}", job.id, started), |
| 1363 | kind: Some(ComputeKind::Workflow), |
| 1364 | cpu_seconds: None, |
| 1365 | // A workflow job on a self-hosted runner: no agent ran it and nobody asked an agent for it. |
| 1366 | agent: None, |
| 1367 | asked_by: None, |
| 1368 | reservation_id: None, |
| 1369 | self_hosted: true, |
| 1370 | instance: None, |
| 1371 | }, |
| 1372 | ) |
| 1373 | .await; |
| 1374 | if let Err(error) = recorded { |
| 1375 | worker::console_error!("actions: self-hosted time not recorded for {}: {error}", job.id); |
| 1376 | } |
| 1377 | Ok(()) |
| 1378 | } |
| 1379 | |
| 1380 | /// `runner` and `RUNNER_*` for a job on a self-hosted runner. |
| 1381 | pub async fn runner_context_for(&self, runner_id: &str, variables: &mut Map<String, Value>) -> Result<Value> { |
| 1382 | let mut context = g1t_actions::events::runner_context(); |
| 1383 | let Some(runner) = self.runner_by_id(runner_id).await? else { return Ok(context) }; |
| 1384 | let os = match runner.os.as_str() { |
| 1385 | "macos" => "macOS", |
| 1386 | "windows" => "Windows", |
| 1387 | _ => "Linux", |
| 1388 | }; |
| 1389 | let arch = if runner.arch == "arm64" { "ARM64" } else { "X64" }; |
| 1390 | context["name"] = json!(runner.name); |
| 1391 | context["os"] = json!(os); |
| 1392 | context["arch"] = json!(arch); |
| 1393 | context["environment"] = json!("self-hosted"); |
| 1394 | { |
| 1395 | let vars = variables; |
| 1396 | vars.insert("RUNNER_NAME".into(), json!(runner.name)); |
| 1397 | vars.insert("RUNNER_OS".into(), json!(os)); |
| 1398 | vars.insert("RUNNER_ARCH".into(), json!(arch)); |
| 1399 | vars.insert("RUNNER_ENVIRONMENT".into(), json!("self-hosted")); |
| 1400 | } |
| 1401 | Ok(context) |
| 1402 | } |
| 1403 | |
| 1404 | // --- Agent work, for the runner service ---------------------------------- |
| 1405 | |
| 1406 | /// `runner_route`: the labels g1t's own work in `repo` runs on, when |
| 1407 | /// its workspace (or the repository) sends it to self-hosted runners. |
| 1408 | pub async fn runner_route(&self, a: RouteArgs) -> Result<Option<Vec<String>>> { |
| 1409 | let repo_id = match self.repo_by_path(&a.repo).await? { |
| 1410 | Some(repo) => Some(repo.id), |
| 1411 | None => None, |
| 1412 | }; |
| 1413 | let settings = self.effective_runner_settings(&a.workspace, repo_id.as_deref()).await?; |
| 1414 | Ok(settings.agents_on_self_hosted.then_some(settings.agent_labels)) |
| 1415 | } |
| 1416 | |
| 1417 | async fn repo_by_path(&self, path: &RepoPath) -> Result<Option<Repo>> { |
| 1418 | let Some(actor) = self.workspace_actor(&path.namespace).await? else { return Ok(None) }; |
| 1419 | self.visible_repo(path, &Some(actor)).await |
| 1420 | } |
| 1421 | |
| 1422 | /// `enqueue_task`. |
| 1423 | pub async fn enqueue_task(&self, a: EnqueueTaskArgs) -> Result<Outcome<String>> { |
| 1424 | let Some(sealer) = &self.sealer else { |
| 1425 | return Ok(fail(FailureCode::Conflict, "Agent work cannot be handed to self-hosted runners yet: g1t's key for it is not set.")); |
| 1426 | }; |
| 1427 | let id = new_id("rtk", now_ms()); |
| 1428 | let sealed = sealer.seal(&serde_json::to_string(&a.env)?, &id); |
| 1429 | let repo_id = self.repo_by_path(&a.repo).await?.map(|r| r.id); |
| 1430 | let labels = match model::runner_labels(&a.labels, model::SELF_HOSTED, model::SELF_HOSTED) { |
| 1431 | Ok(labels) => labels, |
| 1432 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 1433 | }; |
| 1434 | let at = now(); |
| 1435 | let inserted = self |
| 1436 | .db |
| 1437 | .prepare( |
| 1438 | "INSERT OR IGNORE INTO runner_tasks (id, sandbox, workspace, repo_id, repo, kind, title, labels, env, timeout_minutes, status, created_at, run_id) |
| 1439 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?) RETURNING id", |
| 1440 | ) |
| 1441 | .bind(&[ |
| 1442 | id.as_str().into(), |
| 1443 | a.sandbox.as_str().into(), |
| 1444 | a.workspace.to_lowercase().into(), |
| 1445 | optional(repo_id.as_deref()), |
| 1446 | format!("{}/{}", a.repo.namespace, a.repo.name).into(), |
| 1447 | a.kind.as_str().into(), |
| 1448 | a.title.chars().take(300).collect::<String>().into(), |
| 1449 | serde_json::to_string(&labels)?.into(), |
| 1450 | sealed.into(), |
| 1451 | a.timeout_minutes.max(1).into(), |
| 1452 | at.into(), |
| 1453 | optional(a.run_id.as_deref().filter(|id| !id.is_empty())), |
| 1454 | ])? |
| 1455 | .first::<Value>(None) |
| 1456 | .await?; |
| 1457 | if inserted.is_none() { |
| 1458 | return Ok(fail(FailureCode::Conflict, "That sandbox already handed work to a self-hosted runner.")); |
| 1459 | } |
| 1460 | Ok(Outcome::Ok(id)) |
| 1461 | } |
| 1462 | |
| 1463 | /// `cancel_task`: the sandbox stopped it. Its runner hears on its next |
| 1464 | /// poll; the sandbox does its own ending. |
| 1465 | pub async fn cancel_task(&self, a: CancelTaskArgs) -> Result<Outcome<bool>> { |
| 1466 | let done = self |
| 1467 | .db |
| 1468 | .prepare( |
| 1469 | "UPDATE runner_tasks SET status = 'completed', exit_code = -1, reason = ?, env = NULL, finished_at = ? |
| 1470 | WHERE sandbox = ? AND status != 'completed' RETURNING id", |
| 1471 | ) |
| 1472 | .bind(&[optional(a.reason.as_deref()), now().into(), a.sandbox.as_str().into()])? |
| 1473 | .first::<Value>(None) |
| 1474 | .await?; |
| 1475 | Ok(Outcome::Ok(done.is_some())) |
| 1476 | } |
| 1477 | |
| 1478 | /// Ends a task and tells the sandbox waiting on it, once. |
| 1479 | async fn end_task(&self, task: &TaskRow, exit_code: i32, reason: &str) -> Result<()> { |
| 1480 | let ended = self |
| 1481 | .db |
| 1482 | .prepare( |
| 1483 | "UPDATE runner_tasks SET status = 'completed', exit_code = ?, reason = ?, env = NULL, finished_at = ? |
| 1484 | WHERE id = ? AND status != 'completed' RETURNING id", |
| 1485 | ) |
| 1486 | .bind(&[exit_code.into(), optional(Some(reason).filter(|r| !r.is_empty())), now().into(), task.id.as_str().into()])? |
| 1487 | .first::<Value>(None) |
| 1488 | .await?; |
| 1489 | if ended.is_none() { |
| 1490 | return Ok(()); |
| 1491 | } |
| 1492 | if let Some(runner_id) = &task.runner_id { |
| 1493 | self.db |
| 1494 | .prepare("UPDATE runners SET work_id = NULL, work_kind = NULL WHERE id = ? AND work_id = ?") |
| 1495 | .bind(&[runner_id.as_str().into(), task.id.as_str().into()])? |
| 1496 | .run() |
| 1497 | .await?; |
| 1498 | } |
| 1499 | let told: Result<Value> = g1t_kit::call( |
| 1500 | &self.runner, |
| 1501 | "task_ended", |
| 1502 | &json!({ |
| 1503 | "sandbox": task.sandbox, |
| 1504 | "exitCode": exit_code, |
| 1505 | "reason": if reason.is_empty() { Value::Null } else { json!(reason) }, |
| 1506 | "runner": task.runner_name, |
| 1507 | }), |
| 1508 | ) |
| 1509 | .await; |
| 1510 | if let Err(error) = told { |
| 1511 | worker::console_error!("actions: the sandbox waiting on task {} was not told: {error}", task.id); |
| 1512 | } |
| 1513 | Ok(()) |
| 1514 | } |
| 1515 | |
| 1516 | /// `runner_activity`: what runs for the workspace now, in g1t's |
| 1517 | /// sandboxes and handed to its own runners. Its owners only, as its |
| 1518 | /// runners are. |
| 1519 | pub async fn runner_activity(&self, a: RunnerActivityArgs) -> Result<Outcome<RunnerActivity>> { |
| 1520 | let owner = RunnersOwner { repo: None, workspace: Some(a.workspace.clone()) }; |
| 1521 | let place = check!(self.runner_place(&a.actor, &owner, false).await?); |
| 1522 | let workspace = place.workspace; |
| 1523 | let cloud_jobs = self |
| 1524 | .db |
| 1525 | .prepare( |
| 1526 | "SELECT jobs.id, jobs.name, jobs.run_id, runs.repo, jobs.started_at FROM jobs JOIN runs ON runs.id = jobs.run_id |
| 1527 | WHERE lower(jobs.namespace) = ? AND jobs.status = 'in_progress' AND jobs.labels IS NULL |
| 1528 | ORDER BY jobs.started_at DESC LIMIT ?", |
| 1529 | ) |
| 1530 | .bind(&[workspace.as_str().into(), ACTIVITY_LIMIT.into()])? |
| 1531 | .all() |
| 1532 | .await? |
| 1533 | .results::<CloudJob>()?; |
| 1534 | let cloud_jobs_queued = self.queued_jobs(&workspace, false).await?; |
| 1535 | let self_hosted_jobs_queued = self.queued_jobs(&workspace, true).await?; |
| 1536 | let handed_over = self |
| 1537 | .db |
| 1538 | .prepare( |
| 1539 | "SELECT id, run_id, kind, title, repo, status, runner_name, created_at, started_at FROM runner_tasks |
| 1540 | WHERE status IN ('queued', 'in_progress') AND workspace = ? |
| 1541 | ORDER BY created_at DESC LIMIT ?", |
| 1542 | ) |
| 1543 | .bind(&[workspace.as_str().into(), ACTIVITY_LIMIT.into()])? |
| 1544 | .all() |
| 1545 | .await? |
| 1546 | .results::<HandedOverTask>()?; |
| 1547 | Ok(Outcome::Ok(RunnerActivity { cloud_jobs, cloud_jobs_queued, handed_over, self_hosted_jobs_queued })) |
| 1548 | } |
| 1549 | |
| 1550 | /// The workspace's workflow jobs queued for its own runners |
| 1551 | /// (`labelled`), or for g1t's sandboxes. |
| 1552 | async fn queued_jobs(&self, workspace: &str, labelled: bool) -> Result<u32> { |
| 1553 | let sql = if labelled { |
| 1554 | "SELECT COUNT(*) AS n FROM jobs WHERE lower(namespace) = ? AND status = 'queued' AND labels IS NOT NULL" |
| 1555 | } else { |
| 1556 | "SELECT COUNT(*) AS n FROM jobs WHERE lower(namespace) = ? AND status = 'queued' AND labels IS NULL" |
| 1557 | }; |
| 1558 | Ok(self.db.prepare(sql).bind(&[workspace.into()])?.first::<Count>(None).await?.map_or(0, |c| c.n)) |
| 1559 | } |
| 1560 | |
| 1561 | /// `stuck_jobs`: jobs of the viewer's workspaces that have waited ten |
| 1562 | /// minutes or more for a self-hosted runner while none that could take |
| 1563 | /// them is online, for Mission control's Needs you. |
| 1564 | pub async fn stuck_jobs(&self, a: StuckJobsArgs) -> Result<Vec<StuckJob>> { |
| 1565 | let Some(viewer) = a.viewer else { return Ok(Vec::new()) }; |
| 1566 | let at = now_ms(); |
| 1567 | let mut out = Vec::new(); |
| 1568 | for membership in viewer.workspaces.iter().take(20) { |
| 1569 | let workspace = membership.slug.to_lowercase(); |
| 1570 | #[derive(Deserialize)] |
| 1571 | struct Waiting { |
| 1572 | id: String, |
| 1573 | name: String, |
| 1574 | run_id: String, |
| 1575 | labels: String, |
| 1576 | queued_at: Option<String>, |
| 1577 | repo: String, |
| 1578 | repo_id: String, |
| 1579 | } |
| 1580 | let waiting = self |
| 1581 | .db |
| 1582 | .prepare( |
| 1583 | "SELECT jobs.id, jobs.name, jobs.run_id, jobs.labels, jobs.queued_at, runs.repo, jobs.repo_id FROM jobs JOIN runs ON runs.id = jobs.run_id |
| 1584 | WHERE jobs.status = 'queued' AND jobs.labels IS NOT NULL AND lower(jobs.namespace) = ? AND jobs.queued_at < ? |
| 1585 | ORDER BY jobs.rowid LIMIT 20", |
| 1586 | ) |
| 1587 | .bind(&[workspace.as_str().into(), rfc3339(at.saturating_sub(10 * 60 * 1000)).into()])? |
| 1588 | .all() |
| 1589 | .await? |
| 1590 | .results::<Waiting>()?; |
| 1591 | if waiting.is_empty() { |
| 1592 | continue; |
| 1593 | } |
| 1594 | let groups = self.group_rows(&workspace).await?; |
| 1595 | let online: Vec<RunnerRow> = self.runner_rows(&workspace).await?.into_iter().filter(|r| r.online(at)).collect(); |
| 1596 | // A repository's visibility is asked only when a host-mode runner |
| 1597 | // is online, since only those pass over a public repository's jobs. |
| 1598 | let any_host = online.iter().any(|runner| runner.isolation.as_deref() == Some("host")); |
| 1599 | let mut looked = Looked::default(); |
| 1600 | for job in waiting { |
| 1601 | let wanted = Wanted::from_stored(&serde_json::from_str::<Vec<String>>(&job.labels).unwrap_or_default()); |
| 1602 | let visibility = if any_host { self.visibility_once(&mut looked, &job.repo_id).await } else { None }; |
| 1603 | let served = online.iter().any(|runner| { |
| 1604 | let group = runner.group_id.as_ref().and_then(|id| groups.iter().find(|g| &g.id == id)); |
| 1605 | wanted.matches(&runner.labels(), group.map(|g| g.name.as_str())) |
| 1606 | && model::isolation_allows(runner.isolation.as_deref(), visibility.as_deref()) |
| 1607 | }); |
| 1608 | if !served { |
| 1609 | out.push(StuckJob { |
| 1610 | id: job.id, |
| 1611 | name: job.name, |
| 1612 | run_id: job.run_id, |
| 1613 | repo: job.repo, |
| 1614 | labels: wanted.describe(), |
| 1615 | queued_at: job.queued_at.unwrap_or_default(), |
| 1616 | }); |
| 1617 | } |
| 1618 | } |
| 1619 | } |
| 1620 | Ok(out) |
| 1621 | } |
| 1622 | |
| 1623 | // --- Every minute ---------------------------------------------------------- |
| 1624 | |
| 1625 | /// Jobs and tasks that waited too long, tasks whose runner went quiet, |
| 1626 | /// runners offline for weeks, and registration tokens long expired. |
| 1627 | pub async fn sweep_runners(&self, at: u64) -> Result<()> { |
| 1628 | let before = |ms: u64| rfc3339(at.saturating_sub(ms)); |
| 1629 | let waited = model::MAX_WAIT_HOURS * 60 * 60 * 1000; |
| 1630 | let stale = self |
| 1631 | .db |
| 1632 | .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NOT NULL AND queued_at < ? LIMIT 50") |
| 1633 | .bind(&[before(waited).into()])? |
| 1634 | .all() |
| 1635 | .await? |
| 1636 | .results::<JobRow>()?; |
| 1637 | for job in stale { |
| 1638 | let stored: Vec<String> = job.labels.as_deref().and_then(|l| serde_json::from_str(l).ok()).unwrap_or_default(); |
| 1639 | let why = format!( |
| 1640 | "No self-hosted runner with labels {} took it within {} hours.", |
| 1641 | Wanted::from_stored(&stored).describe(), |
| 1642 | model::MAX_WAIT_HOURS |
| 1643 | ); |
| 1644 | self.finish_job(&job.id, "failure", Some(&why), None).await?; |
| 1645 | } |
| 1646 | let tasks = self |
| 1647 | .db |
| 1648 | .prepare( |
| 1649 | "SELECT * FROM runner_tasks WHERE (status = 'queued' AND created_at < ?) OR (status = 'in_progress' AND seen_at < ?) LIMIT 50", |
| 1650 | ) |
| 1651 | .bind(&[before(waited).into(), before(SILENT_MS).into()])? |
| 1652 | .all() |
| 1653 | .await? |
| 1654 | .results::<TaskRow>()?; |
| 1655 | for task in tasks { |
| 1656 | let why = if task.status == "queued" { |
| 1657 | format!("No self-hosted runner with labels {} took it within {} hours.", Wanted::of(&serde_json::from_str::<Vec<String>>(&task.labels).unwrap_or_default(), None).describe(), model::MAX_WAIT_HOURS) |
| 1658 | } else { |
| 1659 | format!("The self-hosted runner {} stopped answering.", task.runner_name.clone().unwrap_or_default()) |
| 1660 | }; |
| 1661 | self.end_task(&task, 1, &why).await?; |
| 1662 | } |
| 1663 | let forgotten = self |
| 1664 | .db |
| 1665 | .prepare("SELECT * FROM runners WHERE COALESCE(last_seen_at, created_at) < ? LIMIT 50") |
| 1666 | .bind(&[before(FORGET_OFFLINE_MS).into()])? |
| 1667 | .all() |
| 1668 | .await? |
| 1669 | .results::<RunnerRow>()?; |
| 1670 | for row in forgotten { |
| 1671 | self.forget_runner(&row, "The runner was offline for 14 days.").await?; |
| 1672 | } |
| 1673 | // An ephemeral runner that took its job and then went quiet. |
| 1674 | let spent = self |
| 1675 | .db |
| 1676 | .prepare("SELECT * FROM runners WHERE ephemeral = 1 AND spent = 1 AND work_id IS NULL AND COALESCE(last_seen_at, created_at) < ? LIMIT 50") |
| 1677 | .bind(&[before(60 * 60 * 1000).into()])? |
| 1678 | .all() |
| 1679 | .await? |
| 1680 | .results::<RunnerRow>()?; |
| 1681 | for row in spent { |
| 1682 | self.forget_runner(&row, "The ephemeral runner finished its job.").await?; |
| 1683 | } |
| 1684 | self.db |
| 1685 | .prepare("DELETE FROM runner_registrations WHERE expires_at < ?") |
| 1686 | .bind(&[before(24 * 60 * 60 * 1000).into()])? |
| 1687 | .run() |
| 1688 | .await?; |
| 1689 | Ok(()) |
| 1690 | } |
| 1691 | } |
| 1692 | |
| 1693 | #[derive(Deserialize)] |
| 1694 | struct IdRow { |
| 1695 | id: String, |
| 1696 | } |
| 1697 | |
| 1698 | /// Whether `runner` is online and could take a public repository's job: |
| 1699 | /// not host-mode, and its repository, group and labels let it. |
| 1700 | fn takes_public_job(runner: &RunnerRow, groups: &[GroupRow], repo_id: &str, run_repo: &str, wanted: &Wanted, at: u64) -> bool { |
| 1701 | let group = runner.group_id.as_ref().and_then(|id| groups.iter().find(|g| &g.id == id)); |
| 1702 | runner.online(at) |
| 1703 | && model::isolation_allows(runner.isolation.as_deref(), Some("public")) |
| 1704 | && runner.repo_id.as_deref().is_none_or(|repo| repo == repo_id) |
| 1705 | && (runner.repo_id.is_some() || group.is_none_or(|g| g.allows(run_repo))) |
| 1706 | && wanted.matches(&runner.labels(), group.map(|g| g.name.as_str())) |
| 1707 | } |
| 1708 | |
| 1709 | fn group_view(group: &GroupRow, runners: &[RunnerRow]) -> RunnerGroup { |
| 1710 | RunnerGroup { |
| 1711 | id: group.id.clone(), |
| 1712 | name: group.name.clone(), |
| 1713 | default: group.is_default != 0, |
| 1714 | repositories: group.repositories(), |
| 1715 | runners: runners.iter().filter(|r| r.group_id.as_deref() == Some(group.id.as_str())).count() as u32, |
| 1716 | updated_at: group.updated_at.clone(), |
| 1717 | } |
| 1718 | } |
| 1719 | |
| 1720 | #[cfg(test)] |
| 1721 | mod tests { |
| 1722 | use super::*; |
| 1723 | |
| 1724 | fn group(repositories: &[&str]) -> GroupRow { |
| 1725 | GroupRow { |
| 1726 | id: "rng_1".into(), |
| 1727 | name: "Default".into(), |
| 1728 | is_default: 1, |
| 1729 | repositories: serde_json::to_string(&repositories).unwrap(), |
| 1730 | updated_at: "2026-10-06T00:00:00Z".into(), |
| 1731 | } |
| 1732 | } |
| 1733 | |
| 1734 | #[test] |
| 1735 | fn being_seen_is_written_once_a_minute() { |
| 1736 | let at = 2_000_000_000_000; |
| 1737 | assert!(seen_is_stale(None, at)); |
| 1738 | assert!(!seen_is_stale(Some(&rfc3339(at - 59_000)), at)); |
| 1739 | assert!(seen_is_stale(Some(&rfc3339(at - 61_000)), at)); |
| 1740 | // A runner written down at the last moment is still online until the |
| 1741 | // next write, a poll later. |
| 1742 | assert!(SEEN_RESOLUTION_MS + model::MAX_POLL_WAIT_MS < model::ONLINE_WITHIN_MS); |
| 1743 | assert!(SEEN_RESOLUTION_MS < crate::SILENT_MS); |
| 1744 | } |
| 1745 | |
| 1746 | #[test] |
| 1747 | fn a_group_lets_its_repositories_use_it() { |
| 1748 | assert!(group(&[]).allows("acme/web")); |
| 1749 | assert!(group(&["web", "api"]).allows("acme/WEB")); |
| 1750 | assert!(group(&["web"]).allows("web")); |
| 1751 | assert!(!group(&["web"]).allows("acme/docs")); |
| 1752 | } |
| 1753 | |
| 1754 | #[test] |
| 1755 | fn names_are_checked() { |
| 1756 | assert_eq!(valid_name(" build-01 ").unwrap(), "build-01"); |
| 1757 | assert!(valid_name("").is_err()); |
| 1758 | assert!(valid_name("has space").is_err()); |
| 1759 | assert!(valid_name(&"x".repeat(65)).is_err()); |
| 1760 | assert!(valid_group_name("GPU machines").is_ok()); |
| 1761 | assert!(valid_group_name("").is_err()); |
| 1762 | } |
| 1763 | |
| 1764 | #[test] |
| 1765 | fn only_a_sandboxed_runner_that_could_take_it_serves_a_public_job() { |
| 1766 | let at = 1_000_000_000; |
| 1767 | let runner = |isolation: Option<&str>| RunnerRow { |
| 1768 | id: "rnr_1".into(), |
| 1769 | workspace: "acme".into(), |
| 1770 | repo_id: None, |
| 1771 | repo: None, |
| 1772 | group_id: None, |
| 1773 | name: "a".into(), |
| 1774 | labels: r#"["self-hosted","linux","x64"]"#.into(), |
| 1775 | os: "linux".into(), |
| 1776 | arch: "x64".into(), |
| 1777 | version: String::new(), |
| 1778 | ephemeral: 0, |
| 1779 | credential_hash: String::new(), |
| 1780 | previous_hash: None, |
| 1781 | rotated_at: String::new(), |
| 1782 | work_id: None, |
| 1783 | work_kind: None, |
| 1784 | spent: 0, |
| 1785 | last_seen_at: Some(rfc3339(at)), |
| 1786 | created_at: String::new(), |
| 1787 | created_by: None, |
| 1788 | isolation: isolation.map(str::to_owned), |
| 1789 | }; |
| 1790 | let wanted = Wanted::of(&["self-hosted".to_owned(), "linux".to_owned()], None); |
| 1791 | let takes = |row: &RunnerRow| takes_public_job(row, &[], "rep_1", "acme/web", &wanted, at); |
| 1792 | assert!(!takes(&runner(Some("host")))); |
| 1793 | assert!(takes(&runner(Some("container")))); |
| 1794 | assert!(takes(&runner(Some("nested-container")))); |
| 1795 | assert!(takes(&runner(None)), "an older runner, which refuses for itself"); |
| 1796 | assert!(!takes(&RunnerRow { repo_id: Some("rep_2".into()), ..runner(None) }), "another repository's"); |
| 1797 | assert!(!takes(&RunnerRow { labels: r#"["self-hosted","macos"]"#.into(), ..runner(None) })); |
| 1798 | assert!(!takes(&RunnerRow { last_seen_at: None, ..runner(None) }), "offline"); |
| 1799 | } |
| 1800 | |
| 1801 | #[test] |
| 1802 | fn online_means_seen_in_the_last_ninety_seconds() { |
| 1803 | let row = RunnerRow { |
| 1804 | id: "rnr_1".into(), |
| 1805 | workspace: "acme".into(), |
| 1806 | repo_id: None, |
| 1807 | repo: None, |
| 1808 | group_id: None, |
| 1809 | name: "a".into(), |
| 1810 | labels: "[]".into(), |
| 1811 | os: "linux".into(), |
| 1812 | arch: "x64".into(), |
| 1813 | version: String::new(), |
| 1814 | ephemeral: 0, |
| 1815 | credential_hash: String::new(), |
| 1816 | previous_hash: None, |
| 1817 | rotated_at: String::new(), |
| 1818 | work_id: None, |
| 1819 | work_kind: None, |
| 1820 | spent: 0, |
| 1821 | last_seen_at: Some(rfc3339(1_000_000_000)), |
| 1822 | created_at: String::new(), |
| 1823 | created_by: None, |
| 1824 | isolation: None, |
| 1825 | }; |
| 1826 | assert!(row.online(1_000_000_000 + 60_000)); |
| 1827 | assert!(!row.online(1_000_000_000 + 120_000)); |
| 1828 | assert_eq!(row.workspace, "acme"); |
| 1829 | } |
| 1830 | } |