Skip to content
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 // Once any of its traffic is intercepted, Cloudflare passes all of it
385 // through this Worker, and HTTPS is re-signed on the way
386 // (`interceptHttps`); so a sandbox with either handler trusts that
387 // certificate, or every HTTPS request it makes fails
388 // (SELF_SIGNED_CERT_IN_CHAIN, as on 2026-10-13).
389 const intercepted = brokering || this.env.EGRESS?.trim().toLowerCase() !== "off";
390 const trust = intercepted ? EGRESS_ENV : {};
391 const watch: Record<string, string> = this.env.ABUSE_WATCH === "off" ? { G1T_ABUSE: "off" } : {};
392 await this.start({ envVars: { ...vars, ...harness, ...trust, ...watch }, enableInternet: true });
393 // A backup has no guardrails, but still a time cap.
394 const cap = guard?.minutes ?? (run.kind === "backup" ? BACKUP_MINUTES : null);
395 if (cap) {
396 await this.ctx.storage.put("timeCap", cap);
397 await this.schedule(cap * 60 + ALARM_GRACE_SECONDS, "timeUp");
398 }
399 } catch (error) {
400 console.error("sandbox not started", run.kind, describeError(error));
401 await revokeCredentials(this.env.IDENTITY, this.ctx.storage);
402 if (tracked) await this.closeRun("failed", `The sandbox could not start: ${String(error)}`);
403 await this.settle(0);
404 throw error;
405 }
406 }
407
408 /** Settles what billing reserved for this sandbox at `micros`, once. */
409 private async settle(micros: number): Promise<void> {
410 const held = await this.ctx.storage.get<Held>("reservation");
411 if (!held) return;
412 await this.ctx.storage.delete("reservation");
413 await gateFor(this.env).settle(held.id, micros);
414 }
415
416 /** Settles the reservation at what the sandbox cost: its seconds at the price billing reserved at. */
417 private async settleStopped(started: number | undefined): Promise<void> {
418 const held = await this.ctx.storage.get<Held>("reservation");
419 if (!held) return;
420 const seconds = started ? Math.max(1, Math.ceil((Date.now() - started) / 1000)) : 0;
421 await this.settle(actualMicros(seconds, held.microsPerSecond));
422 }
423
424 /**
425 * The sandbox stopped itself because it looked like it was mining
426 * (crates/runner abuse.rs), or exited saying so. Stops the run with
427 * `ABUSE_MESSAGE`, tells g1t's staff with `abuse.flagged`, and destroys
428 * the sandbox. Once.
429 */
430 async flagAbuse(verdict: unknown): Promise<void> {
431 if (await this.ctx.storage.get<boolean>("abuse")) return;
432 await this.ctx.storage.put("abuse", true);
433 const tracked = await this.ctx.storage.get<TrackedRun>("agentRun");
434 if (tracked) await reportRun(this.env.WORK, tracked, { halt: "abuse", error: ABUSE_MESSAGE });
435 const owner = await this.ctx.storage.get<{ workspace: string; repo: string | null; kind: string }>("owner");
436 console.log("abuse flagged", owner?.workspace, owner?.repo, owner?.kind, JSON.stringify(verdict));
437 if (this.env.EVENTS && owner) {
438 await eventsClient(this.env.EVENTS)
439 .publish([
440 {
441 type: "abuse.flagged",
442 source: "runner",
443 // Never on a repository's timeline or its webhooks.
444 repoId: null,
445 actor: null,
446 data: {
447 workspace: owner.workspace,
448 repo: owner.repo ?? null,
449 run: tracked?.runId ?? null,
450 kind: owner.kind,
451 sandbox: this.ctx.id.toString(),
452 metrics: verdict && typeof verdict === "object" ? (verdict as Record<string, unknown>) : null,
453 },
454 },
455 ])
456 .catch((error: unknown) => console.log("abuse.flagged not published", String(error)));
457 }
458 await this.destroy().catch((error: unknown) => console.log("sandbox not destroyed for abuse", String(error)));
459 }
460
461 /**
462 * Records the run, which the sandbox then reports its steps to. Never
463 * stops the sandbox from starting: without a record it just goes unseen.
464 */
465 private async openRun(track: Track, envVars: Record<string, string>, guard: RunGuard | null): Promise<TrackedRun | null> {
466 const opened = await agentsClient(this.env.WORK)
467 .openRun({
468 ...track,
469 model: null,
470 sandbox: this.ctx.id.toString(),
471 timeCapMinutes: guard?.minutes ?? null,
472 })
473 .catch((error: unknown) => ({ ok: false as const, error: { message: String(error) } }));
474 if (!opened.ok) {
475 console.log("run not recorded", track.kind, opened.error.message);
476 return null;
477 }
478 await this.ctx.storage.put("agentRun", opened.value);
479 return opened.value;
480 }
481
482 /**
483 * A workflow job sent a brokered secret's stand-in to a host it is not
484 * for: said once per secret and host, in the job's log.
485 */
486 async noteMisdirected(name: string, host: string): Promise<void> {
487 const run = await this.ctx.storage.get<Run>("run");
488 if (run?.kind !== "actions") return;
489 const noted = newlyMisdirected((await this.ctx.storage.get<string[]>("misdirected")) ?? [], name, host);
490 if (!noted) return;
491 await this.ctx.storage.put("misdirected", noted.seen);
492 try {
493 const response = await this.env.ACTIONS.fetch("https://actions/rpc/job_report", {
494 method: "POST",
495 headers: { "content-type": "application/json" },
496 body: JSON.stringify({ job: run.jobId, token: run.token, report: { kind: "note", text: noted.note } }),
497 });
498 await answered("job_report", response);
499 } catch (error) {
500 console.log("misdirected secret not noted", name, String(error));
501 }
502 }
503
504 /** The run's time cap has passed: stop it, as stopped for time. */
505 async timeUp(): Promise<void> {
506 // It already stopped (nothing to stop), or the run in it now is a later
507 // attempt whose cap has not passed: this alarm is an earlier attempt's,
508 // left by a version of g1t that did not clear it, and the run now has
509 // its own.
510 const started = await this.ctx.storage.get<number>("started");
511 const minutes = (await this.ctx.storage.get<number>("timeCap")) ?? 0;
512 if (!capReached(started, minutes, Date.now())) {
513 if (started) console.log("an earlier attempt's time cap passed; the run now keeps going", this.ctx.id.toString());
514 return;
515 }
516 const tracked = await this.ctx.storage.get<TrackedRun>("agentRun");
517 // A workflow job or a build has no run to halt: it fails saying why.
518 await this.ctx.storage.put("stopReason", timeCapMessage(minutes));
519 if (tracked) await reportRun(this.env.WORK, tracked, { halt: "time", error: timeCapMessage(minutes) });
520 if (await this.ctx.storage.get<boolean>("remote")) {
521 await cancelTask(this.env.ACTIONS, this.ctx.id.toString(), timeCapMessage(minutes));
522 await this.remoteEnded(1, null);
523 return;
524 }
525 await this.destroy().catch((error: unknown) => console.log("sandbox not destroyed at its time cap", String(error)));
526 }
527
528 /**
529 * Stops this sandbox's work: its container, or the task a self-hosted
530 * runner holds, which it hears about on its next poll.
531 */
532 async halt(reason: string | null): Promise<void> {
533 if (await this.ctx.storage.get<boolean>("remote")) {
534 await cancelTask(this.env.ACTIONS, this.ctx.id.toString(), reason);
535 await this.remoteEnded(1, reason);
536 return;
537 }
538 // A container already gone has nothing to stop; one that will not stop
539 // is logged, and its time cap still ends it.
540 await this.destroy().catch((error: unknown) => console.error("sandbox not destroyed", reason, describeError(error)));
541 }
542
543 /**
544 * The library's `sleepAfter` has passed. The runner never fetches its
545 * container, so to the library every sandbox looks idle: a run inside its
546 * time cap keeps going, and the cap's own alarm (`timeUp`) ends it. Only
547 * a sandbox with no cap is stopped for inactivity.
548 */
549 override async onActivityExpired(): Promise<void> {
550 const started = await this.ctx.storage.get<number>("started");
551 const cap = await this.ctx.storage.get<number>("timeCap");
552 if (withinTimeCap(started, cap, Date.now(), ALARM_GRACE_SECONDS)) return;
553 await super.onActivityExpired();
554 }
555
556 /**
557 * A container that crashed or could not be reached, as the library tells
558 * it. Logged at error level with the sandbox, never thrown: the library
559 * ignores what this throws, and the stop that follows is handled by
560 * `onStop` or by `run`, which says why the work did not start.
561 */
562 override onError(error: unknown): void {
563 console.error("sandbox container error", this.ctx.id.toString(), describeError(error));
564 }
565
566 /**
567 * The library's alarm: scheduled callbacks, the container's keep-alive,
568 * and `onStop` once it has stopped. A failure is logged with the retry it
569 * was, then thrown so Cloudflare tries the alarm again.
570 */
571 override async alarm(alarmProps?: AlarmInvocationInfo): Promise<void> {
572 try {
573 await super.alarm(alarmProps);
574 } catch (error) {
575 console.error("sandbox alarm failed", this.ctx.id.toString(), `retry ${alarmProps?.retryCount ?? 0}`, describeError(error));
576 throw error;
577 }
578 }
579
580 /**
581 * A self-hosted runner's task ended (the actions service says so, or g1t
582 * stopped it): everything a container's stop does, once.
583 */
584 async remoteEnded(exitCode: number, reason: string | null): Promise<void> {
585 if (!(await this.ctx.storage.get<boolean>("remote"))) return;
586 await this.ctx.storage.delete("remote");
587 await this.ctx.storage.put("selfHostedEnded", true);
588 if (exitCode !== 0 && reason && !(await this.ctx.storage.get<string>("stopReason"))) {
589 await this.ctx.storage.put("stopReason", reason);
590 }
591 await this.onStop({ exitCode, reason: "exit" } as StopParams);
592 await this.ctx.storage.delete("selfHostedEnded");
593 }
594
595 /**
596 * Ends the run's record, once. Returns the status it ended with:
597 * `stopped` when a person stopped it first.
598 */
599 private async closeRun(outcome: "succeeded" | "failed", error?: string): Promise<string | null> {
600 const tracked = await this.ctx.storage.get<TrackedRun>("agentRun");
601 if (!tracked) return null;
602 await this.ctx.storage.delete("agentRun");
603 const closed = await agentsClient(this.env.WORK)
604 .closeRun(tracked.runId, tracked.token, outcome, error)
605 .catch(() => null);
606 return closed?.ok ? closed.value : null;
607 }
608
609 /** Reports how long the sandbox ran, once, whatever it exited with. */
610 private async meterStop(): Promise<void> {
611 const metered = await this.ctx.storage.get<Meter & { started: number }>("meter");
612 if (!metered) return;
613 await this.ctx.storage.delete("meter");
614 const seconds = Math.max(1, Math.ceil((Date.now() - metered.started) / 1000));
615 const run = await this.ctx.storage.get<Run>("run");
616 // On the workspace's own runner: its minutes, at $0.
617 const selfHosted = (await this.ctx.storage.get<boolean>("selfHostedEnded")) ?? false;
618 const recorded = await billingClient(this.env.BILLING)
619 .recordSandbox({
620 workspace: metered.workspace,
621 seconds,
622 description: selfHosted ? `${metered.description} on a self-hosted runner` : metered.description,
623 repo: metered.repo,
624 reference: `sandbox/${this.ctx.id.toString()}/${metered.started}`,
625 // Whether g1t's open-source pool may pay for it.
626 kind: run ? computeKindOf(run.kind) : null,
627 selfHosted,
628 instance: metered.instance ?? null,
629 })
630 .catch((error: unknown) => ({ ok: false as const, error: { message: String(error) } }));
631 if (!recorded.ok) console.log("sandbox time not recorded", metered.workspace, seconds, recorded.error.message);
632 }
633
634 override async onStop({ exitCode, reason }: StopParams): Promise<void> {
635 await revokeCredentials(this.env.IDENTITY, this.ctx.storage);
636 const started = await this.ctx.storage.get<number>("started");
637 await this.meterStop();
638 // It stopped itself for mining, and could not say so before it went.
639 if (exitCode === ABUSE_EXIT_CODE && !(await this.ctx.storage.get<boolean>("abuse"))) {
640 await this.flagAbuse(null);
641 }
642 const flagged = (await this.ctx.storage.get<boolean>("abuse")) ?? false;
643 // Why it stopped, when g1t stopped it: said in place of a plain failure.
644 const why = flagged ? ABUSE_MESSAGE : ((await this.ctx.storage.get<string>("stopReason")) ?? null);
645 // Stopped by g1t, it failed whatever exit code the library passed on.
646 const short = stoppedShort(exitCode, why);
647 await this.closeRun(
648 short ? "failed" : "succeeded",
649 short ? (why ?? `The sandbox exited with ${exitCode}.`) : undefined,
650 );
651 await this.settleStopped(started);
652 await this.ctx.storage.delete("started");
653 if (!short) return;
654 const run = await this.ctx.storage.get<Run>("run");
655 console.log("sandbox stopped", run?.kind, "exit", exitCode, reason);
656 if (!run) return;
657 // Never thrown: this runs in the sandbox's alarm, which a throw would
658 // fail, retry and count as an error, running all of the above again.
659 // What could not be told is logged, and the sweep catches it up.
660 await tellStopped(run.kind, () => this.reportStopped(run, why, exitCode));
661 }
662
663 /**
664 * Tells whoever is waiting on the sandbox's work that it stopped without
665 * finishing it. Each is refused harmlessly when the sandbox reported its
666 * end before it stopped. Throws when the service could not be reached.
667 */
668 private async reportStopped(run: Run, why: string | null, exitCode: number): Promise<void> {
669 if (run.kind === "actions") {
670 // Refused harmlessly if the job reported its end before it stopped.
671 const response = await this.env.ACTIONS.fetch("https://actions/rpc/job_report", {
672 method: "POST",
673 headers: { "content-type": "application/json" },
674 body: JSON.stringify({
675 job: run.jobId,
676 token: run.token,
677 report: { kind: "done", conclusion: "failure", reason: why ?? "The runner stopped before the job finished." },
678 }),
679 });
680 await answered("job_report", response);
681 return;
682 }
683 // Nothing was pushed, so no pull request opens; why is in its log.
684 if (run.kind === "bump") return;
685 if (run.kind === "backup") {
686 // Refused harmlessly if the sandbox reported before it stopped; the
687 // job is otherwise tried again later tonight.
688 await reposClient(this.env.REPOS).failBackup(run.jobId, run.token, why ?? `The sandbox exited with ${exitCode}.`);
689 return;
690 }
691 if (run.kind === "deploy") {
692 // Refused harmlessly if the build reported its end before it stopped.
693 const response = await this.env.DEPLOYMENTS.fetch(`https://deployments/jobs/${run.deployId}/fail`, {
694 method: "POST",
695 headers: { "content-type": "application/json" },
696 body: JSON.stringify({ token: run.token, message: why ?? "The build stopped before it finished." }),
697 });
698 await answered("deploy fail", response);
699 return;
700 }
701 const work = workClient(this.env.WORK);
702 if (run.kind === "checks") {
703 // Refused harmlessly if the run did report before it stopped.
704 await work.reportChecks(run.runId, run.token, {
705 error: why ?? "The sandbox stopped before the checks finished.",
706 });
707 return;
708 }
709 if (run.kind === "queue") {
710 // Refused harmlessly if the state was reported before it stopped.
711 await work.failQueue(run.entryId, run.token, why ?? "The sandbox stopped before the state was checked.");
712 return;
713 }
714 if (run.kind === "mergecheck") {
715 // Refused harmlessly if the probe reported before it stopped.
716 await work.failMergecheck(run.pullId, run.token, why ?? "The sandbox stopped before the merge check finished.");
717 return;
718 }
719 // A sandbox of a kind g1t no longer starts, which ran from before: it
720 // has nobody left to tell.
721 }
722}
723
724/**
725 * What the compute gate decided for one start: go, with what billing
726 * reserved and the plan's caps; or not, refused with what to tell people.
727 */
728type Granted = {
729 ok: true;
730 held: Held | null;
731 limits: PlanLimits;
732 /** The labels of the self-hosted runners it goes to; null for a sandbox. */
733 route: string[] | null;
734};
735type Admitted = Granted | { ok: false; code: string; message: string };
736
737/**
738 * The guardrails' default time cap of each kind of run, for estimating what
739 * it may cost before it starts; the sandbox applies the project's own.
740 */
741const DEFAULT_MINUTES: Record<"queue" | "mergecheck", number> = {
742 queue: 45,
743 mergecheck: 10,
744};
745
746/** A plan's cap on one run, for the sandbox to apply under its guardrails'. */
747function limitsOf(ent: ComputeEntitlements | null): PlanLimits {
748 return { minutes: ent && ent.maxRunMinutes > 0 ? ent.maxRunMinutes : null };
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}