fix: prevent stale connections

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

@@ -26,6 +26,10 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
@@ -70,6 +74,33 @@ public final class PlayerIdentityLock {
     }
   }
 
+  @NonNull CompletableFuture<LockHandle> acquire(@NonNull UUID uuid,
+                                                 @NonNull String name,
+                                                 long timeout,
+                                                 @NonNull TimeUnit unit,
+                                                 @NonNull ScheduledExecutorService scheduler) {
+    synchronized (monitor) {
+      if (!isLocked(uuid, name)) {
+        lockedUuids.add(uuid);
+        lockedNames.add(name);
+        return CompletableFuture.completedFuture(new LockHandle(uuid, name));
+      }
+
+      CompletableFuture<LockHandle> future = new CompletableFuture<>();
+
+      Waiter waiter = new Waiter(uuid, name, future);
+      waiters.add(waiter);
+      try {
+        waiter.timeout = scheduler.schedule(() -> timeout(waiter), timeout, unit);
+      } catch (RuntimeException exception) {
+        waiters.remove(waiter);
+        future.completeExceptionally(exception);
+      }
+
+      return future;
+    }
+  }
+
   private boolean isLocked(UUID uuid, String name) {
     return lockedUuids.contains(uuid) || lockedNames.contains(name);
   }
@@ -87,20 +118,36 @@ private void release(UUID uuid, String name) {
           lockedUuids.add(w.uuid);
           lockedNames.add(w.name);
           it.remove();
+          w.cancelTimeout();
           if (granted == null) {
             granted = new ArrayList<>(2);
           }
           granted.add(w);
         }
       }
     }
+
     if (granted != null) {
       for (Waiter w : granted) {
-        w.future.complete(new LockHandle(w.uuid, w.name));
+        LockHandle lock = new LockHandle(w.uuid, w.name);
+
+        if (!w.future.complete(lock)) {
+          lock.release();
+        }
       }
     }
   }
 
+  private void timeout(Waiter waiter) {
+    synchronized (monitor) {
+      if (!waiters.remove(waiter)) {
+        return;
+      }
+    }
+
+    waiter.future.completeExceptionally(new LockAcquisitionTimeoutException());
+  }
+
   /**
    * A held lock for a (uuid, name) pair. Release the lock exactly once via {@link #release()};
    * subsequent calls are no-ops.
@@ -131,6 +178,33 @@ public boolean isReleased() {
     }
   }
 
-  private record Waiter(UUID uuid, String name, CompletableFuture<LockHandle> future) {
+  static final class LockAcquisitionTimeoutException extends TimeoutException {
+
+    private static final long serialVersionUID = 1L;
+
+  }
+
+  private static final class Waiter {
+
+    private final UUID uuid;
+    private final String name;
+    private final CompletableFuture<LockHandle> future;
+    private volatile ScheduledFuture<?> timeout;
+
+    private Waiter(UUID uuid, String name, CompletableFuture<LockHandle> future) {
+      this.uuid = uuid;
+      this.name = name;
+      this.future = future;
+    }
+
+    private void cancelTimeout() {
+      ScheduledFuture<?> currentTimeout = timeout;
+
+      if (currentTimeout != null) {
+        currentTimeout.cancel(false);
+      }
+    }
+
   }
+
 }

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

@@ -32,6 +32,7 @@
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -64,7 +65,9 @@ public final class PlayerRegistry {
    * Maximum time to wait for {@link DisconnectEvent} handlers to complete before unregistering
    * the player and continuing.
    */
-  private static final long DISCONNECT_EVENT_TIMEOUT_SECONDS = 30;
+  private static final long DISCONNECT_EVENT_TIMEOUT_SECONDS = 10;
+
+  private static final long IDENTITY_LOCK_ACQUIRE_TIMEOUT_SECONDS = 10;
 
   /**
    * Maximum time the identity lock may be held while a connection completes its login sequence.
@@ -105,8 +108,21 @@ public PlayerRegistry(@NonNull VelocityServer server) {
   public @NonNull CompletableFuture<Boolean> registerConnection(@NonNull ConnectedPlayer player) {
     UUID uuid = player.getUniqueId();
     String name = player.getUsername().toLowerCase(Locale.ROOT);
-    return identityLock.acquire(uuid, name)
-        .thenCompose(lock -> withLockReleasedOnFailure(lock, () -> tryRegisterLocked(player, lock)));
+
+    return identityLock.acquire(uuid, name, IDENTITY_LOCK_ACQUIRE_TIMEOUT_SECONDS,
+            TimeUnit.SECONDS, loginTimeoutScheduler)
+        .thenCompose(lock -> withLockReleasedOnFailure(lock, () -> tryRegisterLocked(player, lock)))
+        .exceptionallyCompose(error -> {
+          if (isLockAcquisitionTimeout(error)) {
+            LOGGER.warn("Player {} waited over {}s for the identity lock; closing the connection.",
+                player, IDENTITY_LOCK_ACQUIRE_TIMEOUT_SECONDS);
+
+            player.disconnect(Component.translatable("multiplayer.disconnect.slow_login"));
+            return completedFuture(false);
+          }
+
+          return failedFuture(error);
+        });
   }
 
   /**
@@ -264,12 +280,18 @@ private CompletableFuture<Void> forciblyEvict(ConnectedPlayer existing) {
 
   private CompletableFuture<Void> doUnregisterLocked(ConnectedPlayer player) {
     if (!player.markDisconnectFired()) {
-      // DisconnectEvent was already fired (e.g. by the kick path). Cleanup has already run
-      // and teardownFuture has already been completed by the kicker; nothing more to do.
       return completedFuture(null);
     }
 
-    return fireDisconnectAndCleanup(player, "player", computeDisconnectStatus(player));
+    LoginStatus status;
+    try {
+      status = computeDisconnectStatus(player);
+    } catch (Throwable t) {
+      runCleanup(player, t, "player");
+      return failedFuture(t);
+    }
+
+    return fireDisconnectAndCleanup(player, "player", status);
   }
 
   /**
@@ -287,13 +309,23 @@ private CompletableFuture<Void> fireDisconnectAndCleanup(ConnectedPlayer player,
       return completedFuture(null);
     }
 
-    DisconnectEvent event = new DisconnectEvent(player, status);
-    return server.getEventManager().fire(event)
+    DisconnectEvent event;
+    CompletableFuture<DisconnectEvent> eventFuture;
+    try {
+      event = new DisconnectEvent(player, status);
+      eventFuture = server.getEventManager().fire(event);
+    } catch (Throwable t) {
+      runCleanup(player, t, label);
+      return failedFuture(t);
+    }
+
+    return eventFuture
         .completeOnTimeout(event, DISCONNECT_EVENT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
         .handle((v, ex) -> {
           if (ex != null) {
             LOGGER.error("Exception firing DisconnectEvent for {} {}", label, player, ex);
           }
+
           runCleanup(player, ex, label);
           return null;
         });
@@ -336,6 +368,15 @@ private void removeFromMaps(ConnectedPlayer player) {
     byUuid.remove(player.getUniqueId(), player);
   }
 
+  private static boolean isLockAcquisitionTimeout(Throwable error) {
+    Throwable cause = error;
+    while (cause instanceof CompletionException && cause.getCause() != null) {
+      cause = cause.getCause();
+    }
+
+    return cause instanceof PlayerIdentityLock.LockAcquisitionTimeoutException;
+  }
+
   public Optional<ConnectedPlayer> getPlayer(@NonNull UUID uuid) {
     return Optional.ofNullable(byUuid.get(uuid));
   }