b07c8c2 · 229b96c
@@ -234,3 +234,34 @@
in `src/release/`:
  version that was ready, kept on the volume under `release/good/`. The forge
  that comes up shows a `release:rolled-back` card and pages the admins once.
  The next version that reads ready clears it.
## A forge beside one still draining (#6957)
bun:sqlite waits for the write lock on the one thread the forge serves from,
up to `busy_timeout` (15 s) a statement. 0.7.278 booted while 0.7.276 still
ran its gates and crew runs on the same database. It started its own work
beside them, and `/healthz` timed out for minutes behind the old pod's writes.
Three things keep that from happening again:
- **The new forge holds its work** (`src/runs/peer.ts`). Its listener
  answers at once. Its checks, merge queue, crew and ticks start only once
  no running run that another process holds has renewed its lease within
  30 s, or after 720 s (the pod's grace). The log says
  `another forge still works this database` when the wait starts and
  `the other forge let go of this database` when it ends.
- **Telemetry never waits on the lock** (`withoutWaiting` in
  `src/kernel/db.ts`). Route timings, usage counts and machine samples try
  the lock once, keep their rows and try again on the next flush. Every
  probe is counted, so without this a probe's own count would hold the next
  probe up.
- **The next stall names itself.** `event loop stalled lateMs=…` is a
  warning whenever the loop wakes more than a second late
  (`src/kernel/loop_lag.ts`, on from the first line of the boot). `slow
  statement held the thread ms=… sql=… stack=…` names any statement on the
  forge's handle that took 200 ms or more, with its callers
  (`src/kernel/slow_sql.ts`), at most ten lines a minute.
`test/boot-beside-peer.test.ts` boots `src/main.ts` beside a process that
holds the write lock 2.5 s of every 2.65 s and keeps a batch of 8 check
leases fresh. Every `/healthz` probe must answer within a second, and the
forge's work must start once that process lets go.
@@ -6,6 +6,7 @@
 * 24 h / 15 s + 30 d / 5 min = 5,760 + 8,640 rows per target.
 */
import type { Database } from "bun:sqlite";
import { withoutWaiting } from "../kernel/db.ts";
export const FINE_MS = 15_000;
export const COARSE_MS = 300_000;
@@ -92,7 +93,10 @@
type FineRow = {
  lag_ms: number | null; checks: number | null; agents: number | null;
};
/** Keep `points` (taken at `now`), roll up the windows that closed, drop what is past its window. One transaction. */
/**
 * Keep `points` (taken at `now`), roll up the windows that closed, drop what is past its window. One transaction,
 * never waiting on another process's write lock: behind one it throws at once and the sampler's next tick tries again (#6957).
 */
export function record(db: Database, points: readonly Point[], now: number): void {
  const put = db.query(
    `INSERT OR REPLACE INTO machine_sample
@@ -100,7 +104,7 @@
export function record(db: Database, points: readonly Point[], now: number): voi
        lag_ms, lag_ms_max, checks, agents)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
  );
  db.transaction(() => {
  withoutWaiting(db, db.transaction(() => {
    for (const p of points) {
      const c = p.celsius ?? null;
      const lag = p.lagMs ?? null;
@@ -133,5 +137,5 @@
export function record(db: Database, points: readonly Point[], now: number): voi
    }
    db.query("DELETE FROM machine_sample WHERE step = 15 AND at < ?").run(iso(now - FINE_KEEP_MS));
    db.query("DELETE FROM machine_sample WHERE step = 300 AND at < ?").run(iso(now - COARSE_KEEP_MS));
  })();
  }));
}
@@ -5,7 +5,8 @@
 * key in one transaction as an upsert that adds to `n` and `sum_us`; it runs
 * by itself about a second after the first pending record, once 500 keys are
 * pending, and on `stop()`. The timer is `unref`'d. A failed write loses
 * nothing: the records go back and the timer retries. Rows older than
 * nothing: the records go back and the timer retries. A flush never waits on
      
        junipercrew
        18d ago
        
      
      The header comment now wraps awkwardly and runs past the line width. Re-flow it.
    
 * another process's write lock (`withoutWaiting`): it fails and retries. Rows older than
 * `KEEP_DAYS` are deleted, checked once per hour at flush.
 *
 * `record` also adds to totals kept in memory per database for the life of
@@ -18,6 +19,7 @@
 */
import type { Database } from "bun:sqlite";
import { withoutWaiting } from "../kernel/db.ts";
/** Upper edges in ms; bucket `i` holds durations `<= EDGES_MS[i]`, the last is everything slower. */
export const EDGES_MS = [1, 2, 5, 10, 25, 50, 100, 250, 500, 1000, 2500] as const;
@@ -126,7 +128,8 @@
export function createRouteTiming(
         DO UPDATE SET n = n + excluded.n, sum_us = sum_us + excluded.sum_us`,
      );
      const hourNo = Math.floor(now() / HOUR_MS);
      db.transaction(() => {
      // Behind another process's lock this throws at once and the records go back (#6957).
      withoutWaiting(db, db.transaction(() => {
        for (const [key, v] of batch) {
          const [h, pattern, b] = key.split(SEP) as [string, string, string];
          upsert.run(hourText(Number(h)), pattern, Number(b), v.n, v.us);
@@ -137,7 +140,7 @@
export function createRouteTiming(
          );
          prunedHour = hourNo;
        }
      })();
      }));
    } catch (err) {
      prunedHour = -1;
      for (const [key, v] of batch) {
@@ -14,6 +14,7 @@
 */
import type { Database } from "bun:sqlite";
import { withoutWaiting } from "../kernel/db.ts";
export const ROLES = ["anon", "user", "admin", "agent"] as const;
export const SURFACES = ["api", "mcp", "web"] as const;
@@ -123,14 +124,15 @@
export function createUsageCounter(
         ON CONFLICT (hour, pattern, method, status, role, surface)
         DO UPDATE SET n = n + excluded.n`,
      );
      db.transaction(() => {
      // Behind another process's lock this throws at once and the counts go back (#6957).
      withoutWaiting(db, db.transaction(() => {
        for (const [key, n] of batch) {
          const [hour, pattern, method, status, role, surface] = key.split(
            SEP,
          ) as [string, string, string, string, string, string];
          upsert.run(hour, pattern, method, status, role, surface, n);
        }
      })();
      }));
    } catch (err) {
      // Put the counts back so a failed write loses nothing.
      for (const [key, n] of batch)
@@ -171,6 +171,23 @@
export function transact<T>(db: Database, work: () => T): T {
  return withRetry(() => tx.immediate());
}
/**
 * A write not worth the thread (#6957): telemetry that keeps its rows and
 * tries again. `busy_timeout` waits on the one thread every request is
 * served from, so a write behind another process's lock (a draining forge
 * on the same data) stops /healthz with it. Here the lock is tried once and
 * SQLITE_BUSY comes back at once; the handle's own timeout is put back after.
 */
export function withoutWaiting<T>(db: Database, work: () => T): T {
  const was = db.query<{ timeout: number }, []>("PRAGMA busy_timeout").get()?.timeout ?? BUSY_TIMEOUT_MS;
  db.exec("PRAGMA busy_timeout = 0");
  try {
    return work();
  } finally {
    db.exec(`PRAGMA busy_timeout = ${Math.max(0, Math.floor(was))}`);
  }
}
export function isBusy(error: unknown): boolean {
  if (error === null || typeof error !== "object") return false;
  const code = (error as { code?: unknown }).code;
@@ -9,6 +9,8 @@
 * take, in ms, and starts a new window; null when no timer fired in it.
 */
import type { Log } from "./log.ts";
export type LoopLag = { take(): number | null; stop(): void };
/** How often the timer asks to wake, ms: 20 wakes a second, no measurable cost. */
@@ -29,15 +31,21 @@
export function nearestRank(xs: readonly number[], p: number): number | null {
  return sorted[rank - 1]!;
}
/** Start measuring; `now` is a clock in ms (`performance.now` in production). */
export function startLoopLag(everyMs = LAG_EVERY_MS, now: () => number = () => performance.now()): LoopLag {
/** Start measuring; `now` is a clock in ms (`performance.now` in production); `onLate` hears every wake's lateness. */
export function startLoopLag(
  everyMs = LAG_EVERY_MS,
  now: () => number = () => performance.now(),
  onLate?: (lateMs: number) => void,
): LoopLag {
  let samples: number[] = [];
  let timer: ReturnType<typeof setTimeout> | undefined;
  const arm = (): void => {
    const due = now() + everyMs;
    timer = setTimeout(() => {
      // Early by a fraction of a ms is timer rounding, not a fast loop.
      if (samples.length < MAX_SAMPLES) samples.push(Math.max(0, now() - due));
      const late = Math.max(0, now() - due);
      if (samples.length < MAX_SAMPLES) samples.push(late);
      onLate?.(late);
      arm();
    }, everyMs);
    timer.unref?.();
@@ -52,3 +60,23 @@
export function startLoopLag(everyMs = LAG_EVERY_MS, now: () => number = () => p
    stop: () => clearTimeout(timer),
  };
}
/** A wake this late is a stall worth a warning (#6957): a readiness probe gives up at 1 s. */
export const STALL_WARN_MS = 1_000;
/**
 * The always-on gauge (#6957): from boot, every wake later than `warnMs` is
 * one warning saying how late, so a stall that keeps /healthz from answering
 * names its minute in the log. Returns the stop.
 */
export function watchStalls(
  log: Pick<Log, "warn">,
  warnMs = STALL_WARN_MS,
  everyMs = LAG_EVERY_MS,
  now: () => number = () => performance.now(),
): () => void {
  const lag = startLoopLag(everyMs, now, (late) => {
    if (late >= warnMs) log.warn("event loop stalled", { lateMs: Math.round(late) });
  });
  return lag.stop;
}
@@ -0,0 +1,83 @@
/**
 * Statements that hold the thread (#6957). bun:sqlite is synchronous: a
 * statement waiting on another process's write lock (`busy_timeout`, 15 s)
 * or walking a big table stops every request, `/healthz` too, for as long
 * as it takes. `timeStatements` wraps one handle so a statement or exec
 * that takes `overMs` or more is one warning with the first words of its
 * SQL and the stack that called it: the stall names itself.
 *
 * SQL text is the code's own (values are bound, never spliced in), and the
 * log redacts anything credential-shaped besides. At most `PER_MINUTE`
 * lines a minute, so a lock held for minutes is a few lines, not thousands.
 */
import type { Database } from "bun:sqlite";
import type { Log } from "./log.ts";
/** A statement this slow is a stall a person waits through. */
export const SLOW_SQL_MS = 200;
const PER_MINUTE = 10;
const METHODS = ["run", "get", "all", "values"] as const;
/** The SQL's first 100 characters on one line. */
export function sqlHead(sql: unknown): string {
  return String(sql).replace(/\s+/g, " ").trim().slice(0, 100);
}
/** The callers above this module, nearest first, on one line. */
function callers(): string {
  return (new Error().stack ?? "")
    .split("\n")
    .slice(1)
    .map((l) => l.trim().replace(/^at /, ""))
    .filter((l) => l !== "" && !l.includes("slow_sql.ts"))
    .slice(0, 6)
    .join(" < ");
}
/** Wrap `db`'s query, prepare, run and exec; `now` is a clock in ms. */
export function timeStatements(
  db: Database,
  log: Pick<Log, "warn">,
  overMs = SLOW_SQL_MS,
  now: () => number = () => performance.now(),
): void {
  let minute = 0;
  let told = 0;
  const report = (sql: unknown, ms: number): void => {
    const at = Math.floor(Date.now() / 60_000);
    if (at !== minute) [minute, told] = [at, 0];
    if (++told > PER_MINUTE) return;
    log.warn("slow statement held the thread", { ms: Math.round(ms), sql: sqlHead(sql), stack: callers() });
  };
  const timed = <A extends unknown[], R>(sql: unknown, fn: (...args: A) => R) =>
    (...args: A): R => {
      const start = now();
      try {
        return fn(...args);
      } finally {
        const ms = now() - start;
        if (ms >= overMs) report(sql, ms);
      }
    };
  const wrapped = new WeakSet<object>();
  const wrap = <S extends object>(stmt: S, sql: unknown): S => {
    if (wrapped.has(stmt)) return stmt;
    wrapped.add(stmt);
    const methods = stmt as unknown as Record<string, unknown>;
    for (const m of METHODS) {
      const fn = methods[m];
      if (typeof fn === "function") methods[m] = timed(sql, (fn as (...a: unknown[]) => unknown).bind(stmt));
    }
    return stmt;
  };
  const handle = db as unknown as Record<string, (...args: unknown[]) => unknown>;
  for (const name of ["query", "prepare"] as const) {
    const make = handle[name]!.bind(db);
    handle[name] = (sql: unknown, ...rest: unknown[]) => wrap(make(sql, ...rest) as object, sql);
  }
  for (const name of ["run", "exec"] as const) {
    const exec = handle[name]!.bind(db);
    handle[name] = (sql: unknown, ...rest: unknown[]) => timed(sql, exec)(sql, ...rest);
  }
}
@@ -72,6 +72,10 @@
import { bootCounts, recordBoot, recordCleanExit } from "./kernel/boots.ts";
import { census, censusWords, processRows } from "./kernel/census.ts";
import { reapLeftoverBridges } from "./crew/leftovers.ts";
import { crewRoot } from "./crew/workdir.ts";
import { createLog } from "./kernel/log.ts";
import { watchStalls } from "./kernel/loop_lag.ts";
import { timeStatements } from "./kernel/slow_sql.ts";
import { waitForPeer } from "./runs/peer.ts";
// `forge hook <name>` re-enters here from a git hook stub. It relays to the
// running forge over its socket and exits — it must never boot a server.
@@ -191,6 +195,11 @@
if (parentProblem !== null) {
  process.exit(1);
}
// The loop's lateness, from the first line of the boot: a stall over a
// second is a warning that names its minute (kernel/loop_lag.ts, #6957),
// draining included.
watchStalls(createLog("loop"));
const app = createApp({
  dataDir,
  // A new forge starts with signup = invite (FORGE_SIGNUP below still wins).
@@ -272,6 +281,8 @@
const app = createApp({
      }),
});
// A statement that holds the thread past 200 ms names its caller (#6957).
timeStatements(app.db, createLog("sql"));
loadFlags(app.db);
loadTrustedHosts(app.db);
// FORGE_SIGNUP (checked in configProblem) sets who may sign up; kept like any change.
@@ -361,6 +372,24 @@
if (apps && presence) warmBehind(app.log, presence, () => apps.startingCount());
void backfillCarried(app.db, app.reposRoot).catch((error) =>
  app.log.error("could not backfill what deploys carried", { error: String(error) }),
);
// A forge still draining on this database keeps writing for its gates and
// runs; starting ours beside it stalls the listener behind its write lock
// (#6957). The listener answers meanwhile; a signal now stops what is up.
const leaveEarly = (): void => {
  void Promise.resolve(apps?.stop())
    .catch(() => {})
    .finally(() => {
      presence?.stop();
      ssh?.stop();
      servers.stop();
      app.close();
      process.exit(0);
    });
};
const earlySignals = ["SIGINT", "SIGTERM", "SIGHUP"] as const;
for (const signal of earlySignals) process.once(signal, leaveEarly);
await waitForPeer(app.db, app.log);
for (const signal of earlySignals) process.off(signal, leaveEarly);
// A repository's own check, on every push a pull or the default branch takes.
const checks =
  process.env.FORGE_CHECKS === "0"
@@ -60,3 +60,8 @@
export function holdLease(db: Database, runId: string): () => void {
    }
  };
}
/** The runs this process holds the lease of on `db`: whatever else is renewed is another forge's. */
export function heldRuns(db: Database): ReadonlySet<string> {
  return held.get(db) ?? new Set();
}
@@ -0,0 +1,69 @@
/**
 * Another forge still working this database (#6957). A roll stops the old
 * pod and starts the new one on the same data, and since #6732 the old one
 * keeps its gates and crew runs going through its drain: it keeps writing.
 * bun:sqlite waits for the write lock on the one thread, up to
 * `busy_timeout` a statement, so a new forge that starts its own gates and
 * ticks beside it stalls every request behind the old one's writes: 0.7.278
 * failed readiness for minutes that way.
 *
 * The old forge says it is there the way it always has: it renews the leases
 * of the runs it holds (`lease.ts`). A running run renewed within
 * `PEER_FRESH_MS` that this process does not hold is that forge's, whatever
 * its version. `waitForPeer` lets the listener answer and holds the rest of
 * the boot until no such run is left, or `PEER_WAIT_MAX_MS` has passed (the
 * platform has stopped the old pod by then).
 */
import type { Database } from "bun:sqlite";
import type { Log } from "../kernel/log.ts";
import { HEARTBEAT_MS, heldRuns } from "./lease.ts";
/** Two renewals missed: the lease is not being kept. */
export const PEER_FRESH_MS = 2 * HEARTBEAT_MS;
/** The pod's grace (charts/app, 720 s): an old pod is gone by then. */
export const PEER_WAIT_MAX_MS = 720_000;
const POLL_MS = 5_000;
/** Running runs whose lease another process renewed within `PEER_FRESH_MS` of `now`. */
export function peerRuns(db: Database, now = Date.now()): number {
  const here = heldRuns(db);
  const rows = db
    .query<{ id: string }, [string]>("SELECT id FROM run WHERE state = 'running' AND heartbeat_at >= ?1")
    .all(new Date(now - PEER_FRESH_MS).toISOString());
  return rows.filter((r) => !here.has(r.id)).length;
}
export interface PeerWait {
  readonly pollMs?: number;
  readonly maxMs?: number;
  readonly now?: () => number;
  readonly sleep?: (ms: number) => Promise<unknown>;
}
/** How the wait ended: nobody was there, they finished, or the ceiling passed. */
export type PeerOutcome = "alone" | "gone" | "waited out";
/** Resolve once no other forge's run is live on `db`; said in the log when it waits at all. */
export async function waitForPeer(db: Database, log: Log, opts: PeerWait = {}): Promise<PeerOutcome> {
  const now = opts.now ?? Date.now;
  const sleep = opts.sleep ?? Bun.sleep;
  const maxMs = opts.maxMs ?? PEER_WAIT_MAX_MS;
  const started = now();
  let live = peerRuns(db, started);
  if (live === 0) return "alone";
  log.warn("another forge still works this database: this one serves, and starts its own work once it lets go", {
    runs: live,
    maxSeconds: Math.round(maxMs / 1000),
  });
  while (live > 0) {
    if (now() - started >= maxMs) {
      log.warn("another forge still works this database; starting beside it", { runs: live });
      return "waited out";
    }
    await sleep(opts.pollMs ?? POLL_MS);
    live = peerRuns(db, now());
  }
  log.info("the other forge let go of this database; starting work", { waitedSeconds: Math.round((now() - started) / 1000) });
  return "gone";
}
@@ -0,0 +1,116 @@
/**
 * A forge booting beside one still draining on the same database answers
 * /healthz within a second throughout (#6957). 0.7.278 came up while
 * 0.7.276 finished a batch of gates on the shared data, started its own
 * work, and waited on the old pod's write lock on the one thread it serves
 * from: readiness timed out for minutes.
 *
 * The peer here is that old pod in miniature: a process that keeps the
 * leases of a batch of 8 running checks fresh and holds the write lock
 * for most of every second, then finishes and lets go.
 */
import { expect, test } from "bun:test";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp } from "../src/app.ts";
const PEER_MS = 20_000;
const HOLD_MS = 2_500;
const GAP_MS = 150;
/** Holds the write lock `HOLD_MS` of every `HOLD_MS + GAP_MS`, renewing the 8 leases each time; then lands them and exits. */
const peerScript = `
const { Database } = require("bun:sqlite");
const db = new Database(process.argv[2]);
db.run("PRAGMA busy_timeout = 15000");
const ids = ${JSON.stringify(Array.from({ length: 8 }, (_, n) => `gate-${n}`))};
const until = Date.now() + ${PEER_MS};
const spin = (ms) => { const end = Date.now() + ms; while (Date.now() < end) {} };
console.log("peer up");
const tick = () => {
  db.run("BEGIN IMMEDIATE");
  db.run("UPDATE run SET heartbeat_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id IN (" + ids.map((i) => "'" + i + "'").join(",") + ")");
  spin(${HOLD_MS});
  db.run("COMMIT");
  if (Date.now() < until) return void setTimeout(tick, ${GAP_MS});
  db.run("UPDATE run SET state = 'ok' WHERE state = 'running'");
  console.log("peer done");
  process.exit(0);
};
tick();
`;
test("healthz answers within 1 s throughout a boot beside a draining forge's batch of 8, and work starts once it lets go", async () => {
  const dataDir = await mkdtemp(join(tmpdir(), "forge-bootpeer-"));
  const seed = createApp({ dataDir, port: 0 });
  seed.db.run("PRAGMA foreign_keys = OFF");
  const now = new Date().toISOString();
  for (let n = 0; n < 8; n++)
    seed.db
      .query("INSERT INTO run (id, repo_id, kind, state, started_at, heartbeat_at) VALUES (?1, 'r', 'check', 'running', ?2, ?2)")
      .run(`gate-${n}`, now);
  seed.close();
  const script = join(dataDir, "..", `${dataDir.split("/").pop()}-peer.js`);
  await writeFile(script, peerScript);
  const peer = Bun.spawn(["bun", script, join(dataDir, "forge.db")], { stdout: "pipe", stderr: "inherit" });
  await peer.stdout.getReader().read();
  const web = 20000 + Math.floor(Math.random() * 20000);
  const env: Record<string, string> = {
    ...(process.env as Record<string, string>),
    FORGE_DATA: dataDir,
    PORT: String(web),
    FORGE_SSH_PORT: "0",
    FORGE_APPS: "0",
  };
  delete env.FORGE_URL;
  const forge = Bun.spawn(["bun", "run", "src/main.ts"], { env, stdout: "pipe", stderr: "pipe" });
  let out = "";
  const drain = async (stream: ReadableStream<Uint8Array>) => {
    for await (const chunk of stream) out += new TextDecoder().decode(chunk);
  };
  void drain(forge.stdout);
  void drain(forge.stderr);
  try {
    // Up: the first answer, however long the boot behind the lock took.
    let up = false;
    for (let i = 0; i < 300 && !up; i++) {
      try {
        up = (await fetch(`http://127.0.0.1:${web}/healthz`, { signal: AbortSignal.timeout(1_000) })).ok;
      } catch {
        await Bun.sleep(100);
      }
    }
    expect(up).toBe(true);
    // Then every probe answers within a second while the peer holds on, and after.
    const slow: number[] = [];
    const end = performance.now() + PEER_MS;
    while (performance.now() < end) {
      const t = performance.now();
      try {
        const res = await fetch(`http://127.0.0.1:${web}/healthz`, { signal: AbortSignal.timeout(1_000) });
        if (!res.ok) slow.push(-res.status);
      } catch {
        slow.push(Math.round(performance.now() - t));
      }
      await Bun.sleep(200);
    }
    // A probe that timed out: the forge's own log names what held it.
    if (slow.length > 0) console.log(out);
    expect(slow).toEqual([]);
    expect(out).toContain("another forge still works this database");
    await peer.exited;
    for (let i = 0; i < 100 && !out.includes("the other forge let go of this database"); i++) await Bun.sleep(100);
    expect(out).toContain("the other forge let go of this database");
    expect(out).not.toContain("event loop stalled");
  } finally {
    peer.kill();
    forge.kill();
    await Promise.all([peer.exited, forge.exited]);
      
        junipercrew
        18d ago
        
      
      Wall-clock assertion that no stall line appears is the flake risk on a busy check machine. Prefer a tolerance, or assert only the probe latency.
    
    await rm(dataDir, { recursive: true, force: true });
    await rm(script, { force: true });
  }
}, 60_000);
@@ -0,0 +1,185 @@
/**
 * The next stall names itself (#6957): the always-on gauge warns when the
 * loop wakes over a second late, a statement that holds the thread past
 * 200 ms is logged with the stack that called it, and a forge booting
 * beside another one's live runs holds its own work until they are gone.
 */
import { describe, expect, test } from "bun:test";
import { Database } from "bun:sqlite";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp } from "../src/app.ts";
import { watchStalls } from "../src/kernel/loop_lag.ts";
import { sqlHead, timeStatements } from "../src/kernel/slow_sql.ts";
import { holdLease } from "../src/runs/lease.ts";
import { PEER_FRESH_MS, peerRuns, waitForPeer } from "../src/runs/peer.ts";
import type { LogFields } from "../src/kernel/log.ts";
type Line = { message: string; fields?: LogFields };
const recorder = () => {
  const lines: Line[] = [];
  const push = (message: string, fields?: LogFields) => void lines.push({ message, ...(fields ? { fields } : {}) });
  return { lines, log: { debug: push, info: push, warn: push, error: push } };
};
describe("the stall gauge", () => {
  test("a wake later than the floor is one warning saying how late; an on-time loop says nothing", async () => {
    const { lines, log } = recorder();
    const stop = watchStalls(log, 150, 10);
    try {
      await Bun.sleep(100);
      expect(lines).toEqual([]);
      const until = performance.now() + 300;
      while (performance.now() < until) {
        /* hold the thread */
      }
      await Bun.sleep(50);
    } finally {
      stop();
    }
    expect(lines).toHaveLength(1);
    expect(lines[0]!.message).toBe("event loop stalled");
    expect(Number(lines[0]!.fields?.lateMs)).toBeGreaterThanOrEqual(150);
  });
});
describe("slow statements", () => {
  test("a statement over the floor is logged with its SQL and the caller's stack; a fast one is not", () => {
    const db = new Database(":memory:");
    const { lines, log } = recorder();
    // A clock that says every call took 250 ms, until it is told otherwise.
    let step = 250;
    let t = 0;
    timeStatements(db, log, 200, () => (t += step));
    db.run("CREATE TABLE t (x INTEGER)");
    db.query("INSERT INTO t VALUES (?1)").run(1);
    db.prepare("SELECT   x\n FROM t").all();
    db.transaction(() => db.query("SELECT 1").get()).immediate();
    step = 0;
    db.query("SELECT x FROM t").all();
    db.exec("SELECT 1");
    const sqls = lines.map((l) => l.fields?.sql);
    expect(lines.every((l) => l.message === "slow statement held the thread")).toBe(true);
    expect(sqls).toContain("CREATE TABLE t (x INTEGER)");
    expect(sqls).toContain("INSERT INTO t VALUES (?1)");
    expect(sqls).toContain("SELECT x FROM t");
    expect(sqls).toContain("BEGIN IMMEDIATE");
    expect(lines.every((l) => String(l.fields?.stack).includes("loop-stall.test.ts"))).toBe(true);
    expect(lines.some((l) => String(l.fields?.stack).includes("slow_sql.ts"))).toBe(false);
    expect(lines).toHaveLength(6);
    db.close();
  });
  test("a lock held by another process is the slow statement, named", async () => {
    const dir = await mkdtemp(join(tmpdir(), "forge-slowsql-"));
    const file = join(dir, "db.sqlite");
    const db = new Database(file, { create: true });
    db.run("PRAGMA journal_mode = WAL");
    db.run("PRAGMA busy_timeout = 5000");
    db.run("CREATE TABLE t (x INTEGER)");
    const holder = Bun.spawn(
      ["bun", "-e", `const { Database } = require("bun:sqlite"); const d = new Database(${JSON.stringify(file)}); d.run("BEGIN IMMEDIATE"); console.log("held"); setTimeout(() => { d.run("COMMIT"); process.exit(0); }, 400);`],
      { stdout: "pipe" },
    );
    try {
      const reader = holder.stdout.getReader();
      await reader.read();
      const { lines, log } = recorder();
      timeStatements(db, log);
      db.query("INSERT INTO t VALUES (1)").run();
      expect(lines).toHaveLength(1);
      expect(lines[0]!.fields?.sql).toBe("INSERT INTO t VALUES (1)");
      expect(Number(lines[0]!.fields?.ms)).toBeGreaterThanOrEqual(200);
    } finally {
      await holder.exited;
      db.close();
      await rm(dir, { recursive: true, force: true });
    }
  });
  test("the SQL is one line, at most 100 characters", () => {
    expect(sqlHead("SELECT a,\n   b FROM c")).toBe("SELECT a, b FROM c");
    expect(sqlHead(`SELECT ${"x".repeat(200)}`)).toHaveLength(100);
  });
});
describe("another forge's live runs", () => {
  const withApp = async (body: (db: Database) => Promise<void> | void) => {
    const dataDir = await mkdtemp(join(tmpdir(), "forge-peer-"));
    const app = createApp({ dataDir, port: 0 });
    app.db.run("PRAGMA foreign_keys = OFF");
    try {
      await body(app.db);
    } finally {
      app.close();
      await rm(dataDir, { recursive: true, force: true });
    }
  };
  const run = (db: Database, id: string, state: string, heartbeatAt: string | null) =>
    db.query("INSERT INTO run (id, repo_id, kind, state, heartbeat_at) VALUES (?1, 'r', 'check', ?2, ?3)").run(id, state, heartbeatAt);
  test("a running run renewed lately counts; a lapsed one, a finished one and one this process holds do not", () =>
    withApp((db) => {
      const now = Date.parse("2026-10-27T11:37:30.000Z");
      const at = (ago: number) => new Date(now - ago).toISOString();
      run(db, "fresh", "running", at(1_000));
      run(db, "edge", "running", at(PEER_FRESH_MS - 1));
      run(db, "lapsed", "running", at(PEER_FRESH_MS + 1));
      run(db, "never", "running", null);
      run(db, "done", "ok", at(1_000));
      run(db, "mine", "running", at(1_000));
      expect(peerRuns(db, now)).toBe(3);
      const release = holdLease(db, "mine");
      // Holding renews it now, which is after `now`: still fresh, and still not a peer's.
      expect(peerRuns(db, now)).toBe(2);
      release();
    }));
  test("nobody there: no wait, no line", () =>
    withApp(async (db) => {
      const { lines, log } = recorder();
      expect(await waitForPeer(db, log, { sleep: async () => {} })).toBe("alone");
      expect(lines).toEqual([]);
    }));
  test("the wait says so once, polls until the peer's runs finish, and says it let go", () =>
    withApp(async (db) => {
      run(db, "a", "running", new Date().toISOString());
      run(db, "b", "running", new Date().toISOString());
      const { lines, log } = recorder();
      let polls = 0;
      const outcome = await waitForPeer(db, log, {
        sleep: async () => {
          polls++;
          if (polls === 1) db.run("UPDATE run SET state = 'ok' WHERE id = 'a'");
          if (polls === 2) db.run("UPDATE run SET state = 'ok' WHERE id = 'b'");
        },
      });
      expect(outcome).toBe("gone");
      expect(polls).toBe(2);
      expect(lines.map((l) => l.message)).toEqual([
        "another forge still works this database: this one serves, and starts its own work once it lets go",
        "the other forge let go of this database; starting work",
      ]);
      expect(lines[0]!.fields?.runs).toBe(2);
    }));
  test("a peer that never lets go is waited out at the ceiling", () =>
    withApp(async (db) => {
      const { lines, log } = recorder();
      let clock = Date.now();
      run(db, "a", "running", new Date(clock).toISOString());
      const outcome = await waitForPeer(db, log, {
        maxMs: 10_000,
        now: () => clock,
        sleep: async (ms) => {
          clock += ms;
          db.run("UPDATE run SET heartbeat_at = ?1 WHERE id = 'a'", [new Date(clock).toISOString()]);
        },
      });
      expect(outcome).toBe("waited out");
      expect(lines.at(-1)?.message).toBe("another forge still works this database; starting beside it");
    }));
});
		@@ -234,3 +234,34 @@ in `src/release/`:
234	234	version that was ready, kept on the volume under `release/good/`. The forge
235	235	that comes up shows a `release:rolled-back` card and pages the admins once.
236	236	The next version that reads ready clears it.
	237
	238	## A forge beside one still draining (#6957)
	239
	240	bun:sqlite waits for the write lock on the one thread the forge serves from,
	241	up to `busy_timeout` (15 s) a statement. 0.7.278 booted while 0.7.276 still
	242	ran its gates and crew runs on the same database. It started its own work
	243	beside them, and `/healthz` timed out for minutes behind the old pod's writes.
	244	Three things keep that from happening again:
	245
	246	- The new forge holds its work (`src/runs/peer.ts`). Its listener
	247	answers at once. Its checks, merge queue, crew and ticks start only once
	248	no running run that another process holds has renewed its lease within
	249	30 s, or after 720 s (the pod's grace). The log says
	250	`another forge still works this database` when the wait starts and
	251	`the other forge let go of this database` when it ends.
	252	- Telemetry never waits on the lock (`withoutWaiting` in
	253	`src/kernel/db.ts`). Route timings, usage counts and machine samples try
	254	the lock once, keep their rows and try again on the next flush. Every
	255	probe is counted, so without this a probe's own count would hold the next
	256	probe up.
	257	- The next stall names itself. `event loop stalled lateMs=…` is a
	258	warning whenever the loop wakes more than a second late
	259	(`src/kernel/loop_lag.ts`, on from the first line of the boot). `slow
	260	statement held the thread ms=… sql=… stack=…` names any statement on the
	261	forge's handle that took 200 ms or more, with its callers
	262	(`src/kernel/slow_sql.ts`), at most ten lines a minute.
	263
	264	`test/boot-beside-peer.test.ts` boots `src/main.ts` beside a process that
	265	holds the write lock 2.5 s of every 2.65 s and keeps a batch of 8 check
	266	leases fresh. Every `/healthz` probe must answer within a second, and the
	267	forge's work must start once that process lets go.
		@@ -6,6 +6,7 @@
6	6	* 24 h / 15 s + 30 d / 5 min = 5,760 + 8,640 rows per target.
7	7	*/
8	8	import type { Database } from "bun:sqlite";
	9	import { withoutWaiting } from "../kernel/db.ts";
9	10
10	11	export const FINE_MS = 15_000;
11	12	export const COARSE_MS = 300_000;
		@@ -92,7 +93,10 @@ type FineRow = {
92	93	lag_ms: number \| null; checks: number \| null; agents: number \| null;
93	94	};
94	95
95		/** Keep `points` (taken at `now`), roll up the windows that closed, drop what is past its window. One transaction. */
	96	/**
	97	* Keep `points` (taken at `now`), roll up the windows that closed, drop what is past its window. One transaction,
	98	* never waiting on another process's write lock: behind one it throws at once and the sampler's next tick tries again (#6957).
	99	*/
96	100	export function record(db: Database, points: readonly Point[], now: number): void {
97	101	const put = db.query(
98	102	`INSERT OR REPLACE INTO machine_sample
		@@ -100,7 +104,7 @@ export function record(db: Database, points: readonly Point[], now: number): voi
100	104	lag_ms, lag_ms_max, checks, agents)
101	105	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
102	106	);
103		db.transaction(() => {
	107	withoutWaiting(db, db.transaction(() => {
104	108	for (const p of points) {
105	109	const c = p.celsius ?? null;
106	110	const lag = p.lagMs ?? null;
		@@ -133,5 +137,5 @@ export function record(db: Database, points: readonly Point[], now: number): voi
133	137	}
134	138	db.query("DELETE FROM machine_sample WHERE step = 15 AND at < ?").run(iso(now - FINE_KEEP_MS));
135	139	db.query("DELETE FROM machine_sample WHERE step = 300 AND at < ?").run(iso(now - COARSE_KEEP_MS));
136		})();
	140	}));
137	141	}
		@@ -5,7 +5,8 @@
5	5	* key in one transaction as an upsert that adds to `n` and `sum_us`; it runs
6	6	* by itself about a second after the first pending record, once 500 keys are
7	7	* pending, and on `stop()`. The timer is `unref`'d. A failed write loses
8		* nothing: the records go back and the timer retries. Rows older than
	8	* nothing: the records go back and the timer retries. A flush never waits on
		junipercrew 18d ago The header comment now wraps awkwardly and runs past the line width. Re-flow it.
	9	* another process's write lock (`withoutWaiting`): it fails and retries. Rows older than
9	10	* `KEEP_DAYS` are deleted, checked once per hour at flush.
10	11	*
11	12	* `record` also adds to totals kept in memory per database for the life of
		@@ -18,6 +19,7 @@
18	19	*/
19	20
20	21	import type { Database } from "bun:sqlite";
	22	import { withoutWaiting } from "../kernel/db.ts";
21	23
22	24	/** Upper edges in ms; bucket `i` holds durations `<= EDGES_MS[i]`, the last is everything slower. */
23	25	export const EDGES_MS = [1, 2, 5, 10, 25, 50, 100, 250, 500, 1000, 2500] as const;
		@@ -126,7 +128,8 @@ export function createRouteTiming(
126	128	DO UPDATE SET n = n + excluded.n, sum_us = sum_us + excluded.sum_us`,
127	129	);
128	130	const hourNo = Math.floor(now() / HOUR_MS);
129		db.transaction(() => {
	131	// Behind another process's lock this throws at once and the records go back (#6957).
	132	withoutWaiting(db, db.transaction(() => {
130	133	for (const [key, v] of batch) {
131	134	const [h, pattern, b] = key.split(SEP) as [string, string, string];
132	135	upsert.run(hourText(Number(h)), pattern, Number(b), v.n, v.us);
		@@ -137,7 +140,7 @@ export function createRouteTiming(
137	140	);
138	141	prunedHour = hourNo;
139	142	}
140		})();
	143	}));
141	144	} catch (err) {
142	145	prunedHour = -1;
143	146	for (const [key, v] of batch) {
		@@ -14,6 +14,7 @@
14	14	*/
15	15
16	16	import type { Database } from "bun:sqlite";
	17	import { withoutWaiting } from "../kernel/db.ts";
17	18
18	19	export const ROLES = ["anon", "user", "admin", "agent"] as const;
19	20	export const SURFACES = ["api", "mcp", "web"] as const;
		@@ -123,14 +124,15 @@ export function createUsageCounter(
123	124	ON CONFLICT (hour, pattern, method, status, role, surface)
124	125	DO UPDATE SET n = n + excluded.n`,
125	126	);
126		db.transaction(() => {
	127	// Behind another process's lock this throws at once and the counts go back (#6957).
	128	withoutWaiting(db, db.transaction(() => {
127	129	for (const [key, n] of batch) {
128	130	const [hour, pattern, method, status, role, surface] = key.split(
129	131	SEP,
130	132	) as [string, string, string, string, string, string];
131	133	upsert.run(hour, pattern, method, status, role, surface, n);
132	134	}
133		})();
	135	}));
134	136	} catch (err) {
135	137	// Put the counts back so a failed write loses nothing.
136	138	for (const [key, n] of batch)
		@@ -171,6 +171,23 @@ export function transact<T>(db: Database, work: () => T): T {
171	171	return withRetry(() => tx.immediate());
172	172	}
173	173
	174	/**
	175	* A write not worth the thread (#6957): telemetry that keeps its rows and
	176	* tries again. `busy_timeout` waits on the one thread every request is
	177	* served from, so a write behind another process's lock (a draining forge
	178	* on the same data) stops /healthz with it. Here the lock is tried once and
	179	* SQLITE_BUSY comes back at once; the handle's own timeout is put back after.
	180	*/
	181	export function withoutWaiting<T>(db: Database, work: () => T): T {
	182	const was = db.query<{ timeout: number }, []>("PRAGMA busy_timeout").get()?.timeout ?? BUSY_TIMEOUT_MS;
	183	db.exec("PRAGMA busy_timeout = 0");
	184	try {
	185	return work();
	186	} finally {
	187	db.exec(`PRAGMA busy_timeout = ${Math.max(0, Math.floor(was))}`);
	188	}
	189	}
	190
174	191	export function isBusy(error: unknown): boolean {
175	192	if (error === null \|\| typeof error !== "object") return false;
176	193	const code = (error as { code?: unknown }).code;
		@@ -9,6 +9,8 @@
9	9	* take, in ms, and starts a new window; null when no timer fired in it.
10	10	*/
11	11
	12	import type { Log } from "./log.ts";
	13
12	14	export type LoopLag = { take(): number \| null; stop(): void };
13	15
14	16	/** How often the timer asks to wake, ms: 20 wakes a second, no measurable cost. */
		@@ -29,15 +31,21 @@ export function nearestRank(xs: readonly number[], p: number): number \| null {
29	31	return sorted[rank - 1]!;
30	32	}
31	33
32		/** Start measuring; `now` is a clock in ms (`performance.now` in production). */
33		export function startLoopLag(everyMs = LAG_EVERY_MS, now: () => number = () => performance.now()): LoopLag {
	34	/** Start measuring; `now` is a clock in ms (`performance.now` in production); `onLate` hears every wake's lateness. */
	35	export function startLoopLag(
	36	everyMs = LAG_EVERY_MS,
	37	now: () => number = () => performance.now(),
	38	onLate?: (lateMs: number) => void,
	39	): LoopLag {
34	40	let samples: number[] = [];
35	41	let timer: ReturnType<typeof setTimeout> \| undefined;
36	42	const arm = (): void => {
37	43	const due = now() + everyMs;
38	44	timer = setTimeout(() => {
39	45	// Early by a fraction of a ms is timer rounding, not a fast loop.
40		if (samples.length < MAX_SAMPLES) samples.push(Math.max(0, now() - due));
	46	const late = Math.max(0, now() - due);
	47	if (samples.length < MAX_SAMPLES) samples.push(late);
	48	onLate?.(late);
41	49	arm();
42	50	}, everyMs);
43	51	timer.unref?.();
		@@ -52,3 +60,23 @@ export function startLoopLag(everyMs = LAG_EVERY_MS, now: () => number = () => p
52	60	stop: () => clearTimeout(timer),
53	61	};
54	62	}
	63
	64	/** A wake this late is a stall worth a warning (#6957): a readiness probe gives up at 1 s. */
	65	export const STALL_WARN_MS = 1_000;
	66
	67	/**
	68	* The always-on gauge (#6957): from boot, every wake later than `warnMs` is
	69	* one warning saying how late, so a stall that keeps /healthz from answering
	70	* names its minute in the log. Returns the stop.
	71	*/
	72	export function watchStalls(
	73	log: Pick<Log, "warn">,
	74	warnMs = STALL_WARN_MS,
	75	everyMs = LAG_EVERY_MS,
	76	now: () => number = () => performance.now(),
	77	): () => void {
	78	const lag = startLoopLag(everyMs, now, (late) => {
	79	if (late >= warnMs) log.warn("event loop stalled", { lateMs: Math.round(late) });
	80	});
	81	return lag.stop;
	82	}
		@@ -0,0 +1,83 @@
	1	/**
	2	* Statements that hold the thread (#6957). bun:sqlite is synchronous: a
	3	* statement waiting on another process's write lock (`busy_timeout`, 15 s)
	4	* or walking a big table stops every request, `/healthz` too, for as long
	5	* as it takes. `timeStatements` wraps one handle so a statement or exec
	6	* that takes `overMs` or more is one warning with the first words of its
	7	* SQL and the stack that called it: the stall names itself.
	8	*
	9	* SQL text is the code's own (values are bound, never spliced in), and the
	10	* log redacts anything credential-shaped besides. At most `PER_MINUTE`
	11	* lines a minute, so a lock held for minutes is a few lines, not thousands.
	12	*/
	13
	14	import type { Database } from "bun:sqlite";
	15	import type { Log } from "./log.ts";
	16
	17	/** A statement this slow is a stall a person waits through. */
	18	export const SLOW_SQL_MS = 200;
	19	const PER_MINUTE = 10;
	20	const METHODS = ["run", "get", "all", "values"] as const;
	21
	22	/** The SQL's first 100 characters on one line. */
	23	export function sqlHead(sql: unknown): string {
	24	return String(sql).replace(/\s+/g, " ").trim().slice(0, 100);
	25	}
	26
	27	/** The callers above this module, nearest first, on one line. */
	28	function callers(): string {
	29	return (new Error().stack ?? "")
	30	.split("\n")
	31	.slice(1)
	32	.map((l) => l.trim().replace(/^at /, ""))
	33	.filter((l) => l !== "" && !l.includes("slow_sql.ts"))
	34	.slice(0, 6)
	35	.join(" < ");
	36	}
	37
	38	/** Wrap `db`'s query, prepare, run and exec; `now` is a clock in ms. */
	39	export function timeStatements(
	40	db: Database,
	41	log: Pick<Log, "warn">,
	42	overMs = SLOW_SQL_MS,
	43	now: () => number = () => performance.now(),
	44	): void {
	45	let minute = 0;
	46	let told = 0;
	47	const report = (sql: unknown, ms: number): void => {
	48	const at = Math.floor(Date.now() / 60_000);
	49	if (at !== minute) [minute, told] = [at, 0];
	50	if (++told > PER_MINUTE) return;
	51	log.warn("slow statement held the thread", { ms: Math.round(ms), sql: sqlHead(sql), stack: callers() });
	52	};
	53	const timed = <A extends unknown[], R>(sql: unknown, fn: (...args: A) => R) =>
	54	(...args: A): R => {
	55	const start = now();
	56	try {
	57	return fn(...args);
	58	} finally {
	59	const ms = now() - start;
	60	if (ms >= overMs) report(sql, ms);
	61	}
	62	};
	63	const wrapped = new WeakSet<object>();
	64	const wrap = <S extends object>(stmt: S, sql: unknown): S => {
	65	if (wrapped.has(stmt)) return stmt;
	66	wrapped.add(stmt);
	67	const methods = stmt as unknown as Record<string, unknown>;
	68	for (const m of METHODS) {
	69	const fn = methods[m];
	70	if (typeof fn === "function") methods[m] = timed(sql, (fn as (...a: unknown[]) => unknown).bind(stmt));
	71	}
	72	return stmt;
	73	};
	74	const handle = db as unknown as Record<string, (...args: unknown[]) => unknown>;
	75	for (const name of ["query", "prepare"] as const) {
	76	const make = handle[name]!.bind(db);
	77	handle[name] = (sql: unknown, ...rest: unknown[]) => wrap(make(sql, ...rest) as object, sql);
	78	}
	79	for (const name of ["run", "exec"] as const) {
	80	const exec = handle[name]!.bind(db);
	81	handle[name] = (sql: unknown, ...rest: unknown[]) => timed(sql, exec)(sql, ...rest);
	82	}
	83	}
		@@ -72,6 +72,10 @@ import { bootCounts, recordBoot, recordCleanExit } from "./kernel/boots.ts";
72	72	import { census, censusWords, processRows } from "./kernel/census.ts";
73	73	import { reapLeftoverBridges } from "./crew/leftovers.ts";
74	74	import { crewRoot } from "./crew/workdir.ts";
	75	import { createLog } from "./kernel/log.ts";
	76	import { watchStalls } from "./kernel/loop_lag.ts";
	77	import { timeStatements } from "./kernel/slow_sql.ts";
	78	import { waitForPeer } from "./runs/peer.ts";
75	79
76	80	// `forge hook <name>` re-enters here from a git hook stub. It relays to the
77	81	// running forge over its socket and exits — it must never boot a server.
		@@ -191,6 +195,11 @@ if (parentProblem !== null) {
191	195	process.exit(1);
192	196	}
193	197
	198	// The loop's lateness, from the first line of the boot: a stall over a
	199	// second is a warning that names its minute (kernel/loop_lag.ts, #6957),
	200	// draining included.
	201	watchStalls(createLog("loop"));
	202
194	203	const app = createApp({
195	204	dataDir,
196	205	// A new forge starts with signup = invite (FORGE_SIGNUP below still wins).
		@@ -272,6 +281,8 @@ const app = createApp({
272	281	}),
273	282	});
274	283
	284	// A statement that holds the thread past 200 ms names its caller (#6957).
	285	timeStatements(app.db, createLog("sql"));
275	286	loadFlags(app.db);
276	287	loadTrustedHosts(app.db);
277	288	// FORGE_SIGNUP (checked in configProblem) sets who may sign up; kept like any change.
		@@ -361,6 +372,24 @@ if (apps && presence) warmBehind(app.log, presence, () => apps.startingCount());
361	372	void backfillCarried(app.db, app.reposRoot).catch((error) =>
362	373	app.log.error("could not backfill what deploys carried", { error: String(error) }),
363	374	);
	375	// A forge still draining on this database keeps writing for its gates and
	376	// runs; starting ours beside it stalls the listener behind its write lock
	377	// (#6957). The listener answers meanwhile; a signal now stops what is up.
	378	const leaveEarly = (): void => {
	379	void Promise.resolve(apps?.stop())
	380	.catch(() => {})
	381	.finally(() => {
	382	presence?.stop();
	383	ssh?.stop();
	384	servers.stop();
	385	app.close();
	386	process.exit(0);
	387	});
	388	};
	389	const earlySignals = ["SIGINT", "SIGTERM", "SIGHUP"] as const;
	390	for (const signal of earlySignals) process.once(signal, leaveEarly);
	391	await waitForPeer(app.db, app.log);
	392	for (const signal of earlySignals) process.off(signal, leaveEarly);
364	393	// A repository's own check, on every push a pull or the default branch takes.
365	394	const checks =
366	395	process.env.FORGE_CHECKS === "0"
		@@ -60,3 +60,8 @@ export function holdLease(db: Database, runId: string): () => void {
60	60	}
61	61	};
62	62	}
	63
	64	/** The runs this process holds the lease of on `db`: whatever else is renewed is another forge's. */
	65	export function heldRuns(db: Database): ReadonlySet<string> {
	66	return held.get(db) ?? new Set();
	67	}
		@@ -0,0 +1,69 @@
	1	/**
	2	* Another forge still working this database (#6957). A roll stops the old
	3	* pod and starts the new one on the same data, and since #6732 the old one
	4	* keeps its gates and crew runs going through its drain: it keeps writing.
	5	* bun:sqlite waits for the write lock on the one thread, up to
	6	* `busy_timeout` a statement, so a new forge that starts its own gates and
	7	* ticks beside it stalls every request behind the old one's writes: 0.7.278
	8	* failed readiness for minutes that way.
	9	*
	10	* The old forge says it is there the way it always has: it renews the leases
	11	* of the runs it holds (`lease.ts`). A running run renewed within
	12	* `PEER_FRESH_MS` that this process does not hold is that forge's, whatever
	13	* its version. `waitForPeer` lets the listener answer and holds the rest of
	14	* the boot until no such run is left, or `PEER_WAIT_MAX_MS` has passed (the
	15	* platform has stopped the old pod by then).
	16	*/
	17
	18	import type { Database } from "bun:sqlite";
	19	import type { Log } from "../kernel/log.ts";
	20	import { HEARTBEAT_MS, heldRuns } from "./lease.ts";
	21
	22	/** Two renewals missed: the lease is not being kept. */
	23	export const PEER_FRESH_MS = 2 * HEARTBEAT_MS;
	24	/** The pod's grace (charts/app, 720 s): an old pod is gone by then. */
	25	export const PEER_WAIT_MAX_MS = 720_000;
	26	const POLL_MS = 5_000;
	27
	28	/** Running runs whose lease another process renewed within `PEER_FRESH_MS` of `now`. */
	29	export function peerRuns(db: Database, now = Date.now()): number {
	30	const here = heldRuns(db);
	31	const rows = db
	32	.query<{ id: string }, [string]>("SELECT id FROM run WHERE state = 'running' AND heartbeat_at >= ?1")
	33	.all(new Date(now - PEER_FRESH_MS).toISOString());
	34	return rows.filter((r) => !here.has(r.id)).length;
	35	}
	36
	37	export interface PeerWait {
	38	readonly pollMs?: number;
	39	readonly maxMs?: number;
	40	readonly now?: () => number;
	41	readonly sleep?: (ms: number) => Promise<unknown>;
	42	}
	43
	44	/** How the wait ended: nobody was there, they finished, or the ceiling passed. */
	45	export type PeerOutcome = "alone" \| "gone" \| "waited out";
	46
	47	/** Resolve once no other forge's run is live on `db`; said in the log when it waits at all. */
	48	export async function waitForPeer(db: Database, log: Log, opts: PeerWait = {}): Promise<PeerOutcome> {
	49	const now = opts.now ?? Date.now;
	50	const sleep = opts.sleep ?? Bun.sleep;
	51	const maxMs = opts.maxMs ?? PEER_WAIT_MAX_MS;
	52	const started = now();
	53	let live = peerRuns(db, started);
	54	if (live === 0) return "alone";
	55	log.warn("another forge still works this database: this one serves, and starts its own work once it lets go", {
	56	runs: live,
	57	maxSeconds: Math.round(maxMs / 1000),
	58	});
	59	while (live > 0) {
	60	if (now() - started >= maxMs) {
	61	log.warn("another forge still works this database; starting beside it", { runs: live });
	62	return "waited out";
	63	}
	64	await sleep(opts.pollMs ?? POLL_MS);
	65	live = peerRuns(db, now());
	66	}
	67	log.info("the other forge let go of this database; starting work", { waitedSeconds: Math.round((now() - started) / 1000) });
	68	return "gone";
	69	}