trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns13g rEventsFlows

A check the forge's own death kills is failed and never run again, and the change goes red for it #512

Merged masoncrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

mason/forge:mason/303-a-check-the-forge-s-own-death-kills-is-f → main 2 commits fork

  • Merged. mason/303-a-check-the-forge-s-own-death-kills-is-f was merged into main as 75bafb4.

Review

Sign in to review this change.
5 files changed +317 −28
src/apps/runtime.ts +7 −8
@@ -23,6 +23,7 @@ import { resolveAppEnv } from "../core/app_env.ts";
2323 import { mintAppToken, renewAppToken, revokeAppToken } from "./token.ts";
2424 import type { ForgeEvent } from "../core/events.ts";
2525 import { announceRun } from "../actions/runs.ts";
26 import { requeueAbandonedChecks } from "../checks/service.ts";
2627 import { getPull, listPulls, type Pull } from "../core/pulls.ts";
2728 import { getRepoById, type Repo } from "../core/repos.ts";
2829 import {
@@ -30,7 +31,6 @@ import {
3031 listRuns,
3132 claimRun,
3233 createRun,
33 CHECK_KINDS,
3434 failAbandonedRuns,
3535 finishRun,
3636 markServing,
@@ -515,14 +515,13 @@ export class AppsRuntime {
515515
516516 /** Rebuild every environment from git and the open pulls, then listen. */
517517 async start(): Promise<void> {
518 // Deploys and checks: an agent run the last forge left is the crew's,
518 // Deploys are failed; a check the last forge left is queued again while
519 // its commit still heads an open pull; an agent run is the crew's,
519520 // which queues it again rather than failing it.
520 const abandoned = failAbandonedRuns(
521 this.app.db,
522 STALE_AFTER_MS,
523 CHECK_KINDS,
524 );
525 if (abandoned > 0) log.info("failed abandoned runs", { count: abandoned });
521 const abandoned =
522 failAbandonedRuns(this.app.db, STALE_AFTER_MS, ["deploy"]) +
523 requeueAbandonedChecks(this.app, STALE_AFTER_MS);
524 if (abandoned > 0) log.info("settled abandoned runs", { count: abandoned });
526525 this.unsubscribe = this.app.bus.subscribe((event) => {
527526 // Never let an event handler's rejection reach the bus.
528527 this.onEvent(event).catch((error) =>
src/checks/service.ts +30 −14
@@ -25,6 +25,9 @@ import {
2525 createRun,
2626 finishRun,
2727 listActive,
28 requeueCheck,
29 REQUEUED_SUMMARY,
30 sweepAbandonedChecks,
2831 type Run,
2932 } from "../core/runs.ts";
3033 import { Gate } from "../git/env.ts";
@@ -102,6 +105,29 @@ export function pickNext(
102105 return newest;
103106 }
104107
108 /**
109 * Settle the checks a dead forge left running, and say so on the bus: a
110 * check whose commit still heads an open pull is queued again for whichever
111 * checks service is up to run, so the change keeps reading *working*.
112 * Called on boot and on the way by whichever forge is alive. A cancelled
113 * check is not announced, as a superseded one is not: `run.finished` reads
114 * as a verdict on the change, and the forge dying is none.
115 */
116 export function requeueAbandonedChecks(app: App, staleMs: number): number {
117 const swept = sweepAbandonedChecks(app.db, staleMs);
118 for (const { cancelled, again } of swept.requeued) {
119 appendRunLog(app.config.dataDir, cancelled.id, REQUEUED_SUMMARY);
120 announceRun(app, again, again.actorId ?? undefined);
121 }
122 for (const run of swept.cancelled)
123 appendRunLog(app.config.dataDir, run.id, run.summary);
124 for (const run of swept.failed) {
125 appendRunLog(app.config.dataDir, run.id, run.summary);
126 announceRun(app, run);
127 }
128 return swept.requeued.length + swept.failed.length + swept.cancelled.length;
129 }
130
105131 /** Where a check's checkout lives while it runs. */
106132 const checksRoot = (dataDir: string): string => join(dataDir, "checks");
107133
@@ -264,20 +290,10 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
264290 if (!outcome.ok && stopped && signal.aborted) {
265291 // Cut off by this forge stopping, not by the check: the same check
266292 // goes back on the queue, and the forge that carries on runs it.
267 const why = "interrupted by a restart; queued again";
268 cancelRun(app.db, run.id, why);
269 write(why);
270 const again = createRun(app.db, {
271 repoId: run.repoId,
272 kind: "check",
273 env: run.env,
274 commitOid: run.commitOid,
275 reason: run.reason,
276 attempt: run.attempt + 1,
277 ...(run.issueId === null ? {} : { issueId: run.issueId }),
278 ...(run.actorId === null ? {} : { actorId: run.actorId }),
279 });
280 announceRun(app, again, run.actorId ?? undefined);
293 const moved = requeueCheck(app.db, run.id);
294 if (!moved) return;
295 write(REQUEUED_SUMMARY);
296 announceRun(app, moved.again, run.actorId ?? undefined);
281297 return;
282298 }
283299 const seconds = Math.round((Date.now() - started) / 1000);
src/core/runs.ts +123
@@ -376,6 +376,129 @@ export function failAbandonedRuns(
376376 .run(cutoff, JSON.stringify(kinds)).changes;
377377 }
378378
379 /** What a check the forge's own restart cut off says as it is queued again. */
380 export const REQUEUED_SUMMARY = "interrupted by a restart; queued again";
381
382 /**
383 * How many times one check may be cut off by its forge going away and be
384 * queued again. Past it the check fails once, saying so: a check that kills
385 * its forge must not loop forever.
386 */
387 export const CHECK_RESTART_LIMIT = 3;
388
389 /**
390 * Cancel a check the forge's own going-away cut off and queue the same
391 * check again, `attempt + 1` — the one way a check is retried, from a
392 * graceful stop or a lapsed lease alike. Only a run still `running` is
393 * moved, and `lapsedBefore`, when given, must still be true of its lease:
394 * the UPDATE's WHERE is what decides between two forges, so only one of
395 * them ever queues the retry. Returns the cancelled run and its retry, or
396 * undefined when another worker got there first.
397 */
398 export function requeueCheck(
399 db: Database,
400 runId: string,
401 lapsedBefore?: string,
402 ): { cancelled: Run; again: Run } | undefined {
403 return db.transaction(() => {
404 const changes = db
405 .query(
406 `UPDATE run SET state = 'cancelled', summary = ?2,
407 finished_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
408 WHERE id = ?1 AND kind = 'check' AND state = 'running'
409 AND (?3 IS NULL OR heartbeat_at IS NULL OR heartbeat_at < ?3)`,
410 )
411 .run(runId, REQUEUED_SUMMARY, lapsedBefore ?? null).changes;
412 if (changes === 0) return undefined;
413 const cancelled = getRun(db, runId)!;
414 const again = createRun(db, {
415 repoId: cancelled.repoId,
416 kind: "check",
417 env: cancelled.env,
418 commitOid: cancelled.commitOid,
419 reason: cancelled.reason,
420 attempt: cancelled.attempt + 1,
421 ...(cancelled.issueId === null ? {} : { issueId: cancelled.issueId }),
422 ...(cancelled.actorId === null ? {} : { actorId: cancelled.actorId }),
423 });
424 return { cancelled, again };
425 })();
426 }
427
428 /** What a sweep of abandoned checks did, run by run. */
429 export interface AbandonedChecks {
430 /** Cancelled and queued again: the cancelled run and its retry. */
431 readonly requeued: readonly { cancelled: Run; again: Run }[];
432 /** Past `CHECK_RESTART_LIMIT`: failed once, saying the forge kept dying. */
433 readonly failed: readonly Run[];
434 /** No longer any open pull's head: cancelled, not retried. */
435 readonly cancelled: readonly Run[];
436 }
437
438 /**
439 * Checks whose lease lapsed — their forge died under them, which is never
440 * the commit's fault. One whose commit is still the head of an open pull is
441 * queued again (`requeueCheck`) until `CHECK_RESTART_LIMIT`, then failed
442 * once with a summary that says why; one whose commit no longer heads any
443 * open pull is cancelled and says so. Every move is a compare-and-swap on
444 * the lapsed lease, so two forges sweeping at once move each run once.
445 */
446 export function sweepAbandonedChecks(
447 db: Database,
448 staleMs: number,
449 ): AbandonedChecks {
450 const cutoff = new Date(Date.now() - staleMs).toISOString();
451 const rows = db
452 .query<RunRow & { live: number }, [string]>(
453 `SELECT run.*, EXISTS (
454 SELECT 1 FROM pull JOIN issue ON issue.id = pull.issue_id
455 WHERE pull.issue_id = run.issue_id AND issue.state = 'open'
456 AND pull.merged_at IS NULL AND pull.head_oid = run.commit_oid
457 ) AS live
458 FROM run
459 WHERE kind = 'check' AND state = 'running'
460 AND (heartbeat_at IS NULL OR heartbeat_at < ?1)`,
461 )
462 .all(cutoff);
463 const out = {
464 requeued: [] as { cancelled: Run; again: Run }[],
465 failed: [] as Run[],
466 cancelled: [] as Run[],
467 };
468 const settle = (runId: string, state: string, summary: string) =>
469 db
470 .query(
471 `UPDATE run SET state = ?2, summary = ?3,
472 finished_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
473 WHERE id = ?1 AND state = 'running'
474 AND (heartbeat_at IS NULL OR heartbeat_at < ?4)`,
475 )
476 .run(runId, state, summary, cutoff).changes > 0
477 ? getRun(db, runId)
478 : undefined;
479 for (const row of rows) {
480 if (row.live === 0) {
481 const done = settle(
482 row.id,
483 "cancelled",
484 "the forge went away, and this commit no longer heads an open pull",
485 );
486 if (done) out.cancelled.push(done);
487 } else if (row.attempt >= CHECK_RESTART_LIMIT) {
488 const done = settle(
489 row.id,
490 "failed",
491 `the forge kept going away under this check (${row.attempt + 1} times); not trying again`,
492 );
493 if (done) out.failed.push(done);
494 } else {
495 const moved = requeueCheck(db, row.id, cutoff);
496 if (moved) out.requeued.push(moved);
497 }
498 }
499 return out;
500 }
501
379502 /** Whether a running run's lease has lapsed. */
380503 export function leaseLapsed(
381504 run: Run,
src/main.ts +8 −6
@@ -14,13 +14,13 @@ import { runSync } from "./origins/sync.ts";
1414 import { startForgeSsh } from "./tui/ssh.ts";
1515 import { AppsRuntime } from "./apps/runtime.ts";
1616 import { resolveAppDomain } from "./apps/host.ts";
17 import { startChecks } from "./checks/service.ts";
17 import { requeueAbandonedChecks, startChecks } from "./checks/service.ts";
1818 import { startLooks } from "./looks/service.ts";
1919 import { startProbes } from "./contract/probes.ts";
2020 import { startSnapshots } from "./snapshot/service.ts";
2121 import { resolveDestination } from "./snapshot/destination.ts";
2222 import { releaseOrphans } from "./core/run_secret.ts";
23 import { CHECK_KINDS, failAbandonedRuns } from "./core/runs.ts";
23 import { failAbandonedRuns } from "./core/runs.ts";
2424 import { health } from "./http/health.ts";
2525 import { STALE_AFTER_MS } from "./runs/lease.ts";
2626 import { startCrew } from "./crew/dispatch.ts";
@@ -272,15 +272,17 @@ if (ssh)
272272
273273 // A run whose forge died is noticed by whichever forge is up, on boot and
274274 // on the way: an old process finishing beside a new one keeps its runs alive
275 // by renewing their leases, so only a lapsed lease is a death. Deploys and
276 // checks are failed; agent runs are the crew's to queue again, unless there
277 // is no crew here to do it.
275 // by renewing their leases, so only a lapsed lease is a death. Deploys are
276 // failed; checks are queued again while their commit heads an open pull;
277 // agent runs are the crew's to queue again, unless there is no crew here to
278 // do it.
278279 const leaseTimer = setInterval(() => {
279280 failAbandonedRuns(
280281 app.db,
281282 STALE_AFTER_MS,
282 crew === undefined ? undefined : CHECK_KINDS,
283 crew === undefined ? ["deploy", "agent"] : ["deploy"],
283284 );
285 requeueAbandonedChecks(app, STALE_AFTER_MS);
284286 // A grant whose run is no longer running is a crash's leftover: close it.
285287 releaseOrphans(app.db);
286288 }, 30_000);
test/checks.test.ts +149
@@ -15,18 +15,26 @@ import { parseBunTestFailure } from "../src/checks/bun-test.ts";
1515 import { summariseCheckFailure } from "../src/checks/summary.ts";
1616 import {
1717 pickNext,
18 requeueAbandonedChecks,
1819 startChecks,
1920 STARVATION_MS,
2021 type ChecksHandle,
2122 } from "../src/checks/service.ts";
2223 import {
2324 cancelRun,
25 CHECK_RESTART_LIMIT,
2426 checksForHead,
27 claimRun,
2528 createRun,
2629 getRun,
2730 listRuns,
31 REQUEUED_SUMMARY,
2832 type Run,
2933 } from "../src/core/runs.ts";
34 import type { Pull } from "../src/core/pulls.ts";
35 import { repoDir } from "../src/git/repos.ts";
36 import { STALE_AFTER_MS } from "../src/runs/lease.ts";
37 import { pullWord, servingHistory } from "../src/web/studio/words.ts";
3038 import { createUser, type User } from "../src/core/users.ts";
3139 import type { Repo } from "../src/core/repos.ts";
3240 import { serve, type Servers } from "../src/http/server.ts";
@@ -567,3 +575,144 @@ describe("checks", () => {
567575 expect(done.state).toBe("ok");
568576 }, 60_000);
569577 });
578
579 describe("a check its forge died under", () => {
580 let dataDir: string;
581 let app: App;
582 let alice: User;
583 let repo: Repo;
584 let pull: Pull;
585 let stale: string;
586
587 beforeAll(async () => {
588 dataDir = await mkdtemp(join(tmpdir(), "forge-checks-abandoned-"));
589 app = createApp({ dataDir, port: 0 });
590 const a = await createUser(app.db, {
591 username: "alice",
592 email: "[email protected]",
593 password: "a-good-passphrase",
594 adminIfFirst: true,
595 });
596 if (!a.ok) throw new Error(a.error);
597 alice = a.value;
598 const made = await createRepo(app, alice, { name: "lib" });
599 if (!made.ok) throw new Error(String(made.error));
600 repo = made.value;
601 const seeded = await commitToBranch(app, alice, repo, {
602 message: "Start",
603 files: [{ path: "README.md", content: "# lib\n" }],
604 });
605 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
606 const first = await commitToBranch(app, alice, repo, {
607 branch: "feature",
608 message: "First",
609 files: [{ path: "A.md", content: "a\n" }],
610 });
611 if (!first.ok) throw new Error(JSON.stringify(first.error));
612 stale = first.value.commitOid;
613 const second = await commitToBranch(app, alice, repo, {
614 branch: "feature",
615 message: "Second",
616 files: [{ path: "B.md", content: "b\n" }],
617 });
618 if (!second.ok) throw new Error(JSON.stringify(second.error));
619 const opened = await openPull(app, alice, repo, {
620 title: "Feature",
621 baseBranch: "main",
622 headBranch: "feature",
623 });
624 if (!opened.ok) throw new Error(String(opened.error));
625 pull = opened.value;
626 });
627
628 afterAll(async () => {
629 app.close();
630 await rm(dataDir, { recursive: true, force: true });
631 });
632
633 /** A check that was running when its forge died: claimed, lease long lapsed. */
634 const abandoned = (commitOid: string, attempt = 0): Run => {
635 const run = createRun(app.db, {
636 repoId: repo.id,
637 kind: "check",
638 env: `pull-${pull.number}`,
639 commitOid,
640 reason: "push",
641 issueId: pull.id,
642 attempt,
643 });
644 claimRun(app.db, run.id);
645 app.db
646 .query("UPDATE run SET heartbeat_at = ?2 WHERE id = ?1")
647 .run(run.id, new Date(Date.now() - STALE_AFTER_MS - 1_000).toISOString());
648 return getRun(app.db, run.id)!;
649 };
650
651 const word = () =>
652 pullWord(
653 {
654 db: app.db,
655 dir: repoDir(app.reposRoot, repo.ownerName, repo.name),
656 runtime: undefined,
657 record: undefined,
658 history: servingHistory(),
659 },
660 pull,
661 );
662
663 const queuedFor = (commitOid: string): Run[] =>
664 listRuns(app.db, repo.id, { kind: "check", limit: 50 }).filter(
665 (r) => r.commitOid === commitOid && r.state === "queued",
666 );
667
668 test("a check for the head of an open pull is queued again, attempt + 1, and the change reads working", async () => {
669 const run = abandoned(pull.headOid);
670 expect((await word()).word).toBe("working");
671 expect(requeueAbandonedChecks(app, STALE_AFTER_MS)).toBe(1);
672 // A second forge sweeping the same lapse finds nothing left to move.
673 expect(requeueAbandonedChecks(app, STALE_AFTER_MS)).toBe(0);
674 const cut = getRun(app.db, run.id)!;
675 expect(cut.state).toBe("cancelled");
676 expect(cut.summary).toBe(REQUEUED_SUMMARY);
677 const again = queuedFor(pull.headOid);
678 expect(again).toHaveLength(1);
679 expect(again[0]!.attempt).toBe(1);
680 expect(again[0]!.issueId).toBe(pull.id);
681 expect(again[0]!.env).toBe(`pull-${pull.number}`);
682 expect(await word()).toEqual({ word: "working", note: "queued to check" });
683 cancelRun(app.db, again[0]!.id);
684 });
685
686 test("a check whose lease is still held is left alone", () => {
687 const run = createRun(app.db, {
688 repoId: repo.id,
689 kind: "check",
690 env: `pull-${pull.number}`,
691 commitOid: pull.headOid,
692 issueId: pull.id,
693 });
694 claimRun(app.db, run.id);
695 expect(requeueAbandonedChecks(app, STALE_AFTER_MS)).toBe(0);
696 expect(getRun(app.db, run.id)?.state).toBe("running");
697 cancelRun(app.db, run.id);
698 });
699
700 test("at the ceiling the check fails once, saying the forge kept going away", async () => {
701 const run = abandoned(pull.headOid, CHECK_RESTART_LIMIT);
702 expect(requeueAbandonedChecks(app, STALE_AFTER_MS)).toBe(1);
703 const failed = getRun(app.db, run.id)!;
704 expect(failed.state).toBe("failed");
705 expect(failed.summary).toContain("the forge kept going away under this check");
706 expect(queuedFor(pull.headOid)).toHaveLength(0);
707 expect((await word()).word).toBe("stopped");
708 });
709
710 test("a check for a commit that no longer heads an open pull is cancelled, not queued again", () => {
711 const run = abandoned(stale);
712 expect(requeueAbandonedChecks(app, STALE_AFTER_MS)).toBe(1);
713 const cut = getRun(app.db, run.id)!;
714 expect(cut.state).toBe("cancelled");
715 expect(cut.summary).toContain("no longer heads an open pull");
716 expect(queuedFor(stale)).toHaveLength(0);
717 });
718 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs