e9d761d
@@ -348,6 +348,12 @@
door: an admin with a notify URL is paged once when a drain passes five
minutes and once, by the draining process as it exits, when it ends. A process
that dies mid-drain says nothing more.
The same push pages the vitals causes `errors`, `internal`, `queue:*` and `disk`
once each has held five minutes, and says Recovered once it has been gone five
(a flap in between is no news). A cause present at boot is taken as already
paged. Every other cause (`drill`, `mirror`, `stuck:*`, `red:*`, `snapshots`,
`budget:*`) is paged elsewhere or is a setting, and is ignored here.
## Origins
Your forge is the index of everything you work on, wherever it actually lives.
@@ -42,3 +42,59 @@
export function nextPage(
    return { line: RECOVERED, paged: undefined };
  return undefined;
}
/** How long a cause must hold to be paged, and stay gone to be recovered. */
export const HOLD_MS = 5 * 60_000;
/** The vitals causes this pager owns; the rest are paged elsewhere or are settings. */
export function owned(cause: string): boolean {
  return cause === "errors" || cause === "internal" || cause === "disk" || cause.startsWith("queue:");
}
/** One owned cause being watched. */
export interface Watched {
  readonly since: number;
  readonly sentence: string;
  /** Set once the page was delivered (or was already sent before a restart). */
  paged: boolean;
  /** When it was last seen gone, until it is seen again. */
  absentSince?: number;
}
export type Watch = Map<string, Watched>;
/**
 * Looks at the causes present now. Returns the pages due and the Recovereds
 * due; nothing is marked, so a send that failed is due again next time.
 * `seed` takes every owned cause present as already paged, whatever its age.
 */
export function observe(
  watch: Watch,
  present: readonly { cause: string; sentence: string }[],
  now: number,
  seed = false,
): { pages: string[]; recovered: string[] } {
  const here = new Set<string>();
  for (const p of present) {
    if (!owned(p.cause)) continue;
    here.add(p.cause);
    const w = watch.get(p.cause);
    if (w === undefined) watch.set(p.cause, { since: now, sentence: p.sentence, paged: seed });
    else delete w.absentSince;
  }
  const pages: string[] = [];
  const recovered: string[] = [];
  for (const [key, w] of [...watch]) {
    if (here.has(key)) {
      if (!w.paged && now - w.since >= HOLD_MS) pages.push(key);
      continue;
    }
    w.absentSince ??= now;
    if (now - w.absentSince < HOLD_MS) continue;
    if (w.paged) recovered.push(key);
    else watch.delete(key);
  }
  return { pages, recovered };
}
export const recoveredLine = (cause: string): string => `Recovered: ${cause} has cleared`;
@@ -33,7 +33,8 @@
import { err, ok, type Result } from "../kernel/result.ts";
import { lastMirror, mirrorNeed } from "../snapshot/mirror.ts";
import { drillNeed, firstSeen, lastVerdict } from "../snapshot/verdict.ts";
import { health } from "../http/health.ts";
import { RECOVERED, drainIncident, nextPage } from "./incidents.ts";
import { forgeVitals } from "../actions/health.ts";
import { RECOVERED, drainIncident, nextPage, observe, recoveredLine, type Watch } from "./incidents.ts";
import { askLinesFor } from "../web/studio/asks.ts";
const KEY = "notify_url";
@@ -204,6 +205,8 @@
export interface NotifyOptions {
  clockMs?: number;
  /** When this process began draining; the live health unless a test says. */
  drainingSince?: () => number | null;
  /** The causes the vitals hold now; the live vitals unless a test says. */
  causes?: () => readonly { cause: string; sentence: string }[];
}
const oneLine = (s: string): string =>
@@ -220,6 +223,8 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
  const pause = new Map<string, { until: number; next: number }>();
  /** Per admin: the drain last paged and not yet recovered, set only after delivery. */
  const paged = new Map<string, string | undefined>();
  /** Per admin: the owned vitals causes being watched. */
  const watching = new Map<string, Watch>();
  /** Admins this process itself paged and has not yet told Recovered. */
  const mine = new Set<string>();
  let queue: Promise<void> = Promise.resolve();
@@ -251,8 +256,60 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
    const url = notifyUrlOf(app.db, userId);
    if (user === undefined || url === undefined || !user.isAdmin || user.kind !== "user") {
      paged.delete(userId);
      watching.delete(userId);
      return;
    }
    await pageDrain(userId, url, baseline);
    await pageCauses(userId, url, baseline);
  }
  /** Sends one line under the hourly cap and the backoff; whether it was delivered. */
  async function push(userId: string, url: string, line: string): Promise<boolean> {
    const t = now();
    const recent = (sent.get(userId) ?? []).filter((s) => t - s < HOUR_MS);
    sent.set(userId, recent);
    if (recent.length >= HOURLY || t < (pause.get(userId)?.until ?? 0)) return false;
    recent.push(t);
    let delivered = false;
    try {
      delivered = await send(url, `${oneLine(line)}\n${link}`, {
        click: link,
        title: "forge",
      });
    } catch {
      delivered = false;
    }
    if (delivered) {
      pause.delete(userId);
      return true;
    }
    const wait = Math.min(pause.get(userId)?.next ?? BACKOFF_FIRST_MS, BACKOFF_MAX_MS);
    pause.set(userId, { until: now() + wait, next: wait * 2 });
    app.log.warn("notify endpoint failed; pausing", {
      host: notifyHost(url) ?? "",
      pauseMs: wait,
    });
    return false;
  }
  /** Errors, internal faults, a long queue or a full disk: paged once held, recovered once gone. */
  async function pageCauses(userId: string, url: string, baseline: boolean): Promise<void> {
    const present = options.causes ? options.causes() : forgeVitals(app).degraded;
    let watch = watching.get(userId);
    const seed = baseline || watch === undefined;
    if (watch === undefined) watching.set(userId, (watch = new Map()));
    const due = observe(watch, present, now(), seed);
    for (const key of due.pages) {
      if (!(await push(userId, url, watch.get(key)!.sentence))) return;
      watch.get(key)!.paged = true;
    }
    for (const key of due.recovered) {
      if (!(await push(userId, url, recoveredLine(key)))) return;
      watch.delete(key);
    }
  }
  async function pageDrain(userId: string, url: string, baseline: boolean): Promise<void> {
    const current = drainIncident(drainStart(), now());
    if (baseline || !paged.has(userId)) {
      // A restart has paged what it paged: start from what is true now.
@@ -261,34 +318,11 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
    }
    const next = nextPage(current, paged.get(userId));
    if (next === undefined) return;
    const t = now();
    const recent = (sent.get(userId) ?? []).filter((s) => t - s < HOUR_MS);
    sent.set(userId, recent);
    if (recent.length >= HOURLY || t < (pause.get(userId)?.until ?? 0)) return;
    recent.push(t);
    let delivered = false;
    try {
      delivered = await send(url, `${oneLine(next.line)}\n${link}`, {
        click: link,
        title: "forge",
      });
    } catch {
      delivered = false;
    }
    if (delivered) {
      paged.set(userId, next.paged);
      // Only a page this process delivered is this process's to recover.
      if (next.paged === undefined) mine.delete(userId);
      else mine.add(userId);
      pause.delete(userId);
      return;
    }
    const wait = Math.min(pause.get(userId)?.next ?? BACKOFF_FIRST_MS, BACKOFF_MAX_MS);
    pause.set(userId, { until: now() + wait, next: wait * 2 });
    app.log.warn("notify endpoint failed; pausing", {
      host: notifyHost(url) ?? "",
      pauseMs: wait,
    });
    if (!(await push(userId, url, next.line))) return;
    paged.set(userId, next.paged);
    // Only a page this process delivered is this process's to recover.
    if (next.paged === undefined) mine.delete(userId);
    else mine.add(userId);
  }
  async function tell(userId: string, baseline: boolean): Promise<void> {
@@ -0,0 +1,116 @@
/**
 * The vitals causes the pager owns (`errors`, `internal`, `queue:*`, `disk`)
 * page once they have held five minutes and say Recovered once they have
 * been gone five; every other cause is left to whoever already pages it.
 */
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { createUser } from "../src/core/users.ts";
import { owned } from "../src/crew/incidents.ts";
import { setNotifyUrl, startNotifier, type Notifier } from "../src/crew/notify.ts";
const MIN = 60_000;
describe("owned causes", () => {
  test("an explicit list", () => {
    for (const c of ["errors", "internal", "disk", "queue:check", "queue:run"])
      expect(owned(c)).toBe(true);
    for (const c of ["drill", "mirror", "stuck:r1", "red:r1", "snapshots", "budget:r1", "shared:r:t"])
      expect(owned(c)).toBe(false);
  });
});
describe("the cause pager", () => {
  let dataDir: string;
  let app: App;
  let notifier: Notifier;
  let clock = 1_000_000;
  let causes: string[] = [];
  const posts: string[] = [];
  async function boot(): Promise<void> {
    notifier = startNotifier(app, {
      forgeUrl: "https://forge.example.test",
      now: () => clock,
      drainingSince: () => null,
      causes: () => causes.map((cause) => ({ cause, sentence: `${cause} is bad` })),
      send: async (_u, body) => {
        posts.push(body);
        return true;
      },
    });
    await notifier.settled();
  }
  const at = async (min: number): Promise<void> => {
    clock += min * MIN;
    await notifier.tick();
  };
  beforeEach(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-causes-"));
    app = createApp({ dataDir, port: 0 });
    const made = await createUser(app.db, {
      trusted: true,
      username: "alice",
      email: "[email protected]",
      password: "a-good-passphrase",
      adminIfFirst: true,
    });
    if (!made.ok) throw new Error(made.error);
    setNotifyUrl(app.db, made.value.id, "https://ntfy.example.test/root-topic");
    posts.length = 0;
    clock = 1_000_000;
    causes = [];
    await boot();
  });
  afterEach(async () => {
    await notifier.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  test("a flap pages once; six minutes gone says Recovered once", async () => {
    causes = ["errors"];
    await at(0);
    await at(4);
    expect(posts).toHaveLength(0);
    await at(2);
    expect(posts).toHaveLength(1);
    expect(posts[0]).toContain("errors is bad");
    causes = [];
    await at(1);
    causes = ["errors"];
    await at(1);
    await at(6);
    expect(posts).toHaveLength(1);
    causes = [];
    await at(0);
    await at(4);
    expect(posts).toHaveLength(1);
    await at(2);
    await at(1);
    expect(posts).toHaveLength(2);
    expect(posts[1]).toContain("Recovered");
  });
  test("causes someone else pages are never paged", async () => {
    causes = ["drill", "stuck:r1", "red:r1", "mirror", "snapshots", "budget:r1"];
    await at(0);
    await at(10);
    await at(10);
    expect(posts).toHaveLength(0);
    causes = [];
    await at(10);
    expect(posts).toHaveLength(0);
  });
  test("a restart with disk present pages nothing", async () => {
    causes = ["disk"];
    await notifier.close();
    await boot();
    await at(20);
    expect(posts).toHaveLength(0);
  });
});
		@@ -348,6 +348,12 @@ door: an admin with a notify URL is paged once when a drain passes five
348	348	minutes and once, by the draining process as it exits, when it ends. A process
349	349	that dies mid-drain says nothing more.
350	350
	351	The same push pages the vitals causes `errors`, `internal`, `queue:*` and `disk`
	352	once each has held five minutes, and says Recovered once it has been gone five
	353	(a flap in between is no news). A cause present at boot is taken as already
	354	paged. Every other cause (`drill`, `mirror`, `stuck:`, `red:`, `snapshots`,
	355	`budget:*`) is paged elsewhere or is a setting, and is ignored here.
	356
351	357	## Origins
352	358
353	359	Your forge is the index of everything you work on, wherever it actually lives.
		@@ -42,3 +42,59 @@ export function nextPage(
42	42	return { line: RECOVERED, paged: undefined };
43	43	return undefined;
44	44	}
	45
	46	/** How long a cause must hold to be paged, and stay gone to be recovered. */
	47	export const HOLD_MS = 5 * 60_000;
	48
	49	/** The vitals causes this pager owns; the rest are paged elsewhere or are settings. */
	50	export function owned(cause: string): boolean {
	51	return cause === "errors" \|\| cause === "internal" \|\| cause === "disk" \|\| cause.startsWith("queue:");
	52	}
	53
	54	/** One owned cause being watched. */
	55	export interface Watched {
	56	readonly since: number;
	57	readonly sentence: string;
	58	/** Set once the page was delivered (or was already sent before a restart). */
	59	paged: boolean;
	60	/** When it was last seen gone, until it is seen again. */
	61	absentSince?: number;
	62	}
	63
	64	export type Watch = Map<string, Watched>;
	65
	66	/**
	67	* Looks at the causes present now. Returns the pages due and the Recovereds
	68	* due; nothing is marked, so a send that failed is due again next time.
	69	* `seed` takes every owned cause present as already paged, whatever its age.
	70	*/
	71	export function observe(
	72	watch: Watch,
	73	present: readonly { cause: string; sentence: string }[],
	74	now: number,
	75	seed = false,
	76	): { pages: string[]; recovered: string[] } {
	77	const here = new Set<string>();
	78	for (const p of present) {
	79	if (!owned(p.cause)) continue;
	80	here.add(p.cause);
	81	const w = watch.get(p.cause);
	82	if (w === undefined) watch.set(p.cause, { since: now, sentence: p.sentence, paged: seed });
	83	else delete w.absentSince;
	84	}
	85	const pages: string[] = [];
	86	const recovered: string[] = [];
	87	for (const [key, w] of [...watch]) {
	88	if (here.has(key)) {
	89	if (!w.paged && now - w.since >= HOLD_MS) pages.push(key);
	90	continue;
	91	}
	92	w.absentSince ??= now;
	93	if (now - w.absentSince < HOLD_MS) continue;
	94	if (w.paged) recovered.push(key);
	95	else watch.delete(key);
	96	}
	97	return { pages, recovered };
	98	}
	99
	100	export const recoveredLine = (cause: string): string => `Recovered: ${cause} has cleared`;
		@@ -33,7 +33,8 @@ import { err, ok, type Result } from "../kernel/result.ts";
33	33	import { lastMirror, mirrorNeed } from "../snapshot/mirror.ts";
34	34	import { drillNeed, firstSeen, lastVerdict } from "../snapshot/verdict.ts";
35	35	import { health } from "../http/health.ts";
36		import { RECOVERED, drainIncident, nextPage } from "./incidents.ts";
	36	import { forgeVitals } from "../actions/health.ts";
	37	import { RECOVERED, drainIncident, nextPage, observe, recoveredLine, type Watch } from "./incidents.ts";
37	38	import { askLinesFor } from "../web/studio/asks.ts";
38	39
39	40	const KEY = "notify_url";
		@@ -204,6 +205,8 @@ export interface NotifyOptions {
204	205	clockMs?: number;
205	206	/** When this process began draining; the live health unless a test says. */
206	207	drainingSince?: () => number \| null;
	208	/** The causes the vitals hold now; the live vitals unless a test says. */
	209	causes?: () => readonly { cause: string; sentence: string }[];
207	210	}
208	211
209	212	const oneLine = (s: string): string =>
		@@ -220,6 +223,8 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
220	223	const pause = new Map<string, { until: number; next: number }>();
221	224	/** Per admin: the drain last paged and not yet recovered, set only after delivery. */
222	225	const paged = new Map<string, string \| undefined>();
	226	/** Per admin: the owned vitals causes being watched. */
	227	const watching = new Map<string, Watch>();
223	228	/** Admins this process itself paged and has not yet told Recovered. */
224	229	const mine = new Set<string>();
225	230	let queue: Promise<void> = Promise.resolve();
		@@ -251,8 +256,60 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
251	256	const url = notifyUrlOf(app.db, userId);
252	257	if (user === undefined \|\| url === undefined \|\| !user.isAdmin \|\| user.kind !== "user") {
253	258	paged.delete(userId);
	259	watching.delete(userId);
254	260	return;
255	261	}
	262	await pageDrain(userId, url, baseline);
	263	await pageCauses(userId, url, baseline);
	264	}
	265
	266	/** Sends one line under the hourly cap and the backoff; whether it was delivered. */
	267	async function push(userId: string, url: string, line: string): Promise<boolean> {
	268	const t = now();
	269	const recent = (sent.get(userId) ?? []).filter((s) => t - s < HOUR_MS);
	270	sent.set(userId, recent);
	271	if (recent.length >= HOURLY \|\| t < (pause.get(userId)?.until ?? 0)) return false;
	272	recent.push(t);
	273	let delivered = false;
	274	try {
	275	delivered = await send(url, `${oneLine(line)}\n${link}`, {
	276	click: link,
	277	title: "forge",
	278	});
	279	} catch {
	280	delivered = false;
	281	}
	282	if (delivered) {
	283	pause.delete(userId);
	284	return true;
	285	}
	286	const wait = Math.min(pause.get(userId)?.next ?? BACKOFF_FIRST_MS, BACKOFF_MAX_MS);
	287	pause.set(userId, { until: now() + wait, next: wait * 2 });
	288	app.log.warn("notify endpoint failed; pausing", {
	289	host: notifyHost(url) ?? "",
	290	pauseMs: wait,
	291	});
	292	return false;
	293	}
	294
	295	/** Errors, internal faults, a long queue or a full disk: paged once held, recovered once gone. */
	296	async function pageCauses(userId: string, url: string, baseline: boolean): Promise<void> {
	297	const present = options.causes ? options.causes() : forgeVitals(app).degraded;
	298	let watch = watching.get(userId);
	299	const seed = baseline \|\| watch === undefined;
	300	if (watch === undefined) watching.set(userId, (watch = new Map()));
	301	const due = observe(watch, present, now(), seed);
	302	for (const key of due.pages) {
	303	if (!(await push(userId, url, watch.get(key)!.sentence))) return;
	304	watch.get(key)!.paged = true;
	305	}
	306	for (const key of due.recovered) {
	307	if (!(await push(userId, url, recoveredLine(key)))) return;
	308	watch.delete(key);
	309	}
	310	}
	311
	312	async function pageDrain(userId: string, url: string, baseline: boolean): Promise<void> {
256	313	const current = drainIncident(drainStart(), now());
257	314	if (baseline \|\| !paged.has(userId)) {
258	315	// A restart has paged what it paged: start from what is true now.
		@@ -261,34 +318,11 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
261	318	}
262	319	const next = nextPage(current, paged.get(userId));
263	320	if (next === undefined) return;
264		const t = now();
265		const recent = (sent.get(userId) ?? []).filter((s) => t - s < HOUR_MS);
266		sent.set(userId, recent);
267		if (recent.length >= HOURLY \|\| t < (pause.get(userId)?.until ?? 0)) return;
268		recent.push(t);
269		let delivered = false;
270		try {
271		delivered = await send(url, `${oneLine(next.line)}\n${link}`, {
272		click: link,
273		title: "forge",
274		});
275		} catch {
276		delivered = false;
277		}
278		if (delivered) {
279		paged.set(userId, next.paged);
280		// Only a page this process delivered is this process's to recover.
281		if (next.paged === undefined) mine.delete(userId);
282		else mine.add(userId);
283		pause.delete(userId);
284		return;
285		}
286		const wait = Math.min(pause.get(userId)?.next ?? BACKOFF_FIRST_MS, BACKOFF_MAX_MS);
287		pause.set(userId, { until: now() + wait, next: wait * 2 });
288		app.log.warn("notify endpoint failed; pausing", {
289		host: notifyHost(url) ?? "",
290		pauseMs: wait,
291		});
	321	if (!(await push(userId, url, next.line))) return;
	322	paged.set(userId, next.paged);
	323	// Only a page this process delivered is this process's to recover.
	324	if (next.paged === undefined) mine.delete(userId);
	325	else mine.add(userId);
292	326	}
293	327
294	328	async function tell(userId: string, baseline: boolean): Promise<void> {
		@@ -0,0 +1,116 @@
	1	/**
	2	* The vitals causes the pager owns (`errors`, `internal`, `queue:*`, `disk`)
	3	* page once they have held five minutes and say Recovered once they have
	4	* been gone five; every other cause is left to whoever already pages it.
	5	*/
	6	import { afterEach, beforeEach, describe, expect, test } from "bun:test";
	7	import { mkdtemp, rm } from "node:fs/promises";
	8	import { tmpdir } from "node:os";
	9	import { join } from "node:path";
	10	import { createApp, type App } from "../src/app.ts";
	11	import { createUser } from "../src/core/users.ts";
	12	import { owned } from "../src/crew/incidents.ts";
	13	import { setNotifyUrl, startNotifier, type Notifier } from "../src/crew/notify.ts";
	14
	15	const MIN = 60_000;
	16
	17	describe("owned causes", () => {
	18	test("an explicit list", () => {
	19	for (const c of ["errors", "internal", "disk", "queue:check", "queue:run"])
	20	expect(owned(c)).toBe(true);
	21	for (const c of ["drill", "mirror", "stuck:r1", "red:r1", "snapshots", "budget:r1", "shared:r:t"])
	22	expect(owned(c)).toBe(false);
	23	});
	24	});
	25
	26	describe("the cause pager", () => {
	27	let dataDir: string;
	28	let app: App;
	29	let notifier: Notifier;
	30	let clock = 1_000_000;
	31	let causes: string[] = [];
	32	const posts: string[] = [];
	33
	34	async function boot(): Promise<void> {
	35	notifier = startNotifier(app, {
	36	forgeUrl: "https://forge.example.test",
	37	now: () => clock,
	38	drainingSince: () => null,
	39	causes: () => causes.map((cause) => ({ cause, sentence: `${cause} is bad` })),
	40	send: async (_u, body) => {
	41	posts.push(body);
	42	return true;
	43	},
	44	});
	45	await notifier.settled();
	46	}
	47	const at = async (min: number): Promise<void> => {
	48	clock += min * MIN;
	49	await notifier.tick();
	50	};
	51
	52	beforeEach(async () => {
	53	dataDir = await mkdtemp(join(tmpdir(), "forge-causes-"));
	54	app = createApp({ dataDir, port: 0 });
	55	const made = await createUser(app.db, {
	56	trusted: true,
	57	username: "alice",
	58	email: "[email protected]",
	59	password: "a-good-passphrase",
	60	adminIfFirst: true,
	61	});
	62	if (!made.ok) throw new Error(made.error);
	63	setNotifyUrl(app.db, made.value.id, "https://ntfy.example.test/root-topic");
	64	posts.length = 0;
	65	clock = 1_000_000;
	66	causes = [];
	67	await boot();
	68	});
	69	afterEach(async () => {
	70	await notifier.close();
	71	await rm(dataDir, { recursive: true, force: true });
	72	});
	73
	74	test("a flap pages once; six minutes gone says Recovered once", async () => {
	75	causes = ["errors"];
	76	await at(0);
	77	await at(4);
	78	expect(posts).toHaveLength(0);
	79	await at(2);
	80	expect(posts).toHaveLength(1);
	81	expect(posts[0]).toContain("errors is bad");
	82	causes = [];
	83	await at(1);
	84	causes = ["errors"];
	85	await at(1);
	86	await at(6);
	87	expect(posts).toHaveLength(1);
	88	causes = [];
	89	await at(0);
	90	await at(4);
	91	expect(posts).toHaveLength(1);
	92	await at(2);
	93	await at(1);
	94	expect(posts).toHaveLength(2);
	95	expect(posts[1]).toContain("Recovered");
	96	});
	97
	98	test("causes someone else pages are never paged", async () => {
	99	causes = ["drill", "stuck:r1", "red:r1", "mirror", "snapshots", "budget:r1"];
	100	await at(0);
	101	await at(10);
	102	await at(10);
	103	expect(posts).toHaveLength(0);
	104	causes = [];
	105	await at(10);
	106	expect(posts).toHaveLength(0);
	107	});
	108
	109	test("a restart with disk present pages nothing", async () => {
	110	causes = ["disk"];
	111	await notifier.close();
	112	await boot();
	113	await at(20);
	114	expect(posts).toHaveLength(0);
	115	});
	116	});