diff --git a/server/src/__tests__/heartbeat-workspace-busy.test.ts b/server/src/__tests__/heartbeat-workspace-busy.test.ts index f21701617da..8f055b90ce0 100644 --- a/server/src/__tests__/heartbeat-workspace-busy.test.ts +++ b/server/src/__tests__/heartbeat-workspace-busy.test.ts @@ -1303,6 +1303,18 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => { expect(retryRuns).toBe(0); }); + it("still defers behind a silent holder whose controller renews its lease", async () => { + const fixture = await seedWorkspaceFixture({ + holderActivityAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 60_000), + }); + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 10 * 60_000) }) + .where(eq(heartbeatRuns.id, fixture.holderRunId)); + const run = await heartbeat.invoke(fixture.agentId, "assignment", { issueId: fixture.issueId, wakeReason: "issue_assigned" }, "system"); + const finishedRun = await waitForRunToLeaveActiveStates(run!.id); + expect(finishedRun?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE); + expect(executedRunIds).not.toContain(run!.id); + }); + it("does not defer when the only holder has been silent past the staleness threshold", async () => { const fixture = await seedWorkspaceFixture({ holderActivityAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 60_000), diff --git a/server/src/modules/run-dispatch/adapters/workspace-busy-wait.ts b/server/src/modules/run-dispatch/adapters/workspace-busy-wait.ts index 95553a097f1..96b5548e87f 100644 --- a/server/src/modules/run-dispatch/adapters/workspace-busy-wait.ts +++ b/server/src/modules/run-dispatch/adapters/workspace-busy-wait.ts @@ -3,7 +3,7 @@ import { executionWorkspaces, heartbeatRuns, issues, type Db } from "@paperclipa import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { notifyScheduledRetryDispatch } from "../../../services/run-dispatch-notifications.js"; import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; -import { readRemoteWorkspaceWriterHold } from "../../../services/remote-workspace-writer-hold.js"; +import { readRemoteWorkspaceWriterHold, runOwnerFenceHeldSql } from "../../../services/remote-workspace-writer-hold.js"; import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WORKSPACE_BUSY_PARKED_RECHECK_MS, @@ -42,6 +42,7 @@ export async function parkWorkspaceBusyRetryInTx(tx: Db, run: HeartbeatRun, now: .select({ status: heartbeatRuns.status, lastActivityAt: sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt})`, + ownerFenceHeld: runOwnerFenceHeldSql(), }) .from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, wait.holderRunId))); @@ -50,7 +51,8 @@ export async function parkWorkspaceBusyRetryInTx(tx: Db, run: HeartbeatRun, now: // Reservation defers on that hold, so the wait stays parked through it too. const settlingSince = holder && holder.status !== "running" ? await settlingWriterActivityAt(tx, run, wait) : null; const holderStaleAt = holder?.status === "running" - ? new Date(holder.lastActivityAt).getTime() + WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS + ? Math.max(new Date(holder.lastActivityAt).getTime() + WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, + holder.ownerFenceHeld === true ? now.getTime() + WORKSPACE_BUSY_PARKED_RECHECK_MS : 0) : settlingSince ? settlingSince.getTime() + WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS : null; const holderLive = holderStaleAt !== null && holderStaleAt > now.getTime(); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 7e5a51ea8e5..4308d403e98 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,3 +1,4 @@ +import { runOwnerFenceHeldSql } from "../remote-workspace-writer-hold.js"; import { isInterruptedNativePreparation } from "../cancelled-native-startup.js"; import { isPreDispatchExecutionWaitVerified } from "../pre-dispatch-execution-wait.js"; import { notifyAttentionRetentionWork } from "../retention-work-notifications.js"; @@ -1440,7 +1441,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen ne(heartbeatRuns.id, input.excludeRunId), // Last observed activity: output beats start beats creation. A run // that started recently but has not written output yet is live. - sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) >= ${staleCutoff.toISOString()}::timestamptz`, + or(sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) >= ${staleCutoff.toISOString()}::timestamptz`, + runOwnerFenceHeldSql()), eq(issues.projectWorkspaceId, input.projectWorkspaceId), ne(sql`${issues.id}::text`, input.excludeIssueId), ...(input.honorIsolatedWorkspaceModes @@ -1480,7 +1482,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen (input.now ?? new Date()).getTime() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, ); - const lastActivityBefore = sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) < ${staleCutoff.toISOString()}::timestamptz`; + const lastActivityBefore = sql`(coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) < ${staleCutoff.toISOString()}::timestamptz and not ${runOwnerFenceHeldSql()})`; return await db .select({ runId: heartbeatRuns.id, diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index 802934c00df..487e4a09a5b 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -516,6 +516,20 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace await db.update(heartbeatRuns).set({ lastOutputAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 1_000) }) .where(eq(heartbeatRuns.id, writer.id)); + await db.insert(nativeRunFinalizations).values({ runId: writer.id, companyId: parent.companyId, issueId: parent.id, + phase: "observed", attempt: 1, controllerGeneration: 1, leaseOwner: "silent-native-controller", + leaseExpiresAt: new Date(Date.now() + 60_000) }); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ + constructor: WorkspaceBusyDeferral, code: "workspace_busy", holder: { runId: writer.id } }); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1_000) }) + .where(eq(nativeRunFinalizations.runId, writer.id)); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ status: 409 }); + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, writer.id)); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ + constructor: WorkspaceBusyDeferral, code: "workspace_busy", holder: { runId: writer.id } }); + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() - 1_000) }) + .where(eq(heartbeatRuns.id, writer.id)); await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ status: 409 }); // An ended writer whose save has not settled still holds the workspace: // the next run waits from the writer's end, within the same holder cutoff. diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index d42489f5649..f34e4cdd289 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -1056,7 +1056,8 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace } : {}), }).catch(error => { if (!(error instanceof RemoteWorkspaceWriterActiveError) - || error.writer.lastActivityAt.getTime() < Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS) throw error; + || (!error.writer.ownerFenceHeld + && error.writer.lastActivityAt.getTime() < Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS)) throw error; throw new WorkspaceBusyDeferral({ holder: error.writer, executionWorkspaceId: error.workspaceId, deferralAttempt: run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON ? run.scheduledRetryAttempt ?? 0 : 0, wasIssueAssignee: input.task.wasIssueAssignee, holderEnded: error.writer.ended === true }); diff --git a/server/src/services/remote-workspace-writer-hold.ts b/server/src/services/remote-workspace-writer-hold.ts index 6fa682ab7c3..721239f9777 100644 --- a/server/src/services/remote-workspace-writer-hold.ts +++ b/server/src/services/remote-workspace-writer-hold.ts @@ -1,5 +1,5 @@ import { and, eq, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, issueRecoveryActions, issues, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import type { RemoteWorkspaceAuthority } from "@paperclipai/shared"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canonicalExecutionUuid } from "./execution-identity.js"; @@ -12,8 +12,15 @@ export type RemoteWorkspaceWriter = { runId: string; agentId: string; issueId: string; issueIdentifier: string | null; lastActivityAt: Date; /** The writer run ended; its workspace save or sandbox release has not settled. */ ended?: boolean; + ownerFenceHeld?: boolean; }; +export function runOwnerFenceHeldSql() { + return sql`(coalesce(${heartbeatRuns.controllerLeaseExpiresAt} > now(), false) or exists ( + select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.leaseOwner} is not null and ${nativeRunFinalizations.leaseExpiresAt} > now()))`; +} + /** * Who still holds a remote workspace through its recorded writer run, other * than `runId`. A running writer holds it. An ended writer keeps holding it @@ -33,7 +40,7 @@ export async function readRemoteWorkspaceWriterHold(tx: Db, input: { errorCode: heartbeatRuns.errorCode, executionRecovery: sql`${heartbeatRuns.resultJson}->'executionRecovery'`, contextIssueId: sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`, lastOutputAt: heartbeatRuns.lastOutputAt, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, - finishedAt: heartbeatRuns.finishedAt, + finishedAt: heartbeatRuns.finishedAt, ownerFenceHeld: runOwnerFenceHeldSql(), }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, authority.writerRunId))); const writerIssueId = prior ? prior.nativeIssueId ?? prior.issueId ?? prior.contextIssueId : null; @@ -41,7 +48,7 @@ export async function readRemoteWorkspaceWriterHold(tx: Db, input: { .where(and(eq(issues.companyId, companyId), eq(issues.id, canonicalExecutionUuid(sql`${writerIssueId}::text`)))) : []; if (prior?.status === "running" && writerIssue) return { kind: "held", writer: { runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, - lastActivityAt: prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, + lastActivityAt: prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, ownerFenceHeld: prior.ownerFenceHeld === true, } }; const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, authority.writerRunId)));