From c20c2246b263b22a6a84c8ac10d9f320197a8fb6 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Fri, 18 Sep 2026 13:20:03 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=9B=BA=E5=8C=96=E9=95=BF=E7=9F=AD?= =?UTF-8?q?=E4=BF=A1=E5=9B=9E=E6=89=A7=E7=BB=88=E6=80=81=E5=B9=B6=E9=9A=94?= =?UTF-8?q?=E7=A6=BB=E5=8F=91=E9=80=81=E5=B0=9D=E8=AF=95=E5=BD=92=E5=B1=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../receipt-attempt-resolver.spec.ts | 109 ++++++ .../send-chain/receipt-attempt-resolver.ts | 125 +++++++ api/src/send-chain/send-chain.service.spec.ts | 138 +++++--- .../send-chain/send-gateway-result.service.ts | 9 +- api/src/send-chain/send-receipt.service.ts | 212 +++--------- .../first-version-development-requirements.md | 5 + docs/long-sms-receipt-review-20260918.md | 48 +++ docs/phase-4-send-pipeline-redesign.md | 11 + docs/system-functional-test-cases.md | 15 + docs/testing-progress.md | 13 + tools/testing/verify-receipt-finality.mjs | 315 ++++++++++++++++++ 11 files changed, 791 insertions(+), 209 deletions(-) create mode 100644 api/src/send-chain/receipt-attempt-resolver.spec.ts create mode 100644 api/src/send-chain/receipt-attempt-resolver.ts create mode 100644 docs/long-sms-receipt-review-20260918.md create mode 100644 tools/testing/verify-receipt-finality.mjs diff --git a/api/src/send-chain/receipt-attempt-resolver.spec.ts b/api/src/send-chain/receipt-attempt-resolver.spec.ts new file mode 100644 index 0000000..86ef27e --- /dev/null +++ b/api/src/send-chain/receipt-attempt-resolver.spec.ts @@ -0,0 +1,109 @@ +import { resolveReceiptAttempt } from './receipt-attempt-resolver'; +import { PrismaService } from '../prisma/prisma.service'; + +const channel = { + account: 'supplier', + gatewayHost: 'localhost', + gatewayPort: 7890, + protocol: 'CMPP', + cmppVersion: '3.0', +}; +const message = { id: 'm', messageId: 'MSG', phoneNumber: '13800138000', tenantId: 'tenant' }; +const source = (id = 's', channelId = 'c') => ({ + id, + submitId: id, + channelId, + tenantId: 'tenant', + messageRecordId: 'm', + messageRecord: message, + channel, +}); +const fragment = (submit = source()) => ({ + messageRecordId: 'm', + messageRecord: message, + channelId: submit.channelId, + channel, + submitId: submit.submitId, + submitRecord: submit, +}); +const event = { + messageId: 'MSG', + channelId: 'c', + gatewayMessageId: 'GW', + phoneNumber: message.phoneNumber, + receiptStatus: 'delivered' as const, + rawStatus: 'DELIVRD', +}; +function fixture(submits: unknown[] = [], fragments: unknown[] = []) { + return { + smsMessageRecord: { findUnique: jest.fn().mockResolvedValue(message) }, + smsSubmitRecord: { + findMany: jest.fn().mockResolvedValue(submits), + findUnique: jest.fn().mockResolvedValue(source()), + }, + smsMessageSegmentAudit: { findMany: jest.fn().mockResolvedValue(fragments) }, + smsChannel: { findUnique: jest.fn().mockResolvedValue(channel) }, + }; +} +const resolve = (db: ReturnType, data = event) => + resolveReceiptAttempt(db as unknown as PrismaService, data); +describe('receipt attempt identity', () => { + it('deduplicates primary and fragment evidence for one attempt', async () => { + expect((await resolve(fixture([source()], [fragment()]))).submitRecordId).toBe('s'); + }); + it('rejects a primary ID colliding with another attempt fragment on the same channel', async () => { + await expect(resolve(fixture([source()], [fragment(source('other'))]))).rejects.toThrow('提交尝试关联'); + }); + it('selects the incoming channel, irrespective of newest candidate order', async () => { + expect((await resolve(fixture([source('new', 'other'), source()]))).submitRecordId).toBe('s'); + }); + it('rejects two same-channel submit candidates', async () => { + await expect(resolve(fixture([source(), source('other')]))).rejects.toThrow('提交尝试关联'); + }); + it('accepts one other connection of the same supplier', async () => { + expect((await resolve(fixture([], [fragment(source('s', 'other'))]))).channelId).toBe('other'); + }); + it('rejects ambiguous connections of the same supplier', async () => { + await expect(resolve(fixture([], [fragment(source('a', 'a')), fragment(source('b', 'b'))]))).rejects.toThrow( + '提交尝试关联', + ); + }); + it('does not silently accept changed supplier credentials captured by Inbox', async () => { + const db = fixture([source()]); + await expect( + resolveReceiptAttempt(db as unknown as PrismaService, event, { ...channel, account: 'old-supplier' }), + ).rejects.toThrow('提交尝试关联'); + }); + it('rejects an exact business ID with a different destination', async () => { + await expect(resolve(fixture([source()]), { ...event, phoneNumber: '13900139000' })).rejects.toThrow( + '提交尝试关联', + ); + }); + it('rejects malformed cross-tenant fragment relations', async () => { + await expect(resolve(fixture([], [fragment({ ...source(), tenantId: 'other' })]))).rejects.toThrow('提交尝试关联'); + }); + it('recovers a legacy fragment relation from its globally unique submit ID', async () => { + const db = fixture([], [{ ...fragment(), submitRecord: null }]); + expect((await resolve(db)).submitRecordId).toBe('s'); + expect(db.smsSubmitRecord.findUnique).toHaveBeenCalledWith({ where: { submitId: 's' } }); + }); + it('rejects a truncated candidate set instead of pretending it is unique', async () => { + await expect(resolve(fixture(Array.from({ length: 101 }, () => source())))).rejects.toThrow('提交尝试关联'); + }); + it('limits submit-response-loss recovery to one timed-out submit in 72 hours', async () => { + const db = fixture(); + db.smsSubmitRecord.findMany.mockResolvedValueOnce([]).mockResolvedValueOnce([source()]); + expect((await resolve(db)).submitRecordId).toBe('s'); + expect(db.smsSubmitRecord.findMany).toHaveBeenLastCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + submitStatus: 'timeout', + gatewayMessageId: null, + channelId: 'c', + messageRecordId: 'm', + }), + take: 2, + }), + ); + }); +}); diff --git a/api/src/send-chain/receipt-attempt-resolver.ts b/api/src/send-chain/receipt-attempt-resolver.ts new file mode 100644 index 0000000..3d86752 --- /dev/null +++ b/api/src/send-chain/receipt-attempt-resolver.ts @@ -0,0 +1,125 @@ +import { NotFoundException } from '@nestjs/common'; +import { SmsMessageRecord } from '@prisma/client'; +import { PrismaService } from '../prisma/prisma.service'; +import { GatewayReceiptEventDto } from './send-chain.contracts'; +import { isSameUpstreamEndpointIdentity } from './send-chain.helpers'; + +type UpstreamIdentity = { + account: string; + gatewayHost: string; + gatewayPort: number; + protocol: string; + cmppVersion: string; +}; +type Candidate = { + message: SmsMessageRecord; + messageId: string; + submitRecordId: string; + submitId: string; + channelId: string; + channel: UpstreamIdentity | null; +}; +const unmatched = () => new NotFoundException('回执缺少唯一且可信的提交尝试关联'); + +/** A supplier Msg_Id is not globally unique. Combine submit and fragment evidence + * before accepting a candidate; a fragment can collide with another attempt's + * primary Msg_Id, including on the same logical channel. */ +export async function resolveReceiptAttempt( + db: PrismaService, + data: GatewayReceiptEventDto, + identity?: UpstreamIdentity, +) { + if (!data.gatewayMessageId) throw unmatched(); + const exact = data.messageId ? await db.smsMessageRecord.findUnique({ where: { messageId: data.messageId } }) : null; + const phone = data.phoneNumber?.trim(); + if (exact && phone && exact.phoneNumber !== phone) throw unmatched(); + const scope = exact + ? { messageRecordId: exact.id } + : phone + ? { messageRecord: { phoneNumber: phone } } + : { channelId: data.channelId }; + const [submits, segments] = await Promise.all([ + db.smsSubmitRecord.findMany({ + where: { ...scope, gatewayMessageId: data.gatewayMessageId }, + include: { messageRecord: true, channel: true }, + take: 101, + }), + db.smsMessageSegmentAudit.findMany({ + where: { ...scope, gatewayMessageId: data.gatewayMessageId }, + include: { messageRecord: true, submitRecord: true, channel: true }, + take: 101, + }), + ]); + // A truncated set must never look unique after filtering. + if (submits.length > 100 || segments.length > 100) throw unmatched(); + const candidates = new Map(); + for (const submit of submits) + candidates.set(submit.id, { + message: submit.messageRecord, + messageId: submit.messageRecord.messageId, + submitRecordId: submit.id, + submitId: submit.submitId, + channelId: submit.channelId, + channel: submit.channel, + }); + for (const segment of segments) { + const submit = + segment.submitRecord ?? (await db.smsSubmitRecord.findUnique({ where: { submitId: segment.submitId } })); + if ( + !submit || + submit.messageRecordId !== segment.messageRecordId || + submit.channelId !== segment.channelId || + submit.tenantId !== segment.messageRecord.tenantId + ) + throw unmatched(); + candidates.set(submit.id, { + message: segment.messageRecord, + messageId: segment.messageRecord.messageId, + submitRecordId: submit.id, + submitId: submit.submitId, + channelId: submit.channelId, + channel: segment.channel, + }); + } + const all = [...candidates.values()]; + const direct = all.filter((c) => c.channelId === data.channelId); + if (direct.length > 1) throw unmatched(); + if (direct.length === 1) { + // Inbox supplies the identity captured at intake; changed channel credentials + // cannot silently reassign an older supplier's receipt. + if (identity && (!direct[0].channel || !isSameUpstreamEndpointIdentity(identity, direct[0].channel))) + throw unmatched(); + return direct[0]; + } + const incoming = identity ?? (await db.smsChannel.findUnique({ where: { id: data.channelId } })); + if (!incoming) throw unmatched(); + const shared = all.filter((c) => c.channel && isSameUpstreamEndpointIdentity(incoming, c.channel)); + if (shared.length > 1) throw unmatched(); + if (shared.length === 1 && (exact || phone)) return shared[0]; + if (!phone) throw unmatched(); + // Preserve the existing narrowly bounded recovery of one timed-out submission + // whose provider identity was not recorded before its first receipt arrived. + const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date(); + const legacy = await db.smsSubmitRecord.findMany({ + where: { + channelId: data.channelId, + gatewayMessageId: null, + submitStatus: 'timeout', + ...(exact ? { messageRecordId: exact.id } : {}), + submittedAt: { gte: new Date(deliveredAt.getTime() - 72 * 60 * 60 * 1000), lte: deliveredAt }, + messageRecord: { phoneNumber: phone }, + }, + include: { messageRecord: true, channel: true }, + take: 2, + }); + if (legacy.length !== 1 || (identity && !isSameUpstreamEndpointIdentity(identity, legacy[0].channel))) + throw unmatched(); + const source = legacy[0]; + return { + message: source.messageRecord, + messageId: source.messageRecord.messageId, + submitRecordId: source.id, + submitId: source.submitId, + channelId: source.channelId, + }; +} diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 43514ed..c919b79 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -466,6 +466,34 @@ function createPrismaMock() { return prisma; } +// Supply relational receipt evidence separately from aggregate result fixtures. +async function receiptEvidence( + prisma: ReturnType, + overrides: Record = {}, + attempt: Record = {}, +) { + const message = { ...(await prisma.smsMessageRecord.findUnique()), ...overrides }; + const prior = await prisma.smsSubmitRecord.findFirst(); + const source = { + ...prior, + messageRecordId: message.id, + tenantId: message.tenantId, + channelId: overrides.channelId ?? message.channelId, + submitId: message.submitId ?? prior.submitId, + messageRecord: message, + channel: await prisma.smsChannel.findUnique(), + ...attempt, + }; + prisma.smsSubmitRecord.findMany.mockResolvedValue([source]); + prisma.smsSubmitRecord.findUnique.mockImplementation(({ where }) => + Promise.resolve(where.retryOfSubmitRecordId ? null : source), + ); + const aggregate = prisma.smsMessageSegmentAudit.findMany; + prisma.smsMessageSegmentAudit.findMany = jest + .fn() + .mockImplementation((args) => (args.include?.messageRecord ? Promise.resolve([]) : aggregate(args))); +} + function createService(prisma = createPrismaMock(), openApi?: { queueWebhookEvent: jest.Mock }) { const billing = { estimateSmsCost: jest.fn().mockReturnValue({ @@ -3547,6 +3575,7 @@ describe('SendChainService', () => { expect.objectContaining({ remark: expect.stringContaining('提交失败释放冻结') }), ); + await receiptEvidence(prisma); await service.handleReceipt({ messageId: 'MSG-1', channelId: 'channel-1', @@ -3601,6 +3630,7 @@ describe('SendChainService', () => { }, }); + await receiptEvidence(prisma, { ...(await prisma.smsMessageRecord.findFirst()), submitId: 'SUB-1' }); await service.handleReceipt({ messageId: 'MSG-1', channelId: 'channel-1', @@ -3636,6 +3666,10 @@ describe('SendChainService', () => { unitPrice: 3, }); + await receiptEvidence(prisma, await prisma.smsMessageRecord.findFirst(), { + channelId: 'channel-old', + submitId: 'SUB-OLD', + }); await service.handleReceipt({ messageId: 'MSG-1', channelId: 'channel-old', @@ -3661,29 +3695,28 @@ describe('SendChainService', () => { it('matches receipt to a unique timed-out submit attempt when the upstream submit response was lost', async () => { const { service, prisma } = createService(); prisma.smsMessageRecord.findUnique.mockResolvedValue(null); - prisma.smsSubmitRecord.findMany - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([ - { - id: 'submit-timeout-1', + prisma.smsSubmitRecord.findMany.mockResolvedValueOnce([]).mockResolvedValueOnce([ + { + id: 'submit-timeout-1', + submitId: 'SUB-1', + messageRecordId: 'record-1', + channelId: 'channel-1', + gatewayMessageId: null, + submitStatus: 'timeout', + submittedAt: new Date('2026-07-01T10:00:00.000Z'), + messageRecord: { + id: 'record-1', + tenantId: 'tenant-1', + batchTaskId: 'task-1', + applicationId: 'app-1', + messageId: 'MSG-1', + phoneNumber: '13800000001', channelId: 'channel-1', gatewayMessageId: null, - submitStatus: 'timeout', - submittedAt: new Date('2026-07-01T10:00:00.000Z'), - messageRecord: { - id: 'record-1', - tenantId: 'tenant-1', - batchTaskId: 'task-1', - applicationId: 'app-1', - messageId: 'MSG-1', - phoneNumber: '13800000001', - channelId: 'channel-1', - gatewayMessageId: null, - status: 'timeout', - }, + status: 'timeout', }, - ]); + }, + ]); await service.handleReceipt({ messageId: 'receipt-123456789', @@ -3721,6 +3754,8 @@ describe('SendChainService', () => { prisma.smsSubmitRecord.findMany.mockResolvedValueOnce([ { id: 'submit-channel-b', + submitId: 'SUB-B', + messageRecordId: 'record-channel-b', channelId: 'channel-b', gatewayMessageId: 'SHARED-UPSTREAM-ID', messageRecord: { @@ -3737,6 +3772,12 @@ describe('SendChainService', () => { }, ]); + prisma.smsSubmitRecord.findUnique.mockResolvedValue({ + id: 'submit-channel-b', + submitId: 'SUB-B', + messageRecordId: 'record-channel-b', + channelId: 'channel-b', + }); await service.handleReceipt({ messageId: 'receipt-SHARED-UPSTREAM-ID', channelId: 'channel-b', @@ -3750,7 +3791,6 @@ describe('SendChainService', () => { expect(prisma.smsSubmitRecord.findMany).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ - channelId: 'channel-b', gatewayMessageId: 'SHARED-UPSTREAM-ID', messageRecord: { phoneNumber: '15601992925' }, }), @@ -3788,7 +3828,14 @@ describe('SendChainService', () => { submitRecordId: 'submit-original', channelId: 'channel-original', gatewayMessageId: '736070230367350788', - submitRecord: { id: 'submit-original', submitId: 'SUB-LONG-1' }, + messageRecordId: 'record-long', + submitRecord: { + id: 'submit-original', + submitId: 'SUB-LONG-1', + messageRecordId: 'record-long', + channelId: 'channel-original', + tenantId: 'tenant-1', + }, channel: { id: 'channel-original', account: 'C59748', @@ -3814,6 +3861,12 @@ describe('SendChainService', () => { ]) .mockResolvedValueOnce([]); + prisma.smsSubmitRecord.findUnique.mockResolvedValue({ + id: 'submit-original', + submitId: 'SUB-LONG-1', + messageRecordId: 'record-long', + channelId: 'channel-original', + }); await service.handleReceipt({ messageId: 'receipt-736070230367350788', channelId: 'channel-copy', @@ -3851,7 +3904,13 @@ describe('SendChainService', () => { submitId: 'SUB-ORIGINAL', channelId: 'channel-original', gatewayMessageId: 'SHARED-ID', - submitRecord: { id: 'submit-original', submitId: 'SUB-ORIGINAL' }, + messageRecordId: 'record-original', + submitRecord: { + id: 'submit-original', + submitId: 'SUB-ORIGINAL', + messageRecordId: 'record-original', + channelId: 'channel-original', + }, channel: { id: 'channel-original', account: 'C59748', @@ -3877,7 +3936,7 @@ describe('SendChainService', () => { receiptStatus: 'delivered', rawStatus: 'DELIVRD', }), - ).rejects.toThrow('SMS message record not found'); + ).rejects.toThrow('提交尝试关联'); expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled(); }); @@ -3925,6 +3984,7 @@ describe('SendChainService', () => { { segmentIndex: 2, segmentTotal: 2, receiptStatus: 'delivered', rawStatus: 'DELIVRD', deliveredAt: new Date() }, ]); + await receiptEvidence(prisma); await service.handleReceipt({ messageId: 'MSG-LONG-1', channelId: 'channel-1', @@ -4014,6 +4074,7 @@ describe('SendChainService', () => { }, ]); + await receiptEvidence(prisma); await service.handleReceipt({ messageId: 'MSG-MESSAGE-LEVEL', channelId: 'channel-1', @@ -4078,6 +4139,7 @@ describe('SendChainService', () => { { segmentIndex: 2, segmentTotal: 2, receiptStatus: 'undelivered', rawStatus: 'UNDELIV', deliveredAt: new Date() }, ]); + await receiptEvidence(prisma); await service.handleReceipt({ messageId: 'MSG-CONFLICT', channelId: 'channel-1', @@ -4238,6 +4300,7 @@ describe('SendChainService', () => { }, }); + await receiptEvidence(prisma); await service.handleReceipt({ messageId: 'MSG-LONG-FAIL', channelId: 'channel-1', @@ -4279,6 +4342,7 @@ describe('SendChainService', () => { rawStatus: 'DELIVRD', deliveredAt: '2026-07-01T10:01:00.000Z', }; + await receiptEvidence(prisma); await service.handleReceipt(receipt); await service.handleReceipt(receipt); @@ -4289,19 +4353,16 @@ describe('SendChainService', () => { it('rejects ambiguous receipt heuristic matches to avoid binding to the wrong message', async () => { const { service, prisma } = createService(); prisma.smsMessageRecord.findUnique.mockResolvedValue(null); - prisma.smsSubmitRecord.findMany - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([ - { - id: 'submit-timeout-1', - messageRecord: { id: 'record-1', messageId: 'MSG-1', phoneNumber: '13800000001' }, - }, - { - id: 'submit-timeout-2', - messageRecord: { id: 'record-2', messageId: 'MSG-2', phoneNumber: '13800000001' }, - }, - ]); + prisma.smsSubmitRecord.findMany.mockResolvedValueOnce([]).mockResolvedValueOnce([ + { + id: 'submit-timeout-1', + messageRecord: { id: 'record-1', messageId: 'MSG-1', phoneNumber: '13800000001' }, + }, + { + id: 'submit-timeout-2', + messageRecord: { id: 'record-2', messageId: 'MSG-2', phoneNumber: '13800000001' }, + }, + ]); await expect( service.handleReceipt({ @@ -4312,7 +4373,7 @@ describe('SendChainService', () => { receiptStatus: 'delivered', rawStatus: 'DELIVRD', }), - ).rejects.toThrow('SMS message record not found'); + ).rejects.toThrow('提交尝试关联'); expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled(); }); @@ -4424,6 +4485,7 @@ describe('SendChainService', () => { it('records receipts and uplink messages from gateway events', async () => { const { service, prisma } = createService(); + await receiptEvidence(prisma); await service.handleReceipt({ messageId: 'MSG-1', channelId: 'channel-1', diff --git a/api/src/send-chain/send-gateway-result.service.ts b/api/src/send-chain/send-gateway-result.service.ts index 5c27e44..bbdad55 100644 --- a/api/src/send-chain/send-gateway-result.service.ts +++ b/api/src/send-chain/send-gateway-result.service.ts @@ -190,7 +190,8 @@ export class SendGatewayResultService { data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed'; if ( data.submitStatus !== 'accepted' && - (message.status === 'delivered' || (message.status === 'timeout' && message.errorCode === 'RECEIPT_TIMEOUT')) + (['delivered', 'failed'].includes(message.status) || + (message.status === 'timeout' && message.errorCode === 'RECEIPT_TIMEOUT')) ) { await this.markSubmitResultProcessed(submitRecord.id, data.eventId, submittedAt); return message; @@ -381,10 +382,8 @@ export class SendGatewayResultService { const submitRecord = await this.prisma.smsSubmitRecord.findFirst({ where: { messageRecordId: message.id, - OR: [ - data.submitId ? { submitId: data.submitId } : undefined, - data.gatewayMessageId ? { gatewayMessageId: data.gatewayMessageId } : undefined, - ].filter(Boolean) as Array<{ submitId?: string; gatewayMessageId?: string }>, + channelId: data.channelId, + ...(data.submitId ? { submitId: data.submitId } : { gatewayMessageId: data.gatewayMessageId }), }, orderBy: { createdAt: 'desc' }, }); diff --git a/api/src/send-chain/send-receipt.service.ts b/api/src/send-chain/send-receipt.service.ts index 4ae2b7e..af0845c 100644 --- a/api/src/send-chain/send-receipt.service.ts +++ b/api/src/send-chain/send-receipt.service.ts @@ -1,3 +1,4 @@ +import { resolveReceiptAttempt } from './receipt-attempt-resolver'; import { completionContext } from './completion-context'; import { Logger, NotFoundException } from '@nestjs/common'; import { Prisma } from '@prisma/client'; @@ -14,7 +15,6 @@ import { DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, aggregateReceiptSegmentState, - isSameUpstreamEndpointIdentity, receiptEventKey, longMessageReceiptMode, } from './send-chain.helpers'; @@ -282,6 +282,34 @@ export class SendReceiptService { throw error; } const logicalReceipt = { ...data, channelId: logicalChannelId }; + // The message row is locked by AttemptCompletion. A committed final decision + // also commits accounting and notifications; later evidence cannot undo it. + const sameAttempt = + (!message.submitId || message.submitId === resolved.submitId) && + (!message.channelId || message.channelId === logicalChannelId); + const frozen = + ['failed', 'delivered'].includes(message.status) || + (message.status === 'timeout' && message.errorCode === 'RECEIPT_TIMEOUT'); + if (sameAttempt && frozen) { + const contradicts = + message.status === 'delivered' + ? !['delivered', 'unknown'].includes(data.receiptStatus) + : data.receiptStatus === 'delivered'; + if (contradicts) { + if (!existingReceipt) + await this.recordReceiptConflict({ + message, + submitRecordId: resolved.submitRecordId, + submitId: resolved.submitId, + receiptRecordId, + receiptKey, + data: logicalReceipt, + }); + } else if (data.receiptStatus !== 'unknown') { + await this.facade.recordReceiptSegment(message, logicalReceipt, deliveredAt, resolved.submitRecordId); + } + return message; + } await this.facade.recordReceiptSegment(message, logicalReceipt, deliveredAt, resolved.submitRecordId); const receiptMode = Number(message.billingUnits ?? 1) > 1 ? await this.getLongMessageReceiptMode(logicalChannelId) : 'per_segment'; @@ -455,7 +483,10 @@ export class SendReceiptService { receiptKey: input.receiptKey, gatewayMessageId: input.data.gatewayMessageId, phoneNumber: input.data.phoneNumber, - reason: 'message_level_success_followed_by_failure', + reason: + input.message.status === 'delivered' + ? 'message_level_success_followed_by_failure' + : 'final_failure_followed_by_success', }; await this.prisma.smsReceiptAnomaly.upsert({ where: { anomalyKey }, @@ -479,7 +510,8 @@ export class SendReceiptService { messageRecordId: input.message.id, submitRecordId: input.submitRecordId, receiptRecordId: input.receiptRecordId, - anomalyType: 'aggregate_success_then_failure', + anomalyType: + input.message.status === 'delivered' ? 'aggregate_success_then_failure' : 'final_failure_then_success', previousStatus: input.message.status, incomingStatus: input.data.receiptStatus, rawStatus: input.data.rawStatus, @@ -505,12 +537,21 @@ export class SendReceiptService { submitRecordId?: string, ) { const segmentAudits = this.facade.smsMessageSegmentAuditDelegate(); + const source = submitRecordId + ? await this.prisma.smsSubmitRecord.findUnique({ where: { id: submitRecordId } }) + : null; + if (submitRecordId && (!source || source.messageRecordId !== message.id || source.channelId !== data.channelId)) + throw new Error('completion_receipt_source_mismatch'); + if (!source) throw new NotFoundException('回执缺少可确认的提交尝试关联'); const updated = await segmentAudits.updateMany({ where: { messageRecordId: message.id, + channelId: source.channelId, + OR: [{ submitRecordId: source.id }, { submitRecordId: null, submitId: source.submitId }], gatewayMessageId: data.gatewayMessageId, }, data: { + submitRecordId: source.id, receiptStatus: data.receiptStatus, rawStatus: data.rawStatus, errorCode: data.errorCode ?? null, @@ -520,12 +561,7 @@ export class SendReceiptService { if (updated.count > 0) { return; } - const submitRecord = submitRecordId - ? await this.prisma.smsSubmitRecord.findUnique({ where: { id: submitRecordId } }) - : await this.prisma.smsSubmitRecord.findFirst({ - where: { messageRecordId: message.id, gatewayMessageId: data.gatewayMessageId }, - orderBy: { createdAt: 'desc' }, - }); + const submitRecord = source; await segmentAudits.upsert({ where: { messageRecordId_submitId_segmentIndex: { @@ -600,162 +636,6 @@ export class SendReceiptService { cmppVersion: string; }, ) { - const exactMessage = data.messageId - ? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } }) - : null; - if (exactMessage) { - const segmentAudit = data.gatewayMessageId - ? await this.facade.smsMessageSegmentAuditDelegate().findFirst({ - where: { - messageRecordId: exactMessage.id, - gatewayMessageId: data.gatewayMessageId, - }, - orderBy: { updatedAt: 'desc' }, - }) - : null; - if (segmentAudit) { - return { - message: exactMessage, - messageId: exactMessage.messageId, - submitRecordId: segmentAudit.submitRecordId ?? undefined, - submitId: segmentAudit.submitId, - channelId: segmentAudit.channelId ?? data.channelId, - }; - } - const submitRecord = await this.prisma.smsSubmitRecord.findFirst({ - where: { - messageRecordId: exactMessage.id, - channelId: data.channelId, - gatewayMessageId: data.gatewayMessageId, - }, - orderBy: { createdAt: 'desc' }, - }); - return { - message: exactMessage, - messageId: exactMessage.messageId, - submitRecordId: submitRecord?.id, - submitId: submitRecord?.submitId, - channelId: submitRecord?.channelId ?? data.channelId, - }; - } - - const phoneNumber = data.phoneNumber?.trim(); - const exactSubmits = await this.prisma.smsSubmitRecord.findMany({ - where: { - channelId: data.channelId, - gatewayMessageId: data.gatewayMessageId, - ...(phoneNumber ? { messageRecord: { phoneNumber } } : {}), - }, - include: { messageRecord: true }, - orderBy: { createdAt: 'desc' }, - take: 2, - }); - if (exactSubmits.length === 1 && exactSubmits[0]?.messageRecord) { - return { - message: exactSubmits[0].messageRecord, - messageId: exactSubmits[0].messageRecord.messageId, - submitRecordId: exactSubmits[0].id, - submitId: exactSubmits[0].submitId, - channelId: exactSubmits[0].channelId, - }; - } - - if (!phoneNumber) { - throw new NotFoundException('SMS message record not found'); - } - - const incomingChannel = - incomingIdentity ?? (await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } })); - if (!incomingChannel) { - throw new NotFoundException('SMS message record not found'); - } - const segmentMatches = await this.facade.smsMessageSegmentAuditDelegate().findMany({ - where: { - gatewayMessageId: data.gatewayMessageId, - messageRecord: { phoneNumber }, - }, - include: { messageRecord: true, submitRecord: true, channel: true }, - orderBy: { createdAt: 'desc' }, - take: 10, - }); - const exactSegmentMatches = segmentMatches.filter((candidate) => candidate.channelId === data.channelId); - if (exactSegmentMatches.length === 1 && exactSegmentMatches[0]?.messageRecord) { - return { - message: exactSegmentMatches[0].messageRecord, - messageId: exactSegmentMatches[0].messageRecord.messageId, - submitRecordId: exactSegmentMatches[0].submitRecordId ?? undefined, - submitId: exactSegmentMatches[0].submitRecord?.submitId ?? exactSegmentMatches[0].submitId, - channelId: exactSegmentMatches[0].channelId, - }; - } - const sameSupplierSegments = segmentMatches.filter( - (candidate) => candidate.channel && isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel), - ); - if (sameSupplierSegments.length === 1 && sameSupplierSegments[0]?.messageRecord) { - return { - message: sameSupplierSegments[0].messageRecord, - messageId: sameSupplierSegments[0].messageRecord.messageId, - submitRecordId: sameSupplierSegments[0].submitRecordId ?? undefined, - submitId: sameSupplierSegments[0].submitRecord?.submitId ?? sameSupplierSegments[0].submitId, - channelId: sameSupplierSegments[0].channelId, - }; - } - const crossConnectionSubmits = await this.prisma.smsSubmitRecord.findMany({ - where: { - gatewayMessageId: data.gatewayMessageId, - messageRecord: { phoneNumber }, - }, - include: { messageRecord: true, channel: true }, - orderBy: { createdAt: 'desc' }, - take: 10, - }); - const sameSupplierSubmits = crossConnectionSubmits.filter( - (candidate) => candidate.channel && isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel), - ); - if (sameSupplierSubmits.length === 1 && sameSupplierSubmits[0]?.messageRecord) { - return { - message: sameSupplierSubmits[0].messageRecord, - messageId: sameSupplierSubmits[0].messageRecord.messageId, - submitRecordId: sameSupplierSubmits[0].id, - submitId: sameSupplierSubmits[0].submitId, - channelId: sameSupplierSubmits[0].channelId, - }; - } - - const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date(); - const submittedAfter = new Date(deliveredAt.getTime() - 72 * 60 * 60 * 1000); - const candidates = await this.prisma.smsSubmitRecord.findMany({ - where: { - channelId: data.channelId, - gatewayMessageId: null, - submitStatus: 'timeout', - submittedAt: { - gte: submittedAfter, - lte: deliveredAt, - }, - messageRecord: { - phoneNumber, - }, - }, - include: { - messageRecord: true, - }, - orderBy: { - submittedAt: 'desc', - }, - take: 10, - }); - - if (candidates.length !== 1 || !candidates[0]?.messageRecord) { - throw new NotFoundException('SMS message record not found'); - } - - return { - message: candidates[0].messageRecord, - messageId: candidates[0].messageRecord.messageId, - submitRecordId: candidates[0].id, - submitId: candidates[0].submitId, - channelId: candidates[0].channelId, - }; + return resolveReceiptAttempt(this.prisma, data, incomingIdentity); } } diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index ab95474..b82f2b6 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -2369,3 +2369,8 @@ Webhook需在当前受支持Node运行时通过真实HTTPS投递；SSRF校验后 ## 2026-09-17 有效签名名称唯一性 同一企业、同一应用、相同完整签名名称只能存在一条有效签名；未绑定应用单独作为一个范围。有效状态包括草稿、待审、通过、驳回，停用及删除不占用名称；新增、改名、换应用、审核和恢复均不可绕过，接口及数据库同时防重。批量导入仍更新已有资料，保留未映射字段、用途及关联记录，并发创建后重查已有签名补资料。历史重复不自动删除或合并。设计见 [通道与报备方案](phase-4-channel-reporting-plan.md#2026-09-17-有效签名名称唯一性)。本轮授权代码修改及本地提交，不含推送和部署。 + + +## 2026-09-18 长短信回执终态与归属补充 + +最终失败（含明确回执超时）与成功必须保持消息、账务、客户通知一致。后续同次失败分片不重复选路或退款；矛盾成功/失败回执只留原始事实并生成异常，不自动改账或重发客户通知。unknown、缺分片、普通提交超时仍可接续。回执须按业务消息、手机号、逻辑通道/上游身份和唯一发送尝试共同匹配；相同Msg_Id不能跨尝试批量更新，有歧义留待匹配。设计见phase-4-send-pipeline-redesign.md第10.13节，测试见TC-RC-20260918-01～07。此次不改协议、数据库结构和线上历史数据。 diff --git a/docs/long-sms-receipt-review-20260918.md b/docs/long-sms-receipt-review-20260918.md new file mode 100644 index 0000000..e9ceac6 --- /dev/null +++ b/docs/long-sms-receipt-review-20260918.md @@ -0,0 +1,48 @@ +# 2026-09-18 长短信回执优化复核 + +范围：用户要求检查最近长短信回执优化仍有无Bug。本轮审查、隔离复现与预生产只读核对，不修改业务代码，不提交/推送/部署，不触发线上短信发送、补发、重投、账务或配置变更。实际应用1676cfe，关注phase-4-send-pipeline-redesign.md第10节。 + +## 确认问题（修复前1676cfe基线，行号为当时版本） + +### P1：最终失败后，矛盾成功回执可以改成成功，但不恢复账务或客户通知 + +位置：api/src/send-chain/send-receipt.service.ts:307～356。当前只有RECEIPT_TIMEOUT终态和delivered→failed方向保护，没有failed→delivered终态规则。recordReceiptSegment直接覆盖分片状态，当同次尝试的失败分片随后变成成功、其余分片也成功时，aggregate成为delivered，主消息直接更新成功。成功分支不处理已退款账务；CMPP dedupeKey与HTTP eventId固定，旧失败通知继续保留。 + +独立真实PG复现：构造合法的“最终失败＋refunded账单＋已生成失败通知”快照；同次两分片随后各到成功回执，通过实际SendChainService/耐久工作协调器处理。结果消息delivered、账单refunded、通知payload.receiptStatus=undelivered。事务和通知唯一键无法修复互相矛盾的终态规则。本轮使用真实持久层、实际回执/通知代码，不运行网络投递进程。该复现从已退款快照开始，不宣称执行了实际线上退款。 + +建议：明确并固化最终失败后的矛盾回执策略。若最终结果不可逆，保留审计并生成异常；若业务允许纠正，必须设计账务与通知的完整补偿，不可只改消息状态。 + +### P1：同一业务短信不同尝试复用上游Msg_Id时，回执可能跨通道匹配且同时更新两次尝试 + +位置：send-receipt.service.ts:604～620 exactMessage分支，仅按messageRecordId+gatewayMessageId选最新分片，未限定incoming channel/上游身份；494～519 recordReceiptSegment的updateMany同样未使用已传入submitRecordId或channelId。 + +独立PG复现两个不同通道、同业务短信两次尝试的相同gatewayMessageId，通过实际chain.handleReceipt入口：resolved.channelId指向另一通道，两次尝试的分片都被改成delivered（预期仅一条，无法唯一确认应拒绝匹配）。Msg_Id仅供应商作用域内有意义，不能以全平台无碰撞为前提。这会污染历史审计及当前成功汇总；工作表按sourceSubmitRecordId加锁不能修正此前错误归属。 + +建议：关联时结合逻辑通道/真实上游身份和唯一发送尝试；分片更新必须带resolved.submitRecordId（历史兼容使用可确认的submitId及身份），禁止一条回执批量跨尝试修改。 + +### P2：已经最终失败的后续分片仍重新执行补发选路 + +位置：send-receipt.service.ts:334～342、send-retry.service.ts:334～406。仅存在retryOfSubmitRecordId时复用已有补发，未保存“本次已确认无路可补、最终失败”的不可重复决策。后续合法失败分片成为新事件，会再次执行findApplicationRoute/selectChannelForMessage，并重新走终态副作用。新工作revision不意味着应该重复执行同一终态决策。 + +独立PG复现两分片先后失败：首片处理后消息已经failed，第二片仍调用选路；选路计数2。该用例仅将选路边界隔离为确定的BadRequest“无可用通道”，其余实际SendReceipt/SendRetry、工作表及数据库/通知路径真实执行；不是全量真实路由配置验收，不启动供应商网络。 + +预生产证据：2026-09-18 10:12～10:20，回调/发送worker stderr共719条sms_retry_route_failed、关联692条消息，错误均“无已报备通过且在线的可用通道”。21条消息重复2～3次，核验当前全部failed；逐条比对最早CMPP最终通知createdAt和日志秒时间，至少5条在最终通知已创建后仍出现选路失败日志。日志时间只有秒，不能据此否定其余同秒的重复。该问题增加查询/日志/事务开销，不能把1655次正常补发尝试全部归因于它，也不能量化其占CPU56.8%的比例。 + +建议：给发送尝试保存明确终态决策，后续事实仍审计/异常检测，但复用已提交的最终失败、退款及通知事实，不再选路。和第一项统一状态机处理，不能粗暴丢弃所有后到回执或unknown转明确结果。 + +## 已排除及线上边界 + +- 怀疑“第二片回执先于其SubmitSegmentResult造成永久丢失”未复现：实际外层缺少可靠sourceSubmit关联时抛出404，由Inbox等待；补齐分片元数据后重处理可正确齐段成功。保留初次该断言失败日志，最终作为通过用例而非Bug。 +- 以昨晚21:24上线后为边界查询：当前delivered且billingStatus=refunded计0；新分片中同messageRecordId+gatewayMessageId跨submitRecordId碰撞组0。只能说该窗口当前数据未发现上述两类命中，不是永久不可能发生，也不是全历史完整审计。 +- 当时18136个SmsAttemptCompletionWork全部idle；上一轮CPU诊断未发现旧的重复补发/下游回执唯一键冲突。上一轮原子认领与防重复创建机制有效，但不等同所有状态机/关联边界正确。 + +## 验证和保护 + +新建本机独立PostgreSQL16集群与cmpp_qa_receipt_review数据库，监听127.0.0.1:16439，全部111迁移成功；实际API TypeScript构建通过。未启动Gateway、HTTP投递、API调度或真实短信发送。首次尝试复用旧隔离集群时发现默认端口/角色不同，未修改旧数据库，关闭本轮启动的旧集群，改建独立集群；两套本轮启动进程均已关闭，数据及失败日志保留。 + +隔离证据.local-data/receipt-review-20260918/repro-final.log确认3项缺陷及1项通过；online.json、repeated-routes-stderr.json、route-confirm.json为只读现场聚合。日志ANSI导致首次关联统计为0，去除ANSI后重新核验21组/至少5组，原失败与修正结果明确区分。未重跑与只读审查无关的全量前端/Go测试，不宣称修复完成。 + + +## 后续整改（2026-09-18） + +用户后续授权修改并本地提交，三项按主设计第10.13节实施；上文保留原审查证据，不代表修复后实现。真实PG回归复现结果已改变：终态之后不重复选路，矛盾成功不改失败/退款/通知，跨尝试仅更新目标记录；详情、全部测试及未验证边界见testing-progress.md同日“长短信三项缺陷整改”。未推送、未部署，线上CPU改善尚未验证。 diff --git a/docs/phase-4-send-pipeline-redesign.md b/docs/phase-4-send-pipeline-redesign.md index 2943d3b..96449aa 100644 --- a/docs/phase-4-send-pipeline-redesign.md +++ b/docs/phase-4-send-pipeline-redesign.md @@ -352,3 +352,14 @@ GatewaySubmitOutbox a350aca测试环境长短信验收发现两条消息首尝试分别仅写出2/4、1/3段，Gateway在已收到即时应答时仍等待60秒后误判SUBMIT_TIMEOUT；补发及最终账务/通知收尾正常，但不能据此视作无异常验收。只读代码证据：submitPart在SendReqPkt、异步日志启动之后才登记pending[seq]，readLoop可能提前消费应答并因不存在等待者丢弃。新增真实TCP回归在修改前分别于CMPP2.0第206次、3.0第7次复现。 最小修复保持现有协议、接口、存储和补发策略：使用与heartbeat一致的mu→sendMu锁序，将连接有效性检查、写包和登记pending置于同一临界区，响应读取须等登记完成。网络失败仍走原连接关闭/失败处理。锁内不得执行日志、业务回调或数据库操作。验证两种协议各500次即时应答、Gateway全量test/vet及测试环境新的长短信样本；原异常证据保留，不将旧样本改成无补发成功。 + +### 10.13 2026-09-18 终态与回执归属整改（本轮授权修改并提交） + +本节修订10.2、10.5的实现约束，针对[专项复核](long-sms-receipt-review-20260918.md)三项问题；本轮不推送、部署或修正线上历史数据。 + +- 复用现有消息终态与SmsAttemptCompletionWork.decision作为已提交决定，不新增数据库结构。原有工作→消息事务锁保证最终失败、退款、通知和工作决定同时提交。消息failed/delivered和明确RECEIPT_TIMEOUT不再因后续供应商回执自动逆转；后续同向分片仅补审计，不再选路/退款/创建通知。unknown、未齐片、普通提交timeout仍可继续处理；旧尝试事实不修改当前尝试决定。 +- 同次终态的矛盾明确回执保留SmsCompletionEvent和SmsReceiptRecord原始事实，按消息尝试生成稳定异常键，不覆盖已用于结算的规范分片结果，不改账务或已生成客户通知。失败转成功异常独立标识；重复同一回执不能增加异常计数或重复业务动作。已有message_level成功后失败异常类型兼容。 +- 回执关联同时核验可用的业务消息身份、手机号、逻辑通道/真实上游身份、上游Msg_Id、发送尝试。优先精确逻辑通道；同供应商跨连接仅在相同账号/主机/端口/协议/版本且唯一候选时匹配。Submit与Segment候选须共同消歧，不能分别取最新。多个尝试冲突保留Inbox待匹配；候选超出查询上限时保守拒绝。历史缺submitRecordId时只按可确认submitId回读归属，不能猜当前尝试。 +- 分片写入限制messageRecordId、submitRecordId/明确submitId及channelId；不得仅凭messageRecordId+gatewayMessageId批量跨尝试更新。提交分片记录同样在明确submitId存在时优先精确匹配，避免OR条件被另一尝试同Msg_Id干扰。 +- API和协议不变、鉴权/租户规则不变、无新权限。冲突/未确认回执继续走现有Inbox恢复与人工排查。发布回退只涉及应用；不自动重放已完成事件或历史退款。 +- 验收：真实PG验证顺序/并发失败分片仅一次终态选路，矛盾回执账务/通知/分片均不逆转，跨通道与同通道碰撞拒绝或准确关联，同供应商跨连接、早到回执、unknown转成功、旧尝试迟到及工作故障恢复。隔离固定输入比较选路次数；不能将隔离开销降低推算为线上CPU降幅。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 21766a7..6c3873d 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -5678,3 +5678,18 @@ TC-SQA-01～14：真实隔离PG覆盖核心日期/日报/长短信/事务/分页 | 07 | 迁移遇历史有效重复明确失败并回滚；全部历史记录保留，没有留下半套索引；新库全部111迁移成功。 | 本机真实Nest/PG/Redis覆盖01、02、04～07及03恢复分支；四种有效/两种无效状态和错误分类另由定向单元测试覆盖。线上重复盘点、目标环境迁移及浏览器验收未执行，不等同已发布。 + + +## 2026-09-18 长短信终态与回执归属回归 + +| 编号 | 场景与预期 | 本地证据 | +|---|---|---| +| TC-RC-20260918-01 | 已最终失败后另一失败分片及并发重复到达；选路共一次，终态不重开 | verify-receipt-finality.mjs，真实PG、仅无可用路由边界隔离 | +| TC-RC-20260918-02 | 已失败/已退款/已有失败通知后两片成功；消息、账单、通知不变，原始回执及一条尝试异常保留，规范分片不改成功 | 同上，真实PG退款快照 | +| TC-RC-20260918-03 | 同业务不同通道尝试复用Msg_Id；仅命中通道的那次分片更新 | 同上，真实PG | +| TC-RC-20260918-04 | 同通道某尝试主Msg_Id与另一尝试分片Msg_Id碰撞；拒绝歧义，不写规范回执或分片 | 同上，真实PG | +| TC-RC-20260918-05 | 非首片回执先于提交分片元数据；先拒绝关联、补元数据后齐片成功；unknown仍可并发齐片完成 | 同上，真实PG | +| TC-RC-20260918-06 | 最终失败后迟到SubmitResult拒绝；不得重开终态或创建补发；旧尝试不得覆盖当前决定 | 同上真实PG；send-chain.service.spec.ts旧尝试单测 | +| TC-RC-20260918-07 | 同供应商跨连接唯一匹配、歧义拒绝、身份变更、手机号不符、跨租户关系、历史submitId、候选截断和72小时超时恢复 | receipt-attempt-resolver.spec.ts，12项隔离单测；历史分片补关联并齐片完成另有真实PG用例 | + +既有TC-RC-20260916收尾并发与故障用例继续执行tools/testing/verify-attempt-completion.mjs（真实PG、双OS进程、事务回滚、fence、一次补发Outbox、非零扣退费、通知持久化）。本轮不启动Gateway、Redis消费者或网络通知投递，不以数据库集成代替线上完整短信链路验收；目标环境与线上CPU改善待另行授权发布后验证。 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index b28c42d..5428efb 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -5173,3 +5173,16 @@ CUA本轮可用，实际后端文档三尺寸1600×1000/1366×768/390×844无页 - 原始证据在忽略目录.local-data/signature-uniqueness-20260917/。保留初次测试夹具缺邮箱、映射字段键错误、根目录Prisma配置路径错误及真实驱动冲突识别失败的日志，修正后新库v3验收通过；Redis5.0版本建议及pg查询弃用提示仍存在，未扩展升级依赖。 - 历史兼容：线上有效重名尚未盘点；如存在则迁移明确失败并整体回滚，不能自动删除或合并。未执行测试/预生产部署、线上迁移、浏览器页面验收、文件上传/解析及MinIO回归；本轮真实导入验收覆盖资料暂存/审核应用层，文件解析路径未改变。 - 本地修改、测试和需求/设计/用例同步完成，提交前仅本轮文件及共享文档精确追加进入暂存；其他原始修改保留。提交号在交付回复中报告；推送、测试部署、预生产部署均未执行。 + + +## 2026-09-18 长短信三项缺陷整改（本地修改及提交） + +授权范围：修复、测试、文档及本地提交，不推送/部署。开始核验本地main1676cfe、实际远端main5e4d644，暂存区为空；版本、metrics、发布工具/脚本及其他文档草稿保留，不纳入本轮。设计先补第10.13节。 + +已修复：同次最终失败/成功及明确回执超时不再重开收尾，后续失败片仅审计；矛盾回执留原始记录并产生异常，保持账务和通知；统一Submit和Segment候选按通道/上游身份消歧，分片更新限定发送尝试，显式submitId不再被同Msg_Id另一尝试覆盖。保留unknown、未齐片、历史超时恢复和同供应商跨连接匹配。无需迁移，无线上历史数据修正。 + +验证（2026-09-18）：新隔离cmpp_qa_receipt_fix全111迁移通过；verify-receipt-finality.mjs真实PG八组场景通过，重复终态选路由复现的2次变为1次，跨尝试更新2条变为目标1条，失败/退款/失败通知保持一致。选路仅在确定无可用路由边界隔离；账单用已退款快照，非线上退款。verify-attempt-completion.mjs十一组真实PG回归通过，包括24并发、双OS进程、回滚/接管/fence、三段通知、非零账务与唯一补发Outbox。 + +API全量81套/880项通过并达覆盖率门禁（语句67.73%、分支52.89%、函数68.76%、行70.60%）；最终小幅兼容调整后定向2套/150项及八组真实PG、十一组并发恢复复跑通过。TypeScript生产构建通过；本轮文件ESLint零错误、30条既有any警告，Prettier及diff检查通过。新匹配器单测行覆盖100%、分支89.65%。早期旧mock未提供真实关联造成14项失败，补全关系及查询形状后通过；未降低关联约束。真实PG保留pg驱动并发query弃用警告，未出现事务失败。 + +可重复运行：先构建api并迁移独立本机cmpp_qa_*库，设置COMPLETION_TEST_DATABASE_URL后运行tools/testing/verify-receipt-finality.mjs和verify-attempt-completion.mjs；脚本拒绝非本机或非隔离库。原始日志保留.local-data/receipt-fix-20260918/，不进Git。此轮仅调用真实后端服务/持久层，不启动Gateway、HTTP通知投递或在线发送。前端/Go未变，未重跑其验收；目标环境、Redis/Gateway完整网络闭环、线上CPU降幅未验证。不得把本地通过视作测试/预生产已修复。本地隔离数据库进程收尾关闭，数据及日志保留。 diff --git a/tools/testing/verify-receipt-finality.mjs b/tools/testing/verify-receipt-finality.mjs new file mode 100644 index 0000000..7e3f320 --- /dev/null +++ b/tools/testing/verify-receipt-finality.mjs @@ -0,0 +1,315 @@ +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { randomUUID } from 'node:crypto'; +const url = new URL(process.env.COMPLETION_TEST_DATABASE_URL || ''); +assert(['localhost', '127.0.0.1'].includes(url.hostname) && url.pathname.startsWith('/cmpp_qa_')); +process.env.NODE_ENV = 'test'; +process.env.DATABASE_URL = url.toString(); +const require = createRequire(new URL('../../api/package.json', import.meta.url)); +require('reflect-metadata'); +const { PrismaService } = require('./dist/prisma/prisma.service'); +const { SendChainService } = require('./dist/send-chain/send-chain.service'); +const { BillingService } = require('./dist/billing/billing.service'); +const db = new PrismaService(); +const chain = new SendChainService(db, new BillingService(db), {}, {}); +const key = () => randomUUID(); +const results = []; +try { + const tenant = await db.tenant.create({ data: { name: '隔离审查', code: key() } }); + const app = await db.smsApplication.create({ + data: { + tenantId: tenant.id, + name: '隔离', + cmppAccount: key(), + cmppEnterpriseCode: '000001', + secretHash: 'disabled', + interfaceEnabled: true, + }, + }); + const channel = await db.smsChannel.create({ + data: { + name: '隔离不联网', + code: key(), + gatewayHost: '127.0.0.1', + gatewayPort: 1, + account: key(), + passwordCipher: 'unused', + srcId: '1069', + carriers: ['mobile'], + status: 'disabled', + }, + }); + async function fixture(business = false) { + const mid = key(), + sid = key(); + const m = await db.smsMessageRecord.create({ + data: { + messageId: mid, + phoneNumber: '13800138000', + content: '隔离'.repeat(90), + billingUnits: 2, + status: 'submitted', + channelId: channel.id, + submitId: sid, + ...(business + ? { tenantId: tenant.id, applicationId: app.id, cmppSubmitSequenceId: '42', cmppRegisteredDelivery: true } + : {}), + }, + }); + const s = await db.smsSubmitRecord.create({ + data: { + messageRecordId: m.id, + tenantId: m.tenantId, + channelId: channel.id, + submitId: sid, + submitStatus: 'accepted', + }, + }); + return { m, s }; + } + const receipt = (m, g, status = 'delivered') => ({ + messageId: m.messageId, + channelId: channel.id, + gatewayMessageId: g, + phoneNumber: m.phoneNumber, + receiptStatus: status, + rawStatus: status === 'delivered' ? 'DELIVRD' : 'FAIL', + deliveredAt: new Date().toISOString(), + }); + async function segment(m, s, index, g) { + return chain.handleSubmitSegmentResult({ + messageId: m.messageId, + channelId: channel.id, + submitId: s.submitId, + gatewayMessageId: g, + sequenceId: index, + segmentIndex: index, + segmentTotal: 2, + submitStatus: 'accepted', + }); + } + // 1. Two attempts reuse one supplier ID: only target attempt should change. + { + const { m, s } = await fixture(); + const g = key(); + await segment(m, s, 1, g); + const ch2 = await db.smsChannel.create({ + data: { + name: '隔离通道2', + code: key(), + gatewayHost: '127.0.0.1', + gatewayPort: 2, + account: key(), + passwordCipher: 'unused', + srcId: '1069', + carriers: ['mobile'], + status: 'disabled', + }, + }); + const other = await db.smsSubmitRecord.create({ + data: { messageRecordId: m.id, channelId: ch2.id, submitId: key(), submitStatus: 'accepted' }, + }); + await db.smsMessageSegmentAudit.create({ + data: { + messageRecordId: m.id, + submitRecordId: other.id, + submitId: other.submitId, + channelId: ch2.id, + gatewayMessageId: g, + segmentIndex: 1, + segmentTotal: 2, + }, + }); + const resolved = await chain.resolveReceiptMessage(receipt(m, g)); + await chain.handleReceipt(receipt(m, g)); + const rows = await db.smsMessageSegmentAudit.findMany({ where: { messageRecordId: m.id } }); + assert.equal(rows.filter((r) => r.receiptStatus === 'delivered').length, 1); + assert.equal(resolved.channelId, channel.id); + assert.equal(rows.find((r) => r.submitRecordId === other.id).receiptStatus, null); + results.push({ + passed: 'cross_attempt_update', + updatedAttempts: 1, + expected: 1, + matchedWrongChannel: resolved.channelId !== channel.id, + }); + } + // 2. Final failure preserves accounting and downstream result despite contradictory success. + { + const { m, s } = await fixture(true); + const gs = [key(), key()]; + await segment(m, s, 1, gs[0]); + await segment(m, s, 2, gs[1]); + await db.smsMessageRecord.update({ + where: { id: m.id }, + data: { status: 'failed', receiptStatus: 'undelivered', amountCents: 100n }, + }); + await db.smsBillingRecord.create({ + data: { + tenantId: tenant.id, + applicationId: app.id, + messageId: m.messageId, + phoneNumber: m.phoneNumber, + contentLength: 180, + billingUnits: 2, + unitPrice: 50n, + amountCents: 100n, + billingStatus: 'refunded', + }, + }); + await db.cmppDownstreamDelivery.create({ + data: { + tenantId: tenant.id, + applicationId: app.id, + messageRecordId: m.id, + messageId: m.messageId, + dedupeKey: 'receipt:' + m.id, + deliveryType: 'receipt', + payload: { receiptStatus: 'undelivered' }, + status: 'delivered', + }, + }); + await chain.handleReceipt(receipt(m, gs[0])); + await chain.handleReceipt(receipt(m, gs[1])); + const updated = await db.smsMessageRecord.findUnique({ where: { id: m.id } }); + const bill = await db.smsBillingRecord.findFirst({ where: { messageId: m.messageId } }); + const notice = await db.cmppDownstreamDelivery.findUnique({ where: { dedupeKey: 'receipt:' + m.id } }); + assert.equal(updated.status, 'failed'); + assert.equal(await db.smsReceiptAnomaly.count({ where: { messageRecordId: m.id } }), 1); + assert.equal( + await db.smsMessageSegmentAudit.count({ where: { messageRecordId: m.id, receiptStatus: 'delivered' } }), + 0, + ); + assert.equal(bill.billingStatus, 'refunded'); + assert.equal(notice.payload.receiptStatus, 'undelivered'); + results.push({ + passed: 'contradictory_final', + message: updated.status, + billing: bill.billingStatus, + notice: notice.payload.receiptStatus, + }); + } + // 3. Second-fragment receipt precedes its SubmitSegmentResult metadata. + { + const { m, s } = await fixture(); + const gs = [key(), key()]; + await segment(m, s, 1, gs[0]); + await assert.rejects(() => chain.handleReceipt(receipt(m, gs[1])), /提交尝试关联/); + await segment(m, s, 2, gs[1]); + await chain.handleReceipt(receipt(m, gs[1])); + await chain.handleReceipt(receipt(m, gs[0])); + const rows = await db.smsMessageSegmentAudit.findMany({ + where: { messageRecordId: m.id }, + orderBy: { segmentIndex: 'asc' }, + }); + const updated = await db.smsMessageRecord.findUnique({ where: { id: m.id } }); + const count = await db.smsReceiptRecord.count({ where: { messageRecordId: m.id, receiptStatus: 'delivered' } }); + assert.equal(count, 2); + assert.equal(updated.status, 'delivered'); + results.push({ + passed: 'early_fragment_defers_then_recovers', + receiptFacts: count, + message: updated.status, + segments: rows.map((r) => ({ index: r.segmentIndex, status: r.receiptStatus })), + }); + } + // 4. Late failed fragments must not repeat route decisions after final failure. + { + const { m, s } = await fixture(true); + const task = await db.smsBatchTask.create({ + data: { tenantId: tenant.id, applicationId: app.id, taskNo: key(), content: '隔离', phoneTotal: 1 }, + }); + await db.smsMessageRecord.update({ where: { id: m.id }, data: { batchTaskId: task.id } }); + const gs = [key(), key()]; + await segment(m, s, 1, gs[0]); + await segment(m, s, 2, gs[1]); + const priorFind = chain.findApplicationRoute, + priorSelect = chain.selectChannelForMessage; + let selections = 0; + chain.findApplicationRoute = async () => ({ group: { retryEnabled: true, retryTimeLimitMinutes: 60 } }); + chain.selectChannelForMessage = async () => { + selections++; + throw new (require('@nestjs/common').BadRequestException)('isolated no route'); + }; + try { + await chain.handleReceipt(receipt(m, gs[0], 'undelivered')); + const afterFirst = await db.smsMessageRecord.findUnique({ where: { id: m.id } }); + assert.equal(afterFirst.status, 'failed'); + await Promise.all(Array.from({ length: 8 }, () => chain.handleReceipt(receipt(m, gs[1], 'undelivered')))); + assert.equal(selections, 1); + results.push({ + passed: 'repeated_terminal_routing', + routeSelections: selections, + statusAfterFirst: afterFirst.status, + routingBoundaryIsolated: true, + }); + } finally { + chain.findApplicationRoute = priorFind; + chain.selectChannelForMessage = priorSelect; + } + } + // 5. Same-channel collision between a primary ID and another attempt's fragment is ambiguous. + { + const { m, s } = await fixture(); + const g = key(); + await segment(m, s, 1, g); + const other = await db.smsSubmitRecord.create({ + data: { + messageRecordId: m.id, + channelId: channel.id, + submitId: key(), + gatewayMessageId: g, + submitStatus: 'accepted', + }, + }); + await assert.rejects(() => chain.handleReceipt(receipt(m, g)), /提交尝试关联/); + assert.equal(await db.smsReceiptRecord.count({ where: { messageRecordId: m.id } }), 0); + assert.equal( + await db.smsMessageSegmentAudit.count({ where: { messageRecordId: m.id, receiptStatus: 'delivered' } }), + 0, + ); + results.push({ passed: 'ambiguous_primary_fragment_rejected', otherAttempt: other.id }); + } + // 6. Unknown is not a final failure; all actual fragments may still complete it. + { + const { m, s } = await fixture(); + const gs = [key(), key()]; + await segment(m, s, 1, gs[0]); + await segment(m, s, 2, gs[1]); + await db.smsMessageRecord.update({ where: { id: m.id }, data: { status: 'unknown' } }); + await Promise.all(gs.map((g) => chain.handleReceipt(receipt(m, g)))); + assert.equal((await db.smsMessageRecord.findUnique({ where: { id: m.id } })).status, 'delivered'); + results.push({ passed: 'unknown_and_concurrent_fragments_recover' }); + } + // 7. Late submit rejection cannot reopen a final failure or create a successor. + { + const { m, s } = await fixture(); + const g = key(); + await segment(m, s, 1, g); + await db.smsMessageRecord.update({ where: { id: m.id }, data: { status: 'failed', receiptStatus: 'undelivered' } }); + await chain.handleSubmitResult({ + messageId: m.messageId, + channelId: channel.id, + submitId: s.submitId, + gatewayMessageId: g, + submitStatus: 'rejected', + }); + assert.equal((await db.smsMessageRecord.findUnique({ where: { id: m.id } })).status, 'failed'); + assert.equal(await db.smsSubmitRecord.count({ where: { messageRecordId: m.id } }), 1); + results.push({ passed: 'late_submit_rejection_preserves_final' }); + } + // 8. Legacy audits missing the FK must join the verified attempt before aggregation. + { + const { m, s } = await fixture(); + const gs = [key(), key()]; + await segment(m, s, 1, gs[0]); + await segment(m, s, 2, gs[1]); + await db.smsMessageSegmentAudit.updateMany({ where: { messageRecordId: m.id }, data: { submitRecordId: null } }); + for (const g of gs) await chain.handleReceipt(receipt(m, g)); + assert.equal((await db.smsMessageRecord.findUnique({ where: { id: m.id } })).status, 'delivered'); + assert.equal(await db.smsMessageSegmentAudit.count({ where: { messageRecordId: m.id, submitRecordId: s.id } }), 2); + results.push({ passed: 'legacy_fragment_relation_recovered' }); + } + console.log(JSON.stringify(results)); +} finally { + await db.$disconnect(); +}