mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 08:28:48 +01:00
Park unchanged shutdown cleanup refusals without self-waking
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
554341a79a
commit
a4fe24a34a
3 files changed
+89
-6
No files matched your search
@@ -55,3 +55,7 @@ The actual private-pool control prewarms the connection and requires exactly fiv
|
||||
Agent settlement retains a bounded retry only after its actual `FOR UPDATE SKIP LOCKED` attempt misses an existing candidate. Releasing a transaction that only held the lock emits no eligibility hint; the retained entry still settles exactly once and becomes quiet.
|
||||
|
||||
A native question marker can commit before execution changes the run to failed/retryable failure. Both reconstruction and consumption select this exact unfinished class. A Stop may also advance the coordinator to `terminal_failure/native_retry_cancelled` before the final run-status commit fails. The owning pending predicate includes only the existing exactly bound, audited pending/acknowledged cancellation receipt; cancellation candidate discovery uses `nativeRetryCancellationEligible` again. Existing locked native dispatch and the unchanged acknowledged-receipt final CAS still authorize all effects. A one-shot PostgreSQL failure proves this continuation completes with the same intent/audit receipt and no provider execution, while marked failed history with misbound receipts remains unchanged and quiet.
|
||||
|
||||
Shutdown workspace cleanup publishes resource-work intent only after its competing-allocation and in-flight-cleanup refusal checks, immediately before the existing write. An unchanged allocation conflict or cleanup owner without a finite expiry parks until an owning resource write; inspecting that gate cannot wake the worker again. A finite pendingCleanupLeaseExpiresAtMs remains a concrete obligation: the shutdown owner returns that exact stored deadline through its existing callback and re-reads a moved expiry under the original issue/run/lease lock order.
|
||||
|
||||
Actual registered-owner controls use a confirmed stop-only released lease and a pending ordinary-cleanup lease. Both refusal states retain the pending shutdown marker while twenty virtual minutes produce zero driver SQL. A rolled-back owning gate release preserves that state; a committed release reactivates the original preservation path, retains the allocation and recovery metadata, and expires the marker without certifying file export. Separate controls cross the original and moved cleanup-claim deadlines and the existing active shutdown-marker deadline. Transaction uncertainty is joined through the queue-specific work-fence reconciliation before the idle measurement; no timeout or recurring recovery scan is added.
|
||||
@@ -7,7 +7,7 @@ import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, closeLivenessDb, closeCheckpointDb,
|
||||
heartbeatRuns, heartbeatRunEvents, nativeRunFinalizations, environmentLeases, issueThreadInteractions, applyPendingMigrations,
|
||||
issueRecoveryActions, issues, activityLog, agentWakeupRequests } from "@paperclipai/db";
|
||||
issueRecoveryActions, issues, activityLog, agentWakeupRequests, reconcileDatabaseWork } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
@@ -16,6 +16,7 @@ import { notifyLiveRunRecovery, notifyLiveResourceRecovery, notifyStrandedAgentS
|
||||
import { notifyExecutionRunWork } from "../services/execution-control-notifications.js";
|
||||
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js";
|
||||
import { prepareSandboxStopAndRetain, settleStopOnlyCleanup, hasConfirmedSandboxStopAndRetain } from "../services/sandbox-stop-and-retain.js";
|
||||
import { beginLegacyShutdownWorkspaceSettlement } from "../services/legacy-shutdown-workspace-settlement.js";
|
||||
import { renewLegacyControllerLease, legacyControllerBootId } from "../services/legacy-controller-lease.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
@@ -388,6 +389,79 @@ describe("residual recovery production matrix", () => {
|
||||
await vi.waitFor(async () => expect((await getLease(orphan.id)).status).toBe("pending_cleanup"), { timeout: 40_000 });
|
||||
await quiet();
|
||||
});
|
||||
async function shutdownSource(released: boolean, cleanupExpiry?: number) {
|
||||
const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id,
|
||||
status: "failed", runtimeMode: "legacy", contextSnapshot: { issueId: f.issue.id }, finishedAt: old }).returning();
|
||||
let [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id,
|
||||
issueId: f.issue.id, status: "active", provider: "fixture", providerLeaseId: randomUUID(), acquiredAt: old, updatedAt: old,
|
||||
metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "shutdown-fixture" } }).returning();
|
||||
await beginLegacyShutdownWorkspaceSettlement(db, run, new Date(Date.now() - 1000));
|
||||
if (released) {
|
||||
const prepared = await prepareSandboxStopAndRetain(db, await getLease(lease.id));
|
||||
expect(prepared).toBeTruthy();
|
||||
const settled = await settleStopOnlyCleanup(db, prepared!, { attemptId: String(prepared!.metadata?.pendingCleanupAttemptId),
|
||||
receipt: { providerLeaseId: lease.providerLeaseId, state: "stopped" } });
|
||||
expect(settled).toBeTruthy(); lease = settled!;
|
||||
expect(hasConfirmedSandboxStopAndRetain(lease)).toBe(true);
|
||||
} else {
|
||||
[lease] = await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed",
|
||||
metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ pendingCleanupInFlight: true,
|
||||
...(cleanupExpiry === undefined ? {} : { pendingCleanupLeaseExpiresAtMs: cleanupExpiry }) })}::jsonb`,
|
||||
}).where(eq(environmentLeases.id, lease.id)).returning();
|
||||
}
|
||||
return { ...f, run, lease };
|
||||
}
|
||||
it.each(["competing allocation", "cleanup in flight"] as const)("parks unchanged shutdown refusal for %s until its owning commit", async gate => {
|
||||
const f = await shutdownSource(gate === "competing allocation");
|
||||
const [competing] = gate === "competing allocation" ? await db.insert(environmentLeases).values({ companyId: f.companyId,
|
||||
status: "pending_cleanup", provider: f.lease.provider, providerLeaseId: f.lease.providerLeaseId,
|
||||
acquiredAt: new Date(f.lease.acquiredAt.getTime() - 1000) }).returning() : [];
|
||||
await start();
|
||||
expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" });
|
||||
await quiet();
|
||||
const releaseGate = async (database: typeof db) => writeEnvironmentCleanup(database, tx => tx.update(environmentLeases).set(
|
||||
competing ? { status: "released", releasedAt: new Date() }
|
||||
: { metadata: sql`${environmentLeases.metadata} || '{"pendingCleanupInFlight":false}'::jsonb` },
|
||||
).where(eq(environmentLeases.id, competing?.id ?? f.lease.id)), { companyId: f.companyId });
|
||||
await expect(db.transaction(async tx => {
|
||||
await releaseGate(tx as unknown as typeof db);
|
||||
throw new Error("roll back shutdown gate release");
|
||||
})).rejects.toThrow("roll back shutdown gate release");
|
||||
await reconcileDatabaseWork(db, "legacy-shutdown-settlement"); await Promise.allSettled([...passes]); await immediate();
|
||||
expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" });
|
||||
await quiet();
|
||||
await releaseGate(db);
|
||||
await vi.waitFor(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
const current = await getLease(f.lease.id);
|
||||
expect(current.metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "expired" });
|
||||
expect(current.leasePolicy).toBe("retain_on_failure");
|
||||
expect(current.metadata?.workspaceRestoreRecovery).toMatchObject({ runId: f.run.id });
|
||||
expect(current.providerLeaseId).toBe(f.lease.providerLeaseId);
|
||||
}, { timeout: 15_000 });
|
||||
await reconcileDatabaseWork(db, "legacy-shutdown-settlement"); await Promise.allSettled([...passes]); await immediate();
|
||||
await quiet();
|
||||
});
|
||||
it("retains the actual cleanup claim deadline after shutdown expiry and honors a moved expiry", async () => {
|
||||
const initial = Date.now() + 1500, f = await shutdownSource(false, initial);
|
||||
await start();
|
||||
expect(deadlines.get("legacy-shutdown-settlement")!()).toBe(initial);
|
||||
const moved = Date.now() + 4000;
|
||||
await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({
|
||||
metadata: sql`jsonb_set(${environmentLeases.metadata}, '{pendingCleanupLeaseExpiresAtMs}', ${JSON.stringify(moved)}::jsonb)`,
|
||||
}).where(eq(environmentLeases.id, f.lease.id)), { companyId: f.companyId, leaseId: f.lease.id });
|
||||
await advance(initial - Date.now());
|
||||
expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" });
|
||||
expect(deadlines.get("legacy-shutdown-settlement")!()).toBe(moved);
|
||||
await advance(moved - Date.now() - 1);
|
||||
expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" });
|
||||
await advance(1);
|
||||
await vi.waitFor(async () => expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement)
|
||||
.toMatchObject({ state: "expired" }));
|
||||
expect((await getLease(f.lease.id)).metadata?.workspaceRestoreRecovery).toMatchObject({ runId: f.run.id });
|
||||
await quiet();
|
||||
});
|
||||
it("expires the exact shutdown settlement marker into preserved recovery instead of inferring file export", async () => {
|
||||
const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id,
|
||||
|
||||
@@ -75,7 +75,7 @@ export async function legacyShutdownWorkspaceResourceProtected(db: Db,
|
||||
}
|
||||
|
||||
/** Released stop receipts still need reconciliation if export never settled. */
|
||||
export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date(), scope: { companyId?: string; leaseId?: string } = {}) {
|
||||
export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date(), scope: { companyId?: string; leaseId?: string; onDeferredDeadline?: (at: number) => void } = {}) {
|
||||
const rows = await db.select().from(environmentLeases).where(and(
|
||||
scope.companyId ? eq(environmentLeases.companyId, scope.companyId) : undefined,
|
||||
scope.leaseId ? eq(environmentLeases.id, scope.leaseId) : undefined,
|
||||
@@ -83,7 +83,7 @@ export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new
|
||||
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`,
|
||||
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt' <= ${now.toISOString()}`,
|
||||
)).orderBy(sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt'`).limit(50);
|
||||
for (const lease of rows) await allowLegacyShutdownWorkspaceCleanup(db, lease, now);
|
||||
for (const lease of rows) await allowLegacyShutdownWorkspaceCleanup(db, lease, now, { onDeferredDeadline: scope.onDeferredDeadline });
|
||||
}
|
||||
|
||||
/** Commit the fence while the run is still live, before exposing interruption. */
|
||||
@@ -115,7 +115,7 @@ export async function finishLegacyShutdownWorkspaceSettlement(db: Db, run: Run,
|
||||
|
||||
/** A restart or another cleanup worker must never infer saved files from timeout. */
|
||||
export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, now = new Date(),
|
||||
options: { ownerStopOnly?: boolean } = {}) {
|
||||
options: { ownerStopOnly?: boolean; onDeferredDeadline?: (at: number) => void } = {}) {
|
||||
if (!pendingLegacyShutdownWorkspaceSettlement(lease)) return true;
|
||||
return db.transaction(async tx => {
|
||||
const [hint] = await tx.select().from(environmentLeases).where(and(
|
||||
@@ -148,7 +148,6 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease,
|
||||
if (!Number.isFinite(deadline) || deadline > now.getTime()) return false;
|
||||
// A killed container cannot certify export. Preserve source and hold in
|
||||
// this same transaction before any worker may perform provider cleanup.
|
||||
await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
|
||||
if (current.status !== "active") {
|
||||
const { hasConfirmedSandboxStopAndRetain, readStopOnlyCleanup } = await import("./sandbox-stop-and-retain.js");
|
||||
if (!(current.status === "released" && hasConfirmedSandboxStopAndRetain(current))
|
||||
@@ -165,12 +164,17 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease,
|
||||
if (current.status === "pending_cleanup" && !readStopOnlyCleanup(current)) {
|
||||
const expiresAt = current.metadata?.pendingCleanupLeaseExpiresAtMs;
|
||||
if (current.metadata?.pendingCleanupInFlight === true
|
||||
&& (typeof expiresAt !== "number" || !Number.isFinite(expiresAt) || expiresAt > now.getTime())) return false;
|
||||
&& (typeof expiresAt !== "number" || !Number.isFinite(expiresAt) || expiresAt > now.getTime())) {
|
||||
if (typeof expiresAt === "number" && Number.isFinite(expiresAt)) options.onDeferredDeadline?.(expiresAt);
|
||||
return false;
|
||||
}
|
||||
// A failed ordinary release can race the shutdown marker. Re-enter the
|
||||
// recorder's active-source path within this claim transaction so it
|
||||
// replaces destructive retry with an exact stop-only recovery intent.
|
||||
await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
|
||||
await tx.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, current.id));
|
||||
} else {
|
||||
await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
|
||||
await tx.update(environmentLeases).set({ leasePolicy: "retain_on_failure",
|
||||
metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ workspaceRestoreRecovery: {
|
||||
schema: LEGACY_WORKSPACE_RECOVERY_SCHEMA, runId: run.id,
|
||||
@@ -178,6 +182,7 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease,
|
||||
}).where(eq(environmentLeases.id, current.id));
|
||||
}
|
||||
}
|
||||
if (current.status === "active") await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
|
||||
const { terminalizeLegacyExecution } = await import("./legacy-execution-recovery.js");
|
||||
const preserved = await terminalizeLegacyExecution({ db: tx as unknown as Db, run, status: run.status,
|
||||
recordRestoreFailureOnly: true, patch: { resultJson: { workspaceRestoreFailure: "restore_failed" } } });
|
||||
|
||||
Reference in new issue
Block a user