From 2863de7f3047e7b1bfd880d81647db51ae495c38 Mon Sep 17 00:00:00 2001 From: Sage Kummer Date: Fri, 2 Oct 2026 15:32:56 -0400 Subject: [PATCH] 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. --- .../java/net/coreprotect/consumer/Queue.java | 17 +++++++++---- .../process/ItemTransactionProcess.java | 18 +++++--------- .../database/logger/ItemLogger.java | 24 ++++++++++--------- .../entity/EntityPickupItemListener.java | 8 +------ .../listener/player/CraftItemListener.java | 19 ++++----------- .../player/InventoryChangeListener.java | 10 ++------ .../player/PlayerDropItemListener.java | 8 +------ .../player/PlayerItemBreakListener.java | 8 +------ .../player/ProjectileLaunchListener.java | 12 +++------- 9 files changed, 45 insertions(+), 79 deletions(-) diff --git a/src/main/java/net/coreprotect/consumer/Queue.java b/src/main/java/net/coreprotect/consumer/Queue.java index 00c95740..a372a097 100755 --- a/src/main/java/net/coreprotect/consumer/Queue.java +++ b/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> pendingItems, String id, ItemStack... items) { + pendingItems.compute(id, (key, list) -> { + List 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) { diff --git a/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java b/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java index b11a3938..b8f5a882 100644 --- a/src/main/java/net/coreprotect/consumer/process/ItemTransactionProcess.java +++ b/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); diff --git a/src/main/java/net/coreprotect/database/logger/ItemLogger.java b/src/main/java/net/coreprotect/database/logger/ItemLogger.java index e5ce4c66..c4fda982 100644 --- a/src/main/java/net/coreprotect/database/logger/ItemLogger.java +++ b/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> source, String key) { - List 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> source, String key) { + List 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); } } } diff --git a/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java b/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java index 9f679690..007298fc 100644 --- a/src/main/java/net/coreprotect/listener/entity/EntityPickupItemListener.java +++ b/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 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); diff --git a/src/main/java/net/coreprotect/listener/player/CraftItemListener.java b/src/main/java/net/coreprotect/listener/player/CraftItemListener.java index b93ccf14..e65c80c2 100644 --- a/src/main/java/net/coreprotect/listener/player/CraftItemListener.java +++ b/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 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 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 list = ConfigHandler.itemsCreate.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsCreate.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsCreate, loggingItemId, itemStack); } else { - List 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; diff --git a/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java b/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java index 2649f2c2..04185816 100644 --- a/src/main/java/net/coreprotect/listener/player/InventoryChangeListener.java +++ b/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 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 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); diff --git a/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java b/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java index 8247934d..7b3c4955 100644 --- a/src/main/java/net/coreprotect/listener/player/PlayerDropItemListener.java +++ b/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 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); diff --git a/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java b/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java index 45648561..04ed3144 100644 --- a/src/main/java/net/coreprotect/listener/player/PlayerItemBreakListener.java +++ b/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 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); diff --git a/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java b/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java index 101e6977..5990a9e0 100644 --- a/src/main/java/net/coreprotect/listener/player/ProjectileLaunchListener.java +++ b/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 list = ConfigHandler.itemsShot.getOrDefault(loggingItemId, new ArrayList<>()); - list.add(itemStack); - ConfigHandler.itemsShot.put(loggingItemId, list); + itemId = addPendingItems(ConfigHandler.itemsShot, loggingItemId, itemStack); } else { - List 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;