fix(porticus): 修复发送线程、超时终态与分包完整性

module/minecraft/minecraft-porticus/build.gradle.kts

@@ -5,4 +5,8 @@ dependencies {
     compileOnly(project(":common-util"))
     compileOnly(project(":common-env"))
     compileOnly(project(":common-platform-api"))
+    testImplementation(project(":common"))
+    testImplementation(project(":common-util"))
+    testImplementation(project(":common-platform-api"))
+    testImplementation("ink.ptms.core:v12004:12004-minimize:mapped")
 }

module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/PorticusMission.java

@@ -6,6 +6,7 @@
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import java.util.function.LongSupplier;
 
 /**
  * Porticus
@@ -24,7 +25,9 @@ public abstract class PorticusMission {
     protected Runnable runnable;
     protected String[] command;
     protected long timeout = TimeUnit.SECONDS.toMillis(10);
-    private long start;
+    private volatile long start;
+    private volatile boolean started;
+    LongSupplier timeSource = System::currentTimeMillis;
 
     public PorticusMission() {
         this(UUID.randomUUID());
@@ -38,19 +41,35 @@ public PorticusMission(UUID uid) {
      * 通讯任务是否超时
      */
     public boolean isTimeout() {
-        return start + timeout < System.currentTimeMillis();
+        long startedAt = start;
+        return started && timeSource.getAsLong() - startedAt >= timeout;
     }
 
     /**
      * 运行通讯任务
      *
      * @param target 发送目标，根据服务端类型传入对应玩家对象，当 API 类型为 SERVER 时传入 ProxyPlayer 类型，为 CLIENT 时则传入 Player 类型。
      */
-    public void run(@NotNull Object target) {
-        if (consumer != null || runnable != null) {
-            Porticus.INSTANCE.getMissions().add(this);
+    public synchronized void run(@NotNull Object target) {
+        if (started) {
+            throw new IllegalStateException("Porticus missions can only be run once");
+        }
+        boolean trackCompletion = consumer != null || runnable != null;
+        if (trackCompletion) {
+            synchronized (Porticus.INSTANCE.getMissions()) {
+                for (PorticusMission mission : Porticus.INSTANCE.getMissions()) {
+                    if (mission.getUID().equals(uid)) {
+                        throw new IllegalStateException("A Porticus mission with the same UID is already pending");
+                    }
+                }
+                this.start = timeSource.getAsLong();
+                this.started = true;
+                Porticus.INSTANCE.getMissions().add(this);
+            }
+        } else {
+            this.start = timeSource.getAsLong();
+            this.started = true;
         }
-        this.start = System.currentTimeMillis();
     }
 
     /**

module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/MissionBukkit.java

@@ -10,7 +10,10 @@
 import taboolib.module.porticus.common.MessageBuilder;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.List;
 import java.util.UUID;
+import java.util.function.Consumer;
 
 /**
  * Porticus
@@ -32,23 +35,97 @@ public MissionBukkit(UUID uid) {
 
     @Override
     public void run(@NotNull Object target) {
-        super.run(target);
-        if (target instanceof Player) {
-            sendBukkitMessage((Player) target, command);
-        } else {
+        if (!(target instanceof Player)) {
             throw new IllegalStateException("target must be Player");
         }
+        if (command == null) {
+            throw new IllegalStateException("command must be set before running mission");
+        }
+        List<byte[]> messages;
+        try {
+            messages = MessageBuilder.create(command);
+        } catch (IOException e) {
+            throw new IllegalStateException("failed to encode mission command", e);
+        }
+        boolean tracked = consumer != null || runnable != null;
+        super.run(target);
+        try {
+            scheduleBukkitMessage((Player) target, messages, tracked);
+        } catch (Throwable t) {
+            Porticus.INSTANCE.getMissions().remove(this);
+            throw new IllegalStateException("failed to schedule mission message", t);
+        }
     }
 
     public void sendBukkitMessage(Player player, String[] command) {
-        Bukkit.getScheduler().runTaskAsynchronously(plugin, () -> {
+        try {
+            if (player == null) {
+                throw new IllegalArgumentException("player cannot be null");
+            }
+            scheduleBukkitMessage(player, MessageBuilder.create(command), false);
+        } catch (Throwable t) {
+            t.printStackTrace();
+        }
+    }
+
+    private void scheduleBukkitMessage(Player player, List<byte[]> messages, boolean tracked) throws Exception {
+        Runnable failure = tracked ? () -> Porticus.INSTANCE.getMissions().remove(this) : () -> {
+        };
+        Runnable sendTask = () -> {
+            if (tracked && !Porticus.INSTANCE.getMissions().contains(this)) {
+                return;
+            }
             try {
-                for (byte[] bytes : MessageBuilder.create(command)) {
+                for (byte[] bytes : messages) {
+                    if (tracked && !Porticus.INSTANCE.getMissions().contains(this)) {
+                        return;
+                    }
                     player.sendPluginMessage(plugin, Porticus.INSTANCE.getChannelId(), bytes);
                 }
-            } catch (IOException e) {
-                e.printStackTrace();
+            } catch (Throwable t) {
+                failure.run();
+                t.printStackTrace();
+            }
+        };
+        if (isFolia()) {
+            runOnEntityScheduler(player, sendTask, failure);
+        } else if (Bukkit.isPrimaryThread()) {
+            sendTask.run();
+        } else {
+            Bukkit.getScheduler().runTask(plugin, sendTask);
+        }
+    }
+
+    private static boolean isFolia() {
+        try {
+            Class.forName("io.papermc.paper.threadedregions.RegionizedServer", false, playerClassLoader());
+            return true;
+        } catch (Throwable ignored) {
+            return false;
+        }
+    }
+
+    private static ClassLoader playerClassLoader() {
+        ClassLoader classLoader = Player.class.getClassLoader();
+        return classLoader == null ? ClassLoader.getSystemClassLoader() : classLoader;
+    }
+
+    private void runOnEntityScheduler(Player player, Runnable sendTask, Runnable retired) throws Exception {
+        Object scheduler = player.getClass().getMethod("getScheduler").invoke(player);
+        Method runMethod = null;
+        for (Method method : scheduler.getClass().getMethods()) {
+            if (method.getName().equals("run") && method.getParameterTypes().length == 3) {
+                runMethod = method;
+                break;
             }
-        });
+        }
+        if (runMethod == null) {
+            throw new NoSuchMethodException("EntityScheduler#run");
+        }
+        Consumer<Object> task = ignored -> sendTask.run();
+        Object scheduled = runMethod.invoke(scheduler, plugin, task, retired);
+        if (scheduled == null) {
+            throw new IllegalStateException("EntityScheduler rejected Porticus message task");
+        }
     }
 }

module/minecraft/minecraft-porticus/src/main/java/taboolib/module/porticus/bukkitside/PorticusListener.java

@@ -15,6 +15,9 @@
 import taboolib.module.porticus.common.MessageReader;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 
 /**
  * @author 坏黑
@@ -23,39 +26,47 @@
 @SuppressWarnings("DuplicatedCode")
 public class PorticusListener implements Listener, PluginMessageListener {
 
+    private final Plugin plugin;
+    private final AtomicLong nextCacheWarning = new AtomicLong();
+
     public PorticusListener() {
-        Plugin plugin = JavaPlugin.getProvidingPlugin(Porticus.class);
+        plugin = JavaPlugin.getProvidingPlugin(Porticus.class);
         Bukkit.getPluginManager().registerEvents(this, plugin);
         Bukkit.getMessenger().registerIncomingPluginChannel(plugin, Porticus.INSTANCE.getChannelId(), this);
         Bukkit.getMessenger().registerOutgoingPluginChannel(plugin, Porticus.INSTANCE.getChannelId());
-        Bukkit.getScheduler().runTaskTimer(plugin, () -> {
+        Runnable timeoutTask = () -> {
             for (PorticusMission mission : Porticus.INSTANCE.getMissions()) {
-                if (mission.isTimeout()) {
+                if (mission.isTimeout() && Porticus.INSTANCE.getMissions().remove(mission)) {
                     if (mission.getTimeoutRunnable() != null) {
                         try {
                             mission.getTimeoutRunnable().run();
                         } catch (Throwable t) {
                             t.printStackTrace();
                         }
                     }
-                    Porticus.INSTANCE.getMissions().remove(mission);
                 }
             }
-        }, 0, 20);
+            MessageReader.cleanUp();
+        };
+        if (isFolia()) {
+            runGlobalTimer(plugin, timeoutTask);
+        } else {
+            Bukkit.getScheduler().runTaskTimer(plugin, timeoutTask, 0, 20);
+        }
     }
 
     @EventHandler(priority = EventPriority.MONITOR, ignoreCancelled = true)
     public void e(PorticusBukkitEvent e) {
         for (PorticusMission mission : Porticus.INSTANCE.getMissions()) {
-            if (mission.getUID().equals(e.getUID())) {
+            if (mission.getUID().equals(e.getUID()) && Porticus.INSTANCE.getMissions().remove(mission)) {
                 if (mission.getResponseConsumer() != null) {
                     try {
                         mission.getResponseConsumer().accept(e.getArgs());
                     } catch (Throwable t) {
                         t.printStackTrace();
                     }
                 }
-                Porticus.INSTANCE.getMissions().remove(mission);
+                break;
             }
         }
     }
@@ -66,11 +77,57 @@ public void onPluginMessageReceived(@NotNull String channel, @NotNull Player pla
             try {
                 Message message = MessageReader.read(bytes);
                 if (message.isCompleted()) {
-                    PorticusBukkitEvent.call(player, message.getMessages().get(0).getUID(), message.build());
+                    String[] args = message.buildOnce();
+                    if (args != null) {
+                        PorticusBukkitEvent.call(player, message.getUID(), args);
+                    }
                 }
+            } catch (MessageReader.ProtocolException ignored) {
+                // Malformed or oversized plugin messages are rejected without flooding the server log.
+            } catch (MessageReader.CapacityException ex) {
+                warnCacheCapacity(ex);
             } catch (IOException ex) {
                 ex.printStackTrace();
+            } catch (Throwable t) {
+                t.printStackTrace();
+            }
+        }
+    }
+
+    private void warnCacheCapacity(IOException exception) {
+        long now = System.currentTimeMillis();
+        long next = nextCacheWarning.get();
+        if (now >= next && nextCacheWarning.compareAndSet(next, now + 10_000)) {
+            plugin.getLogger().warning("Porticus message cache rejected input: " + exception.getMessage());
+        }
+    }
+
+    private static boolean isFolia() {
+        try {
+            Class.forName("io.papermc.paper.threadedregions.RegionizedServer");
+            return true;
+        } catch (Throwable ignored) {
+            return false;
+        }
+    }
+
+    private static void runGlobalTimer(Plugin plugin, Runnable runnable) {
+        try {
+            Object scheduler = Bukkit.class.getMethod("getGlobalRegionScheduler").invoke(null);
+            Method runAtFixedRate = null;
+            for (Method method : scheduler.getClass().getMethods()) {
+                if (method.getName().equals("runAtFixedRate") && method.getParameterTypes().length == 4) {
+                    runAtFixedRate = method;
+                    break;
+                }
+            }
+            if (runAtFixedRate == null) {
+                throw new NoSuchMethodException("GlobalRegionScheduler#runAtFixedRate");
             }
+            Consumer<Object> task = ignored -> runnable.run();
+            runAtFixedRate.invoke(scheduler, plugin, task, 1L, 20L);
+        } catch (Throwable t) {
+            throw new IllegalStateException("Unable to schedule Porticus timeout task on Folia", t);
         }
     }
 }