mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 12:59:13 +01:00
fix: make disposition repair wakeups atomic
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
2a6eaab1c1
commit
e1121d790c
3 files changed
+79
No files matched your search
@@ -4370,6 +4370,56 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(repairRun?.contextSnapshot).not.toHaveProperty("allowDeliverableWork");
|
||||
});
|
||||
|
||||
it("deduplicates concurrent immediate disposition-repair wakes inside the issue lock", async () => {
|
||||
const { companyId, agentId, issueId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
runStatus: "cancelled",
|
||||
retryReason: "issue_continuation_needed",
|
||||
runErrorCode: "issue_continuation_waiting_on_review",
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
const idempotencyKey = `issue_disposition_repair:${issueId}:test-fingerprint:1`;
|
||||
const wakeOptions = {
|
||||
source: "automation" as const,
|
||||
triggerDetail: "system" as const,
|
||||
reason: "issue_disposition_repair",
|
||||
idempotencyKey,
|
||||
requestedByActorType: "system",
|
||||
requestedByActorId: null,
|
||||
payload: {
|
||||
issueId,
|
||||
dispositionRepairFingerprint: "test-fingerprint",
|
||||
dispositionRepairAttempt: 1,
|
||||
},
|
||||
contextSnapshot: {
|
||||
issueId,
|
||||
taskId: issueId,
|
||||
wakeReason: "issue_disposition_repair",
|
||||
retryReason: "issue_disposition_repair",
|
||||
dispositionRepairFingerprint: "test-fingerprint",
|
||||
dispositionRepairAttempt: 1,
|
||||
bypassContinuationSummaryPark: true,
|
||||
},
|
||||
};
|
||||
|
||||
const [firstRun, secondRun] = await Promise.all([
|
||||
heartbeat.wakeup(agentId, wakeOptions),
|
||||
heartbeat.wakeup(agentId, wakeOptions),
|
||||
]);
|
||||
|
||||
expect(firstRun?.id).toBeTruthy();
|
||||
expect(secondRun?.id).toBe(firstRun?.id);
|
||||
const requests = await db
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.idempotencyKey, idempotencyKey),
|
||||
));
|
||||
expect(requests).toHaveLength(1);
|
||||
expect(requests[0]?.runId).toBe(firstRun?.id);
|
||||
});
|
||||
|
||||
it("does not reset disposition repair for prose but does reset for durable source state", async () => {
|
||||
const { companyId, agentId, runId, issueId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
|
||||
@@ -17580,6 +17580,28 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return { kind: "skipped" as const };
|
||||
}
|
||||
|
||||
// Issue wakes are serialized by the source-row lock above. Recheck an
|
||||
// idempotency key inside that lock so concurrent callers cannot both
|
||||
// pass an earlier advisory lookup and create separate runs. Keep
|
||||
// skipped requests retryable because they never dispatched work.
|
||||
if (opts.idempotencyKey?.startsWith("issue_disposition_repair:")) {
|
||||
const existingIdempotentWake = await tx
|
||||
.select({ run: heartbeatRuns })
|
||||
.from(agentWakeupRequests)
|
||||
.leftJoin(heartbeatRuns, eq(heartbeatRuns.wakeupRequestId, agentWakeupRequests.id))
|
||||
.where(and(
|
||||
eq(agentWakeupRequests.companyId, agent.companyId),
|
||||
eq(agentWakeupRequests.idempotencyKey, opts.idempotencyKey),
|
||||
ne(agentWakeupRequests.status, "skipped"),
|
||||
))
|
||||
.orderBy(desc(agentWakeupRequests.requestedAt))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (existingIdempotentWake) {
|
||||
return { kind: "idempotent" as const, run: existingIdempotentWake.run };
|
||||
}
|
||||
}
|
||||
|
||||
if (worktreeExecutionCutoff && issue.createdAt < worktreeExecutionCutoff) {
|
||||
await tx.insert(agentWakeupRequests).values({
|
||||
companyId: agent.companyId,
|
||||
@@ -18308,6 +18330,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
});
|
||||
|
||||
if (outcome.kind === "deferred" || outcome.kind === "skipped") return null;
|
||||
if (outcome.kind === "idempotent") return outcome.run;
|
||||
if (outcome.kind === "coalesced") {
|
||||
await startNextQueuedRunForAgent(agent.id);
|
||||
return outcome.run;
|
||||
|
||||
@@ -3492,6 +3492,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
}
|
||||
} else {
|
||||
scheduledRun = await db.transaction(async (tx) => {
|
||||
// Serialize every attempt for this source before the in-transaction
|
||||
// idempotency recheck. This covers delayed attempts that bypass the
|
||||
// normal heartbeat enqueue path and therefore do not acquire its lock.
|
||||
await tx.execute(
|
||||
sql`select id from issues where id = ${input.issue.id} and company_id = ${input.issue.companyId} for update`,
|
||||
);
|
||||
const existing = await tx
|
||||
.select({ run: heartbeatRuns })
|
||||
.from(agentWakeupRequests)
|
||||
|
||||
Reference in new issue
Block a user