Integrated hydrology updates

- Riverbed generation includes upstream shutdown fixes

adapters/bukkit/plugin/src/main/java/art/arcane/iris/Iris.java

@@ -989,6 +989,7 @@ public class Iris extends VolmitPlugin implements Listener, ReloadAware {
     }
 
     public void onDisable() {
+        pendingWorldReplacements.awaitStagingForShutdown();
         GuiHost.shutdown();
         preservePendingBlockDrops();
         papiLifecycleActive = false;
@@ -1006,7 +1007,8 @@ public class Iris extends VolmitPlugin implements Listener, ReloadAware {
         }
         IrisLanguage.shutdown();
         teardownPapi();
-        boolean deferToServerStop = FailedEnableLock.defersTeardownToServerStop(enableFailed, IrisToolbelt.isServerStopping());
+        boolean deferToServerStop = FailedEnableLock.defersTeardownToServerStop(enableFailed, IrisToolbelt.isServerStopping())
+                && INMS.get().deferPluginClassLoaderClose();
         boolean restartingAtStartupBoundary = startupBoundaryRestart.get();
         if (restartingAtStartupBoundary) {
             teardownRuntime("startup-boundary-restart", 30L);
@@ -1038,7 +1040,7 @@ public class Iris extends VolmitPlugin implements Listener, ReloadAware {
         preservePendingBlockDrops();
         papiLifecycleActive = false;
         teardownPapi();
-        if (IrisToolbelt.isServerStopping()) {
+        if (IrisToolbelt.isServerStopping() && INMS.get().deferPluginClassLoaderClose()) {
             quiesceRuntimeForServerShutdown("pre-unload:" + reason);
             startPostStopFinisher();
             Iris.debug("Pre-unload hook deferred generator teardown until Paper closes its chunk schedulers.");
@@ -1130,7 +1132,6 @@ public class Iris extends VolmitPlugin implements Listener, ReloadAware {
 
     private void quiesceRuntimeForServerShutdown(String reason) {
         serverStopTeardownDeferred.set(true);
-        INMS.get().deferPluginClassLoaderClose();
         JigsawStudioService jigsawStudioService = IrisServices.getOrNull(JigsawStudioService.class);
         if (jigsawStudioService != null) {
             try {

adapters/bukkit/plugin/src/main/java/art/arcane/iris/world/PendingWorldReplacementManager.java

@@ -58,6 +58,7 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -78,6 +79,9 @@ public final class PendingWorldReplacementManager implements Listener {
     private volatile UUID overworldEntryAwaitingVerification;
     private volatile World overworldEntryWorld;
     private volatile boolean paperEntryListenerRegistered;
+    private final Object stagingMonitor = new Object();
+    private boolean stagingAdmissionOpen = true;
+    private StagingAdmission activeStaging;
 
     public PendingWorldReplacementManager(Iris plugin) {
         this.plugin = Objects.requireNonNull(plugin, "plugin");
@@ -135,7 +139,8 @@ public final class PendingWorldReplacementManager implements Listener {
                 LifecycleOperationCoordinator.Domain.WORLD_MUTATION,
                 LifecycleOperationCoordinator.OperationKind.WORLD_REPLACE,
                 requiredWorldKey.toString()
-        ); WorldReplacementProgress progress = WorldReplacementProgress.start(
+        ); StagingAdmission ignoredAdmission = admitStaging(requiredWorldKey);
+             WorldReplacementProgress progress = WorldReplacementProgress.start(
                 requiredSender, requiredWorldKey.toString())) {
             synchronized (this) {
                 IrisStartupValidation.requireWorldReplacementStagingReady();
@@ -271,6 +276,29 @@ public final class PendingWorldReplacementManager implements Listener {
         }
     }
 
+    public void awaitStagingForShutdown() {
+        StagingAdmission staging;
+        synchronized (stagingMonitor) {
+            stagingAdmissionOpen = false;
+            staging = activeStaging;
+        }
+        if (staging == null) {
+            return;
+        }
+        Iris.warn("Waiting for the replacement of " + staging.worldKey + " to finish staging before Iris shuts down.");
+        boolean interrupted = false;
+        while (staging.finished.getCount() > 0L) {
+            try {
+                staging.finished.await();
+            } catch (InterruptedException e) {
+                interrupted = true;
+            }
+        }
+        if (interrupted) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
     public synchronized void processPendingStartupReplacements() {
         ArrayList<String> failures = new ArrayList<>();
         ArrayList<String> restartBoundaries = new ArrayList<>();
@@ -1048,6 +1076,17 @@ public final class PendingWorldReplacementManager implements Listener {
         return plugin.getDataFolder().toPath().toAbsolutePath().normalize();
     }
 
+    private StagingAdmission admitStaging(NamespacedKey worldKey) throws IOException {
+        synchronized (stagingMonitor) {
+            if (!stagingAdmissionOpen) {
+                throw new IOException("Iris is shutting down. Run the replacement again after the server starts.");
+            }
+            StagingAdmission staging = new StagingAdmission(worldKey);
+            activeStaging = staging;
+            return staging;
+        }
+    }
+
     private synchronized void loadOverworldEntryGuard() throws IOException {
         ExactWorldSlotPathPolicy.Target target = resolveTarget(OVERWORLD_KEY);
         Optional<WorldReplacementEntryGuard.Entry> loaded = WorldReplacementEntryGuard.load(target.worldDirectory());
@@ -1218,6 +1257,25 @@ public final class PendingWorldReplacementManager implements Listener {
     record SavedLocationInspection(boolean safe, Throwable failure) {
     }
 
+    private final class StagingAdmission implements AutoCloseable {
+        private final NamespacedKey worldKey;
+        private final CountDownLatch finished = new CountDownLatch(1);
+
+        private StagingAdmission(NamespacedKey worldKey) {
+            this.worldKey = worldKey;
+        }
+
+        @Override
+        public void close() {
+            synchronized (stagingMonitor) {
+                if (activeStaging == this) {
+                    activeStaging = null;
+                }
+            }
+            finished.countDown();
+        }
+    }
+
     private static final class RestartBoundaryRequired extends IOException {
         private RestartBoundaryRequired(String message) {
             super(message);

adapters/bukkit/plugin/src/test/java/art/arcane/iris/IrisShutdownMaintenanceTest.java

@@ -16,6 +16,7 @@ import art.arcane.iris.world.IrisWorld;
 import art.arcane.iris.world.WorldMaintenance;
 import art.arcane.iris.world.runtime.BukkitEnginePlatformHooks;
 import art.arcane.iris.world.task.J;
+import art.arcane.volmlib.integration.ReloadAware;
 import art.arcane.volmlib.nativelib.terrain.ServerShutdownBoundary;
 import org.bukkit.Bukkit;
 import org.bukkit.NamespacedKey;
@@ -37,9 +38,11 @@ import java.util.concurrent.atomic.AtomicReference;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.never;
@@ -109,8 +112,6 @@ public class IrisShutdownMaintenanceTest {
                 }
                 verify(fixture.generator).quiesceForServerShutdown();
                 verify(fixture.generator, never()).closeAsync();
-                fixture.nms.verify(INMS::get);
-                verify(fixture.binding).deferPluginClassLoaderClose();
                 verify(fixture.binding, never()).releasePluginClassLoaderClose();
             }
             assertEquals(0, sessions.activeLeases());
@@ -151,6 +152,37 @@ public class IrisShutdownMaintenanceTest {
         }
     }
 
+    @Test
+    public void serverStopPreUnloadDefersTeardownWhileTheClassLoaderIsHeld() throws Exception {
+        try (Fixture fixture = new Fixture()) {
+            when(fixture.binding.deferPluginClassLoaderClose()).thenReturn(true);
+            setField(fixture.plugin, "postStopFinisherStarted", new AtomicBoolean(true));
+            try {
+                fixture.commitReloadDuringServerStop();
+                verify(fixture.generator).quiesceForServerShutdown();
+                verify(fixture.generator, never()).closeAsync();
+                assertTrue(fixture.serverStopTeardownDeferred());
+            } finally {
+                WorldMaintenance.endWorldMaintenance(fixture.identity, "pre-unload:HOT_UNLOAD");
+            }
+        }
+    }
+
+    @Test
+    public void serverStopPreUnloadDrainsInlineWhenTheClassLoaderCannotBeHeld() throws Exception {
+        try (Fixture fixture = new Fixture()) {
+            when(fixture.binding.deferPluginClassLoaderClose()).thenReturn(false);
+            try {
+                fixture.commitReloadDuringServerStop();
+                verify(fixture.generator).closeAsync();
+                verify(fixture.generator, never()).quiesceForServerShutdown();
+                assertFalse(fixture.serverStopTeardownDeferred());
+            } finally {
+                WorldMaintenance.endWorldMaintenance(fixture.identity, "pre-unload:HOT_UNLOAD", true);
+            }
+        }
+    }
+
     private static void setField(Iris plugin, String name, Object value) throws ReflectiveOperationException {
         Field field = Iris.class.getDeclaredField(name);
         field.setAccessible(true);
@@ -206,6 +238,19 @@ public class IrisShutdownMaintenanceTest {
             method.invoke(plugin, "manual_disable", 5L);
         }
 
+        private void commitReloadDuringServerStop() throws ReflectiveOperationException {
+            setField(plugin, "alreadyDrained", new AtomicBoolean());
+            toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true);
+            doCallRealMethod().when(plugin).commitReload(any());
+            plugin.commitReload(ReloadAware.PreUnloadReason.HOT_UNLOAD);
+        }
+
+        private boolean serverStopTeardownDeferred() throws ReflectiveOperationException {
+            Field field = Iris.class.getDeclaredField("serverStopTeardownDeferred");
+            field.setAccessible(true);
+            return ((AtomicBoolean) field.get(plugin)).get();
+        }
+
         private void startFinisher() throws ReflectiveOperationException {
             Method method = Iris.class.getDeclaredMethod("startPostStopFinisher");
             method.setAccessible(true);

adapters/bukkit/plugin/src/test/java/art/arcane/iris/world/PendingWorldReplacementAdmissionTest.java

@@ -11,12 +11,16 @@ import art.arcane.iris.platform.bukkit.plugin.VolmitSender;
 import org.bukkit.NamespacedKey;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertThrows;
 import static org.mockito.Mockito.mock;
 
@@ -42,4 +46,55 @@ public class PendingWorldReplacementAdmissionTest {
             executor.awaitTermination(5L, TimeUnit.SECONDS);
         }
     }
+
+    @Test
+    public void shutdownRejectsReplacementsThatStartAfterIt() {
+        PendingWorldReplacementManager manager = new PendingWorldReplacementManager(mock(Iris.class));
+        manager.awaitStagingForShutdown();
+
+        IOException rejected = assertThrows(IOException.class, () -> manager.stageReplacement(
+                mock(VolmitSender.class), NamespacedKey.minecraft("overworld"), mock(IrisDimension.class), null));
+
+        assertTrue(rejected.getMessage().contains("shutting down"));
+        assertTrue(LifecycleOperationCoordinator.get().isIdle());
+    }
+
+    @Test
+    public void shutdownWaitsForTheReplacementBeingStaged() throws Exception {
+        PendingWorldReplacementManager manager = new PendingWorldReplacementManager(mock(Iris.class));
+        AtomicReference<Throwable> stagingFailure = new AtomicReference<>();
+        Thread staging = new Thread(() -> {
+            try {
+                manager.stageReplacement(mock(VolmitSender.class), NamespacedKey.minecraft("overworld"),
+                        mock(IrisDimension.class), null);
+            } catch (Throwable failure) {
+                stagingFailure.set(failure);
+            }
+        });
+        Thread shutdown = new Thread(manager::awaitStagingForShutdown);
+        synchronized (manager) {
+            staging.start();
+            awaitState(staging, Thread.State.BLOCKED);
+            shutdown.start();
+            awaitState(shutdown, Thread.State.WAITING);
+            assertTrue(staging.isAlive());
+        }
+        shutdown.join(TimeUnit.SECONDS.toMillis(10L));
+        staging.join(TimeUnit.SECONDS.toMillis(10L));
+
+        assertFalse(shutdown.isAlive());
+        assertFalse(staging.isAlive());
+        assertTrue(stagingFailure.get() instanceof Exception);
+        assertTrue(LifecycleOperationCoordinator.get().isIdle());
+    }
+
+    private static void awaitState(Thread thread, Thread.State state) throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5L);
+        while (thread.getState() != state) {
+            if (System.nanoTime() > deadline) {
+                throw new AssertionError(thread.getName() + " never reached " + state + "; it is " + thread.getState());
+            }
+            Thread.sleep(5L);
+        }
+    }
 }

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

@@ -301,7 +301,8 @@ public interface INMSBinding extends NativeTerrainAccess {
     default void uninjectBukkit() {
     }
 
-    default void deferPluginClassLoaderClose() {
+    default boolean deferPluginClassLoaderClose() {
+        return false;
     }
 
     default void releasePluginClassLoaderClose() {

core/src/test/java/art/arcane/iris/platform/bukkit/nms/BukkitBindingTest.java

@@ -30,7 +30,8 @@ public class BukkitBindingTest {
         Fixture fixture = new Fixture();
         when(fixture.lifecycle.isServerStopping()).thenReturn(true);
         assertTrue(fixture.binding.isServerStopping());
-        fixture.binding.deferPluginClassLoaderClose();
+        when(fixture.lifecycle.deferPluginClassLoaderClose()).thenReturn(true);
+        assertTrue(fixture.binding.deferPluginClassLoaderClose());
         fixture.binding.releasePluginClassLoaderClose();
         verify(fixture.lifecycle).deferPluginClassLoaderClose();
         verify(fixture.lifecycle).releasePluginClassLoaderClose();