From a3f83874f17f832ea4861b06670272799c64cb52 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 12:10:04 +0100 Subject: [PATCH 1/4] fix(gecloud): keep live data flowing during slow settings refreshes; a stale read no longer overwrites a newer write GECloudDirect.run() awaited the whole periodic settings refresh inline. A register the inverter lists but cannot serve (AC charge slots 8/9 on a three-phase inverter) burns its full retry budget on every refresh, so one refresh could hold the component for ten minutes or more with no status or meter updates (a 656 s gap was observed in production). - The periodic refresh now runs as its own task. run() waits up to SETTINGS_REFRESH_WAIT_SECONDS (20 s) for it, then carries on with live data. A refresh that is still running is never overlapped. Startup and device-list changes still read the settings inline, because automatic config is built from them; a running background refresh is cancelled first. final() cancels the task and waits for it, and a cancelled settings read cancels and waits for its own read tasks. - A write confirmed while a read of the same register is in flight now wins over the older value read back. Pending writes are kept while a refresh is running, applied when the results are merged, and also written into the cached register by async_write_inverter_setting, so a register replaced mid-write is not left stale. A register written twice in one cycle reads back as its latest value. - A device removed from the account while its settings are read in the background is not put back into the cache, the entity map or the backoff state. - A register whose read fails keeps its last known value, now marked read_failed and published with a read_failed attribute rather than passed off as a fresh readback. - On periodic refreshes, a register that has failed 3 refreshes in a row is backed off: skipped for 1, 2, 4 and then at most 6 refreshes, then tried again. It is never hidden for good, and startup and new-device reads always read every register. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gecloud.py | 241 +++++++++++++--- apps/predbat/tests/test_ge_cloud.py | 418 +++++++++++++++++++++++++++- 2 files changed, 609 insertions(+), 50 deletions(-) diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 76455a365..f29f173c4 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -72,6 +72,15 @@ # from the account is picked up without a restart (#5294). Must be a multiple of 60, as run() is only # called once a minute; a cycle that finds a change polls and reads settings whatever else is due. DEVICE_REFRESH_SECONDS = 60 * 60 +# A periodic settings refresh reads every register one by one, and a register the inverter cannot serve +# burns its whole retry budget, so a refresh can run for many minutes. It runs as its own task and the +# cycle waits this long for it before carrying on, so live status and meter data keep being published. +SETTINGS_REFRESH_WAIT_SECONDS = 20 +# A setting whose read has failed this many periodic refreshes in a row is backed off: skipped for the +# next refresh, then 2, 4... up to SETTING_READ_BACKOFF_MAX_SKIP refreshes, and tried again after that. +# Its last known value is kept and published as a failed readback. +SETTING_READ_BACKOFF_FAILURES = 3 +SETTING_READ_BACKOFF_MAX_SKIP = 6 # An inverter under EMS control is expected to leave its own slot 1 windows spanning the whole day so # that they never override a command from the EMS (see #3781). Anything else is reported. @@ -630,6 +639,11 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e self.evc_devices_dict = {} self.evc_device_list = [] self.settings_from_cache = False + # The periodic settings refresh running in the background, and when it started + self.settings_refresh_task = None + self.settings_refresh_started = None + # serial -> setting id -> {"failures": consecutive failed reads, "skip": refreshes left to skip} + self.setting_read_backoff = {} self.default_options_stamp = None # Devices found by the periodic device check still waiting for their default options, kept across # cycles so a device found while read only is set up once read only ends @@ -1445,6 +1459,9 @@ async def publish_registers(self, device, registers, select_key=None): ha_name = regname_to_ha(reg_name) attributes = {} attributes["friendly_name"] = reg_name + if registers[key].get("read_failed", False): + # The last read of this register failed, so the state is its last known value, not a fresh readback + attributes["read_failed"] = True is_select_time = False is_select_options = False @@ -2243,7 +2260,7 @@ def apply_devices(self, devices, evc_devices): # Every store rather than just the devices polled until now, so entries that never were polled # (settings restored from the storage cache for hardware that has since gone) are dropped too - stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list) + stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list, self.setting_read_backoff) for device in set().union(*stores, self.ems_slot_warned) - set(device_list): for store in stores: store.pop(device, None) @@ -2437,24 +2454,24 @@ async def run(self, seconds, first): if first or devices_changed or (seconds % (10 * 60) == 0): # Get All registers every now and again in case user changes them - settings_updated = False - for device in self.device_list: - # Under EMS control (polling_mode off) the battery inverters are not the control device, but - # their own registers still have to be re-read periodically or a slot changed behind Predbat's - # back stays invisible until a restart (#5103). `first` rather than `seconds == 0` so a startup - # that only succeeds on a backoff retry still takes the snapshot. - if first or (device in new_devices) or (seconds % SETTINGS_SLOW_REFRESH_SECONDS == 0) or self.polling_mode or (device == self.ems_device) or (device == self.gateway_device): - if first and self.settings_from_cache and device in self.settings: - # Fresh cache loaded on startup — skip the slow poll for this device - await self.publish_registers(device, self.settings[device]) - else: - self.settings[device] = await self.async_get_inverter_settings(device, first=False, previous=self.settings.get(device, {})) - await self.publish_registers(device, self.settings[device]) - settings_updated = True - self.check_ems_inverter_slots(device, self.settings.get(device, {})) - - if settings_updated and self.storage: - await self.storage.save("gecloud", "settings", self.settings, format="json", expiry=None) + # Under EMS control (polling_mode off) the battery inverters are not the control device, but + # their own registers still have to be re-read periodically or a slot changed behind Predbat's + # back stays invisible until a restart (#5103). `first` rather than `seconds == 0` so a startup + # that only succeeds on a backoff retry still takes the snapshot. + refresh_devices = [device for device in self.device_list if first or (device in new_devices) or (seconds % SETTINGS_SLOW_REFRESH_SECONDS == 0) or self.polling_mode or (device == self.ems_device) or (device == self.gateway_device)] + if first or devices_changed: + # Automatic config below is built from these settings, so they are read before carrying on. A + # periodic refresh still running was started for the old device list, so it is stopped first. + await self.cancel_settings_refresh() + await self.refresh_settings(refresh_devices, use_cache=first, backoff=False) + elif self.settings_refresh_running(): + self.log("GECloud: Warn: Settings refresh started {:.0f} seconds ago is still running, not starting another".format((self.now_utc_exact - self.settings_refresh_started).total_seconds())) + else: + self.settings_refresh_started = self.now_utc_exact + self.settings_refresh_task = asyncio.get_running_loop().create_task(self.refresh_settings_background(refresh_devices), name="GECloudSettingsRefresh") + done, _ = await asyncio.wait({self.settings_refresh_task}, timeout=SETTINGS_REFRESH_WAIT_SECONDS) + if not done: + self.log("GECloud: Settings refresh of {} is taking longer than {} seconds, carrying on with live data while it completes".format(refresh_devices, SETTINGS_REFRESH_WAIT_SECONDS)) # Run at startup, and again whenever the periodic device check finds the hardware has changed if first or devices_changed: @@ -2495,14 +2512,77 @@ async def run(self, seconds, first): if self.force_charge_on(device): await self.ensure_ac_charge_gate(device) - # Clear pending writes - for device in self.device_list: - if device in self.pending_writes: - self.pending_writes[device] = [] + # Clear pending writes - unless a settings refresh is still reading, as it needs them to tell a write + # made while it was in flight from the older value it read back + if not self.settings_refresh_running(): + for device in self.device_list: + if device in self.pending_writes: + self.pending_writes[device] = [] self.update_success_timestamp() return True + def settings_refresh_running(self): + """Is a periodic settings refresh still running in the background""" + task = self.settings_refresh_task + return task is not None and not task.done() + + async def cancel_settings_refresh(self): + """Stop a background settings refresh and wait for it, and its reads, to finish""" + task = self.settings_refresh_task + self.settings_refresh_task = None + if task is not None and not task.done(): + task.cancel() + # asyncio.wait rather than awaiting the task, so cancelling this caller is not swallowed + await asyncio.wait({task}) + + async def final(self): + """Stop the background settings refresh when the component stops""" + await self.cancel_settings_refresh() + + async def refresh_settings(self, devices, use_cache=False, backoff=False): + """ + Read, cache and publish the settings of devices, then save them to storage + + use_cache publishes the settings restored from storage at startup in place of reading them. A device + dropped by a device list refresh while its settings were being read is not put back. + """ + settings_updated = False + for device in devices: + if device not in self.device_list: + continue + if use_cache and self.settings_from_cache and device in self.settings: + # Fresh cache loaded on startup — skip the slow poll for this device + await self.publish_registers(device, self.settings[device]) + else: + results = await self.async_get_inverter_settings(device, first=False, previous=self.settings.get(device, {}), backoff=backoff) + if device not in self.device_list: + # Gone from the account while it was being read: storing it would bring it back (#5294) + continue + self.settings[device] = results + await self.publish_registers(device, results) + settings_updated = True + failed = sorted((sid for sid, register in results.items() if register.get("read_failed", False)), key=str) + if failed: + self.log("GECloud: Warn: Device {} settings refresh could not read setting ids {}, their last known values are kept and marked read_failed".format(device, failed)) + self.check_ems_inverter_slots(device, self.settings.get(device, {})) + + if settings_updated and self.storage: + await self.storage.save("gecloud", "settings", self.settings, format="json", expiry=None) + + async def refresh_settings_background(self, devices): + """The periodic settings refresh, run as its own task so a slow one does not hold up live data""" + try: + await self.refresh_settings(devices, backoff=True) + except asyncio.CancelledError: + raise + except Exception as e: + self.log("GECloud: Error: Settings refresh failed: {}".format(e)) + return + elapsed = (self.now_utc_exact - self.settings_refresh_started).total_seconds() if self.settings_refresh_started else 0 + if elapsed > SETTINGS_REFRESH_WAIT_SECONDS: + self.log("GECloud: Settings refresh of {} completed after {:.0f} seconds".format(devices, elapsed)) + async def async_send_evc_command(self, uuid, command, params): """ Send a command to the EVC @@ -2545,7 +2625,8 @@ async def async_read_inverter_setting(self, serial, setting_id): """ if serial in self.pending_writes: - for pending in self.pending_writes[serial]: + # Newest first, so a setting written twice in a cycle reads back as its latest value + for pending in reversed(self.pending_writes[serial]): if pending["setting_id"] == setting_id: return {"value": pending["value"], "context": "predbat"} @@ -2597,15 +2678,27 @@ async def async_write_inverter_setting(self, serial, setting_id, value): await asyncio.sleep(random.random() * (3 + retry)) else: self.pending_writes[serial].append({"setting_id": setting_id, "value": data_value}) + # Update the register held in the cache now rather than leaving it to the caller: a settings + # refresh can replace the cached register while this write is in flight, and the caller + # would then update the replaced one and the cache would keep the value read before the write + setting = self.settings.get(serial, {}).get(setting_id, None) + if setting is not None: + setting["value"] = data_value + setting.pop("read_failed", None) break await asyncio.sleep(RETRY_FACTOR * (retry + 1)) if data is None: self.log("GECloud: Warn: Failed to write setting id {}, value {}".format(setting_id, value)) return data - async def async_get_inverter_settings(self, serial, first=False, previous=None): + async def async_get_inverter_settings(self, serial, first=False, previous=None, backoff=False): """ Get settings for account + + A register whose read fails keeps its previous value, marked read_failed so it is not passed off as a + fresh readback. With backoff, a register that has failed SETTING_READ_BACKOFF_FAILURES refreshes in a + row is skipped for a growing number of refreshes rather than costing its whole retry budget each time. + A setting written while the read was in flight keeps the written value, not the older one read back. """ if previous is None: previous = {} @@ -2622,6 +2715,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): pending = [] complete = [] loop = asyncio.get_running_loop() + device_backoff = self.setting_read_backoff.setdefault(serial, {}) # Create the read tasks for setting in self.register_list[serial]: @@ -2645,25 +2739,43 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): future["name"] = name future["validation_rules"] = validation_rules future["validation"] = validation - pending.append(future) + future["skipped"] = False + state = device_backoff.get(sid, None) + if backoff and not first and state and state["skip"] > 0: + state["skip"] -= 1 + future["skipped"] = True + future["data"] = None + complete.append(future) + else: + pending.append(future) # Perform all the reads in parallel - while pending or futures: - while len(futures) < MAX_THREADS and pending: - future = pending.pop(0) + try: + while pending or futures: + while len(futures) < MAX_THREADS and pending: + future = pending.pop(0) + if not first: + future["future"] = loop.create_task(self.async_read_inverter_setting(future["serial"], future["sid"]), name=f"ReadSetting-{future['sid']}") + futures.append(future) + if futures: + future = futures[0] + if first: + future["data"] = None + else: + future["data"] = await future["future"] + futures.pop(0) + future["future"] = None + complete.append(future) if not first: - future["future"] = loop.create_task(self.async_read_inverter_setting(future["serial"], future["sid"]), name=f"ReadSetting-{future['sid']}") - futures.append(future) - if futures: - future = futures.pop(0) - if first: - future["data"] = None - else: - future["data"] = await future["future"] - future["future"] = None - complete.append(future) - if not first: - await asyncio.sleep(0.2) + await asyncio.sleep(0.2) + except asyncio.CancelledError: + # Stopped part way (shutdown, or the device list changed): do not leave reads running unowned + outstanding = [future["future"] for future in futures if future.get("future", None) is not None] + for task in outstanding: + task.cancel() + if outstanding: + await asyncio.wait(outstanding) + raise # Wait for all the futures to complete and store results for future in complete: @@ -2677,9 +2789,19 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): else: value = None - if value is None and sid in results: - # Keep previous failure on read failure - pass + if not first and not future["skipped"]: + self.record_setting_read(device_backoff, serial, sid, name, value is not None) + + if value is None: + # Keep the previous value on a read failure, but say it is not a fresh readback (with first + # nothing was read at all, so there is no failure to report) + if sid in results: + failed = dict(results[sid]) + else: + failed = {"name": name, "value": None, "validation_rules": validation_rules, "validation": validation} + if not first: + failed["read_failed"] = True + results[sid] = failed else: results[sid] = { "name": name, @@ -2687,8 +2809,39 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None): "validation_rules": validation_rules, "validation": validation, } + + # A write confirmed while these reads were in flight is newer than what they read back, so it wins + # (the same rule async_read_inverter_setting applies to a read issued after the write) + for pending_write in self.pending_writes.get(serial, []): + sid = pending_write["setting_id"] + if sid in results: + written = dict(results[sid]) + written["value"] = pending_write["value"] + written.pop("read_failed", None) + results[sid] = written return results + def record_setting_read(self, device_backoff, serial, sid, name, success): + """ + Track consecutive failed reads of a setting and back it off once it keeps failing + + An inverter can list a register it cannot serve (for example AC charge slots 8 and 9), and every + read of it times out through the whole retry budget. Backed off, it is skipped for 1, 2, 4... up + to SETTING_READ_BACKOFF_MAX_SKIP refreshes and then tried again, so it is never hidden for good. + device_backoff is the device's entry in setting_read_backoff taken when the reads started, so a + device dropped by a device list refresh meanwhile is not added back. + """ + if success: + state = device_backoff.pop(sid, None) + if state and state["failures"] >= SETTING_READ_BACKOFF_FAILURES: + self.log("GECloud: Device {} setting id {} ({}) read again after {} failed refreshes".format(serial, sid, name, state["failures"])) + return + state = device_backoff.setdefault(sid, {"failures": 0, "skip": 0}) + state["failures"] += 1 + if state["failures"] >= SETTING_READ_BACKOFF_FAILURES: + state["skip"] = min(2 ** (state["failures"] - SETTING_READ_BACKOFF_FAILURES), SETTING_READ_BACKOFF_MAX_SKIP) + self.log("GECloud: Warn: Device {} setting id {} ({}) has failed {} refreshes in a row, skipping it for the next {} refreshes and keeping its last known value".format(serial, sid, name, state["failures"], state["skip"])) + async def async_get_smart_device_data(self, uuid): """ Get smart device data points diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index 9d4d9d55e..618d8ee5e 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -11,6 +11,7 @@ from gecloud import GECloudDirect, GECloudData, regname_to_ha from gecloud import GE_API_ACCOUNT, GE_API_DEVICES, GE_API_EVC_SEND_COMMAND, GE_API_INVERTER_WRITE_SETTING, GE_API_SITE +from gecloud import GE_API_INVERTER_READ_SETTING, GE_API_INVERTER_SETTINGS, SETTING_READ_BACKOFF_MAX_SKIP from gecloud import EVCCommandFailed, GECloudTerminalError, SITE_MAX_AGE_MINUTES, parse_site_export_limit from gecloud import DEVICE_REFRESH_SECONDS, SETTINGS_SLOW_REFRESH_SECONDS, find_ems_slot_overrides, normalise_register_time from utils import dp4 @@ -76,6 +77,9 @@ def __init__(self): self._discovery_report = None self._now_utc_exact = datetime.now(timezone.utc) self.settings_from_cache = False + self.settings_refresh_task = None + self.settings_refresh_started = None + self.setting_read_backoff = {} self.default_options_stamp = None self.pending_default_options = set() self.ems_disabled_ge_cloud_data = False @@ -292,6 +296,11 @@ def test_ge_cloud(my_predbat=None): ("settings_refetch_failed_register_list", _test_inverter_settings_refetch_after_failed_register_list, "Failed register list fetch is retried"), ("settings_saved_to_storage", _test_settings_saved_to_storage, "Settings saved to storage after poll"), ("settings_restored_from_cache", _test_settings_restored_from_fresh_cache, "Settings restored from fresh storage cache"), + ("settings_refresh_live_data", _test_settings_refresh_does_not_hold_up_live_data, "Slow settings refresh does not hold up live data in the serial loop"), + ("settings_refresh_cancelled_on_stop", _test_settings_refresh_cancelled_on_stop, "Stopping cancels and awaits the background settings refresh"), + ("settings_read_vs_write", _test_settings_read_does_not_overwrite_newer_write, "Settings read in flight does not overwrite a newer write"), + ("settings_read_backoff", _test_settings_read_backoff, "Persistently failing setting read is backed off and marked read_failed"), + ("settings_refresh_departed_device", _test_settings_refresh_departed_device, "Departed device does not come back from a background refresh"), ("inverter_status", _test_async_get_inverter_status, "Get inverter status"), ("inverter_meter", _test_async_get_inverter_meter, "Get inverter meter"), ("status_null_leaves", _test_inverter_status_null_leaves_retained, "Null status leaves retain previous reading"), @@ -2559,7 +2568,7 @@ async def mock_get_device_info(device, previous): async def mock_publish_info(device, info): pass - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): return expected_settings async def mock_publish_registers(device, settings): @@ -2641,7 +2650,7 @@ async def mock_get_device_info(device, previous): async def mock_publish_info(device, info): pass - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): poll_calls.append(device) return {} @@ -2749,7 +2758,7 @@ async def mock_publish_evc_data(serial, data): async def mock_publish_evc_device(serial, device): call_order.append(f"publish_evc_device:{serial}") - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): call_order.append(f"async_get_inverter_settings:{device}") return {} @@ -3022,7 +3031,7 @@ def capture_status(message, had_errors=False, **kwargs): async def benign(*args, **kwargs): return {} - async def mock_get_inverter_settings(device, first, previous): + async def mock_get_inverter_settings(device, first, previous, backoff=False): settings_reads.append(device) return dict(inverter_registers) if device == "inv001" else {} @@ -3157,7 +3166,7 @@ async def get_status(device, previous): calls["status"].append(device) return {"device": device} - async def get_settings(device, first, previous): + async def get_settings(device, first, previous, backoff=False): calls["settings"].append(device) return {} @@ -5509,7 +5518,7 @@ async def mock_publish_info(device, device_info): """No-op info publish.""" return None - async def mock_get_inverter_settings(device, first=False, previous=None): + async def mock_get_inverter_settings(device, first=False, previous=None, backoff=False): """Return this device's fixture register settings, the same shape self.settings[device] holds for real.""" return settings.get(device, {}) @@ -8510,3 +8519,400 @@ async def run_all(): return await check() + await check_storage() + await check_auth() return run_async(run_all()) + + +# ============================================================================= +# Settings refresh isolation: a slow settings read must not hold up live data, and must not overwrite a +# newer confirmed write in the cache + + +ISOLATION_REGISTERS = [ + {"id": 17, "name": "Enable Discharge", "validation_rules": ["boolean"], "validation": None}, + {"id": 120, "name": "AC Charge 8 Start Time", "validation_rules": ["date_format:H:i"], "validation": None}, +] +ISOLATION_SWITCH = "switch.predbat_gecloud_inv001_enable_discharge" + + +def _make_isolation_ge_cloud(devices=("inv001",)): + """ + A component with the real settings, read, write, refresh and publish code, over a fake GE Cloud API. + + Reads of setting id ge.hold_id (120) can be held (ge.hold_reads) and writes held (ge.hold_writes) to order a read + against a write; ge.read_values gives what each setting id reads back, None for a failed read. + """ + ge = MockGECloudDirect() + ge.automatic = False + ge._read_only = True + ge.device_list_reported = list(devices) + ge.read_values = {17: False, 120: "02:00"} + ge.read_calls = [] + ge.read_cancelled = [] + ge.status_calls = [] + ge.hold_id = 120 + ge.hold_reads = None + ge.hold_writes = None + ge.read_started = asyncio.Event() + + async def get_inverter_data(endpoint, serial="", setting_id="", post=False, datain=None, *args, **kwargs): + if endpoint == GE_API_INVERTER_SETTINGS: + return copy.deepcopy(ISOLATION_REGISTERS) + if endpoint == GE_API_INVERTER_READ_SETTING: + ge.read_calls.append((serial, setting_id)) + if setting_id == ge.hold_id and ge.hold_reads is not None: + ge.read_started.set() + try: + await ge.hold_reads.wait() + except asyncio.CancelledError: + ge.read_cancelled.append((serial, setting_id)) + raise + value = ge.read_values.get(setting_id, None) + return None if value is None else {"value": value} + if endpoint == GE_API_INVERTER_WRITE_SETTING: + if ge.hold_writes is not None: + await ge.hold_writes.wait() + return {"value": datain["value"] == "True"} + return None + + async def get_devices(): + return {"battery": list(ge.device_list_reported), "ems": None, "gateway": None, "pv": [], "battery_meters": {}} + + async def get_evc_devices(previous=None): + return [] + + async def get_inverter_status(device, previous): + ge.status_calls.append(device) + return {"time": "fresh"} + + async def get_empty(device, previous): + return {} + + async def nothing(*args, **kwargs): + return None + + ge.async_get_inverter_data = get_inverter_data + ge.async_get_devices = get_devices + ge.async_get_evc_devices = get_evc_devices + ge.async_get_inverter_status = get_inverter_status + ge.async_get_inverter_meter = get_empty + ge.async_get_device_info = get_empty + ge.publish_status = nothing + ge.publish_meter = nothing + ge.publish_info = nothing + ge.publish_site_export_limit = nothing + ge.update_account = nothing + ge.update_site = nothing + return ge + + +class _FastComponentLoop: + """ + Stands in for the asyncio module inside component_base, so ComponentBase.start() runs its real serial + loop with each 5 second tick taking a millisecond. on_tick(seconds) is awaited once per tick. + """ + + def __init__(self, component, on_tick): + self.component = component + self.on_tick = on_tick + self.seconds = 0 + self.real_sleep = asyncio.sleep + + def __getattr__(self, name): + return getattr(asyncio, name) + + async def sleep(self, delay): + if delay == 5: + self.seconds += 5 + await self.on_tick(self.seconds) + await self.real_sleep(0.001) + + +async def _start_component_loop(ge, on_tick, limit=10): + """Run the real ComponentBase.start() loop for ge until on_tick sets api_stop; False if it never finishes""" + import component_base + + ge.api_stop = False + ge.api_started = False + ge.run_timeout = 60 * 60 + ge.count_errors = 0 + ge.base.fatal_error = False + ge.non_fatal_error_occurred = lambda: None + loop = _FastComponentLoop(ge, on_tick) + with patch.object(component_base, "asyncio", loop): + try: + await asyncio.wait_for(ge.start(), timeout=limit) + except asyncio.TimeoutError: + return False + return True + + +def _test_settings_refresh_does_not_hold_up_live_data(my_predbat): + """ + Through the real serial ComponentBase.start() loop: while a periodic settings refresh is stuck on a + register read, the following cycles still fetch live status, no second refresh is started on top of it, + and a write made meanwhile is still in the cache once the held refresh completes. Before the fix the + cycle at 600 seconds awaited the refresh, so the loop never got past it. + """ + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud() + ge.hold_reads = asyncio.Event() + marks = {} + + async def on_tick(seconds): + if seconds == 600: + # The first cycle has finished; hold the next refresh on setting id 120 + ge.hold_reads.clear() + marks["status_before"] = len(ge.status_calls) + if seconds == 900: + marks["status_during"] = len(ge.status_calls) + marks["reads_during"] = len([call for call in ge.read_calls if call[1] == 120]) + # Predbat turns discharge on while the refresh is still holding its older read of the switch + await ge.switch_event(ISOLATION_SWITCH, "turn_on") + if seconds == 1500: + ge.hold_reads.set() + if seconds == 1560: + ge.api_stop = True + + ge.hold_reads.set() + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + finished = await _start_component_loop(ge, on_tick) + if not finished: + print("ERROR: The component loop stalled behind the settings refresh") + return 1 + + # Status every 120 seconds from 600 to 900 is 720 and 840, so live data kept coming + if marks.get("status_during", 0) - marks.get("status_before", 0) < 2: + print("ERROR: Live status was not fetched while the settings refresh was held: {}".format(marks)) + failed += 1 + # Two reads of id 120: startup, then the refresh at 600 - the cycle at 1200 must not start another + if marks.get("reads_during") != 2 or len([call for call in ge.read_calls if call[1] == 120]) != 2: + print("ERROR: Expected one held read of id 120 after startup, got reads {}".format(ge.read_calls)) + failed += 1 + if not any("still running, not starting another" in message for message in ge.log_messages): + print("ERROR: The skipped overlapping refresh was not logged") + failed += 1 + if ge.settings["inv001"][17]["value"] is not True: + print("ERROR: The held refresh replaced the newer write: {}".format(ge.settings["inv001"][17])) + failed += 1 + if ge.dashboard_items.get(ISOLATION_SWITCH, {}).get("state") != "on": + print("ERROR: The switch was not left published on: {}".format(ge.dashboard_items.get(ISOLATION_SWITCH))) + failed += 1 + if ge.settings_refresh_running(): + print("ERROR: Refresh still running after the component stopped") + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_refresh_cancelled_on_stop(my_predbat): + """Stopping the component cancels a held background refresh and waits for its read to finish""" + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud() + ge.hold_reads = asyncio.Event() + ge.hold_reads.set() + + async def on_tick(seconds): + if seconds == 600: + ge.hold_reads.clear() + if seconds == 720: + ge.api_stop = True + + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + finished = await _start_component_loop(ge, on_tick) + if not finished: + print("ERROR: The component loop stalled behind the settings refresh") + return 1 + if ge.settings_refresh_task is not None: + print("ERROR: final() left the refresh task in place") + failed += 1 + if ge.read_cancelled != [("inv001", 120)]: + print("ERROR: The held read was not cancelled and awaited: {}".format(ge.read_cancelled)) + failed += 1 + leftover = [task for task in asyncio.all_tasks() if task is not asyncio.current_task() and task.get_name().startswith(("ReadSetting-", "GECloudSettingsRefresh"))] + if leftover: + print("ERROR: Tasks left running after stop: {}".format(leftover)) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_read_does_not_overwrite_newer_write(my_predbat): + """ + A write confirmed while a settings read of the same register is in flight is not replaced in the cache + by the older value read back - whether the write lands before the read returns, or after the refresh + has stored its results + """ + + async def test(): + failed = 0 + + async def refresh(ge): + """Read the settings and store and publish them, as a settings refresh does""" + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings.get("inv001", {})) + await ge.publish_registers("inv001", ge.settings["inv001"]) + + # The write lands while the read of the register is held + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + await refresh(ge) + ge.hold_id = 17 + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + reading = asyncio.ensure_future(ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"])) + await asyncio.wait_for(ge.read_started.wait(), 1) + await ge.switch_event(ISOLATION_SWITCH, "turn_on") + ge.hold_reads.set() + ge.settings["inv001"] = await asyncio.wait_for(reading, 2) + if ge.settings["inv001"][17]["value"] is not True: + print("ERROR: Read in flight across a write replaced it: {}".format(ge.settings["inv001"][17])) + failed += 1 + + # The refresh stores its results while the write is still in flight + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + await refresh(ge) + ge.hold_writes = asyncio.Event() + write = asyncio.ensure_future(ge.switch_event(ISOLATION_SWITCH, "turn_on")) + await asyncio.sleep(0) + await refresh(ge) + ge.hold_writes.set() + await asyncio.wait_for(write, 2) + if ge.settings["inv001"][17]["value"] is not True: + print("ERROR: A write completing after the refresh stored its results was lost: {}".format(ge.settings["inv001"][17])) + failed += 1 + if ge.dashboard_items.get(ISOLATION_SWITCH, {}).get("state") != "on": + print("ERROR: The switch was published from the replaced register: {}".format(ge.dashboard_items.get(ISOLATION_SWITCH))) + failed += 1 + + # Written twice in one cycle, the setting reads back as the later write + ge.pending_writes["inv001"] = [{"setting_id": 17, "value": True}, {"setting_id": 17, "value": False}] + data = await ge.async_read_inverter_setting("inv001", 17) + if data != {"value": False, "context": "predbat"}: + print("ERROR: Pending write read back as the older value: {}".format(data)) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_read_backoff(my_predbat): + """ + A setting whose reads keep failing is backed off on the periodic refresh, keeps its last known value + marked read_failed (and published so), is tried again after the backoff, and recovers on a good read + """ + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + with patch("gecloud.RETRIES", 1), patch("gecloud.RETRY_FACTOR", 0): + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous={}) + ge.read_values[120] = None + read_pattern = [] + for _ in range(9): + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + read_pattern.append(any(call[1] == 120 for call in ge.read_calls[before:])) + # Fails 1, 2, 3 (back off 1), skip, fails 4 (back off 2), skip, skip, fails 5 (back off 4), skip + expected = [True, True, True, False, True, False, False, True, False] + if read_pattern != expected: + print("ERROR: Backoff read pattern {} expected {}".format(read_pattern, expected)) + failed += 1 + register = ge.settings["inv001"][120] + if register.get("value") != "02:00" or not register.get("read_failed"): + print("ERROR: Failed register should keep its last value marked read_failed: {}".format(register)) + failed += 1 + await ge.publish_registers("inv001", ge.settings["inv001"]) + published = ge.dashboard_items.get("select.predbat_gecloud_inv001_ac_charge_8_start_time", {}) + if not published.get("attributes", {}).get("read_failed"): + print("ERROR: read_failed not published: {}".format(published)) + failed += 1 + + # Without backoff (startup, a new device) every register is read + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"]) + if not any(call[1] == 120 for call in ge.read_calls[before:]): + print("ERROR: A read without backoff skipped the backed off register") + failed += 1 + + # Good again: read on the next try after the backoff, and the flag and backoff are cleared + ge.read_values[120] = "03:00" + for _ in range(SETTING_READ_BACKOFF_MAX_SKIP + 1): + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + register = ge.settings["inv001"][120] + if register.get("value") != "03:00" or register.get("read_failed") or 120 in ge.setting_read_backoff.get("inv001", {}): + print("ERROR: Register did not recover: {} backoff {}".format(register, ge.setting_read_backoff)) + failed += 1 + if ge.settings["inv001"][17].get("read_failed"): + print("ERROR: A good register was marked read_failed") + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_refresh_departed_device(my_predbat): + """ + A device removed from the account while its settings are being read in the background is not put + back in the cache or published, and a device list change stops the background refresh before reading + the settings itself + """ + + async def test(): + failed = 0 + ge = _make_isolation_ge_cloud(devices=("inv001", "inv002")) + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + await ge.run(seconds=0, first=True) + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + try: + await asyncio.wait_for(ge.run(seconds=600, first=False), 5) + except asyncio.TimeoutError: + print("ERROR: run() at 600 seconds waited for the held settings refresh") + return 1 + await asyncio.wait_for(ge.read_started.wait(), 1) + if not ge.settings_refresh_running(): + print("ERROR: Expected the refresh to be held") + return 1 + held = ge.settings_refresh_task + + # inv001 is held mid read; take it off the account and let the read finish + ge.apply_devices({"battery": ["inv002"], "ems": None, "gateway": None, "pv": [], "battery_meters": {}}, []) + for entity_id in [entity_id for entity_id in ge.dashboard_items if "inv001" in entity_id]: + del ge.dashboard_items[entity_id] + ge.hold_reads.set() + await asyncio.wait_for(held, 2) + if "inv001" in ge.settings or "inv001" in ge.setting_read_backoff or any("inv001" in entity_id for entity_id in ge.dashboard_items) or any(entry["device"] == "inv001" for entry in ge.register_entity_map.values()): + print("ERROR: Departed device came back: settings {} entities {}".format(list(ge.settings), [e for e in ge.dashboard_items if "inv001" in e])) + failed += 1 + + # A device list change while a refresh is held cancels it, then reads the settings in line + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + try: + await asyncio.wait_for(ge.run(seconds=1200, first=False), 5) + except asyncio.TimeoutError: + print("ERROR: run() at 1200 seconds waited for the held settings refresh") + return 1 + await asyncio.wait_for(ge.read_started.wait(), 1) + held = ge.settings_refresh_task + ge.hold_reads = None + ge.device_list_reported = ["inv002", "inv003"] + await asyncio.wait_for(ge.run(seconds=DEVICE_REFRESH_SECONDS, first=False), 5) + if not held.cancelled() or ge.read_cancelled[-1:] != [("inv002", 120)]: + print("ERROR: Device list change did not cancel the held refresh: {} {}".format(held, ge.read_cancelled)) + failed += 1 + if "inv003" not in ge.settings or ge.settings_refresh_running(): + print("ERROR: New device's settings were not read in line: {}".format(list(ge.settings))) + failed += 1 + return failed + + return run_async(test()) From 9df6f46493a41288ad57401cafc1aaca307dfcf0 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 12:30:29 +0100 Subject: [PATCH 2/4] fix(gecloud): address review of the settings refresh isolation - A write now wins over a settings read only when the write was confirmed after that register's read was issued, using a write sequence number. pending_writes is cleared every cycle again, as before, so Predbat's own earlier write no longer masks a change made in the GivEnergy app for the length of a slow refresh. - A read failure only counts towards backoff when the device answered other reads in the same refresh. An inverter or GE Cloud outage no longer backs every register off and blinds the refreshes after it. - Devices that were due a read while a refresh was still running (such as the hourly re-read of EMS member inverters) are read by the next refresh instead of waiting for their next slot. - The default options pass and the AC charge gate run on fresh settings. When the refresh outlasts the wait, it runs them itself on completion instead of the cycle using the old snapshot. - A failed background refresh logs a traceback and fails the next cycle, so it is counted as an error as it was when the refresh ran inline. - The register list of a device removed during a refresh is dropped. - Test: the held refresh in the serial loop test now runs to completion and must store the new value it reads. Previously it was cancelled at stop and the test could not see the merge. Added outage, owed-device, deferred-defaults, failure-reporting and external-change cases. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gecloud.py | 134 +++++++++++++++++++--------- apps/predbat/tests/test_ge_cloud.py | 116 +++++++++++++++++++++++- 2 files changed, 204 insertions(+), 46 deletions(-) diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index f29f173c4..19b119c43 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -23,6 +23,7 @@ import math import json import random +import traceback from component_base import ComponentBase from coordinator import inverter_record from mock_base import MockBase as SharedMockBase @@ -642,6 +643,16 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e # The periodic settings refresh running in the background, and when it started self.settings_refresh_task = None self.settings_refresh_started = None + # Devices due a settings read while a refresh was still running, read by the next one + self.settings_refresh_owed = set() + # Set when the cycle that started the refresh stopped waiting for it + self.settings_refresh_detached = False + # Set when a background refresh failed, so the next cycle reports the failure + self.settings_refresh_error = False + # Every confirmed write is numbered; serial -> setting id -> (number, value) of the latest one, so + # a settings read can tell whether a write landed after it was issued + self.write_seq = 0 + self.recent_writes = {} # serial -> setting id -> {"failures": consecutive failed reads, "skip": refreshes left to skip} self.setting_read_backoff = {} self.default_options_stamp = None @@ -2260,7 +2271,7 @@ def apply_devices(self, devices, evc_devices): # Every store rather than just the devices polled until now, so entries that never were polled # (settings restored from the storage cache for hardware that has since gone) are dropped too - stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list, self.setting_read_backoff) + stores = (self.status, self.meter, self.info, self.settings, self.pending_writes, self.register_list, self.setting_read_backoff, self.recent_writes) for device in set().union(*stores, self.ems_slot_warned) - set(device_list): for store in stores: store.pop(device, None) @@ -2459,18 +2470,29 @@ async def run(self, seconds, first): # back stays invisible until a restart (#5103). `first` rather than `seconds == 0` so a startup # that only succeeds on a backoff retry still takes the snapshot. refresh_devices = [device for device in self.device_list if first or (device in new_devices) or (seconds % SETTINGS_SLOW_REFRESH_SECONDS == 0) or self.polling_mode or (device == self.ems_device) or (device == self.gateway_device)] + settings_fresh = True if first or devices_changed: # Automatic config below is built from these settings, so they are read before carrying on. A # periodic refresh still running was started for the old device list, so it is stopped first. await self.cancel_settings_refresh() await self.refresh_settings(refresh_devices, use_cache=first, backoff=False) elif self.settings_refresh_running(): + # Remember what was due (the hourly re-read of EMS member inverters, say), so it is read by the + # next refresh rather than waiting for its next turn + self.settings_refresh_owed.update(refresh_devices) + settings_fresh = False self.log("GECloud: Warn: Settings refresh started {:.0f} seconds ago is still running, not starting another".format((self.now_utc_exact - self.settings_refresh_started).total_seconds())) else: + refresh_devices = [device for device in self.device_list if device in refresh_devices or device in self.settings_refresh_owed] + self.settings_refresh_owed.clear() self.settings_refresh_started = self.now_utc_exact + self.settings_refresh_detached = False self.settings_refresh_task = asyncio.get_running_loop().create_task(self.refresh_settings_background(refresh_devices), name="GECloudSettingsRefresh") done, _ = await asyncio.wait({self.settings_refresh_task}, timeout=SETTINGS_REFRESH_WAIT_SECONDS) if not done: + # The refresh applies the default options and AC charge gate itself once it has fresh settings + self.settings_refresh_detached = True + settings_fresh = False self.log("GECloud: Settings refresh of {} is taking longer than {} seconds, carrying on with live data while it completes".format(refresh_devices, SETTINGS_REFRESH_WAIT_SECONDS)) # Run at startup, and again whenever the periodic device check finds the hardware has changed @@ -2489,39 +2511,51 @@ async def run(self, seconds, first): # so a failed report is retried rather than lost - see refresh_discovery(). self.refresh_discovery() - now_utc = self.now_utc_exact - options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24) - # read_only_now() rather than the switch alone, so an active Axle event (which forces read only - # through the attribute) defers the pass instead of writing over Axle's control (GH#5269) - if not self.read_only_now(): - if options_due: - self.default_options_stamp = now_utc - option_devices = self.device_list - else: - # A device found by the periodic device check is set up now, not at the next 24 hour pass - - # or, when it was found while read only, on the first settings cycle after read only ends - option_devices = [device for device in self.device_list if device in self.pending_default_options] - self.pending_default_options.clear() - for device in option_devices: - await self.enable_default_options(device, self.settings[device]) - - # The 24 hour pass above is only a backstop for the AC charge gate: while a force charge is - # running, re-assert it on every settings refresh in case it was cleared mid-charge (GH#5269). - # Nothing is written while Predbat is not force charging, so an external controller is not fought. - for device in self.device_list: - if self.force_charge_on(device): - await self.ensure_ac_charge_gate(device) + if settings_fresh: + await self.apply_settings_defaults() - # Clear pending writes - unless a settings refresh is still reading, as it needs them to tell a write - # made while it was in flight from the older value it read back - if not self.settings_refresh_running(): - for device in self.device_list: - if device in self.pending_writes: - self.pending_writes[device] = [] + # Clear pending writes + for device in self.device_list: + if device in self.pending_writes: + self.pending_writes[device] = [] + + if self.settings_refresh_error: + # A background settings refresh failed since the last cycle: report it as this cycle failing, as it + # was when the refresh ran inline + self.settings_refresh_error = False + return False self.update_success_timestamp() return True + async def apply_settings_defaults(self): + """ + Once fresh settings are in hand: set the default options (every 24 hours, or for a new device) and + re-assert the AC charge gate while force charging + """ + now_utc = self.now_utc_exact + options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24) + # read_only_now() rather than the switch alone, so an active Axle event (which forces read only + # through the attribute) defers the pass instead of writing over Axle's control (GH#5269) + if not self.read_only_now(): + if options_due: + self.default_options_stamp = now_utc + option_devices = self.device_list + else: + # A device found by the periodic device check is set up now, not at the next 24 hour pass - + # or, when it was found while read only, on the first settings cycle after read only ends + option_devices = [device for device in self.device_list if device in self.pending_default_options] + self.pending_default_options.clear() + for device in option_devices: + await self.enable_default_options(device, self.settings[device]) + + # The 24 hour pass above is only a backstop for the AC charge gate: while a force charge is + # running, re-assert it on every settings refresh in case it was cleared mid-charge (GH#5269). + # Nothing is written while Predbat is not force charging, so an external controller is not fought. + for device in self.device_list: + if self.force_charge_on(device): + await self.ensure_ac_charge_gate(device) + def settings_refresh_running(self): """Is a periodic settings refresh still running in the background""" task = self.settings_refresh_task @@ -2558,6 +2592,7 @@ async def refresh_settings(self, devices, use_cache=False, backoff=False): results = await self.async_get_inverter_settings(device, first=False, previous=self.settings.get(device, {}), backoff=backoff) if device not in self.device_list: # Gone from the account while it was being read: storing it would bring it back (#5294) + self.register_list.pop(device, None) continue self.settings[device] = results await self.publish_registers(device, results) @@ -2574,14 +2609,17 @@ async def refresh_settings_background(self, devices): """The periodic settings refresh, run as its own task so a slow one does not hold up live data""" try: await self.refresh_settings(devices, backoff=True) + elapsed = (self.now_utc_exact - self.settings_refresh_started).total_seconds() if self.settings_refresh_started else 0 + if self.settings_refresh_detached: + self.log("GECloud: Settings refresh of {} completed after {:.0f} seconds".format(devices, elapsed)) + # The cycle that started this refresh moved on without it, so apply the defaults here + await self.apply_settings_defaults() except asyncio.CancelledError: raise except Exception as e: self.log("GECloud: Error: Settings refresh failed: {}".format(e)) - return - elapsed = (self.now_utc_exact - self.settings_refresh_started).total_seconds() if self.settings_refresh_started else 0 - if elapsed > SETTINGS_REFRESH_WAIT_SECONDS: - self.log("GECloud: Settings refresh of {} completed after {:.0f} seconds".format(devices, elapsed)) + self.log("Error: " + traceback.format_exc()) + self.settings_refresh_error = True async def async_send_evc_command(self, uuid, command, params): """ @@ -2678,6 +2716,8 @@ async def async_write_inverter_setting(self, serial, setting_id, value): await asyncio.sleep(random.random() * (3 + retry)) else: self.pending_writes[serial].append({"setting_id": setting_id, "value": data_value}) + self.write_seq += 1 + self.recent_writes.setdefault(serial, {})[setting_id] = (self.write_seq, data_value) # Update the register held in the cache now rather than leaving it to the caller: a settings # refresh can replace the cached register while this write is in flight, and the caller # would then update the replaced one and the cache would keep the value read before the write @@ -2716,6 +2756,8 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, complete = [] loop = asyncio.get_running_loop() device_backoff = self.setting_read_backoff.setdefault(serial, {}) + # Writes numbered after this were confirmed after the reads below began + refresh_seq = self.write_seq # Create the read tasks for setting in self.register_list[serial]: @@ -2740,6 +2782,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, future["validation_rules"] = validation_rules future["validation"] = validation future["skipped"] = False + future["issue_seq"] = refresh_seq state = device_backoff.get(sid, None) if backoff and not first and state and state["skip"] > 0: state["skip"] -= 1 @@ -2754,6 +2797,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, while pending or futures: while len(futures) < MAX_THREADS and pending: future = pending.pop(0) + future["issue_seq"] = self.write_seq if not first: future["future"] = loop.create_task(self.async_read_inverter_setting(future["serial"], future["sid"]), name=f"ReadSetting-{future['sid']}") futures.append(future) @@ -2777,6 +2821,10 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, await asyncio.wait(outstanding) raise + # A failure only counts towards backing a register off when the device answered other reads in the + # same refresh: when nothing reads at all the device or GE Cloud is down, not that one register + device_answered = any(future["data"] and ("value" in future["data"]) for future in complete) + # Wait for all the futures to complete and store results for future in complete: sid = future["sid"] @@ -2789,7 +2837,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, else: value = None - if not first and not future["skipped"]: + if not first and not future["skipped"] and (value is not None or device_answered): self.record_setting_read(device_backoff, serial, sid, name, value is not None) if value is None: @@ -2810,15 +2858,15 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, "validation": validation, } - # A write confirmed while these reads were in flight is newer than what they read back, so it wins - # (the same rule async_read_inverter_setting applies to a read issued after the write) - for pending_write in self.pending_writes.get(serial, []): - sid = pending_write["setting_id"] - if sid in results: - written = dict(results[sid]) - written["value"] = pending_write["value"] - written.pop("read_failed", None) - results[sid] = written + # A write confirmed after a register's read was issued is newer than what that read returns, so it wins + # (a read issued after the write already reads back the pending write, see async_read_inverter_setting) + issued = {future["sid"]: future["issue_seq"] for future in complete} + for sid, (seq, value) in self.recent_writes.get(serial, {}).items(): + if sid in results and seq > issued.get(sid, refresh_seq): + written = dict(results[sid]) + written["value"] = value + written.pop("read_failed", None) + results[sid] = written return results def record_setting_read(self, device_backoff, serial, sid, name, success): diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index 618d8ee5e..bf0272bba 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -80,6 +80,11 @@ def __init__(self): self.settings_refresh_task = None self.settings_refresh_started = None self.setting_read_backoff = {} + self.settings_refresh_owed = set() + self.settings_refresh_detached = False + self.settings_refresh_error = False + self.write_seq = 0 + self.recent_writes = {} self.default_options_stamp = None self.pending_default_options = set() self.ems_disabled_ge_cloud_data = False @@ -301,6 +306,7 @@ def test_ge_cloud(my_predbat=None): ("settings_read_vs_write", _test_settings_read_does_not_overwrite_newer_write, "Settings read in flight does not overwrite a newer write"), ("settings_read_backoff", _test_settings_read_backoff, "Persistently failing setting read is backed off and marked read_failed"), ("settings_refresh_departed_device", _test_settings_refresh_departed_device, "Departed device does not come back from a background refresh"), + ("settings_refresh_follow_ups", _test_settings_refresh_follow_ups, "Owed devices, deferred default options and failure reporting for the background refresh"), ("inverter_status", _test_async_get_inverter_status, "Get inverter status"), ("inverter_meter", _test_async_get_inverter_meter, "Get inverter meter"), ("status_null_leaves", _test_inverter_status_null_leaves_retained, "Null status leaves retain previous reading"), @@ -8670,8 +8676,11 @@ async def on_tick(seconds): # Predbat turns discharge on while the refresh is still holding its older read of the switch await ge.switch_event(ISOLATION_SWITCH, "turn_on") if seconds == 1500: + marks["held_task"] = ge.settings_refresh_task + # The slot changed in the meantime, so the completed refresh must bring the new value in + ge.read_values[120] = "05:30" ge.hold_reads.set() - if seconds == 1560: + if seconds > 1500 and not ge.settings_refresh_running(): ge.api_stop = True ge.hold_reads.set() @@ -8698,8 +8707,12 @@ async def on_tick(seconds): if ge.dashboard_items.get(ISOLATION_SWITCH, {}).get("state") != "on": print("ERROR: The switch was not left published on: {}".format(ge.dashboard_items.get(ISOLATION_SWITCH))) failed += 1 - if ge.settings_refresh_running(): - print("ERROR: Refresh still running after the component stopped") + held = marks.get("held_task") + if held is None or held.cancelled() or not held.done(): + print("ERROR: The held refresh did not run to completion: {}".format(held)) + failed += 1 + if ge.settings["inv001"][120]["value"] != "05:30" or ge.settings["inv001"][120].get("read_failed"): + print("ERROR: The completed refresh did not store the value it read: {}".format(ge.settings["inv001"][120])) failed += 1 return failed @@ -8797,6 +8810,15 @@ async def refresh(ge): if data != {"value": False, "context": "predbat"}: print("ERROR: Pending write read back as the older value: {}".format(data)) failed += 1 + + # Once the write is no longer pending, a read issued after it is the inverter's own state: a change + # made outside Predbat since (in the GivEnergy app, say) is not masked by Predbat's earlier write + ge.pending_writes["inv001"] = [] + ge.read_values[17] = False + await refresh(ge) + if ge.settings["inv001"][17]["value"] is not False: + print("ERROR: An earlier write masked a later change read from the inverter: {}".format(ge.settings["inv001"][17])) + failed += 1 return failed return run_async(test()) @@ -8854,6 +8876,94 @@ async def test(): if ge.settings["inv001"][17].get("read_failed"): print("ERROR: A good register was marked read_failed") failed += 1 + + # The device or GE Cloud being down fails every read: that is not a register it cannot serve, so + # nothing is backed off and every register is read again as soon as it is back + ge.read_values = {17: None, 120: None} + for _ in range(SETTING_READ_BACKOFF_MAX_SKIP + 2): + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if ge.setting_read_backoff.get("inv001"): + print("ERROR: An outage backed registers off: {}".format(ge.setting_read_backoff)) + failed += 1 + ge.read_values = {17: True, 120: "04:00"} + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if sorted(call[1] for call in ge.read_calls[before:]) != [17, 120] or ge.settings["inv001"][120]["value"] != "04:00": + print("ERROR: Registers not read on the first refresh after an outage: {}".format(ge.read_calls[before:])) + failed += 1 + return failed + + return run_async(test()) + + +def _test_settings_refresh_follow_ups(my_predbat): + """ + Devices due while a refresh is still running are read by the next one; the default options pass waits + for fresh settings rather than running on the old ones; a failed background refresh fails the cycle + """ + + async def test(): + failed = 0 + with patch("gecloud.SETTINGS_REFRESH_WAIT_SECONDS", 0.05): + # Under an EMS the member inverter inv001 is only re-read hourly; that slot falls while a refresh + # of the EMS is held, so the next refresh must read it + ge = _make_isolation_ge_cloud(devices=("inv001", "inv002")) + await ge.run(seconds=0, first=True) + ge.polling_mode = False + ge.ems_device = "inv002" + ge.hold_reads = asyncio.Event() + ge.read_started.clear() + await asyncio.wait_for(ge.run(seconds=SETTINGS_SLOW_REFRESH_SECONDS - 600, first=False), 5) + await asyncio.wait_for(ge.read_started.wait(), 1) + await asyncio.wait_for(ge.run(seconds=SETTINGS_SLOW_REFRESH_SECONDS, first=False), 5) + ge.hold_reads.set() + await asyncio.wait_for(ge.settings_refresh_task, 5) + ge.hold_reads = None + before = len(ge.read_calls) + await asyncio.wait_for(ge.run(seconds=SETTINGS_SLOW_REFRESH_SECONDS + 600, first=False), 5) + await asyncio.wait_for(ge.settings_refresh_task, 5) + if not any(call[0] == "inv001" for call in ge.read_calls[before:]): + print("ERROR: The hourly re-read skipped behind a running refresh was lost: {}".format(ge.read_calls[before:])) + failed += 1 + + # The default options pass is due; the refresh is slow, so it runs once that refresh completes + ge = _make_isolation_ge_cloud() + option_calls = [] + + async def record_options(device, registers): + option_calls.append((device, registers[120]["value"])) + + ge.enable_default_options = record_options + await ge.run(seconds=0, first=True) + ge._read_only = False + ge.default_options_stamp = ge.now_utc_exact - timedelta(hours=25) + ge.hold_reads = asyncio.Event() + ge.read_values[120] = "06:00" + await asyncio.wait_for(ge.run(seconds=600, first=False), 5) + if option_calls: + print("ERROR: Default options ran on the settings from before the refresh: {}".format(option_calls)) + failed += 1 + ge.hold_reads.set() + await asyncio.wait_for(ge.settings_refresh_task, 5) + if option_calls != [("inv001", "06:00")]: + print("ERROR: Default options did not run on the fresh settings: {}".format(option_calls)) + failed += 1 + + # A refresh that raises fails the cycle, with the traceback logged, and the next cycle is clean + ge = _make_isolation_ge_cloud() + await ge.run(seconds=0, first=True) + + async def broken(*args, **kwargs): + raise ValueError("bad register list") + + ge.async_get_inverter_settings = broken + result = await asyncio.wait_for(ge.run(seconds=600, first=False), 5) + if result is not False or not any("Traceback" in message for message in ge.log_messages): + print("ERROR: A failed background refresh was not reported: result {}".format(result)) + failed += 1 + if await ge.run(seconds=660, first=False) is not True: + print("ERROR: The cycle after a reported refresh failure should succeed") + failed += 1 return failed return run_async(test()) From 51eaca85ee38ca17cc134b640d6c4edaa06b6a9b Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 12:38:42 +0100 Subject: [PATCH 3/4] fix(gecloud): time-based settings read backoff capped at an hour; an outage clears it The backoff was counted in refreshes, so on an EMS member inverter (read once an hour) the 6-refresh cap left a register unread for six hours. It is now time-based: not read for 15, then 30, then at most 60 minutes, whatever the device's refresh cadence. A refresh in which no register answers (the inverter or GE Cloud is down) now clears the device's backoff, so every register is read again the moment the device is back. Such a refresh already did not count towards backing a register off. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gecloud.py | 30 +++++++++++++++++------------ apps/predbat/tests/test_ge_cloud.py | 27 +++++++++++++++++++------- 2 files changed, 38 insertions(+), 19 deletions(-) diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 19b119c43..28460d2d4 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -77,11 +77,14 @@ # burns its whole retry budget, so a refresh can run for many minutes. It runs as its own task and the # cycle waits this long for it before carrying on, so live status and meter data keep being published. SETTINGS_REFRESH_WAIT_SECONDS = 20 -# A setting whose read has failed this many periodic refreshes in a row is backed off: skipped for the -# next refresh, then 2, 4... up to SETTING_READ_BACKOFF_MAX_SKIP refreshes, and tried again after that. -# Its last known value is kept and published as a failed readback. +# A setting whose read has failed this many periodic refreshes in a row, while the device answered its +# other reads, is backed off: not read for SETTING_READ_BACKOFF_MINUTES, doubling with each further +# failure up to SETTING_READ_BACKOFF_MAX_MINUTES, then tried again. Its last known value is kept and +# published as a failed readback. Time-based, so no register goes unread for more than the cap whatever +# the device's refresh cadence. SETTING_READ_BACKOFF_FAILURES = 3 -SETTING_READ_BACKOFF_MAX_SKIP = 6 +SETTING_READ_BACKOFF_MINUTES = 15 +SETTING_READ_BACKOFF_MAX_MINUTES = 60 # An inverter under EMS control is expected to leave its own slot 1 windows spanning the whole day so # that they never override a command from the EMS (see #3781). Anything else is reported. @@ -653,7 +656,7 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e # a settings read can tell whether a write landed after it was issued self.write_seq = 0 self.recent_writes = {} - # serial -> setting id -> {"failures": consecutive failed reads, "skip": refreshes left to skip} + # serial -> setting id -> {"failures": consecutive failed reads, "until": not read again before this time} self.setting_read_backoff = {} self.default_options_stamp = None # Devices found by the periodic device check still waiting for their default options, kept across @@ -2784,8 +2787,7 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, future["skipped"] = False future["issue_seq"] = refresh_seq state = device_backoff.get(sid, None) - if backoff and not first and state and state["skip"] > 0: - state["skip"] -= 1 + if backoff and not first and state and state.get("until") and self.now_utc_exact < state["until"]: future["skipped"] = True future["data"] = None complete.append(future) @@ -2824,6 +2826,9 @@ async def async_get_inverter_settings(self, serial, first=False, previous=None, # A failure only counts towards backing a register off when the device answered other reads in the # same refresh: when nothing reads at all the device or GE Cloud is down, not that one register device_answered = any(future["data"] and ("value" in future["data"]) for future in complete) + if not first and not device_answered and any(not future["skipped"] for future in complete): + # Nothing answered: start every register afresh, so all are read as soon as the device is back + device_backoff.clear() # Wait for all the futures to complete and store results for future in complete: @@ -2874,8 +2879,8 @@ def record_setting_read(self, device_backoff, serial, sid, name, success): Track consecutive failed reads of a setting and back it off once it keeps failing An inverter can list a register it cannot serve (for example AC charge slots 8 and 9), and every - read of it times out through the whole retry budget. Backed off, it is skipped for 1, 2, 4... up - to SETTING_READ_BACKOFF_MAX_SKIP refreshes and then tried again, so it is never hidden for good. + read of it times out through the whole retry budget. Backed off, it is not read for 15, 30, then + at most 60 minutes before it is tried again, so it is never hidden for good. device_backoff is the device's entry in setting_read_backoff taken when the reads started, so a device dropped by a device list refresh meanwhile is not added back. """ @@ -2884,11 +2889,12 @@ def record_setting_read(self, device_backoff, serial, sid, name, success): if state and state["failures"] >= SETTING_READ_BACKOFF_FAILURES: self.log("GECloud: Device {} setting id {} ({}) read again after {} failed refreshes".format(serial, sid, name, state["failures"])) return - state = device_backoff.setdefault(sid, {"failures": 0, "skip": 0}) + state = device_backoff.setdefault(sid, {"failures": 0, "until": None}) state["failures"] += 1 if state["failures"] >= SETTING_READ_BACKOFF_FAILURES: - state["skip"] = min(2 ** (state["failures"] - SETTING_READ_BACKOFF_FAILURES), SETTING_READ_BACKOFF_MAX_SKIP) - self.log("GECloud: Warn: Device {} setting id {} ({}) has failed {} refreshes in a row, skipping it for the next {} refreshes and keeping its last known value".format(serial, sid, name, state["failures"], state["skip"])) + minutes = min(SETTING_READ_BACKOFF_MINUTES * 2 ** (state["failures"] - SETTING_READ_BACKOFF_FAILURES), SETTING_READ_BACKOFF_MAX_MINUTES) + state["until"] = self.now_utc_exact + timedelta(minutes=minutes) + self.log("GECloud: Warn: Device {} setting id {} ({}) has failed {} refreshes in a row, not reading it for {} minutes and keeping its last known value".format(serial, sid, name, state["failures"], minutes)) async def async_get_smart_device_data(self, uuid): """ diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index bf0272bba..f6a0ed333 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -11,7 +11,7 @@ from gecloud import GECloudDirect, GECloudData, regname_to_ha from gecloud import GE_API_ACCOUNT, GE_API_DEVICES, GE_API_EVC_SEND_COMMAND, GE_API_INVERTER_WRITE_SETTING, GE_API_SITE -from gecloud import GE_API_INVERTER_READ_SETTING, GE_API_INVERTER_SETTINGS, SETTING_READ_BACKOFF_MAX_SKIP +from gecloud import GE_API_INVERTER_READ_SETTING, GE_API_INVERTER_SETTINGS, SETTING_READ_BACKOFF_MAX_MINUTES from gecloud import EVCCommandFailed, GECloudTerminalError, SITE_MAX_AGE_MINUTES, parse_site_export_limit from gecloud import DEVICE_REFRESH_SECONDS, SETTINGS_SLOW_REFRESH_SECONDS, find_ems_slot_overrides, normalise_register_time from utils import dp4 @@ -8839,12 +8839,15 @@ async def test(): ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous={}) ge.read_values[120] = None read_pattern = [] - for _ in range(9): + for _ in range(17): + # A refresh every 10 minutes, as in polling mode + ge._now_utc_exact += timedelta(minutes=10) before = len(ge.read_calls) ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) read_pattern.append(any(call[1] == 120 for call in ge.read_calls[before:])) - # Fails 1, 2, 3 (back off 1), skip, fails 4 (back off 2), skip, skip, fails 5 (back off 4), skip - expected = [True, True, True, False, True, False, False, True, False] + # Fails 1, 2, 3 (15 min off: skip 1), fails 4 (30 min: skip 2), fails 5 (60 min: skip 5), + # fails 6 (capped at 60 min: skip 5), read again + expected = [True, True, True, False, True, False, False, True, False, False, False, False, False, True, False, False, False] if read_pattern != expected: print("ERROR: Backoff read pattern {} expected {}".format(read_pattern, expected)) failed += 1 @@ -8867,7 +8870,8 @@ async def test(): # Good again: read on the next try after the backoff, and the flag and backoff are cleared ge.read_values[120] = "03:00" - for _ in range(SETTING_READ_BACKOFF_MAX_SKIP + 1): + for _ in range(SETTING_READ_BACKOFF_MAX_MINUTES // 10 + 1): + ge._now_utc_exact += timedelta(minutes=10) ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) register = ge.settings["inv001"][120] if register.get("value") != "03:00" or register.get("read_failed") or 120 in ge.setting_read_backoff.get("inv001", {}): @@ -8878,15 +8882,24 @@ async def test(): failed += 1 # The device or GE Cloud being down fails every read: that is not a register it cannot serve, so - # nothing is backed off and every register is read again as soon as it is back + # nothing is backed off - even a register already backed off is read the moment the device is back + ge.read_values = {17: True, 120: None} + for _ in range(3): + ge._now_utc_exact += timedelta(minutes=10) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if not ge.setting_read_backoff.get("inv001", {}).get(120, {}).get("until"): + print("ERROR: Expected id 120 backed off before the outage: {}".format(ge.setting_read_backoff)) + failed += 1 ge.read_values = {17: None, 120: None} - for _ in range(SETTING_READ_BACKOFF_MAX_SKIP + 2): + for _ in range(8): + ge._now_utc_exact += timedelta(minutes=1) ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) if ge.setting_read_backoff.get("inv001"): print("ERROR: An outage backed registers off: {}".format(ge.setting_read_backoff)) failed += 1 ge.read_values = {17: True, 120: "04:00"} before = len(ge.read_calls) + ge._now_utc_exact += timedelta(minutes=1) ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) if sorted(call[1] for call in ge.read_calls[before:]) != [17, 120] or ge.settings["inv001"][120]["value"] != "04:00": print("ERROR: Registers not read on the first refresh after an outage: {}".format(ge.read_calls[before:])) From fb81519b13a67b8c3172c7705a24e4cd7a8af845 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 12:47:19 +0100 Subject: [PATCH 4/4] fix(gecloud): no default options on stale settings after a failed refresh; test the hourly backoff cap If the background settings refresh raised within the wait, the cycle still treated the settings as fresh, and ran the default options pass and the AC charge gate re-check on the settings left from before. They now wait for a cycle with fresh settings. Added a test that a register which keeps failing on an hourly-refreshed EMS member inverter is still read on every hourly refresh and is never left unread for more than an hour. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gecloud.py | 3 +++ apps/predbat/tests/test_ge_cloud.py | 33 ++++++++++++++++++++++++++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 28460d2d4..2bf32ac40 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -2497,6 +2497,9 @@ async def run(self, seconds, first): self.settings_refresh_detached = True settings_fresh = False self.log("GECloud: Settings refresh of {} is taking longer than {} seconds, carrying on with live data while it completes".format(refresh_devices, SETTINGS_REFRESH_WAIT_SECONDS)) + elif self.settings_refresh_error: + # It failed, so the settings are the old ones: leave the defaults to a cycle with fresh settings + settings_fresh = False # Run at startup, and again whenever the periodic device check finds the hardware has changed if first or devices_changed: diff --git a/apps/predbat/tests/test_ge_cloud.py b/apps/predbat/tests/test_ge_cloud.py index f6a0ed333..23eca895f 100644 --- a/apps/predbat/tests/test_ge_cloud.py +++ b/apps/predbat/tests/test_ge_cloud.py @@ -8904,6 +8904,28 @@ async def test(): if sorted(call[1] for call in ge.read_calls[before:]) != [17, 120] or ge.settings["inv001"][120]["value"] != "04:00": print("ERROR: Registers not read on the first refresh after an outage: {}".format(ge.read_calls[before:])) failed += 1 + + # An EMS member inverter is only re-read once an hour: a register that keeps failing there is still read on + # every one of those refreshes, as the backoff never keeps a register unread for more than an hour + ge = _make_isolation_ge_cloud() + ge.device_list = ["inv001"] + ge.pending_writes = {"inv001": []} + with patch("gecloud.RETRIES", 1), patch("gecloud.RETRY_FACTOR", 0): + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous={}) + ge.read_values[120] = None + last_read = ge.now_utc_exact + longest_gap = timedelta(0) + for _ in range(8): + ge._now_utc_exact += timedelta(minutes=SETTING_READ_BACKOFF_MAX_MINUTES) + before = len(ge.read_calls) + ge.settings["inv001"] = await ge.async_get_inverter_settings("inv001", previous=ge.settings["inv001"], backoff=True) + if any(call[1] == 120 for call in ge.read_calls[before:]): + longest_gap = max(longest_gap, ge.now_utc_exact - last_read) + last_read = ge.now_utc_exact + longest_gap = max(longest_gap, ge.now_utc_exact - last_read) + if longest_gap > timedelta(minutes=SETTING_READ_BACKOFF_MAX_MINUTES) or ge.setting_read_backoff["inv001"][120]["failures"] != 8: + print("ERROR: Hourly refreshed register went {} unread: {}".format(longest_gap, ge.setting_read_backoff)) + failed += 1 return failed return run_async(test()) @@ -8962,9 +8984,15 @@ async def record_options(device, registers): print("ERROR: Default options did not run on the fresh settings: {}".format(option_calls)) failed += 1 - # A refresh that raises fails the cycle, with the traceback logged, and the next cycle is clean + # A refresh that raises fails the cycle, with the traceback logged, and the next cycle is clean. + # The default options pass is due, but must not run on the settings the failed refresh left behind ge = _make_isolation_ge_cloud() + option_calls = [] + ge.enable_default_options = record_options await ge.run(seconds=0, first=True) + option_calls.clear() + ge._read_only = False + ge.default_options_stamp = ge.now_utc_exact - timedelta(hours=25) async def broken(*args, **kwargs): raise ValueError("bad register list") @@ -8974,6 +9002,9 @@ async def broken(*args, **kwargs): if result is not False or not any("Traceback" in message for message in ge.log_messages): print("ERROR: A failed background refresh was not reported: result {}".format(result)) failed += 1 + if option_calls: + print("ERROR: Default options ran on stale settings after the refresh failed: {}".format(option_calls)) + failed += 1 if await ge.run(seconds=660, first=False) is not True: print("ERROR: The cycle after a reported refresh failure should succeed") failed += 1