hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: handle cmpp2 deliver receipts

Browse Source
This commit is contained in:
hectorzhao
2026-07-09 16:14:11 +08:00
parent 20757203d9
commit dc7d015219
3 changed files with 146 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/testing-progress.md
+1
View File
@@ -8,6 +8,7 @@
- 已执行：`npm --prefix api test -- channels.service.spec.ts --runInBand`、`npm --prefix api run build`、`npm run build`。待生产部署后用指定号码做一次真实发送验证，并回查 DB/短信记录。
- 2026-07-09 追加：按产品边界收窄通道测试短信，`SmsMessageRecord/SmsSubmitRecord/SmsReceiptRecord/SmsMessageSegmentAudit` 支持 `tenantId/batchTaskId` 为空；通道测试只记录短信、提交结果和回执，不进入企业账务、发送任务进度、客户侧 Deliver 推送或业务补发。
- 2026-07-09 追加：生产验证短信进入 Gateway 后出现 `SUBMIT_TIMEOUT/context deadline exceeded`，上游平台疑似内容乱码；排查确认 API、数据库和 Redis Stream 中中文内容正常，`SubmitCommand.cmpp.msgFmt=8`，根因为 Gateway 在 CMPP 2.0 通道上仍固定构造 `Cmpp3SubmitReqPkt`。已修复为按通道 `cmppVersion` 分别构造 `Cmpp2SubmitReqPkt`/`Cmpp3SubmitReqPkt`，并同时处理 CMPP 2.0/3.0 SubmitResp。新增 `submit_packet_test.go` 覆盖 CMPP 2.0 中文 UCS2 Submit 包类型和长度。
- 2026-07-09 追加：生产验证 Submit 已 accepted 后未见 `SmsReceiptRecord`，结合上游为 CMPP 2.0 排查 Gateway readLoop，确认只处理 `Cmpp3DeliverReqPkt`，CMPP 2.0 回执 Deliver 即使到达连接也不会 ACK 或进入 receipt 事件链路。已修复为同时处理 `Cmpp2DeliverReqPkt`/`Cmpp3DeliverReqPkt`，分别返回 `Cmpp2DeliverRspPkt`/`Cmpp3DeliverRspPkt`，并统一 Deliver 解码、回执和上行处理。新增 `deliver_test.go` 覆盖 CMPP 2.0 `DELIVRD` 回执写入事件链路。
## 2026-07-09 通道复制默认停用与真实连接池状态回写修复
gateway/internal/upstream/deliver_test.go
+88
View File
@@ -0,0 +1,88 @@
package upstream
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"cmpp-platform/gateway/internal/queue"
cmpp "github.com/bigwhite/gocmpp"
)
func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) {
events := make(chan queue.ReceiptEvent, 1)
api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/gateway/events/receipt" {
t.Fatalf("unexpected path: %s", r.URL.Path)
}
var event queue.ReceiptEvent
if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
t.Fatalf("decode receipt event: %v", err)
}
events <- event
w.WriteHeader(http.StatusOK)
}))
defer api.Close()
receipt := cmpp.CmppReceiptPkt{
MsgId: 8412634832294102675,
Stat: "DELIVRD",
SubmitTime: "2607091558",
DoneTime: "2607091559",
DestTerminalId: "18821203795",
SmscSequence: 1,
}
payload, err := receipt.Pack()
if err != nil {
t.Fatalf("pack receipt: %v", err)
}
conn := &connection{
channelID: "channel-1",
apiBaseURL: api.URL,
httpClient: api.Client(),
tracker: map[uint64]queue.SubmitCommand{
receipt.MsgId: {
Envelope: queue.Envelope{
SchemaVersion: queue.SchemaVersion,
MessageType: queue.MessageTypeSubmitCommand,
TraceID: "trace-1",
MessageID: "message-1",
ChannelID: "channel-1",
CreatedAt: time.Now(),
},
SubmitID: "submit-1",
},
},
}
conn.handleDeliver(deliverPacketFromCMPP2(&cmpp.Cmpp2DeliverReqPkt{
SeqId: 7,
MsgId: 999,
RegisterDelivery: 1,
MsgContent: string(payload),
}))
select {
case event := <-events:
if event.MessageID != "message-1" || event.ChannelID != "channel-1" || event.TraceID != "trace-1" {
t.Fatalf("unexpected envelope: %+v", event.Envelope)
}
if event.SequenceID != 7 {
t.Fatalf("SequenceID = %d, want 7", event.SequenceID)
}
if event.GatewayMessageID != "8412634832294102675" {
t.Fatalf("GatewayMessageID = %q", event.GatewayMessageID)
}
if event.PhoneNumber != "18821203795" {
t.Fatalf("PhoneNumber = %q", event.PhoneNumber)
}
if event.ReceiptStatus != "delivered" || event.RawStatus != "DELIVRD" {
t.Fatalf("unexpected receipt status: %+v", event)
}
case <-time.After(time.Second):
t.Fatal("timed out waiting for receipt event")
}
}
gateway/internal/upstream/manager.go
+57 -18
View File
@@ -605,25 +605,64 @@ func (c *connection) readLoop() {
if ch != nil {
ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result}
}
case *cmpp.Cmpp2DeliverReqPkt:
_ = c.client.SendRspPkt(&cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
c.handleDeliver(deliverPacketFromCMPP2(p))
case *cmpp.Cmpp3DeliverReqPkt:
c.handleDeliver(p)
_ = c.client.SendRspPkt(&cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
c.handleDeliver(deliverPacketFromCMPP3(p))
case *cmpp.CmppActiveTestReqPkt:
_ = c.client.SendRspPkt(&cmpp.CmppActiveTestRspPkt{}, p.SeqId)
}
}
}
func (c *connection) handleDeliver(pkt *cmpp.Cmpp3DeliverReqPkt) {
_ = c.client.SendRspPkt(&cmpp.Cmpp3DeliverRspPkt{MsgId: pkt.MsgId, Result: 0}, pkt.SeqId)
type deliverPacket struct {
seqID uint32
msgID uint64
destID string
tpUdhi uint8
msgFmt uint8
srcTerminalID string
registerDelivery uint8
msgContent string
}
if pkt.RegisterDelivery == 1 {
func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket {
return deliverPacket{
seqID: pkt.SeqId,
msgID: pkt.MsgId,
destID: pkt.DestId,
tpUdhi: pkt.TpUdhi,
msgFmt: pkt.MsgFmt,
srcTerminalID: pkt.SrcTerminalId,
registerDelivery: pkt.RegisterDelivery,
msgContent: pkt.MsgContent,
}
}
func deliverPacketFromCMPP3(pkt *cmpp.Cmpp3DeliverReqPkt) deliverPacket {
return deliverPacket{
seqID: pkt.SeqId,
msgID: pkt.MsgId,
destID: pkt.DestId,
tpUdhi: pkt.TpUdhi,
msgFmt: pkt.MsgFmt,
srcTerminalID: pkt.SrcTerminalId,
registerDelivery: pkt.RegisterDelivery,
msgContent: pkt.MsgContent,
}
}
func (c *connection) handleDeliver(pkt deliverPacket) {
if pkt.registerDelivery == 1 {
var receipt cmpp.CmppReceiptPkt
if err := receipt.Unpack([]byte(pkt.MsgContent)); err != nil {
if err := receipt.Unpack([]byte(pkt.msgContent)); err != nil {
return
}
cmd, ok := c.commandFor(receipt.MsgId)
if !ok {
cmd, ok = c.commandFor(pkt.MsgId)
cmd, ok = c.commandFor(pkt.msgID)
}
traceID := fmt.Sprintf("receipt-%d", receipt.MsgId)
messageID := fmt.Sprintf("receipt-%d", receipt.MsgId)
@@ -642,7 +681,7 @@ func (c *connection) handleDeliver(pkt *cmpp.Cmpp3DeliverReqPkt) {
ChannelID: channelID,
CreatedAt: time.Now().UTC(),
},
SequenceID: pkt.SeqId,
SequenceID: pkt.seqID,
GatewayMessageID: fmt.Sprint(receipt.MsgId),
PhoneNumber: strings.TrimSpace(receipt.DestTerminalId),
ReceiptStatus: receiptStatus(receipt.Stat),
@@ -660,7 +699,7 @@ func (c *connection) handleDeliver(pkt *cmpp.Cmpp3DeliverReqPkt) {
if !complete {
return
}
cmd, _ := c.commandFor(pkt.MsgId)
cmd, _ := c.commandFor(pkt.msgID)
event := queue.UplinkEvent{
Envelope: queue.Envelope{
SchemaVersion: queue.SchemaVersion,
@@ -670,32 +709,32 @@ func (c *connection) handleDeliver(pkt *cmpp.Cmpp3DeliverReqPkt) {
ChannelID: c.channelID,
CreatedAt: time.Now().UTC(),
},
SequenceID: pkt.SeqId,
PhoneNumber: strings.TrimSpace(pkt.SrcTerminalId),
DestID: strings.TrimSpace(pkt.DestId),
SequenceID: pkt.seqID,
PhoneNumber: strings.TrimSpace(pkt.srcTerminalID),
DestID: strings.TrimSpace(pkt.destID),
Content: content,
ReceivedAt: time.Now().UTC(),
}
_ = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/uplink", event)
}
func (c *connection) decodeUplinkContent(pkt *cmpp.Cmpp3DeliverReqPkt) (string, bool, error) {
if pkt.TpUdhi != 1 {
content, err := decodeContent(pkt.MsgFmt, pkt.MsgContent)
func (c *connection) decodeUplinkContent(pkt deliverPacket) (string, bool, error) {
if pkt.tpUdhi != 1 {
content, err := decodeContent(pkt.msgFmt, pkt.msgContent)
return content, true, err
}
ref, total, number, payload, ok := parseConcatSegment(pkt.MsgContent)
ref, total, number, payload, ok := parseConcatSegment(pkt.msgContent)
if !ok {
content, err := decodeContent(pkt.MsgFmt, pkt.MsgContent)
content, err := decodeContent(pkt.msgFmt, pkt.msgContent)
return content, true, err
}
key := fmt.Sprintf("%s:%s:%s:%d:%d", c.channelID, strings.TrimSpace(pkt.SrcTerminalId), strings.TrimSpace(pkt.DestId), ref, total)
key := fmt.Sprintf("%s:%s:%s:%d:%d", c.channelID, strings.TrimSpace(pkt.srcTerminalID), strings.TrimSpace(pkt.destID), ref, total)
c.mu.Lock()
if c.longUplink == nil {
c.longUplink = make(map[string]*longUplinkAssembly)
}
pruneLongUplinkAssemblies(c.longUplink, time.Now(), 10*time.Minute)
content, complete, err := assembleLongUplink(c.longUplink, key, pkt.MsgFmt, total, number, payload)
content, complete, err := assembleLongUplink(c.longUplink, key, pkt.msgFmt, total, number, payload)
c.mu.Unlock()
return content, complete, err
}