Skip to content

fix: join contending daemon startups before reuse #3130

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thymikee wants to merge 3 commits into from
Closed

fix: join contending daemon startups before reuse #3130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5ec91d7
fix: join contending daemon startups within one deadline
thymikee Oct 3, 2026
604939b
fix: observe only the current daemon registration owner
thymikee Oct 3, 2026
d2c7a19
test: bound held-claim startup deadline simulation
thymikee Oct 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 10 additions & 7 deletions src/__tests__/test-utils/registered-daemon-fixture.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ const actualCommand = await vi.importActual<typeof import('@agent-device/host-ki
);
const children = new Map<
string,
{ launch: ReturnType<typeof runCmdDetachedMonitored>; startTime: string | null }
Array<{ launch: ReturnType<typeof runCmdDetachedMonitored>; startTime: string | null }>
>();

/** A real registration owner, advertising the caller's HTTP fixture and joining before deletion. */
Expand All @@ -29,10 +29,10 @@ export function registeredDaemonFixtureArgs(
entry,
`import fs from 'node:fs';
import path from 'node:path';
import { tryAcquireDaemonRegistration } from ${JSON.stringify(registrationUrl)};
import { DAEMON_STARTUP_EXIT_CODES, tryAcquireDaemonRegistration } from ${JSON.stringify(registrationUrl)};
const paths = ${JSON.stringify(paths)};
const acquired = await tryAcquireDaemonRegistration(paths);
if (acquired.status !== 'acquired') process.exit(75);
if (acquired.status !== 'acquired') process.exit(DAEMON_STARTUP_EXIT_CODES[acquired.status]);
process.on('SIGTERM', async () => {
const deferred = path.join(paths.baseDir, 'repair-on-shutdown.json');
if (fs.existsSync(deferred)) {
Expand All @@ -43,6 +43,8 @@ process.on('SIGTERM', async () => {
await acquired.owner.finish();
process.exit(0);
});
fs.writeFileSync(path.join(paths.baseDir, 'registration-held'), 'ready');
while (fs.existsSync(path.join(paths.baseDir, 'defer-publication'))) await new Promise(resolve => setTimeout(resolve, 10));
acquired.owner.publish(${JSON.stringify(fields)});
setInterval(() => {}, 1000);
`,
Expand All @@ -60,22 +62,23 @@ export function spawnRegisteredDaemonFixture(
registeredDaemonFixtureArgs(paths, fields),
options,
);
children.set(paths.baseDir, { launch: child, startTime: readProcessStartTime(child.pid) });
const owned = children.get(paths.baseDir) ?? [];
owned.push({ launch: child, startTime: readProcessStartTime(child.pid) });
children.set(paths.baseDir, owned);
return child;
}

export async function finishRegisteredDaemonFixture(stateDir: string): Promise<void> {
const owned = children.get(stateDir);
if (owned) {
for (const owned of children.get(stateDir) ?? []) {
const child = owned.launch;
const termination = await stopDaemonProcess(
{ pid: child.pid, startTime: owned.startTime },
{ mode: 'force', termTimeoutMs: 0, killTimeoutMs: 2_000 },
);
assert.notEqual(termination.status, 'retained', JSON.stringify(termination));
await child.exited;
children.delete(stateDir);
}
children.delete(stateDir);
fs.rmSync(stateDir, { recursive: true, force: true });
}

Expand Down
56 changes: 31 additions & 25 deletions src/daemon-client/__tests__/daemon-client-lifecycle.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import {
currentDaemonCodeSignature,
} from '../../__tests__/test-utils/daemon-http-fixture.ts';
import { AppError } from '@agent-device/kernel/errors';
import { tryAcquireProcessLock, inspectProcessLock } from '@agent-device/host-kit/file';
import { runCmdDetachedMonitored, runCmdSync } from '@agent-device/host-kit/command';
import { shellQuoteIfNeeded } from '@agent-device/kernel/device-shell';
import { readProcessStartTime } from '@agent-device/host-kit/process';
Expand Down Expand Up @@ -208,14 +209,20 @@ async function startHangingHttpDaemonFixture(): Promise<HttpDaemonFixture> {
}

function installSpawnedHttpDaemon(paths: DaemonPaths, httpPort: number): void {
mockSleep.mockImplementation(actualRetry.sleep);
mockRunCmdDetached.mockImplementation((_command, _args, options) => {
assert.equal(options?.env?.AGENT_DEVICE_STATE_DIR, paths.baseDir);
writeDaemonInfo(paths, { httpPort, transport: 'http' });
writeDaemonLock(paths, {
pid: process.pid,
processStartTime: readProcessStartTime(process.pid) ?? undefined,
});
return { pid: process.pid, exited: new Promise(() => {}) };
return spawnRegisteredDaemonFixture(
paths,
{
httpPort,
token: 'local-secret',
version: readVersion(),
codeOrigin: 'checkout',
codeSignature: currentDaemonCodeSignature(),
},
options,
);
});
}

Expand Down Expand Up @@ -357,7 +364,7 @@ test('sendToDaemon retains unknown metadata after a spawn failure', async () =>
assert.equal(results[0]?.removedInfo, false);
assert.equal(attempts, 1);
} finally {
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down Expand Up @@ -402,11 +409,11 @@ test('sendToDaemon reports early daemon exit with log tail and startup paths', a
assert.match(String(thrown.details?.daemonLogTail), /early daemon failure 2/);
assert.equal(attempts, 2);
} finally {
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

test('sendToDaemon removes stale daemon lock before spawning a fresh daemon', async (t) => {
test('daemon acquisition reclaims a proven reused owner before publication', async (t) => {
if (!(await supportsLoopbackBind())) {
t.skip('loopback listeners are not permitted in this environment');
return;
Expand All @@ -416,10 +423,11 @@ test('sendToDaemon removes stale daemon lock before spawning a fresh daemon', as
const paths = resolveDaemonPaths(stateDir);
const daemon = await startHttpDaemonFixture({ via: 'fresh-daemon' });
vi.stubEnv('AGENT_DEVICE_STATE_DIR', stateDir);
writeDaemonLock(paths, {
pid: process.pid,
processStartTime: 'stale-start-time',
const stale = tryAcquireProcessLock({
Comment thread
thymikee marked this conversation as resolved.
lockDirPath: paths.lockPath,
owner: { pid: process.pid, startTime: 'stale-start-time', acquiredAtMs: Date.now() },
});
assert.equal(stale.status, 'acquired');
installSpawnedHttpDaemon(paths, daemon.port);

try {
Expand All @@ -431,18 +439,16 @@ test('sendToDaemon removes stale daemon lock before spawning a fresh daemon', as
meta: { requestId: 'req-stale-lock' },
});

const freshLock = JSON.parse(fs.readFileSync(paths.lockPath, 'utf8')) as {
pid?: number;
processStartTime?: string;
};
const freshLock = inspectProcessLock(paths.lockPath);
assert.deepEqual(response, { ok: true, data: { via: 'fresh-daemon' } });
assert.equal(mockRunCmdDetached.mock.calls.length, 1);
assert.equal(freshLock.pid, process.pid);
assert.notEqual(freshLock.processStartTime, 'stale-start-time');
assert.equal(freshLock.state, 'held');
if (freshLock.state === 'held') assert.notEqual(freshLock.owner.startTime, 'stale-start-time');
assert.deepEqual(daemon.seenPaths, ['GET /health', 'POST /rpc']);
} finally {
if (stale.status === 'acquired') await stale.acquisition.release();
await closeLoopbackServer(daemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down Expand Up @@ -510,7 +516,7 @@ test('sendToDaemon does not reuse reachable daemon metadata with mismatched vers
stderrCapture.restore();
await closeLoopbackServer(staleDaemon.server);
await closeLoopbackServer(freshDaemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
vi.unstubAllEnvs();
}
}
Expand Down Expand Up @@ -553,7 +559,7 @@ test('sendToDaemon prints a takeover notice before replacing an unreachable daem
} finally {
stderrCapture.restore();
await closeLoopbackServer(freshDaemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down Expand Up @@ -594,7 +600,7 @@ test('sendToDaemon replaces socket-only daemon metadata when HTTP transport is r
} finally {
stderrCapture.restore();
await closeLoopbackServer(freshDaemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down Expand Up @@ -697,7 +703,7 @@ test('sendToDaemon falls back from failed socket transport to HTTP using daemon
} finally {
socketFailures.restore();
await closeLoopbackServer(daemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down Expand Up @@ -741,7 +747,7 @@ test('sendToDaemon does not replay over HTTP after the socket request is written
} finally {
socket.restore();
await closeLoopbackServer(daemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down Expand Up @@ -1275,7 +1281,7 @@ test('issue #1384: sendToDaemon does not stop a client-started daemon at an expl
assert.equal(fs.existsSync(paths.lockPath), true);
} finally {
await closeLoopbackServer(daemon.server);
fs.rmSync(stateDir, { recursive: true, force: true });
await finishRegisteredDaemonFixture(stateDir);
}
});

Expand Down
Loading
Loading