From b5005f21d5092b7e2759efdce0ebd02798e4552f Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Tue, 25 Aug 2026 17:53:05 +0800 Subject: [PATCH] fix: make downstream connection events idempotent --- .../application-lifecycle.service.ts | 9 ++++-- api/src/sms-config/sms-config.service.spec.ts | 28 +++++++++++++++++-- .../first-version-development-requirements.md | 1 + docs/system-functional-test-cases.md | 1 + 4 files changed, 35 insertions(+), 4 deletions(-) diff --git a/api/src/sms-config/application-lifecycle.service.ts b/api/src/sms-config/application-lifecycle.service.ts index df03e5c..aa13fbd 100644 --- a/api/src/sms-config/application-lifecycle.service.ts +++ b/api/src/sms-config/application-lifecycle.service.ts @@ -203,7 +203,8 @@ export class SmsApplicationLifecycleService { orderBy: [{ connectedAt: 'asc' }, { connectionId: 'asc' }], }); const allowedConnectionIds = activeConnections.slice(0, application.cmppMaxConnections).map((item) => item.connectionId); - if ((!existing && activeConnections.length >= application.cmppMaxConnections) + const currentConnectionAlreadyActive = activeConnections.some((item) => item.connectionId === data.connectionId); + if ((!existing && !currentConnectionAlreadyActive && activeConnections.length >= application.cmppMaxConnections) || (existing && activeConnections.length > application.cmppMaxConnections && !allowedConnectionIds.includes(data.connectionId))) { throw new ForbiddenException(`CMPP connection limit exceeded (${application.cmppMaxConnections})`); } @@ -224,7 +225,11 @@ export class SmsApplicationLifecycleService { }; const connection = existing ? await this.prisma.cmppDownstreamConnection.update({ where: { id: existing.id }, data: payload }) - : await this.prisma.cmppDownstreamConnection.create({ data: { connectionId: data.connectionId, ...payload } }); + : await this.prisma.cmppDownstreamConnection.upsert({ + where: { connectionId: data.connectionId }, + create: { connectionId: data.connectionId, ...payload }, + update: payload, + }); if (data.status === 'connected') { await this.writeOperationLog(application.tenantId, undefined, `cmpp_downstream_connection.${data.status}`, 'cmpp_downstream_connection', data.connectionId, { applicationId: application.id, diff --git a/api/src/sms-config/sms-config.service.spec.ts b/api/src/sms-config/sms-config.service.spec.ts index 84fd7fb..351b5a7 100644 --- a/api/src/sms-config/sms-config.service.spec.ts +++ b/api/src/sms-config/sms-config.service.spec.ts @@ -147,6 +147,7 @@ function createPrismaMock() { findMany: jest.fn().mockResolvedValue([{ id: 'downstream-1', applicationId: 'app-1', tenantId: 'tenant-1', account: '100001', enterpriseCode: 'APP-EC', connectionId: 'gateway-1-1', status: 'connected', connectedAt: new Date(), lastHeartbeatAt: new Date() }]), findUnique: jest.fn().mockResolvedValue(null), create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'downstream-1', ...data })), + upsert: jest.fn().mockImplementation(({ create, update }) => Promise.resolve({ id: 'downstream-1', ...(create ?? update) })), update: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'downstream-1', ...data })), delete: jest.fn().mockResolvedValue({ id: 'downstream-1' }), deleteMany: jest.fn().mockResolvedValue({ count: 0 }), @@ -646,8 +647,9 @@ describe('SmsConfigService', () => { observedAt: '2026-07-11T11:00:00.000Z', }); - expect(prisma.cmppDownstreamConnection.create).toHaveBeenCalledWith({ - data: expect.objectContaining({ + expect(prisma.cmppDownstreamConnection.upsert).toHaveBeenCalledWith({ + where: { connectionId: 'gateway-1-1' }, + create: expect.objectContaining({ applicationId: 'app-1', tenantId: 'tenant-1', account: '100001', @@ -656,6 +658,10 @@ describe('SmsConfigService', () => { status: 'connected', remoteIp: '127.0.0.1', }), + update: expect.objectContaining({ + applicationId: 'app-1', + status: 'connected', + }), }); expect(prisma.cmppDownstreamConnection.deleteMany).toHaveBeenCalledWith({ where: { @@ -1103,6 +1109,24 @@ describe('SmsConfigService', () => { expect(prisma.smsSignature.update).not.toHaveBeenCalled(); }); + it('accepts concurrent state callbacks for the same newly connected Gateway session', async () => { + const prisma = createPrismaMock(); + prisma.smsApplication.findUnique.mockResolvedValue({ + id: 'app-1', tenantId: 'tenant-1', cmppEnterpriseCode: 'APP-EC', cmppMaxConnections: 1, + interfaceEnabled: true, status: 'active', ipAllowlist: [], + }); + prisma.cmppDownstreamConnection.findUnique.mockResolvedValue(null); + prisma.cmppDownstreamConnection.findMany.mockResolvedValue([{ connectionId: 'gateway-current' }]); + const service = new SmsConfigService(prisma as never); + + await expect(service.recordDownstreamConnectionEvent({ + account: '100001', connectionId: 'gateway-current', status: 'submit', remoteIp: '127.0.0.1', + })).resolves.toEqual(expect.objectContaining({ connectionId: 'gateway-current' })); + expect(prisma.cmppDownstreamConnection.upsert).toHaveBeenCalledWith(expect.objectContaining({ + where: { connectionId: 'gateway-current' }, + })); + }); + it('applies the audit submission range to signatures, templates and drainage records', async () => { const prisma = createPrismaMock(); const service = new SmsConfigService(prisma as never); diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 1ddbe85..ba1df3b 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -2168,3 +2168,4 @@ - 原生SQL对Prisma的无时区时间列统一使用`NOW() AT TIME ZONE 'UTC'`，覆盖Submit Outbox领取、租约、发布、重试及关联消息更新时间，禁止用数据库会话时区污染时延审计。 - 验收必须保持正价，分别执行单企业100/150 TPS和至少两个独立企业合计200 TPS，按非补发首次供应商Submit、回执、上行、计费、主备补发、业务拦截及全队列排空对账；触发拒绝、连接错误、持续积压、数据库异常或账务不一致立即停止。多Gateway P2不在本阶段范围。 - Gateway到主API的回环HTTP连接池空闲时长不得超过API服务端keep-alive生命周期；API默认keep-alive 120秒、headers timeout 125秒，并由发布环境显式设置和校验，防止负载期复用已被Node关闭的连接而把本可受理的Submit误回Result 9。 +- 同一Gateway下游会话的connected、heartbeat、submit和deliver状态回调允许并发到达；连接上限按不同`connectionId`计数，同一`connectionId`的首次状态必须幂等upsert，不得因查询/创建竞态误报超过连接数并主动断开客户连接。不同连接超过应用上限时仍返回403并由Gateway断开。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index bbfabbd..6961466 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4886,3 +4886,4 @@ npm run verify:phase8 | TC-CMPP-PHASE5-015 | Outbox UTC时间语义 | Asia/Shanghai数据库会话下领取、租约、发布、重试均写UTC无时区值；publishedAt-createdAt不再出现约8小时偏差 | | TC-CMPP-PHASE5-016 | 微批发布边界与停止线 | 仅单Gateway；分别验证单企业100/150与多企业200，发生拒绝、连接错误、持续积压、数据库异常或账务不一致立即停止 | | TC-CMPP-PHASE5-017 | Gateway到API长连接生命周期 | API keep-alive 120秒大于Gateway连接池90秒，headers timeout更大；跨越Node原默认5秒空闲边界后继续压测，不得出现loopback connection reset或Result 9 | +| TC-CMPP-PHASE5-018 | 同连接并发状态回调 | 同一connectionId的connected与submit并发时幂等upsert且保持在线；不同connectionId超过cmppMaxConnections仍403，不能误断当前连接或漏SubmitResp |