From d570c5f9614869aaa72f1f4e32cfecf00c6881b2 Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Fri, 10 Jul 2026 23:20:32 +0200 Subject: [PATCH 1/2] Detect stuck WebSocket connections via heartbeat and read timeout Previously listen() blocked in ReadMessage() with no read deadline, so a half-open connection (peer stops sending data but no TCP FIN/RST arrives) was never detected: onDisconnected never fired and the reconnection loop never engaged. This adds two complementary mechanisms: - Heartbeat: a goroutine sends a Home Assistant ping every PingInterval (default 30s); HA replies with pong. This generates traffic during quiet periods so a healthy-but-idle connection is not mistaken for a dead one. - Staleness detection: listen() sets a read deadline (ReadTimeout, default 60s) that resets on every received frame. If nothing arrives within the window, ReadMessage errors out and flows into the existing reconnection path. Both are configurable via hal.yaml (pingInterval, readTimeout). Also adds a writeMutex around all socket writes, since the heartbeat introduces a concurrent writer and gorilla/websocket forbids concurrent writes. Co-Authored-By: Claude Opus 4.8 (1M context) --- config.go | 9 +++ connection.go | 6 +- connection_heartbeat_test.go | 94 +++++++++++++++++++++++++++ hassws/client.go | 121 +++++++++++++++++++++++++++++++++-- hassws/message_types.go | 2 + hassws/server.go | 26 ++++++++ 6 files changed, 252 insertions(+), 6 deletions(-) create mode 100644 connection_heartbeat_test.go diff --git a/config.go b/config.go index af91112..c55978f 100644 --- a/config.go +++ b/config.go @@ -15,6 +15,15 @@ type Config struct { Location LocationConfig `yaml:"location"` DatabasePath string `yaml:"databasePath"` ReconnectInterval time.Duration `yaml:"reconnectInterval"` + + // PingInterval is how often to send a heartbeat ping to Home Assistant to + // keep the connection active. Defaults to 30s if unset. + PingInterval time.Duration `yaml:"pingInterval"` + + // ReadTimeout is how long to wait for any data from Home Assistant before + // treating the connection as stale and reconnecting. Should be larger than + // PingInterval. Defaults to 60s if unset. + ReadTimeout time.Duration `yaml:"readTimeout"` } type HomeAssistantConfig struct { diff --git a/connection.go b/connection.go index 63c3a13..2b008f4 100644 --- a/connection.go +++ b/connection.go @@ -58,8 +58,10 @@ func NewConnection(cfg Config) *Connection { } api := hassws.NewClient(hassws.ClientConfig{ - Host: cfg.HomeAssistant.Host, - Token: cfg.HomeAssistant.Token, + Host: cfg.HomeAssistant.Host, + Token: cfg.HomeAssistant.Token, + PingInterval: cfg.PingInterval, + ReadTimeout: cfg.ReadTimeout, }) // Set the database on the global logger diff --git a/connection_heartbeat_test.go b/connection_heartbeat_test.go new file mode 100644 index 0000000..fe3e4b1 --- /dev/null +++ b/connection_heartbeat_test.go @@ -0,0 +1,94 @@ +package hal + +import ( + "testing" + "time" + + "github.com/dansimau/hal/homeassistant" + "gotest.tools/v3/assert" +) + +// TestHeartbeatKeepsConnectionAlive verifies that heartbeat pings keep the +// connection alive during a quiet period with no state change events, i.e. the +// read timeout does not trip as long as pong responses keep arriving. +func TestHeartbeatKeepsConnectionAlive(t *testing.T) { + conn, server, cleanup := newClientServerWithConfig(t, Config{ + DatabasePath: ":memory:", + ReconnectInterval: 100 * time.Millisecond, + PingInterval: 50 * time.Millisecond, + ReadTimeout: 300 * time.Millisecond, + }) + defer cleanup() + + testEntity := NewEntity("test.entity") + conn.RegisterEntities(testEntity) + + // Wait well beyond the read timeout with no events. Because the server + // responds to heartbeat pings, the connection should stay alive and not + // reconnect. + time.Sleep(600 * time.Millisecond) + + assert.Equal(t, 0, conn.GetReconnectAttempts(), "connection should not reconnect while pings are answered") + + // Connection should still be functional. + server.SendEvent(homeassistant.Event{ + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{ + EntityID: "test.entity", + State: "on", + }, + }, + }) + + waitFor(t, "event received after quiet period", func() bool { + return testEntity.GetState().State == "on" + }, func() { + t.Logf("Entity state: %v", testEntity.GetState()) + }) +} + +// TestStaleConnectionTriggersReconnect verifies that a "stuck" connection (one +// that remains open but stops delivering data) is detected via the read timeout +// and triggers a reconnection. +func TestStaleConnectionTriggersReconnect(t *testing.T) { + conn, server, cleanup := newClientServerWithConfig(t, Config{ + DatabasePath: ":memory:", + ReconnectInterval: 100 * time.Millisecond, + PingInterval: 50 * time.Millisecond, + ReadTimeout: 200 * time.Millisecond, + }) + defer cleanup() + + testEntity := NewEntity("test.entity") + conn.RegisterEntities(testEntity) + + // Simulate a stuck connection: the socket stays open but the server stops + // replying to pings, so no data reaches the client. + server.SetRespondToPings(false) + + // The client should detect the staleness via the read timeout and reconnect. + waitForReconnection(t, conn, 1) + + // Restore normal behaviour so the reconnected connection stays healthy. + server.SetRespondToPings(true) + + // Wait for the subscription to be re-established, then verify events flow. + waitForEventSubscription(t, server, 1) + + server.SendEvent(homeassistant.Event{ + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{ + EntityID: "test.entity", + State: "recovered", + }, + }, + }) + + waitFor(t, "event received after stale reconnect", func() bool { + return testEntity.GetState().State == "recovered" + }, func() { + t.Logf("Entity state: %v", testEntity.GetState()) + }) +} diff --git a/hassws/client.go b/hassws/client.go index 8ac2a48..4a91f3b 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -15,6 +15,17 @@ import ( const readTimeoutSeconds = 3 +const ( + // defaultPingInterval is how often a heartbeat ping is sent to Home + // Assistant to keep the connection active and prove it is still alive. + defaultPingInterval = 30 * time.Second + + // defaultReadTimeout is how long we wait to receive any data (including + // pong responses) before considering the connection stale and closing it + // so it can be re-established. Must be larger than defaultPingInterval. + defaultReadTimeout = 60 * time.Second +) + // Connection states type connectionState int @@ -35,6 +46,11 @@ type Client struct { cfg ClientConfig conn *websocket.Conn + // writeMutex serializes writes to the websocket. Gorilla does not support + // concurrent writers, and the heartbeat goroutine writes concurrently with + // service calls and subscriptions. + writeMutex sync.Mutex + msgID atomic.Int64 // Each request has a unique ID and any response will have the same ID. To @@ -51,9 +67,26 @@ type Client struct { type ClientConfig struct { Host string Token string + + // PingInterval is how often to send a heartbeat ping to Home Assistant. + // Defaults to defaultPingInterval if zero. + PingInterval time.Duration + + // ReadTimeout is how long to wait for any data before considering the + // connection stale and reconnecting. Defaults to defaultReadTimeout if + // zero. Should be larger than PingInterval. + ReadTimeout time.Duration } func NewClient(config ClientConfig) *Client { + if config.PingInterval == 0 { + config.PingInterval = defaultPingInterval + } + + if config.ReadTimeout == 0 { + config.ReadTimeout = defaultReadTimeout + } + c := &Client{ cfg: config, responses: make(map[int]chan []byte), @@ -121,6 +154,10 @@ func (c *Client) authenticate() error { func (c *Client) Close() error { c.setState(stateDisconnected) + + c.writeMutex.Lock() + defer c.writeMutex.Unlock() + return c.conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "bye")) } @@ -151,17 +188,73 @@ func (c *Client) Connect() error { c.setState(stateConnected) - // Start listening for messages - go c.listen() + // done is closed when the listen loop for this connection exits. It is used + // to stop the heartbeat goroutine so it does not outlive the connection. + done := make(chan struct{}) + + // Start listening for messages and sending heartbeat pings. Both capture + // the current connection so they are unaffected by later reconnections + // reassigning c.conn. + go c.listen(conn, done) + go c.heartbeat(conn, done) return nil } +// heartbeat periodically sends a ping to Home Assistant to keep the connection +// active and generate traffic during quiet periods. The pong response resets +// the read deadline in listen(). If a ping fails to send, the connection is +// closed so listen() returns and reconnection is triggered. +func (c *Client) heartbeat(conn *websocket.Conn, done chan struct{}) { + ticker := time.NewTicker(c.cfg.PingInterval) + defer ticker.Stop() + + for { + select { + case <-done: + return + case <-ticker.C: + if err := c.ping(conn); err != nil { + logger.Error("Heartbeat ping failed, closing connection", "", "error", err) + + // Force the connection closed so the listen loop returns and + // the reconnection logic kicks in. + _ = conn.Close() + + return + } + } + } +} + +// ping sends a Home Assistant ping message on the given connection. +func (c *Client) ping(conn *websocket.Conn) error { + msg := CommandMessage{ + ID: c.nextMsgID(), + Type: MessageTypePing, + } + + msgBytes, err := json.Marshal(msg) + if err != nil { + return err + } + + logger.DebugJSON("Writing message", "", string(msgBytes)) + + c.writeMutex.Lock() + defer c.writeMutex.Unlock() + + return conn.WriteMessage(websocket.TextMessage, msgBytes) +} + // Listen for messages from the websocket and dispatch to listener channels. -func (c *Client) listen() { +func (c *Client) listen(conn *websocket.Conn, done chan struct{}) { logger.Info("Connection established", "") defer func() { + // Signal the heartbeat goroutine for this connection to stop. + close(done) + // Close all pending response channels to unblock waiting callers c.mutex.Lock() for msgID, ch := range c.responses { @@ -186,7 +279,17 @@ func (c *Client) listen() { }() for { - _, msgBytes, err := c.conn.ReadMessage() + // Reset the read deadline before each read. If no data (including + // heartbeat pong responses) arrives within ReadTimeout, ReadMessage + // returns an error, the loop exits and reconnection is triggered. This + // detects "stuck" connections that remain open but stop delivering data. + if err := conn.SetReadDeadline(time.Now().Add(c.cfg.ReadTimeout)); err != nil { + logger.Error("Error setting read deadline", "", "error", err) + + return + } + + _, msgBytes, err := conn.ReadMessage() if err != nil { if websocket.IsCloseError(err, websocket.CloseNormalClosure) { logger.Info("Received close message", "") @@ -213,6 +316,13 @@ func (c *Client) listen() { continue } + // Pong responses to our heartbeat pings have no registered listener; + // receiving one has already reset the read deadline above, so just + // move on without logging a spurious "no listeners" warning. + if msg.Type == MessageTypePong { + continue + } + c.mutex.RLock() responseListenerCh, ok := c.responses[msg.ID] c.mutex.RUnlock() @@ -262,6 +372,9 @@ func (c *Client) send(msg any) error { logger.DebugJSON("Writing message", "", string(msgBytes)) + c.writeMutex.Lock() + defer c.writeMutex.Unlock() + return c.conn.WriteMessage(websocket.TextMessage, msgBytes) } diff --git a/hassws/message_types.go b/hassws/message_types.go index eb3d881..57048e6 100644 --- a/hassws/message_types.go +++ b/hassws/message_types.go @@ -13,6 +13,8 @@ const ( MessageTypeCallService MessageType = "call_service" MessageTypeEvent MessageType = "event" MessageTypeGetStates MessageType = "get_states" + MessageTypePing MessageType = "ping" + MessageTypePong MessageType = "pong" MessageTypeResult MessageType = "result" MessageTypeStateChanged MessageType = "state_changed" MessageTypeSubscribeEvents MessageType = "subscribe_events" diff --git a/hassws/server.go b/hassws/server.go index b263682..658b5c9 100644 --- a/hassws/server.go +++ b/hassws/server.go @@ -7,6 +7,7 @@ import ( "net" "net/http" "sync" + "sync/atomic" "time" "github.com/dansimau/hal/homeassistant" @@ -33,6 +34,11 @@ type Server struct { // authenticatedUserID stores the user ID of the authenticated client authenticatedUserID string + // respondToPings controls whether the server replies to ping messages. + // Setting it to false simulates a "stuck" connection that remains open but + // stops delivering data, exercising the client's staleness detection. + respondToPings atomic.Bool + lock sync.RWMutex } @@ -44,6 +50,8 @@ func NewServer(validUsers map[string]string) (*Server, error) { validUsers: validUsers, } + server.respondToPings.Store(true) + server.http.Handler = http.HandlerFunc(server.handler) listener, err := net.Listen("tcp", "127.0.0.1:0") @@ -195,6 +203,17 @@ func (s *Server) listen() { // crash. Result: json.RawMessage("[]"), }) + + case MessageTypePing: + // Simulate a stuck connection by not responding when disabled. + if !s.respondToPings.Load() { + continue + } + + s.SendMessage(CommandMessage{ + ID: cmd.ID, + Type: MessageTypePong, + }) default: panic("[Server] Unknown message type: " + cmd.Type) } @@ -321,6 +340,13 @@ func (s *Server) SendEvent(event homeassistant.Event) { } } +// SetRespondToPings controls whether the server replies to ping messages. +// Passing false simulates a "stuck" connection that remains open but stops +// delivering data. +func (s *Server) SetRespondToPings(respond bool) { + s.respondToPings.Store(respond) +} + // GetSubscriptionCount returns the number of active event subscriptions. func (s *Server) GetSubscriptionCount() int { s.lock.RLock() From 54e252448abef9de205726fad0340b1072a81c6c Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Fri, 10 Jul 2026 23:36:12 +0200 Subject: [PATCH 2/2] Address PR review: bound writes, close stale sockets, derive read timeout - Close the captured socket in listen()'s deferred cleanup so a stale/ half-open connection is not leaked server-side across reconnects. - Add a write deadline to all websocket writes via a shared writeMessage helper, so a blackholed connection cannot wedge a write while holding writeMutex and block reconnection or shutdown. - Derive the default ReadTimeout from PingInterval (2x) and guard against ReadTimeout <= PingInterval, which would otherwise reconnect continuously on quiet connections when a large custom PingInterval is set. - Pass PingInterval/ReadTimeout to the `hal events` websocket client so it honours the configured heartbeat settings. - Add unit tests for the read-timeout derivation and guard logic. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/hal/commands/events.go | 6 ++-- hassws/client.go | 63 ++++++++++++++++++++++++---------- hassws/client_internal_test.go | 60 ++++++++++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 21 deletions(-) create mode 100644 hassws/client_internal_test.go diff --git a/cmd/hal/commands/events.go b/cmd/hal/commands/events.go index 6b7c37f..49e5217 100644 --- a/cmd/hal/commands/events.go +++ b/cmd/hal/commands/events.go @@ -51,8 +51,10 @@ func runEventsCommand(excludes []string, jqFilter string) error { } client := hassws.NewClient(hassws.ClientConfig{ - Host: cfg.HomeAssistant.Host, - Token: cfg.HomeAssistant.Token, + Host: cfg.HomeAssistant.Host, + Token: cfg.HomeAssistant.Token, + PingInterval: cfg.PingInterval, + ReadTimeout: cfg.ReadTimeout, }) if err := client.Connect(); err != nil { diff --git a/hassws/client.go b/hassws/client.go index 4a91f3b..96e3a9f 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -20,10 +20,11 @@ const ( // Assistant to keep the connection active and prove it is still alive. defaultPingInterval = 30 * time.Second - // defaultReadTimeout is how long we wait to receive any data (including - // pong responses) before considering the connection stale and closing it - // so it can be re-established. Must be larger than defaultPingInterval. - defaultReadTimeout = 60 * time.Second + // writeTimeout bounds how long a single websocket write may block. Without + // it, a half-open or blackholed connection could wedge a heartbeat write + // (or any other write) indefinitely while holding writeMutex, preventing + // reconnection and shutdown. + writeTimeout = 10 * time.Second ) // Connection states @@ -73,8 +74,9 @@ type ClientConfig struct { PingInterval time.Duration // ReadTimeout is how long to wait for any data before considering the - // connection stale and reconnecting. Defaults to defaultReadTimeout if - // zero. Should be larger than PingInterval. + // connection stale and reconnecting. When zero it defaults to twice + // PingInterval so pong responses keep the deadline alive. Must be larger + // than PingInterval, or a quiet connection would reconnect continuously. ReadTimeout time.Duration } @@ -83,8 +85,20 @@ func NewClient(config ClientConfig) *Client { config.PingInterval = defaultPingInterval } + // Derive the read timeout from the ping interval so a custom PingInterval + // larger than the default read timeout does not cause continuous + // reconnects on quiet connections. if config.ReadTimeout == 0 { - config.ReadTimeout = defaultReadTimeout + config.ReadTimeout = 2 * config.PingInterval + } + + // Guard against a misconfiguration where the read deadline would expire + // before a heartbeat pong could arrive. + if config.ReadTimeout <= config.PingInterval { + logger.Warn("ReadTimeout must be larger than PingInterval; adjusting", "", + "pingInterval", config.PingInterval, "readTimeout", config.ReadTimeout, "adjustedTo", 2*config.PingInterval) + + config.ReadTimeout = 2 * config.PingInterval } c := &Client{ @@ -155,10 +169,7 @@ func (c *Client) authenticate() error { func (c *Client) Close() error { c.setState(stateDisconnected) - c.writeMutex.Lock() - defer c.writeMutex.Unlock() - - return c.conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "bye")) + return c.writeMessage(c.conn, websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "bye")) } func (c *Client) shutdown() error { @@ -241,10 +252,7 @@ func (c *Client) ping(conn *websocket.Conn) error { logger.DebugJSON("Writing message", "", string(msgBytes)) - c.writeMutex.Lock() - defer c.writeMutex.Unlock() - - return conn.WriteMessage(websocket.TextMessage, msgBytes) + return c.writeMessage(conn, websocket.TextMessage, msgBytes) } // Listen for messages from the websocket and dispatch to listener channels. @@ -255,6 +263,11 @@ func (c *Client) listen(conn *websocket.Conn, done chan struct{}) { // Signal the heartbeat goroutine for this connection to stop. close(done) + // Close the underlying socket so a stale/half-open connection is not + // leaked server-side while we reconnect. Idempotent if already closed + // (e.g. by shutdown() or the heartbeat). + _ = conn.Close() + // Close all pending response channels to unblock waiting callers c.mutex.Lock() for msgID, ch := range c.responses { @@ -363,6 +376,21 @@ func (c *Client) read(target any) error { return json.Unmarshal(msgBytes, target) } +// writeMessage serializes writes to the given connection (gorilla forbids +// concurrent writers) and bounds each write with a deadline. Without the +// deadline, a half-open or blackholed connection could wedge a write +// indefinitely while holding writeMutex, blocking reconnection and shutdown. +func (c *Client) writeMessage(conn *websocket.Conn, messageType int, data []byte) error { + c.writeMutex.Lock() + defer c.writeMutex.Unlock() + + if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + return err + } + + return conn.WriteMessage(messageType, data) +} + // Send a message to the websocket. func (c *Client) send(msg any) error { msgBytes, err := json.Marshal(msg) @@ -372,10 +400,7 @@ func (c *Client) send(msg any) error { logger.DebugJSON("Writing message", "", string(msgBytes)) - c.writeMutex.Lock() - defer c.writeMutex.Unlock() - - return c.conn.WriteMessage(websocket.TextMessage, msgBytes) + return c.writeMessage(c.conn, websocket.TextMessage, msgBytes) } // Add a listener channel for a response to a specific sent message. diff --git a/hassws/client_internal_test.go b/hassws/client_internal_test.go new file mode 100644 index 0000000..e14a0ec --- /dev/null +++ b/hassws/client_internal_test.go @@ -0,0 +1,60 @@ +package hassws + +import ( + "testing" + "time" + + "gotest.tools/v3/assert" +) + +func TestNewClientReadTimeoutDefaults(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + pingInterval time.Duration + readTimeout time.Duration + wantPingInterval time.Duration + wantReadTimeout time.Duration + }{ + { + name: "defaults applied when unset", + wantPingInterval: defaultPingInterval, + wantReadTimeout: 2 * defaultPingInterval, + }, + { + name: "read timeout derived from custom ping interval", + pingInterval: 90 * time.Second, + wantPingInterval: 90 * time.Second, + wantReadTimeout: 180 * time.Second, + }, + { + name: "explicit read timeout above ping interval is kept", + pingInterval: 30 * time.Second, + readTimeout: 45 * time.Second, + wantPingInterval: 30 * time.Second, + wantReadTimeout: 45 * time.Second, + }, + { + name: "read timeout not larger than ping interval is bumped", + pingInterval: 30 * time.Second, + readTimeout: 20 * time.Second, + wantPingInterval: 30 * time.Second, + wantReadTimeout: 60 * time.Second, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + c := NewClient(ClientConfig{ + PingInterval: tc.pingInterval, + ReadTimeout: tc.readTimeout, + }) + + assert.Equal(t, tc.wantPingInterval, c.cfg.PingInterval) + assert.Equal(t, tc.wantReadTimeout, c.cfg.ReadTimeout) + }) + } +}