From 67b760a5992d7ae10d7889fe6765e895d238ab3d Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Thu, 20 Aug 2026 14:19:42 +0800 Subject: [PATCH] perf(cmpp): decouple supplier result callbacks --- .env.example | 4 + .../migration.sql | 6 + api/prisma/schema.prisma | 2 + api/src/send-chain/send-chain.contracts.ts | 2 + api/src/send-chain/send-chain.service.spec.ts | 40 +++ .../send-chain/send-gateway-result.service.ts | 28 ++ docs/codebase-modularization-roadmap.md | 1 + .../examples/submit-result-outbox-event.json | 24 ++ .../gateway-queue-messages.schema.json | 23 +- docs/contracts/send-chain-r10-completion.json | 2 +- docs/contracts/upstream-r7-declarations.json | 6 +- .../first-version-development-requirements.md | 2 + docs/production-deployment.md | 9 +- docs/system-functional-test-cases.md | 8 + docs/testing-progress.md | 14 + gateway/README.md | 1 + gateway/cmd/gateway/main.go | 48 +++ gateway/internal/metrics/metrics.go | 12 + gateway/internal/metrics/metrics_test.go | 6 +- gateway/internal/resultoutbox/outbox.go | 260 +++++++++++++++++ gateway/internal/resultoutbox/outbox_test.go | 129 ++++++++ gateway/internal/resultoutbox/worker.go | 275 ++++++++++++++++++ gateway/internal/submitworker/worker.go | 53 +++- gateway/internal/submitworker/worker_test.go | 25 +- gateway/internal/upstream/manager.go | 12 +- gateway/internal/upstream/submit.go | 63 +--- .../spike/validate-gateway-queue-contract.mjs | 11 + 27 files changed, 1000 insertions(+), 66 deletions(-) create mode 100644 api/prisma/migrations/20260820130000_add_submit_result_idempotency/migration.sql create mode 100644 docs/contracts/examples/submit-result-outbox-event.json create mode 100644 gateway/internal/resultoutbox/outbox.go create mode 100644 gateway/internal/resultoutbox/outbox_test.go create mode 100644 gateway/internal/resultoutbox/worker.go diff --git a/.env.example b/.env.example index d922d97..e93f6a5 100644 --- a/.env.example +++ b/.env.example @@ -43,5 +43,9 @@ GATEWAY_CMPP_ADDR=127.0.0.1:7890 GATEWAY_CMPP_USER=900001 GATEWAY_CMPP_PASSWORD=888888 GATEWAY_SUBMIT_WORKER_CONCURRENCY=64 +GATEWAY_SUBMIT_RESULT_STREAM=gateway.submit.results +GATEWAY_SUBMIT_RESULT_GROUP=cmpp-api-callback +GATEWAY_SUBMIT_RESULT_CONSUMER=gateway-1 +GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY=8 GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY=64 CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS=30 diff --git a/api/prisma/migrations/20260820130000_add_submit_result_idempotency/migration.sql b/api/prisma/migrations/20260820130000_add_submit_result_idempotency/migration.sql new file mode 100644 index 0000000..54b1391 --- /dev/null +++ b/api/prisma/migrations/20260820130000_add_submit_result_idempotency/migration.sql @@ -0,0 +1,6 @@ +ALTER TABLE "SmsSubmitRecord" +ADD COLUMN "resultEventId" TEXT, +ADD COLUMN "resultProcessedAt" TIMESTAMP(3); + +CREATE UNIQUE INDEX "SmsSubmitRecord_resultEventId_key" +ON "SmsSubmitRecord"("resultEventId"); diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index d07323f..ddbd519 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -1760,6 +1760,8 @@ model SmsSubmitRecord { sequenceId Int? gatewayMessageId String? submitStatus String @default("queued") + resultEventId String? @unique + resultProcessedAt DateTime? costUnitPrice BigInt @default(0) costAmountCents BigInt @default(0) errorCode String? diff --git a/api/src/send-chain/send-chain.contracts.ts b/api/src/send-chain/send-chain.contracts.ts index 976b092..4ef2b0d 100644 --- a/api/src/send-chain/send-chain.contracts.ts +++ b/api/src/send-chain/send-chain.contracts.ts @@ -58,6 +58,7 @@ export interface GatewayInboundSingleSubmitResult { } export interface GatewaySubmitResultDto { + eventId?: string; traceId?: string; messageId: string; channelId: string; @@ -81,6 +82,7 @@ export interface GatewaySubmitResultDto { } export interface GatewaySubmitSegmentResultDto { + eventId?: string; traceId?: string; messageId: string; channelId: string; diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 5cf1cb2..7040514 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -2216,6 +2216,46 @@ describe('SendChainService', () => { }); }); + it('treats a redelivered Outbox aggregate event as idempotent', async () => { + const { service, prisma, billing } = createService(); + const baseSubmit = { + id: 'submit-1', + messageRecordId: 'record-1', + channelId: 'channel-1', + submitId: 'SUB-1', + submitStatus: 'accepted', + }; + prisma.smsSubmitRecord.findUnique + .mockResolvedValueOnce({ ...baseSubmit, resultEventId: null }) + .mockResolvedValueOnce({ ...baseSubmit, resultEventId: 'submit:SUB-1:aggregate' }); + const event = { + eventId: 'submit:SUB-1:aggregate', + messageId: 'MSG-1', + channelId: 'channel-1', + submitId: 'SUB-1', + sequenceId: 7, + gatewayMessageId: 'GW-1', + submitStatus: 'accepted' as const, + submittedAt: '2026-07-01T10:00:00.000Z', + }; + + await service.handleSubmitResult(event); + await service.handleSubmitResult(event); + + expect(billing.charge).toHaveBeenCalledTimes(1); + expect(billing.release).toHaveBeenCalledTimes(1); + expect(prisma.smsSubmitRecord.updateMany).toHaveBeenCalledWith({ + where: { + id: 'submit-1', + OR: [{ resultEventId: null }, { resultEventId: 'submit:SUB-1:aggregate' }], + }, + data: { + resultEventId: 'submit:SUB-1:aggregate', + resultProcessedAt: new Date('2026-07-01T10:00:00.000Z'), + }, + }); + }); + it('rejects a legacy aggregate SubmitResult when it cannot match one submit attempt uniquely', async () => { const { service, prisma } = createService(); prisma.smsSubmitRecord.findMany.mockResolvedValue([ diff --git a/api/src/send-chain/send-gateway-result.service.ts b/api/src/send-chain/send-gateway-result.service.ts index b28dc4f..c8b9763 100644 --- a/api/src/send-chain/send-gateway-result.service.ts +++ b/api/src/send-chain/send-gateway-result.service.ts @@ -128,6 +128,12 @@ export class SendGatewayResultService { async handleSubmitResult(data: GatewaySubmitResultDto) { const message = await this.facade.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId); const submitRecord = await this.facade.resolveSubmitRecordForGatewayResult(message.id, data); + if (data.eventId && submitRecord.resultEventId) { + if (submitRecord.resultEventId !== data.eventId) { + throw new BadRequestException('Gateway SubmitResult eventId conflicts with the submit attempt'); + } + return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); + } const effectiveData = { ...data, submitId: submitRecord.submitId }; const batchTask = message.batchTaskId ? await this.prisma.smsBatchTask.findUnique({ where: { id: message.batchTaskId }, select: { sourceType: true } }) @@ -147,6 +153,7 @@ export class SendGatewayResultService { await this.facade.recordSubmitSegments(message, effectiveData, submittedAt); const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId; if (message.submitId && effectiveData.submitId !== message.submitId) { + await this.markSubmitResultProcessed(submitRecord.id, data.eventId, submittedAt); return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); } const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed'; @@ -166,6 +173,7 @@ export class SendGatewayResultService { ); if (retried) { await this.facade.refreshTaskProgress(businessMessage.batchTaskId); + await this.markSubmitResultProcessed(submitRecord.id, data.eventId, submittedAt); return retried; } await this.facade.releaseMessageReservation(businessMessage, data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结'); @@ -218,9 +226,29 @@ export class SendGatewayResultService { if (message.batchTaskId) { await this.facade.refreshTaskProgress(message.batchTaskId); } + await this.markSubmitResultProcessed(submitRecord.id, data.eventId, submittedAt); return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); } + private async markSubmitResultProcessed(submitRecordId: string, eventId: string | undefined, processedAt: Date) { + if (!eventId) { + return; + } + const updated = await this.prisma.smsSubmitRecord.updateMany({ + where: { + id: submitRecordId, + OR: [{ resultEventId: null }, { resultEventId: eventId }], + }, + data: { + resultEventId: eventId, + resultProcessedAt: processedAt, + }, + }); + if (updated.count === 0) { + throw new BadRequestException('Gateway SubmitResult eventId conflicts with the submit attempt'); + } + } + async resolveSubmitRecordForGatewayResult(messageRecordId: string, data: GatewaySubmitResultDto) { if (data.submitId) { const exact = await this.prisma.smsSubmitRecord.findUnique({ where: { submitId: data.submitId } }); diff --git a/docs/codebase-modularization-roadmap.md b/docs/codebase-modularization-roadmap.md index ae2082f..61f5dac 100644 --- a/docs/codebase-modularization-roadmap.md +++ b/docs/codebase-modularization-roadmap.md @@ -1153,5 +1153,6 @@ global，不能错误归入client。`client-signature-*`、发送页、企业认 - CMPP性能分段沿用上述边界：业务模块只在原调用边界提交固定阶段名、成功标志和单调时钟耗时；指标模块拒绝未知阶段。`supplier_rtt`在供应商连接调用结束时立即停止，API结果回写另计`api_callback`，防止后续优化依据混杂总耗时。 - V2提交工作池只归`gateway/internal/submitworker/`治理：Redis领取、全局槽位、在途消息ID和逐条ACK不能渗入上游连接池；`gateway/internal/upstream/`继续只负责通道连接、窗口和供应商协议往返。这样Worker吞吐调优不会改写CMPP连接状态机，连接池也不能自行确认Redis消息。 - V3客户入站窗口只归`gateway/internal/inbound/`与项目内受控的`third_party/gocmpp`服务循环治理：API认证只返回应用窗口，inbound会话负责收紧窗口，协议服务循环负责受限派发和断线等待；不得把客户入站槽位与`submitworker`供应商槽位或`upstream`供应商窗口合并成同一并发计数。 +- V4供应商结果异步边界只归`gateway/internal/resultoutbox/`治理：`upstream`在每个真实分片SubmitResp后只调用持久化接口，`submitworker`只负责聚合结果入Outbox与命令ACK的原子边界，Outbox回调Worker独立控制API并发和PEL恢复。API的`SmsSubmitRecord.resultEventId`是跨重启持久幂等事实；不得把HTTP回调重新放回供应商连接池或Submit工作槽，也不得让Outbox承担业务计费、补发或状态机判断。 - `api/src/infrastructure-monitoring/`是运营端监控聚合与固定阈值应用边界：只消费代码白名单 PromQL，并通过版本化 PostgreSQL 单例、promtool 校验和原子规则热加载管理数值阈值；Exporter安装、端口隔离、固定规则模板和权限仍归`tools/monitoring/`治理。 - 活动告警已读也归该边界：Prometheus保留告警事实，Prisma仅持久化逐管理员、逐触发周期的阅读状态；全局布局只消费轻量未读汇总，不复制指纹、activeAt或用户隔离逻辑。 diff --git a/docs/contracts/examples/submit-result-outbox-event.json b/docs/contracts/examples/submit-result-outbox-event.json new file mode 100644 index 0000000..11fdb63 --- /dev/null +++ b/docs/contracts/examples/submit-result-outbox-event.json @@ -0,0 +1,24 @@ +{ + "schemaVersion": "v1", + "eventId": "submit:submit-20260820-0001:aggregate", + "eventType": "submit_result", + "path": "/gateway/events/submit-result", + "traceId": "trace-20260820-0001", + "messageId": "message-20260820-0001", + "channelId": "channel-test-1", + "submitId": "submit-20260820-0001", + "payload": { + "eventId": "submit:submit-20260820-0001:aggregate", + "schemaVersion": "v1", + "messageType": "SubmitResult", + "traceId": "trace-20260820-0001", + "messageId": "message-20260820-0001", + "channelId": "channel-test-1", + "createdAt": "2026-08-20T05:00:00Z", + "sequenceId": 1, + "gatewayMessageId": "1000000000000001", + "submitStatus": "accepted", + "submittedAt": "2026-08-20T05:00:00Z" + }, + "createdAt": "2026-08-20T05:00:00Z" +} diff --git a/docs/contracts/gateway-queue-messages.schema.json b/docs/contracts/gateway-queue-messages.schema.json index b782693..696998b 100644 --- a/docs/contracts/gateway-queue-messages.schema.json +++ b/docs/contracts/gateway-queue-messages.schema.json @@ -6,7 +6,8 @@ { "$ref": "#/$defs/SubmitCommand" }, { "$ref": "#/$defs/SubmitResult" }, { "$ref": "#/$defs/ReceiptEvent" }, - { "$ref": "#/$defs/UplinkEvent" } + { "$ref": "#/$defs/UplinkEvent" }, + { "$ref": "#/$defs/SubmitResultOutboxEvent" } ], "$defs": { "Envelope": { @@ -139,6 +140,26 @@ } ] }, + "SubmitResultOutboxEvent": { + "type": "object", + "required": ["schemaVersion", "eventId", "eventType", "path", "messageId", "channelId", "submitId", "payload", "createdAt"], + "properties": { + "schemaVersion": { "const": "v1" }, + "eventId": { "type": "string", "pattern": "^submit:.+:(aggregate|segment:[1-9][0-9]*)$" }, + "eventType": { "enum": ["submit_result", "submit_segment_result"] }, + "path": { "enum": ["/gateway/events/submit-result", "/gateway/events/submit-segment-result"] }, + "traceId": { "type": "string" }, + "messageId": { "type": "string", "minLength": 1 }, + "channelId": { "type": "string", "minLength": 1 }, + "submitId": { "type": "string", "minLength": 1 }, + "payload": { + "type": "object", + "required": ["eventId"], + "properties": { "eventId": { "type": "string", "minLength": 1 } } + }, + "createdAt": { "type": "string", "format": "date-time" } + } + }, "ReceiptEvent": { "allOf": [ { "$ref": "#/$defs/Envelope" }, diff --git a/docs/contracts/send-chain-r10-completion.json b/docs/contracts/send-chain-r10-completion.json index 84911f7..5828263 100644 --- a/docs/contracts/send-chain-r10-completion.json +++ b/docs/contracts/send-chain-r10-completion.json @@ -17,7 +17,7 @@ { "name": "handleSubmitResult", "file": "send-gateway-result.service.ts", - "bodySha256": "6184170748eb3934496b060ea3b6704f1d1a831e7ffde7e7f8b73a7c223a869c" + "bodySha256": "9414f8d4e5e1c3cf6d0747e47ce01e101e142772af4f56900932bc6d4ac11ec7" }, { "name": "resolveSubmitRecordForGatewayResult", diff --git a/docs/contracts/upstream-r7-declarations.json b/docs/contracts/upstream-r7-declarations.json index 6c1e917..59ee6e5 100644 --- a/docs/contracts/upstream-r7-declarations.json +++ b/docs/contracts/upstream-r7-declarations.json @@ -173,7 +173,7 @@ "name": "Manager", "kind": "type", "file": "manager.go", - "sha256": "c7ead8b037bc87ad9800650ce191f5fcee9648b8805a133b9506743c762ea608" + "sha256": "b08726201c34da87ae735abd7fc56d2b904104d4b7f6a6a6365263ec5b0a43bc" }, { "name": "defaultChannelConnectionID", @@ -389,7 +389,7 @@ "kind": "func", "receiver": "Manager", "file": "submit.go", - "sha256": "a25d30aeb87c1fe123929330fdabc99261e63d5b51e1cb6207197c43a15de07d" + "sha256": "9be92b0d32626ce93d0b2d721bf008305c4b9e5919e287d49b91c6f2474ddeb1" }, { "name": "submitPart", @@ -410,7 +410,7 @@ "kind": "func", "receiver": "connectionPool", "file": "submit.go", - "sha256": "e3dbf2d433ef4ff95b2fe7a8494b17c6a8e185bdace017cce9c7d2e8c785bfb3" + "sha256": "577b7f942f4b95ea3868d7cf9aadae800df8edda0e7ca94a205b4124be58e9e3" }, { "name": "defaultInt", diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 374a7b7..6a1b68d 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -240,6 +240,7 @@ 9. Gateway 必须按通道连接和窗口容量控制并发，处理窗口满、SMSC 慢响应、sequence 回绕、连接断开时的在途消息状态。 10. Gateway 必须对每个物理通道执行 Redis 分布式 TPS 限速。`SmsChannel.rateLimitPerSecond` 是单通道上限，不是平台总上限；同一通道被多个通道组或多个 Gateway 实例使用时共享同一额度，不同通道独立计数。通道连接命令下发的配置值是最终上限，提交命令携带的值只能进一步降低、不能放大该上限。超流速消息必须继续保留在 Redis Stream pending 中等待可用时隙，不能因等待直接标记发送失败；Gateway 重启后仍可由 consumer group 恢复。worker 必须使用持续补位的全局有界工作池，任一任务完成后立即领取后续消息，不得以“读取10条、等待整批结束”形成批次屏障；每条消息只在自身提交结果已持久回传或完成死信处理后独立`XACK`。全局并发默认64、可由`GATEWAY_SUBMIT_WORKER_CONCURRENCY`配置且上限1024；单通道实际并发仍由 Redis 限速、连接数和 CMPP 窗口共同约束。恢复pending时必须防止超过`MinIdle`的在途消息被同一进程重复提交。 11. Gateway 不承担业务审核、计费、签名报备、通道组路由、黑名单或敏感词判断；这些由 NestJS 完成，Gateway 只执行已授权通道提交与协议事件回传。 +12. 供应商SubmitResp及长短信分片结果必须先进入共享Redis的幂等Outbox，再由独立有界回调Worker异步上报API；供应商Submit工作槽只等待连接窗口、限速和真实供应商往返，不得等待API结果回写。每个分片结果必须在继续发送下一分片前写入Outbox；聚合结果入Outbox与原`gateway.submit.commands`消息`XACK`必须原子完成。事件ID按submitId和分片序号确定生成，重复发布不得产生第二个事件；API必须按事件ID持久幂等，重复回调不得重复扣费、释放余额、补发或推进状态。回调失败保留在Outbox PEL并可在进程重启后恢复，成功后逐条ACK并删除；去重键和并发必须有界配置，不得形成无限内存或Stream归档。 #### 4.8.2 下游客户 CMPP 接入能力 @@ -2100,6 +2101,7 @@ - Gateway供应商下发必须分别记录`stream_wait/rate_limit_wait/connection_wait/supplier_rtt/api_callback`；其中`supplier_rtt`只覆盖供应商连接上的Submit请求与SubmitResp往返，不得包含结果回写API的耗时。阶段名和结果值必须使用代码固定白名单，不得添加手机号、企业、应用、通道、消息、任务或连接ID标签。 - V2有界工作池必须暴露配置槽位数和当前在途槽位数，使用固定`state=configured|in_flight`标签；该指标用于区分Worker容量耗尽与供应商窗口/限速等待，不得增加通道或消息标签。 - V3入站并发必须只并发Submit业务处理，连接认证保持串行先完成，心跳和Deliver ACK不得被长耗时Submit阻塞；每个SubmitResp继续使用原请求Sequence_Id关联，允许按实际完成顺序返回。同一连接关闭时必须先等待已接受的在途处理收尾，再清理会话和回执映射，避免迟到处理重新注册已断开的连接。`cmpp_gateway_inbound_submit_slots{state=configured|in_flight}`只暴露全部在线连接的聚合窗口与在途数量，不得增加账号、应用、连接或消息标签。 +- V4结果Outbox必须暴露独立回调Worker的configured/in_flight槽位和Outbox pending/lag，仍只使用固定状态标签。`api_callback`从V4起只在Outbox回调Worker计时，不再混入供应商Submit工作槽；压测结束必须同时核对命令Stream和结果Outbox均`pending=0/lag=0`，并证明API回调故障时供应商槽继续释放、结果不丢失且恢复后只处理一次。 - 系统监控标题说明必须明确标注数据来自 Prometheus；“服务关键指标”位于趋势/核心服务区域之后、活动告警之前，并提供统一的“告警阈值设置”入口。安全检测页不重复渲染大号标题，说明文字必须明确标注使用 Fail2ban。 - 告警阈值仅开放固定指标的警告/严重数值，不允许前端提交 PromQL、标签、文件路径或持续时间；必须满足警告值小于严重值。配置以 PostgreSQL 保存版本、期望值、生效值和应用状态，经 `promtool check rules` 校验、同目录原子替换和 Prometheus 热加载成功后才标记生效，失败保留上一生效规则并展示原因。 - 右上角预警中心增加“系统监控告警”，通过独立轻量接口统计 Prometheus 当前 firing/pending 告警及严重数，跳转系统监控活动告警区；任一预警域失败不得清空其他域。 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 5dfde1f..393217c 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -61,6 +61,10 @@ CMPP_PUBLIC_HOST=8.160.169.106 CMPP_PUBLIC_PORT=17890 GATEWAY_STARTUP_RECONNECT_DELAY_MS=1000 GATEWAY_SUBMIT_WORKER_CONCURRENCY=64 +GATEWAY_SUBMIT_RESULT_STREAM=gateway.submit.results +GATEWAY_SUBMIT_RESULT_GROUP=cmpp-api-callback +GATEWAY_SUBMIT_RESULT_CONSUMER=gateway-1 +GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY=8 GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY=64 OBJECT_STORAGE_DRIVER=minio OBJECT_STORAGE_LOCAL_ROOT=/var/lib/cmpp-platform/object-storage @@ -85,6 +89,8 @@ Gateway 的最终 TPS 防线依赖与 API 相同的 Redis。通道连接时会 V3起客户CMPP入站Submit按应用`cmppWindowSize`在单连接内受限并发，全局上限`GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY`缺省64、最大1024。发布后必须先以已认证测试连接确认`cmpp_gateway_inbound_submit_slots{state="configured"}`等于应用有效窗口，再执行阶梯压测；不得通过调高全局值绕过应用窗口，也不得在未验证Sequence_Id关联、心跳/ACK活性和断线清理时直接提高生产并发。 +V4起供应商分片和聚合结果写入`gateway.submit.results`幂等Outbox，由默认8槽、最大1024的独立回调Worker上报API。聚合结果XADD与原命令XACK由Redis Lua原子执行；回调成功后结果事件XACK+XDEL，失败事件留在PEL。发布时必须保留同一Redis数据和AOF，不得清理`gateway.submit.results`、其consumer group或`gateway.submit.results:dedupe:*`；调整`GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY`前须核对API/PostgreSQL承载能力。第91条向前兼容migration增加`SmsSubmitRecord.resultEventId/resultProcessedAt`，用于回调跨重启幂等，不删除历史字段或数据。 + 服务重启顺序必须是 Gateway 在前、API 在后。API 启动后等待 `GATEWAY_STARTUP_RECONNECT_DELAY_MS`（默认 1 秒），从 PostgreSQL 读取全部 active 通道并重新下发真实连接命令，同时恢复 Gateway 内存连接池和 Redis 权威 TPS key；禁止沿用数据库中重启前的 connected 状态冒充当前连接。 日报任务默认启用，并由 `REPORT_REFRESH_INTERVAL_MS` 每小时检查一次北京时间业务日是否变化；每个业务日只执行一次 T-4 至 T-1 重算。服务重启后也会自动补跑最近四个完整自然日，确保 72 小时回执更新反映到对账和利润报表。 @@ -145,9 +151,10 @@ curl http://127.0.0.1:8090/health curl http://127.0.0.1:12026/ redis-cli -h 127.0.0.1 -p 6379 ping pg_isready -d "$(grep '^DATABASE_URL=' /etc/cmpp-platform/cmpp-platform.env | cut -d= -f2-)" -grep -E '^(API_ENABLE_SEND_WORKER|API_SEND_WORKER_CONCURRENCY|GATEWAY_SUBMIT_WORKER_CONCURRENCY|GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY)=' /etc/cmpp-platform/cmpp-platform.env +grep -E '^(API_ENABLE_SEND_WORKER|API_SEND_WORKER_CONCURRENCY|GATEWAY_SUBMIT_WORKER_CONCURRENCY|GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY|GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY)=' /etc/cmpp-platform/cmpp-platform.env redis-cli --scan --pattern 'rate:gateway:channel:*' redis-cli XINFO GROUPS gateway.submit.commands +redis-cli XINFO GROUPS gateway.submit.results ``` ## 回滚 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index c083a1f..c0c5aba 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4682,6 +4682,14 @@ npm run verify:phase8 | TC-CMPP-PERF-V3-004 | 断线在途清理 | Submit进入API后由客户端断开TCP，随后让API处理完成 | Gateway等待已接受处理收尾后再执行连接关闭回调；会话、Submit barrier和消息映射最终清理，不重新出现幽灵连接，不发生panic | | TC-CMPP-PERF-V3-005 | 入站槽位聚合指标 | 建立不同窗口的测试连接并制造部分在途Submit后抓取metrics | `cmpp_gateway_inbound_submit_slots`的configured等于在线连接有效窗口合计、in_flight等于当前业务处理数；只有固定state标签 | | TC-CMPP-PERF-V3-006 | V3阶梯容量复测 | 在与V2相同的隔离真实后端/数据库/Redis/供应商模拟器中执行10、20、30、40、50条/秒各60秒 | 与V2按同口径比较SubmitResp P50/P95/P99、API阶段、Stream pending/lag、供应商吞吐和资源；遇P95超过5秒、持续积压或服务异常立即停止，未执行档位不记为通过 | +| TC-CMPP-PERF-V4-001 | 分片结果幂等入Outbox | 对同一submitId和segmentIndex重复发布两次分片结果 | `gateway.submit.results`只新增一个确定性eventId事件；下一分片仅在前一分片Outbox写入成功后发送 | +| TC-CMPP-PERF-V4-002 | 聚合结果与命令ACK原子性 | 让供应商返回成功，在聚合结果写入与命令ACK边界注入Redis失败并重启Gateway | 不存在“命令已ACK但结果Outbox缺失”状态；重试同一发布脚本不会产生重复聚合事件 | +| TC-CMPP-PERF-V4-003 | API回调不占供应商槽 | API submit-result接口延迟10秒，同时持续让供应商快速返回SubmitResp | 供应商Worker槽在结果写入Outbox后立即释放；API延迟只增加独立回调Worker和Outbox积压，不降低供应商Submit槽可继续补位的能力 | +| TC-CMPP-PERF-V4-004 | 回调失败恢复与逐条ACK | 结果回调第一次返回503，随后恢复201并重启回调Worker | 失败事件保留PEL且未删除；恢复后重新投递，成功时单事件原子ACK+删除，其他事件不受整批等待 | +| TC-CMPP-PERF-V4-005 | API持久幂等 | 对同一聚合eventId重复回调两次，并检查提交记录、计费、余额、重试和任务进度 | `SmsSubmitRecord.resultEventId`只记录一次；第二次直接返回当前结果，不重复扣费、释放、补发或推进状态；不同eventId占用同一submit尝试时拒绝 | +| TC-CMPP-PERF-V4-006 | Outbox有界指标 | 制造回调在途和积压后抓取Gateway metrics | 回调Worker configured/in_flight与真实槽位一致，Outbox pending/lag与Redis consumer group一致，指标不含手机号、消息、submit、企业、应用或通道标签 | +| TC-CMPP-PERF-V4-007 | 双Stream发布后排空 | 完成一档隔离压测并等待异步处理结束 | `gateway.submit.commands`和`gateway.submit.results`均为`pending=0/lag=0`，结果Outbox成功事件已删除，无Gateway Submit死信；数据库业务数与客户端完全一致 | +| TC-CMPP-PERF-V4-008 | V4阶梯容量复测 | 在V3相同隔离供应商模拟器与真实API/PostgreSQL/Redis/Gateway中依次执行10、20、30、40、50条/秒各60秒 | 对比V3的SubmitResp分位、供应商RTT、命令Stream、结果Outbox、API阶段和资源；任一档出现拒绝、连接错误、P95超过5秒、双Stream持续增长或服务异常立即停止升档 | | TC-GLOBAL-ALERT-001 | 铃铛分域预警菜单 | 准备签名清退未读消息和安全待处置告警后点击右上角铃铛 | 弹层分开显示“签名清退预警”和“安全检测与封禁”，分别展示真实数量和摘要，角标等于两项之和 | | TC-GLOBAL-ALERT-002 | 预警菜单跳转 | 分别点击铃铛中的两个菜单项 | 签名项跳转`/admin/signature-retirement`，安全项跳转`/admin/security-detection`，弹层关闭且对应页面读取真实后端数据 | | TC-GLOBAL-ALERT-003 | 域间故障隔离与轻量轮询 | 分别让一个汇总接口失败并观察30秒轮询请求 | 失败域显示0且另一域数据保留；安全预警使用专用汇总接口，不调用完整overview、规则、代理状态或告警大列表 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 5c55d57..4144389 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -3747,3 +3747,17 @@ git diff --check - 优先级隔离仍未通过：40条/秒priority P95约3912ms、normal P95约3911ms，且两次SubmitResp拒绝都发生在priority连接；全部5996条均为mobile，联通/电信及六通道容量仍待号段识别修复后测试。原始目录名继续沿用既有`lg-v2-*`脚本标签，但被测运行标识和代码均为V3修复版，正式结论以运行标识为准。 - V3回归已通过Gateway全量`go test ./... -count=1`、`go vet ./...`、项目内gocmpp测试/vet、API TypeScript正式编译、真实隔离Redis上的SendChain 113/113、4份Stream契约、R6 102声明/14项关键测试、R7、SendChain R10及`git diff --check`。Linux测试机补跑`-race`时因网络无法下载仅供测试的`miniredis`和`golang.org/x/text`依赖而阻塞，未伪报通过，也未伪造外部依赖。 - 本轮只使用隔离供应商模拟器，没有发送、补发或重投真实短信，没有修改通道账号、密码、启停状态、企业余额或客户连接。`api/tsconfig.build.tsbuildinfo`、`tsconfig.tsbuildinfo`、`outputs/`、`pnpm-lock.yaml`和空文件`=`继续作为受保护项排除提交。 + +# 2026-08-20 CMPP压测优化V4：供应商结果幂等Outbox、测试环境发布与阶梯复测 + +- V4将供应商分片结果和聚合结果先写入Redis Stream `gateway.submit.results`，API回调由独立持续有界工作池执行，供应商Submit工作槽不再等待API往返。分片事件ID为`submit::segment:`，聚合事件ID为`submit::aggregate`；Lua脚本保证分片去重键与XADD原子、聚合XADD与原命令XACK原子、成功回调XACK与XDEL原子。失败事件保留在PEL并由`XAUTOCLAIM`恢复，回收阈值30秒高于10秒HTTP超时，避免多Gateway实例在回调仍执行时并发重领。 +- 第91条向前兼容migration `20260820130000_add_submit_result_idempotency`为`SmsSubmitRecord`增加唯一可空`resultEventId`和`resultProcessedAt`。API重复收到同一已完成事件直接返回当前消息，冲突事件ID拒绝；测试覆盖重复回调不重复计费、重试或下游业务副作用。新增第5类Gateway队列契约示例，Outbox指标覆盖回调Worker槽位和结果Stream pending/lag。 +- 本地验证通过：Gateway全量`go test ./... -count=1`及`go vet ./...`，API 42套/483项断言全部通过，API与前端TypeScript正式编译、Vite生产构建、5份队列契约、R0/R6/R7及SendChain R10结构门禁、`git diff --check`均通过。全量Jest在断言完成后仍因仓库既有异步句柄不自行退出，本轮未把人工终止后的进程伪报为完整退出码0；使用本机真实Redis补跑，不伪造外部依赖。 +- 仅发布到`100.93.204.60`虚拟机测试环境；预生产`8.160.169.106`只读复核标识保持`433b2ee56f6016ad8afff1bac73f510b8fd53083+gateway-v2.cd7bb8d05e7b`，未发布、未回退、未压测。测试环境发布前恢复资产位于`/opt/cmpp-platform-backups/v4-20260820T052241Z`，包含PostgreSQL自定义格式备份、运行源码、环境和systemd配置；`pg_restore --list`、源码tar可读性及`SHA256SUMS`全部通过。 +- 最终运行包`outputs/cmpp-v4-runtime-final-20260820-141323.tar.gz`共827项、1641927字节，本地与测试机SHA-256均为`a63885439ed53d53a3a012048d5fd1a9aa67772503fd778ec510ec6a38a9944f`，排除依赖、构建产物、`outputs`、`*.tsbuildinfo`、`pnpm-lock.yaml`和空文件`=`。测试环境运行标识为`485af688d21ee0bdb98281f4c20d151d69f7889f+workspace.v4.a63885439ed5`；第91条migration仅应用一次，API、Gateway、安全代理、MinIO、PostgreSQL和Redis健康，最终供应商连接6/6，两条Stream均`pending=0/lag=0`。 +- 首轮默认32槽10条/秒恰逢重启积压回执回放，599/599受理但回执2536、P95=3294ms；积压排空后再测仍为P95=671ms。把测试配置收敛为8槽后，10条/秒599/599、P50/P95/P99=`39/146/304ms`、回执603。基于该对照，代码、示例环境和发布文档的默认值同步改为8；32槽不作为推荐配置。 +- 8槽正式阶梯结果：20条/秒1199/1199受理、P50/P95/P99=`45/1129/1259ms`；30条/秒1799/1799受理、`1081/1626/1885ms`；40条/秒2399/2399受理、`2049/2330/2386ms`。三档均无连接错误；相对V3，20/30/40条/秒P95分别由1427/2033/3912ms降至1129/1626/2330ms，且40条/秒从2条10秒超时改进为零拒绝，确认安全档位由30提升到40条/秒。 +- 解耦证据：20条/秒命令Stream未投递lag峰值0，结果Outbox峰值`pending=8/lag=242`后归零；30条/秒峰值约`pending=8/lag=1076`；40条/秒峰值约`pending=9/lag=1639`，压测结束后约36秒归零并连续保持0/0。40条/秒时供应商命令与结果回调已分开排队，结果回调积压不再占用供应商Submit槽；但Outbox回落时间已成为容量判定的一部分，不能只看客户SubmitResp。 +- 50条/秒触发停止线：因客户端背压只生成2790条而非约2999条，2787条成功、3条等待API满10秒后返回result=9，P50/P95/P99=`6765/7236/7789ms`，throttled ticks=3184。结束后命令Stream一度`pending=64/lag=489`、结果Outbox`pending=8/lag=1690`；命令约1分钟内、结果随后约20秒内归零。日志显示API饱和时协议遥测和连接状态回调超时，并暴露既有`CmppDownstreamConnection`创建/更新的P2002/P2025竞态；未出现`resultEventId`唯一键冲突或Outbox事件失败。停止后未继续上探。 +- 整个窗口客户侧共9984条业务提交，真实PostgreSQL按`queuedAt`精确新增9984条；窗口内7922条实际供应商提交记录全部具有非空且互不重复的`resultEventId`，重复事件组0、Gateway Submit死信0。提交结果为accepted 7392、rejected 147、timeout 383；客户有限回执收集窗口和重连积压回执不替代数据库/Stream对账。最终确认40条/秒为当前测试环境安全档，50条/秒瓶颈转为API/数据库同步入站及遥测争用，后续应继续V1的重复查询/零散写入合并和P1分阶段指标分析。 +- 本轮只连接隔离供应商模拟器`100.91.249.119:17900`，没有发送、补发或重投真实短信，没有修改真实通道账号、密码、启停状态、企业余额或客户连接。压测原始目录沿用既有`lg-v2-*`名称，但被测运行标识与结论均为V4；完整V4报告保存在短信平台测试项目。受保护的`api/tsconfig.build.tsbuildinfo`、`tsconfig.tsbuildinfo`、`outputs/`、`pnpm-lock.yaml`和空文件`=`继续排除提交、不删除、不归因。 diff --git a/gateway/README.md b/gateway/README.md index 208ffe7..330834b 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -13,6 +13,7 @@ - 暴露健康检查和最小指标。 - Redis Stream Submit Worker 使用持续补位的有界工作池并逐条 ACK；默认并发64，可用`GATEWAY_SUBMIT_WORKER_CONCURRENCY`调整，最大1024。供应商通道的真实上限仍由TPS限速、连接数和CMPP窗口共同决定。 - 客户CMPP入站Submit按认证接口返回的应用`cmppWindowSize`在单连接内并发，Gateway再以`GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY`实施默认64、最大1024的全局单连接保护。登录保持串行，心跳和Deliver ACK不等待慢Submit；SubmitResp依靠Sequence_Id关联，允许按完成顺序返回。 +- 供应商SubmitResp先写入Redis Stream幂等Outbox`gateway.submit.results`，聚合结果入Outbox与原Submit命令ACK使用同一Lua脚本；独立默认8槽回调Worker再调用API，供应商工作槽不等待API。每个事件使用确定性`eventId`和7天Redis去重键，API在`SmsSubmitRecord.resultEventId`完成持久幂等；成功回调后Outbox事件原子ACK+删除，失败事件留在PEL恢复。 ## 建议骨架 diff --git a/gateway/cmd/gateway/main.go b/gateway/cmd/gateway/main.go index 7459f23..6b73018 100644 --- a/gateway/cmd/gateway/main.go +++ b/gateway/cmd/gateway/main.go @@ -13,7 +13,9 @@ import ( "cmpp-platform/gateway/internal/health" "cmpp-platform/gateway/internal/inbound" platformmetrics "cmpp-platform/gateway/internal/metrics" + "cmpp-platform/gateway/internal/queue" "cmpp-platform/gateway/internal/ratelimit" + "cmpp-platform/gateway/internal/resultoutbox" "cmpp-platform/gateway/internal/submitworker" "cmpp-platform/gateway/internal/upstream" @@ -32,6 +34,7 @@ func main() { apiBaseURL := os.Getenv("API_BASE_URL") upstreamManager := &upstream.Manager{APIBaseURL: apiBaseURL} var worker *submitworker.Worker + var resultOutbox *resultoutbox.Outbox channelLimiter, err := ratelimit.New(os.Getenv("REDIS_URL")) if err != nil { log.Fatalf("gateway channel rate limiter init failed: %v", err) @@ -70,12 +73,26 @@ func main() { worker.Consumer = getenv("GATEWAY_SUBMIT_CONSUMER", "gateway-1") worker.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_WORKER_CONCURRENCY", 64) worker.APIBaseURL = apiBaseURL + resultOutbox = resultoutbox.New(worker.Redis) + resultOutbox.Stream = getenv("GATEWAY_SUBMIT_RESULT_STREAM", "gateway.submit.results") + resultOutbox.Group = getenv("GATEWAY_SUBMIT_RESULT_GROUP", "cmpp-api-callback") + resultOutbox.Consumer = getenv("GATEWAY_SUBMIT_RESULT_CONSUMER", "gateway-1") + resultOutbox.APIBaseURL = apiBaseURL + resultOutbox.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY", 8) + worker.ResultOutbox = resultOutbox + upstreamManager.SubmitSegmentPublisher = resultOutbox go func() { log.Printf("cmpp gateway submit worker consuming stream=%s group=%s consumer=%s", worker.Stream, worker.Group, worker.Consumer) if err := worker.Run(context.Background()); err != nil { log.Printf("gateway submit worker stopped: %v", err) } }() + go func() { + log.Printf("cmpp gateway result Outbox consuming stream=%s group=%s consumer=%s", resultOutbox.StreamName(), resultOutbox.GroupName(), resultOutbox.Consumer) + if err := resultOutbox.Run(context.Background()); err != nil { + log.Printf("gateway result Outbox worker stopped: %v", err) + } + }() } } @@ -91,6 +108,11 @@ func main() { snapshot.SubmitWorkerConcurrency = worker.ConfiguredConcurrency() snapshot.SubmitWorkerInFlight = worker.InFlight() } + if resultOutbox != nil { + snapshot.ResultWorkerUp = true + snapshot.ResultWorkerConcurrency = resultOutbox.ConfiguredConcurrency() + snapshot.ResultWorkerInFlight = resultOutbox.InFlight() + } snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight = inbound.SubmitSlotSnapshot() if worker == nil || worker.Redis == nil { return snapshot @@ -117,12 +139,38 @@ func main() { snapshot.QueueOldestAgeSeconds = max(0, time.Since(time.UnixMilli(milliseconds)).Seconds()) } } + if resultOutbox != nil { + resultPending, resultErr := worker.Redis.XPending(ctx, resultOutbox.StreamName(), resultOutbox.GroupName()).Result() + if resultErr == nil { + snapshot.ResultQueueAvailable = true + snapshot.ResultQueuePending = resultPending.Count + } + resultGroups, resultErr := worker.Redis.XInfoGroups(ctx, resultOutbox.StreamName()).Result() + if resultErr == nil { + for _, group := range resultGroups { + if group.Name == resultOutbox.GroupName() { + snapshot.ResultQueueLag = group.Lag + break + } + } + } + } return snapshot })) control.Register(mux, control.Server{ APIBaseURL: apiBaseURL, Upstream: upstreamManager, Limiter: channelLimiter, + Submit: func(ctx context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) { + result, submitErr := upstreamManager.Submit(ctx, command) + if resultOutbox == nil { + return result, submitErr + } + if publishErr := resultOutbox.PublishSubmitResult(ctx, command, result); publishErr != nil { + return result, publishErr + } + return result, submitErr + }, RecoveryCandidates: func(ctx context.Context) ([]inbound.DownstreamPresence, error) { return inbound.ListRecoveryCandidates(ctx, presenceStore) }, diff --git a/gateway/internal/metrics/metrics.go b/gateway/internal/metrics/metrics.go index cd34b94..f3f2d2b 100644 --- a/gateway/internal/metrics/metrics.go +++ b/gateway/internal/metrics/metrics.go @@ -43,12 +43,18 @@ type Snapshot struct { SubmitWorkerUp bool SubmitWorkerConcurrency int SubmitWorkerInFlight int64 + ResultWorkerUp bool + ResultWorkerConcurrency int + ResultWorkerInFlight int64 InboundSubmitConcurrency int InboundSubmitInFlight int64 QueueAvailable bool QueuePending int64 QueueLag int64 QueueOldestAgeSeconds float64 + ResultQueueAvailable bool + ResultQueuePending int64 + ResultQueueLag int64 } type SnapshotFunc func(context.Context) Snapshot @@ -118,12 +124,18 @@ func Handler(load SnapshotFunc) http.Handler { fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp)) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_slots Configured and active bounded submit worker slots.\n# TYPE cmpp_gateway_submit_worker_slots gauge\ncmpp_gateway_submit_worker_slots{state=\"configured\"} %d\ncmpp_gateway_submit_worker_slots{state=\"in_flight\"} %d\n", snapshot.SubmitWorkerConcurrency, snapshot.SubmitWorkerInFlight) + fmt.Fprintf(response, "# HELP cmpp_gateway_result_callback_worker_up Whether the asynchronous result callback worker was initialized.\n# TYPE cmpp_gateway_result_callback_worker_up gauge\ncmpp_gateway_result_callback_worker_up %d\n", boolNumber(snapshot.ResultWorkerUp)) + fmt.Fprintf(response, "# HELP cmpp_gateway_result_callback_worker_slots Configured and active result callback worker slots.\n# TYPE cmpp_gateway_result_callback_worker_slots gauge\ncmpp_gateway_result_callback_worker_slots{state=\"configured\"} %d\ncmpp_gateway_result_callback_worker_slots{state=\"in_flight\"} %d\n", snapshot.ResultWorkerConcurrency, snapshot.ResultWorkerInFlight) fmt.Fprintf(response, "# HELP cmpp_gateway_inbound_submit_slots Configured and active authenticated client Submit slots.\n# TYPE cmpp_gateway_inbound_submit_slots gauge\ncmpp_gateway_inbound_submit_slots{state=\"configured\"} %d\ncmpp_gateway_inbound_submit_slots{state=\"in_flight\"} %d\n", snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight) if snapshot.QueueAvailable { fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_pending Pending entries owned by the consumer group.\n# TYPE cmpp_gateway_submit_queue_pending gauge\ncmpp_gateway_submit_queue_pending %d\n", snapshot.QueuePending) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_lag Undelivered entries for the consumer group.\n# TYPE cmpp_gateway_submit_queue_lag gauge\ncmpp_gateway_submit_queue_lag %d\n", snapshot.QueueLag) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_oldest_pending_age_seconds Age of the oldest pending entry.\n# TYPE cmpp_gateway_submit_queue_oldest_pending_age_seconds gauge\ncmpp_gateway_submit_queue_oldest_pending_age_seconds %f\n", snapshot.QueueOldestAgeSeconds) } + if snapshot.ResultQueueAvailable { + fmt.Fprintf(response, "# HELP cmpp_gateway_result_outbox_pending Pending result callbacks owned by the consumer group.\n# TYPE cmpp_gateway_result_outbox_pending gauge\ncmpp_gateway_result_outbox_pending %d\n", snapshot.ResultQueuePending) + fmt.Fprintf(response, "# HELP cmpp_gateway_result_outbox_lag Undelivered result callbacks for the consumer group.\n# TYPE cmpp_gateway_result_outbox_lag gauge\ncmpp_gateway_result_outbox_lag %d\n", snapshot.ResultQueueLag) + } }) } diff --git a/gateway/internal/metrics/metrics_test.go b/gateway/internal/metrics/metrics_test.go index 95b165a..edeace9 100644 --- a/gateway/internal/metrics/metrics_test.go +++ b/gateway/internal/metrics/metrics_test.go @@ -16,7 +16,7 @@ func TestMetricsHandlerExportsOnlyAggregateGatewayState(t *testing.T) { request := httptest.NewRequest(http.MethodGet, "/metrics", nil) response := httptest.NewRecorder() Handler(func(context.Context) Snapshot { - return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, SubmitWorkerConcurrency: 64, SubmitWorkerInFlight: 7, InboundSubmitConcurrency: 96, InboundSubmitInFlight: 9, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12} + return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, SubmitWorkerConcurrency: 64, SubmitWorkerInFlight: 7, ResultWorkerUp: true, ResultWorkerConcurrency: 32, ResultWorkerInFlight: 5, InboundSubmitConcurrency: 96, InboundSubmitInFlight: 9, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12, ResultQueueAvailable: true, ResultQueuePending: 6, ResultQueueLag: 8} }).ServeHTTP(response, request) body := response.Body.String() @@ -28,6 +28,10 @@ func TestMetricsHandlerExportsOnlyAggregateGatewayState(t *testing.T) { `cmpp_gateway_submit_stage_duration_seconds_count{stage="rate_limit_wait",result="success"} 1`, `cmpp_gateway_submit_worker_slots{state="configured"} 64`, `cmpp_gateway_submit_worker_slots{state="in_flight"} 7`, + `cmpp_gateway_result_callback_worker_slots{state="configured"} 32`, + `cmpp_gateway_result_callback_worker_slots{state="in_flight"} 5`, + `cmpp_gateway_result_outbox_pending 6`, + `cmpp_gateway_result_outbox_lag 8`, `cmpp_gateway_inbound_submit_slots{state="configured"} 96`, `cmpp_gateway_inbound_submit_slots{state="in_flight"} 9`, } { diff --git a/gateway/internal/resultoutbox/outbox.go b/gateway/internal/resultoutbox/outbox.go new file mode 100644 index 0000000..5e90dbb --- /dev/null +++ b/gateway/internal/resultoutbox/outbox.go @@ -0,0 +1,260 @@ +package resultoutbox + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync/atomic" + "time" + + "cmpp-platform/gateway/internal/queue" + + "github.com/redis/go-redis/v9" +) + +const ( + defaultStream = "gateway.submit.results" + defaultGroup = "cmpp-api-callback" + defaultConsumer = "gateway-1" + defaultDedupeTTL = 7 * 24 * time.Hour +) + +var publishScript = redis.NewScript(` +local inserted = redis.call('SET', KEYS[2], '1', 'NX', 'EX', ARGV[1]) +if inserted then + redis.call('XADD', KEYS[1], '*', 'data', ARGV[2]) + return 1 +end +return 0 +`) + +var publishAndAckScript = redis.NewScript(` +local inserted = redis.call('SET', KEYS[2], '1', 'NX', 'EX', ARGV[1]) +if inserted then + redis.call('XADD', KEYS[1], '*', 'data', ARGV[2]) +end +redis.call('XACK', KEYS[3], ARGV[3], ARGV[4]) +if inserted then return 1 end +return 0 +`) + +type Event struct { + SchemaVersion string `json:"schemaVersion"` + EventID string `json:"eventId"` + EventType string `json:"eventType"` + Path string `json:"path"` + TraceID string `json:"traceId,omitempty"` + MessageID string `json:"messageId"` + ChannelID string `json:"channelId"` + SubmitID string `json:"submitId"` + Payload json.RawMessage `json:"payload"` + CreatedAt time.Time `json:"createdAt"` +} + +type Outbox struct { + Redis *redis.Client + Stream string + Group string + Consumer string + DedupeTTL time.Duration + APIBaseURL string + HTTPTimeout time.Duration + Concurrency int + MinIdle time.Duration + inFlight atomic.Int64 +} + +func New(client *redis.Client) *Outbox { + return &Outbox{Redis: client} +} + +func (o *Outbox) PublishSubmitSegment(ctx context.Context, command queue.SubmitCommand, segment queue.SubmitSegmentResult) error { + payload := struct { + queue.Envelope + SubmitID string `json:"submitId,omitempty"` + queue.SubmitSegmentResult + }{ + Envelope: command.Envelope, + SubmitID: command.SubmitID, + SubmitSegmentResult: segment, + } + event, err := newEvent( + fmt.Sprintf("submit:%s:segment:%d", command.SubmitID, segment.SegmentIndex), + "submit_segment_result", + "/gateway/events/submit-segment-result", + command, + payload, + ) + if err != nil { + return err + } + return o.publish(ctx, event) +} + +func (o *Outbox) PublishSubmitResultAndAck( + ctx context.Context, + commandStream string, + commandGroup string, + commandMessageID string, + command queue.SubmitCommand, + result queue.SubmitResult, +) error { + if o.Redis == nil { + return fmt.Errorf("result Outbox Redis client is required") + } + event, err := newEvent( + fmt.Sprintf("submit:%s:aggregate", command.SubmitID), + "submit_result", + "/gateway/events/submit-result", + command, + result, + ) + if err != nil { + return err + } + data, err := json.Marshal(event) + if err != nil { + return err + } + // XADD and command XACK share one Redis script so a process crash cannot leave + // an acknowledged supplier command without its aggregate result in the Outbox. + _, err = publishAndAckScript.Run( + ctx, + o.Redis, + []string{o.stream(), o.dedupeKey(event.EventID), commandStream}, + int64(o.dedupeTTL().Seconds()), + string(data), + commandGroup, + commandMessageID, + ).Result() + return err +} + +func (o *Outbox) PublishSubmitResult(ctx context.Context, command queue.SubmitCommand, result queue.SubmitResult) error { + event, err := newEvent( + fmt.Sprintf("submit:%s:aggregate", command.SubmitID), + "submit_result", + "/gateway/events/submit-result", + command, + result, + ) + if err != nil { + return err + } + return o.publish(ctx, event) +} + +func (o *Outbox) publish(ctx context.Context, event Event) error { + if o.Redis == nil { + return fmt.Errorf("result Outbox Redis client is required") + } + data, err := json.Marshal(event) + if err != nil { + return err + } + _, err = publishScript.Run( + ctx, + o.Redis, + []string{o.stream(), o.dedupeKey(event.EventID)}, + int64(o.dedupeTTL().Seconds()), + string(data), + ).Result() + return err +} + +func newEvent(eventID string, eventType string, path string, command queue.SubmitCommand, payload any) (Event, error) { + data, err := json.Marshal(payload) + if err != nil { + return Event{}, err + } + var object map[string]interface{} + if err := json.Unmarshal(data, &object); err != nil { + return Event{}, err + } + // The API stores the deterministic ID on the submit attempt after successful + // processing, making callback redelivery idempotent across Gateway restarts. + object["eventId"] = eventID + data, err = json.Marshal(object) + if err != nil { + return Event{}, err + } + return Event{ + SchemaVersion: queue.SchemaVersion, + EventID: eventID, + EventType: eventType, + Path: path, + TraceID: command.TraceID, + MessageID: command.MessageID, + ChannelID: command.ChannelID, + SubmitID: command.SubmitID, + Payload: data, + CreatedAt: time.Now().UTC(), + }, nil +} + +func EventFromStreamValues(values map[string]interface{}) (Event, error) { + raw, ok := values["data"] + if !ok { + return Event{}, fmt.Errorf("result Outbox data field is required") + } + var data string + switch value := raw.(type) { + case string: + data = value + case []byte: + data = string(value) + default: + data = fmt.Sprint(value) + } + var event Event + if err := json.Unmarshal([]byte(data), &event); err != nil { + return Event{}, err + } + if event.SchemaVersion != queue.SchemaVersion || event.EventID == "" || event.MessageID == "" || event.SubmitID == "" { + return Event{}, fmt.Errorf("invalid result Outbox envelope") + } + if event.Path != "/gateway/events/submit-result" && event.Path != "/gateway/events/submit-segment-result" { + return Event{}, fmt.Errorf("unsupported result Outbox path %q", event.Path) + } + if len(event.Payload) == 0 { + return Event{}, fmt.Errorf("result Outbox payload is required") + } + return event, nil +} + +func (o *Outbox) stream() string { + if strings.TrimSpace(o.Stream) != "" { + return o.Stream + } + return defaultStream +} + +func (o *Outbox) group() string { + if strings.TrimSpace(o.Group) != "" { + return o.Group + } + return defaultGroup +} + +func (o *Outbox) consumer() string { + if strings.TrimSpace(o.Consumer) != "" { + return o.Consumer + } + return defaultConsumer +} + +func (o *Outbox) dedupeTTL() time.Duration { + if o.DedupeTTL > 0 { + return o.DedupeTTL + } + return defaultDedupeTTL +} + +func (o *Outbox) dedupeKey(eventID string) string { + return o.stream() + ":dedupe:" + eventID +} + +func (o *Outbox) StreamName() string { return o.stream() } +func (o *Outbox) GroupName() string { return o.group() } +func (o *Outbox) InFlight() int64 { return o.inFlight.Load() } diff --git a/gateway/internal/resultoutbox/outbox_test.go b/gateway/internal/resultoutbox/outbox_test.go new file mode 100644 index 0000000..eb73567 --- /dev/null +++ b/gateway/internal/resultoutbox/outbox_test.go @@ -0,0 +1,129 @@ +package resultoutbox + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "cmpp-platform/gateway/internal/queue" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +func TestPublishSubmitSegmentIsIdempotent(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + outbox := New(client) + command := testCommand() + segment := queue.SubmitSegmentResult{SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted"} + + if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil { + t.Fatalf("first publish: %v", err) + } + if err := outbox.PublishSubmitSegment(context.Background(), command, segment); err != nil { + t.Fatalf("duplicate publish: %v", err) + } + if got := client.XLen(context.Background(), outbox.StreamName()).Val(); got != 1 { + t.Fatalf("stream length = %d, want 1", got) + } +} + +func TestPublishAggregateAndCommandAckAreAtomicAndIdempotent(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + ctx := context.Background() + commandStream := "gateway.submit.commands" + commandGroup := "cmpp-gateway" + if err := client.XGroupCreateMkStream(ctx, commandStream, commandGroup, "0").Err(); err != nil { + t.Fatalf("create command group: %v", err) + } + commandID, err := client.XAdd(ctx, &redis.XAddArgs{Stream: commandStream, Values: map[string]interface{}{"data": "command"}}).Result() + if err != nil { + t.Fatalf("add command: %v", err) + } + if _, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: commandGroup, Consumer: "gateway-1", Streams: []string{commandStream, ">"}, Count: 1}).Result(); err != nil { + t.Fatalf("claim command: %v", err) + } + outbox := New(client) + command := testCommand() + result := queue.SubmitResult{Envelope: command.Envelope, SubmitID: command.SubmitID, GatewayMessageID: "99", SubmitStatus: "accepted"} + + for attempt := 0; attempt < 2; attempt++ { + if err := outbox.PublishSubmitResultAndAck(ctx, commandStream, commandGroup, commandID, command, result); err != nil { + t.Fatalf("publish attempt %d: %v", attempt+1, err) + } + } + pending, err := client.XPending(ctx, commandStream, commandGroup).Result() + if err != nil { + t.Fatalf("command pending: %v", err) + } + if pending.Count != 0 { + t.Fatalf("command pending = %d, want 0", pending.Count) + } + if got := client.XLen(ctx, outbox.StreamName()).Val(); got != 1 { + t.Fatalf("result stream length = %d, want 1", got) + } +} + +func TestCallbackWorkerRetriesAndOnlyDeletesAfterSuccess(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.Header.Get("X-CMPP-Result-Event-ID") == "" { + t.Error("missing result event id header") + } + if calls.Add(1) == 1 { + http.Error(response, "temporary failure", http.StatusServiceUnavailable) + return + } + response.WriteHeader(http.StatusCreated) + })) + defer server.Close() + + outbox := New(client) + outbox.APIBaseURL = server.URL + outbox.MinIdle = 10 * time.Millisecond + outbox.Concurrency = 1 + command := testCommand() + if err := outbox.PublishSubmitSegment(context.Background(), command, queue.SubmitSegmentResult{ + SegmentTotal: 1, SegmentIndex: 1, SequenceID: 7, GatewayMessageID: "88", SubmitStatus: "accepted", + }); err != nil { + t.Fatalf("publish segment: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- outbox.Run(ctx) }() + deadline := time.Now().Add(5 * time.Second) + for calls.Load() < 2 || client.XLen(context.Background(), outbox.StreamName()).Val() != 0 { + if time.Now().After(deadline) { + t.Fatalf("calls=%d streamLength=%d", calls.Load(), client.XLen(context.Background(), outbox.StreamName()).Val()) + } + time.Sleep(10 * time.Millisecond) + } + cancel() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("worker did not stop") + } +} + +func testCommand() queue.SubmitCommand { + return queue.SubmitCommand{ + Envelope: queue.Envelope{ + SchemaVersion: queue.SchemaVersion, + MessageType: queue.MessageTypeSubmitCommand, + TraceID: "trace-1", + MessageID: "message-1", + ChannelID: "channel-1", + CreatedAt: time.Now().UTC(), + }, + SubmitID: "submit-1", + } +} diff --git a/gateway/internal/resultoutbox/worker.go b/gateway/internal/resultoutbox/worker.go new file mode 100644 index 0000000..d0af11b --- /dev/null +++ b/gateway/internal/resultoutbox/worker.go @@ -0,0 +1,275 @@ +package resultoutbox + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log" + "net/http" + "strings" + "sync" + "time" + + "cmpp-platform/gateway/internal/metrics" + + "github.com/redis/go-redis/v9" +) + +const ( + // Eight callbacks matched the test VM's API/PostgreSQL capacity through + // 40 rps. A larger default caused callback bursts to contend with inbound + // persistence; operators can still raise it after measuring both queues. + defaultConcurrency = 8 + // Keep the reclaim threshold above the callback timeout. Otherwise another + // Gateway replica could reclaim a still-running callback and execute the same + // business transition concurrently before the API stores its idempotency ID. + defaultMinIdle = 30 * time.Second + defaultBlock = 2 * time.Second + defaultHTTPTimeout = 10 * time.Second +) + +var acknowledgeAndDeleteScript = redis.NewScript(` +redis.call('XACK', KEYS[1], ARGV[1], ARGV[2]) +redis.call('XDEL', KEYS[1], ARGV[2]) +return 1 +`) + +func (o *Outbox) Run(ctx context.Context) error { + if o.Redis == nil { + return fmt.Errorf("result Outbox Redis client is required") + } + if strings.TrimSpace(o.APIBaseURL) == "" { + return fmt.Errorf("result Outbox API base URL is required") + } + if err := o.ensureGroup(ctx); err != nil { + return err + } + pool := newCallbackPool(ctx, o, o.concurrency()) + defer pool.wait() + for { + if err := o.recoverPending(ctx, pool); err != nil && ctx.Err() == nil { + log.Printf("gateway result Outbox pending recovery failed: %v", err) + sleep(ctx, time.Second) + continue + } + if ctx.Err() != nil { + return ctx.Err() + } + if err := o.consumeOnce(ctx, pool); err != nil && ctx.Err() == nil { + log.Printf("gateway result Outbox consume failed: %v", err) + sleep(ctx, time.Second) + continue + } + if ctx.Err() != nil { + return ctx.Err() + } + } +} + +func (o *Outbox) ensureGroup(ctx context.Context) error { + err := o.Redis.XGroupCreateMkStream(ctx, o.stream(), o.group(), "0").Err() + if err == nil || strings.Contains(err.Error(), "BUSYGROUP") { + return nil + } + return err +} + +func (o *Outbox) consumeOnce(ctx context.Context, pool *callbackPool) error { + available, err := pool.waitForCapacity(ctx) + if err != nil { + return err + } + streams, err := o.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{ + Group: o.group(), Consumer: o.consumer(), Streams: []string{o.stream(), ">"}, + Count: int64(available), Block: defaultBlock, + }).Result() + if errors.Is(err, redis.Nil) { + return nil + } + if err != nil { + return err + } + for _, stream := range streams { + for _, message := range stream.Messages { + if !pool.dispatch(message) { + return fmt.Errorf("gateway result Outbox capacity accounting mismatch") + } + } + } + return nil +} + +func (o *Outbox) recoverPending(ctx context.Context, pool *callbackPool) error { + if pool.available() == 0 { + return nil + } + messages, _, err := o.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + Stream: o.stream(), Group: o.group(), Consumer: o.consumer(), MinIdle: o.minIdle(), + Start: "0-0", Count: int64(pool.available()), + }).Result() + if errors.Is(err, redis.Nil) { + return nil + } + if err != nil { + return err + } + for _, message := range messages { + if !pool.dispatch(message) { + return fmt.Errorf("gateway result Outbox recovery capacity accounting mismatch") + } + } + return nil +} + +func (o *Outbox) processMessage(ctx context.Context, message redis.XMessage) error { + event, err := EventFromStreamValues(message.Values) + if err != nil { + // Malformed internal events cannot be delivered. Keep them pending for operator + // evidence instead of ACKing and silently losing a supplier result. + return err + } + startedAt := time.Now() + err = o.post(ctx, event) + metrics.ObserveSubmitStage("api_callback", err == nil, time.Since(startedAt)) + if err != nil { + return err + } + // Result events have a bounded dedupe key, so successful callbacks can be + // ACKed and deleted atomically instead of turning the Outbox into an archive. + return acknowledgeAndDeleteScript.Run( + ctx, + o.Redis, + []string{o.stream()}, + o.group(), + message.ID, + ).Err() +} + +func (o *Outbox) post(ctx context.Context, event Event) error { + client := &http.Client{Timeout: o.httpTimeout()} + req, err := http.NewRequestWithContext( + ctx, + http.MethodPost, + strings.TrimRight(o.APIBaseURL, "/")+event.Path, + bytes.NewReader(event.Payload), + ) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-CMPP-Result-Event-ID", event.EventID) + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("result callback returned %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + return nil +} + +type callbackPool struct { + ctx context.Context + outbox *Outbox + slots chan struct{} + completed chan struct{} + group sync.WaitGroup + mu sync.Mutex + active map[string]struct{} +} + +func newCallbackPool(ctx context.Context, outbox *Outbox, concurrency int) *callbackPool { + return &callbackPool{ + ctx: ctx, outbox: outbox, slots: make(chan struct{}, concurrency), completed: make(chan struct{}, concurrency), active: make(map[string]struct{}), + } +} + +func (p *callbackPool) available() int { return cap(p.slots) - len(p.slots) } + +func (p *callbackPool) waitForCapacity(ctx context.Context) (int, error) { + for p.available() == 0 { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-p.completed: + } + } + return p.available(), nil +} + +func (p *callbackPool) dispatch(message redis.XMessage) bool { + p.mu.Lock() + if _, exists := p.active[message.ID]; exists { + p.mu.Unlock() + return true + } + select { + case p.slots <- struct{}{}: + p.active[message.ID] = struct{}{} + p.outbox.inFlight.Add(1) + p.group.Add(1) + p.mu.Unlock() + case <-p.ctx.Done(): + p.mu.Unlock() + return false + default: + p.mu.Unlock() + return false + } + go func() { + defer func() { + p.mu.Lock() + delete(p.active, message.ID) + p.mu.Unlock() + <-p.slots + p.outbox.inFlight.Add(-1) + select { + case p.completed <- struct{}{}: + default: + } + p.group.Done() + }() + if err := p.outbox.processMessage(p.ctx, message); err != nil { + log.Printf("gateway result Outbox event %s failed: %v", message.ID, err) + } + }() + return true +} + +func (p *callbackPool) wait() { p.group.Wait() } + +func (o *Outbox) concurrency() int { + if o.Concurrency > 0 { + return min(o.Concurrency, 1024) + } + return defaultConcurrency +} + +func (o *Outbox) minIdle() time.Duration { + if o.MinIdle > 0 { + return o.MinIdle + } + return defaultMinIdle +} + +func (o *Outbox) httpTimeout() time.Duration { + if o.HTTPTimeout > 0 { + return o.HTTPTimeout + } + return defaultHTTPTimeout +} + +func (o *Outbox) ConfiguredConcurrency() int { return o.concurrency() } + +func sleep(ctx context.Context, duration time.Duration) { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + } +} diff --git a/gateway/internal/submitworker/worker.go b/gateway/internal/submitworker/worker.go index f06a1d5..00dffc0 100644 --- a/gateway/internal/submitworker/worker.go +++ b/gateway/internal/submitworker/worker.go @@ -36,6 +36,7 @@ type Worker struct { Limiter ratelimit.Limiter Submit func(context.Context, queue.SubmitCommand) (queue.SubmitResult, error) ReportDeadLetter func(context.Context, DeadLetterEvent) error + ResultOutbox SubmitResultOutbox Stream string Group string Consumer string @@ -50,6 +51,17 @@ type Worker struct { inFlight atomic.Int64 } +type SubmitResultOutbox interface { + PublishSubmitResultAndAck( + context.Context, + string, + string, + string, + queue.SubmitCommand, + queue.SubmitResult, + ) error +} + type DeadLetterEvent struct { StreamMessageID string `json:"streamMessageId"` TraceID string `json:"traceId,omitempty"` @@ -82,6 +94,9 @@ func (w *Worker) Run(ctx context.Context) error { if w.Upstream == nil { return fmt.Errorf("upstream manager is required") } + if w.ResultOutbox == nil { + return fmt.Errorf("submit result Outbox is required") + } pool := newMessageWorkPool(ctx, w, w.concurrency()) defer pool.wait() for { @@ -292,7 +307,8 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err if !command.CreatedAt.IsZero() { metrics.ObserveSubmitStage("stream_wait", true, time.Since(command.CreatedAt)) } - if err := w.handleCommand(ctx, command); err != nil { + result, err := w.executeCommand(ctx, command) + if err != nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") { if ctx.Err() != nil { return ctx.Err() } @@ -308,29 +324,50 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err } return err } - return w.ackAndClearFailure(ctx, message.ID) + if err := w.ResultOutbox.PublishSubmitResultAndAck( + ctx, + w.stream(), + w.group(), + message.ID, + command, + result, + ); err != nil { + return fmt.Errorf("persist aggregate submit result: %w", err) + } + // The command was ACKed atomically with the aggregate result event. Clearing the + // auxiliary retry counter may lag without affecting delivery correctness. + w.clearFailure(ctx, message.ID) + return nil } -func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error { +func (w *Worker) executeCommand(ctx context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) { startedAt := time.Now() limitStartedAt := time.Now() if w.Limiter != nil { if _, err := w.Limiter.Wait(ctx, command.ChannelID, command.Route.RateLimitPerSecond); err != nil { metrics.ObserveSubmitStage("rate_limit_wait", false, time.Since(limitStartedAt)) - return err + return queue.SubmitResult{}, err } } metrics.ObserveSubmitStage("rate_limit_wait", true, time.Since(limitStartedAt)) submit := w.Submit if submit == nil { if w.Upstream == nil { - return fmt.Errorf("upstream manager is required") + return queue.SubmitResult{}, fmt.Errorf("upstream manager is required") } submit = w.Upstream.Submit } result, err := submit(ctx, command) accepted := err == nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") metrics.ObserveSubmit(accepted, time.Since(startedAt)) + return result, err +} + +// handleCommand remains a narrow compatibility seam for focused tests and the +// control path. Stream consumption uses executeCommand so it can persist the +// exact terminal result before acknowledging the command. +func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error { + result, err := w.executeCommand(ctx, command) if err != nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") { return err } @@ -456,10 +493,14 @@ func (w *Worker) ackAndClearFailure(ctx context.Context, messageID string) error if err := w.Redis.XAck(ctx, w.stream(), w.group(), messageID).Err(); err != nil { return err } + w.clearFailure(ctx, messageID) + return nil +} + +func (w *Worker) clearFailure(ctx context.Context, messageID string) { if err := w.Redis.HDel(ctx, w.failureAttemptsKey(), messageID).Err(); err != nil { w.logf("gateway submit worker clear failure %s failed: %v", messageID, err) } - return nil } func (w *Worker) stream() string { diff --git a/gateway/internal/submitworker/worker_test.go b/gateway/internal/submitworker/worker_test.go index 44b6fb6..311dc38 100644 --- a/gateway/internal/submitworker/worker_test.go +++ b/gateway/internal/submitworker/worker_test.go @@ -17,6 +17,24 @@ type recordingLimiter struct { called bool } +type acknowledgingResultOutbox struct { + client *redis.Client +} + +func (o acknowledgingResultOutbox) PublishSubmitResultAndAck( + ctx context.Context, + commandStream string, + commandGroup string, + commandMessageID string, + _ queue.SubmitCommand, + _ queue.SubmitResult, +) error { + if o.client == nil { + return nil + } + return o.client.XAck(ctx, commandStream, commandGroup, commandMessageID).Err() +} + func (l *recordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) { l.called = true l.channelID = channelID @@ -139,7 +157,8 @@ func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testi startedC := make(chan struct{}) releaseA := make(chan struct{}) worker := &Worker{ - Redis: client, + Redis: client, + ResultOutbox: acknowledgingResultOutbox{client: client}, Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) { switch command.ChannelID { case "channel-a": @@ -188,7 +207,7 @@ func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testi func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *testing.T) { mr := miniredis.RunT(t) client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) - worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway"} + worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", ResultOutbox: acknowledgingResultOutbox{client: client}} ctx := context.Background() if err := worker.ensureGroup(ctx); err != nil { t.Fatalf("ensureGroup: %v", err) @@ -246,7 +265,7 @@ func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *tes func TestPendingRecoveryDoesNotDuplicateAnActiveOrAlreadyAcknowledgedMessage(t *testing.T) { mr := miniredis.RunT(t) client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) - worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond} + worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond, ResultOutbox: acknowledgingResultOutbox{client: client}} ctx := context.Background() if err := worker.ensureGroup(ctx); err != nil { t.Fatalf("ensureGroup: %v", err) diff --git a/gateway/internal/upstream/manager.go b/gateway/internal/upstream/manager.go index 84e78ee..a2ecb91 100644 --- a/gateway/internal/upstream/manager.go +++ b/gateway/internal/upstream/manager.go @@ -22,13 +22,21 @@ const ( ) type Manager struct { - APIBaseURL string - HTTPClient *http.Client + APIBaseURL string + HTTPClient *http.Client + SubmitSegmentPublisher SubmitSegmentPublisher mu sync.Mutex conns map[string]*connectionPool } +// SubmitSegmentPublisher persists each supplier response before the next long-message +// segment is sent. The boundary is intentionally storage-only: HTTP callbacks belong to +// the result Outbox worker and must not consume a supplier Submit window slot. +type SubmitSegmentPublisher interface { + PublishSubmitSegment(context.Context, queue.SubmitCommand, queue.SubmitSegmentResult) error +} + type ConnectionState struct { ChannelID string `json:"channelId"` ConnectionID string `json:"connectionId"` diff --git a/gateway/internal/upstream/submit.go b/gateway/internal/upstream/submit.go index 94158a1..cb8f35b 100644 --- a/gateway/internal/upstream/submit.go +++ b/gateway/internal/upstream/submit.go @@ -7,7 +7,6 @@ import ( "fmt" cmpp "github.com/bigwhite/gocmpp" cmpputils "github.com/bigwhite/gocmpp/utils" - "log" "strings" "time" ) @@ -17,58 +16,30 @@ import ( func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) { if err := validateSubmitCommand(cmd); err != nil { - result := submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error()) - if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil { - return result, postErr - } - return result, err + return submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error()), err } pool, err := m.connectionFor(cmd) if err != nil { - result := submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error()) - if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil { - return result, postErr - } - return result, err + return submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error()), err } - result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) { - payload := struct { - queue.Envelope - SubmitID string `json:"submitId,omitempty"` - queue.SubmitSegmentResult - }{ - Envelope: cmd.Envelope, - SubmitID: cmd.SubmitID, - SubmitSegmentResult: segment, - } - callbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - postErr := m.postSubmitCallback(callbackCtx, "/gateway/events/submit-segment-result", payload) - cancel() - if postErr != nil { - log.Printf( - "protocol_event protocol=cmpp direction=gateway_to_api event=submit_segment_result status=forward_failed channel_id=%s message_id=%s segment=%d/%d error=%q", - cmd.ChannelID, cmd.MessageID, segment.SegmentIndex, segment.SegmentTotal, postErr, - ) + result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) error { + if m.SubmitSegmentPublisher == nil { + return fmt.Errorf("submit segment result publisher is required") } + // A segment result must reach durable local storage before the next segment. + // This cannot make the supplier/Redis boundary globally atomic, but it avoids + // holding the supplier slot for an API round trip and minimizes untracked sends. + return m.SubmitSegmentPublisher.PublishSubmitSegment(ctx, cmd, segment) }) - if err != nil { - if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil { - return result, postErr - } - return result, err - } - if err := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); err != nil { - return result, err - } - return result, nil + return result, err } func (p *connectionPool) submit( ctx context.Context, cmd queue.SubmitCommand, - onSegment func(queue.SubmitSegmentResult), + onSegment func(queue.SubmitSegmentResult) error, ) (queue.SubmitResult, error) { parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content) if err != nil { @@ -95,7 +66,10 @@ func (p *connectionPool) submit( segment := submitSegmentResult(part, seq, gatewayMessageID, result) segments = append(segments, segment) if onSegment != nil { - onSegment(segment) + if publishErr := onSegment(segment); publishErr != nil { + result.Segments = segments + return result, publishErr + } } if firstSequence == 0 { firstSequence = seq @@ -118,13 +92,6 @@ func (p *connectionPool) submit( return result, nil } -func (m *Manager) postSubmitCallback(ctx context.Context, path string, payload any) error { - startedAt := time.Now() - err := m.post(ctx, path, payload) - metrics.ObserveSubmitStage("api_callback", err == nil, time.Since(startedAt)) - return err -} - func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) { rspCh := make(chan submitPartResponse, 1) pkt := c.submitRequestPacket(cmd, part) diff --git a/tools/spike/validate-gateway-queue-contract.mjs b/tools/spike/validate-gateway-queue-contract.mjs index 4c338a5..e91319e 100644 --- a/tools/spike/validate-gateway-queue-contract.mjs +++ b/tools/spike/validate-gateway-queue-contract.mjs @@ -40,6 +40,17 @@ function assertIsoDateTime(value, field, file) { function validateExample(fileName) { const fullPath = join(examplesDir, fileName); const payload = JSON.parse(readFileSync(fullPath, 'utf8')); + if (fileName === 'submit-result-outbox-event.json') { + for (const field of ['schemaVersion', 'eventId', 'eventType', 'path', 'messageId', 'channelId', 'submitId', 'payload', 'createdAt']) { + assert(Object.hasOwn(payload, field), `${fileName}: missing ${field}`); + } + assert(payload.schemaVersion === 'v1', `${fileName}: schemaVersion must be v1`); + assert(['submit_result', 'submit_segment_result'].includes(payload.eventType), `${fileName}: unsupported eventType`); + assert(['/gateway/events/submit-result', '/gateway/events/submit-segment-result'].includes(payload.path), `${fileName}: unsupported path`); + assert(payload.payload?.eventId === payload.eventId, `${fileName}: payload.eventId must match eventId`); + assertIsoDateTime(payload.createdAt, 'createdAt', fileName); + return `${fileName}: ${payload.eventType} ok`; + } const type = payload.messageType; assert(validTypes.has(type), `${fileName}: unsupported messageType ${type}`);