diff --git a/doc/plugins/PLUGIN_SPEC.md b/doc/plugins/PLUGIN_SPEC.md index fc96a2351e5..715a0476d5a 100644 --- a/doc/plugins/PLUGIN_SPEC.md +++ b/doc/plugins/PLUGIN_SPEC.md @@ -529,6 +529,20 @@ The host is responsible for: - secret resolution - UI route registration +An activation whose worker `initialize` call times out is transient. The host +removes the failed worker, keeps the plugin `ready`, and retries the activation +after 5, 15, 45 and 120 seconds on the shared work scheduler. While it retries, +the plugin's `lastError` starts with `Activation timed out` and names the next +retry; a successful retry clears it. Status writes are conditional on the +expected current status, so a disable or uninstall during a retry wins. Server +shutdown cancels pending retries and waits for running ones. Runs that need the plugin see a +worker that is not running, which setup retries within its bounded budget. After +the last retry, the plugin becomes `error` with that same prefix. A run that +needs the plugin then re-activates it on demand, at most once a minute. Any +other activation failure, such as a bad manifest, a worker that exits during +`initialize`, or a version mismatch, makes the plugin `error` at once. The next +server boot retries it once. + ## 12.3 Worker Responsibilities The plugin worker is responsible for: diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 3f73a1c6c98..509646dc29b 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -1178,6 +1178,26 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(offlineManager.call).not.toHaveBeenCalled(); }); + it("treats a sandbox plugin errored by activation timeouts as restarting while it re-activates", async () => { + const { companyId, environment, runId, pluginId } = await seedReusablePluginSandboxLease(); + const input = { companyId, environment, issueId: null, heartbeatRunId: runId, persistedExecutionWorkspace: null }; + await db.update(plugins).set({ status: "error", lastError: "Activation timed out after 4 retries: Worker initialize failed" }) + .where(eq(plugins.id, pluginId)); + const requestActivationTimeoutRecovery = vi.fn(() => true); + const manager = { isRunning: () => false, call: vi.fn(), requestActivationTimeoutRecovery } as unknown as PluginWorkerManager; + const refusal = await environmentRuntimeService(db, { pluginWorkerManager: manager, pluginWorkerReadyTimeoutMs: 0 }) + .acquireRunLease(input).catch(error => error); + expect(refusal).toBeInstanceOf(PluginWorkerUnavailableError); + expect(requestActivationTimeoutRecovery).toHaveBeenCalledWith(expect.objectContaining({ id: pluginId, status: "error" })); + + requestActivationTimeoutRecovery.mockReturnValue(false); + const terminal = await environmentRuntimeService(db, { pluginWorkerManager: manager, pluginWorkerReadyTimeoutMs: 0 }) + .acquireRunLease(input).catch(error => error); + expect(terminal).not.toBeInstanceOf(PluginWorkerUnavailableError); + expect((terminal as Error).message).toContain("but that plugin is currently error"); + expect(manager.call).not.toHaveBeenCalled(); + }); + it("retains a successful reusable sandbox lease without stopping the provider resource", async () => { const { pluginId, runId, reusableLease } = await seedReusablePluginSandboxLease(); const workerManager = { @@ -4008,6 +4028,21 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(call).toHaveBeenCalledOnce(); }); + it.each(["ready", "error"] as const)("refuses an exact resume as a retryable stopped worker while the %s plugin re-activates", async status => { + const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); + const lease = (await environmentService(db).getLeaseById(seeded.reusableLease.id))!; + await db.update(plugins).set({ status, lastError: status === "error" ? "Activation timed out after 4 retries: initialize" : null }) + .where(eq(plugins.id, seeded.pluginId)); + const call = vi.fn(); + const requestActivationTimeoutRecovery = vi.fn(() => true); + const workerManager = { isRunning: () => false, call, getWorker: () => undefined, requestActivationTimeoutRecovery } as unknown as PluginWorkerManager; + const error = await environmentRuntimeService(db, { pluginWorkerManager: workerManager }) + .resumeRunLease({ environment: seeded.environment, lease }).catch(caught => caught); + expect(error).toBeInstanceOf(PluginWorkerUnavailableError); + expect(call).not.toHaveBeenCalled(); + expect(requestActivationTimeoutRecovery).toHaveBeenCalledTimes(status === "error" ? 1 : 0); + }); + it.each([ { timeoutMs: 300_000, rpcTimeoutMs: 330_000 }, { timeoutMs: 1234, rpcTimeoutMs: 31_234 }, diff --git a/server/src/__tests__/helpers/bundled-image-fixture.ts b/server/src/__tests__/helpers/bundled-image-fixture.ts index dcb958a5235..44d2481bb48 100644 --- a/server/src/__tests__/helpers/bundled-image-fixture.ts +++ b/server/src/__tests__/helpers/bundled-image-fixture.ts @@ -7,7 +7,7 @@ import { distributionBundleDigest, readDistributionPluginCatalog } from "../../s import { derivePluginDatabaseNamespace } from "../../services/plugin-database.js"; import { createBundledPluginPackageResolver } from "../../services/bundled-plugin-package.js"; -export function bundledImageFixture(manifestOverrides: Partial = {}) { +export function bundledImageFixture(manifestOverrides: Partial = {}, options: { workerPrelude?: string } = {}) { const root = realpathSync(mkdtempSync(path.join(os.tmpdir(), "bundled-image-"))); const packageRoot = path.join(root, "distribution", "fixture"); const manifest: PaperclipPluginManifestV1 = { @@ -22,6 +22,7 @@ export function bundledImageFixture(manifestOverrides: Partial new Error(`Worker initialize failed for "${pluginId}": RPC call "initialize" timed out after 15000ms`, + { cause: new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.TIMEOUT, message: 'RPC call "initialize" timed out after 15000ms' }) }); + +describePostgres("plugin activation after a worker initialize timeout", () => { + let database: Awaited>; + let db: ReturnType; + const roots: string[] = []; + beforeAll(async () => { + if (externalDatabaseUrl) { + await applyPendingMigrations(externalDatabaseUrl); + database = { connectionString: externalDatabaseUrl, cleanup: () => closeRegisteredClients(externalDatabaseUrl) }; + } else { + database = await startEmbeddedPostgresTestDatabase("plugin-activation-timeout-"); + } + db = createDb(database.connectionString); + }, 90_000); + afterEach(async () => { + await db.delete(plugins); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); + }); + afterAll(async () => { await database?.cleanup(); }); + + async function setup(startOutcomes: Array<"timeout" | "crash" | "ok">, beforeRecoveredStart?: (pluginId: string) => Promise) { + const image = bundledImageFixture(); + roots.push(image.root); + const options = { + localPluginDir: path.join(image.root, "home"), prebuiltOnly: true, + resolvePackageRoot: image.resolver.resolve, assertPackageActivation: image.resolver.assertActivation, + enableLocalFilesystem: false, enableNpmDiscovery: false, + activationTimeoutRetryDelaysMs: [20, 20], activationTimeoutOnDemandIntervalMs: 0, + }; + await pluginLoader(db, options).installPlugin({ localPath: image.packageRoot }); + const registry = pluginRegistryService(db); + const row = (await registry.getByKey(image.manifest.id))!; + await registry.updateStatus(row.id, { status: "ready" }); + let running = false; + let handle: { status: "crashed" | "running" } | undefined; + const startWorker = vi.fn(async (pluginId: string) => { + if (handle) throw new Error(`Worker already registered for plugin ${pluginId}`); + const outcome = startOutcomes.shift() ?? "ok"; + if (outcome !== "ok") { + handle = { status: "crashed" }; + if (outcome === "timeout") throw initializeTimeout(pluginId); + throw new Error(`Worker initialize failed for "${pluginId}": worker exited with code 1`); + } + if (startWorker.mock.calls.length > 1) await beforeRecoveredStart?.(pluginId); + handle = { status: "running" }; + running = true; + }); + const markError = vi.fn(async (pluginId: string, error: string) => registry.updateStatus(pluginId, { status: "error", lastError: error })); + const services = { + workerManager: { + startWorker, call: vi.fn(async () => ({})), isRunning: vi.fn(() => running), + stopWorker: vi.fn(async () => { running = false; handle = undefined; }), stopAll: vi.fn(async () => { running = false; handle = undefined; }), + getWorker: vi.fn(() => handle), + }, + lifecycleManager: { markError }, + jobStore: { syncJobDeclarations: vi.fn() }, jobScheduler: { registerPlugin: vi.fn(), unregisterPlugin: vi.fn(), stop: vi.fn() }, + toolDispatcher: { unregisterPluginTools: vi.fn() }, + eventBus: { forPlugin: vi.fn(), subscriptionCount: vi.fn(() => 0), clearPlugin: vi.fn() }, + buildHostHandlers: vi.fn(() => ({})), instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, + } as unknown as PluginRuntimeServices; + const loader = pluginLoader(db, options, services); + return { loader, registry, row, startWorker, markError, isRunning: () => running }; + } + + it("retries an activation whose worker initialize timed out and ends ready with a running worker", async () => { + const { loader, registry, row, startWorker, markError, isRunning } = await setup(["timeout", "ok"]); + const result = await loader.loadAll(); + expect(result.failed).toBe(1); + expect((await registry.getById(row.id))!).toMatchObject({ status: "ready", + lastError: expect.stringContaining(PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX) }); + await vi.waitFor(() => expect(isRunning()).toBe(true), { timeout: 5_000 }); + expect(startWorker).toHaveBeenCalledTimes(2); + expect(markError).not.toHaveBeenCalled(); + await vi.waitFor(async () => expect((await registry.getById(row.id))!).toMatchObject({ status: "ready", lastError: null })); + await loader.shutdownAll(); + }); + + it("does not bring back a plugin an operator disabled while its activation was timing out", async () => { + const { loader, registry, row, startWorker } = await setup(["timeout", "ok"]); + await loader.loadAll(); + await registry.updateStatus(row.id, { status: "disabled" }); + await new Promise(resolve => setTimeout(resolve, 200)); + expect(startWorker).toHaveBeenCalledOnce(); + expect((await registry.getById(row.id))!.status).toBe("disabled"); + expect(loader.requestActivationTimeoutRecovery({ id: row.id, status: "error", lastError: `${PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX} after 2 retries` })).toBe(true); + await loader.stopActivationTimeoutRecovery(); + expect(startWorker).toHaveBeenCalledOnce(); + expect((await registry.getById(row.id))!.status).toBe("disabled"); + }); + + it("stops a recovered worker when an operator disabled the plugin during its activation", async () => { + let disable: ((pluginId: string) => Promise) | undefined; + const { loader, registry, row, startWorker, isRunning } = await setup(["timeout", "ok"], async pluginId => { await disable?.(pluginId); }); + disable = async pluginId => { await registry.updateStatus(pluginId, { status: "disabled" }); }; + await loader.loadAll(); + await vi.waitFor(() => expect(startWorker).toHaveBeenCalledTimes(2), { timeout: 5_000 }); + await vi.waitFor(() => expect(isRunning()).toBe(false), { timeout: 5_000 }); + await loader.stopActivationTimeoutRecovery(); + expect(isRunning()).toBe(false); + expect((await registry.getById(row.id))!.status).toBe("disabled"); + }); + + it("cancels a pending retry on shutdown", async () => { + const { loader, startWorker } = await setup(["timeout", "ok"]); + await loader.loadAll(); + await loader.stopActivationTimeoutRecovery(); + await new Promise(resolve => setTimeout(resolve, 200)); + expect(startWorker).toHaveBeenCalledOnce(); + }); + + it("marks a plugin error once its activation keeps timing out past the bound", async () => { + const { loader, registry, row, startWorker, markError, isRunning } = await setup(["timeout", "timeout", "timeout"]); + await loader.loadAll(); + await vi.waitFor(() => expect(markError).toHaveBeenCalledOnce(), { timeout: 5_000 }); + expect(startWorker).toHaveBeenCalledTimes(3); + expect(markError).toHaveBeenCalledWith(row.id, expect.stringMatching(new RegExp(`^${PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX} after 2 retries`))); + expect(isRunning()).toBe(false); + const errored = (await registry.getById(row.id))!; + expect(errored.status).toBe("error"); + + expect(loader.requestActivationTimeoutRecovery(errored)).toBe(true); + await vi.waitFor(() => expect(isRunning()).toBe(true), { timeout: 5_000 }); + expect(startWorker).toHaveBeenCalledTimes(4); + expect((await registry.getById(row.id))!.status).toBe("ready"); + await loader.shutdownAll(); + }); + + it("recovers a real worker whose first initialize stalled past the host timeout", async () => { + const markerRoot = mkdtempSync(path.join(os.tmpdir(), "plugin-stall-marker-")); + roots.push(markerRoot); + const marker = path.join(markerRoot, "stalled"); + const image = bundledImageFixture({}, { workerPrelude: `import { existsSync as stalledBefore, writeFileSync as markStalled } from "node:fs"; +if (!process.argv.includes("--source") && !stalledBefore(${JSON.stringify(marker)})) { markStalled(${JSON.stringify(marker)}, "stalled"); setInterval(() => {}, 60_000); await new Promise(() => {}); }` }); + roots.push(image.root); + const options = { + localPluginDir: path.join(image.root, "home"), prebuiltOnly: true, + resolvePackageRoot: image.resolver.resolve, assertPackageActivation: image.resolver.assertActivation, + enableLocalFilesystem: false, enableNpmDiscovery: false, activationTimeoutRetryDelaysMs: [50], + }; + await pluginLoader(db, options).installPlugin({ localPath: image.packageRoot }); + const registry = pluginRegistryService(db); + const row = (await registry.getByKey(image.manifest.id))!; + await registry.updateStatus(row.id, { status: "ready" }); + const manager = createPluginWorkerManager(); + const markError = vi.fn(async (pluginId: string, error: string) => registry.updateStatus(pluginId, { status: "error", lastError: error })); + const services = { + workerManager: manager, lifecycleManager: { markError }, + jobStore: { syncJobDeclarations: vi.fn() }, jobScheduler: { registerPlugin: vi.fn(), unregisterPlugin: vi.fn(), stop: vi.fn() }, + toolDispatcher: { unregisterPluginTools: vi.fn() }, + eventBus: { forPlugin: vi.fn(), subscriptionCount: vi.fn(() => 0), clearPlugin: vi.fn() }, + buildHostHandlers: vi.fn(() => ({})), instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, + } as unknown as PluginRuntimeServices; + const loader = pluginLoader(db, options, services); + try { + const result = await loader.loadAll(); + expect(result.failed).toBe(1); + expect(existsSync(marker)).toBe(true); + await vi.waitFor(() => expect(manager.isRunning(row.id)).toBe(true), { timeout: 10_000 }); + expect(await manager.call(row.id, "health", {})).toMatchObject({ status: "ok" }); + expect(markError).not.toHaveBeenCalled(); + await vi.waitFor(async () => expect((await registry.getById(row.id))!).toMatchObject({ status: "ready", lastError: null })); + } finally { + await loader.shutdownAll(); + } + }, 60_000); + + it("keeps a deterministic activation failure a permanent error without retrying it", async () => { + const { loader, registry, row, startWorker, markError } = await setup(["crash", "ok"]); + await loader.loadAll(); + expect(markError).toHaveBeenCalledWith(row.id, expect.stringMatching(/^Activation failed: /)); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(startWorker).toHaveBeenCalledOnce(); + const errored = (await registry.getById(row.id))!; + expect(errored.status).toBe("error"); + expect(loader.requestActivationTimeoutRecovery(errored)).toBe(false); + await loader.shutdownAll(); + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index 1ef01538921..1d3192e90ec 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1010,6 +1010,7 @@ export async function createApp( }, ); runtimePluginLoader = loader; + workerManager.setActivationTimeoutRecovery?.(plugin => loader.requestActivationTimeoutRecovery(plugin)); const browserUse = browserUseService(db, undefined, { cancelWorkForScope: heartbeatService(db, { pluginWorkerManager: workerManager }).cancelBudgetScopeWork }, (session, run) => toolGateway.browserUseSessionAuthorized({ ...session, runId: run.heartbeatRunId, invocationId: run.invocationId })); @@ -1400,6 +1401,7 @@ export async function createApp( const shutdownAppServices = (): Promise => { if (appServicesShutdown) return appServicesShutdown; appServicesShutdown = finishServiceShutdown(() => scheduler.stop(), async () => { + await loader.stopActivationTimeoutRecovery(); await agentLifecycle.stop(); unregisterMcpOwners(); await publicMcpEvents?.stop(); diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index b2fb67bb003..a567bcffcc8 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1506,7 +1506,10 @@ function createSandboxEnvironmentDriver( if (input.pluginId) { const pinned = await resolvePluginSandboxProviderDriverById({ db, pluginId: input.pluginId, driverKey: input.provider }); if (!pinned) return { state: "missing" as const, resolved: null }; - if (pinned.plugin.status !== "ready") return { state: "not_ready" as const, resolved: pinned }; + if (pinned.plugin.status !== "ready") { + return pluginWorkerManager?.requestActivationTimeoutRecovery?.(pinned.plugin) + ? { state: "worker_unavailable" as const, resolved: pinned } : { state: "not_ready" as const, resolved: pinned }; + } if (!pluginWorkerManager?.isRunning(pinned.plugin.id)) return { state: "worker_unavailable" as const, resolved: pinned }; return { state: "running" as const, resolved: pinned }; } @@ -1531,7 +1534,8 @@ function createSandboxEnvironmentDriver( } if (installed.plugin.status !== "ready") { - return { state: "not_ready" as const, resolved: installed }; + return pluginWorkerManager?.requestActivationTimeoutRecovery?.(installed.plugin) + ? { state: "worker_unavailable" as const, resolved: installed } : { state: "not_ready" as const, resolved: installed }; } if (!pluginWorkerManager) { @@ -2694,6 +2698,14 @@ function createSandboxEnvironmentDriver( async resumeRunLease(input) { const pluginId = readString(input.lease.metadata?.pluginId); const providerKey = readString(input.lease.metadata?.provider); + if (input.lease.metadata?.sandboxProviderPlugin && pluginWorkerManager && pluginId && providerKey + && !pluginWorkerManager.isRunning(pluginId)) { + const plugin = await resolvePluginSandboxProviderDriverById({ db, pluginId, driverKey: providerKey }); + if (plugin && (plugin.plugin.status === "ready" || pluginWorkerManager.requestActivationTimeoutRecovery?.(plugin.plugin))) { + throw new PluginWorkerUnavailableError({ pluginId, workerStatus: pluginWorkerManager.getWorker(pluginId)?.status ?? null, + message: `Sandbox provider "${providerKey}" is installed via plugin "${plugin.plugin.pluginKey}", but its worker is not running.` }); + } + } if (!input.lease.metadata?.sandboxProviderPlugin || !pluginWorkerManager || !pluginId || !providerKey || !input.lease.providerLeaseId || input.lease.environmentId !== input.environment.id || !pluginWorkerVerifiesLifecycleMethod(pluginId, "environmentResumeLease") diff --git a/server/src/services/plugin-loader.ts b/server/src/services/plugin-loader.ts index 909bece5edd..82294301469 100644 --- a/server/src/services/plugin-loader.ts +++ b/server/src/services/plugin-loader.ts @@ -54,6 +54,7 @@ import type { PluginLifecycleManager } from "./plugin-lifecycle.js"; import { pluginDatabaseService } from "./plugin-database.js"; import { resolveBundledCatalogRoot } from "./bundled-plugins.js"; import { managedPluginWorkerEnvGrant } from "./managed-plugin-worker-env.js"; +import { createWorkScheduler } from "./work-scheduler.js"; const execFileAsync = promisify(execFile); const __dirname = path.dirname(fileURLToPath(import.meta.url)); @@ -318,6 +319,29 @@ export interface PluginLoaderOptions { * Registry support is not yet implemented; this field is reserved. */ registryUrl?: string; + + /** Delays between retries of an activation whose worker `initialize` timed out. */ + activationTimeoutRetryDelaysMs?: readonly number[]; + + /** Shortest interval between on-demand re-activations of a plugin errored by activation timeouts. */ + activationTimeoutOnDemandIntervalMs?: number; +} + +export const PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX = "Activation timed out"; +const DEFAULT_ACTIVATION_TIMEOUT_RETRY_DELAYS_MS = [5_000, 15_000, 45_000, 120_000] as const; +const DEFAULT_ACTIVATION_TIMEOUT_ON_DEMAND_INTERVAL_MS = 60_000; + +export function isPluginActivationTimeout(error: unknown): boolean { + let current: unknown = error; + for (let depth = 0; depth < 6 && current instanceof Error; depth++) { + if ((current as { code?: unknown }).code === PLUGIN_RPC_ERROR_CODES.TIMEOUT) return true; + current = current.cause; + } + return false; +} + +export function isPluginActivationTimeoutError(plugin: { status: string; lastError: string | null }): boolean { + return plugin.status === "error" && plugin.lastError?.startsWith(PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX) === true; } // --------------------------------------------------------------------------- @@ -445,6 +469,7 @@ export interface PluginLoadSingleOptions { interface PluginActivateOptions { markErrorOnFailure: boolean; + activationTimeoutRetry?: number; } /** @@ -646,6 +671,16 @@ export interface PluginLoader { */ unloadSingle(pluginId: string, pluginKey: string): Promise; + /** + * Re-activate a plugin that is in `error` only because its worker's + * `initialize` timed out, at most once per interval. Returns whether an + * activation is now in flight; the caller treats the plugin as starting. + */ + requestActivationTimeoutRecovery(plugin: { id: string; status: string; lastError: string | null }): boolean; + + /** Cancel pending activation-timeout retries and await the ones already running. */ + stopActivationTimeoutRecovery(): Promise; + /** * Stop all managed plugin workers. Called during server shutdown. * @@ -1168,9 +1203,45 @@ export function pluginLoader( assertPackageActivation, resolvePackageRoot, prebuiltOnly = false, + activationTimeoutRetryDelaysMs = DEFAULT_ACTIVATION_TIMEOUT_RETRY_DELAYS_MS, + activationTimeoutOnDemandIntervalMs = DEFAULT_ACTIVATION_TIMEOUT_ON_DEMAND_INTERVAL_MS, } = options; const registry = pluginRegistryService(db); + const activationRetryScheduler = createWorkScheduler(); + const activeActivationRecoveries = new Set>(); + const onDemandActivations = new Map(); + let shuttingDown = false; + + function trackActivationRecovery(work: () => Promise, label: string, pluginId: string) { + const running = work().catch((err: unknown) => { + log.warn({ pluginId, err: err instanceof Error ? err.message : String(err) }, label); + }); + activeActivationRecoveries.add(running); + void running.finally(() => activeActivationRecoveries.delete(running)); + } + + async function stopActivationTimeoutRecovery() { + shuttingDown = true; + activationRetryScheduler.stop(); + while (activeActivationRecoveries.size > 0) await Promise.allSettled([...activeActivationRecoveries]); + } + + async function activateForRecovery(plugin: PluginRecord, activationTimeoutRetry: number) { + const result = await activatePlugin(plugin, { markErrorOnFailure: true, activationTimeoutRetry }); + if (!result.success) return; + const latest = (await registry.getById(plugin.id)) as PluginRecord | null; + if (latest?.status === "ready" && !shuttingDown) return; + log.info({ pluginId: plugin.id, pluginKey: plugin.pluginKey, status: latest?.status ?? null }, + "plugin-loader: stopping a recovered plugin that an operator changed during its activation"); + await teardownPluginRuntime(plugin.id, plugin.pluginKey); + } + + async function discardFailedWorkerHandle(pluginId: string) { + const handle = runtimeServices?.workerManager.getWorker(pluginId); + if (!handle || (handle.status !== "crashed" && handle.status !== "stopped")) return; + await runtimeServices!.workerManager.stopWorker(pluginId); + } const manifestValidator = pluginManifestValidator(); const capabilityValidator = pluginCapabilityValidator(); const log = logger.child({ service: "plugin-loader" }); @@ -2143,7 +2214,31 @@ export function pluginLoader( // shutdownAll // ----------------------------------------------------------------------- + requestActivationTimeoutRecovery(plugin) { + if (!runtimeServices || shuttingDown || !isPluginActivationTimeoutError(plugin)) return false; + const previous = onDemandActivations.get(plugin.id); + if (previous?.inFlight) return true; + if (previous && Date.now() - previous.at < activationTimeoutOnDemandIntervalMs) return false; + const attempt = { at: Date.now(), inFlight: true }; + onDemandActivations.set(plugin.id, attempt); + trackActivationRecovery(async () => { + try { + const flipped = (await registry.updateStatusIf(plugin.id, + { status: "error", lastErrorPrefix: PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX }, { status: "ready" })) as PluginRecord | null; + if (!flipped || shuttingDown) return; + log.info({ pluginId: plugin.id, pluginKey: flipped.pluginKey }, + "plugin-loader: re-activating a plugin errored by activation timeouts on demand"); + await discardFailedWorkerHandle(plugin.id); + await activateForRecovery(flipped, activationTimeoutRetryDelaysMs.length); + } finally { attempt.inFlight = false; } + }, "plugin-loader: on-demand re-activation failed", plugin.id); + return true; + }, + + stopActivationTimeoutRecovery, + async shutdownAll(): Promise { + await stopActivationTimeoutRecovery(); if (!runtimeServices) { throw new Error( "Cannot shutdownAll: no PluginRuntimeServices provided.", @@ -2506,6 +2601,12 @@ export function pluginLoader( "plugin-loader: plugin activated successfully", ); + if (activePlugin.lastError?.startsWith(PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX)) { + activePlugin = ((await registry.updateStatusIf(pluginId, + { status: "ready", lastErrorPrefix: PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX }, { status: "ready", lastError: null })) as PluginRecord | null) + ?? activePlugin; + } + return { plugin: activePlugin, success: true, registered }; } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); @@ -2515,13 +2616,39 @@ export function pluginLoader( "plugin-loader: failed to activate plugin", ); - if (options.markErrorOnFailure) { + const timedOut = isPluginActivationTimeout(err); + const timeoutRetry = options.activationTimeoutRetry ?? 0; + if (options.markErrorOnFailure && timedOut && timeoutRetry < activationTimeoutRetryDelaysMs.length && !shuttingDown) { + const delayMs = activationTimeoutRetryDelaysMs[timeoutRetry]!; + try { + if (registered.worker) await teardownPluginRuntime(pluginId, pluginKey); + await discardFailedWorkerHandle(pluginId); + await registry.updateStatusIf(pluginId, { status: "ready" }, { status: "ready", lastError: + `${PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX}; retrying in ${Math.round(delayMs / 1000)}s ` + + `(retry ${timeoutRetry + 1} of ${activationTimeoutRetryDelaysMs.length}): ${errorMessage}` }); + } catch (retryErr) { + log.warn({ pluginId, pluginKey, err: retryErr instanceof Error ? retryErr.message : String(retryErr) }, + "plugin-loader: could not record a timed-out activation before retrying it"); + } + log.warn({ pluginId, pluginKey, delayMs, retry: timeoutRetry + 1 }, + "plugin-loader: plugin activation timed out; retrying"); + activationRetryScheduler.schedule(`plugin-activation:${pluginId}`, Date.now() + delayMs, () => { + trackActivationRecovery(async () => { + const current = (await registry.getById(pluginId)) as PluginRecord | null; + if (shuttingDown || !current || current.status !== "ready" || workerManager.isRunning(pluginId)) return; + await discardFailedWorkerHandle(pluginId); + await activateForRecovery(current, timeoutRetry + 1); + }, "plugin-loader: timed-out activation retry failed", pluginId); + }); + } else if (options.markErrorOnFailure) { // Mark the plugin as errored in the database. The running process // leaves it inactive; the next boot's loadAll retries it once, and the // lifecycle enable() path can revive it sooner by hand. // markError also deactivates the plugin runtime in this process. try { - await lifecycleManager.markError(pluginId, `Activation failed: ${errorMessage}`); + await lifecycleManager.markError(pluginId, timedOut + ? `${PLUGIN_ACTIVATION_TIMEOUT_ERROR_PREFIX} after ${timeoutRetry} retries: ${errorMessage}` + : `Activation failed: ${errorMessage}`); } catch (markErr) { log.error( { diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index 5f5c84470d3..ab9eabc658d 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -279,6 +279,26 @@ export function pluginRegistryService(db: Db) { .then((rows) => rows[0] ?? null)); }, + /** + * Update status and error only while the row still has the expected status + * (and, when given, an error with the expected prefix). Returns null when a + * concurrent change such as a disable or an uninstall won. + */ + updateStatusIf: async (id: string, expected: { status: string; lastErrorPrefix?: string }, input: UpdatePluginStatus) => { + const plugin = await getById(id); + if (!plugin || plugin.status !== expected.status) return null; + const workspaceProviderActivated = Boolean(plugin.manifestJson.workspaceProvider) + && plugin.status !== "ready" && input.status === "ready"; + return writeWorkspaceProviderChange(workspaceProviderActivated, database => database + .update(plugins) + .set({ status: input.status, lastError: input.lastError ?? null, updatedAt: new Date() }) + .where(and(eq(plugins.id, id), eq(plugins.status, expected.status as typeof plugin.status), + expected.lastErrorPrefix === undefined ? undefined + : sql`starts_with(${plugins.lastError}, ${expected.lastErrorPrefix})`)) + .returning() + .then((rows) => rows[0] ?? null)); + }, + // ----- Uninstall / Remove -------------------------------------------- /** diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index ceb257fec44..d28b800d5be 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -829,6 +829,10 @@ export interface WorkerDiagnostics { * for starting/stopping all workers and routing RPC calls. */ export interface PluginWorkerManager { + /** Installs the loader's on-demand re-activation for plugins errored by activation timeouts. */ + setActivationTimeoutRecovery?(handler: ((plugin: { id: string; status: string; lastError: string | null }) => boolean) | null): void; + /** Whether a re-activation of a plugin errored by activation timeouts is now in flight. */ + requestActivationTimeoutRecovery?(plugin: { id: string; status: string; lastError: string | null }): boolean; drainStartup?(contextId: string): Promise; recoverStartupAcquisition?(contextId: string, creationOperationId: string): void; retireStartup?(contextId: string): Promise; @@ -3415,7 +3419,7 @@ export function createPluginWorkerHandle( log.error({ err: msg }, "worker initialize failed"); await killProcess(); setStatus("crashed"); - throw new Error(`Worker initialize failed for "${pluginId}": ${msg}`); + throw new Error(`Worker initialize failed for "${pluginId}": ${msg}`, { cause: err }); } // Reset crash counter on successful start @@ -4089,7 +4093,14 @@ export function createPluginWorkerManager( const idleWorkerManifests = new WeakMap(); const idleManagerGeneration = randomUUID(); let nextIdleWorkerIdentity = 0; + let activationTimeoutRecovery: ((plugin: { id: string; status: string; lastError: string | null }) => boolean) | null = null; return { + setActivationTimeoutRecovery(handler) { + activationTimeoutRecovery = handler; + }, + requestActivationTimeoutRecovery(plugin) { + return activationTimeoutRecovery?.(plugin) ?? false; + }, onConfigApplied(listener) { configApplied.add(listener); return () => { configApplied.delete(listener); };