Fixed race when processing container transactions

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

@@ -20,57 +20,70 @@ 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) {
-                        clearContainerTransaction(transactingChestId, loggingChestIdSuffix, loggingChestId);
-                        return;
+        if (!(object instanceof Location)) {
+            return;
+        }
+
+        Location location = (Location) object;
+        Map<Integer, Object> inventories = Consumer.consumerInventories.get(processId);
+        Object inventory = inventories.remove(id);
+        if (inventory == null) {
+            return;
+        }
+
+        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));
+    }
+
+    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];
                     }
-                    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));
-                                }
-                            }
-                        }
+                    if (destinationContents != null) {
+                        Object[] lastAbort = ConfigHandler.hopperAbort.get(hopperPush);
+                        ConfigHandler.hopperAbort.put(hopperPush, HopperTransactionUtils.createAbortState(lastAbort, destinationContents, movedItem));
                     }
                 }
             }

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

@@ -77,11 +77,12 @@ public final class InventoryChangeListener extends Queue implements Listener {
         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, () -> {
+        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) {
@@ -172,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);

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) {
-            int forceSize = Queue.getForceContainerSize(loggingChestId);
-            if (forceSize > 0) {
-                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

@@ -523,18 +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[]> oldList = ConfigHandler.oldContainer.get(loggingContainerId);
-        boolean hasPendingBaseline = oldList != null && !oldList.isEmpty();
+            ItemStack[] oldestSnapshot = Queue.peekForceContainer(loggingContainerId);
+            boolean hasStaleForceSnapshots = oldestSnapshot != null && oldestSnapshot.length != snapshot.length;
 
-        ItemStack[] oldestSnapshot = Queue.peekForceContainer(loggingContainerId);
-        boolean hasStaleForceSnapshots = oldestSnapshot != null && oldestSnapshot.length != snapshot.length;
+            if (!hasPendingBaseline || hasStaleForceSnapshots) {
+                Queue.removeForceContainer(loggingContainerId);
+            }
 
-        if (!hasPendingBaseline || hasStaleForceSnapshots) {
-            Queue.removeForceContainer(loggingContainerId);
-        }
-
-        Queue.addForceContainer(loggingContainerId, snapshot);
+            Queue.addForceContainer(loggingContainerId, snapshot);
+        });
     }
 
     private void cleanupOpenInteractions(long nowMillis) {

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

@@ -9,8 +9,10 @@ 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;
@@ -18,8 +20,14 @@ import org.bukkit.inventory.ItemStack;
 public final class HopperTransactionUtils {
 
     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");
@@ -49,6 +57,20 @@ public final class HopperTransactionUtils {
         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);
     }
@@ -260,6 +282,11 @@ public final class HopperTransactionUtils {
         }
     }
 
+    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;