mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
Bound managed plugin worker recovery latency
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
445b7c68d5
commit
4b53868e90
2 files changed
+82
-15
No files matched your search
@@ -37,10 +37,10 @@ const manifest: PaperclipPluginManifestV1 = {
|
||||
],
|
||||
};
|
||||
|
||||
function createPlugin(status: string) {
|
||||
function createPlugin(status: string, options: { id?: string; pluginKey?: string } = {}) {
|
||||
return {
|
||||
id: PLUGIN_ID,
|
||||
pluginKey: PLUGIN_KEY,
|
||||
id: options.id ?? PLUGIN_ID,
|
||||
pluginKey: options.pluginKey ?? PLUGIN_KEY,
|
||||
status,
|
||||
manifestJson: manifest,
|
||||
};
|
||||
@@ -213,4 +213,38 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => {
|
||||
]);
|
||||
expect(secondDrivers).toEqual(firstDrivers);
|
||||
});
|
||||
|
||||
it("bounds slow managed bundled recovery attempts without serial waits", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const plugins = [
|
||||
createPlugin("ready"),
|
||||
createPlugin("ready", {
|
||||
id: "plugin-modal",
|
||||
pluginKey: "paperclip.modal-sandbox-provider",
|
||||
}),
|
||||
];
|
||||
mockRegistry.list.mockResolvedValue(plugins);
|
||||
const worker = createWorkerManager();
|
||||
const startWorker = vi.fn(() => new Promise<boolean>(() => {}));
|
||||
|
||||
const driversPromise = listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY, "paperclip.modal-sandbox-provider"],
|
||||
startWorker,
|
||||
timeoutMs: 25,
|
||||
},
|
||||
});
|
||||
await Promise.resolve();
|
||||
|
||||
expect(startWorker).toHaveBeenCalledTimes(2);
|
||||
await vi.advanceTimersByTimeAsync(25);
|
||||
|
||||
await expect(driversPromise).resolves.toEqual([]);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -33,6 +33,7 @@ import type { PluginWorkerManager } from "./plugin-worker-manager.js";
|
||||
export interface ReadyPluginWorkerRecovery {
|
||||
pluginKeys: readonly string[];
|
||||
startWorker(plugin: { id: string; pluginKey: string }): Promise<boolean>;
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
export interface ReadyPluginEnvironmentDriver {
|
||||
@@ -55,6 +56,23 @@ export function pluginDriverProviderKey(config: Pick<PluginEnvironmentConfig, "p
|
||||
return `${config.pluginKey}:${config.driverKey}`;
|
||||
}
|
||||
|
||||
const DEFAULT_READY_PLUGIN_WORKER_RECOVERY_TIMEOUT_MS = 2_000;
|
||||
|
||||
async function resolveWithTimeout<T>(promise: Promise<T>, timeoutMs: number, timeoutValue: T): Promise<T> {
|
||||
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return await promise;
|
||||
let timeout: ReturnType<typeof setTimeout> | null = null;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<T>((resolve) => {
|
||||
timeout = setTimeout(() => resolve(timeoutValue), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timeout) clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
export async function resolvePluginEnvironmentDriver(input: {
|
||||
db: Db;
|
||||
workerManager: PluginWorkerManager;
|
||||
@@ -121,20 +139,35 @@ export async function listReadyPluginEnvironmentDrivers(input: {
|
||||
const pluginRegistry = pluginRegistryService(input.db);
|
||||
const plugins = await pluginRegistry.list();
|
||||
const recoverablePluginKeys = new Set(input.recoverMissingWorker?.pluginKeys ?? []);
|
||||
const readyPlugins = plugins.filter((plugin) => plugin.status === "ready");
|
||||
const recoveryAttempts: Promise<boolean>[] = [];
|
||||
|
||||
for (const plugin of readyPlugins) {
|
||||
const canRecover =
|
||||
!input.workerManager.isRunning(plugin.id)
|
||||
&& recoverablePluginKeys.has(plugin.pluginKey)
|
||||
&& !input.workerManager.getWorker(plugin.id);
|
||||
if (!canRecover || !input.recoverMissingWorker) continue;
|
||||
const timeoutMs =
|
||||
input.recoverMissingWorker.timeoutMs ?? DEFAULT_READY_PLUGIN_WORKER_RECOVERY_TIMEOUT_MS;
|
||||
recoveryAttempts.push(resolveWithTimeout(
|
||||
input.recoverMissingWorker.startWorker({
|
||||
id: plugin.id,
|
||||
pluginKey: plugin.pluginKey,
|
||||
}).catch(() => false),
|
||||
timeoutMs,
|
||||
false,
|
||||
));
|
||||
}
|
||||
|
||||
if (recoveryAttempts.length > 0) {
|
||||
await Promise.all(recoveryAttempts);
|
||||
}
|
||||
|
||||
const rows: ReadyPluginEnvironmentDriver[] = [];
|
||||
for (const plugin of plugins) {
|
||||
if (plugin.status !== "ready") continue;
|
||||
for (const plugin of readyPlugins) {
|
||||
if (!input.workerManager.isRunning(plugin.id)) {
|
||||
const canRecover =
|
||||
recoverablePluginKeys.has(plugin.pluginKey)
|
||||
&& !input.workerManager.getWorker(plugin.id);
|
||||
if (canRecover) {
|
||||
await input.recoverMissingWorker?.startWorker({
|
||||
id: plugin.id,
|
||||
pluginKey: plugin.pluginKey,
|
||||
}).catch(() => false);
|
||||
}
|
||||
if (!input.workerManager.isRunning(plugin.id)) continue;
|
||||
continue;
|
||||
}
|
||||
rows.push(
|
||||
...(plugin.manifestJson.environmentDrivers ?? [])
|
||||
|
||||
Reference in new issue
Block a user