From ff19cf66b108768b7dd0ad3487396e415d1b20fa Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 9 May 2026 23:58:17 -0500 Subject: [PATCH] Serialize source recovery action upserts --- server/src/services/issue-recovery-actions.ts | 36 +++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 1e39c066f2b..cf6083e6967 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -94,6 +94,32 @@ function isUniqueRecoveryActionConflict(error: unknown) { } export function issueRecoveryActionService(db: Db) { + const upsertQueues = new Map>(); + + async function runExclusiveUpsert( + input: UpsertIssueRecoveryActionInput, + task: () => Promise, + ): Promise { + const key = `${input.companyId}:${input.sourceIssueId}`; + const previous = upsertQueues.get(key) ?? Promise.resolve(); + let release: () => void = () => {}; + const current = new Promise((resolve) => { + release = resolve; + }); + const next = previous.catch(() => undefined).then(() => current); + upsertQueues.set(key, next); + + await previous.catch(() => undefined); + try { + return await task(); + } finally { + release(); + if (upsertQueues.get(key) === next) { + upsertQueues.delete(key); + } + } + } + async function getActiveForIssue(companyId: string, sourceIssueId: string): Promise { const row = await db .select() @@ -142,10 +168,10 @@ export function issueRecoveryActionService(db: Db) { `Failed to upsert active recovery action for issue ${input.sourceIssueId} after ${MAX_UPSERT_RETRIES} retries`, ); } - return upsertSourceScoped(input, retryCount + 1); + return upsertSourceScopedUnlocked(input, retryCount + 1); } - async function upsertSourceScoped( + async function upsertSourceScopedUnlocked( input: UpsertIssueRecoveryActionInput, retryCount = 0, ): Promise { @@ -225,6 +251,12 @@ export function issueRecoveryActionService(db: Db) { } } + async function upsertSourceScoped( + input: UpsertIssueRecoveryActionInput, + ): Promise { + return runExclusiveUpsert(input, () => upsertSourceScopedUnlocked(input)); + } + async function resolveActiveForIssue( input: ResolveIssueRecoveryActionInput, dbOrTx: DbOrTransaction = db,