Complete pending placements when plugin host closes

README.md

@@ -6,7 +6,7 @@ StrataProxy 是面向 Minecraft 1.7.10 Forge 的玩家会话代理。项目正
 
 - **玩家会话**：代理负责入服、连接后端、转发、切换后端和断线清理。玩家身份包含 UUID 与连接代次，避免旧连接的异步结果作用到新连接。在线玩家视图由 Session 持有，监听器只保留按 UUID 查找会话的索引。
 - **服务器目录**：静态配置与插件调用同一个注册接口。目录维护地址、声明的容量，以及本代理连接和入服预留的数量；它不采集后端 CPU、TPS、内存等负载。同名重注册生成新句柄，旧异步结果不能修改新条目；同名同地址的旧连接仍占用容量，地址变更后旧连接只计入旧地址。
-- **插件 API**：插件可查询 `PlayerView`、`ServerView`，动态注册后端，并以一个异步回调决定初始落点。空岛分配、实例唤醒和玩法数据由插件自行实现。单次选服失败或超时只结束当前请求，不停用插件；插件返回的拒绝原因会作为登录断开消息发给玩家。
+- **插件 API**：插件可查询 `PlayerView`、`ServerView`，动态注册后端，并以一个异步回调决定初始落点。空岛分配、实例唤醒和玩法数据由插件自行实现。单次选服失败或超时只结束当前请求，不停用插件；宿主关闭时会结束未决选服请求并撤销插件注册。插件返回的拒绝原因会作为登录断开消息发给玩家。
 - **发现方式**：DNS 是独立插件，核心不认识 DNS 或 Agent。配置只启用列出的插件；插件关闭时其注册会清理。
 - **协议热路径**：会话始终保留协议 5 的帧边界，普通数据包不重新编码并尽量复用 Netty `ByteBuf`；Keep Alive ID 在会话内转换，转服后旧后端的迟到回复会被丢弃；转服后只改写少数与玩家实体 ID 相关的包，并按目标通道可写状态控制读取。后端主机名由 Netty 异步解析，避免把 DNS 等待放在玩家 I/O 线程上。
 - **日志**：SLF4J API 与 Logback 运行时。

proxy-core/src/main/java/dev/strataproxy/core/plugin/PluginHost.java

@@ -31,11 +31,14 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.ServiceLoader;
+import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -55,6 +58,7 @@ public final class PluginHost implements AutoCloseable {
     private final ScheduledExecutorService timer;
     private final List<URLClassLoader> classLoaders = new ArrayList<>();
     private final List<LoadedPlugin> plugins = new ArrayList<>();
+    private final Set<PendingPlacement> pendingPlacements = ConcurrentHashMap.newKeySet();
     private State state = State.LOADING;
     private InitialPlacementHandler placementHandler;
 
@@ -173,26 +177,37 @@ public final class PluginHost implements AutoCloseable {
     public CompletionStage<Optional<PlacementDecision>> placeInitial(PlayerView player) {
         Objects.requireNonNull(player, "player");
         final InitialPlacementHandler handler;
+        final CompletableFuture<Optional<PlacementDecision>> result;
+        final AtomicBoolean decided;
+        final PendingPlacement request;
+        final ScheduledFuture<?> timeoutTask;
         synchronized (this) {
             if (state != State.ENABLED) {
                 return CompletableFuture.failedFuture(new IllegalStateException("Plugin host is not enabled"));
             }
             handler = placementHandler;
+            if (handler == null) {
+                return CompletableFuture.completedFuture(Optional.empty());
+            }
+            result = new CompletableFuture<>();
+            decided = new AtomicBoolean();
+            request = new PendingPlacement(result, decided);
+            pendingPlacements.add(request);
+            try {
+                timeoutTask = timer.schedule(
+                        () -> {
+                            if (!decided.compareAndSet(false, true)) return;
+                            result.completeExceptionally(new PlacementTimeoutException(placementTimeout));
+                        }, placementTimeout.toNanos(), TimeUnit.NANOSECONDS);
+            } catch (RuntimeException failure) {
+                pendingPlacements.remove(request);
+                return CompletableFuture.failedFuture(failure);
+            }
         }
-        if (handler == null) {
-            return CompletableFuture.completedFuture(Optional.empty());
-        }
-
-        CompletableFuture<Optional<PlacementDecision>> result = new CompletableFuture<>();
-        AtomicBoolean decided = new AtomicBoolean();
-        var timeoutTask = timer.schedule(
-                () -> {
-                    if (!decided.compareAndSet(false, true)) return;
-                    PlacementTimeoutException timeout = new PlacementTimeoutException(placementTimeout);
-                    result.completeExceptionally(timeout);
-                },
-                placementTimeout.toNanos(), TimeUnit.NANOSECONDS);
-        result.whenComplete((ignored, failure) -> timeoutTask.cancel(false));
+        result.whenComplete((ignored, failure) -> {
+            timeoutTask.cancel(false);
+            pendingPlacements.remove(request);
+        });
         try {
             callbacks.execute(() -> {
                 if (decided.get()) {
@@ -232,6 +247,10 @@ public final class PluginHost implements AutoCloseable {
             return;
         }
         state = State.CLOSED;
+        for (PendingPlacement request : pendingPlacements) {
+            request.decided().set(true);
+            request.result().completeExceptionally(new IllegalStateException("Plugin host closed"));
+        }
         for (int index = plugins.size() - 1; index >= 0; index--) {
             LoadedPlugin loaded = plugins.get(index);
             if (loaded.enabled) {
@@ -302,6 +321,9 @@ public final class PluginHost implements AutoCloseable {
 
     private enum State { LOADING, ENABLED, CLOSED }
 
+    private record PendingPlacement(CompletableFuture<Optional<PlacementDecision>> result,
+                                    AtomicBoolean decided) { }
+
     private static final class LoadedPlugin {
         private final Plugin plugin;
         private final BackendOwner owner;

proxy-core/src/test/java/dev/strataproxy/core/plugin/PluginHostTest.java

@@ -130,6 +130,23 @@ class PluginHostTest {
         host.close();
     }
 
+    @Test
+    void closingHostCompletesPendingPlacementAndIgnoresLatePluginDecision() throws Exception {
+        PendingPlacementPlugin plugin = new PendingPlacementPlugin();
+        PluginHost host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(30));
+        host.load(List.of(plugin));
+        host.enable();
+        CompletableFuture<Optional<PlacementDecision>> placement = host.placeInitial(PLAYER).toCompletableFuture();
+        assertTrue(plugin.invoked.await(5, TimeUnit.SECONDS));
+
+        host.close();
+        var failure = assertThrows(java.util.concurrent.ExecutionException.class,
+                () -> placement.get(1, TimeUnit.SECONDS));
+        assertInstanceOf(IllegalStateException.class, failure.getCause());
+        plugin.decision.complete(PlacementDecision.select("late"));
+        assertTrue(placement.isCompletedExceptionally());
+    }
+
     @Test
     void onlyOneInitialPlacementHandlerCanBeInstalled() {
         PluginHost host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1));
@@ -204,6 +221,20 @@ class PluginHostTest {
         @Override public void onDisable() { disableCount.incrementAndGet(); }
     }
 
+    private static final class PendingPlacementPlugin implements Plugin {
+        private final CountDownLatch invoked = new CountDownLatch(1);
+        private final CompletableFuture<PlacementDecision> decision = new CompletableFuture<>();
+
+        @Override public void onLoad(PluginContext context) { }
+
+        @Override public Optional<InitialPlacementHandler> initialPlacementHandler() {
+            return Optional.of((player, servers) -> {
+                invoked.countDown();
+                return decision;
+            });
+        }
+    }
+
     private static final class BlockingCatalog implements BackendCatalog {
         private final InMemoryBackendCatalog delegate = new InMemoryBackendCatalog();
         private final CountDownLatch registerEntered = new CountDownLatch(1);