From 55019443c0015c65da94f0ad17ed274d77e6290a Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Wed, 22 Jul 2026 17:09:44 +0800 Subject: [PATCH] fix: reject CMPP daily limit synchronously --- api/src/send-chain/send-chain.service.spec.ts | 11 ++-- api/src/send-chain/send-chain.service.ts | 38 +++++++++---- .../first-version-development-requirements.md | 4 +- docs/system-functional-test-cases.md | 6 +- docs/testing-progress.md | 6 ++ gateway/internal/inbound/server.go | 11 +++- gateway/internal/inbound/server_test.go | 56 +++++++++++++++++++ 7 files changed, 112 insertions(+), 20 deletions(-) diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index d760470..212d02c 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -850,7 +850,7 @@ describe('SendChainService', () => { expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledTimes(2); }); - it('rejects every destination in one CMPP Submit with auditable receipts when the daily limit is exceeded', async () => { + it('rejects the whole CMPP Submit synchronously while keeping per-destination audit records when the daily limit is exceeded', async () => { const { service, prisma, billing } = createService(); prisma.$queryRaw.mockResolvedValueOnce([{ dailyLimit: 1, usedCount: null }]); let taskIndex = 0; @@ -866,12 +866,13 @@ describe('SendChainService', () => { remoteIp: '127.0.0.1', }); - expect(result).toEqual(expect.objectContaining({ accepted: true, phoneCount: 2 })); + expect(result).toEqual(expect.objectContaining({ accepted: false, result: 8, phoneCount: 2 })); expect(prisma.smsMessageRecord.create).toHaveBeenCalledTimes(2); - expect(prisma.smsReceiptRecord.create).toHaveBeenCalledTimes(2); - expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({ - data: expect.objectContaining({ errorCode: 'DAILY_LIMIT', receiptStatus: 'undelivered' }), + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ status: 'rejected', errorCode: 'DAILY_LIMIT' }), }); + expect(prisma.smsReceiptRecord.create).not.toHaveBeenCalled(); + expect(prisma.cmppDownstreamDelivery.create).not.toHaveBeenCalled(); expect(billing.freeze).not.toHaveBeenCalled(); }); diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 44f1d10..9116374 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -2028,9 +2028,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { throw new BadRequestException('CMPP account is invalid'); } const dailyQuota = await this.tryReserveDailySendQuota(application.id, phoneNumbers.length); - const dailyLimitFailure = dailyQuota.reserved + const dailyLimitRejection = dailyQuota.reserved ? undefined - : `应用当日发送上限${dailyQuota.dailyLimit}条，本次${phoneNumbers.length}条超出剩余配额`; + : { + code: 'DAILY_LIMIT', + reason: `应用当日发送上限${dailyQuota.dailyLimit}条，本次${phoneNumbers.length}条超出剩余配额`, + }; const submitGroupMessageId = `MSG-${randomUUID()}`; const submissions = phoneNumbers.map((phoneNumber, index) => ({ @@ -2045,11 +2048,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { ...data, phoneNumber: submission.phoneNumber, phoneNumbers: undefined, - }, submission.messageId, submitGroupMessageId, dailyLimitFailure)))); + }, submission.messageId, submitGroupMessageId, dailyLimitRejection)))); } const first = results[0]; return { ...first, + result: dailyLimitRejection ? 8 : undefined, phoneCount: results.length, messages: results.map((result, index) => ({ phoneNumber: phoneNumbers[index], @@ -2065,7 +2069,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { data: GatewayInboundSubmitDto & { phoneNumber: string }, messageId: string, submitGroupMessageId: string, - dailyLimitFailure?: string, + synchronousRejection?: { code: string; reason: string }, ) { const application = await this.findInboundApplication(data.account); if (!application) { @@ -2098,7 +2102,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { sourceType: 'cmpp', content: data.content, phoneTotal: 1, - status: 'validating', + status: synchronousRejection ? 'rejected' : 'validating', + auditStatus: synchronousRejection ? 'rejected' : undefined, + rejectReason: synchronousRejection?.reason, progressTotal: 1, }, }); @@ -2110,7 +2116,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { sourceIp: data.remoteIp, userAgent: 'cmpp-gateway', payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account }, - status: 'accepted', + status: synchronousRejection ? 'rejected' : 'accepted', }, }); const message = await this.prisma.smsMessageRecord.create({ @@ -2130,10 +2136,24 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { cmppSubmitGroupMessageId: submitGroupMessageId, clientSrcId, applicationExtension: application.cmppApplicationExtension, - status: 'validating', + status: synchronousRejection ? 'rejected' : 'validating', + errorCode: synchronousRejection?.code, + errorMessage: synchronousRejection?.reason, }, }); + if (synchronousRejection) { + return { + accepted: false, + tenantId: application.tenantId, + applicationId: application.id, + taskId: task.id, + messageId: message.messageId, + messageRecordId: message.id, + status: 'rejected', + }; + } + const reject = async (code: string, reason: string) => { await this.prisma.smsBatchTask.update({ where: { id: task.id }, @@ -2203,9 +2223,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }); await this.enqueueBatchTask(task.id); }; - if (dailyLimitFailure) { - await reject('DAILY_LIMIT', dailyLimitFailure); - } else if (application.status !== 'active' || application.tenant.status !== 'active') { + if (application.status !== 'active' || application.tenant.status !== 'active') { await reject('ACCOUNT', '企业或短信应用已停用'); } else if (!application.interfaceEnabled) { await reject('INTERFACE', '短信应用 CMPP 接口已停用'); diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index cd09cc9..81c3026 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -1635,5 +1635,7 @@ ## 2026-07-22 应用日发送上限与HTTP参数默认值补充 - 每个短信应用的日发送上限默认为100000条，按北京时间自然日和去重后目标号码数计数。客户端、公开HTTP和下游CMPP入站必须共用PostgreSQL原子配额计数，多API实例并发不得突破上限。 -- 客户端/HTTP整批超限时不创建任务、短信记录或账务冻结，HTTP返回429及`DAILY_SEND_LIMIT_EXCEEDED`。CMPP合法Submit超限时仍保留每个号码的可审计失败主记录，不冻结/扣费，并以`DAILY_LIMIT`失败回执通知客户。 +- 客户端/HTTP整批超限时不创建任务、短信记录或账务冻结，HTTP返回429及`DAILY_SEND_LIMIT_EXCEEDED`。CMPP合法Submit整包超限时返回唯一一个非0 `SUBMIT_RESP`（日限额映射`result=8`），每个目的号码仍保留`rejected/DAILY_LIMIT`审计主记录；不冻结、不扣费、不占用额度，也不再生成或投递异步`DELIVER`回执。 +- 日额度在任务正式受理时按北京时间占用；待审核和定时任务占用受理日额度，后续审核拒绝、取消或发送失败均不返还。发送校验固定遵循身份/应用权限、请求结构和全部号码基础格式、签名/模板/引流、禁发时段及风控、路由通道、余额原子校验与冻结、日额度原子占用、任务落库入队的业务优先级。余额早期读取只能用于提示，最终资格判断与冻结必须紧邻任务受理执行。 +- 号码基础校验只判断空值、字符、长度、数量上限、重复及多号码完整性；号段识别用于运营商和地区快照及路由提示，未知号段不得据此拒绝，必须继续按全国或三网兼容通道处理。 - 首次开通HTTP接口时，后端默认开启单条发送、状态查询、回执回调、上行查询、上行回调和客户端自助密钥六项能力，回执/上行投递默认为HTTP Webhook。参数复制必须包含应用名称、AppID、六项能力、基础地址、文档、QPS、白名单和真实投递方式。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 1a82115..d421df2 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3714,6 +3714,10 @@ npm run verify:phase8 | TC-SEND-DAILY-001 | 新建应用不传dailyLimit | PostgreSQL保存100000，返回值与页面均显示100000 | | TC-SEND-DAILY-002 | 当日剩余1条时，客户端或HTTP同时发2个号码 | 整批返回429/DAILY_SEND_LIMIT_EXCEEDED，不新建任务、短信记录、冻结或队列作业 | | TC-SEND-DAILY-003 | 两个API实例并发争抢最后配额 | 依赖`applicationId+usageDate`唯一索引与条件upsert，只有不突破上限的请求成功 | -| TC-SEND-DAILY-004 | 多号码CMPP Submit整包超限 | 仅一个SubmitResp；每个号码均有rejected主记录和DAILY_LIMIT回执，无冻结和扣费 | +| TC-SEND-DAILY-004 | 多号码CMPP Submit整包超限 | 仅一个非0 SubmitResp且`result=8`、Msg_Id为0；每个号码均有`rejected/DAILY_LIMIT`主记录，无冻结、扣费、额度占用、SmsReceiptRecord和下游DELIVER投递 | +| TC-SEND-DAILY-005 | Bind后历史待回执拉取完成，再提交日限额超限包 | Bind阶段允许拉取既有pending；超限Submit不得新增pending回执或建立Msg_Id/Sequence映射，连接保持可用 | +| TC-SEND-DAILY-006 | 待审核/定时任务受理后再拒绝或取消 | 受理日额度已占用且不返还；失败、取消和最终送达统计不得反向修改日用量 | +| TC-SEND-ORDER-001 | 请求同时存在号码格式、模板和余额错误 | 按固定业务顺序先返回号码基础错误；修复号码后返回签名/模板错误，只有业务资格通过后才执行最终余额原子校验/冻结 | +| TC-SEND-ORDER-002 | 号段库无法识别但号码基础格式合法 | 不以未知号段拒绝；记录未知运营商快照并继续匹配全国或三网兼容通道，正常产生提交、消费和回执数据 | | TC-HTTP-PARAM-002 | 首次开通HTTP后查看并复制参数 | 六项能力默认开启，回执/上行为HTTP Webhook；复制文本含AppID和“客户端自助密钥”，与真实API/DB一致 | | TC-HTTP-PARAM-003 | 升级前已开通HTTP且Webhook能力开启，投递模式仍为cmpp | migration将对应回执/上行模式回填为http，参数复制不再显示CMPP长连接 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index aabbcf5..3c481b8 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -2219,3 +2219,9 @@ git diff --check - 发布后`cmpp-gateway`、`cmpp-api`、Nginx、PostgreSQL、Redis和MinIO均active，`12026/17890/8090/3000/9000/6379/5432`监听；API/Gateway health、Redis PONG均通过。2个active上游通道均恢复`connected/currentConnections=1`，Redis中7个通道权威TPS配置存在；`gateway.submit.commands` consumer group为`pending=0、lag=0`，部署后API/Gateway error journal均为0。 - 外部首页、运营登录页、客户端登录页和API health均返回HTTP 200，公网CMPP `8.160.169.106:17890` TCP连接成功。应用内Browser两次在导航/DOM读取阶段控制超时并重置，因此只记录外部HTTP入口通过，不虚报浏览器DOM、交互或console验收通过。 - 本次业务数据变更仅来自两条已审查migration；未发送或重投短信，未充值、审核、删除、禁用账号、改密或修改真实通道配置。 + +## 2026-07-22 CMPP日限额同步整包拒绝口径修正（待提交发布） + +- 产品确认日额度按任务正式受理的北京时间自然日占用，待审核和定时任务计入受理日；后续审核拒绝、取消或发送失败不返还。号码基础校验不以号段库是否识别作为合法性条件，未知号段继续走全国或三网兼容通道。 +- 修正CMPP整包日限额语义：API为每个目的号码建立`rejected/DAILY_LIMIT`审计主记录，但返回`accepted=false/result=8`；Gateway同步返回唯一一个非0 `SUBMIT_RESP`且Msg_Id为0，不建立下游会话映射、不生成`SmsReceiptRecord`、不创建`CmppDownstreamDelivery`，不冻结或扣费。 +- 新增API回归覆盖双号码整包超限、每号码审计记录及无回执/无冻结；新增Gateway真实CMPP2.0协议回归覆盖Bind后的`result=8`响应和不新增pending回执拉取。失败测试先在旧实现上分别暴露`accepted=true`和Gateway响应结构缺少业务结果码，修改后SendChain定向78项及Gateway inbound包通过。 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 1017e2d..a378e2f 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -61,6 +61,7 @@ type submitResponseMessage struct { type submitResponse struct { Accepted bool `json:"accepted"` + Result uint32 `json:"result,omitempty"` MessageID string `json:"messageId"` Messages []submitResponseMessage `json:"messages,omitempty"` } @@ -316,11 +317,15 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge if err != nil { reason = err.Error() } + responseResult := result.Result + if responseResult == 0 { + responseResult = 9 + } logger.Printf( - "cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=9 stage=business duration_ms=%d content_chars=%d content_hash=%s reason=%q", - clientProtocol, req.protocol, account, remote, req.sequenceID, phone, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason, + "cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=%d stage=business duration_ms=%d content_chars=%d content_hash=%s reason=%q", + clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason, ) - setInboundSubmitResponse(response.Packer, 0, 9) + setInboundSubmitResponse(response.Packer, 0, responseResult) return false, nil } gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID) diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index b096e7f..a7049ba 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -12,6 +12,7 @@ import ( "reflect" "strings" "sync" + "sync/atomic" "testing" "time" @@ -310,6 +311,61 @@ func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) { } } +func TestDailyLimitRejectsSubmitSynchronouslyWithoutPendingReceipt(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + account := "100001" + password := "secret-hash" + var pendingCalls 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}) + case "/api/gateway/events/inbound/submit": + _ = json.NewEncoder(w).Encode(submitResponse{Accepted: false, Result: 8, MessageID: "MSG-LIMIT"}) + case "/api/gateway/events/downstream/pending": + pendingCalls.Add(1) + _ = json.NewEncoder(w).Encode([]pendingDelivery{}) + case "/api/gateway/events/inbound/connection": + w.WriteHeader(http.StatusOK) + default: + t.Fatalf("unexpected api path: %s", r.URL.Path) + } + })) + defer api.Close() + + addr := reserveTCPAddr(t) + go func() { _ = (Server{Addr: addr, APIBaseURL: api.URL + "/api"}).ListenAndServe() }() + time.Sleep(300 * time.Millisecond) + + client := cmpp.NewClient(cmpp.V20) + defer client.Disconnect() + if err := client.Connect(addr, account, password, 2*time.Second); err != nil { + t.Fatalf("connect CMPP2 inbound: %v", err) + } + time.Sleep(100 * time.Millisecond) + pendingCallsAfterBind := pendingCalls.Load() + content, _ := cmpputils.Utf8ToUcs2("日限额拒绝") + if _, err := client.SendReqPkt(&cmpp.Cmpp2SubmitReqPkt{ + PkTotal: 1, PkNumber: 1, RegisteredDelivery: 1, MsgLevel: 1, + ServiceId: "cmpp", FeeUserType: 2, FeeTerminalId: "13500002696", + MsgFmt: 8, MsgSrc: account, FeeType: "02", FeeCode: "0", SrcId: "10690000", + DestUsrTl: 1, DestTerminalId: []string{"13500002696"}, MsgLength: uint8(len(content)), MsgContent: content, + }); err != nil { + t.Fatalf("send submit: %v", err) + } + + rsp := recvSubmitRsp20(t, client) + if rsp.Result != 8 || rsp.MsgId != 0 { + t.Fatalf("expected synchronous daily-limit result=8 without Msg_Id, got %+v", rsp) + } + time.Sleep(100 * time.Millisecond) + if pendingCalls.Load() != pendingCallsAfterBind { + t.Fatalf("daily-limit rejection must not add downstream receipt polling, before=%d after=%d", pendingCallsAfterBind, pendingCalls.Load()) + } +} + func TestReceiptWithoutOriginalSequenceIsUnrecoverable(t *testing.T) { resetDownstreamRegistry() defer resetDownstreamRegistry()