From f20c6dacc34fcc91cfee85a6e2f0d63084fa9a15 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 20 Aug 2026 03:18:39 -0700 Subject: [PATCH] fix(gateway): own plugin reload generations (#126627) --- src/gateway/server-core-runtime.ts | 268 ++++++----- src/gateway/server-lifecycle.ts | 27 +- .../server-plugin-runtime-generation.test.ts | 88 ++++ .../server-plugin-runtime-generation.ts | 81 ++++ src/gateway/server-plugins.lifecycle.test.ts | 90 +++- src/gateway/server-reload-contracts.ts | 6 +- src/gateway/server-reload-handlers.test.ts | 234 +++++++++- src/gateway/server-reload-hot.ts | 14 +- src/gateway/server-startup-finish.ts | 17 +- src/gateway/server-startup-plugins.ts | 14 + .../server-startup-post-attach.test.ts | 189 ++++++++ src/gateway/server-startup-post-attach.ts | 76 ++- src/plugins/http-registry.test.ts | 226 +++++++++ src/plugins/http-registry.ts | 47 +- src/plugins/services.replacement.test.ts | 441 ++++++++++++++++++ src/plugins/services.ts | 240 +++++++--- 16 files changed, 1839 insertions(+), 219 deletions(-) create mode 100644 src/gateway/server-plugin-runtime-generation.test.ts create mode 100644 src/gateway/server-plugin-runtime-generation.ts create mode 100644 src/plugins/services.replacement.test.ts diff --git a/src/gateway/server-core-runtime.ts b/src/gateway/server-core-runtime.ts index 5b9c1f5688d5..6806e6539774 100644 --- a/src/gateway/server-core-runtime.ts +++ b/src/gateway/server-core-runtime.ts @@ -5,7 +5,6 @@ import { } from "../channels/plugins/registry-loaded.js"; import type { ChannelId } from "../channels/plugins/types.public.js"; import { getRuntimeConfig } from "../config/io.js"; -import type { OpenClawConfig } from "../config/types.openclaw.js"; import type { createSubsystemLogger } from "../logging/subsystem.js"; import { setCurrentPluginMetadataSnapshot } from "../plugins/current-plugin-metadata-snapshot.js"; import { completePluginMetadataSnapshot } from "../plugins/plugin-metadata-snapshot.js"; @@ -27,7 +26,11 @@ import { isLoopbackHost } from "./net.js"; import { resolveGatewayStartupPluginActivationConfig } from "./plugin-activation-runtime-config.js"; import type { prepareGatewayLifecycle } from "./server-lifecycle.js"; import type { GatewayRequestHandlers } from "./server-methods/types.js"; -import type { GatewayPluginReloadResult } from "./server-reload-handlers.js"; +import type { GatewayPluginRuntimeClaim } from "./server-plugin-runtime-generation.js"; +import type { + GatewayPluginReloadResult, + GatewayReloadHandlerParams, +} from "./server-reload-contracts.js"; import { getHealthVersion, getPresenceVersion, @@ -43,9 +46,8 @@ type GatewayStartupChannelPlugin = { meta: { aliases?: readonly string[] }; }; -function listGatewayStartupChannelPlugins(): GatewayStartupChannelPlugin[] { - return listLoadedChannelPlugins() as GatewayStartupChannelPlugin[]; -} +const listGatewayStartupChannelPlugins = (): GatewayStartupChannelPlugin[] => + listLoadedChannelPlugins() as GatewayStartupChannelPlugin[]; const MAX_MEDIA_TTL_HOURS = 24 * 7; @@ -465,52 +467,51 @@ export async function startGatewayCoreRuntime(input: { }; const refreshAttachedGatewayDiscovery = async ( nextPluginRegistry: typeof pluginRuntime.registry, + claim: GatewayPluginRuntimeClaim, ) => { if (minimalTestGateway) { return; } try { - const stopPreviousDiscovery = kernel.swapBonjourStop(null); - if (stopPreviousDiscovery) { - try { - await stopPreviousDiscovery(); - } catch (err) { - logDiscovery.warn(`gateway discovery stop failed before plugin refresh: ${String(err)}`); - } + if (!(await claim.waitForUnblocked())) { + return; } + const stopPreviousDiscovery = kernel.swapBonjourStop(null); + await stopPreviousDiscovery?.().catch((err: unknown) => { + logDiscovery.warn(`gateway discovery stop failed before plugin refresh: ${String(err)}`); + }); const { startGatewayPluginDiscovery } = await loadGatewayStartupEarlyModule(); - kernel.swapBonjourStop( - await startGatewayPluginDiscovery({ - minimalTestGateway, - cfgAtStart, - port, - gatewayTls, - gatewayDirectReachable: !isLoopbackHost(bindHost), - tailscaleMode, - logDiscovery, - pluginRegistry: nextPluginRegistry, - }), - ); + if (!(await claim.waitForUnblocked())) { + return; + } + const stopNextDiscovery = await startGatewayPluginDiscovery({ + minimalTestGateway, + cfgAtStart, + port, + gatewayTls, + gatewayDirectReachable: !isLoopbackHost(bindHost), + tailscaleMode, + logDiscovery, + pluginRegistry: nextPluginRegistry, + }); + if ( + !(await claim.waitForUnblocked()) || + !claim.publish(() => kernel.swapBonjourStop(stopNextDiscovery)) + ) { + await stopNextDiscovery?.(); + } } catch (err) { logDiscovery.warn(`gateway discovery refresh failed after plugin load: ${String(err)}`); } }; - const reloadAttachedGatewayPlugins = async (params: { - nextConfig: OpenClawConfig; - changedPaths: readonly string[]; - beforeReplace: ( - channels: ReadonlySet, - accounts?: ReadonlyMap>, - ) => Promise; - commitRuntime: () => Promise; - env: NodeJS.ProcessEnv; - isAborted?: () => boolean; - }): Promise => { + const reloadAttachedGatewayPlugins: GatewayReloadHandlerParams["reloadPlugins"] = async ( + params, + ) => { const [ { loadPluginLookUpTable }, { listAmbientOnlyConfiguredChannelIds }, { prepareGatewayPluginLoad }, - { startPluginServices }, + { startPluginServices, PLUGIN_SERVICE_REPLACEMENT_STOP_TIMEOUT_MS }, { listChannelPluginConfigTargetIds, pluginConfigTargetsChanged }, ] = await Promise.all([ import("../plugins/plugin-lookup-table.js"), @@ -519,6 +520,11 @@ export async function startGatewayCoreRuntime(input: { import("../plugins/services.js"), import("./plugin-channel-reload-targets.js"), ]); + const cancelledReload = (activeChannels: Iterable): GatewayPluginReloadResult => ({ + restartChannels: new Set(), + activeChannels: new Set(activeChannels), + cancelled: true, + }); const listAttachedChannelConfigTargets = () => new Map( listGatewayStartupChannelPlugins().map((plugin) => [ @@ -534,7 +540,7 @@ export async function startGatewayCoreRuntime(input: { const beforeChannelIds = new Set(beforeChannelTargets.keys()); const nextPluginActivationConfig = resolveGatewayStartupPluginActivationConfig({ runtimeConfig: params.nextConfig, - activationSourceConfig: params.nextConfig, + activationSourceConfig: params.sourceConfig, env: params.env, ambientEnvTriggers, }); @@ -542,7 +548,7 @@ export async function startGatewayCoreRuntime(input: { config: nextPluginActivationConfig, workspaceDir: pluginWorkspaceDir, env: params.env, - activationSourceConfig: params.nextConfig, + activationSourceConfig: params.sourceConfig, // Workers can be created after startup; reload planning needs the live durable set. workerProviderIds: workerEnvironmentStartup?.listDurableProviderIds() ?? [], ambientEnvTriggers, @@ -552,7 +558,7 @@ export async function startGatewayCoreRuntime(input: { ? new Set( listAmbientOnlyConfiguredChannelIds({ config: params.nextConfig, - activationSourceConfig: params.nextConfig, + activationSourceConfig: params.sourceConfig, env: params.env, includePersistedAuthState: false, manifestRecords: nextPluginLookUpTable.manifestRegistry.plugins, @@ -560,22 +566,13 @@ export async function startGatewayCoreRuntime(input: { ) : new Set(); const nextStartupPluginIds = new Set(nextPluginLookUpTable.startup.pluginIds); - const nextStartupChannelIds = new Set(); - for (const plugin of nextPluginLookUpTable.manifestRegistry.plugins) { - if (!nextStartupPluginIds.has(plugin.id)) { - continue; - } - if (plugin.channels.length === 0) { - nextStartupChannelIds.add(plugin.id); - continue; - } - for (const channelId of plugin.channels) { - nextStartupChannelIds.add(channelId); - } - } + const nextStartupChannelIds = new Set( + nextPluginLookUpTable.manifestRegistry.plugins.flatMap(({ id, channels }) => + nextStartupPluginIds.has(id) ? (channels.length > 0 ? channels : [id]) : [], + ), + ); const channelsToStopBeforeReplace = new Set(); - for (const channelId of beforeChannelIds) { - const targetIds = beforeChannelTargets.get(channelId) ?? new Set([channelId]); + for (const [channelId, targetIds] of beforeChannelTargets) { if ( !nextStartupChannelIds.has(channelId) || pluginConfigTargetsChanged(targetIds, params.changedPaths) @@ -583,80 +580,117 @@ export async function startGatewayCoreRuntime(input: { channelsToStopBeforeReplace.add(channelId); } } - await params.beforeReplace( - channelsToStopBeforeReplace, - channelManager.getPluginCommandCatalogAccounts(), - ); - // If an in-process restart signalled abort during beforeReplace, - // stop before any plugin metadata/runtime side effects continue. - if (params.isAborted?.()) { - return { - restartChannels: new Set(), - activeChannels: new Set(beforeChannelIds), - cancelled: true, - }; - } - const previousPluginServices = runtimeState.pluginServices; - await params.commitRuntime(); - channelManager.setAmbientAutostartSuppressedChannelIds( - nextAmbientAutostartSuppressedChannelIds, - ); - const loaded = prepareGatewayPluginLoad({ - cfg: params.nextConfig, - workspaceDir: pluginWorkspaceDir, - log, - coreGatewayMethodNames, - hostServices: pluginHostServices, - baseMethods, - pluginLookUpTable: nextPluginLookUpTable, - ambientEnvTriggers, - resolveGatewayContext: resolvePluginGatewayContext, - }); - const nextPluginMetadataSnapshot = completePluginMetadataSnapshot({ - snapshot: nextPluginLookUpTable, - config: params.nextConfig, - env: params.env, - workspaceDir: pluginWorkspaceDir, - }); - setCurrentPluginMetadataSnapshot(nextPluginMetadataSnapshot, { - config: params.nextConfig, - env: params.env, - workspaceDir: pluginWorkspaceDir, - }); - currentPluginMetadataSnapshot = nextPluginMetadataSnapshot; - replaceAttachedPluginRuntime(loaded); - kernel.setPluginServices(null); - if (previousPluginServices) { - await previousPluginServices.stop(); - } - await refreshAttachedGatewayDiscovery(loaded.pluginRegistry); - kernel.setPluginServices( - await startPluginServices({ + const pluginRuntimeGeneration = kernel.pluginRuntimeGeneration; + const replacement = pluginRuntimeGeneration.reserve(); + let recoverFromReplacementTeardown: ((error: unknown) => void) | undefined; + try { + await params.beforeReplace( + channelsToStopBeforeReplace, + channelManager.getPluginCommandCatalogAccounts(), + ); + // A rejected reservation restores startup authority; a committed replacement never does. + if (params.isAborted?.()) { + replacement.reject(); + return cancelledReload(beforeChannelIds); + } + const previousServices = pluginRuntimeGeneration.currentServices(); + if (previousServices) { + // Service shutdown is irreversible; only the synchronous runtime commit releases recovery. + recoverFromReplacementTeardown = params.onReplacementTeardownFailure; + await previousServices.stop({ + strict: true, + deadlineAtMs: Date.now() + PLUGIN_SERVICE_REPLACEMENT_STOP_TIMEOUT_MS, + }); + if (params.isAborted?.()) { + throw new Error( + "Gateway plugin runtime replacement was superseded after service teardown", + ); + } + } + await params.commitRuntime(() => { + replacement.commit(); + pluginRuntimeGeneration.publishServices(replacement.claim, null); + recoverFromReplacementTeardown = undefined; + }); + if (!(await replacement.claim.waitForUnblocked())) { + return cancelledReload(beforeChannelIds); + } + + let loaded: ReturnType | undefined; + if ( + !replacement.claim.publish(() => { + channelManager.setAmbientAutostartSuppressedChannelIds( + nextAmbientAutostartSuppressedChannelIds, + ); + loaded = prepareGatewayPluginLoad({ + cfg: params.nextConfig, + activationSourceConfig: params.sourceConfig, + workspaceDir: pluginWorkspaceDir, + log, + coreGatewayMethodNames, + hostServices: pluginHostServices, + baseMethods, + pluginLookUpTable: nextPluginLookUpTable, + ambientEnvTriggers, + resolveGatewayContext: resolvePluginGatewayContext, + }); + const nextPluginMetadataSnapshot = completePluginMetadataSnapshot({ + snapshot: nextPluginLookUpTable, + config: params.sourceConfig, + env: params.env, + workspaceDir: pluginWorkspaceDir, + }); + setCurrentPluginMetadataSnapshot(nextPluginMetadataSnapshot, { + config: params.sourceConfig, + compatibleConfigs: [params.nextConfig], + env: params.env, + workspaceDir: pluginWorkspaceDir, + }); + currentPluginMetadataSnapshot = nextPluginMetadataSnapshot; + replaceAttachedPluginRuntime(loaded); + }) || + !loaded + ) { + return cancelledReload(listAttachedChannelConfigTargets().keys()); + } + await refreshAttachedGatewayDiscovery(loaded.pluginRegistry, replacement.claim); + if (!(await replacement.claim.waitForUnblocked())) { + return cancelledReload(listAttachedChannelConfigTargets().keys()); + } + const nextServices = await startPluginServices({ registry: loaded.pluginRegistry, config: params.nextConfig, workspaceDir: pluginWorkspaceDir, broadcastPluginEvent, - }), - ); - const afterChannelTargets = listAttachedChannelConfigTargets(); - const afterChannelIds = new Set(afterChannelTargets.keys()); - const restartChannels = new Set(); - for (const channelId of new Set([...beforeChannelIds, ...afterChannelIds])) { - const targetIds = - afterChannelTargets.get(channelId) ?? - beforeChannelTargets.get(channelId) ?? - new Set([channelId]); + onHandle: (handle) => pluginRuntimeGeneration.publishServices(replacement.claim, handle), + }); if ( - afterChannelIds.has(channelId) && - (beforeChannelIds.has(channelId) !== afterChannelIds.has(channelId) || - pluginConfigTargetsChanged(targetIds, params.changedPaths)) + !(await replacement.claim.waitForUnblocked()) || + !pluginRuntimeGeneration.publishServices(replacement.claim, nextServices) + ) { + await nextServices.stop({ + strict: true, + deadlineAtMs: Date.now() + PLUGIN_SERVICE_REPLACEMENT_STOP_TIMEOUT_MS, + }); + } + } catch (error) { + replacement.reject(); + recoverFromReplacementTeardown?.(error); + throw error; + } + const afterChannelTargets = listAttachedChannelConfigTargets(); + const restartChannels = new Set(); + for (const [channelId, targetIds] of afterChannelTargets) { + if ( + !beforeChannelIds.has(channelId) || + pluginConfigTargetsChanged(targetIds, params.changedPaths) ) { restartChannels.add(channelId); } } return { restartChannels, - activeChannels: afterChannelIds, + activeChannels: new Set(afterChannelTargets.keys()), }; }; diff --git a/src/gateway/server-lifecycle.ts b/src/gateway/server-lifecycle.ts index 1efe0aeb6b27..78e147b91c7a 100644 --- a/src/gateway/server-lifecycle.ts +++ b/src/gateway/server-lifecycle.ts @@ -24,6 +24,10 @@ import { createGatewayCronReconciliation } from "./server-cron-reconciled.js"; import { applyGatewayLaneConcurrency, resolveGatewayLaneConcurrency } from "./server-lanes.js"; import { createGatewayServerLiveState } from "./server-live-state.js"; import type { GatewayRequestContext } from "./server-methods/types.js"; +import { + createGatewayPluginRuntimeGeneration, + type GatewayPluginRuntimeClaim, +} from "./server-plugin-runtime-generation.js"; import type { GatewayCloseOptions } from "./server-public.js"; import type { prepareGatewayKernelState } from "./server-runtime-state-prepare.js"; import { runGatewayShutdownSteps } from "./server-shutdown.js"; @@ -212,12 +216,19 @@ export async function prepareGatewayLifecycle(params: { gatewayMethods: listActiveGatewayMethods(pluginRuntime.baseGatewayMethods), }); const runtimeState = runtimeStateRef.current; + const pluginRuntimeGeneration = createGatewayPluginRuntimeGeneration({ + getServices: () => runtimeState.pluginServices, + setServices: (services) => { + runtimeState.pluginServices = services; + }, + }); const unavailableGatewayMethods = new Set( minimalTestGateway ? [] : STARTUP_UNAVAILABLE_GATEWAY_METHODS, ); // Kernel methods are the only writers for readiness and advertised-method state. // Residents use this surface so later ownership splits cannot mutate shared state directly. const kernel = { + pluginRuntimeGeneration, setDispatchReady: (ready: boolean) => { startupState.dispatchReady = ready; }, @@ -253,19 +264,19 @@ export async function prepareGatewayLifecycle(params: { runtimeState.heartbeatRunner = handles.heartbeatRunner; runtimeState.stopOutboundDeliveryRecovery = handles.stopOutboundDeliveryRecovery; }, - setPostAttachHandles: (handles: { - stopGatewayUpdateCheck: typeof runtimeState.stopGatewayUpdateCheck; - pluginServices: typeof runtimeState.pluginServices; - }) => { + setPostAttachHandles: ( + handles: { + stopGatewayUpdateCheck: typeof runtimeState.stopGatewayUpdateCheck; + pluginServices: typeof runtimeState.pluginServices; + }, + claim: GatewayPluginRuntimeClaim, + ) => { runtimeState.stopGatewayUpdateCheck = handles.stopGatewayUpdateCheck; - runtimeState.pluginServices = handles.pluginServices; + pluginRuntimeGeneration.publishServices(claim, handles.pluginServices); }, setTailscaleCleanup: (cleanup: typeof runtimeState.tailscaleCleanup) => { runtimeState.tailscaleCleanup = cleanup; }, - setPluginServices: (pluginServices: typeof runtimeState.pluginServices) => { - runtimeState.pluginServices = pluginServices; - }, setConfigReloaderHandle: (configReloader: typeof runtimeState.configReloader) => { runtimeState.configReloader = configReloader; }, diff --git a/src/gateway/server-plugin-runtime-generation.test.ts b/src/gateway/server-plugin-runtime-generation.test.ts new file mode 100644 index 000000000000..d41337addbe8 --- /dev/null +++ b/src/gateway/server-plugin-runtime-generation.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, it, vi } from "vitest"; +import type { PluginServicesHandle } from "../plugins/services.js"; +import { createGatewayPluginRuntimeGeneration } from "./server-plugin-runtime-generation.js"; + +describe("Gateway plugin runtime generation", () => { + it("blocks stale publication during reservation, restores rejected claims, and commits winners", async () => { + let currentServices: PluginServicesHandle | null = null; + const owner = createGatewayPluginRuntimeGeneration({ + getServices: () => currentServices, + setServices: (services) => { + currentServices = services; + }, + }); + const startupClaim = owner.currentClaim(); + const published = vi.fn(); + + expect(startupClaim.publish(published)).toBe(true); + + const rejectedReplacement = owner.reserve(); + expect(startupClaim.isCurrent()).toBe(false); + expect(startupClaim.publish(published)).toBe(false); + let startupUnblocked = false; + const startupCanContinue = startupClaim.waitForUnblocked().then(() => { + startupUnblocked = true; + }); + await Promise.resolve(); + expect(startupUnblocked).toBe(false); + rejectedReplacement.reject(); + await startupCanContinue; + expect(startupClaim.isCurrent()).toBe(true); + + const acceptedReplacement = owner.reserve(); + expect(acceptedReplacement.claim.publish(published)).toBe(false); + acceptedReplacement.commit(); + expect(owner.currentClaim()).toBe(acceptedReplacement.claim); + expect(startupClaim.publish(published)).toBe(false); + expect(acceptedReplacement.claim.publish(published)).toBe(true); + expect(published).toHaveBeenCalledTimes(2); + + const winningServices: PluginServicesHandle = { stop: vi.fn(async () => {}) }; + expect(owner.publishServices(startupClaim, winningServices)).toBe(false); + expect(owner.publishServices(acceptedReplacement.claim, winningServices)).toBe(true); + expect(owner.currentServices()).toBe(winningServices); + }); + + it.each([ + { successor: "rejects", survives: true }, + { successor: "commits", survives: false }, + ])( + "settles a pending successor that $successor before deciding discovery and service ownership", + async ({ survives }) => { + let currentServices: PluginServicesHandle | null = null; + const owner = createGatewayPluginRuntimeGeneration({ + getServices: () => currentServices, + setServices: (services) => { + currentServices = services; + }, + }); + const committed = owner.reserve(); + committed.commit(); + const pendingSuccessor = owner.reserve(); + const discoveryStop = vi.fn(); + const services: PluginServicesHandle = { stop: vi.fn(async () => {}) }; + let settled = false; + const publication = committed.claim.waitForUnblocked().then((isCurrent) => { + settled = true; + if (isCurrent) { + owner.publishServices(committed.claim, services); + } else { + discoveryStop(); + } + return isCurrent; + }); + + await Promise.resolve(); + expect(settled).toBe(false); + if (survives) { + pendingSuccessor.reject(); + } else { + pendingSuccessor.commit(); + } + + await expect(publication).resolves.toBe(survives); + expect(owner.currentServices()).toBe(survives ? services : null); + expect(discoveryStop).toHaveBeenCalledTimes(survives ? 0 : 1); + }, + ); +}); diff --git a/src/gateway/server-plugin-runtime-generation.ts b/src/gateway/server-plugin-runtime-generation.ts new file mode 100644 index 000000000000..3aeb88fe712b --- /dev/null +++ b/src/gateway/server-plugin-runtime-generation.ts @@ -0,0 +1,81 @@ +import type { PluginServicesHandle } from "../plugins/services.js"; +import { createDeferredCore } from "../shared/deferred.js"; + +export type GatewayPluginRuntimeClaim = Readonly<{ + isCurrent: () => boolean; + waitForUnblocked: () => Promise; + publish: (publication: () => void) => boolean; +}>; + +type GatewayPluginRuntimeReservation = Readonly<{ + claim: GatewayPluginRuntimeClaim; + commit: () => void; + reject: () => void; +}>; + +/** One Gateway owner fences every plugin publication across startup and hot replacement. */ +export function createGatewayPluginRuntimeGeneration(params: { + getServices: () => PluginServicesHandle | null; + setServices: (services: PluginServicesHandle | null) => void; +}) { + let current: GatewayPluginRuntimeClaim; + let pending: + | { + claim: GatewayPluginRuntimeClaim; + settled: ReturnType>; + } + | undefined; + + const createClaim = (): GatewayPluginRuntimeClaim => { + const claim: GatewayPluginRuntimeClaim = Object.freeze({ + isCurrent: () => current === claim && pending === undefined, + waitForUnblocked: async () => { + for (;;) { + const reservation = pending; + if (current !== claim || !reservation) { + return claim.isCurrent(); + } + await reservation.settled.promise; + } + }, + publish: (publication: () => void) => { + if (!claim.isCurrent()) { + return false; + } + publication(); + return true; + }, + }); + return claim; + }; + current = createClaim(); + + return { + currentClaim: () => current, + currentServices: () => params.getServices(), + publishServices: (claim: GatewayPluginRuntimeClaim, services: PluginServicesHandle | null) => + claim.publish(() => params.setServices(services)), + reserve: (): GatewayPluginRuntimeReservation => { + if (pending) { + throw new Error("a Gateway plugin runtime replacement is already pending"); + } + const reservation = { claim: createClaim(), settled: createDeferredCore() }; + pending = reservation; + const settle = (accepted: boolean) => { + if (pending !== reservation) { + return; + } + if (accepted) { + current = reservation.claim; + } + pending = undefined; + reservation.settled.resolve(); + }; + return Object.freeze({ + claim: reservation.claim, + commit: () => settle(true), + reject: () => settle(false), + }); + }, + }; +} diff --git a/src/gateway/server-plugins.lifecycle.test.ts b/src/gateway/server-plugins.lifecycle.test.ts index e0f4eee22486..8b5a02355ec2 100644 --- a/src/gateway/server-plugins.lifecycle.test.ts +++ b/src/gateway/server-plugins.lifecycle.test.ts @@ -5,7 +5,9 @@ import fs from "node:fs/promises"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { getActivePluginRegistry } from "../plugins/runtime.js"; import type { PluginRuntime } from "../plugins/runtime/types.js"; +import { getActiveSecretsRuntimeConfigSnapshot } from "../secrets/runtime-state.js"; import { getFreePort } from "../test-utils/ports.js"; import { connectWebchatClient, @@ -30,9 +32,14 @@ type InstanceBindingProbeCoordinator = { identify: (value: object) => number; nextRegistryId: number; runtimes: PluginRuntime[]; + serviceStarts: number; + serviceStops: number; + serviceStopFailure?: "rejection" | "timeout"; }; -function installInstanceBindingProbeCoordinator(): InstanceBindingProbeCoordinator { +function installInstanceBindingProbeCoordinator(options?: { + serviceStopFailure?: InstanceBindingProbeCoordinator["serviceStopFailure"]; +}): InstanceBindingProbeCoordinator { const ids = new WeakMap(); let nextId = 1; const coordinator: InstanceBindingProbeCoordinator = { @@ -47,6 +54,9 @@ function installInstanceBindingProbeCoordinator(): InstanceBindingProbeCoordinat }, nextRegistryId: 1, runtimes: [], + serviceStarts: 0, + serviceStops: 0, + ...(options?.serviceStopFailure ? { serviceStopFailure: options.serviceStopFailure } : {}), }; (globalThis as Record)[INSTANCE_BINDING_PROBE_KEY] = coordinator; return coordinator; @@ -104,6 +114,23 @@ async function writeInstanceBindingProbePlugin(): Promise<{ bundledRoot: string const coordinator = globalThis[Symbol.for("openclaw.test.gatewayInstanceBindingProbe")]; const registryId = coordinator.nextRegistryId++; coordinator.runtimes.push(api.runtime); + if (coordinator.serviceStopFailure) { + api.registerService({ + id: "instance-binding-service", + start() { + coordinator.serviceStarts += 1; + }, + stop() { + coordinator.serviceStops += 1; + if (coordinator.serviceStopFailure === "rejection") { + return Promise.reject(new Error("instance-binding service cleanup rejected")); + } + if (coordinator.serviceStopFailure === "timeout") { + return new Promise(() => {}); + } + }, + }); + } api.registerGatewayMethod("${INSTANCE_BINDING_PROBE_METHOD}", ({ context, respond }) => { respond(true, { registryId, @@ -118,8 +145,10 @@ async function writeInstanceBindingProbePlugin(): Promise<{ bundledRoot: string return { bundledRoot }; } -async function prepareInstanceBindingTest() { - const coordinator = installInstanceBindingProbeCoordinator(); +async function prepareInstanceBindingTest(options?: { + serviceStopFailure?: InstanceBindingProbeCoordinator["serviceStopFailure"]; +}) { + const coordinator = installInstanceBindingProbeCoordinator(options); const plugin = await writeInstanceBindingProbePlugin(); process.env.OPENCLAW_TEST_MINIMAL_GATEWAY = "0"; delete process.env.OPENCLAW_DISABLE_BUNDLED_PLUGINS; @@ -284,4 +313,59 @@ describe("gateway plugin instance bindings", () => { ).resolves.toEqual({ messages: [] }); }, ); + + it.each(["rejection", "timeout"] as const)( + "keeps the active Gateway runtime when real plugin replacement cleanup fails by %s", + { timeout: 600_000 }, + async (serviceStopFailure) => { + const { coordinator } = await prepareInstanceBindingTest({ serviceStopFailure }); + const hotReloadRecovery = vi.fn(() => ({ status: "emitted" as const })); + const port = await getFreePort(); + const server = await startTestGatewayServer(port, { + auth: { mode: "none" }, + controlUiEnabled: false, + hotReloadRecovery, + sidecarStartup: "start", + }); + started.push(server); + await server.startupSettled; + + const initialRegistry = getActivePluginRegistry(); + const initialRuntimeConfig = getActiveSecretsRuntimeConfigSnapshot()?.config; + const initialRegistrationCount = coordinator.runtimes.length; + const initialHandler = initialRegistry?.gatewayHandlers[INSTANCE_BINDING_PROBE_METHOD]; + expect(initialRegistry).toBeDefined(); + expect(initialRuntimeConfig).toBeDefined(); + expect(initialHandler).toBeTypeOf("function"); + expect(coordinator.serviceStarts).toBe(1); + + const socket = await connectWebchatClient({ port, scopes: ["operator.admin"] }); + sockets.push(socket); + const currentConfig = await rpcReq<{ hash?: string }>(socket, "config.get", {}); + expect(currentConfig.ok).toBe(true); + const reload = await rpcReq(socket, "config.patch", { + raw: JSON.stringify({ + plugins: { + entries: { + "instance-binding-probe": { + subagent: { allowModelOverride: true }, + }, + }, + }, + }), + baseHash: currentConfig.payload?.hash, + }); + expect(reload.ok, reload.error?.message).toBe(true); + + await expect.poll(() => hotReloadRecovery.mock.calls.length, { timeout: 30_000 }).toBe(1); + expect(coordinator.serviceStops).toBe(1); + expect(coordinator.serviceStarts).toBe(1); + expect(coordinator.runtimes).toHaveLength(initialRegistrationCount); + expect(getActiveSecretsRuntimeConfigSnapshot()?.config).toBe(initialRuntimeConfig); + expect(getActivePluginRegistry()).toBe(initialRegistry); + expect(getActivePluginRegistry()?.gatewayHandlers[INSTANCE_BINDING_PROBE_METHOD]).toBe( + initialHandler, + ); + }, + ); }); diff --git a/src/gateway/server-reload-contracts.ts b/src/gateway/server-reload-contracts.ts index d5d3051478bc..bceb7d0d5b22 100644 --- a/src/gateway/server-reload-contracts.ts +++ b/src/gateway/server-reload-contracts.ts @@ -69,9 +69,9 @@ export type GatewayGmailRestartAbortController = { export type GatewayHotReloadPublication = { publish: (commit: () => Promise, isCommitted: () => boolean) => Promise; isCurrent: () => boolean; + sourceConfig: OpenClawConfig; prepareRestartRuntimeConfig?: () => Promise; runtimeEnv?: NodeJS.ProcessEnv; - sourceConfig?: OpenClawConfig; }; export type GatewayRestartTransactionState = "pending" | "committed" | "rejected"; @@ -151,12 +151,14 @@ export type GatewayReloadHandlerParams = { stopPostReadySidecars?: () => Promise | void; reloadPlugins: (params: { nextConfig: OpenClawConfig; + sourceConfig: OpenClawConfig; changedPaths: readonly string[]; beforeReplace: ( channels: ReadonlySet, accounts?: ReadonlyMap>, ) => Promise; - commitRuntime: () => Promise; + commitRuntime: (onCommit?: () => void) => Promise; + onReplacementTeardownFailure: (error: unknown) => void; env: NodeJS.ProcessEnv; isAborted?: () => boolean; }) => Promise; diff --git a/src/gateway/server-reload-handlers.test.ts b/src/gateway/server-reload-handlers.test.ts index e946cbdbfcf4..dd075ecdd335 100644 --- a/src/gateway/server-reload-handlers.test.ts +++ b/src/gateway/server-reload-handlers.test.ts @@ -1369,7 +1369,11 @@ describe("gateway hot reload model state", () => { applyHotReload( createHotTailPlan({ restartHeartbeat: true }), { agents: { defaults: { maxConcurrent: 1 } } } as OpenClawConfig, - { publish, isCurrent: () => true }, + { + sourceConfig: { agents: { defaults: { maxConcurrent: 1 } } }, + publish, + isCurrent: () => true, + }, ), ).rejects.toThrow("heartbeat update failed"); @@ -1480,6 +1484,7 @@ describe("gateway hot reload model state", () => { createCronRestartPlan(), { cron: { enabled: true } }, { + sourceConfig: { cron: { enabled: true } }, publish, isCurrent: () => true, }, @@ -1796,6 +1801,7 @@ describe("gateway hot reload superseded tail recovery", () => { plan, { agents: { defaults: { workspace: "/tmp/a" } } }, { + sourceConfig: { agents: { defaults: { workspace: "/tmp/a" } } }, isCurrent: () => false, publish: async (commit) => await commit(), }, @@ -1845,6 +1851,7 @@ describe("gateway hot reload superseded tail recovery", () => { try { const staleTail = handlers.applyHotReload(plan, configA, { + sourceConfig: configA, isCurrent: () => false, publish: async (commit) => await commit(), }); @@ -1962,6 +1969,7 @@ describe("gateway hot reload superseded tail recovery", () => { agents: { defaults: { workspace: "/tmp/a" } }, } as OpenClawConfig; const reloadA = handlers.applyHotReload(plan, configA, { + sourceConfig: configA, isCurrent, publish: async (commit) => await commit(), }); @@ -1980,6 +1988,7 @@ describe("gateway hot reload superseded tail recovery", () => { const configC = { logging: { level: "debug" as const } } satisfies OpenClawConfig; pendingConfig = configC; await handlers.applyHotReload(createHotTailPlan(), configC, { + sourceConfig: configC, isCurrent: () => pendingConfig === configC, publish: async (commit) => await commit(), }); @@ -2011,6 +2020,7 @@ describe("gateway hot reload superseded tail recovery", () => { createHotTailPlan({ restartChannels: new Set(["discord"]) }), {}, { + sourceConfig: {}, isCurrent: () => current, publish: async (commit) => await commit(), }, @@ -2807,6 +2817,10 @@ describe("gateway restart deferral preflight", () => { channels: { discord: { token: "token" } }, }, { + sourceConfig: { + gateway: { reload: {} }, + channels: { discord: { token: "token" } }, + }, isCurrent: () => true, publish: async (commit) => { runtimePublished = true; @@ -4852,6 +4866,7 @@ describe("gateway plugin hot reload handlers", () => { }), {}, { + sourceConfig: {}, runtimeEnv: runtimeEnv.env, isCurrent: () => true, publish: async (commit) => { @@ -4909,6 +4924,7 @@ describe("gateway plugin hot reload handlers", () => { }), nextConfig, { + sourceConfig: nextConfig, runtimeEnv: runtimeEnv.env, isCurrent: () => true, publish: async (commit) => { @@ -5069,6 +5085,7 @@ describe("gateway plugin hot reload handlers", () => { }), { hooks: { enabled: true, token: "token", path: "/next" } }, { + sourceConfig: { hooks: { enabled: true, token: "token", path: "/next" } }, isCurrent: () => true, publish: async (commit) => { events.push("runtime:publish"); @@ -5089,6 +5106,202 @@ describe("gateway plugin hot reload handlers", () => { expect(handlers.setState).toHaveBeenCalledTimes(1); }); + it("passes authored plugin config separately from synthesized runtime trust to replacement planning", async () => { + const sourceConfig = { + plugins: { enabled: true }, + } satisfies OpenClawConfig; + const runtimeConfig = { + plugins: { + enabled: true, + allow: ["external-plugin"], + entries: { "external-plugin": { enabled: true } }, + }, + } satisfies OpenClawConfig; + const reloadPlugins = vi.fn( + async (params: { + commitRuntime: () => Promise; + }): Promise => { + await params.commitRuntime(); + return makePluginReloadResult(); + }, + ); + const handlers = createReloadHandlersForTest(undefined, undefined, reloadPlugins); + + await handlers.applyHotReload(createPluginReloadPlan(), runtimeConfig, { + sourceConfig, + isCurrent: () => true, + publish: async (commit) => await commit(), + }); + + const reloadParams = reloadPlugins.mock.calls[0]?.[0] as + | { nextConfig: OpenClawConfig; sourceConfig?: OpenClawConfig } + | undefined; + expect(reloadParams?.nextConfig).toBe(runtimeConfig); + expect(reloadParams?.sourceConfig).toBe(sourceConfig); + }); + + it("requests recovery when runtime publication rejects after successful service teardown", async () => { + const events: string[] = []; + const publicationFailure = new Error("runtime publication rejected"); + const requestRecoveryRestart = vi.fn(() => ({ status: "emitted" as const })); + const activateCandidate = vi.fn(); + const startCandidateServices = vi.fn(); + const publish = vi.fn(async () => { + events.push("runtime:publish"); + throw publicationFailure; + }); + const reloadPlugins = vi.fn( + async (params: { + beforeReplace: (channels: ReadonlySet) => Promise; + commitRuntime: () => Promise; + onReplacementTeardownFailure: (error: unknown) => void; + }): Promise => { + await params.beforeReplace(new Set(["discord"])); + events.push("services:strict-stop"); + try { + await params.commitRuntime(); + } catch (error) { + params.onReplacementTeardownFailure(error); + throw error; + } + activateCandidate(); + startCandidateServices(); + return makePluginReloadResult(); + }, + ); + const handlers = createReloadHandlersForTest( + undefined, + { + stop: vi.fn(async (channel) => { + events.push(`channel:stop:${channel}`); + }), + start: vi.fn(async (channel) => { + events.push(`channel:start:${channel}`); + }), + }, + reloadPlugins, + undefined, + requestRecoveryRestart, + ); + + await expect( + handlers.applyHotReload( + createPluginReloadPlan(), + { plugins: { enabled: true } }, + { + sourceConfig: { plugins: { enabled: true } }, + publish, + isCurrent: () => true, + }, + ), + ).rejects.toBe(publicationFailure); + + expect(events).toEqual(["channel:stop:discord", "services:strict-stop", "runtime:publish"]); + expect(requestRecoveryRestart).toHaveBeenCalledOnce(); + expect(activateCandidate).not.toHaveBeenCalled(); + expect(startCandidateServices).not.toHaveBeenCalled(); + expect(handlers.setState).not.toHaveBeenCalled(); + }); + + it.each([ + { label: "service rejection", failure: "service rejected cleanup" }, + { label: "service timeout", failure: "service cleanup timed out" }, + ])( + "requests recovery without committing replacement after strict $label", + async ({ failure }) => { + await withGatewayRestartSignal(async (signalSpy) => { + const events: string[] = []; + const cleanupFailure = new AggregateError( + [new Error(failure)], + "plugin service stop failed", + ); + const publish = vi.fn(async (commit: () => Promise) => await commit()); + const reloadPlugins = vi.fn( + async (params: { + beforeReplace: (channels: ReadonlySet) => Promise; + commitRuntime: () => Promise; + onReplacementTeardownFailure: (error: unknown) => void; + }): Promise => { + await params.beforeReplace(new Set(["discord"])); + events.push("services:strict-stop"); + params.onReplacementTeardownFailure(cleanupFailure); + throw cleanupFailure; + }, + ); + const handlers = createReloadHandlersForTest( + undefined, + { + stop: vi.fn(async (channel) => { + events.push(`channel:stop:${channel}`); + }), + start: vi.fn(async (channel) => { + events.push(`channel:start:${channel}`); + }), + }, + reloadPlugins, + ); + + await expect( + handlers.applyHotReload( + createPluginReloadPlan(), + { plugins: { enabled: true } }, + { + sourceConfig: { plugins: { enabled: true } }, + publish, + isCurrent: () => true, + }, + ), + ).rejects.toBe(cleanupFailure); + + expect(events).toEqual(["channel:stop:discord", "services:strict-stop"]); + expect(publish).not.toHaveBeenCalled(); + expect(handlers.setState).not.toHaveBeenCalled(); + expect(signalSpy).toHaveBeenCalledOnce(); + expect(isGatewayWorkAdmissionClosed()).toBe(true); + markGatewaySigusr1RestartHandled(); + }); + }, + ); + + it("rolls back unrelated aggregate plugin failures without requesting cleanup recovery", async () => { + const events: string[] = []; + const planningFailure = new AggregateError( + [new Error("candidate rejected")], + "plugin planning failed", + ); + const requestRecoveryRestart = vi.fn(() => ({ status: "emitted" as const })); + const reloadPlugins = vi.fn( + async (params: { + beforeReplace: (channels: ReadonlySet) => Promise; + }): Promise => { + await params.beforeReplace(new Set(["discord"])); + throw planningFailure; + }, + ); + const handlers = createReloadHandlersForTest( + undefined, + { + stop: vi.fn(async (channel) => { + events.push(`stop:${channel}`); + }), + start: vi.fn(async (channel) => { + events.push(`start:${channel}`); + }), + }, + reloadPlugins, + undefined, + requestRecoveryRestart, + ); + + await expect( + handlers.applyHotReload(createPluginReloadPlan(), { plugins: { enabled: true } }), + ).rejects.toBe(planningFailure); + + expect(events).toEqual(["stop:discord", "start:discord"]); + expect(requestRecoveryRestart).not.toHaveBeenCalled(); + expect(handlers.setState).not.toHaveBeenCalled(); + }); + it("restarts only the account retaining a command catalog on plugin replacement", async () => { const discordPlugin = createChannelTestPluginBase({ id: "discord", @@ -5159,7 +5372,11 @@ describe("gateway plugin hot reload handlers", () => { handlers.applyHotReload( createPluginReloadPlan(), { plugins: { enabled: true } }, - { publish: async (commit) => await commit(), isCurrent: () => true }, + { + sourceConfig: { plugins: { enabled: true } }, + publish: async (commit) => await commit(), + isCurrent: () => true, + }, ), ); @@ -5222,7 +5439,11 @@ describe("gateway plugin hot reload handlers", () => { handlers.applyHotReload( createPluginReloadPlan(), { plugins: { enabled: true } }, - { publish: async (commit) => await commit(), isCurrent: () => true }, + { + sourceConfig: { plugins: { enabled: true } }, + publish: async (commit) => await commit(), + isCurrent: () => true, + }, ), ).resolves.toBeUndefined(); @@ -5253,7 +5474,7 @@ describe("gateway plugin hot reload handlers", () => { handlers.applyHotReload( createPluginReloadPlan(), { plugins: { enabled: true } }, - { publish, isCurrent: () => true }, + { sourceConfig: { plugins: { enabled: true } }, publish, isCurrent: () => true }, ), ).resolves.toBeUndefined(); @@ -5330,7 +5551,7 @@ describe("gateway plugin hot reload handlers", () => { handlers.applyHotReload( plan, { plugins: { enabled: true } }, - { publish, isCurrent: () => true }, + { sourceConfig: { plugins: { enabled: true } }, publish, isCurrent: () => true }, ), ).rejects.toThrow( "config reload requires a managed gateway restart owner for irreversible hot reload", @@ -5385,7 +5606,7 @@ describe("gateway plugin hot reload handlers", () => { handlers.applyHotReload( createPluginReloadPlan(), { plugins: { enabled: true } }, - { publish, isCurrent: () => true }, + { sourceConfig: { plugins: { enabled: true } }, publish, isCurrent: () => true }, ), ).rejects.toThrow("publication failed"); @@ -5775,6 +5996,7 @@ describe("deferred channel reload abort generation", () => { abortChannelReloadPlan, {}, { + sourceConfig: {}, isCurrent: () => transactionCurrent, publish: async (commit) => await commit(), }, diff --git a/src/gateway/server-reload-hot.ts b/src/gateway/server-reload-hot.ts index cbfcf08b74c8..c17d36a1631e 100644 --- a/src/gateway/server-reload-hot.ts +++ b/src/gateway/server-reload-hot.ts @@ -164,7 +164,7 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) `${action} suppressed by crash-loop breaker for channels: ${[...channels].join(", ")}`, ); }; - const commitRuntime = async () => { + const commitRuntime = async (onCommit?: () => void) => { if (runtimeCommitted) { return; } @@ -191,6 +191,7 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) } applyGatewayLaneConcurrency(laneConcurrency); runtimeCommitted = true; + onCommit?.(); setGatewaySigusr1RestartPolicy({ allowExternal: isRestartEnabled(nextConfig) }); if (plan.restartCron) { params.cronReconciliation.invalidate(); @@ -301,7 +302,8 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) }); return; } - params.logReload.warn(`${surface} failed after config commit${detail}; restarting gateway`); + const commitState = runtimeCommitted ? "after config commit" : "before config commit"; + params.logReload.warn(`${surface} failed ${commitState}${detail}; restarting gateway`); if (recoveryRestartScheduled) { return; } @@ -503,14 +505,22 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) try { pluginReloadResult = await params.reloadPlugins({ nextConfig, + // Without a managed publication, the direct caller's input is itself authored. + sourceConfig: publication ? publication.sourceConfig : nextConfig, changedPaths: plan.changedPaths, beforeReplace: stopChannelsBeforePluginReplace, commitRuntime, + onReplacementTeardownFailure: (error) => + scheduleRecoveryRestart("plugin service replacement teardown", error), env: publication?.runtimeEnv ?? process.env, isAborted: isPluginReloadAborted, }); } catch (err) { if (!runtimeCommitted) { + // Once replacement teardown begins, old services cannot safely be rolled back. + if (recoveryRestartScheduled) { + throw err; + } const rollbackFailures = await rollbackStoppedPluginTargets( "failed plugin runtime publication", ); diff --git a/src/gateway/server-startup-finish.ts b/src/gateway/server-startup-finish.ts index 006314df77d6..44b9dfd1a34e 100644 --- a/src/gateway/server-startup-finish.ts +++ b/src/gateway/server-startup-finish.ts @@ -130,6 +130,7 @@ export async function finishGatewayStartup(params: { residentRegistry, getPluginMetadataSnapshot, } = runtime; + const startupPluginRuntimeClaim = kernel.pluginRuntimeGeneration.currentClaim(); const unregisterGatewayLifetimeSidecar = (sidecar: GatewayPostReadySidecarHandle) => { kernel.setGatewayLifetimeSidecars( runtimeState.gatewayLifetimeSidecars.filter((registered) => registered !== sidecar), @@ -260,6 +261,9 @@ export async function finishGatewayStartup(params: { activationSourceConfig: startupActivationSourceConfig, pluginManifestRecords, ...(pluginMetadataSnapshot ? { pluginMetadataSnapshot } : {}), + pluginRuntimeClaim: startupPluginRuntimeClaim, + getCurrentPluginRegistry: () => pluginRuntime.registry, + getCurrentPluginMetadataSnapshot: getPluginMetadataSnapshot, ambientEnvTriggers, pluginRegistry: pluginRuntime.registry, defaultWorkspaceDir, @@ -286,15 +290,20 @@ export async function finishGatewayStartup(params: { startupTrace, ambientEnvTriggers, resolveGatewayContext: resolvePluginGatewayContext, + pluginRuntimeClaim: startupPluginRuntimeClaim, + getCurrentPluginRegistry: () => pluginRuntime.registry, }); }, onStartupPluginsLoading: () => { startupState.pendingReason = "startup-sidecars"; }, onStartupPluginsLoaded: async (loaded) => { - replaceAttachedPluginRuntime(loaded); + if (!startupPluginRuntimeClaim.publish(() => replaceAttachedPluginRuntime(loaded))) { + loaded.retireGatewayRuntimeBindings?.(); + return; + } startupState.pendingReason = "startup-sidecars"; - await refreshAttachedGatewayDiscovery(loaded.pluginRegistry); + await refreshAttachedGatewayDiscovery(loaded.pluginRegistry, startupPluginRuntimeClaim); }, getCronService: () => runtimeState?.cronState.cron as PluginHookGatewayCronService | undefined, @@ -302,7 +311,7 @@ export async function finishGatewayStartup(params: { releaseStartupAccountStarts(); }, onPluginServices: (pluginServices) => { - kernel.setPluginServices(pluginServices); + kernel.pluginRuntimeGeneration.publishServices(startupPluginRuntimeClaim, pluginServices); }, onPostReadySidecars: registerPostReadySidecars, onGatewayLifetimeSidecars: registerGatewayLifetimeSidecars, @@ -342,7 +351,7 @@ export async function finishGatewayStartup(params: { }), ), ); - kernel.setPostAttachHandles(postAttachHandles); + kernel.setPostAttachHandles(postAttachHandles, startupPluginRuntimeClaim); startupTrace.detail("memory.ready", collectGatewayProcessMemoryUsageMb()); startupTrace.mark("ready"); if (sidecarStartup === "defer") { diff --git a/src/gateway/server-startup-plugins.ts b/src/gateway/server-startup-plugins.ts index e18f25b34920..4ddc87e884b4 100644 --- a/src/gateway/server-startup-plugins.ts +++ b/src/gateway/server-startup-plugins.ts @@ -17,6 +17,7 @@ import { getActivePluginRegistry, setActivePluginRegistry } from "../plugins/run import { resolveGatewayStartupPluginActivationConfig } from "./plugin-activation-runtime-config.js"; import { listGatewayMethods } from "./server-methods-list.js"; import type { GatewayContextResolver } from "./server-methods/types.js"; +import type { GatewayPluginRuntimeClaim } from "./server-plugin-runtime-generation.js"; type GatewayPluginBootstrapLog = { info: (message: string) => void; @@ -227,10 +228,23 @@ export async function loadGatewayStartupPluginRuntime(params: { startupTrace?: GatewayStartupTrace; ambientEnvTriggers?: AmbientEnvTriggerPolicy; resolveGatewayContext?: GatewayContextResolver; + pluginRuntimeClaim?: GatewayPluginRuntimeClaim; + getCurrentPluginRegistry?: () => PluginRegistry; }) { // Keep server-plugin-bootstrap behind one lazy boundary; startup config tests can exercise // planning without importing plugin package runtimes. const { loadGatewayStartupPlugins } = await import("./server-plugin-bootstrap.js"); + await params.pluginRuntimeClaim?.waitForUnblocked(); + if (params.pluginRuntimeClaim && !params.pluginRuntimeClaim.isCurrent()) { + const currentPluginRegistry = params.getCurrentPluginRegistry?.(); + if (!currentPluginRegistry) { + throw new Error("superseded Gateway startup cannot resolve the current plugin runtime"); + } + return { + pluginRegistry: currentPluginRegistry, + gatewayMethods: params.baseMethods, + }; + } const loaded = loadGatewayStartupPlugins({ cfg: params.cfg, activationSourceConfig: params.activationSourceConfig, diff --git a/src/gateway/server-startup-post-attach.test.ts b/src/gateway/server-startup-post-attach.test.ts index 732ef50188e5..1346e2452163 100644 --- a/src/gateway/server-startup-post-attach.test.ts +++ b/src/gateway/server-startup-post-attach.test.ts @@ -10,7 +10,10 @@ import type { PluginHookGatewayContext, PluginHookGatewayStartEvent, } from "../plugins/hook-types.js"; +import { registerPluginHttpRoute } from "../plugins/http-registry.js"; +import { createEmptyPluginRegistry } from "../plugins/registry-empty.js"; import type { PluginServicesHandle } from "../plugins/services.js"; +import type { OpenClawPluginServiceContext } from "../plugins/types.js"; import { getActiveGatewayRootWorkCount, resetGatewayWorkAdmission, @@ -1442,6 +1445,72 @@ describe("startGatewayPostAttachRuntime", () => { expect(events).toEqual(["startup-loaded-start", "startup-loaded-end", "sidecars"]); }); + it("adopts a winning plugin generation without publishing stale deferred startup state", async () => { + const startupRegistry = { + plugins: [{ id: "startup", status: "loaded" }], + typedHooks: [], + } as never; + const winningRegistry = { + plugins: [{ id: "replacement", status: "loaded" }], + typedHooks: [], + } as never; + let startupClaimCurrent = true; + let releasePluginLoad: (() => void) | undefined; + const pluginLoadReady = new Promise((resolve) => { + releasePluginLoad = resolve; + }); + const pluginRuntimeClaim = { + isCurrent: () => startupClaimCurrent, + waitForUnblocked: async () => true, + publish: (publish: () => void) => { + if (!startupClaimCurrent) { + return false; + } + publish(); + return true; + }, + }; + const onStartupPluginsLoaded = vi.fn(); + const onPluginServices = vi.fn(); + const onSidecarsReady = vi.fn(); + const unlockStartupMethods = vi.fn(); + const startGatewaySidecarsCandidate = vi.fn( + async (params: Parameters[0]) => { + expect(params.pluginRegistry).toBe(winningRegistry); + expect(params.shouldStartPluginServices?.()).toBe(false); + return { pluginServices: null, postReadySidecars: [] }; + }, + ); + const loadStartupPlugins = vi.fn(async () => { + await pluginLoadReady; + return { pluginRegistry: startupRegistry, gatewayMethods: ["startup.method"] }; + }); + + const runtime = await startGatewayPostAttachRuntime( + createPostAttachParams({ + sidecarStartup: "defer", + loadStartupPlugins, + onStartupPluginsLoaded, + onPluginServices, + onSidecarsReady, + unlockStartupMethods, + pluginRuntimeClaim, + getCurrentPluginRegistry: () => winningRegistry, + }), + createPostAttachRuntimeDeps({ startGatewaySidecars: startGatewaySidecarsCandidate }), + ); + await waitForGatewayTestState(() => expect(loadStartupPlugins).toHaveBeenCalledOnce()); + startupClaimCurrent = false; + releasePluginLoad?.(); + await expect(runtime.startupSettled).resolves.toBeUndefined(); + + expect(onStartupPluginsLoaded).not.toHaveBeenCalled(); + expect(startGatewaySidecarsCandidate).toHaveBeenCalledOnce(); + expect(onPluginServices).not.toHaveBeenCalled(); + expect(unlockStartupMethods).toHaveBeenCalledOnce(); + expect(onSidecarsReady).toHaveBeenCalledOnce(); + }); + it("waits for sidecars by default before returning", async () => { let resumeSidecars: (() => void) | undefined; const sidecarsReady = new Promise<{ pluginServices: null; postReadySidecars: [] }>( @@ -2077,6 +2146,126 @@ describe("startGatewayPostAttachRuntime", () => { expect(onPluginServices).toHaveBeenLastCalledWith(null); }); + it("forwards strict replacement cleanup through the deferred plugin service owner", async () => { + const serviceStop = vi.fn(async () => {}); + const startedServices = { stop: serviceStop }; + const publishedOwner: { current: PluginServicesHandle | null } = { current: null }; + hoisted.startPluginServices.mockImplementationOnce(async (params) => { + params.onHandle?.(startedServices); + return startedServices; + }); + + await startGatewaySidecars({ + cfg: { hooks: { internal: { enabled: false } } } as never, + pluginRegistry: createPostAttachParams().pluginRegistry, + defaultWorkspaceDir: "/tmp/openclaw-workspace", + deps: {} as never, + startChannels: vi.fn(async () => {}), + onPluginServices: (handle) => { + publishedOwner.current = handle; + }, + log: { warn: vi.fn() }, + logHooks: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + logChannels: { info: vi.fn(), error: vi.fn() }, + }); + + expect(publishedOwner.current).not.toBeNull(); + const replacement = { strict: true, deadlineAtMs: Date.now() + 5_000 } as const; + await publishedOwner.current?.stop(replacement); + expect(serviceStop).toHaveBeenCalledWith(replacement); + }); + + it("fences late service capabilities when deferred ownership consumes the replacement deadline", async () => { + vi.useFakeTimers(); + const actualServices = + await vi.importActual("../plugins/services.js"); + const registry = createEmptyPluginRegistry(); + const broadcastPluginEvent = vi.fn(); + let context: OpenClawPluginServiceContext | undefined; + let releaseCleanup: (() => void) | undefined; + const cleanupReleased = new Promise((resolve) => { + releaseCleanup = resolve; + }); + registry.services.push({ + pluginId: "deferred-deadline", + source: "test", + origin: "workspace", + service: { + id: "deferred-deadline-service", + start: (serviceContext) => { + context = serviceContext; + registerPluginHttpRoute({ + path: "/deferred-deadline-route", + auth: "plugin", + handler: vi.fn(), + }); + }, + stop: async () => { + await cleanupReleased; + }, + }, + }); + hoisted.startPluginServices.mockImplementationOnce(async (params) => { + await new Promise((resolve) => { + setTimeout(resolve, 4_900); + }); + return await actualServices.startPluginServices( + params as Parameters[0], + ); + }); + const publishedOwner: { current: PluginServicesHandle | null } = { current: null }; + let stopping: Promise | undefined; + let sidecars: ReturnType | undefined; + + try { + sidecars = startGatewaySidecars({ + cfg: { hooks: { internal: { enabled: false } } } as never, + pluginRegistry: registry, + defaultWorkspaceDir: "/tmp/openclaw-workspace", + deps: {} as never, + startChannels: vi.fn(async () => {}), + broadcastPluginEvent, + onPluginServices: (handle) => { + publishedOwner.current = handle; + }, + log: { warn: vi.fn() }, + logHooks: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + logChannels: { info: vi.fn(), error: vi.fn() }, + }); + await waitForGatewayTestState(() => { + expect(hoisted.startPluginServices).toHaveBeenCalledOnce(); + }); + if (!publishedOwner.current) { + throw new Error("deferred plugin service owner was not published"); + } + + const deadlineAtMs = Date.now() + 5_000; + let failure: unknown; + stopping = publishedOwner.current + .stop({ strict: true, deadlineAtMs }) + .catch((error: unknown) => { + failure = error; + }); + + await vi.advanceTimersByTimeAsync(4_900); + expect(registry.httpRoutes).toHaveLength(1); + expect(failure).toBeUndefined(); + + await vi.advanceTimersByTimeAsync(100); + expect(failure).toBeInstanceOf(AggregateError); + expect(registry.httpRoutes).toEqual([]); + expect(() => context?.gatewayEvents?.emit("late", {}, { scope: "operator.read" })).toThrow( + "no longer active", + ); + expect(broadcastPluginEvent).not.toHaveBeenCalled(); + } finally { + releaseCleanup?.(); + await stopping; + await sidecars; + vi.useRealTimers(); + } + }); + it("reports deferred plugin services after core startup returns", async () => { await withEnvAsync( { OPENCLAW_SKIP_CHANNELS: undefined, OPENCLAW_SKIP_PROVIDERS: undefined }, diff --git a/src/gateway/server-startup-post-attach.ts b/src/gateway/server-startup-post-attach.ts index cace17fe8195..a8a52835eba7 100644 --- a/src/gateway/server-startup-post-attach.ts +++ b/src/gateway/server-startup-post-attach.ts @@ -35,6 +35,7 @@ import type { GatewayBroadcastToConnIdsFn } from "./server-broadcast-types.js"; import type { GatewayControlUiRootLifecycle } from "./server-control-ui-root.js"; import type { GatewayRecoveryRuntime } from "./server-instance-runtime.types.js"; import type { GatewayClient, GatewayContextResolver } from "./server-methods/shared-types.js"; +import type { GatewayPluginRuntimeClaim } from "./server-plugin-runtime-generation.js"; import type { GatewayResidentRegistry } from "./server-resident-registry.js"; import type { refreshLatestUpdateRestartSentinel } from "./server-restart-sentinel.js"; import type { GatewaySidecarStartupMode } from "./server-sidecar-startup-mode.js"; @@ -579,6 +580,7 @@ export async function startGatewaySidecars(params: { onPostReadySidecars?: (sidecars: GatewayPostReadySidecarHandle[]) => void; shouldCreatePostReadySidecars?: () => boolean; shouldStartPluginServices?: () => boolean; + pluginRuntimeClaim?: GatewayPluginRuntimeClaim; broadcastPluginEvent?: import("./server-broadcast-types.js").GatewayPluginEventBroadcastFn; log: { warn: (msg: string) => void }; logHooks: { @@ -694,7 +696,10 @@ export async function startGatewaySidecars(params: { await Promise.all([accountStartGateRelease, channelStart]); }); - const shouldStartPluginServices = params.shouldStartPluginServices?.() !== false; + await params.pluginRuntimeClaim?.waitForUnblocked(); + const shouldStartPluginServices = + params.pluginRuntimeClaim?.isCurrent() !== false && + params.shouldStartPluginServices?.() !== false; let pluginServicesOwner: PluginServicesHandle | null = null; let pluginServicesStopRequested = false; let resolvePluginServicesOwner: ((handle: PluginServicesHandle | null) => void) | undefined; @@ -702,11 +707,37 @@ export async function startGatewaySidecars(params: { const ownedPluginServices = createDeferredCore(); let stopPromise: Promise | undefined; pluginServicesOwner = { - stop: () => { + stop: (options) => { pluginServicesStopRequested = true; - return (stopPromise ??= ownedPluginServices.promise.then(async (handle) => { - await handle?.stop(); - })); + stopPromise ??= ownedPluginServices.promise.then(async (handle) => { + await handle?.stop(options); + }); + if (!options?.strict) { + return stopPromise; + } + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => { + reject( + new AggregateError( + [new Error("Gateway plugin service startup did not settle before replacement")], + "Gateway plugin service replacement cleanup failed", + ), + ); + }, + Math.max(0, options.deadlineAtMs - Date.now()), + ); + void stopPromise?.then( + () => { + clearTimeout(timer); + resolve(); + }, + (error: unknown) => { + clearTimeout(timer); + reject(error instanceof Error ? error : new Error(String(error))); + }, + ); + }); }, }; resolvePluginServicesOwner = ownedPluginServices.resolve; @@ -1126,12 +1157,17 @@ export async function startGatewayPostAttachRuntime( loadStartupPlugins?: () => Awaitable<{ pluginRegistry: PluginRegistry; gatewayMethods: string[]; + retireGatewayRuntimeBindings?: () => void; }>; onStartupPluginsLoading?: () => void; onStartupPluginsLoaded?: (result: { pluginRegistry: PluginRegistry; gatewayMethods: string[]; + retireGatewayRuntimeBindings?: () => void; }) => Awaitable; + pluginRuntimeClaim?: GatewayPluginRuntimeClaim; + getCurrentPluginRegistry?: () => PluginRegistry; + getCurrentPluginMetadataSnapshot?: () => PluginMetadataSnapshot | undefined; getCronService?: () => PluginHookGatewayCronService | null | undefined; onChannelsStarted?: () => Awaitable; onPluginServices?: (pluginServices: PluginServicesHandle | null) => void; @@ -1195,6 +1231,7 @@ export async function startGatewayPostAttachRuntime( let startupPluginsLoadPromise: Promise<{ pluginRegistry: PluginRegistry; gatewayMethods: string[]; + retireGatewayRuntimeBindings?: () => void; }> | null = null; const loadStartupPluginsIfNeeded = async () => { if (params.minimalTestGateway || !params.loadStartupPlugins) { @@ -1208,6 +1245,13 @@ export async function startGatewayPostAttachRuntime( const loaded = await measureStartup(params.startupTrace, "plugins.runtime-post-bind", () => params.loadStartupPlugins!(), ); + await params.pluginRuntimeClaim?.waitForUnblocked(); + if (params.pluginRuntimeClaim?.isCurrent() === false) { + loaded.retireGatewayRuntimeBindings?.(); + pluginRegistry = params.getCurrentPluginRegistry?.() ?? pluginRegistry; + startupPluginsLoaded = true; + return { pluginRegistry, gatewayMethods: [] }; + } pluginRegistry = loaded.pluginRegistry; startupPluginsLoaded = true; params.startupTrace?.detail("plugins.runtime-post-bind", [ @@ -1293,6 +1337,9 @@ export async function startGatewayPostAttachRuntime( let pluginServicesReported = false; let reportedPluginServices: PluginServicesHandle | null = null; const reportPluginServices = (pluginServices: PluginServicesHandle | null) => { + if (params.pluginRuntimeClaim?.isCurrent() === false) { + return; + } pluginServicesReported = true; reportedPluginServices = pluginServices; params.onPluginServices?.(pluginServices); @@ -1348,12 +1395,16 @@ export async function startGatewayPostAttachRuntime( const loaderStatsBefore = getPluginModuleLoaderStats(); const result = await (async () => { try { + const startupRuntimeCurrent = params.pluginRuntimeClaim?.isCurrent() !== false; + const pluginMetadataSnapshot = startupRuntimeCurrent + ? params.pluginMetadataSnapshot + : params.getCurrentPluginMetadataSnapshot?.(); return await measureStartup(params.startupTrace, "sidecars.total", () => runtimeDeps.startGatewaySidecars({ - cfg: params.gatewayPluginConfigAtStart, - ...(params.pluginMetadataSnapshot - ? { pluginMetadataSnapshot: params.pluginMetadataSnapshot } - : {}), + cfg: startupRuntimeCurrent + ? params.gatewayPluginConfigAtStart + : params.getConfig(), + ...(pluginMetadataSnapshot ? { pluginMetadataSnapshot } : {}), pluginRegistry, defaultWorkspaceDir: params.defaultWorkspaceDir, deps: params.deps, @@ -1370,7 +1421,12 @@ export async function startGatewayPostAttachRuntime( params.onPostReadySidecars?.(sidecars); }, shouldCreatePostReadySidecars: () => params.isClosing?.() !== true, - shouldStartPluginServices: () => params.isClosing?.() !== true, + shouldStartPluginServices: () => + params.isClosing?.() !== true && + params.pluginRuntimeClaim?.isCurrent() !== false, + ...(params.pluginRuntimeClaim + ? { pluginRuntimeClaim: params.pluginRuntimeClaim } + : {}), broadcastPluginEvent: params.broadcastPluginEvent, startupOutcomes, waitForPostReadyWork: params.waitForPostReadyWork, diff --git a/src/plugins/http-registry.test.ts b/src/plugins/http-registry.test.ts index 7723f5a6e732..141d7dd7b2fd 100644 --- a/src/plugins/http-registry.test.ts +++ b/src/plugins/http-registry.test.ts @@ -81,6 +81,33 @@ function createLoggedRouteHarness() { }; } +function createTrackedRouteLease() { + let active = true; + const cleanups = new Set<() => void>(); + const lease = { + isActive: () => active, + retain: vi.fn((cleanup: () => void) => { + const release = () => { + if (cleanups.delete(release)) { + cleanup(); + } + }; + cleanups.add(release); + return release; + }), + }; + return { + lease, + cleanups, + revoke: () => { + active = false; + for (const cleanup of cleanups) { + cleanup(); + } + }, + }; +} + describe("registerPluginHttpRoute", () => { afterEach(() => { resetPluginRuntimeStateForTest(); @@ -641,4 +668,203 @@ describe("registerPluginHttpRoute", () => { unregister(); expect(scopedRegistry.httpRoutes).toHaveLength(0); }); + + it("tracks exact scoped route cleanup without removing unrelated routes", () => { + const registry = createEmptyPluginRegistry(); + const owner = createTrackedRouteLease(); + registerPluginHttpRoute({ + path: "/unrelated-webhook", + auth: "plugin", + handler: vi.fn(), + registry, + }); + + const cleanup = withPluginHttpRouteRegistry( + registry, + () => [ + registerPluginHttpRoute({ + path: "/leased-anonymous-webhook", + auth: "plugin", + handler: vi.fn(), + }), + registerPluginHttpRoute({ + path: "/leased-owned-webhook", + auth: "plugin", + handler: vi.fn(), + pluginId: "demo", + }), + ], + owner.lease, + ); + + expect(owner.cleanups.size).toBe(2); + cleanup[0]?.(); + expect(owner.cleanups.size).toBe(1); + expect(registry.httpRoutes.map((route) => route.path)).toEqual([ + "/unrelated-webhook", + "/leased-owned-webhook", + ]); + owner.revoke(); + cleanup[0]?.(); + cleanup[1]?.(); + owner.revoke(); + expect(owner.cleanups.size).toBe(0); + expect(registry.httpRoutes.map((route) => route.path)).toEqual(["/unrelated-webhook"]); + }); + + it.each([ + { name: "anonymous", pluginId: undefined, explicitRegistry: false, nestedScope: false }, + { name: "plugin-owned", pluginId: "demo", explicitRegistry: false, nestedScope: false }, + { + name: "anonymous with an explicit registry", + pluginId: undefined, + explicitRegistry: true, + nestedScope: false, + }, + { + name: "anonymous through a nested scope", + pluginId: undefined, + explicitRegistry: false, + nestedScope: true, + }, + ])( + "rejects late $name route registration after its async service lease expires", + async ({ pluginId, explicitRegistry, nestedScope }) => { + const registry = createEmptyPluginRegistry(); + const cleanups: Array<() => void> = []; + let active = true; + let releaseContinuation: (() => void) | undefined; + const continuation = new Promise((resolve) => { + releaseContinuation = resolve; + }); + + const lateRegistration = withPluginHttpRouteRegistry( + registry, + async () => { + await continuation; + const register = () => + registerPluginHttpRoute({ + path: "/late-webhook", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + ...(pluginId ? { pluginId } : {}), + ...(explicitRegistry ? { registry } : {}), + }); + if (nestedScope) { + withPluginHttpRouteRegistry(registry, register); + } else { + register(); + } + }, + { + isActive: () => active, + retain: (unregister) => { + cleanups.push(unregister); + return unregister; + }, + }, + ); + + active = false; + releaseContinuation?.(); + + await expect(lateRegistration).rejects.toThrow( + "plugin service HTTP route lease is no longer active", + ); + expect(registry.httpRoutes).toHaveLength(0); + expect(cleanups).toHaveLength(0); + }, + ); + + it("preserves non-throwing registration behavior for an expired route lease", () => { + const registry = createEmptyPluginRegistry(); + const messages: string[] = []; + + const unregister = withPluginHttpRouteRegistry( + registry, + () => + registerPluginHttpRoute({ + path: "/late-webhook", + auth: "plugin", + handler: vi.fn(), + log: (message) => messages.push(message), + }), + { isActive: () => false, retain: vi.fn() }, + ); + + expect(messages).toEqual(["plugin service HTTP route lease is no longer active"]); + expect(registry.httpRoutes).toHaveLength(0); + expect(() => unregister()).not.toThrow(); + }); + + it.each([ + { name: "a different nested registry", childLease: false, expiredOwner: "parent" }, + { name: "a supplied replacement lease", childLease: true, expiredOwner: "parent" }, + { name: "an expired child under an active parent", childLease: true, expiredOwner: "child" }, + ])("rejects expired ambient route authority through $name", ({ childLease, expiredOwner }) => { + const parentRegistry = createEmptyPluginRegistry(); + const nestedRegistry = createEmptyPluginRegistry(); + const parent = createTrackedRouteLease(); + const child = createTrackedRouteLease(); + (expiredOwner === "parent" ? parent : child).revoke(); + + expect(() => + withPluginHttpRouteRegistry( + parentRegistry, + () => + withPluginHttpRouteRegistry( + nestedRegistry, + () => + registerPluginHttpRoute({ + path: "/independent-webhook", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + }), + childLease ? child.lease : undefined, + ), + parent.lease, + ), + ).toThrow("plugin service HTTP route lease is no longer active"); + + expect(parentRegistry.httpRoutes).toHaveLength(0); + expect(nestedRegistry.httpRoutes).toHaveLength(0); + expect(parent.lease.retain).not.toHaveBeenCalled(); + expect(child.lease.retain).not.toHaveBeenCalled(); + }); + + it("releases nested routes from every ambient lease when any owner revokes", () => { + const parentRegistry = createEmptyPluginRegistry(); + const nestedRegistry = createEmptyPluginRegistry(); + const parent = createTrackedRouteLease(); + const child = createTrackedRouteLease(); + + const unregister = withPluginHttpRouteRegistry( + parentRegistry, + () => + withPluginHttpRouteRegistry( + nestedRegistry, + () => + registerPluginHttpRoute({ + path: "/nested-webhook", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + }), + child.lease, + ), + parent.lease, + ); + + expect(parent.cleanups.size).toBe(1); + expect(child.cleanups.size).toBe(1); + parent.revoke(); + + expect(nestedRegistry.httpRoutes).toHaveLength(0); + expect(parent.cleanups.size).toBe(0); + expect(child.cleanups.size).toBe(0); + unregister(); + child.revoke(); + }); }); diff --git a/src/plugins/http-registry.ts b/src/plugins/http-registry.ts index c0d034a2ef0a..d003b3049ef4 100644 --- a/src/plugins/http-registry.ts +++ b/src/plugins/http-registry.ts @@ -12,11 +12,25 @@ type PluginHttpRouteHandler = ( res: ServerResponse, ) => Promise | boolean | void; -const pluginHttpRouteRegistryScope = new AsyncLocalStorage(); +type PluginHttpRouteRegistrationLease = { + isActive: () => boolean; + retain: (unregister: () => void) => () => void; +}; + +const pluginHttpRouteRegistryScope = new AsyncLocalStorage<{ + registry: PluginRegistry; + leases: readonly PluginHttpRouteRegistrationLease[]; +}>(); const noopUnregister = () => {}; -export function withPluginHttpRouteRegistry(registry: PluginRegistry, run: () => T): T { - return pluginHttpRouteRegistryScope.run(registry, run); +export function withPluginHttpRouteRegistry( + registry: PluginRegistry, + run: () => T, + lease?: PluginHttpRouteRegistrationLease, +): T { + const inherited = pluginHttpRouteRegistryScope.getStore()?.leases ?? []; + const leases = lease && !inherited.includes(lease) ? [...inherited, lease] : inherited; + return pluginHttpRouteRegistryScope.run({ registry, leases }, run); } export function registerPluginHttpRoute(params: { @@ -39,14 +53,8 @@ export function registerPluginHttpRoute(params: { log?: (message: string) => void; registry?: PluginRegistry; }): () => void { - const registry = - params.registry ?? - pluginHttpRouteRegistryScope.getStore() ?? - requireActivePluginHttpRouteRegistry(); - const routes = registry.httpRoutes ?? []; - registry.httpRoutes = routes; - - const normalizedPath = normalizePluginHttpPath(params.path, params.fallbackPath); + const scope = pluginHttpRouteRegistryScope.getStore(); + const registry = params.registry ?? scope?.registry ?? requireActivePluginHttpRouteRegistry(); const suffix = params.accountId ? ` for account "${params.accountId}"` : ""; const rejectRegistration = (message: string): (() => void) => { params.log?.(message); @@ -55,6 +63,15 @@ export function registerPluginHttpRoute(params: { } return noopUnregister; }; + // AsyncLocalStorage survives timed-out service callbacks; expired continuations must not + // regain route authority, even when they retained an explicit registry reference. + if (scope?.leases.some((lease) => !lease.isActive())) { + return rejectRegistration("plugin service HTTP route lease is no longer active"); + } + + const routes = registry.httpRoutes ?? []; + registry.httpRoutes = routes; + const normalizedPath = normalizePluginHttpPath(params.path, params.fallbackPath); if (!normalizedPath) { return rejectRegistration(`plugin: webhook path missing${suffix}`); } @@ -146,10 +163,16 @@ export function registerPluginHttpRoute(params: { }; routes.push(entry); - return () => { + const releases: Array<() => void> = []; + const unregister = () => { const index = routes.indexOf(entry); if (index >= 0) { routes.splice(index, 1); } + for (const release of releases.splice(0)) { + release(); + } }; + scope?.leases.forEach((lease) => releases.push(lease.retain(unregister))); + return unregister; } diff --git a/src/plugins/services.replacement.test.ts b/src/plugins/services.replacement.test.ts new file mode 100644 index 000000000000..9bf99ea5c64d --- /dev/null +++ b/src/plugins/services.replacement.test.ts @@ -0,0 +1,441 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { + emitTrustedDiagnosticEvent, + resetDiagnosticEventsForTest, +} from "../infra/diagnostic-events.js"; +import { + formatPropagatedDiagnosticTraceparent, + resetDiagnosticTracePropagationForTest, +} from "../infra/diagnostic-trace-propagation.js"; +import { + getDiagnosticStabilitySnapshot, + resetDiagnosticStabilityRecorderForTest, + type DiagnosticExporterHealthUpdate, +} from "../logging/diagnostic-stability.js"; +import { queuePluginSessionsChanged } from "./gateway-events.js"; +import { registerPluginHttpRoute, withPluginHttpRouteRegistry } from "./http-registry.js"; +import type { PluginOrigin } from "./plugin-origin.types.js"; +import { createEmptyPluginRegistry } from "./registry.js"; +import { resetPluginRuntimeStateForTest } from "./runtime.js"; +import { listPluginServiceHealthFailures } from "./service-health.js"; +import { startPluginServices, type PluginServicesHandle } from "./services.js"; +import type { OpenClawPluginService, OpenClawPluginServiceContext } from "./types.js"; + +const mockedLogger = vi.hoisted(() => ({ + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + child: vi.fn(() => mockedLogger), +})); + +vi.mock("../logging/subsystem.js", () => ({ + createSubsystemLogger: () => mockedLogger, +})); + +function createRegistry( + services: OpenClawPluginService[], + pluginId = "plugin:test", + origin: PluginOrigin = "workspace", +) { + const registry = createEmptyPluginRegistry(); + registry.services = services.map((service) => ({ + pluginId, + service, + source: "test", + origin, + rootDir: "/plugins/test-plugin", + })) as typeof registry.services; + return registry; +} + +const createServiceConfig = () => ({}) as Parameters[0]["config"]; + +describe("plugin service replacement", () => { + beforeEach(() => { + vi.clearAllMocks(); + resetDiagnosticEventsForTest(); + resetDiagnosticTracePropagationForTest(); + resetDiagnosticStabilityRecorderForTest(); + resetPluginRuntimeStateForTest(); + }); + + it("strictly aggregates ordinary and exporter failures while draining producers first", async () => { + const order: string[] = []; + const ordinaryFailure = new Error("ordinary cleanup rejected"); + const exporterFailure = new Error("exporter cleanup rejected"); + const registry = createRegistry([ + { + id: "ordinary-first", + start: () => {}, + stop: () => { + order.push("ordinary-first"); + emitTrustedDiagnosticEvent({ + type: "log.record", + level: "INFO", + message: "queued before exporter shutdown", + }); + throw ordinaryFailure; + }, + }, + { + id: "ordinary-second", + start: () => {}, + stop: () => { + order.push("ordinary-second"); + }, + }, + ]); + registry.services.push( + ...createRegistry( + [ + { + id: "diagnostics-prometheus", + start: () => {}, + stop: () => { + order.push("prometheus"); + }, + }, + ], + "diagnostics-prometheus", + "bundled", + ).services, + ...createRegistry( + [ + { + id: "diagnostics-otel", + start: (ctx) => { + ctx.internalDiagnostics?.onEvent((event) => { + if (event.type === "log.record") { + order.push("drained"); + } + }); + }, + stop: () => { + order.push("otel"); + throw exporterFailure; + }, + }, + ], + "diagnostics-otel", + "bundled", + ).services, + ); + const handle = await startPluginServices({ registry, config: createServiceConfig() }); + const failure = await handle + .stop({ strict: true, deadlineAtMs: Date.now() + 5_000 }) + .catch((error: unknown) => error); + + expect(failure).toBeInstanceOf(AggregateError); + expect((failure as AggregateError).errors).toEqual([ + expect.objectContaining({ + cause: ordinaryFailure, + message: expect.stringContaining("plugin=plugin:test, service=ordinary-first"), + }), + expect.objectContaining({ + cause: exporterFailure, + message: expect.stringContaining("plugin=diagnostics-otel, service=diagnostics-otel"), + }), + ]); + expect(order).toEqual(["ordinary-second", "ordinary-first", "drained", "otel", "prometheus"]); + }); + + it("bounds strict cleanup and fences timed-out service routes, events, and health", async () => { + vi.useFakeTimers(); + let releaseCleanup: (() => void) | undefined; + const cleanupReleased = new Promise((resolve) => { + releaseCleanup = resolve; + }); + const received = vi.fn(); + const siblingStop = vi.fn(); + const broadcastPluginEvent = vi.fn(); + const lateFailures: unknown[] = []; + const nestedRegistry = createEmptyPluginRegistry(); + let context: OpenClawPluginServiceContext | undefined; + const registry = createRegistry([ + { id: "sibling", start: () => {}, stop: siblingStop }, + { + id: "blocked-cleanup", + start: (ctx) => { + context = ctx; + ctx.gatewayEvents?.onSessionsChanged(received); + registerPluginHttpRoute({ path: "/owned-route", auth: "plugin", handler: vi.fn() }); + }, + stop: async (ctx) => { + await cleanupReleased; + ctx.serviceHealth?.reportFailure(new Error("late stale failure")); + for (const run of [ + () => ctx.gatewayEvents?.emit("late", {}, { scope: "operator.read" }), + () => + registerPluginHttpRoute({ + path: "/late-anonymous-route", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + }), + () => + withPluginHttpRouteRegistry(nestedRegistry, () => + registerPluginHttpRoute({ + path: "/late-nested-route", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + }), + ), + () => + withPluginHttpRouteRegistry( + nestedRegistry, + () => + registerPluginHttpRoute({ + path: "/late-replacement-lease-route", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + }), + { isActive: () => true, retain: (cleanup) => cleanup }, + ), + ]) { + try { + run(); + } catch (error) { + lateFailures.push(error); + } + } + }, + }, + ]); + let stopping: Promise | undefined; + + try { + const handle = await startPluginServices({ + registry, + config: createServiceConfig(), + broadcastPluginEvent, + }); + let failure: unknown; + stopping = handle + .stop({ strict: true, deadlineAtMs: Date.now() + 5_000 }) + .catch((error: unknown) => { + failure = error; + }); + await vi.advanceTimersByTimeAsync(5_000); + + expect(failure).toBeInstanceOf(AggregateError); + expect((failure as AggregateError).errors).toEqual([ + expect.objectContaining({ + message: expect.stringMatching(/plugin=plugin:test, service=blocked-cleanup.*timed out/), + }), + ]); + expect(siblingStop).toHaveBeenCalledOnce(); + expect(registry.httpRoutes).toEqual([]); + expect(() => context?.gatewayEvents?.onSessionsChanged(received)).toThrow("no longer active"); + + releaseCleanup?.(); + await Promise.resolve(); + await Promise.resolve(); + queuePluginSessionsChanged({ sessionKey: "agent:main:main" }); + await Promise.resolve(); + + expect(lateFailures).toHaveLength(4); + expect(received).not.toHaveBeenCalled(); + expect(broadcastPluginEvent).not.toHaveBeenCalled(); + expect(listPluginServiceHealthFailures(registry)).toEqual([]); + expect(registry.httpRoutes).toEqual([]); + expect(nestedRegistry.httpRoutes).toEqual([]); + } finally { + releaseCleanup?.(); + await stopping; + vi.useRealTimers(); + } + }); + + it("honors a replacement deadline inherited after ownership consumed most of its budget", async () => { + vi.useFakeTimers(); + const broadcastPluginEvent = vi.fn(); + let releaseCleanup: (() => void) | undefined; + const cleanupReleased = new Promise((resolve) => { + releaseCleanup = resolve; + }); + let context: OpenClawPluginServiceContext | undefined; + const registry = createRegistry([ + { + id: "late-owner", + start: (serviceContext) => { + context = serviceContext; + registerPluginHttpRoute({ path: "/deadline-route", auth: "plugin", handler: vi.fn() }); + }, + stop: async (serviceContext) => { + await cleanupReleased; + serviceContext.gatewayEvents?.emit("late", {}, { scope: "operator.read" }); + }, + }, + ]); + let stopping: Promise | undefined; + + try { + const handle = await startPluginServices({ + registry, + config: createServiceConfig(), + broadcastPluginEvent, + }); + const deadlineAtMs = Date.now() + 100; + let failure: unknown; + stopping = handle.stop({ strict: true, deadlineAtMs }).catch((error: unknown) => { + failure = error; + }); + + await vi.advanceTimersByTimeAsync(99); + expect(failure).toBeUndefined(); + expect(registry.httpRoutes).toHaveLength(1); + + await vi.advanceTimersByTimeAsync(1); + expect(failure).toBeInstanceOf(AggregateError); + expect(registry.httpRoutes).toEqual([]); + expect(() => context?.gatewayEvents?.emit("late", {}, { scope: "operator.read" })).toThrow( + "no longer active", + ); + expect(broadcastPluginEvent).not.toHaveBeenCalled(); + } finally { + releaseCleanup?.(); + await stopping; + vi.useRealTimers(); + } + }); + + it("bounds strict shutdown while startup is unsettled and revokes its late continuation", async () => { + vi.useFakeTimers(); + let releaseStartup: (() => void) | undefined; + const startupReleased = new Promise((resolve) => { + releaseStartup = resolve; + }); + const broadcastPluginEvent = vi.fn(); + const lateFailures: unknown[] = []; + let lifecycleHandle: PluginServicesHandle | undefined; + const registry = createRegistry([ + { + id: "blocked-startup", + start: async (ctx) => { + await startupReleased; + ctx.serviceHealth?.reportFailure(new Error("late startup failure")); + for (const run of [ + () => ctx.gatewayEvents?.emit("late", {}, { scope: "operator.read" }), + () => + registerPluginHttpRoute({ + path: "/late-startup-route", + auth: "plugin", + handler: vi.fn(), + throwOnFailure: true, + }), + ]) { + try { + run(); + } catch (error) { + lateFailures.push(error); + } + } + }, + }, + ]); + const starting = startPluginServices({ + registry, + config: createServiceConfig(), + broadcastPluginEvent, + onHandle: (handle) => { + lifecycleHandle = handle; + }, + }); + let stopping: Promise | undefined; + + try { + let failure: unknown; + stopping = lifecycleHandle! + .stop({ strict: true, deadlineAtMs: Date.now() + 5_000 }) + .catch((error: unknown) => { + failure = error; + }); + await vi.advanceTimersByTimeAsync(5_000); + + expect(failure).toBeInstanceOf(AggregateError); + expect((failure as AggregateError).errors[0]).toMatchObject({ + message: expect.stringContaining("plugin service startup settlement timed out"), + }); + + releaseStartup?.(); + await starting; + await stopping; + expect(lateFailures).toHaveLength(2); + expect(broadcastPluginEvent).not.toHaveBeenCalled(); + expect(listPluginServiceHealthFailures(registry)).toEqual([]); + expect(registry.httpRoutes).toEqual([]); + } finally { + releaseStartup?.(); + await starting; + await stopping; + vi.useRealTimers(); + } + }); + + it("revokes trusted diagnostics listeners, emitters, bridges, and exporter health with their service", async () => { + const listener = vi.fn(); + const lateListener = vi.fn(); + const traceContext = { + traceId: "1234567890abcdef1234567890abcdef", + spanId: "1234567890abcdef", + }; + let context: OpenClawPluginServiceContext | undefined; + const registry = createRegistry( + [ + { + id: "diagnostics-otel", + start: (ctx) => { + context = ctx; + ctx.internalDiagnostics?.onEvent(listener); + ctx.internalDiagnostics?.registerTracePropagationBridge?.({ + resolveTraceContext: () => undefined, + }); + registerPluginHttpRoute({ path: "/exporter-route", auth: "plugin", handler: vi.fn() }); + }, + }, + ], + "diagnostics-otel", + "bundled", + ); + const handle = await startPluginServices({ registry, config: createServiceConfig() }); + + expect(formatPropagatedDiagnosticTraceparent(traceContext)).toBeUndefined(); + await handle.stop(); + + expect(() => + context?.internalDiagnostics?.emit({ type: "log.record", level: "INFO", message: "late" }), + ).toThrow("no longer active"); + expect(() => context?.internalDiagnostics?.onEvent(lateListener)).toThrow("no longer active"); + expect(() => + context?.internalDiagnostics?.registerTracePropagationBridge?.({ + resolveTraceContext: () => undefined, + }), + ).toThrow("no longer active"); + ( + context?.internalDiagnostics as + | (NonNullable & { + reportExporterHealth?: (update: DiagnosticExporterHealthUpdate) => void; + }) + | undefined + )?.reportExporterHealth?.({ + signal: "traces", + transport: "otlp-http-protobuf", + status: "failure", + reason: "export_failed", + }); + emitTrustedDiagnosticEvent({ type: "log.record", level: "INFO", message: "still active" }); + + expect(listener).not.toHaveBeenCalled(); + expect(lateListener).not.toHaveBeenCalled(); + expect(formatPropagatedDiagnosticTraceparent(traceContext)).toBe( + "00-1234567890abcdef1234567890abcdef-1234567890abcdef-01", + ); + expect( + getDiagnosticStabilitySnapshot({ type: "telemetry.exporter", limit: 1000 }).events, + ).toEqual([]); + expect(registry.httpRoutes).toEqual([]); + }); +}); diff --git a/src/plugins/services.ts b/src/plugins/services.ts index 427c152a012a..d01b812ab1ea 100644 --- a/src/plugins/services.ts +++ b/src/plugins/services.ts @@ -25,12 +25,55 @@ import { encodeStartupTraceSegment } from "./startup-trace-segment.js"; import type { OpenClawPluginServiceContext, PluginLogger } from "./types.js"; const log = createSubsystemLogger("plugins"); +export const PLUGIN_SERVICE_REPLACEMENT_STOP_TIMEOUT_MS = 5_000; + +class PluginServiceReplacementTimeoutError extends Error {} + type TrustedExporterInternalDiagnostics = NonNullable< OpenClawPluginServiceContext["internalDiagnostics"] > & { reportExporterHealth: (update: DiagnosticExporterHealthUpdate) => void; }; +function createPluginServiceCapabilityLease() { + let active = true; + const cleanups = new Set<() => void>(); + const assertActive = (capability: string) => { + if (!active) { + throw new Error(`plugin service ${capability} is no longer active`); + } + }; + const retain = (cleanup: () => void): (() => void) => { + if (!active) { + cleanup(); + assertActive("capability lease"); + } + const release = () => { + if (cleanups.delete(release)) { + cleanup(); + } + }; + cleanups.add(release); + return release; + }; + return { + isActive: () => active, + assertActive, + retain, + revoke: () => { + if (!active) { + return; + } + active = false; + for (const cleanup of cleanups) { + cleanup(); + } + }, + }; +} + +type PluginServiceCapabilityLease = ReturnType; + function createPluginLogger(): PluginLogger { return { info: (msg) => log.info(msg), @@ -47,6 +90,7 @@ function createServiceContext(params: { service: PluginServiceRegistration; serviceHealth: NonNullable; gatewayEvents?: OpenClawPluginServiceContext["gatewayEvents"]; + lease: PluginServiceCapabilityLease; }): OpenClawPluginServiceContext { const isDiagnosticsExporter = params.service?.pluginId === params.service?.service.id && @@ -59,14 +103,26 @@ function createServiceContext(params: { const internalDiagnostics: TrustedExporterInternalDiagnostics | undefined = grantsInternalDiagnostics ? { - emit: emitTrustedDiagnosticEventWithPrivateData, - onEvent: isOtelExporter - ? (listener) => - onTrustedInternalDiagnosticEvent(markTrustedOtelDiagnosticListener(listener)) - : onTrustedInternalDiagnosticEvent, - registerTracePropagationBridge: registerDiagnosticTracePropagationBridge, - reportExporterHealth: (update) => - recordDiagnosticExporterHealth(params.service.service.id, update), + emit: (event, privateData) => { + params.lease.assertActive("internal diagnostic emitter"); + emitTrustedDiagnosticEventWithPrivateData(event, privateData); + }, + onEvent: (listener) => { + params.lease.assertActive("internal diagnostic listener"); + const trustedListener = isOtelExporter + ? markTrustedOtelDiagnosticListener(listener) + : listener; + return params.lease.retain(onTrustedInternalDiagnosticEvent(trustedListener)); + }, + registerTracePropagationBridge: (bridge) => { + params.lease.assertActive("diagnostic trace propagation bridge"); + return params.lease.retain(registerDiagnosticTracePropagationBridge(bridge)); + }, + reportExporterHealth: (update) => { + if (params.lease.isActive()) { + recordDiagnosticExporterHealth(params.service.service.id, update); + } + }, } : undefined; @@ -92,26 +148,22 @@ function createServiceContext(params: { function createScopedGatewayEvents(params: { pluginId: string; broadcast?: GatewayPluginEventBroadcastFn; + lease: PluginServiceCapabilityLease; }): { gatewayEvents?: OpenClawPluginServiceContext["gatewayEvents"]; - revoke: () => void; } { // No broadcaster means no gateway events at all: emits have nowhere to go and // sessions.changed is queued by the broadcaster itself. Omitting the facade // keeps `ctx.gatewayEvents` presence as the capability signal plugins // feature-detect; a silently dropping emit would defeat their fallbacks. if (!params.broadcast) { - return { revoke: () => undefined }; + return {}; } const broadcast = params.broadcast; - let active = true; - const subscriptions = new Set<() => void>(); return { gatewayEvents: { emit: (event, payload: PluginJsonValue, opts) => { - if (!active) { - throw new Error("plugin service gateway event emitter is no longer active"); - } + params.lease.assertActive("gateway event emitter"); if (!/^[a-z][a-z0-9_-]*$/u.test(event)) { throw new Error(`invalid plugin gateway event name: ${event}`); } @@ -128,29 +180,10 @@ function createScopedGatewayEvents(params: { broadcast(`plugin.${params.pluginId}.${event}`, payload, opts.scope); }, onSessionsChanged: (handler) => { - if (!active) { - throw new Error("plugin service gateway event subscriber is no longer active"); - } - const unsubscribe = subscribePluginSessionsChanged(handler); - let subscribed = true; - const release = () => { - if (!subscribed) { - return; - } - subscribed = false; - subscriptions.delete(release); - unsubscribe(); - }; - subscriptions.add(release); - return release; + params.lease.assertActive("gateway event subscriber"); + return params.lease.retain(subscribePluginSessionsChanged(handler)); }, }, - revoke: () => { - active = false; - for (const unsubscribe of subscriptions) { - unsubscribe(); - } - }, }; } @@ -178,7 +211,7 @@ function createScopedPluginServiceStartupTrace( } export type PluginServicesHandle = { - stop: () => Promise; + stop: (options?: { strict: true; deadlineAtMs: number }) => Promise; }; type PluginServiceStartupTrace = { @@ -197,22 +230,71 @@ export async function startPluginServices(params: { const healthGeneration = createPluginServiceHealthGeneration(params.registry); const running: Array<{ id: string; + pluginId: string; diagnosticsExporter: boolean; stop?: () => void | Promise; - revokeGatewayEvents: () => void; - revokeServiceHealth: () => void; + lease: PluginServiceCapabilityLease; }> = []; - const stopService = async (entry: (typeof running)[number], failures?: unknown[]) => { + const runBeforeDeadline = async ( + run: () => void | Promise, + deadline: number, + label: string, + owner?: string, + ): Promise => { + const operation = Promise.resolve(run()); + const remaining = deadline - Date.now(); + const timeoutError = () => + new PluginServiceReplacementTimeoutError( + `${label} timed out after ${PLUGIN_SERVICE_REPLACEMENT_STOP_TIMEOUT_MS}ms${owner ? ` (${owner})` : ""}`, + ); + if (remaining <= 0) { + await Promise.race([operation, Promise.reject(timeoutError())]); + return; + } + let timer: ReturnType | undefined; + try { + await Promise.race([ + operation, + new Promise((_, reject) => { + timer = setTimeout(() => reject(timeoutError()), remaining); + timer.unref?.(); + }), + ]); + } finally { + clearTimeout(timer); + } + }; + const stopService = async ( + entry: (typeof running)[number], + failures?: unknown[], + deadline?: number, + ) => { try { if (entry.stop) { - await withPluginHttpRouteRegistry(params.registry, () => entry.stop?.()); + const cleanup = () => + withPluginHttpRouteRegistry(params.registry, () => entry.stop?.(), entry.lease); + if (deadline === undefined) { + await cleanup(); + } else { + await runBeforeDeadline(cleanup, deadline, "plugin service stop"); + } } } catch (err) { log.warn(`plugin service stop failed (${entry.id}): ${String(err)}`); - failures?.push(err); + failures?.push( + deadline === undefined + ? err + : new Error( + `plugin service stop failed (plugin=${entry.pluginId}, service=${entry.id}): ${ + err instanceof PluginServiceReplacementTimeoutError + ? err.message + : `rejected: ${String(err)}` + }`, + { cause: err }, + ), + ); } finally { - entry.revokeGatewayEvents(); - entry.revokeServiceHealth(); + entry.lease.revoke(); } }; const startupSettled = createDeferredCore(); @@ -220,28 +302,70 @@ export async function startPluginServices(params: { let stopRequested = false; let stopPromise: Promise | undefined; const handle: PluginServicesHandle = { - stop: () => { + stop: (options) => { stopRequested = true; // Store the shared promise before plugin cleanup runs so shutdown cannot start twice. if (!stopPromise) { + const strict = options?.strict === true; + const deadline = strict ? options.deadlineAtMs : undefined; stopPromise = Promise.resolve().then(async () => { - await startupSettled.promise.catch(() => {}); + const failures: unknown[] = []; + if (deadline === undefined) { + await startupSettled.promise.catch(() => {}); + } else { + try { + const starting = running.at(-1); + await runBeforeDeadline( + () => startupSettled.promise.catch(() => {}), + deadline, + "plugin service startup settlement", + starting ? `plugin=${starting.pluginId}, service=${starting.id}` : undefined, + ); + } catch (error) { + failures.push(error); + // Startup may resume after replacement timed out; its issued capabilities die now. + for (const entry of running) { + entry.lease.revoke(); + } + } + } const reversed = running.toReversed(); const diagnosticsExporters = reversed.filter((entry) => entry.diagnosticsExporter); - const exporterFailures: unknown[] = []; - const stopServices = async (services: typeof reversed, failures?: unknown[]) => { + const exporterFailures = strict ? failures : []; + const stopServices = async (services: typeof reversed, collected?: unknown[]) => { for (const entry of services) { - await stopService(entry, failures); + await stopService(entry, collected, deadline); } }; - await stopServices(reversed.filter((entry) => !entry.diagnosticsExporter)); + await stopServices( + reversed.filter((entry) => !entry.diagnosticsExporter), + strict ? failures : undefined, + ); if (diagnosticsExporters.length > 0) { // Producers stop first; this barrier preserves their queued tail before exporters detach. - await waitForDiagnosticEventsDrained(); + if (deadline === undefined) { + await waitForDiagnosticEventsDrained(); + } else { + try { + await runBeforeDeadline( + waitForDiagnosticEventsDrained, + deadline, + "plugin diagnostic event drain", + diagnosticsExporters + .map((entry) => `plugin=${entry.pluginId}, service=${entry.id}`) + .join("; "), + ); + } catch (error) { + failures.push(error); + } + } } // Ordinary plugin cleanup stays warn-and-continue. Trusted diagnostics // exporter failures propagate because they can mean telemetry was lost. await stopServices(diagnosticsExporters, exporterFailures); + if (strict && failures.length > 0) { + throw new AggregateError(failures, "plugin service replacement cleanup failed"); + } if (exporterFailures.length === 1) { throw exporterFailures[0]; } @@ -267,11 +391,14 @@ export async function startPluginServices(params: { } const service = entry.service; const traceName = createPluginServiceTraceName(entry); + const lease = createPluginServiceCapabilityLease(); const scopedGatewayEvents = createScopedGatewayEvents({ pluginId: entry.pluginId, broadcast: params.broadcastPluginEvent, + lease, }); const serviceHealth = healthGeneration.createReporter(entry); + lease.retain(serviceHealth.revoke); const serviceContext = createServiceContext({ config: params.config, startupTrace: params.startupTrace, @@ -279,24 +406,27 @@ export async function startPluginServices(params: { service: entry, serviceHealth: serviceHealth.health, gatewayEvents: scopedGatewayEvents.gatewayEvents, + lease, }); const runningService = { id: service.id, + pluginId: entry.pluginId, diagnosticsExporter: serviceContext.internalDiagnostics !== undefined, stop: service.stop ? () => service.stop?.(serviceContext) : undefined, - revokeGatewayEvents: scopedGatewayEvents.revoke, - revokeServiceHealth: serviceHealth.revoke, + lease, }; + // Own capabilities before startup yields so a bounded replacement can revoke stale work. + running.push(runningService); try { const startService = () => - withPluginHttpRouteRegistry(params.registry, () => service.start(serviceContext)); + withPluginHttpRouteRegistry(params.registry, () => service.start(serviceContext), lease); if (params.startupTrace) { await params.startupTrace.measure(traceName, startService); } else { await startService(); } - running.push(runningService); } catch (err) { + running.splice(running.indexOf(runningService), 1); failedCount += 1; serviceContext.serviceHealth?.reportFailure(err); const error = err as Error;