fix(net): stop BackupServer.close() hanging while bind is in flight

framework/src/main/java/org/tron/common/backup/socket/BackupServer.java

@@ -27,12 +27,12 @@ public class BackupServer implements AutoCloseable {
 
   private BackupManager backupManager;
 
-  private Channel channel;
-
   private volatile boolean shutdown = false;
 
   private final String name = "BackupServer";
-  private ExecutorService executor;
+  private volatile ExecutorService executor;
+
+  private volatile NioEventLoopGroup group;
 
   @Autowired
   public BackupServer(final BackupManager backupManager) {
@@ -41,6 +41,14 @@ public BackupServer(final BackupManager backupManager) {
 
   public void initServer() {
     if (port > 0 && commonParameter.getBackupMembers().size() > 0) {
+      try {
+        // Let close() reach the group before bind() completes.
+        group = new NioEventLoopGroup(1);
+      } catch (RuntimeException e) {
+        // Do not propagate selector failures to block processing.
+        logger.error("Start backup server with port {} failed.", port, e);
+        return;
+      }
       executor = ExecutorServiceManager.newSingleThreadExecutor(name);
       executor.submit(() -> {
         try {
@@ -53,7 +61,6 @@ public void initServer() {
   }
 
   private void start() throws Exception {
-    NioEventLoopGroup group = new NioEventLoopGroup(1);
     try {
       while (!shutdown) {
         Bootstrap b = new Bootstrap();
@@ -73,7 +80,7 @@ public void initChannel(NioDatagramChannel ch)
               }
             });
 
-        channel = b.bind(port).sync().channel();
+        Channel channel = b.bind(port).sync().channel();
 
         logger.info("Backup server started, bind port {}", port);
 
@@ -84,10 +91,15 @@ public void initChannel(NioDatagramChannel ch)
         }
         logger.warn("Restart backup server ...");
       }
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      if (!shutdown) {
+        logger.error("Backup server interrupted.", e);
+      }
     } catch (Exception e) {
       logger.error("Start backup server with port {} failed.", port, e);
     } finally {
-      group.shutdownGracefully().sync();
+      group.shutdownGracefully();
     }
   }
 
@@ -96,14 +108,12 @@ public void close() {
     logger.info("Closing backup server...");
     shutdown = true;
     backupManager.stop();
-    if (channel != null) {
-      try {
-        channel.close().await(10, TimeUnit.SECONDS);
-      } catch (Exception e) {
-        logger.warn("Closing backup server failed.", e);
-      }
+    if (executor != null) {
+      executor.shutdownNow();
+    }
+    if (group != null) {
+      group.shutdownGracefully().awaitUninterruptibly(10, TimeUnit.SECONDS);
     }
-    ExecutorServiceManager.shutdownAndAwaitTermination(executor, name);
     logger.info("Backup server closed.");
   }
 }

framework/src/test/java/org/tron/common/backup/BackupManagerTest.java

@@ -9,7 +9,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.BiFunction;
 import org.junit.After;
@@ -138,15 +137,8 @@ public void testSendKeepAliveMessage() throws Exception {
 
     Thread.sleep(parameter.getKeepAliveInterval() + 1000);//test send KeepAliveMessage
 
-    field = manager.getClass().getDeclaredField("executorService");
-    field.setAccessible(true);
-    ScheduledExecutorService executorService = (ScheduledExecutorService) field.get(manager);
-    executorService.shutdown();
-
-    Field field2 = backupServer.getClass().getDeclaredField("executor");
-    field2.setAccessible(true);
-    ExecutorService executorService2 = (ExecutorService) field2.get(backupServer);
-    executorService2.shutdown();
+    // also stops the manager's keep-alive executor
+    backupServer.close();
 
     Assert.assertEquals(BackupManager.BackupStatusEnum.INIT, manager.getStatus());
   }

framework/src/test/java/org/tron/common/backup/BackupServerTest.java

@@ -1,17 +1,34 @@
 package org.tron.common.backup;
 
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mockStatic;
+
+import io.netty.channel.Channel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.SingleThreadEventExecutor;
+import java.io.IOException;
+import java.net.DatagramSocket;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 import org.junit.rules.Timeout;
+import org.mockito.MockedStatic;
+import org.springframework.test.util.ReflectionTestUtils;
 import org.tron.common.TestConstants;
 import org.tron.common.backup.socket.BackupServer;
+import org.tron.common.es.ExecutorServiceManager;
 import org.tron.common.parameter.CommonParameter;
-import org.tron.common.utils.PublicMethod;
 import org.tron.core.config.args.Args;
 
 
@@ -22,17 +39,18 @@ public class BackupServerTest {
 
   @Rule
   public Timeout globalTimeout = Timeout.seconds(60);
+  private BackupManager backupManager;
   private BackupServer backupServer;
 
   @Before
   public void setUp() throws Exception {
     Args.setParam(new String[]{"-d", temporaryFolder.newFolder().toString()},
         TestConstants.TEST_CONF);
-    CommonParameter.getInstance().setBackupPort(PublicMethod.chooseRandomPort());
+    CommonParameter.getInstance().setBackupPort(freeUdpPort());
     List<String> members = new ArrayList<>();
     members.add("127.0.0.2");
     CommonParameter.getInstance().setBackupMembers(members);
-    BackupManager backupManager = new BackupManager();
+    backupManager = new BackupManager();
     backupManager.init();
     backupServer = new BackupServer(backupManager);
   }
@@ -43,10 +61,104 @@ public void tearDown() {
     Args.clearParam();
   }
 
-  @Test(timeout = 60_000)
-  public void test() throws InterruptedException {
+  @Test
+  public void closeAfterStarted() throws Exception {
     backupServer.initServer();
-    // wait for the server to start so channel is assigned before close() is called
-    Thread.sleep(1000);
+    assertTrue("server did not bind in time", waitUntil(() -> {
+      Channel channel = serverChannel();
+      return channel != null && channel.isActive();
+    }, 30));
+
+    Channel channel = serverChannel();
+    ExecutorService executor =
+        (ExecutorService) ReflectionTestUtils.getField(backupServer, "executor");
+    backupServer.close();
+
+    Assert.assertNotNull(channel);
+    assertFalse("server channel is still open", channel.isOpen());
+    assertTrue("event loop group did not terminate", eventLoopGroup().isTerminated());
+    Assert.assertNotNull(executor);
+    assertTrue("server executor did not terminate",
+        executor.awaitTermination(5, TimeUnit.SECONDS));
+  }
+
+  @Test
+  public void closeWhileBindIsPending() throws Exception {
+    ExecutorService executor = ExecutorServiceManager.newSingleThreadExecutor("BackupServer");
+    CountDownLatch parked = new CountDownLatch(1);
+    CountDownLatch release = new CountDownLatch(1);
+    FutureTask<Void> closed = new FutureTask<>(backupServer::close, null);
+    Thread closer = new Thread(closed, "backup-test-closer");
+    try {
+      try (MockedStatic<ExecutorServiceManager> factory =
+          mockStatic(ExecutorServiceManager.class)) {
+        factory.when(() -> ExecutorServiceManager.newSingleThreadExecutor("BackupServer"))
+            .thenAnswer(invocation -> {
+              // The group exists, but the server task has not been submitted yet.
+              eventLoopGroup().next().execute(() -> {
+                parked.countDown();
+                try {
+                  release.await();
+                } catch (InterruptedException e) {
+                  Thread.currentThread().interrupt();
+                }
+              });
+              assertTrue("event loop did not park", parked.await(5, TimeUnit.SECONDS));
+              return executor;
+            });
+        backupServer.initServer();
+      }
+
+      NioEventLoopGroup group = eventLoopGroup();
+      SingleThreadEventExecutor loop = (SingleThreadEventExecutor) group.next();
+      assertTrue("bind registration did not queue", waitUntil(() -> loop.pendingTasks() > 0, 5));
+      closer.start();
+      assertTrue("close did not shut down the group", waitUntil(group::isShuttingDown, 5));
+      // Let bind proceed only after close() has requested shutdown.
+      release.countDown();
+      closed.get(15, TimeUnit.SECONDS);
+      assertTrue("server executor did not terminate",
+          executor.awaitTermination(5, TimeUnit.SECONDS));
+      assertTrue("event loop group did not terminate", group.isTerminated());
+    } finally {
+      // Release the event loop even when a preparation or shutdown assertion fails.
+      release.countDown();
+      closer.interrupt();
+      executor.shutdownNow();
+      NioEventLoopGroup group = eventLoopGroup();
+      if (group != null) {
+        assertTrue("event loop cleanup failed",
+            group.shutdownGracefully().awaitUninterruptibly(10, TimeUnit.SECONDS));
+      }
+      closer.join(10_000);
+      assertTrue("executor cleanup failed", executor.awaitTermination(10, TimeUnit.SECONDS));
+    }
+  }
+
+  private Channel serverChannel() {
+    Object handler = ReflectionTestUtils.getField(backupManager, "messageHandler");
+    return handler == null ? null : (Channel) ReflectionTestUtils.getField(handler, "channel");
+  }
+
+  private NioEventLoopGroup eventLoopGroup() {
+    return (NioEventLoopGroup) ReflectionTestUtils.getField(backupServer, "group");
+  }
+
+  private boolean waitUntil(BooleanSupplier condition, long timeoutSeconds)
+      throws InterruptedException {
+    long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds);
+    while (!condition.getAsBoolean()) {
+      if (System.nanoTime() - deadline >= 0) {
+        return false;
+      }
+      TimeUnit.MILLISECONDS.sleep(10);
+    }
+    return true;
+  }
+
+  private int freeUdpPort() throws IOException {
+    try (DatagramSocket socket = new DatagramSocket(0)) {
+      return socket.getLocalPort();
+    }
   }
 }