mariusz.pacek/home-pump
1
1 Fork 0
Code Issues Pull Requests 44 Actions Packages Projects Releases Wiki Activity

fix(restore): back off failed restores and respect the write rate limit

Browse Source 
A failed restore was retried every tick, writing the EEPROM every 30 s.
Retries now wait 1, 2, 5, 15 minutes, then 15 minutes each; every
failure is audited and logged, the restore stays pending. Restore
writes count as last writes and wait for the per-register minimum
interval. CommandService takes a monotonic time source for this.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
rootandClaude Sonnet 5.5 committed 2026-10-24 05:13:44 +02:00
1 parent 6d7704964a
commit 17b65dc1c0
2 files changed
+96 -19
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/main/java/home/gateway/application/CommandService.java
+41 -16
View File
@@ -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<Duration> RETRY_WAITS = List.of(Duration.ofMinutes(1), Duration.ofMinutes(2),
Duration.ofMinutes(5), Duration.ofMinutes(15));
private final Map<String, Setpoint> 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<Integer, Restore> restores = new LinkedHashMap<>();
private final Map<Integer, Instant> lastWrite = new HashMap<>();
private final Map<Integer, String> restoreProblems = new HashMap<>();
private final Map<Integer, Long> lastWrite = new HashMap<>();
private final Map<Integer, Backoff> 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<Setpoint> 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();
}
src/test/java/home/gateway/application/CommandServiceTest.java
+55 -3
View File
@@ -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