Improve connection aliveness check

matterclient.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -82,7 +83,9 @@ type Client struct {
 	lruCache    *lru.Cache
 	aliveChan   chan bool
 	loginCancel context.CancelFunc
-	lastPong    time.Time
+
+	lastWsActivity atomic.Int64
+	connectedAt    atomic.Int64
 }
 
 var Matterircd bool
@@ -569,40 +572,52 @@ func (m *Client) wsConnect() {
 
 	m.WsClient.Listen()
 
-	m.lastPong = time.Now()
+	m.lastWsActivity.Store(time.Now().Unix())
+	m.connectedAt.Store(time.Now().Unix())
 
 	m.logger.Debug("WsClient: connected")
 
 	// only start to parse WS messages when login is completely done
 	m.WsConnected = true
 }
 
-func (m *Client) doCheckAlive() error {
-	if _, _, err := m.Client.GetPing(context.TODO()); err != nil {
-		return err
+func (m *Client) doCheckAlive(ctx context.Context) error {
+	if m.WsClient != nil && m.WsClient.ListenError != nil {
+		return fmt.Errorf("websocket listen error: %w", m.WsClient.ListenError)
 	}
 
-	if m.reconnectBusy {
+	connectedUnix := m.connectedAt.Load()
+	uptime := time.Since(time.Unix(connectedUnix, 0)).Round(time.Second)
+	lastActiveUnix := m.lastWsActivity.Load()
+	timeSinceActivity := time.Since(time.Unix(lastActiveUnix, 0))
+
+	if timeSinceActivity < 20*time.Second {
+		m.logger.Tracef("websocket is active (last event %v ago; up %s), skipping ping", timeSinceActivity.Round(time.Second), uptime)
 		return nil
 	}
 
-	if m.WsClient.ListenError == nil {
-		m.WsClient.SendMessage("ping", nil)
-	} else {
-		m.logger.Errorf("got a listen error: %#v", m.WsClient.ListenError)
-
-		return m.WsClient.ListenError
+	if timeSinceActivity < 55*time.Second {
+		// Send a ping down the websocket to try to keep it active/alive
+		if m.WsClient != nil {
+			m.logger.Tracef("websocket has been quiet (last event %v ago; up %s), sending websocket ping", timeSinceActivity.Round(time.Second), uptime)
+			m.WsClient.SendMessage("ping", nil)
+			return nil
+		}
 	}
 
-	if time.Since(m.lastPong) > 90*time.Second {
-		return errors.New("no pong received in 90 seconds")
+	m.logger.Tracef("websocket has been quiet (last event %v ago; up %s), falling back to HTTP GetPing", timeSinceActivity.Round(time.Second), uptime)
+	if _, _, err := m.Client.GetPing(ctx); err != nil {
+		m.logger.Warnf("fallback HTTP ping failed (up %s): %s", uptime, err)
+		return fmt.Errorf("fallback HTTP ping failed (up %s): %w", uptime, err)
 	}
 
+	m.lastWsActivity.Store(time.Now().Unix())
+
 	return nil
 }
 
 func (m *Client) checkAlive(ctx context.Context) {
-	ticker := time.NewTicker(time.Second * 45)
+	ticker := time.NewTicker(time.Second * 30)
 
 	for {
 		select {
@@ -611,11 +626,25 @@ func (m *Client) checkAlive(ctx context.Context) {
 
 			return
 		case <-ticker.C:
+			var err error
+
 			// check if session still is valid
-			err := m.doCheckAlive()
+			for i := 0; i < 3; i++ { //nolint:intrange
+				err = m.doCheckAlive(ctx)
+				if err == nil {
+					break
+				}
+
+				if i < 2 {
+					m.logger.Warnf("alive check failed, retrying %d/3: %s", i+1, err)
+					time.Sleep(time.Second * 2)
+				}
+			}
+
 			if err != nil {
 				m.logger.Errorf("connection not alive: %s", err)
 				m.aliveChan <- false
+				continue
 			}
 
 			m.aliveChan <- true
@@ -632,7 +661,7 @@ func (m *Client) checkConnection(ctx context.Context) {
 			if !alive {
 				time.Sleep(time.Second * 10)
 
-				if m.doCheckAlive() != nil {
+				if m.doCheckAlive(ctx) != nil {
 					m.Reconnect()
 				}
 			}
@@ -663,6 +692,8 @@ func (m *Client) WsReceiver(ctx context.Context) {
 				continue
 			}
 
+			m.lastWsActivity.Store(time.Now().Unix())
+
 			m.logger.Debugf("WsReceiver event: %#v", event)
 
 			eventType := event.EventType()
@@ -680,19 +711,21 @@ func (m *Client) WsReceiver(ctx context.Context) {
 				m.parseMessage(msg)
 			}
 
-			m.MessageChan <- msg
+			select {
+			case m.MessageChan <- msg:
+				// Message sent successfully
+			default:
+				// Message channel buffer full, drop/discard
+				m.logger.Errorf("CRITICAL: MessageChan is blocked! Downstream processor is hung. Dropping event: %s", event.EventType())
+			}
 		case response := <-m.WsClient.ResponseChannel:
 			if response == nil || !response.IsValid() {
 				continue
 			}
 
 			m.logger.Debugf("WsReceiver response: %#v", response)
 
-			if text, ok := response.Data["text"].(string); ok {
-				if text == "pong" {
-					m.lastPong = time.Now()
-				}
-			}
+			m.lastWsActivity.Store(time.Now().Unix())
 
 			m.parseResponse(response)
 		case <-m.WsClient.PingTimeoutChannel: