Add central guarded transfer lifecycle

docs/plugins.md

@@ -143,6 +143,12 @@ context.events().subscribe(PlayerDisconnectedEvent.class, event ->
 
 `InitialPlacementHandler` 仍是单提供者选服策略。命令、服务器目录和转服 API 保持原有职责，不通过事件替代。
 
+## 转服状态保护
+
+插件可以在 `onLoad` 或 `onEnable` 中调用 `PluginContext.registerTransferGuard(TransferGuard)` 注册唯一的转服保护器。所有 `Players.transfer(...)` 调用都会经过它，包括第三方插件发起的转服。MoonBridge 在打开目标后端连接、发送登录包之前异步调用 `prepare(TransferContext)`；只有 `ALLOW` 才继续。上下文包含本次尝试 UUID、精确的玩家连接身份、源后端和目标后端。
+
+每次 prepare 已被调用、但转服未成功时，核心都会调用 `failed(context, reason)`，包括明确拒绝、异常和超时；插件可能已经改动了源端状态，所以必须按 transfer UUID 幂等处理迟到的 prepare 结果。核心只在回调返回 `SOURCE_RESTORED` 时恢复旧转发；拒绝、超时、异常或 `DISCONNECT` 都会断开连接。成功切换后调用 `completed(context)` 清理插件状态。所有回调都在 I/O 线程之外运行，每个阶段受 `plugins.eventTimeoutSeconds` 限制。注册保护器但没有配置要求时仍会启用它；`plugins.requiredTransferGuard` 可指定必须注册保护器的插件 provider class，适用于必须 fail closed 的部署。配置了该项但 provider 未加载或未注册时，代理会在开始监听前失败启动。未配置且没有注册保护器时，保持通用 MoonBridge 原有转服行为。
+
 统一事件模型让准入决策和会话通知共享同一类型、监听器、订阅表与派发实现，同时保留各类事件所需的运行策略。准入阶段执行有序且失败关闭的结果聚合；生命周期通知异步 FIFO 尽力交付并与会话处理隔离。没有 PLAY 包事件。目前没有事件 API 的专门基准，不能据此宣称性能收益或无开销。
 
 源码入口：[`PluginContext`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/PluginContext.java)、[`Events`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/event/Events.java)、[`AccessDecision`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/AccessDecision.java)。

proxy-core/src/main/java/dev/moonbridge/app/ProxyConfiguration.java

@@ -218,7 +218,11 @@ public record ProxyConfiguration(String listen, Authentication authentication, L
     }
 
     public record Plugins(String directory, Map<String, Map<String, String>> enabled,
-                          Integer eventTimeoutSeconds) {
+                          Integer eventTimeoutSeconds, String requiredTransferGuard) {
+        public Plugins(String directory, Map<String, Map<String, String>> enabled, Integer eventTimeoutSeconds) {
+            this(directory, enabled, eventTimeoutSeconds, null);
+        }
+
         public Plugins {
             if (directory == null || directory.isBlank()) {
                 throw new IllegalArgumentException("plugin directory is required");
@@ -227,6 +231,9 @@ public record ProxyConfiguration(String listen, Authentication authentication, L
             if (eventTimeoutSeconds < 1 || eventTimeoutSeconds > 30) {
                 throw new IllegalArgumentException("eventTimeoutSeconds must be between 1 and 30");
             }
+            if (requiredTransferGuard != null && requiredTransferGuard.isBlank()) {
+                throw new IllegalArgumentException("requiredTransferGuard must be a plugin provider class name");
+            }
             enabled = enabled == null ? Map.of() : enabled.entrySet().stream().collect(
                     java.util.stream.Collectors.toUnmodifiableMap(
                             entry -> {

proxy-core/src/main/java/dev/moonbridge/app/ProxyMain.java

@@ -93,10 +93,14 @@ public final class ProxyMain {
                      : new BackendControlService(configuration.backendChannel(), catalog,
                      plugins.localMessaging(), proxyEpoch)) {
             if (control != null) plugins.setBackendChannelTransport(control);
+            if (configuration.plugins().requiredTransferGuard() != null) {
+                plugins.requireTransferGuard(configuration.plugins().requiredTransferGuard(), eventTimeout);
+            }
             listener.setMaxConnections(configuration.maxConnections());
             plugins.loadPlugins(pluginDirectory(configPath, configuration.plugins().directory()),
                     configuration.plugins().enabled());
             plugins.enable();
+            listener.setTransferGuard(plugins.transferGuard().orElse(null));
             if (control != null) control.start();
             listener.setEvents(plugins, eventTimeout);
             listener.setPermissions(plugins.permissionService());

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

@@ -40,6 +40,10 @@ import dev.moonbridge.api.ServerRegistration;
 import dev.moonbridge.api.ServerView;
 import dev.moonbridge.api.Servers;
 import dev.moonbridge.api.TransferResult;
+import dev.moonbridge.api.TransferGuard;
+import dev.moonbridge.api.TransferContext;
+import dev.moonbridge.api.TransferGuardDecision;
+import dev.moonbridge.api.TransferFailureDisposition;
 import dev.moonbridge.api.permission.PermissionProvider;
 import dev.moonbridge.api.permission.PermissionResult;
 import net.kyori.adventure.text.Component;
@@ -139,6 +143,9 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
     private volatile BackendChannelTransport backendChannelTransport;
     private ScheduledFuture<?> notificationDropReporter;
     private InitialPlacementHandler placementHandler;
+    private String requiredTransferGuardProvider;
+    private RegisteredTransferGuard registeredTransferGuard;
+    private Duration transferGuardTimeout = Duration.ofSeconds(5);
     private volatile Map<Class<?>, List<EventRegistration<?, ?>>> eventListeners = Map.of();
 
     public PluginHost(BackendCatalog catalog, Players players, Duration placementTimeout) {
@@ -290,6 +297,86 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         }
     }
 
+    /** Requires the named plugin provider to register the transfer guard before startup proceeds. */
+    public synchronized void requireTransferGuard(String providerClassName, Duration timeout) {
+        requireState(State.LOADING);
+        if (providerClassName == null || providerClassName.isBlank()) {
+            throw new IllegalArgumentException("providerClassName must be a plugin provider class name");
+        }
+        Objects.requireNonNull(timeout, "timeout");
+        if (timeout.isNegative() || timeout.isZero() || timeout.compareTo(Duration.ofSeconds(30)) > 0) {
+            throw new IllegalArgumentException("transfer guard timeout must be between 1 and 30 seconds");
+        }
+        this.requiredTransferGuardProvider = providerClassName;
+        this.transferGuardTimeout = timeout;
+    }
+
+    /** Async dispatch facade used by player sessions; every callback runs off Netty I/O loops. */
+    public synchronized Optional<TransferGuard> transferGuard() {
+        if (registeredTransferGuard == null) return Optional.empty();
+        TransferGuard provider = registeredTransferGuard.guard;
+        return Optional.of(new TransferGuard() {
+            @Override public CompletionStage<TransferGuardDecision> prepare(TransferContext transfer) {
+                return invokeTransferGuard(() -> provider.prepare(transfer), "prepare");
+            }
+            @Override public CompletionStage<TransferFailureDisposition> failed(TransferContext transfer,
+                                                                                 String reason) {
+                return invokeTransferGuard(() -> provider.failed(transfer, reason), "failed");
+            }
+            @Override public CompletionStage<Void> completed(TransferContext transfer) {
+                return invokeTransferGuard(() -> provider.completed(transfer), "completed");
+            }
+        });
+    }
+
+    private <T> CompletionStage<T> invokeTransferGuard(
+            java.util.function.Supplier<CompletionStage<T>> callback, String phase) {
+        CompletableFuture<T> result = new CompletableFuture<>();
+        ScheduledFuture<?> deadline;
+        try {
+            deadline = timer.schedule(() -> result.completeExceptionally(
+                    new java.util.concurrent.TimeoutException("transfer guard " + phase + " timed out")),
+                    transferGuardTimeout.toNanos(), TimeUnit.NANOSECONDS);
+            callbacks.execute(() -> {
+                if (result.isDone()) return;
+                try {
+                    CompletionStage<T> stage = Objects.requireNonNull(callback.get(),
+                            "transfer guard " + phase + " stage");
+                    stage.whenComplete((value, failure) -> {
+                        if (failure == null) result.complete(value);
+                        else result.completeExceptionally(failure);
+                    });
+                } catch (Throwable failure) {
+                    result.completeExceptionally(failure);
+                }
+            });
+        } catch (Throwable failure) {
+            return CompletableFuture.failedFuture(failure);
+        }
+        result.whenComplete((ignored, failure) -> deadline.cancel(false));
+        return result;
+    }
+
+    private synchronized void registerTransferGuard(PluginContextImpl context, TransferGuard guard) {
+        Objects.requireNonNull(guard, "guard");
+        synchronized (context) {
+            context.requireActive();
+            if (!context.eventRegistrationOpen) {
+                throw new IllegalStateException("Transfer guard registration is only allowed during onLoad or onEnable");
+            }
+            String provider = plugins.stream().filter(loaded -> loaded.context == context)
+                    .map(loaded -> loaded.plugin.getClass().getName()).findFirst().orElseThrow();
+            if (requiredTransferGuardProvider != null && !requiredTransferGuardProvider.equals(provider)) {
+                throw new IllegalStateException("Only configured transfer guard provider may register: "
+                        + requiredTransferGuardProvider);
+            }
+            if (registeredTransferGuard != null) {
+                throw new IllegalStateException("Only one plugin may register a transfer guard; conflict at " + provider);
+            }
+            registeredTransferGuard = new RegisteredTransferGuard(provider, guard);
+        }
+    }
+
     private static List<String> declaredProviders(Path jar) throws IOException {
         try (JarFile archive = new JarFile(jar.toFile())) {
             var service = archive.getJarEntry("META-INF/services/" + Plugin.class.getName());
@@ -347,6 +434,11 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
                     loaded.context.endEventRegistration();
                 }
             }
+            if (requiredTransferGuardProvider != null && (registeredTransferGuard == null
+                    || !requiredTransferGuardProvider.equals(registeredTransferGuard.providerClassName))) {
+                throw new IllegalStateException("Required transfer guard provider did not register: "
+                        + requiredTransferGuardProvider);
+            }
             PermissionProvider selectedPermissions = null;
             String permissionProviderOwner = null;
             for (LoadedPlugin loaded : plugins) {
@@ -1183,6 +1275,8 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         }
     }
 
+    private record RegisteredTransferGuard(String providerClassName, TransferGuard guard) { }
+
     /** URLClassLoader with a supported, narrowly-scoped hook for plugin-owned extension libraries. */
     private static final class HostPluginClassLoader extends URLClassLoader {
         private final Set<Path> libraries = new HashSet<>();
@@ -1262,6 +1356,9 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         @Override public UUID proxyEpoch() { return PluginHost.this.proxyEpoch; }
 
         @Override public Players players() { return pluginPlayers; }
+        @Override public void registerTransferGuard(TransferGuard guard) {
+            PluginHost.this.registerTransferGuard(this, guard);
+        }
         @Override public Servers servers() { return servers; }
         @Override public Commands commands() { return pluginCommands; }
         @Override public Events events() { return pluginEvents; }
@@ -1285,6 +1382,12 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         private synchronized void deactivateAndRemove() {
             active = false;
             eventRegistrationOpen = false;
+            if (registeredTransferGuard != null && plugins.stream()
+                    .filter(loaded -> loaded.context == this)
+                    .anyMatch(loaded -> registeredTransferGuard.providerClassName.equals(
+                            loaded.plugin.getClass().getName()))) {
+                registeredTransferGuard = null;
+            }
             pluginMessaging.close();
             revokeEventSubscriptions();
             for (RegisteredCommand command : commands.values()) {

proxy-core/src/main/java/dev/moonbridge/core/session/ProxySessionListener.java

@@ -14,6 +14,7 @@ import dev.moonbridge.api.DisconnectResult;
 import dev.moonbridge.api.Players;
 import dev.moonbridge.api.TransferResult;
 import dev.moonbridge.api.TransferStatus;
+import dev.moonbridge.api.TransferGuard;
 import dev.moonbridge.core.auth.MinecraftEncryptionRequest;
 import dev.moonbridge.core.auth.SessionVerifier;
 import dev.moonbridge.core.backend.BackendCatalog;
@@ -89,6 +90,7 @@ public final class ProxySessionListener implements Players {
     private java.util.function.BiFunction<PlayerView, String, List<String>> playerCommandNames;
     private java.util.function.BiPredicate<PlayerView, String> commandVisibility = (player, suggestion) -> true;
     private PermissionService permissions;
+    private volatile TransferGuard transferGuard;
     private java.util.function.BiFunction<PlayerView, String, Optional<CompletionStage<List<String>>>> completions;
 
     private EventDispatcher events;
@@ -183,6 +185,16 @@ public final class ProxySessionListener implements Players {
         this.placement = Objects.requireNonNull(placement, "placement");
     }
 
+    /** Installs the single plugin-registered state guard before accepting players. */
+    public synchronized void setTransferGuard(TransferGuard transferGuard) {
+        if (started || closed || this.transferGuard != null) {
+            throw new IllegalStateException("Transfer guard must be configured once before listener start");
+        }
+        this.transferGuard = transferGuard;
+    }
+
+    TransferGuard transferGuard() { return transferGuard; }
+
     /** Configures the host boot epoch and per-backend forwarding keys before accepting clients. */
     public synchronized void setSessionBinding(UUID proxyEpoch, Map<String, String> secretsByBackend) {
         if (started || closed || listener != null) {

proxy-core/src/main/java/dev/moonbridge/core/session/Session.java

@@ -9,6 +9,10 @@ import dev.moonbridge.api.MessageResult;
 import dev.moonbridge.api.DisconnectResult;
 import dev.moonbridge.api.TransferResult;
 import dev.moonbridge.api.TransferStatus;
+import dev.moonbridge.api.TransferContext;
+import dev.moonbridge.api.TransferGuard;
+import dev.moonbridge.api.TransferGuardDecision;
+import dev.moonbridge.api.TransferFailureDisposition;
 import dev.moonbridge.core.auth.AuthenticatedEncryption;
 import dev.moonbridge.core.auth.MinecraftCipherDecoder;
 import dev.moonbridge.core.auth.MinecraftCipherEncoder;
@@ -50,6 +54,7 @@ import java.security.GeneralSecurityException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Optional;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -1062,10 +1067,37 @@ final class Session extends ChannelInboundHandlerAdapter {
             result.complete(TransferResult.of(TransferStatus.SERVER_UNAVAILABLE));
             return;
         }
-        TransferAttempt attempt = new TransferAttempt(target, result, relay);
+        TransferContext guardContext = owner.transferGuard() == null ? null : new TransferContext(
+                UUID.randomUUID(), identity, selected.handle().id().value(), target.handle().id().value());
+        TransferAttempt attempt = new TransferAttempt(target, result, relay, guardContext);
         transfer = attempt;
+        if (guardContext != null) {
+            CompletionStage<TransferGuardDecision> prepared;
+            try {
+                prepared = Objects.requireNonNull(owner.transferGuard().prepare(guardContext),
+                        "transfer guard prepare stage");
+            } catch (Throwable failure) {
+                failTransfer(attempt, "transfer guard preparation failed");
+                return;
+            }
+            withGuardDeadline(prepared, "prepare").whenComplete((decision, failure) -> frontend.eventLoop().execute(() -> {
+                if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return;
+                if (failure != null || decision != TransferGuardDecision.ALLOW) {
+                    failTransfer(attempt, failure == null ? "transfer guard rejected transfer"
+                            : "transfer guard preparation failed");
+                    return;
+                }
+                openReplacementBackend(attempt);
+            }));
+        } else {
+            openReplacementBackend(attempt);
+        }
+    }
+
+    private void openReplacementBackend(TransferAttempt attempt) {
+        if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return;
         LOGGER.debug("Starting replacement backend connection for player {} to {}",
-                view.username(), target.address());
+                view.username(), attempt.target.address());
         attempt.candidate = new TransferCandidate(identity.playerId(), view.username(), new TransferCandidate.Listener() {
             @Override public void ready(TransferCandidate candidate) { candidateReady(attempt); }
             @Override public void failed(TransferCandidate candidate, String reason) { failTransfer(attempt, reason); }
@@ -1104,6 +1136,19 @@ final class Session extends ChannelInboundHandlerAdapter {
         }
     }
 
+    private <T> CompletionStage<T> withGuardDeadline(CompletionStage<T> source, String phase) {
+        CompletableFuture<T> bounded = new CompletableFuture<>();
+        ScheduledFuture<?> deadline = frontend.eventLoop().schedule(() -> bounded.completeExceptionally(
+                new java.util.concurrent.TimeoutException("transfer guard " + phase + " timed out")),
+                owner.eventTimeout().toNanos(), TimeUnit.NANOSECONDS);
+        source.whenComplete((value, failure) -> {
+            if (failure == null) bounded.complete(value);
+            else bounded.completeExceptionally(failure);
+        });
+        bounded.whenComplete((ignored, failure) -> deadline.cancel(false));
+        return bounded;
+    }
+
     private void tryStartPendingTransfer() {
         PendingTransfer waiting = pendingTransfer;
         if (disconnecting || waiting == null || relay == null || !playObservation.ready().isDone()) return;
@@ -1188,15 +1233,14 @@ final class Session extends ChannelInboundHandlerAdapter {
         if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return;
         attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> {
             if (transfer != attempt || attempt.result.isDone()) return;
-            attempt.result.complete(TransferResult.failed("replacement backend cutover timed out"));
-            closePair();
+            attempt.forceDisconnect = true;
+            failTransfer(attempt, "replacement backend cutover timed out");
         }, owner.transferCutoverTimeout().toNanos(), TimeUnit.NANOSECONDS);
         try {
             attempt.clientBuffer = installTransferBuffer(frontend, "transfer-client-buffer");
             attempt.oldBackendBuffer = installTransferBuffer(backend, "transfer-old-backend-buffer");
         } catch (RuntimeException failure) {
             failTransfer(attempt, "could not buffer frames for transfer");
-            closePair();
             return;
         }
         attempt.pauseInProgress = true;
@@ -1206,8 +1250,9 @@ final class Session extends ChannelInboundHandlerAdapter {
         pause.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
             attempt.pauseInProgress = false;
             if (failure != null) {
+                attempt.forceDisconnect = true;
                 if (!attempt.finished) failTransfer(attempt, "old backend stopped during transfer");
-                closePair();
+                else resumeAfterFailedTransfer(attempt);
                 return;
             }
             attempt.paused = true;
@@ -1229,14 +1274,14 @@ final class Session extends ChannelInboundHandlerAdapter {
                         if (detachFailure != null) {
                             boolean oldRelayStillAttached = frontend.pipeline().get("raw-relay") != null
                                     && backend.pipeline().get("raw-relay") != null;
+                            if (!oldRelayStillAttached) attempt.forceDisconnect = true;
                             failTransfer(attempt, "could not detach old backend relay");
-                            if (!oldRelayStillAttached) closePair();
                             return;
                         }
                         if (closed.get() || disconnecting || !attempt.channel.isActive()) {
                             if (disconnecting) return;
+                            attempt.forceDisconnect = true;
                             failTransfer(attempt, "replacement backend closed during transfer");
-                            closePair();
                             return;
                         }
                         attempt.detached = true;
@@ -1268,16 +1313,16 @@ final class Session extends ChannelInboundHandlerAdapter {
         try {
             attempt.candidate.handOff();
         } catch (RuntimeException invalid) {
+            attempt.forceDisconnect = true;
             failTransfer(attempt, "replacement backend was not ready");
-            closePair();
             return;
         }
         removeHandshakeCodecs(attempt.channel).whenComplete((ignored, failure) ->
                 frontend.eventLoop().execute(() -> {
                     if (failure != null || closed.get() || disconnecting || !attempt.channel.isActive()) {
                         if (disconnecting) return;
+                        attempt.forceDisconnect = true;
                         failTransfer(attempt, "replacement backend pipeline failed");
-                        closePair();
                         return;
                     }
                     installCandidateRelay(attempt);
@@ -1301,15 +1346,15 @@ final class Session extends ChannelInboundHandlerAdapter {
             RawRelay.Link observedLink = next;
             nextObservation.ready().whenComplete((ignored, failure) -> observedLink.stopObserving());
         } catch (RuntimeException failure) {
+            attempt.forceDisconnect = true;
             failTransfer(attempt, "could not attach replacement relay");
-            closePair();
             return;
         }
         next.ready().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
             if (failure != null || closed.get() || disconnecting) {
                 if (disconnecting) return;
+                attempt.forceDisconnect = true;
                 failTransfer(attempt, "replacement relay was not ready");
-                closePair();
                 return;
             }
             ByteBuf opening;
@@ -1317,15 +1362,15 @@ final class Session extends ChannelInboundHandlerAdapter {
                 keepAlives.switchBackend();
                 opening = transferOpening(attempt);
             } catch (RuntimeException malformed) {
+                attempt.forceDisconnect = true;
                 failTransfer(attempt, "could not prepare replacement world transition");
-                closePair();
                 return;
             }
             frontend.writeAndFlush(opening).addListener(write -> frontend.eventLoop().execute(() -> {
                 if (!write.isSuccess() || closed.get() || disconnecting || !attempt.channel.isActive()) {
                     if (disconnecting) return;
+                    attempt.forceDisconnect = true;
                     failTransfer(attempt, "could not send replacement world transition");
-                    closePair();
                     return;
                 }
                 Channel oldBackend = backend;
@@ -1345,7 +1390,8 @@ final class Session extends ChannelInboundHandlerAdapter {
                 try {
                     attempt.clientBuffer.drainAndRemove();
                 } catch (RuntimeException replayFailure) {
-                    closePair();
+                    attempt.forceDisconnect = true;
+                    failTransfer(attempt, "could not replay buffered client frames");
                     return;
                 }
                 if (closed.get()) return;
@@ -1355,15 +1401,15 @@ final class Session extends ChannelInboundHandlerAdapter {
                     attempt.cutoverDeadline.cancel(false);
                     attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> {
                         if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return;
+                        attempt.forceDisconnect = true;
                         failTransfer(attempt, "replacement Forge handshake timed out");
-                        closePair();
                     }, FORGE_TRANSFER_HANDSHAKE_TIMEOUT.toNanos(), TimeUnit.NANOSECONDS);
                     CompletableFuture.allOf(nextObservation.ready(), attempt.frameState.worldReady())
                             .whenComplete((negotiated, negotiationFailure) -> frontend.eventLoop().execute(() -> {
                                 if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return;
                                 if (negotiationFailure != null || !attempt.channel.isActive()) {
+                                    attempt.forceDisconnect = true;
                                     failTransfer(attempt, "replacement Forge handshake failed");
-                                    closePair();
                                 } else finishTransfer(attempt);
                             }));
                 } else finishTransfer(attempt);
@@ -1376,6 +1422,20 @@ final class Session extends ChannelInboundHandlerAdapter {
         transfer = null;
         attempt.finished = true;
         attempt.cutoverDeadline.cancel(false);
+        if (attempt.guardContext != null) {
+            TransferGuard guard = owner.transferGuard();
+            if (guard != null) {
+                try {
+                    guard.completed(attempt.guardContext).whenComplete((ignored, failure) -> {
+                        if (failure != null) LOGGER.warn("Transfer guard completion callback failed for transfer {}",
+                                attempt.guardContext.transferId(), failure);
+                    });
+                } catch (Throwable failure) {
+                    LOGGER.warn("Transfer guard completion callback failed for transfer {}",
+                            attempt.guardContext.transferId(), failure);
+                }
+            }
+        }
         attempt.result.complete(TransferResult.of(TransferStatus.NETWORK_READY));
     }
 
@@ -1444,14 +1504,49 @@ final class Session extends ChannelInboundHandlerAdapter {
                 attempt.channel != null && attempt.channel.isActive());
         attempt.finished = true;
         attempt.failureReason = reason;
+        if (attempt.detached) attempt.forceDisconnect = true;
         if (attempt.channel != null) attempt.channel.close();
         if (attempt.candidate != null) attempt.candidate.close();
         if (attempt.frameState != null) attempt.frameState.close();
-        if (disconnecting) return;
-        if (!attempt.pauseInProgress) resumeAfterFailedTransfer(attempt);
+        TransferGuard guard = owner.transferGuard();
+        if (attempt.guardContext == null || guard == null) {
+            attempt.guardResolved = true;
+            if (attempt.forceDisconnect) closePair();
+            else if (!attempt.pauseInProgress) resumeAfterFailedTransfer(attempt);
+            return;
+        }
+        attempt.abortStarted = true;
+        CompletionStage<TransferFailureDisposition> aborted;
+        try {
+            aborted = Objects.requireNonNull(guard.failed(attempt.guardContext, reason),
+                    "transfer guard failed stage");
+        } catch (Throwable failure) {
+            attempt.guardResolved = true;
+            attempt.forceDisconnect = true;
+            if (!attempt.pauseInProgress) closePair();
+            return;
+        }
+        withGuardDeadline(aborted, "failed").whenComplete((disposition, failure) -> {
+            Runnable resume = () -> {
+                if (transfer != attempt || closed.get()) return;
+                attempt.guardResolved = true;
+                if (failure != null || disposition != TransferFailureDisposition.SOURCE_RESTORED) {
+                    attempt.forceDisconnect = true;
+                }
+                if (attempt.forceDisconnect || disconnecting) closePair();
+                else if (!attempt.pauseInProgress) resumeAfterFailedTransfer(attempt);
+            };
+            try { frontend.eventLoop().execute(resume); }
+            catch (RejectedExecutionException ignored) { closePair(); }
+        });
     }
 
     private void resumeAfterFailedTransfer(TransferAttempt attempt) {
+        if (!attempt.guardResolved || closed.get()) return;
+        if (attempt.forceDisconnect || attempt.detached) {
+            closePair();
+            return;
+        }
         if (attempt.paused && !attempt.detached && !closed.get() && !disconnecting) {
             attempt.oldRelay.resume().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
                 if (failure != null) closePair();
@@ -1477,6 +1572,18 @@ final class Session extends ChannelInboundHandlerAdapter {
     }
 
     void closePair() {
+        if (!frontend.eventLoop().inEventLoop() && !frontend.eventLoop().isTerminated()) {
+            try {
+                frontend.eventLoop().execute(this::closePair);
+                return;
+            } catch (RejectedExecutionException ignored) { }
+        }
+        TransferAttempt active = transfer;
+        if (!closed.get() && active != null && active.guardContext != null && !active.abortStarted) {
+            active.forceDisconnect = true;
+            failTransfer(active, "player session closed during transfer");
+            return;
+        }
         if (!closed.compareAndSet(false, true)) return;
         Runnable cleanup = () -> {
             if (tabCompletion != null) tabCompletion.close();
@@ -1505,13 +1612,22 @@ final class Session extends ChannelInboundHandlerAdapter {
             }
             TransferAttempt activeTransfer = transfer;
             if (activeTransfer != null) {
+                TransferGuard guard = owner.transferGuard();
+                if (activeTransfer.guardContext != null && !activeTransfer.abortStarted && guard != null) {
+                    activeTransfer.abortStarted = true;
+                    try {
+                        guard.failed(activeTransfer.guardContext, "player session closed during transfer");
+                    } catch (Throwable ignored) { }
+                }
                 transfer = null;
                 activeTransfer.finished = true;
                 if (activeTransfer.cutoverDeadline != null) activeTransfer.cutoverDeadline.cancel(false);
                 if (activeTransfer.channel != null) activeTransfer.channel.close();
                 if (activeTransfer.candidate != null) activeTransfer.candidate.close();
                 if (activeTransfer.frameState != null) activeTransfer.frameState.close();
-                activeTransfer.result.complete(TransferResult.failed("player session closed during transfer"));
+                String transferFailure = activeTransfer.failureReason == null
+                        ? "player session closed during transfer" : activeTransfer.failureReason;
+                activeTransfer.result.complete(TransferResult.failed(transferFailure));
             }
             if (identityClaimed) {
                 owner.releasePermissions(identity);
@@ -1576,6 +1692,7 @@ final class Session extends ChannelInboundHandlerAdapter {
         private final BackendView target;
         private final CompletableFuture<TransferResult> result;
         private final RawRelay.Link oldRelay;
+        private final TransferContext guardContext;
         private TransferCandidate candidate;
         private TransferFrameHandler.State frameState;
         private TransferFrameBuffer clientBuffer;
@@ -1585,14 +1702,18 @@ final class Session extends ChannelInboundHandlerAdapter {
         private boolean paused;
         private boolean detached;
         private boolean finished;
+        private boolean guardResolved;
+        private boolean abortStarted;
+        private boolean forceDisconnect;
         private String failureReason;
         private ScheduledFuture<?> cutoverDeadline;
 
-        private TransferAttempt(BackendView target,
-                                CompletableFuture<TransferResult> result, RawRelay.Link oldRelay) {
+        private TransferAttempt(BackendView target, CompletableFuture<TransferResult> result,
+                                RawRelay.Link oldRelay, TransferContext guardContext) {
             this.target = target;
             this.result = result;
             this.oldRelay = oldRelay;
+            this.guardContext = guardContext;
         }
     }
 

proxy-core/src/main/resources/config/moonbridge.yml

@@ -12,6 +12,8 @@ plugins:
   directory: "../plugins"
   eventTimeoutSeconds: 5
   enabled: {}
+  # Optional fail-closed deployment requirement; omit for generic MoonBridge installs.
+  # requiredTransferGuard: "example.plugin.TransferGuardPlugin"
   # To enable the bundled native LuckPerms platform, replace the empty map above:
   # enabled:
   #   dev.moonbridge.luckperms.LuckPermsMoonBridgePlugin:

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

@@ -68,6 +68,17 @@ class PluginHostTest {
     private static final PlayerView PLAYER = new PlayerView(
             new PlayerIdentity(UUID.randomUUID(), 1), "TestPlayer", Optional.empty());
 
+    @Test
+    void configuredRequiredTransferGuardMustRegisterBeforeEnable() {
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1))) {
+            host.requireTransferGuard("example.RequiredTransferGuard", Duration.ofSeconds(1));
+            host.load(List.of(new Plugin() {
+                @Override public void onLoad(PluginContext context) { }
+            }));
+            assertThrows(IllegalStateException.class, host::enable);
+        }
+    }
+
     @Test
     void accessChecksRunInLoadOrderOffCallerAndStopAtFirstDenial() throws Exception {
         var calls = new java.util.concurrent.CopyOnWriteArrayList<String>();

proxy-core/src/test/java/dev/moonbridge/core/session/SessionTransferTest.java

@@ -3,6 +3,10 @@ package dev.moonbridge.core.session;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import dev.moonbridge.api.PlacementDecision;
 import dev.moonbridge.api.TransferStatus;
+import dev.moonbridge.api.TransferGuard;
+import dev.moonbridge.api.TransferContext;
+import dev.moonbridge.api.TransferGuardDecision;
+import dev.moonbridge.api.TransferFailureDisposition;
 import dev.moonbridge.api.event.Event;
 import dev.moonbridge.api.event.PlayerDisconnectedEvent;
 import dev.moonbridge.api.event.ServerConnectedEvent;
@@ -58,6 +62,135 @@ final class SessionTransferTest {
     private static final UUID PLAYER_ID = UUID.nameUUIDFromBytes(
             ("OfflinePlayer:" + USERNAME).getBytes(StandardCharsets.UTF_8));
 
+    @Test
+    void guardAbortMustRestoreSourceBeforeThirdPartyTransferResumesRelay() throws Exception {
+        var catalog = new InMemoryBackendCatalog();
+        try (ServerSocket oldServer = server(); ServerSocket newServer = server()) {
+            var oldReceived = new CompletableFuture<byte[]>();
+            var abortEntered = new CompletableFuture<TransferContext>();
+            var abort = new CompletableFuture<TransferFailureDisposition>();
+            backendThread(oldServer, () -> {
+                try (Socket socket = oldServer.accept()) {
+                    socket.setSoTimeout(5000);
+                    var input = new DataInputStream(socket.getInputStream());
+                    var output = new DataOutputStream(socket.getOutputStream());
+                    acceptLogin(input, output, 0);
+                    byte[] request = readFrame(input);
+                    oldReceived.complete(request);
+                    writeFrame(output, new byte[]{0x03, 0x66});
+                    while (input.read() != -1) { }
+                } catch (Throwable failure) { oldReceived.completeExceptionally(failure); }
+            });
+            register(catalog, "old", oldServer);
+            register(catalog, "new", newServer);
+            var listener = listener(catalog);
+            listener.setTransferGuard(new TransferGuard() {
+                @Override public CompletionStage<TransferGuardDecision> prepare(TransferContext transfer) {
+                    return CompletableFuture.completedFuture(TransferGuardDecision.REJECT);
+                }
+
+                @Override public CompletionStage<TransferFailureDisposition> failed(
+                        TransferContext transfer, String reason) {
+                    abortEntered.complete(transfer);
+                    return abort;
+                }
+            });
+            try {
+                int port = ((InetSocketAddress) listener.start().toCompletableFuture()
+                        .get(5, TimeUnit.SECONDS).localAddress()).getPort();
+                try (Socket client = new Socket(InetAddress.getLoopbackAddress(), port)) {
+                    client.setSoTimeout(5000);
+                    var input = new DataInputStream(client.getInputStream());
+                    var output = new DataOutputStream(client.getOutputStream());
+                    sendLogin(output);
+                    readFrame(input); readFrame(input); readFrame(input);
+                    var player = awaitPlayer(listener);
+
+                    // Calling the public Players API directly represents any plugin caller.
+                    var transfer = listener.transfer(player.identity(), "new").toCompletableFuture();
+                    TransferContext context = abortEntered.get(5, TimeUnit.SECONDS);
+                    assertEquals(player.identity(), context.player());
+                    assertEquals("old", context.sourceBackend());
+                    assertEquals("new", context.targetBackend());
+                    assertTrue(!transfer.isDone(), "transfer must wait for the recovery callback");
+                    newServer.setSoTimeout(150);
+                    assertThrows(java.net.SocketTimeoutException.class, newServer::accept,
+                            "target login must not start after guard rejection");
+
+                    abort.complete(TransferFailureDisposition.SOURCE_RESTORED);
+                    assertEquals(TransferStatus.FAILED, transfer.get(5, TimeUnit.SECONDS).status());
+                    writeFrame(output, new byte[]{0x01, 0x55});
+                    assertArrayEquals(new byte[]{0x01, 0x55}, oldReceived.get(5, TimeUnit.SECONDS));
+                    assertArrayEquals(new byte[]{0x03, 0x66}, readFrame(input));
+                }
+            } finally { listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS); }
+        }
+    }
+
+    @Test
+    void prepareTimeoutStillWaitsForGuardRecoveryBeforeResumingSource() throws Exception {
+        var catalog = new InMemoryBackendCatalog();
+        try (ServerSocket oldServer = server(); ServerSocket newServer = server()) {
+            var oldReceived = new CompletableFuture<byte[]>();
+            var abortEntered = new CompletableFuture<Void>();
+            var abort = new CompletableFuture<TransferFailureDisposition>();
+            backendThread(oldServer, () -> {
+                try (Socket socket = oldServer.accept()) {
+                    socket.setSoTimeout(5000);
+                    var input = new DataInputStream(socket.getInputStream());
+                    var output = new DataOutputStream(socket.getOutputStream());
+                    acceptLogin(input, output, 0);
+                    byte[] request = readFrame(input);
+                    oldReceived.complete(request);
+                    writeFrame(output, new byte[]{0x03, 0x66});
+                    while (input.read() != -1) { }
+                } catch (Throwable failure) { oldReceived.completeExceptionally(failure); }
+            });
+            register(catalog, "old", oldServer);
+            register(catalog, "new", newServer);
+            var listener = listener(catalog);
+            listener.setEvents(new EventDispatcher() {
+                @Override public boolean hasSubscribers(Class<?> eventType) { return false; }
+                @Override public <R> CompletionStage<R> dispatch(Event<R> event) {
+                    return CompletableFuture.failedFuture(new AssertionError("no events expected"));
+                }
+            }, Duration.ofMillis(500));
+            listener.setTransferGuard(new TransferGuard() {
+                @Override public CompletionStage<TransferGuardDecision> prepare(TransferContext transfer) {
+                    return new CompletableFuture<>(); // Simulate a handoff that may have committed without its ACK.
+                }
+                @Override public CompletionStage<TransferFailureDisposition> failed(
+                        TransferContext transfer, String reason) {
+                    abortEntered.complete(null);
+                    return abort;
+                }
+            });
+            try {
+                int port = ((InetSocketAddress) listener.start().toCompletableFuture()
+                        .get(5, TimeUnit.SECONDS).localAddress()).getPort();
+                try (Socket client = new Socket(InetAddress.getLoopbackAddress(), port)) {
+                    client.setSoTimeout(5000);
+                    var input = new DataInputStream(client.getInputStream());
+                    var output = new DataOutputStream(client.getOutputStream());
+                    sendLogin(output);
+                    readFrame(input); readFrame(input); readFrame(input);
+                    var player = awaitPlayer(listener);
+                    var transfer = listener.transfer(player.identity(), "new").toCompletableFuture();
+                    abortEntered.get(5, TimeUnit.SECONDS);
+                    assertTrue(!transfer.isDone(), "timed out prepare must await recovery");
+                    newServer.setSoTimeout(100);
+                    assertThrows(java.net.SocketTimeoutException.class, newServer::accept,
+                            "target login must not start after prepare timeout");
+                    abort.complete(TransferFailureDisposition.SOURCE_RESTORED);
+                    assertEquals(TransferStatus.FAILED, transfer.get(5, TimeUnit.SECONDS).status());
+                    writeFrame(output, new byte[]{0x01, 0x55});
+                    assertArrayEquals(new byte[]{0x01, 0x55}, oldReceived.get(5, TimeUnit.SECONDS));
+                    assertArrayEquals(new byte[]{0x03, 0x66}, readFrame(input));
+                }
+            } finally { listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS); }
+        }
+    }
+
     @Test
     void clientFrameWaitsForWorldTransitionDuringCutover() throws Exception {
         var catalog = new InMemoryBackendCatalog();
@@ -459,7 +592,8 @@ final class SessionTransferTest {
                     var transfer = listener.transfer(player.identity(), "new").toCompletableFuture()
                             .get(5, TimeUnit.SECONDS);
                     assertEquals(TransferStatus.FAILED, transfer.status());
-                    assertTrue(transfer.detail().orElse("").contains("cutover timed out"));
+                    String detail = transfer.detail().orElse("");
+                    assertTrue(detail.contains("cutover timed out") || detail.contains("old backend stopped"), detail);
                     assertEquals(-1, input.read());
                     oldClosed.get(5, TimeUnit.SECONDS);
                     newClosed.get(5, TimeUnit.SECONDS);

proxy-plugin-api/src/main/java/dev/moonbridge/api/PluginContext.java

@@ -16,6 +16,9 @@ public interface PluginContext {
 
     Players players();
 
+    /** Registers this plugin as the proxy's transfer state guard. Registration is unique. */
+    void registerTransferGuard(TransferGuard guard);
+
     Servers servers();
 
     Commands commands();

proxy-plugin-api/src/main/java/dev/moonbridge/api/TransferContext.java

@@ -0,0 +1,22 @@
+package dev.moonbridge.api;
+
+import java.util.Objects;
+import java.util.UUID;
+
+/** Immutable identity and route snapshot for one backend transfer attempt. */
+public record TransferContext(UUID transferId, PlayerIdentity player, String sourceBackend,
+                              String targetBackend) {
+    public TransferContext {
+        Objects.requireNonNull(transferId, "transferId");
+        Objects.requireNonNull(player, "player");
+        validateName(sourceBackend, "sourceBackend");
+        validateName(targetBackend, "targetBackend");
+    }
+
+    private static void validateName(String value, String name) {
+        Objects.requireNonNull(value, name);
+        if (value.isBlank() || !value.equals(value.trim()) || value.length() > 128) {
+            throw new IllegalArgumentException(name + " must be a nonblank backend name without surrounding whitespace");
+        }
+    }
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/TransferFailureDisposition.java

@@ -0,0 +1,9 @@
+package dev.moonbridge.api;
+
+/** Safe action after a guarded transfer failed. */
+public enum TransferFailureDisposition {
+    /** The exact source backend has been restored and may receive the resumed relay. */
+    SOURCE_RESTORED,
+    /** Source ownership is uncertain or was not restored; MoonBridge must close the player. */
+    DISCONNECT
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/TransferGuard.java

@@ -0,0 +1,25 @@
+package dev.moonbridge.api;
+
+import java.util.concurrent.CompletionStage;
+
+/** Coordinates application state that must move with a player backend transfer. */
+public interface TransferGuard {
+    /**
+     * Prepares the source before MoonBridge opens the replacement backend connection. Returning
+     * ALLOW means source handoff is complete. This callback may have side effects even when it
+     * rejects, fails, or times out; MoonBridge calls {@link #failed} unless cutover succeeds.
+     */
+    CompletionStage<TransferGuardDecision> prepare(TransferContext transfer);
+
+    /**
+     * Resolves a transfer that did not complete. Return SOURCE_RESTORED only after the exact
+     * source has been safely restored. The callback must be idempotent by transferId because a
+     * prepare acknowledgement may be lost while the source operation is still finishing.
+     */
+    CompletionStage<TransferFailureDisposition> failed(TransferContext transfer, String reason);
+
+    /** Called after network cutover succeeds, for plugin-owned cleanup of transfer state. */
+    default CompletionStage<Void> completed(TransferContext transfer) {
+        return java.util.concurrent.CompletableFuture.completedFuture(null);
+    }
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/TransferGuardDecision.java

@@ -0,0 +1,7 @@
+package dev.moonbridge.api;
+
+/** Result of the source preparation step. */
+public enum TransferGuardDecision {
+    ALLOW,
+    REJECT
+}