GH-2191 - Fix state machine for more complex scenarios.

Author: meistermeier

src/main/java/org/springframework/data/neo4j/core/Neo4jTemplate.java

@@ -470,14 +470,15 @@ private <T> T processNestedRelations(Neo4jPersistentEntity<?> sourceEntity, Obje
 			}
 
 			// break recursive procession and deletion of previously created relationships
-			ProcessState processState = stateMachine.getStateOf(relationshipDescriptionObverse, relatedValuesToStore);
+			ProcessState processState = stateMachine.getStateOf(fromId, relationshipDescriptionObverse, relatedValuesToStore);
 			if (processState == ProcessState.PROCESSED_ALL_RELATIONSHIPS || processState == ProcessState.PROCESSED_BOTH) {
 				return;
 			}
 
-			// remove all relationships before creating all new if the entity is not new
-			// this avoids the usage of cache but might have significant impact on overall performance
-			if (!isParentObjectNew) {
+			// Remove all relationships before creating all new if the entity is not new and the relationship
+			// has not been processed before.
+			// This avoids the usage of cache but might have significant impact on overall performance
+			if (!isParentObjectNew && !stateMachine.hasProcessedRelationship(fromId, relationshipDescription)) {
 
 				List<Long> knownRelationshipsIds = new ArrayList<>();
 				if (idProperty != null) {
@@ -508,7 +509,7 @@ private <T> T processNestedRelations(Neo4jPersistentEntity<?> sourceEntity, Obje
 				return;
 			}
 
-			stateMachine.markAsProcessed(relationshipDescription, relatedValuesToStore);
+			stateMachine.markRelationshipAsProcessed(fromId, relationshipDescription);
 
 			for (Object relatedValueToStore : relatedValuesToStore) {
 
@@ -522,12 +523,13 @@ private <T> T processNestedRelations(Neo4jPersistentEntity<?> sourceEntity, Obje
 
 				Long relatedInternalId;
 				// No need to save values if processed
-				if (processState == ProcessState.PROCESSED_ALL_VALUES) {
+				if (stateMachine.hasProcessedValue(relatedValueToStore)) {
 					relatedInternalId = queryRelatedNode(relatedNode, targetEntity, inDatabase);
 				} else {
 					relatedInternalId = saveRelatedNode(relatedNode, relationshipContext.getAssociationTargetType(),
 							targetEntity, inDatabase);
 				}
+				stateMachine.markValueAsProcessed(relatedValueToStore);
 
 				CreateRelationshipStatementHolder statementHolder = neo4jMappingContext.createStatement(
 						sourceEntity, relationshipContext, relatedValueToStore);
@@ -556,7 +558,6 @@ private <T> T processNestedRelations(Neo4jPersistentEntity<?> sourceEntity, Obje
 				}
 			}
 
-
 		});
 
 		return (T) propertyAccessor.getBean();

src/main/java/org/springframework/data/neo4j/core/ReactiveNeo4jTemplate.java

@@ -601,14 +601,15 @@ private Mono<Void> processNestedRelations(Neo4jPersistentEntity<?> sourceEntity,
 				}
 
 				// break recursive procession and deletion of previously created relationships
-				ProcessState processState = stateMachine.getStateOf(relationshipDescriptionObverse, relatedValuesToStore);
+				ProcessState processState = stateMachine.getStateOf(fromId, relationshipDescriptionObverse, relatedValuesToStore);
 				if (processState == ProcessState.PROCESSED_ALL_RELATIONSHIPS || processState == ProcessState.PROCESSED_BOTH) {
 					return;
 				}
 
-				// remove all relationships before creating all new if the entity is not new
-				// this avoids the usage of cache but might have significant impact on overall performance
-				if (!isParentObjectNew) {
+				// Remove all relationships before creating all new if the entity is not new and the relationship
+				// has not been processed before.
+				// This avoids the usage of cache but might have significant impact on overall performance
+				if (!isParentObjectNew && !stateMachine.hasProcessedRelationship(fromId, relationshipDescription)) {
 
 					List<Long> knownRelationshipsIds = new ArrayList<>();
 					if (idProperty != null) {
@@ -642,7 +643,7 @@ private Mono<Void> processNestedRelations(Neo4jPersistentEntity<?> sourceEntity,
 					return;
 				}
 
-				stateMachine.markAsProcessed(relationshipDescription, relatedValuesToStore);
+				stateMachine.markRelationshipAsProcessed(fromId, relationshipDescription);
 
 				for (Object relatedValueToStore : relatedValuesToStore) {
 
@@ -655,12 +656,13 @@ private Mono<Void> processNestedRelations(Neo4jPersistentEntity<?> sourceEntity,
 								return Mono.just(targetEntity.isNew(relatedNode)).flatMap(isNew -> {
 									Mono<Long> relatedIdMono;
 
-									if (processState == ProcessState.PROCESSED_ALL_VALUES) {
+									if (stateMachine.hasProcessedValue(relatedValueToStore)) {
 										relatedIdMono = queryRelatedNode(relatedNode, targetEntity, inDatabase);
 									} else {
 										relatedIdMono = saveRelatedNode(relatedNode, relationshipContext.getAssociationTargetType(),
 												targetEntity, inDatabase);
 									}
+									stateMachine.markValueAsProcessed(relatedValueToStore);
 									return relatedIdMono.flatMap(relatedInternalId -> {
 
 											// if an internal id is used this must get set to link this entity in the next iteration

src/main/java/org/springframework/data/neo4j/core/mapping/NestedRelationshipProcessingStateMachine.java

@@ -17,6 +17,7 @@
 
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -48,7 +49,7 @@ public enum ProcessState {
 	/**
 	 * The set of already processed relationships.
 	 */
-	private final Set<RelationshipDescription> processedRelationshipDescriptions = new HashSet<>();
+	private final Set<RelationshipDescriptionWithSourceId> processedRelationshipDescriptions = new HashSet<>();
 
 	/**
 	 * The set of already processed related objects.
@@ -64,11 +65,11 @@ public NestedRelationshipProcessingStateMachine(Object initialObject) {
 	 * @param valuesToStore Check whether all the values in the collection have been processed
 	 * @return The state of things processed
 	 */
-	public ProcessState getStateOf(RelationshipDescription relationshipDescription, @Nullable Collection<?> valuesToStore) {
+	public ProcessState getStateOf(Object fromId, RelationshipDescription relationshipDescription, @Nullable Collection<?> valuesToStore) {
 
 		try {
 			read.lock();
-			boolean hasProcessedRelationship = hasProcessed(relationshipDescription);
+			boolean hasProcessedRelationship = hasProcessedRelationship(fromId, relationshipDescription);
 			boolean hasProcessedAllValues = hasProcessedAllOf(valuesToStore);
 			if (hasProcessedRelationship && hasProcessedAllValues) {
 				return ProcessState.PROCESSED_BOTH;
@@ -85,38 +86,97 @@ public ProcessState getStateOf(RelationshipDescription relationshipDescription,
 		}
 	}
 
+	/**
+	 * Combination of relationship description and fromId to differentiate between `equals`-wise equal relationship
+	 * descriptions by their source identifier. This is needed because sometimes the very same relationship definition
+	 * can get processed for different objects of the same entity.
+	 * One could say that this is a Tuple but it has a nicer name.
+	 */
+	private static class RelationshipDescriptionWithSourceId {
+		private final Object id;
+		private final RelationshipDescription relationshipDescription;
+
+		RelationshipDescriptionWithSourceId(Object id, RelationshipDescription relationshipDescription) {
+			this.id = id;
+			this.relationshipDescription = relationshipDescription;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+			RelationshipDescriptionWithSourceId that = (RelationshipDescriptionWithSourceId) o;
+			return id.equals(that.id) && relationshipDescription.equals(that.relationshipDescription);
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(id, relationshipDescription);
+		}
+	}
+
 	/**
 	 * Marks the passed objects as processed
 	 *
 	 * @param relationshipDescription To be marked as processed
-	 * @param valuesToStore If not {@literal null}, all non-null values will be marked as processed
 	 */
-	public void markAsProcessed(RelationshipDescription relationshipDescription, @Nullable Collection<?> valuesToStore) {
+	public void markRelationshipAsProcessed(Object fromId, RelationshipDescription relationshipDescription) {
 
 		try {
 			write.lock();
-			this.processedRelationshipDescriptions.add(relationshipDescription);
-			if (valuesToStore != null) {
-				valuesToStore.stream().filter(v -> v != null).forEach(processedObjects::add);
-			}
+			this.processedRelationshipDescriptions.add(new RelationshipDescriptionWithSourceId(fromId, relationshipDescription));
 		} finally {
 			write.unlock();
 		}
 	}
+	/**
+	 * Marks the passed objects as processed
+	 *
+	 * @param valueToStore If not {@literal null}, all non-null values will be marked as processed
+	 */
+	public void markValueAsProcessed(Object valueToStore) {
 
-	private boolean hasProcessedAllOf(@Nullable Collection<?> valuesToStore) {
-		// there can be null elements in the unified collection of values to store.
-		if (valuesToStore == null) {
-			return false;
+		try {
+			write.lock();
+			this.processedObjects.add(valueToStore);
+		} finally {
+			write.unlock();
 		}
-		return processedObjects.containsAll(valuesToStore);
 	}
 
-	private boolean hasProcessed(RelationshipDescription relationshipDescription) {
+	/**
+	 * Checks if the value has already been processed.
+	 *
+	 * @param value the object that should be looked for in the registry.
+ 	 * @return processed yes (true) / no (false)
+	 */
+	public boolean hasProcessedValue(Object value) {
+		return processedObjects.contains(value);
+	}
 
+	/**
+	 * Checks if the relationship has already been processed.
+	 *
+	 * @param relationshipDescription the relationship that should be looked for in the registry.
+	 * @return processed yes (true) / no (false)
+	 */
+	public boolean hasProcessedRelationship(Object fromId, @Nullable RelationshipDescription relationshipDescription) {
 		if (relationshipDescription != null) {
-			return processedRelationshipDescriptions.contains(relationshipDescription);
+			return processedRelationshipDescriptions.contains(new RelationshipDescriptionWithSourceId(fromId, relationshipDescription));
 		}
 		return false;
 	}
+
+	private boolean hasProcessedAllOf(@Nullable Collection<?> valuesToStore) {
+		// there can be null elements in the unified collection of values to store.
+		if (valuesToStore == null) {
+			return false;
+		}
+		return processedObjects.containsAll(valuesToStore);
+	}
+
 }

src/test/java/org/springframework/data/neo4j/integration/imperative/OptimisticLockingIT.java

@@ -18,6 +18,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -307,6 +308,39 @@ void shouldDoThings(@Autowired VersionedThingRepository repository) {
 		}
 	}
 
+	@Test // GH-2191
+	void shouldNotTraverseToBidiRelatedThingsWithOldVersion(@Autowired VersionedThingRepository repository) {
+		VersionedThing thing1 = new VersionedThing("Thing1");
+		VersionedThing thing2 = new VersionedThing("Thing2");
+		VersionedThing thing3 = new VersionedThing("Thing3");
+		VersionedThing thing4 = new VersionedThing("Thing4");
+
+		List<VersionedThing> thing1Relationships = new ArrayList<>();
+		thing1Relationships.add(thing2);
+		thing1Relationships.add(thing3);
+		thing1Relationships.add(thing4);
+		thing1.setOtherVersionedThings(thing1Relationships);
+		repository.save(thing1);
+		// Initially creates:
+		// Thing1-[:HAS]->Thing2
+		// Thing1-[:HAS]->Thing3
+		// Thing1-[:HAS]->Thing4
+
+		thing1 = repository.findById(thing1.getId()).get();
+		thing3 = repository.findById(thing3.getId()).get();
+		thing3.setOtherVersionedThings(Collections.singletonList(thing1));
+		repository.save(thing3);
+		// adds
+		// Thing3-[:HAS]->Thing1
+
+		try (Session session = driver.session()) {
+			Long relationshipCount = session
+					.run("MATCH (:VersionedThing)-[r:HAS]->(:VersionedThing) return count(r) as relationshipCount")
+					.single().get("relationshipCount").asLong();
+			assertThat(relationshipCount).isEqualTo(4);
+		}
+	}
+
 	interface VersionedThingRepository extends Neo4jRepository<VersionedThing, Long> {}
 
 	interface VersionedThingWithAssignedIdRepository extends Neo4jRepository<VersionedThingWithAssignedId, Long> {}

src/test/java/org/springframework/data/neo4j/integration/reactive/ReactiveOptimisticLockingIT.java

@@ -17,6 +17,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 import java.util.ArrayList;
@@ -267,6 +268,39 @@ void shouldFailOnDeleteByEntityWithWrongVersion(@Autowired VersionedThingWithAss
 
 	}
 
+	@Test // GH-2191
+	void shouldNotTraverseToBidiRelatedThingsWithOldVersion(@Autowired VersionedThingRepository repository) {
+		VersionedThing thing1 = new VersionedThing("Thing1");
+		VersionedThing thing2 = new VersionedThing("Thing2");
+		VersionedThing thing3 = new VersionedThing("Thing3");
+		VersionedThing thing4 = new VersionedThing("Thing4");
+
+		List<VersionedThing> thing1Relationships = new ArrayList<>();
+		thing1Relationships.add(thing2);
+		thing1Relationships.add(thing3);
+		thing1Relationships.add(thing4);
+		thing1.setOtherVersionedThings(thing1Relationships);
+		StepVerifier.create(repository.save(thing1))
+				.expectNextCount(1)
+				.verifyComplete();
+
+		Flux.zip(repository.findById(thing1.getId()), repository.findById(thing3.getId()))
+				.flatMap(tuple -> {
+				tuple.getT2().setOtherVersionedThings(Collections.singletonList(tuple.getT1()));
+				return repository.save(tuple.getT2());
+				})
+				.as(StepVerifier::create)
+				.expectNextCount(1)
+				.verifyComplete();
+
+		try (Session session = driver.session()) {
+			Long relationshipCount = session
+					.run("MATCH (:VersionedThing)-[r:HAS]->(:VersionedThing) return count(r) as relationshipCount")
+					.single().get("relationshipCount").asLong();
+			assertThat(relationshipCount).isEqualTo(4);
+		}
+	}
+
 	interface VersionedThingRepository extends ReactiveNeo4jRepository<VersionedThing, Long> {}
 
 	interface VersionedThingWithAssignedIdRepository