Offload Marstek poll handler so a slow powermeter can't stall listener

The MQTT Insights listener is a single async-for loop. When a Marstek
poll handler awaited binding.get_values() inline it could block every
subsequent message (other CT polls, Insights commands) for as long as
the powermeter took to yield a reading. Spawn each response in its own
task instead, and track/cancel those tasks on disconnect/shutdown.

Also snapshot _marstek_bindings under the lock before scanning in
_find_marstek_binding, and drop the type: ignore on the topic helpers
by returning an explicit 2-tuple.

Adds two integration tests: register-before-start populates subscriptions
on first connect, and a slow handler on one binding doesn't block a
concurrent fast poll on another binding.

https://claude.ai/code/session_01K5ypPxYASWXJewf7Lk9a1e

Author: claude

src/astrameter/mqtt_insights/marstek_mqtt.py

@@ -80,15 +80,18 @@ def parse_app_topic(topic: str) -> tuple[str, str] | None:
 
 
 def app_topics_for(binding: MarstekMqttBinding) -> tuple[str, str]:
-    return tuple(  # type: ignore[return-value]
-        t.format(ct_type=binding.ct_type, mac=binding.mac) for t in APP_TOPIC_TEMPLATES
+    old, new = APP_TOPIC_TEMPLATES
+    return (
+        old.format(ct_type=binding.ct_type, mac=binding.mac),
+        new.format(ct_type=binding.ct_type, mac=binding.mac),
     )
 
 
 def device_topics_for(binding: MarstekMqttBinding) -> tuple[str, str]:
-    return tuple(  # type: ignore[return-value]
-        t.format(ct_type=binding.ct_type, mac=binding.mac)
-        for t in DEVICE_TOPIC_TEMPLATES
+    old, new = DEVICE_TOPIC_TEMPLATES
+    return (
+        old.format(ct_type=binding.ct_type, mac=binding.mac),
+        new.format(ct_type=binding.ct_type, mac=binding.mac),
     )
 
 

src/astrameter/mqtt_insights/mqtt_insights_test.py

@@ -1058,7 +1058,99 @@ async def test_marstek_get_values_failure_suppressed(mqtt_broker):
             await _collect_messages(client, received, timeout=1)
             assert received == []
         # get_values was called but no reply was published
+        await _poll(lambda: binding.device_id in service._marstek_get_values_failed)
         assert calls == [("called",)]
-        assert binding.device_id in service._marstek_get_values_failed
     finally:
         await service.stop()
+
+
+@needs_mosquitto
+async def test_marstek_register_before_start_subscribes_on_connect(mqtt_broker):
+    """A binding registered before start() must get its App topics
+    subscribed on the first connect."""
+    port = mqtt_broker
+    service = _make_service(port)
+    binding, _ = _make_binding()
+    # Register *before* start — the service must pick this up on connect.
+    await service.register_marstek(binding)
+    await service.start()
+
+    try:
+        await service.wait_connected()
+        await _poll(lambda: service._client is not None)
+
+        async with aiomqtt.Client(hostname="127.0.0.1", port=port) as client:
+            await client.subscribe(f"hame_energy/HME-4/device/{binding.mac}/ctrl")
+            await client.publish(
+                f"hame_energy/HME-4/App/{binding.mac}/ctrl", payload=b"cd=1"
+            )
+            received = []
+            await _collect_messages(
+                client, received, timeout=5, stop=lambda _: len(received) >= 1
+            )
+        assert len(received) == 1
+        assert received[0].payload.startswith(
+            b"pwr_a=100,pwr_b=200,pwr_c=300,pwr_t=600"
+        )
+    finally:
+        await service.stop()
+
+
+@needs_mosquitto
+async def test_marstek_slow_handler_does_not_stall_listener(mqtt_broker):
+    """A slow get_values for one binding must not block polls for another.
+
+    With the offload-to-task design, the listener stays responsive even
+    while a prior poll handler is still awaiting its powermeter.
+    """
+    port = mqtt_broker
+    service = _make_service(port)
+
+    slow_gate = asyncio.Event()
+
+    async def _slow_values() -> list[float]:
+        # Block until the test explicitly releases this handler.
+        await slow_gate.wait()
+        return [1.0, 2.0, 3.0]
+
+    slow = MarstekMqttBinding(
+        device_id="slow-ct",
+        ct_type="HME-4",
+        mac="02b250111111",
+        get_values=_slow_values,
+        wifi_rssi=-50,
+    )
+    fast, _ = _make_binding(
+        device_id="fast-ct", mac="02b250222222", values=[10.0, 20.0, 30.0]
+    )
+
+    await service.register_marstek(slow)
+    await service.register_marstek(fast)
+    await service.start()
+
+    try:
+        await service.wait_connected()
+        await _poll(lambda: service._client is not None)
+
+        async with aiomqtt.Client(hostname="127.0.0.1", port=port) as client:
+            await client.subscribe(f"hame_energy/HME-4/device/{fast.mac}/ctrl")
+
+            # Trigger the slow poll first — its handler will block in get_values.
+            await client.publish(
+                f"hame_energy/HME-4/App/{slow.mac}/ctrl", payload=b"cd=1"
+            )
+            # Immediately trigger the fast poll — if the listener were
+            # stalled, we'd never see its reply.
+            await client.publish(
+                f"hame_energy/HME-4/App/{fast.mac}/ctrl", payload=b"cd=1"
+            )
+            received = []
+            await _collect_messages(
+                client, received, timeout=5, stop=lambda _: len(received) >= 1
+            )
+
+        assert len(received) == 1
+        assert received[0].payload.startswith(b"pwr_a=10,pwr_b=20,pwr_c=30,pwr_t=60")
+    finally:
+        slow_gate.set()
+        await service.stop()

src/astrameter/mqtt_insights/service.py

@@ -104,6 +104,10 @@ def __init__(self, config: MqttInsightsConfig) -> None:
         # Rate-limit per-device get_values failure logging so a broken
         # powermeter doesn't flood the log at hm2mqtt's poll cadence.
         self._marstek_get_values_failed: set[str] = set()
+        # In-flight poll handlers — tracked so one slow powermeter doesn't
+        # block the listener loop, and so we can cancel pending tasks on
+        # reconnect / shutdown.
+        self._marstek_tasks: set[asyncio.Task[None]] = set()
 
     # ── Public API (called from device event listeners) ───────────────
 
@@ -293,6 +297,7 @@ async def _run(self) -> None:
                     finally:
                         async with self._marstek_lock:
                             self._client = None
+                        await self._cancel_marstek_tasks()
 
             except asyncio.CancelledError:
                 self._connected.clear()
@@ -687,12 +692,14 @@ def _handle_device_command(self, device_id: str, cmd: dict) -> None:
     async def _handle_marstek_message(
         self, client: aiomqtt.Client, message: aiomqtt.Message
     ) -> None:
+        """Dispatch a poll quickly; offload the response to a task so a
+        slow powermeter can't stall the listener loop."""
         topic = str(message.topic)
         parsed = parse_app_topic(topic)
         if parsed is None:
             return
         ct_type, mac = parsed
-        binding = self._find_marstek_binding(ct_type, mac)
+        binding = await self._find_marstek_binding(ct_type, mac)
         if binding is None:
             logger.debug("Marstek MQTT: no binding for %s/%s", ct_type, mac)
             return
@@ -702,6 +709,13 @@ async def _handle_marstek_message(
             logger.debug("Marstek MQTT: non-poll payload on %s", topic)
             return
 
+        task = asyncio.create_task(self._serve_marstek_poll(client, binding))
+        self._marstek_tasks.add(task)
+        task.add_done_callback(self._marstek_tasks.discard)
+
+    async def _serve_marstek_poll(
+        self, client: aiomqtt.Client, binding: MarstekMqttBinding
+    ) -> None:
         try:
             watts = await binding.get_values()
         except Exception:
@@ -722,10 +736,23 @@ async def _handle_marstek_message(
             with contextlib.suppress(aiomqtt.MqttError):
                 await client.publish(reply_topic, payload=payload, qos=0, retain=False)
 
-    def _find_marstek_binding(
+    async def _cancel_marstek_tasks(self) -> None:
+        pending = tuple(self._marstek_tasks)
+        for task in pending:
+            task.cancel()
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)
+        self._marstek_tasks.clear()
+
+    async def _find_marstek_binding(
         self, ct_type: str, mac: str
     ) -> MarstekMqttBinding | None:
-        for binding in self._marstek_bindings.values():
-            if binding.ct_type == ct_type and binding.mac == mac.lower():
+        # Snapshot under the lock so a concurrent (un)register can't mutate
+        # the dict mid-scan.
+        async with self._marstek_lock:
+            candidates = tuple(self._marstek_bindings.values())
+        mac_lower = mac.lower()
+        for binding in candidates:
+            if binding.ct_type == ct_type and binding.mac == mac_lower:
                 return binding
         return None