From ebe4225c1eeea5f31b6a4ebcaeabf5a444619432 Mon Sep 17 00:00:00 2001 From: Chroma <68839442+DarkChromaMC@users.noreply.github.com> Date: Mon, 19 Oct 2026 22:42:07 +1300 Subject: [PATCH] Recover DuckDB entity UUID conflicts without dropping or stalling A DuckDB unique-index conflict on co_entity_spawn.uuid can happen when the identity row was committed after the consumer transaction's snapshot began (or a delete is still visible to an older reader). - Entity removals: on the exact duplicate-UUID constraint error, roll back, start a fresh transaction and retry once; reuse the identity that is now visible. If it is still unresolved, drop only that removal and continue, instead of retrying the same event forever and blocking the queue. - Entity interactions: same refresh-and-retry, loading the existing identity so the interaction is recorded instead of dropped (#1002 only dropped it). - Match the conflict on the event's own UUID, so other constraint errors are still retried and reported. - Reopen the entity_spawn prepared statement when DuckDB closed it after a constraint violation. - Summarize skipped records in one console warning every 5 minutes instead of a stack trace per event. Refs #998 Co-Authored-By: Claude Opus 5.5 --- .../process/EntityUuidConflictReporter.java | 50 ++++++ .../coreprotect/consumer/process/Process.java | 154 +++++++++++++++--- .../RelationalConsumerWriteBatch.java | 3 +- 3 files changed, 186 insertions(+), 21 deletions(-) create mode 100644 src/main/java/net/coreprotect/consumer/process/EntityUuidConflictReporter.java diff --git a/src/main/java/net/coreprotect/consumer/process/EntityUuidConflictReporter.java b/src/main/java/net/coreprotect/consumer/process/EntityUuidConflictReporter.java new file mode 100644 index 000000000..e3b645f6a --- /dev/null +++ b/src/main/java/net/coreprotect/consumer/process/EntityUuidConflictReporter.java @@ -0,0 +1,50 @@ +package net.coreprotect.consumer.process; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.function.LongSupplier; +import java.util.logging.Logger; + +final class EntityUuidConflictReporter { + private static final long INTERVAL_NANOS = TimeUnit.MINUTES.toNanos(5); + private final LongSupplier clock; + private final Consumer warning; + private boolean warned; + private long lastWarning; + private long interactions; + private long removals; + private UUID lastUuid; + + EntityUuidConflictReporter() { + this(System::nanoTime, message -> Logger.getLogger("CoreProtect").warning("[CoreProtect] " + message)); + } + + EntityUuidConflictReporter(LongSupplier clock, Consumer warning) { + this.clock = clock; + this.warning = warning; + } + + synchronized void record(boolean removal, UUID uuid) { + if (removal) removals++; + else interactions++; + lastUuid = uuid; + flush(); + } + + synchronized void flush() { + if (interactions == 0 && removals == 0) return; + long now = clock.getAsLong(); + if (warned && now - lastWarning < INTERVAL_NANOS) return; + String skipped = interactions == 0 ? "" : interactions + " entity interaction record(s)"; + if (removals > 0) skipped += (skipped.isEmpty() ? "" : " and ") + removals + " entity removal record(s)"; + warning.accept("Skipped " + skipped + " due to unresolved DuckDB UUID conflicts. " + + "These records were not saved; consumer processing continues. Last UUID: " + lastUuid + + ". Repeated warnings are summarized every 5 minutes."); + warned = true; + lastWarning = now; + interactions = 0; + removals = 0; + lastUuid = null; + } +} diff --git a/src/main/java/net/coreprotect/consumer/process/Process.java b/src/main/java/net/coreprotect/consumer/process/Process.java index 786482118..a45690f93 100755 --- a/src/main/java/net/coreprotect/consumer/process/Process.java +++ b/src/main/java/net/coreprotect/consumer/process/Process.java @@ -6,6 +6,7 @@ import java.sql.SQLNonTransientException; import java.sql.Statement; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.IdentityHashMap; import java.util.LinkedHashMap; @@ -80,6 +81,7 @@ public class Process { public static int lastLockUpdate = 0; private static volatile int currentConsumerSize = 0; + private static EntityUuidConflictReporter entityUuidConflicts = new EntityUuidConflictReporter(); private static final int MAX_PREPARATION_FAILURES = 3; private static final int CLICKHOUSE_ACCESS_DENIED = 497; private static final Map preparationFailures = new IdentityHashMap<>(); @@ -97,6 +99,7 @@ public static int getCurrentConsumerSize() { protected static void resetPreparationFailures() { preparationFailures.clear(); + entityUuidConflicts = new EntityUuidConflictReporter(); } protected static int consumerDelay(boolean backlog) { @@ -133,6 +136,7 @@ protected static void updateLockTable(ConsumerWriteBatch batch, int locked) { } protected static void processConsumer(int processId, boolean lastRun) { + entityUuidConflicts.flush(); List pendingEntitySpawnLogs = new ArrayList<>(); Map entitySpawnIdentities = new LinkedHashMap<>(); Map pendingEntityIdentityConfirmations = new LinkedHashMap<>(); @@ -361,19 +365,14 @@ else if (ConfigHandler.databaseType.isColumnar()) { EntitySpawnIdentity[] loggedIdentity = { existingIdentity }; boolean[] identityActive = new boolean[1]; try { - writeBatch.executeAtomically("entity_interaction_log", () -> { - if (loggedIdentity[0] == null) { - loggedIdentity[0] = EntitySpawnStatement.insertIdentity(batch, interaction.getTime(), interaction.getEntityUuid(), interaction.getOrigin(), interaction.getCurrentLocation()); - } - identityActive[0] = EntityInteractionLogger.log(batch, loggedIdentity[0], interaction, logContext); - }); + logEntityInteraction(connection, writeBatch, interaction, logContext, loggedIdentity, identityActive); + } + catch (UnresolvedEntityInteractionException e) { + Database.acknowledgeRollbackOnlyTransaction(); + duplicateEntityUuidFailure = e; + break; } catch (Exception e) { - if (shouldDiscardFailedEvent(ConfigHandler.databaseType, action, e)) { - Database.acknowledgeRollbackOnlyTransaction(); - duplicateEntityUuidFailure = e; - break; - } pendingEntityInteractions.add(new PendingEntityInteraction(user, interaction, false, true)); throw e; } @@ -518,7 +517,18 @@ else if (existingSpawnIdentity == null) { entitySpawnIdentities.putIfAbsent(update.getUuid(), previousIdentity); } } - EntitySpawnIdentity createdIdentity = entitySpawnUpdates.apply(update); + EntitySpawnIdentity createdIdentity; + try { + createdIdentity = applyEntitySpawnUpdate(connection, writeBatch, entitySpawnUpdates, update); + } + catch (UnresolvedEntityRemovalException e) { + // The event owns this isolated transaction. Only retire it + // after the acknowledged rollback below succeeds. + Database.acknowledgeRollbackOnlyTransaction(); + entitySpawnUpdates.afterRetain(); + duplicateEntityUuidFailure = e; + break; + } if (createdIdentity != null) { entitySpawnIdentities.put(createdIdentity.getUuid(), createdIdentity); promotedEntityIdentities.add(createdIdentity.getUuid()); @@ -600,9 +610,18 @@ else if (existingSpawnIdentity == null) { return; } if (duplicateEntityUuidFailure != null) { - EntityInteraction interaction = (EntityInteraction) consumerObject.get(id); - cancelEntityInteractionPromotion(interaction); - ErrorReporter.report(new IllegalStateException("Dropped entity interaction after a duplicate DuckDB entity UUID prevented identity creation: " + interaction.getEntityUuid(), duplicateEntityUuidFailure)); + if (action == ENTITY_SPAWN_UPDATE) { + EntitySpawnData removal = (EntitySpawnData) consumerObject.get(id); + EntitySpawnTracking.clearTracking(removal.getUuid()); + entityUuidConflicts.record(true, removal.getUuid()); + ErrorReporter.report(new IllegalStateException("Dropped entity removal after a duplicate DuckDB entity UUID remained unresolved in a fresh transaction; terminal removal was not recorded: " + removal.getUuid(), duplicateEntityUuidFailure), false); + } + else { + EntityInteraction interaction = (EntityInteraction) consumerObject.get(id); + cancelEntityInteractionPromotion(interaction); + entityUuidConflicts.record(false, interaction.getEntityUuid()); + ErrorReporter.report(new IllegalStateException("Dropped entity interaction after a duplicate DuckDB entity UUID prevented identity creation: " + interaction.getEntityUuid(), duplicateEntityUuidFailure), false); + } retryConsumerBatch(processId, consumerData, users, consumerObject, processedThrough); return; } @@ -1109,17 +1128,112 @@ private static boolean isPermanentPreparationFailure(Throwable failure) { return !sqlFailure; } - static boolean shouldDiscardFailedEvent(DatabaseType databaseType, int action, Throwable failure) { - if (!databaseType.isDuckDB() || action != ENTITY_INTERACTION) { - return false; + private static void logEntityInteraction(Connection connection, ConsumerWriteBatch batch, + EntityInteraction interaction, EntityInteractionLogger.LogContext context, + EntitySpawnIdentity[] identity, boolean[] active) throws Exception { + Database.SavepointOperation operation = () -> { + if (identity[0] == null) { + identity[0] = EntitySpawnStatement.insertIdentity(batch, interaction.getTime(), interaction.getEntityUuid(), interaction.getOrigin(), interaction.getCurrentLocation()); + } + active[0] = EntityInteractionLogger.log(batch, identity[0], interaction, context); + }; + try { + batch.executeAtomically("entity_interaction_log", operation); + return; + } + catch (Exception failure) { + if (identity[0] != null || !isDuplicateEntityUuidFailure(interaction.getEntityUuid(), failure)) { + throw failure; + } + // This interaction owns an isolated DuckDB transaction. Refresh the + // snapshot only after confirming rollback of the failed insert. + Database.acknowledgeRollbackOnlyTransaction(); + if (!batch.commit()) { + throw new SQLException("Unable to roll back conflicting entity interaction", failure); + } + batch.begin(); + } + + identity[0] = EntitySpawnStatement.loadIdentities(connection, Collections.singleton(interaction.getEntityUuid())).get(interaction.getEntityUuid()); + try { + batch.executeAtomically("entity_interaction_retry", operation); } + catch (Exception failure) { + if (identity[0] != null || !isDuplicateEntityUuidFailure(interaction.getEntityUuid(), failure)) { + throw failure; + } + throw new UnresolvedEntityInteractionException(failure); + } + } + private static final class UnresolvedEntityInteractionException extends SQLException { + private static final long serialVersionUID = 1L; + + private UnresolvedEntityInteractionException(Exception cause) { + super("Entity interaction UUID remains unresolved after transaction refresh", cause); + } + } + + private static EntitySpawnIdentity applyEntitySpawnUpdate(Connection connection, ConsumerWriteBatch batch, + ConsumerEntitySpawnUpdates updates, EntitySpawnData data) throws Exception { + if (!ConfigHandler.databaseType.isDuckDB() || data.getOperation() != EntitySpawnData.Operation.REMOVED) { + return updates.apply(data); + } + + EntitySpawnIdentity[] identity = new EntitySpawnIdentity[1]; + try { + batch.executeAtomically("entity_removal", () -> identity[0] = updates.apply(data)); + return identity[0]; + } + catch (Exception failure) { + if (!isDuplicateEntityRemovalFailure(data, failure)) { + throw failure; + } + // DuckDB constraint failures abort the transaction. A lookup here + // cannot reconcile a row committed after the old snapshot began. + Database.acknowledgeRollbackOnlyTransaction(); + if (!batch.commit()) { + throw new SQLException("Unable to roll back conflicting entity removal", failure); + } + updates.afterRetain(); + batch.begin(); + } + + try { + batch.executeAtomically("entity_removal_retry", () -> identity[0] = updates.apply(data)); + } + catch (Exception failure) { + if (!isDuplicateEntityRemovalFailure(data, failure)) { + throw failure; + } + throw new UnresolvedEntityRemovalException(failure); + } + // Reuse the reconciled identity for later events in this consumer batch. + return identity[0] != null ? identity[0] + : EntitySpawnStatement.loadIdentities(connection, Collections.singleton(data.getUuid())).get(data.getUuid()); + } + + private static final class UnresolvedEntityRemovalException extends SQLException { + private static final long serialVersionUID = 1L; + + private UnresolvedEntityRemovalException(Exception cause) { + super("Entity removal UUID remains unresolved after transaction refresh", cause); + } + } + + private static boolean isDuplicateEntityRemovalFailure(EntitySpawnData data, Throwable failure) { + return data.getOperation() == EntitySpawnData.Operation.REMOVED && isDuplicateEntityUuidFailure(data.getUuid(), failure); + } + + private static boolean isDuplicateEntityUuidFailure(UUID uuid, Throwable failure) { + if (!ConfigHandler.databaseType.isDuckDB()) { + return false; + } Set visited = new HashSet<>(); while (failure != null && visited.add(failure)) { String message = failure.getMessage(); if (failure instanceof SQLException && message != null - && message.contains("Constraint Error: Duplicate key \"uuid: ") - && message.contains("violates unique constraint")) { + && message.contains("Constraint Error: Duplicate key \"uuid: " + uuid + "\" violates unique constraint")) { return true; } failure = failure.getCause(); diff --git a/src/main/java/net/coreprotect/database/RelationalConsumerWriteBatch.java b/src/main/java/net/coreprotect/database/RelationalConsumerWriteBatch.java index 66e6483ad..47f5facb2 100644 --- a/src/main/java/net/coreprotect/database/RelationalConsumerWriteBatch.java +++ b/src/main/java/net/coreprotect/database/RelationalConsumerWriteBatch.java @@ -609,7 +609,8 @@ private PreparedStatement batchStatement(int type, int index) throws SQLExceptio } private PreparedStatement entitySpawnStatement() throws SQLException { - if (entitySpawnStatement == null) { + // DuckDB closes the prepared statement after a constraint violation. + if (entitySpawnStatement == null || (databaseType.isDuckDB() && entitySpawnStatement.isClosed())) { String sql = "INSERT INTO " + ConfigHandler.prefix + "entity_spawn (time,block_rowid,kill_rowid,uuid,wid,current_wid,origin_x,origin_y,origin_z,x,y,z,yaw,pitch,data,removed) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"; entitySpawnStatement = own(prepare(sql, true)); }