Merge branch 'stable' into jbr-conflits-02272025-stable-to-release12

Author: wartraxx51

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.")

backend/infrahub/core/node/__init__.py

@@ -7,15 +7,23 @@
 from infrahub_sdk.uuidt import UUIDT
 
 from infrahub.core import registry
+<<<<<<< HEAD
 from infrahub.core.changelog.models import NodeChangelog
 from infrahub.core.constants import (
     OBJECT_TEMPLATE_NAME_ATTR,
     OBJECT_TEMPLATE_RELATIONSHIP_NAME,
+=======
+from infrahub.core.constants import (
+    GLOBAL_BRANCH_NAME,
+>>>>>>> stable
     BranchSupportType,
     ComputedAttributeKind,
     InfrahubKind,
     RelationshipCardinality,
+<<<<<<< HEAD
     RelationshipKind,
+=======
+>>>>>>> stable
 )
 from infrahub.core.constants.schema import SchemaElementPathType
 from infrahub.core.protocols import CoreNumberPool, CoreObjectTemplate
@@ -28,6 +36,7 @@
 
 from ...graphql.constants import KIND_GRAPHQL_FIELD_NAME
 from ...graphql.models import OrderModel
+from ..query.relationship import RelationshipDeleteAllQuery
 from ..relationship import RelationshipManager
 from ..utils import update_relationships_to
 from .base import BaseNode, BaseNodeMeta, BaseNodeOptions
@@ -690,6 +699,7 @@ async def delete(self, db: InfrahubDatabase, at: Optional[Timestamp] = None) ->
             if deleted_attribute:
                 node_changelog.add_attribute(attribute=deleted_attribute)
 
+<<<<<<< HEAD
         # Go over the list of relationships and update them one by one
         for name in self._relationships:
             rel: RelationshipManager = getattr(self, name)
@@ -698,8 +708,15 @@ async def delete(self, db: InfrahubDatabase, at: Optional[Timestamp] = None) ->
 
         # Need to check if there are some unidirectional relationship as well
         # For example, if we delete a tag, we must check the permissions and update all the relationships pointing at it
+=======
+>>>>>>> stable
         branch = self.get_branch_based_on_support_type()
 
+        delete_query = await RelationshipDeleteAllQuery.init(
+            db=db, node_id=self.get_id(), branch=branch, at=delete_at, branch_agnostic=branch.name == GLOBAL_BRANCH_NAME
+        )
+        await delete_query.execute(db=db)
+
         # Update the relationship to the branch itself
         query = await NodeGetListQuery.init(
             db=db,