import { Test, TestingModule } from '@nestjs/testing'; import { Prisma } from '../generated/prisma'; import { claimBatch } from '../src/common/outbox/outbox-claims'; import { PrismaModule } from '../src/prisma/prisma.module'; import { PrismaService } from '../src/prisma/prisma.service'; const ROUNDS = 1000; /** * "availableAt" is TIMESTAMP(3): a value written from SQL now() (stale * release, retry backoff, requeue) is ROUNDED to milliseconds, so it can lie * up to half a millisecond after the next statement's now(). A claim right * behind such a write must still find the row. */ describe('claimBatch and millisecond rounding (e2e)', () => { const correlationId = `claim-rounding-${Date.now()}-${Math.round(Math.random() * 1e6)}`; let moduleRef: TestingModule; let prisma: PrismaService; beforeAll(async () => { moduleRef = await Test.createTestingModule({ imports: [PrismaModule], }).compile(); await moduleRef.init(); prisma = moduleRef.get(PrismaService); }); afterAll(async () => { await prisma.domainEvent.deleteMany({ where: { correlationId } }); await moduleRef.close(); }); it('claims an entry whose availableAt was just set to now()', async () => { const event = await prisma.domainEvent.create({ data: { type: 'TaskDue', aggregateType: 'Task', aggregateId: correlationId, payload: {}, correlationId, }, }); let missed = 0; for (let round = 0; round < ROUNDS; round++) { await prisma.$executeRaw`UPDATE "DomainEvent" SET "status" = 'PENDING', "claimedAt" = NULL, "availableAt" = now() WHERE "id" = ${event.id}`; const claimed = await claimBatch(prisma, 'DomainEvent', { limit: 1, filter: Prisma.sql`"correlationId" = ${correlationId}`, orderBy: 'occurredAt', }); if (claimed.length !== 1) { missed++; } } expect(missed).toBe(0); }); });