mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(environments): close Apex findings on managed plugin environments (#19526)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A managed instance gets its instance-level environments from the `PAPERCLIP_MANAGED_CONFIG` document, and #19508 added plugin environment entries to it > - Greptile's Apex review of #19508 arrived after it merged, with three findings > - A credential inside an array nested in an array passed the secret check and the response redaction > - Recovery of an archived environment listened on one worker handle, which a restart replaces, and its attempts were not tracked through idle checks and shutdown > - This pull request closes all three findings > - The benefit is that credentials never pass the check, an environment comes back after any worker restart, and shutdown never cuts off a recovery write ## Linked Issues or Issue Description Refs #19508 (the Apex findings this pull request fixes) **Subsystem affected** Server environments: the managed-config parser, the environment routes' response floor, the managed environment bootstrap, and the plugin worker manager. **Problem or motivation** Greptile's Apex review of #19508 found three problems. (1) `findSecretLikeConfigKey` and `redactSecretLikeConfigKeys` skip array elements that are arrays, so `{"groups":[[{"apiToken":"..."}]]}` passes the boot check and is echoed to restricted readers. (2) Recovery subscribes to `ready` on one worker handle. A restart such as a settings save replaces that handle, so the new worker's `ready` never reaches the listener, and the environment stays archived. (3) A recovery attempt is fire-and-forget. The worker manager's work token ends when the worker replies, but the database write continues, so an idle check or the database shutdown can overlap it. **Proposed solution** Walk arrays at any depth in both functions. Add a manager-wide `onWorkerReady` subscription and use it for recovery. Run every recovery attempt through a controller that tracks it as idle work and that shutdown stops and awaits before the database closes. **Alternatives considered** Re-subscribing on each new handle after a restart. That needs a hook in every restart path. The manager-wide subscription covers all of them in one place. **Roadmap alignment** This only fixes code that #19508 added or touched. It adds no product surface. ## What Changed - `server/src/services/managed-config.ts`: `findSecretLikeConfigKey` walks arrays at any depth (new `findSecretLikeConfigKeyInValue`). This covers sandbox `config` and plugin `driverConfig`. - `server/src/routes/environments.ts`: `redactSecretLikeConfigKeys` walks arrays at any depth (new `redactSecretLikeConfigValue`). - `server/src/services/plugin-worker-manager.ts`: new optional `onWorkerReady(listener)` on `PluginWorkerManager`. It fires on every worker `ready`, including a replacement handle from a restart. A listener error is logged and does not affect the worker. - `server/src/services/managed-environments.ts`: - Sandbox and plugin recovery subscribe through `subscribeWorkerReady`. It prefers `onWorkerReady` and falls back to the handle when the manager has no such method. - New `createManagedEnvironmentRecovery()` controller. Each attempt runs inside `trackIdleWork` for its whole duration. `stop()` removes every `ready` subscription, refuses new attempts, and awaits unfinished ones. - `server/src/index.ts`: creates the controller, passes it to `applyManagedEnvironments`, and awaits `stop()` during shutdown, after the delivery workers stop and before the application teardown and database close. ## Verification - `cd server && npx vitest run src/__tests__/managed-config.test.ts`: a secret-like key in an array nested in an array refuses boot, for plugin `driverConfig` and sandbox `config`. - `npx vitest run src/__tests__/environment-routes.test.ts`: a nested-array credential is removed from the response, and other nested values stay. - `npx vitest run src/services/managed-environments-plugin-entries.test.ts`: - With `onWorkerReady`, recovery never consults the handle. It ignores another plugin's `ready`, re-ensures on this plugin's `ready`, and then unsubscribes. - An attempt counts in `idleWorkSnapshot()` until its ensure settles. `stop()` unsubscribes, waits for the unfinished attempt, and later `ready` events start nothing. - `npx vitest run src/services/managed-environments.test.ts src/__tests__/managed-plugin-environments.test.ts`: pass. - `pnpm --filter @paperclipai/server typecheck` passes. - I did not run the full `pnpm test:run`, `pnpm -r typecheck` or `pnpm build`. CI covers them. ## Risks - The redaction now also walks nested arrays in sandbox config responses. Only credential-shaped keys are removed; other values are unchanged. - The boot check now refuses a document that hides a secret-like key in a nested array. A valid document has no such key. - `onWorkerReady` is optional, so other `PluginWorkerManager` implementations and test fakes keep working. - Shutdown now waits for an unfinished recovery attempt. An attempt is one worker call plus one short transaction. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use (file edits, shell, test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
3b3037aab5
commit
1206373a95
10 files changed
+266
-54
No files matched your search
@@ -577,6 +577,23 @@ describe("environment routes", () => {
|
||||
expect(JSON.stringify(res.body)).not.toContain("config-credential-must-never-echo");
|
||||
});
|
||||
|
||||
it("redacts credential-shaped keys inside arrays nested in arrays", async () => {
|
||||
mockEnvironmentService.getById.mockResolvedValue({
|
||||
...createPlatformSandboxEnvironment(),
|
||||
config: {
|
||||
provider: "daytona",
|
||||
groups: [[{ name: "a", apiToken: "nested-credential-must-never-echo" }], ["plain"]],
|
||||
},
|
||||
});
|
||||
const app = createApp(ownerAdminActor);
|
||||
|
||||
const res = await request(app).get("/api/environments/env-managed-1");
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.config).toEqual({ provider: "daytona", groups: [[{ name: "a" }], ["plain"]] });
|
||||
expect(JSON.stringify(res.body)).not.toContain("nested-credential-must-never-echo");
|
||||
});
|
||||
|
||||
it("keeps blanking env vars on legacy kubernetes-marker rows", async () => {
|
||||
// Pre-generalization builds may have written platform values into
|
||||
// these rows' env vars, so the legacy floor stays absolute there.
|
||||
|
||||
@@ -411,6 +411,11 @@ describe("parseManagedConfigEnv environments section", () => {
|
||||
),
|
||||
),
|
||||
).toThrow(/adapters\[0\].clientSecret/);
|
||||
expect(() =>
|
||||
parseManagedConfigEnv(
|
||||
envWith(validDoc({ environments: [entry({ config: { pools: [[], [{ privateKey: { password: "s" } }]] } })] })),
|
||||
),
|
||||
).toThrow(/pools\[1\]\[0\].privateKey.password/);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -482,6 +487,8 @@ describe("parseManagedConfigEnv plugin environment entries", () => {
|
||||
.toThrow(/"environments\[0\].driverConfig" key "apiToken" looks secret-bearing/);
|
||||
expect(() => parseManagedConfigEnv(docWith([remoteHost({ driverConfig: { auth: [{ password: "p" }] } })])))
|
||||
.toThrow(/auth\[0\].password/);
|
||||
expect(() => parseManagedConfigEnv(docWith([remoteHost({ driverConfig: { groups: [[{ apiToken: "t" }]] } })])))
|
||||
.toThrow(/groups\[0\]\[0\].apiToken/);
|
||||
});
|
||||
|
||||
it("refuses unknown keys, including sandbox keys on a plugin entry and the reverse", () => {
|
||||
|
||||
@@ -37,6 +37,10 @@ vi.mock("../services/adapter-settings.js", async () => ({
|
||||
initializeAdapterSettings: vi.fn(async () => null),
|
||||
}));
|
||||
vi.mock("../services/cloud-runtime-config.js", () => ({ initializeCloudRuntimeConfig: cloudRuntimeStartupMocks.initialize }));
|
||||
const managedEnvironmentMocks = vi.hoisted(() => {
|
||||
const recovery = { run: vi.fn(), onStop: vi.fn(), stop: vi.fn(async () => {}) };
|
||||
return { recovery, apply: vi.fn(async () => null), factory: vi.fn(() => recovery) };
|
||||
});
|
||||
const cloudMigrationMocks = vi.hoisted(() => ({ migrate: vi.fn(async () => null), qualify: vi.fn(async () => null) }));
|
||||
vi.mock("../services/cloud-configuration-migration.js", () => ({ migrateCloudConfiguration: cloudMigrationMocks.migrate }));
|
||||
vi.mock("../services/cloud-configuration-qualification.js", () => ({ qualifyCloudConfigurationForCurrentImage: cloudMigrationMocks.qualify }));
|
||||
@@ -482,7 +486,8 @@ vi.mock("../services/index.js", () => ({
|
||||
})),
|
||||
feedbackService: feedbackServiceFactoryMock,
|
||||
bootstrapExecutionPolicyFromEnv: vi.fn(async () => null),
|
||||
applyManagedEnvironments: vi.fn(async () => null),
|
||||
applyManagedEnvironments: managedEnvironmentMocks.apply,
|
||||
createManagedEnvironmentRecovery: managedEnvironmentMocks.factory,
|
||||
environmentCustomImageService: environmentCustomImagesServiceFactoryMock,
|
||||
executionWorkspaceService: executionWorkspaceServiceFactoryMock,
|
||||
externalObjectService: externalObjectsServiceFactoryMock,
|
||||
@@ -1369,6 +1374,14 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(createAppMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("stops managed environment recovery when a later startup step fails", async () => {
|
||||
managedEnvironmentMocks.recovery.stop.mockClear();
|
||||
managedEnvironmentMocks.apply.mockRejectedValueOnce(new Error("managed environments refused"));
|
||||
await expect(startServer()).rejects.toThrow("managed environments refused");
|
||||
expect(managedEnvironmentMocks.factory).toHaveBeenCalled();
|
||||
expect(managedEnvironmentMocks.recovery.stop).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("passes the feedback export service into createApp so pending traces flush in runtime", async () => {
|
||||
const started = await startServer();
|
||||
|
||||
|
||||
@@ -137,6 +137,7 @@ import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "./mi
|
||||
import {
|
||||
feedbackService,
|
||||
applyManagedEnvironments,
|
||||
createManagedEnvironmentRecovery,
|
||||
backfillPrincipalAccessCompatibility,
|
||||
backfillLegacyToolOAuthTokens,
|
||||
bootstrapExecutionPolicyFromEnv,
|
||||
@@ -1175,6 +1176,17 @@ async function startServerWithDatabaseTeardown(
|
||||
throw err;
|
||||
}
|
||||
|
||||
// Post-boot recovery of managed environments (re-ensuring a row when its
|
||||
// plugin worker becomes ready). Shutdown stops it before the database closes,
|
||||
// and so does the startup-failure teardown: a later startup step that throws
|
||||
// must not leave an attempt writing through a closed database.
|
||||
const managedEnvironmentRecovery = createManagedEnvironmentRecovery();
|
||||
const closeDatabaseAfterStartupFailure = startupDatabase.close;
|
||||
startupDatabase.close = async () => {
|
||||
await managedEnvironmentRecovery.stop();
|
||||
await closeDatabaseAfterStartupFailure?.();
|
||||
};
|
||||
|
||||
// Ensure the sandbox and plugin environments declared in the managed-config
|
||||
// document (`environments` section) before the heartbeat resumes queued runs. The
|
||||
// document already parsed fail-closed above; the ensure step itself is
|
||||
@@ -1190,6 +1202,7 @@ async function startServerWithDatabaseTeardown(
|
||||
locals?: { bundledPluginsStartup?: Promise<unknown>; managedBundledPluginKeys?: readonly string[] };
|
||||
}).locals;
|
||||
const managedEnvironmentsResult = await applyManagedEnvironments(db as any, managedConfig, {
|
||||
recovery: managedEnvironmentRecovery,
|
||||
pluginsReady: appLocals?.bundledPluginsStartup,
|
||||
autoInstalledPluginKeys: appLocals?.managedBundledPluginKeys,
|
||||
workerManager: pluginWorkerManager,
|
||||
@@ -2043,6 +2056,7 @@ async function startServerWithDatabaseTeardown(
|
||||
await runtimeResponseDelivery?.stop();
|
||||
await deploymentHints?.stop();
|
||||
await deliveryWork.stop();
|
||||
await managedEnvironmentRecovery.stop();
|
||||
|
||||
// Whatever the drain did not finalize (timed-out runs, the hot-restart
|
||||
// skip path) still has a local-only tail when the in-flight run-log
|
||||
|
||||
@@ -93,19 +93,18 @@ function redactSecretLikeConfigKeys(value: Record<string, unknown>): Record<stri
|
||||
const result: Record<string, unknown> = {};
|
||||
for (const [key, child] of Object.entries(value)) {
|
||||
if (SECRET_LIKE_CONFIG_KEY_PATTERN.test(key)) continue;
|
||||
if (isPlainRecord(child)) {
|
||||
result[key] = redactSecretLikeConfigKeys(child);
|
||||
} else if (Array.isArray(child)) {
|
||||
result[key] = child.map((element) =>
|
||||
isPlainRecord(element) ? redactSecretLikeConfigKeys(element) : element,
|
||||
);
|
||||
} else {
|
||||
result[key] = child;
|
||||
}
|
||||
result[key] = redactSecretLikeConfigValue(child);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Objects and arrays at any depth, including arrays nested in arrays. */
|
||||
function redactSecretLikeConfigValue(value: unknown): unknown {
|
||||
if (isPlainRecord(value)) return redactSecretLikeConfigKeys(value);
|
||||
if (Array.isArray(value)) return value.map(redactSecretLikeConfigValue);
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Floor view of a platform-provisioned environment on a cloud-managed
|
||||
* instance: credential-shaped config keys are never echoed — to ANY actor,
|
||||
|
||||
@@ -145,7 +145,7 @@ export {
|
||||
type ManagedInstanceConfig,
|
||||
} from "./managed-config.js";
|
||||
export { bootstrapExecutionPolicyFromEnv } from "./execution-policy-bootstrap.js";
|
||||
export { applyManagedEnvironments } from "./managed-environments.js";
|
||||
export { applyManagedEnvironments, createManagedEnvironmentRecovery } from "./managed-environments.js";
|
||||
export { buildExportFidelityReport, collectExportFidelityCounts } from "./export-fidelity.js";
|
||||
export { companyPortabilityService } from "./company-portability.js";
|
||||
export { teamsCatalogService } from "./teams-catalog.js";
|
||||
|
||||
@@ -146,16 +146,19 @@ function findSecretLikeConfigKey(value: Record<string, unknown>, path: string):
|
||||
for (const [key, child] of Object.entries(value)) {
|
||||
const childPath = path.length > 0 ? `${path}.${key}` : key;
|
||||
if (SECRET_LIKE_CONFIG_KEY_PATTERN.test(key)) return childPath;
|
||||
if (isPlainObject(child)) {
|
||||
const nested = findSecretLikeConfigKey(child, childPath);
|
||||
if (nested) return nested;
|
||||
} else if (Array.isArray(child)) {
|
||||
for (const [index, element] of child.entries()) {
|
||||
if (!isPlainObject(element)) continue;
|
||||
const nested = findSecretLikeConfigKey(element, `${childPath}[${index}]`);
|
||||
if (nested) return nested;
|
||||
}
|
||||
}
|
||||
const nested = findSecretLikeConfigKeyInValue(child, childPath);
|
||||
if (nested) return nested;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Objects and arrays at any depth, including arrays nested in arrays. */
|
||||
function findSecretLikeConfigKeyInValue(value: unknown, path: string): string | null {
|
||||
if (isPlainObject(value)) return findSecretLikeConfigKey(value, path);
|
||||
if (!Array.isArray(value)) return null;
|
||||
for (const [index, element] of value.entries()) {
|
||||
const nested = findSecretLikeConfigKeyInValue(element, `${path}[${index}]`);
|
||||
if (nested) return nested;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { unprocessable } from "../errors.js";
|
||||
import { applyManagedEnvironments } from "./managed-environments.js";
|
||||
import { applyManagedEnvironments, createManagedEnvironmentRecovery } from "./managed-environments.js";
|
||||
import { idleWorkSnapshot } from "./task-admission.js";
|
||||
import {
|
||||
PLUGIN_KEY,
|
||||
environmentsSeam,
|
||||
@@ -316,6 +317,77 @@ describe("applyManagedEnvironments plugin entries", () => {
|
||||
expect(manager.getWorker).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps waiting through a restart that replaces the worker handle", async () => {
|
||||
// Manager-wide `ready` subscription: the worker handle is never consulted,
|
||||
// so a replacement handle's `ready` still reaches the listener.
|
||||
const readyListeners = new Set<(pluginId: string) => void>();
|
||||
const manager = {
|
||||
isRunning: vi.fn(() => false),
|
||||
call: vi.fn(),
|
||||
getWorker: vi.fn(() => undefined),
|
||||
onWorkerReady: vi.fn((listener: (pluginId: string) => void) => {
|
||||
readyListeners.add(listener);
|
||||
return () => { readyListeners.delete(listener); };
|
||||
}),
|
||||
};
|
||||
const environments = environmentsSeam();
|
||||
await applyManagedEnvironments(noDb, configWith([pluginEntry()]), {
|
||||
env: {},
|
||||
workerManager: manager,
|
||||
environments,
|
||||
instanceSettings: instanceSettingsSeam(),
|
||||
resolveEnvironmentPlugin: readyPlugin(),
|
||||
autoInstalledPluginKeys: [PLUGIN_KEY],
|
||||
});
|
||||
expect(manager.getWorker).not.toHaveBeenCalled();
|
||||
|
||||
manager.isRunning.mockReturnValue(true);
|
||||
for (const listener of [...readyListeners]) listener("another-plugin");
|
||||
await tick();
|
||||
expect(environments.ensureManagedPluginEnvironment).not.toHaveBeenCalled();
|
||||
for (const listener of [...readyListeners]) listener("remote-host-plugin");
|
||||
await tick();
|
||||
expect(environments.ensureManagedPluginEnvironment).toHaveBeenCalledTimes(1);
|
||||
expect(readyListeners.size).toBe(0);
|
||||
});
|
||||
|
||||
it("tracks recovery attempts as idle work and lets shutdown await and unsubscribe them", async () => {
|
||||
const { handle, fireReady } = recoveryHandle();
|
||||
let finishEnsure!: () => void;
|
||||
const ensureManagedPluginEnvironment = vi.fn(() => new Promise((resolve) => {
|
||||
finishEnsure = () => resolve({ environment: pluginEnvironmentRow(), action: "updated" });
|
||||
}));
|
||||
const recovery = createManagedEnvironmentRecovery();
|
||||
const manager = workerManager(false, handle);
|
||||
await applyManagedEnvironments(noDb, configWith([pluginEntry()]), {
|
||||
env: {},
|
||||
workerManager: manager,
|
||||
environments: environmentsSeam({ ensureManagedPluginEnvironment: ensureManagedPluginEnvironment as never }),
|
||||
instanceSettings: instanceSettingsSeam(),
|
||||
resolveEnvironmentPlugin: readyPlugin(),
|
||||
autoInstalledPluginKeys: [PLUGIN_KEY],
|
||||
recovery,
|
||||
});
|
||||
const idleBefore = idleWorkSnapshot().active;
|
||||
manager.isRunning.mockReturnValue(true);
|
||||
fireReady();
|
||||
expect(idleWorkSnapshot().active).toBe(idleBefore + 1);
|
||||
|
||||
let stopped = false;
|
||||
const stopping = recovery.stop().then(() => { stopped = true; });
|
||||
await tick();
|
||||
expect(stopped).toBe(false);
|
||||
expect(handle.off).toHaveBeenCalled();
|
||||
finishEnsure();
|
||||
await stopping;
|
||||
expect(stopped).toBe(true);
|
||||
expect(idleWorkSnapshot().active).toBe(idleBefore);
|
||||
|
||||
fireReady();
|
||||
await tick();
|
||||
expect(ensureManagedPluginEnvironment).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("treats a missing plugin, a plugin without the driver, or a worker manager without call as unavailable", async () => {
|
||||
for (const [resolveEnvironmentPlugin, manager] of [
|
||||
[vi.fn(async () => null), workerManager()],
|
||||
|
||||
@@ -63,10 +63,50 @@ import type { ManagedResourceStockStatus } from "./managed-resource-drift.js";
|
||||
import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js";
|
||||
import { resolvePluginSandboxProviderDriverByKey } from "./plugin-environment-driver.js";
|
||||
import { pluginRegistryService } from "./plugin-registry.js";
|
||||
import { trackIdleWork } from "./task-admission.js";
|
||||
import { resolveBundledCatalogRoot, resolveBundledPluginInstalls } from "./bundled-plugins.js";
|
||||
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
|
||||
import type { ManagedPluginEnvironmentReconcileAction } from "./managed-plugin-environments.js";
|
||||
|
||||
/**
|
||||
* Lifecycle of the recovery work `applyManagedEnvironments` leaves behind
|
||||
* after boot (re-ensuring an archived environment when its plugin worker
|
||||
* becomes ready).
|
||||
*/
|
||||
export interface ManagedEnvironmentRecovery {
|
||||
/** Run one attempt as idle-tracked work, unless stopped. */
|
||||
run(work: () => Promise<void>): void;
|
||||
/** Register an unsubscribe function for `stop()`. Calls it at once when stopped. */
|
||||
onStop(unsubscribe: () => void): void;
|
||||
/** Unsubscribe every listener, refuse new attempts, and await unfinished ones. */
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
export function createManagedEnvironmentRecovery(): ManagedEnvironmentRecovery {
|
||||
const inFlight = new Set<Promise<void>>();
|
||||
const unsubscribers = new Set<() => void>();
|
||||
let stopped = false;
|
||||
return {
|
||||
run(work) {
|
||||
if (stopped) return;
|
||||
const attempt: Promise<void> = trackIdleWork(work())
|
||||
.catch(() => undefined)
|
||||
.finally(() => { inFlight.delete(attempt); });
|
||||
inFlight.add(attempt);
|
||||
},
|
||||
onStop(unsubscribe) {
|
||||
if (stopped) unsubscribe();
|
||||
else unsubscribers.add(unsubscribe);
|
||||
},
|
||||
async stop() {
|
||||
stopped = true;
|
||||
for (const unsubscribe of unsubscribers) unsubscribe();
|
||||
unsubscribers.clear();
|
||||
while (inFlight.size > 0) await Promise.allSettled([...inFlight]);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export interface ApplyManagedEnvironmentsOptions {
|
||||
env?: Record<string, string | undefined>;
|
||||
/**
|
||||
@@ -90,7 +130,7 @@ export interface ApplyManagedEnvironmentsOptions {
|
||||
* Plugin entries also need `call`, to run the plugin's own config
|
||||
* validation; without it every plugin entry is treated as unavailable.
|
||||
*/
|
||||
workerManager?: Pick<PluginWorkerManager, "isRunning"> & Partial<Pick<PluginWorkerManager, "call">> & {
|
||||
workerManager?: Pick<PluginWorkerManager, "isRunning"> & Partial<Pick<PluginWorkerManager, "call" | "onWorkerReady">> & {
|
||||
getWorker(pluginId: string):
|
||||
| {
|
||||
on(event: "ready", listener: (payload: { pluginId: string }) => void): void;
|
||||
@@ -134,6 +174,13 @@ export interface ApplyManagedEnvironmentsOptions {
|
||||
* startup. When omitted, the list is resolved here the same way.
|
||||
*/
|
||||
autoInstalledPluginKeys?: readonly string[];
|
||||
/**
|
||||
* Owns the post-boot recovery attempts and their `ready` subscriptions:
|
||||
* each attempt counts as idle-tracked work, and `stop()` (called on
|
||||
* shutdown before the database closes) unsubscribes and awaits unfinished
|
||||
* attempts. Defaults to a controller nobody stops (tests).
|
||||
*/
|
||||
recovery?: ManagedEnvironmentRecovery;
|
||||
}
|
||||
|
||||
export interface ManagedPluginEnvironmentReconciliationOutcome {
|
||||
@@ -341,27 +388,50 @@ export async function applyManagedEnvironments(
|
||||
};
|
||||
|
||||
// Recovery path for a `ready` plugin whose worker was down at check time:
|
||||
// that shape is usually a crash in restart-backoff, and the manager's
|
||||
// respawn re-emits `ready` on the same handle. A one-shot listener re-runs
|
||||
// the idempotent ensure so the recovered provider becomes selectable again
|
||||
// without waiting for the next boot (this pass has already archived the
|
||||
// row). The post-subscribe `isRunning` re-check closes the race where the
|
||||
// worker recovered between the gate check and the subscription. No handle
|
||||
// means no respawn is coming (activation never started a worker), so the
|
||||
// degraded-until-next-boot posture stands.
|
||||
const scheduleRecoveryReactivation = (pluginId: string, reactivateEntry: () => Promise<void>): void => {
|
||||
// that shape is usually a crash in restart-backoff, and the manager
|
||||
// re-emits `ready` when it respawns or replaces the worker. A one-shot
|
||||
// listener re-runs the idempotent ensure so the recovered provider becomes
|
||||
// selectable again without waiting for the next boot (this pass has already
|
||||
// archived the row). The post-subscribe `isRunning` re-check closes the race
|
||||
// where the worker recovered between the gate check and the subscription.
|
||||
// Each attempt runs through `recovery`, so idle checks count it and shutdown
|
||||
// awaits it.
|
||||
const recovery = opts.recovery ?? createManagedEnvironmentRecovery();
|
||||
|
||||
/**
|
||||
* Call `listener` on each `ready` of the plugin's worker. Prefers the
|
||||
* manager-wide subscription, which survives a restart replacing the worker
|
||||
* handle (for example a settings save that restarts the worker). Returns
|
||||
* the unsubscribe function, or null when there is no way to hear a `ready`
|
||||
* (no manager-wide subscription and no handle: activation never started a
|
||||
* worker, so no respawn is coming).
|
||||
*/
|
||||
const subscribeWorkerReady = (pluginId: string, listener: () => void): (() => void) | null => {
|
||||
const manager = opts.workerManager;
|
||||
if (manager?.onWorkerReady) {
|
||||
return manager.onWorkerReady((readyPluginId) => {
|
||||
if (readyPluginId === pluginId) listener();
|
||||
});
|
||||
}
|
||||
const handle = manager?.getWorker(pluginId);
|
||||
if (!handle) return;
|
||||
if (!handle) return null;
|
||||
handle.on("ready", listener);
|
||||
return () => handle.off("ready", listener);
|
||||
};
|
||||
|
||||
const scheduleRecoveryReactivation = (pluginId: string, reactivateEntry: () => Promise<void>): void => {
|
||||
let reactivated = false;
|
||||
let unsubscribe: (() => void) | null = null;
|
||||
const reactivate = (): void => {
|
||||
if (reactivated) return;
|
||||
reactivated = true;
|
||||
handle.off("ready", reactivate);
|
||||
void reactivateEntry();
|
||||
unsubscribe?.();
|
||||
recovery.run(reactivateEntry);
|
||||
};
|
||||
handle.on("ready", reactivate);
|
||||
if (manager?.isRunning(pluginId)) reactivate();
|
||||
unsubscribe = subscribeWorkerReady(pluginId, reactivate);
|
||||
if (!unsubscribe) return;
|
||||
recovery.onStop(unsubscribe);
|
||||
if (opts.workerManager?.isRunning(pluginId)) reactivate();
|
||||
};
|
||||
|
||||
const reactivateSandboxEntry = async (spec: ManagedInstanceConfig["environments"][number]): Promise<void> => {
|
||||
@@ -525,37 +595,35 @@ export async function applyManagedEnvironments(
|
||||
* worker that crashes during an attempt (for example inside
|
||||
* `environmentValidateConfig`) leaves the listener armed, so its next
|
||||
* respawn retries. A `ready` that arrives during an attempt queues one more
|
||||
* attempt. No handle means no respawn is coming (activation never started a
|
||||
* worker), so the degraded-until-next-boot posture stands.
|
||||
* attempt. The subscription survives a restart that replaces the worker
|
||||
* handle (see `subscribeWorkerReady`).
|
||||
*/
|
||||
const schedulePluginRecovery = (spec: PluginSpec, pluginId: string, workerAccess: PluginWorkerAccess): void => {
|
||||
const manager = opts.workerManager;
|
||||
const handle = manager?.getWorker(pluginId);
|
||||
if (!handle) return;
|
||||
const logBase = { name: spec.name, pluginKey: spec.pluginKey, driverKey: spec.driverKey };
|
||||
let settled = false;
|
||||
let inFlight = false;
|
||||
let queued = false;
|
||||
let unsubscribe: (() => void) | null = null;
|
||||
const settle = (): void => {
|
||||
settled = true;
|
||||
handle.off("ready", attempt);
|
||||
unsubscribe?.();
|
||||
};
|
||||
function attempt(): void {
|
||||
const attempt = (): void => {
|
||||
if (settled) return;
|
||||
if (inFlight) {
|
||||
queued = true;
|
||||
return;
|
||||
}
|
||||
inFlight = true;
|
||||
void ensurePluginEntry(spec, workerAccess)
|
||||
.then((result) => {
|
||||
recovery.run(async () => {
|
||||
try {
|
||||
const result = await ensurePluginEntry(spec, workerAccess);
|
||||
settle();
|
||||
logger.info(
|
||||
{ ...logBase, environmentId: result.environment.id, action: result.action },
|
||||
"managed plugin environment reactivated after plugin worker recovery",
|
||||
);
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
} catch (err) {
|
||||
if (isPermanentConfigRejection(err)) {
|
||||
settle();
|
||||
logger.error(
|
||||
@@ -568,17 +636,19 @@ export async function applyManagedEnvironments(
|
||||
"failed to reactivate managed plugin environment; retrying when the plugin worker is ready again",
|
||||
);
|
||||
}
|
||||
})
|
||||
.finally(() => {
|
||||
} finally {
|
||||
inFlight = false;
|
||||
if (queued && !settled) {
|
||||
queued = false;
|
||||
attempt();
|
||||
}
|
||||
});
|
||||
}
|
||||
handle.on("ready", attempt);
|
||||
if (manager?.isRunning(pluginId)) attempt();
|
||||
}
|
||||
});
|
||||
};
|
||||
unsubscribe = subscribeWorkerReady(pluginId, attempt);
|
||||
if (!unsubscribe) return;
|
||||
recovery.onStop(unsubscribe);
|
||||
if (opts.workerManager?.isRunning(pluginId)) attempt();
|
||||
};
|
||||
|
||||
for (const spec of pluginSpecs) {
|
||||
|
||||
@@ -832,6 +832,12 @@ export interface PluginWorkerManager {
|
||||
retireStartup?(contextId: string): Promise<void>;
|
||||
/** Subscribe after a worker successfully applies live company configuration. */
|
||||
onConfigApplied(listener: () => void): () => void;
|
||||
/**
|
||||
* Subscribe to every worker `ready`: first start, crash respawn, and a
|
||||
* replacement handle from a restart. Unlike `getWorker(id).on("ready")`, it
|
||||
* survives the handle being replaced. Returns the unsubscribe function.
|
||||
*/
|
||||
onWorkerReady?(listener: (pluginId: string) => void): () => void;
|
||||
inspectIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<{
|
||||
backgroundWork: "none" | "present" | "unknown"; pluginIds: string[];
|
||||
catalogGeneration?: string;
|
||||
@@ -4044,6 +4050,7 @@ export function createPluginWorkerManager(
|
||||
};
|
||||
const startupWorkers = () => new Set([...workers.values(), ...retiredStartupWorkers]);
|
||||
const configApplied = new Set<() => void>();
|
||||
const workerReady = new Set<(pluginId: string) => void>();
|
||||
let activeInspections = 0;
|
||||
/** Per-plugin startup locks to prevent concurrent spawn races. */
|
||||
const startupLocks = new Map<string, Promise<PluginWorkerHandle>>();
|
||||
@@ -4063,6 +4070,10 @@ export function createPluginWorkerManager(
|
||||
configApplied.add(listener);
|
||||
return () => { configApplied.delete(listener); };
|
||||
},
|
||||
onWorkerReady(listener) {
|
||||
workerReady.add(listener);
|
||||
return () => { workerReady.delete(listener); };
|
||||
},
|
||||
async inspectIdleSleep(hold) {
|
||||
if (startupLocks.size) return { backgroundWork: "unknown", pluginIds: [] };
|
||||
const snapshot = [...workers.entries()];
|
||||
@@ -4113,6 +4124,12 @@ export function createPluginWorkerManager(
|
||||
idleWorkerManifests.set(handle, pluginIdleManifestGeneration(options.manifest));
|
||||
if (existing) retainStartupWorker(existing);
|
||||
workers.set(pluginId, handle);
|
||||
handle.on("ready", (payload) => {
|
||||
for (const listener of [...workerReady]) {
|
||||
try { listener(payload.pluginId); }
|
||||
catch (err) { log.warn({ err, pluginId: payload.pluginId }, "worker ready listener failed"); }
|
||||
}
|
||||
});
|
||||
|
||||
// Subscribe to crash/ready events for live event forwarding
|
||||
if (managerOptions?.onWorkerEvent) {
|
||||
|
||||
Reference in new issue
Block a user