mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 09:07:46 +02:00
test: release held adapters even when recovery assertions fail
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip committed
1 parent
722db815da
commit
d442fc431f
1 file changed
+129
-125
@@ -819,13 +819,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
}
|
||||
|
||||
it("does not reap active adapter executions started by another heartbeat service instance", async () => {
|
||||
let releaseAdapter: (() => void) | null = null;
|
||||
let releaseAdapter!: () => void;
|
||||
const adapterRelease = new Promise<void>(resolve => { releaseAdapter = resolve; });
|
||||
const adapterStarted = new Promise<void>((resolve) => {
|
||||
mockAdapterExecute.mockImplementationOnce(async () => {
|
||||
resolve();
|
||||
await new Promise<void>((release) => {
|
||||
releaseAdapter = release;
|
||||
});
|
||||
await adapterRelease;
|
||||
return {
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
@@ -849,52 +848,55 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
const executorHeartbeat = heartbeatService(db);
|
||||
const reaperHeartbeat = heartbeatService(db);
|
||||
|
||||
await executorHeartbeat.resumeQueuedRuns();
|
||||
await Promise.race([
|
||||
adapterStarted,
|
||||
new Promise<never>((_, reject) => {
|
||||
setTimeout(
|
||||
() =>
|
||||
reject(
|
||||
new Error("Timed out waiting for adapter execution to start"),
|
||||
),
|
||||
3_000,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
try {
|
||||
await executorHeartbeat.resumeQueuedRuns();
|
||||
await Promise.race([
|
||||
adapterStarted,
|
||||
new Promise<never>((_, reject) => {
|
||||
setTimeout(
|
||||
() =>
|
||||
reject(
|
||||
new Error("Timed out waiting for adapter execution to start"),
|
||||
),
|
||||
3_000,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
updatedAt: new Date("2026-03-19T00:00:00.000Z"),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
|
||||
const result = await reaperHeartbeat.reapOrphanedRuns({
|
||||
staleThresholdMs: 1,
|
||||
});
|
||||
expect(result).toEqual({ reaped: 0, runIds: [] });
|
||||
const result = await reaperHeartbeat.reapOrphanedRuns({
|
||||
staleThresholdMs: 1,
|
||||
});
|
||||
expect(result).toEqual({ reaped: 0, runIds: [] });
|
||||
|
||||
const activeRun = await reaperHeartbeat.getRun(runId);
|
||||
expect(activeRun?.status).toBe("running");
|
||||
expect(activeRun?.errorCode).toBeNull();
|
||||
const activeRun = await reaperHeartbeat.getRun(runId);
|
||||
expect(activeRun?.status).toBe("running");
|
||||
expect(activeRun?.errorCode).toBeNull();
|
||||
|
||||
const wakeup = await db
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(wakeup?.status).toBe("claimed");
|
||||
const wakeup = await db
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.id, wakeupRequestId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(wakeup?.status).toBe("claimed");
|
||||
|
||||
if (!releaseAdapter)
|
||||
throw new Error("Adapter release handle was not captured");
|
||||
releaseAdapter();
|
||||
const settledRun = await waitForRunToSettle(
|
||||
executorHeartbeat,
|
||||
runId,
|
||||
5_000,
|
||||
);
|
||||
expect(settledRun?.status).toBe("succeeded");
|
||||
releaseAdapter();
|
||||
const settledRun = await waitForRunToSettle(
|
||||
executorHeartbeat,
|
||||
runId,
|
||||
5_000,
|
||||
);
|
||||
expect(settledRun?.status).toBe("succeeded");
|
||||
} finally {
|
||||
releaseAdapter();
|
||||
await executorHeartbeat.drainActiveRunExecutions();
|
||||
}
|
||||
});
|
||||
|
||||
async function seedStrandedIssueFixture(input: {
|
||||
@@ -3407,7 +3409,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
});
|
||||
|
||||
it("persists codex_local spawn identity before hot restart and never loses the live run for missing metadata", async () => {
|
||||
let releaseAdapter: (() => void) | null = null;
|
||||
let releaseAdapter!: () => void;
|
||||
const adapterRelease = new Promise<void>(resolve => { releaseAdapter = resolve; });
|
||||
let spawnedPid: number | null = null;
|
||||
const adapterStarted = new Promise<void>((resolve) => {
|
||||
mockAdapterExecute.mockImplementationOnce(async (rawInput?: unknown) => {
|
||||
@@ -3429,9 +3432,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
startedAt: new Date("2026-07-30T07:00:00.000Z").toISOString(),
|
||||
});
|
||||
resolve();
|
||||
await new Promise<void>((release) => {
|
||||
releaseAdapter = release;
|
||||
});
|
||||
await adapterRelease;
|
||||
return {
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
@@ -3456,87 +3457,90 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
includeIssue: false,
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await Promise.race([
|
||||
adapterStarted,
|
||||
new Promise<never>((_, reject) => {
|
||||
setTimeout(
|
||||
() =>
|
||||
reject(
|
||||
new Error("Timed out waiting for codex_local spawn identity"),
|
||||
),
|
||||
3_000,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
|
||||
const running = await waitForValue(async () =>
|
||||
db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.then((rows) => {
|
||||
const row = rows[0] ?? null;
|
||||
return row?.status === "running" && row.processPid ? row : null;
|
||||
try {
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await Promise.race([
|
||||
adapterStarted,
|
||||
new Promise<never>((_, reject) => {
|
||||
setTimeout(
|
||||
() =>
|
||||
reject(
|
||||
new Error("Timed out waiting for codex_local spawn identity"),
|
||||
),
|
||||
3_000,
|
||||
);
|
||||
}),
|
||||
);
|
||||
const observedProcessStartedAt = await readProcessStartedAt(spawnedPid!);
|
||||
expect(observedProcessStartedAt).not.toBeNull();
|
||||
expect(running).toMatchObject({
|
||||
id: runId,
|
||||
status: "running",
|
||||
processPid: spawnedPid,
|
||||
processGroupId: null,
|
||||
processStartedAt: new Date(observedProcessStartedAt!),
|
||||
});
|
||||
]);
|
||||
|
||||
await withTempPaperclipHome(async (home) => {
|
||||
await writeHotRestartIntent({
|
||||
previousServerPid: process.pid,
|
||||
previousServerVersion: "old-version",
|
||||
requestedAt: new Date("2026-07-30T07:01:00.000Z"),
|
||||
});
|
||||
await heartbeat.prepareHotRestartShutdown(
|
||||
"SIGTERM",
|
||||
new Date("2026-07-30T07:02:00.000Z"),
|
||||
);
|
||||
|
||||
const adoption = await heartbeat.reconcileHotRestartAdoption(
|
||||
new Date("2026-07-30T07:03:00.000Z"),
|
||||
);
|
||||
expect(adoption).toMatchObject({
|
||||
mode: "reported",
|
||||
adoptedRunIds: [runId],
|
||||
finalizedWhileDownRunIds: [],
|
||||
lostRunIds: [],
|
||||
});
|
||||
const report = JSON.parse(
|
||||
await fs.readFile(resolveHotRestartReportPath(home), "utf8"),
|
||||
) as { runs?: Array<Record<string, unknown>> };
|
||||
expect(report.runs).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
runId,
|
||||
classification: "adopted",
|
||||
reason: "process_pid_alive",
|
||||
const running = await waitForValue(async () =>
|
||||
db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.then((rows) => {
|
||||
const row = rows[0] ?? null;
|
||||
return row?.status === "running" && row.processPid ? row : null;
|
||||
}),
|
||||
]),
|
||||
);
|
||||
expect(report.runs).not.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
runId,
|
||||
reason: "missing_process_metadata",
|
||||
}),
|
||||
]),
|
||||
);
|
||||
});
|
||||
const observedProcessStartedAt = await readProcessStartedAt(spawnedPid!);
|
||||
expect(observedProcessStartedAt).not.toBeNull();
|
||||
expect(running).toMatchObject({
|
||||
id: runId,
|
||||
status: "running",
|
||||
processPid: spawnedPid,
|
||||
processGroupId: null,
|
||||
processStartedAt: new Date(observedProcessStartedAt!),
|
||||
});
|
||||
|
||||
if (!releaseAdapter)
|
||||
throw new Error("Adapter release handle was not captured");
|
||||
releaseAdapter();
|
||||
const settled = await waitForRunToSettle(heartbeat, runId, 5_000);
|
||||
expect(settled?.status).toBe("succeeded");
|
||||
await withTempPaperclipHome(async (home) => {
|
||||
await writeHotRestartIntent({
|
||||
previousServerPid: process.pid,
|
||||
previousServerVersion: "old-version",
|
||||
requestedAt: new Date("2026-07-30T07:01:00.000Z"),
|
||||
});
|
||||
await heartbeat.prepareHotRestartShutdown(
|
||||
"SIGTERM",
|
||||
new Date("2026-07-30T07:02:00.000Z"),
|
||||
);
|
||||
|
||||
const adoption = await heartbeat.reconcileHotRestartAdoption(
|
||||
new Date("2026-07-30T07:03:00.000Z"),
|
||||
);
|
||||
expect(adoption).toMatchObject({
|
||||
mode: "reported",
|
||||
adoptedRunIds: [runId],
|
||||
finalizedWhileDownRunIds: [],
|
||||
lostRunIds: [],
|
||||
});
|
||||
const report = JSON.parse(
|
||||
await fs.readFile(resolveHotRestartReportPath(home), "utf8"),
|
||||
) as { runs?: Array<Record<string, unknown>> };
|
||||
expect(report.runs).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
runId,
|
||||
classification: "adopted",
|
||||
reason: "process_pid_alive",
|
||||
}),
|
||||
]),
|
||||
);
|
||||
expect(report.runs).not.toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
runId,
|
||||
reason: "missing_process_metadata",
|
||||
}),
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
releaseAdapter();
|
||||
const settled = await waitForRunToSettle(heartbeat, runId, 5_000);
|
||||
expect(settled?.status).toBe("succeeded");
|
||||
} finally {
|
||||
releaseAdapter();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
}
|
||||
});
|
||||
|
||||
it("reports adopted hot-restart runs before startup reap can mark them process_lost", async () => {
|
||||
|
||||
Reference in new issue
Block a user