4a95b08 · 47c423e · f52c3aa · ebc9b8b
@@ -328,8 +328,10 @@
entry, and the terminal dashboard prints each with its start. When the guard
view holds something that needs a person (a held or stuck rollout, an OOM kill,
or a platform view gone stale), the board's strip and the terminal dashboard
both lead with one line, `needs you: forge <what>`, and do not repeat it as a
rollout line. A drain that runs too long is not covered on either: the forge
keeps no start time for it.
rollout line. A drain that runs too long is not on either; it is a push, not a
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.
## Origins
@@ -0,0 +1,44 @@
/**
 * What is wrong with the forge itself that is worth a page. Pure: the
 * drain's start and the clock in, plain data out. Only a drain that has
 * outlived a release is paged; the vitals and `forge-guard`'s `held`,
 * `stuck` and `oom:*` already have their own place on the board.
 */
import { DRAIN_LIMIT_MS } from "../kernel/guard_verdict.ts";
export interface Incident {
  /** Stable for one drain, so a repeat tick is not news. */
  readonly key: string;
  readonly line: string;
}
/** The drain as one incident once it has lasted more than five minutes. */
export function drainIncident(
  drainingSince: number | null | undefined,
  now: number,
): Incident | undefined {
  if (typeof drainingSince !== "number" || now - drainingSince <= DRAIN_LIMIT_MS)
    return undefined;
  return {
    key: `drain:${drainingSince}`,
    line: `forge has been draining for ${Math.floor((now - drainingSince) / 60_000)} minutes`,
  };
}
export const RECOVERED = "Recovered: the forge is no longer draining";
      
        sentinelcrew
        28d ago
        
      
      In production this line can't be reached (see the comment on notify.ts). Either carry the paged key across the restart, for example in the db after delivery so the next boot sends Recovered once, or drop Recovered and say so in the pull.
    
/**
 * What to say, given the incident now and the key last paged: a page for an
 * incident not yet paged, "Recovered" for a paged one that ended, else nothing.
 */
export function nextPage(
  current: Incident | undefined,
  paged: string | undefined,
): { line: string; paged: string | undefined } | undefined {
  if (current !== undefined && current.key !== paged)
    return { line: current.line, paged: current.key };
  if (current === undefined && paged !== undefined)
    return { line: RECOVERED, paged: undefined };
  return undefined;
}
@@ -32,6 +32,8 @@
import { landedLines } from "../core/watches.ts";
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 { askLinesFor } from "../web/studio/asks.ts";
const KEY = "notify_url";
@@ -39,6 +41,7 @@
export const MAX_URL = 200;
const HOURLY = 10;
const HOUR_MS = 3_600_000;
const CLOCK_MS = 10 * 60_000;
const PAGE_MS = 60_000;
const SEND_TIMEOUT_MS = 5_000;
const BACKOFF_FIRST_MS = 60_000;
const BACKOFF_MAX_MS = HOUR_MS;
@@ -184,7 +187,12 @@
export function mirrorLines(
export interface Notifier {
  /** Resolves when the work queued so far is done. For tests and shutdown. */
  settled(): Promise<void>;
  /** Looks at the forge's own health now, as the minute's clock does. */
  tick(): Promise<void>;
  /** Stops listening and the asks' clock; the drain pager keeps watching. */
  stop(): void;
  /** Stops everything, the pager too: the last step before the process goes. */
  close(): Promise<void>;
}
export interface NotifyOptions {
@@ -194,6 +202,8 @@
export interface NotifyOptions {
  now?: () => number;
  /** How often the clock wakes the notifier; ten minutes unless a test says. */
  clockMs?: number;
  /** When this process began draining; the live health unless a test says. */
  drainingSince?: () => number | null;
}
const oneLine = (s: string): string =>
@@ -208,6 +218,10 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
  const seen = new Map<string, Set<string>>();
  const sent = new Map<string, number[]>();
  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>();
  /** Admins this process itself paged and has not yet told Recovered. */
  const mine = new Set<string>();
  let queue: Promise<void> = Promise.resolve();
  let stopped = false;
@@ -220,6 +234,63 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
      .all(KEY)
      .map((r) => r.user_id);
  const drainStart = (): number | null =>
    options.drainingSince
      ? options.drainingSince()
      : health.draining
        ? health.drainingSince
        : null;
  /**
   * The admin is paged once for a drain past five minutes and once when it
   * ends. The key is recorded after delivery only, so a page the cap or the
   * backoff held back is tried again next tick and its Recovered is kept.
   */
  async function page(userId: string, baseline: boolean): Promise<void> {
    const user = getUserById(app.db, userId);
    const url = notifyUrlOf(app.db, userId);
    if (user === undefined || url === undefined || !user.isAdmin || user.kind !== "user") {
      paged.delete(userId);
      return;
    }
    const current = drainIncident(drainStart(), now());
      
        sentinelcrew
        28d ago
        outdated
      
      The stored key is read only here, at boot or on first sight. A drain that a still-draining previous process pages after this process has booted (the swap order in the README) is never seen, so its Recovered is lost until some later restart sends a stale one. Send Recovered from the draining process's close() instead.
    
    if (baseline || !paged.has(userId)) {
      // A restart has paged what it paged: start from what is true now.
      paged.set(userId, current?.key);
      return;
    }
    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,
    });
  }
  async function tell(userId: string, baseline: boolean): Promise<void> {
    const user = getUserById(app.db, userId);
    const url = notifyUrlOf(app.db, userId);
@@ -300,7 +371,10 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
  // A forge that was down has told what it told: start from what asks now.
  queue = queue.then(async () => {
    for (const id of people()) await tell(id, true).catch(() => undefined);
    for (const id of people()) {
      await tell(id, true).catch(() => undefined);
      await page(id, true).catch(() => undefined);
    }
  });
  // Eight days of silence is no event: the clock wakes the notifier too.
@@ -315,12 +389,45 @@
export function startNotifier(app: App, options: NotifyOptions): Notifier {
  }, options.clockMs ?? CLOCK_MS);
  clock.unref();
  // A drain is paged within a minute of its fifth, not within ten.
  const pageAll = (): Promise<void> => {
    queue = queue.then(async () => {
      try {
      
        sentinelcrew
        28d ago
        outdated
      
      This interval is cleared by notifier.stop(), which drain() in src/main.ts calls in the same tick as startDraining. While a drain is running, nothing is left to look at drainingSince, so the page never fires. Keep the pager alive until close().
    
        for (const id of people()) await page(id, false);
      } catch (error) {
        app.log.error("notify failed", { error: String(error) });
      }
    });
    return queue;
  };
  const pager = setInterval(() => void pageAll(), Math.min(options.clockMs ?? CLOCK_MS, PAGE_MS));
  pager.unref();
  return {
    settled: () => queue,
    tick: pageAll,
    stop() {
      stopped = true;
      clearInterval(clock);
      unsubscribe();
    },
    async close() {
      this.stop();
      clearInterval(pager);
      
        sentinelcrew
        28d ago
        outdated
      
      This is where the drain truly ends. If paged holds a key, send the Recovered here (send is already capped at SEND_TIMEOUT_MS) before clearing the pager, and delete notify_drain_paged on delivery.
    
      await queue;
      // The process that paged is the one whose exit ends the drain: it says
      // Recovered itself, once, however the release was ordered. A process
      // that dies without closing says nothing.
      for (const userId of [...mine]) {
        const url = notifyUrlOf(app.db, userId);
      
        sentinelcrew
        28d ago
        
      
      Low: at exit, Recovered goes to everyone in mine without checking again whether they are still an admin, unlike page(). An admin demoted mid-drain would still get one Recovered at their own URL. That is harmless, but you could call getUserById and check isAdmin here the same way page() does.
    
        if (url === undefined) continue;
        try {
          if (await send(url, `${RECOVERED}\n${link}`, { click: link, title: "forge" }))
            mine.delete(userId);
        } catch {
          // a dead endpoint costs nothing more at exit
        }
      }
    },
  };
}
@@ -16,7 +16,10 @@
import type { Log } from "../kernel/log.ts";
export const health = { draining: false };
export const health: { draining: boolean; drainingSince: number | null } = {
  draining: false,
  drainingSince: null,
};
/** Whether the last liveness answer was a 503, so the log says so once. */
let databaseDown = false;
@@ -28,6 +31,7 @@
let databaseDown = false;
export function startDraining(log: Log, why: string, fields: Record<string, number> = {}): void {
  if (health.draining) return;
  health.draining = true;
  health.drainingSince = Date.now();
      
        sentinelcrew
        28d ago
        
      
      Setting this is right. But it's the only thing that changes when a drain starts, and the same call site stops the notifier immediately afterwards (see main.ts drain()).
    
  log.info("forge draining: /readyz answers 503 from now", { why, ...fields });
}
@@ -579,8 +579,10 @@
function drain(signal: NodeJS.Signals): Promise<unknown> {
}
/** The last step of a drain: cut what is left and exit. */
function close(): void {
async function close(): Promise<void> {
  app.log.info("forge shutting down");
  // Says Recovered if this process paged a drain: its exit is the end of it.
  await notifier.close().catch(() => {});
  ssh?.stop();
  servers.stop();
  clearInterval(flagTimer);
@@ -0,0 +1,195 @@
/**
 * A drain past five minutes pages the admin once, its end says Recovered
 * once; a page the cap or the backoff held back is sent later; what the
 * guard already shows (`stuck`, `drill`) is not paged here.
 */
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 { drainIncident, nextPage } from "../src/crew/incidents.ts";
import { health, startDraining } from "../src/http/health.ts";
import { setNotifyUrl, startNotifier, type Notifier } from "../src/crew/notify.ts";
const MIN = 60_000;
describe("incidents (pure)", () => {
  test("only a drain past five minutes is an incident, keyed by its start", () => {
    expect(drainIncident(null, 10 * MIN)).toBeUndefined();
    expect(drainIncident(0, 5 * MIN)).toBeUndefined();
    expect(drainIncident(0, 6 * MIN)?.key).toBe("drain:0");
  });
  test("a page, then nothing, then Recovered", () => {
    const inc = drainIncident(0, 6 * MIN);
    const page = nextPage(inc, undefined)!;
    expect(page.paged).toBe("drain:0");
    expect(nextPage(inc, page.paged)).toBeUndefined();
    expect(nextPage(undefined, page.paged)?.line).toContain("Recovered");
    expect(nextPage(undefined, undefined)).toBeUndefined();
  });
});
describe("the drain page", () => {
  let dataDir: string;
  let app: App;
  let notifier: Notifier;
  let clock = 1_000_000;
  let since: number | null = null;
  let accept = true;
  const posts: string[] = [];
  async function boot(): Promise<void> {
    notifier = startNotifier(app, {
      forgeUrl: "https://forge.example.test",
      now: () => clock,
      drainingSince: () => since,
      send: async (_u, body) => {
        posts.push(body);
        return accept;
      },
    });
    await notifier.settled();
  }
  beforeEach(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-drain-"));
    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;
    since = null;
    accept = true;
    await boot();
  });
  afterEach(async () => {
    await notifier.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  test("six minutes of draining sends one page, its end one Recovered", async () => {
    since = clock;
    clock += 4 * MIN;
    await notifier.tick();
    expect(posts).toHaveLength(0);
    clock += 2 * MIN;
    await notifier.tick();
    await notifier.tick();
    expect(posts).toHaveLength(1);
    expect(posts[0]).toContain("draining for 6 minutes");
    since = null;
    await notifier.tick();
    await notifier.tick();
    expect(posts).toHaveLength(2);
    expect(posts[1]).toContain("Recovered");
  });
  test("a page the endpoint refused is sent on a later tick, and Recovered is not lost", async () => {
      
        sentinelcrew
        28d ago
        outdated
      
      This covers the backoff. The issue asks for a page held back by the hourly cap and sent on the next tick, and no test fills HOURLY. Since sent is shared with tell(), add a test where 10 sends in the hour hold the drain page back and it goes out once the window rolls over.
    
    since = clock;
    clock += 6 * MIN;
    accept = false;
    await notifier.tick();
    expect(posts).toHaveLength(1);
    // Backed off: nothing is tried at once.
    await notifier.tick();
    expect(posts).toHaveLength(1);
    accept = true;
    clock += 2 * MIN;
    await notifier.tick();
    expect(posts).toHaveLength(2);
    since = null;
    await notifier.tick();
    expect(posts).toHaveLength(3);
    expect(posts[2]).toContain("Recovered");
  });
  test("a restart in a paged drain does not page again", async () => {
      
        sentinelcrew
        28d ago
        
      
      Low: this test never pages before the restart. close() runs before any tick, so mine is empty. It does check that a boot mid-drain starts silent, but the name says "a paged drain". Rename it, or add a tick() before close() and expect exactly one page and one Recovered from the close.
    
    since = clock;
    clock += 6 * MIN;
    await notifier.close();
    await boot();
    await notifier.tick();
    expect(posts).toHaveLength(0);
  });
  test("a drain started the way main.ts starts it, with the notifier stopped beside it, is still paged", async () => {
    const real = Date.now();
    clock = real;
    await notifier.close();
    notifier = startNotifier(app, {
      forgeUrl: "https://forge.example.test",
      now: () => clock,
      send: async (_u, body) => {
        posts.push(body);
        return true;
      },
    });
    await notifier.settled();
    startDraining(app.log, "SIGTERM");
    notifier.stop();
    try {
      clock = real + 6 * MIN;
      await notifier.tick();
      await notifier.tick();
      expect(posts).toHaveLength(1);
      expect(posts[0]).toContain("draining");
    } finally {
      health.draining = false;
      health.drainingSince = null;
    }
  });
  test("a swap: the old process pages after the new one booted, and its exit says Recovered once", async () => {
    // The new process booted first and sees no drain of its own.
    const next = startNotifier(app, {
      forgeUrl: "https://forge.example.test",
      now: () => clock,
      drainingSince: () => null,
      send: async (_u, body) => {
        posts.push(body);
        return true;
      },
    });
    await next.settled();
    since = clock;
    clock += 6 * MIN;
    await notifier.tick();
    await next.tick();
    expect(posts).toHaveLength(1);
    await notifier.close();
    expect(posts).toHaveLength(2);
    expect(posts[1]).toContain("Recovered");
    await next.tick();
    await next.close();
    expect(posts).toHaveLength(2);
  });
  test("a page held back by the hourly cap is sent once the window rolls over", async () => {
    // Five pages and five Recovered fill the ten an hour.
    for (let i = 0; i < 5; i++) {
      since = clock;
      clock += 6 * MIN;
      await notifier.tick();
      since = null;
      await notifier.tick();
    }
    expect(posts).toHaveLength(10);
    since = clock;
    clock += 6 * MIN;
    await notifier.tick();
    expect(posts).toHaveLength(10);
      
        sentinelcrew
        28d ago
        outdated
      
      Nothing in the change reads the guard, so this test cannot fail and guards nothing. Drop it, or make it assert something the code actually decides.
    
    clock += 61 * MIN;
    await notifier.tick();
    expect(posts).toHaveLength(11);
    expect(posts[10]).toContain("draining");
  });
});
@@ -109,7 +109,7 @@
describe("notify", () => {
    await notifier.settled();
  });
  afterAll(async () => {
    notifier.stop();
    await notifier.close();
    await rm(dataDir, { recursive: true, force: true });
  });
		@@ -328,8 +328,10 @@ entry, and the terminal dashboard prints each with its start. When the guard
328	328	view holds something that needs a person (a held or stuck rollout, an OOM kill,
329	329	or a platform view gone stale), the board's strip and the terminal dashboard
330	330	both lead with one line, `needs you: forge <what>`, and do not repeat it as a
331		rollout line. A drain that runs too long is not covered on either: the forge
332		keeps no start time for it.
	331	rollout line. A drain that runs too long is not on either; it is a push, not a
	332	door: an admin with a notify URL is paged once when a drain passes five
	333	minutes and once, by the draining process as it exits, when it ends. A process
	334	that dies mid-drain says nothing more.
333	335
334	336	## Origins
335	337
		@@ -0,0 +1,44 @@
	1	/**
	2	* What is wrong with the forge itself that is worth a page. Pure: the
	3	* drain's start and the clock in, plain data out. Only a drain that has
	4	* outlived a release is paged; the vitals and `forge-guard`'s `held`,
	5	* `stuck` and `oom:*` already have their own place on the board.
	6	*/
	7
	8	import { DRAIN_LIMIT_MS } from "../kernel/guard_verdict.ts";
	9
	10	export interface Incident {
	11	/** Stable for one drain, so a repeat tick is not news. */
	12	readonly key: string;
	13	readonly line: string;
	14	}
	15
	16	/** The drain as one incident once it has lasted more than five minutes. */
	17	export function drainIncident(
	18	drainingSince: number \| null \| undefined,
	19	now: number,
	20	): Incident \| undefined {
	21	if (typeof drainingSince !== "number" \|\| now - drainingSince <= DRAIN_LIMIT_MS)
	22	return undefined;
	23	return {
	24	key: `drain:${drainingSince}`,
	25	line: `forge has been draining for ${Math.floor((now - drainingSince) / 60_000)} minutes`,
	26	};
	27	}
	28
	29	export const RECOVERED = "Recovered: the forge is no longer draining";
	30
		sentinelcrew 28d ago In production this line can't be reached (see the comment on `notify.ts`). Either carry the paged key across the restart, for example in the db after delivery so the next boot sends Recovered once, or drop Recovered and say so in the pull.
	31	/**
	32	* What to say, given the incident now and the key last paged: a page for an
	33	* incident not yet paged, "Recovered" for a paged one that ended, else nothing.
	34	*/
	35	export function nextPage(
	36	current: Incident \| undefined,
	37	paged: string \| undefined,
	38	): { line: string; paged: string \| undefined } \| undefined {
	39	if (current !== undefined && current.key !== paged)
	40	return { line: current.line, paged: current.key };
	41	if (current === undefined && paged !== undefined)
	42	return { line: RECOVERED, paged: undefined };
	43	return undefined;
	44	}
		@@ -32,6 +32,8 @@ import { landedLines } from "../core/watches.ts";
32	32	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	import { health } from "../http/health.ts";
	36	import { RECOVERED, drainIncident, nextPage } from "./incidents.ts";
35	37	import { askLinesFor } from "../web/studio/asks.ts";
36	38
37	39	const KEY = "notify_url";
		@@ -39,6 +41,7 @@ export const MAX_URL = 200;
39	41	const HOURLY = 10;
40	42	const HOUR_MS = 3_600_000;
41	43	const CLOCK_MS = 10 * 60_000;
	44	const PAGE_MS = 60_000;
42	45	const SEND_TIMEOUT_MS = 5_000;
43	46	const BACKOFF_FIRST_MS = 60_000;
44	47	const BACKOFF_MAX_MS = HOUR_MS;
		@@ -184,7 +187,12 @@ export function mirrorLines(
184	187	export interface Notifier {
185	188	/** Resolves when the work queued so far is done. For tests and shutdown. */
186	189	settled(): Promise<void>;
	190	/** Looks at the forge's own health now, as the minute's clock does. */
	191	tick(): Promise<void>;
	192	/** Stops listening and the asks' clock; the drain pager keeps watching. */
187	193	stop(): void;
	194	/** Stops everything, the pager too: the last step before the process goes. */
	195	close(): Promise<void>;
188	196	}
189	197
190	198	export interface NotifyOptions {
		@@ -194,6 +202,8 @@ export interface NotifyOptions {
194	202	now?: () => number;
195	203	/** How often the clock wakes the notifier; ten minutes unless a test says. */
196	204	clockMs?: number;
	205	/** When this process began draining; the live health unless a test says. */
	206	drainingSince?: () => number \| null;
197	207	}
198	208
199	209	const oneLine = (s: string): string =>
		@@ -208,6 +218,10 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
208	218	const seen = new Map<string, Set<string>>();
209	219	const sent = new Map<string, number[]>();
210	220	const pause = new Map<string, { until: number; next: number }>();
	221	/** Per admin: the drain last paged and not yet recovered, set only after delivery. */
	222	const paged = new Map<string, string \| undefined>();
	223	/** Admins this process itself paged and has not yet told Recovered. */
	224	const mine = new Set<string>();
211	225	let queue: Promise<void> = Promise.resolve();
212	226	let stopped = false;
213	227
		@@ -220,6 +234,63 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
220	234	.all(KEY)
221	235	.map((r) => r.user_id);
222	236
	237	const drainStart = (): number \| null =>
	238	options.drainingSince
	239	? options.drainingSince()
	240	: health.draining
	241	? health.drainingSince
	242	: null;
	243
	244	/**
	245	* The admin is paged once for a drain past five minutes and once when it
	246	* ends. The key is recorded after delivery only, so a page the cap or the
	247	* backoff held back is tried again next tick and its Recovered is kept.
	248	*/
	249	async function page(userId: string, baseline: boolean): Promise<void> {
	250	const user = getUserById(app.db, userId);
	251	const url = notifyUrlOf(app.db, userId);
	252	if (user === undefined \|\| url === undefined \|\| !user.isAdmin \|\| user.kind !== "user") {
	253	paged.delete(userId);
	254	return;
	255	}
	256	const current = drainIncident(drainStart(), now());
		sentinelcrew 28d ago outdated The stored key is read only here, at boot or on first sight. A drain that a still-draining previous process pages after this process has booted (the swap order in the README) is never seen, so its Recovered is lost until some later restart sends a stale one. Send Recovered from the draining process's `close()` instead.
	257	if (baseline \|\| !paged.has(userId)) {
	258	// A restart has paged what it paged: start from what is true now.
	259	paged.set(userId, current?.key);
	260	return;
	261	}
	262	const next = nextPage(current, paged.get(userId));
	263	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	});
	292	}
	293
223	294	async function tell(userId: string, baseline: boolean): Promise<void> {
224	295	const user = getUserById(app.db, userId);
225	296	const url = notifyUrlOf(app.db, userId);
		@@ -300,7 +371,10 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
300	371
301	372	// A forge that was down has told what it told: start from what asks now.
302	373	queue = queue.then(async () => {
303		for (const id of people()) await tell(id, true).catch(() => undefined);
	374	for (const id of people()) {
	375	await tell(id, true).catch(() => undefined);
	376	await page(id, true).catch(() => undefined);
	377	}
304	378	});
305	379
306	380	// Eight days of silence is no event: the clock wakes the notifier too.
		@@ -315,12 +389,45 @@ export function startNotifier(app: App, options: NotifyOptions): Notifier {
315	389	}, options.clockMs ?? CLOCK_MS);
316	390	clock.unref();
317	391
	392	// A drain is paged within a minute of its fifth, not within ten.
	393	const pageAll = (): Promise<void> => {
	394	queue = queue.then(async () => {
	395	try {
		sentinelcrew 28d ago outdated This interval is cleared by `notifier.stop()`, which `drain()` in `src/main.ts` calls in the same tick as `startDraining`. While a drain is running, nothing is left to look at `drainingSince`, so the page never fires. Keep the pager alive until `close()`.
	396	for (const id of people()) await page(id, false);
	397	} catch (error) {
	398	app.log.error("notify failed", { error: String(error) });
	399	}
	400	});
	401	return queue;
	402	};
	403	const pager = setInterval(() => void pageAll(), Math.min(options.clockMs ?? CLOCK_MS, PAGE_MS));
	404	pager.unref();
	405
318	406	return {
319	407	settled: () => queue,
	408	tick: pageAll,
320	409	stop() {
321	410	stopped = true;
322	411	clearInterval(clock);
323	412	unsubscribe();
324	413	},
	414	async close() {
	415	this.stop();
	416	clearInterval(pager);
		sentinelcrew 28d ago outdated This is where the drain truly ends. If `paged` holds a key, send the Recovered here (`send` is already capped at SEND_TIMEOUT_MS) before clearing the pager, and delete `notify_drain_paged` on delivery.
	417	await queue;
	418	// The process that paged is the one whose exit ends the drain: it says
	419	// Recovered itself, once, however the release was ordered. A process
	420	// that dies without closing says nothing.
	421	for (const userId of [...mine]) {
	422	const url = notifyUrlOf(app.db, userId);
		sentinelcrew 28d ago Low: at exit, Recovered goes to everyone in `mine` without checking again whether they are still an admin, unlike `page()`. An admin demoted mid-drain would still get one Recovered at their own URL. That is harmless, but you could call `getUserById` and check `isAdmin` here the same way `page()` does.
	423	if (url === undefined) continue;
	424	try {
	425	if (await send(url, `${RECOVERED}\n${link}`, { click: link, title: "forge" }))
	426	mine.delete(userId);
	427	} catch {
	428	// a dead endpoint costs nothing more at exit
	429	}
	430	}
	431	},
325	432	};
326	433	}
		@@ -16,7 +16,10 @@
16	16
17	17	import type { Log } from "../kernel/log.ts";
18	18
19		export const health = { draining: false };
	19	export const health: { draining: boolean; drainingSince: number \| null } = {
	20	draining: false,
	21	drainingSince: null,
	22	};
20	23
21	24	/** Whether the last liveness answer was a 503, so the log says so once. */
22	25	let databaseDown = false;
		@@ -28,6 +31,7 @@ let databaseDown = false;
28	31	export function startDraining(log: Log, why: string, fields: Record<string, number> = {}): void {
29	32	if (health.draining) return;
30	33	health.draining = true;
	34	health.drainingSince = Date.now();
		sentinelcrew 28d ago Setting this is right. But it's the only thing that changes when a drain starts, and the same call site stops the notifier immediately afterwards (see `main.ts` `drain()`).
31	35	log.info("forge draining: /readyz answers 503 from now", { why, ...fields });
32	36	}
33	37
		@@ -579,8 +579,10 @@ function drain(signal: NodeJS.Signals): Promise<unknown> {
579	579	}
580	580
581	581	/** The last step of a drain: cut what is left and exit. */
582		function close(): void {
	582	async function close(): Promise<void> {
583	583	app.log.info("forge shutting down");
	584	// Says Recovered if this process paged a drain: its exit is the end of it.
	585	await notifier.close().catch(() => {});
584	586	ssh?.stop();
585	587	servers.stop();
586	588	clearInterval(flagTimer);
		@@ -0,0 +1,195 @@
	1	/**
	2	* A drain past five minutes pages the admin once, its end says Recovered
	3	* once; a page the cap or the backoff held back is sent later; what the
	4	* guard already shows (`stuck`, `drill`) is not paged here.
	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 { drainIncident, nextPage } from "../src/crew/incidents.ts";
	13	import { health, startDraining } from "../src/http/health.ts";
	14	import { setNotifyUrl, startNotifier, type Notifier } from "../src/crew/notify.ts";
	15
	16	const MIN = 60_000;
	17
	18	describe("incidents (pure)", () => {
	19	test("only a drain past five minutes is an incident, keyed by its start", () => {
	20	expect(drainIncident(null, 10 * MIN)).toBeUndefined();
	21	expect(drainIncident(0, 5 * MIN)).toBeUndefined();
	22	expect(drainIncident(0, 6 * MIN)?.key).toBe("drain:0");
	23	});
	24	test("a page, then nothing, then Recovered", () => {
	25	const inc = drainIncident(0, 6 * MIN);
	26	const page = nextPage(inc, undefined)!;
	27	expect(page.paged).toBe("drain:0");
	28	expect(nextPage(inc, page.paged)).toBeUndefined();
	29	expect(nextPage(undefined, page.paged)?.line).toContain("Recovered");
	30	expect(nextPage(undefined, undefined)).toBeUndefined();
	31	});
	32	});
	33
	34	describe("the drain page", () => {
	35	let dataDir: string;
	36	let app: App;
	37	let notifier: Notifier;
	38	let clock = 1_000_000;
	39	let since: number \| null = null;
	40	let accept = true;
	41	const posts: string[] = [];
	42
	43	async function boot(): Promise<void> {
	44	notifier = startNotifier(app, {
	45	forgeUrl: "https://forge.example.test",
	46	now: () => clock,
	47	drainingSince: () => since,
	48	send: async (_u, body) => {
	49	posts.push(body);
	50	return accept;
	51	},
	52	});
	53	await notifier.settled();
	54	}
	55
	56	beforeEach(async () => {
	57	dataDir = await mkdtemp(join(tmpdir(), "forge-drain-"));
	58	app = createApp({ dataDir, port: 0 });
	59	const made = await createUser(app.db, {
	60	trusted: true,
	61	username: "alice",
	62	email: "[email protected]",
	63	password: "a-good-passphrase",
	64	adminIfFirst: true,
	65	});
	66	if (!made.ok) throw new Error(made.error);
	67	setNotifyUrl(app.db, made.value.id, "https://ntfy.example.test/root-topic");
	68	posts.length = 0;
	69	clock = 1_000_000;
	70	since = null;
	71	accept = true;
	72	await boot();
	73	});
	74	afterEach(async () => {
	75	await notifier.close();
	76	await rm(dataDir, { recursive: true, force: true });
	77	});
	78
	79	test("six minutes of draining sends one page, its end one Recovered", async () => {
	80	since = clock;
	81	clock += 4 * MIN;
	82	await notifier.tick();
	83	expect(posts).toHaveLength(0);
	84	clock += 2 * MIN;
	85	await notifier.tick();
	86	await notifier.tick();
	87	expect(posts).toHaveLength(1);
	88	expect(posts[0]).toContain("draining for 6 minutes");
	89	since = null;
	90	await notifier.tick();
	91	await notifier.tick();
	92	expect(posts).toHaveLength(2);
	93	expect(posts[1]).toContain("Recovered");
	94	});
	95
	96	test("a page the endpoint refused is sent on a later tick, and Recovered is not lost", async () => {
		sentinelcrew 28d ago outdated This covers the backoff. The issue asks for a page held back by the hourly cap and sent on the next tick, and no test fills `HOURLY`. Since `sent` is shared with `tell()`, add a test where 10 sends in the hour hold the drain page back and it goes out once the window rolls over.
	97	since = clock;
	98	clock += 6 * MIN;
	99	accept = false;
	100	await notifier.tick();
	101	expect(posts).toHaveLength(1);
	102	// Backed off: nothing is tried at once.
	103	await notifier.tick();
	104	expect(posts).toHaveLength(1);
	105	accept = true;
	106	clock += 2 * MIN;
	107	await notifier.tick();
	108	expect(posts).toHaveLength(2);
	109	since = null;
	110	await notifier.tick();
	111	expect(posts).toHaveLength(3);
	112	expect(posts[2]).toContain("Recovered");
	113	});
	114
	115	test("a restart in a paged drain does not page again", async () => {
		sentinelcrew 28d ago Low: this test never pages before the restart. `close()` runs before any tick, so `mine` is empty. It does check that a boot mid-drain starts silent, but the name says "a paged drain". Rename it, or add a `tick()` before `close()` and expect exactly one page and one Recovered from the close.
	116	since = clock;
	117	clock += 6 * MIN;
	118	await notifier.close();
	119	await boot();
	120	await notifier.tick();
	121	expect(posts).toHaveLength(0);
	122	});
	123
	124	test("a drain started the way main.ts starts it, with the notifier stopped beside it, is still paged", async () => {
	125	const real = Date.now();
	126	clock = real;
	127	await notifier.close();
	128	notifier = startNotifier(app, {
	129	forgeUrl: "https://forge.example.test",
	130	now: () => clock,
	131	send: async (_u, body) => {
	132	posts.push(body);
	133	return true;
	134	},
	135	});
	136	await notifier.settled();
	137	startDraining(app.log, "SIGTERM");
	138	notifier.stop();
	139	try {
	140	clock = real + 6 * MIN;
	141	await notifier.tick();
	142	await notifier.tick();
	143	expect(posts).toHaveLength(1);
	144	expect(posts[0]).toContain("draining");
	145	} finally {
	146	health.draining = false;
	147	health.drainingSince = null;
	148	}
	149	});
	150
	151	test("a swap: the old process pages after the new one booted, and its exit says Recovered once", async () => {
	152	// The new process booted first and sees no drain of its own.
	153	const next = startNotifier(app, {
	154	forgeUrl: "https://forge.example.test",
	155	now: () => clock,
	156	drainingSince: () => null,
	157	send: async (_u, body) => {
	158	posts.push(body);
	159	return true;
	160	},
	161	});
	162	await next.settled();
	163	since = clock;
	164	clock += 6 * MIN;
	165	await notifier.tick();
	166	await next.tick();
	167	expect(posts).toHaveLength(1);
	168	await notifier.close();
	169	expect(posts).toHaveLength(2);
	170	expect(posts[1]).toContain("Recovered");
	171	await next.tick();
	172	await next.close();
	173	expect(posts).toHaveLength(2);
	174	});
	175
	176	test("a page held back by the hourly cap is sent once the window rolls over", async () => {
	177	// Five pages and five Recovered fill the ten an hour.
	178	for (let i = 0; i < 5; i++) {
	179	since = clock;
	180	clock += 6 * MIN;
	181	await notifier.tick();
	182	since = null;
	183	await notifier.tick();
	184	}
	185	expect(posts).toHaveLength(10);
	186	since = clock;
	187	clock += 6 * MIN;
	188	await notifier.tick();
	189	expect(posts).toHaveLength(10);
		sentinelcrew 28d ago outdated Nothing in the change reads the guard, so this test cannot fail and guards nothing. Drop it, or make it assert something the code actually decides.
	190	clock += 61 * MIN;
	191	await notifier.tick();
	192	expect(posts).toHaveLength(11);
	193	expect(posts[10]).toContain("draining");
	194	});
	195	});
		@@ -109,7 +109,7 @@ describe("notify", () => {
109	109	await notifier.settled();
110	110	});
111	111	afterAll(async () => {
112		notifier.stop();
	112	await notifier.close();
113	113	await rm(dataDir, { recursive: true, force: true });
114	114	});
115	115