From 58d0a98134b201cf582d52c745e7898185d4e974 Mon Sep 17 00:00:00 2001 From: saqib mir Date: Wed, 9 Sep 2026 17:10:10 +0530 Subject: [PATCH] =?UTF-8?q?fix(016-load-concurrency-testing):=20US1=20?= =?UTF-8?q?=E2=80=94=20retry=20assignment=20creation=20on=20race=20conflic?= =?UTF-8?q?t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AssignmentRepository.createAssignment now catches the assignments_one_current_per_ticket unique-violation and retries the whole supersede-then-create transaction (bounded, with jitter) instead of propagating a raw P2002 to the caller. Verified against real Postgres: before this fix, 20 genuinely concurrent assignment attempts on the same ticket reliably threw an unhandled unique-constraint error; after it, exactly one current assignment results every time across 10 repeated runs. Co-Authored-By: Claude Sonnet 5 --- .../repository/assignment.repository.ts | 67 +++++-- tests/concurrency/assignment-race.test.ts | 178 ++++++++++++++++++ 2 files changed, 230 insertions(+), 15 deletions(-) create mode 100644 tests/concurrency/assignment-race.test.ts diff --git a/src/modules/orchestration/assignments/repository/assignment.repository.ts b/src/modules/orchestration/assignments/repository/assignment.repository.ts index df4a662..080cc5f 100644 --- a/src/modules/orchestration/assignments/repository/assignment.repository.ts +++ b/src/modules/orchestration/assignments/repository/assignment.repository.ts @@ -8,6 +8,24 @@ export interface CreateAssignmentData { reason?: string | undefined; } +// Bounded, but generous: under N genuinely concurrent attempts on the same ticket, a given +// attempt can collide with a different still-in-flight one on each of several retries before +// the field of contenders drains — 3 was observed to be too few under a 20-way race in this +// project's own concurrency test (tests/concurrency/assignment-race.test.ts). +const MAX_CREATE_ASSIGNMENT_ATTEMPTS = 20; + +function isCurrentAssignmentConflict(error: unknown): boolean { + return ( + error instanceof Prisma.PrismaClientKnownRequestError && + error.code === 'P2002' && + (error.meta?.target as string[] | undefined)?.includes('ticketId') === true + ); +} + +function jitterDelayMs(): number { + return Math.floor(Math.random() * 15); +} + export class AssignmentRepository { constructor(private readonly prisma = prismaClient) {} @@ -16,23 +34,42 @@ export class AssignmentRepository { * supersedes any existing current row for this ticket (isCurrent: false, unassignedAt: now()) * and inserts the new current row — the same "never overwrite, always a new row" guarantee * 004's KnowledgeEntry versioning already established for a different entity. + * + * 016-load-concurrency-testing research.md §1: at Postgres's default READ COMMITTED + * isolation, two concurrent calls can each see "nothing current to supersede" and both + * attempt to `create` their own `isCurrent: true` row. The `assignments_one_current_per_ticket` + * partial unique index (migration 20260909120000) makes the second one fail fast with `P2002` + * instead of silently succeeding — caught here and retried (bounded) so the loser's own + * request still applies, correctly superseding the winner's row on the next attempt, rather + * than surfacing a raw conflict to a caller that did nothing wrong. */ async createAssignment(data: CreateAssignmentData): Promise { - return this.prisma.$transaction(async (tx) => { - await tx.assignment.updateMany({ - where: { ticketId: data.ticketId, isCurrent: true }, - data: { isCurrent: false, unassignedAt: new Date() }, - }); - return tx.assignment.create({ - data: { - ticketId: data.ticketId, - agentId: data.agentId, - strategy: data.strategy, - reason: data.reason, - isCurrent: true, - } as Prisma.AssignmentUncheckedCreateInput, - }); - }); + for (let attempt = 1; attempt <= MAX_CREATE_ASSIGNMENT_ATTEMPTS; attempt++) { + try { + return await this.prisma.$transaction(async (tx) => { + await tx.assignment.updateMany({ + where: { ticketId: data.ticketId, isCurrent: true }, + data: { isCurrent: false, unassignedAt: new Date() }, + }); + return tx.assignment.create({ + data: { + ticketId: data.ticketId, + agentId: data.agentId, + strategy: data.strategy, + reason: data.reason, + isCurrent: true, + } as Prisma.AssignmentUncheckedCreateInput, + }); + }); + } catch (error) { + if (!isCurrentAssignmentConflict(error) || attempt === MAX_CREATE_ASSIGNMENT_ATTEMPTS) { + throw error; + } + await new Promise((resolve) => setTimeout(resolve, jitterDelayMs())); + } + } + /* istanbul ignore next -- unreachable: the loop above always returns or throws */ + throw new Error('createAssignment: exhausted retry attempts unexpectedly.'); } async findCurrent(ticketId: string): Promise { diff --git a/tests/concurrency/assignment-race.test.ts b/tests/concurrency/assignment-race.test.ts new file mode 100644 index 0000000..edd969d --- /dev/null +++ b/tests/concurrency/assignment-race.test.ts @@ -0,0 +1,178 @@ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { buildApp } from '@/app'; +import { prismaClient } from '@/infrastructure/database'; +import { FastifyInstance } from 'fastify'; +import { loginAs, authHeader } from '../helpers/auth'; +import { + encryptCredential, + generateCredentialSecret, + issueIntegrationToken, +} from '@/modules/catalog/products'; +import { assignmentEngine } from '@/modules/orchestration/assignments'; + +/** + * specs/016-load-concurrency-testing User Story 1 / FR-001 / SC-001: two or more concurrent + * assignment attempts on the same ticket must never leave more than one Assignment row marked + * `isCurrent`. Before research.md §1's fix (a Postgres partial unique index on + * `assignments(ticketId) WHERE isCurrent = true`, plus a bounded retry in + * AssignmentRepository.createAssignment), Postgres's default READ COMMITTED isolation let two + * concurrent `assignmentEngine.assignToSpecificNode` calls each see "nothing current to + * supersede" and both successfully create their own `isCurrent: true` row. + */ +describe('Assignment double-assignment race (User Story 1)', () => { + let app: FastifyInstance; + let authToken: string; + const externalProductId = `TEST_ASSIGN_RACE_PROD_${Date.now()}`; + const skillTag = `assign_race_skill_${Date.now()}`; + let productId: string; + let teamId: string; + const agentIds: string[] = []; + let nodeId: string; + let secret: string; + const createdTicketIds: string[] = []; + + async function createTicket(): Promise { + const token = issueIntegrationToken(secret, { + externalProductId, + tenantId: 'tenant-1', + userId: 'user-1', + }); + const created = await app.inject({ + method: 'POST', + url: '/v1/support/requests', + headers: { authorization: `Bearer ${token}` }, + payload: { + productId: externalProductId, + tenantId: 'tenant-1', + userId: 'user-1', + source: 'test', + problem: `Assignment race test ${Date.now()}-${Math.random()}`, + }, + }); + expect(created.statusCode).toBe(202); + const ticketId = created.json().data.ticketId as string; + createdTicketIds.push(ticketId); + return ticketId; + } + + beforeAll(async () => { + app = await buildApp(); + authToken = await loginAs(app, 'ADMIN'); + + const product = await prismaClient.product.create({ + data: { externalProductId, name: 'Assignment Race Test Product', status: 'active' }, + }); + productId = product.id; + secret = generateCredentialSecret(); + await prismaClient.productIntegration.create({ + data: { + productId, + credentialRef: encryptCredential(secret), + authMechanism: 'signed_token', + allowedScope: { tenantIds: ['tenant-1'] }, + status: 'active', + rateLimitPerMinute: 1000, + rateLimitPerUserPerMinute: 1000, + }, + }); + + const team = await app.inject({ + method: 'POST', + url: '/admin/teams', + headers: authHeader(authToken), + payload: { name: `Assign Race Team ${Date.now()}` }, + }); + teamId = team.json().data.id; + + // Several eligible agents so a race has real agents to (incorrectly) double-assign across, + // not just one candidate every attempt would trivially agree on. + for (let i = 0; i < 5; i++) { + const agent = await app.inject({ + method: 'POST', + url: `/admin/teams/${teamId}/agents`, + headers: authHeader(authToken), + payload: { name: `Assign Race Agent ${i}` }, + }); + const agentId = agent.json().data.id; + agentIds.push(agentId); + await app.inject({ + method: 'PUT', + url: `/admin/agents/${agentId}/skills/${skillTag}`, + headers: authHeader(authToken), + payload: { level: 3 }, + }); + } + + const node = await app.inject({ + method: 'POST', + url: '/admin/hierarchy-nodes', + headers: authHeader(authToken), + payload: { + name: 'Assign Race Node', + order: 0, + productScope: [externalProductId], + skills: [skillTag], + assignmentStrategy: 'ROUND_ROBIN', + }, + }); + nodeId = node.json().data.id; + }); + + afterAll(async () => { + const ticketFilter = { ticketId: { in: createdTicketIds } }; + await prismaClient.assignmentHistory.deleteMany({ where: ticketFilter }); + await prismaClient.assignment.deleteMany({ where: ticketFilter }); + await prismaClient.hierarchyNode.deleteMany({ where: { id: nodeId } }); + await prismaClient.agentSkill.deleteMany({ where: { agentId: { in: agentIds } } }); + await prismaClient.agent.deleteMany({ where: { teamId } }); + await prismaClient.team.deleteMany({ where: { id: teamId } }); + await prismaClient.ticketMessage.deleteMany({ where: ticketFilter }); + await prismaClient.ticket.deleteMany({ where: { id: { in: createdTicketIds } } }); + await prismaClient.problem.deleteMany({ where: { productId } }); + await prismaClient.productIntegration.deleteMany({ where: { productId } }); + await prismaClient.product.deleteMany({ where: { id: productId } }); + await app.close(); + }); + + it('leaves exactly one current assignment after 20 genuinely concurrent assignment attempts', async () => { + const ticketId = await createTicket(); + + const attempts = 20; + await Promise.all( + Array.from({ length: attempts }, () => + assignmentEngine.assignToSpecificNode(ticketId, nodeId, 'system', 'concurrency test'), + ), + ); + + const currentAssignments = await prismaClient.assignment.findMany({ + where: { ticketId, isCurrent: true }, + }); + expect(currentAssignments.length).toBe(1); + + // Every attempt was still recorded (superseded or current) — the race must not have + // silently dropped attempts, only converged them onto a single current row. + const allAssignments = await prismaClient.assignment.findMany({ where: { ticketId } }); + expect(allAssignments.length).toBe(attempts); + }); + + it( + 'holds consistently across 10 repeated runs (SC-001: zero exceptions)', + async () => { + for (let run = 0; run < 10; run++) { + const ticketId = await createTicket(); + + await Promise.all( + Array.from({ length: 20 }, () => + assignmentEngine.assignToSpecificNode(ticketId, nodeId, 'system', `run ${run}`), + ), + ); + + const currentAssignments = await prismaClient.assignment.findMany({ + where: { ticketId, isCurrent: true }, + }); + expect(currentAssignments.length).toBe(1); + } + }, + 60000, + ); +});