From 9cfa55c9b231e3907cef5ea0dbe505b4fdc57f9c Mon Sep 17 00:00:00 2001 From: Saty <5752427+SatyPardus@users.noreply.github.com> Date: Sun, 18 Oct 2026 05:00:59 +0200 Subject: [PATCH] Remote checks fail fast when the worker is down and no longer pile up While the worker was stopped (maintenance), checks were queued with nobody to run them; agents timed out ("fetch failed", 502 after a restart) and every retry queued another copy. Now: the worker writes a heartbeat and a check is refused with 503 at once while it is down; a retry of the same files joins the job already queued or running; a job nobody waits for any more is dropped. TD_KEEP_WORKER=1 deploys without touching the worker. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RNwtXPYqFu3oKaaXwYwNWW --- deploy/update.sh | 10 +++++-- src/server/app.js | 10 ++++--- src/server/checks.js | 63 +++++++++++++++++++++++++++++++++++--------- src/server/db.js | 3 ++- test/server.test.js | 4 +++ tools/deploy.sh | 3 ++- 6 files changed, 73 insertions(+), 20 deletions(-) diff --git a/deploy/update.sh b/deploy/update.sh index 3e3e0c7..0a286ea 100644 --- a/deploy/update.sh +++ b/deploy/update.sh @@ -1,6 +1,7 @@ #!/bin/sh # Updates the coordinator's code without taking the site down. Run as root on the server: # sh deploy/update.sh +# TD_KEEP_WORKER=1 leaves the worker alone (during maintenance with the worker stopped). # # The new release is copied and syntax-checked next to the running one; nothing changes for the # running site until then. /opt/togetherdecomp is a symlink that switches to the new release in @@ -67,8 +68,13 @@ else systemctl is-active --quiet togetherdecomp.service || { journalctl -u togetherdecomp -n 20 --no-pager; exit 1; } # The worker finishes its running verification, merge or check before it stops (no work is # redone), then starts on the new release. That can take minutes, so don't wait for it here. - echo "== the worker restarts once its running work is done" - systemctl restart --no-block togetherdecomp-worker.service + if [ -n "${TD_KEEP_WORKER:-}" ]; then + # Maintenance (e.g. a rebuild with the worker stopped): leave it; it runs the new code when started. + echo "== worker left as it is (TD_KEEP_WORKER)" + else + echo "== the worker restarts once its running work is done" + systemctl restart --no-block togetherdecomp-worker.service + fi fi # Keep the last three releases for a quick rollback (ln -sfn /opt/togetherdecomp). diff --git a/src/server/app.js b/src/server/app.js index d5241df..dccb929 100644 --- a/src/server/app.js +++ b/src/server/app.js @@ -15,7 +15,7 @@ import { Originals } from './original.js'; import { downloadsDir, installPs1, installSh } from './install.js'; import { ServerGhidra } from './ghidra.js'; import { functionRegions } from './split.js'; -import { CheckQueue } from './checks.js'; +import { CheckQueue, WorkerUnavailable } from './checks.js'; import { withGuidance } from './guidance.js'; import { Router, HttpError, readJson, sendJson, RateLimiter } from './http.js'; import { PEImage } from '../shared/pe.js'; @@ -494,7 +494,7 @@ export function createApp(ctx) { const checkLimiter = new RateLimiter(config.checksPerHour ?? 60, 3600_000); - router.on('POST', '/api/check', async ({ req }) => { + router.on('POST', '/api/check', async ({ req, res }) => { const user = auth(req); checkLimiter.check(user.id); const body = await readJson(req); @@ -506,10 +506,12 @@ export function createApp(ctx) { if (!repo.isAncestor(body.baseCommit)) throw new HttpError(409, 'your base commit is not part of main; run td sync'); const merged = repo.mergeOntoHead(body.baseCommit, body.files); if (merged.conflicts.length) return { gate: 'conflict', conflicts: merged.conflicts }; + const gone = new AbortController(); + res.on('close', () => res.writableFinished || gone.abort()); try { - return guided(await ctx.checks.run(user.id, merged.files)); + return guided(await ctx.checks.run(user.id, merged.files, { signal: gone.signal })); } catch (e) { - throw new HttpError(503, e.message); + throw new HttpError(503, e.message, e instanceof WorkerUnavailable ? { retryAfterSeconds: 120 } : undefined); } }); diff --git a/src/server/checks.js b/src/server/checks.js index 4a2c24e..ac05de9 100644 --- a/src/server/checks.js +++ b/src/server/checks.js @@ -2,36 +2,74 @@ // The web process queues them in the database and waits for the result, so it never compiles // anything itself and either process can restart without the other. +import crypto from 'node:crypto'; import { now } from './db.js'; import { summarizeCheck } from '../shared/check.js'; const POLL_MS = 1000; const KEEP_MS = 24 * 3600_000; +// The worker records that it is alive; without it, checks are refused at once instead of waiting. +const HEARTBEAT_KEY = 'worker_seen'; +const HEARTBEAT_MS = 5000; +const WORKER_GONE_MS = 30_000; + +export class WorkerUnavailable extends Error {} export class CheckQueue { constructor({ store, repo, verifier, config, log = console }) { Object.assign(this, { store, repo, verifier, config, log }); this.running = 0; + this.waiters = new Map(); // web side: job id -> requests waiting for it } - /** Web side: queues a check of already merged files and waits for its result. */ - async run(userId, files) { - const id = Number(this.store.run(`INSERT INTO check_jobs (user_id, files_json, state, created_at) VALUES (?, ?, 'queued', ?)`, userId, JSON.stringify(files), now()).lastInsertRowid); - // A queued check waits for the ones ahead of it, then builds; give up well after both. - const deadline = Date.now() + 3 * (this.config.buildTimeoutMs ?? 300_000) + 60_000; - while (Date.now() < deadline) { - await new Promise((r) => setTimeout(r, POLL_MS)); - const job = this.store.get('SELECT state, result_json FROM check_jobs WHERE id = ?', id); - if (job.state === 'done') return JSON.parse(job.result_json); - if (job.state === 'failed') throw new Error(JSON.parse(job.result_json).error); + workerAlive() { + return now() - Number(this.store.getMeta(HEARTBEAT_KEY) ?? 0) < WORKER_GONE_MS; + } + + /** + * Web side: queues a check of already merged files and waits for its result. A retry of the same + * files joins the job already on its way; a job nobody waits for any more is dropped. + * @param {AbortSignal} [signal] aborted when the client disconnects + */ + async run(userId, files, { signal } = {}) { + if (!this.workerAlive()) throw new WorkerUnavailable('the compile service is paused for maintenance; try again in a few minutes'); + const filesJson = JSON.stringify(files); + const hash = crypto.createHash('sha256').update(filesJson).digest('hex'); + const existing = this.store.get(`SELECT id FROM check_jobs WHERE user_id = ? AND files_hash = ? AND state IN ('queued', 'running') ORDER BY id DESC LIMIT 1`, userId, hash); + const id = existing?.id ?? Number(this.store.run(`INSERT INTO check_jobs (user_id, files_json, files_hash, state, created_at) VALUES (?, ?, ?, 'queued', ?)`, userId, filesJson, hash, now()).lastInsertRowid); + this.waiters.set(id, (this.waiters.get(id) ?? 0) + 1); + try { + // A queued check waits for the ones ahead of it, then builds; give up well after both. + const deadline = Date.now() + 3 * (this.config.buildTimeoutMs ?? 300_000) + 60_000; + while (Date.now() < deadline && !signal?.aborted) { + await new Promise((r) => setTimeout(r, POLL_MS)); + const job = this.store.get('SELECT state, result_json FROM check_jobs WHERE id = ?', id); + if (job.state === 'done') return JSON.parse(job.result_json); + if (job.state === 'failed') throw new Error(JSON.parse(job.result_json).error); + if (job.state === 'queued' && !this.workerAlive()) throw new WorkerUnavailable('the compile service is paused for maintenance; try again in a few minutes'); + } + throw new Error(signal?.aborted ? 'the client went away' : 'the check did not finish in time; the server may be busy, try again later'); + } finally { + const left = this.waiters.get(id) - 1; + if (left > 0) this.waiters.set(id, left); + else { + this.waiters.delete(id); + this.store.run(`UPDATE check_jobs SET state = 'abandoned' WHERE id = ? AND state = 'queued'`, id); + } } - this.store.run(`UPDATE check_jobs SET state = 'abandoned' WHERE id = ? AND state = 'queued'`, id); - throw new Error('the check did not finish in time; the server may be busy, try again later'); } /** Worker side. */ start() { this.store.run(`UPDATE check_jobs SET state = 'queued' WHERE state = 'running'`); + this.store.setMeta(HEARTBEAT_KEY, now()); + this.heartbeat = setInterval(() => { + try { + this.store.setMeta(HEARTBEAT_KEY, now()); + } catch (e) { + this.log.error(`worker heartbeat: ${e.message}`); + } + }, HEARTBEAT_MS); this.timer = setInterval(() => { try { this.pump(); @@ -43,6 +81,7 @@ export class CheckQueue { stop() { clearInterval(this.timer); + clearInterval(this.heartbeat); } pump() { diff --git a/src/server/db.js b/src/server/db.js index c021d6c..1f331ae 100644 --- a/src/server/db.js +++ b/src/server/db.js @@ -232,7 +232,8 @@ export class Store { this.addColumn('func_status', 'confirmed', 'INTEGER NOT NULL DEFAULT 0'); // Fingerprint of main's compiled code per function (submissions may only change their own tasks). this.addColumn('func_status', 'code_hash', 'TEXT'); - this.addColumn('versions', 'label', 'TEXT'); // human-readable name, e.g. "World of Warcraft 5.0.3 (build 15890, x64)" + this.addColumn('versions', 'label', 'TEXT'); + this.addColumn('check_jobs', 'files_hash', 'TEXT'); // a retry of the same check joins the queued one // human-readable name, e.g. "World of Warcraft 5.0.3 (build 15890, x64)" this.addColumn('attempts', 'batch_id', 'INTEGER REFERENCES batches(id)'); this.db.exec('CREATE INDEX IF NOT EXISTS attempts_batch ON attempts(batch_id)'); } diff --git a/test/server.test.js b/test/server.test.js index fd8b498..0a68fda 100644 --- a/test/server.test.js +++ b/test/server.test.js @@ -383,6 +383,10 @@ test('remote checks are queued in the database and run by the worker', async () check: { annotationErrors: [], units: [{ file: 'src/rc.cpp', compiled: true, lintErrors: [], unannotated: [], bindings: {}, inferred: [], includes: [], errors: [], functions: [] }] }, rec: { conflicts: [], unknown: [], learned: [], learnedAddresses: [] }, }); + // No worker running: refused at once instead of waiting. + const paused = await call('POST', '/api/check', { baseCommit: ctx.repo.head(), files: { 'src/rc.cpp': 'int rc() { return 0; }\n' } }, user); + assert.equal(paused.status, 503); + assert.match(paused.body.error, /paused for maintenance/); ctx.checks.start(); try { const r = await call('POST', '/api/check', { baseCommit: ctx.repo.head(), files: { 'src/rc.cpp': 'int rc() { return 0; }\n' } }, user); diff --git a/tools/deploy.sh b/tools/deploy.sh index 63d2e00..66d9c77 100644 --- a/tools/deploy.sh +++ b/tools/deploy.sh @@ -1,6 +1,7 @@ #!/bin/sh # Deploys the committed HEAD to the coordinator without taking the site down (deploy/update.sh). # tools/deploy.sh [--td] +# TD_KEEP_WORKER=1 tools/deploy.sh leaves the worker as it is (see deploy/update.sh). # --td also publishes the td binaries from dist/ (node tools/build-td/build.mjs) for download and # self-update. TD_HOST names the ssh host (default: td-server). set -eu @@ -30,5 +31,5 @@ ssh "$HOST" "set -e chmod 0755 \$D/td-linux-x64 echo 'td binaries published' fi - sh \"\$d/deploy/update.sh\" \"\$d\" + TD_KEEP_WORKER='${TD_KEEP_WORKER:-}' sh \"\$d/deploy/update.sh\" \"\$d\" rm -rf \"\$d\""