From 8093d88a419de17d430ad7ad82ac1d5e390c6aa7 Mon Sep 17 00:00:00 2001 From: Kevin Herron Date: Fri, 23 Oct 2026 10:42:14 -0700 Subject: [PATCH 1/2] Complete client disconnect when the channel closes The ChannelFsm disconnect action in the TCP and WebSocket client transports completed its future from a channelInactive handler that it added to the pipeline from the transport executor. A channel that was already closed never calls a newly added handler, and a closing channel skips one whose handlerAdded has not run yet. disconnect() then waited for the 5 second fallback timeout. This happens when the application calls disconnect() at about the same moment the connection closes, so the ChannelFsm evaluates Disconnect before the ChannelInactive event the close queued. Complete the disconnect from a close future listener instead. Netty notifies it even when it is added after the channel closed, and it runs before channelInactive propagates, which keeps the previous order. The timeout now only closes the channel, and the listener completes the future. Fixes #2134 --- .../OpcWebSocketClientTransport.java | 36 +++--- .../client/tcp/OpcTcpClientTransport.java | 38 +++---- .../client/tcp/OpcTcpTransportTest.java | 105 ++++++++++++++++++ 3 files changed, 132 insertions(+), 47 deletions(-) diff --git a/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java b/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java index 5e9ff4ad11..5eca42b843 100644 --- a/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java +++ b/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java @@ -24,9 +24,6 @@ import io.netty.buffer.PooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ConnectTimeoutException; @@ -45,7 +42,6 @@ import io.netty.handler.ssl.SslContextBuilder; import io.netty.handler.ssl.util.InsecureTrustManagerFactory; import io.netty.util.Timeout; -import io.netty.util.TimerTask; import java.net.ConnectException; import java.net.InetSocketAddress; import java.net.URI; @@ -256,27 +252,21 @@ protected void initChannel(SocketChannel channel) throws Exception { public CompletableFuture disconnect(FsmContext ctx, Channel channel) { var disconnectFuture = new CompletableFuture(); - TimerTask onTimeout = - t -> - channel - .close() - .addListener( - (ChannelFutureListener) channelFuture -> disconnectFuture.complete(null)); - - Timeout timeout = config.getWheelTimer().newTimeout(onTimeout, 5, TimeUnit.SECONDS); + // Close the channel ourselves if the server doesn't close it after CloseSecureChannel. + Timeout timeout = + config.getWheelTimer().newTimeout(t -> channel.close(), 5, TimeUnit.SECONDS); + // Netty notifies a close future listener even when it's added after the channel closed, so + // the disconnect completes without waiting for the timeout when this action runs against a + // channel that is already closed or closing. channel - .pipeline() - .addFirst( - new ChannelInboundHandlerAdapter() { - @Override - public void channelInactive(ChannelHandlerContext channelContext) throws Exception { - // TODO MDC? - logger.debug("channelInactive() disconnect complete"); - timeout.cancel(); - disconnectFuture.complete(null); - super.channelInactive(channelContext); - } + .closeFuture() + .addListener( + f -> { + // TODO MDC? + logger.debug("Channel closed, disconnect complete"); + timeout.cancel(); + disconnectFuture.complete(null); }); var requestHeader = diff --git a/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java b/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java index e1805cc491..6b5951c73f 100644 --- a/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java +++ b/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java @@ -23,16 +23,12 @@ import io.netty.buffer.PooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ConnectTimeoutException; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.util.Timeout; -import io.netty.util.TimerTask; import java.net.ConnectException; import java.net.InetSocketAddress; import java.util.List; @@ -365,29 +361,23 @@ protected void initChannel(SocketChannel ch) { public CompletableFuture disconnect(FsmContext ctx, Channel channel) { var disconnectFuture = new CompletableFuture(); - TimerTask onTimeout = - t -> - channel - .close() - .addListener( - (ChannelFutureListener) channelFuture -> disconnectFuture.complete(null)); - - Timeout timeout = config.getWheelTimer().newTimeout(onTimeout, 5, TimeUnit.SECONDS); + // Close the channel ourselves if the server doesn't close it after CloseSecureChannel. + Timeout timeout = + config.getWheelTimer().newTimeout(t -> channel.close(), 5, TimeUnit.SECONDS); + // Netty notifies a close future listener even when it's added after the channel closed, so + // the disconnect completes without waiting for the timeout when this action runs against a + // channel that is already closed or closing. channel - .pipeline() - .addFirst( - new ChannelInboundHandlerAdapter() { - @Override - public void channelInactive(ChannelHandlerContext channelContext) throws Exception { - try (MDC.MDCCloseable ignored = MDC.putCloseable("instance-id", instanceId)) { - logger.debug("channelInactive() disconnect complete"); - } - - timeout.cancel(); - disconnectFuture.complete(null); - super.channelInactive(channelContext); + .closeFuture() + .addListener( + f -> { + try (MDC.MDCCloseable ignored = MDC.putCloseable("instance-id", instanceId)) { + logger.debug("Channel closed, disconnect complete"); } + + timeout.cancel(); + disconnectFuture.complete(null); }); var requestHeader = diff --git a/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java b/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java index 588102164d..3d674031db 100644 --- a/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java +++ b/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java @@ -14,6 +14,7 @@ import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint; import static org.eclipse.milo.opcua.stack.transport.TestPortAllocator.allocatePort; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -21,6 +22,10 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; import java.io.EOFException; import java.io.IOException; import java.io.InputStream; @@ -38,6 +43,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; @@ -89,6 +96,7 @@ import org.eclipse.milo.opcua.stack.core.types.structured.UserTokenPolicy; import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateBuilder; import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateGenerator; +import org.eclipse.milo.opcua.stack.core.util.Unit; import org.eclipse.milo.opcua.stack.transport.client.ClientApplicationContext; import org.eclipse.milo.opcua.stack.transport.client.uasc.ClientSecureChannel; import org.eclipse.milo.opcua.stack.transport.server.OpcServerTransport; @@ -416,6 +424,103 @@ public UInteger getRequestTimeout() { LOGGER.debug("server transport unbound"); } + // The disconnect action falls back to closing the channel after 5 seconds, so these tests bound + // disconnect() well inside that timeout to show the channel close completed it. + @Test + void disconnectSendsCloseSecureChannelAndCompletesWhenChannelCloses() throws Exception { + OpcServerTransport serverTransport = + bindServerTransport(SecurityPolicy.None, MessageSecurityMode.None); + + var closeSecureChannelSent = new CountDownLatch(1); + + OpcTcpClientTransportConfig config = + OpcTcpClientTransportConfig.newBuilder() + .setChannelPipelineCustomizer( + pipeline -> + pipeline.addFirst( + new ChannelOutboundHandlerAdapter() { + @Override + public void write( + ChannelHandlerContext ctx, Object msg, ChannelPromise promise) + throws Exception { + + if (msg instanceof ByteBuf buffer + && buffer.readableBytes() >= 3 + && buffer.getByte(buffer.readerIndex()) == 'C' + && buffer.getByte(buffer.readerIndex() + 1) == 'L' + && buffer.getByte(buffer.readerIndex() + 2) == 'O') { + + closeSecureChannelSent.countDown(); + } + super.write(ctx, msg, promise); + } + })) + .build(); + + var transport = new OpcTcpClientTransport(config); + + try { + transport.connect(noSecurityClientApplicationContext()).get(); + Channel channel = transport.getChannelFsm().getChannel().get(); + + transport.disconnect().get(2, TimeUnit.SECONDS); + + assertEquals(0, closeSecureChannelSent.getCount()); + assertFalse(channel.isOpen()); + } finally { + serverTransport.unbind(); + } + } + + // When the application calls disconnect() at about the same moment the connection closes, the + // ChannelFsm can evaluate Disconnect before the ChannelInactive event queued by the close. The + // disconnect action then runs against a channel that is already closed. + @Test + void disconnectCompletesWhenChannelClosedBeforeDisconnectActionRuns() throws Exception { + OpcServerTransport serverTransport = + bindServerTransport(SecurityPolicy.None, MessageSecurityMode.None); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try { + var transport = + new OpcTcpClientTransport( + OpcTcpClientTransportConfig.newBuilder().setExecutor(executor).build()); + + transport.connect(noSecurityClientApplicationContext()).get(); + Channel channel = transport.getChannelFsm().getChannel().get(); + + // Hold the ChannelFsm's only executor thread so Disconnect is queued before the + // ChannelInactive event from the close below. + var release = new CountDownLatch(1); + executor.execute( + () -> { + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + CompletableFuture disconnected = transport.disconnect(); + channel.close().sync(); + release.countDown(); + + disconnected.get(2, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + serverTransport.unbind(); + } + } + + private ClientApplicationContext noSecurityClientApplicationContext() { + return clientApplicationContext( + SecurityPolicy.None, + new CertificateMaterial(clientKeyPair, clientCertificate, clientCertificateBytes), + new CertificateMaterial(serverKeyPair, serverCertificate, serverCertificateBytes), + MessageSecurityMode.None); + } + @Test void securityPolicyRejectedOnUnsecuredChannel() throws Exception { // We opened an unsecured channel even though only secured endpoints are available. From 8551915ac91f46d2899210e649ee8b28cba3c219 Mon Sep 17 00:00:00 2001 From: Kevin Herron Date: Fri, 23 Oct 2026 10:53:08 -0700 Subject: [PATCH 2/2] Update netty-channel-fsm to 1.0.4 1.0.3 and 1.0.4 fix cases where the ChannelFsm could leak a channel, get stuck, or end in the wrong state. Events from a previous channel no longer affect a newer connection, connect() and disconnect() requests apply in the order they were made, and disconnect() closes a channel from a connection attempt that is still pending. The public API Milo uses is unchanged. getChannel() now returns a copy of the FSM's future, which the transport already made itself before sending requests. The FSM also reports Disconnecting to Disconnecting transitions after a close, which the transport's Connected-boundary listener ignores. --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index fef4eab1ce..b50a770da8 100644 --- a/pom.xml +++ b/pom.xml @@ -66,7 +66,7 @@ 4.0.5 4.0.9 4.1.139.Final - 1.0.2 + 1.0.4 2.0.20 1.0.1 2.14.0