diff --git a/CLAUDE.md b/CLAUDE.md index 92a6959..87ac819 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,6 +32,8 @@ The slow tests are the session integration tests (`ProxySessionListenerTest`, `S **Session threading.** `ProxySessionListener` owns the listener, event loops and the UUID→session index. Each `Session` (one per player connection) confines its mutable state to the frontend event loop, and its backend channels are created on that same loop (`SessionChannels.backendBootstrap` enforces it). Login verification, backend DNS and plugin callbacks run off that loop and re-enter it by posting a task; results are checked against the current connection identity so a stale async result cannot act on a reconnected player. Plugin work goes through bounded pools that reject when full. +**Session layout.** `core.session` holds `Session` (the Netty handler and lifecycle owner) plus the pieces it delegates to: `FrontendLogin` (handshake, status, encryption, Mojang verification), `InitialRouter` (placement decision and first backend dial), `BackendHandshakes`. `core.session.transfer` is `TransferCoordinator`, which reaches the session only through the `TransferHost` port (Session supplies it as an adapter) and owns all transfer state. `core.session.play` has the per-frame PLAY handlers and relay start-up buffering; `core.session.channel` has backend dialing and pipeline lookups. Dependencies point one way: `session` -> `transfer` -> `play`/`channel`. + **Data path.** After login, `RawRelay` forwards already-framed PLAY frames between the two channels without copying. Per-frame concerns are separate inbound handlers (`KeepAliveBridge`, `PlayerCommandInterceptor`, `TabCompletionBridge`, `TransferFrameHandler`) in front of the relay. Backpressure works by pausing reads on the source when the peer is not writable; the relay flushes once per read batch (`channelReadComplete`). **Transfers.** `Session.beginTransfer` dials and logs into the candidate backend while the old link keeps serving, buffers frames from both sides (`TransferFrameBuffer`, `TransitionBuffer`), swaps relays, and rewrites the player's entity IDs afterwards (`Minecraft1710EntityIds`). Forge clients also need an FML reset/handshake before the world switch; `TransferCandidate` and `PlayObservation` track that state. Failure before the client-visible switch restores the old link; after it, the session is closed. diff --git a/benchmarks/RelayBenchmark.java b/benchmarks/RelayBenchmark.java index ad199b6..d471bca 100644 --- a/benchmarks/RelayBenchmark.java +++ b/benchmarks/RelayBenchmark.java @@ -1,5 +1,7 @@ package dev.moonbridge.core.session; +import dev.moonbridge.core.session.play.KeepAliveBridge; + import dev.moonbridge.core.protocol.MinecraftFrameDecoder; import dev.moonbridge.core.protocol.ProtocolProfile; import dev.moonbridge.core.protocol.ProtocolVarInt; diff --git a/docs/architecture.md b/docs/architecture.md index ffa8b99..3d6e01c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -28,7 +28,7 @@ flowchart LR F --> G[PLAY / Forge 握手与转发] ``` -普通转发沿用协议 5 帧边界，不重新编码普通帧；同一次 socket 读取解码出的帧逐个写入对端，在读取完成时合并为一次 flush，读取之外转发的帧由每批一次的事件循环任务补发 flush。前后端通道绑定同一 Netty EventLoop，会话控制状态在该循环串行变更。Keep Alive ID 由代理转换并跟踪未完成请求；必要的玩家实体 ID 数据包在转服后改写。通道不可写时通过暂停对端读取施加背压。实现见 [`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`RawRelay`](../proxy-core/src/main/java/dev/moonbridge/core/relay/RawRelay.java) 与 [`KeepAliveBridge`](../proxy-core/src/main/java/dev/moonbridge/core/session/KeepAliveBridge.java)。 +普通转发沿用协议 5 帧边界，不重新编码普通帧；同一次 socket 读取解码出的帧逐个写入对端，在读取完成时合并为一次 flush，读取之外转发的帧由每批一次的事件循环任务补发 flush。前后端通道绑定同一 Netty EventLoop，会话控制状态在该循环串行变更。Keep Alive ID 由代理转换并跟踪未完成请求；必要的玩家实体 ID 数据包在转服后改写。通道不可写时通过暂停对端读取施加背压。实现见 [`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`RawRelay`](../proxy-core/src/main/java/dev/moonbridge/core/relay/RawRelay.java) 与 [`KeepAliveBridge`](../proxy-core/src/main/java/dev/moonbridge/core/session/play/KeepAliveBridge.java)。 ## 后端目录与发现所有权 @@ -62,6 +62,6 @@ flowchart LR ## 相关源码 - 插件 API：[`PluginContext`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/PluginContext.java)、[`Servers`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/Servers.java)、[`Players`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/Players.java) -- 会话和转服：[`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`TransferCandidate`](../proxy-core/src/main/java/dev/moonbridge/core/session/TransferCandidate.java)、[`TransferFrameBuffer`](../proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameBuffer.java) +- 会话和转服：[`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`TransferCandidate`](../proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCandidate.java)、[`TransferFrameBuffer`](../proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameBuffer.java) - 目录：[`BackendCatalog`](../proxy-core/src/main/java/dev/moonbridge/core/backend/BackendCatalog.java)、[`InMemoryBackendCatalog`](../proxy-core/src/main/java/dev/moonbridge/core/backend/InMemoryBackendCatalog.java) - 后端控制通道：[`BackendControlService`](../proxy-core/src/main/java/dev/moonbridge/core/control/BackendControlService.java) diff --git a/docs/operations.md b/docs/operations.md index f39dc8b..a1b82c8 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -71,7 +71,7 @@ Docker 容器间的监听、服务名和地址配置见[配置与部署](backend 现有测试可用于定位行为范围： - 登录、认证和协议：`proxy-core/src/test/java/dev/moonbridge/core/auth/` 与 `proxy-core/src/test/java/dev/moonbridge/core/protocol/` -- 会话、断开、转服与缓冲：`proxy-core/src/test/java/dev/moonbridge/core/session/`，重点包括 `SessionTransferTest`、`SessionTransferDispatchTest`、`TransferFrameBufferTest` +- 会话、断开、转服与缓冲：`proxy-core/src/test/java/dev/moonbridge/core/session/`，重点包括 `SessionTransferTest`、`SessionTransferDispatchTest`、`transfer/TransferFrameBufferTest` - 插件生命周期、访问检查和 API：`proxy-core/src/test/java/dev/moonbridge/core/plugin/PluginHostTest.java`、`proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java`、`proxy-core/src/test/java/dev/moonbridge/app/ProxyConfigurationTest.java` 与 `proxy-plugin-api/src/test/` - 控制通道：`proxy-core/src/test/java/dev/moonbridge/core/control/` 与 `backend-channel-client/src/test/` diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/BackendHandshakes.java b/proxy-core/src/main/java/dev/moonbridge/core/session/BackendHandshakes.java new file mode 100644 index 0000000..c0d08e6 --- /dev/null +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/BackendHandshakes.java @@ -0,0 +1,51 @@ +package dev.moonbridge.core.session; + +import dev.moonbridge.api.PlayerIdentity; +import dev.moonbridge.core.auth.VerifiedProfile; +import dev.moonbridge.core.forwarding.BungeeLegacyForwarding; +import dev.moonbridge.core.protocol.MinecraftHandshake; +import dev.moonbridge.core.protocol.ProtocolProfile; +import dev.moonbridge.messaging.session.ForwardedSessionProof; +import io.netty.buffer.ByteBuf; +import io.netty.channel.Channel; + +import java.net.InetSocketAddress; +import java.util.Arrays; +import java.util.List; +import java.util.UUID; + +/** Builds the handshake a backend receives: forwarded identity plus, when bound, a session proof. */ +final class BackendHandshakes { + private BackendHandshakes() { } + + static ByteBuf encode(ProxySessionListener owner, Channel frontend, MinecraftHandshake handshake, + PlayerIdentity identity, String username, VerifiedProfile verifiedProfile, + String backendName, long backendEpoch) { + byte[] secret = owner.sessionBindingSecret(backendName); + try { + if (owner.onlineMode()) { + return forward(owner, frontend, handshake, identity, verifiedProfile, secret, backendName, backendEpoch); + } + if (secret == null) return handshake.encode(frontend.alloc(), ProtocolProfile.minecraft1710()); + // OFFLINE identity is derived by this proxy and is not a Mojang-authenticated profile. + // Only forward it when this backend has explicitly configured session binding. + VerifiedProfile offlineProfile = new VerifiedProfile(identity.playerId(), username, List.of()); + return forward(owner, frontend, handshake, identity, offlineProfile, secret, backendName, backendEpoch); + } finally { + if (secret != null) Arrays.fill(secret, (byte) 0); + } + } + + private static ByteBuf forward(ProxySessionListener owner, Channel frontend, MinecraftHandshake handshake, + PlayerIdentity identity, VerifiedProfile profile, byte[] secret, + String backendName, long backendEpoch) { + String proof = null; + if (secret != null && backendEpoch > 0) { + long expiresAt = Math.addExact(System.currentTimeMillis(), ForwardedSessionProof.MAX_LIFETIME_MILLIS); + proof = ForwardedSessionProof.create(owner.proxyEpoch(), identity.playerId(), identity.connectionId(), + backendName, backendEpoch, UUID.randomUUID(), expiresAt, secret); + } + return BungeeLegacyForwarding.encode(frontend.alloc(), handshake, + (InetSocketAddress) frontend.remoteAddress(), profile, proof); + } +} diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/FrontendLogin.java b/proxy-core/src/main/java/dev/moonbridge/core/session/FrontendLogin.java new file mode 100644 index 0000000..1d6846f --- /dev/null +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/FrontendLogin.java @@ -0,0 +1,184 @@ +package dev.moonbridge.core.session; + +import dev.moonbridge.core.auth.AuthenticatedEncryption; +import dev.moonbridge.core.auth.MinecraftCipherDecoder; +import dev.moonbridge.core.auth.MinecraftCipherEncoder; +import dev.moonbridge.core.auth.MinecraftEncryptionRequest; +import dev.moonbridge.core.auth.MinecraftEncryptionResponse; +import dev.moonbridge.core.auth.OnlineModeCrypto; +import dev.moonbridge.core.auth.VerifiedProfile; +import dev.moonbridge.core.protocol.LoginStart; +import dev.moonbridge.core.protocol.MinecraftFrameDecoder; +import dev.moonbridge.core.protocol.MinecraftHandshake; +import dev.moonbridge.core.protocol.ProtocolProfile; +import dev.moonbridge.core.protocol.ProtocolVarInt; +import dev.moonbridge.core.session.play.FrameTransformHandler; +import io.netty.buffer.ByteBuf; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPipeline; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.security.GeneralSecurityException; +import java.util.Arrays; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.function.BiConsumer; +import java.util.function.BooleanSupplier; + +/** + * The client-facing half of connecting: handshake, server-list status, login start and (online mode) + * encryption and Mojang verification. It ends by reporting who the player is; everything after that + * (permissions, admission, placement) belongs to the session. Confined to the frontend event loop. + */ +final class FrontendLogin { + private static final String LOGIN_WAIT_GUARD = "login-wait-guard"; + + private final ProxySessionListener owner; + private final Channel frontend; + private final Runnable closeSession; + private final BooleanSupplier shuttingDown; + private final BiConsumer authenticated; + private MinecraftHandshake handshake; + private boolean statusRequest; + private LoginStart loginStart; + private MinecraftEncryptionRequest encryptionRequest; + private VerifiedProfile verifiedProfile; + private CompletableFuture> verification; + + FrontendLogin(ProxySessionListener owner, Channel frontend, Runnable closeSession, + BooleanSupplier shuttingDown, BiConsumer authenticated) { + this.owner = owner; + this.frontend = frontend; + this.closeSession = closeSession; + this.shuttingDown = shuttingDown; + this.authenticated = authenticated; + } + + MinecraftHandshake handshake() { return handshake; } + LoginStart loginStart() { return loginStart; } + VerifiedProfile verifiedProfile() { return verifiedProfile; } + + void cancel() { + if (verification != null) verification.cancel(true); + } + + /** Any frame the client sends while the proxy is still deciding about the login is a protocol error. */ + void guardPendingLogin() { + ChannelPipeline pipeline = frontend.pipeline(); + if (pipeline.get(LOGIN_WAIT_GUARD) == null) { + pipeline.addFirst(LOGIN_WAIT_GUARD, new FrameTransformHandler(closeSession) { + @Override protected ByteBuf transform(ChannelHandlerContext ctx, ByteBuf frame) { + throw new IllegalStateException("client sent data while login was pending"); + } + }); + } + frontend.config().setAutoRead(true); + } + + void releaseGuard() { + if (frontend.pipeline().get(LOGIN_WAIT_GUARD) != null) frontend.pipeline().remove(LOGIN_WAIT_GUARD); + } + + void receive(ByteBuf packet) { + if (handshake == null) { + receiveHandshake(packet); + return; + } + if (statusRequest) { + receiveStatus(packet); + return; + } + if (encryptionRequest != null) { + receiveEncryptionResponse(packet); + return; + } + if (loginStart != null) { + closeSession.run(); + return; + } + loginStart = LoginStart.decode(packet, ProtocolProfile.minecraft1710()); + if (owner.onlineMode()) { + encryptionRequest = owner.newEncryptionRequest(); + frontend.writeAndFlush(encryptionRequest.encode(frontend.alloc())).addListener(write -> { + if (!write.isSuccess()) closeSession.run(); + }); + return; + } + UUID uuid = UUID.nameUUIDFromBytes(("OfflinePlayer:" + loginStart.username()).getBytes(StandardCharsets.UTF_8)); + authenticated.accept(uuid, loginStart.username()); + } + + private void receiveHandshake(ByteBuf packet) { + handshake = MinecraftHandshake.decode(packet, ProtocolProfile.minecraft1710()); + if (handshake.nextState() == MinecraftHandshake.NextState.LOGIN) { + int separator = handshake.serverAddress().indexOf('\0'); + if (separator >= 0) { + handshake = new MinecraftHandshake(handshake.protocolVersion(), + handshake.serverAddress().substring(0, separator), handshake.serverPort(), handshake.nextState()); + } + } + if (handshake.protocolVersion() != ProtocolProfile.PROTOCOL_1_7_10) closeSession.run(); + else statusRequest = handshake.nextState() == MinecraftHandshake.NextState.STATUS; + } + + private void receiveStatus(ByteBuf packet) { + int id = ProtocolVarInt.read(packet.duplicate()); + if (id == 0) { + frontend.writeAndFlush(owner.serverListStatus().encode(frontend.alloc(), owner.onlineCount())); + } else if (id == 1) { + ByteBuf input = packet.duplicate(); + ProtocolVarInt.read(input); + if (input.readableBytes() != 8) throw new IllegalArgumentException("bad status ping"); + ByteBuf pong = frontend.alloc().buffer(9); + ProtocolVarInt.write(pong, 1); + pong.writeLong(input.readLong()); + frontend.writeAndFlush(pong).addListener(ignored -> frontend.close()); + } else closeSession.run(); + } + + private void receiveEncryptionResponse(ByteBuf packet) { + MinecraftEncryptionResponse response = MinecraftEncryptionResponse.decode(packet); + AuthenticatedEncryption accepted; + try { + accepted = OnlineModeCrypto.decrypt(owner.encryptionKeys().getPrivate(), encryptionRequest, response); + } catch (GeneralSecurityException failure) { + closeSession.run(); + return; + } + byte[] secret = accepted.sharedSecret(); + String serverHash = OnlineModeCrypto.serverHash(encryptionRequest.serverId(), secret, + encryptionRequest.publicKey()); + encryptionRequest = null; + ChannelPipeline pipeline = frontend.pipeline(); + try { + pipeline.addAfter("minecraft-frame-decoder", "minecraft-cipher-decoder", new MinecraftCipherDecoder(secret)); + pipeline.addAfter("minecraft-cipher-decoder", "encrypted-frame-decoder", + new MinecraftFrameDecoder(ProtocolProfile.minecraft1710(), true, + ProtocolProfile.MAX_LOGIN_FRAME_BYTES)); + pipeline.addBefore("minecraft-frame-encoder", "minecraft-cipher-encoder", new MinecraftCipherEncoder(secret)); + pipeline.remove("minecraft-frame-decoder"); + } finally { + Arrays.fill(secret, (byte) 0); + } + guardPendingLogin(); + String clientIp = ((InetSocketAddress) frontend.remoteAddress()).getAddress().getHostAddress(); + verification = owner.verifier().verify(loginStart.username(), serverHash, clientIp).toCompletableFuture(); + verification.whenComplete((profile, failure) -> frontend.eventLoop().execute(() -> { + verification = null; + if (shuttingDown.getAsBoolean()) return; + if (failure != null || profile == null || profile.isEmpty()) { + closeSession.run(); + return; + } + verifiedProfile = profile.get(); + if (!verifiedProfile.username().equalsIgnoreCase(loginStart.username())) { + closeSession.run(); + return; + } + authenticated.accept(verifiedProfile.uuid(), verifiedProfile.username()); + })); + } +} diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/InitialRouter.java b/proxy-core/src/main/java/dev/moonbridge/core/session/InitialRouter.java new file mode 100644 index 0000000..3ff89bb --- /dev/null +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/InitialRouter.java @@ -0,0 +1,159 @@ +package dev.moonbridge.core.session; + +import dev.moonbridge.api.PlacementDecision; +import dev.moonbridge.core.backend.BackendId; +import dev.moonbridge.core.backend.BackendView; +import dev.moonbridge.core.session.channel.SessionChannels; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import net.kyori.adventure.text.Component; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Optional; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +/** + * Picks the player's first backend: turns a placement decision into an ordered candidate list and dials + * it until one connects. Retries only before a backend has received any Minecraft handshake/login bytes. + * Confined to the frontend event loop. + */ +final class InitialRouter { + private static final Logger LOGGER = LoggerFactory.getLogger(InitialRouter.class); + + /** Outcome callbacks; {@code connected} gets a live channel that nothing has been written to yet. */ + interface Listener { + void connected(Channel backend, BackendView target); + void reject(Component reason); + void close(); + } + + private final ProxySessionListener owner; + private final Channel frontend; + private final BooleanSupplier shuttingDown; + private final Listener listener; + private List candidates = List.of(); + private int nextCandidate; + private long deadlineNanos; + private String username; + private Channel dialing; + + InitialRouter(ProxySessionListener owner, Channel frontend, BooleanSupplier shuttingDown, Listener listener) { + this.owner = owner; + this.frontend = frontend; + this.shuttingDown = shuttingDown; + this.listener = listener; + } + + /** Call when placement starts; the whole routing budget runs from here. */ + void startClock() { deadlineNanos = System.nanoTime() + owner.placementTimeout().toNanos(); } + + void route(String username, Optional decision) { + this.username = username; + if (decision.isPresent()) { + if (decision.get() instanceof PlacementDecision.Reject rejected) { + listener.reject(rejected.reason()); + return; + } + if (!(decision.get() instanceof PlacementDecision.Select selected)) { + listener.close(); + return; + } + candidates = selected.backendNames(); + } else { + candidates = owner.initialServers(); + } + nextCandidate = 0; + if (candidates.isEmpty()) { + reject("No entry servers are configured."); + return; + } + dialNext(); + } + + /** Closes an in-flight dial that never became the session's backend. */ + void cancel() { + Channel channel = dialing; + if (channel != null) channel.close(); + } + + private void reject(String reason) { listener.reject(Component.text(reason)); } + + private void dialNext() { + if (shuttingDown.getAsBoolean()) return; + while (nextCandidate < candidates.size()) { + long remaining = deadlineNanos - System.nanoTime(); + if (remaining <= 0) { + reject("Initial server routing timed out."); + return; + } + BackendView target = owner.catalog().find(new BackendId(candidates.get(nextCandidate++))).orElse(null); + if (target == null || !SessionChannels.isTcpAddress(target.address())) continue; + dial(target, remaining); + return; + } + reject("No entry server could be reached."); + } + + private void dial(BackendView target, long remainingNanos) { + // The session is not attached to this channel until the attempt has won. A failed + // dial's inactive/exception callbacks must not close a later connection. + Bootstrap bootstrap = SessionChannels.backendBootstrap(frontend, owner.transport(), owner.backendResolver(), + (int) Math.max(1, Math.min(5000, TimeUnit.NANOSECONDS.toMillis(remainingNanos))), false, + pipeline -> { }); + ChannelFuture connect; + try { + connect = bootstrap.connect(SessionChannels.socketAddress(target.address())); + } catch (RuntimeException failure) { + LOGGER.debug("Initial backend dial could not start for {} to {}", username, target.address(), failure); + dialNext(); + return; + } + dialing = connect.channel(); + connect.addListener(ignored -> finish(connect, target)); + } + + private void finish(ChannelFuture connect, BackendView target) { + if (!frontend.eventLoop().inEventLoop()) { + try { + frontend.eventLoop().execute(() -> finish(connect, target)); + } catch (RejectedExecutionException shutdown) { + connect.channel().close(); + } + return; + } + if (shuttingDown.getAsBoolean() || dialing != connect.channel()) { + connect.channel().close(); + return; + } + if (!connect.isSuccess()) { + LOGGER.debug("Initial backend connection failed for player {} to {}", username, target.address(), + connect.cause()); + retry(connect); + return; + } + BackendView current = owner.catalog().find(target.handle().id()).orElse(null); + if (current == null || !current.handle().equals(target.handle()) + || !current.address().equals(target.address())) { + retry(connect); + return; + } + if (deadlineNanos - System.nanoTime() <= 0) { + reject("Initial server routing timed out."); + return; + } + dialing = null; + candidates = List.of(); + listener.connected(connect.channel(), current); + } + + private void retry(ChannelFuture connect) { + dialing = null; + connect.channel().close(); + dialNext(); + } +} diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/ProxySessionListener.java b/proxy-core/src/main/java/dev/moonbridge/core/session/ProxySessionListener.java index 27c19e0..b26d6f2 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/ProxySessionListener.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/ProxySessionListener.java @@ -1,5 +1,7 @@ package dev.moonbridge.core.session; +import dev.moonbridge.core.session.channel.SessionFrameEncoder; + import dev.moonbridge.api.PlacementDecision; import dev.moonbridge.api.event.Event; import dev.moonbridge.api.event.ConnectionAdmissionEvent; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/Session.java b/proxy-core/src/main/java/dev/moonbridge/core/session/Session.java index aaca6c5..49b8853 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/Session.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/Session.java @@ -1,45 +1,40 @@ package dev.moonbridge.core.session; +import dev.moonbridge.core.session.channel.SessionChannels; +import dev.moonbridge.core.session.play.CommandRateLimiter; +import dev.moonbridge.core.session.play.KeepAliveBridge; +import dev.moonbridge.core.session.play.PlayObservation; +import dev.moonbridge.core.session.play.PlayerCommandInterceptor; +import dev.moonbridge.core.session.play.TabCompletionBridge; +import dev.moonbridge.core.session.play.TransitionBuffer; +import dev.moonbridge.core.session.play.TransitionFrames; +import dev.moonbridge.core.session.transfer.TransferCoordinator; +import dev.moonbridge.core.session.transfer.TransferHost; + import dev.moonbridge.api.PlacementDecision; import dev.moonbridge.api.AccessDecision; -import dev.moonbridge.api.ServerView; -import dev.moonbridge.api.event.TransferContext; -import dev.moonbridge.api.event.TransferPreparingEvent; import dev.moonbridge.core.event.TransferPreparation; -import dev.moonbridge.core.event.PreparedTransfer; import dev.moonbridge.api.event.PlayerAdmissionEvent; import dev.moonbridge.api.PlayerIdentity; import dev.moonbridge.api.PlayerView; import dev.moonbridge.api.MessageResult; import dev.moonbridge.api.DisconnectResult; import dev.moonbridge.api.TransferResult; -import dev.moonbridge.api.TransferStatus; -import dev.moonbridge.core.auth.AuthenticatedEncryption; -import dev.moonbridge.core.auth.MinecraftCipherDecoder; -import dev.moonbridge.core.auth.MinecraftCipherEncoder; -import dev.moonbridge.core.auth.MinecraftEncryptionRequest; -import dev.moonbridge.core.auth.MinecraftEncryptionResponse; -import dev.moonbridge.core.auth.OnlineModeCrypto; -import dev.moonbridge.core.auth.VerifiedProfile; import dev.moonbridge.core.backend.BackendView; -import dev.moonbridge.core.backend.BackendId; -import dev.moonbridge.core.forwarding.BungeeLegacyForwarding; -import dev.moonbridge.messaging.session.ForwardedSessionProof; -import dev.moonbridge.core.protocol.ByteBufs; import dev.moonbridge.core.protocol.LoginStart; -import dev.moonbridge.core.protocol.Minecraft1710EntityIds; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.MinecraftLoginSuccess; import dev.moonbridge.core.protocol.MinecraftLoginDisconnect; -import dev.moonbridge.core.protocol.MinecraftHandshake; -import dev.moonbridge.core.protocol.MinecraftFrameDecoder; import dev.moonbridge.core.protocol.ProtocolProfile; import dev.moonbridge.core.protocol.ProtocolVarInt; import dev.moonbridge.core.relay.RawRelay; -import io.netty.bootstrap.Bootstrap; +import dev.moonbridge.core.backend.BackendCatalog; +import dev.moonbridge.core.event.TransferPreparation; +import dev.moonbridge.core.net.NetworkTransport; +import io.netty.resolver.AddressResolverGroup; +import java.util.function.BiConsumer; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; -import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; @@ -49,17 +44,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; -import java.security.GeneralSecurityException; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.time.Instant; import java.util.Optional; import java.util.UUID; -import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.Arrays; import java.util.Objects; -import java.util.List; import net.kyori.adventure.text.Component; import dev.moonbridge.core.protocol.MinecraftText; import java.util.concurrent.CompletableFuture; @@ -77,26 +67,16 @@ final class Session extends ChannelInboundHandlerAdapter { private static final int MAX_TRANSITION_BUFFER_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes(); private static final int MAX_TRANSITION_BUFFER_FRAMES = 1024; private static final int MAX_PENDING_MESSAGES = 64; - private static final Duration FORGE_TRANSFER_HANDSHAKE_TIMEOUT = Duration.ofSeconds(30); private static final Duration LOGIN_DISCONNECT_DRAIN_TIMEOUT = Duration.ofSeconds(5); private final ProxySessionListener owner; private final Channel frontend; private volatile Channel backend; private boolean backendConnected; - private MinecraftHandshake handshake; - private boolean statusRequest; - private LoginStart loginStart; - private MinecraftEncryptionRequest encryptionRequest; - private VerifiedProfile verifiedProfile; - private CompletableFuture> verification; private PlayerIdentity identity; private boolean identityClaimed; private volatile PlayerView view; private BackendView selected; private CompletableFuture> placementRequest; - private List initialCandidates = List.of(); - private int initialCandidateIndex; - private long initialRoutingDeadlineNanos; private CompletableFuture admissionRequest; private boolean loginDisconnectStarted; private volatile boolean published; @@ -119,15 +99,24 @@ final class Session extends ChannelInboundHandlerAdapter { private PlayObservation playObservation; private KeepAliveBridge.State keepAlives; private RawRelay.Link relay; - private Integer clientEntityId; - private boolean clientFmlAwaitingServerHello; - private TransferFrameHandler.State frameState; - private PendingTransfer pendingTransfer; - private TransferAttempt transfer; + private final FrontendLogin login; + private final InitialRouter router; + private final TransferCoordinator transfers; Session(ProxySessionListener owner, Channel frontend) { this.owner = owner; this.frontend = frontend; + this.login = new FrontendLogin(owner, frontend, this::closePair, () -> closed.get() || disconnecting, + this::beginLogin); + this.router = new InitialRouter(owner, frontend, () -> closed.get() || disconnecting, + new InitialRouter.Listener() { + @Override public void connected(Channel channel, BackendView target) { + initialBackendConnected(channel, target); + } + @Override public void reject(Component reason) { disconnectLogin(reason); } + @Override public void close() { closePair(); } + }); + this.transfers = new TransferCoordinator(new Host()); } @Override public void handlerAdded(ChannelHandlerContext ctx) { @@ -160,7 +149,7 @@ final class Session extends ChannelInboundHandlerAdapter { bufferTransitionFrame(ctx.channel() == frontend, body); return; } - if (ctx.channel() == frontend) receiveFrontend(body); + if (ctx.channel() == frontend) login.receive(body); else receiveBackend(body); } catch (RuntimeException failure) { LOGGER.debug("Closing session after {} frame handling failed", @@ -171,104 +160,6 @@ final class Session extends ChannelInboundHandlerAdapter { } } - private void receiveFrontend(ByteBuf packet) { - if (handshake == null) { - handshake = MinecraftHandshake.decode(packet, ProtocolProfile.minecraft1710()); - if (handshake.nextState() == MinecraftHandshake.NextState.LOGIN) { - int separator = handshake.serverAddress().indexOf('\0'); - if (separator >= 0) { - handshake = new MinecraftHandshake(handshake.protocolVersion(), - handshake.serverAddress().substring(0, separator), handshake.serverPort(), handshake.nextState()); - } - } - if (handshake.protocolVersion() != ProtocolProfile.PROTOCOL_1_7_10) { - closePair(); - return; - } - if (handshake.nextState() == MinecraftHandshake.NextState.STATUS) { - statusRequest = true; - return; - } - return; - } - if (statusRequest) { - int id = ProtocolVarInt.read(packet.duplicate()); - if (id == 0) { - sendStatus(); - } else if (id == 1) { - ByteBuf input = packet.duplicate(); - ProtocolVarInt.read(input); - if (input.readableBytes() != 8) throw new IllegalArgumentException("bad status ping"); - ByteBuf pong = frontend.alloc().buffer(9); - ProtocolVarInt.write(pong, 1); - pong.writeLong(input.readLong()); - frontend.writeAndFlush(pong).addListener(ignored -> frontend.close()); - } else closePair(); - return; - } - if (encryptionRequest != null) { - receiveEncryptionResponse(packet); - return; - } - if (loginStart != null) { - closePair(); - return; - } - loginStart = LoginStart.decode(packet, ProtocolProfile.minecraft1710()); - if (owner.onlineMode()) { - encryptionRequest = owner.newEncryptionRequest(); - frontend.writeAndFlush(encryptionRequest.encode(frontend.alloc())).addListener(write -> { - if (!write.isSuccess()) closePair(); - }); - return; - } - UUID uuid = UUID.nameUUIDFromBytes(("OfflinePlayer:" + loginStart.username()).getBytes(StandardCharsets.UTF_8)); - beginLogin(uuid, loginStart.username()); - } - - private void receiveEncryptionResponse(ByteBuf packet) { - MinecraftEncryptionResponse response = MinecraftEncryptionResponse.decode(packet); - AuthenticatedEncryption accepted; - try { - accepted = OnlineModeCrypto.decrypt(owner.encryptionKeys().getPrivate(), encryptionRequest, response); - } catch (GeneralSecurityException failure) { - closePair(); - return; - } - byte[] secret = accepted.sharedSecret(); - String serverHash = OnlineModeCrypto.serverHash(encryptionRequest.serverId(), secret, - encryptionRequest.publicKey()); - encryptionRequest = null; - ChannelPipeline pipeline = frontend.pipeline(); - try { - pipeline.addAfter("minecraft-frame-decoder", "minecraft-cipher-decoder", new MinecraftCipherDecoder(secret)); - pipeline.addAfter("minecraft-cipher-decoder", "encrypted-frame-decoder", - new MinecraftFrameDecoder(ProtocolProfile.minecraft1710(), true, - ProtocolProfile.MAX_LOGIN_FRAME_BYTES)); - pipeline.addBefore("minecraft-frame-encoder", "minecraft-cipher-encoder", new MinecraftCipherEncoder(secret)); - pipeline.remove("minecraft-frame-decoder"); - } finally { - Arrays.fill(secret, (byte) 0); - } - observeWaitingClient(); - String clientIp = ((InetSocketAddress) frontend.remoteAddress()).getAddress().getHostAddress(); - verification = owner.verifier().verify(loginStart.username(), serverHash, clientIp).toCompletableFuture(); - verification.whenComplete((profile, failure) -> frontend.eventLoop().execute(() -> { - verification = null; - if (closed.get() || disconnecting) return; - if (failure != null || profile == null || profile.isEmpty()) { - closePair(); - return; - } - verifiedProfile = profile.get(); - if (!verifiedProfile.username().equalsIgnoreCase(loginStart.username())) { - closePair(); - return; - } - beginLogin(verifiedProfile.uuid(), verifiedProfile.username()); - })); - } - private void beginLogin(UUID uuid, String username) { identity = new PlayerIdentity(uuid, owner.allocateConnectionId()); if (!owner.claimIdentity(uuid, this)) { @@ -277,28 +168,25 @@ final class Session extends ChannelInboundHandlerAdapter { } identityClaimed = true; view = new PlayerView(identity, username, Optional.empty()); - observeWaitingClient(); + login.guardPendingLogin(); resetLoginDeadline(owner.eventTimeout().plusSeconds(1)); + CompletionStage prepared; try { - owner.preparePermissions(view).whenComplete((ignored, failure) -> { - try { - frontend.eventLoop().execute(() -> { - if (closed.get() || disconnecting || loginDisconnectStarted) return; - if (failure != null) { - LOGGER.debug("Permission preparation failed for {}", view.username(), failure); - disconnectLogin("Could not load permissions. Please try again."); - } else { - beginPlayerAdmission(); - } - }); - } catch (RejectedExecutionException shutdown) { - closePair(); - } - }); + prepared = owner.preparePermissions(view); } catch (RuntimeException failure) { LOGGER.debug("Could not prepare permissions for {}", view.username(), failure); disconnectLogin("Could not load permissions. Please try again."); + return; } + resumeOnLoop(prepared, (ignored, failure) -> { + if (closed.get() || disconnecting || loginDisconnectStarted) return; + if (failure != null) { + LOGGER.debug("Permission preparation failed for {}", view.username(), failure); + disconnectLogin("Could not load permissions. Please try again."); + } else { + beginPlayerAdmission(); + } + }); } private void beginPlayerAdmission() { @@ -307,169 +195,63 @@ final class Session extends ChannelInboundHandlerAdapter { return; } resetLoginDeadline(owner.eventTimeout().plusSeconds(1)); + CompletableFuture request; try { - CompletableFuture request = Objects.requireNonNull(owner.dispatchEvent( - new PlayerAdmissionEvent(view, (InetSocketAddress) frontend.remoteAddress(), verifiedProfile != null)), + request = Objects.requireNonNull(owner.dispatchEvent( + new PlayerAdmissionEvent(view, (InetSocketAddress) frontend.remoteAddress(), + login.verifiedProfile() != null)), "player admission stage").toCompletableFuture(); - admissionRequest = request; - request.whenComplete((decision, failure) -> { - try { - frontend.eventLoop().execute(() -> { - if (admissionRequest == request) admissionRequest = null; - if (closed.get() || loginDisconnectStarted) return; - if (failure != null || decision == null) { - LOGGER.debug("Player admission failed for {}", view.username(), failure); - disconnectLogin("Could not check login access. Please try again."); - } else if (decision instanceof AccessDecision.Denied denied) { - disconnectLogin(denied.reason()); - } else { - beginPlacement(); - } - }); - } catch (RejectedExecutionException shutdown) { - closePair(); - } - }); } catch (RuntimeException failure) { LOGGER.debug("Could not start player admission for {}", view.username(), failure); disconnectLogin("Could not check login access. Please try again."); + return; } + admissionRequest = request; + resumeOnLoop(request, (decision, failure) -> { + if (admissionRequest == request) admissionRequest = null; + if (closed.get() || loginDisconnectStarted) return; + if (failure != null || decision == null) { + LOGGER.debug("Player admission failed for {}", view.username(), failure); + disconnectLogin("Could not check login access. Please try again."); + } else if (decision instanceof AccessDecision.Denied denied) { + disconnectLogin(denied.reason()); + } else { + beginPlacement(); + } + }); } private void beginPlacement() { - initialRoutingDeadlineNanos = System.nanoTime() + owner.placementTimeout().toNanos(); + router.startClock(); resetLoginDeadline(owner.placementTimeout()); CompletableFuture> request = owner.placement().apply(view).toCompletableFuture(); placementRequest = request; - request.whenComplete((decision, failure) -> { + resumeOnLoop(request, (decision, failure) -> { + if (placementRequest == request) placementRequest = null; if (closed.get() || disconnecting) return; + if (failure != null || decision == null) { + disconnectLogin("Could not select a server. Please try again."); + return; + } + router.route(view.username(), decision); + }); + } + + /** Re-enters the frontend event loop from whichever thread completed {@code stage}. */ + private void resumeOnLoop(CompletionStage stage, BiConsumer handler) { + stage.whenComplete((value, failure) -> { try { - frontend.eventLoop().execute(() -> { - if (placementRequest == request) placementRequest = null; - if (closed.get() || disconnecting) return; - if (failure != null || decision == null) { - disconnectLogin("Could not select a server. Please try again."); - return; - } - selectBackend(decision); - }); + frontend.eventLoop().execute(() -> handler.accept(value, failure)); } catch (RejectedExecutionException shutdown) { closePair(); } }); } - private void observeWaitingClient() { - ChannelPipeline pipeline = frontend.pipeline(); - if (pipeline.get("login-wait-guard") == null) { - pipeline.addFirst("login-wait-guard", new FrameTransformHandler(this::closePair) { - @Override protected ByteBuf transform(ChannelHandlerContext ctx, ByteBuf frame) { - throw new IllegalStateException("client sent data while login was pending"); - } - }); - } - frontend.config().setAutoRead(true); - } - - private void sendStatus() { - frontend.writeAndFlush(owner.serverListStatus().encode(frontend.alloc(), owner.onlineCount())); - } - - private void selectBackend(Optional decision) { - if (decision.isPresent()) { - if (decision.get() instanceof PlacementDecision.Reject rejected) { - disconnectLogin(rejected.reason()); - return; - } - if (!(decision.get() instanceof PlacementDecision.Select selectedDecision)) { - closePair(); - return; - } - initialCandidates = selectedDecision.backendNames(); - } else { - initialCandidates = owner.initialServers(); - } - initialCandidateIndex = 0; - if (initialCandidates.isEmpty()) { - disconnectLogin("No entry servers are configured."); - return; - } - connectNextInitialBackend(); - } - - /** Retries only before a backend has received any Minecraft handshake/login bytes. */ - private void connectNextInitialBackend() { - if (closed.get() || disconnecting) return; - while (initialCandidateIndex < initialCandidates.size()) { - long remaining = initialRoutingDeadlineNanos - System.nanoTime(); - if (remaining <= 0) { - disconnectLogin("Initial server routing timed out."); - return; - } - String name = initialCandidates.get(initialCandidateIndex++); - BackendView target = owner.catalog().find(new BackendId(name)).orElse(null); - if (target == null) continue; - if (!SessionChannels.isTcpAddress(target.address())) continue; - connectInitialBackend(target, remaining); - return; - } - disconnectLogin("No entry server could be reached."); - } - - private void connectInitialBackend(BackendView target, long remainingNanos) { - // Do not attach the shared Session until the attempt has won. A failed - // dial's inactive/exception callbacks must not close a later connection. - Bootstrap bootstrap = SessionChannels.backendBootstrap(frontend, owner.transport(), owner.backendResolver(), - (int) Math.max(1, Math.min(5000, TimeUnit.NANOSECONDS.toMillis(remainingNanos))), false, - pipeline -> { }); - ChannelFuture connect; - try { - connect = bootstrap.connect(SessionChannels.socketAddress(target.address())); - } catch (RuntimeException failure) { - LOGGER.debug("Initial backend dial could not start for {} to {}", view.username(), target.address(), failure); - connectNextInitialBackend(); - return; - } - backend = connect.channel(); - connect.addListener(ignored -> finishInitialConnection(connect, target)); - } - - private void finishInitialConnection(ChannelFuture connect, BackendView target) { - if (!frontend.eventLoop().inEventLoop()) { - try { - frontend.eventLoop().execute(() -> finishInitialConnection(connect, target)); - } catch (RejectedExecutionException shutdown) { - connect.channel().close(); - } - return; - } - if (closed.get() || disconnecting || backend != connect.channel()) { - connect.channel().close(); - return; - } - if (!connect.isSuccess()) { - LOGGER.debug("Initial backend connection failed for player {} to {}", view.username(), target.address(), - connect.cause()); - backend = null; - connect.channel().close(); - connectNextInitialBackend(); - return; - } - BackendView current = owner.catalog().find(target.handle().id()).orElse(null); - if (current == null || !current.handle().equals(target.handle()) - || !current.address().equals(target.address())) { - backend = null; - connect.channel().close(); - connectNextInitialBackend(); - return; - } - if (initialRoutingDeadlineNanos - System.nanoTime() <= 0) { - disconnectLogin("Initial server routing timed out."); - return; - } - selected = current; + private void initialBackendConnected(Channel channel, BackendView target) { + backend = channel; + selected = target; backendConnected = true; - initialCandidates = List.of(); resetLoginDeadline(owner.loginStageTimeout()); backend.pipeline().addLast("initial-session", Session.this); backend.config().setAutoRead(true); @@ -518,7 +300,7 @@ final class Session extends ChannelInboundHandlerAdapter { LOGGER.debug("Login phase deadline expired for player {} with backend {}; disconnect started={}", view == null ? "" : view.username(), selected == null ? "" : selected.handle().id().value(), loginDisconnectStarted); - if (loginStart != null && !loginDisconnectStarted && !published) { + if (login.loginStart() != null && !loginDisconnectStarted && !published) { disconnectLogin("Login timed out."); } else { closePair(); @@ -557,7 +339,7 @@ final class Session extends ChannelInboundHandlerAdapter { closePair(); return; } - UUID expected = owner.onlineMode() ? verifiedProfile.uuid() + UUID expected = owner.onlineMode() ? login.verifiedProfile().uuid() : UUID.nameUUIDFromBytes(("OfflinePlayer:" + success.username()).getBytes(StandardCharsets.UTF_8)); if (!expected.equals(success.playerId())) { LOGGER.debug("Backend login UUID mismatch for player {}", view.username()); @@ -575,7 +357,7 @@ final class Session extends ChannelInboundHandlerAdapter { if (closed.get()) return; frontend.eventLoop().execute(() -> { initialPlayDeadline.cancel(false); - if (failure == null) tryStartPendingTransfer(); + if (failure == null) transfers.startPending(); }); }); view = new PlayerView(identity, view.username(), selected.handle().id().value()); @@ -585,9 +367,7 @@ final class Session extends ChannelInboundHandlerAdapter { if (closed.get() || disconnecting) return; relayStarting = true; frontend.config().setAutoRead(false); - if (frontend.pipeline().get("login-wait-guard") != null) { - frontend.pipeline().remove("login-wait-guard"); - } + login.releaseGuard(); backend.config().setAutoRead(false); // Outbound order is the protocol boundary: anything submitted after Login Success // must use PLAY packet identifiers, even while its write promise is still pending. @@ -648,7 +428,7 @@ final class Session extends ChannelInboundHandlerAdapter { relay = link; relayStarting = false; link.start(); - tryStartPendingTransfer(); + transfers.startPending(); } }); }); @@ -716,7 +496,7 @@ final class Session extends ChannelInboundHandlerAdapter { return; } if (!published || !frontendPlayPhase || relayStarting || relay == null - || transfer != null || pendingTransfer != null || playObservation == null + || transfers.busy() || playObservation == null || !playObservation.ready().isDone() || playObservation.ready().isCompletedExceptionally()) { finishMessage(); result.complete(MessageResult.NOT_READY); @@ -944,593 +724,13 @@ final class Session extends ChannelInboundHandlerAdapter { return removed; } - CompletionStage transferTo(String backendName) { - CompletableFuture result = new CompletableFuture<>(); - Runnable command = () -> { - try { - beginTransfer(backendName, result); - } catch (RuntimeException failure) { - result.completeExceptionally(failure); - closePair(); - } - }; - if (frontend.eventLoop().inEventLoop()) command.run(); - else { - try { - frontend.eventLoop().execute(command); - } catch (RejectedExecutionException shutdown) { - result.complete(TransferResult.of(TransferStatus.PLAYER_NOT_CONNECTED)); - } - } - return result; - } - - private void beginTransfer(String backendName, CompletableFuture result) { - if (closed.get() || disconnecting || !published) { - result.complete(TransferResult.of(TransferStatus.PLAYER_NOT_CONNECTED)); - return; - } - if (transfer != null || pendingTransfer != null) { - result.complete(TransferResult.failed("a backend transfer is already in progress")); - return; - } - if (selected.handle().id().value().equals(backendName)) { - BackendView current = owner.catalog().find(selected.handle().id()).orElse(null); - if (current != null && current.address().equals(selected.address())) { - result.complete(TransferResult.of(TransferStatus.NETWORK_READY)); - return; - } - } - if (playObservation.ready().isCompletedExceptionally()) { - result.complete(TransferResult.failed("backend login or Forge negotiation failed")); - return; - } - if (relay == null || !playObservation.ready().isDone()) { - PendingTransfer waiting = new PendingTransfer(backendName, result); - pendingTransfer = waiting; - waiting.deadline = frontend.eventLoop().schedule(() -> { - if (pendingTransfer == waiting) { - pendingTransfer = null; - result.complete(TransferResult.failed("backend login or Forge negotiation timed out")); - } - }, 15, TimeUnit.SECONDS); - return; - } - BackendView target; - try { - target = owner.catalog().find(new BackendId(backendName)).orElse(null); - } catch (IllegalArgumentException invalidName) { - target = null; - } - if (target == null || !SessionChannels.isTcpAddress(target.address())) { - result.complete(TransferResult.of(TransferStatus.SERVER_UNAVAILABLE)); - return; - } - TransferAttempt attempt = new TransferAttempt(target, result, relay); - attempt.preparation = owner.selectTransferPreparation(); - if (attempt.preparation != null) { - Duration totalBudget = owner.eventTimeout().multipliedBy(2).plusSeconds(20) - .plus(owner.transferCutoverTimeout()).plus(FORGE_TRANSFER_HANDSHAKE_TIMEOUT); - attempt.context = new TransferContext(UUID.randomUUID(), identity, serverView(selected), serverView(target), - selected.handle().generation(), target.handle().generation(), - selected.owner().instanceGeneration(), target.owner().instanceGeneration(), - Instant.now().plus(totalBudget)); - attempt.totalDeadline = frontend.eventLoop().schedule(() -> { - if (currentAttempt(attempt)) failTransfer(attempt, "coordinated transfer timed out"); - }, totalBudget.toNanos(), TimeUnit.NANOSECONDS); - } - transfer = attempt; - LOGGER.debug("Starting replacement backend connection for player {} to {}", - view.username(), 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); } - }, attempt.preparation != null); - try { - Bootstrap bootstrap = SessionChannels.backendBootstrap(frontend, owner.transport(), - owner.backendResolver(), 5000, true, - pipeline -> pipeline.addLast("transfer-candidate", attempt.candidate)); - ChannelFuture connect = bootstrap.connect(SessionChannels.socketAddress(attempt.target.address())); - attempt.channel = connect.channel(); - connect.addListener(future -> { - if (attempt.finished || closed.get() || disconnecting) { connect.channel().close(); return; } - if (!future.isSuccess()) { - LOGGER.debug("Replacement backend connection failed for player {} to {}", - view.username(), attempt.target.address(), future.cause()); - failTransfer(attempt, "could not connect to replacement backend"); - return; - } - LOGGER.debug("Replacement backend TCP connection established for player {} to {}", - view.username(), attempt.target.address()); - if (attempt.preparation == null) writeBackendLogin(connect.channel()); - else prepareTransfer(attempt); - }); - } catch (RuntimeException failure) { - failTransfer(attempt, "could not start replacement backend connection"); - } - } - - private static ServerView serverView(BackendView backend) { - return new ServerView(backend.handle().id().value(), backend.address(), backend.tags(), backend.metadata()); - } - - private boolean currentAttempt(TransferAttempt attempt) { - return transfer == attempt && !attempt.finished && !closed.get() && !disconnecting; - } - - private boolean targetStillCurrent(TransferAttempt attempt) { - BackendView current = owner.catalog().find(attempt.target.handle().id()).orElse(null); - return current != null && current.handle().equals(attempt.target.handle()) - && current.owner().equals(attempt.target.owner()) && current.address().equals(attempt.target.address()) - && attempt.channel != null && attempt.channel.isActive(); - } - - private void prepareTransfer(TransferAttempt attempt) { - try { - CompletionStage preparation = attempt.preparation.prepare( - new TransferPreparingEvent(attempt.context)); - attempt.coordination = preparation.toCompletableFuture(); - preparation.whenComplete((prepared, failure) -> frontend.eventLoop().execute(() -> { - if (!currentAttempt(attempt)) return; - attempt.coordination = null; - if (failure != null || prepared == null) { - LOGGER.debug("Transfer preparation failed for {}", view.username(), failure); - failTransfer(attempt, "transfer preparation failed"); - } else if (!targetStillCurrent(attempt)) { - failTransfer(attempt, "replacement backend registration changed"); - } else if (prepared.requiresSourceRelease()) { - releaseTransferSource(attempt, prepared); - } else { - writeBackendLogin(attempt.channel); - } - })); - } catch (RuntimeException failure) { - failTransfer(attempt, "could not prepare backend transfer"); - } - } - - private void releaseTransferSource(TransferAttempt attempt, PreparedTransfer prepared) { - Channel source = backend; - pauseSource(attempt, source, "could not stop source gameplay for transfer", - "could not drain source relay", () -> { - if (!currentAttempt(attempt)) { - if (!closed.get() && !disconnecting) resumeAfterFailedTransfer(attempt); - return; - } - if (clientHasPartialFrame()) { - failTransfer(attempt, "client packet was incomplete at the transfer boundary"); - return; - } - if (!targetStillCurrent(attempt)) { - failTransfer(attempt, "replacement backend became unavailable"); - return; - } - attempt.oldRelay.detach().whenComplete((removed, detachFailure) -> frontend.eventLoop().execute(() -> { - if (detachFailure != null) { - boolean attached = frontend.pipeline().get("raw-relay") != null - && source.pipeline().get("raw-relay") != null; - failTransfer(attempt, "could not detach source relay"); - if (!attached) closePair(); - return; - } - attempt.detached = true; - if (!currentAttempt(attempt) || !targetStillCurrent(attempt)) { - failTransfer(attempt, "replacement backend closed during source release"); - closePair(); - return; - } - // Disarm exact old-channel EOF before closing it; no destination exists yet. - attempt.sourceReleased = true; - attempt.clientBuffer.discardFrames(); - attempt.oldBackendBuffer.discardFrames(); - backend = null; - relay = null; - view = new PlayerView(identity, view.username(), Optional.empty()); - owner.playerContextChanged(view); - if (source.pipeline().get("transfer-frame-handler") != null) { - source.pipeline().remove("transfer-frame-handler"); - } - source.close().addListener(closedSource -> { - if (!currentAttempt(attempt)) return; - if (!closedSource.isSuccess()) { - failTransfer(attempt, "could not close source connection"); - return; - } - confirmSourceReleased(attempt, prepared); - }); - })); - }); - } - - private void confirmSourceReleased(TransferAttempt attempt, PreparedTransfer prepared) { - try { - CompletionStage confirmation = prepared.sourceClosed(); - attempt.coordination = confirmation.toCompletableFuture(); - confirmation.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { - if (!currentAttempt(attempt)) return; - attempt.coordination = null; - if (failure != null) { - LOGGER.debug("Source release confirmation failed for {}", view.username(), failure); - failTransfer(attempt, "source release confirmation failed"); - } else if (!targetStillCurrent(attempt)) { - failTransfer(attempt, "replacement backend registration changed"); - } else writeBackendLogin(attempt.channel); - })); - } catch (RuntimeException failure) { - failTransfer(attempt, "could not confirm source release"); - } - } - - private void tryStartPendingTransfer() { - PendingTransfer waiting = pendingTransfer; - if (disconnecting || waiting == null || relay == null || !playObservation.ready().isDone()) return; - pendingTransfer = null; - waiting.deadline.cancel(false); - beginTransfer(waiting.backendName, waiting.result); - } - - private void writeBackendLogin(Channel channel) { - transfer.candidate.loginStarted(); - ByteBuf handshakeBody = encodeBackendHandshake( - transfer.target.handle().id().value(), transfer.target.owner().instanceGeneration()); - ByteBuf loginBody = new LoginStart(view.username()).encode(frontend.alloc(), ProtocolProfile.minecraft1710()); - channel.write(handshakeBody); - channel.writeAndFlush(loginBody).addListener(write -> { - if (write.isSuccess()) { - LOGGER.debug("Replacement backend login frames flushed for player {} to {}", - view.username(), channel.remoteAddress()); - } else { - LOGGER.debug("Replacement backend login write failed for player {}", - view.username(), write.cause()); - } - if (!write.isSuccess() && transfer != null && transfer.channel == channel) { - failTransfer(transfer, "could not write replacement backend login"); - } - }); - } - - private ByteBuf encodeForwardedHandshake(String backendName, long backendEpoch) { - byte[] secret = owner.sessionBindingSecret(backendName); - try { - return encodeForwardedHandshake(backendName, backendEpoch, verifiedProfile, secret); - } finally { - if (secret != null) Arrays.fill(secret, (byte) 0); - } - } private ByteBuf encodeBackendHandshake(String backendName, long backendEpoch) { - if (owner.onlineMode()) return encodeForwardedHandshake(backendName, backendEpoch); - byte[] secret = owner.sessionBindingSecret(backendName); - if (secret == null) return handshake.encode(frontend.alloc(), ProtocolProfile.minecraft1710()); - try { - // OFFLINE identity is derived by this proxy and is not a Mojang-authenticated profile. - // Only forward it when this backend has explicitly configured session binding. - VerifiedProfile offlineProfile = new VerifiedProfile(identity.playerId(), view.username(), List.of()); - return encodeForwardedHandshake(backendName, backendEpoch, offlineProfile, secret); - } finally { - Arrays.fill(secret, (byte) 0); - } + return BackendHandshakes.encode(owner, frontend, login.handshake(), identity, view.username(), login.verifiedProfile(), + backendName, backendEpoch); } - private ByteBuf encodeForwardedHandshake(String backendName, long backendEpoch, - VerifiedProfile profile, byte[] secret) { - String proof = null; - if (secret != null && backendEpoch > 0) { - long expiresAt = Math.addExact(System.currentTimeMillis(), ForwardedSessionProof.MAX_LIFETIME_MILLIS); - proof = ForwardedSessionProof.create(owner.proxyEpoch(), identity.playerId(), identity.connectionId(), - backendName, backendEpoch, UUID.randomUUID(), expiresAt, secret); - } - return BungeeLegacyForwarding.encode(frontend.alloc(), handshake, - (InetSocketAddress) frontend.remoteAddress(), profile, proof); - } - - private void candidateReady(TransferAttempt attempt) { - if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return; - if (attempt.preparation != null && !targetStillCurrent(attempt)) { - failTransfer(attempt, "replacement backend registration changed during login"); - return; - } - attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> { - if (transfer != attempt || attempt.result.isDone()) return; - attempt.result.complete(TransferResult.failed("replacement backend cutover timed out")); - closePair(); - }, owner.transferCutoverTimeout().toNanos(), TimeUnit.NANOSECONDS); - if (attempt.sourceReleased) { - if (clientHasPartialFrame()) { - failTransfer(attempt, "client packet was incomplete at replacement cutover"); - return; - } - activateCandidate(attempt); - return; - } - pauseSource(attempt, backend, "could not buffer frames for transfer", - "old backend stopped during transfer", () -> { - if (attempt.finished || closed.get() || disconnecting) { - if (disconnecting) return; - resumeAfterFailedTransfer(attempt); - return; - } - if (clientHasPartialFrame()) { - failTransfer(attempt, "client packet was incomplete at the transfer boundary"); - return; - } - if (!attempt.channel.isActive()) { - failTransfer(attempt, "replacement backend became unavailable"); - return; - } - attempt.oldRelay.detach().whenComplete((removed, detachFailure) -> - frontend.eventLoop().execute(() -> { - if (detachFailure != null) { - boolean oldRelayStillAttached = frontend.pipeline().get("raw-relay") != null - && backend.pipeline().get("raw-relay") != null; - failTransfer(attempt, "could not detach old backend relay"); - if (!oldRelayStillAttached) closePair(); - return; - } - if (closed.get() || disconnecting || !attempt.channel.isActive()) { - if (disconnecting) return; - failTransfer(attempt, "replacement backend closed during transfer"); - closePair(); - return; - } - attempt.detached = true; - activateCandidate(attempt); - })); - }); - } - - /** - * Buffers both old-link directions and pauses the old relay. {@code onPaused} runs on the session - * event loop after accepted writes drain; buffer or drain failures fail the attempt and close. - */ - private void pauseSource(TransferAttempt attempt, Channel source, String bufferFailure, String drainFailure, - Runnable onPaused) { - try { - attempt.clientBuffer = installTransferBuffer(frontend, "transfer-client-buffer"); - attempt.oldBackendBuffer = installTransferBuffer(source, "transfer-old-backend-buffer"); - } catch (RuntimeException failure) { - failTransfer(attempt, bufferFailure); - closePair(); - return; - } - attempt.pauseInProgress = true; - CompletionStage pause = attempt.oldRelay.pause(); - frontend.eventLoop().execute(attempt.clientBuffer::readUntilRemoved); - source.eventLoop().execute(attempt.oldBackendBuffer::readUntilRemoved); - pause.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { - attempt.pauseInProgress = false; - if (failure != null) { - failTransfer(attempt, drainFailure); - closePair(); - return; - } - attempt.paused = true; - onPaused.run(); - })); - } - - private TransferFrameBuffer installTransferBuffer(Channel channel, String name) { - var buffer = new TransferFrameBuffer(this::closePair); - channel.pipeline().addAfter(SessionChannels.frameDecoderName(channel.pipeline()), name, buffer); - return buffer; - } - - private boolean clientHasPartialFrame() { - return SessionChannels.frameDecoder(frontend.pipeline(), "client frame decoder missing during transfer") - .hasPartialFrame(); - } - - private void activateCandidate(TransferAttempt attempt) { - // Keep READY ownership until removal. A decoder can emit more target frames after - // JoinGame in the same read batch; marking HANDED_OFF early would silently drop them. - removeHandshakeCodecs(attempt.channel, attempt.candidate::handOff).whenComplete((ignored, failure) -> - frontend.eventLoop().execute(() -> { - if (failure != null || closed.get() || disconnecting || !attempt.channel.isActive()) { - if (disconnecting) return; - failTransfer(attempt, "replacement backend pipeline failed"); - closePair(); - return; - } - installCandidateRelay(attempt); - })); - } - - private void installCandidateRelay(TransferAttempt attempt) { - PlayObservation nextObservation = attempt.candidate.observation(); - RawRelay.Link next; - try { - if (clientEntityId == null) clientEntityId = playObservation.entityId().orElseThrow(); - attempt.frameState = new TransferFrameHandler.State(frontend, attempt.channel, nextObservation, - clientEntityId, attempt.candidate.joinGame(), this::closePair); - attempt.channel.pipeline().addLast("keep-alive-bridge", - new KeepAliveBridge(keepAlives, false, this::closePair)); - installTransferFrameHandlers(attempt.frameState, attempt.channel); - installTabCompletion(attempt.channel); - next = RawRelay.attach(frontend, attempt.channel, - bytes -> nextObservation.observeFrame(false, bytes), - bytes -> nextObservation.observeFrame(true, bytes)); - RawRelay.Link observedLink = next; - nextObservation.ready().whenComplete((ignored, failure) -> observedLink.stopObserving()); - } catch (RuntimeException failure) { - failTransfer(attempt, "could not attach replacement relay"); - closePair(); - return; - } - next.ready().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { - if (failure != null || closed.get() || disconnecting - || (attempt.preparation != null && !targetStillCurrent(attempt))) { - if (disconnecting) return; - failTransfer(attempt, "replacement relay was not ready"); - closePair(); - return; - } - ByteBuf opening; - try { - keepAlives.switchBackend(); - opening = transferOpening(attempt); - } catch (RuntimeException malformed) { - 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() - || (attempt.preparation != null && !targetStillCurrent(attempt))) { - if (disconnecting) return; - failTransfer(attempt, "could not send replacement world transition"); - closePair(); - return; - } - Channel oldBackend = backend; - Optional previousServer = attempt.sourceReleased - ? Optional.of(attempt.context.source().name()) : view.currentServer(); - PlayObservation oldObservation = playObservation; - TransferFrameHandler.State oldFrameState = frameState; - backend = attempt.channel; - selected = attempt.target; - playObservation = nextObservation; - relay = next; - frameState = attempt.frameState; - view = new PlayerView(identity, view.username(), selected.handle().id().value()); - owner.serverConnected(view, previousServer); - if (oldBackend != null) oldBackend.close(); - oldObservation.close(); - if (oldFrameState != null) oldFrameState.close(); - try { - attempt.clientBuffer.drainAndRemove(); - } catch (RuntimeException replayFailure) { - closePair(); - return; - } - if (closed.get()) return; - next.start(); - if (closed.get()) return; - if (nextObservation.forgeSeen()) { - attempt.cutoverDeadline.cancel(false); - attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> { - if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return; - 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()) { - failTransfer(attempt, "replacement Forge handshake failed"); - closePair(); - } else finishTransfer(attempt); - })); - } else finishTransfer(attempt); - })); - })); - } - - private void finishTransfer(TransferAttempt attempt) { - if (transfer != attempt || attempt.finished || closed.get() || disconnecting) return; - transfer = null; - attempt.finished = true; - attempt.cutoverDeadline.cancel(false); - cancelCoordination(attempt); - attempt.result.complete(TransferResult.of(TransferStatus.NETWORK_READY)); - } - - private void installTransferFrameHandlers(TransferFrameHandler.State state, Channel target) { - ChannelPipeline clientPipeline = frontend.pipeline(); - if (clientPipeline.get("transfer-frame-handler") != null) clientPipeline.remove("transfer-frame-handler"); - if (SessionChannels.frameDecoderName(clientPipeline) == null) { - throw new IllegalStateException("client frame decoder missing during transfer"); - } - clientPipeline.addLast("transfer-frame-handler", new TransferFrameHandler(state, false)); - target.pipeline().addLast("transfer-frame-handler", new TransferFrameHandler(state, true)); - } - - private ByteBuf transferOpening(TransferAttempt attempt) { - List queued = attempt.candidate.takeQueuedPackets(); - try { - return ByteBufs.fill(frontend.alloc().buffer(), output -> { - boolean nextForge = attempt.candidate.observation().forgeSeen(); - // FML's reset also restores the client's frozen registry. A Forge -> vanilla - // switch needs it even though the replacement server has no ServerHello. - // After that reset the client remains in HELLO until a later Forge switch. - if (!clientFmlAwaitingServerHello && (playObservation.forgeSeen() || nextForge)) { - ByteBuf reset = Minecraft1710PlayPackets.forgeReset(frontend.alloc()); - try { output.writeBytes(reset); } finally { reset.release(); } - clientFmlAwaitingServerHello = true; - } - if (nextForge) clientFmlAwaitingServerHello = false; - if (attempt.candidate.joinGame() != null) { - int targetDimension = attempt.candidate.observation().dimension() - .orElse(attempt.candidate.joinGame().dimension()); - ByteBuf respawns = Minecraft1710PlayPackets.respawnSequence(frontend.alloc(), - attempt.candidate.joinGame(), targetDimension); - try { output.writeBytes(respawns); } finally { respawns.release(); } - } - for (ByteBuf packet : queued) { - ByteBuf mapped = keepAlives.body(frontend.alloc(), packet, false); - if (mapped == null) continue; - try { - TransitionFrames.appendClientbound(output, frontend.alloc(), mapped, - attempt.candidate.joinGame() == null ? null : attempt.candidate.joinGame().entityId(), - clientEntityId); - } finally { - if (mapped != packet) mapped.release(); - } - } - }); - } finally { - queued.forEach(ByteBuf::release); - } - } - - private void failTransfer(TransferAttempt attempt, String reason) { - if (attempt.finished) return; - LOGGER.debug("Replacement backend failed for player {} to {}: {}; channel registered={}, active={}", - view.username(), attempt.target.address(), reason, - attempt.channel != null && attempt.channel.isRegistered(), - attempt.channel != null && attempt.channel.isActive()); - attempt.finished = true; - attempt.failureReason = reason; - cancelCoordination(attempt); - 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); - } - - private void resumeAfterFailedTransfer(TransferAttempt attempt) { - if (attempt.paused && !attempt.detached && !closed.get() && !disconnecting) { - attempt.oldRelay.resume().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { - if (failure != null) closePair(); - completeFailedTransfer(attempt); - })); - } else { - completeFailedTransfer(attempt); - } - } - - private void completeFailedTransfer(TransferAttempt attempt) { - if (!attempt.detached && !closed.get() && !disconnecting) { - try { - if (attempt.oldBackendBuffer != null) attempt.oldBackendBuffer.drainAndRemove(); - if (attempt.clientBuffer != null) attempt.clientBuffer.drainAndRemove(); - } catch (RuntimeException failure) { - closePair(); - } - } - if (attempt.cutoverDeadline != null) attempt.cutoverDeadline.cancel(false); - if (transfer == attempt) transfer = null; - attempt.result.complete(TransferResult.failed(attempt.failureReason)); - if (attempt.detached) closePair(); - } - - private static void cancelCoordination(TransferAttempt attempt) { - if (attempt.totalDeadline != null) attempt.totalDeadline.cancel(false); - CompletableFuture work = attempt.coordination; - attempt.coordination = null; - // Cancellation may run arbitrary CompletionStage continuations, never on session I/O. - if (work != null) Thread.startVirtualThread(() -> work.cancel(false)); - } + CompletionStage transferTo(String backendName) { return transfers.transferTo(backendName); } void closePair() { if (!closed.compareAndSet(false, true)) return; @@ -1539,7 +739,8 @@ final class Session extends ChannelInboundHandlerAdapter { if (initialLoginDeadline != null) initialLoginDeadline.cancel(false); if (initialPlayDeadline != null) initialPlayDeadline.cancel(false); if (disconnectDeadline != null) disconnectDeadline.cancel(false); - if (verification != null) verification.cancel(true); + login.cancel(); + router.cancel(); if (admissionRequest != null) { admissionRequest.cancel(false); admissionRequest = null; @@ -1552,24 +753,7 @@ final class Session extends ChannelInboundHandlerAdapter { Channel upstream = backend; if (upstream != null) upstream.close(); if (playObservation != null) playObservation.close(); - if (frameState != null) frameState.close(); - PendingTransfer waiting = pendingTransfer; - if (waiting != null) { - pendingTransfer = null; - waiting.deadline.cancel(false); - waiting.result.complete(TransferResult.failed("player session closed during transfer")); - } - TransferAttempt activeTransfer = transfer; - if (activeTransfer != null) { - transfer = null; - activeTransfer.finished = true; - cancelCoordination(activeTransfer); - 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")); - } + transfers.close(); if (identityClaimed) { owner.releasePermissions(identity); owner.releaseIdentity(identity.playerId(), this); @@ -1598,12 +782,12 @@ final class Session extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) { if (ctx.channel() != frontend && ctx.channel() != backend) return; - if (!published && loginStart != null) { + if (!published && login.loginStart() != null) { LOGGER.debug("{} channel closed during initial login for player {} with backend {}", - ctx.channel() == frontend ? "Client" : "Backend", loginStart.username(), + ctx.channel() == frontend ? "Client" : "Backend", login.loginStart().username(), selected == null ? "" : selected.handle().id().value()); } - if (ctx.channel() == backend && backendConnected && !published && loginStart != null + if (ctx.channel() == backend && backendConnected && !published && login.loginStart() != null && !loginDisconnectStarted) { disconnectLogin("Selected server closed during login."); return; @@ -1627,4 +811,49 @@ final class Session extends ChannelInboundHandlerAdapter { return Optional.of(view); } + /** Adapter exposing this session's connection state to its {@link TransferCoordinator}. */ + private final class Host implements TransferHost { + @Override public Channel frontend() { return frontend; } + @Override public Channel backend() { return backend; } + @Override public RawRelay.Link relay() { return relay; } + @Override public BackendView selected() { return selected; } + @Override public PlayerView view() { return view; } + @Override public PlayerIdentity identity() { return identity; } + @Override public PlayObservation observation() { return playObservation; } + @Override public KeepAliveBridge.State keepAlives() { return keepAlives; } + @Override public boolean published() { return published; } + @Override public boolean closed() { return closed.get(); } + @Override public boolean disconnecting() { return disconnecting; } + @Override public void closeSession() { closePair(); } + @Override public BackendCatalog catalog() { return owner.catalog(); } + @Override public TransferPreparation selectPreparation() { return owner.selectTransferPreparation(); } + @Override public Duration eventTimeout() { return owner.eventTimeout(); } + @Override public Duration cutoverTimeout() { return owner.transferCutoverTimeout(); } + @Override public NetworkTransport transport() { return owner.transport(); } + @Override public AddressResolverGroup resolver() { return owner.backendResolver(); } + @Override public ByteBuf backendHandshake(String backendName, long backendEpoch) { + return encodeBackendHandshake(backendName, backendEpoch); + } + @Override public void installTabCompletion(Channel target) { Session.this.installTabCompletion(target); } + @Override public CompletableFuture swapHandshakeCodecs(Channel channel, Runnable beforeRemoval) { + return removeHandshakeCodecs(channel, beforeRemoval); + } + + @Override public void sourceReleased() { + backend = null; + relay = null; + view = new PlayerView(identity, view.username(), Optional.empty()); + owner.playerContextChanged(view); + } + + @Override public void cutover(Channel next, BackendView target, PlayObservation observation, + RawRelay.Link link, Optional previousServer) { + backend = next; + selected = target; + playObservation = observation; + relay = link; + view = new PlayerView(identity, view.username(), selected.handle().id().value()); + owner.serverConnected(view, previousServer); + } + } } diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/SessionChannels.java b/proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionChannels.java similarity index 81% rename from proxy-core/src/main/java/dev/moonbridge/core/session/SessionChannels.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionChannels.java index 7605bb7..68a8cd5 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/SessionChannels.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionChannels.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.channel; import dev.moonbridge.core.net.NetworkTransport; import dev.moonbridge.core.protocol.MinecraftFrameDecoder; @@ -17,19 +17,19 @@ import java.net.URI; import java.util.function.Consumer; /** Backend dialing and pipeline lookups shared by initial login and transfer. */ -final class SessionChannels { - static final String FRAME_DECODER = "minecraft-frame-decoder"; - static final String ENCRYPTED_FRAME_DECODER = "encrypted-frame-decoder"; - static final String FRAME_ENCODER = "minecraft-frame-encoder"; +public final class SessionChannels { + public static final String FRAME_DECODER = "minecraft-frame-decoder"; + public static final String ENCRYPTED_FRAME_DECODER = "encrypted-frame-decoder"; + public static final String FRAME_ENCODER = "minecraft-frame-encoder"; private SessionChannels() { } - static boolean isTcpAddress(URI address) { + public static boolean isTcpAddress(URI address) { return "tcp".equalsIgnoreCase(address.getScheme()) && address.getHost() != null && address.getPort() >= 1 && address.getPort() <= 65535; } - static InetSocketAddress socketAddress(URI address) { + public static InetSocketAddress socketAddress(URI address) { String host = address.getHost(); if (host.startsWith("[") && host.endsWith("]")) { host = host.substring(1, host.length() - 1); @@ -42,7 +42,7 @@ final class SessionChannels { } /** Dials on the player's event loop so session state stays confined to one thread. */ - static Bootstrap backendBootstrap(Channel frontend, NetworkTransport transport, + public static Bootstrap backendBootstrap(Channel frontend, NetworkTransport transport, AddressResolverGroup resolver, int connectTimeoutMillis, boolean autoRead, Consumer extraHandlers) { @@ -62,19 +62,19 @@ final class SessionChannels { }); } - static void installCodecs(ChannelPipeline pipeline) { + public static void installCodecs(ChannelPipeline pipeline) { pipeline.addLast(FRAME_DECODER, new MinecraftFrameDecoder(ProtocolProfile.minecraft1710(), true)); pipeline.addLast(FRAME_ENCODER, new SessionFrameEncoder()); } /** Name of the active frame decoder: the encrypted one replaces the plain one in online mode. */ - static String frameDecoderName(ChannelPipeline pipeline) { + public static String frameDecoderName(ChannelPipeline pipeline) { if (pipeline.get(ENCRYPTED_FRAME_DECODER) != null) return ENCRYPTED_FRAME_DECODER; if (pipeline.get(FRAME_DECODER) != null) return FRAME_DECODER; return null; } - static MinecraftFrameDecoder frameDecoder(ChannelPipeline pipeline, String failure) { + public static MinecraftFrameDecoder frameDecoder(ChannelPipeline pipeline, String failure) { String name = frameDecoderName(pipeline); if (name == null || !(pipeline.get(name) instanceof MinecraftFrameDecoder decoder)) { throw new IllegalStateException(failure); diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/SessionFrameEncoder.java b/proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionFrameEncoder.java similarity index 79% rename from proxy-core/src/main/java/dev/moonbridge/core/session/SessionFrameEncoder.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionFrameEncoder.java index c588eb9..4618fef 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/SessionFrameEncoder.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionFrameEncoder.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.channel; import dev.moonbridge.core.protocol.MinecraftFrameEncoder; import dev.moonbridge.core.protocol.ProtocolProfile; @@ -8,7 +8,7 @@ import io.netty.handler.codec.MessageToMessageEncoder; import java.util.List; -final class SessionFrameEncoder extends MessageToMessageEncoder { +public final class SessionFrameEncoder extends MessageToMessageEncoder { @Override protected void encode(ChannelHandlerContext ctx, ByteBuf message, List output) { output.add(MinecraftFrameEncoder.encode(ctx.alloc(), message, ProtocolProfile.minecraft1710())); } diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/CommandRateLimiter.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/CommandRateLimiter.java similarity index 67% rename from proxy-core/src/main/java/dev/moonbridge/core/session/CommandRateLimiter.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/CommandRateLimiter.java index 11df196..a5bf076 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/CommandRateLimiter.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/CommandRateLimiter.java @@ -1,25 +1,25 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import java.util.concurrent.TimeUnit; /** Token bucket for proxy-owned player commands; confined to the session event loop. */ -final class CommandRateLimiter { - enum Decision { ADMIT, DENY, DENY_WITH_NOTICE } +public final class CommandRateLimiter { + public enum Decision { ADMIT, DENY, DENY_WITH_NOTICE } - static final int BURST = 10; - static final long TOKEN_NANOS = TimeUnit.MILLISECONDS.toNanos(200); - static final long NOTICE_NANOS = TimeUnit.SECONDS.toNanos(2); + public static final int BURST = 10; + public static final long TOKEN_NANOS = TimeUnit.MILLISECONDS.toNanos(200); + public static final long NOTICE_NANOS = TimeUnit.SECONDS.toNanos(2); private int tokens = BURST; private long refillNanos; private long lastNoticeNanos; private boolean noticeSent; - CommandRateLimiter(long nowNanos) { + public CommandRateLimiter(long nowNanos) { refillNanos = nowNanos; } - Decision admit(long nowNanos) { + public Decision admit(long nowNanos) { long elapsed = nowNanos - refillNanos; if (elapsed >= TOKEN_NANOS) { long replenished = Math.min(BURST, elapsed / TOKEN_NANOS); diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/FrameTransformHandler.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/FrameTransformHandler.java similarity index 88% rename from proxy-core/src/main/java/dev/moonbridge/core/session/FrameTransformHandler.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/FrameTransformHandler.java index f116591..21982f5 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/FrameTransformHandler.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/FrameTransformHandler.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.relay.RawRelay; import io.netty.buffer.ByteBuf; @@ -11,13 +11,13 @@ import io.netty.util.ReferenceCountUtil; * An inbound PLAY handler that sees one complete frame at a time and owns its reference counting, so * subclasses never call {@code release()}: {@link #transform} borrows the frame and answers what to forward. */ -abstract class FrameTransformHandler extends ChannelInboundHandlerAdapter { +public abstract class FrameTransformHandler extends ChannelInboundHandlerAdapter { /** Returned by {@link #transform} when the frame was consumed and the relay resumes on its own later. */ - static final ByteBuf CONSUMED = Unpooled.EMPTY_BUFFER; + public static final ByteBuf CONSUMED = Unpooled.EMPTY_BUFFER; private final Runnable closeSession; - FrameTransformHandler(Runnable closeSession) { + public FrameTransformHandler(Runnable closeSession) { this.closeSession = closeSession; } diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/KeepAliveBridge.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/KeepAliveBridge.java similarity index 86% rename from proxy-core/src/main/java/dev/moonbridge/core/session/KeepAliveBridge.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/KeepAliveBridge.java index fdd4be0..0315e9d 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/KeepAliveBridge.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/KeepAliveBridge.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufAllocator; @@ -9,18 +9,18 @@ import java.util.Map; import java.util.concurrent.ThreadLocalRandom; /** Keeps connection-local PLAY keep-alive replies on the backend that issued them. */ -final class KeepAliveBridge extends FrameTransformHandler { - static final class State { +public final class KeepAliveBridge extends FrameTransformHandler { + public static final class State { private static final int MAX_PENDING = 1024; private int nextClientId = ThreadLocalRandom.current().nextInt(); private final Map pending = new HashMap<>(); - void switchBackend() { + public void switchBackend() { pending.clear(); } /** Transforms a packet body while the login frame encoder is still installed. */ - ByteBuf body(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) { + public ByteBuf body(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) { int start = packet.readerIndex(); if (packet.readableBytes() == 0 || packet.getUnsignedByte(start) != 0) return packet; if (packet.readableBytes() != 5) throw new IllegalArgumentException("invalid keep-alive packet"); @@ -30,7 +30,7 @@ final class KeepAliveBridge extends FrameTransformHandler { } /** Transforms a complete length-prefixed PLAY frame without copying ordinary packets. */ - ByteBuf frame(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) { + public ByteBuf frame(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) { int start = packet.readerIndex(); if (packet.readableBytes() < 2 || packet.getUnsignedByte(start) != 5 || packet.getUnsignedByte(start + 1) != 0) return packet; @@ -60,7 +60,7 @@ final class KeepAliveBridge extends FrameTransformHandler { private final boolean fromFrontend; private final Runnable closeSession; - KeepAliveBridge(State state, boolean fromFrontend, Runnable closeSession) { + public KeepAliveBridge(State state, boolean fromFrontend, Runnable closeSession) { super(closeSession); this.state = state; this.fromFrontend = fromFrontend; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/PlayObservation.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayObservation.java similarity index 86% rename from proxy-core/src/main/java/dev/moonbridge/core/session/PlayObservation.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayObservation.java index f314477..0ac66e3 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/PlayObservation.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayObservation.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.ProtocolProfile; @@ -9,7 +9,7 @@ import java.util.OptionalInt; import java.util.concurrent.CompletableFuture; /** PLAY facts for a session; callers run on the player's event loop. */ -final class PlayObservation implements AutoCloseable { +public final class PlayObservation implements AutoCloseable { private static final int MAX_FRAME_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes(); private final CompletableFuture ready = new CompletableFuture<>(); private boolean joinSeen; @@ -20,7 +20,7 @@ final class PlayObservation implements AutoCloseable { private OptionalInt forgeDimensionOverride = OptionalInt.empty(); private OptionalInt entityId = OptionalInt.empty(); /** Called after MinecraftFrameDecoder with one complete, length-prefixed frame. */ - void observeFrame(boolean clientbound, ByteBuf frame) { + public void observeFrame(boolean clientbound, ByteBuf frame) { if (ready.isDone()) return; ByteBuf packet = frame.duplicate(); int length = ProtocolVarInt.read(packet); @@ -30,7 +30,7 @@ final class PlayObservation implements AutoCloseable { observePacket(clientbound, packet); } - void observePacket(boolean clientbound, ByteBuf packet) { + public void observePacket(boolean clientbound, ByteBuf packet) { if (ready.isDone()) return; int id = ProtocolVarInt.read(packet.duplicate()); if (clientbound) { @@ -65,10 +65,10 @@ final class PlayObservation implements AutoCloseable { if (joinSeen && forgeSeen && backendComplete && clientComplete) ready.complete(null); } - CompletableFuture ready() { return ready; } - boolean forgeSeen() { return forgeSeen; } - OptionalInt dimension() { return dimension; } - OptionalInt entityId() { return entityId; } + public CompletableFuture ready() { return ready; } + public boolean forgeSeen() { return forgeSeen; } + public OptionalInt dimension() { return dimension; } + public OptionalInt entityId() { return entityId; } @Override public void close() { if (!ready.isDone()) ready.completeExceptionally(new IllegalStateException("session closed")); diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/PlayerCommandInterceptor.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayerCommandInterceptor.java similarity index 75% rename from proxy-core/src/main/java/dev/moonbridge/core/session/PlayerCommandInterceptor.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayerCommandInterceptor.java index b844dda..656dd15 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/PlayerCommandInterceptor.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayerCommandInterceptor.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import io.netty.buffer.ByteBuf; @@ -7,10 +7,10 @@ import io.netty.channel.ChannelHandlerContext; import java.util.function.Predicate; /** Consumes only registered proxy commands; all other PLAY frames keep their wire bytes. */ -final class PlayerCommandInterceptor extends FrameTransformHandler { +public final class PlayerCommandInterceptor extends FrameTransformHandler { private final Predicate dispatch; - PlayerCommandInterceptor(Predicate dispatch, Runnable closeSession) { + public PlayerCommandInterceptor(Predicate dispatch, Runnable closeSession) { super(closeSession); this.dispatch = dispatch; } diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TabCompletionBridge.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/TabCompletionBridge.java similarity index 94% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TabCompletionBridge.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/TabCompletionBridge.java index f5e1fa2..9b87270 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TabCompletionBridge.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/TabCompletionBridge.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.protocol.MinecraftTabCompletion; import dev.moonbridge.core.relay.RawRelay; @@ -20,7 +20,7 @@ import java.util.function.Function; import java.util.function.Predicate; /** One backend generation, confined to the shared frontend/backend event loop. */ -final class TabCompletionBridge implements AutoCloseable { +public final class TabCompletionBridge implements AutoCloseable { private final Channel frontend; private final Function> roots; private final Function>>> complete; @@ -44,13 +44,13 @@ final class TabCompletionBridge implements AutoCloseable { private ScheduledFuture timeout; private CompletableFuture pluginResult; - TabCompletionBridge(Channel frontend, Function> roots, + public TabCompletionBridge(Channel frontend, Function> roots, Function>>> complete, Runnable closeSession) { this(frontend, roots, complete, closeSession, ignored -> true); } - TabCompletionBridge(Channel frontend, Function> roots, + public TabCompletionBridge(Channel frontend, Function> roots, Function>>> complete, Runnable closeSession, Predicate visibleRoot) { this.frontend = frontend; @@ -60,8 +60,8 @@ final class TabCompletionBridge implements AutoCloseable { this.visibleRoot = visibleRoot; } - ChannelInboundHandlerAdapter frontendHandler() { return new Handler(true); } - ChannelInboundHandlerAdapter backendHandler() { return new Handler(false); } + public ChannelInboundHandlerAdapter frontendHandler() { return new Handler(true); } + public ChannelInboundHandlerAdapter backendHandler() { return new Handler(false); } private final class Handler extends ChannelInboundHandlerAdapter { private final boolean fromClient; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TransitionBuffer.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionBuffer.java similarity index 73% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TransitionBuffer.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionBuffer.java index 7cd97ff..89ceb2f 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TransitionBuffer.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionBuffer.java @@ -1,25 +1,25 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import io.netty.buffer.ByteBuf; import java.util.ArrayDeque; /** Bounded queue of frames read while the relay starts; it owns every buffer it holds until polled. */ -final class TransitionBuffer { - record Frame(boolean fromFrontend, ByteBuf payload) { } +public final class TransitionBuffer { + public record Frame(boolean fromFrontend, ByteBuf payload) { } private final int maxFrames; private final int maxBytes; private final ArrayDeque frames = new ArrayDeque<>(); private int bytes; - TransitionBuffer(int maxFrames, int maxBytes) { + public TransitionBuffer(int maxFrames, int maxBytes) { this.maxFrames = maxFrames; this.maxBytes = maxBytes; } /** Retains a duplicate of {@code packet}; returns false, retaining nothing, when a bound would be exceeded. */ - boolean add(boolean fromFrontend, ByteBuf packet) { + public boolean add(boolean fromFrontend, ByteBuf packet) { int size = packet.readableBytes(); if (frames.size() >= maxFrames || bytes + size > maxBytes) return false; frames.addLast(new Frame(fromFrontend, packet.retainedDuplicate())); @@ -27,19 +27,19 @@ final class TransitionBuffer { return true; } - boolean isEmpty() { return frames.isEmpty(); } + public boolean isEmpty() { return frames.isEmpty(); } - int size() { return frames.size(); } + public int size() { return frames.size(); } /** Transfers ownership of the oldest frame to the caller, or returns null when empty. */ - Frame poll() { + public Frame poll() { Frame frame = frames.pollFirst(); if (frame != null) bytes -= frame.payload().readableBytes(); return frame; } /** Releases everything still queued. */ - void release() { + public void release() { Frame frame; while ((frame = frames.pollFirst()) != null) frame.payload().release(); bytes = 0; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TransitionFrames.java b/proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionFrames.java similarity index 86% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TransitionFrames.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionFrames.java index 1b96de7..e1246c1 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TransitionFrames.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionFrames.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.protocol.ByteBufs; import dev.moonbridge.core.protocol.Minecraft1710EntityIds; @@ -8,14 +8,14 @@ import io.netty.buffer.ByteBufAllocator; import io.netty.channel.Channel; /** Maps frames buffered during relay start-up before they are written to their destination. */ -final class TransitionFrames { +public final class TransitionFrames { private TransitionFrames() { } /** * Takes ownership of {@code payload} and returns the buffer to send, or null when the frame is * swallowed. The payload is released on every failure, so callers never touch it after an exception. */ - static ByteBuf map(PlayObservation observation, KeepAliveBridge.State keepAlives, boolean fromFrontend, + public static ByteBuf map(PlayObservation observation, KeepAliveBridge.State keepAlives, boolean fromFrontend, ByteBuf payload, Channel target) { try { if (observation != null) observation.observePacket(!fromFrontend, payload); @@ -33,7 +33,7 @@ final class TransitionFrames { * Frames {@code mapped} (borrowed), swaps the player's entity ids when the replacement backend assigned a * different one, and appends the result to {@code output}. Every temporary buffer is released. */ - static void appendClientbound(ByteBuf output, ByteBufAllocator allocator, ByteBuf mapped, + public static void appendClientbound(ByteBuf output, ByteBufAllocator allocator, ByteBuf mapped, Integer serverEntityId, Integer clientEntityId) { ByteBufs.use(Minecraft1710PlayPackets.frame(allocator, mapped), frame -> { ByteBuf outgoing = serverEntityId == null ? frame diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/PendingTransfer.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/PendingTransfer.java similarity index 92% rename from proxy-core/src/main/java/dev/moonbridge/core/session/PendingTransfer.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/transfer/PendingTransfer.java index 5fe8366..63a4c6f 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/PendingTransfer.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/PendingTransfer.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; import dev.moonbridge.api.TransferResult; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferAttempt.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferAttempt.java similarity index 96% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TransferAttempt.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferAttempt.java index 4bb2fb1..b2dc0b4 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferAttempt.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferAttempt.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; import dev.moonbridge.api.TransferResult; import dev.moonbridge.api.event.TransferContext; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferCandidate.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCandidate.java similarity index 90% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TransferCandidate.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCandidate.java index 904d17b..b295fbe 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferCandidate.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCandidate.java @@ -1,4 +1,6 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; + +import dev.moonbridge.core.session.play.PlayObservation; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.MinecraftLoginSuccess; @@ -17,8 +19,8 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; /** Logs into a replacement backend while the existing player session keeps its old backend. */ -final class TransferCandidate extends ChannelInboundHandlerAdapter { - interface Listener { +public final class TransferCandidate extends ChannelInboundHandlerAdapter { + public interface Listener { void ready(TransferCandidate candidate); void failed(TransferCandidate candidate, String reason); } @@ -38,11 +40,11 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter { private State state = State.LOGIN; private int queuedBytes; - TransferCandidate(UUID expectedId, String username, Listener listener) { + public TransferCandidate(UUID expectedId, String username, Listener listener) { this(expectedId, username, listener, false); } - TransferCandidate(UUID expectedId, String username, Listener listener, boolean deferLogin) { + public TransferCandidate(UUID expectedId, String username, Listener listener, boolean deferLogin) { this.expectedId = expectedId; this.username = username; this.listener = listener; @@ -55,7 +57,7 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter { } /** Coordinated waits must not consume the destination's login budget. */ - void loginStarted() { + public void loginStarted() { if (loginStarted) return; if (context == null || !context.executor().inEventLoop()) { throw new IllegalStateException("candidate login must start on its event loop"); @@ -141,23 +143,23 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter { else fail(ctx, "replacement backend failed"); } - Minecraft1710PlayPackets.JoinGame joinGame() { return joinGame; } - PlayObservation observation() { return observation; } + public Minecraft1710PlayPackets.JoinGame joinGame() { return joinGame; } + public PlayObservation observation() { return observation; } - List takeQueuedPackets() { + public List takeQueuedPackets() { var packets = new ArrayList<>(queued); queued.clear(); queuedBytes = 0; return packets; } - void handOff() { + public void handOff() { if (state != State.READY) throw new IllegalStateException("replacement backend is not ready"); state = State.HANDED_OFF; deadline.cancel(false); } - void close() { + public void close() { if (deadline != null) deadline.cancel(false); ByteBuf packet; while ((packet = queued.pollFirst()) != null) packet.release(); diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCoordinator.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCoordinator.java new file mode 100644 index 0000000..0dcb823 --- /dev/null +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCoordinator.java @@ -0,0 +1,629 @@ +package dev.moonbridge.core.session.transfer; + +import dev.moonbridge.api.ServerView; +import dev.moonbridge.api.TransferResult; +import dev.moonbridge.api.TransferStatus; +import dev.moonbridge.api.event.TransferContext; +import dev.moonbridge.api.event.TransferPreparingEvent; +import dev.moonbridge.core.backend.BackendId; +import dev.moonbridge.core.backend.BackendView; +import dev.moonbridge.core.event.PreparedTransfer; +import dev.moonbridge.core.protocol.ByteBufs; +import dev.moonbridge.core.protocol.LoginStart; +import dev.moonbridge.core.protocol.ProtocolProfile; +import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; +import dev.moonbridge.core.relay.RawRelay; +import dev.moonbridge.core.session.channel.SessionChannels; +import dev.moonbridge.core.session.play.KeepAliveBridge; +import dev.moonbridge.core.session.play.PlayObservation; +import dev.moonbridge.core.session.play.TransitionFrames; +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelPipeline; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; + +/** + * Moves a connected player to another backend while the old link keeps serving: dials and logs into the + * candidate, buffers both directions, swaps relays and rewrites the player's entity ids. Everything runs on + * the player's frontend event loop; the session is reached only through {@link TransferHost}. + */ +public final class TransferCoordinator { + private static final Logger LOGGER = LoggerFactory.getLogger(TransferCoordinator.class); + private static final Duration FORGE_TRANSFER_HANDSHAKE_TIMEOUT = Duration.ofSeconds(30); + + private final TransferHost host; + private final Channel frontend; + private PendingTransfer pendingTransfer; + private TransferAttempt transfer; + private TransferFrameHandler.State frameState; + private Integer clientEntityId; + private boolean clientFmlAwaitingServerHello; + + public TransferCoordinator(TransferHost host) { + this.host = host; + this.frontend = host.frontend(); + } + + /** True while a transfer is queued or running; other writers to the client must wait. */ + public boolean busy() { return transfer != null || pendingTransfer != null; } + + public CompletionStage transferTo(String backendName) { + CompletableFuture result = new CompletableFuture<>(); + Runnable command = () -> { + try { + beginTransfer(backendName, result); + } catch (RuntimeException failure) { + result.completeExceptionally(failure); + host.closeSession(); + } + }; + if (frontend.eventLoop().inEventLoop()) command.run(); + else { + try { + frontend.eventLoop().execute(command); + } catch (RejectedExecutionException shutdown) { + result.complete(TransferResult.of(TransferStatus.PLAYER_NOT_CONNECTED)); + } + } + return result; + } + + private void beginTransfer(String backendName, CompletableFuture result) { + if (host.closed() || host.disconnecting() || !host.published()) { + result.complete(TransferResult.of(TransferStatus.PLAYER_NOT_CONNECTED)); + return; + } + if (transfer != null || pendingTransfer != null) { + result.complete(TransferResult.failed("a backend transfer is already in progress")); + return; + } + if (host.selected().handle().id().value().equals(backendName)) { + BackendView current = host.catalog().find(host.selected().handle().id()).orElse(null); + if (current != null && current.address().equals(host.selected().address())) { + result.complete(TransferResult.of(TransferStatus.NETWORK_READY)); + return; + } + } + if (host.observation().ready().isCompletedExceptionally()) { + result.complete(TransferResult.failed("backend login or Forge negotiation failed")); + return; + } + if (host.relay() == null || !host.observation().ready().isDone()) { + PendingTransfer waiting = new PendingTransfer(backendName, result); + pendingTransfer = waiting; + waiting.deadline = frontend.eventLoop().schedule(() -> { + if (pendingTransfer == waiting) { + pendingTransfer = null; + result.complete(TransferResult.failed("backend login or Forge negotiation timed out")); + } + }, 15, TimeUnit.SECONDS); + return; + } + BackendView target; + try { + target = host.catalog().find(new BackendId(backendName)).orElse(null); + } catch (IllegalArgumentException invalidName) { + target = null; + } + if (target == null || !SessionChannels.isTcpAddress(target.address())) { + result.complete(TransferResult.of(TransferStatus.SERVER_UNAVAILABLE)); + return; + } + TransferAttempt attempt = new TransferAttempt(target, result, host.relay()); + attempt.preparation = host.selectPreparation(); + if (attempt.preparation != null) { + Duration totalBudget = host.eventTimeout().multipliedBy(2).plusSeconds(20) + .plus(host.cutoverTimeout()).plus(FORGE_TRANSFER_HANDSHAKE_TIMEOUT); + attempt.context = new TransferContext(UUID.randomUUID(), host.identity(), serverView(host.selected()), serverView(target), + host.selected().handle().generation(), target.handle().generation(), + host.selected().owner().instanceGeneration(), target.owner().instanceGeneration(), + Instant.now().plus(totalBudget)); + attempt.totalDeadline = frontend.eventLoop().schedule(() -> { + if (currentAttempt(attempt)) failTransfer(attempt, "coordinated transfer timed out"); + }, totalBudget.toNanos(), TimeUnit.NANOSECONDS); + } + transfer = attempt; + LOGGER.debug("Starting replacement backend connection for player {} to {}", + host.view().username(), target.address()); + attempt.candidate = new TransferCandidate(host.identity().playerId(), host.view().username(), new TransferCandidate.Listener() { + @Override public void ready(TransferCandidate candidate) { candidateReady(attempt); } + @Override public void failed(TransferCandidate candidate, String reason) { failTransfer(attempt, reason); } + }, attempt.preparation != null); + try { + Bootstrap bootstrap = SessionChannels.backendBootstrap(frontend, host.transport(), + host.resolver(), 5000, true, + pipeline -> pipeline.addLast("transfer-candidate", attempt.candidate)); + ChannelFuture connect = bootstrap.connect(SessionChannels.socketAddress(attempt.target.address())); + attempt.channel = connect.channel(); + connect.addListener(future -> { + if (attempt.finished || host.closed() || host.disconnecting()) { connect.channel().close(); return; } + if (!future.isSuccess()) { + LOGGER.debug("Replacement backend connection failed for player {} to {}", + host.view().username(), attempt.target.address(), future.cause()); + failTransfer(attempt, "could not connect to replacement backend"); + return; + } + LOGGER.debug("Replacement backend TCP connection established for player {} to {}", + host.view().username(), attempt.target.address()); + if (attempt.preparation == null) writeBackendLogin(connect.channel()); + else prepareTransfer(attempt); + }); + } catch (RuntimeException failure) { + failTransfer(attempt, "could not start replacement backend connection"); + } + } + + private static ServerView serverView(BackendView backend) { + return new ServerView(backend.handle().id().value(), backend.address(), backend.tags(), backend.metadata()); + } + + private boolean currentAttempt(TransferAttempt attempt) { + return transfer == attempt && !attempt.finished && !host.closed() && !host.disconnecting(); + } + + private boolean targetStillCurrent(TransferAttempt attempt) { + BackendView current = host.catalog().find(attempt.target.handle().id()).orElse(null); + return current != null && current.handle().equals(attempt.target.handle()) + && current.owner().equals(attempt.target.owner()) && current.address().equals(attempt.target.address()) + && attempt.channel != null && attempt.channel.isActive(); + } + + private void prepareTransfer(TransferAttempt attempt) { + try { + CompletionStage preparation = attempt.preparation.prepare( + new TransferPreparingEvent(attempt.context)); + attempt.coordination = preparation.toCompletableFuture(); + preparation.whenComplete((prepared, failure) -> frontend.eventLoop().execute(() -> { + if (!currentAttempt(attempt)) return; + attempt.coordination = null; + if (failure != null || prepared == null) { + LOGGER.debug("Transfer preparation failed for {}", host.view().username(), failure); + failTransfer(attempt, "transfer preparation failed"); + } else if (!targetStillCurrent(attempt)) { + failTransfer(attempt, "replacement backend registration changed"); + } else if (prepared.requiresSourceRelease()) { + releaseTransferSource(attempt, prepared); + } else { + writeBackendLogin(attempt.channel); + } + })); + } catch (RuntimeException failure) { + failTransfer(attempt, "could not prepare backend transfer"); + } + } + + private void releaseTransferSource(TransferAttempt attempt, PreparedTransfer prepared) { + Channel source = host.backend(); + pauseSource(attempt, source, "could not stop source gameplay for transfer", + "could not drain source relay", () -> { + if (!currentAttempt(attempt)) { + if (!host.closed() && !host.disconnecting()) resumeAfterFailedTransfer(attempt); + return; + } + if (clientHasPartialFrame()) { + failTransfer(attempt, "client packet was incomplete at the transfer boundary"); + return; + } + if (!targetStillCurrent(attempt)) { + failTransfer(attempt, "replacement backend became unavailable"); + return; + } + attempt.oldRelay.detach().whenComplete((removed, detachFailure) -> frontend.eventLoop().execute(() -> { + if (detachFailure != null) { + boolean attached = frontend.pipeline().get("raw-relay") != null + && source.pipeline().get("raw-relay") != null; + failTransfer(attempt, "could not detach source relay"); + if (!attached) host.closeSession(); + return; + } + attempt.detached = true; + if (!currentAttempt(attempt) || !targetStillCurrent(attempt)) { + failTransfer(attempt, "replacement backend closed during source release"); + host.closeSession(); + return; + } + // Disarm exact old-channel EOF before closing it; no destination exists yet. + attempt.sourceReleased = true; + attempt.clientBuffer.discardFrames(); + attempt.oldBackendBuffer.discardFrames(); + host.sourceReleased(); + if (source.pipeline().get("transfer-frame-handler") != null) { + source.pipeline().remove("transfer-frame-handler"); + } + source.close().addListener(closedSource -> { + if (!currentAttempt(attempt)) return; + if (!closedSource.isSuccess()) { + failTransfer(attempt, "could not close source connection"); + return; + } + confirmSourceReleased(attempt, prepared); + }); + })); + }); + } + + private void confirmSourceReleased(TransferAttempt attempt, PreparedTransfer prepared) { + try { + CompletionStage confirmation = prepared.sourceClosed(); + attempt.coordination = confirmation.toCompletableFuture(); + confirmation.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { + if (!currentAttempt(attempt)) return; + attempt.coordination = null; + if (failure != null) { + LOGGER.debug("Source release confirmation failed for {}", host.view().username(), failure); + failTransfer(attempt, "source release confirmation failed"); + } else if (!targetStillCurrent(attempt)) { + failTransfer(attempt, "replacement backend registration changed"); + } else writeBackendLogin(attempt.channel); + })); + } catch (RuntimeException failure) { + failTransfer(attempt, "could not confirm source release"); + } + } + + /** Starts the queued transfer once the relay is attached and the PLAY/Forge handshake has finished. */ + public void startPending() { + PendingTransfer waiting = pendingTransfer; + if (host.disconnecting() || waiting == null || host.relay() == null || !host.observation().ready().isDone()) return; + pendingTransfer = null; + waiting.deadline.cancel(false); + beginTransfer(waiting.backendName, waiting.result); + } + + private void writeBackendLogin(Channel channel) { + transfer.candidate.loginStarted(); + ByteBuf handshakeBody = host.backendHandshake( + transfer.target.handle().id().value(), transfer.target.owner().instanceGeneration()); + ByteBuf loginBody = new LoginStart(host.view().username()).encode(frontend.alloc(), ProtocolProfile.minecraft1710()); + channel.write(handshakeBody); + channel.writeAndFlush(loginBody).addListener(write -> { + if (write.isSuccess()) { + LOGGER.debug("Replacement backend login frames flushed for player {} to {}", + host.view().username(), channel.remoteAddress()); + } else { + LOGGER.debug("Replacement backend login write failed for player {}", + host.view().username(), write.cause()); + } + if (!write.isSuccess() && transfer != null && transfer.channel == channel) { + failTransfer(transfer, "could not write replacement backend login"); + } + }); + } + + private void candidateReady(TransferAttempt attempt) { + if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return; + if (attempt.preparation != null && !targetStillCurrent(attempt)) { + failTransfer(attempt, "replacement backend registration changed during login"); + return; + } + attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> { + if (transfer != attempt || attempt.result.isDone()) return; + attempt.result.complete(TransferResult.failed("replacement backend cutover timed out")); + host.closeSession(); + }, host.cutoverTimeout().toNanos(), TimeUnit.NANOSECONDS); + if (attempt.sourceReleased) { + if (clientHasPartialFrame()) { + failTransfer(attempt, "client packet was incomplete at replacement cutover"); + return; + } + activateCandidate(attempt); + return; + } + pauseSource(attempt, host.backend(), "could not buffer frames for transfer", + "old backend stopped during transfer", () -> { + if (attempt.finished || host.closed() || host.disconnecting()) { + if (host.disconnecting()) return; + resumeAfterFailedTransfer(attempt); + return; + } + if (clientHasPartialFrame()) { + failTransfer(attempt, "client packet was incomplete at the transfer boundary"); + return; + } + if (!attempt.channel.isActive()) { + failTransfer(attempt, "replacement backend became unavailable"); + return; + } + attempt.oldRelay.detach().whenComplete((removed, detachFailure) -> + frontend.eventLoop().execute(() -> { + if (detachFailure != null) { + boolean oldRelayStillAttached = frontend.pipeline().get("raw-relay") != null + && host.backend().pipeline().get("raw-relay") != null; + failTransfer(attempt, "could not detach old backend relay"); + if (!oldRelayStillAttached) host.closeSession(); + return; + } + if (host.closed() || host.disconnecting() || !attempt.channel.isActive()) { + if (host.disconnecting()) return; + failTransfer(attempt, "replacement backend closed during transfer"); + host.closeSession(); + return; + } + attempt.detached = true; + activateCandidate(attempt); + })); + }); + } + + /** + * Buffers both old-link directions and pauses the old host.relay(). {@code onPaused} runs on the session + * event loop after accepted writes drain; buffer or drain failures fail the attempt and close. + */ + private void pauseSource(TransferAttempt attempt, Channel source, String bufferFailure, String drainFailure, + Runnable onPaused) { + try { + attempt.clientBuffer = installTransferBuffer(frontend, "transfer-client-buffer"); + attempt.oldBackendBuffer = installTransferBuffer(source, "transfer-old-backend-buffer"); + } catch (RuntimeException failure) { + failTransfer(attempt, bufferFailure); + host.closeSession(); + return; + } + attempt.pauseInProgress = true; + CompletionStage pause = attempt.oldRelay.pause(); + frontend.eventLoop().execute(attempt.clientBuffer::readUntilRemoved); + source.eventLoop().execute(attempt.oldBackendBuffer::readUntilRemoved); + pause.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { + attempt.pauseInProgress = false; + if (failure != null) { + failTransfer(attempt, drainFailure); + host.closeSession(); + return; + } + attempt.paused = true; + onPaused.run(); + })); + } + + private TransferFrameBuffer installTransferBuffer(Channel channel, String name) { + var buffer = new TransferFrameBuffer(host::closeSession); + channel.pipeline().addAfter(SessionChannels.frameDecoderName(channel.pipeline()), name, buffer); + return buffer; + } + + private boolean clientHasPartialFrame() { + return SessionChannels.frameDecoder(frontend.pipeline(), "client frame decoder missing during transfer") + .hasPartialFrame(); + } + + private void activateCandidate(TransferAttempt attempt) { + // Keep READY ownership until removal. A decoder can emit more target frames after + // JoinGame in the same read batch; marking HANDED_OFF early would silently drop them. + host.swapHandshakeCodecs(attempt.channel, attempt.candidate::handOff).whenComplete((ignored, failure) -> + frontend.eventLoop().execute(() -> { + if (failure != null || host.closed() || host.disconnecting() || !attempt.channel.isActive()) { + if (host.disconnecting()) return; + failTransfer(attempt, "replacement backend pipeline failed"); + host.closeSession(); + return; + } + installCandidateRelay(attempt); + })); + } + + private void installCandidateRelay(TransferAttempt attempt) { + PlayObservation nextObservation = attempt.candidate.observation(); + RawRelay.Link next; + try { + if (clientEntityId == null) clientEntityId = host.observation().entityId().orElseThrow(); + attempt.frameState = new TransferFrameHandler.State(frontend, attempt.channel, nextObservation, + clientEntityId, attempt.candidate.joinGame(), host::closeSession); + attempt.channel.pipeline().addLast("keep-alive-bridge", + new KeepAliveBridge(host.keepAlives(), false, host::closeSession)); + installTransferFrameHandlers(attempt.frameState, attempt.channel); + host.installTabCompletion(attempt.channel); + next = RawRelay.attach(frontend, attempt.channel, + bytes -> nextObservation.observeFrame(false, bytes), + bytes -> nextObservation.observeFrame(true, bytes)); + RawRelay.Link observedLink = next; + nextObservation.ready().whenComplete((ignored, failure) -> observedLink.stopObserving()); + } catch (RuntimeException failure) { + failTransfer(attempt, "could not attach replacement relay"); + host.closeSession(); + return; + } + next.ready().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { + if (failure != null || host.closed() || host.disconnecting() + || (attempt.preparation != null && !targetStillCurrent(attempt))) { + if (host.disconnecting()) return; + failTransfer(attempt, "replacement relay was not ready"); + host.closeSession(); + return; + } + ByteBuf opening; + try { + host.keepAlives().switchBackend(); + opening = transferOpening(attempt); + } catch (RuntimeException malformed) { + failTransfer(attempt, "could not prepare replacement world transition"); + host.closeSession(); + return; + } + frontend.writeAndFlush(opening).addListener(write -> frontend.eventLoop().execute(() -> { + if (!write.isSuccess() || host.closed() || host.disconnecting() || !attempt.channel.isActive() + || (attempt.preparation != null && !targetStillCurrent(attempt))) { + if (host.disconnecting()) return; + failTransfer(attempt, "could not send replacement world transition"); + host.closeSession(); + return; + } + Channel oldBackend = host.backend(); + Optional previousServer = attempt.sourceReleased + ? Optional.of(attempt.context.source().name()) : host.view().currentServer(); + PlayObservation oldObservation = host.observation(); + TransferFrameHandler.State oldFrameState = frameState; + frameState = attempt.frameState; + host.cutover(attempt.channel, attempt.target, nextObservation, next, previousServer); + if (oldBackend != null) oldBackend.close(); + oldObservation.close(); + if (oldFrameState != null) oldFrameState.close(); + try { + attempt.clientBuffer.drainAndRemove(); + } catch (RuntimeException replayFailure) { + host.closeSession(); + return; + } + if (host.closed()) return; + next.start(); + if (host.closed()) return; + if (nextObservation.forgeSeen()) { + attempt.cutoverDeadline.cancel(false); + attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> { + if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return; + failTransfer(attempt, "replacement Forge handshake timed out"); + host.closeSession(); + }, 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 || host.closed() || host.disconnecting()) return; + if (negotiationFailure != null || !attempt.channel.isActive()) { + failTransfer(attempt, "replacement Forge handshake failed"); + host.closeSession(); + } else finishTransfer(attempt); + })); + } else finishTransfer(attempt); + })); + })); + } + + private void finishTransfer(TransferAttempt attempt) { + if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return; + transfer = null; + attempt.finished = true; + attempt.cutoverDeadline.cancel(false); + cancelCoordination(attempt); + attempt.result.complete(TransferResult.of(TransferStatus.NETWORK_READY)); + } + + private void installTransferFrameHandlers(TransferFrameHandler.State state, Channel target) { + ChannelPipeline clientPipeline = frontend.pipeline(); + if (clientPipeline.get("transfer-frame-handler") != null) clientPipeline.remove("transfer-frame-handler"); + if (SessionChannels.frameDecoderName(clientPipeline) == null) { + throw new IllegalStateException("client frame decoder missing during transfer"); + } + clientPipeline.addLast("transfer-frame-handler", new TransferFrameHandler(state, false)); + target.pipeline().addLast("transfer-frame-handler", new TransferFrameHandler(state, true)); + } + + private ByteBuf transferOpening(TransferAttempt attempt) { + List queued = attempt.candidate.takeQueuedPackets(); + try { + return ByteBufs.fill(frontend.alloc().buffer(), output -> { + boolean nextForge = attempt.candidate.observation().forgeSeen(); + // FML's reset also restores the client's frozen registry. A Forge -> vanilla + // switch needs it even though the replacement server has no ServerHello. + // After that reset the client remains in HELLO until a later Forge switch. + if (!clientFmlAwaitingServerHello && (host.observation().forgeSeen() || nextForge)) { + ByteBuf reset = Minecraft1710PlayPackets.forgeReset(frontend.alloc()); + try { output.writeBytes(reset); } finally { reset.release(); } + clientFmlAwaitingServerHello = true; + } + if (nextForge) clientFmlAwaitingServerHello = false; + if (attempt.candidate.joinGame() != null) { + int targetDimension = attempt.candidate.observation().dimension() + .orElse(attempt.candidate.joinGame().dimension()); + ByteBuf respawns = Minecraft1710PlayPackets.respawnSequence(frontend.alloc(), + attempt.candidate.joinGame(), targetDimension); + try { output.writeBytes(respawns); } finally { respawns.release(); } + } + for (ByteBuf packet : queued) { + ByteBuf mapped = host.keepAlives().body(frontend.alloc(), packet, false); + if (mapped == null) continue; + try { + TransitionFrames.appendClientbound(output, frontend.alloc(), mapped, + attempt.candidate.joinGame() == null ? null : attempt.candidate.joinGame().entityId(), + clientEntityId); + } finally { + if (mapped != packet) mapped.release(); + } + } + }); + } finally { + queued.forEach(ByteBuf::release); + } + } + + private void failTransfer(TransferAttempt attempt, String reason) { + if (attempt.finished) return; + LOGGER.debug("Replacement backend failed for player {} to {}: {}; channel registered={}, active={}", + host.view().username(), attempt.target.address(), reason, + attempt.channel != null && attempt.channel.isRegistered(), + attempt.channel != null && attempt.channel.isActive()); + attempt.finished = true; + attempt.failureReason = reason; + cancelCoordination(attempt); + if (attempt.channel != null) attempt.channel.close(); + if (attempt.candidate != null) attempt.candidate.close(); + if (attempt.frameState != null) attempt.frameState.close(); + if (host.disconnecting()) return; + if (!attempt.pauseInProgress) resumeAfterFailedTransfer(attempt); + } + + private void resumeAfterFailedTransfer(TransferAttempt attempt) { + if (attempt.paused && !attempt.detached && !host.closed() && !host.disconnecting()) { + attempt.oldRelay.resume().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> { + if (failure != null) host.closeSession(); + completeFailedTransfer(attempt); + })); + } else { + completeFailedTransfer(attempt); + } + } + + private void completeFailedTransfer(TransferAttempt attempt) { + if (!attempt.detached && !host.closed() && !host.disconnecting()) { + try { + if (attempt.oldBackendBuffer != null) attempt.oldBackendBuffer.drainAndRemove(); + if (attempt.clientBuffer != null) attempt.clientBuffer.drainAndRemove(); + } catch (RuntimeException failure) { + host.closeSession(); + } + } + if (attempt.cutoverDeadline != null) attempt.cutoverDeadline.cancel(false); + if (transfer == attempt) transfer = null; + attempt.result.complete(TransferResult.failed(attempt.failureReason)); + if (attempt.detached) host.closeSession(); + } + + private static void cancelCoordination(TransferAttempt attempt) { + if (attempt.totalDeadline != null) attempt.totalDeadline.cancel(false); + CompletableFuture work = attempt.coordination; + attempt.coordination = null; + // Cancellation may run arbitrary CompletionStage continuations, never on session I/O. + if (work != null) Thread.startVirtualThread(() -> work.cancel(false)); + } + /** Session teardown: fails any queued or running transfer and releases the frame state. */ + public void close() { + if (frameState != null) frameState.close(); + PendingTransfer waiting = pendingTransfer; + if (waiting != null) { + pendingTransfer = null; + waiting.deadline.cancel(false); + waiting.result.complete(TransferResult.failed("player session closed during transfer")); + } + TransferAttempt activeTransfer = transfer; + if (activeTransfer != null) { + transfer = null; + activeTransfer.finished = true; + cancelCoordination(activeTransfer); + 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")); + } + } +} diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameBuffer.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameBuffer.java similarity index 92% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameBuffer.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameBuffer.java index 557bc9f..7cd7862 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameBuffer.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameBuffer.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; import dev.moonbridge.core.protocol.ProtocolProfile; import io.netty.buffer.ByteBuf; @@ -9,7 +9,7 @@ import io.netty.util.ReferenceCountUtil; import java.util.ArrayDeque; /** Holds decoded frames on either side of the old session during backend cutover. */ -final class TransferFrameBuffer extends ChannelInboundHandlerAdapter { +public final class TransferFrameBuffer extends ChannelInboundHandlerAdapter { // The decoder retains the three-byte length prefix on a maximum-sized protocol 5 frame. private static final int MAX_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes() + 3; private static final int MAX_MESSAGES = 1024; @@ -22,7 +22,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter { private boolean reading; private boolean discard; - TransferFrameBuffer(Runnable closeSession) { + public TransferFrameBuffer(Runnable closeSession) { this.closeSession = closeSession; } @@ -31,7 +31,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter { } /** Keep reading bounded frames so a paused transfer also observes peer disconnects. */ - void readUntilRemoved() { + public void readUntilRemoved() { ChannelHandlerContext ctx = context; if (ctx == null || !ctx.executor().inEventLoop()) { throw new IllegalStateException("transfer buffer must read on its channel event loop"); @@ -65,7 +65,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter { } /** Replays frames after this handler, preserving the already decoded frame boundary. */ - void drainAndRemove() { + public void drainAndRemove() { var ctx = context; if (ctx == null || !ctx.executor().inEventLoop()) { throw new IllegalStateException("transfer buffer must drain on its channel event loop"); @@ -84,7 +84,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter { } /** After irreversible source detach, discard queued and future old-world frames. */ - void discardFrames() { + public void discardFrames() { var ctx = context; if (ctx == null || !ctx.executor().inEventLoop()) { throw new IllegalStateException("transfer buffer must switch mode on its channel event loop"); diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameHandler.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameHandler.java similarity index 93% rename from proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameHandler.java rename to proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameHandler.java index 9373535..7abb599 100644 --- a/proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameHandler.java +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameHandler.java @@ -1,4 +1,7 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; + +import dev.moonbridge.core.session.play.FrameTransformHandler; +import dev.moonbridge.core.session.play.PlayObservation; import dev.moonbridge.core.protocol.Minecraft1710EntityIds; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; @@ -12,8 +15,8 @@ import io.netty.channel.ChannelHandlerContext; import java.util.concurrent.CompletableFuture; /** Handles the few framed packets that cannot pass unchanged after a backend switch. */ -final class TransferFrameHandler extends FrameTransformHandler { - static final class State implements AutoCloseable { +public final class TransferFrameHandler extends FrameTransformHandler { + public static final class State implements AutoCloseable { private final Channel frontend; private final Channel backend; private final PlayObservation observation; @@ -78,7 +81,7 @@ final class TransferFrameHandler extends FrameTransformHandler { private final State state; private final boolean clientbound; - TransferFrameHandler(State state, boolean clientbound) { + public TransferFrameHandler(State state, boolean clientbound) { super(state.closeSession); this.state = state; this.clientbound = clientbound; diff --git a/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferHost.java b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferHost.java new file mode 100644 index 0000000..da55b76 --- /dev/null +++ b/proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferHost.java @@ -0,0 +1,58 @@ +package dev.moonbridge.core.session.transfer; + +import dev.moonbridge.api.PlayerIdentity; +import dev.moonbridge.api.PlayerView; +import dev.moonbridge.core.backend.BackendCatalog; +import dev.moonbridge.core.backend.BackendView; +import dev.moonbridge.core.event.TransferPreparation; +import dev.moonbridge.core.net.NetworkTransport; +import dev.moonbridge.core.relay.RawRelay; +import dev.moonbridge.core.session.play.KeepAliveBridge; +import dev.moonbridge.core.session.play.PlayObservation; +import io.netty.buffer.ByteBuf; +import io.netty.channel.Channel; +import io.netty.resolver.AddressResolverGroup; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +/** + * What a {@link TransferCoordinator} needs from its session and the proxy. The session keeps ownership of + * the connection state; the coordinator only reads it and commits the two transitions below. + * All methods are called on the session's frontend event loop. + */ +public interface TransferHost { + Channel frontend(); + Channel backend(); + RawRelay.Link relay(); + BackendView selected(); + PlayerView view(); + PlayerIdentity identity(); + /** True once the player has been admitted to a backend (post login success). */ + boolean published(); + PlayObservation observation(); + KeepAliveBridge.State keepAlives(); + boolean closed(); + boolean disconnecting(); + void closeSession(); + + BackendCatalog catalog(); + TransferPreparation selectPreparation(); + Duration eventTimeout(); + Duration cutoverTimeout(); + NetworkTransport transport(); + AddressResolverGroup resolver(); + + ByteBuf backendHandshake(String backendName, long backendEpoch); + void installTabCompletion(Channel backend); + /** Strips login-phase handlers from {@code channel} (after {@code beforeRemoval}) and installs lifecycle handling. */ + CompletableFuture swapHandshakeCodecs(Channel channel, Runnable beforeRemoval); + + /** The source backend is gone and no destination exists yet: the player has no current server. */ + void sourceReleased(); + /** The replacement backend now serves the player. */ + void cutover(Channel backend, BackendView selected, PlayObservation observation, RawRelay.Link relay, + Optional previousServer); +} diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java index f1d40d8..11bef86 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java @@ -1,5 +1,7 @@ package dev.moonbridge.core.session; +import dev.moonbridge.core.session.play.PlayObservation; + import com.fasterxml.jackson.databind.ObjectMapper; import dev.moonbridge.api.PlacementDecision; import dev.moonbridge.api.AccessDecision; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/CommandRateLimiterTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/CommandRateLimiterTest.java similarity index 84% rename from proxy-core/src/test/java/dev/moonbridge/core/session/CommandRateLimiterTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/CommandRateLimiterTest.java index 006031e..9772359 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/CommandRateLimiterTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/CommandRateLimiterTest.java @@ -1,10 +1,10 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import org.junit.jupiter.api.Test; -import static dev.moonbridge.core.session.CommandRateLimiter.Decision.ADMIT; -import static dev.moonbridge.core.session.CommandRateLimiter.Decision.DENY; -import static dev.moonbridge.core.session.CommandRateLimiter.Decision.DENY_WITH_NOTICE; +import static dev.moonbridge.core.session.play.CommandRateLimiter.Decision.ADMIT; +import static dev.moonbridge.core.session.play.CommandRateLimiter.Decision.DENY; +import static dev.moonbridge.core.session.play.CommandRateLimiter.Decision.DENY_WITH_NOTICE; import static org.junit.jupiter.api.Assertions.assertEquals; final class CommandRateLimiterTest { diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/FrameTransformHandlerTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/FrameTransformHandlerTest.java similarity index 99% rename from proxy-core/src/test/java/dev/moonbridge/core/session/FrameTransformHandlerTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/FrameTransformHandlerTest.java index b02680c..fc26edd 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/FrameTransformHandlerTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/FrameTransformHandlerTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import io.netty.buffer.ByteBuf; import io.netty.buffer.DefaultByteBufHolder; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/KeepAliveBridgeTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/KeepAliveBridgeTest.java similarity index 99% rename from proxy-core/src/test/java/dev/moonbridge/core/session/KeepAliveBridgeTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/KeepAliveBridgeTest.java index da34610..8dd297f 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/KeepAliveBridgeTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/KeepAliveBridgeTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.relay.RawRelay; import io.netty.buffer.ByteBuf; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/PlayObservationTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/PlayObservationTest.java similarity index 99% rename from proxy-core/src/test/java/dev/moonbridge/core/session/PlayObservationTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/PlayObservationTest.java index ecd1fde..983844e 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/PlayObservationTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/PlayObservationTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.ProtocolVarInt; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/TabCompletionBridgeTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/TabCompletionBridgeTest.java similarity index 99% rename from proxy-core/src/test/java/dev/moonbridge/core/session/TabCompletionBridgeTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/TabCompletionBridgeTest.java index 82a7c38..2c36c30 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/TabCompletionBridgeTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/TabCompletionBridgeTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.MinecraftTabCompletion; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/TransitionBufferTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionBufferTest.java similarity index 98% rename from proxy-core/src/test/java/dev/moonbridge/core/session/TransitionBufferTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionBufferTest.java index f25ec91..3e7bb67 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/TransitionBufferTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionBufferTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/TransitionFramesTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionFramesTest.java similarity index 99% rename from proxy-core/src/test/java/dev/moonbridge/core/session/TransitionFramesTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionFramesTest.java index daebc3b..f3a0253 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/TransitionFramesTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionFramesTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.play; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/TransferCandidateTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferCandidateTest.java similarity index 99% rename from proxy-core/src/test/java/dev/moonbridge/core/session/TransferCandidateTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferCandidateTest.java index ee83d53..fb4d60f 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/TransferCandidateTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferCandidateTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.ProtocolVarInt; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/TransferFrameBufferTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameBufferTest.java similarity index 98% rename from proxy-core/src/test/java/dev/moonbridge/core/session/TransferFrameBufferTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameBufferTest.java index 17ba971..c902dea 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/TransferFrameBufferTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameBufferTest.java @@ -1,4 +1,4 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; import dev.moonbridge.core.protocol.ProtocolProfile; import io.netty.buffer.ByteBuf; diff --git a/proxy-core/src/test/java/dev/moonbridge/core/session/TransferFrameHandlerTest.java b/proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameHandlerTest.java similarity index 96% rename from proxy-core/src/test/java/dev/moonbridge/core/session/TransferFrameHandlerTest.java rename to proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameHandlerTest.java index aca9fd8..442b5c9 100644 --- a/proxy-core/src/test/java/dev/moonbridge/core/session/TransferFrameHandlerTest.java +++ b/proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameHandlerTest.java @@ -1,4 +1,6 @@ -package dev.moonbridge.core.session; +package dev.moonbridge.core.session.transfer; + +import dev.moonbridge.core.session.play.PlayObservation; import dev.moonbridge.core.protocol.Minecraft1710PlayPackets; import dev.moonbridge.core.protocol.ProtocolVarInt;