Skip to content
Sign up

flagon-io/g1t

public
00

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