diff --git a/backend/infrahub/core/branch/data_deleter.py b/backend/infrahub/core/branch/data_deleter.py new file mode 100644 index 00000000000..477341bbe01 --- /dev/null +++ b/backend/infrahub/core/branch/data_deleter.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Protocol + +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.constants.database import DatabaseEdgeType +from infrahub.core.query.branch import ( + DeleteBranchAgnosticAttributesQuery, + DeleteBranchAgnosticRelationshipsQuery, + DeleteBranchEdgesQuery, +) +from infrahub.core.query.standard_node import StandardNodeDeleteQuery +from infrahub.exceptions import ValidationError +from infrahub.log import get_logger + +if TYPE_CHECKING: + from infrahub.core.branch.models import Branch + from infrahub.database import InfrahubDatabase + +# The agnostic cleanup batches Nodes, and each one can drag an unbounded number of peer vertices +# into the transaction with it, so its batch is capped low. +MAX_AGNOSTIC_PEER_BATCH_SIZE = 500 + + +@dataclass(frozen=True) +class BranchDeleteResult: + """What a delete attempt actually did. + + `branch_deleted` is false when the branch had already been removed by the time this attempt got + to it, which is how a caller knows not to repeat the work that follows a delete. + """ + + branch_deleted: bool + edges_removed: int + + +class BranchDataDeleterInterface(Protocol): + """The database side of a branch delete.""" + + async def delete(self, branch: Branch) -> BranchDeleteResult: ... + + +class LoggerInterface(Protocol): + """Just enough of a logger for progress reporting.""" + + def info(self, message: str, /) -> Any: ... + + +class BranchDataDeleter: + """Remove a branch, every edge belonging to it, and the vertices that only it kept alive. + + The graph work is split into one bounded query per batch so that no single transaction has to + hold the whole branch in memory. Each query is its own auto-commit transaction, which also means + an interrupted delete can be resumed by running the whole thing again. + """ + + def __init__(self, db: InfrahubDatabase, batch_size: int, log: LoggerInterface | None = None) -> None: + self.db = db + self.batch_size = batch_size + self.log = log or get_logger() + + async def delete(self, branch: Branch) -> BranchDeleteResult: + """Remove the branch's data and then the branch itself. + + Returns whether the Branch object was actually deleted in case multiple processes try to + delete concurrently so the caller can know which delete really succeeded. + + Raises: + ValidationError: When the branch is the default branch or an internal one. + + """ + if branch.is_default: + raise ValidationError(f"Unable to delete {branch.name} it is the default branch.") + if branch.is_global: + raise ValidationError(f"Unable to delete {branch.name} this is an internal branch.") + + if branch.status != BranchStatus.DELETING: + branch.status = BranchStatus.DELETING + await branch.save(db=self.db) + + edges_removed = await self.delete_branch_data(branch_name=branch.name) + + query = await StandardNodeDeleteQuery.init(db=self.db, node=branch) + await query.execute(db=self.db) + branch_deleted = query.stats.get_counter("nodes_deleted") > 0 + + return BranchDeleteResult(branch_deleted=branch_deleted, edges_removed=edges_removed) + + async def delete_branch_data(self, branch_name: str) -> int: + """Remove a branch's data without requiring the branch itself to still exist. + + Returns the number of edges removed, so a caller whose own logging is the only thing the + operator can see is able to report progress. + """ + agnostic_edges_count = await self._delete_agnostic_peers(branch_name=branch_name) + branch_edges_count = await self._delete_edges(branch_name=branch_name) + return agnostic_edges_count + branch_edges_count + + async def _delete_agnostic_peers(self, branch_name: str) -> int: + """Drop the agnostic attributes and relationships of Nodes that exist on no other branch. + + Both queries locate those Nodes through the branch's IS_PART_OF edges, so this has to + finish before the edge deletion starts removing them. Resuming a delete that failed part + way through this stage is safe for the same reason: no IS_PART_OF edge has been touched yet. + + Returns the number of edges removed, which is every edge of the peers detached here, not + only the agnostic ones that led to them. + """ + batch_size = min(self.batch_size, MAX_AGNOSTIC_PEER_BATCH_SIZE) + + relationships_query = await DeleteBranchAgnosticRelationshipsQuery.init( + db=self.db, branch_name=branch_name, batch_size=batch_size + ) + await relationships_query.execute(db=self.db) + + attributes_query = await DeleteBranchAgnosticAttributesQuery.init( + db=self.db, branch_name=branch_name, batch_size=batch_size + ) + await attributes_query.execute(db=self.db) + + edges_removed = relationships_query.stats.get_counter( + "relationships_deleted" + ) + attributes_query.stats.get_counter("relationships_deleted") + if edges_removed: + self.log.info( + f"Deleted agnostic peers of nodes only on branch '{branch_name}', {edges_removed} edge(s) removed" + ) + return edges_removed + + async def _delete_edges(self, branch_name: str) -> int: + edges_removed = 0 + for edge_type in DatabaseEdgeType: + deleted_total = 0 + while True: + # A fresh query per batch: the stats counters accumulate per instance, so a reused + # one would never report zero again and the loop would not end. + query = await DeleteBranchEdgesQuery.init( + db=self.db, branch_name=branch_name, edge_type=edge_type, batch_size=self.batch_size + ) + await query.execute(db=self.db) + deleted = query.deleted_edge_count() + if not deleted: + break + deleted_total += deleted + + if deleted_total: + edges_removed += deleted_total + self.log.info(f"Deleted {deleted_total} {edge_type.value} edge(s) on branch '{branch_name}'") + + return edges_removed diff --git a/backend/infrahub/core/branch/delete_coordinator.py b/backend/infrahub/core/branch/delete_coordinator.py new file mode 100644 index 00000000000..5e66cbd495e --- /dev/null +++ b/backend/infrahub/core/branch/delete_coordinator.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol + +from infrahub.events.branch_action import BranchDeletedEvent +from infrahub.events.models import EventMeta +from infrahub.exceptions import ValidationError +from infrahub.workflows.catalogue import BRANCH_CANCEL_PROPOSED_CHANGES, GIT_REPOSITORIES_DELETE_BRANCH + +if TYPE_CHECKING: + from infrahub.context import InfrahubContext + from infrahub.core.branch.data_deleter import BranchDataDeleterInterface, BranchDeleteResult, LoggerInterface + from infrahub.core.branch.models import Branch + from infrahub.services.adapters.event import InfrahubEventService + from infrahub.services.adapters.workflow import InfrahubWorkflow + + +class DiffFreezerInterface(Protocol): + """Interface for freezing diffs.""" + + async def freeze_diffs_for_branch(self, branch_name: str) -> None: ... + + +class BranchDeleteOrchestrator: + """Delete a branch and do the work that follows from it. + + Holds no database of its own: the deletion is delegated, which is what keeps the ordering and the + post-delete decisions here testable without one. + """ + + def __init__( + self, + data_deleter: BranchDataDeleterInterface, + diff_freezer: DiffFreezerInterface, + event_service: InfrahubEventService, + workflow: InfrahubWorkflow, + log: LoggerInterface, + global_branch: Branch, + delete_git_branch_after_merge: bool, + ) -> None: + self.data_deleter = data_deleter + self.diff_freezer = diff_freezer + self.event_service = event_service + self.workflow = workflow + self.log = log + self.global_branch = global_branch + self.delete_git_branch_after_merge = delete_git_branch_after_merge + + async def delete( + self, + branch: Branch, + context: InfrahubContext, + delete_from_git: bool = False, + proposed_change_id: str | None = None, + ) -> BranchDeleteResult: + """Remove the branch, then cancel its proposed changes, announce it, and drop its Git branch. + + Raises: + ValidationError: When the branch is the default branch or an internal one. + + """ + # Before the freeze, not after: a refused delete has to leave the branch's diffs alone. + if branch.is_default: + raise ValidationError(f"Unable to delete {branch.name} it is the default branch.") + if branch.is_global: + raise ValidationError(f"Unable to delete {branch.name} this is an internal branch.") + + # Freezing has to precede the deletion, which takes away the branch name they are found by. + await self.diff_freezer.freeze_diffs_for_branch(branch_name=branch.name) + + result = await self.data_deleter.delete(branch=branch) + + if result.branch_deleted: + await self.workflow.submit_workflow( + workflow=BRANCH_CANCEL_PROPOSED_CHANGES, context=context, parameters={"branch_name": branch.name} + ) + await self.event_service.send( + event=BranchDeletedEvent( + branch_name=branch.name, + branch_id=str(branch.uuid), + sync_with_git=branch.sync_with_git, + meta=EventMeta.from_context(context=context.to_event_context(), branch=self.global_branch), + proposed_change_id=proposed_change_id, + ) + ) + else: + # Another attempt removed the branch, so the work above is already its responsibility. + self.log.info(f"Branch '{branch.name}' was already deleted") + + # Always execute in case concurrent delete process with delete_from_git=False won the delete race. + if (self.delete_git_branch_after_merge or delete_from_git) and branch.sync_with_git: + await self.workflow.submit_workflow( + workflow=GIT_REPOSITORIES_DELETE_BRANCH, context=context, parameters={"branch": branch.name} + ) + + return result diff --git a/backend/infrahub/core/branch/models.py b/backend/infrahub/core/branch/models.py index 83c68e709e7..f5241bcfdf2 100644 --- a/backend/infrahub/core/branch/models.py +++ b/backend/infrahub/core/branch/models.py @@ -14,7 +14,6 @@ from infrahub.core.query import Query, QueryType from infrahub.core.query.branch import ( BranchNodeGetListQuery, - DeleteBranchRelationshipsQuery, RebaseBranchQuery, ) from infrahub.core.registry import registry @@ -315,17 +314,16 @@ async def create(self, db: InfrahubDatabase, user_id: str = SYSTEM_USER_ID) -> b return await super().create(db=db, user_id=user_id) async def delete(self, db: InfrahubDatabase) -> None: - if self.is_default: - raise ValidationError(f"Unable to delete {self.name} it is the default branch.") - if self.is_global: - raise ValidationError(f"Unable to delete {self.name} this is an internal branch.") + """Not supported on a Branch. - self.status = BranchStatus.DELETING - await self.save(db=db) + The inherited implementation would drop the Branch vertex and silently leave every edge and + vertex belonging to the branch behind, so it is refused rather than overridden. - query = await DeleteBranchRelationshipsQuery.init(db=db, branch_name=self.name) - await query.execute(db=db) - await super().delete(db=db) + Raises: + NotImplementedError: Always. + + """ + raise NotImplementedError("Unable to delete a Branch directly, use BranchDataDeleter instead.") def get_query_filter_relationships( self, rel_labels: list, at: Optional[Timestamp] = None, include_outside_parentheses: bool = False diff --git a/backend/infrahub/core/branch/tasks.py b/backend/infrahub/core/branch/tasks.py index 38258febc3e..35daeaddeaa 100644 --- a/backend/infrahub/core/branch/tasks.py +++ b/backend/infrahub/core/branch/tasks.py @@ -13,6 +13,8 @@ from infrahub.core import registry from infrahub.core.branch import Branch from infrahub.core.branch.creator import BranchCreator +from infrahub.core.branch.data_deleter import BranchDataDeleter +from infrahub.core.branch.delete_coordinator import BranchDeleteOrchestrator from infrahub.core.branch.enums import BranchStatus from infrahub.core.changelog.diff import DiffChangelogCollector, MigrationTracker from infrahub.core.constants import DiffAction, MutationAction @@ -36,7 +38,6 @@ from infrahub.core.validators.tasks import schema_validate_migrations from infrahub.dependencies.registry import get_component_registry from infrahub.events.branch_action import ( - BranchDeletedEvent, BranchMergedEvent, BranchMigratedEvent, BranchRebasedEvent, @@ -54,7 +55,6 @@ BRANCH_MERGE_POST_PROCESS, DIFF_REFRESH_ALL, DIFF_UPDATE, - GIT_REPOSITORIES_DELETE_BRANCH, IPAM_RECONCILIATION, TRIGGER_ARTIFACT_DEFINITION_GENERATE, TRIGGER_GENERATOR_DEFINITION_RUN, @@ -538,37 +538,30 @@ async def delete_branch( ) -> None: await add_tags(branches=[branch], nodes=[proposed_change_id] if proposed_change_id else None) database = await get_database() + workflow = get_workflow() + event_service = await get_event_service() async with database.start_session() as db: - obj = await Branch.get_by_name(db=db, name=str(branch)) + # ignore_deleting=False so that a delete which failed part way through can be run again: + obj = await Branch.get_by_name(db=db, name=str(branch), ignore_deleting=False) component_registry = get_component_registry() diff_repository = await component_registry.get_component(DiffRepository, db=db, branch=obj) - await diff_repository.freeze_diffs_for_branch(branch_name=branch) - - await obj.delete(db=db) - event_context = context.to_event_context() - event = BranchDeletedEvent( - branch_name=branch, - branch_id=str(obj.uuid), - sync_with_git=obj.sync_with_git, - meta=EventMeta.from_context(context=event_context, branch=registry.get_global_branch()), - proposed_change_id=proposed_change_id, - ) - - await get_workflow().submit_workflow( - workflow=BRANCH_CANCEL_PROPOSED_CHANGES, context=context, parameters={"branch_name": branch} + log = get_run_logger() + orchestrator = BranchDeleteOrchestrator( + data_deleter=BranchDataDeleter(db=db, batch_size=config.SETTINGS.database.query_size_limit, log=log), + diff_freezer=diff_repository, + event_service=event_service, + workflow=workflow, + log=log, + global_branch=registry.get_global_branch(), + delete_git_branch_after_merge=config.SETTINGS.git.delete_git_branch_after_merge, ) - - event_service = await get_event_service() - await event_service.send(event=event) - - should_delete_git = (config.SETTINGS.git.delete_git_branch_after_merge or delete_from_git) and obj.sync_with_git - if should_delete_git: - await get_workflow().submit_workflow( - workflow=GIT_REPOSITORIES_DELETE_BRANCH, + await orchestrator.delete( + branch=obj, context=context, - parameters={"branch": branch}, + delete_from_git=delete_from_git, + proposed_change_id=proposed_change_id, ) diff --git a/backend/infrahub/core/graph/__init__.py b/backend/infrahub/core/graph/__init__.py index 947747f069f..6e250218fee 100644 --- a/backend/infrahub/core/graph/__init__.py +++ b/backend/infrahub/core/graph/__init__.py @@ -1 +1 @@ -GRAPH_VERSION = 74 +GRAPH_VERSION = 75 diff --git a/backend/infrahub/core/migrations/graph/m032_cleanup_orphaned_branch_relationships.py b/backend/infrahub/core/migrations/graph/m032_cleanup_orphaned_branch_relationships.py index c77ccfadc39..e43075f18fb 100644 --- a/backend/infrahub/core/migrations/graph/m032_cleanup_orphaned_branch_relationships.py +++ b/backend/infrahub/core/migrations/graph/m032_cleanup_orphaned_branch_relationships.py @@ -2,9 +2,10 @@ from typing import TYPE_CHECKING, Any +from infrahub import config +from infrahub.core.branch.data_deleter import BranchDataDeleter from infrahub.core.migrations.shared import MigrationInput, MigrationResult from infrahub.core.query import Query, QueryType -from infrahub.core.query.branch import DeleteBranchRelationshipsQuery from infrahub.log import get_logger from ..shared import ArbitraryMigration @@ -83,10 +84,10 @@ async def execute(self, migration_input: MigrationInput) -> MigrationResult: log.info(f"Found {len(orphaned_branch_names)} orphaned branch names: {orphaned_branch_names}") + deleter = BranchDataDeleter(db=db, batch_size=config.SETTINGS.database.query_size_limit) for branch_name in orphaned_branch_names: log.info(f"Cleaning up branch '{branch_name}'...") - delete_query = await DeleteBranchRelationshipsQuery.init(db=db, branch_name=branch_name) - await delete_query.execute(db=db) + await deleter.delete_branch_data(branch_name=branch_name) log.info(f"Branch '{branch_name}' cleaned up.") log.info("Deleting orphaned relationships...") diff --git a/backend/infrahub/core/migrations/graph/m075_finish_deleting_branches.py b/backend/infrahub/core/migrations/graph/m075_finish_deleting_branches.py new file mode 100644 index 00000000000..efe96dc0e19 --- /dev/null +++ b/backend/infrahub/core/migrations/graph/m075_finish_deleting_branches.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from infrahub import config +from infrahub.core.branch import Branch +from infrahub.core.branch.data_deleter import BranchDataDeleter, BranchDataDeleterInterface +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.migrations.shared import ArbitraryMigration, MigrationInput, MigrationResult, get_migration_console +from infrahub.core.query import Query, QueryType + +if TYPE_CHECKING: + from infrahub.database import InfrahubDatabase + +console = get_migration_console() + + +class DeletingBranchNamesQuery(Query): + """Find the branches whose delete never finished.""" + + name: str = "deleting_branch_names" + insert_return: bool = False + type: QueryType = QueryType.READ + + async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None: # noqa: ARG002 + query = """ +MATCH (b:Branch) +WHERE b.status = $deleting_status +RETURN b.name AS branch_name + """ + self.params["deleting_status"] = BranchStatus.DELETING.value + self.add_to_query(query) + self.update_return_labels("branch_name") + self.order_by = ["branch_name"] + + def get_branch_names(self) -> list[str]: + return [result.get_as_type(label="branch_name", return_type=str) for result in self.get_results()] + + +class Migration075(ArbitraryMigration): + """Finish deleting branches that a previous branch delete left unfinished. + + A branch delete used to run as one query whose memory use grew with the size of the branch, so + on a large branch it could exhaust the transaction memory pool and fail part way through. The + branch was left with the DELETING status, which hides it from the branch list, and with however + much of its data the failed run had not yet reached. Nothing retried it, and the branch could not + be deleted again because it was no longer possible to look up. + """ + + name: str = "075_finish_deleting_branches" + description: str = "Finish deleting branches whose delete failed part way through." + minimum_version: int = 74 + + async def validate_migration(self, db: InfrahubDatabase) -> MigrationResult: # noqa: ARG002 + return MigrationResult() + + def build_deleter(self, db: InfrahubDatabase) -> BranchDataDeleterInterface: + return BranchDataDeleter(db=db, batch_size=config.SETTINGS.database.query_size_limit) + + async def execute(self, migration_input: MigrationInput) -> MigrationResult: + db = migration_input.db + + try: + names_query = await DeletingBranchNamesQuery.init(db=db) + await names_query.execute(db=db) + branch_names = names_query.get_branch_names() + except Exception as exc: + return MigrationResult(errors=[f"Unable to look up branches in the DELETING state: {exc}"]) + + if not branch_names: + return MigrationResult() + + console.log(f"Found {len(branch_names)} branch(es) left in the DELETING state: {branch_names}") + + # One branch failing must not hide the others: each is deleted in its own right, and the + # names of those that failed are reported so a re-run has something to act on. + errors: list[str] = [] + deleter = self.build_deleter(db=db) + for branch_name in branch_names: + console.log(f"Cleaning up branch '{branch_name}' left in the DELETING state...") + try: + branch = await Branch.get_by_name(db=db, name=branch_name, ignore_deleting=False) + delete_result = await deleter.delete(branch=branch) + except Exception as exc: + console.log(f"Branch '{branch_name}' could not be deleted: {exc}") + errors.append(f"branch '{branch_name}': {exc}") + continue + if delete_result.branch_deleted: + console.log(f"Branch '{branch_name}' deleted, {delete_result.edges_removed} edge(s) removed.") + else: + console.log(f"Branch '{branch_name}' was already gone, {delete_result.edges_removed} edge(s) removed.") + + return MigrationResult(errors=errors) diff --git a/backend/infrahub/core/query/branch.py b/backend/infrahub/core/query/branch.py index d99ff4d02b1..6a13a11c266 100644 --- a/backend/infrahub/core/query/branch.py +++ b/backend/infrahub/core/query/branch.py @@ -10,83 +10,137 @@ from infrahub.core.timestamp import Timestamp if TYPE_CHECKING: + from infrahub.core.constants.database import DatabaseEdgeType from infrahub.database import InfrahubDatabase -class DeleteBranchRelationshipsQuery(Query): - name: str = "delete_branch_relationships" +class DeleteBranchAgnosticRelationshipsQuery(Query): + """Delete the agnostic Relationship vertices attached to Nodes that only exist on this branch. + + Must run before any IS_PART_OF edge of the branch is deleted: the branch-only determination + reads those edges, so once they are gone the affected Nodes can no longer be found and their + agnostic peers leak. + """ + + name: str = "delete_branch_agnostic_relationships" insert_return: bool = False + insert_limit: bool = False type: QueryType = QueryType.WRITE - def __init__(self, branch_name: str, **kwargs: Any) -> None: + def __init__(self, branch_name: str, batch_size: int, **kwargs: Any) -> None: self.branch_name = branch_name + self.batch_size = batch_size super().__init__(**kwargs) async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None: # noqa: ARG002 query = """ -// -------------- -// for every Node that only exists on this branch (it's about to be deleted), -// find any agnostic relationships or attributes connected to the Node and delete them -// -------------- -OPTIONAL MATCH (:Root)<-[e:IS_PART_OF {status: "active"}]-(n:Node) +MATCH (:Root)<-[e:IS_PART_OF {status: "active"}]-(n:Node) WHERE e.branch = $branch_name -// does the node only exist on this branch? -CALL (n) { - OPTIONAL MATCH (n)-[ipo:IS_PART_OF {status: "active"}]->(:Root) +AND NOT EXISTS { + MATCH (n)-[ipo:IS_PART_OF {status: "active"}]->(:Root) WHERE ipo.branch <> $branch_name - LIMIT 1 - RETURN ipo IS NOT NULL AS node_exists_on_other_branch } -// if so, delete any linked agnostic relationships or attributes -CALL (n, node_exists_on_other_branch) { - WITH n, node_exists_on_other_branch - WHERE node_exists_on_other_branch = FALSE - OPTIONAL MATCH (n)-[:IS_RELATED {branch: $global_branch_name}]-(rel:Relationship) +CALL (n) { + MATCH (n)-[:IS_RELATED {branch: $global_branch_name}]-(rel:Relationship) DETACH DELETE rel -} IN TRANSACTIONS OF 500 ROWS -CALL (n, node_exists_on_other_branch) { - WITH n, node_exists_on_other_branch - WHERE node_exists_on_other_branch = FALSE - OPTIONAL MATCH (n)-[:HAS_ATTRIBUTE {branch: $global_branch_name}]-(attr:Attribute) - DETACH DELETE attr -} IN TRANSACTIONS OF 500 ROWS +} IN TRANSACTIONS OF %(batch_size)s ROWS + """ % {"batch_size": self.batch_size} + self.params["branch_name"] = self.branch_name + self.params["global_branch_name"] = GLOBAL_BRANCH_NAME + self.add_to_query(query) -// reduce the results to a single row -WITH 1 AS one -LIMIT 1 -// -------------- -// for every edge on this branch, delete it -// -------------- -MATCH (s)-[r]->(d) -WHERE r.branch = $branch_name -CALL (r) { - DELETE r -} IN TRANSACTIONS OF 500 ROWS +class DeleteBranchAgnosticAttributesQuery(Query): + """Delete the agnostic Attribute vertices attached to Nodes that only exist on this branch. -// -------------- -// get the database IDs of every vertex linked to a deleted edge -// -------------- -WITH DISTINCT elementId(s) AS s_id, elementId(d) AS d_id -WITH collect(s_id) + collect(d_id) AS vertex_ids -UNWIND vertex_ids AS vertex_id + Carries the same ordering requirement as the agnostic Relationship query. + """ -// -------------- -// delete any vertices that are now orphaned -// -------------- -CALL (vertex_id) { - MATCH (n) - WHERE elementId(n) = vertex_id - AND NOT exists((n)--()) - DELETE n -} IN TRANSACTIONS OF 500 ROWS - """ + name: str = "delete_branch_agnostic_attributes" + insert_return: bool = False + insert_limit: bool = False + + type: QueryType = QueryType.WRITE + + def __init__(self, branch_name: str, batch_size: int, **kwargs: Any) -> None: + self.branch_name = branch_name + self.batch_size = batch_size + super().__init__(**kwargs) + + async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None: # noqa: ARG002 + query = """ +MATCH (:Root)<-[e:IS_PART_OF {status: "active"}]-(n:Node) +WHERE e.branch = $branch_name +AND NOT EXISTS { + MATCH (n)-[ipo:IS_PART_OF {status: "active"}]->(:Root) + WHERE ipo.branch <> $branch_name +} +CALL (n) { + MATCH (n)-[:HAS_ATTRIBUTE {branch: $global_branch_name}]-(attr:Attribute) + DETACH DELETE attr +} IN TRANSACTIONS OF %(batch_size)s ROWS + """ % {"batch_size": self.batch_size} self.params["branch_name"] = self.branch_name self.params["global_branch_name"] = GLOBAL_BRANCH_NAME self.add_to_query(query) +class DeleteBranchEdgesQuery(Query): + """Delete one batch of edges of a single type belonging to a branch, plus any vertex left bare. + + Every edge on the branch is removed by this query's DELETE, and both endpoints of each one are + then re-examined, so a vertex is examined once per edge it had. The batch that removes its last + edge is therefore the one that sees it at degree zero and deletes it. Nothing can be stranded, + because no edge is ever removed by any other means -- which is why the vertices need no separate + sweep afterwards, and why the vertex delete must not be a DETACH DELETE. A DETACH DELETE would + take out the branch edges the vertex still had, and those edges would then never reach a batch + of their own, leaving the vertices on their far side unexamined and orphaned. + + The DISTINCT is what makes this sound: it forces the whole batch's edge deletes to complete + before the first vertex is examined, so degree zero means degree zero. + + Naming the edge type is what lets the `branch` range index be used for the match; the type + cannot be a query parameter, so it is interpolated from the closed DatabaseEdgeType enum. + + Run repeatedly until it stops deleting edges. + """ + + name: str = "delete_branch_edges" + insert_return: bool = False + insert_limit: bool = False + + type: QueryType = QueryType.WRITE + + def __init__(self, branch_name: str, edge_type: DatabaseEdgeType, batch_size: int, **kwargs: Any) -> None: + self.branch_name = branch_name + self.edge_type = edge_type + self.batch_size = batch_size + super().__init__(**kwargs) + + async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None: # noqa: ARG002 + query = """ +MATCH (s)-[r:%(edge_type)s]->(d) +WHERE r.branch = $branch_name +WITH s, r, d +LIMIT $batch_size +DELETE r + +WITH s, d +UNWIND [s, d] AS v +WITH DISTINCT v +WHERE NOT v:Root +AND NOT EXISTS { MATCH (v)--() } +DELETE v + """ % {"edge_type": self.edge_type.value} + self.params["branch_name"] = self.branch_name + self.params["batch_size"] = self.batch_size + self.add_to_query(query) + + def deleted_edge_count(self) -> int: + return self.stats.get_counter("relationships_deleted") + + class RebaseBranchQuery(Query): """Rebase a branch onto the default branch by updating edge timestamps. diff --git a/backend/infrahub/graphql/mutations/branch.py b/backend/infrahub/graphql/mutations/branch.py index a07e23ea089..c98838d0d00 100644 --- a/backend/infrahub/graphql/mutations/branch.py +++ b/backend/infrahub/graphql/mutations/branch.py @@ -148,7 +148,8 @@ async def mutate( wait_until_completion: bool = True, ) -> Self: graphql_context: GraphqlContext = info.context - obj = await Branch.get_by_name(db=graphql_context.db, name=str(data.name)) + # ignore_deleting=False so a delete that failed part way through can be retried: the first + obj = await Branch.get_by_name(db=graphql_context.db, name=str(data.name), ignore_deleting=False) await apply_external_context(graphql_context=graphql_context, context_input=context) parameters = { diff --git a/backend/tests/component/core/migrations/graph/test_024.py b/backend/tests/component/core/migrations/graph/test_024.py index 9beae34f476..eb022e2a513 100644 --- a/backend/tests/component/core/migrations/graph/test_024.py +++ b/backend/tests/component/core/migrations/graph/test_024.py @@ -3,6 +3,7 @@ import pytest from infrahub.core import registry +from infrahub.core.branch.data_deleter import BranchDataDeleter from infrahub.core.branch.models import Branch from infrahub.core.constants import RelationshipHierarchyDirection from infrahub.core.diff.coordinator import DiffCoordinator @@ -42,7 +43,7 @@ async def test_hierarchy_fix_migration( await diff_merger.merge_graph(at=at) # delete the branch - await branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=branch) # remove the hierarchy property on main query = """ diff --git a/backend/tests/component/core/migrations/graph/test_067_freeze_orphaned_branch_tracking_diffs.py b/backend/tests/component/core/migrations/graph/test_067_freeze_orphaned_branch_tracking_diffs.py index d4d06e51729..d68a2888877 100644 --- a/backend/tests/component/core/migrations/graph/test_067_freeze_orphaned_branch_tracking_diffs.py +++ b/backend/tests/component/core/migrations/graph/test_067_freeze_orphaned_branch_tracking_diffs.py @@ -20,6 +20,7 @@ from infrahub_sdk.timestamp import Timestamp from infrahub.core.branch import Branch +from infrahub.core.branch.data_deleter import BranchDataDeleter from infrahub.core.branch.enums import BranchStatus from infrahub.core.diff.model.path import BranchTrackingId, EnrichedDiffs, FrozenTrackingId from infrahub.core.diff.repository.repository import DiffRepository @@ -143,7 +144,7 @@ async def test_migration_067( from_time=deleted_from, to_time=deleted_from.add(seconds=60), ) - await deleted_branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=deleted_branch) expectations.append( DiffExpectation( name="deleted branch frozen", @@ -184,7 +185,7 @@ async def test_migration_067( from_time=v1_from, to_time=v1_from.add(seconds=60), ) - await reused_v1.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=reused_v1) reused_v2 = await create_branch(db=db, branch_name=reused_name) v2_from = Timestamp(reused_v2.get_branched_from()) reused_v2_diff, reused_v2_base = await self._create_diff_pair( @@ -220,7 +221,7 @@ async def test_migration_067( to_time=frozen_from.add(seconds=60), ) await diff_repository.freeze_diffs_for_branch(branch_name=frozen_branch.name) - await frozen_branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=frozen_branch) expectations.append( DiffExpectation( name="already frozen unchanged", @@ -243,7 +244,7 @@ async def test_migration_067( lifecycle_v1.status = BranchStatus.MERGED await lifecycle_v1.save(db=db) await diff_repository.mark_tracking_ids_merged(tracking_ids=[BranchTrackingId(name=lifecycle_name)]) - await lifecycle_v1.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=lifecycle_v1) lifecycle_v2 = await create_branch(db=db, branch_name=lifecycle_name) lc_v2_from = Timestamp(lifecycle_v2.get_branched_from()) lc_v2_diff, lc_v2_base = await self._create_diff_pair( diff --git a/backend/tests/component/core/migrations/graph/test_075_finish_deleting_branches.py b/backend/tests/component/core/migrations/graph/test_075_finish_deleting_branches.py new file mode 100644 index 00000000000..f4b2e99bf96 --- /dev/null +++ b/backend/tests/component/core/migrations/graph/test_075_finish_deleting_branches.py @@ -0,0 +1,184 @@ +import pytest + +from infrahub.core.branch.data_deleter import BranchDataDeleter, BranchDataDeleterInterface, BranchDeleteResult +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.branch.models import Branch +from infrahub.core.initialization import create_branch +from infrahub.core.manager import NodeManager +from infrahub.core.migrations.graph.m075_finish_deleting_branches import Migration075 +from infrahub.core.migrations.shared import MigrationInput +from infrahub.core.node import Node +from infrahub.database import InfrahubDatabase +from infrahub.exceptions import BranchNotFoundError, NodeNotFoundError + + +class FailingBranchDeleter: + """Deletes for real, except for one branch, where it raises instead. + + Delegating for the others is what lets a test tell "the loop carried on" apart from "the loop + called delete again but nothing was reclaimed". + """ + + def __init__(self, deleter: BranchDataDeleter, failing_branch_name: str) -> None: + self.deleter = deleter + self.failing_branch_name = failing_branch_name + self.attempted: list[str] = [] + + async def delete(self, branch: Branch) -> BranchDeleteResult: + self.attempted.append(branch.name) + if branch.name == self.failing_branch_name: + raise ValueError("FAILED") + return await self.deleter.delete(branch=branch) + + +class Migration075WithFailingDeleter(Migration075): + """Migration075 wired to a deleter that fails on one nominated branch.""" + + failing_branch_name: str = "" + deleter: FailingBranchDeleter | None = None + + model_config = {"arbitrary_types_allowed": True} + + def build_deleter(self, db: InfrahubDatabase) -> BranchDataDeleterInterface: + self.deleter = FailingBranchDeleter( + deleter=BranchDataDeleter(db=db, batch_size=5), failing_branch_name=self.failing_branch_name + ) + return self.deleter + + +async def _branch_edge_count(db: InfrahubDatabase, branch_name: str) -> int: + results = await db.execute_query( + query="MATCH ()-[e]->() WHERE e.branch = $branch_name RETURN count(e) AS count", + params={"branch_name": branch_name}, + ) + return results[0]["count"] + + +async def _add_tag(db: InfrahubDatabase, branch: Branch, name: str) -> Node: + node = await Node.init(db=db, branch=branch, schema="BuiltinTag") + await node.new(db=db, name=name) + await node.save(db=db) + return node + + +async def test_migration_075(db: InfrahubDatabase, default_branch: Branch, person_tag_schema: None) -> None: + """A branch abandoned in DELETING loses every edge, while the other branches keep all of theirs.""" + healthy_branch = await create_branch(db=db, branch_name="healthy-branch") + stalled_branch = await create_branch(db=db, branch_name="stalled-branch") + + node_on_main = await _add_tag(db=db, branch=default_branch, name="node-on-main") + node_on_healthy = await _add_tag(db=db, branch=healthy_branch, name="node-on-healthy-branch") + node_on_stalled = await _add_tag(db=db, branch=stalled_branch, name="node-on-stalled-branch") + + # Reproduce what a failed delete leaves behind: the status set, the data still present. + stalled_branch.status = BranchStatus.DELETING + await stalled_branch.save(db=db) + + edges_before = { + name: await _branch_edge_count(db=db, branch_name=name) + for name in (default_branch.name, healthy_branch.name, stalled_branch.name) + } + # Every branch has to start with edges, otherwise the assertions below prove nothing. + assert all(count > 0 for count in edges_before.values()), edges_before + # Likewise the stalled branch's node has to be readable to begin with, so that it disappearing + # afterwards is attributable to the migration. + node_before = await NodeManager.get_one(db=db, branch=stalled_branch, id=node_on_stalled.id) + assert node_before is not None + + migration = Migration075() + execution_result = await migration.execute(migration_input=MigrationInput(db=db)) + assert not execution_result.errors + + validation_result = await migration.validate_migration(db=db) + assert not validation_result.errors + + edges_after = { + name: await _branch_edge_count(db=db, branch_name=name) + for name in (default_branch.name, healthy_branch.name, stalled_branch.name) + } + + # The abandoned branch is emptied; the untouched branches keep exactly what they had. + assert edges_after == { + default_branch.name: edges_before[default_branch.name], + healthy_branch.name: edges_before[healthy_branch.name], + stalled_branch.name: 0, + } + + # The branch node is gone too, along with the data that hung off it. + with pytest.raises(BranchNotFoundError): + await Branch.get_by_name(db=db, name=stalled_branch.name, ignore_deleting=False) + with pytest.raises(NodeNotFoundError): + await NodeManager.get_one(db=db, branch=stalled_branch, id=node_on_stalled.id, raise_on_error=True) + + # The surviving branches are still usable, not merely still edged. + reloaded_healthy = await Branch.get_by_name(db=db, name=healthy_branch.name) + assert reloaded_healthy.status == BranchStatus.OPEN + retrieved_on_healthy = await NodeManager.get_one(db=db, branch=healthy_branch, id=node_on_healthy.id) + assert retrieved_on_healthy is not None + assert retrieved_on_healthy.get_attribute("name").value == "node-on-healthy-branch" + retrieved_on_main = await NodeManager.get_one(db=db, branch=default_branch, id=node_on_main.id) + assert retrieved_on_main is not None + assert retrieved_on_main.get_attribute("name").value == "node-on-main" + + +async def test_migration_075_no_deleting_branches( + db: InfrahubDatabase, default_branch: Branch, person_tag_schema: None +) -> None: + """With nothing to finish, every branch keeps every edge.""" + branch = await create_branch(db=db, branch_name="untouched-branch") + await _add_tag(db=db, branch=default_branch, name="node-on-main") + await _add_tag(db=db, branch=branch, name="node-on-untouched-branch") + + edges_before = { + name: await _branch_edge_count(db=db, branch_name=name) for name in (default_branch.name, branch.name) + } + assert all(count > 0 for count in edges_before.values()), edges_before + + migration = Migration075() + execution_result = await migration.execute(migration_input=MigrationInput(db=db)) + assert not execution_result.errors + + edges_after = { + name: await _branch_edge_count(db=db, branch_name=name) for name in (default_branch.name, branch.name) + } + assert edges_after == edges_before + + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + + +async def test_migration_075_one_failing_branch_does_not_block_the_others( + db: InfrahubDatabase, default_branch: Branch, person_tag_schema: None +) -> None: + """A branch that cannot be deleted is reported by name; the rest are still reclaimed.""" + branch_names = ["stalled-a", "stalled-b", "stalled-c"] + for branch_name in branch_names: + branch = await create_branch(db=db, branch_name=branch_name) + await _add_tag(db=db, branch=branch, name=f"node-on-{branch_name}") + branch.status = BranchStatus.DELETING + await branch.save(db=db) + + edges_before = {name: await _branch_edge_count(db=db, branch_name=name) for name in branch_names} + assert all(count > 0 for count in edges_before.values()), edges_before + + migration = Migration075WithFailingDeleter(failing_branch_name="stalled-b") + execution_result = await migration.execute(migration_input=MigrationInput(db=db)) + + # The failure is surfaced against the branch it belongs to, not as a bare exception string. + assert execution_result.errors == ["branch 'stalled-b': FAILED"] + + # Every branch was attempted, including the ones queued behind the failure. + assert migration.deleter is not None + assert migration.deleter.attempted == branch_names + + # The other two are genuinely reclaimed; the failed one keeps everything it had. + edges_after = {name: await _branch_edge_count(db=db, branch_name=name) for name in branch_names} + assert edges_after == {"stalled-a": 0, "stalled-b": edges_before["stalled-b"], "stalled-c": 0} + + for deleted_name in ("stalled-a", "stalled-c"): + with pytest.raises(BranchNotFoundError): + await Branch.get_by_name(db=db, name=deleted_name, ignore_deleting=False) + + # The failed branch is left exactly as it was, so a re-run can pick it up. + still_stalled = await Branch.get_by_name(db=db, name="stalled-b", ignore_deleting=False) + assert still_stalled.status == BranchStatus.DELETING diff --git a/backend/tests/component/core/resource_manager/test_number_pool_query.py b/backend/tests/component/core/resource_manager/test_number_pool_query.py index d06500c3175..01bd9dccf94 100644 --- a/backend/tests/component/core/resource_manager/test_number_pool_query.py +++ b/backend/tests/component/core/resource_manager/test_number_pool_query.py @@ -4,6 +4,7 @@ from infrahub.core import registry from infrahub.core.branch import Branch +from infrahub.core.branch.data_deleter import BranchDataDeleter from infrahub.core.constants import InfrahubKind from infrahub.core.diff.coordinator import DiffCoordinator from infrahub.core.diff.data_check_synchronizer import DiffDataCheckSynchronizer @@ -128,7 +129,7 @@ async def test_NumberPoolGetUsed( assert await get_used_numbers_in_pool(db=db, pool=incident_pool, branch=default_branch) == [1, 2, 3, 4, 5, 6, 7] # Delete the branch and validate that the numbers allocated previously are available - await branch2.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=branch2) assert await get_used_numbers_in_pool(db=db, pool=incident_pool, branch=default_branch) == [1, 2, 3] # Create a new branch and add more incidents @@ -138,7 +139,7 @@ async def test_NumberPoolGetUsed( assert await get_used_numbers_in_pool(db=db, pool=incident_pool, branch=default_branch) == [1, 2, 3, 4, 5, 6] # Delete the branch and validate that the numbers allocated previously are available - await branch3.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=branch3) assert await get_used_numbers_in_pool(db=db, pool=incident_pool, branch=default_branch) == [1, 2, 3] # Delete nodes in main and ensure the numbers are reallocated diff --git a/backend/tests/component/core/test_branch.py b/backend/tests/component/core/test_branch.py index 0182f4caca8..709654af98f 100644 --- a/backend/tests/component/core/test_branch.py +++ b/backend/tests/component/core/test_branch.py @@ -5,6 +5,8 @@ from pydantic import ValidationError as PydanticValidationError from infrahub.core.branch import Branch +from infrahub.core.branch.data_deleter import BranchDataDeleter +from infrahub.core.branch.enums import BranchStatus from infrahub.core.constants import GLOBAL_BRANCH_NAME from infrahub.core.diff.coordinator import DiffCoordinator from infrahub.core.diff.data_check_synchronizer import DiffDataCheckSynchronizer @@ -295,6 +297,56 @@ async def test_is_isolated(db: InfrahubDatabase, base_dataset_02: dict) -> None: assert cars[0].name.value == "volt" +async def test_branch_delete_method_is_refused(db: InfrahubDatabase, default_branch: Branch) -> None: + """Deleting through the model would drop the Branch node and orphan all of its data.""" + branch = await create_branch(branch_name="refuse-me", db=db) + + with pytest.raises( + NotImplementedError, match=r"^Unable to delete a Branch directly, use BranchDataDeleter instead\.$" + ): + await branch.delete(db=db) + + # The branch is untouched: still listed, still OPEN. + reloaded = await Branch.get_by_name(name="refuse-me", db=db) + assert reloaded.status == BranchStatus.OPEN + + +async def test_branch_deleter_refuses_default_and_global_branches(db: InfrahubDatabase, default_branch: Branch) -> None: + """The guards that used to live on Branch.delete still apply on the deleter.""" + deleter = BranchDataDeleter(db=db, batch_size=5) + + with pytest.raises(ValidationError, match=r"Unable to delete .* it is the default branch\."): + await deleter.delete(branch=default_branch) + + global_branch = registry.get_global_branch() + with pytest.raises(ValidationError, match=r"Unable to delete .* this is an internal branch\."): + await deleter.delete(branch=global_branch) + + # Neither branch was altered before the guard fired. + assert (await Branch.get_by_name(name=default_branch.name, db=db)).status == BranchStatus.OPEN + + +async def test_branch_deleter_reports_who_removed_the_branch( + db: InfrahubDatabase, default_branch: Branch, repos_in_main: dict, car_person_schema: SchemaBranch +) -> None: + """Only the attempt that removes the branch reports having done so.""" + branch = await create_branch(branch_name="claim-me", db=db) + person = await Node.init(schema="TestPerson", branch=branch.name, db=db) + await person.new(name="Bobby", height=175, db=db) + await person.save(db=db) + + deleter = BranchDataDeleter(db=db, batch_size=5) + + first = await deleter.delete(branch=branch) + assert first.branch_deleted is True + assert first.edges_removed > 0 + + # Deleting the same branch again is harmless, but must not claim to have done it. + second = await deleter.delete(branch=branch) + assert second.branch_deleted is False + assert second.edges_removed == 0 + + async def test_delete_branch( db: InfrahubDatabase, default_branch: Branch, repos_in_main: dict, car_person_schema: SchemaBranch ) -> None: @@ -314,7 +366,9 @@ async def test_delete_branch( params = {"branch_name": branch_name} pre_delete = await db.execute_query(query=relationship_query, params=params) - await branch.delete(db=db) + # A batch size well below the number of edges on the branch, so the batching loop has to run + # more than once to finish. + await BranchDataDeleter(db=db, batch_size=5).delete(branch=branch) post_delete = await db.execute_query(query=relationship_query, params=params) assert branch.id == found.id @@ -362,7 +416,7 @@ async def test_delete_branch_with_agnostic_attrs_and_rels( rel_uuid = agnostic_rel.id # Delete the branch - await branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=branch) # Verify the branch is deleted with pytest.raises(BranchNotFoundError): @@ -425,7 +479,7 @@ async def test_delete_branch_after_merge_preserves_node( assert device_on_main.get_attribute("serial_number").value == "SN-67890" # Delete the branch - await branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=branch) # Verify the branch is deleted with pytest.raises(BranchNotFoundError): diff --git a/backend/tests/component/graphql/diff/test_diff_tree_terminal_branch.py b/backend/tests/component/graphql/diff/test_diff_tree_terminal_branch.py index 99427afe1fe..e38684361de 100644 --- a/backend/tests/component/graphql/diff/test_diff_tree_terminal_branch.py +++ b/backend/tests/component/graphql/diff/test_diff_tree_terminal_branch.py @@ -5,6 +5,7 @@ from infrahub.core import registry from infrahub.core.branch import Branch +from infrahub.core.branch.data_deleter import BranchDataDeleter from infrahub.core.branch.enums import BranchStatus from infrahub.core.diff.coordinator import DiffCoordinator from infrahub.core.diff.data_check_synchronizer import DiffDataCheckSynchronizer @@ -390,7 +391,7 @@ async def deleted_branch( merged_branch: Branch, ) -> Branch: """Delete the branch and remove it from registry.""" - await merged_branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=merged_branch) registry.branch.pop(merged_branch.name, None) return merged_branch diff --git a/backend/tests/component/graphql/mutations/test_branch.py b/backend/tests/component/graphql/mutations/test_branch.py index d15144c7490..0658b2a24ea 100644 --- a/backend/tests/component/graphql/mutations/test_branch.py +++ b/backend/tests/component/graphql/mutations/test_branch.py @@ -554,6 +554,37 @@ async def test_branch_delete_own_branch_succeeds( assert delete_result.data["BranchDelete"]["ok"] is True +async def test_branch_delete_retries_a_branch_left_deleting( + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + first_account: Node, + session_first_account: AccountSession, + local_services: InfrahubServices, +) -> None: + """A delete that failed part way through can be retried. + + The first attempt leaves the branch in DELETING, which the default branch lookup hides. Without + accepting that status the retry would report the branch as missing and its data would be unreachable. + """ + branch = await _create_branch(branch_name="stuck-deleting-branch", db=db, owner=first_account) + branch.status = BranchStatus.DELETING + await branch.save(db=db) + + with patch.object(local_services.workflow, "execute_workflow", new=AsyncMock(return_value=None)): + delete_result = await graphql_mutation( + query='mutation { BranchDelete(data: { name: "stuck-deleting-branch" }) { ok } }', + db=db, + branch=default_branch, + account_session=session_first_account, + service=local_services, + ) + + assert delete_result.errors is None + assert delete_result.data + assert delete_result.data["BranchDelete"]["ok"] is True + + async def test_branch_delete_others_branch_denied( db: InfrahubDatabase, default_branch: Branch, diff --git a/backend/tests/integration/diff/test_diff_update.py b/backend/tests/integration/diff/test_diff_update.py index 4d03f654d1e..7ae7197132d 100644 --- a/backend/tests/integration/diff/test_diff_update.py +++ b/backend/tests/integration/diff/test_diff_update.py @@ -7,6 +7,7 @@ from infrahub_sdk.exceptions import GraphQLError from infrahub.core import registry +from infrahub.core.branch.data_deleter import BranchDataDeleter from infrahub.core.constants import NULL_VALUE, BranchConflictKeep, DiffAction, InfrahubKind from infrahub.core.constants.database import DatabaseEdgeType from infrahub.core.diff.model.path import BranchTrackingId, ConflictSelection, EnrichedDiffRoot @@ -215,7 +216,7 @@ async def diff_on_deleted_branch( diff = await self.get_branch_diff(db=db, branch=deleted_branch) assert len(diff.nodes) == 1 - await deleted_branch.delete(db=db) + await BranchDataDeleter(db=db, batch_size=5).delete(branch=deleted_branch) return diff @staticmethod diff --git a/backend/tests/unit/core/branch/test_delete_coordinator.py b/backend/tests/unit/core/branch/test_delete_coordinator.py new file mode 100644 index 00000000000..1e9261e0943 --- /dev/null +++ b/backend/tests/unit/core/branch/test_delete_coordinator.py @@ -0,0 +1,151 @@ +from uuid import uuid4 + +import pytest + +from infrahub.auth.session import AccountSession +from infrahub.auth.types import AuthType +from infrahub.context import BranchContext, InfrahubContext +from infrahub.core.branch import Branch +from infrahub.core.branch.data_deleter import BranchDeleteResult +from infrahub.core.branch.delete_coordinator import BranchDeleteOrchestrator +from infrahub.core.constants import GLOBAL_BRANCH_NAME +from infrahub.events.branch_action import BranchDeletedEvent +from infrahub.workflows.catalogue import BRANCH_CANCEL_PROPOSED_CHANGES, GIT_REPOSITORIES_DELETE_BRANCH +from tests.adapters.event import MemoryInfrahubEvent +from tests.adapters.log import FakeLogger +from tests.adapters.workflow import WorkflowRecorder + + +class RecordingDataDeleter: + """Reports a fixed outcome and remembers which branches it was asked to delete.""" + + def __init__(self, result: BranchDeleteResult) -> None: + self.result = result + self.deleted: list[str] = [] + + async def delete(self, branch: Branch) -> BranchDeleteResult: + self.deleted.append(branch.name) + return self.result + + +class RecordingDiffFreezer: + def __init__(self) -> None: + self.frozen: list[str] = [] + + async def freeze_diffs_for_branch(self, branch_name: str) -> None: + self.frozen.append(branch_name) + + +@pytest.fixture +def context() -> InfrahubContext: + return InfrahubContext( + account=AccountSession(account_id=str(uuid4()), auth_type=AuthType.NONE), + branch=BranchContext(name="main", id="placeholder"), + ) + + +def _build( + *, + branch_deleted: bool, + delete_git_branch_after_merge: bool = False, +) -> tuple[ + BranchDeleteOrchestrator, + RecordingDataDeleter, + RecordingDiffFreezer, + WorkflowRecorder, + MemoryInfrahubEvent, + FakeLogger, +]: + data_deleter = RecordingDataDeleter( + result=BranchDeleteResult(branch_deleted=branch_deleted, edges_removed=7 if branch_deleted else 0) + ) + diff_freezer = RecordingDiffFreezer() + workflow = WorkflowRecorder() + events = MemoryInfrahubEvent() + log = FakeLogger() + orchestrator = BranchDeleteOrchestrator( + data_deleter=data_deleter, + diff_freezer=diff_freezer, + event_service=events, + workflow=workflow, + log=log, + global_branch=Branch(name=GLOBAL_BRANCH_NAME, is_global=True, uuid=uuid4()), + delete_git_branch_after_merge=delete_git_branch_after_merge, + ) + return orchestrator, data_deleter, diff_freezer, workflow, events, log + + +def _branch(name: str = "some-branch", sync_with_git: bool = True) -> Branch: + return Branch(name=name, sync_with_git=sync_with_git, uuid=uuid4()) + + +async def test_delete_runs_post_delete_work(context: InfrahubContext) -> None: + """The attempt that removes the branch cancels its proposed changes and announces it.""" + orchestrator, data_deleter, diff_freezer, workflow, events, _ = _build(branch_deleted=True) + branch = _branch() + + result = await orchestrator.delete(branch=branch, context=context, delete_from_git=True) + + assert result == BranchDeleteResult(branch_deleted=True, edges_removed=7) + # The diffs are frozen before the delete, since they are found by a branch name it takes away. + assert diff_freezer.frozen == [branch.name] + assert data_deleter.deleted == [branch.name] + assert [type(event) for event in events.events] == [BranchDeletedEvent] + assert workflow.get_submit_calls_for(BRANCH_CANCEL_PROPOSED_CHANGES) == [ + {"workflow": BRANCH_CANCEL_PROPOSED_CHANGES, "parameters": {"branch_name": branch.name}} + ] + assert workflow.get_submit_calls_for(GIT_REPOSITORIES_DELETE_BRANCH) == [ + {"workflow": GIT_REPOSITORIES_DELETE_BRANCH, "parameters": {"branch": branch.name}} + ] + + +async def test_delete_skips_post_delete_work_when_another_attempt_won(context: InfrahubContext) -> None: + """An attempt that removed nothing must not repeat what belongs to the one that did.""" + orchestrator, _, _, workflow, events, log = _build(branch_deleted=False) + branch = _branch() + + result = await orchestrator.delete(branch=branch, context=context, delete_from_git=False) + + assert result.branch_deleted is False + assert events.events == [] + assert workflow.submit_calls == [] + assert log.info_logs == [f"Branch '{branch.name}' was already deleted"] + + +async def test_delete_from_git_survives_losing_the_race(context: InfrahubContext) -> None: + """The attempt that won may not have been asked to remove the Git branch.""" + orchestrator, _, _, workflow, events, _ = _build(branch_deleted=False) + branch = _branch() + + await orchestrator.delete(branch=branch, context=context, delete_from_git=True) + + assert workflow.get_submit_calls_for(GIT_REPOSITORIES_DELETE_BRANCH) == [ + {"workflow": GIT_REPOSITORIES_DELETE_BRANCH, "parameters": {"branch": branch.name}} + ] + # Still nothing that belongs to the winning attempt. + assert events.events == [] + assert workflow.get_submit_calls_for(BRANCH_CANCEL_PROPOSED_CHANGES) == [] + + +async def test_delete_from_git_is_ignored_for_a_branch_that_does_not_track_git(context: InfrahubContext) -> None: + orchestrator, _, _, workflow, events, _ = _build(branch_deleted=True) + branch = _branch(sync_with_git=False) + + await orchestrator.delete(branch=branch, context=context, delete_from_git=True) + + assert workflow.get_submit_calls_for(GIT_REPOSITORIES_DELETE_BRANCH) == [] + # The rest of the post-delete work still ran. + assert [type(event) for event in events.events] == [BranchDeletedEvent] + + +async def test_delete_git_branch_after_merge_setting_deletes_without_an_explicit_request( + context: InfrahubContext, +) -> None: + orchestrator, _, _, workflow, _, _ = _build(branch_deleted=True, delete_git_branch_after_merge=True) + branch = _branch() + + await orchestrator.delete(branch=branch, context=context, delete_from_git=False) + + assert workflow.get_submit_calls_for(GIT_REPOSITORIES_DELETE_BRANCH) == [ + {"workflow": GIT_REPOSITORIES_DELETE_BRANCH, "parameters": {"branch": branch.name}} + ] diff --git a/changelog/9889.fixed.md b/changelog/9889.fixed.md new file mode 100644 index 00000000000..0d3e734e99e --- /dev/null +++ b/changelog/9889.fixed.md @@ -0,0 +1 @@ +Fixed deleting a large branch failing with a database out-of-memory error and leaving the branch and its data behind. Branches left behind by an earlier failure are now cleaned up on upgrade.