d0ad41c · 3ea9c29
@@ -29,6 +29,8 @@
import { cpuTarget, readCpus, readPoint, unreadable } from "./machine.ts";
import { FINE_MS, record, type Point } from "./ring.ts";
import { readHwmon, readTopology, temperatures } from "./thermal.ts";
import { during } from "../perf/jobs.ts";
import { withoutWaiting } from "../kernel/db.ts";
import { pruneEvents, TICK_ROWS } from "../system/retention_events.ts";
export type Paths = { cgroup: string; proc: string; sys: string };
export const SYSTEM: Paths = { cgroup: "/sys/fs/cgroup", proc: "/proc", sys: "/sys" };
@@ -169,6 +171,13 @@
export function startSampler(
    } catch (error) {
      app.log.warn("machine sample failed", { error: String(error) });
    }
    // The same tick prunes one batch of old events (`system/retention_events.ts`); a busy
    // database skips it, and the next tick tries again.
    try {
      withoutWaiting(app.db, () => pruneEvents(app.db, now, TICK_ROWS));
    } catch (error) {
      app.log.warn("event prune failed", { error: String(error) });
    }
  };
  during("sampler", tick);
  const timer = setInterval(() => during("sampler", tick), options.everyMs ?? FINE_MS);
@@ -0,0 +1,1 @@
7a89596dcc5066e8627a0f4ed7a811e794740bc9d9d90ce91306acbe58db8c4e
@@ -0,0 +1,6 @@
// Old events are pruned in batches (`system/retention_events.ts`, #12008), and `webhook_delivery`
// cascades from `event`: without an index on `event_id` every deleted event scans the deliveries.
// One index makes the cascade a lookup. Additive.
export default `
  CREATE INDEX IF NOT EXISTS delivery_event ON webhook_delivery(event_id);
`;
@@ -0,0 +1,129 @@
/**
 * How long an `event` row is kept, and the batch that deletes what is past it (#12008).
 *
 * Every reader of `event`, and the oldest row it can reach:
 *
 * Windowed by age — the longest decides KEEP_MS:
 * - `flow/estimate_base.ts`: reopened issues, 90 days (the base rate's widest level).
 * - `core/team-metrics.ts`: reopens for the team ledger, 7 days.
 * - `web/pages/security.ts`: deleted repositories, 30 days.
 * - `web/studio/since.ts`: "since you were here", the digest window (days).
 * - `feed/items.ts`, `feed/quiet.ts`: the home feed, its newest 300 and a 24 h quiet window.
 * - `feed/affinity.ts`: a viewer's newest 200, weighed by a 7-day half-life.
 * - `actions/mail.ts`: mail ceiling notices, 24 h. `crew/stale_pulls.ts`: closes, 72 h.
 * - `crew/landed.ts`: comments since a pull merged (read when it merges).
 * - `boards/sources.ts`: spawn timings by day, a board's range.
 *
 * Feeds, audit views and replay cursors — they page back through every row, so the rule sets
 * their reach: they answer the last KEEP_MS, and nothing they list newer than that changes:
 * - `core/events.ts` `listEvents`/`listEventsVisibleTo`, under the events page and
 *   `/api/v1/repos/…/events` (`core/events_page.ts`), the MCP `events` tool
 *   (`http/mcp/tools/events.ts`), the TUI events screen, the Atom feed (`web/feeds.ts`, newest
 *   200), the home timeline (`feed/timeline.ts`) and the SSE replay from Last-Event-ID
 *   (`http/sse.ts`): a cursor older than the rule resumes from the oldest row kept.
 * - `core/cloud_account.ts`: the newest cloud secret reads, an audit list.
 * - `core/webhooks.ts`: a hook's recent deliveries (they cascade with their event); a delivery
 *   still `pending` keeps its event, so a retry never loses what it sends.
 *
 * At any age — state read back out of the log, so these kinds are never pruned (KEPT):
 * - `issue.*`, `pull.*`: thread timelines (`web/pages/thread.ts`), assignees and blockers
 *   (`actions/issues.ts`, `crew/dispatch.ts`, `crew/stranded.ts`), team waits, reopen counts
 *   (`flow/read.ts`), splits (`core/issues.ts`, `core/limb_replan.ts`), merge closes
 *   (`web/studio/tell.ts`, `core/done_on_main.ts`), the gate (`core/gate_hold.ts`), last push
 *   to a pull (`flow/idle_pulls.ts`, `web/pages/flow-waiting.ts`).
 * - `repo.*`: detached forks (`actions/fork_chain.ts`), the vision kept (`web/pages/thread.ts`).
 * - `user.*`: how an account was made (`core/accounts.ts`).
 * - `push`: when a repository was last pushed (`core/listings.ts`, `core/repos.ts`,
 *   `core/fork_follow.ts`, `flow/alarm_stuck.ts`).
 * - `app.egress.*`: the grants an app holds (`actions/egress_grants.ts`).
 * - `keeper_*`, `forge_down`, `forge_up`: what the keeper said once, tried, or last saw
 *   (`maintain/keeper/*.ts`).
 * - `pinned.check`, `clog.cleared`, `webhook.failing`: the newest of each, or a streak's mark.
 * Newest-row readers (`web/map_cache.ts`, `spawn/clients.ts`) read MAX and never reach the old end.
 *
 * A new reader that reaches past KEEP_MS names its kind in KEPT, or it is windowed.
 */
import type { Database } from "bun:sqlite";
const DAY_MS = 86_400_000;
/** Twice the longest windowed reader (estimate_base's 90 days). */
export const KEEP_MS = 180 * DAY_MS;
/** One batch: small enough to hold the write lock for milliseconds. */
export const BATCH_ROWS = 5_000;
/** Rows deleted per transaction: about 7 µs a row across the table's indexes. */
export const CHUNK_ROWS = 1_000;
/** What the sampler's 15-s tick asks for: 2.4M old rows in about ten hours, a few ms a tick. */
export const TICK_ROWS = 1_000;
/** After a pass reaches rows inside the rule it rests: the next pass starts a day later. */
export const REST_MS = DAY_MS;
export const KEPT_PREFIXES: readonly string[] = ["issue.", "pull.", "repo.", "user.", "app.egress.", "keeper_"];
export const KEPT_KINDS: readonly string[] = ["push", "forge_down", "forge_up", "pinned.check", "clog.cleared", "webhook.failing"];
/** Whether a row of `kind` is kept at any age. */
export function keptKind(kind: string): boolean {
  return KEPT_KINDS.includes(kind) || KEPT_PREFIXES.some((p) => kind.startsWith(p));
}
type Pass = { after: number; restUntil: number };
const passes = new WeakMap<Database, Pass>();
/**
 * One batch: walks at most `rowCap` rows from where the last batch stopped, in `seq` order (the
 * order they were written, so oldest first), and deletes those older than KEEP_MS that no reader
 * reaches at any age. `more` is whether rows past the rule are still to walk; when a pass meets
 * a row inside the rule it ends and rests REST_MS before starting from the oldest again. `lockMs`
 * is the longest the delete held the write lock (one CHUNK_ROWS transaction).
 */
export function pruneEvents(db: Database, now: number, rowCap: number = BATCH_ROWS): { deleted: number; more: boolean; lockMs: number } {
  const cap = Math.max(1, Math.min(BATCH_ROWS, Math.floor(rowCap)));
  const pass = passes.get(db) ?? { after: 0, restUntil: 0 };
  passes.set(db, pass);
  if (now < pass.restUntil) return { deleted: 0, more: false, lockMs: 0 };
  const cutoff = new Date(now - KEEP_MS).toISOString();
  const rows = db
    .query<{ seq: number; rowid: number; kind: string; created_at: string }, [number, number]>(
      "SELECT seq, rowid, kind, created_at FROM event WHERE seq > ?1 ORDER BY seq LIMIT ?2",
    )
    .all(pass.after, cap);
  const old: typeof rows = [];
  let reached = rows.length < cap;
  for (const row of rows) {
    if (row.created_at >= cutoff) {
      reached = true;
      break;
    }
    old.push(row);
  }
  const doomed = old.filter((r) => !keptKind(r.kind)).map((r) => r.rowid);
  let deleted = 0;
  let lockMs = 0;
  // A chunk per transaction, so the write lock is never held for the whole batch.
  for (let at = 0; at < doomed.length; at += CHUNK_ROWS) {
    const chunk = doomed.slice(at, at + CHUNK_ROWS);
    const started = performance.now();
    db.transaction(() => {
      // A delivery still pending keeps its event: the retry sends it.
      const pending = new Set(
        db
          .query<{ rowid: number }, []>(
            "SELECT e.rowid FROM webhook_delivery d JOIN event e ON e.id = d.event_id WHERE d.state = 'pending'",
          )
          .all()
          .map((r) => r.rowid),
      );
      const gone = pending.size === 0 ? chunk : chunk.filter((r) => !pending.has(r));
      // Counted from the list, not `changes`: that counts the deliveries cascading with them too.
      db.query<unknown, [string]>("DELETE FROM event WHERE rowid IN (SELECT value FROM json_each(?1))").run(JSON.stringify(gone));
      deleted += gone.length;
    })();
    lockMs = Math.max(lockMs, performance.now() - started);
  }
  if (reached) {
    pass.after = 0;
    pass.restUntil = now + REST_MS;
    return { deleted, more: false, lockMs };
  }
  pass.after = old[old.length - 1]!.seq;
  return { deleted, more: true, lockMs };
}
@@ -0,0 +1,137 @@
import { describe, expect, test } from "bun:test";
import type { Database } from "bun:sqlite";
import { migrations } from "../src/core/schema.ts";
import { listEvents } from "../src/core/events.ts";
import { countNotes } from "../src/maintain/keeper/record.ts";
import { openDb } from "../src/kernel/db.ts";
import { BATCH_ROWS, KEEP_MS, keptKind, pruneEvents, REST_MS } from "../src/system/retention_events.ts";
const DAY = 86_400_000;
const NOW = Date.parse("2026-11-13T12:00:00.000Z");
const iso = (ms: number) => new Date(ms).toISOString();
function seeded(): Database {
  const db = openDb(":memory:", migrations);
  // Seeded rows name a repo and a webhook by id alone; the tables they point at are not under test.
  db.exec("PRAGMA foreign_keys = OFF");
  return db;
}
let n = 0;
function put(db: Database, kind: string, ageMs: number, repo: string | null = "r1", payload: Record<string, unknown> = {}): string {
  const id = `e${String(++n).padStart(8, "0")}`;
  db.query("INSERT INTO event (id, kind, repo_id, payload, created_at) VALUES (?1, ?2, ?3, ?4, ?5)").run(
    id, kind, repo, JSON.stringify(payload), iso(NOW - ageMs),
  );
  return id;
}
const ids = (rows: { id: string }[]) => rows.map((r) => r.id);
describe("pruneEvents", () => {
  test("deletes old rows of windowed kinds; every named reader answers the same for its window", () => {
    const db = seeded();
    // Oldest first, as the log is written: old windowed rows, old kept rows, then newer rows.
    const oldWindowed = [put(db, "secret.read", KEEP_MS + 30 * DAY), put(db, "forge.spawn.live", KEEP_MS + 2 * DAY, null)];
    const oldPending = put(db, "run.finished", KEEP_MS + DAY);
    db.query("INSERT INTO webhook_delivery (id, webhook_id, event_id, state) VALUES ('d1', 'w1', ?1, 'pending')").run(oldPending);
    const oldDone = put(db, "run.started", KEEP_MS + DAY);
    db.query("INSERT INTO webhook_delivery (id, webhook_id, event_id, state) VALUES ('d2', 'w1', ?1, 'ok')").run(oldDone);
    put(db, "repo.detached", KEEP_MS + 20 * DAY);
    put(db, "user.created", KEEP_MS + 20 * DAY, null);
    put(db, "issue.closed", KEEP_MS + 10 * DAY, "r1", { number: 1, reason: "split" });
    put(db, "keeper_said", KEEP_MS + 10 * DAY, null, { key: "k1" });
    put(db, "push", KEEP_MS + 5 * DAY);
    const recent: string[] = [];
    for (let d = 170; d >= 0; d -= 10) recent.push(put(db, d % 20 === 0 ? "secret.read" : "issue.reopened", d * DAY, "r1", { number: d }));
    const cutoff = iso(NOW - KEEP_MS);
    const readers = () => ({
      // Feeds, the events page, the API, MCP and SSE replay all page `listEvents`:
      // their window is the rule.
      feed: ids(listEvents(db, { limit: 500 })).filter((id) => db.query("SELECT 1 FROM event WHERE id = ?1 AND created_at >= ?2").get(id, cutoff)),
      repoFeed: ids(listEvents(db, { repoId: "r1", limit: 50 })).filter((id) => db.query("SELECT 1 FROM event WHERE id = ?1 AND created_at >= ?2").get(id, cutoff)),
      replay: ids(listEvents(db, { afterId: recent[3]!, limit: 500 })),
      // estimate_base: reopened issues within 90 days.
      estimate: db.query("SELECT DISTINCT json_extract(payload, '$.number') AS n FROM event WHERE repo_id = 'r1' AND kind = 'issue.reopened' AND created_at >= ?1").all(iso(NOW - 90 * DAY)),
      // security: 30 days of repo.deleted; mail: 24 h.
      windowed: db.query("SELECT id FROM event WHERE created_at >= ?1 ORDER BY seq").all(cutoff),
      // At any age: fork_chain, accounts, core/issues splits, the keeper's once-said, last push.
      detached: db.query("SELECT 1 FROM event WHERE repo_id = 'r1' AND kind = 'repo.detached' LIMIT 1").get() !== null,
      created: db.query("SELECT COUNT(*) AS n FROM event WHERE kind = 'user.created'").get(),
      splits: db.query("SELECT COUNT(*) AS n FROM event WHERE kind = 'issue.closed' AND json_extract(payload, '$.reason') = 'split'").get(),
      said: countNotes(db, "keeper_said", "key", "k1"),
      lastPush: db.query("SELECT MAX(created_at) AS at FROM event WHERE repo_id = 'r1' AND kind = 'push'").get(),
    });
    const before = readers();
    expect(before.said).toBe(1);
    expect(before.detached).toBe(true);
    db.exec("PRAGMA foreign_keys = ON"); // deliveries cascade with their event, as in production
    const first = pruneEvents(db, NOW, BATCH_ROWS);
    expect(first).toMatchObject({ deleted: 3, more: false });
    expect(readers()).toEqual(before);
    for (const id of oldWindowed) expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(id)).toBeNull();
    // A delivery still pending keeps its event; a finished one goes with it.
    expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(oldPending)).not.toBeNull();
    expect(db.query("SELECT 1 FROM webhook_delivery WHERE id = 'd2'").get()).toBeNull();
  });
  test("walks in capped batches, then rests a day before the next pass", () => {
    const db = seeded();
    for (let i = 0; i < 25; i++) put(db, "secret.read", KEEP_MS + DAY);
    put(db, "secret.read", DAY);
    expect(pruneEvents(db, NOW, 10)).toMatchObject({ deleted: 10, more: true });
    expect(pruneEvents(db, NOW, 10)).toMatchObject({ deleted: 10, more: true });
    expect(pruneEvents(db, NOW, 10)).toMatchObject({ deleted: 5, more: false });
    put(db, "secret.read", KEEP_MS + DAY);
    expect(pruneEvents(db, NOW + 1000, 10)).toMatchObject({ deleted: 0, more: false });
    // The next pass, once rows written later have aged past the rule too.
    expect(pruneEvents(db, NOW + Math.max(REST_MS, KEEP_MS), 10)).toMatchObject({ deleted: 2, more: false });
    expect((db.query("SELECT COUNT(*) AS n FROM event").get() as { n: number }).n).toBe(0);
  });
  test("the rule keeps state kinds at any age and windows the rest", () => {
    for (const k of ["issue.closed", "pull.pushed", "repo.detached", "user.created", "push", "keeper_lull", "forge_down", "app.egress.granted"])
      expect(keptKind(k)).toBe(true);
    for (const k of ["secret.read", "forge.spawn.live", "mail.ceiling", "measurement"]) expect(keptKind(k)).toBe(false);
    expect(KEEP_MS).toBeGreaterThanOrEqual(90 * DAY);
  });
  test("one 5,000-row batch holds the write lock under 50 ms", () => {
    const db = seeded();
    const add = db.query("INSERT INTO event (id, kind, repo_id, payload, created_at) VALUES (?1, 'secret.read', 'r1', ?2, ?3)");
    db.transaction(() => {
      for (let i = 0; i < 20_000; i++) add.run(`b${String(i).padStart(6, "0")}`, `{"name":"CLOUD_${i}","number":${i}}`, iso(NOW - KEEP_MS - DAY - (20_000 - i)));
      for (let i = 0; i < 2_000; i++) db.query("INSERT INTO webhook_delivery (id, webhook_id, event_id, state) VALUES (?1, 'w1', ?2, 'ok')").run(`d${i}`, `b${String(i * 7).padStart(6, "0")}`);
    })();
    db.exec("PRAGMA foreign_keys = ON");
    pruneEvents(db, NOW, 100); // warm the statements
    const { deleted, lockMs } = pruneEvents(db, NOW, BATCH_ROWS);
    expect(deleted).toBe(BATCH_ROWS);
    expect(lockMs).toBeGreaterThan(0);
    expect(lockMs).toBeLessThan(50);
  });
});
describe("the sampler's tick", () => {
  test("prunes a batch of old events, the same periodic job as the machine_sample rule", async () => {
    const { startSampler } = await import("../src/boards/sampler.ts");
    const db = seeded();
    // The sampler ticks once at start with the clock's own now, so the rows are aged from it.
    const now = Date.now();
    const gone = put(db, "secret.read", KEEP_MS + DAY + (NOW - now), null);
    const kept = put(db, "secret.read", DAY + (NOW - now), null);
    db.exec("PRAGMA foreign_keys = ON");
    const app = { db, log: { warn: () => {}, info: () => {} } } as unknown as import("../src/app.ts").App;
    const nowhere = { cgroup: "/nowhere", proc: "/nowhere", sys: "/nowhere" };
    const s = startSampler(app, { paths: nowhere, everyMs: 3_600_000, cells: () => [] });
    try {
      s.tick(now);
      expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(gone)).toBeNull();
      expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(kept)).not.toBeNull();
    } finally {
      s.stop();
    }
  });
});
		@@ -29,6 +29,8 @@ import { cpuTarget, readCpus, readPoint, unreadable } from "./machine.ts";
29	29	import { FINE_MS, record, type Point } from "./ring.ts";
30	30	import { readHwmon, readTopology, temperatures } from "./thermal.ts";
31	31	import { during } from "../perf/jobs.ts";
	32	import { withoutWaiting } from "../kernel/db.ts";
	33	import { pruneEvents, TICK_ROWS } from "../system/retention_events.ts";
32	34
33	35	export type Paths = { cgroup: string; proc: string; sys: string };
34	36	export const SYSTEM: Paths = { cgroup: "/sys/fs/cgroup", proc: "/proc", sys: "/sys" };
		@@ -169,6 +171,13 @@ export function startSampler(
169	171	} catch (error) {
170	172	app.log.warn("machine sample failed", { error: String(error) });
171	173	}
	174	// The same tick prunes one batch of old events (`system/retention_events.ts`); a busy
	175	// database skips it, and the next tick tries again.
	176	try {
	177	withoutWaiting(app.db, () => pruneEvents(app.db, now, TICK_ROWS));
	178	} catch (error) {
	179	app.log.warn("event prune failed", { error: String(error) });
	180	}
172	181	};
173	182	during("sampler", tick);
174	183	const timer = setInterval(() => during("sampler", tick), options.everyMs ?? FINE_MS);
		@@ -0,0 +1,1 @@
	1	7a89596dcc5066e8627a0f4ed7a811e794740bc9d9d90ce91306acbe58db8c4e
		@@ -0,0 +1,6 @@
	1	// Old events are pruned in batches (`system/retention_events.ts`, #12008), and `webhook_delivery`
	2	// cascades from `event`: without an index on `event_id` every deleted event scans the deliveries.
	3	// One index makes the cascade a lookup. Additive.
	4	export default `
	5	CREATE INDEX IF NOT EXISTS delivery_event ON webhook_delivery(event_id);
	6	`;
		@@ -0,0 +1,129 @@
	1	/**
	2	* How long an `event` row is kept, and the batch that deletes what is past it (#12008).
	3	*
	4	* Every reader of `event`, and the oldest row it can reach:
	5	*
	6	* Windowed by age — the longest decides KEEP_MS:
	7	* - `flow/estimate_base.ts`: reopened issues, 90 days (the base rate's widest level).
	8	* - `core/team-metrics.ts`: reopens for the team ledger, 7 days.
	9	* - `web/pages/security.ts`: deleted repositories, 30 days.
	10	* - `web/studio/since.ts`: "since you were here", the digest window (days).
	11	* - `feed/items.ts`, `feed/quiet.ts`: the home feed, its newest 300 and a 24 h quiet window.
	12	* - `feed/affinity.ts`: a viewer's newest 200, weighed by a 7-day half-life.
	13	* - `actions/mail.ts`: mail ceiling notices, 24 h. `crew/stale_pulls.ts`: closes, 72 h.
	14	* - `crew/landed.ts`: comments since a pull merged (read when it merges).
	15	* - `boards/sources.ts`: spawn timings by day, a board's range.
	16	*
	17	* Feeds, audit views and replay cursors — they page back through every row, so the rule sets
	18	* their reach: they answer the last KEEP_MS, and nothing they list newer than that changes:
	19	* - `core/events.ts` `listEvents`/`listEventsVisibleTo`, under the events page and
	20	* `/api/v1/repos/…/events` (`core/events_page.ts`), the MCP `events` tool
	21	* (`http/mcp/tools/events.ts`), the TUI events screen, the Atom feed (`web/feeds.ts`, newest
	22	* 200), the home timeline (`feed/timeline.ts`) and the SSE replay from Last-Event-ID
	23	* (`http/sse.ts`): a cursor older than the rule resumes from the oldest row kept.
	24	* - `core/cloud_account.ts`: the newest cloud secret reads, an audit list.
	25	* - `core/webhooks.ts`: a hook's recent deliveries (they cascade with their event); a delivery
	26	* still `pending` keeps its event, so a retry never loses what it sends.
	27	*
	28	* At any age — state read back out of the log, so these kinds are never pruned (KEPT):
	29	* - `issue.`, `pull.`: thread timelines (`web/pages/thread.ts`), assignees and blockers
	30	* (`actions/issues.ts`, `crew/dispatch.ts`, `crew/stranded.ts`), team waits, reopen counts
	31	* (`flow/read.ts`), splits (`core/issues.ts`, `core/limb_replan.ts`), merge closes
	32	* (`web/studio/tell.ts`, `core/done_on_main.ts`), the gate (`core/gate_hold.ts`), last push
	33	* to a pull (`flow/idle_pulls.ts`, `web/pages/flow-waiting.ts`).
	34	* - `repo.*`: detached forks (`actions/fork_chain.ts`), the vision kept (`web/pages/thread.ts`).
	35	* - `user.*`: how an account was made (`core/accounts.ts`).
	36	* - `push`: when a repository was last pushed (`core/listings.ts`, `core/repos.ts`,
	37	* `core/fork_follow.ts`, `flow/alarm_stuck.ts`).
	38	* - `app.egress.*`: the grants an app holds (`actions/egress_grants.ts`).
	39	* - `keeper_*`, `forge_down`, `forge_up`: what the keeper said once, tried, or last saw
	40	* (`maintain/keeper/*.ts`).
	41	* - `pinned.check`, `clog.cleared`, `webhook.failing`: the newest of each, or a streak's mark.
	42	* Newest-row readers (`web/map_cache.ts`, `spawn/clients.ts`) read MAX and never reach the old end.
	43	*
	44	* A new reader that reaches past KEEP_MS names its kind in KEPT, or it is windowed.
	45	*/
	46	import type { Database } from "bun:sqlite";
	47
	48	const DAY_MS = 86_400_000;
	49	/** Twice the longest windowed reader (estimate_base's 90 days). */
	50	export const KEEP_MS = 180 * DAY_MS;
	51	/** One batch: small enough to hold the write lock for milliseconds. */
	52	export const BATCH_ROWS = 5_000;
	53	/** Rows deleted per transaction: about 7 µs a row across the table's indexes. */
	54	export const CHUNK_ROWS = 1_000;
	55	/** What the sampler's 15-s tick asks for: 2.4M old rows in about ten hours, a few ms a tick. */
	56	export const TICK_ROWS = 1_000;
	57	/** After a pass reaches rows inside the rule it rests: the next pass starts a day later. */
	58	export const REST_MS = DAY_MS;
	59
	60	export const KEPT_PREFIXES: readonly string[] = ["issue.", "pull.", "repo.", "user.", "app.egress.", "keeper_"];
	61	export const KEPT_KINDS: readonly string[] = ["push", "forge_down", "forge_up", "pinned.check", "clog.cleared", "webhook.failing"];
	62
	63	/** Whether a row of `kind` is kept at any age. */
	64	export function keptKind(kind: string): boolean {
	65	return KEPT_KINDS.includes(kind) \|\| KEPT_PREFIXES.some((p) => kind.startsWith(p));
	66	}
	67
	68	type Pass = { after: number; restUntil: number };
	69	const passes = new WeakMap<Database, Pass>();
	70
	71	/**
	72	* One batch: walks at most `rowCap` rows from where the last batch stopped, in `seq` order (the
	73	* order they were written, so oldest first), and deletes those older than KEEP_MS that no reader
	74	* reaches at any age. `more` is whether rows past the rule are still to walk; when a pass meets
	75	* a row inside the rule it ends and rests REST_MS before starting from the oldest again. `lockMs`
	76	* is the longest the delete held the write lock (one CHUNK_ROWS transaction).
	77	*/
	78	export function pruneEvents(db: Database, now: number, rowCap: number = BATCH_ROWS): { deleted: number; more: boolean; lockMs: number } {
	79	const cap = Math.max(1, Math.min(BATCH_ROWS, Math.floor(rowCap)));
	80	const pass = passes.get(db) ?? { after: 0, restUntil: 0 };
	81	passes.set(db, pass);
	82	if (now < pass.restUntil) return { deleted: 0, more: false, lockMs: 0 };
	83	const cutoff = new Date(now - KEEP_MS).toISOString();
	84	const rows = db
	85	.query<{ seq: number; rowid: number; kind: string; created_at: string }, [number, number]>(
	86	"SELECT seq, rowid, kind, created_at FROM event WHERE seq > ?1 ORDER BY seq LIMIT ?2",
	87	)
	88	.all(pass.after, cap);
	89	const old: typeof rows = [];
	90	let reached = rows.length < cap;
	91	for (const row of rows) {
	92	if (row.created_at >= cutoff) {
	93	reached = true;
	94	break;
	95	}
	96	old.push(row);
	97	}
	98	const doomed = old.filter((r) => !keptKind(r.kind)).map((r) => r.rowid);
	99	let deleted = 0;
	100	let lockMs = 0;
	101	// A chunk per transaction, so the write lock is never held for the whole batch.
	102	for (let at = 0; at < doomed.length; at += CHUNK_ROWS) {
	103	const chunk = doomed.slice(at, at + CHUNK_ROWS);
	104	const started = performance.now();
	105	db.transaction(() => {
	106	// A delivery still pending keeps its event: the retry sends it.
	107	const pending = new Set(
	108	db
	109	.query<{ rowid: number }, []>(
	110	"SELECT e.rowid FROM webhook_delivery d JOIN event e ON e.id = d.event_id WHERE d.state = 'pending'",
	111	)
	112	.all()
	113	.map((r) => r.rowid),
	114	);
	115	const gone = pending.size === 0 ? chunk : chunk.filter((r) => !pending.has(r));
	116	// Counted from the list, not `changes`: that counts the deliveries cascading with them too.
	117	db.query<unknown, [string]>("DELETE FROM event WHERE rowid IN (SELECT value FROM json_each(?1))").run(JSON.stringify(gone));
	118	deleted += gone.length;
	119	})();
	120	lockMs = Math.max(lockMs, performance.now() - started);
	121	}
	122	if (reached) {
	123	pass.after = 0;
	124	pass.restUntil = now + REST_MS;
	125	return { deleted, more: false, lockMs };
	126	}
	127	pass.after = old[old.length - 1]!.seq;
	128	return { deleted, more: true, lockMs };
	129	}
		@@ -0,0 +1,137 @@
	1	import { describe, expect, test } from "bun:test";
	2	import type { Database } from "bun:sqlite";
	3	import { migrations } from "../src/core/schema.ts";
	4	import { listEvents } from "../src/core/events.ts";
	5	import { countNotes } from "../src/maintain/keeper/record.ts";
	6	import { openDb } from "../src/kernel/db.ts";
	7	import { BATCH_ROWS, KEEP_MS, keptKind, pruneEvents, REST_MS } from "../src/system/retention_events.ts";
	8
	9	const DAY = 86_400_000;
	10	const NOW = Date.parse("2026-11-13T12:00:00.000Z");
	11	const iso = (ms: number) => new Date(ms).toISOString();
	12
	13	function seeded(): Database {
	14	const db = openDb(":memory:", migrations);
	15	// Seeded rows name a repo and a webhook by id alone; the tables they point at are not under test.
	16	db.exec("PRAGMA foreign_keys = OFF");
	17	return db;
	18	}
	19
	20	let n = 0;
	21	function put(db: Database, kind: string, ageMs: number, repo: string \| null = "r1", payload: Record<string, unknown> = {}): string {
	22	const id = `e${String(++n).padStart(8, "0")}`;
	23	db.query("INSERT INTO event (id, kind, repo_id, payload, created_at) VALUES (?1, ?2, ?3, ?4, ?5)").run(
	24	id, kind, repo, JSON.stringify(payload), iso(NOW - ageMs),
	25	);
	26	return id;
	27	}
	28
	29	const ids = (rows: { id: string }[]) => rows.map((r) => r.id);
	30
	31	describe("pruneEvents", () => {
	32	test("deletes old rows of windowed kinds; every named reader answers the same for its window", () => {
	33	const db = seeded();
	34	// Oldest first, as the log is written: old windowed rows, old kept rows, then newer rows.
	35	const oldWindowed = [put(db, "secret.read", KEEP_MS + 30 * DAY), put(db, "forge.spawn.live", KEEP_MS + 2 * DAY, null)];
	36	const oldPending = put(db, "run.finished", KEEP_MS + DAY);
	37	db.query("INSERT INTO webhook_delivery (id, webhook_id, event_id, state) VALUES ('d1', 'w1', ?1, 'pending')").run(oldPending);
	38	const oldDone = put(db, "run.started", KEEP_MS + DAY);
	39	db.query("INSERT INTO webhook_delivery (id, webhook_id, event_id, state) VALUES ('d2', 'w1', ?1, 'ok')").run(oldDone);
	40	put(db, "repo.detached", KEEP_MS + 20 * DAY);
	41	put(db, "user.created", KEEP_MS + 20 * DAY, null);
	42	put(db, "issue.closed", KEEP_MS + 10 * DAY, "r1", { number: 1, reason: "split" });
	43	put(db, "keeper_said", KEEP_MS + 10 * DAY, null, { key: "k1" });
	44	put(db, "push", KEEP_MS + 5 * DAY);
	45	const recent: string[] = [];
	46	for (let d = 170; d >= 0; d -= 10) recent.push(put(db, d % 20 === 0 ? "secret.read" : "issue.reopened", d * DAY, "r1", { number: d }));
	47
	48	const cutoff = iso(NOW - KEEP_MS);
	49	const readers = () => ({
	50	// Feeds, the events page, the API, MCP and SSE replay all page `listEvents`:
	51	// their window is the rule.
	52	feed: ids(listEvents(db, { limit: 500 })).filter((id) => db.query("SELECT 1 FROM event WHERE id = ?1 AND created_at >= ?2").get(id, cutoff)),
	53	repoFeed: ids(listEvents(db, { repoId: "r1", limit: 50 })).filter((id) => db.query("SELECT 1 FROM event WHERE id = ?1 AND created_at >= ?2").get(id, cutoff)),
	54	replay: ids(listEvents(db, { afterId: recent[3]!, limit: 500 })),
	55	// estimate_base: reopened issues within 90 days.
	56	estimate: db.query("SELECT DISTINCT json_extract(payload, '$.number') AS n FROM event WHERE repo_id = 'r1' AND kind = 'issue.reopened' AND created_at >= ?1").all(iso(NOW - 90 * DAY)),
	57	// security: 30 days of repo.deleted; mail: 24 h.
	58	windowed: db.query("SELECT id FROM event WHERE created_at >= ?1 ORDER BY seq").all(cutoff),
	59	// At any age: fork_chain, accounts, core/issues splits, the keeper's once-said, last push.
	60	detached: db.query("SELECT 1 FROM event WHERE repo_id = 'r1' AND kind = 'repo.detached' LIMIT 1").get() !== null,
	61	created: db.query("SELECT COUNT(*) AS n FROM event WHERE kind = 'user.created'").get(),
	62	splits: db.query("SELECT COUNT(*) AS n FROM event WHERE kind = 'issue.closed' AND json_extract(payload, '$.reason') = 'split'").get(),
	63	said: countNotes(db, "keeper_said", "key", "k1"),
	64	lastPush: db.query("SELECT MAX(created_at) AS at FROM event WHERE repo_id = 'r1' AND kind = 'push'").get(),
	65	});
	66	const before = readers();
	67	expect(before.said).toBe(1);
	68	expect(before.detached).toBe(true);
	69
	70	db.exec("PRAGMA foreign_keys = ON"); // deliveries cascade with their event, as in production
	71	const first = pruneEvents(db, NOW, BATCH_ROWS);
	72	expect(first).toMatchObject({ deleted: 3, more: false });
	73	expect(readers()).toEqual(before);
	74	for (const id of oldWindowed) expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(id)).toBeNull();
	75	// A delivery still pending keeps its event; a finished one goes with it.
	76	expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(oldPending)).not.toBeNull();
	77	expect(db.query("SELECT 1 FROM webhook_delivery WHERE id = 'd2'").get()).toBeNull();
	78	});
	79
	80	test("walks in capped batches, then rests a day before the next pass", () => {
	81	const db = seeded();
	82	for (let i = 0; i < 25; i++) put(db, "secret.read", KEEP_MS + DAY);
	83	put(db, "secret.read", DAY);
	84	expect(pruneEvents(db, NOW, 10)).toMatchObject({ deleted: 10, more: true });
	85	expect(pruneEvents(db, NOW, 10)).toMatchObject({ deleted: 10, more: true });
	86	expect(pruneEvents(db, NOW, 10)).toMatchObject({ deleted: 5, more: false });
	87	put(db, "secret.read", KEEP_MS + DAY);
	88	expect(pruneEvents(db, NOW + 1000, 10)).toMatchObject({ deleted: 0, more: false });
	89	// The next pass, once rows written later have aged past the rule too.
	90	expect(pruneEvents(db, NOW + Math.max(REST_MS, KEEP_MS), 10)).toMatchObject({ deleted: 2, more: false });
	91	expect((db.query("SELECT COUNT(*) AS n FROM event").get() as { n: number }).n).toBe(0);
	92	});
	93
	94	test("the rule keeps state kinds at any age and windows the rest", () => {
	95	for (const k of ["issue.closed", "pull.pushed", "repo.detached", "user.created", "push", "keeper_lull", "forge_down", "app.egress.granted"])
	96	expect(keptKind(k)).toBe(true);
	97	for (const k of ["secret.read", "forge.spawn.live", "mail.ceiling", "measurement"]) expect(keptKind(k)).toBe(false);
	98	expect(KEEP_MS).toBeGreaterThanOrEqual(90 * DAY);
	99	});
	100
	101	test("one 5,000-row batch holds the write lock under 50 ms", () => {
	102	const db = seeded();
	103	const add = db.query("INSERT INTO event (id, kind, repo_id, payload, created_at) VALUES (?1, 'secret.read', 'r1', ?2, ?3)");
	104	db.transaction(() => {
	105	for (let i = 0; i < 20_000; i++) add.run(`b${String(i).padStart(6, "0")}`, `{"name":"CLOUD_${i}","number":${i}}`, iso(NOW - KEEP_MS - DAY - (20_000 - i)));
	106	for (let i = 0; i < 2_000; i++) db.query("INSERT INTO webhook_delivery (id, webhook_id, event_id, state) VALUES (?1, 'w1', ?2, 'ok')").run(`d${i}`, `b${String(i * 7).padStart(6, "0")}`);
	107	})();
	108	db.exec("PRAGMA foreign_keys = ON");
	109	pruneEvents(db, NOW, 100); // warm the statements
	110	const { deleted, lockMs } = pruneEvents(db, NOW, BATCH_ROWS);
	111	expect(deleted).toBe(BATCH_ROWS);
	112	expect(lockMs).toBeGreaterThan(0);
	113	expect(lockMs).toBeLessThan(50);
	114	});
	115	});
	116
	117	describe("the sampler's tick", () => {
	118	test("prunes a batch of old events, the same periodic job as the machine_sample rule", async () => {
	119	const { startSampler } = await import("../src/boards/sampler.ts");
	120	const db = seeded();
	121	// The sampler ticks once at start with the clock's own now, so the rows are aged from it.
	122	const now = Date.now();
	123	const gone = put(db, "secret.read", KEEP_MS + DAY + (NOW - now), null);
	124	const kept = put(db, "secret.read", DAY + (NOW - now), null);
	125	db.exec("PRAGMA foreign_keys = ON");
	126	const app = { db, log: { warn: () => {}, info: () => {} } } as unknown as import("../src/app.ts").App;
	127	const nowhere = { cgroup: "/nowhere", proc: "/nowhere", sys: "/nowhere" };
	128	const s = startSampler(app, { paths: nowhere, everyMs: 3_600_000, cells: () => [] });
	129	try {
	130	s.tick(now);
	131	expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(gone)).toBeNull();
	132	expect(db.query("SELECT 1 FROM event WHERE id = ?1").get(kept)).not.toBeNull();
	133	} finally {
	134	s.stop();
	135	}
	136	});
	137	});