From eb0b23637bb22b9a0eccf3a880c944821db25af1 Mon Sep 17 00:00:00 2001 From: Mariusz Pacek Date: Sat, 24 Oct 2026 00:46:02 +0200 Subject: [PATCH] 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 --- .../java/home/gateway/application/Poller.java | 51 +++++++++++++++---- .../home/gateway/application/PollerTest.java | 39 +++++++++++++- 2 files changed, 78 insertions(+), 12 deletions(-) diff --git a/src/main/java/home/gateway/application/Poller.java b/src/main/java/home/gateway/application/Poller.java index 69c175f..f85278f 100644 --- a/src/main/java/home/gateway/application/Poller.java +++ b/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 threads = new CopyOnWriteArrayList<>(); + private final Set 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 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 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()); + } + } } } diff --git a/src/test/java/home/gateway/application/PollerTest.java b/src/test/java/home/gateway/application/PollerTest.java index 35253c0..abc98f0 100644 --- a/src/test/java/home/gateway/application/PollerTest.java +++ b/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 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(); + 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));