Fix entity state stuck out-of-sync via out-of-order event delivery

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
 	})
 

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())
+}

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)