CoreProtect Community Edition v24.1

README.md

@@ -41,7 +41,7 @@ Maven:
 <dependency>
     <groupId>net.coreprotect</groupId>
     <artifactId>coreprotect</artifactId>
-    <version>24.0</version>
+    <version>24.1</version>
     <scope>provided</scope>
 </dependency>
 ```

pom.xml

@@ -2,7 +2,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>net.coreprotect</groupId>
   <artifactId>CoreProtect</artifactId>
-  <version>24.0</version>
+  <version>24.1</version>
   <properties>
     <project.branch></project.branch>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

src/main/java/net/coreprotect/api/MessageAPI.java

@@ -92,11 +92,11 @@ public class MessageAPI {
         }
 
         String collate = Config.getGlobal().MYSQL ? "" : " COLLATE NOCASE";
-        try (PreparedStatement statement = connection.prepareStatement("SELECT rowid FROM " + ConfigHandler.prefix + "user WHERE user = ?" + collate + " ORDER BY rowid ASC LIMIT 0, 1")) {
+        try (PreparedStatement statement = connection.prepareStatement("SELECT rowid AS id FROM " + ConfigHandler.prefix + "user WHERE user = ?" + collate + " ORDER BY rowid ASC LIMIT 0, 1")) {
             statement.setString(1, user);
             try (ResultSet results = statement.executeQuery()) {
                 if (results.next()) {
-                    int id = results.getInt("rowid");
+                    int id = results.getInt("id");
                     ConfigHandler.playerIdCache.put(user.toLowerCase(java.util.Locale.ROOT), id);
                     ConfigHandler.playerIdCacheReversed.put(id, user);
                     return id;

src/main/java/net/coreprotect/command/TabHandler.java

@@ -43,7 +43,7 @@ public class TabHandler implements TabCompleter {
         String currentArg = args[args.length - 1].toLowerCase(Locale.ROOT).trim();
         String lastArg = args.length > 1 ? args[args.length - 2].toLowerCase(Locale.ROOT).trim() : "";
 
-        ParamState paramState = getParamState(args);
+        ParamState paramState = getParamState(sender, args);
 
         // Handle param-specific completions
         if (isActionParam(lastArg, currentArg) && hasLookupPermission(sender)) {
@@ -160,10 +160,11 @@ public class TabHandler implements TabCompleter {
         boolean pageLookup;
     }
 
-    private ParamState getParamState(String[] args) {
+    private ParamState getParamState(CommandSender sender, String[] args) {
         ParamState state = new ParamState();
+        String senderName = sender.getName();
 
-        if (ConfigHandler.lookupType.get(args[0]) != null && ConfigHandler.lookupPage.get(args[0]) != null) {
+        if (ConfigHandler.lookupType.get(senderName) != null && ConfigHandler.lookupPage.get(senderName) != null) {
             state.pageLookup = true;
         }
 
@@ -202,13 +203,14 @@ public class TabHandler implements TabCompleter {
         }
 
         if (!state.hasContainer) {
-            if (ConfigHandler.lookupType.get(args[0]) != null) {
-                int lookupType = ConfigHandler.lookupType.get(args[0]);
+            Integer lookupType = ConfigHandler.lookupType.get(senderName);
+            if (lookupType != null) {
                 if (lookupType == 1) {
                     state.validContainer = true;
                 }
                 else if (lookupType == 5) {
-                    if (ConfigHandler.lookupUlist.get(args[0]).contains("#container")) {
+                    List<String> lookupUsers = ConfigHandler.lookupUlist.get(senderName);
+                    if (lookupUsers != null && lookupUsers.contains("#container")) {
                         state.validContainer = true;
                     }
                 }

src/main/java/net/coreprotect/config/ConfigHandler.java

@@ -55,7 +55,7 @@ public class ConfigHandler extends Queue {
     public static final String COMMUNITY_EDITION = "Community Edition";
     public static final String JAVA_VERSION = "11.0";
     public static final String MINECRAFT_VERSION = "1.16.5";
-    public static final String PATCH_VERSION = "24.0";
+    public static final String PATCH_VERSION = "24.1";
     public static final String LATEST_VERSION = "26.2";
     public static String path = "plugins/CoreProtect/";
     public static String sqlite = "database.db";

src/main/java/net/coreprotect/consumer/Queue.java

@@ -36,20 +36,50 @@ import net.coreprotect.utility.ErrorReporter;
 
 public class Queue {
 
-    protected static synchronized int modifyForceContainer(String id, ItemStack[] container) {
+    public static synchronized void addForceContainer(String id, ItemStack[] container) {
+        if (container == null) {
+            return;
+        }
+
+        ConfigHandler.forceContainer.computeIfAbsent(id, k -> new ArrayList<>()).add(container);
+    }
+
+    public static synchronized void setForceContainer(String id, ItemStack[] container) {
+        if (container == null) {
+            return;
+        }
+
+        List<ItemStack[]> forceList = new ArrayList<>();
+        forceList.add(container);
+        ConfigHandler.forceContainer.put(id, forceList);
+    }
+
+    public static synchronized ItemStack[] pollForceContainer(String id) {
         List<ItemStack[]> forceList = ConfigHandler.forceContainer.get(id);
         if (forceList == null) {
-            return 0;
+            return null;
         }
 
-        if (container == null) {
-            forceList.remove(0);
-        }
-        else {
-            forceList.add(container);
+        ItemStack[] container = forceList.isEmpty() ? null : forceList.remove(0);
+        if (forceList.isEmpty()) {
+            ConfigHandler.forceContainer.remove(id);
         }
 
-        return forceList.size();
+        return container;
+    }
+
+    public static synchronized ItemStack[] peekForceContainer(String id) {
+        List<ItemStack[]> forceList = ConfigHandler.forceContainer.get(id);
+        return (forceList == null || forceList.isEmpty()) ? null : forceList.get(0);
+    }
+
+    public static synchronized int getForceContainerSize(String id) {
+        List<ItemStack[]> forceList = ConfigHandler.forceContainer.get(id);
+        return forceList == null ? 0 : forceList.size();
+    }
+
+    public static synchronized void removeForceContainer(String id) {
+        ConfigHandler.forceContainer.remove(id);
     }
 
     protected static synchronized int getChestId(String id) {

src/main/java/net/coreprotect/consumer/process/ContainerBreakProcess.java

@@ -9,6 +9,7 @@ import org.bukkit.inventory.ItemStack;
 
 import net.coreprotect.consumer.Consumer;
 import net.coreprotect.database.logger.ContainerBreakLogger;
+import net.coreprotect.utility.HopperTransactionUtils;
 
 class ContainerBreakProcess {
 
@@ -18,7 +19,8 @@ class ContainerBreakProcess {
             Map<Integer, ItemStack[]> containers = Consumer.consumerContainers.get(processId);
             if (containers.get(id) != null) {
                 ItemStack[] container = containers.get(id);
-                ContainerBreakLogger.log(preparedStmt, batchCount, user, location, type, container);
+                String transactionId = HopperTransactionUtils.getTransactionId(location);
+                HopperTransactionUtils.synchronizeTransaction(transactionId, () -> ContainerBreakLogger.log(preparedStmt, batchCount, user, location, type, container));
                 containers.remove(id);
             }
         }

src/main/java/net/coreprotect/consumer/process/ContainerTransactionProcess.java

@@ -2,12 +2,9 @@ package net.coreprotect.consumer.process;
 
 import java.sql.PreparedStatement;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 import org.bukkit.Location;
 import org.bukkit.Material;
@@ -16,81 +13,88 @@ import org.bukkit.inventory.ItemStack;
 
 import net.coreprotect.config.ConfigHandler;
 import net.coreprotect.consumer.Consumer;
+import net.coreprotect.consumer.Queue;
 import net.coreprotect.database.logger.ContainerLogger;
 import net.coreprotect.utility.HopperTransactionUtils;
 
 class ContainerTransactionProcess {
 
     static void process(PreparedStatement preparedStmtContainer, PreparedStatement preparedStmtItems, int batchCount, int processId, int id, Material type, int forceData, String user, Object object) {
-        if (object instanceof Location) {
-            Location location = (Location) object;
-            Map<Integer, Object> inventories = Consumer.consumerInventories.get(processId);
-            Object inventory = inventories.remove(id);
-            if (inventory != null) {
-                String transactingChestId = HopperTransactionUtils.getTransactionId(location);
-                String loggingChestIdSuffix = HopperTransactionUtils.getLoggingIdSuffix(location);
-                String loggingChestId = HopperTransactionUtils.getLoggingId(user, loggingChestIdSuffix);
-                if (ConfigHandler.loggingChest.get(loggingChestId) != null) {
-                    int current_chest = ConfigHandler.loggingChest.get(loggingChestId);
-                    if (ConfigHandler.oldContainer.get(loggingChestId) == null) {
-                        ConfigHandler.removeOldContainerViewer(loggingChestIdSuffix, loggingChestId);
-                        ConfigHandler.loggingChest.remove(loggingChestId);
-                        ConfigHandler.forceContainer.remove(loggingChestId);
-                        HopperTransactionUtils.removeOwner(transactingChestId, loggingChestId);
-                        return;
-                    }
-                    int force_size = 0;
-                    if (ConfigHandler.forceContainer.get(loggingChestId) != null) {
-                        force_size = ConfigHandler.forceContainer.get(loggingChestId).size();
-                    }
-                    if (current_chest == forceData || force_size > 0) { // This prevents client side chest sorting mods from messing things up.
-                        ContainerLogger.log(preparedStmtContainer, preparedStmtItems, batchCount, user, type, inventory, location);
-                        List<ItemStack[]> old = ConfigHandler.oldContainer.get(loggingChestId);
-                        if (old.size() == 0) {
-                            ConfigHandler.oldContainer.remove(loggingChestId);
-                            ConfigHandler.removeOldContainerViewer(loggingChestIdSuffix, loggingChestId);
-                            ConfigHandler.loggingChest.remove(loggingChestId);
-                            ConfigHandler.forceContainer.remove(loggingChestId);
-                            HopperTransactionUtils.removeOwner(transactingChestId, loggingChestId);
-                        }
-                    }
-                    else if (loggingChestId.startsWith("#hopper")) {
-                        if (force_size == 0 && ConfigHandler.oldContainer.getOrDefault(loggingChestId, Collections.synchronizedList(new ArrayList<>())).size() == 1 && HopperTransactionUtils.pendingDeltaCount(transactingChestId) == 0) {
-                            int loopCount = ConfigHandler.loggingChest.getOrDefault(loggingChestId, 0);
-                            int maxInventorySize = (99 * 54);
-                            try {
-                                Inventory checkInventory = (Inventory) inventory;
-                                maxInventorySize = checkInventory.getSize() * checkInventory.getMaxStackSize();
-                            }
-                            catch (Exception e) {
-                                // use default of 5,346
-                            }
+        if (!(object instanceof Location)) {
+            return;
+        }
 
-                            if (loopCount > maxInventorySize) {
-                                ItemStack[] destinationContents = null;
-                                ItemStack movedItem = null;
+        Location location = (Location) object;
+        Map<Integer, Object> inventories = Consumer.consumerInventories.get(processId);
+        Object inventory = inventories.remove(id);
+        if (inventory == null) {
+            return;
+        }
 
-                                String hopperPush = "#hopper-push." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-                                Object[] hopperPushData = ConfigHandler.hopperSuccess.remove(hopperPush);
-                                if (hopperPushData != null) {
-                                    destinationContents = (ItemStack[]) hopperPushData[0];
-                                    movedItem = (ItemStack) hopperPushData[1];
-                                }
+        String transactingChestId = HopperTransactionUtils.getTransactionId(location);
+        String loggingChestIdSuffix = HopperTransactionUtils.getLoggingIdSuffix(location);
+        String loggingChestId = HopperTransactionUtils.getLoggingId(user, loggingChestIdSuffix);
+        HopperTransactionUtils.synchronizeTransaction(transactingChestId,
+                () -> processTransaction(preparedStmtContainer, preparedStmtItems, batchCount, type, forceData, user, inventory, location, transactingChestId, loggingChestIdSuffix, loggingChestId));
+    }
 
-                                if (destinationContents != null) {
-                                    Set<ItemStack> movedItems = new HashSet<>();
-                                    Object[] lastAbort = ConfigHandler.hopperAbort.get(hopperPush);
-                                    if (lastAbort != null && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
-                                        ((Set<?>) lastAbort[0]).forEach(itemStack -> movedItems.add((ItemStack) itemStack));
-                                    }
-                                    movedItems.add(movedItem);
-                                    ConfigHandler.hopperAbort.put(hopperPush, new Object[] { movedItems, destinationContents });
-                                }
-                            }
-                        }
+    private static void processTransaction(PreparedStatement preparedStmtContainer, PreparedStatement preparedStmtItems, int batchCount, Material type, int forceData, String user, Object inventory,
+            Location location, String transactingChestId, String loggingChestIdSuffix, String loggingChestId) {
+        if (ConfigHandler.loggingChest.get(loggingChestId) == null) {
+            return;
+        }
+
+        int current_chest = ConfigHandler.loggingChest.get(loggingChestId);
+        if (ConfigHandler.oldContainer.get(loggingChestId) == null) {
+            clearContainerTransaction(transactingChestId, loggingChestIdSuffix, loggingChestId);
+            return;
+        }
+        int force_size = Queue.getForceContainerSize(loggingChestId);
+        if (current_chest == forceData || force_size > 0) { // This prevents client side chest sorting mods from messing things up.
+            ContainerLogger.log(preparedStmtContainer, preparedStmtItems, batchCount, user, type, inventory, location);
+            List<ItemStack[]> old = ConfigHandler.oldContainer.get(loggingChestId);
+            if (old == null || old.isEmpty()) {
+                clearContainerTransaction(transactingChestId, loggingChestIdSuffix, loggingChestId);
+            }
+        }
+        else if (loggingChestId.startsWith("#hopper")) {
+            if (force_size == 0 && ConfigHandler.oldContainer.getOrDefault(loggingChestId, Collections.synchronizedList(new ArrayList<>())).size() == 1
+                    && HopperTransactionUtils.pendingDeltaCount(transactingChestId) == 0) {
+                int loopCount = ConfigHandler.loggingChest.getOrDefault(loggingChestId, 0);
+                int maxInventorySize = (99 * 54);
+                try {
+                    Inventory checkInventory = (Inventory) inventory;
+                    maxInventorySize = checkInventory.getSize() * checkInventory.getMaxStackSize();
+                }
+                catch (Exception e) {
+                    // use default of 5,346
+                }
+
+                if (loopCount > maxInventorySize) {
+                    ItemStack[] destinationContents = null;
+                    ItemStack movedItem = null;
+
+                    String hopperPush = HopperTransactionUtils.getHopperPushId(location);
+                    Object[] hopperPushData = ConfigHandler.hopperSuccess.remove(hopperPush);
+                    if (hopperPushData != null) {
+                        destinationContents = (ItemStack[]) hopperPushData[0];
+                        movedItem = (ItemStack) hopperPushData[1];
+                    }
+
+                    if (destinationContents != null) {
+                        Object[] lastAbort = ConfigHandler.hopperAbort.get(hopperPush);
+                        ConfigHandler.hopperAbort.put(hopperPush, HopperTransactionUtils.createAbortState(lastAbort, destinationContents, movedItem));
                     }
                 }
             }
         }
     }
+
+    private static void clearContainerTransaction(String transactionId, String locationSuffix, String loggingId) {
+        ConfigHandler.oldContainer.remove(loggingId);
+        ConfigHandler.removeOldContainerViewer(locationSuffix, loggingId);
+        ConfigHandler.loggingChest.remove(loggingId);
+        Queue.removeForceContainer(loggingId);
+        HopperTransactionUtils.removeOwner(transactionId, loggingId);
+    }
 }

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;
 
@@ -139,17 +139,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) {
-                        List<ItemStack[]> forceList = new ArrayList<>();
-                        forceList.add(ItemUtils.getContainerState(contents));
-                        ConfigHandler.forceContainer.put(HopperTransactionUtils.getLoggingId(user, location), forceList);
-                        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/logger/ContainerBreakLogger.java

@@ -6,7 +6,7 @@ import org.bukkit.Location;
 import org.bukkit.Material;
 import org.bukkit.inventory.ItemStack;
 
-import net.coreprotect.config.ConfigHandler;
+import net.coreprotect.consumer.Queue;
 import net.coreprotect.model.item.ItemTransactionActions;
 import net.coreprotect.utility.HopperTransactionUtils;
 import net.coreprotect.utility.ItemUtils;
@@ -25,9 +25,7 @@ public class ContainerBreakLogger {
             String loggingContainerId = HopperTransactionUtils.getLoggingId(player, l);
 
             // If there was a pending chest transaction, it would have already been processed.
-            if (ConfigHandler.forceContainer.get(loggingContainerId) != null) {
-                ConfigHandler.forceContainer.remove(loggingContainerId);
-            }
+            Queue.removeForceContainer(loggingContainerId);
         }
         catch (Exception e) {
             ErrorReporter.report(e);

src/main/java/net/coreprotect/database/logger/ContainerLogger.java

@@ -70,6 +70,9 @@ public class ContainerLogger extends Queue {
             String loggingContainerId = HopperTransactionUtils.getLoggingId(player, location);
             String transactingChestId = HopperTransactionUtils.getTransactionId(location);
             List<ItemStack[]> oldList = ConfigHandler.oldContainer.get(loggingContainerId);
+            if (oldList == null || oldList.isEmpty()) {
+                return;
+            }
             ItemStack[] oi1 = oldList.get(0);
             ItemStack[] oldInventory = ItemUtils.getContainerState(oi1);
             ItemStack[] newInventory = ItemUtils.getContainerState(contents);
@@ -79,7 +82,7 @@ public class ContainerLogger extends Queue {
             boolean duplicateSuppression = Config.getConfig(location.getWorld()).DUPLICATE_SUPPRESSION;
 
             // Check if this is a dispenser with no actual changes
-            if (duplicateSuppression && "#dispenser".equals(player) && ItemUtils.compareContainers(oldInventory, newInventory)) {
+            if (duplicateSuppression && "#dispenser".equals(player) && getForceContainerSize(loggingContainerId) == 0 && ItemUtils.compareContainers(oldInventory, newInventory)) {
                 // No changes detected, mark this dispenser in the dispenserNoChange map
                 // Extract the location key from the loggingContainerId
                 // Format: #dispenser.x.y.z
@@ -102,6 +105,9 @@ public class ContainerLogger extends Queue {
                         ConfigHandler.dispenserNoChange.computeIfAbsent(locationKey, k -> new ConcurrentHashMap<>()).put(eventKey, System.currentTimeMillis());
                     }
                 }
+                oldList.remove(0);
+                ConfigHandler.oldContainer.put(loggingContainerId, oldList);
+                HopperTransactionUtils.consumeSnapshot(transactingChestId, loggingContainerId);
                 return;
             }
 
@@ -124,16 +130,9 @@ public class ContainerLogger extends Queue {
                 }
             }
 
-            List<ItemStack[]> forceList = ConfigHandler.forceContainer.get(loggingContainerId);
-            if (forceList != null) {
-                int forceSize = 0;
-                if (!forceList.isEmpty()) {
-                    newInventory = ItemUtils.getContainerState(forceList.get(0));
-                    forceSize = modifyForceContainer(loggingContainerId, null);
-                }
-                if (forceSize == 0) {
-                    ConfigHandler.forceContainer.remove(loggingContainerId);
-                }
+            ItemStack[] forceState = peekForceContainer(loggingContainerId);
+            if (forceState != null) {
+                newInventory = ItemUtils.getContainerState(forceState);
             }
             else {
                 long snapshotMark = HopperTransactionUtils.peekSnapshotMark(transactingChestId, loggingContainerId);
@@ -141,11 +140,14 @@ public class ContainerLogger extends Queue {
             }
 
             if (duplicateSuppression && shouldSuppressContainerDuplicate(player, location, oldInventory, newInventory)) {
+                if (forceState != null) {
+                    pollForceContainer(loggingContainerId);
+                }
                 oldList.remove(0);
                 ConfigHandler.oldContainer.put(loggingContainerId, oldList);
                 HopperTransactionUtils.consumeSnapshot(transactingChestId, loggingContainerId);
                 if ("#hopper".equals(player)) {
-                    String hopperPush = "#hopper-push." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
+                    String hopperPush = HopperTransactionUtils.getHopperPushId(location);
                     ConfigHandler.hopperSuccess.remove(hopperPush);
                     ConfigHandler.hopperAbort.remove(hopperPush);
                 }
@@ -185,6 +187,9 @@ public class ContainerLogger extends Queue {
                 ItemLogger.logTransaction(preparedStmtItems, batchCount, 0, player, location, newInventory, ItemTransactionActions.ADD_ENDER);
             }
 
+            if (forceState != null) {
+                pollForceContainer(loggingContainerId);
+            }
             oldList.remove(0);
             ConfigHandler.oldContainer.put(loggingContainerId, oldList);
             HopperTransactionUtils.consumeSnapshot(transactingChestId, loggingContainerId);
@@ -242,7 +247,7 @@ public class ContainerLogger extends Queue {
             }
 
             if (success && user.equals("#hopper")) {
-                String hopperPush = "#hopper-push." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
+                String hopperPush = HopperTransactionUtils.getHopperPushId(location);
                 ConfigHandler.hopperSuccess.remove(hopperPush);
                 ConfigHandler.hopperAbort.remove(hopperPush);
             }

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/block/BlockDispenseListener.java

@@ -9,6 +9,7 @@ import org.bukkit.block.Block;
 import org.bukkit.block.BlockState;
 import org.bukkit.block.data.BlockData;
 import org.bukkit.block.data.Lightable;
+import org.bukkit.block.data.Waterlogged;
 import org.bukkit.block.data.type.Dispenser;
 import org.bukkit.event.EventHandler;
 import org.bukkit.event.EventPriority;
@@ -18,6 +19,7 @@ import org.bukkit.inventory.InventoryHolder;
 import org.bukkit.inventory.ItemStack;
 import org.bukkit.util.Vector;
 
+import net.coreprotect.CoreProtect;
 import net.coreprotect.bukkit.BukkitAdapter;
 import net.coreprotect.config.Config;
 import net.coreprotect.consumer.Queue;
@@ -25,6 +27,9 @@ import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.paper.listener.BlockPreDispenseListener;
 import net.coreprotect.thread.CacheHandler;
+import net.coreprotect.thread.Scheduler;
+import net.coreprotect.utility.BlockUtils;
+import net.coreprotect.utility.ErrorReporter;
 
 public final class BlockDispenseListener extends Queue implements Listener {
 
@@ -98,33 +103,61 @@ public final class BlockDispenseListener extends Queue implements Listener {
                     return;
                 }
 
-                if (!type.equals(Material.AIR) || !newBlock.getType().equals(Material.AIR)) {
-                    if (type == Material.FIRE) { // lit a lightable block
-                        type = newBlock.getType();
-                        if (BlockGroup.LIGHTABLES.contains(type)) {
-                            Lightable lightable = (Lightable) newBlockData;
-                            lightable.setLit(true);
+                if (type == Material.FIRE) { // lit a lightable block
+                    type = newBlock.getType();
+                    if (BlockGroup.LIGHTABLES.contains(type)) {
+                        Lightable lightable = (Lightable) newBlockData;
+                        lightable.setLit(true);
 
-                            queueBlockPlace(user, newBlock.getState(), newBlock.getType(), newBlock.getState(), type, -1, 0, newBlockData.getAsString());
-                        }
+                        queueBlockPlace(user, newBlock.getState(), newBlock.getType(), newBlock.getState(), type, -1, 0, newBlockData.getAsString());
                     }
-                    else if (dispenseRelative) {
-                        BlockState blockState = newBlock.getState();
-                        if (config.DUPLICATE_SUPPRESSION && shouldSuppressDispenseLiquidDuplicate(user, newBlock, type)) {
-                            return;
-                        }
-                        if (!type.equals(Material.AIR)) {
-                            queueBlockPlaceValidate(user, blockState, newBlock, blockState, type, 1, 1, null, 0);
-                        }
-                        else {
-                            Queue.queueBlockBreak(user, newBlock.getState(), newBlock.getType(), newBlock.getBlockData().getAsString(), 0);
-                        }
+                }
+                else if (dispenseRelative && !type.equals(Material.AIR)) {
+                    BlockState blockState = newBlock.getState();
+                    if (config.DUPLICATE_SUPPRESSION && shouldSuppressDispenseLiquidDuplicate(user, newBlock, type)) {
+                        return;
                     }
+                    queueBlockPlaceValidate(user, blockState, newBlock, blockState, type, 1, 1, null, 0);
+                }
+                else if (dispenseRelative && material == Material.BUCKET) {
+                    BlockState blockState = newBlock.getState();
+                    if (config.DUPLICATE_SUPPRESSION && shouldSuppressDispenseLiquidDuplicate(user, newBlock, type)) {
+                        return;
+                    }
+                    queueBucketRemovalValidate(user, newBlock, blockState);
                 }
             }
         }
     }
 
+    private static void queueBucketRemovalValidate(String user, Block block, BlockState blockState) {
+        Material originalType = blockState.getType();
+        BlockData originalBlockData = blockState.getBlockData();
+        String blockData = originalBlockData.getAsString();
+        boolean waterlogged = originalBlockData instanceof Waterlogged && ((Waterlogged) originalBlockData).isWaterlogged();
+
+        Scheduler.scheduleSyncDelayedTask(CoreProtect.getInstance(), () -> {
+            try {
+                Material currentType = block.getType();
+                Material removedType = originalType;
+                boolean removed = !BlockUtils.isAir(originalType) && BlockUtils.isAir(currentType);
+
+                if (waterlogged) {
+                    BlockData currentBlockData = block.getBlockData();
+                    removed = BlockUtils.isAir(currentType) || (currentType == originalType && currentBlockData instanceof Waterlogged && !((Waterlogged) currentBlockData).isWaterlogged());
+                    removedType = Material.WATER;
+                }
+
+                if (removed) {
+                    Queue.queueBlockBreak(user, blockState, removedType, blockData, 0);
+                }
+            }
+            catch (Exception e) {
+                ErrorReporter.report(e);
+            }
+        }, block.getLocation(), 0);
+    }
+
     private boolean shouldSuppressDispenseLiquidDuplicate(String user, Block targetBlock, Material newType) {
         Material oldType = targetBlock.getType();
         boolean placeLiquid = ("#water".equals(user) || "#lava".equals(user) || "#dispenser".equals(user)) && (newType == Material.WATER || newType == Material.LAVA);

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,11 +1,10 @@
 package net.coreprotect.listener.player;
 
 import java.util.Arrays;
-import java.util.HashSet;
+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;
@@ -22,145 +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) {
-        String loggingChestId = "#hopper-pull." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
+        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) {
-            Set<ItemStack> movedItems = new HashSet<>();
-            ItemStack[] destinationContents = destinationHolder.getInventory().getContents();
-            if (lastAbort != null && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
-                ((Set<?>) lastAbort[0]).forEach(itemStack -> movedItems.add((ItemStack) itemStack));
-            }
-            movedItems.add(movedItem);
-
-            ConfigHandler.hopperAbort.put(loggingChestId, new Object[] { movedItems, ItemUtils.getContainerState(destinationContents) });
-            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

@@ -1,170 +1,62 @@
 package net.coreprotect.listener.player;
 
 import java.util.Arrays;
-import java.util.HashSet;
 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) {
-            return;
-        }
-
-        String loggingChestId = "#hopper-push." + destinationLocation.getBlockX() + "." + destinationLocation.getBlockY() + "." + destinationLocation.getBlockZ();
-        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) {
-            return;
-        }
-
-        boolean abort = false;
-        boolean addedInventory = ItemUtils.canAddContainer(destinationContainer, movedItem, destinationHolder.getInventory().getMaxStackSize());
-        if (!addedInventory) {
-            abort = true;
-        }
-
-        if (abort) {
-            Set<ItemStack> movedItems = new HashSet<>();
-            ItemStack[] destinationContents = destinationHolder.getInventory().getContents();
-            if (lastAbort != null && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
-                ((Set<?>) lastAbort[0]).forEach(itemStack -> movedItems.add((ItemStack) itemStack));
-            }
-            movedItems.add(movedItem);
-
-            ConfigHandler.hopperAbort.put(loggingChestId, new Object[] { movedItems, ItemUtils.getContainerState(destinationContents) });
-            return;
-        }
-        else {
-            ConfigHandler.hopperSuccess.put(loggingChestId, new Object[] { destinationContainer, movedItem });
-        }
-
-        HopperTransactionUtils.recordItemRemoved(HopperTransactionUtils.getTransactionId(location), movedItem);
-
-        if (Config.getConfig(location.getWorld()).HOPPER_FILTER_META && !movedItem.hasItemMeta()) {
+        if (location == null || location.getWorld() == null || sourceHolder == null || destinationHolder == null || item == null || item.getAmount() <= 0) {
             return;
         }
 
         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;
-            }
+        Location destinationLocation = destinationInventory.getLocation();
+        if (destinationLocation == null || destinationLocation.getWorld() == null) {
+            return;
         }
 
-        InventoryChangeListener.onHopperInventoryInteract(user, destinationInventory, originalDestination, destinationInventory.getLocation(), movedItem);
+        String loggingChestId = HopperTransactionUtils.getHopperPushId(destinationLocation);
+        Object[] lastAbort = ConfigHandler.hopperAbort.get(loggingChestId);
+        ItemStack[] destinationContents = destinationInventory.getContents();
+        if (lastAbort != null && ((Set<?>) lastAbort[0]).contains(item) && Arrays.equals(destinationContents, (ItemStack[]) lastAbort[1])) {
+            return;
+        }
+
+        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;
+        }
+
+        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

@@ -5,7 +5,6 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
 
 import org.bukkit.Location;
 import org.bukkit.Material;
@@ -38,7 +37,6 @@ import net.coreprotect.config.ConfigHandler;
 import net.coreprotect.consumer.Queue;
 import net.coreprotect.model.BlockGroup;
 import net.coreprotect.paper.PaperAdapter;
-import net.coreprotect.thread.Scheduler;
 import net.coreprotect.utility.HopperTransactionUtils;
 import net.coreprotect.utility.ItemUtils;
 import net.coreprotect.utility.Validate;
@@ -48,38 +46,7 @@ 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 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) {
@@ -102,6 +69,44 @@ 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);
+        String transactionId = HopperTransactionUtils.getTransactionId(capturedLocation);
+        HopperPullListener.flushPendingPull(boundaryLocation, null, capturedContents);
+        ContainerTransactionDispatcher.submit(boundaryLocation, () -> HopperTransactionUtils.synchronizeTransaction(transactionId, () -> {
+            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();
@@ -147,10 +152,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;
                 }
             }
         }
@@ -160,6 +173,12 @@ public final class InventoryChangeListener extends Queue implements Listener {
 
     private static boolean queueContainerTransaction(String user, Location playerLocation, Material type, Object inventory, ItemStack[] inventoryData, ItemStack[] forceInventoryData, ItemStack batchItem) {
         String transactingChestId = HopperTransactionUtils.getTransactionId(playerLocation);
+        return HopperTransactionUtils.synchronizeTransaction(transactingChestId,
+                () -> queueContainerTransactionLocked(user, playerLocation, type, inventory, inventoryData, forceInventoryData, batchItem, transactingChestId));
+    }
+
+    private static boolean queueContainerTransactionLocked(String user, Location playerLocation, Material type, Object inventory, ItemStack[] inventoryData, ItemStack[] forceInventoryData, ItemStack batchItem,
+            String transactingChestId) {
         String loggingChestIdSuffix = HopperTransactionUtils.getLoggingIdSuffix(playerLocation);
         String loggingChestId = HopperTransactionUtils.getLoggingId(user, loggingChestIdSuffix);
         Set<String> locationViewers = ConfigHandler.oldContainerViewers.get(loggingChestIdSuffix);
@@ -172,15 +191,16 @@ public final class InventoryChangeListener extends Queue implements Listener {
                 List<ItemStack[]> viewerOldList = ConfigHandler.oldContainer.get(loggingChestIdViewer);
                 if (viewerOldList != null) { // viewer has pending consumer item
                     int sizeOld = viewerOldList.size();
-                    List<ItemStack[]> list = ConfigHandler.forceContainer.computeIfAbsent(loggingChestIdViewer, k -> new ArrayList<>());
+                    int forceSize = getForceContainerSize(loggingChestIdViewer);
 
-                    if (list.size() < sizeOld) {
+                    if (forceSize < sizeOld) {
                         ItemStack[] containerState = ItemUtils.getContainerState(inventoryData);
 
-                        long snapshotMark = HopperTransactionUtils.getSnapshotMark(transactingChestId, loggingChestIdViewer, list.size());
+                        long snapshotMark = HopperTransactionUtils.getSnapshotMark(transactingChestId, loggingChestIdViewer, forceSize);
                         containerState = HopperTransactionUtils.applyPendingChanges(containerState, transactingChestId, snapshotMark);
 
-                        modifyForceContainer(loggingChestIdViewer, containerState);
+                        ItemStack[] previousState = viewerOldList.get(forceSize);
+                        addForceContainer(loggingChestIdViewer, ItemUtils.getSharedContainerState(containerState, previousState));
                     }
                 }
             }
@@ -189,19 +209,27 @@ public final class InventoryChangeListener extends Queue implements Listener {
         if (forceInventoryData == null && batchItem != null && HopperTransactionUtils.shouldForceBatchBoundary(transactingChestId, loggingChestId, batchItem)) {
             forceInventoryData = inventoryData;
         }
+        List<ItemStack[]> existingOldList = ConfigHandler.oldContainer.get(loggingChestId);
+        ItemStack[] previousState = existingOldList == null || existingOldList.isEmpty() ? null : existingOldList.get(existingOldList.size() - 1);
+        ItemStack[] oldState = null;
+        ItemStack[] forceState = null;
         if (forceInventoryData != null) {
-            ConfigHandler.forceContainer.computeIfAbsent(loggingChestId, k -> new ArrayList<>()).add(ItemUtils.getContainerState(forceInventoryData));
+            oldState = ItemUtils.getSharedContainerState(inventoryData, previousState);
+            forceState = forceInventoryData == inventoryData ? oldState : ItemUtils.getSharedContainerState(forceInventoryData, oldState);
+            addForceContainer(loggingChestId, forceState);
         }
 
         int chestId = getChestId(loggingChestId);
         if (chestId > 0) {
-            List<ItemStack[]> forceList = ConfigHandler.forceContainer.get(loggingChestId);
-            if (forceList != null) {
-                int forceSize = forceList.size();
+            int forceSize = getForceContainerSize(loggingChestId);
+            if (forceSize > 0) {
                 List<ItemStack[]> list = ConfigHandler.oldContainer.get(loggingChestId);
 
                 if (list != null && list.size() <= forceSize) {
-                    list.add(ItemUtils.getContainerState(inventoryData));
+                    if (oldState == null) {
+                        oldState = ItemUtils.getSharedContainerState(inventoryData, previousState);
+                    }
+                    list.add(oldState);
                     ConfigHandler.oldContainer.put(loggingChestId, list);
                     HopperTransactionUtils.registerSnapshot(transactingChestId, loggingChestId, false);
                 }
@@ -209,7 +237,10 @@ public final class InventoryChangeListener extends Queue implements Listener {
         }
         else {
             List<ItemStack[]> list = new ArrayList<>();
-            list.add(ItemUtils.getContainerState(inventoryData));
+            if (oldState == null) {
+                oldState = ItemUtils.getSharedContainerState(inventoryData, previousState);
+            }
+            list.add(oldState);
             ConfigHandler.oldContainer.put(loggingChestId, list);
             ConfigHandler.addOldContainerViewer(loggingChestIdSuffix, loggingChestId);
             HopperTransactionUtils.registerSnapshot(transactingChestId, loggingChestId, true);
@@ -261,8 +292,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);
@@ -271,13 +303,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);
@@ -573,7 +603,25 @@ public final class InventoryChangeListener extends Queue implements Listener {
             user = "#hopper";
         }
 
-        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) {

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

@@ -122,28 +122,30 @@ public final class PlayerInteractEntityListener extends Queue implements Listene
         String transactingChestId = HopperTransactionUtils.getTransactionId(location);
         String loggingChestIdSuffix = HopperTransactionUtils.getLoggingIdSuffix(location);
         String loggingChestId = HopperTransactionUtils.getLoggingId(user, loggingChestIdSuffix);
-        int chestId = Queue.getChestId(loggingChestId);
-        if (chestId > 0) {
-            if (ConfigHandler.forceContainer.get(loggingChestId) != null) {
-                int forceSize = ConfigHandler.forceContainer.get(loggingChestId).size();
-                List<ItemStack[]> list = ConfigHandler.oldContainer.get(loggingChestId);
+        HopperTransactionUtils.synchronizeTransaction(transactingChestId, () -> {
+            int chestId = Queue.getChestId(loggingChestId);
+            if (chestId > 0) {
+                int forceSize = Queue.getForceContainerSize(loggingChestId);
+                if (forceSize > 0) {
+                    List<ItemStack[]> list = ConfigHandler.oldContainer.get(loggingChestId);
 
-                if (list.size() <= forceSize) {
-                    list.add(ItemUtils.getContainerState(contents));
-                    ConfigHandler.oldContainer.put(loggingChestId, list);
-                    HopperTransactionUtils.registerSnapshot(transactingChestId, loggingChestId, false);
+                    if (list.size() <= forceSize) {
+                        list.add(ItemUtils.getContainerState(contents));
+                        ConfigHandler.oldContainer.put(loggingChestId, list);
+                        HopperTransactionUtils.registerSnapshot(transactingChestId, loggingChestId, false);
+                    }
                 }
             }
-        }
-        else {
-            List<ItemStack[]> list = new ArrayList<>();
-            list.add(ItemUtils.getContainerState(contents));
-            ConfigHandler.oldContainer.put(loggingChestId, list);
-            ConfigHandler.addOldContainerViewer(loggingChestIdSuffix, loggingChestId);
-            HopperTransactionUtils.registerSnapshot(transactingChestId, loggingChestId, true);
-        }
+            else {
+                List<ItemStack[]> list = new ArrayList<>();
+                list.add(ItemUtils.getContainerState(contents));
+                ConfigHandler.oldContainer.put(loggingChestId, list);
+                ConfigHandler.addOldContainerViewer(loggingChestIdSuffix, loggingChestId);
+                HopperTransactionUtils.registerSnapshot(transactingChestId, loggingChestId, true);
+            }
 
-        Queue.queueContainerTransaction(user, location, type, container, chestId);
+            Queue.queueContainerTransaction(user, location, type, container, chestId);
+        });
 
         if (logDrop) {
             ItemStack dropItem = contents[0];

src/main/java/net/coreprotect/paper/listener/CopperGolemChestListener.java

@@ -1,7 +1,5 @@
 package net.coreprotect.paper.listener;
 
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -29,6 +27,7 @@ import net.coreprotect.CoreProtect;
 import net.coreprotect.bukkit.BukkitAdapter;
 import net.coreprotect.config.Config;
 import net.coreprotect.config.ConfigHandler;
+import net.coreprotect.consumer.Queue;
 import net.coreprotect.listener.player.InventoryChangeListener;
 import net.coreprotect.thread.Scheduler;
 import net.coreprotect.utility.HopperTransactionUtils;
@@ -524,24 +523,20 @@ public final class CopperGolemChestListener implements Listener {
         }
 
         String loggingContainerId = HopperTransactionUtils.getLoggingId(USERNAME, location);
+        String transactionId = HopperTransactionUtils.getTransactionId(location);
+        HopperTransactionUtils.synchronizeTransaction(transactionId, () -> {
+            List<ItemStack[]> oldList = ConfigHandler.oldContainer.get(loggingContainerId);
+            boolean hasPendingBaseline = oldList != null && !oldList.isEmpty();
 
-        List<ItemStack[]> forceList = ConfigHandler.forceContainer.get(loggingContainerId);
-        List<ItemStack[]> oldList = ConfigHandler.oldContainer.get(loggingContainerId);
+            ItemStack[] oldestSnapshot = Queue.peekForceContainer(loggingContainerId);
+            boolean hasStaleForceSnapshots = oldestSnapshot != null && oldestSnapshot.length != snapshot.length;
 
-        boolean hasPendingBaseline = oldList != null && !oldList.isEmpty();
-        boolean hasStaleForceSnapshots = forceList != null && !forceList.isEmpty() && (forceList.get(0) == null || forceList.get(0).length != snapshot.length);
+            if (!hasPendingBaseline || hasStaleForceSnapshots) {
+                Queue.removeForceContainer(loggingContainerId);
+            }
 
-        if (!hasPendingBaseline || hasStaleForceSnapshots) {
-            ConfigHandler.forceContainer.remove(loggingContainerId);
-            forceList = null;
-        }
-
-        if (forceList == null) {
-            forceList = Collections.synchronizedList(new ArrayList<>());
-            ConfigHandler.forceContainer.put(loggingContainerId, forceList);
-        }
-
-        forceList.add(snapshot);
+            Queue.addForceContainer(loggingContainerId, snapshot);
+        });
     }
 
     private void cleanupOpenInteractions(long nowMillis) {

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

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

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

@@ -2,22 +2,32 @@ package net.coreprotect.utility;
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Deque;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
 
 import org.bukkit.Location;
 import org.bukkit.inventory.ItemStack;
 
 public final class HopperTransactionUtils {
 
-    private static final int MAX_PENDING_DELTAS = 1000000;
     private static final long APPLY_ALL_MARK = -1L;
+    private static final int TRANSACTION_LOCK_COUNT = 256;
 
     private static final ConcurrentHashMap<String, PendingTransaction> pendingTransactions = new ConcurrentHashMap<>();
+    private static final Object[] transactionLocks = new Object[TRANSACTION_LOCK_COUNT];
+
+    static {
+        Arrays.setAll(transactionLocks, index -> new Object());
+    }
 
     private HopperTransactionUtils() {
         throw new IllegalStateException("Utility class");
@@ -39,6 +49,28 @@ public final class HopperTransactionUtils {
         return "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ() + "." + location.getWorld().getUID().toString();
     }
 
+    public static String getHopperPushId(Location location) {
+        return "#hopper-push" + getLoggingIdSuffix(location);
+    }
+
+    public static String getHopperPullId(Location location) {
+        return "#hopper-pull" + getLoggingIdSuffix(location);
+    }
+
+    public static void synchronizeTransaction(String transactionId, Runnable operation) {
+        Objects.requireNonNull(operation);
+        synchronized (getTransactionLock(transactionId)) {
+            operation.run();
+        }
+    }
+
+    public static <T> T synchronizeTransaction(String transactionId, Supplier<T> operation) {
+        Objects.requireNonNull(operation);
+        synchronized (getTransactionLock(transactionId)) {
+            return operation.get();
+        }
+    }
+
     public static boolean hasTransaction(String transactionId) {
         return pendingTransactions.containsKey(transactionId);
     }
@@ -132,7 +164,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) {
@@ -152,6 +184,22 @@ public final class HopperTransactionUtils {
         }
     }
 
+    public static Object[] createAbortState(Object[] previous, ItemStack[] destinationContents, ItemStack movedItem) {
+        Set<ItemStack> movedItems = new HashSet<>();
+        if (previous != null && previous.length >= 2 && previous[0] instanceof Set && previous[1] instanceof ItemStack[] && Arrays.equals(destinationContents, (ItemStack[]) previous[1])) {
+            for (Object item : (Set<?>) previous[0]) {
+                if (item instanceof ItemStack) {
+                    movedItems.add((ItemStack) item);
+                }
+            }
+        }
+        if (movedItem != null) {
+            movedItems.add(movedItem.clone());
+        }
+
+        return new Object[] { movedItems, ItemUtils.getContainerState(destinationContents) };
+    }
+
     public static ItemStack[] applyPendingChanges(ItemStack[] containerState, String transactionId, long sinceMark) {
         if (containerState == null) {
             return null;
@@ -198,7 +246,7 @@ public final class HopperTransactionUtils {
                 count++;
             }
             else {
-                removeFirstSimilarItem(result, item);
+                removeSimilarItems(result, item);
             }
         }
 
@@ -220,18 +268,25 @@ public final class HopperTransactionUtils {
 
             Delta tail = transaction.deltas.peekLast();
             if (tail != null && tail.addBack == addBack && tail.seq > transaction.lastMarkSeq && tail.item.isSimilar(item)) {
-                tail.amount = (int) Math.min((long) tail.amount + amount, Integer.MAX_VALUE);
-                return;
-            }
+                long combinedAmount = (long) tail.amount + amount;
+                if (combinedAmount <= Integer.MAX_VALUE) {
+                    tail.amount = (int) combinedAmount;
+                    return;
+                }
 
-            if (transaction.deltas.size() >= MAX_PENDING_DELTAS) {
-                return;
+                tail.amount = Integer.MAX_VALUE;
+                amount = (int) (combinedAmount - Integer.MAX_VALUE);
             }
 
             transaction.deltas.addLast(new Delta(item.clone(), addBack, amount, transaction.nextSeq));
         }
     }
 
+    private static Object getTransactionLock(String transactionId) {
+        Objects.requireNonNull(transactionId);
+        return transactionLocks[Math.floorMod(transactionId.hashCode(), TRANSACTION_LOCK_COUNT)];
+    }
+
     private static void pruneDeltas(PendingTransaction transaction) {
         long minMark = Long.MAX_VALUE;
         boolean hasMark = false;
@@ -255,14 +310,22 @@ public final class HopperTransactionUtils {
         }
     }
 
-    private static void removeFirstSimilarItem(ItemStack[] items, ItemStack item) {
-        for (ItemStack check : items) {
-            if (check == null || !check.isSimilar(item)) {
+    private static void removeSimilarItems(ItemStack[] items, ItemStack item) {
+        int remaining = item.getAmount();
+        for (int i = 0; i < items.length; i++) {
+            if (remaining <= 0) {
+                return;
+            }
+            ItemStack check = items[i];
+            if (check == null || check.getAmount() <= 0 || !check.isSimilar(item)) {
                 continue;
             }
 
-            check.setAmount(check.getAmount() - item.getAmount());
-            return;
+            int removed = Math.min(remaining, check.getAmount());
+            ItemStack updated = check.clone();
+            updated.setAmount(check.getAmount() - removed);
+            items[i] = updated;
+            remaining -= removed;
         }
     }
 

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

@@ -120,6 +120,31 @@ public class ItemUtils {
         return result;
     }
 
+    public static ItemStack[] getSharedContainerState(ItemStack[] array, ItemStack[] previousState) {
+        if (array == null) {
+            return null;
+        }
+        if (previousState == null || previousState.length != array.length) {
+            return getContainerState(array);
+        }
+
+        ItemStack[] result = new ItemStack[array.length];
+        boolean unchanged = true;
+        for (int i = 0; i < array.length; i++) {
+            ItemStack itemStack = array[i];
+            ItemStack previousItem = previousState[i];
+            if (itemStack == null ? previousItem == null : itemStack.equals(previousItem)) {
+                result[i] = previousItem;
+            }
+            else {
+                result[i] = itemStack == null ? null : itemStack.clone();
+                unchanged = false;
+            }
+        }
+
+        return unchanged ? previousState : result;
+    }
+
     public static ItemStack[] sortContainerState(ItemStack[] array) {
         if (array == null) {
             return null;

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) {