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/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..96e3a9f 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -15,6 +15,18 @@ 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 + + // 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 type connectionState int @@ -35,6 +47,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 +68,39 @@ 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. 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 } func NewClient(config ClientConfig) *Client { + if config.PingInterval == 0 { + 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 = 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{ cfg: config, responses: make(map[int]chan []byte), @@ -121,7 +168,8 @@ func (c *Client) authenticate() error { func (c *Client) Close() error { c.setState(stateDisconnected) - 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 { @@ -151,17 +199,75 @@ 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)) + + return c.writeMessage(conn, 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 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 { @@ -186,7 +292,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 +329,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() @@ -253,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) @@ -262,7 +400,7 @@ func (c *Client) send(msg any) error { logger.DebugJSON("Writing message", "", string(msgBytes)) - 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) + }) + } +} 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()