From 957770324960a120865a346d0d2ea2abc887ce45 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Mon, 27 Apr 2026 22:26:41 -0700 Subject: [PATCH] test(gateway): trim cron server memory hotspots Move pure cron coverage off websocket server RPC loops and clean up timeout listener retention in gateway test helpers. --- src/gateway/server.cron.test.ts | 156 ++++++++++++++++------------- src/gateway/test-helpers.server.ts | 23 +++-- 2 files changed, 100 insertions(+), 79 deletions(-) diff --git a/src/gateway/server.cron.test.ts b/src/gateway/server.cron.test.ts index d0fe19f22f7..0c13612377f 100644 --- a/src/gateway/server.cron.test.ts +++ b/src/gateway/server.cron.test.ts @@ -4,6 +4,7 @@ import path from "node:path"; import { setImmediate as setImmediatePromise } from "node:timers/promises"; import { afterAll, beforeEach, describe, expect, test, vi } from "vitest"; import type WebSocket from "ws"; +import { resetConfigRuntimeState } from "../config/config.js"; import type { GuardedFetchOptions } from "../infra/net/fetch-guard.js"; import { connectOk, @@ -185,29 +186,37 @@ async function directCronReq( let result: | { ok: boolean; payload?: unknown; error?: { code?: string; message?: string } } | undefined; - await cronHandlers[method]({ - req: {} as never, - params, - respond: (ok, payload, error) => { - result = { - ok, - payload, - error, - }; - }, - context: { - cron: cronState.cron, - cronStorePath: cronState.storePath, - logGateway: { - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - }, - getRuntimeConfig: cronState.getRuntimeConfig, - } as never, - client: null, - isWebchatConnect: () => false, - }); + const respond = (ok: boolean, payload?: unknown, error?: { code?: string; message?: string }) => { + result = { + ok, + payload, + error, + }; + }; + try { + await cronHandlers[method]({ + req: {} as never, + params, + respond, + context: { + cron: cronState.cron, + cronStorePath: cronState.storePath, + logGateway: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, + getRuntimeConfig: cronState.getRuntimeConfig, + } as never, + client: null, + isWebchatConnect: () => false, + }); + } catch (err) { + respond(false, undefined, { + code: "unavailable", + message: err instanceof Error ? err.message : String(err), + }); + } if (!result) { throw new Error(`${method} did not respond`); } @@ -222,8 +231,12 @@ function expectCronJobIdFromResponse(response: { ok?: unknown; payload?: unknown return id; } -async function addMainSystemEventCronJob(params: { ws: WebSocket; name: string; text?: string }) { - const response = await rpcReq(params.ws, "cron.add", { +async function addMainSystemEventCronJobDirect(params: { + cronState: DirectCronState; + name: string; + text?: string; +}) { + const response = await directCronReq(params.cronState, "cron.add", { name: params.name, enabled: true, schedule: { kind: "every", everyMs: 60_000 }, @@ -263,6 +276,7 @@ async function writeCronConfig(config: unknown) { expect(typeof configPath).toBe("string"); await fs.mkdir(path.dirname(configPath as string), { recursive: true }); await fs.writeFile(configPath as string, JSON.stringify(config, null, 2), "utf-8"); + resetConfigRuntimeState(); } async function runCronJobForce(ws: WebSocket, id: string) { @@ -322,11 +336,10 @@ describe("gateway server cron", () => { cronEnabled: false, }); - const { server, ws } = await startServerWithClient(); - await connectOk(ws); + const cronState = await createDirectCronState(); try { - const addRes = await rpcReq(ws, "cron.add", { + const addRes = await directCronReq(cronState, "cron.add", { name: "daily", enabled: true, schedule: { kind: "every", everyMs: 60_000 }, @@ -339,7 +352,7 @@ describe("gateway server cron", () => { const dailyJobId = (addRes.payload as { id?: unknown } | null)?.id; expect(typeof dailyJobId).toBe("string"); - const listRes = await rpcReq(ws, "cron.list", { + const listRes = await directCronReq(cronState, "cron.list", { includeDisabled: true, }); expect(listRes.ok).toBe(true); @@ -362,7 +375,7 @@ describe("gateway server cron", () => { }); const routeAtMs = Date.now() - 1; - const routeRes = await rpcReq(ws, "cron.add", { + const routeRes = await directCronReq(cronState, "cron.add", { name: "route test", enabled: true, schedule: { kind: "at", at: new Date(routeAtMs).toISOString() }, @@ -375,14 +388,14 @@ describe("gateway server cron", () => { const routeJobId = typeof routeJobIdValue === "string" ? routeJobIdValue : ""; expect(routeJobId.length > 0).toBe(true); - const runRes = await rpcReq(ws, "cron.run", { id: routeJobId, mode: "force" }, 20_000); + const runRes = await directCronReq(cronState, "cron.run", { id: routeJobId, mode: "force" }); expect(runRes.ok).toBe(true); expect(runRes.payload).toEqual({ ok: true, enqueued: true, runId: expect.any(String) }); const events = await waitForSystemEvent(); expect(events.some((event) => event.includes("cron route check"))).toBe(true); const wrappedAtMs = Date.now() + 1000; - const wrappedRes = await rpcReq(ws, "cron.add", { + const wrappedRes = await directCronReq(cronState, "cron.add", { data: { name: "wrapped", schedule: { at: new Date(wrappedAtMs).toISOString() }, @@ -397,10 +410,13 @@ describe("gateway server cron", () => { expect(wrappedPayload?.wakeMode).toBe("now"); expect((wrappedPayload?.schedule as { kind?: unknown } | undefined)?.kind).toBe("at"); - const patchJobId = await addMainSystemEventCronJob({ ws, name: "patch test" }); + const patchJobId = await addMainSystemEventCronJobDirect({ + cronState, + name: "patch test", + }); const atMs = Date.now() + 1_000; - const updateRes = await rpcReq(ws, "cron.update", { + const updateRes = await directCronReq(cronState, "cron.update", { id: patchJobId, patch: { schedule: { at: new Date(atMs).toISOString() }, @@ -414,7 +430,7 @@ describe("gateway server cron", () => { expect(updated?.schedule?.kind).toBe("at"); expect(updated?.payload?.kind).toBe("systemEvent"); - const mergeRes = await rpcReq(ws, "cron.add", { + const mergeRes = await directCronReq(cronState, "cron.add", { name: "patch merge", enabled: true, schedule: { kind: "every", everyMs: 60_000 }, @@ -427,7 +443,7 @@ describe("gateway server cron", () => { const mergeJobId = typeof mergeJobIdValue === "string" ? mergeJobIdValue : ""; expect(mergeJobId.length > 0).toBe(true); - const noTimeoutRes = await rpcReq(ws, "cron.add", { + const noTimeoutRes = await directCronReq(cronState, "cron.add", { name: "no-timeout payload", enabled: true, schedule: { kind: "every", everyMs: 60_000 }, @@ -447,7 +463,7 @@ describe("gateway server cron", () => { expect(noTimeoutPayload?.payload?.kind).toBe("agentTurn"); expect(noTimeoutPayload?.payload?.timeoutSeconds).toBe(0); - const mergeUpdateRes = await rpcReq(ws, "cron.update", { + const mergeUpdateRes = await directCronReq(cronState, "cron.update", { id: mergeJobId, patch: { delivery: { mode: "announce", channel: "telegram", to: "19098680" }, @@ -467,7 +483,7 @@ describe("gateway server cron", () => { expect(merged?.delivery?.channel).toBe("telegram"); expect(merged?.delivery?.to).toBe("19098680"); - const modelOnlyPatchRes = await rpcReq(ws, "cron.update", { + const modelOnlyPatchRes = await directCronReq(cronState, "cron.update", { id: mergeJobId, patch: { payload: { @@ -489,7 +505,7 @@ describe("gateway server cron", () => { expect(modelOnlyPatched?.payload?.message).toBe("hello"); expect(modelOnlyPatched?.payload?.model).toBe("anthropic/claude-sonnet-4-6"); - const deliveryPatchRes = await rpcReq(ws, "cron.update", { + const deliveryPatchRes = await directCronReq(cronState, "cron.update", { id: mergeJobId, patch: { delivery: { @@ -514,9 +530,12 @@ describe("gateway server cron", () => { expect(deliveryPatched?.delivery?.to).toBe("+15550001111"); expect(deliveryPatched?.delivery?.bestEffort).toBe(true); - const rejectJobId = await addMainSystemEventCronJob({ ws, name: "patch reject" }); + const rejectJobId = await addMainSystemEventCronJobDirect({ + cronState, + name: "patch reject", + }); - const rejectUpdateRes = await rpcReq(ws, "cron.update", { + const rejectUpdateRes = await directCronReq(cronState, "cron.update", { id: rejectJobId, patch: { payload: { kind: "agentTurn", message: "nope" }, @@ -524,9 +543,12 @@ describe("gateway server cron", () => { }); expect(rejectUpdateRes.ok).toBe(false); - const jobId = await addMainSystemEventCronJob({ ws, name: "jobId test" }); + const jobId = await addMainSystemEventCronJobDirect({ + cronState, + name: "jobId test", + }); - const jobIdUpdateRes = await rpcReq(ws, "cron.update", { + const jobIdUpdateRes = await directCronReq(cronState, "cron.update", { jobId, patch: { schedule: { at: new Date(Date.now() + 2_000).toISOString() }, @@ -535,9 +557,12 @@ describe("gateway server cron", () => { }); expect(jobIdUpdateRes.ok).toBe(true); - const disableJobId = await addMainSystemEventCronJob({ ws, name: "disable test" }); + const disableJobId = await addMainSystemEventCronJobDirect({ + cronState, + name: "disable test", + }); - const disableUpdateRes = await rpcReq(ws, "cron.update", { + const disableUpdateRes = await directCronReq(cronState, "cron.update", { id: disableJobId, patch: { enabled: false }, }); @@ -546,8 +571,7 @@ describe("gateway server cron", () => { expect(disabled?.enabled).toBe(false); } finally { await cleanupCronTestRun({ - ws, - server, + cronState, prevSkipCron, clearSessionConfig: true, }); @@ -619,11 +643,10 @@ describe("gateway server cron", () => { }, }); - const { server, ws } = await startServerWithClient(); - await connectOk(ws); + const cronState = await createDirectCronState(); try { - const addRes = await rpcReq(ws, "cron.add", { + const addRes = await directCronReq(cronState, "cron.add", { name: "main default agent", enabled: true, schedule: { kind: "every", everyMs: 60_000 }, @@ -637,7 +660,7 @@ describe("gateway server cron", () => { const jobId = typeof jobIdValue === "string" ? jobIdValue : ""; expect(jobId.length > 0).toBe(true); - const updateRes = await rpcReq(ws, "cron.update", { + const updateRes = await directCronReq(cronState, "cron.update", { id: jobId, patch: { delivery: { mode: "announce", channel: "telegram", to: "19098680" }, @@ -648,7 +671,7 @@ describe("gateway server cron", () => { const updated = updateRes.payload as { delivery?: unknown } | undefined; expect(updated?.delivery).toBeUndefined(); } finally { - await cleanupCronTestRun({ ws, server, prevSkipCron }); + await cleanupCronTestRun({ cronState, prevSkipCron }); } }); @@ -716,11 +739,10 @@ describe("gateway server cron", () => { }, }); - const { server, ws } = await startServerWithClient(); - await connectOk(ws); + const cronState = await createDirectCronState(); try { - const addRes = await rpcReq(ws, "cron.add", { + const addRes = await directCronReq(cronState, "cron.add", { name: "main default agent drift", enabled: true, schedule: { kind: "every", everyMs: 60_000 }, @@ -734,8 +756,7 @@ describe("gateway server cron", () => { const jobId = typeof jobIdValue === "string" ? jobIdValue : ""; expect(jobId.length > 0).toBe(true); - const { writeConfigFile } = await import("../config/config.js"); - await writeConfigFile({ + await writeCronConfig({ session: { mainKey: "main", }, @@ -749,22 +770,15 @@ describe("gateway server cron", () => { }, }); - let agentIds: string[] = []; - for (let i = 0; i < 20; i += 1) { - const agentsRes = await rpcReq<{ agents?: Array<{ id?: string }> }>(ws, "agents.list", {}); - expect(agentsRes.ok).toBe(true); - agentIds = (agentsRes.payload?.agents ?? []) - .map((agent) => agent.id) - .filter((id): id is string => typeof id === "string"); - if (agentIds.includes("main") && agentIds.includes("ops")) { - break; - } - await yieldToEventLoop(); - } + const agentIds = + cronState + .getRuntimeConfig() + .agents?.list?.map((agent) => agent.id) + .filter((id): id is string => typeof id === "string") ?? []; expect(agentIds).toContain("main"); expect(agentIds).toContain("ops"); - const updateRes = await rpcReq(ws, "cron.update", { + const updateRes = await directCronReq(cronState, "cron.update", { id: jobId, patch: { delivery: { mode: "announce", channel: "telegram", to: "19098680" }, @@ -776,7 +790,7 @@ describe("gateway server cron", () => { } expect(updateRes.ok).toBe(true); } finally { - await cleanupCronTestRun({ ws, server, prevSkipCron }); + await cleanupCronTestRun({ cronState, prevSkipCron }); } }); diff --git a/src/gateway/test-helpers.server.ts b/src/gateway/test-helpers.server.ts index 8a89ed594ce..e3002bbf38d 100644 --- a/src/gateway/test-helpers.server.ts +++ b/src/gateway/test-helpers.server.ts @@ -555,21 +555,28 @@ export function onceMessage( timeoutMs = 10_000, ): Promise { return new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error("timeout")), timeoutMs); - const closeHandler = (code: number, reason: Buffer) => { + let timer: ReturnType; + function cleanup() { clearTimeout(timer); ws.off("message", handler); + ws.off("close", closeHandler); + } + function closeHandler(code: number, reason: Buffer) { + cleanup(); reject(new Error(`closed ${code}: ${reason.toString()}`)); - }; - const handler = (data: WebSocket.RawData) => { + } + function handler(data: WebSocket.RawData) { const obj = JSON.parse(rawDataToString(data)) as T; if (filter(obj)) { - clearTimeout(timer); - ws.off("message", handler); - ws.off("close", closeHandler); + cleanup(); resolve(obj); } - }; + } + timer = setTimeout(() => { + cleanup(); + reject(new Error("timeout")); + }, timeoutMs); + timer.unref?.(); ws.on("message", handler); ws.once("close", closeHandler); });