From ae43c2ed9ce581ce5b59ece170307e1c7d8973da Mon Sep 17 00:00:00 2001 From: Aaron McCarty Date: Tue, 18 Aug 2026 13:43:32 -0700 Subject: [PATCH 1/6] fix(backend): retire branch-agnostic field edges on object deletion (#9762) A branch-agnostic attribute or relationship keeps its value on edges carrying the global branch name. Deleting the owning object tombstoned the object but left those edges open and active, so the value stayed allocated with no owner until uniqueness validation failed on ids that no longer resolve. This is the first enforcement point: single object deletion. - agnostic_retention.py holds the retention judgement as one shared Cypher fragment, so the enforcement points still to come cannot each re-derive it slightly differently. A branch retains a field only where the same linked vertex is live and reaches the field over a live edge, under that branch's own view with isolation applied; retention across branches is a disjunction taken after that per-branch conjunction. A relationship needs two live peers, counted by uuid so that the copies a kind or inheritance change leaves behind count once between them. The branch windows are derived from (:Branch) inside the query rather than marshalled through Python, which removes a paginated branch read whose default limit would quietly turn the branches past it into branches that retain nothing. - node_agnostic_retirement.py anchors on one node's open, active global owning edges and time-closes every open, active global edge of each field vertex no branch retains. Never a deleted-status edge and never a vertex; IS_RESERVED is out of reach because it is never incident to an attribute or relationship vertex, and closing HAS_VALUE and HAS_ATTRIBUTE is what frees a pooled value. - Node.delete invokes it after the existence tombstone, inside the caller's still-open transaction, and a failure propagates. Raising rolls the tombstone back and the caller retries; swallowing would commit a deleted object still holding a live branch-agnostic value, which no user or operator action repairs. Still to come, each as its own slice: branch merge, branch rebase, branch deletion, the two schema-removal paths, and the repair migration for the backlog of orphans already in customer databases. Co-Authored-By: Claude Opus 5 (1M context) --- backend/infrahub/core/node/__init__.py | 13 + .../infrahub/core/query/agnostic_retention.py | 112 +++++ .../core/query/node_agnostic_retirement.py | 73 ++++ .../core/test_agnostic_retirement.py | 357 ++++++++++++++++ backend/tests/component/query/__init__.py | 0 .../test_node_agnostic_retirement_query.py | 401 ++++++++++++++++++ backend/tests/helpers/agnostic_edges.py | 266 ++++++++++++ .../helpers/schema/agnostic_retirement.py | 46 ++ 8 files changed, 1268 insertions(+) create mode 100644 backend/infrahub/core/query/agnostic_retention.py create mode 100644 backend/infrahub/core/query/node_agnostic_retirement.py create mode 100644 backend/tests/component/core/test_agnostic_retirement.py create mode 100644 backend/tests/component/query/__init__.py create mode 100644 backend/tests/component/query/test_node_agnostic_retirement_query.py create mode 100644 backend/tests/helpers/agnostic_edges.py create mode 100644 backend/tests/helpers/schema/agnostic_retirement.py diff --git a/backend/infrahub/core/node/__init__.py b/backend/infrahub/core/node/__init__.py index d36677addfb..d479e8d0c33 100644 --- a/backend/infrahub/core/node/__init__.py +++ b/backend/infrahub/core/node/__init__.py @@ -25,6 +25,7 @@ from infrahub.core.protocols import CoreNumberPool, CoreObjectTemplate from infrahub.core.protocols_base import CoreNode from infrahub.core.query.node import NodeCheckIDQuery, NodeCreateAllQuery, NodeDeleteQuery, NodeUpdateMetadataQuery +from infrahub.core.query.node_agnostic_retirement import RetireNodeAgnosticFieldsQuery from infrahub.core.schema import ( AttributeSchema, NodeSchema, @@ -1250,6 +1251,18 @@ async def delete(self, db: InfrahubDatabase, user_id: str = SYSTEM_USER_ID, at: query = await NodeDeleteQuery.init(db=db, node=self, at=delete_at, user_id=user_id) await query.execute(db=db) + retirement_query = await RetireNodeAgnosticFieldsQuery.init(db=db, node_uuid=self.get_id(), at=delete_at) + await retirement_query.execute(db=db) + retired = retirement_query.get_data() + if retired.edges_closed: + log.debug( + "Retired branch-agnostic fields of a deleted node", + node_id=self.get_id(), + node_kind=self.get_kind(), + edges_closed=retired.edges_closed, + at=delete_at.to_string(), + ) + self._node_changelog = node_changelog async def to_graphql( diff --git a/backend/infrahub/core/query/agnostic_retention.py b/backend/infrahub/core/query/agnostic_retention.py new file mode 100644 index 00000000000..735ee28f0ea --- /dev/null +++ b/backend/infrahub/core/query/agnostic_retention.py @@ -0,0 +1,112 @@ +"""The one predicate that decides whether a branch-agnostic field is still retained by any branch. + +A branch-agnostic field keeps its value on edges carrying the global branch name, which every branch +reads. Those edges may only be closed once **NO** branch can still reach a live owner over a live field +edge. Cypher for this is consolidated here to ensure consistency. + +Retention is decided per branch **and** per linked vertex: under that branch's view, the vertex's +existence edge and its edge to the field must both resolve to `active`. Surviving edges are summed +per branch, and only then is the maximum taken across branches. Retention is a disjunction of what +each branch holds live on its own, never a pool the branches contribute to jointly. + +A `:Relationship` needs two qualifying field edges rather than one, because a relationship missing a +peer is not a relationship. A `:Relationship` must also have two distinct active peers to be +considered active. + +Assumption: every branch forks from the default branch. A branch-of-branch feature would not extend +this logic. +""" + +# Expects `field` in scope, one row per candidate vertex, plus the `$global_branch_name` and `$at` +# parameters. Emits the candidates no branch retains, with `field` as the only variable in scope. +UNRETAINED_AGNOSTIC_FIELD_PREDICATE = """ +WITH collect(field) AS candidates + +// ---------------------- +// The branches are read once for the whole run and carried as a list. +// ---------------------- +MATCH (branch:Branch) +WHERE branch.name <> $global_branch_name +WITH + candidates, + collect({ + name: branch.name, + origin_name: CASE WHEN branch.is_default THEN NULL ELSE branch.origin_branch END, + origin_at: CASE + WHEN branch.is_default THEN NULL + WHEN branch.branched_from < $at THEN branch.branched_from + ELSE $at + END + }) AS branch_windows + +UNWIND candidates AS field +WITH field, branch_windows, CASE WHEN field:Relationship THEN 2 ELSE 1 END AS required_live_peers + +CALL (field, branch_windows) { + UNWIND branch_windows AS branch_window + WITH + field, + branch_window.name AS branch_name, + branch_window.origin_name AS origin_name, + branch_window.origin_at AS origin_at + + // ---------------------- + // Count this `field`'s active links to :Node vertices on this branch + // ---------------------- + MATCH (node:Node)-[field_edge:HAS_ATTRIBUTE|IS_RELATED]-(field) + WHERE (field_edge.branch IN [$global_branch_name, branch_name] + AND field_edge.from <= $at + AND (field_edge.to IS NULL OR field_edge.to > $at)) + OR (field_edge.branch = origin_name + AND field_edge.from <= origin_at + AND (field_edge.to IS NULL OR field_edge.to > origin_at)) + WITH + branch_name, + origin_name, + origin_at, + node, + field_edge.branch_level AS field_edge_level, + field_edge.from AS field_edge_from, + field_edge.status AS field_edge_status + ORDER BY field_edge_level DESC, field_edge_from DESC, field_edge_status ASC + WITH + branch_name, + origin_name, + origin_at, + node, + collect(field_edge_status)[0] AS winning_field_edge_status + WHERE winning_field_edge_status = "active" + + // ---------------------- + // Check that each linked :Node vertex is active on this branch. + // The global branch stays in the existence match so that an owner which is itself branch-agnostic + // reads as live on every branch and is therefore retained. + // ---------------------- + MATCH (node)-[existence:IS_PART_OF]->(:Root) + WHERE (existence.branch IN [$global_branch_name, branch_name] + AND existence.from <= $at + AND (existence.to IS NULL OR existence.to > $at)) + OR (existence.branch = origin_name + AND existence.from <= origin_at + AND (existence.to IS NULL OR existence.to > origin_at)) + WITH + branch_name, + node, + existence.branch_level AS existence_level, + existence.from AS existence_from, + existence.status AS existence_status + ORDER BY existence_level DESC, existence_from DESC, existence_status ASC + WITH branch_name, node, collect(existence_status)[0] AS winning_existence + WHERE winning_existence = "active" + + // ---------------------- + // Peers are counted by uuid. Kind/inheritance migration leaves multiple Node vertices with + // the same uuid for a single entity + // ---------------------- + WITH branch_name, count(DISTINCT node.uuid) AS live_peer_count + RETURN max(live_peer_count) AS most_live_peers +} + +WITH field, required_live_peers, coalesce(most_live_peers, 0) AS live_peers +WHERE live_peers < required_live_peers +""" diff --git a/backend/infrahub/core/query/node_agnostic_retirement.py b/backend/infrahub/core/query/node_agnostic_retirement.py new file mode 100644 index 00000000000..09e7a08c38f --- /dev/null +++ b/backend/infrahub/core/query/node_agnostic_retirement.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from infrahub.core.constants import GLOBAL_BRANCH_NAME +from infrahub.core.query import Query, QueryType +from infrahub.core.query.agnostic_retention import UNRETAINED_AGNOSTIC_FIELD_PREDICATE + +if TYPE_CHECKING: + from infrahub.core.timestamp import Timestamp + from infrahub.database import InfrahubDatabase + + +@dataclass(frozen=True) +class NodeAgnosticRetirementResult: + """What retiring one node's branch-agnostic fields changed.""" + + edges_closed: int + """Global edges given a `to` timestamp. Zero means every field is still retained somewhere.""" + + +_RETIRE_UNRETAINED_FIELDS_OF_NODE = """ +// ----------------- +// MATCH on the branch-agnostic edges we care about to start with. +// ----------------- +MATCH (anchor_node:Node {uuid: $node_uuid})-[anchor:HAS_ATTRIBUTE|IS_RELATED]-(field:Attribute|Relationship) +WHERE anchor.branch = $global_branch_name + AND anchor.status = "active" + AND anchor.to IS NULL +WITH DISTINCT field +%(unretained_predicate)s + +MATCH (field)-[e]-() +WHERE e.branch = $global_branch_name + AND e.status = "active" + AND e.to IS NULL +SET e.to = $at +RETURN count(e) AS edges_closed +""" % {"unretained_predicate": UNRETAINED_AGNOSTIC_FIELD_PREDICATE} + + +class RetireNodeAgnosticFieldsQuery(Query): + """Close the open global edges of one node's branch-agnostic fields that no branch retains. + + Checks if the field is reachable from ANY branch. It is only deleted if it is completely + unreachable. + """ + + name: str = "retire_node_agnostic_fields" + type: QueryType = QueryType.WRITE + + insert_return: bool = False + insert_limit: bool = False + + def __init__(self, node_uuid: str, at: Timestamp, **kwargs: Any) -> None: + self.node_uuid = node_uuid + super().__init__(at=at, **kwargs) + + async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None: # noqa: ARG002 + self.params["global_branch_name"] = GLOBAL_BRANCH_NAME + self.params["node_uuid"] = self.node_uuid + self.params["at"] = self.at.to_string() + + self.add_to_query(_RETIRE_UNRETAINED_FIELDS_OF_NODE) + self.update_return_labels(["edges_closed"]) + + def get_data(self) -> NodeAgnosticRetirementResult: + """Return what the run closed.""" + result = self.get_result() + if result: + return NodeAgnosticRetirementResult(edges_closed=result.get_as_type("edges_closed", int)) + return NodeAgnosticRetirementResult(edges_closed=0) diff --git a/backend/tests/component/core/test_agnostic_retirement.py b/backend/tests/component/core/test_agnostic_retirement.py new file mode 100644 index 00000000000..9a1de8dfb06 --- /dev/null +++ b/backend/tests/component/core/test_agnostic_retirement.py @@ -0,0 +1,357 @@ +"""Test for deleting branch-aware objects with branch-agnostic attributes and relationships. + +Branch-agnostic fields on branch-aware objects should only be deleted when the cannot be accessed +from any branch. + +Every assertion reads the edges directly rather than going through the node manager: the subject is +which edges carry a `to` timestamp and which do not, and a read through the manager would hide the +very states these tests exist to pin down. Where a branch is expected to go on reading the object, the +manager is used as well, because that is the claim being made. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest + +from infrahub.core import registry +from infrahub.core.constants import GLOBAL_BRANCH_NAME +from infrahub.core.initialization import create_branch +from infrahub.core.manager import NodeManager +from infrahub.core.node import Node +from infrahub.core.query.node_agnostic_retirement import RetireNodeAgnosticFieldsQuery +from infrahub.core.timestamp import Timestamp +from infrahub.database import InfrahubDatabase, InfrahubDatabaseMode + +if TYPE_CHECKING: + from neo4j import Record + + from infrahub.core.branch import Branch + from infrahub.core.query import QueryType + +from tests.helpers.agnostic_edges import ( + attribute_global_edges, + attribute_owning_edges, + edge_summary, + existence_edges, + expected_closed_at, + open_edge_types, + open_edges, + relationship_global_edges, + remove_attribute_on_branch, +) +from tests.helpers.schema.agnostic_retirement import ( + AGNOSTIC_RETIREMENT_SCHEMA, + GADGET_KIND, + RELATIONSHIP_IDENTIFIER, + WIDGET_KIND, +) + + +class RetirementFailureError(Exception): + """Stands in for whatever the retirement run could fail with.""" + + +class FailingRetirementDatabase(InfrahubDatabase): + """Database that fails the branch-agnostic retirement query and passes every other query through. + + A real database is what makes the claim testable: the deletion's own writes have to reach the + transaction so that the rollback has something to undo. + """ + + @classmethod + def from_db(cls, db: InfrahubDatabase) -> FailingRetirementDatabase: + return cls( + mode=InfrahubDatabaseMode.DRIVER, + driver=db._driver, + db_type=db.db_type, + default_neo4j_runtime=db.default_neo4j_runtime, + queries_names_to_config=db.queries_names_to_config, + ) + + async def execute_query_with_metadata( + self, + query: str, + params: dict[str, Any] | None = None, + name: str = "undefined", + context: dict[str, str] | None = None, + type: QueryType | None = None, + timeout_seconds: float | None = None, + ) -> tuple[list[Record], dict[str, Any]]: + if name == RetireNodeAgnosticFieldsQuery.name: + raise RetirementFailureError("the retirement run could not complete") + return await super().execute_query_with_metadata( + query=query, params=params, name=name, context=context, type=type, timeout_seconds=timeout_seconds + ) + + +async def _create_widget(db: InfrahubDatabase, branch: Branch, name: str, serial: int, **kwargs: Any) -> Node: + widget = await Node.init(db=db, schema=WIDGET_KIND, branch=branch) + await widget.new(db=db, name=name, serial=serial, **kwargs) + await widget.save(db=db) + return widget + + +async def _delete(db: InfrahubDatabase, node_id: str, branch: Branch, at: Timestamp) -> None: + to_delete = await NodeManager.get_one(db=db, id=node_id, branch=branch, raise_on_error=True) + await to_delete.delete(db=db, at=at) + + +class TestAgnosticRetirementOnDelete: + @pytest.fixture(scope="class") + async def default_branch(self, default_branch_scope_class: Branch) -> Branch: + return default_branch_scope_class + + @pytest.fixture(scope="class") + async def agnostic_schema(self, db: InfrahubDatabase, default_branch: Branch) -> None: + registry.schema.register_schema(schema=AGNOSTIC_RETIREMENT_SCHEMA, branch=default_branch.name) + + async def test_a_field_created_and_deleted_on_the_same_user_branch_is_closed_by_the_delete( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """An object that only ever existed on one branch cannot also exist elsewhere through a merge. + + A branch that has been merged is permanently read-only, so a delete running on a branch proves the + branch was never merged. Nothing else can be holding the field, so the close is unconditional. + """ + branch = await create_branch(db=db, branch_name="creates-and-deletes") + widget = await _create_widget(db=db, branch=branch, name="branch-only", serial=100) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + deleted_at = Timestamp() + await _delete(db=db, node_id=widget.id, branch=branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert {edge.status for edge in after} == {"active"}, "retirement is a time-close, never a status tombstone" + + async def test_a_field_stays_open_while_the_default_branch_still_holds_the_object( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Deleting on a branch says nothing about the object's fate on the branch it forked from.""" + widget = await _create_widget(db=db, branch=default_branch, name="deleted-on-a-branch-only", serial=200) + branch = await create_branch(db=db, branch_name="deletes-its-own-copy") + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + + await _delete(db=db, node_id=widget.id, branch=branch, at=Timestamp()) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ) + still_on_default = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) + assert still_on_default is not None + assert still_on_default.get_attribute(name="serial").value == 200 + + async def test_a_field_is_closed_when_the_default_branch_deletes_the_last_holder( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """No branch forked between the creation and the deletion, so no branch can still read the object.""" + widget = await _create_widget(db=db, branch=default_branch, name="last-holder", serial=300) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + deleted_at = Timestamp() + await _delete(db=db, node_id=widget.id, branch=default_branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert {edge.status for edge in after} == {"active"}, "retirement is a time-close, never a status tombstone" + + async def test_a_field_stays_open_for_a_branch_that_forked_between_creation_and_deletion( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """A branch that forked while the object was live reads it as live, and keeps reading its value.""" + widget = await _create_widget(db=db, branch=default_branch, name="retained-by-a-fork", serial=400) + branch = await create_branch(db=db, branch_name="forked-before-the-delete") + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + await _delete(db=db, node_id=widget.id, branch=default_branch, at=Timestamp()) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ) + assert await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) is None + on_branch = await NodeManager.get_one(db=db, id=widget.id, branch=branch) + assert on_branch is not None + assert on_branch.get_attribute(name="serial").value == 400 + + async def test_a_field_stays_open_until_every_retaining_branch_has_deleted_the_object( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Verify that an object's agnostic fields are only deleted when NO branch can reach them""" + widget = await _create_widget(db=db, branch=default_branch, name="held-by-two", serial=800) + first = await create_branch(db=db, branch_name="first-of-two-holders") + second = await create_branch(db=db, branch_name="second-of-two-holders") + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + await _delete(db=db, node_id=widget.id, branch=default_branch, at=Timestamp()) + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ), "two branches still read the object, so nothing is released" + + await _delete(db=db, node_id=widget.id, branch=first, at=Timestamp()) + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ), "the second branch alone is still enough to retain it" + assert await NodeManager.get_one(db=db, id=widget.id, branch=second) is not None + + last_delete = Timestamp() + await _delete(db=db, node_id=widget.id, branch=second, at=last_delete) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edges(after) == [], "the last holder released it" + assert {edge.to_time for edge in after} == {last_delete.to_string()} + + async def test_a_relationship_is_closed_when_its_peers_are_live_on_different_branches( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Both peers must be live under one branch's view, not one peer each on two branches. + + After deleting each peer on a separate branch, the object is NOT retired. Only when the + relationship is broken on _every_ branch will it be retired. + """ + gadget = await Node.init(db=db, schema=GADGET_KIND, branch=default_branch) + await gadget.new(db=db, name="peer-on-its-own-branch") + await gadget.save(db=db) + widget = await _create_widget(db=db, branch=default_branch, name="split-peers", serial=1100, gadget=gadget) + + keeps_the_gadget = await create_branch(db=db, branch_name="keeps-the-gadget") + keeps_the_widget = await create_branch(db=db, branch_name="keeps-the-widget") + + before = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert [edge.edge_type for edge in open_edges(before)].count("IS_RELATED") == 2 + + # each branch loses one half of the relationship, so neither branch holds both peers + await _delete(db=db, node_id=widget.id, branch=keeps_the_gadget, at=Timestamp()) + await _delete(db=db, node_id=gadget.id, branch=keeps_the_widget, at=Timestamp()) + + assert edge_summary( + await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + ) == edge_summary(before), "the default branch still holds both peers" + + last_delete = Timestamp() + await _delete(db=db, node_id=gadget.id, branch=default_branch, at=last_delete) + + after = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert open_edges(after) == [], ( + "no branch reads both peers as live, so the relationship is released even though each peer " + "survives somewhere" + ) + + async def test_a_field_removed_on_the_only_retaining_branch_is_closed_with_the_object( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """Test an attribute removed from the schema on a branch is retired when its object is deleted.""" + widget = await _create_widget(db=db, branch=default_branch, name="field-removed-on-the-fork", serial=500) + branch = await create_branch(db=db, branch_name="removed-the-attribute") + await remove_attribute_on_branch( + db=db, node_id=widget.id, attribute_name="serial", branch=branch, at=Timestamp() + ) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED"} + + deleted_at = Timestamp() + await _delete(db=db, node_id=widget.id, branch=default_branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert {edge.status for edge in after} == {"active"}, "retirement is a time-close, never a status tombstone" + + owning_edges = await attribute_owning_edges(db=db, node_id=widget.id, attribute_name="serial") + assert sorted((edge.branch, edge.status, edge.to_time or "") for edge in owning_edges) == [ + (GLOBAL_BRANCH_NAME, "active", deleted_at.to_string()), + (default_branch.name, "deleted", ""), + (branch.name, "deleted", ""), + ], "only the global edge is closed; the branch-scoped tombstones are left exactly as they were" + + still_on_branch = await NodeManager.get_one(db=db, id=widget.id, branch=branch) + assert still_on_branch is not None, "the branch retains the object, which is why only the field was released" + assert still_on_branch.get_attribute(name="serial").value is None, ( + "the branch that dropped the field reads no value for it, which is what made it unretained" + ) + + async def test_a_relationship_is_closed_when_one_of_its_peers_is_deleted( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """A relationship missing a peer is not a relationship, so both of its peer edges are closed.""" + gadget = await Node.init(db=db, schema=GADGET_KIND, branch=default_branch) + await gadget.new(db=db, name="doomed-peer") + await gadget.save(db=db) + widget = await _create_widget(db=db, branch=default_branch, name="surviving-peer", serial=600, gadget=gadget) + + before = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert [edge.edge_type for edge in open_edges(before)].count("IS_RELATED") == 2 + + deleted_at = Timestamp() + await _delete(db=db, node_id=gadget.id, branch=default_branch, at=deleted_at) + + after = await relationship_global_edges(db=db, node_id=widget.id, identifier=RELATIONSHIP_IDENTIFIER) + assert open_edges(after) == [] + assert {edge.status for edge in after} == {"active"} + assert {edge.to_time for edge in after} == {deleted_at.to_string()} + + async def test_a_retirement_failure_propagates_and_leaves_the_graph_untouched( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + ) -> None: + """The failure has to reach the caller, because the deletion is what rolls back with it. + + Reporting the failure as a zero commits a deleted object still holding a live branch-agnostic + value, which is the shape retirement exists to prevent and which no later action repairs. + """ + widget = await _create_widget(db=db, branch=default_branch, name="delete-rolls-back", serial=700) + + attribute_before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + existence_before = await existence_edges(db=db, node_id=widget.id) + + failing_db = FailingRetirementDatabase.from_db(db=db) + with pytest.raises(RetirementFailureError, match=r"^the retirement run could not complete$"): + async with failing_db.start_transaction() as dbt: + to_delete = await NodeManager.get_one(db=dbt, id=widget.id, branch=default_branch, raise_on_error=True) + await to_delete.delete(db=dbt, at=Timestamp()) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(attribute_before) + ) + assert sorted((edge.branch, edge.status, edge.to_time or "") for edge in existence_before) == sorted( + (edge.branch, edge.status, edge.to_time or "") for edge in await existence_edges(db=db, node_id=widget.id) + ), "the existence tombstone rolled back with the retirement that failed after it" + still_there = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) + assert still_there is not None + assert still_there.get_attribute(name="serial").value == 700 diff --git a/backend/tests/component/query/__init__.py b/backend/tests/component/query/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/backend/tests/component/query/test_node_agnostic_retirement_query.py b/backend/tests/component/query/test_node_agnostic_retirement_query.py new file mode 100644 index 00000000000..4f5851565aa --- /dev/null +++ b/backend/tests/component/query/test_node_agnostic_retirement_query.py @@ -0,0 +1,401 @@ +"""Graph-shape assertions for the retirement of one node's branch-agnostic fields. + +Every assertion reads the edges directly rather than going through the node manager: the subject is +which edges carry a `to` timestamp and which do not, and a read through the manager would hide the +very states these tests exist to pin down. +""" + +from typing import Any + +import pytest + +from infrahub.core import registry +from infrahub.core.branch import Branch +from infrahub.core.constants import ( + SchemaPathType, +) +from infrahub.core.manager import NodeManager +from infrahub.core.migrations.schema.node_kind_update import ( + NodeKindUpdateMigration, + NodeKindUpdateMigrationQuery01, +) +from infrahub.core.node import Node +from infrahub.core.path import SchemaPath +from infrahub.core.query.node_agnostic_retirement import ( + NodeAgnosticRetirementResult, + RetireNodeAgnosticFieldsQuery, +) +from infrahub.core.timestamp import Timestamp +from infrahub.database import InfrahubDatabase +from tests.helpers.agnostic_edges import ( + attribute_global_edges, + attribute_vertex_count, + edge_summary, + node_vertex_count, + open_active_edges, + open_edges, + relationship_global_edges, + relationship_peer_shape, + tombstone_existence_only, + tombstone_relationship_peer_edge, + values_reachable_over_open_edges, +) +from tests.helpers.schema.agnostic_retirement import ( + AGNOSTIC_RETIREMENT_SCHEMA, + BEACON_KIND, + GADGET_KIND, + RELATIONSHIP_IDENTIFIER, + WIDGET_KIND, +) + + +async def _rename_widget_kind(db: InfrahubDatabase, branch: Branch) -> None: + """Rename the widget kind in the graph, leaving a superseded node vertex under the same uuid. + + The copy shares the original's attribute and relationship vertices, so a field vertex is linked to + both the live copy and the superseded one. + """ + previous_schema = registry.schema.get_node_schema(name=WIDGET_KIND, branch=branch, duplicate=False) + renamed_schema = registry.schema.get_node_schema(name=WIDGET_KIND, branch=branch, duplicate=True) + renamed_schema.name = "RenamedWidget" + + migration = NodeKindUpdateMigration( + previous_node_schema=previous_schema, + new_node_schema=renamed_schema, + schema_path=SchemaPath(path_type=SchemaPathType.ATTRIBUTE, schema_kind=renamed_schema.kind, field_name="name"), + ) + query = await NodeKindUpdateMigrationQuery01.init(db=db, branch=branch, migration=migration) + await query.execute(db=db) + + +async def _retire(db: InfrahubDatabase, node_id: str, at: Timestamp) -> NodeAgnosticRetirementResult: + query = await RetireNodeAgnosticFieldsQuery.init(db=db, node_uuid=node_id, at=at) + await query.execute(db=db) + return query.get_data() + + +async def _create_widget(db: InfrahubDatabase, branch: Branch, name: str, serial: int, **kwargs: Any) -> Node: + widget = await Node.init(db=db, schema=WIDGET_KIND, branch=branch) + await widget.new(db=db, name=name, serial=serial, **kwargs) + await widget.save(db=db) + return widget + + +async def _create_gadget(db: InfrahubDatabase, branch: Branch, name: str) -> Node: + gadget = await Node.init(db=db, schema=GADGET_KIND, branch=branch) + await gadget.new(db=db, name=name) + await gadget.save(db=db) + return gadget + + +class TestRetireNodeAgnosticFields: + @pytest.fixture(scope="class") + async def default_branch(self, default_branch_scope_class: Branch) -> Branch: + return default_branch_scope_class + + @pytest.fixture(scope="class") + async def nodedel_schema(self, db: InfrahubDatabase, default_branch: Branch) -> None: + registry.schema.register_schema(schema=AGNOSTIC_RETIREMENT_SCHEMA, branch=default_branch.name) + + async def test_an_anchor_that_matches_nothing_reports_a_measured_zero( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Ensure the query returns 0 when it does nothing, and that the anchor is what limited it. + + A retirable object is present and left alone, so the zero says "this uuid owns nothing" rather + than "the database held nothing to find". The row itself matters too: the query ends in an + aggregation with no grouping key, and a missing row would leave the outcome unknown. + """ + bystander = await _create_widget(db=db, branch=default_branch, name="not-the-anchor", serial=1000) + await tombstone_existence_only(db=db, node_id=bystander.get_id(), branch=default_branch, at=Timestamp()) + bystander_before = await attribute_global_edges(db=db, node_id=bystander.get_id(), attribute_name="serial") + assert open_active_edges(bystander_before) != [], ( + "the bystander has to be retirable, or the anchor is not what spared it" + ) + + query = await RetireNodeAgnosticFieldsQuery.init(db=db, node_uuid="no-node-carries-this-uuid", at=Timestamp()) + await query.execute(db=db) + + assert query.get_result() is not None + assert query.get_data() == NodeAgnosticRetirementResult(edges_closed=0) + assert edge_summary( + await attribute_global_edges(db=db, node_id=bystander.get_id(), attribute_name="serial") + ) == edge_summary(bystander_before), "the unrelated retirable object is untouched" + + async def test_branch_agnostic_object_is_not_deleted_while_active( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify branch-agnostic object is not improperly deleted by the retirement query.""" + beacon = await Node.init(db=db, schema=BEACON_KIND, branch=default_branch) + await beacon.new(db=db, name="beacon-alive") + await beacon.save(db=db) + + before = await attribute_global_edges(db=db, node_id=beacon.get_id(), attribute_name="name") + assert open_edges(before) != [] + + assert await _retire(db=db, node_id=beacon.get_id(), at=Timestamp()) == NodeAgnosticRetirementResult( + edges_closed=0 + ) + assert edge_summary( + await attribute_global_edges(db=db, node_id=beacon.get_id(), attribute_name="name") + ) == edge_summary(before) + + async def test_an_owner_that_is_itself_branch_agnostic_is_closed_once_by_its_own_deletion( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Ensure deleting a branch-agnostic object leaves nothing for the retirement query to address.""" + beacon = await Node.init(db=db, schema=BEACON_KIND, branch=default_branch) + await beacon.new(db=db, name="beacon-deleted") + await beacon.save(db=db) + beacon_id = beacon.get_id() + + before = await attribute_global_edges(db=db, node_id=beacon_id, attribute_name="name") + open_before = open_edges(before) + assert open_before != [] + + deleted_at = Timestamp() + to_delete = await NodeManager.get_one(db=db, id=beacon_id, branch=default_branch) + await to_delete.delete(db=db, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=beacon_id, attribute_name="name") + + assert {(edge.edge_type, edge.status) for edge in open_edges(after)} == { + ("HAS_ATTRIBUTE", "deleted"), + ("HAS_VALUE", "deleted"), + ("IS_PROTECTED", "deleted"), + } + assert {edge.to_time for edge in after if edge.status == "active"} == {deleted_at.to_string()} + + assert await _retire(db=db, node_id=beacon_id, at=Timestamp()) == NodeAgnosticRetirementResult(edges_closed=0) + + async def test_partially_deleted_object_is_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Object with a closed IS_PART_OF edge and an active agnostic field is retired.""" + widget = await _create_widget(db=db, branch=default_branch, name="orphan-holder", serial=8100) + before = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + open_before = open_edges(before) + assert open_before != [] + + await tombstone_existence_only(db=db, node_id=widget.get_id(), branch=default_branch, at=Timestamp()) + + unchanged = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert edge_summary(unchanged) == edge_summary(before) + + retired_at = Timestamp() + assert await _retire(db=db, node_id=widget.get_id(), at=retired_at) == NodeAgnosticRetirementResult( + edges_closed=len(open_before) + ) + + after = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert open_edges(after) == [] + assert {edge.to_time for edge in after if edge.status == "active"} == {retired_at.to_string()} + + async def test_a_relationship_stays_open_while_both_peers_are_live_on_one_branch( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + gadget = await _create_gadget(db=db, branch=default_branch, name="live-peer") + widget = await _create_widget(db=db, branch=default_branch, name="live-owner", serial=500, gadget=gadget) + + before = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + assert [edge.edge_type for edge in open_edges(before)].count("IS_RELATED") == 2 + + at = Timestamp() + assert await _retire(db=db, node_id=widget.id, at=at) == NodeAgnosticRetirementResult(edges_closed=0) + assert await _retire(db=db, node_id=gadget.id, at=at) == NodeAgnosticRetirementResult(edges_closed=0) + + assert edge_summary( + await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + ) == (edge_summary(before)) + + async def test_a_sourced_and_owned_attribute_closes_every_property_edge_type( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify HAS_OWNER and HAS_SOURCE edges are retired correctly.""" + source = await _create_gadget(db=db, branch=default_branch, name="the-source") + owner = await _create_gadget(db=db, branch=default_branch, name="the-owner") + + widget = await Node.init(db=db, schema=WIDGET_KIND, branch=default_branch) + await widget.new( + db=db, + name="sourced-and-owned", + serial={"value": 2200, "source": source.get_id(), "owner": owner.get_id()}, + ) + await widget.save(db=db) + + before = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert {edge.edge_type for edge in open_edges(before)} == { + "HAS_ATTRIBUTE", + "HAS_VALUE", + "IS_PROTECTED", + "HAS_SOURCE", + "HAS_OWNER", + }, "the attribute is expected to hold all four property edge types plus its owning edge" + + await tombstone_existence_only(db=db, node_id=widget.get_id(), branch=default_branch, at=Timestamp()) + + at = Timestamp() + assert await _retire(db=db, node_id=widget.get_id(), at=at) == NodeAgnosticRetirementResult( + edges_closed=len(open_edges(before)) + ) + + after = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert open_edges(after) == [] + + async def test_a_partially_closed_relationship_is_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify an illegal Relationship with 1 closed IS_RELATED edge is correctly retired""" + widget = await _create_widget(db=db, branch=default_branch, name="tombstoned-peer-edge", serial=2000) + gadget = await _create_gadget(db=db, branch=default_branch, name="tombstoned-peer-edge-gadget") + await widget.get_relationship(name="gadget").update(db=db, data=gadget) + await widget.save(db=db) + assert await relationship_peer_shape(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) == ( + 2, + 2, + ), "the relationship is expected to reach two distinct live peers before one edge is tombstoned" + + await tombstone_relationship_peer_edge( + db=db, + node_id=widget.get_id(), + identifier=RELATIONSHIP_IDENTIFIER, + peer_id=gadget.get_id(), + at=Timestamp(), + ) + before = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + assert edge_summary(before) == [ + ("IS_PROTECTED", "active", ""), + ("IS_RELATED", "active", ""), + ("IS_RELATED", "active", ""), + ("IS_RELATED", "deleted", ""), + ], "the tombstone is a new edge alongside the active one it supersedes, and both are open" + attribute_before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + + at = Timestamp() + assert await _retire(db=db, node_id=widget.id, at=at) == NodeAgnosticRetirementResult(edges_closed=3) + + assert edge_summary( + await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + ) == [ + ("IS_PROTECTED", "active", at.to_string()), + ("IS_RELATED", "active", at.to_string()), + ("IS_RELATED", "active", at.to_string()), + ("IS_RELATED", "deleted", ""), + ], "every active edge is closed and the tombstone is left exactly as it was" + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(attribute_before) + ), "the live node's own attribute is retained, so only the relationship was released" + + async def test_relationship_with_renamed_peer_is_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Relationship linked to a peer with an updated kind needs to still be retired correctly.""" + widget = await _create_widget(db=db, branch=default_branch, name="renamed-peer", serial=3300) + gadget = await _create_gadget(db=db, branch=default_branch, name="renamed-peer-gadget") + await widget.get_relationship(name="gadget").update(db=db, data=gadget) + await widget.save(db=db) + + await _rename_widget_kind(db=db, branch=default_branch) + assert await node_vertex_count(db=db, node_id=widget.id) == 2, ( + "the rename is expected to leave a superseded node vertex sharing the uuid" + ) + + await tombstone_existence_only(db=db, node_id=gadget.get_id(), branch=default_branch, at=Timestamp()) + + before = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + open_active_before = [edge for edge in open_edges(before) if edge.status == "active"] + assert open_active_before != [] + attribute_before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + + at = Timestamp() + assert await _retire(db=db, node_id=widget.id, at=at) == NodeAgnosticRetirementResult( + edges_closed=len(open_active_before) + ) + + after = await relationship_global_edges(db=db, node_id=widget.get_id(), identifier=RELATIONSHIP_IDENTIFIER) + assert [edge for edge in open_edges(after) if edge.status == "active"] == [], ( + "every active edge of the relationship is closed once its second peer is gone" + ) + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(attribute_before) + ), "the surviving `widget` still reads its own attribute, so only the relationship was released" + + async def test_a_renamed_kind_is_not_improperly_retired( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify a kind-migrated object is not improperly retired""" + widget = await _create_widget(db=db, branch=default_branch, name="renamed-kind", serial=1100) + + await _rename_widget_kind(db=db, branch=default_branch) + + assert await node_vertex_count(db=db, node_id=widget.id) == 2, ( + "the rename is expected to leave a superseded node vertex sharing the uuid" + ) + assert await attribute_vertex_count(db=db, node_id=widget.id, attribute_name="serial") == 1, ( + "both node vertices are expected to share one attribute vertex" + ) + + before = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert await values_reachable_over_open_edges(db=db, node_id=widget.id, attribute_name="serial") == [1100] + + assert await _retire(db=db, node_id=widget.id, at=Timestamp()) == NodeAgnosticRetirementResult(edges_closed=0) + + assert await values_reachable_over_open_edges(db=db, node_id=widget.id, attribute_name="serial") == [1100] + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(before) + ) + + async def test_a_second_run_over_an_already_retired_field_closes_nothing( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Verify that retirement is idempotent.""" + widget = await _create_widget(db=db, branch=default_branch, name="retired-once", serial=2500) + + deleted_at = Timestamp() + to_delete = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch, raise_on_error=True) + await to_delete.delete(db=db, at=deleted_at) + + after_first_run = await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial") + assert edge_summary(after_first_run) == sorted( + (edge_type, "active", deleted_at.to_string()) + for edge_type in ("HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED") + ) + + second_run_at = Timestamp() + assert second_run_at > deleted_at, "the second run's stamp is expected to be distinguishable from the first" + + assert await _retire(db=db, node_id=widget.id, at=second_run_at) == NodeAgnosticRetirementResult(edges_closed=0) + + assert edge_summary(await attribute_global_edges(db=db, node_id=widget.id, attribute_name="serial")) == ( + edge_summary(after_first_run) + ) diff --git a/backend/tests/helpers/agnostic_edges.py b/backend/tests/helpers/agnostic_edges.py new file mode 100644 index 00000000000..5fec7399ecd --- /dev/null +++ b/backend/tests/helpers/agnostic_edges.py @@ -0,0 +1,266 @@ +"""Graph-shape readers for the branch-agnostic retirement tests. + +These read edges directly rather than going through the node manager: the subject of the assertions +is which edges carry a `to` timestamp and which do not, and a read through the manager would hide the +very states the tests exist to pin down. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from infrahub.core.constants import GLOBAL_BRANCH_NAME + +if TYPE_CHECKING: + from infrahub.core.branch import Branch + from infrahub.core.timestamp import Timestamp + from infrahub.database import InfrahubDatabase + + +@dataclass(frozen=True) +class EdgeState: + """One edge as these assertions care about it: what it is, where it sits, and whether it is open.""" + + edge_type: str + branch: str + status: str + to_time: str | None + + @property + def is_open(self) -> bool: + return self.to_time is None + + @property + def is_active(self) -> bool: + return self.status == "active" + + +def open_edges(edges: list[EdgeState]) -> list[EdgeState]: + """Every open edge, one entry each. + + Counting distinct types instead undercounts a vertex holding two edges of one type, which is every + relationship vertex, so an expected closure count is read from this and never from the types. + """ + return [edge for edge in edges if edge.is_open] + + +def open_active_edges(edges: list[EdgeState]) -> list[EdgeState]: + """The edges a retirement run is expected to close: open, and not already a tombstone.""" + return [edge for edge in edges if edge.is_open and edge.is_active] + + +def open_edge_types(edges: list[EdgeState]) -> set[str]: + return {edge.edge_type for edge in open_edges(edges)} + + +def edge_summary(edges: list[EdgeState]) -> list[tuple[str, str, str]]: + """Order-independent view of a vertex's edges, for before/after comparison.""" + return sorted((edge.edge_type, edge.status, edge.to_time or "") for edge in edges) + + +def branch_summary(edges: list[EdgeState]) -> list[tuple[str, str, str]]: + """Order-independent view keyed on the branch, for the assertions where that is the point.""" + return sorted((edge.branch, edge.status, edge.to_time or "") for edge in edges) + + +def expected_closed_at(edges: list[EdgeState], at: Timestamp) -> list[tuple[str, str, str]]: + """What `edge_summary` should return once every type present has been closed at `at`.""" + return sorted((edge_type, "active", at.to_string()) for edge_type in {edge.edge_type for edge in edges}) + + +def to_times(edges: list[EdgeState]) -> set[str | None]: + return {edge.to_time for edge in edges} + + +async def attribute_global_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[EdgeState]: + """Every global-branch edge touching the attribute vertex, owning edge included.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WITH DISTINCT a + MATCH (a)-[e]-() + WHERE e.branch = $global_branch + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id, "attribute_name": attribute_name, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def relationship_global_edges(db: InfrahubDatabase, node_id: str, identifier: str) -> list[EdgeState]: + """Every global-branch edge of the relationship vertex reached from this node. + + Reached through the node rather than matched on the identifier alone, so a relationship built by + another test sharing the database is not counted. The traversal is unfiltered on purpose: the edge + it arrives over may itself have been closed by the run under assertion. + """ + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:IS_RELATED]-(r:Relationship {name: $identifier}) + WITH DISTINCT r + MATCH (r)-[e]-() + WHERE e.branch = $global_branch + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id, "identifier": identifier, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def attribute_owning_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[EdgeState]: + """Every owning edge of the attribute vertex, on any branch.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WITH DISTINCT a + MATCH (:Node)-[e:HAS_ATTRIBUTE]->(a) + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id, "attribute_name": attribute_name}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def existence_edges(db: InfrahubDatabase, node_id: str) -> list[EdgeState]: + """Every existence edge of the node, on any branch.""" + results = await db.execute_query( + query=""" + MATCH (n:Node {uuid: $node_id})-[e:IS_PART_OF]->(:Root) + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"node_id": node_id}, + ) + return [EdgeState(**dict(result)) for result in results] + + +async def relationship_peer_shape(db: InfrahubDatabase, node_id: str, identifier: str) -> tuple[int, int]: + """How many live global `IS_RELATED` edges this node's relationship has, and how many peers.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:IS_RELATED]-(r:Relationship {name: $identifier}) + WITH DISTINCT r + MATCH (r)-[e:IS_RELATED]-(peer:Node) + WHERE e.branch = $global_branch AND e.status = "active" AND e.to IS NULL + RETURN count(e) AS edge_count, count(DISTINCT peer.uuid) AS peer_count + """, + params={"node_id": node_id, "identifier": identifier, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return results[0]["edge_count"], results[0]["peer_count"] + + +async def node_vertex_count(db: InfrahubDatabase, node_id: str) -> int: + """How many `:Node` vertices carry this uuid; more than one after a kind or inheritance change.""" + results = await db.execute_query( + query="MATCH (n:Node {uuid: $node_id}) RETURN count(n) AS vertex_count", + params={"node_id": node_id}, + ) + return results[0]["vertex_count"] + + +async def attribute_vertex_count(db: InfrahubDatabase, node_id: str, attribute_name: str) -> int: + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + RETURN count(DISTINCT a) AS attribute_count + """, + params={"node_id": node_id, "attribute_name": attribute_name}, + ) + return results[0]["attribute_count"] + + +async def values_reachable_over_open_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[Any]: + """The attribute values a node still reads, following only open, active global edges.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[owning:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WHERE owning.branch = $global_branch AND owning.status = "active" AND owning.to IS NULL + MATCH (a)-[value_edge:HAS_VALUE]->(value:AttributeValue) + WHERE value_edge.branch = $global_branch AND value_edge.status = "active" AND value_edge.to IS NULL + RETURN value.value AS value + """, + params={"node_id": node_id, "attribute_name": attribute_name, "global_branch": GLOBAL_BRANCH_NAME}, + ) + return [result["value"] for result in results] + + +async def tombstone_existence_only(db: InfrahubDatabase, node_id: str, branch: Branch, at: Timestamp) -> None: + """Mark the owner deleted while leaving every field edge exactly as it was. + + The ordinary delete tombstones an attribute's edges alongside the existence edge, so both axes flip + together and either one alone would reach the right answer. This builds the state where they + disagree -- an owner no branch reads as live still holding an open, active global value edge -- + which is the orphan shape this feature repairs. + """ + results = await db.execute_query( + query=""" + MATCH (n:Node {uuid: $node_id})-[existing:IS_PART_OF]->(root:Root) + WHERE existing.branch = $branch AND existing.status = "active" AND existing.to IS NULL + SET existing.to = $at + CREATE (n)-[:IS_PART_OF {branch: $branch, branch_level: $branch_level, status: "deleted", from: $at}]->(root) + RETURN count(existing) AS tombstoned + """, + params={ + "node_id": node_id, + "branch": branch.name, + "branch_level": branch.hierarchy_level, + "at": at.to_string(), + }, + ) + assert results[0]["tombstoned"] == 1 + + +async def tombstone_relationship_peer_edge( + db: InfrahubDatabase, node_id: str, identifier: str, peer_id: str, at: Timestamp +) -> None: + """Add a `deleted` global peer edge that supersedes the active one, leaving the active one open. + + A tombstone is a new, more recent edge rather than a rewrite of the one it supersedes -- the graph + never holds a `deleted` edge where the `active` one it replaced has vanished. Both stay open, and + which of them speaks for the branch is decided by the ordering, which is what puts the status under + test here. + """ + results = await db.execute_query( + query=""" + MATCH (peer:Node {uuid: $peer_id})-[active:IS_RELATED]-(r:Relationship {name: $identifier}) + WHERE active.branch = $global_branch AND active.status = "active" AND active.to IS NULL + AND EXISTS { MATCH (:Node {uuid: $node_id})-[:IS_RELATED]-(r) } + CREATE (peer)-[:IS_RELATED {branch: $global_branch, branch_level: active.branch_level, + status: "deleted", from: $at}]->(r) + RETURN count(active) AS tombstoned + """, + params={ + "node_id": node_id, + "identifier": identifier, + "peer_id": peer_id, + "global_branch": GLOBAL_BRANCH_NAME, + "at": at.to_string(), + }, + ) + assert results[0]["tombstoned"] == 1 + + +async def remove_attribute_on_branch( + db: InfrahubDatabase, node_id: str, attribute_name: str, branch: Branch, at: Timestamp +) -> None: + """Mirror the owning edge with a branch-level `deleted` one, as a schema attribute removal does. + + A removal never closes the global edges; it writes a more specific `deleted` edge that wins under + the removing branch's view only, which is what makes the field disappear there while the object + goes on existing everywhere. + """ + await db.execute_query( + query=""" + MATCH (n:Node {uuid: $node_id})-[owning:HAS_ATTRIBUTE]->(a:Attribute {name: $attribute_name}) + WHERE owning.branch = $global_branch AND owning.status = "active" AND owning.to IS NULL + CREATE (n)-[:HAS_ATTRIBUTE {branch: $branch_name, branch_level: $branch_level, status: "deleted", from: $at}]->(a) + """, + params={ + "node_id": node_id, + "attribute_name": attribute_name, + "global_branch": GLOBAL_BRANCH_NAME, + "branch_name": branch.name, + "branch_level": branch.hierarchy_level, + "at": at.to_string(), + }, + ) diff --git a/backend/tests/helpers/schema/agnostic_retirement.py b/backend/tests/helpers/schema/agnostic_retirement.py new file mode 100644 index 00000000000..9cc86a851f1 --- /dev/null +++ b/backend/tests/helpers/schema/agnostic_retirement.py @@ -0,0 +1,46 @@ +from infrahub.core.constants import BranchSupportType, RelationshipCardinality, RelationshipKind +from infrahub.core.schema import AttributeSchema, NodeSchema, RelationshipSchema, SchemaRoot + +RELATIONSHIP_IDENTIFIER = "agnosticretire_widget__agnosticretire_gadget" + +WIDGET_KIND = "AgnosticretireWidget" +GADGET_KIND = "AgnosticretireGadget" +BEACON_KIND = "AgnosticretireBeacon" + +AGNOSTIC_WIDGET = NodeSchema( + name="Widget", + namespace="Agnosticretire", + branch=BranchSupportType.AWARE, + attributes=[ + AttributeSchema(name="name", kind="Text", unique=True), + AttributeSchema(name="serial", kind="Number", branch=BranchSupportType.AGNOSTIC), + ], + relationships=[ + RelationshipSchema( + name="gadget", + kind=RelationshipKind.GENERIC, + peer=GADGET_KIND, + identifier=RELATIONSHIP_IDENTIFIER, + cardinality=RelationshipCardinality.ONE, + optional=True, + branch=BranchSupportType.AGNOSTIC, + ), + ], +) + +AGNOSTIC_GADGET = NodeSchema( + name="Gadget", + namespace="Agnosticretire", + branch=BranchSupportType.AWARE, + attributes=[AttributeSchema(name="name", kind="Text", unique=True)], +) + +# A kind that is itself branch-agnostic, so its own existence edge lives on the global branch. +AGNOSTIC_BEACON = NodeSchema( + name="Beacon", + namespace="Agnosticretire", + branch=BranchSupportType.AGNOSTIC, + attributes=[AttributeSchema(name="name", kind="Text", unique=True)], +) + +AGNOSTIC_RETIREMENT_SCHEMA = SchemaRoot(nodes=[AGNOSTIC_WIDGET, AGNOSTIC_GADGET, AGNOSTIC_BEACON]) From 43cabda1aa6929b4ab4bb6985605e247f1efc297 Mon Sep 17 00:00:00 2001 From: Aaron McCarty Date: Tue, 18 Aug 2026 15:41:27 -0700 Subject: [PATCH 2/6] fix(backend): roll back on retirement failure and bound the closure in time (#9762) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the object-deletion enforcement point. `git/tasks.py` deleted repository check nodes in session mode, so a retirement failure left the earlier deletions committed — the partial state this feature exists to prevent. The cleanup now runs in a transaction. The session was not a decision against one: history shows it was applied by a 2024 sweep that replaced an unscoped `db` across eight files, and the multi-node delete loop it wraps was already non-atomic. The retirement query now requires `from <= $at` on both the anchor and the closure, so it cannot write a `to` that precedes an edge's own `from`. In practice `at` is the deletion's own timestamp and this cannot arise; one predicate makes the inverted interval unrepresentable. The test builds a candidate holding one edge dated later than the requested time, and removing the closure bound fails it. Removing the anchor bound fails nothing, because the closure bound already prevents the write, so it is a candidate-set narrowing rather than verified behaviour. Atomicity still depends on the caller elsewhere. `ConvertObjectType` deletes in session mode and is left alone deliberately, being a separate mutation whose multi-step conversion was already non-atomic. Making `Node.delete` refuse to retire outside transaction mode would settle it generally. Two further review points are unchanged by decision: non-isolated branches, where the API deprecates the flag, and self-referencing relationships, which are not supported. Co-Authored-By: Claude Opus 5 (1M context) --- .../core/query/node_agnostic_retirement.py | 2 + backend/infrahub/git/tasks.py | 4 +- .../test_node_agnostic_retirement_query.py | 52 +++++++++++++++++++ 3 files changed, 56 insertions(+), 2 deletions(-) diff --git a/backend/infrahub/core/query/node_agnostic_retirement.py b/backend/infrahub/core/query/node_agnostic_retirement.py index 09e7a08c38f..031b5169ec8 100644 --- a/backend/infrahub/core/query/node_agnostic_retirement.py +++ b/backend/infrahub/core/query/node_agnostic_retirement.py @@ -27,6 +27,7 @@ class NodeAgnosticRetirementResult: MATCH (anchor_node:Node {uuid: $node_uuid})-[anchor:HAS_ATTRIBUTE|IS_RELATED]-(field:Attribute|Relationship) WHERE anchor.branch = $global_branch_name AND anchor.status = "active" + AND anchor.from <= $at AND anchor.to IS NULL WITH DISTINCT field %(unretained_predicate)s @@ -34,6 +35,7 @@ class NodeAgnosticRetirementResult: MATCH (field)-[e]-() WHERE e.branch = $global_branch_name AND e.status = "active" + AND e.from <= $at AND e.to IS NULL SET e.to = $at RETURN count(e) AS edges_closed diff --git a/backend/infrahub/git/tasks.py b/backend/infrahub/git/tasks.py index c44205a997a..355dc9fd0b8 100644 --- a/backend/infrahub/git/tasks.py +++ b/backend/infrahub/git/tasks.py @@ -1170,8 +1170,8 @@ async def run_check_merge_conflicts(model: CheckRepositoryMergeConflicts) -> Val await check.save() database = await get_database() - async with database.start_session() as db: - await NodeManager.delete(db=db, nodes=list(existing_checks.values())) + async with database.start_transaction() as dbt: + await NodeManager.delete(db=dbt, nodes=list(existing_checks.values())) return validator_conclusion diff --git a/backend/tests/component/query/test_node_agnostic_retirement_query.py b/backend/tests/component/query/test_node_agnostic_retirement_query.py index 4f5851565aa..dd3f3d76c9e 100644 --- a/backend/tests/component/query/test_node_agnostic_retirement_query.py +++ b/backend/tests/component/query/test_node_agnostic_retirement_query.py @@ -12,6 +12,7 @@ from infrahub.core import registry from infrahub.core.branch import Branch from infrahub.core.constants import ( + GLOBAL_BRANCH_NAME, SchemaPathType, ) from infrahub.core.manager import NodeManager @@ -372,6 +373,57 @@ async def test_a_renamed_kind_is_not_improperly_retired( edge_summary(before) ) + async def test_an_edge_that_begins_after_the_requested_time_is_left_alone( + self, + db: InfrahubDatabase, + default_branch: Branch, + nodedel_schema: None, + ) -> None: + """Closing an edge at a time before it began would invert its interval, so it is not a candidate.""" + widget = await _create_widget(db=db, branch=default_branch, name="edge-from-the-future", serial=4200) + await tombstone_existence_only(db=db, node_id=widget.get_id(), branch=default_branch, at=Timestamp()) + + before = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert open_active_edges(before) != [], "the field has to be retirable, or the bounds are not what spared it" + + # the anchor half: nothing is even a candidate at a time before its owning edge began + assert await _retire( + db=db, node_id=widget.get_id(), at=Timestamp().subtract(hours=1) + ) == NodeAgnosticRetirementResult(edges_closed=0) + assert edge_summary( + await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + ) == edge_summary(before), "no candidate, so nothing is closed" + + # the closure half: a field that IS a candidate, holding one edge that begins later + future = Timestamp().add_delta(hours=1) + await db.execute_query( + query=""" + MATCH (:Node {uuid: $node_id})-[:HAS_ATTRIBUTE]->(a:Attribute {name: "serial"}) + WITH DISTINCT a + MATCH (owner:Node {uuid: $node_id}) + CREATE (a)-[:HAS_SOURCE {branch: $global_branch, branch_level: 1, status: "active", from: $future}]->(owner) + """, + params={ + "node_id": widget.get_id(), + "global_branch": GLOBAL_BRANCH_NAME, + "future": future.to_string(), + }, + ) + with_future = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + assert len(open_active_edges(with_future)) == len(open_active_edges(before)) + 1 + + at = Timestamp() + assert await _retire(db=db, node_id=widget.get_id(), at=at) == NodeAgnosticRetirementResult( + edges_closed=len(open_active_edges(before)) + ), "every edge that had begun is closed, and the one that had not is not counted" + + after = await attribute_global_edges(db=db, node_id=widget.get_id(), attribute_name="serial") + still_open = open_active_edges(after) + assert [edge.edge_type for edge in still_open] == ["HAS_SOURCE"], ( + "the edge that begins later is left open rather than closed before it existed" + ) + assert still_open[0].to_time is None + async def test_a_second_run_over_an_already_retired_field_closes_nothing( self, db: InfrahubDatabase, From 6a862257fb4421bd134897800cf785afca8e92d0 Mon Sep 17 00:00:00 2001 From: Aaron McCarty Date: Tue, 18 Aug 2026 21:10:55 -0700 Subject: [PATCH 3/6] test(backend): pool re-allocation after an object holding a pooled value is deleted (#9762) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Allocate a branch-agnostic value from a number pool, delete the object holding it, allocate again, and assert the same value comes back — along with the graph state that accompanies it: the global HAS_VALUE and HAS_ATTRIBUTE closed, and the IS_RESERVED edge deliberately untouched, since a reservation is the pool's to manage and allocation joins it to the live node by identifier rather than by edge validity. The test disproved the premise it was written under. Re-allocation does not depend on retirement: `BaseAttribute.get_branch_for_delete` returns the node's branch for an agnostic attribute on an aware node, so an ordinary delete already writes branch-scoped `deleted` HAS_VALUE and HAS_ATTRIBUTE edges, and the pool's used-value queries run `branch_agnostic=True`, which collapses the filter to a pure time predicate under which those tombstones win. Neutralising the closure leaves re-allocation working. What retirement fixes is the uniqueness-validation leak, which reads the graph differently. So the test keeps its place through the graph assertions rather than through SC-007, and it is mutation-proven on those: neutralising the closure fails it. The success criterion is amended separately. Co-Authored-By: Claude Opus 5 (1M context) --- .../core/test_agnostic_retirement.py | 74 ++++++++++++++++++- backend/tests/helpers/agnostic_edges.py | 12 +++ 2 files changed, 85 insertions(+), 1 deletion(-) diff --git a/backend/tests/component/core/test_agnostic_retirement.py b/backend/tests/component/core/test_agnostic_retirement.py index 9a1de8dfb06..a1bc2bd16ee 100644 --- a/backend/tests/component/core/test_agnostic_retirement.py +++ b/backend/tests/component/core/test_agnostic_retirement.py @@ -16,10 +16,11 @@ import pytest from infrahub.core import registry -from infrahub.core.constants import GLOBAL_BRANCH_NAME +from infrahub.core.constants import GLOBAL_BRANCH_NAME, InfrahubKind from infrahub.core.initialization import create_branch from infrahub.core.manager import NodeManager from infrahub.core.node import Node +from infrahub.core.node.resource_manager.number_pool import CoreNumberPool from infrahub.core.query.node_agnostic_retirement import RetireNodeAgnosticFieldsQuery from infrahub.core.timestamp import Timestamp from infrahub.database import InfrahubDatabase, InfrahubDatabaseMode @@ -29,8 +30,10 @@ from infrahub.core.branch import Branch from infrahub.core.query import QueryType + from infrahub.core.schema.schema_branch import SchemaBranch from tests.helpers.agnostic_edges import ( + EdgeState, attribute_global_edges, attribute_owning_edges, edge_summary, @@ -38,6 +41,7 @@ expected_closed_at, open_edge_types, open_edges, + pool_reservation_edges, relationship_global_edges, remove_attribute_on_branch, ) @@ -48,6 +52,9 @@ WIDGET_KIND, ) +SERIAL_POOL_START = 9001 +SERIAL_POOL_END = 9002 + class RetirementFailureError(Exception): """Stands in for whatever the retirement run could fail with.""" @@ -107,6 +114,32 @@ async def default_branch(self, default_branch_scope_class: Branch) -> Branch: async def agnostic_schema(self, db: InfrahubDatabase, default_branch: Branch) -> None: registry.schema.register_schema(schema=AGNOSTIC_RETIREMENT_SCHEMA, branch=default_branch.name) + @pytest.fixture(scope="class") + async def serial_pool( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + register_core_models_schema_scope_class: SchemaBranch, + ) -> CoreNumberPool: + """A pool of two numbers backing the widget's branch-agnostic serial. + + Two rather than one so that an allocation which fails to reuse the freed value still succeeds, + and reports the number it handed out instead of an exhausted pool. + """ + registry.node[InfrahubKind.NUMBERPOOL] = CoreNumberPool + pool = await CoreNumberPool.init(db=db, schema=InfrahubKind.NUMBERPOOL) + await pool.new( + db=db, + name="agnostic-serial-pool", + node=WIDGET_KIND, + node_attribute="serial", + start_range=SERIAL_POOL_START, + end_range=SERIAL_POOL_END, + ) + await pool.save(db=db) + return pool + async def test_a_field_created_and_deleted_on_the_same_user_branch_is_closed_by_the_delete( self, db: InfrahubDatabase, @@ -355,3 +388,42 @@ async def test_a_retirement_failure_propagates_and_leaves_the_graph_untouched( still_there = await NodeManager.get_one(db=db, id=widget.id, branch=default_branch) assert still_there is not None assert still_there.get_attribute(name="serial").value == 700 + + async def test_a_value_freed_by_retirement_is_allocatable_again_from_its_pool( + self, + db: InfrahubDatabase, + default_branch: Branch, + agnostic_schema: None, + serial_pool: CoreNumberPool, + ) -> None: + """Deleting the object holding a pooled value returns that value to the pool.""" + holder = await Node.init(db=db, schema=WIDGET_KIND, branch=default_branch) + await holder.new(db=db, name="holds-a-pooled-serial", serial={"from_pool": {"id": serial_pool.id}}) + await holder.save(db=db) + + assert holder.get_attribute(name="serial").value == SERIAL_POOL_START + assert await serial_pool.get_used(db=db, branch=default_branch) == [SERIAL_POOL_START] + + before = await attribute_global_edges(db=db, node_id=holder.id, attribute_name="serial") + assert open_edge_types(before) == {"HAS_ATTRIBUTE", "HAS_VALUE", "IS_PROTECTED", "HAS_SOURCE"} + reserved_before = await pool_reservation_edges(db=db, pool_id=serial_pool.id, identifier=holder.id) + assert reserved_before == [ + EdgeState(edge_type="IS_RESERVED", branch=GLOBAL_BRANCH_NAME, status="active", to_time=None) + ] + + deleted_at = Timestamp() + await _delete(db=db, node_id=holder.id, branch=default_branch, at=deleted_at) + + after = await attribute_global_edges(db=db, node_id=holder.id, attribute_name="serial") + assert edge_summary(after) == expected_closed_at(before, deleted_at) + assert await pool_reservation_edges(db=db, pool_id=serial_pool.id, identifier=holder.id) == reserved_before, ( + "the reservation is never cleaned up on delete, and does not need to be" + ) + assert await serial_pool.get_used(db=db, branch=default_branch) == [] + + reallocated = await Node.init(db=db, schema=WIDGET_KIND, branch=default_branch) + await reallocated.new(db=db, name="takes-the-freed-serial", serial={"from_pool": {"id": serial_pool.id}}) + await reallocated.save(db=db) + + assert reallocated.get_attribute(name="serial").value == SERIAL_POOL_START + assert await serial_pool.get_used(db=db, branch=default_branch) == [SERIAL_POOL_START] diff --git a/backend/tests/helpers/agnostic_edges.py b/backend/tests/helpers/agnostic_edges.py index 5fec7399ecd..a758834b946 100644 --- a/backend/tests/helpers/agnostic_edges.py +++ b/backend/tests/helpers/agnostic_edges.py @@ -264,3 +264,15 @@ async def remove_attribute_on_branch( "at": at.to_string(), }, ) + + +async def pool_reservation_edges(db: InfrahubDatabase, pool_id: str, identifier: str) -> list[EdgeState]: + """Every reservation edge a pool holds under one identifier, on any branch.""" + results = await db.execute_query( + query=""" + MATCH (:Node {uuid: $pool_id})-[e:IS_RESERVED {identifier: $identifier}]->(:AttributeValue) + RETURN type(e) AS edge_type, e.branch AS branch, e.status AS status, e.to AS to_time + """, + params={"pool_id": pool_id, "identifier": identifier}, + ) + return [EdgeState(**dict(result)) for result in results] From 446b432bebbf351f4fba5cb721a78c4664275643 Mon Sep 17 00:00:00 2001 From: Aaron McCarty Date: Thu, 20 Aug 2026 11:21:52 -0700 Subject: [PATCH 4/6] delete each check in its own txn --- backend/infrahub/git/tasks.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/backend/infrahub/git/tasks.py b/backend/infrahub/git/tasks.py index 355dc9fd0b8..b3be6e63e55 100644 --- a/backend/infrahub/git/tasks.py +++ b/backend/infrahub/git/tasks.py @@ -1170,8 +1170,9 @@ async def run_check_merge_conflicts(model: CheckRepositoryMergeConflicts) -> Val await check.save() database = await get_database() - async with database.start_transaction() as dbt: - await NodeManager.delete(db=dbt, nodes=list(existing_checks.values())) + for check in existing_checks.values(): + async with database.start_transaction() as dbt: + await NodeManager.delete(db=dbt, nodes=[check]) return validator_conclusion From e6fa96aa14da8e03de6c8a5ba85addf1f944dffd Mon Sep 17 00:00:00 2001 From: Aaron McCarty Date: Thu, 20 Aug 2026 11:22:58 -0700 Subject: [PATCH 5/6] remove unused test functions --- backend/tests/helpers/agnostic_edges.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/backend/tests/helpers/agnostic_edges.py b/backend/tests/helpers/agnostic_edges.py index a758834b946..06ff0d92147 100644 --- a/backend/tests/helpers/agnostic_edges.py +++ b/backend/tests/helpers/agnostic_edges.py @@ -59,20 +59,11 @@ def edge_summary(edges: list[EdgeState]) -> list[tuple[str, str, str]]: return sorted((edge.edge_type, edge.status, edge.to_time or "") for edge in edges) -def branch_summary(edges: list[EdgeState]) -> list[tuple[str, str, str]]: - """Order-independent view keyed on the branch, for the assertions where that is the point.""" - return sorted((edge.branch, edge.status, edge.to_time or "") for edge in edges) - - def expected_closed_at(edges: list[EdgeState], at: Timestamp) -> list[tuple[str, str, str]]: """What `edge_summary` should return once every type present has been closed at `at`.""" return sorted((edge_type, "active", at.to_string()) for edge_type in {edge.edge_type for edge in edges}) -def to_times(edges: list[EdgeState]) -> set[str | None]: - return {edge.to_time for edge in edges} - - async def attribute_global_edges(db: InfrahubDatabase, node_id: str, attribute_name: str) -> list[EdgeState]: """Every global-branch edge touching the attribute vertex, owning edge included.""" results = await db.execute_query( From 0e27d7b69f5faf3345962fe75e80c29b1bb46ebc Mon Sep 17 00:00:00 2001 From: Aaron McCarty Date: Thu, 20 Aug 2026 15:27:42 -0700 Subject: [PATCH 6/6] refactor(backend): stream winning edge statuses in agnostic retention (#9762) Collect the candidate fields in the caller, drop branch-aware fields early with an EXISTS gate, and resolve each peer's winning field and existence edges with OPTIONAL CALL ... LIMIT 1 subqueries instead of sorting and collecting the full row set. Top-1 per peer and branch keeps memory flat where the global sort grew with edge history and branch count. Co-Authored-By: Claude Fable 5 --- .../infrahub/core/query/agnostic_retention.py | 122 ++++++++++-------- .../core/query/node_agnostic_retirement.py | 2 +- 2 files changed, 67 insertions(+), 57 deletions(-) diff --git a/backend/infrahub/core/query/agnostic_retention.py b/backend/infrahub/core/query/agnostic_retention.py index 735ee28f0ea..9d623f39edc 100644 --- a/backend/infrahub/core/query/agnostic_retention.py +++ b/backend/infrahub/core/query/agnostic_retention.py @@ -13,22 +13,25 @@ peer is not a relationship. A `:Relationship` must also have two distinct active peers to be considered active. +The winner lookups are OPTIONAL CALL subqueries and the peers are counted conditionally rather than +filtered so that a field with no live edge on a branch reaches the end of the predicate carrying a +count of zero -- a plain CALL or MATCH would drop its row instead. + Assumption: every branch forks from the default branch. A branch-of-branch feature would not extend this logic. """ -# Expects `field` in scope, one row per candidate vertex, plus the `$global_branch_name` and `$at` -# parameters. Emits the candidates no branch retains, with `field` as the only variable in scope. +# Expects `agnostic_candidates` in scope: a list of the candidate `:Attribute` / `:Relationship` +# vertices, plus the `$global_branch_name` and `$at` parameters. Emits one row per candidate that no +# branch retains, with `field` as the only variable in scope. UNRETAINED_AGNOSTIC_FIELD_PREDICATE = """ -WITH collect(field) AS candidates - // ---------------------- // The branches are read once for the whole run and carried as a list. // ---------------------- MATCH (branch:Branch) WHERE branch.name <> $global_branch_name WITH - candidates, + agnostic_candidates, collect({ name: branch.name, origin_name: CASE WHEN branch.is_default THEN NULL ELSE branch.origin_branch END, @@ -39,49 +42,56 @@ END }) AS branch_windows -UNWIND candidates AS field +UNWIND agnostic_candidates AS field +WITH DISTINCT field, branch_windows +// ---------------------- +// Quick filter to remove all fields with no active edges on the global branch, covers most fields +// ---------------------- +WHERE EXISTS { + MATCH (field)-[global_edge]-() + WHERE global_edge.branch = $global_branch_name + AND global_edge.status = "active" + AND global_edge.to IS NULL +} WITH field, branch_windows, CASE WHEN field:Relationship THEN 2 ELSE 1 END AS required_live_peers -CALL (field, branch_windows) { - UNWIND branch_windows AS branch_window - WITH - field, - branch_window.name AS branch_name, - branch_window.origin_name AS origin_name, - branch_window.origin_at AS origin_at +// ---------------------- +// Get all the possible peer :Nodes for each field. Usually 1 for :Attribute and 2 for :Relationship +// ---------------------- +OPTIONAL MATCH (node:Node)-[:HAS_ATTRIBUTE|IS_RELATED]-(field) +WITH DISTINCT field, branch_windows, required_live_peers, node - // ---------------------- - // Count this `field`'s active links to :Node vertices on this branch - // ---------------------- - MATCH (node:Node)-[field_edge:HAS_ATTRIBUTE|IS_RELATED]-(field) +UNWIND branch_windows AS branch_window +WITH + field, + required_live_peers, + node, + branch_window.name AS branch_name, + branch_window.origin_name AS origin_name, + branch_window.origin_at AS origin_at + +// ---------------------- +// Resolve the status of this peer's latest edge to the field on this branch. +// ---------------------- +OPTIONAL CALL (field, node, branch_name, origin_name, origin_at) { + MATCH (node)-[field_edge:HAS_ATTRIBUTE|IS_RELATED]-(field) WHERE (field_edge.branch IN [$global_branch_name, branch_name] AND field_edge.from <= $at AND (field_edge.to IS NULL OR field_edge.to > $at)) OR (field_edge.branch = origin_name AND field_edge.from <= origin_at AND (field_edge.to IS NULL OR field_edge.to > origin_at)) - WITH - branch_name, - origin_name, - origin_at, - node, - field_edge.branch_level AS field_edge_level, - field_edge.from AS field_edge_from, - field_edge.status AS field_edge_status - ORDER BY field_edge_level DESC, field_edge_from DESC, field_edge_status ASC - WITH - branch_name, - origin_name, - origin_at, - node, - collect(field_edge_status)[0] AS winning_field_edge_status - WHERE winning_field_edge_status = "active" + RETURN field_edge.status AS latest_field_edge_status + ORDER BY field_edge.branch_level DESC, field_edge.from DESC, field_edge.status ASC + LIMIT 1 +} - // ---------------------- - // Check that each linked :Node vertex is active on this branch. - // The global branch stays in the existence match so that an owner which is itself branch-agnostic - // reads as live on every branch and is therefore retained. - // ---------------------- +// ---------------------- +// Resolve whether this peer exists on this branch. +// The global branch stays in the existence match so that an owner which is itself branch-agnostic +// reads as live on every branch and is therefore retained. +// ---------------------- +OPTIONAL CALL (node, branch_name, origin_name, origin_at) { MATCH (node)-[existence:IS_PART_OF]->(:Root) WHERE (existence.branch IN [$global_branch_name, branch_name] AND existence.from <= $at @@ -89,24 +99,24 @@ OR (existence.branch = origin_name AND existence.from <= origin_at AND (existence.to IS NULL OR existence.to > origin_at)) - WITH - branch_name, - node, - existence.branch_level AS existence_level, - existence.from AS existence_from, - existence.status AS existence_status - ORDER BY existence_level DESC, existence_from DESC, existence_status ASC - WITH branch_name, node, collect(existence_status)[0] AS winning_existence - WHERE winning_existence = "active" - - // ---------------------- - // Peers are counted by uuid. Kind/inheritance migration leaves multiple Node vertices with - // the same uuid for a single entity - // ---------------------- - WITH branch_name, count(DISTINCT node.uuid) AS live_peer_count - RETURN max(live_peer_count) AS most_live_peers + RETURN existence.status AS latest_existence + ORDER BY existence.branch_level DESC, existence.from DESC, existence.status ASC + LIMIT 1 } -WITH field, required_live_peers, coalesce(most_live_peers, 0) AS live_peers -WHERE live_peers < required_live_peers +// ---------------------- +// Peers are counted by uuid. Kind/inheritance migration leaves multiple Node vertices with +// the same uuid for a single entity. +// ---------------------- +WITH + field, + required_live_peers, + branch_name, + count(DISTINCT CASE + WHEN latest_field_edge_status = "active" AND latest_existence = "active" THEN node.uuid + END) AS live_peer_count + +WITH field, required_live_peers, max(live_peer_count) AS most_live_peers +WHERE most_live_peers < required_live_peers +WITH field """ diff --git a/backend/infrahub/core/query/node_agnostic_retirement.py b/backend/infrahub/core/query/node_agnostic_retirement.py index 031b5169ec8..dd62bd90309 100644 --- a/backend/infrahub/core/query/node_agnostic_retirement.py +++ b/backend/infrahub/core/query/node_agnostic_retirement.py @@ -29,7 +29,7 @@ class NodeAgnosticRetirementResult: AND anchor.status = "active" AND anchor.from <= $at AND anchor.to IS NULL -WITH DISTINCT field +WITH collect(DISTINCT field) AS agnostic_candidates %(unretained_predicate)s MATCH (field)-[e]-()