Run the website and the worker as separate processes

td-server start web | worker | all. The web process serves the website, API
and git; the worker verifies, merges, runs remote checks and refreshes the
Ghidra project. They share the database and the repository, so restarting
one never interrupts the other. Remote checks are queued in check_jobs and
picked up by the worker. Web files are read from disk per request, so page
changes need no restart at all. The web unit no longer has docker access.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RNwtXPYqFu3oKaaXwYwNWW
This commit is contained in:
Saty 2026-10-18 03:48:50 +02:00
commit 2ec1bda2cc
8 changed files with 201 additions and 43 deletions

View file

@ -84,6 +84,7 @@ fs.writeFileSync(file, JSON.stringify(c, null, 2) + "\n");
echo "== systemd"
install -m 0644 "$SRC/deploy/togetherdecomp.service" /etc/systemd/system/togetherdecomp.service
install -m 0644 "$SRC/deploy/togetherdecomp-worker.service" /etc/systemd/system/togetherdecomp-worker.service
install -m 0644 "$SRC/deploy/nftables-td.nft" /etc/nftables-td.nft
cat > /etc/systemd/system/td-firewall.service <<'EOF'
[Unit]
@ -106,8 +107,8 @@ caddy validate --config /etc/caddy/Caddyfile --adapter caddyfile >/dev/null
systemctl daemon-reload
systemctl enable --now td-firewall.service
systemctl restart td-firewall.service
systemctl enable togetherdecomp.service caddy.service
systemctl restart togetherdecomp.service caddy.service
systemctl enable togetherdecomp.service togetherdecomp-worker.service caddy.service
systemctl restart togetherdecomp.service togetherdecomp-worker.service caddy.service
sleep 2
systemctl --no-pager --lines=5 status togetherdecomp.service | head -12
echo "== done: https://$HOST"

View file

@ -0,0 +1,43 @@
[Unit]
Description=Clank Together worker: verification, merges, remote checks, Ghidra refresh
After=network-online.target docker.service
Wants=network-online.target
[Service]
Type=simple
User=togetherdecomp
Group=togetherdecomp
SupplementaryGroups=docker
WorkingDirectory=/var/lib/togetherdecomp
ExecStart=/opt/node/bin/node --disable-warning=ExperimentalWarning /opt/togetherdecomp/src/server/cli.js start worker --data /var/lib/togetherdecomp
Restart=on-failure
# Restarting re-queues whatever was being verified.
RestartSec=5
Environment=NODE_ENV=production
# Hardening. The process needs: its data dir, the docker socket, git.
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/var/lib/togetherdecomp
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectKernelLogs=true
ProtectControlGroups=true
ProtectClock=true
ProtectHostname=true
RestrictSUIDSGID=true
RestrictRealtime=true
RestrictNamespaces=true
LockPersonality=true
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
CapabilityBoundingSet=
SystemCallArchitectures=native
# 0022 so the sandboxed (uid-remapped) compiler can read job files; the data
# dir itself is 0700, which keeps everything private to this user.
UMask=0022
[Install]
WantedBy=multi-user.target

View file

@ -1,20 +1,19 @@
[Unit]
Description=Clank Together coordinator
After=network-online.target docker.service
Description=Clank Together coordinator: website, API and git
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=togetherdecomp
Group=togetherdecomp
SupplementaryGroups=docker
WorkingDirectory=/var/lib/togetherdecomp
ExecStart=/opt/node/bin/node --disable-warning=ExperimentalWarning /opt/togetherdecomp/src/server/cli.js start --data /var/lib/togetherdecomp
ExecStart=/opt/node/bin/node --disable-warning=ExperimentalWarning /opt/togetherdecomp/src/server/cli.js start web --data /var/lib/togetherdecomp
Restart=on-failure
RestartSec=5
Environment=NODE_ENV=production
# Hardening. The process needs: its data dir, the docker socket, git.
# Hardening. The process needs: its data dir and git. Compiling happens in togetherdecomp-worker.
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true

View file

@ -15,11 +15,11 @@ 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 { Router, HttpError, readJson, sendJson, RateLimiter } from './http.js';
import { PEImage } from '../shared/pe.js';
import { loadProject, loadToolchainLock } from '../shared/project.js';
import { DEFAULT_LIMITS, lintSubmission } from '../shared/lint.js';
import { summarizeCheck } from '../shared/check.js';
import { hex } from '../shared/annotations.js';
const WEB_DIR = path.join(path.dirname(fileURLToPath(import.meta.url)), 'web');
@ -80,12 +80,15 @@ export function openServer(dataDir, { log = console } = {}) {
const originals = new Originals({ store });
const ghidra = new ServerGhidra({ config, store, originals, log });
merger.onMerged = (commit) => ghidra.scheduleRefresh(commit);
return { config, store, repo, project, tasks, verifier, merger, getPe, accounts, peers, originals, ghidra };
const checks = new CheckQueue({ store, repo, verifier, config, log });
return { config, store, repo, project, tasks, verifier, merger, getPe, accounts, peers, originals, ghidra, checks };
}
export function createApp(ctx) {
const { config, store, repo, project, tasks, merger, accounts, peers } = ctx;
const router = new Router();
// In a separate web process the worker picks new work up from the database by itself.
const runsWorker = ctx.runsWorker ?? true;
downloadsRoot = downloadsDir(config);
const submitLimiter = new RateLimiter(config.submitsPerHour, 3600_000);
const symbolLimiter = new RateLimiter(120, 3600_000);
@ -293,7 +296,7 @@ export function createApp(ctx) {
agentMeta: body.agentMeta ? JSON.stringify(body.agentMeta).slice(0, 1000) : null,
envDigest: body.envDigest ? String(body.envDigest).slice(0, 200) : null,
});
ctx.verifier.pump();
if (runsWorker) ctx.verifier.pump();
return r;
};
@ -452,19 +455,6 @@ export function createApp(ctx) {
// (contributors without a local MSVC toolchain). Same static gate as a submission.
const checkLimiter = new RateLimiter(config.checksPerHour ?? 60, 3600_000);
let checksRunning = 0;
const checkWaiters = [];
const withCheckSlot = async (fn) => {
const max = config.remoteCheckConcurrency ?? 2;
if (checksRunning >= max) await new Promise((r) => checkWaiters.push(r));
checksRunning++;
try {
return await fn();
} finally {
checksRunning--;
checkWaiters.shift()?.();
}
};
router.on('POST', '/api/check', async ({ req }) => {
const user = auth(req);
@ -478,10 +468,11 @@ 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 };
return withCheckSlot(async () => {
const { check, rec } = await ctx.verifier.check(merged.files, `check-${user.id}-${Date.now()}`);
return { gate: 'ok', head: repo.head(), ...summarizeCheck(check, rec, new Set(Object.keys(merged.files))) };
});
try {
return await ctx.checks.run(user.id, merged.files);
} catch (e) {
throw new HttpError(503, e.message);
}
});
// --- peer review (agents, API token or session)
@ -535,7 +526,7 @@ export function createApp(ctx) {
const user = auth(req, { role: 'reviewer', session: true });
const body = await readJson(req, 16 * 1024);
const r = merger.review(user, intParam(params.id), String(body.action ?? ''), body.comment ? String(body.comment) : null);
if (r.state === 'approved') setImmediate(() => merger.pump());
if (r.state === 'approved' && runsWorker) setImmediate(() => merger.pump());
return r;
});
@ -544,7 +535,7 @@ export function createApp(ctx) {
const user = auth(req, { role: 'reviewer', session: true });
const body = await readJson(req, 16 * 1024);
const r = merger.reviewBatch(user, intParam(params.id), String(body.action ?? ''), body.comment ? String(body.comment) : null);
setImmediate(() => merger.pump());
if (runsWorker) setImmediate(() => merger.pump());
return r;
});
@ -664,20 +655,36 @@ function serveStatic(pathname, res) {
fs.createReadStream(path.join(WEB_DIR, file)).pipe(res);
}
export async function startServer(dataDir) {
/**
* role "web": website, API and git; "worker": verification, merges, remote checks and the Ghidra
* refresh; "all": both in one process. The two share only the database and the repository, so
* either restarts without interrupting the other.
*/
export async function startServer(dataDir, { role = 'all' } = {}) {
if (!['all', 'web', 'worker'].includes(role)) throw new Error(`unknown role ${role}; use web, worker or all`);
const ctx = openServer(dataDir);
ctx.tasks.generate();
ctx.verifier.start();
ctx.merger.start();
const sweeper = setInterval(() => { try { ctx.tasks.sweepExpired(); } catch (e) { console.error(`lease sweep: ${e.message}`); } }, 60_000);
const server = createApp(ctx);
await new Promise((r) => server.listen(ctx.config.listen.port, ctx.config.listen.host, r));
console.log(`clank together coordinator listening on http://${ctx.config.listen.host}:${ctx.config.listen.port} (public url ${ctx.config.publicUrl})`);
ctx.runsWorker = role !== 'web';
const stops = [];
if (role !== 'web') {
ctx.tasks.generate();
ctx.verifier.start();
ctx.merger.start();
ctx.checks.start();
stops.push(() => ctx.verifier.stop(), () => ctx.merger.stop(), () => ctx.checks.stop());
console.log(`clank together worker running (${role})`);
}
let server = null;
if (role !== 'worker') {
const sweeper = setInterval(() => { try { ctx.tasks.sweepExpired(); } catch (e) { console.error(`lease sweep: ${e.message}`); } }, 60_000);
stops.push(() => clearInterval(sweeper));
server = createApp(ctx);
await new Promise((r) => server.listen(ctx.config.listen.port, ctx.config.listen.host, r));
console.log(`clank together coordinator listening on http://${ctx.config.listen.host}:${ctx.config.listen.port} (public url ${ctx.config.publicUrl})`);
}
const shutdown = () => {
clearInterval(sweeper);
ctx.verifier.stop();
ctx.merger.stop();
server.close(() => process.exit(0));
for (const stop of stops) stop();
if (server) server.close(() => process.exit(0));
else process.exit(0);
};
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);

77
src/server/checks.js Normal file
View file

@ -0,0 +1,77 @@
// Remote checks (td check for contributors without a local toolchain) run in the worker process.
// 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 { now } from './db.js';
import { summarizeCheck } from '../shared/check.js';
const POLL_MS = 1000;
const KEEP_MS = 24 * 3600_000;
export class CheckQueue {
constructor({ store, repo, verifier, config, log = console }) {
Object.assign(this, { store, repo, verifier, config, log });
this.running = 0;
}
/** 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);
}
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.timer = setInterval(() => {
try {
this.pump();
} catch (e) {
this.log.error(`check pump: ${e.message}`);
}
}, POLL_MS);
}
stop() {
clearInterval(this.timer);
}
pump() {
const max = this.config.remoteCheckConcurrency ?? 2;
while (this.running < max) {
const job = this.store.tx(() => {
const row = this.store.get(`SELECT * FROM check_jobs WHERE state = 'queued' ORDER BY id LIMIT 1`);
if (row) this.store.run(`UPDATE check_jobs SET state = 'running' WHERE id = ?`, row.id);
return row;
});
if (!job) break;
this.running++;
this.execute(job).finally(() => this.running--);
}
this.store.run(`DELETE FROM check_jobs WHERE created_at < ? AND state != 'running'`, now() - KEEP_MS);
}
async execute(job) {
const files = JSON.parse(job.files_json);
let state = 'done';
let result;
try {
const { check, rec } = await this.verifier.check(files, `check-${job.user_id}-${job.id}`);
result = { gate: 'ok', head: this.repo.head(), ...summarizeCheck(check, rec, new Set(Object.keys(files))) };
} catch (e) {
state = 'failed';
result = { error: e.message };
this.log.error(`remote check ${job.id} failed: ${e.message}`);
}
this.store.run('UPDATE check_jobs SET state = ?, result_json = ?, finished_at = ? WHERE id = ?', state, JSON.stringify(result), now(), job.id);
}
}

View file

@ -19,7 +19,7 @@ const HELP = `td-server - Clank Together coordinator
Usage: td-server <command> [options] (all commands accept --data <dir>, default ./data)
init create data dir, server.json and the canonical repo
start run the coordinator
start [web|worker] run the coordinator (default: both in one process)
user add <name> [--role R] [--password-stdin]
create an active account (contributor|reviewer|maintainer|admin);
without --password-stdin a password is generated and printed
@ -77,7 +77,7 @@ async function main() {
if (!cmd || values.help) return console.log(HELP);
if (cmd === 'init') return init(dataDir);
if (cmd === 'start') return startServer(dataDir);
if (cmd === 'start') return startServer(dataDir, { role: sub ?? 'all' });
if (cmd === 'user') {
const accounts = new Accounts(new Store(path.join(dataDir, 'td.sqlite')));

View file

@ -199,6 +199,17 @@ CREATE TABLE IF NOT EXISTS original_code (
PRIMARY KEY (version_id, addr)
);
-- Remote checks, queued by the web process and run by the worker process.
CREATE TABLE IF NOT EXISTS check_jobs (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id),
files_json TEXT NOT NULL,
state TEXT NOT NULL, -- queued, running, done, failed, abandoned
result_json TEXT,
created_at INTEGER NOT NULL,
finished_at INTEGER
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY,
at INTEGER NOT NULL,

View file

@ -344,6 +344,26 @@ test('peer review: three other agents must agree, by reading only', async () =>
ctx.store.run(`UPDATE tasks SET state = 'done' WHERE id = ?`, task.id);
});
test('remote checks are queued in the database and run by the worker', async () => {
const user = await activeUser('rcheck');
const realCheck = ctx.verifier.check;
ctx.verifier.check = async () => ({
check: { annotationErrors: [], units: [{ file: 'src/rc.cpp', compiled: true, lintErrors: [], unannotated: [], bindings: {}, inferred: [], includes: [], errors: [], functions: [] }] },
rec: { conflicts: [], unknown: [], learned: [], learnedAddresses: [] },
});
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);
assert.equal(r.status, 200, JSON.stringify(r.body));
assert.equal(r.body.gate, 'ok');
assert.equal(r.body.units[0].file, 'src/rc.cpp');
assert.equal(ctx.store.get(`SELECT state FROM check_jobs ORDER BY id DESC LIMIT 1`).state, 'done');
} finally {
ctx.checks.stop();
ctx.verifier.check = realCheck;
}
});
test('maintainers can decide a function still in peer review; reviewers cannot', async () => {
await ctx.accounts.createUser('pmaint', 'maintainer', PASSWORD);
await ctx.accounts.createUser('previewer', 'reviewer', PASSWORD);