From 0757a699ff7138e5a4b0d6075a79cd2cae0ab32b Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Thu, 20 Aug 2026 11:45:16 +0800 Subject: [PATCH] perf(cmpp): process inbound submits within client window --- .env.example | 1 + api/src/send-chain/send-chain.service.spec.ts | 2 + .../send-chain/send-inbound-entry.service.ts | 1 + docs/codebase-modularization-roadmap.md | 1 + docs/contracts/inbound-r6-declarations.json | 72 ++++++++++-- .../first-version-development-requirements.md | 3 +- docs/production-deployment.md | 5 +- docs/system-functional-test-cases.md | 6 + docs/testing-progress.md | 8 ++ gateway/README.md | 1 + gateway/cmd/gateway/main.go | 14 ++- gateway/internal/inbound/authentication.go | 4 + gateway/internal/inbound/server.go | 13 +- gateway/internal/inbound/server_test.go | 111 ++++++++++++++++++ gateway/internal/inbound/sessions.go | 36 ++++++ gateway/internal/inbound/submit.go | 2 + gateway/internal/metrics/metrics.go | 23 ++-- gateway/internal/metrics/metrics_test.go | 4 +- gateway/third_party/gocmpp/server.go | 107 ++++++++++++++--- tools/quality/verify-inbound-r6.go | 22 ++++ 20 files changed, 389 insertions(+), 47 deletions(-) diff --git a/.env.example b/.env.example index 969a36b..d922d97 100644 --- a/.env.example +++ b/.env.example @@ -43,4 +43,5 @@ GATEWAY_CMPP_ADDR=127.0.0.1:7890 GATEWAY_CMPP_USER=900001 GATEWAY_CMPP_PASSWORD=888888 GATEWAY_SUBMIT_WORKER_CONCURRENCY=64 +GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY=64 CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS=30 diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index a0a40a3..5cf1cb2 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -81,6 +81,7 @@ function createPrismaMock() { status: 'active', interfaceEnabled: true, cmppMaxConnections: 2, + cmppWindowSize: 32, queuePriority: 'normal', ipAllowlist: [{ ipCidr: '127.0.0.1/32' }], tenant: { id: 'tenant-1', status: 'active', certificationStatus: 'approved' }, @@ -1057,6 +1058,7 @@ describe('SendChainService', () => { account: '100001', enterpriseCode: 'SP0001', maxConnections: 2, + windowSize: 32, status: 'authenticated', })); expect(prisma.operationLog.create).toHaveBeenCalledWith({ diff --git a/api/src/send-chain/send-inbound-entry.service.ts b/api/src/send-chain/send-inbound-entry.service.ts index 76036f5..a8c8a33 100644 --- a/api/src/send-chain/send-inbound-entry.service.ts +++ b/api/src/send-chain/send-inbound-entry.service.ts @@ -102,6 +102,7 @@ async authenticateInboundApplication(data: GatewayInboundAuthDto) { enterpriseCode: application.cmppEnterpriseCode, passwordCipher: application.secretHash, maxConnections: application.cmppMaxConnections, + windowSize: application.cmppWindowSize, status: 'authenticated', }; } catch (error) { diff --git a/docs/codebase-modularization-roadmap.md b/docs/codebase-modularization-roadmap.md index e9590ce..ae2082f 100644 --- a/docs/codebase-modularization-roadmap.md +++ b/docs/codebase-modularization-roadmap.md @@ -1152,5 +1152,6 @@ global，不能错误归入client。`client-signature-*`、发送页、企业认 - `gateway/internal/metrics/`只负责线程安全聚合和Prometheus文本暴露；上游连接池、下游Session和Submit Worker仅提供总数或有界结果，不把实体ID或凭据交给监控模块。 - 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`供应商窗口合并成同一并发计数。 - `api/src/infrastructure-monitoring/`是运营端监控聚合与固定阈值应用边界：只消费代码白名单 PromQL，并通过版本化 PostgreSQL 单例、promtool 校验和原子规则热加载管理数值阈值；Exporter安装、端口隔离、固定规则模板和权限仍归`tools/monitoring/`治理。 - 活动告警已读也归该边界：Prometheus保留告警事实，Prisma仅持久化逐管理员、逐触发周期的阅读状态；全局布局只消费轻量未读汇总，不复制指纹、activeAt或用户隔离逻辑。 diff --git a/docs/contracts/inbound-r6-declarations.json b/docs/contracts/inbound-r6-declarations.json index 19e68b5..6903b8d 100644 --- a/docs/contracts/inbound-r6-declarations.json +++ b/docs/contracts/inbound-r6-declarations.json @@ -78,19 +78,19 @@ "name": "authRequest", "kind": "type", "file": "authentication.go", - "sha256": "742c3df9a2743741b679a5e1524a8826168385e2af5442fc9019ffc8139ea159" + "sha256": "742c3df9a2743741b679a5e1524a8826168385e2af5442fc9019ffc8139ea159" }, { "name": "authResponse", "kind": "type", "file": "authentication.go", - "sha256": "387ff32b3067a0eb342d09e1e5d3308f8359415171a2045f10dc56ca01205727" + "sha256": "049a0b49b97fca25e6093948732f33c9896020c8f3ef8ca4303b05796047ff2b" }, { "name": "authenticate", "kind": "func", "file": "authentication.go", - "sha256": "fbd2a9aab7116a66f4a9cdc4538d1ccd71f822dda37f3f92c08d62417c1f4418" + "sha256": "fbd2a9aab7116a66f4a9cdc4538d1ccd71f822dda37f3f92c08d62417c1f4418" }, { "name": "cmppVersionName", @@ -102,7 +102,13 @@ "name": "handleLogin", "kind": "func", "file": "authentication.go", - "sha256": "b40ab9a3d1230f71af983a0b1a7c510f56f9e1e1d09817f8705b55c6412ee11f" + "sha256": "4d38524f2e4efc3bd19d7e319268b611c306341e18cb8b92f693922c1a21eee0" + }, + { + "name": "reportSecurityEvent", + "kind": "func", + "file": "authentication.go", + "sha256": "8d20e1b5b80f1a8a7c8948bc5dcab283f4c14d9d516008afdfc061b3f74cbaeb" }, { "name": "setInboundConnectResponse", @@ -182,6 +188,12 @@ "file": "delivery.go", "sha256": "4acf5485100e1bc6bb19a8004a3b1b7796fd8a6192fe2899e163abed7d134b69" }, + { + "name": "downstreamReceiptMessageID", + "kind": "func", + "file": "delivery.go", + "sha256": "8b9d214452acdaebfb193c79bd810219fc2400a4c10784f1c3e3322e70a7ad9c" + }, { "name": "errorMessageWithCode", "kind": "func", @@ -342,13 +354,19 @@ "name": "ListenAndServe", "kind": "func", "file": "server.go", - "sha256": "c57e5d860f6bdbb7d47a5414e0d186fe34cf61c42f94fc2b1ab5739b6272c272" + "sha256": "8415f1a682d28c5a64f3fbe26972b1cab07cd04c43c74d7e9da1065255a06e5d" }, { "name": "Server", "kind": "type", "file": "server.go", - "sha256": "b43c6128e784432a544631754f449c49fa1ad3e2cd456ccf795cb5fef69fceaf" + "sha256": "aa919bb151d551a47e1c3fc1191fdd90ee9757f58ef10a718000ff4e8961e4d3" + }, + { + "name": "boundedSubmitWindow", + "kind": "func", + "file": "server.go", + "sha256": "b5ef665e15e9c4b6bad345c94cdc19c28cce4037fb9f484a9d8ca7319c8418c3" }, { "name": "defaultHTTPTimeout", @@ -356,12 +374,30 @@ "file": "server.go", "sha256": "247aaf1070a6886f01059677dad739e69cb9dd10bd55d6397ba5fb9d385fe39a" }, + { + "name": "ActiveConnectionCount", + "kind": "func", + "file": "sessions.go", + "sha256": "3df8a36ca047e9b0d2fb6e2cc4f54b5623b54edacb28397c7e49d2452ee40450" + }, { "name": "DisconnectAccount", "kind": "func", "file": "sessions.go", "sha256": "306c822ca618b5d47dd002595e0896b87d86c26a0b35ef7ea6b032df8cefe66b" }, + { + "name": "SubmitSlotSnapshot", + "kind": "func", + "file": "sessions.go", + "sha256": "4d80321cc1634c4c15880f335144c4c4927cb94df0b4390da1499aa649feb317" + }, + { + "name": "beginInboundSubmit", + "kind": "func", + "file": "sessions.go", + "sha256": "9a25ff062c976af7bddd07b2ace5b0e0de3a562a5e26f8df03fb0030a7ac470b" + }, { "name": "downstreamConnectionEvent", "kind": "type", @@ -378,7 +414,7 @@ "name": "downstreamSession", "kind": "type", "file": "sessions.go", - "sha256": "36d85eab72c673e4e4d47a2b60e68d96561ecb93b0a851de850280706e6779bc" + "sha256": "3f4383f1c5406906df5065f32dc38f3a1be84f6fd678a2533101ab35d90f5d68" }, { "name": "findSessionByConn", @@ -452,6 +488,12 @@ "file": "sessions.go", "sha256": "61cf0ec3ac4278be5ccbd112d196aec662ae20f6aebd6079ac7102e83b1e7008" }, + { + "name": "submitWindowByConn", + "kind": "func", + "file": "sessions.go", + "sha256": "d17509dc36a19c5d1910bb8588ef875fd4f6225675643f994886d0f48aa15f08" + }, { "name": "touchPresence", "kind": "func", @@ -474,7 +516,7 @@ "name": "handleSubmit", "kind": "func", "file": "submit.go", - "sha256": "76550096e4047d8bcfbb14f693294e86c7a1bf0be2db953f8afebaecd661c577" + "sha256": "9d6590964ba080a55758ee7fee0b3acefc398ce81d3d13fd7eac4660b2d90e68" }, { "name": "inboundLongMessageFragment", @@ -500,6 +542,12 @@ "file": "submit.go", "sha256": "977d81193fdc2adfd402d6b1ddb142ea2fdf44f7c8f8d2d00ca8cf20d2cf0830" }, + { + "name": "observeInboundSubmitResponse", + "kind": "func", + "file": "submit.go", + "sha256": "645537f6f6e6eb69296f756fb3cb291ccecf2b82d48ce0bea427bb898dff6188" + }, { "name": "setInboundSubmitResponse", "kind": "func", @@ -548,11 +596,17 @@ "file": "transport.go", "sha256": "fd069087ad8497197f55a98ecfb9e5b1234ce641a0fb0a9161f8c320f436dd37" }, + { + "name": "maxAPIResponseBodyBytes", + "kind": "const", + "file": "transport.go", + "sha256": "3b238a26101aa6a3b50d6651479502adc2be2fb98b338c933cbfd4c90510995f" + }, { "name": "post", "kind": "func", "file": "transport.go", - "sha256": "7982e33b65432936ef97e85357e662a1547c4a93a9852f145127b74ab50609c3" + "sha256": "7982e33b65432936ef97e85357e662a1547c4a93a9852f145127b74ab50609c3" }, { "name": "remoteIP", diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 9136f0e..374a7b7 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -124,7 +124,7 @@ 9. 短信应用必须有应用级客户侧企业代码 `cmppEnterpriseCode`，运营端添加/编辑应用时可自定义；不得从上游通道 `SmsChannel.enterpriseCode` 透传。 10. 短信应用接口密码 `passwordCipher` 新建时默认随机生成 16 位 UUID 片段，运营端可手工修改；编辑时留空不覆盖原密码。 11. 应用 `AppID` 是平台内部应用标识，用于页面展示、复制参数和工单定位，不作为 CMPP bind/login 认证参数。 -12. 短信应用必须可配置客户侧 CMPP 最大连接数 `cmppMaxConnections`；客户侧提交窗口 `cmppWindowSize` 后端保留默认值，当前第一版不在运营端展示或要求运营配置，待 Gateway 入站侧按应用窗口真正限流后再开放为高级配置。 +12. 短信应用必须可配置客户侧 CMPP 最大连接数 `cmppMaxConnections`；客户侧提交窗口 `cmppWindowSize` 由真实后端保存并在认证成功时下发 Gateway。V3 起 Gateway 对同一已认证连接最多并发处理该窗口数量的 Submit，超过窗口的报文通过停止继续读取形成 TCP 背压，不得无界创建协程；Gateway 全局保护上限由 `GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY` 控制，默认 64、最大 1024，应用窗口只能进一步收紧而不能放大全局上限。 13. 短信应用必须恢复设计基线中的“短信接口”开关，字段为 `interfaceEnabled`，默认开通；关闭后客户端/API 发送链路、客户侧 CMPP Gateway bind/login 和 submit 都必须被真实后端拒绝，不允许只在前端隐藏入口。 14. CMPP 协议类型当前第一版仅允许 `CMPP2.0`，字段保持 `interfaceType=cmpp20`；HTTP 不写入该字段，而是通过独立的 `SmsApplicationHttpConfig` 总开关和子能力配置开通。前后端仍必须拒绝把 `interfaceType` 直接改成 `http` 等无效协议值。 @@ -2099,6 +2099,7 @@ - CMPP入站性能优化第一步只增加观测，不改变SubmitResp、持久化、风控、计费、路由、入队或回执语义。API必须用固定低基数阶段记录`application_lookup/long_message_fragment/submission_precheck/template_match/task_persist/api_request_persist/content_detection/message_persist/risk_frequency/billing/queue_publish/complete_submit/total`耗时及成功/失败；Gateway入站必须拆分`decode/api_roundtrip/response_write/handler_total`。 - 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}`只暴露全部在线连接的聚合窗口与在途数量，不得增加账号、应用、连接或消息标签。 - 系统监控标题说明必须明确标注数据来自 Prometheus；“服务关键指标”位于趋势/核心服务区域之后、活动告警之前，并提供统一的“告警阈值设置”入口。安全检测页不重复渲染大号标题，说明文字必须明确标注使用 Fail2ban。 - 告警阈值仅开放固定指标的警告/严重数值，不允许前端提交 PromQL、标签、文件路径或持续时间；必须满足警告值小于严重值。配置以 PostgreSQL 保存版本、期望值、生效值和应用状态，经 `promtool check rules` 校验、同目录原子替换和 Prometheus 热加载成功后才标记生效，失败保留上一生效规则并展示原因。 - 右上角预警中心增加“系统监控告警”，通过独立轻量接口统计 Prometheus 当前 firing/pending 告警及严重数，跳转系统监控活动告警区；任一预警域失败不得清空其他域。 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index a0d92f8..5dfde1f 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -61,6 +61,7 @@ CMPP_PUBLIC_HOST=8.160.169.106 CMPP_PUBLIC_PORT=17890 GATEWAY_STARTUP_RECONNECT_DELAY_MS=1000 GATEWAY_SUBMIT_WORKER_CONCURRENCY=64 +GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY=64 OBJECT_STORAGE_DRIVER=minio OBJECT_STORAGE_LOCAL_ROOT=/var/lib/cmpp-platform/object-storage PROD_ADMIN_EMAIL=admin@example.com @@ -82,6 +83,8 @@ HTTP容量边界固定为：NestJS普通JSON/URL-encoded请求体`2 MiB`，仅`/ Gateway 的最终 TPS 防线依赖与 API 相同的 Redis。通道连接时会写入 `rate:gateway:channel:config:` 权威上限，实际预约使用 `rate:gateway:channel:`；这些 key 不应在正常发布时清理。多 Gateway 实例必须指向同一 Redis，才能共享单通道额度。超速的 `gateway.submit.commands` 消息会保持在 consumer group pending 中等待，不应通过手工 `XACK` 或删除 Stream 处理积压；先检查通道配置、Redis key、consumer group 和 Gateway 日志。V2起Submit Worker使用持续补位有界池并逐条ACK，`GATEWAY_SUBMIT_WORKER_CONCURRENCY`缺省64、最大1024；调整前必须同时核对供应商连接数、窗口、TPS限制、Gateway RSS和`cmpp_gateway_submit_worker_slots`，不能用放大并发绕过通道限速。 +V3起客户CMPP入站Submit按应用`cmppWindowSize`在单连接内受限并发，全局上限`GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY`缺省64、最大1024。发布后必须先以已认证测试连接确认`cmpp_gateway_inbound_submit_slots{state="configured"}`等于应用有效窗口，再执行阶梯压测；不得通过调高全局值绕过应用窗口，也不得在未验证Sequence_Id关联、心跳/ACK活性和断线清理时直接提高生产并发。 + 服务重启顺序必须是 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 小时回执更新反映到对账和利润报表。 @@ -142,7 +145,7 @@ 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)=' /etc/cmpp-platform/cmpp-platform.env +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 redis-cli --scan --pattern 'rate:gateway:channel:*' redis-cli XINFO GROUPS gateway.submit.commands ``` diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index bff0cd1..c083a1f 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4676,6 +4676,12 @@ npm run verify:phase8 | TC-CMPP-PERF-V2-005 | Worker槽位指标 | 工作池空闲、部分占用和满载时抓取Gateway metrics | `cmpp_gateway_submit_worker_slots`的`configured/in_flight`与真实配置和在途数一致，不包含通道、消息或客户标识 | | TC-CMPP-PERF-V2-006 | 失败、死信和重启兼容 | 构造提交失败至最大次数、畸形命令、进程重启后的pending恢复 | 失败次数、死信上报、独立ACK和failure hash清理保持既有语义；重启不丢消息、不把未完成消息误报成功 | | TC-CMPP-PERF-V2-007 | 隔离环境阶梯持续压测 | 在供应商模拟器、真实API/PostgreSQL/Redis/Gateway链路中，先做100条受控突发，再依次执行10、20、30、40、50条/秒各60秒；每档等待Stream排空并核对数据库、模拟器和Prometheus | 每档SubmitResp拒绝和连接错误为0，Stream最终`pending=0/lag=0`，无死信或服务异常；任一档SubmitResp P95超过5秒、Stream持续增长或服务异常时立即停止升档并保留该档证据，不把未执行档位记为通过 | +| TC-CMPP-PERF-V3-001 | 同连接窗口内并发 | 应用窗口设为2，让第1个Submit的API处理阻塞，再发送第2个Submit | 第2个请求无需等待第1个完成即可进入API并按自身Sequence_Id先返回SubmitResp；释放第1个后仍返回其原Sequence_Id | +| TC-CMPP-PERF-V3-002 | 应用窗口与全局上限 | 分别设置应用窗口1、32、2048及Gateway全局上限16、64 | 窗口1保持串行；有效并发为`min(应用窗口, Gateway上限, 1024)`，超过窗口时停止继续读取形成背压，不产生无界协程 | +| TC-CMPP-PERF-V3-003 | 非Submit协议活性 | 在多个慢Submit在途时发送ActiveTest并接收Deliver ACK | 心跳和ACK仍可被读取和处理，不因业务Submit串行处理而超时；登录必须在任何并发Submit前串行完成 | +| 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-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 ac70659..1e2cf90 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -3723,3 +3723,11 @@ git diff --check - 数据库按实际首条`queuedAt=2026-08-20 03:01:47.980`对账，恰好新增1898条：最终delivered 1859、failed 11、submitted 28；28条submitted与模拟器配置的28次不回执一致。供应商模拟器同期收到2048次Submit尝试，其中2013次接受、35次拒绝，发送并收到ACK的回执均为1985次，错误0；Gateway重试解释了尝试数高于业务消息数。客户端进程在收集窗口内看到的receipt数量包含异步到达，不能直接替代数据库和模拟器最终对账。 - 本轮所有1898条记录仍识别为`mobile`，未覆盖联通、电信，优先级在20条/秒下也未表现出隔离优势：priority P95约5991ms，normal P95约4791ms。因此“修复号段/运营商识别后验证移动、联通、电信六通道容量及优先级隔离”继续保持P1未完成。 - 本轮新增测试辅助脚本`lg-cmpp-stress-lab/scripts/run-v2-stage.ps1`，只根据档位和持续时间生成一次性客户端配置并调用既有真实CMPP压测客户端，不引入mock、静态结果或localStorage。完整复测报告及原始结果保存在短信平台测试项目；代码仍未提交、未推送，预生产保持原状。 + +# 2026-08-20 CMPP压测优化V3：受窗口约束的连接内并发（开发中） + +- V0/V2已按用户授权提交为`b9a71fe`，提交范围为入站/供应商阶段指标、持续有界Submit Worker、契约和同步文档；`*.tsbuildinfo`、`outputs/`、自动产生的`pnpm-lock.yaml`及空文件`=`继续排除并保护，未推送。提交前Gateway全量测试、`go vet`、API TypeScript正式编译和`git diff --check`通过。 +- V3保持API风控、计费、路由、持久化和SubmitResp业务结果语义不变。认证接口新增返回真实`cmppWindowSize`；Gateway对同一认证连接按`min(应用窗口, GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY, 1024)`并发处理Submit，窗口满时停止继续读取形成TCP背压。登录、心跳和Deliver ACK不纳入Submit槽位，SubmitResp继续以原Sequence_Id关联并允许按完成顺序返回。 +- 项目内gocmpp服务循环只对CMPP2/3 Submit启用并发；连接退出前等待已接受的在途处理完成后才调用会话清理，代码注释解释了这是为了防止迟到API处理重新注册已关闭连接。新增聚合指标`cmpp_gateway_inbound_submit_slots{state=configured|in_flight}`，不带账号、应用、连接或消息标签。 +- 专项集成测试已证明窗口2时，第1个Submit被API阻塞后，第2个Submit可进入API并以自身Sequence_Id先返回；释放后第1个响应仍按原Sequence_Id返回。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`通过。Windows本机`go test -race`因Go工具链未启用CGO而无法运行，未伪报通过；将在Linux测试环境部署前补跑。测试环境恢复资产、部署和阶梯压测待后续补记。 +- 当前只修改本地代码，没有连接或改变预生产运行版本；预生产只读标识仍为`433b2ee56f6016ad8afff1bac73f510b8fd53083+gateway-v2.cd7bb8d05e7b`。V3只允许发布到`100.93.204.60`虚拟机测试环境。 diff --git a/gateway/README.md b/gateway/README.md index 2406fa6..208ffe7 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -12,6 +12,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关联，允许按完成顺序返回。 ## 建议骨架 diff --git a/gateway/cmd/gateway/main.go b/gateway/cmd/gateway/main.go index b5ca069..7459f23 100644 --- a/gateway/cmd/gateway/main.go +++ b/gateway/cmd/gateway/main.go @@ -48,12 +48,13 @@ func main() { go func() { log.Printf("cmpp gateway inbound server listening on %s", cmppAddr) if err := (inbound.Server{ - Addr: cmppAddr, - APIBaseURL: apiBaseURL, - PresenceStore: presenceStore, - RecoveryStore: recoveryStore, - GatewayInstanceID: getenv("GATEWAY_INSTANCE_ID", hostname()), - SecurityEventToken: os.Getenv("SECURITY_EVENT_TOKEN"), + Addr: cmppAddr, + APIBaseURL: apiBaseURL, + PresenceStore: presenceStore, + RecoveryStore: recoveryStore, + GatewayInstanceID: getenv("GATEWAY_INSTANCE_ID", hostname()), + MaxSubmitConcurrency: positiveEnvInt("GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY", 64), + SecurityEventToken: os.Getenv("SECURITY_EVENT_TOKEN"), }).ListenAndServe(); err != nil { log.Fatalf("gateway inbound server stopped: %v", err) } @@ -90,6 +91,7 @@ func main() { snapshot.SubmitWorkerConcurrency = worker.ConfiguredConcurrency() snapshot.SubmitWorkerInFlight = worker.InFlight() } + snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight = inbound.SubmitSlotSnapshot() if worker == nil || worker.Redis == nil { return snapshot } diff --git a/gateway/internal/inbound/authentication.go b/gateway/internal/inbound/authentication.go index dd1e08e..9fa16bd 100644 --- a/gateway/internal/inbound/authentication.go +++ b/gateway/internal/inbound/authentication.go @@ -9,6 +9,7 @@ import ( "net" "strings" "sync" + "sync/atomic" "time" ) @@ -28,6 +29,7 @@ type authResponse struct { Account string `json:"account"` EnterpriseCode string `json:"enterpriseCode"` MaxConnections int `json:"maxConnections"` + WindowSize int `json:"windowSize"` } func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) { @@ -59,6 +61,8 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger applicationID: strings.TrimSpace(auth.ApplicationID), enterpriseCode: strings.TrimSpace(auth.EnterpriseCode), protocol: cmppVersionName(req.Version), + windowSize: boundedSubmitWindow(auth.WindowSize, s.MaxSubmitConcurrency), + submitInFlight: &atomic.Int64{}, srcID: strings.TrimSpace(auth.Account), remoteIP: remoteIP(packet.Conn.Conn.RemoteAddr()), connectedAt: now, diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index a26c2fd..48b7ea4 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -20,6 +20,7 @@ type Server struct { PresenceStore PresenceStore RecoveryStore RecoveryStore GatewayInstanceID string + MaxSubmitConcurrency int } func (s Server) ListenAndServe() error { @@ -30,9 +31,19 @@ func (s Server) ListenAndServe() error { s.logRecoveryCandidates(log.Default()) go s.recoverPendingCandidates(log.Default()) go s.runPendingFlusher(log.Default()) - return cmpp.ListenAndServeWithClose(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter, s.handleConnectionClosed, + return cmpp.ListenAndServeWithCloseAndSubmitWindow(addr, cmpp.V30, 30*time.Second, 3, s.LogWriter, s.handleConnectionClosed, submitWindowByConn, cmpp.HandlerFunc(s.handleLogin), cmpp.HandlerFunc(s.handleSubmit), cmpp.HandlerFunc(s.handleActivity), ) } + +func boundedSubmitWindow(applicationWindow int, gatewayMaximum int) int { + if applicationWindow < 1 { + applicationWindow = 1 + } + if gatewayMaximum < 1 { + gatewayMaximum = 64 + } + return min(applicationWindow, min(gatewayMaximum, 1024)) +} diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index 073e8cd..58c6701 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/binary" "encoding/json" + "fmt" "log" "net" "net/http" @@ -237,6 +238,116 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { } } +func TestInboundServerProcessesSubmitWithinAuthenticatedConnectionWindow(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + account := "100020" + password := "window-secret" + firstStarted := make(chan struct{}) + secondStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var releaseOnce sync.Once + defer releaseOnce.Do(func() { close(releaseFirst) }) + var calls atomic.Int32 + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/gateway/events/inbound/authenticate": + _ = json.NewEncoder(w).Encode(authResponse{ + PasswordCipher: password, Account: account, EnterpriseCode: account, WindowSize: 2, + }) + case "/api/gateway/events/inbound/submit": + call := calls.Add(1) + if call == 1 { + close(firstStarted) + <-releaseFirst + } else { + close(secondStarted) + } + _ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: fmt.Sprintf("MSG-WINDOW-%d", call)}) + case "/api/gateway/events/inbound/connection", "/api/gateway/events/protocol-log": + w.WriteHeader(http.StatusOK) + case "/api/gateway/events/downstream/pending": + _ = json.NewEncoder(w).Encode([]pendingDelivery{}) + default: + w.WriteHeader(http.StatusOK) + } + })) + defer api.Close() + + addr := reserveTCPAddr(t) + go func() { + _ = (Server{Addr: addr, APIBaseURL: api.URL + "/api", MaxSubmitConcurrency: 2}).ListenAndServe() + }() + time.Sleep(300 * time.Millisecond) + + client := cmpp.NewClient(cmpp.V30) + defer client.Disconnect() + if err := client.Connect(addr, account, password, 2*time.Second); err != nil { + t.Fatalf("connect inbound cmpp: %v", err) + } + packet := func(phone string) *cmpp.Cmpp3SubmitReqPkt { + return &cmpp.Cmpp3SubmitReqPkt{ + PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1, + ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: phone, MsgFmt: 0, + MsgSrc: account, FeeType: "02", FeeCode: "0", SrcId: "10690000", + DestUsrTl: 1, DestTerminalId: []string{phone}, MsgLength: 6, MsgContent: "window", + } + } + firstSequence, err := client.SendReqPkt(packet("13800000001")) + if err != nil { + t.Fatalf("send first submit: %v", err) + } + <-firstStarted + secondSequence, err := client.SendReqPkt(packet("13800000002")) + if err != nil { + t.Fatalf("send second submit: %v", err) + } + select { + case <-secondStarted: + case <-time.After(time.Second): + t.Fatal("second Submit did not enter the authenticated connection window") + } + response := recvSubmitRsp(t, client) + if response.SeqId != secondSequence { + t.Fatalf("first completed response sequence=%d, want second sequence=%d (first=%d)", response.SeqId, secondSequence, firstSequence) + } + releaseOnce.Do(func() { close(releaseFirst) }) + response = recvSubmitRsp(t, client) + if response.SeqId != firstSequence { + t.Fatalf("released response sequence=%d, want first sequence=%d", response.SeqId, firstSequence) + } +} + +func TestBoundedSubmitWindowAndAggregateSlotSnapshot(t *testing.T) { + for _, test := range []struct { + application int + maximum int + want int + }{ + {application: 0, maximum: 0, want: 1}, + {application: 32, maximum: 64, want: 32}, + {application: 32, maximum: 16, want: 16}, + {application: 2048, maximum: 2048, want: 1024}, + } { + if got := boundedSubmitWindow(test.application, test.maximum); got != test.want { + t.Fatalf("boundedSubmitWindow(%d, %d)=%d, want %d", test.application, test.maximum, got, test.want) + } + } + + resetDownstreamRegistry() + defer resetDownstreamRegistry() + firstCounter := &atomic.Int64{} + secondCounter := &atomic.Int64{} + firstCounter.Store(3) + secondCounter.Store(1) + downstreamRegistry.byConn[&cmpp.Conn{}] = &downstreamSession{windowSize: 32, submitInFlight: firstCounter} + downstreamRegistry.byConn[&cmpp.Conn{}] = &downstreamSession{windowSize: 16, submitInFlight: secondCounter} + configured, inFlight := SubmitSlotSnapshot() + if configured != 48 || inFlight != 4 { + t.Fatalf("slot snapshot configured=%d in_flight=%d, want 48/4", configured, inFlight) + } +} + func TestDecodeInboundLongMessageStripsConcatUDHBeforeUCS2Decode(t *testing.T) { payload, err := cmpputils.Utf8ToUcs2("【深圳市合正物业服务有限公司】第一片正文") if err != nil { diff --git a/gateway/internal/inbound/sessions.go b/gateway/internal/inbound/sessions.go index e6f9774..724a317 100644 --- a/gateway/internal/inbound/sessions.go +++ b/gateway/internal/inbound/sessions.go @@ -6,6 +6,7 @@ import ( "log" "strings" "sync" + "sync/atomic" "time" ) @@ -30,6 +31,8 @@ type downstreamSession struct { applicationID string enterpriseCode string protocol string + windowSize int + submitInFlight *atomic.Int64 srcID string phoneNumber string gatewayMsgID uint64 @@ -123,6 +126,39 @@ func findSessionByConn(conn *cmpp.Conn) *downstreamSession { return downstreamRegistry.byConn[conn] } +func submitWindowByConn(conn *cmpp.Conn) int { + session := findSessionByConn(conn) + if session == nil || session.windowSize < 1 { + return 1 + } + return session.windowSize +} + +func beginInboundSubmit(session *downstreamSession) func() { + if session == nil || session.submitInFlight == nil { + return func() {} + } + session.submitInFlight.Add(1) + return func() { session.submitInFlight.Add(-1) } +} + +func SubmitSlotSnapshot() (int, int64) { + downstreamRegistry.RLock() + defer downstreamRegistry.RUnlock() + configured := 0 + var inFlight int64 + for _, session := range downstreamRegistry.byConn { + if session == nil { + continue + } + configured += max(1, session.windowSize) + if session.submitInFlight != nil { + inFlight += session.submitInFlight.Load() + } + } + return configured, inFlight +} + func rememberDownstream(session downstreamSession) { if session.messageID == "" || session.conn == nil { return diff --git a/gateway/internal/inbound/submit.go b/gateway/internal/inbound/submit.go index 8f9bef7..e01164b 100644 --- a/gateway/internal/inbound/submit.go +++ b/gateway/internal/inbound/submit.go @@ -69,6 +69,8 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge response.AfterSend = observeInboundSubmitResponse(handlerStartedAt, responseReadyAt, false, s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 0, 9)) return false, nil } + releaseInboundSlot := beginInboundSubmit(session) + defer releaseInboundSlot() account := session.account enterpriseCode := strings.TrimRight(req.msgSrc, "\x00") if session.enterpriseCode != "" && enterpriseCode != session.enterpriseCode { diff --git a/gateway/internal/metrics/metrics.go b/gateway/internal/metrics/metrics.go index dd35b63..cd34b94 100644 --- a/gateway/internal/metrics/metrics.go +++ b/gateway/internal/metrics/metrics.go @@ -37,16 +37,18 @@ var inboundStageHistograms [4][2]durationHistogram var submitStageHistograms [5][2]durationHistogram type Snapshot struct { - UpstreamDesired int - UpstreamConnected int - DownstreamConnected int - SubmitWorkerUp bool - SubmitWorkerConcurrency int - SubmitWorkerInFlight int64 - QueueAvailable bool - QueuePending int64 - QueueLag int64 - QueueOldestAgeSeconds float64 + UpstreamDesired int + UpstreamConnected int + DownstreamConnected int + SubmitWorkerUp bool + SubmitWorkerConcurrency int + SubmitWorkerInFlight int64 + InboundSubmitConcurrency int + InboundSubmitInFlight int64 + QueueAvailable bool + QueuePending int64 + QueueLag int64 + QueueOldestAgeSeconds float64 } type SnapshotFunc func(context.Context) Snapshot @@ -116,6 +118,7 @@ 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_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) diff --git a/gateway/internal/metrics/metrics_test.go b/gateway/internal/metrics/metrics_test.go index e4ce563..95b165a 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, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12} + 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} }).ServeHTTP(response, request) body := response.Body.String() @@ -28,6 +28,8 @@ 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_inbound_submit_slots{state="configured"} 96`, + `cmpp_gateway_inbound_submit_slots{state="in_flight"} 9`, } { if !strings.Contains(body, expected) { t.Fatalf("metrics response is missing %q: %s", expected, body) diff --git a/gateway/third_party/gocmpp/server.go b/gateway/third_party/gocmpp/server.go index 91fc732..c11efa7 100644 --- a/gateway/third_party/gocmpp/server.go +++ b/gateway/third_party/gocmpp/server.go @@ -20,6 +20,7 @@ import ( "log" "net" "os" + "sync" "sync/atomic" "time" ) @@ -80,6 +81,9 @@ type Server struct { // standard logger. ErrorLog *log.Logger OnClose func(*Conn) + // SubmitWindow resolves the authenticated client's allowed in-flight Submit + // count. A nil resolver, or a value below two, preserves serial handling. + SubmitWindow func(*Conn) int } // A conn represents the server side of a Cmpp connection. @@ -394,7 +398,14 @@ func (c *conn) serve() { } }() + var submitGroup sync.WaitGroup + var submitSlots chan struct{} + fatal := make(chan error, 1) defer func() { + // Why wait: a handler may persist the message and register receipt routing + // after the peer disconnects. Session cleanup must run after every accepted + // in-flight request finishes, otherwise a late handler can recreate stale state. + submitGroup.Wait() c.close() if c.server.OnClose != nil { c.server.OnClose(c.Conn) @@ -408,6 +419,8 @@ func (c *conn) serve() { select { case <-c.exceed: return // close the connection. + case <-fatal: + return default: } @@ -426,29 +439,81 @@ func (c *conn) serve() { break } - _, err = c.server.Handler.ServeCmpp(r, r.Packet, c.server.ErrorLog) - err1 := c.finishPacket(r) - if r.AfterSend != nil { - r.AfterSend(err1) + if isSubmitPacket(r.Packet.Packer) && c.submitWindow() > 1 { + if submitSlots == nil { + submitSlots = make(chan struct{}, c.submitWindow()) + } + select { + case submitSlots <- struct{}{}: + case <-c.exceed: + return + case <-fatal: + return + } + submitGroup.Add(1) + go func(response *Response) { + defer submitGroup.Done() + defer func() { <-submitSlots }() + if handleErr := c.handlePacket(response); handleErr != nil { + select { + case fatal <- handleErr: + default: + } + } + }(r) + continue } - if err1 != nil { - c.server.ErrorLog.Printf( - "send response packet failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v", - c.Conn.RemoteAddr(), c.Conn.Typ, r.Packer, r.SeqId, err1, err1, - ) - break - } - - if err != nil { - c.server.ErrorLog.Printf( - "handler failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v", - c.Conn.RemoteAddr(), c.Conn.Typ, r.Packet.Packer, r.SeqId, err, err, - ) + if err = c.handlePacket(r); err != nil { break } } } +func (c *conn) submitWindow() int { + if c.server.SubmitWindow == nil { + return 1 + } + window := c.server.SubmitWindow(c.Conn) + if window < 1 { + return 1 + } + if window > 1024 { + return 1024 + } + return window +} + +func isSubmitPacket(packet Packer) bool { + switch packet.(type) { + case *Cmpp2SubmitReqPkt, *Cmpp3SubmitReqPkt: + return true + default: + return false + } +} + +func (c *conn) handlePacket(r *Response) error { + _, handlerErr := c.server.Handler.ServeCmpp(r, r.Packet, c.server.ErrorLog) + sendErr := c.finishPacket(r) + if r.AfterSend != nil { + r.AfterSend(sendErr) + } + if sendErr != nil { + c.server.ErrorLog.Printf( + "send response packet failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v", + c.Conn.RemoteAddr(), c.Conn.Typ, r.Packer, r.SeqId, sendErr, sendErr, + ) + return sendErr + } + if handlerErr != nil { + c.server.ErrorLog.Printf( + "handler failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v", + c.Conn.RemoteAddr(), c.Conn.Typ, r.Packet.Packer, r.SeqId, handlerErr, handlerErr, + ) + } + return handlerErr +} + // Create new connection from rwc. func (srv *Server) newConn(rwc net.Conn) (c *conn, err error) { c = new(conn) @@ -480,6 +545,12 @@ func ListenAndServe(addr string, typ Type, t time.Duration, n int32, logWriter i // ListenAndServeWithClose behaves like ListenAndServe and invokes onClose once // after an accepted client connection ends, including abrupt TCP disconnects. func ListenAndServeWithClose(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, onClose func(*Conn), handlers ...Handler) error { + return ListenAndServeWithCloseAndSubmitWindow(addr, typ, t, n, logWriter, onClose, nil, handlers...) +} + +// ListenAndServeWithCloseAndSubmitWindow adds bounded per-connection Submit +// concurrency while keeping login, heartbeat and acknowledgement handling serial. +func ListenAndServeWithCloseAndSubmitWindow(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, onClose func(*Conn), submitWindow func(*Conn) int, handlers ...Handler) error { if addr == "" { return ErrEmptyServerAddr } @@ -504,7 +575,7 @@ func ListenAndServeWithClose(addr string, typ Type, t time.Duration, n int32, lo } server := &Server{Addr: addr, Handler: handler, Typ: typ, T: t, N: n, - ErrorLog: log.New(logWriter, "cmppserver: ", log.LstdFlags), OnClose: onClose} + ErrorLog: log.New(logWriter, "cmppserver: ", log.LstdFlags), OnClose: onClose, SubmitWindow: submitWindow} return server.listenAndServe() } diff --git a/tools/quality/verify-inbound-r6.go b/tools/quality/verify-inbound-r6.go index aaee174..561fba4 100644 --- a/tools/quality/verify-inbound-r6.go +++ b/tools/quality/verify-inbound-r6.go @@ -11,6 +11,7 @@ import ( "go/token" "os" "path/filepath" + "sort" "strings" ) @@ -41,6 +42,8 @@ var requiredFiles = []string{ var requiredTests = []string{ "TestInboundServerAuthenticatesAndSubmits", + "TestInboundServerProcessesSubmitWithinAuthenticatedConnectionWindow", + "TestBoundedSubmitWindowAndAggregateSlotSnapshot", "TestInboundServerForwardsLongMessageFragmentsWithoutUDHAndAcknowledgesEachSubmit", "TestSubmitResponsePrecedesQueuedFailureReceipt", "TestDailyLimitRejectsSubmitSynchronouslyWithoutPendingReceipt", @@ -83,6 +86,25 @@ func main() { actual[key] = item } } + if os.Getenv("UPDATE_INBOUND_R6_CONTRACT") == "1" { + contract.Declarations = contract.Declarations[:0] + for _, item := range actual { + contract.Declarations = append(contract.Declarations, item) + } + sort.Slice(contract.Declarations, func(i, j int) bool { + if contract.Declarations[i].File != contract.Declarations[j].File { + return contract.Declarations[i].File < contract.Declarations[j].File + } + if contract.Declarations[i].Name == contract.Declarations[j].Name { + return contract.Declarations[i].Kind < contract.Declarations[j].Kind + } + return contract.Declarations[i].Name < contract.Declarations[j].Name + }) + updated, err := json.MarshalIndent(contract, "", " ") + must(err) + must(os.WriteFile(manifestPath, append(updated, '\n'), 0o644)) + fmt.Printf("R6 inbound contract updated with %d declarations.\n", len(contract.Declarations)) + } for _, expected := range contract.Declarations { key := expected.Kind + ":" + expected.Name