From 7ff7248e7dbcdd3874ae031b0c67515f84eb766c Mon Sep 17 00:00:00 2001 From: Marc Froehlich Date: Thu, 27 Aug 2026 01:26:50 +0200 Subject: [PATCH] Fixed KST4Contest caused disconnect on getting no activities of ON4KST chatservers if no new lines arriving --- docs/PROJECT_CONTEXT.md | 10 +- .../MessageBusManagementThread.java | 15 +- .../controller/On4KstConnectionManager.java | 170 +++++++++++++++--- .../controller/On4KstProtocol.java | 23 ++- .../controller/On4KstConnectionProbeTest.java | 132 ++++++++++++++ 5 files changed, 321 insertions(+), 29 deletions(-) create mode 100644 src/test/java/kst4contest/controller/On4KstConnectionProbeTest.java diff --git a/docs/PROJECT_CONTEXT.md b/docs/PROJECT_CONTEXT.md index f867d839..d8eb3bf0 100644 --- a/docs/PROJECT_CONTEXT.md +++ b/docs/PROJECT_CONTEXT.md @@ -1,6 +1,6 @@ # KST4Contest Project Context -Last reviewed: 2026-08-25 +Last reviewed: 2026-08-27 This file is the durable technical project context for KST4Contest. It is not a user manual and not a replacement for the changelog. Current code, tests and authoritative external specifications remain the source of truth when this document is stale or ambiguous. @@ -65,6 +65,14 @@ Known integration areas include: CR/LF framing, XML framing, ports/transports, callsign normalization and frequency formatting are protocol behaviour and must not be changed as incidental cleanup. +### ON4KST session liveness + +- After 90 seconds without inbound data, the application keeps the established empty CRLF heartbeat. +- At about 180 seconds of inbound idle time, the TCP session sends one `RDXQ|
|` probe. The probe state belongs to the session, so a two-category login still sends only one probe per idle phase. +- Any subsequent inbound server frame confirms the probe. `DXQ` is accepted as the expected internal response and is not published as chat content. +- If no inbound frame arrives by about 210 seconds, the existing reconnect flow remains responsible for replacing the session. +- Probe diagnostics contain the session id, main category, opcode and timing only. They must not include credentials, complete server frames or normal chat messages. + ## User Workflow / UI Invariants - Contest operating speed and low-friction interaction are primary goals. diff --git a/src/main/java/kst4contest/controller/MessageBusManagementThread.java b/src/main/java/kst4contest/controller/MessageBusManagementThread.java index 41fcc847..6735e926 100644 --- a/src/main/java/kst4contest/controller/MessageBusManagementThread.java +++ b/src/main/java/kst4contest/controller/MessageBusManagementThread.java @@ -877,6 +877,12 @@ public class MessageBusManagementThread extends Thread { || messageToProcess.getMessageText().isEmpty()) { // No processable data. } else { + if (On4KstProtocol.isConnectionProbeResponse( + messageToProcess.getMessageText())) { + // DXQ is the internal response to the active connection probe. + // Liveness was already recorded by the session manager. + return; + } if (messageToProcess.getMessageText().startsWith(SRVR_LOGSTAT + "|")) { String[] logstatMessage = @@ -2177,8 +2183,11 @@ public class MessageBusManagementThread extends Thread { // e.printStackTrace(); // } - System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at - // Console#######################################################TODO:Wichtig + if (!On4KstProtocol.isConnectionProbeResponse( + messageTextRaw.getMessageText())) { + System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at + // Console#######################################################TODO:Wichtig + } try { processRXMessage23001(messageTextRaw); @@ -2207,4 +2216,4 @@ public class MessageBusManagementThread extends Thread { System.out.println("Msgbusmgt: interrupt"); this.interrupt(); } -} \ No newline at end of file +} diff --git a/src/main/java/kst4contest/controller/On4KstConnectionManager.java b/src/main/java/kst4contest/controller/On4KstConnectionManager.java index ae74860a..a60e351a 100644 --- a/src/main/java/kst4contest/controller/On4KstConnectionManager.java +++ b/src/main/java/kst4contest/controller/On4KstConnectionManager.java @@ -48,6 +48,8 @@ final class On4KstConnectionManager { static final long LOGIN_FALLBACK_MILLIS = 2_000L; //Login-Fallback static final long HANDSHAKE_TIMEOUT_MILLIS = 45_000L; //Handshake-Timeout static final long APPLICATION_HEARTBEAT_AFTER_MILLIS = 90_000L; //Application-Heartbeat + /** Idle duration after which the server is asked for current DX data. */ + static final long CONNECTION_PROBE_AFTER_MILLIS = 180_000L; //Active connection probe static final long INBOUND_STALE_AFTER_MILLIS = 210_000L; //Stale-Timeout - time without rxed data static final List RECONNECT_DELAYS_MILLIS = List.of(2_000L, 5_000L, 10_000L, 20_000L, 30_000L); //Reconnect-Backoff if no connection possible @@ -178,7 +180,19 @@ final class On4KstConnectionManager { session.lastInboundMillis.set(now); session.lastProgressMillis.set(now); - String opcode = opcode(line); + String opcode = On4KstProtocol.opcode(line); + long probeResponseMillis = session.connectionProbe.acknowledge(now); + if (probeResponseMillis >= 0L) { + LOGGER.log(Level.INFO, + "ON4KST connection probe confirmed: session {0}, " + + "received opcode {1}, response time {2} ms", + new Object[] { + sessionId, + opcode, + probeResponseMillis + }); + } + if ("CK".equals(opcode)) { sendHeartbeat(session); } @@ -244,7 +258,15 @@ final class On4KstConnectionManager { new LinkedBlockingQueue<>(); LinkedBlockingQueue transmitQueue = new LinkedBlockingQueue<>(); - Session session = new Session(token, socket, receiveQueue, transmitQueue); + int mainCategory = controller.getChatPreferences() + .getLoginChatCategoryMain() + .getCategoryNumber(); + Session session = new Session( + token, + socket, + receiveQueue, + transmitQueue, + mainCategory); ReadThread readThread = new ReadThread( token, socket, receiveQueue, this::isActiveSession, @@ -252,8 +274,7 @@ final class On4KstConnectionManager { failure -> onConnectionFailure(token, failure)); WriteThread writeThread = new WriteThread( token, socket, transmitQueue, - controller.getChatPreferences().getLoginChatCategoryMain() - .getCategoryNumber(), + mainCategory, this::isActiveSession, failure -> onConnectionFailure(token, failure), controller::onOn4KstOutboundFrameRejected); @@ -535,6 +556,29 @@ final class On4KstConnectionManager { session.transmitQueue.offer(heartbeat); } + private void sendConnectionProbe( + Session session, + long now, + long inboundIdle + ) { + if (session == null || !isActiveSession(session.id) + || !session.connectionProbe.tryStart(now)) { + return; + } + + LOGGER.log(Level.INFO, + "Sending ON4KST connection probe: session {0}, main category " + + "{1}, inbound idle {2} seconds", + new Object[] { + session.id, + session.mainCategory, + inboundIdle / 1_000L + }); + sendControl( + session, + On4KstProtocol.connectionProbe(session.mainCategory)); + } + private void onConnectionFailure(long sessionId, Throwable failure) { scheduler.execute(() -> failSession(sessionId, failure)); } @@ -634,18 +678,47 @@ final class On4KstConnectionManager { return; } - long inboundIdle = now - session.lastInboundMillis.get(); - if (inboundIdle > INBOUND_STALE_AFTER_MILLIS) { - failSession(session.id, - new SocketException("No ON4KST data received for " - + inboundIdle / 1_000L + " seconds")); + long lastInboundMillis = session.lastInboundMillis.get(); + long inboundIdle = now - lastInboundMillis; + IdleAction idleAction = determineIdleAction( + inboundIdle, + session.lastHeartbeatMillis.get() >= lastInboundMillis, + session.connectionProbe.isOutstanding()); + + if (session.lastInboundMillis.get() != lastInboundMillis) { return; } - if (inboundIdle > APPLICATION_HEARTBEAT_AFTER_MILLIS - && session.lastHeartbeatMillis.get() - < session.lastInboundMillis.get()) { - sendHeartbeat(session); + switch (idleAction) { + case TIMEOUT -> { + if (session.lastInboundMillis.get() != lastInboundMillis) { + return; + } + long probeWaitMillis = + session.connectionProbe.responseWaitMillis(now); + if (probeWaitMillis >= 0L) { + LOGGER.log(Level.WARNING, + "ON4KST connection probe timed out: session " + + "{0}, main category {1}, no response " + + "for {2} ms, inbound idle {3} seconds; " + + "reconnecting", + new Object[] { + session.id, + session.mainCategory, + probeWaitMillis, + inboundIdle / 1_000L + }); + } + failSession(session.id, + new SocketException("No ON4KST data received for " + + inboundIdle / 1_000L + " seconds")); + } + case CONNECTION_PROBE -> + sendConnectionProbe(session, now, inboundIdle); + case HEARTBEAT -> sendHeartbeat(session); + case NONE -> { + // The session is active or already has the required idle action. + } } } catch (RuntimeException exception) { LOGGER.log(Level.WARNING, @@ -653,6 +726,28 @@ final class On4KstConnectionManager { } } + /** + * Selects at most one maintenance action for the current inbound idle phase. + */ + static IdleAction determineIdleAction( + long inboundIdleMillis, + boolean heartbeatSentForIdlePhase, + boolean probeOutstanding + ) { + if (inboundIdleMillis > INBOUND_STALE_AFTER_MILLIS) { + return IdleAction.TIMEOUT; + } + if (inboundIdleMillis >= CONNECTION_PROBE_AFTER_MILLIS + && !probeOutstanding) { + return IdleAction.CONNECTION_PROBE; + } + if (inboundIdleMillis > APPLICATION_HEARTBEAT_AFTER_MILLIS + && !heartbeatSentForIdlePhase) { + return IdleAction.HEARTBEAT; + } + return IdleAction.NONE; + } + private void validateConfiguration() { ChatPreferences preferences = controller.getChatPreferences(); On4KstProtocol.login( @@ -782,15 +877,6 @@ final class On4KstConnectionManager { } } - private String opcode(String line) { - if (line == null) { - return ""; - } - int separator = line.indexOf('|'); - return (separator < 0 ? line : line.substring(0, separator)) - .trim().toUpperCase(Locale.ROOT); - } - private String describeFailure(Throwable failure) { if (failure == null) { return "unknown error"; @@ -808,12 +894,15 @@ final class On4KstConnectionManager { private final Socket socket; private final LinkedBlockingQueue receiveQueue; private final LinkedBlockingQueue transmitQueue; + private final int mainCategory; private final long connectedMillis = System.currentTimeMillis(); private final AtomicLong lastInboundMillis = new AtomicLong(connectedMillis); private final AtomicLong lastProgressMillis = new AtomicLong(connectedMillis); private final AtomicLong lastHeartbeatMillis = new AtomicLong(); + private final ConnectionProbeState connectionProbe = + new ConnectionProbeState(); private final Map> initialMembers = new ConcurrentHashMap<>(); @@ -831,12 +920,45 @@ final class On4KstConnectionManager { long id, Socket socket, LinkedBlockingQueue receiveQueue, - LinkedBlockingQueue transmitQueue + LinkedBlockingQueue transmitQueue, + int mainCategory ) { this.id = id; this.socket = socket; this.receiveQueue = receiveQueue; this.transmitQueue = transmitQueue; + this.mainCategory = mainCategory; } } -} \ No newline at end of file + + /** Maintenance action selected by the session monitor. */ + enum IdleAction { + NONE, + HEARTBEAT, + CONNECTION_PROBE, + TIMEOUT + } + + /** Tracks one outstanding liveness probe for the complete TCP session. */ + static final class ConnectionProbeState { + private final AtomicLong sentMillis = new AtomicLong(); + + boolean tryStart(long now) { + return now > 0L && sentMillis.compareAndSet(0L, now); + } + + long acknowledge(long now) { + long sent = sentMillis.getAndSet(0L); + return sent == 0L ? -1L : Math.max(0L, now - sent); + } + + boolean isOutstanding() { + return sentMillis.get() > 0L; + } + + long responseWaitMillis(long now) { + long sent = sentMillis.get(); + return sent == 0L ? -1L : Math.max(0L, now - sent); + } + } +} diff --git a/src/main/java/kst4contest/controller/On4KstProtocol.java b/src/main/java/kst4contest/controller/On4KstProtocol.java index 932575c2..99a3ea84 100644 --- a/src/main/java/kst4contest/controller/On4KstProtocol.java +++ b/src/main/java/kst4contest/controller/On4KstProtocol.java @@ -54,6 +54,27 @@ final class On4KstProtocol { + "|0|"; } + /** Builds the active liveness probe for the session's main chat. */ + static String connectionProbe(int category) { + return "RDXQ|" + category(category) + "|"; + } + + /** Returns whether a server frame is the expected liveness-probe response. */ + static boolean isConnectionProbeResponse(String frame) { + return "DXQ".equals(opcode(frame)); + } + + /** Extracts and normalizes the opcode without exposing the remaining frame. */ + static String opcode(String frame) { + if (frame == null) { + return ""; + } + int separator = frame.indexOf('|'); + return (separator < 0 ? frame : frame.substring(0, separator)) + .trim() + .toUpperCase(Locale.ROOT); + } + /** Builds a category-qualified locator command after validating Maidenhead syntax. */ static String setLocator(int category, String locator) { return command(category, "/SETLOC " + locator(locator)); @@ -189,4 +210,4 @@ final class On4KstProtocol { } return category; } -} \ No newline at end of file +} diff --git a/src/test/java/kst4contest/controller/On4KstConnectionProbeTest.java b/src/test/java/kst4contest/controller/On4KstConnectionProbeTest.java new file mode 100644 index 00000000..e067d308 --- /dev/null +++ b/src/test/java/kst4contest/controller/On4KstConnectionProbeTest.java @@ -0,0 +1,132 @@ +package kst4contest.controller; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import kst4contest.model.ChatMessage; + +class On4KstConnectionProbeTest { + + @Test + void buildsMainChatProbeAndAcceptsExpectedResponse() { + assertEquals("RDXQ|2|", On4KstProtocol.connectionProbe(2)); + assertTrue(On4KstProtocol.isConnectionProbeResponse("DXQ|2|data|")); + assertFalse(On4KstProtocol.isConnectionProbeResponse( + "CH|2|123|DL1ABC|Name|0|text|0|")); + } + + @Test + void selectsHeartbeatProbeAndTimeoutAtIdleBoundaries() { + assertEquals( + On4KstConnectionManager.IdleAction.NONE, + idleAction(90_000L, false, false)); + assertEquals( + On4KstConnectionManager.IdleAction.HEARTBEAT, + idleAction(90_001L, false, false)); + assertEquals( + On4KstConnectionManager.IdleAction.NONE, + idleAction(179_999L, true, false)); + assertEquals( + On4KstConnectionManager.IdleAction.CONNECTION_PROBE, + idleAction(180_000L, true, false)); + assertEquals( + On4KstConnectionManager.IdleAction.NONE, + idleAction(210_000L, true, true)); + assertEquals( + On4KstConnectionManager.IdleAction.TIMEOUT, + idleAction(210_001L, true, true)); + } + + @Test + void oneSessionProbeIsAcknowledgedByAnyInboundTraffic() { + On4KstConnectionManager.ConnectionProbeState probe = + new On4KstConnectionManager.ConnectionProbeState(); + + assertTrue(probe.tryStart(1_000L)); + assertFalse(probe.tryStart(1_001L), + "A second category must not start another session probe"); + assertTrue(probe.isOutstanding()); + + assertEquals(250L, probe.acknowledge(1_250L)); + assertFalse(probe.isOutstanding()); + assertEquals(-1L, probe.acknowledge(1_500L)); + + assertTrue(probe.tryStart(2_000L), + "New inbound activity starts a new idle phase"); + } + + @Test + @Timeout(5) + void writerUsesExactCrLfForHeartbeatAndConnectionProbe() throws Exception { + byte[] expected = "\r\nRDXQ|2|\r\n".getBytes(StandardCharsets.UTF_8); + + try (ServerSocket server = new ServerSocket(0)) { + CompletableFuture received = CompletableFuture.supplyAsync(() -> { + try (Socket accepted = server.accept()) { + accepted.setSoTimeout(2_000); + return accepted.getInputStream().readNBytes(expected.length); + } catch (Exception exception) { + throw new RuntimeException(exception); + } + }); + + try (Socket client = new Socket("127.0.0.1", server.getLocalPort())) { + LinkedBlockingQueue queue = + new LinkedBlockingQueue<>(); + AtomicBoolean active = new AtomicBoolean(true); + WriteThread writer = new WriteThread( + 11L, + client, + queue, + 2, + ignored -> active.get(), + ignored -> { }, + ignored -> { }); + writer.start(); + + queue.add(serverFrame("")); + queue.add(serverFrame(On4KstProtocol.connectionProbe(2))); + + assertArrayEquals( + expected, + received.get(2, TimeUnit.SECONDS)); + + active.set(false); + writer.interrupt(); + writer.join(Duration.ofSeconds(2).toMillis()); + } + } + } + + private On4KstConnectionManager.IdleAction idleAction( + long inboundIdleMillis, + boolean heartbeatSent, + boolean probeOutstanding + ) { + return On4KstConnectionManager.determineIdleAction( + inboundIdleMillis, + heartbeatSent, + probeOutstanding); + } + + private ChatMessage serverFrame(String text) { + ChatMessage message = new ChatMessage(); + message.setMessageDirectedToServer(true); + message.setMessageText(text); + return message; + } +}