Added native chunk draining

- Native chunk queues can advance on their owning server thread
- Folia and foreign threads cannot run global chunk promotion

adapters/bukkit/nms/v26_2_R1/src/main/java/art/arcane/iris/core/nms/v26_2_R1/NMSBinding.java

@@ -38,6 +38,7 @@ import art.arcane.volmlib.util.json.JSONObject;
 import art.arcane.volmlib.util.mantle.runtime.Mantle;
 import art.arcane.volmlib.util.matter.Matter;
 import art.arcane.volmlib.util.math.Vector3d;
+import art.arcane.volmlib.util.scheduling.FoliaScheduler;
 import art.arcane.volmlib.util.matter.MatterBiomeInject;
 import art.arcane.iris.spi.PlatformBiome;
 import art.arcane.iris.util.nbt.common.mca.NBTWorld;
@@ -1140,6 +1141,18 @@ public class NMSBinding implements INMSBinding {
         }
     }
 
+    @Override
+    public boolean pollChunkTask(World world) {
+        if (FoliaScheduler.isRegionizedRuntime(Bukkit.getServer())) {
+            return false;
+        }
+        ServerLevel level = ((CraftWorld) world).getHandle();
+        if (Thread.currentThread() != level.getServer().getRunningThread()) {
+            throw new IllegalStateException("Native chunk tasks require the server lifecycle thread");
+        }
+        return level.getChunkSource().pollTask();
+    }
+
     @Override
     public void flushChunkIO(World world) {
         try {

adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/core/nms/v26_2_R1/NmsChunkTaskPollingTest.java

@@ -0,0 +1,95 @@
+package art.arcane.iris.core.nms.v26_2_R1;
+
+import art.arcane.volmlib.util.scheduling.FoliaScheduler;
+import net.minecraft.SharedConstants;
+import net.minecraft.server.Bootstrap;
+import net.minecraft.server.MinecraftServer;
+import net.minecraft.server.level.ServerChunkCache;
+import net.minecraft.server.level.ServerLevel;
+import org.bukkit.Bukkit;
+import org.bukkit.Server;
+import org.bukkit.craftbukkit.CraftWorld;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+public class NmsChunkTaskPollingTest {
+    @BeforeClass
+    public static void bootstrapMinecraftRegistries() {
+        SharedConstants.tryDetectVersion();
+        Bootstrap.bootStrap();
+    }
+
+    @Test
+    public void serverOwnerPumpsOnlyTheTargetWorldQueue() {
+        Server bukkitServer = mock(Server.class);
+        CraftWorld world = mock(CraftWorld.class);
+        ServerLevel level = mock(ServerLevel.class);
+        MinecraftServer server = mock(MinecraftServer.class);
+        ServerChunkCache chunks = mock(ServerChunkCache.class);
+        NMSBinding binding = mock(NMSBinding.class, CALLS_REAL_METHODS);
+        CompletableFuture<Void> completed = new CompletableFuture<>();
+        when(world.getHandle()).thenReturn(level);
+        when(level.getServer()).thenReturn(server);
+        when(server.getRunningThread()).thenReturn(Thread.currentThread());
+        when(level.getChunkSource()).thenReturn(chunks);
+        when(chunks.pollTask()).thenAnswer(invocation -> completed.complete(null)).thenReturn(false);
+        try (MockedStatic<Bukkit> bukkit = mockStatic(Bukkit.class);
+             MockedStatic<FoliaScheduler> folia = mockStatic(FoliaScheduler.class)) {
+            bukkit.when(Bukkit::getServer).thenReturn(bukkitServer);
+            folia.when(() -> FoliaScheduler.isRegionizedRuntime(bukkitServer)).thenReturn(false);
+            assertTrue(binding.pollChunkTask(world));
+            assertTrue(completed.isDone());
+            assertFalse(binding.pollChunkTask(world));
+        }
+        verify(chunks, times(2)).pollTask();
+        verify(server, never()).executeAllRecentInternalTasks();
+    }
+
+    @Test
+    public void nonOwnerIsRejectedBeforeAccessingTheChunkQueue() {
+        Server bukkitServer = mock(Server.class);
+        CraftWorld world = mock(CraftWorld.class);
+        ServerLevel level = mock(ServerLevel.class);
+        MinecraftServer server = mock(MinecraftServer.class);
+        NMSBinding binding = mock(NMSBinding.class, CALLS_REAL_METHODS);
+        when(world.getHandle()).thenReturn(level);
+        when(level.getServer()).thenReturn(server);
+        when(server.getRunningThread()).thenReturn(new Thread("Other server owner"));
+        try (MockedStatic<Bukkit> bukkit = mockStatic(Bukkit.class);
+             MockedStatic<FoliaScheduler> folia = mockStatic(FoliaScheduler.class)) {
+            bukkit.when(Bukkit::getServer).thenReturn(bukkitServer);
+            folia.when(() -> FoliaScheduler.isRegionizedRuntime(bukkitServer)).thenReturn(false);
+            assertThrows(IllegalStateException.class, () -> binding.pollChunkTask(world));
+        }
+        verify(level, never()).getChunkSource();
+    }
+
+    @Test
+    public void foliaNeverUsesTheWorldWideTaskPump() {
+        Server bukkitServer = mock(Server.class);
+        CraftWorld world = mock(CraftWorld.class);
+        NMSBinding binding = mock(NMSBinding.class, CALLS_REAL_METHODS);
+        try (MockedStatic<Bukkit> bukkit = mockStatic(Bukkit.class);
+             MockedStatic<FoliaScheduler> folia = mockStatic(FoliaScheduler.class)) {
+            bukkit.when(Bukkit::getServer).thenReturn(bukkitServer);
+            folia.when(() -> FoliaScheduler.isRegionizedRuntime(bukkitServer)).thenReturn(true);
+            assertFalse(binding.pollChunkTask(world));
+        }
+        verifyNoInteractions(world);
+    }
+}

core/src/main/java/art/arcane/iris/core/nms/INMSBinding.java

@@ -242,6 +242,10 @@ public interface INMSBinding {
         return false;
     }
 
+    default boolean pollChunkTask(World world) {
+        return false;
+    }
+
     default void flushChunkIO(World world) {
     }