diff --git a/api/prisma/migrations/20260726223000_prevent_duplicate_retry_side_effects/migration.sql b/api/prisma/migrations/20260726223000_prevent_duplicate_retry_side_effects/migration.sql new file mode 100644 index 0000000..2240fae --- /dev/null +++ b/api/prisma/migrations/20260726223000_prevent_duplicate_retry_side_effects/migration.sql @@ -0,0 +1,38 @@ +ALTER TABLE "SmsSubmitRecord" +ADD COLUMN "retryOfSubmitRecordId" TEXT; + +CREATE UNIQUE INDEX "SmsSubmitRecord_retryOfSubmitRecordId_key" +ON "SmsSubmitRecord"("retryOfSubmitRecordId"); + +ALTER TABLE "SmsSubmitRecord" +ADD CONSTRAINT "SmsSubmitRecord_retryOfSubmitRecordId_fkey" +FOREIGN KEY ("retryOfSubmitRecordId") REFERENCES "SmsSubmitRecord"("id") +ON DELETE SET NULL ON UPDATE CASCADE; + +ALTER TABLE "AccountTransaction" +ADD COLUMN "idempotencyKey" TEXT; + +CREATE UNIQUE INDEX "AccountTransaction_idempotencyKey_key" +ON "AccountTransaction"("idempotencyKey"); + +ALTER TABLE "CmppDownstreamDelivery" +ADD COLUMN "dedupeKey" TEXT; + +WITH ranked_receipts AS ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY "messageRecordId", "deliveryType" + ORDER BY "createdAt", id + ) AS row_number + FROM "CmppDownstreamDelivery" + WHERE "messageRecordId" IS NOT NULL + AND "deliveryType" = 'receipt' +) +UPDATE "CmppDownstreamDelivery" AS delivery +SET "dedupeKey" = 'receipt:' || delivery."messageRecordId" +FROM ranked_receipts +WHERE delivery.id = ranked_receipts.id + AND ranked_receipts.row_number = 1; + +CREATE UNIQUE INDEX "CmppDownstreamDelivery_dedupeKey_key" +ON "CmppDownstreamDelivery"("dedupeKey"); diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 23ce076..7f0d19f 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -318,6 +318,7 @@ model AccountTransaction { id String @id @default(cuid()) tenantId String transactionType String + idempotencyKey String? @unique amountCents BigInt @default(0) balanceAfter BigInt @default(0) relatedType String? @@ -1441,30 +1442,33 @@ model CmppSubmitSession { } model SmsSubmitRecord { - id String @id @default(cuid()) - tenantId String? - batchTaskId String? - messageRecordId String - channelId String - sessionId String? - submitId String @unique - sequenceId Int? - gatewayMessageId String? - submitStatus String @default("queued") - costUnitPrice BigInt @default(0) - costAmountCents BigInt @default(0) - errorCode String? - errorMessage String? - submittedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String? + batchTaskId String? + messageRecordId String + channelId String + sessionId String? + retryOfSubmitRecordId String? @unique + submitId String @unique + sequenceId Int? + gatewayMessageId String? + submitStatus String @default("queued") + costUnitPrice BigInt @default(0) + costAmountCents BigInt @default(0) + errorCode String? + errorMessage String? + submittedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant? @relation(fields: [tenantId], references: [id]) - batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade) - messageRecord SmsMessageRecord @relation(fields: [messageRecordId], references: [id], onDelete: Cascade) - channel SmsChannel @relation(fields: [channelId], references: [id]) - session CmppSubmitSession? @relation(fields: [sessionId], references: [id]) - segmentAudits SmsMessageSegmentAudit[] + tenant Tenant? @relation(fields: [tenantId], references: [id]) + batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade) + messageRecord SmsMessageRecord @relation(fields: [messageRecordId], references: [id], onDelete: Cascade) + channel SmsChannel @relation(fields: [channelId], references: [id]) + session CmppSubmitSession? @relation(fields: [sessionId], references: [id]) + retryOfSubmitRecord SmsSubmitRecord? @relation("SmsSubmitRetry", fields: [retryOfSubmitRecordId], references: [id]) + retrySubmit SmsSubmitRecord? @relation("SmsSubmitRetry") + segmentAudits SmsMessageSegmentAudit[] @@index([tenantId, createdAt]) @@index([messageRecordId]) @@ -1728,6 +1732,7 @@ model CmppDownstreamDelivery { applicationId String messageRecordId String? messageId String? + dedupeKey String? @unique deliveryType String status String @default("pending") payload Json diff --git a/api/src/billing/billing.service.spec.ts b/api/src/billing/billing.service.spec.ts index 9ff61f6..4c8213a 100644 --- a/api/src/billing/billing.service.spec.ts +++ b/api/src/billing/billing.service.spec.ts @@ -18,7 +18,8 @@ function createPrismaMock() { return Promise.resolve({ count: 1 }); }), update: jest.fn().mockImplementation(({ data }) => { - if (data.balanceCents !== undefined) accountState.balanceCents = data.balanceCents; + if (data.balanceCents?.increment !== undefined) accountState.balanceCents += data.balanceCents.increment; + else if (data.balanceCents !== undefined) accountState.balanceCents = data.balanceCents; if (data.creditCents !== undefined) accountState.creditCents = data.creditCents; return Promise.resolve({ ...accountState }); }), @@ -26,6 +27,7 @@ function createPrismaMock() { accountTransaction: { findMany: jest.fn(), findFirst: jest.fn(), + findUnique: jest.fn().mockResolvedValue(null), create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: `tx-${data.transactionType}`, ...data })), }, rechargeOrder: { @@ -125,7 +127,7 @@ describe('BillingService', () => { expect(order).toEqual(expect.objectContaining({ amountCents: 500, status: 'paid' })); expect(prisma.tenantAccount.update).toHaveBeenCalledWith({ where: { tenantId: 'tenant-1' }, - data: { balanceCents: 1500 }, + data: { balanceCents: { increment: 500 } }, }); expect(prisma.accountTransaction.create).toHaveBeenCalledWith({ data: expect.objectContaining({ @@ -302,6 +304,46 @@ describe('BillingService', () => { expect(prisma.accountState).toEqual(expect.objectContaining({ balanceCents: 890 })); }); + it('serializes and replays concurrent refunds with one balance mutation', async () => { + const prisma = createPrismaMock(); + let transactionChain = Promise.resolve(undefined); + let persistedTransaction: Record | null = null; + prisma.$transaction.mockImplementation((callback) => { + const run = transactionChain.then(() => callback(prisma)); + transactionChain = run.then(() => undefined, () => undefined); + return run; + }); + prisma.accountTransaction.findUnique.mockImplementation(() => Promise.resolve(persistedTransaction)); + prisma.accountTransaction.create.mockImplementation(({ data }) => { + persistedTransaction = { id: 'tx-refund-once', ...data }; + return Promise.resolve(persistedTransaction); + }); + const service = new BillingService(prisma as never); + const refund = { + tenantId: 'tenant-1', + amountCents: 1053, + idempotencyKey: 'sms-refund:MSG-LONG-RACE', + relatedType: 'sms_message_record', + relatedId: 'MSG-LONG-RACE', + remark: '最终失败退款', + }; + + const results = await Promise.all([ + service.refund(refund), + service.refund(refund), + service.refund(refund), + ]); + + expect(results.map((result) => result.id)).toEqual([ + 'tx-refund-once', + 'tx-refund-once', + 'tx-refund-once', + ]); + expect(prisma.tenantAccount.update).toHaveBeenCalledTimes(1); + expect(prisma.accountTransaction.create).toHaveBeenCalledTimes(1); + expect(prisma.accountState.balanceCents).toBe(2053); + }); + it('creates SMS billing records linked to message and task identifiers', async () => { const prisma = createPrismaMock(); const service = new BillingService(prisma as never); diff --git a/api/src/billing/billing.service.ts b/api/src/billing/billing.service.ts index 2b8a74b..d276fa4 100644 --- a/api/src/billing/billing.service.ts +++ b/api/src/billing/billing.service.ts @@ -19,6 +19,7 @@ export interface UpdateCreditLimitDto { export interface CreateAccountTransactionDto { tenantId: string; transactionType: string; + idempotencyKey?: string; amountCents?: number; balanceAfter?: number; relatedType?: string; @@ -67,6 +68,7 @@ export interface EstimateSmsCostDto { export interface BillingActionDto { tenantId: string; + idempotencyKey?: string; amountCents?: number; relatedType?: string; relatedId?: string; @@ -454,25 +456,46 @@ export class BillingService { } private async applyAccountDelta(data: CreateAccountTransactionDto) { - const account = await this.getAccountOrCreate(data.tenantId); - const nextBalance = moneyToNumber(account.balanceCents) + (data.amountCents ?? 0); - await this.prisma.tenantAccount.update({ - where: { tenantId: data.tenantId }, - data: { - balanceCents: nextBalance, - }, - }); - - return this.prisma.accountTransaction.create({ - data: { - tenantId: data.tenantId, - transactionType: data.transactionType, - amountCents: data.amountCents ?? 0, - balanceAfter: nextBalance, - relatedType: data.relatedType, - relatedId: data.relatedId, - remark: data.remark, - }, + const amountCents = data.amountCents ?? 0; + const idempotencyKey = data.idempotencyKey?.trim() || null; + return this.prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'tenant-account:' + data.tenantId}, 0))`; + if (idempotencyKey) { + const existing = await tx.accountTransaction.findUnique({ where: { idempotencyKey } }); + if (existing) { + if ( + existing.tenantId !== data.tenantId + || existing.transactionType !== data.transactionType + || moneyToNumber(existing.amountCents) !== amountCents + || existing.relatedType !== (data.relatedType ?? null) + || existing.relatedId !== (data.relatedId ?? null) + ) { + throw new ConflictException('账务幂等键已用于另一笔交易'); + } + return existing; + } + } + await tx.tenantAccount.upsert({ + where: { tenantId: data.tenantId }, + update: {}, + create: { tenantId: data.tenantId, balanceCents: 0, creditCents: 0, status: 'active' }, + }); + const account = await tx.tenantAccount.update({ + where: { tenantId: data.tenantId }, + data: { balanceCents: { increment: amountCents } }, + }); + return tx.accountTransaction.create({ + data: { + tenantId: data.tenantId, + transactionType: data.transactionType, + idempotencyKey, + amountCents, + balanceAfter: account.balanceCents, + relatedType: data.relatedType, + relatedId: data.relatedId, + remark: data.remark, + }, + }); }); } } diff --git a/api/src/open-api/open-api.service.spec.ts b/api/src/open-api/open-api.service.spec.ts index 18cc038..f06ae54 100644 --- a/api/src/open-api/open-api.service.spec.ts +++ b/api/src/open-api/open-api.service.spec.ts @@ -65,13 +65,20 @@ describe('OpenApiService', () => { const prisma = { smsApplication: { findUnique: jest.fn().mockResolvedValue({ httpConfig: { enabled: true, receiptWebhookEnabled: true, receiptDeliveryMode: 'http' } }) }, httpWebhookEndpoint: { findUnique: jest.fn().mockResolvedValue({ id: 'endpoint-1', status: 'active' }) }, - httpWebhookEvent: { create: jest.fn().mockResolvedValue({ id: 'event-row-1' }) }, - httpWebhookDelivery: { create: jest.fn().mockResolvedValue({ id: 'delivery-1' }) }, + httpWebhookEvent: { upsert: jest.fn().mockResolvedValue({ id: 'event-row-1' }) }, + httpWebhookDelivery: { upsert: jest.fn().mockResolvedValue({ id: 'delivery-1', status: 'pending' }) }, }; const service = new OpenApiService(prisma as never, {} as never); - await service.queueWebhookEvent({ tenantId: 'tenant-1', applicationId: 'app-1', eventType: 'receipt', messageId: 'MSG-1', payload: { receiptStatus: 'delivered' } }); - expect(prisma.httpWebhookEvent.create).toHaveBeenCalled(); - expect(prisma.httpWebhookDelivery.create).toHaveBeenCalledWith({ data: { eventId: 'event-row-1', endpointId: 'endpoint-1' } }); + const input = { tenantId: 'tenant-1', applicationId: 'app-1', eventType: 'receipt' as const, messageRecordId: 'record-1', messageId: 'MSG-1', payload: { receiptStatus: 'delivered' } }; + await service.queueWebhookEvent(input); + await service.queueWebhookEvent(input); + expect(prisma.httpWebhookEvent.upsert).toHaveBeenCalledWith(expect.objectContaining({ + where: { eventId: 'evt_receipt_record-1' }, + })); + expect(prisma.httpWebhookEvent.upsert).toHaveBeenCalledTimes(2); + expect(prisma.httpWebhookDelivery.upsert).toHaveBeenCalledWith(expect.objectContaining({ + create: { eventId: 'event-row-1', endpointId: 'endpoint-1' }, + })); }); it('defaults a newly enabled HTTP interface to all six capabilities and automatic dual delivery', async () => { diff --git a/api/src/open-api/open-api.service.ts b/api/src/open-api/open-api.service.ts index ef476cc..52c95b3 100644 --- a/api/src/open-api/open-api.service.ts +++ b/api/src/open-api/open-api.service.ts @@ -315,10 +315,22 @@ export class OpenApiService implements OnModuleInit, OnModuleDestroy { if (!config?.enabled || !enabled) return null; const endpoint = await this.prisma.httpWebhookEndpoint.findUnique({ where: { applicationId_eventType: { applicationId: data.applicationId, eventType: data.eventType } } }); if (!endpoint || endpoint.status !== 'active') return null; - const event = await this.prisma.httpWebhookEvent.create({ - data: { eventId: `evt_${randomUUID()}`, tenantId: data.tenantId, applicationId: data.applicationId, eventType: data.eventType, messageRecordId: data.messageRecordId, messageId: data.messageId, uplinkMessageId: data.uplinkMessageId, payload: data.payload as Prisma.InputJsonValue }, + const eventId = data.eventType === 'receipt' && data.messageRecordId + ? `evt_receipt_${data.messageRecordId}` + : data.eventType === 'uplink' && data.uplinkMessageId + ? `evt_uplink_${data.uplinkMessageId}` + : `evt_${randomUUID()}`; + const event = await this.prisma.httpWebhookEvent.upsert({ + where: { eventId }, + update: {}, + create: { eventId, tenantId: data.tenantId, applicationId: data.applicationId, eventType: data.eventType, messageRecordId: data.messageRecordId, messageId: data.messageId, uplinkMessageId: data.uplinkMessageId, payload: data.payload as Prisma.InputJsonValue }, }); - const delivery = await this.prisma.httpWebhookDelivery.create({ data: { eventId: event.id, endpointId: endpoint.id } }); + const delivery = await this.prisma.httpWebhookDelivery.upsert({ + where: { eventId_endpointId: { eventId: event.id, endpointId: endpoint.id } }, + update: {}, + create: { eventId: event.id, endpointId: endpoint.id }, + }); + if (delivery.status === 'delivered') return delivery; await this.queue?.add('deliver', { deliveryId: delivery.id }, { jobId: delivery.id, removeOnComplete: 1000, removeOnFail: 1000 }); return delivery; } diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 03cec1c..1ad12c3 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -1,4 +1,5 @@ import { createHash } from 'node:crypto'; +import { Prisma } from '@prisma/client'; import { BillingService } from '../billing/billing.service'; import { RiskReviewService } from '../risk-review/risk-review.service'; import { SendChainService } from './send-chain.service'; @@ -153,7 +154,11 @@ function createPrismaMock() { create: jest.fn().mockResolvedValue({ id: 'submit-1' }), updateMany: jest.fn().mockResolvedValue({ count: 1 }), findFirst: jest.fn().mockResolvedValue({ id: 'submit-1', submitId: 'SUB-1', submitStatus: 'accepted', createdAt: new Date('2026-07-01T10:00:00.000Z') }), - findUnique: jest.fn().mockResolvedValue({ id: 'submit-1', submitId: 'SUB-1', submitStatus: 'accepted' }), + findUnique: jest.fn().mockImplementation(({ where }) => Promise.resolve( + where.retryOfSubmitRecordId + ? null + : { id: 'submit-1', messageRecordId: 'record-1', channelId: 'channel-1', submitId: 'SUB-1', submitStatus: 'accepted' }, + )), count: jest.fn().mockResolvedValue(1), findMany: jest.fn().mockResolvedValue([]), }, @@ -1971,7 +1976,7 @@ describe('SendChainService', () => { }, }); prisma.smsSubmitRecord.findMany.mockResolvedValue([ - { id: 'submit-1', submitId: 'SUB-1', channelId: primary.id, createdAt: new Date() }, + { id: 'submit-1', submitId: 'SUB-1', messageRecordId: 'record-1', channelId: primary.id, createdAt: new Date() }, ]); const submitMessageToGateway = jest.spyOn(service as any, 'submitMessageToGateway') .mockResolvedValue({ submitted: true, messageRecordId: 'record-1', channelId: backup.id, attempt: 1 }); @@ -1997,9 +2002,67 @@ describe('SendChainService', () => { expect.objectContaining({ signatureId: 'sig-direct' }), expect.objectContaining({ channel: expect.objectContaining({ id: backup.id }) }), 1, + 'submit-1', ); }); + it('allows only one retry submit when three long-message failure receipts race', async () => { + const { service, prisma } = createService(); + const channel = await prisma.smsChannel.findUnique(); + const queueAdd = jest.fn().mockResolvedValue(undefined); + jest.spyOn(service as any, 'getGatewayQueue').mockReturnValue({ add: queueAdd }); + jest.spyOn(service as any, 'waitForChannelRateLimit').mockResolvedValue(undefined); + let claimedRetry: Record | null = null; + prisma.smsSubmitRecord.create.mockImplementation(async ({ data }) => { + if (data.retryOfSubmitRecordId) { + if (claimedRetry) { + throw new Prisma.PrismaClientKnownRequestError('duplicate retry claim', { + code: 'P2002', + clientVersion: '7.9.0', + meta: { target: ['retryOfSubmitRecordId'] }, + }); + } + claimedRetry = { id: 'retry-submit-1', ...data }; + return claimedRetry; + } + return { id: 'submit-1', ...data }; + }); + prisma.smsSubmitRecord.findUnique.mockImplementation(({ where }) => Promise.resolve( + where.retryOfSubmitRecordId ? claimedRetry : null, + )); + const message = { + id: 'record-long-race', + tenantId: 'tenant-1', + batchTaskId: 'task-1', + applicationId: 'app-1', + templateId: 'tpl-1', + messageId: 'MSG-LONG-RACE', + phoneNumber: '13800000001', + content: '长短信'.repeat(136), + billingUnits: 3, + template: { signature: { id: 'sig-1', name: '签名' } }, + }; + const routed = { + channel, + groupId: 'group-1', + carrier: 'mobile', + province: '山东', + routeScope: 'national', + }; + + const results = await Promise.all([ + (service as any).submitMessageToGateway(message, routed, 1, 'source-submit-1'), + (service as any).submitMessageToGateway(message, routed, 1, 'source-submit-1'), + (service as any).submitMessageToGateway(message, routed, 1, 'source-submit-1'), + ]); + + expect(results.filter((result) => result.submitted)).toHaveLength(1); + expect(results.filter((result) => result.duplicateRetry)).toHaveLength(2); + expect(queueAdd).toHaveBeenCalledTimes(1); + expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledTimes(1); + expect(prisma.smsMessageRecord.update).toHaveBeenCalledTimes(1); + }); + it('releases reservation for rejected submit result and refunds failed receipts', async () => { const { service, prisma, billing } = createService(); prisma.smsBillingRecord.findFirst @@ -2031,7 +2094,10 @@ describe('SendChainService', () => { receiptStatus: 'undelivered', rawStatus: 'UNDELIV', }); - expect(billing.refund).toHaveBeenCalledWith(expect.objectContaining({ remark: '最终失败退款' })); + expect(billing.refund).toHaveBeenCalledWith(expect.objectContaining({ + idempotencyKey: 'sms-refund:MSG-1', + remark: '最终失败退款', + })); }); it('stops failed receipt retry after the configured minute limit', async () => { @@ -2403,6 +2469,56 @@ describe('SendChainService', () => { expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(1); }); + it('creates and sends only one downstream final receipt under concurrent completion', async () => { + const { service, prisma } = createService(); + let claimedDelivery: Record | null = null; + prisma.cmppDownstreamDelivery.create.mockImplementation(async ({ data }) => { + if (claimedDelivery) { + throw new Prisma.PrismaClientKnownRequestError('duplicate downstream receipt', { + code: 'P2002', + clientVersion: '7.9.0', + meta: { target: ['dedupeKey'] }, + }); + } + claimedDelivery = { + id: 'delivery-once', + ...data, + createdAt: new Date(), + updatedAt: new Date(), + }; + return claimedDelivery; + }); + prisma.cmppDownstreamDelivery.findUnique.mockImplementation(({ where }) => Promise.resolve( + where.dedupeKey || where.id === 'delivery-once' ? claimedDelivery : null, + )); + const payload = { + tenantId: 'tenant-1', + applicationId: 'app-1', + messageRecordId: 'record-long-race', + messageId: 'MSG-LONG-RACE', + deliveryType: 'receipt' as const, + payload: { + messageId: 'MSG-LONG-RACE', + receiptStatus: 'undelivered', + rawStatus: 'FLNIGLK', + }, + }; + + const results = await Promise.all([ + (service as any).queueAndTryDownstreamDelivery(payload), + (service as any).queueAndTryDownstreamDelivery(payload), + (service as any).queueAndTryDownstreamDelivery(payload), + ]); + + expect(results.map((result) => result.id)).toEqual([ + 'delivery-once', + 'delivery-once', + 'delivery-once', + ]); + expect(service['postGatewayControl']).toHaveBeenCalledTimes(1); + expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(3); + }); + it('marks a long message failed when a non-primary segment returns an explicit failure', async () => { const { service, prisma, billing } = createService(); prisma.smsMessageRecord.findUnique.mockResolvedValue({ @@ -2434,6 +2550,17 @@ describe('SendChainService', () => { { segmentIndex: 1, segmentTotal: 2, receiptStatus: null, rawStatus: null, deliveredAt: null }, { segmentIndex: 2, segmentTotal: 2, receiptStatus: 'undelivered', rawStatus: 'YL:1014', deliveredAt: new Date() }, ]); + prisma.smsSubmitRecord.findUnique.mockImplementation(({ where }) => Promise.resolve( + where.id === 'submit-long' + ? { + id: 'submit-long', + messageRecordId: 'record-long', + channelId: 'channel-1', + submitId: 'SUB-LONG-FAIL', + submitStatus: 'accepted', + } + : null, + )); prisma.smsBillingRecord.findFirst .mockResolvedValueOnce(null) .mockResolvedValueOnce({ id: 'billing-charged', billingStatus: 'charged' }); diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 8b737b6..b38805e 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -1108,7 +1108,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } } else if (data.submitStatus !== 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) { const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string }; - const retried = await this.retryMessageIfAllowed(businessMessage, data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发'); + const retried = await this.retryMessageIfAllowed( + businessMessage, + data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发', + submitRecord.id, + ); if (retried) { await this.refreshTaskProgress(businessMessage.batchTaskId); return retried; @@ -1465,7 +1469,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId; if (status === 'failed' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) { const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string }; - const retried = await this.retryMessageIfAllowed(businessMessage, '回执失败补发'); + const retried = await this.retryMessageIfAllowed( + businessMessage, + '回执失败补发', + resolved.submitRecordId, + ); if (retried) { await this.refreshTaskProgress(businessMessage.batchTaskId); return retried; @@ -2400,21 +2408,51 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return null; } const payload = { account: application?.cmppAccount, applicationId: data.applicationId, ...data.payload }; - const delivery = await this.prisma.cmppDownstreamDelivery.create({ - data: { - tenantId: data.tenantId, - applicationId: data.applicationId, - messageRecordId: data.messageRecordId, - messageId: data.messageId, - deliveryType: data.deliveryType, - payload, - retryEnabled: deliveryAllowed && (data.deliveryType === 'uplink' - ? application?.downstreamUplinkRetryEnabled ?? true - : application?.downstreamReceiptRetryEnabled ?? true), - status: deliveryAllowed ? 'pending' : 'abandoned', - lastError: deliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送', - }, - }); + const dedupeKey = data.deliveryType === 'receipt' && data.messageRecordId + ? `receipt:${data.messageRecordId}` + : data.deliveryType === 'uplink' && typeof data.payload.uplinkMessageId === 'string' + ? `uplink:${data.payload.uplinkMessageId}` + : null; + let delivery; + try { + delivery = await this.prisma.cmppDownstreamDelivery.create({ + data: { + tenantId: data.tenantId, + applicationId: data.applicationId, + messageRecordId: data.messageRecordId, + messageId: data.messageId, + dedupeKey, + deliveryType: data.deliveryType, + payload, + retryEnabled: deliveryAllowed && (data.deliveryType === 'uplink' + ? application?.downstreamUplinkRetryEnabled ?? true + : application?.downstreamReceiptRetryEnabled ?? true), + status: deliveryAllowed ? 'pending' : 'abandoned', + lastError: deliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送', + }, + }); + } catch (error) { + if ( + dedupeKey + && error instanceof Prisma.PrismaClientKnownRequestError + && error.code === 'P2002' + ) { + const existing = await this.prisma.cmppDownstreamDelivery.findUnique({ + where: { dedupeKey }, + }); + if (existing) { + this.logger.warn(`downstream_delivery_deduplicated ${JSON.stringify({ + deliveryType: data.deliveryType, + messageRecordId: data.messageRecordId, + messageId: data.messageId, + dedupeKey, + deliveryId: existing.id, + })}`); + return existing; + } + } + throw error; + } if (!deliveryAllowed) { return delivery; } @@ -3302,6 +3340,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationId?: string | null; templateId?: string | null; signatureId?: string | null; + submitId?: string | null; messageId: string; phoneNumber: string; content: string; @@ -3314,44 +3353,87 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }, routed: RoutedChannel, attempt: number, + retryOfSubmitRecordId?: string, ) { const channel = routed.channel; const upstreamSrcId = composeUpstreamSrcId(channel.srcId, message.applicationExtension); await this.ensureSignatureReportedForChannel(message, channel.id); await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond); const submitId = `SUB-${randomUUID()}`; - const session = await this.prisma.cmppSubmitSession.upsert({ - where: { sessionNo: `OPEN-${channel.id}` }, - update: { submitTotal: { increment: 1 } }, - create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 }, - }); - await this.prisma.smsSubmitRecord.create({ - data: { - tenantId: message.tenantId, - batchTaskId: message.batchTaskId, - messageRecordId: message.id, - channelId: channel.id, - sessionId: session.id, - submitId, - submitStatus: 'queued', - costUnitPrice: channel.unitPrice ?? 0, - costAmountCents: moneyToNumber(channel.unitPrice) * Math.max(1, message.billingUnits ?? 1), - }, - }); - await this.prisma.smsMessageRecord.update({ - where: { id: message.id }, - data: { - channelId: channel.id, - carrier: routed.carrier, - province: routed.province, - submitId, - status: 'submit_queued', - submitStatus: 'queued', - receiptStatus: null, - errorCode: null, - errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined, - }, - }); + try { + await this.prisma.$transaction(async (tx) => { + const session = await tx.cmppSubmitSession.upsert({ + where: { sessionNo: `OPEN-${channel.id}` }, + update: { submitTotal: { increment: 1 } }, + create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 }, + }); + await tx.smsSubmitRecord.create({ + data: { + tenantId: message.tenantId, + batchTaskId: message.batchTaskId, + messageRecordId: message.id, + channelId: channel.id, + sessionId: session.id, + retryOfSubmitRecordId, + submitId, + submitStatus: 'queued', + costUnitPrice: channel.unitPrice ?? 0, + costAmountCents: moneyToNumber(channel.unitPrice) * Math.max(1, message.billingUnits ?? 1), + }, + }); + await tx.smsMessageRecord.update({ + where: { id: message.id }, + data: { + channelId: channel.id, + carrier: routed.carrier, + province: routed.province, + submitId, + status: 'submit_queued', + submitStatus: 'queued', + receiptStatus: null, + errorCode: null, + errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined, + }, + }); + }); + if (retryOfSubmitRecordId) { + this.logger.log(`sms_retry_claim_acquired ${JSON.stringify({ + messageId: message.messageId, + messageRecordId: message.id, + retryOfSubmitRecordId, + submitId, + channelId: channel.id, + })}`); + } + } catch (error) { + if ( + retryOfSubmitRecordId + && error instanceof Prisma.PrismaClientKnownRequestError + && error.code === 'P2002' + ) { + const existingRetry = await this.prisma.smsSubmitRecord.findUnique({ + where: { retryOfSubmitRecordId }, + }); + if (existingRetry) { + this.logger.warn(`sms_retry_claim_reused ${JSON.stringify({ + messageId: message.messageId, + messageRecordId: message.id, + retryOfSubmitRecordId, + submitId: existingRetry.submitId, + channelId: existingRetry.channelId, + })}`); + return { + submitted: false, + duplicateRetry: true, + messageRecordId: message.id, + channelId: existingRetry.channelId, + attempt, + submitId: existingRetry.submitId, + }; + } + } + throw error; + } const command = { schemaVersion: 'v1', messageType: 'SubmitCommand', @@ -3414,6 +3496,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { batchTaskId: string; applicationId?: string | null; templateId?: string | null; + signatureId?: string | null; + submitId?: string | null; messageId: string; phoneNumber: string; content: string; @@ -3423,6 +3507,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationExtension?: string | null; }, reason: string, + sourceSubmitRecordId?: string, ) { const attempts = await this.prisma.smsSubmitRecord.findMany({ where: { messageRecordId: message.id }, @@ -3430,6 +3515,38 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { take: 200, }); const attemptedChannelIds = attempts.map((attempt) => attempt.channelId); + let sourceAttempt = sourceSubmitRecordId + ? attempts.find((attempt) => attempt.id === sourceSubmitRecordId) + : attempts.find((attempt) => attempt.submitId === message.submitId) ?? attempts[attempts.length - 1]; + if (!sourceAttempt && sourceSubmitRecordId) { + sourceAttempt = await this.prisma.smsSubmitRecord.findUnique({ + where: { id: sourceSubmitRecordId }, + }) ?? undefined; + } + if (!sourceAttempt || sourceAttempt.messageRecordId !== message.id) { + this.logger.error(`sms_retry_route_failed ${JSON.stringify({ + messageId: message.messageId, + messageRecordId: message.id, + reason, + sourceSubmitRecordId, + sourceMessageRecordId: sourceAttempt?.messageRecordId, + error: sourceAttempt ? 'retry_source_submit_record_mismatch' : 'retry_source_submit_record_missing', + })}`); + return null; + } + const existingRetry = await this.prisma.smsSubmitRecord.findUnique({ + where: { retryOfSubmitRecordId: sourceAttempt.id }, + }); + if (existingRetry) { + this.logger.warn(`sms_retry_claim_reused ${JSON.stringify({ + messageId: message.messageId, + messageRecordId: message.id, + retryOfSubmitRecordId: sourceAttempt.id, + submitId: existingRetry.submitId, + channelId: existingRetry.channelId, + })}`); + return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); + } const ageMinutes = (Date.now() - new Date(message.queuedAt ?? Date.now()).getTime()) / 60_000; this.logger.log(`sms_retry_route_started ${JSON.stringify({ messageId: message.messageId, @@ -3469,7 +3586,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { where: { id: message.id }, data: { errorMessage: reason }, }); - const retried = await this.submitMessageToGateway(message, routed, attempts.length); + const retried = await this.submitMessageToGateway( + message, + routed, + attempts.length, + sourceAttempt.id, + ); this.logger.log(`sms_retry_route_selected ${JSON.stringify({ messageId: message.messageId, messageRecordId: message.id, @@ -3985,6 +4107,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { await this.billing.release({ tenantId: message.tenantId, amountCents, + idempotencyKey: `sms-charge-release:${message.messageId}`, relatedType: 'sms_batch_task', relatedId: message.batchTaskId, remark: `短信 ${message.messageId} 提交成功释放冻结并转扣费`, @@ -3993,6 +4116,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { const transaction = await this.billing.charge({ tenantId: message.tenantId, amountCents, + idempotencyKey: `sms-charge:${message.messageId}`, relatedType: 'sms_message_record', relatedId: message.messageId, remark: '提交成功扣费', @@ -4038,6 +4162,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { await this.billing.release({ tenantId: message.tenantId, amountCents, + idempotencyKey: `sms-reservation-release:${message.messageId}`, relatedType: 'sms_message_record', relatedId: message.messageId, remark: `${remark}: ${message.messageId}`, @@ -4063,6 +4188,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { const transaction = await this.billing.refund({ tenantId: message.tenantId, amountCents, + idempotencyKey: `sms-refund:${message.messageId}`, relatedType: 'sms_message_record', relatedId: message.messageId, remark, diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index bff0383..56b59e0 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -1750,3 +1750,12 @@ 3. 迟到的旧尝试结果只允许更新其对应提交尝试和分片审计，不得覆盖短信主记录当前尝试的通道、上游消息号或终态。 4. 发送详情中的“通道发送与回执”必须优先按`SmsMessageSegmentAudit.submitId`重建逐次尝试，显示每次真实通道、发送时间、提交结果及各分片回执；不得用短信主记录最终通道回填所有历史尝试。 5. 通道组成员顺序、优先级、权重或主备关系变更必须写操作审计，保存修改前后有序成员及真实通道名称，便于解释某条短信发送当时使用的路由配置。 + +## 2026-07-26 长短信失败并发补发与账务幂等补充要求 + +1. 同一`SmsSubmitRecord`无论收到多少个分片失败回执、重复聚合结果或并发工作线程，只允许创建一个下一跳补发记录。数据库必须以来源提交记录建立唯一补发关系，不能只依靠进程内锁、先查后建或短信主记录当前状态判断。 +2. 任一分片明确失败仍可及时判定本次长短信尝试失败，不要求等待其余失败回执；后续分片回执继续完整落库和写通讯日志，但只能复用已取得的补发决定，不得再次向Gateway发布提交命令。 +3. 短信扣费、冻结释放和最终失败退款必须使用稳定业务幂等键。企业账户余额变更在数据库事务内按企业串行并使用原子增量，禁止读取旧余额后覆盖写入；相同幂等键重复调用必须返回原交易，不得再次改变余额。 +4. 同一短信记录只允许生成一条最终回执投递。CMPP下游投递使用数据库唯一去重键，HTTP Webhook使用稳定事件号和事件/端点唯一关系；重复终态处理返回原投递，不得再次向客户发送。 +5. 补发抢占成功、并发复用及下游投递去重必须写结构化日志，至少包含平台消息号、短信记录、来源提交记录、下一跳`submitId`、通道和复用的投递记录。 +6. 历史重复提交、通讯报文、回执、退款及客户ACK属于事故审计证据，不得在功能migration中删除或覆盖。历史余额修正必须先完成账户与流水专项对账，再通过可审计冲正处理。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 50c06e2..ba109c6 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3895,3 +3895,15 @@ npm run verify:phase8 | TC-SUBMIT-ATTR-005 | 两分片长短信在通道A失败后由通道B补发 | 每个分片结果归属正确`submitId`和通道；任一迟到结果不污染另一尝试 | | TC-SMS-DETAIL-006 | 查看先经富泷失败、再经铁布衫失败的历史短信详情 | “通道发送与回执”显示两次真实通道及各自分片回执，不把两行都显示为最终通道 | | TC-CHANNEL-GROUP-007 | 调整通道组成员顺序、优先级、权重或主备 | 操作日志保存修改前后有序成员、通道编号和名称，可还原短信发送时配置 | + +## 2026-07-26 长短信并发失败幂等用例 + +| 用例编号 | 场景 | 预期结果 | +|---|---|---| +| TC-RETRY-RACE-001 | 三分片长短信的三个失败回执并发进入API，备用通道可用 | 三个回执和通讯报文全部保存；来源提交记录只关联一个补发记录，只发布一个Gateway命令、三个补发分片 | +| TC-RETRY-RACE-002 | 三个线程在唯一补发记录提交前后交错执行 | 只有一个线程取得`retryOfSubmitRecordId`唯一关系；其他线程返回同一下一跳`submitId`并写复用日志，不退款、不生成最终回执 | +| TC-RETRY-RACE-003 | 三个失败回执并发处理且没有可用备用通道 | 主记录最终失败；只产生一笔退款交易和一次余额增量，只生成一个CMPP最终失败回执及一个HTTP回调事件 | +| TC-BILLING-IDEM-004 | 三个线程使用同一短信退款幂等键并发退款 | 三次调用返回同一交易ID，`AccountTransaction`只有一条，账户余额只增加一次 | +| TC-BILLING-ATOMIC-005 | 同一企业同时发生扣费、退款和充值 | 账户级事务锁串行化余额变更，使用数据库原子增量；每条流水`balanceAfter`连续且最终余额与流水一致 | +| TC-DOWNSTREAM-IDEM-006 | 同一短信终态被重复处理，企业同时启用CMPP和HTTP | CMPP只有一条`CmppDownstreamDelivery`且只发送一次；HTTP只有一个稳定事件和一条端点投递 | +| TC-MIGRATION-IDEM-007 | 在含历史重复最终回执的预生产数据上执行migration | 历史行全部保留；每个短信只给最早一条历史回执设置唯一键，其余保持空键；新数据开始强制唯一 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 3f6026f..eabf7be 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -2529,3 +2529,13 @@ git diff --check - 标准部署脚本完成两套干净依赖安装、安全门禁、Prisma生成/迁移检查、前端/API/Gateway构建和服务重启；72条migration齐全且无待执行项，`.deployed-commit=0857de09d82aec7233108fb6700c40a6183c21f8`。生产依赖审计仍报告前端2个high（未使用的React Router RSC路径）和API 3个moderate（Prisma CLI工具链），自动安全门禁确认既定缓解继续有效，未执行破坏兼容性的`audit fix --force`。 - 发布后API、Gateway、Nginx、PostgreSQL、MinIO均active，Redis`PONG`；`12026/17890/8090/3000/6379/5432/9000`均监听，内外健康页及运营/客户端入口HTTP 200，公网CMPP 17890可连接；Redis提交流消费者1、`pending=0`、`lag=0`，发布后API/Gateway无error级日志。 - Gateway重启时供应商曾对“会员营销-富泷”首次登录返回`auth failed`，系统按鉴权失败5分钟慢重试策略等待而未高频重连；21:42:17自动重试成功，最终4个启用通道全部`connected/currentConnections=1/desiredConnections=1`。历史17:53短信数据库只读验证仍为富泷两个`FLBLACK`分片、铁布衫两个`WL:FSNM`分片，证明部署未改写证据且新详情具备正确重建数据。 + +## 2026-07-26 `MSG-65a47514`长短信并发重复补发P0修复（发布前） + +- 预生产只读证据确认`MSG-65a47514-7f30-49af-8113-b76747550199`正文408字、3个计费/协议分片。12:33:33经会员营销-铁布衫提交一次；12:58:17三个`WL:CGMT`失败回执在9毫秒内到达，三个处理线程分别触发整条补发，在约13毫秒内创建三个富泷`submitId`，形成“铁布衫1次+富泷3次”、共4次整条尝试和12个真实Submit分片。所有尝试均为失败回执，没有`DELIVRD`成功证据。 +- 并发影响还包括三个已被客户`Result=0`确认的最终失败回执、三条各1053分的退款流水，以及非原子余额覆盖导致的实际多退。通道尝试记录成本为铁布衫993分、富泷3168分；是否形成供应商真实账单需另行对账。本轮不删除历史提交、回执、投递、ACK或账务证据，不自动冲正现有余额。 +- 新migration`20260726223000_prevent_duplicate_retry_side_effects`为`SmsSubmitRecord`增加唯一`retryOfSubmitRecordId`自关联、为`AccountTransaction`增加唯一`idempotencyKey`、为`CmppDownstreamDelivery`增加唯一`dedupeKey`。历史重复下游回执只给每条短信最早的回执设置稳定键，其余历史行原样保留。 +- 补发记录、提交会话计数和短信当前尝试改为同一数据库事务创建；相同来源提交记录的并发线程只有一个能取得下一跳，其他线程读取并复用已存在的补发，禁止再次向Gateway发布。缺少可审计来源提交记录时安全停止补发并写结构化错误。 +- 企业账户通用余额变更改为企业级PostgreSQL事务锁加原子`increment`；短信扣费、提交成功释放冻结、最终释放和退款使用平台消息号稳定幂等键。三次并发退款专项用例返回同一交易ID，只创建一条流水并只改变一次余额。 +- CMPP最终回执使用`receipt:{messageRecordId}`唯一键；HTTP回执事件使用`evt_receipt_{messageRecordId}`稳定事件号并复用事件/端点唯一投递。并发专项用例证明三个完成线程只向Gateway发送一次最终回执；所有抢占、复用和投递去重均写结构化日志。 +- 当前专项门禁：发送链、账务和HTTP API共3 suites / 121 tests通过，包含三个长短信失败处理线程并发抢占、三次并发退款及三次最终回执创建。全量API回归26 suites / 346 tests通过，API TypeScript构建检查、Prisma schema校验、前端TypeScript/Vite/生产依赖安全门禁及Gateway Go测试/vet均通过；提交、推送和预生产部署结果待完成后补记。