Fix races on the pending item transaction lists (#1013)

* Fix races on the pending item transaction lists

Item listeners appended to the shared per-player lists with get, add and put while the consumer read and then cleared them, so items could be logged twice or not at all, and the anvil path replaced pending items outright. Listeners now append inside ConcurrentHashMap.compute and the consumer takes each list with remove. The prepared copy is kept for every database type so a retried relational batch still writes its rows.

* Close discard and clone-failure gaps in pending item transactions

Listeners now append their item and register its generation in one
Queue.addPendingItems call that holds the Queue lock, the same lock
ItemTransactionProcess.discard takes, so discard can no longer clear an
item whose transaction has not been queued yet. ItemLogger.prepare now
only detaches the pending lists; cloning and merging move into
PreparedTransaction.log, which runs after the prepared transaction is
stored for the batch, so a cloning failure leaves the detached items
available to the retried transaction.

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

@@ -1,9 +1,11 @@
 package net.coreprotect.consumer;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.bukkit.Bukkit;
 import org.bukkit.Location;
@@ -103,10 +105,17 @@ public class Queue {
         return chestId;
     }
 
-    protected static synchronized int getItemId(String id) {
-        int chestId = ConfigHandler.loggingItem.getOrDefault(id, -1) + 1;
-        ConfigHandler.loggingItem.put(id, chestId);
-        return chestId;
+    // The Queue lock keeps ItemTransactionProcess.discard from clearing an item before its generation is registered,
+    // and compute keeps the append atomic with the consumer's take in ItemLogger
+    protected static synchronized int addPendingItems(ConcurrentHashMap<String, List<ItemStack>> pendingItems, String id, ItemStack... items) {
+        pendingItems.compute(id, (key, list) -> {
+            List<ItemStack> result = list == null ? new ArrayList<>(items.length) : list;
+            Collections.addAll(result, items);
+            return result;
+        });
+        int itemId = ConfigHandler.loggingItem.getOrDefault(id, -1) + 1;
+        ConfigHandler.loggingItem.put(id, itemId);
+        return itemId;
     }
 
     private static boolean queueStandardData(Object[] data, String[] user, Object object, boolean first, long reservation) {

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

@@ -26,23 +26,17 @@ class ItemTransactionProcess extends Queue {
             if (ConfigHandler.loggingItem.get(loggingItemId) != null) {
                 int current_chest = ConfigHandler.loggingItem.get(loggingItemId);
                 if (ConfigHandler.itemsPickup.get(loggingItemId) == null && ConfigHandler.itemsDrop.get(loggingItemId) == null && ConfigHandler.itemsThrown.get(loggingItemId) == null && ConfigHandler.itemsShot.get(loggingItemId) == null && ConfigHandler.itemsBreak.get(loggingItemId) == null && ConfigHandler.itemsDestroy.get(loggingItemId) == null && ConfigHandler.itemsCreate.get(loggingItemId) == null && ConfigHandler.itemsSell.get(loggingItemId) == null && ConfigHandler.itemsBuy.get(loggingItemId) == null) {
+                    ConfigHandler.loggingItem.remove(loggingItemId, forceData);
                     return;
                 }
                 if (current_chest == forceData) {
                     int currentTime = (int) (System.currentTimeMillis() / 1000L);
                     if (currentTime > time) {
-                        ItemLogger.PreparedTransaction prepared = null;
-                        if (ConfigHandler.databaseType.isColumnar()) {
-                            prepared = ItemLogger.prepare(location, offset, user);
-                            Consumer.consumerObjects.get(processId).put(id, prepared);
-                        }
-                        else {
-                            ItemLogger.log(preparedStmt, batchCount, location, offset, user);
-                        }
-                        clearItemTransaction(loggingItemId);
-                        if (prepared != null) {
-                            prepared.log(preparedStmt, batchCount, user);
-                        }
+                        // prepare takes the pending lists, so a retried batch has to log from the stored copy
+                        ItemLogger.PreparedTransaction prepared = ItemLogger.prepare(location, offset, user);
+                        Consumer.consumerObjects.get(processId).put(id, prepared);
+                        ConfigHandler.loggingItem.remove(loggingItemId, forceData);
+                        prepared.log(preparedStmt, batchCount, user);
                     }
                     else {
                         Queue.queueItemTransaction(user, location, time, offset, forceData);

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

@@ -59,20 +59,19 @@ public class ItemLogger {
     public static PreparedTransaction prepare(Location location, int offset, String user) {
         String key = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
         ItemStack[][] items = {
-                snapshot(ConfigHandler.itemsPickup, key), snapshot(ConfigHandler.itemsDrop, key),
-                snapshot(ConfigHandler.itemsThrown, key), snapshot(ConfigHandler.itemsShot, key),
-                snapshot(ConfigHandler.itemsBreak, key), snapshot(ConfigHandler.itemsDestroy, key),
-                snapshot(ConfigHandler.itemsCreate, key), snapshot(ConfigHandler.itemsSell, key), snapshot(ConfigHandler.itemsBuy, key)
+                take(ConfigHandler.itemsPickup, key), take(ConfigHandler.itemsDrop, key),
+                take(ConfigHandler.itemsThrown, key), take(ConfigHandler.itemsShot, key),
+                take(ConfigHandler.itemsBreak, key), take(ConfigHandler.itemsDestroy, key),
+                take(ConfigHandler.itemsCreate, key), take(ConfigHandler.itemsSell, key), take(ConfigHandler.itemsBuy, key)
         };
-        for (ItemStack[] group : items) {
-            ItemUtils.mergeItems(null, group);
-        }
         return new PreparedTransaction(location, (int) (System.currentTimeMillis() / 1000L) - offset, items);
     }
 
-    private static ItemStack[] snapshot(Map<String, List<ItemStack>> source, String key) {
-        List<ItemStack> values = source.get(key);
-        return values == null ? new ItemStack[0] : ItemUtils.getContainerState(values.toArray(new ItemStack[0]));
+    // Removing the whole list means an item appended after this point starts a new list for the next transaction.
+    // Cloning waits for PreparedTransaction.log, so the detached items are stored for a retry before anything can throw
+    private static ItemStack[] take(Map<String, List<ItemStack>> source, String key) {
+        List<ItemStack> values = source.remove(key);
+        return values == null ? new ItemStack[0] : values.toArray(new ItemStack[0]);
     }
 
     public static final class PreparedTransaction {
@@ -92,7 +91,10 @@ public class ItemLogger {
                 return;
             }
             for (int index = 0; index < items.length; index++) {
-                logTransaction(batch, batchCount, 0, user, location.clone(), ItemUtils.getContainerState(items[index]), ACTIONS[index], time);
+                // Clone before merging so a retry logs from the untouched detached items
+                ItemStack[] group = ItemUtils.getContainerState(items[index]);
+                ItemUtils.mergeItems(null, group);
+                logTransaction(batch, batchCount, 0, user, location.clone(), group, ACTIONS[index], time);
             }
         }
     }

src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java

@@ -1,7 +1,5 @@
 package net.coreprotect.listener.entity;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Locale;
 
 import org.bukkit.Location;
@@ -26,11 +24,7 @@ public final class EntityPickupItemListener extends Queue implements Listener {
         }
 
         String loggingItemId = player.getName().toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-        int itemId = getItemId(loggingItemId);
-
-        List<ItemStack> list = ConfigHandler.itemsPickup.getOrDefault(loggingItemId, new ArrayList<>());
-        list.add(itemStack.clone());
-        ConfigHandler.itemsPickup.put(loggingItemId, list);
+        int itemId = addPendingItems(ConfigHandler.itemsPickup, loggingItemId, itemStack.clone());
 
         int time = (int) (System.currentTimeMillis() / 1000L) + 1;
         Queue.queueItemTransaction(player.getName(), location.clone(), time, 0, itemId);

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

@@ -39,27 +39,18 @@ public final class CraftItemListener extends Queue implements Listener {
         }
 
         String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-        int itemId = getItemId(loggingItemId);
-
+        int itemId;
         if (action == ItemLogger.ITEM_BUY) {
-            List<ItemStack> list = ConfigHandler.itemsBuy.getOrDefault(loggingItemId, new ArrayList<>());
-            list.add(itemStack);
-            ConfigHandler.itemsBuy.put(loggingItemId, list);
+            itemId = addPendingItems(ConfigHandler.itemsBuy, loggingItemId, itemStack);
         }
         else if (action == ItemLogger.ITEM_SELL) {
-            List<ItemStack> list = ConfigHandler.itemsSell.getOrDefault(loggingItemId, new ArrayList<>());
-            list.add(itemStack);
-            ConfigHandler.itemsSell.put(loggingItemId, list);
+            itemId = addPendingItems(ConfigHandler.itemsSell, loggingItemId, itemStack);
         }
         else if (action == ItemLogger.ITEM_CREATE) {
-            List<ItemStack> list = ConfigHandler.itemsCreate.getOrDefault(loggingItemId, new ArrayList<>());
-            list.add(itemStack);
-            ConfigHandler.itemsCreate.put(loggingItemId, list);
+            itemId = addPendingItems(ConfigHandler.itemsCreate, loggingItemId, itemStack);
         }
         else {
-            List<ItemStack> list = ConfigHandler.itemsDestroy.getOrDefault(loggingItemId, new ArrayList<>());
-            list.add(itemStack);
-            ConfigHandler.itemsDestroy.put(loggingItemId, list);
+            itemId = addPendingItems(ConfigHandler.itemsDestroy, loggingItemId, itemStack);
         }
 
         int time = (int) (System.currentTimeMillis() / 1000L) + 1;

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

@@ -547,18 +547,12 @@ public final class InventoryChangeListener extends Queue implements Listener {
         // Process the enchantment operation
         Location location = player.getLocation();
         String loggingItemId = player.getName().toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-        int itemId = getItemId(loggingItemId);
 
         // Log the input items as removed
-        List<ItemStack> removedItems = new ArrayList<>();
-        removedItems.add(firstItem.clone());
-        removedItems.add(secondItem.clone());
-        ConfigHandler.itemsDestroy.put(loggingItemId, removedItems);
+        addPendingItems(ConfigHandler.itemsDestroy, loggingItemId, firstItem.clone(), secondItem.clone());
 
         // Log the output item as created
-        List<ItemStack> createdItems = new ArrayList<>();
-        createdItems.add(resultItem.clone());
-        ConfigHandler.itemsCreate.put(loggingItemId, createdItems);
+        int itemId = addPendingItems(ConfigHandler.itemsCreate, loggingItemId, resultItem.clone());
 
         int time = (int) (System.currentTimeMillis() / 1000L) + 1;
         Queue.queueItemTransaction(player.getName(), location.clone(), time, 0, itemId);

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

@@ -1,7 +1,5 @@
 package net.coreprotect.listener.player;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.UUID;
@@ -41,11 +39,7 @@ public final class PlayerDropItemListener extends Queue implements Listener {
         }
 
         String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-        int itemId = getItemId(loggingItemId);
-
-        List<ItemStack> list = ConfigHandler.itemsDrop.getOrDefault(loggingItemId, new ArrayList<>());
-        list.add(itemStack.clone());
-        ConfigHandler.itemsDrop.put(loggingItemId, list);
+        int itemId = addPendingItems(ConfigHandler.itemsDrop, loggingItemId, itemStack.clone());
 
         int time = (int) (System.currentTimeMillis() / 1000L) + 1;
         Queue.queueItemTransaction(user, location.clone(), time, 0, itemId);

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

@@ -1,7 +1,5 @@
 package net.coreprotect.listener.player;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Locale;
 
 import org.bukkit.Location;
@@ -23,11 +21,7 @@ public final class PlayerItemBreakListener extends Queue implements Listener {
         }
 
         String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-        int itemId = getItemId(loggingItemId);
-
-        List<ItemStack> list = ConfigHandler.itemsBreak.getOrDefault(loggingItemId, new ArrayList<>());
-        list.add(itemStack.clone());
-        ConfigHandler.itemsBreak.put(loggingItemId, list);
+        int itemId = addPendingItems(ConfigHandler.itemsBreak, loggingItemId, itemStack.clone());
 
         int time = (int) (System.currentTimeMillis() / 1000L) + 1;
         Queue.queueItemTransaction(user, location.clone(), time, 0, itemId);

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

@@ -1,10 +1,8 @@
 package net.coreprotect.listener.player;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -36,22 +34,18 @@ public final class ProjectileLaunchListener extends Queue implements Listener {
         }
 
         String loggingItemId = user.toLowerCase(Locale.ROOT) + "." + location.getBlockX() + "." + location.getBlockY() + "." + location.getBlockZ();
-        int itemId = getItemId(loggingItemId);
 
         itemStack = itemStack.clone();
         if (amount > 0) {
             itemStack.setAmount(amount);
         }
 
+        int itemId;
         if (action == ItemLogger.ITEM_SHOOT) {
-            List<ItemStack> list = ConfigHandler.itemsShot.getOrDefault(loggingItemId, new ArrayList<>());
-            list.add(itemStack);
-            ConfigHandler.itemsShot.put(loggingItemId, list);
+            itemId = addPendingItems(ConfigHandler.itemsShot, loggingItemId, itemStack);
         }
         else {
-            List<ItemStack> list = ConfigHandler.itemsThrown.getOrDefault(loggingItemId, new ArrayList<>());
-            list.add(itemStack);
-            ConfigHandler.itemsThrown.put(loggingItemId, list);
+            itemId = addPendingItems(ConfigHandler.itemsThrown, loggingItemId, itemStack);
         }
 
         int time = (int) (System.currentTimeMillis() / 1000L) + delay;