fix(output): serialise status recording across threads

Component threads (GE Cloud, Solis, ...) call record_status() alongside the
main loop. The error_count attribute is a read-modify-write of the sensor's
own state, so two concurrent calls could both read N and both write N+1.
record_status() and record_status_under_warning() now share a status lock,
which also keeps each call's current_status/status_warning update whole.

It does not cover record_final_run_status()'s choice between them, which
reads had_errors outside the lock: a component warning landing between
that read and the write is still overwritten until the next run, as
before this change.

The status-change notification is sent after the lock is released: over
the websocket it waits on another thread for up to two minutes, which would
otherwise stall every other status write. The lock is kept out of the debug
dump, which serialises PredBat's attributes and can't pickle a lock.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: chalfontchubby

apps/predbat/output.py

@@ -2760,67 +2760,77 @@ def record_status(self, message, debug="", had_errors=False, notify=False, extra
         """
         Records status to HA sensor
         """
-        if not extra:
-            extra = ""
-
-        # A warning that repeats the one already shown behind the previous run's state keeps that
-        # state in front of it, rather than flipping the sensor back to the bare warning for the few
-        # seconds until this run's own state is put back - two state changes every cycle for as long
-        # as the warning recurs. Any different warning or error, including the ones a run records
-        # when it bails out without executing, is shown bare.
-        state_message = message
-        if had_errors and self.current_status and self.current_status != message and self.current_status.endswith(", " + message):
-            state_message = self.current_status
-
-        self.current_status = state_message + extra
-        if notify and self.previous_status != message and self.set_status_notify:
-            if self.had_errors and had_errors:
-                # Already in error state, do not notify second error in a single run (spam)
-                pass
-            else:
-                self.call_notify(f"{self.prefix.capitalize()} status change to: {message}{extra}")
-                self.previous_status = message
-
-        error_count = self.get_state_wrapper(self.prefix + ".status", attribute="error_count", default=0)
-        try:
-            error_count = int(error_count)
-        except (ValueError, TypeError):
-            error_count = 0
-
-        if had_errors:
-            error_count += 1
+        # Component threads (GE Cloud, Solis, ...) record status too, so the read-modify-write of
+        # error_count and the current_status/status_warning pair must not interleave with the main loop.
+        # The sensor write under the lock is a REST call with a timeout, so a slow Home Assistant does
+        # hold up other status writes. The notification is sent after the lock is released: over the
+        # websocket it waits on another thread for up to two minutes, which would stall them far longer.
+        notify_message = None
+        with self.status_lock:
+            if not extra:
+                extra = ""
+
+            # A warning that repeats the one already shown behind the previous run's state keeps that
+            # state in front of it, rather than flipping the sensor back to the bare warning for the few
+            # seconds until this run's own state is put back - two state changes every cycle for as long
+            # as the warning recurs. Any different warning or error, including the ones a run records
+            # when it bails out without executing, is shown bare.
+            state_message = message
+            if had_errors and self.current_status and self.current_status != message and self.current_status.endswith(", " + message):
+                state_message = self.current_status
+
+            self.current_status = state_message + extra
+            if notify and self.previous_status != message and self.set_status_notify:
+                if self.had_errors and had_errors:
+                    # Already in error state, do not notify second error in a single run (spam)
+                    pass
+                else:
+                    notify_message = f"{self.prefix.capitalize()} status change to: {message}{extra}"
+                    self.previous_status = message
+
+            error_count = self.get_state_wrapper(self.prefix + ".status", attribute="error_count", default=0)
+            try:
+                error_count = int(error_count)
+            except (ValueError, TypeError):
+                error_count = 0
+
+            if had_errors:
+                error_count += 1
+
+            # Home Assistant rejects entity states over 255 characters, and this message is the state
+            # of the status sensor. Clamp what is written as the state - the full text survives in
+            # current_status, the log line and the notification, and attributes have no such cap.
+            # Motivated by the window warnings listing every configured inverter component (#4990):
+            # three or more of those push past 255, so the dashboard would keep a stale status on
+            # exactly the cycles the warning matters.
+            self.dashboard_item(
+                self.prefix + ".status",
+                state=state_message[:255],
+                attributes={
+                    "friendly_name": "Status",
+                    "detail": extra,
+                    "icon": "mdi:information",
+                    "last_updated": self.now_utc_real.strftime(TIME_FORMAT),
+                    "debug": debug,
+                    "version": THIS_VERSION_DISPLAY,
+                    "error": (had_errors or self.had_errors),
+                    "error_count": error_count,
+                },
+            )
 
-        # Home Assistant rejects entity states over 255 characters, and this message is the state
-        # of the status sensor. Clamp what is written as the state - the full text survives in
-        # current_status, the log line and the notification, and attributes have no such cap.
-        # Motivated by the window warnings listing every configured inverter component (#4990):
-        # three or more of those push past 255, so the dashboard would keep a stale status on
-        # exactly the cycles the warning matters.
-        self.dashboard_item(
-            self.prefix + ".status",
-            state=state_message[:255],
-            attributes={
-                "friendly_name": "Status",
-                "detail": extra,
-                "icon": "mdi:information",
-                "last_updated": self.now_utc_real.strftime(TIME_FORMAT),
-                "debug": debug,
-                "version": THIS_VERSION_DISPLAY,
-                "error": (had_errors or self.had_errors),
-                "error_count": error_count,
-            },
-        )
+            if had_errors:
+                self.log("Warn: record_status {}".format(message + extra))
+            else:
+                self.log("Info: record_status {}".format(message + extra))
 
-        if had_errors:
-            self.log("Warn: record_status {}".format(message + extra))
-        else:
-            self.log("Info: record_status {}".format(message + extra))
+            if had_errors:
+                self.had_errors = True
+                # Kept so the end of the run can put the state it executed back in front of it
+                self.status_warning = message
+                self.status_warning_debug = debug
 
-        if had_errors:
-            self.had_errors = True
-            # Kept so the end of the run can put the state it executed back in front of it
-            self.status_warning = message
-            self.status_warning_debug = debug
+        if notify_message:
+            self.call_notify(notify_message)
 
     def record_status_under_warning(self, run_status):
         """
@@ -2837,29 +2847,31 @@ def record_status_under_warning(self, run_status):
         Does nothing when no warning was recorded this run (e.g. a component thread set had_errors
         on its own), leaving the sensor as it was.
         """
-        if not self.status_warning:
-            return
-        error_count = self.get_state_wrapper(self.prefix + ".status", attribute="error_count", default=0)
-        try:
-            error_count = int(error_count)
-        except (ValueError, TypeError):
-            error_count = 0
-        message = "{}, {}".format(run_status, self.status_warning)
-        self.current_status = message
-        self.dashboard_item(
-            self.prefix + ".status",
-            state=message[:255],
-            attributes={
-                "friendly_name": "Status",
-                "detail": "",
-                "icon": "mdi:information",
-                "last_updated": self.now_utc_real.strftime(TIME_FORMAT),
-                "debug": self.status_warning_debug,
-                "version": THIS_VERSION_DISPLAY,
-                "error": True,
-                "error_count": error_count,
-            },
-        )
+        # Same lock as record_status() - see there.
+        with self.status_lock:
+            if not self.status_warning:
+                return
+            error_count = self.get_state_wrapper(self.prefix + ".status", attribute="error_count", default=0)
+            try:
+                error_count = int(error_count)
+            except (ValueError, TypeError):
+                error_count = 0
+            message = "{}, {}".format(run_status, self.status_warning)
+            self.current_status = message
+            self.dashboard_item(
+                self.prefix + ".status",
+                state=message[:255],
+                attributes={
+                    "friendly_name": "Status",
+                    "detail": "",
+                    "icon": "mdi:information",
+                    "last_updated": self.now_utc_real.strftime(TIME_FORMAT),
+                    "debug": self.status_warning_debug,
+                    "version": THIS_VERSION_DISPLAY,
+                    "error": True,
+                    "error_count": error_count,
+                },
+            )
 
     def publish_last_started(self):
         """

apps/predbat/predbat.py

@@ -25,6 +25,7 @@
 import gc
 import random
 import time
+import threading
 
 # from memory_profiler import profile
 
@@ -369,6 +370,7 @@ def reset(self):
         self.had_errors = False
         self.status_warning = None
         self.status_warning_debug = ""
+        self.status_lock = threading.RLock()
         self.plan_valid = False
         self.plan_preclip = None
         self.plan_last_updated = None

apps/predbat/tests/test_component_health_status.py

@@ -9,6 +9,9 @@
 # pylint: disable=attribute-defined-outside-init
 # fmt on
 
+import threading
+import time
+
 from output import split_status_warning
 
 
@@ -182,6 +185,101 @@ def test_record_status_under_warning(my_predbat):
     return failed
 
 
+def test_record_status_concurrent(my_predbat):
+    """
+    Verify record_status() calls from several threads at once don't lose an error_count increment,
+    and that the status notification is sent with the lock released.
+
+    Component threads (GE Cloud, Solis, ...) record status alongside the main loop, and error_count is
+    a read-modify-write of the sensor's own attribute. The read is slowed here so that, without the
+    status lock, the threads reliably read the same count and overwrite each other's increment.
+    """
+    print("*** Running test: concurrent record_status calls keep every error_count increment")
+    failed = 0
+    status_entity = my_predbat.prefix + ".status"
+    saved_item = my_predbat.ha_interface.dummy_items.get(status_entity)
+    saved_value = my_predbat.dashboard_values.get(status_entity)
+    saved = (my_predbat.current_status, my_predbat.had_errors, my_predbat.status_warning, my_predbat.status_warning_debug)
+    original_get_state_wrapper = my_predbat.get_state_wrapper
+    original_call_notify = my_predbat.call_notify
+    saved_notify_flag = my_predbat.set_status_notify
+    saved_previous_status = my_predbat.previous_status
+    threads_count, calls_each = 4, 10
+
+    def _slow_get_state_wrapper(*args, **kwargs):
+        value = original_get_state_wrapper(*args, **kwargs)
+        time.sleep(0.001)
+        return value
+
+    def _record_warnings():
+        for _ in range(calls_each):
+            my_predbat.record_status("Warn: concurrent test", had_errors=True)
+
+    try:
+        my_predbat.dashboard_values.pop(status_entity, None)
+        my_predbat.ha_interface.dummy_items.pop(status_entity, None)
+        my_predbat.get_state_wrapper = _slow_get_state_wrapper
+        threads = [threading.Thread(target=_record_warnings) for _ in range(threads_count)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        error_count = my_predbat.dashboard_values[status_entity]["attributes"]["error_count"]
+        if error_count != threads_count * calls_each:
+            print("ERROR: expected error_count {} after {} concurrent warnings, got {}".format(threads_count * calls_each, threads_count * calls_each, error_count))
+            failed = 1
+        else:
+            print("OK: all {} concurrent warnings counted".format(error_count))
+        my_predbat.get_state_wrapper = original_get_state_wrapper
+
+        # The notification is sent with the lock released - over the websocket it can wait minutes
+        # on another thread, and every other status would stall behind it
+        notified = []
+
+        def _fake_notify(message):
+            result = {}
+
+            def _try_lock():
+                result["free"] = my_predbat.status_lock.acquire(blocking=False)
+                if result["free"]:
+                    my_predbat.status_lock.release()
+
+            checker = threading.Thread(target=_try_lock)
+            checker.start()
+            checker.join()
+            notified.append((message, result["free"]))
+
+        my_predbat.call_notify = _fake_notify
+        my_predbat.set_status_notify = True
+        my_predbat.previous_status = None
+        my_predbat.had_errors = False
+        my_predbat.record_status("Exporting", notify=True)
+        if len(notified) != 1 or "Exporting" not in notified[0][0]:
+            print("ERROR: expected one 'Exporting' notification, got {}".format(notified))
+            failed = 1
+        elif not notified[0][1]:
+            print("ERROR: the notification was sent while the status lock was held")
+            failed = 1
+        else:
+            print("OK: status notification sent with the lock released")
+    finally:
+        my_predbat.get_state_wrapper = original_get_state_wrapper
+        my_predbat.call_notify = original_call_notify
+        my_predbat.set_status_notify = saved_notify_flag
+        my_predbat.previous_status = saved_previous_status
+        my_predbat.current_status, my_predbat.had_errors, my_predbat.status_warning, my_predbat.status_warning_debug = saved
+        if saved_item is None:
+            my_predbat.ha_interface.dummy_items.pop(status_entity, None)
+        else:
+            my_predbat.ha_interface.dummy_items[status_entity] = saved_item
+        if saved_value is None:
+            my_predbat.dashboard_values.pop(status_entity, None)
+        else:
+            my_predbat.dashboard_values[status_entity] = saved_value
+
+    return failed
+
+
 def test_component_health_status(my_predbat):
     """
     Verify record_final_run_status() marks the run as an error, naming the failed component(s),

apps/predbat/unit_test.py

@@ -67,7 +67,7 @@
 from tests.test_plugin_startup import test_plugin_startup_order
 from tests.test_active_flag import test_active_flag
 from tests.test_time_loop_guard import test_time_loop_guard
-from tests.test_component_health_status import test_component_health_status, test_record_status_state_clamped, test_record_status_under_warning
+from tests.test_component_health_status import test_component_health_status, test_record_status_state_clamped, test_record_status_under_warning, test_record_status_concurrent
 from tests.test_optimise_levels import run_optimise_levels_tests
 from tests.test_trim_export import run_trim_export_tests
 from tests.test_plan_tiebreak import run_plan_tiebreak_tests
@@ -516,6 +516,7 @@ def main():
         ("component_health_status", test_component_health_status, "Component errors fail the recorded run status tests", False),
         ("record_status_state_clamped", test_record_status_state_clamped, "Status sensor state is clamped at the 255 characters Home Assistant accepts", False),
         ("record_status_under_warning", test_record_status_under_warning, "A run's state is recorded in front of the warning it raised", False),
+        ("record_status_concurrent", test_record_status_concurrent, "Concurrent record_status calls keep every error_count increment", False),
         ("dynamic_load_car", test_dynamic_load_car_slot_cancellation, "Dynamic load car slot cancellation tests", False),
         ("dynamic_load_high", test_dynamic_load_high_load_baseline, "Dynamic load high-load baseline tests", False),
         ("dynamic_load_car_not_charging", test_dynamic_load_car_not_charging, "Dynamic load car not charging tests", False),

apps/predbat/utils.py

@@ -231,7 +231,8 @@ def copy(self):
 
 
 # Predbat member variables never included in a debug dump or served over MCP - live object
-# graphs, the HA interface, loaded secrets and the URL caches. Shared with is_debug_excluded_key().
+# graphs, the HA interface, loaded secrets, the URL caches and locks (which can't be serialised).
+# Shared with is_debug_excluded_key().
 DEBUG_EXCLUDE_LIST = [
     "ha_interface",
     "components",
@@ -252,6 +253,7 @@ def copy(self):
     "github_url_cache",
     "octopus_url_cache",
     "secrets",
+    "status_lock",
 ]