Stabilize transport network tests

opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/server/tcp/OpcTcpServerTransport.java

@@ -88,23 +88,19 @@ public synchronized void unbind() {
 
     boundAddresses.clear();
 
-    channelReferences.forEach(
-        channel -> {
-          try {
-            channel.close().sync();
-          } catch (InterruptedException ignored) {
-          }
-        });
+    channelReferences.forEach(channel -> channel.close().syncUninterruptibly());
     channelReferences.clear();
 
+    Set<Channel> childChannels;
     synchronized (childChannelReferences) {
-      childChannelReferences.forEach(
-          channel -> {
-            LoggerFactory.getLogger(getClass()).info("Closing child channel: {}", channel);
-            channel.close();
-          });
+      childChannels = new HashSet<>(childChannelReferences);
       childChannelReferences.clear();
     }
+    childChannels.forEach(
+        channel -> {
+          LoggerFactory.getLogger(getClass()).info("Closing child channel: {}", channel);
+          channel.close().syncUninterruptibly();
+        });
 
     serverBootstrap.reset();
   }

opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/TestPortAllocator.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2026 the Eclipse Milo Authors
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+
+package org.eclipse.milo.opcua.stack.transport;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.channels.OverlappingFileLockException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Allocates loopback ports that remain reserved for Milo tests for the life of the test JVM. */
+public final class TestPortAllocator {
+
+  private static final int MAX_ALLOCATION_ATTEMPTS = 100;
+  private static final Path LOCK_DIRECTORY =
+      Path.of(System.getProperty("java.io.tmpdir"), "milo-test-port-locks");
+
+  private static final List<PortReservation> RESERVATIONS = new ArrayList<>();
+
+  private TestPortAllocator() {}
+
+  /**
+   * Allocate a currently available loopback port.
+   *
+   * <p>The allocation is coordinated with other Milo test JVMs and remains reserved to this JVM.
+   * The caller may bind a server to the returned port immediately.
+   *
+   * @return an available loopback port.
+   * @throws IOException if a port cannot be allocated.
+   */
+  public static synchronized int allocatePort() throws IOException {
+    Files.createDirectories(LOCK_DIRECTORY);
+
+    for (int attempt = 0; attempt < MAX_ALLOCATION_ATTEMPTS; attempt++) {
+      try (var socket = new ServerSocket()) {
+        socket.setReuseAddress(false);
+        socket.bind(new InetSocketAddress(InetAddress.getByName("localhost"), 0));
+
+        int port = socket.getLocalPort();
+        FileChannel channel =
+            FileChannel.open(
+                LOCK_DIRECTORY.resolve(port + ".lock"),
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE);
+
+        FileLock lock = tryLock(channel);
+        if (lock != null) {
+          RESERVATIONS.add(new PortReservation(channel, lock));
+          return port;
+        }
+
+        channel.close();
+      }
+    }
+
+    throw new IOException(
+        "unable to allocate a test port after " + MAX_ALLOCATION_ATTEMPTS + " attempts");
+  }
+
+  private static FileLock tryLock(FileChannel channel) throws IOException {
+    try {
+      return channel.tryLock();
+    } catch (OverlappingFileLockException ignored) {
+      return null;
+    }
+  }
+
+  @SuppressWarnings("unused")
+  private record PortReservation(FileChannel channel, FileLock lock) {}
+}

opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java

@@ -12,6 +12,7 @@
 
 import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.ubyte;
 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.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -92,6 +93,7 @@
 import org.eclipse.milo.opcua.stack.transport.server.ServiceRequestContext;
 import org.eclipse.milo.opcua.stack.transport.server.tcp.OpcTcpServerTransport;
 import org.eclipse.milo.opcua.stack.transport.server.tcp.OpcTcpServerTransportConfig;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -104,11 +106,21 @@ class OpcTcpTransportTest extends SecurityFixture {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(OpcTcpTransportTest.class);
 
+  private InetSocketAddress serverAddress;
+  private OpcServerTransport serverTransport;
+
   static {
     // Required for SecurityPolicy.Aes256_Sha256_RsaPss
     Security.addProvider(new BouncyCastleProvider());
   }
 
+  @AfterEach
+  void unbindServerTransport() throws Exception {
+    if (serverTransport != null) {
+      serverTransport.unbind();
+    }
+  }
+
   private static Stream<Arguments> provideSecurityParameters() {
     return Stream.of(
         Arguments.of(SecurityPolicy.None, MessageSecurityMode.None),
@@ -765,7 +777,7 @@ void openSecureChannelRejectsRepeatIssueOnEstablishedChannel() throws Exception
     }
   }
 
-  private static void createSession(OpcTcpClientTransport transport) throws Exception {
+  private void createSession(OpcTcpClientTransport transport) throws Exception {
     var header =
         new RequestHeader(
             NodeId.NULL_VALUE, DateTime.now(), uint(0), uint(0), null, uint(5_000), null);
@@ -776,7 +788,7 @@ private static void createSession(OpcTcpClientTransport transport) throws Except
             new ApplicationDescription(
                 "", "", LocalizedText.NULL_VALUE, ApplicationType.Client, null, null, null),
             null,
-            "opc.tcp://localhost:12685",
+            endpointUrl(),
             "sessionName",
             ByteString.NULL_VALUE,
             ByteString.NULL_VALUE,
@@ -1056,13 +1068,14 @@ public SecurityKeysListener getSecurityKeysListener() {
           }
         };
 
-    var transport = new OpcTcpServerTransport(config);
-    transport.bind(applicationContext, new InetSocketAddress("localhost", 12685));
-    return transport;
+    serverAddress = new InetSocketAddress("localhost", allocatePort());
+    serverTransport = new OpcTcpServerTransport(config);
+    serverTransport.bind(applicationContext, serverAddress);
+    return serverTransport;
   }
 
-  private static ChannelParameters openRawTcpChannel(Socket socket) throws Exception {
-    socket.connect(new InetSocketAddress("localhost", 12685));
+  private ChannelParameters openRawTcpChannel(Socket socket) throws Exception {
+    socket.connect(serverAddress);
     socket.setSoTimeout(3_000);
 
     EncodingLimits encodingLimits = DefaultEncodingContext.INSTANCE.getEncodingLimits();
@@ -1073,7 +1086,7 @@ private static ChannelParameters openRawTcpChannel(Socket socket) throws Excepti
             encodingLimits.getMaxChunkSize(),
             encodingLimits.getMaxMessageSize(),
             encodingLimits.getMaxChunkCount(),
-            "opc.tcp://localhost:12685");
+            endpointUrl());
 
     ByteBuf helloBuffer = TcpMessageEncoder.encode(hello);
     try {
@@ -1354,15 +1367,15 @@ private EndpointDescription newEndpointDescription(
       SecurityPolicy securityPolicy, MessageSecurityMode messageSecurityMode, byte[] certificate) {
 
     return new EndpointDescription(
-        "opc.tcp://localhost:12685",
+        endpointUrl(),
         new ApplicationDescription(
             "uri:server",
             "productUri",
             LocalizedText.NULL_VALUE,
             ApplicationType.Server,
             null,
             null,
-            new String[] {"opc.tcp://localhost:12685"}),
+            new String[] {endpointUrl()}),
         ByteString.of(certificate),
         messageSecurityMode,
         securityPolicy.getUri(),
@@ -1373,6 +1386,10 @@ private EndpointDescription newEndpointDescription(
         ubyte(0));
   }
 
+  private String endpointUrl() {
+    return "opc.tcp://localhost:" + serverAddress.getPort();
+  }
+
   private static CertificateMaterial nistP256Certificate(String commonName) throws Exception {
     return eccCertificate(SelfSignedCertificateGenerator.generateNistP256KeyPair(), commonName);
   }

opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/package-info.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2026 the Eclipse Milo Authors
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+
+/**
+ * Shared support for transport tests that cross the JVM/network boundary.
+ *
+ * <p>Tests should let the operating system choose listener ports when the bound address is
+ * observable. When an API requires the port before binding, {@link
+ * org.eclipse.milo.opcua.stack.transport.TestPortAllocator} coordinates allocation across Milo test
+ * JVMs and keeps each assigned port exclusive to one JVM for its lifetime.
+ */
+package org.eclipse.milo.opcua.stack.transport;

opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/server/tcp/OpcTcpServerTransportTest.java

@@ -10,17 +10,24 @@
 
 package org.eclipse.milo.opcua.stack.transport.server.tcp;
 
+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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
+import java.net.Socket;
 import java.net.SocketTimeoutException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import org.eclipse.milo.opcua.stack.core.encoding.EncodingContext;
 import org.eclipse.milo.opcua.stack.core.security.CertificateManager;
 import org.eclipse.milo.opcua.stack.core.types.UaRequestMessageType;
@@ -77,6 +84,43 @@ void unbindRejectsLaterReverseConnectWithoutOpeningSocket() throws Exception {
     }
   }
 
+  // Unbind is the server teardown barrier; returning with accepted channels still open lets one
+  // test invocation leak network activity into the next.
+  @Test
+  void unbindClosesAcceptedChannelsBeforeReturning() throws Exception {
+    eventLoop = new NioEventLoopGroup(1);
+    var acceptedChannel = new AtomicReference<Channel>();
+    var channelAccepted = new CountDownLatch(1);
+
+    OpcTcpServerTransportConfig config =
+        OpcTcpServerTransportConfig.newBuilder()
+            .setEventLoop(eventLoop)
+            .setChannelPipelineCustomizer(
+                pipeline -> {
+                  acceptedChannel.set(pipeline.channel());
+                  channelAccepted.countDown();
+                })
+            .build();
+
+    var transport = new OpcTcpServerTransport(config);
+    var bindAddress = new InetSocketAddress("localhost", allocatePort());
+
+    try (var socket = new Socket()) {
+      transport.bind(newServerApplicationContext(), bindAddress);
+      socket.connect(bindAddress);
+
+      assertTrue(channelAccepted.await(3, TimeUnit.SECONDS));
+      Channel channel = acceptedChannel.get();
+
+      transport.unbind();
+
+      assertFalse(channel.isOpen());
+      assertTrue(channel.closeFuture().isDone());
+    } finally {
+      transport.unbind();
+    }
+  }
+
   private static ServerApplicationContext newServerApplicationContext() {
     return new ServerApplicationContext() {