mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 14:12:51 +00:00
fix(server): start configured plugins concurrently (#8427)
The daemon accepts connections only after its configured plugins have started, and PluginService.start started them one at a time, so each plugin's build and process launch added to daemon startup. Configured plugins now start together. Their providers and usage sources are still published in config order: a plugin that finishes early waits for the plugins listed before it, so when two plugins claim the same provider or usage-source ID, the one listed first keeps it, as before. A failing plugin still fails on its own. Fixes #7979
This commit is contained in:
@@ -20,6 +20,7 @@ import { DaemonConfigStore } from "../daemon-config-store.js";
|
||||
import { PluginService } from "./index.js";
|
||||
import { BuiltinPluginLoader } from "./builtin/index.js";
|
||||
import { ManagedPluginSources } from "./managed-source.js";
|
||||
import type { PluginProviderMetadata } from "./plugin-process-protocol.js";
|
||||
import {
|
||||
startNpmRegistry,
|
||||
npmPluginPackages,
|
||||
@@ -179,6 +180,55 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) {
|
||||
return { runtime, started, releaseStart, starts };
|
||||
}
|
||||
|
||||
function createGatedRuntime(providers: Record<string, readonly PluginProviderMetadata[]> = {}) {
|
||||
const gates = new Map<string, PromiseWithResolvers<void>>();
|
||||
const gate = (pluginId: string) => {
|
||||
const existing = gates.get(pluginId);
|
||||
if (existing) return existing;
|
||||
const created = Promise.withResolvers<void>();
|
||||
gates.set(pluginId, created);
|
||||
return created;
|
||||
};
|
||||
const begun: string[] = [];
|
||||
const running = new Set<string>();
|
||||
const runtime: TestPluginRuntime = {
|
||||
...emptyUsageRuntime,
|
||||
catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })),
|
||||
invoke: async () => undefined,
|
||||
getLogs: () => [],
|
||||
clearLogs: () => undefined,
|
||||
getProviderRegistrations: (pluginId) =>
|
||||
running.has(pluginId) ? (providers[pluginId] ?? []) : [],
|
||||
startPlugin: async (pluginId) => {
|
||||
begun.push(pluginId);
|
||||
await gate(pluginId).promise;
|
||||
running.add(pluginId);
|
||||
},
|
||||
stopPluginById: async (pluginId) => running.delete(pluginId),
|
||||
stopAll: async () => {
|
||||
running.clear();
|
||||
},
|
||||
subscribe: () => () => undefined,
|
||||
bindPaseoSessionHost: () => undefined,
|
||||
};
|
||||
return {
|
||||
runtime,
|
||||
begun,
|
||||
finish: (pluginId: string) => gate(pluginId).resolve(),
|
||||
fail: (pluginId: string, error: Error) => gate(pluginId).reject(error),
|
||||
};
|
||||
}
|
||||
|
||||
function nextNotification(service: PluginService, pluginId: string): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const unsubscribe = service.subscribe((notified) => {
|
||||
if (notified !== pluginId) return;
|
||||
unsubscribe();
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
describe("PluginService", () => {
|
||||
it.each([
|
||||
{ shadow: false, runtime: "subprocess" },
|
||||
@@ -372,15 +422,127 @@ describe("PluginService", () => {
|
||||
"startup-first": { source: "directory", path: first },
|
||||
"startup-second": { source: "directory", path: second },
|
||||
});
|
||||
const snapshots: string[][] = [];
|
||||
service.subscribe(() => snapshots.push(catalogIds(service)));
|
||||
const notifications: { pluginId: string; running: boolean }[] = [];
|
||||
service.subscribe((pluginId) =>
|
||||
notifications.push({ pluginId, running: catalogIds(service).includes(pluginId) }),
|
||||
);
|
||||
|
||||
await service.start();
|
||||
|
||||
expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]);
|
||||
expect(notifications).toEqual([
|
||||
{ pluginId: "startup-first", running: true },
|
||||
{ pluginId: "startup-second", running: true },
|
||||
]);
|
||||
expect(catalogIds(service)).toEqual(["startup-first", "startup-second"]);
|
||||
await service.stopAllPlugins();
|
||||
}, 20_000);
|
||||
|
||||
it("starts configured plugins concurrently and publishes them in config order", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
const gated = createGatedRuntime();
|
||||
const service = createService(
|
||||
home,
|
||||
{
|
||||
alpha: { source: "directory", path: "/plugins/alpha" },
|
||||
beta: { source: "directory", path: "/plugins/beta" },
|
||||
gamma: { source: "directory", path: "/plugins/gamma" },
|
||||
},
|
||||
{ runtime: gated.runtime },
|
||||
);
|
||||
const notified: string[] = [];
|
||||
service.subscribe((pluginId) => notified.push(pluginId));
|
||||
let started = false;
|
||||
const starting = (async () => {
|
||||
await service.start();
|
||||
started = true;
|
||||
})();
|
||||
|
||||
await expect.poll(() => gated.begun).toEqual(["alpha", "beta", "gamma"]);
|
||||
|
||||
gated.finish("gamma");
|
||||
const alphaPublished = nextNotification(service, "alpha");
|
||||
gated.finish("alpha");
|
||||
await alphaPublished;
|
||||
expect(notified).toEqual(["alpha"]);
|
||||
expect(started).toBe(false);
|
||||
|
||||
gated.finish("beta");
|
||||
await starting;
|
||||
expect(notified).toEqual(["alpha", "beta", "gamma"]);
|
||||
expect(catalogIds(service)).toEqual(["alpha", "beta", "gamma"]);
|
||||
await service.stopAllPlugins();
|
||||
});
|
||||
|
||||
it("starts the other configured plugins when one fails while they are still starting", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
const gated = createGatedRuntime();
|
||||
const service = createService(
|
||||
home,
|
||||
{
|
||||
broken: { source: "directory", path: "/plugins/broken" },
|
||||
healthy: { source: "directory", path: "/plugins/healthy" },
|
||||
},
|
||||
{ runtime: gated.runtime },
|
||||
);
|
||||
const notified: string[] = [];
|
||||
service.subscribe((pluginId) => notified.push(pluginId));
|
||||
const starting = service.start();
|
||||
|
||||
await expect.poll(() => gated.begun).toEqual(["broken", "healthy"]);
|
||||
const brokenPublished = nextNotification(service, "broken");
|
||||
gated.fail("broken", new Error("startup exploded"));
|
||||
await brokenPublished;
|
||||
gated.finish("healthy");
|
||||
await starting;
|
||||
|
||||
expect(notified).toEqual(["broken", "healthy"]);
|
||||
expect(catalogIds(service)).toEqual(["healthy"]);
|
||||
expect(await service.listPlugins()).toMatchObject([
|
||||
{ id: "broken", status: "failed", error: "startup exploded" },
|
||||
{ id: "healthy", status: "running" },
|
||||
]);
|
||||
await service.stopAllPlugins();
|
||||
});
|
||||
|
||||
it("keeps a provider ID with the plugin listed first when both start at once", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
const gated = createGatedRuntime({
|
||||
first: [{ id: "shared-agent", label: "First agent" }],
|
||||
second: [{ id: "shared-agent", label: "Second agent" }],
|
||||
});
|
||||
const service = createService(
|
||||
home,
|
||||
{
|
||||
first: { source: "directory", path: "/plugins/first" },
|
||||
second: { source: "directory", path: "/plugins/second" },
|
||||
},
|
||||
{ runtime: gated.runtime },
|
||||
);
|
||||
const starting = service.start();
|
||||
|
||||
await expect.poll(() => gated.begun).toEqual(["first", "second"]);
|
||||
gated.finish("second");
|
||||
gated.finish("first");
|
||||
await starting;
|
||||
|
||||
expect(service.getProviderRegistrations()).toMatchObject([
|
||||
{ id: "shared-agent", label: "First agent" },
|
||||
]);
|
||||
expect(catalogIds(service)).toEqual(["first"]);
|
||||
expect(await service.listPlugins()).toMatchObject([
|
||||
{ id: "first", status: "running" },
|
||||
{
|
||||
id: "second",
|
||||
status: "failed",
|
||||
error: 'Plugin second cannot register provider ID "shared-agent" twice',
|
||||
},
|
||||
]);
|
||||
await service.stopAllPlugins();
|
||||
});
|
||||
|
||||
it("uses an explicit config key, exposes reload failure, and retries from disk", async () => {
|
||||
const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
|
||||
roots.push(home);
|
||||
|
||||
@@ -230,11 +230,18 @@ export class PluginService {
|
||||
}
|
||||
});
|
||||
if (config.pluginsEnabled === true) {
|
||||
for (const [pluginId, source] of Object.entries(config.plugins ?? {})) {
|
||||
if (source.enabled === false) continue;
|
||||
await this.startConfigured(pluginId);
|
||||
this.notify(pluginId);
|
||||
}
|
||||
let earlierPublished: Promise<unknown> = Promise.resolve();
|
||||
await Promise.all(
|
||||
Object.entries(config.plugins ?? {})
|
||||
.filter(([, source]) => source.enabled !== false)
|
||||
.map(async ([pluginId]) => {
|
||||
const publishTurn = earlierPublished;
|
||||
const starting = this.startConfigured(pluginId, publishTurn);
|
||||
earlierPublished = Promise.all([publishTurn, starting]);
|
||||
await starting;
|
||||
this.notify(pluginId);
|
||||
}),
|
||||
);
|
||||
}
|
||||
this.configStore.onFieldChange("pluginsEnabled", (value) => {
|
||||
this.handleGlobalSwitch(value === true);
|
||||
@@ -560,12 +567,12 @@ export class PluginService {
|
||||
}).catch((error) => this.logger.error({ err: error }, "Failed to enable plugins"));
|
||||
}
|
||||
|
||||
private async startConfigured(pluginId: string): Promise<void> {
|
||||
private async startConfigured(pluginId: string, publishTurn?: Promise<unknown>): Promise<void> {
|
||||
const source = this.configStore.get().plugins?.[pluginId];
|
||||
if (!source || source.enabled === false || !this.canPublish(pluginId)) return;
|
||||
this.errors.delete(pluginId);
|
||||
try {
|
||||
await this.startPlugin(pluginId, source.path);
|
||||
await this.startPlugin(pluginId, source.path, publishTurn);
|
||||
} catch (error) {
|
||||
if (this.canPublish(pluginId)) this.recordFailure(pluginId, error);
|
||||
}
|
||||
@@ -594,8 +601,13 @@ export class PluginService {
|
||||
);
|
||||
}
|
||||
|
||||
private async startPlugin(pluginId: string, sourcePath: string): Promise<void> {
|
||||
private async startPlugin(
|
||||
pluginId: string,
|
||||
sourcePath: string,
|
||||
publishTurn?: Promise<unknown>,
|
||||
): Promise<void> {
|
||||
await this.runtime.startPlugin(pluginId, sourcePath, () => this.canPublish(pluginId));
|
||||
await publishTurn;
|
||||
try {
|
||||
await this.publishProviderRegistrations(pluginId, sourcePath);
|
||||
this.publishUsageSources(pluginId);
|
||||
|
||||
Reference in New Issue
Block a user