diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 60000f1..e3964c4 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -2119,6 +2119,24 @@ describe('SendChainService', () => { try { const { service, prisma, billing, riskReview, phoneFrequency } = createService(); service.enqueueBatchTask = jest.fn(); + prisma.$queryRaw.mockImplementationOnce((query) => { + const payloadHash = query.values.find((value: unknown) => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value)); + return Promise.resolve([{ + validationError: null, + payloadHash, + response: { + accepted: true, + tenantId: 'tenant-1', + applicationId: 'app-1', + taskId: '', + messageId: 'MSG-fast', + messageRecordId: '', + status: 'accepted_pending', + phoneCount: 2, + messages: [], + }, + }]); + }); const result = await service.submitInboundMessage({ requestId: 'cmpp-inbound:test-fast-path', @@ -2134,13 +2152,12 @@ describe('SendChainService', () => { status: 'accepted_pending', phoneCount: 2, })); - expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledWith({ - data: expect.objectContaining({ - requestKey: 'cmpp-inbound:test-fast-path', - tenantId: 'tenant-1', - applicationId: 'app-1', - }), - }); + expect(prisma.smsApplication.findFirst).not.toHaveBeenCalled(); + expect(prisma.$queryRaw).toHaveBeenCalledTimes(1); + const sql = prisma.$queryRaw.mock.calls[0][0].strings.join(' '); + expect(sql).toContain('INSERT INTO "CmppInboundSubmissionInbox"'); + expect(sql).toContain('JOIN "Tenant"'); + expect(sql).toContain('ON CONFLICT ("requestKey") DO NOTHING'); expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); expect(riskReview.evaluateTask).not.toHaveBeenCalled(); @@ -2158,7 +2175,6 @@ describe('SendChainService', () => { process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true'; try { const { service, prisma } = createService(); - let originalHash = ''; const storedResponse = { accepted: true, tenantId: 'tenant-1', @@ -2170,19 +2186,10 @@ describe('SendChainService', () => { phoneCount: 1, messages: [{ phoneNumber: '13800000001', messageId: 'MSG-stable', messageRecordId: '', taskId: '', status: 'accepted_pending' }], }; - prisma.cmppInboundSubmissionInbox.create - .mockImplementationOnce(({ data }) => { - originalHash = data.payloadHash; - return Promise.resolve({ id: 'inbox-1' }); - }) - .mockRejectedValueOnce(new Prisma.PrismaClientKnownRequestError('duplicate request key', { - code: 'P2002', - clientVersion: '7.9.0', - })); - prisma.cmppInboundSubmissionInbox.findUnique.mockImplementation(() => Promise.resolve({ - payloadHash: originalHash, - response: storedResponse, - })); + prisma.$queryRaw.mockImplementation((query) => { + const payloadHash = query.values.find((value: unknown) => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value)); + return Promise.resolve([{ validationError: null, payloadHash, response: storedResponse }]); + }); const request = { requestId: 'cmpp-inbound:test-retry', account: '100001', @@ -2194,7 +2201,8 @@ describe('SendChainService', () => { await service.submitInboundMessage(request); await expect(service.submitInboundMessage(request)).resolves.toEqual(storedResponse); - expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledTimes(2); + expect(prisma.$queryRaw).toHaveBeenCalledTimes(2); + expect(prisma.cmppInboundSubmissionInbox.findUnique).not.toHaveBeenCalled(); expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); } finally { if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED; @@ -2202,6 +2210,58 @@ describe('SendChainService', () => { } }); + it('rejects a disabled application before the merged Inbox statement can insert', async () => { + const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true'; + try { + const { service, prisma } = createService(); + prisma.$queryRaw.mockResolvedValueOnce([{ + validationError: 'CMPP account is disabled for new submissions', + payloadHash: null, + response: null, + }]); + + await expect(service.submitInboundMessage({ + requestId: 'cmpp-inbound:disabled', + account: '100001', + phoneNumber: '13800000001', + content: 'hello', + })).rejects.toThrow('CMPP account is disabled for new submissions'); + expect(prisma.cmppInboundSubmissionInbox.findUnique).not.toHaveBeenCalled(); + } finally { + if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous; + } + }); + + it('uses a read-only recovery only for a concurrent Inbox insert outside the CTE snapshot', async () => { + const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true'; + try { + const { service, prisma } = createService(); + let payloadHash = ''; + prisma.$queryRaw.mockImplementationOnce((query) => { + payloadHash = query.values.find((value: unknown) => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value)); + return Promise.resolve([{ validationError: null, payloadHash: null, response: null }]); + }); + prisma.cmppInboundSubmissionInbox.findUnique.mockImplementationOnce(() => Promise.resolve({ + payloadHash, + response: { accepted: true, messageId: 'MSG-concurrent', status: 'accepted_pending' }, + })); + + await expect(service.submitInboundMessage({ + requestId: 'cmpp-inbound:concurrent', + account: '100001', + phoneNumber: '13800000001', + content: 'hello', + })).resolves.toEqual(expect.objectContaining({ messageId: 'MSG-concurrent' })); + expect(prisma.cmppInboundSubmissionInbox.findUnique).toHaveBeenCalledTimes(1); + } finally { + if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous; + } + }); + it('persists an idempotent daily quota reservation with a Prisma Date value', async () => { const { service, prisma } = createService(); prisma.$queryRaw.mockResolvedValueOnce([{ tenantId: 'tenant-1', dailyLimit: 100000, usedCount: 1 }]); diff --git a/api/src/send-chain/send-inbound-entry.service.ts b/api/src/send-chain/send-inbound-entry.service.ts index e5d59d2..3b22ee6 100644 --- a/api/src/send-chain/send-inbound-entry.service.ts +++ b/api/src/send-chain/send-inbound-entry.service.ts @@ -33,6 +33,26 @@ type ClaimedInboundWorkflow = { payload: Prisma.JsonValue; }; +type PersistedInboundWorkflowRow = { + validationError: string | null; + payloadHash: string | null; + response: Prisma.JsonValue | null; +}; + +type InboundApplication = NonNullable>>; + +type InboundWorkflowResponse = { + accepted: boolean; + tenantId: string; + applicationId: string; + taskId: string; + messageId: string; + messageRecordId: string; + status: string; + phoneCount: number; + messages: Array<{ phoneNumber: string; messageId: string; messageRecordId: string; taskId: string; status: string }>; +}; + /** * R9 inboundEntry implementation. Cross-method calls return through the stable SendChainService seam. */ @@ -183,6 +203,13 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { throw new BadRequestException('CMPP submit phone number is invalid'); } + if (this.inboundFastPathEnabled() && !data.longMessage) { + return this.measureInboundStage( + 'inbox_persist', + () => this.persistValidatedInboundWorkflow(data, phoneNumbers), + ); + } + const application = await this.measureInboundStage( 'application_lookup', () => this.facade.findInboundApplication(data.account), @@ -276,12 +303,6 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { throw error; } } - if (this.inboundFastPathEnabled()) { - return this.measureInboundStage( - 'inbox_persist', - () => this.persistInboundWorkflow(data, phoneNumbers, application), - ); - } return this.measureInboundStage( 'complete_submit', () => this.facade.submitCompleteInboundMessage(data, phoneNumbers, application), @@ -292,6 +313,162 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { return process.env.CMPP_INBOUND_FAST_PATH_ENABLED === 'true'; } + private async persistValidatedInboundWorkflow( + data: GatewayInboundSubmitDto, + phoneNumbers: string[], + ) { + const requestKey = data.requestId?.trim(); + if (!requestKey || requestKey.length > 160) { + throw new BadRequestException('CMPP inbound requestId is required for fast-path idempotency'); + } + const submitGroupMessageId = `MSG-${randomUUID()}`; + const messageIds = phoneNumbers.map((_, index) => index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`); + const payload: InboundWorkflowPayload = { + data: JSON.parse(JSON.stringify(data)) as GatewayInboundSubmitDto, + phoneNumbers, + submitGroupMessageId, + messageIds, + }; + const payloadHash = createHash('sha256').update(JSON.stringify({ + data: payload.data, + phoneNumbers, + requestedGroupMessageId: null, + })).digest('hex'); + const responseMessages = phoneNumbers.map((phoneNumber, index) => ({ + phoneNumber, + messageId: messageIds[index], + messageRecordId: '', + taskId: '', + status: 'accepted_pending', + })); + const remoteIp = data.remoteIp?.replace(/^::ffff:/, '').trim() || null; + const submittedSrcId = data.srcId?.trim() ?? ''; + + // One indexed statement must both validate the current application state and create the + // durable Inbox row. Keeping those operations in one database snapshot prevents a disable + // racing between a separate SELECT and INSERT, while the unique request key remains the + // authoritative idempotency boundary. + const rows = await this.prisma.$queryRaw(Prisma.sql` + WITH application AS ( + SELECT app.id, + app."tenantId", + app.status, + app."interfaceEnabled", + app."queuePriority", + app."cmppApplicationExtension", + app."cmppAccessNumberFillEnabled", + app."cmppAccessNumberFillPrefix", + app."cmppClientSrcId", + tenant.status AS "tenantStatus" + FROM "SmsApplication" AS app + JOIN "Tenant" AS tenant ON tenant.id = app."tenantId" + WHERE app."cmppAccount" = ${data.account} + LIMIT 1 + ), validation AS ( + SELECT application.*, + CASE + WHEN application.status <> 'active' + OR application."tenantStatus" <> 'active' + OR NOT application."interfaceEnabled" + THEN 'CMPP account is disabled for new submissions' + WHEN ${remoteIp}::text IS NOT NULL + AND EXISTS ( + SELECT 1 FROM "SmsApplicationIpAllowlist" allowlist + WHERE allowlist."applicationId" = application.id + ) + AND NOT EXISTS ( + SELECT 1 + FROM "SmsApplicationIpAllowlist" allowlist + WHERE allowlist."applicationId" = application.id + AND ( + (position('/' IN trim(allowlist."ipCidr")) = 0 + AND regexp_replace(trim(allowlist."ipCidr"), '^::ffff:', '') = ${remoteIp}) + OR ( + trim(allowlist."ipCidr") ~ '^[0-9]{1,3}(\\.[0-9]{1,3}){3}/([0-9]|[12][0-9]|3[0-2])$' + AND ${remoteIp} ~ '^[0-9]{1,3}(\\.[0-9]{1,3}){3}$' + AND ${remoteIp}::inet <<= trim(allowlist."ipCidr")::cidr + ) + ) + ) + THEN 'CMPP source IP is not in application allowlist' + WHEN coalesce(trim(application."cmppApplicationExtension"), '') <> '' + AND ${submittedSrcId} <> coalesce( + nullif(trim(application."cmppClientSrcId"), ''), + CASE WHEN application."cmppAccessNumberFillEnabled" + THEN coalesce(trim(application."cmppAccessNumberFillPrefix"), '') + ELSE '' + END || trim(application."cmppApplicationExtension") + ) + THEN 'CMPP Src_Id must equal the access number assigned to this application: ' || coalesce( + nullif(trim(application."cmppClientSrcId"), ''), + CASE WHEN application."cmppAccessNumberFillEnabled" + THEN coalesce(trim(application."cmppAccessNumberFillPrefix"), '') + ELSE '' + END || trim(application."cmppApplicationExtension") + ) + ELSE NULL + END AS "validationError" + FROM application + ), inserted AS ( + INSERT INTO "CmppInboundSubmissionInbox" ( + id, "requestKey", "payloadHash", "tenantId", "applicationId", "queuePriority", + payload, response, status, attempts, "nextAttemptAt", "createdAt", "updatedAt" + ) + SELECT ${randomUUID()}, ${requestKey}, ${payloadHash}, validation."tenantId", validation.id, + CASE WHEN validation."queuePriority" = 'priority' THEN 'priority' ELSE 'normal' END, + ${JSON.stringify(payload)}::jsonb, + jsonb_build_object( + 'accepted', true, + 'tenantId', validation."tenantId", + 'applicationId', validation.id, + 'taskId', '', + 'messageId', ${submitGroupMessageId}, + 'messageRecordId', '', + 'status', 'accepted_pending', + 'phoneCount', ${phoneNumbers.length}, + 'messages', ${JSON.stringify(responseMessages)}::jsonb + ), + 'pending', 0, + (NOW() AT TIME ZONE 'UTC'), + (NOW() AT TIME ZONE 'UTC'), + (NOW() AT TIME ZONE 'UTC') + FROM validation + WHERE validation."validationError" IS NULL + ON CONFLICT ("requestKey") DO NOTHING + RETURNING "payloadHash", response + ) + SELECT validation."validationError", + coalesce(inserted."payloadHash", existing."payloadHash") AS "payloadHash", + coalesce(inserted.response, existing.response) AS response + FROM validation + LEFT JOIN inserted ON true + LEFT JOIN "CmppInboundSubmissionInbox" existing + ON existing."requestKey" = ${requestKey} + LIMIT 1 + `); + const row = rows[0]; + if (!row) { + throw new BadRequestException('CMPP account is invalid'); + } + if (row.validationError) { + throw new BadRequestException(row.validationError); + } + if (row.response == null) { + // A concurrent insert that wins after this statement's MVCC snapshot is not visible to + // the CTE. The unique key has already prevented duplication, so only that rare retry path + // needs a second read; normal submissions still use exactly one database round trip. + const concurrent = await this.prisma.cmppInboundSubmissionInbox.findUnique({ where: { requestKey } }); + if (!concurrent || concurrent.payloadHash !== payloadHash) { + throw new BadRequestException('CMPP inbound requestId conflicts with another payload'); + } + return concurrent.response as InboundWorkflowResponse; + } + if (row.payloadHash !== payloadHash) { + throw new BadRequestException('CMPP inbound requestId conflicts with another payload'); + } + return row.response as InboundWorkflowResponse; + } + private async persistInboundWorkflow( data: GatewayInboundSubmitDto, phoneNumbers: string[], @@ -1051,10 +1228,17 @@ startInboundWorkflowWorker() { try { await this.refreshInboundWorkflowMetrics(); const claimed = await this.claimInboundWorkflows(available); + const applications = claimed.length === 0 + ? [] + : await this.prisma.smsApplication.findMany({ + where: { id: { in: [...new Set(claimed.map((item) => item.applicationId))] } }, + include: { tenant: true, ipAllowlist: true }, + }); + const applicationById = new Map(applications.map((application) => [application.id, application])); for (const item of claimed) { this.inboundWorkflowInFlight += 1; this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); - const task = this.processClaimedInboundWorkflow(item) + const task = this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId)) .catch((error) => this.logger.error(`CMPP inbound workflow ${item.id} failed to settle: ${String(error)}`)) .finally(() => { this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - 1); @@ -1110,11 +1294,10 @@ startInboundWorkflowWorker() { `); } - private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow) { + private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow, application?: InboundApplication) { try { const payload = parseInboundWorkflowPayload(item.payload); - const application = await this.facade.findInboundApplication(payload.data.account); - if (!application || application.id !== item.applicationId) { + if (!application || application.cmppAccount !== payload.data.account) { throw new Error('CMPP inbound application no longer matches persisted workflow'); } const result = await this.facade.submitCompleteInboundMessage( diff --git a/docs/codebase-modularization-roadmap.md b/docs/codebase-modularization-roadmap.md index 86db44c..0470816 100644 --- a/docs/codebase-modularization-roadmap.md +++ b/docs/codebase-modularization-roadmap.md @@ -1160,3 +1160,5 @@ global，不能错误归入client。`client-signature-*`、发送页、企业认 - 进程隔离同时包含容量隔离：`PrismaService`按进程角色选择独立、有界的连接池上限；Gateway inbound分别持有与受限Submit窗口匹配的Submit API Transport和小型后台Transport，协议日志/回执流量不得占用Submit连接。连接复用、池上限和超时只属于传输/基础设施边界，不得渗入风控、计费、路由或消息状态机。 - `api/src/infrastructure-monitoring/`是运营端监控聚合与固定阈值应用边界：只消费代码白名单 PromQL，并通过版本化 PostgreSQL 单例、promtool 校验和原子规则热加载管理数值阈值；Exporter安装、端口隔离、固定规则模板和权限仍归`tools/monitoring/`治理。 - 活动告警已读也归该边界：Prometheus保留告警事实，Prisma仅持久化逐管理员、逐触发周期的阅读状态；全局布局只消费轻量未读汇总，不复制指纹、activeAt或用户隔离逻辑。 +- 500条/秒第二阶段仍限制在`send-inbound-entry`和既有进程容量边界：短短信用一个PostgreSQL CTE完成当前应用校验和Inbox写入；长短信继续走既有分片域。Worker可按一次领取批次预取应用快照，但不得引入跨批应用状态缓存，也不得把模板、风控、计费、路由状态塞回Gateway同步入口。 +- 容量参数继续归进程组装层：Inbox业务槽、BullMQ发送槽、Gateway供应商槽和结果Outbox槽分别有界、分别观测。代码模块不得假定测试环境参数就是生产默认值；生产调优必须重新基于PostgreSQL连接预算、六通道TPS/窗口和回调承载证据。 diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index df61df7..d781c2f 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -2114,3 +2114,11 @@ - 右上角预警中心增加“系统监控告警”，通过独立轻量接口统计 Prometheus 当前 firing/pending 告警及严重数，跳转系统监控活动告警区；任一预警域失败不得清空其他域。 - 活动告警列表必须提供逐条“标记已读”。已读状态按管理员和“告警指纹 + 本次 activeAt”持久化到 PostgreSQL；仅从当前管理员的预警中心数量中扣减，不改变 Prometheus firing/pending 状态，也不减少页面活动告警总数。同标签告警恢复后再次触发时必须重新成为未读。 - 服务端只能确认 Prometheus 当前仍存在且 activeAt 一致的告警，过期、已恢复或已重新触发的请求必须拒绝；重复点击同一次告警应幂等，并写操作日志。阈值设置弹窗只保留通用 Modal 外层滚动，不得嵌套第二个独立滚动区域。 + +## 完整处理500条/秒第二阶段：合并入口SQL与有界全链扩容（2026-08-20） + +- 普通短短信快路径必须把应用、企业当前启停状态、接口开关、IP白名单、Src_Id校验与Inbox幂等写入合并为同一个PostgreSQL语句和同一MVCC快照；正常新请求只允许一次数据库往返，不得先查应用再写Inbox。应用账号唯一索引和Inbox请求键唯一索引继续作为查询与幂等边界，不增加无证据索引。 +- 合并语句不得缓存应用启停或白名单；应用不存在、已停用、接口关闭、IP或Src_Id不匹配时不得产生Inbox。相同请求键同载荷返回原响应，不同载荷拒绝；并发唯一键竞争因快照不可见时只允许一次只读恢复，不得通过无意义`ON CONFLICT DO UPDATE`制造写放大。 +- 长短信仍在完整重组并完成既有校验后写Inbox，不为追求吞吐改写分片状态机。Inbox Worker每次领取后应按本批应用ID一次读取当前应用/企业/白名单快照，不能对同一批每条消息重复查应用；领取、实际业务处理和逐条完成仍保持原短事务与租约边界。 +- 测试环境可在PostgreSQL连接总预算内分别提高Worker业务槽、BullMQ发送槽、Gateway供应商槽及结果Outbox槽，但每个池必须显式有界，且总数据库连接须为运维保留余量。供应商真实TPS、连接数和窗口仍是硬上限，禁止用并发参数绕过单通道限速。 +- 第二阶段仍以“完整处理”验收：除500条/秒SubmitResp零拒绝、零缺失和延迟停止线外，还必须等待Inbox、BullMQ、命令Stream、结果Outbox与回执链排空，并按数据库业务记录、唯一MessageId、供应商尝试/接受和最终状态对账。入口达到500而全链排空速率不足500条/秒时必须明确判失败。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 1971045..dcf83e2 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4745,3 +4745,14 @@ npm run verify:phase8 | TC-CMPP-500-P1-012 | API/Worker数据库池隔离 | API与Worker分别配置32/8连接并在Worker积压时持续提交 | 两进程使用各自有界连接池，API受理连接不被Worker抢占；总连接数不超过PostgreSQL上限且压后无`idle in transaction`泄漏 | 执行记录（2026-08-20，测试环境）：P1-001至009已由API/Gateway自动化、真实PostgreSQL迁移和独立Worker恢复验证覆盖；P1-011专用Submit传输隔离后100、200条/秒分别2999/2999、3998/3998成功，零拒绝、零节流、零连接错误。P1-010在500条/秒档失败：测试环境池调优后10秒仅2979条，实际297.9条/秒且节流611次；全部2979条最终完成并排空，但完整链仅约24条/秒。故第一阶段不通过500条/秒总目标，不执行“完整500条/秒已达标”的结论。 + +### 完整处理500条/秒第二阶段 + +| 编号 | 场景 | 操作 | 预期 | +|---|---|---|---| +| TC-CMPP-500-P2-001 | 合并校验与Inbox写入 | 普通短短信走快路径并检查SQL与阶段指标 | 使用账号唯一索引在同一SQL校验应用/企业/接口/IP/Src_Id并幂等插入；正常请求仅一次DB往返且不再单独记录`application_lookup` | +| TC-CMPP-500-P2-002 | 合并SQL拒绝安全 | 分别停用应用、停用企业、关闭接口、使用错误IP和Src_Id | 返回对应拒绝且Inbox没有新增；不读取或缓存旧应用快照 | +| TC-CMPP-500-P2-003 | 幂等与并发唯一键竞争 | 同载荷重试、冲突载荷重试，并并发提交相同请求键 | 同载荷返回原MessageId；冲突载荷拒绝；并发竞争最多执行一次只读恢复，不发生重复Inbox或无意义更新 | +| TC-CMPP-500-P2-004 | Worker批量应用快照 | 同批领取多个相同及不同应用Inbox | 每个领取批次按唯一应用ID一次查询，逐条继续使用匹配快照；应用不匹配时安全退避，不跨租约持有事务 | +| TC-CMPP-500-P2-005 | 有界容量配置 | 在测试环境提高Worker、BullMQ、Gateway Submit和Outbox并发并观测连接 | 各池配置值与在途数可观测，PostgreSQL连接低于`max_connections`并无长事务；单通道限速不被绕过 | +| TC-CMPP-500-P2-006 | 100→200→500完整链阶梯 | 使用全新测试号段、隔离六通道模拟器，逐档注入并等待全部队列排空 | 每档报告入口实际速率/分位、Inbox完成、各队列峰值与排空、供应商阶段和数据库对账；500档只有入口及完整链均达到500条/秒且零丢重才通过 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index b33e1d1..cf85781 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -3805,3 +3805,10 @@ git diff --check - 专用Submit池后100条/秒30秒发满2999条，2999/2999成功、零拒绝/节流/连接错误，P50/P95/P99=`42/147/173ms`，API平均约37ms、Gateway API往返平均约54ms；2999条Inbox全部completed，完整供应商链从开始到双Stream排空约132秒，折算约22.7条/秒。200条/秒20秒发满3998条，3998/3998成功、零拒绝/节流/连接错误，分位=`134/370/837ms`；Inbox在开始后约75秒全部完成，双Stream在开始后约166秒排空，完整链折算约24.1条/秒。 - 默认API池32、Submit池64下首次500条/秒10秒只生成1716条，1716/1716成功但节流615次，P95=5217ms，实际171.6条/秒，失败停止。基于PostgreSQL`max_connections=100`和进程连接证据，仅在测试环境把`API_DB_POOL_MAX`调为48、`GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY`调为128，Worker仍为8；复测生成2979条，2979/2979成功、零拒绝/连接错误，P50/P95/P99=`942/1162/3221ms`，但仍节流611次，实际297.9条/秒，500入口仍失败。10个测试连接各窗口32形成320总在途，Gateway平均往返约965ms，实测上限与窗口理论值一致；API平均约322ms，继续只加连接会放大PostgreSQL竞争。 - 最终数据库按四个新号码前缀对账：`1390005/6/7/8`的Inbox completed分别为`2999/3998/1716/2979`，对应`SmsMessageRecord`及唯一MessageId数量完全相同，非completed Inbox为0，命令Stream和结果Outbox均`pending=0/lag=0`，相关服务日志无panic/fatal/Prisma错误或Inbox retry。第一阶段结论为200条/秒入口可靠受理通过、500条/秒入口未通过、完整异步链约24条/秒；不得宣称已完整处理500条/秒。下一阶段必须合并每Submit应用查询与Inbox写入、提高测试连接/窗口总量，并把Worker业务事务、供应商六通道总TPS及结果/回执写入水平扩容，不能再靠单机连接池参数解决。 + +## 2026-08-20 完整处理500条/秒第二阶段：合并入口SQL与全链有界扩容（实施中） + +- 本轮重新核对本地`HEAD=633e7a705539cc706aa0a6dbabfd47cc61265e7f`、`origin/main=c4f36fc50d7906dfb2f97c881e9ea43c6a64c370`；测试环境运行标识为`229a0b28fd8b84d910c359ff9ac442fc3e843cb4+workspace.p1submitpool.f50505ad9ee9`，预生产仍为`433b2ee56f6016ad8afff1bac73f510b8fd53083+gateway-v2.cd7bb8d05e7b`。受保护的构建产物、`outputs/`、`pnpm-lock.yaml`和空文件`=`保持原状，不归因本轮。 +- Worker累计真实指标4695条显示：预检、模板、消息持久化、风控频次、计费和入队平均约`83.9/74.5/28.3/92.2/25.2/64.7ms`，与上一阶段完整链约24条/秒一致。该证据确认不能把入口吞吐冒充完整处理能力。 +- 普通短短信快路径已改为单个PostgreSQL CTE：使用账号唯一索引读取应用和企业当前状态，在同一快照校验接口、IP白名单及Src_Id，并以请求键`ON CONFLICT DO NOTHING`幂等写Inbox和稳定响应。正常路径不再先执行Prisma应用查询；仅并发唯一键竞争且当前快照看不到胜者时执行一次只读恢复，避免用自更新制造表膨胀。长短信分片状态机保持不变。 +- Inbox Worker领取后改为按本批唯一`applicationId`一次查询应用、企业和白名单，再逐条处理；不做跨批应用状态缓存，领取事务仍不包住风控、计费、Redis或供应商调用。专项SendChain 121项、API全量42套493项、API与前端TypeScript、Vite生产构建、Prisma validate、Gateway全量测试/vet、5份Stream契约、R0/R6/R7/R10、安全/部署门禁和`git diff --check`均通过；R9仍只被HEAD既有`dispatchDueScheduledTasks`哈希漂移阻断，未改写该方法。提交、恢复资产、测试环境发布和100→200→500阶梯结果待后续补记。