diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index ae92ce4b2a0..65cf2148a2b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -498,6 +498,9 @@ A new run that finds such a recently ended writer waits for the workspace. It do not fail setup. The wait uses the same one-hour holder cutoff as a quiet live writer, measured from the time the writer ended. After the cutoff, preparation fails with the unfinished-writer error and leaves the held files unchanged. +This wait is also one parked retry, for example after a reassignment stops the +writer. It is re-checked every minute while the ended writer's hold settles, is +admitted once the hold releases, and stops parking at the same cutoff. An instance admission drain keeps accepted new turns queued and displays **Waiting for maintenance**. Ending the audited drain requests admission for diff --git a/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts b/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts index 05c13b25a85..b0d106027f1 100644 --- a/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts +++ b/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts @@ -163,6 +163,56 @@ describe.skipIf(!support.supported)("inherited remote workspace heartbeat wait", expect(readRemoteWorkspaceAuthority(f.current.metadata)).toEqual(f.authority); }); + async function workspaceBusyRunIds(companyId: string) { + return (await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))) + .filter(row => row.status === "cancelled" && row.errorCode === "workspace_busy").map(row => row.id); + } + + it("parks one waiter while an ended writer's save settles, then admits it once the hold releases", async () => { + // A task reassigned mid-run: the stopped writer ended, but its save has + // not settled. The waiter must stay one parked retry, not a new run per check. + const f = await terminalWriterFixture(new Date()); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", contextSnapshot: { workspaceBusyWait: { + key: `execution:${f.workspace.id}`, holderRunId: f.writer.id } } }); + let checkAt = new Date(retry.scheduledRetryAt!.getTime() + 1_000); + for (let check = 0; check < 3; check += 1) { + expect((await heartbeat.promoteDueScheduledRetries(checkAt)).runIds).not.toContain(retry.id); + const parked = await heartbeat.getRun(retry.id); + expect(parked?.status).toBe("scheduled_retry"); + const recheckAt = new Date(parked!.scheduledRetryAt!); + expect(recheckAt.getTime()).toBeGreaterThan(checkAt.getTime()); + // Bounded by the holder cutoff measured from the writer's end. + expect(recheckAt.getTime()).toBeLessThanOrEqual(f.writer.finishedAt!.getTime() + 60 * 60_000 + 1_000); + checkAt = new Date(recheckAt.getTime() + 1_000); + } + expect(await workspaceBusyRunIds(f.run.companyId)).toEqual([f.run.id]); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, retry.id))).toEqual([]); + + // The save settles: the writer hold is released. + await db.update(executionWorkspaces).set({ metadata: { remoteWorkspace: { ...f.authority, writerRunId: null, dirty: false } } }) + .where(eq(executionWorkspaces.id, f.workspace.id)); + expect((await heartbeat.promoteDueScheduledRetries(checkAt)).runIds).toContain(retry.id); + await heartbeat.cancelRun(retry.id); + await heartbeat.drainActiveRunExecutions(); + expect(await workspaceBusyRunIds(f.run.companyId)).toEqual([f.run.id]); + }); + + it("stops parking behind an ended writer once its unsettled save reaches the holder cutoff", async () => { + const f = await terminalWriterFixture(new Date(Date.now() - 58 * 60_000)); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id)); + expect(retry?.status).toBe("scheduled_retry"); + const cutoff = f.writer.finishedAt!.getTime() + 60 * 60_000; + if (retry.scheduledRetryAt!.getTime() < cutoff) { + expect((await heartbeat.promoteDueScheduledRetries(new Date(retry.scheduledRetryAt!.getTime() + 1))).runIds) + .not.toContain(retry.id); + } + expect((await heartbeat.promoteDueScheduledRetries(new Date(cutoff + 2_000))).runIds).toContain(retry.id); + await heartbeat.cancelRun(retry.id); + await heartbeat.drainActiveRunExecutions(); + }); + it("stops waiting for an ended writer whose save stayed unsettled past the holder cutoff", async () => { const f = await terminalWriterFixture(new Date(Date.now() - 2 * 60 * 60_000)); expect(f.run).toMatchObject({ status: "failed", error: "The workspace still has an unfinished writer or save operation." }); 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 c62b7153208..95553a097f1 100644 --- a/server/src/modules/run-dispatch/adapters/workspace-busy-wait.ts +++ b/server/src/modules/run-dispatch/adapters/workspace-busy-wait.ts @@ -1,11 +1,14 @@ import { and, eq, gt, inArray, ne, sql } from "drizzle-orm"; -import { heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { executionWorkspaces, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +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 { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WORKSPACE_BUSY_PARKED_RECHECK_MS, WORKSPACE_BUSY_RETRY_REASON, + WORKSPACE_BUSY_SETTLING_RECHECK_MS, WORKSPACE_BUSY_WAIT_KEY, readWorkspaceBusyWait, type WorkspaceBusyWait, @@ -42,9 +45,13 @@ export async function parkWorkspaceBusyRetryInTx(tx: Db, run: HeartbeatRun, now: }) .from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, wait.holderRunId))); + // A remote writer that ended still holds its workspace until its save and + // sandbox release settle (a run stopped by reassignment, for example). + // 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 - : null; + : settlingSince ? settlingSince.getTime() + WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS : null; const holderLive = holderStaleAt !== null && holderStaleAt > now.getTime(); // Arrival order among waiters for one workspace: an earlier waiter that is @@ -68,7 +75,7 @@ export async function parkWorkspaceBusyRetryInTx(tx: Db, run: HeartbeatRun, now: if (!holderLive && !earlierWaiter) return false; const recheckAt = new Date(Math.min( - now.getTime() + WORKSPACE_BUSY_PARKED_RECHECK_MS, + now.getTime() + (holderLive && settlingSince ? WORKSPACE_BUSY_SETTLING_RECHECK_MS : WORKSPACE_BUSY_PARKED_RECHECK_MS), holderLive ? holderStaleAt! + 1_000 : Number.POSITIVE_INFINITY, )); await notifyScheduledRetryCancellation(tx, { companyId: run.companyId, runId: run.id }); @@ -88,6 +95,25 @@ export async function parkWorkspaceBusyRetryInTx(tx: Db, run: HeartbeatRun, now: return Boolean(parked); } +/** + * When the recorded holder is an ended remote writer that still holds the + * waited execution workspace, the time its hold-cutoff clock started (the + * same value reservation measures). Null when that hold has settled or moved. + */ +async function settlingWriterActivityAt(tx: Db, run: HeartbeatRun, wait: WorkspaceBusyWait): Promise { + if (!wait.key.startsWith("execution:")) return null; + const [workspace] = await tx + .select({ metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(and(eq(executionWorkspaces.companyId, run.companyId), eq(executionWorkspaces.id, wait.key.slice("execution:".length)))); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (!authority || authority.writerRunId !== wait.holderRunId) return null; + const hold = await readRemoteWorkspaceWriterHold(tx, { companyId: run.companyId, authority }); + return hold.kind === "held" && hold.writer.ended && hold.writer.runId === wait.holderRunId + ? new Date(hold.writer.lastActivityAt) + : null; +} + async function taskStillUsesWaitedWorkspace(tx: Db, run: HeartbeatRun, wait: WorkspaceBusyWait): Promise { const context = run.contextSnapshot ?? {}; const issueId = typeof context.issueId === "string" ? context.issueId : null; diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 4790660945a..a941139bc80 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -36,6 +36,13 @@ export const WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS = 60 * 60 * 1000; // Fallback re-check for a parked workspace wait. The holder ending wakes the // waiter directly; this only bounds a missed wake. export const WORKSPACE_BUSY_PARKED_RECHECK_MS = 5 * 60 * 1000; +// Re-check for a wait behind an ended remote writer whose save or sandbox +// release has not settled. Settling has no single completion point to wake +// from, so the parked run re-checks on this cadence. Each check creates no run. +// It reads the workspace, the writer run, its task, and the writer's leases +// (one row per sandbox lease of that run, so a small set) in the promotion +// transaction. The checks last only while the wait exists, up to the cutoff. +export const WORKSPACE_BUSY_SETTLING_RECHECK_MS = 60 * 1000; export const WORKSPACE_BUSY_WAIT_KEY = "workspaceBusyWait"; /** What a workspace-busy retry is waiting behind, recorded at deferral. */ diff --git a/server/src/services/remote-workspace-writer-hold.ts b/server/src/services/remote-workspace-writer-hold.ts new file mode 100644 index 00000000000..fa4c69bd4a1 --- /dev/null +++ b/server/src/services/remote-workspace-writer-hold.ts @@ -0,0 +1,64 @@ +import { and, eq, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, 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"; +import { REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE } from "./remote-workspace-transport.js"; + +const object = (value: unknown): Record => + typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record : {}; + +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; +}; + +/** + * 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 + * until its save and sandbox release settle. `unattributed` is a hold whose + * writer cannot be named (no run or task); `free` means another run may + * become the writer. Workspace reservation and the parked workspace wait read + * this one rule, so a wait never ends before reservation would succeed. + */ +export async function readRemoteWorkspaceWriterHold(tx: Db, input: { + companyId: string; authority: RemoteWorkspaceAuthority; runId?: string | null; +}): Promise<{ kind: "free" } | { kind: "unattributed" } | { kind: "held"; writer: RemoteWorkspaceWriter }> { + const { authority, companyId } = input; + if (!authority.writerRunId || authority.writerRunId === input.runId) return { kind: "free" }; + const [prior] = await tx.select({ status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, + issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, + runtimeMode: heartbeatRuns.runtimeMode, nativeSessionId: heartbeatRuns.nativeSessionId, runnerInstanceId: heartbeatRuns.runnerInstanceId, + 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, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, authority.writerRunId))); + const writerIssueId = prior ? prior.nativeIssueId ?? prior.issueId ?? prior.contextIssueId : null; + const [writerIssue] = writerIssueId ? await tx.select({ id: issues.id, identifier: issues.identifier }).from(issues) + .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, + } }; + const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, companyId), + eq(environmentLeases.heartbeatRunId, authority.writerRunId))); + const bootstrap = object(prior?.executionRecovery); + const pristinePreparation = !authority.location && !authority.snapshotId && prior?.status === "failed" + && prior.runtimeMode === "legacy" && prior.nativeSessionId === null && prior.runnerInstanceId === null + && prior.errorCode === REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE + && bootstrap.kind === "bootstrap" && bootstrap.providerWorkStarted === false; + const unusedReservation = !authority.dirty && prior && ["failed", "cancelled", "interrupted", "timed_out", "succeeded"].includes(prior.status) + && leases.every(lease => hasRemoteTerminationReceipt(lease) || hasRecoveredCleanupStopReceipt(lease) || (pristinePreparation + && lease.releasedAt && lease.cleanupStatus === "success" && ["released", "expired", "failed"].includes(lease.status))); + if (unusedReservation) return { kind: "free" }; + // An ended writer still holds the workspace until its save and sandbox + // release settle, within the same bounded cutoff as a live writer. + if (prior && writerIssue) return { kind: "held", writer: { + runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, + lastActivityAt: prior.finishedAt ?? prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, ended: true, + } }; + return { kind: "unattributed" }; +} diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index 835c67ae8c3..0b0494e5055 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -22,12 +22,11 @@ import { logger } from "../middleware/logger.js"; import { createGitRemoteAuthProvider } from "./git-credentials.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { assertExecutionWorkspaceRecordAccess, assertTaskWorkspaceExecutionAccess } from "./task-workspace-access.js"; -import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; -import { canonicalExecutionUuid } from "./execution-identity.js"; +import { readRemoteWorkspaceWriterHold, type RemoteWorkspaceWriter } from "./remote-workspace-writer-hold.js"; import { persistActivity } from "./activity-log.js"; import { captureRemoteWorkspaceArchive, cleanupRemoteWorkspaceArchive, prepareRemoteWorkspaceDirectory, prepareRemoteRepository, remoteWorkspaceRepositoryPath, - REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE, RemoteWorkspaceOperationTimeoutError, RemoteWorkspacePreparationTimeoutError, remoteWorkspaceFailureCause, + RemoteWorkspaceOperationTimeoutError, RemoteWorkspacePreparationTimeoutError, remoteWorkspaceFailureCause, requireRemoteWorkspaceTarget, restoreRemoteWorkspaceDirectory, RemoteWorkspaceRestoreError, type RemoteWorkspaceRestoreStep, } from "./remote-workspace-transport.js"; import { @@ -153,11 +152,7 @@ async function lockWorkspaceRunParents(tx: Transaction, companyId: string, runId export class RemoteWorkspaceWriterActiveError extends HttpError { constructor( readonly workspaceId: string, - readonly writer: { - 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; - }, + readonly writer: RemoteWorkspaceWriter, ) { super(409, "The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); } @@ -215,44 +210,9 @@ export async function reserveRemoteWorkspace(input: { sql`${workspaceOperations.heartbeatRunId} is distinct from ${input.runId}::uuid`, )).limit(1); if (uncertain) throw conflict("The prior provisioning operation needs verification before this workspace can run again.", { code: "remote_workspace_provision_unverified" }); - if (authority.writerRunId && authority.writerRunId !== input.runId) { - const [prior] = await tx.select({ status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, - issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, - runtimeMode: heartbeatRuns.runtimeMode, nativeSessionId: heartbeatRuns.nativeSessionId, runnerInstanceId: heartbeatRuns.runnerInstanceId, - 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, - }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, authority.writerRunId))); - const writerIssueId = prior ? prior.nativeIssueId ?? prior.issueId ?? prior.contextIssueId : null; - const [writerIssue] = writerIssueId ? await tx.select({ id: issues.id, identifier: issues.identifier }).from(issues) - .where(and(eq(issues.companyId, input.companyId), eq(issues.id, canonicalExecutionUuid(sql`${writerIssueId}::text`)))) : []; - if (prior?.status === "running" && writerIssue) throw new RemoteWorkspaceWriterActiveError(existing.id, { - runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, - lastActivityAt: prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, - }); - const leases = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), - eq(environmentLeases.heartbeatRunId, authority.writerRunId))); - const bootstrap = object(prior?.executionRecovery); - const pristinePreparation = !authority.location && !authority.snapshotId && prior?.status === "failed" - && prior.runtimeMode === "legacy" && prior.nativeSessionId === null && prior.runnerInstanceId === null - && prior.errorCode === REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE - && bootstrap.kind === "bootstrap" && bootstrap.providerWorkStarted === false; - const unusedReservation = !authority.dirty && prior && ["failed", "cancelled", "interrupted", "timed_out", "succeeded"].includes(prior.status) - && leases.every(lease => hasRemoteTerminationReceipt(lease) || hasRecoveredCleanupStopReceipt(lease) || (pristinePreparation - && lease.releasedAt && lease.cleanupStatus === "success" && ["released", "expired", "failed"].includes(lease.status))); - if (!unusedReservation) { - // An ended writer still holds the workspace until its save and sandbox - // release settle. Report it as the holder so preparation waits within - // the same bounded cutoff as a live writer instead of failing setup. - if (prior && writerIssue) throw new RemoteWorkspaceWriterActiveError(existing.id, { - runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, - lastActivityAt: prior.finishedAt ?? prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, ended: true, - }); - throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); - } - } + const hold = await readRemoteWorkspaceWriterHold(tx as unknown as Db, { companyId: input.companyId, authority, runId: input.runId }); + if (hold.kind === "held") throw new RemoteWorkspaceWriterActiveError(existing.id, hold.writer); + if (hold.kind === "unattributed") throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); if (authority.writerRunId === input.runId) return existing; const next: RemoteWorkspaceAuthority = { ...authority, source, epoch: authority.epoch + 1, writerRunId: input.runId }; const [saved] = await tx.update(executionWorkspaces).set({ metadata: { ...existing.metadata, remoteWorkspace: next,