src/task-runs.ts

Functions

Imports

It imports src/env.ts, src/log.ts, src/ops-types/feed.ts, src/time.ts. Imported by src/index.ts, src/ops-feed.ts, src/site-repair.ts, src/tick.ts.

Source

1import type { Env } from "./env";
2import type { OpsTask, OpsTaskRun, OpsTaskRunOutcome } from "./ops-types/feed";
3import { logEvent } from "./log";
4import { MINUTE_MS, DAY_MS } from "./time";
5
6// The run ledger (migrations/0029_task_runs.sql). Every scheduled task goes through runTask,
7// which writes one task_runs row per run: start, finish, outcome and a one-line
8// reason. The Portal reads the newest runs of each task (readTaskRuns) and flags a
9// task that has failed or gone quiet (taskStates).
10//
11// Adapted from Foxhound's runCronJob (app/lib/cron/run.ts): one wrapper, one row per
12// executed run, and the reason column, because "ok" alone cannot tell a sweep that
13// found nothing from one that never looked.
14
15export type TaskId =
16  | "backup"
17  | "skills-refresh"
18  | "tick"
19  | "auto-merge"
20  | "watcher"
21  | "outcome-sweep"
22  | "merge-resume"
23  | "maintenance"
24  | "site-repair"
25  | "public-build"
26  | "push"
27  | "digest"
28  | "merge-units"
29  | "merge-safety"
30  | "doc-index"
31  | "supersede";
32
33// Each task, and how long it may go without a run before the Portal says it has
34// gone quiet: twice its period, Foxhound's rule (app/lib/cron/alerts.server.ts),
35// because one missed run is normal and a flag on it would teach the reader to
36// ignore the flag. Null where runs come and go with the work, so a quiet spell
37// means nothing: those are flagged only when they fail.
38//   tick           */5 * * * *, which carries the five steps below it
39//   backup         0 9 * * *
40//   skills-refresh 30 9 * * *, doing its work on one UTC day a week
41//   watcher        on the tick, every cadence_min minutes (src/watcher.ts); its
42//                  period is read from the watcher's own snapshot
43export const TASKS: ReadonlyArray<{ id: TaskId; label: string; period_ms: number | null }> = [
44  { id: "tick", label: "Five-minute tick", period_ms: 5 * MINUTE_MS },
45  { id: "watcher", label: "Watcher pass", period_ms: null },
46  { id: "backup", label: "Backup", period_ms: DAY_MS },
47  { id: "skills-refresh", label: "Skills refresh", period_ms: 7 * DAY_MS },
48  { id: "auto-merge", label: "Auto-merge step", period_ms: null },
49  { id: "outcome-sweep", label: "Merge-state sweep", period_ms: null },
50  { id: "merge-resume", label: "Stale-job merge resume", period_ms: null },
51  { id: "maintenance", label: "Daily maintenance pass", period_ms: DAY_MS },
52  // One row per call to a site's operator API (src/site-repair.ts), from the watcher or a
53  // person, so its runs come and go with the drift.
54  { id: "site-repair", label: "Site repairs", period_ms: null },
55  // The public system model (src/public-data.ts), built once a UTC day on the tick.
56  { id: "public-build", label: "Public model build", period_ms: DAY_MS },
57  // Phone alerts (src/push.ts), on the tick; recorded only when an alert was due.
58  { id: "push", label: "Phone alerts", period_ms: null },
59  // The daily email digest (src/digest.ts), on the tick; off until Email Routing is on,
60  // so a quiet spell means nothing by itself and only a refusal is flagged.
61  { id: "digest", label: "Daily email digest", period_ms: null },
62  // Merge unit promotion and demotion (src/merge-units.ts), on the tick; recorded only
63  // when its report or a suspension changed.
64  { id: "merge-units", label: "Merge unit proposals", period_ms: null },
65  // The auto-merge safety meter and shutoff (src/merge-safety.ts), on the tick; recorded
66  // only when a revert was recorded, a namespace paused or the meter changed.
67  { id: "merge-safety", label: "Auto-merge safety meter", period_ms: null },
68  // The meaning index (src/doc-index.ts), on the tick; recorded only when it sent or
69  // removed something, so a quiet spell means nothing changed.
70  { id: "doc-index", label: "Search index", period_ms: null },
71  // The daily supersede pass (src/supersede.ts), with the backup; recorded only when it
72  // marked something.
73  { id: "supersede", label: "Supersede pass", period_ms: null },
74];
75
76const STALE_PERIODS = 2;
77// Rows older than this are pruned, PRUNE_BATCH at a time on each write.
78export const RUN_RETENTION_DAYS = 14;
79const PRUNE_BATCH = 50;
80// How many of each task's newest runs the Portal shows.
81export const RUNS_PER_TASK = 5;
82// A reason is one line: what the run did or why it did not.
83export const REASON_MAX = 300;
84
85/** What a task's body reports about the run it made. Null means it was not due and
86 *  made no run (the watcher between passes, the sweep before its day), so nothing is
87 *  recorded. */
88export type TaskResult = { outcome: Exclude<OpsTaskRunOutcome, "threw">; reason: string } | null;
89
90function oneLine(text: string): string {
91  const line = text.replace(/\s+/g, " ").trim();
92  return line.length > REASON_MAX ? `${line.slice(0, REASON_MAX - 1)}…` : line;
93}
94
95/** The insert and a bounded prune, in one batch. */
96export function taskRunStatements(db: D1Database, task: TaskId, started: Date, finished: Date, outcome: OpsTaskRunOutcome, reason: string): D1PreparedStatement[] {
97  const cutoff = new Date(finished.getTime() - RUN_RETENTION_DAYS * DAY_MS).toISOString();
98  return [
99    db
100      .prepare(`INSERT INTO task_runs (task, started_at, finished_at, outcome, reason) VALUES (?1, ?2, ?3, ?4, ?5)`)
101      .bind(task, started.toISOString(), finished.toISOString(), outcome, oneLine(reason)),
102    // The oldest PRUNE_BATCH rows by id, of which those past retention go: rows are
103    // appended in time order, so this stays a short index walk however large the
104    // table is (the same prune as session_events, src/ops-hooks.ts).
105    db
106      .prepare(`DELETE FROM task_runs WHERE id IN (SELECT id FROM task_runs ORDER BY id LIMIT ?1) AND finished_at < ?2`)
107      .bind(PRUNE_BATCH, cutoff),
108  ];
109}
110
111/** Write one run. Never throws, so a failed ledger write cannot stop the task it
112 *  records or the tasks after it. A write that fails is said in the log, by task, and
113 *  the Portal then shows that task as quiet: the ledger's own failure is visible. */
114async function record(env: Env, task: TaskId, started: Date, outcome: OpsTaskRunOutcome, reason: string): Promise<void> {
115  try {
116    await env.DB.batch(taskRunStatements(env.DB, task, started, new Date(), outcome, reason));
117  } catch (err) {
118    logEvent("error", "TASK_RUN_UNRECORDED", { message: `TASK_RUN_UNRECORDED ${task} ${outcome}: ${err instanceof Error ? err.message : String(err)}` });
119  }
120}
121
122/**
123 * Run one scheduled task and record it. A body that throws is recorded as `threw`
124 * with the error, logged with `tag`, and then rethrown when `rethrow` is set (a cron
125 * branch, so the invocation shows as failed) or swallowed into the record otherwise
126 * (a tick step, so one step's throw does not stop the steps after it; the record and
127 * the log line are where it is said).
128 */
129export async function runTask<T>(
130  env: Env,
131  task: TaskId,
132  body: () => Promise<T>,
133  judge: (value: T) => TaskResult,
134  opts: { tag: string; rethrow: boolean }
135): Promise<T | null> {
136  const started = new Date();
137  let value: T;
138  try {
139    value = await body();
140  } catch (err) {
141    const message = err instanceof Error ? err.message : String(err);
142    logEvent("error", opts.tag, {
143      message: `${opts.tag} ${err instanceof Error && err.stack ? `${message}\n${err.stack}` : message}`,
144    });
145    await record(env, task, started, "threw", message);
146    if (opts.rethrow) throw err;
147    return null;
148  }
149  const result = judge(value);
150  if (result) await record(env, task, started, result.outcome, result.reason);
151  return value;
152}
153
154interface TaskRunRow {
155  task: string;
156  started_at: string;
157  finished_at: string;
158  outcome: OpsTaskRunOutcome;
159  reason: string;
160}
161
162/** Each task's newest runs: one keyed read per task, through task_runs_task, in one
163 *  batch. TASKS.length statements, counted in OPS_FEED_READS (src/ops-feed.ts). */
164export async function readTaskRuns(db: D1Database): Promise<Map<TaskId, OpsTaskRun[]>> {
165  const results = await db.batch<TaskRunRow>(
166    TASKS.map((t) =>
167      db
168        .prepare(`SELECT task, started_at, finished_at, outcome, reason FROM task_runs WHERE task = ?1 ORDER BY id DESC LIMIT ?2`)
169        .bind(t.id, RUNS_PER_TASK)
170    )
171  );
172  const out = new Map<TaskId, OpsTaskRun[]>();
173  TASKS.forEach((t, i) => {
174    out.set(
175      t.id,
176      (results[i]?.results ?? []).map((r) => ({ started_at: r.started_at, finished_at: r.finished_at, outcome: r.outcome, reason: r.reason }))
177    );
178  });
179  return out;
180}
181
182/** Each task with its newest runs and its flag, decided at `now`.
183 *  - `failing`: the newest run threw or was refused.
184 *  - `quiet`: a periodic task whose newest run is more than STALE_PERIODS periods old.
185 *  - `never`: no run in the ledger. A notice, not a fault: after the migration every
186 *    task starts here, and the daily ones stay here until their hour. */
187export function taskStates(runs: Map<TaskId, OpsTaskRun[]>, now: Date, watcherCadenceMin: number): OpsTask[] {
188  return TASKS.map((t) => {
189    const recent = runs.get(t.id) ?? [];
190    const last = recent[0] ?? null;
191    const period = t.id === "watcher" ? watcherCadenceMin * MINUTE_MS : t.period_ms;
192    let flag: OpsTask["flag"] = null;
193    if (last === null) flag = "never";
194    else if (last.outcome === "threw" || last.outcome === "refused") flag = "failing";
195    else if (period !== null && now.getTime() - Date.parse(last.finished_at) > STALE_PERIODS * period) flag = "quiet";
196    return { id: t.id, label: t.label, period_ms: period, flag, recent };
197  });
198}