trevato/forge

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

More
Forks1g fMapg aRuns3g rEventsFlows

A pull run records an adapter's /pull answer: counts, cursor and duration, validated at the boundary #10058

Merged torquecrew opened this

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

ConversationFiles
Merged

torque/forge:torque/10042-a-pull-run-records-an-adapter-s-pull-ans → main 2 commits fork

  • Merged. torque/10042-a-pull-run-records-an-adapter-s-pull-ans was merged into main as 2b47779.

Review

Sign in to review this change.
7 files changed +445 −3
docs/ADAPTERS.md +4 −3
@@ -11,9 +11,10 @@ answer and records each pull as a run.
1111 ## Status
1212
1313 This is the contract, not a description of running code. Today the forge does
14 not read the `adapter` block, does not call `POST /pull` and does not record
15 pull runs. An adapter written to this page is an ordinary app until that work
16 lands. The rest of the forge's app rules apply to it now: see
14 not read the `adapter` block and does not call `POST /pull`. The answer's
15 checks and the pull run record exist (`src/apps/pull.ts`: `parsePullAnswer`,
16 `recordPull`, kept in the `pull_run` table), but nothing calls them yet. An
17 adapter written to this page is an ordinary app until that work lands. The rest of the forge's app rules apply to it now: see
1718 [`docs/APPS.md`](APPS.md).
1819
1920 ## The `adapter` block in forge.json
src/apps/pull.ts added +144
@@ -0,0 +1,144 @@
1 /**
2 * An adapter's `POST /pull` answer, judged and recorded (docs/ADAPTERS.md,
3 * #10042). `parsePullAnswer` is the boundary: pure, given the status and the
4 * raw body, it answers four checked fields or a failed pull with a short
5 * reason naming what was wrong. `recordPull` writes the outcome as a run of
6 * kind `pull` through `core/runs.ts`. Nothing here reaches the network: the
7 * caller that sends the request and times it comes later (#437).
8 */
9
10 import type { Database } from "bun:sqlite";
11 import { createPullRun, lastPullCursor, type PullRun } from "../core/runs.ts";
12
13 /** A body over this many bytes is a failed pull, never parsed. */
14 export const PULL_BODY_MAX = 64 * 1024;
15 /** Each count is an integer from 0 to this. */
16 export const PULL_COUNT_MAX = 1_000_000;
17 /** The cursor is at most this many characters (code points). */
18 export const PULL_CURSOR_MAX = 200;
19 /** A failed pull's reason is cut to this many characters. */
20 const REASON_MAX = 200;
21
22 const FIELDS = ["pulled", "skipped", "cursor", "errors"] as const;
23 const COUNTS = ["pulled", "skipped", "errors"] as const;
24
25 /** What a repeat run's reason says. */
26 export const REPEAT_REASON = "same cursor as the last good pull: nothing new written";
27
28 export type PullAnswer =
29 | {
30 readonly ok: true;
31 readonly pulled: number;
32 readonly skipped: number;
33 readonly errors: number;
34 readonly cursor: string;
35 }
36 | { readonly ok: false; readonly reason: string };
37
38 /** An answer and how long the request took, as `recordPull` takes it. */
39 export type PullOutcome = PullAnswer & { readonly durationMs: number };
40
41 const fail = (reason: string): PullAnswer => ({ ok: false, reason: reason.slice(0, REASON_MAX) });
42
43 /**
44 * Any control, format, private-use, unassigned or surrogate character, or a
45 * line or paragraph separator: what "one line of printable characters" excludes.
46 */
47 const NOT_PRINTABLE = /[\p{C}\p{Zl}\p{Zp}]/u;
48
49 /** Whether `value` is a cursor the contract allows: one line, at most 200 printable characters. */
50 export function isPullCursor(value: unknown): value is string {
51 return (
52 typeof value === "string" &&
53 [...value].length <= PULL_CURSOR_MAX &&
54 !NOT_PRINTABLE.test(value)
55 );
56 }
57
58 const isCount = (value: unknown): value is number =>
59 typeof value === "number" && Number.isInteger(value) && value >= 0 && value <= PULL_COUNT_MAX;
60
61 /**
62 * Judge one answer. A non-2xx status, a body over 64 KB, a body that is not a
63 * JSON object of exactly `pulled`, `skipped`, `cursor` and `errors`, or any
64 * field outside its rule is a failed pull, and the reason names the first
65 * thing wrong — in a fixed order, so the same body always gives the same one.
66 */
67 export function parsePullAnswer(status: number, body: string): PullAnswer {
68 if (!Number.isInteger(status) || status < 200 || status > 299) {
69 return fail(`status ${Number.isInteger(status) ? status : "unknown"} is not 2xx`);
70 }
71 if (Buffer.byteLength(body, "utf8") > PULL_BODY_MAX) {
72 return fail(`body is over ${PULL_BODY_MAX} bytes`);
73 }
74 let parsed: unknown;
75 try {
76 parsed = JSON.parse(body);
77 } catch {
78 return fail("body is not JSON");
79 }
80 if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
81 return fail("body is not a JSON object");
82 }
83 const record = parsed as Record<string, unknown>;
84 const keys = Object.keys(record);
85 const extra = keys.find((k) => !(FIELDS as readonly string[]).includes(k));
86 if (extra !== undefined) return fail(`unexpected field ${JSON.stringify(extra.slice(0, 40))}`);
87 const missing = FIELDS.find((f) => !Object.hasOwn(record, f));
88 if (missing !== undefined) return fail(`${missing} is missing`);
89 for (const field of COUNTS) {
90 if (!isCount(record[field])) return fail(`${field} is not an integer from 0 to ${PULL_COUNT_MAX}`);
91 }
92 if (!isPullCursor(record.cursor)) {
93 return fail(`cursor is not one line of at most ${PULL_CURSOR_MAX} printable characters`);
94 }
95 return {
96 ok: true,
97 pulled: record.pulled as number,
98 skipped: record.skipped as number,
99 errors: record.errors as number,
100 cursor: record.cursor,
101 };
102 }
103
104 /** A duration the table accepts: a whole, non-negative number of milliseconds. */
105 const durationOf = (ms: number): number => (Number.isFinite(ms) && ms > 0 ? Math.round(ms) : 0);
106
107 /**
108 * Record one pull as a run. A failed answer keeps its reason and no cursor, so
109 * the last good cursor stays where the next pull starts. A good answer whose
110 * cursor equals the last good pull's is a repeat: it is recorded, with its
111 * `pulled` counted as skipped and nothing written, so the same records never
112 * count as written twice. The read and the write are one transaction, so two
113 * recordings of the same answer cannot both see a fresh cursor.
114 */
115 export function recordPull(db: Database, appId: string, outcome: PullOutcome): PullRun {
116 const durationMs = durationOf(outcome.durationMs);
117 return db.transaction((): PullRun => {
118 if (!outcome.ok) {
119 return createPullRun(db, {
120 appId,
121 state: "failed",
122 written: 0,
123 skipped: 0,
124 failed: 0,
125 cursor: "",
126 durationMs,
127 reason: outcome.reason.slice(0, REASON_MAX),
128 repeat: false,
129 });
130 }
131 const repeat = lastPullCursor(db, appId) === outcome.cursor;
132 return createPullRun(db, {
133 appId,
134 state: "ok",
135 written: repeat ? 0 : outcome.pulled,
136 skipped: repeat ? Math.min(PULL_COUNT_MAX, outcome.skipped + outcome.pulled) : outcome.skipped,
137 failed: outcome.errors,
138 cursor: outcome.cursor,
139 durationMs,
140 reason: repeat ? REPEAT_REASON : "",
141 repeat,
142 });
143 }).immediate();
144 }
src/core/apps.ts +1
@@ -236,6 +236,7 @@ export function deleteApp(db: Database, appId: string): boolean {
236236 db.run("DELETE FROM app_feature WHERE app_id = ?1", [appId]);
237237 db.run("DELETE FROM app_feature_setting WHERE app_id = ?1", [appId]);
238238 db.run("DELETE FROM feature_fire WHERE app_id = ?1", [appId]);
239 db.run("DELETE FROM pull_run WHERE app_id = ?1", [appId]);
239240 return db.query("DELETE FROM app WHERE id = ?1").run(appId).changes > 0;
240241 })();
241242 }
src/core/migration-pins/20261109120000-pull-run.pin added +1
@@ -0,0 +1,1 @@
1 b247f3e8d2a52ab9b79dfc409b7926250a5306df926f0ff5f3feefd24b1a5034
src/core/migrations/20261109120000-pull-run.ts added +28
@@ -0,0 +1,28 @@
1 // A run of kind `pull` (#10042): one row per `POST /pull` the forge made to an
2 // adapter (docs/ADAPTERS.md), written by `recordPull` (`apps/pull.ts`) through
3 // `createPullRun` (`core/runs.ts`). Its own table, not a `run` row: `run.kind`'s
4 // CHECK can only be widened by rebuilding `run`, and a migration only adds
5 // (`test/schema-additive.test.ts`). `written`, `skipped` and `failed` are the
6 // answer's `pulled`, `skipped` and `errors`, 0..1,000,000; `cursor` is empty on
7 // a failed pull, whose `reason` says what was wrong; `repeat` marks an answer
8 // whose cursor matched the last good one, recorded with nothing written.
9 // `pull_run_app` serves an app's list newest first; `pull_run_app_ok` serves
10 // the last good cursor. No foreign key, like `feature_fire`: `deleteApp`
11 // clears the rows. Additive.
12 export default `
13 CREATE TABLE pull_run (
14 id TEXT PRIMARY KEY,
15 app_id TEXT NOT NULL,
16 state TEXT NOT NULL CHECK (state IN ('ok','failed')),
17 written INTEGER NOT NULL DEFAULT 0 CHECK (written BETWEEN 0 AND 1000000),
18 skipped INTEGER NOT NULL DEFAULT 0 CHECK (skipped BETWEEN 0 AND 1000000),
19 failed INTEGER NOT NULL DEFAULT 0 CHECK (failed BETWEEN 0 AND 1000000),
20 cursor TEXT NOT NULL DEFAULT '',
21 duration_ms INTEGER NOT NULL DEFAULT 0 CHECK (duration_ms >= 0),
22 reason TEXT NOT NULL DEFAULT '',
23 repeat INTEGER NOT NULL DEFAULT 0,
24 created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
25 );
26 CREATE INDEX pull_run_app ON pull_run(app_id, id DESC);
27 CREATE INDEX pull_run_app_ok ON pull_run(app_id, id DESC) WHERE state = 'ok';
28 `;
src/core/runs.ts +113
@@ -1474,3 +1474,116 @@ export function runningCount(db: Database, repoId: string): number {
14741474 .get(repoId)?.n ?? 0
14751475 );
14761476 }
1477
1478 /**
1479 * A run of kind `pull`: one `POST /pull` the forge made to an adapter
1480 * (docs/ADAPTERS.md), kept in `pull_run` because `run.kind`'s CHECK cannot
1481 * widen without a rebuild (migration `20261109120000-pull-run`). Plain data:
1482 * what the answer is worth is decided in `apps/pull.ts`, which writes here.
1483 */
1484 export interface PullRun {
1485 readonly id: string;
1486 readonly appId: string;
1487 readonly kind: "pull";
1488 readonly state: "ok" | "failed";
1489 /** Records the pull wrote — the answer's `pulled`; 0 on a repeat or a failure. */
1490 readonly written: number;
1491 /** Records read and not written — the answer's `skipped`. */
1492 readonly skipped: number;
1493 /** Records that failed — the answer's `errors`. */
1494 readonly failed: number;
1495 /** Where the next pull starts; empty on a failed pull. */
1496 readonly cursor: string;
1497 readonly durationMs: number;
1498 /** Why a failed pull failed, or what a repeat was; empty otherwise. */
1499 readonly reason: string;
1500 /** The cursor matched the last good pull's, so nothing counted as written. */
1501 readonly repeat: boolean;
1502 readonly createdAt: string;
1503 }
1504
1505 interface PullRunRow {
1506 id: string;
1507 app_id: string;
1508 state: "ok" | "failed";
1509 written: number;
1510 skipped: number;
1511 failed: number;
1512 cursor: string;
1513 duration_ms: number;
1514 reason: string;
1515 repeat: number;
1516 created_at: string;
1517 }
1518
1519 const toPullRun = (r: PullRunRow): PullRun => ({
1520 id: r.id,
1521 appId: r.app_id,
1522 kind: "pull",
1523 state: r.state,
1524 written: r.written,
1525 skipped: r.skipped,
1526 failed: r.failed,
1527 cursor: r.cursor,
1528 durationMs: r.duration_ms,
1529 reason: r.reason,
1530 repeat: r.repeat !== 0,
1531 createdAt: r.created_at,
1532 });
1533
1534 export interface CreatePullRunInput {
1535 appId: string;
1536 state: "ok" | "failed";
1537 written: number;
1538 skipped: number;
1539 failed: number;
1540 cursor: string;
1541 durationMs: number;
1542 reason: string;
1543 repeat: boolean;
1544 }
1545
1546 /** Write one finished pull run. The table's CHECKs refuse what `apps/pull.ts` should already have. */
1547 export function createPullRun(db: Database, input: CreatePullRunInput): PullRun {
1548 const runId = id();
1549 db.query(
1550 `INSERT INTO pull_run (id, app_id, state, written, skipped, failed, cursor, duration_ms, reason, repeat)
1551 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)`,
1552 ).run(
1553 runId,
1554 input.appId,
1555 input.state,
1556 input.written,
1557 input.skipped,
1558 input.failed,
1559 input.cursor,
1560 input.durationMs,
1561 input.reason.slice(0, 500),
1562 input.repeat ? 1 : 0,
1563 );
1564 return toPullRun(
1565 db.query<PullRunRow, [string]>("SELECT * FROM pull_run WHERE id = ?1").get(runId)!,
1566 );
1567 }
1568
1569 /** The cursor of an app's last good pull, from `pull_run_app_ok`; undefined before the first. */
1570 export function lastPullCursor(db: Database, appId: string): string | undefined {
1571 return (
1572 db
1573 .query<{ cursor: string }, [string]>(
1574 `SELECT cursor FROM pull_run
1575 WHERE app_id = ?1 AND state = 'ok' ORDER BY id DESC LIMIT 1`,
1576 )
1577 .get(appId)?.cursor ?? undefined
1578 );
1579 }
1580
1581 /** An app's pull runs, newest first, from `pull_run_app`. */
1582 export function listPullRuns(db: Database, appId: string, limit = 50): PullRun[] {
1583 return db
1584 .query<PullRunRow, [string, number]>(
1585 `SELECT * FROM pull_run WHERE app_id = ?1 ORDER BY id DESC LIMIT ?2`,
1586 )
1587 .all(appId, Math.max(1, Math.min(500, Math.trunc(limit))))
1588 .map(toPullRun);
1589 }
test/apps-pull.test.ts added +154
@@ -0,0 +1,154 @@
1 /**
2 * An adapter's `/pull` answer is judged at the boundary and every pull leaves
3 * a run of kind `pull` (#10042, docs/ADAPTERS.md): a failing answer is
4 * recorded failed with its reason, a repeated cursor is recorded and writes
5 * nothing twice, and out-of-range counts or a multi-line cursor are refused.
6 * No network: answers are strings.
7 */
8
9 import { Database } from "bun:sqlite";
10 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
11 import {
12 parsePullAnswer,
13 PULL_BODY_MAX,
14 recordPull,
15 REPEAT_REASON,
16 type PullAnswer,
17 } from "../src/apps/pull.ts";
18 import { lastPullCursor, listPullRuns } from "../src/core/runs.ts";
19 import { migrationFiles } from "../src/core/schema.ts";
20 import { migrate } from "../src/kernel/migrate.ts";
21
22 let db: Database;
23 beforeEach(() => {
24 db = new Database(":memory:");
25 migrate(db, migrationFiles);
26 });
27 afterEach(() => db.close());
28
29 const body = (over: Record<string, unknown> = {}) =>
30 JSON.stringify({ pulled: 12, skipped: 3, cursor: "2026-11-09T07:00:00Z", errors: 0, ...over });
31
32 const reasonOf = (answer: PullAnswer): string => (answer.ok ? "" : answer.reason);
33
34 describe("parsePullAnswer", () => {
35 test("a good answer gives its four fields", () => {
36 expect(parsePullAnswer(200, body())).toEqual({
37 ok: true,
38 pulled: 12,
39 skipped: 3,
40 errors: 0,
41 cursor: "2026-11-09T07:00:00Z",
42 });
43 // The edges of the range are inside it.
44 expect(parsePullAnswer(204, body({ pulled: 0, skipped: 1_000_000 })).ok).toBe(true);
45 expect(parsePullAnswer(200, body({ cursor: "x".repeat(200) })).ok).toBe(true);
46 });
47
48 test("out-of-range counts are refused, each naming its field", () => {
49 expect(reasonOf(parsePullAnswer(200, body({ pulled: 1_000_001 })))).toBe(
50 "pulled is not an integer from 0 to 1000000",
51 );
52 expect(reasonOf(parsePullAnswer(200, body({ skipped: -1 })))).toBe(
53 "skipped is not an integer from 0 to 1000000",
54 );
55 expect(reasonOf(parsePullAnswer(200, body({ errors: 1.5 })))).toBe(
56 "errors is not an integer from 0 to 1000000",
57 );
58 expect(parsePullAnswer(200, body({ pulled: "12" })).ok).toBe(false);
59 expect(parsePullAnswer(200, '{"pulled":1e400,"skipped":0,"cursor":"a","errors":0}').ok).toBe(false);
60 });
61
62 test("a multi-line, over-long or unprintable cursor is refused", () => {
63 const why = "cursor is not one line of at most 200 printable characters";
64 expect(reasonOf(parsePullAnswer(200, body({ cursor: "page-1\npage-2" })))).toBe(why);
65 expect(reasonOf(parsePullAnswer(200, body({ cursor: "a\rb" })))).toBe(why);
66 expect(reasonOf(parsePullAnswer(200, body({ cursor: "a
b" })))).toBe(why);
67 expect(reasonOf(parsePullAnswer(200, body({ cursor: "a\u0000b" })))).toBe(why);
68 expect(reasonOf(parsePullAnswer(200, body({ cursor: "x".repeat(201) })))).toBe(why);
69 expect(reasonOf(parsePullAnswer(200, body({ cursor: 7 })))).toBe(why);
70 });
71
72 test("a non-2xx status, an oversized body or a wrong shape is a failed pull with a reason", () => {
73 expect(reasonOf(parsePullAnswer(500, body()))).toBe("status 500 is not 2xx");
74 expect(reasonOf(parsePullAnswer(301, body()))).toBe("status 301 is not 2xx");
75 const huge = body({ cursor: "c" }).replace("}", `,"pad":"${"x".repeat(PULL_BODY_MAX)}"}`);
76 expect(reasonOf(parsePullAnswer(200, huge))).toBe("body is over 65536 bytes");
77 expect(reasonOf(parsePullAnswer(200, "<html>"))).toBe("body is not JSON");
78 expect(reasonOf(parsePullAnswer(200, "[1,2]"))).toBe("body is not a JSON object");
79 expect(reasonOf(parsePullAnswer(200, "null"))).toBe("body is not a JSON object");
80 expect(reasonOf(parsePullAnswer(200, body({ extra: 1 })))).toBe('unexpected field "extra"');
81 expect(reasonOf(parsePullAnswer(200, '{"pulled":1,"skipped":0,"cursor":"a"}'))).toBe("errors is missing");
82 });
83 });
84
85 describe("recordPull", () => {
86 test("a failing answer is recorded as failed with its reason, and keeps the last good cursor", () => {
87 recordPull(db, "app1", { ...parsePullAnswer(200, body({ cursor: "c1" })), durationMs: 40 });
88 const run = recordPull(db, "app1", { ...parsePullAnswer(503, "busy"), durationMs: 12.6 });
89 expect(run).toMatchObject({
90 kind: "pull",
91 appId: "app1",
92 state: "failed",
93 reason: "status 503 is not 2xx",
94 written: 0,
95 cursor: "",
96 durationMs: 13,
97 repeat: false,
98 });
99 expect(listPullRuns(db, "app1").map((r) => r.state)).toEqual(["failed", "ok"]);
100 expect(lastPullCursor(db, "app1")).toBe("c1");
101 });
102
103 test("a repeated cursor is recorded and writes nothing twice", () => {
104 const first = recordPull(db, "app1", { ...parsePullAnswer(200, body({ cursor: "c1" })), durationMs: 50 });
105 expect(first).toMatchObject({ state: "ok", written: 12, skipped: 3, failed: 0, cursor: "c1", durationMs: 50, repeat: false });
106 // The forge repeats the pull and the adapter answers the same cursor with the same counts.
107 const again = recordPull(db, "app1", { ...parsePullAnswer(200, body({ cursor: "c1" })), durationMs: 30 });
108 expect(again).toMatchObject({ state: "ok", written: 0, skipped: 15, cursor: "c1", repeat: true, reason: REPEAT_REASON });
109 const runs = listPullRuns(db, "app1");
110 expect(runs).toHaveLength(2);
111 expect(runs.reduce((n, r) => n + r.written, 0)).toBe(12);
112 // A failure between does not reset what counts as the last good cursor.
113 recordPull(db, "app1", { ...parsePullAnswer(200, "oops"), durationMs: 1 });
114 expect(recordPull(db, "app1", { ...parsePullAnswer(200, body({ cursor: "c1" })), durationMs: 1 }).repeat).toBe(true);
115 // A new cursor writes again; another app's cursor is its own.
116 expect(recordPull(db, "app1", { ...parsePullAnswer(200, body({ cursor: "c2", pulled: 4 })), durationMs: 1 }).written).toBe(4);
117 expect(recordPull(db, "app2", { ...parsePullAnswer(200, body({ cursor: "c1" })), durationMs: 1 }).written).toBe(12);
118 });
119
120 test("a repeat's skipped stays inside the range", () => {
121 recordPull(db, "a", { ...parsePullAnswer(200, body({ cursor: "c", pulled: 1_000_000, skipped: 1_000_000 })), durationMs: 0 });
122 const again = recordPull(db, "a", { ...parsePullAnswer(200, body({ cursor: "c", pulled: 1_000_000, skipped: 1_000_000 })), durationMs: 0 });
123 expect(again).toMatchObject({ written: 0, skipped: 1_000_000 });
124 });
125
126 test("an app's runs and its last good cursor are read from an index, never a scan or a sort", () => {
127 // The plan of the SQL the readers really run, its parameters bound.
128 const plan = (read: () => void): string => {
129 const original = db.query.bind(db);
130 const sqls: string[] = [];
131 (db as unknown as { query: unknown }).query = (sql: string) => (sqls.push(sql), original(sql));
132 try {
133 read();
134 } finally {
135 (db as unknown as { query: unknown }).query = original;
136 }
137 return sqls
138 .flatMap((sql) => {
139 const n = Math.max(0, ...[...sql.matchAll(/\?(\d+)/g)].map((m) => Number(m[1])));
140 return original(`EXPLAIN QUERY PLAN ${sql}`).all(...Array.from({ length: n }, () => "x"));
141 })
142 .map((r) => (r as { detail: string }).detail)
143 .join("\n");
144 };
145 const last = plan(() => void lastPullCursor(db, "x"));
146 expect(last).toContain("pull_run_app_ok");
147 const list = plan(() => void listPullRuns(db, "x", 5));
148 expect(list).toContain("pull_run_app");
149 for (const p of [last, list]) {
150 expect(p).not.toMatch(/^SCAN (?!.*USING (COVERING )?INDEX)/m);
151 expect(p).not.toContain("TEMP B-TREE");
152 }
153 });
154 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs