Add support for processing tasks in order within groups

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultMigrationManager.java

@@ -94,7 +94,21 @@ class DefaultMigrationManager {
               "Update length of invocation column on outbox for MySQL dialects only.",
               "ALTER TABLE TXNO_OUTBOX MODIFY COLUMN invocation MEDIUMTEXT",
               Map.of(
-                  Dialect.POSTGRESQL_9, "", Dialect.H2, "", Dialect.ORACLE, "SELECT * FROM dual")));
+                  Dialect.POSTGRESQL_9, "", Dialect.H2, "", Dialect.ORACLE, "SELECT * FROM dual")),
+          new Migration(
+              9,
+              "Add createTime column to outbox",
+              "ALTER TABLE TXNO_OUTBOX ADD COLUMN createTime TIMESTAMP(6) NULL AFTER invocation",
+              Map.of(
+                  Dialect.POSTGRESQL_9,
+                  "ALTER TABLE TXNO_OUTBOX ADD COLUMN createTime TIMESTAMP(6)",
+                  Dialect.ORACLE,
+                  "ALTER TABLE TXNO_OUTBOX ADD createTime TIMESTAMP(6)")),
+          new Migration(
+              10,
+              "Add groupId column to outbox",
+              "ALTER TABLE TXNO_OUTBOX ADD COLUMN groupId VARCHAR(250)",
+              Map.of(Dialect.ORACLE, "ALTER TABLE TXNO_OUTBOX ADD groupId VARCHAR2(250)")));
 
   static void migrate(TransactionManager transactionManager, Dialect dialect) {
     transactionManager.inTransaction(

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultPersistor.java

@@ -36,7 +36,8 @@
 public class DefaultPersistor implements Persistor, Validatable {
 
   private static final String ALL_FIELDS =
-      "id, uniqueRequestId, invocation, lastAttemptTime, nextAttemptTime, attempts, blocked, processed, version";
+      "id, uniqueRequestId, invocation, lastAttemptTime, nextAttemptTime, attempts, blocked, processed, version, "
+          + "createTime, groupId";
 
   /**
    * @param writeLockTimeoutSeconds How many seconds to wait before timing out on obtaining a write
@@ -98,7 +99,11 @@ public void migrate(TransactionManager transactionManager) {
   public void save(Transaction tx, TransactionOutboxEntry entry)
       throws SQLException, AlreadyScheduledException {
     var insertSql =
-        "INSERT INTO " + tableName + " (" + ALL_FIELDS + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
+        "INSERT INTO "
+            + tableName
+            + " ("
+            + ALL_FIELDS
+            + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
     var writer = new StringWriter();
     serializer.serializeInvocation(entry.getInvocation(), writer);
     if (entry.getUniqueRequestId() == null) {
@@ -139,6 +144,9 @@ private void setupInsert(
     stmt.setBoolean(7, entry.isBlocked());
     stmt.setBoolean(8, entry.isProcessed());
     stmt.setInt(9, entry.getVersion());
+    stmt.setTimestamp(
+        10, entry.getCreateTime() == null ? null : Timestamp.from(entry.getCreateTime()));
+    stmt.setString(11, entry.getGroupId());
   }
 
   @Override
@@ -248,6 +256,13 @@ public boolean unblock(Transaction tx, String entryId) throws Exception {
   @Override
   public List<TransactionOutboxEntry> selectBatch(Transaction tx, int batchSize, Instant now)
       throws Exception {
+    List<TransactionOutboxEntry> results = selectBatchUnordered(tx, batchSize, now);
+    results.addAll(selectBatchOrdered(tx, batchSize - results.size(), now));
+    return results;
+  }
+
+  public List<TransactionOutboxEntry> selectBatchUnordered(
+      Transaction tx, int batchSize, Instant now) throws Exception {
     String forUpdate = dialect.isSupportsSkipLock() ? " FOR UPDATE SKIP LOCKED" : "";
     //noinspection resource
     try (PreparedStatement stmt =
@@ -262,6 +277,7 @@ public List<TransactionOutboxEntry> selectBatch(Transaction tx, int batchSize, I
                     + dialect.booleanValue(false)
                     + " AND processed = "
                     + dialect.booleanValue(false)
+                    + " AND groupid IS NULL "
                     + dialect.getLimitCriteria()
                     + forUpdate)) {
       stmt.setTimestamp(1, Timestamp.from(now));
@@ -270,6 +286,56 @@ public List<TransactionOutboxEntry> selectBatch(Transaction tx, int batchSize, I
     }
   }
 
+  public List<TransactionOutboxEntry> selectBatchOrdered(
+      final Transaction tx, final int batchSize, final Instant now) throws Exception {
+    String forUpdate = dialect.isSupportsSkipLock() ? " FOR UPDATE SKIP LOCKED" : "";
+    try (PreparedStatement stmt =
+        tx.connection()
+            .prepareStatement(
+                dialect.isSupportsWindowFunctions()
+                    ? "WITH t AS"
+                        + " ("
+                        + "   SELECT RANK() OVER (PARTITION BY groupid ORDER BY createtime) AS rn, "
+                        + ALL_FIELDS
+                        + "   FROM "
+                        + tableName
+                        + "   WHERE processed = "
+                        + dialect.booleanValue(false)
+                        + " )"
+                        + " SELECT "
+                        + ALL_FIELDS
+                        + " FROM t "
+                        + " WHERE rn = 1 AND nextAttemptTime < ? AND blocked = "
+                        + dialect.booleanValue(false)
+                        + " AND groupid IS NOT NULL "
+                        + dialect.getLimitCriteria()
+                        + (dialect.isSupportsWindowFunctionsForUpdate() ? forUpdate : "")
+                    : "SELECT "
+                        + ALL_FIELDS
+                        + " FROM"
+                        + " ("
+                        + "   SELECT groupid AS group_id, MIN(createtime) AS min_create_time"
+                        + "   FROM "
+                        + tableName
+                        + "   WHERE processed = "
+                        + dialect.booleanValue(false)
+                        + "   GROUP BY group_id"
+                        + " ) AS t"
+                        + " JOIN "
+                        + tableName
+                        + " t1"
+                        + " ON t1.groupid = t.group_id AND t1.createtime = t.min_create_time"
+                        + " WHERE nextAttemptTime < ? AND blocked = "
+                        + dialect.booleanValue(false)
+                        + " AND groupid IS NOT NULL "
+                        + dialect.getLimitCriteria()
+                        + forUpdate)) {
+      stmt.setTimestamp(1, Timestamp.from(now));
+      stmt.setInt(2, batchSize);
+      return gatherResults(batchSize, stmt);
+    }
+  }
+
   @Override
   public int deleteProcessedAndExpired(Transaction tx, int batchSize, Instant now)
       throws Exception {
@@ -311,6 +377,11 @@ private TransactionOutboxEntry map(ResultSet rs) throws SQLException, IOExceptio
               .blocked(rs.getBoolean("blocked"))
               .processed(rs.getBoolean("processed"))
               .version(rs.getInt("version"))
+              .createTime(
+                  rs.getTimestamp("createTime") == null
+                      ? null
+                      : rs.getTimestamp("createTime").toInstant())
+              .groupId(rs.getString("groupId"))
               .build();
       log.debug("Found {}", entry);
       return entry;

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/Dialect.java

@@ -13,15 +13,34 @@
 @Getter
 @Beta
 public enum Dialect {
-  MY_SQL_5(false, Constants.DEFAULT_DELETE_EXPIRED_STMT, Constants.DEFAULT_LIMIT_CRITERIA), //
-  MY_SQL_8(true, Constants.DEFAULT_DELETE_EXPIRED_STMT, Constants.DEFAULT_LIMIT_CRITERIA), //
+  MY_SQL_5(
+      false,
+      false,
+      false,
+      Constants.DEFAULT_DELETE_EXPIRED_STMT,
+      Constants.DEFAULT_LIMIT_CRITERIA), //
+  MY_SQL_8(
+      true,
+      true,
+      true,
+      Constants.DEFAULT_DELETE_EXPIRED_STMT,
+      Constants.DEFAULT_LIMIT_CRITERIA), //
   POSTGRESQL_9(
+      true,
+      true,
       true,
       "DELETE FROM {{table}} WHERE id IN (SELECT id FROM {{table}} WHERE nextAttemptTime < ? AND processed = true AND blocked = false LIMIT ?)",
       Constants.DEFAULT_LIMIT_CRITERIA), //
-  H2(false, Constants.DEFAULT_DELETE_EXPIRED_STMT, Constants.DEFAULT_LIMIT_CRITERIA), //
+  H2(
+      false,
+      true,
+      true,
+      Constants.DEFAULT_DELETE_EXPIRED_STMT,
+      Constants.DEFAULT_LIMIT_CRITERIA), //
   ORACLE(
       true,
+      true,
+      false,
       "DELETE FROM {{table}} WHERE nextAttemptTime < ? AND processed = 1 AND blocked = 0 AND ROWNUM <= ?",
       Constants.ORACLE_LIMIT_CRITERIA);
 
@@ -33,6 +52,19 @@ public enum Dialect {
   @SuppressWarnings("JavaDoc")
   private final boolean supportsSkipLock;
 
+  /**
+   * @return True if dialect supports use of SQL window functions.
+   */
+  @SuppressWarnings("JavaDoc")
+  private final boolean supportsWindowFunctions;
+
+  /**
+   * @return True if dialect supports use of ({@code FOR UPDATE}) in queries using SQL window
+   *     functions.
+   */
+  @SuppressWarnings("JavaDoc")
+  private final boolean supportsWindowFunctionsForUpdate;
+
   /**
    * @return Format string for the SQL required to delete expired retained records.
    */

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutbox.java

@@ -312,6 +312,18 @@ interface ParameterizedScheduleBuilder {
      */
     ParameterizedScheduleBuilder uniqueRequestId(String uniqueRequestId);
 
+    /**
+     * Specifies a group id for the request. This defaults to {@code null}, but if non-null, enables
+     * requests to be executed in the order created within the group of requests using the same
+     * group id. If non-null, the request will not be sumbitted immediately. Instead, ordered
+     * requests are processed only with {@link TransactionOutbox#flush()}.
+     *
+     * @param groupId The group id identifying requests that must be processed in order. May be
+     *     {@code null}, in which case no ordering is enforced.
+     * @return Builder.
+     */
+    ParameterizedScheduleBuilder groupId(String groupId);
+
     /**
      * Equivalent to {@link TransactionOutbox#schedule(Class)}, but applying additional parameters
      * to the request as configured using {@link TransactionOutbox#with()}.

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutboxEntry.java

@@ -101,6 +101,10 @@ public class TransactionOutboxEntry implements Validatable {
   @Setter
   private int version;
 
+  @Getter private Instant createTime;
+
+  @Getter private String groupId;
+
   @EqualsAndHashCode.Exclude @ToString.Exclude private volatile boolean initialized;
   @EqualsAndHashCode.Exclude @ToString.Exclude private String description;
 

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutboxImpl.java

@@ -106,7 +106,7 @@ public void initialize() {
 
   @Override
   public <T> T schedule(Class<T> clazz) {
-    return schedule(clazz, null);
+    return schedule(clazz, null, null);
   }
 
   @Override
@@ -220,7 +220,7 @@ public boolean unblock(String entryId, Object transactionContext) {
     }
   }
 
-  private <T> T schedule(Class<T> clazz, String uniqueRequestId) {
+  private <T> T schedule(Class<T> clazz, String uniqueRequestId, String groupId) {
     if (!initialized.get()) {
       throw new IllegalStateException("Not initialized");
     }
@@ -236,16 +236,19 @@ private <T> T schedule(Class<T> clazz, String uniqueRequestId) {
                           extracted.getMethodName(),
                           extracted.getParameters(),
                           extracted.getArgs(),
-                          uniqueRequestId);
+                          uniqueRequestId,
+                          groupId);
                   validator.validate(entry);
                   persistor.save(extracted.getTransaction(), entry);
-                  extracted
-                      .getTransaction()
-                      .addPostCommitHook(
-                          () -> {
-                            listener.scheduled(entry);
-                            submitNow(entry);
-                          });
+                  if (groupId == null) {
+                    extracted
+                        .getTransaction()
+                        .addPostCommitHook(
+                            () -> {
+                              listener.scheduled(entry);
+                              submitNow(entry);
+                            });
+                  }
                   log.debug(
                       "Scheduled {} for running after transaction commit", entry.description());
                   return null;
@@ -311,7 +314,12 @@ private void invoke(TransactionOutboxEntry entry, Transaction transaction)
   }
 
   private TransactionOutboxEntry newEntry(
-      Class<?> clazz, String methodName, Class<?>[] params, Object[] args, String uniqueRequestId) {
+      Class<?> clazz,
+      String methodName,
+      Class<?>[] params,
+      Object[] args,
+      String uniqueRequestId,
+      String groupId) {
     return TransactionOutboxEntry.builder()
         .id(UUID.randomUUID().toString())
         .invocation(
@@ -324,6 +332,8 @@ private TransactionOutboxEntry newEntry(
         .lastAttemptTime(null)
         .nextAttemptTime(after(attemptFrequency))
         .uniqueRequestId(uniqueRequestId)
+        .createTime(clockProvider.get().instant())
+        .groupId(groupId)
         .build();
   }
 
@@ -409,18 +419,29 @@ private class ParameterizedScheduleBuilderImpl implements ParameterizedScheduleB
 
     private String uniqueRequestId;
 
+    private String groupId;
+
     @Override
     public ParameterizedScheduleBuilder uniqueRequestId(String uniqueRequestId) {
       this.uniqueRequestId = uniqueRequestId;
       return this;
     }
 
+    @Override
+    public ParameterizedScheduleBuilder groupId(final String groupId) {
+      this.groupId = groupId;
+      return this;
+    }
+
     @Override
     public <T> T schedule(Class<T> clazz) {
       if (uniqueRequestId != null && uniqueRequestId.length() > 250) {
         throw new IllegalArgumentException("uniqueRequestId may be up to 250 characters");
       }
-      return TransactionOutboxImpl.this.schedule(clazz, uniqueRequestId);
+      if (groupId != null && groupId.length() > 250) {
+        throw new IllegalArgumentException("groupId may be up to 250 characters");
+      }
+      return TransactionOutboxImpl.this.schedule(clazz, uniqueRequestId, groupId);
     }
   }
 }

transactionoutbox-core/src/test/java/com/gruelbox/transactionoutbox/acceptance/AbstractAcceptanceTest.java

@@ -2,6 +2,7 @@
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
 import static org.junit.Assert.*;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -243,6 +244,55 @@ public void success(TransactionOutboxEntry entry) {
     MatcherAssert.assertThat(ids, containsInAnyOrder("1", "2", "4", "6"));
   }
 
+  @Test
+  void orderedRequests() {
+
+    TransactionManager transactionManager = simpleTxnManager();
+
+    List<String> ids = new ArrayList<>();
+    AtomicReference<Clock> clockProvider = new AtomicReference<>(Clock.systemDefaultZone());
+
+    TransactionOutbox outbox =
+        TransactionOutbox.builder()
+            .transactionManager(transactionManager)
+            .listener(
+                new TransactionOutboxListener() {
+                  @Override
+                  public void success(TransactionOutboxEntry entry) {
+                    ids.add((String) entry.getInvocation().getArgs()[0]);
+                  }
+                })
+            .submitter(Submitter.withExecutor(Runnable::run))
+            .persistor(Persistor.forDialect(connectionDetails().dialect()))
+            .clockProvider(clockProvider::get)
+            .build();
+
+    clearOutbox();
+
+    String groupId = "groupId1";
+    transactionManager.inTransaction(
+        () -> outbox.with().groupId(groupId).schedule(ClassProcessor.class).process("2"));
+    clockProvider.set(
+        Clock.fixed(clockProvider.get().instant().minusSeconds(1), clockProvider.get().getZone()));
+    transactionManager.inTransaction(
+        () -> outbox.with().groupId(groupId).schedule(ClassProcessor.class).process("1"));
+    clockProvider.set(
+        Clock.fixed(clockProvider.get().instant().plusSeconds(2), clockProvider.get().getZone()));
+    transactionManager.inTransaction(
+        () -> outbox.with().groupId(groupId).schedule(ClassProcessor.class).process("3"));
+
+    // Ensure no tasks were processed immediately, as a group id was provided
+    MatcherAssert.assertThat(ids, equalTo(new ArrayList<>(List.of())));
+
+    // Run the clock over the threshold
+    clockProvider.set(
+        Clock.fixed(clockProvider.get().instant().plusSeconds(120), clockProvider.get().getZone()));
+    while (outbox.flush()) {}
+
+    // Ensure tasks were processed in order, as all shared the same group id
+    MatcherAssert.assertThat(ids, equalTo(new ArrayList<>(List.of("1", "2", "3"))));
+  }
+
   /**
    * Uses a simple data source transaction manager and attempts to fire a concrete class via
    * reflection.