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