diff --git a/scripts/bench-gateway-startup.ts b/scripts/bench-gateway-startup.ts index e7e928adf01..2aa2f101246 100644 --- a/scripts/bench-gateway-startup.ts +++ b/scripts/bench-gateway-startup.ts @@ -15,6 +15,7 @@ import { readProcessTreeCpuMs, requestProbeStatus, } from "./lib/gateway-bench-probes.ts"; +import { selectSlowStartupTraceDurations } from "./lib/gateway-startup-trace-ranking.js"; type GatewayBenchCase = { config: Record; @@ -540,7 +541,7 @@ function formatStats(stats: SummaryStats | null): string { return `p50=${formatMs(stats.p50)} avg=${formatMs(stats.avg)} min=${formatMs(stats.min)} max=${formatMs(stats.max)}`; } -function formatMemoryStats(stats: SummaryStats | null): string { +function formatMemoryStats(stats: SummaryStats | null | undefined): string { if (!stats) { return "n/a"; } @@ -554,13 +555,6 @@ function formatRatioStats(stats: SummaryStats | null): string { return `p50=${formatRatio(stats.p50)} avg=${formatRatio(stats.avg)} min=${formatRatio(stats.min)} max=${formatRatio(stats.max)}`; } -function getStartupTraceStat( - startupTrace: Record, - key: string, -): SummaryStats | null { - return startupTrace[key] ?? null; -} - async function waitForProbe(params: { deadlineAt: number; isDone?: () => boolean; @@ -939,15 +933,12 @@ function printResult(result: CaseResult): void { console.log(` /readyz: ${formatStats(result.summary.readyzMs)}`); console.log(` max RSS: ${formatMemoryStats(result.summary.maxRssMb)}`); console.log( - ` ready memory: rss=${formatMemoryStats(getStartupTraceStat(result.summary.startupTrace, "memory.ready.rssMb"))} heap=${formatMemoryStats(getStartupTraceStat(result.summary.startupTrace, "memory.ready.heapUsedMb"))} external=${formatMemoryStats(getStartupTraceStat(result.summary.startupTrace, "memory.ready.externalMb"))}`, + ` ready memory: rss=${formatMemoryStats(result.summary.startupTrace["memory.ready.rssMb"])} heap=${formatMemoryStats(result.summary.startupTrace["memory.ready.heapUsedMb"])} external=${formatMemoryStats(result.summary.startupTrace["memory.ready.externalMb"])}`, ); console.log( - ` post-ready memory: rss=${formatMemoryStats(getStartupTraceStat(result.summary.startupTrace, "memory.post-ready.rssMb"))} heap=${formatMemoryStats(getStartupTraceStat(result.summary.startupTrace, "memory.post-ready.heapUsedMb"))} external=${formatMemoryStats(getStartupTraceStat(result.summary.startupTrace, "memory.post-ready.externalMb"))}`, + ` post-ready memory: rss=${formatMemoryStats(result.summary.startupTrace["memory.post-ready.rssMb"])} heap=${formatMemoryStats(result.summary.startupTrace["memory.post-ready.heapUsedMb"])} external=${formatMemoryStats(result.summary.startupTrace["memory.post-ready.externalMb"])}`, ); - const trace = Object.entries(result.summary.startupTrace) - .filter(([name]) => !name.endsWith(".total") && !name.startsWith("memory.")) - .toSorted((a, b) => (b[1].avg ?? 0) - (a[1].avg ?? 0)) - .slice(0, 8); + const trace = selectSlowStartupTraceDurations(result.summary.startupTrace, 8); if (trace.length > 0) { console.log(" trace top:"); for (const [name, stats] of trace) { diff --git a/scripts/deadcode-exports.baseline.mjs b/scripts/deadcode-exports.baseline.mjs index df4c5d8c6fd..038c9e4fcbb 100644 --- a/scripts/deadcode-exports.baseline.mjs +++ b/scripts/deadcode-exports.baseline.mjs @@ -1534,6 +1534,7 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [ "extensions/sms/src/twilio.ts: parseTwilioFormBody", "extensions/sms/src/twilio.ts: TwilioSmsApiError", "extensions/sms/src/webhook.ts: createSmsWebhookReplayGuard", + "extensions/sms/src/webhook.ts: resetSmsWebhookRateLimiterForTest", "extensions/sms/src/webhook.ts: resetSmsWebhookReplayGuardsForTest", "extensions/synology-chat/src/channel.ts: createSynologyChatPlugin", "extensions/synology-chat/src/channel.ts: synologyChatMessageAdapter", @@ -3634,7 +3635,6 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [ "src/gateway/session-utils.fs.ts: archiveSessionTranscripts", "src/gateway/session-utils.fs.ts: cleanupArchivedSessionTranscripts", "src/gateway/session-utils.fs.ts: ReadSessionMessagesPageOptions", - "src/gateway/session-utils.fs.ts: resolveSessionTranscriptResetArchiveCandidatesAsync", "src/gateway/session-utils.types.ts: SessionCompactionCheckpointPreview", "src/gateway/startup-auth.ts: assertGatewayAuthNotKnownWeak", "src/gateway/talk-handoff.ts: clearTalkHandoffsForTest", @@ -3686,6 +3686,7 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [ "src/gateway/worker-environments/store.ts: WorkerEnvironmentTeardownTerminalState", "src/gateway/worker-environments/transcript-commit-store.ts: WorkerTranscriptCommitBeginResult", "src/gateway/worker-environments/transcript-commit.ts: WorkerTranscriptCommitter", + "src/gateway/worker-environments/tunnel.ts: createWorkerSshRunner", "src/gateway/ws-log.ts: shortId", "src/hooks/bundled/session-memory/transcript.ts: getRecentSessionContent", "src/hooks/bundled/session-memory/transcript.ts: sanitizeSessionMemoryTranscriptText", diff --git a/scripts/lib/gateway-startup-trace-ranking.ts b/scripts/lib/gateway-startup-trace-ranking.ts new file mode 100644 index 00000000000..d66d40572c6 --- /dev/null +++ b/scripts/lib/gateway-startup-trace-ranking.ts @@ -0,0 +1,17 @@ +export function isStartupTraceDuration(name: string): boolean { + if (name.endsWith(".total") || name.startsWith("memory.")) { + return false; + } + const metricName = name.slice(name.lastIndexOf(".") + 1); + return !metricName.endsWith("Count") && !metricName.endsWith("Mb"); +} + +export function selectSlowStartupTraceDurations( + trace: Record, + limit: number, +): Array<[string, T]> { + return Object.entries(trace) + .filter(([name]) => isStartupTraceDuration(name)) + .toSorted((left, right) => (right[1].avg ?? 0) - (left[1].avg ?? 0)) + .slice(0, limit); +} diff --git a/scripts/ts-max-loc-baseline-v2.json b/scripts/ts-max-loc-baseline-v2.json index cb28e63d362..dc9d2bc0024 100644 --- a/scripts/ts-max-loc-baseline-v2.json +++ b/scripts/ts-max-loc-baseline-v2.json @@ -453,7 +453,7 @@ "scripts/apple-app-i18n.ts": 965, "scripts/bench-cli-startup.ts": 1316, "scripts/bench-gateway-restart.ts": 1697, - "scripts/bench-gateway-startup.ts": 1035, + "scripts/bench-gateway-startup.ts": 1026, "scripts/bench-sqlite-state.ts": 674, "scripts/control-ui-i18n.ts": 1749, "scripts/control-ui-mock-dev.ts": 1401, @@ -933,7 +933,7 @@ "src/gateway/server-restart-sentinel.ts": 674, "src/gateway/server-startup-config.ts": 624, "src/gateway/server-startup-post-attach.ts": 1471, - "src/gateway/server.impl.ts": 2370, + "src/gateway/server.impl.ts": 2271, "src/gateway/server/ws-connection.ts": 699, "src/gateway/server/ws-connection/message-handler.ts": 2866, "src/gateway/server/ws-connection/worker-connection.ts": 616, @@ -955,7 +955,7 @@ "src/gateway/worker-environments/live-events.ts": 778, "src/gateway/worker-environments/service.ts": 1270, "src/gateway/worker-environments/store.ts": 888, - "src/gateway/worker-environments/tunnel.ts": 558, + "src/gateway/worker-environments/tunnel.ts": 553, "src/hooks/install.ts": 775, "src/hooks/message-hook-mappers.ts": 614, "src/infra/agent-events.ts": 741, diff --git a/src/gateway/server-import-boundary.test.ts b/src/gateway/server-import-boundary.test.ts index 932d3faf6a7..e1757d8424a 100644 --- a/src/gateway/server-import-boundary.test.ts +++ b/src/gateway/server-import-boundary.test.ts @@ -55,30 +55,65 @@ describe("gateway startup import boundaries", () => { ); expect(validation).not.toContain("legacy-secretref-env-marker"); expect(validation).not.toContain("commands/doctor"); + const workerStartup = readSource("src/gateway/server-worker-environment-startup.ts"); + expect(serverImpl).toContain('import("./server-worker-environment-startup.js")'); + for (const workerModule of ["live-events", "service", "store", "transcript-commit"]) { + expect(serverImpl).not.toContain(`from "./worker-environments/${workerModule}.js"`); + expect(workerStartup).toContain(`import("./worker-environments/${workerModule}.js")`); + } + expect(serverImpl).not.toContain('from "../plugins/worker-provider-registry.js"'); + expect(workerStartup).toContain('import("../plugins/worker-provider-registry.js")'); + expect(serverImpl).not.toContain( + 'from "../../packages/gateway-protocol/src/schema/worker-admission.js"', + ); + expect(workerStartup).toContain( + 'import("../../packages/gateway-protocol/src/schema/worker-admission.js")', + ); + }); + + it("defers retained plugin generation cleanup to the post-ready idle scheduler", () => { + const serverImpl = readSource("src/gateway/server.impl.ts"); + const cleanup = readSource("src/gateway/server-retained-plugin-cleanup.ts"); + const importBoundary = serverImpl.indexOf("type LoadGatewayModelCatalog"); + const serverStart = serverImpl.indexOf("export async function startGatewayServer"); + const postReadyStart = serverImpl.indexOf("scheduleGatewayPostReadyMaintenance({", serverStart); + const cleanupCall = serverImpl.lastIndexOf("cleanupRetainedPluginInstallGenerations("); + + expect(importBoundary).toBeGreaterThan(-1); + expect(serverImpl.slice(0, importBoundary)).not.toContain("managed-npm-retention"); + expect(serverImpl.slice(0, importBoundary)).not.toContain("installed-plugin-index-records"); + expect(cleanup).toContain('import("../plugins/managed-npm-retention.js")'); + expect(cleanup).toContain('import("../plugins/installed-plugin-index-records.js")'); + expect(postReadyStart).toBeGreaterThan(serverStart); + expect(cleanupCall).toBeGreaterThan(postReadyStart); + expect(serverImpl.slice(postReadyStart, cleanupCall + 300)).not.toContain( + "startupConfigLoad.pluginMetadataSnapshot?.index.installRecords", + ); + expect(cleanup).toContain("loadInstalledPluginIndexInstallRecordsSync()"); }); it("loads the worker bootstrap runtime only when an operation needs it", () => { - const serverImpl = readSource("src/gateway/server.impl.ts"); - const runtimeLoad = "await loadWorkerEnvironmentRuntimeModule()"; - const prepareStart = serverImpl.indexOf("const prepareWorkerInstallation = async"); - const serviceStart = serverImpl.indexOf("const workerEnvironmentService =", prepareStart); - const identityStart = serverImpl.indexOf("resolveSshIdentity: async", serviceStart); - const bootstrapStart = serverImpl.indexOf("bootstrapWorker: async", serviceStart); - const loggerStart = serverImpl.indexOf("logger: log.child", bootstrapStart); + const workerStartup = readSource("src/gateway/server-worker-environment-startup.ts"); + const runtimeLoad = "loadWorkerEnvironmentRuntimeModule()"; + const prepareStart = workerStartup.indexOf("const prepareInstallation = async"); + const serviceStart = workerStartup.indexOf("const workerEnvironmentService =", prepareStart); + const identityStart = workerStartup.indexOf("resolveSshIdentity: async", serviceStart); + const bootstrapStart = workerStartup.indexOf("bootstrapWorker: async", serviceStart); + const loggerStart = workerStartup.indexOf("logger: params.log.child", bootstrapStart); expect(prepareStart).toBeGreaterThan(-1); expect(serviceStart).toBeGreaterThan(prepareStart); expect(identityStart).toBeGreaterThan(serviceStart); expect(bootstrapStart).toBeGreaterThan(serviceStart); expect(loggerStart).toBeGreaterThan(bootstrapStart); - expect(serverImpl.slice(0, prepareStart)).not.toContain(runtimeLoad); - expect(serverImpl.slice(prepareStart, serviceStart)).toContain(runtimeLoad); - expect(serverImpl.slice(identityStart, bootstrapStart)).toContain(runtimeLoad); - expect(serverImpl.slice(bootstrapStart, loggerStart)).toContain(runtimeLoad); - expect(serverImpl.slice(bootstrapStart, loggerStart)).toContain( + expect(workerStartup.slice(0, prepareStart)).not.toContain(runtimeLoad); + expect(workerStartup.slice(prepareStart, serviceStart)).toContain(runtimeLoad); + expect(workerStartup.slice(identityStart, bootstrapStart)).toContain(runtimeLoad); + expect(workerStartup.slice(bootstrapStart, loggerStart)).toContain(runtimeLoad); + expect(workerStartup.slice(bootstrapStart, loggerStart)).toContain( "pinnedHostKey: sshEndpoint.hostKey", ); - expect(serverImpl.match(/await loadWorkerEnvironmentRuntimeModule\(\)/gu)).toHaveLength(3); + expect(workerStartup.match(/loadWorkerEnvironmentRuntimeModule\(\)/gu)).toHaveLength(3); }); it("fences config reload before gateway teardown and gateway_stop hooks", () => { diff --git a/src/gateway/server-methods/environments.test.ts b/src/gateway/server-methods/environments.test.ts index 8dfe4f143c7..872afea49a5 100644 --- a/src/gateway/server-methods/environments.test.ts +++ b/src/gateway/server-methods/environments.test.ts @@ -6,7 +6,7 @@ import { ErrorCodes } from "../../../packages/gateway-protocol/src/index.js"; import { listDevicePairing } from "../../infra/device-pairing.js"; import { listNodePairing } from "../../infra/node-pairing.js"; import type { WorkerEnvironmentRecord } from "../worker-environments/store.js"; -import type { WorkerTunnelStatus } from "../worker-environments/tunnel.js"; +import type { WorkerTunnelStatus } from "../worker-environments/tunnel-contract.js"; import { environmentsHandlers, summarizeWorkerEnvironment } from "./environments.js"; vi.mock("../../infra/device-pairing.js", () => ({ diff --git a/src/gateway/server-retained-plugin-cleanup.ts b/src/gateway/server-retained-plugin-cleanup.ts new file mode 100644 index 00000000000..88cbcf84257 --- /dev/null +++ b/src/gateway/server-retained-plugin-cleanup.ts @@ -0,0 +1,30 @@ +type RetainedPluginCleanupLogger = { + info: (message: string) => void; + warn: (message: string) => void; +}; + +export async function cleanupRetainedPluginInstallGenerations(params: { + log: RetainedPluginCleanupLogger; +}): Promise { + try { + // The idle delay spans plugin installs and reloads; protect the current paths, + // not the startup snapshot, before deleting any retained generation. + const records = ( + await import("../plugins/installed-plugin-index-records.js") + ).loadInstalledPluginIndexInstallRecordsSync(); + const { cleanupRetainedManagedNpmInstallGenerations } = + await import("../plugins/managed-npm-retention.js"); + const removedGenerations = await cleanupRetainedManagedNpmInstallGenerations({ + activeInstallPaths: Object.values(records).flatMap((record) => + record.installPath ? [record.installPath] : [], + ), + onError: (error, projectRoot) => + params.log.warn(`failed to clean retained npm generation ${projectRoot}: ${String(error)}`), + }); + if (removedGenerations > 0) { + params.log.info(`cleaned ${removedGenerations} retained npm plugin generation(s)`); + } + } catch (error) { + params.log.warn(`retained npm generation cleanup unavailable: ${String(error)}`); + } +} diff --git a/src/gateway/server-runtime-services.test.ts b/src/gateway/server-runtime-services.test.ts index 1ff5ee40161..7008fc9cd38 100644 --- a/src/gateway/server-runtime-services.test.ts +++ b/src/gateway/server-runtime-services.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { getActiveGatewayRootWorkCount, resetGatewayWorkAdmission, + tryBeginGatewayRootWorkAdmission, } from "../process/gateway-work-admission.js"; const hoisted = vi.hoisted(() => { @@ -67,6 +68,7 @@ vi.mock("./model-pricing-cache.js", () => ({ const { activateGatewayScheduledServices, runGatewayPostReadyMaintenance, + scheduleGatewayIdleTask, scheduleGatewayPostReadyMaintenance, startGatewayCronWithLogging, startGatewayRuntimeServices, @@ -375,6 +377,75 @@ describe("server-runtime-services", () => { expect(startMaintenance).not.toHaveBeenCalled(); }); + it("runs a scheduled idle task in an independent admitted root", async () => { + vi.useFakeTimers(); + const activeRootCounts: number[] = []; + const run = vi.fn(async () => { + activeRootCounts.push(getActiveGatewayRootWorkCount()); + }); + + scheduleGatewayIdleTask({ + delayMs: 25, + retryDelayMs: 50, + isClosing: () => false, + isBusy: () => getActiveGatewayRootWorkCount({ excludeCurrent: true }) > 0, + run, + log: createLog(), + errorMessage: "idle task failed", + }); + + await vi.advanceTimersByTimeAsync(25); + await vi.waitFor(() => expect(run).toHaveBeenCalledOnce()); + expect(activeRootCounts).toEqual([1]); + await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + }); + + it("retries a scheduled idle task while request work is active", async () => { + vi.useFakeTimers(); + const admission = tryBeginGatewayRootWorkAdmission(); + if (!admission) { + throw new Error("Expected request work admission"); + } + const run = vi.fn(async () => undefined); + + scheduleGatewayIdleTask({ + delayMs: 25, + retryDelayMs: 50, + isClosing: () => false, + isBusy: () => getActiveGatewayRootWorkCount({ excludeCurrent: true }) > 0, + run, + log: createLog(), + errorMessage: "idle task failed", + }); + + await vi.advanceTimersByTimeAsync(25); + expect(run).not.toHaveBeenCalled(); + admission.release(); + await vi.advanceTimersByTimeAsync(49); + expect(run).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(() => expect(run).toHaveBeenCalledOnce()); + }); + + it("cancels a scheduled idle task before its delay elapses", async () => { + vi.useFakeTimers(); + const run = vi.fn(async () => undefined); + const handle = scheduleGatewayIdleTask({ + delayMs: 25, + retryDelayMs: 50, + isClosing: () => false, + isBusy: () => false, + run, + log: createLog(), + errorMessage: "idle task failed", + }); + + handle.stop(); + await vi.advanceTimersByTimeAsync(25); + + expect(run).not.toHaveBeenCalled(); + }); + it("clears delayed maintenance handles when close starts during maintenance startup", async () => { vi.useFakeTimers(); let closing = false; diff --git a/src/gateway/server-runtime-services.ts b/src/gateway/server-runtime-services.ts index 529ea9538f5..b2a985d69cd 100644 --- a/src/gateway/server-runtime-services.ts +++ b/src/gateway/server-runtime-services.ts @@ -23,6 +23,9 @@ export { type GatewayPostReadyLogger = { warn: (message: string) => void; }; +export type GatewayIdleTaskHandle = { + stop: () => void; +}; export type GatewayMaintenanceHandles = NonNullable< Awaited> >; @@ -170,6 +173,58 @@ export function scheduleGatewayPostReadyMaintenance(params: { return timer; } +/** Schedules one low-priority task, retrying until the gateway has no active request roots. */ +export function scheduleGatewayIdleTask(params: { + delayMs: number; + retryDelayMs: number; + isClosing: () => boolean; + isBusy: () => boolean; + run: () => Promise; + log: GatewayPostReadyLogger; + errorMessage: string; +}): GatewayIdleTaskHandle { + let stopped = false; + let timer: ReturnType | null = null; + const schedule = (delayMs: number) => { + if (stopped || params.isClosing()) { + return; + } + timer = setTimeout(() => { + timer = null; + if (stopped || params.isClosing()) { + return; + } + if (params.isBusy()) { + schedule(params.retryDelayMs); + return; + } + void runWithGatewayIndependentRootWorkAdmission(async () => { + if (stopped || params.isClosing()) { + return; + } + // Recheck inside admission so work that arrived while this task was + // joining the root set gets priority over non-urgent maintenance. + if (params.isBusy()) { + schedule(params.retryDelayMs); + return; + } + await params.run(); + }).catch((error: unknown) => params.log.warn(`${params.errorMessage}: ${String(error)}`)); + }, delayMs); + timer.unref?.(); + }; + schedule(params.delayMs); + return { + stop: () => { + stopped = true; + if (timer) { + clearTimeout(timer); + timer = null; + } + }, + }; +} + function recoverPendingOutboundDeliveries(params: { cfg: OpenClawConfig; log: GatewayRuntimeServiceLogger; diff --git a/src/gateway/server-worker-environment-startup.ts b/src/gateway/server-worker-environment-startup.ts new file mode 100644 index 00000000000..bba7cee2807 --- /dev/null +++ b/src/gateway/server-worker-environment-startup.ts @@ -0,0 +1,162 @@ +import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; +import { getRuntimeConfig } from "../config/config.js"; +import type { PluginRegistry } from "../plugins/registry-types.js"; +import { + getActiveSecretsRuntimeConfigSnapshot, + getActiveSecretsRuntimeEnv, +} from "../secrets/runtime-state.js"; +import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; +import type { WorkerBundleProducer, WorkerNpmArtifact } from "./worker-environments/bundle.js"; +import type { WorkerLiveEventReceiver } from "./worker-environments/live-events.js"; +import type { WorkerEnvironmentService } from "./worker-environments/service.js"; + +type WorkerEnvironmentStore = ReturnType< + typeof import("./worker-environments/store.js").createWorkerEnvironmentStore +>; +type WorkerEnvironmentRecord = ReturnType[number]; +type WorkerGatewayEndpoint = { host: "127.0.0.1" | "::1"; port: number } | undefined; +type WorkerEnvironmentLogger = { + child: (name: string) => { warn: (message: string) => void }; +}; + +export type GatewayWorkerEnvironmentStartupState = { + durableProviderIds: string[]; + listDurableProviderIds: () => string[]; + records: WorkerEnvironmentRecord[]; + store: WorkerEnvironmentStore; +}; + +export type GatewayWorkerEnvironmentRuntime = { + workerEnvironmentService?: WorkerEnvironmentService; + workerLiveEvents?: WorkerLiveEventReceiver; +}; + +const loadWorkerEnvironmentRuntimeModule = createLazyRuntimeModule( + () => import("./worker-environments/runtime.js"), +); +const loadWorkerInferenceRuntimeModule = createLazyRuntimeModule( + () => import("./worker-environments/inference-runtime.js"), +); + +export async function loadGatewayWorkerEnvironmentStartupState(): Promise { + const { createWorkerEnvironmentStore } = await import("./worker-environments/store.js"); + const store = createWorkerEnvironmentStore(); + const records = store.list(); + const durableProviderIds = uniqueStrings( + records.flatMap((record) => + record.state === "destroyed" || record.state === "failed" || record.state === "orphaned" + ? [] + : [record.providerId], + ), + ); + const listDurableProviderIds = () => + uniqueStrings(store.listForReconcile().map((record) => record.providerId)); + return { durableProviderIds, listDurableProviderIds, records, store }; +} + +export async function createGatewayWorkerEnvironmentRuntime(params: { + getPluginRegistry: () => Pick; + resolveWorkerGateway: () => WorkerGatewayEndpoint; + startup: GatewayWorkerEnvironmentStartupState; + log: WorkerEnvironmentLogger; +}): Promise { + const [ + { createWorkerEnvironmentService }, + { createWorkerLiveEventReceiver }, + { createWorkerTranscriptCommitter }, + { createWorkerTunnelManager }, + { resolveWorkerProvider }, + ] = await Promise.all([ + import("./worker-environments/service.js"), + import("./worker-environments/live-events.js"), + import("./worker-environments/transcript-commit.js"), + import("./worker-environments/tunnel.js"), + import("../plugins/worker-provider-registry.js"), + ]); + let workerBundleProducer: WorkerBundleProducer | undefined; + let workerNpmArtifact: Promise | undefined; + const prepareInstallation = async (install: "bundle" | "npm") => { + const [workerRuntime, { WORKER_PROTOCOL_FEATURES }] = await Promise.all([ + loadWorkerEnvironmentRuntimeModule(), + import("../../packages/gateway-protocol/src/schema/worker-admission.js"), + ]); + workerBundleProducer ??= workerRuntime.createWorkerBundleProducer({ + protocolFeatures: WORKER_PROTOCOL_FEATURES, + }); + const bundle = await workerBundleProducer.prepare(); + if (install === "bundle") { + return bundle; + } + workerNpmArtifact ??= workerRuntime + .resolveWorkerNpmInstallationArtifact({ bundle }) + .catch((error: unknown) => { + workerNpmArtifact = undefined; + throw error; + }); + return await workerNpmArtifact; + }; + const startupBindings = params.startup.records.flatMap((record) => + record.state === "attached" && record.attachedSessionIds.length === 1 + ? [ + { + environmentId: record.environmentId, + runEpoch: record.ownerEpoch, + sessionId: record.attachedSessionIds[0]!, + }, + ] + : [], + ); + const workerLiveEvents = createWorkerLiveEventReceiver({ + getConfig: getRuntimeConfig, + startupBindings, + startupOwners: new Map( + startupBindings.map((binding) => [binding.environmentId, binding.runEpoch] as const), + ), + }); + const workerEnvironmentService = createWorkerEnvironmentService({ + store: params.startup.store, + getConfig: getRuntimeConfig, + // Plugin reload replaces the registry object; resolve against the live binding. + resolveProvider: (providerId) => resolveWorkerProvider(params.getPluginRegistry(), providerId), + prepareInstallation, + tunnelManager: createWorkerTunnelManager(), + resolveWorkerGateway: params.resolveWorkerGateway, + applyTranscriptCommit: createWorkerTranscriptCommitter({ + getConfig: getRuntimeConfig, + }).commit, + executeInference: async (inferenceParams) => { + const workerInferenceRuntime = await loadWorkerInferenceRuntimeModule(); + return await workerInferenceRuntime.executeWorkerInference(inferenceParams); + }, + liveEvents: workerLiveEvents, + resolveSshIdentity: async ({ provider, leaseId, profile, keyRef }) => { + const workerRuntime = await loadWorkerEnvironmentRuntimeModule(); + return await workerRuntime.resolveWorkerSshIdentity({ + provider, + leaseId, + profile, + keyRef, + resolveGeneric: async (genericKeyRef) => ({ + kind: "material", + contents: await workerRuntime.resolveSecretRefString(genericKeyRef, { + config: getActiveSecretsRuntimeConfigSnapshot()?.sourceConfig ?? getRuntimeConfig(), + env: getActiveSecretsRuntimeEnv(), + }), + }), + }); + }, + bootstrapWorker: async ({ sshEndpoint, installation, resolveIdentity, signal }) => { + const workerRuntime = await loadWorkerEnvironmentRuntimeModule(); + return await workerRuntime.bootstrapWorker( + { + ssh: sshEndpoint, + artifact: installation, + pinnedHostKey: sshEndpoint.hostKey, + }, + { signal, resolveIdentity }, + ); + }, + logger: params.log.child("worker-environments"), + }); + return { workerEnvironmentService, workerLiveEvents }; +} diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index d762643680b..7ed421633c8 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -3,7 +3,6 @@ import type { IncomingMessage, ServerResponse } from "node:http"; // and WebSocket surfaces, config reload hooks, and graceful restart/shutdown. import { monitorEventLoopDelay, performance } from "node:perf_hooks"; import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; -import { WORKER_PROTOCOL_FEATURES } from "../../packages/gateway-protocol/src/schema/worker-admission.js"; import { getActiveBackgroundExecSessionCount } from "../agents/bash-process-registry.js"; import { getActiveEmbeddedRunCount, @@ -65,21 +64,17 @@ import { startDiagnosticHeartbeat, stopDiagnosticHeartbeat } from "../logging/di import { createSubsystemLogger, runtimeForLogger } from "../logging/subsystem.js"; import { setCurrentPluginMetadataSnapshot } from "../plugins/current-plugin-metadata-snapshot.js"; import type { PluginHookGatewayCronService } from "../plugins/hook-types.js"; -import { loadInstalledPluginIndexInstallRecordsSync } from "../plugins/installed-plugin-index-records.js"; -import { cleanupRetainedManagedNpmInstallGenerations } from "../plugins/managed-npm-retention.js"; import { clearPluginMetadataLifecycleCaches } from "../plugins/plugin-metadata-lifecycle.js"; import { pinActivePluginChannelRegistry, pinActivePluginHttpRouteRegistry, pinActivePluginSessionExtensionRegistry, } from "../plugins/runtime.js"; -import { resolveWorkerProvider } from "../plugins/worker-provider-registry.js"; import { getTotalQueueSize, isGatewayDraining } from "../process/command-queue.js"; import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js"; import type { RuntimeEnv } from "../runtime.js"; import { clearSecretsRuntimeSnapshot, - getActiveSecretsRuntimeEnv, getActiveSecretsRuntimeConfigSnapshot, } from "../secrets/runtime-state.js"; import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; @@ -157,11 +152,6 @@ import { loadGatewayTlsRuntime } from "./server/tls.js"; import { resolveSharedGatewaySessionGeneration } from "./server/ws-shared-generation.js"; import { mergeGatewayAuthConfig, mergeGatewayTailscaleConfig } from "./startup-auth.js"; import { maybeSeedControlUiAllowedOriginsAtStartup } from "./startup-control-ui-origins.js"; -import type { WorkerBundleProducer, WorkerNpmArtifact } from "./worker-environments/bundle.js"; -import { createWorkerLiveEventReceiver } from "./worker-environments/live-events.js"; -import { createWorkerEnvironmentService } from "./worker-environments/service.js"; -import { createWorkerEnvironmentStore } from "./worker-environments/store.js"; -import { createWorkerTranscriptCommitter } from "./worker-environments/transcript-commit.js"; type LoadGatewayModelCatalog = typeof import("./server-model-catalog.js").loadGatewayModelCatalog; type LoadGatewayModelCatalogSnapshot = @@ -170,14 +160,8 @@ type LoadGatewayModelCatalogSnapshot = const loadGatewayModelCatalogModule = createLazyRuntimeModule( () => import("./server-model-catalog.js"), ); -const loadWorkerEnvironmentRuntimeModule = createLazyRuntimeModule( - () => import("./worker-environments/runtime.js"), -); -const loadWorkerInferenceRuntimeModule = createLazyRuntimeModule( - () => import("./worker-environments/inference-runtime.js"), -); -const loadWorkerTunnelRuntimeModule = createLazyRuntimeModule( - () => import("./worker-environments/tunnel.js"), +const loadWorkerEnvironmentStartupModule = createLazyRuntimeModule( + () => import("./server-worker-environment-startup.js"), ); export async function resetModelCatalogCacheForTest(): Promise { @@ -190,6 +174,7 @@ ensureOpenClawCliOnPath(); const MAX_MEDIA_TTL_HOURS = 24 * 7; const POST_READY_MAINTENANCE_DELAY_MS = 250; +const RETAINED_PLUGIN_CLEANUP_DELAY_MS = 30_000; type GatewayStartupChannelPlugin = { id: ChannelId; @@ -583,23 +568,6 @@ export async function startGatewayServer( opts: GatewayServerOptions = {}, ): Promise { normalizeStateDirEnv(process.env); - // runGatewayLoop calls this after closing the previous server on both fresh - // and in-process restarts, making retired plugin generations safe to remove. - try { - const installRecords = loadInstalledPluginIndexInstallRecordsSync(); - const removedGenerations = await cleanupRetainedManagedNpmInstallGenerations({ - activeInstallPaths: Object.values(installRecords).flatMap((record) => - record.installPath ? [record.installPath] : [], - ), - onError: (error, projectRoot) => - log.warn(`failed to clean retained npm generation ${projectRoot}: ${String(error)}`), - }); - if (removedGenerations > 0) { - log.info(`cleaned ${removedGenerations} retained npm plugin generation(s)`); - } - } catch (error) { - log.warn(`retained npm generation cleanup unavailable: ${String(error)}`); - } const { bootstrapGatewayNetworkRuntime } = await import("./server-network-runtime.js"); bootstrapGatewayNetworkRuntime(); @@ -876,13 +844,12 @@ export async function startGatewayServer( ), preserveExistingOwnership: true, }); - const workerEnvironmentStore = minimalTestGateway ? undefined : createWorkerEnvironmentStore(); - const hasWorkerEnvironmentRecords = (workerEnvironmentStore?.list().length ?? 0) > 0; - // Durable rows can outlive profiles. Startup planning still enforces plugin trust/disable gates. - const listDurableWorkerProviderIds = () => - uniqueStrings( - workerEnvironmentStore?.listForReconcile().map((record) => record.providerId) ?? [], - ); + const workerEnvironmentStartup = minimalTestGateway + ? undefined + : await startupTrace.measure("worker-environments.store-import", async () => { + const workerModule = await loadWorkerEnvironmentStartupModule(); + return await workerModule.loadGatewayWorkerEnvironmentStartupState(); + }); const { prepareGatewayPluginBootstrap } = await loadStartupPluginsModule(); const pluginBootstrap = await startupTrace.measure("plugins.bootstrap", () => prepareGatewayPluginBootstrap({ @@ -890,7 +857,7 @@ export async function startGatewayServer( activationSourceConfig: startupActivationSourceConfig, startupRuntimeConfig, pluginMetadataSnapshot: startupConfigLoad.pluginMetadataSnapshot, - workerProviderIds: listDurableWorkerProviderIds(), + workerProviderIds: workerEnvironmentStartup?.durableProviderIds ?? [], minimalTestGateway, log, loadRuntimePlugins: false, @@ -935,109 +902,23 @@ export async function startGatewayServer( // Unconfigured clean installs get no service; durable rows still need list/status projection. const shouldStartWorkerEnvironmentService = Object.keys(gatewayPluginConfigAtStart.cloudWorkers?.profiles ?? {}).length > 0 || - hasWorkerEnvironmentRecords; - let workerBundleProducer: WorkerBundleProducer | undefined; - let workerNpmArtifact: Promise | undefined; + Boolean(workerEnvironmentStartup?.records.length); let resolveWorkerGatewayEndpoint: () => | { host: "127.0.0.1" | "::1"; port: number } | undefined = () => undefined; - const prepareWorkerInstallation = async (install: "bundle" | "npm") => { - const workerEnvironmentRuntime = await loadWorkerEnvironmentRuntimeModule(); - workerBundleProducer ??= workerEnvironmentRuntime.createWorkerBundleProducer({ - protocolFeatures: WORKER_PROTOCOL_FEATURES, - }); - const bundle = await workerBundleProducer.prepare(); - if (install === "bundle") { - return bundle; - } - workerNpmArtifact ??= workerEnvironmentRuntime - .resolveWorkerNpmInstallationArtifact({ bundle }) - .catch((error: unknown) => { - workerNpmArtifact = undefined; - throw error; - }); - return await workerNpmArtifact; - }; - const workerTunnelManager = - workerEnvironmentStore && shouldStartWorkerEnvironmentService - ? (await loadWorkerTunnelRuntimeModule()).createWorkerTunnelManager() - : undefined; - // Freeze restart-owned identities before any new attach can advance its environment. - // Only an exact pre-start owner may seed an ephemeral ACK after gateway state loss. - const workerEnvironmentRecordsAtStartup = workerEnvironmentStore?.list() ?? []; - const workerStartupBindings = workerEnvironmentRecordsAtStartup.flatMap((record) => - record.state === "attached" && record.attachedSessionIds.length === 1 - ? [ - { - environmentId: record.environmentId, - runEpoch: record.ownerEpoch, - sessionId: record.attachedSessionIds[0]!, - }, - ] - : [], - ); - const workerStartupOwners = new Map( - workerStartupBindings.map((binding) => [binding.environmentId, binding.runEpoch] as const), - ); - const workerLiveEvents = - workerEnvironmentStore && shouldStartWorkerEnvironmentService - ? createWorkerLiveEventReceiver({ - getConfig: getRuntimeConfig, - startupBindings: workerStartupBindings, - startupOwners: workerStartupOwners, + const workerEnvironmentRuntime = + workerEnvironmentStartup && shouldStartWorkerEnvironmentService + ? await startupTrace.measure("worker-environments.runtime-imports", async () => { + const workerModule = await loadWorkerEnvironmentStartupModule(); + return await workerModule.createGatewayWorkerEnvironmentRuntime({ + getPluginRegistry: () => pluginRegistry, + resolveWorkerGateway: () => resolveWorkerGatewayEndpoint(), + startup: workerEnvironmentStartup, + log, + }); }) - : undefined; - const workerEnvironmentService = - workerEnvironmentStore && shouldStartWorkerEnvironmentService - ? createWorkerEnvironmentService({ - store: workerEnvironmentStore, - getConfig: getRuntimeConfig, - resolveProvider: (providerId) => resolveWorkerProvider(pluginRegistry, providerId), - prepareInstallation: prepareWorkerInstallation, - tunnelManager: workerTunnelManager, - resolveWorkerGateway: () => resolveWorkerGatewayEndpoint(), - applyTranscriptCommit: createWorkerTranscriptCommitter({ - getConfig: getRuntimeConfig, - }).commit, - executeInference: async (params) => { - const workerInferenceRuntime = await loadWorkerInferenceRuntimeModule(); - return await workerInferenceRuntime.executeWorkerInference(params); - }, - ...(workerLiveEvents ? { liveEvents: workerLiveEvents } : {}), - resolveSshIdentity: async ({ provider, leaseId, profile, keyRef }) => { - const workerEnvironmentRuntime = await loadWorkerEnvironmentRuntimeModule(); - return await workerEnvironmentRuntime.resolveWorkerSshIdentity({ - provider, - leaseId, - profile, - keyRef, - resolveGeneric: async (genericKeyRef) => ({ - kind: "material", - contents: await workerEnvironmentRuntime.resolveSecretRefString(genericKeyRef, { - config: - getActiveSecretsRuntimeConfigSnapshot()?.sourceConfig ?? getRuntimeConfig(), - env: getActiveSecretsRuntimeEnv(), - }), - }), - }); - }, - bootstrapWorker: async ({ sshEndpoint, installation, resolveIdentity, signal }) => { - const workerEnvironmentRuntime = await loadWorkerEnvironmentRuntimeModule(); - return await workerEnvironmentRuntime.bootstrapWorker( - { - ssh: sshEndpoint, - artifact: installation, - pinnedHostKey: sshEndpoint.hostKey, - }, - { - signal, - resolveIdentity, - }, - ); - }, - logger: log.child("worker-environments"), - }) - : undefined; + : {}; + const { workerEnvironmentService, workerLiveEvents } = workerEnvironmentRuntime; const channelLogs = Object.fromEntries( listGatewayStartupChannelPlugins().map((plugin) => [plugin.id, logChannels.child(plugin.id)]), ) as Record>; @@ -1369,6 +1250,7 @@ export async function startGatewayServer( }, }); let postReadyMaintenanceTimer: ReturnType | null = null; + let retainedPluginCleanupHandle: { stop: () => void } | null = null; const clearPostReadyMaintenanceTimer = () => { if (!postReadyMaintenanceTimer) { return; @@ -1380,6 +1262,8 @@ export async function startGatewayServer( closePreludeStarted = true; cronReconciliation.invalidate(); clearPostReadyMaintenanceTimer(); + retainedPluginCleanupHandle?.stop(); + retainedPluginCleanupHandle = null; }; let configReloaderStopPromise: Promise | null = null; const stopConfigReloaderForClose = () => { @@ -1796,7 +1680,8 @@ export async function startGatewayServer( workspaceDir: defaultWorkspaceDir, env: params.env, activationSourceConfig: params.nextConfig, - workerProviderIds: listDurableWorkerProviderIds(), + // Workers can be created after startup; reload planning needs the live durable set. + workerProviderIds: workerEnvironmentStartup?.listDurableProviderIds() ?? [], }); const nextStartupPluginIds = new Set(nextPluginLookUpTable.startup.pluginIds); const nextStartupChannelIds = new Set(); @@ -2335,6 +2220,22 @@ export async function startGatewayServer( startupTrace.detail("memory.post-ready", collectGatewayProcessMemoryUsageMb()); }, }); + // The loop closes the previous server before this generation starts, so retired + // plugin installs are safe to remove. Wait for an idle window and resolve current + // install paths at execution time so cleanup cannot remove active code or delay a turn. + retainedPluginCleanupHandle = gatewayRuntimeServices.scheduleGatewayIdleTask({ + delayMs: RETAINED_PLUGIN_CLEANUP_DELAY_MS, + retryDelayMs: RETAINED_PLUGIN_CLEANUP_DELAY_MS, + isClosing: () => closePreludeStarted, + isBusy: () => getActiveGatewayRootWorkCount({ excludeCurrent: true }) > 0, + run: async () => { + const { cleanupRetainedPluginInstallGenerations } = + await import("./server-retained-plugin-cleanup.js"); + await cleanupRetainedPluginInstallGenerations({ log }); + }, + log, + errorMessage: "retained npm generation cleanup failed", + }); } else { startupTrace.detail("memory.post-ready", collectGatewayProcessMemoryUsageMb()); } diff --git a/src/gateway/session-transcript-readers.ts b/src/gateway/session-transcript-readers.ts index 86e714017a5..b1d251b83e8 100644 --- a/src/gateway/session-transcript-readers.ts +++ b/src/gateway/session-transcript-readers.ts @@ -78,7 +78,7 @@ type ReadSessionMessageByIdResult = { found: boolean; }; -export type ResolvedTranscriptReadTarget = { +type ResolvedTranscriptReadTarget = { agentId?: string; sessionFile: string; sessionId: string; diff --git a/src/gateway/worker-environments/tunnel-contract.ts b/src/gateway/worker-environments/tunnel-contract.ts index 48f45400d30..636254b8771 100644 --- a/src/gateway/worker-environments/tunnel-contract.ts +++ b/src/gateway/worker-environments/tunnel-contract.ts @@ -7,7 +7,7 @@ export type WorkerTunnelRequest = { ownerEpoch: number; }; -export type WorkerWorkspaceCommand = { +type WorkerWorkspaceCommand = { argv: readonly string[]; input?: string; timeoutMs?: number; diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index 82f13e1a522..7750d0a1624 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -1,13 +1,12 @@ import { describe, expect, it, vi } from "vitest"; import type { WorkerSshEndpoint } from "../../plugins/types.js"; import type { CommandOptions, SpawnResult } from "../../process/exec.js"; -import { - createWorkerSshRunner, - createWorkerTunnelManager, - type WorkerSshProcess, - type WorkerSshProcessExit, - type WorkerSshRunner, -} from "./tunnel.js"; +import { createWorkerSshRunner, createWorkerTunnelManager } from "./tunnel.js"; + +type WorkerTunnelOptions = NonNullable[0]>; +type WorkerSshRunner = NonNullable; +type WorkerSshProcess = ReturnType; +type WorkerSshProcessExit = Awaited; const HOST_KEY = [["ssh", "ed25519"].join("-"), "AAAA"].join(" "); const SSH: WorkerSshEndpoint = { diff --git a/src/gateway/worker-environments/tunnel.ts b/src/gateway/worker-environments/tunnel.ts index 577f7886ed0..40ec66c0577 100644 --- a/src/gateway/worker-environments/tunnel.ts +++ b/src/gateway/worker-environments/tunnel.ts @@ -23,12 +23,7 @@ import type { WorkerTunnelStatus, } from "./tunnel-contract.js"; -export type { - WorkerTunnelHandle, - WorkerTunnelRequest, - WorkerTunnelStatus, - WorkerWorkspaceCommand, -} from "./tunnel-contract.js"; +export type { WorkerTunnelHandle } from "./tunnel-contract.js"; const READY_MARKER = "OPENCLAW_WORKER_TUNNEL_READY"; const REMOTE_SOCKET_NAME = "gateway.sock"; @@ -68,23 +63,23 @@ rm -f -- "$socket" rmdir -- "$directory" 2>/dev/null || true `; -export type WorkerSshProcessExit = { +type WorkerSshProcessExit = { code: number | null; signal: NodeJS.Signals | null; }; -export type WorkerSshProcess = { +type WorkerSshProcess = { ready: Promise; exited: Promise; stop(): Promise; }; -export type WorkerSshRunner = { +type WorkerSshRunner = { start(argv: string[], options: CommandOptions): WorkerSshProcess; run(argv: string[], options: CommandOptions): Promise; }; -export type WorkerTunnelStartRequest = WorkerTunnelRequest & { +type WorkerTunnelStartRequest = WorkerTunnelRequest & { gateway: { host: "127.0.0.1" | "::1"; port: number }; ssh: WorkerSshEndpoint; resolveIdentity: WorkerSshIdentityResolver; @@ -110,7 +105,7 @@ type TunnelEntry = { workspaceTasks: Set>; }; -export type WorkerTunnelManagerOptions = { +type WorkerTunnelManagerOptions = { runner?: WorkerSshRunner; sleep?: (ms: number, signal?: AbortSignal) => Promise; backoff?: BackoffPolicy; diff --git a/src/plugins/managed-npm-retention.ts b/src/plugins/managed-npm-retention.ts index 3cc45881fd6..e083941b857 100644 --- a/src/plugins/managed-npm-retention.ts +++ b/src/plugins/managed-npm-retention.ts @@ -173,8 +173,8 @@ export async function cleanupRetainedManagedNpmInstallGenerations( onError?: (error: unknown, projectRoot: string) => void; } = {}, ): Promise { - // Callers run this after the previous gateway server has closed and before - // the next one loads plugins, so retired module graphs no longer need these trees. + // Callers run this only after the previous gateway server has closed and preserve + // every active install path, so retired module graphs no longer need these trees. const npmDir = params.npmDir ?? resolveDefaultPluginNpmDir(params.env); const projectsDir = resolvePluginNpmProjectsDir(npmDir); const activeInstallPaths = Array.from(params.activeInstallPaths ?? [], (installPath) => diff --git a/test/scripts/bench-gateway-startup.test.ts b/test/scripts/bench-gateway-startup.test.ts index fd0939acdad..0dfe901a5d8 100644 --- a/test/scripts/bench-gateway-startup.test.ts +++ b/test/scripts/bench-gateway-startup.test.ts @@ -7,6 +7,7 @@ import path from "node:path"; import { performance } from "node:perf_hooks"; import { beforeAll, describe, expect, it } from "vitest"; import { testing } from "../../scripts/bench-gateway-startup.ts"; +import { isStartupTraceDuration } from "../../scripts/lib/gateway-startup-trace-ranking.js"; import { registerStopChildBehaviorTests } from "./bench-gateway-child-test-support.js"; async function listenOnLoopback(handler: RequestListener) { @@ -383,6 +384,15 @@ describe("gateway startup benchmark script", () => { expect(startupTrace["sidecars.acp.runtime-ready.readyCount"]).toBe(1); }); + it("keeps counts and memory metrics out of the slow-duration ranking", () => { + expect(isStartupTraceDuration("plugins.runtime-post-bind")).toBe(true); + expect(isStartupTraceDuration("plugins.gateway-load.loadMs")).toBe(true); + expect(isStartupTraceDuration("ready.eventLoopMax")).toBe(true); + expect(isStartupTraceDuration("plugins.runtime-post-bind.gatewayMethodCount")).toBe(false); + expect(isStartupTraceDuration("memory.ready.rssMb")).toBe(false); + expect(isStartupTraceDuration("ready.total")).toBe(false); + }); + it("records probe state transitions, first error kind, and first recovery", async () => { let calls = 0; const { port, server } = await listenOnLoopback((_req, res) => {