From b6fdda950faf522ccc64592876e993d7bc05d94f Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 02:22:05 -0500 Subject: [PATCH] Wake pending shutdown peers after scoped resource release Include exact pending shutdown markers in physical-resource peer invalidation, preserving active orphan notifications and canonical cleanup authority. Cover different lease/run scoped release, rollback, finite deadlines and quiet refusal states with real registered workers. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-live-recovery-sleep.md | 2 ++ server/src/__tests__/live-recovery-work.test.ts | 16 ++++++++++++---- .../environment-cleanup-notifications.ts | 17 +++++++++++++---- 3 files changed, 27 insertions(+), 8 deletions(-) diff --git a/doc/plans/2026-10-30-live-recovery-sleep.md b/doc/plans/2026-10-30-live-recovery-sleep.md index fac3d192b6c..218903f3388 100644 --- a/doc/plans/2026-10-30-live-recovery-sleep.md +++ b/doc/plans/2026-10-30-live-recovery-sleep.md @@ -59,3 +59,5 @@ A native question marker can commit before execution changes the run to failed/r 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. + +A scoped environment release also invalidates pending shutdown markers on peers of the exact provider allocation, including released stop-only sources. Peer lookup is fenced by the changed company/run or lease and physical provider/resource identity; each pending marker receives its own company/lease shutdown hint. The marker must retain the canonical schema, pending state and matching run identity. Active orphan hints remain limited to active leases. Registered-worker regressions park for twenty minutes, roll back a different lease/run-scoped competitor release, then prove that its ordinary committed release alone records exactly one workspace recovery action and expires the source marker. No company-wide wake or direct target hint is supplied by the test. diff --git a/server/src/__tests__/live-recovery-work.test.ts b/server/src/__tests__/live-recovery-work.test.ts index fefe4bb0b2a..cb495a3df59 100644 --- a/server/src/__tests__/live-recovery-work.test.ts +++ b/server/src/__tests__/live-recovery-work.test.ts @@ -412,9 +412,14 @@ describe("residual recovery production matrix", () => { } 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, + it.each(["competing allocation lease", "competing allocation run", "cleanup in flight"] as const)("parks unchanged shutdown refusal for %s until its owning commit", async gate => { + const competingGate = gate !== "cleanup in flight"; + const f = await shutdownSource(competingGate); + const [otherRun] = gate === "competing allocation run" ? await db.insert(heartbeatRuns).values({ + companyId: f.companyId, agentId: f.agent.id, status: "failed", runtimeMode: "legacy", finishedAt: new Date(), + }).returning() : []; + const [competing] = competingGate ? await db.insert(environmentLeases).values({ companyId: f.companyId, + heartbeatRunId: otherRun?.id, status: "pending_cleanup", provider: f.lease.provider, providerLeaseId: f.lease.providerLeaseId, acquiredAt: new Date(f.lease.acquiredAt.getTime() - 1000) }).returning() : []; await start(); @@ -423,7 +428,8 @@ describe("residual recovery production matrix", () => { 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 }); + ).where(eq(environmentLeases.id, competing?.id ?? f.lease.id)), { companyId: f.companyId, + ...(otherRun ? { runId: otherRun.id } : { leaseId: competing?.id ?? f.lease.id }) }); await expect(db.transaction(async tx => { await releaseGate(tx as unknown as typeof db); throw new Error("roll back shutdown gate release"); @@ -440,6 +446,8 @@ describe("residual recovery production matrix", () => { expect(current.metadata?.workspaceRestoreRecovery).toMatchObject({ runId: f.run.id }); expect(current.providerLeaseId).toBe(f.lease.providerLeaseId); }, { timeout: 15_000 }); + expect(await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId), + eq(issueRecoveryActions.fingerprint, `legacy-execution:${f.run.id}`)))).toHaveLength(1); await reconcileDatabaseWork(db, "legacy-shutdown-settlement"); await Promise.allSettled([...passes]); await immediate(); await quiet(); }); diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index ce0ded6fcc0..76ba2521ea1 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -1,4 +1,4 @@ -import { and, eq, sql } from "drizzle-orm"; +import { and, eq, or, sql } from "drizzle-orm"; import { environmentLeases } from "@paperclipai/db"; import { notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScope } from "./native-environment-settlement-notifications.js"; @@ -10,13 +10,22 @@ export type EnvironmentCleanupWorkScope = NativeEnvironmentSettlementScope & { l export async function notifyEnvironmentCleanup(transaction: object, scope?: EnvironmentCleanupWorkScope): Promise { await notifyLiveResourceRecovery(transaction, scope); if (scope && (scope.runId || scope.leaseId) && typeof (transaction as Db).select === "function") { + // A stopped/released allocation can still owe shutdown workspace preservation. + // Its peer's owning release must reactivate that exact marker as well. + const pendingShutdown = sql`${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,schema}' = 'paperclip.legacy-shutdown-workspace-settlement.v1' + and ${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,state}' = 'pending' + and ${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,runId}' = ${environmentLeases.heartbeatRunId}::text`; const peers = await (transaction as Db).select({ id: environmentLeases.id, companyId: environmentLeases.companyId, - runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and(eq(environmentLeases.status, "active"), + status: environmentLeases.status, pendingShutdown }).from(environmentLeases).where(and( + or(eq(environmentLeases.status, "active"), pendingShutdown), sql`exists (select 1 from environment_leases changed where changed.company_id = ${scope.companyId} and ${scope.leaseId ? sql`changed.id = ${scope.leaseId}` : sql`changed.heartbeat_run_id = ${scope.runId}`} and changed.provider_lease_id is not null and changed.provider = ${environmentLeases.provider} and changed.provider_lease_id = ${environmentLeases.providerLeaseId})`)); - for (const peer of peers) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.orphanActiveLeases, - JSON.stringify({ companyId: peer.companyId, leaseId: peer.id })); + for (const peer of peers) { + const hint = JSON.stringify({ companyId: peer.companyId, leaseId: peer.id }); + if (peer.status === "active") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.orphanActiveLeases, hint); + if (peer.pendingShutdown) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.legacyShutdownSettlement, hint); + } } await notifyNativeEnvironmentSettlement(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup);