Merge pull request #5883 from opsmill/jbr-conflits-02272025-stable-to-release12

Merge stable into release-1.2

Author: LucasG0

backend/infrahub/core/changelog/diff.py

@@ -177,7 +177,7 @@ def _process_node_cardinality_one_relationship(
                             value_previous=rel_prop.previous_value,
                         )
 
-        node.add_relationship(relationship=changelog_rel)
+        node.add_relationship(relationship_changelog=changelog_rel)
 
     def _convert_string_boolean_value(self, value: str | None) -> bool | None:
         """Convert string based boolean for is_protected and is_visible."""
@@ -227,7 +227,7 @@ def _process_node_cardinality_many_relationship(
 
             changelog_rel.peers.append(peer_log)
 
-        node.add_relationship(relationship=changelog_rel)
+        node.add_relationship(relationship_changelog=changelog_rel)
 
     def collect_changelogs(self) -> Sequence[tuple[DiffAction, NodeChangelog]]:
         self._populate_diff_nodes()

backend/infrahub/core/changelog/models.py

@@ -129,8 +129,8 @@ def add_property(self, name: str, value_current: bool | str | None, value_previo
     def set_parent(self, parent_id: str, parent_kind: str) -> None:
         self._parent = ChangelogRelatedNode(node_id=parent_id, node_kind=parent_kind)
 
-    def set_parent_from_relationship(self, relationship: Relationship) -> None:
-        if relationship.schema.kind == RelationshipKind.PARENT:
+    def set_parent_from_relationship(self, rel_kind: RelationshipKind) -> None:
+        if rel_kind == RelationshipKind.PARENT:
             if (
                 self.peer_status in [DiffAction.ADDED, DiffAction.UNCHANGED, DiffAction.UPDATED]
                 and self.peer_id
@@ -308,14 +308,14 @@ def add_attribute(self, attribute: AttributeChangelog) -> None:
             self.attributes[attribute.name] = attribute
 
     def add_relationship(
-        self, relationship: RelationshipCardinalityOneChangelog | RelationshipCardinalityManyChangelog
+        self, relationship_changelog: RelationshipCardinalityOneChangelog | RelationshipCardinalityManyChangelog
     ) -> None:
-        if isinstance(relationship, RelationshipCardinalityOneChangelog) and relationship.parent:
-            self.add_parent(parent=relationship.parent)
-        if relationship.is_empty:
+        if isinstance(relationship_changelog, RelationshipCardinalityOneChangelog) and relationship_changelog.parent:
+            self.add_parent(parent=relationship_changelog.parent)
+        if relationship_changelog.is_empty:
             return
 
-        self.relationships[relationship.name] = relationship
+        self.relationships[relationship_changelog.name] = relationship_changelog
 
     def create_attribute(self, attribute: BaseAttribute) -> None:
         changelog_attribute = AttributeChangelog(
@@ -383,7 +383,7 @@ def remove_peer(self, peer_data: RelationshipPeerData) -> None:
     def _set_cardinality_one_peer(self, relationship: Relationship) -> None:
         self.cardinality_one_relationship.peer_id = relationship.peer_id
         self.cardinality_one_relationship.peer_kind = relationship.get_peer_kind()
-        self.cardinality_one_relationship.set_parent_from_relationship(relationship=relationship)
+        self.cardinality_one_relationship.set_parent_from_relationship(rel_kind=relationship.schema.kind)
 
     def add_parent_from_relationship(self, relationship: Relationship) -> None:
         if self.schema.cardinality == RelationshipCardinality.ONE:
@@ -420,18 +420,16 @@ def add_updated_relationship(
                     value_current=getattr(relationship, property_name),
                     value_previous=previous_value,
                 )
-            self.cardinality_one_relationship.set_parent_from_relationship(relationship=relationship)
+            self.cardinality_one_relationship.set_parent_from_relationship(rel_kind=relationship.schema.kind)
 
-    def delete_relationship(self, relationship: Relationship) -> None:
+    def delete_relationship(self, peer_id: str, peer_kind: str, rel_schema: RelationshipSchema) -> None:
         if self.schema.cardinality == RelationshipCardinality.ONE:
-            self.cardinality_one_relationship.peer_id_previous = relationship.get_peer_id()
-            self.cardinality_one_relationship.peer_kind_previous = relationship.get_peer_kind()
-            self.cardinality_one_relationship.set_parent_from_relationship(relationship=relationship)
+            self.cardinality_one_relationship.peer_id_previous = peer_id
+            self.cardinality_one_relationship.peer_kind_previous = peer_kind
+            self.cardinality_one_relationship.set_parent_from_relationship(rel_kind=rel_schema.kind)
 
         elif self.schema.cardinality == RelationshipCardinality.MANY:
-            self.cardinality_many_relationship.remove_peer(
-                peer_id=relationship.get_peer_id(), peer_kind=relationship.get_peer_kind()
-            )
+            self.cardinality_many_relationship.remove_peer(peer_id=peer_id, peer_kind=peer_kind)
 
     @property
     def changelog(self) -> RelationshipCardinalityOneChangelog | RelationshipCardinalityManyChangelog:

backend/infrahub/core/constants/schema.py

@@ -1,5 +1,7 @@
 from enum import Enum, Flag, auto
 
+PARENT_CHILD_IDENTIFIER = "parent__child"
+
 
 class FlagProperty(Enum):
     IS_VISIBLE = "is_visible"

backend/infrahub/core/diff/model/path.py

@@ -163,6 +163,10 @@ class EnrichedDiffAttribute(BaseSummary):
     def __hash__(self) -> int:
         return hash(self.name)
 
+    @property
+    def num_properties(self) -> int:
+        return len(self.properties)
+
     def get_all_conflicts(self) -> dict[str, EnrichedDiffConflict]:
         return {prop.path_identifier: prop.conflict for prop in self.properties if prop.conflict}
 
@@ -202,6 +206,10 @@ class EnrichedDiffSingleRelationship(BaseSummary):
     def __hash__(self) -> int:
         return hash(self.peer_id)
 
+    @property
+    def num_properties(self) -> int:
+        return len(self.properties)
+
     def get_all_conflicts(self) -> dict[str, EnrichedDiffConflict]:
         all_conflicts: dict[str, EnrichedDiffConflict] = {}
         if self.conflict:
@@ -248,6 +256,10 @@ class EnrichedDiffRelationship(BaseSummary):
     def __hash__(self) -> int:
         return hash(self.name)
 
+    @property
+    def num_properties(self) -> int:
+        return sum(r.num_properties for r in self.relationships)
+
     def get_all_conflicts(self) -> dict[str, EnrichedDiffConflict]:
         all_conflicts: dict[str, EnrichedDiffConflict] = {}
         for element in self.relationships:
@@ -308,6 +320,10 @@ class EnrichedDiffNode(BaseSummary):
     def __hash__(self) -> int:
         return hash(self.uuid)
 
+    @property
+    def num_properties(self) -> int:
+        return sum(a.num_properties for a in self.attributes) + sum(r.num_properties for r in self.relationships)
+
     def get_all_conflicts(self) -> dict[str, EnrichedDiffConflict]:
         all_conflicts: dict[str, EnrichedDiffConflict] = {}
         if self.conflict:

backend/infrahub/core/diff/query/save.py

@@ -83,33 +83,55 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:  # noqa
         query = """
 UNWIND $node_details_list AS node_details
 WITH node_details.root_uuid AS root_uuid, node_details.node_map AS node_map
+MATCH (diff_root:DiffRoot {uuid: root_uuid})
+MERGE (diff_root)-[:DIFF_HAS_NODE]->(diff_node:DiffNode {uuid: node_map.node_properties.uuid})
+WITH root_uuid, node_map, diff_node, (node_map.conflict_params IS NOT NULL) AS has_node_conflict
+SET
+    diff_node.kind = node_map.node_properties.kind,
+    diff_node.label = node_map.node_properties.label,
+    diff_node.changed_at = node_map.node_properties.changed_at,
+    diff_node.action = node_map.node_properties.action,
+    diff_node.path_identifier = node_map.node_properties.path_identifier
+WITH root_uuid, node_map, diff_node, has_node_conflict
 CALL {
-    WITH root_uuid, node_map
-    MATCH (diff_root {uuid: root_uuid})
-    MERGE (diff_root)-[:DIFF_HAS_NODE]->(diff_node:DiffNode {uuid: node_map.node_properties.uuid})
-    SET
-        diff_node.kind = node_map.node_properties.kind,
-        diff_node.label = node_map.node_properties.label,
-        diff_node.changed_at = node_map.node_properties.changed_at,
-        diff_node.action = node_map.node_properties.action,
-        diff_node.path_identifier = node_map.node_properties.path_identifier
     // -------------------------
-    // add/remove node-level conflict
+    // delete parent-child relationships for included nodes, they will be added in EnrichedNodesLinkQuery
     // -------------------------
-    WITH diff_node, node_map
-    OPTIONAL MATCH (diff_node)-[:DIFF_HAS_CONFLICT]->(current_diff_node_conflict:DiffConflict)
-    WITH diff_node, node_map, current_diff_node_conflict, (node_map.conflict_params IS NOT NULL) AS has_node_conflict
-    FOREACH (i in CASE WHEN has_node_conflict = FALSE THEN [1] ELSE [] END |
-        DETACH DELETE current_diff_node_conflict
-    )
-    FOREACH (i in CASE WHEN has_node_conflict = TRUE THEN [1] ELSE [] END |
-        MERGE (diff_node)-[:DIFF_HAS_CONFLICT]->(diff_node_conflict:DiffConflict)
-        SET diff_node_conflict = node_map.conflict_params
-    )
+    WITH diff_node
+    MATCH (diff_node)-[:DIFF_HAS_RELATIONSHIP]->(:DiffRelationship)-[parent_rel:DIFF_HAS_NODE]->(:DiffNode)
+    DELETE parent_rel
+}
+OPTIONAL MATCH (diff_node)-[:DIFF_HAS_CONFLICT]->(current_node_conflict:DiffConflict)
+CALL {
+    // -------------------------
+    // create a node-level conflict, if necessary
+    // -------------------------
+    WITH diff_node, current_node_conflict, has_node_conflict
+    WITH diff_node, current_node_conflict, has_node_conflict
+    WHERE current_node_conflict IS NULL AND has_node_conflict = TRUE
+    CREATE (diff_node)-[:DIFF_HAS_CONFLICT]->(:DiffConflict)
+}
+CALL {
+    // -------------------------
+    // delete a node-level conflict, if necessary
+    // -------------------------
+    WITH current_node_conflict, has_node_conflict
+    WITH current_node_conflict, has_node_conflict
+    WHERE current_node_conflict IS NOT NULL AND has_node_conflict = FALSE
+    DETACH DELETE current_node_conflict
+}
+WITH root_uuid, node_map, diff_node, has_node_conflict, node_map.conflict_params AS node_conflict_params
+CALL {
+    // -------------------------
+    // set the properties of the node-level conflict, if necessary
+    // -------------------------
+    WITH diff_node, has_node_conflict, node_conflict_params
+    WITH diff_node, has_node_conflict, node_conflict_params
+    WHERE has_node_conflict = TRUE
+    OPTIONAL MATCH (diff_node)-[:DIFF_HAS_CONFLICT]->(node_conflict:DiffConflict)
+    SET node_conflict = node_conflict_params
 }
 CALL {
-    WITH root_uuid, node_map
-    MATCH (diff_root {uuid: root_uuid})-[:DIFF_HAS_NODE]->(diff_node:DiffNode {uuid: node_map.node_properties.uuid})
     // -------------------------
     // remove stale attributes for this node
     // -------------------------
@@ -426,14 +448,14 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:  # noqa
     node_link_details.root_uuid AS root_uuid,
     node_link_details.parent_uuid AS parent_uuid,
     node_link_details.child_uuid AS child_uuid,
-    node_link_details.relationship_name AS relationship_name
+    node_link_details.child_relationship_name AS relationship_name
 CALL {
     WITH root_uuid, parent_uuid, child_uuid, relationship_name
     MATCH (diff_root {uuid: root_uuid})
-    MATCH (diff_root)-[:DIFF_HAS_NODE]->(parent_node:DiffNode {uuid: parent_uuid})
-        -[:DIFF_HAS_RELATIONSHIP]->(diff_rel_group:DiffRelationship {name: relationship_name})
     MATCH (diff_root)-[:DIFF_HAS_NODE]->(child_node:DiffNode {uuid: child_uuid})
-    MERGE (diff_rel_group)-[:DIFF_HAS_NODE]->(child_node)
+        -[:DIFF_HAS_RELATIONSHIP]->(diff_rel_group:DiffRelationship {name: relationship_name})
+    MATCH (diff_root)-[:DIFF_HAS_NODE]->(parent_node:DiffNode {uuid: parent_uuid})
+    MERGE (diff_rel_group)-[:DIFF_HAS_NODE]->(parent_node)
 }
         """
         self.add_to_query(query)
@@ -443,14 +465,14 @@ def _build_node_parent_links(self, enriched_node: EnrichedDiffNode, root_uuid: s
             return []
         parent_links = []
         for relationship in enriched_node.relationships:
-            for child_node in relationship.nodes:
+            for parent_node in relationship.nodes:
                 parent_links.append(
                     {
-                        "parent_uuid": enriched_node.uuid,
-                        "relationship_name": relationship.name,
-                        "child_uuid": child_node.uuid,
+                        "child_uuid": enriched_node.uuid,
+                        "child_relationship_name": relationship.name,
+                        "parent_uuid": parent_node.uuid,
                         "root_uuid": root_uuid,
                     }
                 )
-                parent_links.extend(self._build_node_parent_links(enriched_node=child_node, root_uuid=root_uuid))
+                parent_links.extend(self._build_node_parent_links(enriched_node=parent_node, root_uuid=root_uuid))
         return parent_links

backend/infrahub/core/diff/repository/repository.py

@@ -1,6 +1,8 @@
 from collections import defaultdict
 from typing import AsyncGenerator, Generator, Iterable
 
+from neo4j.exceptions import TransientError
+
 from infrahub import config
 from infrahub.core import registry
 from infrahub.core.diff.query.field_summary import EnrichedDiffNodeFieldSummaryQuery
@@ -42,11 +44,10 @@
 
 
 class DiffRepository:
-    MAX_SAVE_BATCH_SIZE: int = 100
-
-    def __init__(self, db: InfrahubDatabase, deserializer: EnrichedDiffDeserializer):
+    def __init__(self, db: InfrahubDatabase, deserializer: EnrichedDiffDeserializer, max_save_batch_size: int = 1000):
         self.db = db
         self.deserializer = deserializer
+        self.max_save_batch_size = max_save_batch_size
 
     async def _run_get_diff_query(
         self,
@@ -230,20 +231,44 @@ def _get_node_create_request_batch(
     ) -> Generator[list[EnrichedNodeCreateRequest], None, None]:
         node_requests = []
         for diff_root in (enriched_diffs.base_branch_diff, enriched_diffs.diff_branch_diff):
+            size_count = 0
             for node in diff_root.nodes:
-                node_requests.append(EnrichedNodeCreateRequest(node=node, root_uuid=diff_root.uuid))
-                if len(node_requests) == self.MAX_SAVE_BATCH_SIZE:
+                node_size_count = node.num_properties
+                if size_count + node_size_count < self.max_save_batch_size:
+                    node_requests.append(EnrichedNodeCreateRequest(node=node, root_uuid=diff_root.uuid))
+                    size_count += node_size_count
+                else:
+                    log.info(f"Num nodes in batch: {len(node_requests)}, num properties in batch: {size_count}")
                     yield node_requests
-                    node_requests = []
+                    size_count = node_size_count
+                    node_requests = [EnrichedNodeCreateRequest(node=node, root_uuid=diff_root.uuid)]
         if node_requests:
+            log.info(f"Num nodes in batch: {len(node_requests)}, num properties in batch: {size_count}")
             yield node_requests
 
-    @retry_db_transaction(name="enriched_diff_save")
-    async def save(self, enriched_diffs: EnrichedDiffs | EnrichedDiffsMetadata, do_summary_counts: bool = True) -> None:
+    @retry_db_transaction(name="enriched_diff_metadata_save")
+    async def _save_root_metadata(self, enriched_diffs: EnrichedDiffsMetadata) -> None:
         log.info("Updating diff metadata...")
         root_query = await EnrichedDiffRootsUpsertQuery.init(db=self.db, enriched_diffs=enriched_diffs)
         await root_query.execute(db=self.db)
         log.info("Diff metadata updated.")
+
+    async def _save_node_batch(self, node_create_batch: list[EnrichedNodeCreateRequest]) -> None:
+        node_query = await EnrichedNodeBatchCreateQuery.init(db=self.db, node_create_batch=node_create_batch)
+        try:
+            await node_query.execute(db=self.db)
+        except TransientError as exc:
+            if not exc.code or "OutOfMemoryError".lower() not in str(exc.code).lower():
+                raise
+            log.exception("Database memory error during save. Trying smaller transactions")
+            for node_request in node_create_batch:
+                single_node_query = await EnrichedNodeBatchCreateQuery.init(
+                    db=self.db, node_create_batch=[node_request]
+                )
+                await single_node_query.execute(db=self.db)
+
+    async def save(self, enriched_diffs: EnrichedDiffs | EnrichedDiffsMetadata, do_summary_counts: bool = True) -> None:
+        await self._save_root_metadata(enriched_diffs=enriched_diffs)
         if not isinstance(enriched_diffs, EnrichedDiffs):
             return
         num_nodes = len(enriched_diffs.base_branch_diff.nodes) + len(enriched_diffs.diff_branch_diff.nodes)
@@ -252,9 +277,8 @@ async def save(self, enriched_diffs: EnrichedDiffs | EnrichedDiffsMetadata, do_s
             self._get_node_create_request_batch(enriched_diffs=enriched_diffs)
         ):
             log.info(f"Saving node batch #{batch_num}...")
-            node_query = await EnrichedNodeBatchCreateQuery.init(db=self.db, node_create_batch=node_create_batch)
-            await node_query.execute(db=self.db)
-            log.info(f"Batch #{batch_num} saved")
+            await self._save_node_batch(node_create_batch=node_create_batch)
+            log.info(f"Batch saved. num_nodes={len(node_create_batch)}")
         link_query = await EnrichedNodesLinkQuery.init(db=self.db, enriched_diffs=enriched_diffs)
         await link_query.execute(db=self.db)
         log.info("Diff saved.")
@@ -444,6 +468,7 @@ async def get_node_field_specifiers(self, diff_id: str) -> dict[str, set[str]]:
             offset += limit
         return specifiers
 
+    @retry_db_transaction(name="enriched_diff_summary_counts")
     async def add_summary_counts(
         self,
         diff_branch_name: str,
@@ -460,4 +485,4 @@ async def add_summary_counts(
             node_uuids=node_uuids,
         )
         await query.execute(db=self.db)
-        log.info("Summary counts updated...")
+        log.info("Summary counts updated.")