49c292c · c01e0a7
@@ -114,3 +114,11 @@
For a bad answer the forge:
A good answer is recorded as a run with its counts, its cursor and its
duration.
The schedule is kept by `src/apps/pull_schedule.ts`: once a minute the forge
pulls each app whose production is running, whose `forge.json` on the default
branch has an `adapter` block, and whose last pull, good or failed, is at least
`schedule.every` seconds old. A sleeping app is never woken to be pulled; it is
pulled on the first minute it is running and due. The request goes over
loopback, waits at most 10 seconds and reads at most 64 KB. Two pulls of the
same app never overlap.
@@ -4,7 +4,7 @@
 * raw body, it answers four checked fields or a failed pull with a short
 * reason naming what was wrong. `recordPull` writes the outcome as a run of
 * kind `pull` through `core/runs.ts`. Nothing here reaches the network: the
 * caller that sends the request and times it comes later (#437).
 * caller that sends the request and times it is `pull_schedule.ts` (#10043).
 */
import type { Database } from "bun:sqlite";
@@ -0,0 +1,176 @@
/**
 * An adapter pulled on its schedule (#10043, docs/ADAPTERS.md): on a tick the
 * forge POSTs `/pull` to each app whose production is running, whose
 * `forge.json` on the default branch has an `adapter` block
 * (`apps/adapter.ts`), and whose last pull is at least `schedule.every`
 * seconds old. The answer is judged by `parsePullAnswer` and kept by
 * `recordPull` (`apps/pull.ts`). The request goes over the same loopback reach
 * the scheduled tools use (`feature_tools.ts`): 64 KB, 10 s. An app that is
 * not running is skipped, never woken; two pulls of one app never overlap.
 */
import type { Database } from "bun:sqlite";
import type { App } from "../app.ts";
import { listApps, type AppRecord } from "../core/apps.ts";
import { getRepoById } from "../core/repos.ts";
import { lastPullAt } from "../core/runs.ts";
import { resolveOid } from "../git/merge.ts";
import { repoDir } from "../git/repos.ts";
import { parseAdapter, type AdapterSpec } from "./adapter.ts";
import { readFileAt } from "./checkout.ts";
import { parsePullAnswer, PULL_BODY_MAX, recordPull, type PullAnswer } from "./pull.ts";
import { appsRuntime } from "./runtime.ts";
/** How long the forge waits for one `/pull` answer, as for a scheduled tool. */
export const PULL_WAIT_MS = 10_000;
/** Whether a pull is due: never pulled, or at least `everySeconds` since the last one (ms clock). */
export function isPullDue(lastAt: number | undefined, everySeconds: number, now: number): boolean {
  return lastAt === undefined || now - lastAt >= everySeconds * 1000;
}
/** What the forge heard from `POST /pull`: a status and the body as text, or why there was none. */
export type PullReply = { readonly status: number; readonly body: string } | { readonly error: string };
/** What `pullApp` needs from the running forge for one app; the seam a test replaces. */
export interface PullReach {
  /** The production port when it is running; undefined otherwise (asleep, stopped, never deployed). */
  port(): number | undefined;
  /** The `adapter` block the default branch declares now; undefined when it is not an adapter. */
  adapter(): Promise<AdapterSpec | undefined>;
  /** Send `POST /pull` to the app on `port`. */
  post(port: number): Promise<PullReply>;
}
const inFlight = new Set<string>();
/** The answer a reply is worth: a reply with no status is a failed pull naming why. */
export function answerOf(reply: PullReply): PullAnswer {
  if ("error" in reply) return { ok: false, reason: reply.error };
  return parsePullAnswer(reply.status, reply.body);
}
/**
 * Pull one app if it is a running adapter and due: whether a pull was made.
 * The checks run cheapest first — running, then in flight, then declared,
 * then due — so a sleeping app costs no git read and is never reached.
 */
export async function pullApp(db: Database, appId: string, reach: PullReach, now: () => number = Date.now): Promise<boolean> {
  const port = reach.port();
  if (port === undefined) return false;
  if (inFlight.has(appId)) return false;
  inFlight.add(appId);
  try {
    const adapter = await reach.adapter();
    if (adapter === undefined) return false;
    if (!isPullDue(lastPullAt(db, appId), adapter.schedule.every, now())) return false;
    const started = now();
    const answer = answerOf(await reach.post(port));
    recordPull(db, appId, { ...answer, durationMs: Math.max(0, now() - started) });
    return true;
  } finally {
    inFlight.delete(appId);
  }
}
/** POST `/pull` on loopback: at most 64 KB read (one byte over, so the judge says so), 10 s. */
export async function postPull(port: number, waitMs: number = PULL_WAIT_MS): Promise<PullReply> {
  try {
    const res = await fetch(`http://127.0.0.1:${port}/pull`, {
      method: "POST",
      headers: { "content-type": "application/json", accept: "application/json" },
      body: "{}",
      redirect: "manual",
      signal: AbortSignal.timeout(waitMs),
    });
    const reader = res.body?.getReader();
    const chunks: Uint8Array[] = [];
    let total = 0;
    while (reader !== undefined) {
      const { done, value } = await reader.read();
      if (done) break;
      total += value.length;
      if (total > PULL_BODY_MAX) {
        await reader.cancel();
        return { error: `body is over ${PULL_BODY_MAX} bytes` };
      }
      chunks.push(value);
    }
    return { status: res.status, body: Buffer.concat(chunks).toString("utf8") };
  } catch {
    return { error: "the app did not answer" };
  }
}
/** The adapter block by default-branch tip: a tick costs a git read only when a tip is new. */
const declaredByTip = new Map<string, AdapterSpec | undefined>();
const DECLARED_MAX = 256;
async function declaredAdapter(app: App, record: AppRecord): Promise<AdapterSpec | undefined> {
  const repo = getRepoById(app.db, record.repoId);
  if (repo === undefined) return undefined;
  const dir = repoDir(app.reposRoot, repo.ownerName, repo.name);
  const tip = await resolveOid(dir, `refs/heads/${repo.defaultBranch}`);
  if (tip === undefined) return undefined;
  if (declaredByTip.has(tip)) return declaredByTip.get(tip);
  const text = await readFileAt(dir, tip, "forge.json");
  let spec: AdapterSpec | undefined;
  if (text !== undefined) {
    try {
      const parsed: unknown = JSON.parse(text);
      if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) {
        const read = parseAdapter((parsed as Record<string, unknown>)["adapter"]);
        spec = read.ok ? read.value : undefined; // a bad block is the deploy's to report
      }
    } catch {
      spec = undefined;
    }
  }
  if (declaredByTip.size >= DECLARED_MAX) declaredByTip.clear();
  declaredByTip.set(tip, spec);
  return spec;
}
/** What `pullApp` reaches for one app in the real runtime: only a production that is running, never woken. */
export function runtimePullReach(app: App, record: AppRecord): PullReach {
  return {
    port: () => {
      const status = appsRuntime(app)?.envStatus(record, "production");
      return status?.state === "running" ? status.port : undefined;
    },
    adapter: () => declaredAdapter(app, record),
    post: (port) => postPull(port),
  };
}
/** Every app, one after another; the apps pulled. `reachOf` is the seam a test replaces. */
export async function pullDue(
  app: App,
  now: () => number = Date.now,
  reachOf: (record: AppRecord) => PullReach = (record) => runtimePullReach(app, record),
): Promise<string[]> {
  const pulled: string[] = [];
  for (const record of listApps(app.db)) {
    try {
      if (await pullApp(app.db, record.id, reachOf(record), now)) pulled.push(record.slug);
    } catch (error) {
      app.log.warn("could not pull an adapter", { app: record.slug, error: String(error) });
    }
  }
  return pulled;
}
const TICK_MS = 60_000;
/** The timer. Off under `bun test` (NODE_ENV=test) so no sweep outlives a test; a test calls `pullDue`. */
export function startPullSchedule(app: App, tickMs: number = TICK_MS): { stop(): void } | undefined {
  if (process.env["NODE_ENV"] === "test") return undefined;
  let running = false;
  const tick = setInterval(() => {
    if (running) return;
    running = true;
    void pullDue(app).finally(() => (running = false));
  }, tickMs);
  tick.unref();
  return { stop: () => clearInterval(tick) };
}
@@ -1587,3 +1587,15 @@
export function listPullRuns(db: Database, appId: string, limit = 50): PullRun[]
    .all(appId, Math.max(1, Math.min(500, Math.trunc(limit))))
    .map(toPullRun);
}
/** When an app's last pull, good or failed, was recorded, in ms since the epoch; undefined before the first. */
export function lastPullAt(db: Database, appId: string): number | undefined {
  const row = db
    .query<{ created_at: string }, [string]>(
      `SELECT created_at FROM pull_run WHERE app_id = ?1 ORDER BY id DESC LIMIT 1`,
    )
    .get(appId);
  if (row === null) return undefined;
  const at = Date.parse(row.created_at);
  return Number.isFinite(at) ? at : undefined;
}
@@ -471,6 +471,7 @@
import { captureStore, loopWatch, perfRecord, readLedger } from "../perf/live.ts
import { classOf, countryOf, repoOf, signatureOf } from "../perf/traffic.ts";
import { createTrafficWriter } from "../perf/traffic_store.ts";
import { startFeatureSchedule } from "../apps/feature_schedule.ts";
import { startPullSchedule } from "../apps/pull_schedule.ts";
import { startGithubPull } from "../perf/github_traffic.ts";
import { appSourcesPage } from "../web/pages/sources.ts";
import { createViewWriter } from "../perf/views_store.ts";
@@ -2312,6 +2313,7 @@
export async function serve(app: App): Promise<Servers> {
  const viewWriter = createViewWriter(app.db, app.log);
  const githubPull = startGithubPull(app.db, app.log);
  const featureSchedule = startFeatureSchedule(app);
  const pullSchedule = startPullSchedule(app);
  const readWriter = createReadWriter(app.db, readLedger, app.log);
  const counted = (
    owner: App,
@@ -2635,6 +2637,7 @@
export async function serve(app: App): Promise<Servers> {
      viewWriter.stop();
      githubPull?.stop();
      featureSchedule?.stop();
      pullSchedule?.stop();
      readWriter.stop();
      // A child's hooks stop answering before its database closes.
      for (const key of [...new Set([...childHttps.keys(), ...childInternals.keys()])]) letGo(key);
@@ -0,0 +1,180 @@
/**
 * The forge pulls a running adapter on its schedule (#10043): a due adapter
 * is pulled once, one not yet due is not, a sleeping app is never reached, a
 * failing answer is recorded failed, and two pulls of one app never overlap.
 * The clock and the reach are fakes; one test posts to a loopback server to
 * prove the real request's path, size cap and shape.
 */
import { Database } from "bun:sqlite";
import { afterAll, afterEach, beforeEach, describe, expect, test } from "bun:test";
import { PULL_BODY_MAX } from "../src/apps/pull.ts";
import {
  isPullDue,
  postPull,
  pullApp,
  pullDue,
  startPullSchedule,
  type PullReach,
  type PullReply,
} from "../src/apps/pull_schedule.ts";
import { listPullRuns } from "../src/core/runs.ts";
import { migrationFiles } from "../src/core/schema.ts";
import { migrate } from "../src/kernel/migrate.ts";
const EVERY = 900; // seconds
const good = JSON.stringify({ pulled: 4, skipped: 1, cursor: "c1", errors: 0 });
let db: Database;
let posts: number;
let reply: () => Promise<PullReply>;
const reach = (over: Partial<PullReach> = {}): PullReach => ({
  port: () => 4321,
  adapter: async () => ({ schedule: { every: EVERY } }),
  post: async () => {
    posts += 1;
    return reply();
  },
  ...over,
});
/** A pull recorded `agoMs` before the real clock, so "last pull" is under the test's control. */
const pulledAgo = (appId: string, agoMs: number): void => {
  db.run(
    `INSERT INTO pull_run (id, app_id, state, cursor, created_at) VALUES (?1, ?2, 'ok', 'c0', ?3)`,
    ["00000000000000000000000000", appId, new Date(Date.now() - agoMs).toISOString()],
  );
};
beforeEach(() => {
  db = new Database(":memory:");
  migrate(db, migrationFiles);
  posts = 0;
  reply = async () => ({ status: 200, body: good });
});
afterEach(() => db.close());
describe("isPullDue", () => {
  test("never pulled is due; inside the interval is not; at it is", () => {
    expect(isPullDue(undefined, EVERY, 5)).toBe(true);
    expect(isPullDue(1000, EVERY, 1000 + EVERY * 1000 - 1)).toBe(false);
    expect(isPullDue(1000, EVERY, 1000 + EVERY * 1000)).toBe(true);
  });
});
describe("pullApp", () => {
  test("a due adapter is pulled once and its answer recorded; the next tick inside the interval does nothing", async () => {
    let t = Date.now();
    const clock = () => t;
    expect(await pullApp(db, "a1", reach(), clock)).toBe(true);
    expect(posts).toBe(1);
    const [run] = listPullRuns(db, "a1");
    expect(run).toMatchObject({ state: "ok", written: 4, skipped: 1, cursor: "c1", failed: 0 });
    t += 60_000; // one tick later: not due
    expect(await pullApp(db, "a1", reach(), clock)).toBe(false);
    expect(posts).toBe(1);
    t += EVERY * 1000; // a whole interval later: due again
    expect(await pullApp(db, "a1", reach(), clock)).toBe(true);
    expect(posts).toBe(2);
  });
  test("an adapter pulled less than `every` seconds ago is not pulled", async () => {
    pulledAgo("a1", (EVERY - 1) * 1000);
    expect(await pullApp(db, "a1", reach())).toBe(false);
    expect(posts).toBe(0);
    expect(listPullRuns(db, "a1")).toHaveLength(1);
  });
  test("a failing answer is recorded as failed with its reason", async () => {
    reply = async () => ({ status: 500, body: "no" });
    expect(await pullApp(db, "a1", reach())).toBe(true);
    const [run] = listPullRuns(db, "a1");
    expect(run?.state).toBe("failed");
    expect(run?.reason).toBe("status 500 is not 2xx");
  });
  test("no answer at all is recorded as failed", async () => {
    reply = async () => ({ error: "the app did not answer" });
    await pullApp(db, "a1", reach());
    expect(listPullRuns(db, "a1")[0]).toMatchObject({ state: "failed", reason: "the app did not answer" });
  });
  test("a sleeping app is never reached, and its manifest is not read", async () => {
    let read = false;
    const asleep = reach({ port: () => undefined, adapter: async () => ((read = true), { schedule: { every: EVERY } }) });
    expect(await pullApp(db, "a1", asleep)).toBe(false);
    expect(posts).toBe(0);
    expect(read).toBe(false);
    expect(listPullRuns(db, "a1")).toEqual([]);
  });
  test("an app without an adapter block is not pulled", async () => {
    expect(await pullApp(db, "a1", reach({ adapter: async () => undefined }))).toBe(false);
    expect(posts).toBe(0);
  });
  test("a second pull of the same app while one is in flight does not overlap it", async () => {
    let release!: () => void;
    reply = () => new Promise((resolve) => (release = () => resolve({ status: 200, body: good })));
    const first = pullApp(db, "a1", reach());
    await Bun.sleep(0);
    expect(await pullApp(db, "a1", reach())).toBe(false);
    release();
    expect(await first).toBe(true);
    expect(posts).toBe(1);
    expect(listPullRuns(db, "a1")).toHaveLength(1);
  });
});
describe("pullDue", () => {
  test("sweeps every app: the due one pulled once, the one not due left, one that throws logged", async () => {
    db.run("PRAGMA foreign_keys = OFF");
    for (const [id, slug] of [["a1", "due"], ["a2", "fresh"], ["a3", "broken"]] as const)
      db.run("INSERT INTO app (id, repo_id, slug, access) VALUES (?1, ?2, ?3, 'private')", [id, `repo-${id}`, slug]);
    pulledAgo("a2", 1000);
    const warned: string[] = [];
    const forge = { db, log: { warn: (m: string) => warned.push(m) } } as never;
    const reachOf = (r: { id: string }): PullReach =>
      r.id === "a3" ? reach({ adapter: () => Promise.reject(new Error("boom")) }) : reach();
    expect(await pullDue(forge, Date.now, reachOf)).toEqual(["due"]);
    expect(posts).toBe(1);
    expect(warned).toEqual(["could not pull an adapter"]);
    expect(await pullDue(forge, Date.now, reachOf)).toEqual([]);
    expect(posts).toBe(1);
  });
});
describe("postPull", () => {
  const seen: { method: string; path: string }[] = [];
  let answer: () => Response = () => new Response(good);
  const server = Bun.serve({
    port: 0,
    hostname: "127.0.0.1",
    fetch: (req) => {
      seen.push({ method: req.method, path: new URL(req.url).pathname });
      return answer();
    },
  });
  const port = server.port ?? 0;
  afterAll(() => server.stop(true));
  test("POSTs /pull on loopback and hands back the status and body", async () => {
    expect(await postPull(port)).toEqual({ status: 200, body: good });
    expect(seen).toEqual([{ method: "POST", path: "/pull" }]);
  });
  test("a body over 64 KB is cut off and named", async () => {
    answer = () => new Response("x".repeat(PULL_BODY_MAX + 1));
    expect(await postPull(port)).toEqual({ error: `body is over ${PULL_BODY_MAX} bytes` });
  });
  test("an app that answers too slowly did not answer", async () => {
    answer = () => new Response(new ReadableStream({ start() {} }));
    expect(await postPull(port, 50)).toEqual({ error: "the app did not answer" });
  });
});
test("the timer does not start under the test harness", () => {
  expect(startPullSchedule({} as never)).toBeUndefined();
});
		@@ -114,3 +114,11 @@ For a bad answer the forge:
114	114
115	115	A good answer is recorded as a run with its counts, its cursor and its
116	116	duration.
	117
	118	The schedule is kept by `src/apps/pull_schedule.ts`: once a minute the forge
	119	pulls each app whose production is running, whose `forge.json` on the default
	120	branch has an `adapter` block, and whose last pull, good or failed, is at least
	121	`schedule.every` seconds old. A sleeping app is never woken to be pulled; it is
	122	pulled on the first minute it is running and due. The request goes over
	123	loopback, waits at most 10 seconds and reads at most 64 KB. Two pulls of the
	124	same app never overlap.
		@@ -4,7 +4,7 @@
4	4	* raw body, it answers four checked fields or a failed pull with a short
5	5	* reason naming what was wrong. `recordPull` writes the outcome as a run of
6	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).
	7	* caller that sends the request and times it is `pull_schedule.ts` (#10043).
8	8	*/
9	9
10	10	import type { Database } from "bun:sqlite";
		@@ -0,0 +1,176 @@
	1	/**
	2	* An adapter pulled on its schedule (#10043, docs/ADAPTERS.md): on a tick the
	3	* forge POSTs `/pull` to each app whose production is running, whose
	4	* `forge.json` on the default branch has an `adapter` block
	5	* (`apps/adapter.ts`), and whose last pull is at least `schedule.every`
	6	* seconds old. The answer is judged by `parsePullAnswer` and kept by
	7	* `recordPull` (`apps/pull.ts`). The request goes over the same loopback reach
	8	* the scheduled tools use (`feature_tools.ts`): 64 KB, 10 s. An app that is
	9	* not running is skipped, never woken; two pulls of one app never overlap.
	10	*/
	11
	12	import type { Database } from "bun:sqlite";
	13	import type { App } from "../app.ts";
	14	import { listApps, type AppRecord } from "../core/apps.ts";
	15	import { getRepoById } from "../core/repos.ts";
	16	import { lastPullAt } from "../core/runs.ts";
	17	import { resolveOid } from "../git/merge.ts";
	18	import { repoDir } from "../git/repos.ts";
	19	import { parseAdapter, type AdapterSpec } from "./adapter.ts";
	20	import { readFileAt } from "./checkout.ts";
	21	import { parsePullAnswer, PULL_BODY_MAX, recordPull, type PullAnswer } from "./pull.ts";
	22	import { appsRuntime } from "./runtime.ts";
	23
	24	/** How long the forge waits for one `/pull` answer, as for a scheduled tool. */
	25	export const PULL_WAIT_MS = 10_000;
	26
	27	/** Whether a pull is due: never pulled, or at least `everySeconds` since the last one (ms clock). */
	28	export function isPullDue(lastAt: number \| undefined, everySeconds: number, now: number): boolean {
	29	return lastAt === undefined \|\| now - lastAt >= everySeconds * 1000;
	30	}
	31
	32	/** What the forge heard from `POST /pull`: a status and the body as text, or why there was none. */
	33	export type PullReply = { readonly status: number; readonly body: string } \| { readonly error: string };
	34
	35	/** What `pullApp` needs from the running forge for one app; the seam a test replaces. */
	36	export interface PullReach {
	37	/** The production port when it is running; undefined otherwise (asleep, stopped, never deployed). */
	38	port(): number \| undefined;
	39	/** The `adapter` block the default branch declares now; undefined when it is not an adapter. */
	40	adapter(): Promise<AdapterSpec \| undefined>;
	41	/** Send `POST /pull` to the app on `port`. */
	42	post(port: number): Promise<PullReply>;
	43	}
	44
	45	const inFlight = new Set<string>();
	46
	47	/** The answer a reply is worth: a reply with no status is a failed pull naming why. */
	48	export function answerOf(reply: PullReply): PullAnswer {
	49	if ("error" in reply) return { ok: false, reason: reply.error };
	50	return parsePullAnswer(reply.status, reply.body);
	51	}
	52
	53	/**
	54	* Pull one app if it is a running adapter and due: whether a pull was made.
	55	* The checks run cheapest first — running, then in flight, then declared,
	56	* then due — so a sleeping app costs no git read and is never reached.
	57	*/
	58	export async function pullApp(db: Database, appId: string, reach: PullReach, now: () => number = Date.now): Promise<boolean> {
	59	const port = reach.port();
	60	if (port === undefined) return false;
	61	if (inFlight.has(appId)) return false;
	62	inFlight.add(appId);
	63	try {
	64	const adapter = await reach.adapter();
	65	if (adapter === undefined) return false;
	66	if (!isPullDue(lastPullAt(db, appId), adapter.schedule.every, now())) return false;
	67	const started = now();
	68	const answer = answerOf(await reach.post(port));
	69	recordPull(db, appId, { ...answer, durationMs: Math.max(0, now() - started) });
	70	return true;
	71	} finally {
	72	inFlight.delete(appId);
	73	}
	74	}
	75
	76	/** POST `/pull` on loopback: at most 64 KB read (one byte over, so the judge says so), 10 s. */
	77	export async function postPull(port: number, waitMs: number = PULL_WAIT_MS): Promise<PullReply> {
	78	try {
	79	const res = await fetch(`http://127.0.0.1:${port}/pull`, {
	80	method: "POST",
	81	headers: { "content-type": "application/json", accept: "application/json" },
	82	body: "{}",
	83	redirect: "manual",
	84	signal: AbortSignal.timeout(waitMs),
	85	});
	86	const reader = res.body?.getReader();
	87	const chunks: Uint8Array[] = [];
	88	let total = 0;
	89	while (reader !== undefined) {
	90	const { done, value } = await reader.read();
	91	if (done) break;
	92	total += value.length;
	93	if (total > PULL_BODY_MAX) {
	94	await reader.cancel();
	95	return { error: `body is over ${PULL_BODY_MAX} bytes` };
	96	}
	97	chunks.push(value);
	98	}
	99	return { status: res.status, body: Buffer.concat(chunks).toString("utf8") };
	100	} catch {
	101	return { error: "the app did not answer" };
	102	}
	103	}
	104
	105	/** The adapter block by default-branch tip: a tick costs a git read only when a tip is new. */
	106	const declaredByTip = new Map<string, AdapterSpec \| undefined>();
	107	const DECLARED_MAX = 256;
	108
	109	async function declaredAdapter(app: App, record: AppRecord): Promise<AdapterSpec \| undefined> {
	110	const repo = getRepoById(app.db, record.repoId);
	111	if (repo === undefined) return undefined;
	112	const dir = repoDir(app.reposRoot, repo.ownerName, repo.name);
	113	const tip = await resolveOid(dir, `refs/heads/${repo.defaultBranch}`);
	114	if (tip === undefined) return undefined;
	115	if (declaredByTip.has(tip)) return declaredByTip.get(tip);
	116	const text = await readFileAt(dir, tip, "forge.json");
	117	let spec: AdapterSpec \| undefined;
	118	if (text !== undefined) {
	119	try {
	120	const parsed: unknown = JSON.parse(text);
	121	if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) {
	122	const read = parseAdapter((parsed as Record<string, unknown>)["adapter"]);
	123	spec = read.ok ? read.value : undefined; // a bad block is the deploy's to report
	124	}
	125	} catch {
	126	spec = undefined;
	127	}
	128	}
	129	if (declaredByTip.size >= DECLARED_MAX) declaredByTip.clear();
	130	declaredByTip.set(tip, spec);
	131	return spec;
	132	}
	133
	134	/** What `pullApp` reaches for one app in the real runtime: only a production that is running, never woken. */
	135	export function runtimePullReach(app: App, record: AppRecord): PullReach {
	136	return {
	137	port: () => {
	138	const status = appsRuntime(app)?.envStatus(record, "production");
	139	return status?.state === "running" ? status.port : undefined;
	140	},
	141	adapter: () => declaredAdapter(app, record),
	142	post: (port) => postPull(port),
	143	};
	144	}
	145
	146	/** Every app, one after another; the apps pulled. `reachOf` is the seam a test replaces. */
	147	export async function pullDue(
	148	app: App,
	149	now: () => number = Date.now,
	150	reachOf: (record: AppRecord) => PullReach = (record) => runtimePullReach(app, record),
	151	): Promise<string[]> {
	152	const pulled: string[] = [];
	153	for (const record of listApps(app.db)) {
	154	try {
	155	if (await pullApp(app.db, record.id, reachOf(record), now)) pulled.push(record.slug);
	156	} catch (error) {
	157	app.log.warn("could not pull an adapter", { app: record.slug, error: String(error) });
	158	}
	159	}
	160	return pulled;
	161	}
	162
	163	const TICK_MS = 60_000;
	164
	165	/** The timer. Off under `bun test` (NODE_ENV=test) so no sweep outlives a test; a test calls `pullDue`. */
	166	export function startPullSchedule(app: App, tickMs: number = TICK_MS): { stop(): void } \| undefined {
	167	if (process.env["NODE_ENV"] === "test") return undefined;
	168	let running = false;
	169	const tick = setInterval(() => {
	170	if (running) return;
	171	running = true;
	172	void pullDue(app).finally(() => (running = false));
	173	}, tickMs);
	174	tick.unref();
	175	return { stop: () => clearInterval(tick) };
	176	}
		@@ -1587,3 +1587,15 @@ export function listPullRuns(db: Database, appId: string, limit = 50): PullRun[]
1587	1587	.all(appId, Math.max(1, Math.min(500, Math.trunc(limit))))
1588	1588	.map(toPullRun);
1589	1589	}
	1590
	1591	/** When an app's last pull, good or failed, was recorded, in ms since the epoch; undefined before the first. */
	1592	export function lastPullAt(db: Database, appId: string): number \| undefined {
	1593	const row = db
	1594	.query<{ created_at: string }, [string]>(
	1595	`SELECT created_at FROM pull_run WHERE app_id = ?1 ORDER BY id DESC LIMIT 1`,
	1596	)
	1597	.get(appId);
	1598	if (row === null) return undefined;
	1599	const at = Date.parse(row.created_at);
	1600	return Number.isFinite(at) ? at : undefined;
	1601	}
		@@ -471,6 +471,7 @@ import { captureStore, loopWatch, perfRecord, readLedger } from "../perf/live.ts
471	471	import { classOf, countryOf, repoOf, signatureOf } from "../perf/traffic.ts";
472	472	import { createTrafficWriter } from "../perf/traffic_store.ts";
473	473	import { startFeatureSchedule } from "../apps/feature_schedule.ts";
	474	import { startPullSchedule } from "../apps/pull_schedule.ts";
474	475	import { startGithubPull } from "../perf/github_traffic.ts";
475	476	import { appSourcesPage } from "../web/pages/sources.ts";
476	477	import { createViewWriter } from "../perf/views_store.ts";
		@@ -2312,6 +2313,7 @@ export async function serve(app: App): Promise<Servers> {
2312	2313	const viewWriter = createViewWriter(app.db, app.log);
2313	2314	const githubPull = startGithubPull(app.db, app.log);
2314	2315	const featureSchedule = startFeatureSchedule(app);
	2316	const pullSchedule = startPullSchedule(app);
2315	2317	const readWriter = createReadWriter(app.db, readLedger, app.log);
2316	2318	const counted = (
2317	2319	owner: App,
		@@ -2635,6 +2637,7 @@ export async function serve(app: App): Promise<Servers> {
2635	2637	viewWriter.stop();
2636	2638	githubPull?.stop();
2637	2639	featureSchedule?.stop();
	2640	pullSchedule?.stop();
2638	2641	readWriter.stop();
2639	2642	// A child's hooks stop answering before its database closes.
2640	2643	for (const key of [...new Set([...childHttps.keys(), ...childInternals.keys()])]) letGo(key);
		@@ -0,0 +1,180 @@
	1	/**
	2	* The forge pulls a running adapter on its schedule (#10043): a due adapter
	3	* is pulled once, one not yet due is not, a sleeping app is never reached, a
	4	* failing answer is recorded failed, and two pulls of one app never overlap.
	5	* The clock and the reach are fakes; one test posts to a loopback server to
	6	* prove the real request's path, size cap and shape.
	7	*/
	8
	9	import { Database } from "bun:sqlite";
	10	import { afterAll, afterEach, beforeEach, describe, expect, test } from "bun:test";
	11	import { PULL_BODY_MAX } from "../src/apps/pull.ts";
	12	import {
	13	isPullDue,
	14	postPull,
	15	pullApp,
	16	pullDue,
	17	startPullSchedule,
	18	type PullReach,
	19	type PullReply,
	20	} from "../src/apps/pull_schedule.ts";
	21	import { listPullRuns } from "../src/core/runs.ts";
	22	import { migrationFiles } from "../src/core/schema.ts";
	23	import { migrate } from "../src/kernel/migrate.ts";
	24
	25	const EVERY = 900; // seconds
	26	const good = JSON.stringify({ pulled: 4, skipped: 1, cursor: "c1", errors: 0 });
	27
	28	let db: Database;
	29	let posts: number;
	30	let reply: () => Promise<PullReply>;
	31
	32	const reach = (over: Partial<PullReach> = {}): PullReach => ({
	33	port: () => 4321,
	34	adapter: async () => ({ schedule: { every: EVERY } }),
	35	post: async () => {
	36	posts += 1;
	37	return reply();
	38	},
	39	...over,
	40	});
	41
	42	/** A pull recorded `agoMs` before the real clock, so "last pull" is under the test's control. */
	43	const pulledAgo = (appId: string, agoMs: number): void => {
	44	db.run(
	45	`INSERT INTO pull_run (id, app_id, state, cursor, created_at) VALUES (?1, ?2, 'ok', 'c0', ?3)`,
	46	["00000000000000000000000000", appId, new Date(Date.now() - agoMs).toISOString()],
	47	);
	48	};
	49
	50	beforeEach(() => {
	51	db = new Database(":memory:");
	52	migrate(db, migrationFiles);
	53	posts = 0;
	54	reply = async () => ({ status: 200, body: good });
	55	});
	56	afterEach(() => db.close());
	57
	58	describe("isPullDue", () => {
	59	test("never pulled is due; inside the interval is not; at it is", () => {
	60	expect(isPullDue(undefined, EVERY, 5)).toBe(true);
	61	expect(isPullDue(1000, EVERY, 1000 + EVERY * 1000 - 1)).toBe(false);
	62	expect(isPullDue(1000, EVERY, 1000 + EVERY * 1000)).toBe(true);
	63	});
	64	});
	65
	66	describe("pullApp", () => {
	67	test("a due adapter is pulled once and its answer recorded; the next tick inside the interval does nothing", async () => {
	68	let t = Date.now();
	69	const clock = () => t;
	70	expect(await pullApp(db, "a1", reach(), clock)).toBe(true);
	71	expect(posts).toBe(1);
	72	const [run] = listPullRuns(db, "a1");
	73	expect(run).toMatchObject({ state: "ok", written: 4, skipped: 1, cursor: "c1", failed: 0 });
	74	t += 60_000; // one tick later: not due
	75	expect(await pullApp(db, "a1", reach(), clock)).toBe(false);
	76	expect(posts).toBe(1);
	77	t += EVERY * 1000; // a whole interval later: due again
	78	expect(await pullApp(db, "a1", reach(), clock)).toBe(true);
	79	expect(posts).toBe(2);
	80	});
	81
	82	test("an adapter pulled less than `every` seconds ago is not pulled", async () => {
	83	pulledAgo("a1", (EVERY - 1) * 1000);
	84	expect(await pullApp(db, "a1", reach())).toBe(false);
	85	expect(posts).toBe(0);
	86	expect(listPullRuns(db, "a1")).toHaveLength(1);
	87	});
	88
	89	test("a failing answer is recorded as failed with its reason", async () => {
	90	reply = async () => ({ status: 500, body: "no" });
	91	expect(await pullApp(db, "a1", reach())).toBe(true);
	92	const [run] = listPullRuns(db, "a1");
	93	expect(run?.state).toBe("failed");
	94	expect(run?.reason).toBe("status 500 is not 2xx");
	95	});
	96
	97	test("no answer at all is recorded as failed", async () => {
	98	reply = async () => ({ error: "the app did not answer" });
	99	await pullApp(db, "a1", reach());
	100	expect(listPullRuns(db, "a1")[0]).toMatchObject({ state: "failed", reason: "the app did not answer" });
	101	});
	102
	103	test("a sleeping app is never reached, and its manifest is not read", async () => {
	104	let read = false;
	105	const asleep = reach({ port: () => undefined, adapter: async () => ((read = true), { schedule: { every: EVERY } }) });
	106	expect(await pullApp(db, "a1", asleep)).toBe(false);
	107	expect(posts).toBe(0);
	108	expect(read).toBe(false);
	109	expect(listPullRuns(db, "a1")).toEqual([]);
	110	});
	111
	112	test("an app without an adapter block is not pulled", async () => {
	113	expect(await pullApp(db, "a1", reach({ adapter: async () => undefined }))).toBe(false);
	114	expect(posts).toBe(0);
	115	});
	116
	117	test("a second pull of the same app while one is in flight does not overlap it", async () => {
	118	let release!: () => void;
	119	reply = () => new Promise((resolve) => (release = () => resolve({ status: 200, body: good })));
	120	const first = pullApp(db, "a1", reach());
	121	await Bun.sleep(0);
	122	expect(await pullApp(db, "a1", reach())).toBe(false);
	123	release();
	124	expect(await first).toBe(true);
	125	expect(posts).toBe(1);
	126	expect(listPullRuns(db, "a1")).toHaveLength(1);
	127	});
	128	});
	129
	130	describe("pullDue", () => {
	131	test("sweeps every app: the due one pulled once, the one not due left, one that throws logged", async () => {
	132	db.run("PRAGMA foreign_keys = OFF");
	133	for (const [id, slug] of [["a1", "due"], ["a2", "fresh"], ["a3", "broken"]] as const)
	134	db.run("INSERT INTO app (id, repo_id, slug, access) VALUES (?1, ?2, ?3, 'private')", [id, `repo-${id}`, slug]);
	135	pulledAgo("a2", 1000);
	136	const warned: string[] = [];
	137	const forge = { db, log: { warn: (m: string) => warned.push(m) } } as never;
	138	const reachOf = (r: { id: string }): PullReach =>
	139	r.id === "a3" ? reach({ adapter: () => Promise.reject(new Error("boom")) }) : reach();
	140	expect(await pullDue(forge, Date.now, reachOf)).toEqual(["due"]);
	141	expect(posts).toBe(1);
	142	expect(warned).toEqual(["could not pull an adapter"]);
	143	expect(await pullDue(forge, Date.now, reachOf)).toEqual([]);
	144	expect(posts).toBe(1);
	145	});
	146	});
	147
	148	describe("postPull", () => {
	149	const seen: { method: string; path: string }[] = [];
	150	let answer: () => Response = () => new Response(good);
	151	const server = Bun.serve({
	152	port: 0,
	153	hostname: "127.0.0.1",
	154	fetch: (req) => {
	155	seen.push({ method: req.method, path: new URL(req.url).pathname });
	156	return answer();
	157	},
	158	});
	159	const port = server.port ?? 0;
	160	afterAll(() => server.stop(true));
	161
	162	test("POSTs /pull on loopback and hands back the status and body", async () => {
	163	expect(await postPull(port)).toEqual({ status: 200, body: good });
	164	expect(seen).toEqual([{ method: "POST", path: "/pull" }]);
	165	});
	166
	167	test("a body over 64 KB is cut off and named", async () => {
	168	answer = () => new Response("x".repeat(PULL_BODY_MAX + 1));
	169	expect(await postPull(port)).toEqual({ error: `body is over ${PULL_BODY_MAX} bytes` });
	170	});
	171
	172	test("an app that answers too slowly did not answer", async () => {
	173	answer = () => new Response(new ReadableStream({ start() {} }));
	174	expect(await postPull(port, 50)).toEqual({ error: "the app did not answer" });
	175	});
	176	});
	177
	178	test("the timer does not start under the test harness", () => {
	179	expect(startPullSchedule({} as never)).toBeUndefined();
	180	});