GH-1321 - Fix application of FailedCriteria in failed publication lookups.

We now correctly build up limit clauses for database requiring the ANSI syntax. Also, we convert the reference Instant into a Timestamp before binding it as parameter value.

Author: odrotbohm

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

@@ -17,7 +17,6 @@
 
 import java.time.Clock;
 import java.time.Duration;
-import java.time.Instant;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
@@ -35,6 +34,7 @@
 import org.springframework.modulith.events.EventPublication;
 import org.springframework.modulith.events.EventPublication.Status;
 import org.springframework.modulith.events.ResubmissionOptions;
+import org.springframework.modulith.events.core.EventPublicationRepository.FailedCriteria;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
@@ -234,18 +234,9 @@ public void processFailedPublications(ResubmissionOptions options, Consumer<Targ
 			return;
 		}
 
-		var criteria = new EventPublicationRepository.FailedCriteria() {
-
-			@Override
-			public Instant getPublicationDateReference() {
-				return clock.instant().minus(options.getMinAge());
-			}
-
-			@Override
-			public int getMaxItemsToRead() {
-				return Math.min(options.getBatchSize(), options.getBatchSize() - currentlyResubmitted);
-			}
-		};
+		var criteria = FailedCriteria.ALL
+				.withPublicationsPublishedBefore(clock.instant().minus(options.getMinAge()))
+				.withItemsToRead(Math.min(options.getBatchSize(), options.getBatchSize() - currentlyResubmitted));
 
 		processPublications(events.findFailedPublications(criteria), options.getFilter(), consumer);
 	}

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

@@ -209,20 +209,17 @@ default int countByStatus(Status status) {
 		return 0;
 	}
 
-	interface FailedCriteria {
+	static class FailedCriteria {
 
-		public static FailedCriteria ALL = new FailedCriteria() {
+		public static FailedCriteria ALL = new FailedCriteria(-1, null);
 
-			@Override
-			public int getMaxItemsToRead() {
-				return -1;
-			}
+		private final long maxItemsToRead;
+		private final @Nullable Instant publicationDateReference;
 
-			@Override
-			public @Nullable Instant getPublicationDateReference() {
-				return null;
-			}
-		};
+		private FailedCriteria(long maxItemsToRead, @Nullable Instant publicationDateReference) {
+			this.maxItemsToRead = maxItemsToRead;
+			this.publicationDateReference = publicationDateReference;
+		}
 
 		/**
 		 * The reference date to use as cutoff when selecting failed
@@ -231,12 +228,24 @@ public int getMaxItemsToRead() {
 		 * @return can be {@literal null}.
 		 */
 		@Nullable
-		Instant getPublicationDateReference();
+		public Instant getPublicationDateReference() {
+			return publicationDateReference;
+		}
+
+		public FailedCriteria withPublicationsPublishedBefore(Instant reference) {
+			return new FailedCriteria(maxItemsToRead, reference);
+		}
 
 		/**
 		 * The number of {@link org.springframework.modulith.events.EventPublication}s to read. Return -1 to indicate you
 		 * want to read all items.
 		 */
-		int getMaxItemsToRead();
+		public long getMaxItemsToRead() {
+			return maxItemsToRead;
+		}
+
+		public FailedCriteria withItemsToRead(long itemsToRead) {
+			return new FailedCriteria(itemsToRead, publicationDateReference);
+		}
 	}
 }

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

@@ -87,6 +87,11 @@ UUID databaseToUUID(Object id) {
 		boolean isSchemaSupported() {
 			return false;
 		}
+
+		@Override
+		String getLimitClause(long limit) {
+			return " OFFSET 0 ROWS FETCH NEXT " + limit + " ROWS ONLY";
+		}
 	},
 
 	ORACLE("oracle", "Oracle") {
@@ -105,6 +110,11 @@ UUID databaseToUUID(Object id) {
 		boolean isSchemaSupported() {
 			return false;
 		}
+
+		@Override
+		String getLimitClause(long limit) {
+			return " FETCH FIRST " + limit + " ROWS ONLY";
+		}
 	};
 
 	static final String SCHEMA_NOT_SUPPORTED = "Setting the schema name is not supported!";
@@ -143,6 +153,10 @@ String getArchiveSchemaResourceFilename(boolean legacy) {
 		return getSchemaBase(legacy) + "-archive.sql";
 	}
 
+	String getLimitClause(long limit) {
+		return " LIMIT " + limit;
+	}
+
 	private String getSchemaBase(boolean legacy) {
 		return "/schemas/" + (legacy ? "v1" : "v2") + "/schema-" + value;
 	}

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

@@ -533,19 +533,18 @@ public List<TargetEventPublication> findFailedPublications(FailedCriteria criter
 		if (instant != null) {
 
 			sql += """
-					AND PUBLICATION_DATE < ?
+					 AND PUBLICATION_DATE < ?
 					""";
 
-			args.add(instant);
+			args.add(Timestamp.from(instant));
 		}
 
+		sql += " ORDER BY PUBLICATION_DATE ASC";
+
 		var itemsToRead = criteria.getMaxItemsToRead();
 
 		if (itemsToRead != -1) {
-
-			sql += """
-					LIMIT %s
-					""".formatted(itemsToRead);
+			sql += settings.getDatabaseType().getLimitClause(itemsToRead);
 		}
 
 		var result = operations.query(asOneLine(sql), this::resultSetToPublications, args.toArray());

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

@@ -439,6 +439,37 @@ void marksPublicationAsProcessing() {
 			repository.markProcessing(publication.getIdentifier());
 		}
 
+		@Test // GH-1321
+		void looksUpFailedPublicationInBatch() {
+
+			var event = new TestEvent("first");
+			var publication = createPublication(event);
+
+			repository.markFailed(publication.getIdentifier());
+
+			assertThat(repository.findFailedPublications(FailedCriteria.ALL.withItemsToRead(10)))
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(publication.getIdentifier());
+		}
+
+		@Test // GH-1321
+		void looksUpFailedPublicationWithReferenceDate() throws Exception {
+
+			var event = new TestEvent("first");
+			var publication = createPublication(event);
+
+			repository.markFailed(publication.getIdentifier());
+
+			Thread.sleep(200);
+
+			var criteria = FailedCriteria.ALL
+					.withPublicationsPublishedBefore(publication.getPublicationDate().plusMillis(50));
+
+			assertThat(repository.findFailedPublications(criteria))
+					.extracting(TargetEventPublication::getIdentifier)
+					.containsExactly(publication.getIdentifier());
+		}
+
 		private void assertOneByStatus(Status reference) {
 
 			for (var status : Status.values()) {