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.
2026-05-06 06:20:43 +00:00 · 2026-04-27 22:26:41 -07:00
parent 2c58c5d4ec
commit 9577703249
2 changed files with 100 additions and 79 deletions
--- 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 });
    }
  });

--- a/src/gateway/test-helpers.server.ts
+++ b/src/gateway/test-helpers.server.ts
@@ -555,21 +555,28 @@ export function onceMessage<T extends GatewayTestMessage = GatewayTestMessage>(
  timeoutMs = 10_000,
 ): Promise<T> {
  return new Promise<T>((resolve, reject) => {
-    const timer = setTimeout(() => reject(new Error("timeout")), timeoutMs);
-    const closeHandler = (code: number, reason: Buffer) => {
+    let timer: ReturnType<typeof setTimeout>;
+    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);
  });