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