From 445b7c68d540c973e61c1f7ea4527187b940ef08 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Wed, 29 Jul 2026 04:44:39 +0000 Subject: [PATCH] Single-flight managed plugin worker recovery Co-Authored-By: Paperclip --- ...-environment-driver-ready-recovery.test.ts | 216 ++++++++++++++++++ server/src/app.ts | 103 +++++++-- server/src/routes/environments.ts | 11 +- .../src/services/plugin-environment-driver.ts | 79 +++++-- 4 files changed, 373 insertions(+), 36 deletions(-) create mode 100644 server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts diff --git a/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts new file mode 100644 index 00000000000..e41c4c787e0 --- /dev/null +++ b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts @@ -0,0 +1,216 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; +import { createManagedBundledPluginWorkerRecovery } from "../app.js"; +import { listReadyPluginEnvironmentDrivers } from "../services/plugin-environment-driver.js"; +import type { PluginLoader } from "../services/plugin-loader.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; + +const mockRegistry = vi.hoisted(() => ({ + list: vi.fn(), +})); + +vi.mock("../services/plugin-registry.js", () => ({ + pluginRegistryService: () => mockRegistry, +})); + +const PLUGIN_ID = "plugin-daytona"; +const PLUGIN_KEY = "paperclip.daytona-sandbox-provider"; + +const manifest: PaperclipPluginManifestV1 = { + id: PLUGIN_KEY, + apiVersion: 1, + version: "1.0.0", + displayName: "Daytona Sandbox Provider", + description: "Provides Daytona-backed sandboxes.", + author: "Paperclip", + categories: ["automation"], + capabilities: ["environment.drivers.register"], + entrypoints: { worker: "dist/worker.js" }, + environmentDrivers: [ + { + driverKey: "daytona", + kind: "sandbox_provider", + displayName: "Daytona", + description: "Daytona sandbox provider", + configSchema: { type: "object", properties: {} }, + }, + ], +}; + +function createPlugin(status: string) { + return { + id: PLUGIN_ID, + pluginKey: PLUGIN_KEY, + status, + manifestJson: manifest, + }; +} + +function createWorkerManager(options: { + running?: boolean; + hasHandle?: boolean; +} = {}) { + let running = options.running ?? false; + const workerManager = { + isRunning: vi.fn(() => running), + getWorker: vi.fn(() => (options.hasHandle ? { status: "backoff" } : undefined)), + } as unknown as PluginWorkerManager; + return { + workerManager, + markRunning: () => { + running = true; + }, + }; +} + +function createDeferred() { + let resolve!: (value?: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +describe("listReadyPluginEnvironmentDrivers worker recovery", () => { + beforeEach(() => { + mockRegistry.list.mockReset(); + }); + + it("recovers a managed bundled provider that was installed by a sibling process after this process skipped it", async () => { + const plugin = createPlugin("installed"); + mockRegistry.list.mockImplementation(async () => [plugin]); + const worker = createWorkerManager(); + const startWorker = vi.fn(async () => { + worker.markRunning(); + return true; + }); + + await expect( + listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }), + ).resolves.toEqual([]); + expect(startWorker).not.toHaveBeenCalled(); + + plugin.status = "ready"; + const drivers = await listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + + expect(startWorker).toHaveBeenCalledWith({ + id: PLUGIN_ID, + pluginKey: PLUGIN_KEY, + }); + expect(drivers).toEqual([ + expect.objectContaining({ + pluginId: PLUGIN_ID, + pluginKey: PLUGIN_KEY, + driverKey: "daytona", + displayName: "Daytona", + }), + ]); + }); + + it("does not lazy-start ready plugins outside the managed bundled allowlist", async () => { + const plugin = createPlugin("ready"); + mockRegistry.list.mockResolvedValue([plugin]); + const worker = createWorkerManager(); + const startWorker = vi.fn(async () => { + worker.markRunning(); + return true; + }); + + const drivers = await listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: ["paperclip.kubernetes-sandbox-provider"], + startWorker, + }, + }); + + expect(startWorker).not.toHaveBeenCalled(); + expect(drivers).toEqual([]); + }); + + it("leaves existing worker-manager recovery handles alone", async () => { + const plugin = createPlugin("ready"); + mockRegistry.list.mockResolvedValue([plugin]); + const worker = createWorkerManager({ hasHandle: true }); + const startWorker = vi.fn(async () => true); + + const drivers = await listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + + expect(startWorker).not.toHaveBeenCalled(); + expect(drivers).toEqual([]); + }); + + it("single-flights concurrent managed bundled recovery starts", async () => { + const plugin = createPlugin("ready"); + mockRegistry.list.mockResolvedValue([plugin]); + const worker = createWorkerManager(); + const loadStarted = createDeferred(); + const releaseLoad = createDeferred(); + const loadSingle = vi.fn(async () => { + loadStarted.resolve(); + await releaseLoad.promise; + worker.markRunning(); + return { success: true }; + }); + const startWorker = createManagedBundledPluginWorkerRecovery({ + managedBundledPluginKeys: [PLUGIN_KEY], + workerManager: worker.workerManager, + getLoader: () => ({ loadSingle }) as Pick, + }); + + const first = listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + await loadStarted.promise; + const second = listReadyPluginEnvironmentDrivers({ + db: {} as never, + workerManager: worker.workerManager, + recoverMissingWorker: { + pluginKeys: [PLUGIN_KEY], + startWorker, + }, + }); + + releaseLoad.resolve(); + const [firstDrivers, secondDrivers] = await Promise.all([first, second]); + + expect(loadSingle).toHaveBeenCalledTimes(1); + expect(loadSingle).toHaveBeenCalledWith(PLUGIN_ID); + expect(firstDrivers).toEqual([ + expect.objectContaining({ + pluginId: PLUGIN_ID, + driverKey: "daytona", + }), + ]); + expect(secondDrivers).toEqual(firstDrivers); + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index 5eeeb751c0b..2d7b76837fe 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -64,7 +64,7 @@ import { pluginUiStaticRoutes } from "./routes/plugin-ui-static.js"; import { readBrandedStaticIndexHtml } from "./static-index-html.js"; import { applyUiBranding } from "./ui-branding.js"; import { logger } from "./middleware/logger.js"; -import { DEFAULT_LOCAL_PLUGIN_DIR, pluginLoader } from "./services/plugin-loader.js"; +import { DEFAULT_LOCAL_PLUGIN_DIR, pluginLoader, type PluginLoader } from "./services/plugin-loader.js"; import { SELF_HOSTED_AUTO_INSTALL_KEYS, ensureBundledPlugins, @@ -148,6 +148,55 @@ export function shouldEnablePrivateHostnameGuard(opts: { ); } +export function createManagedBundledPluginWorkerRecovery(input: { + managedBundledPluginKeys: readonly string[]; + workerManager: Pick; + getLoader: () => Pick | null; +}): (plugin: { id: string; pluginKey: string }) => Promise { + const recoverablePluginKeys = new Set(input.managedBundledPluginKeys); + const inFlightStarts = new Map>(); + + return async (plugin) => { + if (!recoverablePluginKeys.has(plugin.pluginKey)) return false; + + const inFlight = inFlightStarts.get(plugin.id); + if (inFlight) return inFlight; + + const startPromise = (async () => { + if (input.workerManager.getWorker(plugin.id)) { + return input.workerManager.isRunning(plugin.id); + } + + const loader = input.getLoader(); + if (!loader) return false; + + try { + const result = await loader.loadSingle(plugin.id); + return result.success === true || input.workerManager.isRunning(plugin.id); + } catch (err) { + logger.warn( + { + pluginId: plugin.id, + pluginKey: plugin.pluginKey, + err: err instanceof Error ? err.message : String(err), + }, + "managed bundled plugin lazy worker recovery failed", + ); + throw err; + } + })(); + + inFlightStarts.set(plugin.id, startPromise); + try { + return await startPromise; + } finally { + if (inFlightStarts.get(plugin.id) === startPromise) { + inFlightStarts.delete(plugin.id); + } + } + }; +} + export async function createApp( db: Db, opts: { @@ -238,6 +287,34 @@ export async function createApp( const hostServicesDisposers = new Map void>(); const workerManager = opts.pluginWorkerManager ?? createPluginWorkerManager(); + const managedAutoInstallKeys = opts.managedPluginAutoInstall ?? null; + const bundledCatalogRoot = + opts.bundledPluginCatalogRoot ?? resolveBundledCatalogRoot(process.env); + const bundledPluginInstalls = resolveBundledPluginInstalls( + managedAutoInstallKeys ?? SELF_HOSTED_AUTO_INSTALL_KEYS, + { + catalogRoot: bundledCatalogRoot, + env: process.env, + enforceCatalogRoot: managedAutoInstallKeys !== null, + }, + ); + const managedBundledPluginKeys = + managedAutoInstallKeys !== null + ? bundledPluginInstalls.map((install) => install.pluginKey) + : []; + let runtimePluginLoader: Pick | null = null; + // A sibling process can install a managed bundled plugin while this process + // skips the mid-install row, then finish the row after this process's + // loadAll() pass. The capabilities route may recover only those managed + // bundles by starting their ready-but-unstarted worker lazily. + const recoverManagedBundledPluginWorker = + managedAutoInstallKeys !== null + ? createManagedBundledPluginWorkerRecovery({ + managedBundledPluginKeys, + workerManager, + getLoader: () => runtimePluginLoader, + }) + : undefined; // Mount API routes const api = Router(); @@ -271,7 +348,15 @@ export async function createApp( api.use(fileResourceRoutes(db)); api.use(routineRoutes(db, { pluginWorkerManager: workerManager })); api.use(pipelineRoutes(db)); - api.use(environmentRoutes(db, { pluginWorkerManager: workerManager })); + api.use(environmentRoutes(db, { + pluginWorkerManager: workerManager, + recoverMissingPluginWorker: recoverManagedBundledPluginWorker + ? { + pluginKeys: managedBundledPluginKeys, + startWorker: recoverManagedBundledPluginWorker, + } + : undefined, + })); api.use(executionWorkspaceRoutes(db, { pluginWorkerManager: workerManager })); api.use(goalRoutes(db)); api.use(boardChatRoutes(db, { deploymentMode: opts.deploymentMode })); @@ -380,6 +465,7 @@ export async function createApp( }, }, ); + runtimePluginLoader = loader; api.use( toolGatewayRoutes(db, toolGateway), ); @@ -570,21 +656,10 @@ export async function createApp( // drive the key list from the control plane; self-hosted instances keep // the pre-existing behavior of ensuring only the kubernetes bundle. // - // Resolution below is deliberately synchronous and NOT fail-safe: an + // Resolution is deliberately synchronous and NOT fail-safe: an // unknown key or a path escaping the bundled catalog root throws out of // createApp so a managed instance refuses to start (positive allowlist, // fail closed). - const managedAutoInstallKeys = opts.managedPluginAutoInstall ?? null; - const bundledCatalogRoot = - opts.bundledPluginCatalogRoot ?? resolveBundledCatalogRoot(process.env); - const bundledPluginInstalls = resolveBundledPluginInstalls( - managedAutoInstallKeys ?? SELF_HOSTED_AUTO_INSTALL_KEYS, - { - catalogRoot: bundledCatalogRoot, - env: process.env, - enforceCatalogRoot: managedAutoInstallKeys !== null, - }, - ); // SAFETY: installation is fully fail-safe. Any failure // (missing bundle, install error, load error) is caught, logged, and // swallowed per plugin so the server ALWAYS finishes booting. A degraded diff --git a/server/src/routes/environments.ts b/server/src/routes/environments.ts index e26f687ccc5..e83cc07c184 100644 --- a/server/src/routes/environments.ts +++ b/server/src/routes/environments.ts @@ -47,7 +47,10 @@ import { } from "../services/environment-config.js"; import { probeEnvironment } from "../services/environment-probe.js"; import { secretService } from "../services/secrets.js"; -import { listReadyPluginEnvironmentDrivers } from "../services/plugin-environment-driver.js"; +import { + listReadyPluginEnvironmentDrivers, + type ReadyPluginWorkerRecovery, +} from "../services/plugin-environment-driver.js"; import { getConfiguredSecretProvider } from "../secrets/configured-provider.js"; import { assertBoardOrgAccess, getActorInfo } from "./authz.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; @@ -261,7 +264,10 @@ function assertNoClientPlatformProvisionedMarkers(metadata: unknown): void { export function environmentRoutes( db: Db, - options: { pluginWorkerManager?: PluginWorkerManager } = {}, + options: { + pluginWorkerManager?: PluginWorkerManager; + recoverMissingPluginWorker?: ReadyPluginWorkerRecovery; + } = {}, ) { const router = Router(); const svc = environmentService(db); @@ -602,6 +608,7 @@ export function environmentRoutes( const pluginDrivers = await listReadyPluginEnvironmentDrivers({ db, workerManager: options.pluginWorkerManager, + recoverMissingWorker: options.recoverMissingPluginWorker, }); res.json(getEnvironmentCapabilities( AGENT_ADAPTER_TYPES, diff --git a/server/src/services/plugin-environment-driver.ts b/server/src/services/plugin-environment-driver.ts index 72d2425f19b..966998bca2d 100644 --- a/server/src/services/plugin-environment-driver.ts +++ b/server/src/services/plugin-environment-driver.ts @@ -30,6 +30,27 @@ import { import { pluginRegistryService } from "./plugin-registry.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +export interface ReadyPluginWorkerRecovery { + pluginKeys: readonly string[]; + startWorker(plugin: { id: string; pluginKey: string }): Promise; +} + +export interface ReadyPluginEnvironmentDriver { + pluginId: string; + pluginKey: string; + driverKey: string; + displayName: string; + description?: string; + configSchema: PluginEnvironmentDriverDeclaration["configSchema"]; + supportsReusableLeases?: PluginEnvironmentDriverDeclaration["supportsReusableLeases"]; + supportsInteractiveSetup?: PluginEnvironmentDriverDeclaration["supportsInteractiveSetup"]; + interactiveSetupConnectionTypes?: PluginEnvironmentDriverDeclaration["interactiveSetupConnectionTypes"]; + supportsTemplateCapture?: PluginEnvironmentDriverDeclaration["supportsTemplateCapture"]; + templateRefKind?: PluginEnvironmentDriverDeclaration["templateRefKind"]; + templateConfigBinding?: PluginEnvironmentDriverDeclaration["templateConfigBinding"]; + supportsTemplateDelete?: PluginEnvironmentDriverDeclaration["supportsTemplateDelete"]; +} + export function pluginDriverProviderKey(config: Pick): string { return `${config.pluginKey}:${config.driverKey}`; } @@ -94,30 +115,48 @@ export async function resolvePluginSandboxProviderDriverByKey(input: { export async function listReadyPluginEnvironmentDrivers(input: { db: Db; workerManager?: PluginWorkerManager; + recoverMissingWorker?: ReadyPluginWorkerRecovery; }) { if (!input.workerManager) return []; const pluginRegistry = pluginRegistryService(input.db); const plugins = await pluginRegistry.list(); - return plugins.flatMap((plugin) => { - if (plugin.status !== "ready" || !input.workerManager?.isRunning(plugin.id)) return []; - return (plugin.manifestJson.environmentDrivers ?? []) - .filter((driver) => driver.kind === "sandbox_provider") - .map((driver) => ({ - pluginId: plugin.id, - pluginKey: plugin.pluginKey, - driverKey: driver.driverKey, - displayName: driver.displayName, - description: driver.description, - configSchema: driver.configSchema, - supportsReusableLeases: driver.supportsReusableLeases, - supportsInteractiveSetup: driver.supportsInteractiveSetup, - interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes, - supportsTemplateCapture: driver.supportsTemplateCapture, - templateRefKind: driver.templateRefKind, - templateConfigBinding: driver.templateConfigBinding, - supportsTemplateDelete: driver.supportsTemplateDelete, - })); - }); + const recoverablePluginKeys = new Set(input.recoverMissingWorker?.pluginKeys ?? []); + const rows: ReadyPluginEnvironmentDriver[] = []; + for (const plugin of plugins) { + if (plugin.status !== "ready") continue; + 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; + } + rows.push( + ...(plugin.manifestJson.environmentDrivers ?? []) + .filter((driver) => driver.kind === "sandbox_provider") + .map((driver) => ({ + pluginId: plugin.id, + pluginKey: plugin.pluginKey, + driverKey: driver.driverKey, + displayName: driver.displayName, + description: driver.description, + configSchema: driver.configSchema, + supportsReusableLeases: driver.supportsReusableLeases, + supportsInteractiveSetup: driver.supportsInteractiveSetup, + interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes, + supportsTemplateCapture: driver.supportsTemplateCapture, + templateRefKind: driver.templateRefKind, + templateConfigBinding: driver.templateConfigBinding, + supportsTemplateDelete: driver.supportsTemplateDelete, + })), + ); + } + return rows; } export async function validatePluginSandboxProviderConfig(input: {