fix: 修复 CMPP 协议字段容量与版本兼容性

gateway/internal/inbound/acknowledgement.go

@@ -80,6 +80,10 @@ func registerDownstreamAck(session *downstreamSession, deliveryID string, claimI
 	}
 	key := downstreamAckKey(session.conn, sequenceID)
 	downstreamAckRegistry.Lock()
+	if downstreamAckRegistry.items[key] != nil {
+		downstreamAckRegistry.Unlock()
+		return nil
+	}
 	downstreamAckRegistry.items[key] = tracker
 	tracker.timer = time.AfterFunc(time.Until(deadline), func() {
 		timedOut := takeDownstreamAck(session.conn, sequenceID)

gateway/internal/inbound/delivery.go

@@ -15,19 +15,19 @@ import (
 // arbitrary account session would acknowledge a message with the wrong Msg_Id.
 
 type DownstreamReceipt struct {
-	DeliveryID           string `json:"deliveryId,omitempty"`
-	ClaimID              string `json:"claimId,omitempty"`
-	Account              string `json:"account,omitempty"`
-	ApplicationID        string `json:"applicationId,omitempty"`
-	MessageID            string `json:"messageId"`
-	GatewayMessageID     string `json:"gatewayMessageId,omitempty"`
-	PhoneNumber          string `json:"phoneNumber,omitempty"`
-	ReceiptStatus        string `json:"receiptStatus"`
-	RawStatus            string `json:"rawStatus,omitempty"`
-	ErrorCode            string `json:"errorCode,omitempty"`
-	SubmitSequenceID     uint32 `json:"submitSequenceId,omitempty"`
-	SubmitGroupMessageID string `json:"submitGroupMessageId,omitempty"`
-	DeliveredAt          string `json:"deliveredAt,omitempty"`
+	DeliveryID           string  `json:"deliveryId,omitempty"`
+	ClaimID              string  `json:"claimId,omitempty"`
+	Account              string  `json:"account,omitempty"`
+	ApplicationID        string  `json:"applicationId,omitempty"`
+	MessageID            string  `json:"messageId"`
+	GatewayMessageID     string  `json:"gatewayMessageId,omitempty"`
+	PhoneNumber          string  `json:"phoneNumber,omitempty"`
+	ReceiptStatus        string  `json:"receiptStatus"`
+	RawStatus            string  `json:"rawStatus,omitempty"`
+	ErrorCode            string  `json:"errorCode,omitempty"`
+	SubmitSequenceID     *uint32 `json:"submitSequenceId,omitempty"`
+	SubmitGroupMessageID string  `json:"submitGroupMessageId,omitempty"`
+	DeliveredAt          string  `json:"deliveredAt,omitempty"`
 }
 
 type DownstreamUplink struct {
@@ -108,7 +108,7 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
 		session = recoverReceiptSession(event)
 	}
 	if session == nil {
-		if event.SubmitSequenceID == 0 {
+		if event.SubmitSequenceID == nil {
 			return DownstreamSendResult{
 				Retryable:    false,
 				ReasonCode:   "MISSING_SUBMIT_SEQUENCE_ID",
@@ -158,7 +158,11 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
 		DestTerminalId: defaultString(event.PhoneNumber, session.phoneNumber),
 		SmscSequence:   uint32(time.Now().UnixNano() & 0xffffffff),
 	}
-	receiptBytes, err := receipt.Pack()
+	version := cmpp.V30
+	if session.protocol == "cmpp20" || session.protocol == "cmpp21" {
+		version = cmpp.V20
+	}
+	receiptBytes, err := receipt.PackVersion(version)
 	if err != nil {
 		return DownstreamSendResult{}, err
 	}
@@ -167,8 +171,8 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
 }
 
 func downstreamReceiptMessageID(event DownstreamReceipt, session *downstreamSession) uint64 {
-	if event.SubmitSequenceID != 0 {
-		return messageIDFrom(defaultString(event.SubmitGroupMessageID, event.MessageID), event.SubmitSequenceID)
+	if event.SubmitSequenceID != nil {
+		return messageIDFrom(defaultString(event.SubmitGroupMessageID, event.MessageID), *event.SubmitSequenceID)
 	}
 	if session == nil {
 		return 0
@@ -189,7 +193,7 @@ func findReceiptSession(messageID string, account string) *downstreamSession {
 }
 
 func recoverReceiptSession(event DownstreamReceipt) *downstreamSession {
-	if event.MessageID == "" || event.SubmitSequenceID == 0 || event.Account == "" {
+	if event.MessageID == "" || event.SubmitSequenceID == nil || event.Account == "" {
 		return nil
 	}
 	downstreamRegistry.RLock()
@@ -200,7 +204,7 @@ func recoverReceiptSession(event DownstreamReceipt) *downstreamSession {
 	}
 	recovered := *accountSession
 	recovered.messageID = event.MessageID
-	recovered.gatewayMsgID = messageIDFrom(defaultString(event.SubmitGroupMessageID, event.MessageID), event.SubmitSequenceID)
+	recovered.gatewayMsgID = messageIDFrom(defaultString(event.SubmitGroupMessageID, event.MessageID), *event.SubmitSequenceID)
 	return &recovered
 }
 
@@ -293,6 +297,12 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID
 		AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
 	}
 	tracker := registerDownstreamAck(session, deliveryID, claimID, sequenceID, messageID, ackDeadlineAt)
+	// A wrapped sequence cannot overwrite another unacknowledged delivery.
+	if deliveryID != "" && tracker == nil {
+		result.Retryable = true
+		result.ReasonCode = "SEQUENCE_IN_USE"
+		return result, nil
+	}
 	if err := session.conn.SendPkt(deliver, sequenceID); err != nil {
 		removeDownstreamAck(tracker)
 		session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "failed", "SEND_FAILED", err)

gateway/internal/inbound/protocol_fields_test.go

@@ -0,0 +1,146 @@
+package inbound
+
+import (
+	"encoding/json"
+	cmpp "github.com/bigwhite/gocmpp"
+	"net"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestZeroSequenceReceiptRecoversAndAcknowledgesOverTCP(t *testing.T) {
+	for _, version := range []cmpp.Type{cmpp.V20, cmpp.V30} {
+		t.Run(version.String(), func(t *testing.T) {
+			resetDownstreamRegistry()
+			defer resetDownstreamRegistry()
+			listener, err := net.Listen("tcp", "127.0.0.1:0")
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer listener.Close()
+			tcp, err := net.Dial("tcp", listener.Addr().String())
+			if err != nil {
+				t.Fatal(err)
+			}
+			remote, err := listener.Accept()
+			if err != nil {
+				t.Fatal(err)
+			}
+			server, client := cmpp.NewConn(remote, version), cmpp.NewConn(tcp, version)
+			server.SetState(cmpp.CONN_AUTHOK)
+			client.SetState(cmpp.CONN_AUTHOK)
+			defer server.Close()
+			defer client.Close()
+			acknowledged := make(chan downstreamDeliveryLifecycleEvent, 2)
+			session := &downstreamSession{account: "qa", conn: server, protocol: version.String(), connectionID: "reconnected", mu: &sync.Mutex{}, deliveryReport: func(e downstreamDeliveryLifecycleEvent) {
+				if e.Kind == "acknowledged" {
+					acknowledged <- e
+				}
+			}}
+			// Simulate a fresh connection with no in-memory original Submit mapping.
+			downstreamRegistry.Lock()
+			downstreamRegistry.byAccount["qa"] = session
+			downstreamRegistry.Unlock()
+			var event DownstreamReceipt
+			if err = json.Unmarshal([]byte(`{"deliveryId":"qa-zero","account":"qa","messageId":"original","submitSequenceId":0,"receiptStatus":"delivered","phoneNumber":"13800138000"}`), &event); err != nil {
+				t.Fatal(err)
+			}
+			result, err := PushReceiptWithResult(event)
+			if err != nil || !result.Sent {
+				t.Fatalf("zero cannot recover: %+v %v", result, err)
+			}
+			p, err := client.RecvAndUnpackPkt(time.Second)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var content string
+			var seq uint32
+			var msgID uint64
+			switch pkt := p.(type) {
+			case *cmpp.Cmpp2DeliverReqPkt:
+				content, seq, msgID = pkt.MsgContent, pkt.SeqId, pkt.MsgId
+			case *cmpp.Cmpp3DeliverReqPkt:
+				content, seq, msgID = pkt.MsgContent, pkt.SeqId, pkt.MsgId
+			default:
+				t.Fatalf("bad packet %T", p)
+			}
+			var receipt cmpp.CmppReceiptPkt
+			if err = receipt.UnpackVersion([]byte(content), version); err != nil {
+				t.Fatal(err)
+			}
+			if receipt.MsgId != messageIDFrom("original", 0) || msgID != receipt.MsgId {
+				t.Fatalf("wrong original Msg_Id: %d", receipt.MsgId)
+			}
+			var response cmpp.Packer = &cmpp.Cmpp3DeliverRspPkt{MsgId: msgID, Result: 0}
+			if version == cmpp.V20 {
+				response = &cmpp.Cmpp2DeliverRspPkt{MsgId: msgID, Result: 0}
+			}
+			if err = client.SendPkt(response, seq); err != nil {
+				t.Fatal(err)
+			}
+			p, err = server.RecvAndUnpackPkt(time.Second)
+			if err != nil {
+				t.Fatal(err)
+			}
+			switch pkt := p.(type) {
+			case *cmpp.Cmpp2DeliverRspPkt:
+				handleDownstreamAcknowledgement(server, pkt.SeqId, pkt.MsgId, uint32(pkt.Result), nil)
+			case *cmpp.Cmpp3DeliverRspPkt:
+				handleDownstreamAcknowledgement(server, pkt.SeqId, pkt.MsgId, pkt.Result, nil)
+			}
+			select {
+			case ack := <-acknowledged:
+				if ack.Result != 0 || ack.MessageID != receipt.MsgId {
+					t.Fatal("wrong ACK")
+				}
+			case <-time.After(time.Second):
+				t.Fatal("missing ACK")
+			}
+			raw, err := json.Marshal(submitRequest{SequenceID: 0})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var fields map[string]any
+			json.Unmarshal(raw, &fields)
+			if value, ok := fields["sequenceId"]; !ok || value != float64(0) {
+				t.Fatal("zero omitted from Submit callback")
+			}
+		})
+	}
+}
+
+func TestDownstreamAckCollisionDoesNotReplacePendingDelivery(t *testing.T) {
+	conn := &cmpp.Conn{}
+	session := &downstreamSession{conn: conn}
+	for _, sequence := range []uint32{^uint32(0), 0} {
+		first := registerDownstreamAck(session, "first", "a", sequence, 1, time.Now().Add(time.Minute))
+		if first == nil {
+			t.Fatal("first registration failed")
+		}
+		if registerDownstreamAck(session, "second", "b", sequence, 2, time.Now().Add(time.Minute)) != nil {
+			t.Fatal("overwrote pending delivery")
+		}
+		if takeDownstreamAck(conn, sequence) != first {
+			t.Fatal("lost original delivery")
+		}
+	}
+}
+
+func TestAbsentAndZeroSubmitSequenceRemainDistinct(t *testing.T) {
+	resetDownstreamRegistry()
+	defer resetDownstreamRegistry()
+	for _, payload := range []string{`{}`, `{"submitSequenceId":null}`, `{"submitSequenceId":0}`} {
+		event := DownstreamReceipt{Account: "qa", MessageID: "original"}
+		if err := json.Unmarshal([]byte(payload), &event); err != nil {
+			t.Fatal(err)
+		}
+		result, err := PushReceiptWithResult(event)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if result.Retryable != (event.SubmitSequenceID != nil) {
+			t.Fatalf("missing confused with zero: %s %+v", payload, result)
+		}
+	}
+}

gateway/internal/inbound/server_test.go

@@ -212,7 +212,7 @@ func TestInboundServerAuthenticatesAndSubmits(t *testing.T) {
 		t.Fatalf("expected receipt deliver, got %+v", deliver)
 	}
 	var receipt cmpp.CmppReceiptPkt
-	if err := receipt.Unpack([]byte(deliver.MsgContent)); err != nil {
+	if err := receipt.UnpackVersion([]byte(deliver.MsgContent), cmpp.V30); err != nil {
 		t.Fatalf("unpack pushed receipt: %v", err)
 	}
 	if receipt.Stat != "DELIVRD" || receipt.DestTerminalId != "13600002696" || receipt.MsgId != rsp.MsgId {
@@ -535,7 +535,7 @@ func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) {
 			}
 			payload, _ := json.Marshal(DownstreamReceipt{
 				Account: account, MessageID: "MSG-ORDER", PhoneNumber: "13500002696",
-				ReceiptStatus: "undelivered", RawStatus: "REJECTD", SubmitSequenceID: submit.SequenceID,
+				ReceiptStatus: "undelivered", RawStatus: "REJECTD", SubmitSequenceID: sequencePointer(submit.SequenceID),
 			})
 			pendingReturned = true
 			_ = json.NewEncoder(w).Encode([]pendingDelivery{{
@@ -667,7 +667,7 @@ func TestRecoverableReceiptWaitsForClientConnection(t *testing.T) {
 		DeliveryID:       "delivery-retry",
 		Account:          "100001",
 		MessageID:        "MSG-RETRY",
-		SubmitSequenceID: 77,
+		SubmitSequenceID: sequencePointer(77),
 		ReceiptStatus:    "delivered",
 	})
 	if err != nil {
@@ -1161,7 +1161,7 @@ func TestReceiptLookupDoesNotFallbackToAccountBeforeSubmitMappingExists(t *testi
 	}
 
 	recovered := recoverReceiptSession(DownstreamReceipt{
-		MessageID: "MSG-NOT-REMEMBERED", Account: "100001", SubmitSequenceID: 1216579149,
+		MessageID: "MSG-NOT-REMEMBERED", Account: "100001", SubmitSequenceID: sequencePointer(1216579149),
 	})
 	if recovered == nil {
 		t.Fatal("expected persisted submit sequence to recover receipt session")
@@ -1170,10 +1170,10 @@ func TestReceiptLookupDoesNotFallbackToAccountBeforeSubmitMappingExists(t *testi
 		t.Fatalf("unexpected recovered Msg_Id: %d", recovered.gatewayMsgID)
 	}
 	first := recoverReceiptSession(DownstreamReceipt{
-		MessageID: "MSG-FIRST", SubmitGroupMessageID: "MSG-GROUP", Account: "100001", SubmitSequenceID: 77,
+		MessageID: "MSG-FIRST", SubmitGroupMessageID: "MSG-GROUP", Account: "100001", SubmitSequenceID: sequencePointer(77),
 	})
 	second := recoverReceiptSession(DownstreamReceipt{
-		MessageID: "MSG-SECOND", SubmitGroupMessageID: "MSG-GROUP", Account: "100001", SubmitSequenceID: 77,
+		MessageID: "MSG-SECOND", SubmitGroupMessageID: "MSG-GROUP", Account: "100001", SubmitSequenceID: sequencePointer(77),
 	})
 	if first == nil || second == nil || first.gatewayMsgID != second.gatewayMsgID || first.gatewayMsgID != messageIDFrom("MSG-GROUP", 77) {
 		t.Fatalf("multi-destination recovery did not preserve the original Msg_Id: first=%+v second=%+v", first, second)
@@ -1183,10 +1183,10 @@ func TestReceiptLookupDoesNotFallbackToAccountBeforeSubmitMappingExists(t *testi
 func TestLongMessageReceiptsUseEachOriginalFragmentMsgID(t *testing.T) {
 	session := &downstreamSession{gatewayMsgID: messageIDFrom("MSG-GROUP", 101)}
 	first := downstreamReceiptMessageID(DownstreamReceipt{
-		MessageID: "MSG-CHILD", SubmitGroupMessageID: "MSG-GROUP", SubmitSequenceID: 101,
+		MessageID: "MSG-CHILD", SubmitGroupMessageID: "MSG-GROUP", SubmitSequenceID: sequencePointer(101),
 	}, session)
 	second := downstreamReceiptMessageID(DownstreamReceipt{
-		MessageID: "MSG-CHILD", SubmitGroupMessageID: "MSG-GROUP", SubmitSequenceID: 102,
+		MessageID: "MSG-CHILD", SubmitGroupMessageID: "MSG-GROUP", SubmitSequenceID: sequencePointer(102),
 	}, session)
 	if first != messageIDFrom("MSG-GROUP", 101) || second != messageIDFrom("MSG-GROUP", 102) {
 		t.Fatalf("fragment receipt Msg_Id mismatch: first=%d second=%d", first, second)
@@ -1324,3 +1324,5 @@ func recvSubmitRsp20(t *testing.T, client *cmpp.Client) *cmpp.Cmpp2SubmitRspPkt
 	t.Fatal("timed out waiting CMPP2 submit response")
 	return nil
 }
+
+func sequencePointer(n uint32) *uint32 { return &n }

gateway/internal/inbound/submit.go

@@ -27,7 +27,7 @@ type submitRequest struct {
 	Content            string                      `json:"content"`
 	SrcID              string                      `json:"srcId,omitempty"`
 	DestID             string                      `json:"destId,omitempty"`
-	SequenceID         uint32                      `json:"sequenceId,omitempty"`
+	SequenceID         uint32                      `json:"sequenceId"`
 	RegisteredDelivery uint8                       `json:"registeredDelivery"`
 	RemoteIP           string                      `json:"remoteIp,omitempty"`
 	LongMessage        *inboundLongMessageFragment `json:"longMessage,omitempty"`

gateway/internal/upstream/connection.go

@@ -16,6 +16,13 @@ import (
 // The reader and heartbeat goroutines share one connection lifecycle. Closing
 // the connection must wake pending submitters before scheduling pool recovery.
 
+// Caller holds c.mu across sequence selection, wire write and pending registration.
+func (c *connection) sequenceAvailable(sequence uint32) bool {
+	_, submit := c.pending[sequence]
+	_, heartbeat := c.heartbeatPending[sequence]
+	return !submit && !heartbeat
+}
+
 type connection struct {
 	channelID            string
 	config               queue.UpstreamConfig

gateway/internal/upstream/deliver.go

@@ -15,6 +15,7 @@ import (
 // mobile-originated content uses a separate long-message assembly path.
 
 type deliverPacket struct {
+	version          cmpp.Type
 	seqID            uint32
 	msgID            uint64
 	destID           string
@@ -26,7 +27,7 @@ type deliverPacket struct {
 }
 
 func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket {
-	return deliverPacket{
+	return deliverPacket{version: cmpp.V20,
 		seqID:            pkt.SeqId,
 		msgID:            pkt.MsgId,
 		destID:           pkt.DestId,
@@ -39,7 +40,7 @@ func deliverPacketFromCMPP2(pkt *cmpp.Cmpp2DeliverReqPkt) deliverPacket {
 }
 
 func deliverPacketFromCMPP3(pkt *cmpp.Cmpp3DeliverReqPkt) deliverPacket {
-	return deliverPacket{
+	return deliverPacket{version: cmpp.V30,
 		seqID:            pkt.SeqId,
 		msgID:            pkt.MsgId,
 		destID:           pkt.DestId,
@@ -54,7 +55,7 @@ func deliverPacketFromCMPP3(pkt *cmpp.Cmpp3DeliverReqPkt) deliverPacket {
 func (c *connection) handleDeliver(pkt deliverPacket) error {
 	if pkt.registerDelivery == 1 {
 		var receipt cmpp.CmppReceiptPkt
-		if err := receipt.Unpack([]byte(pkt.msgContent)); err != nil {
+		if err := receipt.UnpackVersion([]byte(pkt.msgContent), pkt.version); err != nil {
 			log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_receipt status=parse_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err)
 			return err
 		}

gateway/internal/upstream/flow_control.go

@@ -166,7 +166,7 @@ func (c *connection) sendHeartbeat() bool {
 		return false
 	}
 	c.sendMu.Lock()
-	seq, err := c.client.SendReqPkt(&cmpp.CmppActiveTestReqPkt{})
+	seq, err := c.client.SendReqPktAvailable(&cmpp.CmppActiveTestReqPkt{}, c.sequenceAvailable)
 	c.sendMu.Unlock()
 	if err != nil {
 		c.mu.Unlock()

gateway/internal/upstream/protocol_fields_test.go

@@ -0,0 +1,50 @@
+package upstream
+
+import (
+	"cmpp-platform/gateway/internal/queue"
+	"encoding/json"
+	cmpp "github.com/bigwhite/gocmpp"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestCMPP3ReceiptKeepsUnsignedFieldsAndFullDestination(t *testing.T) {
+	events := make(chan queue.ReceiptEvent, 1)
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var e queue.ReceiptEvent
+		if err := json.NewDecoder(r.Body).Decode(&e); err != nil {
+			t.Error(err)
+		}
+		events <- e
+		w.WriteHeader(200)
+	}))
+	defer api.Close()
+	receipt := cmpp.CmppReceiptPkt{MsgId: ^uint64(0), Stat: "DELIVRD", DestTerminalId: strings.Repeat("9", 32), SmscSequence: ^uint32(0)}
+	raw, err := receipt.PackVersion(cmpp.V30)
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn := &connection{channelID: "qa", apiBaseURL: api.URL, httpClient: api.Client()}
+	packet := deliverPacketFromCMPP3(&cmpp.Cmpp3DeliverReqPkt{SeqId: ^uint32(0), RegisterDelivery: 1, MsgContent: string(raw)})
+	if err = conn.handleDeliver(packet); err != nil {
+		t.Fatal(err)
+	}
+	event := <-events
+	if event.GatewayMessageID != "18446744073709551615" || event.SequenceID != ^uint32(0) || event.PhoneNumber != receipt.DestTerminalId {
+		t.Fatalf("truncated callback: %+v", event)
+	}
+	packet.msgContent = string(raw[:60])
+	if conn.handleDeliver(packet) == nil {
+		t.Fatal("legacy 60-byte body silently accepted on CMPP3")
+	}
+}
+
+func TestUpstreamSequenceAvailabilityIncludesHeartbeatAndSubmit(t *testing.T) {
+	conn := &connection{pending: map[uint32]chan submitPartResponse{0: make(chan submitPartResponse)}, heartbeatPending: map[uint32]time.Time{^uint32(0): time.Now()}}
+	if conn.sequenceAvailable(0) || conn.sequenceAvailable(^uint32(0)) || !conn.sequenceAvailable(1) {
+		t.Fatal("wrapped sequence overwrites outstanding request")
+	}
+}

gateway/internal/upstream/protocol_log.go

@@ -23,7 +23,7 @@ func (c *connection) emitDeliverResponse(pkt deliverPacket, responseErr error) {
 	channelID := c.channelID
 	if pkt.registerDelivery == 1 {
 		var receipt cmpp.CmppReceiptPkt
-		if err := receipt.Unpack([]byte(pkt.msgContent)); err == nil {
+		if err := receipt.UnpackVersion([]byte(pkt.msgContent), pkt.version); err == nil {
 			gatewayMessageID = fmt.Sprint(receipt.MsgId)
 			phone = strings.TrimSpace(receipt.DestTerminalId)
 			if cmd, ok := c.commandFor(receipt.MsgId); ok {

gateway/internal/upstream/submit.go

@@ -101,7 +101,7 @@ func (p *connectionPool) submit(
 				return result, publishErr
 			}
 		}
-		if firstSequence == 0 {
+		if len(segments) == 1 {
 			firstSequence = seq
 		}
 		if firstGatewayMessageID == "" {
@@ -146,7 +146,7 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
 	}
 	c.sendMu.Lock()
 	wireSource = "write_uncertain"
-	seq, err := client.SendReqPkt(pkt)
+	seq, err := client.SendReqPktAvailable(pkt, c.sequenceAvailable)
 	if err == nil {
 		at := time.Now().UTC()
 		wireAt = &at

gateway/third_party/gocmpp/client.go

@@ -15,6 +15,7 @@ package cmpp
 
 import (
 	"errors"
+	"fmt"
 	"net"
 	"time"
 )
@@ -72,7 +73,7 @@ func (cli *Client) Connect(servAddr, user, password string, timeout time.Duratio
 	}
 
 	var ok bool
-	var status uint8
+	var status uint32
 	if cli.typ == V20 || cli.typ == V21 {
 		var rsp *Cmpp2ConnRspPkt
 		rsp, ok = p.(*Cmpp2ConnRspPkt)
@@ -80,7 +81,7 @@ func (cli *Client) Connect(servAddr, user, password string, timeout time.Duratio
 			err = ErrRespNotMatch
 			return err
 		}
-		status = rsp.Status
+		status = uint32(rsp.Status)
 	} else {
 		var rsp *Cmpp3ConnRspPkt
 		rsp, ok = p.(*Cmpp3ConnRspPkt)
@@ -88,15 +89,16 @@ func (cli *Client) Connect(servAddr, user, password string, timeout time.Duratio
 			err = ErrRespNotMatch
 			return err
 		}
-		status = uint8(rsp.Status)
+		status = rsp.Status
 	}
 
 	if status != 0 {
-		if status <= ErrnoConnOthers { //ErrnoConnOthers = 5
-			err = ConnRspStatusErrMap[status]
+		if status <= uint32(ErrnoConnOthers) { //ErrnoConnOthers = 5
+			err = ConnRspStatusErrMap[uint8(status)]
 		} else {
 			err = ConnRspStatusErrMap[ErrnoConnOthers]
 		}
+		err = fmt.Errorf("CMPP CONNECT_RESP status=%d: %w", status, err)
 		return err
 	}
 
@@ -112,8 +114,21 @@ func (cli *Client) Disconnect() {
 
 // SendReqPkt pack the cmpp request packet structure and send it to the other peer.
 func (cli *Client) SendReqPkt(packet Packer) (uint32, error) {
-	seq := <-cli.conn.SeqId
-	return seq, cli.conn.SendPkt(packet, seq)
+	return cli.SendReqPktAvailable(packet, nil)
+}
+
+// The caller holds its pending-request lock until the returned sequence is registered.
+// A wrapped sequence must never replace an outstanding request.
+func (cli *Client) SendReqPktAvailable(packet Packer, available func(uint32) bool) (uint32, error) {
+	for {
+		seq, ok := <-cli.conn.SeqId
+		if !ok {
+			return 0, ErrConnIsClosed
+		}
+		if available == nil || available(seq) {
+			return seq, cli.conn.SendPkt(packet, seq)
+		}
+	}
 }
 
 // SendRspPkt pack the cmpp response packet structure and send it to the other peer.

gateway/third_party/gocmpp/packet.go

@@ -46,7 +46,7 @@ const (
 	CMPP_HEADER_LEN  uint32 = 12
 	CMPP2_PACKET_MAX uint32 = 2477
 	CMPP2_PACKET_MIN uint32 = 12
-	CMPP3_PACKET_MAX uint32 = 3335
+	CMPP3_PACKET_MAX uint32 = Cmpp3SubmitReqPktMaxLen
 	CMPP3_PACKET_MIN uint32 = 12
 )
 

gateway/third_party/gocmpp/protocol_compatibility_test.go

@@ -0,0 +1,181 @@
+package cmpp
+
+import (
+	"encoding/binary"
+	"fmt"
+	"net"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestReceiptVersionLayout(t *testing.T) {
+	for _, version := range []Type{V20, V21, V30} {
+		width, size := 21, 60
+		if version == V30 {
+			width, size = 32, 71
+		}
+		original := CmppReceiptPkt{MsgId: ^uint64(0), Stat: "DELIVRD", SubmitTime: "2609201200", DoneTime: "2609201201", DestTerminalId: strings.Repeat("9", width), SmscSequence: ^uint32(0)}
+		raw, err := original.PackVersion(version)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(raw) != size || binary.BigEndian.Uint32(raw[size-4:]) != ^uint32(0) {
+			t.Fatalf("wrong layout: %x", raw)
+		}
+		var decoded CmppReceiptPkt
+		if err = decoded.UnpackVersion(raw, version); err != nil || !reflect.DeepEqual(decoded, original) {
+			t.Fatalf("roundtrip: %+v %v", decoded, err)
+		}
+		for _, bad := range [][]byte{raw[:len(raw)-1], append(append([]byte{}, raw...), 0)} {
+			if decoded.UnpackVersion(bad, version) == nil {
+				t.Fatal("invalid length accepted")
+			}
+		}
+		other := V30
+		if version == V30 {
+			other = V20
+		}
+		if decoded.UnpackVersion(raw, other) == nil {
+			t.Fatal("wrong version accepted")
+		}
+		original.DestTerminalId += "1"
+		if _, err = original.PackVersion(version); err == nil {
+			t.Fatal("truncated destination")
+		}
+	}
+}
+
+// Real TCP peers exercise the bounded reader, not only Pack/Unpack in memory.
+func tcpPair(t *testing.T, version Type) (*Conn, net.Conn) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+	client, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	server, err := listener.Accept()
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn := NewConn(server, version)
+	conn.SetState(CONN_AUTHOK)
+	t.Cleanup(func() { conn.Close(); client.Close() })
+	return conn, client
+}
+
+func TestCMPP3LargeSubmitAndMalformedPackets(t *testing.T) {
+	for _, length := range []int{140, 159} {
+		p := Cmpp3SubmitReqPkt{DestUsrTl: 99, DestTerminalId: make([]string, 99), MsgLength: uint8(length), MsgContent: strings.Repeat("x", length)}
+		if length == 140 {
+			p.MsgFmt = 8
+		}
+		raw, err := p.Pack(^uint32(0))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(raw) != 3331+length {
+			t.Fatalf("size %d", len(raw))
+		}
+		conn, peer := tcpPair(t, V30)
+		go peer.Write(raw)
+		pkt, err := conn.RecvAndUnpackPkt(time.Second)
+		if err != nil {
+			t.Fatal(err)
+		}
+		decoded := pkt.(*Cmpp3SubmitReqPkt)
+		if decoded.SeqId != ^uint32(0) || len(decoded.DestTerminalId) != 99 || decoded.MsgContent != p.MsgContent {
+			t.Fatal("wire mismatch")
+		}
+		var d Cmpp3SubmitReqPkt
+		if d.Unpack(raw[8:len(raw)-1]) == nil || d.Unpack(append(raw[8:], 0)) == nil {
+			t.Fatal("malformed body accepted")
+		}
+		p.DestUsrTl = 100
+		p.DestTerminalId = append(p.DestTerminalId, "")
+		if _, err = p.Pack(0); err == nil {
+			t.Fatal("100 destinations accepted")
+		}
+	}
+	for _, size := range []uint32{0, 11, CMPP3_PACKET_MAX + 1, ^uint32(0)} {
+		conn, peer := tcpPair(t, V30)
+		raw := make([]byte, 4)
+		binary.BigEndian.PutUint32(raw, size)
+		go peer.Write(raw)
+		if _, err := conn.RecvAndUnpackPkt(time.Second); err == nil {
+			t.Fatalf("accepted length %d", size)
+		}
+	}
+	p := Cmpp3SubmitReqPkt{DestUsrTl: 1, DestTerminalId: []string{"1"}, MsgFmt: 8, MsgLength: 141, MsgContent: strings.Repeat("a", 141)}
+	if _, err := p.Pack(0); err == nil {
+		t.Fatal("oversized non-ASCII accepted")
+	}
+	p.MsgFmt, p.MsgLength, p.MsgContent = 0, 160, strings.Repeat("a", 160)
+	if _, err := p.Pack(0); err == nil {
+		t.Fatal("ASCII must be strictly shorter than 160 bytes")
+	}
+}
+
+func TestConnectStatusKeepsAll32Bits(t *testing.T) {
+	for _, status := range []uint32{0, 5, 255, 256, ^uint32(0)} {
+		t.Run(fmt.Sprint(status), func(t *testing.T) {
+			listener, err := net.Listen("tcp", "127.0.0.1:0")
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer listener.Close()
+			done := make(chan error, 1)
+			go func() {
+				peer, e := listener.Accept()
+				if e != nil {
+					done <- e
+					return
+				}
+				conn := NewConn(peer, V30)
+				defer conn.Close()
+				conn.SetState(CONN_CONNECTED)
+				req, e := conn.RecvAndUnpackPkt(time.Second)
+				if e == nil {
+					e = conn.SendPkt(&Cmpp3ConnRspPkt{Status: status, Version: V30}, req.(*CmppConnReqPkt).SeqId)
+				}
+				done <- e
+			}()
+			client := NewClient(V30)
+			defer client.Disconnect()
+			err = client.Connect(listener.Addr().String(), "123456", "secret", time.Second)
+			if status == 0 && err != nil {
+				t.Fatal(err)
+			}
+			if status != 0 && (err == nil || !strings.Contains(err.Error(), fmt.Sprintf("status=%d", status))) {
+				t.Fatalf("status truncated: %v", err)
+			}
+			if e := <-done; e != nil {
+				t.Fatal(e)
+			}
+		})
+	}
+}
+
+func TestRequestSequenceSkipsInFlightAcrossWrap(t *testing.T) {
+	conn, peer := tcpPair(t, V30)
+	sequences := make(chan uint32, 3)
+	sequences <- ^uint32(0)
+	sequences <- 0
+	sequences <- 1
+	conn.SeqId = sequences
+	client := &Client{conn: conn, typ: V30}
+	read := make(chan error, 1)
+	go func() { raw := make([]byte, 12); _, err := peer.Read(raw); read <- err }()
+	seq, err := client.SendReqPktAvailable(&CmppActiveTestReqPkt{}, func(n uint32) bool { return n != ^uint32(0) })
+	if err != nil || seq != 0 {
+		t.Fatalf("zero lost: %d %v", seq, err)
+	}
+	if err = <-read; err != nil {
+		t.Fatal(err)
+	}
+}

gateway/third_party/gocmpp/receipt.go

@@ -13,11 +13,15 @@
 
 package cmpp
 
-import "encoding/binary"
+import (
+	"encoding/binary"
+	"fmt"
+)
 
 // Packet length const for cmpp receipt packet.
 const (
-	CmppReceiptPktLen uint32 = 60 //60d, 0x3c
+	Cmpp3ReceiptPktLen uint32 = 71
+	CmppReceiptPktLen  uint32 = 60 //60d, 0x3c
 )
 
 type CmppReceiptPkt struct {
@@ -31,7 +35,18 @@ type CmppReceiptPkt struct {
 
 // Pack packs the CmppReceiptPkt to bytes stream for client side.
 func (p *CmppReceiptPkt) Pack() ([]byte, error) {
-	var pktLen uint32 = CmppReceiptPktLen
+	return p.PackVersion(V20)
+}
+
+// PackVersion uses the negotiated connection version, never a body-length guess.
+func (p *CmppReceiptPkt) PackVersion(version Type) ([]byte, error) {
+	pktLen, width, err := receiptLayout(version)
+	if err != nil {
+		return nil, err
+	}
+	if len(p.Stat) > 7 || len(p.SubmitTime) > 10 || len(p.DoneTime) > 10 || len(p.DestTerminalId) > width {
+		return nil, fmt.Errorf("receipt field exceeds protocol width")
+	}
 
 	var w = newPacketWriter(pktLen)
 
@@ -39,7 +54,7 @@ func (p *CmppReceiptPkt) Pack() ([]byte, error) {
 	w.WriteFixedSizeString(p.Stat, 7)
 	w.WriteFixedSizeString(p.SubmitTime, 10)
 	w.WriteFixedSizeString(p.DoneTime, 10)
-	w.WriteFixedSizeString(p.DestTerminalId, 21)
+	w.WriteFixedSizeString(p.DestTerminalId, width)
 	w.WriteInt(binary.BigEndian, p.SmscSequence)
 
 	return w.Bytes()
@@ -49,6 +64,27 @@ func (p *CmppReceiptPkt) Pack() ([]byte, error) {
 // After unpack, you will get all value of fields in
 // CmppReceiptPkt struct.
 func (p *CmppReceiptPkt) Unpack(data []byte) error {
+	return p.UnpackVersion(data, V20)
+}
+
+func receiptLayout(version Type) (uint32, int, error) {
+	switch version {
+	case V20, V21:
+		return CmppReceiptPktLen, 21, nil
+	case V30:
+		return Cmpp3ReceiptPktLen, 32, nil
+	}
+	return 0, 0, fmt.Errorf("unsupported receipt version: %v", version)
+}
+
+func (p *CmppReceiptPkt) UnpackVersion(data []byte, version Type) error {
+	size, width, err := receiptLayout(version)
+	if err != nil {
+		return err
+	}
+	if len(data) != int(size) {
+		return fmt.Errorf("invalid receipt length %d for version %v (expected %d)", len(data), version, size)
+	}
 	var r = newPacketReader(data)
 
 	r.ReadInt(binary.BigEndian, &p.MsgId)
@@ -62,7 +98,7 @@ func (p *CmppReceiptPkt) Unpack(data []byte) error {
 	doneTime := r.ReadCString(10)
 	p.DoneTime = string(doneTime)
 
-	destTerminalId := r.ReadCString(21)
+	destTerminalId := r.ReadCString(width)
 	p.DestTerminalId = string(destTerminalId)
 
 	r.ReadInt(binary.BigEndian, &p.SmscSequence)

gateway/third_party/gocmpp/submit.go

@@ -152,6 +152,9 @@ type Cmpp3SubmitRspPkt struct {
 // Before calling Pack, you should initialize a Cmpp2SubmitReqPkt variable
 // with correct field value.
 func (p *Cmpp2SubmitReqPkt) Pack(seqId uint32) ([]byte, error) {
+	if err := validateSubmit(p.DestUsrTl, p.DestTerminalId, p.MsgFmt, p.MsgLength, p.MsgContent); err != nil {
+		return nil, err
+	}
 	var pktLen uint32 = CMPP_HEADER_LEN + 117 + uint32(p.DestUsrTl)*21 + 1 + uint32(p.MsgLength) + 8
 
 	var w = newPacketWriter(pktLen)
@@ -200,6 +203,8 @@ func (p *Cmpp2SubmitReqPkt) Pack(seqId uint32) ([]byte, error) {
 // Usually it is used in server side. After unpack, you will get all value of fields in
 // Cmpp2SubmitReqPkt struct.
 func (p *Cmpp2SubmitReqPkt) Unpack(data []byte) error {
+	p.DestTerminalId = nil
+
 	var r = newPacketReader(data)
 
 	// Sequence Id
@@ -259,7 +264,13 @@ func (p *Cmpp2SubmitReqPkt) Unpack(data []byte) error {
 	reserve := r.ReadCString(8)
 	p.Reserve = string(reserve)
 
-	return r.Error()
+	if err := r.Error(); err != nil {
+		return err
+	}
+	if len(data) != 130+int(p.DestUsrTl)*21+int(p.MsgLength) {
+		return errSubmitInvalidStruct
+	}
+	return validateSubmit(p.DestUsrTl, p.DestTerminalId, p.MsgFmt, p.MsgLength, p.MsgContent)
 }
 
 // Pack packs the Cmpp2SubmitRspPkt to bytes stream for Server side.
@@ -302,6 +313,9 @@ func (p *Cmpp2SubmitRspPkt) Unpack(data []byte) error {
 // Before calling Pack, you should initialize a Cmpp3SubmitReqPkt variable
 // with correct field value.
 func (p *Cmpp3SubmitReqPkt) Pack(seqId uint32) ([]byte, error) {
+	if err := validateSubmit(p.DestUsrTl, p.DestTerminalId, p.MsgFmt, p.MsgLength, p.MsgContent); err != nil {
+		return nil, err
+	}
 	var pktLen uint32 = CMPP_HEADER_LEN + 129 + uint32(p.DestUsrTl)*32 + 1 + 1 + uint32(p.MsgLength) + 20
 
 	var w = newPacketWriter(pktLen)
@@ -352,6 +366,8 @@ func (p *Cmpp3SubmitReqPkt) Pack(seqId uint32) ([]byte, error) {
 // Usually it is used in server side. After unpack, you will get all value of fields in
 // Cmpp3SubmitReqPkt struct.
 func (p *Cmpp3SubmitReqPkt) Unpack(data []byte) error {
+	p.DestTerminalId = nil
+
 	var r = newPacketReader(data)
 
 	// Sequence Id
@@ -413,7 +429,13 @@ func (p *Cmpp3SubmitReqPkt) Unpack(data []byte) error {
 	linkId := r.ReadCString(20)
 	p.LinkId = string(linkId)
 
-	return r.Error()
+	if err := r.Error(); err != nil {
+		return err
+	}
+	if len(data) != 155+int(p.DestUsrTl)*32+int(p.MsgLength) {
+		return errSubmitInvalidStruct
+	}
+	return validateSubmit(p.DestUsrTl, p.DestTerminalId, p.MsgFmt, p.MsgLength, p.MsgContent)
 }
 
 // Pack packs the Cmpp3SubmitRspPkt to bytes stream for Server side.
@@ -451,3 +473,18 @@ func (p *Cmpp3SubmitRspPkt) Unpack(data []byte) error {
 
 	return r.Error()
 }
+
+// The receive buffer is bounded separately; validate counts before accepting the body.
+func validateSubmit(count uint8, destinations []string, format, length uint8, content string) error {
+	if count == 0 || count > 99 || int(count) != len(destinations) {
+		return errSubmitInvalidStruct
+	}
+	limit := 140
+	if format == 0 {
+		limit = 159 // CMPP specifies ASCII <160 bytes; other formats <=140.
+	}
+	if int(length) != len(content) || len(content) > limit {
+		return errSubmitInvalidMsgLength
+	}
+	return nil
+}