Further improvements to pipeline in the worst-case scenario.

proxy/src/main/java/com/velocitypowered/proxy/connection/MinecraftConnection.java

@@ -16,7 +16,6 @@ import com.velocitypowered.natives.encryption.VelocityCipher;
 import com.velocitypowered.natives.encryption.VelocityCipherFactory;
 import com.velocitypowered.natives.util.Natives;
 import com.velocitypowered.proxy.VelocityServer;
-import com.velocitypowered.proxy.connection.client.InitialInboundConnection;
 import com.velocitypowered.proxy.connection.client.StatusSessionHandler;
 import com.velocitypowered.proxy.protocol.MinecraftPacket;
 import com.velocitypowered.proxy.protocol.StateRegistry;
@@ -26,6 +25,7 @@ import com.velocitypowered.proxy.protocol.netty.MinecraftCompressDecoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftCompressEncoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftDecoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftEncoder;
+import com.velocitypowered.proxy.util.except.QuietException;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
@@ -39,6 +39,7 @@ import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.security.GeneralSecurityException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import javax.crypto.SecretKey;
 import javax.crypto.spec.SecretKeySpec;
 import org.apache.logging.log4j.LogManager;
@@ -52,6 +53,8 @@ import org.checkerframework.checker.nullness.qual.Nullable;
 public class MinecraftConnection extends ChannelInboundHandlerAdapter {
 
   private static final Logger logger = LogManager.getLogger(MinecraftConnection.class);
+  private static final AtomicLong lastQuietError = new AtomicLong();
+  private static final AtomicLong quietErrorsSent = new AtomicLong();
 
   private final Channel channel;
   private SocketAddress remoteAddress;
@@ -111,6 +114,10 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
         return;
       }
 
+      if (this.isClosed()) {
+        return;
+      }
+
       if (msg instanceof MinecraftPacket) {
         MinecraftPacket pkt = (MinecraftPacket) msg;
         if (!pkt.handle(sessionHandler)) {
@@ -151,6 +158,11 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
         if (cause instanceof ReadTimeoutException) {
           logger.error("{}: read timed out", association);
         } else {
+          if (cause instanceof QuietException && willThrottleQuietErrorLogging()) {
+            // Silence the disconnect
+            this.knownDisconnect = true;
+            return;
+          }
           logger.error("{}: exception encountered in {}", association, sessionHandler, cause);
         }
       }
@@ -428,4 +440,17 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
   public void setType(ConnectionType connectionType) {
     this.connectionType = connectionType;
   }
+
+  private static boolean willThrottleQuietErrorLogging() {
+    long lastErrorAt = lastQuietError.get();
+    long now = System.currentTimeMillis();
+
+    if (lastErrorAt + 2000 >= now) {
+      return quietErrorsSent.incrementAndGet() >= 5;
+    } else {
+      lastQuietError.set(now);
+      quietErrorsSent.set(0);
+      return false;
+    }
+  }
 }

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftDecoder.java

@@ -8,11 +8,10 @@ import com.velocitypowered.proxy.protocol.StateRegistry;
 import com.velocitypowered.proxy.util.except.QuietException;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.CorruptedFrameException;
-import io.netty.handler.codec.MessageToMessageDecoder;
-import java.util.List;
 
-public class MinecraftDecoder extends MessageToMessageDecoder<ByteBuf> {
+public class MinecraftDecoder extends ChannelInboundHandlerAdapter {
 
   public static final boolean DEBUG = Boolean.getBoolean("velocity.packet-decode-logging");
   private static final QuietException DECODE_FAILED =
@@ -36,33 +35,41 @@ public class MinecraftDecoder extends MessageToMessageDecoder<ByteBuf> {
   }
 
   @Override
-  protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
+  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+    if (msg instanceof ByteBuf) {
+      ByteBuf buf = (ByteBuf) msg;
+      try {
+        tryDecode(ctx, buf);
+      } finally {
+        buf.release();
+      }
+    } else {
+      ctx.fireChannelRead(msg);
+    }
+  }
+
+  private void tryDecode(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {
     if (!ctx.channel().isActive()) {
       return;
     }
 
-    if (!msg.isReadable()) {
+    int originalReaderIndex = buf.readerIndex();
-      return;
+    int packetId = ProtocolUtils.readVarInt(buf);
-    }
-
-    ByteBuf slice = msg.slice();
-
-    int packetId = ProtocolUtils.readVarInt(msg);
     MinecraftPacket packet = this.registry.createPacket(packetId);
     if (packet == null) {
-      msg.skipBytes(msg.readableBytes());
+      buf.readerIndex(originalReaderIndex);
-      out.add(slice.retain());
+      ctx.fireChannelRead(buf.retain());
     } else {
       try {
-        packet.decode(msg, direction, registry.version);
+        packet.decode(buf, direction, registry.version);
       } catch (Exception e) {
         throw handleDecodeFailure(e, packet, packetId);
       }
 
-      if (msg.isReadable()) {
+      if (buf.isReadable()) {
         throw handleNotReadEnough(packet, packetId);
       }
-      out.add(packet);
+      ctx.fireChannelRead(packet);
     }
   }
 

proxy/src/main/java/com/velocitypowered/proxy/protocol/packet/ServerLogin.java

@@ -5,11 +5,14 @@ import com.velocitypowered.api.network.ProtocolVersion;
 import com.velocitypowered.proxy.connection.MinecraftSessionHandler;
 import com.velocitypowered.proxy.protocol.MinecraftPacket;
 import com.velocitypowered.proxy.protocol.ProtocolUtils;
+import com.velocitypowered.proxy.util.except.QuietException;
 import io.netty.buffer.ByteBuf;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 public class ServerLogin implements MinecraftPacket {
 
+  private static final QuietException EMPTY_USERNAME = new QuietException("Empty username!");
+
   private @Nullable String username;
 
   public ServerLogin() {
@@ -36,6 +39,9 @@ public class ServerLogin implements MinecraftPacket {
   @Override
   public void decode(ByteBuf buf, ProtocolUtils.Direction direction, ProtocolVersion version) {
     username = ProtocolUtils.readString(buf, 16);
+    if (username.isEmpty()) {
+      throw EMPTY_USERNAME;
+    }
   }
 
   @Override