commit 838cff4a97f9c3a279827b20004344f2f6cefc5f
parent b7f7a1195135bf9ec0f122f820985c4bbb582aa0
Author: I Mean I'm Just Saying <imeanimjustsaying@kiwifarms.st>
Date: Tue, 22 Sep 2026 16:08:50 -0400
ops: --wait survives a poll that fails, and can be bounded
The editor is one process, so a busy in-process build-index starves the event
loop until `fetch` rejects outright — and followJob treated that rejection as
the job's problem, printing a stack trace about work that was running fine and
went on to finish. Anyone who had run `pnpm ops build-site --wait` had seen it;
the note in plans/curated-tags.md telling the operator to go poll
/api/jobs/active by hand is the workaround written down.
Now every poll is caught, the wait backs off 1→2→…→30 s rather than hammering a
server that is already struggling, and `from` survives the failure so no log
text is lost. After three consecutive failures the question changes from "what
does the log say" to "is the job still there": /api/jobs/active is cheap and is
what the editor's own head polls. Listed ⇒ keep waiting. Absent ⇒ it went
terminal unobserved, so one last log poll reads the status it ended with — and
if that fails too the follow errors rather than claiming an outcome it never
read.
--wait-timeout <seconds> is for a caller that cannot hang. Default none: the
honest answer for a queue that may hold a job behind hours of other work is to
wait. followJob takes an injected {fetch, sleep, now} so the four paths are
unit-testable without a server.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Diffstat:
2 files changed, 221 insertions(+), 15 deletions(-)
diff --git a/scripts/archilyzer-ops.mjs b/scripts/archilyzer-ops.mjs
@@ -8,7 +8,8 @@
//
// USAGE
//
-// pnpm ops <action> [--json '<body>' | --file <path>] [--wait] [--quiet]
+// pnpm ops <action> [--json '<body>' | --file <path>] [--wait]
+// [--wait-timeout <seconds>] [--quiet]
// pnpm ops get channel <slug> [--counts]
// pnpm ops get tags [<tagId>]
// pnpm ops list
@@ -42,7 +43,9 @@
// --wait follows /api/jobs/<jobId>/log to the end for a job-starting action and
// exits 0 only if the job finished `done`. Without it the command returns as
// soon as the job is QUEUED, which is the honest answer: the queue may hold it
-// behind other work for hours.
+// behind other work for hours. A poll that fails does NOT end the follow — see
+// followJob — and --wait-timeout <seconds> is there for a caller that cannot
+// wait indefinitely.
//
// The response JSON is printed verbatim on stdout (log lines from --wait go to
// stderr), so `pnpm ops … | jq` works.
@@ -104,10 +107,29 @@ export function parseArgs(argv) {
let wait = false;
let quiet = false;
let counts = false;
+ let waitTimeout = null;
+ const readTimeout = (raw) => {
+ const n = Number(raw);
+ if (!Number.isFinite(n) || n <= 0) {
+ return { error: "--wait-timeout needs a positive number of seconds" };
+ }
+ waitTimeout = n;
+ return null;
+ };
for (let i = 0; i < argv.length; i++) {
const arg = argv[i];
if (arg === "--wait") {
wait = true;
+ } else if (arg === "--wait-timeout") {
+ const raw = argv[++i];
+ if (raw === undefined) {
+ return { error: "--wait-timeout needs a number of seconds" };
+ }
+ const err = readTimeout(raw);
+ if (err) return err;
+ } else if (arg.startsWith("--wait-timeout=")) {
+ const err = readTimeout(arg.slice("--wait-timeout=".length));
+ if (err) return err;
} else if (arg === "--quiet") {
quiet = true;
} else if (arg === "--counts") {
@@ -167,6 +189,7 @@ export function parseArgs(argv) {
path: GETTERS[noun](positional[2], counts),
wait: false,
quiet,
+ waitTimeout,
};
}
const action = positional[0];
@@ -194,18 +217,26 @@ export function parseArgs(argv) {
...(action === "tag-videos" ? { defaultSource: agentSource() } : {}),
wait,
quiet,
+ waitTimeout,
};
}
export function usage() {
return [
"Usage: pnpm ops <action> [--json '<body>' | --file <path>] [--wait]",
+ " [--wait-timeout <seconds>] [--quiet]",
" pnpm ops get channel <slug> [--counts]",
" pnpm ops get tags [<tagId>]",
" pnpm ops list",
"",
`Actions: ${ACTIONS.join(", ")}`,
"",
+ "--wait follows the job's log and survives a poll that fails (a busy",
+ " in-process build starves the server): it backs off and, after three",
+ " failures, asks /api/jobs/active whether the job is still there.",
+ "--wait-timeout <seconds> gives up and exits 1 instead of waiting forever.",
+ " Default: no timeout — the queue may legitimately hold a job for hours.",
+ "",
"Env: ARCHILYZER_EDITOR_URL (default http://localhost:3001), WORKER_TOKEN,",
" ARCHILYZER_AGENT (provenance of a tag write; default \"cli\")",
].join("\n");
@@ -229,19 +260,98 @@ function authHeaders() {
// the browser polls it same-origin with no token. Sending one here implied a
// gate that does not exist, which is worse than sending nothing: the next
// person to read this would conclude the endpoint was protected.
-async function followJob(jobId, quiet) {
+//
+// A POLL FAILURE IS NOT A JOB FAILURE, and this used to treat them as the same
+// thing. The editor is single-process: a busy in-process build-index starves
+// the event loop for long enough that `fetch` rejects outright, and one
+// rejection ended the follow with a stack trace about a job that was running
+// fine and went on to finish. So every poll is caught, the wait backs off
+// 1→2→4…→30 s instead of hammering a server that is already struggling, and
+// `from` is kept across the failure so not one line of log text is lost.
+//
+// After three consecutive failures the endpoint is no longer trusted to answer
+// at all, and the question becomes a different one — IS THE JOB STILL THERE?
+// `/api/jobs/active` is cheap and is what the editor's own head polls. Listed
+// ⇒ keep waiting, however long that takes. Absent ⇒ it ended while we could not
+// see it, so one last log poll reads the terminal status; if even that fails,
+// the follow gives up rather than claiming an outcome it never read.
+//
+// `--wait-timeout` bounds the whole thing for a caller that cannot hang (CI,
+// an agent). Default none, because the honest default for a queue that may hold
+// a job behind hours of other work is to wait.
+export async function followJob(jobId, quiet, opts = {}) {
+ const doFetch = opts.fetch ?? fetch;
+ const sleep = opts.sleep ?? ((ms) => new Promise((r) => setTimeout(r, ms)));
+ const now = opts.now ?? (() => Date.now());
+ const deadline =
+ opts.timeoutSeconds > 0 ? now() + opts.timeoutSeconds * 1000 : null;
+ const base = opts.baseUrl ?? baseUrl();
+ const id = encodeURIComponent(jobId);
+
let from = 0;
+ let failures = 0;
+ let backoff = 1000;
+
+ // One log poll. Returns the status, or null when the poll itself failed —
+ // never throws, so a transient fetch rejection cannot end the follow.
+ const pollLog = async () => {
+ try {
+ const res = await doFetch(`${base}/api/jobs/${id}/log?from=${from}`);
+ if (!res.ok) return null;
+ const payload = await res.json();
+ if (payload.content && !quiet) process.stderr.write(payload.content);
+ // Only advance once the chunk is in hand: a poll that failed halfway
+ // must re-ask for the same offset.
+ from = payload.nextOffset ?? from;
+ return payload.status ?? null;
+ } catch {
+ return null;
+ }
+ };
+
+ const stillListed = async () => {
+ try {
+ const res = await doFetch(`${base}/api/jobs/active`);
+ if (!res.ok) return null;
+ const payload = await res.json();
+ const jobs = Array.isArray(payload.jobs) ? payload.jobs : [];
+ return jobs.some((j) => j && j.id === jobId);
+ } catch {
+ return null;
+ }
+ };
+
for (;;) {
- const res = await fetch(
- `${baseUrl()}/api/jobs/${encodeURIComponent(jobId)}/log?from=${from}`,
- );
- if (!res.ok) throw new Error(`log poll failed: HTTP ${res.status}`);
- const payload = await res.json();
- if (payload.content && !quiet) process.stderr.write(payload.content);
- from = payload.nextOffset ?? from;
- const status = payload.status;
- if (status !== "queued" && status !== "running") return status;
- await new Promise((r) => setTimeout(r, 1000));
+ const status = await pollLog();
+ if (status !== null) {
+ failures = 0;
+ backoff = 1000;
+ if (status !== "queued" && status !== "running") return status;
+ } else {
+ failures++;
+ if (failures >= 3) {
+ const listed = await stillListed();
+ if (listed === false) {
+ // Gone from the active list: it went terminal while the log endpoint
+ // was unreachable. One more try at the status it ended with.
+ const final = await pollLog();
+ if (final !== null) return final;
+ throw new Error(
+ `lost contact with job ${jobId}: it is no longer active and its log could not be read`,
+ );
+ }
+ // Listed (or the probe failed too) — keep waiting. A job we can still
+ // see is a job to wait for.
+ if (listed === true) failures = 0;
+ }
+ backoff = Math.min(backoff * 2, 30_000);
+ }
+ if (deadline !== null && now() >= deadline) {
+ throw new Error(
+ `--wait-timeout: gave up after ${opts.timeoutSeconds}s waiting for job ${jobId}; it is still running and can be followed on /jobs`,
+ );
+ }
+ await sleep(status !== null ? 1000 : backoff);
}
}
@@ -325,7 +435,9 @@ async function main() {
}
let worst = 0;
for (const jobId of jobIds) {
- const status = await followJob(jobId, parsed.quiet);
+ const status = await followJob(jobId, parsed.quiet, {
+ timeoutSeconds: parsed.waitTimeout ?? 0,
+ });
console.error(`[${jobId}] ${status}`);
if (status !== "done") worst = 1;
}
diff --git a/scripts/archilyzer-ops.test.mjs b/scripts/archilyzer-ops.test.mjs
@@ -5,7 +5,7 @@
// Run with: pnpm test:scripts
import assert from "node:assert/strict";
import test from "node:test";
-import { parseArgs, usage } from "./archilyzer-ops.mjs";
+import { followJob, parseArgs, usage } from "./archilyzer-ops.mjs";
test("no arguments prints usage", () => {
assert.equal(parseArgs([]).help, true);
@@ -118,3 +118,97 @@ test("get tags reads the whole vocabulary, or one tag's assignments", () => {
// The optional argument is per-noun: `get channel` still demands a slug.
assert.match(parseArgs(["get", "channel"]).error, /needs an argument/);
});
+
+// ─── --wait, and the poll that fails ───
+
+test("--wait-timeout is parsed, and refuses a non-number", () => {
+ assert.equal(parseArgs(["build-index", "--wait", "--wait-timeout", "90"]).waitTimeout, 90);
+ assert.equal(parseArgs(["build-index", "--wait-timeout=90"]).waitTimeout, 90);
+ // Default is NONE: the queue may legitimately hold a job for hours.
+ assert.equal(parseArgs(["build-index", "--wait"]).waitTimeout, null);
+ assert.match(parseArgs(["build-index", "--wait-timeout"]).error, /needs a number/);
+ assert.match(parseArgs(["build-index", "--wait-timeout", "soon"]).error, /positive number/);
+ assert.match(parseArgs(["build-index", "--wait-timeout", "0"]).error, /positive number/);
+ assert.match(usage(), /--wait-timeout/);
+});
+
+// A fake editor. `log` is the sequence of log-poll outcomes — an Error is
+// thrown at the caller the way a starved server makes `fetch` reject.
+function fakeEditor({ log = [], active = [] } = {}) {
+ const calls = { log: 0, active: 0 };
+ const reply = (body) => ({ ok: true, json: async () => body });
+ const doFetch = async (url) => {
+ if (url.includes("/api/jobs/active")) {
+ const next = active[Math.min(calls.active, active.length - 1)];
+ calls.active++;
+ if (next instanceof Error) throw next;
+ return reply({ jobs: next });
+ }
+ const next = log[Math.min(calls.log, log.length - 1)];
+ calls.log++;
+ if (next instanceof Error) throw next;
+ return reply(next);
+ };
+ return { doFetch, calls };
+}
+
+const follow = (jobId, editor, opts = {}) =>
+ followJob(jobId, true, {
+ fetch: editor.doFetch,
+ sleep: async () => {},
+ baseUrl: "http://editor",
+ ...opts,
+ });
+
+test("a poll that rejects does not end the follow", async () => {
+ // THE BUG THIS FIXES: one `fetch failed` while an in-process build-index
+ // starved the server used to abort --wait and report a job that was fine.
+ const editor = fakeEditor({
+ log: [
+ { content: "a", nextOffset: 1, status: "running" },
+ new Error("fetch failed"),
+ new Error("fetch failed"),
+ { content: "b", nextOffset: 2, status: "done" },
+ ],
+ });
+ assert.equal(await follow("j1", editor), "done");
+ // Two failures is below the probe threshold, so it never asked.
+ assert.equal(editor.calls.active, 0);
+});
+
+test("after three failures it asks whether the job is still there", async () => {
+ const boom = new Error("fetch failed");
+ // Still listed ⇒ keep waiting, and the follow ends on the status it finally
+ // reads rather than on a guess.
+ const waiting = fakeEditor({
+ log: [boom, boom, boom, { content: "", nextOffset: 0, status: "done" }],
+ active: [[{ id: "j1" }]],
+ });
+ assert.equal(await follow("j1", waiting), "done");
+ assert.equal(waiting.calls.active, 1);
+
+ // Absent from the active list ⇒ it ended while the log was unreachable, so
+ // one final poll reads the terminal status. `failed` is reported, not hidden.
+ const gone = fakeEditor({
+ log: [boom, boom, boom, { content: "", nextOffset: 0, status: "failed" }],
+ active: [[]],
+ });
+ assert.equal(await follow("j1", gone), "failed");
+});
+
+test("a job that is gone AND unreadable refuses to claim an outcome", async () => {
+ const boom = new Error("fetch failed");
+ const editor = fakeEditor({ log: [boom], active: [[]] });
+ await assert.rejects(follow("j1", editor), /lost contact with job j1/);
+});
+
+test("--wait-timeout gives up on a job that never ends", async () => {
+ const editor = fakeEditor({
+ log: [{ content: "", nextOffset: 0, status: "running" }],
+ });
+ let clock = 0;
+ await assert.rejects(
+ follow("j1", editor, { timeoutSeconds: 5, now: () => (clock += 3000) }),
+ /--wait-timeout: gave up after 5s/,
+ );
+});
