reset proposed change status on unexpected merge failure (#9164)

* reset proposed change state on unexpected merge failure

* add changelog

* improve error handling

Author: ajtmccarty

backend/infrahub/proposed_change/tasks.py

@@ -226,16 +226,27 @@ async def merge_proposed_change(
                         )
 
         log.info("Proposed change is eligible to be merged")
+
+        merge_succeeded = False
         try:
             await merge_branch(branch=source_branch.name, context=context, proposed_change_id=proposed_change_id)
+            merge_succeeded = True
         except MergeFailedError as exc:
-            await _proposed_change_transition_state(
-                proposed_change=proposed_change,
-                state=ProposedChangeState.OPEN,
-                database=db,
-                user_id=context.account.account_id,
-            )
-            return Failed(message=f"Merge failure when trying to merge {exc.message}")
+            return Failed(message=exc.message)
+        except Exception as exc:
+            log.exception("Unexpected failure during proposed change merge")
+            return Failed(message=f"Merge failure when trying to merge {source_branch.name}: {exc}")
+        except BaseException as exc:
+            log.exception("Unexpected failure during proposed change merge")
+            raise exc
+        finally:
+            if not merge_succeeded:
+                await _proposed_change_transition_state(
+                    proposed_change=proposed_change,
+                    state=ProposedChangeState.OPEN,
+                    database=db,
+                    user_id=context.account.account_id,
+                )
 
         log.info(f"Branch {source_branch.name} has been merged successfully")
 

backend/tests/component/graphql/mutations/test_proposed_change.py

@@ -16,17 +16,14 @@
 from infrahub.core.manager import NodeManager
 from infrahub.core.node import Node
 from infrahub.core.schema.schema_branch import SchemaBranch
-from infrahub.core.timestamp import Timestamp
 from infrahub.database import InfrahubDatabase
 from infrahub.graphql.initialization import prepare_graphql_params
-from infrahub.message_bus.types import KVTTL
 from infrahub.permissions import AssignedPermissions, PermissionBackend
 from infrahub.proposed_change.constants import ProposedChangeState
 from infrahub.proposed_change.models import RequestProposedChangePipeline
 from infrahub.services import InfrahubServices
 from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
 from infrahub.services.component import InfrahubComponent
-from infrahub.worker import WORKER_IDENTITY
 from infrahub.workers.dependencies import build_client
 from infrahub.workflows.catalogue import REQUEST_PROPOSED_CHANGE_PIPELINE
 from infrahub.workflows.initialization import setup_deployments, setup_worker_pools
@@ -617,18 +614,7 @@ async def test_merge_proposed_change_permission_failure(
                 await setup_deployments(prefect_client)
 
             branch_name = "merge-proposed-change-perm"
-            branch = await create_branch(branch_name=branch_name, db=db)
-            await service.cache.set(
-                key=f"workers:schema_hash:branch:{str(branch.get_uuid)}:{service.component_type.value}:worker:{WORKER_IDENTITY}",
-                value=branch.active_schema_hash.main,
-                expires=KVTTL.TWO_HOURS,
-            )
-            await service.cache.set(
-                key=f"workers:active:{service.component_type.value}:worker:{WORKER_IDENTITY}",
-                value=Timestamp().to_string(),
-                expires=KVTTL.FIFTEEN,
-            )
-            await service.component.refresh_heartbeat()
+            await create_branch(branch_name=branch_name, db=db)
 
             proposed_change = await Node.init(db=db, schema=InfrahubKind.PROPOSEDCHANGE)
             await proposed_change.new(
@@ -658,6 +644,66 @@ async def test_merge_proposed_change_permission_failure(
             assert not update_status.errors
 
 
+class TestMergeProposedChangeUnexpectedFailure(TestInfrahubApp):
+    async def test_merge_proposed_change_unexpected_failure_resets_state(
+        self,
+        db: InfrahubDatabase,
+        default_permission_backend: None,
+        register_core_models_schema: None,
+        session_admin: AccountSession,
+        client: InfrahubClient,
+        dependency_provider: Provider,
+    ) -> None:
+        with dependency_provider.scope(build_client, lambda: client):
+            cache = MemoryCache()
+            message_bus = BusRecorder()
+            service = await InfrahubServices.new(
+                database=db,
+                message_bus=message_bus,
+                workflow=WorkflowLocalExecution(),
+                cache=cache,
+                client=client,
+                component=InfrahubComponent(
+                    cache=cache, db=db, message_bus=message_bus, component_type=ComponentType.NONE
+                ),
+            )
+
+            async with get_client(sync_client=False) as prefect_client:
+                await setup_worker_pools(client=prefect_client)
+                await setup_deployments(prefect_client)
+
+            branch_name = "merge-proposed-change-fail"
+            await create_branch(branch_name=branch_name, db=db)
+
+            proposed_change = await Node.init(db=db, schema=InfrahubKind.PROPOSEDCHANGE)
+            await proposed_change.new(
+                db=db, name="pc-merge-fail-1234", destination_branch="main", source_branch=branch_name, state="open"
+            )
+            await proposed_change.save(db=db)
+
+            with patch(
+                "infrahub.proposed_change.tasks.merge_branch",
+                side_effect=RuntimeError("simulated post-merge failure"),
+            ):
+                update_status = await graphql_mutation(
+                    query=UPDATE_PROPOSED_CHANGE,
+                    db=db,
+                    variables={"proposed_change": proposed_change.id, "state": "merged"},
+                    account_session=session_admin,
+                    service=service,
+                )
+
+            assert update_status.errors
+            assert update_status.errors[0].message == (
+                f"Merge failure when trying to merge {branch_name}: simulated post-merge failure"
+            )
+
+            refreshed_pc = await NodeManager.get_one(db=db, id=proposed_change.id, raise_on_error=True)
+            assert refreshed_pc.get_attribute("state").value.value == ProposedChangeState.OPEN.value
+            branch = await Branch.get_by_name(db=db, name=branch_name)
+            assert branch.status == BranchStatus.OPEN
+
+
 async def test_create_thread(
     db: InfrahubDatabase,
     register_core_models_schema: None,

changelog/+53a1cf44.fixed.md

@@ -0,0 +1 @@
+Reset a Proposed Change's status to "open" following a merge failure that raises an unexpected error.