perf: expand gateway capacity and prevent receipt replay

gateway/cmd/gateway/main.go

@@ -13,6 +13,7 @@ import (
 	"cmpp-platform/gateway/internal/health"
 	"cmpp-platform/gateway/internal/inbound"
 	platformmetrics "cmpp-platform/gateway/internal/metrics"
+	"cmpp-platform/gateway/internal/protocollog"
 	"cmpp-platform/gateway/internal/queue"
 	"cmpp-platform/gateway/internal/ratelimit"
 	"cmpp-platform/gateway/internal/resultoutbox"
@@ -33,7 +34,17 @@ func main() {
 	}
 	apiBaseURL := os.Getenv("API_BASE_URL")
 	callbackBaseURL := getenv("GATEWAY_CALLBACK_API_BASE_URL", apiBaseURL)
-	upstreamManager := &upstream.Manager{APIBaseURL: apiBaseURL, EventAPIBaseURL: callbackBaseURL}
+	gatewayInstanceID := getenv("GATEWAY_INSTANCE_ID", hostname())
+	protocolLogPublisher, protocolLogErr := protocollog.New(os.Getenv("REDIS_URL"))
+	if protocolLogErr != nil {
+		log.Printf("gateway protocol log Redis publisher init failed: %v", protocolLogErr)
+	} else {
+		protocolLogPublisher.Stream = getenv("GATEWAY_PROTOCOL_LOG_STREAM", "gateway.protocol.logs")
+		protocolLogPublisher.GatewayInstanceID = gatewayInstanceID
+		protocolLogPublisher.SuccessSampleRate = positiveEnvInt("GATEWAY_PROTOCOL_LOG_SUCCESS_SAMPLE_PERCENT", 10)
+		protocolLogPublisher.MaxLen = int64(positiveEnvInt("GATEWAY_PROTOCOL_LOG_STREAM_MAX_LEN", 200000))
+	}
+	upstreamManager := &upstream.Manager{APIBaseURL: apiBaseURL, EventAPIBaseURL: callbackBaseURL, ProtocolLogPublisher: protocolLogPublisher, GatewayInstanceID: gatewayInstanceID}
 	var worker *submitworker.Worker
 	var resultOutbox *resultoutbox.Outbox
 	channelLimiter, err := ratelimit.New(os.Getenv("REDIS_URL"))
@@ -59,9 +70,10 @@ func main() {
 			SubmitHTTPClient:     inbound.NewAPIHTTPClient(inboundConcurrency),
 			PresenceStore:        presenceStore,
 			RecoveryStore:        recoveryStore,
-			GatewayInstanceID:    getenv("GATEWAY_INSTANCE_ID", hostname()),
+			GatewayInstanceID:    gatewayInstanceID,
 			MaxSubmitConcurrency: inboundConcurrency,
 			SecurityEventToken:   os.Getenv("SECURITY_EVENT_TOKEN"),
+			ProtocolLogPublisher: protocolLogPublisher,
 		}).ListenAndServe(); err != nil {
 			log.Fatalf("gateway inbound server stopped: %v", err)
 		}
@@ -83,8 +95,14 @@ func main() {
 			resultOutbox.Consumer = getenv("GATEWAY_SUBMIT_RESULT_CONSUMER", "gateway-1")
 			resultOutbox.APIBaseURL = callbackBaseURL
 			resultOutbox.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY", 8)
+			resultOutbox.BatchEnabled = os.Getenv("GATEWAY_CALLBACK_BATCH_ENABLED") != "false"
+			resultOutbox.BatchSize = positiveEnvInt("GATEWAY_CALLBACK_BATCH_SIZE", 50)
+			resultOutbox.BatchWait = time.Duration(positiveEnvInt("GATEWAY_CALLBACK_BATCH_WAIT_MS", 10)) * time.Millisecond
+			resultOutbox.GatewayInstanceID = gatewayInstanceID
+			resultOutbox.DeadLetterStream = getenv("GATEWAY_CALLBACK_DEAD_LETTER_STREAM", "gateway.submit.results.dead")
 			worker.ResultOutbox = resultOutbox
 			upstreamManager.SubmitSegmentPublisher = resultOutbox
+			upstreamManager.EventPublisher = resultOutbox
 			go func() {
 				log.Printf("cmpp gateway submit worker consuming stream=%s group=%s consumer=%s", worker.Stream, worker.Group, worker.Consumer)
 				if err := worker.Run(context.Background()); err != nil {
@@ -108,6 +126,13 @@ func main() {
 			UpstreamDesired: desired, UpstreamConnected: connected,
 			DownstreamConnected: inbound.ActiveConnectionCount(), SubmitWorkerUp: worker != nil,
 		}
+		snapshot.UpstreamWindowConfigured, snapshot.UpstreamWindowInFlight = upstreamManager.WindowCounts()
+		if protocolLogPublisher != nil {
+			snapshot.ProtocolLogPublished, snapshot.ProtocolLogSampled, snapshot.ProtocolLogErrors = protocolLogPublisher.Counts()
+			if length, err := protocolLogPublisher.Redis.XLen(ctx, protocolLogPublisher.StreamName()).Result(); err == nil {
+				snapshot.ProtocolLogQueueLength = length
+			}
+		}
 		if worker != nil {
 			snapshot.SubmitWorkerConcurrency = worker.ConfiguredConcurrency()
 			snapshot.SubmitWorkerInFlight = worker.InFlight()
@@ -116,6 +141,7 @@ func main() {
 			snapshot.ResultWorkerUp = true
 			snapshot.ResultWorkerConcurrency = resultOutbox.ConfiguredConcurrency()
 			snapshot.ResultWorkerInFlight = resultOutbox.InFlight()
+			snapshot.CallbackBatchRequests, snapshot.CallbackBatchEvents, snapshot.CallbackBatchRetries, snapshot.CallbackDeadLetters = resultOutbox.BatchCounts()
 		}
 		snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight = inbound.SubmitSlotSnapshot()
 		if worker == nil || worker.Redis == nil {

gateway/internal/control/server.go

@@ -41,6 +41,10 @@ type ChannelConfig struct {
 	WindowSize               int    `json:"windowSize,omitempty"`
 	HeartbeatIntervalSeconds int    `json:"heartbeatIntervalSeconds,omitempty"`
 	HeartbeatMissThreshold   int    `json:"heartbeatMissThreshold,omitempty"`
+	ConnectionWarmupSeconds  int    `json:"connectionWarmupSeconds,omitempty"`
+	ConnectionDrainSeconds   int    `json:"connectionDrainTimeoutSeconds,omitempty"`
+	SubmitTimeoutSeconds     int    `json:"submitResponseTimeoutSeconds,omitempty"`
+	FailureCooldownSeconds   int    `json:"connectionFailureCooldownSeconds,omitempty"`
 }
 
 type DisconnectChannelCommand struct {
@@ -351,6 +355,10 @@ func (s Server) connectChannel(ctx context.Context, command ConnectChannelComman
 			WindowSize:               command.Channel.WindowSize,
 			HeartbeatIntervalSeconds: command.Channel.HeartbeatIntervalSeconds,
 			HeartbeatMissThreshold:   command.Channel.HeartbeatMissThreshold,
+			ConnectionWarmupSeconds:  command.Channel.ConnectionWarmupSeconds,
+			ConnectionDrainSeconds:   command.Channel.ConnectionDrainSeconds,
+			SubmitTimeoutSeconds:     command.Channel.SubmitTimeoutSeconds,
+			FailureCooldownSeconds:   command.Channel.FailureCooldownSeconds,
 		},
 	})
 	if err != nil {
@@ -411,6 +419,12 @@ func validateConnectChannelCommand(command ConnectChannelCommand) error {
 	if command.Channel.Account == "" || command.Channel.PasswordCipher == "" {
 		return fmt.Errorf("account and passwordCipher are required")
 	}
+	if command.DesiredConnections < 1 || command.DesiredConnections > 8 {
+		return fmt.Errorf("desiredConnections must be between 1 and 8")
+	}
+	if command.Channel.WindowSize != 0 && (command.Channel.WindowSize < 1 || command.Channel.WindowSize > 64) {
+		return fmt.Errorf("windowSize must be between 1 and 64")
+	}
 	return nil
 }
 

gateway/internal/control/server_test.go

@@ -140,6 +140,22 @@ func TestConnectChannelRejectsInvalidCommand(t *testing.T) {
 	}
 }
 
+func TestConnectChannelRejectsCapacityOutsideSupportedMatrix(t *testing.T) {
+	handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
+		return ConnectionStateCallback{}, nil
+	})
+	for _, body := range []string{
+		`{"schemaVersion":"v1","messageType":"ConnectChannel","channelId":"c","connectionId":"x","desiredConnections":9,"channel":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"a","passwordCipher":"p","windowSize":16}}`,
+		`{"schemaVersion":"v1","messageType":"ConnectChannel","channelId":"c","connectionId":"x","desiredConnections":1,"channel":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"a","passwordCipher":"p","windowSize":65}}`,
+	} {
+		response := httptest.NewRecorder()
+		handler.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/connections/connect", strings.NewReader(body)))
+		if response.Code != http.StatusBadRequest {
+			t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
+		}
+	}
+}
+
 func TestDisconnectChannelStopsSupplierPool(t *testing.T) {
 	var received DisconnectChannelCommand
 	handler := handlerWithServer(Server{

gateway/internal/inbound/acknowledgement.go

@@ -18,6 +18,7 @@ const defaultDownstreamAckTimeout = 30 * time.Second
 
 type downstreamAckTracker struct {
 	deliveryID   string
+	claimID      string
 	connectionID string
 	sequenceID   uint32
 	messageID    uint64
@@ -69,29 +70,29 @@ func downstreamAckKey(conn *cmpp.Conn, sequenceID uint32) string {
 	return fmt.Sprintf("%p:%d", conn, sequenceID)
 }
 
-func registerDownstreamAck(session *downstreamSession, deliveryID string, sequenceID uint32, messageID uint64, deadline time.Time) *downstreamAckTracker {
+func registerDownstreamAck(session *downstreamSession, deliveryID string, claimID string, sequenceID uint32, messageID uint64, deadline time.Time) *downstreamAckTracker {
 	if session == nil || session.conn == nil || strings.TrimSpace(deliveryID) == "" {
 		return nil
 	}
 	tracker := &downstreamAckTracker{
-		deliveryID: deliveryID, connectionID: session.connectionID,
+		deliveryID: deliveryID, claimID: claimID, connectionID: session.connectionID,
 		sequenceID: sequenceID, messageID: messageID, session: session,
 	}
 	key := downstreamAckKey(session.conn, sequenceID)
 	downstreamAckRegistry.Lock()
 	downstreamAckRegistry.items[key] = tracker
-	downstreamAckRegistry.Unlock()
 	tracker.timer = time.AfterFunc(time.Until(deadline), func() {
 		timedOut := takeDownstreamAck(session.conn, sequenceID)
 		if timedOut == nil || timedOut.session == nil || timedOut.session.deliveryReport == nil {
 			return
 		}
 		timedOut.session.deliveryReport(downstreamDeliveryLifecycleEvent{
-			Kind: "failed", DeliveryID: timedOut.deliveryID, ConnectionID: timedOut.connectionID,
+			Kind: "failed", DeliveryID: timedOut.deliveryID, ClaimID: timedOut.claimID, ConnectionID: timedOut.connectionID,
 			SequenceID: timedOut.sequenceID, MessageID: timedOut.messageID, ObservedAt: time.Now().UTC(),
 			FailureType: "ack_timeout", ErrorMessage: "CMPP_DELIVER_RESP timeout",
 		})
 	})
+	downstreamAckRegistry.Unlock()
 	return tracker
 }
 
@@ -149,7 +150,7 @@ func handleDownstreamAcknowledgement(conn *cmpp.Conn, sequenceID uint32, message
 	}
 	if tracker.session != nil && tracker.session.deliveryReport != nil {
 		go tracker.session.deliveryReport(downstreamDeliveryLifecycleEvent{
-			Kind: "acknowledged", DeliveryID: tracker.deliveryID, ConnectionID: tracker.connectionID,
+			Kind: "acknowledged", DeliveryID: tracker.deliveryID, ClaimID: tracker.claimID, ConnectionID: tracker.connectionID,
 			SequenceID: sequenceID, MessageID: messageID, Result: result, ObservedAt: time.Now().UTC(),
 		})
 	}

gateway/internal/inbound/delivery.go

@@ -16,6 +16,7 @@ import (
 
 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"`
@@ -31,6 +32,7 @@ type DownstreamReceipt struct {
 
 type DownstreamUplink 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,omitempty"`
@@ -55,6 +57,7 @@ type DownstreamSendResult struct {
 type downstreamDeliveryLifecycleEvent struct {
 	Kind          string
 	DeliveryID    string
+	ClaimID       string
 	ConnectionID  string
 	SequenceID    uint32
 	MessageID     uint64
@@ -70,7 +73,7 @@ func (s Server) reportDownstreamDelivery(event downstreamDeliveryLifecycleEvent)
 		return
 	}
 	payload := map[string]any{
-		"id": event.DeliveryID, "connectionId": event.ConnectionID,
+		"id": event.DeliveryID, "claimId": event.ClaimID, "connectionId": event.ConnectionID,
 		"sequenceId": strconv.FormatUint(uint64(event.SequenceID), 10),
 		"messageId":  strconv.FormatUint(event.MessageID, 10),
 	}
@@ -160,7 +163,7 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
 		return DownstreamSendResult{}, err
 	}
 	deliver := downstreamDeliverPacket(session, receiptMessageID, session.srcID, defaultString(event.PhoneNumber, session.phoneNumber), 0, 1, string(receiptBytes))
-	return sendDownstream(session, deliver, event.DeliveryID)
+	return sendDownstream(session, deliver, event.DeliveryID, event.ClaimID)
 }
 
 func downstreamReceiptMessageID(event DownstreamReceipt, session *downstreamSession) uint64 {
@@ -228,7 +231,7 @@ func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error)
 		0,
 		content,
 	)
-	return sendDownstream(session, deliver, event.DeliveryID)
+	return sendDownstream(session, deliver, event.DeliveryID, event.ClaimID)
 }
 
 func errorMessageWithCode(message string, code string) string {
@@ -272,7 +275,7 @@ func findSession(messageID string, account string) *downstreamSession {
 	return nil
 }
 
-func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID string) (DownstreamSendResult, error) {
+func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID string, claimID string) (DownstreamSendResult, error) {
 	session.mu.Lock()
 	defer session.mu.Unlock()
 	messageID := downstreamDeliverMessageID(deliver)
@@ -289,7 +292,7 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID
 		SentAt:        formatRFC3339Nano(sentAt),
 		AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt),
 	}
-	tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt)
+	tracker := registerDownstreamAck(session, deliveryID, claimID, sequenceID, messageID, ackDeadlineAt)
 	if err := session.conn.SendPkt(deliver, sequenceID); err != nil {
 		removeDownstreamAck(tracker)
 		session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "failed", "SEND_FAILED", err)
@@ -306,7 +309,7 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID
 	result.Sent = true
 	if deliveryID != "" && session.deliveryReport != nil {
 		go session.deliveryReport(downstreamDeliveryLifecycleEvent{
-			Kind: "sent", DeliveryID: deliveryID, ConnectionID: session.connectionID,
+			Kind: "sent", DeliveryID: deliveryID, ClaimID: claimID, ConnectionID: session.connectionID,
 			SequenceID: sequenceID, MessageID: messageID, ObservedAt: sentAt, AckDeadlineAt: ackDeadlineAt,
 		})
 	}

gateway/internal/inbound/pending_recovery.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -15,6 +16,8 @@ import (
 type pendingDeliveryRequest struct {
 	Account string `json:"account"`
 	Limit   int    `json:"limit,omitempty"`
+	ClaimID string `json:"claimId"`
+	LeaseMS int    `json:"leaseMs"`
 }
 
 type pendingDelivery struct {
@@ -22,6 +25,7 @@ type pendingDelivery struct {
 	DeliveryType string          `json:"deliveryType"`
 	Payload      json.RawMessage `json:"payload"`
 	CreatedAt    time.Time       `json:"createdAt"`
+	ClaimID      string          `json:"claimId"`
 }
 
 type pendingFlushResult struct {
@@ -33,25 +37,96 @@ type pendingFlushResult struct {
 	LastError      string
 }
 
+type pendingFlushCall struct {
+	done   chan struct{}
+	result pendingFlushResult
+	err    error
+}
+
+var pendingFlushSingleflight = struct {
+	sync.Mutex
+	byAccount map[string]*pendingFlushCall
+}{byAccount: make(map[string]*pendingFlushCall)}
+
+type pendingFlushSchedule struct {
+	first time.Time
+	timer *time.Timer
+}
+
+var pendingFlushDebouncer = struct {
+	sync.Mutex
+	byAccount map[string]*pendingFlushSchedule
+}{byAccount: make(map[string]*pendingFlushSchedule)}
+
+func (s Server) schedulePendingFlush(account string, logger *log.Logger) {
+	account = strings.TrimSpace(account)
+	if account == "" {
+		return
+	}
+	pendingFlushDebouncer.Lock()
+	if scheduled := pendingFlushDebouncer.byAccount[account]; scheduled != nil {
+		if time.Since(scheduled.first) < 250*time.Millisecond {
+			scheduled.timer.Reset(25 * time.Millisecond)
+		}
+		pendingFlushDebouncer.Unlock()
+		return
+	}
+	scheduled := &pendingFlushSchedule{first: time.Now()}
+	scheduled.timer = time.AfterFunc(25*time.Millisecond, func() {
+		pendingFlushDebouncer.Lock()
+		if pendingFlushDebouncer.byAccount[account] == scheduled {
+			delete(pendingFlushDebouncer.byAccount, account)
+		}
+		pendingFlushDebouncer.Unlock()
+		if _, err := s.flushPending(account, logger); err != nil {
+			logger.Printf("cmpp inbound event=scheduled_pending_flush_failed account=%s error=%q", account, err.Error())
+		}
+	})
+	pendingFlushDebouncer.byAccount[account] = scheduled
+	pendingFlushDebouncer.Unlock()
+}
+
 func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushResult, error) {
+	account = strings.TrimSpace(account)
 	result := pendingFlushResult{Account: account}
 	if account == "" {
 		return result, nil
 	}
+	pendingFlushSingleflight.Lock()
+	if active := pendingFlushSingleflight.byAccount[account]; active != nil {
+		pendingFlushSingleflight.Unlock()
+		<-active.done
+		return active.result, active.err
+	}
+	call := &pendingFlushCall{done: make(chan struct{})}
+	pendingFlushSingleflight.byAccount[account] = call
+	pendingFlushSingleflight.Unlock()
+	call.result, call.err = s.flushPendingClaimed(account, logger)
+	pendingFlushSingleflight.Lock()
+	delete(pendingFlushSingleflight.byAccount, account)
+	close(call.done)
+	pendingFlushSingleflight.Unlock()
+	return call.result, call.err
+}
+
+func (s Server) flushPendingClaimed(account string, logger *log.Logger) (pendingFlushResult, error) {
+	result := pendingFlushResult{Account: account}
+	claimID := fmt.Sprintf("%s:%s:%d", s.gatewayInstanceID(), account, time.Now().UnixNano())
 	var deliveries []pendingDelivery
-	if err := s.post(context.Background(), "/gateway/events/downstream/pending", pendingDeliveryRequest{Account: account, Limit: 100}, &deliveries); err != nil {
+	if err := s.post(context.Background(), "/gateway/events/downstream/pending", pendingDeliveryRequest{Account: account, Limit: 100, ClaimID: claimID, LeaseMS: 30000}, &deliveries); err != nil {
 		logger.Printf("cmpp inbound pending delivery fetch failed account=%s err=%v", account, err)
 		result.LastError = err.Error()
 		return result, err
 	}
 	result.Deliveries = len(deliveries)
 	for _, delivery := range deliveries {
+		delivery.ClaimID = defaultString(delivery.ClaimID, claimID)
 		sendResult, err := s.pushPendingDelivery(account, delivery)
 		if err != nil {
 			result.FailedCount++
 			result.LastError = err.Error()
 			_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
-				"id": delivery.ID, "errorMessage": err.Error(), "failureType": "send_failed",
+				"id": delivery.ID, "claimId": delivery.ClaimID, "errorMessage": err.Error(), "failureType": "send_failed",
 				"connectionId": sendResult.ConnectionID, "sequenceId": sendResult.SequenceID,
 				"messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
 			}, nil)
@@ -63,6 +138,7 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
 		}
 		if sendResult.ReasonCode == "SUBMIT_RESPONSE_PENDING" {
 			result.WaitingCount++
+			_ = s.releasePendingClaim(delivery, sendResult)
 			continue
 		}
 		errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery")
@@ -74,8 +150,12 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
 			result.FailedCount++
 		}
 		result.LastError = errorMessage
+		if sendResult.Retryable {
+			_ = s.releasePendingClaim(delivery, sendResult)
+			continue
+		}
 		_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
-			"id": delivery.ID, "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
+			"id": delivery.ID, "claimId": delivery.ClaimID, "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
 			"failureType": failureType, "connectionId": sendResult.ConnectionID,
 			"sequenceId": sendResult.SequenceID, "messageId": sendResult.MessageID, "sentAt": sendResult.SentAt,
 		}, nil)
@@ -83,6 +163,14 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
 	return result, nil
 }
 
+func (s Server) releasePendingClaim(delivery pendingDelivery, sendResult DownstreamSendResult) error {
+	return s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{
+		"id": delivery.ID, "claimId": delivery.ClaimID,
+		"errorMessage": errorMessageWithCode(defaultString(sendResult.ErrorMessage, "Gateway released downstream delivery claim"), sendResult.ReasonCode),
+		"failureType":  "claim_released",
+	}, nil)
+}
+
 func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (DownstreamSendResult, error) {
 	switch delivery.DeliveryType {
 	case "receipt":
@@ -91,6 +179,7 @@ func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (D
 			return DownstreamSendResult{}, err
 		}
 		event.DeliveryID = delivery.ID
+		event.ClaimID = delivery.ClaimID
 		event.Account = defaultString(event.Account, account)
 		allowRecovery := !delivery.CreatedAt.IsZero() && time.Since(delivery.CreatedAt) >= 5*time.Second
 		return pushReceiptWithResult(event, allowRecovery)
@@ -100,6 +189,7 @@ func (s Server) pushPendingDelivery(account string, delivery pendingDelivery) (D
 			return DownstreamSendResult{}, err
 		}
 		event.DeliveryID = delivery.ID
+		event.ClaimID = delivery.ClaimID
 		event.Account = defaultString(event.Account, account)
 		return PushUplinkWithResult(event)
 	default:

gateway/internal/inbound/protocol_log.go

@@ -1,6 +1,7 @@
 package inbound
 
 import (
+	"cmpp-platform/gateway/internal/protocollog"
 	"context"
 	"fmt"
 	cmpp "github.com/bigwhite/gocmpp"
@@ -8,20 +9,7 @@ import (
 	"strconv"
 )
 
-type protocolLogEvent struct {
-	Protocol         string         `json:"protocol"`
-	Direction        string         `json:"direction"`
-	EventType        string         `json:"eventType"`
-	Status           string         `json:"status"`
-	TenantID         string         `json:"tenantId,omitempty"`
-	ApplicationID    string         `json:"applicationId,omitempty"`
-	Account          string         `json:"account,omitempty"`
-	MessageID        string         `json:"messageId,omitempty"`
-	GatewayMessageID string         `json:"gatewayMessageId,omitempty"`
-	Phone            string         `json:"phone,omitempty"`
-	ResultCode       string         `json:"resultCode,omitempty"`
-	Detail           map[string]any `json:"detail,omitempty"`
-}
+type protocolLogEvent = protocollog.Event
 
 func (s Server) submitResponseProtocolLogger(
 	account string,
@@ -71,6 +59,15 @@ func protocolSubmitResponseDetail(sequenceID uint32, sendErr error) map[string]a
 }
 
 func (s Server) emitProtocolLog(event protocolLogEvent) {
+	event.GatewayInstanceID = s.GatewayInstanceID
+	if s.ProtocolLogPublisher != nil {
+		go func() {
+			if err := s.ProtocolLogPublisher.Publish(context.Background(), event); err != nil {
+				log.Printf("cmpp inbound protocol log Redis publish failed account=%s message_id=%s error=%q", event.Account, event.MessageID, err)
+			}
+		}()
+		return
+	}
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout)
 		defer cancel()

gateway/internal/inbound/server.go

@@ -1,6 +1,8 @@
 package inbound
 
 import (
+	"cmpp-platform/gateway/internal/protocollog"
+	"context"
 	cmpp "github.com/bigwhite/gocmpp"
 	"io"
 	"log"
@@ -22,6 +24,9 @@ type Server struct {
 	RecoveryStore        RecoveryStore
 	GatewayInstanceID    string
 	MaxSubmitConcurrency int
+	ProtocolLogPublisher interface {
+		Publish(context.Context, protocollog.Event) error
+	}
 }
 
 func (s Server) ListenAndServe() error {

gateway/internal/inbound/server_test.go

@@ -930,6 +930,42 @@ func TestFlushOnlineAccountsFetchesPendingDeliveries(t *testing.T) {
 	}
 }
 
+func TestFlushPendingCoalescesConcurrentRequestsPerAccount(t *testing.T) {
+	var calls atomic.Int32
+	release := make(chan struct{})
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/api/gateway/events/downstream/pending" {
+			t.Fatalf("unexpected api path: %s", r.URL.Path)
+		}
+		calls.Add(1)
+		<-release
+		_ = json.NewEncoder(w).Encode([]pendingDelivery{})
+	}))
+	defer api.Close()
+
+	server := Server{APIBaseURL: api.URL + "/api", GatewayInstanceID: "gateway-a"}
+	results := make(chan error, 8)
+	for range 8 {
+		go func() {
+			_, err := server.flushPending("100001", log.Default())
+			results <- err
+		}()
+	}
+	deadline := time.Now().Add(time.Second)
+	for calls.Load() == 0 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	close(release)
+	for range 8 {
+		if err := <-results; err != nil {
+			t.Fatalf("flush failed: %v", err)
+		}
+	}
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("pending fetch calls = %d, want 1", got)
+	}
+}
+
 func TestRecoverPendingCandidatesFetchesPresenceAccounts(t *testing.T) {
 	resetDownstreamRegistry()
 	defer resetDownstreamRegistry()
@@ -1085,7 +1121,7 @@ func TestDownstreamDeliveryRequiresAcknowledgement(t *testing.T) {
 		messageID: "MSG-LONG-1", phoneNumber: "18821203795",
 		protocolLog: func(event protocolLogEvent) { protocolEvents <- event },
 	}
-	registerDownstreamAck(session, "delivery-1", 37, 9016479179509871733, time.Now().Add(time.Second))
+	registerDownstreamAck(session, "delivery-1", "", 37, 9016479179509871733, time.Now().Add(time.Second))
 	handleDownstreamAcknowledgement(conn, 37, 9016479179509871733, 0, log.Default())
 
 	select {
@@ -1165,6 +1201,7 @@ func TestSendDownstreamRejectsZeroMessageID(t *testing.T) {
 		&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)
@@ -1180,7 +1217,7 @@ func TestDownstreamDeliveryReportsAckTimeout(t *testing.T) {
 		conn: &cmpp.Conn{}, connectionID: "conn-1",
 		deliveryReport: func(event downstreamDeliveryLifecycleEvent) { events <- event },
 	}
-	registerDownstreamAck(session, "delivery-timeout", 38, 9017467844344255865, time.Now().Add(20*time.Millisecond))
+	registerDownstreamAck(session, "delivery-timeout", "", 38, 9017467844344255865, time.Now().Add(20*time.Millisecond))
 
 	select {
 	case event := <-events:

gateway/internal/inbound/submit.go

@@ -212,11 +212,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge
 		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())
-			}
-		}()
+		s.schedulePendingFlush(account, logger)
 	})
 	logger.Printf(
 		"cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s dest_count=%d accepted_count=%d result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s",

gateway/internal/metrics/metrics.go

@@ -39,6 +39,8 @@ var submitStageHistograms [5][2]durationHistogram
 type Snapshot struct {
 	UpstreamDesired          int
 	UpstreamConnected        int
+	UpstreamWindowConfigured int
+	UpstreamWindowInFlight   int
 	DownstreamConnected      int
 	SubmitWorkerUp           bool
 	SubmitWorkerConcurrency  int
@@ -55,6 +57,14 @@ type Snapshot struct {
 	ResultQueueAvailable     bool
 	ResultQueuePending       int64
 	ResultQueueLag           int64
+	CallbackBatchRequests    int64
+	CallbackBatchEvents      int64
+	CallbackBatchRetries     int64
+	CallbackDeadLetters      int64
+	ProtocolLogPublished     int64
+	ProtocolLogSampled       int64
+	ProtocolLogErrors        int64
+	ProtocolLogQueueLength   int64
 }
 
 type SnapshotFunc func(context.Context) Snapshot
@@ -121,6 +131,10 @@ func Handler(load SnapshotFunc) http.Handler {
 			writeDurationHistogram(response, "cmpp_gateway_submit_stage_duration_seconds", stage, &submitStageHistograms[index])
 		}
 		fmt.Fprintf(response, "# HELP cmpp_gateway_upstream_connections Desired and live supplier connections.\n# TYPE cmpp_gateway_upstream_connections gauge\ncmpp_gateway_upstream_connections{state=\"desired\"} %d\ncmpp_gateway_upstream_connections{state=\"connected\"} %d\n", snapshot.UpstreamDesired, snapshot.UpstreamConnected)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_upstream_window_slots Configured and in-flight supplier window slots.\n# TYPE cmpp_gateway_upstream_window_slots gauge\ncmpp_gateway_upstream_window_slots{state=\"configured\"} %d\ncmpp_gateway_upstream_window_slots{state=\"in_flight\"} %d\n", snapshot.UpstreamWindowConfigured, snapshot.UpstreamWindowInFlight)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_callback_batch_total Batch callback requests, events, retries and dead letters.\n# TYPE cmpp_gateway_callback_batch_total counter\ncmpp_gateway_callback_batch_total{result=\"requests\"} %d\ncmpp_gateway_callback_batch_total{result=\"events\"} %d\ncmpp_gateway_callback_batch_total{result=\"retries\"} %d\ncmpp_gateway_callback_batch_total{result=\"dead_letter\"} %d\n", snapshot.CallbackBatchRequests, snapshot.CallbackBatchEvents, snapshot.CallbackBatchRetries, snapshot.CallbackDeadLetters)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_protocol_log_total Protocol log Stream outcomes.\n# TYPE cmpp_gateway_protocol_log_total counter\ncmpp_gateway_protocol_log_total{result=\"published\"} %d\ncmpp_gateway_protocol_log_total{result=\"sampled_out\"} %d\ncmpp_gateway_protocol_log_total{result=\"error\"} %d\n", snapshot.ProtocolLogPublished, snapshot.ProtocolLogSampled, snapshot.ProtocolLogErrors)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_protocol_log_stream_length Current protocol log Stream length.\n# TYPE cmpp_gateway_protocol_log_stream_length gauge\ncmpp_gateway_protocol_log_stream_length %d\n", snapshot.ProtocolLogQueueLength)
 		fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected)
 		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp))
 		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_slots Configured and active bounded submit worker slots.\n# TYPE cmpp_gateway_submit_worker_slots gauge\ncmpp_gateway_submit_worker_slots{state=\"configured\"} %d\ncmpp_gateway_submit_worker_slots{state=\"in_flight\"} %d\n", snapshot.SubmitWorkerConcurrency, snapshot.SubmitWorkerInFlight)

gateway/internal/protocollog/publisher.go

@@ -0,0 +1,128 @@
+package protocollog
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"hash/fnv"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const defaultStream = "gateway.protocol.logs"
+
+type Event struct {
+	EventID           string         `json:"eventId"`
+	GatewayInstanceID string         `json:"gatewayInstanceId,omitempty"`
+	ConnectionID      string         `json:"connectionId,omitempty"`
+	Protocol          string         `json:"protocol"`
+	Direction         string         `json:"direction"`
+	EventType         string         `json:"eventType"`
+	Status            string         `json:"status"`
+	TenantID          string         `json:"tenantId,omitempty"`
+	ApplicationID     string         `json:"applicationId,omitempty"`
+	ChannelID         string         `json:"channelId,omitempty"`
+	Account           string         `json:"account,omitempty"`
+	MessageID         string         `json:"messageId,omitempty"`
+	SubmitID          string         `json:"submitId,omitempty"`
+	GatewayMessageID  string         `json:"gatewayMessageId,omitempty"`
+	Phone             string         `json:"phone,omitempty"`
+	ResultCode        string         `json:"resultCode,omitempty"`
+	DurationMs        int            `json:"durationMs,omitempty"`
+	PayloadBytes      int            `json:"payloadBytes,omitempty"`
+	Detail            map[string]any `json:"detail,omitempty"`
+	CreatedAt         time.Time      `json:"createdAt"`
+}
+
+type Publisher struct {
+	Redis             *redis.Client
+	Stream            string
+	GatewayInstanceID string
+	SuccessSampleRate int
+	MaxLen            int64
+	published         atomic.Int64
+	sampled           atomic.Int64
+	errors            atomic.Int64
+}
+
+func New(redisURL string) (*Publisher, error) {
+	if strings.TrimSpace(redisURL) == "" {
+		redisURL = "redis://127.0.0.1:6379"
+	}
+	options, err := redis.ParseURL(redisURL)
+	if err != nil {
+		return nil, err
+	}
+	return &Publisher{Redis: redis.NewClient(options)}, nil
+}
+
+func (p *Publisher) Publish(ctx context.Context, event Event) error {
+	if p == nil || p.Redis == nil {
+		return fmt.Errorf("protocol log Redis publisher is unavailable")
+	}
+	if event.CreatedAt.IsZero() {
+		event.CreatedAt = time.Now().UTC()
+	}
+	if event.GatewayInstanceID == "" {
+		event.GatewayInstanceID = p.GatewayInstanceID
+	}
+	if event.EventID == "" {
+		event.EventID = fmt.Sprintf("PL-%d-%d", event.CreatedAt.UnixNano(), p.published.Load()+p.sampled.Load()+1)
+	}
+	if !p.mustKeep(event) && !p.sample(event) {
+		p.sampled.Add(1)
+		return nil
+	}
+	data, err := json.Marshal(event)
+	if err != nil {
+		p.errors.Add(1)
+		return err
+	}
+	args := &redis.XAddArgs{Stream: p.stream(), Values: map[string]any{"data": string(data)}}
+	if p.MaxLen > 0 {
+		args.MaxLen = p.MaxLen
+		args.Approx = true
+	}
+	if err := p.Redis.XAdd(ctx, args).Err(); err != nil {
+		p.errors.Add(1)
+		return err
+	}
+	p.published.Add(1)
+	return nil
+}
+
+func (p *Publisher) mustKeep(event Event) bool {
+	if event.Status != "success" {
+		return true
+	}
+	code := strings.TrimSpace(event.ResultCode)
+	return code != "" && code != "0"
+}
+
+func (p *Publisher) sample(event Event) bool {
+	rate := p.SuccessSampleRate
+	if rate <= 0 {
+		rate = 10
+	}
+	if rate >= 100 {
+		return true
+	}
+	h := fnv.New32a()
+	_, _ = h.Write([]byte(event.ChannelID + "|" + event.MessageID + "|" + event.EventType + "|" + strconv.FormatInt(event.CreatedAt.UnixNano()/int64(time.Millisecond), 10)))
+	return int(h.Sum32()%100) < rate
+}
+
+func (p *Publisher) stream() string {
+	if strings.TrimSpace(p.Stream) != "" {
+		return p.Stream
+	}
+	return defaultStream
+}
+func (p *Publisher) StreamName() string { return p.stream() }
+func (p *Publisher) Counts() (int64, int64, int64) {
+	return p.published.Load(), p.sampled.Load(), p.errors.Load()
+}

gateway/internal/protocollog/publisher_test.go

@@ -0,0 +1,34 @@
+package protocollog
+
+import (
+	"context"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+func TestNewUsesLocalRedisWhenURLIsEmpty(t *testing.T) {
+	publisher, err := New("")
+	if err != nil {
+		t.Fatalf("New returned error: %v", err)
+	}
+	if publisher == nil || publisher.Redis == nil {
+		t.Fatal("expected an initialized Redis client")
+	}
+	if got := publisher.Redis.Options().Addr; got != "127.0.0.1:6379" {
+		t.Fatalf("expected local Redis fallback, got %q", got)
+	}
+}
+
+func TestFailuresAreNeverSampledOut(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	p := &Publisher{Redis: client, SuccessSampleRate: 1, MaxLen: 100}
+	if err := p.Publish(context.Background(), Event{Protocol: "cmpp", EventType: "submit", Status: "failed", ResultCode: "TIMEOUT"}); err != nil {
+		t.Fatal(err)
+	}
+	if client.XLen(context.Background(), p.StreamName()).Val() != 1 {
+		t.Fatal("failed protocol event must be retained")
+	}
+}

gateway/internal/queue/messages.go

@@ -70,6 +70,10 @@ type UpstreamConfig struct {
 	WindowSize               int    `json:"windowSize,omitempty"`
 	HeartbeatIntervalSeconds int    `json:"heartbeatIntervalSeconds,omitempty"`
 	HeartbeatMissThreshold   int    `json:"heartbeatMissThreshold,omitempty"`
+	ConnectionWarmupSeconds  int    `json:"connectionWarmupSeconds,omitempty"`
+	ConnectionDrainSeconds   int    `json:"connectionDrainTimeoutSeconds,omitempty"`
+	SubmitTimeoutSeconds     int    `json:"submitResponseTimeoutSeconds,omitempty"`
+	FailureCooldownSeconds   int    `json:"connectionFailureCooldownSeconds,omitempty"`
 }
 
 type Retry struct {
@@ -146,6 +150,10 @@ type ConnectChannelConfig struct {
 	WindowSize               int    `json:"windowSize,omitempty"`
 	HeartbeatIntervalSeconds int    `json:"heartbeatIntervalSeconds,omitempty"`
 	HeartbeatMissThreshold   int    `json:"heartbeatMissThreshold,omitempty"`
+	ConnectionWarmupSeconds  int    `json:"connectionWarmupSeconds,omitempty"`
+	ConnectionDrainSeconds   int    `json:"connectionDrainTimeoutSeconds,omitempty"`
+	SubmitTimeoutSeconds     int    `json:"submitResponseTimeoutSeconds,omitempty"`
+	FailureCooldownSeconds   int    `json:"connectionFailureCooldownSeconds,omitempty"`
 }
 
 type DisconnectChannelCommand struct {

gateway/internal/resultoutbox/batch.go

@@ -0,0 +1,170 @@
+package resultoutbox
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+type callbackBatch struct {
+	BatchID           string               `json:"batchId"`
+	GatewayInstanceID string               `json:"gatewayInstanceId"`
+	CreatedAt         time.Time            `json:"createdAt"`
+	Events            []callbackBatchEvent `json:"events"`
+}
+type callbackBatchEvent struct {
+	EventID string          `json:"eventId"`
+	Type    string          `json:"type"`
+	Payload json.RawMessage `json:"payload"`
+}
+type callbackBatchResponse struct {
+	BatchID string                `json:"batchId"`
+	Results []callbackEventResult `json:"results"`
+}
+type callbackEventResult struct {
+	EventID   string `json:"eventId"`
+	Accepted  bool   `json:"accepted"`
+	Retryable bool   `json:"retryable,omitempty"`
+	ErrorCode string `json:"errorCode,omitempty"`
+}
+
+func (o *Outbox) runBatches(ctx context.Context) error {
+	for ctx.Err() == nil {
+		messages, _, err := o.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{Stream: o.stream(), Group: o.group(), Consumer: o.consumer(), MinIdle: o.minIdle(), Start: "0-0", Count: int64(o.batchSize())}).Result()
+		if err != nil && !errors.Is(err, redis.Nil) {
+			return err
+		}
+		if len(messages) == 0 {
+			streams, readErr := o.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{Group: o.group(), Consumer: o.consumer(), Streams: []string{o.stream(), ">"}, Count: int64(o.batchSize()), Block: o.batchWait()}).Result()
+			if errors.Is(readErr, redis.Nil) {
+				continue
+			}
+			if readErr != nil {
+				return readErr
+			}
+			for _, stream := range streams {
+				messages = append(messages, stream.Messages...)
+			}
+		}
+		if len(messages) == 0 {
+			continue
+		}
+		if err := o.processBatch(ctx, messages); err != nil {
+			o.batchRetries.Add(int64(len(messages)))
+			sleep(ctx, 100*time.Millisecond)
+		}
+	}
+	return ctx.Err()
+}
+
+func (o *Outbox) processBatch(ctx context.Context, messages []redis.XMessage) error {
+	batch := callbackBatch{BatchID: fmt.Sprintf("CB-%d", time.Now().UnixNano()), GatewayInstanceID: o.GatewayInstanceID, CreatedAt: time.Now().UTC()}
+	byEvent := make(map[string]redis.XMessage, len(messages))
+	for _, message := range messages {
+		event, err := EventFromStreamValues(message.Values)
+		if err != nil {
+			_ = o.deadLetter(ctx, message, "INVALID_ENVELOPE", err.Error())
+			continue
+		}
+		batch.Events = append(batch.Events, callbackBatchEvent{EventID: event.EventID, Type: event.EventType, Payload: event.Payload})
+		byEvent[event.EventID] = message
+	}
+	if len(batch.Events) == 0 {
+		return nil
+	}
+	body, err := json.Marshal(batch)
+	if err != nil {
+		return err
+	}
+	if len(body) > 1024*1024 {
+		return fmt.Errorf("callback batch exceeds 1MB")
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(o.APIBaseURL, "/")+"/gateway/events/batch", bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	client := &http.Client{Timeout: o.httpTimeout()}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		data, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return fmt.Errorf("batch callback returned %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
+	}
+	var result callbackBatchResponse
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 1024*1024)).Decode(&result); err != nil {
+		return err
+	}
+	if result.BatchID != batch.BatchID {
+		return fmt.Errorf("callback batchId mismatch")
+	}
+	o.batchRequests.Add(1)
+	o.batchEvents.Add(int64(len(batch.Events)))
+	seen := make(map[string]struct{}, len(result.Results))
+	for _, item := range result.Results {
+		message, ok := byEvent[item.EventID]
+		if !ok {
+			continue
+		}
+		seen[item.EventID] = struct{}{}
+		if item.Accepted {
+			if err := o.ackDelete(ctx, message.ID); err != nil {
+				return err
+			}
+			continue
+		}
+		if !item.Retryable {
+			if err := o.deadLetter(ctx, message, item.ErrorCode, "non-retryable callback result"); err != nil {
+				return err
+			}
+		}
+	}
+	for eventID := range byEvent {
+		if _, ok := seen[eventID]; !ok {
+			return fmt.Errorf("batch response omitted event %s", eventID)
+		}
+	}
+	return nil
+}
+
+func (o *Outbox) ackDelete(ctx context.Context, id string) error {
+	return acknowledgeAndDeleteScript.Run(ctx, o.Redis, []string{o.stream()}, o.group(), id).Err()
+}
+func (o *Outbox) deadLetter(ctx context.Context, message redis.XMessage, code, detail string) error {
+	stream := o.DeadLetterStream
+	if stream == "" {
+		stream = o.stream() + ".dead"
+	}
+	pipe := o.Redis.TxPipeline()
+	pipe.XAdd(ctx, &redis.XAddArgs{Stream: stream, Values: map[string]any{"sourceId": message.ID, "errorCode": code, "detail": detail, "data": fmt.Sprint(message.Values["data"])}})
+	pipe.XAck(ctx, o.stream(), o.group(), message.ID)
+	pipe.XDel(ctx, o.stream(), message.ID)
+	if _, err := pipe.Exec(ctx); err != nil {
+		return err
+	}
+	o.deadLetters.Add(1)
+	return nil
+}
+func (o *Outbox) batchSize() int {
+	if o.BatchSize < 1 {
+		return 50
+	}
+	return min(o.BatchSize, 100)
+}
+func (o *Outbox) batchWait() time.Duration {
+	if o.BatchWait <= 0 {
+		return 10 * time.Millisecond
+	}
+	return o.BatchWait
+}

gateway/internal/resultoutbox/outbox.go

@@ -53,16 +53,25 @@ type Event struct {
 }
 
 type Outbox struct {
-	Redis       *redis.Client
-	Stream      string
-	Group       string
-	Consumer    string
-	DedupeTTL   time.Duration
-	APIBaseURL  string
-	HTTPTimeout time.Duration
-	Concurrency int
-	MinIdle     time.Duration
-	inFlight    atomic.Int64
+	Redis             *redis.Client
+	Stream            string
+	Group             string
+	Consumer          string
+	DedupeTTL         time.Duration
+	APIBaseURL        string
+	HTTPTimeout       time.Duration
+	Concurrency       int
+	MinIdle           time.Duration
+	BatchEnabled      bool
+	BatchSize         int
+	BatchWait         time.Duration
+	GatewayInstanceID string
+	DeadLetterStream  string
+	inFlight          atomic.Int64
+	batchRequests     atomic.Int64
+	batchEvents       atomic.Int64
+	batchRetries      atomic.Int64
+	deadLetters       atomic.Int64
 }
 
 func New(client *redis.Client) *Outbox {
@@ -145,6 +154,18 @@ func (o *Outbox) PublishSubmitResult(ctx context.Context, command queue.SubmitCo
 	return o.publish(ctx, event)
 }
 
+func (o *Outbox) PublishReceipt(ctx context.Context, event queue.ReceiptEvent) error {
+	return o.publishRaw(ctx, Event{SchemaVersion: queue.SchemaVersion, EventID: fmt.Sprintf("receipt:%s:%s:%d", event.GatewayMessageID, event.RawStatus, event.SequenceID), EventType: "receipt_intake", Path: "/gateway/events/receipt/intake", TraceID: event.TraceID, MessageID: event.MessageID, ChannelID: event.ChannelID, Payload: mustMarshal(event), CreatedAt: time.Now().UTC()})
+}
+
+func (o *Outbox) PublishUplink(ctx context.Context, event queue.UplinkEvent) error {
+	return o.publishRaw(ctx, Event{SchemaVersion: queue.SchemaVersion, EventID: fmt.Sprintf("uplink:%s:%d:%d", event.ChannelID, event.SequenceID, event.ReceivedAt.UnixNano()), EventType: "uplink", Path: "/gateway/events/uplink", TraceID: event.TraceID, MessageID: event.MessageID, ChannelID: event.ChannelID, Payload: mustMarshal(event), CreatedAt: time.Now().UTC()})
+}
+
+func mustMarshal(value any) json.RawMessage { data, _ := json.Marshal(value); return data }
+
+func (o *Outbox) publishRaw(ctx context.Context, event Event) error { return o.publish(ctx, event) }
+
 func (o *Outbox) publish(ctx context.Context, event Event) error {
 	if o.Redis == nil {
 		return fmt.Errorf("result Outbox Redis client is required")
@@ -211,10 +232,10 @@ func EventFromStreamValues(values map[string]interface{}) (Event, error) {
 	if err := json.Unmarshal([]byte(data), &event); err != nil {
 		return Event{}, err
 	}
-	if event.SchemaVersion != queue.SchemaVersion || event.EventID == "" || event.MessageID == "" || event.SubmitID == "" {
+	if event.SchemaVersion != queue.SchemaVersion || event.EventID == "" || event.MessageID == "" {
 		return Event{}, fmt.Errorf("invalid result Outbox envelope")
 	}
-	if event.Path != "/gateway/events/submit-result" && event.Path != "/gateway/events/submit-segment-result" {
+	if event.Path != "/gateway/events/submit-result" && event.Path != "/gateway/events/submit-segment-result" && event.Path != "/gateway/events/receipt/intake" && event.Path != "/gateway/events/uplink" && event.Path != "/gateway/events/dead-letter" {
 		return Event{}, fmt.Errorf("unsupported result Outbox path %q", event.Path)
 	}
 	if len(event.Payload) == 0 {
@@ -258,3 +279,6 @@ func (o *Outbox) dedupeKey(eventID string) string {
 func (o *Outbox) StreamName() string { return o.stream() }
 func (o *Outbox) GroupName() string  { return o.group() }
 func (o *Outbox) InFlight() int64    { return o.inFlight.Load() }
+func (o *Outbox) BatchCounts() (int64, int64, int64, int64) {
+	return o.batchRequests.Load(), o.batchEvents.Load(), o.batchRetries.Load(), o.deadLetters.Load()
+}

gateway/internal/resultoutbox/outbox_test.go

@@ -2,6 +2,7 @@ package resultoutbox
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
 	"sync/atomic"
@@ -32,6 +33,99 @@ func TestPublishSubmitSegmentIsIdempotent(t *testing.T) {
 	}
 }
 
+func TestBatchCallbackSendsMultipleEventsInOneRequest(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	var requests atomic.Int32
+	var eventCount atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		requests.Add(1)
+		var batch callbackBatch
+		if err := json.NewDecoder(request.Body).Decode(&batch); err != nil {
+			t.Errorf("decode batch: %v", err)
+			return
+		}
+		eventCount.Store(int32(len(batch.Events)))
+		result := callbackBatchResponse{BatchID: batch.BatchID}
+		for _, event := range batch.Events {
+			result.Results = append(result.Results, callbackEventResult{EventID: event.EventID, Accepted: true})
+		}
+		_ = json.NewEncoder(response).Encode(result)
+	}))
+	defer server.Close()
+	outbox := New(client)
+	outbox.APIBaseURL = server.URL
+	outbox.BatchEnabled = true
+	outbox.BatchSize = 50
+	outbox.BatchWait = 10 * time.Millisecond
+	outbox.GatewayInstanceID = "gateway-test"
+	command := testCommand()
+	for index := 1; index <= 2; index++ {
+		if err := outbox.PublishSubmitSegment(context.Background(), command, queue.SubmitSegmentResult{SegmentTotal: 2, SegmentIndex: index, SequenceID: uint32(index), GatewayMessageID: "88", SubmitStatus: "accepted"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- outbox.Run(ctx) }()
+	deadline := time.Now().Add(3 * time.Second)
+	for client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("batch did not drain")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	<-done
+	if requests.Load() != 1 || eventCount.Load() != 2 {
+		t.Fatalf("requests/events=%d/%d, want 1/2", requests.Load(), eventCount.Load())
+	}
+}
+
+func TestBatchCallbackReplaysWholeRequestAfterHTTPFailure(t *testing.T) {
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	var calls atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		var batch callbackBatch
+		_ = json.NewDecoder(request.Body).Decode(&batch)
+		if calls.Add(1) == 1 {
+			http.Error(response, "busy", http.StatusServiceUnavailable)
+			return
+		}
+		result := callbackBatchResponse{BatchID: batch.BatchID}
+		for _, event := range batch.Events {
+			result.Results = append(result.Results, callbackEventResult{EventID: event.EventID, Accepted: true})
+		}
+		_ = json.NewEncoder(response).Encode(result)
+	}))
+	defer server.Close()
+	outbox := New(client)
+	outbox.APIBaseURL = server.URL
+	outbox.BatchEnabled = true
+	outbox.BatchWait = 5 * time.Millisecond
+	outbox.MinIdle = 5 * time.Millisecond
+	outbox.GatewayInstanceID = "g"
+	if err := outbox.PublishSubmitSegment(context.Background(), testCommand(), queue.SubmitSegmentResult{SegmentTotal: 1, SegmentIndex: 1, SequenceID: 1, GatewayMessageID: "1", SubmitStatus: "accepted"}); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- outbox.Run(ctx) }()
+	deadline := time.Now().Add(3 * time.Second)
+	for client.XLen(context.Background(), outbox.StreamName()).Val() != 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("replayed batch did not drain")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	<-done
+	if calls.Load() < 2 {
+		t.Fatalf("calls=%d want replay", calls.Load())
+	}
+}
+
 func TestPublishAggregateAndCommandAckAreAtomicAndIdempotent(t *testing.T) {
 	mr := miniredis.RunT(t)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})

gateway/internal/resultoutbox/worker.go

@@ -46,6 +46,9 @@ func (o *Outbox) Run(ctx context.Context) error {
 	if err := o.ensureGroup(ctx); err != nil {
 		return err
 	}
+	if o.BatchEnabled {
+		return o.runBatches(ctx)
+	}
 	pool := newCallbackPool(ctx, o, o.concurrency())
 	defer pool.wait()
 	for {

gateway/internal/upstream/connection.go

@@ -1,6 +1,7 @@
 package upstream
 
 import (
+	"cmpp-platform/gateway/internal/protocollog"
 	"cmpp-platform/gateway/internal/queue"
 	"context"
 	"fmt"
@@ -16,24 +17,38 @@ import (
 // the connection must wake pending submitters before scheduling pool recovery.
 
 type connection struct {
-	channelID  string
-	config     queue.UpstreamConfig
-	index      int
-	pool       *connectionPool
-	apiBaseURL string
-	httpClient *http.Client
+	channelID            string
+	config               queue.UpstreamConfig
+	index                int
+	pool                 *connectionPool
+	apiBaseURL           string
+	httpClient           *http.Client
+	protocolLogPublisher interface {
+		Publish(context.Context, protocollog.Event) error
+	}
+	gatewayInstanceID string
+	eventPublisher    interface {
+		PublishReceipt(context.Context, queue.ReceiptEvent) error
+		PublishUplink(context.Context, queue.UplinkEvent) error
+	}
 
-	mu               sync.Mutex
-	sendMu           sync.Mutex
-	client           *cmpp.Client
-	window           chan struct{}
-	pending          map[uint32]chan submitPartResponse
-	tracker          map[uint64]queue.SubmitCommand
-	longUplink       map[string]*longUplinkAssembly
-	readOnce         sync.Once
-	closed           bool
-	heartbeatCancel  context.CancelFunc
-	heartbeatPending map[uint32]time.Time
+	mu                  sync.Mutex
+	sendMu              sync.Mutex
+	client              *cmpp.Client
+	window              chan struct{}
+	windowLimit         int
+	draining            bool
+	retired             bool
+	lastSubmitRTT       time.Duration
+	consecutiveFailures int
+	cooldownUntil       time.Time
+	pending             map[uint32]chan submitPartResponse
+	tracker             map[uint64]queue.SubmitCommand
+	longUplink          map[string]*longUplinkAssembly
+	readOnce            sync.Once
+	closed              bool
+	heartbeatCancel     context.CancelFunc
+	heartbeatPending    map[uint32]time.Time
 }
 
 type submitPartResponse struct {
@@ -161,6 +176,14 @@ func (c *connection) identity() string {
 	return fmt.Sprintf("%s-%d", c.channelID, c.index)
 }
 
+func (c *connection) retire() {
+	c.mu.Lock()
+	c.draining = true
+	c.retired = true
+	c.mu.Unlock()
+	c.handleConnectionLoss(fmt.Errorf("connection retired after drain"))
+}
+
 func (c *connection) close() {
 	c.handleConnectionLoss(fmt.Errorf("connection closed"))
 }
@@ -191,7 +214,7 @@ func (c *connection) handleConnectionLoss(err error) {
 		default:
 		}
 	}
-	if c.pool != nil {
+	if c.pool != nil && !c.retired {
 		status := "disconnected"
 		if c.pool.countActiveConnections() > 0 {
 			status = "reconnecting"

gateway/internal/upstream/deliver.go

@@ -88,7 +88,16 @@ func (c *connection) handleDeliver(pkt deliverPacket) error {
 			DeliveredAt:      time.Now().UTC(),
 			ConnectionID:     c.identity(),
 		}
-		if err := postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt/intake", event); err != nil {
+		if c.protocolLogPublisher != nil {
+			c.emitProtocolLog(protocolLogEvent{Protocol: "cmpp", Direction: "channel_to_platform", EventType: "deliver_receipt", Status: "success", ChannelID: channelID, Account: c.config.Account, MessageID: messageID, GatewayMessageID: fmt.Sprint(receipt.MsgId), Phone: strings.TrimSpace(receipt.DestTerminalId), ResultCode: strings.TrimSpace(receipt.Stat), Detail: map[string]any{"sequenceId": pkt.seqID}})
+		}
+		var publishErr error
+		if c.eventPublisher != nil {
+			publishErr = c.eventPublisher.PublishReceipt(context.Background(), event)
+		} else {
+			publishErr = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt/intake", event)
+		}
+		if err := publishErr; 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 {
@@ -121,7 +130,16 @@ func (c *connection) handleDeliver(pkt deliverPacket) error {
 		Content:     content,
 		ReceivedAt:  time.Now().UTC(),
 	}
-	if err := postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/uplink", event); err != nil {
+	if c.protocolLogPublisher != nil {
+		c.emitProtocolLog(protocolLogEvent{Protocol: "cmpp", Direction: "channel_to_platform", EventType: "deliver_uplink", Status: "success", ChannelID: c.channelID, Account: c.config.Account, MessageID: cmd.MessageID, Phone: strings.TrimSpace(pkt.srcTerminalID), Detail: map[string]any{"sequenceId": pkt.seqID}})
+	}
+	var publishErr error
+	if c.eventPublisher != nil {
+		publishErr = c.eventPublisher.PublishUplink(context.Background(), event)
+	} else {
+		publishErr = postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/uplink", event)
+	}
+	if err := publishErr; err != nil {
 		log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_uplink status=forward_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err)
 	} 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)

gateway/internal/upstream/flow_control.go

@@ -45,11 +45,20 @@ func (p *connectionPool) tryAcquireConnection() (*connection, func()) {
 	if len(p.conns) == 0 {
 		return nil, nil
 	}
-	for i := 0; i < len(p.conns); i++ {
-		index := (p.next + i) % len(p.conns)
-		conn := p.conns[index]
+	bestIndex := -1
+	bestInFlight := int(^uint(0) >> 1)
+	var bestRTT time.Duration
+	for index, conn := range p.conns {
+		inFlight, rtt, usable := conn.capacitySnapshot()
+		if !usable || inFlight > bestInFlight || (inFlight == bestInFlight && bestIndex >= 0 && bestRTT > 0 && rtt >= bestRTT) {
+			continue
+		}
+		bestIndex, bestInFlight, bestRTT = index, inFlight, rtt
+	}
+	if bestIndex >= 0 {
+		conn := p.conns[bestIndex]
 		if conn.tryAcquireWindow() {
-			p.next = (index + 1) % len(p.conns)
+			p.next = (bestIndex + 1) % len(p.conns)
 			return conn, conn.releaseWindow
 		}
 	}
@@ -57,8 +66,17 @@ func (p *connectionPool) tryAcquireConnection() (*connection, func()) {
 }
 
 func (c *connection) tryAcquireWindow() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	if c.window == nil {
-		c.window = make(chan struct{}, defaultWindowSize)
+		c.window = make(chan struct{}, maximumWindowSize)
+	}
+	limit := c.windowLimit
+	if limit < 1 {
+		limit = defaultWindowSize
+	}
+	if c.closed || c.draining || c.client == nil || len(c.window) >= limit {
+		return false
 	}
 	select {
 	case c.window <- struct{}{}:
@@ -68,6 +86,37 @@ func (c *connection) tryAcquireWindow() bool {
 	}
 }
 
+func (c *connection) capacitySnapshot() (int, time.Duration, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	limit := c.windowLimit
+	if limit < 1 {
+		limit = defaultWindowSize
+	}
+	inFlight := len(c.window)
+	return inFlight, c.lastSubmitRTT, !c.closed && !c.draining && c.client != nil && inFlight < limit && !time.Now().Before(c.cooldownUntil)
+}
+
+func (c *connection) markSubmitFailure() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.consecutiveFailures++
+	if c.consecutiveFailures >= 3 {
+		seconds := c.config.FailureCooldownSeconds
+		if seconds <= 0 {
+			seconds = 30
+		}
+		c.cooldownUntil = time.Now().Add(time.Duration(seconds) * time.Second)
+	}
+}
+
+func (c *connection) markSubmitSuccess() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.consecutiveFailures = 0
+	c.cooldownUntil = time.Time{}
+}
+
 func (c *connection) releaseWindow() {
 	if c.window == nil {
 		return

gateway/internal/upstream/manager.go

@@ -1,6 +1,7 @@
 package upstream
 
 import (
+	"cmpp-platform/gateway/internal/protocollog"
 	"cmpp-platform/gateway/internal/queue"
 	"context"
 	"fmt"
@@ -19,6 +20,8 @@ const (
 	defaultReconnectInitialDelay  = 5 * time.Second
 	defaultReconnectMaximumDelay  = 5 * time.Minute
 	defaultAuthReconnectDelay     = 5 * time.Minute
+	maximumConnections            = 8
+	maximumWindowSize             = 64
 )
 
 type Manager struct {
@@ -26,6 +29,14 @@ type Manager struct {
 	EventAPIBaseURL        string
 	HTTPClient             *http.Client
 	SubmitSegmentPublisher SubmitSegmentPublisher
+	EventPublisher         interface {
+		PublishReceipt(context.Context, queue.ReceiptEvent) error
+		PublishUplink(context.Context, queue.UplinkEvent) error
+	}
+	ProtocolLogPublisher interface {
+		Publish(context.Context, protocollog.Event) error
+	}
+	GatewayInstanceID string
 
 	mu    sync.Mutex
 	conns map[string]*connectionPool
@@ -72,8 +83,14 @@ func (m *Manager) ConnectChannel(ctx context.Context, command queue.ConnectChann
 		WindowSize:               command.Channel.WindowSize,
 		HeartbeatIntervalSeconds: command.Channel.HeartbeatIntervalSeconds,
 		HeartbeatMissThreshold:   command.Channel.HeartbeatMissThreshold,
+		ConnectionWarmupSeconds:  command.Channel.ConnectionWarmupSeconds,
+		ConnectionDrainSeconds:   command.Channel.ConnectionDrainSeconds,
+		SubmitTimeoutSeconds:     command.Channel.SubmitTimeoutSeconds,
+		FailureCooldownSeconds:   command.Channel.FailureCooldownSeconds,
 	})
-	if pool == nil || !pool.matches(config) {
+	if pool != nil && !pool.matches(config) && pool.sameEndpoint(config) {
+		pool.reconfigure(config)
+	} else if pool == nil || !pool.matches(config) {
 		if pool != nil {
 			pool.close()
 		}
@@ -130,7 +147,9 @@ func (m *Manager) connectionFor(cmd queue.SubmitCommand) (*connectionPool, error
 	m.ensureDefaultsLocked()
 
 	pool := m.conns[cmd.ChannelID]
-	if pool == nil || !pool.matches(cmd.Upstream) {
+	if pool != nil && !pool.matches(cmd.Upstream) && pool.sameEndpoint(cmd.Upstream) {
+		pool.reconfigure(cmd.Upstream)
+	} else if pool == nil || !pool.matches(cmd.Upstream) {
 		if pool != nil {
 			pool.close()
 		}
@@ -170,6 +189,27 @@ func (m *Manager) ConnectionCounts() (desired int, connected int) {
 	return desired, connected
 }
 
+func (m *Manager) WindowCounts() (configured int, inFlight int) {
+	m.mu.Lock()
+	pools := make([]*connectionPool, 0, len(m.conns))
+	for _, pool := range m.conns {
+		pools = append(pools, pool)
+	}
+	m.mu.Unlock()
+	for _, pool := range pools {
+		pool.mu.Lock()
+		conns := append([]*connection(nil), pool.conns...)
+		pool.mu.Unlock()
+		for _, conn := range conns {
+			conn.mu.Lock()
+			configured += max(1, conn.windowLimit)
+			inFlight += len(conn.window)
+			conn.mu.Unlock()
+		}
+	}
+	return configured, inFlight
+}
+
 func (m *Manager) newConnectionPool(channelID string, connectionID string, config queue.UpstreamConfig) *connectionPool {
 	eventAPIBaseURL := m.EventAPIBaseURL
 	if eventAPIBaseURL == "" {
@@ -184,8 +224,11 @@ func (m *Manager) newConnectionPool(channelID string, connectionID string, confi
 		reporter: func(ctx context.Context, state ConnectionState) error {
 			return m.post(ctx, "/admin/gateway/connections", state)
 		},
-		reconnectSignal: make(chan struct{}, 1),
-		stopCh:          make(chan struct{}),
+		protocolLogPublisher: m.ProtocolLogPublisher,
+		gatewayInstanceID:    m.GatewayInstanceID,
+		eventPublisher:       m.EventPublisher,
+		reconnectSignal:      make(chan struct{}, 1),
+		stopCh:               make(chan struct{}),
 	}
 }
 
@@ -202,6 +245,12 @@ func validateConnectChannelCommand(command queue.ConnectChannelCommand) error {
 	if command.Channel.Account == "" || command.Channel.PasswordCipher == "" {
 		return fmt.Errorf("account and passwordCipher are required")
 	}
+	if command.DesiredConnections < 1 || command.DesiredConnections > maximumConnections {
+		return fmt.Errorf("desiredConnections must be between 1 and %d", maximumConnections)
+	}
+	if command.Channel.WindowSize != 0 && (command.Channel.WindowSize < 1 || command.Channel.WindowSize > maximumWindowSize) {
+		return fmt.Errorf("windowSize must be between 1 and %d", maximumWindowSize)
+	}
 	return nil
 }
 
@@ -222,5 +271,17 @@ func normalizeUpstreamConfig(config queue.UpstreamConfig) queue.UpstreamConfig {
 	if config.HeartbeatMissThreshold <= 0 {
 		config.HeartbeatMissThreshold = defaultHeartbeatMissThreshold
 	}
+	if config.ConnectionWarmupSeconds < 0 {
+		config.ConnectionWarmupSeconds = 30
+	}
+	if config.ConnectionDrainSeconds <= 0 {
+		config.ConnectionDrainSeconds = 60
+	}
+	if config.SubmitTimeoutSeconds <= 0 {
+		config.SubmitTimeoutSeconds = 60
+	}
+	if config.FailureCooldownSeconds <= 0 {
+		config.FailureCooldownSeconds = 30
+	}
 	return config
 }

gateway/internal/upstream/pool.go

@@ -1,6 +1,7 @@
 package upstream
 
 import (
+	"cmpp-platform/gateway/internal/protocollog"
 	"cmpp-platform/gateway/internal/queue"
 	"context"
 	"net/http"
@@ -9,12 +10,20 @@ import (
 )
 
 type connectionPool struct {
-	channelID    string
-	connectionID string
-	config       queue.UpstreamConfig
-	apiBaseURL   string
-	httpClient   *http.Client
-	reporter     func(context.Context, ConnectionState) error
+	channelID            string
+	connectionID         string
+	config               queue.UpstreamConfig
+	apiBaseURL           string
+	httpClient           *http.Client
+	reporter             func(context.Context, ConnectionState) error
+	protocolLogPublisher interface {
+		Publish(context.Context, protocollog.Event) error
+	}
+	gatewayInstanceID string
+	eventPublisher    interface {
+		PublishReceipt(context.Context, queue.ReceiptEvent) error
+		PublishUplink(context.Context, queue.UplinkEvent) error
+	}
 
 	mu                     sync.Mutex
 	connectMu              sync.Mutex
@@ -34,6 +43,103 @@ func (p *connectionPool) matches(config queue.UpstreamConfig) bool {
 	return p.config == normalizeUpstreamConfig(config)
 }
 
+func (p *connectionPool) sameEndpoint(config queue.UpstreamConfig) bool {
+	next := normalizeUpstreamConfig(config)
+	current := p.config
+	return current.GatewayHost == next.GatewayHost && current.GatewayPort == next.GatewayPort &&
+		current.Account == next.Account && current.PasswordCipher == next.PasswordCipher && current.CMPPVersion == next.CMPPVersion
+}
+
+// reconfigure changes only runtime capacity on the existing pool. Existing
+// sequence mappings remain attached to their physical connection. Scale down
+// marks surplus connections draining before closing them; scale up is serialized.
+func (p *connectionPool) reconfigure(config queue.UpstreamConfig) {
+	next := normalizeUpstreamConfig(config)
+	p.mu.Lock()
+	p.config = next
+	for _, conn := range p.conns {
+		conn.mu.Lock()
+		conn.config = next
+		conn.windowLimit = next.WindowSize
+		conn.mu.Unlock()
+	}
+	p.mu.Unlock()
+	go p.reconcileCapacity()
+}
+
+func (p *connectionPool) reconcileCapacity() {
+	p.connectMu.Lock()
+	defer p.connectMu.Unlock()
+	if p.stopped() {
+		return
+	}
+
+	p.mu.Lock()
+	desired := max(1, min(maximumConnections, p.config.DesiredConnections))
+	if len(p.conns) > desired {
+		retiring := append([]*connection(nil), p.conns[desired:]...)
+		p.conns = p.conns[:desired]
+		for _, conn := range retiring {
+			conn.mu.Lock()
+			conn.draining = true
+			conn.mu.Unlock()
+		}
+		p.mu.Unlock()
+		deadline := time.Now().Add(time.Duration(p.config.ConnectionDrainSeconds) * time.Second)
+		for _, conn := range retiring {
+			for len(conn.window) > 0 && time.Now().Before(deadline) {
+				time.Sleep(10 * time.Millisecond)
+			}
+			conn.retire()
+		}
+		_ = p.reportState(context.Background(), "connected", nil)
+		return
+	}
+	p.mu.Unlock()
+
+	for {
+		p.mu.Lock()
+		if len(p.conns) >= desired || p.stopped() {
+			p.mu.Unlock()
+			break
+		}
+		index := len(p.conns)
+		config := p.config
+		p.mu.Unlock()
+		conn := p.newConnection(index, config)
+		if _, err := conn.ensureConnected(); err != nil {
+			p.scheduleReconnect(err)
+			_ = p.reportState(context.Background(), "failed", err)
+			return
+		}
+		p.mu.Lock()
+		p.conns = append(p.conns, conn)
+		p.mu.Unlock()
+		_ = p.reportState(context.Background(), "connected", nil)
+		if len(p.conns) < desired && config.ConnectionWarmupSeconds > 0 {
+			timer := time.NewTimer(time.Duration(config.ConnectionWarmupSeconds) * time.Second)
+			select {
+			case <-p.stopCh:
+				timer.Stop()
+				return
+			case <-timer.C:
+			}
+		}
+	}
+}
+
+func (p *connectionPool) newConnection(index int, config queue.UpstreamConfig) *connection {
+	return &connection{
+		channelID: p.channelID, config: config, index: index, pool: p,
+		apiBaseURL: p.apiBaseURL, httpClient: p.httpClient,
+		protocolLogPublisher: p.protocolLogPublisher, gatewayInstanceID: p.gatewayInstanceID,
+		eventPublisher: p.eventPublisher,
+		window:         make(chan struct{}, maximumWindowSize), windowLimit: config.WindowSize,
+		pending: make(map[uint32]chan submitPartResponse), tracker: make(map[uint64]queue.SubmitCommand),
+		longUplink: make(map[string]*longUplinkAssembly), heartbeatPending: make(map[uint32]time.Time),
+	}
+}
+
 func (p *connectionPool) ensureConnected() error {
 	p.connectMu.Lock()
 	defer p.connectMu.Unlock()
@@ -58,20 +164,12 @@ func (p *connectionPool) ensureConnected() error {
 			p.mu.Unlock()
 			break
 		}
-		index := len(p.conns)
-		conn := &connection{
-			channelID:        p.channelID,
-			config:           p.config,
-			index:            index,
-			pool:             p,
-			apiBaseURL:       p.apiBaseURL,
-			httpClient:       p.httpClient,
-			window:           make(chan struct{}, p.config.WindowSize),
-			pending:          make(map[uint32]chan submitPartResponse),
-			tracker:          make(map[uint64]queue.SubmitCommand),
-			longUplink:       make(map[string]*longUplinkAssembly),
-			heartbeatPending: make(map[uint32]time.Time),
+		if connectedAny {
+			p.mu.Unlock()
+			break
 		}
+		index := len(p.conns)
+		conn := p.newConnection(index, p.config)
 		p.mu.Unlock()
 
 		connected, err := conn.ensureConnected()
@@ -87,6 +185,7 @@ func (p *connectionPool) ensureConnected() error {
 	if connectedAny {
 		_ = p.reportState(context.Background(), "connected", nil)
 	}
+	go p.reconcileCapacity()
 	return nil
 }
 

gateway/internal/upstream/pool_test.go

@@ -2,7 +2,9 @@ package upstream
 
 import (
 	"context"
+	cmpp "github.com/bigwhite/gocmpp"
 	"testing"
+	"time"
 
 	"cmpp-platform/gateway/internal/queue"
 )
@@ -10,8 +12,8 @@ import (
 func TestConnectionPoolAcquiresAcrossConnections(t *testing.T) {
 	pool := &connectionPool{
 		conns: []*connection{
-			{window: make(chan struct{}, 1)},
-			{window: make(chan struct{}, 1)},
+			{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 1},
+			{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 1},
 		},
 	}
 
@@ -40,6 +42,51 @@ func TestConnectionPoolAcquiresAcrossConnections(t *testing.T) {
 	releaseSecond()
 }
 
+func TestPoolContinuesOnHealthyConnectionWhenPeerIsDraining(t *testing.T) {
+	draining := &connection{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 16, draining: true}
+	healthy := &connection{client: &cmpp.Client{}, window: make(chan struct{}, 64), windowLimit: 16}
+	pool := &connectionPool{conns: []*connection{draining, healthy}}
+	selected, release := pool.tryAcquireConnection()
+	if selected != healthy {
+		t.Fatal("expected healthy peer connection")
+	}
+	release()
+}
+
+func TestRuntimeCapacityMatrixAndSmoothScaleDown(t *testing.T) {
+	for _, connections := range []int{1, 2, 4, 8} {
+		for _, window := range []int{1, 16, 32, 64} {
+			config := normalizeUpstreamConfig(queue.UpstreamConfig{GatewayHost: "127.0.0.1", GatewayPort: 17890, Account: "a", PasswordCipher: "p", CMPPVersion: "3.0", DesiredConnections: connections, WindowSize: window})
+			if config.DesiredConnections != connections || config.WindowSize != window {
+				t.Fatalf("matrix normalized incorrectly: %+v", config)
+			}
+		}
+	}
+	pool := &connectionPool{channelID: "channel-1", connectionID: "primary", config: normalizeUpstreamConfig(queue.UpstreamConfig{DesiredConnections: 2, WindowSize: 16, ConnectionDrainSeconds: 1}), stopCh: make(chan struct{}), reconnectSignal: make(chan struct{}, 1)}
+	pool.conns = []*connection{{pool: pool, window: make(chan struct{}, 64), windowLimit: 16}, {pool: pool, window: make(chan struct{}, 64), windowLimit: 16}}
+	pool.reconfigure(queue.UpstreamConfig{DesiredConnections: 1, WindowSize: 32, ConnectionDrainSeconds: 1})
+	deadline := time.Now().Add(time.Second)
+	for {
+		pool.mu.Lock()
+		count := len(pool.conns)
+		first := pool.conns[0]
+		pool.mu.Unlock()
+		if count == 1 {
+			first.mu.Lock()
+			limit := first.windowLimit
+			first.mu.Unlock()
+			if limit != 32 {
+				t.Fatalf("window limit=%d want32", limit)
+			}
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("scale down did not complete")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
 func TestManagerDisconnectChannelRemovesPoolAndStopsReconnects(t *testing.T) {
 	pool := &connectionPool{
 		channelID:       "channel-1",

gateway/internal/upstream/protocol_log.go

@@ -1,6 +1,7 @@
 package upstream
 
 import (
+	"cmpp-platform/gateway/internal/protocollog"
 	"context"
 	"fmt"
 	cmpp "github.com/bigwhite/gocmpp"
@@ -8,22 +9,7 @@ import (
 	"strings"
 )
 
-type protocolLogEvent struct {
-	Protocol         string         `json:"protocol"`
-	Direction        string         `json:"direction"`
-	EventType        string         `json:"eventType"`
-	Status           string         `json:"status"`
-	TenantID         string         `json:"tenantId,omitempty"`
-	ApplicationID    string         `json:"applicationId,omitempty"`
-	ChannelID        string         `json:"channelId,omitempty"`
-	Account          string         `json:"account,omitempty"`
-	MessageID        string         `json:"messageId,omitempty"`
-	GatewayMessageID string         `json:"gatewayMessageId,omitempty"`
-	Phone            string         `json:"phone,omitempty"`
-	ResultCode       string         `json:"resultCode,omitempty"`
-	PayloadBytes     int            `json:"payloadBytes,omitempty"`
-	Detail           map[string]any `json:"detail,omitempty"`
-}
+type protocolLogEvent = protocollog.Event
 
 func (c *connection) emitDeliverResponse(pkt deliverPacket, responseErr error) {
 	status := "success"
@@ -71,6 +57,16 @@ func (c *connection) emitDeliverResponse(pkt deliverPacket, responseErr error) {
 }
 
 func (c *connection) emitProtocolLog(event protocolLogEvent) {
+	event.ConnectionID = c.identity()
+	event.GatewayInstanceID = c.gatewayInstanceID
+	if c.protocolLogPublisher != nil {
+		go func() {
+			if err := c.protocolLogPublisher.Publish(context.Background(), event); err != nil {
+				log.Printf("protocol log Redis publish failed channel_id=%s message_id=%s error=%q", event.ChannelID, event.MessageID, err)
+			}
+		}()
+		return
+	}
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), defaultHTTPTimeout)
 		defer cancel()

gateway/internal/upstream/submit.go

@@ -97,6 +97,12 @@ func (p *connectionPool) submit(
 }
 
 func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) {
+	startedAt := time.Now()
+	defer func() {
+		c.mu.Lock()
+		c.lastSubmitRTT = time.Since(startedAt)
+		c.mu.Unlock()
+	}()
 	rspCh := make(chan submitPartResponse, 1)
 	pkt := c.submitRequestPacket(cmd, part)
 
@@ -163,14 +169,20 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
 		c.mu.Unlock()
 	}()
 
-	waitCtx, cancel := context.WithTimeout(ctx, defaultSubmitTimeout)
+	timeout := time.Duration(c.config.SubmitTimeoutSeconds) * time.Second
+	if timeout <= 0 {
+		timeout = defaultSubmitTimeout
+	}
+	waitCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 	select {
 	case <-waitCtx.Done():
+		c.markSubmitFailure()
 		result := submitResult(cmd, seq, "", "timeout", "SUBMIT_TIMEOUT", waitCtx.Err().Error())
 		return seq, "", result, waitCtx.Err()
 	case rsp := <-rspCh:
 		if rsp.err != nil {
+			c.markSubmitFailure()
 			result := submitResult(cmd, seq, "", "timeout", "CONNECTION_LOST", rsp.err.Error())
 			return seq, "", result, rsp.err
 		}
@@ -203,6 +215,7 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
 			},
 		})
 		if rsp.result == 0 {
+			c.markSubmitSuccess()
 			c.mu.Lock()
 			c.tracker[rsp.msgID] = cmd
 			c.mu.Unlock()
@@ -345,6 +358,12 @@ func validateSubmitCommand(cmd queue.SubmitCommand) error {
 	if cmd.Upstream.Account == "" || cmd.Upstream.PasswordCipher == "" {
 		return fmt.Errorf("upstream account and passwordCipher are required")
 	}
+	if cmd.Upstream.DesiredConnections != 0 && (cmd.Upstream.DesiredConnections < 1 || cmd.Upstream.DesiredConnections > maximumConnections) {
+		return fmt.Errorf("upstream desiredConnections must be between 1 and %d", maximumConnections)
+	}
+	if cmd.Upstream.WindowSize != 0 && (cmd.Upstream.WindowSize < 1 || cmd.Upstream.WindowSize > maximumWindowSize) {
+		return fmt.Errorf("upstream windowSize must be between 1 and %d", maximumWindowSize)
+	}
 	if len(cmd.PhoneNumber) == 0 || len(cmd.Content) == 0 {
 		return fmt.Errorf("phoneNumber and content are required")
 	}