From aabb58856b1d7774e78d94f3fce9d959d57a6360 Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Sat, 18 Jul 2026 10:50:17 +0200 Subject: [PATCH 1/3] Fix entity state stuck out-of-sync via out-of-order event delivery Local entity state could get stuck disagreeing with Home Assistant (e.g. HAL reports a light "on" while it is really "off"). The state dump showed a mix of two events (state/last_changed from an "on" event, last_updated from a later "off" event), i.e. a newer state overwritten by an older one. Root cause was three things together: - hassws/client.go dispatched every incoming frame in a new goroutine onto a shared unbuffered channel. Go gives no ordering among goroutines racing on a channel send, so rapid on/off events (plus the Hue strip's post-on attribute-settle events) could reach the handler reordered. - Entity.SetState is an unconditional overwrite and there was no staleness check anywhere, so processing was strictly last-writer-wins. - The state only self-healed on the next reconnect via syncStates. Fix (two layers): - Deliver events in order: buffer the per-listener response channel (eventChannelBufferSize) and send sequentially from the single read loop instead of spawning a goroutine per message. The buffer keeps the read loop from blocking so synchronous CallService responses still flow. - Add a staleness guard in StateChangeEvent: drop an update whose LastUpdated is older than the currently held state (equal still applies). Also stop a nil NewState from overwriting the stored/persisted state with nil. Adds tests for in-order delivery, stale-update rejection, and nil-state handling. Co-Authored-By: Claude Opus 4.8 (1M context) --- connection.go | 23 ++++- connection_test.go | 203 +++++++++++++++++++++++++++++++++++++++++++++ hassws/client.go | 16 +++- 3 files changed, 236 insertions(+), 6 deletions(-) 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..e553e71 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -15,6 +15,14 @@ import ( const readTimeoutSeconds = 3 +// eventChannelBufferSize bounds the per-listener response channel. The read loop +// is the single sender, so a buffered channel preserves message order (FIFO) +// while absorbing bursts that arrive while a handler is busy (e.g. blocked on a +// synchronous CallService round-trip). It must be large enough that the read +// loop rarely backpressures, which would otherwise stall CallService responses +// that flow through the same read loop. +const eventChannelBufferSize = 256 + 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 +354,11 @@ func (c *Client) listen(conn *websocket.Conn, done chan struct{}) { continue } - go func(responseListenerCh chan []byte) { + // Deliver sequentially from this single read loop so events reach the + // handler in the order Home Assistant sent them. The channel is buffered + // (eventChannelBufferSize) so this rarely 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 +420,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 From 5b3bb28a9d74eef948055c5c57cbfaf115d7baf5 Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Sat, 18 Jul 2026 15:19:18 +0200 Subject: [PATCH 2/3] Decouple event dispatch from the read loop with an ordered drainer Addresses PR review (Codex): buffering the per-listener channel and sending sequentially preserved order but did not fully decouple the read loop. A subscription handler that makes a synchronous service call (e.g. Light.TurnOn -> CallService) waits for a result delivered by the same read loop; if more than eventChannelBufferSize frames were queued ahead of that result, the read loop would block on the full event channel and the call would time out. Introduce dispatchInOrder: a drainer pulls frames off the read-loop channel immediately into an unbounded, order-preserving queue, and a single feeder invokes the handler in receipt order. The reader can never be stalled by a slow or reentrant handler, while per-subscription order is preserved. Both SubscribeEvents and SubscribeEventsRaw now use it; the channel buffer is now just a small handoff/startup smoother. Adds TestReaderNotBlockedByBlockedHandler, which blocks a handler while a backlog larger than the buffer arrives and asserts a concurrent service call still completes (fails against the previous buffer-only approach). Co-Authored-By: Claude Opus 4.8 (1M context) --- connection_test.go | 70 ++++++++++++++++++++++++++++++ hassws/client.go | 105 ++++++++++++++++++++++++++++++++++----------- 2 files changed, 149 insertions(+), 26 deletions(-) diff --git a/connection_test.go b/connection_test.go index 2d66277..88a19fc 100644 --- a/connection_test.go +++ b/connection_test.go @@ -159,6 +159,76 @@ func TestEventsProcessedInOrder(t *testing.T) { assert.Equal(t, expected[n-1], testEntity.GetState().State) } +// TestReaderNotBlockedByBlockedHandler verifies that a slow/blocked subscription +// handler cannot stall the websocket read loop. A service call's result is +// delivered by the same read loop, so if the reader could block handing a backlog +// of events to a stuck handler, the call could never receive its result. Here one +// event blocks the handler while a backlog larger than the channel buffer piles +// up behind it; a concurrent service call must still complete. +func TestReaderNotBlockedByBlockedHandler(t *testing.T) { + t.Parallel() + + conn, server, cleanup := testutil.NewClientServer(t) + defer cleanup() + + testEntity := hal.NewEntity("test.entity") + light := hal.NewLight("light.test") + conn.RegisterEntities(testEntity, light) + + gate := make(chan struct{}) + + var blocked atomic.Bool + + // Released before cleanup (defers run LIFO) so the handler goroutine unblocks. + defer close(gate) + + conn.RegisterAutomations( + hal.NewAutomation(). + WithName("test.block"). + WithEntities(testEntity). + WithAction(func(_ context.Context, _ hal.EntityInterface) { + // Block the handler on the very first event; later events must + // still be read off the socket while it waits here. + if blocked.CompareAndSwap(false, true) { + <-gate + } + }), + ) + + // Far more events than eventChannelBufferSize (256), so the backlog would + // overflow a buffered channel and stall the reader without a drainer. + for range 300 { + server.SendEvent(homeassistant.Event{ + EventType: "state_changed", + EventData: homeassistant.EventData{ + EntityID: "test.entity", + NewState: &homeassistant.State{State: "on"}, + }, + }) + } + + testutil.WaitFor(t, "verify handler is blocked", func() bool { + return blocked.Load() + }, func() {}) + + // Give the reader time to take the whole backlog off the socket. + time.Sleep(200 * time.Millisecond) + + // The service call's result flows back through the same read loop, which must + // not be stuck behind the backlog feeding the blocked handler. + done := make(chan error, 1) + go func() { + done <- light.TurnOn() + }() + + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("service call blocked behind event backlog: read loop stalled") + } +} + // 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. diff --git a/hassws/client.go b/hassws/client.go index e553e71..f7a2cd6 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -15,12 +15,12 @@ import ( const readTimeoutSeconds = 3 -// eventChannelBufferSize bounds the per-listener response channel. The read loop -// is the single sender, so a buffered channel preserves message order (FIFO) -// while absorbing bursts that arrive while a handler is busy (e.g. blocked on a -// synchronous CallService round-trip). It must be large enough that the read -// loop rarely backpressures, which would otherwise stall CallService responses -// that flow through the same read loop. +// eventChannelBufferSize is a small smoothing buffer on the per-listener +// response channel. The read loop is the single sender, so the channel preserves +// message order (FIFO). Decoupling the read loop from slow handlers is handled by +// dispatchInOrder (which drains this channel eagerly into an unbounded queue), so +// this buffer only needs to cover brief handoff/startup windows, not a full +// backlog. const eventChannelBufferSize = 256 const ( @@ -354,10 +354,11 @@ func (c *Client) listen(conn *websocket.Conn, done chan struct{}) { continue } - // Deliver sequentially from this single read loop so events reach the - // handler in the order Home Assistant sent them. The channel is buffered - // (eventChannelBufferSize) so this rarely blocks; the recover() tolerates - // a send to a channel closed during shutdown/reconnect. + // Deliver sequentially from this single read loop so messages reach the + // listener in the order Home Assistant sent them. Subscription listeners + // drain this channel eagerly via dispatchInOrder, so a slow/reentrant + // handler cannot stall the reader here; the recover() tolerates a send to + // a channel closed during shutdown/reconnect. func(responseListenerCh chan []byte) { defer func() { if r := recover(); r != nil { @@ -516,19 +517,17 @@ 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 - go func(ch chan []byte) { - for b := range ch { - var msg EventMessage - if err := json.Unmarshal(b, &msg); err != nil { - logger.Error("Error unmarshalling event message", "", "error", err) + // Dispatch events to the handler in order, decoupled from the read loop. + go c.dispatchInOrder(responseChan, func(b []byte) { + var msg EventMessage + if err := json.Unmarshal(b, &msg); err != nil { + logger.Error("Error unmarshalling event message", "", "error", err) - continue - } - - handler(msg) + return } - }(responseChan) + + handler(msg) + }) logger.Info("Listening for state changes", "") @@ -575,15 +574,69 @@ func (c *Client) SubscribeEventsRaw(eventType string, handler func([]byte)) erro return fmt.Errorf("%w: %s", ErrUnexpectedResponse, resBytes) } - go func(ch chan []byte) { - for b := range ch { - handler(b) - } - }(responseChan) + go c.dispatchInOrder(responseChan, handler) return nil } +// dispatchInOrder decouples the shared websocket read loop from a subscription +// handler. A drainer moves frames off ch immediately (never running handler work +// inline) into an unbounded, order-preserving queue, and a feeder invokes deliver +// in receipt order. +// +// This matters because a handler may call a synchronous service method (e.g. +// Light.TurnOn -> CallService) whose result is delivered by this same read loop. +// If the reader blocked while handing off a backlog of subscription frames, that +// result could never arrive and the call would time out. Draining eagerly into an +// unbounded queue keeps the reader free no matter how slow or reentrant a handler +// is, while a single feeder preserves per-subscription order. +func (c *Client) dispatchInOrder(ch chan []byte, deliver func([]byte)) { + var ( + mu sync.Mutex + cond = sync.NewCond(&mu) + queue [][]byte + closed bool + ) + + // Feeder: invoke the handler for each frame, in receipt order. + go func() { + for { + mu.Lock() + for len(queue) == 0 && !closed { + cond.Wait() + } + + if len(queue) == 0 && closed { + mu.Unlock() + + return + } + + frame := queue[0] + queue[0] = nil // release the reference so the frame can be GC'd + queue = queue[1:] + mu.Unlock() + + deliver(frame) + } + }() + + // Drainer: move frames off the read loop's channel without blocking on + // handler work, preserving order. Exits when the channel is closed on + // disconnect/shutdown, then wakes the feeder to finish and return. + for b := range ch { + mu.Lock() + queue = append(queue, b) + cond.Signal() + mu.Unlock() + } + + mu.Lock() + closed = true + cond.Signal() + mu.Unlock() +} + func (c *Client) CallService(msg CallServiceRequest) (CallServiceResponse, error) { if c.getState() != stateConnected { return CallServiceResponse{}, ErrNotConnected From 63605fe7e87f7d63b9180238ed9e55b071ed6550 Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Sat, 18 Jul 2026 15:48:32 +0200 Subject: [PATCH 3/3] Simplify event dispatch: large bounded channel instead of a drainer The dispatchInOrder drainer (a goroutine feeding an unbounded mutex/cond queue) was more machinery than the failure it guarded against warrants. If the event channel fills while a handler is blocked on a synchronous CallService, the reader backpressures and the call can time out - but that is rare (needs a burst larger than the buffer within the 3s call timeout), and it self-heals: the call times out, the handler unblocks, the backlog drains. No crash, no lost events, no state corruption. Drop the drainer and instead size the per-listener channel buffer generously (256 -> 8192; a chan []byte only preallocates slice headers, so this is cheap). SubscribeEvents/SubscribeEventsRaw consume the channel directly in a single goroutine again, which still preserves order (single consumer, FIFO channel). The sequential send in the read loop and the staleness guard in StateChangeEvent are unchanged, so the original out-of-order-state fix stands. Remove TestReaderNotBlockedByBlockedHandler, which asserted a "reader never stalls" guarantee we are intentionally no longer providing. Co-Authored-By: Claude Opus 4.8 (1M context) --- connection_test.go | 70 ----------------------------- hassws/client.go | 110 +++++++++++++-------------------------------- 2 files changed, 31 insertions(+), 149 deletions(-) diff --git a/connection_test.go b/connection_test.go index 88a19fc..2d66277 100644 --- a/connection_test.go +++ b/connection_test.go @@ -159,76 +159,6 @@ func TestEventsProcessedInOrder(t *testing.T) { assert.Equal(t, expected[n-1], testEntity.GetState().State) } -// TestReaderNotBlockedByBlockedHandler verifies that a slow/blocked subscription -// handler cannot stall the websocket read loop. A service call's result is -// delivered by the same read loop, so if the reader could block handing a backlog -// of events to a stuck handler, the call could never receive its result. Here one -// event blocks the handler while a backlog larger than the channel buffer piles -// up behind it; a concurrent service call must still complete. -func TestReaderNotBlockedByBlockedHandler(t *testing.T) { - t.Parallel() - - conn, server, cleanup := testutil.NewClientServer(t) - defer cleanup() - - testEntity := hal.NewEntity("test.entity") - light := hal.NewLight("light.test") - conn.RegisterEntities(testEntity, light) - - gate := make(chan struct{}) - - var blocked atomic.Bool - - // Released before cleanup (defers run LIFO) so the handler goroutine unblocks. - defer close(gate) - - conn.RegisterAutomations( - hal.NewAutomation(). - WithName("test.block"). - WithEntities(testEntity). - WithAction(func(_ context.Context, _ hal.EntityInterface) { - // Block the handler on the very first event; later events must - // still be read off the socket while it waits here. - if blocked.CompareAndSwap(false, true) { - <-gate - } - }), - ) - - // Far more events than eventChannelBufferSize (256), so the backlog would - // overflow a buffered channel and stall the reader without a drainer. - for range 300 { - server.SendEvent(homeassistant.Event{ - EventType: "state_changed", - EventData: homeassistant.EventData{ - EntityID: "test.entity", - NewState: &homeassistant.State{State: "on"}, - }, - }) - } - - testutil.WaitFor(t, "verify handler is blocked", func() bool { - return blocked.Load() - }, func() {}) - - // Give the reader time to take the whole backlog off the socket. - time.Sleep(200 * time.Millisecond) - - // The service call's result flows back through the same read loop, which must - // not be stuck behind the backlog feeding the blocked handler. - done := make(chan error, 1) - go func() { - done <- light.TurnOn() - }() - - select { - case err := <-done: - assert.NilError(t, err) - case <-time.After(2 * time.Second): - t.Fatal("service call blocked behind event backlog: read loop stalled") - } -} - // 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. diff --git a/hassws/client.go b/hassws/client.go index f7a2cd6..2d5619e 100644 --- a/hassws/client.go +++ b/hassws/client.go @@ -15,13 +15,15 @@ import ( const readTimeoutSeconds = 3 -// eventChannelBufferSize is a small smoothing buffer on the per-listener -// response channel. The read loop is the single sender, so the channel preserves -// message order (FIFO). Decoupling the read loop from slow handlers is handled by -// dispatchInOrder (which drains this channel eagerly into an unbounded queue), so -// this buffer only needs to cover brief handoff/startup windows, not a full -// backlog. -const eventChannelBufferSize = 256 +// 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 @@ -355,10 +357,10 @@ func (c *Client) listen(conn *websocket.Conn, done chan struct{}) { } // Deliver sequentially from this single read loop so messages reach the - // listener in the order Home Assistant sent them. Subscription listeners - // drain this channel eagerly via dispatchInOrder, so a slow/reentrant - // handler cannot stall the reader here; the recover() tolerates a send to - // a channel closed during shutdown/reconnect. + // 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 { @@ -517,17 +519,20 @@ func (c *Client) SubscribeEvents(eventType string, handler func(EventMessage)) e return fmt.Errorf("%w: %s", ErrUnexpectedResponse, resBytes) } - // Dispatch events to the handler in order, decoupled from the read loop. - go c.dispatchInOrder(responseChan, func(b []byte) { - var msg EventMessage - if err := json.Unmarshal(b, &msg); err != nil { - logger.Error("Error unmarshalling event message", "", "error", err) + // 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 + if err := json.Unmarshal(b, &msg); err != nil { + logger.Error("Error unmarshalling event message", "", "error", err) - return - } + continue + } - handler(msg) - }) + handler(msg) + } + }(responseChan) logger.Info("Listening for state changes", "") @@ -574,67 +579,14 @@ func (c *Client) SubscribeEventsRaw(eventType string, handler func([]byte)) erro return fmt.Errorf("%w: %s", ErrUnexpectedResponse, resBytes) } - go c.dispatchInOrder(responseChan, handler) - - return nil -} - -// dispatchInOrder decouples the shared websocket read loop from a subscription -// handler. A drainer moves frames off ch immediately (never running handler work -// inline) into an unbounded, order-preserving queue, and a feeder invokes deliver -// in receipt order. -// -// This matters because a handler may call a synchronous service method (e.g. -// Light.TurnOn -> CallService) whose result is delivered by this same read loop. -// If the reader blocked while handing off a backlog of subscription frames, that -// result could never arrive and the call would time out. Draining eagerly into an -// unbounded queue keeps the reader free no matter how slow or reentrant a handler -// is, while a single feeder preserves per-subscription order. -func (c *Client) dispatchInOrder(ch chan []byte, deliver func([]byte)) { - var ( - mu sync.Mutex - cond = sync.NewCond(&mu) - queue [][]byte - closed bool - ) - - // Feeder: invoke the handler for each frame, in receipt order. - go func() { - for { - mu.Lock() - for len(queue) == 0 && !closed { - cond.Wait() - } - - if len(queue) == 0 && closed { - mu.Unlock() - - return - } - - frame := queue[0] - queue[0] = nil // release the reference so the frame can be GC'd - queue = queue[1:] - mu.Unlock() - - deliver(frame) + // Consume events in order (single consumer over a FIFO channel). + go func(ch chan []byte) { + for b := range ch { + handler(b) } - }() - - // Drainer: move frames off the read loop's channel without blocking on - // handler work, preserving order. Exits when the channel is closed on - // disconnect/shutdown, then wakes the feeder to finish and return. - for b := range ch { - mu.Lock() - queue = append(queue, b) - cond.Signal() - mu.Unlock() - } + }(responseChan) - mu.Lock() - closed = true - cond.Signal() - mu.Unlock() + return nil } func (c *Client) CallService(msg CallServiceRequest) (CallServiceResponse, error) {