fix(workspaces): recover degraded runtime databases (#11651)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - Managed workspaces run local web services and embedded PostgreSQL
databases
> - A listening process could return an unhealthy response and still be
reused
> - Embedded PostgreSQL failures had no bounded restart owner
> - Cleanup inferred ownership from a branch slug instead of exact
persisted instance data
> - This pull request validates runtime health, supervises database
recovery, and uses exact cleanup ownership
> - The benefit is reliable replacement of degraded services without
deleting active instances
## Linked Issues or Issue Description
**What happened?**
Workspace reconciliation could reuse a degraded Paperclip process after
any successful HTTP response. Embedded PostgreSQL could stop without
bounded recovery. Cleanup could infer database ownership from a branch
slug and select the wrong instance.
**Expected behavior**
Paperclip must require a semantic healthy response from the assigned
loopback listener. It must replace degraded processes. It must supervise
embedded PostgreSQL with bounded restarts. Cleanup must use exact
persisted worktree and instance-root ownership.
**Steps to reproduce**
1. Start a managed workspace runtime.
2. Make its health endpoint return HTTP 200 with an unhealthy status, or
stop its embedded PostgreSQL process.
3. Reconcile the workspace or run instance cleanup.
4. Observe that the old implementation can reuse the degraded runtime or
infer ownership from its branch slug.
**Paperclip version or commit**
Current `master` before this pull request.
**Deployment mode**
Local dev with managed workspace services.
## What Changed
- Require `{ "status": "ok" }` from the assigned loopback health
endpoint before runtime reuse or adoption.
- Refresh persisted runtime health and replace degraded managed
processes.
- Add bounded embedded PostgreSQL restart supervision with coordinated
shutdown and hot-restart support.
- Stop the unhealthy web process when PostgreSQL recovery is exhausted
so reconciliation can replace it.
- Require exact persisted instance-root ownership before cleanup can
reclaim an embedded database.
- Add focused regression tests for degraded HTTP responses, ownership
mismatches, bounded recovery, active instance preservation, and
confirmed orphan reclamation.
## Verification
- `pnpm --filter @paperclipai/server typecheck`
- `pnpm --filter @paperclipai/server test --
src/embedded-postgres-supervisor.test.ts
src/services/workspace-instance-cleanup.test.ts
src/services/workspace-runtime.test.ts
src/services/execution-workspaces-service.test.ts`
- `pnpm -r typecheck`
- `pnpm build`
- `git diff --check`
- The full local stable test runner also found host-owned listeners on
ports 42000 and 52000. Those listeners conflict with the exposure test
fixture. The focused changed suites pass, and CI runs on a clean host.
## Risks
- A custom process that returns HTTP 2xx without the Paperclip health
contract is now degraded by design.
- Restart exhaustion terminates the managed web process. The runtime
reconciler then starts a clean process.
- Cleanup now fails closed when persisted ownership is missing. This can
retain an ambiguous orphan for manual review.
> 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
- OpenAI GPT-5 Codex. The exact serving revision and context-window size
are not exposed. The model used agentic reasoning, repository tools,
code execution, and test execution.
## 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)
- [ ] 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
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
---------
Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
e434829889
commit
1b4de0b65c
|
|
@ -0,0 +1,61 @@
|
|||
import { EventEmitter } from "node:events";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createEmbeddedPostgresSupervisor, type SupervisedEmbeddedPostgres } from "../embedded-postgres-supervisor.js";
|
||||
|
||||
function createInstance(startError?: Error) {
|
||||
const process = new EventEmitter();
|
||||
const instance: SupervisedEmbeddedPostgres = {
|
||||
process,
|
||||
start: vi.fn(async () => { if (startError) throw startError; }),
|
||||
stop: vi.fn(async () => undefined),
|
||||
};
|
||||
return { instance, process };
|
||||
}
|
||||
|
||||
describe("embedded PostgreSQL supervisor", () => {
|
||||
it("restarts PostgreSQL after its managed child exits unexpectedly", async () => {
|
||||
const initial = createInstance();
|
||||
const replacement = createInstance();
|
||||
const onRestarted = vi.fn();
|
||||
const supervisor = createEmbeddedPostgresSupervisor({
|
||||
initialInstance: initial.instance,
|
||||
createInstance: () => replacement.instance,
|
||||
restartDelaysMs: [0],
|
||||
onRestarted,
|
||||
});
|
||||
initial.process.emit("exit", 137, "SIGKILL");
|
||||
await supervisor.waitForRecovery();
|
||||
expect(replacement.instance.start).toHaveBeenCalledOnce();
|
||||
expect(supervisor.current()).toBe(replacement.instance);
|
||||
expect(onRestarted).toHaveBeenCalledWith(1);
|
||||
});
|
||||
|
||||
it("does not restart PostgreSQL during orderly shutdown", async () => {
|
||||
const initial = createInstance();
|
||||
const createReplacement = vi.fn(() => createInstance().instance);
|
||||
const supervisor = createEmbeddedPostgresSupervisor({
|
||||
initialInstance: initial.instance,
|
||||
createInstance: createReplacement,
|
||||
restartDelaysMs: [0],
|
||||
});
|
||||
await supervisor.shutdown();
|
||||
expect(initial.instance.stop).toHaveBeenCalledOnce();
|
||||
expect(createReplacement).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("bounds recovery attempts and reports the final failure", async () => {
|
||||
const initial = createInstance();
|
||||
const failures = [new Error("first"), new Error("second"), new Error("third")];
|
||||
const onRecoveryExhausted = vi.fn();
|
||||
const supervisor = createEmbeddedPostgresSupervisor({
|
||||
initialInstance: initial.instance,
|
||||
createInstance: () => createInstance(failures.shift()).instance,
|
||||
restartDelaysMs: [0, 0, 0],
|
||||
onRecoveryExhausted,
|
||||
});
|
||||
initial.process.emit("exit", 1, null);
|
||||
await supervisor.waitForRecovery();
|
||||
expect(onRecoveryExhausted).toHaveBeenCalledOnce();
|
||||
expect(onRecoveryExhausted).toHaveBeenCalledWith(expect.objectContaining({ message: "third" }));
|
||||
});
|
||||
});
|
||||
|
|
@ -114,7 +114,7 @@ describe("worktree instance cleanup", () => {
|
|||
await expect(fs.stat(instanceRoot)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("falls back to deterministic instance ownership when persisted root metadata is absent", async () => {
|
||||
it("preserves a collision-resistant active instance when persisted ownership is absent", async () => {
|
||||
const worktreesDir = await makeTempRoot("paperclip-managed-worktrees-");
|
||||
const workspacePath = await makeTempRoot("paperclip-cleanup-workspace-");
|
||||
const instanceId = deriveWorktreeInstanceId(workspacePath);
|
||||
|
|
@ -133,8 +133,9 @@ describe("worktree instance cleanup", () => {
|
|||
worktreesDir,
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({ status: "removed", instanceRoot });
|
||||
await expect(fs.stat(instanceRoot)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(result).toMatchObject({ status: "refused", instanceRoot });
|
||||
expect((result as { warning: string }).warning).toContain("no persisted instance root");
|
||||
await expect(fs.readFile(path.join(instanceRoot, "marker"), "utf8")).resolves.toBe("remove me");
|
||||
});
|
||||
|
||||
it("refuses and logs an instance pointer outside the managed worktree root", async () => {
|
||||
|
|
|
|||
|
|
@ -3919,6 +3919,86 @@ describe("ensureRuntimeServicesForRun", () => {
|
|||
}
|
||||
});
|
||||
|
||||
it("replaces a reused Paperclip dev runtime whose 2xx health payload is unhealthy", async () => {
|
||||
const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-misreported-health-"));
|
||||
const workspace = buildWorkspace(workspaceRoot);
|
||||
const serviceCommand =
|
||||
"node -e \"let healthy=true;const http=require('node:http');http.createServer((req,res)=>{if(req.url==='/misreport'){healthy=false;res.end('failed');return;}if(req.url==='/api/health'){res.setHeader('content-type','application/json');res.end(JSON.stringify(healthy?{status:'ok'}:{status:'unhealthy',error:'database_unreachable'}));return;}res.end('ok')}).listen(Number(process.env.PORT),'127.0.0.1')\"";
|
||||
const input = {
|
||||
actor: { id: "agent-1", name: "Codex Coder", companyId: "company-1" },
|
||||
issue: null,
|
||||
workspace,
|
||||
executionWorkspaceId: "execution-workspace-health",
|
||||
config: { workspaceRuntime: { services: [{
|
||||
name: "paperclip-dev",
|
||||
command: serviceCommand,
|
||||
cwd: ".",
|
||||
port: { type: "auto" as const },
|
||||
readiness: { type: "http" as const, urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 3, intervalMs: 100 },
|
||||
expose: { type: "url" as const, urlTemplate: "http://127.0.0.1:{{port}}" },
|
||||
lifecycle: "shared" as const,
|
||||
stopPolicy: { type: "manual" as const },
|
||||
}] } },
|
||||
adapterEnv: {},
|
||||
};
|
||||
try {
|
||||
const [first] = await startRuntimeServicesForWorkspaceControl(input);
|
||||
await expect(fetch(`${first!.url}/misreport`)).resolves.toMatchObject({ ok: true });
|
||||
await expect(fetch(`${first!.url}/api/health`)).resolves.toMatchObject({ ok: true });
|
||||
const [[replacement], [concurrentReuse]] = await Promise.all([
|
||||
startRuntimeServicesForWorkspaceControl(input),
|
||||
startRuntimeServicesForWorkspaceControl(input),
|
||||
]);
|
||||
expect(replacement?.id).not.toBe(first?.id);
|
||||
expect(replacement?.reused).toBe(false);
|
||||
expect(concurrentReuse?.id).toBe(replacement?.id);
|
||||
expect(concurrentReuse?.reused).toBe(true);
|
||||
} finally {
|
||||
await stopRuntimeServicesForExecutionWorkspace({
|
||||
executionWorkspaceId: "execution-workspace-health",
|
||||
workspaceCwd: workspaceRoot,
|
||||
});
|
||||
await fs.rm(workspaceRoot, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reuses a shared Paperclip dev runtime after one transient unhealthy response", async () => {
|
||||
const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-transient-health-"));
|
||||
const workspace = buildWorkspace(workspaceRoot);
|
||||
const serviceCommand =
|
||||
"node -e \"let failNext=false;const http=require('node:http');http.createServer((req,res)=>{if(req.url==='/fail-next'){failNext=true;res.end('armed');return;}if(req.url==='/api/health'){res.setHeader('content-type','application/json');const healthy=!failNext;failNext=false;res.end(JSON.stringify({status:healthy?'ok':'unhealthy'}));return;}res.end('ok')}).listen(Number(process.env.PORT),'127.0.0.1')\"";
|
||||
const input = {
|
||||
actor: { id: "agent-1", name: "Codex Coder", companyId: "company-1" },
|
||||
issue: null,
|
||||
workspace,
|
||||
executionWorkspaceId: "execution-workspace-transient-health",
|
||||
config: { workspaceRuntime: { services: [{
|
||||
name: "paperclip-dev",
|
||||
command: serviceCommand,
|
||||
cwd: ".",
|
||||
port: { type: "auto" as const },
|
||||
readiness: { type: "http" as const, urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 3, intervalMs: 100 },
|
||||
expose: { type: "url" as const, urlTemplate: "http://127.0.0.1:{{port}}" },
|
||||
lifecycle: "shared" as const,
|
||||
stopPolicy: { type: "manual" as const },
|
||||
}] } },
|
||||
adapterEnv: {},
|
||||
};
|
||||
try {
|
||||
const [first] = await startRuntimeServicesForWorkspaceControl(input);
|
||||
await expect(fetch(`${first!.url}/fail-next`)).resolves.toMatchObject({ ok: true });
|
||||
const [reused] = await startRuntimeServicesForWorkspaceControl(input);
|
||||
expect(reused?.id).toBe(first?.id);
|
||||
expect(reused?.reused).toBe(true);
|
||||
} finally {
|
||||
await stopRuntimeServicesForExecutionWorkspace({
|
||||
executionWorkspaceId: "execution-workspace-transient-health",
|
||||
workspaceCwd: workspaceRoot,
|
||||
});
|
||||
await fs.rm(workspaceRoot, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("uses explicit readiness URL when exposed URL is not the local probe address", async () => {
|
||||
const workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-runtime-explicit-readiness-"));
|
||||
const workspace = buildWorkspace(workspaceRoot);
|
||||
|
|
@ -6803,7 +6883,7 @@ describeEmbeddedPostgres("workspace runtime startup reconciliation", () => {
|
|||
const projectWorkspaceId = randomUUID();
|
||||
// Binds the app port and its HMR companion, both loopback-only.
|
||||
const command =
|
||||
"node -e \"const http=require('node:http');const p=Number(process.env.PORT);for(const q of [p,p+10000])http.createServer((req,res)=>res.end('ok')).listen(q,'127.0.0.1');setInterval(()=>{},1000)\"";
|
||||
"node -e \"const http=require('node:http');const p=Number(process.env.PORT);for(const q of [p,p+10000])http.createServer((req,res)=>{if(req.url==='/api/health'){res.setHeader('content-type','application/json');res.end(JSON.stringify({status:'ok'}));return;}res.end('ok')}).listen(q,'127.0.0.1');setInterval(()=>{},1000)\"";
|
||||
const workspaceRuntime = {
|
||||
services: [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -0,0 +1,92 @@
|
|||
export type EmbeddedPostgresExitListener = (code: number | null, signal: NodeJS.Signals | null) => void;
|
||||
|
||||
export interface SupervisedEmbeddedPostgres {
|
||||
start(): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
process?: { once(event: "exit", listener: EmbeddedPostgresExitListener): unknown };
|
||||
}
|
||||
|
||||
export interface EmbeddedPostgresSupervisor {
|
||||
current(): SupervisedEmbeddedPostgres;
|
||||
shutdown(): Promise<void>;
|
||||
waitForRecovery(): Promise<void>;
|
||||
}
|
||||
|
||||
type Options = {
|
||||
initialInstance: SupervisedEmbeddedPostgres;
|
||||
createInstance: () => SupervisedEmbeddedPostgres;
|
||||
beforeRestart?: (attempt: number) => Promise<void> | void;
|
||||
restartDelaysMs?: number[];
|
||||
delay?: (milliseconds: number) => Promise<void>;
|
||||
onUnexpectedExit?: EmbeddedPostgresExitListener;
|
||||
onRestartAttemptFailed?: (error: unknown, attempt: number) => void;
|
||||
onRestarted?: (attempt: number) => void;
|
||||
onRecoveryExhausted?: (error: unknown) => void;
|
||||
};
|
||||
|
||||
const defaultDelay = (milliseconds: number) => new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
|
||||
|
||||
export function createEmbeddedPostgresSupervisor(options: Options): EmbeddedPostgresSupervisor {
|
||||
const restartDelaysMs = options.restartDelaysMs ?? [0, 250, 1_000];
|
||||
const wait = options.delay ?? defaultDelay;
|
||||
let activeInstance = options.initialInstance;
|
||||
let activeInstanceExited = false;
|
||||
let shuttingDown = false;
|
||||
let recoveryPromise: Promise<void> | null = null;
|
||||
|
||||
const recover = async () => {
|
||||
let lastError: unknown = new Error("Embedded PostgreSQL exited unexpectedly");
|
||||
for (let index = 0; index < restartDelaysMs.length; index += 1) {
|
||||
if (shuttingDown) return;
|
||||
const attempt = index + 1;
|
||||
const delayMs = restartDelaysMs[index] ?? 0;
|
||||
if (delayMs > 0) await wait(delayMs);
|
||||
if (shuttingDown) return;
|
||||
try {
|
||||
await options.beforeRestart?.(attempt);
|
||||
const replacement = options.createInstance();
|
||||
await replacement.start();
|
||||
if (shuttingDown) {
|
||||
await replacement.stop();
|
||||
return;
|
||||
}
|
||||
activeInstance = replacement;
|
||||
activeInstanceExited = false;
|
||||
monitor(replacement);
|
||||
options.onRestarted?.(attempt);
|
||||
return;
|
||||
} catch (error) {
|
||||
lastError = error;
|
||||
options.onRestartAttemptFailed?.(error, attempt);
|
||||
}
|
||||
}
|
||||
if (!shuttingDown) options.onRecoveryExhausted?.(lastError);
|
||||
};
|
||||
|
||||
const monitor = (instance: SupervisedEmbeddedPostgres) => {
|
||||
const child = instance.process;
|
||||
if (!child) {
|
||||
options.onRecoveryExhausted?.(new Error("Embedded PostgreSQL started without a child process to monitor"));
|
||||
return;
|
||||
}
|
||||
child.once("exit", (code, signal) => {
|
||||
if (activeInstance !== instance) return;
|
||||
activeInstanceExited = true;
|
||||
if (shuttingDown) return;
|
||||
options.onUnexpectedExit?.(code, signal);
|
||||
recoveryPromise = recover().finally(() => { recoveryPromise = null; });
|
||||
});
|
||||
};
|
||||
|
||||
monitor(activeInstance);
|
||||
return {
|
||||
current: () => activeInstance,
|
||||
waitForRecovery: async () => { await recoveryPromise; },
|
||||
shutdown: async () => {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
await recoveryPromise;
|
||||
if (!activeInstanceExited) await activeInstance.stop();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -96,6 +96,11 @@ import {
|
|||
} from "./shutdown.js";
|
||||
import { systemdNotify } from "./services/systemd-notify.js";
|
||||
import { flushInFlightRunLogMirrors } from "./services/run-log-store.js";
|
||||
import {
|
||||
createEmbeddedPostgresSupervisor,
|
||||
type EmbeddedPostgresSupervisor,
|
||||
type SupervisedEmbeddedPostgres,
|
||||
} from "./embedded-postgres-supervisor.js";
|
||||
import type {
|
||||
InstanceDatabaseBackupRunResult,
|
||||
InstanceDatabaseBackupTrigger,
|
||||
|
|
@ -112,10 +117,8 @@ type BetterAuthSessionResult = {
|
|||
user: BetterAuthSessionUser | null;
|
||||
};
|
||||
|
||||
type EmbeddedPostgresInstance = {
|
||||
type EmbeddedPostgresInstance = SupervisedEmbeddedPostgres & {
|
||||
initialise(): Promise<void>;
|
||||
start(): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
};
|
||||
|
||||
type EmbeddedPostgresCtor = new (opts: {
|
||||
|
|
@ -326,6 +329,7 @@ export async function startServer(): Promise<StartedServer> {
|
|||
let db;
|
||||
let pluginMigrationDb;
|
||||
let embeddedPostgres: EmbeddedPostgresInstance | null = null;
|
||||
let embeddedPostgresSupervisor: EmbeddedPostgresSupervisor | null = null;
|
||||
let embeddedPostgresStartedByThisProcess = false;
|
||||
let migrationSummary: MigrationSummary = "skipped";
|
||||
let activeDatabaseConnectionString: string;
|
||||
|
|
@ -450,7 +454,7 @@ export async function startServer(): Promise<StartedServer> {
|
|||
}
|
||||
port = detectedPort;
|
||||
logger.info(`Using embedded PostgreSQL because no DATABASE_URL set (dataDir=${dataDir}, port=${port})`);
|
||||
embeddedPostgres = new EmbeddedPostgres({
|
||||
const createEmbeddedPostgres = () => new EmbeddedPostgres({
|
||||
databaseDir: dataDir,
|
||||
user: "paperclip",
|
||||
password: "paperclip",
|
||||
|
|
@ -460,6 +464,7 @@ export async function startServer(): Promise<StartedServer> {
|
|||
onLog: appendEmbeddedPostgresLog,
|
||||
onError: appendEmbeddedPostgresLog,
|
||||
});
|
||||
embeddedPostgres = createEmbeddedPostgres();
|
||||
|
||||
if (!clusterAlreadyInitialized) {
|
||||
try {
|
||||
|
|
@ -489,6 +494,36 @@ export async function startServer(): Promise<StartedServer> {
|
|||
});
|
||||
}
|
||||
embeddedPostgresStartedByThisProcess = true;
|
||||
embeddedPostgresSupervisor = createEmbeddedPostgresSupervisor({
|
||||
initialInstance: embeddedPostgres,
|
||||
createInstance: createEmbeddedPostgres,
|
||||
beforeRestart: () => {
|
||||
const runningPostgresPid = getRunningPid();
|
||||
if (runningPostgresPid) {
|
||||
throw new Error(`Refusing embedded PostgreSQL recovery because the data directory reports a live process (pid=${runningPostgresPid})`);
|
||||
}
|
||||
if (existsSync(postmasterPidFile)) rmSync(postmasterPidFile, { force: true });
|
||||
},
|
||||
onUnexpectedExit: (code, signal) => logger.error(
|
||||
{ code, signal, recentLogs: logBuffer.getRecentLogs() },
|
||||
"Embedded PostgreSQL exited unexpectedly; attempting recovery",
|
||||
),
|
||||
onRestartAttemptFailed: (err, attempt) => logger.error(
|
||||
{ err, attempt, recentLogs: logBuffer.getRecentLogs() },
|
||||
"Embedded PostgreSQL recovery attempt failed",
|
||||
),
|
||||
onRestarted: (attempt) => logger.info(
|
||||
{ attempt, port },
|
||||
"Embedded PostgreSQL recovered after unexpected exit",
|
||||
),
|
||||
onRecoveryExhausted: (err) => {
|
||||
logger.fatal(
|
||||
{ err, recentLogs: logBuffer.getRecentLogs() },
|
||||
"Embedded PostgreSQL recovery exhausted; stopping the unhealthy server",
|
||||
);
|
||||
process.kill(process.pid, "SIGTERM");
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1635,8 +1670,9 @@ export async function startServer(): Promise<StartedServer> {
|
|||
|
||||
const appShutdown = (app as { locals?: { paperclipShutdown?: () => Promise<void> } }).locals
|
||||
?.paperclipShutdown;
|
||||
const embeddedPostgresToStop =
|
||||
embeddedPostgres && embeddedPostgresStartedByThisProcess ? embeddedPostgres : null;
|
||||
const stopEmbeddedPostgres = embeddedPostgres && embeddedPostgresStartedByThisProcess
|
||||
? () => embeddedPostgresSupervisor?.shutdown() ?? embeddedPostgres!.stop()
|
||||
: null;
|
||||
|
||||
// Await the ordered application teardown before the process exits. A live
|
||||
// setup-token login session must stop and release its sandbox lease before
|
||||
|
|
@ -1645,7 +1681,7 @@ export async function startServer(): Promise<StartedServer> {
|
|||
await finalizeServerShutdown({
|
||||
signal,
|
||||
shutdownAppServices: appShutdown,
|
||||
stopEmbeddedPostgres: embeddedPostgresToStop ? () => embeddedPostgresToStop.stop() : null,
|
||||
stopEmbeddedPostgres,
|
||||
shutdownInstrumentation,
|
||||
log: logger,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -2218,6 +2218,13 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
|||
.where(eq(executionWorkspaces.id, id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row) return null;
|
||||
const { refreshPersistedRuntimeServiceHealth } = await import("./workspace-runtime.js");
|
||||
await refreshPersistedRuntimeServiceHealth({
|
||||
db,
|
||||
companyId: row.companyId,
|
||||
executionWorkspaceId: row.id,
|
||||
projectWorkspaceId: row.projectWorkspaceId,
|
||||
});
|
||||
const runtimeServicesByWorkspaceId = await loadEffectiveRuntimeServicesByExecutionWorkspace(db, row.companyId, [row]);
|
||||
return hydrateWorkspace(
|
||||
row,
|
||||
|
|
|
|||
|
|
@ -289,9 +289,12 @@ export async function cleanupWorktreeInstanceArtifacts(input: {
|
|||
return { status: "refused", instanceRoot: configuredInstanceRoot, warning };
|
||||
}
|
||||
|
||||
const expectedInstanceRoot = input.expectedInstanceRoot
|
||||
? path.resolve(input.expectedInstanceRoot)
|
||||
: null;
|
||||
const expectedInstanceRoot = input.expectedInstanceRoot ? path.resolve(input.expectedInstanceRoot) : null;
|
||||
if (!expectedInstanceRoot) {
|
||||
warning = `Refusing to remove instance directory "${configuredInstanceRoot}" because execution workspace ${input.workspaceId} has no persisted instance root.`;
|
||||
await recordRefusal({ refusalReason: "persisted_instance_root_missing" });
|
||||
return { status: "refused", instanceRoot: configuredInstanceRoot, warning };
|
||||
}
|
||||
if (expectedInstanceRoot && configuredInstanceRoot !== expectedInstanceRoot) {
|
||||
warning = `Refusing to remove instance directory "${configuredInstanceRoot}" because it does not match execution workspace ${input.workspaceId}'s persisted instance root "${expectedInstanceRoot}".`;
|
||||
await recordRefusal({
|
||||
|
|
|
|||
|
|
@ -232,6 +232,8 @@ const runtimeServicesById = new Map<string, RuntimeServiceRecord>();
|
|||
const runtimeServicesByReuseKey = new Map<string, string>();
|
||||
const runtimeServiceLeasesByRun = new Map<string, string[]>();
|
||||
const runtimeProvisionByWorkspace = new Map<string, Promise<void>>();
|
||||
const runtimeControlStartByOwner = new Map<string, Promise<void>>();
|
||||
const runtimeReplacementClaimsByReuseKey = new Map<string, number>();
|
||||
const quarantinedRuntimeExposurePorts = new Set<number>();
|
||||
/**
|
||||
* Pair-atomic in-process claims for exposure allocations that have not bound a
|
||||
|
|
@ -477,6 +479,8 @@ export async function resetRuntimeServicesForTests(
|
|||
runtimeServicesByReuseKey.clear();
|
||||
runtimeServiceLeasesByRun.clear();
|
||||
runtimeProvisionByWorkspace.clear();
|
||||
runtimeControlStartByOwner.clear();
|
||||
runtimeReplacementClaimsByReuseKey.clear();
|
||||
quarantinedRuntimeExposurePorts.clear();
|
||||
exposurePortPairClaims.clear();
|
||||
workspaceRuntimeExposureDeps = defaultWorkspaceRuntimeExposureDeps();
|
||||
|
|
@ -4726,14 +4730,26 @@ function resolveRuntimeServiceHealthUrl(
|
|||
|
||||
async function isRuntimeServiceUrlHealthy(
|
||||
url: string | null,
|
||||
input?: { serviceName?: string | null; command?: string | null },
|
||||
input?: {
|
||||
serviceName?: string | null;
|
||||
command?: string | null;
|
||||
provider?: string | null;
|
||||
port?: number | null;
|
||||
},
|
||||
) {
|
||||
if (!url) return true;
|
||||
const healthUrl = resolveRuntimeServiceHealthUrl(url, input);
|
||||
const localProbeUrl = input?.provider === "local_process" && input.port && isPaperclipDevRuntimeService(input)
|
||||
? `http://127.0.0.1:${input.port}`
|
||||
: null;
|
||||
const probeUrl = localProbeUrl ?? url;
|
||||
if (!probeUrl) return true;
|
||||
const healthUrl = resolveRuntimeServiceHealthUrl(probeUrl, input);
|
||||
if (!healthUrl) return false;
|
||||
try {
|
||||
const response = await fetch(healthUrl, { signal: AbortSignal.timeout(2_000) });
|
||||
return response.ok;
|
||||
if (!response.ok) return false;
|
||||
if (!isPaperclipDevRuntimeService(input ?? {})) return true;
|
||||
const payload = await response.json().catch(() => null) as { status?: unknown } | null;
|
||||
return payload?.status === "ok";
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
|
|
@ -5931,6 +5947,37 @@ async function stopRuntimeService(serviceId: string) {
|
|||
await persistRuntimeServiceRecord(record.db, record);
|
||||
}
|
||||
|
||||
async function findHealthyRunningRuntimeService(reuseKey: string | null) {
|
||||
const existingId = reuseKey ? runtimeServicesByReuseKey.get(reuseKey) : null;
|
||||
const existing = existingId ? runtimeServicesById.get(existingId) : null;
|
||||
if (!existing || existing.status !== "running") return null;
|
||||
const healthInput = {
|
||||
serviceName: existing.serviceName,
|
||||
command: existing.command,
|
||||
provider: existing.provider,
|
||||
port: existing.port,
|
||||
};
|
||||
let healthy = await isRuntimeServiceUrlHealthy(existing.url, healthInput);
|
||||
if (!healthy) {
|
||||
// A single timeout or connection reset is not enough evidence to destroy a
|
||||
// shared runtime that active runs may still use. Confirm the failure after
|
||||
// a short bounded delay before entering the destructive replacement path.
|
||||
await delay(250);
|
||||
healthy = await isRuntimeServiceUrlHealthy(existing.url, healthInput);
|
||||
}
|
||||
if (healthy) return existing;
|
||||
if (existing.leaseRunIds.size > 0) {
|
||||
existing.healthStatus = "unhealthy";
|
||||
if (reuseKey && runtimeServicesByReuseKey.get(reuseKey) === existing.id) {
|
||||
runtimeServicesByReuseKey.delete(reuseKey);
|
||||
}
|
||||
await persistRuntimeServiceRecord(existing.db, existing);
|
||||
return null;
|
||||
}
|
||||
await stopRuntimeService(existing.id);
|
||||
return null;
|
||||
}
|
||||
|
||||
async function markPersistedRuntimeServicesStoppedForExecutionWorkspace(input: {
|
||||
db: Db;
|
||||
executionWorkspaceId: string;
|
||||
|
|
@ -6254,7 +6301,7 @@ async function isPersistedIsolatedExecutionWorkspace(input: {
|
|||
return row?.mode === "isolated_workspace";
|
||||
}
|
||||
|
||||
export async function ensureRuntimeServicesForRun(input: {
|
||||
type EnsureRuntimeServicesForRunInput = {
|
||||
db?: Db;
|
||||
runId: string;
|
||||
agent: ExecutionWorkspaceAgentRef;
|
||||
|
|
@ -6265,7 +6312,11 @@ export async function ensureRuntimeServicesForRun(input: {
|
|||
adapterEnv: Record<string, string>;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
recorder?: WorkspaceOperationRecorder | null;
|
||||
}): Promise<RuntimeServiceRef[]> {
|
||||
};
|
||||
|
||||
async function ensureRuntimeServicesForRunInvocation(
|
||||
input: EnsureRuntimeServicesForRunInput,
|
||||
): Promise<RuntimeServiceRef[]> {
|
||||
const rawServices = selectRuntimeServiceEntries({
|
||||
config: input.config,
|
||||
respectDesiredStates: true,
|
||||
|
|
@ -6304,9 +6355,8 @@ export async function ensureRuntimeServicesForRun(input: {
|
|||
}).reuseKey;
|
||||
|
||||
if (reuseKey) {
|
||||
const existingId = runtimeServicesByReuseKey.get(reuseKey);
|
||||
const existing = existingId ? runtimeServicesById.get(existingId) : null;
|
||||
if (existing && existing.status === "running") {
|
||||
const existing = await findHealthyRunningRuntimeService(reuseKey);
|
||||
if (existing) {
|
||||
existing.leaseRunIds.add(input.runId);
|
||||
existing.lastUsedAt = new Date().toISOString();
|
||||
existing.stoppedAt = null;
|
||||
|
|
@ -6354,6 +6404,124 @@ export async function ensureRuntimeServicesForRun(input: {
|
|||
return refs;
|
||||
}
|
||||
|
||||
async function withRuntimeStartMutex<T>(ownerKey: string, start: () => Promise<T>): Promise<T> {
|
||||
const previous = runtimeControlStartByOwner.get(ownerKey) ?? Promise.resolve();
|
||||
let release!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const queued = previous.then(() => current);
|
||||
runtimeControlStartByOwner.set(ownerKey, queued);
|
||||
await previous;
|
||||
try {
|
||||
return await start();
|
||||
} finally {
|
||||
release();
|
||||
if (runtimeControlStartByOwner.get(ownerKey) === queued) runtimeControlStartByOwner.delete(ownerKey);
|
||||
}
|
||||
}
|
||||
|
||||
function resolveRuntimeStartMutexPlan(input: {
|
||||
services: Array<Record<string, unknown>>;
|
||||
workspace: RealizedExecutionWorkspace;
|
||||
executionWorkspaceId?: string | null;
|
||||
issue: ExecutionWorkspaceIssueRef | null;
|
||||
runId: string;
|
||||
agent: ExecutionWorkspaceAgentRef;
|
||||
adapterEnv: Record<string, string>;
|
||||
}) {
|
||||
const fallbackOwnerId = input.executionWorkspaceId
|
||||
?? input.workspace.workspaceId
|
||||
?? path.resolve(input.workspace.cwd);
|
||||
const replacementReuseKeys: string[] = [];
|
||||
const keys = input.services.map((service) => {
|
||||
const { scopeType, scopeId } = resolveServiceScopeId({
|
||||
service,
|
||||
workspace: input.workspace,
|
||||
executionWorkspaceId: input.executionWorkspaceId,
|
||||
issue: input.issue,
|
||||
runId: input.runId,
|
||||
agent: input.agent,
|
||||
});
|
||||
const reuseKey = resolveRuntimeServiceReuseIdentity({
|
||||
service,
|
||||
workspace: input.workspace,
|
||||
agent: input.agent,
|
||||
issue: input.issue,
|
||||
adapterEnv: input.adapterEnv,
|
||||
scopeType,
|
||||
scopeId,
|
||||
}).reuseKey;
|
||||
// Converge all callers that can replace an existing shared runtime on its
|
||||
// reuse identity. For an initial start, retain owner-level concurrency so
|
||||
// the exposure allocator's in-flight pair claims remain authoritative.
|
||||
if (
|
||||
reuseKey
|
||||
&& (runtimeServicesByReuseKey.has(reuseKey) || runtimeReplacementClaimsByReuseKey.has(reuseKey))
|
||||
) {
|
||||
runtimeReplacementClaimsByReuseKey.set(
|
||||
reuseKey,
|
||||
(runtimeReplacementClaimsByReuseKey.get(reuseKey) ?? 0) + 1,
|
||||
);
|
||||
replacementReuseKeys.push(reuseKey);
|
||||
return `reuse:${reuseKey}`;
|
||||
}
|
||||
return `${input.agent.companyId}:owner:${fallbackOwnerId}`;
|
||||
});
|
||||
return {
|
||||
ownerKeys: [...new Set(keys)].sort(),
|
||||
replacementReuseKeys,
|
||||
};
|
||||
}
|
||||
|
||||
function releaseRuntimeReplacementClaims(reuseKeys: string[]) {
|
||||
for (const reuseKey of reuseKeys) {
|
||||
const next = (runtimeReplacementClaimsByReuseKey.get(reuseKey) ?? 1) - 1;
|
||||
if (next <= 0) runtimeReplacementClaimsByReuseKey.delete(reuseKey);
|
||||
else runtimeReplacementClaimsByReuseKey.set(reuseKey, next);
|
||||
}
|
||||
}
|
||||
|
||||
async function withRuntimeStartMutexes<T>(
|
||||
ownerKeys: string[],
|
||||
start: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const acquire = async (index: number): Promise<T> => {
|
||||
const ownerKey = ownerKeys[index];
|
||||
if (!ownerKey) return await start();
|
||||
return await withRuntimeStartMutex(ownerKey, () => acquire(index + 1));
|
||||
};
|
||||
return await acquire(0);
|
||||
}
|
||||
|
||||
export async function ensureRuntimeServicesForRun(
|
||||
input: EnsureRuntimeServicesForRunInput,
|
||||
): Promise<RuntimeServiceRef[]> {
|
||||
const services = selectRuntimeServiceEntries({
|
||||
config: input.config,
|
||||
respectDesiredStates: true,
|
||||
defaultDesiredState: readDesiredRuntimeState(input.config.desiredState) ?? "running",
|
||||
serviceStates: readConfiguredServiceStates(input.config),
|
||||
});
|
||||
const mutexPlan = resolveRuntimeStartMutexPlan({
|
||||
services,
|
||||
workspace: input.workspace,
|
||||
executionWorkspaceId: input.executionWorkspaceId,
|
||||
issue: input.issue,
|
||||
runId: input.runId,
|
||||
agent: input.agent,
|
||||
adapterEnv: input.adapterEnv,
|
||||
});
|
||||
try {
|
||||
return await withRuntimeStartMutexes(
|
||||
mutexPlan.ownerKeys,
|
||||
() => ensureRuntimeServicesForRunInvocation(input),
|
||||
);
|
||||
} finally {
|
||||
releaseRuntimeReplacementClaims(mutexPlan.replacementReuseKeys);
|
||||
}
|
||||
}
|
||||
|
||||
type StartRuntimeServicesForWorkspaceControlInput = {
|
||||
db?: Db;
|
||||
invocationId?: string;
|
||||
|
|
@ -6417,9 +6585,8 @@ async function startRuntimeServicesForWorkspaceControlUnlocked(
|
|||
}).reuseKey;
|
||||
|
||||
if (reuseKey) {
|
||||
const existingId = runtimeServicesByReuseKey.get(reuseKey);
|
||||
const existing = existingId ? runtimeServicesById.get(existingId) : null;
|
||||
if (existing && existing.status === "running") {
|
||||
const existing = await findHealthyRunningRuntimeService(reuseKey);
|
||||
if (existing) {
|
||||
const prepared = options?.preparedProvisioning;
|
||||
if (prepared?.service === service && prepared.record.id !== existing.id && persistenceDb) {
|
||||
await persistenceDb
|
||||
|
|
@ -6543,7 +6710,7 @@ async function discardFailedDeferredRuntimeStart(db: Db, record: RuntimeServiceR
|
|||
await persistRuntimeServiceRecord(db, record);
|
||||
}
|
||||
|
||||
export async function startRuntimeServicesForWorkspaceControl(
|
||||
async function startRuntimeServicesForWorkspaceControlInvocation(
|
||||
input: StartRuntimeServicesForWorkspaceControlInput,
|
||||
): Promise<RuntimeServiceRef[]> {
|
||||
const rawServices = selectRuntimeServiceEntries({
|
||||
|
|
@ -6608,9 +6775,8 @@ export async function startRuntimeServicesForWorkspaceControl(
|
|||
scopeType,
|
||||
scopeId,
|
||||
}).reuseKey;
|
||||
const existingId = reuseKey ? runtimeServicesByReuseKey.get(reuseKey) : null;
|
||||
const existing = existingId ? runtimeServicesById.get(existingId) : null;
|
||||
if (existing?.status === "running") continue;
|
||||
const existing = await findHealthyRunningRuntimeService(reuseKey);
|
||||
if (existing) continue;
|
||||
|
||||
const record = await prepareRuntimeProvisioning({
|
||||
db: input.db,
|
||||
|
|
@ -6762,6 +6928,36 @@ export async function startRuntimeServicesForWorkspaceControl(
|
|||
}
|
||||
}
|
||||
|
||||
export async function startRuntimeServicesForWorkspaceControl(
|
||||
input: StartRuntimeServicesForWorkspaceControlInput,
|
||||
): Promise<RuntimeServiceRef[]> {
|
||||
const services = selectRuntimeServiceEntries({
|
||||
config: input.config,
|
||||
serviceIndex: input.serviceIndex,
|
||||
respectDesiredStates: input.respectDesiredStates,
|
||||
defaultDesiredState: readDesiredRuntimeState(input.config.desiredState) ?? "stopped",
|
||||
serviceStates: readConfiguredServiceStates(input.config),
|
||||
});
|
||||
const invocationId = input.invocationId ?? "workspace_control";
|
||||
const mutexPlan = resolveRuntimeStartMutexPlan({
|
||||
services,
|
||||
workspace: input.workspace,
|
||||
executionWorkspaceId: input.executionWorkspaceId,
|
||||
issue: input.issue,
|
||||
runId: invocationId,
|
||||
agent: input.actor,
|
||||
adapterEnv: input.adapterEnv,
|
||||
});
|
||||
try {
|
||||
return await withRuntimeStartMutexes(
|
||||
mutexPlan.ownerKeys,
|
||||
() => startRuntimeServicesForWorkspaceControlInvocation(input),
|
||||
);
|
||||
} finally {
|
||||
releaseRuntimeReplacementClaims(mutexPlan.replacementReuseKeys);
|
||||
}
|
||||
}
|
||||
|
||||
export async function releaseRuntimeServicesForRun(runId: string) {
|
||||
const acquired = runtimeServiceLeasesByRun.get(runId) ?? [];
|
||||
runtimeServiceLeasesByRun.delete(runId);
|
||||
|
|
@ -6773,7 +6969,14 @@ export async function releaseRuntimeServicesForRun(runId: string) {
|
|||
const stopType = asString(record.stopPolicy?.type, record.lifecycle === "ephemeral" ? "on_run_finish" : "manual");
|
||||
await persistRuntimeServiceRecord(record.db, record);
|
||||
if (record.leaseRunIds.size === 0) {
|
||||
if (record.lifecycle === "ephemeral" || stopType === "on_run_finish") {
|
||||
const detachedUnhealthySharedRuntime = record.healthStatus === "unhealthy"
|
||||
&& Boolean(record.reuseKey)
|
||||
&& runtimeServicesByReuseKey.get(record.reuseKey!) !== record.id;
|
||||
if (
|
||||
record.lifecycle === "ephemeral"
|
||||
|| stopType === "on_run_finish"
|
||||
|| detachedUnhealthySharedRuntime
|
||||
) {
|
||||
await stopRuntimeService(serviceId);
|
||||
continue;
|
||||
}
|
||||
|
|
@ -7004,6 +7207,59 @@ async function buildPersistedRuntimeExposureIntentLookup(db: Db) {
|
|||
};
|
||||
}
|
||||
|
||||
export async function refreshPersistedRuntimeServiceHealth(input: {
|
||||
db: Db;
|
||||
companyId: string;
|
||||
executionWorkspaceId: string;
|
||||
projectWorkspaceId?: string | null;
|
||||
}) {
|
||||
const ownershipCondition = input.projectWorkspaceId
|
||||
? or(
|
||||
eq(workspaceRuntimeServices.executionWorkspaceId, input.executionWorkspaceId),
|
||||
and(
|
||||
eq(workspaceRuntimeServices.projectWorkspaceId, input.projectWorkspaceId),
|
||||
eq(workspaceRuntimeServices.scopeType, "project_workspace"),
|
||||
),
|
||||
)
|
||||
: eq(workspaceRuntimeServices.executionWorkspaceId, input.executionWorkspaceId);
|
||||
const rows = await input.db
|
||||
.select({
|
||||
id: workspaceRuntimeServices.id,
|
||||
serviceName: workspaceRuntimeServices.serviceName,
|
||||
command: workspaceRuntimeServices.command,
|
||||
provider: workspaceRuntimeServices.provider,
|
||||
port: workspaceRuntimeServices.port,
|
||||
url: workspaceRuntimeServices.url,
|
||||
healthStatus: workspaceRuntimeServices.healthStatus,
|
||||
})
|
||||
.from(workspaceRuntimeServices)
|
||||
.where(and(
|
||||
eq(workspaceRuntimeServices.companyId, input.companyId),
|
||||
eq(workspaceRuntimeServices.provider, "local_process"),
|
||||
eq(workspaceRuntimeServices.status, "running"),
|
||||
ownershipCondition,
|
||||
));
|
||||
const results = await Promise.all(rows.map(async (row) => ({
|
||||
row,
|
||||
healthStatus: await isRuntimeServiceUrlHealthy(row.url, row) ? "healthy" as const : "unhealthy" as const,
|
||||
})));
|
||||
await Promise.all(results.map(async ({ row, healthStatus }) => {
|
||||
const liveRecord = runtimeServicesById.get(row.id);
|
||||
if (liveRecord) liveRecord.healthStatus = healthStatus;
|
||||
if (row.healthStatus === healthStatus) return;
|
||||
await input.db.update(workspaceRuntimeServices).set({ healthStatus, updatedAt: new Date() }).where(and(
|
||||
eq(workspaceRuntimeServices.id, row.id),
|
||||
eq(workspaceRuntimeServices.companyId, input.companyId),
|
||||
eq(workspaceRuntimeServices.status, "running"),
|
||||
));
|
||||
}));
|
||||
return {
|
||||
checked: results.length,
|
||||
healthy: results.filter((result) => result.healthStatus === "healthy").length,
|
||||
unhealthy: results.filter((result) => result.healthStatus === "unhealthy").length,
|
||||
};
|
||||
}
|
||||
|
||||
export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
|
||||
const rows = await db
|
||||
.select()
|
||||
|
|
@ -7185,7 +7441,12 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
|
|||
if (
|
||||
backfillDecision.action === "reprovision"
|
||||
|| !exposureHealthMatches
|
||||
|| !(await isRuntimeServiceUrlHealthy(adoptedUrl, { serviceName: row.serviceName, command: row.command }))
|
||||
|| !(await isRuntimeServiceUrlHealthy(adoptedUrl, {
|
||||
serviceName: row.serviceName,
|
||||
command: row.command,
|
||||
provider: "local_process",
|
||||
port: adoptedRecord.port ?? row.port,
|
||||
}))
|
||||
) {
|
||||
if (backfillDecision.action === "reprovision") backfilled += 1;
|
||||
await terminateLocalService(adoptedRecord);
|
||||
|
|
|
|||
Loading…
Reference in New Issue