From 6606aa972f547a684506198a5eec4d28ded0d8bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arnaud=20Cogolu=C3=A8gnes?= <514737+acogoluegnes@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:52:38 +0200 Subject: [PATCH] Defer main loop start until after the header is sent A previous commit moved _frameHandler.initialize(this) before sendHeader() in AMQConnection.start() to fix a race where the I/O thread could dereference a still-null connection reference on a fast broker reply. For SocketFrameHandler, though, initialize() also starts the MainLoop thread, which begins blocking-reading the socket right away. That reader thread now runs concurrently with the header write. For TLS sockets the handshake is lazy and triggered by whichever operation touches the socket first, read or write. When the peer certificate is rejected, the thread that loses the race can see a generic SocketException ("Connection or outbound has closed") instead of the real SSLHandshakeException, since the other thread already tore down the connection. This showed up as a flaky SslContextFactoryTest failure on CI. Split the two concerns: FrameHandler.initialize() still runs before sendHeader(), so the connection reference is set in time for an early reply, but a new FrameHandler.startProcessing() hook - called only after sendHeader() succeeds - is now responsible for actually starting a dedicated reader thread. SocketFrameHandler moves its connection.startMainLoop() call there. Netty and NIO ignore the new hook, since they already read asynchronously and aren't affected by this race. References #2018 --- .../rabbitmq/client/impl/AMQConnection.java | 6 ++++++ .../com/rabbitmq/client/impl/FrameHandler.java | 18 +++++++++++++----- .../client/impl/NettyFrameHandlerFactory.java | 5 +++++ .../client/impl/SocketFrameHandler.java | 14 +++++++++++++- .../impl/nio/SocketChannelFrameHandler.java | 10 ++++++++++ .../client/test/AMQConnectionTest.java | 17 ++++++++++++++++- .../rabbitmq/client/test/BrokenFramesTest.java | 15 +++++++++++++++ 7 files changed, 78 insertions(+), 7 deletions(-) diff --git a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java index 61a25c20ed..08143f01ad 100644 --- a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java +++ b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java @@ -334,6 +334,12 @@ public void start() _frameHandler.close(); throw ioe; } + // Only start reading frames once the header has actually gone out: for + // frame handlers with their own dedicated I/O thread (e.g. blocking IO), + // starting it any earlier would race with sendHeader() over who drives + // the TLS handshake, and the loser gets a confusing generic SocketException + // instead of the real SSLHandshakeException. + this._frameHandler.startProcessing(); AMQP.Connection.Start connStart; AMQP.Connection.Tune connTune = null; diff --git a/src/main/java/com/rabbitmq/client/impl/FrameHandler.java b/src/main/java/com/rabbitmq/client/impl/FrameHandler.java index ce7e8bbfb2..0eba50ec30 100644 --- a/src/main/java/com/rabbitmq/client/impl/FrameHandler.java +++ b/src/main/java/com/rabbitmq/client/impl/FrameHandler.java @@ -55,14 +55,22 @@ default boolean internalHearbeat() { void initialize(AMQConnection connection); - default void finishConnectionNegotiation() { + /** + * Start processing incoming frames, once the connection header has been sent. + * Frame handlers that already read asynchronously (e.g. Netty, NIO) can ignore this, + * as reading may already be under way by the time {@link #initialize(AMQConnection)} runs. + * Frame handlers with their own dedicated reader thread should wait for this call before + * starting it, so that thread does not race {@link #sendHeader()} over the TLS handshake. + */ + void startProcessing(); - } + /** + * Callback to signal the end of the negotiation phase. + */ + void finishConnectionNegotiation(); /** Cap inbound frame payloads, applied once frame_max is negotiated. */ - default void setFrameMax(int frameMax) { - - } + void setFrameMax(int frameMax); /** * Read a {@link Frame} from the underlying data connection. diff --git a/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java index 23f298cd05..2dac670260 100644 --- a/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java +++ b/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java @@ -332,6 +332,11 @@ public void initialize(AMQConnection connection) { this.handler.connection = connection; } + @Override + public void startProcessing() { + // no op + } + @Override public void finishConnectionNegotiation() { maybeRemoveHandler(HANDLER_PROTOCOL_VERSION_MISMATCH); diff --git a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java index 0fef5d0a63..e10b7897ba 100644 --- a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java +++ b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java @@ -61,6 +61,8 @@ public class SocketFrameHandler implements FrameHandler { private volatile int framePayloadLimit; private final IntSupplier payloadLimitSupplier; + private volatile AMQConnection connection; + /** Time to linger before closing the socket forcefully. */ public static final int SOCKET_CLOSING_TIMEOUT = 1; @@ -193,7 +195,17 @@ public void sendHeader() throws IOException { @Override public void initialize(AMQConnection connection) { - connection.startMainLoop(); + this.connection = connection; + } + + @Override + public void startProcessing() { + this.connection.startMainLoop(); + } + + @Override + public void finishConnectionNegotiation() { + // no op } @Override diff --git a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java index 51d5d2f1b3..c10a06883d 100644 --- a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java +++ b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java @@ -83,6 +83,16 @@ public void initialize(AMQConnection connection) { state.setConnection(connection); } + @Override + public void startProcessing() { + // no op + } + + @Override + public void finishConnectionNegotiation() { + // no op + } + @Override public void setFrameMax(int frameMax) { this.state.setFrameMax(frameMax); diff --git a/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java b/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java index 7e5d54b6cd..47f77731f1 100644 --- a/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java +++ b/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java @@ -231,7 +231,7 @@ public Frame readFrame() throws IOException { return null; // simulate a socket timeout } - public void sendHeader() throws IOException { + public void sendHeader() { _numHeadersSent++; } @@ -240,6 +240,21 @@ public void initialize(AMQConnection connection) { connection.startMainLoop(); } + @Override + public void startProcessing() { + + } + + @Override + public void finishConnectionNegotiation() { + + } + + @Override + public void setFrameMax(int frameMax) { + + } + public void setTimeout(int timeoutMs) throws SocketException { this.timeout = timeoutMs; } diff --git a/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java b/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java index 7b7c2b7085..cf5c8533dd 100644 --- a/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java +++ b/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java @@ -226,6 +226,21 @@ public void initialize(AMQConnection connection) { connection.startMainLoop(); } + @Override + public void startProcessing() { + + } + + @Override + public void finishConnectionNegotiation() { + + } + + @Override + public void setFrameMax(int frameMax) { + + } + public void setTimeout(int timeoutMs) { // no need to implement this: don't bother changing the timeout }