From 37c3a57aff44720c5961fbb9b7c7d23b4611f2d1 Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 3 Jul 2026 15:16:59 +1000 Subject: [PATCH 1/3] Retry before declaring connection not alive --- matterclient.go | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/matterclient.go b/matterclient.go index 60100ee..5c2d1de 100644 --- a/matterclient.go +++ b/matterclient.go @@ -568,11 +568,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 From feb62d1ecdfa3834a8ed0a19ead07c746bc8777d Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 3 Jul 2026 16:46:40 +1000 Subject: [PATCH 2/3] Switch to hybrid WebSocket and HTTP ping for checking connection is alive --- matterclient.go | 44 +++++++++++++++++++++++++------------------- 1 file changed, 25 insertions(+), 19 deletions(-) diff --git a/matterclient.go b/matterclient.go index 5c2d1de..e12746d 100644 --- a/matterclient.go +++ b/matterclient.go @@ -78,7 +78,8 @@ type Client struct { lruCache *lru.Cache aliveChan chan bool loginCancel context.CancelFunc - lastPong time.Time + + lastWsActivity time.Time } var Matterircd bool @@ -526,7 +527,7 @@ func (m *Client) wsConnect() { m.WsClient.Listen() - m.lastPong = time.Now() + m.lastWsActivity = time.Now() m.logger.Debug("WsClient: connected") @@ -535,25 +536,28 @@ 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 { + m.RLock() + timeSinceActivity := time.Since(m.lastWsActivity) + m.RUnlock() + + 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.Lock() + m.lastWsActivity = time.Now() + m.Unlock() return nil } @@ -634,6 +638,10 @@ func (m *Client) WsReceiver(ctx context.Context) { continue } + m.Lock() + m.lastWsActivity = time.Now() + m.Unlock() + m.logger.Debugf("WsReceiver event: %#v", event) msg := &Message{ @@ -651,13 +659,11 @@ func (m *Client) WsReceiver(ctx context.Context) { continue } - m.logger.Debugf("WsReceiver response: %#v", response) + m.Lock() + m.lastWsActivity = time.Now() + m.Unlock() - 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: From 58f3288de760e827230e15f7d534f433297d889c Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 3 Jul 2026 20:53:57 +1000 Subject: [PATCH 3/3] Remove mutex --- matterclient.go | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/matterclient.go b/matterclient.go index e12746d..ef8a750 100644 --- a/matterclient.go +++ b/matterclient.go @@ -11,6 +11,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -79,7 +80,7 @@ type Client struct { aliveChan chan bool loginCancel context.CancelFunc - lastWsActivity time.Time + lastWsActivity atomic.Int64 } var Matterircd bool @@ -527,7 +528,7 @@ func (m *Client) wsConnect() { m.WsClient.Listen() - m.lastWsActivity = time.Now() + m.lastWsActivity.Store(time.Now().Unix()) m.logger.Debug("WsClient: connected") @@ -540,9 +541,8 @@ func (m *Client) doCheckAlive() error { return fmt.Errorf("websocket listen error: %w", m.WsClient.ListenError) } - m.RLock() - timeSinceActivity := time.Since(m.lastWsActivity) - m.RUnlock() + 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)) @@ -555,9 +555,7 @@ func (m *Client) doCheckAlive() error { return fmt.Errorf("fallback HTTP ping failed: %w", err) } - m.Lock() - m.lastWsActivity = time.Now() - m.Unlock() + m.lastWsActivity.Store(time.Now().Unix()) return nil } @@ -638,9 +636,7 @@ func (m *Client) WsReceiver(ctx context.Context) { continue } - m.Lock() - m.lastWsActivity = time.Now() - m.Unlock() + m.lastWsActivity.Store(time.Now().Unix()) m.logger.Debugf("WsReceiver event: %#v", event) @@ -653,15 +649,18 @@ 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.Lock() - m.lastWsActivity = time.Now() - m.Unlock() + m.lastWsActivity.Store(time.Now().Unix()) m.logger.Debugf("WsReceiver response: %#v", response)