mirror of
https://github.com/layue13/MoonBridge.git
synced 2026-10-25 18:46:18 +00:00
Preserve login disconnect frames across backend close
This commit is contained in:
layue13 committed
1 parent
5cb322c2b1
commit
0b15bd85e0
3 files changed
+134
-6
No files matched your search
@@ -13,7 +13,7 @@ StrataProxy 是面向 Minecraft 1.7.10 Forge 的玩家会话代理。项目正
|
||||
|
||||
## 当前可运行范围
|
||||
|
||||
配置必须显式选择 `authentication: OFFLINE` 或 `ONLINE_BUNGEE`。两种模式均解析协议 5 Handshake 与 Login Start，等待异步落点，预留容量，连接后端，确认 Login Success，然后转发普通字节流。在线模式执行加密握手、异步会话校验，并向可信的 Bungee 兼容后端转发已验证身份。状态查询由代理直接回答：在线数是本代理已登录的会话数，最大值按当前已注册后端的声明容量计算，且不低于在线数。后端关闭或读失败时，已提交给客户端的写入会先排空再关闭前端，最多等待 5 秒。`Players.transfer` 会在旧后端继续服务时登录候选后端；普通候选后端收到 Join Game 后即可开始交接，Forge 候选后端收到 ServerHello 后先切换握手链路，待客户端完成握手并收到 Join Game 时再合成 Respawn。`NETWORK_READY` 等待这次协议握手和世界切换包写出，但不表示目标服的插件已完成空岛或副本加载。切换后的玩家实体 ID 会映射到原客户端 ID；候选登录失败时保留旧链路。Forge 重置包写出后若目标拒绝握手，代理结束会话并返回失败，不能恢复旧链路。DNS 与 Agent 是调用通用注册 API 的独立插件。
|
||||
配置必须显式选择 `authentication: OFFLINE` 或 `ONLINE_BUNGEE`。两种模式均解析协议 5 Handshake 与 Login Start，等待异步落点，预留容量，连接后端，确认 Login Success，然后转发普通字节流。在线模式执行加密握手、异步会话校验，并向可信的 Bungee 兼容后端转发已验证身份。状态查询由代理直接回答：在线数是本代理已登录的会话数，最大值按当前已注册后端的声明容量计算，且不低于在线数。登录阶段的后端拒绝消息会在前端写出后再清理会话；后端关闭或读失败时，已提交给客户端的写入也会先排空再关闭前端，这两种等待最多 5 秒。`Players.transfer` 会在旧后端继续服务时登录候选后端；普通候选后端收到 Join Game 后即可开始交接，Forge 候选后端收到 ServerHello 后先切换握手链路，待客户端完成握手并收到 Join Game 时再合成 Respawn。`NETWORK_READY` 等待这次协议握手和世界切换包写出，但不表示目标服的插件已完成空岛或副本加载。切换后的玩家实体 ID 会映射到原客户端 ID；候选登录失败时保留旧链路。Forge 重置包写出后若目标拒绝握手，代理结束会话并返回失败，不能恢复旧链路。DNS 与 Agent 是调用通用注册 API 的独立插件。
|
||||
|
||||
**验证边界**：跨后端切换已通过普通协议与 Forge 握手的合成 TCP 测试，包括切换前失败回退、候选服 Join Game 后立即断开时保留旧会话、切换后 Forge 握手拒绝、维度变化后再转服、Forge 到普通后端的切换、握手后 Join Game、维度覆盖、玩家实体 ID 映射和中途断线清理。从 Forge 切换到普通后端会发送一次 FML 重置以恢复客户端注册表；随后进入 Forge 时不会重复发送重置。转服固定发送中间维度和目标维度两次 Respawn，避免依赖已经过期的客户端维度记录；交接期间客户端与旧后端的帧分别进入有上限的缓冲区。成功时，客户端帧在世界切换包写出后转发，旧后端帧丢弃；切换前失败恢复旧链路后，两侧暂存的帧都会回放。这些时序已通过合成 TCP 测试。在线认证仅通过注入会话校验器的合成 TCP 测试。真实 Mojang 服务、Uranium/Bungee 兼容后端和 1.7.10 Forge 整合包尚未联机验证，也未做同条件性能验收；当前不能宣称生产可用或实服 Forge 转服兼容。
|
||||
|
||||
|
||||
@@ -59,6 +59,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
||||
final class Session extends ChannelInboundHandlerAdapter {
|
||||
private static final int MAX_TRANSITION_BUFFER_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes();
|
||||
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;
|
||||
@@ -74,7 +75,7 @@ final class Session extends ChannelInboundHandlerAdapter {
|
||||
private BackendView selected;
|
||||
private CapacityReservation reservation;
|
||||
private boolean placementInProgress;
|
||||
private boolean loginDisconnectSent;
|
||||
private boolean loginDisconnectStarted;
|
||||
private volatile boolean published;
|
||||
private boolean relayStarting;
|
||||
private int transitionBufferBytes;
|
||||
@@ -360,9 +361,10 @@ final class Session extends ChannelInboundHandlerAdapter {
|
||||
}
|
||||
|
||||
private void disconnectLogin(String reason) {
|
||||
if (closed.get() || loginDisconnectSent) return;
|
||||
loginDisconnectSent = true;
|
||||
if (closed.get() || loginDisconnectStarted) return;
|
||||
loginDisconnectStarted = true;
|
||||
frontend.config().setAutoRead(false);
|
||||
resetLoginDeadline(LOGIN_DISCONNECT_DRAIN_TIMEOUT);
|
||||
if (!frontend.isActive()) { closePair(); return; }
|
||||
try {
|
||||
frontend.writeAndFlush(MinecraftLoginDisconnect.encode(frontend.alloc(), reason))
|
||||
@@ -379,6 +381,7 @@ final class Session extends ChannelInboundHandlerAdapter {
|
||||
}
|
||||
|
||||
private void receiveBackend(ByteBuf packet) {
|
||||
if (loginDisconnectStarted) return;
|
||||
ByteBuf input = packet.duplicate();
|
||||
int id = ProtocolVarInt.read(input);
|
||||
if (id == 1) { // Encryption Request means backend is not in offline mode.
|
||||
@@ -386,6 +389,9 @@ final class Session extends ChannelInboundHandlerAdapter {
|
||||
return;
|
||||
}
|
||||
if (id == 0) { // Login Disconnect.
|
||||
loginDisconnectStarted = true;
|
||||
frontend.config().setAutoRead(false);
|
||||
resetLoginDeadline(LOGIN_DISCONNECT_DRAIN_TIMEOUT);
|
||||
frontend.writeAndFlush(packet.copy()).addListener(ignored -> closePair());
|
||||
return;
|
||||
}
|
||||
@@ -1041,11 +1047,12 @@ final class Session extends ChannelInboundHandlerAdapter {
|
||||
}
|
||||
|
||||
@Override public void channelInactive(ChannelHandlerContext ctx) {
|
||||
closePair();
|
||||
if (ctx.channel() != backend || !loginDisconnectStarted) closePair();
|
||||
}
|
||||
|
||||
@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
|
||||
closePair();
|
||||
if (ctx.channel() == backend && loginDisconnectStarted) ctx.close();
|
||||
else closePair();
|
||||
}
|
||||
|
||||
Optional<PlayerView> onlineView() {
|
||||
|
||||
@@ -989,6 +989,127 @@ final class ProxySessionListenerTest {
|
||||
verifyBackendReadFailureDrain(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
void backendLoginDisconnectSurvivesImmediateBackendClose() throws Exception {
|
||||
verifyBackendLoginDisconnectDrain(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
void failedBackendConnectSendsLoginDisconnectReason() throws Exception {
|
||||
int unavailablePort;
|
||||
try (ServerSocket unused = new ServerSocket(0, 8, InetAddress.getLoopbackAddress())) {
|
||||
unavailablePort = unused.getLocalPort();
|
||||
}
|
||||
var catalog = new InMemoryBackendCatalog();
|
||||
catalog.register(new BackendRegistration(new BackendId("lobby"), new BackendOwner("static", 0),
|
||||
URI.create("tcp://127.0.0.1:" + unavailablePort), 1));
|
||||
var listener = new ProxySessionListener(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), catalog);
|
||||
listener.setPlacement(player -> CompletableFuture.completedFuture(Optional.of(PlacementDecision.select("lobby"))));
|
||||
try {
|
||||
int port = ((InetSocketAddress) listener.start().toCompletableFuture()
|
||||
.get(5, TimeUnit.SECONDS).localAddress()).getPort();
|
||||
try (Socket client = new Socket(InetAddress.getLoopbackAddress(), port)) {
|
||||
client.setSoTimeout(5000);
|
||||
sendLogin(client, "ConnectFailure");
|
||||
var response = new java.io.ByteArrayInputStream(readFrame(new DataInputStream(client.getInputStream())));
|
||||
assertEquals(0, readVarInt(response));
|
||||
assertTrue(readString(response, 32767).contains("Could not connect"));
|
||||
}
|
||||
} finally {
|
||||
listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void backendLoginDisconnectClosesStalledClientWriteAfterDeadline() throws Exception {
|
||||
verifyBackendLoginDisconnectDrain(false);
|
||||
}
|
||||
|
||||
private void verifyBackendLoginDisconnectDrain(boolean completeClientWrite) throws Exception {
|
||||
var catalog = new InMemoryBackendCatalog();
|
||||
try (ServerSocket backendServer = new ServerSocket(0, 8, InetAddress.getLoopbackAddress())) {
|
||||
var backendReady = new CompletableFuture<Void>();
|
||||
var sendDisconnect = new CompletableFuture<Void>();
|
||||
var backendDone = new CompletableFuture<Void>();
|
||||
var disconnectBody = new ByteArrayOutputStream();
|
||||
writeVarInt(disconnectBody, 0);
|
||||
writeString(disconnectBody, "{\"text\":\"Backend refused login\"}");
|
||||
byte[] disconnect = disconnectBody.toByteArray();
|
||||
var backendThread = new Thread(() -> {
|
||||
try (Socket socket = backendServer.accept()) {
|
||||
socket.setSoTimeout(5000);
|
||||
var input = new DataInputStream(socket.getInputStream());
|
||||
var output = new DataOutputStream(socket.getOutputStream());
|
||||
readFrame(input);
|
||||
readFrame(input);
|
||||
backendReady.complete(null);
|
||||
sendDisconnect.get(5, TimeUnit.SECONDS);
|
||||
writeFrame(output, disconnect);
|
||||
} catch (Throwable failure) { backendDone.completeExceptionally(failure); return; }
|
||||
backendDone.complete(null);
|
||||
}, "fake-login-disconnect-backend");
|
||||
backendThread.setDaemon(true);
|
||||
backendThread.start();
|
||||
|
||||
catalog.register(new BackendRegistration(new BackendId("lobby"), new BackendOwner("static", 0),
|
||||
URI.create("tcp://127.0.0.1:" + backendServer.getLocalPort()), 1));
|
||||
var listener = new ProxySessionListener(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), catalog);
|
||||
listener.setPlacement(player -> CompletableFuture.completedFuture(Optional.of(PlacementDecision.select("lobby"))));
|
||||
var heldContext = new AtomicReference<ChannelHandlerContext>();
|
||||
var heldMessage = new AtomicReference<Object>();
|
||||
var heldPromise = new AtomicReference<ChannelPromise>();
|
||||
var heldWrite = new CompletableFuture<Void>();
|
||||
try {
|
||||
int port = ((InetSocketAddress) listener.start().toCompletableFuture()
|
||||
.get(5, TimeUnit.SECONDS).localAddress()).getPort();
|
||||
try (Socket client = new Socket(InetAddress.getLoopbackAddress(), port)) {
|
||||
client.setSoTimeout(5000);
|
||||
sendLogin(client, "RejectedPlayer");
|
||||
backendReady.get(5, TimeUnit.SECONDS);
|
||||
Session session = listener.allSessions().iterator().next();
|
||||
var frontendField = Session.class.getDeclaredField("frontend");
|
||||
frontendField.setAccessible(true);
|
||||
Channel frontend = (Channel) frontendField.get(session);
|
||||
var backendField = Session.class.getDeclaredField("backend");
|
||||
backendField.setAccessible(true);
|
||||
Channel backendChannel = (Channel) backendField.get(session);
|
||||
frontend.eventLoop().submit(() -> frontend.pipeline().addFirst("hold-login-disconnect",
|
||||
new ChannelOutboundHandlerAdapter() {
|
||||
@Override public void write(ChannelHandlerContext ctx, Object message,
|
||||
ChannelPromise promise) {
|
||||
heldContext.set(ctx);
|
||||
heldMessage.set(message);
|
||||
heldPromise.set(promise);
|
||||
heldWrite.complete(null);
|
||||
}
|
||||
})).get(5, TimeUnit.SECONDS);
|
||||
sendDisconnect.complete(null);
|
||||
heldWrite.get(5, TimeUnit.SECONDS);
|
||||
backendDone.get(5, TimeUnit.SECONDS);
|
||||
assertTrue(backendChannel.closeFuture().await(5, TimeUnit.SECONDS));
|
||||
backendChannel.eventLoop().submit(() -> { }).get(5, TimeUnit.SECONDS);
|
||||
assertTrue(frontend.isOpen(), "backend EOF dropped an accepted login disconnect");
|
||||
var input = new DataInputStream(client.getInputStream());
|
||||
if (completeClientWrite) {
|
||||
frontend.eventLoop().submit(() -> heldContext.get().writeAndFlush(
|
||||
heldMessage.getAndSet(null), heldPromise.get())).get(5, TimeUnit.SECONDS);
|
||||
assertArrayEquals(disconnect, readFrame(input));
|
||||
} else {
|
||||
assertTrue(frontend.closeFuture().await(7, TimeUnit.SECONDS),
|
||||
"the stalled login disconnect write must not keep the session open forever");
|
||||
}
|
||||
assertEquals(-1, input.read());
|
||||
assertEquals(0, catalog.find(new BackendId("lobby")).orElseThrow().reservedCapacity());
|
||||
}
|
||||
} finally {
|
||||
sendDisconnect.complete(null);
|
||||
ReferenceCountUtil.release(heldMessage.getAndSet(null));
|
||||
listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS);
|
||||
backendDone.get(5, TimeUnit.SECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void backendReadFailureClosesStalledClientWriteAfterDeadline() throws Exception {
|
||||
verifyBackendReadFailureDrain(false);
|
||||
|
||||
Reference in new issue
Block a user