Improve connection aliveness check

matterclient.go

@@ -11,6 +11,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -78,7 +79,8 @@ type Client struct {
 	lruCache    *lru.Cache
 	aliveChan   chan bool
 	loginCancel context.CancelFunc
-	lastPong    time.Time
+
+	lastWsActivity atomic.Int64
 }
 
 var Matterircd bool
@@ -526,7 +528,7 @@ func (m *Client) wsConnect() {
 
 	m.WsClient.Listen()
 
-	m.lastPong = time.Now()
+	m.lastWsActivity.Store(time.Now().Unix())
 
 	m.logger.Debug("WsClient: connected")
 
@@ -535,25 +537,25 @@ func (m *Client) wsConnect() {
 }
 
 func (m *Client) doCheckAlive() error {
-	if _, _, err := m.Client.GetPing(context.TODO()); err != nil {
-		return err
+	if m.WsClient != nil && m.WsClient.ListenError != nil {
+		return fmt.Errorf("websocket listen error: %w", m.WsClient.ListenError)
 	}
 
-	if m.reconnectBusy {
+	lastActiveUnix := m.lastWsActivity.Load()
+	timeSinceActivity := time.Since(time.Unix(lastActiveUnix, 0))
+
+	if timeSinceActivity < 90*time.Second {
+		m.logger.Debugf("websocket is active (last event %v ago), skipping HTTP ping", timeSinceActivity.Round(time.Second))
 		return nil
 	}
 
-	if m.WsClient.ListenError == nil {
-		m.WsClient.SendMessage("ping", nil)
-	} else {
-		m.logger.Errorf("got a listen error: %#v", m.WsClient.ListenError)
+	m.logger.Debug("websocket has been quiet, falling back to HTTP GetPing")
 
-		return m.WsClient.ListenError
+	if _, _, err := m.Client.GetPing(context.TODO()); err != nil {
+		return fmt.Errorf("fallback HTTP ping failed: %w", err)
 	}
 
-	if time.Since(m.lastPong) > 90*time.Second {
-		return errors.New("no pong received in 90 seconds")
-	}
+	m.lastWsActivity.Store(time.Now().Unix())
 
 	return nil
 }
@@ -568,11 +570,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++ {
+				err = m.doCheckAlive()
+				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
@@ -620,6 +636,8 @@ func (m *Client) WsReceiver(ctx context.Context) {
 				continue
 			}
 
+			m.lastWsActivity.Store(time.Now().Unix())
+
 			m.logger.Debugf("WsReceiver event: %#v", event)
 
 			msg := &Message{
@@ -631,19 +649,20 @@ func (m *Client) WsReceiver(ctx context.Context) {
 				m.parseMessage(msg)
 			}
 
-			m.MessageChan <- msg
+			select {
+			case m.MessageChan <- msg:
+				// Message sent successfully
+			case <-time.After(15*time.Second):
+				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)
+			m.lastWsActivity.Store(time.Now().Unix())
 
-			if text, ok := response.Data["text"].(string); ok {
-				if text == "pong" {
-					m.lastPong = time.Now()
-				}
-			}
+			m.logger.Debugf("WsReceiver response: %#v", response)
 
 			m.parseResponse(response)
 		case <-m.WsClient.PingTimeoutChannel: