Additional hopper transaction logging hardening

.gitignore

@@ -181,6 +181,7 @@ CLAUDE.md
 .clinerules
 context.json
 context.yml
+issues.md
 run.sh
 run_*.sh
 no-log4j2.xml

src/main/java/net/coreprotect/database/ContainerRollback.java

@@ -15,6 +15,7 @@ import org.bukkit.entity.ArmorStand;
 import org.bukkit.entity.Entity;
 import org.bukkit.entity.ItemFrame;
 import org.bukkit.entity.LivingEntity;
+import org.bukkit.inventory.Inventory;
 import org.bukkit.inventory.ItemStack;
 
 import net.coreprotect.CoreProtect;
@@ -24,6 +25,7 @@ import net.coreprotect.consumer.process.Process;
 import net.coreprotect.database.rollback.Rollback;
 import net.coreprotect.database.rollback.RollbackComplete;
 import net.coreprotect.language.Phrase;
+import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.thread.Scheduler;
 import net.coreprotect.utility.BlockUtils;
@@ -91,6 +93,9 @@ public class ContainerRollback extends Rollback {
 
                         int modifyCount = 0;
                         if (container != null) {
+                            if (container instanceof Inventory) {
+                                InventoryChangeListener.flushPendingContainer((Inventory) container, location);
+                            }
                             for (Object[] lookupRow : lookupList) {
                                 // int unixtimestamp = (int) (System.currentTimeMillis() / 1000L);
                                 // int rowId = lookupRow[0];

src/main/java/net/coreprotect/database/Database.java

@@ -24,11 +24,11 @@ import net.coreprotect.consumer.Consumer;
 import net.coreprotect.consumer.Queue;
 import net.coreprotect.consumer.process.Process;
 import net.coreprotect.language.Phrase;
+import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.model.rollback.RollbackUpdateTargets;
 import net.coreprotect.utility.Chat;
 import net.coreprotect.utility.Color;
-import net.coreprotect.utility.HopperTransactionUtils;
 import net.coreprotect.utility.ItemUtils;
 import net.coreprotect.utility.ErrorReporter;
 
@@ -189,15 +189,19 @@ public class Database extends Queue {
     }
 
     public static void containerBreakCheck(String user, Material type, Object container, ItemStack[] contents, Location location) {
-        if (BlockGroup.CONTAINERS.contains(type) && !BlockGroup.SHULKER_BOXES.contains(type)) {
+        if (BlockGroup.CONTAINERS.contains(type)) {
             if (Config.getConfig(location.getWorld()).ITEM_TRANSACTIONS) {
                 try {
                     if (contents == null) {
                         contents = ItemUtils.getContainerContents(type, container, location);
                     }
                     if (contents != null) {
-                        setForceContainer(HopperTransactionUtils.getLoggingId(user, location), ItemUtils.getContainerState(contents));
-                        Queue.queueContainerBreak(user, location, type, contents);
+                        if (BlockGroup.SHULKER_BOXES.contains(type)) {
+                            InventoryChangeListener.flushPendingContainer(location, contents);
+                        }
+                        else {
+                            InventoryChangeListener.queueContainerBreak(user, location, type, contents);
+                        }
                     }
                 }
                 catch (Exception e) {

src/main/java/net/coreprotect/database/rollback/RollbackBlockHandler.java

@@ -41,6 +41,7 @@ import org.bukkit.inventory.ItemStack;
 
 import net.coreprotect.bukkit.BukkitAdapter;
 import net.coreprotect.consumer.Queue;
+import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.model.PendingBlockChange;
 import net.coreprotect.paper.PaperAdapter;
@@ -168,6 +169,7 @@ public class RollbackBlockHandler extends Queue {
                         if (BlockGroup.CONTAINERS.contains(changeType)) {
                             Inventory inventory = BlockUtils.getContainerInventory(block.getState(), false);
                             if (inventory != null) {
+                                InventoryChangeListener.flushPendingContainer(inventory, block.getLocation());
                                 inventory.clear();
                             }
                         }
@@ -284,6 +286,7 @@ public class RollbackBlockHandler extends Queue {
                     }
                     if (meta != null) {
                         Inventory inventory = BlockUtils.getContainerInventory(block.getState(), false);
+                        InventoryChangeListener.flushPendingContainer(inventory, block.getLocation());
                         for (Object value : meta) {
                             ItemStack item = ItemUtils.unserializeItemStackLegacy(value);
                             if (item != null) {

src/main/java/net/coreprotect/database/rollback/RollbackProcessor.java

@@ -24,12 +24,14 @@ import org.bukkit.entity.Entity;
 import org.bukkit.entity.ItemFrame;
 import org.bukkit.entity.LivingEntity;
 import org.bukkit.entity.Player;
+import org.bukkit.inventory.Inventory;
 import org.bukkit.inventory.ItemStack;
 
 import net.coreprotect.CoreProtect;
 import net.coreprotect.bukkit.BukkitAdapter;
 import net.coreprotect.config.Config;
 import net.coreprotect.config.ConfigHandler;
+import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.model.PendingBlockChange;
 import net.coreprotect.model.item.ItemTransactionActions;
@@ -313,6 +315,9 @@ public class RollbackProcessor {
                                 }
                                 else {
                                     container = BlockUtils.getContainerInventory(blockState, false);
+                                    if (container instanceof Inventory) {
+                                        InventoryChangeListener.flushPendingContainer((Inventory) container, block.getLocation());
+                                    }
                                 }
 
                                 containerType = block.getType();

src/main/java/net/coreprotect/listener/player/ContainerTransactionDispatcher.java

@@ -0,0 +1,91 @@
+package net.coreprotect.listener.player;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.bukkit.Location;
+
+import net.coreprotect.CoreProtect;
+import net.coreprotect.thread.Scheduler;
+import net.coreprotect.utility.ErrorReporter;
+import net.coreprotect.utility.HopperTransactionUtils;
+
+final class ContainerTransactionDispatcher {
+
+    private static final int STRIPE_COUNT = 4;
+    private static final Stripe[] STRIPES = new Stripe[STRIPE_COUNT];
+
+    static {
+        for (int i = 0; i < STRIPE_COUNT; i++) {
+            STRIPES[i] = new Stripe();
+        }
+    }
+
+    private ContainerTransactionDispatcher() {
+        throw new IllegalStateException("Utility class");
+    }
+
+    static void submit(Location location, Runnable task) {
+        if (location == null || location.getWorld() == null || task == null) {
+            return;
+        }
+
+        String transactionId = HopperTransactionUtils.getTransactionId(location);
+        Stripe stripe = STRIPES[(transactionId.hashCode() & Integer.MAX_VALUE) % STRIPE_COUNT];
+        stripe.submit(task);
+    }
+
+    static int pendingTasks() {
+        int pending = 0;
+        for (Stripe stripe : STRIPES) {
+            pending += stripe.queue.size();
+        }
+        return pending;
+    }
+
+    static void drainForShutdown() {
+        for (Stripe stripe : STRIPES) {
+            stripe.drain();
+        }
+    }
+
+    private static final class Stripe {
+        private final ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();
+        private final AtomicBoolean running = new AtomicBoolean(false);
+
+        private void submit(Runnable task) {
+            queue.add(task);
+            start();
+        }
+
+        private void start() {
+            if (!running.compareAndSet(false, true)) {
+                return;
+            }
+
+            Scheduler.runTaskAsynchronously(CoreProtect.getInstance(), this::drain);
+        }
+
+        private void drain() {
+            synchronized (this) {
+                try {
+                    Runnable task;
+                    while ((task = queue.poll()) != null) {
+                        try {
+                            task.run();
+                        }
+                        catch (Exception e) {
+                            ErrorReporter.report(e);
+                        }
+                    }
+                }
+                finally {
+                    running.set(false);
+                    if (!queue.isEmpty()) {
+                        start();
+                    }
+                }
+            }
+        }
+    }
+}

src/main/java/net/coreprotect/listener/player/HopperPullAccumulator.java

@@ -0,0 +1,175 @@
+package net.coreprotect.listener.player;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.bukkit.inventory.ItemStack;
+
+final class HopperPullAccumulator {
+
+    private final List<MoveRun> runs = new ArrayList<>();
+    private long moveCount;
+
+    void add(String user, ItemStack item) {
+        if (user == null || user.isEmpty() || item == null || item.getAmount() <= 0) {
+            return;
+        }
+
+        int amount = item.getAmount();
+        MoveRun tail = runs.isEmpty() ? null : runs.get(runs.size() - 1);
+        if (tail != null && tail.user.equals(user) && tail.amount == amount && tail.item.isSimilar(item)) {
+            tail.repetitions++;
+        }
+        else {
+            ItemStack prototype = item.clone();
+            prototype.setAmount(amount);
+            runs.add(new MoveRun(user, prototype, amount));
+        }
+        moveCount++;
+    }
+
+    boolean isEmpty() {
+        return runs.isEmpty();
+    }
+
+    int runCount() {
+        return runs.size();
+    }
+
+    long moveCount() {
+        return moveCount;
+    }
+
+    void replay(ItemStack[] checkpoint, int inventoryMaxStackSize, MoveConsumer consumer) {
+        if (checkpoint == null || consumer == null || runs.isEmpty()) {
+            return;
+        }
+
+        ItemStack[] state = checkpoint.clone();
+        for (int i = runs.size() - 1; i >= 0; i--) {
+            MoveRun run = runs.get(i);
+            state = addToState(state, run.item, run.totalAmount(), inventoryMaxStackSize);
+        }
+
+        for (MoveRun run : runs) {
+            for (long repetition = 0; repetition < run.repetitions; repetition++) {
+                ItemStack movedItem = run.item.clone();
+                movedItem.setAmount(run.amount);
+                consumer.accept(run.user, state, movedItem);
+                state = removeFromState(state, movedItem);
+            }
+        }
+    }
+
+    private static ItemStack[] addToState(ItemStack[] state, ItemStack item, long amount, int inventoryMaxStackSize) {
+        ItemStack[] result = state.clone();
+        long remaining = amount;
+        int maxStackSize = maxStackSize(item, inventoryMaxStackSize);
+
+        for (int i = 0; i < result.length && remaining > 0; i++) {
+            ItemStack existing = result[i];
+            if (existing == null || existing.getAmount() <= 0 || !existing.isSimilar(item) || existing.getAmount() >= maxStackSize) {
+                continue;
+            }
+
+            int added = (int) Math.min(remaining, maxStackSize - existing.getAmount());
+            ItemStack updated = existing.clone();
+            updated.setAmount(existing.getAmount() + added);
+            result[i] = updated;
+            remaining -= added;
+        }
+
+        for (int i = 0; i < result.length && remaining > 0; i++) {
+            ItemStack existing = result[i];
+            if (existing != null && existing.getAmount() > 0) {
+                continue;
+            }
+
+            int added = (int) Math.min(remaining, maxStackSize);
+            ItemStack inserted = item.clone();
+            inserted.setAmount(added);
+            result[i] = inserted;
+            remaining -= added;
+        }
+
+        if (remaining <= 0) {
+            return result;
+        }
+
+        long additionalSlots = (remaining + maxStackSize - 1L) / maxStackSize;
+        if (additionalSlots > Integer.MAX_VALUE - result.length) {
+            throw new IllegalStateException("Hopper pull reconstruction exceeds maximum inventory size");
+        }
+
+        ItemStack[] expanded = new ItemStack[result.length + (int) additionalSlots];
+        System.arraycopy(result, 0, expanded, 0, result.length);
+        int slot = result.length;
+        while (remaining > 0) {
+            int added = (int) Math.min(remaining, maxStackSize);
+            ItemStack inserted = item.clone();
+            inserted.setAmount(added);
+            expanded[slot] = inserted;
+            slot++;
+            remaining -= added;
+        }
+        return expanded;
+    }
+
+    private static ItemStack[] removeFromState(ItemStack[] state, ItemStack item) {
+        ItemStack[] result = state.clone();
+        int remaining = item.getAmount();
+        for (int i = 0; i < result.length && remaining > 0; i++) {
+            ItemStack existing = result[i];
+            if (existing == null || existing.getAmount() <= 0 || !existing.isSimilar(item)) {
+                continue;
+            }
+
+            int removed = Math.min(remaining, existing.getAmount());
+            if (removed == existing.getAmount()) {
+                result[i] = null;
+            }
+            else {
+                ItemStack updated = existing.clone();
+                updated.setAmount(existing.getAmount() - removed);
+                result[i] = updated;
+            }
+            remaining -= removed;
+        }
+        return result;
+    }
+
+    private static int maxStackSize(ItemStack item, int inventoryMaxStackSize) {
+        int maxStackSize = item.getMaxStackSize();
+        if (inventoryMaxStackSize > 0 && (inventoryMaxStackSize < maxStackSize || maxStackSize < 1)) {
+            maxStackSize = inventoryMaxStackSize;
+        }
+        return Math.max(1, maxStackSize);
+    }
+
+    @FunctionalInterface
+    interface MoveConsumer {
+        void accept(String user, ItemStack[] sourceBefore, ItemStack movedItem);
+    }
+
+    private static final class MoveRun {
+        private final String user;
+        private final ItemStack item;
+        private final int amount;
+        private long repetitions = 1;
+
+        private MoveRun(String user, ItemStack item, int amount) {
+            this.user = user;
+            this.item = item;
+            this.amount = amount;
+        }
+
+        private long totalAmount() {
+            try {
+                return Math.multiplyExact((long) amount, repetitions);
+            }
+            catch (ArithmeticException e) {
+                throw new IllegalStateException("Hopper pull amount overflow", e);
+            }
+        }
+    }
+}

src/main/java/net/coreprotect/listener/player/HopperPullListener.java

@@ -1,10 +1,10 @@
 package net.coreprotect.listener.player;
 
 import java.util.Arrays;
+import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.bukkit.Location;
 import org.bukkit.inventory.Inventory;
@@ -21,139 +21,157 @@ import net.coreprotect.utility.ErrorReporter;
 
 public final class HopperPullListener {
 
-    private static final ConcurrentLinkedQueue<Object[]> hopperQueue = new ConcurrentLinkedQueue<>();
-    private static final AtomicBoolean processorRunning = new AtomicBoolean(false);
-    private static final int BATCH_SIZE = 100;
-    private static final int MAX_CONCURRENT_PROCESSORS = 4;
-    private static final AtomicInteger activeProcessors = new AtomicInteger(0);
+    private static final Map<String, PendingPull> pendingPulls = new ConcurrentHashMap<>();
+
+    private HopperPullListener() {
+        throw new IllegalStateException("Listener class");
+    }
 
     static void processHopperPull(Location location, String user, InventoryHolder sourceHolder, InventoryHolder destinationHolder, ItemStack item) {
+        if (location == null || location.getWorld() == null || sourceHolder == null || destinationHolder == null || item == null || item.getAmount() <= 0) {
+            return;
+        }
+
+        Inventory sourceInventory = sourceHolder.getInventory();
+        Inventory destinationInventory = destinationHolder.getInventory();
         String loggingChestId = HopperTransactionUtils.getHopperPullId(location);
         Object[] lastAbort = ConfigHandler.hopperAbort.get(loggingChestId);
-        if (lastAbort != null) {
-            ItemStack[] destinationContents = destinationHolder.getInventory().getContents();
-            if (((Set<?>) lastAbort[0]).contains(item) && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
-                return;
-            }
+        ItemStack[] destinationContents = destinationInventory.getContents();
+        if (lastAbort != null && ((Set<?>) lastAbort[0]).contains(item) && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
+            return;
         }
 
-        ItemStack[] destinationContainer = ItemUtils.getContainerState(destinationHolder.getInventory().getContents());
+        ItemStack[] destinationContainer = ItemUtils.getContainerState(destinationContents);
         ItemStack movedItem = item.clone();
-
-        // Queue the hopper pull operation instead of creating a new thread immediately
-        hopperQueue.add(new Object[] { location, user, sourceHolder, destinationHolder, movedItem, destinationContainer, loggingChestId, lastAbort });
-
-        // Start the processor if it's not already running
-        if (processorRunning.compareAndSet(false, true)) {
-            startHopperProcessor();
-        }
-    }
-
-    private static void startHopperProcessor() {
-        if (activeProcessors.incrementAndGet() <= MAX_CONCURRENT_PROCESSORS) {
-            Scheduler.runTaskAsynchronously(CoreProtect.getInstance(), () -> {
-                try {
-                    // Use the same server running check as Consumer class
-                    while (!hopperQueue.isEmpty() && (ConfigHandler.serverRunning || ConfigHandler.converterRunning)) {
-                        processHopperBatch();
-                    }
-                }
-                catch (Exception e) {
-                    ErrorReporter.report(e);
-                }
-                finally {
-                    activeProcessors.decrementAndGet();
-                    processorRunning.set(false);
-
-                    // If more items were added and we're still running, restart the processor
-                    if (!hopperQueue.isEmpty() && (ConfigHandler.serverRunning || ConfigHandler.converterRunning) && activeProcessors.get() == 0) {
-                        startHopperProcessor();
-                    }
-                }
-            });
-        }
-        else {
-            activeProcessors.decrementAndGet();
-        }
-    }
-
-    private static void processHopperBatch() {
-        int processed = 0;
-        final long taskStarted = InventoryChangeListener.tasksStarted.incrementAndGet();
-
-        while (!hopperQueue.isEmpty() && processed < BATCH_SIZE && (ConfigHandler.serverRunning || ConfigHandler.converterRunning)) {
-            Object[] data = hopperQueue.poll();
-            if (data == null)
-                continue;
-
-            Location location = (Location) data[0];
-            String user = (String) data[1];
-            InventoryHolder sourceHolder = (InventoryHolder) data[2];
-            InventoryHolder destinationHolder = (InventoryHolder) data[3];
-            ItemStack movedItem = (ItemStack) data[4];
-            ItemStack[] destinationContainer = (ItemStack[]) data[5];
-            String loggingChestId = (String) data[6];
-            Object[] lastAbort = (Object[]) data[7];
-
-            processSingleHopperPull(location, user, sourceHolder, destinationHolder, movedItem, destinationContainer, loggingChestId, lastAbort);
-
-            processed++;
-        }
-
-        // Ensure sequential processing just like the original code
-        InventoryChangeListener.checkTasks(taskStarted);
-    }
-
-    private static void processSingleHopperPull(Location location, String user, InventoryHolder sourceHolder, InventoryHolder destinationHolder, ItemStack movedItem, ItemStack[] destinationContainer, String loggingChestId, Object[] lastAbort) {
-
-        if (sourceHolder == null || destinationHolder == null) {
+        if (!ItemUtils.canAddContainer(destinationContainer, movedItem, destinationInventory.getMaxStackSize())) {
+            ConfigHandler.hopperAbort.put(loggingChestId, HopperTransactionUtils.createAbortState(lastAbort, destinationContainer, movedItem));
             return;
         }
 
-        boolean abort = false;
-        boolean addedInventory = ItemUtils.canAddContainer(destinationContainer, movedItem, destinationHolder.getInventory().getMaxStackSize());
-        if (!addedInventory) {
-            abort = true;
-        }
-
-        if (abort) {
-            ItemStack[] destinationContents = destinationHolder.getInventory().getContents();
-            ConfigHandler.hopperAbort.put(loggingChestId, HopperTransactionUtils.createAbortState(lastAbort, destinationContents, movedItem));
-            return;
-        }
-
-        if (ConfigHandler.hopperAbort.get(loggingChestId) != null) {
-            ConfigHandler.hopperAbort.remove(loggingChestId);
-        }
-
+        ConfigHandler.hopperAbort.remove(loggingChestId);
         boolean hopperTransactions = Config.getConfig(location.getWorld()).HOPPER_TRANSACTIONS;
         if (!hopperTransactions) {
             HopperTransactionUtils.recordItemRemoved(HopperTransactionUtils.getTransactionId(location), movedItem);
             return;
         }
 
-        Location destinationLocation = destinationHolder.getInventory().getLocation();
+        Location destinationLocation = destinationInventory.getLocation();
         if (destinationLocation == null) {
             return;
         }
 
         HopperTransactionUtils.recordItemAdded(HopperTransactionUtils.getTransactionId(destinationLocation), movedItem);
-
+        if (!Config.getConfig(location.getWorld()).ITEM_TRANSACTIONS) {
+            return;
+        }
         if (Config.getConfig(location.getWorld()).HOPPER_FILTER_META && !movedItem.hasItemMeta()) {
+            HopperTransactionUtils.recordItemRemoved(HopperTransactionUtils.getTransactionId(location), movedItem);
             return;
         }
 
-        Inventory sourceInventory = sourceHolder.getInventory();
-        ItemStack[] inventoryContents = sourceInventory.getContents();
-        ItemStack[] originalSource = new ItemStack[inventoryContents.length + 1];
-        for (int i = 0; i < inventoryContents.length; i++) {
-            ItemStack itemStack = inventoryContents[i];
-            if (itemStack != null) {
-                originalSource[i] = itemStack.clone();
+        String transactionId = HopperTransactionUtils.getTransactionId(location);
+        PendingPull pending = pendingPulls.compute(transactionId, (key, current) -> {
+            PendingPull value = current;
+            if (value == null) {
+                value = new PendingPull(sourceInventory, location.clone(), sourceInventory.getMaxStackSize());
             }
+            value.accumulator.add(user, movedItem);
+            return value;
+        });
+
+        if (pending.scheduled.compareAndSet(false, true)) {
+            Scheduler.runTask(CoreProtect.getInstance(), () -> capturePendingPull(transactionId, pending), pending.location);
+        }
+    }
+
+    static void flushPendingPull(Location location, Inventory inventory, ItemStack[] checkpoint) {
+        if (location == null || location.getWorld() == null || checkpoint == null) {
+            return;
         }
 
-        originalSource[inventoryContents.length] = movedItem;
-        InventoryChangeListener.onInventoryInteract(user, sourceInventory, originalSource, null, sourceInventory.getLocation(), true);
+        String transactionId = HopperTransactionUtils.getTransactionId(location);
+        PendingPull pending = pendingPulls.remove(transactionId);
+        if (pending == null) {
+            return;
+        }
+
+        Inventory sourceInventory = inventory == null ? pending.inventory : inventory;
+        submitPendingPull(pending, sourceInventory, checkpoint);
+    }
+
+    static void flushPendingPull(Location location, Inventory inventory) {
+        if (location == null || location.getWorld() == null || inventory == null) {
+            return;
+        }
+
+        String transactionId = HopperTransactionUtils.getTransactionId(location);
+        PendingPull pending = pendingPulls.get(transactionId);
+        if (pending == null) {
+            return;
+        }
+
+        ItemStack[] checkpoint = ItemUtils.getContainerState(inventory.getContents());
+        if (pendingPulls.remove(transactionId, pending)) {
+            submitPendingPull(pending, inventory, checkpoint);
+        }
+    }
+
+    static void flushPendingPullsForShutdown() {
+        for (Map.Entry<String, PendingPull> entry : pendingPulls.entrySet()) {
+            PendingPull pending = entry.getValue();
+            try {
+                ItemStack[] checkpoint = ItemUtils.getContainerState(pending.inventory.getContents());
+                if (pendingPulls.remove(entry.getKey(), pending)) {
+                    submitPendingPull(pending, pending.inventory, checkpoint);
+                }
+            }
+            catch (Exception e) {
+                ErrorReporter.report(e);
+            }
+        }
+    }
+
+    static int pendingLocations() {
+        return pendingPulls.size();
+    }
+
+    static long pendingMoves() {
+        long count = 0;
+        for (PendingPull pending : pendingPulls.values()) {
+            count += pending.accumulator.moveCount();
+        }
+        return count;
+    }
+
+    private static void capturePendingPull(String transactionId, PendingPull pending) {
+        if (!pendingPulls.remove(transactionId, pending)) {
+            return;
+        }
+
+        ItemStack[] checkpoint = ItemUtils.getContainerState(pending.inventory.getContents());
+        submitPendingPull(pending, pending.inventory, checkpoint);
+    }
+
+    private static void submitPendingPull(PendingPull pending, Inventory sourceInventory, ItemStack[] checkpoint) {
+        if (pending == null || sourceInventory == null || checkpoint == null || pending.accumulator.isEmpty()) {
+            return;
+        }
+
+        ContainerTransactionDispatcher.submit(pending.location, () -> pending.accumulator.replay(checkpoint, pending.maxStackSize,
+                (user, sourceBefore, movedItem) -> InventoryChangeListener.onHopperInventoryInteract(user, sourceInventory, sourceBefore, pending.location, movedItem)));
+    }
+
+    private static final class PendingPull {
+        private final Inventory inventory;
+        private final Location location;
+        private final int maxStackSize;
+        private final HopperPullAccumulator accumulator = new HopperPullAccumulator();
+        private final AtomicBoolean scheduled = new AtomicBoolean(false);
+
+        private PendingPull(Inventory inventory, Location location, int maxStackSize) {
+            this.inventory = inventory;
+            this.location = location;
+            this.maxStackSize = maxStackSize;
+        }
     }
 }

src/main/java/net/coreprotect/listener/player/HopperPushListener.java

@@ -2,161 +2,61 @@ package net.coreprotect.listener.player;
 
 import java.util.Arrays;
 import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.bukkit.Location;
 import org.bukkit.inventory.Inventory;
 import org.bukkit.inventory.InventoryHolder;
 import org.bukkit.inventory.ItemStack;
 
-import net.coreprotect.CoreProtect;
 import net.coreprotect.config.Config;
 import net.coreprotect.config.ConfigHandler;
-import net.coreprotect.thread.Scheduler;
 import net.coreprotect.utility.HopperTransactionUtils;
 import net.coreprotect.utility.ItemUtils;
-import net.coreprotect.utility.ErrorReporter;
 
 public final class HopperPushListener {
 
-    private static final ConcurrentLinkedQueue<Object[]> hopperQueue = new ConcurrentLinkedQueue<>();
-    private static final AtomicBoolean processorRunning = new AtomicBoolean(false);
-    private static final int BATCH_SIZE = 100;
-    private static final int PROCESS_DELAY_TICKS = 1;
-    private static final int MAX_CONCURRENT_PROCESSORS = 4;
-    private static final AtomicInteger activeProcessors = new AtomicInteger(0);
+    private HopperPushListener() {
+        throw new IllegalStateException("Listener class");
+    }
 
     static void processHopperPush(Location location, String user, InventoryHolder sourceHolder, InventoryHolder destinationHolder, ItemStack item) {
-        Location destinationLocation = destinationHolder.getInventory().getLocation();
-        if (destinationLocation == null) {
+        if (location == null || location.getWorld() == null || sourceHolder == null || destinationHolder == null || item == null || item.getAmount() <= 0) {
+            return;
+        }
+
+        Inventory destinationInventory = destinationHolder.getInventory();
+        Location destinationLocation = destinationInventory.getLocation();
+        if (destinationLocation == null || destinationLocation.getWorld() == null) {
             return;
         }
 
         String loggingChestId = HopperTransactionUtils.getHopperPushId(destinationLocation);
         Object[] lastAbort = ConfigHandler.hopperAbort.get(loggingChestId);
-        if (lastAbort != null) {
-            ItemStack[] destinationContents = destinationHolder.getInventory().getContents();
-            if (((Set<?>) lastAbort[0]).contains(item) && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
-                return;
-            }
-        }
-
-        ItemStack[] destinationContainer = ItemUtils.getContainerState(destinationHolder.getInventory().getContents());
-        ItemStack movedItem = item.clone();
-
-        hopperQueue.add(new Object[] { location, user, sourceHolder, destinationHolder, movedItem, destinationContainer, loggingChestId, lastAbort });
-
-        // Start the processor if it's not already running
-        if (processorRunning.compareAndSet(false, true)) {
-            startHopperProcessor();
-        }
-    }
-
-    private static void startHopperProcessor() {
-        if (activeProcessors.incrementAndGet() <= MAX_CONCURRENT_PROCESSORS) {
-            Scheduler.runTaskAsynchronously(CoreProtect.getInstance(), () -> {
-                try {
-                    // Use the same server running check as Consumer class
-                    while (!hopperQueue.isEmpty() && (ConfigHandler.serverRunning || ConfigHandler.converterRunning)) {
-                        processHopperBatch();
-                    }
-                }
-                catch (Exception e) {
-                    ErrorReporter.report(e);
-                }
-                finally {
-                    activeProcessors.decrementAndGet();
-                    processorRunning.set(false);
-
-                    // If more items were added and we're still running, restart the processor
-                    if (!hopperQueue.isEmpty() && (ConfigHandler.serverRunning || ConfigHandler.converterRunning) && activeProcessors.get() == 0) {
-                        startHopperProcessor();
-                    }
-                }
-            });
-        }
-        else {
-            activeProcessors.decrementAndGet();
-        }
-    }
-
-    private static void processHopperBatch() {
-        int processed = 0;
-        final long taskStarted = InventoryChangeListener.tasksStarted.incrementAndGet();
-
-        while (!hopperQueue.isEmpty() && processed < BATCH_SIZE && (ConfigHandler.serverRunning || ConfigHandler.converterRunning)) {
-            Object[] data = hopperQueue.poll();
-            if (data == null)
-                continue;
-
-            Location location = (Location) data[0];
-            String user = (String) data[1];
-            InventoryHolder sourceHolder = (InventoryHolder) data[2];
-            InventoryHolder destinationHolder = (InventoryHolder) data[3];
-            ItemStack movedItem = (ItemStack) data[4];
-            ItemStack[] destinationContainer = (ItemStack[]) data[5];
-            String loggingChestId = (String) data[6];
-            Object[] lastAbort = (Object[]) data[7];
-
-            processSingleHopperPush(location, user, sourceHolder, destinationHolder, movedItem, destinationContainer, loggingChestId, lastAbort);
-
-            processed++;
-        }
-
-        InventoryChangeListener.checkTasks(taskStarted);
-    }
-
-    private static void processSingleHopperPush(Location location, String user, InventoryHolder sourceHolder, InventoryHolder destinationHolder, ItemStack movedItem, ItemStack[] destinationContainer, String loggingChestId, Object[] lastAbort) {
-
-        if (sourceHolder == null || destinationHolder == null) {
+        ItemStack[] destinationContents = destinationInventory.getContents();
+        if (lastAbort != null && ((Set<?>) lastAbort[0]).contains(item) && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
             return;
         }
 
-        boolean abort = false;
-        boolean addedInventory = ItemUtils.canAddContainer(destinationContainer, movedItem, destinationHolder.getInventory().getMaxStackSize());
-        if (!addedInventory) {
-            abort = true;
-        }
-
-        if (abort) {
-            ItemStack[] destinationContents = destinationHolder.getInventory().getContents();
-            ConfigHandler.hopperAbort.put(loggingChestId, HopperTransactionUtils.createAbortState(lastAbort, destinationContents, movedItem));
+        ItemStack[] destinationContainer = ItemUtils.getContainerState(destinationContents);
+        ItemStack movedItem = item.clone();
+        if (!ItemUtils.canAddContainer(destinationContainer, movedItem, destinationInventory.getMaxStackSize())) {
+            ConfigHandler.hopperAbort.put(loggingChestId, HopperTransactionUtils.createAbortState(lastAbort, destinationContainer, movedItem));
             return;
         }
 
         HopperTransactionUtils.recordItemRemoved(HopperTransactionUtils.getTransactionId(location), movedItem);
-
+        HopperPullListener.flushPendingPull(destinationLocation, destinationInventory, destinationContainer);
+        if (!Config.getConfig(location.getWorld()).ITEM_TRANSACTIONS) {
+            return;
+        }
         if (Config.getConfig(location.getWorld()).HOPPER_FILTER_META && !movedItem.hasItemMeta()) {
+            HopperTransactionUtils.recordItemAdded(HopperTransactionUtils.getTransactionId(destinationLocation), movedItem);
             return;
         }
 
-        ConfigHandler.hopperSuccess.put(loggingChestId, new Object[] { destinationContainer, movedItem });
-
-        Inventory destinationInventory = destinationHolder.getInventory();
-        ItemStack[] originalDestination = destinationInventory.getContents().clone();
-        int removeAmount = movedItem.getAmount();
-        for (int i = 0; i < originalDestination.length; i++) {
-            if (removeAmount == 0) {
-                break;
-            }
-
-            ItemStack itemStack = (originalDestination[i] != null ? originalDestination[i].clone() : null);
-            if (itemStack != null && itemStack.isSimilar(movedItem)) {
-                if (itemStack.getAmount() >= removeAmount) {
-                    itemStack.setAmount(itemStack.getAmount() - removeAmount);
-                    removeAmount = 0;
-                }
-                else {
-                    removeAmount = removeAmount - itemStack.getAmount();
-                    itemStack = null;
-                }
-
-                originalDestination[i] = itemStack;
-            }
-        }
-
-        InventoryChangeListener.onHopperInventoryInteract(user, destinationInventory, originalDestination, destinationInventory.getLocation(), movedItem);
+        ContainerTransactionDispatcher.submit(destinationLocation, () -> {
+            ConfigHandler.hopperSuccess.put(loggingChestId, new Object[] { destinationContainer, movedItem });
+            InventoryChangeListener.onHopperInventoryInteract(user, destinationInventory, destinationContainer, destinationLocation, movedItem);
+        });
     }
 }

src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java

@@ -8,7 +8,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
 
 import org.bukkit.Location;
 import org.bukkit.Material;
@@ -53,39 +52,8 @@ import us.lynuxcraft.deadsilenceiv.advancedchests.chest.AdvancedChest;
 
 public final class InventoryChangeListener extends Queue implements Listener {
 
-    protected static AtomicLong tasksStarted = new AtomicLong();
-    protected static AtomicLong tasksCompleted = new AtomicLong();
     private static ConcurrentHashMap<String, Boolean> inventoryProcessing = new ConcurrentHashMap<>();
     private static final Map<UUID, PendingEntityContainerTransaction> pendingEntityTransactions = new HashMap<>();
-    private static final Object taskCompletionLock = new Object();
-    private static final long TASK_WAIT_MAX_MS = 50; // Maximum wait time in milliseconds
-
-    protected static void checkTasks(long taskStarted) {
-        try {
-            // Skip checking if this is the first task or we're already caught up
-            if (taskStarted <= 1 || tasksCompleted.get() >= (taskStarted - 1L)) {
-                tasksCompleted.set(taskStarted);
-                return;
-            }
-
-            // Try to update without waiting if possible
-            if (tasksCompleted.compareAndSet(taskStarted - 1L, taskStarted)) {
-                return;
-            }
-
-            // Use proper synchronization instead of busy waiting
-            synchronized (taskCompletionLock) {
-                if (tasksCompleted.get() < (taskStarted - 1L)) {
-                    taskCompletionLock.wait(TASK_WAIT_MAX_MS);
-                }
-                tasksCompleted.set(taskStarted);
-                taskCompletionLock.notifyAll(); // Notify other waiting threads
-            }
-        }
-        catch (Exception e) {
-            ErrorReporter.report(e);
-        }
-    }
 
     public static boolean inventoryTransaction(String user, Location location, ItemStack[] inventoryData) {
         if (location != null) {
@@ -108,6 +76,43 @@ public final class InventoryChangeListener extends Queue implements Listener {
         return false;
     }
 
+    public static void queueContainerBreak(String user, Location location, Material type, ItemStack[] contents) {
+        if (user == null || user.isEmpty() || location == null || location.getWorld() == null || type == null || contents == null) {
+            return;
+        }
+
+        Location capturedLocation = location.clone();
+        Location boundaryLocation = getCanonicalContainerLocation(location);
+        ItemStack[] capturedContents = ItemUtils.getContainerState(contents);
+        HopperPullListener.flushPendingPull(boundaryLocation, null, capturedContents);
+        ContainerTransactionDispatcher.submit(boundaryLocation, () -> {
+            setForceContainer(HopperTransactionUtils.getLoggingId(user, capturedLocation), ItemUtils.getContainerState(capturedContents));
+            Queue.queueContainerBreak(user, capturedLocation, type, capturedContents);
+        });
+    }
+
+    public static void flushPendingContainer(Location location, ItemStack[] contents) {
+        if (location == null || location.getWorld() == null || contents == null) {
+            return;
+        }
+
+        HopperPullListener.flushPendingPull(getCanonicalContainerLocation(location), null, ItemUtils.getContainerState(contents));
+    }
+
+    public static void flushPendingContainer(Inventory inventory, Location location) {
+        if (inventory == null || location == null || location.getWorld() == null) {
+            return;
+        }
+
+        Location inventoryLocation = inventory.getLocation();
+        HopperPullListener.flushPendingPull(inventoryLocation == null ? location : inventoryLocation, inventory);
+    }
+
+    public static void flushPendingTransactionsForShutdown() {
+        HopperPullListener.flushPendingPullsForShutdown();
+        ContainerTransactionDispatcher.drainForShutdown();
+    }
+
     static boolean onInventoryInteract(String user, final Inventory inventory, ItemStack[] inventoryData, Material containerType, Location location, boolean aSync) {
         if (inventory != null && location != null) {
             World world = location.getWorld();
@@ -153,10 +158,18 @@ public final class InventoryChangeListener extends Queue implements Listener {
 
                 if (playerLocation != null) {
                     if (inventoryData == null) {
-                        inventoryData = inventory.getContents();
+                        inventoryData = ItemUtils.getContainerState(inventory.getContents());
+                    }
+                    else if (!aSync) {
+                        inventoryData = ItemUtils.getContainerState(inventoryData);
                     }
 
-                    return queueContainerTransaction(user, playerLocation, type, inventory, inventoryData, null, null);
+                    ItemStack[] capturedState = inventoryData;
+                    Location capturedLocation = playerLocation.clone();
+                    Material capturedType = type;
+                    HopperPullListener.flushPendingPull(capturedLocation, inventory, capturedState);
+                    ContainerTransactionDispatcher.submit(capturedLocation, () -> queueContainerTransaction(user, capturedLocation, capturedType, inventory, capturedState, null, null));
+                    return true;
                 }
             }
         }
@@ -285,8 +298,9 @@ public final class InventoryChangeListener extends Queue implements Listener {
             return;
         }
 
-        Location inventoryLocation = location;
+        Location inventoryLocation = location.clone();
         ItemStack[] containerState = ItemUtils.getContainerState(inventory.getContents());
+        HopperPullListener.flushPendingPull(inventoryLocation, inventory, containerState);
 
         String loggingChestId = player.getName() + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
         Boolean lastTransaction = inventoryProcessing.get(loggingChestId);
@@ -295,13 +309,11 @@ public final class InventoryChangeListener extends Queue implements Listener {
         }
         inventoryProcessing.put(loggingChestId, true);
 
-        final long taskStarted = InventoryChangeListener.tasksStarted.incrementAndGet();
-        Scheduler.runTaskAsynchronously(CoreProtect.getInstance(), () -> {
+        Material containerType = (enderChest != true ? null : Material.ENDER_CHEST);
+        ContainerTransactionDispatcher.submit(inventoryLocation, () -> {
             try {
-                Material containerType = (enderChest != true ? null : Material.ENDER_CHEST);
-                InventoryChangeListener.checkTasks(taskStarted);
                 inventoryProcessing.remove(loggingChestId);
-                onInventoryInteract(player.getName(), inventory, containerState, containerType, inventoryLocation, true);
+                queueContainerTransaction(player.getName(), inventoryLocation, containerType == null ? Material.CHEST : containerType, inventory, containerState, null, null);
             }
             catch (Exception e) {
                 ErrorReporter.report(e);
@@ -785,7 +797,8 @@ public final class InventoryChangeListener extends Queue implements Listener {
             onInventoryInteract(user, sourceInventory, sourceBefore, null, sourceLocation, true);
         }
         if (destinationBlockContainer && (Validate.isHopper(sourceHolder) || Validate.isDropper(sourceHolder)) && !Validate.isHopper(destinationHolder)) {
-            onHopperInventoryInteract(user, destinationInventory, destinationBefore, destinationLocation, moved);
+            HopperPullListener.flushPendingPull(destinationLocation, destinationInventory, destinationBefore);
+            ContainerTransactionDispatcher.submit(destinationLocation, () -> onHopperInventoryInteract(user, destinationInventory, destinationBefore, destinationLocation, moved));
         }
     }
 
@@ -852,10 +865,28 @@ public final class InventoryChangeListener extends Queue implements Listener {
             queueEntityContainerDelta(user, entityContainer, oldContents, newContents);
         }
         else {
-            queueContainerTransaction(user, location, Material.HOPPER, newContents, oldContents, newContents, null);
+            Location inventoryLocation = location.clone();
+            String transactionUser = user;
+            ContainerTransactionDispatcher.submit(inventoryLocation, () -> queueContainerTransaction(transactionUser, inventoryLocation, Material.HOPPER, newContents, oldContents, newContents, null));
         }
     }
 
+    private static Location getCanonicalContainerLocation(Location location) {
+        try {
+            BlockState state = location.getBlock().getState();
+            if (state instanceof InventoryHolder) {
+                Location inventoryLocation = ((InventoryHolder) state).getInventory().getLocation();
+                if (inventoryLocation != null && inventoryLocation.getWorld() != null) {
+                    return inventoryLocation.clone();
+                }
+            }
+        }
+        catch (Exception e) {
+            ErrorReporter.report(e);
+        }
+        return location.clone();
+    }
+
     private static ItemStack[] addPickedItem(ItemStack[] contents, ItemStack itemStack, int inventoryMaxStackSize) {
         if (contents == null || itemStack == null || itemStack.getAmount() <= 0 || itemStack.getType() == Material.AIR) {
             return null;

src/main/java/net/coreprotect/services/ShutdownService.java

@@ -13,6 +13,7 @@ import net.coreprotect.consumer.Consumer;
 import net.coreprotect.consumer.process.Process;
 import net.coreprotect.language.Phrase;
 import net.coreprotect.listener.player.PlayerQuitListener;
+import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.paper.PaperAdapter;
 import net.coreprotect.utility.Chat;
 import net.coreprotect.utility.Extensions;
@@ -59,6 +60,8 @@ public class ShutdownService {
                 EntitySpawnTracking.queueLoadedLocationsForShutdown();
             }
 
+            InventoryChangeListener.flushPendingTransactionsForShutdown();
+
             ConfigHandler.serverRunning = false;
             long shutdownTime = System.currentTimeMillis();
             long nextAlertTime = shutdownTime + ALERT_INTERVAL_MS;

src/main/java/net/coreprotect/utility/HopperTransactionUtils.java

@@ -142,7 +142,7 @@ public final class HopperTransactionUtils {
     }
 
     public static void recordItemAdded(String transactionId, ItemStack item) {
-        recordDelta(transactionId, item, false, 1);
+        recordDelta(transactionId, item, false, item == null ? 0 : item.getAmount());
     }
 
     public static void recordItemRemoved(String transactionId, ItemStack item) {

src/main/java/net/coreprotect/worldedit/WorldEditLogger.java

@@ -29,6 +29,7 @@ import com.sk89q.worldedit.world.block.BlockStateHolder;
 
 import net.coreprotect.config.Config;
 import net.coreprotect.consumer.Queue;
+import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.action.SignActions;
 import net.coreprotect.utility.BlockUtils;
 import net.coreprotect.utility.EntityUtils;
@@ -100,7 +101,7 @@ public class WorldEditLogger extends Queue {
                     }
                 }
                 if (containerContents != null) {
-                    Queue.queueContainerBreak(actor.getName(), location, oldType, containerContents);
+                    InventoryChangeListener.queueContainerBreak(actor.getName(), location, oldType, containerContents);
                 }
             }
             catch (Exception e) {