Extended probe worker drain

- Offline generation checks allow bounded hydrology prefetch completion

- Worker timeout failures include active pool state

- Tests cover timeout and interrupt handling

probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java

@@ -57,6 +57,7 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.LockSupport;
 
 final class RealPackProbeSupport {
+    private static final long WORKER_DRAIN_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(120);
     private static final List<Throwable> REPORTED = Collections.synchronizedList(new ArrayList<>());
 
     private RealPackProbeSupport() {
@@ -155,7 +156,7 @@ final class RealPackProbeSupport {
     }
 
     private static void awaitGenerationWorkers() throws InterruptedException {
-        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+        long deadline = System.nanoTime() + WORKER_DRAIN_TIMEOUT_NANOS;
         List<ForkJoinPool> pools = new ArrayList<>(3);
         for (MultiBurst burst : List.of(MultiBurst.hydrology, MultiBurst.ioBurst, MultiBurst.burst)) {
             try {
@@ -168,6 +169,10 @@ final class RealPackProbeSupport {
                 throw new IllegalStateException("Probe generation worker barrier failed", failure);
             }
         }
+        awaitGenerationWorkers(pools, deadline);
+    }
+
+    static void awaitGenerationWorkers(List<ForkJoinPool> pools, long deadline) throws InterruptedException {
         long[] previous = quietWorkerSnapshot(pools, deadline);
         while (true) {
             long[] current = quietWorkerSnapshot(pools, deadline);
@@ -187,7 +192,8 @@ final class RealPackProbeSupport {
                     throw new InterruptedException("Probe generation worker barrier interrupted");
                 }
                 if (System.nanoTime() >= deadline) {
-                    throw new IllegalStateException("Probe generation workers did not settle within 15 seconds");
+                    throw new IllegalStateException("Probe generation workers did not settle before the deadline: pool "
+                            + (index + 1) + "/" + pools.size() + " " + pool);
                 }
                 long count = pool.getStealCount();
                 boolean quiet = pool.isQuiescent();

probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java

@@ -17,13 +17,17 @@ import java.io.File;
 import java.net.URL;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ForkJoinPool;
+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.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public final class RealPackProbeSupportBarrierTest {
@@ -68,6 +72,89 @@ public final class RealPackProbeSupportBarrierTest {
         PlatformLeakGuard.requireClean("closed real-pack workspace", "end");
     }
 
+    @Test(timeout = 10_000L)
+    public void waitsForActiveWorkWithinItsDeadline() throws Exception {
+        try (ForkJoinPool pool = new ForkJoinPool(1)) {
+            CountDownLatch entered = new CountDownLatch(1);
+            Future<?> task = pool.submit(() -> {
+                entered.countDown();
+                Thread.sleep(100L);
+                return null;
+            });
+            assertTrue(entered.await(5L, TimeUnit.SECONDS));
+            RealPackProbeSupport.awaitGenerationWorkers(List.of(pool),
+                    System.nanoTime() + TimeUnit.SECONDS.toNanos(5));
+            assertTrue(task.isDone());
+            task.get();
+        }
+    }
+
+    @Test(timeout = 10_000L)
+    public void rejectsUnsettledWorkAtItsDeadlineWithoutCancellingIt() throws Exception {
+        CountDownLatch release = new CountDownLatch(1);
+        try (ForkJoinPool pool = new ForkJoinPool(1)) {
+            CountDownLatch entered = new CountDownLatch(1);
+            Future<?> task = pool.submit(() -> {
+                entered.countDown();
+                release.await();
+                return null;
+            });
+            try {
+                assertTrue(entered.await(5L, TimeUnit.SECONDS));
+                IllegalStateException failure = assertThrows(IllegalStateException.class,
+                        () -> RealPackProbeSupport.awaitGenerationWorkers(List.of(pool),
+                                System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(50)));
+                assertTrue(failure.getMessage().contains("pool 1/1"));
+                assertTrue(failure.getMessage().contains("active ="));
+                assertFalse(task.isDone());
+            } finally {
+                release.countDown();
+            }
+            RealPackProbeSupport.awaitGenerationWorkers(List.of(pool),
+                    System.nanoTime() + TimeUnit.SECONDS.toNanos(5));
+            task.get();
+        }
+    }
+
+    @Test(timeout = 10_000L)
+    public void interruptionStopsAnActiveWorkerWait() throws Exception {
+        CountDownLatch release = new CountDownLatch(1);
+        try (ForkJoinPool pool = new ForkJoinPool(1)) {
+            CountDownLatch entered = new CountDownLatch(1);
+            Future<?> task = pool.submit(() -> {
+                entered.countDown();
+                release.await();
+                return null;
+            });
+            AtomicReference<Throwable> failure = new AtomicReference<>();
+            CountDownLatch waiting = new CountDownLatch(1);
+            Thread waiter = new Thread(() -> {
+                waiting.countDown();
+                try {
+                    RealPackProbeSupport.awaitGenerationWorkers(List.of(pool),
+                            System.nanoTime() + TimeUnit.SECONDS.toNanos(5));
+                } catch (Throwable interrupted) {
+                    failure.set(interrupted);
+                }
+            });
+            try {
+                assertTrue(entered.await(5L, TimeUnit.SECONDS));
+                waiter.start();
+                assertTrue(waiting.await(5L, TimeUnit.SECONDS));
+                waiter.interrupt();
+                waiter.join(5_000L);
+                assertFalse(waiter.isAlive());
+                assertTrue(failure.get() instanceof InterruptedException);
+                assertFalse(task.isDone());
+            } finally {
+                release.countDown();
+                waiter.interrupt();
+                waiter.join(5_000L);
+            }
+            task.get();
+        }
+    }
+
     private static void verifyCrossPoolContinuation(RealPackProbeSupport.EngineSession session,
                                                      MultiBurst first, MultiBurst second) throws Exception {
         IllegalStateException failure = new IllegalStateException("nested generation failure");