Shared pending terrain samples

- Overlapping terrain batches and point requests share pending coordinate samples
- Failed terrain loads release reservations for retry
- Provisional worker results mark every caller and stay out of the shared cache
- Concurrency tests cover overlap, recursion, interruption, failure and single-worker batching

core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyBasisCache.java

@@ -1,8 +1,15 @@
 package art.arcane.iris.generation.hydrology;
 
+import art.arcane.iris.generation.stream.ProvisionalSampling;
 import art.arcane.volmlib.util.cache.CacheKey;
 import it.unimi.dsi.fastutil.longs.Long2ObjectLinkedOpenHashMap;
 import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
+
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.RejectedExecutionException;
 
 final class HydrologyBasisCache {
     private static final int MAXIMUM_ENTRIES = 32_768;
@@ -39,23 +46,52 @@ final class HydrologyBasisCache {
         seed(samples, full);
     }
 
-    boolean findWithoutSlope(long packed, HydrologyTerrainSample[] samples, int index) {
+    HydrologyTerrainSample[] sampleWithoutSlopeBatch(long[] coordinates, int count,
-        Stripe stripe = stripe(withoutSlope, packed);
+                                                    HydrologyNaturalTerrainSampler sampler) {
-        synchronized (stripe) {
+        Objects.checkFromIndexSize(0, count, coordinates.length);
-            samples[index] = stripe.samples.getAndMoveToFirst(packed);
+        if (count > HydrologyTerrainSampler.MAXIMUM_BATCH_SIZE) {
-            return samples[index] != null || stripe.samples.containsKey(packed);
+            throw new IllegalArgumentException("Terrain basis batch exceeds its sample limit");
         }
-    }
+        HydrologyTerrainSample[] samples = new HydrologyTerrainSample[count];
-
+        Pending[] pending = new Pending[count];
-    HydrologyTerrainSample rememberWithoutSlope(long packed, HydrologyTerrainSample sample) {
+        long[] ownedCoordinates = new long[count];
-        Stripe stripe = stripe(withoutSlope, packed);
+        Pending[] owned = new Pending[count];
-        synchronized (stripe) {
+        int ownedCount = 0;
-            if (stripe.samples.containsKey(packed)) {
+        try {
-                return stripe.samples.getAndMoveToFirst(packed);
+            for (int index = 0; index < count; index++) {
+                long packed = coordinates[index];
+                Stripe stripe = stripe(withoutSlope, packed);
+                synchronized (stripe) {
+                    HydrologyTerrainSample cached = stripe.samples.getAndMoveToFirst(packed);
+                    if (cached != null || stripe.samples.containsKey(packed)) {
+                        samples[index] = cached;
+                        continue;
+                    }
+                    Pending loading = stripe.pending.get(packed);
+                    if (loading == null) {
+                        loading = new Pending();
+                        stripe.pending.put(packed, loading);
+                        ownedCoordinates[ownedCount] = packed;
+                        owned[ownedCount++] = loading;
+                    }
+                    pending[index] = loading;
+                }
             }
-            stripe.put(packed, sample);
+            if (ownedCount > 0) {
-            return sample;
+                loadBatch(ownedCoordinates, owned, ownedCount, sampler);
+            }
+        } catch (RuntimeException | Error failure) {
+            for (int index = 0; index < ownedCount; index++) {
+                fail(withoutSlope, ownedCoordinates[index], owned[index], failure);
+            }
+            throw failure;
         }
+        for (int index = 0; index < count; index++) {
+            if (pending[index] != null) {
+                samples[index] = pending[index].await();
+            }
+        }
+        return samples;
     }
 
     void seedWithoutSlope(Long2ObjectMap<HydrologyTerrainSample> samples) {
@@ -64,15 +100,78 @@ final class HydrologyBasisCache {
 
     private HydrologyTerrainSample sample(int x, int z, HydrologyNaturalTerrainSampler sampler, boolean omitSlope) {
         long packed = RiverFootprint.pack(x, z);
-        Stripe stripe = stripe(omitSlope ? withoutSlope : full, packed);
+        Stripe[] stripes = omitSlope ? withoutSlope : full;
+        Stripe stripe = stripe(stripes, packed);
+        Pending loading;
+        boolean owner = false;
         synchronized (stripe) {
             HydrologyTerrainSample cached = stripe.samples.getAndMoveToFirst(packed);
             if (cached != null || stripe.samples.containsKey(packed)) {
                 return cached;
             }
-            HydrologyTerrainSample sampled = omitSlope ? sampler.sampleBasisWithoutSlope(x, z) : sampler.sampleBasis(x, z);
+            loading = stripe.pending.get(packed);
-            stripe.put(packed, sampled);
+            if (loading == null) {
-            return sampled;
+                loading = new Pending();
+                stripe.pending.put(packed, loading);
+                owner = true;
+            }
+        }
+        if (owner) {
+            try {
+                HydrologyTerrainSample sampled = ProvisionalSampling.memoizable(() -> omitSlope
+                        ? sampler.sampleBasisWithoutSlope(x, z) : sampler.sampleBasis(x, z));
+                complete(stripes, packed, loading, sampled, false);
+            } catch (ProvisionalSampling.Unmemoizable provisional) {
+                complete(stripes, packed, loading, provisional.value(), true);
+            } catch (RuntimeException | Error failure) {
+                fail(stripes, packed, loading, failure);
+                throw failure;
+            }
+        }
+        return loading.await();
+    }
+
+    private void loadBatch(long[] coordinates, Pending[] pending, int count, HydrologyNaturalTerrainSampler sampler) {
+        HydrologyTerrainSample[] samples;
+        boolean provisional = false;
+        try {
+            samples = ProvisionalSampling.memoizable(() -> sampler.sampleBasisWithoutSlopeBatch(coordinates, count));
+        } catch (ProvisionalSampling.Unmemoizable value) {
+            samples = value.value();
+            provisional = true;
+        }
+        if (samples.length < count) {
+            throw new IllegalStateException("Terrain basis batch returned too few samples");
+        }
+        for (int index = 0; index < count; index++) {
+            complete(withoutSlope, coordinates[index], pending[index], samples[index], provisional);
+        }
+    }
+
+    private void complete(Stripe[] stripes, long packed, Pending pending, HydrologyTerrainSample sample,
+                          boolean provisional) {
+        Stripe stripe = stripe(stripes, packed);
+        synchronized (stripe) {
+            if (stripe.samples.containsKey(packed)) {
+                sample = stripe.samples.getAndMoveToFirst(packed);
+                provisional = false;
+            } else if (!provisional) {
+                stripe.put(packed, sample);
+            }
+            stripe.pending.remove(packed);
+            pending.sample = sample;
+            pending.provisional = provisional;
+            pending.complete.countDown();
+        }
+    }
+
+    private void fail(Stripe[] stripes, long packed, Pending pending, Throwable failure) {
+        Stripe stripe = stripe(stripes, packed);
+        synchronized (stripe) {
+            if (stripe.pending.remove(packed, pending)) {
+                pending.failure = failure;
+                pending.complete.countDown();
+            }
         }
     }
 
@@ -92,6 +191,7 @@ final class HydrologyBasisCache {
     private static final class Stripe {
         private final int maximumEntries;
         private final Long2ObjectLinkedOpenHashMap<HydrologyTerrainSample> samples;
+        private final Long2ObjectOpenHashMap<Pending> pending = new Long2ObjectOpenHashMap<>();
 
         private Stripe(int maximumEntries) {
             this.maximumEntries = maximumEntries;
@@ -105,4 +205,61 @@ final class HydrologyBasisCache {
             samples.putAndMoveToFirst(packed, value);
         }
     }
+
+    private static final class Pending implements ForkJoinPool.ManagedBlocker {
+        private final Thread owner = Thread.currentThread();
+        private final CountDownLatch complete = new CountDownLatch(1);
+        private HydrologyTerrainSample sample;
+        private Throwable failure;
+        private boolean provisional;
+
+        private HydrologyTerrainSample await() {
+            if (!isReleasable() && owner == Thread.currentThread()) {
+                throw new IllegalStateException("Recursive terrain basis sampling");
+            }
+            boolean interrupted = false;
+            boolean managed = true;
+            try {
+                while (!isReleasable()) {
+                    try {
+                        if (managed) {
+                            ForkJoinPool.managedBlock(this);
+                        } else {
+                            complete.await();
+                        }
+                    } catch (InterruptedException interruption) {
+                        interrupted = true;
+                        managed = false;
+                    } catch (RejectedExecutionException unavailable) {
+                        managed = false;
+                    }
+                }
+            } finally {
+                if (interrupted) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+            if (failure instanceof Error error) {
+                throw error;
+            }
+            if (failure instanceof RuntimeException runtime) {
+                throw runtime;
+            }
+            if (provisional) {
+                ProvisionalSampling.mark();
+            }
+            return sample;
+        }
+
+        @Override
+        public boolean block() throws InterruptedException {
+            complete.await();
+            return true;
+        }
+
+        @Override
+        public boolean isReleasable() {
+            return complete.getCount() == 0;
+        }
+    }
 }

core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyFootprintCompiler.java

@@ -946,8 +946,6 @@ final class HydrologyFootprintCompiler {
             HydrologyTerrainSample cached = terrainBases.get(key);
             if (cached != null) {
                 samples[index] = cached;
-            } else if (sharedBasis != null && sharedBasis.findWithoutSlope(key, samples, index)) {
-                continue;
             } else if (naturalSampler == null) {
                 samples[index] = sampleTerrainBasis(RiverFootprint.unpackX(key), RiverFootprint.unpackZ(key));
             } else {
@@ -956,10 +954,11 @@ final class HydrologyFootprintCompiler {
             }
         }
         if (missingCount > 0) {
-            HydrologyTerrainSample[] loaded = naturalSampler.sampleBasisWithoutSlopeBatch(missing, missingCount);
+            HydrologyTerrainSample[] loaded = sharedBasis == null
+                    ? naturalSampler.sampleBasisWithoutSlopeBatch(missing, missingCount)
+                    : sharedBasis.sampleWithoutSlopeBatch(missing, missingCount, naturalSampler);
             for (int index = 0; index < missingCount; index++) {
-                samples[positions[index]] = sharedBasis == null ? loaded[index]
+                samples[positions[index]] = loaded[index];
-                        : sharedBasis.rememberWithoutSlope(missing[index], loaded[index]);
             }
         }
         for (int index = 0; index < count; index++) {

core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyForkJoin.java

@@ -1,5 +1,7 @@
 package art.arcane.iris.generation.hydrology;
 
+import art.arcane.iris.generation.stream.ProvisionalSampling;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -91,6 +93,7 @@ public final class HydrologyForkJoin {
         private final CountDownLatch complete = new CountDownLatch(1);
         private T result;
         private Throwable failure;
+        private boolean provisional;
 
         Task(Callable<T> work) {
             this.work = Objects.requireNonNull(work);
@@ -136,6 +139,9 @@ public final class HydrologyForkJoin {
             if (failure != null) {
                 throw unwrap(failure);
             }
+            if (provisional) {
+                ProvisionalSampling.mark();
+            }
             return result;
         }
 
@@ -156,7 +162,10 @@ public final class HydrologyForkJoin {
                 return;
             }
             try {
-                result = call(work);
+                result = ProvisionalSampling.memoizable(() -> call(work));
+            } catch (ProvisionalSampling.Unmemoizable value) {
+                result = value.value();
+                provisional = true;
             } catch (RuntimeException | Error failed) {
                 failure = failed;
             } finally {

core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyBasisCacheTest.java

@@ -1,5 +1,6 @@
 package art.arcane.iris.generation.hydrology;
 
+import art.arcane.iris.generation.stream.ProvisionalSampling;
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import org.junit.Test;
 
@@ -9,29 +10,347 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public class HydrologyBasisCacheTest {
     @Test
-    public void batchLookupPreservesNullAndFirstPublishedSample() {
+    public void batchLookupPreservesNullDuplicatesAndEviction() {
+        CountingSampler sampler = new CountingSampler();
         HydrologyBasisCache cache = new HydrologyBasisCache(1);
-        HydrologyTerrainSample[] samples = new HydrologyTerrainSample[1];
+        long key = RiverFootprint.pack(Integer.MIN_VALUE, Integer.MAX_VALUE);
-        long key = RiverFootprint.pack(-17, Integer.MAX_VALUE);
+        assertArrayEquals(new HydrologyTerrainSample[2],
-        assertTrue(!cache.findWithoutSlope(key, samples, 0));
+                cache.sampleWithoutSlopeBatch(new long[]{key, key}, 2, sampler));
-        assertNull(cache.rememberWithoutSlope(key, null));
+        assertNull(cache.sampleWithoutSlope(Integer.MIN_VALUE, Integer.MAX_VALUE, sampler));
-        assertTrue(cache.findWithoutSlope(key, samples, 0));
+        assertEquals(1, sampler.slopeFreeCalls.get());
-        assertNull(samples[0]);
+        long later = RiverFootprint.pack(73, -17);
-        HydrologyTerrainSample later = HydrologyTerrainSample.openLand(73, 0D, "land");
+        assertEquals(87, cache.sampleWithoutSlopeBatch(new long[]{later}, 1, sampler)[0].naturalHeight());
-        assertNull(cache.rememberWithoutSlope(key, later));
+        assertNull(cache.sampleWithoutSlopeBatch(new long[]{key}, 1, sampler)[0]);
-        cache.rememberWithoutSlope(key + 1, later);
+        assertEquals(3, sampler.slopeFreeCalls.get());
-        assertTrue(!cache.findWithoutSlope(key, samples, 0));
+    }
-        assertEquals(later, cache.rememberWithoutSlope(key + 1, null));
+
+    @Test(timeout = 10_000)
+    public void overlappingBatchesDispatchOnlyTheirOwnedCoordinates() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        CountDownLatch firstStarted = new CountDownLatch(1);
+        CountDownLatch secondStarted = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        AtomicInteger batchCalls = new AtomicInteger();
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) {
+                if (batchCalls.getAndIncrement() == 0) {
+                    assertEquals(2, count);
+                    firstStarted.countDown();
+                } else {
+                    assertEquals(1, count);
+                    assertEquals(RiverFootprint.pack(3, 0), coordinates[0]);
+                    secondStarted.countDown();
+                }
+                await(release);
+                return super.sampleBasisWithoutSlopeBatch(coordinates, count);
+            }
+        };
+        ExecutorService workers = Executors.newFixedThreadPool(2);
+        try {
+            Future<HydrologyTerrainSample[]> first = workers.submit(() -> cache.sampleWithoutSlopeBatch(
+                    new long[]{RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}, 2, sampler));
+            assertTrue(firstStarted.await(5, TimeUnit.SECONDS));
+            Future<HydrologyTerrainSample[]> second = workers.submit(() -> cache.sampleWithoutSlopeBatch(
+                    new long[]{RiverFootprint.pack(2, 0), RiverFootprint.pack(3, 0)}, 2, sampler));
+            assertTrue(secondStarted.await(5, TimeUnit.SECONDS));
+            release.countDown();
+            assertSame(first.get(5, TimeUnit.SECONDS)[1], second.get(5, TimeUnit.SECONDS)[0]);
+            assertEquals(3, sampler.slopeFreeCalls.get());
+            assertEquals(2, batchCalls.get());
+        } finally {
+            release.countDown();
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test(timeout = 10_000)
+    public void pointAndBatchSharePendingSamplesWithoutHoldingStripeLocks() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache(1);
+        CountDownLatch pointStarted = new CountDownLatch(1);
+        CountDownLatch batchStarted = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample sampleBasisWithoutSlope(int x, int z) {
+                if (x == 1) {
+                    pointStarted.countDown();
+                    await(release);
+                }
+                return super.sampleBasisWithoutSlope(x, z);
+            }
+
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) {
+                assertEquals(1, count);
+                assertEquals(RiverFootprint.pack(2, 0), coordinates[0]);
+                batchStarted.countDown();
+                return super.sampleBasisWithoutSlopeBatch(coordinates, count);
+            }
+        };
+        ExecutorService workers = Executors.newFixedThreadPool(2);
+        try {
+            Future<HydrologyTerrainSample> point = workers.submit(() -> cache.sampleWithoutSlope(1, 0, sampler));
+            assertTrue(pointStarted.await(5, TimeUnit.SECONDS));
+            Future<HydrologyTerrainSample[]> batch = workers.submit(() -> cache.sampleWithoutSlopeBatch(
+                    new long[]{RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}, 2, sampler));
+            assertTrue(batchStarted.await(5, TimeUnit.SECONDS));
+            release.countDown();
+            assertSame(point.get(5, TimeUnit.SECONDS), batch.get(5, TimeUnit.SECONDS)[0]);
+            assertEquals(2, sampler.slopeFreeCalls.get());
+        } finally {
+            release.countDown();
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test(timeout = 10_000)
+    public void nestedBatchesFinishOnSingleWorkerPool() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        ForkJoinPool workers = new ForkJoinPool(1);
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) {
+                List<HydrologyTerrainSample> values = HydrologyForkJoin.invokeAll(List.of(
+                        () -> sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[0]), 0),
+                        () -> sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[1]), 0)), workers);
+                return values.toArray(HydrologyTerrainSample[]::new);
+            }
+        };
+        try {
+            workers.submit(() -> HydrologyForkJoin.invokeAll(List.of(
+                    () -> cache.sampleWithoutSlopeBatch(new long[]{RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}, 2, sampler),
+                    () -> cache.sampleWithoutSlopeBatch(new long[]{RiverFootprint.pack(2, 0), RiverFootprint.pack(1, 0)}, 2, sampler)), workers))
+                    .get(5, TimeUnit.SECONDS);
+            assertEquals(2, sampler.slopeFreeCalls.get());
+        } finally {
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test
+    public void failedMalformedAndRecursiveBatchesReleaseEveryReservation() {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)};
+        CountingSampler failed = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) {
+                throw new IllegalArgumentException("failure");
+            }
+        };
+        assertThrows(IllegalArgumentException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, failed));
+        CountingSampler malformed = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) {
+                return new HydrologyTerrainSample[1];
+            }
+        };
+        assertThrows(IllegalStateException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, malformed));
+        CountingSampler absent = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) {
+                return null;
+            }
+        };
+        assertThrows(NullPointerException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, absent));
+        CountingSampler recursive = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) {
+                cache.sampleWithoutSlope(2, 0, this);
+                throw new AssertionError("Recursive sampling must not wait on itself");
+            }
+        };
+        assertThrows(IllegalStateException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, recursive));
+        CountingSampler successful = new CountingSampler();
+        assertEquals(2, cache.sampleWithoutSlopeBatch(coordinates, 2, successful).length);
+        assertEquals(2, successful.slopeFreeCalls.get());
+    }
+
+    @Test(timeout = 10_000)
+    public void interruptedPointWaiterReceivesBatchResultAndRestoresInterrupt() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        AtomicReference<Thread> waiting = new AtomicReference<>();
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) {
+                started.countDown();
+                await(release);
+                return super.sampleBasisWithoutSlopeBatch(coordinates, count);
+            }
+        };
+        ExecutorService workers = Executors.newFixedThreadPool(2);
+        try {
+            Future<HydrologyTerrainSample[]> batch = workers.submit(() -> cache.sampleWithoutSlopeBatch(
+                    new long[]{RiverFootprint.pack(1, 0)}, 1, sampler));
+            assertTrue(started.await(5, TimeUnit.SECONDS));
+            Future<HydrologyTerrainSample> point = workers.submit(() -> {
+                waiting.set(Thread.currentThread());
+                HydrologyTerrainSample sample = cache.sampleWithoutSlope(1, 0, sampler);
+                assertTrue(Thread.interrupted());
+                return sample;
+            });
+            awaitWaiting(waiting);
+            waiting.get().interrupt();
+            release.countDown();
+            assertSame(batch.get(5, TimeUnit.SECONDS)[0], point.get(5, TimeUnit.SECONDS));
+            assertEquals(1, sampler.slopeFreeCalls.get());
+        } finally {
+            release.countDown();
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test(timeout = 10_000)
+    public void batchFailureReachesPointWaiterAndAllowsAllCoordinatesToRetry() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        AtomicReference<Thread> waiting = new AtomicReference<>();
+        IllegalArgumentException failure = new IllegalArgumentException("retry");
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) {
+                started.countDown();
+                await(release);
+                throw failure;
+            }
+        };
+        long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)};
+        ExecutorService workers = Executors.newFixedThreadPool(2);
+        try {
+            Future<IllegalArgumentException> batch = workers.submit(() -> assertThrows(IllegalArgumentException.class,
+                    () -> cache.sampleWithoutSlopeBatch(coordinates, 2, sampler)));
+            assertTrue(started.await(5, TimeUnit.SECONDS));
+            Future<IllegalArgumentException> point = workers.submit(() -> {
+                waiting.set(Thread.currentThread());
+                return assertThrows(IllegalArgumentException.class, () -> cache.sampleWithoutSlope(2, 0, sampler));
+            });
+            awaitWaiting(waiting);
+            release.countDown();
+            assertSame(failure, batch.get(5, TimeUnit.SECONDS));
+            assertSame(failure, point.get(5, TimeUnit.SECONDS));
+            CountingSampler successful = new CountingSampler();
+            assertEquals(2, cache.sampleWithoutSlopeBatch(coordinates, 2, successful).length);
+            assertEquals(2, successful.slopeFreeCalls.get());
+        } finally {
+            release.countDown();
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test
+    public void provisionalResultsAreNotRetained() {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample sampleBasisWithoutSlope(int x, int z) {
+                ProvisionalSampling.mark();
+                return super.sampleBasisWithoutSlope(x, z);
+            }
+        };
+        long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)};
+        cache.sampleWithoutSlopeBatch(coordinates, 2, sampler);
+        cache.sampleWithoutSlopeBatch(coordinates, 2, sampler);
+        cache.sampleWithoutSlope(1, 0, sampler);
+        cache.sampleWithoutSlope(1, 0, sampler);
+        assertEquals(6, sampler.slopeFreeCalls.get());
+    }
+
+    @Test(timeout = 10_000)
+    public void provisionalWorkerResultMarksSharedCacheWaitersAndIsNotRetained() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        AtomicReference<Thread> waiting = new AtomicReference<>();
+        ExecutorService providerWorkers = Executors.newFixedThreadPool(2);
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) {
+                List<HydrologyTerrainSample> values = HydrologyForkJoin.invokeAll(List.of(() -> {
+                    started.countDown();
+                    await(release);
+                    ProvisionalSampling.mark();
+                    return sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[0]), 0);
+                }, () -> sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[1]), 0)), providerWorkers);
+                return values.toArray(HydrologyTerrainSample[]::new);
+            }
+        };
+        ExecutorService callers = Executors.newFixedThreadPool(2);
+        long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)};
+        try {
+            Future<HydrologyTerrainSample[]> batch = callers.submit(() -> cache.sampleWithoutSlopeBatch(coordinates, 2, sampler));
+            assertTrue(started.await(5, TimeUnit.SECONDS));
+            Future<ProvisionalSampling.Unmemoizable> point = callers.submit(() -> {
+                waiting.set(Thread.currentThread());
+                return assertThrows(ProvisionalSampling.Unmemoizable.class,
+                        () -> ProvisionalSampling.memoizable(() -> cache.sampleWithoutSlope(1, 0, sampler)));
+            });
+            awaitWaiting(waiting);
+            release.countDown();
+            assertSame(batch.get(5, TimeUnit.SECONDS)[0], point.get(5, TimeUnit.SECONDS).value());
+            cache.sampleWithoutSlopeBatch(coordinates, 2, sampler);
+            assertEquals(4, sampler.slopeFreeCalls.get());
+        } finally {
+            release.countDown();
+            callers.shutdownNow();
+            providerWorkers.shutdownNow();
+            assertTrue(callers.awaitTermination(5, TimeUnit.SECONDS));
+            assertTrue(providerWorkers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test(timeout = 10_000)
+    public void seedWinsPendingSample() throws Exception {
+        HydrologyBasisCache cache = new HydrologyBasisCache();
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        CountingSampler sampler = new CountingSampler() {
+            @Override
+            public HydrologyTerrainSample sampleBasisWithoutSlope(int x, int z) {
+                started.countDown();
+                await(release);
+                return super.sampleBasisWithoutSlope(x, z);
+            }
+        };
+        ExecutorService workers = Executors.newSingleThreadExecutor();
+        try {
+            Future<HydrologyTerrainSample[]> result = workers.submit(() -> cache.sampleWithoutSlopeBatch(
+                    new long[]{RiverFootprint.pack(1, 0)}, 1, sampler));
+            assertTrue(started.await(5, TimeUnit.SECONDS));
+            HydrologyTerrainSample seeded = HydrologyTerrainSample.openLand(73, 0D, "seed");
+            Long2ObjectOpenHashMap<HydrologyTerrainSample> seeds = new Long2ObjectOpenHashMap<>();
+            seeds.put(RiverFootprint.pack(1, 0), seeded);
+            cache.seedWithoutSlope(seeds);
+            release.countDown();
+            assertSame(seeded, result.get(5, TimeUnit.SECONDS)[0]);
+            assertSame(seeded, cache.sampleWithoutSlope(1, 0, sampler));
+        } finally {
+            release.countDown();
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
     }
 
     @Test
@@ -122,7 +441,28 @@ public class HydrologyBasisCacheTest {
         }
     }
 
-    private static final class CountingSampler implements HydrologyNaturalTerrainSampler {
+    private static void awaitWaiting(AtomicReference<Thread> thread) {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (System.nanoTime() < deadline) {
+            Thread current = thread.get();
+            if (current != null && current.getState() == Thread.State.WAITING) {
+                return;
+            }
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+        }
+        throw new AssertionError("Cache waiter did not wait for its owner");
+    }
+
+    private static void await(CountDownLatch latch) {
+        try {
+            assertTrue(latch.await(5, TimeUnit.SECONDS));
+        } catch (InterruptedException interruption) {
+            Thread.currentThread().interrupt();
+            throw new AssertionError(interruption);
+        }
+    }
+
+    private static class CountingSampler implements HydrologyNaturalTerrainSampler {
         private final AtomicInteger fullCalls = new AtomicInteger();
         private final AtomicInteger slopeFreeCalls = new AtomicInteger();
         private final AtomicInteger failures = new AtomicInteger();

core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyForkJoinTest.java

@@ -1,5 +1,6 @@
 package art.arcane.iris.generation.hydrology;
 
+import art.arcane.iris.generation.stream.ProvisionalSampling;
 import org.junit.Test;
 
 import java.util.ArrayList;
@@ -23,6 +24,44 @@ import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public class HydrologyForkJoinTest {
+    @Test(timeout = 10_000)
+    public void workerProvisionalAnswersReachEveryAwaitingCaller() throws Exception {
+        HydrologyForkJoin.Task<Integer> task = new HydrologyForkJoin.Task<>(() -> {
+            ProvisionalSampling.mark();
+            return 73;
+        });
+        ExecutorService workers = Executors.newFixedThreadPool(2);
+        try {
+            workers.submit(task).get(5, TimeUnit.SECONDS);
+            for (int attempt = 0; attempt < 2; attempt++) {
+                ProvisionalSampling.Unmemoizable result = workers.submit(() -> assertThrows(
+                        ProvisionalSampling.Unmemoizable.class,
+                        () -> ProvisionalSampling.memoizable(task::await))).get(5, TimeUnit.SECONDS);
+                assertEquals(Integer.valueOf(73), result.value());
+            }
+        } finally {
+            workers.shutdownNow();
+            assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS));
+        }
+    }
+
+    @Test(timeout = 10_000)
+    public void provisionalWorkerMarksTheBatchCaller() {
+        ExecutorService workers = Executors.newFixedThreadPool(2);
+        try {
+            ProvisionalSampling.Unmemoizable result = assertThrows(ProvisionalSampling.Unmemoizable.class,
+                    () -> ProvisionalSampling.memoizable(() -> HydrologyForkJoin.invokeAll(List.of(
+                            () -> 1,
+                            () -> {
+                                ProvisionalSampling.mark();
+                                return 2;
+                            }), workers)));
+            assertEquals(List.of(1, 2), result.value());
+        } finally {
+            workers.shutdownNow();
+        }
+    }
+
     @Test
     public void blockedGridBatchNeverHelpsUnrelatedOwnerWork() throws Exception {
         ForkJoinPool pool = new ForkJoinPool(2, ForkJoinPool.defaultForkJoinWorkerThreadFactory,