fix(qmd): let only one concurrent launch start the daemon
An exclusive daemon.lock (stale after 2 minutes) is held from before the health check until the daemon listens; a launch that loses exits 0. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
691ad6a69f
commit
3012e1666e
2 files changed
+47
-5
No files matched your search
@@ -2,10 +2,10 @@
|
|||||||
import { test } from 'node:test';
|
import { test } from 'node:test';
|
||||||
import assert from 'node:assert/strict';
|
import assert from 'node:assert/strict';
|
||||||
import { execFileSync } from 'node:child_process';
|
import { execFileSync } from 'node:child_process';
|
||||||
import { mkdtempSync, rmSync, utimesSync, writeFileSync } from 'node:fs';
|
import { existsSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from 'node:fs';
|
||||||
import { tmpdir } from 'node:os';
|
import { tmpdir } from 'node:os';
|
||||||
import { join, resolve } from 'node:path';
|
import { join, resolve } from 'node:path';
|
||||||
import { createServer, gitFingerprint, ignoreGlobs, keepModelsLoaded } from '../../shared/qmd/qmd-daemon.mjs';
|
import { acquireStartLock, createServer, gitFingerprint, ignoreGlobs, keepModelsLoaded } from '../../shared/qmd/qmd-daemon.mjs';
|
||||||
import { collectionName } from '../../shared/qmd/qmd-lib.mjs';
|
import { collectionName } from '../../shared/qmd/qmd-lib.mjs';
|
||||||
|
|
||||||
const defaultHits = () => [{ displayPath: `${collectionName('/r')}/src/a.ts`, body: 'x\nretry here', bestChunkPos: 2, bestChunk: 'retry here', score: 0.91 }];
|
const defaultHits = () => [{ displayPath: `${collectionName('/r')}/src/a.ts`, body: 'x\nretry here', bestChunkPos: 2, bestChunk: 'retry here', score: 0.91 }];
|
||||||
@@ -201,3 +201,17 @@ test('git ignored paths reach addCollection and are persisted only when they cha
|
|||||||
assert.deepEqual(persisted[1], [collectionName('/r'), ['dist/**', 'tmp/**']]);
|
assert.deepEqual(persisted[1], [collectionName('/r'), ['dist/**', 'tmp/**']]);
|
||||||
assert.equal(s.store.calls.filter((c) => c[0] === 'update').length, 3);
|
assert.equal(s.store.calls.filter((c) => c[0] === 'update').length, 3);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('the start lock admits one launcher and is taken over after 2 minutes', (t) => {
|
||||||
|
const dir = mkdtempSync(join(tmpdir(), 'qmd-lock-'));
|
||||||
|
t.after(() => rmSync(dir, { recursive: true, force: true }));
|
||||||
|
const lock = join(dir, 'daemon.lock');
|
||||||
|
const release = acquireStartLock(lock);
|
||||||
|
assert.equal(typeof release, 'function');
|
||||||
|
assert.equal(acquireStartLock(lock), null, 'a concurrent launch loses');
|
||||||
|
release();
|
||||||
|
assert.equal(existsSync(lock), false);
|
||||||
|
assert.equal(typeof acquireStartLock(lock), 'function');
|
||||||
|
assert.equal(acquireStartLock(lock, () => Date.now() + 60_000), null, 'a fresh lock holds');
|
||||||
|
assert.equal(typeof acquireStartLock(lock, () => Date.now() + 121_000), 'function', 'a stale lock is taken over');
|
||||||
|
});
|
||||||
@@ -1,9 +1,9 @@
|
|||||||
// shared/qmd/qmd-daemon.mjs — keeps QMD's models loaded for fast searches; exits when idle.
|
// shared/qmd/qmd-daemon.mjs — keeps QMD's models loaded for fast searches; exits when idle.
|
||||||
import { execFileSync } from 'node:child_process';
|
import { execFileSync } from 'node:child_process';
|
||||||
import { randomBytes, timingSafeEqual } from 'node:crypto';
|
import { randomBytes, timingSafeEqual } from 'node:crypto';
|
||||||
import { existsSync, readFileSync, statSync } from 'node:fs';
|
import { closeSync, existsSync, mkdirSync, openSync, readFileSync, rmSync, statSync } from 'node:fs';
|
||||||
import { createServer as createHttpServer } from 'node:http';
|
import { createServer as createHttpServer } from 'node:http';
|
||||||
import { join, resolve } from 'node:path';
|
import { dirname, join, resolve } from 'node:path';
|
||||||
import { pathToFileURL } from 'node:url';
|
import { pathToFileURL } from 'node:url';
|
||||||
import { setCollectionIgnore } from './qmd-config.mjs';
|
import { setCollectionIgnore } from './qmd-config.mjs';
|
||||||
import { MASK, collectionName, idleMinutes, importQmd, qmdConfigPath, qmdDbPath, readJson, runQmdCli, stateDir, writePrivateJson, writeTextAtomic } from './qmd-lib.mjs';
|
import { MASK, collectionName, idleMinutes, importQmd, qmdConfigPath, qmdDbPath, readJson, runQmdCli, stateDir, writePrivateJson, writeTextAtomic } from './qmd-lib.mjs';
|
||||||
@@ -42,6 +42,29 @@ export function gitIgnored(repo, escape) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Exclusive start lock: returns a release function, or null while another launch holds a lock younger than 2 minutes.
|
||||||
|
// ponytail: two launchers that find the same stale lock can both take it over; the daemon.json health check covers the rest.
|
||||||
|
export function acquireStartLock(path, now = Date.now) {
|
||||||
|
mkdirSync(dirname(path), { recursive: true });
|
||||||
|
for (let attempt = 0; attempt < 2; attempt++) {
|
||||||
|
try {
|
||||||
|
closeSync(openSync(path, 'wx'));
|
||||||
|
let held = true;
|
||||||
|
return () => {
|
||||||
|
if (held) rmSync(path, { force: true });
|
||||||
|
held = false;
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
if (error.code !== 'EEXIST') throw error;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
if (now() - statSync(path).mtimeMs < 120_000) return null;
|
||||||
|
rmSync(path, { force: true });
|
||||||
|
} catch { /* the holder released it meanwhile: try again */ }
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
// QMD 2.8.3 createStore unloads models after 5 idle minutes (store.internal.llm); the daemon's own idle exit replaces that.
|
// QMD 2.8.3 createStore unloads models after 5 idle minutes (store.internal.llm); the daemon's own idle exit replaces that.
|
||||||
export function keepModelsLoaded(store) {
|
export function keepModelsLoaded(store) {
|
||||||
const llm = store.internal?.llm;
|
const llm = store.internal?.llm;
|
||||||
@@ -165,11 +188,15 @@ export function createServer({ store, token, idleMs, now = Date.now, onIdle, ext
|
|||||||
async function main() {
|
async function main() {
|
||||||
const dir = stateDir();
|
const dir = stateDir();
|
||||||
const daemonFile = join(dir, 'daemon.json');
|
const daemonFile = join(dir, 'daemon.json');
|
||||||
|
// The lock comes before the health check, so a launch that waited for another one sees its daemon.json.
|
||||||
|
const releaseLock = acquireStartLock(join(dir, 'daemon.lock'));
|
||||||
|
if (!releaseLock) return; // another launch is starting the daemon
|
||||||
|
process.on('exit', releaseLock);
|
||||||
const current = readJson(daemonFile, null);
|
const current = readJson(daemonFile, null);
|
||||||
if (current) {
|
if (current) {
|
||||||
try {
|
try {
|
||||||
const health = await fetch(`http://127.0.0.1:${current.port}/health`, { headers: { 'x-qmd-token': current.token }, signal: AbortSignal.timeout(2000) });
|
const health = await fetch(`http://127.0.0.1:${current.port}/health`, { headers: { 'x-qmd-token': current.token }, signal: AbortSignal.timeout(2000) });
|
||||||
if (health.ok && (await health.json()).service === 'ai-config-qmd') return;
|
if (health.ok && (await health.json()).service === 'ai-config-qmd') return releaseLock();
|
||||||
} catch { /* stale file: start a new daemon */ }
|
} catch { /* stale file: start a new daemon */ }
|
||||||
}
|
}
|
||||||
const { createStore, extractSnippet } = await importQmd();
|
const { createStore, extractSnippet } = await importQmd();
|
||||||
@@ -214,6 +241,7 @@ async function main() {
|
|||||||
});
|
});
|
||||||
daemon.server.listen(0, '127.0.0.1', () => {
|
daemon.server.listen(0, '127.0.0.1', () => {
|
||||||
writePrivateJson(daemonFile, { port: daemon.server.address().port, token, pid: process.pid });
|
writePrivateJson(daemonFile, { port: daemon.server.address().port, token, pid: process.pid });
|
||||||
|
releaseLock();
|
||||||
});
|
});
|
||||||
setInterval(daemon.checkIdle, 60_000).unref();
|
setInterval(daemon.checkIdle, 60_000).unref();
|
||||||
process.on('SIGTERM', stop);
|
process.on('SIGTERM', stop);
|
||||||
|
|||||||
Reference in new issue
Block a user