diff --git a/changelog/572.fixed.md b/changelog/572.fixed.md new file mode 100644 index 00000000..c9aaca61 --- /dev/null +++ b/changelog/572.fixed.md @@ -0,0 +1,3 @@ +Tracking groups now reconcile correctly when a run saves no nodes at all. Previously `update_group()` returned early on an empty member list, so a generator that produced nothing (a decommissioning run) or a repository whose last object file was removed left every previously tracked node behind as an orphan, still listed in the group. A run that tracks nothing but has an existing group now prunes it; a run that tracks nothing and has no group still creates none. + +Cleanup is also no longer aborted by a single refused delete. `delete_unused()` attempts every unused member and reports the failures together as `TrackingGroupCleanupError` instead of propagating the first `GraphQLError` and silently skipping the rest. Members that could not be deleted are kept in the tracking group so a later run retries them, and the sync client now has the same error tolerance as the async one. diff --git a/infrahub_sdk/client.py b/infrahub_sdk/client.py index d600ea2f..3eba70d8 100644 --- a/infrahub_sdk/client.py +++ b/infrahub_sdk/client.py @@ -2139,10 +2139,13 @@ async def __aexit__( exc_value: BaseException | None, traceback: TracebackType | None, ) -> None: - if exc_type is None and self.mode == InfrahubClientMode.TRACKING: - await self.group_context.update_group() - - self.mode = InfrahubClientMode.DEFAULT + try: + if exc_type is None and self.mode == InfrahubClientMode.TRACKING: + await self.group_context.update_group() + finally: + # update_group() can raise, and leaving the client in tracking mode would + # silently enroll every later save into the stale context. + self.mode = InfrahubClientMode.DEFAULT async def convert_object_type( self, @@ -3883,10 +3886,13 @@ def __exit__( exc_value: BaseException | None, traceback: TracebackType | None, ) -> None: - if exc_type is None and self.mode == InfrahubClientMode.TRACKING: - self.group_context.update_group() - - self.mode = InfrahubClientMode.DEFAULT + try: + if exc_type is None and self.mode == InfrahubClientMode.TRACKING: + self.group_context.update_group() + finally: + # update_group() can raise, and leaving the client in tracking mode would + # silently enroll every later save into the stale context. + self.mode = InfrahubClientMode.DEFAULT def convert_object_type( self, diff --git a/infrahub_sdk/exceptions.py b/infrahub_sdk/exceptions.py index 02111b9a..df263c39 100644 --- a/infrahub_sdk/exceptions.py +++ b/infrahub_sdk/exceptions.py @@ -66,6 +66,19 @@ def __init__(self, errors: list[dict[str, Any]], query: str | None = None, varia super().__init__(self.message) +class TrackingGroupCleanupError(Error): + """Raised when unused members of a tracking group could not be deleted. + + Every unused member is attempted before this is raised, and the ones that failed are + kept in the tracking group so a later run retries them. + """ + + def __init__(self, failures: dict[str, str]) -> None: + self.failures = failures + details = "; ".join(f"{node_id} ({reason})" for node_id, reason in failures.items()) + super().__init__(f"Unable to delete {len(failures)} unused member(s) of the tracking group: {details}") + + class VersionNotSupportedError(Error): """Raised when a feature is used against an Infrahub server version that does not support it.""" diff --git a/infrahub_sdk/query_groups.py b/infrahub_sdk/query_groups.py index 3fcde346..1093ee57 100644 --- a/infrahub_sdk/query_groups.py +++ b/infrahub_sdk/query_groups.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any from .constants import InfrahubClientMode -from .exceptions import GraphQLError, NodeNotFoundError +from .exceptions import Error, NodeNotFoundError, TrackingGroupCleanupError from .utils import dict_hash if TYPE_CHECKING: @@ -108,17 +108,32 @@ async def get_group(self, store_peers: bool = False) -> InfrahubNode | None: self.previous_members = group._get_relationship_many(name="members").peers return group - async def delete_unused(self) -> None: - if self.previous_members and self.unused_member_ids: - for member in self.previous_members: - if member.id in self.unused_member_ids and member.typename: - try: - await self.client.delete(kind=member.typename, id=member.id) - except GraphQLError as exc: - if not exc.message or "Unable to find the node" not in exc.message: - # If the node already has been deleted, skip the error as it would have been deleted - # by the cascade delete of another node - raise + async def delete_unused(self) -> dict[str, str]: + """Delete the members that this run no longer uses. + + Every candidate is attempted even when some deletes are refused, so one refusal + cannot leave the rest of the unused members behind. + + Returns: + The id of each member that could not be deleted, mapped to the reason. + + """ + failures: dict[str, str] = {} + if not self.previous_members or not self.unused_member_ids: + return failures + + for member in self.previous_members: + if member.id not in self.unused_member_ids or not member.typename: + continue + try: + await self.client.delete(kind=member.typename, id=member.id, branch=self.branch) + except Error as exc: + if exc.message and "Unable to find the node" in exc.message: + # The node was already removed by the cascade delete of another node + continue + failures[member.id] = exc.message or str(exc) + + return failures async def add_related_nodes(self, ids: list[str], update_group_context: bool | None = None) -> None: """Add related Nodes IDs to the context. @@ -147,42 +162,49 @@ async def add_related_groups(self, ids: list[str], update_group_context: bool | self.related_group_ids.extend(ids) async def update_group(self) -> None: - """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution.""" + """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution. + + Raises: + TrackingGroupCleanupError: When one or more unused members could not be deleted. + + """ members: list[str] = self.related_group_ids + self.related_node_ids - if not members: + existing_group = None + if self.delete_unused_nodes: + existing_group = await self.get_group(store_peers=True) + + # A run that tracked nothing and has no group to reconcile must not create an empty one. + if not members and existing_group is None: return + failures: dict[str, str] = {} + if existing_group: + previous_member_ids: list[str] = existing_group._get_relationship_many(name="members").peer_ids + self.unused_member_ids = list(set(previous_member_ids) - set(members)) + failures = await self.delete_unused() + + # An already-empty group that stays empty needs no upsert. + if not members and not previous_member_ids: + return + group_name = self._generate_group_name() schema = await self.client.schema.get(kind=self.group_type) description = self._generate_group_description(schema=schema) - existing_group = None - if self.delete_unused_nodes: - existing_group = await self.get_group(store_peers=True) - + # Members that could not be deleted stay in the group so a later run retries them. group = await self.client.create( kind=self.group_type, name=group_name, description=description, - members=members, + members=members + list(failures), branch=self.branch, **self.group_params, ) await group.save(allow_upsert=True, update_group_context=False) - if not existing_group: - return - - # Calculate how many nodes should be deleted - self.unused_member_ids = list( - set(existing_group._get_relationship_many(name="members").peer_ids) - set(members) - ) - - if not self.delete_unused_nodes: - return - - await self.delete_unused() + if failures: + raise TrackingGroupCleanupError(failures=failures) # TODO : create anoter "read" group. Could be based of the store items # Need to filters the store items inherited from CoreGroup to add them as children # Need to validate that it's UUIDas "key" if we want to implement other methods to store item @@ -198,7 +220,9 @@ def __init__(self, client: InfrahubClientSync) -> None: def get_group(self, store_peers: bool = False) -> InfrahubNodeSync | None: group_name = self._generate_group_name() try: - group = self.client.get(kind=self.group_type, name__value=group_name, include=["members"]) + group = self.client.get( + kind=self.group_type, name__value=group_name, include=["members"], branch=self.branch + ) except NodeNotFoundError: return None @@ -208,11 +232,32 @@ def get_group(self, store_peers: bool = False) -> InfrahubNodeSync | None: self.previous_members = group._get_relationship_many(name="members").peers return group - def delete_unused(self) -> None: - if self.previous_members and self.unused_member_ids: - for member in self.previous_members: - if member.id in self.unused_member_ids and member.typename: - self.client.delete(kind=member.typename, id=member.id) + def delete_unused(self) -> dict[str, str]: + """Delete the members that this run no longer uses. + + Every candidate is attempted even when some deletes are refused, so one refusal + cannot leave the rest of the unused members behind. + + Returns: + The id of each member that could not be deleted, mapped to the reason. + + """ + failures: dict[str, str] = {} + if not self.previous_members or not self.unused_member_ids: + return failures + + for member in self.previous_members: + if member.id not in self.unused_member_ids or not member.typename: + continue + try: + self.client.delete(kind=member.typename, id=member.id, branch=self.branch) + except Error as exc: + if exc.message and "Unable to find the node" in exc.message: + # The node was already removed by the cascade delete of another node + continue + failures[member.id] = exc.message or str(exc) + + return failures def add_related_nodes(self, ids: list[str], update_group_context: bool | None = None) -> None: """Add related Nodes IDs to the context. @@ -241,42 +286,49 @@ def add_related_groups(self, ids: list[str], update_group_context: bool | None = self.related_group_ids.extend(ids) def update_group(self) -> None: - """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution.""" + """Create or update (using upsert) a CoreStandardGroup to store all the Nodes and Groups used during an execution. + + Raises: + TrackingGroupCleanupError: When one or more unused members could not be deleted. + + """ members: list[str] = self.related_node_ids + self.related_group_ids - if not members: + existing_group = None + if self.delete_unused_nodes: + existing_group = self.get_group(store_peers=True) + + # A run that tracked nothing and has no group to reconcile must not create an empty one. + if not members and existing_group is None: return + failures: dict[str, str] = {} + if existing_group: + previous_member_ids: list[str] = existing_group._get_relationship_many(name="members").peer_ids + self.unused_member_ids = list(set(previous_member_ids) - set(members)) + failures = self.delete_unused() + + # An already-empty group that stays empty needs no upsert. + if not members and not previous_member_ids: + return + group_name = self._generate_group_name() schema = self.client.schema.get(kind=self.group_type) description = self._generate_group_description(schema=schema) - existing_group = None - if self.delete_unused_nodes: - existing_group = self.get_group(store_peers=True) - + # Members that could not be deleted stay in the group so a later run retries them. group = self.client.create( kind=self.group_type, name=group_name, description=description, - members=members, + members=members + list(failures), branch=self.branch, **self.group_params, ) group.save(allow_upsert=True, update_group_context=False) - if not existing_group: - return - - # Calculate how many nodes should be deleted - self.unused_member_ids = list( - set(existing_group._get_relationship_many(name="members").peer_ids) - set(members) - ) - - if not self.delete_unused_nodes: - return - - self.delete_unused() + if failures: + raise TrackingGroupCleanupError(failures=failures) # TODO : create anoter "read" group. Could be based of the store items # Need to filters the store items inherited from CoreGroup to add them as children diff --git a/tests/integration/test_tracking_zero_members.py b/tests/integration/test_tracking_zero_members.py new file mode 100644 index 00000000..5eb61606 --- /dev/null +++ b/tests/integration/test_tracking_zero_members.py @@ -0,0 +1,267 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from infrahub_sdk.constants import InfrahubClientMode +from infrahub_sdk.exceptions import NodeNotFoundError, TrackingGroupCleanupError +from infrahub_sdk.testing.docker import TestInfrahubDockerClient +from infrahub_sdk.testing.schemas.animal import TESTING_CAT, TESTING_PERSON, SchemaAnimal + +if TYPE_CHECKING: + from infrahub_sdk import InfrahubClient, InfrahubClientSync + + +class TestTrackingZeroMembers(TestInfrahubDockerClient, SchemaAnimal): + @pytest.fixture(scope="class") + async def base_dataset(self, client: InfrahubClient, load_schema: None) -> None: + return None + + async def test_zero_member_run_prunes_previous_members(self, client: InfrahubClient, base_dataset: None) -> None: + person_name = "TrackingZeroMemberPerson" + tag_name = "tracking-zero-TAG" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + tag = await clt.create(kind="BuiltinTag", name=tag_name) + await tag.save(allow_upsert=True) + person = await clt.create(kind=TESTING_PERSON, name=person_name, tags=[tag]) + await person.save(allow_upsert=True) + + group_name = client.group_context._generate_group_name() + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # A run that saves nothing must still prune everything the previous run tracked. + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + + with pytest.raises(NodeNotFoundError): + await client.get(kind="BuiltinTag", name__value=tag_name) + with pytest.raises(NodeNotFoundError): + await client.get(kind=TESTING_PERSON, name__value=person_name) + + async def test_zero_member_run_without_existing_group_creates_nothing( + self, client: InfrahubClient, base_dataset: None + ) -> None: + params = {"person_name": "TrackingNeverAnyMembers"} + + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + group_name = client.group_context._generate_group_name() + with pytest.raises(NodeNotFoundError): + await client.get(kind="CoreStandardGroup", name__value=group_name) + + async def test_refused_delete_does_not_abort_remaining_reaps( + self, client: InfrahubClient, base_dataset: None + ) -> None: + person_name = "TrackingRefusedPerson" + doomed_tag_name = "tracking-refused-DOOMED" + keeper_tag_name = "tracking-refused-KEEPER" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = await clt.create(kind=TESTING_PERSON, name=person_name) + await person.save(allow_upsert=True) + doomed_tag = await clt.create(kind="BuiltinTag", name=doomed_tag_name) + await doomed_tag.save(allow_upsert=True) + + group_name = client.group_context._generate_group_name() + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # An animal outside the tracking group makes its owner undeletable, + # because Animal.owner is a mandatory relationship. + cat = await client.create(kind=TESTING_CAT, name="TrackingRefusedCat", breed="Bengal", owner=person) + await cat.save() + + # Second run saves only a new tag, so the person and the first tag both + # become reap candidates. The person's delete is refused by the server. + with pytest.raises(TrackingGroupCleanupError) as exc_info: + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + keeper_tag = await clt.create(kind="BuiltinTag", name=keeper_tag_name) + await keeper_tag.save(allow_upsert=True) + + assert list(exc_info.value.failures) == [person.id] + + # The refused delete must not prevent the other unused member from being reaped. + with pytest.raises(NodeNotFoundError): + await client.get(kind="BuiltinTag", name__value=doomed_tag_name) + + # The person survived, and must still be a group member so a later run can retry it. + await client.get(kind=TESTING_PERSON, name__value=person_name) + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert sorted(group.members.peer_ids) == sorted([person.id, keeper_tag.id]) + + +class TestTrackingRefusedDeleteOnZeroMemberRun(TestInfrahubDockerClient, SchemaAnimal): + @pytest.fixture(scope="class") + async def base_dataset(self, client: InfrahubClient, load_schema: None) -> None: + return None + + async def test_zero_member_run_keeps_undeletable_member(self, client: InfrahubClient, base_dataset: None) -> None: + person_name = "TrackingRetryPerson" + params = {"person_name": person_name} + + async with client.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = await clt.create(kind=TESTING_PERSON, name=person_name) + await person.save(allow_upsert=True) + + cat = await client.create(kind=TESTING_CAT, name="TrackingRetryCat", breed="Bengal", owner=person) + await cat.save() + + # A zero-member run now attempts the reap; the person's delete is refused. + with pytest.raises(TrackingGroupCleanupError): + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + assert client.mode == InfrahubClientMode.DEFAULT + + group_name = client.group_context._generate_group_name() + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert group.members.peer_ids == [person.id] + + # Once the blocking node is gone, the next zero-member run reaps the person. + await client.delete(kind=TESTING_CAT, id=cat.id) + async with client.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = await client.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + with pytest.raises(NodeNotFoundError): + await client.get(kind=TESTING_PERSON, name__value=person_name) + + +class TestTrackingZeroMembersSync(TestInfrahubDockerClient, SchemaAnimal): + @pytest.fixture(scope="class") + async def base_dataset(self, client: InfrahubClient, load_schema: None) -> None: + return None + + def test_zero_member_run_prunes_previous_members(self, client_sync: InfrahubClientSync, base_dataset: None) -> None: + person_name = "SyncTrackingZeroMemberPerson" + tag_name = "sync-tracking-zero-TAG" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt: + tag = clt.create(kind="BuiltinTag", name=tag_name) + tag.save(allow_upsert=True) + person = clt.create(kind=TESTING_PERSON, name=person_name, tags=[tag]) + person.save(allow_upsert=True) + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # A run that saves nothing must still prune everything the previous run tracked. + with client_sync.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + + with pytest.raises(NodeNotFoundError): + client_sync.get(kind="BuiltinTag", name__value=tag_name) + with pytest.raises(NodeNotFoundError): + client_sync.get(kind=TESTING_PERSON, name__value=person_name) + + def test_zero_member_run_without_existing_group_creates_nothing( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + params = {"person_name": "SyncTrackingNeverAnyMembers"} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True): + pass + + group_name = client_sync.group_context._generate_group_name() + with pytest.raises(NodeNotFoundError): + client_sync.get(kind="CoreStandardGroup", name__value=group_name) + + def test_refused_delete_does_not_abort_remaining_reaps( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + person_name = "SyncTrackingRefusedPerson" + doomed_tag_name = "sync-tracking-refused-DOOMED" + keeper_tag_name = "sync-tracking-refused-KEEPER" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = clt.create(kind=TESTING_PERSON, name=person_name) + person.save(allow_upsert=True) + doomed_tag = clt.create(kind="BuiltinTag", name=doomed_tag_name) + doomed_tag.save(allow_upsert=True) + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 2 + + # An animal outside the tracking group makes its owner undeletable, + # because Animal.owner is a mandatory relationship. + cat = client_sync.create(kind=TESTING_CAT, name="SyncTrackingRefusedCat", breed="Bengal", owner=person) + cat.save() + + # Second run saves only a new tag, so the person and the first tag both + # become reap candidates. The person's delete is refused by the server. + with ( + pytest.raises(TrackingGroupCleanupError) as exc_info, + client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt, + ): + keeper_tag = clt.create(kind="BuiltinTag", name=keeper_tag_name) + keeper_tag.save(allow_upsert=True) + + assert list(exc_info.value.failures) == [person.id] + + # The refused delete must not prevent the other unused member from being reaped. + with pytest.raises(NodeNotFoundError): + client_sync.get(kind="BuiltinTag", name__value=doomed_tag_name) + + # The person survived, and must still be a group member so a later run can retry it. + client_sync.get(kind=TESTING_PERSON, name__value=person_name) + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert sorted(group.members.peer_ids) == sorted([person.id, keeper_tag.id]) + + +class TestTrackingRefusedDeleteOnZeroMemberRunSync(TestInfrahubDockerClient, SchemaAnimal): + @pytest.fixture(scope="class") + async def base_dataset(self, client: InfrahubClient, load_schema: None) -> None: + return None + + def test_zero_member_run_keeps_undeletable_member( + self, client_sync: InfrahubClientSync, base_dataset: None + ) -> None: + person_name = "SyncTrackingRetryPerson" + params = {"person_name": person_name} + + with client_sync.start_tracking(params=params, delete_unused_nodes=True) as clt: + person = clt.create(kind=TESTING_PERSON, name=person_name) + person.save(allow_upsert=True) + + cat = client_sync.create(kind=TESTING_CAT, name="SyncTrackingRetryCat", breed="Bengal", owner=person) + cat.save() + + # A zero-member run now attempts the reap; the person's delete is refused. + with ( + pytest.raises(TrackingGroupCleanupError), + client_sync.start_tracking(params=params, delete_unused_nodes=True), + ): + pass + + assert client_sync.mode == InfrahubClientMode.DEFAULT + + group_name = client_sync.group_context._generate_group_name() + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert group.members.peer_ids == [person.id] + + # Once the blocking node is gone, the next zero-member run reaps the person. + client_sync.delete(kind=TESTING_CAT, id=cat.id) + with client_sync.start_tracking(params=params, delete_unused_nodes=True): + pass + + group = client_sync.get(kind="CoreStandardGroup", name__value=group_name, include=["members"]) + assert len(group.members.peers) == 0 + with pytest.raises(NodeNotFoundError): + client_sync.get(kind=TESTING_PERSON, name__value=person_name)