commit 0e88af4ad9f9bad7da64710469749dd53dbf05cd
parent eb19ce1fda4f1cd3330dc53566ef88bf47e67b03
Author: finwo <finwo@pm.me>
Date: Tue, 29 Sep 2026 11:28:37 +0200
Stop a worker holding tasks the conductor thinks are running
Diffstat:
3 files changed, 181 insertions(+), 17 deletions(-)
diff --git a/src/worker/agent.js b/src/worker/agent.js
@@ -97,6 +97,19 @@ export async function main(argv = process.argv.slice(2)) {
};
for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => shutdown(signal));
+ async function handBack(task) {
+ try {
+ await client.done(task.endpoints.done, {
+ success: false,
+ state: 'failed',
+ exitCode: null,
+ error: 'worker shut down before the task started',
+ });
+ } catch (e) {
+ logger.warn(`could not hand back ${task.name}: ${e.message}`);
+ }
+ }
+
// One loop per concurrency slot. Each polls independently, so a slow task
// in one slot does not stall the others.
async function slot(index) {
@@ -122,6 +135,12 @@ export async function main(argv = process.argv.slice(2)) {
continue;
}
+ // Signal landed mid-claim; starting it would orphan it until the reaper.
+ if (stopping) {
+ await handBack(task);
+ return;
+ }
+
running.add(task.id);
logger.info(`[${index}] ${task.name} (${task.image})`);
const started = Date.now();
diff --git a/src/worker/task.js b/src/worker/task.js
@@ -110,6 +110,22 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
let failure = null;
let flushScanners = null;
+ // Before setup: the reaper clock starts at claim, and a pull can outrun it.
+ const beatEvery = Math.max(5, Number(task.heartbeat_interval) || 30) * 1000;
+ heartbeat = setInterval(async () => {
+ try {
+ const result = await client.heartbeat(task.endpoints.heartbeat);
+ if (result.cancelled && !cancelled) {
+ cancelled = true;
+ log.note('[conductor] cancelled, stopping the container');
+ if (containerCreated) await runtime.kill(taskContainer);
+ }
+ } catch (e) {
+ logger.warn?.(`heartbeat failed for ${task.name}: ${e.message}`);
+ }
+ }, beatEvery);
+ heartbeat.unref?.();
+
try {
scratch = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-task-'));
@@ -165,6 +181,8 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
await fs.writeFile(scriptPath, buildScript(task.script), { mode: 0o755 });
await runtime.copyIn(taskContainer, scriptPath, SCRIPT_PATH);
+ if (cancelled) throw new Error('cancelled during setup');
+
const { child } = runtime.startCreated(taskContainer);
// One per stream; the two interleave.
@@ -175,22 +193,6 @@ export async function runTask({ cfg, client, runtime, task, logger = console })
for (const scanner of scanners) log.write(scanner.end());
};
- // Report in, and find out whether the run was cancelled underneath us.
- const beatEvery = Math.max(5, Number(task.heartbeat_interval) || 30) * 1000;
- heartbeat = setInterval(async () => {
- try {
- const result = await client.heartbeat(task.endpoints.heartbeat);
- if (result.cancelled && !cancelled) {
- cancelled = true;
- log.note('[conductor] cancelled, stopping the container');
- await runtime.kill(taskContainer);
- }
- } catch (e) {
- logger.warn?.(`heartbeat failed for ${task.name}: ${e.message}`);
- }
- }, beatEvery);
- heartbeat.unref?.();
-
const limit = (Number(task.timeout) || cfg.default_timeout) * 1000;
timeoutTimer = setTimeout(async () => {
timedOut = true;
diff --git a/test/worker.test.js b/test/worker.test.js
@@ -9,11 +9,15 @@ import os from 'node:os';
import path from 'node:path';
import { spawn } from 'node:child_process';
import { fileURLToPath } from 'node:url';
+import { EventEmitter } from 'node:events';
+import { PassThrough, Readable } from 'node:stream';
+import { gzipSync } from 'node:zlib';
+import { createServer } from 'node:http';
import { loadWorkerConfig, featureNames } from '../src/worker/config.js';
import { buildScript, shellQuote } from '../src/worker/script.js';
import { containerName, createRuntime } from '../src/worker/docker.js';
import { createLogStream } from '../src/worker/logstream.js';
-import { resolveFeatures, patternPrefix, copyOutPaths } from '../src/worker/task.js';
+import { resolveFeatures, patternPrefix, copyOutPaths, runTask } from '../src/worker/task.js';
import { collectArtifacts, shouldCollect } from '../src/worker/artifacts.js';
async function tempDir(prefix = 'conductor-worker-') {
@@ -460,3 +464,142 @@ test('a failing conductor does not break the task', async () => {
await log.close();
assert.match(log.error.message, /unreachable/);
});
+
+const quiet = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
+
+function setupTask(overrides = {}) {
+ return {
+ id: 'task-setup',
+ name: 'build',
+ image: 'busybox',
+ sha: 'a'.repeat(40),
+ attempt: 1,
+ script: 'true',
+ env: {},
+ masked: [],
+ timeout: 600,
+ heartbeat_interval: 5,
+ workdir: '/workspace',
+ endpoints: {
+ log: 'http://x/log',
+ source: 'http://x/source.tar.gz',
+ heartbeat: 'http://x/heartbeat',
+ done: 'http://x/done',
+ artifact: 'http://x/artifacts',
+ },
+ ...overrides,
+ };
+}
+
+function exitingChild(code = 0) {
+ const child = new EventEmitter();
+ child.stdout = new PassThrough();
+ child.stderr = new PassThrough();
+ setImmediate(() => {
+ child.stdout.end();
+ child.stderr.end();
+ child.emit('close', code);
+ });
+ return child;
+}
+
+test('a heartbeat goes out while a task is still being set up', async () => {
+ const beats = [];
+ const client = {
+ async appendLog(url, chunk, offset) { return { size: offset + chunk.length }; },
+ async heartbeat() { beats.push(Date.now()); return { cancelled: false }; },
+ async source() { return Readable.from([gzipSync(Buffer.alloc(0))]); },
+ async done() { return {}; },
+ };
+
+ // Stands in for an image pull that outlasts the conductor's reaper clock.
+ const runtime = {
+ async createNetwork() {},
+ async removeNetwork() {},
+ async create() {
+ const until = Date.now() + 8000;
+ while (Date.now() < until && beats.length === 0) {
+ await new Promise((r) => { setTimeout(r, 25); });
+ }
+ },
+ async copyStreamIn(name, stream) {
+ stream.resume();
+ await new Promise((r) => { stream.on('end', r); stream.on('error', r); });
+ },
+ async copyIn() {},
+ async copyOut() {},
+ async kill() {},
+ async remove() {},
+ startCreated: () => ({ child: exitingChild(0), args: [] }),
+ };
+
+ const result = await runTask({
+ cfg: { name: 'w', shell: 'sh', default_timeout: 600, features: {} },
+ client,
+ runtime,
+ task: setupTask(),
+ logger: quiet,
+ });
+
+ assert.ok(beats.length > 0, 'the conductor heard nothing while the image was pulling');
+ assert.equal(result.success, true);
+});
+
+test('a task claimed as the worker stops is handed back, not started', async () => {
+ const events = [];
+ let task = null;
+ const server = createServer((req, res) => {
+ events.push(req.url);
+ res.setHeader('content-type', 'application/json');
+ if (req.url === '/api/v1/tasks/claim') {
+ // Slow enough that the signal lands while the claim is in flight.
+ setTimeout(() => res.end(JSON.stringify({ ok: true, data: { task } })), 1200);
+ return;
+ }
+ res.end(JSON.stringify({ ok: true, data: {} }));
+ });
+
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const port = server.address().port;
+ const at = (p) => `http://127.0.0.1:${port}${p}`;
+ task = setupTask({
+ endpoints: {
+ log: at('/log'),
+ source: at('/source.tar.gz'),
+ heartbeat: at('/heartbeat'),
+ done: at('/done'),
+ artifact: at('/artifacts'),
+ },
+ });
+
+ const dir = await tempDir();
+ const config = path.join(dir, 'worker.json');
+ await fs.writeFile(config, JSON.stringify({
+ conductor_url: `http://127.0.0.1:${port}`,
+ token: 'irrelevant',
+ poll_interval: 60,
+ docker: 'true',
+ }));
+
+ const agent = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '../src/worker/agent.js');
+ const child = spawn(process.execPath, [agent, '--config', config], { stdio: 'ignore' });
+
+ await new Promise((r) => { setTimeout(r, 700); });
+ child.kill('SIGTERM');
+
+ const code = await new Promise((resolve) => {
+ const timer = setTimeout(() => resolve('timed out'), 10000);
+ child.on('exit', (c) => { clearTimeout(timer); resolve(c); });
+ });
+ child.kill('SIGKILL');
+ server.close();
+
+ assert.equal(code, 0, 'should still exit cleanly');
+ assert.ok(events.includes('/done'), `expected the task to be handed back, saw ${events.join(', ')}`);
+ assert.ok(
+ !events.includes('/source.tar.gz'),
+ `the task was started despite the signal, saw ${events.join(', ')}`
+ );
+
+ await fs.rm(dir, { recursive: true, force: true });
+});
