From 1ce02ef2066aa1ecd995b0a3b884304218adc008 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Tue, 14 Jul 2026 16:22:41 +0800 Subject: [PATCH] fix: correlate downstream receipts with submit responses --- .../migration.sql | 11 ++ api/prisma/schema.prisma | 1 + api/src/send-chain/send-chain.service.spec.ts | 39 +++- api/src/send-chain/send-chain.service.ts | 28 ++- .../first-version-development-requirements.md | 2 + docs/system-functional-test-cases.md | 3 +- docs/testing-progress.md | 9 + gateway/internal/inbound/server.go | 67 ++++++- gateway/internal/inbound/server_test.go | 110 ++++++++++++ gateway/third_party/gocmpp/server.go | 9 +- .../admin/AdminDownstreamDeliveriesPage.tsx | 166 ++++++++++-------- src/styles/global.css | 133 ++++++++++++++ 12 files changed, 487 insertions(+), 91 deletions(-) create mode 100644 api/prisma/migrations/20260714153000_fix_downstream_receipt_message_id/migration.sql diff --git a/api/prisma/migrations/20260714153000_fix_downstream_receipt_message_id/migration.sql b/api/prisma/migrations/20260714153000_fix_downstream_receipt_message_id/migration.sql new file mode 100644 index 0000000..83d9d2d --- /dev/null +++ b/api/prisma/migrations/20260714153000_fix_downstream_receipt_message_id/migration.sql @@ -0,0 +1,11 @@ +ALTER TABLE "SmsMessageRecord" +ADD COLUMN "cmppSubmitSequenceId" TEXT; + +UPDATE "CmppDownstreamDelivery" +SET + "status" = 'unconfirmed', + "deliveredAt" = NULL, + "lastError" = 'CMPP_DELIVER_RESP Msg_Id=0，仅确认协议收包，未确认业务回执关联' +WHERE "deliveryType" = 'receipt' + AND "status" = 'delivered' + AND COALESCE("ackMessageId", '0') = '0'; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 67698be..9337250 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -983,6 +983,7 @@ model SmsMessageRecord { channelId String? submitId String? gatewayMessageId String? + cmppSubmitSequenceId String? status String @default("queued") submitStatus String? receiptStatus String? diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index b0a8a4a..c06d2d6 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -551,15 +551,20 @@ describe('SendChainService', () => { account: '100001', phoneNumber: '13800000001', content: 'unreported content', + sequenceId: 1216579149, remoteIp: '127.0.0.1', })).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' })); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ cmppSubmitSequenceId: '1216579149' }), + }); + expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({ data: expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }), }); expect(service['postGatewayControl']).toHaveBeenCalledWith( '/downstream/receipt', - expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }), + expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE', submitSequenceId: 1216579149 }), ); expect(riskReview.aggregateTemplateMismatch).not.toHaveBeenCalled(); }); @@ -1484,6 +1489,29 @@ describe('SendChainService', () => { })); }); + it('does not treat Result=0 with Msg_Id=0 as a business acknowledgement', async () => { + const { service, prisma } = createService(); + + await service.acknowledgeDownstreamDelivery({ + id: 'delivery-1', + connectionId: 'conn-1', + sequenceId: '91', + messageId: '0', + result: 0, + acknowledgedAt: '2026-07-14T07:07:49.336Z', + }); + + expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ ackResult: 0, ackMessageId: '0' }), + })); + expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ status: 'pending', lastError: expect.stringContaining('Msg_Id=0') }), + })); + expect(prisma.cmppDownstreamDelivery.updateMany).not.toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ status: 'delivered' }), + })); + }); + it('does not automatically retry an unacknowledged delivery when its policy snapshot is disabled', async () => { const { service, prisma } = createService(); prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue({ @@ -1564,6 +1592,15 @@ describe('SendChainService', () => { resourceId: 'delivery-1', }), }); + expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ + status: 'pending', + acknowledgedAt: null, + ackResult: null, + ackMessageId: null, + deliveredAt: null, + }), + })); }); it('supports batch requeue of downstream deliveries', async () => { diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 11fb5ed..65ed9f5 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -119,7 +119,7 @@ export interface GatewayDownstreamAcknowledgedDto extends GatewayDownstreamSentD acknowledgedAt?: string; } -export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'connection_lost'; +export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'ack_invalid' | 'connection_lost'; type GatewayControlDeliveryResult = { sent?: boolean; @@ -987,7 +987,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { async acknowledgeDownstreamDelivery(data: GatewayDownstreamAcknowledgedDto) { const acknowledgedAt = asDateOrNull(data.acknowledgedAt) ?? new Date(); - if (data.result === 0) { + const acknowledgedMessageId = String(data.messageId ?? '').trim(); + if (data.result === 0 && acknowledgedMessageId !== '' && acknowledgedMessageId !== '0') { await this.prisma.cmppDownstreamDelivery.updateMany({ where: { id: data.id, status: { not: 'delivered' } }, data: { @@ -1015,6 +1016,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { connectionId: data.connectionId, }, }); + if (data.result === 0) { + return this.markDownstreamDeliveryFailed(data.id, 'CMPP_DELIVER_RESP Msg_Id=0，客户端仅确认协议收包，无法关联原短信', 'ack_invalid'); + } return this.markDownstreamDeliveryFailed(data.id, `downstream CMPP_DELIVER_RESP result=${data.result}`, 'ack_rejected'); } @@ -1027,7 +1031,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return delivery; } const retryCount = (delivery.retryCount ?? 0) + 1; - const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'connection_lost'; + const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'ack_invalid' || failureType === 'connection_lost'; const retryAllowed = !acknowledgementFailure || delivery.retryEnabled !== false; const finalFailure = !retryAllowed || retryCount >= downstreamMaxRetries(); const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed'; @@ -1265,7 +1269,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }; await this.prisma.cmppDownstreamDelivery.update({ where: { id: delivery.id }, - data: { status: 'pending', retryCount: 0, nextRetryAt: null, ackDeadlineAt: null, lastError: null }, + data: { + status: 'pending', + retryCount: 0, + nextRetryAt: null, + sentAt: null, + acknowledgedAt: null, + ackDeadlineAt: null, + ackResult: null, + ackSequenceId: null, + ackMessageId: null, + connectionId: null, + deliveredAt: null, + lastError: null, + }, }); try { const result = await this.postGatewayControl(path, requestPayload) as GatewayControlDeliveryResult; @@ -1649,6 +1666,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { unitPrice: billing.unitPrice, amountCents: billing.amountCents, queuePriority, + cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId), status: 'validating', }, }); @@ -2140,6 +2158,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationId?: string | null; messageId: string; phoneNumber: string; + cmppSubmitSequenceId?: string | null; }, errorCode: string, reason: string, @@ -2181,6 +2200,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { rawStatus: 'REJECTD', errorCode, errorMessage: reason, + submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined, deliveredAt: deliveredAt.toISOString(), }, }); diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 189d6ed..bc1eaaa 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -280,6 +280,8 @@ - 下游投递重试已改为指数退避第一版：首次失败后按基础间隔重试，随后按 2 倍递增，并受最大退避上限约束，避免客户长时间离线时平台每分钟机械重试。 - 下游状态必须以客户确认作为终态：Gateway `SendPkt` 成功后只能写 `awaiting_ack`，仅收到匹配连接、`Sequence_Id`、`Msg_Id` 且 `CMPP_DELIVER_RESP.Result=0` 后才能写 `delivered`；超时、非零 Result 和历史未留存 ACK 的记录分别按未确认、拒绝或历史未确认展示，不能再把 TCP 写出冒充客户已收到。 - 企业应用必须分别提供“回执自动重试投递”和“上行短信自动重试投递”开关，默认开启。开关按投递创建时快照保存；关闭只阻止已经写出但未获 ACK/被拒绝后的自动重发，不阻止离线队列在客户首次上线时完成首次投递。手工重投不受开关限制，但必须提示重复业务处理风险并二次确认。 +- 客户 Submit 的失败状态回执必须严格晚于对应 `CMPP_SUBMIT_RESP` 写出，且 Deliver 中的业务 `Msg_Id` 必须非 0、与该 SubmitResp 返回的 `Msg_Id` 完全一致；`Result=0` 但 `Msg_Id=0` 只能表示客户端协议栈收包，不能标记业务回执已确认。平台必须持久化原 Submit Sequence_Id，使 Gateway 重启或客户重连后的补投仍可重建相同业务 `Msg_Id`。 +- 运营端允许对 `delivered`（客户端已确认）记录再次手工重投，但单条和批量入口都必须明确提示可能造成下游重复处理；`awaiting_ack` 状态在确认窗口内不得并发重投。 - 已实现客户侧最终 Deliver 推送的第一版能力：Gateway 在下游 Submit 被接受后记录 messageId 到客户连接的内存映射；NestJS 收到最终 receipt/uplink 并入库后调用 Gateway `/downstream/receipt`、`/downstream/uplink`，Gateway 向仍在线的客户 CMPP 连接下发 Deliver Receipt 或普通 Deliver。 - 已实现客户侧 Deliver 持久化第一版能力：NestJS 收到最终 receipt/uplink 后写入 `CmppDownstreamDelivery` 待投递记录；在线推送成功标记 delivered，客户断线或 Gateway 不可达时保留 pending 并记录 retry 信息；客户重新 bind 后 Gateway 按账号拉取 pending 记录补发。 - 已实现普通上行匹配与人工认领第一版：优先按 messageId 精确匹配；无 messageId 时按接入号匹配应用路由；仍无唯一应用时按手机号和最近下发时间窗口匹配；多候选标记 ambiguous 并写入 `SmsUplinkMatchCandidate` 候选，运营端可人工认领候选应用/下发记录；认领后更新上行记录、保留候选审计，并创建真实客户侧上行 Deliver 投递记录。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 2302e1b..774a482 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3290,4 +3290,5 @@ npm run verify:phase8 | TC-GW-ACK-001 | 客户在线时分别触发一条状态回执和一条上行短信，Gateway `SendPkt` 成功后延迟返回 `CMPP_DELIVER_RESP`。 | 写出后 `CmppDownstreamDelivery.status=awaiting_ack`；只有匹配连接、Sequence_Id、Msg_Id 且 Result=0 后才变为 `delivered`，并保存写出时间、确认时间、ACK 字段和连接 ID。 | | TC-GW-ACK-002 | 分别返回非零 Result、不返回响应直到超时、重启 Gateway 后让旧 `awaiting_ack` 超时。 | 非零 Result 和超时不会误记为 delivered；Gateway 重启后 NestJS 能恢复过期 ACK，按策略进入退避重试或最终 `rejected/unconfirmed`。 | | TC-GW-ACK-003 | 在企业应用中分别关闭“回执自动重试”和“上行短信自动重试”，各制造一次 ACK 超时，再重新开启并创建新投递。 | 关闭只影响对应类型的新投递策略快照；离线后的首次投递仍会在重连时执行；已写出未确认的记录不自动重发；重新开启后新记录按退避策略重试。 | -| TC-GW-ACK-004 | 对 `unconfirmed/rejected/failed` 记录执行单条和批量手工重投。 | 页面提示重复处理风险并二次确认；真实调用 Gateway；`awaiting_ack/delivered` 不允许重投；重发复用同一业务 Msg_Id。 | +| TC-GW-ACK-004 | 对 `unconfirmed/rejected/failed/delivered` 记录执行单条和批量手工重投。 | 页面提示重复处理风险并二次确认；真实调用 Gateway；`awaiting_ack` 不允许并发重投；重发复用同一业务 Msg_Id。 | +| TC-GW-ACK-005 | 客户 Submit 后由业务校验立即生成失败回执，并覆盖在线即时投递、Gateway 重启后恢复投递；另模拟客户端对 `Msg_Id=0` 返回 Result=0。 | 客户收到的第一个响应包必须是对应 `CMPP_SUBMIT_RESP`，之后 Deliver 的 `Msg_Id` 非 0 且与 SubmitResp 完全一致；重启后根据持久化 Submit Sequence_Id 重建同一 Msg_Id；Result=0/Msg_Id=0 不得写为 delivered。 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 3b12aa5..561ad7f 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1727,3 +1727,12 @@ git diff --check - 功能提交 `8c03663f` 已 push 并部署生产。部署前 PostgreSQL 备份为 `/opt/cmpp-platform/backups/cmpp-20260714-142059.sql`（约 65MB），运行源码备份为 `/opt/cmpp-platform/backups/source-20260714-142059.tar.gz`（约 21MB）；发布包本地和服务器 SHA-256 均为 `3e969aaf00e828da4167018bc349795413743414bc827d452859f0fd45776cd1`。 - 生产 migration `20260714130000_add_downstream_delivery_ack_tracking` 成功应用，37 条 migration 全部齐全；历史 7 条 `delivered` 已按新口径迁移为 `unconfirmed`，204 个企业应用的回执和上行自动重试开关均为默认开启。生产 `.deployed-commit=8c03663f245c12cd5e6aab28ad9eeb68ea60ed57`，`cmpp-api`、`cmpp-gateway`、Nginx、MinIO 均 active，`12026/17890/8090/3000` 监听，API/Gateway health、Redis、外部首页和运营端登录页均正常，部署后近期日志无新增 error；前端产物已确认包含“回执自动重试”和“客户端已确认”。未擅自发送或重投客户短信。 - 生产验证站点当前为 HTTP，已按发布约束显式设置 `SESSION_COOKIE_SECURE=false`，ACK 超时设置为 30 秒；无 Cookie 和无效 Cookie 访问受保护/会话接口均返回预期 401。服务器凭据文件对存量管理员只记录 `password=unchanged`，不是可用明文密码，因此未继续进行真实登录和开关点击，且已清除本次诊断产生的一次失败计数；登录后的页面交互由用户使用现有账号验收。 + +## 2026-07-14 下游回执 Msg_Id=0 与 SubmitResp 时序修复 + +- 生产核查 15:07、15:10 两条余额失败短信：`CmppDownstreamDelivery` 均记录 Result=0，但 ACK Msg_Id 都是 0；Gateway 日志中的原 SubmitResp Msg_Id 分别为 `9677414932210841325`、`9687782471788644609`。这只能证明下游协议栈收到了 Deliver，不能证明下游平台已将状态回执关联到原短信，与下游页面仍显示“未回执”一致。 +- 根因为 NestJS 在处理入站 Submit 时同步生成失败回执，而 Gateway 尚未建立 messageId 映射；原查找逻辑在精确消息未命中时回退账号会话，使用 bind 会话的零值 Msg_Id 提前写出 Deliver。上午 11:40 的回执手工重投后才显示，是因为重投时对应 Submit 映射已经存在，能携带正确 Msg_Id。 +- 修复为：精确消息未登记时禁止回退账号会话；Gateway 仅在 ConnectResp/SubmitResp 成功写出后冲刷 pending；新建短信记录持久化原 CMPP Submit Sequence_Id，重启恢复时结合平台 MessageId 重建相同 Msg_Id；发送层拒绝任何 Msg_Id=0 的 Deliver，NestJS 也不再把 Result=0/Msg_Id=0 标记为 delivered。 +- migration `20260714153000_fix_downstream_receipt_message_id` 新增 `SmsMessageRecord.cmppSubmitSequenceId`，并把存量 `deliveryType=receipt/status=delivered/ackMessageId=0` 纠正为 `unconfirmed`，保留 ACK 证据但不冒充业务确认。 +- 下游投递记录列表改为固定信息分组：投递类型/时间、企业/应用、消息 ID、状态、重试、最后错误和操作各自保持可读宽度，最后错误不再被挤成竖排。`delivered` 记录开放单条和批量手工重投，继续保留重复业务处理二次确认；`awaiting_ack` 仍禁止并发重投。 +- 已新增 TC-GW-ACK-005 和 Gateway 回归：首个返回包必须是 SubmitResp，随后失败回执 Deliver Msg_Id 与 SubmitResp 完全相同；另覆盖精确映射不回退、持久化 Sequence_Id 恢复和 Msg_Id=0 拒绝。API 全量 15 suites、154 项、Gateway 全量 Go 测试、Prisma validate、真实本地 PostgreSQL migration、API build、前端 build 和 `git diff --check` 均通过；前端仅有既有 Vite chunk size warning。经用户授权在应用内浏览器完成一次本地图形验证码登录，使用真实 NestJS API、PostgreSQL 和临时投递记录在 1440×1000 视口验证列表无横向挤压、无 `NaN`、console 无 error/warn，`delivered` 行的重投按钮可用且点击确实进入真实后端重投链路；因本地未运行 Gateway，请求按预期变为待重试而非伪造成功。临时投递记录和验收账号已清理。当前未提交、未推送、未部署生产。 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 9d47824..00e59a1 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -6,6 +6,7 @@ import ( "crypto/md5" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "log" @@ -75,6 +76,7 @@ type DownstreamReceipt struct { ReceiptStatus string `json:"receiptStatus"` RawStatus string `json:"rawStatus,omitempty"` ErrorCode string `json:"errorCode,omitempty"` + SubmitSequenceID uint32 `json:"submitSequenceId,omitempty"` DeliveredAt string `json:"deliveredAt,omitempty"` } @@ -220,7 +222,11 @@ func (s Server) handleLogin(response *cmpp.Response, packet *cmpp.Packet, logger } rememberAccount(session) go s.reportConnection(&session, "connected", "") - go s.flushPending(defaultString(auth.Account, account), logger) + response.AfterSend = func(sendErr error) { + if sendErr == nil { + go s.flushPending(defaultString(auth.Account, account), logger) + } + } logger.Printf( "cmpp inbound event=login_accepted protocol=%s requested_version=0x%02x response_version=0x%02x account=%s remote=%s", cmppVersionName(req.Version), uint8(req.Version), uint8(req.Version), account, packet.Conn.Conn.RemoteAddr(), @@ -319,11 +325,16 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge if current := findSessionByConn(packet.Conn); current != nil && current.report != nil { go current.report(current, "submit", "") } - go func() { - if _, err := s.flushPending(account, logger); err != nil { - logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error()) + response.AfterSend = func(sendErr error) { + if sendErr != nil { + return } - }() + go func() { + if _, err := s.flushPending(account, logger); err != nil { + logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error()) + } + }() + } logger.Printf( "cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s", clientProtocol, req.protocol, account, remote, req.sequenceID, phone, result.MessageID, gatewayMsgID, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, @@ -493,6 +504,7 @@ type pendingDelivery struct { ID string `json:"id"` DeliveryType string `json:"deliveryType"` Payload json.RawMessage `json:"payload"` + CreatedAt time.Time `json:"createdAt"` } type pendingFlushResult struct { @@ -546,7 +558,8 @@ func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (D } event.DeliveryID = delivery.ID event.Account = defaultString(event.Account, account) - return PushReceiptWithResult(event) + allowRecovery := !delivery.CreatedAt.IsZero() && time.Since(delivery.CreatedAt) >= 5*time.Second + return pushReceiptWithResult(event, allowRecovery) case "uplink": var event DownstreamUplink if err := json.Unmarshal(delivery.Payload, &event); err != nil { @@ -848,7 +861,14 @@ func PushReceipt(event DownstreamReceipt) (bool, error) { } func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error) { - session := findSession(event.MessageID, event.Account) + return pushReceiptWithResult(event, false) +} + +func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (DownstreamSendResult, error) { + session := findReceiptSession(event.MessageID, event.Account) + if session == nil && allowRecovery { + session = recoverReceiptSession(event) + } if session == nil { return DownstreamSendResult{}, nil } @@ -878,6 +898,34 @@ func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error return sendDownstream(session, deliver, event.DeliveryID) } +func findReceiptSession(messageID string, account string) *downstreamSession { + downstreamRegistry.RLock() + defer downstreamRegistry.RUnlock() + if messageID != "" { + return downstreamRegistry.byMessageID[messageID] + } + if account != "" { + return downstreamRegistry.byAccount[account] + } + return nil +} + +func recoverReceiptSession(event DownstreamReceipt) *downstreamSession { + if event.MessageID == "" || event.SubmitSequenceID == 0 || event.Account == "" { + return nil + } + downstreamRegistry.RLock() + accountSession := downstreamRegistry.byAccount[event.Account] + downstreamRegistry.RUnlock() + if accountSession == nil || accountSession.conn == nil { + return nil + } + recovered := *accountSession + recovered.messageID = event.MessageID + recovered.gatewayMsgID = messageIDFrom(event.MessageID, event.SubmitSequenceID) + return &recovered +} + func PushUplink(event DownstreamUplink) (bool, error) { result, err := PushUplinkWithResult(event) return result.Sent, err @@ -936,8 +984,11 @@ func findSession(messageID string, account string) *downstreamSession { func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID string) (DownstreamSendResult, error) { session.mu.Lock() defer session.mu.Unlock() - sequenceID := <-session.conn.SeqId messageID := downstreamDeliverMessageID(deliver) + if messageID == 0 { + return DownstreamSendResult{}, errors.New("refusing downstream CMPP_DELIVER with Msg_Id=0") + } + sequenceID := <-session.conn.SeqId sentAt := time.Now().UTC() ackDeadlineAt := sentAt.Add(downstreamAckTimeout()) tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt) diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index f520508..521b920 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -226,6 +226,82 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) { } } +func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + account := "100001" + password := "secret-hash" + var mu sync.Mutex + var submit submitRequest + pendingReturned := false + + 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": + mu.Lock() + defer mu.Unlock() + if err := json.NewDecoder(r.Body).Decode(&submit); err != nil { + t.Fatalf("decode submit: %v", err) + } + _ = json.NewEncoder(w).Encode(submitResponse{Accepted: true, MessageID: "MSG-ORDER"}) + case "/api/gateway/events/downstream/pending": + mu.Lock() + defer mu.Unlock() + if submit.SequenceID == 0 || pendingReturned { + _ = json.NewEncoder(w).Encode([]pendingDelivery{}) + return + } + payload, _ := json.Marshal(DownstreamReceipt{ + Account: account, MessageID: "MSG-ORDER", PhoneNumber: "13500002696", + ReceiptStatus: "undelivered", RawStatus: "REJECTD", SubmitSequenceID: submit.SequenceID, + }) + pendingReturned = true + _ = json.NewEncoder(w).Encode([]pendingDelivery{{ + ID: "delivery-order", DeliveryType: "receipt", Payload: payload, CreatedAt: time.Now().UTC(), + }}) + case "/api/gateway/events/inbound/connection", "/api/gateway/events/downstream/sent": + 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) + } + 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) + } + + first, err := client.RecvAndUnpackPkt(2 * time.Second) + if err != nil { + t.Fatalf("receive first packet: %v", err) + } + submitResponse, ok := first.(*cmpp.Cmpp2SubmitRspPkt) + if !ok || submitResponse.Result != 0 || submitResponse.MsgId == 0 { + t.Fatalf("first packet must be successful SUBMIT_RESP, got %T %+v", first, first) + } + deliver := recvDeliver20(t, client) + if deliver.MsgId != submitResponse.MsgId { + t.Fatalf("receipt Msg_Id=%d does not match SUBMIT_RESP Msg_Id=%d", deliver.MsgId, submitResponse.MsgId) + } +} + func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) { resetDownstreamRegistry() defer resetDownstreamRegistry() @@ -582,6 +658,40 @@ func TestDownstreamDeliveryRequiresAcknowledgement(t *testing.T) { } } +func TestReceiptLookupDoesNotFallbackToAccountBeforeSubmitMappingExists(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + + conn := &cmpp.Conn{} + rememberAccount(downstreamSession{ + account: "100001", protocol: "cmpp20", conn: conn, mu: &sync.Mutex{}, connectionID: "conn-1", + }) + if session := findReceiptSession("MSG-NOT-REMEMBERED", "100001"); session != nil { + t.Fatalf("receipt unexpectedly fell back to account session: %+v", session) + } + + recovered := recoverReceiptSession(DownstreamReceipt{ + MessageID: "MSG-NOT-REMEMBERED", Account: "100001", SubmitSequenceID: 1216579149, + }) + if recovered == nil { + t.Fatal("expected persisted submit sequence to recover receipt session") + } + if recovered.gatewayMsgID != messageIDFrom("MSG-NOT-REMEMBERED", 1216579149) || recovered.gatewayMsgID == 0 { + t.Fatalf("unexpected recovered Msg_Id: %d", recovered.gatewayMsgID) + } +} + +func TestSendDownstreamRejectsZeroMessageID(t *testing.T) { + _, err := sendDownstream( + &downstreamSession{mu: &sync.Mutex{}}, + &cmpp.Cmpp2DeliverReqPkt{MsgId: 0}, + "delivery-zero", + ) + if err == nil || !strings.Contains(err.Error(), "Msg_Id=0") { + t.Fatalf("expected zero Msg_Id rejection, got %v", err) + } +} + func TestDownstreamDeliveryReportsAckTimeout(t *testing.T) { resetDownstreamRegistry() defer resetDownstreamRegistry() diff --git a/gateway/third_party/gocmpp/server.go b/gateway/third_party/gocmpp/server.go index d8467bf..aa3d8d3 100644 --- a/gateway/third_party/gocmpp/server.go +++ b/gateway/third_party/gocmpp/server.go @@ -39,7 +39,8 @@ type Packet struct { type Response struct { *Packet Packer - SeqId uint32 + SeqId uint32 + AfterSend func(error) } type Handler interface { @@ -420,7 +421,11 @@ func (c *conn) serve() { } _, err = c.server.Handler.ServeCmpp(r, r.Packet, c.server.ErrorLog) - if err1 := c.finishPacket(r); err1 != nil { + err1 := c.finishPacket(r) + if r.AfterSend != nil { + r.AfterSend(err1) + } + 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, diff --git a/src/apps/admin/AdminDownstreamDeliveriesPage.tsx b/src/apps/admin/AdminDownstreamDeliveriesPage.tsx index f35f378..b9da470 100644 --- a/src/apps/admin/AdminDownstreamDeliveriesPage.tsx +++ b/src/apps/admin/AdminDownstreamDeliveriesPage.tsx @@ -1,7 +1,8 @@ import { useCallback, useEffect, useMemo, useState } from 'react'; import { AlertTriangle, BarChart3, CheckCircle2, Eye, RefreshCw, Search, TimerReset } from 'lucide-react'; import { adminApi, type DownstreamDeliveryDashboard, type DownstreamDeliveryRecord, type EnterpriseApplication } from '@/api/adminApi'; -import { Breadcrumb, Button, Input, Modal, Pagination, Select, Table, Tag, type TableColumn } from '@/components/ui'; +import { Breadcrumb, Button, Input, Modal, Pagination, Select, Tag } from '@/components/ui'; +import { formatDateTime } from '@/utils/dateTime'; const statusTone: Record = { pending: 'warning', @@ -113,9 +114,10 @@ export function AdminDownstreamDeliveriesPage() { loadData(); }, [loadData]); + const replayableStatuses = useMemo(() => new Set(['pending', 'delivered', 'failed', 'unconfirmed', 'rejected']), []); const selectableIds = useMemo( - () => records.filter((item) => ['pending', 'failed', 'unconfirmed', 'rejected'].includes(item.status)).map((item) => item.id), - [records], + () => records.filter((item) => replayableStatuses.has(item.status)).map((item) => item.id), + [records, replayableStatuses], ); const allSelected = selectableIds.length > 0 && selectableIds.every((id) => selectedIds.includes(id)); const summary = dashboard?.summary; @@ -123,65 +125,16 @@ export function AdminDownstreamDeliveriesPage() { const retryBuckets = dashboard?.retryBuckets ?? []; const topApplications = dashboard?.topApplications ?? []; - const columns: Array> = [ - { - key: 'select', - title: '选择', - width: '52px', - align: 'center', - render: (record) => ( - { - setSelectedIds((current) => - event.target.checked - ? [...current, record.id] - : current.filter((item) => item !== record.id), - ); - }} - aria-label={`选择${record.id}`} - /> - ), - }, - { key: 'createdAt', title: '投递时间', width: '180px', render: (record) => record.createdAt }, - { key: 'tenant', title: '企业', width: '180px', render: (record) => record.tenant?.name ?? '-' }, - { key: 'application', title: '应用', width: '180px', render: (record) => record.application?.name ?? '-' }, - { key: 'type', title: '类型', width: '110px', render: (record) => deliveryTypeLabel[record.deliveryType] ?? record.deliveryType }, - { key: 'messageId', title: '消息 ID', width: '180px', render: (record) => {record.messageId ?? '-'} }, - { key: 'status', title: '状态', width: '150px', render: (record) => {statusLabel[record.status] ?? record.status} }, - { key: 'retry', title: '重试', width: '90px', align: 'center', render: (record) => record.retryCount }, - { key: 'error', title: '最后错误', render: (record) => record.lastError ?? '-' }, - { - key: 'actions', - title: '操作', - width: '170px', - align: 'right', - render: (record) => ( -
- - -
- ), - }, - ]; - const totalPages = Math.max(1, Math.ceil(total / pageSize)); + const replayRecord = (record: DownstreamDeliveryRecord) => { + const acknowledgedWarning = record.status === 'delivered' ? '该记录已收到客户端确认，' : ''; + if (!window.confirm(`${acknowledgedWarning}重投可能导致下游业务重复处理，确认继续吗？`)) return; + adminApi.requeueDownstreamDelivery(record.id) + .then(() => loadData()) + .catch((failure: Error) => setError(failure.message || '人工重投失败')); + }; + return (
@@ -306,12 +259,12 @@ export function AdminDownstreamDeliveriesPage() { {typeBreakdown.map((item) => (
{deliveryTypeLabel[item.deliveryType] ?? item.deliveryType} - {item.total} - {item.pending} - {item.awaitingAck} - {item.delivered} - {item.unconfirmed + item.rejected} - {item.failed} + {item.total ?? 0} + {item.pending ?? 0} + {item.awaitingAck ?? 0} + {item.delivered ?? 0} + {(item.unconfirmed ?? 0) + (item.rejected ?? 0)} + {item.failed ?? 0}
))}
@@ -346,10 +299,10 @@ export function AdminDownstreamDeliveriesPage() { {topApplications.length > 0 ? topApplications.map((item) => (
{item.name} - {item.pending} - {item.failed + item.unconfirmed + item.rejected} - {item.delivered} - {item.alertCount} + {item.pending ?? 0} + {(item.failed ?? 0) + (item.unconfirmed ?? 0) + (item.rejected ?? 0)} + {item.delivered ?? 0} + {item.alertCount ?? 0}
)) : (
暂无应用告警数据
@@ -358,11 +311,11 @@ export function AdminDownstreamDeliveriesPage() {
-
-

- 已选择 {selectedIds.length} 条，可对未确认或失败记录执行批量重投 +

+

+ 已选择 {selectedIds.length} 条；客户端已确认的记录也允许人工重投，请注意重复处理风险。

-
+
- +
+
+ 选择 + 投递信息 + 企业 / 应用 + 消息 ID + 状态 + 重试 + 最后错误 + 操作 +
+ {records.length > 0 ? records.map((record) => ( +
+
+ { + setSelectedIds((current) => event.target.checked + ? [...current, record.id] + : current.filter((item) => item !== record.id)); + }} + aria-label={`选择${record.id}`} + /> +
+
+ {deliveryTypeLabel[record.deliveryType] ?? record.deliveryType} + {formatDateTime(record.createdAt)} +
+
+ {record.tenant?.name ?? '-'} + {record.application?.name ?? '-'} +
+
+ {record.messageId ?? '-'} +
+
+ {statusLabel[record.status] ?? record.status} +
+
+ {record.retryCount} + 次 +
+
+ {record.lastError ?? '无'} +
+
+ + +
+
+ )) : ( +
{loading ? '加载中...' : '暂无下游投递记录'}
+ )} +
div, +.downstream-delivery-list__actions { + align-items: center; + display: flex; + flex-shrink: 0; + gap: var(--space-2); +} + +.downstream-delivery-list { + overflow-x: auto; +} + +.downstream-delivery-list__header, +.downstream-delivery-list__row { + align-items: center; + display: grid; + gap: var(--space-3); + grid-template-columns: 36px 140px minmax(150px, 1fr) minmax(165px, 1.05fr) 126px 54px minmax(170px, 1.2fr) 136px; + min-width: 1080px; + padding: var(--space-4); +} + +.downstream-delivery-list__header { + background: var(--color-bg-subtle); + color: var(--color-text-muted); + font-size: var(--font-size-sm); + font-weight: var(--font-weight-semibold); +} + +.downstream-delivery-list__row { + border-top: 1px solid var(--color-border); + min-height: 104px; +} + +.downstream-delivery-list__row:first-of-type { + border-top: 0; +} + +.downstream-delivery-list__row:hover { + background: color-mix(in srgb, var(--color-selected-soft) 28%, transparent); +} + +.downstream-delivery-list__select, +.downstream-delivery-list__retry { + text-align: center; +} + +.downstream-delivery-list__meta, +.downstream-delivery-list__owner, +.downstream-delivery-list__retry { + display: grid; + gap: 6px; +} + +.downstream-delivery-list__meta strong, +.downstream-delivery-list__owner strong, +.downstream-delivery-list__retry strong { + color: var(--color-text-strong); +} + +.downstream-delivery-list__meta span, +.downstream-delivery-list__owner span, +.downstream-delivery-list__retry span { + color: var(--color-text-muted); + font-size: var(--font-size-sm); +} + +.downstream-delivery-list__message strong { + color: var(--color-text-strong); + font-family: ui-monospace, SFMono-Regular, Consolas, monospace; + font-size: var(--font-size-sm); + overflow-wrap: anywhere; +} + +.downstream-delivery-list__error { + background: #fff7ed; + border: 1px solid #fed7aa; + border-radius: var(--radius-md); + color: #9a3412; + display: -webkit-box; + line-height: 1.55; + overflow: hidden; + padding: var(--space-3); + -webkit-box-orient: vertical; + -webkit-line-clamp: 2; +} + +.downstream-delivery-list__error.is-empty { + background: var(--color-bg-subtle); + border-color: var(--color-border); + color: var(--color-text-subtle); +} + +.downstream-delivery-list__actions { + justify-content: flex-end; +} + +.downstream-delivery-list__empty { + color: var(--color-text-muted); + padding: 48px var(--space-5); + text-align: center; +} + +@media (max-width: 760px) { + .downstream-delivery-toolbar { + align-items: stretch; + flex-direction: column; + } + + .downstream-delivery-toolbar > div { + justify-content: flex-end; + } +} + .admin-task-enterprise, .admin-task-counts, .admin-task-send-type {