diff --git a/connection.go b/connection.go index 2b008f4..57422ef 100644 --- a/connection.go +++ b/connection.go @@ -274,19 +274,34 @@ func (h *Connection) StateChangeEvent(event hassws.EventMessage) { logger.InfoDiff("State changed for", event.Event.EventData.EntityID, cmp.Diff(event.Event.EventData.OldState, event.Event.EventData.NewState)) - if event.Event.EventData.NewState != nil { - entity.SetState(*event.Event.EventData.NewState) + // A nil NewState carries no state to apply (e.g. entity removed); leave the + // stored state untouched rather than overwriting it with nil. + if event.Event.EventData.NewState == nil { + return } + newState := *event.Event.EventData.NewState + + // Drop stale updates that arrive out of order. Home Assistant events can be + // reordered in transit, and a reconnect resync can race with live events; an + // older state must never overwrite a newer one. Equal timestamps still apply. + current := entity.GetState() + if !current.LastUpdated.IsZero() && newState.LastUpdated.Before(current.LastUpdated) { + logger.Debug("Skipping stale state update", event.Event.EventData.EntityID) + + return + } + + entity.SetState(newState) + // Update database asynchronously entityID := event.Event.EventData.EntityID - newState := event.Event.EventData.NewState h.db.EnqueueWrite(func(db *gorm.DB) error { return db.Clauses(clause.OnConflict{ UpdateAll: true, }).Create(&store.Entity{ ID: entityID, - State: newState, + State: &newState, }).Error }) diff --git a/connection_test.go b/connection_test.go index 96d009a..2d66277 100644 --- a/connection_test.go +++ b/connection_test.go @@ -2,13 +2,16 @@ package hal_test import ( "context" + "sync" "sync/atomic" "testing" + "time" "github.com/dansimau/hal" "github.com/dansimau/hal/homeassistant" "github.com/dansimau/hal/testutil" "github.com/davecgh/go-spew/spew" + "gotest.tools/v3/assert" ) func TestConnection(t *testing.T) { @@ -84,3 +87,203 @@ func TestLoopProtection(t *testing.T) { spew.Dump(automationTriggered.Load()) }) } + +// TestEventsProcessedInOrder verifies that state change events are delivered to +// the handler in the order Home Assistant sent them. A regression here (e.g. +// dispatching each frame in its own goroutine) would let a rapid on/off sequence +// be reordered, leaving local state out of sync with reality. +func TestEventsProcessedInOrder(t *testing.T) { + t.Parallel() + + conn, server, cleanup := testutil.NewClientServer(t) + defer cleanup() + + testEntity := hal.NewEntity("test.entity") + conn.RegisterEntities(testEntity) + + var ( + mu sync.Mutex + observed []string + ) + + conn.RegisterAutomations( + hal.NewAutomation(). + WithName("test.order"). + WithEntities(testEntity). + WithAction(func(_ context.Context, e hal.EntityInterface) { + mu.Lock() + observed = append(observed, e.GetState().State) + mu.Unlock() + }), + ) + + const n = 50 + + expected := make([]string, 0, n) + + for i := range n { + state := "on" + if i%2 == 1 { + state = "off" + } + + expected = append(expected, state) + + // Zero timestamps so the staleness guard never applies, isolating the + // ordering behaviour of the transport layer. + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: state}, + }, + }) + } + + testutil.WaitFor(t, "verify all events processed", func() bool { + mu.Lock() + defer mu.Unlock() + + return len(observed) == n + }, func() { + mu.Lock() + defer mu.Unlock() + + spew.Dump(observed) + }) + + mu.Lock() + defer mu.Unlock() + + assert.DeepEqual(t, expected, observed) + assert.Equal(t, expected[n-1], testEntity.GetState().State) +} + +// TestStaleStateIgnored verifies that a state update carrying an older +// LastUpdated than the currently held state is dropped, so an out-of-order event +// can never overwrite newer state. +func TestStaleStateIgnored(t *testing.T) { + t.Parallel() + + conn, server, cleanup := testutil.NewClientServer(t) + defer cleanup() + + var automationTriggered atomic.Int32 + + testEntity := hal.NewEntity("test.entity") + conn.RegisterEntities(testEntity) + + conn.RegisterAutomations( + hal.NewAutomation(). + WithName("test.automation"). + WithEntities(testEntity). + WithAction(func(_ context.Context, _ hal.EntityInterface) { + automationTriggered.Add(1) + }), + ) + + t1 := time.Date(2026, 7, 17, 23, 0, 15, 0, time.UTC) + t2 := time.Date(2026, 7, 17, 23, 0, 23, 0, time.UTC) + t3 := time.Date(2026, 7, 17, 23, 0, 30, 0, time.UTC) + + // Newer "off" state (the truth). + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: "off", LastUpdated: t2}, + }, + }) + + // Older "on" state arriving out of order - must be dropped. + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: "on", LastUpdated: t1}, + }, + }) + + // Newer "off" barrier: events are delivered in order, so once this fires the + // stale "on" above has definitely been processed (and dropped). + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: "off", LastUpdated: t3}, + }, + }) + + testutil.WaitFor(t, "verify barrier event processed", func() bool { + return automationTriggered.Load() == 2 + }, func() { + spew.Dump(automationTriggered.Load(), testEntity.GetState()) + }) + + // The stale "on" neither changed the state nor fired an automation. + assert.Equal(t, "off", testEntity.GetState().State) + assert.Equal(t, int32(2), automationTriggered.Load()) +} + +// TestNilStateLeavesStateUnchanged verifies that an event carrying a nil +// NewState is ignored rather than wiping the stored state. +func TestNilStateLeavesStateUnchanged(t *testing.T) { + t.Parallel() + + conn, server, cleanup := testutil.NewClientServer(t) + defer cleanup() + + var automationTriggered atomic.Int32 + + testEntity := hal.NewEntity("test.entity") + conn.RegisterEntities(testEntity) + + conn.RegisterAutomations( + hal.NewAutomation(). + WithName("test.automation"). + WithEntities(testEntity). + WithAction(func(_ context.Context, _ hal.EntityInterface) { + automationTriggered.Add(1) + }), + ) + + t1 := time.Date(2026, 7, 17, 23, 0, 15, 0, time.UTC) + t2 := time.Date(2026, 7, 17, 23, 0, 30, 0, time.UTC) + + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: "on", LastUpdated: t1}, + }, + }) + + // nil NewState must be ignored, not overwrite the stored state. + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: nil, + }, + }) + + // Newer barrier so we can deterministically wait for the nil event to have + // been processed. + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: "on", LastUpdated: t2}, + }, + }) + + testutil.WaitFor(t, "verify barrier event processed", func() bool { + return automationTriggered.Load() == 2 + }, func() { + spew.Dump(automationTriggered.Load(), testEntity.GetState()) + }) + + // State survived the nil event, and the nil event fired no automation. + assert.Equal(t, "on", testEntity.GetState().State) + assert.Equal(t, int32(2), automationTriggered.Load()) +} diff --git a/hassws/client.go b/hassws/client.go index 96e3a9f..2d5619e 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -15,6 +15,16 @@ import ( const readTimeoutSeconds = 3 +// eventChannelBufferSize is the buffer on each per-listener response channel. The +// read loop is the single sender, so the channel preserves message order (FIFO), +// and the large buffer absorbs event bursts so the reader keeps making progress +// even while a handler is busy (e.g. blocked on a synchronous CallService whose +// reply is delivered by this same read loop). If the buffer ever fills the reader +// backpressures, which could time out an in-flight service call, but this is +// self-healing (the call times out, the handler unblocks, the backlog drains) and +// needs a burst larger than this buffer within the call timeout to occur at all. +const eventChannelBufferSize = 8192 + const ( // defaultPingInterval is how often a heartbeat ping is sent to Home // Assistant to keep the connection active and prove it is still alive. @@ -346,7 +356,12 @@ func (c *Client) listen(conn *websocket.Conn, done chan struct{}) { continue } - go func(responseListenerCh chan []byte) { + // Deliver sequentially from this single read loop so messages reach the + // listener in the order Home Assistant sent them. The channel is heavily + // buffered (eventChannelBufferSize) so this send effectively never blocks; + // the recover() tolerates a send to a channel closed during + // shutdown/reconnect. + func(responseListenerCh chan []byte) { defer func() { if r := recover(); r != nil { c.removeMessageResponseListener(msg.ID) @@ -408,7 +423,7 @@ func (c *Client) addMessageResponseListener(msgID int) (ch chan []byte) { c.mutex.Lock() defer c.mutex.Unlock() - ch = make(chan []byte) + ch = make(chan []byte, eventChannelBufferSize) c.responses[msgID] = ch return ch @@ -504,7 +519,8 @@ func (c *Client) SubscribeEvents(eventType string, handler func(EventMessage)) e return fmt.Errorf("%w: %s", ErrUnexpectedResponse, resBytes) } - // Create Goroutine to event messages and dispatch to handler + // Consume events in order. A single consumer over a FIFO channel preserves + // the order Home Assistant sent them. go func(ch chan []byte) { for b := range ch { var msg EventMessage @@ -563,6 +579,7 @@ func (c *Client) SubscribeEventsRaw(eventType string, handler func([]byte)) erro return fmt.Errorf("%w: %s", ErrUnexpectedResponse, resBytes) } + // Consume events in order (single consumer over a FIFO channel). go func(ch chan []byte) { for b := range ch { handler(b)