Merge branch 'dev/3.0.0' into usb/add-incremental-compiling

api/src/main/java/com/velocitypowered/api/command/CommandManager.java

@@ -45,7 +45,9 @@ public interface CommandManager {
    * @throws IllegalArgumentException if one of the given aliases is already registered, or
    *         the given command does not implement a registrable {@link Command} subinterface
    * @see Command for a list of registrable Command subinterfaces
+   * @deprecated use {@link #register(CommandMeta, Command)} instead with a plugin specified
    */
+  @Deprecated
   default void register(String alias, Command command, String... otherAliases) {
     register(metaBuilder(alias).aliases(otherAliases).build(), command);
   }
@@ -55,7 +57,9 @@ public interface CommandManager {
    *
    * @param command the command to register
    * @throws IllegalArgumentException if the node alias is already registered
+   * @deprecated use {@link #register(CommandMeta, Command)} instead with a plugin specified
    */
+  @Deprecated
   void register(BrigadierCommand command);
 
   /**

api/src/main/java/com/velocitypowered/api/event/EventManager.java

@@ -45,10 +45,28 @@ public interface EventManager {
    * @param postOrder the order in which events should be posted to the handler
    * @param handler the handler to register
    * @param <E> the event type to handle
+   * @deprecated use {@link #register(Object, Class, short, EventHandler)} instead
    */
+  @Deprecated
   <E> void register(Object plugin, Class<E> eventClass, PostOrder postOrder,
       EventHandler<E> handler);
 
+  /**
+   * Requests that the specified {@code handler} listen for events and associate it with the {@code
+   * plugin}.
+   *
+   * <p>Note that this method will register a non-asynchronous listener by default. If you want to
+   * use an asynchronous event handler, return {@link EventTask#async(Runnable)} from the handler.</p>
+   *
+   * @param plugin the plugin to associate with the handler
+   * @param eventClass the class for the event handler to register
+   * @param postOrder the relative order in which events should be posted to the handler
+   * @param handler the handler to register
+   * @param <E> the event type to handle
+   */
+  <E> void register(Object plugin, Class<E> eventClass, short postOrder,
+      EventHandler<E> handler);
+
   /**
    * Fires the specified event to the event bus asynchronously. This allows Velocity to continue
    * servicing connections while a plugin handles a potentially long-running operation such as a

api/src/main/java/com/velocitypowered/api/event/PostOrder.java

@@ -12,6 +12,6 @@ package com.velocitypowered.api.event;
  */
 public enum PostOrder {
 
-  FIRST, EARLY, NORMAL, LATE, LAST
+  FIRST, EARLY, NORMAL, LATE, LAST, CUSTOM
 
 }

api/src/main/java/com/velocitypowered/api/event/Subscribe.java

@@ -22,24 +22,38 @@ public @interface Subscribe {
   /**
    * The order events will be posted to this listener.
    *
+   * @deprecated specify the order using {@link #priority()} instead
    * @return the order
    */
+  @Deprecated
   PostOrder order() default PostOrder.NORMAL;
 
   /**
-   * Whether the handler must be called asynchronously.
+   * The priority of this event handler. Priorities are used to determine the order in which event
+   * handlers are called. The higher the priority, the earlier the event handler will be called.
    *
-   * <p><strong>This option currently has no effect, but in the future it will. In Velocity 3.0.0,
+   * <p>Note that due to compatibility constraints, you must specify {@link PostOrder#CUSTOM}
-   * all event handlers run asynchronously by default. You are encouraged to determine whether or
+   * in order to use this field.</p>
-   * not to enable it now. This option is being provided as a migration aid.</strong></p>
    *
-   * <p>If this method returns {@code true}, the method is guaranteed to be executed
+   * @return the priority
-   * asynchronously. Otherwise, the handler may be executed on the current thread or
+   */
-   * asynchronously. <strong>This still means you must consider thread-safety in your
+  short priority() default Short.MIN_VALUE;
-   * event listeners</strong> as the "current thread" can and will be different each time.</p>
+
+  /**
+   * Whether the handler must be called asynchronously. By default, all event handlers are called
+   * asynchronously.
    *
-   * <p>If any method handler targeting an event type is marked with {@code true}, then every
+   * <p>For performance (for instance, if you use {@link EventTask#withContinuation}), you can
-   * handler targeting that event type will be executed asynchronously.</p>
+   * optionally specify <code>false</code>. This option will become {@code false} by default
+   * in a future release of Velocity.</p>
+   *
+   * <p>If this is {@code true}, the method is guaranteed to be executed asynchronously. Otherwise,
+   * the handler may be executed on the current thread or asynchronously. <strong>This still means
+   * you must consider thread-safety in your event listeners</strong> as the "current thread" can
+   * and will be different each time.</p>
+   *
+   * <p>Note that if any method handler targeting an event type is marked with {@code true}, then
+   * every handler targeting that event type will be executed asynchronously.</p>
    *
    * @return Requires async
    */

gradle.properties

@@ -1,2 +1,2 @@
 group=com.velocitypowered
-version=3.3.0-SNAPSHOT
+version=3.4.0-SNAPSHOT

native/README.md

@@ -18,13 +18,11 @@ CommonCrypto library.
 - macOS aarch64 ("Apple Silicon")
 
 For Linux platforms, we provide two versions of the native library: one built against OpenSSL 1.1.x and one built against OpenSSL 3.x.x.
-All native libraries are built on various versions of Ubuntu:
+All native libraries are built on various versions of Ubuntu and Alpine:
 
 - Ubuntu 20.04 for OpenSSL 1.1.x support and for compression
 - Ubuntu 22.04 for OpenSSL 3.x.x support
-
+- Alpine 3.18 for OpenSSL 3.x.x support and compression (musl libc users only)
-Currently, we do not provide native libraries for distributions based on musl libc, like Alpine Linux. You might be able to use `apk add libc6-compat` to fake it, but this is not officially supported.
-In the future we may provide a musl libc build.
 
 ## Building
 

native/build-support/alpine.Dockerfile

@@ -0,0 +1,9 @@
+
+FROM amazoncorretto:17.0.12-alpine3.18
+
+# Install required dependencies
+RUN apk add --no-cache bash alpine-sdk cmake openssl-dev openssl
+
+# Create a non-root user
+RUN adduser -D user
+USER user

native/build-support/build-all-linux-natives.sh

@@ -6,7 +6,8 @@ set -e
 cd "$(dirname "$0")/.." || exit 1
 
 ARCHS=(x86_64 aarch64)
-BASE_DOCKERFILE_VARIANTS=(ubuntu-focal ubuntu-jammy)
+BASE_DOCKERFILE_VARIANTS=(ubuntu-focal ubuntu-jammy alpine)
+COMPRESSION_VARIANTS=(ubuntu-focal alpine)
 
 for variant in "${BASE_DOCKERFILE_VARIANTS[@]}"; do
   docker_platforms=""
@@ -25,8 +26,8 @@ for arch in "${ARCHS[@]}"; do
     docker run --rm -v "$(pwd)":/app --platform linux/${arch} velocity-native-build:$variant /bin/bash -c "cd /app && ./build-support/compile-linux-crypto.sh"
   done
 
-  # Use only the oldest variant for the compression library
+  for variant in "${COMPRESSION_VARIANTS[@]}"; do
-  variant=${BASE_DOCKERFILE_VARIANTS[0]}
+    echo "Building native compression for $arch on $variant..."
-  echo "Building native compression for $arch on $variant..."
+    docker run --rm -v "$(pwd)":/app --platform linux/${arch} velocity-native-build:$variant /bin/bash -c "cd /app && ./build-support/compile-linux-compress.sh"
-  docker run --rm -v "$(pwd)":/app --platform linux/${arch} velocity-native-build:$variant /bin/bash -c "cd /app && ./build-support/compile-linux-compress.sh"
+  done
 done

native/build-support/compile-linux-compress.sh

@@ -16,8 +16,14 @@ cd libdeflate || exit
 rm -rf build && cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -B build && cmake --build build --target libdeflate_static
 cd ..
 
+# Determine if we are on musl libc or glibc
+suffix=""
+if ldd --version 2>&1 | grep -q musl; then
+  suffix="-musl"
+fi
+
 CFLAGS="-O2 -I$JAVA_HOME/include/ -I$JAVA_HOME/include/linux/ -fPIC -shared -Wl,-z,noexecstack -Wall -Werror -fomit-frame-pointer"
 ARCH=$(uname -m)
 mkdir -p src/main/resources/linux_$ARCH
 $CC $CFLAGS -Ilibdeflate src/main/c/jni_util.c src/main/c/jni_zlib_deflate.c src/main/c/jni_zlib_inflate.c \
-    libdeflate/build/libdeflate.a -o src/main/resources/linux_$ARCH/velocity-compress.so
+    libdeflate/build/libdeflate.a -o src/main/resources/linux_$ARCH/velocity-compress$suffix.so

native/build-support/compile-linux-crypto.sh

@@ -20,9 +20,11 @@ if [ ! "$CC" ]; then
   export CC=gcc
 fi
 
-output_file="velocity-cipher.so"
+# Determine if we are on musl libc or glibc
-if [ -n "$OPENSSL_VERSION" ]; then
+suffix=""
-  output_file="velocity-cipher-ossl${OPENSSL_VERSION}.so"
+if ldd --version 2>&1 | grep -q musl; then
+  suffix="-musl"
+  filename=$(echo "$filename" | sed "s/\.so/$suffix.so/")
 fi
 
 CFLAGS="-O2 -I$JAVA_HOME/include/ -I$JAVA_HOME/include/linux/ -fPIC -shared -Wl,-z,noexecstack -Wall -Werror -fomit-frame-pointer"

native/src/main/java/com/velocitypowered/natives/compression/CompressorUtils.java

@@ -17,29 +17,12 @@
 
 package com.velocitypowered.natives.compression;
 
-import io.netty.buffer.ByteBuf;
-import java.util.zip.DataFormatException;
-
 class CompressorUtils {
   /**
    * The default preferred output buffer size for zlib.
    */
   static final int ZLIB_BUFFER_SIZE = 8192;
 
-  /**
-   * Ensures that the buffer does not go over {@code max}.
-   *
-   * @param buf the buffer for check
-   * @param max the maximum size for the buffer
-   * @throws DataFormatException if the buffer becomes too bug
-   */
-  static void ensureMaxSize(ByteBuf buf, int max) throws DataFormatException {
-    int len = buf.readableBytes();
-    if (len > max) {
-      throw new DataFormatException("Got too much data (" + len + " > " + max + ")");
-    }
-  }
-
   private CompressorUtils() {
     throw new AssertionError();
   }

native/src/main/java/com/velocitypowered/natives/util/MoreByteBufUtils.java

@@ -52,18 +52,13 @@ public class MoreByteBufUtils {
 
   private static boolean isCompatible(Native nativeStuff, ByteBuf buf) {
     BufferPreference preferred = nativeStuff.preferredBufferType();
-    switch (preferred) {
+    return switch (preferred) {
-      case DIRECT_PREFERRED:
+      case DIRECT_PREFERRED, HEAP_PREFERRED ->
-      case HEAP_PREFERRED:
         // The native prefers this type, but doesn't strictly require we provide it.
-        return true;
+        true;
-      case DIRECT_REQUIRED:
+      case DIRECT_REQUIRED -> buf.hasMemoryAddress();
-        return buf.hasMemoryAddress();
+      case HEAP_REQUIRED -> buf.hasArray();
-      case HEAP_REQUIRED:
+    };
-        return buf.hasArray();
-      default:
-        throw new AssertionError("Preferred buffer type unknown");
-    }
   }
 
   /**
@@ -77,15 +72,9 @@ public class MoreByteBufUtils {
    */
   public static ByteBuf preferredBuffer(ByteBufAllocator alloc, Native nativeStuff,
       int initialCapacity) {
-    switch (nativeStuff.preferredBufferType()) {
+    return switch (nativeStuff.preferredBufferType()) {
-      case HEAP_REQUIRED:
+      case HEAP_REQUIRED, HEAP_PREFERRED -> alloc.heapBuffer(initialCapacity);
-      case HEAP_PREFERRED:
+      case DIRECT_PREFERRED, DIRECT_REQUIRED -> alloc.directBuffer(initialCapacity);
-        return alloc.heapBuffer(initialCapacity);
+    };
-      case DIRECT_PREFERRED:
-      case DIRECT_REQUIRED:
-        return alloc.directBuffer(initialCapacity);
-      default:
-        throw new AssertionError("Preferred buffer type unknown");
-    }
   }
 }

native/src/main/java/com/velocitypowered/natives/util/NativeConstraints.java

@@ -19,6 +19,7 @@ package com.velocitypowered.natives.util;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import java.nio.charset.StandardCharsets;
 import java.util.function.BooleanSupplier;
 
 /**
@@ -29,6 +30,8 @@ public class NativeConstraints {
   private static final boolean IS_AMD64;
   private static final boolean IS_AARCH64;
   private static final boolean CAN_GET_MEMORYADDRESS;
+  private static final boolean IS_LINUX;
+  private static final boolean IS_MUSL_LIBC;
 
   static {
     ByteBuf test = Unpooled.directBuffer();
@@ -39,21 +42,46 @@ public class NativeConstraints {
     }
 
     String osArch = System.getProperty("os.arch", "");
-    // HotSpot on Intel macOS prefers x86_64, but OpenJ9 on macOS and HotSpot/OpenJ9 elsewhere
-    // give amd64.
     IS_AMD64 = osArch.equals("amd64") || osArch.equals("x86_64");
     IS_AARCH64 = osArch.equals("aarch64") || osArch.equals("arm64");
+
+    IS_LINUX = System.getProperty("os.name", "").equalsIgnoreCase("Linux");
+
+    // Determine if we're using musl libc by invoking `ldd --version`.
+    if (IS_LINUX) {
+      boolean isMusl;
+      try {
+        Process process = new ProcessBuilder("ldd", "--version")
+            .redirectErrorStream(true)
+            .start();
+        process.waitFor();
+        try (var reader = process.getInputStream()) {
+          byte[] outputRaw = reader.readAllBytes();
+          String output = new String(outputRaw, StandardCharsets.UTF_8);
+          isMusl = output.contains("musl");
+        }
+      } catch (Exception e) {
+        isMusl = false;
+      }
+      IS_MUSL_LIBC = isMusl;
+    } else {
+      IS_MUSL_LIBC = false;
+    }
   }
 
   static final BooleanSupplier NATIVE_BASE = () -> NATIVES_ENABLED && CAN_GET_MEMORYADDRESS;
 
   static final BooleanSupplier LINUX_X86_64 = () -> NATIVE_BASE.getAsBoolean()
-      && System.getProperty("os.name", "").equalsIgnoreCase("Linux")
+      && IS_LINUX && IS_AMD64 && !IS_MUSL_LIBC;
-      && IS_AMD64;
+
+  static final BooleanSupplier LINUX_X86_64_MUSL = () -> NATIVE_BASE.getAsBoolean()
+      && IS_LINUX && IS_AMD64 && IS_MUSL_LIBC;
 
   static final BooleanSupplier LINUX_AARCH64 = () -> NATIVE_BASE.getAsBoolean()
-      && System.getProperty("os.name", "").equalsIgnoreCase("Linux")
+      && IS_LINUX && IS_AARCH64 && !IS_MUSL_LIBC;
-      && IS_AARCH64;
+
+  static final BooleanSupplier LINUX_AARCH64_MUSL = () -> NATIVE_BASE.getAsBoolean()
+      && IS_LINUX && IS_AARCH64 && IS_MUSL_LIBC;
 
   static final BooleanSupplier MACOS_AARCH64 = () -> NATIVE_BASE.getAsBoolean()
       && System.getProperty("os.name", "").equalsIgnoreCase("Mac OS X")

native/src/main/java/com/velocitypowered/natives/util/Natives.java

@@ -83,11 +83,21 @@ public class Natives {
           new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_X86_64,
               copyAndLoadNative("/linux_x86_64/velocity-compress.so"),
               "libdeflate (Linux x86_64)",
-              LibdeflateVelocityCompressor.FACTORY), // compiled with Debian 10
+              LibdeflateVelocityCompressor.FACTORY), // compiled with Ubuntu 20.04
+          new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_X86_64_MUSL,
+              copyAndLoadNative("/linux_x86_64/velocity-compress-musl.so"),
+              "libdeflate (Linux x86_64, musl)",
+              LibdeflateVelocityCompressor.FACTORY), // compiled with Alpine 3.18
+
           new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_AARCH64,
               copyAndLoadNative("/linux_aarch64/velocity-compress.so"),
               "libdeflate (Linux aarch64)",
-              LibdeflateVelocityCompressor.FACTORY), // compiled with Fedora 36
+              LibdeflateVelocityCompressor.FACTORY), // compiled with Ubuntu 20.04
+          new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_AARCH64_MUSL,
+              copyAndLoadNative("/linux_aarch64/velocity-compress-musl.so"),
+              "libdeflate (Linux aarch64, musl)",
+              LibdeflateVelocityCompressor.FACTORY), // compiled with Alpine 3.18
+
           new NativeCodeLoader.Variant<>(NativeConstraints.MACOS_AARCH64,
               copyAndLoadNative("/macos_arm64/velocity-compress.dylib"),
               "libdeflate (macOS ARM64 / Apple Silicon)",
@@ -108,6 +118,9 @@ public class Natives {
           new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_X86_64,
               copyAndLoadNative("/linux_x86_64/velocity-cipher-ossl11x.so"), // Ubuntu 20.04
               "OpenSSL 1.1.x (Linux x86_64)", NativeVelocityCipher.FACTORY),
+          new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_X86_64_MUSL,
+              copyAndLoadNative("/linux_x86_64/velocity-cipher-ossl30x-musl.so"), // Alpine 3.18
+              "OpenSSL 3.x.x (Linux x86_64, musl)", NativeVelocityCipher.FACTORY),
 
           new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_AARCH64,
               copyAndLoadNative("/linux_aarch64/velocity-cipher.so"),
@@ -118,6 +131,9 @@ public class Natives {
           new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_AARCH64,
               copyAndLoadNative("/linux_aarch64/velocity-cipher-ossl11x.so"),
               "OpenSSL 1.1.x (Linux aarch64)", NativeVelocityCipher.FACTORY), // Ubuntu 20.04
+          new NativeCodeLoader.Variant<>(NativeConstraints.LINUX_AARCH64_MUSL,
+              copyAndLoadNative("/linux_aarch64/velocity-cipher-ossl30x-musl.so"),
+              "OpenSSL 3.x.x (Linux aarch64, musl)", NativeVelocityCipher.FACTORY), // Alpine 3.18
 
           new NativeCodeLoader.Variant<>(NativeConstraints.MACOS_AARCH64,
               copyAndLoadNative("/macos_arm64/velocity-cipher.dylib"),

proxy/build.gradle.kts

@@ -92,6 +92,15 @@ tasks {
         dependsOn(configurateBuildTask)
         from(zipTree(configurateBuildTask.map { it.outputs.files.singleFile }))
     }
+
+    runShadow {
+        workingDir = file("run").also(File::mkdirs)
+        standardInput = System.`in`
+    }
+    named<JavaExec>("run") {
+        workingDir = file("run").also(File::mkdirs)
+        standardInput = System.`in` // Doesn't work?
+    }
 }
 
 dependencies {

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

@@ -22,11 +22,13 @@ import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
+import com.velocitypowered.api.command.BrigadierCommand;
 import com.velocitypowered.api.event.proxy.ProxyInitializeEvent;
 import com.velocitypowered.api.event.proxy.ProxyReloadEvent;
 import com.velocitypowered.api.event.proxy.ProxyShutdownEvent;
 import com.velocitypowered.api.network.ProtocolVersion;
 import com.velocitypowered.api.plugin.PluginContainer;
+import com.velocitypowered.api.plugin.PluginDescription;
 import com.velocitypowered.api.plugin.PluginManager;
 import com.velocitypowered.api.proxy.Player;
 import com.velocitypowered.api.proxy.ProxyServer;
@@ -52,6 +54,9 @@ import com.velocitypowered.proxy.crypto.EncryptionUtils;
 import com.velocitypowered.proxy.event.VelocityEventManager;
 import com.velocitypowered.proxy.network.ConnectionManager;
 import com.velocitypowered.proxy.plugin.VelocityPluginManager;
+import com.velocitypowered.proxy.plugin.loader.VelocityPluginContainer;
+import com.velocitypowered.proxy.plugin.loader.VelocityPluginDescription;
+import com.velocitypowered.proxy.plugin.virtual.VelocityVirtualPlugin;
 import com.velocitypowered.proxy.protocol.ProtocolUtils;
 import com.velocitypowered.proxy.protocol.util.FaviconSerializer;
 import com.velocitypowered.proxy.protocol.util.GameProfileSerializer;
@@ -77,6 +82,7 @@ import java.nio.file.Path;
 import java.security.KeyPair;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
@@ -100,6 +106,7 @@ import net.kyori.adventure.translation.GlobalTranslator;
 import net.kyori.adventure.translation.TranslationRegistry;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.bstats.MetricsBase;
 import org.checkerframework.checker.nullness.qual.EnsuresNonNull;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -110,6 +117,8 @@ import org.checkerframework.checker.nullness.qual.Nullable;
  */
 public class VelocityServer implements ProxyServer, ForwardingAudience {
 
+  public static final String VELOCITY_URL = "https://velocitypowered.com";
+
   private static final Logger logger = LogManager.getLogger(VelocityServer.class);
   public static final Gson GENERAL_GSON = new GsonBuilder()
       .registerTypeHierarchyAdapter(Favicon.class, FaviconSerializer.INSTANCE)
@@ -162,7 +171,7 @@ public class VelocityServer implements ProxyServer, ForwardingAudience {
   VelocityServer(final ProxyOptions options) {
     pluginManager = new VelocityPluginManager(this);
     eventManager = new VelocityEventManager(pluginManager);
-    commandManager = new VelocityCommandManager(eventManager);
+    commandManager = new VelocityCommandManager(eventManager, pluginManager);
     scheduler = new VelocityScheduler(pluginManager);
     console = new VelocityConsole(this);
     cm = new ConnectionManager(this);
@@ -199,6 +208,16 @@ public class VelocityServer implements ProxyServer, ForwardingAudience {
     return new ProxyVersion(implName, implVendor, implVersion);
   }
 
+  private VelocityPluginContainer createVirtualPlugin() {
+    ProxyVersion version = getVersion();
+    PluginDescription description = new VelocityPluginDescription(
+        "velocity", version.getName(), version.getVersion(), "The Velocity proxy",
+        VELOCITY_URL, ImmutableList.of(version.getVendor()), Collections.emptyList(), null);
+    VelocityPluginContainer container = new VelocityPluginContainer(description);
+    container.setInstance(VelocityVirtualPlugin.INSTANCE);
+    return container;
+  }
+
   @Override
   public VelocityCommandManager getCommandManager() {
     return commandManager;
@@ -213,6 +232,7 @@ public class VelocityServer implements ProxyServer, ForwardingAudience {
   void start() {
     logger.info("Booting up {} {}...", getVersion().getName(), getVersion().getVersion());
     console.setupStreams();
+    pluginManager.registerPlugin(this.createVirtualPlugin());
 
     registerTranslations();
 
@@ -221,11 +241,35 @@ public class VelocityServer implements ProxyServer, ForwardingAudience {
     cm.logChannelInformation();
 
     // Initialize commands first
-    commandManager.register(VelocityCommand.create(this));
+    final BrigadierCommand velocityParentCommand = VelocityCommand.create(this);
-    commandManager.register(CallbackCommand.create());
+    commandManager.register(
-    commandManager.register(ServerCommand.create(this));
+        commandManager.metaBuilder(velocityParentCommand)
-    commandManager.register("shutdown", ShutdownCommand.command(this),
+            .plugin(VelocityVirtualPlugin.INSTANCE)
-        "end", "stop");
+            .build(),
+        velocityParentCommand
+    );
+    final BrigadierCommand callbackCommand = CallbackCommand.create();
+    commandManager.register(
+        commandManager.metaBuilder(callbackCommand)
+            .plugin(VelocityVirtualPlugin.INSTANCE)
+            .build(),
+        velocityParentCommand
+    );
+    final BrigadierCommand serverCommand = ServerCommand.create(this);
+    commandManager.register(
+        commandManager.metaBuilder(serverCommand)
+            .plugin(VelocityVirtualPlugin.INSTANCE)
+            .build(),
+        serverCommand
+    );
+    final BrigadierCommand shutdownCommand = ShutdownCommand.command(this);
+    commandManager.register(
+        commandManager.metaBuilder(shutdownCommand)
+            .plugin(VelocityVirtualPlugin.INSTANCE)
+            .aliases("end", "stop")
+            .build(),
+        shutdownCommand
+    );
     new GlistCommand(this).register();
     new SendCommand(this).register();
 
@@ -263,7 +307,13 @@ public class VelocityServer implements ProxyServer, ForwardingAudience {
       this.cm.queryBind(configuration.getBind().getHostString(), configuration.getQueryPort());
     }
 
-    Metrics.VelocityMetrics.startMetrics(this, configuration.getMetrics());
+    final String defaultPackage = new String(
+        new byte[] { 'o', 'r', 'g', '.', 'b', 's', 't', 'a', 't', 's' });
+    if (!MetricsBase.class.getPackage().getName().startsWith(defaultPackage)) {
+      Metrics.VelocityMetrics.startMetrics(this, configuration.getMetrics());
+    } else {
+      logger.warn("debug environment, metrics is disabled!");
+    }
   }
 
   private void registerTranslations() {
@@ -533,7 +583,6 @@ public class VelocityServer implements ProxyServer, ForwardingAudience {
 
         eventManager.fire(new ProxyShutdownEvent()).join();
 
-        timedOut = !eventManager.shutdown() || timedOut;
         timedOut = !scheduler.shutdown() || timedOut;
 
         if (timedOut) {

proxy/src/main/java/com/velocitypowered/proxy/command/VelocityCommandManager.java

@@ -20,6 +20,7 @@ package com.velocitypowered.proxy.command;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.mojang.brigadier.CommandDispatcher;
 import com.mojang.brigadier.Message;
 import com.mojang.brigadier.ParseResults;
@@ -37,11 +38,15 @@ import com.velocitypowered.api.command.CommandSource;
 import com.velocitypowered.api.command.VelocityBrigadierMessage;
 import com.velocitypowered.api.event.command.CommandExecuteEvent;
 import com.velocitypowered.api.event.command.PostCommandInvocationEvent;
+import com.velocitypowered.api.plugin.PluginManager;
+import com.velocitypowered.proxy.command.brigadier.VelocityBrigadierCommandWrapper;
 import com.velocitypowered.proxy.command.registrar.BrigadierCommandRegistrar;
 import com.velocitypowered.proxy.command.registrar.CommandRegistrar;
 import com.velocitypowered.proxy.command.registrar.RawCommandRegistrar;
 import com.velocitypowered.proxy.command.registrar.SimpleCommandRegistrar;
 import com.velocitypowered.proxy.event.VelocityEventManager;
+import com.velocitypowered.proxy.plugin.virtual.VelocityVirtualPlugin;
+import io.netty.util.concurrent.FastThreadLocalThread;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -49,6 +54,7 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.stream.Collectors;
@@ -71,13 +77,16 @@ public class VelocityCommandManager implements CommandManager {
   private final SuggestionsProvider<CommandSource> suggestionsProvider;
   private final CommandGraphInjector<CommandSource> injector;
   private final Map<String, CommandMeta> commandMetas;
+  private final PluginManager pluginManager;
 
   /**
    * Constructs a command manager.
    *
    * @param eventManager the event manager
    */
-  public VelocityCommandManager(final VelocityEventManager eventManager) {
+  public VelocityCommandManager(final VelocityEventManager eventManager,
+      PluginManager pluginManager) {
+    this.pluginManager = pluginManager;
     this.lock = new ReentrantReadWriteLock();
     this.dispatcher = new CommandDispatcher<>();
     this.eventManager = Preconditions.checkNotNull(eventManager);
@@ -218,16 +227,13 @@ public class VelocityCommandManager implements CommandManager {
     return eventManager.fire(new CommandExecuteEvent(source, cmdLine));
   }
 
-  private boolean executeImmediately0(final CommandSource source, final String cmdLine) {
+  private boolean executeImmediately0(final CommandSource source, final ParseResults<CommandSource> parsed) {
     Preconditions.checkNotNull(source, "source");
-    Preconditions.checkNotNull(cmdLine, "cmdLine");
 
-    final String normalizedInput = VelocityCommands.normalizeInput(cmdLine, true);
     CommandResult result = CommandResult.EXCEPTION;
     try {
       // The parse can fail if the requirement predicates throw
-      final ParseResults<CommandSource> parse = this.parse(normalizedInput, source);
+      boolean executed = dispatcher.execute(parsed) != BrigadierCommand.FORWARD;
-      boolean executed = dispatcher.execute(parse) != BrigadierCommand.FORWARD;
       result = executed ? CommandResult.EXECUTED : CommandResult.FORWARDED;
       return executed;
     } catch (final CommandSyntaxException e) {
@@ -249,9 +255,9 @@ public class VelocityCommandManager implements CommandManager {
       }
     } catch (final Throwable e) {
       // Ugly, ugly swallowing of everything Throwable, because plugins are naughty.
-      throw new RuntimeException("Unable to invoke command " + cmdLine + " for " + source, e);
+      throw new RuntimeException("Unable to invoke command  " + parsed.getReader().getString() + "for " + source, e);
     } finally {
-      eventManager.fireAndForget(new PostCommandInvocationEvent(source, cmdLine, result));
+      eventManager.fireAndForget(new PostCommandInvocationEvent(source, parsed.getReader().getString(), result));
     }
   }
 
@@ -260,13 +266,17 @@ public class VelocityCommandManager implements CommandManager {
     Preconditions.checkNotNull(source, "source");
     Preconditions.checkNotNull(cmdLine, "cmdLine");
 
-    return callCommandEvent(source, cmdLine).thenApplyAsync(event -> {
+    return callCommandEvent(source, cmdLine).thenComposeAsync(event -> {
       CommandExecuteEvent.CommandResult commandResult = event.getResult();
       if (commandResult.isForwardToServer() || !commandResult.isAllowed()) {
-        return false;
+        return CompletableFuture.completedFuture(false);
       }
-      return executeImmediately0(source, commandResult.getCommand().orElse(event.getCommand()));
+      final ParseResults<CommandSource> parsed = this.parse(
-    }, eventManager.getAsyncExecutor());
+          commandResult.getCommand().orElse(cmdLine), source);
+      return CompletableFuture.supplyAsync(
+          () -> executeImmediately0(source, parsed), this.getAsyncExecutor(parsed)
+      );
+    }, figureAsyncExecutorForParsing());
   }
 
   @Override
@@ -276,7 +286,12 @@ public class VelocityCommandManager implements CommandManager {
     Preconditions.checkNotNull(cmdLine, "cmdLine");
 
     return CompletableFuture.supplyAsync(
-        () -> executeImmediately0(source, cmdLine), eventManager.getAsyncExecutor());
+        () -> this.parse(cmdLine, source), figureAsyncExecutorForParsing()
+    ).thenCompose(
+        parsed -> CompletableFuture.supplyAsync(
+            () -> executeImmediately0(source, parsed), this.getAsyncExecutor(parsed)
+        )
+    );
   }
 
   /**
@@ -324,9 +339,10 @@ public class VelocityCommandManager implements CommandManager {
    * @return the parse results
    */
   private ParseResults<CommandSource> parse(final String input, final CommandSource source) {
+    final String normalizedInput = VelocityCommands.normalizeInput(input, true);
     lock.readLock().lock();
     try {
-      return dispatcher.parse(input, source);
+      return dispatcher.parse(normalizedInput, source);
     } finally {
       lock.readLock().unlock();
     }
@@ -370,4 +386,25 @@ public class VelocityCommandManager implements CommandManager {
   public CommandGraphInjector<CommandSource> getInjector() {
     return injector;
   }
+
+  private Executor getAsyncExecutor(ParseResults<CommandSource> parse) {
+    Object registrant;
+    if (parse.getContext().getCommand() instanceof VelocityBrigadierCommandWrapper vbcw) {
+      registrant = vbcw.registrant() == null ? VelocityVirtualPlugin.INSTANCE : vbcw.registrant();
+    } else {
+      registrant = VelocityVirtualPlugin.INSTANCE;
+    }
+    return pluginManager.ensurePluginContainer(registrant).getExecutorService();
+  }
+
+  private Executor figureAsyncExecutorForParsing() {
+    final Thread thread = Thread.currentThread();
+    if (thread instanceof FastThreadLocalThread) {
+      // we *never* want to block the Netty event loop, so use the async executor
+      return pluginManager.ensurePluginContainer(VelocityVirtualPlugin.INSTANCE).getExecutorService();
+    } else {
+      // it's some other thread that isn't a Netty event loop thread. direct execution it is!
+      return MoreExecutors.directExecutor();
+    }
+  }
 }

proxy/src/main/java/com/velocitypowered/proxy/command/VelocityCommands.java

@@ -24,6 +24,7 @@ import com.mojang.brigadier.context.CommandContext;
 import com.mojang.brigadier.context.CommandContextBuilder;
 import com.mojang.brigadier.context.ParsedArgument;
 import com.mojang.brigadier.context.ParsedCommandNode;
+import com.mojang.brigadier.tree.ArgumentCommandNode;
 import com.mojang.brigadier.tree.CommandNode;
 import com.mojang.brigadier.tree.LiteralCommandNode;
 import com.mojang.brigadier.tree.RootCommandNode;
@@ -32,6 +33,7 @@ import com.velocitypowered.api.command.CommandManager;
 import com.velocitypowered.api.command.CommandSource;
 import com.velocitypowered.api.command.InvocableCommand;
 import com.velocitypowered.proxy.command.brigadier.VelocityArgumentCommandNode;
+import com.velocitypowered.proxy.command.brigadier.VelocityBrigadierCommandWrapper;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -44,6 +46,59 @@ import org.checkerframework.checker.nullness.qual.Nullable;
  */
 public final class VelocityCommands {
 
+  // Wrapping
+
+  /**
+   * Walks the command node tree and wraps all {@link Command} instances in a {@link VelocityBrigadierCommandWrapper},
+   * to indicate the plugin that registered the command. This also has the side effect of cloning
+   * the command node tree.
+   *
+   * @param delegate the command node to wrap
+   * @param registrant the plugin that registered the command
+   * @return the wrapped command node
+   */
+  public static CommandNode<CommandSource> wrap(final CommandNode<CommandSource> delegate,
+      final @Nullable Object registrant) {
+    Preconditions.checkNotNull(delegate, "delegate");
+    if (registrant == null) {
+      // the registrant is null if the `plugin` was absent when we try to register the command
+      return delegate;
+    }
+
+    com.mojang.brigadier.Command<CommandSource> maybeCommand = delegate.getCommand();
+    if (maybeCommand != null && !(maybeCommand instanceof VelocityBrigadierCommandWrapper)) {
+      maybeCommand = VelocityBrigadierCommandWrapper.wrap(delegate.getCommand(), registrant);
+    }
+
+    if (delegate instanceof LiteralCommandNode<CommandSource> lcn) {
+      var literalBuilder = shallowCopyAsBuilder(lcn, delegate.getName(), true);
+      literalBuilder.executes(maybeCommand);
+      // we also need to wrap any children
+      for (final CommandNode<CommandSource> child : delegate.getChildren()) {
+        literalBuilder.then(wrap(child, registrant));
+      }
+      if (delegate.getRedirect() != null) {
+        literalBuilder.redirect(wrap(delegate.getRedirect(), registrant));
+      }
+      return literalBuilder.build();
+    } else if (delegate instanceof VelocityArgumentCommandNode<CommandSource, ?> vacn) {
+      return vacn.withCommand(maybeCommand)
+          .withRedirect(delegate.getRedirect() != null ? wrap(delegate.getRedirect(), registrant) : null);
+    } else if (delegate instanceof ArgumentCommandNode) {
+      var argBuilder = delegate.createBuilder().executes(maybeCommand);
+      // we also need to wrap any children
+      for (final CommandNode<CommandSource> child : delegate.getChildren()) {
+        argBuilder.then(wrap(child, registrant));
+      }
+      if (delegate.getRedirect() != null) {
+        argBuilder.redirect(wrap(delegate.getRedirect(), registrant));
+      }
+      return argBuilder.build();
+    } else {
+      throw new IllegalArgumentException("Unsupported node type: " + delegate.getClass());
+    }
+  }
+
   // Normalization
 
   /**
@@ -135,6 +190,33 @@ public final class VelocityCommands {
    */
   public static LiteralCommandNode<CommandSource> shallowCopy(
       final LiteralCommandNode<CommandSource> original, final String newName) {
+    return shallowCopy(original, newName, original.getCommand());
+  }
+
+  /**
+   * Creates a copy of the given literal with the specified name.
+   *
+   * @param original   the literal node to copy
+   * @param newName    the name of the returned literal node
+   * @param newCommand the new command to set on the copied node
+   * @return a copy of the literal with the given name
+   */
+  private static LiteralCommandNode<CommandSource> shallowCopy(
+      final LiteralCommandNode<CommandSource> original, final String newName,
+      final com.mojang.brigadier.Command<CommandSource> newCommand) {
+    return shallowCopyAsBuilder(original, newName, false).executes(newCommand).build();
+  }
+
+  /**
+   * Creates a copy of the given literal with the specified name.
+   *
+   * @param original the literal node to copy
+   * @param newName  the name of the returned literal node
+   * @return a copy of the literal with the given name
+   */
+  private static LiteralArgumentBuilder<CommandSource> shallowCopyAsBuilder(
+      final LiteralCommandNode<CommandSource> original, final String newName,
+      final boolean skipChildren) {
     // Brigadier resolves the redirect of a node if further input can be parsed.
     // Let <bar> be a literal node having a redirect to a <foo> literal. Then,
     // the context returned by CommandDispatcher#parseNodes when given the input
@@ -150,10 +232,12 @@ public final class VelocityCommands {
         .requiresWithContext(original.getContextRequirement())
         .forward(original.getRedirect(), original.getRedirectModifier(), original.isFork())
         .executes(original.getCommand());
-    for (final CommandNode<CommandSource> child : original.getChildren()) {
+    if (!skipChildren) {
-      builder.then(child);
+      for (final CommandNode<CommandSource> child : original.getChildren()) {
+        builder.then(child);
+      }
     }
-    return builder.build();
+    return builder;
   }
 
   // Arguments node

proxy/src/main/java/com/velocitypowered/proxy/command/brigadier/VelocityArgumentCommandNode.java

@@ -93,6 +93,16 @@ public class VelocityArgumentCommandNode<S, T> extends ArgumentCommandNode<S, St
     throw new UnsupportedOperationException();
   }
 
+  public VelocityArgumentCommandNode<S, T> withCommand(Command<S> command) {
+    return new VelocityArgumentCommandNode<>(getName(), type, command, getRequirement(),
+        getContextRequirement(), getRedirect(), getRedirectModifier(), isFork(), getCustomSuggestions());
+  }
+
+  public VelocityArgumentCommandNode<S, T> withRedirect(CommandNode<S> target) {
+    return new VelocityArgumentCommandNode<>(getName(), type, getCommand(), getRequirement(),
+        getContextRequirement(), target, getRedirectModifier(), isFork(), getCustomSuggestions());
+  }
+
   @Override
   public boolean isValidInput(final String input) {
     return true;

proxy/src/main/java/com/velocitypowered/proxy/command/brigadier/VelocityBrigadierCommandWrapper.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright (C) 2024 Velocity 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 <https://www.gnu.org/licenses/>.
+ */
+
+package com.velocitypowered.proxy.command.brigadier;
+
+import com.mojang.brigadier.Command;
+import com.mojang.brigadier.context.CommandContext;
+import com.mojang.brigadier.exceptions.CommandSyntaxException;
+import com.velocitypowered.api.command.CommandSource;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Wraps a Brigadier command to allow us to track the registrant.
+ */
+public class VelocityBrigadierCommandWrapper implements Command<CommandSource> {
+
+  private final Command<CommandSource> delegate;
+  private final Object registrant;
+
+  private VelocityBrigadierCommandWrapper(Command<CommandSource> delegate, Object registrant) {
+    this.delegate = delegate;
+    this.registrant = registrant;
+  }
+
+  /**
+   * Transforms the given command into a {@code VelocityBrigadierCommandWrapper} if the registrant
+   * is not null and if the command is not already wrapped.
+   *
+   * @param delegate the command to wrap
+   * @param registrant the registrant of the command
+   * @return the wrapped command, if necessary
+   */
+  public static Command<CommandSource> wrap(Command<CommandSource> delegate, @Nullable Object registrant) {
+    if (registrant == null) {
+      // nothing to wrap
+      return delegate;
+    }
+    if (delegate instanceof VelocityBrigadierCommandWrapper) {
+      // already wrapped
+      return delegate;
+    }
+    return new VelocityBrigadierCommandWrapper(delegate, registrant);
+  }
+
+  @Override
+  public int run(CommandContext<CommandSource> context) throws CommandSyntaxException {
+    return delegate.run(context);
+  }
+
+  public Object registrant() {
+    return registrant;
+  }
+}

proxy/src/main/java/com/velocitypowered/proxy/command/builtin/GlistCommand.java

@@ -31,6 +31,7 @@ import com.velocitypowered.api.permission.Tristate;
 import com.velocitypowered.api.proxy.Player;
 import com.velocitypowered.api.proxy.ProxyServer;
 import com.velocitypowered.api.proxy.server.RegisteredServer;
+import com.velocitypowered.proxy.plugin.virtual.VelocityVirtualPlugin;
 import java.util.List;
 import java.util.Optional;
 import net.kyori.adventure.text.Component;
@@ -80,7 +81,13 @@ public class GlistCommand {
         .executes(this::serverCount)
         .build();
     rootNode.then(serverNode);
-    server.getCommandManager().register(new BrigadierCommand(rootNode));
+    final BrigadierCommand command = new BrigadierCommand(rootNode);
+    server.getCommandManager().register(
+        server.getCommandManager().metaBuilder(command)
+            .plugin(VelocityVirtualPlugin.INSTANCE)
+            .build(),
+        command
+    );
   }
 
   private int totalCount(final CommandContext<CommandSource> context) {

proxy/src/main/java/com/velocitypowered/proxy/command/builtin/SendCommand.java

@@ -30,6 +30,7 @@ import com.velocitypowered.api.proxy.Player;
 import com.velocitypowered.api.proxy.ProxyServer;
 import com.velocitypowered.api.proxy.ServerConnection;
 import com.velocitypowered.api.proxy.server.RegisteredServer;
+import com.velocitypowered.proxy.plugin.virtual.VelocityVirtualPlugin;
 import java.util.Objects;
 import java.util.Optional;
 import net.kyori.adventure.text.Component;
@@ -96,7 +97,13 @@ public class SendCommand {
         .build();
     playerNode.then(serverNode);
     rootNode.then(playerNode.build());
-    server.getCommandManager().register(new BrigadierCommand(rootNode.build()));
+    final BrigadierCommand command = new BrigadierCommand(rootNode);
+    server.getCommandManager().register(
+        server.getCommandManager().metaBuilder(command)
+            .plugin(VelocityVirtualPlugin.INSTANCE)
+            .build(),
+        command
+    );
   }
 
   private int usage(final CommandContext<CommandSource> context) {

proxy/src/main/java/com/velocitypowered/proxy/command/builtin/ShutdownCommand.java

@@ -23,6 +23,7 @@ import com.mojang.brigadier.builder.LiteralArgumentBuilder;
 import com.mojang.brigadier.builder.RequiredArgumentBuilder;
 import com.velocitypowered.api.command.BrigadierCommand;
 import com.velocitypowered.api.command.CommandSource;
+import com.velocitypowered.api.proxy.ConsoleCommandSource;
 import com.velocitypowered.proxy.VelocityServer;
 import net.kyori.adventure.text.minimessage.MiniMessage;
 import net.kyori.adventure.text.serializer.legacy.LegacyComponentSerializer;
@@ -43,7 +44,7 @@ public final class ShutdownCommand {
    */
   public static BrigadierCommand command(final VelocityServer server) {
     return new BrigadierCommand(LiteralArgumentBuilder.<CommandSource>literal("shutdown")
-        .requires(source -> source == server.getConsoleCommandSource())
+        .requires(source -> source instanceof ConsoleCommandSource)
         .executes(context -> {
           server.shutdown(true);
           return Command.SINGLE_SUCCESS;

proxy/src/main/java/com/velocitypowered/proxy/command/registrar/BrigadierCommandRegistrar.java

@@ -40,17 +40,19 @@ public final class BrigadierCommandRegistrar extends AbstractCommandRegistrar<Br
     // Register it (if valid), since it's probably what the user expects.
     // If invalid, the metadata contains the same alias, but in lowercase.
     final LiteralCommandNode<CommandSource> literal = command.getNode();
+    final LiteralCommandNode<CommandSource> wrapped =
+        (LiteralCommandNode<CommandSource>) VelocityCommands.wrap(literal, meta.getPlugin());
     final String primaryAlias = literal.getName();
     if (VelocityCommands.isValidAlias(primaryAlias)) {
       // Register directly without copying
-      this.register(literal);
+      this.register(wrapped);
     }
 
     for (final String alias : meta.getAliases()) {
       if (primaryAlias.equals(alias)) {
         continue;
       }
-      this.register(literal, alias);
+      this.register(wrapped, alias);
     }
 
     // Brigadier commands don't support hinting, ignore

proxy/src/main/java/com/velocitypowered/proxy/command/registrar/InvocableCommandRegistrar.java

@@ -32,6 +32,7 @@ import com.velocitypowered.api.command.InvocableCommand;
 import com.velocitypowered.proxy.command.VelocityCommandMeta;
 import com.velocitypowered.proxy.command.VelocityCommands;
 import com.velocitypowered.proxy.command.brigadier.VelocityArgumentBuilder;
+import com.velocitypowered.proxy.command.brigadier.VelocityBrigadierCommandWrapper;
 import com.velocitypowered.proxy.command.invocation.CommandInvocationFactory;
 import java.util.Iterator;
 import java.util.concurrent.locks.Lock;
@@ -76,11 +77,11 @@ abstract class InvocableCommandRegistrar<T extends InvocableCommand<I>,
       final I invocation = invocationFactory.create(context);
       return command.hasPermission(invocation);
     };
-    final Command<CommandSource> callback = context -> {
+    final Command<CommandSource> callback = VelocityBrigadierCommandWrapper.wrap(context -> {
       final I invocation = invocationFactory.create(context);
       command.execute(invocation);
       return 1; // handled
-    };
+    }, meta.getPlugin());
 
     final LiteralCommandNode<CommandSource> literal = LiteralArgumentBuilder
         .<CommandSource>literal(alias)

proxy/src/main/java/com/velocitypowered/proxy/connection/backend/VelocityServerConnection.java

@@ -53,8 +53,6 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import net.kyori.adventure.nbt.CompoundBinaryTag;
-import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.jetbrains.annotations.NotNull;
 
@@ -72,7 +70,6 @@ public class VelocityServerConnection implements MinecraftConnectionAssociation,
   private boolean gracefulDisconnect = false;
   private BackendConnectionPhase connectionPhase = BackendConnectionPhases.UNKNOWN;
   private final Map<Long, Long> pendingPings = new HashMap<>();
-  private @MonotonicNonNull CompoundBinaryTag activeDimensionRegistry;
 
   /**
    * Initializes a new server connection.
@@ -366,12 +363,4 @@ public class VelocityServerConnection implements MinecraftConnectionAssociation,
   public boolean hasCompletedJoin() {
     return hasCompletedJoin;
   }
-
-  public CompoundBinaryTag getActiveDimensionRegistry() {
-    return activeDimensionRegistry;
-  }
-
-  public void setActiveDimensionRegistry(CompoundBinaryTag activeDimensionRegistry) {
-    this.activeDimensionRegistry = activeDimensionRegistry;
-  }
 }

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

@@ -565,8 +565,6 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
       }
     }
 
-    destination.setActiveDimensionRegistry(joinGame.getRegistry()); // 1.16
-
     // Remove previous boss bars. These don't get cleared when sending JoinGame, thus the need to
     // track them.
     for (UUID serverBossBar : serverBossBars) {

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

@@ -30,9 +30,9 @@ import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
 import java.net.InetSocketAddress;
-import java.util.ArrayDeque;
 import java.util.Optional;
 import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import net.kyori.adventure.text.Component;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
@@ -58,7 +58,7 @@ public class LoginInboundConnection implements LoginPhaseConnection, KeyIdentifi
       InitialInboundConnection delegate) {
     this.delegate = delegate;
     this.outstandingResponses = Int2ObjectSyncMap.hashmap();
-    this.loginMessagesToSend = new ArrayDeque<>();
+    this.loginMessagesToSend = new ConcurrentLinkedQueue<>();
   }
 
   @Override

proxy/src/main/java/com/velocitypowered/proxy/event/EventTypeTracker.java

@@ -33,8 +33,9 @@ class EventTypeTracker {
   }
 
   public Collection<Class<?>> getFriendsOf(final Class<?> eventType) {
-    if (friends.containsKey(eventType)) {
+    ImmutableSet<Class<?>> existingFriends = friends.get(eventType);
-      return friends.get(eventType);
+    if (existingFriends != null) {
+      return existingFriends;
     }
 
     final Collection<Class<?>> types = getEventTypes(eventType);

proxy/src/main/java/com/velocitypowered/proxy/event/VelocityEventManager.java

@@ -25,7 +25,6 @@ import com.google.common.base.VerifyException;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ListMultimap;
 import com.google.common.reflect.TypeToken;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.velocitypowered.api.event.Continuation;
 import com.velocitypowered.api.event.EventHandler;
 import com.velocitypowered.api.event.EventManager;
@@ -38,6 +37,7 @@ import com.velocitypowered.api.plugin.PluginManager;
 import com.velocitypowered.proxy.event.UntargetedEventHandler.EventTaskHandler;
 import com.velocitypowered.proxy.event.UntargetedEventHandler.VoidHandler;
 import com.velocitypowered.proxy.event.UntargetedEventHandler.WithContinuationHandler;
+import com.velocitypowered.proxy.util.collect.Enum2IntMap;
 import java.lang.invoke.MethodHandle;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.VarHandle;
@@ -55,9 +55,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiConsumer;
@@ -76,6 +73,14 @@ import org.lanternpowered.lmbda.LambdaType;
  */
 public class VelocityEventManager implements EventManager {
 
+  private static final Enum2IntMap<PostOrder> POST_ORDER_MAP = new Enum2IntMap.Builder<>(PostOrder.class)
+      .put(PostOrder.FIRST, Short.MAX_VALUE - 1)
+      .put(PostOrder.EARLY, Short.MAX_VALUE / 2)
+      .put(PostOrder.NORMAL, 0)
+      .put(PostOrder.LATE, Short.MIN_VALUE / 2)
+      .put(PostOrder.LAST, Short.MIN_VALUE + 1)
+      .put(PostOrder.CUSTOM, 0)
+      .build();
   private static final Logger logger = LogManager.getLogger(VelocityEventManager.class);
 
   private static final MethodHandles.Lookup methodHandlesLookup = MethodHandles.lookup();
@@ -87,9 +92,8 @@ public class VelocityEventManager implements EventManager {
       LambdaType.of(WithContinuationHandler.class);
 
   private static final Comparator<HandlerRegistration> handlerComparator =
-      Comparator.comparingInt(o -> o.order);
+      Collections.reverseOrder(Comparator.comparingInt(o -> o.order));
 
-  private final ExecutorService asyncExecutor;
   private final PluginManager pluginManager;
 
   private final ListMultimap<Class<?>, HandlerRegistration> handlersByType =
@@ -112,9 +116,6 @@ public class VelocityEventManager implements EventManager {
    */
   public VelocityEventManager(final PluginManager pluginManager) {
     this.pluginManager = pluginManager;
-    this.asyncExecutor = Executors
-        .newFixedThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactoryBuilder()
-            .setNameFormat("Velocity Async Event Executor - #%d").setDaemon(true).build());
   }
 
   /**
@@ -140,6 +141,7 @@ public class VelocityEventManager implements EventManager {
     final short order;
     final Class<?> eventType;
     final EventHandler<Object> handler;
+    final AsyncType asyncType;
 
     /**
      * The instance of the {@link EventHandler} or the listener instance that was registered.
@@ -147,31 +149,40 @@ public class VelocityEventManager implements EventManager {
     final Object instance;
 
     public HandlerRegistration(final PluginContainer plugin, final short order,
-        final Class<?> eventType, final Object instance, final EventHandler<Object> handler) {
+        final Class<?> eventType, final Object instance, final EventHandler<Object> handler,
+        final AsyncType asyncType) {
       this.plugin = plugin;
       this.order = order;
       this.eventType = eventType;
       this.instance = instance;
       this.handler = handler;
+      this.asyncType = asyncType;
     }
   }
 
   enum AsyncType {
-    /**
-     * The complete event will be handled on an async thread.
-     */
-    ALWAYS,
     /**
      * The event will never run async, everything is handled on the netty thread.
      */
-    NEVER
+    NEVER,
+    /**
+     * The event will initially start on the thread calling the {@code fire} method, and possibly
+     * switch over to an async thread.
+     */
+    SOMETIMES,
+    /**
+     * The complete event will be handled on an async thread.
+     */
+    ALWAYS
   }
 
   static final class HandlersCache {
 
+    final AsyncType asyncType;
     final HandlerRegistration[] handlers;
 
-    HandlersCache(final HandlerRegistration[] handlers) {
+    HandlersCache(AsyncType asyncType, final HandlerRegistration[] handlers) {
+      this.asyncType = asyncType;
       this.handlers = handlers;
     }
   }
@@ -194,7 +205,15 @@ public class VelocityEventManager implements EventManager {
     }
 
     baked.sort(handlerComparator);
-    return new HandlersCache(baked.toArray(new HandlerRegistration[0]));
+
+    AsyncType asyncType = AsyncType.NEVER;
+    for (HandlerRegistration registration : baked) {
+      if (registration.asyncType.compareTo(asyncType) > 0) {
+        asyncType = registration.asyncType;
+      }
+    }
+
+    return new HandlersCache(asyncType, baked.toArray(new HandlerRegistration[0]));
   }
 
   /**
@@ -230,15 +249,17 @@ public class VelocityEventManager implements EventManager {
   static final class MethodHandlerInfo {
 
     final Method method;
+    final AsyncType asyncType;
     final @Nullable Class<?> eventType;
     final short order;
     final @Nullable String errors;
     final @Nullable Class<?> continuationType;
 
-    private MethodHandlerInfo(final Method method, final @Nullable Class<?> eventType,
+    private MethodHandlerInfo(final Method method, final AsyncType asyncType,
-        final short order, final @Nullable String errors,
+        final @Nullable Class<?> eventType, final short order, final @Nullable String errors,
         final @Nullable Class<?> continuationType) {
       this.method = method;
+      this.asyncType = asyncType;
       this.eventType = eventType;
       this.order = order;
       this.errors = errors;
@@ -302,17 +323,41 @@ public class VelocityEventManager implements EventManager {
           }
         }
       }
+      AsyncType asyncType = AsyncType.NEVER;
+      final Class<?> returnType = method.getReturnType();
       if (handlerAdapter == null) {
-        final Class<?> returnType = method.getReturnType();
         if (returnType != void.class && continuationType == Continuation.class) {
           errors.add("method return type must be void if a continuation parameter is provided");
         } else if (returnType != void.class && returnType != EventTask.class) {
-          errors.add("method return type must be void or EventTask");
+          errors.add("method return type must be void, AsyncTask, "
+              + "EventTask.Basic or EventTask.WithContinuation");
+        } else if (returnType == EventTask.class) {
+          // technically, for compatibility, we *should* assume that the method must be invoked
+          // async, however, from examining some publicly-available plugins, developers did
+          // generally follow the contract and returned an EventTask only if they wanted this
+          // behavior. enable it for them.
+          asyncType = AsyncType.SOMETIMES;
         }
+      } else {
+        // for custom handlers, we always expect a return type of EventTask. this feature appears
+        // to have not been used in the wild AFAIK, so it gets the new behavior by default
+        asyncType = AsyncType.SOMETIMES;
+      }
+
+      if (paramCount == 1 && returnType == void.class && subscribe.async()) {
+        // these are almost always a dead giveaway of a plugin that will need its handlers
+        // run async, so unless we're told otherwise, we'll assume that's the case
+        asyncType = AsyncType.ALWAYS;
+      }
+
+      final short order;
+      if (subscribe.order() == PostOrder.CUSTOM) {
+        order = subscribe.priority();
+      } else {
+        order = (short) POST_ORDER_MAP.get(subscribe.order());
       }
-      final short order = (short) subscribe.order().ordinal();
       final String errorsJoined = errors.isEmpty() ? null : String.join(",", errors);
-      collected.put(key, new MethodHandlerInfo(method, eventType, order, errorsJoined,
+      collected.put(key, new MethodHandlerInfo(method, asyncType, eventType, order, errorsJoined,
           continuationType));
     }
     final Class<?> superclass = targetClass.getSuperclass();
@@ -351,12 +396,29 @@ public class VelocityEventManager implements EventManager {
   @SuppressWarnings("unchecked")
   public <E> void register(final Object plugin, final Class<E> eventClass,
       final PostOrder order, final EventHandler<E> handler) {
+    if (order == PostOrder.CUSTOM) {
+      throw new IllegalArgumentException(
+          "This method does not support custom post orders. Use the overload with short instead."
+      );
+    }
+    register(plugin, eventClass, (short) POST_ORDER_MAP.get(order), handler, AsyncType.ALWAYS);
+  }
+
+  @Override
+  public <E> void register(Object plugin, Class<E> eventClass, short postOrder,
+      EventHandler<E> handler) {
+    register(plugin, eventClass, postOrder, handler, AsyncType.SOMETIMES);
+  }
+
+  private  <E> void register(Object plugin, Class<E> eventClass, short postOrder,
+      EventHandler<E> handler, AsyncType asyncType) {
     final PluginContainer pluginContainer = pluginManager.ensurePluginContainer(plugin);
     requireNonNull(eventClass, "eventClass");
     requireNonNull(handler, "handler");
 
     final HandlerRegistration registration = new HandlerRegistration(pluginContainer,
-        (short) order.ordinal(), eventClass, handler, (EventHandler<Object>) handler);
+        postOrder, eventClass, handler, (EventHandler<Object>) handler,
+        AsyncType.ALWAYS);
     register(Collections.singletonList(registration));
   }
 
@@ -386,7 +448,7 @@ public class VelocityEventManager implements EventManager {
 
       final EventHandler<Object> handler = untargetedHandler.buildHandler(listener);
       registrations.add(new HandlerRegistration(pluginContainer, info.order,
-          info.eventType, listener, handler));
+          info.eventType, listener, handler, info.asyncType));
     }
 
     register(registrations);
@@ -473,10 +535,13 @@ public class VelocityEventManager implements EventManager {
 
   private <E> void fire(final @Nullable CompletableFuture<E> future,
       final E event, final HandlersCache handlersCache) {
-    // In Velocity 1.1.0, all events were fired asynchronously. As Velocity 3.0.0 is intended to be
+    final HandlerRegistration registration = handlersCache.handlers[0];
-    // largely (albeit not 100%) compatible with 1.1.x, we also fire events async. This behavior
+    if (registration.asyncType == AsyncType.ALWAYS) {
-    // will go away in Velocity Polymer.
+      registration.plugin.getExecutorService().execute(
-    asyncExecutor.execute(() -> fire(future, event, 0, true, handlersCache.handlers));
+          () -> fire(future, event, 0, true, handlersCache.handlers));
+    } else {
+      fire(future, event, 0, false, handlersCache.handlers);
+    }
   }
 
   private static final int TASK_STATE_DEFAULT = 0;
@@ -505,6 +570,7 @@ public class VelocityEventManager implements EventManager {
     private final @Nullable CompletableFuture<E> future;
     private final boolean currentlyAsync;
     private final E event;
+    private final Thread firedOnThread;
 
     // This field is modified via a VarHandle, so this field is used and cannot be final.
     @SuppressWarnings({"UnusedVariable", "FieldMayBeFinal", "FieldCanBeLocal"})
@@ -527,6 +593,7 @@ public class VelocityEventManager implements EventManager {
       this.event = event;
       this.index = index;
       this.currentlyAsync = currentlyAsync;
+      this.firedOnThread = Thread.currentThread();
     }
 
     @Override
@@ -537,8 +604,8 @@ public class VelocityEventManager implements EventManager {
     }
 
     /**
-     * Executes the task and returns whether the next one should be executed immediately after this
+     * Executes the task and returns whether the next handler should be executed immediately
-     * one without scheduling.
+     * after this one, without additional scheduling.
      */
     boolean execute() {
       state = TASK_STATE_EXECUTING;
@@ -580,7 +647,18 @@ public class VelocityEventManager implements EventManager {
       }
       if (!CONTINUATION_TASK_STATE.compareAndSet(
           this, TASK_STATE_EXECUTING, TASK_STATE_CONTINUE_IMMEDIATELY)) {
-        asyncExecutor.execute(() -> fire(future, event, index + 1, true, registrations));
+        // We established earlier that registrations[index + 1] is a valid index.
+        // If we are remaining in the same thread for the next handler, fire
+        // the next event immediately, else fire it within the executor service
+        // of the plugin with the next handler.
+        final HandlerRegistration next = registrations[index + 1];
+        final Thread currentThread = Thread.currentThread();
+        if (currentThread == firedOnThread && next.asyncType != AsyncType.ALWAYS) {
+          fire(future, event, index + 1, currentlyAsync, registrations);
+        } else {
+          next.plugin.getExecutorService().execute(() ->
+              fire(future, event, index + 1, true, registrations));
+        }
       }
     }
 
@@ -606,7 +684,7 @@ public class VelocityEventManager implements EventManager {
             continue;
           }
         } else {
-          asyncExecutor.execute(continuationTask);
+          registration.plugin.getExecutorService().execute(continuationTask);
         }
         // fire will continue in another thread once the async task is
         // executed and the continuation is resumed
@@ -626,13 +704,4 @@ public class VelocityEventManager implements EventManager {
     logger.error("Couldn't pass {} to {} {}", registration.eventType.getSimpleName(),
             pluginDescription.getId(), pluginDescription.getVersion().orElse(""), t);
   }
-
-  public boolean shutdown() throws InterruptedException {
-    asyncExecutor.shutdown();
-    return asyncExecutor.awaitTermination(10, TimeUnit.SECONDS);
-  }
-
-  public ExecutorService getAsyncExecutor() {
-    return asyncExecutor;
-  }
 }

proxy/src/main/java/com/velocitypowered/proxy/network/BackendChannelInitializer.java

@@ -49,7 +49,7 @@ public class BackendChannelInitializer extends ChannelInitializer<Channel> {
   }
 
   @Override
-  protected void initChannel(Channel ch) throws Exception {
+  protected void initChannel(Channel ch) {
     ch.pipeline()
         .addLast(FRAME_DECODER, new MinecraftVarintFrameDecoder())
         .addLast(READ_TIMEOUT,

proxy/src/main/java/com/velocitypowered/proxy/plugin/VelocityPluginManager.java

@@ -68,7 +68,12 @@ public class VelocityPluginManager implements PluginManager {
     this.server = checkNotNull(server, "server");
   }
 
-  private void registerPlugin(PluginContainer plugin) {
+  /**
+   * Registers a plugin with the plugin manager.
+   *
+   * @param plugin the plugin to register
+   */
+  public void registerPlugin(PluginContainer plugin) {
     pluginsById.put(plugin.getDescription().getId(), plugin);
     Optional<?> instance = plugin.getInstance();
     instance.ifPresent(o -> pluginInstances.put(o, plugin));

proxy/src/main/java/com/velocitypowered/proxy/plugin/virtual/VelocityVirtualPlugin.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (C) 2024 Velocity 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 <https://www.gnu.org/licenses/>.
+ */
+
+package com.velocitypowered.proxy.plugin.virtual;
+
+/**
+ * A singleton plugin object that represents the Velocity proxy itself.
+ */
+public class VelocityVirtualPlugin {
+  @SuppressWarnings("InstantiationOfUtilityClass")
+  public static final VelocityVirtualPlugin INSTANCE = new VelocityVirtualPlugin();
+
+  private VelocityVirtualPlugin() {
+  }
+}

proxy/src/main/java/com/velocitypowered/proxy/protocol/ProtocolUtils.java

@@ -104,6 +104,7 @@ public enum ProtocolUtils {
           .build();
 
   public static final int DEFAULT_MAX_STRING_SIZE = 65536; // 64KiB
+  private static final int MAXIMUM_VARINT_SIZE = 5;
   private static final BinaryTagType<? extends BinaryTag>[] BINARY_TAG_TYPES = new BinaryTagType[] {
       BinaryTagTypes.END, BinaryTagTypes.BYTE, BinaryTagTypes.SHORT, BinaryTagTypes.INT,
       BinaryTagTypes.LONG, BinaryTagTypes.FLOAT, BinaryTagTypes.DOUBLE,
@@ -111,13 +112,18 @@ public enum ProtocolUtils {
       BinaryTagTypes.COMPOUND, BinaryTagTypes.INT_ARRAY, BinaryTagTypes.LONG_ARRAY};
   private static final QuietDecoderException BAD_VARINT_CACHED =
       new QuietDecoderException("Bad VarInt decoded");
-  private static final int[] VARINT_EXACT_BYTE_LENGTHS = new int[33];
+  private static final int[] VAR_INT_LENGTHS = new int[65];
 
   static {
     for (int i = 0; i <= 32; ++i) {
-      VARINT_EXACT_BYTE_LENGTHS[i] = (int) Math.ceil((31d - (i - 1)) / 7d);
+      VAR_INT_LENGTHS[i] = (int) Math.ceil((31d - (i - 1)) / 7d);
     }
-    VARINT_EXACT_BYTE_LENGTHS[32] = 1; // Special case for the number 0.
+    VAR_INT_LENGTHS[32] = 1; // Special case for the number 0.
+  }
+
+  private static DecoderException badVarint() {
+    return MinecraftDecoder.DEBUG ? new CorruptedFrameException("Bad VarInt decoded")
+        : BAD_VARINT_CACHED;
   }
 
   /**
@@ -127,56 +133,29 @@ public enum ProtocolUtils {
    * @return the decoded VarInt
    */
   public static int readVarInt(ByteBuf buf) {
-    int read = readVarIntSafely(buf);
+    int readable = buf.readableBytes();
-    if (read == Integer.MIN_VALUE) {
+    if (readable == 0) {
-      throw MinecraftDecoder.DEBUG ? new CorruptedFrameException("Bad VarInt decoded")
+      // special case for empty buffer
-          : BAD_VARINT_CACHED;
+      throw badVarint();
     }
-    return read;
-  }
 
-  /**
+    // we can read at least one byte, and this should be a common case
-   * Reads a Minecraft-style VarInt from the specified {@code buf}. The difference between this
+    int k = buf.readByte();
-   * method and {@link #readVarInt(ByteBuf)} is that this function returns a sentinel value if the
+    if ((k & 0x80) != 128) {
-   * varint is invalid.
+      return k;
-   *
+    }
-   * @param buf the buffer to read from
+
-   * @return the decoded VarInt, or {@code Integer.MIN_VALUE} if the varint is invalid
+    // in case decoding one byte was not enough, use a loop to decode up to the next 4 bytes
-   */
+    int maxRead = Math.min(MAXIMUM_VARINT_SIZE, readable);
-  public static int readVarIntSafely(ByteBuf buf) {
+    int i = k & 0x7F;
-    int i = 0;
+    for (int j = 1; j < maxRead; j++) {
-    int maxRead = Math.min(5, buf.readableBytes());
+      k = buf.readByte();
-    for (int j = 0; j < maxRead; j++) {
-      int k = buf.readByte();
       i |= (k & 0x7F) << j * 7;
       if ((k & 0x80) != 128) {
         return i;
       }
     }
-    return Integer.MIN_VALUE;
+    throw badVarint();
-  }
-
-  /**
-   * Reads a Minecraft-style VarInt from the specified {@code buf}. The difference between this
-   * method and {@link #readVarInt(ByteBuf)} is that this function returns a sentinel value if the
-   * varint is invalid.
-   *
-   * @param buf the buffer to read from
-   * @return the decoded VarInt
-   * @throws DecoderException if the varint is invalid
-   */
-  public static int readVarIntSafelyOrThrow(ByteBuf buf) {
-    int i = 0;
-    int maxRead = Math.min(5, buf.readableBytes());
-    for (int j = 0; j < maxRead; j++) {
-      int k = buf.readByte();
-      i |= (k & 0x7F) << j * 7;
-      if ((k & 0x80) != 128) {
-        return i;
-      }
-    }
-    throw MinecraftDecoder.DEBUG ? new CorruptedFrameException("Bad VarInt decoded")
-        : BAD_VARINT_CACHED;
   }
 
   /**
@@ -186,7 +165,7 @@ public enum ProtocolUtils {
    * @return the byte size of {@code value} if encoded as a VarInt
    */
   public static int varIntBytes(int value) {
-    return VARINT_EXACT_BYTE_LENGTHS[Integer.numberOfLeadingZeros(value)];
+    return VAR_INT_LENGTHS[Integer.numberOfLeadingZeros(value)];
   }
 
   /**
@@ -210,6 +189,8 @@ public enum ProtocolUtils {
 
   private static void writeVarIntFull(ByteBuf buf, int value) {
     // See https://steinborn.me/posts/performance/how-fast-can-you-write-a-varint/
+
+    // This essentially is an unrolled version of the "traditional" VarInt encoding.
     if ((value & (0xFFFFFFFF << 7)) == 0) {
       buf.writeByte(value);
     } else if ((value & (0xFFFFFFFF << 14)) == 0) {

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

@@ -17,7 +17,8 @@
 
 package com.velocitypowered.proxy.protocol.netty;
 
-import com.velocitypowered.proxy.protocol.netty.VarintByteDecoder.DecodeResult;
+import static io.netty.util.ByteProcessor.FIND_NON_NUL;
+
 import com.velocitypowered.proxy.util.except.QuietDecoderException;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -29,53 +30,114 @@ import java.util.List;
  */
 public class MinecraftVarintFrameDecoder extends ByteToMessageDecoder {
 
-  private static final QuietDecoderException BAD_LENGTH_CACHED =
+  private static final QuietDecoderException BAD_PACKET_LENGTH =
       new QuietDecoderException("Bad packet length");
-  private static final QuietDecoderException VARINT_BIG_CACHED =
+  private static final QuietDecoderException VARINT_TOO_BIG =
       new QuietDecoderException("VarInt too big");
 
   @Override
-  protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+  protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
+      throws Exception {
     if (!ctx.channel().isActive()) {
       in.clear();
       return;
     }
 
-    final VarintByteDecoder reader = new VarintByteDecoder();
+    // skip any runs of 0x00 we might find
-
+    int packetStart = in.forEachByte(FIND_NON_NUL);
-    int varintEnd = in.forEachByte(reader);
+    if (packetStart == -1) {
-    if (varintEnd == -1) {
-      // We tried to go beyond the end of the buffer. This is probably a good sign that the
-      // buffer was too short to hold a proper varint.
-      if (reader.getResult() == DecodeResult.RUN_OF_ZEROES) {
-        // Special case where the entire packet is just a run of zeroes. We ignore them all.
-        in.clear();
-      }
       return;
     }
+    in.readerIndex(packetStart);
 
-    if (reader.getResult() == DecodeResult.RUN_OF_ZEROES) {
+    // try to read the length of the packet
-      // this will return to the point where the next varint starts
+    in.markReaderIndex();
-      in.readerIndex(varintEnd);
+    int preIndex = in.readerIndex();
-    } else if (reader.getResult() == DecodeResult.SUCCESS) {
+    int length = readRawVarInt21(in);
-      int readVarint = reader.getReadVarint();
+    if (preIndex == in.readerIndex()) {
-      int bytesRead = reader.getBytesRead();
+      return;
-      if (readVarint < 0) {
+    }
-        in.clear();
+    if (length < 0) {
-        throw BAD_LENGTH_CACHED;
+      throw BAD_PACKET_LENGTH;
-      } else if (readVarint == 0) {
+    }
-        // skip over the empty packet(s) and ignore it
+
-        in.readerIndex(varintEnd + 1);
+    // note that zero-length packets are ignored
+    if (length > 0) {
+      if (in.readableBytes() < length) {
+        in.resetReaderIndex();
       } else {
-        int minimumRead = bytesRead + readVarint;
+        out.add(in.readRetainedSlice(length));
-        if (in.isReadable(minimumRead)) {
-          out.add(in.retainedSlice(varintEnd + 1, readVarint));
-          in.skipBytes(minimumRead);
-        }
       }
-    } else if (reader.getResult() == DecodeResult.TOO_BIG) {
-      in.clear();
-      throw VARINT_BIG_CACHED;
     }
   }
+
+  /**
+   * Reads a VarInt from the buffer of up to 21 bits in size.
+   *
+   * @param buffer the buffer to read from
+   * @return the VarInt decoded, {@code 0} if no varint could be read
+   * @throws QuietDecoderException if the VarInt is too big to be decoded
+   */
+  private static int readRawVarInt21(ByteBuf buffer) {
+    if (buffer.readableBytes() < 4) {
+      // we don't have enough that we can read a potentially full varint, so fall back to
+      // the slow path.
+      return readRawVarintSmallBuf(buffer);
+    }
+    int wholeOrMore = buffer.getIntLE(buffer.readerIndex());
+
+    // take the last three bytes and check if any of them have the high bit set
+    int atStop = ~wholeOrMore & 0x808080;
+    if (atStop == 0) {
+      // all bytes have the high bit set, so the varint we are trying to decode is too wide
+      throw VARINT_TOO_BIG;
+    }
+
+    int bitsToKeep = Integer.numberOfTrailingZeros(atStop) + 1;
+    buffer.skipBytes(bitsToKeep >> 3);
+
+    // remove all bits we don't need to keep, a trick from
+    // https://github.com/netty/netty/pull/14050#issuecomment-2107750734:
+    //
+    // > The idea is that thisVarintMask has 0s above the first one of firstOneOnStop, and 1s at
+    // > and below it. For example if firstOneOnStop is 0x800080 (where the last 0x80 is the only
+    // > one that matters), then thisVarintMask is 0xFF.
+    //
+    // this is also documented in Hacker's Delight, section 2-1 "Manipulating Rightmost Bits"
+    int preservedBytes = wholeOrMore & (atStop ^ (atStop - 1));
+
+    // merge together using this trick: https://github.com/netty/netty/pull/14050#discussion_r1597896639
+    preservedBytes = (preservedBytes & 0x007F007F) | ((preservedBytes & 0x00007F00) >> 1);
+    preservedBytes = (preservedBytes & 0x00003FFF) | ((preservedBytes & 0x3FFF0000) >> 2);
+    return preservedBytes;
+  }
+
+  private static int readRawVarintSmallBuf(ByteBuf buffer) {
+    if (!buffer.isReadable()) {
+      return 0;
+    }
+    buffer.markReaderIndex();
+
+    byte tmp = buffer.readByte();
+    if (tmp >= 0) {
+      return tmp;
+    }
+    int result = tmp & 0x7F;
+    if (!buffer.isReadable()) {
+      buffer.resetReaderIndex();
+      return 0;
+    }
+    if ((tmp = buffer.readByte()) >= 0) {
+      return result | tmp << 7;
+    }
+    result |= (tmp & 0x7F) << 7;
+    if (!buffer.isReadable()) {
+      buffer.resetReaderIndex();
+      return 0;
+    }
+    if ((tmp = buffer.readByte()) >= 0) {
+      return result | tmp << 14;
+    }
+    return result | (tmp & 0x7F) << 14;
+  }
 }

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

@@ -23,33 +23,34 @@ import com.velocitypowered.proxy.protocol.ProtocolUtils;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.MessageToByteEncoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import java.util.List;
 
 /**
  * Handler for appending a length for Minecraft packets.
  */
 @ChannelHandler.Sharable
-public class MinecraftVarintLengthEncoder extends MessageToByteEncoder<ByteBuf> {
+public class MinecraftVarintLengthEncoder extends MessageToMessageEncoder<ByteBuf> {
 
   public static final MinecraftVarintLengthEncoder INSTANCE = new MinecraftVarintLengthEncoder();
-  public static final boolean IS_JAVA_CIPHER = Natives.cipher.get() == JavaVelocityCipher.FACTORY;
+
+  static final boolean IS_JAVA_CIPHER = Natives.cipher.get() == JavaVelocityCipher.FACTORY;
 
   private MinecraftVarintLengthEncoder() {
   }
 
   @Override
-  protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
+  protected void encode(ChannelHandlerContext ctx, ByteBuf buf,
-    ProtocolUtils.writeVarInt(out, msg.readableBytes());
+      List<Object> list) throws Exception {
-    out.writeBytes(msg);
+    final int length = buf.readableBytes();
-  }
+    final int varintLength = ProtocolUtils.varIntBytes(length);
 
-  @Override
+    final ByteBuf lenBuf = IS_JAVA_CIPHER
-  protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, ByteBuf msg, boolean preferDirect)
+        ? ctx.alloc().heapBuffer(varintLength)
-      throws Exception {
+        : ctx.alloc().directBuffer(varintLength);
-    int anticipatedRequiredCapacity = ProtocolUtils.varIntBytes(msg.readableBytes())
+
-        + msg.readableBytes();
+    ProtocolUtils.writeVarInt(lenBuf, length);
-    return IS_JAVA_CIPHER
+    list.add(lenBuf);
-        ? ctx.alloc().heapBuffer(anticipatedRequiredCapacity)
+    list.add(buf.retain());
-        : ctx.alloc().directBuffer(anticipatedRequiredCapacity);
   }
 }

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

@@ -1,68 +0,0 @@
-/*
- * Copyright (C) 2020-2021 Velocity 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 <https://www.gnu.org/licenses/>.
- */
-
-package com.velocitypowered.proxy.protocol.netty;
-
-import io.netty.util.ByteProcessor;
-
-class VarintByteDecoder implements ByteProcessor {
-
-  private int readVarint;
-  private int bytesRead;
-  private DecodeResult result = DecodeResult.TOO_SHORT;
-
-  @Override
-  public boolean process(byte k) {
-    if (k == 0 && bytesRead == 0) {
-      // tentatively say it's invalid, but there's a possibility of redemption
-      result = DecodeResult.RUN_OF_ZEROES;
-      return true;
-    }
-    if (result == DecodeResult.RUN_OF_ZEROES) {
-      return false;
-    }
-    readVarint |= (k & 0x7F) << bytesRead++ * 7;
-    if (bytesRead > 3) {
-      result = DecodeResult.TOO_BIG;
-      return false;
-    }
-    if ((k & 0x80) != 128) {
-      result = DecodeResult.SUCCESS;
-      return false;
-    }
-    return true;
-  }
-
-  public int getReadVarint() {
-    return readVarint;
-  }
-
-  public int getBytesRead() {
-    return bytesRead;
-  }
-
-  public DecodeResult getResult() {
-    return result;
-  }
-
-  public enum DecodeResult {
-    SUCCESS,
-    TOO_SHORT,
-    TOO_BIG,
-    RUN_OF_ZEROES
-  }
-}

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

@@ -63,7 +63,7 @@ public class LoginPluginMessagePacket extends DeferredByteBufHolder implements M
 
   @Override
   public void decode(ByteBuf buf, ProtocolUtils.Direction direction, ProtocolVersion version) {
-    this.id = ProtocolUtils.readVarIntSafelyOrThrow(buf);
+    this.id = ProtocolUtils.readVarInt(buf);
     this.channel = ProtocolUtils.readString(buf);
     if (buf.isReadable()) {
       this.replace(buf.readRetainedSlice(buf.readableBytes()));

proxy/src/main/java/com/velocitypowered/proxy/protocol/packet/chat/ComponentHolder.java

@@ -173,21 +173,21 @@ public class ComponentHolder {
         case 1://BinaryTagTypes.BYTE:
           byte[] bytes = new byte[jsonArray.size()];
           for (int i = 0; i < bytes.length; i++) {
-            bytes[i] = (Byte) jsonArray.get(i).getAsNumber();
+            bytes[i] = jsonArray.get(i).getAsNumber().byteValue();
           }
 
           return ByteArrayBinaryTag.byteArrayBinaryTag(bytes);
         case 3://BinaryTagTypes.INT:
           int[] ints = new int[jsonArray.size()];
           for (int i = 0; i < ints.length; i++) {
-            ints[i] = (Integer) jsonArray.get(i).getAsNumber();
+            ints[i] = jsonArray.get(i).getAsNumber().intValue();
           }
 
           return IntArrayBinaryTag.intArrayBinaryTag(ints);
         case 4://BinaryTagTypes.LONG:
           long[] longs = new long[jsonArray.size()];
           for (int i = 0; i < longs.length; i++) {
-            longs[i] = (Long) jsonArray.get(i).getAsNumber();
+            longs[i] = jsonArray.get(i).getAsNumber().longValue();
           }
 
           return LongArrayBinaryTag.longArrayBinaryTag(longs);

proxy/src/main/java/com/velocitypowered/proxy/protocol/packet/chat/session/SessionChatHandler.java

@@ -71,7 +71,8 @@ public class SessionChatHandler implements ChatHandler<SessionPlayerChatPacket>
                   invalidChange(logger, player);
                   return null;
                 }
-                return this.player.getChatBuilderFactory().builder().message(packet.message)
+                return this.player.getChatBuilderFactory().builder()
+                    .message(chatResult.getMessage().orElse(packet.getMessage()))
                     .setTimestamp(packet.timestamp)
                     .setLastSeenMessages(newLastSeenMessages)
                     .toServer();

proxy/src/test/java/com/velocitypowered/proxy/command/BrigadierCommandTests.java

@@ -230,7 +230,7 @@ public class BrigadierCommandTests extends CommandTestSuite {
     final Exception wrapper = assertThrows(CompletionException.class, () ->
         manager.executeAsync(source, "hello").join());
 
-    assertSame(expected, wrapper.getCause().getCause());
+    assertSame(expected, wrapper.getCause());
   }
 
   // Suggestions

proxy/src/test/java/com/velocitypowered/proxy/command/CommandTestSuite.java

@@ -29,9 +29,9 @@ import com.velocitypowered.api.permission.Tristate;
 import com.velocitypowered.api.proxy.Player;
 import com.velocitypowered.proxy.event.MockEventManager;
 import com.velocitypowered.proxy.event.VelocityEventManager;
+import com.velocitypowered.proxy.testutil.FakePluginManager;
 import java.util.Arrays;
 import java.util.Collection;
-import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 
@@ -47,19 +47,9 @@ abstract class CommandTestSuite {
     eventManager = new MockEventManager();
   }
 
-  @AfterAll
-  static void afterAll() {
-    try {
-      eventManager.shutdown();
-      eventManager = null;
-    } catch (final InterruptedException e) {
-      Thread.currentThread().interrupt();
-    }
-  }
-
   @BeforeEach
   void setUp() {
-    this.manager = new VelocityCommandManager(eventManager);
+    this.manager = new VelocityCommandManager(eventManager, new FakePluginManager());
   }
 
   final void assertHandled(final String input) {

proxy/src/test/java/com/velocitypowered/proxy/event/EventTest.java

@@ -41,20 +41,24 @@ import org.junit.jupiter.api.TestInstance;
 public class EventTest {
 
   public static final String CONTINUATION_TEST_THREAD_NAME = "Continuation test thread";
-  private final VelocityEventManager eventManager =
+  private final FakePluginManager pluginManager = new FakePluginManager();
-      new VelocityEventManager(new FakePluginManager());
+  private final VelocityEventManager eventManager = new VelocityEventManager(pluginManager);
 
   @AfterAll
   void shutdown() throws Exception {
-    eventManager.shutdown();
+    pluginManager.shutdown();
   }
 
   static final class TestEvent {
 
   }
 
+  static void assertSyncThread(final Thread thread) {
+    assertEquals(Thread.currentThread(), thread);
+  }
+
   static void assertAsyncThread(final Thread thread) {
-    assertTrue(thread.getName().contains("Velocity Async Event Executor"));
+    assertTrue(thread.getName().contains("Test Async Thread"));
   }
 
   static void assertContinuationThread(final Thread thread) {
@@ -90,6 +94,7 @@ public class EventTest {
       eventManager.fire(new TestEvent()).get();
     } finally {
       eventManager.unregisterListeners(FakePluginManager.PLUGIN_A);
+      eventManager.unregisterListeners(FakePluginManager.PLUGIN_B);
     }
 
     // Check that the order is A < B < C.
@@ -119,6 +124,7 @@ public class EventTest {
       eventManager.fire(new TestEvent()).get();
     } finally {
       eventManager.unregisterListeners(FakePluginManager.PLUGIN_A);
+      eventManager.unregisterListeners(FakePluginManager.PLUGIN_B);
     }
 
     // Check that the order is A < B < C.
@@ -126,6 +132,26 @@ public class EventTest {
     assertTrue(listener2Invoked.get() < listener3Invoked.get(), "Listener C invoked before B!");
   }
 
+  @Test
+  void testAlwaysSync() throws Exception {
+    final AlwaysSyncListener listener = new AlwaysSyncListener();
+    handleMethodListener(listener);
+    assertSyncThread(listener.thread);
+    assertEquals(1, listener.result);
+  }
+
+  static final class AlwaysSyncListener {
+
+    @MonotonicNonNull Thread thread;
+    int result;
+
+    @Subscribe(async = false)
+    void sync(TestEvent event) {
+      result++;
+      thread = Thread.currentThread();
+    }
+  }
+
   @Test
   void testAlwaysAsync() throws Exception {
     final AlwaysAsyncListener listener = new AlwaysAsyncListener();
@@ -143,7 +169,7 @@ public class EventTest {
     @MonotonicNonNull Thread threadC;
     int result;
 
-    @Subscribe
+    @Subscribe(async = true, order = PostOrder.EARLY)
     void firstAsync(TestEvent event) {
       result++;
       threadA = Thread.currentThread();
@@ -155,50 +181,93 @@ public class EventTest {
       return EventTask.async(() -> result++);
     }
 
-    @Subscribe
+    @Subscribe(order = PostOrder.LATE)
     void thirdAsync(TestEvent event) {
       result++;
       threadC = Thread.currentThread();
     }
   }
 
+  @Test
+  void testSometimesAsync() throws Exception {
+    final SometimesAsyncListener listener = new SometimesAsyncListener();
+    handleMethodListener(listener);
+    assertSyncThread(listener.threadA);
+    assertSyncThread(listener.threadB);
+    assertAsyncThread(listener.threadC);
+    assertAsyncThread(listener.threadD);
+    assertEquals(3, listener.result);
+  }
+
+  static final class SometimesAsyncListener {
+
+    @MonotonicNonNull Thread threadA;
+    @MonotonicNonNull Thread threadB;
+    @MonotonicNonNull Thread threadC;
+    @MonotonicNonNull Thread threadD;
+    int result;
+
+    @Subscribe(order = PostOrder.EARLY, async = false)
+    void notAsync(TestEvent event) {
+      result++;
+      threadA = Thread.currentThread();
+    }
+
+    @Subscribe
+    EventTask notAsyncUntilTask(TestEvent event) {
+      threadB = Thread.currentThread();
+      return EventTask.async(() -> {
+        threadC = Thread.currentThread();
+        result++;
+      });
+    }
+
+    @Subscribe(order = PostOrder.LATE, async = false)
+    void stillAsyncAfterTask(TestEvent event) {
+      threadD = Thread.currentThread();
+      result++;
+    }
+  }
+
   @Test
   void testContinuation() throws Exception {
     final ContinuationListener listener = new ContinuationListener();
     handleMethodListener(listener);
-    assertAsyncThread(listener.thread1);
+    assertSyncThread(listener.threadA);
-    assertAsyncThread(listener.thread2);
+    assertSyncThread(listener.threadB);
-    assertContinuationThread(listener.thread2Custom);
+    assertAsyncThread(listener.threadC);
-    assertAsyncThread(listener.thread3);
     assertEquals(2, listener.value.get());
   }
 
   static final class ContinuationListener {
 
-    @MonotonicNonNull Thread thread1;
+    @MonotonicNonNull Thread threadA;
-    @MonotonicNonNull Thread thread2;
+    @MonotonicNonNull Thread threadB;
-    @MonotonicNonNull Thread thread2Custom;
+    @MonotonicNonNull Thread threadC;
-    @MonotonicNonNull Thread thread3;
 
     final AtomicInteger value = new AtomicInteger();
 
     @Subscribe(order = PostOrder.EARLY)
     EventTask continuation(TestEvent event) {
-      thread1 = Thread.currentThread();
+      threadA = Thread.currentThread();
       return EventTask.withContinuation(continuation -> {
         value.incrementAndGet();
-        thread2 = Thread.currentThread();
+        threadB = Thread.currentThread();
         new Thread(() -> {
-          thread2Custom = Thread.currentThread();
+          try {
+            Thread.sleep(100);
+          } catch (InterruptedException e) {
+            e.printStackTrace();
+          }
           value.incrementAndGet();
           continuation.resume();
-        }, CONTINUATION_TEST_THREAD_NAME).start();
+        }).start();
       });
     }
 
     @Subscribe(order = PostOrder.LATE)
     void afterContinuation(TestEvent event) {
-      thread3 = Thread.currentThread();
+      threadC = Thread.currentThread();
     }
   }
 
@@ -207,9 +276,9 @@ public class EventTest {
     final ResumeContinuationImmediatelyListener listener =
         new ResumeContinuationImmediatelyListener();
     handleMethodListener(listener);
-    assertAsyncThread(listener.threadA);
+    assertSyncThread(listener.threadA);
-    assertAsyncThread(listener.threadB);
+    assertSyncThread(listener.threadB);
-    assertAsyncThread(listener.threadC);
+    assertSyncThread(listener.threadC);
     assertEquals(2, listener.result);
   }
 
@@ -241,42 +310,44 @@ public class EventTest {
   void testContinuationParameter() throws Exception {
     final ContinuationParameterListener listener = new ContinuationParameterListener();
     handleMethodListener(listener);
-    assertAsyncThread(listener.thread1);
+    assertSyncThread(listener.threadA);
-    assertAsyncThread(listener.thread2);
+    assertSyncThread(listener.threadB);
-    assertContinuationThread(listener.thread2Custom);
+    assertAsyncThread(listener.threadC);
-    assertAsyncThread(listener.thread3);
     assertEquals(3, listener.result.get());
   }
 
   static final class ContinuationParameterListener {
 
-    @MonotonicNonNull Thread thread1;
+    @MonotonicNonNull Thread threadA;
-    @MonotonicNonNull Thread thread2;
+    @MonotonicNonNull Thread threadB;
-    @MonotonicNonNull Thread thread2Custom;
+    @MonotonicNonNull Thread threadC;
-    @MonotonicNonNull Thread thread3;
 
     final AtomicInteger result = new AtomicInteger();
 
     @Subscribe
     void resume(TestEvent event, Continuation continuation) {
-      thread1 = Thread.currentThread();
+      threadA = Thread.currentThread();
       result.incrementAndGet();
       continuation.resume();
     }
 
     @Subscribe(order = PostOrder.LATE)
     void resumeFromCustomThread(TestEvent event, Continuation continuation) {
-      thread2 = Thread.currentThread();
+      threadB = Thread.currentThread();
       new Thread(() -> {
-        thread2Custom = Thread.currentThread();
+        try {
+          Thread.sleep(100);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        }
         result.incrementAndGet();
         continuation.resume();
-      }, CONTINUATION_TEST_THREAD_NAME).start();
+      }).start();
     }
 
     @Subscribe(order = PostOrder.LAST)
     void afterCustomThread(TestEvent event, Continuation continuation) {
-      thread3 = Thread.currentThread();
+      threadC = Thread.currentThread();
       result.incrementAndGet();
       continuation.resume();
     }
@@ -328,8 +399,7 @@ public class EventTest {
                 + "the second is the fancy continuation");
           }
         },
-        new TypeToken<TriConsumer<Object, Object, FancyContinuation>>() {
+        new TypeToken<TriConsumer<Object, Object, FancyContinuation>>() {},
-        },
         invokeFunction -> (instance, event) ->
             EventTask.withContinuation(continuation ->
                 invokeFunction.accept(instance, event, new FancyContinuationImpl(continuation))
@@ -349,4 +419,4 @@ public class EventTest {
       continuation.resume();
     }
   }
-}
+}

proxy/src/test/java/com/velocitypowered/proxy/protocol/ProtocolUtilsTest.java

@@ -70,7 +70,7 @@ public class ProtocolUtilsTest {
   private void writeReadTestOld(ByteBuf buf, int test) {
     buf.clear();
     writeVarIntOld(buf, test);
-    assertEquals(test, ProtocolUtils.readVarIntSafely(buf));
+    assertEquals(test, ProtocolUtils.readVarInt(buf));
   }
 
   @Test
@@ -103,7 +103,7 @@ public class ProtocolUtilsTest {
           "Encoding of " + i + " was invalid");
 
       assertEquals(i, oldReadVarIntSafely(varintNew));
-      assertEquals(i, ProtocolUtils.readVarIntSafely(varintOld));
+      assertEquals(i, ProtocolUtils.readVarInt(varintOld));
 
       varintNew.clear();
       varintOld.clear();

proxy/src/test/java/com/velocitypowered/proxy/testutil/FakePluginManager.java

@@ -18,14 +18,16 @@
 package com.velocitypowered.proxy.testutil;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.velocitypowered.api.plugin.PluginContainer;
 import com.velocitypowered.api.plugin.PluginDescription;
 import com.velocitypowered.api.plugin.PluginManager;
+import com.velocitypowered.proxy.plugin.virtual.VelocityVirtualPlugin;
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.Executors;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 /**
@@ -36,15 +38,23 @@ public class FakePluginManager implements PluginManager {
   public static final Object PLUGIN_A = new Object();
   public static final Object PLUGIN_B = new Object();
 
-  private static final PluginContainer PC_A = new FakePluginContainer("a", PLUGIN_A);
+  private final PluginContainer containerA = new FakePluginContainer("a", PLUGIN_A);
-  private static final PluginContainer PC_B = new FakePluginContainer("b", PLUGIN_B);
+  private final PluginContainer containerB = new FakePluginContainer("b", PLUGIN_B);
+  private final PluginContainer containerVelocity = new FakePluginContainer("velocity",
+      VelocityVirtualPlugin.INSTANCE);
+
+  private ExecutorService service = Executors.newCachedThreadPool(
+      new ThreadFactoryBuilder().setNameFormat("Test Async Thread").setDaemon(true).build()
+  );
 
   @Override
   public @NonNull Optional<PluginContainer> fromInstance(@NonNull Object instance) {
     if (instance == PLUGIN_A) {
-      return Optional.of(PC_A);
+      return Optional.of(containerA);
     } else if (instance == PLUGIN_B) {
-      return Optional.of(PC_B);
+      return Optional.of(containerB);
+    } else if (instance == VelocityVirtualPlugin.INSTANCE) {
+      return Optional.of(containerVelocity);
     } else {
       return Optional.empty();
     }
@@ -54,9 +64,11 @@ public class FakePluginManager implements PluginManager {
   public @NonNull Optional<PluginContainer> getPlugin(@NonNull String id) {
     switch (id) {
       case "a":
-        return Optional.of(PC_A);
+        return Optional.of(containerA);
       case "b":
-        return Optional.of(PC_B);
+        return Optional.of(containerB);
+      case "velocity":
+        return Optional.of(containerVelocity);
       default:
         return Optional.empty();
     }
@@ -64,7 +76,7 @@ public class FakePluginManager implements PluginManager {
 
   @Override
   public @NonNull Collection<PluginContainer> getPlugins() {
-    return ImmutableList.of(PC_A, PC_B);
+    return ImmutableList.of(containerVelocity, containerA, containerB);
   }
 
   @Override
@@ -77,16 +89,18 @@ public class FakePluginManager implements PluginManager {
     throw new UnsupportedOperationException();
   }
 
-  private static class FakePluginContainer implements PluginContainer {
+  public void shutdown() {
+    this.service.shutdownNow();
+  }
+
+  private class FakePluginContainer implements PluginContainer {
 
     private final String id;
     private final Object instance;
-    private final ExecutorService service;
 
     private FakePluginContainer(String id, Object instance) {
       this.id = id;
       this.instance = instance;
-      this.service = ForkJoinPool.commonPool();
     }
 
     @Override