fix: re-handshake in place so a dropped session costs 0 polls, not 1

.github/workflows/docker-build.yml

@@ -3,6 +3,8 @@ name: Build and Push
 on:
   push:
     branches: [main]
+  pull_request:
+    branches: [main]
 
 env:
   REGISTRY: ghcr.io
@@ -11,7 +13,22 @@ env:
   IMAGE_NAME: programmer-network/tapo-exporter
 
 jobs:
+  # The poller talks to real hardware, so the only thing worth testing is its
+  # error handling — and that is exactly where the ~50% poll-success bug lived.
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - run: pip install --no-cache-dir prometheus-client==0.21.1
+      - run: python3 test_poller.py
+
   build:
+    needs: test
+    # Pull requests only validate; publishing :latest happens on main.
+    if: github.event_name == 'push'
     runs-on: ubuntu-latest
     permissions:
       contents: read

tapo-exporter.py

@@ -13,9 +13,10 @@
 import logging
 import os
 import sys
+import time
 
 from tapo import ApiClient
-from prometheus_client import CollectorRegistry, Gauge, start_http_server
+from prometheus_client import CollectorRegistry, Counter, Gauge, start_http_server
 
 logging.basicConfig(
     level=logging.INFO,
@@ -41,6 +42,12 @@
 metric_on_since = Gauge("tapo_plug_on_since_seconds", "Seconds since powered on",     ["plug"], registry=registry)
 metric_today    = Gauge("tapo_plug_energy_today_kwh", "Energy today in kWh",          ["plug"], registry=registry)
 metric_month    = Gauge("tapo_plug_energy_month_kwh", "Energy this month in kWh",     ["plug"], registry=registry)
+# On failure the gauges above keep their last good values rather than going
+# stale-but-plausible in silence. This is how a consumer tells the difference:
+# time() - tapo_plug_last_success_seconds is the true age of every reading.
+metric_last_ok  = Gauge("tapo_plug_last_success_seconds", "Unix time of the last successful poll", ["plug"], registry=registry)
+metric_reauths  = Counter("tapo_plug_reauth_total",   "Mid-poll re-handshakes after a dropped session", ["plug"], registry=registry)
+metric_fails    = Counter("tapo_plug_poll_failures_total", "Polls that failed even after a re-handshake", ["plug"], registry=registry)
 
 
 def load_plugs() -> dict:
@@ -56,22 +63,45 @@ def load_plugs() -> dict:
 
 
 class PlugPoller:
-    """Holds a reused device session per plug; re-handshakes only on error."""
+    """Holds a reused device session per plug, re-handshaking in place on error.
 
-    def __init__(self, client: ApiClient, name: str, host: str):
-        self.client = client
+    Each poller owns its own ApiClient. Sharing one across plugs makes their
+    handshakes race — the plugs answer 403 Forbidden for every request carrying
+    a session another plug's handshake has since superseded.
+    """
+
+    def __init__(self, user: str, password: str, name: str, host: str):
+        self.client = ApiClient(user, password)
         self.name = name
         self.host = host
         self.device = None
 
+    async def _read(self):
+        """One full read against the current session, handshaking if needed."""
+        if self.device is None:
+            self.device = await asyncio.wait_for(self.client.p110(self.host), OP_TIMEOUT)
+        info = await asyncio.wait_for(self.device.get_device_info(), OP_TIMEOUT)
+        energy = await asyncio.wait_for(self.device.get_energy_usage(), OP_TIMEOUT)
+        power = await asyncio.wait_for(self.device.get_current_power(), OP_TIMEOUT)
+        return info, energy, power
+
     async def poll(self):
         labels = {"plug": self.name}
         try:
-            if self.device is None:
-                self.device = await asyncio.wait_for(self.client.p110(self.host), OP_TIMEOUT)
-            info = await asyncio.wait_for(self.device.get_device_info(), OP_TIMEOUT)
-            energy = await asyncio.wait_for(self.device.get_energy_usage(), OP_TIMEOUT)
-            power = await asyncio.wait_for(self.device.get_current_power(), OP_TIMEOUT)
+            try:
+                info, energy, power = await self._read()
+            except Exception as first:
+                # A dropped session is the overwhelmingly common failure here and
+                # it is recoverable immediately — the plug is fine, our token is
+                # not. Retrying on the NEXT cycle instead of this one is what
+                # pinned poll success at ~50%: every session survived exactly one
+                # poll, so the loop alternated success, expiry, re-handshake,
+                # expiry, forever. Reconnect and retry once, in place.
+                log.info("session lost for plug %s (%s): %s — re-handshaking",
+                         self.name, self.host, first)
+                metric_reauths.labels(**labels).inc()
+                self.device = None
+                info, energy, power = await self._read()
 
             metric_up.labels(**labels).set(1)
             metric_state.labels(**labels).set(1 if info.device_on else 0)
@@ -80,10 +110,16 @@ async def poll(self):
             metric_on_since.labels(**labels).set(info.on_time)
             metric_today.labels(**labels).set(energy.today_energy / 1000.0)    # Wh -> kWh
             metric_month.labels(**labels).set(energy.month_energy / 1000.0)    # Wh -> kWh
+            metric_last_ok.labels(**labels).set(time.time())
         except Exception as e:
+            # Both the original attempt and the retry failed, so this is not a
+            # stale session — the plug is genuinely unreachable, off the Wi-Fi,
+            # or the credentials are wrong.
             metric_up.labels(**labels).set(0)
+            metric_fails.labels(**labels).inc()
             self.device = None  # force a fresh handshake next cycle
-            log.warning("poll failed for plug %s (%s): %s", self.name, self.host, e)
+            log.warning("poll failed for plug %s (%s) after re-handshake: %s",
+                        self.name, self.host, e)
 
 
 async def poll_loop(pollers, interval: int):
@@ -105,8 +141,7 @@ def main():
         sys.exit(1)
 
     plugs = load_plugs()
-    client = ApiClient(user, password)
-    pollers = [PlugPoller(client, name, host) for name, host in plugs.items()]
+    pollers = [PlugPoller(user, password, name, host) for name, host in plugs.items()]
 
     # Start the HTTP server FIRST (daemon thread) so /metrics — and thus the k8s
     # readiness probe — is serving immediately, even if the plugs are unreachable.

test_poller.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Regression test for the ~50% poll-success bug.
+
+In the field, every Tapo session survived exactly one poll and then answered
+403 Forbidden / Tapo(SessionTimeout). The old poll() reacted by clearing the
+device and giving up until the NEXT cycle, so the loop alternated success,
+expiry, re-handshake, expiry — pinning `avg_over_time(tapo_plug_up[3h])` at
+0.503 on all three plugs while power gauges kept updating from the good half.
+
+This test models exactly that plug: a session that dies after one full read.
+The poller must absorb it and report every poll as successful.
+
+Run: python3 test_poller.py   (needs prometheus-client; no network, no plugs)
+"""
+import asyncio
+import importlib.util
+import os
+import sys
+import types
+
+
+# ── a plug whose session dies after exactly one complete read ────────────────
+class FakeInfo:
+    device_on, rssi, on_time = True, -55, 1234
+
+
+class FakeEnergy:
+    today_energy, month_energy = 612.0, 18000.0
+
+
+class FakePower:
+    current_power = 50.0
+
+
+class FakeDevice:
+    def __init__(self):
+        self.spent = False
+
+    async def _guard(self):
+        if self.spent:
+            raise RuntimeError("Tapo(SessionTimeout): 403 Forbidden")
+
+    async def get_device_info(self):
+        await self._guard()
+        return FakeInfo()
+
+    async def get_energy_usage(self):
+        await self._guard()
+        return FakeEnergy()
+
+    async def get_current_power(self):
+        await self._guard()
+        self.spent = True          # the read that consumes the session
+        return FakePower()
+
+
+class FakeApiClient:
+    def __init__(self, user, password):
+        self.handshakes = 0
+
+    async def p110(self, host):
+        self.handshakes += 1
+        return FakeDevice()
+
+
+# The exporter is a single script, not a package — stub `tapo` before loading it.
+fake_tapo = types.ModuleType("tapo")
+fake_tapo.ApiClient = FakeApiClient
+sys.modules["tapo"] = fake_tapo
+
+_here = os.path.dirname(os.path.abspath(__file__))
+_spec = importlib.util.spec_from_file_location(
+    "tapo_exporter", os.path.join(_here, "tapo-exporter.py"))
+exporter = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(exporter)
+
+PLUG = "DGX Spark"
+
+
+def gauge(metric):
+    return metric.labels(plug=PLUG)._value.get()
+
+
+async def test_expiring_session_still_polls_cleanly():
+    poller = exporter.PlugPoller("user", "password", PLUG, "192.168.10.217")
+
+    ups = []
+    for _ in range(10):
+        await poller.poll()
+        ups.append(gauge(exporter.metric_up))
+
+    rate = sum(ups) / len(ups)
+    print(f"up per poll  : {[int(u) for u in ups]}")
+    print(f"success rate : {rate:.0%}")
+    print(f"reauths      : {int(gauge(exporter.metric_reauths))}")
+    print(f"hard failures: {int(gauge(exporter.metric_fails))}")
+
+    assert rate == 1.0, f"expected every poll to succeed, got {rate:.0%}"
+    assert gauge(exporter.metric_fails) == 0, "a recoverable expiry was reported as a failure"
+    assert gauge(exporter.metric_power) == 50.0, "power gauge not populated"
+    assert gauge(exporter.metric_last_ok) > 0, "last-success timestamp not set"
+
+
+async def test_unreachable_plug_is_reported_down():
+    """The retry must not paper over a plug that is genuinely gone."""
+    class DeadClient(FakeApiClient):
+        async def p110(self, host):
+            raise RuntimeError("connection refused")
+
+    poller = exporter.PlugPoller("user", "password", PLUG, "192.168.10.217")
+    poller.client = DeadClient("user", "password")
+    before = gauge(exporter.metric_fails)
+    await poller.poll()
+
+    assert gauge(exporter.metric_up) == 0, "unreachable plug should report up=0"
+    assert gauge(exporter.metric_fails) == before + 1, "failure counter not incremented"
+    print("unreachable plug correctly reported down")
+
+
+async def main():
+    await test_expiring_session_still_polls_cleanly()
+    await test_unreachable_plug_is_reported_down()
+    print("\nOK")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())