From 13e78d962f3618a6afacd063adf7cd713eff3a2e Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 15:46:00 +0100 Subject: [PATCH 01/11] fix(gateway): send each hub command once and wait for its ack The generic write_and_poll loop re-calls the gateway's select/number/switch handlers every 2 s until the read-back matches, and each call published the command again. A slow dongle write therefore queued up to ten Modbus writes on the hub, and PredBat reported writes as failed that the hub was still working through. The hub already publishes an ack per command on predbat/devices//ack/. Subscribe to it and track control writes per entity: once an ack has been matched on this connection, an identical command is published once per 30 s window (while waiting for the ack, after an ok ack, and after a refusal), a different value goes out immediately, and an unanswered command is re-sent once before acks stop being relied on. Ok acks update cached slot times and enable switches from the applied values; refusals are logged with the hub's reason; a "replay" rejection re-sends under a new id. Without the ack subscription, or before any ack is seen, behaviour is unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- .cspell/custom-dictionary-workspace.txt | 1 + apps/predbat/gateway.py | 221 ++++++++++++++++-- apps/predbat/tests/test_gateway.py | 299 ++++++++++++++++++++++++ 3 files changed, 504 insertions(+), 17 deletions(-) diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index b6d763b77..b99adc822 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -1,6 +1,7 @@ # Custom Dictionary Words 3hy aaaabbbbccccdddd +acks activepower adata adbase diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index d783cc3de..c9ff72050 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -12,6 +12,7 @@ import math import os import ssl +import threading import time import uuid import traceback @@ -55,6 +56,11 @@ # Telemetry staleness threshold (seconds) _TELEMETRY_STALE_THRESHOLD = 120 +# How long an identical control command waits for the hub's ack (or, after an ack, +# for telemetry to catch up) before it may be published again (seconds). Longer than +# the generic write_and_poll loop (10 x 2 s), so one write sends the command once. +_COMMAND_ACK_WINDOW = 30 + # Total startup wait budget, in 0.5 s ticks, shared by the connection and auto-config waits _STARTUP_WAIT_TICKS = 120 * 2 _STARTUP_WAIT_SECONDS = _STARTUP_WAIT_TICKS * 0.5 @@ -184,6 +190,10 @@ class GatewayMQTT(ComponentBase): Instance methods handle MQTT lifecycle and ComponentBase integration. """ + # Guards the command id counter and the in-flight command map, which are touched both by + # control writes (engine thread) and by acks arriving on the MQTT listener's loop. + _command_lock = threading.Lock() + def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqtt_token=None, gateway_inverter_serial=None, gateway_evc_automatic=False, gateway_evc_control=False, **kwargs): """Initialize gateway configuration and build MQTT topic strings. @@ -240,6 +250,8 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self.topic_schedule = f"{self._topic_base}/schedule" self.topic_command = f"{self._topic_base}/command" self.topic_ev_command = f"{self._topic_base}/ev/command" + self.topic_ack = f"{self._topic_base}/ack/+" + self._ack_topic_prefix = f"{self._topic_base}/ack/" # Runtime state self._mqtt_client = None @@ -281,6 +293,14 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self._suffix_to_serial = {} # maps entity suffix (last 6 chars of serial) -> full serial string self._command_id = 0 # incrementing counter included in every published command + # Command acks (predbat/devices//ack/). Control writes are + # tracked per entity so an identical re-send can wait for the hub's ack + # instead of queueing another Modbus write every poll. See _send_control(). + self._pending_commands = {} # entity_id -> {command_id, command, kwargs, sent_at, state, resent} + self._ack_subscribed = False # broker granted the ack subscription on this connection + self._acks_seen = False # at least one ack matched a tracked command on this connection + self._ack_subscribe_warned = False # the denied-subscription warning is logged once + # Predbat data publish state (price/timeline for device display) self._last_predbat_data = None @@ -646,6 +666,12 @@ async def _mqtt_loop(self): await client.subscribe(self.topic_online, qos=1) self.log(f"Info: GatewayMQTT: Subscribed to {self.topic_status} and {self.topic_online}") + # Command acks are optional: without them control writes behave as before + with self._command_lock: + self._pending_commands = {} + self._acks_seen = False + self._ack_subscribed = await self._subscribe_acks(client) + async for message in client.messages: if self.api_stop: break @@ -659,6 +685,7 @@ async def _mqtt_loop(self): self.log(f"Warn: GatewayMQTT: MQTT connection error: {e}") self._mqtt_connected = False self._mqtt_client = None + self._ack_subscribed = False self._first_connection_attempted = True if self.api_stop: @@ -675,6 +702,28 @@ async def _mqtt_loop(self): self._mqtt_connected = False self._mqtt_client = None + self._ack_subscribed = False + + async def _subscribe_acks(self, client): + """Subscribe to the hub's per-command acks, tolerating a broker that refuses it. + + Returns: + bool: True if the broker granted the subscription. + """ + try: + result = await client.subscribe(self.topic_ack, qos=0) + code = result[0] if result else 0x80 + code = int(getattr(code, "value", code)) + if code < 0x80: + self.log(f"Info: GatewayMQTT: Subscribed to {self.topic_ack}") + return True + reason = f"refused by broker, code {code}" + except Exception as e: + reason = str(e) + if not self._ack_subscribe_warned: + self._ack_subscribe_warned = True + self.log(f"Warn: GatewayMQTT: Cannot subscribe to command acks ({reason}) - control writes will be re-sent until telemetry confirms them") + return False async def _handle_message(self, message): """Dispatch an incoming MQTT message to the appropriate handler. @@ -687,6 +736,8 @@ async def _handle_message(self, message): try: if topic == self.topic_status: self._process_telemetry(message.payload) + elif topic.startswith(getattr(self, "_ack_topic_prefix", "\0")): + self._process_ack(message.payload) elif topic == self.topic_online: payload = message.payload.decode("utf-8", errors="replace").strip() was_online = self._gateway_online @@ -919,12 +970,7 @@ def _inject_inverter_entities(self, inv, suffix): ("discharge_end", "discharge_slot1_end"), ]: hhmm = getattr(sched, field, 0) if sched else 0 - hours = hhmm // 100 - minutes = hhmm % 100 - if hours >= 24: - hours = 0 # firmware sends 2400 for midnight end-of-day - time_str = f"{hours:02d}:{minutes:02d}:00" - self.dashboard_item(f"select.{pfx}_{name}", time_str, attributes=GATEWAY_ATTRIBUTE_TABLE.get(name, {}), app="gateway") + self.dashboard_item(f"select.{pfx}_{name}", self._hhmm_to_time_str(hhmm), attributes=GATEWAY_ATTRIBUTE_TABLE.get(name, {}), app="gateway") # Inverter time (from GatewayStatus timestamp for clock drift detection) if self._last_status and self._last_status.timestamp: @@ -1667,15 +1713,23 @@ async def _republish_plan_if_stale(self): self._last_plan_publish_time = time.time() self.log("Info: GatewayMQTT: Re-published execution plan (refreshed timestamp)") - async def publish_command(self, command, **kwargs): + def _next_command_id(self): + """Allocate the next command id number (callers may be on different threads).""" + with self._command_lock: + self._command_id += 1 + return self._command_id + + async def publish_command(self, command, command_id=None, **kwargs): """Build and publish a JSON command to the gateway. Args: command: Command name (set_charge_rate, set_reserve, etc.) + command_id: Pre-allocated id number (from _next_command_id); allocated here when None. **kwargs: Command-specific fields (power_w, target_soc, etc.). """ - self._command_id += 1 - cmd_json = self.build_command(command, command_id=self._command_id, **kwargs) + if command_id is None: + command_id = self._next_command_id() + cmd_json = self.build_command(command, command_id=command_id, **kwargs) self.log("Info: GatewayMQTT: publish_command: command={}, payload={}".format(command, cmd_json)) @@ -1797,7 +1851,7 @@ async def _update_charge_slot(self, entity_id, hhmm, serial): schedule = {"start": hhmm} else: schedule = {"end": hhmm} - await self.publish_command("set_charge_slot", schedule_json=json.dumps(schedule), serial=serial) + await self._send_control(entity_id, "set_charge_slot", schedule_json=json.dumps(schedule), serial=serial) self.log(f"Info: GatewayMQTT: Charge slot update: {schedule}") async def _update_discharge_slot(self, entity_id, hhmm, serial): @@ -1806,7 +1860,7 @@ async def _update_discharge_slot(self, entity_id, hhmm, serial): schedule = {"start": hhmm} else: schedule = {"end": hhmm} - await self.publish_command("set_discharge_slot", schedule_json=json.dumps(schedule), serial=serial) + await self._send_control(entity_id, "set_discharge_slot", schedule_json=json.dumps(schedule), serial=serial) self.log(f"Info: GatewayMQTT: Discharge slot update: {schedule}") async def number_event(self, entity_id, value): @@ -1829,13 +1883,13 @@ async def number_event(self, entity_id, value): self.log(f"Warn: GatewayMQTT: number_event: cannot resolve serial for entity '{entity_id}' — command not sent") return if "_discharge_rate" in entity_id: - await self.publish_command("set_discharge_rate", power_w=val, serial=serial) + await self._send_control(entity_id, "set_discharge_rate", power_w=val, serial=serial) elif "_charge_rate" in entity_id: - await self.publish_command("set_charge_rate", power_w=val, serial=serial) + await self._send_control(entity_id, "set_charge_rate", power_w=val, serial=serial) elif "_reserve" in entity_id: - await self.publish_command("set_reserve", target_soc=val, serial=serial) + await self._send_control(entity_id, "set_reserve", target_soc=val, serial=serial) elif "_target_soc" in entity_id: - await self.publish_command("set_target_soc", target_soc=val, serial=serial) + await self._send_control(entity_id, "set_target_soc", target_soc=val, serial=serial) async def switch_event(self, entity_id, service): """Handle switch entity service calls (charge/discharge enable). @@ -1864,12 +1918,145 @@ async def switch_event(self, entity_id, service): self.log(f"Warn: GatewayMQTT: switch_event: cannot resolve serial for entity '{entity_id}' — command not sent") return if "_charge_enabled" in entity_id: - await self.publish_command("set_charge_enable", enable=is_on, serial=serial) + await self._send_control(entity_id, "set_charge_enable", enable=is_on, serial=serial) self.log(f"Info: GatewayMQTT: Charge {'enabled' if is_on else 'disabled'}") elif "_discharge_enabled" in entity_id: - await self.publish_command("set_discharge_enable", enable=is_on, serial=serial) + await self._send_control(entity_id, "set_discharge_enable", enable=is_on, serial=serial) self.log(f"Info: GatewayMQTT: Discharge {'enabled' if is_on else 'disabled'}") + async def _send_control(self, entity_id, command, **kwargs): + """Publish a control write once and let the hub's ack confirm it. + + The generic write_and_poll loop calls the event handlers again every couple of + seconds until the read-back matches. Publishing on every call queued a fresh + Modbus write on the hub each time, so a slow dongle write turned into a storm. + Once the hub has shown it acks commands on this connection, an identical command + (same entity, command and payload) is published only once per _COMMAND_ACK_WINDOW: + while it waits for its ack, after an ok ack (telemetry may lag), and after a + refusal (re-sending cannot change the answer). A different value is published + immediately and replaces the tracked one. A command still unanswered after the window + is re-sent once; if that goes unanswered too, acks stop being relied on until the + next one arrives. With no ack subscription, or before any ack has been seen, + every call publishes exactly as before. + + Args: + entity_id: The entity being written, used as the tracking key. + command: Command name (set_charge_slot, set_charge_enable, ...). + **kwargs: Command fields passed on to publish_command. + """ + if not (getattr(self, "_ack_subscribed", False) and self._mqtt_connected): + await self.publish_command(command, **kwargs) + return + + now = time.time() + resent = False + with self._command_lock: + entry = self._pending_commands.get(entity_id) + same = entry is not None and entry["command"] == command and entry["kwargs"] == kwargs + if same and self._acks_seen: + if now - entry["sent_at"] < _COMMAND_ACK_WINDOW: + self.log(f"Info: GatewayMQTT: {command} for {kwargs.get('serial')} already sent as {entry['command_id']} ({entry['state']}), not re-sending") + return + if entry["state"] == "sent": + if entry["resent"]: + self._acks_seen = False + self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") + else: + resent = True + command_id = self._command_id = self._command_id + 1 + self._pending_commands[entity_id] = {"command_id": f"PBAT{command_id}", "command": command, "kwargs": dict(kwargs), "sent_at": now, "state": "sent", "resent": resent} + + try: + await self.publish_command(command, command_id=command_id, **kwargs) + except BaseException: + # Never sent, so nothing is in flight - do not let it suppress the next attempt + with self._command_lock: + if self._pending_commands.get(entity_id, {}).get("command_id") == f"PBAT{command_id}": + del self._pending_commands[entity_id] + raise + + def _process_ack(self, data): + """Handle a command ack from the hub. + + Args: + data: Raw JSON payload from predbat/devices//ack/. + """ + try: + ack = json.loads(data) + except (ValueError, TypeError) as e: + self.log(f"Warn: GatewayMQTT: Failed to decode command ack: {e}") + return + if not isinstance(ack, dict): + return + command_id = str(ack.get("command_id", "")) + ok = ack.get("ok") is True + error = ack.get("error") or "unknown" + + with self._command_lock: + entity_id, entry = next(((e, r) for e, r in self._pending_commands.items() if r["command_id"] == command_id), (None, None)) + # Acks for other senders, superseded values or already-answered commands are ignored + if entry is None or entry["state"] != "sent" or ack.get("command", entry["command"]) != entry["command"]: + return + self._acks_seen = True + if ok: + entry["state"] = "applied" + elif error == "replay": + # The hub has seen this id before (e.g. after a PredBat restart reset the + # counter), so the value was never considered: let the next attempt go out + # under a fresh id straight away. + del self._pending_commands[entity_id] + else: + entry["state"] = "refused" + entry["sent_at"] = time.time() + command = entry["command"] + serial = entry["kwargs"].get("serial") + + if ok: + self.log(f"Info: GatewayMQTT: {command} for {serial} acknowledged by hub ({command_id}) applied={ack.get('applied')}") + self._apply_ack(entity_id, command, ack.get("applied")) + elif error == "replay": + self.log(f"Info: GatewayMQTT: {command} for {serial} rejected as a replay of {command_id}, will re-send with a new id") + else: + self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") + + def _apply_ack(self, entity_id, command, applied): + """Update the cached entity from an ok ack so the write's read-back matches without waiting for telemetry. + + Only the values that map cleanly are applied: slot start/end actually written and + charge/discharge enables. Staged or recorded EMS slot endpoints are not written to + the inverter, and numbers wait for telemetry. + """ + if not isinstance(applied, dict): + return + if command in ("set_charge_slot", "set_discharge_slot"): + if applied.get("staged") or applied.get("recorded"): + return + start, end = applied.get("start"), applied.get("end") + if not all(isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= 2400 for v in (start, end)): + return + name = "discharge_slot1" if command == "set_discharge_slot" else "charge_slot1" + marker = f"_{name}_" + if marker not in entity_id: + return + base = entity_id[: entity_id.index(marker)] + for field, hhmm in (("start", start), ("end", end)): + self.dashboard_item(f"{base}_{name}_{field}", self._hhmm_to_time_str(hhmm), attributes=GATEWAY_ATTRIBUTE_TABLE.get(f"{name}_{field}", {}), app="gateway") + elif command in ("set_charge_enable", "set_discharge_enable"): + enable = applied.get("enable") + if not isinstance(enable, bool): + return + name = "discharge_enabled" if command == "set_discharge_enable" else "charge_enabled" + self.dashboard_item(entity_id, "on" if enable else "off", attributes=GATEWAY_ATTRIBUTE_TABLE.get(name, {}), app="gateway") + + @staticmethod + def _hhmm_to_time_str(hhmm): + """Convert an HHMM integer to the HH:MM:SS string the schedule selects use (2400 -> 00:00:00).""" + hours = hhmm // 100 + minutes = hhmm % 100 + if hours >= 24: + hours = 0 # firmware sends 2400 for midnight end-of-day + return f"{hours:02d}:{minutes:02d}:00" + async def final(self): """Cleanup: cancel listener task, disconnect.""" # Cancel the MQTT listener task diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 6d7d5d8e5..fb30ba910 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -4683,6 +4683,304 @@ def test_ev_soc_battery_size_through_get_arg(my_predbat): my_predbat.current_status = original_status +class TestCommandAck: + """Control writes are sent once and confirmed by the hub's ack instead of being re-sent every poll.""" + + SLOT_START = "select.predbat_gateway_456789_charge_slot1_start" + SLOT_END = "select.predbat_gateway_456789_charge_slot1_end" + CHARGE_ENABLED = "switch.predbat_gateway_456789_charge_enabled" + CHARGE_RATE = "number.predbat_gateway_456789_charge_rate" + + def _make_gateway(self, subscribed=True, acks_seen=True): + from gateway import GatewayMQTT + from unittest.mock import MagicMock + + gw = GatewayMQTT.__new__(GatewayMQTT) + gw.log = MagicMock() + gw.prefix = "predbat" + gw._suffix_to_serial = {"456789": "CE123456789"} + gw._mqtt_connected = True + gw._command_id = 0 + gw._pending_commands = {} + gw._ack_subscribed = subscribed + gw._acks_seen = acks_seen + gw._ack_subscribe_warned = False + gw.topic_status = "predbat/devices/pbgw_test/status" + gw.topic_online = "predbat/devices/pbgw_test/online" + gw.topic_ack = "predbat/devices/pbgw_test/ack/+" + gw._ack_topic_prefix = "predbat/devices/pbgw_test/ack/" + gw._published = [] # (command, command_id, kwargs) + gw.cache = {} # entity_id -> cached state, as the generic write loop would read it back + + async def fake_publish_command(command, command_id=None, **kwargs): + gw._published.append((command, command_id, kwargs)) + + def fake_dashboard_item(entity_id, state=None, attributes=None, app=None): + gw.cache[entity_id] = state + + gw.publish_command = fake_publish_command + gw.dashboard_item = fake_dashboard_item + gw.get_state_wrapper = lambda entity_id=None, **kwargs: gw.cache.get(entity_id) + return gw + + def _run(self, coro): + import asyncio + + return asyncio.run(coro) + + def _clock(self): + """A controllable time.time() for the gateway module.""" + from unittest.mock import patch + + clock = {"now": 1000.0} + patcher = patch("gateway.time.time", lambda: clock["now"]) + return clock, patcher + + def _ack(self, gw, command_id, command, ok=True, error=None, applied=None): + import json + + ack = {"command_id": command_id, "command": command, "ok": ok} + if error: + ack["error"] = error + if applied is not None: + ack["applied"] = applied + gw._process_ack(json.dumps(ack).encode("utf-8")) + + def _logged(self, gw, text): + return any(text in str(call.args[0]) for call in gw.log.call_args_list) + + def test_slow_ack_publishes_once_and_confirms(self): + """The generic loop polls every 2 s; an ack arriving after 10 s means one publish, then a matching read-back.""" + gw = self._make_gateway() + gw.cache[self.SLOT_START] = "00:00:00" + clock, patcher = self._clock() + with patcher: + matched = False + for step in range(10): + clock["now"] = 1000.0 + step * 2 + if clock["now"] >= 1010.0 and not matched: + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + if gw.cache.get(self.SLOT_START) == "00:30:00": + matched = True + break + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + assert matched + assert len(gw._published) == 1 + assert gw._published[0][1] == 1 + assert gw.cache[self.SLOT_END] == "18:00:00" + + def test_refusal_logged_and_not_resent_within_window(self): + """A refused command logs the hub's reason and is not re-sent until the window has passed.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_END, "00:00:00")) + clock["now"] = 1001.0 + self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="would_disable_charge_zero_length_slot") + for step in range(1, 15): + clock["now"] = 1001.0 + step * 2 + self._run(gw.select_event(self.SLOT_END, "00:00:00")) + assert len(gw._published) == 1 + assert self._logged(gw, "Warn: GatewayMQTT: set_charge_slot for CE123456789 refused by hub: would_disable_charge_zero_length_slot") + # The refusal does not update the cached value + assert self.SLOT_END not in gw.cache + # After the window a new attempt goes out + clock["now"] = 1031.0 + self._run(gw.select_event(self.SLOT_END, "00:00:00")) + assert len(gw._published) == 2 + + def test_no_ack_resends_once_after_window_then_falls_back(self): + """No ack: suppressed for the window, re-sent once, then acks stop being relied on.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + for step in range(1, 15): + clock["now"] = 1000.0 + step * 2 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 1 + clock["now"] = 1030.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert [p[1] for p in gw._published] == [1, 2] + clock["now"] = 1040.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 2 + # The re-send is unanswered too: fall back to today's behaviour + clock["now"] = 1060.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 3 + assert gw._acks_seen is False + clock["now"] = 1062.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 4 + + def test_new_value_while_in_flight_publishes_immediately(self): + """A different value is not held back by the pending one, and the stale ack is ignored.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + clock["now"] = 1001.0 + self._run(gw.select_event(self.SLOT_START, "01:00:00")) + assert [p[1] for p in gw._published] == [1, 2] + # The superseded command's ack must not overwrite the cache + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + assert self.SLOT_START not in gw.cache + self._ack(gw, "PBAT2", "set_charge_slot", applied={"start": 100, "end": 1800, "slot": 0}) + assert gw.cache[self.SLOT_START] == "01:00:00" + + def test_ok_ack_updates_switch_and_suppresses_duplicates(self): + """An ok ack updates an enable switch from `applied` and later identical calls are not re-sent.""" + gw = self._make_gateway() + gw.cache[self.CHARGE_ENABLED] = "off" + clock, patcher = self._clock() + with patcher: + self._run(gw.switch_event(self.CHARGE_ENABLED, "turn_on")) + self._ack(gw, "PBAT1", "set_charge_enable", applied={"enable": True, "slot": 0}) + assert gw.cache[self.CHARGE_ENABLED] == "on" + clock["now"] = 1002.0 + self._run(gw.switch_event(self.CHARGE_ENABLED, "turn_on")) + assert len(gw._published) == 1 + + def test_staged_or_recorded_slot_ack_does_not_update_cache(self): + """EMS staged/recorded endpoints were not written to the inverter, so the cache is left alone.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"staged": True, "start": 30, "slot": 0}) + self._run(gw.select_event(self.SLOT_END, "05:00:00")) + self._ack(gw, "PBAT2", "set_charge_slot", applied={"start": 30, "end": 500, "slot": 0, "recorded": True}) + assert gw.cache == {} + + def test_number_ack_does_not_update_cache(self): + """Numbers wait for telemetry; only slots and enables are applied from the ack.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + assert gw.cache == {} + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + + def test_replay_error_resends_with_new_id(self): + """A 'replay' rejection means the value was never considered: the next call publishes under a new id.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="replay") + clock["now"] = 1002.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert [p[1] for p in gw._published] == [1, 2] + assert not self._logged(gw, "refused by hub") + + def test_without_ack_subscription_behaviour_is_unchanged(self): + """No ack subscription: every call publishes, with the same arguments as before.""" + gw = self._make_gateway(subscribed=False, acks_seen=False) + clock, patcher = self._clock() + with patcher: + for _ in range(3): + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert gw._published == [("set_charge_rate", None, {"power_w": 2000, "serial": "CE123456789"})] * 3 + assert gw._pending_commands == {} + + def test_no_suppression_until_an_ack_has_been_seen(self): + """Subscribed but no ack matched yet (e.g. firmware without acks): publish every call, as before.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + clock["now"] = 1002.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 2 + # The first matched ack switches suppression on + self._ack(gw, "PBAT2", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + assert gw._acks_seen is True + clock["now"] = 1004.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 2 + + def test_unknown_or_mismatched_acks_ignored(self): + """Acks for ids PredBat is not tracking, or for a different command, change nothing.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT99", "set_charge_rate") + self._ack(gw, "PBAT1", "set_reserve") + gw._process_ack(b"not json") + gw._process_ack(b"[1, 2]") + assert gw._acks_seen is False + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "sent" + + def test_failed_publish_is_not_left_in_flight(self): + """A publish that raises must not suppress the next attempt.""" + gw = self._make_gateway() + + async def failing_publish_command(command, command_id=None, **kwargs): + raise RuntimeError("broker gone") + + gw.publish_command = failing_publish_command + try: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + except RuntimeError: + pass + assert gw._pending_commands == {} + + def test_ack_topic_routed_to_ack_handler(self): + """Messages on ack/ reach the ack handler.""" + import json + from unittest.mock import MagicMock + + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + message = MagicMock() + message.topic = "predbat/devices/pbgw_test/ack/PBAT1" + message.payload = json.dumps({"command_id": "PBAT1", "command": "set_charge_rate", "ok": True}).encode("utf-8") + self._run(gw._handle_message(message)) + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + + def test_denied_ack_subscription_warns_once(self): + """A refused ack subscription is tolerated and logged once, not on every reconnect.""" + from unittest.mock import AsyncMock, MagicMock + + gw = self._make_gateway(subscribed=False, acks_seen=False) + client = MagicMock() + client.subscribe = AsyncMock(return_value=(0x80,)) + assert self._run(gw._subscribe_acks(client)) is False + assert self._run(gw._subscribe_acks(client)) is False + warnings = [c for c in gw.log.call_args_list if "Cannot subscribe to command acks" in str(c.args[0])] + assert len(warnings) == 1 + client.subscribe = AsyncMock(side_effect=Exception("not authorised")) + assert self._run(gw._subscribe_acks(client)) is False + client.subscribe = AsyncMock(return_value=(0,)) + assert self._run(gw._subscribe_acks(client)) is True + + def test_publish_command_uses_preallocated_id(self): + """publish_command sends the id it is given, and allocates the next one when not given one.""" + import json + from gateway import GatewayMQTT + from unittest.mock import MagicMock + + gw = GatewayMQTT.__new__(GatewayMQTT) + gw.log = MagicMock() + gw._mqtt_connected = True + gw._command_id = 7 + gw.topic_command = "predbat/devices/pbgw_test/command" + sent = [] + + async def fake_publish_raw(topic, payload, retain=False): + sent.append(json.loads(payload)) + + gw._publish_raw = fake_publish_raw + self._run(gw.publish_command("set_charge_rate", command_id=42, power_w=100)) + self._run(gw.publish_command("set_charge_rate", power_w=100)) + assert [s["command_id"] for s in sent] == ["PBAT42", "PBAT8"] + + def run_gateway_tests(my_predbat=None): """Run all GatewayMQTT tests. Returns True on failure, False on success.""" from tests.test_gateway_token_refresh import TestIsAuthFailure, TestApplyRefreshResponse, TestMaybeRefreshOnAuthError @@ -4705,6 +5003,7 @@ def run_gateway_tests(my_predbat=None): TestSelectEvent, TestNumberEvent, TestSwitchEvent, + TestCommandAck, TestTokenRefresh, TestPlanHookConversion, TestMQTTIntegration, From e45e78b1a6f5d06b6ccf10e5451bf8fd7f01d465 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 15:54:05 +0100 Subject: [PATCH 02/11] fix(gateway): match acks for any send of a value, keep refusals, seed command ids Review follow-ups: - keep the recent ids of a tracked value, so an ack for an earlier send (before suppression is on) confirms it instead of being ignored; - a refusal after an ok from another unit is still reported and kept, and only the first ok ack updates the cached slot times; - seed the command id counter from the clock so ids do not restart at PBAT1 after a restart (the hub's replay check, and a late ack for the previous process's id); - time the ack window with a monotonic clock. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 67 ++++++++++++++++++++---------- apps/predbat/tests/test_gateway.py | 57 ++++++++++++++++++++++++- 2 files changed, 101 insertions(+), 23 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index c9ff72050..129123531 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -61,6 +61,16 @@ # the generic write_and_poll loop (10 x 2 s), so one write sends the command once. _COMMAND_ACK_WINDOW = 30 +# Ids kept per tracked command, so an ack for an earlier send of the same value still +# counts (matches the number of recent ids the hub remembers for replay protection) +_COMMAND_ACK_IDS_KEPT = 16 + + +def _monotonic(): + """Clock for the ack window: immune to wall-clock corrections, and patchable in tests.""" + return time.monotonic() + + # Total startup wait budget, in 0.5 s ticks, shared by the connection and auto-config waits _STARTUP_WAIT_TICKS = 120 * 2 _STARTUP_WAIT_SECONDS = _STARTUP_WAIT_TICKS * 0.5 @@ -291,12 +301,15 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self._ev_charging_active: bool = False # last commanded state; avoids duplicate start/stop sends self._ev_max_current: dict = {} # charge_point_id → last known max_current_a from telemetry self._suffix_to_serial = {} # maps entity suffix (last 6 chars of serial) -> full serial string - self._command_id = 0 # incrementing counter included in every published command + # Incrementing counter included in every published command, seeded from the clock so + # ids do not restart at PBAT1 after a restart: the hub rejects ids it has recently seen + # as replays, and a late ack for the previous process's PBAT1 must not match ours. + self._command_id = int(time.time() * 1000) # Command acks (predbat/devices//ack/). Control writes are # tracked per entity so an identical re-send can wait for the hub's ack # instead of queueing another Modbus write every poll. See _send_control(). - self._pending_commands = {} # entity_id -> {command_id, command, kwargs, sent_at, state, resent} + self._pending_commands = {} # entity_id -> {command_ids, command, kwargs, sent_at, state, resent, answered} self._ack_subscribed = False # broker granted the ack subscription on this connection self._acks_seen = False # at least one ack matched a tracked command on this connection self._ack_subscribe_warned = False # the denied-subscription warning is logged once @@ -1948,30 +1961,34 @@ async def _send_control(self, entity_id, command, **kwargs): await self.publish_command(command, **kwargs) return - now = time.time() - resent = False + now = _monotonic() with self._command_lock: entry = self._pending_commands.get(entity_id) same = entry is not None and entry["command"] == command and entry["kwargs"] == kwargs if same and self._acks_seen: if now - entry["sent_at"] < _COMMAND_ACK_WINDOW: - self.log(f"Info: GatewayMQTT: {command} for {kwargs.get('serial')} already sent as {entry['command_id']} ({entry['state']}), not re-sending") + self.log(f"Info: GatewayMQTT: {command} for {kwargs.get('serial')} already sent as {entry['command_ids'][-1]} ({entry['state']}), not re-sending") return - if entry["state"] == "sent": - if entry["resent"]: - self._acks_seen = False - self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") - else: - resent = True + if entry["state"] == "sent" and entry["resent"]: + self._acks_seen = False + self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") + if not same or entry["state"] != "sent": + # New value, or a fresh attempt after the previous one was answered + entry = {"command_ids": [], "command": command, "kwargs": dict(kwargs), "state": "sent", "resent": False, "answered": False} + self._pending_commands[entity_id] = entry + elif self._acks_seen: + entry["resent"] = True # the one re-send after an unanswered window command_id = self._command_id = self._command_id + 1 - self._pending_commands[entity_id] = {"command_id": f"PBAT{command_id}", "command": command, "kwargs": dict(kwargs), "sent_at": now, "state": "sent", "resent": resent} + # Earlier ids of the same value stay matchable: an ack for any of them confirms it + entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [f"PBAT{command_id}"] + entry["sent_at"] = now try: await self.publish_command(command, command_id=command_id, **kwargs) except BaseException: # Never sent, so nothing is in flight - do not let it suppress the next attempt with self._command_lock: - if self._pending_commands.get(entity_id, {}).get("command_id") == f"PBAT{command_id}": + if self._pending_commands.get(entity_id) is entry: del self._pending_commands[entity_id] raise @@ -1993,27 +2010,35 @@ def _process_ack(self, data): error = ack.get("error") or "unknown" with self._command_lock: - entity_id, entry = next(((e, r) for e, r in self._pending_commands.items() if r["command_id"] == command_id), (None, None)) - # Acks for other senders, superseded values or already-answered commands are ignored - if entry is None or entry["state"] != "sent" or ack.get("command", entry["command"]) != entry["command"]: + entity_id, entry = next(((e, r) for e, r in self._pending_commands.items() if command_id in r["command_ids"]), (None, None)) + # Acks for other senders or superseded values are ignored + if entry is None or ack.get("command", entry["command"]) != entry["command"]: return + if entry["state"] == "refused" or (ok and entry["state"] == "applied"): + return # already answered (a command reaching several units acks once per unit) self._acks_seen = True + first_ok = False if ok: entry["state"] = "applied" + first_ok = not entry["answered"] elif error == "replay": - # The hub has seen this id before (e.g. after a PredBat restart reset the - # counter), so the value was never considered: let the next attempt go out - # under a fresh id straight away. + if entry["state"] != "sent": + return + # The hub has seen this id before, so the value was never considered: + # let the next attempt go out under a fresh id straight away. del self._pending_commands[entity_id] else: + # A refusal outranks an ok from another unit for the same command entry["state"] = "refused" - entry["sent_at"] = time.time() + entry["sent_at"] = _monotonic() + entry["answered"] = True command = entry["command"] serial = entry["kwargs"].get("serial") if ok: self.log(f"Info: GatewayMQTT: {command} for {serial} acknowledged by hub ({command_id}) applied={ack.get('applied')}") - self._apply_ack(entity_id, command, ack.get("applied")) + if first_ok: + self._apply_ack(entity_id, command, ack.get("applied")) elif error == "replay": self.log(f"Info: GatewayMQTT: {command} for {serial} rejected as a replay of {command_id}, will re-send with a new id") else: diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index fb30ba910..2543ae56f 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -4729,11 +4729,11 @@ def _run(self, coro): return asyncio.run(coro) def _clock(self): - """A controllable time.time() for the gateway module.""" + """A controllable clock for the ack window.""" from unittest.mock import patch clock = {"now": 1000.0} - patcher = patch("gateway.time.time", lambda: clock["now"]) + patcher = patch("gateway._monotonic", lambda: clock["now"]) return clock, patcher def _ack(self, gw, command_id, command, ok=True, error=None, applied=None): @@ -4901,6 +4901,59 @@ def test_no_suppression_until_an_ack_has_been_seen(self): self._run(gw.number_event(self.CHARGE_RATE, 2000)) assert len(gw._published) == 2 + def test_ack_for_an_earlier_send_of_the_same_value_counts(self): + """Before suppression is on, each poll re-sends under a new id; an ack for any of them confirms the value.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + for step in range(3): + clock["now"] = 1000.0 + step * 2 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert [p[1] for p in gw._published] == [1, 2, 3] + # Only the first send's ack arrives + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + assert gw._acks_seen is True + clock["now"] = 1006.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 3 + + def test_refusal_after_ok_from_another_unit_wins(self): + """A command reaching two units acks twice; a refusal after an ok is still reported and kept.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="modbus_write_failed") + assert gw._pending_commands[self.SLOT_START]["state"] == "refused" + assert self._logged(gw, "refused by hub: modbus_write_failed") + # A later ok for the same id changes nothing + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 45, "end": 1800, "slot": 1}) + assert gw._pending_commands[self.SLOT_START]["state"] == "refused" + assert gw.cache[self.SLOT_START] == "00:30:00" + + def test_second_ok_ack_does_not_overwrite_cache(self): + """Only the first ok ack for a command updates the cached slot times.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 900, "slot": 1}) + assert gw.cache[self.SLOT_END] == "18:00:00" + + def test_command_ids_do_not_restart_at_one(self): + """initialize() seeds the id counter from the clock so a restart does not reuse recent ids.""" + from gateway import GatewayMQTT + from unittest.mock import MagicMock + + gw = GatewayMQTT.__new__(GatewayMQTT) + gw.base = MagicMock() + gw.args = {} + gw.initialize(gateway_device_id="pbgw_test", mqtt_host="mqtt.example.com", mqtt_token="tok") + assert gw._command_id > 1_000_000_000_000 + assert gw.topic_ack == "predbat/devices/pbgw_test/ack/+" + def test_unknown_or_mismatched_acks_ignored(self): """Acks for ids PredBat is not tracking, or for a different command, change nothing.""" gw = self._make_gateway(acks_seen=False) From c02d76c57c320a0775c87cb69c3d78c9b279b7dc Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 16:06:14 +0100 Subject: [PATCH 03/11] fix(gateway): track ack outcomes per command id Review follow-ups: - record ok/refused per id: an ok for any send of a value means it was applied, while within one id (several units) a refusal outranks an ok; - a failed publish drops only its own id, so earlier sends stay matchable; - derive the slot entities from the entity's own suffix. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 59 ++++++++++++++++++------------ apps/predbat/tests/test_gateway.py | 55 ++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 23 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 129123531..eee34ec66 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -309,7 +309,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt # Command acks (predbat/devices//ack/). Control writes are # tracked per entity so an identical re-send can wait for the hub's ack # instead of queueing another Modbus write every poll. See _send_control(). - self._pending_commands = {} # entity_id -> {command_ids, command, kwargs, sent_at, state, resent, answered} + self._pending_commands = {} # entity_id -> {command_ids, outcomes, command, kwargs, state, sent_at, resent, cached} self._ack_subscribed = False # broker granted the ack subscription on this connection self._acks_seen = False # at least one ack matched a tracked command on this connection self._ack_subscribe_warned = False # the denied-subscription warning is logged once @@ -1974,27 +1974,38 @@ async def _send_control(self, entity_id, command, **kwargs): self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") if not same or entry["state"] != "sent": # New value, or a fresh attempt after the previous one was answered - entry = {"command_ids": [], "command": command, "kwargs": dict(kwargs), "state": "sent", "resent": False, "answered": False} + entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": False} self._pending_commands[entity_id] = entry - elif self._acks_seen: + previous = (entry["sent_at"], entry["resent"]) + if self._acks_seen and entry["command_ids"]: entry["resent"] = True # the one re-send after an unanswered window command_id = self._command_id = self._command_id + 1 + command_ref = f"PBAT{command_id}" # Earlier ids of the same value stay matchable: an ack for any of them confirms it - entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [f"PBAT{command_id}"] + entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref] entry["sent_at"] = now try: await self.publish_command(command, command_id=command_id, **kwargs) except BaseException: - # Never sent, so nothing is in flight - do not let it suppress the next attempt + # This id was never sent, so it must not hold back the next attempt; ids that + # were sent earlier stay matchable with self._command_lock: - if self._pending_commands.get(entity_id) is entry: - del self._pending_commands[entity_id] + if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"]: + entry["command_ids"].remove(command_ref) + if entry["command_ids"]: + entry["sent_at"], entry["resent"] = previous + else: + del self._pending_commands[entity_id] raise def _process_ack(self, data): """Handle a command ack from the hub. + Every id sent for a tracked value carries the same payload, so an ok for any of + them means the value was applied. A command reaching several units acks once per + unit under the same id: there a refusal outranks an ok. + Args: data: Raw JSON payload from predbat/devices//ack/. """ @@ -2008,38 +2019,40 @@ def _process_ack(self, data): command_id = str(ack.get("command_id", "")) ok = ack.get("ok") is True error = ack.get("error") or "unknown" + replay = not ok and error == "replay" + apply_cache = False with self._command_lock: entity_id, entry = next(((e, r) for e, r in self._pending_commands.items() if command_id in r["command_ids"]), (None, None)) # Acks for other senders or superseded values are ignored if entry is None or ack.get("command", entry["command"]) != entry["command"]: return - if entry["state"] == "refused" or (ok and entry["state"] == "applied"): - return # already answered (a command reaching several units acks once per unit) self._acks_seen = True - first_ok = False - if ok: - entry["state"] = "applied" - first_ok = not entry["answered"] - elif error == "replay": + if replay: if entry["state"] != "sent": return # The hub has seen this id before, so the value was never considered: # let the next attempt go out under a fresh id straight away. del self._pending_commands[entity_id] else: - # A refusal outranks an ok from another unit for the same command - entry["state"] = "refused" - entry["sent_at"] = _monotonic() - entry["answered"] = True + outcome = entry["outcomes"].get(command_id) + if outcome == "refused" or (ok and outcome == "ok"): + return # this id is already answered + entry["outcomes"][command_id] = "ok" if ok else "refused" + was_refused = entry["state"] == "refused" + entry["state"] = "applied" if "ok" in entry["outcomes"].values() else "refused" + if entry["state"] == "refused" and not was_refused: + entry["sent_at"] = _monotonic() # the window runs from the refusal + if ok and entry["state"] == "applied" and not entry["cached"]: + entry["cached"] = apply_cache = True command = entry["command"] serial = entry["kwargs"].get("serial") if ok: self.log(f"Info: GatewayMQTT: {command} for {serial} acknowledged by hub ({command_id}) applied={ack.get('applied')}") - if first_ok: + if apply_cache: self._apply_ack(entity_id, command, ack.get("applied")) - elif error == "replay": + elif replay: self.log(f"Info: GatewayMQTT: {command} for {serial} rejected as a replay of {command_id}, will re-send with a new id") else: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") @@ -2060,10 +2073,10 @@ def _apply_ack(self, entity_id, command, applied): if not all(isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= 2400 for v in (start, end)): return name = "discharge_slot1" if command == "set_discharge_slot" else "charge_slot1" - marker = f"_{name}_" - if marker not in entity_id: + suffix = next((f"_{name}_{field}" for field in ("start", "end") if entity_id.endswith(f"_{name}_{field}")), None) + if suffix is None: return - base = entity_id[: entity_id.index(marker)] + base = entity_id[: -len(suffix)] for field, hhmm in (("start", start), ("end", end)): self.dashboard_item(f"{base}_{name}_{field}", self._hhmm_to_time_str(hhmm), attributes=GATEWAY_ATTRIBUTE_TABLE.get(f"{name}_{field}", {}), app="gateway") elif command in ("set_charge_enable", "set_discharge_enable"): diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 2543ae56f..fe6c87d6a 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -4932,6 +4932,61 @@ def test_refusal_after_ok_from_another_unit_wins(self): assert gw._pending_commands[self.SLOT_START]["state"] == "refused" assert gw.cache[self.SLOT_START] == "00:30:00" + def test_ok_for_a_later_send_beats_refusal_of_an_earlier_one(self): + """Distinct sends of the same value: an ok for any of them means the value was applied.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + clock["now"] = 1002.0 + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="modbus_write_failed") + assert gw._pending_commands[self.SLOT_START]["state"] == "refused" + self._ack(gw, "PBAT2", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + assert gw._pending_commands[self.SLOT_START]["state"] == "applied" + assert gw.cache[self.SLOT_START] == "00:30:00" + # A delayed refusal of an earlier send does not undo the later success + clock["now"] = 1004.0 + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + assert gw._pending_commands[self.SLOT_START]["state"] == "applied" + assert len(gw._published) == 2 + + def test_failed_retry_keeps_earlier_ids_matchable(self): + """A publish that raises drops only its own id; an ack for an earlier send still counts.""" + gw = self._make_gateway(acks_seen=False) + real_publish = gw.publish_command + + async def failing_publish_command(command, command_id=None, **kwargs): + raise RuntimeError("broker gone") + + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + clock["now"] = 1002.0 + gw.publish_command = failing_publish_command + try: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + except RuntimeError: + pass + gw.publish_command = real_publish + assert gw._pending_commands[self.CHARGE_RATE]["command_ids"] == ["PBAT1"] + assert gw._pending_commands[self.CHARGE_RATE]["sent_at"] == 1000.0 + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + assert gw._acks_seen is True + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + + def test_slot_cache_uses_the_entity_suffix_not_the_first_match(self): + """A prefix that itself contains 'charge_slot1_' does not redirect the cache update.""" + gw = self._make_gateway() + gw.prefix = "house_charge_slot1_x" + entity = "select.house_charge_slot1_x_gateway_456789_charge_slot1_start" + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(entity, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + assert gw.cache == {entity: "00:30:00", "select.house_charge_slot1_x_gateway_456789_charge_slot1_end": "18:00:00"} + def test_second_ok_ack_does_not_overwrite_cache(self): """Only the first ok ack for a command updates the cached slot times.""" gw = self._make_gateway() From b0fae8c4d0b353b736aa628b17be4d2bfcea6ef3 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 16:15:29 +0100 Subject: [PATCH 04/11] fix(gateway): keep the ack cache update honest across multi-unit acks Review follow-ups: - only mark the cache as updated when an ack actually carried usable values, so a staged/recorded first ok does not block a later plain one; - a refusal for the id whose ok updated the cache restores the last telemetry instead of leaving the optimistic value in place; - every refusal restarts the cooldown window. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 48 +++++++++++++++++++----------- apps/predbat/tests/test_gateway.py | 47 +++++++++++++++++++++++++++-- 2 files changed, 75 insertions(+), 20 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index eee34ec66..ffd4747d1 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -309,7 +309,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt # Command acks (predbat/devices//ack/). Control writes are # tracked per entity so an identical re-send can wait for the hub's ack # instead of queueing another Modbus write every poll. See _send_control(). - self._pending_commands = {} # entity_id -> {command_ids, outcomes, command, kwargs, state, sent_at, resent, cached} + self._pending_commands = {} # entity_id -> {command_ids, outcomes, command, kwargs, state, sent_at, resent, cached (id that updated the cache)} self._ack_subscribed = False # broker granted the ack subscription on this connection self._acks_seen = False # at least one ack matched a tracked command on this connection self._ack_subscribe_warned = False # the denied-subscription warning is logged once @@ -1974,7 +1974,7 @@ async def _send_control(self, entity_id, command, **kwargs): self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") if not same or entry["state"] != "sent": # New value, or a fresh attempt after the previous one was answered - entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": False} + entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None} self._pending_commands[entity_id] = entry previous = (entry["sent_at"], entry["resent"]) if self._acks_seen and entry["command_ids"]: @@ -2020,7 +2020,7 @@ def _process_ack(self, data): ok = ack.get("ok") is True error = ack.get("error") or "unknown" replay = not ok and error == "replay" - apply_cache = False + apply_cache = restore_status = False with self._command_lock: entity_id, entry = next(((e, r) for e, r in self._pending_commands.items() if command_id in r["command_ids"]), (None, None)) @@ -2035,27 +2035,33 @@ def _process_ack(self, data): # let the next attempt go out under a fresh id straight away. del self._pending_commands[entity_id] else: - outcome = entry["outcomes"].get(command_id) - if outcome == "refused" or (ok and outcome == "ok"): - return # this id is already answered + if entry["outcomes"].get(command_id) == "refused": + return # a refusal for this id outranks any other unit's ok entry["outcomes"][command_id] = "ok" if ok else "refused" - was_refused = entry["state"] == "refused" entry["state"] = "applied" if "ok" in entry["outcomes"].values() else "refused" - if entry["state"] == "refused" and not was_refused: - entry["sent_at"] = _monotonic() # the window runs from the refusal - if ok and entry["state"] == "applied" and not entry["cached"]: - entry["cached"] = apply_cache = True + if not ok: + entry["sent_at"] = _monotonic() # the window runs from the latest refusal + # Another unit refused the id whose ok updated the cache: that update overstated + # what was applied, so put back the last telemetry until the next status arrives + restore_status = entry["cached"] == command_id + if restore_status: + entry["cached"] = None + apply_cache = ok and entry["cached"] is None command = entry["command"] serial = entry["kwargs"].get("serial") if ok: self.log(f"Info: GatewayMQTT: {command} for {serial} acknowledged by hub ({command_id}) applied={ack.get('applied')}") - if apply_cache: - self._apply_ack(entity_id, command, ack.get("applied")) + if apply_cache and self._apply_ack(entity_id, command, ack.get("applied")): + with self._command_lock: + if self._pending_commands.get(entity_id) is entry and entry["outcomes"].get(command_id) == "ok": + entry["cached"] = command_id elif replay: self.log(f"Info: GatewayMQTT: {command} for {serial} rejected as a replay of {command_id}, will re-send with a new id") else: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") + if restore_status and self._last_status is not None: + self._inject_entities(self._last_status) def _apply_ack(self, entity_id, command, applied): """Update the cached entity from an ok ack so the write's read-back matches without waiting for telemetry. @@ -2063,28 +2069,34 @@ def _apply_ack(self, entity_id, command, applied): Only the values that map cleanly are applied: slot start/end actually written and charge/discharge enables. Staged or recorded EMS slot endpoints are not written to the inverter, and numbers wait for telemetry. + + Returns: + bool: True if a cached entity was updated. """ if not isinstance(applied, dict): - return + return False if command in ("set_charge_slot", "set_discharge_slot"): if applied.get("staged") or applied.get("recorded"): - return + return False start, end = applied.get("start"), applied.get("end") if not all(isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= 2400 for v in (start, end)): - return + return False name = "discharge_slot1" if command == "set_discharge_slot" else "charge_slot1" suffix = next((f"_{name}_{field}" for field in ("start", "end") if entity_id.endswith(f"_{name}_{field}")), None) if suffix is None: - return + return False base = entity_id[: -len(suffix)] for field, hhmm in (("start", start), ("end", end)): self.dashboard_item(f"{base}_{name}_{field}", self._hhmm_to_time_str(hhmm), attributes=GATEWAY_ATTRIBUTE_TABLE.get(f"{name}_{field}", {}), app="gateway") + return True elif command in ("set_charge_enable", "set_discharge_enable"): enable = applied.get("enable") if not isinstance(enable, bool): - return + return False name = "discharge_enabled" if command == "set_discharge_enable" else "charge_enabled" self.dashboard_item(entity_id, "on" if enable else "off", attributes=GATEWAY_ATTRIBUTE_TABLE.get(name, {}), app="gateway") + return True + return False @staticmethod def _hhmm_to_time_str(hhmm): diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index fe6c87d6a..f863c38c7 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -4711,6 +4711,7 @@ def _make_gateway(self, subscribed=True, acks_seen=True): gw._ack_topic_prefix = "predbat/devices/pbgw_test/ack/" gw._published = [] # (command, command_id, kwargs) gw.cache = {} # entity_id -> cached state, as the generic write loop would read it back + gw._last_status = None async def fake_publish_command(command, command_id=None, **kwargs): gw._published.append((command, command_id, kwargs)) @@ -4918,20 +4919,62 @@ def test_ack_for_an_earlier_send_of_the_same_value_counts(self): assert len(gw._published) == 3 def test_refusal_after_ok_from_another_unit_wins(self): - """A command reaching two units acks twice; a refusal after an ok is still reported and kept.""" + """A command reaching two units acks twice; a refusal after an ok is reported, kept, and undoes the cache update.""" gw = self._make_gateway() + gw.cache[self.SLOT_START] = "00:00:00" + gw._last_status = object() + + def fake_inject_entities(status): + assert status is gw._last_status + gw.cache[self.SLOT_START] = "00:00:00" # last telemetry: the old value + gw.cache[self.SLOT_END] = "00:00:00" + + gw._inject_entities = fake_inject_entities clock, patcher = self._clock() with patcher: self._run(gw.select_event(self.SLOT_START, "00:30:00")) self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + assert gw.cache[self.SLOT_START] == "00:30:00" self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="modbus_write_failed") assert gw._pending_commands[self.SLOT_START]["state"] == "refused" assert self._logged(gw, "refused by hub: modbus_write_failed") + assert gw.cache[self.SLOT_START] == "00:00:00" # A later ok for the same id changes nothing - self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 45, "end": 1800, "slot": 1}) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 1}) assert gw._pending_commands[self.SLOT_START]["state"] == "refused" + assert gw.cache[self.SLOT_START] == "00:00:00" + + def test_unusable_first_ok_does_not_block_a_later_cache_update(self): + """A staged ok from one unit, then a plain ok from another: the plain one still updates the cache.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"staged": True, "start": 30, "slot": 0}) + assert gw.cache == {} + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 1}) assert gw.cache[self.SLOT_START] == "00:30:00" + def test_refusal_restarts_the_window_even_when_another_send_succeeded(self): + """Any refusal restarts the cooldown, so an identical call is not re-sent within 30 s of it.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + gw._acks_seen = False # force a second send of the same value + clock["now"] = 1002.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT2", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + clock["now"] = 1025.0 + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="modbus_write_failed") + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + clock["now"] = 1040.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 2 + clock["now"] = 1056.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 3 + def test_ok_for_a_later_send_beats_refusal_of_an_earlier_one(self): """Distinct sends of the same value: an ok for any of them means the value was applied.""" gw = self._make_gateway(acks_seen=False) From 2e9996fe0c54a2287d6c019822ef3f301c445d8d Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 16:21:46 +0100 Subject: [PATCH 05/11] fix(gateway): keep earlier attempts' acks matchable and handle every refusal Review follow-ups: - a fresh attempt for an already-answered value keeps its earlier ids, so a late ack from another unit is still logged and acted on; a failed publish only removes its own id; - every refusal for an id is logged and restarts the window, not just the first. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 20 ++++++++++------- apps/predbat/tests/test_gateway.py | 36 ++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 8 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index ffd4747d1..ae45db5e4 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1972,12 +1972,15 @@ async def _send_control(self, entity_id, command, **kwargs): if entry["state"] == "sent" and entry["resent"]: self._acks_seen = False self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") - if not same or entry["state"] != "sent": - # New value, or a fresh attempt after the previous one was answered + if not same: entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None} self._pending_commands[entity_id] = entry - previous = (entry["sent_at"], entry["resent"]) - if self._acks_seen and entry["command_ids"]: + previous = {"outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} + if entry["state"] != "sent": + # A fresh attempt after the previous one was answered. Earlier ids stay + # matchable, so a late ack from another unit is still logged and acted on. + entry.update(outcomes={}, state="sent", resent=False, cached=None) + elif self._acks_seen and entry["command_ids"]: entry["resent"] = True # the one re-send after an unanswered window command_id = self._command_id = self._command_id + 1 command_ref = f"PBAT{command_id}" @@ -1993,10 +1996,11 @@ async def _send_control(self, entity_id, command, **kwargs): with self._command_lock: if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"]: entry["command_ids"].remove(command_ref) - if entry["command_ids"]: - entry["sent_at"], entry["resent"] = previous - else: + if not entry["command_ids"]: del self._pending_commands[entity_id] + elif entry["state"] == "sent" and not entry["outcomes"]: + # No ack arrived meanwhile: back to how the earlier attempt left it + entry.update({key: previous[key] for key in ("outcomes", "state", "sent_at", "resent", "cached")}) raise def _process_ack(self, data): @@ -2035,7 +2039,7 @@ def _process_ack(self, data): # let the next attempt go out under a fresh id straight away. del self._pending_commands[entity_id] else: - if entry["outcomes"].get(command_id) == "refused": + if ok and entry["outcomes"].get(command_id) == "refused": return # a refusal for this id outranks any other unit's ok entry["outcomes"][command_id] = "ok" if ok else "refused" entry["state"] = "applied" if "ok" in entry["outcomes"].values() else "refused" diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index f863c38c7..fe6cb6dd0 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5030,6 +5030,42 @@ def test_slot_cache_uses_the_entity_suffix_not_the_first_match(self): self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) assert gw.cache == {entity: "00:30:00", "select.house_charge_slot1_x_gateway_456789_charge_slot1_end": "18:00:00"} + def test_late_refusal_for_an_earlier_attempt_is_still_handled(self): + """After an answered attempt, a fresh attempt keeps earlier ids: a late refusal is logged and restarts the window.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + clock["now"] = 1031.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert gw._pending_commands[self.CHARGE_RATE]["command_ids"] == ["PBAT1", "PBAT2"] + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "sent" + clock["now"] = 1032.0 + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="not_managed") + assert self._logged(gw, "refused by hub: not_managed") + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "refused" + assert gw._pending_commands[self.CHARGE_RATE]["sent_at"] == 1032.0 + + def test_each_refusal_for_the_same_id_is_logged_and_restarts_the_window(self): + """Two units refusing one command: both reasons are logged and the window runs from the second.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + clock["now"] = 1001.0 + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="not_managed") + clock["now"] = 1025.0 + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="modbus_write_failed") + assert self._logged(gw, "refused by hub: not_managed") + assert self._logged(gw, "refused by hub: modbus_write_failed") + clock["now"] = 1031.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 1 + clock["now"] = 1055.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 2 + def test_second_ok_ack_does_not_overwrite_cache(self): """Only the first ok ack for a command updates the cached slot times.""" gw = self._make_gateway() From c78efed13f27dc07079f177478383b8984242ed1 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 22:09:20 +0100 Subject: [PATCH 06/11] fix(gateway): keep cache provenance across attempts and confirmed ids on publish errors Review follow-ups: - remember every id whose ok updated the cache, across fresh attempts, so a late refusal from another unit still restores telemetry; - a publish error no longer removes an id whose ack already arrived while the publish awaited the broker's confirmation; - a replay rejection is logged as "refused by hub: replay" (and still re-sent under a new id). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 18 +++++++---- apps/predbat/tests/test_gateway.py | 49 +++++++++++++++++++++++++++++- 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index ae45db5e4..e49430803 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1973,7 +1973,7 @@ async def _send_control(self, entity_id, command, **kwargs): self._acks_seen = False self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") if not same: - entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None} + entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": []} self._pending_commands[entity_id] = entry previous = {"outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} if entry["state"] != "sent": @@ -1994,7 +1994,9 @@ async def _send_control(self, entity_id, command, **kwargs): # This id was never sent, so it must not hold back the next attempt; ids that # were sent earlier stay matchable with self._command_lock: - if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"]: + # An ack for this id may already have arrived while the publish was awaiting its + # broker confirmation: then the command did reach the hub, so keep it + if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"] and command_ref not in entry["outcomes"]: entry["command_ids"].remove(command_ref) if not entry["command_ids"]: del self._pending_commands[entity_id] @@ -2045,10 +2047,13 @@ def _process_ack(self, data): entry["state"] = "applied" if "ok" in entry["outcomes"].values() else "refused" if not ok: entry["sent_at"] = _monotonic() # the window runs from the latest refusal - # Another unit refused the id whose ok updated the cache: that update overstated - # what was applied, so put back the last telemetry until the next status arrives - restore_status = entry["cached"] == command_id + # Another unit refused an id whose ok updated the cache (in this attempt or an + # earlier one): that update overstated what was applied, so put back the last + # telemetry until the next status arrives + restore_status = command_id in entry["cached_ids"] if restore_status: + entry["cached_ids"].remove(command_id) + if entry["cached"] == command_id: entry["cached"] = None apply_cache = ok and entry["cached"] is None command = entry["command"] @@ -2060,8 +2065,9 @@ def _process_ack(self, data): with self._command_lock: if self._pending_commands.get(entity_id) is entry and entry["outcomes"].get(command_id) == "ok": entry["cached"] = command_id + entry["cached_ids"] = entry["cached_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_id] elif replay: - self.log(f"Info: GatewayMQTT: {command} for {serial} rejected as a replay of {command_id}, will re-send with a new id") + self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: replay ({command_id} already used), re-sending with a new id") else: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") if restore_status and self._last_status is not None: diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index fe6cb6dd0..43e8d2b91 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -4874,7 +4874,8 @@ def test_replay_error_resends_with_new_id(self): clock["now"] = 1002.0 self._run(gw.number_event(self.CHARGE_RATE, 2000)) assert [p[1] for p in gw._published] == [1, 2] - assert not self._logged(gw, "refused by hub") + assert self._logged(gw, "Warn: GatewayMQTT: set_charge_rate for CE123456789 refused by hub: replay") + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "sent" def test_without_ack_subscription_behaviour_is_unchanged(self): """No ack subscription: every call publishes, with the same arguments as before.""" @@ -5066,6 +5067,52 @@ def test_each_refusal_for_the_same_id_is_logged_and_restarts_the_window(self): self._run(gw.number_event(self.CHARGE_RATE, 2000)) assert len(gw._published) == 2 + def test_late_refusal_undoes_a_cache_update_from_an_earlier_attempt(self): + """PBAT1 ok updates the slot cache, PBAT2 is a fresh attempt, then another unit refuses PBAT1: telemetry is restored.""" + gw = self._make_gateway() + gw._last_status = object() + restored = [] + + def fake_inject_entities(status): + restored.append(status) + gw.cache[self.SLOT_START] = "00:00:00" + + gw._inject_entities = fake_inject_entities + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + assert gw.cache[self.SLOT_START] == "00:30:00" + clock["now"] = 1031.0 + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + assert [p[1] for p in gw._published] == [1, 2] + self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="modbus_write_failed") + assert restored == [gw._last_status] + assert gw.cache[self.SLOT_START] == "00:00:00" + assert self._logged(gw, "refused by hub: modbus_write_failed") + + def test_publish_failure_after_its_ack_arrived_keeps_the_entry(self): + """The hub acknowledged the command while the publish was still awaiting the broker; the later publish error must not undo that.""" + gw = self._make_gateway() + + async def ack_then_fail(command, command_id=None, **kwargs): + gw._published.append((command, command_id, kwargs)) + self._ack(gw, f"PBAT{command_id}", command, applied={"power_w": 2000, "slot": 0}) + raise RuntimeError("connection lost before the broker confirmed") + + gw.publish_command = ack_then_fail + clock, patcher = self._clock() + with patcher: + try: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + except RuntimeError: + pass + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + assert gw._pending_commands[self.CHARGE_RATE]["command_ids"] == ["PBAT1"] + clock["now"] = 1002.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 1 + def test_second_ok_ack_does_not_overwrite_cache(self): """Only the first ok ack for a command updates the cached slot times.""" gw = self._make_gateway() From a16364b0920fdfdaacf7ac64e39c9798b6894355 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 22:18:12 +0100 Subject: [PATCH 07/11] fix(gateway): keep per-id outcomes across attempts; log late replays Review follow-ups: - a fresh attempt keeps earlier ids' outcomes (pruned with the ids), so a refusal from an earlier attempt still outranks that id's late ok; - a replay arriving after the command was answered is treated as another unit's refusal: logged and restarts the window. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 22 +++++++------- apps/predbat/tests/test_gateway.py | 46 ++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 10 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index e49430803..dd4006fc5 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1977,15 +1977,17 @@ async def _send_control(self, entity_id, command, **kwargs): self._pending_commands[entity_id] = entry previous = {"outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} if entry["state"] != "sent": - # A fresh attempt after the previous one was answered. Earlier ids stay - # matchable, so a late ack from another unit is still logged and acted on. - entry.update(outcomes={}, state="sent", resent=False, cached=None) + # A fresh attempt after the previous one was answered. Earlier ids and their + # outcomes stay, so a late ack from another unit is still logged and acted on and + # a refusal still outranks that id's later ok. + entry.update(state="sent", resent=False, cached=None) elif self._acks_seen and entry["command_ids"]: entry["resent"] = True # the one re-send after an unanswered window command_id = self._command_id = self._command_id + 1 command_ref = f"PBAT{command_id}" # Earlier ids of the same value stay matchable: an ack for any of them confirms it entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref] + entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]} entry["sent_at"] = now try: @@ -2000,7 +2002,7 @@ async def _send_control(self, entity_id, command, **kwargs): entry["command_ids"].remove(command_ref) if not entry["command_ids"]: del self._pending_commands[entity_id] - elif entry["state"] == "sent" and not entry["outcomes"]: + elif entry["state"] == "sent" and entry["outcomes"] == previous["outcomes"]: # No ack arrived meanwhile: back to how the earlier attempt left it entry.update({key: previous[key] for key in ("outcomes", "state", "sent_at", "resent", "cached")}) raise @@ -2034,11 +2036,11 @@ def _process_ack(self, data): if entry is None or ack.get("command", entry["command"]) != entry["command"]: return self._acks_seen = True - if replay: - if entry["state"] != "sent": - return - # The hub has seen this id before, so the value was never considered: - # let the next attempt go out under a fresh id straight away. + # While still waiting, a replay means the hub has seen this id before and never + # considered the value: let the next attempt go out under a fresh id straight away. + # Once the command has an answer, a replay is just another unit's refusal. + drop_replay = replay and entry["state"] == "sent" + if drop_replay: del self._pending_commands[entity_id] else: if ok and entry["outcomes"].get(command_id) == "refused": @@ -2066,7 +2068,7 @@ def _process_ack(self, data): if self._pending_commands.get(entity_id) is entry and entry["outcomes"].get(command_id) == "ok": entry["cached"] = command_id entry["cached_ids"] = entry["cached_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_id] - elif replay: + elif drop_replay: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: replay ({command_id} already used), re-sending with a new id") else: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 43e8d2b91..c6c1b7227 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5113,6 +5113,52 @@ async def ack_then_fail(command, command_id=None, **kwargs): self._run(gw.number_event(self.CHARGE_RATE, 2000)) assert len(gw._published) == 1 + def test_refusal_from_an_earlier_attempt_still_outranks_its_late_ok(self): + """PBAT1 refused, fresh attempt PBAT2, then another unit's late ok for PBAT1: it must not count as applied.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="modbus_write_failed") + clock["now"] = 1031.0 + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + assert [p[1] for p in gw._published] == [1, 2] + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 1}) + assert gw._pending_commands[self.SLOT_START]["state"] == "sent" + assert gw.cache == {} + # PBAT2's own ok still confirms the value + self._ack(gw, "PBAT2", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + assert gw._pending_commands[self.SLOT_START]["state"] == "applied" + assert gw.cache[self.SLOT_START] == "00:30:00" + + def test_replay_after_an_answer_is_logged_as_a_refusal(self): + """A replay from another unit after the command was answered is logged and restarts the window.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + clock["now"] = 1020.0 + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="replay") + assert self._logged(gw, "Warn: GatewayMQTT: set_charge_rate for CE123456789 refused by hub: replay") + assert gw._pending_commands[self.CHARGE_RATE]["sent_at"] == 1020.0 + clock["now"] = 1040.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 1 + + def test_outcomes_do_not_grow_past_the_kept_ids(self): + """Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + for attempt in range(40): + clock["now"] = 1000.0 + attempt * 31 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, f"PBAT{attempt + 1}", "set_charge_rate", ok=False, error="not_managed") + entry = gw._pending_commands[self.CHARGE_RATE] + assert len(entry["command_ids"]) == 16 + assert set(entry["outcomes"]) <= set(entry["command_ids"]) + def test_second_ok_ack_does_not_overwrite_cache(self): """Only the first ok ack for a command updates the cached slot times.""" gw = self._make_gateway() From 195e738fb5c2ae425579f7b1a508743f456bdb5f Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 22:26:10 +0100 Subject: [PATCH 08/11] fix(gateway): only a replay for the id awaiting its first answer drops the entry A late replay for an earlier id (e.g. from another unit after a fresh attempt) is handled as a refusal instead of discarding the current attempt. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 8 ++++---- apps/predbat/tests/test_gateway.py | 23 +++++++++++++++++++++++ 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index dd4006fc5..8c2dd2140 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -2036,10 +2036,10 @@ def _process_ack(self, data): if entry is None or ack.get("command", entry["command"]) != entry["command"]: return self._acks_seen = True - # While still waiting, a replay means the hub has seen this id before and never - # considered the value: let the next attempt go out under a fresh id straight away. - # Once the command has an answer, a replay is just another unit's refusal. - drop_replay = replay and entry["state"] == "sent" + # A replay for the id still awaiting its first answer means the hub has seen that id + # before and never considered the value: let the next attempt go out under a fresh id + # straight away. For an older or already-answered id it is just another refusal. + drop_replay = replay and entry["state"] == "sent" and entry["command_ids"][-1] == command_id and command_id not in entry["outcomes"] if drop_replay: del self._pending_commands[entity_id] else: diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index c6c1b7227..31bcab487 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5146,6 +5146,29 @@ def test_replay_after_an_answer_is_logged_as_a_refusal(self): self._run(gw.number_event(self.CHARGE_RATE, 2000)) assert len(gw._published) == 1 + def test_late_replay_for_an_earlier_id_does_not_drop_the_fresh_attempt(self): + """PBAT1 ok, fresh attempt PBAT2, then a late replay for PBAT1: PBAT2 stays tracked and the window restarts.""" + gw = self._make_gateway() + gw._last_status = object() + restored = [] + gw._inject_entities = lambda status: restored.append(status) + clock, patcher = self._clock() + with patcher: + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + self._ack(gw, "PBAT1", "set_charge_slot", applied={"start": 30, "end": 1800, "slot": 0}) + clock["now"] = 1031.0 + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + clock["now"] = 1032.0 + self._ack(gw, "PBAT1", "set_charge_slot", ok=False, error="replay") + entry = gw._pending_commands[self.SLOT_START] + assert entry["command_ids"] == ["PBAT1", "PBAT2"] + assert entry["sent_at"] == 1032.0 + assert restored == [gw._last_status] + assert self._logged(gw, "refused by hub: replay") + clock["now"] = 1034.0 + self._run(gw.select_event(self.SLOT_START, "00:30:00")) + assert len(gw._published) == 2 + def test_outcomes_do_not_grow_past_the_kept_ids(self): """Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry.""" gw = self._make_gateway() From 504e512aa8db27abb6b818bfcf7238a8cebd2344 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 22:34:55 +0100 Subject: [PATCH 09/11] fix(gateway): scope replay handling to the current attempt; full rollback on publish errors Review follow-ups: - a replay for the newest unanswered id drops the entry unless another send in the same attempt already applied the value, in which case only that id is forgotten (acks from earlier attempts do not count); - a failed publish restores the id history as it was, including an id pruned to make room. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 42 +++++++++++++------- apps/predbat/tests/test_gateway.py | 63 ++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 15 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 8c2dd2140..4df184452 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1973,14 +1973,14 @@ async def _send_control(self, entity_id, command, **kwargs): self._acks_seen = False self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") if not same: - entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": []} + entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": []} self._pending_commands[entity_id] = entry - previous = {"outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} + previous = {"command_ids": list(entry["command_ids"]), "attempt_ids": list(entry["attempt_ids"]), "outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} if entry["state"] != "sent": # A fresh attempt after the previous one was answered. Earlier ids and their # outcomes stay, so a late ack from another unit is still logged and acted on and # a refusal still outranks that id's later ok. - entry.update(state="sent", resent=False, cached=None) + entry.update(state="sent", resent=False, cached=None, attempt_ids=[]) elif self._acks_seen and entry["command_ids"]: entry["resent"] = True # the one re-send after an unanswered window command_id = self._command_id = self._command_id + 1 @@ -1988,7 +1988,9 @@ async def _send_control(self, entity_id, command, **kwargs): # Earlier ids of the same value stay matchable: an ack for any of them confirms it entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref] entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]} + entry["attempt_ids"] = entry["attempt_ids"] + [command_ref] entry["sent_at"] = now + outcomes_sent = dict(entry["outcomes"]) try: await self.publish_command(command, command_id=command_id, **kwargs) @@ -1999,12 +2001,15 @@ async def _send_control(self, entity_id, command, **kwargs): # An ack for this id may already have arrived while the publish was awaiting its # broker confirmation: then the command did reach the hub, so keep it if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"] and command_ref not in entry["outcomes"]: - entry["command_ids"].remove(command_ref) - if not entry["command_ids"]: + if not previous["command_ids"]: del self._pending_commands[entity_id] - elif entry["state"] == "sent" and entry["outcomes"] == previous["outcomes"]: - # No ack arrived meanwhile: back to how the earlier attempt left it - entry.update({key: previous[key] for key in ("outcomes", "state", "sent_at", "resent", "cached")}) + elif entry["outcomes"] == outcomes_sent: + # No ack arrived meanwhile: back to how the earlier attempt left it, + # including any id pruned to make room for this one + entry.update(previous) + else: + entry["command_ids"].remove(command_ref) + entry["attempt_ids"].remove(command_ref) raise def _process_ack(self, data): @@ -2028,7 +2033,7 @@ def _process_ack(self, data): ok = ack.get("ok") is True error = ack.get("error") or "unknown" replay = not ok and error == "replay" - apply_cache = restore_status = False + apply_cache = restore_status = keep_entry = False with self._command_lock: entity_id, entry = next(((e, r) for e, r in self._pending_commands.items() if command_id in r["command_ids"]), (None, None)) @@ -2036,11 +2041,17 @@ def _process_ack(self, data): if entry is None or ack.get("command", entry["command"]) != entry["command"]: return self._acks_seen = True - # A replay for the id still awaiting its first answer means the hub has seen that id - # before and never considered the value: let the next attempt go out under a fresh id - # straight away. For an older or already-answered id it is just another refusal. - drop_replay = replay and entry["state"] == "sent" and entry["command_ids"][-1] == command_id and command_id not in entry["outcomes"] - if drop_replay: + # A replay for the newest id, before any unit answered it, means the hub has seen that + # id before and never considered this send. Drop the id; if nothing else has answered + # the value, let the next attempt go out under a fresh id straight away. For an older + # or already-answered id a replay is just another unit's refusal. + drop_replay = replay and entry["command_ids"][-1] == command_id and command_id not in entry["outcomes"] + keep_entry = drop_replay and any(entry["outcomes"].get(ref) == "ok" for ref in entry["attempt_ids"]) + if keep_entry: + # Another send in this attempt already applied the value: only forget this id + entry["command_ids"].remove(command_id) + entry["attempt_ids"].remove(command_id) + elif drop_replay: del self._pending_commands[entity_id] else: if ok and entry["outcomes"].get(command_id) == "refused": @@ -2069,7 +2080,8 @@ def _process_ack(self, data): entry["cached"] = command_id entry["cached_ids"] = entry["cached_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_id] elif drop_replay: - self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: replay ({command_id} already used), re-sending with a new id") + action = "already applied by another send" if keep_entry else "re-sending with a new id" + self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: replay ({command_id} already used), {action}") else: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") if restore_status and self._last_status is not None: diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 31bcab487..2878f3040 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5169,6 +5169,69 @@ def test_late_replay_for_an_earlier_id_does_not_drop_the_fresh_attempt(self): self._run(gw.select_event(self.SLOT_START, "00:30:00")) assert len(gw._published) == 2 + def test_replay_for_the_newest_id_after_an_earlier_send_applied(self): + """PBAT1 unanswered, PBAT2 re-sent, PBAT1 ok, then PBAT2 replay: PBAT2 is forgotten and the applied value stays suppressed.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + clock["now"] = 1030.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + self._ack(gw, "PBAT2", "set_charge_rate", ok=False, error="replay") + entry = gw._pending_commands[self.CHARGE_RATE] + assert entry["command_ids"] == ["PBAT1"] + assert entry["state"] == "applied" + assert self._logged(gw, "already applied by another send") + clock["now"] = 1040.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 2 + + def test_replay_for_the_newest_id_with_only_an_older_attempts_ok_drops_the_entry(self): + """An ok from an earlier attempt does not count: a replay of the fresh attempt's id re-sends straight away.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + clock["now"] = 1031.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT2", "set_charge_rate", ok=False, error="replay") + assert self.CHARGE_RATE not in gw._pending_commands + clock["now"] = 1032.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert [p[1] for p in gw._published] == [1, 2, 3] + + def test_failed_publish_with_full_id_history_restores_the_pruned_id(self): + """With 16 ids kept, a failed 17th send puts back the pruned oldest id and the earlier timing.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + for step in range(16): + clock["now"] = 1000.0 + step * 2 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + before = dict(gw._pending_commands[self.CHARGE_RATE]) + before_ids = list(before["command_ids"]) + assert len(before_ids) == 16 + + async def failing_publish_command(command, command_id=None, **kwargs): + raise RuntimeError("broker gone") + + real_publish = gw.publish_command + gw.publish_command = failing_publish_command + clock["now"] = 1100.0 + try: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + except RuntimeError: + pass + gw.publish_command = real_publish + entry = gw._pending_commands[self.CHARGE_RATE] + assert entry["command_ids"] == before_ids + assert entry["sent_at"] == before["sent_at"] + # The oldest id is still matchable + self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + def test_outcomes_do_not_grow_past_the_kept_ids(self): """Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry.""" gw = self._make_gateway() From 58cec7838e2f5e26faf58effdcc6b490b314fc7a Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 22:44:07 +0100 Subject: [PATCH 10/11] fix(gateway): bound attempt ids, detect acks during a failing publish, time replays Review follow-ups: - prune the per-attempt id list with the kept ids; - roll back a failed publish only if no ack of any kind arrived meanwhile (an ack counter, not an outcome comparison); - a replay of the newest id after another send applied the value still restarts the refusal window. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 10 +++--- apps/predbat/tests/test_gateway.py | 51 ++++++++++++++++++++++++++++-- 2 files changed, 55 insertions(+), 6 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 4df184452..f92059187 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1973,7 +1973,7 @@ async def _send_control(self, entity_id, command, **kwargs): self._acks_seen = False self.log(f"Warn: GatewayMQTT: {command} for {kwargs.get('serial')} was never acknowledged, re-sending until telemetry confirms it") if not same: - entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": []} + entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": [], "acks": 0} self._pending_commands[entity_id] = entry previous = {"command_ids": list(entry["command_ids"]), "attempt_ids": list(entry["attempt_ids"]), "outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} if entry["state"] != "sent": @@ -1988,9 +1988,9 @@ async def _send_control(self, entity_id, command, **kwargs): # Earlier ids of the same value stay matchable: an ack for any of them confirms it entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref] entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]} - entry["attempt_ids"] = entry["attempt_ids"] + [command_ref] + entry["attempt_ids"] = [ref for ref in entry["attempt_ids"] if ref in entry["command_ids"]] + [command_ref] entry["sent_at"] = now - outcomes_sent = dict(entry["outcomes"]) + acks_before = entry["acks"] try: await self.publish_command(command, command_id=command_id, **kwargs) @@ -2003,7 +2003,7 @@ async def _send_control(self, entity_id, command, **kwargs): if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"] and command_ref not in entry["outcomes"]: if not previous["command_ids"]: del self._pending_commands[entity_id] - elif entry["outcomes"] == outcomes_sent: + elif entry["acks"] == acks_before: # No ack arrived meanwhile: back to how the earlier attempt left it, # including any id pruned to make room for this one entry.update(previous) @@ -2041,6 +2041,7 @@ def _process_ack(self, data): if entry is None or ack.get("command", entry["command"]) != entry["command"]: return self._acks_seen = True + entry["acks"] += 1 # lets a failing publish tell whether any ack arrived meanwhile # A replay for the newest id, before any unit answered it, means the hub has seen that # id before and never considered this send. Drop the id; if nothing else has answered # the value, let the next attempt go out under a fresh id straight away. For an older @@ -2051,6 +2052,7 @@ def _process_ack(self, data): # Another send in this attempt already applied the value: only forget this id entry["command_ids"].remove(command_id) entry["attempt_ids"].remove(command_id) + entry["sent_at"] = _monotonic() # still a refusal: the window runs from it elif drop_replay: del self._pending_commands[entity_id] else: diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 2878f3040..16784ff74 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5177,15 +5177,21 @@ def test_replay_for_the_newest_id_after_an_earlier_send_applied(self): self._run(gw.number_event(self.CHARGE_RATE, 2000)) clock["now"] = 1030.0 self._run(gw.number_event(self.CHARGE_RATE, 2000)) + clock["now"] = 1031.0 self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) + clock["now"] = 1059.0 self._ack(gw, "PBAT2", "set_charge_rate", ok=False, error="replay") entry = gw._pending_commands[self.CHARGE_RATE] assert entry["command_ids"] == ["PBAT1"] assert entry["state"] == "applied" assert self._logged(gw, "already applied by another send") - clock["now"] = 1040.0 + # The replay is a refusal too: the window runs from it + clock["now"] = 1060.0 self._run(gw.number_event(self.CHARGE_RATE, 2000)) - assert len(gw._published) == 2 + assert len(gw._published) == 2 + clock["now"] = 1089.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 3 def test_replay_for_the_newest_id_with_only_an_older_attempts_ok_drops_the_entry(self): """An ok from an earlier attempt does not count: a replay of the fresh attempt's id re-sends straight away.""" @@ -5232,6 +5238,34 @@ async def failing_publish_command(command, command_id=None, **kwargs): self._ack(gw, "PBAT1", "set_charge_rate", applied={"power_w": 2000, "slot": 0}) assert gw._pending_commands[self.CHARGE_RATE]["state"] == "applied" + def test_repeated_refusal_during_a_failing_publish_is_not_rolled_back(self): + """A refusal repeated while a later publish awaits the broker keeps its restarted window when that publish fails.""" + gw = self._make_gateway() + clock, patcher = self._clock() + with patcher: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="not_managed") + + async def refusal_then_fail(command, command_id=None, **kwargs): + clock["now"] = 1035.0 + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="not_managed") + raise RuntimeError("broker gone") + + real_publish = gw.publish_command + gw.publish_command = refusal_then_fail + clock["now"] = 1031.0 + try: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + except RuntimeError: + pass + gw.publish_command = real_publish + entry = gw._pending_commands[self.CHARGE_RATE] + assert entry["command_ids"] == ["PBAT1"] + assert entry["sent_at"] == 1035.0 + clock["now"] = 1036.0 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + assert len(gw._published) == 1 + def test_outcomes_do_not_grow_past_the_kept_ids(self): """Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry.""" gw = self._make_gateway() @@ -5245,6 +5279,19 @@ def test_outcomes_do_not_grow_past_the_kept_ids(self): assert len(entry["command_ids"]) == 16 assert set(entry["outcomes"]) <= set(entry["command_ids"]) + def test_attempt_ids_do_not_grow_without_acks(self): + """Subscribed but no acks ever: every call re-sends, and the id lists stay bounded.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + for step in range(50): + clock["now"] = 1000.0 + step * 2 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + entry = gw._pending_commands[self.CHARGE_RATE] + assert len(gw._published) == 50 + assert len(entry["command_ids"]) == 16 + assert len(entry["attempt_ids"]) == 16 + def test_second_ok_ack_does_not_overwrite_cache(self): """Only the first ok ack for a command updates the cached slot times.""" gw = self._make_gateway() From 756cb42f7bcc855143b65fc19559205059d6a228 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 25 Sep 2026 22:52:29 +0100 Subject: [PATCH 11/11] fix(gateway): prune the oldest command id only after the new send is out A failing publish then never loses an earlier id, even when the history is full and acks for other ids arrive while it awaits the broker. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 28 +++++++++++++++++----------- apps/predbat/tests/test_gateway.py | 27 +++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 11 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index f92059187..e58bfd20d 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1975,7 +1975,7 @@ async def _send_control(self, entity_id, command, **kwargs): if not same: entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": [], "acks": 0} self._pending_commands[entity_id] = entry - previous = {"command_ids": list(entry["command_ids"]), "attempt_ids": list(entry["attempt_ids"]), "outcomes": dict(entry["outcomes"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} + previous = {"attempt_ids": list(entry["attempt_ids"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]} if entry["state"] != "sent": # A fresh attempt after the previous one was answered. Earlier ids and their # outcomes stay, so a late ack from another unit is still logged and acted on and @@ -1985,10 +1985,10 @@ async def _send_control(self, entity_id, command, **kwargs): entry["resent"] = True # the one re-send after an unanswered window command_id = self._command_id = self._command_id + 1 command_ref = f"PBAT{command_id}" - # Earlier ids of the same value stay matchable: an ack for any of them confirms it - entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref] - entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]} - entry["attempt_ids"] = [ref for ref in entry["attempt_ids"] if ref in entry["command_ids"]] + [command_ref] + # Earlier ids of the same value stay matchable: an ack for any of them confirms it. + # The oldest id is only pruned once this send has gone out (see below). + entry["command_ids"] = entry["command_ids"] + [command_ref] + entry["attempt_ids"] = entry["attempt_ids"] + [command_ref] entry["sent_at"] = now acks_before = entry["acks"] @@ -2001,17 +2001,23 @@ async def _send_control(self, entity_id, command, **kwargs): # An ack for this id may already have arrived while the publish was awaiting its # broker confirmation: then the command did reach the hub, so keep it if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"] and command_ref not in entry["outcomes"]: - if not previous["command_ids"]: + entry["command_ids"].remove(command_ref) + entry["attempt_ids"].remove(command_ref) + if not entry["command_ids"]: del self._pending_commands[entity_id] elif entry["acks"] == acks_before: - # No ack arrived meanwhile: back to how the earlier attempt left it, - # including any id pruned to make room for this one + # No ack arrived meanwhile: back to how the earlier attempt left it entry.update(previous) - else: - entry["command_ids"].remove(command_ref) - entry["attempt_ids"].remove(command_ref) raise + with self._command_lock: + # Keep the id history bounded now that this send is out + if self._pending_commands.get(entity_id) is entry and len(entry["command_ids"]) > _COMMAND_ACK_IDS_KEPT: + entry["command_ids"] = entry["command_ids"][-_COMMAND_ACK_IDS_KEPT:] + entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]} + entry["attempt_ids"] = [ref for ref in entry["attempt_ids"] if ref in entry["command_ids"]] + entry["cached_ids"] = [ref for ref in entry["cached_ids"] if ref in entry["command_ids"]] + def _process_ack(self, data): """Handle a command ack from the hub. diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 16784ff74..fa1021724 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5266,6 +5266,33 @@ async def refusal_then_fail(command, command_id=None, **kwargs): self._run(gw.number_event(self.CHARGE_RATE, 2000)) assert len(gw._published) == 1 + def test_full_history_failing_publish_with_concurrent_ack_keeps_all_earlier_ids(self): + """16 ids kept, a 17th send fails while an ack for PBAT2 arrives: PBAT1..PBAT16 all stay matchable.""" + gw = self._make_gateway(acks_seen=False) + clock, patcher = self._clock() + with patcher: + for step in range(16): + clock["now"] = 1000.0 + step * 2 + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + + async def ack_other_then_fail(command, command_id=None, **kwargs): + self._ack(gw, "PBAT2", "set_charge_rate", ok=False, error="not_managed") + self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="modbus_write_failed") + raise RuntimeError("broker gone") + + real_publish = gw.publish_command + gw.publish_command = ack_other_then_fail + clock["now"] = 1100.0 + try: + self._run(gw.number_event(self.CHARGE_RATE, 2000)) + except RuntimeError: + pass + gw.publish_command = real_publish + entry = gw._pending_commands[self.CHARGE_RATE] + assert entry["command_ids"] == [f"PBAT{n}" for n in range(1, 17)] + assert entry["outcomes"] == {"PBAT2": "refused", "PBAT1": "refused"} + assert self._logged(gw, "refused by hub: modbus_write_failed") + def test_outcomes_do_not_grow_past_the_kept_ids(self): """Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry.""" gw = self._make_gateway()