fix(poller): no catch-up burst, sink errors do not mark devices down

After a stall the deadline stayed in the past and several polls ran
back to back; it now resets so one poll follows. A ReadingSink failure
was counted as a device failure; it is now logged once per outage.

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

src/main/java/home/gateway/application/Poller.java

@@ -1,10 +1,13 @@
 package home.gateway.application;
 
+import home.gateway.domain.Reading;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Clock;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.locks.LockSupport;
 
@@ -18,6 +21,7 @@ public class Poller implements AutoCloseable {
     private final ReadingSink sink;
     private final Clock clock;
     private final List<Thread> threads = new CopyOnWriteArrayList<>();
+    private final Set<String> devicesWithSinkFailure = ConcurrentHashMap.newKeySet();
     // A read may swallow the interrupt, so the loop cannot rely on the thread's flag alone.
     private volatile boolean closed;
 
@@ -54,24 +58,29 @@ public class Poller implements AutoCloseable {
         while (!closed) {
             up = pollOnce(device, up);
             next += intervalNanos;
-            // An overrun leaves next in the past, so the following poll starts at once.
-            long wait = next - System.nanoTime();
-            if (wait > 0) {
-                LockSupport.parkNanos(wait);
+            if (next < System.nanoTime()) {
+                // After an overrun one poll follows at once; keeping the old deadline would
+                // fire several back to back and hammer a device that just stalled.
+                next = System.nanoTime();
+            } else {
+                parkUntil(next);
             }
         }
     }
 
+    // parkNanos may return early, and an early return would shorten the interval.
+    private void parkUntil(long deadline) {
+        long wait;
+        while (!closed && (wait = deadline - System.nanoTime()) > 0) {
+            LockSupport.parkNanos(wait);
+        }
+    }
+
     private boolean pollOnce(Device device, boolean wasUp) {
         var now = clock.instant();
+        List<Reading> readings;
         try {
-            var readings = device.read(now);
-            snapshot.recordSuccess(device.name(), now, readings);
-            sink.write(readings);
-            if (!wasUp) {
-                LOGGER.info("{} is back up", device.name());
-            }
-            return true;
+            readings = device.read(now);
         } catch (Exception e) {
             snapshot.recordFailure(device.name(), now, String.valueOf(e.getMessage()));
             if (wasUp) {
@@ -79,5 +88,25 @@ public class Poller implements AutoCloseable {
             }
             return false;
         }
+        snapshot.recordSuccess(device.name(), now, readings);
+        writeToSink(device.name(), readings);
+        if (!wasUp) {
+            LOGGER.info("{} is back up", device.name());
+        }
+        return true;
+    }
+
+    // A storage problem says nothing about the device, so it must not mark it down.
+    private void writeToSink(String deviceName, List<Reading> readings) {
+        try {
+            sink.write(readings);
+            if (devicesWithSinkFailure.remove(deviceName)) {
+                LOGGER.info("Writing readings of {} works again", deviceName);
+            }
+        } catch (Exception e) {
+            if (devicesWithSinkFailure.add(deviceName)) {
+                LOGGER.warn("Cannot write readings of {}: {}", deviceName, e.toString());
+            }
+        }
     }
 }

src/test/java/home/gateway/application/PollerTest.java

@@ -10,10 +10,12 @@ import java.time.Instant;
 import java.time.ZoneOffset;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BooleanSupplier;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class PollerTest {
@@ -24,6 +26,7 @@ class PollerTest {
     private final Clock clock = Clock.fixed(NOW, ZoneOffset.UTC);
     private final Snapshot snapshot = new Snapshot();
     private final List<Reading> written = new CopyOnWriteArrayList<>();
+    private ReadingSink sink = written::addAll;
     private Poller poller;
 
     @AfterEach
@@ -32,7 +35,7 @@ class PollerTest {
     }
 
     private void startPolling(Device... devices) {
-        poller = new Poller(List.of(devices), snapshot, written::addAll, clock);
+        poller = new Poller(List.of(devices), snapshot, sink, clock);
         poller.start();
     }
 
@@ -118,6 +121,40 @@ class PollerTest {
         assertTrue(Duration.ofNanos(System.nanoTime() - start).toMillis() < 380);
     }
 
+    @Test
+    void stallLongerThanTheIntervalIsFollowedByOneImmediatePollOnly() throws Exception {
+        var starts = new CopyOnWriteArrayList<Long>();
+        var device = new FakeDevice("pump", Duration.ofMillis(200), call -> {
+            starts.add(System.nanoTime());
+            if (call == 1) {
+                Thread.sleep(700);
+            }
+        });
+        startPolling(device);
+
+        await(() -> starts.size() >= 3);
+
+        long afterStall = Duration.ofNanos(starts.get(1) - starts.get(0)).toMillis();
+        long afterCatchUp = Duration.ofNanos(starts.get(2) - starts.get(1)).toMillis();
+        assertTrue(afterStall < 800);
+        assertTrue(afterCatchUp >= 150, "polls ran back to back after the stall");
+    }
+
+    @Test
+    void sinkFailureDoesNotMarkTheDeviceDown() throws Exception {
+        var attempts = new AtomicInteger();
+        sink = readings -> {
+            attempts.incrementAndGet();
+            throw new IllegalStateException("influx exploded");
+        };
+        startPolling(FakeDevice.working("pump", FAST));
+
+        await(() -> attempts.get() >= 3);
+
+        assertNotNull(state("pump").lastSuccess());
+        assertNull(state("pump").lastError());
+    }
+
     @Test
     void closeInterruptsSlowReadAndJoins() throws Exception {
         var slow = new FakeDevice("slow", FAST, call -> Thread.sleep(60_000));