Skip to content

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
1,398 linesCodeBlameRaw
1import { Container, type StopParams } from "@cloudflare/containers";
2import { WorkerEntrypoint } from "cloudflare:workers";
3
4import {
5 type BumpArgs,
6 UPDATE_BRANCH_PREFIX,
7 type RunKind,
8 agentsClient,
9 type G1tEvent,
10 type QueueJob,
11 type RepoPath,
12 type Result,
13 type ServiceBinding,
14 type User,
15 type Viewer,
16 type ComputeEntitlements,
17 type ComputeKind,
18 ComputeGate,
19 actualMicros,
20 eventsClient,
21 refusalMessage,
22 sandboxEstimateMicros,
23 billingClient,
24 fail,
25 ok,
26 reposClient,
27 workClient,
28 type InstanceType,
29 STANDARD_INSTANCE,
30 instanceNamed,
31 priceScale,
32 runnerMinutes,
33} from "@g1t/contracts";
34
35import { BUMP_MINUTES, BUMP_TOKEN_TTL_SECONDS, bumpEnv, bumpProblem, bumpSandboxName, systemActor } from "./bump";
36import { BACKUP_MINUTES, backupEnv, backupPace, backupSandboxName } from "./backup";
37import { holdCredentials, revokeCredentials, runCredential } from "./credentials";
38import { cancelTask, enqueueTask, handedOverStep, selfHostedRoute, taskEnv, taskRepo } from "./self-hosted";
39import { answered, capReached, describeError, stoppedShort, tellStopped, withinTimeCap } from "./lifecycle";
40import {
41 ABUSE_EXIT_CODE,
42 ABUSE_HOST,
43 ABUSE_MESSAGE,
44 ALARM_GRACE_SECONDS,
45 type PlanLimits,
46 type RunGuard,
47 abuse,
48 buildGuard,
49 dockerFor,
50 egress,
51 guardFor,
52 harnessEnv,
53 reportRun,
54 SANDBOX_BINDINGS,
55 sandboxNamespace,
56 throughWorker,
57 timeCapMessage,
58 withPlanLimits,
59} from "./guard";
60import { newlyMisdirected } from "./broker";
61import { relay, relayDeps } from "./broker-relay";
62import { forward } from "./forward";
63import { EGRESS_ENV, jobNamespaceOf } from "./egress";
64
65// Outbound interception, which brokered secrets and abuse reports use,
66// needs this exported.
67export { ContainerProxy } from "@cloudflare/containers";
68
69export interface RunnerEnv {
70 SANDBOX: DurableObjectNamespace<AttemptSandbox>;
71 /**
72 * Workflow jobs that ask with `runs-on` for `g1t-2core` (the standard
73 * machine, kept as a name so the label keeps working) or `g1t-4core` (a
74 * larger instance type): the same image, in classes of their own.
75 */
76 SANDBOX_2CORE?: DurableObjectNamespace<Sandbox2Core>;
77 SANDBOX_4CORE?: DurableObjectNamespace<Sandbox4Core>;
78 IDENTITY: ServiceBinding;
79 REPOS: ServiceBinding;
80 WORK: ServiceBinding;
81 BILLING: ServiceBinding;
82 /** GitHub Actions jobs: told when a job's sandbox dies without reporting. */
83 ACTIONS: ServiceBinding;
84 /** Told when a deploy sandbox dies without reporting. */
85 DEPLOYMENTS: ServiceBinding;
86 /** The event bus: `abuse.flagged`, for g1t's staff. */
87 EVENTS?: ServiceBinding;
88 CLOUDFLARE_ACCOUNT_ID: string;
89 /**
90 * `off` stops passing any sandbox's HTTP and HTTPS through the Worker: a
91 * switch for the operator, should it misbehave. Brokered secrets' stand-
92 * ins then go out as they are, unfilled. Anything else passes the
93 * requests of workflow jobs that may be given brokered secrets.
94 */
95 EGRESS?: string;
96 /**
97 * `off` stops sandboxes watching themselves for mining (crates/runner
98 * abuse.rs): a switch for the operator, should it stop real work.
99 * Anything else leaves it on. Miners named in commands are refused
100 * either way.
101 */
102 ABUSE_WATCH?: string;
103 /**
104 * `off` leaves workflow jobs without a Docker Engine of their own
105 * (crates/runner docker/): a switch for the operator. Anything else
106 * gives each job one, started the first time it is used.
107 */
108 DOCKER?: string;
109 /**
110 * Nightly backups (backup.ts): how many queued backups one sweep starts
111 * (`0`: none, backups off here), and how many may run at once.
112 */
113 BACKUPS_PER_SWEEP?: string;
114 BACKUPS_RUNNING?: string;
115}
116
117/**
118 * What a sandbox is doing. Runs of an issue's commands (`checks`) are from
119 * before checks were workflows; a sandbox that was started for work g1t no
120 * longer does is told apart by its kind and only cleaned up.
121 */
122type Run =
123 | { kind: "checks"; runId: string; token: string }
124 /** One combined state of a merge queue, being built and checked. */
125 | { kind: "queue"; entryId: string; token: string }
126 /** Whether a pull request merges cleanly: two commits merged, nothing pushed. */
127 | { kind: "mergecheck"; pullId: string; token: string }
128 /** One job of a GitHub Actions workflow. */
129 | { kind: "actions"; jobId: string; token: string }
130 /** A build of one commit, deployed to g1t.page. */
131 | { kind: "deploy"; deployId: string; token: string }
132 /**
133 * A security update: one package raised in its lockfiles and pushed to
134 * its branch. The security service opens the pull request when it hears
135 * the push, so a failure has no one to tell.
136 */
137 | { kind: "bump"; repo: RepoPath; branch: string }
138 /**
139 * A repository's nightly backup: a bundle cut and sent to the repos
140 * service. g1t's own work, never charged to the workspace.
141 */
142 | { kind: "backup"; jobId: string; token: string };
143/**
144 * Whose sandbox time it is, reported when the sandbox stops, and the
145 * machine it ran on when it was not the standard one.
146 */
147type Meter = {
148 workspace: string;
149 repo: string;
150 description: string;
151 instance?: string | null;
152};
153/**
154 * What billing reserved for a sandbox's work (`ComputeGate.admit`), settled
155 * when it stops at what it cost: its seconds.
156 */
157type Held = { id: string; workspace: string; microsPerSecond: number };
158/**
159 * A sandbox that is not tracked as a run but still has a time cap: a
160 * workflow job, a deploy build or an update, for `minutes` at most.
161 */
162type Build = {
163 kind: "actions" | "deploy" | "bump";
164 minutes: number;
165 /**
166 * A workflow job that may be given brokered secrets: its HTTP and HTTPS
167 * pass the Worker, which fills them in.
168 */
169 brokered?: boolean;
170};
171/** Deploy builds are metered by the Deployments plan, not here. */
172type RunRequest = Run & {
173 envVars: Record<string, string>;
174 meter?: Meter;
175 track?: Track;
176 /** The workspace's plan's caps, applied under its guardrails' (lower of each). */
177 limits?: PlanLimits;
178 reservation?: Held | null;
179 build?: Build;
180 /** Whose sandbox it is, when it has no meter: for `abuse.flagged`. */
181 owner?: { workspace: string; repo: string };
182 /**
183 * The labels of the workspace's self-hosted runners this work goes to
184 * instead of a container (self-hosted.ts). Null or absent: a container.
185 */
186 selfHosted?: string[] | null;
187};
188
189/** What a sandbox is, as billing meters it. */
190function computeKindOf(kind: Run["kind"]): ComputeKind | null {
191 switch (kind) {
192 case "checks":
193 case "mergecheck":
194 // A security update resolves lockfiles, as cheap as a check.
195 case "bump":
196 return "check";
197 case "queue":
198 return "queue";
199 case "actions":
200 return "workflow";
201 case "deploy":
202 return "deploy";
203 // Not metered: a backup is g1t's own cost; nor is a sandbox of a kind
204 // g1t no longer starts, which ran before it stopped starting them.
205 default:
206 return null;
207 }
208}
209
210/** One gate per isolate, so entitlements and prices are kept between calls. */
211let gate: ComputeGate | null = null;
212function gateFor(env: { BILLING: ServiceBinding }): ComputeGate {
213 gate ??= new ComputeGate(env.BILLING);
214 return gate;
215}
216
217/**
218 * What to record the sandbox as, so people can watch it: a run of the
219 * merge queue.
220 */
221type Track = {
222 actor: User;
223 repo: RepoPath;
224 kind: RunKind;
225 number?: number | null;
226 pullId?: string | null;
227 title?: string | null;
228 startedBy?: string | null;
229};
230/** The run a sandbox reports to, kept so it can be closed when it stops. */
231type TrackedRun = { runId: string; token: string };
232
233function meter(repo: RepoPath, description: string): Meter {
234 return { workspace: repo.namespace, repo: `${repo.namespace}/${repo.name}`, description };
235}
236
237/** What the deployments service asks a sandbox to build. */
238type DeployJob = {
239 deployId: string;
240 /** The workspace the project is in, which pays. */
241 workspace?: string;
242 /** What the deployments service reserved for the build, settled when it stops. */
243 reservation?: string | null;
244 /** The price it reserved at, per second. */
245 microsPerSecond?: number | null;
246 /** The plan's longest run, in minutes; the build gets the lower of this and its own. */
247 maxRunMinutes?: number | null;
248 /** Lets the sandbox, and nothing else, report this build. */
249 token: string;
250 /** Whose access reads the commit. */
251 actor: User;
252 /** The repository the commit is in: the pull request's fork, or the repository. */
253 source: RepoPath;
254 /**
255 * The project's repository, whose guardrails the build runs under, and
256 * its id. A preview's `source` is its pull request's working copy, so
257 * the two differ. Older callers send only `source`.
258 */
259 repo?: RepoPath | null;
260 repoId?: string | null;
261 commit: string;
262 /** Where in the repository the project lives; empty for all of it. */
263 rootDir?: string;
264 buildCommand?: string | null;
265 outputDir?: string | null;
266 /** What it is built as (`auto`, `static`, `workers` or `docker`), and its Dockerfile's path when set. */
267 buildAs?: string | null;
268 dockerfile?: string | null;
269 /** The repository's variables for deploy builds. */
270 buildEnv?: Record<string, string>;
271 /** Its secrets for deploy builds: set like variables, and redacted from the log. */
272 buildSecrets?: Record<string, string>;
273};
274
275/** Long enough to install and build; then the read token stops working. */
276const DEPLOY_TOKEN_TTL_SECONDS = 30 * 60;
277
278/** Long enough to clone, build and test a merge queue's state; then the token stops working. */
279const CHECKS_TOKEN_TTL_SECONDS = 45 * 60;
280
281/** Long enough to clone and merge two commits; then the read token stops working. */
282const MERGECHECK_TOKEN_TTL_SECONDS = 10 * 60;
283
284/**
285 * One sandbox, for one piece of work: a workflow job, a build, a state of
286 * the merge queue, a merge check, an update or a backup. The image's entrypoint
287 * is the g1t runner, which does the work and exits; this class only starts
288 * it and cleans up if it dies without reporting.
289 */
290export class AttemptSandbox extends Container<RunnerEnv> {
291 // Past the longest default time cap (90 minutes) and its alarm. A run whose guardrails allow longer (up to 240 minutes) is kept
292 // past it by `onActivityExpired`, so only its own cap ends it. A finished
293 // run's process exits and stops the sandbox well before this.
294 sleepAfter = "100m";
295 // HTTPS to an intercepted host goes through its handler too (guard.ts).
296 interceptHttps = true;
297 static {
298 // Assigned, not declared: a class field would hide the setter that
299 // registers the handler with the containers library. Which one a
300 // sandbox uses is chosen as it starts (`run`).
301 AttemptSandbox.outboundHandlers = { egress, abuse };
302 }
303
304 async run(request: RunRequest): Promise<void> {
305 const { envVars, meter, track, limits, reservation, build, owner, selfHosted, ...run } = request;
306 // What billing reserved is settled however this ends, once.
307 if (reservation) await this.ctx.storage.put("reservation", reservation);
308 let guard: RunGuard | null;
309 try {
310 // A tracked run gets its project's guardrails; no sandbox for one
311 // starts without them. Workflow jobs, deploy builds and updates get
312 // the time cap they were given. The plan's caps apply under them:
313 // the lower of each.
314 guard = track
315 ? withPlanLimits(await guardFor(this.env.WORK, track.repo, track.kind), limits)
316 : build
317 ? withPlanLimits(buildGuard(build.minutes), limits)
318 : null;
319 } catch (error) {
320 // Thrown to the caller, which says why the work did not start; logged
321 // here too, so a sandbox that never started is traceable on its own.
322 console.error("sandbox not started", run.kind, describeError(error));
323 await this.settle(0);
324 throw error;
325 }
326 await this.ctx.storage.put("run", run);
327 // What an earlier attempt in this object left: its reason, its cap
328 // and its cap's alarm, which would otherwise stop this attempt when
329 // the earlier one's time ran out (re-runs use the same object).
330 await this.ctx.storage.delete(["abuse", "stopReason", "remote", "misdirected", "timeCap"]);
331 this.deleteSchedules("timeUp");
332 if (meter) await this.ctx.storage.put("meter", { ...meter, started: Date.now() });
333 await this.ctx.storage.put("started", Date.now());
334 const who = meter ? { workspace: meter.workspace, repo: meter.repo } : owner;
335 if (who) await this.ctx.storage.put("owner", { ...who, kind: track?.kind ?? run.kind });
336 const tracked = track ? await this.openRun(track, envVars, guard) : null;
337 // Its credentials are tied to the run, and revoked when it stops.
338 await holdCredentials(this.env.IDENTITY, this.ctx.storage, envVars, tracked?.runId ?? null);
339 try {
340 const vars = tracked ? { ...envVars, AGENT_RUN: tracked.runId, AGENT_RUN_TOKEN: tracked.token } : envVars;
341 // The workspace's own runner, not a container: the same environment,
342 // handed over as a task.
343 const repo = selfHosted?.length ? taskRepo(track, meter, owner) : null;
344 if (selfHosted?.length && repo) {
345 const harness = guard ? harnessEnv(guard, vars) : {};
346 const minutes = guard?.minutes ?? limits?.minutes ?? 60;
347 await enqueueTask(this.env.ACTIONS, {
348 sandbox: this.ctx.id.toString(),
349 repo,
350 kind: track?.kind ?? run.kind,
351 title: track?.title ?? meter?.description ?? `${run.kind} in ${repo.namespace}/${repo.name}`,
352 labels: selfHosted,
353 env: taskEnv({ ...vars, ...harness }),
354 timeoutMinutes: minutes,
355 runId: tracked?.runId ?? null,
356 });
357 await this.ctx.storage.put("remote", true);
358 if (tracked) await reportRun(this.env.WORK, tracked, { steps: [handedOverStep(selfHosted)] });
359 if (guard) {
360 await this.ctx.storage.put("timeCap", guard.minutes);
361 await this.schedule(guard.minutes * 60 + ALARM_GRACE_SECONDS, "timeUp");
362 }
363 return;
364 }
365 // Every sandbox reaches the internet. A workflow job that may be
366 // given brokered secrets sends its HTTP and HTTPS through `egress`,
367 // which fills them in from the job's stand-ins (asked for with the
368 // job's token) and sends every request on, to any host; any other
369 // sandbox has only `ABUSE_HOST` intercepted, so it can report that it
370 // stopped itself for mining. Both are chosen here, as it starts: hosts
371 // named on the class up front (`outboundByHost` in a static block)
372 // left sandboxes that never picked up their jobs (rolled back
373 // 2026-10-13).
374 const brokering = run.kind === "actions" && throughWorker(build, this.env.EGRESS);
375 this.enableInternet = true;
376 if (brokering && run.kind === "actions") {
377 await this.setOutboundHandler("egress", { job: { id: run.jobId, token: run.token } });
378 } else if (this.env.EGRESS?.trim().toLowerCase() !== "off") {
379 await this.setOutboundByHost(ABUSE_HOST, "abuse").catch((error: unknown) =>
380 console.log("abuse reports not routed", String(error)),
381 );
382 }
383 const harness = guard && !build ? harnessEnv(guard, vars) : {};
384 // Its HTTPS is re-signed as it passes the Worker, so it trusts that
385 // certificate.
386 const trust = brokering ? EGRESS_ENV : {};
387 const watch: Record<string, string> = this.env.ABUSE_WATCH === "off" ? { G1T_ABUSE: "off" } : {};
388 await this.start({ envVars: { ...vars, ...harness, ...trust, ...watch }, enableInternet: true });
389 // A backup has no guardrails, but still a time cap.
390 const cap = guard?.minutes ?? (run.kind === "backup" ? BACKUP_MINUTES : null);
391 if (cap) {
392 await this.ctx.storage.put("timeCap", cap);
393 await this.schedule(cap * 60 + ALARM_GRACE_SECONDS, "timeUp");
394 }
395 } catch (error) {
396 console.error("sandbox not started", run.kind, describeError(error));
397 await revokeCredentials(this.env.IDENTITY, this.ctx.storage);
398 if (tracked) await this.closeRun("failed", `The sandbox could not start: ${String(error)}`);
399 await this.settle(0);
400 throw error;
401 }
402 }
403
404 /** Settles what billing reserved for this sandbox at `micros`, once. */
405 private async settle(micros: number): Promise<void> {
406 const held = await this.ctx.storage.get<Held>("reservation");
407 if (!held) return;
408 await this.ctx.storage.delete("reservation");
409 await gateFor(this.env).settle(held.id, micros);
410 }
411
412 /** Settles the reservation at what the sandbox cost: its seconds at the price billing reserved at. */
413 private async settleStopped(started: number | undefined): Promise<void> {
414 const held = await this.ctx.storage.get<Held>("reservation");
415 if (!held) return;
416 const seconds = started ? Math.max(1, Math.ceil((Date.now() - started) / 1000)) : 0;
417 await this.settle(actualMicros(seconds, held.microsPerSecond));
418 }
419
420 /**
421 * The sandbox stopped itself because it looked like it was mining
422 * (crates/runner abuse.rs), or exited saying so. Stops the run with
423 * `ABUSE_MESSAGE`, tells g1t's staff with `abuse.flagged`, and destroys
424 * the sandbox. Once.
425 */
426 async flagAbuse(verdict: unknown): Promise<void> {
427 if (await this.ctx.storage.get<boolean>("abuse")) return;
428 await this.ctx.storage.put("abuse", true);
429 const tracked = await this.ctx.storage.get<TrackedRun>("agentRun");
430 if (tracked) await reportRun(this.env.WORK, tracked, { halt: "abuse", error: ABUSE_MESSAGE });
431 const owner = await this.ctx.storage.get<{ workspace: string; repo: string | null; kind: string }>("owner");
432 console.log("abuse flagged", owner?.workspace, owner?.repo, owner?.kind, JSON.stringify(verdict));
433 if (this.env.EVENTS && owner) {
434 await eventsClient(this.env.EVENTS)
435 .publish([
436 {
437 type: "abuse.flagged",
438 source: "runner",
439 // Never on a repository's timeline or its webhooks.
440 repoId: null,
441 actor: null,
442 data: {
443 workspace: owner.workspace,
444 repo: owner.repo ?? null,
445 run: tracked?.runId ?? null,
446 kind: owner.kind,
447 sandbox: this.ctx.id.toString(),
448 metrics: verdict && typeof verdict === "object" ? (verdict as Record<string, unknown>) : null,
449 },
450 },
451 ])
452 .catch((error: unknown) => console.log("abuse.flagged not published", String(error)));
453 }
454 await this.destroy().catch((error: unknown) => console.log("sandbox not destroyed for abuse", String(error)));
455 }
456
457 /**
458 * Records the run, which the sandbox then reports its steps to. Never
459 * stops the sandbox from starting: without a record it just goes unseen.
460 */
461 private async openRun(track: Track, envVars: Record<string, string>, guard: RunGuard | null): Promise<TrackedRun | null> {
462 const opened = await agentsClient(this.env.WORK)
463 .openRun({
464 ...track,
465 model: null,
466 sandbox: this.ctx.id.toString(),
467 budgetUsd: guard?.policy.budgetUsd ?? null,
468 timeCapMinutes: guard?.minutes ?? null,
469 })
470 .catch((error: unknown) => ({ ok: false as const, error: { message: String(error) } }));
471 if (!opened.ok) {
472 console.log("run not recorded", track.kind, opened.error.message);
473 return null;
474 }
475 await this.ctx.storage.put("agentRun", opened.value);
476 return opened.value;
477 }
478
479 /**
480 * A workflow job sent a brokered secret's stand-in to a host it is not
481 * for: said once per secret and host, in the job's log.
482 */
483 async noteMisdirected(name: string, host: string): Promise<void> {
484 const run = await this.ctx.storage.get<Run>("run");
485 if (run?.kind !== "actions") return;
486 const noted = newlyMisdirected((await this.ctx.storage.get<string[]>("misdirected")) ?? [], name, host);
487 if (!noted) return;
488 await this.ctx.storage.put("misdirected", noted.seen);
489 try {
490 const response = await this.env.ACTIONS.fetch("https://actions/rpc/job_report", {
491 method: "POST",
492 headers: { "content-type": "application/json" },
493 body: JSON.stringify({ job: run.jobId, token: run.token, report: { kind: "note", text: noted.note } }),
494 });
495 await answered("job_report", response);
496 } catch (error) {
497 console.log("misdirected secret not noted", name, String(error));
498 }
499 }
500
501 /** The run's time cap has passed: stop it, as stopped for time. */
502 async timeUp(): Promise<void> {
503 // It already stopped (nothing to stop), or the run in it now is a later
504 // attempt whose cap has not passed: this alarm is an earlier attempt's,
505 // left by a version of g1t that did not clear it, and the run now has
506 // its own.
507 const started = await this.ctx.storage.get<number>("started");
508 const minutes = (await this.ctx.storage.get<number>("timeCap")) ?? 0;
509 if (!capReached(started, minutes, Date.now())) {
510 if (started) console.log("an earlier attempt's time cap passed; the run now keeps going", this.ctx.id.toString());
511 return;
512 }
513 const tracked = await this.ctx.storage.get<TrackedRun>("agentRun");
514 // A workflow job or a build has no run to halt: it fails saying why.
515 await this.ctx.storage.put("stopReason", timeCapMessage(minutes));
516 if (tracked) await reportRun(this.env.WORK, tracked, { halt: "time", error: timeCapMessage(minutes) });
517 if (await this.ctx.storage.get<boolean>("remote")) {
518 await cancelTask(this.env.ACTIONS, this.ctx.id.toString(), timeCapMessage(minutes));
519 await this.remoteEnded(1, null);
520 return;
521 }
522 await this.destroy().catch((error: unknown) => console.log("sandbox not destroyed at its time cap", String(error)));
523 }
524
525 /**
526 * Stops this sandbox's work: its container, or the task a self-hosted
527 * runner holds, which it hears about on its next poll.
528 */
529 async halt(reason: string | null): Promise<void> {
530 if (await this.ctx.storage.get<boolean>("remote")) {
531 await cancelTask(this.env.ACTIONS, this.ctx.id.toString(), reason);
532 await this.remoteEnded(1, reason);
533 return;
534 }
535 // A container already gone has nothing to stop; one that will not stop
536 // is logged, and its time cap still ends it.
537 await this.destroy().catch((error: unknown) => console.error("sandbox not destroyed", reason, describeError(error)));
538 }
539
540 /**
541 * The library's `sleepAfter` has passed. The runner never fetches its
542 * container, so to the library every sandbox looks idle: a run inside its
543 * time cap keeps going, and the cap's own alarm (`timeUp`) ends it. Only
544 * a sandbox with no cap is stopped for inactivity.
545 */
546 override async onActivityExpired(): Promise<void> {
547 const started = await this.ctx.storage.get<number>("started");
548 const cap = await this.ctx.storage.get<number>("timeCap");
549 if (withinTimeCap(started, cap, Date.now(), ALARM_GRACE_SECONDS)) return;
550 await super.onActivityExpired();
551 }
552
553 /**
554 * A container that crashed or could not be reached, as the library tells
555 * it. Logged at error level with the sandbox, never thrown: the library
556 * ignores what this throws, and the stop that follows is handled by
557 * `onStop` or by `run`, which says why the work did not start.
558 */
559 override onError(error: unknown): void {
560 console.error("sandbox container error", this.ctx.id.toString(), describeError(error));
561 }
562
563 /**
564 * The library's alarm: scheduled callbacks, the container's keep-alive,
565 * and `onStop` once it has stopped. A failure is logged with the retry it
566 * was, then thrown so Cloudflare tries the alarm again.
567 */
568 override async alarm(alarmProps?: AlarmInvocationInfo): Promise<void> {
569 try {
570 await super.alarm(alarmProps);
571 } catch (error) {
572 console.error("sandbox alarm failed", this.ctx.id.toString(), `retry ${alarmProps?.retryCount ?? 0}`, describeError(error));
573 throw error;
574 }
575 }
576
577 /**
578 * A self-hosted runner's task ended (the actions service says so, or g1t
579 * stopped it): everything a container's stop does, once.
580 */
581 async remoteEnded(exitCode: number, reason: string | null): Promise<void> {
582 if (!(await this.ctx.storage.get<boolean>("remote"))) return;
583 await this.ctx.storage.delete("remote");
584 await this.ctx.storage.put("selfHostedEnded", true);
585 if (exitCode !== 0 && reason && !(await this.ctx.storage.get<string>("stopReason"))) {
586 await this.ctx.storage.put("stopReason", reason);
587 }
588 await this.onStop({ exitCode, reason: "exit" } as StopParams);
589 await this.ctx.storage.delete("selfHostedEnded");
590 }
591
592 /**
593 * Ends the run's record, once. Returns the status it ended with:
594 * `stopped` when a person stopped it first.
595 */
596 private async closeRun(outcome: "succeeded" | "failed", error?: string): Promise<string | null> {
597 const tracked = await this.ctx.storage.get<TrackedRun>("agentRun");
598 if (!tracked) return null;
599 await this.ctx.storage.delete("agentRun");
600 const closed = await agentsClient(this.env.WORK)
601 .closeRun(tracked.runId, tracked.token, outcome, error)
602 .catch(() => null);
603 return closed?.ok ? closed.value : null;
604 }
605
606 /** Reports how long the sandbox ran, once, whatever it exited with. */
607 private async meterStop(): Promise<void> {
608 const metered = await this.ctx.storage.get<Meter & { started: number }>("meter");
609 if (!metered) return;
610 await this.ctx.storage.delete("meter");
611 const seconds = Math.max(1, Math.ceil((Date.now() - metered.started) / 1000));
612 const run = await this.ctx.storage.get<Run>("run");
613 // On the workspace's own runner: its minutes, at $0.
614 const selfHosted = (await this.ctx.storage.get<boolean>("selfHostedEnded")) ?? false;
615 const recorded = await billingClient(this.env.BILLING)
616 .recordSandbox({
617 workspace: metered.workspace,
618 seconds,
619 description: selfHosted ? `${metered.description} on a self-hosted runner` : metered.description,
620 repo: metered.repo,
621 reference: `sandbox/${this.ctx.id.toString()}/${metered.started}`,
622 // Whether g1t's open-source pool may pay for it.
623 kind: run ? computeKindOf(run.kind) : null,
624 selfHosted,
625 instance: metered.instance ?? null,
626 })
627 .catch((error: unknown) => ({ ok: false as const, error: { message: String(error) } }));
628 if (!recorded.ok) console.log("sandbox time not recorded", metered.workspace, seconds, recorded.error.message);
629 }
630
631 override async onStop({ exitCode, reason }: StopParams): Promise<void> {
632 await revokeCredentials(this.env.IDENTITY, this.ctx.storage);
633 const started = await this.ctx.storage.get<number>("started");
634 await this.meterStop();
635 // It stopped itself for mining, and could not say so before it went.
636 if (exitCode === ABUSE_EXIT_CODE && !(await this.ctx.storage.get<boolean>("abuse"))) {
637 await this.flagAbuse(null);
638 }
639 const flagged = (await this.ctx.storage.get<boolean>("abuse")) ?? false;
640 // Why it stopped, when g1t stopped it: said in place of a plain failure.
641 const why = flagged ? ABUSE_MESSAGE : ((await this.ctx.storage.get<string>("stopReason")) ?? null);
642 // Stopped by g1t, it failed whatever exit code the library passed on.
643 const short = stoppedShort(exitCode, why);
644 await this.closeRun(
645 short ? "failed" : "succeeded",
646 short ? (why ?? `The sandbox exited with ${exitCode}.`) : undefined,
647 );
648 await this.settleStopped(started);
649 await this.ctx.storage.delete("started");
650 if (!short) return;
651 const run = await this.ctx.storage.get<Run>("run");
652 console.log("sandbox stopped", run?.kind, "exit", exitCode, reason);
653 if (!run) return;
654 // Never thrown: this runs in the sandbox's alarm, which a throw would
655 // fail, retry and count as an error, running all of the above again.
656 // What could not be told is logged, and the sweep catches it up.
657 await tellStopped(run.kind, () => this.reportStopped(run, why, exitCode));
658 }
659
660 /**
661 * Tells whoever is waiting on the sandbox's work that it stopped without
662 * finishing it. Each is refused harmlessly when the sandbox reported its
663 * end before it stopped. Throws when the service could not be reached.
664 */
665 private async reportStopped(run: Run, why: string | null, exitCode: number): Promise<void> {
666 if (run.kind === "actions") {
667 // Refused harmlessly if the job reported its end before it stopped.
668 const response = await this.env.ACTIONS.fetch("https://actions/rpc/job_report", {
669 method: "POST",
670 headers: { "content-type": "application/json" },
671 body: JSON.stringify({
672 job: run.jobId,
673 token: run.token,
674 report: { kind: "done", conclusion: "failure", reason: why ?? "The runner stopped before the job finished." },
675 }),
676 });
677 await answered("job_report", response);
678 return;
679 }
680 // Nothing was pushed, so no pull request opens; why is in its log.
681 if (run.kind === "bump") return;
682 if (run.kind === "backup") {
683 // Refused harmlessly if the sandbox reported before it stopped; the
684 // job is otherwise tried again later tonight.
685 await reposClient(this.env.REPOS).failBackup(run.jobId, run.token, why ?? `The sandbox exited with ${exitCode}.`);
686 return;
687 }
688 if (run.kind === "deploy") {
689 // Refused harmlessly if the build reported its end before it stopped.
690 const response = await this.env.DEPLOYMENTS.fetch(`https://deployments/jobs/${run.deployId}/fail`, {
691 method: "POST",
692 headers: { "content-type": "application/json" },
693 body: JSON.stringify({ token: run.token, message: why ?? "The build stopped before it finished." }),
694 });
695 await answered("deploy fail", response);
696 return;
697 }
698 const work = workClient(this.env.WORK);
699 if (run.kind === "checks") {
700 // Refused harmlessly if the run did report before it stopped.
701 await work.reportChecks(run.runId, run.token, {
702 error: why ?? "The sandbox stopped before the checks finished.",
703 });
704 return;
705 }
706 if (run.kind === "queue") {
707 // Refused harmlessly if the state was reported before it stopped.
708 await work.failQueue(run.entryId, run.token, why ?? "The sandbox stopped before the state was checked.");
709 return;
710 }
711 if (run.kind === "mergecheck") {
712 // Refused harmlessly if the probe reported before it stopped.
713 await work.failMergecheck(run.pullId, run.token, why ?? "The sandbox stopped before the merge check finished.");
714 return;
715 }
716 // A sandbox of a kind g1t no longer starts, which ran from before: it
717 // has nobody left to tell.
718 }
719}
720
721/**
722 * What the compute gate decided for one start: go, with what billing
723 * reserved and the plan's caps; or not, refused with what to tell people.
724 */
725type Granted = {
726 ok: true;
727 held: Held | null;
728 limits: PlanLimits;
729 /** The labels of the self-hosted runners it goes to; null for a sandbox. */
730 route: string[] | null;
731};
732type Admitted = Granted | { ok: false; code: string; message: string };
733
734/**
735 * The guardrails' default time cap of each kind of run, for estimating what
736 * it may cost before it starts; the sandbox applies the project's own.
737 */
738const DEFAULT_MINUTES: Record<"queue" | "mergecheck", number> = {
739 queue: 45,
740 mergecheck: 10,
741};
742
743/** A plan's caps on one run, for the sandbox to apply under its guardrails'. */
744function limitsOf(ent: ComputeEntitlements | null): PlanLimits {
745 return {
746 minutes: ent && ent.maxRunMinutes > 0 ? ent.maxRunMinutes : null,
747 budgetUsd: ent && ent.runCapMicros > 0 ? ent.runCapMicros / 1_000_000 : null,
748 };
749}
750
751/** The shorter of a kind's time cap and the plan's, for an estimate. */
752function estimateMinutes(minutes: number, ent: ComputeEntitlements | null): number {
753 return ent && ent.maxRunMinutes > 0 ? Math.min(minutes, ent.maxRunMinutes) : minutes;
754}
755
756/** A start the gate did not let through, as a result for whoever asked. */
757function notAdmitted(admitted: Exclude<Admitted, Granted>): Result<never> {
758 return fail("payment_required", admitted.message);
759}
760
761/**
762 * Larger sandboxes for workflow jobs that ask for one in `runs-on`: the
763 * same image and behaviour on a larger Containers instance type, each a
764 * class of its own (wrangler.jsonc). Outbound handlers are registered by
765 * class, so each registers its own.
766 */
767export class Sandbox2Core extends AttemptSandbox {
768 static {
769 Sandbox2Core.outboundHandlers = { egress, abuse };
770 }
771}
772export class Sandbox4Core extends AttemptSandbox {
773 static {
774 Sandbox4Core.outboundHandlers = { egress, abuse };
775 }
776}
777
778/**
779 * What the actions service sends to start a job (`StartJobArgs`). Callers
780 * from before may also send `workflow`, `environment` and `trusted`, for
781 * workflow-only domains, which are gone: they are ignored.
782 */
783type ActionsJobArgs = {
784 job: string;
785 token: string;
786 repo: RepoPath;
787 timeoutMinutes: number;
788 /** The machine its `runs-on` asked for, by label; absent, the standard one. */
789 instance?: string | null;
790 /** Whether it may be given brokered secrets, which its requests need filled in. */
791 brokered?: boolean;
792};
793
794export default class RunnerService extends WorkerEntrypoint<RunnerEnv> {
795 /**
796 * The JSON protocol the Rust services speak: `POST /rpc/<method>` with the
797 * arguments as the body. Only bound services can.
798 */
799 async fetch(request: Request): Promise<Response> {
800 const { pathname } = new URL(request.url);
801 if (request.method === "POST" && pathname === "/rpc/start_actions_job") {
802 return Response.json(await this.startActionsJob((await request.json()) as ActionsJobArgs));
803 }
804 if (request.method === "POST" && pathname === "/rpc/stop_actions_job") {
805 const args = (await request.json()) as { job: string };
806 // Whichever machine it asked for: the job's object in every namespace.
807 await Promise.all(
808 Object.keys(SANDBOX_BINDINGS).map((className) => {
809 const namespace = sandboxNamespace(this.env, className) as unknown as DurableObjectNamespace<AttemptSandbox>;
810 return namespace
811 .get(namespace.idFromName(`actions:${args.job}`))
812 .destroy()
813 .catch(() => undefined);
814 }),
815 );
816 return Response.json(ok(true));
817 }
818 // A self-hosted runner's task ended: the sandbox that handed it over
819 // does what it does when a container stops.
820 if (request.method === "POST" && pathname === "/rpc/task_ended") {
821 const args = (await request.json()) as { sandbox: string; exitCode: number; reason?: string | null };
822 const sandbox = this.env.SANDBOX.get(this.env.SANDBOX.idFromString(args.sandbox));
823 await sandbox.remoteEnded(args.exitCode, args.reason ?? null);
824 return Response.json(ok(true));
825 }
826 // A self-hosted job's request to a brokered secret's host, which the
827 // API passes on as it came (broker-relay.ts).
828 if (request.method === "POST" && pathname === "/rpc/broker") {
829 try {
830 return await relay(request, relayDeps(this.env.ACTIONS, (upstream) => forward(upstream)));
831 } catch (error) {
832 console.log("broker relay failed", String(error));
833 return Response.json({ error: { code: "unavailable", message: "g1t could not relay the request." } }, { status: 503 });
834 }
835 }
836 if (request.method === "POST" && pathname === "/rpc/bump") {
837 return Response.json(await this.startBump(await request.json()));
838 }
839 if (request.method === "POST" && pathname === "/rpc/start_deploy") {
840 return Response.json(await this.startDeploy((await request.json()) as DeployJob));
841 }
842 return new Response("Not found\n", { status: 404 });
843 }
844
845 // ---- The compute gate (@g1t/contracts compute.ts) -------------------------
846
847 /** Whether `repo` is public: what g1t's open-source pool can pay for. */
848 private async isPublic(repo: RepoPath): Promise<boolean> {
849 const found = await reposClient(this.env.REPOS)
850 .get(repo, null)
851 .catch(() => null);
852 return Boolean(found?.ok && !found.value.isPrivate);
853 }
854
855 /**
856 * Whether a sandbox (the merge queue, a
857 * merge check, a workflow job) may start in `repo`, with what it may cost
858 * for `minutes` reserved. Public repositories' checks, workflows and
859 * queue can be paid by the open-source pool. Never throws.
860 */
861 private async admitSandbox(
862 kind: ComputeKind,
863 repo: RepoPath,
864 minutes: number,
865 instance: InstanceType = STANDARD_INSTANCE,
866 { selfHosted = true }: { selfHosted?: boolean } = {},
867 ): Promise<Admitted> {
868 const workspace = repo.namespace.toLowerCase();
869 const compute = gateFor(this.env);
870 const ent = await compute.entitlements(workspace);
871 if (ent?.paused) return { ok: false, code: "paused", message: refusalMessage("paused", workspace, kind, ent.paused) };
872 // Checks and the merge queue go to the workspace's own runners when it
873 // says so, and cost nothing there. Workflow jobs choose with `runs-on`.
874 const route = selfHosted && (kind === "check" || kind === "queue") ? await selfHostedRoute(this.env.ACTIONS, repo) : null;
875 if (route) return { ok: true, held: null, limits: limitsOf(ent), route };
876 const [standardMicros, isPublic] = await Promise.all([compute.microsPerSecond(), this.isPublic(repo)]);
877 // Reserved for its time limit at the larger of what it may cost (the
878 // sandbox price, scaled to its machine) and its minutes at the
879 // machine's rate, which is what it is charged.
880 const limit = estimateMinutes(minutes, ent);
881 const microsPerSecond = standardMicros * priceScale(instance);
882 const estimateMicros = Math.max(
883 sandboxEstimateMicros(limit, microsPerSecond),
884 runnerMinutes(limit * 60) * instance.microsPerMinute,
885 );
886 const admission = await compute.admit({ workspace, repo, public: isPublic, kind, estimateMicros }, ent);
887 if (!admission.ok) return { ok: false, code: admission.code, message: admission.message };
888 return {
889 ok: true,
890 held: admission.reservation ? { id: admission.reservation.id, workspace, microsPerSecond } : null,
891 limits: limitsOf(ent),
892 route: null,
893 };
894 }
895
896 /** Gives back what was reserved for a start that never reached its sandbox. */
897 private async release(held: Held | null): Promise<void> {
898 if (held) await gateFor(this.env).settle(held.id, 0);
899 }
900
901 /**
902 * Runs `start`, giving back what was reserved if it fails. A sandbox that
903 * could not start has given it back already; settling twice at nothing
904 * is harmless.
905 */
906 private async holding<T>(granted: Granted, start: () => Promise<T>): Promise<T> {
907 try {
908 return await start();
909 } catch (error) {
910 await this.release(granted.held);
911 throw error;
912 }
913 }
914
915 /**
916 * Starts one job of a GitHub Actions workflow in a sandbox of its own.
917 * The sandbox fetches the job, its contexts and its secrets with the
918 * job's token, and reports back to the actions service through the API.
919 * Jobs run on g1t's machines, so only for workspaces that may use them.
920 */
921 private async startActionsJob(args: ActionsJobArgs): Promise<Result<true>> {
922 // The machine its `runs-on` asked for; the standard one otherwise.
923 const instance = instanceNamed(args.instance);
924 // Workflow jobs run on g1t's machines: only as the workspace's plan
925 // allows, or on a public repository, from the open-source pool.
926 const admitted = await this.admitSandbox("workflow", args.repo, args.timeoutMinutes, instance);
927 if (!admitted.ok) return fail("payment_required", `Not started: ${admitted.message}`);
928 const namespace = this.jobNamespace(instance);
929 if (!namespace) {
930 await this.release(admitted.held);
931 return fail("invalid", `Not started: ${instance.label} machines are not available here.`);
932 }
933 const sandbox = namespace.get(namespace.idFromName(`actions:${args.job}`));
934 const on = instance === STANDARD_INSTANCE ? "" : ` on ${instance.label}`;
935 try {
936 await sandbox.run({
937 kind: "actions",
938 jobId: args.job,
939 token: args.token,
940 reservation: admitted.held,
941 limits: admitted.limits,
942 // The job's own time limit, and whether its requests pass the
943 // Worker for its brokered secrets.
944 build: {
945 kind: "actions",
946 minutes: Math.max(1, args.timeoutMinutes),
947 brokered: args.brokered === true,
948 },
949 meter: {
950 ...meter(args.repo, `A workflow job in ${args.repo.namespace}/${args.repo.name}${on}`),
951 instance: instance === STANDARD_INSTANCE ? null : instance.label,
952 },
953 envVars: {
954 MODE: "actions",
955 G1T_API: "https://api.g1t.sh",
956 ACTIONS_JOB: args.job,
957 ACTIONS_TOKEN: args.token,
958 // Docker of the job's own, inside its sandbox (crates/runner docker/).
959 G1T_DOCKER: dockerFor(this.env.DOCKER),
960 },
961 });
962 } catch (error) {
963 // A sandbox that could not start, or stopped at once: the job fails
964 // with why, rather than waiting to be noticed.
965 return {
966 ok: false,
967 error: { code: "conflict", message: `The runner could not start the job: ${String(error).replace(/^Error: /, "")}` },
968 };
969 }
970 // `true`, not null: an outcome needs a value.
971 return ok(true);
972 }
973
974 /** The sandboxes of a label: the class the catalogue names for it, by its binding. */
975 private jobNamespace(instance: InstanceType): DurableObjectNamespace<AttemptSandbox> | null {
976 return jobNamespaceOf(this.env, instance) as DurableObjectNamespace<AttemptSandbox> | null;
977 }
978
979 /**
980 * Builds one commit in a sandbox of its own and deploys it to g1t.page.
981 * Asked by the deployments service, which has already checked that the
982 * workspace pays for Deployments; that plan is what
983 * lets a build use g1t's machines.
984 */
985 private async startDeploy(job: DeployJob): Promise<Result<true>> {
986 // To read the commit, which may be private, as whoever pushed it.
987 const token = await runCredential(this.env.IDENTITY, {
988 onBehalfOf: job.actor,
989 repo: job.source,
990 kind: "deploy",
991 use: "runner",
992 read: [job.source],
993 ttlSeconds: DEPLOY_TOKEN_TTL_SECONDS,
994 });
995 const sandbox = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(`deploy:${job.deployId}`));
996 const workspace = (job.workspace ?? job.source.namespace).toLowerCase();
997 // The project the build is for, and who it is charged to.
998 const project = job.repo ?? job.source;
999 try {
1000 await sandbox.run({
1001 kind: "deploy",
1002 deployId: job.deployId,
1003 token: job.token,
1004 reservation: job.reservation
1005 ? { id: job.reservation, workspace, microsPerSecond: job.microsPerSecond ?? 0 }
1006 : null,
1007 limits: { minutes: job.maxRunMinutes ?? null },
1008 // For as long as its read token lasts.
1009 build: { kind: "deploy", minutes: DEPLOY_TOKEN_TTL_SECONDS / 60 },
1010 owner: { workspace, repo: `${project.namespace}/${project.name}` },
1011 envVars: {
1012 MODE: "deploy",
1013 G1T_API: "https://api.g1t.sh",
1014 DEPLOY_ID: job.deployId,
1015 DEPLOY_TOKEN: job.token,
1016 G1T_USER: job.actor.username,
1017 G1T_TOKEN: token,
1018 GIT_REMOTE: `https://g1t.sh/${job.source.namespace}/${job.source.name}.git`,
1019 GIT_COMMIT: job.commit,
1020 ROOT_DIR: job.rootDir ?? "",
1021 BUILD_COMMAND: job.buildCommand ?? "",
1022 OUTPUT_DIR: job.outputDir ?? "",
1023 BUILD_AS: job.buildAs ?? "auto",
1024 DOCKERFILE: job.dockerfile ?? "",
1025 // A container build's own Docker Engine (crates/runner docker/),
1026 // started only if the project is built as one.
1027 G1T_DOCKER: dockerFor(this.env.DOCKER),
1028 BUILD_ENV: JSON.stringify(job.buildEnv ?? {}),
1029 BUILD_SECRETS: JSON.stringify(job.buildSecrets ?? {}),
1030 },
1031 });
1032 } catch (error) {
1033 return {
1034 ok: false,
1035 error: { code: "conflict", message: `The runner could not start the build: ${String(error).replace(/^Error: /, "")}` },
1036 };
1037 }
1038 return ok(true);
1039 }
1040
1041 /**
1042 * Makes a security update in a sandbox of its own (crates/runner
1043 * bump.rs): raises one package to a fixed version in the lockfiles
1044 * named, commits that as g1t and pushes it to its `g1t/security/…`
1045 * branch. A version update (`kind: "version"`) raises one or more
1046 * packages the same way, to the branch its dependency update file names,
1047 * which is never the default one. Asked by the security service, which
1048 * opens the pull request when it hears the push; nothing here opens one.
1049 * Admitted, reserved and metered like checks, always in g1t's sandbox (a
1050 * self-hosted runner may not know the mode), under the project's network
1051 * list plus the package registries. Returns whether the sandbox started.
1052 */
1053 private async startBump(input: unknown): Promise<Result<boolean>> {
1054 const problem = bumpProblem(input, UPDATE_BRANCH_PREFIX);
1055 if (problem) return fail("invalid", problem);
1056 const args = input as BumpArgs;
1057 const repo = args.repo;
1058 const what = args.kind === "version" ? "version update" : "security update";
1059 const actor = systemActor(repo.namespace);
1060 const closed = await this.closedRepo(actor, repo);
1061 if (closed) return closed;
1062 const admitted = await this.admitSandbox("check", repo, BUMP_MINUTES, STANDARD_INSTANCE, { selfHosted: false });
1063 if (!admitted.ok) return notAdmitted(admitted);
1064 try {
1065 const defaultBranch = await this.defaultBranch(repo, actor);
1066 // From its `target-branch`, which its pull request merges into, or
1067 // the default branch.
1068 const base = args.base ?? defaultBranch;
1069 // A version update names its own branch, which is never the one it
1070 // starts from, nor the default one.
1071 if (args.kind === "version" && (args.branch === defaultBranch || args.branch === base)) {
1072 await this.release(admitted.held);
1073 return fail("invalid", `A version update cannot push to ${args.branch}, the branch it starts from.`);
1074 }
1075 // As g1t, for the workspace: reads the repository and pushes this
1076 // branch only, with no API operations.
1077 const token = await runCredential(this.env.IDENTITY, {
1078 onBehalfOf: actor,
1079 repo,
1080 kind: "bump",
1081 use: "runner",
1082 read: [repo],
1083 push: [{ repo, branch: args.branch }],
1084 ttlSeconds: BUMP_TOKEN_TTL_SECONDS,
1085 });
1086 const sandbox = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(bumpSandboxName(args)));
1087 await sandbox.run({
1088 kind: "bump",
1089 repo,
1090 branch: args.branch,
1091 reservation: admitted.held,
1092 limits: admitted.limits,
1093 build: { kind: "bump", minutes: BUMP_MINUTES },
1094 meter: meter(repo, `${args.kind === "version" ? "Version" : "Security"} update in ${repo.namespace}/${repo.name}`),
1095 envVars: bumpEnv(args, base, token),
1096 });
1097 } catch (error) {
1098 await this.release(admitted.held);
1099 return fail("conflict", `The runner could not start the ${what}: ${String(error).replace(/^Error: /, "")}`);
1100 }
1101 return ok(true);
1102 }
1103
1104 /**
1105 * Events from the bus. Each one that could change what a pull request
1106 * needs next moves it along: the work service decides whether it can
1107 * merge by the repository's settings, and merges it if so. Merge checks
1108 * and the merge queue's states start here too.
1109 */
1110 async queue(batch: MessageBatch<G1tEvent>): Promise<void> {
1111 for (const message of batch.messages) {
1112 const event = message.body;
1113 switch (event.type) {
1114 case "pull.opened":
1115 case "pull.ready":
1116 case "pull.updated":
1117 case "checks.completed":
1118 case "review.completed":
1119 case "pull.mergeability":
1120 await this.advance(event.data.pullId);
1121 break;
1122 // Its head or its target moved and both changed the same files:
1123 // find out whether it still merges cleanly.
1124 case "pull.mergecheck":
1125 await this.startMergecheck(event.data.pullId);
1126 break;
1127 // Something joined, left or landed: test the next batch if none is.
1128 case "queue.changed":
1129 await this.buildQueue(event.data.repoId);
1130 break;
1131 // A person approved or asked for changes, which may let it merge.
1132 case "comment.created":
1133 if (event.data.pullId && event.data.verdict) await this.advance(event.data.pullId);
1134 break;
1135 // The branch the others would land on has moved.
1136 case "pull.merged":
1137 await this.advanceAll(event.data.repoId);
1138 break;
1139 // Read-only or gone: the runs there stop.
1140 case "repo.archived":
1141 case "repo.deleted":
1142 await this.stopRunsIn(event.data.repoId);
1143 break;
1144 }
1145 message.ack();
1146 }
1147 }
1148
1149 /** A sweep, for steps whose trigger was missed or whose sandbox died, and the nightly backups. */
1150 async scheduled(): Promise<void> {
1151 await this.advanceAll();
1152 await this.startBackups().catch((error: unknown) => console.log("backups not started", String(error)));
1153 }
1154
1155 /**
1156 * Starts a few of the nightly backups the repos service queued, each in
1157 * a sandbox of its own that holds only its job's token: the sandbox asks
1158 * for a read-only git credential itself, when it is ready to clone. No
1159 * plan is asked and nothing is metered: backups are g1t's own work.
1160 */
1161 private async startBackups(): Promise<void> {
1162 const pace = backupPace(this.env.BACKUPS_PER_SWEEP, this.env.BACKUPS_RUNNING);
1163 if (pace.perSweep === 0) return;
1164 const repos = reposClient(this.env.REPOS);
1165 for (const claim of await repos.claimBackups(pace.perSweep, pace.running)) {
1166 try {
1167 const sandbox = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(backupSandboxName(claim)));
1168 await sandbox.run({
1169 kind: "backup",
1170 jobId: claim.jobId,
1171 token: claim.token,
1172 // For `abuse.flagged`: whose repository it was.
1173 owner: { workspace: claim.path.namespace, repo: `${claim.path.namespace}/${claim.path.name}` },
1174 envVars: backupEnv(claim, "https://api.g1t.sh"),
1175 });
1176 } catch (error) {
1177 await repos.failBackup(claim.jobId, claim.token, `The sandbox could not start: ${String(error)}`).catch(() => null);
1178 }
1179 }
1180 }
1181
1182 private async advanceAll(repoId?: string): Promise<void> {
1183 const pulls = await workClient(this.env.WORK).managedPulls(repoId);
1184 for (const pullId of pulls) await this.advance(pullId);
1185 }
1186
1187 /**
1188 * Merges a pull request by the repository's settings when it is ready:
1189 * the work service decides and merges, and claims the step, so calling
1190 * this twice merges nothing twice.
1191 */
1192 private async advance(pullId: string): Promise<void> {
1193 const next = await workClient(this.env.WORK).advance(pullId);
1194 // g1t no longer reviews, revises or catches up pull requests itself.
1195 if (next.action !== "none") console.log("lifecycle step not taken", pullId, next.action);
1196 }
1197
1198 /**
1199 * Starts the next batch of a repository's merge queue, if it has one
1200 * ready: a sandbox per entry, all at once, each building the default
1201 * branch with that entry and everything ahead of it.
1202 */
1203 private async buildQueue(repoId: string): Promise<void> {
1204 const work = workClient(this.env.WORK);
1205 const jobs = await work.queueBuild(repoId);
1206 // Merge queue sandboxes, like any other, only as the workspace's plan
1207 // allows: refused states fail at once, saying why. A state whose
1208 // sandbox could not start fails at once too, rather than holding the
1209 // queue until it times out.
1210 await Promise.all(
1211 jobs.map(async (job) => {
1212 const admitted = await this.admitSandbox("queue", job.repo, DEFAULT_MINUTES.queue);
1213 if (!admitted.ok) {
1214 await work.failQueue(job.entryId, job.token, `Not started: ${admitted.message}`);
1215 return;
1216 }
1217 await this.holding(admitted, () => this.startQueueRun(job, admitted)).catch((error: unknown) =>
1218 work.failQueue(job.entryId, job.token, `Its sandbox could not start: ${String(error)}`),
1219 );
1220 }),
1221 );
1222 }
1223
1224 private async startQueueRun(job: QueueJob, granted: Granted): Promise<void> {
1225 // To read the changes and push the tested state, as a member.
1226 // Reads each queued change; pushes only the queue's own branch.
1227 const token = await runCredential(this.env.IDENTITY, {
1228 onBehalfOf: job.actor,
1229 repo: job.repo,
1230 kind: "queue",
1231 use: "runner",
1232 number: job.stack.at(-1)?.number ?? null,
1233 read: job.stack.map((item) => item.source),
1234 push: [{ repo: job.repo, branch: job.branch }],
1235 ttlSeconds: CHECKS_TOKEN_TTL_SECONDS,
1236 });
1237 const remote = (path: RepoPath) => `https://g1t.sh/${path.namespace}/${path.name}.git`;
1238 const sandbox = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(`queue-${job.entryId}-${job.baseCommit}`));
1239 await sandbox.run({
1240 kind: "queue",
1241 entryId: job.entryId,
1242 token: job.token,
1243 reservation: granted.held,
1244 limits: granted.limits,
1245 selfHosted: granted.route,
1246 track: {
1247 actor: job.actor,
1248 repo: job.repo,
1249 kind: "queue",
1250 number: job.stack.at(-1)?.number ?? null,
1251 title: `Merge queue: ${job.stack.map((item) => `#${item.number}`).join(" + ")}`,
1252 },
1253 meter: meter(job.repo, `Merge queue on ${job.repo.namespace}/${job.repo.name}`),
1254 envVars: {
1255 MODE: "queue",
1256 G1T_API: "https://api.g1t.sh",
1257 QUEUE_ENTRY: job.entryId,
1258 QUEUE_TOKEN: job.token,
1259 G1T_USER: job.actor.username,
1260 G1T_TOKEN: token,
1261 BASE_REMOTE: remote(job.repo),
1262 BASE_COMMIT: job.baseCommit,
1263 QUEUE_BRANCH: job.branch,
1264 STACK: JSON.stringify(
1265 job.stack.map((item) => ({
1266 number: item.number,
1267 title: item.title,
1268 remote: remote(item.source),
1269 branch: item.branch,
1270 commit: item.commit,
1271 })),
1272 ),
1273 CHECKS: JSON.stringify(job.checks),
1274 CONTRACT_CHECKS: JSON.stringify(job.contractChecks),
1275 },
1276 });
1277 }
1278
1279 /**
1280 * Merges a pull request's head into its target in a sandbox of its own,
1281 * pushing nothing, to find the files that conflict.
1282 * The work service decides when one is needed and how many may run.
1283 */
1284 private async startMergecheck(pullId: string): Promise<void> {
1285 const work = workClient(this.env.WORK);
1286 const started = await work.startMergecheck(pullId);
1287 if (!started.ok) return;
1288 const job = started.value;
1289 let granted: Granted | null = null;
1290 try {
1291 // Like any sandbox, only as the workspace's plan allows.
1292 const admitted = await this.admitSandbox("check", job.repo, DEFAULT_MINUTES.mergecheck);
1293 if (!admitted.ok) throw new Error(`Not started: ${admitted.message}`);
1294 granted = admitted;
1295 // To read the change, which may be private, as whoever opened it.
1296 const token = await runCredential(this.env.IDENTITY, {
1297 onBehalfOf: job.author,
1298 repo: job.repo,
1299 kind: "mergecheck",
1300 use: "runner",
1301 number: job.number,
1302 read: [job.repo, job.source],
1303 ttlSeconds: MERGECHECK_TOKEN_TTL_SECONDS,
1304 });
1305 const remote = (path: RepoPath) => `https://g1t.sh/${path.namespace}/${path.name}.git`;
1306 // One sandbox per pair of commits: asking twice starts nothing twice.
1307 const sandbox = this.env.SANDBOX.get(
1308 this.env.SANDBOX.idFromName(`mergecheck-${job.pullId}-${job.head}-${job.base}`),
1309 );
1310 await sandbox.run({
1311 kind: "mergecheck",
1312 pullId: job.pullId,
1313 token: job.token,
1314 reservation: granted.held,
1315 limits: granted.limits,
1316 selfHosted: granted.route,
1317 meter: meter(job.repo, `Merge check of ${job.repo.namespace}/${job.repo.name}#${job.number}`),
1318 envVars: {
1319 MODE: "mergecheck",
1320 G1T_API: "https://api.g1t.sh",
1321 MERGECHECK_PULL: job.pullId,
1322 MERGECHECK_TOKEN: job.token,
1323 G1T_USER: job.author.username,
1324 G1T_TOKEN: token,
1325 BASE_REMOTE: remote(job.repo),
1326 BASE_COMMIT: job.base,
1327 HEAD_REMOTE: remote(job.source),
1328 HEAD_BRANCH: job.branch,
1329 HEAD_COMMIT: job.head,
1330 },
1331 });
1332 } catch (error) {
1333 if (granted) await this.release(granted.held);
1334 await work.failMergecheck(job.pullId, job.token, error instanceof Error ? error.message : String(error));
1335 }
1336 }
1337
1338 /**
1339 * A refusal if `repo` takes no work from anyone: it is archived, so
1340 * read-only, or it was deleted (repos hides a deleted one, so it is not
1341 * found). Null when repos cannot answer now; the other checks still run.
1342 */
1343 private async closedRepo(actor: User, repo: RepoPath): Promise<Result<never> | null> {
1344 const repos = reposClient(this.env.REPOS);
1345 const found = await repos.get(repo, actor).catch(() => null);
1346 if (!found) return null;
1347 if (!found.ok) {
1348 return found.error.code === "not_found"
1349 ? fail("not_found", `There is no repository at ${repo.namespace}/${repo.name}, or it was deleted.`)
1350 : null;
1351 }
1352 const status = await repos.statusById(found.value.id).catch(() => null);
1353 if (status?.deleted) {
1354 return fail("not_found", `${found.value.namespace}/${found.value.name} was deleted. An owner can restore it from the workspace's settings.`);
1355 }
1356 if (status?.archived || found.value.archivedAt) {
1357 return fail(
1358 "forbidden",
1359 `${found.value.namespace}/${found.value.name} is archived, so it is read-only. An owner can unarchive it in its settings.`,
1360 );
1361 }
1362 return null;
1363 }
1364
1365 /**
1366 * Stops every run in a repository that was archived or deleted: the
1367 * work service marks them stopped when it hears of it, and lists them
1368 * here (`runs_in_repo`, by id, so a deleted repository's runs are found
1369 * too), and each sandbox is destroyed. Never throws.
1370 */
1371 private async stopRunsIn(repoId: string): Promise<void> {
1372 try {
1373 const response = await this.env.WORK.fetch("https://work/rpc/runs_in_repo", {
1374 method: "POST",
1375 headers: { "content-type": "application/json" },
1376 body: JSON.stringify({ repoId }),
1377 });
1378 if (!response.ok) return;
1379 const runs = (await response.json()) as { runId: string; sandbox: string | null }[];
1380 for (const run of runs) {
1381 if (!run.sandbox) continue;
1382 try {
1383 await this.env.SANDBOX.get(this.env.SANDBOX.idFromString(run.sandbox)).halt("The repository was archived or deleted.");
1384 } catch (error) {
1385 // Already gone, or never started.
1386 console.log("sandbox not destroyed", run.runId, String(error));
1387 }
1388 }
1389 } catch (error) {
1390 console.error("could not stop the runs in", repoId, error);
1391 }
1392 }
1393
1394 private async defaultBranch(repo: RepoPath, viewer: Viewer): Promise<string> {
1395 const found = await reposClient(this.env.REPOS).get(repo, viewer);
1396 return found.ok ? found.value.defaultBranch : "main";
1397 }
1398}