RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: make checkout run validation atomic 

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Chris 2026-09-11 17:36:17 -04:00
parent 79ffbbaae0
commit bac7f2a0e5
3 changed files with 326 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/src/__tests__/issue-stale-execution-lock-routes.test.ts
View File

@ -500,6 +500,10 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => {
runtimeConfig: {},
permissions: {},
});
await db
.update(heartbeatRuns)
.set({ agentId: otherAgentId })
.where(eq(heartbeatRuns.id, currentRunId));
await db.insert(issues).values({
id: issueId,
companyId,

157
server/src/__tests__/issues-service.test.ts
View File

@ -6146,6 +6146,96 @@ describeEmbeddedPostgres("issueService.clearExecutionRunIfTerminal", () => {
});
});
it("rejects checkout when the owning run becomes terminal before the issue mutation", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const checkoutRunId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(heartbeatRuns).values({
id: checkoutRunId,
companyId,
agentId,
status: "running",
invocationSource: "manual",
startedAt: new Date("2026-08-26T11:16:18.000Z"),
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Checkout racing run completion",
status: "todo",
priority: "critical",
assigneeAgentId: agentId,
});
const terminalWriteReady = deferred<void>();
const allowTerminalCommit = deferred<void>();
const terminalWrite = db.transaction(async (tx) => {
await tx
.update(heartbeatRuns)
.set({
status: "succeeded",
finishedAt: new Date("2026-08-26T11:16:19.000Z"),
})
.where(eq(heartbeatRuns.id, checkoutRunId));
terminalWriteReady.resolve();
await allowTerminalCommit.promise;
});
await terminalWriteReady.promise;
const checkout = svc.checkout(
issueId,
agentId,
["todo"],
checkoutRunId,
);
const checkoutAssertion = expect(checkout).rejects.toMatchObject({
status: 409,
details: {
code: "issue_checkout_run_not_live",
checkoutRunId,
runStatus: "succeeded",
},
});
await new Promise<void>((resolve) => setImmediate(resolve));
allowTerminalCommit.resolve();
await terminalWrite;
await checkoutAssertion;
const row = await db
.select({
status: issues.status,
checkoutRunId: issues.checkoutRunId,
executionRunId: issues.executionRunId,
})
.from(issues)
.where(eq(issues.id, issueId))
.then((rows) => rows[0]);
expect(row).toEqual({
status: "todo",
checkoutRunId: null,
executionRunId: null,
});
});
it("checkout adoption of a stale checkoutRunId preserves the issue's assigneeUserId", async () => {
// Regression for PR #2482 checkout-adoption review finding: any adoption
// helper that re-locks an existing in_progress issue (e.g. when the prior
@ -6973,7 +7063,12 @@ describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adopti
async function seedOwnershipIssue(params: {
checkoutStatus: "running" | "failed" | "timed_out";
actorRunStatus?: "running" | "failed" | "timed_out" | "succeeded";
actorRunStatus?:
| "scheduled_retry"
| "running"
| "failed"
| "timed_out"
| "succeeded";
assigneeMatchesActor?: boolean;
}) {
const companyId = randomUUID();
@ -7108,6 +7203,21 @@ describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adopti
});
});
it("does not let scheduled-retry runs adopt checkout ownership", async () => {
const seeded = await seedOwnershipIssue({
checkoutStatus: "failed",
actorRunStatus: "scheduled_retry",
});
await expect(
svc.assertCheckoutOwner(
seeded.issueId,
seeded.actorAgentId,
seeded.actorRunId,
),
).rejects.toMatchObject({ status: 409 });
});
it("adopts unowned checkout after a concurrent stale-checkout clear wins the lock race", async () => {
const seeded = await seedOwnershipIssue({ checkoutStatus: "failed" });
await db
@ -7164,6 +7274,51 @@ describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adopti
});
});
it("serializes concurrent checkout and unowned ownership assertion without a deadlock", async () => {
const seeded = await seedOwnershipIssue({ checkoutStatus: "failed" });
await db
.update(issues)
.set({
checkoutRunId: null,
executionRunId: null,
executionLockedAt: null,
executionAgentNameKey: null,
})
.where(eq(issues.id, seeded.issueId));
const [checkedOut, ownership] = await Promise.all([
svc.checkout(
seeded.issueId,
seeded.actorAgentId,
["in_progress"],
seeded.actorRunId,
),
svc.assertCheckoutOwner(
seeded.issueId,
seeded.actorAgentId,
seeded.actorRunId,
),
]);
expect(checkedOut.checkoutRunId).toBe(seeded.actorRunId);
expect(checkedOut.executionRunId).toBe(seeded.actorRunId);
expect(ownership.checkoutRunId).toBe(seeded.actorRunId);
expect(ownership.executionRunId).toBe(seeded.actorRunId);
const row = await db
.select({
checkoutRunId: issues.checkoutRunId,
executionRunId: issues.executionRunId,
})
.from(issues)
.where(eq(issues.id, seeded.issueId))
.then((rows) => rows[0]);
expect(row).toEqual({
checkoutRunId: seeded.actorRunId,
executionRunId: seeded.actorRunId,
});
});
});
describeEmbeddedPostgres("issueService.addComment createdByRunId", () => {

258
server/src/services/issues.ts
View File

@ -1856,6 +1856,7 @@ type IssueUserContextInput = {
type ProjectGoalReader = Pick<Db, "select">;
type DbReader = Pick<Db, "select">;
type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
type DbOrTransaction = Db | DbTransaction;
type IssueCreateInput = Omit<typeof issues.$inferInsert, "companyId"> & {
labelIds?: string[];
blockedByIssueIds?: string[];
@ -7482,6 +7483,49 @@ export function issueService(db: Db) {
return heartbeatRunIsTerminalOrMissing(dbOrTx, runId);
}
async function withActiveCheckoutRun<T>(input: {
issueId: string;
companyId: string;
agentId: string;
checkoutRunId: string;
operation: (tx: DbTransaction) => Promise<T>;
}): Promise<T> {
return db.transaction(async (tx) => {
// Keep checkout's lock order aligned with stale-lock cleanup and run
// finalization. Holding both rows through the mutation makes run
// liveness and issue ownership one atomic decision.
await tx.execute(
sql`select ${issues.id} from ${issues} where ${issues.id} = ${input.issueId} for update`,
);
await tx.execute(
sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.checkoutRunId} for update`,
);
const checkoutRun = await tx
.select({
status: heartbeatRuns.status,
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
})
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.checkoutRunId))
.then((rows) => rows[0] ?? null);
const isOwnedRun =
checkoutRun &&
checkoutRun.companyId === input.companyId &&
checkoutRun.agentId === input.agentId;
if (!isOwnedRun || !ACTIVE_RUN_STATUSES.includes(checkoutRun.status)) {
throw conflict("Issue checkout requires a live owning run", {
code: "issue_checkout_run_not_live",
issueId: input.issueId,
actorAgentId: input.agentId,
checkoutRunId: input.checkoutRunId,
runStatus: isOwnedRun ? checkoutRun.status : null,
});
}
return input.operation(tx);
});
}
async function adoptStaleCheckoutRun(input: {
issueId: string;
actorAgentId: string;
@ -7536,7 +7580,7 @@ export function issueService(db: Db) {
const stale =
!existingRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(existingRun.status);
const actorLive =
actorRun && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status);
actorRun && ACTIVE_RUN_STATUSES.includes(actorRun.status);
if (!stale || !actorLive) {
return { adopted: null, latest: lockedIssue };
}
@ -7591,6 +7635,11 @@ export function issueService(db: Db) {
actorRunId: string;
}) {
return db.transaction(async (tx) => {
// Match checkout's issue -> heartbeat lock order to avoid a deadlock when
// an ownership assertion races a checkout for the same issue and run.
await tx.execute(
sql`select ${issues.id} from ${issues} where ${issues.id} = ${input.issueId} for update`,
);
await tx.execute(
sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.actorRunId} for update`,
);
@ -7599,7 +7648,7 @@ export function issueService(db: Db) {
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.actorRunId))
.then((rows) => rows[0] ?? null);
if (!actorRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status))
if (!actorRun || !ACTIVE_RUN_STATUSES.includes(actorRun.status))
return null;
const now = new Date();
@ -11357,33 +11406,13 @@ export function issueService(db: Db) {
}
if (checkoutRunId) {
const checkoutRun = await db
.select({
status: heartbeatRuns.status,
agentId: heartbeatRuns.agentId,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.id, checkoutRunId),
eq(heartbeatRuns.companyId, issueCompany.companyId),
),
)
.then((rows) => rows[0] ?? null);
if (
!checkoutRun ||
checkoutRun.agentId !== agentId ||
!ACTIVE_RUN_STATUSES.includes(checkoutRun.status)
) {
throw conflict("Issue checkout requires a live owning run", {
code: "issue_checkout_run_not_live",
issueId: id,
actorAgentId: agentId,
checkoutRunId,
runStatus:
checkoutRun?.agentId === agentId ? checkoutRun.status : null,
});
}
await withActiveCheckoutRun({
issueId: id,
companyId: issueCompany.companyId,
agentId,
checkoutRunId,
operation: async () => undefined,
});
}
await clearExecutionRunIfTerminal(id);
@ -11428,27 +11457,37 @@ export function issueService(db: Db) {
eq(issues.executionRunId, checkoutRunId),
)
: isNull(issues.executionRunId);
const updated = await db
.update(issues)
.set({
assigneeAgentId: agentId,
assigneeUserId: null,
checkoutRunId,
executionRunId: checkoutRunId,
status: "in_progress",
startedAt: now,
updatedAt: now,
})
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition),
executionLockCondition,
),
)
.returning()
.then((rows) => rows[0] ?? null);
const updateIssue = (dbOrTx: DbOrTransaction) =>
dbOrTx
.update(issues)
.set({
assigneeAgentId: agentId,
assigneeUserId: null,
checkoutRunId,
executionRunId: checkoutRunId,
status: "in_progress",
startedAt: now,
updatedAt: now,
})
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition),
executionLockCondition,
),
)
.returning()
.then((rows) => rows[0] ?? null);
const updated = checkoutRunId
? await withActiveCheckoutRun({
issueId: id,
companyId: issueCompany.companyId,
agentId,
checkoutRunId,
operation: updateIssue,
})
: await updateIssue(db);
if (updated) {
const [enriched] = await withIssueLabels(db, [updated]);
@ -11477,27 +11516,34 @@ export function issueService(db: Db) {
current.executionRunId === checkoutRunId) &&
checkoutRunId
) {
const adopted = await db
.update(issues)
.set({
checkoutRunId,
executionRunId: checkoutRunId,
updatedAt: new Date(),
})
.where(
and(
eq(issues.id, id),
eq(issues.status, "in_progress"),
eq(issues.assigneeAgentId, agentId),
isNull(issues.checkoutRunId),
or(
isNull(issues.executionRunId),
eq(issues.executionRunId, checkoutRunId),
),
),
)
.returning()
.then((rows) => rows[0] ?? null);
const adopted = await withActiveCheckoutRun({
issueId: id,
companyId: issueCompany.companyId,
agentId,
checkoutRunId,
operation: (tx) =>
tx
.update(issues)
.set({
checkoutRunId,
executionRunId: checkoutRunId,
updatedAt: new Date(),
})
.where(
and(
eq(issues.id, id),
eq(issues.status, "in_progress"),
eq(issues.assigneeAgentId, agentId),
isNull(issues.checkoutRunId),
or(
isNull(issues.executionRunId),
eq(issues.executionRunId, checkoutRunId),
),
),
)
.returning()
.then((rows) => rows[0] ?? null),
});
if (adopted) return adopted;
}
@ -11539,6 +11585,7 @@ export function issueService(db: Db) {
current.executionRunId,
);
if (stale) {
const previousExecutionRunId = current.executionRunId;
const now = new Date();
const adoptionSet: Record<string, unknown> = {
assigneeAgentId: agentId,
@ -11552,22 +11599,29 @@ export function issueService(db: Db) {
if (current.status !== "in_progress") {
adoptionSet.startedAt = now;
}
const adopted = await db
.update(issues)
.set(adoptionSet)
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
eq(issues.executionRunId, current.executionRunId),
or(
isNull(issues.assigneeAgentId),
eq(issues.assigneeAgentId, agentId),
),
),
)
.returning()
.then((rows) => rows[0] ?? null);
const adopted = await withActiveCheckoutRun({
issueId: id,
companyId: issueCompany.companyId,
agentId,
checkoutRunId,
operation: (tx) =>
tx
.update(issues)
.set(adoptionSet)
.where(
and(
eq(issues.id, id),
inArray(issues.status, expectedStatuses),
eq(issues.executionRunId, previousExecutionRunId),
or(
isNull(issues.assigneeAgentId),
eq(issues.assigneeAgentId, agentId),
),
),
)
.returning()
.then((rows) => rows[0] ?? null),
});
if (adopted) {
const [enriched] = await withIssueLabels(db, [adopted]);
return enriched;
@ -11581,11 +11635,31 @@ export function issueService(db: Db) {
current.status === "in_progress" &&
sameRunLock(current.checkoutRunId, checkoutRunId)
) {
const row = await db
.select()
.from(issues)
.where(eq(issues.id, id))
.then((rows) => rows[0] ?? null);
const row = checkoutRunId
? await withActiveCheckoutRun({
issueId: id,
companyId: issueCompany.companyId,
agentId,
checkoutRunId,
operation: (tx) =>
tx
.select()
.from(issues)
.where(
and(
eq(issues.id, id),
eq(issues.status, "in_progress"),
eq(issues.assigneeAgentId, agentId),
eq(issues.checkoutRunId, checkoutRunId),
),
)
.then((rows) => rows[0] ?? null),
})
: await db
.select()
.from(issues)
.where(eq(issues.id, id))
.then((rows) => rows[0] ?? null);
if (!row) throw notFound("Issue not found");
const [enriched] = await withIssueLabels(db, [row]);
return enriched;