From 6f74c50bc2162c91dee2ad5f70e83a60a1e58100 Mon Sep 17 00:00:00 2001 From: Matt <4009945+MattBDev@users.noreply.github.com> Date: Wed, 15 Jul 2026 01:51:03 -0400 Subject: [PATCH 1/3] Make FaweStreamChangeSet.blockSize atomic and document read reentrancy blockSize was a plain long incremented from six add* methods that run on pipeline worker threads and read unsynchronized from isEmpty()/longSize()/ size(). Neither the increments nor the 64-bit reads were atomic, so counts could be lost and readers could observe a torn value. Use a LongAdder, which stays cheap under the concurrent writers this path actually sees. RollbackOptimizedHistory's constructor is updated for the new field type. Its (int) truncation of the incoming long size is deliberately preserved here: it is a separate known bug and fixing it is out of scope for this change. Also documents that a single FaweStreamChangeSet is not safe for concurrent or overlapping read traversals: posDel, idDel, originX, originZ and version hold decoder state on the instance rather than per traversal, so a second concurrent read, or a read racing readHeader(), corrupts the running deltas. This is documentation only; a runtime guard needs the per-traversal codec refactor and is not attempted here. Adds FaweStreamChangeSetBlockSizeTest covering concurrent add() counting. Co-Authored-By: Claude Opus 4.8 --- worldedit-core/build.gradle.kts | 1 + .../history/RollbackOptimizedHistory.java | 7 +- .../changeset/FaweStreamChangeSet.java | 47 ++++++++-- .../FaweStreamChangeSetBlockSizeTest.java | 91 +++++++++++++++++++ 4 files changed, 135 insertions(+), 11 deletions(-) create mode 100644 worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java diff --git a/worldedit-core/build.gradle.kts b/worldedit-core/build.gradle.kts index dba8ca0823..597a69bd93 100644 --- a/worldedit-core/build.gradle.kts +++ b/worldedit-core/build.gradle.kts @@ -62,6 +62,7 @@ dependencies { // Tests testRuntimeOnly(libs.log4j.core) testImplementation(libs.parallelgzip) + testImplementation(libs.lz4Java) } tasks.test { diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java index 7f677b0d86..117887ed32 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java @@ -51,7 +51,12 @@ public RollbackOptimizedHistory( this.maxX = region.getMaximumX(); this.maxY = region.getMaximumY(); this.maxZ = region.getMaximumZ(); - this.blockSize = (int) size; + // NOTE: this truncates `size` to an int before storing it, same as the historic + // `this.blockSize = (int) size;` assignment did. That truncation is a known, separate + // bug tracked elsewhere - it is preserved here intentionally and not fixed as part of + // the blockSize -> LongAdder migration. + this.blockSize.reset(); + this.blockSize.add((int) size); this.command = command; this.closed = true; } diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java index 8c9da71ae5..19639cf0f4 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java @@ -37,10 +37,37 @@ import java.util.NoSuchElementException; import java.util.Queue; import java.util.concurrent.Exchanger; +import java.util.concurrent.atomic.LongAdder; import java.util.function.BiConsumer; /** * FAWE stream ChangeSet offering support for extended-height worlds + * + *

Thread-safety / reentrancy warning: a single {@code FaweStreamChangeSet} + * instance is not safe for concurrent read traversals. The decoder state used while + * reading back changes - {@link #posDel}, {@link #idDel}, {@link #originX}, {@link #originZ} and + * {@link #version} - is stored directly on the instance rather than being scoped to a single + * traversal. These fields are (re)initialized by {@link #readHeader(InputStream)} / + * {@link #setupStreamDelegates(int)} and are then mutated as each iterator or + * {@link com.fastasyncworldedit.core.history.change.ChangePopulator} advances (e.g. the + * running {@code lx}/{@code ly}/{@code lz} delta-decoding state captured by the position + * delegate).

+ * + *

As a result:

+ * + * + *

This is documentation only for now: a proper fix that gives each traversal its own, + * independent decoder state is planned as a future, larger refactor. No runtime locking or + * guard has been added here to enforce the above.

*/ public abstract class FaweStreamChangeSet extends AbstractChangeSet { @@ -52,7 +79,7 @@ public abstract class FaweStreamChangeSet extends AbstractChangeSet { private final int compression; private final int minY; - protected long blockSize; + protected final LongAdder blockSize = new LongAdder(); private int originX; private int originZ; private int version; @@ -292,21 +319,21 @@ public FaweOutputStream getCompressedOS(OutputStream os) throws IOException { @Override public boolean isEmpty() { - if (blockSize > 0) { + if (blockSize.sum() > 0) { return false; } if (!super.isEmpty()) { return false; } flush(); - return blockSize == 0; + return blockSize.sum() == 0; } @Override public long longSize() { // Flush so we can accurately get the size flush(); - return blockSize; + return blockSize.sum(); } @Override @@ -361,7 +388,7 @@ public int getOriginZ() { @Override public void add(int x, int y, int z, int combinedFrom, int combinedTo) { - blockSize++; + blockSize.increment(); try { FaweOutputStream stream = getBlockOS(x, y, z); //x @@ -374,7 +401,7 @@ public void add(int x, int y, int z, int combinedFrom, int combinedTo) { @Override public void addBiomeChange(int bx, int by, int bz, BiomeType from, BiomeType to) { - blockSize++; + blockSize.increment(); try { int x = bx >> 2; int y = by >> 2; @@ -400,7 +427,7 @@ public void addBiomeChange(int bx, int by, int bz, BiomeType from, BiomeType to) @Override public void addTileCreate(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getTileCreateOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); @@ -411,7 +438,7 @@ public void addTileCreate(final FaweCompoundTag tag) { @Override public void addTileRemove(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getTileRemoveOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); @@ -422,7 +449,7 @@ public void addTileRemove(final FaweCompoundTag tag) { @Override public void addEntityRemove(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getEntityRemoveOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); @@ -433,7 +460,7 @@ public void addEntityRemove(final FaweCompoundTag tag) { @Override public void addEntityCreate(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getEntityCreateOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); diff --git a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java new file mode 100644 index 0000000000..3929f6d239 --- /dev/null +++ b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java @@ -0,0 +1,91 @@ +package com.fastasyncworldedit.core.history.changeset; + +import com.fastasyncworldedit.core.configuration.Settings; +import com.fastasyncworldedit.core.history.MemoryOptimizedHistory; +import com.sk89q.worldedit.world.World; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Regression test for the {@code blockSize} counter on {@link FaweStreamChangeSet}. Historically + * this was a plain, unsynchronized {@code long} incremented from multiple pipeline worker + * threads via {@code blockSize++}, which is not atomic and can silently lose increments under + * real contention. It is now a {@link java.util.concurrent.atomic.LongAdder}, which this test + * verifies is accurate under many concurrent writers. + * + *

Mutates the process-wide {@link Settings} singleton (to avoid needing the {@code lz4-java} + * compression codec, which is {@code compileOnly} and not present on the unit test runtime + * classpath), so this class is marked {@link Isolated} to avoid interference with other tests + * that may run concurrently in the same JVM.

+ */ +@Isolated +class FaweStreamChangeSetBlockSizeTest { + + private static final int THREADS = 16; + private static final int CALLS_PER_THREAD = 1000; + + @Test + void addIsAccurateUnderConcurrentWriters() throws InterruptedException { + int previousCompressionLevel = Settings.settings().HISTORY.COMPRESSION_LEVEL; + // Compression level 0 skips the LZ4/Zstd codecs entirely (see MainUtil#getCompressedOS), + // which keeps this test independent of the compileOnly lz4-java dependency. + Settings.settings().HISTORY.COMPRESSION_LEVEL = 0; + + World world = mock(World.class); + when(world.getMinY()).thenReturn(-64); + when(world.getMaxY()).thenReturn(319); + + try { + MemoryOptimizedHistory changeSet = new MemoryOptimizedHistory(world); + + ExecutorService executor = Executors.newFixedThreadPool(THREADS); + CountDownLatch startLatch = new CountDownLatch(1); + + List> futures = new ArrayList<>(THREADS); + for (int t = 0; t < THREADS; t++) { + final int threadIndex = t; + Callable task = () -> { + startLatch.await(); + for (int i = 0; i < CALLS_PER_THREAD; i++) { + changeSet.add(threadIndex, 0, i, 0, 1); + } + return null; + }; + futures.add(executor.submit(task)); + } + + startLatch.countDown(); + for (Future future : futures) { + try { + future.get(30, TimeUnit.SECONDS); + } catch (ExecutionException e) { + throw new AssertionError("worker thread failed while calling add()", e.getCause()); + } catch (java.util.concurrent.TimeoutException e) { + throw new AssertionError("worker thread did not finish in time", e); + } + } + executor.shutdown(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS), "executor did not terminate in time"); + + assertEquals((long) THREADS * CALLS_PER_THREAD, changeSet.longSize()); + } finally { + Settings.settings().HISTORY.COMPRESSION_LEVEL = previousCompressionLevel; + } + } + +} From 5531a0de36a2bb44947c69409a13e3d44ad43caa Mon Sep 17 00:00:00 2001 From: Matt <4009945+MattBDev@users.noreply.github.com> Date: Wed, 15 Jul 2026 07:58:31 -0400 Subject: [PATCH 2/3] Address Copilot review: executor cleanup on failure paths, accurate javadoc If a worker future threw or timed out, the test rethrew immediately from the catch block, skipping the graceful shutdown()/awaitTermination below it and leaking non-daemon threads into the rest of the suite. Move executor construction outside the try and unconditionally shutdownNow() in a finally, so a stuck or failed worker no longer leaks threads. Also corrects the class javadoc, which claimed COMPRESSION_LEVEL was forced to 0 because lz4-java is missing from the test classpath - inaccurate, since this PR adds lz4-java as a test dependency. The real reason: level 0 bypasses the compression backend entirely so the test only measures counter accuracy, and @Isolated is needed because Settings is process-global mutable state. Co-Authored-By: Claude Sonnet 5 --- .../FaweStreamChangeSetBlockSizeTest.java | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java index 3929f6d239..7ad7c0ee2b 100644 --- a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java +++ b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java @@ -28,10 +28,11 @@ * real contention. It is now a {@link java.util.concurrent.atomic.LongAdder}, which this test * verifies is accurate under many concurrent writers. * - *

Mutates the process-wide {@link Settings} singleton (to avoid needing the {@code lz4-java} - * compression codec, which is {@code compileOnly} and not present on the unit test runtime - * classpath), so this class is marked {@link Isolated} to avoid interference with other tests - * that may run concurrently in the same JVM.

+ *

Forces {@code Settings.settings().HISTORY.COMPRESSION_LEVEL} to 0, which bypasses the + * compression backend entirely (see {@code MainUtil#getCompressedOS}), so this test only measures + * counter accuracy and isn't coupled to compression behavior. Because {@link Settings} is + * process-global mutable state, this class is marked {@link Isolated} so no other test running + * concurrently in the same JVM observes the temporarily-changed level.

*/ @Isolated class FaweStreamChangeSetBlockSizeTest { @@ -50,10 +51,9 @@ void addIsAccurateUnderConcurrentWriters() throws InterruptedException { when(world.getMinY()).thenReturn(-64); when(world.getMaxY()).thenReturn(319); + MemoryOptimizedHistory changeSet = new MemoryOptimizedHistory(world); + ExecutorService executor = Executors.newFixedThreadPool(THREADS); try { - MemoryOptimizedHistory changeSet = new MemoryOptimizedHistory(world); - - ExecutorService executor = Executors.newFixedThreadPool(THREADS); CountDownLatch startLatch = new CountDownLatch(1); List> futures = new ArrayList<>(THREADS); @@ -84,6 +84,10 @@ void addIsAccurateUnderConcurrentWriters() throws InterruptedException { assertEquals((long) THREADS * CALLS_PER_THREAD, changeSet.longSize()); } finally { + // shutdownNow() runs even on the failure paths above (a worker throwing or timing + // out), so a stuck/failed worker never leaks non-daemon threads into the rest of the + // suite. It's a no-op once the graceful shutdown() above has already succeeded. + executor.shutdownNow(); Settings.settings().HISTORY.COMPRESSION_LEVEL = previousCompressionLevel; } } From f7bc001dcc9e769ebd405af637ac4ba03b5d7954 Mon Sep 17 00:00:00 2001 From: Matt <4009945+MattBDev@users.noreply.github.com> Date: Wed, 15 Jul 2026 08:35:50 -0400 Subject: [PATCH 3/3] Pre-initialize the lazy block stream before the concurrent phase This test's PR only migrates blockSize to a LongAdder; it doesn't include the (separately-fixed, sibling-PR) double-checked-locking fix for getBlockOS(). Without a single-threaded warm-up call first, the first wave of concurrent add() calls below would race on that unrelated, still-present lazy-init bug too, rather than exercising only the counter accuracy this test targets. getBlockOS() itself doesn't touch blockSize, so the expected total asserted at the end is unaffected. Co-Authored-By: Claude Sonnet 5 --- .../changeset/FaweStreamChangeSetBlockSizeTest.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java index 7ad7c0ee2b..d49ea91a8e 100644 --- a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java +++ b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java @@ -41,7 +41,7 @@ class FaweStreamChangeSetBlockSizeTest { private static final int CALLS_PER_THREAD = 1000; @Test - void addIsAccurateUnderConcurrentWriters() throws InterruptedException { + void addIsAccurateUnderConcurrentWriters() throws InterruptedException, java.io.IOException { int previousCompressionLevel = Settings.settings().HISTORY.COMPRESSION_LEVEL; // Compression level 0 skips the LZ4/Zstd codecs entirely (see MainUtil#getCompressedOS), // which keeps this test independent of the compileOnly lz4-java dependency. @@ -54,6 +54,15 @@ void addIsAccurateUnderConcurrentWriters() throws InterruptedException { MemoryOptimizedHistory changeSet = new MemoryOptimizedHistory(world); ExecutorService executor = Executors.newFixedThreadPool(THREADS); try { + // Pre-initialize the lazy block-output stream single-threaded, before any concurrent + // add() calls. getBlockOS()'s own double-checked-locking race is fixed independently + // in a sibling PR, not this one - on this branch it's still present, so without this + // warm-up the first wave of concurrent add() calls below would race on that unrelated + // lazy-init path too, rather than exercising only the blockSize counter this test + // targets. getBlockOS() itself doesn't touch blockSize, so this doesn't affect the + // expected total asserted below. + changeSet.getBlockOS(0, 0, 0); + CountDownLatch startLatch = new CountDownLatch(1); List> futures = new ArrayList<>(THREADS);