mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 08:28:48 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
cfa4e96ad4
commit
b6fdda950f
3 files changed
+27
-8
No files matched your search
@@ -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.
|
||||
@@ -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();
|
||||
});
|
||||
|
||||
@@ -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<void> {
|
||||
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<boolean>`${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);
|
||||
|
||||
Reference in new issue
Block a user