From 2e48e1fe5a4f1efe619aa0607105e7285bbf545a Mon Sep 17 00:00:00 2001 From: Gabriel Hall Date: Sun, 18 Oct 2026 01:40:00 -0700 Subject: [PATCH] GH-1971 - Only claim event publications for resubmission that are still failed. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FailedEventPublications.resubmit(…) reads failed publications and claims each one via markResubmitted(…), which only required the publication to be incomplete and not RESUBMITTED. A publication that a concurrent resubmission had claimed and moved to PROCESSING could thus be claimed again, and its listener was invoked a second time. markResubmitted(…) now takes the status the caller expects the publication to have, like markAbandoned(…). processFailedPublications(…) passes FAILED, and the JDBC, JPA, MongoDB and Neo4j repositories then only claim publications with that status or without one. The resubmission of incomplete publications passes null and keeps claiming PUBLISHED and PROCESSING ones. Signed-off-by: Gabriel Hall --- .../core/DefaultEventPublicationRegistry.java | 9 ++-- .../core/EventPublicationRepository.java | 9 +++- ...aultEventPublicationRegistryUnitTests.java | 14 ++++++ .../JdbcEventPublicationRepositoryV2.java | 12 +++-- ...blicationRepositoryV2IntegrationTests.java | 29 +++++++++--- .../jpa/JpaEventPublicationRepository.java | 21 ++++++--- ...PublicationRepositoryIntegrationTests.java | 29 +++++++++--- .../MongoDbEventPublicationRepository.java | 9 ++-- ...MongoDbEventPublicationRepositoryTest.java | 43 +++++++++++++---- .../Neo4jEventPublicationRepository.java | 47 +++++++++++++------ .../Neo4jEventPublicationRepositoryTest.java | 29 +++++++++--- 11 files changed, 188 insertions(+), 63 deletions(-) diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java index b7ca18155..39734482e 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java @@ -236,7 +236,7 @@ public void processIncompletePublications(Predicate filter, LOGGER.debug(getConfirmationMessage(publications) + " found."); - processPublications(publications, filter, consumer); + processPublications(publications, null, filter, consumer); } /* @@ -262,7 +262,7 @@ public void processFailedPublications(ResubmissionOptions options, Consumer publications, Predicate filter, + private void processPublications(Collection publications, + @Nullable Status expectedCurrentStatus, Predicate filter, Consumer consumer) { publications.stream() // .filter(filter) // .forEach(it -> { - if (!events.markResubmitted(it.getIdentifier(), clock.instant())) { + if (!events.markResubmitted(it.getIdentifier(), clock.instant(), expectedCurrentStatus)) { return; } diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java index 288b2008f..6aaef82ec 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java @@ -101,15 +101,22 @@ default void markFailed(UUID identifier) {} * Marks the {@link org.springframework.modulith.events.EventPublication} with the given identifier as resubmitted. * Returns {@literal false} in case the update was not successful, which usually means that the instance identified * might have been claimed for re-submission by another application instance. + *

+ * If {@code expectedCurrentStatus} is given, the publication is only claimed if it currently has that status or no + * status at all. This guards against claiming a publication that a concurrent resubmission has claimed and moved on + * (e.g. to {@link org.springframework.modulith.events.EventPublication.Status#PROCESSING}) since it was read; used + * when resubmitting previously-read {@link org.springframework.modulith.events.EventPublication.Status#FAILED} + * publications. Pass {@literal null} to claim any incomplete publication not currently resubmitted. * * @param identifier must not be {@literal null}. * @param resubmissionDate must not be {@literal null}. + * @param expectedCurrentStatus can be {@literal null}. * @return whether the {@link org.springframework.modulith.events.EventPublication} with the given identifier was * marked as resubmitted. * @since 2.0 * @see org.springframework.modulith.events.EventPublication.Status#RESUBMITTED */ - default boolean markResubmitted(UUID identifier, Instant resubmissionDate) { + default boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) { return true; } diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java index 7712ab8b1..baad54c53 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistryUnitTests.java +++ b/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-1971 + void claimsFailedPublicationsForResubmissionOnlyIfStillFailed() { + + var identifier = UUID.randomUUID(); + var publication = mock(TargetEventPublication.class); + when(publication.getIdentifier()).thenReturn(identifier); + + when(repository.findFailedPublications(any())).thenReturn(List.of(publication)); + + createRegistry(Instant.now()).processFailedPublications(ResubmissionOptions.defaults(), __ -> {}); + + verify(repository).markResubmitted(eq(identifier), any(), eq(Status.FAILED)); + } + @Test // GH-1764 void marksPublicationAsAbandonedWhenAbandonPolicyApplies() { diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java index 90775f448..651e3f421 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2.java @@ -406,10 +406,14 @@ public void markFailed(UUID identifier) { /* * (non-Javadoc) - * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status) */ @Override - public boolean markResubmitted(UUID identifier, Instant instant) { + public boolean markResubmitted(UUID identifier, Instant instant, @Nullable Status expectedCurrentStatus) { + + var guard = expectedCurrentStatus == null + ? "STATUS != 'RESUBMITTED'" + : "STATUS = '%s'".formatted(expectedCurrentStatus.name()); var sql = asOneLine(""" UPDATE %s @@ -420,8 +424,8 @@ public boolean markResubmitted(UUID identifier, Instant instant) { WHERE ID = ? AND COMPLETION_DATE IS NULL - AND (STATUS IS NULL OR STATUS != 'RESUBMITTED') - """.formatted(settings.getTable())); + AND (STATUS IS NULL OR %s) + """.formatted(settings.getTable(), guard)); var timestamp = Timestamp.from(instant); var id = uuidToDatabase(identifier); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java index d561ea055..d0bfedbd1 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryV2IntegrationTests.java @@ -538,8 +538,8 @@ void claimsResubmissionOnce() { var now = Instant.now(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isFalse(); } @Test // GH-1321 @@ -553,7 +553,7 @@ void countsByStatus() { repository.markFailed(publication.getIdentifier()); assertOneByStatus(Status.FAILED); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertOneByStatus(Status.RESUBMITTED); } @@ -626,7 +626,7 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { var publication = createPublication(event); repository.markFailed(publication.getIdentifier()); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertThat(repository.markAbandoned(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse(); assertThat(repository.findByStatus(Status.RESUBMITTED)) @@ -635,7 +635,22 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { assertThat(repository.findByStatus(Status.ABANDONED)).isEmpty(); } - @Test // GH-1565 + @Test // GH-1971 + void doesNotResubmitFailedPublicationThatHasBeenConcurrentlyResubmitted() { + + var publication = createPublication(new TestEvent("first")); + + repository.markFailed(publication.getIdentifier()); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue(); + + repository.markProcessing(publication.getIdentifier()); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse(); + assertOneByStatus(Status.PROCESSING); + } + + @Test // GH-1565, GH-1971 void resubmitsPublicationWithNullStatusColumn() { var event = new TestEvent("first"); @@ -643,7 +658,7 @@ void resubmitsPublicationWithNullStatusColumn() { setStatusColumnToNull(publication); - assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue(); } @Test // GH-1565 @@ -668,7 +683,7 @@ void doesNotResubmitCompletedPublication() { repository.markCompleted(publication.getIdentifier(), Instant.now()); - assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), null)).isFalse(); } @Test // GH-1925 diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java index 689e3c518..011411511 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java @@ -165,7 +165,7 @@ class JpaEventPublicationRepository implements EventPublicationRepository { p.lastResubmissionDate = ?1 where p.id = ?2 and p.completionDate is null - and (p.status is null or p.status != org.springframework.modulith.events.EventPublication$Status.RESUBMITTED) + and (p.status is null or %s) """; private static final String COUNT_BY_STATUS = """ @@ -374,17 +374,24 @@ public void markFailed(UUID identifier) { /* * (non-Javadoc) - * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status) */ @Override - public boolean markResubmitted(UUID identifier, Instant resubmissionDate) { + public boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) { - var result = entityManager.createQuery(MARK_RESUBMITTED) + var guard = expectedCurrentStatus == null + ? "p.status != org.springframework.modulith.events.EventPublication$Status.RESUBMITTED" + : "p.status = ?3"; + + var query = entityManager.createQuery(MARK_RESUBMITTED.formatted(guard)) .setParameter(1, resubmissionDate) - .setParameter(2, identifier) - .executeUpdate(); + .setParameter(2, identifier); + + if (expectedCurrentStatus != null) { + query.setParameter(3, expectedCurrentStatus); + } - return result == 1; + return query.executeUpdate() == 1; } /* diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java b/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java index 301cac968..231e770c9 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java @@ -461,8 +461,8 @@ void claimsResubmissionOnce() { var now = Instant.now(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isFalse(); } @Test // GH-1375 @@ -476,7 +476,7 @@ void countsByStatus() { repository.markFailed(publication.getIdentifier()); assertOneByStatus(Status.FAILED); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertOneByStatus(Status.RESUBMITTED); } @@ -575,7 +575,7 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { var publication = createPublication(event); repository.markFailed(publication.getIdentifier()); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); em.flush(); em.clear(); @@ -591,7 +591,22 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { assertThat(repository.findByStatus(Status.ABANDONED)).isEmpty(); } - @Test // GH-1565 + @Test // GH-1971 + void doesNotResubmitFailedPublicationThatHasBeenConcurrentlyResubmitted() { + + var publication = createPublication(new TestEvent("first")); + + repository.markFailed(publication.getIdentifier()); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue(); + + repository.markProcessing(publication.getIdentifier()); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse(); + assertOneByStatus(Status.PROCESSING); + } + + @Test // GH-1565, GH-1971 void resubmitsPublicationWithNullStatusColumn() { var event = new TestEvent("first"); @@ -599,7 +614,7 @@ void resubmitsPublicationWithNullStatusColumn() { setStatusColumnToNull(publication); - assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue(); } @Test // GH-1565 @@ -628,7 +643,7 @@ void doesNotResubmitCompletedPublication() { em.flush(); em.clear(); - assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), null)).isFalse(); } @Test // GH-1925 diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java index dcc3fdcfe..15101f1f2 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java @@ -214,12 +214,15 @@ public void markFailed(UUID identifier) { /* * (non-Javadoc) - * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status) */ @Override - public boolean markResubmitted(UUID identifier, Instant resubmissionDate) { + public boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) { - var query = query(where(ID).is(identifier).and(STATUS).ne(Status.RESUBMITTED).and(COMPLETION_DATE).isNull()); + var criteria = where(ID).is(identifier).and(COMPLETION_DATE).isNull().and(STATUS); + var query = query(expectedCurrentStatus == null + ? criteria.ne(Status.RESUBMITTED) + : criteria.in(expectedCurrentStatus, null)); var update = Update.update(STATUS, Status.RESUBMITTED) .inc(COMPLETION_ATTEMPTS, 1) .set(LAST_RESUBMISSION_DATE, resubmissionDate); diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java b/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java index 353a4e047..7095d65ca 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java @@ -17,6 +17,8 @@ import static org.assertj.core.api.Assertions.*; import static org.junit.jupiter.api.Assumptions.*; +import static org.springframework.data.mongodb.core.query.Criteria.*; +import static org.springframework.data.mongodb.core.query.Query.*; import java.time.Clock; import java.time.Duration; @@ -41,6 +43,7 @@ import org.springframework.core.env.Environment; import org.springframework.data.mongodb.MongoTransactionManager; import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.query.Update; import org.springframework.modulith.events.EventPublication.Status; import org.springframework.modulith.events.ResubmissionOptions; import org.springframework.modulith.events.core.DefaultEventPublicationRegistry; @@ -405,8 +408,8 @@ void claimsResubmissionOnce() { var now = Instant.now(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isFalse(); } @Test // GH-1336, GH-1855 @@ -420,7 +423,7 @@ void countsByStatus() { repository.markFailed(publication.getIdentifier()); assertByStatus(Status.FAILED, publication.getIdentifier()); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertByStatus(Status.RESUBMITTED, publication.getIdentifier()); repository.markCompleted(publication.getIdentifier(), Instant.now()); @@ -470,7 +473,7 @@ void recognizesPreviouslyCompletedPublications(Status status) { assertThat(repository.findFailedPublications(FailedCriteria.ALL)).isEmpty(); repository.markFailed(publication.id); - assertThat(repository.markResubmitted(publication.id, now.plusSeconds(1))).isFalse(); + assertThat(repository.markResubmitted(publication.id, now.plusSeconds(1), null)).isFalse(); assertThat(mongoTemplate.findAll(MongoDbEventPublication.class, collection)).singleElement() .satisfies(it -> assertThat(it.status).isEqualTo(status)); @@ -502,7 +505,7 @@ void doesNotResubmitPublicationCompletedAfterFailedLookup() { assertThat(candidates).hasSize(1); repository.markCompleted(publication.getIdentifier(), Instant.now()); - candidates.forEach(it -> assertThat(repository.markResubmitted(it.getIdentifier(), Instant.now())).isFalse()); + candidates.forEach(it -> assertThat(repository.markResubmitted(it.getIdentifier(), Instant.now(), null)).isFalse()); assertCompleted(publication.getIdentifier()); } @@ -518,8 +521,8 @@ void resubmitsOnlyStaleIncompletePublications() { var recent = savePublicationAt(old.plusSeconds(3), Status.PUBLISHED); var completed = savePublicationAt(old.plusSeconds(4), Status.PUBLISHED); - assertThat(repository.markResubmitted(resubmitted.id, old.plusSeconds(30))).isTrue(); - assertThat(repository.markResubmitted(recent.id, now.minusSeconds(5))).isTrue(); + assertThat(repository.markResubmitted(resubmitted.id, old.plusSeconds(30), null)).isTrue(); + assertThat(repository.markResubmitted(recent.id, now.minusSeconds(5), null)).isTrue(); repository.markCompleted(completed.id, now.minusSeconds(30)); var registry = new DefaultEventPublicationRegistry(repository, Clock.fixed(now, ZoneOffset.UTC)); @@ -610,7 +613,7 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { var publication = createPublication(event); repository.markFailed(publication.getIdentifier()); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertThat(repository.markAbandoned(publication.getIdentifier(), Instant.now(), Status.FAILED)).isFalse(); assertThat(repository.findByStatus(Status.RESUBMITTED)) @@ -619,6 +622,30 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { assertThat(repository.findByStatus(Status.ABANDONED)).isEmpty(); } + @Test // GH-1971 + void resubmitsPublicationOnlyIfInExpectedStatus() { + + var processing = savePublicationAt(Instant.now(), Status.PROCESSING); + var failed = savePublicationAt(Instant.now(), Status.FAILED); + + assertThat(repository.markResubmitted(processing.id, Instant.now(), Status.FAILED)).isFalse(); + assertThat(repository.markResubmitted(failed.id, Instant.now(), Status.FAILED)).isTrue(); + + assertByStatusResult(Status.PROCESSING, processing.id); + assertByStatusResult(Status.RESUBMITTED, failed.id); + } + + @Test // GH-1971 + void resubmitsFailedPublicationWithoutStatus() { + + var publication = createPublication(new TestEvent("first")); + + mongoTemplate.updateFirst(query(where("_id").is(publication.getIdentifier())), new Update().unset("status"), + MongoDbEventPublication.class); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), Status.FAILED)).isTrue(); + } + @Test // GH-1412 void archivesPublicationsInsideTransaction() { diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java index cf33cb38e..d1513952c 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java @@ -31,6 +31,7 @@ import java.util.function.Function; import org.jspecify.annotations.Nullable; +import org.neo4j.cypherdsl.core.Condition; import org.neo4j.cypherdsl.core.Cypher; import org.neo4j.cypherdsl.core.Node; import org.neo4j.cypherdsl.core.ResultStatement; @@ -72,6 +73,7 @@ class Neo4jEventPublicationRepository 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 EXPECTED_STATUS = "expectedStatus"; // return references private static final String STATUS_COUNT = "statusCount"; @@ -166,16 +168,24 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .set(EVENT_PUBLICATION_NODE.property(STATUS).to(parameter(STATUS))) .build(); - private static final Statement RESUBMIT_STATEMENT = match(EVENT_PUBLICATION_NODE) - .where(EVENT_PUBLICATION_NODE.property(ID).eq(parameter(ID))) - .and(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).isNull()) - .and(EVENT_PUBLICATION_NODE.property(STATUS).isNull() - .or(EVENT_PUBLICATION_NODE.property(STATUS).ne(literalOf(Status.RESUBMITTED.name())))) - .set(EVENT_PUBLICATION_NODE.property(STATUS).to(parameter(STATUS))) - .set(EVENT_PUBLICATION_NODE.property(COMPLETION_ATTEMPTS) - .to(EVENT_PUBLICATION_NODE.property(COMPLETION_ATTEMPTS).add(literalOf(1)))) - .set(EVENT_PUBLICATION_NODE.property(LAST_RESUBMISSION_DATE).to(parameter(LAST_RESUBMISSION_DATE))) - .build(); + private static final Statement RESUBMIT_STATEMENT = resubmitStatement( + EVENT_PUBLICATION_NODE.property(STATUS).ne(literalOf(Status.RESUBMITTED.name()))); + + private static final Statement RESUBMIT_IF_IN_EXPECTED_STATUS_STATEMENT = resubmitStatement( + EVENT_PUBLICATION_NODE.property(STATUS).eq(parameter(EXPECTED_STATUS))); + + private static Statement resubmitStatement(Condition status) { + + return match(EVENT_PUBLICATION_NODE) + .where(EVENT_PUBLICATION_NODE.property(ID).eq(parameter(ID))) + .and(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).isNull()) + .and(EVENT_PUBLICATION_NODE.property(STATUS).isNull().or(status)) + .set(EVENT_PUBLICATION_NODE.property(STATUS).to(parameter(STATUS))) + .set(EVENT_PUBLICATION_NODE.property(COMPLETION_ATTEMPTS) + .to(EVENT_PUBLICATION_NODE.property(COMPLETION_ATTEMPTS).add(literalOf(1)))) + .set(EVENT_PUBLICATION_NODE.property(LAST_RESUBMISSION_DATE).to(parameter(LAST_RESUBMISSION_DATE))) + .build(); + } private static final Lazy COMPLETE_IN_ARCHIVE_BY_ID_STATEMENT = Lazy.of( () -> applyProperties(match(EVENT_PUBLICATION_NODE).where(EVENT_PUBLICATION_NODE.property(ID).eq(parameter(ID))) @@ -705,16 +715,23 @@ public void markProcessing(UUID identifier) { /* * (non-Javadoc) - * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markResubmitted(java.util.UUID, java.time.Instant, org.springframework.modulith.events.EventPublication.Status) */ @Override - public boolean markResubmitted(UUID identifier, Instant resubmissionDate) { + public boolean markResubmitted(UUID identifier, Instant resubmissionDate, @Nullable Status expectedCurrentStatus) { + + var statement = expectedCurrentStatus == null ? RESUBMIT_STATEMENT : RESUBMIT_IF_IN_EXPECTED_STATUS_STATEMENT; - var update = neo4jClient.query(renderer.render(RESUBMIT_STATEMENT)) + var query = neo4jClient.query(renderer.render(statement)) .bind(Values.value(identifier.toString())).to(ID) .bind(Status.RESUBMITTED.name()).to(STATUS) - .bind(Values.value(resubmissionDate.atOffset(ZoneOffset.UTC))).to(LAST_RESUBMISSION_DATE) - .run(); + .bind(Values.value(resubmissionDate.atOffset(ZoneOffset.UTC))).to(LAST_RESUBMISSION_DATE); + + if (expectedCurrentStatus != null) { + query = query.bind(expectedCurrentStatus.name()).to(EXPECTED_STATUS); + } + + var update = query.run(); return update.counters().propertiesSet() > 1; } diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java b/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java index 8ee4adbf7..3072a1529 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java @@ -376,7 +376,7 @@ void countsByStatus() { repository.markFailed(publication.getIdentifier()); assertOneByStatus(EventPublication.Status.FAILED); - var resubmitted = repository.markResubmitted(publication.getIdentifier(), Instant.now()); + var resubmitted = repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertThat(resubmitted).isTrue(); assertOneByStatus(EventPublication.Status.RESUBMITTED); @@ -405,8 +405,8 @@ void claimsResubmissionOnce() { var now = Instant.now(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue(); - assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), now, null)).isFalse(); } @Test // GH-1337 @@ -546,7 +546,7 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { var publication = createPublication(event); repository.markFailed(publication.getIdentifier()); - repository.markResubmitted(publication.getIdentifier(), Instant.now()); + repository.markResubmitted(publication.getIdentifier(), Instant.now(), null); assertThat(repository.markAbandoned(publication.getIdentifier(), Instant.now(), EventPublication.Status.FAILED)).isFalse(); assertThat(repository.findByStatus(EventPublication.Status.RESUBMITTED)) @@ -555,7 +555,22 @@ void doesNotAbandonPublicationThatHasBeenConcurrentlyResubmitted() { assertThat(repository.findByStatus(EventPublication.Status.ABANDONED)).isEmpty(); } - @Test // GH-1565 + @Test // GH-1971 + void doesNotResubmitFailedPublicationThatHasBeenConcurrentlyResubmitted() { + + var publication = createPublication(new TestEvent("first")); + + repository.markFailed(publication.getIdentifier()); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), EventPublication.Status.FAILED)).isTrue(); + + repository.markProcessing(publication.getIdentifier()); + + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), EventPublication.Status.FAILED)).isFalse(); + assertOneByStatus(EventPublication.Status.PROCESSING); + } + + @Test // GH-1565, GH-1971 void resubmitsPublicationWithNullStatusProperty() { var event = new TestEvent("first"); @@ -563,7 +578,7 @@ void resubmitsPublicationWithNullStatusProperty() { removeStatusProperty(publication); - assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isTrue(); + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), EventPublication.Status.FAILED)).isTrue(); } @Test // GH-1565 @@ -588,7 +603,7 @@ void doesNotResubmitCompletedPublication() { repository.markCompleted(publication.getIdentifier(), Instant.now()); - assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now())).isFalse(); + assertThat(repository.markResubmitted(publication.getIdentifier(), Instant.now(), null)).isFalse(); } @Test // GH-1925