mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 21:59:38 +01:00
fix: prevent inbox archive deadlocks during completion (#15615)
Take a company-scoped parent lock before writing archive state and reuse the caller transaction. Preserve newer archive timestamps and attribution when a request waits, so completed tasks remain archived. Verified with real PostgreSQL race, rollback, company isolation, concurrency, timestamp and visibility tests, independent review, full CI and Greptile 5/5. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3d1d5294d9
commit
1881894973
3 files changed
+219
-25
No files matched your search
@@ -25,6 +25,7 @@ import {
|
||||
isNull,
|
||||
like,
|
||||
lt,
|
||||
lte,
|
||||
ne,
|
||||
notExists,
|
||||
notInArray,
|
||||
@@ -7951,35 +7952,64 @@ export function issueService(db: Db) {
|
||||
},
|
||||
dbOrTx: any = db,
|
||||
) {
|
||||
const now = new Date();
|
||||
const [row] = await dbOrTx
|
||||
.insert(issueInboxArchives)
|
||||
.values({
|
||||
companyId,
|
||||
issueId,
|
||||
userId,
|
||||
archivedByActorType: attribution?.archivedByActorType ?? "user",
|
||||
archivedByAgentId: attribution?.archivedByAgentId ?? null,
|
||||
archivedByRunId: attribution?.archivedByRunId ?? null,
|
||||
archivedAt,
|
||||
updatedAt: now,
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
issueInboxArchives.companyId,
|
||||
issueInboxArchives.issueId,
|
||||
issueInboxArchives.userId,
|
||||
],
|
||||
set: {
|
||||
archivedAt,
|
||||
const runArchive = async (tx: typeof dbOrTx) => {
|
||||
// Completion locks the issue before archiving. Take the FK's parent lock
|
||||
// first too, or an insert can hold the archive key while waiting on that
|
||||
// issue and deadlock with completion's archive UPSERT. SHARE also holds
|
||||
// companyId stable; different users can still archive concurrently.
|
||||
const [issue] = await tx
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)))
|
||||
.for("share");
|
||||
if (!issue) throw notFound("Issue not found");
|
||||
|
||||
const now = new Date();
|
||||
const [row] = await tx
|
||||
.insert(issueInboxArchives)
|
||||
.values({
|
||||
companyId,
|
||||
issueId,
|
||||
userId,
|
||||
archivedByActorType: attribution?.archivedByActorType ?? "user",
|
||||
archivedByAgentId: attribution?.archivedByAgentId ?? null,
|
||||
archivedByRunId: attribution?.archivedByRunId ?? null,
|
||||
archivedAt,
|
||||
updatedAt: now,
|
||||
},
|
||||
})
|
||||
.returning();
|
||||
return row;
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
issueInboxArchives.companyId,
|
||||
issueInboxArchives.issueId,
|
||||
issueInboxArchives.userId,
|
||||
],
|
||||
set: {
|
||||
archivedAt,
|
||||
archivedByActorType: attribution?.archivedByActorType ?? "user",
|
||||
archivedByAgentId: attribution?.archivedByAgentId ?? null,
|
||||
archivedByRunId: attribution?.archivedByRunId ?? null,
|
||||
updatedAt: now,
|
||||
},
|
||||
// A request that waited behind completion must not replace the newer
|
||||
// archive with its earlier request time and resurface the done task.
|
||||
setWhere: lte(issueInboxArchives.archivedAt, archivedAt),
|
||||
})
|
||||
.returning();
|
||||
if (row) return row;
|
||||
// ON CONFLICT holds this row lock even when setWhere skips the update.
|
||||
// Return the newer state, including its matching actor attribution.
|
||||
const [existing] = await tx
|
||||
.select()
|
||||
.from(issueInboxArchives)
|
||||
.where(and(
|
||||
eq(issueInboxArchives.companyId, companyId),
|
||||
eq(issueInboxArchives.issueId, issueId),
|
||||
eq(issueInboxArchives.userId, userId),
|
||||
));
|
||||
if (!existing) throw new Error("Inbox archive conflict row missing");
|
||||
return existing;
|
||||
};
|
||||
return dbOrTx === db ? db.transaction(runArchive) : runArchive(dbOrTx);
|
||||
}
|
||||
|
||||
const service = {
|
||||
|
||||
Reference in new issue
Block a user