mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 12:56:01 -06:00
fix(gateway): own plugin reload generations (#126627)
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
a9ddc2fb1d
commit
f20c6dacc3
+151
-117
@@ -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<ChannelId>,
|
||||
accounts?: ReadonlyMap<ChannelId, ReadonlySet<string>>,
|
||||
) => Promise<void>;
|
||||
commitRuntime: () => Promise<void>;
|
||||
env: NodeJS.ProcessEnv;
|
||||
isAborted?: () => boolean;
|
||||
}): Promise<GatewayPluginReloadResult> => {
|
||||
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<ChannelId>): 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<string>();
|
||||
const nextStartupPluginIds = new Set(nextPluginLookUpTable.startup.pluginIds);
|
||||
const nextStartupChannelIds = new Set<ChannelId>();
|
||||
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<ChannelId>(
|
||||
nextPluginLookUpTable.manifestRegistry.plugins.flatMap(({ id, channels }) =>
|
||||
nextStartupPluginIds.has(id) ? (channels.length > 0 ? channels : [id]) : [],
|
||||
),
|
||||
);
|
||||
const channelsToStopBeforeReplace = new Set<ChannelId>();
|
||||
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<typeof prepareGatewayPluginLoad> | 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<ChannelId>();
|
||||
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<ChannelId>();
|
||||
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()),
|
||||
};
|
||||
};
|
||||
|
||||
|
||||
@@ -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<string>(
|
||||
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;
|
||||
},
|
||||
|
||||
@@ -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);
|
||||
},
|
||||
);
|
||||
});
|
||||
@@ -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<boolean>;
|
||||
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<typeof createDeferredCore<void>>;
|
||||
}
|
||||
| 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),
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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<object, number>();
|
||||
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<PropertyKey, unknown>)[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,
|
||||
);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
@@ -69,9 +69,9 @@ export type GatewayGmailRestartAbortController = {
|
||||
export type GatewayHotReloadPublication = {
|
||||
publish: (commit: () => Promise<void>, isCommitted: () => boolean) => Promise<void>;
|
||||
isCurrent: () => boolean;
|
||||
sourceConfig: OpenClawConfig;
|
||||
prepareRestartRuntimeConfig?: () => Promise<OpenClawConfig>;
|
||||
runtimeEnv?: NodeJS.ProcessEnv;
|
||||
sourceConfig?: OpenClawConfig;
|
||||
};
|
||||
|
||||
export type GatewayRestartTransactionState = "pending" | "committed" | "rejected";
|
||||
@@ -151,12 +151,14 @@ export type GatewayReloadHandlerParams = {
|
||||
stopPostReadySidecars?: () => Promise<void> | void;
|
||||
reloadPlugins: (params: {
|
||||
nextConfig: OpenClawConfig;
|
||||
sourceConfig: OpenClawConfig;
|
||||
changedPaths: readonly string[];
|
||||
beforeReplace: (
|
||||
channels: ReadonlySet<ChannelKind>,
|
||||
accounts?: ReadonlyMap<ChannelKind, ReadonlySet<string>>,
|
||||
) => Promise<void>;
|
||||
commitRuntime: () => Promise<void>;
|
||||
commitRuntime: (onCommit?: () => void) => Promise<void>;
|
||||
onReplacementTeardownFailure: (error: unknown) => void;
|
||||
env: NodeJS.ProcessEnv;
|
||||
isAborted?: () => boolean;
|
||||
}) => Promise<GatewayPluginReloadResult>;
|
||||
|
||||
@@ -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<void>;
|
||||
}): Promise<GatewayPluginReloadResult> => {
|
||||
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<ChannelKind>) => Promise<void>;
|
||||
commitRuntime: () => Promise<void>;
|
||||
onReplacementTeardownFailure: (error: unknown) => void;
|
||||
}): Promise<GatewayPluginReloadResult> => {
|
||||
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<void>) => await commit());
|
||||
const reloadPlugins = vi.fn(
|
||||
async (params: {
|
||||
beforeReplace: (channels: ReadonlySet<ChannelKind>) => Promise<void>;
|
||||
commitRuntime: () => Promise<void>;
|
||||
onReplacementTeardownFailure: (error: unknown) => void;
|
||||
}): Promise<GatewayPluginReloadResult> => {
|
||||
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<ChannelKind>) => Promise<void>;
|
||||
}): Promise<GatewayPluginReloadResult> => {
|
||||
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(),
|
||||
},
|
||||
|
||||
@@ -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",
|
||||
);
|
||||
|
||||
@@ -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") {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<void>((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<typeof startGatewaySidecarsImpl>[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<typeof import("../plugins/services.js")>("../plugins/services.js");
|
||||
const registry = createEmptyPluginRegistry();
|
||||
const broadcastPluginEvent = vi.fn();
|
||||
let context: OpenClawPluginServiceContext | undefined;
|
||||
let releaseCleanup: (() => void) | undefined;
|
||||
const cleanupReleased = new Promise<void>((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<void>((resolve) => {
|
||||
setTimeout(resolve, 4_900);
|
||||
});
|
||||
return await actualServices.startPluginServices(
|
||||
params as Parameters<typeof actualServices.startPluginServices>[0],
|
||||
);
|
||||
});
|
||||
const publishedOwner: { current: PluginServicesHandle | null } = { current: null };
|
||||
let stopping: Promise<void> | undefined;
|
||||
let sidecars: ReturnType<typeof startGatewaySidecars> | 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 },
|
||||
|
||||
@@ -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<PluginServicesHandle | null>();
|
||||
let stopPromise: Promise<void> | 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<void>((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<void>;
|
||||
pluginRuntimeClaim?: GatewayPluginRuntimeClaim;
|
||||
getCurrentPluginRegistry?: () => PluginRegistry;
|
||||
getCurrentPluginMetadataSnapshot?: () => PluginMetadataSnapshot | undefined;
|
||||
getCronService?: () => PluginHookGatewayCronService | null | undefined;
|
||||
onChannelsStarted?: () => Awaitable<void>;
|
||||
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,
|
||||
|
||||
@@ -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<void>((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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -12,11 +12,25 @@ type PluginHttpRouteHandler = (
|
||||
res: ServerResponse,
|
||||
) => Promise<boolean | void> | boolean | void;
|
||||
|
||||
const pluginHttpRouteRegistryScope = new AsyncLocalStorage<PluginRegistry>();
|
||||
type PluginHttpRouteRegistrationLease = {
|
||||
isActive: () => boolean;
|
||||
retain: (unregister: () => void) => () => void;
|
||||
};
|
||||
|
||||
const pluginHttpRouteRegistryScope = new AsyncLocalStorage<{
|
||||
registry: PluginRegistry;
|
||||
leases: readonly PluginHttpRouteRegistrationLease[];
|
||||
}>();
|
||||
const noopUnregister = () => {};
|
||||
|
||||
export function withPluginHttpRouteRegistry<T>(registry: PluginRegistry, run: () => T): T {
|
||||
return pluginHttpRouteRegistryScope.run(registry, run);
|
||||
export function withPluginHttpRouteRegistry<T>(
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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<typeof startPluginServices>[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<void>((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<void> | 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<void>((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<void> | 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<void>((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<void> | 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<OpenClawPluginServiceContext["internalDiagnostics"]> & {
|
||||
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([]);
|
||||
});
|
||||
});
|
||||
+185
-55
@@ -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<typeof createPluginServiceCapabilityLease>;
|
||||
|
||||
function createPluginLogger(): PluginLogger {
|
||||
return {
|
||||
info: (msg) => log.info(msg),
|
||||
@@ -47,6 +90,7 @@ function createServiceContext(params: {
|
||||
service: PluginServiceRegistration;
|
||||
serviceHealth: NonNullable<OpenClawPluginServiceContext["serviceHealth"]>;
|
||||
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<void>;
|
||||
stop: (options?: { strict: true; deadlineAtMs: number }) => Promise<void>;
|
||||
};
|
||||
|
||||
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<void>;
|
||||
revokeGatewayEvents: () => void;
|
||||
revokeServiceHealth: () => void;
|
||||
lease: PluginServiceCapabilityLease;
|
||||
}> = [];
|
||||
const stopService = async (entry: (typeof running)[number], failures?: unknown[]) => {
|
||||
const runBeforeDeadline = async (
|
||||
run: () => void | Promise<void>,
|
||||
deadline: number,
|
||||
label: string,
|
||||
owner?: string,
|
||||
): Promise<void> => {
|
||||
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<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
operation,
|
||||
new Promise<never>((_, 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<void> | 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;
|
||||
|
||||
Reference in New Issue
Block a user