diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 21553967b5d..85d6bf6afc0 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -90,6 +90,24 @@ function makeRecoveryActionRow(overrides: Record = {}) { const LATEST_RUN_TIMES = { startedAt: new Date(), createdAt: new Date() }; describe("issueRecoveryActionService", () => { + it.each(["unrelated unique constraint", "other SQLSTATE", "cyclic cause"])("does not retry %s errors", async kind => { + const constraint = "issue_recovery_actions_active_source_uq"; + const error: Error & { cause?: unknown } = new Error("Wrapped database failure"); + error.cause = kind === "unrelated unique constraint" ? { code: "23505", constraint_name: "other_unique_constraint" } + : kind === "other SQLSTATE" ? { code: "23503", constraint_name: constraint } : error; + const query = { from() { return this; }, where() { return this; }, orderBy() { return this; }, + limit() { return Promise.resolve([]); } }; + const fakeDb = { + async transaction(fn: (tx: unknown) => Promise): Promise { return fn(fakeDb); }, + select: vi.fn(() => query), + insert: vi.fn(() => ({ values: () => ({ returning: async () => { throw error; } }) })), + }; + await expect(issueRecoveryActionService(fakeDb as never).upsertSourceScoped({ companyId: "company", sourceIssueId: "issue", + kind: "missing_disposition", cause: "saved-debt", fingerprint: "same-source", nextAction: "Restore existing work." })) + .rejects.toBe(error); + expect(fakeDb.select).toHaveBeenCalledOnce(); + expect(fakeDb.insert).toHaveBeenCalledOnce(); + }); it("does not reactivate an action resolved between the active read and update", async () => { const existingRow = makeRecoveryActionRow({ id: "existing-action", attemptCount: 1 }); const createdRow = makeRecoveryActionRow({ id: "new-action", attemptCount: 1 }); @@ -413,6 +431,49 @@ describeEmbeddedPostgres("issue recovery actions", () => { }, ); + it("converges two service instances after their real inserts compete for one active source", async () => { + const { companyId, managerId, sourceIssueId } = await seedCompany(); + const input = { companyId, sourceIssueId, kind: "stranded_assigned_issue" as const, ownerType: "agent" as const, + ownerAgentId: managerId, cause: "stranded_assigned_issue", fingerprint: "concurrent-source", + evidence: { sourceRunId: "concurrent-source-run" }, lastAttemptAt: new Date("2026-11-03T12:00:00Z"), + nextAction: "Restore the existing execution path." }; + let release!: () => void, acquired!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { acquired = resolve; }); + const lock = db.transaction(async tx => { + await tx.execute(sql`lock table issue_recovery_actions in share mode`); + acquired(); await gate; + }); + await locked; + const outcomes = Promise.allSettled([issueRecoveryActionService(db).upsertSourceScoped(input), + issueRecoveryActionService(db).upsertSourceScoped(input)]); + try { + await vi.waitFor(async () => { + const [waiting] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and state = 'active' and wait_event_type = 'Lock' + and query like 'insert into "issue_recovery_actions"%'`); + expect(waiting.count).toBe(2); + }, { timeout: 5000, interval: 20 }); + } finally { release(); await lock; await outcomes; } + const results = await outcomes; + const active = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId), + eq(issueRecoveryActions.sourceIssueId, sourceIssueId), sql`${issueRecoveryActions.status} in ('active', 'escalated')`)); + expect(active).toHaveLength(1); + expect(results.flatMap(result => result.status === "rejected" ? [{ + name: result.reason?.constructor?.name, code: result.reason?.code, + causeName: result.reason?.cause?.constructor?.name, causeCode: result.reason?.cause?.code, + causeConstraint: result.reason?.cause?.constraint_name, + }] : [])).toEqual([]); + expect(results.map(result => result.status === "fulfilled" ? result.value.id : null)).toEqual([active[0]!.id, active[0]!.id]); + // The lost insert uses the existing upsert merge, just like a second sequential call. + expect(active[0]).toMatchObject({ + status: "active", attemptCount: 2, maxAttempts: null, timeoutAt: null, + ownerType: "agent", ownerAgentId: managerId, ownerUserId: null, + evidence: input.evidence, lastAttemptAt: input.lastAttemptAt, + nextAction: input.nextAction, wakePolicy: null, monitorPolicy: null, + }); + }); + it("upserts one active source-scoped action per issue and keeps company scoping explicit", async () => { const { companyId, managerId, sourceIssueId } = await seedCompany(); const svc = issueRecoveryActionService(db); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index e276dc88f89..ae868406aa6 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -116,19 +116,27 @@ function toReadModel(row: IssueRecoveryActionRow): IssueRecoveryAction { } function isUniqueRecoveryActionConflict(error: unknown) { - const maybe = error as { code?: string; constraint?: string; message?: string } | null; - return Boolean( - maybe && - maybe.code === "23505" && - ( + const seen = new Set(); + let candidate = error; + // Drizzle wraps the driver error in cause. Only this specific uniqueness + // conflict can reuse the existing bounded source-scoped upsert retry. + for (let depth = 0; depth < 8 && candidate && typeof candidate === "object"; depth++) { + if (seen.has(candidate)) break; + seen.add(candidate); + const maybe = candidate as { code?: string; constraint?: string; constraint_name?: string; message?: string; cause?: unknown }; + if (maybe.code === "23505" && ( maybe.constraint === "issue_recovery_actions_active_source_uq" || maybe.constraint === "issue_recovery_actions_active_fingerprint_uq" || + maybe.constraint_name === "issue_recovery_actions_active_source_uq" || + maybe.constraint_name === "issue_recovery_actions_active_fingerprint_uq" || typeof maybe.message === "string" && ( maybe.message.includes("issue_recovery_actions_active_source_uq") || maybe.message.includes("issue_recovery_actions_active_fingerprint_uq") ) - ), - ); + )) return true; + candidate = maybe.cause; + } + return false; } export function issueRecoveryActionService(db: DbOrTransaction) {