fix: properly track delayedDbTasks

What happened:
If two different UniqueData instances wanted to update the same column (in different rows) at the same time, and they both were doing so with a delay (for example if they used @UpdateInterval), then one of them would replace the other in the queue. The last enqueued update (within the delay window) would be the only one executed on the PG DB, both would be executed on the H2 DB.

Author: Sammster10

core/src/main/java/net/staticstudios/data/DataManager.java

@@ -1548,7 +1548,7 @@ public void set(String schema, String table, String column, ColumnValuePairs idC
             values.add(columnValuePair.value());
         }
         try {
-            dataAccessor.executeUpdate(SQLTransaction.Statement.of(h2Sql, pgSql), values, delay);
+            dataAccessor.executeUpdate(idColumns, SQLTransaction.Statement.of(h2Sql, pgSql), values, delay);
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }

core/src/main/java/net/staticstudios/data/UniqueData.java

@@ -64,7 +64,7 @@ public synchronized void delete() {
         @Language("SQL") String sql = stringBuilder.toString();
 
         try {
-            dataManager.getDataAccessor().executeUpdate(SQLTransaction.Statement.of(sql, sql), values, 0);
+            dataManager.getDataAccessor().executeUpdate(idColumns, SQLTransaction.Statement.of(sql, sql), values, 0);
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }

core/src/main/java/net/staticstudios/data/impl/DataAccessor.java

@@ -17,8 +17,8 @@ public interface DataAccessor {
 
     ResultSet executeQuery(@Language("SQL") String sql, List<Object> values) throws SQLException;
 
-    default void executeUpdate(SQLTransaction.Statement statement, List<Object> values, int delay) throws SQLException {
-        executeTransaction(new SQLTransaction().update(statement, values), delay);
+    default void executeUpdate(@Nullable ColumnValuePairs idColumns, SQLTransaction.Statement statement, List<Object> values, int delay) throws SQLException {
+        executeTransaction(new SQLTransaction(idColumns).update(statement, values), delay);
     }
 
     void executeTransaction(SQLTransaction transaction, int delay) throws SQLException;

core/src/main/java/net/staticstudios/data/impl/data/ReferenceImpl.java

@@ -232,7 +232,7 @@ private void setLocal(@Nullable T value) {
         @Language("SQL") String sql = sqlBuilder.toString();
 
         try {
-            holder.getDataManager().getDataAccessor().executeUpdate(SQLTransaction.Statement.of(sql, sql), values, 0);
+            holder.getDataManager().getDataAccessor().executeUpdate(holder.getIdColumns(), SQLTransaction.Statement.of(sql, sql), values, 0);
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }
@@ -315,7 +315,7 @@ private void setReferenced(@Nullable T value) {
         @Language("SQL") String updateSql = updateSqlBuilder.toString();
 
         List<Object> existingIdValues = new ArrayList<>();
-        SQLTransaction transaction = new SQLTransaction();
+        SQLTransaction transaction = new SQLTransaction(holder.getIdColumns());
 
         transaction.query(SQLTransaction.Statement.of(selectExistingSql, selectExistingSql), () -> holderLinkValues, rs -> {
             try {

core/src/main/java/net/staticstudios/data/impl/h2/H2DataAccessor.java

@@ -57,7 +57,7 @@ public class H2DataAccessor implements DataAccessor {
     private final Set<SchemaTable> knownTables = new HashSet<>();
     private final DataManager dataManager;
     private final PostgresListener postgresListener;
-    private final Map<List<Pair<String, String>>, Runnable> delayedTasks = new ConcurrentHashMap<>();
+    private final Map<EnqueuedDatabaseTaskKey, Runnable> delayedTasks = new ConcurrentHashMap<>();
     private final Map<String, Runnable> delayedRedisTasks = new ConcurrentHashMap<>();
     private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(thread -> {
         Thread t = new Thread(thread);
@@ -798,10 +798,10 @@ private List<String> getColumnsInTable(String schema, String table) throws SQLEx
     }
 
     private void runDatabaseTask(SQLTransaction transaction, int delay) {
-        List<Pair<String, String>> key = new ArrayList<>(transaction.getOperations().size());
-        for (SQLTransaction.Operation operation : transaction.getOperations()) {
-            SQLTransaction.Statement sqlStatement = operation.getStatement();
-            key.add(Pair.of(sqlStatement.getH2Sql(), sqlStatement.getPgSql()));
+        if (transaction.getHolderIds() == null) {
+            // has to run now, since we don't know who we are waiting on. Future updates need to replace only their holder's tasks, so id cols are a must.
+            logger.warn("Transaction {} does not have holder IDs, running immediately. Requested delay was {}ms", transaction, delay);
+            delay = -1;
         }
 
         Runnable runnable = () -> taskQueue.submitTask(connection -> {
@@ -839,6 +839,14 @@ private void runDatabaseTask(SQLTransaction transaction, int delay) {
             runnable.run();
             return;
         }
+
+        SQLTransaction.Statement[] statements = new SQLTransaction.Statement[transaction.getOperations().size()];
+        for (int i = 0; i < transaction.getOperations().size(); i++) {
+            statements[i] = transaction.getOperations().get(i).getStatement();
+        }
+
+        EnqueuedDatabaseTaskKey key = new EnqueuedDatabaseTaskKey(transaction.getHolderIds(), statements);
+
         if (delayedTasks.put(key, runnable) == null) {
             scheduledExecutorService.schedule(() -> {
                 Runnable removed = delayedTasks.remove(key);

core/src/main/java/net/staticstudios/data/util/EnqueuedDatabaseTaskKey.java

@@ -0,0 +1,6 @@
+package net.staticstudios.data.util;
+
+import org.jetbrains.annotations.NotNull;
+
+public record EnqueuedDatabaseTaskKey(@NotNull ColumnValuePairs holderPairs, SQLTransaction.Statement[] statements) {
+}

core/src/main/java/net/staticstudios/data/util/SQLTransaction.java

@@ -12,15 +12,24 @@
 import java.util.function.Supplier;
 
 public class SQLTransaction {
+    private @Nullable ColumnValuePairs holderIds;
     private final List<Operation> operations = new ArrayList<>();
 
     public SQLTransaction() {
     }
 
+    public SQLTransaction(@Nullable ColumnValuePairs holderIds) {
+        this.holderIds = holderIds;
+    }
+
     public List<Operation> getOperations() {
         return operations;
     }
 
+    public @Nullable ColumnValuePairs getHolderIds() {
+        return holderIds;
+    }
+
     public SQLTransaction query(Statement statement, Supplier<List<Object>> valuesSupplier, @NotNull Consumer<ResultSet> resultHandler) {
         Preconditions.checkNotNull(resultHandler, "Use update() method for statements without result handlers");
         operations.add(new Operation(statement, valuesSupplier, resultHandler, () -> true));