From a35ecca7bd8b8081bb06cc179e668879455329e7 Mon Sep 17 00:00:00 2001 From: Brian Neumann-Fopiano Date: Mon, 28 Sep 2026 11:41:17 -0400 Subject: [PATCH] Group commit generation history - Semantic journal and saved-biome appends write without forcing; a shared flusher forces touched files within 2s outside every store lock - GenerationHistory.sync forces pending history at world save, engine shutdown, pregen completion and before activation promotion - A failed deferred force poisons the store the same way an uncertain append did - Cache the active epoch pack root instead of regex-validating it on every stage open - Tests pin append-without-force, sync barriers, failure poisoning and the save, shutdown and pregen hooks --- .../modded/command/ModdedPregenMethod.java | 5 + .../iris/generation/runtime/Engine.java | 4 + .../iris/generation/runtime/IrisEngine.java | 15 ++ .../iris/world/history/GenerationHistory.java | 22 +++ .../world/history/GenerationHistoryPaths.java | 12 +- .../GenerationHistoryRuntimeRouter.java | 5 + .../history/GenerationSemanticIndex.java | 22 ++- .../iris/world/history/SavedBiomeStore.java | 15 +- .../iris/world/pregen/AsyncPregenMethod.java | 14 ++ .../world/pregen/MedievalPregenMethod.java | 14 ++ .../world/storage/DeferredDurability.java | 138 +++++++++++++++ .../IrisEngineGenerationHistorySyncTest.java | 54 ++++++ .../runtime/IrisEngineWorldSaveTest.java | 1 + .../GenerationHistoryAdmissionTest.java | 21 ++- .../history/GenerationHistoryPathsTest.java | 15 ++ ...tionHistoryRuntimeRouterLifecycleTest.java | 78 +++++++++ .../GenerationSemanticBatchWriteTest.java | 11 +- .../GenerationSemanticJournalWriteTest.java | 75 ++++---- .../GenerationSemanticPointReadTest.java | 28 +-- .../GenerationUnpublishedHistoryTest.java | 10 +- .../history/SavedBiomeStoreBatchTest.java | 112 ++++++------ .../history/SavedBiomeStoreWriteTest.java | 97 +++++++++-- .../pregen/AsyncPregenHistorySyncTest.java | 84 +++++++++ .../world/storage/DeferredDurabilityTest.java | 161 ++++++++++++++++++ 24 files changed, 877 insertions(+), 136 deletions(-) create mode 100644 core/src/main/java/art/arcane/iris/world/storage/DeferredDurability.java create mode 100644 core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationHistorySyncTest.java create mode 100644 core/src/test/java/art/arcane/iris/world/pregen/AsyncPregenHistorySyncTest.java create mode 100644 core/src/test/java/art/arcane/iris/world/storage/DeferredDurabilityTest.java diff --git a/adapters/modded-common/src/main/java/art/arcane/iris/modded/command/ModdedPregenMethod.java b/adapters/modded-common/src/main/java/art/arcane/iris/modded/command/ModdedPregenMethod.java index 011fed5e3..058066888 100644 --- a/adapters/modded-common/src/main/java/art/arcane/iris/modded/command/ModdedPregenMethod.java +++ b/adapters/modded-common/src/main/java/art/arcane/iris/modded/command/ModdedPregenMethod.java @@ -116,6 +116,11 @@ public final class ModdedPregenMethod implements PregeneratorMethod { } ModdedIrisLog.info("Iris modded pregen done: dim={} completed={} peakInFlight={} finalLimit={}", runtime.worldIdentity(), completed.get(), inFlightPeak.get(), adaptiveLimit.get()); + try { + engine.syncGenerationHistory(); + } catch (RuntimeException e) { + ModdedIrisLog.error("Iris modded pregen could not force generation history to stable storage", e); + } if (deferFinalSaveIfRequested()) { return; } diff --git a/core/src/main/java/art/arcane/iris/generation/runtime/Engine.java b/core/src/main/java/art/arcane/iris/generation/runtime/Engine.java index 02e35e4a8..23846e0ae 100644 --- a/core/src/main/java/art/arcane/iris/generation/runtime/Engine.java +++ b/core/src/main/java/art/arcane/iris/generation/runtime/Engine.java @@ -250,6 +250,7 @@ public interface Engine extends DataProvider, Fallible, BlockUpdater, Renderer, default void save() { NativeStructureOwnershipStore.flush(this); + syncGenerationHistory(); getMantle().save(); getWorldManager().onSave(); saveEngineData(); @@ -259,6 +260,7 @@ public interface Engine extends DataProvider, Fallible, BlockUpdater, Renderer, default void saveNow() { NativeStructureOwnershipStore.flush(this); + syncGenerationHistory(); getMantle().saveAllNow(); saveEngineData(); } @@ -267,6 +269,8 @@ public interface Engine extends DataProvider, Fallible, BlockUpdater, Renderer, void saveEngineData(); + void syncGenerationHistory(); + default String getName() { return getDimension().getName(); } diff --git a/core/src/main/java/art/arcane/iris/generation/runtime/IrisEngine.java b/core/src/main/java/art/arcane/iris/generation/runtime/IrisEngine.java index 1a14334f6..bdce1e93b 100644 --- a/core/src/main/java/art/arcane/iris/generation/runtime/IrisEngine.java +++ b/core/src/main/java/art/arcane/iris/generation/runtime/IrisEngine.java @@ -1529,6 +1529,7 @@ public class IrisEngine implements Engine { private void scheduleWorldSave() { backgroundTasks.scheduleAdmittedTask(this, () -> NativeStructureOwnershipStore.flush(this)); + backgroundTasks.scheduleAdmittedTask(this, this::syncGenerationHistory); getMantle().save(); getWorldManager().onSave(); saveEngineData(); @@ -1539,6 +1540,20 @@ public class IrisEngine implements Engine { engineDataStore.saveEngineData(); } + @Override + public void syncGenerationHistory() { + GenerationHistoryRuntimeRouter router = getGenerationHistoryRuntimeRouter().orElse(null); + if (router == null) { + return; + } + try { + router.history().sync(); + } catch (IOException failure) { + throw new IllegalStateException("Unable to force Iris generation history to stable storage for " + + getWorld().name() + ".", failure); + } + } + @Override public void blockUpdatedMetric() { bud.incrementAndGet(); diff --git a/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java b/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java index 2b1df3d07..82a5d6c50 100644 --- a/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java +++ b/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java @@ -82,6 +82,27 @@ public final class GenerationHistory { return savedBiomes; } + public void sync() throws IOException { + IOException failure = null; + try { + semantics.sync(); + } catch (IOException semanticFailure) { + failure = semanticFailure; + } + try { + savedBiomes.sync(); + } catch (IOException biomeFailure) { + if (failure == null) { + failure = biomeFailure; + } else { + failure.addSuppressed(biomeFailure); + } + } + if (failure != null) { + throw failure; + } + } + public static GenerationHistory create( Path dimensionRoot, Path packSource, @@ -478,6 +499,7 @@ public final class GenerationHistory { private GenerationActivation promotePendingLocked( BoundarySignatureCapture signatureCapture, boolean startupRecovery, boolean validateSemantics ) throws IOException { + sync(); WorldChunkInventory inventory = recoverUnstoredClaims(startupRecovery); validateReferencedState(Optional.empty(), validateSemantics); Optional pending = store.pendingActivation(); diff --git a/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryPaths.java b/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryPaths.java index da0a54b27..7fe3dfc57 100644 --- a/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryPaths.java +++ b/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryPaths.java @@ -26,6 +26,7 @@ public final class GenerationHistoryPaths { private final Path legacyPackRoot; private final Path legacyMantleRoot; private final Path regionRoot; + private volatile PackRoot lastPackRoot; private GenerationHistoryPaths(Path dimensionRoot) { this.dimensionRoot = Objects.requireNonNull(dimensionRoot, "dimensionRoot") @@ -71,7 +72,13 @@ public final class GenerationHistoryPaths { } public Path packRoot(String epochId) { - return epochRoot(epochId).resolve(LEGACY_PACK_DIRECTORY_NAME); + PackRoot cached = lastPackRoot; + if (cached != null && cached.epochId().equals(epochId)) { + return cached.path(); + } + Path resolved = epochRoot(epochId).resolve(LEGACY_PACK_DIRECTORY_NAME); + lastPackRoot = new PackRoot(epochId, resolved); + return resolved; } public Path activationsRoot() { @@ -108,4 +115,7 @@ public final class GenerationHistoryPaths { public Path regionRoot() { return regionRoot; } + + private record PackRoot(String epochId, Path path) { + } } diff --git a/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouter.java b/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouter.java index f8107ddea..919d219c7 100644 --- a/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouter.java +++ b/core/src/main/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouter.java @@ -500,6 +500,11 @@ public final class GenerationHistoryRuntimeRouter implements AutoCloseable { } catch (Throwable detachFailure) { failure = detachFailure; } + try { + history.sync(); + } catch (Throwable syncFailure) { + failure = appendFailure(failure, syncFailure); + } try { biomes.close(); } catch (Throwable biomeFailure) { diff --git a/core/src/main/java/art/arcane/iris/world/history/GenerationSemanticIndex.java b/core/src/main/java/art/arcane/iris/world/history/GenerationSemanticIndex.java index 5bf84ddd2..b99888127 100644 --- a/core/src/main/java/art/arcane/iris/world/history/GenerationSemanticIndex.java +++ b/core/src/main/java/art/arcane/iris/world/history/GenerationSemanticIndex.java @@ -2,6 +2,7 @@ package art.arcane.iris.world.history; import art.arcane.iris.generation.hydrology.HydrologyFeatureType; import art.arcane.iris.spi.IrisLogging; +import art.arcane.iris.world.storage.DeferredDurability; import art.arcane.iris.world.storage.Durability; import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap; import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; @@ -93,6 +94,7 @@ public final class GenerationSemanticIndex { private final PointerPublisher pointerPublisher; private final CatalogPublisher catalogPublisher; private final boolean unpublished; + private final DeferredDurability durability; private long regionDecodeCount; private volatile IOException journalFailure; private long appendingRegionKey; @@ -118,6 +120,8 @@ public final class GenerationSemanticIndex { this.pointerPublisher = pointerPublisher; this.catalogPublisher = catalogPublisher; this.unpublished = unpublished; + this.durability = new DeferredDurability( + DeferredDurability.DEFAULT_FLUSH_DELAY_MILLIS, failure -> journalFailure = failure); } public static GenerationSemanticIndex load(Path dimensionRoot) throws IOException { @@ -265,7 +269,7 @@ public final class GenerationSemanticIndex { appendingBase = baseRegion; publicationLock.unlock(); try { - SemanticJournal.append(directory, regionX, regionZ, updates, unpublished); + SemanticJournal.append(directory, regionX, regionZ, updates, unpublished, durability); } catch (JournalAppendFailure failure) { journalFailure = failure; throw failure; @@ -289,6 +293,15 @@ public final class GenerationSemanticIndex { } } + public void sync() throws IOException { + try { + durability.sync(); + } catch (IOException failure) { + journalFailure = failure; + throw failure; + } + } + public int discardUnstoredClaims(WorldChunkInventory stored, Set activationIds) throws IOException { Objects.requireNonNull(stored, "stored chunks"); Set selected = Set.copyOf(activationIds); @@ -1940,7 +1953,8 @@ public final class GenerationSemanticIndex { int regionX, int regionZ, List claims, - boolean unpublished + boolean unpublished, + DeferredDurability durability ) throws IOException { Path file = directory.resolve(journalFileName(regionX, regionZ)); boolean created = !Files.exists(file, LinkOption.NOFOLLOW_LINKS); @@ -1974,10 +1988,10 @@ public final class GenerationSemanticIndex { } } if (!unpublished) { - Durability.force(channel); + durability.written(file); } if (created) { - RegionShard.forceDirectory(directory); + durability.linked(directory); } } catch (IOException failure) { try { diff --git a/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java b/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java index fd6615089..713dc7905 100644 --- a/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java +++ b/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java @@ -1,5 +1,6 @@ package art.arcane.iris.world.history; +import art.arcane.iris.world.storage.DeferredDurability; import art.arcane.iris.world.storage.Durability; import java.io.BufferedInputStream; @@ -52,6 +53,7 @@ public final class SavedBiomeStore { private final Path dimensionRoot; private final Path directory; private final boolean unpublished; + private final DeferredDurability durability; private final WriteStripe[] regionLocks = new WriteStripe[64]; private final LinkedHashMap regions = new LinkedHashMap<>(64, 0.75F, true); private final LinkedHashMap> chunks = new LinkedHashMap<>(128, 0.75F, true); @@ -61,6 +63,8 @@ public final class SavedBiomeStore { private SavedBiomeStore(Path dimensionRoot, boolean unpublished) throws IOException { this.dimensionRoot = Objects.requireNonNull(dimensionRoot, "dimensionRoot").toAbsolutePath().normalize(); this.unpublished = unpublished; + this.durability = new DeferredDurability( + DeferredDurability.DEFAULT_FLUSH_DELAY_MILLIS, failure -> writeFailure = failure); directory = this.dimensionRoot.resolve("iris/generation/biomes"); requireSafeAncestors(); for (int index = 0; index < regionLocks.length; index++) { @@ -227,7 +231,7 @@ public final class SavedBiomeStore { } } if (!unpublished) { - Durability.force(output.getChannel()); + durability.written(region.path); } appendCompleted = true; } catch (IOException failure) { @@ -269,6 +273,15 @@ public final class SavedBiomeStore { } } + public void sync() throws IOException { + try { + durability.sync(); + } catch (IOException failure) { + writeFailure = failure; + throw failure; + } + } + public int discardUnstoredClaims(WorldChunkInventory inventory, Set selectedActivations) throws IOException { Objects.requireNonNull(inventory, "inventory"); Set selected = Set.copyOf(selectedActivations); diff --git a/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java index b470b333d..1f909b6dd 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java @@ -779,6 +779,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { TimeUnit.SECONDS, () -> IrisLogging.warn("Async pregen is still draining outstanding chunks. " + metricsSnapshot()) ); + syncGenerationHistory(); mantleCleanup.close(MANTLE_CLEANUP_DRAIN_SECONDS, TimeUnit.SECONDS); flushAllRemainingChunks(); @@ -807,6 +808,19 @@ public class AsyncPregenMethod implements PregeneratorMethod { } } + private void syncGenerationHistory() { + Engine engine = resolveMetricsEngine(); + if (engine == null) { + return; + } + try { + engine.syncGenerationHistory(); + } catch (Throwable e) { + IrisLogging.reportError("Async pregen could not force generation history to stable storage for world " + + world.getName() + ".", e); + } + } + private boolean isCancelled() { return closing.get() || Thread.currentThread().isInterrupted(); } diff --git a/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java index 51cb6c07c..26687e469 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java @@ -112,6 +112,19 @@ public class MedievalPregenMethod implements PregeneratorMethod { return cached; } + private void syncGenerationHistory() { + Engine engine = resolveEngine(); + if (engine == null) { + return; + } + try { + engine.syncGenerationHistory(); + } catch (Throwable e) { + IrisLogging.reportError("Medieval pregen could not force generation history to stable storage for world " + + world.getName() + ".", e); + } + } + private void waitForChunks() { for (CompletableFuture i : futures) { try { @@ -179,6 +192,7 @@ public class MedievalPregenMethod implements PregeneratorMethod { if (prefetchPool != null) { prefetchPool.shutdownNow(); } + syncGenerationHistory(); unloadAndSaveAllChunks(true); chunkFlush.close(UNLOAD_TIMEOUT_SECONDS, TimeUnit.SECONDS); } finally { diff --git a/core/src/main/java/art/arcane/iris/world/storage/DeferredDurability.java b/core/src/main/java/art/arcane/iris/world/storage/DeferredDurability.java new file mode 100644 index 000000000..cd385083c --- /dev/null +++ b/core/src/main/java/art/arcane/iris/world/storage/DeferredDurability.java @@ -0,0 +1,138 @@ +package art.arcane.iris.world.storage; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; + +/** + * Group commit for append-only stores: writers record which files and directory entries they + * changed, and one background pass forces them shortly afterwards, outside every store lock. + * {@link #sync()} is the durability barrier for saves, shutdown and pregeneration completion. + */ +public final class DeferredDurability { + public static final long DEFAULT_FLUSH_DELAY_MILLIS = 2_000L; + private static final ScheduledThreadPoolExecutor FLUSHER = createFlusher(); + + private final Set files = ConcurrentHashMap.newKeySet(); + private final Set directories = ConcurrentHashMap.newKeySet(); + private final ReentrantLock flushLock = new ReentrantLock(); + private final AtomicBoolean scheduled = new AtomicBoolean(); + private final long flushDelayMillis; + private final Consumer backgroundFailure; + + public DeferredDurability(long flushDelayMillis, Consumer backgroundFailure) { + if (flushDelayMillis < 0L) { + throw new IllegalArgumentException("Flush delay must not be negative."); + } + this.flushDelayMillis = flushDelayMillis; + this.backgroundFailure = Objects.requireNonNull(backgroundFailure, "backgroundFailure"); + } + + public void written(Path file) { + if (!Durability.enabled()) { + return; + } + files.add(file); + schedule(); + } + + public void linked(Path directory) { + if (!Durability.enabled() || File.separatorChar == '\\') { + return; + } + directories.add(directory); + schedule(); + } + + public void sync() throws IOException { + flushLock.lock(); + try { + IOException failure = null; + for (Path file : files) { + files.remove(file); + try { + forceFile(file); + } catch (IOException fileFailure) { + failure = append(failure, fileFailure); + } + } + for (Path directory : directories) { + directories.remove(directory); + try { + forceDirectory(directory); + } catch (IOException directoryFailure) { + failure = append(failure, directoryFailure); + } + } + if (failure != null) { + throw failure; + } + } finally { + flushLock.unlock(); + } + } + + private void schedule() { + if (scheduled.compareAndSet(false, true)) { + FLUSHER.schedule(this::flushInBackground, flushDelayMillis, TimeUnit.MILLISECONDS); + } + } + + private void flushInBackground() { + scheduled.set(false); + try { + sync(); + } catch (IOException failure) { + backgroundFailure.accept(failure); + } catch (RuntimeException failure) { + backgroundFailure.accept(new IOException("Deferred durability sync failed unexpectedly.", failure)); + } + } + + private static void forceFile(Path file) throws IOException { + try (FileChannel channel = FileChannel.open(file, StandardOpenOption.WRITE)) { + Durability.force(channel); + } catch (NoSuchFileException replaced) { + // Compaction or rewrite already published this file's contents durably under another name. + } + } + + private static void forceDirectory(Path directory) throws IOException { + try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) { + Durability.force(channel); + } catch (UnsupportedOperationException failure) { + throw new IOException("Directory cannot be durability-synced: " + directory, failure); + } + } + + private static IOException append(IOException failure, IOException additional) { + if (failure == null) { + return additional; + } + failure.addSuppressed(additional); + return failure; + } + + private static ScheduledThreadPoolExecutor createFlusher() { + ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, runnable -> { + Thread thread = new Thread(runnable, "Iris Durability Flush"); + thread.setDaemon(true); + return thread; + }); + executor.setKeepAliveTime(30L, TimeUnit.SECONDS); + executor.allowCoreThreadTimeOut(true); + executor.setRemoveOnCancelPolicy(true); + return executor; + } +} diff --git a/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationHistorySyncTest.java b/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationHistorySyncTest.java new file mode 100644 index 000000000..d3af9c5a1 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationHistorySyncTest.java @@ -0,0 +1,54 @@ +package art.arcane.iris.generation.runtime; + +import art.arcane.iris.world.history.GenerationHistory; +import art.arcane.iris.world.history.GenerationHistoryRuntimeRouter; +import art.arcane.iris.world.IrisWorld; +import org.junit.Test; + +import java.io.IOException; +import java.util.Optional; + +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.doCallRealMethod; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class IrisEngineGenerationHistorySyncTest { + @Test + public void syncForcesTheAttachedHistory() throws Exception { + GenerationHistory history = mock(GenerationHistory.class); + IrisEngine engine = engineWith(history); + engine.syncGenerationHistory(); + verify(history).sync(); + } + + @Test + public void syncWithoutAnAttachedHistoryDoesNothing() { + IrisEngine engine = mock(IrisEngine.class); + when(engine.getGenerationHistoryRuntimeRouter()).thenReturn(Optional.empty()); + doCallRealMethod().when(engine).syncGenerationHistory(); + engine.syncGenerationHistory(); + } + + @Test + public void failedSyncSurfacesTheStorageFailure() throws Exception { + GenerationHistory history = mock(GenerationHistory.class); + IOException failure = new IOException("history force failed"); + doThrow(failure).when(history).sync(); + IrisEngine engine = engineWith(history); + when(engine.getWorld()).thenReturn(IrisWorld.builder().name("world").build()); + assertSame(failure, assertThrows(IllegalStateException.class, engine::syncGenerationHistory).getCause()); + } + + private static IrisEngine engineWith(GenerationHistory history) { + GenerationHistoryRuntimeRouter router = mock(GenerationHistoryRuntimeRouter.class); + when(router.history()).thenReturn(history); + IrisEngine engine = mock(IrisEngine.class); + when(engine.getGenerationHistoryRuntimeRouter()).thenReturn(Optional.of(router)); + doCallRealMethod().when(engine).syncGenerationHistory(); + return engine; + } +} diff --git a/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineWorldSaveTest.java b/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineWorldSaveTest.java index b824cfdec..418ee0d09 100644 --- a/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineWorldSaveTest.java +++ b/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineWorldSaveTest.java @@ -68,6 +68,7 @@ public class IrisEngineWorldSaveTest { fixture.release().countDown(); close.get(5, TimeUnit.SECONDS).requireComplete("close"); assertEquals(0, fixture.sessions().activeLeases()); + verify(fixture.engine()).syncGenerationHistory(); } finally { fixture.release().countDown(); } diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java index 627a76b46..b3eddb747 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java @@ -193,7 +193,7 @@ public class GenerationHistoryAdmissionTest { } @Test - public void queuedClaimsShareOneForceAndRetainIndividualValidation() throws Exception { + public void queuedClaimsShareOneAppendPerRegionWithoutForcingAndRetainIndividualValidation() throws Exception { GenerationHistory history = history(); GenerationAdmission.RuntimeLease runtime = history.retainRuntime(); List stages = new ArrayList<>(); @@ -207,6 +207,7 @@ public class GenerationHistoryAdmissionTest { workers.add(thread); return thread; }); + AtomicInteger appends = new AtomicInteger(); AtomicInteger forces = new AtomicInteger(); List> results = new ArrayList<>(); try { @@ -217,7 +218,7 @@ public class GenerationHistoryAdmissionTest { ? ChunkGenerationSemantics.builder(1, 0, stage.activation().activationId()) .addSurfaceBiome("iris:conflict").seal().build() : claim(stage); - results.add(executor.submit(() -> persistCountingForces(history, stage, update, forces))); + results.add(executor.submit(() -> persistCounting(history, stage, update, appends, forces))); awaitBlockedOnHistory(history, workers, index + 1); } stages.get(5).close(); @@ -232,7 +233,8 @@ public class GenerationHistoryAdmissionTest { .getCause() instanceof IllegalStateException); assertTrue(results.get(6).get(5, TimeUnit.SECONDS)); assertTrue(results.get(7).get(5, TimeUnit.SECONDS)); - assertEquals(2, forces.get()); + assertEquals(2, appends.get()); + assertEquals(0, forces.get()); GenerationSemanticIndex loaded = GenerationSemanticIndex.loadRequired(history.paths().dimensionRoot()); assertEquals(5, loaded.recordCount()); assertEquals(claim(stages.getFirst()), loaded.get(1, 0).orElseThrow()); @@ -348,8 +350,9 @@ public class GenerationHistoryAdmissionTest { } } - private static boolean persistCountingForces(GenerationHistory history, GenerationHistory.GenerationStage stage, - ChunkGenerationSemantics claim, AtomicInteger forces) throws Exception { + private static boolean persistCounting(GenerationHistory history, GenerationHistory.GenerationStage stage, + ChunkGenerationSemantics claim, AtomicInteger appends, + AtomicInteger forces) throws Exception { try (MockedStatic ignored = mockStatic(FileChannel.class, invocation -> { FileChannel source = (FileChannel) invocation.callRealMethod(); Path path = invocation.getArgument(0); @@ -357,6 +360,7 @@ public class GenerationHistoryAdmissionTest { || !path.getFileName().toString().endsWith(".iswal")) { return source; } + appends.incrementAndGet(); FileChannel intercepted = mock(FileChannel.class, delegatesTo(source)); doAnswer(force -> { forces.incrementAndGet(); @@ -396,14 +400,13 @@ public class GenerationHistoryAdmissionTest { return channel; } FileChannel intercepted = mock(FileChannel.class, delegatesTo(channel)); - doAnswer(force -> { + doAnswer(write -> { entered.countDown(); if (!release.await(5, TimeUnit.SECONDS)) { throw new IOException("Timed out waiting to resume journal persistence"); } - channel.force(true); - return null; - }).when(intercepted).force(true); + return channel.write((ByteBuffer) write.getArgument(0)); + }).when(intercepted).write(any(ByteBuffer.class)); return intercepted; })) { return history.claimGeneratedSemantics(stage, claim); diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryPathsTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryPathsTest.java index cefee05d3..c9432804a 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryPathsTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryPathsTest.java @@ -5,6 +5,7 @@ import org.junit.Test; import java.nio.file.Path; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThrows; public final class GenerationHistoryPathsTest { @@ -34,4 +35,18 @@ public final class GenerationHistoryPathsTest { assertThrows(IllegalArgumentException.class, () -> paths.epochRoot("../escape")); assertThrows(IllegalArgumentException.class, () -> paths.activationRoot(0L)); } + + @Test + public void repeatedPackRootsStayValidatedAndFollowTheRequestedEpoch() { + GenerationHistoryPaths paths = GenerationHistoryPaths.forDimension(Path.of("build", "world")); + String first = "a".repeat(64); + String second = "b".repeat(64); + + Path firstRoot = paths.packRoot(first); + assertSame(firstRoot, paths.packRoot(first)); + assertEquals(paths.epochRoot(second).resolve("pack"), paths.packRoot(second)); + assertEquals(firstRoot, paths.packRoot(first)); + assertThrows(IllegalArgumentException.class, () -> paths.packRoot("../escape")); + assertThrows(NullPointerException.class, () -> paths.packRoot(null)); + } } diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouterLifecycleTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouterLifecycleTest.java index 524ce19e6..364f43c63 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouterLifecycleTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryRuntimeRouterLifecycleTest.java @@ -1,17 +1,24 @@ package art.arcane.iris.world.history; import art.arcane.iris.generation.runtime.IrisEngine; +import art.arcane.iris.world.storage.Durability; import org.junit.Test; +import org.mockito.MockedStatic; +import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; +import java.util.List; +import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; @@ -21,8 +28,12 @@ import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import static org.mockito.AdditionalAnswers.delegatesTo; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -67,6 +78,73 @@ public final class GenerationHistoryRuntimeRouterLifecycleTest extends Generatio router.close(); } + @Test + public void closeForcesDeferredSemanticAndBiomeWrites() throws Exception { + Path world = temporaryFolder.newFolder("router-close-sync-world").toPath(); + GenerationHistory history = createHistory(world, createPack("router-close-sync-pack", "alpha")); + IrisEngine engine = mock(IrisEngine.class); + FakeRuntimeFactory runtimes = new FakeRuntimeFactory(); + IrisEngine.GenerationRuntimeBinding active = runtimes.binding(history, history.activeActivation()); + when(engine.getActiveGenerationRuntimeBinding()).thenReturn(active); + GenerationHistoryRuntimeRouter router = GenerationHistoryRuntimeRouter.attach( + engine, history, (ignored, x, z) -> signature(x, z), runtimes); + Path journal = history.paths().generationRoot().resolve("semantics/r.0.0.iswal"); + Path biomes = history.paths().generationRoot().resolve("biomes/r.0.0.ibio"); + Map forces = new ConcurrentHashMap<>(); + try (MockedStatic durability = mockStatic(Durability.class, CALLS_REAL_METHODS); + MockedStatic ignored = countingForces(Set.of(journal, biomes), forces)) { + durability.when(Durability::enabled).thenReturn(true); + try (GenerationAdmission.RuntimeLease runtime = history.retainRuntime(); + GenerationHistory.GenerationStage stage = history.openStage(1, 0)) { + long activation = stage.activation().activationId(); + assertTrue(history.claimGeneratedSemantics(stage, ChunkGenerationSemantics.builder(1, 0, activation) + .addSurfaceBiome("iris:forest").seal().build())); + assertTrue(history.savedBiomes().claimAndPersist(savedBiomes(1, activation))); + } + assertEquals(0, forced(forces, journal)); + assertEquals(0, forced(forces, biomes)); + router.close(); + assertEquals(1, forced(forces, journal)); + assertEquals(1, forced(forces, biomes)); + } + verify(engine).detachGenerationHistoryRuntimeRouter(router); + } + + private static int forced(Map forces, Path path) { + AtomicInteger count = forces.get(path); + return count == null ? 0 : count.get(); + } + + private static MockedStatic countingForces(Set watched, Map forces) { + return mockStatic(FileChannel.class, invocation -> { + FileChannel source = (FileChannel) invocation.callRealMethod(); + if (invocation.getMethod().getParameterCount() != 2 || !watched.contains(invocation.getArgument(0))) { + return source; + } + Path path = invocation.getArgument(0); + FileChannel intercepted = mock(FileChannel.class, delegatesTo(source)); + doAnswer(force -> { + forces.computeIfAbsent(path, unused -> new AtomicInteger()).incrementAndGet(); + source.force(true); + return null; + }).when(intercepted).force(true); + return intercepted; + }); + } + + private static SavedBiomeChunk savedBiomes(int chunkX, long activation) { + SavedBiomeChunk.Cell cell = new SavedBiomeChunk.Cell(activation, "biome/main", "region/main"); + SavedBiomeChunk.Column column = new SavedBiomeChunk.Column(cell, cell, + List.of(new SavedBiomeChunk.Span(-64, 320, cell))); + SavedBiomeChunk.Builder builder = SavedBiomeChunk.builder(new SavedBiomeChunk.Header(chunkX, 0, activation, -64, 384)); + for (int z = 0; z < 16; z++) { + for (int x = 0; x < 16; x++) { + builder.column(x, z, column); + } + } + return builder.build(); + } + @Test public void closeRejectsNewRoutesAndAwaitsExistingLeases() throws Exception { Path world = temporaryFolder.newFolder("router-close-await-world").toPath(); diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticBatchWriteTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticBatchWriteTest.java index 6f435686c..2fb2e65bd 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticBatchWriteTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticBatchWriteTest.java @@ -95,19 +95,18 @@ public class GenerationSemanticBatchWriteTest { GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root); List claims = List.of( pending(0, "first"), pending(32, "failed"), pending(64, "last")); - IOException failure = new IOException("Middle region force failed"); - AtomicInteger forces = new AtomicInteger(); + IOException failure = new IOException("Middle region write failed"); + AtomicInteger writes = new AtomicInteger(); try (MockedStatic ignored = channels((path, source, intercepted) -> { if (!path.getFileName().toString().equals("r.1.0.iswal")) { return; } doAnswer(invocation -> { - if (forces.incrementAndGet() == 1) { + if (writes.incrementAndGet() == 1) { throw failure; } - source.force(true); - return null; - }).when(intercepted).force(true); + return source.write((ByteBuffer) invocation.getArgument(0)); + }).when(intercepted).write(any(ByteBuffer.class)); })) { index.claimAndPersistBatch(claims); } diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticJournalWriteTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticJournalWriteTest.java index e3743c60a..239531c3a 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticJournalWriteTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticJournalWriteTest.java @@ -92,44 +92,64 @@ public class GenerationSemanticJournalWriteTest { } @Test - public void failedForceRestoresBytesAndLeavesTheClaimRetryable() throws Exception { + public void claimReturnsBeforeTheJournalIsForcedAndSyncForcesIt() throws Exception { Path root = temporaryFolder.newFolder().toPath(); GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root); index.claimAndPersist(claim(0, "first")); - Path journal = index.storageDirectory().resolve("r.0.0.iswal"); - byte[] original = Files.readAllBytes(journal); - IOException failure = new IOException("Journal force failed"); + index.sync(); AtomicInteger forces = new AtomicInteger(); try (MockedStatic ignored = journalChannels((source, intercepted) -> doAnswer(invocation -> { - if (forces.incrementAndGet() == 1) { - throw failure; - } + forces.incrementAndGet(); source.force(true); return null; }).when(intercepted).force(true))) { - assertSame(failure, assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "second")))); + assertTrue(index.claimAndPersist(claim(1, "second"))); + assertEquals(0, forces.get()); + assertEquals(Optional.of(claim(1, "second")), index.get(1, 0)); + index.sync(); + assertEquals(1, forces.get()); + index.sync(); + assertEquals(1, forces.get()); } - assertEquals(2, forces.get()); - assertArrayEquals(original, Files.readAllBytes(journal)); - assertTrue(index.get(1, 0).isEmpty()); - assertTrue(index.claimAndPersist(claim(1, "second"))); assertEquals(Optional.of(claim(1, "second")), GenerationSemanticIndex.loadRequired(root).get(1, 0)); } + @Test + public void failedSyncRejectsFurtherMutationButKeepsWrittenClaims() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root); + index.claimAndPersist(claim(0, "first")); + IOException failure = new IOException("Deferred journal force failed"); + try (MockedStatic ignored = journalChannels((source, intercepted) -> + doAnswer(invocation -> { + throw failure; + }).when(intercepted).force(true))) { + assertSame(failure, assertThrows(IOException.class, index::sync)); + } + assertSame(failure, assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "second"))).getCause()); + assertThrows(IllegalStateException.class, () -> index.get(0, 0)); + GenerationSemanticIndex replayed = GenerationSemanticIndex.loadRequired(root); + assertEquals(Optional.of(claim(0, "first")), replayed.get(0, 0)); + assertTrue(replayed.get(1, 0).isEmpty()); + } + @Test public void rollbackFailurePreservesCausesAndRejectsFurtherMutation() throws Exception { Path root = temporaryFolder.newFolder().toPath(); GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root); index.claimAndPersist(claim(0, "first")); - IOException failure = new IOException("Journal force failed"); + IOException failure = new IOException("Journal write failed"); IOException rollback = new IOException("Rollback force failed"); - AtomicInteger forces = new AtomicInteger(); IOException reported; - try (MockedStatic ignored = journalChannels((source, intercepted) -> - doAnswer(invocation -> { - throw forces.incrementAndGet() == 1 ? failure : rollback; - }).when(intercepted).force(true))) { + try (MockedStatic ignored = journalChannels((source, intercepted) -> { + doAnswer(invocation -> { + throw failure; + }).when(intercepted).write(any(ByteBuffer.class)); + doAnswer(invocation -> { + throw rollback; + }).when(intercepted).force(true); + })) { reported = assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "failed"))); } assertSame(failure, reported.getCause()); @@ -176,7 +196,7 @@ public class GenerationSemanticJournalWriteTest { } @Test - public void newJournalDirectoryFailureRollsBackBeforeRetry() throws Exception { + public void newJournalDirectoryIsForcedBySyncAndItsFailureRejectsFurtherMutation() throws Exception { assumeTrue(File.separatorChar != '\\'); Path root = temporaryFolder.newFolder().toPath(); GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root); @@ -190,20 +210,17 @@ public class GenerationSemanticJournalWriteTest { } FileChannel intercepted = mock(FileChannel.class, delegatesTo(source)); doAnswer(force -> { - if (directoryForces.incrementAndGet() == 1) { - throw failure; - } - source.force(true); - return null; + directoryForces.incrementAndGet(); + throw failure; }).when(intercepted).force(true); return intercepted; })) { - assertSame(failure, assertThrows(IOException.class, () -> index.claimAndPersist(claim(0, "first")))); + assertTrue(index.claimAndPersist(claim(0, "first"))); + assertEquals(0, directoryForces.get()); + assertSame(failure, assertThrows(IOException.class, index::sync)); + assertEquals(1, directoryForces.get()); } - assertEquals(2, directoryForces.get()); - assertEquals(0L, Files.size(index.storageDirectory().resolve("r.0.0.iswal"))); - assertTrue(index.get(0, 0).isEmpty()); - assertTrue(index.claimAndPersist(claim(0, "first"))); + assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "second"))); assertEquals(Optional.of(claim(0, "first")), GenerationSemanticIndex.loadRequired(root).get(0, 0)); } diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticPointReadTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticPointReadTest.java index e2ce86e0e..f6ab222c5 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticPointReadTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationSemanticPointReadTest.java @@ -6,6 +6,7 @@ import org.junit.rules.TemporaryFolder; import org.mockito.MockedStatic; import java.io.IOException; +import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; @@ -29,6 +30,7 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.AdditionalAnswers.delegatesTo; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; @@ -69,7 +71,7 @@ public class GenerationSemanticPointReadTest { } @Test - public void newRegionRemainsAbsentUntilItsFirstForceCompletes() throws Exception { + public void newRegionRemainsAbsentUntilItsFirstAppendCompletes() throws Exception { GenerationSemanticIndex index = index(); ExecutorService reader = Executors.newSingleThreadExecutor(); try (AppendGate gate = append(() -> index.claimAndPersist(claim(-32, 1)), Failure.NONE)) { @@ -233,7 +235,7 @@ public class GenerationSemanticPointReadTest { private static AppendGate append(Callable operation, Failure mode) throws InterruptedException { AppendGate gate = new AppendGate(operation, mode); gate.worker.start(); - assertTrue(gate.forcing.await(5, TimeUnit.SECONDS)); + assertTrue(gate.appending.await(5, TimeUnit.SECONDS)); return gate; } @@ -250,7 +252,7 @@ public class GenerationSemanticPointReadTest { } private static final class AppendGate implements AutoCloseable { - private final CountDownLatch forcing = new CountDownLatch(1); + private final CountDownLatch appending = new CountDownLatch(1); private final CountDownLatch release = new CountDownLatch(1); private final AtomicReference failure = new AtomicReference<>(); private final IOException originalFailure = new IOException("Semantic journal failed"); @@ -266,24 +268,24 @@ public class GenerationSemanticPointReadTest { return source; } FileChannel intercepted = mock(FileChannel.class, delegatesTo(source)); - AtomicInteger forces = new AtomicInteger(); + AtomicInteger writes = new AtomicInteger(); doAnswer(call -> { - int count = forces.incrementAndGet(); - if (count == 1) { - forcing.countDown(); + if (writes.incrementAndGet() == 1) { + appending.countDown(); if (!release.await(10, TimeUnit.SECONDS)) { - throw new IOException("Timed out waiting to resume semantic force"); + throw new IOException("Timed out waiting to resume semantic append"); } if (mode == Failure.ROLLBACK_CONFIRMED || mode == Failure.ROLLBACK_UNCERTAIN) { throw originalFailure; } } - if (count == 2 && mode == Failure.ROLLBACK_UNCERTAIN) { + return source.write((ByteBuffer) call.getArgument(0)); + }).when(intercepted).write(any(ByteBuffer.class)); + if (mode == Failure.ROLLBACK_UNCERTAIN) { + doAnswer(call -> { throw new IOException("Semantic rollback failed"); - } - source.force(true); - return null; - }).when(intercepted).force(true); + }).when(intercepted).force(true); + } if (mode == Failure.CLOSE_UNCERTAIN) { doAnswer(call -> { source.close(); diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationUnpublishedHistoryTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationUnpublishedHistoryTest.java index 4bc16b410..4230c9e84 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationUnpublishedHistoryTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationUnpublishedHistoryTest.java @@ -46,6 +46,7 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup GenerationHistory reopened = GenerationHistory.open(root); assertEquals(biomes(1), reopened.savedBiomes().get(1, 0).orElseThrow()); assertBiomeAppendForces(reopened.savedBiomes(), root, 2, 1); + assertEquals(biomes(2), GenerationHistory.open(root).savedBiomes().get(2, 0).orElseThrow()); } @Test @@ -61,7 +62,10 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup GenerationSemanticIndex reopened = GenerationSemanticIndex.loadRequired(root); assertEquals(semantics(1), reopened.get(1, 0).orElseThrow()); try (MockedStatic forces = mockStatic(Durability.class, CALLS_REAL_METHODS)) { + forces.when(Durability::enabled).thenReturn(true); assertTrue(reopened.claimAndPersist(semantics(2))); + forces.verify(() -> Durability.force(any(FileChannel.class)), times(0)); + reopened.sync(); forces.verify(() -> Durability.force(any(FileChannel.class)), times(1)); } reopened.compactJournals(); @@ -91,7 +95,7 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup assertTrue(store.claimAndPersist(biomes(1))); } - private static void assertBiomeAppendForces(SavedBiomeStore store, Path root, int x, int count) throws Exception { + private static void assertBiomeAppendForces(SavedBiomeStore store, Path root, int x, int syncForces) throws Exception { try (MockedStatic durability = mockStatic(Durability.class, CALLS_REAL_METHODS); RandomAccessFile actual = new RandomAccessFile(region(root).toFile(), "rw")) { durability.when(Durability::enabled).thenReturn(true); @@ -101,7 +105,9 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup (file, context) -> doReturn(channel).when(file).getChannel())) { assertTrue(store.claimAndPersist(biomes(x))); } - verify(channel, times(count)).force(true); + verify(channel, times(0)).force(true); + store.sync(); + durability.verify(() -> Durability.force(any(FileChannel.class)), times(syncForces)); } } diff --git a/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreBatchTest.java b/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreBatchTest.java index 3128fc873..aaf2fe9fd 100644 --- a/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreBatchTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreBatchTest.java @@ -60,9 +60,7 @@ public class SavedBiomeStoreBatchTest { List claims = List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L)); try { List> results = enqueue(store, root, control, workers, claims); - assertTrue(control.forcing.await(5, TimeUnit.SECONDS)); - assertEquals(3, control.maximumGathered.get()); - assertTrue(control.writes.get() > claims.size()); + assertTrue(control.writing.await(5, TimeUnit.SECONDS)); for (SavedBiomeChunk claim : claims) { assertTrue(store.cached(claim.chunkX(), claim.chunkZ()).isEmpty()); } @@ -72,7 +70,9 @@ public class SavedBiomeStoreBatchTest { assertTrue(completed.persisted()); assertNull(completed.failure()); } - assertEquals(1, control.forces.get()); + assertEquals(3, control.maximumGathered.get()); + assertTrue(control.writes.get() > claims.size()); + assertEquals(0, control.forces.get()); SavedBiomeStore reopened = SavedBiomeStore.open(root); for (SavedBiomeChunk claim : claims) { assertEquals(claim, reopened.get(claim.chunkX(), claim.chunkZ()).orElseThrow()); @@ -94,6 +94,7 @@ public class SavedBiomeStoreBatchTest { BatchControl control = new BatchControl(); control.maximumWriteBytes = 11; control.writeFailure = new IOException("Partial gathered append failed"); + control.failingWrite = 2; control.release.countDown(); ExecutorService workers = Executors.newFixedThreadPool(4); List claims = List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L)); @@ -167,7 +168,7 @@ public class SavedBiomeStoreBatchTest { } @Test - public void queuedClaimsShareOneForceAndPublishAfterItCompletes() throws Exception { + public void queuedClaimsShareOneAppendAndPublishAfterItCompletes() throws Exception { Path root = temporaryFolder.newFolder().toPath(); SavedBiomeStore store = SavedBiomeStore.open(root); store.claimAndPersist(chunk(0, 3L)); @@ -177,7 +178,7 @@ public class SavedBiomeStoreBatchTest { List> results = enqueue(store, root, control, workers, List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L), chunk(4, 3L), chunk(5, 3L), chunk(6, 3L), chunk(7, 3L), chunk(8, 3L))); - assertTrue(control.forcing.await(5, TimeUnit.SECONDS)); + assertTrue(control.writing.await(5, TimeUnit.SECONDS)); for (int index = 1; index <= 8; index++) { assertTrue(store.cached(index, 0).isEmpty()); assertFalse(results.get(index - 1).isDone()); @@ -188,7 +189,8 @@ public class SavedBiomeStoreBatchTest { assertNull(claim.failure); assertTrue(claim.persisted); } - assertEquals(1, control.forces.get()); + assertEquals(1, control.writes.get()); + assertEquals(0, control.forces.get()); SavedBiomeStore reopened = SavedBiomeStore.open(root); for (int index = 1; index <= 8; index++) { assertEquals(chunk(index, 3L), reopened.get(index, 0).orElseThrow()); @@ -201,19 +203,20 @@ public class SavedBiomeStoreBatchTest { } @Test - public void batchForceFailureRollsBackEveryNewClaimButPreservesExistingDuplicates() throws Exception { + public void batchWriteFailureRollsBackEveryNewClaimButPreservesExistingDuplicates() throws Exception { Path root = temporaryFolder.newFolder().toPath(); SavedBiomeStore store = SavedBiomeStore.open(root); store.claimAndPersist(chunk(0, 3L)); Path region = root.resolve("iris/generation/biomes/r.0.0.ibio"); byte[] original = Files.readAllBytes(region); BatchControl control = new BatchControl(); - control.failure = new IOException("Batch force failed"); + control.writeFailure = new IOException("Batch write failed"); + control.failingWrite = 1; ExecutorService workers = Executors.newFixedThreadPool(4); try { List> results = enqueue(store, root, control, workers, List.of(chunk(0, 3L), chunk(1, 3L), chunk(1, 3L), chunk(2, 3L))); - assertTrue(control.forcing.await(5, TimeUnit.SECONDS)); + assertTrue(control.writing.await(5, TimeUnit.SECONDS)); control.release.countDown(); ClaimResult existing = results.getFirst().get(5, TimeUnit.SECONDS); assertFalse(existing.persisted); @@ -221,9 +224,9 @@ public class SavedBiomeStoreBatchTest { for (int index = 1; index < results.size(); index++) { ClaimResult result = results.get(index).get(5, TimeUnit.SECONDS); assertFalse(result.persisted); - assertSame(control.failure, result.failure); + assertSame(control.writeFailure, result.failure); } - assertEquals(2, control.forces.get()); + assertEquals(1, control.forces.get()); assertArrayEquals(original, Files.readAllBytes(region)); assertTrue(store.cached(1, 0).isEmpty()); assertTrue(store.claimAndPersist(chunk(1, 3L))); @@ -245,7 +248,7 @@ public class SavedBiomeStoreBatchTest { try { List> results = enqueue(store, root, control, workers, List.of(chunk(1, 3L), chunk(1, 3L), chunk(1, 4L))); - assertTrue(control.forcing.await(5, TimeUnit.SECONDS)); + assertTrue(control.writing.await(5, TimeUnit.SECONDS)); control.release.countDown(); int winners = 0; SavedBiomeChunk actual = null; @@ -257,7 +260,8 @@ public class SavedBiomeStoreBatchTest { } } assertEquals(1, winners); - assertEquals(1, control.forces.get()); + assertEquals(1, control.writes.get()); + assertEquals(0, control.forces.get()); assertEquals(actual, SavedBiomeStore.open(root).get(1, 0).orElseThrow()); for (int index = 0; index < results.size(); index++) { ClaimResult result = results.get(index).get(5, TimeUnit.SECONDS); @@ -277,12 +281,13 @@ public class SavedBiomeStoreBatchTest { SavedBiomeStore store = SavedBiomeStore.open(root); store.claimAndPersist(chunk(0, 3L)); BatchControl control = new BatchControl(); - control.uncheckedFailure = new IllegalStateException("Batch force failed unexpectedly"); + control.uncheckedFailure = new IllegalStateException("Batch write failed unexpectedly"); + control.failingWrite = 1; ExecutorService workers = Executors.newFixedThreadPool(4); try { List> results = enqueue(store, root, control, workers, List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L), chunk(4, 3L))); - assertTrue(control.forcing.await(5, TimeUnit.SECONDS)); + assertTrue(control.writing.await(5, TimeUnit.SECONDS)); Object writingRegion = regionLock(store, 0, 0); for (int region = 1; region <= 132; region++) { if (regionLock(store, region << 5, 0) != writingRegion) { @@ -295,7 +300,8 @@ public class SavedBiomeStoreBatchTest { assertFalse(result.persisted); assertSame(control.uncheckedFailure, rootCause(result.failure)); } - assertEquals(1, control.forces.get()); + assertEquals(1, control.writes.get()); + assertEquals(0, control.forces.get()); IOException retry = assertThrows(IOException.class, () -> store.claimAndPersist(chunk(1, 3L))); assertSame(control.uncheckedFailure, rootCause(retry)); IOException read = assertThrows(IOException.class, () -> store.get(1, 0)); @@ -348,45 +354,43 @@ public class SavedBiomeStoreBatchTest { try (RandomAccessFile actual = new RandomAccessFile( root.resolve("iris/generation/biomes/r.0.0.ibio").toFile(), "rw")) { FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel())); - if (control.maximumWriteBytes > 0) { - doAnswer(invocation -> { - ByteBuffer[] records = invocation.getArgument(0); - int offset = invocation.getArgument(1); - int count = invocation.getArgument(2); - control.maximumGathered.accumulateAndGet(count, Math::max); - int write = control.writes.incrementAndGet(); - if (control.zeroFirstWrite && write == 1) { - return 0L; - } - if (control.writeFailure != null && write == 2) { - throw control.writeFailure; - } - long written = 0L; - for (int index = offset; index < offset + count && written < control.maximumWriteBytes; index++) { - ByteBuffer record = records[index]; - int previousLimit = record.limit(); - record.limit(record.position() + Math.min(record.remaining(), control.maximumWriteBytes - (int) written)); - try { - written += actual.getChannel().write(record); - } finally { - record.limit(previousLimit); - } - } - return written; - }).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt()); - } doAnswer(invocation -> { - int force = control.forces.incrementAndGet(); - if (force == 1) { - control.forcing.countDown(); + ByteBuffer[] records = invocation.getArgument(0); + int offset = invocation.getArgument(1); + int count = invocation.getArgument(2); + control.maximumGathered.accumulateAndGet(count, Math::max); + int write = control.writes.incrementAndGet(); + if (write == 1) { + control.writing.countDown(); assertTrue(control.release.await(5, TimeUnit.SECONDS)); - if (control.failure != null) { - throw control.failure; - } - if (control.uncheckedFailure != null) { - throw control.uncheckedFailure; + } + if (write == control.failingWrite && control.uncheckedFailure != null) { + throw control.uncheckedFailure; + } + if (write == control.failingWrite && control.writeFailure != null) { + throw control.writeFailure; + } + if (control.zeroFirstWrite && write == 1) { + return 0L; + } + if (control.maximumWriteBytes <= 0) { + return actual.getChannel().write(records, offset, count); + } + long written = 0L; + for (int index = offset; index < offset + count && written < control.maximumWriteBytes; index++) { + ByteBuffer record = records[index]; + int previousLimit = record.limit(); + record.limit(record.position() + Math.min(record.remaining(), control.maximumWriteBytes - (int) written)); + try { + written += actual.getChannel().write(record); + } finally { + record.limit(previousLimit); } } + return written; + }).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt()); + doAnswer(invocation -> { + control.forces.incrementAndGet(); actual.getChannel().force(true); return null; }).when(channel).force(true); @@ -426,15 +430,15 @@ public class SavedBiomeStoreBatchTest { } private static final class BatchControl { - private final CountDownLatch forcing = new CountDownLatch(1); + private final CountDownLatch writing = new CountDownLatch(1); private final CountDownLatch release = new CountDownLatch(1); private final AtomicInteger forces = new AtomicInteger(); private int maximumWriteBytes; private boolean zeroFirstWrite; + private int failingWrite; private IOException writeFailure; private final AtomicInteger writes = new AtomicInteger(); private final AtomicInteger maximumGathered = new AtomicInteger(); - private IOException failure; private RuntimeException uncheckedFailure; } diff --git a/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java b/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java index 4a88b8b32..d7eaa570e 100644 --- a/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java @@ -5,6 +5,7 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; @@ -21,7 +22,7 @@ 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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.CRC32; import java.util.zip.Deflater; import java.util.zip.DeflaterOutputStream; @@ -41,6 +42,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; @@ -112,40 +114,41 @@ public class SavedBiomeStoreWriteTest { } @Test - public void appendWritesOneFrameAndPublishesOnlyAfterForce() throws Exception { + public void appendWritesOneFrameAndPublishesOnceTheWriteCompletes() throws Exception { Path root = temporaryFolder.newFolder().toPath(); SavedBiomeStore store = SavedBiomeStore.open(root); store.claimAndPersist(chunk(0, 3L)); - CountDownLatch forcing = new CountDownLatch(1); + CountDownLatch writing = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); ExecutorService workers = Executors.newSingleThreadExecutor(); try { - Future writing = workers.submit(() -> { + Future claimed = workers.submit(() -> { try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) { FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel())); doAnswer(invocation -> { - forcing.countDown(); + writing.countDown(); await(release); - actual.getChannel().force(true); - return null; - }).when(channel).force(true); + return actual.getChannel().write((ByteBuffer[]) invocation.getArgument(0), + (int) invocation.getArgument(1), (int) invocation.getArgument(2)); + }).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt()); try (MockedConstruction files = mockConstruction(RandomAccessFile.class, withSettings().defaultAnswer(delegatesTo(actual)), (file, context) -> doReturn(channel).when(file).getChannel())) { boolean result = store.claimAndPersist(chunk(1, 3L)); assertEquals(1, files.constructed().size()); verify(channel).write(any(ByteBuffer[].class), anyInt(), anyInt()); + verify(channel, never()).force(true); verify(files.constructed().getFirst(), never()).write(any(byte[].class)); verify(files.constructed().getFirst(), never()).writeInt(anyInt()); return result; } } }); - assertTrue(forcing.await(5, TimeUnit.SECONDS)); + assertTrue(writing.await(5, TimeUnit.SECONDS)); assertTrue(store.cached(1, 0).isEmpty()); - assertFalse(writing.isDone()); + assertFalse(claimed.isDone()); release.countDown(); - assertTrue(writing.get(5, TimeUnit.SECONDS)); + assertTrue(claimed.get(5, TimeUnit.SECONDS)); assertEquals(chunk(1, 3L), store.cached(1, 0).orElseThrow()); assertEquals(chunk(1, 3L), SavedBiomeStore.open(root).get(1, 0).orElseThrow()); } finally { @@ -155,21 +158,63 @@ public class SavedBiomeStoreWriteTest { } @Test - public void failedForceRollsBackWithoutPublishingAndRetainsRollbackFailure() throws Exception { + public void syncForcesAppendedRegionsOnce() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + store.claimAndPersist(chunk(0, 3L)); + store.sync(); + AtomicInteger forces = new AtomicInteger(); + try (MockedStatic ignored = regionChannels(root, (source, intercepted) -> + doAnswer(invocation -> { + forces.incrementAndGet(); + source.force(true); + return null; + }).when(intercepted).force(true))) { + assertTrue(store.claimAndPersist(chunk(1, 3L))); + assertTrue(store.claimAndPersist(chunk(2, 3L))); + assertEquals(0, forces.get()); + store.sync(); + assertEquals(1, forces.get()); + store.sync(); + assertEquals(1, forces.get()); + } + assertEquals(chunk(2, 3L), SavedBiomeStore.open(root).get(2, 0).orElseThrow()); + } + + @Test + public void failedSyncRejectsFurtherClaimsButKeepsWrittenRecords() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + store.claimAndPersist(chunk(0, 3L)); + IOException failure = new IOException("Deferred saved biome force failed"); + try (MockedStatic ignored = regionChannels(root, (source, intercepted) -> + doAnswer(invocation -> { + throw failure; + }).when(intercepted).force(true))) { + assertSame(failure, assertThrows(IOException.class, store::sync)); + } + assertSame(failure, assertThrows(IOException.class, () -> store.claimAndPersist(chunk(1, 3L))).getCause()); + assertSame(failure, assertThrows(IOException.class, () -> store.get(5, 0)).getCause()); + SavedBiomeStore reopened = SavedBiomeStore.open(root); + assertEquals(chunk(0, 3L), reopened.get(0, 0).orElseThrow()); + assertTrue(reopened.get(1, 0).isEmpty()); + } + + @Test + public void failedWriteRollsBackWithoutPublishingAndRetainsRollbackFailure() throws Exception { for (boolean failRollback : List.of(false, true)) { Path root = temporaryFolder.newFolder().toPath(); SavedBiomeStore store = SavedBiomeStore.open(root); store.claimAndPersist(chunk(0, 3L)); byte[] original = Files.readAllBytes(regionPath(root)); - IOException failure = new IOException("Saved biome force failed"); + IOException failure = new IOException("Saved biome write failed"); IOException rollback = new IOException("Saved biome rollback force failed"); - AtomicBoolean first = new AtomicBoolean(true); try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) { FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel())); doAnswer(invocation -> { - if (first.getAndSet(false)) { - throw failure; - } + throw failure; + }).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt()); + doAnswer(invocation -> { if (failRollback) { throw rollback; } @@ -262,6 +307,24 @@ public class SavedBiomeStoreWriteTest { } } + private static MockedStatic regionChannels(Path root, ChannelMutation mutation) { + Path region = regionPath(root); + return mockStatic(FileChannel.class, invocation -> { + FileChannel source = (FileChannel) invocation.callRealMethod(); + if (invocation.getMethod().getParameterCount() != 2 || !region.equals(invocation.getArgument(0))) { + return source; + } + FileChannel intercepted = mock(FileChannel.class, delegatesTo(source)); + mutation.apply(source, intercepted); + return intercepted; + }); + } + + @FunctionalInterface + private interface ChannelMutation { + void apply(FileChannel source, FileChannel intercepted) throws IOException; + } + private static ClaimResult claimResult(SavedBiomeStore store, SavedBiomeChunk chunk) { try { return new ClaimResult(store.claimAndPersist(chunk), null); diff --git a/core/src/test/java/art/arcane/iris/world/pregen/AsyncPregenHistorySyncTest.java b/core/src/test/java/art/arcane/iris/world/pregen/AsyncPregenHistorySyncTest.java new file mode 100644 index 000000000..50a6b2f3b --- /dev/null +++ b/core/src/test/java/art/arcane/iris/world/pregen/AsyncPregenHistorySyncTest.java @@ -0,0 +1,84 @@ +package art.arcane.iris.world.pregen; + +import art.arcane.iris.generation.runtime.Engine; +import art.arcane.iris.spi.IrisLogging; +import art.arcane.iris.world.IrisToolbelt; +import org.bukkit.World; +import org.junit.Test; +import org.mockito.InOrder; +import org.mockito.MockedStatic; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.same; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AsyncPregenHistorySyncTest { + @Test + public void closeForcesGenerationHistoryOnceOutstandingChunksDrain() throws Exception { + Engine engine = mock(Engine.class); + PregenAdmissionGate admission = mock(PregenAdmissionGate.class); + AsyncPregenMethod method = closable(engine, admission); + try (MockedStatic toolbelt = mockStatic(IrisToolbelt.class)) { + toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true); + method.close(); + } + InOrder order = inOrder(admission, engine); + order.verify(admission).awaitDrain(anyLong(), any(), any()); + order.verify(engine).syncGenerationHistory(); + } + + @Test + public void closeReportsAHistorySyncFailureAndStillFinishes() throws Exception { + Engine engine = mock(Engine.class); + IllegalStateException failure = new IllegalStateException("history force failed"); + doThrow(failure).when(engine).syncGenerationHistory(); + AsyncPregenMethod method = closable(engine, mock(PregenAdmissionGate.class)); + try (MockedStatic toolbelt = mockStatic(IrisToolbelt.class); + MockedStatic logging = mockStatic(IrisLogging.class)) { + toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true); + method.close(); + logging.verify(() -> IrisLogging.reportError(anyString(), same(failure))); + } + verify(engine).syncGenerationHistory(); + } + + private static AsyncPregenMethod closable(Engine engine, PregenAdmissionGate admission) throws Exception { + AsyncPregenMethod method = mock(AsyncPregenMethod.class, CALLS_REAL_METHODS); + World world = mock(World.class); + when(world.getName()).thenReturn("world"); + set(method, "world", world); + set(method, "closing", new AtomicBoolean()); + set(method, "holdsWorkerBoost", new AtomicBoolean()); + set(method, "admission", admission); + set(method, "mantleCleanup", new PregenSerialWorker("Test Mantle Cleanup", "world")); + set(method, "chunkFlush", new PregenSerialWorker("Test Chunk Flush", "world")); + set(method, "regionChunks", new ConcurrentHashMap<>()); + set(method, "pendingEvictions", new ConcurrentLinkedQueue<>()); + set(method, "metricsEngine", engine); + Class ticketExecutor = Class.forName(AsyncPregenMethod.class.getName() + "$TicketExecutor"); + Constructor constructor = ticketExecutor.getDeclaredConstructor(AsyncPregenMethod.class); + constructor.setAccessible(true); + set(method, "executor", constructor.newInstance(method)); + return method; + } + + private static void set(AsyncPregenMethod method, String name, Object value) throws Exception { + Field field = AsyncPregenMethod.class.getDeclaredField(name); + field.setAccessible(true); + field.set(method, value); + } +} diff --git a/core/src/test/java/art/arcane/iris/world/storage/DeferredDurabilityTest.java b/core/src/test/java/art/arcane/iris/world/storage/DeferredDurabilityTest.java new file mode 100644 index 000000000..449a349e6 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/world/storage/DeferredDurabilityTest.java @@ -0,0 +1,161 @@ +package art.arcane.iris.world.storage; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.MockedStatic; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; +import static org.mockito.AdditionalAnswers.delegatesTo; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; + +public class DeferredDurabilityTest { + private static final long NEVER_MILLIS = TimeUnit.HOURS.toMillis(1L); + + @Rule + public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void syncForcesEachWrittenFileOnceAndThenNothing() throws Exception { + Path first = temporaryFolder.newFile().toPath(); + Path second = temporaryFolder.newFile().toPath(); + DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> { + throw new AssertionError(failure); + }); + durability.written(first); + durability.written(second); + durability.written(first); + Map forces = new ConcurrentHashMap<>(); + try (MockedStatic ignored = countingForces(forces, null)) { + durability.sync(); + assertEquals(Map.of(first, 1, second, 1), snapshot(forces)); + durability.sync(); + assertEquals(Map.of(first, 1, second, 1), snapshot(forces)); + } + } + + @Test + public void syncForcesLinkedDirectories() throws Exception { + assumeTrue(File.separatorChar != '\\'); + Path directory = temporaryFolder.newFolder().toPath(); + DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> { + throw new AssertionError(failure); + }); + durability.linked(directory); + Map forces = new ConcurrentHashMap<>(); + try (MockedStatic ignored = countingForces(forces, null)) { + durability.sync(); + } + assertEquals(Map.of(directory, 1), snapshot(forces)); + } + + @Test + public void relaxedModeTracksNothing() throws Exception { + Path file = temporaryFolder.newFile().toPath(); + String previous = System.getProperty(Durability.MODE_PROPERTY); + System.setProperty(Durability.MODE_PROPERTY, "relaxed"); + try { + DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> { + throw new AssertionError(failure); + }); + durability.written(file); + durability.linked(file.getParent()); + Map forces = new ConcurrentHashMap<>(); + try (MockedStatic ignored = countingForces(forces, null)) { + durability.sync(); + } + assertTrue(forces.isEmpty()); + } finally { + if (previous == null) { + System.clearProperty(Durability.MODE_PROPERTY); + } else { + System.setProperty(Durability.MODE_PROPERTY, previous); + } + } + } + + @Test + public void replacedFilesAreSkippedAndFailuresReachTheSyncCaller() throws Exception { + Path removed = temporaryFolder.newFile().toPath(); + Path failing = temporaryFolder.newFile().toPath(); + DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> { + throw new AssertionError(failure); + }); + durability.written(removed); + durability.written(failing); + Files.delete(removed); + IOException failure = new IOException("force failed"); + Map forces = new ConcurrentHashMap<>(); + try (MockedStatic ignored = countingForces(forces, path -> path.equals(failing) ? failure : null)) { + assertSame(failure, assertThrows(IOException.class, durability::sync)); + durability.sync(); + } + assertEquals(Map.of(failing, 1), snapshot(forces)); + } + + @Test + public void backgroundFlushRunsAfterTheDelayAndReportsFailures() throws Exception { + Path directory = temporaryFolder.newFolder().toPath(); + CountDownLatch reported = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + DeferredDurability durability = new DeferredDurability(10L, background -> { + failure.set(background); + reported.countDown(); + }); + durability.written(directory); + assertTrue(reported.await(5L, TimeUnit.SECONDS)); + assertNotNull(failure.get()); + durability.sync(); + } + + private static Map snapshot(Map forces) { + Map values = new ConcurrentHashMap<>(); + forces.forEach((path, count) -> values.put(path, count.get())); + return values; + } + + private static MockedStatic countingForces(Map forces, FailurePlan failures) { + return mockStatic(FileChannel.class, invocation -> { + FileChannel source = (FileChannel) invocation.callRealMethod(); + if (invocation.getMethod().getParameterCount() != 2) { + return source; + } + Path path = invocation.getArgument(0); + FileChannel intercepted = mock(FileChannel.class, delegatesTo(source)); + doAnswer(force -> { + forces.computeIfAbsent(path, ignored -> new AtomicInteger()).incrementAndGet(); + IOException planned = failures == null ? null : failures.failureFor(path); + if (planned != null) { + throw planned; + } + source.force(true); + return null; + }).when(intercepted).force(true); + return intercepted; + }); + } + + @FunctionalInterface + private interface FailurePlan { + IOException failureFor(Path path); + } +}