diff --git a/core/src/main/java/com/minekube/connect/network/netty/TunnelHandler.java b/core/src/main/java/com/minekube/connect/network/netty/TunnelHandler.java index 3db64af16..fbd99f431 100644 --- a/core/src/main/java/com/minekube/connect/network/netty/TunnelHandler.java +++ b/core/src/main/java/com/minekube/connect/network/netty/TunnelHandler.java @@ -35,7 +35,6 @@ import io.netty.channel.EventLoop; import java.util.Arrays; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import lombok.RequiredArgsConstructor; @@ -48,13 +47,6 @@ class TunnelHandler implements Handler { private final AtomicLong tunnelToBackendPackets = new AtomicLong(); private final AtomicLong tunnelToBackendBytes = new AtomicLong(); - // Coalesces flushes across an EventLoop tick: one flush() per batch of - // onReceive calls instead of one per packet. The CAS lives inside the - // write task so the flush is always enqueued after the write that needs - // it — scheduling the CAS outside the EventLoop races, because a later - // write can be enqueued behind an already-scheduled flush. - private final AtomicBoolean flushScheduled = new AtomicBoolean(false); - @Override public void onReceive(byte[] data) { tunnelToBackendPackets.incrementAndGet(); @@ -66,19 +58,11 @@ public void onReceive(byte[] data) { Channel ch = downstreamServerConn; EventLoop el = ch.eventLoop(); try { - el.execute(() -> { - ch.write(Unpooled.wrappedBuffer(payload), ch.voidPromise()); - if (flushScheduled.compareAndSet(false, true)) { - try { - el.execute(() -> { - flushScheduled.set(false); - ch.flush(); - }); - } catch (RejectedExecutionException ignored) { - flushScheduled.set(false); - } - } - }); + // Keep acceptance and delivery in one FIFO event-loop task. Scheduling flush as a + // second task allows unrelated channel work to observe the packet before it is + // delivered, which can stall time-sensitive protocol responses such as keepalives. + el.execute(() -> ch.writeAndFlush( + Unpooled.wrappedBuffer(payload), ch.voidPromise())); } catch (RejectedExecutionException ignored) { // Event loop is shutting down; the channel is going away anyway. } @@ -105,10 +89,8 @@ public void onClose() { playerName, sessionId, downstreamServerConn.localAddress(), downstreamServerConn.remoteAddress(), tunnelToBackendPackets.get(), tunnelToBackendBytes.get()); } - // Flush before closing: deferred writes from onReceive() may still be - // sitting in the channel's outbound buffer with the flush scheduled as - // a separate EventLoop task, so closing without a final flush can drop - // the last payload. + // Flush before closing as a final safeguard for any outbound data written by another + // channel handler. Accepted onReceive tasks are FIFO-ordered ahead of this close task. Channel ch = downstreamServerConn; try { ch.eventLoop().execute(() -> { diff --git a/core/src/test/java/com/minekube/connect/network/netty/TunnelHandlerTest.java b/core/src/test/java/com/minekube/connect/network/netty/TunnelHandlerTest.java index b2ee570c0..faa97c521 100644 --- a/core/src/test/java/com/minekube/connect/network/netty/TunnelHandlerTest.java +++ b/core/src/test/java/com/minekube/connect/network/netty/TunnelHandlerTest.java @@ -49,7 +49,22 @@ void onReceiveWritesPayloadAndFlushesOnce() throws Exception { } @Test - void burstOfReceivesCoalescesIntoOneFlush() throws Exception { + void onReceiveFlushesBeforeTheNextEventLoopTaskCanObserveDelivery() throws Exception { + TunnelHandler handler = newHandler(); + byte[] payload = new byte[] {4, 5, 6}; + + runWithEventLoopBlocked(() -> { + handler.onReceive(payload); + eventLoop.execute(() -> events.add(new RecordedEvent(Event.OBSERVE))); + }); + awaitEventLoop(); + + assertEventTypes(Event.WRITE, Event.FLUSH, Event.OBSERVE); + assertArrayEquals(payload, events.get(0).payload); + } + + @Test + void burstOfReceivesFlushesEveryPayloadInOrder() throws Exception { TunnelHandler handler = newHandler(); List payloads = new ArrayList<>(); @@ -63,7 +78,7 @@ void burstOfReceivesCoalescesIntoOneFlush() throws Exception { awaitEventLoop(); assertEquals(50, count(Event.WRITE)); - assertEquals(1, count(Event.FLUSH)); + assertEquals(50, count(Event.FLUSH)); List actualPayloads = writePayloads(); assertEquals(50, actualPayloads.size()); for (int i = 0; i < payloads.size(); i++) { @@ -116,6 +131,19 @@ private TunnelHandler newHandler() { } }).when(channel).write(any(ByteBuf.class), any(ChannelPromise.class)); + doAnswer(invocation -> { + ByteBuf buf = invocation.getArgument(0); + try { + byte[] payload = new byte[buf.readableBytes()]; + buf.getBytes(buf.readerIndex(), payload); + events.add(new RecordedEvent(Event.WRITE, payload)); + events.add(new RecordedEvent(Event.FLUSH)); + return invocation.getArgument(1); + } finally { + buf.release(); + } + }).when(channel).writeAndFlush(any(ByteBuf.class), any(ChannelPromise.class)); + doAnswer(invocation -> { events.add(new RecordedEvent(Event.FLUSH)); return channel; @@ -186,7 +214,7 @@ private int count(Event type) { return count; } - private enum Event { WRITE, FLUSH, CLOSE } + private enum Event { WRITE, FLUSH, CLOSE, OBSERVE } private static final class RecordedEvent { private final Event type; diff --git a/velocity/src/main/java/com/minekube/connect/inject/velocity/VelocityInjector.java b/velocity/src/main/java/com/minekube/connect/inject/velocity/VelocityInjector.java index 707fd761a..47e46b96d 100644 --- a/velocity/src/main/java/com/minekube/connect/inject/velocity/VelocityInjector.java +++ b/velocity/src/main/java/com/minekube/connect/inject/velocity/VelocityInjector.java @@ -50,12 +50,12 @@ import io.netty.channel.local.LocalAddress; import io.netty.channel.local.LocalIoHandler; import io.netty.channel.nio.NioIoHandler; -import io.netty.util.concurrent.ThreadAwareExecutor; import io.netty.util.concurrent.DefaultThreadFactory; +import io.netty.util.concurrent.ThreadAwareExecutor; +import java.lang.reflect.Method; import java.util.concurrent.Executor; -import java.util.function.Supplier; import java.util.concurrent.ThreadFactory; -import java.lang.reflect.Method; +import java.util.function.Supplier; import lombok.Getter; import lombok.RequiredArgsConstructor; @@ -80,8 +80,11 @@ public boolean inject() { ChannelInitializer serverInitializer = castedInvoke(serverInitializerHolder, "get"); Method serverSetter = getMethod(serverInitializerHolder, "set", ChannelInitializer.class); - invoke(serverInitializerHolder, serverSetter, - new VelocityChannelInitializer(this, serverInitializer, false, false)); + VelocityChannelInitializer connectServerInitializer = + new VelocityChannelInitializer(this, serverInitializer, false, false); + invoke(serverInitializerHolder, serverSetter, connectServerInitializer); + Supplier> currentServerInitializer = + () -> castedInvoke(serverInitializerHolder, "get"); // Proxy <-> Server // Object backendInitializerHolder = getValue(connectionManager, "backendChannelInitializer"); @@ -132,7 +135,10 @@ protected IoEventLoop newChild(Executor executor, IoHandlerFactory ioHandlerFact ChannelFuture channelFuture = (new ServerBootstrap() .channel(LocalServerChannelWrapper.class) - .childHandler(new VelocityChannelInitializer(this, serverInitializer, false, true)) + // Resolve Velocity's current frontend initializer for every tunnel connection. + // PacketEvents, ViaVersion, and auth plugins can wrap the holder after Connect + // starts; keeping the startup snapshot here silently omits those later handlers. + .childHandler(new CurrentVelocityChannelInitializer(currentServerInitializer)) .group(localBossGroup, localWorkerGroup) .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, serverWriteMark) // Required or else rare network freezes can occur @@ -187,4 +193,26 @@ protected void initChannel(Channel channel) { injector.addInjectedClient(channel); } } + + static final class CurrentVelocityChannelInitializer extends ChannelInitializer { + private static final Method INIT_CHANNEL = + getMethod(ChannelInitializer.class, "initChannel", Channel.class); + + private final Supplier> currentInitializer; + + CurrentVelocityChannelInitializer( + Supplier> currentInitializer) { + this.currentInitializer = currentInitializer; + } + + @Override + protected void initChannel(Channel channel) { + ChannelInitializer initializer = currentInitializer.get(); + if (initializer == null) { + throw new IllegalStateException("Velocity server channel initializer is unavailable"); + } + invoke(initializer, INIT_CHANNEL, channel); + VelocityChatSessionPacketFilter.inject(channel, true); + } + } } diff --git a/velocity/src/test/java/com/minekube/connect/inject/velocity/VelocityInjectorLateBindingTest.java b/velocity/src/test/java/com/minekube/connect/inject/velocity/VelocityInjectorLateBindingTest.java new file mode 100644 index 000000000..d4082ac3d --- /dev/null +++ b/velocity/src/test/java/com/minekube/connect/inject/velocity/VelocityInjectorLateBindingTest.java @@ -0,0 +1,49 @@ +package com.minekube.connect.inject.velocity; + +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.embedded.EmbeddedChannel; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +class VelocityInjectorLateBindingTest { + @Test + void localTunnelUsesFrontendInitializerInstalledAfterConnectStartup() { + AtomicReference> current = new AtomicReference<>(initializer("velocity-decoder")); + VelocityInjector.CurrentVelocityChannelInitializer tunnelInitializer = + new VelocityInjector.CurrentVelocityChannelInitializer(current::get); + + ChannelInitializer connectInitializer = current.get(); + current.set(wrappingInitializer(connectInitializer, "packetevents-decoder")); + + EmbeddedChannel channel = new EmbeddedChannel(tunnelInitializer); + + assertNotNull(channel.pipeline().get("velocity-decoder")); + assertNotNull(channel.pipeline().get("packetevents-decoder")); + channel.finishAndReleaseAll(); + } + + private static ChannelInitializer initializer(String handlerName) { + return new ChannelInitializer<>() { + @Override + protected void initChannel(Channel channel) { + channel.pipeline().addLast(handlerName, new ChannelInboundHandlerAdapter()); + } + }; + } + + private static ChannelInitializer wrappingInitializer( + ChannelInitializer wrapped, + String handlerName) { + return new ChannelInitializer<>() { + @Override + protected void initChannel(Channel channel) { + channel.pipeline().addLast(wrapped); + channel.pipeline().addLast(handlerName, new ChannelInboundHandlerAdapter()); + } + }; + } +}