fix(qmd): handle a lost daemon start race and stale daemon tokens
startDaemon accepts the daemon of another launch when its own child exits 0. A 403 from the daemon marks daemon.json as stale: the wrapper deletes it and starts a new daemon once. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
3012e1666e
commit
06c4848336
2 files changed
+51
-8
No files matched your search
@@ -1,7 +1,10 @@
|
|||||||
import { test } from 'node:test';
|
import { test } from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
import { EventEmitter } from 'node:events';
|
import { EventEmitter } from 'node:events';
|
||||||
import { runSearch, startDaemon, toCliResults } from '../../shared/qmd/qmd-search.mjs';
|
import { mkdtempSync, rmSync, writeFileSync, existsSync } from 'node:fs';
|
||||||
|
import { tmpdir } from 'node:os';
|
||||||
|
import { join } from 'node:path';
|
||||||
|
import { callDaemon, runSearch, startDaemon, toCliResults } from '../../shared/qmd/qmd-search.mjs';
|
||||||
import { collectionName } from '../../shared/qmd/qmd-lib.mjs';
|
import { collectionName } from '../../shared/qmd/qmd-lib.mjs';
|
||||||
|
|
||||||
const hit = [{ path: 'src/a.ts', line: 3, score: 0.9, snippet: 's' }];
|
const hit = [{ path: 'src/a.ts', line: 3, score: 0.9, snippet: 's' }];
|
||||||
@@ -76,3 +79,33 @@ test('startDaemon rejects when the child cannot be spawned', async () => {
|
|||||||
return child;
|
return child;
|
||||||
} }), /spawn failed/);
|
} }), /spawn failed/);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
function tempDaemonFile(t, value) {
|
||||||
|
const dir = mkdtempSync(join(tmpdir(), 'qmd-search-'));
|
||||||
|
t.after(() => rmSync(dir, { recursive: true, force: true }));
|
||||||
|
const file = join(dir, 'daemon.json');
|
||||||
|
if (value) writeFileSync(file, JSON.stringify(value));
|
||||||
|
return file;
|
||||||
|
}
|
||||||
|
|
||||||
|
test('startDaemon resolves when its child loses the start race to a running daemon', async (t) => {
|
||||||
|
const file = tempDaemonFile(t, { port: 4000, token: 'other', pid: 999 });
|
||||||
|
await startDaemon({ file, spawnChild: () => {
|
||||||
|
const child = new EventEmitter(); child.unref = () => {}; child.pid = 1;
|
||||||
|
setImmediate(() => child.emit('exit', 0));
|
||||||
|
return child;
|
||||||
|
} });
|
||||||
|
});
|
||||||
|
|
||||||
|
test('a 403 means a stale token: daemon.json is deleted and one new daemon is started', async (t) => {
|
||||||
|
const file = tempDaemonFile(t, { port: 4000, token: 'stale', pid: 999 });
|
||||||
|
const fetchImpl = async (url, options) => (options.headers['x-qmd-token'] === 'fresh'
|
||||||
|
? { ok: true, status: 200, json: async () => ({ results: hit }) } : { ok: false, status: 403 });
|
||||||
|
let started = 0;
|
||||||
|
const results = await runSearch({ root: '/r', query: 'q', topK: 5,
|
||||||
|
daemon: (path, body) => callDaemon(path, body, { file, fetchImpl }),
|
||||||
|
startDaemon: async () => { started += 1; assert.equal(existsSync(file), false); writeFileSync(file, JSON.stringify({ port: 4001, token: 'fresh', pid: 998 })); },
|
||||||
|
cli: async () => assert.fail('must not fall back') });
|
||||||
|
assert.deepEqual(results, hit);
|
||||||
|
assert.equal(started, 1);
|
||||||
|
});
|
||||||
@@ -1,5 +1,6 @@
|
|||||||
// shared/qmd/qmd-search.mjs — search the current repository through the QMD daemon, with a CLI fallback.
|
// shared/qmd/qmd-search.mjs — search the current repository through the QMD daemon, with a CLI fallback.
|
||||||
import { execFileSync, spawn } from 'node:child_process';
|
import { execFileSync, spawn } from 'node:child_process';
|
||||||
|
import { rmSync } from 'node:fs';
|
||||||
import { dirname, join, resolve } from 'node:path';
|
import { dirname, join, resolve } from 'node:path';
|
||||||
import { fileURLToPath, pathToFileURL } from 'node:url';
|
import { fileURLToPath, pathToFileURL } from 'node:url';
|
||||||
import { parseArgs } from 'node:util';
|
import { parseArgs } from 'node:util';
|
||||||
@@ -18,12 +19,12 @@ export function toCliResults(items, collection) {
|
|||||||
|
|
||||||
const notRunning = (message, cause) => Object.assign(new Error(message, { cause }), { notRunning: true });
|
const notRunning = (message, cause) => Object.assign(new Error(message, { cause }), { notRunning: true });
|
||||||
|
|
||||||
async function callDaemon(path, body) {
|
export async function callDaemon(path, body, { file = daemonFile(), fetchImpl = fetch } = {}) {
|
||||||
const info = readJson(daemonFile(), null);
|
const info = readJson(file, null);
|
||||||
if (!info?.port || !info?.token) throw notRunning('daemon not running');
|
if (!info?.port || !info?.token) throw notRunning('daemon not running');
|
||||||
let response;
|
let response;
|
||||||
try {
|
try {
|
||||||
response = await fetch(`http://127.0.0.1:${info.port}${path}`, {
|
response = await fetchImpl(`http://127.0.0.1:${info.port}${path}`, {
|
||||||
method: 'POST', headers: { 'x-qmd-token': info.token, 'content-type': 'application/json' },
|
method: 'POST', headers: { 'x-qmd-token': info.token, 'content-type': 'application/json' },
|
||||||
body: JSON.stringify(body), signal: AbortSignal.timeout(600_000) });
|
body: JSON.stringify(body), signal: AbortSignal.timeout(600_000) });
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -31,23 +32,32 @@ async function callDaemon(path, body) {
|
|||||||
if (error instanceof TypeError) throw notRunning('daemon not reachable', error);
|
if (error instanceof TypeError) throw notRunning('daemon not reachable', error);
|
||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
|
if (response.status === 403) {
|
||||||
|
// Whatever listens there does not know this token, so daemon.json is stale.
|
||||||
|
rmSync(file, { force: true });
|
||||||
|
throw notRunning('daemon rejected the token');
|
||||||
|
}
|
||||||
if (!response.ok) throw new Error(`daemon answered ${response.status}`);
|
if (!response.ok) throw new Error(`daemon answered ${response.status}`);
|
||||||
return response.json();
|
return response.json();
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function startDaemon({ spawnChild = spawn } = {}) {
|
export async function startDaemon({ spawnChild = spawn, file = daemonFile() } = {}) {
|
||||||
const device = readJson(setupStatePath(), {}).device || 'cpu';
|
const device = readJson(setupStatePath(), {}).device || 'cpu';
|
||||||
const child = spawnChild(process.execPath, [join(here, 'qmd-daemon.mjs')], { detached: true, stdio: 'ignore', windowsHide: true, env: deviceEnv(device) });
|
const child = spawnChild(process.execPath, [join(here, 'qmd-daemon.mjs')], { detached: true, stdio: 'ignore', windowsHide: true, env: deviceEnv(device) });
|
||||||
let failure = null;
|
let failure = null;
|
||||||
|
let lostRace = false; // exit code 0: another launch holds the start lock or a daemon already runs
|
||||||
child.once('error', (error) => { failure = error; });
|
child.once('error', (error) => { failure = error; });
|
||||||
child.once('exit', (code) => { failure ||= new Error(`daemon exited with code ${code}`); });
|
child.once('exit', (code) => {
|
||||||
|
if (code === 0) lostRace = true;
|
||||||
|
else failure ||= new Error(`daemon exited with code ${code}`);
|
||||||
|
});
|
||||||
child.unref();
|
child.unref();
|
||||||
const deadline = Date.now() + 120_000;
|
const deadline = Date.now() + 120_000;
|
||||||
while (Date.now() < deadline) {
|
while (Date.now() < deadline) {
|
||||||
await new Promise((done) => setTimeout(done, 500));
|
await new Promise((done) => setTimeout(done, 500));
|
||||||
if (failure) throw failure;
|
if (failure) throw failure;
|
||||||
const info = readJson(daemonFile(), null);
|
const info = readJson(file, null);
|
||||||
if (info?.port && info?.token && info.pid === child.pid) return;
|
if (info?.port && info?.token && (lostRace || info.pid === child.pid)) return;
|
||||||
}
|
}
|
||||||
throw new Error('daemon did not start within 120 s');
|
throw new Error('daemon did not start within 120 s');
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user