Merge PR #64: fix(modbus): close an idle connection, so no poll meets a reset socket

On the Pi every other heat pump poll failed with 'Connection reset by
peer': MK01 drops a connection left idle for minutes, and the gateway
kept one open between polls (300 s). The socket now closes after 5 idle
seconds and the next poll connects anew, which also frees the single
MK01 connection for the vendor's app between polls.

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

gateway/src/main/java/home/gateway/adapters/modbus/TcpTransport.java

@@ -7,11 +7,16 @@ import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * Modbus TCP (MBAP) over one long-lived socket. The MK01 gateway serves a single client and a
- * second connection drops the first, so the socket is reused and requests never overlap.
+ * Modbus TCP (MBAP). The MK01 gateway serves a single client and a second connection drops the
+ * first, so requests never overlap and the socket is reused for the requests of one poll. It is
+ * closed after a few idle seconds: MK01 resets a connection left idle for minutes, which cost every
+ * other poll, and while one is open the vendor's app cannot connect.
  */
 public final class TcpTransport implements ModbusTransport {
 
@@ -20,17 +25,26 @@ public final class TcpTransport implements ModbusTransport {
     }
 
     private static final int HEADER_LENGTH = 7;
+    static final Duration IDLE_CLOSE = Duration.ofSeconds(5);
 
     // Not synchronized: a virtual thread blocked in a monitor pins its carrier thread.
     private final ReentrantLock lock = new ReentrantLock();
     private final Settings settings;
+    private final Duration idleClose;
+    private final ScheduledExecutorService idleCloser = Executors.newSingleThreadScheduledExecutor(
+            Thread.ofVirtual().name("modbus-idle-close").factory());
     private Socket socket;
     private DataInputStream in;
     private int transactionId;
     private long lastResponseNanos;
 
     public TcpTransport(Settings settings) {
+        this(settings, IDLE_CLOSE);
+    }
+
+    TcpTransport(Settings settings, Duration idleClose) {
         this.settings = settings;
+        this.idleClose = idleClose;
         this.lastResponseNanos = System.nanoTime() - settings.minimumPause().toNanos();
     }
 
@@ -52,6 +66,9 @@ public final class TcpTransport implements ModbusTransport {
             socket.getOutputStream().write(frame(id, unitId, requestPdu));
             byte[] response = readResponse(id, unitId);
             lastResponseNanos = System.nanoTime();
+            if (!idleCloser.isShutdown()) {
+                idleCloser.schedule(this::closeIfIdle, idleClose.toNanos(), TimeUnit.NANOSECONDS);
+            }
             return response;
         } catch (IOException e) {
             closeSocket();
@@ -63,12 +80,25 @@ public final class TcpTransport implements ModbusTransport {
     public void close() {
         lock.lock();
         try {
+            idleCloser.shutdownNow();
             closeSocket();
         } finally {
             lock.unlock();
         }
     }
 
+    /** A later request in the same poll moved lastResponseNanos on, and the socket stays. */
+    private void closeIfIdle() {
+        lock.lock();
+        try {
+            if (System.nanoTime() - lastResponseNanos >= idleClose.toNanos()) {
+                closeSocket();
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
     private void awaitPause() throws InterruptedIOException {
         long remaining = settings.minimumPause().toNanos() - (System.nanoTime() - lastResponseNanos);
         if (remaining <= 0) {

gateway/src/test/java/home/gateway/adapters/modbus/MockModbusServer.java

@@ -21,6 +21,7 @@ public final class MockModbusServer implements Closeable {
     private final Map<Integer, Integer> holding = new ConcurrentHashMap<>();
     private final Map<Integer, Integer> input = new ConcurrentHashMap<>();
     private final AtomicInteger connections = new AtomicInteger();
+    private final AtomicInteger open = new AtomicInteger();
     private final List<int[]> writes = new CopyOnWriteArrayList<>();
     private volatile Duration responseDelay = Duration.ZERO;
     private volatile boolean dropNextRequest;
@@ -42,6 +43,11 @@ public final class MockModbusServer implements Closeable {
         return connections.get();
     }
 
+    /** Connections the client has not closed yet. */
+    public int openConnections() {
+        return open.get();
+    }
+
     public void holding(int address, int value) {
         holding.put(address, value);
     }
@@ -94,6 +100,7 @@ public final class MockModbusServer implements Closeable {
             try {
                 Socket client = server.accept();
                 connections.incrementAndGet();
+                open.incrementAndGet();
                 Thread.ofVirtual().start(() -> serve(client));
             } catch (IOException e) {
                 return;
@@ -128,6 +135,8 @@ public final class MockModbusServer implements Closeable {
             }
         } catch (IOException ignored) {
             // the client went away, which is a normal end of a test
+        } finally {
+            open.decrementAndGet();
         }
     }
 

gateway/src/test/java/home/gateway/adapters/modbus/TcpTransportTest.java

@@ -61,6 +61,28 @@ class TcpTransportTest {
         assertEquals(1, server.connections());
     }
 
+    @Test
+    void closesTheConnectionWhenIdleAndReconnectsForTheNextPoll() throws Exception {
+        transport.close();
+        transport = new TcpTransport(new TcpTransport.Settings(
+                "localhost", server.port(), Duration.ofSeconds(2), Duration.ofSeconds(2), Duration.ZERO),
+                Duration.ofMillis(100));
+
+        transport.exchange(1, READ_HOLDING_10);
+        transport.exchange(1, READ_HOLDING_10);
+        assertEquals(1, server.openConnections());
+
+        // Idle: the controller is free for the vendor's app, and no socket goes stale.
+        long deadline = System.nanoTime() + Duration.ofSeconds(3).toNanos();
+        while (server.openConnections() > 0 && System.nanoTime() < deadline) {
+            Thread.sleep(20);
+        }
+        assertEquals(0, server.openConnections());
+
+        assertArrayEquals(HOLDING_10_RESPONSE, transport.exchange(1, READ_HOLDING_10));
+        assertEquals(2, server.connections());
+    }
+
     @Test
     void reconnectsAfterServerDropsConnection() throws IOException {
         transport.exchange(1, READ_HOLDING_10);