app/lib/health/ask-converge.mjs

Functions

Imports

It imports no other module of this repo. Imported by app/lib/health/repair.mjs.

Source

1/**
2 * One row of /api/health, as ship and the watchdog read it. `waitedMs` is set only by settleAskDrift.
3 *
4 * @typedef {{ name?: string, ok?: boolean, expected?: number, present?: number, waitedMs?: number }} HealthCheckRow
5 */
6
7export const ASK_POLL_ATTEMPTS = 12;
8export const ASK_POLL_INTERVAL_MS = 10_000;
9export const ASK_POLL_WINDOW_MS = 120_000;
10
11/**
12 * @param {object} options
13 * @param {() => Promise<{ ok: boolean, expected?: number, present?: number } | null>} options.reading
14 * @param {(ms: number) => Promise<unknown>} options.sleep
15 * @param {() => number} [options.now]
16 * @param {number} [options.attempts]
17 * @param {number} [options.intervalMs]
18 * @param {number} [options.windowMs]
19 * @param {(event: { poll: number, reading: any, error: string | null }) => void} [options.onPoll]
20 * @returns {Promise<{ converged: boolean, polls: number, latest: any, lastError: string | null }>}
21 */
22export async function awaitAskConvergence({
23  reading,
24  sleep,
25  now = Date.now,
26  attempts = ASK_POLL_ATTEMPTS,
27  intervalMs = ASK_POLL_INTERVAL_MS,
28  windowMs = ASK_POLL_WINDOW_MS,
29  onPoll = () => {},
30}) {
31  const deadline = now() + windowMs;
32  /** @type {any} */
33  let latest = null;
34  let polls = 0;
35  /** A thrown read is an unreadable poll, and its reason is carried out rather than dropped. */
36  /** @type {string | null} */
37  let lastError = null;
38
39  for (let i = 0; i < attempts && now() < deadline; i += 1) {
40    await sleep(intervalMs);
41    polls += 1;
42
43    /** @type {any} */
44    let observed = null;
45    /** @type {string | null} */
46    let error = null;
47    try {
48      observed = await reading();
49    } catch (thrown) {
50      // A failing HTTP status is not this: it is a readable answer, and the caller parses the body.
51      error = thrown instanceof Error ? thrown.message : String(thrown);
52      lastError = error;
53    }
54
55    const usable = observed && typeof observed.ok === "boolean" ? observed : null;
56    if (usable) latest = usable;
57    onPoll({ poll: polls, reading: usable, error });
58
59    if (usable && usable.ok) return { converged: true, polls, latest, lastError };
60  }
61
62  return { converged: false, polls, latest, lastError };
63}
64
65/**
66 * Ship's post-repair assertion reads health seconds after the Ask upload, and the index can list
67 * short then even when the upload's own write-back read was full (157 then 155, deploy run
68 * 36274700184). A failing row gets the same bounded window as the write-back; one that converges
69 * is ok, and one that does not keeps its latest counts and the time waited, for the miss.
70 *
71 * @param {object} options
72 * @param {HealthCheckRow | undefined} options.row the ask-index-drift row
73 * @param {() => Promise<{ ok: boolean, expected?: number, present?: number } | null>} options.reading
74 * @param {(ms: number) => Promise<unknown>} options.sleep
75 * @param {() => number} [options.now]
76 * @param {(event: { poll: number, reading: any, error: string | null }) => void} [options.onPoll]
77 * @returns {Promise<HealthCheckRow | undefined>}
78 */
79export async function settleAskDrift({ row, reading, sleep, now = Date.now, onPoll }) {
80  if (!row || row.ok === true) return row;
81
82  const started = now();
83  const { converged, latest } = await awaitAskConvergence({ reading, sleep, now, onPoll });
84  // The row's own counts are the stale ones; a converged row carries none, as health's ok rows do.
85  if (converged) return { name: row.name, ok: true };
86  const counts =
87    latest && typeof latest.expected === "number" && typeof latest.present === "number"
88      ? { expected: latest.expected, present: latest.present }
89      : {};
90  return { ...row, ...counts, waitedMs: now() - started };
91}