Skip to content

Commit 9817813

Browse files
steipetesteipete
committed
fix: refresh deferred plugin runtime surfaces
1 parent 4620e66 commit 9817813

7 files changed

Lines changed: 208 additions & 40 deletions

‎src/gateway/server-runtime-state.test.ts‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,10 @@
11
import { afterEach, describe, expect, it } from "vitest";
22
import { createEmptyPluginRegistry } from "../plugins/registry.js";
33
import {
4+
getActivePluginChannelRegistry,
45
pinActivePluginHttpRouteRegistry,
6+
pinActivePluginChannelRegistry,
7+
releasePinnedPluginChannelRegistry,
58
releasePinnedPluginHttpRouteRegistry,
69
resetPluginRuntimeStateForTest,
710
resolveActivePluginHttpRouteRegistry,
@@ -25,10 +28,11 @@ function createRegistryWithRoute(path: string) {
2528
describe("createGatewayRuntimeState", () => {
2629
afterEach(() => {
2730
releasePinnedPluginHttpRouteRegistry();
31+
releasePinnedPluginChannelRegistry();
2832
resetPluginRuntimeStateForTest();
2933
});
3034

31-
it("releases a post-bootstrap repinned HTTP route registry on cleanup", async () => {
35+
it("releases post-bootstrap repinned plugin registries on cleanup", async () => {
3236
const startupRegistry = createRegistryWithRoute("/startup");
3337
const loadedRegistry = createRegistryWithRoute("/loaded");
3438
const fallbackRegistry = createRegistryWithRoute("/fallback");
@@ -57,10 +61,13 @@ describe("createGatewayRuntimeState", () => {
5761
});
5862

5963
pinActivePluginHttpRouteRegistry(loadedRegistry);
64+
pinActivePluginChannelRegistry(loadedRegistry);
6065
expect(resolveActivePluginHttpRouteRegistry(fallbackRegistry)).toBe(loadedRegistry);
66+
expect(getActivePluginChannelRegistry()).toBe(loadedRegistry);
6167

6268
runtimeState.releasePluginRouteRegistry();
6369

6470
expect(resolveActivePluginHttpRouteRegistry(fallbackRegistry)).toBe(startupRegistry);
71+
expect(getActivePluginChannelRegistry()).toBe(startupRegistry);
6572
});
6673
});

‎src/gateway/server-startup-early.test.ts‎

Lines changed: 59 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,27 @@
1-
import { describe, expect, it } from "vitest";
2-
import { startGatewayEarlyRuntime } from "./server-startup-early.js";
1+
import { beforeEach, describe, expect, it, vi } from "vitest";
2+
3+
const mocks = vi.hoisted(() => ({
4+
getMachineDisplayName: vi.fn(async () => "Test Machine"),
5+
startGatewayDiscovery: vi.fn(async () => ({ bonjourStop: null })),
6+
}));
7+
8+
vi.mock("../infra/machine-name.js", () => ({
9+
getMachineDisplayName: mocks.getMachineDisplayName,
10+
}));
11+
12+
vi.mock("./server-discovery-runtime.js", () => ({
13+
startGatewayDiscovery: mocks.startGatewayDiscovery,
14+
}));
15+
16+
import { startGatewayEarlyRuntime, startGatewayPluginDiscovery } from "./server-startup-early.js";
317

418
describe("startGatewayEarlyRuntime", () => {
19+
beforeEach(() => {
20+
mocks.getMachineDisplayName.mockClear();
21+
mocks.startGatewayDiscovery.mockClear();
22+
mocks.startGatewayDiscovery.mockResolvedValue({ bonjourStop: null });
23+
});
24+
525
it("does not eagerly start the MCP loopback server", async () => {
626
const earlyRuntime = await startGatewayEarlyRuntime({
727
minimalTestGateway: true,
@@ -41,4 +61,41 @@ describe("startGatewayEarlyRuntime", () => {
4161

4262
expect(earlyRuntime).not.toHaveProperty("mcpServer");
4363
});
64+
65+
it("starts discovery with the current plugin registry services", async () => {
66+
const stop = vi.fn(async () => {});
67+
mocks.startGatewayDiscovery.mockResolvedValueOnce({ bonjourStop: stop } as never);
68+
const service = {
69+
pluginId: "bonjour",
70+
service: { id: "bonjour", advertise: vi.fn() },
71+
};
72+
73+
await expect(
74+
startGatewayPluginDiscovery({
75+
minimalTestGateway: false,
76+
cfgAtStart: { discovery: { mdns: { mode: "full" } } } as never,
77+
port: 19_001,
78+
gatewayTls: { enabled: true, fingerprintSha256: "abc123" },
79+
tailscaleMode: "serve" as never,
80+
logDiscovery: {
81+
info: () => {},
82+
warn: () => {},
83+
},
84+
pluginRegistry: {
85+
gatewayDiscoveryServices: [service],
86+
} as never,
87+
}),
88+
).resolves.toBe(stop);
89+
90+
expect(mocks.startGatewayDiscovery).toHaveBeenCalledWith(
91+
expect.objectContaining({
92+
machineDisplayName: "Test Machine",
93+
port: 19_001,
94+
gatewayTls: { enabled: true, fingerprintSha256: "abc123" },
95+
tailscaleMode: "serve",
96+
mdnsMode: "full",
97+
gatewayDiscoveryServices: [service],
98+
}),
99+
);
100+
});
44101
});

‎src/gateway/server-startup-early.ts‎

Lines changed: 33 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,38 @@ import {
1616
import { startGatewayDiscovery } from "./server-discovery-runtime.js";
1717
import { startGatewayMaintenanceTimers } from "./server-maintenance.js";
1818

19+
export async function startGatewayPluginDiscovery(params: {
20+
minimalTestGateway: boolean;
21+
cfgAtStart: OpenClawConfig;
22+
port: number;
23+
gatewayTls: { enabled: boolean; fingerprintSha256?: string };
24+
tailscaleMode: GatewayTailscaleMode;
25+
logDiscovery: {
26+
info: (msg: string) => void;
27+
warn: (msg: string) => void;
28+
};
29+
pluginRegistry?: PluginRegistry;
30+
}): Promise<(() => Promise<void>) | null> {
31+
if (params.minimalTestGateway) {
32+
return null;
33+
}
34+
const machineDisplayName = await getMachineDisplayName();
35+
const discovery = await startGatewayDiscovery({
36+
machineDisplayName,
37+
port: params.port,
38+
gatewayTls: params.gatewayTls.enabled
39+
? { enabled: true, fingerprintSha256: params.gatewayTls.fingerprintSha256 }
40+
: undefined,
41+
wideAreaDiscoveryEnabled: params.cfgAtStart.discovery?.wideArea?.enabled === true,
42+
wideAreaDiscoveryDomain: params.cfgAtStart.discovery?.wideArea?.domain,
43+
tailscaleMode: params.tailscaleMode,
44+
mdnsMode: params.cfgAtStart.discovery?.mdns?.mode,
45+
gatewayDiscoveryServices: params.pluginRegistry?.gatewayDiscoveryServices,
46+
logDiscovery: params.logDiscovery,
47+
});
48+
return discovery.bonjourStop;
49+
}
50+
1951
export async function startGatewayEarlyRuntime(params: {
2052
minimalTestGateway: boolean;
2153
cfgAtStart: OpenClawConfig;
@@ -59,24 +91,7 @@ export async function startGatewayEarlyRuntime(params: {
5991
setSkillsRefreshTimer: (timer: ReturnType<typeof setTimeout> | null) => void;
6092
getRuntimeConfig: () => OpenClawConfig;
6193
}) {
62-
let bonjourStop: (() => Promise<void>) | null = null;
63-
if (!params.minimalTestGateway) {
64-
const machineDisplayName = await getMachineDisplayName();
65-
const discovery = await startGatewayDiscovery({
66-
machineDisplayName,
67-
port: params.port,
68-
gatewayTls: params.gatewayTls.enabled
69-
? { enabled: true, fingerprintSha256: params.gatewayTls.fingerprintSha256 }
70-
: undefined,
71-
wideAreaDiscoveryEnabled: params.cfgAtStart.discovery?.wideArea?.enabled === true,
72-
wideAreaDiscoveryDomain: params.cfgAtStart.discovery?.wideArea?.domain,
73-
tailscaleMode: params.tailscaleMode,
74-
mdnsMode: params.cfgAtStart.discovery?.mdns?.mode,
75-
gatewayDiscoveryServices: params.pluginRegistry?.gatewayDiscoveryServices,
76-
logDiscovery: params.logDiscovery,
77-
});
78-
bonjourStop = discovery.bonjourStop;
79-
}
94+
const bonjourStop = await startGatewayPluginDiscovery(params);
8095

8196
if (!params.minimalTestGateway) {
8297
setSkillsRemoteRegistry(params.nodeRegistry);

‎src/gateway/server-startup-post-attach.test.ts‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -309,6 +309,57 @@ describe("startGatewayPostAttachRuntime", () => {
309309
);
310310
});
311311

312+
it("waits for deferred startup plugin attachment before channel sidecars", async () => {
313+
const events: string[] = [];
314+
let finishAttachment!: () => void;
315+
const attachmentFinished = new Promise<void>((resolve) => {
316+
finishAttachment = () => {
317+
events.push("startup-loaded-end");
318+
resolve();
319+
};
320+
});
321+
const loadedPluginRegistry = {
322+
plugins: [{ id: "acpx", status: "loaded" }],
323+
typedHooks: [],
324+
} as never;
325+
const loadStartupPlugins = vi.fn(async () => ({
326+
pluginRegistry: loadedPluginRegistry,
327+
gatewayMethods: ["ping", "acp.spawn"],
328+
}));
329+
const onStartupPluginsLoaded = vi.fn(() => {
330+
events.push("startup-loaded-start");
331+
return attachmentFinished;
332+
});
333+
const startGatewaySidecars = vi.fn(async () => {
334+
events.push("sidecars");
335+
return { pluginServices: null };
336+
});
337+
338+
const runtimePromise = startGatewayPostAttachRuntime(
339+
{
340+
...createPostAttachParams({
341+
pluginRegistry: {
342+
plugins: [],
343+
typedHooks: [],
344+
} as never,
345+
loadStartupPlugins,
346+
onStartupPluginsLoaded,
347+
}),
348+
},
349+
createPostAttachRuntimeDeps({ startGatewaySidecars }),
350+
);
351+
352+
await vi.waitFor(() => {
353+
expect(events).toEqual(["startup-loaded-start"]);
354+
});
355+
expect(startGatewaySidecars).not.toHaveBeenCalled();
356+
357+
finishAttachment();
358+
await runtimePromise;
359+
360+
expect(events).toEqual(["startup-loaded-start", "startup-loaded-end", "sidecars"]);
361+
});
362+
312363
it("keeps the qmd memory backend lazy by default", async () => {
313364
await startGatewayPostAttachRuntime({
314365
...createPostAttachParams(),

‎src/gateway/server-startup-post-attach.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -588,7 +588,7 @@ export async function startGatewayPostAttachRuntime(
588588
onStartupPluginsLoaded?: (result: {
589589
pluginRegistry: PluginRegistry;
590590
gatewayMethods: string[];
591-
}) => void;
591+
}) => Awaitable<void>;
592592
getCronService?: () => PluginHookGatewayCronService | null | undefined;
593593
onPluginServices?: (pluginServices: PluginServicesHandle | null) => void;
594594
onSidecarsReady?: () => void;
@@ -612,7 +612,7 @@ export async function startGatewayPostAttachRuntime(
612612
params.loadStartupPlugins!(),
613613
);
614614
pluginRegistry = loaded.pluginRegistry;
615-
params.onStartupPluginsLoaded?.(loaded);
615+
await params.onStartupPluginsLoaded?.(loaded);
616616
}
617617

618618
await measureStartup(params.startupTrace, "post-attach.log", () =>

‎src/gateway/server-startup.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
export { startGatewayEarlyRuntime } from "./server-startup-early.js";
1+
export { startGatewayEarlyRuntime, startGatewayPluginDiscovery } from "./server-startup-early.js";
22
export {
33
__testing,
44
startGatewayPostAttachRuntime,

‎src/gateway/server.impl.ts‎

Lines changed: 54 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,10 @@ import {
4242
} from "../plugins/current-plugin-metadata-snapshot.js";
4343
import { runGlobalGatewayStopSafely } from "../plugins/hook-runner-global.js";
4444
import type { PluginHookGatewayCronService } from "../plugins/hook-types.js";
45-
import { pinActivePluginHttpRouteRegistry } from "../plugins/runtime.js";
45+
import {
46+
pinActivePluginChannelRegistry,
47+
pinActivePluginHttpRouteRegistry,
48+
} from "../plugins/runtime.js";
4649
import type { PluginRuntime } from "../plugins/runtime/types.js";
4750
import { getTotalQueueSize } from "../process/command-queue.js";
4851
import type { RuntimeEnv } from "../runtime.js";
@@ -91,7 +94,11 @@ import {
9194
prepareGatewayPluginBootstrap,
9295
} from "./server-startup-plugins.js";
9396
import { STARTUP_UNAVAILABLE_GATEWAY_METHODS } from "./server-startup-unavailable-methods.js";
94-
import { startGatewayEarlyRuntime, startGatewayPostAttachRuntime } from "./server-startup.js";
97+
import {
98+
startGatewayEarlyRuntime,
99+
startGatewayPluginDiscovery,
100+
startGatewayPostAttachRuntime,
101+
} from "./server-startup.js";
95102
import { createWizardSessionTracker } from "./server-wizard-sessions.js";
96103
import { attachGatewayWsHandlers } from "./server-ws-runtime.js";
97104
import { createGatewayEventLoopHealthMonitor } from "./server/event-loop-health.js";
@@ -1001,6 +1008,36 @@ export async function startGatewayServer(
10011008
Object.assign(attachedGatewayExtraHandlers, pluginRegistry.gatewayHandlers);
10021009
attachedPluginGatewayHandlerKeys = new Set(Object.keys(pluginRegistry.gatewayHandlers));
10031010
pinActivePluginHttpRouteRegistry(pluginRegistry);
1011+
pinActivePluginChannelRegistry(pluginRegistry);
1012+
};
1013+
const refreshAttachedGatewayDiscovery = async (nextPluginRegistry: typeof pluginRegistry) => {
1014+
if (minimalTestGateway) {
1015+
return;
1016+
}
1017+
try {
1018+
const stopPreviousDiscovery = runtimeState.bonjourStop;
1019+
runtimeState.bonjourStop = null;
1020+
if (stopPreviousDiscovery) {
1021+
try {
1022+
await stopPreviousDiscovery();
1023+
} catch (err) {
1024+
logDiscovery.warn(
1025+
`gateway discovery stop failed before plugin refresh: ${String(err)}`,
1026+
);
1027+
}
1028+
}
1029+
runtimeState.bonjourStop = await startGatewayPluginDiscovery({
1030+
minimalTestGateway,
1031+
cfgAtStart,
1032+
port,
1033+
gatewayTls,
1034+
tailscaleMode,
1035+
logDiscovery,
1036+
pluginRegistry: nextPluginRegistry,
1037+
});
1038+
} catch (err) {
1039+
logDiscovery.warn(`gateway discovery refresh failed after plugin load: ${String(err)}`);
1040+
}
10041041
};
10051042

10061043
const canvasHostServerPort = (canvasHostServer as CanvasHostServer | null)?.port;
@@ -1084,19 +1121,19 @@ export async function startGatewayServer(
10841121
if (!minimalTestGateway) {
10851122
if (runtimePluginsLoaded && deferredConfiguredChannelPluginIds.length > 0) {
10861123
const { reloadDeferredGatewayPlugins } = await import("./server-plugin-bootstrap.js");
1087-
replaceAttachedPluginRuntime(
1088-
reloadDeferredGatewayPlugins({
1089-
cfg: gatewayPluginConfigAtStart,
1090-
activationSourceConfig: startupActivationSourceConfig,
1091-
workspaceDir: defaultWorkspaceDir,
1092-
log,
1093-
coreGatewayMethodNames: baseMethods,
1094-
baseMethods,
1095-
pluginIds: startupPluginIds,
1096-
pluginLookUpTable,
1097-
logDiagnostics: false,
1098-
}),
1099-
);
1124+
const loaded = reloadDeferredGatewayPlugins({
1125+
cfg: gatewayPluginConfigAtStart,
1126+
activationSourceConfig: startupActivationSourceConfig,
1127+
workspaceDir: defaultWorkspaceDir,
1128+
log,
1129+
coreGatewayMethodNames: baseMethods,
1130+
baseMethods,
1131+
pluginIds: startupPluginIds,
1132+
pluginLookUpTable,
1133+
logDiagnostics: false,
1134+
});
1135+
replaceAttachedPluginRuntime(loaded);
1136+
await refreshAttachedGatewayDiscovery(loaded.pluginRegistry);
11001137
}
11011138
}
11021139

@@ -1171,9 +1208,10 @@ export async function startGatewayServer(
11711208
onStartupPluginsLoading: () => {
11721209
startupPendingReason = "plugin-runtime-deps";
11731210
},
1174-
onStartupPluginsLoaded: (loaded) => {
1211+
onStartupPluginsLoaded: async (loaded) => {
11751212
replaceAttachedPluginRuntime(loaded);
11761213
startupPendingReason = "startup-sidecars";
1214+
await refreshAttachedGatewayDiscovery(loaded.pluginRegistry);
11771215
},
11781216
getCronService: () =>
11791217
runtimeState?.cronState.cron as PluginHookGatewayCronService | undefined,

0 commit comments

Comments
 (0)