Fixed auto read race conditions

src/main/java/net/raphimc/viaproxy/proxy/packethandler/CompressionPacketHandler.java

@@ -31,6 +31,7 @@ import net.raphimc.netminecraft.packet.impl.login.S2CLoginSuccessPacket1_7;
 import net.raphimc.vialoader.util.VersionEnum;
 import net.raphimc.viaproxy.cli.options.Options;
 import net.raphimc.viaproxy.proxy.session.ProxyConnection;
+import net.raphimc.viaproxy.proxy.util.ChannelUtil;
 
 import java.util.List;
 
@@ -58,11 +59,11 @@ public class CompressionPacketHandler extends PacketHandler {
         } else if (packet instanceof S2CLoginSuccessPacket1_7) {
             if (this.proxyConnection.getClientVersion().isNewerThanOrEqualTo(VersionEnum.r1_8)) {
                 if (Options.COMPRESSION_THRESHOLD > -1 && this.proxyConnection.getC2P().attr(MCPipeline.COMPRESSION_THRESHOLD_ATTRIBUTE_KEY).get() == -1) {
-                    this.proxyConnection.getChannel().config().setAutoRead(false);
+                    ChannelUtil.disableAutoRead(this.proxyConnection.getChannel());
                     this.proxyConnection.getC2P().writeAndFlush(new S2CLoginCompressionPacket(Options.COMPRESSION_THRESHOLD)).addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE, (ChannelFutureListener) f -> {
                         if (f.isSuccess()) {
                             this.proxyConnection.getC2P().attr(MCPipeline.COMPRESSION_THRESHOLD_ATTRIBUTE_KEY).set(Options.COMPRESSION_THRESHOLD);
-                            this.proxyConnection.getChannel().config().setAutoRead(true);
+                            ChannelUtil.restoreAutoRead(this.proxyConnection.getChannel());
                         }
                     });
                 }

src/main/java/net/raphimc/viaproxy/proxy/packethandler/ConfigurationPacketHandler.java

@@ -26,6 +26,7 @@ import net.raphimc.netminecraft.packet.impl.configuration.C2SConfigFinishConfigu
 import net.raphimc.netminecraft.packet.impl.configuration.S2CConfigFinishConfiguration1_20_2;
 import net.raphimc.netminecraft.packet.impl.login.C2SLoginStartConfiguration1_20_2;
 import net.raphimc.viaproxy.proxy.session.ProxyConnection;
+import net.raphimc.viaproxy.proxy.util.ChannelUtil;
 import net.raphimc.viaproxy.util.logging.Logger;
 
 import java.util.List;
@@ -51,7 +52,7 @@ public class ConfigurationPacketHandler extends PacketHandler {
                     if (f.isSuccess()) {
                         Logger.u_info("session", this.proxyConnection.getC2P().remoteAddress(), this.proxyConnection.getGameProfile(), "Switching to CONFIGURATION state");
                         this.proxyConnection.setP2sConnectionState(ConnectionState.CONFIGURATION);
-                        this.proxyConnection.getChannel().config().setAutoRead(true);
+                        ChannelUtil.restoreAutoRead(this.proxyConnection.getChannel());
                     }
                 });
             }
@@ -60,7 +61,7 @@ public class ConfigurationPacketHandler extends PacketHandler {
             listeners.add(f -> {
                 if (f.isSuccess()) {
                     this.proxyConnection.setP2sConnectionState(ConnectionState.CONFIGURATION);
-                    this.proxyConnection.getChannel().config().setAutoRead(true);
+                    ChannelUtil.restoreAutoRead(this.proxyConnection.getChannel());
                 }
             });
         } else if (packet instanceof C2SConfigFinishConfiguration1_20_2) {
@@ -69,7 +70,7 @@ public class ConfigurationPacketHandler extends PacketHandler {
                 if (f.isSuccess()) {
                     Logger.u_info("session", this.proxyConnection.getC2P().remoteAddress(), this.proxyConnection.getGameProfile(), "Configuration finished! Switching to PLAY state");
                     this.proxyConnection.setP2sConnectionState(ConnectionState.PLAY);
-                    this.proxyConnection.getChannel().config().setAutoRead(true);
+                    ChannelUtil.restoreAutoRead(this.proxyConnection.getChannel());
                 }
             });
         }
@@ -81,10 +82,10 @@ public class ConfigurationPacketHandler extends PacketHandler {
     public boolean handleP2S(IPacket packet, List<ChannelFutureListener> listeners) {
         if (packet instanceof UnknownPacket unknownPacket && this.proxyConnection.getP2sConnectionState() == ConnectionState.PLAY) {
             if (unknownPacket.packetId == this.startConfigurationId) {
-                this.proxyConnection.getChannel().config().setAutoRead(false);
+                ChannelUtil.disableAutoRead(this.proxyConnection.getChannel());
             }
         } else if (packet instanceof S2CConfigFinishConfiguration1_20_2) {
-            this.proxyConnection.getChannel().config().setAutoRead(false);
+            ChannelUtil.disableAutoRead(this.proxyConnection.getChannel());
         }
 
         return true;

src/main/java/net/raphimc/viaproxy/proxy/packethandler/LoginPacketHandler.java

@@ -36,6 +36,7 @@ import net.raphimc.viaproxy.proxy.LoginState;
 import net.raphimc.viaproxy.proxy.external_interface.AuthLibServices;
 import net.raphimc.viaproxy.proxy.external_interface.ExternalInterface;
 import net.raphimc.viaproxy.proxy.session.ProxyConnection;
+import net.raphimc.viaproxy.proxy.util.ChannelUtil;
 import net.raphimc.viaproxy.proxy.util.CloseAndReturn;
 import net.raphimc.viaproxy.util.logging.Logger;
 
@@ -183,12 +184,12 @@ public class LoginPacketHandler extends PacketHandler {
             this.proxyConnection.setGameProfile(new GameProfile(loginSuccessPacket.uuid, loginSuccessPacket.name));
             Logger.u_info("session", this.proxyConnection.getC2P().remoteAddress(), this.proxyConnection.getGameProfile(), "Connected successfully! Switching to " + nextState + " state");
 
-            this.proxyConnection.getChannel().config().setAutoRead(false);
+            ChannelUtil.disableAutoRead(this.proxyConnection.getChannel());
             listeners.add(f -> {
                 if (f.isSuccess() && nextState != ConnectionState.CONFIGURATION) {
                     this.proxyConnection.setC2pConnectionState(nextState);
                     this.proxyConnection.setP2sConnectionState(nextState);
-                    this.proxyConnection.getChannel().config().setAutoRead(true);
+                    ChannelUtil.restoreAutoRead(this.proxyConnection.getChannel());
                 }
             });
         }

src/main/java/net/raphimc/viaproxy/proxy/proxy2server/Proxy2ServerChannelInitializer.java

@@ -23,12 +23,12 @@ import com.viaversion.viaversion.protocol.ProtocolPipelineImpl;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.handler.codec.haproxy.HAProxyMessageEncoder;
-import io.netty.handler.flow.FlowControlHandler;
 import io.netty.handler.proxy.HttpProxyHandler;
 import io.netty.handler.proxy.ProxyHandler;
 import io.netty.handler.proxy.Socks4ProxyHandler;
 import io.netty.handler.proxy.Socks5ProxyHandler;
 import net.raphimc.netminecraft.constants.MCPipeline;
+import net.raphimc.netminecraft.netty.codec.NoReadFlowControlHandler;
 import net.raphimc.netminecraft.netty.connection.MinecraftChannelInitializer;
 import net.raphimc.netminecraft.packet.registry.PacketRegistryUtil;
 import net.raphimc.vialoader.netty.VLPipeline;
@@ -76,8 +76,9 @@ public class Proxy2ServerChannelInitializer extends MinecraftChannelInitializer
 
         super.initChannel(channel);
         channel.attr(MCPipeline.PACKET_REGISTRY_ATTRIBUTE_KEY).set(PacketRegistryUtil.getHandshakeRegistry(true));
+
         channel.pipeline().addLast(new ViaProxyVLPipeline(user, proxyConnection.getServerVersion()));
-        channel.pipeline().addAfter(VLPipeline.VIA_CODEC_NAME, "via-" + MCPipeline.FLOW_CONTROL_HANDLER_NAME, new FlowControlHandler());
+        channel.pipeline().addAfter(VLPipeline.VIA_CODEC_NAME, "via-" + MCPipeline.FLOW_CONTROL_HANDLER_NAME, new NoReadFlowControlHandler());
 
         if (PluginManager.EVENT_MANAGER.call(new Proxy2ServerChannelInitializeEvent(ITyped.Type.POST, channel, false)).isCancelled()) {
             channel.close();

src/main/java/net/raphimc/viaproxy/proxy/util/ChannelUtil.java

@@ -0,0 +1,48 @@
+/*
+ * This file is part of ViaProxy - https://github.com/RaphiMC/ViaProxy
+ * Copyright (C) 2023 RK_01/RaphiMC and contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package net.raphimc.viaproxy.proxy.util;
+
+import io.netty.channel.Channel;
+import io.netty.util.AttributeKey;
+
+import java.util.Stack;
+
+public class ChannelUtil {
+
+    private static final AttributeKey<Stack<Boolean>> LAST_AUTO_READ = AttributeKey.valueOf("last-auto-read");
+
+    public static void disableAutoRead(final Channel channel) {
+        if (channel.attr(LAST_AUTO_READ).get() == null) {
+            channel.attr(LAST_AUTO_READ).set(new Stack<>());
+        }
+
+        channel.attr(LAST_AUTO_READ).get().push(channel.config().isAutoRead());
+        channel.config().setAutoRead(false);
+    }
+
+    public static void restoreAutoRead(final Channel channel) {
+        if (channel.attr(LAST_AUTO_READ).get() == null) {
+            throw new IllegalStateException("Tried to restore auto read, but it was never disabled");
+        }
+        if (channel.config().isAutoRead()) {
+            throw new IllegalStateException("Race condition detected: Auto read has been enabled somewhere else");
+        }
+        channel.config().setAutoRead(channel.attr(LAST_AUTO_READ).get().pop());
+    }
+
+}