Connection handling improvements

proxy/src/main/java/com/velocitypowered/proxy/VelocityServer.java

@@ -1062,25 +1062,27 @@ public HttpClient getSharedHttpClient() {
   }
 
   /**
-   * Checks if the {@code connection} can be registered with the proxy.
+   * Checks if a connection identified by the given profile can be registered with the proxy.
+   * Called before the {@link ConnectedPlayer} instance is constructed so that rejected
+   * connections never create a phantom player object.
    *
-   * @param connection the connection to check
+   * @param profile the resolved game profile of the incoming connection
    * @return {@code true} if we can register the connection, {@code false} if not
    */
-  public boolean canRegisterConnection(ConnectedPlayer connection) {
+  public boolean canRegisterConnection(GameProfile profile) {
     // When kick-existing-players is enabled, skip duplicate checks here.
     // registerConnection() handles kicking the existing player and enforcing IP rules.
     if (configuration.isKickExistingPlayers()) {
       return true;
     }
 
-    String lowerName = connection.getUsername().toLowerCase(Locale.US);
+    String lowerName = profile.getName().toLowerCase(Locale.US);
 
     if (connectionsByName.containsKey(lowerName)) {
       return false;
     }
 
-    return !connectionsByUuid.containsKey(connection.getUniqueId());
+    return !connectionsByUuid.containsKey(profile.getId());
   }
 
   /**

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

@@ -77,6 +77,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import javax.crypto.SecretKey;
 import javax.crypto.spec.SecretKeySpec;
@@ -103,6 +104,13 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
    */
   public static final int MAX_CLIENT_PACKET_SIZE = Integer.getInteger("velocity.max-client-packet-size", 2097152);
 
+  /**
+   * Maximum time to wait for {@link #closeWith(Object)}'s write-and-flush to complete before
+   * forcibly closing the channel. Guards against a stuck outbound buffer leaving the connection
+   * alive until Netty's read timeout.
+   */
+  private static final long HARD_CLOSE_TIMEOUT_SECONDS = 5;
+
   private final Channel channel;
 
   public boolean pendingConfigurationSwitch = false;
@@ -318,25 +326,35 @@ public void flush() {
    */
   public void closeWith(Object msg) {
     if (channel.isActive()) {
+      knownDisconnect = true;
+
       boolean is17 = this.getProtocolVersion().lessThan(ProtocolVersion.MINECRAFT_1_8)
           && this.getProtocolVersion().noLessThan(ProtocolVersion.MINECRAFT_1_7_2);
+
       if (is17 && this.getState() != StateRegistry.STATUS) {
         channel.eventLoop().execute(() -> {
           // 1.7.x versions have a race condition with switching protocol states, so explicitly
           // close the connection after a short while.
           this.setAutoReading(false);
-          channel.eventLoop().schedule(() -> {
-            knownDisconnect = true;
-            channel.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
-          }, 250, TimeUnit.MILLISECONDS);
+          channel.eventLoop().schedule(() -> writeAndCloseChannel(msg), 250, TimeUnit.MILLISECONDS);
         });
       } else {
-        knownDisconnect = true;
-        channel.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
+        writeAndCloseChannel(msg);
       }
     }
   }
 
+  private void writeAndCloseChannel(Object msg) {
+    ChannelFuture writeFuture = channel.writeAndFlush(msg);
+    writeFuture.addListener(ChannelFutureListener.CLOSE);
+    ScheduledFuture<?> hardClose = channel.eventLoop().schedule(() -> {
+      if (channel.isActive()) {
+        channel.close();
+      }
+    }, HARD_CLOSE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+    writeFuture.addListener(f -> hardClose.cancel(false));
+  }
+
   public void close() {
     close(true);
   }

proxy/src/main/java/com/velocitypowered/proxy/connection/client/AuthSessionHandler.java

@@ -140,12 +140,21 @@ public void activated() {
         return CompletableFuture.completedFuture(null);
       }
 
+      GameProfile resolvedProfile = profileEvent.getGameProfile();
+
+      if (!server.canRegisterConnection(resolvedProfile)) {
+        inbound.disconnect(
+            Component.translatable("velocity.error.already-connected-proxy", NamedTextColor.RED));
+        return CompletableFuture.completedFuture(null);
+      }
+
       // Initiate a regular connection and move over to it.
-      ConnectedPlayer player = new ConnectedPlayer(server, profileEvent.getGameProfile(),
+      ConnectedPlayer player = new ConnectedPlayer(server, resolvedProfile,
           mcConnection, inbound.getVirtualHost().orElse(null), inbound.getRawVirtualHost().orElse(null), onlineMode,
           inbound.getHandshakeIntent(), inbound.getIdentifiedKey());
       this.connectedPlayer = player;
-      if (!server.canRegisterConnection(player)) {
+
+      if (!server.registerConnection(player)) {
         player.disconnect0(
             Component.translatable("velocity.error.already-connected-proxy", NamedTextColor.RED),
             true);
@@ -291,11 +300,6 @@ private void completeLoginProtocolPhaseAndInitialize(ConnectedPlayer player) {
       if (reason.isPresent()) {
         player.disconnect0(reason.get(), true);
       } else {
-        if (!server.registerConnection(player)) {
-          player.disconnect0(Component.translatable("velocity.error.already-connected-proxy"), true);
-          return;
-        }
-
         if (!this.server.getClusterPlayerService().onPlayerConnect(player)) {
           return;
         }

proxy/src/main/java/com/velocitypowered/proxy/connection/client/ConnectedPlayer.java

@@ -166,6 +166,12 @@ public class ConnectedPlayer implements MinecraftConnectionAssociation, Player,
 
   public static final int MAX_CLIENTSIDE_PLUGIN_CHANNELS = Integer.getInteger("velocity.max-clientside-plugin-channels", 1024);
 
+  /**
+   * Maximum time to wait for {@link DisconnectEvent} handlers to complete during {@link #teardown()}
+   * before unregistering the player and completing the teardown future.
+   */
+  private static final long DISCONNECT_EVENT_TIMEOUT_SECONDS = 30;
+
   private static final PlainTextComponentSerializer PASS_THRU_TRANSLATE =
       PlainTextComponentSerializer.builder().flattener(TranslatableMapper.FLATTENER).build();
 
@@ -248,9 +254,10 @@ public class ConnectedPlayer implements MinecraftConnectionAssociation, Player,
 
   /**
    * Whether the player has fully connected to the first server it's connecting to.
-   * This flag will be {@code true} after the first call to
-   * {@link #setConnectedServer(VelocityServerConnection)} with a non-null
-   * {@link VelocityServerConnection} as its argument.
+   * Set on the first non-null call to {@link #setConnectedServer(VelocityServerConnection)}
+   * and never cleared, so that {@link DisconnectEvent.LoginStatus} correctly reports
+   * {@link DisconnectEvent.LoginStatus#SUCCESSFUL_LOGIN} even when {@code connectedServer}
+   * has been nulled by a kick handler.
    */
   private boolean firstServerConnected = false;
 
@@ -805,6 +812,10 @@ public void disconnect(Component reason) {
    * @param duringLogin whether the disconnect happened during login
    */
   public void disconnect0(Component reason, boolean duringLogin) {
+    if (connection.isKnownDisconnect()) {
+      return;
+    }
+
     Component translated = this.translateMessage(reason);
 
     if (server.getConfiguration().isLogPlayerDisconnections()) {
@@ -840,8 +851,8 @@ public void resetInFlightConnection() {
   public void handleConnectionException(VelocityRegisteredServer server,
                                         Throwable throwable,
                                         boolean safe) {
-    if (!isActive()) {
-      // If the connection is no longer active, it makes no sense to try and recover it.
+    if (!isActive() || connection.isKnownDisconnect()) {
+      // No point trying to recover an inactive connection or one already being torn down.
       return;
     }
 
@@ -885,8 +896,8 @@ public void handleConnectionException(VelocityRegisteredServer server,
   public void handleConnectionException(VelocityRegisteredServer server,
                                         DisconnectPacket disconnect,
                                         boolean safe) {
-    if (!isActive()) {
-      // If the connection is no longer active, it makes no sense to try and recover it.
+    if (!isActive() || connection.isKnownDisconnect()) {
+      // No point trying to recover an inactive connection or one already being torn down.
       return;
     }
 
@@ -929,8 +940,8 @@ private void handleConnectionException(VelocityRegisteredServer rs,
                                          @Nullable Component kickReason,
                                          Component friendlyReason,
                                          boolean safe) {
-    if (!isActive()) {
-      // If the connection is no longer active, it makes no sense to try and recover it.
+    if (!isActive() || connection.isKnownDisconnect()) {
+      // No point trying to recover an inactive connection or one already being torn down.
       return;
     }
 
@@ -977,8 +988,8 @@ private void handleKickEvent(KickedFromServerEvent originalEvent, Component frie
         connectedServer = null;
       }
 
-      if (!isActive()) {
-        // If the connection is no longer active, it makes no sense to try and recover it.
+      if (!isActive() || connection.isKnownDisconnect()) {
+        // No point trying to recover an inactive connection or one already being torn down.
         return;
       }
 
@@ -1304,28 +1315,30 @@ public void teardown() {
       connectedServer.disconnect();
     }
 
-    Optional<ConnectedPlayer> connectedPlayer = server.getPlayer(this.getUniqueId());
-    server.unregisterConnection(this);
+    DisconnectEvent event = new DisconnectEvent(this, computeDisconnectStatus());
+    server.getEventManager().fire(event)
+        .completeOnTimeout(event, DISCONNECT_EVENT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+        .whenComplete((val, ex) -> {
+          server.unregisterConnection(this);
+          if (ex == null) {
+            this.teardownFuture.complete(null);
+          } else {
+            this.teardownFuture.completeExceptionally(ex);
+          }
+        });
+  }
+
+  private DisconnectEvent.LoginStatus computeDisconnectStatus() {
+    Optional<ConnectedPlayer> registered = server.getPlayer(this.getUniqueId());
+    if (registered.isEmpty()) {
+      return connection.isKnownDisconnect() ? LoginStatus.CANCELLED_BY_PROXY : LoginStatus.CANCELLED_BY_USER;
+    }
 
-    DisconnectEvent.LoginStatus status;
-    if (connectedPlayer.isPresent()) {
-      if (connectedPlayer.get().getCurrentServer().isEmpty()) {
-        status = LoginStatus.PRE_SERVER_JOIN;
-      } else {
-        status = connectedPlayer.get() == this ? LoginStatus.SUCCESSFUL_LOGIN : LoginStatus.CONFLICTING_LOGIN;
-      }
-    } else {
-      status = connection.isKnownDisconnect() ? LoginStatus.CANCELLED_BY_PROXY : LoginStatus.CANCELLED_BY_USER;
+    if (registered.get() != this) {
+      return LoginStatus.CONFLICTING_LOGIN;
     }
 
-    DisconnectEvent event = new DisconnectEvent(this, status);
-    server.getEventManager().fire(event).whenComplete((val, ex) -> {
-      if (ex == null) {
-        this.teardownFuture.complete(null);
-      } else {
-        this.teardownFuture.completeExceptionally(ex);
-      }
-    });
+    return this.firstServerConnected ? LoginStatus.SUCCESSFUL_LOGIN : LoginStatus.PRE_SERVER_JOIN;
   }
 
   public CompletableFuture<Void> getTeardownFuture() {