diff --git a/worldedit-core/build.gradle.kts b/worldedit-core/build.gradle.kts index dba8ca0823..bf0fd8124f 100644 --- a/worldedit-core/build.gradle.kts +++ b/worldedit-core/build.gradle.kts @@ -62,6 +62,13 @@ dependencies { // Tests testRuntimeOnly(libs.log4j.core) testImplementation(libs.parallelgzip) + // lz4-java is compileOnly for the main sourceSet (expected to be shaded in by platform jars + // at runtime), but MainUtil#getCompressedOS() references LZ4 stream types even on code paths + // that aren't taken at COMPRESSION_LEVEL 0 - the JVM verifier still needs to resolve those + // types when the method is first invoked, so tests exercising that method (e.g. + // DiskStorageHistory's stream getters, which delegate to it) need the real classes on the + // test runtime classpath. + testImplementation(libs.lz4Java) { isTransitive = false } } tasks.test { diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/DiskStorageHistory.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/DiskStorageHistory.java index eb2e3da59e..cd1779402c 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/DiskStorageHistory.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/DiskStorageHistory.java @@ -57,17 +57,17 @@ public class DiskStorageHistory extends FaweStreamChangeSet { * [contents]... * { short rel x, short rel z, unsigned byte y, short combinedFrom, short combinedTo } */ - private FaweOutputStream osBD; + private volatile FaweOutputStream osBD; // biome - private FaweOutputStream osBIO; + private volatile FaweOutputStream osBIO; // NBT From - private NBTOutputStream osNBTF; + private volatile NBTOutputStream osNBTF; // NBT To - private NBTOutputStream osNBTT; + private volatile NBTOutputStream osNBTT; // Entity Create From - private NBTOutputStream osENTCF; + private volatile NBTOutputStream osENTCF; // Entity Create To - private NBTOutputStream osENTCT; + private volatile NBTOutputStream osENTCT; private int index; @@ -298,16 +298,70 @@ public long getSizeOnDisk() { return total; } + /** + * Closes {@code closeable} (if not null), suppressing rather than propagating or masking any + * failure from the close itself onto {@code primary} - the original failure that triggered + * the cleanup, and the one that must actually reach the caller. Used when a fallible resource + * has to be closed before rethrowing, without losing the real cause if the close itself + * fails too. No-op if {@code closeable} is null (e.g. construction failed before it was + * created). + */ + private static void closeQuietly(AutoCloseable closeable, Throwable primary) { + if (closeable == null) { + return; + } + try { + closeable.close(); + } catch (Throwable suppressed) { + // Throwable, not just Exception: a RuntimeException or Error from close() itself + // must not propagate in place of primary either, or it would mask the original + // failure exactly like the checked-exception case this method exists to prevent. + // This still doesn't swallow anything - it's attached to primary, which callers + // always rethrow. + primary.addSuppressed(suppressed); + } + } + @Override public FaweOutputStream getBlockOS(int x, int y, int z) throws IOException { if (osBD != null) { return osBD; } synchronized (this) { + if (osBD != null) { + return osBD; + } bdFile.getParentFile().mkdirs(); bdFile.createNewFile(); - osBD = getCompressedOS(new FileOutputStream(bdFile)); - writeHeader(osBD, x, y, z); + // Write the header before publishing to the volatile field: osBD is not thread-safe + // (see getCompressedOS()'s javadoc), so another thread's unsynchronized fast-path + // read (`if (osBD != null) return osBD;`) must never observe the stream until it is + // fully initialized, or it could start writing block data concurrently with the + // header write here and corrupt the file. + // + // The raw FileOutputStream is kept in a local so it can be closed directly if + // getCompressedOS() itself throws before returning anything to close; once wrapped, + // closing the returned stream closes the FileOutputStream it wraps. Both catch + // clauses below also cover unchecked RuntimeException/Error, not just IOException: + // MainUtil.getCompressedOS()/writeHeader() can fail that way too (e.g. a linkage + // error constructing a compression stream), and an unchecked failure after fos is + // opened would otherwise leak the file descriptor since it's never published to osBD. + FileOutputStream fos = new FileOutputStream(bdFile); + FaweOutputStream stream; + try { + stream = getCompressedOS(fos); + } catch (IOException | RuntimeException | Error e) { + closeQuietly(fos, e); + throw e; + } + try { + writeHeader(stream, x, y, z); + } catch (IOException | RuntimeException | Error e) { + // Not yet published to osBD, so close() would never close this otherwise. + closeQuietly(stream, e); + throw e; + } + osBD = stream; return osBD; } } @@ -318,9 +372,18 @@ public FaweOutputStream getBiomeOS() throws IOException { return osBIO; } synchronized (this) { + if (osBIO != null) { + return osBIO; + } bioFile.getParentFile().mkdirs(); bioFile.createNewFile(); - osBIO = getCompressedOS(new FileOutputStream(bioFile)); + FileOutputStream fos = new FileOutputStream(bioFile); + try { + osBIO = getCompressedOS(fos); + } catch (IOException | RuntimeException | Error e) { + closeQuietly(fos, e); + throw e; + } return osBIO; } } @@ -330,10 +393,35 @@ public NBTOutputStream getEntityCreateOS() throws IOException { if (osENTCT != null) { return osENTCT; } - enttFile.getParentFile().mkdirs(); - enttFile.createNewFile(); - osENTCT = new NBTOutputStream(getCompressedOS(new FileOutputStream(enttFile))); - return osENTCT; + synchronized (this) { + if (osENTCT != null) { + return osENTCT; + } + enttFile.getParentFile().mkdirs(); + enttFile.createNewFile(); + // Two-step, like getBlockOS(): if getCompressedOS() succeeds but the NBTOutputStream + // constructor then fails, closing only fos would leave the compression wrapper chain + // getCompressedOS() built (buffered/LZ4 streams) unclosed. Close whichever the + // innermost fully-constructed object is at the point of failure - closing the wrapper + // cascades down and closes fos too. + FileOutputStream fos = new FileOutputStream(enttFile); + FaweOutputStream stream; + try { + stream = getCompressedOS(fos); + } catch (IOException | RuntimeException | Error e) { + closeQuietly(fos, e); + throw e; + } + try { + osENTCT = new NBTOutputStream(stream); + } catch (RuntimeException | Error e) { + // NBTOutputStream(OutputStream) doesn't declare IOException - only unchecked + // failures are possible here. + closeQuietly(stream, e); + throw e; + } + return osENTCT; + } } @Override @@ -341,10 +429,29 @@ public NBTOutputStream getEntityRemoveOS() throws IOException { if (osENTCF != null) { return osENTCF; } - entfFile.getParentFile().mkdirs(); - entfFile.createNewFile(); - osENTCF = new NBTOutputStream(getCompressedOS(new FileOutputStream(entfFile))); - return osENTCF; + synchronized (this) { + if (osENTCF != null) { + return osENTCF; + } + entfFile.getParentFile().mkdirs(); + entfFile.createNewFile(); + // See getEntityCreateOS() for why this is two steps. + FileOutputStream fos = new FileOutputStream(entfFile); + FaweOutputStream stream; + try { + stream = getCompressedOS(fos); + } catch (IOException | RuntimeException | Error e) { + closeQuietly(fos, e); + throw e; + } + try { + osENTCF = new NBTOutputStream(stream); + } catch (RuntimeException | Error e) { + closeQuietly(stream, e); + throw e; + } + return osENTCF; + } } @Override @@ -352,10 +459,29 @@ public NBTOutputStream getTileCreateOS() throws IOException { if (osNBTT != null) { return osNBTT; } - nbttFile.getParentFile().mkdirs(); - nbttFile.createNewFile(); - osNBTT = new NBTOutputStream(getCompressedOS(new FileOutputStream(nbttFile))); - return osNBTT; + synchronized (this) { + if (osNBTT != null) { + return osNBTT; + } + nbttFile.getParentFile().mkdirs(); + nbttFile.createNewFile(); + // See getEntityCreateOS() for why this is two steps. + FileOutputStream fos = new FileOutputStream(nbttFile); + FaweOutputStream stream; + try { + stream = getCompressedOS(fos); + } catch (IOException | RuntimeException | Error e) { + closeQuietly(fos, e); + throw e; + } + try { + osNBTT = new NBTOutputStream(stream); + } catch (RuntimeException | Error e) { + closeQuietly(stream, e); + throw e; + } + return osNBTT; + } } @Override @@ -363,10 +489,29 @@ public NBTOutputStream getTileRemoveOS() throws IOException { if (osNBTF != null) { return osNBTF; } - nbtfFile.getParentFile().mkdirs(); - nbtfFile.createNewFile(); - osNBTF = new NBTOutputStream(getCompressedOS(new FileOutputStream(nbtfFile))); - return osNBTF; + synchronized (this) { + if (osNBTF != null) { + return osNBTF; + } + nbtfFile.getParentFile().mkdirs(); + nbtfFile.createNewFile(); + // See getEntityCreateOS() for why this is two steps. + FileOutputStream fos = new FileOutputStream(nbtfFile); + FaweOutputStream stream; + try { + stream = getCompressedOS(fos); + } catch (IOException | RuntimeException | Error e) { + closeQuietly(fos, e); + throw e; + } + try { + osNBTF = new NBTOutputStream(stream); + } catch (RuntimeException | Error e) { + closeQuietly(stream, e); + throw e; + } + return osNBTF; + } } @Override diff --git a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/DiskStorageHistoryConcurrencyTest.java b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/DiskStorageHistoryConcurrencyTest.java new file mode 100644 index 0000000000..4b3d51e095 --- /dev/null +++ b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/DiskStorageHistoryConcurrencyTest.java @@ -0,0 +1,269 @@ +package com.fastasyncworldedit.core.history; + +import com.fastasyncworldedit.core.configuration.Settings; +import com.fastasyncworldedit.core.internal.io.FaweOutputStream; +import com.sk89q.worldedit.world.World; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; + +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +/** + * Regression test for the double-checked locking bug in {@link DiskStorageHistory}'s lazy + * stream getters: racing threads must always observe the same, single, lazily-constructed + * stream instance. + * + *

Runs its test methods on the same thread: {@code @BeforeEach}/{@code @AfterEach} save and + * restore the global {@code Settings.settings().HISTORY.COMPRESSION_LEVEL}, and this suite has + * method-level parallelism enabled by default, so two methods of this class racing on that field + * would corrupt each other's setting.

+ */ +@Execution(ExecutionMode.SAME_THREAD) +class DiskStorageHistoryConcurrencyTest { + + private static final int THREADS = 16; + private static final int ITERATIONS = 20; + + @TempDir + File tempDir; + + private World world; + private int originalCompressionLevel; + + @BeforeEach + void setUp() { + world = mock(World.class); + when(world.getMinY()).thenReturn(-64); + when(world.getMaxY()).thenReturn(319); + when(world.getName()).thenReturn("concurrency-test-world"); + + // MainUtil.getCompressedOS() returns a plain, uncompressed stream at COMPRESSION_LEVEL 0 + // and otherwise builds an LZ4/Zstd stack. This test only cares about the identity of the + // lazily-constructed stream, not compression behavior, so force level 0 to keep it + // independent of the compression backend selection logic. + originalCompressionLevel = Settings.settings().HISTORY.COMPRESSION_LEVEL; + Settings.settings().HISTORY.COMPRESSION_LEVEL = 0; + } + + @AfterEach + void tearDown() { + Settings.settings().HISTORY.COMPRESSION_LEVEL = originalCompressionLevel; + } + + @Test + void getBlockOSReturnsSingleInstanceUnderConcurrentAccess() throws Exception { + for (int iteration = 0; iteration < ITERATIONS; iteration++) { + File folder = new File(tempDir, "block-" + iteration); + DiskStorageHistory history = new DiskStorageHistory(folder, world, UUID.randomUUID(), iteration); + + Set instances = Collections.newSetFromMap(new IdentityHashMap<>()); + try { + runConcurrently(THREADS, () -> { + try { + return history.getBlockOS(0, 0, 0); + } catch (Exception e) { + throw new RuntimeException(e); + } + }, instances); + + assertEquals( + 1, + instances.size(), + "getBlockOS() must return exactly one distinct stream instance across racing threads (iteration " + + iteration + ")" + ); + } finally { + // Must run even on assertion failure, or the open stream(s) from the racing + // getBlockOS() calls above leak - on Windows that can also make @TempDir cleanup + // fail with a confusing secondary error that masks the real assertion failure. + history.close(); + } + } + } + + @Test + void getTileCreateOSReturnsSingleInstanceUnderConcurrentAccess() throws Exception { + for (int iteration = 0; iteration < ITERATIONS; iteration++) { + File folder = new File(tempDir, "tile-" + iteration); + DiskStorageHistory history = new DiskStorageHistory(folder, world, UUID.randomUUID(), iteration); + + Set instances = Collections.newSetFromMap(new IdentityHashMap<>()); + try { + runConcurrently(THREADS, () -> { + try { + return history.getTileCreateOS(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }, instances); + + assertEquals( + 1, + instances.size(), + "getTileCreateOS() must return exactly one distinct stream instance across racing threads (iteration " + + iteration + ")" + ); + } finally { + // See getBlockOSReturnsSingleInstanceUnderConcurrentAccess for why this must run + // even on assertion failure. + history.close(); + } + } + } + + /** + * Spins up {@code threadCount} threads, gates them behind a {@link CyclicBarrier} so they all + * invoke {@code action} at roughly the same instant, and records the identity of every + * returned object into {@code resultsOut}. + */ + private void runConcurrently(int threadCount, java.util.function.Supplier action, Set resultsOut) + throws InterruptedException { + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CyclicBarrier barrier = new CyclicBarrier(threadCount); + CountDownLatch done = new CountDownLatch(threadCount); + AtomicInteger failures = new AtomicInteger(); + + for (int i = 0; i < threadCount; i++) { + executor.submit(() -> { + try { + barrier.await(10, TimeUnit.SECONDS); + Object result = action.get(); + synchronized (resultsOut) { + resultsOut.add(result); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + failures.incrementAndGet(); + } catch (BrokenBarrierException | java.util.concurrent.TimeoutException e) { + failures.incrementAndGet(); + } catch (Throwable e) { + // Throwable, not just RuntimeException: an Error (e.g. an AssertionError from + // a failed assertion inside this worker) would otherwise complete this + // submitted task's Future exceptionally with nothing checking it, and the + // test could still pass since failures wouldn't be incremented. + failures.incrementAndGet(); + } finally { + done.countDown(); + } + }); + } + + boolean completed; + boolean terminated; + try { + completed = done.await(30, TimeUnit.SECONDS); + } finally { + // Must run even if done.await() itself is interrupted, not just on the timeout path + // below - otherwise a stuck/interrupted wait leaks the pool's non-daemon threads. + executor.shutdownNow(); + terminated = executor.awaitTermination(10, TimeUnit.SECONDS); + } + + if (!completed) { + fail("Timed out waiting for racing threads to finish (possible barrier deadlock or hang)"); + } + if (!terminated) { + fail("Executor did not terminate after shutdownNow(); worker threads may still be running"); + } + assertEquals(0, failures.get(), "no thread should have failed while racing to acquire the lazy stream"); + } + + /** + * Deterministic regression test for the cleanup-on-failure paths in {@code getBlockOS()}: if + * {@code writeHeader()} fails, the not-yet-published stream must be closed - but if that + * close attempt itself fails, the resulting exception must not replace (mask) the original + * header-write failure; it must be attached via {@link Throwable#addSuppressed}. + */ + @Test + void getBlockOSCloseFailureIsSuppressedNotMasked() throws Exception { + DiskStorageHistory history = spy(new DiskStorageHistory(tempDir, world, UUID.randomUUID(), 0)); + + IOException closeFailure = new IOException("close failed"); + // Let getCompressedOS() run for real (so the actual FileOutputStream is genuinely wrapped + // and can be genuinely closed, same as production), then spy just its close() so it + // releases the real handle - keeping @TempDir cleanup happy - before throwing the + // synthetic failure this test is targeting. + doAnswer(invocation -> { + FaweOutputStream real = (FaweOutputStream) invocation.callRealMethod(); + FaweOutputStream streamSpy = spy(real); + doAnswer(closeInvocation -> { + real.close(); + throw closeFailure; + }).when(streamSpy).close(); + return streamSpy; + }).when(history).getCompressedOS(any()); + + IOException headerFailure = new IOException("header failed"); + doThrow(headerFailure).when(history).writeHeader(any(), anyInt(), anyInt(), anyInt()); + + IOException thrown = assertThrows(IOException.class, () -> history.getBlockOS(0, 0, 0)); + + assertSame(headerFailure, thrown, "the original header-write failure must propagate, not be masked by a " + + "failure while closing the stream"); + assertEquals(1, thrown.getSuppressed().length, "the close failure must be attached as a suppressed exception"); + assertSame(closeFailure, thrown.getSuppressed()[0]); + } + + /** + * Same as {@link #getBlockOSCloseFailureIsSuppressedNotMasked()}, but the cleanup close() + * itself fails with an {@link Error} rather than an {@code IOException}. {@code + * closeQuietly()} must catch {@link Throwable}, not just {@link Exception} - {@link + * RuntimeException} would not distinguish the two, since it is itself an {@code Exception} - + * or an {@code Error} from close() would propagate in place of the original and mask it + * exactly like the checked-exception case does. + */ + @Test + void getBlockOSErrorFromCloseIsSuppressedNotMasked() throws Exception { + DiskStorageHistory history = spy(new DiskStorageHistory(tempDir, world, UUID.randomUUID(), 0)); + + Error closeFailure = new AssertionError("close failed with an Error"); + doAnswer(invocation -> { + FaweOutputStream real = (FaweOutputStream) invocation.callRealMethod(); + FaweOutputStream streamSpy = spy(real); + doAnswer(closeInvocation -> { + real.close(); + throw closeFailure; + }).when(streamSpy).close(); + return streamSpy; + }).when(history).getCompressedOS(any()); + + IOException headerFailure = new IOException("header failed"); + doThrow(headerFailure).when(history).writeHeader(any(), anyInt(), anyInt(), anyInt()); + + IOException thrown = assertThrows(IOException.class, () -> history.getBlockOS(0, 0, 0)); + + assertSame(headerFailure, thrown, "the original header-write failure must propagate, not be masked by an " + + "unchecked failure while closing the stream"); + assertEquals(1, thrown.getSuppressed().length, "the close failure must be attached as a suppressed exception"); + assertSame(closeFailure, thrown.getSuppressed()[0]); + } + +}