mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 01:27:09 +01:00
perf: retire the aggregate heartbeat maintenance interval
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
4fbdf83b89
commit
0b22686ec0
4 files changed
+22
-65
No files matched your search
@@ -76,6 +76,8 @@ describe("assembled maintenance server production wiring", () => {
|
||||
expect(real.network.nonDatabaseConnections).toBe(before.network.nonDatabaseConnections);
|
||||
expect(real.network.nonDatabaseBytesWritten).toBe(before.network.nonDatabaseBytesWritten);
|
||||
const realFrames = server.proxy.frames, realTransport = server.proxy.transport;
|
||||
expect(realTransport.connectionsOpened).toBeGreaterThan(0);
|
||||
expect(realTransport.connectionsClosed).toBe(realTransport.connectionsOpened);
|
||||
for (let step = 0; step < 240; step++) { await server.advance(30_000); await server.settled(10); }
|
||||
const after = await server.snapshot();
|
||||
expect(after.now - real.now).toBe(2 * 60 * 60_000);
|
||||
@@ -85,6 +87,7 @@ describe("assembled maintenance server production wiring", () => {
|
||||
expect(after.network.nonDatabaseConnections).toBe(before.network.nonDatabaseConnections);
|
||||
expect(after.network.nonDatabaseBytesWritten).toBe(before.network.nonDatabaseBytesWritten);
|
||||
expect(server.proxy.errors).toEqual([]);
|
||||
expect(server.proxy.transport).toEqual(realTransport);
|
||||
await writeFile(path.join(server.directory, "enabled-quiet-measurement.json"), JSON.stringify({ phase: "enabled-quiet", state,
|
||||
wallElapsedMs, realClockElapsedMs: real.now - before.now, qualifiedClockMs: after.now - real.now, stepMs: 30_000, ioSettledBetweenSteps: true,
|
||||
measuredQueries: 0, startupQueries: queryCount, startup: receipt(server.proxy.queries), before, real, after,
|
||||
|
||||
@@ -186,7 +186,6 @@ const {
|
||||
scanSilentActiveRuns: vi.fn(async () => ({ created: 0, escalated: 0 })),
|
||||
sweepStaleIssueLocks: vi.fn(async () => ({ cleared: 0 })),
|
||||
sweepPendingCleanupLeases: vi.fn(async () => ({ swept: 0, destroyed: 0, capped: 0 })),
|
||||
sweepExpiredRuntimeStatuses: vi.fn(() => 0),
|
||||
tickAgentTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })),
|
||||
nextAgentTimerAt: vi.fn(async () => null),
|
||||
tickIssueMonitors: vi.fn(async () => ({ checked: 0, triggered: 0, skipped: 0 })),
|
||||
@@ -836,7 +835,7 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.skillVersionRetention)).toEqual([]);
|
||||
});
|
||||
|
||||
it.each([false, true])("registers status-card evaluation only with maintenance and never ticks it from the heartbeat loop (%s)", async enabled => {
|
||||
it.each([false, true])("registers status-card evaluation only with maintenance and creates no aggregate interval (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 1000 }));
|
||||
let tick: (() => void) | undefined;
|
||||
const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, interval: number) => {
|
||||
@@ -850,8 +849,8 @@ describe("startServer feedback export wiring", () => {
|
||||
if (enabled) {
|
||||
const service = vi.mocked(statusCardEvaluationService).mock.results.at(-1)!.value;
|
||||
expect(service.run).toHaveBeenCalledOnce();
|
||||
expect(tick).toBeDefined();
|
||||
for (let i = 0; i < 3; i++) { tick!(); await settleStartupHousekeeping(); }
|
||||
expect(tick).toBeUndefined();
|
||||
await settleStartupHousekeeping();
|
||||
expect(service.run).toHaveBeenCalledOnce();
|
||||
}
|
||||
expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.statusCardEvaluation)).toEqual([]);
|
||||
@@ -1073,7 +1072,7 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(registration).toBeUndefined();
|
||||
expect(attentionRetentionService).not.toHaveBeenCalled();
|
||||
expect(decisionRetentionService).not.toHaveBeenCalled();
|
||||
expect(tick).toBeDefined(); tick!(); await settleStartupHousekeeping();
|
||||
expect(tick).toBeUndefined(); await settleStartupHousekeeping();
|
||||
expect(decisionRetentionService).not.toHaveBeenCalled();
|
||||
return;
|
||||
}
|
||||
@@ -1089,8 +1088,8 @@ describe("startServer feedback export wiring", () => {
|
||||
const companyReads = () => vi.mocked(db.select).mock.results.filter(({ value }) =>
|
||||
vi.mocked(value.from).mock.calls.some(([table]) => table === companies)).length;
|
||||
const before = companyReads();
|
||||
expect(before).toBe(0); expect(tick).toBeDefined();
|
||||
tick!(); await settleStartupHousekeeping();
|
||||
expect(before).toBe(0); expect(tick).toBeUndefined();
|
||||
await settleStartupHousekeeping();
|
||||
expect(companyReads()).toBe(before);
|
||||
expect(source.load).toHaveBeenCalledOnce();
|
||||
expect(retention.deliverNotifications).toHaveBeenCalledOnce();
|
||||
@@ -1397,7 +1396,7 @@ describe("startServer feedback export wiring", () => {
|
||||
} finally { intervals.mockRestore(); }
|
||||
});
|
||||
|
||||
it("does not rescan idle tool health from the aggregate maintenance tick", async () => {
|
||||
it("registers idle tool health without an aggregate maintenance interval", async () => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, heartbeatSchedulerIntervalMs: 30_000 }));
|
||||
let tick: (() => void) | undefined;
|
||||
const intervals = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, ms: number) => {
|
||||
@@ -1414,9 +1413,8 @@ describe("startServer feedback export wiring", () => {
|
||||
const loads = vi.mocked(toolAccessService).mock.results.map(result => result.value.scheduledHealth.load);
|
||||
const loadCount = () => loads.reduce((total, load) => total + load.mock.calls.length, 0);
|
||||
expect(loadCount()).toBe(1);
|
||||
expect(tick).toEqual(expect.any(Function));
|
||||
tick!(); await new Promise<void>(resolve => setImmediate(resolve));
|
||||
tick!(); await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(tick).toBeUndefined();
|
||||
await new Promise<void>(resolve => setImmediate(resolve));
|
||||
expect(count()).toBe(0);
|
||||
expect(loadCount()).toBe(1);
|
||||
} finally { intervals.mockRestore(); }
|
||||
@@ -1439,8 +1437,7 @@ describe("startServer feedback export wiring", () => {
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled();
|
||||
expect(intervalCallback).not.toBeNull();
|
||||
intervalCallback?.();
|
||||
expect(intervalCallback).toBeNull();
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled();
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled();
|
||||
@@ -1533,8 +1530,7 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1);
|
||||
expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1);
|
||||
|
||||
expect(intervalCallback).not.toBeNull();
|
||||
intervalCallback?.();
|
||||
expect(intervalCallback).toBeNull();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
@@ -1591,14 +1587,11 @@ describe("startServer feedback export wiring", () => {
|
||||
try {
|
||||
await startServer();
|
||||
|
||||
// The disabled path still creates one heartbeat runtime. This runtime owns
|
||||
// the orphan-sandbox cleanup sweep, so a leaked provider sandbox is still
|
||||
// reaped at startup and on the interval.
|
||||
// The disabled path retains startup recovery and cleanup deadlines.
|
||||
expect(heartbeatServiceFactoryMock).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatServiceMock.sweepPendingCleanupLeases).toHaveBeenCalled();
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(intervalCallback).not.toBeNull();
|
||||
intervalCallback?.();
|
||||
expect(intervalCallback).toBeNull();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
@@ -1629,8 +1622,7 @@ describe("startServer feedback export wiring", () => {
|
||||
expect((await startServer()).server).toBe(fakeServer);
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(tick).toBeDefined();
|
||||
tick?.();
|
||||
expect(tick).toBeUndefined();
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.accounting);
|
||||
@@ -1724,10 +1716,12 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(getStartupRecoveryState().phase).toBe("ready");
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatIntervals).toHaveLength(0);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.routineSchedules)).toBe(false);
|
||||
|
||||
releaseCleanup?.();
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatIntervals).toHaveLength(1);
|
||||
expect(heartbeatIntervals).toHaveLength(0);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.routineSchedules)).toBe(true);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
@@ -1755,7 +1749,8 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(await registration![1].hasPending!()).toBe(true);
|
||||
expect(environmentCustomImagesServiceMock.nextSetupCleanupAt).not.toHaveBeenCalled();
|
||||
// The scheduler starts only after every housekeeping step settled.
|
||||
expect(heartbeatIntervals).toHaveLength(1);
|
||||
expect(heartbeatIntervals).toHaveLength(0);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.routineSchedules)).toBe(true);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
|
||||
@@ -1250,7 +1250,6 @@ async function startServerWithDatabaseTeardown(
|
||||
drainRunIds?: string[];
|
||||
}>) | null = null;
|
||||
let heartbeatSchedulerStopped = false;
|
||||
let heartbeatSchedulerInterval: ReturnType<typeof setInterval> | null = null;
|
||||
const heartbeatSchedulerInFlight = new Set<Promise<void>>();
|
||||
const trackHeartbeatSchedulerWork = (work: Promise<unknown>) => {
|
||||
const finishIdleWork = beginIdleTrackedWork();
|
||||
@@ -1268,12 +1267,6 @@ async function startServerWithDatabaseTeardown(
|
||||
await Promise.allSettled([...heartbeatSchedulerInFlight]);
|
||||
}
|
||||
};
|
||||
const startHeartbeatSchedulerInterval = (callback: () => void) => {
|
||||
heartbeatSchedulerInterval = setInterval(() => {
|
||||
if (!isWarmStandby() && !isIdleTaskDrainActive()) callback();
|
||||
}, config.heartbeatSchedulerIntervalMs);
|
||||
heartbeatSchedulerInterval?.unref?.();
|
||||
};
|
||||
const skillSourceSync = skillSourceSyncService(db as any);
|
||||
const nativeHarnessBackupExpiration = nativeHarnessBackupExpirationService(db as any, {
|
||||
duty: "backups",
|
||||
@@ -1841,31 +1834,6 @@ async function startServerWithDatabaseTeardown(
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
};
|
||||
|
||||
const createHeartbeatSchedulerTick = (scheduled: ReturnType<typeof registerScheduledWork>) => () => {
|
||||
// Track the outer async callback as well as the work it starts. Shutdown
|
||||
// can then wait through an already-running suppression check before it
|
||||
// captures the authoritative set of running heartbeat rows.
|
||||
trackHeartbeatSchedulerWork((async () => {
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses();
|
||||
if (sweptRuntimeStatuses > 0) {
|
||||
logger.info(
|
||||
{ swept: sweptRuntimeStatuses },
|
||||
"heartbeat runtime-status sweeper cleared expired entries",
|
||||
);
|
||||
}
|
||||
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
|
||||
|
||||
})().catch((err) => {
|
||||
logger.error({ err }, "heartbeat scheduler tick failed");
|
||||
}));
|
||||
};
|
||||
const startHeartbeatScheduler = () => {
|
||||
if (isCloudManagedInstance()) {
|
||||
const status = registerRoutineExecutionStatusWork(deliveryWork, routineExecutionStatusService(db as any),
|
||||
@@ -1891,7 +1859,6 @@ async function startServerWithDatabaseTeardown(
|
||||
}),
|
||||
});
|
||||
trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.monitors.ready, scheduled.routines.ready]));
|
||||
startHeartbeatSchedulerInterval(createHeartbeatSchedulerTick(scheduled));
|
||||
};
|
||||
trackHeartbeatSchedulerWork(runStartupHousekeeping().then(() => {
|
||||
if (!heartbeatSchedulerStopped) startHeartbeatScheduler();
|
||||
@@ -1902,8 +1869,6 @@ async function startServerWithDatabaseTeardown(
|
||||
// this path registers the same startup recovery and pending-work deadlines
|
||||
// as the enabled path.
|
||||
await startEnvironmentCleanup();
|
||||
startHeartbeatSchedulerInterval(() => {
|
||||
});
|
||||
}
|
||||
|
||||
if (config.databaseBackupEnabled) {
|
||||
@@ -2020,10 +1985,6 @@ async function startServerWithDatabaseTeardown(
|
||||
await workspaceRecoveryWork.stop();
|
||||
unregisterTerminalWorkspaceProbe?.();
|
||||
unsubscribeNativeWorkspaceSyncRetention?.();
|
||||
if (heartbeatSchedulerInterval) {
|
||||
clearInterval(heartbeatSchedulerInterval);
|
||||
heartbeatSchedulerInterval = null;
|
||||
}
|
||||
|
||||
const heartbeatShutdown = await coordinateHeartbeatSchedulerShutdown({
|
||||
signal,
|
||||
|
||||
@@ -524,7 +524,6 @@ import { environmentRunOrchestrator } from "./environment-run-orchestrator.js";
|
||||
|
||||
import {
|
||||
clearHeartbeatRunRuntimeStatus,
|
||||
sweepExpiredHeartbeatRunRuntimeStatuses,
|
||||
} from "./heartbeat-run-runtime-status.js";
|
||||
|
||||
import { resolveAndRetainRunTrustPreset } from "./run-trust-preset.js";
|
||||
@@ -4003,7 +4002,6 @@ export function heartbeatService(
|
||||
|
||||
decorateActiveRunStatus: decorateHeartbeatRunRuntimeStatus,
|
||||
recordRuntimeProgress: recordCurrentHeartbeatRunRuntimeProgress,
|
||||
sweepExpiredRuntimeStatuses: sweepExpiredHeartbeatRunRuntimeStatuses,
|
||||
|
||||
getRunReadAccess,
|
||||
getRunLogAccess,
|
||||
|
||||
Reference in new issue
Block a user