fix(gateway): keep running with commands disabled on a corrupt restore file
Polling, health and the watchdog continue; /commands answers 503 until a restart and never writes. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
b1fce8ad31
commit
85600d70ef
6 files changed
+104
-7
No files matched your search
@@ -12,8 +12,11 @@ import home.gateway.adapters.modbus.ModbusTransport;
|
||||
import home.gateway.adapters.modbus.PduModbusClient;
|
||||
import home.gateway.adapters.modbus.TcpTransport;
|
||||
import home.gateway.adapters.solarman.SolarmanV5Transport;
|
||||
import home.gateway.adapters.store.CorruptRestoreFileException;
|
||||
import home.gateway.adapters.store.JsonRestoreStore;
|
||||
import home.gateway.application.CommandService;
|
||||
import home.gateway.application.Commands;
|
||||
import home.gateway.application.DisabledCommands;
|
||||
import home.gateway.application.Device;
|
||||
import home.gateway.application.Poller;
|
||||
import home.gateway.application.RestoreScheduler;
|
||||
@@ -106,7 +109,7 @@ public final class Gateway implements AutoCloseable {
|
||||
Watchdog watchdog = own(new Watchdog(snapshot, clock, critical,
|
||||
config.watchdog().allowedIntervals(), exit));
|
||||
watchdog.start(WATCHDOG_PERIOD);
|
||||
CommandService commands = config.commands().enabled() ? startCommands(writer) : null;
|
||||
Commands commands = config.commands().enabled() ? startCommands(writer) : null;
|
||||
// Pushed last, so it closes first: no request arrives while the rest shuts down.
|
||||
GatewayHttpServer http = own(new GatewayHttpServer(snapshot, clock, config.http().port(), critical));
|
||||
if (commands != null) {
|
||||
@@ -115,13 +118,25 @@ public final class Gateway implements AutoCloseable {
|
||||
http.start();
|
||||
}
|
||||
|
||||
private CommandService startCommands(InfluxWriter audit) throws IOException {
|
||||
/**
|
||||
* A restore file that cannot be read must not stop the heating side (polling, health), so the
|
||||
* write path stays off instead, and never writes, until someone has looked at the moved file.
|
||||
*/
|
||||
private Commands startCommands(InfluxWriter audit) throws IOException {
|
||||
var settings = config.commands();
|
||||
var registers = new ModbusSetpointRegisters(heatPumpModbus, config.devices().heatPump().unitId());
|
||||
var service = new CommandService(settings.setpoints(), registers,
|
||||
new JsonRestoreStore(settings.restoreFile()), audit, clock, System::nanoTime,
|
||||
EARLIEST_PLAUSIBLE, settings.minInterval(),
|
||||
() -> Thread.sleep(settings.readBackDelayMillis()));
|
||||
CommandService service;
|
||||
try {
|
||||
service = new CommandService(settings.setpoints(), registers,
|
||||
new JsonRestoreStore(settings.restoreFile()), audit, clock, System::nanoTime,
|
||||
EARLIEST_PLAUSIBLE, settings.minInterval(),
|
||||
() -> Thread.sleep(settings.readBackDelayMillis()));
|
||||
} catch (CorruptRestoreFileException e) {
|
||||
LOGGER.error("Commands disabled until restart, pending restores were NOT applied: "
|
||||
+ "restore file moved to {}", e.quarantined(), e);
|
||||
return new DisabledCommands("commands disabled: restore file was corrupt (moved to "
|
||||
+ e.quarantined() + ")");
|
||||
}
|
||||
own(new RestoreScheduler(service)).start(settings.restoreCheckPeriod());
|
||||
return service;
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.sun.net.httpserver.HttpExchange;
|
||||
import com.sun.net.httpserver.HttpHandler;
|
||||
import home.gateway.application.Commands;
|
||||
import home.gateway.application.CommandsUnavailableException;
|
||||
import home.gateway.application.Snapshot;
|
||||
import home.gateway.domain.Command;
|
||||
import home.gateway.domain.CommandResult;
|
||||
@@ -26,6 +27,7 @@ import static home.gateway.adapters.http.GatewayHttpServer.respond;
|
||||
* {@code POST /commands} applies one setpoint change, {@code GET /commands} lists the setpoints
|
||||
* and pending restores. Only the backend may call it (S6): every request needs the shared token.
|
||||
* HTTP status follows the outcome: 200 applied, 422 rejected, 502 failed on the controller side.
|
||||
* 503 when the write path is unavailable ({@link CommandsUnavailableException}).
|
||||
* Repeated bad tokens lead to 429 for a while ({@link FailedLoginThrottle}).
|
||||
*/
|
||||
class CommandsEndpoint implements HttpHandler {
|
||||
@@ -74,7 +76,15 @@ class CommandsEndpoint implements HttpHandler {
|
||||
throttle.recordFailure();
|
||||
exchange.getResponseHeaders().set("WWW-Authenticate", "Bearer");
|
||||
respond(exchange, 401, new GatewayHttpServer.Error("unauthorized"));
|
||||
} else if (method.equals("GET")) {
|
||||
} else {
|
||||
dispatch(exchange, method);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void dispatch(HttpExchange exchange, String method) throws IOException {
|
||||
try {
|
||||
if (method.equals("GET")) {
|
||||
respond(exchange, 200, listing());
|
||||
} else if (method.equals("POST")) {
|
||||
post(exchange);
|
||||
@@ -82,6 +92,8 @@ class CommandsEndpoint implements HttpHandler {
|
||||
exchange.getResponseHeaders().set("Allow", "GET, POST");
|
||||
respond(exchange, 405, new GatewayHttpServer.Error("method not allowed"));
|
||||
}
|
||||
} catch (CommandsUnavailableException e) {
|
||||
respond(exchange, 503, new GatewayHttpServer.Error(e.getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
package home.gateway.application;
|
||||
|
||||
/** The write path is not usable right now; the message tells the caller why (HTTP 503). */
|
||||
public class CommandsUnavailableException extends RuntimeException {
|
||||
|
||||
public CommandsUnavailableException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package home.gateway.application;
|
||||
|
||||
import home.gateway.domain.Command;
|
||||
import home.gateway.domain.CommandResult;
|
||||
import home.gateway.domain.Restore;
|
||||
import home.gateway.domain.Setpoint;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/** Stands in for the write path when it cannot be trusted: it never touches the controller. */
|
||||
public record DisabledCommands(String reason) implements Commands {
|
||||
|
||||
@Override
|
||||
public CommandResult apply(Command command) {
|
||||
throw new CommandsUnavailableException(reason);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Setpoint> setpoints() {
|
||||
throw new CommandsUnavailableException(reason);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Restore> pendingRestores() {
|
||||
throw new CommandsUnavailableException(reason);
|
||||
}
|
||||
}
|
||||
@@ -154,6 +154,24 @@ class GatewayEndToEndTest {
|
||||
assertEquals(List.of(), controllerWrites());
|
||||
}
|
||||
|
||||
@Test
|
||||
void aCorruptRestoreFileDisablesCommandsButPollingAndHealthContinue() throws Exception {
|
||||
Files.writeString(dir.resolve("restores.json"), "{not json");
|
||||
gateway = Gateway.start(loadConfig(commandsSection(true)), Clock.systemUTC(), () -> { });
|
||||
waitForFirstPoll();
|
||||
|
||||
assertEquals(200, get("/health").statusCode());
|
||||
for (var response : List.of(post("{\"setpoint\": \"cwu_economy\", \"value\": 45}"), send("GET", null))) {
|
||||
assertEquals(503, response.statusCode(), response.body());
|
||||
assertTrue(response.body().contains("commands disabled: restore file was corrupt (moved to "
|
||||
+ dir.resolve("restores.json.bad-")), response.body());
|
||||
}
|
||||
assertEquals(List.of(), controllerWrites());
|
||||
try (var files = Files.list(dir)) {
|
||||
assertEquals(1, files.filter(file -> file.getFileName().toString().startsWith("restores.json.bad-")).count());
|
||||
}
|
||||
}
|
||||
|
||||
private String commandsSection(boolean enabled) throws IOException {
|
||||
var tokenFile = Files.writeString(dir.resolve("token"), BEARER + "\n");
|
||||
return """
|
||||
|
||||
@@ -3,6 +3,7 @@ package home.gateway.adapters.http;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import home.gateway.application.Commands;
|
||||
import home.gateway.application.DisabledCommands;
|
||||
import home.gateway.application.Snapshot;
|
||||
import home.gateway.domain.Command;
|
||||
import home.gateway.domain.CommandResult;
|
||||
@@ -332,6 +333,21 @@ class CommandsEndpointTest {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void unavailableCommandsAnswer503ToAnAuthorizedCallerOnly() throws Exception {
|
||||
server.close();
|
||||
server = new GatewayHttpServer(snapshot, clock, 0, Set.of());
|
||||
server.serveCommands(new DisabledCommands("commands disabled: test"), BEARER);
|
||||
server.start();
|
||||
|
||||
for (var method : new String[] { "GET", "POST" }) {
|
||||
var response = authorized(method, method.equals("POST") ? example("request-boost.json") : null);
|
||||
assertEquals(503, response.statusCode());
|
||||
assertEquals("commands disabled: test", json.readTree(response.body()).get("error").asText());
|
||||
}
|
||||
assertEquals(401, send("GET", null).statusCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void emptyOrShortTokenIsRefused() throws Exception {
|
||||
for (var token : new String[] { null, "", "short" }) {
|
||||
|
||||
Reference in new issue
Block a user