feat: improve channel resilience and operations

gateway/internal/upstream/connection_loss_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 )
 
 func TestHandleConnectionLossNotifiesPendingSubmitters(t *testing.T) {
@@ -18,6 +19,8 @@ func TestHandleConnectionLossNotifiesPendingSubmitters(t *testing.T) {
 				reported = state
 				return nil
 			},
+			reconnectSignal: make(chan struct{}, 1),
+			stopCh:          make(chan struct{}),
 		},
 	}
 	conn.pool.conns = []*connection{conn}
@@ -45,6 +48,60 @@ func TestHandleConnectionLossNotifiesPendingSubmitters(t *testing.T) {
 	if reported.Status != "disconnected" || reported.CurrentConnections != 0 {
 		t.Fatalf("unexpected reported state: %+v", reported)
 	}
+	if conn.pool.reconnectCount != 1 || conn.pool.nextReconnectAt.IsZero() {
+		t.Fatalf("expected connection loss to schedule reconnect, got count=%d next=%v", conn.pool.reconnectCount, conn.pool.nextReconnectAt)
+	}
+	conn.pool.close()
+}
+
+func TestHeartbeatTimeoutClosesConnectionAndSchedulesReconnect(t *testing.T) {
+	pool := &connectionPool{
+		channelID:       "channel-1",
+		connectionID:    "channel-1:primary",
+		config:          normalizeUpstreamConfig(queueUpstreamConfigForTest()),
+		reconnectSignal: make(chan struct{}, 1),
+		stopCh:          make(chan struct{}),
+	}
+	conn := &connection{
+		pool:             pool,
+		pending:          make(map[uint32]chan submitPartResponse),
+		heartbeatPending: map[uint32]time.Time{1: time.Now(), 2: time.Now(), 3: time.Now()},
+	}
+	pool.conns = []*connection{conn}
+
+	if conn.sendHeartbeat() {
+		t.Fatal("expected heartbeat timeout to stop heartbeat loop")
+	}
+	if !conn.closed {
+		t.Fatal("expected heartbeat timeout to close the connection")
+	}
+	if pool.reconnectCount != 1 || pool.lastErrorCategory != "heartbeat_timeout" {
+		t.Fatalf("unexpected reconnect state: count=%d category=%s", pool.reconnectCount, pool.lastErrorCategory)
+	}
+	pool.close()
+}
+
+func TestReconnectDelayUsesCappedBackoffAndSlowAuthenticationRetry(t *testing.T) {
+	if got := reconnectDelay(1, "network"); got < 4*time.Second || got > 6*time.Second {
+		t.Fatalf("first reconnect delay = %v", got)
+	}
+	if got := reconnectDelay(6, "network"); got < 4*time.Minute || got > 6*time.Minute {
+		t.Fatalf("capped reconnect delay = %v", got)
+	}
+	if got := reconnectDelay(1, "authentication"); got != 5*time.Minute {
+		t.Fatalf("authentication reconnect delay = %v", got)
+	}
+}
+
+func TestHeartbeatResponseClearsOnlyMatchingRequest(t *testing.T) {
+	conn := &connection{heartbeatPending: map[uint32]time.Time{7: time.Now(), 8: time.Now()}}
+	conn.handleHeartbeatResponse(7)
+	if _, exists := conn.heartbeatPending[7]; exists {
+		t.Fatal("expected matching heartbeat request to be cleared")
+	}
+	if _, exists := conn.heartbeatPending[8]; !exists {
+		t.Fatal("expected unrelated heartbeat request to remain pending")
+	}
 }
 
 func TestTemporaryReadTimeoutDetection(t *testing.T) {

gateway/internal/upstream/manager.go

@@ -19,10 +19,15 @@ import (
 )
 
 const (
-	defaultConnectTimeout = 5 * time.Second
-	defaultSubmitTimeout  = 10 * time.Second
-	defaultHTTPTimeout    = 10 * time.Second
-	defaultWindowSize     = 16
+	defaultConnectTimeout         = 5 * time.Second
+	defaultSubmitTimeout          = 10 * time.Second
+	defaultHTTPTimeout            = 10 * time.Second
+	defaultWindowSize             = 16
+	defaultHeartbeatInterval      = 30 * time.Second
+	defaultHeartbeatMissThreshold = 3
+	defaultReconnectInitialDelay  = 5 * time.Second
+	defaultReconnectMaximumDelay  = 5 * time.Minute
+	defaultAuthReconnectDelay     = 5 * time.Minute
 )
 
 type Manager struct {
@@ -34,16 +39,19 @@ type Manager struct {
 }
 
 type ConnectionState struct {
-	ChannelID          string `json:"channelId"`
-	ConnectionID       string `json:"connectionId"`
-	Status             string `json:"status"`
-	DesiredConnections int    `json:"desiredConnections"`
-	CurrentConnections int    `json:"currentConnections"`
-	LastConnectedAt    string `json:"lastConnectedAt,omitempty"`
-	LastDisconnectedAt string `json:"lastDisconnectedAt,omitempty"`
-	LastHeartbeatAt    string `json:"lastHeartbeatAt,omitempty"`
-	ReconnectCount     int    `json:"reconnectCount,omitempty"`
-	LastError          string `json:"lastError,omitempty"`
+	ChannelID              string `json:"channelId"`
+	ConnectionID           string `json:"connectionId"`
+	Status                 string `json:"status"`
+	DesiredConnections     int    `json:"desiredConnections"`
+	CurrentConnections     int    `json:"currentConnections"`
+	LastConnectedAt        string `json:"lastConnectedAt,omitempty"`
+	LastDisconnectedAt     string `json:"lastDisconnectedAt,omitempty"`
+	LastHeartbeatAt        string `json:"lastHeartbeatAt,omitempty"`
+	ReconnectCount         int    `json:"reconnectCount,omitempty"`
+	LastReconnectAttemptAt string `json:"lastReconnectAttemptAt,omitempty"`
+	NextReconnectAt        string `json:"nextReconnectAt,omitempty"`
+	LastErrorCategory      string `json:"lastErrorCategory,omitempty"`
+	LastError              string `json:"lastError,omitempty"`
 }
 
 func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
@@ -86,13 +94,15 @@ func (m *Manager) ConnectChannel(ctx context.Context, command queue.ConnectChann
 	m.ensureDefaultsLocked()
 	pool := m.conns[command.ChannelID]
 	config := normalizeUpstreamConfig(queue.UpstreamConfig{
-		GatewayHost:        command.Channel.GatewayHost,
-		GatewayPort:        command.Channel.GatewayPort,
-		Account:            command.Channel.Account,
-		PasswordCipher:     command.Channel.PasswordCipher,
-		CMPPVersion:        command.Channel.CMPPVersion,
-		DesiredConnections: command.DesiredConnections,
-		WindowSize:         16,
+		GatewayHost:              command.Channel.GatewayHost,
+		GatewayPort:              command.Channel.GatewayPort,
+		Account:                  command.Channel.Account,
+		PasswordCipher:           command.Channel.PasswordCipher,
+		CMPPVersion:              command.Channel.CMPPVersion,
+		DesiredConnections:       command.DesiredConnections,
+		WindowSize:               command.Channel.WindowSize,
+		HeartbeatIntervalSeconds: command.Channel.HeartbeatIntervalSeconds,
+		HeartbeatMissThreshold:   command.Channel.HeartbeatMissThreshold,
 	})
 	if pool == nil || !pool.matches(config) {
 		if pool != nil {
@@ -103,16 +113,47 @@ func (m *Manager) ConnectChannel(ctx context.Context, command queue.ConnectChann
 	}
 	m.mu.Unlock()
 
+	pool.startSupervisor()
 	if err := pool.ensureConnected(); err != nil {
+		if pool.stopped() {
+			return pool.snapshotState("disconnected", nil), nil
+		}
+		pool.scheduleReconnect(err)
 		_ = pool.reportState(ctx, "failed", err)
-		m.mu.Lock()
-		delete(m.conns, command.ChannelID)
-		m.mu.Unlock()
 		return pool.snapshotState("failed", err), nil
 	}
+	if pool.stopped() {
+		return pool.snapshotState("disconnected", nil), nil
+	}
+	pool.resetReconnectState()
 	return pool.snapshotState("connected", nil), nil
 }
 
+func (m *Manager) DisconnectChannel(ctx context.Context, command queue.DisconnectChannelCommand) (ConnectionState, error) {
+	if command.ChannelID == "" || command.ConnectionID == "" {
+		return ConnectionState{}, fmt.Errorf("channelId and connectionId are required")
+	}
+	m.mu.Lock()
+	m.ensureDefaultsLocked()
+	pool := m.conns[command.ChannelID]
+	delete(m.conns, command.ChannelID)
+	m.mu.Unlock()
+	if pool != nil {
+		pool.close()
+		state := pool.snapshotState("disconnected", nil)
+		_ = pool.reportState(ctx, "disconnected", nil)
+		return state, nil
+	}
+	return ConnectionState{
+		ChannelID:          command.ChannelID,
+		ConnectionID:       command.ConnectionID,
+		Status:             "disconnected",
+		DesiredConnections: 0,
+		CurrentConnections: 0,
+		LastDisconnectedAt: time.Now().UTC().Format(time.RFC3339Nano),
+	}, nil
+}
+
 func (m *Manager) connectionFor(cmd queue.SubmitCommand) (*connectionPool, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -127,10 +168,12 @@ func (m *Manager) connectionFor(cmd queue.SubmitCommand) (*connectionPool, error
 		pool = m.newConnectionPool(cmd.ChannelID, defaultChannelConnectionID(cmd.ChannelID), normalizeUpstreamConfig(cmd.Upstream))
 		m.conns[cmd.ChannelID] = pool
 	}
+	pool.startSupervisor()
 	if err := pool.ensureConnected(); err != nil {
-		delete(m.conns, cmd.ChannelID)
+		pool.scheduleReconnect(err)
 		return nil, err
 	}
+	pool.resetReconnectState()
 	return pool, nil
 }
 
@@ -161,6 +204,8 @@ 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{}),
 	}
 }
 
@@ -172,9 +217,18 @@ type connectionPool struct {
 	httpClient   *http.Client
 	reporter     func(context.Context, ConnectionState) error
 
-	mu    sync.Mutex
-	conns []*connection
-	next  int
+	mu                     sync.Mutex
+	connectMu              sync.Mutex
+	conns                  []*connection
+	next                   int
+	reconnectSignal        chan struct{}
+	stopCh                 chan struct{}
+	stopOnce               sync.Once
+	supervisorOnce         sync.Once
+	reconnectCount         int
+	lastReconnectAttemptAt time.Time
+	nextReconnectAt        time.Time
+	lastErrorCategory      string
 }
 
 func (p *connectionPool) matches(config queue.UpstreamConfig) bool {
@@ -182,6 +236,8 @@ func (p *connectionPool) matches(config queue.UpstreamConfig) bool {
 }
 
 func (p *connectionPool) ensureConnected() error {
+	p.connectMu.Lock()
+	defer p.connectMu.Unlock()
 	desired := p.config.DesiredConnections
 	if desired <= 0 {
 		desired = 1
@@ -189,29 +245,38 @@ func (p *connectionPool) ensureConnected() error {
 	connectedAny := false
 	for {
 		p.mu.Lock()
+		active := p.conns[:0]
+		for _, existing := range p.conns {
+			existing.mu.Lock()
+			usable := existing.client != nil && !existing.closed
+			existing.mu.Unlock()
+			if usable {
+				active = append(active, existing)
+			}
+		}
+		p.conns = active
 		if len(p.conns) >= desired {
 			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),
+			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),
 		}
 		p.mu.Unlock()
 
 		connected, err := conn.ensureConnected()
 		if err != nil {
-			conn.close()
-			p.close()
 			return err
 		}
 
@@ -226,6 +291,118 @@ func (p *connectionPool) ensureConnected() error {
 	return nil
 }
 
+func (p *connectionPool) startSupervisor() {
+	p.supervisorOnce.Do(func() {
+		go p.superviseReconnects()
+	})
+}
+
+func (p *connectionPool) stopped() bool {
+	select {
+	case <-p.stopCh:
+		return true
+	default:
+		return false
+	}
+}
+
+func (p *connectionPool) signalReconnect() {
+	select {
+	case <-p.stopCh:
+		return
+	default:
+	}
+	select {
+	case p.reconnectSignal <- struct{}{}:
+	default:
+	}
+}
+
+func (p *connectionPool) scheduleReconnect(stateErr error) {
+	select {
+	case <-p.stopCh:
+		return
+	default:
+	}
+	p.mu.Lock()
+	now := time.Now().UTC()
+	if !p.nextReconnectAt.IsZero() && p.nextReconnectAt.After(now) {
+		p.mu.Unlock()
+		return
+	}
+	p.reconnectCount++
+	p.lastReconnectAttemptAt = now
+	p.lastErrorCategory = connectionErrorCategory(stateErr)
+	delay := reconnectDelay(p.reconnectCount, p.lastErrorCategory)
+	p.nextReconnectAt = p.lastReconnectAttemptAt.Add(delay)
+	p.mu.Unlock()
+	p.signalReconnect()
+}
+
+func (p *connectionPool) resetReconnectState() {
+	p.mu.Lock()
+	p.reconnectCount = 0
+	p.lastReconnectAttemptAt = time.Time{}
+	p.nextReconnectAt = time.Time{}
+	p.lastErrorCategory = ""
+	p.mu.Unlock()
+	p.signalReconnect()
+}
+
+func (p *connectionPool) superviseReconnects() {
+	for {
+		select {
+		case <-p.stopCh:
+			return
+		case <-p.reconnectSignal:
+		}
+		for {
+			p.mu.Lock()
+			next := p.nextReconnectAt
+			p.mu.Unlock()
+			if next.IsZero() {
+				break
+			}
+			timer := time.NewTimer(time.Until(next))
+			select {
+			case <-p.stopCh:
+				if !timer.Stop() {
+					<-timer.C
+				}
+				return
+			case <-p.reconnectSignal:
+				if !timer.Stop() {
+					<-timer.C
+				}
+				continue
+			case <-timer.C:
+			}
+			_ = p.reportState(context.Background(), "reconnecting", nil)
+			p.mu.Lock()
+			p.lastReconnectAttemptAt = time.Now().UTC()
+			p.mu.Unlock()
+			if err := p.ensureConnected(); err != nil {
+				select {
+				case <-p.stopCh:
+					return
+				default:
+				}
+				p.scheduleReconnect(err)
+				_ = p.reportState(context.Background(), "failed", err)
+				continue
+			}
+			select {
+			case <-p.stopCh:
+				return
+			default:
+			}
+			p.resetReconnectState()
+			_ = p.reportState(context.Background(), "connected", nil)
+			break
+		}
+	}
+}
+
 func (p *connectionPool) submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
 	parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
 	if err != nil {
@@ -314,6 +491,11 @@ func (p *connectionPool) tryAcquireConnection() (*connection, func()) {
 }
 
 func (p *connectionPool) close() {
+	p.connectMu.Lock()
+	defer p.connectMu.Unlock()
+	p.stopOnce.Do(func() {
+		close(p.stopCh)
+	})
 	p.mu.Lock()
 	conns := p.conns
 	p.conns = nil
@@ -339,6 +521,16 @@ func (p *connectionPool) snapshotState(status string, stateErr error) Connection
 		DesiredConnections: p.config.DesiredConnections,
 		CurrentConnections: p.countActiveConnections(),
 	}
+	p.mu.Lock()
+	state.ReconnectCount = p.reconnectCount
+	state.LastErrorCategory = p.lastErrorCategory
+	if !p.lastReconnectAttemptAt.IsZero() {
+		state.LastReconnectAttemptAt = p.lastReconnectAttemptAt.Format(time.RFC3339Nano)
+	}
+	if !p.nextReconnectAt.IsZero() {
+		state.NextReconnectAt = p.nextReconnectAt.Format(time.RFC3339Nano)
+	}
+	p.mu.Unlock()
 	if state.DesiredConnections <= 0 {
 		state.DesiredConnections = 1
 	}
@@ -346,6 +538,8 @@ func (p *connectionPool) snapshotState(status string, stateErr error) Connection
 	case "connected":
 		state.LastConnectedAt = now
 		state.LastHeartbeatAt = now
+	case "heartbeat":
+		state.LastHeartbeatAt = now
 	case "disconnected", "failed":
 		state.LastDisconnectedAt = now
 	}
@@ -378,15 +572,17 @@ type connection struct {
 	apiBaseURL string
 	httpClient *http.Client
 
-	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
+	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
 }
 
 type submitPartResponse struct {
@@ -415,7 +611,11 @@ func (c *connection) ensureConnected() (bool, error) {
 	}
 	c.client = client
 	c.closed = false
+	c.heartbeatPending = make(map[uint32]time.Time)
+	heartbeatCtx, cancelHeartbeat := context.WithCancel(context.Background())
+	c.heartbeatCancel = cancelHeartbeat
 	go c.readLoop()
+	go c.heartbeatLoop(heartbeatCtx)
 	return true, nil
 }
 
@@ -423,8 +623,17 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
 	rspCh := make(chan submitPartResponse, 1)
 	pkt := c.submitRequestPacket(cmd, part)
 
+	c.mu.Lock()
+	client := c.client
+	closed := c.closed
+	c.mu.Unlock()
+	if closed || client == nil {
+		err := fmt.Errorf("supplier connection is not available")
+		result := submitResult(cmd, 0, "", "timeout", "CONNECTION_LOST", err.Error())
+		return 0, "", result, err
+	}
 	c.sendMu.Lock()
-	seq, err := c.client.SendReqPkt(pkt)
+	seq, err := client.SendReqPkt(pkt)
 	c.sendMu.Unlock()
 	if err != nil {
 		c.close()
@@ -576,10 +785,17 @@ func (c *connection) releaseWindow() {
 
 func (c *connection) readLoop() {
 	for {
-		pkt, err := c.client.RecvAndUnpackPkt(time.Second)
+		c.mu.Lock()
+		client := c.client
+		closed := c.closed
+		c.mu.Unlock()
+		if closed || client == nil {
+			return
+		}
+		pkt, err := client.RecvAndUnpackPkt(time.Second)
 		if err != nil {
 			c.mu.Lock()
-			closed := c.closed
+			closed = c.closed
 			c.mu.Unlock()
 			if closed {
 				return
@@ -588,7 +804,7 @@ func (c *connection) readLoop() {
 				continue
 			}
 			c.handleConnectionLoss(err)
-			continue
+			return
 		}
 		switch p := pkt.(type) {
 		case *cmpp.Cmpp2SubmitRspPkt:
@@ -606,17 +822,86 @@ func (c *connection) readLoop() {
 				ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result}
 			}
 		case *cmpp.Cmpp2DeliverReqPkt:
-			_ = c.client.SendRspPkt(&cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
+			_ = c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
 			c.handleDeliver(deliverPacketFromCMPP2(p))
 		case *cmpp.Cmpp3DeliverReqPkt:
-			_ = c.client.SendRspPkt(&cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
+			_ = c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId)
 			c.handleDeliver(deliverPacketFromCMPP3(p))
 		case *cmpp.CmppActiveTestReqPkt:
-			_ = c.client.SendRspPkt(&cmpp.CmppActiveTestRspPkt{}, p.SeqId)
+			_ = c.sendResponse(client, &cmpp.CmppActiveTestRspPkt{}, p.SeqId)
+			_ = c.pool.reportState(context.Background(), "heartbeat", nil)
+		case *cmpp.CmppActiveTestRspPkt:
+			c.handleHeartbeatResponse(p.SeqId)
+			_ = c.pool.reportState(context.Background(), "heartbeat", nil)
 		}
 	}
 }
 
+func (c *connection) sendResponse(client *cmpp.Client, packet cmpp.Packer, sequenceID uint32) error {
+	c.sendMu.Lock()
+	defer c.sendMu.Unlock()
+	return client.SendRspPkt(packet, sequenceID)
+}
+
+func (c *connection) heartbeatLoop(ctx context.Context) {
+	interval := time.Duration(c.config.HeartbeatIntervalSeconds) * time.Second
+	if interval <= 0 {
+		interval = defaultHeartbeatInterval
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if !c.sendHeartbeat() {
+				return
+			}
+		}
+	}
+}
+
+func (c *connection) sendHeartbeat() bool {
+	threshold := c.config.HeartbeatMissThreshold
+	if threshold <= 0 {
+		threshold = defaultHeartbeatMissThreshold
+	}
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return false
+	}
+	if len(c.heartbeatPending) >= threshold {
+		c.mu.Unlock()
+		c.handleConnectionLoss(fmt.Errorf("heartbeat timeout after %d unanswered ACTIVE_TEST requests", threshold))
+		return false
+	}
+	if c.client == nil {
+		c.mu.Unlock()
+		return false
+	}
+	c.sendMu.Lock()
+	seq, err := c.client.SendReqPkt(&cmpp.CmppActiveTestReqPkt{})
+	c.sendMu.Unlock()
+	if err != nil {
+		c.mu.Unlock()
+		c.handleConnectionLoss(fmt.Errorf("send ACTIVE_TEST: %w", err))
+		return false
+	}
+	if !c.closed {
+		c.heartbeatPending[seq] = time.Now().UTC()
+	}
+	c.mu.Unlock()
+	return true
+}
+
+func (c *connection) handleHeartbeatResponse(sequenceID uint32) {
+	c.mu.Lock()
+	delete(c.heartbeatPending, sequenceID)
+	c.mu.Unlock()
+}
+
 type deliverPacket struct {
 	seqID            uint32
 	msgID            uint64
@@ -757,8 +1042,13 @@ func (c *connection) handleConnectionLoss(err error) {
 		return
 	}
 	c.closed = true
+	if c.heartbeatCancel != nil {
+		c.heartbeatCancel()
+		c.heartbeatCancel = nil
+	}
 	pending := c.pending
 	c.pending = make(map[uint32]chan submitPartResponse)
+	c.heartbeatPending = make(map[uint32]time.Time)
 	if c.client != nil {
 		c.client.Disconnect()
 		c.client = nil
@@ -774,8 +1064,9 @@ func (c *connection) handleConnectionLoss(err error) {
 	if c.pool != nil {
 		status := "disconnected"
 		if c.pool.countActiveConnections() > 0 {
-			status = "connected"
+			status = "reconnecting"
 		}
+		c.pool.scheduleReconnect(err)
 		_ = c.pool.reportState(context.Background(), status, err)
 	}
 }
@@ -864,9 +1155,57 @@ func normalizeUpstreamConfig(config queue.UpstreamConfig) queue.UpstreamConfig {
 	if config.WindowSize <= 0 {
 		config.WindowSize = defaultWindowSize
 	}
+	if config.HeartbeatIntervalSeconds <= 0 {
+		config.HeartbeatIntervalSeconds = int(defaultHeartbeatInterval / time.Second)
+	}
+	if config.HeartbeatMissThreshold <= 0 {
+		config.HeartbeatMissThreshold = defaultHeartbeatMissThreshold
+	}
 	return config
 }
 
+func connectionErrorCategory(err error) string {
+	if err == nil {
+		return ""
+	}
+	message := strings.ToLower(err.Error())
+	switch {
+	case strings.Contains(message, "auth"), strings.Contains(message, "password"), strings.Contains(message, "credential"):
+		return "authentication"
+	case strings.Contains(message, "heartbeat"):
+		return "heartbeat_timeout"
+	case strings.Contains(message, "timeout"):
+		return "timeout"
+	default:
+		return "network"
+	}
+}
+
+func reconnectDelay(attempt int, category string) time.Duration {
+	if category == "authentication" {
+		return defaultAuthReconnectDelay
+	}
+	if attempt < 1 {
+		attempt = 1
+	}
+	delays := []time.Duration{
+		defaultReconnectInitialDelay,
+		15 * time.Second,
+		30 * time.Second,
+		time.Minute,
+		2 * time.Minute,
+		defaultReconnectMaximumDelay,
+	}
+	delay := delays[min(attempt-1, len(delays)-1)]
+	// Deterministic ±10% jitter prevents a large set of channels from retrying together.
+	offsetPercent := (attempt*37)%21 - 10
+	delay += time.Duration(int64(delay) * int64(offsetPercent) / 100)
+	if delay < time.Second {
+		return time.Second
+	}
+	return delay
+}
+
 func isTemporaryReadTimeout(err error) bool {
 	var netErr net.Error
 	return errors.As(err, &netErr) && netErr.Timeout()

gateway/internal/upstream/pool_test.go

@@ -1,6 +1,7 @@
 package upstream
 
 import (
+	"context"
 	"testing"
 
 	"cmpp-platform/gateway/internal/queue"
@@ -39,6 +40,38 @@ func TestConnectionPoolAcquiresAcrossConnections(t *testing.T) {
 	releaseSecond()
 }
 
+func TestManagerDisconnectChannelRemovesPoolAndStopsReconnects(t *testing.T) {
+	pool := &connectionPool{
+		channelID:       "channel-1",
+		connectionID:    "channel-1:primary",
+		config:          normalizeUpstreamConfig(queueUpstreamConfigForTest()),
+		reconnectSignal: make(chan struct{}, 1),
+		stopCh:          make(chan struct{}),
+	}
+	manager := &Manager{conns: map[string]*connectionPool{"channel-1": pool}}
+
+	state, err := manager.DisconnectChannel(context.Background(), queue.DisconnectChannelCommand{
+		MessageType:  queue.MessageTypeDisconnectChannel,
+		ChannelID:    "channel-1",
+		ConnectionID: "channel-1:primary",
+	})
+
+	if err != nil {
+		t.Fatalf("disconnect channel: %v", err)
+	}
+	if state.Status != "disconnected" || state.CurrentConnections != 0 {
+		t.Fatalf("unexpected state: %+v", state)
+	}
+	if _, exists := manager.conns["channel-1"]; exists {
+		t.Fatal("expected channel pool to be removed")
+	}
+	select {
+	case <-pool.stopCh:
+	default:
+		t.Fatal("expected reconnect supervisor to be stopped")
+	}
+}
+
 func TestNormalizeUpstreamConfigDefaults(t *testing.T) {
 	config := normalizeUpstreamConfig(queueUpstreamConfigForTest())
 	if config.DesiredConnections != 1 {
@@ -47,6 +80,9 @@ func TestNormalizeUpstreamConfigDefaults(t *testing.T) {
 	if config.WindowSize != defaultWindowSize {
 		t.Fatalf("WindowSize = %d, want %d", config.WindowSize, defaultWindowSize)
 	}
+	if config.HeartbeatIntervalSeconds != 30 || config.HeartbeatMissThreshold != 3 {
+		t.Fatalf("unexpected heartbeat defaults: interval=%d threshold=%d", config.HeartbeatIntervalSeconds, config.HeartbeatMissThreshold)
+	}
 }
 
 func queueUpstreamConfigForTest() queue.UpstreamConfig {

gateway/internal/upstream/reconnect_integration_test.go

@@ -0,0 +1,119 @@
+package upstream
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"cmpp-platform/gateway/internal/queue"
+
+	cmpp "github.com/bigwhite/gocmpp"
+)
+
+func TestFailedSupplierConnectionReconnectsWhenEndpointRecovers(t *testing.T) {
+	address := reserveSupplierAddress(t)
+	states := make(chan ConnectionState, 16)
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var state ConnectionState
+		if err := json.NewDecoder(r.Body).Decode(&state); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		states <- state
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer api.Close()
+
+	host, portText, err := net.SplitHostPort(address)
+	if err != nil {
+		t.Fatalf("split address: %v", err)
+	}
+	var port int
+	if _, err := fmt.Sscanf(portText, "%d", &port); err != nil {
+		t.Fatalf("parse port: %v", err)
+	}
+	manager := &Manager{APIBaseURL: api.URL}
+	command := queue.ConnectChannelCommand{
+		SchemaVersion:      queue.SchemaVersion,
+		MessageType:        queue.MessageTypeConnectChannel,
+		ChannelID:          "channel-reconnect",
+		ConnectionID:       "channel-reconnect:primary",
+		DesiredConnections: 1,
+		Channel: queue.ConnectChannelConfig{
+			GatewayHost:              host,
+			GatewayPort:              port,
+			Account:                  "sp",
+			PasswordCipher:           "secret",
+			CMPPVersion:              "3.0",
+			HeartbeatIntervalSeconds: 1,
+			HeartbeatMissThreshold:   3,
+		},
+	}
+	initial, err := manager.ConnectChannel(context.Background(), command)
+	if err != nil {
+		t.Fatalf("initial connect command: %v", err)
+	}
+	if initial.Status != "failed" {
+		t.Fatalf("initial state = %s, want failed", initial.Status)
+	}
+
+	go func() {
+		_ = cmpp.ListenAndServe(address, cmpp.V30, time.Hour, 3, nil,
+			cmpp.HandlerFunc(func(response *cmpp.Response, packet *cmpp.Packet, _ *log.Logger) (bool, error) {
+				if _, ok := packet.Packer.(*cmpp.CmppConnReqPkt); ok {
+					response.Packer.(*cmpp.Cmpp3ConnRspPkt).Version = 0x30
+				}
+				return false, nil
+			}),
+		)
+	}()
+	time.Sleep(100 * time.Millisecond)
+
+	manager.mu.Lock()
+	pool := manager.conns[command.ChannelID]
+	manager.mu.Unlock()
+	pool.mu.Lock()
+	pool.nextReconnectAt = time.Now()
+	pool.mu.Unlock()
+	pool.signalReconnect()
+
+	deadline := time.After(4 * time.Second)
+	connected := false
+	for {
+		select {
+		case state := <-states:
+			if state.Status == "connected" && state.CurrentConnections == 1 {
+				connected = true
+			}
+			if connected && state.Status == "heartbeat" && state.LastHeartbeatAt != "" {
+				_, _ = manager.DisconnectChannel(context.Background(), queue.DisconnectChannelCommand{
+					MessageType:  queue.MessageTypeDisconnectChannel,
+					ChannelID:    command.ChannelID,
+					ConnectionID: command.ConnectionID,
+				})
+				return
+			}
+		case <-deadline:
+			t.Fatal("timed out waiting for automatic supplier reconnection and active heartbeat")
+		}
+	}
+}
+
+func reserveSupplierAddress(t *testing.T) string {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("reserve address: %v", err)
+	}
+	address := listener.Addr().String()
+	if err := listener.Close(); err != nil {
+		t.Fatalf("close reserved listener: %v", err)
+	}
+	return address
+}