diff --git a/Hemera/HomeAssistant/Connection/HAConnectionManager.swift b/Hemera/HomeAssistant/Connection/HAConnectionManager.swift index 7c18773..f514d0d 100644 --- a/Hemera/HomeAssistant/Connection/HAConnectionManager.swift +++ b/Hemera/HomeAssistant/Connection/HAConnectionManager.swift @@ -61,29 +61,41 @@ final class HAConnectionManager { extension HAConnectionManager: HAConnectionDelegate { nonisolated func connection(_ connection: HAConnection, didTransitionTo state: HAConnectionState) { - Task { @MainActor in - let wasConnected = isConnected - switch state { - case .ready: - isConnected = true - lastError = nil - Log.info("Connected to Home Assistant") - - if !wasConnected && hasConnectedBefore { - Log.info("Reconnected to Home Assistant — triggering resync") - onReconnect?() - } - hasConnectedBefore = true - case .disconnected: - isConnected = false - Log.warning("Connection state: disconnected") - case .connecting: - isConnected = false - Log.info("Connection state: connecting") - case .authenticating: - isConnected = false - Log.info("Connection state: authenticating") + /** + HAKit delivers this callback synchronously on `.main` + (`connection.callbackQueue = .main`, set in `init`), so handle each + transition in delivery order. Wrapping every transition in its own + unstructured `Task` would drop that ordering guarantee, so a + `.disconnected` then `.ready` pair could compute the reconnect check + against stale state and miss or spuriously fire `onReconnect`. + */ + MainActor.assumeIsolated { + handleTransition(state) + } + } + + func handleTransition(_ state: HAConnectionState) { + let wasConnected = isConnected + switch state { + case .ready: + isConnected = true + lastError = nil + Log.info("Connected to Home Assistant") + + if !wasConnected && hasConnectedBefore { + Log.info("Reconnected to Home Assistant — triggering resync") + onReconnect?() } + hasConnectedBefore = true + case .disconnected: + isConnected = false + Log.warning("Connection state: disconnected") + case .connecting: + isConnected = false + Log.info("Connection state: connecting") + case .authenticating: + isConnected = false + Log.info("Connection state: authenticating") } } } diff --git a/Hemera/HomeAssistant/HADataSyncService.swift b/Hemera/HomeAssistant/HADataSyncService.swift index 3a549f7..e4552aa 100644 --- a/Hemera/HomeAssistant/HADataSyncService.swift +++ b/Hemera/HomeAssistant/HADataSyncService.swift @@ -33,6 +33,34 @@ final class HADataSyncService { private var stateChangedToken: HACancellable? + /** + Guards the real-time path until the initial snapshot has been applied. + While `false`, incoming `state_changed` events are buffered instead of + applied, so a change that fires during the snapshot-fetch window is + reconciled against the snapshot rather than lost. + */ + private var isSnapshotApplied = false + + /** + Events received before the snapshot was applied, kept in arrival order. + A plain array is safe: the service is `@MainActor` and HAKit delivers + the subscription callback on `.main` (`connection.callbackQueue = .main`). + */ + private var bufferedEvents: [HAResponseEventStateChanged] = [] + + /** + Cap on `bufferedEvents` so a stalled sync (live socket, but a `getStates` + that never returns) can't grow the buffer without bound. Past the cap the + oldest events are dropped; the snapshot re-baselines every entity on + flush, so only intermediate states of a fast-changing entity during the + stall are lost. + */ + static let maxBufferedEvents = 2000 + + /// Ensures the buffer-overflow warning is logged once per sync window, not + /// once per dropped event. + private var didWarnBufferOverflow = false + deinit { stateChangedToken?.cancel() } private var conn: HAConnection { connectionManager.connection } @@ -55,17 +83,31 @@ final class HADataSyncService { func start() { Log.info("Starting initial data sync") - Task { - await syncAllData() - subscribeToStateChanges() - } + /** + Subscribe before fetching the snapshot so events firing during the + fetch window are buffered (not lost) and flushed once it lands. + */ + subscribeToStateChanges() + Task { await syncAllData() } } /// Re-fetches all data and re-subscribes to real-time changes (e.g. after reconnecting or pull-to-refresh). func resync() async { Log.info("Re-syncing all data") - await syncAllData() + /** + Re-arm buffering for the resync window, then re-subscribe before the + fetch so events during the window are reconciled against the snapshot + (closing the same gap as `start()`, e.g. on reconnect). + + Tradeoff: live updates are held — not applied — until the snapshot + lands and flushes, so a foreground / pull-to-refresh resync briefly + pauses live UI updates instead of showing them immediately. This is + bounded and self-healing: nothing is lost and arrival order is + preserved on flush. + */ + isSnapshotApplied = false subscribeToStateChanges() + await syncAllData() } // MARK: - Data Sync @@ -76,12 +118,18 @@ final class HADataSyncService { Log.info("Fetched \(payload.entities.count) entities, \(payload.areaMappings.count) areas, \(payload.floors?.count ?? 0) floors — applying to main context") await applySyncPayload(payload) + flushBufferedEvents() errorNotifier?.clearSyncFailed() Log.info("Sync complete") onSyncComplete() } catch { Log.error("Failed to sync data", cause: error) + /** + Flush even on failure so buffered events are applied and the + buffer can never grow unbounded across repeated failed syncs. + */ + flushBufferedEvents() errorNotifier?.showError(Localization.syncFailed) errorNotifier?.markSyncFailed() onSyncComplete() @@ -363,13 +411,50 @@ final class HADataSyncService { stateChangedToken = conn.subscribe(to: .stateChanged()) { [weak self] _, event in guard let self else { return } - Task { + /** + HAKit delivers this callback synchronously on `.main` + (`connection.callbackQueue = .main`), so handle it in delivery + order. A per-event `Task` would drop HAKit's serial ordering + guarantee and could apply back-to-back changes for one entity out + of order — and would also race the buffer flush. + */ + MainActor.assumeIsolated { self.handleStateChanged(event) } } } - private func handleStateChanged(_ event: HAResponseEventStateChanged) { + func handleStateChanged(_ event: HAResponseEventStateChanged) { + guard isSnapshotApplied else { + bufferedEvents.append(event) + if bufferedEvents.count > Self.maxBufferedEvents { + bufferedEvents.removeFirst() + if !didWarnBufferOverflow { + didWarnBufferOverflow = true + Log.warning("state_changed buffer exceeded \(Self.maxBufferedEvents) events during sync — dropping oldest; snapshot will re-baseline on completion") + } + } + return + } + applyStateChanged(event) + } + + /** + Applies events buffered during the snapshot-fetch window in arrival + order (preserving last-writer-wins), then switches to applying + subsequent events immediately. + */ + func flushBufferedEvents() { + isSnapshotApplied = true + didWarnBufferOverflow = false + let pending = bufferedEvents + bufferedEvents.removeAll() + for event in pending { + applyStateChanged(event) + } + } + + private func applyStateChanged(_ event: HAResponseEventStateChanged) { guard let entity = event.newState else { return } entityRegistry.upsert(from: entity, in: mainContext) do { diff --git a/HemeraTests/HomeAssistant/Connection/HAConnectionManagerTransitionTests.swift b/HemeraTests/HomeAssistant/Connection/HAConnectionManagerTransitionTests.swift new file mode 100644 index 0000000..9b6eb40 --- /dev/null +++ b/HemeraTests/HomeAssistant/Connection/HAConnectionManagerTransitionTests.swift @@ -0,0 +1,66 @@ +import Foundation +import HAKit +import Testing +@testable import Hemera + +/** + Tests that connection transitions are applied in delivery order so + `onReconnect` fires exactly once per genuine reconnect and never spuriously. + */ +@MainActor +struct HAConnectionManagerTransitionTests { + + private func makeManager() -> HAConnectionManager { + HAConnectionManager(serverURL: URL(string: "http://localhost:8123")!, tokenProvider: { "" }) + } + + @Test + func firstConnect_doesNotFireReconnect() { + let manager = makeManager() + var reconnectCount = 0 + manager.onReconnect = { reconnectCount += 1 } + + manager.handleTransition(.connecting) + manager.handleTransition(.authenticating) + manager.handleTransition(.ready(version: "2024.1")) + + #expect(reconnectCount == 0) + #expect(manager.isConnected) + } + + @Test + func reconnectAfterDrop_firesReconnectExactlyOnce() { + let manager = makeManager() + var reconnectCount = 0 + manager.onReconnect = { reconnectCount += 1 } + + // Initial connect — establishes `hasConnectedBefore`, no reconnect. + manager.handleTransition(.ready(version: "2024.1")) + #expect(reconnectCount == 0) + + // Drop, then recover — a genuine reconnect fires exactly once. + manager.handleTransition(.disconnected(reason: .disconnected)) + #expect(!manager.isConnected) + manager.handleTransition(.ready(version: "2024.1")) + + #expect(reconnectCount == 1) + #expect(manager.isConnected) + } + + @Test + func repeatedReady_doesNotFireSpuriousReconnect() { + let manager = makeManager() + var reconnectCount = 0 + manager.onReconnect = { reconnectCount += 1 } + + /** + Two `.ready` transitions with no intervening disconnect: the second + is already-connected, so the reconnect check must not fire. + */ + manager.handleTransition(.ready(version: "2024.1")) + manager.handleTransition(.ready(version: "2024.1")) + + #expect(reconnectCount == 0) + #expect(manager.isConnected) + } +} diff --git a/HemeraTests/HomeAssistant/HADataSyncBufferingTests.swift b/HemeraTests/HomeAssistant/HADataSyncBufferingTests.swift new file mode 100644 index 0000000..4293a11 --- /dev/null +++ b/HemeraTests/HomeAssistant/HADataSyncBufferingTests.swift @@ -0,0 +1,130 @@ +import Foundation +import SwiftData +import HAKit +import Testing +@testable import Hemera + +/** + Tests the real-time buffering path of `HADataSyncService`: events that arrive + before the initial snapshot is applied are buffered and flushed in arrival + order, so a change during the fetch window is reconciled instead of lost. + */ +@MainActor +struct HADataSyncBufferingTests { + + let container: ModelContainer + let context: ModelContext + let service: HADataSyncService + + init() { + let schema = Schema([LightEntity.self, AreaEntity.self, FloorEntity.self, HomeTile.self]) + let config = ModelConfiguration(isStoredInMemoryOnly: true) + container = try! ModelContainer(for: schema, configurations: config) + context = container.mainContext + + EntityRegistry.shared.register(LightEntity.self) + + let url = URL(string: "http://localhost:8123")! + let connectionManager = HAConnectionManager(serverURL: url, tokenProvider: { "" }) + let restClient = HARESTClient(urlProvider: { url }, tokenProvider: { "" }) + service = HADataSyncService( + connectionManager: connectionManager, + restClient: restClient, + mainContext: context, + entityRegistry: EntityRegistry.shared, + onSyncComplete: {} + ) + } + + // MARK: - Helpers + + private func lightEvent(entityId: String, state: String) throws -> HAResponseEventStateChanged { + let newState = try HAEntity( + entityId: entityId, + domain: "light", + state: state, + lastChanged: Date(), + lastUpdated: Date(), + attributes: ["friendly_name": "Lamp"], + context: .init(id: "", userId: nil, parentId: nil) + ) + let event = HAResponseEvent( + type: .stateChanged, + timeFired: Date(), + data: [:], + origin: .local, + context: .init(id: "", userId: nil, parentId: nil) + ) + return HAResponseEventStateChanged( + event: event, + entityId: entityId, + oldState: nil, + newState: newState + ) + } + + private func storedLight(_ entityId: String) throws -> LightEntity? { + try context.fetch(FetchDescriptor(predicate: LightEntity.entityIdPredicate(entityId))).first + } + + // MARK: - Buffering during the snapshot window + + @Test + func handleStateChanged_beforeSnapshot_buffersUntilFlush() throws { + let light = LightEntity(entityId: "light.lamp", name: "Lamp", state: .off) + context.insert(light) + + // Event arrives before the snapshot is applied — must be buffered, not applied. + service.handleStateChanged(try lightEvent(entityId: "light.lamp", state: "on")) + #expect(try storedLight("light.lamp")?.state == .off) + + // Snapshot lands → buffered event is applied. + service.flushBufferedEvents() + #expect(try storedLight("light.lamp")?.state == .on) + } + + @Test + func flushBufferedEvents_appliesInArrivalOrder_lastWriterWins() throws { + let light = LightEntity(entityId: "light.lamp", name: "Lamp", state: .off) + context.insert(light) + + // Two changes for one entity, buffered in arrival order: on then off. + service.handleStateChanged(try lightEvent(entityId: "light.lamp", state: "on")) + service.handleStateChanged(try lightEvent(entityId: "light.lamp", state: "off")) + #expect(try storedLight("light.lamp")?.state == .off) // untouched until flush + + service.flushBufferedEvents() + // Newest state wins — the older "on" must not overwrite the newer "off". + #expect(try storedLight("light.lamp")?.state == .off) + } + + @Test + func handleStateChanged_bufferExceedsCap_dropsOldest() throws { + // The oldest buffered event (for `light.overflow`) is pushed first, then + // the buffer is filled past its cap so that event is evicted. + let overflowId = "light.overflow" + service.handleStateChanged(try lightEvent(entityId: overflowId, state: "on")) + for index in 0..