fix: harden upstream and downstream receipt delivery

gateway/internal/inbound/server.go

@@ -209,6 +209,11 @@ var downstreamAckRegistry = struct {
 	items map[string]*downstreamAckTracker
 }{items: make(map[string]*downstreamAckTracker)}
 
+var downstreamSubmitBarrier = struct {
+	sync.RWMutex
+	byConn map[*cmpp.Conn]int
+}{byConn: make(map[*cmpp.Conn]int)}
+
 func (s Server) ListenAndServe() error {
 	addr := s.Addr
 	if addr == "" {
@@ -336,6 +341,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
 	}
 	contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content)))
 	startedAt := time.Now()
+	releaseSubmitBarrier := beginDownstreamSubmitBarrier(packet.Conn)
 	result, err := s.submit(remote, submitRequest{
 		Account:      account,
 		PhoneNumber:  phone,
@@ -361,7 +367,11 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
 			clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason,
 		)
 		setInboundSubmitResponse(response.Packer, 0, responseResult)
-		response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult)
+		protocolLogger := s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult)
+		response.AfterSend = func(sendErr error) {
+			releaseSubmitBarrier()
+			protocolLogger(sendErr)
+		}
 		return false, nil
 	}
 	gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID)
@@ -401,6 +411,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
 		go current.report(current, "submit", "")
 	}
 	response.AfterSend = func(sendErr error) {
+		releaseSubmitBarrier()
 		s.emitProtocolLog(protocolLogEvent{
 			Protocol:         "cmpp",
 			Direction:        "platform_to_client",
@@ -538,6 +549,9 @@ func (s Server) reportConnectionOrDisconnect(session *downstreamSession, status
 }
 
 func (s Server) handleConnectionClosed(conn *cmpp.Conn) {
+	downstreamSubmitBarrier.Lock()
+	delete(downstreamSubmitBarrier.byConn, conn)
+	downstreamSubmitBarrier.Unlock()
 	session := findSessionByConn(conn)
 	if session == nil {
 		return
@@ -705,10 +719,10 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
 		if err != nil {
 			result.FailedCount++
 			result.LastError = err.Error()
-			_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{
-				"id":           delivery.ID,
-				"errorMessage": err.Error(),
-				"failureType":  "send_failed",
+			_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
+				"id": delivery.ID, "errorMessage": err.Error(), "failureType": "send_failed",
+				"connectionId": sendResult.ConnectionID, "sequenceId": sendResult.SequenceID,
+				"messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
 			}, nil)
 			continue
 		}
@@ -716,6 +730,10 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
 			result.DeliveredCount++
 			continue
 		}
+		if sendResult.ReasonCode == "SUBMIT_RESPONSE_PENDING" {
+			result.WaitingCount++
+			continue
+		}
 		errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery")
 		failureType := "unrecoverable"
 		if sendResult.Retryable {
@@ -725,10 +743,10 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
 			result.FailedCount++
 		}
 		result.LastError = errorMessage
-		_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{
-			"id":           delivery.ID,
-			"errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
-			"failureType":  failureType,
+		_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
+			"id": delivery.ID, "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
+			"failureType": failureType, "connectionId": sendResult.ConnectionID,
+			"sequenceId": sendResult.SequenceID, "messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
 		}, nil)
 	}
 	return result, nil
@@ -1153,6 +1171,13 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
 			ErrorMessage: "下游客户端当前未连接，等待自动重试",
 		}, nil
 	}
+	if downstreamSubmitResponsePending(session.conn) {
+		return DownstreamSendResult{
+			Retryable:    true,
+			ReasonCode:   "SUBMIT_RESPONSE_PENDING",
+			ErrorMessage: "客户 SubmitResp 尚未完成写出，回执已保留并等待响应后投递",
+		}, nil
+	}
 	stat := strings.TrimSpace(event.RawStatus)
 	if stat == "" {
 		stat = cmppReceiptStatus(event.ReceiptStatus)
@@ -1179,6 +1204,36 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
 	return sendDownstream(session, deliver, event.DeliveryID)
 }
 
+func beginDownstreamSubmitBarrier(conn *cmpp.Conn) func() {
+	if conn == nil {
+		return func() {}
+	}
+	downstreamSubmitBarrier.Lock()
+	downstreamSubmitBarrier.byConn[conn]++
+	downstreamSubmitBarrier.Unlock()
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			downstreamSubmitBarrier.Lock()
+			if downstreamSubmitBarrier.byConn[conn] <= 1 {
+				delete(downstreamSubmitBarrier.byConn, conn)
+			} else {
+				downstreamSubmitBarrier.byConn[conn]--
+			}
+			downstreamSubmitBarrier.Unlock()
+		})
+	}
+}
+
+func downstreamSubmitResponsePending(conn *cmpp.Conn) bool {
+	if conn == nil {
+		return false
+	}
+	downstreamSubmitBarrier.RLock()
+	defer downstreamSubmitBarrier.RUnlock()
+	return downstreamSubmitBarrier.byConn[conn] > 0
+}
+
 func findReceiptSession(messageID string, account string) *downstreamSession {
 	downstreamRegistry.RLock()
 	defer downstreamRegistry.RUnlock()
@@ -1288,6 +1343,13 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID
 	sequenceID := <-session.conn.SeqId
 	sentAt := time.Now().UTC()
 	ackDeadlineAt := sentAt.Add(downstreamAckTimeout())
+	result := DownstreamSendResult{
+		ConnectionID:  session.connectionID,
+		SequenceID:    strconv.FormatUint(uint64(sequenceID), 10),
+		MessageID:     strconv.FormatUint(messageID, 10),
+		SentAt:        formatRFC3339Nano(sentAt),
+		AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
+	}
 	tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt)
 	if err := session.conn.SendPkt(deliver, sequenceID); err != nil {
 		removeDownstreamAck(tracker)
@@ -1296,14 +1358,13 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID
 			go session.report(session, "disconnected", err.Error())
 		}
 		forgetDownstream(session)
-		return DownstreamSendResult{}, err
+		result.Retryable = true
+		result.ReasonCode = "SEND_FAILED"
+		result.ErrorMessage = err.Error()
+		return result, nil
 	}
 	session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "success", "", nil)
-	result := DownstreamSendResult{
-		Sent: true, ConnectionID: session.connectionID,
-		SequenceID: strconv.FormatUint(uint64(sequenceID), 10), MessageID: strconv.FormatUint(messageID, 10),
-		SentAt: formatRFC3339Nano(sentAt), AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
-	}
+	result.Sent = true
 	if deliveryID != "" && session.deliveryReport != nil {
 		go session.deliveryReport(downstreamDeliveryLifecycleEvent{
 			Kind: "sent", DeliveryID: deliveryID, ConnectionID: session.connectionID,
@@ -1433,6 +1494,22 @@ func handleDownstreamAcknowledgement(conn *cmpp.Conn, sequenceID uint32, message
 		if logger != nil {
 			logger.Printf("cmpp inbound event=deliver_ack_unmatched seq=%d message_id=%d result=%d", sequenceID, messageID, result)
 		}
+		if session := findSessionByConn(conn); session != nil && session.protocolLog != nil {
+			session.protocolLog(protocolLogEvent{
+				Protocol:         "cmpp",
+				Direction:        "client_to_platform",
+				EventType:        "deliver_resp",
+				Status:           "failed",
+				TenantID:         session.tenantID,
+				ApplicationID:    session.applicationID,
+				Account:          session.account,
+				MessageID:        session.messageID,
+				GatewayMessageID: strconv.FormatUint(messageID, 10),
+				Phone:            session.phoneNumber,
+				ResultCode:       strconv.FormatUint(uint64(result), 10),
+				Detail:           map[string]any{"sequenceId": sequenceID, "unmatched": true},
+			})
+		}
 		return
 	}
 	if tracker.messageID != messageID {

gateway/internal/inbound/server_test.go

@@ -549,6 +549,41 @@ func TestRecoverableReceiptWaitsForClientConnection(t *testing.T) {
 	}
 }
 
+func TestReceiptWaitsUntilCurrentSubmitResponseHasBeenWritten(t *testing.T) {
+	resetDownstreamRegistry()
+	defer resetDownstreamRegistry()
+
+	conn := &cmpp.Conn{}
+	session := &downstreamSession{
+		messageID: "MSG-LONG-1",
+		account:   "100001",
+		conn:      conn,
+	}
+	downstreamRegistry.Lock()
+	downstreamRegistry.byMessageID[session.messageID] = session
+	downstreamRegistry.byAccount[session.account] = session
+	downstreamRegistry.byConn[conn] = session
+	downstreamRegistry.Unlock()
+
+	release := beginDownstreamSubmitBarrier(conn)
+	result, err := PushReceiptWithResult(DownstreamReceipt{
+		DeliveryID:    "delivery-long-failed",
+		Account:       session.account,
+		MessageID:     session.messageID,
+		ReceiptStatus: "undelivered",
+	})
+	if err != nil {
+		t.Fatalf("push guarded receipt: %v", err)
+	}
+	if result.Sent || !result.Retryable || result.ReasonCode != "SUBMIT_RESPONSE_PENDING" {
+		t.Fatalf("unexpected guarded result: %+v", result)
+	}
+	release()
+	if downstreamSubmitResponsePending(conn) {
+		t.Fatal("submit response barrier remained active after release")
+	}
+}
+
 func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) {
 	resetDownstreamRegistry()
 	defer resetDownstreamRegistry()
@@ -1058,6 +1093,9 @@ func resetDownstreamRegistry() {
 	}
 	downstreamAckRegistry.items = make(map[string]*downstreamAckTracker)
 	downstreamAckRegistry.Unlock()
+	downstreamSubmitBarrier.Lock()
+	downstreamSubmitBarrier.byConn = make(map[*cmpp.Conn]int)
+	downstreamSubmitBarrier.Unlock()
 }
 
 func reserveTCPAddr(t *testing.T) string {

gateway/internal/queue/messages.go

@@ -108,6 +108,7 @@ type ReceiptEvent struct {
 	RawStatus        string    `json:"rawStatus"`
 	ErrorCode        string    `json:"errorCode,omitempty"`
 	DeliveredAt      time.Time `json:"deliveredAt"`
+	ConnectionID     string    `json:"connectionId,omitempty"`
 }
 
 type UplinkEvent struct {

gateway/internal/upstream/deliver_test.go

@@ -15,7 +15,7 @@ import (
 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" {
+		if r.URL.Path != "/gateway/events/receipt/intake" {
 			t.Fatalf("unexpected path: %s", r.URL.Path)
 		}
 		var event queue.ReceiptEvent
@@ -58,12 +58,14 @@ func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) {
 			},
 		},
 	}
-	conn.handleDeliver(deliverPacketFromCMPP2(&cmpp.Cmpp2DeliverReqPkt{
+	if err := conn.handleDeliver(deliverPacketFromCMPP2(&cmpp.Cmpp2DeliverReqPkt{
 		SeqId:            7,
 		MsgId:            999,
 		RegisterDelivery: 1,
 		MsgContent:       string(payload),
-	}))
+	})); err != nil {
+		t.Fatalf("handle receipt: %v", err)
+	}
 
 	select {
 	case event := <-events:
@@ -82,6 +84,9 @@ func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) {
 		if event.ReceiptStatus != "delivered" || event.RawStatus != "DELIVRD" {
 			t.Fatalf("unexpected receipt status: %+v", event)
 		}
+		if event.ConnectionID != "channel-1-0" {
+			t.Fatalf("ConnectionID = %q, want channel-1-0", event.ConnectionID)
+		}
 	case <-time.After(time.Second):
 		t.Fatal("timed out waiting for receipt event")
 	}

gateway/internal/upstream/manager.go

@@ -73,7 +73,26 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su
 		return result, err
 	}
 
-	result, err := pool.submit(ctx, cmd)
+	result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) {
+		payload := struct {
+			queue.Envelope
+			SubmitID string `json:"submitId,omitempty"`
+			queue.SubmitSegmentResult
+		}{
+			Envelope:            cmd.Envelope,
+			SubmitID:            cmd.SubmitID,
+			SubmitSegmentResult: segment,
+		}
+		callbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		postErr := m.post(callbackCtx, "/gateway/events/submit-segment-result", payload)
+		cancel()
+		if postErr != nil {
+			log.Printf(
+				"protocol_event protocol=cmpp direction=gateway_to_api event=submit_segment_result status=forward_failed channel_id=%s message_id=%s segment=%d/%d error=%q",
+				cmd.ChannelID, cmd.MessageID, segment.SegmentIndex, segment.SegmentTotal, postErr,
+			)
+		}
+	})
 	if err != nil {
 		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
 			return result, postErr
@@ -404,7 +423,11 @@ func (p *connectionPool) superviseReconnects() {
 	}
 }
 
-func (p *connectionPool) submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
+func (p *connectionPool) submit(
+	ctx context.Context,
+	cmd queue.SubmitCommand,
+	onSegment func(queue.SubmitSegmentResult),
+) (queue.SubmitResult, error) {
 	parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
 	if err != nil {
 		result := submitResult(cmd, 0, "", "rejected", "ENCODE_FAILED", err.Error())
@@ -423,7 +446,11 @@ func (p *connectionPool) submit(ctx context.Context, cmd queue.SubmitCommand) (q
 		}
 		seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part)
 		release()
-		segments = append(segments, submitSegmentResult(part, seq, gatewayMessageID, result))
+		segment := submitSegmentResult(part, seq, gatewayMessageID, result)
+		segments = append(segments, segment)
+		if onSegment != nil {
+			onSegment(segment)
+		}
 		if firstSequence == 0 {
 			firstSequence = seq
 		}
@@ -878,13 +905,35 @@ func (c *connection) readLoop() {
 				ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result}
 			}
 		case *cmpp.Cmpp2DeliverReqPkt:
-			responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
-			c.emitDeliverResponse(deliverPacketFromCMPP2(p), responseErr)
-			c.handleDeliver(deliverPacketFromCMPP2(p))
+			deliver := deliverPacketFromCMPP2(p)
+			if deliver.registerDelivery == 1 {
+				if err := c.handleDeliver(deliver); err != nil {
+					log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=intake_failed channel_id=%s sequence_id=%d error=%q", c.channelID, deliver.seqID, err)
+					c.handleConnectionLoss(fmt.Errorf("persist upstream receipt before DELIVER_RESP: %w", err))
+					return
+				}
+				responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
+				c.emitDeliverResponse(deliver, responseErr)
+			} else {
+				responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
+				c.emitDeliverResponse(deliver, responseErr)
+				_ = c.handleDeliver(deliver)
+			}
 		case *cmpp.Cmpp3DeliverReqPkt:
-			responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
-			c.emitDeliverResponse(deliverPacketFromCMPP3(p), responseErr)
-			c.handleDeliver(deliverPacketFromCMPP3(p))
+			deliver := deliverPacketFromCMPP3(p)
+			if deliver.registerDelivery == 1 {
+				if err := c.handleDeliver(deliver); err != nil {
+					log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=intake_failed channel_id=%s sequence_id=%d error=%q", c.channelID, deliver.seqID, err)
+					c.handleConnectionLoss(fmt.Errorf("persist upstream receipt before DELIVER_RESP: %w", err))
+					return
+				}
+				responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
+				c.emitDeliverResponse(deliver, responseErr)
+			} else {
+				responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
+				c.emitDeliverResponse(deliver, responseErr)
+				_ = c.handleDeliver(deliver)
+			}
 		case *cmpp.CmppActiveTestReqPkt:
 			_ = c.sendResponse(client, &cmpp.CmppActiveTestRspPkt{}, p.SeqId)
 			_ = c.pool.reportState(context.Background(), "heartbeat", nil)
@@ -1069,12 +1118,12 @@ func deliverPacketFromCMPP3(pkt *cmpp.Cmpp3DeliverReqPkt) deliverPacket {
 	}
 }
 
-func (c *connection) handleDeliver(pkt 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 {
 			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
+			return err
 		}
 		log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_receipt status=received channel_id=%s sequence_id=%d gateway_message_id=%d raw_status=%s", c.channelID, pkt.seqID, receipt.MsgId, strings.TrimSpace(receipt.Stat))
 		cmd, ok := c.commandFor(receipt.MsgId)
@@ -1104,22 +1153,24 @@ func (c *connection) handleDeliver(pkt deliverPacket) {
 			ReceiptStatus:    receiptStatus(receipt.Stat),
 			RawStatus:        strings.TrimSpace(receipt.Stat),
 			DeliveredAt:      time.Now().UTC(),
+			ConnectionID:     c.identity(),
 		}
-		if err := postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt", event); err != nil {
+		if err := postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt/intake", event); err != nil {
 			log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=forward_failed channel_id=%s sequence_id=%d gateway_message_id=%d error=%q", c.channelID, pkt.seqID, receipt.MsgId, err)
+			return err
 		} else {
 			log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=forwarded channel_id=%s sequence_id=%d gateway_message_id=%d", c.channelID, pkt.seqID, receipt.MsgId)
 		}
-		return
+		return nil
 	}
 
 	content, complete, err := c.decodeUplinkContent(pkt)
 	if err != nil {
 		log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_uplink status=decode_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err)
-		return
+		return err
 	}
 	if !complete {
-		return
+		return nil
 	}
 	cmd, _ := c.commandFor(pkt.msgID)
 	event := queue.UplinkEvent{
@@ -1142,6 +1193,14 @@ func (c *connection) handleDeliver(pkt deliverPacket) {
 	} else {
 		log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_uplink status=forwarded channel_id=%s sequence_id=%d packet_msg_id=%d", c.channelID, pkt.seqID, pkt.msgID)
 	}
+	return nil
+}
+
+func (c *connection) identity() string {
+	if c.pool != nil && strings.TrimSpace(c.pool.connectionID) != "" {
+		return fmt.Sprintf("%s-%d", c.pool.connectionID, c.index)
+	}
+	return fmt.Sprintf("%s-%d", c.channelID, c.index)
 }
 
 func (c *connection) decodeUplinkContent(pkt deliverPacket) (string, bool, error) {