Skip to content

Commit 756cb42

Browse files
mgazzaclaude
andcommitted
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf
1 parent 58cec78 commit 756cb42

2 files changed

Lines changed: 44 additions & 11 deletions

File tree

‎apps/predbat/gateway.py‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1975,7 +1975,7 @@ async def _send_control(self, entity_id, command, **kwargs):
19751975
if not same:
19761976
entry = {"command_ids": [], "outcomes": {}, "command": command, "kwargs": dict(kwargs), "state": "sent", "sent_at": now, "resent": False, "cached": None, "cached_ids": [], "attempt_ids": [], "acks": 0}
19771977
self._pending_commands[entity_id] = entry
1978-
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"]}
1978+
previous = {"attempt_ids": list(entry["attempt_ids"]), "state": entry["state"], "sent_at": entry["sent_at"], "resent": entry["resent"], "cached": entry["cached"]}
19791979
if entry["state"] != "sent":
19801980
# A fresh attempt after the previous one was answered. Earlier ids and their
19811981
# 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):
19851985
entry["resent"] = True # the one re-send after an unanswered window
19861986
command_id = self._command_id = self._command_id + 1
19871987
command_ref = f"PBAT{command_id}"
1988-
# Earlier ids of the same value stay matchable: an ack for any of them confirms it
1989-
entry["command_ids"] = entry["command_ids"][-(_COMMAND_ACK_IDS_KEPT - 1) :] + [command_ref]
1990-
entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]}
1991-
entry["attempt_ids"] = [ref for ref in entry["attempt_ids"] if ref in entry["command_ids"]] + [command_ref]
1988+
# Earlier ids of the same value stay matchable: an ack for any of them confirms it.
1989+
# The oldest id is only pruned once this send has gone out (see below).
1990+
entry["command_ids"] = entry["command_ids"] + [command_ref]
1991+
entry["attempt_ids"] = entry["attempt_ids"] + [command_ref]
19921992
entry["sent_at"] = now
19931993
acks_before = entry["acks"]
19941994

@@ -2001,17 +2001,23 @@ async def _send_control(self, entity_id, command, **kwargs):
20012001
# An ack for this id may already have arrived while the publish was awaiting its
20022002
# broker confirmation: then the command did reach the hub, so keep it
20032003
if self._pending_commands.get(entity_id) is entry and command_ref in entry["command_ids"] and command_ref not in entry["outcomes"]:
2004-
if not previous["command_ids"]:
2004+
entry["command_ids"].remove(command_ref)
2005+
entry["attempt_ids"].remove(command_ref)
2006+
if not entry["command_ids"]:
20052007
del self._pending_commands[entity_id]
20062008
elif entry["acks"] == acks_before:
2007-
# No ack arrived meanwhile: back to how the earlier attempt left it,
2008-
# including any id pruned to make room for this one
2009+
# No ack arrived meanwhile: back to how the earlier attempt left it
20092010
entry.update(previous)
2010-
else:
2011-
entry["command_ids"].remove(command_ref)
2012-
entry["attempt_ids"].remove(command_ref)
20132011
raise
20142012

2013+
with self._command_lock:
2014+
# Keep the id history bounded now that this send is out
2015+
if self._pending_commands.get(entity_id) is entry and len(entry["command_ids"]) > _COMMAND_ACK_IDS_KEPT:
2016+
entry["command_ids"] = entry["command_ids"][-_COMMAND_ACK_IDS_KEPT:]
2017+
entry["outcomes"] = {key: value for key, value in entry["outcomes"].items() if key in entry["command_ids"]}
2018+
entry["attempt_ids"] = [ref for ref in entry["attempt_ids"] if ref in entry["command_ids"]]
2019+
entry["cached_ids"] = [ref for ref in entry["cached_ids"] if ref in entry["command_ids"]]
2020+
20152021
def _process_ack(self, data):
20162022
"""Handle a command ack from the hub.
20172023

‎apps/predbat/tests/test_gateway.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5266,6 +5266,33 @@ async def refusal_then_fail(command, command_id=None, **kwargs):
52665266
self._run(gw.number_event(self.CHARGE_RATE, 2000))
52675267
assert len(gw._published) == 1
52685268

5269+
def test_full_history_failing_publish_with_concurrent_ack_keeps_all_earlier_ids(self):
5270+
"""16 ids kept, a 17th send fails while an ack for PBAT2 arrives: PBAT1..PBAT16 all stay matchable."""
5271+
gw = self._make_gateway(acks_seen=False)
5272+
clock, patcher = self._clock()
5273+
with patcher:
5274+
for step in range(16):
5275+
clock["now"] = 1000.0 + step * 2
5276+
self._run(gw.number_event(self.CHARGE_RATE, 2000))
5277+
5278+
async def ack_other_then_fail(command, command_id=None, **kwargs):
5279+
self._ack(gw, "PBAT2", "set_charge_rate", ok=False, error="not_managed")
5280+
self._ack(gw, "PBAT1", "set_charge_rate", ok=False, error="modbus_write_failed")
5281+
raise RuntimeError("broker gone")
5282+
5283+
real_publish = gw.publish_command
5284+
gw.publish_command = ack_other_then_fail
5285+
clock["now"] = 1100.0
5286+
try:
5287+
self._run(gw.number_event(self.CHARGE_RATE, 2000))
5288+
except RuntimeError:
5289+
pass
5290+
gw.publish_command = real_publish
5291+
entry = gw._pending_commands[self.CHARGE_RATE]
5292+
assert entry["command_ids"] == [f"PBAT{n}" for n in range(1, 17)]
5293+
assert entry["outcomes"] == {"PBAT2": "refused", "PBAT1": "refused"}
5294+
assert self._logged(gw, "refused by hub: modbus_write_failed")
5295+
52695296
def test_outcomes_do_not_grow_past_the_kept_ids(self):
52705297
"""Per-id outcomes are pruned with the ids, so a value re-sent for a long time does not grow the entry."""
52715298
gw = self._make_gateway()

0 commit comments

Comments
 (0)