feat: protect mod payloads with main-thread ownership in 1.1.17

tests/payload/ReviewRegression.java

@@ -0,0 +1,81 @@
+package dev.modzuozhi;
+
+import dev.modzuozhi.core.compat.*;
+import dev.modzuozhi.core.dimthread.*;
+import net.minecraft.server.MinecraftServer;
+import net.minecraft.resources.ResourceLocation;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.*;
+
+public final class ReviewRegression {
+    static final List<String> failures=new ArrayList<>();
+    static void expect(boolean ok,String name){if(!ok){failures.add(name);ModZuozhi.LOGGER.error("REVIEW RED: {}",name);}}
+    public static void run(MinecraftServer server) throws Exception {
+        var manager=DimThreadCore.MANAGER;manager.setPaused(server,true);
+        var player=new net.neoforged.neoforge.common.util.FakePlayer(server.overworld(),
+            new com.mojang.authlib.GameProfile(UUID.randomUUID(),"ReviewTest"));
+        var disconnects=new AtomicInteger();
+        new net.minecraft.server.network.ServerGamePacketListenerImpl(server,
+            new net.minecraft.network.Connection(net.minecraft.network.protocol.PacketFlow.SERVERBOUND),player,
+            net.minecraft.server.network.CommonListenerCookie.createInitial(player.getGameProfile(),false)) {
+                @Override public void disconnect(net.minecraft.network.chat.Component reason){disconnects.incrementAndGet();}
+            };
+        var ctx=new net.neoforged.neoforge.network.handling.ServerPayloadContext(player.connection,ResourceLocation.parse("dimensionalripper:review"));
+        var result=new AtomicReference<CompletableFuture<Integer>>();
+        Thread rejected=new Thread(() -> result.set(ctx.enqueueWork(() -> {throw new RejectedExecutionException("handler executor closed");})));
+        rejected.start();rejected.join(5000);PayloadMainThreadDispatch.drain(server);PayloadMainThreadDispatch.drain(server);
+        expect(disconnects.get()==0,"handler rejection must not disconnect player");
+        try{expect(result.get().join()==null,"guard preserves handler rejection result");}
+        catch(CompletionException e){expect(false,"guard preserves handler rejection result");}
+        Thread cancelledHandler=new Thread(() -> result.set(ctx.enqueueWork(() -> {throw new CancellationException("handler cancelled its own computation");})));
+        cancelledHandler.start();cancelledHandler.join(5000);PayloadMainThreadDispatch.drain(server);
+        expect(result.get().join()==null,"guard preserves handler cancellation result");
+        var overflow=new AtomicReference<CompletableFuture<Integer>>();
+        Thread flood=new Thread(() -> {for(int i=0;i<257;i++)overflow.set(ctx.enqueueWork(() -> 1));});
+        flood.start();flood.join(5000);
+        expect(overflow.get().isCompletedExceptionally(),"capacity rejection remains exceptional");
+        PayloadMainThreadDispatch.drain(server);
+        expect(disconnects.get()==1,"genuine capacity rejection disconnects once");
+        for(int i=0;i<10;i++)PayloadMainThreadDispatch.drain(server);
+        String counters=PayloadDiagnostics.snapshot(server);
+        expect(counters.contains("failed=2") && counters.contains("rejected=1"),
+                "diagnostics distinguish handler failures from capacity rejection");
+
+        var other=server.getLevel(net.minecraft.world.level.Level.NETHER);
+        manager.removeLoop(other);
+        manager.runMainThreadPhase(server,"review-new-loop",() -> manager.getLoop(other).scheduleIfNeeded());
+        expect(manager.getLoop(other).isPaused(),"new loop inherits menu pause after phase exit");
+        manager.setPaused(server,true);
+
+        var loop=manager.getLoop(other);var entered=new CountDownLatch(1);var release=new CountDownLatch(1);
+        var finished=new AtomicBoolean();
+        loop.execute(() -> {
+            server.execute(() -> manager.removeLoop(other));entered.countDown();
+            while(release.getCount()!=0) try{release.await();}catch(InterruptedException ignored){}
+            finished.set(true);
+        });
+        manager.setPaused(server,false);PayloadRegression.check(entered.await(5,TimeUnit.SECONDS),"retiring loop fixture start");
+        Thread releaser=new Thread(() -> {try{Thread.sleep(250);}catch(InterruptedException ignored){}release.countDown();});
+        releaser.start();
+        try{manager.runMainThreadPhase(server,"review-retire",() -> expect(finished.get() && !loop.isTicking(),"retired loop must drain before exclusive phase"));}
+        finally{release.countDown();releaser.join(5000);}
+        manager.setPaused(server,true);
+
+        var labelled=new CountDownLatch(1);var labelRelease=new CountDownLatch(1);
+        var late=new AtomicReference<CompletableFuture<Integer>>();
+        Thread submitter=new Thread(() -> {
+            labelled.countDown();try{labelRelease.await(5,TimeUnit.SECONDS);}catch(InterruptedException e){throw new RuntimeException(e);}
+            // Exact late-registration boundary after the submitter had obtained its queue.
+            var ticket=PayloadDiagnostics.begin(server,"dimensionalripper:shutdown_race");
+            PayloadDiagnostics.end(server,ticket,new CancellationException("stopped"));
+            late.set(PayloadMainThreadDispatch.submit(player,ctx.payloadId(),() -> {throw new AssertionError("late action");}));
+        });
+        submitter.start();PayloadRegression.check(labelled.await(5,TimeUnit.SECONDS),"late submit fixture pause");
+        PayloadMainThreadDispatch.stop(server);PayloadMainThreadDispatch.clear(server);labelRelease.countDown();submitter.join(5000);
+        expect(late.get()!=null && late.get().isCompletedExceptionally(),"late future completed");
+        expect(PayloadDiagnostics.snapshot(server).equals("payloads: pending=0 batches=0"),"late submission must not recreate diagnostics after clear");
+        if(!failures.isEmpty())throw new AssertionError(String.join("; ",failures));
+        ModZuozhi.LOGGER.info("PAYLOAD REVIEW PASS");
+    }
+}