diff --git a/pnnl.goss.core/build.gradle b/pnnl.goss.core/build.gradle index 6b47379a..4832d002 100644 --- a/pnnl.goss.core/build.gradle +++ b/pnnl.goss.core/build.gradle @@ -16,6 +16,12 @@ dependencies { // AssertJ for fluent assertions testImplementation 'org.assertj:assertj-core:3.26.3' + + // Embedded broker for GOSS-023: bounded-receive must be proven against a + // real JMS session's wall-clock blocking behavior, not only a mocked + // MessageConsumer. Test-scope only; pnnl.goss.core.runner already carries + // this at implementation scope for its own live-broker itests. + testImplementation 'org.apache.activemq:activemq-broker:6.2.0' } test { diff --git a/pnnl.goss.core/src/pnnl/goss/core/Client.java b/pnnl.goss.core/src/pnnl/goss/core/Client.java index b5f86395..492f3fb3 100644 --- a/pnnl.goss.core/src/pnnl/goss/core/Client.java +++ b/pnnl.goss.core/src/pnnl/goss/core/Client.java @@ -55,6 +55,53 @@ public Serializable getResponse(Serializable request, String destination, public Serializable getResponse(Serializable request, String destination, RESPONSE_FORMAT responseFormat, DESTINATION_TYPE destinationType) throws SystemException, JMSException; + /** + * Makes synchronous call to the server using QUEUE destination (default), + * bounded by an explicit receive timeout. Unlike the no-timeout overloads + * (which block indefinitely waiting for a reply), this returns {@code null} + * once {@code timeoutMillis} elapses without a reply, rather than blocking the + * calling thread forever. Intended for callers that need to retry against a + * service that may not be answerable yet (a boot-order race), where an + * unbounded wait would defeat the retry loop entirely. + * + * @param request + * @param destination + * @param responseFormat + * @param timeoutMillis + * maximum time to wait for a reply, in milliseconds. A value of + * {@code 0} blocks indefinitely (matches + * {@link jakarta.jms.MessageConsumer#receive(long)} semantics). + * @return the response, or {@code null} if no reply arrived within + * {@code timeoutMillis} + * @throws SystemException + */ + public Serializable getResponse(Serializable request, String destination, + RESPONSE_FORMAT responseFormat, long timeoutMillis) throws SystemException, JMSException; + + /** + * Makes synchronous call to the server with specified destination type, bounded + * by an explicit receive timeout. See + * {@link #getResponse(Serializable, String, RESPONSE_FORMAT, long)} for the + * timeout semantics. + * + * @param request + * @param destination + * destination name + * @param responseFormat + * @param destinationType + * TOPIC or QUEUE + * @param timeoutMillis + * maximum time to wait for a reply, in milliseconds. A value of + * {@code 0} blocks indefinitely (matches + * {@link jakarta.jms.MessageConsumer#receive(long)} semantics). + * @return the response, or {@code null} if no reply arrived within + * {@code timeoutMillis} + * @throws SystemException + */ + public Serializable getResponse(Serializable request, String destination, + RESPONSE_FORMAT responseFormat, DESTINATION_TYPE destinationType, long timeoutMillis) + throws SystemException, JMSException; + /** * Lets the client subscribe to a Topic of the given name for event based * communication. diff --git a/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java b/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java index 2b730fb9..77772764 100644 --- a/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java +++ b/pnnl.goss.core/src/pnnl/goss/core/client/GossClient.java @@ -53,6 +53,7 @@ import java.util.List; import java.util.UUID; +import jakarta.jms.BytesMessage; import jakarta.jms.Connection; import jakarta.jms.Destination; import jakarta.jms.JMSException; @@ -93,6 +94,20 @@ public class GossClient implements Client { private static final Logger log = LoggerFactory.getLogger(GossClient.class); + // jakarta.jms.MessageConsumer#receive(long) treats a timeout of 0 as "block + // indefinitely", the same contract as the no-arg receive() it replaces. Used + // by the unbounded getResponse() overloads below so their behavior is + // unchanged after the bounded-timeout overload was added (GADP-051). + private static final long UNBOUNDED_RECEIVE_TIMEOUT_MS = 0L; + + // Sane upper bound on a BytesMessage reply body decoded by getResponse(). + // A topology/response JSON reply is at most a few MB; a length beyond this + // is either a malformed/adversarial wire value or provider corruption, and + // allocating a byte[] directly from an unbounded wire-supplied length is an + // OOM denial-of-service vector (and the (int) cast on getBodyLength() is + // lossy/negative past Integer.MAX_VALUE). Reject rather than allocate. + private static final long MAX_BYTES_MESSAGE_BODY_LENGTH = 8L * 1024 * 1024; + private UUID uuid = null; private String brokerUri = null; private String stompUri = null; @@ -288,7 +303,9 @@ public Serializable getResponse(Serializable message, String destinationName, /** * Sends request and gets response for synchronous communication with specified - * destination type. + * destination type. Blocks indefinitely for a reply; see + * {@link #getResponse(Serializable, String, RESPONSE_FORMAT, DESTINATION_TYPE, long)} + * for a bounded-wait variant. * * @param message * instance of pnnl.goss.core.Request or any of its subclass. @@ -308,6 +325,73 @@ public Serializable getResponse(Serializable message, String destinationName, @Override public Serializable getResponse(Serializable message, String destinationName, RESPONSE_FORMAT responseFormat, DESTINATION_TYPE destinationType) throws SystemException, JMSException { + // Preserve the unbounded-wait contract exactly: 0 means "block indefinitely" + // per jakarta.jms.MessageConsumer#receive(long), the same as the no-arg + // receive() this delegation replaced. + return getResponse(message, destinationName, responseFormat, destinationType, + UNBOUNDED_RECEIVE_TIMEOUT_MS); + } + + /** + * Sends request and gets response for synchronous communication, defaulting to + * QUEUE destination type, bounded by an explicit receive timeout (GADP-051). See + * {@link #getResponse(Serializable, String, RESPONSE_FORMAT, DESTINATION_TYPE, long)} + * for the full timeout semantics. + * + * @param message + * instance of pnnl.goss.core.Request or any of its subclass. + * @param destinationName + * the destination name (topic or queue) + * @param responseFormat + * the response format + * @param timeoutMillis + * maximum time to wait for a reply, in milliseconds. A value of + * {@code 0} blocks indefinitely (matches + * {@link jakarta.jms.MessageConsumer#receive(long)} semantics). + * @return return an Object which could be a pnnl.goss.core.DataResponse, + * pnnl.goss.core.UploadResponse or pnnl.goss.core.DataError, or + * {@code null} if no reply arrived within {@code timeoutMillis}. + * @throws IllegalStateException + * when GossCLient is initialized with an GossResponseEvent. Cannot + * synchronously receive a message when a MessageListener is set. + * @throws JMSException + */ + @Override + public Serializable getResponse(Serializable message, String destinationName, + RESPONSE_FORMAT responseFormat, long timeoutMillis) throws SystemException, JMSException { + return getResponse(message, destinationName, responseFormat, DESTINATION_TYPE.QUEUE, timeoutMillis); + } + + /** + * Sends request and gets response for synchronous communication with specified + * destination type, bounded by an explicit receive timeout (GADP-051). Unlike + * the unbounded overloads, this returns {@code null} once {@code timeoutMillis} + * elapses without a reply, rather than blocking the calling thread forever. + * + * @param message + * instance of pnnl.goss.core.Request or any of its subclass. + * @param destinationName + * the destination name (topic or queue) + * @param responseFormat + * the response format + * @param destinationType + * TOPIC or QUEUE + * @param timeoutMillis + * maximum time to wait for a reply, in milliseconds. A value of + * {@code 0} blocks indefinitely (matches + * {@link jakarta.jms.MessageConsumer#receive(long)} semantics). + * @return return an Object which could be a pnnl.goss.core.DataResponse, + * pnnl.goss.core.UploadResponse or pnnl.goss.core.DataError, or + * {@code null} if no reply arrived within {@code timeoutMillis}. + * @throws IllegalStateException + * when GossCLient is initialized with an GossResponseEvent. Cannot + * synchronously receive a message when a MessageListener is set. + * @throws JMSException + */ + @Override + public Serializable getResponse(Serializable message, String destinationName, + RESPONSE_FORMAT responseFormat, DESTINATION_TYPE destinationType, long timeoutMillis) + throws SystemException, JMSException { if (protocol == null) { protocol = PROTOCOL.OPENWIRE; } @@ -320,33 +404,175 @@ public Serializable getResponse(Serializable message, String destinationName, } Serializable response = null; - Destination replyDestination = getTemporaryDestination(); - Destination destination = getDestination(destinationName, destinationType); - log.debug("Creating consumer for destination " + replyDestination + " (type: " + destinationType + ")"); - DefaultClientConsumer clientConsumer = new DefaultClientConsumer( - session, replyDestination); + // GADP-051 (session-isolation): the synchronous receive() below MUST run + // on a Session that has NO MessageListener attached. The shared `session` + // may carry async subscribe() listeners (for example FieldBusManager's + // device-output subscription, which publishDeviceOutput() registers before + // it issues its synchronous topology request on the same client), and + // jakarta.jms forbids a synchronous receive() on such a session: + // ActiveMQSession.checkMessageListener throws IllegalStateException, + // "Cannot synchronously receive a message when a MessageListener is set". + // That made every topology getResponse() throw instantly regardless of + // timing, so the caller's bounded-retry window could never succeed. + // + // Fix: derive a dedicated, listener-free Session from the SAME Connection + // for this request/reply, and close it in the finally block. A JMS + // Connection supports many Sessions, so this isolates the synchronous + // path from the async listener path without disturbing the shared + // session's subscriptions. The request is still SENT via the shared + // clientPublisher (publishing on a listener-bound session is permitted); + // only the temporary reply destination, its consumer, and the synchronous + // receive move onto the dedicated session. The session is always closed, + // so no session is leaked per call. + getSession(); // ensure the shared Connection exists before deriving a session from it + Session syncSession = null; + DefaultClientConsumer clientConsumer = null; try { + syncSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Destination replyDestination = getTemporaryDestination(syncSession); + Destination destination = getDestination(destinationName, destinationType); + + log.debug("Creating consumer for destination " + replyDestination + " (type: " + destinationType + ")"); + clientConsumer = new DefaultClientConsumer(syncSession, replyDestination); + clientPublisher.sendMessage(message, destination, replyDestination, responseFormat); Message responseMessage = clientConsumer.getMessageConsumer() - .receive(); - response = ((TextMessage) responseMessage).getText(); + .receive(timeoutMillis); + if (responseMessage == null) { + // Timed out waiting for a reply (or the consumer was concurrently + // closed, per the JMS receive(long) contract). A null Message here + // means "no reply arrived in time", not "an empty reply": return + // null explicitly rather than falling through to the TextMessage + // cast below, which would NPE on a null responseMessage. Callers + // that retry (e.g. FieldBusManager's bounded topology request) rely + // on this null to mean "not ready yet, try again". + log.debug("No response received on " + replyDestination + " within " + + timeoutMillis + "ms"); + return response; + } + // GADP-051 (message-type dispatch): check instanceof BEFORE casting. + // The previous code unconditionally cast responseMessage to TextMessage + // before these checks, so any reply that was NOT a TextMessage (for + // example ActiveMQBytesMessage, which is how a reply arrives over the + // STOMP-to-OpenWire bridge, e.g. the topology background service's + // response) threw ClassCastException immediately. That exception was + // previously masked by the session-listener IllegalStateException this + // method used to throw first (GADP-051 session-split fix); once that + // was fixed the latent bad cast surfaced. Handle each real message type + // explicitly, matching the decode convention already used in + // DefaultClientListener.onMessage for the same wire formats, and fail + // loudly (data-invariants Rule 2) rather than defaulting to null/empty + // on an unrecognized type: the body is consumed downstream (e.g. + // TopologyRequestProcess builds its topology map from it), so a wrong + // or empty body would be silent data corruption, not just a crash. if (responseMessage instanceof ObjectMessage) { ObjectMessage objectMessage = (ObjectMessage) responseMessage; if (objectMessage.getObject() instanceof Response) { + // The reply carries a first-class pnnl.goss.core.Response + // (or subclass, e.g. DataResponse) written as the + // ObjectMessage payload by an OpenWire-side sender. Unwrap + // it directly rather than treating it as opaque + // Serializable, so callers get the typed Response contract + // they expect from this method's declared return. response = (Response) objectMessage.getObject(); + } else { + // The reply carries some other Serializable payload (not a + // Response). This method's contract is "return whatever + // was sent", so pass it through unchanged rather than + // rejecting it: the sender, not this client, decides what + // is a valid payload shape. + response = (Serializable) objectMessage.getObject(); } } else if (responseMessage instanceof TextMessage) { response = ((TextMessage) responseMessage).getText(); + } else if (responseMessage instanceof BytesMessage) { + // BytesMessage is used by STOMP clients (Python, JavaScript, etc.) + // and by replies bridged from STOMP to OpenWire. Decode with the + // same UTF-8 convention DefaultClientListener.onMessage uses for + // the equivalent BytesMessage case, so both synchronous and + // asynchronous receive paths interpret the wire body identically. + BytesMessage bytesMessage = (BytesMessage) responseMessage; + long bodyLength = bytesMessage.getBodyLength(); + // Validate the wire-supplied length BEFORE allocating. An + // unbounded/negative bodyLength allocated directly into `new + // byte[(int) bodyLength]` is an OOM denial-of-service vector + // (and the (int) cast is lossy/negative past + // Integer.MAX_VALUE, which would throw + // NegativeArraySizeException from inside the allocation + // rather than a clear diagnostic). A real topology/response + // JSON reply is a few MB at most, so reject anything outside + // a sane cap loudly, matching the else-branch's fail-loud + // contract, rather than allocating from an unvalidated value. + if (bodyLength < 0 || bodyLength > MAX_BYTES_MESSAGE_BODY_LENGTH) { + throw SystemException.wrap(new JMSException( + "BytesMessage reply body length " + bodyLength + + " is negative or exceeds the " + + MAX_BYTES_MESSAGE_BODY_LENGTH + + "-byte cap for getResponse replies")) + .set("destination", destinationName) + .set("message", message); + } + byte[] bytes = new byte[(int) bodyLength]; + bytesMessage.readBytes(bytes); + response = new String(bytes, java.nio.charset.StandardCharsets.UTF_8); + } else { + // An unexpected/unsupported JMS message type. Do not silently + // return null or an empty body here: that would convert a visible + // failure into invisible data loss for whatever consumed the + // response downstream. Fail loudly with the concrete type so the + // caller (and its logs) can diagnose it. + throw SystemException.wrap(new JMSException( + "Unsupported JMS message type for getResponse reply: " + + responseMessage.getClass().getName())) + .set("destination", destinationName) + .set("message", message); } } catch (JMSException e) { - SystemException.wrap(e).set("destination", destinationName).set("message", message); + // GOSS-023: a genuine JMSException here (a real transport/provider + // failure, e.g. the broker or consumer going away mid-receive) must + // never be silently coerced into the same null result as a genuine + // timeout. The previous code constructed a SystemException and threw + // it away without ever calling throw, so this catch block was a dead + // computation: the failure was swallowed and getResponse fell through + // to "return response" (still null) after only the few milliseconds + // it took the provider to report the failure, not the requested + // timeoutMillis. That made a hard error indistinguishable from + // "no reply arrived in time" and collapsed callers' bounded-retry + // windows (see GADP-051's TopologyRequestProcess, whose designed + // ~19s retry budget collapsed to ~4s because every attempt returned + // in a few ms instead of blocking). Surface it instead, matching the + // getTemporaryDestination/getDestination pattern already used + // elsewhere in this class: throw, don't swallow. + throw SystemException.wrap(e).set("destination", destinationName).set("message", message); } finally { + // Guard the consumer close the same way the session close below is + // guarded: if consumer.close() throws, that must not skip the + // syncSession.close() that follows, which would otherwise leak the + // dedicated request/reply session. Log with context rather than + // swallowing silently. if (clientConsumer != null) { - clientConsumer.close(); + try { + clientConsumer.close(); + } catch (Exception e) { + log.warn("Failed to close synchronous request/reply consumer for destination {}", + destinationName, e); + } + } + // Close the dedicated request/reply session so it is not leaked per + // call. Guard the close so a failure tearing down this session cannot + // mask a response already computed above, and log it with context + // rather than swallowing it silently. + if (syncSession != null) { + try { + syncSession.close(); + } catch (JMSException e) { + log.warn("Failed to close synchronous request/reply session for destination {}", + destinationName, e); + } } } @@ -661,19 +887,24 @@ private Session getSession() throws SystemException { return session; } - private Destination getTemporaryDestination() throws SystemException { + // Create the temporary reply queue on the supplied session. A JMS + // TemporaryQueue is scoped to the Session (really the Connection) that + // created it and can only be consumed by that same session, so the + // synchronous getResponse() path passes its dedicated, listener-free session + // here (GADP-051) rather than the shared listener-bearing `session`. + private Destination getTemporaryDestination(Session destinationSession) throws SystemException { Destination destination = null; try { if (protocol.equals(PROTOCOL.SSL)) { - destination = getSession().createTemporaryQueue(); + destination = destinationSession.createTemporaryQueue(); if (destination == null) { throw new SystemException(ConnectionCode.DESTINATION_ERROR); } } else { if (protocol.equals(PROTOCOL.OPENWIRE) || protocol.equals(PROTOCOL.STOMP)) { // Both OPENWIRE and STOMP use standard JMS with ActiveMQ - destination = getSession().createTemporaryQueue(); + destination = destinationSession.createTemporaryQueue(); if (destination == null) { throw new SystemException( ConnectionCode.DESTINATION_ERROR); diff --git a/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientBoundedReceiveTest.java b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientBoundedReceiveTest.java new file mode 100644 index 00000000..95b4edcd --- /dev/null +++ b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientBoundedReceiveTest.java @@ -0,0 +1,327 @@ +package pnnl.goss.core.client.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.Serializable; +import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; + +import jakarta.jms.BytesMessage; +import jakarta.jms.Connection; +import jakarta.jms.MessageConsumer; +import jakarta.jms.ObjectMessage; +import jakarta.jms.Queue; +import jakarta.jms.Session; +import jakarta.jms.TemporaryQueue; +import jakarta.jms.TextMessage; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import pnnl.goss.core.Client.DESTINATION_TYPE; +import pnnl.goss.core.Client.PROTOCOL; +import pnnl.goss.core.ClientPublishser; +import pnnl.goss.core.DataResponse; +import pnnl.goss.core.Request.RESPONSE_FORMAT; +import pnnl.goss.core.Response; +import pnnl.goss.core.client.GossClient; + +/** + * Regression coverage for GADP-051: GossClient.getResponse()'s reply wait was + * unbounded (MessageConsumer.receive() with no timeout), so a caller polling a + * service that has not started answering yet (a boot-order race, e.g. the + * topology-background-service at cold start) blocked forever on the first + * attempt and could never retry. These tests cover the new bounded-timeout + * overload's contract directly against a mocked JMS Session/MessageConsumer (no + * live broker, same pattern as GossClientConsumerLeakTest): a timed-out receive + * returns null and still closes the consumer rather than blocking or throwing, + * and the pre-existing unbounded overloads keep calling receive(0), preserving + * their exact prior "block indefinitely" behavior for other callers. + * + *

GADP-051 (session-isolation): getResponse() now derives a dedicated, + * listener-free Session from the Connection for its synchronous request/reply + * (so an async subscribe() listener on the shared session cannot poison the + * synchronous receive()). The mock wiring therefore stubs the Connection to + * return a mock Session, on which the temporary reply queue and its consumer + * are created; the test additionally asserts that dedicated session is closed + * after the call, matching the leak-safe teardown the fix guarantees. + */ +public class GossClientBoundedReceiveTest { + + private static final String DESTINATION = "goss.gridappsd.request.data.cimtopology"; + private static final long TIMEOUT_MILLIS = 3000L; + + private GossClient client; + private Connection mockConnection; + private Session mockSession; + private MessageConsumer mockConsumer; + private ClientPublishser mockPublisher; + + @BeforeEach + void setUp() throws Exception { + client = new GossClient(PROTOCOL.OPENWIRE, null, "tcp://localhost:61616", + "stomp://localhost:61613"); + + mockConnection = mock(Connection.class); + mockSession = mock(Session.class); + TemporaryQueue mockTempQueue = mock(TemporaryQueue.class); + Queue mockQueue = mock(Queue.class); + mockConsumer = mock(MessageConsumer.class); + mockPublisher = mock(ClientPublishser.class); + + // getResponse() derives its dedicated sync session from the Connection. + when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)).thenReturn(mockSession); + when(mockSession.createTemporaryQueue()).thenReturn(mockTempQueue); + when(mockSession.createQueue(DESTINATION)).thenReturn(mockQueue); + when(mockSession.createConsumer(mockTempQueue)).thenReturn(mockConsumer); + + // Both the shared session and the connection are pre-populated so + // getSession() is a no-op (session already present) and the dedicated + // sync session is derived from the injected connection. + setPrivateField(client, "connection", mockConnection); + setPrivateField(client, "session", mockSession); + setPrivateField(client, "clientPublisher", mockPublisher); + } + + @Test + @DisplayName("a timed-out receive returns null instead of blocking or throwing, and still closes the consumer") + void timedOutReceiveReturnsNullAndClosesConsumer() throws Exception { + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(null); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + + assertThat(result).isNull(); + verify(mockConsumer, times(1)).receive(TIMEOUT_MILLIS); + verify(mockConsumer, times(1)).close(); + // The dedicated request/reply session must be closed (leak-safe teardown). + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("the caller-supplied timeout is honored: receive(timeoutMillis) is called, not the no-arg receive()") + void receiveIsCalledWithTheGivenTimeout() throws Exception { + TextMessage mockMessage = mock(TextMessage.class); + when(mockMessage.getText()).thenReturn("{\"ok\":true}"); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + + assertThat(result).isEqualTo("{\"ok\":true}"); + verify(mockConsumer, times(1)).receive(TIMEOUT_MILLIS); + verify(mockConsumer, never()).receive(); + } + + @Test + @DisplayName("the pre-existing 4-arg (destinationType) overload still calls receive(0): block-indefinitely contract unchanged") + void unboundedDestinationTypeOverloadStillBlocksIndefinitely() throws Exception { + TextMessage mockMessage = mock(TextMessage.class); + when(mockMessage.getText()).thenReturn("{\"ok\":true}"); + when(mockConsumer.receive(0L)).thenReturn(mockMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE); + + assertThat(result).isEqualTo("{\"ok\":true}"); + verify(mockConsumer, times(1)).receive(0L); + } + + @Test + @DisplayName("the pre-existing 3-arg (default QUEUE) overload still calls receive(0): block-indefinitely contract unchanged") + void defaultQueueOverloadStillBlocksIndefinitely() throws Exception { + TextMessage mockMessage = mock(TextMessage.class); + when(mockMessage.getText()).thenReturn("{\"ok\":true}"); + when(mockConsumer.receive(0L)).thenReturn(mockMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON); + + assertThat(result).isEqualTo("{\"ok\":true}"); + verify(mockConsumer, times(1)).receive(0L); + } + + @Test + @DisplayName("the new 4-arg (default QUEUE, timeoutMillis) overload delegates the given timeout through to receive()") + void defaultQueueTimeoutOverloadHonorsTheGivenTimeout() throws Exception { + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(null); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, TIMEOUT_MILLIS); + + assertThat(result).isNull(); + verify(mockConsumer, times(1)).receive(TIMEOUT_MILLIS); + verify(mockConsumer, times(1)).close(); + // The dedicated request/reply session must be closed (leak-safe teardown). + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("GADP-051 (message-type dispatch): a BytesMessage reply (STOMP-to-OpenWire bridged, e.g. the " + + "topology background service) is decoded to its UTF-8 text body instead of throwing ClassCastException " + + "on an unconditional TextMessage cast") + void bytesMessageReplyIsDecodedToItsUtf8TextBody() throws Exception { + String expectedBody = "{\"feeders\":[{\"mrid\":\"abc123\"}]}"; + byte[] expectedBytes = expectedBody.getBytes(StandardCharsets.UTF_8); + + BytesMessage mockBytesMessage = mock(BytesMessage.class); + when(mockBytesMessage.getBodyLength()).thenReturn((long) expectedBytes.length); + when(mockBytesMessage.readBytes(any(byte[].class))).thenAnswer(invocation -> { + byte[] target = invocation.getArgument(0); + System.arraycopy(expectedBytes, 0, target, 0, expectedBytes.length); + return expectedBytes.length; + }); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockBytesMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + + assertThat(result).isEqualTo(expectedBody); + verify(mockConsumer, times(1)).receive(TIMEOUT_MILLIS); + verify(mockConsumer, times(1)).close(); + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("GADP-051 review remediation item 1: an ObjectMessage carrying a Response is unwrapped to that " + + "exact Response instance") + void objectMessageCarryingResponseIsUnwrappedToTheResponseBody() throws Exception { + DataResponse expected = new DataResponse(); + expected.setDestination(DESTINATION); + expected.setUsername("expected-user"); + + ObjectMessage mockObjectMessage = mock(ObjectMessage.class); + when(mockObjectMessage.getObject()).thenReturn(expected); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockObjectMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + + // Value-assert the decoded body is the exact Response instance the + // ObjectMessage carried, not merely "some non-null Response". + assertThat(result).isSameAs(expected); + assertThat(((Response) result).getId()).isEqualTo(expected.getId()); + assertThat(((DataResponse) result).getUsername()).isEqualTo("expected-user"); + verify(mockConsumer, times(1)).receive(TIMEOUT_MILLIS); + verify(mockConsumer, times(1)).close(); + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("GADP-051 review remediation item 1: an ObjectMessage carrying a plain (non-Response) " + + "Serializable is passed through unchanged, per the current else-branch contract") + void objectMessageCarryingPlainSerializableIsPassedThroughUnchanged() throws Exception { + // A plain Serializable that is NOT a Response/DataResponse. The code + // under test's else branch does `response = (Serializable) + // objectMessage.getObject()` with no further transformation, so the + // correct assertion is that the exact same String instance/value + // comes back untouched. + String expectedPayload = "plain-serializable-payload"; + + ObjectMessage mockObjectMessage = mock(ObjectMessage.class); + when(mockObjectMessage.getObject()).thenReturn(expectedPayload); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockObjectMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + + assertThat(result).isEqualTo(expectedPayload); + assertThat(result).isNotInstanceOf(Response.class); + verify(mockConsumer, times(1)).receive(TIMEOUT_MILLIS); + verify(mockConsumer, times(1)).close(); + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("GADP-051 review remediation item 3: a BytesMessage whose reported body length exceeds the " + + "sane cap throws SystemException instead of allocating an unbounded byte[]") + void oversizedBytesMessageBodyLengthThrowsInsteadOfAllocating() throws Exception { + BytesMessage mockBytesMessage = mock(BytesMessage.class); + // One byte over the 8 MiB cap. + when(mockBytesMessage.getBodyLength()).thenReturn((8L * 1024 * 1024) + 1); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockBytesMessage); + + assertThatThrownBy(() -> client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS)) + .isInstanceOf(com.northconcepts.exception.SystemException.class) + .hasMessageContaining("exceeds"); + // readBytes must never be called: the allocation itself is what this + // guard exists to prevent, so it must be rejected before any attempt + // to size or fill a buffer from the wire-supplied length. + verify(mockBytesMessage, never()).readBytes(any(byte[].class)); + verify(mockConsumer, times(1)).close(); + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("GADP-051 review remediation item 3: a BytesMessage reporting a negative body length throws " + + "SystemException instead of a lossy (int) cast feeding NegativeArraySizeException") + void negativeBytesMessageBodyLengthThrowsInsteadOfAllocating() throws Exception { + BytesMessage mockBytesMessage = mock(BytesMessage.class); + when(mockBytesMessage.getBodyLength()).thenReturn(-1L); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockBytesMessage); + + assertThatThrownBy(() -> client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS)) + .isInstanceOf(com.northconcepts.exception.SystemException.class) + .hasMessageContaining("negative"); + verify(mockBytesMessage, never()).readBytes(any(byte[].class)); + verify(mockConsumer, times(1)).close(); + verify(mockSession, times(1)).close(); + } + + @Test + @DisplayName("GADP-051 review remediation item 3 invariant: a valid normal-size BytesMessage still decodes " + + "to the exact same UTF-8 string as before the length-cap guard was added") + void normalSizeBytesMessageStillDecodesToExactUtf8String() throws Exception { + String expectedBody = "{\"feeders\":[{\"mrid\":\"abc123\"}]}"; + byte[] expectedBytes = expectedBody.getBytes(StandardCharsets.UTF_8); + + BytesMessage mockBytesMessage = mock(BytesMessage.class); + when(mockBytesMessage.getBodyLength()).thenReturn((long) expectedBytes.length); + when(mockBytesMessage.readBytes(any(byte[].class))).thenAnswer(invocation -> { + byte[] target = invocation.getArgument(0); + System.arraycopy(expectedBytes, 0, target, 0, expectedBytes.length); + return expectedBytes.length; + }); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockBytesMessage); + + Serializable result = client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + + assertThat(result).isEqualTo(expectedBody); + } + + @Test + @DisplayName("GADP-051 (message-type dispatch): an unsupported JMS message type fails loudly with a " + + "SystemException naming the concrete type, rather than silently returning null/empty (data-invariants " + + "Rule 2: no default that could pass as a valid-but-wrong body downstream)") + void unsupportedMessageTypeFailsLoudlyInsteadOfReturningNullOrEmpty() throws Exception { + jakarta.jms.MapMessage mockMapMessage = mock(jakarta.jms.MapMessage.class); + when(mockConsumer.receive(TIMEOUT_MILLIS)).thenReturn(mockMapMessage); + + assertThatThrownBy(() -> client.getResponse(DESTINATION + "-request", DESTINATION, + RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS)) + .isInstanceOf(com.northconcepts.exception.SystemException.class) + .hasMessageContaining("Unsupported JMS message type") + .hasMessageContaining("MapMessage"); + // The dedicated request/reply session and consumer must still be closed + // (leak-safe teardown) even though the reply body was unusable. + verify(mockConsumer, times(1)).close(); + verify(mockSession, times(1)).close(); + } + + private static void setPrivateField(Object target, String name, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } +} diff --git a/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientBoundedReceiveWallClockTest.java b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientBoundedReceiveWallClockTest.java new file mode 100644 index 00000000..3019aa12 --- /dev/null +++ b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientBoundedReceiveWallClockTest.java @@ -0,0 +1,139 @@ +package pnnl.goss.core.client.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.net.URI; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.TransportConnector; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.northconcepts.exception.SystemException; + +import pnnl.goss.core.Client.DESTINATION_TYPE; +import pnnl.goss.core.Client.PROTOCOL; +import pnnl.goss.core.client.GossClient; + +/** + * GOSS-023 regression coverage: the bounded getResponse(...,timeoutMillis) + * overload must actually BLOCK the calling thread for (approximately) its + * timeout budget when no reply arrives, not return early. The existing + * GossClientBoundedReceiveTest is mock-only: it proves receive(timeoutMillis) + * is the method INVOKED, but a Mockito stub returning null returns immediately + * regardless of the argument, so it cannot catch a regression where the real + * JMS call path returns early (the exact defect Hale's GADP-051 runtime-verify + * caught: the retry window collapsed from a designed ~19s to ~4s because each + * bounded receive() returned in ~5ms instead of blocking ~3000ms). + * + * This test uses a real embedded ActiveMQ broker (same pattern as + * pnnl.goss.core.runner's ClientServerTest) and a destination with no + * responder, so the only way getResponse(...,timeoutMillis) can return before + * the deadline is a genuine defect in the blocking path. + */ +public class GossClientBoundedReceiveWallClockTest { + + private static final String DESTINATION = "goss.gridappsd.test.no.responder"; + private static final long TIMEOUT_MILLIS = 1500L; + // Real scheduling jitter (GC, thread wakeup) is normally low tens of ms; + // allow a generous margin without weakening the assertion enough to miss + // the ~5ms-fast-return defect class this test targets. + private static final long TOLERANCE_MILLIS = 400L; + + private BrokerService broker; + private String brokerUri; + private GossClient client; + + @BeforeEach + void startBrokerAndClient() throws Exception { + broker = new BrokerService(); + broker.setBrokerName("goss-023-test-broker"); + broker.setPersistent(false); + broker.setUseJmx(false); + TransportConnector connector = new TransportConnector(); + connector.setUri(new URI("tcp://0.0.0.0:0")); + broker.addConnector(connector); + broker.start(); + broker.waitUntilStarted(); + brokerUri = broker.getTransportConnectors().get(0).getPublishableConnectString(); + + client = new GossClient(PROTOCOL.OPENWIRE, null, brokerUri, "stomp://localhost:0"); + client.createSession(); + } + + @AfterEach + void stopBrokerAndClient() throws Exception { + if (client != null) { + client.close(); + } + if (broker != null) { + broker.stop(); + broker.waitUntilStopped(); + } + } + + @Test + void boundedGetResponseBlocksForItsFullTimeoutBudgetWhenNoReplyArrives() throws Exception { + long t0 = System.currentTimeMillis(); + Object result = client.getResponse("no-op-request", DESTINATION, + pnnl.goss.core.Request.RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS); + long elapsed = System.currentTimeMillis() - t0; + + assertThat(result).isNull(); + assertThat(elapsed) + .withFailMessage( + "getResponse(...,%dms) returned after only %dms; it must block for approximately " + + "its full timeout budget when no reply arrives (GOSS-023: this collapsed the " + + "FieldBusManager retry window from ~19s to ~4s in GADP-051 runtime-verify)", + TIMEOUT_MILLIS, elapsed) + .isGreaterThanOrEqualTo(TIMEOUT_MILLIS - TOLERANCE_MILLIS); + } + + /** + * GOSS-023 root-cause coverage: the actual mechanism behind Hale's GADP-051 + * finding. The single-attempt "no responder" scenario above proves the ordinary + * receive(timeout) blocking path is healthy; it does NOT reproduce a fast + * return. What DOES reproduce a fast return is a genuine JMSException during + * receive() (e.g. the broker/transport going away mid-call): before the fix, + * GossClient's catch (JMSException e) block constructed a SystemException and + * never threw it, silently falling through to "return null" after only the few + * ms it took the provider to report the failure, indistinguishable from a real + * multi-second timeout. That is exactly what collapsed FieldBusManager's + * designed ~19s retry budget to ~4s: every attempt "timed out" almost + * instantly. + * + * A hard transport failure must be surfaced, not swallowed into null: this test + * kills the broker shortly after the receive() call starts, and asserts + * getResponse propagates a SystemException (wrapping the real JMSException) + * rather than returning null. + */ + @Test + void boundedGetResponseSurfacesAGenuineTransportFailureInsteadOfReturningNull() throws Exception { + Thread killer = new Thread(() -> { + try { + Thread.sleep(300); + broker.stop(); + broker.waitUntilStopped(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + killer.start(); + + try { + assertThatThrownBy(() -> client.getResponse("no-op-request", DESTINATION, + pnnl.goss.core.Request.RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS)) + .withFailMessage( + "getResponse(...) must surface a genuine transport failure (a real " + + "JMSException from a dead broker/consumer) rather than silently " + + "returning null, which is indistinguishable from a real timeout " + + "and defeats bounded-retry callers like FieldBusManager (GOSS-023)") + .isInstanceOf(SystemException.class) + .hasCauseInstanceOf(jakarta.jms.JMSException.class); + } finally { + killer.join(); + } + } +} diff --git a/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientSyncReceiveWithActiveListenerTest.java b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientSyncReceiveWithActiveListenerTest.java new file mode 100644 index 00000000..82264c84 --- /dev/null +++ b/pnnl.goss.core/test/pnnl/goss/core/client/test/GossClientSyncReceiveWithActiveListenerTest.java @@ -0,0 +1,130 @@ +package pnnl.goss.core.client.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +import java.net.URI; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.broker.TransportConnector; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import pnnl.goss.core.Client.DESTINATION_TYPE; +import pnnl.goss.core.Client.PROTOCOL; +import pnnl.goss.core.client.GossClient; + +/** + * GADP-051 (session-isolation) regression coverage. + * + *

Root cause verified at runtime cold start: FieldBusManager registers an + * async device-output {@code MessageListener} via {@code subscribe(...)} and + * then issues a synchronous topology {@code getResponse(...)} on the SAME + * {@link pnnl.goss.core.client.GossClient}. Both operations shared the client's + * single JMS {@code Session}, and {@code jakarta.jms} forbids a synchronous + * {@code receive()} on a session that has a {@code MessageListener} attached: + * {@code org.apache.activemq.ActiveMQSession.checkMessageListener} throws + * {@code jakarta.jms.IllegalStateException} with the message + * "Cannot synchronously receive a message when a MessageListener is set". + * That made every topology attempt throw instantly regardless of timing, so + * the caller's bounded-retry window (GADP-051 / GOSS-023) could never succeed. + * + *

This test reproduces the exact interaction against a real embedded broker + * (a mock cannot exercise {@code ActiveMQSession.checkMessageListener}): it + * subscribes an async listener, then calls the bounded {@code getResponse} + * overload against a destination with no responder. + * + *

+ * + *

Same embedded-broker pattern as {@link GossClientBoundedReceiveWallClockTest}. + */ +public class GossClientSyncReceiveWithActiveListenerTest { + + private static final String LISTENER_TOPIC = "goss.gridappsd.test.device.output"; + private static final String SYNC_DESTINATION = "goss.gridappsd.test.topology.no.responder"; + private static final long TIMEOUT_MILLIS = 1200L; + // Scheduling jitter margin, mirrored from GossClientBoundedReceiveWallClockTest. + private static final long TOLERANCE_MILLIS = 400L; + + private BrokerService broker; + private String brokerUri; + private GossClient client; + + @BeforeEach + void startBrokerAndClient() throws Exception { + broker = new BrokerService(); + broker.setBrokerName("gadp-051-session-isolation-test-broker"); + broker.setPersistent(false); + broker.setUseJmx(false); + TransportConnector connector = new TransportConnector(); + connector.setUri(new URI("tcp://0.0.0.0:0")); + broker.addConnector(connector); + broker.start(); + broker.waitUntilStarted(); + brokerUri = broker.getTransportConnectors().get(0).getPublishableConnectString(); + + client = new GossClient(PROTOCOL.OPENWIRE, null, brokerUri, "stomp://localhost:0"); + client.createSession(); + } + + @AfterEach + void stopBrokerAndClient() throws Exception { + if (client != null) { + client.close(); + } + if (broker != null) { + broker.stop(); + broker.waitUntilStopped(); + } + } + + @Test + void syncGetResponseSucceedsWhileAnAsyncListenerIsActiveOnTheClient() throws Exception { + // Attach an async MessageListener on the client, exactly as + // FieldBusManager.publishDeviceOutput() does before it issues its + // synchronous topology request. Pre-fix this poisons the shared session + // for any later synchronous receive() on the same client. + client.subscribe(LISTENER_TOPIC, response -> { + // no-op: presence of the listener is what matters, not its behavior + }); + + long t0 = System.currentTimeMillis(); + + // Pre-fix this throws jakarta.jms.IllegalStateException + // ("Cannot synchronously receive a message when a MessageListener is set") + // within a few ms. Post-fix it runs on a dedicated listener-free session, + // blocks for its timeout budget, and returns null (no responder). + final Object[] result = new Object[1]; + assertThatCode(() -> result[0] = client.getResponse("topology-request", SYNC_DESTINATION, + pnnl.goss.core.Request.RESPONSE_FORMAT.JSON, DESTINATION_TYPE.QUEUE, TIMEOUT_MILLIS)) + .withFailMessage( + "getResponse must complete a synchronous receive even while an async " + + "MessageListener is active on the same client; pre-fix the shared " + + "session threw IllegalStateException (\"Cannot synchronously receive " + + "a message when a MessageListener is set\"), which made every " + + "FieldBusManager topology attempt fail instantly (GADP-051)") + .doesNotThrowAnyException(); + + long elapsed = System.currentTimeMillis() - t0; + + // No responder on SYNC_DESTINATION, so a healthy synchronous receive + // returns null only after blocking for its timeout budget. This is the + // GREEN signal: the receive actually ran (rather than throwing instantly). + assertThat(result[0]).isNull(); + assertThat(elapsed) + .withFailMessage( + "getResponse returned/threw after only %dms; with an active async listener " + + "and a dedicated sync session it must block for approximately its " + + "%dms timeout budget, not fail instantly (GADP-051)", + elapsed, TIMEOUT_MILLIS) + .isGreaterThanOrEqualTo(TIMEOUT_MILLIS - TOLERANCE_MILLIS); + } +}