Skip to content

Commit 6394ef4

Browse files
thymikeethymikee
committed
fix: rotate app logs after process relaunch
1 parent 5f63821 commit 6394ef4

8 files changed

Lines changed: 335 additions & 142 deletions

File tree

‎packages/capture-kit/src/app-log-pid-monitor.test.ts‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
import { describe, expect, test, vi } from 'vitest';
2+
import type { AppLogBackgroundProcess } from '@agent-device/contracts/platform';
3+
import { monitorPidScopedProcess } from './app-log-pid-monitor.ts';
4+
import { controlledSleeps, deferred, processFixture } from './app-log-pid-process.fixtures.ts';
5+
6+
describe('PID-scoped app-log monitor', () => {
7+
test('rotates a live stream when the app PID changes', async () => {
8+
const first = processFixture(deferred<{ stdout: string; stderr: string; exitCode: number }>());
9+
const second = processFixture(deferred<{ stdout: string; stderr: string; exitCode: number }>());
10+
const sleeps = controlledSleeps();
11+
const resolvePid = vi.fn(async () => '456');
12+
const startProcess = vi.fn(async () => second.process);
13+
let stopped = false;
14+
let active: AppLogBackgroundProcess | undefined;
15+
const monitor = monitorPidScopedProcess({
16+
initialProcess: { pid: '123', process: first.process },
17+
stopped: () => stopped,
18+
setActive: (process) => {
19+
active = process;
20+
},
21+
setState: vi.fn(),
22+
resolvePid,
23+
startProcess,
24+
sleep: sleeps.sleep,
25+
});
26+
27+
await vi.waitFor(() => expect(sleeps.pending()).toBe(1));
28+
sleeps.releaseNext();
29+
await vi.waitFor(() => expect(first.terminate).toHaveBeenCalledOnce());
30+
await vi.waitFor(() => expect(sleeps.pending()).toBe(1));
31+
sleeps.releaseNext();
32+
await vi.waitFor(() => expect(startProcess).toHaveBeenCalledWith('456'));
33+
await vi.waitFor(() => expect(sleeps.pending()).toBe(1));
34+
35+
stopped = true;
36+
await active?.terminate();
37+
await expect(monitor).resolves.toBeUndefined();
38+
expect(resolvePid).toHaveBeenCalledTimes(2);
39+
expect(first.dispose).toHaveBeenCalledOnce();
40+
expect(second.terminate).toHaveBeenCalledOnce();
41+
expect(second.dispose).toHaveBeenCalledOnce();
42+
});
43+
});

‎packages/capture-kit/src/app-log-pid-monitor.ts‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
import type { AppLogBackgroundProcess, AppLogLiveSnapshot } from '@agent-device/contracts/platform';
2+
3+
export type PidScopedProcess = Readonly<{
4+
pid: string;
5+
process: AppLogBackgroundProcess;
6+
}>;
7+
8+
export type PidScopedProcessMonitor = Readonly<{
9+
initialProcess: PidScopedProcess | undefined;
10+
stopped(): boolean;
11+
setActive(process: AppLogBackgroundProcess | undefined): void;
12+
setState(state: AppLogLiveSnapshot['state']): void;
13+
resolvePid(): Promise<string>;
14+
startProcess(pid: string): Promise<AppLogBackgroundProcess>;
15+
sleep(milliseconds: number, signal?: AbortSignal): Promise<void>;
16+
}>;
17+
18+
type SubsequentProcessStart =
19+
| Readonly<{ status: 'stopped' }>
20+
| Readonly<{ status: 'waiting' }>
21+
| Readonly<{ status: 'started'; process: PidScopedProcess }>;
22+
23+
/** Monitors a PID-scoped stream and rotates it when the application process changes. */
24+
export async function monitorPidScopedProcess(input: PidScopedProcessMonitor): Promise<void> {
25+
let process = input.initialProcess;
26+
while (!input.stopped()) {
27+
if (process) {
28+
await settleActiveProcess(input, process);
29+
process = undefined;
30+
await pauseBeforeProcessRestart(input);
31+
continue;
32+
}
33+
34+
const started = await startAndAdoptSubsequentProcess(input);
35+
if (started.status === 'stopped') return;
36+
if (started.status === 'waiting') continue;
37+
process = started.process;
38+
}
39+
}
40+
41+
async function settleActiveProcess(
42+
input: PidScopedProcessMonitor,
43+
scopedProcess: PidScopedProcess,
44+
): Promise<void> {
45+
const { process } = scopedProcess;
46+
input.setActive(process);
47+
try {
48+
input.setState('active');
49+
await observeActiveProcess(input, scopedProcess);
50+
} finally {
51+
await disposeAdoptedProcess(input, process);
52+
}
53+
}
54+
55+
async function observeActiveProcess(
56+
input: PidScopedProcessMonitor,
57+
scopedProcess: PidScopedProcess,
58+
): Promise<void> {
59+
const wait = observeProcessWait(scopedProcess.process);
60+
try {
61+
await pollActiveProcessPid(input, scopedProcess, wait);
62+
await terminateStoppedProcess(input, scopedProcess.process, wait);
63+
await assertProcessWaitSucceeded(wait.outcome);
64+
} finally {
65+
wait.controller.abort();
66+
}
67+
}
68+
69+
type ProcessWaitObservation = Readonly<{
70+
controller: AbortController;
71+
outcome: Promise<Readonly<{ status: 'exited' }> | Readonly<{ status: 'failed'; error: unknown }>>;
72+
settled(): boolean;
73+
}>;
74+
75+
function observeProcessWait(process: AppLogBackgroundProcess): ProcessWaitObservation {
76+
const controller = new AbortController();
77+
let settled = false;
78+
const finish = () => {
79+
settled = true;
80+
controller.abort();
81+
};
82+
const outcome = process.wait.then(
83+
() => {
84+
finish();
85+
return { status: 'exited' } as const;
86+
},
87+
(error: unknown) => {
88+
finish();
89+
return { status: 'failed', error } as const;
90+
},
91+
);
92+
return { controller, outcome, settled: () => settled };
93+
}
94+
95+
async function pollActiveProcessPid(
96+
input: PidScopedProcessMonitor,
97+
scopedProcess: PidScopedProcess,
98+
wait: ProcessWaitObservation,
99+
): Promise<void> {
100+
while (!input.stopped() && !wait.settled()) {
101+
if (!(await waitForPidPoll(input, wait))) return;
102+
const observedPid = await input.resolvePid();
103+
if (input.stopped()) return;
104+
if (observedPid === scopedProcess.pid) continue;
105+
await scopedProcess.process.terminate();
106+
return;
107+
}
108+
}
109+
110+
async function waitForPidPoll(
111+
input: PidScopedProcessMonitor,
112+
wait: ProcessWaitObservation,
113+
): Promise<boolean> {
114+
try {
115+
await input.sleep(500, wait.controller.signal);
116+
} catch (error) {
117+
if (!wait.settled()) throw error;
118+
}
119+
return !wait.settled();
120+
}
121+
122+
async function terminateStoppedProcess(
123+
input: PidScopedProcessMonitor,
124+
process: AppLogBackgroundProcess,
125+
wait: ProcessWaitObservation,
126+
): Promise<void> {
127+
if (input.stopped() && !wait.settled()) await process.terminate();
128+
}
129+
130+
async function assertProcessWaitSucceeded(
131+
outcome: ProcessWaitObservation['outcome'],
132+
): Promise<void> {
133+
const settled = await outcome;
134+
if (settled.status === 'failed') throw settled.error;
135+
}
136+
137+
async function pauseBeforeProcessRestart(input: PidScopedProcessMonitor): Promise<void> {
138+
if (input.stopped()) return;
139+
input.setState('recovering');
140+
await input.sleep(500);
141+
}
142+
143+
async function startAndAdoptSubsequentProcess(
144+
input: PidScopedProcessMonitor,
145+
): Promise<SubsequentProcessStart> {
146+
const pid = await input.resolvePid();
147+
if (input.stopped()) return { status: 'stopped' };
148+
if (!pid) {
149+
input.setState('recovering');
150+
await input.sleep(1_000);
151+
return { status: 'waiting' };
152+
}
153+
const process = await input.startProcess(pid);
154+
input.setActive(process);
155+
if (!input.stopped()) return { status: 'started', process: { pid, process } };
156+
await stopAdoptedProcess(input, process);
157+
return { status: 'stopped' };
158+
}
159+
160+
async function stopAdoptedProcess(
161+
input: PidScopedProcessMonitor,
162+
process: AppLogBackgroundProcess,
163+
): Promise<void> {
164+
try {
165+
await process.terminate();
166+
await process.wait;
167+
} finally {
168+
await disposeAdoptedProcess(input, process);
169+
}
170+
}
171+
172+
async function disposeAdoptedProcess(
173+
input: PidScopedProcessMonitor,
174+
process: AppLogBackgroundProcess,
175+
): Promise<void> {
176+
try {
177+
await process[Symbol.asyncDispose]();
178+
} finally {
179+
input.setActive(undefined);
180+
}
181+
}

‎packages/capture-kit/src/app-log-pid-process.fixtures.ts‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
import { vi } from 'vitest';
2+
import type { AppLogBackgroundProcess } from '@agent-device/contracts/platform';
3+
4+
export function deferred<T>() {
5+
let resolve!: (value: T) => void;
6+
const promise = new Promise<T>((settle) => {
7+
resolve = settle;
8+
});
9+
return { promise, resolve };
10+
}
11+
12+
export function processFixture(
13+
wait: ReturnType<typeof deferred<{ stdout: string; stderr: string; exitCode: number }>>,
14+
) {
15+
const terminate = vi.fn(async () => wait.resolve({ stdout: '', stderr: '', exitCode: 0 }));
16+
const dispose = vi.fn(async () => {});
17+
return {
18+
terminate,
19+
dispose,
20+
process: {
21+
wait: wait.promise,
22+
terminate,
23+
[Symbol.asyncDispose]: dispose,
24+
} satisfies AppLogBackgroundProcess,
25+
};
26+
}
27+
28+
export function controlledSleeps() {
29+
const releases: (() => void)[] = [];
30+
const sleep = (_milliseconds: number, signal?: AbortSignal) =>
31+
new Promise<void>((resolve, reject) => {
32+
const release = () => {
33+
signal?.removeEventListener('abort', onAbort);
34+
resolve();
35+
};
36+
const onAbort = () => {
37+
const index = releases.indexOf(release);
38+
if (index >= 0) releases.splice(index, 1);
39+
reject(signal?.reason);
40+
};
41+
releases.push(release);
42+
signal?.addEventListener('abort', onAbort, { once: true });
43+
});
44+
return {
45+
sleep,
46+
pending: () => releases.length,
47+
releaseNext: () => releases.shift()?.(),
48+
};
49+
}

‎packages/capture-kit/src/app-log-pid-process.test.ts‎

Lines changed: 17 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import type {
44
AppLogOutputSink,
55
AppLogRuntimeHost,
66
} from '@agent-device/contracts/platform';
7+
import { deferred, processFixture } from './app-log-pid-process.fixtures.ts';
78
import { createPidScopedAppLogProcess } from './app-log-pid-process.ts';
89

910
describe('PID-scoped app-log process lifecycle', () => {
@@ -97,7 +98,7 @@ describe('PID-scoped app-log process lifecycle', () => {
9798
});
9899

99100
await vi.waitFor(() => expect(start).toHaveBeenCalledOnce());
100-
expect(resolvePid).toHaveBeenCalledTimes(2);
101+
expect(resolvePid.mock.calls.length).toBeGreaterThanOrEqual(2);
101102
await handle.finish();
102103
expect(started.terminate).toHaveBeenCalledOnce();
103104
});
@@ -151,30 +152,6 @@ describe('PID-scoped app-log process lifecycle', () => {
151152
});
152153
});
153154

154-
function deferred<T>() {
155-
let resolve!: (value: T) => void;
156-
const promise = new Promise<T>((settle) => {
157-
resolve = settle;
158-
});
159-
return { promise, resolve };
160-
}
161-
162-
function processFixture(
163-
wait: ReturnType<typeof deferred<{ stdout: string; stderr: string; exitCode: number }>>,
164-
) {
165-
const terminate = vi.fn(async () => wait.resolve({ stdout: '', stderr: '', exitCode: 0 }));
166-
const dispose = vi.fn(async () => {});
167-
return {
168-
terminate,
169-
dispose,
170-
process: {
171-
wait: wait.promise,
172-
terminate,
173-
[Symbol.asyncDispose]: dispose,
174-
} satisfies AppLogBackgroundProcess,
175-
};
176-
}
177-
178155
function outputFixture() {
179156
const dispose = vi.fn(async () => {});
180157
return {
@@ -216,6 +193,20 @@ function hostFixture(output: AppLogOutputSink, onOpen?: () => void): AppLogRunti
216193
inspect: async () => 'missing',
217194
terminate: async () => 'already-missing',
218195
},
219-
clock: { now: () => 0, sleep: vi.fn(async () => {}) },
196+
clock: { now: () => 0, sleep: vi.fn(testSleep) },
220197
};
221198
}
199+
200+
async function testSleep(_milliseconds: number, signal?: AbortSignal): Promise<void> {
201+
await new Promise<void>((resolve, reject) => {
202+
const onAbort = () => {
203+
clearTimeout(timeout);
204+
reject(signal?.reason);
205+
};
206+
const timeout = setTimeout(() => {
207+
signal?.removeEventListener('abort', onAbort);
208+
resolve();
209+
}, 1);
210+
signal?.addEventListener('abort', onAbort, { once: true });
211+
});
212+
}

0 commit comments

Comments
 (0)