From 1490dc7c89e0f93c1511f81c443f3b343d1bb5cf Mon Sep 17 00:00:00 2001
From: Intelli <contact@intelli.software>
Date: Sun, 4 Oct 2026 11:05:16 -0600
Subject: [PATCH] Reduce ClickHouse consumer payload retention

---
 .../coreprotect/consumer/process/Process.java | 20 ++++++++++++++-----
 .../clickhouse/ClickHouseRowBinaryBuffer.java |  3 +++
 2 files changed, 18 insertions(+), 5 deletions(-)

diff --git a/src/main/java/net/coreprotect/consumer/process/Process.java b/src/main/java/net/coreprotect/consumer/process/Process.java
index a571e9e7..2d48c472 100755
--- a/src/main/java/net/coreprotect/consumer/process/Process.java
+++ b/src/main/java/net/coreprotect/consumer/process/Process.java
@@ -268,6 +268,7 @@ public class Process {
             entitySpawnUpdates = hasEntitySpawnUpdates ? writeBatch.entitySpawnUpdates() : null;
             if (entitySpawnUpdates != null) {
                 entitySpawnUpdates.prefetch(entitySpawnUpdateData);
+                entitySpawnUpdateData.clear();
             }
             processingStarted = true;
             for (int i = 0; i < consumerDataSize; i++) {
@@ -527,6 +528,7 @@ public class Process {
                             boolean interrupted = Consumer.interrupt;
                             boolean batchLimitReached = writeBatch.shouldCommit();
                             if ((interrupted || batchLimitReached) && !isolatedTransaction) {
+                                int batchStart = processedThrough;
                                 TransactionOutcome outcome = commit(writeBatch);
                                 if (outcome == TransactionOutcome.COMMITTED) {
                                     processedThrough = i + 1;
@@ -536,6 +538,9 @@ public class Process {
                                     completeFailedConsumerBatch(processId, consumerData, users, consumerObject, processedThrough, i + 1, outcome == TransactionOutcome.RETAINED);
                                     return;
                                 }
+                                if (ConfigHandler.databaseType.isClickHouse()) {
+                                    releaseProcessedConsumerData(processId, consumerData, users, consumerObject, batchStart, processedThrough);
+                                }
                                 boolean backlog = batchLimitReached && ConfigHandler.databaseType.isClickHouse() && i + 1 < consumerDataSize;
                                 if (interrupted || backlog) {
                                     try {
@@ -938,16 +943,21 @@ public class Process {
 
     private static void discardProcessedConsumerData(int processId, ArrayList<Object[]> consumerData, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, int count) {
         int processed = Math.min(count, consumerData.size());
-        for (int index = 0; index < processed; index++) {
+        releaseProcessedConsumerData(processId, consumerData, users, consumerObject, 0, processed);
+        consumerData.subList(0, processed).clear();
+    }
+
+    private static void releaseProcessedConsumerData(int processId, ArrayList<Object[]> consumerData, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, int fromIndex, int toIndex) {
+        for (int index = fromIndex; index < toIndex; index++) {
             Object[] data = consumerData.get(index);
-            preparationFailures.remove(data);
             if (data == null) {
                 continue;
             }
-            int id = (int) data[0];
+            preparationFailures.remove(data);
+            Integer id = (Integer) data[0];
             discardConsumerData(processId, users, consumerObject, id, (int) data[1]);
+            consumerData.set(index, null);
         }
-        consumerData.subList(0, processed).clear();
     }
 
     private static void discardFailedConsumerData(int processId, ArrayList<Object[]> consumerData, Map<Integer, String[]> users,
@@ -1040,7 +1050,7 @@ public class Process {
         return false;
     }
 
-    private static void discardConsumerData(int processId, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, int id, int action) {
+    private static void discardConsumerData(int processId, Map<Integer, String[]> users, Map<Integer, Object> consumerObject, Integer id, int action) {
         Object object = consumerObject.get(id);
         if (action == -1) {
             if (object instanceof EntityContainerRollbackUpdate) {
diff --git a/src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowBinaryBuffer.java b/src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowBinaryBuffer.java
index aef157fa..78826d13 100644
--- a/src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowBinaryBuffer.java
+++ b/src/main/java/net/coreprotect/database/clickhouse/ClickHouseRowBinaryBuffer.java
@@ -7,6 +7,7 @@ import java.io.InputStream;
 import java.sql.SQLException;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -109,6 +110,7 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
         if (rowStarted) {
             throw new IllegalStateException("ClickHouse RowBinary row is incomplete");
         }
+        Arrays.fill(values, null);
         sealed = true;
     }
 
@@ -161,6 +163,7 @@ final class ClickHouseRowBinaryBuffer implements AutoCloseable {
         sealed = true;
         rows.reset();
         rowSpans.clear();
+        Arrays.fill(values, null);
     }
 
     private void ensureWritable() {