Skip to content
Sign up

flagon-io/g1t

public
0

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

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