src/job-breaker.ts

Functions

Imports

It imports src/env.ts, src/store-guards.ts, src/time.ts. Imported by src/bench.ts, src/controls-perform.ts, src/controls-plan.ts, src/jobs-claim.ts, src/status.ts, src/tools/status.ts.

Source

1import type { Env } from "./env";
2import { auditStatement } from "./store-guards";
3import { DAY_MS } from "./time";
4
5// The queue's circuit breaker (job_9e602b31888f, OWASP ASI08, cascading failures).
6//
7// A namespace whose drivers failed N jobs in the last 24 hours stops handing out work
8// until the seat or Dustin resets it. The breaker holds no "open" flag: it is computed
9// at claim time from the audit log, so there is no state to drift from what happened,
10// and nothing but a person's reset closes it. The reset is a timestamp; only failures
11// after it count, so a reset cannot be undone by the failures that tripped it.
12//
13// What counts is the holder's own `fail` (audit `job-fail`). A seat's admin fail is a
14// person cleaning up, and a claim that fails a tampered or corrupt body is the Worker
15// refusing, not a driver failing, so neither trips it. Nor does a bench run's key: its
16// fail is the honest "not done" its job asks for (src/jobs-seat.ts completeAsCaller), and
17// three of them in a day would otherwise stop the study's runs. A holder's fail that
18// Capsid's own complete refusal forced, or that states its remainder item by item against
19// a diff that shows the rest done, is left out too (src/fail-exemption.ts).
20
21const BREAKER_THRESHOLD_DEFAULT = 3;
22const BREAKER_WINDOW_MS = DAY_MS;
23export const BREAKER_THRESHOLD_KEY = "jobs:breaker:threshold";
24export const breakerResetKey = (namespace: string) => `jobs:breaker:reset:${namespace}`;
25
26export interface BreakerState {
27  namespace: string;
28  open: boolean;
29  // Holder fails counted since `since`.
30  failed: number;
31  threshold: number;
32  // The later of 24 hours ago and the last reset, as audit_log writes times.
33  since: string;
34  reset_at: string | null;
35  // Why a setting could not be read, when one could not. The breaker then uses the
36  // stricter value, so an outage never lets a failing namespace keep claiming.
37  note?: string;
38}
39
40// audit_log.at is datetime('now'): "YYYY-MM-DD HH:MM:SS", UTC.
41function sqlTime(d: Date): string {
42  return d.toISOString().slice(0, 19).replace("T", " ");
43}
44
45/** The threshold, 1 to 50; the default when unset. An unreadable key uses the default and says so. */
46async function readBreakerThreshold(kv: KVNamespace): Promise<{ threshold: number; note?: string }> {
47  let raw: string | null;
48  try {
49    raw = await kv.get(BREAKER_THRESHOLD_KEY);
50  } catch (err) {
51    return { threshold: BREAKER_THRESHOLD_DEFAULT, note: `could not read ${BREAKER_THRESHOLD_KEY}: ${err instanceof Error ? err.message : String(err)}` };
52  }
53  if (raw === null) return { threshold: BREAKER_THRESHOLD_DEFAULT };
54  const n = Number(raw);
55  if (!Number.isInteger(n) || n < 1 || n > 50) {
56    return { threshold: BREAKER_THRESHOLD_DEFAULT, note: `${BREAKER_THRESHOLD_KEY} holds ${JSON.stringify(raw)}, not a whole number from 1 to 50` };
57  }
58  return { threshold: n };
59}
60
61export async function breakerState(env: Env, namespace: string, now: Date): Promise<BreakerState> {
62  const notes: string[] = [];
63  const { threshold, note } = await readBreakerThreshold(env.APP_KV);
64  if (note) notes.push(note);
65  let resetAt: string | null = null;
66  try {
67    resetAt = await env.APP_KV.get(breakerResetKey(namespace));
68  } catch (err) {
69    // No reset read means every failure in the window counts: the stricter reading.
70    notes.push(`could not read the reset for ${namespace}: ${err instanceof Error ? err.message : String(err)}`);
71  }
72  const windowStart = new Date(now.getTime() - BREAKER_WINDOW_MS);
73  const reset = resetAt ? new Date(resetAt) : null;
74  const from = reset && !Number.isNaN(reset.getTime()) && reset > windowStart ? reset : windowStart;
75  if (resetAt && (!reset || Number.isNaN(reset.getTime()))) notes.push(`the reset for ${namespace} holds ${JSON.stringify(resetAt)}, not a time; ignored`);
76  const since = sqlTime(from);
77  // A transition writes two audit rows with this action, one for the job and one for
78  // its mirror document, both on the job's path; counting paths counts each job once.
79  // A fail exempted when it was written leaves out both of its rows (src/fail-exemption.ts).
80  const row = await env.DB.prepare(
81    `SELECT COUNT(DISTINCT path) AS n FROM audit_log WHERE action = 'job-fail' AND namespace = ?1 AND at >= ?2 AND actor NOT LIKE 'agent:bench-run-%'
82     AND path NOT IN (SELECT path FROM audit_log WHERE action = 'job-fail' AND namespace = ?1 AND at >= ?2 AND json_extract(params, '$.breaker_exempt') IS NOT NULL)`
83  )
84    .bind(namespace, since)
85    .first<{ n: number }>();
86  const failed = row?.n ?? 0;
87  return {
88    namespace,
89    open: failed >= threshold,
90    failed,
91    threshold,
92    since,
93    reset_at: resetAt,
94    ...(notes.length ? { note: notes.join("; ") } : {}),
95  };
96}
97
98/** The refusal a claim returns while the breaker is open. */
99export function breakerRefusal(state: BreakerState): string {
100  return (
101    `the circuit breaker for ${state.namespace} is open: ${state.failed} jobs failed by their holders since ${state.since} UTC, ` +
102    `at or over the threshold of ${state.threshold}. Nothing in ${state.namespace} is handed out until the seat or Dustin resets it ` +
103    `(capsid_admin_control_improve action breaker_reset, or Reset in the Portal, under Paused after repeated failures on Needs you).`
104  );
105}
106
107/**
108 * The seat's or Dustin's control (capsid_admin_control_improve action breaker_reset, admin only, and the
109 * Portal's reset_breaker): close one namespace's breaker, so failures before now stop
110 * counting, and optionally set the threshold, which applies to every namespace. Audited
111 * under the caller, and read back.
112 */
113export async function resetBreaker(
114  env: Env,
115  actor: string,
116  now: Date,
117  args: { namespace: string; threshold?: number }
118): Promise<{ ok: true; action: "breaker_reset"; state: BreakerState } | { ok: false; error: string }> {
119  if (args.threshold !== undefined && (!Number.isInteger(args.threshold) || args.threshold < 1 || args.threshold > 50)) {
120    return { ok: false, error: "threshold must be a whole number from 1 to 50." };
121  }
122  const before = await breakerState(env, args.namespace, now);
123  const stamp = now.toISOString();
124  await env.APP_KV.put(breakerResetKey(args.namespace), stamp);
125  if (args.threshold !== undefined) await env.APP_KV.put(BREAKER_THRESHOLD_KEY, String(args.threshold));
126  await auditStatement(env.DB, actor, "job-breaker-reset", args.namespace, null, {
127    reset_at: stamp,
128    was_open: before.open,
129    failed: before.failed,
130    threshold: args.threshold ?? before.threshold,
131  }).run();
132  return { ok: true, action: "breaker_reset", state: await breakerState(env, args.namespace, now) };
133}