Recover DuckDB entity UUID conflicts without dropping or stalling (#1039)

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 <noreply@anthropic.com>

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<String> 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<String> 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;
+    }
+}

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

@@ -6,6 +6,7 @@ import java.sql.SQLNonTransientConnectionException;
 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<Object[], Integer> preparationFailures = new IdentityHashMap<>();
@@ -97,6 +99,7 @@ public class Process {
 
     protected static void resetPreparationFailures() {
         preparationFailures.clear();
+        entityUuidConflicts = new EntityUuidConflictReporter();
     }
 
     protected static int consumerDelay(boolean backlog) {
@@ -133,6 +136,7 @@ public class Process {
     }
 
     protected static void processConsumer(int processId, boolean lastRun) {
+        entityUuidConflicts.flush();
         List<PendingEntitySpawnLog> pendingEntitySpawnLogs = new ArrayList<>();
         Map<UUID, EntitySpawnIdentity> entitySpawnIdentities = new LinkedHashMap<>();
         Map<UUID, Location> pendingEntityIdentityConfirmations = new LinkedHashMap<>();
@@ -361,19 +365,14 @@ public class Process {
                                     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 @@ public class Process {
                                                 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 @@ public class Process {
                             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 @@ public class Process {
         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<Throwable> 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();

src/main/java/net/coreprotect/database/RelationalConsumerWriteBatch.java

@@ -609,7 +609,8 @@ public final class RelationalConsumerWriteBatch implements ConsumerWriteBatch {
     }
 
     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));
         }