fix(connect): preserve tunnel delivery and late wrappers

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(() -> {

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<byte[]> 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<byte[]> 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;

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<ChannelInitializer<Channel>> 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<Channel> {
+        private static final Method INIT_CHANNEL =
+                getMethod(ChannelInitializer.class, "initChannel", Channel.class);
+
+        private final Supplier<ChannelInitializer<Channel>> currentInitializer;
+
+        CurrentVelocityChannelInitializer(
+                Supplier<ChannelInitializer<Channel>> currentInitializer) {
+            this.currentInitializer = currentInitializer;
+        }
+
+        @Override
+        protected void initChannel(Channel channel) {
+            ChannelInitializer<Channel> initializer = currentInitializer.get();
+            if (initializer == null) {
+                throw new IllegalStateException("Velocity server channel initializer is unavailable");
+            }
+            invoke(initializer, INIT_CHANNEL, channel);
+            VelocityChatSessionPacketFilter.inject(channel, true);
+        }
+    }
 }

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<ChannelInitializer<Channel>> current = new AtomicReference<>(initializer("velocity-decoder"));
+        VelocityInjector.CurrentVelocityChannelInitializer tunnelInitializer =
+                new VelocityInjector.CurrentVelocityChannelInitializer(current::get);
+
+        ChannelInitializer<Channel> 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<Channel> initializer(String handlerName) {
+        return new ChannelInitializer<>() {
+            @Override
+            protected void initChannel(Channel channel) {
+                channel.pipeline().addLast(handlerName, new ChannelInboundHandlerAdapter());
+            }
+        };
+    }
+
+    private static ChannelInitializer<Channel> wrappingInitializer(
+            ChannelInitializer<Channel> wrapped,
+            String handlerName) {
+        return new ChannelInitializer<>() {
+            @Override
+            protected void initChannel(Channel channel) {
+                channel.pipeline().addLast(wrapped);
+                channel.pipeline().addLast(handlerName, new ChannelInboundHandlerAdapter());
+            }
+        };
+    }
+}