Make FaweStreamChangeSet.blockSize atomic and document read reentrancy

worldedit-core/build.gradle.kts

@@ -62,6 +62,7 @@ dependencies {
     // Tests
     testRuntimeOnly(libs.log4j.core)
     testImplementation(libs.parallelgzip)
+    testImplementation(libs.lz4Java)
 }
 
 tasks.test {

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;
     }

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
+ *
+ * <p><strong>Thread-safety / reentrancy warning:</strong> a single {@code FaweStreamChangeSet}
+ * instance is <em>not</em> 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).</p>
+ *
+ * <p>As a result:</p>
+ * <ul>
+ *     <li>The same instance must never be read from more than one thread at the same time
+ *     (for example, two undos racing on the same changeset).</li>
+ *     <li>Even from a single thread, one traversal (an iterator obtained from
+ *     {@link #getIterator(boolean)}/{@link #getBlockIterator(boolean)}/etc., or a populator
+ *     obtained from {@link #getCoordinatedChanges}) must be fully drained/closed before a new
+ *     traversal is started on the same instance - starting a second traversal (or calling
+ *     {@link #readHeader(InputStream)} again) while another is still in progress will corrupt
+ *     the shared decoder state.</li>
+ * </ul>
+ *
+ * <p>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.</p>
  */
 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()));

worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java

@@ -0,0 +1,104 @@
+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.
+ *
+ * <p>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.</p>
+ */
+@Isolated
+class FaweStreamChangeSetBlockSizeTest {
+
+    private static final int THREADS = 16;
+    private static final int CALLS_PER_THREAD = 1000;
+
+    @Test
+    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.
+        Settings.settings().HISTORY.COMPRESSION_LEVEL = 0;
+
+        World world = mock(World.class);
+        when(world.getMinY()).thenReturn(-64);
+        when(world.getMaxY()).thenReturn(319);
+
+        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<Future<?>> futures = new ArrayList<>(THREADS);
+            for (int t = 0; t < THREADS; t++) {
+                final int threadIndex = t;
+                Callable<Void> 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 {
+            // 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;
+        }
+    }
+
+}