From 070af0b9a70fec166dc9f55053bf3585daded614 Mon Sep 17 00:00:00 2001 From: Ralf Himmelein Date: Tue, 6 Oct 2026 20:38:46 +0100 Subject: [PATCH 1/3] fix(network): give CoT a read buffer that can hold a TLS record CotProtocol kept the 1 KB packet that protocol detection hands over and read into it for the life of the connection. Over ssl:// a TLS record is indivisible and decrypts to up to 16 KB, so the first record carrying more than 1 KB of CoT made SSLEngine.unwrap return BUFFER_OVERFLOW with nothing consumed. readBuffer did not handle BUFFER_OVERFLOW: the record stayed in encryptedIn, the buffer filled, socket reads returned 0 and the receive queue stopped draining while the reader looped on the buffered byte count. Small records decode, so the feed ran for minutes and then stalled for good on every connection. CotProtocol now reads into its own buffer sized from serverReadBufferSize, never below 32 KB and capped at 1 MB. readBuffer hands decoded data back on BUFFER_OVERFLOW so the caller can drain, and closes the endpoint when an empty buffer cannot hold a record instead of hanging. Also: - the reader stops looping on a read that decodes no payload and waits on the selector instead - the endpoint is closed when the inbound buffer is full on BUFFER_UNDERFLOW rather than spinning - delegated engine tasks requested after the handshake are run, since the finished handshake manager no longer runs them - the pooled reader thread is no longer renamed with an extra prefix on every run Observed on central's IHMC CoT listener: receive queue pinned at ~205 KB minutes after each connect, with the CoT reader burning CPU. NO-ISSUE Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 3332f2844b4591b133bb1acc8deefc35f700907a) --- .../network/io/impl/ssl/SSLEndPoint.java | 49 ++++- .../protocol/impl/cot/CotProtocol.java | 45 ++++- .../impl/ssl/SSLEndPointReadOverflowTest.java | 167 ++++++++++++++++++ .../protocol/impl/cot/CotProtocolTest.java | 139 +++++++++++++++ 4 files changed, 392 insertions(+), 8 deletions(-) create mode 100644 src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java diff --git a/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java b/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java index f1c23e6e5f..107bd0ba38 100644 --- a/src/main/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPoint.java +++ b/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"); diff --git a/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java b/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java index 65ef1a99b7..c7252db982 100644 --- a/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java +++ b/src/main/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocol.java @@ -74,6 +74,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 = " 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)); @@ -137,20 +151,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); diff --git a/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java new file mode 100644 index 0000000000..907057bb99 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java @@ -0,0 +1,167 @@ +/* + * + * Copyright [ 2026 ] Ralf Himmelein and Claude + * + * Licensed under the Apache License, Version 2.0 with the Commons Clause + * (the "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * https://commonsclause.com/ + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.mapsmessaging.network.io.impl.ssl; + +import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; +import io.mapsmessaging.dto.rest.config.network.impl.TcpConfigDTO; +import io.mapsmessaging.network.io.EndPointServerStatus; +import io.mapsmessaging.network.io.impl.Selector; +import org.junit.jupiter.api.Test; + +import javax.net.ssl.SSLEngine; +import javax.net.ssl.SSLEngineResult; +import javax.net.ssl.SSLSession; +import java.io.IOException; +import java.net.InetAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.*; + +/** + * Inbound TLS records larger than the caller's read buffer. The engine mock answers exactly as the + * JDK's TLSv1.3 engine does (checked against a real engine): a record that does not fit in the + * destination returns BUFFER_OVERFLOW with nothing consumed or produced. + */ +class SSLEndPointReadOverflowTest { + + private static final int PACKET_BUFFER_SIZE = 16709; + private static final int APPLICATION_BUFFER_SIZE = 16704; + private static final int RECORD_ENCRYPTED_BYTES = 5038; + private static final int RECORD_PLAINTEXT_BYTES = 5000; + + @Test + void readBuffer_whenRecordCannotFitEmptyBuffer_closesInsteadOfHanging() throws Exception { + SSLEndPoint endPoint = new Fixture().createEndPoint(); + try { + ByteBuffer detectionSized = ByteBuffer.allocate(1024); + + IOException failure = assertThrows(IOException.class, () -> endPoint.readBuffer(detectionSized)); + + assertTrue(failure.getMessage().contains("cannot hold a TLS record")); + assertEquals(0, detectionSized.position()); + } finally { + endPoint.close(); + } + } + + @Test + void readBuffer_whenBufferHoldsTheRecord_decodesIt() throws Exception { + SSLEndPoint endPoint = new Fixture().createEndPoint(); + try { + ByteBuffer applicationIn = ByteBuffer.allocate(100_000); + + endPoint.readBuffer(applicationIn); + + assertEquals(RECORD_PLAINTEXT_BYTES, applicationIn.position()); + } finally { + endPoint.close(); + } + } + + @Test + void readBuffer_whenPartlyFilledBufferOverflows_returnsDecodedDataForTheCallerToDrain() throws Exception { + SSLEndPoint endPoint = new Fixture().createEndPoint(); + try { + ByteBuffer applicationIn = ByteBuffer.allocate(3000); + applicationIn.put(new byte[100]); + + assertDoesNotThrow(() -> endPoint.readBuffer(applicationIn)); + + assertEquals(100, applicationIn.position()); + } finally { + endPoint.close(); + } + } + + private static final class Fixture { + + private final SocketChannel channel = mock(SocketChannel.class); + private final Socket socket = mock(Socket.class); + private final Selector selector = mock(Selector.class); + private final SSLEngine sslEngine = mock(SSLEngine.class); + private final SSLSession sslSession = mock(SSLSession.class); + private final EndPointServerStatus serverStatus = mock(EndPointServerStatus.class); + private final AtomicBoolean recordDelivered = new AtomicBoolean(); + + Fixture() throws Exception { + EndPointServerConfigDTO serverConfig = new EndPointServerConfigDTO(); + TcpConfigDTO tcpConfig = new TcpConfigDTO(); + tcpConfig.setTimeout(60_000); + serverConfig.setEndPointConfig(tcpConfig); + + when(serverStatus.getConfig()).thenReturn(serverConfig); + when(channel.socket()).thenReturn(socket); + when(channel.isConnected()).thenReturn(true); + when(socket.getLocalAddress()).thenReturn(InetAddress.getLoopbackAddress()); + when(sslEngine.getSession()).thenReturn(sslSession); + when(sslSession.getPacketBufferSize()).thenReturn(PACKET_BUFFER_SIZE); + when(sslSession.getApplicationBufferSize()).thenReturn(APPLICATION_BUFFER_SIZE); + when(sslEngine.getHandshakeStatus()).thenReturn(SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING); + + when(sslEngine.wrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn(new SSLEngineResult( + SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0)); + when(channel.write(any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer encrypted = invocation.getArgument(0); + int remaining = encrypted.remaining(); + encrypted.position(encrypted.limit()); + return remaining; + }); + + // The socket holds one encrypted record, delivered once. + when(channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + if (recordDelivered.getAndSet(true)) { + return 0; + } + ByteBuffer encryptedIn = invocation.getArgument(0); + encryptedIn.put(new byte[RECORD_ENCRYPTED_BYTES]); + return RECORD_ENCRYPTED_BYTES; + }); + + when(sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer encryptedIn = invocation.getArgument(0); + ByteBuffer applicationIn = invocation.getArgument(1); + if (encryptedIn.remaining() < RECORD_ENCRYPTED_BYTES) { + return new SSLEngineResult(SSLEngineResult.Status.BUFFER_UNDERFLOW, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0); + } + if (applicationIn.remaining() < RECORD_PLAINTEXT_BYTES) { + return new SSLEngineResult(SSLEngineResult.Status.BUFFER_OVERFLOW, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0); + } + encryptedIn.position(encryptedIn.position() + RECORD_ENCRYPTED_BYTES); + applicationIn.put(new byte[RECORD_PLAINTEXT_BYTES]); + return new SSLEngineResult(SSLEngineResult.Status.OK, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, RECORD_ENCRYPTED_BYTES, RECORD_PLAINTEXT_BYTES); + }); + + doAnswer(invocation -> null).when(selector).register(any(SocketChannel.class), anyInt(), any()); + } + + SSLEndPoint createEndPoint() throws Exception { + return new SSLEndPoint(1, sslEngine, channel, selector, endPoint -> { }, serverStatus, List.of()); + } + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java index 07dc4aae79..c325c979f1 100644 --- a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java @@ -3,6 +3,7 @@ import io.mapsmessaging.api.MessageEvent; import io.mapsmessaging.api.Session; import io.mapsmessaging.api.SessionManager; +import io.mapsmessaging.dto.rest.config.network.EndPointConfigDTO; import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; import io.mapsmessaging.dto.rest.config.protocol.impl.CotProtocolConfigDTO; import io.mapsmessaging.engine.session.security.SecurityContext; @@ -10,13 +11,17 @@ import io.mapsmessaging.network.io.Packet; import io.mapsmessaging.utilities.admin.JMXManager; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.mockito.MockedStatic; import javax.security.auth.Subject; +import java.nio.channels.SelectionKey; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.*; class CotProtocolTest { @@ -62,6 +67,110 @@ void protocolIdentityInboundContractAndCloseUseInternalSession() throws Exceptio } } + @Test + @Timeout(10) + void readTask_whenReadReportsBytesButDecodesNoPayload_parksInsteadOfSpinning() throws Exception { + boolean original = JMXManager.isEnableJMX(); + JMXManager.setEnableJMX(false); + SessionManager manager = mock(SessionManager.class); + Session session = mock(Session.class); + SecurityContext security = mock(SecurityContext.class); + Subject subject = new Subject(); + when(session.getSecurityContext()).thenReturn(security); + when(security.getSubject()).thenReturn(subject); + when(manager.create(any(), any())).thenReturn(session); + + try (MockedStatic mocked = mockStatic(SessionManager.class)) { + mocked.when(SessionManager::getInstance).thenReturn(manager); + EndPoint endpoint = endpoint(); + // Simulate the wedged SSL case: readPacket reports a positive (encrypted) + // count but leaves the application packet empty - no decoded payload. With + // the old while(read > 0) loop this spun the reader at 100% CPU forever. + when(endpoint.readPacket(any())).thenReturn(16708); + + CotProtocol protocol = + new CotProtocol(endpoint, new Packet(0, false), new CotProtocolConfigDTO()); + // The constructor registers OP_READ once; clear that interaction first. + verify(endpoint).register(eq(SelectionKey.OP_READ), eq(protocol)); + clearInvocations(endpoint); + + protocol.selected(protocol, null, SelectionKey.OP_READ); + + // The reader must stop after a bounded number of reads and re-arm the + // selector, rather than looping on the empty positive count. + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + verify(endpoint, atMost(5)).readPacket(any()); + verify(endpoint, never()).close(); + + protocol.close(); + } finally { + JMXManager.setEnableJMX(original); + } + } + + @Test + @Timeout(10) + void readTask_readsIntoBufferThatHoldsATlsRecord_notTheDetectionPacket() throws Exception { + withSession(endpoint -> { + List readTargets = new CopyOnWriteArrayList<>(); + when(endpoint.readPacket(any())).thenAnswer(invocation -> { + readTargets.add(invocation.getArgument(0)); + return 16708; + }); + // Protocol detection hands over its own 1 KB buffer - over ssl:// a TLS record decrypts + // to up to 16 KB and can never be unwrapped into it. + Packet detectionPacket = new Packet(1024, true); + CotProtocol protocol = new CotProtocol(endpoint, detectionPacket, new CotProtocolConfigDTO()); + clearInvocations(endpoint); + + protocol.selected(protocol, null, SelectionKey.OP_READ); + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + + assertFalse(readTargets.isEmpty()); + for (Packet target : readTargets) { + assertNotSame(detectionPacket, target); + assertTrue(target.getRawBuffer().capacity() >= 16704, + "read buffer of " + target.getRawBuffer().capacity() + " bytes cannot hold a TLS record"); + } + protocol.close(); + }); + } + + @Test + void readBufferSize_usesConfiguredSize_withinTlsRecordFloorAndSafetyCap() { + assertEquals(100_000, CotProtocol.readBufferSize(endpointWithReadBufferSize(100_000))); + assertEquals(32 * 1024, CotProtocol.readBufferSize(endpointWithReadBufferSize(10_240))); + assertEquals(32 * 1024, CotProtocol.readBufferSize(endpointWithReadBufferSize(1024))); + assertEquals(1_048_576, CotProtocol.readBufferSize(endpointWithReadBufferSize(4_000_000))); + assertEquals(32 * 1024, CotProtocol.readBufferSize(endpoint())); + } + + @Test + @Timeout(10) + void readTask_doesNotGrowThePooledThreadNameAcrossRuns() throws Exception { + withSession(endpoint -> { + List readerNames = new CopyOnWriteArrayList<>(); + when(endpoint.readPacket(any())).thenAnswer(invocation -> { + readerNames.add(Thread.currentThread().getName()); + return 16708; + }); + CotProtocol protocol = new CotProtocol(endpoint, new Packet(0, false), new CotProtocolConfigDTO()); + + for (int run = 0; run < 3; run++) { + clearInvocations(endpoint); + protocol.selected(protocol, null, SelectionKey.OP_READ); + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + } + + assertEquals(3, readerNames.size()); + for (String name : readerNames) { + assertTrue(name.startsWith("CoT Protocol Reader::"), name); + assertFalse(name.startsWith("CoT Protocol Reader::CoT Protocol Reader::"), name); + } + protocol.close(); + }); + } + @Test void staticFrameSearchHelpersHandleBoundsAndWhitespace() throws Exception { byte[] data = "xx 1tail".getBytes(); @@ -82,6 +191,36 @@ void staticFrameSearchHelpersHandleBoundsAndWhitespace() throws Exception { assertEquals(-1, index.invoke(null, data, "missing".getBytes(), 0)); } + @FunctionalInterface + private interface SessionBody { + void run(EndPoint endpoint) throws Exception; + } + + private static void withSession(SessionBody body) throws Exception { + boolean original = JMXManager.isEnableJMX(); + JMXManager.setEnableJMX(false); + SessionManager manager = mock(SessionManager.class); + Session session = mock(Session.class); + SecurityContext security = mock(SecurityContext.class); + when(session.getSecurityContext()).thenReturn(security); + when(security.getSubject()).thenReturn(new Subject()); + when(manager.create(any(), any())).thenReturn(session); + try (MockedStatic mocked = mockStatic(SessionManager.class)) { + mocked.when(SessionManager::getInstance).thenReturn(manager); + body.run(endpoint()); + } finally { + JMXManager.setEnableJMX(original); + } + } + + private static EndPoint endpointWithReadBufferSize(long size) { + EndPoint endpoint = endpoint(); + EndPointConfigDTO endPointConfig = mock(EndPointConfigDTO.class); + when(endPointConfig.getServerReadBufferSize()).thenReturn(size); + when(endpoint.getConfig().getEndPointConfig()).thenReturn(endPointConfig); + return endpoint; + } + private static EndPoint endpoint() { EndPoint endpoint = mock(EndPoint.class); EndPointServerConfigDTO config = mock(EndPointServerConfigDTO.class); From f88fe1659f7ee98334d158e39ae2af7e285fc8a4 Mon Sep 17 00:00:00 2001 From: Matthew Buckton Date: Thu, 15 Oct 2026 15:32:03 +1100 Subject: [PATCH 2/3] test(network): cover CoT TLS record draining and recovery Exercise real TLS 1.2 and TLS 1.3 records, buffered overflow retries, fragmented input, delegated tasks, full-buffer underflow and CoT detection handoff. Retain existing TLS handshake regressions. Refs: MSG-415 --- .../impl/ssl/SSLEndPointReadOverflowTest.java | 126 +++++++++ .../io/impl/ssl/SSLEndPointTlsRecordTest.java | 265 ++++++++++++++++++ .../protocol/impl/cot/CotProtocolTest.java | 54 +++- 3 files changed, 441 insertions(+), 4 deletions(-) create mode 100644 src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java diff --git a/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java index 907057bb99..a80d0e13e0 100644 --- a/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java +++ b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointReadOverflowTest.java @@ -34,6 +34,7 @@ import java.nio.channels.SocketChannel; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.any; @@ -96,6 +97,131 @@ void readBuffer_whenPartlyFilledBufferOverflows_returnsDecodedDataForTheCallerTo } } + @Test + void overflowing_second_record_is_preserved_until_the_caller_drains_the_first() throws Exception { + Fixture fixture = new Fixture(); + AtomicBoolean delivered = new AtomicBoolean(); + when(fixture.channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + if (delivered.getAndSet(true)) { + return 0; + } + ByteBuffer encrypted = invocation.getArgument(0); + encrypted.put(new byte[2 * RECORD_ENCRYPTED_BYTES]); + return 2 * RECORD_ENCRYPTED_BYTES; + }); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(6000); + endPoint.readBuffer(application); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertTrue(endPoint.hasBufferedReadData()); + + application.clear(); + endPoint.readBuffer(application); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertFalse(endPoint.hasBufferedReadData()); + } finally { + endPoint.close(); + } + } + + @Test + void fragmented_record_is_retained_until_the_remaining_encrypted_bytes_arrive() throws Exception { + Fixture fixture = new Fixture(); + AtomicInteger reads = new AtomicInteger(); + when(fixture.channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + int count = switch (reads.getAndIncrement()) { + case 0 -> 2000; + case 1 -> RECORD_ENCRYPTED_BYTES - 2000; + default -> 0; + }; + ByteBuffer encrypted = invocation.getArgument(0); + encrypted.put(new byte[count]); + return count; + }); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(32768); + endPoint.readBuffer(application); + assertEquals(0, application.position()); + assertTrue(endPoint.hasBufferedReadData()); + + endPoint.readBuffer(application); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertFalse(endPoint.hasBufferedReadData()); + } finally { + endPoint.close(); + } + } + + @Test + void full_encrypted_buffer_underflow_fails_instead_of_spinning() throws Exception { + Fixture fixture = new Fixture(); + when(fixture.channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer encrypted = invocation.getArgument(0); + int count = encrypted.remaining(); + encrypted.put(new byte[count]); + return count; + }); + when(fixture.sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn( + new SSLEngineResult(SSLEngineResult.Status.BUFFER_UNDERFLOW, + SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, 0, 0)); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + IOException failure = assertThrows(IOException.class, () -> endPoint.readBuffer(ByteBuffer.allocate(32768))); + assertTrue(failure.getMessage().contains("inbound buffer full")); + verify(fixture.sslEngine, times(1)).unwrap(any(ByteBuffer.class), any(ByteBuffer.class)); + } finally { + endPoint.close(); + } + } + + @Test + void delegated_task_runs_before_retrying_the_buffered_record() throws Exception { + Fixture fixture = new Fixture(); + AtomicBoolean taskRan = new AtomicBoolean(); + Runnable task = () -> taskRan.set(true); + when(fixture.sslEngine.getDelegatedTask()).thenReturn(task, (Runnable) null); + when(fixture.sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenAnswer(invocation -> { + if (!taskRan.get()) { + return new SSLEngineResult(SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NEED_TASK, 0, 0); + } + ByteBuffer encrypted = invocation.getArgument(0); + ByteBuffer application = invocation.getArgument(1); + encrypted.position(encrypted.position() + RECORD_ENCRYPTED_BYTES); + application.put(new byte[RECORD_PLAINTEXT_BYTES]); + return new SSLEngineResult(SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING, + RECORD_ENCRYPTED_BYTES, RECORD_PLAINTEXT_BYTES); + }); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(32768); + endPoint.readBuffer(application); + assertTrue(taskRan.get()); + assertEquals(RECORD_PLAINTEXT_BYTES, application.position()); + assertFalse(endPoint.hasBufferedReadData()); + } finally { + endPoint.close(); + } + } + + @Test + void need_task_without_a_delegated_task_returns_without_repeated_unwraps() throws Exception { + Fixture fixture = new Fixture(); + when(fixture.sslEngine.unwrap(any(ByteBuffer.class), any(ByteBuffer.class))).thenReturn( + new SSLEngineResult(SSLEngineResult.Status.OK, SSLEngineResult.HandshakeStatus.NEED_TASK, 0, 0)); + SSLEndPoint endPoint = fixture.createEndPoint(); + try { + ByteBuffer application = ByteBuffer.allocate(32768); + endPoint.readBuffer(application); + assertEquals(0, application.position()); + assertTrue(endPoint.hasBufferedReadData()); + verify(fixture.sslEngine, times(1)).unwrap(any(ByteBuffer.class), any(ByteBuffer.class)); + } finally { + endPoint.close(); + } + } + private static final class Fixture { private final SocketChannel channel = mock(SocketChannel.class); diff --git a/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java new file mode 100644 index 0000000000..0eeb6067d0 --- /dev/null +++ b/src/test/java/io/mapsmessaging/network/io/impl/ssl/SSLEndPointTlsRecordTest.java @@ -0,0 +1,265 @@ +/* + * + * Copyright [ 2020 - 2024 ] Matthew Buckton + * Copyright [ 2024 - 2026 ] MapsMessaging B.V. + * + * Licensed under the Apache License, Version 2.0 with the Commons Clause + * (the "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * https://commonsclause.com/ + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.mapsmessaging.network.io.impl.ssl; + +import io.mapsmessaging.dto.rest.config.network.EndPointServerConfigDTO; +import io.mapsmessaging.dto.rest.config.network.impl.TcpConfigDTO; +import io.mapsmessaging.network.io.EndPointServerStatus; +import io.mapsmessaging.network.io.Packet; +import io.mapsmessaging.network.io.impl.Selector; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.KeyStore; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.List; +import java.util.Queue; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLEngine; +import javax.net.ssl.SSLEngineResult; +import javax.net.ssl.TrustManagerFactory; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +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 static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Exercises the endpoint with real JDK records; only the non-blocking socket boundary is mocked. */ +@Timeout(10) +class SSLEndPointTlsRecordTest { + + @ParameterizedTest + @CsvSource({"TLSv1.2,5000", "TLSv1.3,5000", "TLSv1.2,16384", "TLSv1.3,16384"}) + void a_record_larger_than_detection_buffer_decodes_without_losing_payload(String protocol, int size) throws Exception { + try (Fixture fixture = new Fixture(protocol)) { + byte[] payload = payload(size, 11); + fixture.incoming.add(ByteBuffer.wrap(fixture.encrypt(payload))); + Packet packet = new Packet(32768, false); + + assertTrue(fixture.endPoint.readPacket(packet) > 0); + + assertArrayEquals(payload, decoded(packet)); + assertFalse(fixture.endPoint.hasBufferedReadData()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"TLSv1.2", "TLSv1.3"}) + void a_second_buffered_record_survives_overflow_and_decodes_on_the_next_read(String protocol) throws Exception { + try (Fixture fixture = new Fixture(protocol)) { + byte[] first = payload(5000, 17); + byte[] second = payload(5000, 29); + byte[] firstRecord = fixture.encrypt(first); + byte[] secondRecord = fixture.encrypt(second); + ByteBuffer records = ByteBuffer.allocate(firstRecord.length + secondRecord.length); + records.put(firstRecord).put(secondRecord).flip(); + fixture.incoming.add(records); + Packet packet = new Packet(6000, false); + + fixture.endPoint.readPacket(packet); + assertArrayEquals(first, decoded(packet)); + assertTrue(fixture.incoming.isEmpty(), "Both records must already have left the socket"); + assertTrue(fixture.endPoint.hasBufferedReadData()); + + packet.clear(); + assertTrue(fixture.endPoint.readPacket(packet) > 0); + assertArrayEquals(second, decoded(packet)); + assertFalse(fixture.endPoint.hasBufferedReadData()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"TLSv1.2", "TLSv1.3"}) + void a_fragmented_record_produces_payload_only_after_its_remaining_bytes_arrive(String protocol) throws Exception { + try (Fixture fixture = new Fixture(protocol)) { + byte[] payload = payload(5000, 37); + byte[] record = fixture.encrypt(payload); + fixture.incoming.add(ByteBuffer.wrap(Arrays.copyOfRange(record, 0, 7))); + Packet packet = new Packet(32768, false); + + assertTrue(fixture.endPoint.readPacket(packet) > 0); + assertEquals(0, packet.position()); + assertTrue(fixture.endPoint.hasBufferedReadData()); + + fixture.incoming.add(ByteBuffer.wrap(Arrays.copyOfRange(record, 7, record.length))); + assertTrue(fixture.endPoint.readPacket(packet) > 0); + assertArrayEquals(payload, decoded(packet)); + assertFalse(fixture.endPoint.hasBufferedReadData()); + } + } + + private static byte[] payload(int size, int seed) { + byte[] payload = new byte[size]; + for (int index = 0; index < size; index++) { + payload[index] = (byte) (index * 31 + seed); + } + return payload; + } + + private static byte[] decoded(Packet packet) { + packet.flip(); + byte[] payload = new byte[packet.available()]; + packet.get(payload); + return payload; + } + + private static SSLContext context() throws Exception { + char[] password = System.getProperty("javax.net.ssl.keyStorePassword", "password").toCharArray(); + KeyStore keys = KeyStore.getInstance("JKS"); + Path keyStore = Path.of(System.getProperty("javax.net.ssl.keyStore", "my-keystore.jks")); + try (InputStream input = Files.newInputStream(keyStore)) { + keys.load(input, password); + } + KeyManagerFactory keyManagers = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + keyManagers.init(keys, password); + // The fixture trusts the same test identity at both ends; no external trust or network is needed. + TrustManagerFactory trustManagers = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trustManagers.init(keys); + SSLContext context = SSLContext.getInstance("TLS"); + context.init(keyManagers.getKeyManagers(), trustManagers.getTrustManagers(), null); + return context; + } + + private static final class Fixture implements AutoCloseable { + private final Queue incoming = new ArrayDeque<>(); + private final SSLEngine client; + private final SSLEndPoint endPoint; + + private Fixture(String protocol) throws Exception { + SSLContext context = context(); + client = context.createSSLEngine(); + client.setUseClientMode(true); + client.setEnabledProtocols(new String[]{protocol}); + SSLEngine server = context.createSSLEngine(); + server.setEnabledProtocols(new String[]{protocol}); + SocketChannel channel = mock(SocketChannel.class); + Socket socket = mock(Socket.class); + Selector selector = mock(Selector.class); + EndPointServerStatus status = mock(EndPointServerStatus.class); + EndPointServerConfigDTO config = new EndPointServerConfigDTO(); + config.setEndPointConfig(new TcpConfigDTO()); + when(status.getConfig()).thenReturn(config); + when(channel.socket()).thenReturn(socket); + when(channel.isConnected()).thenReturn(true); + when(socket.getLocalAddress()).thenReturn(InetAddress.getLoopbackAddress()); + when(channel.write(any(ByteBuffer.class))).thenAnswer(invocation -> { + ByteBuffer bytes = invocation.getArgument(0); + int count = bytes.remaining(); + bytes.position(bytes.limit()); + return count; + }); + when(channel.read(any(ByteBuffer.class))).thenAnswer(invocation -> { + if (incoming.isEmpty()) { + return 0; + } + ByteBuffer target = invocation.getArgument(0); + ByteBuffer source = incoming.element(); + int count = Math.min(target.remaining(), source.remaining()); + int limit = source.limit(); + source.limit(source.position() + count); + target.put(source); + source.limit(limit); + if (!source.hasRemaining()) { + incoming.remove(); + } + return count; + }); + endPoint = new SSLEndPoint(1, server, channel, selector, null, status, List.of()); + try { + handshake(server); + endPoint.handshakeManager = new SSLHandshakeManagerFinished(ByteBuffer.allocate(server.getSession().getApplicationBufferSize())); + } catch (Exception failure) { + endPoint.close(); + throw failure; + } + } + + private byte[] encrypt(byte[] payload) throws Exception { + ByteBuffer encrypted = ByteBuffer.allocate(65536); + ByteBuffer application = ByteBuffer.wrap(payload); + while (application.hasRemaining()) { + SSLEngineResult result = client.wrap(application, encrypted); + assertEquals(SSLEngineResult.Status.OK, result.getStatus()); + assertTrue(result.bytesConsumed() > 0, "Wrapping application data must make progress"); + } + encrypted.flip(); + byte[] record = new byte[encrypted.remaining()]; + encrypted.get(record); + return record; + } + + private void handshake(SSLEngine server) throws Exception { + client.beginHandshake(); + ByteBuffer clientToServer = ByteBuffer.allocate(131072); + ByteBuffer serverToClient = ByteBuffer.allocate(131072); + ByteBuffer sink = ByteBuffer.allocate(65536); + for (int attempt = 0; attempt < 100; attempt++) { + handshakeStep(client, server, clientToServer, sink); + handshakeStep(server, client, serverToClient, sink); + if (client.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING + && server.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING + && clientToServer.position() == 0 && serverToClient.position() == 0) { + return; + } + } + throw new AssertionError("TLS handshake did not complete"); + } + + private static void handshakeStep(SSLEngine sender, SSLEngine receiver, ByteBuffer wire, ByteBuffer sink) throws Exception { + runTasks(sender); + runTasks(receiver); + if (sender.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NEED_WRAP) { + sender.wrap(ByteBuffer.allocate(0), wire); + } + wire.flip(); + if (wire.hasRemaining()) { + sink.clear(); + receiver.unwrap(wire, sink); + } + wire.compact(); + runTasks(receiver); + } + + private static void runTasks(SSLEngine engine) { + Runnable task; + while ((task = engine.getDelegatedTask()) != null) { + task.run(); + } + } + + @Override + public void close() { + endPoint.close(); + } + } +} diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java index 92079e8262..78a67cba60 100644 --- a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java @@ -1,6 +1,9 @@ package io.mapsmessaging.network.protocol.impl.cot; import io.mapsmessaging.api.MessageEvent; +import io.mapsmessaging.api.Destination; +import io.mapsmessaging.api.message.Message; +import io.mapsmessaging.api.features.DestinationType; import io.mapsmessaging.api.Session; import io.mapsmessaging.api.SessionManager; import io.mapsmessaging.dto.rest.config.network.EndPointConfigDTO; @@ -13,9 +16,14 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.mockito.MockedStatic; +import org.mockito.ArgumentCaptor; import javax.security.auth.Subject; import java.nio.channels.SelectionKey; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -111,7 +119,7 @@ void readTask_whenReadReportsBytesButDecodesNoPayload_parksInsteadOfSpinning() t @Test @Timeout(10) void readTask_readsIntoBufferThatHoldsATlsRecord_notTheDetectionPacket() throws Exception { - withSession(endpoint -> { + withSession((endpoint, session) -> { List readTargets = new CopyOnWriteArrayList<>(); when(endpoint.readPacket(any())).thenAnswer(invocation -> { readTargets.add(invocation.getArgument(0)); @@ -148,7 +156,7 @@ void readBufferSize_usesConfiguredSize_withinTlsRecordFloorAndSafetyCap() { @Test @Timeout(10) void readTask_doesNotGrowThePooledThreadNameAcrossRuns() throws Exception { - withSession(endpoint -> { + withSession((endpoint, session) -> { List readerNames = new CopyOnWriteArrayList<>(); when(endpoint.readPacket(any())).thenAnswer(invocation -> { readerNames.add(Thread.currentThread().getName()); @@ -171,6 +179,44 @@ void readTask_doesNotGrowThePooledThreadNameAcrossRuns() throws Exception { }); } + @Test + @Timeout(10) + void detection_prefix_and_large_following_read_publish_one_complete_event() throws Exception { + withSession((endpoint, session) -> { + Destination destination = mock(Destination.class); + when(session.findDestination("/tak/cot/inbound", DestinationType.TOPIC)) + .thenReturn(CompletableFuture.completedFuture(destination)); + byte[] event = ("" + "payload".repeat(800) + + "").getBytes(StandardCharsets.UTF_8); + int split = 128; + Packet detection = new Packet(1024, false); + detection.put(Arrays.copyOfRange(event, 0, split)); + detection.flip(); + AtomicBoolean delivered = new AtomicBoolean(); + when(endpoint.readPacket(any())).thenAnswer(invocation -> { + if (delivered.getAndSet(true)) { + return 0; + } + Packet target = invocation.getArgument(0); + target.put(Arrays.copyOfRange(event, split, event.length)); + return event.length - split; + }); + CotProtocol protocol = new CotProtocol(endpoint, detection, new CotProtocolConfigDTO()); + try { + verify(destination, never()).storeMessage(any()); + clearInvocations(endpoint); + protocol.selected(protocol, null, SelectionKey.OP_READ); + verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + ArgumentCaptor stored = ArgumentCaptor.forClass(Message.class); + verify(destination, times(1)).storeMessage(stored.capture()); + assertArrayEquals(event, stored.getValue().getOpaqueData()); + verify(endpoint, never()).close(); + } finally { + protocol.close(); + } + }); + } + @Test void staticFrameSearchHelpersHandleBoundsAndWhitespace() throws Exception { byte[] data = "xx 1tail".getBytes(); @@ -193,7 +239,7 @@ void staticFrameSearchHelpersHandleBoundsAndWhitespace() throws Exception { @FunctionalInterface private interface SessionBody { - void run(EndPoint endpoint) throws Exception; + void run(EndPoint endpoint, Session session) throws Exception; } private static void withSession(SessionBody body) throws Exception { @@ -207,7 +253,7 @@ private static void withSession(SessionBody body) throws Exception { when(manager.create(any(), any())).thenReturn(session); try (MockedStatic mocked = mockStatic(SessionManager.class)) { mocked.when(SessionManager::getInstance).thenReturn(manager); - body.run(endpoint()); + body.run(endpoint(), session); } finally { JMXManager.setEnableJMX(original); } From 6da368dde683bd44fe39d3f19184a0be28f03bee Mon Sep 17 00:00:00 2001 From: Matthew Buckton Date: Thu, 15 Oct 2026 15:34:26 +1100 Subject: [PATCH 3/3] test(cot): isolate TLS handoff from daemon singleton state Run the reader under a scoped daemon mock so the handoff assertion remains deterministic after other protocol tests start the server. Refs: MSG-415 --- .../protocol/impl/cot/CotProtocolTest.java | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java index 78a67cba60..63b70dc5eb 100644 --- a/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java +++ b/src/test/java/io/mapsmessaging/network/protocol/impl/cot/CotProtocolTest.java @@ -1,6 +1,7 @@ package io.mapsmessaging.network.protocol.impl.cot; import io.mapsmessaging.api.MessageEvent; +import io.mapsmessaging.MessageDaemon; import io.mapsmessaging.api.Destination; import io.mapsmessaging.api.message.Message; import io.mapsmessaging.api.features.DestinationType; @@ -19,6 +20,7 @@ import org.mockito.ArgumentCaptor; import javax.security.auth.Subject; +import java.lang.reflect.Constructor; import java.nio.channels.SelectionKey; import java.nio.charset.StandardCharsets; import java.util.Arrays; @@ -202,11 +204,18 @@ void detection_prefix_and_large_following_read_publish_one_complete_event() thro return event.length - split; }); CotProtocol protocol = new CotProtocol(endpoint, detection, new CotProtocolConfigDTO()); - try { + try (MockedStatic daemon = mockStatic(MessageDaemon.class)) { + daemon.when(MessageDaemon::getInstance).thenReturn(null); verify(destination, never()).storeMessage(any()); clearInvocations(endpoint); - protocol.selected(protocol, null, SelectionKey.OP_READ); - verify(endpoint, timeout(5000)).register(eq(SelectionKey.OP_READ), eq(protocol)); + // Run the actual reader on this thread so the singleton boundary remains isolated, + // independent of whether another protocol test has started the daemon. + Class readerType = Class.forName(CotProtocol.class.getName() + "$ReadTask"); + Constructor constructor = readerType.getDeclaredConstructor(CotProtocol.class); + constructor.setAccessible(true); + Runnable reader = (Runnable) constructor.newInstance(protocol); + reader.run(); + verify(endpoint).register(eq(SelectionKey.OP_READ), eq(protocol)); ArgumentCaptor stored = ArgumentCaptor.forClass(Message.class); verify(destination, times(1)).storeMessage(stored.capture()); assertArrayEquals(event, stored.getValue().getOpaqueData());