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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RNwtXPYqFu3oKaaXwYwNWW

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 <unpacked-source-dir>
+# 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"
+  if [ -n "${TD_KEEP_WORKER:-}" ]; then
-  systemctl restart --no-block togetherdecomp-worker.service
+    # 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 <release> /opt/togetherdecomp).

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);
     }
   });
 

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. */
+  workerAlive() {
-  async run(userId, files) {
+    return now() - Number(this.store.getMeta(HEARTBEAT_KEY) ?? 0) < WORKER_GONE_MS;
-    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) {
+   * Web side: queues a check of already merged files and waits for its result. A retry of the same
-      await new Promise((r) => setTimeout(r, POLL_MS));
+   * files joins the job already on its way; a job nobody waits for any more is dropped.
-      const job = this.store.get('SELECT state, result_json FROM check_jobs WHERE id = ?', id);
+   * @param {AbortSignal} [signal]  aborted when the client disconnects
-      if (job.state === 'done') return JSON.parse(job.result_json);
+   */
-      if (job.state === 'failed') throw new Error(JSON.parse(job.result_json).error);
+  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() {

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)');
   }

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);

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\""