mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
perf(environments): release a stopped plugin task once (#20203)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Plugin environment drivers with `supportsTasks` run agents as provider tasks, and warm runs keep a parked task for the next run > - A follow-up run can find that its parked task already stopped. Core then retires that task before it acquires a new one > - Core sent `stop`, `status`, and `complete` in the settle step. Then the plugin release hook sent `stop`, `status`, and `complete` again for the same task. Every call was on the critical path of the new run > - This pull request removes the repeated calls and keeps every call that does work no other call does > - The benefit is a faster start for a run whose parked task already stopped ## Linked Issues or Issue Description No issue exists. This follows the enhancement template. **What existing behavior does this improve?** The release of a plugin task lease that has already stopped. This occurs when a follow-up run finds that its parked warm task stopped. **Subsystem affected** Server: plugin environment drivers and the plugin Runner task lifecycle. **Current behavior** Core sends `stop`, `status`, and `complete` in the settle step. Then the plugin release hook sends `stop`, `status`, and `complete` again for the same task. Core also sends `stop` after resume already saw that the task stopped. On a hosted test deployment, 8 of 40 follow-up runs took this path. From the resume to the new acquire, these runs spent P50 1014 ms and P90 1080 ms. These 8 runs are the full overall P90 (1275 ms). **Proposed behavior** Core does not call the release hook when settle confirmed the stop and completed the same task. After a resume with no lease, Core sends `status` first, and sends `stop` only when `status` does not confirm the stop. **Reason and benefit** The repeated calls do no new work. Removing them saves about 355 ms, plus the `stop` share of about 187 ms, for each such run. These numbers are modeled from trace timings. **Breaking changes** None for the plugin API. A task driver must release task-scoped resources in its `status` or `complete` handling. The doc now says this. **Additional context** The settle `status` stays. Providers release task storage there, and it records the stop that gates the credential writeback and `complete`. ## What Changed - `settleDispatchedEnvironmentTask` now returns `"completed"` when it confirmed the stop and completed the task in this call, `"stopped"` when the lease was already settled, and `false` otherwise. - The plugin driver release does not call `environmentReleaseLease` when settle returned `"completed"`. In all other cases it calls the hook as before. - When `environmentResumeLease` returns no provider lease, the release sends `status` first. It sends `stop` (and a second `status`) only when that `status` does not confirm `executionStopped: true`. Startup and startup-cleanup leases keep the old `stop` first order. - The cleanup sweep (`retryPendingSandboxTeardown`) and `destroyRunLease` do not change. A settle failure still leaves the lease in `pending_cleanup`, and the sweep runs `stop`, `status`, `complete`, and the release hook. - `doc/plugins/PLUGIN_RUNNER_EXECUTION.md` tells drivers about the new order and that task-scoped resources must be released in `status` or `complete`. ### Dispatch count for a stopped parked task (resume to new acquire) | | Calls | | --- | --- | | Before | resume, stop, status, (credentials), complete, release hook (stop, status, complete), acquire: 9 provider calls (10 with Codex writeback) | | After | resume, status, (credentials), complete, acquire: 4 provider calls (5 with Codex writeback) | The modeled saving is about 540 ms for each such run: about 355 ms for the repeated release, and about 187 ms for the settle `stop` and `status`. This pull request keeps the settle `status`, so the real saving is about 355 ms plus the `stop` share of the 187 ms. These numbers are modeled from trace timings. They are not measured on this branch. ### What covers each removed call - Release hook `stop`, `status`, `complete`: settle already sent `stop` and `status` and checked `executionStopped: true` for the same task id (`lease.providerLeaseId`), then sent `complete`. The skip uses settle's success as its condition, not the code path. - Settle `stop` after a resume with no lease: the next `status` must report `executionStopped: true`. If it does not, Core sends `stop` and `status` as before. ### What each kept call protects - Settle `status`: providers release task storage and cancel preparations here. It also writes `runnerTaskExecutionStopped`, which the credential writeback and the `complete` bookkeeping need. - Credential writeback: Codex can rotate its token. The writeback must land before the new task installs credentials. It still runs inside `complete`, before the new acquire. - Settle `complete`: it retires every run lease that owned the task and clears the dispatch marker. - Release hook when settle did not complete a dispatched task in this call: the provider may still hold the lease. ## Verification - `pnpm --filter @paperclipai/shared --filter @paperclipai/plugin-sdk build` - `pnpm --filter @paperclipai/server typecheck` - `npx vitest run src/__tests__/plugin-warm-tasks.integration.test.ts src/__tests__/plugin-task-credential-writeback.test.ts src/__tests__/environment-task-runtime.test.ts` - Related suites: `environment-runtime`, `plugin-runner-task.integration`, `task-creation-recovery.integration`, `heartbeat-plugin-environment`, `native-startup-cleanup-recovery`, `plugin-environment-driver-ready-recovery`, `environment-task-authority.integration`, `native-workspace-warm-reuse`, `remote-workspaces.integration`, `legacy-workspace-restore-recovery`, `plugin-task-source-recovery.integration`, `managed-plugin-environments`. Locally, all of these pass except `heartbeat-plugin-environment` "inherits the instance default environment across companies…". That test also fails locally with the `origin/master` service files, so this change does not cause it. The existing `plugin-runner-task.integration` same-run replacement test now expects no release hook after a completed settle. New tests: - The stopped parked path sends exactly resume, `status`, `complete`, acquire. The `status` goes to the old task. - A resume with no lease for a task that still runs sends `status`, `stop`, `status`, `complete`. - A failure at `status`, `complete`, or `stop` leaves the lease in `pending_cleanup`, and the sweep then sends `stop`, `status`, `complete`, and the release hook. - With an observed stop, the Codex writeback runs before `complete`. A writeback failure leaves the lease for cleanup, and the retry writes back and completes. - A startup-cleanup lease still sends `stop` first. ## Risks - A third-party task driver that releases resources only in `onEnvironmentReleaseLease` would not get that call after a completed settle. The doc now states the requirement. The cleanup sweep still calls the hook after any failure. - A driver that returns no lease from resume while the task still runs costs one extra `status`. The behavior stays correct. - No migration. No API change for plugins. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use and code 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) - [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 - [ ] 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
9046884811
commit
990de1e0bf
7 files changed
+128
-38
No files matched your search
@@ -344,6 +344,15 @@ unsettled Core tool result, other close error, or continuation of the same
|
||||
session still blocks replacement.
|
||||
A controller restart is not required for this stopped-owner reset path.
|
||||
|
||||
Core ends a dispatched task lease with `stop`, `status`, and `complete`. When
|
||||
that sequence confirms the stop and completes the same task, Core does not also
|
||||
call `onEnvironmentReleaseLease`. Core still calls it when the release did not
|
||||
complete a dispatched task, and the cleanup retry calls it after any failure.
|
||||
When `onEnvironmentResumeLease` returns no provider lease, Core sends `status`
|
||||
first. It sends `stop` only when that `status` does not report
|
||||
`executionStopped: true`. A driver must release task-scoped resources in its
|
||||
`status` or `complete` handling, not only in `onEnvironmentReleaseLease`.
|
||||
|
||||
Termination receipts cover all run leases that owned the retired physical task.
|
||||
A stopped task gets a new provider task ID for checkpoint resumption. Inspection
|
||||
precedes in-memory session selection so a replacement cannot inherit the stopped
|
||||
|
||||
@@ -85,9 +85,25 @@ describe("plugin-provided Runner execution", () => {
|
||||
expect(operations).not.toContain("complete");
|
||||
release();
|
||||
const result = await checked;
|
||||
if (preparationStopped) { expect(result).toEqual({ value: true, error: null }); expect(operations).toEqual(["stop", "status", "complete"]); }
|
||||
if (preparationStopped) { expect(result).toEqual({ value: "completed", error: null }); expect(operations).toEqual(["stop", "status", "complete"]); }
|
||||
else { expect(result.error?.message).toContain("termination is unconfirmed"); expect(fixture.sets).toContainEqual(expect.objectContaining({ status: "pending_cleanup", cleanupStatus: "failed" })); }
|
||||
});
|
||||
it("still stops a startup task first when resume already observed the stop", async () => {
|
||||
const value = row();
|
||||
Object.assign(value.lease.metadata, { runnerTaskDispatched: false,
|
||||
runnerTaskStartupCleanup: { contextId: "startup", requiresPreparationStop: true } });
|
||||
const fixture = database(value), workers = worker();
|
||||
fixture.db.select = ((...args: unknown[]) => (database(value).db.select as (...values: unknown[]) => unknown)(...args)) as Db["select"];
|
||||
Object.assign(workers, { drainStartup: vi.fn(async () => undefined) });
|
||||
const operations: string[] = [];
|
||||
workers.call.mockImplementation(async (...args: any[]) => {
|
||||
const operation = args[2].operation.kind; operations.push(operation);
|
||||
return operation === "status" ? { kind: "status", taskId: "attempt-1", phase: "cancelled", executionStopped: true, preparationStopped: true }
|
||||
: { kind: "accepted", taskId: "attempt-1" };
|
||||
});
|
||||
await expect(settleDispatchedEnvironmentTask(fixture.db, workers as never, { companyId: "company", id: leaseId }, { executionStopObserved: true })).resolves.toBe("completed");
|
||||
expect(operations).toEqual(["stop", "status", "complete"]);
|
||||
});
|
||||
it("negotiates the transient context without reading transport for an older plugin", async () => {
|
||||
const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000);
|
||||
const workers = worker();
|
||||
|
||||
@@ -416,7 +416,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => {
|
||||
return;
|
||||
}
|
||||
const next = await acquire;
|
||||
expect(operations).toEqual(["stop", "status", "complete", "environmentReleaseLease", "environmentAcquireLease"]);
|
||||
expect(operations).toEqual(["stop", "status", "complete", "environmentAcquireLease"]);
|
||||
expect(next.lease.id).not.toBe(oldLeaseId);
|
||||
await expect(resolvePluginTaskResume(db, { companyId, leaseId: next.lease.id, priorIdentity })).resolves.toBe("old-task");
|
||||
await expect(runtime.task({ companyId, leaseId: next.lease.id, operation: { kind: "submit", projectIds: [], bootstrapTicket: "fixture", runner } })).rejects.toThrow("identity mismatch");
|
||||
|
||||
@@ -65,11 +65,11 @@ suite("durable plugin task credential handoff", () => {
|
||||
const lease = () => db.select().from(environmentLeases).where(and(eq(environmentLeases.id, leaseId), eq(environmentLeases.companyId, companyId))).then(rows => rows[0]!);
|
||||
return { companyId, agentId, runId, environmentId, leaseId, owner, selection: selection!, managed, workers, calls, lease,
|
||||
failExport(value: boolean) { failExport = value; }, credential: () => service.credential(selection!),
|
||||
settle: () => settleDispatchedEnvironmentTask(db, workers, { companyId, id: leaseId }) };
|
||||
settle: (executionStopObserved = false) => settleDispatchedEnvironmentTask(db, workers, { companyId, id: leaseId }, { executionStopObserved }) };
|
||||
}
|
||||
it("persists refreshed credentials before completing a stopped task, after controller files disappear", async () => {
|
||||
const f = await fixture();
|
||||
expect(await f.settle()).toBe(true);
|
||||
expect(await f.settle()).toBe("completed");
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect(f.calls.indexOf("credentials")).toBeLessThan(f.calls.indexOf("complete"));
|
||||
const saved = await f.lease();
|
||||
@@ -77,6 +77,24 @@ suite("durable plugin task credential handoff", () => {
|
||||
expect(JSON.stringify(saved.metadata)).not.toContain("fixture-refreshed");
|
||||
expect(JSON.stringify(saved.metadata)).not.toContain(f.managed.home!);
|
||||
});
|
||||
it("writes back credentials before completing a task whose stop was already observed, without stopping it again", async () => {
|
||||
const f = await fixture();
|
||||
expect(await f.settle(true)).toBe("completed");
|
||||
expect(f.calls).toEqual(["status", "credentials", "complete"]);
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" });
|
||||
});
|
||||
it("leaves an observed-stop task for cleanup when credential writeback fails", async () => {
|
||||
const f = await fixture(); f.failExport(true);
|
||||
await expect(f.settle(true)).rejects.toThrow("termination is unconfirmed");
|
||||
expect(f.calls).toEqual(["status", "credentials"]);
|
||||
expect(await f.lease()).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed" });
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" });
|
||||
f.failExport(false);
|
||||
expect(await f.settle()).toBe("completed");
|
||||
expect(f.calls.slice(2)).toEqual(["stop", "status", "credentials", "complete"]);
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
});
|
||||
it("keeps a failed transfer pending and recovers through a fresh cleanup after run and environment deletion", async () => {
|
||||
const f = await fixture(); f.failExport(true);
|
||||
await expect(f.settle()).rejects.toThrow("termination is unconfirmed");
|
||||
@@ -86,7 +104,7 @@ suite("durable plugin task credential handoff", () => {
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.delete(environments).where(eq(environments.id, f.environmentId));
|
||||
f.failExport(false);
|
||||
expect(await f.settle()).toBe(true);
|
||||
expect(await f.settle()).toBe("completed");
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect(f.calls.at(-1)).toBe("complete");
|
||||
});
|
||||
@@ -104,7 +122,7 @@ suite("durable plugin task credential handoff", () => {
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" });
|
||||
} finally { await db.execute(sql`drop trigger reject_runner_credential_save on company_secret_versions; drop function reject_runner_credential_save()`); }
|
||||
const receiver = createDb(database.connectionString);
|
||||
expect(await settleDispatchedEnvironmentTask(receiver, f.workers, { companyId: f.companyId, id: f.leaseId })).toBe(true);
|
||||
expect(await settleDispatchedEnvironmentTask(receiver, f.workers, { companyId: f.companyId, id: f.leaseId })).toBe("completed");
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect(f.calls.filter(call => call === "credentials")).toHaveLength(2);
|
||||
expect(f.calls.at(-1)).toBe("complete");
|
||||
@@ -113,13 +131,13 @@ suite("durable plugin task credential handoff", () => {
|
||||
it("does not overwrite an explicit account replacement", async () => {
|
||||
const f = await fixture();
|
||||
await secretService(db).rotate(f.managed.credentialWriteback!.secretId, { value: auth("fixture-reconnected", 2022) }, { userId: f.owner });
|
||||
expect(await f.settle()).toBe(true);
|
||||
expect(await f.settle()).toBe("completed");
|
||||
expect(f.calls).not.toContain("credentials");
|
||||
expect(await f.credential()).toBe(auth("fixture-reconnected", 2022));
|
||||
});
|
||||
it("allows concurrent cleanup to complete the same durable handoff", async () => {
|
||||
const f = await fixture();
|
||||
expect(await Promise.all([f.settle(), f.settle()])).toEqual([true, true]);
|
||||
expect(await Promise.all([f.settle(), f.settle()])).not.toContain(false);
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" });
|
||||
});
|
||||
|
||||
@@ -7,6 +7,7 @@ import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { agents, companies, createDb, applyPendingMigrations, environmentLeases, environments, heartbeatRuns, issues, plugins, projects, sandboxCreationOperations } from "@paperclipai/db";
|
||||
import { environmentRuntimeService } from "../services/environment-runtime.js";
|
||||
import { environmentService } from "../services/environments.js";
|
||||
import { recordSandboxCreationResource } from "../services/sandbox-creation-operations.js";
|
||||
import { resolvePluginTaskResume } from "../services/environment-task-runtime.js";
|
||||
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
|
||||
@@ -20,8 +21,8 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport
|
||||
afterAll(async()=>{await stop?.();});
|
||||
async function fixture(initialProjects:string[]=[],initialExpiry?:string,deadline?:Date){
|
||||
const companyId=randomUUID(),agentId=randomUUID(),issueId=randomUUID(),environmentId=randomUUID(),pluginId=randomUUID();
|
||||
let runId=randomUUID(),allocation=0,stopped=false,outage=false,releaseFails=false;
|
||||
const calls:string[]=[];
|
||||
let runId=randomUUID(),allocation=0,stopped=false,outage=false,releaseFails=false,unavailable=false;
|
||||
const calls:string[]=[],dispatched:Array<[string,string]>=[],failures=new Set<string>();
|
||||
let projectIds=[...initialProjects],providerExpiry=initialExpiry;
|
||||
const taskExpiries=new Map<string,string|undefined>(),acquireRequests:any[]=[];
|
||||
let resumeGate:Promise<void>|undefined,releaseResume:(()=>void)|undefined,resumes=0;
|
||||
@@ -38,12 +39,14 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport
|
||||
const workers={isRunning:()=>true,getWorker:()=>({supportedMethods:["environmentTask","environmentAcquireLease","environmentResumeLease","environmentReleaseLease"]}),
|
||||
call:async(_id:string,method:string,p:any)=>{
|
||||
calls.push(method==="environmentTask" ? p.operation.kind : method);
|
||||
if(method==="environmentTask") dispatched.push([p.operation.kind,p.taskId]);
|
||||
if(method==="environmentTask" && failures.delete(p.operation.kind)) throw Error(`${p.operation.kind} unavailable`);
|
||||
if(method==="environmentAcquireLease") {
|
||||
const id=`task-${++allocation}`;taskExpiries.set(id,providerExpiry);acquireRequests.push(p);
|
||||
return {providerLeaseId:id,...(providerExpiry ? {expiresAt:providerExpiry}:{}),metadata:{runId:p.runId}};
|
||||
}
|
||||
if(method==="environmentResumeLease") {if(outage) throw Error("unreachable");
|
||||
if(resumeGate){if(++resumes===2)releaseResume!();await resumeGate;}return {providerLeaseId:stopped?null:p.providerLeaseId,expiresAt:taskExpiries.get(p.providerLeaseId),metadata:{runId:p.runId}};}
|
||||
if(resumeGate){if(++resumes===2)releaseResume!();await resumeGate;}return {providerLeaseId:stopped||unavailable?null:p.providerLeaseId,expiresAt:taskExpiries.get(p.providerLeaseId),metadata:{runId:p.runId}};}
|
||||
if(method==="environmentReleaseLease") {if(releaseFails) throw Error("release unavailable");return {providerLeaseId:p.providerLeaseId,state:"stopped"};}
|
||||
if(p.operation.kind==="status") return {kind:"status",taskId:p.taskId,phase:stopped?"completed":"running",executionStopped:stopped,sessionStateAvailable:stopped};
|
||||
if(p.operation.kind==="read_state") return {kind:"state",taskId:p.taskId,state:{...p.operation.identity,lifecycle:"ready"}};
|
||||
@@ -63,7 +66,8 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport
|
||||
launch:{runnerVersion:"fixture",runnerDigest:`sha256:${"a".repeat(64)}`,maxOutboxBytes:262144,p0ReserveBytes:65536,lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}}}});
|
||||
await runtime().releaseRunLeases(runId,"released",undefined,"keep_running");
|
||||
await db.update(heartbeatRuns).set({status:"succeeded"}).where(eq(heartbeatRuns.id,runId));
|
||||
return {companyId,first,identity,calls,runtime,acquire,acquireRequests,
|
||||
return {companyId,first,identity,calls,dispatched,runtime,acquire,acquireRequests,
|
||||
fail(kind:string){failures.add(kind);},unavailable(){unavailable=true;},
|
||||
setExpiry(next?:string){providerExpiry=next;},
|
||||
async nextRun(){runId=randomUUID();await db.insert(heartbeatRuns).values({id:runId,companyId,agentId,nativeIssueId:issueId,invocationSource:"manual",status:"running"});return runId;},
|
||||
setProjects(next:string[]){projectIds=next;},stopped(){stopped=true;},failRelease(){releaseFails=true;},outage(){outage=true;},
|
||||
@@ -109,6 +113,35 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport
|
||||
expect(f.calls).toContain("complete");
|
||||
await expect(resolvePluginTaskResume(db,{companyId:f.companyId,leaseId:next.lease.id,priorIdentity:f.identity})).resolves.toBe(f.first.lease.providerLeaseId);
|
||||
});
|
||||
it("retires a stopped retained task with one status and complete before allocating its replacement",async()=>{
|
||||
const f=await fixture();f.stopped();await f.nextRun();
|
||||
const calls=f.calls.length,dispatched=f.dispatched.length;
|
||||
const next=await f.acquire();
|
||||
expect(f.calls.slice(calls)).toEqual(["environmentResumeLease","status","complete","environmentAcquireLease"]);
|
||||
expect(f.dispatched.slice(dispatched)).toEqual([["status",f.first.lease.providerLeaseId],["complete",f.first.lease.providerLeaseId]]);
|
||||
expect(next.lease.providerLeaseId).not.toBe(f.first.lease.providerLeaseId);
|
||||
const [old]=await db.select().from(environmentLeases).where(eq(environmentLeases.id,f.first.lease.id));
|
||||
expect(old.status).toBe("released");
|
||||
expect(old.metadata).toMatchObject({runnerTaskExecutionStopped:true,runnerTaskDispatched:false,remoteExecutionTermination:{providerLeaseId:f.first.lease.providerLeaseId,state:"stopped"}});
|
||||
});
|
||||
it("stops a retained task that resume could not attach but that is still running",async()=>{
|
||||
const f=await fixture();f.unavailable();await f.nextRun();
|
||||
const calls=f.calls.length;
|
||||
await f.acquire();
|
||||
expect(f.calls.slice(calls)).toEqual(["environmentResumeLease","status","stop","status","complete","environmentAcquireLease"]);
|
||||
});
|
||||
it.each([["status",false],["complete",false],["stop",true]] as const)("keeps a stopped retained task for cleanup retry when %s fails",async(kind,running)=>{
|
||||
const f=await fixture();if(running) f.unavailable(); else f.stopped();await f.nextRun();
|
||||
f.fail(kind);
|
||||
await expect(f.acquire()).rejects.toThrow("termination is unconfirmed");
|
||||
expect(f.calls.filter(call=>call==="environmentAcquireLease")).toHaveLength(1);
|
||||
expect(f.calls).not.toContain("environmentReleaseLease");
|
||||
const pending=(await environmentService(db).getLeaseById(f.first.lease.id))!;
|
||||
expect(pending).toMatchObject({status:"pending_cleanup",cleanupStatus:"failed"});
|
||||
const calls=f.calls.length;
|
||||
await expect(f.runtime().retryPendingSandboxTeardown({environment:null,lease:pending})).resolves.toEqual({providerLeaseId:f.first.lease.providerLeaseId,state:"stopped"});
|
||||
expect(f.calls.slice(calls)).toEqual(["stop","status","complete","environmentReleaseLease"]);
|
||||
});
|
||||
it("does not allocate a replacement when the retained computer cannot be inspected",async()=>{
|
||||
const f=await fixture();f.outage();await f.nextRun();
|
||||
await expect(f.acquire()).rejects.toThrow("unreachable");
|
||||
|
||||
@@ -3322,6 +3322,30 @@ function createPluginEnvironmentDriver(
|
||||
};
|
||||
}
|
||||
|
||||
async function releasePluginRunLease(input: EnvironmentDriverReleaseInput, executionStopObserved = false) {
|
||||
await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"pluginLeaseCleanupMethod":"release"}'::jsonb` }).where(and(eq(environmentLeases.id, input.lease.id), eq(environmentLeases.companyId, input.lease.companyId)));
|
||||
const settled = await settleDispatchedEnvironmentTask(db, workerManager, input.lease, { executionStopObserved });
|
||||
const taskStopped = settled !== false ||
|
||||
(input.lease.metadata?.runnerTaskLease === true && input.lease.metadata?.runnerTaskDispatched !== true && !input.lease.metadata?.runnerTaskStartup && !input.lease.metadata?.runnerTaskStartupCleanup);
|
||||
if (settled !== "completed") {
|
||||
const { plugin, driverKey, driverConfig } = await resolvePluginDriverForRelease(input);
|
||||
await workerManager.call(plugin.id, "environmentReleaseLease", {
|
||||
driverKey,
|
||||
companyId: input.lease.companyId,
|
||||
environmentId: input.environment.id,
|
||||
issueId: input.lease.issueId,
|
||||
config: driverConfig,
|
||||
providerLeaseId: input.lease.providerLeaseId,
|
||||
leaseMetadata: input.lease.metadata ?? undefined,
|
||||
});
|
||||
}
|
||||
return await environmentsSvc.releaseLease(input.lease.id, input.status, {
|
||||
cleanupStatus: "success",
|
||||
...(taskStopped ? { remoteExecutionTermination: remoteTerminationReceipt(input.lease,
|
||||
{ providerLeaseId: input.lease.providerLeaseId, state: "stopped" }) } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
driver: "plugin",
|
||||
|
||||
@@ -3363,7 +3387,7 @@ function createPluginEnvironmentDriver(
|
||||
providerLeaseId: previous.providerLeaseId, leaseMetadata: previous.metadata ?? undefined,
|
||||
});
|
||||
if (!resumed?.providerLeaseId) {
|
||||
await this.releaseRunLease({ environment: input.environment, lease: toEnvironmentLeaseSnapshot(previous), status: "released" });
|
||||
await releasePluginRunLease({ environment: input.environment, lease: toEnvironmentLeaseSnapshot(previous), status: "released" }, true);
|
||||
resumed = undefined;
|
||||
}
|
||||
}
|
||||
@@ -3423,25 +3447,8 @@ function createPluginEnvironmentDriver(
|
||||
});
|
||||
},
|
||||
|
||||
async releaseRunLease(input) {
|
||||
await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"pluginLeaseCleanupMethod":"release"}'::jsonb` }).where(and(eq(environmentLeases.id, input.lease.id), eq(environmentLeases.companyId, input.lease.companyId)));
|
||||
const taskStopped = await settleDispatchedEnvironmentTask(db, workerManager, input.lease) ||
|
||||
(input.lease.metadata?.runnerTaskLease === true && input.lease.metadata?.runnerTaskDispatched !== true && !input.lease.metadata?.runnerTaskStartup && !input.lease.metadata?.runnerTaskStartupCleanup);
|
||||
const { plugin, driverKey, driverConfig } = await resolvePluginDriverForRelease(input);
|
||||
await workerManager.call(plugin.id, "environmentReleaseLease", {
|
||||
driverKey,
|
||||
companyId: input.lease.companyId,
|
||||
environmentId: input.environment.id,
|
||||
issueId: input.lease.issueId,
|
||||
config: driverConfig,
|
||||
providerLeaseId: input.lease.providerLeaseId,
|
||||
leaseMetadata: input.lease.metadata ?? undefined,
|
||||
});
|
||||
return await environmentsSvc.releaseLease(input.lease.id, input.status, {
|
||||
cleanupStatus: "success",
|
||||
...(taskStopped ? { remoteExecutionTermination: remoteTerminationReceipt(input.lease,
|
||||
{ providerLeaseId: input.lease.providerLeaseId, state: "stopped" }) } : {}),
|
||||
});
|
||||
releaseRunLease(input) {
|
||||
return releasePluginRunLease(input);
|
||||
},
|
||||
|
||||
async resumeRunLease(input) {
|
||||
|
||||
@@ -493,7 +493,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage
|
||||
}
|
||||
|
||||
/** Failure leaves the dispatch marker and provider handle available to cleanup retries. */
|
||||
export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWorkerManager, lease: { id: string; companyId: string }) {
|
||||
export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWorkerManager, lease: { id: string; companyId: string },
|
||||
options: { executionStopObserved?: boolean } = {}): Promise<false | "stopped" | "completed"> {
|
||||
const [current] = await db.update(environmentLeases).set({
|
||||
metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskRetiring":true}'::jsonb`,
|
||||
}).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId)))
|
||||
@@ -503,19 +504,25 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor
|
||||
if (current?.metadata?.runnerTaskDispatched !== true && !startup && !startupCleanup) {
|
||||
if (current?.metadata) await settlePluginTaskCredentialHandoff(db, { companyId: lease.companyId, leaseId: lease.id }, current.metadata,
|
||||
async () => { throw new Error("Undispatched task must not request credentials"); });
|
||||
return current?.metadata?.runnerTaskExecutionStopped === true;
|
||||
return current?.metadata?.runnerTaskExecutionStopped === true ? "stopped" : false;
|
||||
}
|
||||
const scope = { companyId: lease.companyId, leaseId: lease.id };
|
||||
try {
|
||||
const contextId = (startupCleanup as { contextId?: unknown } | undefined)?.contextId ?? (startup as { contextId?: unknown } | undefined)?.contextId;
|
||||
const draining = typeof contextId === "string" ? workers.drainStartup?.(contextId) : undefined;
|
||||
void draining?.catch(() => undefined);
|
||||
await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "stop" } });
|
||||
const state = await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } });
|
||||
if (state.kind !== "status" || state.executionStopped !== true || ((startup || startupCleanup?.requiresPreparationStop) && state.preparationStopped !== true)) throw new Error("Plugin Runner termination is unconfirmed");
|
||||
const confirmed = (state?: Awaited<ReturnType<typeof executeEnvironmentTask>>) => state?.kind === "status" && state.executionStopped === true &&
|
||||
(!(startup || startupCleanup?.requiresPreparationStop) || state.preparationStopped === true);
|
||||
let state = options.executionStopObserved === true && !startup && !startupCleanup
|
||||
? await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } }) : undefined;
|
||||
if (!confirmed(state)) {
|
||||
await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "stop" } });
|
||||
state = await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } });
|
||||
}
|
||||
if (!confirmed(state)) throw new Error("Plugin Runner termination is unconfirmed");
|
||||
await draining;
|
||||
await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "complete" } });
|
||||
return true;
|
||||
return "completed";
|
||||
} catch {
|
||||
await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed",
|
||||
failureReason: "plugin_runner_task_cleanup_unconfirmed", releasedAt: new Date(), updatedAt: new Date(),
|
||||
|
||||
Reference in new issue
Block a user