trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns14g rEventsFlows

watchDrill(fetch, sleep) polls a forge until it is healthy again and returns each distinct line it said while unhealthy #2563

Merged scoutcrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/2424-watchdrill-fetch-sleep-polls-a-forge-unt → main 3 commits fork 3 threads · 3 unresolved

  • Merged. scout/2424-watchdrill-fetch-sleep-polls-a-forge-unt was merged into main as 61f8f1b.

Review

Sign in to review this change.
4 files changed +285 −69
scripts/drills/run.ts +31 −69
@@ -9,6 +9,7 @@
99 */
1010
1111 import { DOMAINS } from "../../src/kernel/domains.ts";
12 import { UNREACHABLE, clean, linesOf, watchPolls, type Poll } from "./watch.ts";
1213
1314 export interface DrillRun {
1415 domain: string;
@@ -27,49 +28,16 @@ export interface DrillRun {
2728 const SEEN_MAX = 8_000;
2829 const DEFAULT_CEILING_MS = 3_600_000;
2930
30 /** C0 and C1 control characters out (tab stays): the forge's words never drive the terminal. */
31 export const clean = (s: string): string => s.replace(/[\x00-\x08\x0a-\x1f\x7f-\x9f]/g, "");
31 export { clean };
3232
33 /** Every string in a JSON value, or the text's own lines when it is not JSON. */
34 function linesOf(text: string, contentType: string): string[] {
35 return rawLinesOf(text, contentType).map(clean).filter((l) => l.trim() !== "");
36 }
37
38 function rawLinesOf(text: string, contentType: string): string[] {
39 if (contentType.includes("json")) {
40 try {
41 const out: string[] = [];
42 const walk = (v: unknown): void => {
43 if (typeof v === "string") out.push(v);
44 else if (Array.isArray(v)) v.forEach(walk);
45 else if (v && typeof v === "object") Object.values(v).forEach(walk);
46 };
47 walk(JSON.parse(text));
48 return out;
49 } catch {
50 /* fall through to plain lines */
51 }
52 }
53 return text
54 .replace(/<(script|style)[\s\S]*?<\/\1>/gi, " ")
55 .replace(/<[^>]*>/g, "\n")
56 .replace(/&amp;/g, "&").replace(/&lt;/g, "<").replace(/&gt;/g, ">")
57 .replace(/&quot;/g, '"').replace(/&#0?39;/g, "'")
58 .split("\n")
59 .map((l) => l.trim())
60 .filter((l) => l !== "");
61 }
62
63 interface Look {
64 healthy: boolean;
65 health: string[];
66 board: string[];
67 }
68
69 async function look(forge: string, token: string): Promise<Look> {
33 /**
34 * One look at the forge: healthy only when `/api/v1/health` reports ok and `/`
35 * answers; what each said is kept under `health` and `board`.
36 */
37 export function forgePoll(forge: string, token: string, fetchIt: typeof fetch = fetch): () => Promise<Poll> {
7038 const get = async (path: string): Promise<{ ok: boolean; lines: string[]; json: unknown }> => {
7139 try {
72 const res = await fetch(`${forge}${path}`, {
40 const res = await fetchIt(`${forge}${path}`, {
7341 headers: { authorization: `Bearer ${token}` },
7442 redirect: "manual",
7543 signal: AbortSignal.timeout(5_000),
@@ -82,17 +50,16 @@ async function look(forge: string, token: string): Promise<Look> {
8250 }
8351 return { ok: res.status < 400, lines: linesOf(text, type), json };
8452 } catch {
85 return { ok: false, lines: [], json: null };
53 return { ok: false, lines: [UNREACHABLE], json: null };
8654 }
8755 };
88 const [h, b] = await Promise.all([get("/api/v1/health"), get("/")]);
89 const reported = (h.json as { ok?: unknown } | null)?.ok === true;
90 return { healthy: h.ok && b.ok && reported, health: h.lines, board: b.lines };
56 return async () => {
57 const [h, b] = await Promise.all([get("/api/v1/health"), get("/")]);
58 const reported = (h.json as { ok?: unknown } | null)?.ok === true;
59 return { healthy: h.ok && b.ok && reported, said: { health: h.lines, board: b.lines } };
60 };
9161 }
9262
93 /** A line with its numbers blanked, so a clock or a counter is not a new line. */
94 const shape = (l: string): string => l.replace(/\d+/g, "#");
95
9663 const cap = (lines: Iterable<string>): string => [...lines].join("\n").slice(0, SEEN_MAX);
9764
9865 /** Runs the drill; resolves to the process exit code. */
@@ -116,29 +83,24 @@ export async function runDrill(run: DrillRun): Promise<number> {
11683 run.say(`The forge should be back within ${domain.backMinutes} minutes. Press Enter when you have caused it.`);
11784 await run.enter();
11885
119 const started = now();
120 const health = new Map<string, string>();
121 const board = new Map<string, string>();
122 let sawDown = false;
123 for (;;) {
124 const seen = await look(run.forge, run.token);
125 if (seen.healthy && sawDown) break;
126 if (!seen.healthy) {
127 sawDown = true;
128 for (const l of seen.health) if (!health.has(shape(l))) { health.set(shape(l), l); run.say(`health: ${l}`); }
129 for (const l of seen.board) if (!board.has(shape(l))) { board.set(shape(l), l); run.say(`board: ${l}`); }
130 }
131 if (now() - started >= ceilingMs) {
132 const ceilingS = Math.round(ceilingMs / 1000);
133 run.say(`The forge did not come back within the ceiling of ${ceilingS}s; nothing was posted.`);
134 run.say(JSON.stringify({ domain: domain.key, backSeconds: null, posted: false, ceilingSeconds: ceilingS }));
135 return 0;
136 }
137 await sleep(intervalMs);
86 const watched = await watchPolls({
87 poll: forgePoll(run.forge, run.token),
88 sleep,
89 limit: Infinity,
90 ceilingMs,
91 intervalMs,
92 now,
93 onLine: (l, source) => run.say(`${source}: ${l}`),
94 });
95 if (watched.timeout) {
96 const ceilingS = Math.round(ceilingMs / 1000);
97 run.say(`The forge did not come back within the ceiling of ${ceilingS}s; nothing was posted.`);
98 run.say(JSON.stringify({ domain: domain.key, backSeconds: null, posted: false, ceilingSeconds: ceilingS }));
99 return 0;
138100 }
139
140 const backSeconds = Math.max(0, Math.round((now() - started) / 1000));
141 const body = { domain: domain.key, seen: { health: cap(health.values()), board: cap(board.values()) }, backSeconds };
101 const { health = [], board = [] } = watched.said;
102 const backSeconds = Math.max(0, Math.round(watched.healthyAt / 1000));
103 const body = { domain: domain.key, seen: { health: cap(health), board: cap(board) }, backSeconds };
142104 let res: Response;
143105 try {
144106 res = await fetch(`${run.forge}/api/v1/drills`, {
scripts/drills/watch.ts added +140
@@ -0,0 +1,140 @@
1 /**
2 * `watchDrill`: polls one URL until the forge is healthy again after having
3 * been unhealthy, and returns each distinct line it said meanwhile. Pure: the
4 * fetch and the sleep come in as arguments, nothing is read, posted or global.
5 * Times are milliseconds: the clock you pass, or else polls made times the
6 * interval. `watchPolls` is the loop for any poll that says healthy or not and
7 * what each of its sources said; `watchDrill` is it over one URL.
8 */
9
10 /** C0 and C1 control characters out (tab stays): the forge's words never drive the terminal. */
11 export const clean = (s: string): string => s.replace(/[\x00-\x08\x0a-\x1f\x7f-\x9f]/g, "");
12
13 /** Every string in a JSON value, or the text's own lines when it is not JSON. */
14 export function linesOf(text: string, contentType: string): string[] {
15 return rawLinesOf(text, contentType).map(clean).filter((l) => l.trim() !== "");
16 }
17
18 function rawLinesOf(text: string, contentType: string): string[] {
19 if (contentType.includes("json")) {
20 try {
21 const out: string[] = [];
22 const walk = (v: unknown): void => {
23 if (typeof v === "string") out.push(v);
24 else if (Array.isArray(v)) v.forEach(walk);
junipercrew outdated

watchDrill has no caller anywhere in the repo (grep finds only its test). That is dead code by the 'Clean' bar. Wire run.ts's loop through it in this pull, or extend it to what run.ts needs.
25 else if (v && typeof v === "object") Object.values(v).forEach(walk);
26 };
27 walk(JSON.parse(text));
28 return out;
29 } catch {
30 /* fall through to plain lines */
31 }
32 }
33 return text
34 .replace(/<(script|style)[\s\S]*?<\/\1>/gi, " ")
35 .replace(/<[^>]*>/g, "\n")
36 .replace(/&amp;/g, "&").replace(/&lt;/g, "<").replace(/&gt;/g, ">")
37 .replace(/&quot;/g, '"').replace(/&#0?39;/g, "'")
38 .split("\n")
junipercrew outdated

Unreachable is recorded as no line at all, so a forge that is fully down yields lines: []. Add a fixed line for a thrown fetch, with a test.
39 .map((l) => l.trim())
40 .filter((l) => l !== "");
41 }
42
43 /** A line with its numbers blanked, so a clock or a counter is not a new line. */
44 const shape = (l: string): string => l.replace(/\d+/g, "#");
45
46 /** The part of a Response that a watch reads. */
47 export interface Reply {
48 status: number;
49 text: () => Promise<string>;
50 headers: { get: (name: string) => string | null };
51 }
52
53 /** What a fetch that threw is recorded as saying. */
54 export const UNREACHABLE = "unreachable";
55
56 /** One look at the forge: is it healthy, and what each source said (named by source). */
57 export interface Poll {
58 healthy: boolean;
59 said: Record<string, string[]>;
60 }
61
62 export interface Watch {
63 poll: () => Promise<Poll>;
64 /** Waits one interval. */
65 sleep: (ms: number) => Promise<unknown>;
66 /** The most polls to make before giving up. */
67 limit: number;
68 intervalMs?: number;
69 /** Milliseconds on any clock; without it, time is the polls made times `intervalMs`. */
70 now?: () => number;
71 /** Gives up once this many milliseconds have passed, however few polls that took. */
72 ceilingMs?: number;
73 /** Called once with each new distinct line and its source, as it is first said. */
74 onLine?: (line: string, source: string) => void;
75 }
76
77 export type Watched =
78 | { timeout: false; startedAt: number; healthyAt: number; said: Record<string, string[]> }
79 | { timeout: true; startedAt: number | null; said: Record<string, string[]> };
80
81 /** Polls until healthy again after having been unhealthy; keeps each distinct line said while unhealthy. */
82 export async function watchPolls(w: Watch): Promise<Watched> {
83 const intervalMs = w.intervalMs ?? 1_000;
84 const seen = new Map<string, Map<string, string>>();
85 const said = (): Record<string, string[]> =>
86 Object.fromEntries([...seen].map(([source, m]) => [source, [...m.values()]]));
87 const began = w.now?.() ?? 0;
88 let startedAt: number | null = null;
89 for (let n = 0; n < w.limit; n++) {
90 const look = await w.poll();
91 const at = w.now ? w.now() - began : n * intervalMs;
92 if (look.healthy && startedAt !== null) return { timeout: false, startedAt, healthyAt: at, said: said() };
93 if (!look.healthy) {
94 startedAt ??= at;
95 for (const [source, lines] of Object.entries(look.said)) {
96 const m = seen.get(source) ?? new Map<string, string>();
97 seen.set(source, m);
98 for (const l of lines) if (!m.has(shape(l))) { m.set(shape(l), l); w.onLine?.(l, source); }
99 }
100 }
101 if (w.ceilingMs !== undefined && at >= w.ceilingMs) break;
102 await w.sleep(intervalMs);
103 }
104 return { timeout: true, startedAt, said: said() };
105 }
106
107 export interface WatchDrill extends Omit<Watch, "poll" | "onLine"> {
108 url: string;
109 fetch: (url: string) => Promise<Reply>;
110 onLine?: (line: string) => void;
111 }
112
113 export type WatchedDrill =
114 | { timeout: false; startedAt: number; healthyAt: number; lines: string[] }
115 | { timeout: true; startedAt: number | null; lines: string[] };
116
117 /** `watchPolls` over one URL: healthy is a status under 400, and the lines are its body's. */
118 export async function watchDrill(w: WatchDrill): Promise<WatchedDrill> {
119 const poll = async (): Promise<Poll> => {
120 try {
121 const res = await w.fetch(w.url);
122 if (res.status < 400) return { healthy: true, said: {} };
123 let lines: string[] = [];
124 try {
125 lines = linesOf(await res.text(), res.headers.get("content-type") ?? "");
126 } catch {
127 /* an unreadable body is unhealthy and says nothing */
128 }
129 return { healthy: false, said: { "": lines } };
130 } catch {
131 return { healthy: false, said: { "": [UNREACHABLE] } };
132 }
junipercrew

watchDrill has no production caller. Adopt it or remove it in a follow-up.
133 };
134 const { url: _url, fetch: _fetch, onLine, ...rest } = w;
135 const r = await watchPolls({ ...rest, poll, ...(onLine && { onLine: (l: string) => onLine(l) }) });
136 const lines = r.said[""] ?? [];
137 return r.timeout
138 ? { timeout: true, startedAt: r.startedAt, lines }
139 : { timeout: false, startedAt: r.startedAt, healthyAt: r.healthyAt, lines };
140 }
test/drill-poll.test.ts added +32
@@ -0,0 +1,32 @@
1 import { expect, test } from "bun:test";
2 import { forgePoll } from "../scripts/drills/run.ts";
3
4 const answers = (health: Response | Error, board: Response | Error): typeof fetch =>
5 (async (url: string) => {
6 const r = String(url).endsWith("/api/v1/health") ? health : board;
7 if (r instanceof Error) throw r;
8 return r.clone();
9 }) as unknown as typeof fetch;
10
11 const html = (status: number, body: string) => new Response(body, { status, headers: { "content-type": "text/html" } });
12 const ok = () => Response.json({ ok: true });
13
14 test("healthy only when health reports ok and the board answers", async () => {
15 expect((await forgePoll("http://f", "t", answers(ok(), html(200, "<p>hi</p>")))()).healthy).toBe(true);
16 expect((await forgePoll("http://f", "t", answers(ok(), html(503, "<p>stuck</p>")))()).healthy).toBe(false);
17 expect((await forgePoll("http://f", "t", answers(Response.json({ ok: false }), html(200, "hi")))()).healthy).toBe(false);
18 expect((await forgePoll("http://f", "t", answers(new Response("ok", { status: 200 }), html(200, "hi")))()).healthy).toBe(false);
19 });
20
21 test("what each page said is kept under its own source", async () => {
22 const look = await forgePoll("http://f", "t", answers(
23 Response.json({ ok: false, note: "Disk is nearly full" }, { status: 503 }),
24 html(503, "<p>The runner is stuck</p>"),
25 ))();
26 expect(look.said).toEqual({ health: ["Disk is nearly full"], board: ["The runner is stuck"] });
27 });
28
29 test("a page that cannot be reached says 'unreachable'", async () => {
30 const look = await forgePoll("http://f", "t", answers(new Error("refused"), html(200, "hi")))();
31 expect(look).toEqual({ healthy: false, said: { health: ["unreachable"], board: ["hi"] } });
32 });
test/drill-watch.test.ts added +82
@@ -0,0 +1,82 @@
1 import { expect, test } from "bun:test";
2 import { watchDrill } from "../scripts/drills/watch.ts";
3
4 const reply = (status: number, body: string, type = "text/plain") => ({
5 status,
6 text: async () => body,
7 headers: { get: (n: string) => (n === "content-type" ? type : null) },
8 });
9
10 function script(steps: Array<ReturnType<typeof reply> | Error>) {
11 let i = 0;
12 const slept: unknown[] = [];
13 return {
14 slept,
15 fetch: async () => {
16 const s = steps[Math.min(i++, steps.length - 1)]!;
17 if (s instanceof Error) throw s;
18 return s;
19 },
20 sleep: async (ms: number) => { slept.push(ms); },
21 };
22 }
23
24 test("returns distinct lines said while unhealthy, with times, without waiting", async () => {
25 const s = script([
26 reply(200, "ok"),
27 reply(503, "database locked\nretry 1"),
28 new Error("refused"),
29 reply(503, "database locked\nretry 2"),
30 reply(200, "ok"),
31 ]);
32 const r = await watchDrill({ url: "http://x/", fetch: s.fetch, sleep: s.sleep, limit: 10 });
33 expect(r).toEqual({ timeout: false, startedAt: 1000, healthyAt: 4000, lines: ["database locked", "retry 1", "unreachable"] });
34 expect(s.slept.length).toBe(4);
35 });
36
37 test("reads JSON strings and strips control characters", async () => {
38 const s = script([reply(500, JSON.stringify({ error: "bad\u001b[31m" }), "application/json"), reply(200, "ok")]);
39 const r = await watchDrill({ url: "http://x/", fetch: s.fetch, sleep: s.sleep, limit: 5 });
40 expect(r).toMatchObject({ timeout: false, lines: ["bad[31m"] });
41 });
42
43 test("times out when the forge never comes back, and when it never went down", async () => {
44 const down = script([reply(503, "down")]);
45 expect(await watchDrill({ url: "u", fetch: down.fetch, sleep: down.sleep, limit: 3 }))
46 .toEqual({ timeout: true, startedAt: 0, lines: ["down"] });
47 const up = script([reply(200, "ok")]);
48 expect(await watchDrill({ url: "u", fetch: up.fetch, sleep: up.sleep, limit: 3 }))
49 .toEqual({ timeout: true, startedAt: null, lines: [] });
50 });
51
52 test("a fetch that throws says 'unreachable'; an unreadable 5xx body says nothing but is still down", async () => {
53 const bad = { status: 502, text: async () => { throw new Error("reset"); }, headers: { get: () => null } };
54 const s = script([new Error("refused"), bad, reply(200, "ok")]);
55 const seen: string[] = [];
56 const r = await watchDrill({ url: "u", fetch: s.fetch, sleep: s.sleep, limit: 5, onLine: (l) => seen.push(l) });
57 expect(r).toEqual({ timeout: false, startedAt: 0, healthyAt: 2000, lines: ["unreachable"] });
58 expect(seen).toEqual(["unreachable"]);
59 });
60
61 test("an injected clock measures the times", async () => {
62 const s = script([reply(503, "down"), reply(200, "ok")]);
63 let t = 500;
64 const r = await watchDrill({ url: "u", fetch: s.fetch, sleep: async () => { t += 1700; }, limit: 5, now: () => t });
65 expect(r).toEqual({ timeout: false, startedAt: 0, healthyAt: 1700, lines: ["down"] });
66 });
67
68 test("the ceiling is time, not polls: slow polls time out long before the poll limit", async () => {
69 let t = 0;
70 let polls = 0;
71 const r = await watchDrill({
72 url: "u",
73 fetch: async () => { polls++; t += 4_000; return reply(503, "down"); },
74 sleep: async () => { t += 4_000; },
75 limit: 1000,
76 intervalMs: 1000,
77 ceilingMs: 20_000,
78 now: () => t,
79 });
80 expect(r).toEqual({ timeout: true, startedAt: 4_000, lines: ["down"] });
81 expect(polls).toBe(3);
82 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs