Skip to content

Commit 36373e6

Browse files
thymikeethymikee
committed
fix(daemon): give each startup its own marker file and retire it at its owner
The integration leak oracle failed because every daemon appended its startup phase markers to one shared `daemon-startup-phases.jsonl` beside the registration and nothing retired that file, so any completed launch left state-dir residue the strict oracle names as a stray entry. Ownership now follows the file name: each launch writes only to `daemon-startup-phases-<pid>.jsonl` through `resolveDaemonStartupPhasesPath(stateDir, pid)`. The writing daemon removes its file once it has registered; the launching client removes it once that process's exit is joined (in `retireStartupAttempt`, after the exit is observed, never before); and each new registration-lock owner sweeps the files of provably dead pids next to the shutdown-report removal. A live pid's file is always kept, so a concurrent or replaced launch can neither add to, hide behind, nor delete another launch's evidence, and the client no longer needs `spawnedAtMs` to attribute markers to its own launch. Along the protocol boundary: - `recordDaemonStartupMarker` takes an optional `reportFailure` callback so producers surface a refused append on stderr (the daemon log tail the waiting client already ships as failure evidence) instead of swallowing it; the append stays non-fatal and a crash-truncated final line is separated so it cannot glue two records together. - The phase list gains real completed boundaries, `composition-loaded` and `recovery-finished`, and no heartbeats, so silence still means a stalled startup and progress extends the stall window exactly as far as the work took. - The launch-exit join drops its redundant `.catch(() => {})`: the `runCmdDetachedMonitored` owner resolves the promise on error and exit, so it never rejects. - Startup code splits out of `daemon-client-lifecycle.ts` into `daemon-client-startup.ts` (budget, wait loop, marker refresh, hints), keeping both modules under the 1000-line tripwire; importers attach and ensure point at the owning module directly. Regression coverage: the budget suite now pins per-launch attribution, including a fresh daemon producing its own markers while an earlier launch's leftover file is swept at the new registration, and a new protocol suite pins the per-pid paths, the dead-pid sweep, removal idempotence, and the write-failure seam. Refs #3782, #3791
1 parent 694a093 commit 36373e6

10 files changed

Lines changed: 1162 additions & 776 deletions

‎packages/daemon-contracts/src/daemon-registration-owner-startup-markers.test.ts‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
import { appendFile, readFile, writeFile } from 'node:fs/promises';
2+
import { spawnSync } from 'node:child_process';
3+
import { existsSync, mkdirSync, readFileSync } from 'node:fs';
4+
import path from 'node:path';
5+
import { describe, expect, it } from 'vitest';
6+
import {
7+
readDaemonStartupPhaseMarkers,
8+
recordDaemonStartupMarker,
9+
removeDaemonStartupPhasesFile,
10+
resolveDaemonStartupPhasesPath,
11+
tryAcquireDaemonRegistration,
12+
} from '@agent-device/daemon-contracts/daemon-registration-owner';
13+
import { resolveDaemonPaths } from '@agent-device/daemon-contracts/daemon-resolution';
14+
import { mkdtempForTest } from '@agent-device/test-kit';
15+
16+
describe('per-launch startup marker files', () => {
17+
it('scopes marker files to the writing pid so a fresh launch never reads an older launch phases', async () => {
18+
const stateDir = await mkdtempForTest('agent-device-markers-');
19+
const first = resolveDaemonStartupPhasesPath(stateDir, 1111);
20+
const second = resolveDaemonStartupPhasesPath(stateDir, 2222);
21+
22+
recordDaemonStartupMarker(first, 'lock-acquired');
23+
24+
expect(second).not.toBe(first);
25+
expect(existsSync(second)).toBe(false);
26+
27+
recordDaemonStartupMarker(second, 'modules-loaded');
28+
const firstPhases = readDaemonStartupPhaseMarkers(readFileSync(first, 'utf8'));
29+
const secondPhases = readDaemonStartupPhaseMarkers(readFileSync(second, 'utf8'));
30+
expect(Object.keys(firstPhases)).toEqual(['lock-acquired']);
31+
expect(Object.keys(secondPhases)).toEqual(['modules-loaded']);
32+
});
33+
34+
it('keeps the earliest record when a replayed phase is appended after a later one', () => {
35+
const text = [
36+
'{"data":{"startupPhase":"listening","timestampMs":500}}',
37+
'{"data":{"startupPhase":"lock-acquired","timestampMs":20}}',
38+
'{"data":{"startupPhase":"lock-acquired","timestampMs":35}}',
39+
].join('\n');
40+
const phases = readDaemonStartupPhaseMarkers(text);
41+
expect(phases['lock-acquired']).toBe(20);
42+
expect(phases.listening).toBe(500);
43+
});
44+
45+
it('ignores unknown phases and unparsable lines so the protocol stays additive', () => {
46+
const text = [
47+
'{"data":{"startupPhase":"heartbeat","timestampMs":10}}',
48+
'{"data":{"startupPhase":"registered","timestampMs":"7"}}',
49+
'not json',
50+
'{"data":{"startupPhase":"listening","timestampMs":30}}',
51+
].join('\n');
52+
expect(readDaemonStartupPhaseMarkers(text)).toEqual({ listening: 30 });
53+
});
54+
});
55+
56+
describe('registration-time sweep', () => {
57+
it('retires marker files of provably-dead pids and keeps files of live pids', async () => {
58+
const paths = resolveDaemonPaths(await mkdtempForTest('agent-device-markers-'));
59+
const deadPid = spawnDeadPid();
60+
const own = resolveDaemonStartupPhasesPath(paths.baseDir, process.pid);
61+
const dead = resolveDaemonStartupPhasesPath(paths.baseDir, deadPid);
62+
const legacy = path.join(paths.baseDir, 'daemon-startup-phases.jsonl');
63+
await Promise.all([
64+
writeFile(own, 'x\n', 'utf8'),
65+
writeFile(dead, 'x\n', 'utf8'),
66+
writeFile(legacy, 'x\n', 'utf8'),
67+
]);
68+
69+
const attempt = await tryAcquireDaemonRegistration(paths);
70+
expect(attempt.status).toBe('acquired');
71+
if (attempt.status !== 'acquired') throw new Error('registration refused');
72+
73+
expect(existsSync(dead)).toBe(false);
74+
expect(existsSync(own)).toBe(true);
75+
// The sweep retires only its own kind: an unknown entry stays as evidence.
76+
expect(existsSync(legacy)).toBe(true);
77+
await attempt.owner.finish();
78+
});
79+
80+
it('retires files whose pid is unparseable', async () => {
81+
const paths = resolveDaemonPaths(await mkdtempForTest('agent-device-markers-'));
82+
const stale = path.join(paths.baseDir, 'daemon-startup-phases-.jsonl');
83+
await writeFile(stale, 'x\n', 'utf8');
84+
85+
const attempt = await tryAcquireDaemonRegistration(paths);
86+
expect(attempt.status).toBe('acquired');
87+
if (attempt.status !== 'acquired') throw new Error('registration refused');
88+
89+
expect(existsSync(stale)).toBe(false);
90+
await attempt.owner.finish();
91+
});
92+
});
93+
94+
describe('removeDaemonStartupPhasesFile', () => {
95+
it('removes a marker file and tolerates a file already gone', async () => {
96+
const stateDir = await mkdtempForTest('agent-device-markers-');
97+
const file = resolveDaemonStartupPhasesPath(stateDir, 4242);
98+
await writeFile(file, 'x\n', 'utf8');
99+
100+
removeDaemonStartupPhasesFile(file);
101+
expect(existsSync(file)).toBe(false);
102+
103+
expect(() => removeDaemonStartupPhasesFile(file)).not.toThrow();
104+
});
105+
});
106+
107+
describe('recordDaemonStartupMarker failure seam', () => {
108+
it('reports a write failure through the callback instead of throwing, keeping startup running', async () => {
109+
const stateDir = await mkdtempForTest('agent-device-markers-');
110+
// A directory in place of the marker file makes every append fail.
111+
const markerPath = resolveDaemonStartupPhasesPath(stateDir, 777);
112+
mkdirSync(markerPath, { recursive: true });
113+
114+
const reported: unknown[] = [];
115+
expect(() =>
116+
recordDaemonStartupMarker(markerPath, 'modules-loaded', (error) => {
117+
reported.push(error);
118+
}),
119+
).not.toThrow();
120+
121+
expect(reported).toHaveLength(1);
122+
expect(reported[0]).toHaveProperty('message');
123+
});
124+
125+
it('keeps appending valid JSONL when an incomplete line precedes a crash-replay append', async () => {
126+
const stateDir = await mkdtempForTest('agent-device-markers-');
127+
const markerPath = resolveDaemonStartupPhasesPath(stateDir, 999);
128+
await appendFile(markerPath, '{"data":{"startupPhase":"modules-lo', 'utf8');
129+
130+
recordDaemonStartupMarker(markerPath, 'composition-loaded');
131+
132+
const tail = await readFile(markerPath, 'utf8');
133+
expect(Object.keys(readDaemonStartupPhaseMarkers(tail))).toEqual(['composition-loaded']);
134+
});
135+
});
136+
137+
function spawnDeadPid(): number {
138+
// A process that has already exited proves the liveness check rejects it,
139+
// the same way the daemon-process takeover guards prove it.
140+
const child = spawnSync(process.execPath, ['-e', ''], { encoding: 'utf8' });
141+
if (child.error) throw child.error;
142+
const pid = child.pid;
143+
if (pid === undefined) throw new Error('spawnSync returned no pid');
144+
return pid;
145+
}

‎packages/daemon-contracts/src/daemon-registration-owner.ts‎

Lines changed: 93 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import {
1010
type DaemonTerminationResult,
1111
} from './daemon-process.ts';
1212
import {
13+
isProcessAlive,
1314
readCurrentOwnerIdentity,
1415
readProcessStartTime,
1516
type OwnerIdentity,
@@ -48,16 +49,24 @@ import {
4849
export const DAEMON_STARTUP_EXIT_CODES = Object.freeze({ busy: 75, unproven: 78 });
4950

5051
/**
51-
* Startup phase markers (#3782): the record a daemon leaves beside its registration of how far its
52+
* Startup phase markers (#3782): the record a daemon writes beside its registration of how far its
5253
* startup got. The client waiting for registration reads this record to tell "slow but progressing"
53-
* from "stuck". Each marker is one JSONL line appended to a dedicated state-directory file, so no
54-
* `daemon.log` reader has to learn a new line shape and `daemon.log` keeps its existing ownership.
55-
* The record is append-only across startups: a waiting client scopes it to its own launch by the
56-
* marker timestamps, never by deleting what an earlier startup left.
54+
* from "stuck". One launch owns exactly one file, named for the pid that wrote it: an earlier or a
55+
* concurrent launch cannot add to, replace, or hide another launch's markers, and neither the file
56+
* nor a waiting client ever needs a timestamp to decide which startup a marker belongs to. A phase
57+
* is recorded when the work it names has finished; the record never carries a heartbeat, so silence
58+
* stays what a stalled startup looks like.
59+
*
60+
* Ownership, so the record never becomes residue: the writer removes the file once its startup
61+
* completes; the client that launched a process removes it once that process's exit is joined; and
62+
* each new registration owner sweeps the files of provably dead pids when it takes the lock. Each
63+
* marker is one JSONL line appended to the launch's own file.
5764
*/
5865
const DAEMON_STARTUP_PHASES = Object.freeze([
5966
'modules-loaded',
67+
'composition-loaded',
6068
'lock-acquired',
69+
'recovery-finished',
6170
'listening',
6271
'registered',
6372
] as const);
@@ -66,35 +75,96 @@ export type DaemonStartupPhase = (typeof DAEMON_STARTUP_PHASES)[number];
6675

6776
const DAEMON_STARTUP_PHASE_SET: ReadonlySet<string> = new Set<string>(DAEMON_STARTUP_PHASES);
6877

69-
const DAEMON_STARTUP_PHASES_FILE = 'daemon-startup-phases.jsonl';
78+
const DAEMON_STARTUP_PHASES_ENTRY = /^daemon-startup-phases-(\d*)\.jsonl$/;
7079

71-
type DaemonStartupMarkerRecord = {
72-
data?: { startupPhase?: unknown; timestampMs?: unknown };
73-
};
74-
75-
export function resolveDaemonStartupPhasesPath(stateDir: string): string {
76-
return path.join(stateDir, DAEMON_STARTUP_PHASES_FILE);
80+
/** The marker file the daemon process `pid` writes in `stateDir`; an absent file is no evidence. */
81+
export function resolveDaemonStartupPhasesPath(stateDir: string, pid: number): string {
82+
return path.join(stateDir, `daemon-startup-phases-${pid}.jsonl`);
7783
}
7884

85+
/**
86+
* Appends one phase marker for the launch owning `startupPhasesPath`. A marker is progress
87+
* evidence, never a startup requirement: a host that refuses the append must not lose a daemon
88+
* that can still register. Producers that own a diagnostics channel pass `reportFailure` so the
89+
* refusal is observable where the waiting client already reads failure evidence (the daemon log
90+
* tail in the startup failure details); a client never receives a false marker from a broken
91+
* channel — it reads a missing record as no evidence.
92+
*/
7993
export function recordDaemonStartupMarker(
8094
startupPhasesPath: string,
8195
phase: DaemonStartupPhase,
96+
reportFailure?: (error: NormalizedError) => void,
8297
): void {
8398
try {
8499
const timestampMs = Date.now();
85100
fs.mkdirSync(path.dirname(startupPhasesPath), { recursive: true });
101+
// A crash mid-append can leave the file without a trailing newline; without a separator the
102+
// next append would glue into one unparsable line and lose both records.
103+
const separator = endsWithoutNewline(startupPhasesPath) ? '\n' : '';
86104
fs.appendFileSync(
87105
startupPhasesPath,
88-
`${JSON.stringify({ data: { startupPhase: phase, timestampMs } })}\n`,
106+
`${separator}${JSON.stringify({ data: { startupPhase: phase, timestampMs } })}\n`,
89107
);
108+
} catch (error) {
109+
reportFailure?.(normalizeError(error));
110+
}
111+
}
112+
113+
function endsWithoutNewline(file: string): boolean {
114+
let fd: number | undefined;
115+
try {
116+
const stats = fs.statSync(file);
117+
if (stats.size === 0) return false;
118+
const buffer = Buffer.alloc(1);
119+
fd = fs.openSync(file, 'r');
120+
if (fs.readSync(fd, buffer, 0, 1, stats.size - 1) !== 1) return false;
121+
return buffer[0] !== 0x0a;
90122
} catch {
91-
// A marker is progress evidence, never a startup requirement: a host that refuses the append
92-
// must not lose a daemon that can still register.
123+
return false;
124+
} finally {
125+
if (fd !== undefined) fs.closeSync(fd);
93126
}
94127
}
95128

96129
/**
97-
* The phases observed in a startup marker tail, keyed by phase, holding each phase's earliest
130+
* Removes one startup's marker file. The writer calls it once its own startup completes; the client
131+
* that launched a process calls it once that process's exit is joined. Both callers can prove no
132+
* other writer can still append to this file, which is what the pid-specific name encodes.
133+
*/
134+
export function removeDaemonStartupPhasesFile(startupPhasesPath: string): void {
135+
try {
136+
fs.rmSync(startupPhasesPath, { force: true });
137+
} catch {
138+
// Evidence removal never fails a daemon that is already up or a client that already joined.
139+
}
140+
}
141+
142+
/**
143+
* Sweeps marker files left by provably dead launches, from the position of a process that holds the
144+
* registration lock. A live pid's file is kept: its launch may still be between phases, and only
145+
* that launch's own completion or a joined exit may remove it. A file whose name carries no usable
146+
* pid cannot be owned by any launch, so this owner retires it too.
147+
*/
148+
function sweepDeadDaemonStartupPhasesFiles(stateDir: string): void {
149+
let entries: string[];
150+
try {
151+
entries = fs.readdirSync(stateDir);
152+
} catch {
153+
return;
154+
}
155+
for (const entry of entries) {
156+
const match = DAEMON_STARTUP_PHASES_ENTRY.exec(entry);
157+
if (match === null) continue;
158+
const deadPid = Number(match[1]);
159+
if (deadPid === process.pid) continue;
160+
if (!Number.isSafeInteger(deadPid) || deadPid <= 0 || !isProcessAlive(deadPid)) {
161+
removeDaemonStartupPhasesFile(path.join(stateDir, entry));
162+
}
163+
}
164+
}
165+
166+
/**
167+
* The phases observed in one launch's marker tail, keyed by phase, holding each phase's earliest
98168
* occurrence. Unparsable and unknown lines are ignored: the protocol is additive, and a client must
99169
* never fail a startup over a line shape it does not know.
100170
*/
@@ -125,6 +195,10 @@ export function readDaemonStartupPhaseMarkers(
125195
}
126196
return phases;
127197
}
198+
199+
type DaemonStartupMarkerRecord = {
200+
data?: { startupPhase?: unknown; timestampMs?: unknown };
201+
};
128202
export type DaemonRegistrationFields = Readonly<{
129203
socketPort?: number;
130204
httpPort?: number;
@@ -164,6 +238,9 @@ export async function tryAcquireDaemonRegistration(
164238
try {
165239
acquisition.assertHeld();
166240
fs.rmSync(resolveDaemonShutdownReportPath(boundPaths.baseDir), { force: true });
241+
// A dead launch's markers are this owner's to sweep: only the launch itself or a client that
242+
// joined its exit may remove a live launch's file, and neither can still reach this one.
243+
sweepDeadDaemonStartupPhasesFiles(boundPaths.baseDir);
167244
} catch (error) {
168245
await releaseRegistrationAfterFailure(acquisition, error, boundPaths.logPath);
169246
}

‎packages/daemon/src/server/http/daemon-runtime.ts‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ import {
9393
tryAcquireDaemonRegistration,
9494
DAEMON_STARTUP_EXIT_CODES,
9595
recordDaemonStartupMarker,
96+
removeDaemonStartupPhasesFile,
9697
resolveDaemonStartupPhasesPath,
9798
type DaemonRegistrationOwner,
9899
} from '@agent-device/daemon-contracts/daemon-registration-owner';
@@ -736,8 +737,16 @@ async function startDaemonRuntimeOnHost(
736737
await finishDaemonRegistration({ registration, infoPath, logPath });
737738
});
738739
// #3782: the client waiting on this state dir reads these markers to tell a slow daemon from a
739-
// stuck one.
740-
recordDaemonStartupMarker(resolveDaemonStartupPhasesPath(baseDir), 'lock-acquired');
740+
// stuck one. Each marker names a phase whose work has finished; this launch's file is named for
741+
// its own pid, so no other launch's evidence can add to or hide it.
742+
const startupPhasesPath = resolveDaemonStartupPhasesPath(baseDir, process.pid);
743+
const recordStartupPhase = (phase: Parameters<typeof recordDaemonStartupMarker>[1]): void =>
744+
recordDaemonStartupMarker(startupPhasesPath, phase, (error) => {
745+
// Same contract as the entry's markers: the refusal goes where the waiting client already
746+
// reads failure evidence (the daemon log), and startup keeps going.
747+
stderr.write(`Daemon startup marker write failed: ${error.message}\n`);
748+
});
749+
recordStartupPhase('lock-acquired');
741750

742751
let servers: DaemonServer[] = [];
743752
rollback.onRollback(async () => {
@@ -794,14 +803,21 @@ async function startDaemonRuntimeOnHost(
794803
data: { error: error instanceof Error ? error.message : String(error) },
795804
});
796805
});
806+
// #3782: the recovery chain above (owner-lifecycle configuration, legacy-marker and app-log
807+
// recovery, owned-process reaping, web-orphan cleanup) is real completed work: its boundary
808+
// extends the stall window exactly as much as it took, never more.
809+
recordStartupPhase('recovery-finished');
797810
const opened = await openDaemonServers();
798811
servers = opened.servers;
799812
const { socketPort, httpPort } = opened;
800813
// #3782: the last two progress markers. A client that polls the registration and the markers
801814
// together can tell "listening but slow to answer" from "never progressed".
802-
recordDaemonStartupMarker(resolveDaemonStartupPhasesPath(baseDir), 'listening');
815+
recordStartupPhase('listening');
803816
publishDaemonInfo(socketPort, httpPort);
804-
recordDaemonStartupMarker(resolveDaemonStartupPhasesPath(baseDir), 'registered');
817+
recordStartupPhase('registered');
818+
// This startup's evidence has done its job: no waiting client needs it anymore, and this
819+
// process is the only writer of the file, so this is the removal its owner contract allows.
820+
removeDaemonStartupPhasesFile(startupPhasesPath);
805821
stopMetadataLossWatch = armDaemonMetadataLossWatch(baseDir, infoPath, logPath, daemonIdentity);
806822
await flushDaemonStartupDiagnostics(logPath, startupDiagnostics);
807823
for (const plugin of deps.pluginInventory.filter(({ state }) => state === 'failed')) {

‎src/__tests__/test-utils/registered-daemon-fixture.ts‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ export function registeredDaemonFixtureArgs(
3434
entry,
3535
`import fs from 'node:fs';
3636
import path from 'node:path';
37-
import { DAEMON_STARTUP_EXIT_CODES, tryAcquireDaemonRegistration } from ${JSON.stringify(registrationUrl)};
37+
import { DAEMON_STARTUP_EXIT_CODES, tryAcquireDaemonRegistration, recordDaemonStartupMarker, resolveDaemonStartupPhasesPath } from ${JSON.stringify(registrationUrl)};
3838
const paths = ${JSON.stringify(paths)};
3939
const barrier = ${JSON.stringify(acquisitionBarrier)};
4040
if (barrier) {
@@ -43,6 +43,10 @@ if (barrier) {
4343
}
4444
const acquired = await tryAcquireDaemonRegistration(paths);
4545
if (acquired.status !== 'acquired') process.exit(DAEMON_STARTUP_EXIT_CODES[acquired.status]);
46+
// A suite that wants this child to behave like the real entry's marker producer drops this flag
47+
// file before spawning; the marker is written through the real protocol, to this pid's own file.
48+
if (fs.existsSync(path.join(paths.baseDir, 'emit-startup-markers')))
49+
recordDaemonStartupMarker(resolveDaemonStartupPhasesPath(paths.baseDir, process.pid), 'lock-acquired');
4650
process.on('SIGTERM', async () => {
4751
if (fs.existsSync(path.join(paths.baseDir, 'ignore-sigterm'))) return;
4852
const deferred = path.join(paths.baseDir, 'repair-on-shutdown.json');

0 commit comments

Comments
 (0)