Fixed KST4Contest caused disconnect on getting no activities of ON4KST chatservers if no new lines arriving

This commit is contained in:
Marc Froehlich
2026-08-27 01:26:50 +02:00
parent 3193e4ac73
commit 7ff7248e7d
5 changed files with 321 additions and 29 deletions
+9 -1
View File
@@ -1,6 +1,6 @@
# KST4Contest Project Context # 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. 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. 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|<main chat id>|` 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 ## User Workflow / UI Invariants
- Contest operating speed and low-friction interaction are primary goals. - Contest operating speed and low-friction interaction are primary goals.
@@ -877,6 +877,12 @@ public class MessageBusManagementThread extends Thread {
|| messageToProcess.getMessageText().isEmpty()) { || messageToProcess.getMessageText().isEmpty()) {
// No processable data. // No processable data.
} else { } 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 + "|")) { if (messageToProcess.getMessageText().startsWith(SRVR_LOGSTAT + "|")) {
String[] logstatMessage = String[] logstatMessage =
@@ -2177,8 +2183,11 @@ public class MessageBusManagementThread extends Thread {
// e.printStackTrace(); // e.printStackTrace();
// } // }
System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at if (!On4KstProtocol.isConnectionProbeResponse(
// Console#######################################################TODO:Wichtig messageTextRaw.getMessageText())) {
System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at
// Console#######################################################TODO:Wichtig
}
try { try {
processRXMessage23001(messageTextRaw); processRXMessage23001(messageTextRaw);
@@ -2207,4 +2216,4 @@ public class MessageBusManagementThread extends Thread {
System.out.println("Msgbusmgt: interrupt"); System.out.println("Msgbusmgt: interrupt");
this.interrupt(); this.interrupt();
} }
} }
@@ -48,6 +48,8 @@ final class On4KstConnectionManager {
static final long LOGIN_FALLBACK_MILLIS = 2_000L; //Login-Fallback static final long LOGIN_FALLBACK_MILLIS = 2_000L; //Login-Fallback
static final long HANDSHAKE_TIMEOUT_MILLIS = 45_000L; //Handshake-Timeout static final long HANDSHAKE_TIMEOUT_MILLIS = 45_000L; //Handshake-Timeout
static final long APPLICATION_HEARTBEAT_AFTER_MILLIS = 90_000L; //Application-Heartbeat 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 long INBOUND_STALE_AFTER_MILLIS = 210_000L; //Stale-Timeout - time without rxed data
static final List<Long> RECONNECT_DELAYS_MILLIS = static final List<Long> RECONNECT_DELAYS_MILLIS =
List.of(2_000L, 5_000L, 10_000L, 20_000L, 30_000L); //Reconnect-Backoff if no connection possible 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.lastInboundMillis.set(now);
session.lastProgressMillis.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)) { if ("CK".equals(opcode)) {
sendHeartbeat(session); sendHeartbeat(session);
} }
@@ -244,7 +258,15 @@ final class On4KstConnectionManager {
new LinkedBlockingQueue<>(); new LinkedBlockingQueue<>();
LinkedBlockingQueue<ChatMessage> transmitQueue = LinkedBlockingQueue<ChatMessage> transmitQueue =
new LinkedBlockingQueue<>(); 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( ReadThread readThread = new ReadThread(
token, socket, receiveQueue, this::isActiveSession, token, socket, receiveQueue, this::isActiveSession,
@@ -252,8 +274,7 @@ final class On4KstConnectionManager {
failure -> onConnectionFailure(token, failure)); failure -> onConnectionFailure(token, failure));
WriteThread writeThread = new WriteThread( WriteThread writeThread = new WriteThread(
token, socket, transmitQueue, token, socket, transmitQueue,
controller.getChatPreferences().getLoginChatCategoryMain() mainCategory,
.getCategoryNumber(),
this::isActiveSession, this::isActiveSession,
failure -> onConnectionFailure(token, failure), failure -> onConnectionFailure(token, failure),
controller::onOn4KstOutboundFrameRejected); controller::onOn4KstOutboundFrameRejected);
@@ -535,6 +556,29 @@ final class On4KstConnectionManager {
session.transmitQueue.offer(heartbeat); 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) { private void onConnectionFailure(long sessionId, Throwable failure) {
scheduler.execute(() -> failSession(sessionId, failure)); scheduler.execute(() -> failSession(sessionId, failure));
} }
@@ -634,18 +678,47 @@ final class On4KstConnectionManager {
return; return;
} }
long inboundIdle = now - session.lastInboundMillis.get(); long lastInboundMillis = session.lastInboundMillis.get();
if (inboundIdle > INBOUND_STALE_AFTER_MILLIS) { long inboundIdle = now - lastInboundMillis;
failSession(session.id, IdleAction idleAction = determineIdleAction(
new SocketException("No ON4KST data received for " inboundIdle,
+ inboundIdle / 1_000L + " seconds")); session.lastHeartbeatMillis.get() >= lastInboundMillis,
session.connectionProbe.isOutstanding());
if (session.lastInboundMillis.get() != lastInboundMillis) {
return; return;
} }
if (inboundIdle > APPLICATION_HEARTBEAT_AFTER_MILLIS switch (idleAction) {
&& session.lastHeartbeatMillis.get() case TIMEOUT -> {
< session.lastInboundMillis.get()) { if (session.lastInboundMillis.get() != lastInboundMillis) {
sendHeartbeat(session); 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) { } catch (RuntimeException exception) {
LOGGER.log(Level.WARNING, 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() { private void validateConfiguration() {
ChatPreferences preferences = controller.getChatPreferences(); ChatPreferences preferences = controller.getChatPreferences();
On4KstProtocol.login( 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) { private String describeFailure(Throwable failure) {
if (failure == null) { if (failure == null) {
return "unknown error"; return "unknown error";
@@ -808,12 +894,15 @@ final class On4KstConnectionManager {
private final Socket socket; private final Socket socket;
private final LinkedBlockingQueue<ChatMessage> receiveQueue; private final LinkedBlockingQueue<ChatMessage> receiveQueue;
private final LinkedBlockingQueue<ChatMessage> transmitQueue; private final LinkedBlockingQueue<ChatMessage> transmitQueue;
private final int mainCategory;
private final long connectedMillis = System.currentTimeMillis(); private final long connectedMillis = System.currentTimeMillis();
private final AtomicLong lastInboundMillis = private final AtomicLong lastInboundMillis =
new AtomicLong(connectedMillis); new AtomicLong(connectedMillis);
private final AtomicLong lastProgressMillis = private final AtomicLong lastProgressMillis =
new AtomicLong(connectedMillis); new AtomicLong(connectedMillis);
private final AtomicLong lastHeartbeatMillis = new AtomicLong(); private final AtomicLong lastHeartbeatMillis = new AtomicLong();
private final ConnectionProbeState connectionProbe =
new ConnectionProbeState();
private final Map<Integer, Map<String, ChatMember>> initialMembers = private final Map<Integer, Map<String, ChatMember>> initialMembers =
new ConcurrentHashMap<>(); new ConcurrentHashMap<>();
@@ -831,12 +920,45 @@ final class On4KstConnectionManager {
long id, long id,
Socket socket, Socket socket,
LinkedBlockingQueue<ChatMessage> receiveQueue, LinkedBlockingQueue<ChatMessage> receiveQueue,
LinkedBlockingQueue<ChatMessage> transmitQueue LinkedBlockingQueue<ChatMessage> transmitQueue,
int mainCategory
) { ) {
this.id = id; this.id = id;
this.socket = socket; this.socket = socket;
this.receiveQueue = receiveQueue; this.receiveQueue = receiveQueue;
this.transmitQueue = transmitQueue; this.transmitQueue = transmitQueue;
this.mainCategory = mainCategory;
} }
} }
}
/** 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);
}
}
}
@@ -54,6 +54,27 @@ final class On4KstProtocol {
+ "|0|"; + "|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. */ /** Builds a category-qualified locator command after validating Maidenhead syntax. */
static String setLocator(int category, String locator) { static String setLocator(int category, String locator) {
return command(category, "/SETLOC " + locator(locator)); return command(category, "/SETLOC " + locator(locator));
@@ -189,4 +210,4 @@ final class On4KstProtocol {
} }
return category; return category;
} }
} }
@@ -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<byte[]> 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<ChatMessage> 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;
}
}