GH-1906 - Allow resubmitting least recently attempted failed publications first

spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/ResubmissionOptions.java

@@ -32,8 +32,10 @@ public class ResubmissionOptions {
 	private final int batchSize;
 	private final Duration minAge;
 	private final Predicate<EventPublication> filter;
+	private final boolean leastRecentlyAttemptedFirst;
 
-	private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Predicate<EventPublication> filter) {
+	private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Predicate<EventPublication> filter,
+			boolean leastRecentlyAttemptedFirst) {
 
 		Assert.isTrue(maxInFlight > 0, "Max in flight number must be greater than zero!");
 		Assert.isTrue(batchSize > 0, "Batch size must be greater than zero!");
@@ -45,16 +47,17 @@ private ResubmissionOptions(int maxInFlight, int batchSize, Duration minAge, Pre
 		this.batchSize = batchSize;
 		this.minAge = minAge;
 		this.filter = filter;
+		this.leastRecentlyAttemptedFirst = leastRecentlyAttemptedFirst;
 	}
 
 	/**
 	 * Creates a new {@link ResubmissionOptions} with no bound for in-flight publications, a batch size of 100, no minimum
-	 * age and including all {@link EventPublication} instances.
+	 * age, including all {@link EventPublication} instances and reading the oldest publications first.
 	 *
 	 * @return will never be {@literal null}.
 	 */
 	public static ResubmissionOptions defaults() {
-		return new ResubmissionOptions(Integer.MAX_VALUE, 100, Duration.ZERO, __ -> true);
+		return new ResubmissionOptions(Integer.MAX_VALUE, 100, Duration.ZERO, __ -> true, false);
 	}
 
 	public int getMaxInFlight() {
@@ -69,7 +72,7 @@ public int getMaxInFlight() {
 	 * @return will never be {@literal null}.
 	 */
 	public ResubmissionOptions withMaxInFlight(int maxInFlight) {
-		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst);
 	}
 
 	public int getBatchSize() {
@@ -83,7 +86,7 @@ public int getBatchSize() {
 	 * @return will never be {@literal null}.
 	 */
 	public ResubmissionOptions withBatchSize(int batchSize) {
-		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst);
 	}
 
 	public Duration getMinAge() {
@@ -97,7 +100,7 @@ public Duration getMinAge() {
 	 * @return will never be {@literal null}.
 	 */
 	public ResubmissionOptions withMinAge(Duration minAge) {
-		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst);
 	}
 
 	public Predicate<EventPublication> getFilter() {
@@ -111,6 +114,24 @@ public Predicate<EventPublication> getFilter() {
 	 * @return will never be {@literal null}.
 	 */
 	public ResubmissionOptions withFilter(Predicate<EventPublication> filter) {
-		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter);
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, leastRecentlyAttemptedFirst);
+	}
+
+	public boolean isLeastRecentlyAttemptedFirst() {
+		return leastRecentlyAttemptedFirst;
+	}
+
+	/**
+	 * Configures failed {@link EventPublication}s to be read in the order of their last re-submission, falling back to
+	 * their publication date for the ones never resubmitted. A publication resubmitted and failed again moves to the end
+	 * of the queue, so that publications that keep failing or are skipped by the {@link #withFilter(Predicate) filter}
+	 * don't fill every batch read. By default, the oldest publications are read first, which retains their publication
+	 * order.
+	 *
+	 * @return will never be {@literal null}.
+	 * @since 2.2
+	 */
+	public ResubmissionOptions withLeastRecentlyAttemptedFirst() {
+		return new ResubmissionOptions(maxInFlight, batchSize, minAge, filter, true);
 	}
 }

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java

@@ -260,7 +260,8 @@ public void processFailedPublications(ResubmissionOptions options, Consumer<Targ
 
 		var criteria = FailedCriteria.ALL
 				.withPublicationsPublishedBefore(clock.instant().minus(options.getMinAge()))
-				.withItemsToRead(itemsToRead);
+				.withItemsToRead(itemsToRead)
+				.withLeastRecentlyAttemptedFirst(options.isLeastRecentlyAttemptedFirst());
 
 		processPublications(events.findFailedPublications(criteria), options.getFilter(), consumer);
 	}

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java

@@ -254,19 +254,22 @@ default int countByStatus(Status status) {
 
 	static class FailedCriteria {
 
-		public static final FailedCriteria ALL = new FailedCriteria(-1, null);
+		public static final FailedCriteria ALL = new FailedCriteria(-1, null, false);
 
 		private final long maxItemsToRead;
 		private final @Nullable Instant publicationDateReference;
+		private final boolean leastRecentlyAttemptedFirst;
 
-		private FailedCriteria(long maxItemsToRead, @Nullable Instant publicationDateReference) {
+		private FailedCriteria(long maxItemsToRead, @Nullable Instant publicationDateReference,
+				boolean leastRecentlyAttemptedFirst) {
 
 			Assert.isTrue(maxItemsToRead == -1 || maxItemsToRead > 0,
 					() -> "Maximum number of items to read must be -1 (unlimited) or positive, but was: %d!"
 							.formatted(maxItemsToRead));
 
 			this.maxItemsToRead = maxItemsToRead;
 			this.publicationDateReference = publicationDateReference;
+			this.leastRecentlyAttemptedFirst = leastRecentlyAttemptedFirst;
 		}
 
 		/**
@@ -281,7 +284,7 @@ public Instant getPublicationDateReference() {
 		}
 
 		public FailedCriteria withPublicationsPublishedBefore(Instant reference) {
-			return new FailedCriteria(maxItemsToRead, reference);
+			return new FailedCriteria(maxItemsToRead, reference, leastRecentlyAttemptedFirst);
 		}
 
 		/**
@@ -299,7 +302,26 @@ public long getMaxItemsToRead() {
 		 * @param itemsToRead {@code -1} for no limit, or a positive value
 		 */
 		public FailedCriteria withItemsToRead(long itemsToRead) {
-			return new FailedCriteria(itemsToRead, publicationDateReference);
+			return new FailedCriteria(itemsToRead, publicationDateReference, leastRecentlyAttemptedFirst);
+		}
+
+		/**
+		 * Whether to read the {@link org.springframework.modulith.events.EventPublication}s in the order of their last
+		 * re-submission, falling back to their publication date for the ones never resubmitted, instead of the oldest
+		 * ones first.
+		 *
+		 * @since 2.2
+		 */
+		public boolean isLeastRecentlyAttemptedFirst() {
+			return leastRecentlyAttemptedFirst;
+		}
+
+		/**
+		 * @param leastRecentlyAttemptedFirst whether to read the publications least recently attempted first.
+		 * @since 2.2
+		 */
+		public FailedCriteria withLeastRecentlyAttemptedFirst(boolean leastRecentlyAttemptedFirst) {
+			return new FailedCriteria(maxItemsToRead, publicationDateReference, leastRecentlyAttemptedFirst);
 		}
 	}
 }

spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java

@@ -136,6 +136,20 @@ void processFailedPublicationsUsesRemainingInFlightWhenLessThanBatchSize() {
 		verify(repository).findFailedPublications(argThat(criteria -> criteria.getMaxItemsToRead() == 50));
 	}
 
+	@Test // GH-1906
+	void processFailedPublicationsReadsLeastRecentlyAttemptedFirstOnlyIfConfigured() {
+
+		when(repository.findFailedPublications(any())).thenReturn(Collections.emptyList());
+
+		var registry = createRegistry(Instant.now());
+
+		registry.processFailedPublications(ResubmissionOptions.defaults(), __ -> {});
+		registry.processFailedPublications(ResubmissionOptions.defaults().withLeastRecentlyAttemptedFirst(), __ -> {});
+
+		verify(repository).findFailedPublications(argThat(criteria -> !criteria.isLeastRecentlyAttemptedFirst()));
+		verify(repository).findFailedPublications(argThat(criteria -> criteria.isLeastRecentlyAttemptedFirst()));
+	}
+
 	@Test // GH-1764
 	void marksPublicationAsAbandonedWhenAbandonPolicyApplies() {
 

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java

@@ -602,7 +602,9 @@ public List<TargetEventPublication> findFailedPublications(FailedCriteria criter
 			args.add(Timestamp.from(instant));
 		}
 
-		sql += " ORDER BY PUBLICATION_DATE ASC";
+		sql += criteria.isLeastRecentlyAttemptedFirst()
+				? " ORDER BY COALESCE(LAST_RESUBMISSION_DATE, PUBLICATION_DATE) ASC"
+				: " ORDER BY PUBLICATION_DATE ASC";
 
 		var itemsToRead = criteria.getMaxItemsToRead();
 

spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java

@@ -579,6 +579,30 @@ void looksUpFailedPublicationInBatch() {
 					.containsExactly(publication.getIdentifier());
 		}
 
+		@Test // GH-1906
+		void looksUpLeastRecentlyAttemptedFailedPublicationsFirstOnlyIfConfigured() {
+
+			var now = Instant.now();
+			var first = createPublication(new TestEvent("first"), now.minusSeconds(2));
+			var second = createPublication(new TestEvent("second"), now.minusSeconds(1));
+
+			repository.markFailed(first.getIdentifier());
+			repository.markFailed(second.getIdentifier());
+
+			repository.markResubmitted(first.getIdentifier(), now);
+			repository.markFailed(first.getIdentifier());
+
+			var criteria = FailedCriteria.ALL.withItemsToRead(1);
+
+			assertThat(repository.findFailedPublications(criteria))
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(first.getIdentifier());
+
+			assertThat(repository.findFailedPublications(criteria.withLeastRecentlyAttemptedFirst(true)))
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(second.getIdentifier());
+		}
+
 		@Test // GH-1321
 		void looksUpFailedPublicationWithReferenceDate() throws Exception {
 
@@ -722,13 +746,17 @@ String archiveTable() {
 		}
 
 		private TargetEventPublication createPublication(Object event) {
+			return createPublication(event, Instant.now());
+		}
+
+		private TargetEventPublication createPublication(Object event, Instant publicationDate) {
 
 			var token = event.toString();
 
 			doReturn(token).when(serializer).serialize(event);
 			doReturn(event).when(serializer).deserialize(token, event.getClass());
 
-			return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER));
+			return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER, publicationDate));
 		}
 	}
 

spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java

@@ -470,7 +470,9 @@ public List<TargetEventPublication> findFailedPublications(FailedCriteria criter
 			args.add(instant);
 		}
 
-		query += " order by p.publicationDate asc";
+		query += criteria.isLeastRecentlyAttemptedFirst()
+				? " order by coalesce(p.lastResubmissionDate, p.publicationDate) asc"
+				: " order by p.publicationDate asc";
 
 		var jpaQuery = entityManager.createQuery(query, DefaultJpaEventPublication.class);
 

spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java

@@ -502,6 +502,30 @@ void looksUpFailedPublicationInBatch() {
 					.containsExactly(publication.getIdentifier());
 		}
 
+		@Test // GH-1906
+		void looksUpLeastRecentlyAttemptedFailedPublicationsFirstOnlyIfConfigured() {
+
+			var now = Instant.now();
+			var first = createPublication(new TestEvent("first"), now.minusSeconds(2));
+			var second = createPublication(new TestEvent("second"), now.minusSeconds(1));
+
+			repository.markFailed(first.getIdentifier());
+			repository.markFailed(second.getIdentifier());
+
+			repository.markResubmitted(first.getIdentifier(), now);
+			repository.markFailed(first.getIdentifier());
+
+			var criteria = FailedCriteria.ALL.withItemsToRead(1);
+
+			assertThat(repository.findFailedPublications(criteria))
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(first.getIdentifier());
+
+			assertThat(repository.findFailedPublications(criteria.withLeastRecentlyAttemptedFirst(true)))
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(second.getIdentifier());
+		}
+
 		@Test // GH-1375
 		void looksUpFailedPublicationWithReferenceDate() throws Exception {
 
@@ -682,10 +706,14 @@ private List<JpaEventPublication> getArchivedPublications() {
 		}
 
 		private TargetEventPublication createPublication(Object event) {
+			return createPublication(event, Instant.now());
+		}
+
+		private TargetEventPublication createPublication(Object event, Instant publicationDate) {
 
 			serializer.registerSerialization(event);
 
-			return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER));
+			return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER, publicationDate));
 		}
 
 		private void savePublicationAt(LocalDateTime date) {

spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java

@@ -15,10 +15,12 @@
  */
 package org.springframework.modulith.events.mongodb;
 
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
 import static org.springframework.data.mongodb.core.query.Criteria.*;
 import static org.springframework.data.mongodb.core.query.Query.*;
 
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
@@ -33,6 +35,8 @@
 import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
 import org.springframework.data.mongodb.core.FindAndModifyOptions;
 import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.ConditionalOperators;
 import org.springframework.data.mongodb.core.aggregation.Fields;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
@@ -63,6 +67,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository {
 	private static final String STATUS = "status";
 	private static final String COMPLETION_ATTEMPTS = "completionAttempts";
 	private static final String LAST_RESUBMISSION_DATE = "lastResubmissionDate";
+	private static final String LAST_ATTEMPT_DATE = "lastAttemptDate";
 
 	private static final Sort DEFAULT_SORT = Sort.by(PUBLICATION_DATE).ascending();
 
@@ -307,6 +312,10 @@ public List<TargetEventPublication> findFailedPublications(FailedCriteria criter
 			throw new IllegalArgumentException("Number of items to read needs to fit into an integer!");
 		}
 
+		if (criteria.isLeastRecentlyAttemptedFirst()) {
+			return readLeastRecentlyAttemptedFirst(baseCriteria, limit);
+		}
+
 		var query = defaultQuery(baseCriteria);
 
 		return readMapped(limit != -1 ? query.limit((int) limit) : query);
@@ -399,6 +408,28 @@ private List<TargetEventPublication> readMapped(Query query, String collection)
 
 	}
 
+	private List<TargetEventPublication> readLeastRecentlyAttemptedFirst(Criteria criteria, long limit) {
+
+		var lastAttemptDate = ConditionalOperators.ifNull(LAST_RESUBMISSION_DATE).thenValueOf(PUBLICATION_DATE);
+
+		var operations = new ArrayList<AggregationOperation>(List.of(
+				match(criteria),
+				addFields().addFieldWithValue(LAST_ATTEMPT_DATE, lastAttemptDate).build(),
+				sort(Sort.by(LAST_ATTEMPT_DATE).ascending())));
+
+		if (limit != -1) {
+			operations.add(limit(limit));
+		}
+
+		var aggregation = newAggregation(MongoDbEventPublication.class, operations);
+
+		return mongoTemplate.aggregate(aggregation, collection, MongoDbEventPublication.class)
+				.getMappedResults()
+				.stream()
+				.map(MongoDbEventPublicationRepository::documentToDomain)
+				.toList();
+	}
+
 	private Criteria byEventAndListenerId(Object event, PublicationTargetIdentifier identifier) {
 
 		var eventAsMongoType = mongoTemplate.getConverter().convertToMongoType(event, TypeInformation.OBJECT);