diff --git a/src/main/java/home/gateway/application/CommandService.java b/src/main/java/home/gateway/application/CommandService.java index 6718b19..47bc5c6 100644 --- a/src/main/java/home/gateway/application/CommandService.java +++ b/src/main/java/home/gateway/application/CommandService.java @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.LongSupplier; import java.util.stream.Stream; /** @@ -34,37 +35,42 @@ public class CommandService implements Commands { private static final String AUDIT_DATABASE = "Dom_Pompa"; private static final String AUDIT_MEASUREMENT = "setpoint_audit"; private static final String RESTORE_USER = "restore"; + // A failed restore is retried after these waits, then every 15 minutes: no EEPROM write storm (S4). + private static final List RETRY_WAITS = List.of(Duration.ofMinutes(1), Duration.ofMinutes(2), + Duration.ofMinutes(5), Duration.ofMinutes(15)); private final Map allowlist = new LinkedHashMap<>(); private final SetpointRegisters registers; private final RestoreStore store; private final ReadingSink audit; private final Clock clock; + private final LongSupplier nanoTime; private final Duration minInterval; private final Pause pause; // Not synchronized: blocking I/O inside a monitor would pin a virtual thread's carrier. private final ReentrantLock lock = new ReentrantLock(); private final Map restores = new LinkedHashMap<>(); - private final Map lastWrite = new HashMap<>(); - private final Map restoreProblems = new HashMap<>(); + private final Map lastWrite = new HashMap<>(); + private final Map backoffs = new HashMap<>(); + + private record Backoff(int failures, long retryAtNanos) { + } /** Fails when the store cannot be read: pending restores must never be dropped silently. */ public CommandService(List setpoints, SetpointRegisters registers, RestoreStore store, - ReadingSink audit, Clock clock, Duration minInterval, Pause pause) throws IOException { + ReadingSink audit, Clock clock, LongSupplier nanoTime, Duration minInterval, Pause pause) + throws IOException { setpoints.forEach(SafetyLimits::require); setpoints.forEach(setpoint -> allowlist.put(setpoint.name(), setpoint)); this.registers = registers; this.store = store; this.audit = audit; this.clock = clock; + this.nanoTime = nanoTime; this.minInterval = minInterval; this.pause = pause; store.load().forEach(restore -> restores.put(restore.register(), restore)); - var loaded = restores.size(); List.copyOf(restores.values()).forEach(this::isValid); - if (restores.size() != loaded) { - persist(); - } } @Override @@ -91,8 +97,7 @@ public class CommandService implements Commands { return rejected(command, setpoint, "a boost may last at most " + Command.MAX_BOOST.toHours() + " h"); } var now = clock.instant(); - var previous = lastWrite.get(setpoint.register()); - if (previous != null && Duration.between(previous, now).compareTo(minInterval) < 0) { + if (tooSoon(setpoint.register())) { return rejected(command, setpoint, "too soon after the last write to " + setpoint.name()); } return write(command, setpoint, now); @@ -112,7 +117,7 @@ public class CommandService implements Commands { // Saved before the write, so a write never happens without its restore on disk. remember(new Restore(setpoint.name(), register, original, command.value(), restoreAt), previous); } - lastWrite.put(register, now); + lastWrite.put(register, nanoTime.getAsLong()); attempted = true; registers.write(register, command.value()); pause.await(); @@ -148,13 +153,22 @@ public class CommandService implements Commands { return failed(command, setpoint, old, message); } - /** Runs the restores whose time has come, oldest first; one that fails stays pending. */ + private boolean tooSoon(int register) { + var last = lastWrite.get(register); + return last != null && nanoTime.getAsLong() - last < minInterval.toNanos(); + } + + /** + * Runs the restores whose time has come, oldest first. One that fails stays pending and is + * retried after a growing wait; the rate limit of S4 holds for restores too. + */ public void runDueRestores() { lock.lock(); try { var now = clock.instant(); restores.values().stream() .filter(restore -> !restore.due().isAfter(now)) + .filter(restore -> !isWaiting(restore)) .sorted(Comparator.comparing(Restore::due)) .toList() .forEach(this::tryRestore); @@ -163,6 +177,12 @@ public class CommandService implements Commands { } } + private boolean isWaiting(Restore restore) { + var backoff = backoffs.get(restore.register()); + var backingOff = backoff != null && backoff.retryAtNanos() - nanoTime.getAsLong() > 0; + return backingOff || tooSoon(restore.register()); + } + /** One restore that is bad or blows up must not keep the others from running. */ private void tryRestore(Restore restore) { try { @@ -180,7 +200,7 @@ public class CommandService implements Commands { if (problem == null) { return true; } - restores.remove(restore.register()); + forget(restore.register()); auditRestore(restore, "INVALID", null); LOGGER.error("restore dropped, never written: {} ({})", problem, restore); return false; @@ -212,6 +232,7 @@ public class CommandService implements Commands { current, restore.appliedValue()); return; } + lastWrite.put(restore.register(), nanoTime.getAsLong()); registers.write(restore.register(), restore.originalValue()); pause.await(); var actual = registers.read(restore.register()); @@ -221,7 +242,6 @@ public class CommandService implements Commands { } forget(restore.register()); auditRestore(restore, "APPLIED", current); - restoreProblems.remove(restore.register()); LOGGER.info("{} restored to {}", restore.setpoint(), restore.originalValue()); } catch (IOException e) { restoreFailed(restore, current, e.getMessage()); @@ -233,9 +253,12 @@ public class CommandService implements Commands { private void restoreFailed(Restore restore, Integer old, String problem) { auditRestore(restore, "FAILED", old); - if (!problem.equals(restoreProblems.put(restore.register(), problem))) { - LOGGER.error("restore of {} failed, will retry: {}", restore.setpoint(), problem); - } + var earlier = backoffs.get(restore.register()); + var failures = earlier == null ? 1 : earlier.failures() + 1; + var wait = RETRY_WAITS.get(Math.min(failures, RETRY_WAITS.size()) - 1); + backoffs.put(restore.register(), new Backoff(failures, nanoTime.getAsLong() + wait.toNanos())); + LOGGER.error("restore of {} failed (attempt {}), next try in {}: {}", restore.setpoint(), failures, + wait, problem); } @Override @@ -257,6 +280,7 @@ public class CommandService implements Commands { restores.put(restore.register(), restore); try { store.save(List.copyOf(restores.values())); + backoffs.remove(restore.register()); } catch (IOException e) { if (previous == null) { restores.remove(restore.register()); @@ -269,6 +293,7 @@ public class CommandService implements Commands { private void forget(int register) { restores.remove(register); + backoffs.remove(register); persist(); } diff --git a/src/test/java/home/gateway/application/CommandServiceTest.java b/src/test/java/home/gateway/application/CommandServiceTest.java index ba400c8..7774d20 100644 --- a/src/test/java/home/gateway/application/CommandServiceTest.java +++ b/src/test/java/home/gateway/application/CommandServiceTest.java @@ -59,7 +59,7 @@ class CommandServiceTest { private CommandService service() throws IOException { registers.values.putIfAbsent(4144, 45); registers.values.putIfAbsent(4132, 20); - return new CommandService(List.of(CWU, CO), registers, store, audit::addAll, clock, + return new CommandService(List.of(CWU, CO), registers, store, audit::addAll, clock, clock::nanos, Duration.ofSeconds(60), () -> { pauses++; inPause.countDown(); @@ -261,7 +261,7 @@ class CommandServiceTest { }; assertThrows(IOException.class, () -> new CommandService(List.of(CWU), registers, broken, - audit::addAll, clock, Duration.ZERO, () -> { })); + audit::addAll, clock, clock::nanos, Duration.ZERO, () -> { })); } @Test @@ -333,6 +333,7 @@ class CommandServiceTest { assertEquals("FAILED", lastStatus()); assertEquals(50, registers.values.get(4144)); + clock.advance(Duration.ofMinutes(1)); registers.failWrites = false; service.runDueRestores(); @@ -538,7 +539,7 @@ class CommandServiceTest { var typo = new Setpoint("cwu_comfort", 4144, 30, 500, 1); assertThrows(IllegalArgumentException.class, () -> new CommandService(List.of(typo), registers, store, - audit::addAll, clock, Duration.ZERO, () -> { })); + audit::addAll, clock, clock::nanos, Duration.ZERO, () -> { })); } @Test @@ -553,11 +554,62 @@ class CommandServiceTest { assertEquals(Status.APPLIED, service.apply(command("cwu_comfort", 50, Duration.ofHours(72))).status()); } + @Test + void failedRestoreBacksOffThenRetriesEveryFifteenMinutesAndNeverGivesUp() throws IOException { + var service = boosted(); + clock.advance(HOUR); + registers.failWrites = true; + + for (var wait : List.of(1, 2, 5, 15, 15, 15)) { + var writes = registers.writes; + service.runDueRestores(); + assertEquals(writes + 1, registers.writes); + clock.advance(Duration.ofMinutes(wait).minusSeconds(1)); + service.runDueRestores(); + assertEquals(writes + 1, registers.writes); + clock.advance(Duration.ofSeconds(1)); + } + + assertEquals(6, audit.stream().filter(reading -> "FAILED".equals(reading.tags().get("status"))).count()); + assertEquals(1, service.pendingRestores().size()); + } + + @Test + void aRestoreWaitsForTheMinimumIntervalSinceTheLastWrite() throws IOException { + var service = service(); + service.apply(command("cwu_comfort", 50, Duration.ofSeconds(10))); + clock.advance(Duration.ofSeconds(10)); + var writes = registers.writes; + + service.runDueRestores(); + + assertEquals(writes, registers.writes); + assertEquals(1, service.pendingRestores().size()); + clock.advance(Duration.ofSeconds(50)); + service.runDueRestores(); + assertEquals(45, registers.values.get(4144)); + } + + @Test + void aRestoreCountsAsALastWrite() throws IOException { + var service = boosted(); + clock.advance(HOUR); + service.runDueRestores(); + + assertEquals(Status.REJECTED, service.apply(command("cwu_comfort", 40, null)).status()); + } + private static final class MutableClock extends Clock { private Instant now = T0; + private long nanos; void advance(Duration by) { now = now.plus(by); + nanos += by.toNanos(); + } + + long nanos() { + return nanos; } @Override