fix(network): prevent CoT TLS read-buffer stalls

src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java

@@ -245,13 +245,41 @@ protected int readBuffer(ByteBuffer applicationIn) throws IOException {
 
         if (result.getStatus() == Status.BUFFER_UNDERFLOW) {
           encryptedIn.compact();
+          if (!encryptedIn.hasRemaining()) {
+            // The inbound buffer is full yet the engine still needs more bytes to
+            // decode a record: we can neither read more (no space) nor make
+            // progress. Left alone this spins the reader at 100% CPU and the
+            // socket never drains. Tear the endpoint down instead - a reconnect
+            // re-handshakes with a fresh buffer.
+            throw new IOException("SSL inbound buffer full on BUFFER_UNDERFLOW; record exceeds "
+                + encryptedIn.capacity() + " byte buffer, closing endpoint");
+          }
           return response;
         }
 
+        if (result.getStatus() == Status.BUFFER_OVERFLOW) {
+          if (applicationIn.position() > 0) {
+            break; // hand back what has been decoded; the caller drains it and reads again
+          }
+          // Not even one record fits in an empty buffer, so it can never be decoded: nothing
+          // leaves encryptedIn, the socket stops draining and the peer stalls. Fail loudly.
+          throw new IOException("SSL read buffer of " + applicationIn.capacity()
+              + " bytes cannot hold a TLS record of up to "
+              + sslEngine.getSession().getApplicationBufferSize() + " bytes, closing endpoint");
+        }
+
         SSLEngineResult.HandshakeStatus hs = result.getHandshakeStatus();
-        if (hs == SSLEngineResult.HandshakeStatus.NEED_TASK
-            || hs == SSLEngineResult.HandshakeStatus.NEED_WRAP) {
-          break; // let handshake manager drive next step
+        if (hs == SSLEngineResult.HandshakeStatus.NEED_TASK) {
+          // After the handshake the manager is SSLHandshakeManagerFinished, which drives
+          // nothing, so a task requested later (e.g. by a TLS 1.3 key update) would never
+          // run. Run it here and retry the unwrap.
+          if (!runDelegatedTasks()) {
+            break; // nothing to run, do not spin on an unchanged status
+          }
+          continue;
+        }
+        if (hs == SSLEngineResult.HandshakeStatus.NEED_WRAP) {
+          break; // needs an outbound record, driven by the write path
         }
 
         boolean noProgress = encryptedIn.position() == beforeIn && applicationIn.position() == beforeOut;
@@ -269,6 +297,21 @@ protected int readBuffer(ByteBuffer applicationIn) throws IOException {
     return response;
   }
 
+  /**
+   * Runs any tasks the engine has delegated, as the handshake manager does during negotiation.
+   *
+   * @return true if at least one task was run, false if the engine had none pending
+   */
+  private boolean runDelegatedTasks() {
+    Runnable runnable;
+    boolean ranTask = false;
+    while ((runnable = sslEngine.getDelegatedTask()) != null) {
+      runnable.run();
+      ranTask = true;
+    }
+    return ranTask;
+  }
+
   private SSLEngineResult handleSSLEngineResult(SSLEngineResult result) throws IOException {
     if (result.getStatus() == Status.CLOSED) {
       throw new IOException("Session closed");

src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java

@@ -88,6 +88,11 @@ public class CotProtocol extends Protocol implements Selectable {
 
   private static final String DESTINATION_NAME = "/tak/cot/inbound";
   private static final int MAX_BUFFER_SIZE = 1_048_576; // 1MB safety cap against a malformed/never-terminated stream
+  // A TLS record is indivisible and decrypts to up to 16 KB, so every read must have room for a
+  // whole record. The packet handed over by protocol detection is only 1 KB and must not be reused
+  // for reads: over ssl:// any record carrying more than 1 KB of CoT would never decode.
+  private static final int MIN_READ_BUFFER_SIZE = 32 * 1024;
+  private static final String READER_THREAD_PREFIX = "CoT Protocol Reader::";
   private static final String UPDATE_SOURCE = "cot-ingest";
 
   private static final byte[] EVENT_START = "<event".getBytes(StandardCharsets.US_ASCII);
@@ -117,16 +122,25 @@ public class CotProtocol extends Protocol implements Selectable {
 
   public CotProtocol(EndPoint endPoint, Packet initialPacket, CotProtocolConfigDTO config) throws IOException {
     super(endPoint, config != null ? config : new CotProtocolConfigDTO());
-    this.packet = initialPacket;
+    this.packet = new Packet(readBufferSize(endPoint), false);
     this.sessionId = "cot-" + UUID.randomUUID();
     createSession();
     logger.debug("CoT passthrough connection created on {}", endPoint.getConfig().getUrl());
-    if (initialPacket.available() > 0) {
+    if (initialPacket != null && initialPacket.available() > 0) {
       appendAndProcess(initialPacket);
     }
     endPoint.register(SelectionKey.OP_READ, this);
   }
 
+  /** The configured server read buffer size, never smaller than one TLS record and never above the 1 MB cap. */
+  static int readBufferSize(EndPoint endPoint) {
+    long configured = 0;
+    if (endPoint.getConfig() != null && endPoint.getConfig().getEndPointConfig() != null) {
+      configured = endPoint.getConfig().getEndPointConfig().getServerReadBufferSize();
+    }
+    return (int) Math.max(MIN_READ_BUFFER_SIZE, Math.min(configured, MAX_BUFFER_SIZE));
+  }
+
   private void createSession() throws IOException {
     try {
       SessionContextBuilder scb = new SessionContextBuilder(sessionId, new ProtocolClientConnection(this));
@@ -168,20 +182,41 @@ public void selected(Selectable selectable, Selector selector, int selection) {
   private final class ReadTask implements Runnable {
     @Override
     public void run() {
-      Thread.currentThread().setName("CoT Protocol Reader::" + Thread.currentThread().getName());
+      // The pool reuses threads, so prefix for this run only - prefixing the current name every
+      // time grew it without bound over the life of the server.
+      Thread current = Thread.currentThread();
+      String poolName = current.getName();
+      current.setName(READER_THREAD_PREFIX + poolName);
+      try {
+        readAvailable();
+      } finally {
+        current.setName(poolName);
+      }
+    }
+
+    private void readAvailable() {
       try {
         int read;
         boolean first = true;
         do {
           packet.clear();
           read = endPoint.readPacket(packet);
-          if (read > 0) {
+          int applicationBytes = packet.position();
+          if (read > 0 && applicationBytes > 0) {
             EndPoint.totalReceived.increment();
             packet.flip();
             appendAndProcess(packet);
-          } else if (first) {
+          } else if (first && read <= 0) {
+            // Nothing readable on the first pass: the peer has gone away.
             closeQuietly();
             return;
+          } else {
+            // read > 0 but no application data was produced - e.g. an SSL read
+            // consumed encrypted bytes for a partial record and decoded nothing
+            // yet (readPacket then reports the encrypted count, not application
+            // payload). Stop draining and wait for the selector to signal fresh
+            // data instead of busy-spinning on a count that carries no payload.
+            break;
           }
           first = false;
         } while (read > 0);