diff --git a/backend/infrahub/cli/__init__.py b/backend/infrahub/cli/__init__.py index cdd751cae62..b845d78319f 100644 --- a/backend/infrahub/cli/__init__.py +++ b/backend/infrahub/cli/__init__.py @@ -9,6 +9,7 @@ from .db import app as db_app from .dev import app as dev_app from .events import app as events_app +from .recover import app as recover_app from .server import app as server_app from .tasks import app as tasks_app from .upgrade import upgrade_cmd @@ -28,6 +29,7 @@ def common(ctx: typer.Context) -> None: app.add_typer(tasks_app, name="tasks", hidden=True) app.add_typer(dev_app, name="dev", help="Internal development commands.") app.command(name="upgrade")(upgrade_cmd) +app.add_typer(recover_app, name="recover", help="Recover from failed operations.") async def _init_shell(config_file: str) -> None: diff --git a/backend/infrahub/cli/recover.py b/backend/infrahub/cli/recover.py new file mode 100644 index 00000000000..df81e85d753 --- /dev/null +++ b/backend/infrahub/cli/recover.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import typer +from infrahub_sdk.async_typer import AsyncTyper +from rich.console import Console + +from infrahub import config +from infrahub.components import ComponentType +from infrahub.core.branch import Branch +from infrahub.core.initialization import initialize_registry +from infrahub.core.merge.failure_identifier import MergeFailureIdentifier +from infrahub.core.merge.failure_recoverer import MergeFailureRecoverer, RecoveryOutcome, RecoveryReport +from infrahub.core.merge.write_blocker import MergeWriteBlocker +from infrahub.core.registry import registry +from infrahub.core.schema import SchemaRoot, core_models, internal_schema +from infrahub.core.schema.manager import SchemaManager +from infrahub.dependencies.registry import build_component_registry +from infrahub.workers.dependencies import get_cache, get_component, set_component_type + +if TYPE_CHECKING: + from infrahub.cli.context import CliContext + +app = AsyncTyper() + + +# The callback registers this Typer app as a command group so that subcommands (e.g. ``merge``) nest +# under ``recover`` and it carries the group help. A single-subcommand group would otherwise collapse +# to the subcommand itself, losing the ``recover`` grouping. +@app.callback() +def callback() -> None: + """Recover from failed operations.""" + + +def _print_report(console: Console, report: RecoveryReport) -> None: + if report.outcome is RecoveryOutcome.NOTHING_TO_RECOVER: + console.print("[green]No failed merge to recover.[/green]") + elif report.outcome is RecoveryOutcome.ORPHANED_CLEARED: + console.print("[yellow]Cleared a stale merge-protection marker; there was no branch to recover.[/yellow]") + elif report.outcome is RecoveryOutcome.RECOVERED: + console.print(f"[bold green]Recovered the failed merge on branch '{report.branch}'.[/bold green]") + if report.proposed_change is not None: + console.print(f"Proposed change '{report.proposed_change}' was reset to OPEN.") + console.print("Writes to the default branch are allowed again.") + elif report.outcome is RecoveryOutcome.FAILED: + console.print( + f"[red]Recovery of branch '{report.branch}' failed. The branch stays protected; " + f"review the logs and retry.[/red]" + ) + + +@app.command(name="merge") +async def recover_cmd( + ctx: typer.Context, + branch: str | None = typer.Argument( + None, help="Name of the branch to recover; if omitted, the failed merge is auto-detected." + ), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip the confirmation prompt."), + force: bool = typer.Option( + False, + "--force", + "-f", + help="Recover even when the merge lock is absent/ambiguous, not just when the worker is confirmed dead.", + ), + config_file: str = typer.Argument("infrahub.toml", envvar="INFRAHUB_CONFIG"), +) -> None: + """Recover a failed branch merge. + + Roll back the partial graph merge and reset the branch (and any associated proposed change) to + OPEN, then lift the write protection so the default branch is writable again. Idempotent: a run + with nothing to recover reports so and makes no changes. By default only a merge whose worker is + confirmed dead is recovered; pass --force to also recover a merge stuck with an absent/ambiguous + lock. + + Raises: + Exit: When the recovery does not complete successfully (raises typer.Exit to signal a + non-zero exit status to the caller). + + """ + logging.getLogger("infrahub").setLevel(logging.WARNING) + logging.getLogger("neo4j").setLevel(logging.ERROR) + logging.getLogger("prefect").setLevel(logging.ERROR) + + console = Console() + + config.load_and_exit(config_file_name=config_file) + + context: CliContext = ctx.obj + db = await context.init_db(retry=1) + + try: + await initialize_registry(db=db) + set_component_type(component_type=ComponentType.API_SERVER) + build_component_registry() + + cache = await get_cache() + component = await get_component() + merge_write_blocker = MergeWriteBlocker(cache=cache) + default_branch = await Branch.get_by_name(db=db, name=registry.default_branch) + + # The proposed-change lookup during recovery resolves its schema through ``db.schema``, which + # prefers a schema attached to the db over the global registry. Build the internal and core + # schema in a fresh, registry-independent manager and attach it to the db, so recovery works + # without depending on (or mutating) the global registry or any user-defined schema. + schema_manager = SchemaManager() + schema_branch = schema_manager.register_schema(schema=SchemaRoot(**internal_schema), branch=default_branch.name) + schema_branch.load_schema(schema=SchemaRoot(**core_models)) + schema_branch.process() + db.add_schema(schema=schema_branch, name=default_branch.name) + + identifier = MergeFailureIdentifier( + db=db, + cache=cache, + component=component, + merge_write_blocker=merge_write_blocker, + default_branch=default_branch, + grace_period_seconds=config.SETTINGS.main.merge_failure_grace_period_seconds, + ) + recoverer = MergeFailureRecoverer( + db=db, + merge_write_blocker=merge_write_blocker, + identifier=identifier, + default_branch=default_branch, + cache=cache, + ) + + preview = await recoverer.preview(force=force, branch_name=branch) + if preview.outcome is RecoveryOutcome.RECOVERABLE: + console.print(f"A failed merge was found on branch [bold]'{preview.branch}'[/bold].") + if preview.merge_started_at is not None: + console.print(f"The merge started at {preview.merge_started_at}.") + if preview.proposed_change is not None: + console.print(f"Associated proposed change: {preview.proposed_change}.") + + if not yes and not typer.confirm("Recover this failed merge?"): + console.print("Aborted; no changes were made.") + raise typer.Exit(code=1) + + # Pin recovery to the branch that was previewed and confirmed + report = await recoverer.recover(force=force, branch_name=branch or preview.branch) + _print_report(console=console, report=report) + if report.outcome is RecoveryOutcome.FAILED: + raise typer.Exit(code=1) + finally: + await db.close() diff --git a/backend/infrahub/core/diff/merger/merger.py b/backend/infrahub/core/diff/merger/merger.py index eedf3c2d145..8c76cb16e60 100644 --- a/backend/infrahub/core/diff/merger/merger.py +++ b/backend/infrahub/core/diff/merger/merger.py @@ -2,7 +2,6 @@ from typing import TYPE_CHECKING -from infrahub.core import registry from infrahub.core.diff.model.path import BranchTrackingId from infrahub.core.diff.query.bulk_merge import ( BulkMergeAttributePropertyEdgesQuery, @@ -123,11 +122,6 @@ async def merge_graph(self, at: Timestamp) -> None: ) await metadata_query.execute(db=self.db) - branched_from = at.subtract(microseconds=1) - self.source_branch.branched_from = branched_from.to_string() - await self.source_branch.save(db=self.db) - registry.branch[self.source_branch.name] = self.source_branch - log.info("Graph merge complete") @retry_db_transaction(name="bulk_merge_node_existence") diff --git a/backend/infrahub/core/merge/failure_identifier.py b/backend/infrahub/core/merge/failure_identifier.py index 9c2ca1d62b7..870dfca1f99 100644 --- a/backend/infrahub/core/merge/failure_identifier.py +++ b/backend/infrahub/core/merge/failure_identifier.py @@ -1,7 +1,5 @@ from __future__ import annotations -from dataclasses import dataclass -from enum import Enum from typing import TYPE_CHECKING from infrahub import config @@ -24,22 +22,6 @@ log = get_logger() -class RecoveryOutcome(Enum): - NOTHING_TO_RECOVER = "nothing_to_recover" - DECLINED = "declined" - RECOVERED = "recovered" - ORPHANED_CLEARED = "orphaned_cleared" - FAILED = "failed" - - -@dataclass(frozen=True) -class RecoveryReport: - outcome: RecoveryOutcome - branch: str | None - proposed_change: str | None - merge_started_at: str | None - - class MergeFailureIdentifier: """Act on a merge whose worker died mid-flight. @@ -55,43 +37,42 @@ def __init__( cache: InfrahubCache, component: InfrahubComponent, merge_write_blocker: MergeWriteBlocker, + default_branch: Branch, grace_period_seconds: int, ) -> None: self.db = db self.cache = cache self.component = component self.merge_write_blocker = merge_write_blocker + self.default_branch = default_branch self.grace_period_seconds = grace_period_seconds - def should_mark_as_failed_merge( - self, - *, - status: BranchStatus, - lock_holder_worker_id: str | None, - active_worker_ids: set[str], - merge_started_at: Timestamp | None, - now: Timestamp, - ) -> bool: - """Decide whether a branch represents a dead merge that must be flagged ``MERGE_FAILED``. + async def should_mark_as_failed_merge(self, *, branch: Branch, now: Timestamp) -> bool: + """Decide whether a branch stuck in ``MERGING`` is a dead merge whose worker died mid-flight. A merge has failed when all of the following hold: - the branch is still ``MERGING`` (it never reached ``MERGED`` or rolled back to ``OPEN``), - - the global merge lock is *held* (``lock_holder_worker_id is not None``) — a dead worker - cannot release it, so a genuine failure leaves it held; an absent lock is ambiguous (a - cache flush during a live merge would look the same) and is deliberately not auto-flagged, + - the global merge lock is *held* — a dead worker cannot release it, so a genuine failure + leaves it held; an absent lock is ambiguous (a cache flush during a live merge would look + the same) and is deliberately not flagged, - the lock holder is not among the live workers (the holder died), and - the merge has been running longer than the grace period, which absorbs a transient worker-heartbeat write blip so a healthy merge is never mis-flagged. """ - if status != BranchStatus.MERGING: + if branch.status != BranchStatus.MERGING: return False + lock_holder_worker_id = await self.merge_lock_holder() if lock_holder_worker_id is None: return False - if lock_holder_worker_id in active_worker_ids: + if lock_holder_worker_id in await self._active_worker_ids(): return False - if merge_started_at is None: + if branch.merge_started_at is None: return False - return merge_started_at.add(seconds=self.grace_period_seconds) < now + return Timestamp(branch.merge_started_at).add(seconds=self.grace_period_seconds) < now + + async def merge_lock_holder(self) -> str | None: + """Return the worker id currently holding the global merge lock, or ``None`` if it is unheld.""" + return await get_merge_lock_holder_worker_id(cache=self.cache) async def scan(self) -> str | None: """Detect a dead merge and reconcile the protection key. @@ -110,21 +91,9 @@ async def _detect_and_mark(self) -> str | None: if not merging: return None - # The liveness signal is the worker holding the global merge lock: a dead holder means a - # crashed merge. - lock_holder_worker_id = await get_merge_lock_holder_worker_id(cache=self.cache) - active_worker_ids = await self._active_worker_ids() now = Timestamp() - for branch in merging: - merge_started_at = Timestamp(branch.merge_started_at) if branch.merge_started_at else None - if not self.should_mark_as_failed_merge( - status=branch.status, - lock_holder_worker_id=lock_holder_worker_id, - active_worker_ids=active_worker_ids, - merge_started_at=merge_started_at, - now=now, - ): + if not await self.should_mark_as_failed_merge(branch=branch, now=now): continue branch.status = BranchStatus.MERGE_FAILED @@ -175,17 +144,19 @@ async def _reconcile_protection_key(self) -> None: await self.merge_write_blocker.set(branch=protected.name, state=expected_state) async def _active_worker_ids(self) -> set[str]: - workers = await self.component.list_workers(branch=registry.default_branch, schema_hash=False) + workers = await self.component.list_workers(branch=self.default_branch.name, schema_hash=False) return {worker.id for worker in workers if worker.active} async def scan_for_failed_merges(db: InfrahubDatabase, service: InfrahubServices) -> str | None: - """Build a ``MergeFailureRecovery`` from the running service and run one detection scan.""" - recovery = MergeFailureIdentifier( + """Build the identifier from the running service and run one detection scan.""" + default_branch = await registry.get_branch(db=db, branch=registry.default_branch) + identifier = MergeFailureIdentifier( db=db, cache=service.cache, component=service.component, merge_write_blocker=MergeWriteBlocker(cache=service.cache), + default_branch=default_branch, grace_period_seconds=config.SETTINGS.main.merge_failure_grace_period_seconds, ) - return await recovery.scan() + return await identifier.scan() diff --git a/backend/infrahub/core/merge/failure_recoverer.py b/backend/infrahub/core/merge/failure_recoverer.py new file mode 100644 index 00000000000..f8121d084d5 --- /dev/null +++ b/backend/infrahub/core/merge/failure_recoverer.py @@ -0,0 +1,293 @@ +from __future__ import annotations + +import time +from dataclasses import dataclass +from enum import Enum +from typing import TYPE_CHECKING + +from infrahub.core.branch import Branch +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.branch.filters import BranchListFilters +from infrahub.core.manager import NodeManager +from infrahub.core.protocols import CoreProposedChange +from infrahub.core.query.rollback import RollbackQuery, RollbackScope +from infrahub.core.registry import registry +from infrahub.core.timestamp import Timestamp +from infrahub.log import get_logger +from infrahub.proposed_change.constants import ProposedChangeState + +from .merge_locker import MERGE_LOCK_KEY +from .write_blocker import MalformedMergeProtectionError + +if TYPE_CHECKING: + from infrahub.database import InfrahubDatabase + from infrahub.services.adapters.cache import InfrahubCache + + from .failure_identifier import MergeFailureIdentifier + from .write_blocker import MergeWriteBlocker + +log = get_logger() + + +class RecoveryOutcome(Enum): + NOTHING_TO_RECOVER = "nothing_to_recover" + RECOVERABLE = "recoverable" + RECOVERED = "recovered" + ORPHANED_CLEARED = "orphaned_cleared" + FAILED = "failed" + + +@dataclass(frozen=True) +class RecoveryReport: + outcome: RecoveryOutcome + branch: str | None + proposed_change: str | None + merge_started_at: str | None + + +class MergeFailureRecoverer: + """Reverse a failed branch merge and return the system to a writable state. + + A merge that dies mid-flight leaves the default branch partially merged and write-protected. + ``recover`` finds that branch, reverses the partial merge with the range rollback, resets any + associated proposed change and then the branch to ``OPEN``, and lifts the write protection. It + is idempotent. A run interrupted before the write protection is lifted leaves the branch + flagged so a re-run re-detects it, re-runs the rollback (a no-op) and finishes the reset. + """ + + def __init__( + self, + db: InfrahubDatabase, + merge_write_blocker: MergeWriteBlocker, + identifier: MergeFailureIdentifier, + default_branch: Branch, + cache: InfrahubCache, + ) -> None: + self.db = db + self.merge_write_blocker = merge_write_blocker + self.identifier = identifier + self.default_branch = default_branch + self.cache = cache + + async def preview(self, *, force: bool = False, branch_name: str | None = None) -> RecoveryReport: + """Report whether a failed merge can be recovered, making no changes. + + Read-only: it never lifts write protection or touches a branch. When a recoverable branch is + found it reports ``RECOVERABLE`` with the branch, its merge start and any associated proposed + change so an operator can decide; otherwise it reports ``NOTHING_TO_RECOVER``. + """ + recoverable = await self._find_recoverable(force=force, branch_name=branch_name) + if recoverable is None: + return RecoveryReport( + outcome=RecoveryOutcome.NOTHING_TO_RECOVER, branch=None, proposed_change=None, merge_started_at=None + ) + + branch, proposed_change = recoverable + return RecoveryReport( + outcome=RecoveryOutcome.RECOVERABLE, + branch=branch.name, + proposed_change=proposed_change.get_id() if proposed_change else None, + merge_started_at=branch.merge_started_at, + ) + + async def recover(self, *, force: bool = False, branch_name: str | None = None) -> RecoveryReport: + """Recover the failed merge. + + Recovery always acts on a branch already flagged ``MERGE_FAILED``. It also acts on a branch + stuck in ``MERGING`` when the merge is provably dead, and with ``force`` on one with an absent + merge lock. + + When nothing needs recovering, clears a stale write-protection cache key if one names a branch + that no longer exists (``ORPHANED_CLEARED``) or reports ``NOTHING_TO_RECOVER`` otherwise. On a + recoverable branch, rolls the merge back, resets the proposed change and the branch to + ``OPEN`` and lifts the write protection, returning ``RECOVERED`` or ``FAILED``. + """ + recoverable = await self._find_recoverable(force=force, branch_name=branch_name) + if recoverable is None: + # A named branch that is not recoverable leaves any cache key belonging to another branch + # in place; only auto-detect mode may clear an orphaned cache key. + if branch_name is not None: + return RecoveryReport( + outcome=RecoveryOutcome.NOTHING_TO_RECOVER, branch=None, proposed_change=None, merge_started_at=None + ) + return await self._clear_orphaned_or_nothing() + + branch, proposed_change = recoverable + merge_started_at = branch.merge_started_at + proposed_change_id = proposed_change.get_id() if proposed_change else None + + log.info( + "merge.recovery.started", + branch=branch.name, + merge_started_at=merge_started_at, + proposed_change=proposed_change_id, + ) + started_at = time.monotonic() + try: + await self._rollback(merge_started_at=merge_started_at) + # Reset the proposed change before the branch in case a later step fails, the branch stays + # flagged so a re-run re-detects it and completes. + await self._reset_proposed_change(proposed_change=proposed_change) + # Drop the stale merge lock while protection is still up, so this unconditional delete can + # only hit the dead worker's lock, not a lock for a fresh merge. + await self._release_merge_lock() + await self._reset_branch(branch=branch) + # Lift the write protection last, so a failure earlier in the sequence leaves it in place. + await self.merge_write_blocker.delete() + except Exception as exc: + log.error("merge.recovery.failed", branch=branch.name, error=str(exc), exc_info=True) + return RecoveryReport( + outcome=RecoveryOutcome.FAILED, + branch=branch.name, + proposed_change=proposed_change_id, + merge_started_at=merge_started_at, + ) + + log.info( + "merge.recovery.completed", + branch=branch.name, + proposed_change=proposed_change_id, + duration_ms=int((time.monotonic() - started_at) * 1000), + ) + return RecoveryReport( + outcome=RecoveryOutcome.RECOVERED, + branch=branch.name, + proposed_change=proposed_change_id, + merge_started_at=merge_started_at, + ) + + async def _find_recoverable( + self, *, force: bool, branch_name: str | None + ) -> tuple[Branch, CoreProposedChange | None] | None: + """Return the recoverable branch and its associated proposed change, or ``None`` if none. + + Shared by the read-only preview and the mutating recovery so both agree on what is + recoverable; the caller decides whether to act on the returned live objects. + """ + branch = await self._find_recoverable_branch(force=force, branch_name=branch_name) + if branch is None: + return None + proposed_change = await self._find_proposed_change(branch_name=branch.name) + return branch, proposed_change + + async def _find_recoverable_branch(self, *, force: bool, branch_name: str | None) -> Branch | None: + """Return the branch whose failed merge must be recovered, or ``None`` if none is found. + + A branch recorded ``MERGE_FAILED`` is always recoverable. A branch still ``MERGING`` is + recoverable when its merge is provably dead — the merge lock is held by a worker that is no + longer active and the grace period has passed. A ``MERGING`` branch whose lock holder is a live + worker is a healthy merge and is left untouched. + + A ``MERGING`` branch with no lock is ambiguous and is gated behind ``force``. The missing lock + could be caused by a cache flush during a healthy merge. + """ + branches = await Branch.get_list( + db=self.db, + branch_filters=BranchListFilters(statuses=[BranchStatus.MERGE_FAILED, BranchStatus.MERGING]), + ) + if branch_name is not None: + branches = [b for b in branches if b.name == branch_name] + + failed = next((b for b in branches if b.status == BranchStatus.MERGE_FAILED), None) + if failed is not None: + return failed + + merging = [b for b in branches if b.status == BranchStatus.MERGING] + if not merging: + return None + + now = Timestamp() + for branch in merging: + if await self.identifier.should_mark_as_failed_merge(branch=branch, now=now): + return branch + if force and await self.identifier.merge_lock_holder() is None: + return branch + return None + + async def _clear_orphaned_or_nothing(self) -> RecoveryReport: + """Clear a stale write-protection cache key when no merge still owns it. + + A key naming a branch still ``MERGING`` is left in place. A key is stale and dropped when its + branch was removed out-of-band, is malformed with no identifiable branch, or is already + ``OPEN``. A cache read error propagates rather than dropping a key whose value could not be read. + """ + try: + protection = await self.merge_write_blocker.get() + except MalformedMergeProtectionError: + # A present-but-unparseable cache key names no branch to recover but still blocks writes; drop it. + log.warning("merge.recovery.orphaned_cleared") + await self.merge_write_blocker.delete() + return RecoveryReport( + outcome=RecoveryOutcome.ORPHANED_CLEARED, branch=None, proposed_change=None, merge_started_at=None + ) + + if protection is None: + return RecoveryReport( + outcome=RecoveryOutcome.NOTHING_TO_RECOVER, branch=None, proposed_change=None, merge_started_at=None + ) + + existing = await Branch.get_list(db=self.db, name=protection.branch) + branch = existing[0] if existing else None + if branch is not None and branch.status == BranchStatus.MERGING: + # A live in-progress merge (or an ambiguous absent-lock merge not auto-recovered) still owns + # the key. A MERGE_FAILED branch would have been recovered before reaching here. + return RecoveryReport( + outcome=RecoveryOutcome.NOTHING_TO_RECOVER, branch=None, proposed_change=None, merge_started_at=None + ) + + log.warning("merge.recovery.orphaned_cleared", branch=protection.branch) + await self.merge_write_blocker.delete() + return RecoveryReport( + outcome=RecoveryOutcome.ORPHANED_CLEARED, + branch=protection.branch, + proposed_change=None, + merge_started_at=None, + ) + + async def _rollback(self, merge_started_at: str | None) -> None: + """Reverse every default-branch write stamped at or after the merge start. + + Scoped to the default branch (the only merge target) and keyed on the merge start; the write + block guarantees the merge owned every default-branch write in that window, so a range revert + restores the pre-merge graph and per-node metadata. A missing timestamp means the cache key + predates the recorded merge start and there is nothing to reverse. + """ + if merge_started_at is None: + return + rollback_query = await RollbackQuery.init( + db=self.db, + branch=self.default_branch, + target_branch=self.default_branch, + at=Timestamp(merge_started_at), + scope=RollbackScope.SINCE_TIMESTAMP, + restore_metadata=True, + ) + await rollback_query.execute(db=self.db) + + async def _reset_branch(self, branch: Branch) -> None: + branch.status = BranchStatus.OPEN + await branch.save(db=self.db) + registry.branch[branch.name] = branch + + async def _release_merge_lock(self) -> None: + """Release the global merge lock a dead merge worker left held by dropping its cache key. + + A no-op when the lock is already unheld. + """ + await self.cache.delete(key=MERGE_LOCK_KEY) + + async def _find_proposed_change(self, branch_name: str) -> CoreProposedChange | None: + proposed_changes = await NodeManager.query( + db=self.db, + schema=CoreProposedChange, + filters={"source_branch__value": branch_name, "state__value": ProposedChangeState.MERGING.value}, + branch=self.default_branch, + ) + return proposed_changes[0] if proposed_changes else None + + async def _reset_proposed_change(self, proposed_change: CoreProposedChange | None) -> None: + if proposed_change is None: + return + # The state attribute is enum-backed: reads coerce back to the enum, writes take the raw value. + proposed_change.state.value = ProposedChangeState.OPEN.value # type: ignore[misc] + await proposed_change.save(db=self.db) diff --git a/backend/infrahub/graphql/mutations/branch.py b/backend/infrahub/graphql/mutations/branch.py index df54601df97..7085214e9da 100644 --- a/backend/infrahub/graphql/mutations/branch.py +++ b/backend/infrahub/graphql/mutations/branch.py @@ -7,7 +7,7 @@ from typing_extensions import Self from infrahub.branch.merge_mutation_checker import verify_branch_merge_mutation_allowed -from infrahub.branch.status_checker import BranchStatusChecker +from infrahub.branch.status_checker import MERGE_RECOVERY_REQUIRED_MESSAGE, BranchStatusChecker from infrahub.core import registry from infrahub.core.account import GlobalPermission from infrahub.core.branch import Branch @@ -17,7 +17,7 @@ from infrahub.core.merge.write_blocker import MergeWriteBlocker from infrahub.core.protocols import CoreProposedChange from infrahub.database import retry_db_transaction -from infrahub.exceptions import BranchNotFoundError, ValidationError +from infrahub.exceptions import BranchNotFoundError, MergeRecoveryRequiredError, ValidationError from infrahub.graphql.context import apply_external_context from infrahub.graphql.field_extractor import extract_graphql_fields from infrahub.graphql.types.context import ContextInput @@ -154,6 +154,13 @@ async def mutate( graphql_context: GraphqlContext = info.context obj = await Branch.get_by_name(db=graphql_context.db, name=str(data.name)) + # A branch left in MERGE_FAILED by a died merge must not be deleted until an administrator has + # recovered it. + if obj.status == BranchStatus.MERGE_FAILED: + raise MergeRecoveryRequiredError( + identifier=obj.name, message=MERGE_RECOVERY_REQUIRED_MESSAGE, merging_branch=obj.name + ) + # Branch deletes are exempt from the merge write gate, but must verify that the branch being deleted is # not the one being merged merge_write_blocker = MergeWriteBlocker(cache=graphql_context.active_service.cache) diff --git a/backend/tests/component/core/merge/conftest.py b/backend/tests/component/core/merge/conftest.py new file mode 100644 index 00000000000..48082cc9ae5 --- /dev/null +++ b/backend/tests/component/core/merge/conftest.py @@ -0,0 +1,81 @@ +"""Shared fakes and builders for the merge-failure recovery tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from infrahub.core.merge.failure_identifier import MergeFailureIdentifier +from infrahub.core.merge.failure_recoverer import MergeFailureRecoverer +from infrahub.core.merge.write_blocker import MergeWriteBlocker + +if TYPE_CHECKING: + from infrahub.core.branch import Branch + from infrahub.database import InfrahubDatabase + from infrahub.services.component import InfrahubComponent + from tests.adapters.cache import MemoryCache + +# Recovery finds the durably flagged branch directly, so the liveness grace period is irrelevant to +# most tests; those that exercise the MERGING liveness gate override it per-call. +GRACE_PERIOD_SECONDS = 180 + + +class FailAtBranchResetRecoverer(MergeFailureRecoverer): + """Real recoverer that raises while resetting the branch, after the graph rollback has run. + + Reproduces a recovery interrupted after the rollback lands but before the branch is reopened and + the protection lifted, so the branch stays flagged and protected for a re-run. + """ + + async def _reset_branch(self, branch: Branch) -> None: + raise RuntimeError("branch reset failed") + + +class FailAtLockReleaseRecoverer(MergeFailureRecoverer): + """Real recoverer whose merge-lock release fails, before the branch is reopened. + + Reproduces a cache backend that cannot drop the stale lock key. The failure must not be swallowed: + a held lock blocks every future merge, so recovery must report failure and leave the branch flagged, + the protection held and the lock in place for a re-run. + """ + + async def _release_merge_lock(self) -> None: + raise RuntimeError("lock release failed") + + +def build_identifier( + db: InfrahubDatabase, + cache: MemoryCache, + component: InfrahubComponent, + default_branch: Branch, + grace_period_seconds: int = GRACE_PERIOD_SECONDS, +) -> MergeFailureIdentifier: + return MergeFailureIdentifier( + db=db, + cache=cache, + component=component, + merge_write_blocker=MergeWriteBlocker(cache=cache), + default_branch=default_branch, + grace_period_seconds=grace_period_seconds, + ) + + +def build_recovery( + db: InfrahubDatabase, + cache: MemoryCache, + component: InfrahubComponent, + default_branch: Branch, + grace_period_seconds: int = GRACE_PERIOD_SECONDS, +) -> MergeFailureRecoverer: + return MergeFailureRecoverer( + db=db, + merge_write_blocker=MergeWriteBlocker(cache=cache), + identifier=build_identifier( + db=db, + cache=cache, + component=component, + default_branch=default_branch, + grace_period_seconds=grace_period_seconds, + ), + default_branch=default_branch, + cache=cache, + ) diff --git a/backend/tests/component/core/merge/test_failure_detection.py b/backend/tests/component/core/merge/test_failure_detection.py index 2b0737b3cf3..abf8bb5a699 100644 --- a/backend/tests/component/core/merge/test_failure_detection.py +++ b/backend/tests/component/core/merge/test_failure_detection.py @@ -46,13 +46,14 @@ async def component(self, db: InfrahubDatabase, cache: MemoryCache, default_bran return component def _identifier( - self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent + self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent, default_branch: Branch ) -> MergeFailureIdentifier: return MergeFailureIdentifier( db=db, cache=cache, component=component, merge_write_blocker=MergeWriteBlocker(cache=cache), + default_branch=default_branch, grace_period_seconds=GRACE_PERIOD_SECONDS, ) @@ -68,10 +69,15 @@ async def merging_branch_for_10m(self, db: InfrahubDatabase, default_branch: Bra return branch async def test_dead_holder_past_grace_is_flagged( - self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent, merging_branch_for_10m: Branch + self, + db: InfrahubDatabase, + cache: MemoryCache, + component: InfrahubComponent, + merging_branch_for_10m: Branch, + default_branch: Branch, ) -> None: await cache.set(MERGE_LOCK_KEY, _lock_token(DEAD_WORKER)) - recovery = self._identifier(db, cache, component) + recovery = self._identifier(db, cache, component, default_branch) flagged = await recovery.scan() @@ -83,17 +89,22 @@ async def test_dead_holder_past_grace_is_flagged( ) async def test_active_holder_is_not_flagged( - self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent, merging_branch_for_10m: Branch + self, + db: InfrahubDatabase, + cache: MemoryCache, + component: InfrahubComponent, + merging_branch_for_10m: Branch, + default_branch: Branch, ) -> None: await cache.set(MERGE_LOCK_KEY, _lock_token(WORKER_IDENTITY)) - recovery = self._identifier(db, cache, component) + recovery = self._identifier(db, cache, component, default_branch) assert await recovery.scan() is None reloaded = await Branch.get_by_name(db=db, name=merging_branch_for_10m.name) assert reloaded.status == BranchStatus.MERGING async def test_within_grace_is_not_flagged( - self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent + self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent, default_branch: Branch ) -> None: branch = Branch( name="failure-detection-young-merge", @@ -103,17 +114,22 @@ async def test_within_grace_is_not_flagged( ) await branch.save(db=db) await cache.set(MERGE_LOCK_KEY, _lock_token(DEAD_WORKER)) - recovery = self._identifier(db, cache, component) + recovery = self._identifier(db, cache, component, default_branch) assert await recovery.scan() is None reloaded = await Branch.get_by_name(db=db, name=branch.name) assert reloaded.status == BranchStatus.MERGING async def test_scan_is_idempotent( - self, db: InfrahubDatabase, cache: MemoryCache, component: InfrahubComponent, merging_branch_for_10m: Branch + self, + db: InfrahubDatabase, + cache: MemoryCache, + component: InfrahubComponent, + merging_branch_for_10m: Branch, + default_branch: Branch, ) -> None: await cache.set(MERGE_LOCK_KEY, _lock_token(DEAD_WORKER)) - recovery = self._identifier(db, cache, component) + recovery = self._identifier(db, cache, component, default_branch) assert await recovery.scan() == merging_branch_for_10m.name # A second pass finds no MERGING branch (it is now MERGE_FAILED), so it is a no-op. diff --git a/backend/tests/component/core/merge/test_recovery.py b/backend/tests/component/core/merge/test_recovery.py new file mode 100644 index 00000000000..6fb8ee4501f --- /dev/null +++ b/backend/tests/component/core/merge/test_recovery.py @@ -0,0 +1,462 @@ +from __future__ import annotations + +import re +from typing import TYPE_CHECKING + +import pytest + +from infrahub.branch.status_checker import MERGE_RECOVERY_REQUIRED_MESSAGE, BranchStatusChecker +from infrahub.components import ComponentType +from infrahub.core.branch import Branch +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.constants import InfrahubKind +from infrahub.core.manager import NodeManager +from infrahub.core.merge.failure_recoverer import RecoveryOutcome +from infrahub.core.merge.merge_locker import MERGE_LOCK_KEY +from infrahub.core.merge.write_blocker import MergeProtection, MergeProtectionState, MergeWriteBlocker +from infrahub.core.node import Node +from infrahub.core.timestamp import Timestamp +from infrahub.exceptions import MergeRecoveryRequiredError +from infrahub.services.component import InfrahubComponent +from infrahub.worker import WORKER_IDENTITY +from tests.adapters.cache import MemoryCache +from tests.adapters.message_bus import BusRecorder + +from .conftest import ( + FailAtBranchResetRecoverer, + FailAtLockReleaseRecoverer, + build_identifier, + build_recovery, +) + +if TYPE_CHECKING: + from infrahub.core.schema.schema_branch import SchemaBranch + from infrahub.database import InfrahubDatabase + +DEAD_WORKER = "dead-worker" + + +def _lock_token(worker_id: str) -> str: + return f"{Timestamp().to_string()}::{worker_id}" + + +class TestRecovery: + """Recovery's non-happy paths: nothing to recover, orphaned marker, stuck merge, preview, delete gate.""" + + @pytest.fixture + async def cache(self) -> MemoryCache: + return MemoryCache() + + @pytest.fixture + async def component(self, db: InfrahubDatabase, cache: MemoryCache, default_branch: Branch) -> InfrahubComponent: + # refresh_heartbeat marks THIS worker active, so a lock held by any other worker id is dead. + component = InfrahubComponent( + cache=cache, db=db, message_bus=BusRecorder(), component_type=ComponentType.API_SERVER + ) + await component.refresh_heartbeat() + return component + + async def test_no_failure_reports_nothing_to_recover( + self, + db: InfrahubDatabase, + default_branch: Branch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.NOTHING_TO_RECOVER + assert report.branch is None + + async def test_orphaned_marker_is_cleared_without_crashing( + self, + db: InfrahubDatabase, + default_branch: Branch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + # The cache still names a protected branch, but that branch was removed out-of-band. + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch="branch-removed-out-of-band", state=MergeProtectionState.MERGE_FAILED) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.ORPHANED_CLEARED + assert report.branch == "branch-removed-out-of-band" + assert await blocker.get() is None + + async def test_stale_marker_for_reopened_branch_is_cleared( + self, + db: InfrahubDatabase, + default_branch: Branch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + # A prior recovery reopened the branch to OPEN but its cache-key delete then failed, leaving a + # stale MERGE_FAILED marker for a branch that is now OPEN. A re-run finishes the cleanup rather + # than leaving default-branch writes blocked until the watcher reconciles. + branch = Branch( + name="recovery-reopened-stale-marker", + status=BranchStatus.OPEN, + branched_from=Timestamp().to_string(), + ) + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.ORPHANED_CLEARED + assert report.branch == branch.name + assert await blocker.get() is None + + async def test_stuck_merging_with_dead_lock_is_recovered( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + # A branch stuck in MERGING whose merge lock is held by a worker that is no longer alive. The + # recurring scan deliberately leaves this ambiguous case alone; operator confirmation recovers. + branch = Branch( + name="recovery-stuck-merging", + status=BranchStatus.MERGING, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + await cache.set(MERGE_LOCK_KEY, _lock_token(DEAD_WORKER)) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGING) + + recovery = build_recovery( + db=db, cache=cache, component=component, default_branch=default_branch, grace_period_seconds=0 + ) + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.RECOVERED + assert report.branch == branch.name + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert await blocker.get() is None + # The stale merge lock the dead worker held is released, so the next merge is not blocked until + # the deadlock-cleanup cron runs. + assert await cache.get(MERGE_LOCK_KEY) is None + + async def test_preview_is_read_only_then_recover_recovers( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + branch = Branch( + name="recovery-preview", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + + # Preview reports the branch as recoverable and makes no changes. + preview = await recovery.preview() + assert preview.outcome == RecoveryOutcome.RECOVERABLE + assert preview.branch == branch.name + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.MERGE_FAILED + assert await blocker.get() == MergeProtection(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + # A subsequent recover actually recovers the branch. + report = await recovery.recover() + assert report.outcome == RecoveryOutcome.RECOVERED + assert report.branch == branch.name + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert await blocker.get() is None + + async def test_failed_branch_delete_is_rejected_then_allowed_after_recovery( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + branch = Branch( + name="recovery-delete-gate", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + # The same gate the mutation middleware runs for BranchDelete refuses the failed branch. + checker = BranchStatusChecker(db=db, merge_write_blocker=blocker) + with pytest.raises(MergeRecoveryRequiredError, match=re.escape(MERGE_RECOVERY_REQUIRED_MESSAGE)): + await checker.check_merging_status(branch=branch) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover() + assert report.outcome == RecoveryOutcome.RECOVERED + + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + # With the branch reopened and the protection lifted, the delete gate no longer blocks it. + await checker.check_merging_status(branch=reloaded) + + async def test_healthy_in_progress_merge_marker_is_retained( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + # A healthy in-progress merge: the branch is MERGING and the merge lock is held by THIS live + # worker (refresh_heartbeat marked it active). The read-only preview must leave it alone. + branch = Branch( + name="recovery-healthy-merging", + status=BranchStatus.MERGING, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + await cache.set(MERGE_LOCK_KEY, _lock_token(WORKER_IDENTITY)) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGING) + + recovery = build_recovery( + db=db, cache=cache, component=component, default_branch=default_branch, grace_period_seconds=0 + ) + report = await recovery.preview() + + assert report.outcome == RecoveryOutcome.NOTHING_TO_RECOVER + # The healthy merge still owns the protection; the preview must not have lifted it. + assert await blocker.get() == MergeProtection(branch=branch.name, state=MergeProtectionState.MERGING) + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.MERGING + + async def test_recovery_failure_holds_protection( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + branch = Branch( + name="recovery-failed-step", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = FailAtBranchResetRecoverer( + db=db, + merge_write_blocker=blocker, + identifier=build_identifier(db=db, cache=cache, component=component, default_branch=default_branch), + default_branch=default_branch, + cache=cache, + ) + + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.FAILED + assert report.branch == branch.name + # The failing step leaves the branch flagged and the protection held for a re-run. + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.MERGE_FAILED + assert await blocker.get() == MergeProtection(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + async def test_lock_release_failure_holds_protection_and_lock( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + branch = Branch( + name="recovery-lock-release-failed", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + lock_token = _lock_token(DEAD_WORKER) + await cache.set(MERGE_LOCK_KEY, lock_token) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = FailAtLockReleaseRecoverer( + db=db, + merge_write_blocker=blocker, + identifier=build_identifier(db=db, cache=cache, component=component, default_branch=default_branch), + default_branch=default_branch, + cache=cache, + ) + + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.FAILED + assert report.branch == branch.name + # A swallowed lock-release failure would report success while leaving merges permanently blocked; + # instead the branch stays flagged, the protection held and the lock in place for a re-run. + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.MERGE_FAILED + assert await blocker.get() == MergeProtection(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + assert await cache.get(MERGE_LOCK_KEY) == lock_token + + async def test_absent_lock_merging_is_gated_by_force( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + # A branch stuck in MERGING with the merge lock absent (ambiguous: no lock token is set). + branch = Branch( + name="recovery-absent-lock", + status=BranchStatus.MERGING, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGING) + + recovery = build_recovery( + db=db, cache=cache, component=component, default_branch=default_branch, grace_period_seconds=0 + ) + + # Without force the ambiguous absent-lock case is left alone and the marker is retained. + without_force = await recovery.recover(force=False) + assert without_force.outcome == RecoveryOutcome.NOTHING_TO_RECOVER + assert await blocker.get() == MergeProtection(branch=branch.name, state=MergeProtectionState.MERGING) + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.MERGING + + # Force lets the operator override and recover it. + with_force = await recovery.recover(force=True) + assert with_force.outcome == RecoveryOutcome.RECOVERED + assert with_force.branch == branch.name + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert await blocker.get() is None + + async def test_proposed_change_is_reset_to_open( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + branch = Branch( + name="recovery-proposed-change", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + ) + await branch.save(db=db) + + proposed_change = await Node.init(db=db, schema=InfrahubKind.PROPOSEDCHANGE) + await proposed_change.new( + db=db, name="pc-recovery", source_branch=branch.name, destination_branch="main", state="merging" + ) + await proposed_change.save(db=db) + + # Captured after the proposed change is created so the rollback window holds none of the setup. + branch.merge_started_at = Timestamp().to_string() + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.RECOVERED + assert report.proposed_change == proposed_change.id + reloaded_pc = await NodeManager.get_one(db=db, id=proposed_change.id, raise_on_error=True) + assert reloaded_pc.state.value.value == "open" # type: ignore[attr-defined] + + async def test_named_branch_is_targeted( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + branch = Branch( + name="recovery-named-target", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover(branch_name=branch.name) + + assert report.outcome == RecoveryOutcome.RECOVERED + assert report.branch == branch.name + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert await blocker.get() is None + + async def test_named_branch_that_is_not_recoverable_leaves_other_markers( + self, + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + cache: MemoryCache, + component: InfrahubComponent, + ) -> None: + # A genuinely failed branch owns the protection marker, but the operator names a different, + # healthy branch. Recovery must not touch the failed branch's marker. + failed_branch = Branch( + name="recovery-named-failed", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await failed_branch.save(db=db) + healthy_branch = Branch( + name="recovery-named-healthy", + status=BranchStatus.OPEN, + branched_from=Timestamp().to_string(), + ) + await healthy_branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=failed_branch.name, state=MergeProtectionState.MERGE_FAILED) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover(branch_name=healthy_branch.name) + + assert report.outcome == RecoveryOutcome.NOTHING_TO_RECOVER + assert report.branch is None + # The failed branch and its marker are untouched. + reloaded = await Branch.get_by_name(db=db, name=failed_branch.name) + assert reloaded.status == BranchStatus.MERGE_FAILED + assert await blocker.get() == MergeProtection( + branch=failed_branch.name, state=MergeProtectionState.MERGE_FAILED + ) diff --git a/backend/tests/component/core/merge/test_recovery_rollback.py b/backend/tests/component/core/merge/test_recovery_rollback.py new file mode 100644 index 00000000000..0fc6323582e --- /dev/null +++ b/backend/tests/component/core/merge/test_recovery_rollback.py @@ -0,0 +1,361 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import pytest + +from infrahub.components import ComponentType +from infrahub.core.branch import Branch +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.diff.coordinator import DiffCoordinator +from infrahub.core.diff.merger.exclusion_plan import MergeExclusionPlan, MergeExclusionPlanBuilder +from infrahub.core.diff.merger.merger import DiffMerger +from infrahub.core.diff.repository.repository import DiffRepository +from infrahub.core.initialization import create_branch +from infrahub.core.manager import NodeManager +from infrahub.core.merge.failure_recoverer import RecoveryOutcome +from infrahub.core.merge.write_blocker import MergeProtectionState, MergeWriteBlocker +from infrahub.core.node import Node +from infrahub.core.timestamp import Timestamp +from infrahub.dependencies.registry import get_component_registry +from infrahub.services.component import InfrahubComponent +from tests.adapters.cache import MemoryCache +from tests.adapters.message_bus import BusRecorder +from tests.helpers.db_validation import count_branch_edges_at, get_node_metadata, verify_graph + +from .conftest import FailAtBranchResetRecoverer, build_identifier, build_recovery + +if TYPE_CHECKING: + from infrahub.core.schema.schema_branch import SchemaBranch + from infrahub.database import InfrahubDatabase + + +async def _branch_edge_fingerprint(db: InfrahubDatabase, branch_name: str) -> list[tuple]: + """Snapshot every edge on a branch, keyed on endpoints and timestamps. + + Two snapshots compare equal only when the graph is byte-for-byte identical for that branch, so + an empty diff between a pre-merge and a post-recovery snapshot proves the rollback restored the + branch exactly (new edges deleted, closed edges reopened to their original open state). + + This branch-scoped edge fingerprint is used instead of a whole-DB snapshot: a whole-DB snapshot + hashes every node property, so the ``Branch`` vertex's ``merge_started_at`` (set at merge start + and deliberately left in place after recovery) would make a pre-merge vs post-recovery snapshot + differ even when the graph rollback was exact. + """ + result = await db.execute_query( + query=( + "MATCH (src)-[r {branch: $branch}]->(dst) " + "RETURN type(r) AS edge_type, elementId(src) AS src, elementId(dst) AS dst, " + "r.from AS edge_from, r.to AS edge_to, r.status AS status" + ), + params={"branch": branch_name}, + ) + return sorted( + ( + row.get("edge_type"), + row.get("src"), + row.get("dst"), + row.get("edge_from"), + row.get("edge_to"), + row.get("status"), + ) + for row in result + ) + + +class _MidMergeFailingDiffMerger(DiffMerger): + """A real DiffMerger that commits the earlier bulk edges then raises before finishing. + + Reproduces a worker dying partway through the graph write phase: the earlier bulk queries have + already committed on the default branch when a later one fails, leaving partially merged data with + no in-process rollback for recovery to clean up. + """ + + async def _bulk_merge_relationship_property_edges(self, at: Timestamp, plan: MergeExclusionPlan) -> None: + await super()._bulk_merge_relationship_property_edges(at=at, plan=plan) + raise ValueError("mid-merge failure injected") + + +@dataclass +class _MergeDataset: + """Data loaded once for the ordered fail->recover cycles that share a single branch.""" + + branch: Branch + alice_id: str + bob_id: str + original_branched_from: str + + +class TestRecoveryRollback: + """Recovery of a failed merge restores the pre-merge default branch and is idempotent (real db). + + The tests intentionally run in definition order against the SAME branch: the class-scoped + ``merge_dataset`` fixture loads the data once, and each test drives a full merge -> fail -> recover + cycle on that branch. This exercises that a branch can fail to merge and be recovered repeatedly, + always leaving the branch OPEN with its data intact so the next test can re-merge it. Because the + default branch changes with every merge/recover, each test recomputes the branch diff before its + merge. ``branched_from`` is set only at branch creation, so neither the merge nor the recovery + touches it; each test asserts it still equals its original value as an invariant guard. + """ + + @pytest.fixture + async def cache(self) -> MemoryCache: + return MemoryCache() + + @pytest.fixture + async def component( + self, db: InfrahubDatabase, cache: MemoryCache, default_branch_scope_class: Branch + ) -> InfrahubComponent: + component = InfrahubComponent( + cache=cache, db=db, message_bus=BusRecorder(), component_type=ComponentType.API_SERVER + ) + await component.refresh_heartbeat() + return component + + @pytest.fixture(scope="class") + async def merge_dataset( + self, + db: InfrahubDatabase, + default_branch_scope_class: Branch, + register_core_models_schema_scope_class: SchemaBranch, + car_person_schema_scope_class: SchemaBranch, + ) -> _MergeDataset: + alice = await Node.init(db=db, schema="TestPerson", branch=default_branch_scope_class) + await alice.new(db=db, name="Alice", height=170) + await alice.save(db=db) + + branch = await create_branch(branch_name="recover-shared", db=db) + original_branched_from = branch.get_branched_from() + + alice_on_branch = await NodeManager.get_one(db=db, id=alice.id, branch=branch, raise_on_error=True) + alice_on_branch.get_attribute("height").value = 200 + await alice_on_branch.save(db=db) + bob = await Node.init(db=db, schema="TestPerson", branch=branch) + await bob.new(db=db, name="Bob", height=150) + await bob.save(db=db) + + return _MergeDataset( + branch=branch, + alice_id=alice.id, + bob_id=bob.id, + original_branched_from=original_branched_from, + ) + + async def _flag_merge_failed( + self, + db: InfrahubDatabase, + branch: Branch, + merge_at: Timestamp, + cache: MemoryCache, + ) -> MergeWriteBlocker: + """Hand-set the failed-merge marker the orchestrator would persist, then raise the write block. + + Keys the marker on the timestamp the merge wrote at so recovery can find and roll back the + partially merged data. + """ + branch.merge_started_at = merge_at.to_string() + branch.status = BranchStatus.MERGE_FAILED + await branch.save(db=db) + blocker = MergeWriteBlocker(cache=cache) + await blocker.set(branch=branch.name, state=MergeProtectionState.MERGE_FAILED) + return blocker + + async def test_recover_restores_graph_and_metadata( + self, + db: InfrahubDatabase, + default_branch_scope_class: Branch, + cache: MemoryCache, + component: InfrahubComponent, + merge_dataset: _MergeDataset, + ) -> None: + default_branch = default_branch_scope_class + branch = await Branch.get_by_name(db=db, name=merge_dataset.branch.name) + + component_registry = get_component_registry() + diff_coordinator = await component_registry.get_component(DiffCoordinator, db=db, branch=branch) + diff_merger = await component_registry.get_component(DiffMerger, db=db, branch=branch) + await diff_coordinator.update_branch_diff(base_branch=default_branch, diff_branch=branch) + + pre_merge_fingerprint = await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) + alice_metadata_before = await get_node_metadata(db=db, node_uuid=merge_dataset.alice_id) + + merge_at = Timestamp() + await diff_merger.merge_graph(at=merge_at) + + # The merge really landed on the default branch before we flag it as failed. + assert await count_branch_edges_at(db=db, branch_name=default_branch.name, at=merge_at.to_string()) > 0 + alice_after_merge = await NodeManager.get_one(db=db, id=merge_dataset.alice_id, raise_on_error=True) + assert alice_after_merge.get_attribute("height").value == 200 + assert (await get_node_metadata(db=db, node_uuid=merge_dataset.alice_id))["previous_updated_at"] is not None + + # The merge does not touch branched_from; it stays at its branch-creation value. + after_merge = await Branch.get_by_name(db=db, name=branch.name) + assert after_merge.branched_from == merge_dataset.original_branched_from + + blocker = await self._flag_merge_failed( + db=db, + branch=after_merge, + merge_at=merge_at, + cache=cache, + ) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.RECOVERED + assert report.branch == branch.name + assert report.merge_started_at == merge_at.to_string() + + # Graph diff versus the pre-merge snapshot is empty. + assert await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) == pre_merge_fingerprint + assert await count_branch_edges_at(db=db, branch_name=default_branch.name, at=merge_at.to_string()) == 0 + + # Touched-node updated_at/by is restored to its pre-merge value. + assert await get_node_metadata(db=db, node_uuid=merge_dataset.alice_id) == alice_metadata_before + + # Data on the default branch is reverted. + alice_main = await NodeManager.get_one(db=db, id=merge_dataset.alice_id, raise_on_error=True) + assert alice_main.get_attribute("height").value == 170 + assert await NodeManager.get_one(db=db, id=merge_dataset.bob_id) is None + + # The branch is reopened and the write protection is lifted; branched_from was never touched by + # the merge or the recovery, so it still equals its branch-creation value. + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert reloaded.branched_from == merge_dataset.original_branched_from + assert await blocker.get() is None + + await verify_graph(db=db) + + # A second run after a completed recovery finds nothing to do and leaves the graph unchanged. + second = await recovery.recover() + assert second.outcome == RecoveryOutcome.NOTHING_TO_RECOVER + assert await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) == pre_merge_fingerprint + + async def test_interrupted_recovery_is_idempotent( + self, + db: InfrahubDatabase, + default_branch_scope_class: Branch, + cache: MemoryCache, + component: InfrahubComponent, + merge_dataset: _MergeDataset, + ) -> None: + default_branch = default_branch_scope_class + branch = await Branch.get_by_name(db=db, name=merge_dataset.branch.name) + + component_registry = get_component_registry() + diff_coordinator = await component_registry.get_component(DiffCoordinator, db=db, branch=branch) + diff_merger = await component_registry.get_component(DiffMerger, db=db, branch=branch) + # The prior test's merge/recover cycle left new edges on the default branch, so the persisted + # diff and the fingerprint must both be recomputed against the current default-branch state + # before this cycle's merge. + await diff_coordinator.update_branch_diff(base_branch=default_branch, diff_branch=branch) + + pre_merge_fingerprint = await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) + + merge_at = Timestamp() + await diff_merger.merge_graph(at=merge_at) + + after_merge = await Branch.get_by_name(db=db, name=branch.name) + blocker = await self._flag_merge_failed( + db=db, + branch=after_merge, + merge_at=merge_at, + cache=cache, + ) + + # First run: the rollback lands, then the branch reset fails, so the branch stays flagged. + failing = FailAtBranchResetRecoverer( + db=db, + merge_write_blocker=blocker, + identifier=build_identifier(db=db, cache=cache, component=component, default_branch=default_branch), + default_branch=default_branch, + cache=cache, + ) + first = await failing.recover() + + assert first.outcome == RecoveryOutcome.FAILED + # The rollback already restored the graph, but the branch stays flagged and protected. + assert await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) == pre_merge_fingerprint + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.MERGE_FAILED + assert await blocker.get() is not None + + # A full re-run re-detects the branch; the second rollback is a safe no-op and it finishes. + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + second = await recovery.recover() + + assert second.outcome == RecoveryOutcome.RECOVERED + assert await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) == pre_merge_fingerprint + assert await count_branch_edges_at(db=db, branch_name=default_branch.name, at=merge_at.to_string()) == 0 + assert await NodeManager.get_one(db=db, id=merge_dataset.bob_id) is None + # Alice's default-branch value is reverted to its pre-merge height, undoing the merged change. + alice_main = await NodeManager.get_one(db=db, id=merge_dataset.alice_id, raise_on_error=True) + assert alice_main.get_attribute("height").value == 170 + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert reloaded.branched_from == merge_dataset.original_branched_from + assert await blocker.get() is None + + await verify_graph(db=db) + + async def test_recover_restores_partial_graph_after_mid_merge_failure( + self, + db: InfrahubDatabase, + default_branch_scope_class: Branch, + cache: MemoryCache, + component: InfrahubComponent, + merge_dataset: _MergeDataset, + ) -> None: + default_branch = default_branch_scope_class + branch = await Branch.get_by_name(db=db, name=merge_dataset.branch.name) + + component_registry = get_component_registry() + diff_coordinator = await component_registry.get_component(DiffCoordinator, db=db, branch=branch) + diff_repository = await component_registry.get_component(DiffRepository, db=db, branch=branch) + diff_merger = _MidMergeFailingDiffMerger( + db=db, + source_branch=branch, + destination_branch=default_branch, + diff_repository=diff_repository, + exclusion_plan_builder=MergeExclusionPlanBuilder(), + ) + await diff_coordinator.update_branch_diff(base_branch=default_branch, diff_branch=branch) + + pre_merge_fingerprint = await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) + + merge_at = Timestamp() + with pytest.raises(ValueError, match=r"^mid-merge failure injected$"): + await diff_merger.merge_graph(at=merge_at) + + # The earlier bulk queries committed partial data on the default branch before the failure. + assert await count_branch_edges_at(db=db, branch_name=default_branch.name, at=merge_at.to_string()) > 0 + + after_merge = await Branch.get_by_name(db=db, name=branch.name) + blocker = await self._flag_merge_failed( + db=db, + branch=after_merge, + merge_at=merge_at, + cache=cache, + ) + + recovery = build_recovery(db=db, cache=cache, component=component, default_branch=default_branch) + report = await recovery.recover() + + assert report.outcome == RecoveryOutcome.RECOVERED + + # The partially merged graph is fully reverted to its pre-merge state. + assert await _branch_edge_fingerprint(db=db, branch_name=default_branch.name) == pre_merge_fingerprint + assert await count_branch_edges_at(db=db, branch_name=default_branch.name, at=merge_at.to_string()) == 0 + # Bob (added on the branch) never lands on the default branch, and Alice's height is reverted. + assert await NodeManager.get_one(db=db, id=merge_dataset.bob_id) is None + alice_main = await NodeManager.get_one(db=db, id=merge_dataset.alice_id, raise_on_error=True) + assert alice_main.get_attribute("height").value == 170 + + reloaded = await Branch.get_by_name(db=db, name=branch.name) + assert reloaded.status == BranchStatus.OPEN + assert reloaded.branched_from == merge_dataset.original_branched_from + assert await blocker.get() is None + + await verify_graph(db=db) diff --git a/backend/tests/component/core/test_branch_merge.py b/backend/tests/component/core/test_branch_merge.py index 1345a7056e9..396c2991528 100644 --- a/backend/tests/component/core/test_branch_merge.py +++ b/backend/tests/component/core/test_branch_merge.py @@ -58,19 +58,18 @@ async def test_merge_graph( assert cars[0].nbr_seats.value == 5 assert cars[0].nbr_seats.is_protected is False - # Query all cars in BRANCH1, AFTER the merge + # Query all cars in BRANCH1, AFTER the merge. BRANCH1 does not see c2, created on main after + # BRANCH1 forked. cars = sorted(await NodeManager.query(schema="TestCar", branch=branch1, db=db), key=lambda c: c.id) - assert len(cars) == 3 - assert cars[2].id == "c3" - assert cars[2].name.value == "volt" + assert [car.id for car in cars] == ["c1", "c3"] + assert cars[1].name.value == "volt" - # Query all cars in BRANCH1, BEFORE the merge + # Query all cars in BRANCH1, BEFORE the merge — same isolated two-car view (still no c2). cars = sorted( await NodeManager.query(schema="TestCar", branch=branch1, at=base_dataset_02["time0"], db=db), key=lambda c: c.id, ) - assert len(cars) == 3 - assert cars[0].id == "c1" + assert [car.id for car in cars] == ["c1", "c3"] assert cars[0].nbr_seats.value == 4 # It should be possible to merge a graph even without changes diff --git a/backend/tests/component/graphql/mutations/test_branch.py b/backend/tests/component/graphql/mutations/test_branch.py index e6a04eb1cc3..d896db2233a 100644 --- a/backend/tests/component/graphql/mutations/test_branch.py +++ b/backend/tests/component/graphql/mutations/test_branch.py @@ -7,14 +7,17 @@ from infrahub.auth.session import AccountSession from infrahub.auth.types import AuthType +from infrahub.branch.status_checker import MERGE_RECOVERY_REQUIRED_MESSAGE from infrahub.core import registry from infrahub.core.branch import Branch from infrahub.core.branch.enums import BranchStatus from infrahub.core.constants import InfrahubKind from infrahub.core.initialization import create_branch from infrahub.core.manager import NodeManager +from infrahub.core.merge.write_blocker import MergeWriteBlocker 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.services import InfrahubServices @@ -318,6 +321,45 @@ async def test_branch_delete( assert delete_before_create.errors[0].message == "Branch: branch3 not found." +async def test_branch_delete_merge_failed_rejected( + db: InfrahubDatabase, + default_branch: Branch, + register_core_models_schema: SchemaBranch, + session_admin: AccountSession, + local_services: InfrahubServices, +) -> None: + # A branch left durably in MERGE_FAILED must not be deletable until an administrator recovers it, + # even when the volatile write-protection cache key is absent (the routine state after a restart or + # cache flush) and regardless of which branch the request targets. No merge:protected key is set + # here, so a passing assertion proves the delete block reads the durable branch status rather than + # the cache key. + failed_branch = Branch( + name="merge-failed-branch", + status=BranchStatus.MERGE_FAILED, + branched_from=Timestamp().to_string(), + merge_started_at=Timestamp().to_string(), + ) + await failed_branch.save(db=db) + + assert await MergeWriteBlocker(cache=local_services.cache).get() is None + + result = await graphql_mutation( + query='mutation { BranchDelete(data: { name: "merge-failed-branch" }) { ok } }', + db=db, + branch=default_branch, + account_session=session_admin, + service=local_services, + ) + + assert result.errors + assert len(result.errors) == 1 + assert result.errors[0].message == MERGE_RECOVERY_REQUIRED_MESSAGE + + # The guard raises before the delete workflow runs, so the branch is left intact for recovery. + reloaded = await Branch.get_by_name(db=db, name=failed_branch.name) + assert reloaded.status == BranchStatus.MERGE_FAILED + + async def test_branch_rebase_wrong_branch( db: InfrahubDatabase, default_branch: Branch, diff --git a/backend/tests/helpers/db_validation.py b/backend/tests/helpers/db_validation.py index 65c5700e605..1679d467819 100644 --- a/backend/tests/helpers/db_validation.py +++ b/backend/tests/helpers/db_validation.py @@ -388,3 +388,18 @@ async def count_branch_edges_at(db: InfrahubDatabase, branch_name: str, at: str) params={"at": at, "branch": branch_name}, ) return result[0].get("c") + + +async def get_node_metadata(db: InfrahubDatabase, node_uuid: str) -> dict[str, str | None]: + """Return a node vertex's ``updated_at``/``previous_updated_at`` metadata.""" + result = await db.execute_query( + query=( + "MATCH (n:Node {uuid: $uuid}) " + "RETURN n.updated_at AS updated_at, n.previous_updated_at AS previous_updated_at" + ), + params={"uuid": node_uuid}, + ) + return { + "updated_at": result[0].get("updated_at"), + "previous_updated_at": result[0].get("previous_updated_at"), + } diff --git a/backend/tests/integration_docker/test_merge_kill_recovery.py b/backend/tests/integration_docker/test_merge_kill_recovery.py index ad13deffcfa..9513260ec50 100644 --- a/backend/tests/integration_docker/test_merge_kill_recovery.py +++ b/backend/tests/integration_docker/test_merge_kill_recovery.py @@ -35,63 +35,97 @@ async def _wait_for_status(client: InfrahubClient, branch_name: str, target: str return status +async def _drive_branch_to_merge_failed( + client: InfrahubClient, + infrahub_compose: InfrahubDockerCompose, + branch_name: str, +) -> None: + """Take a fresh branch all the way to MERGE_FAILED by killing its merge worker mid-flight. + + Creates the branch with enough changes that the merge stays in MERGING long enough to be caught, + fires the merge without awaiting it (it runs on a task worker), SIGKILLs the worker(s) while the + merge is in flight so the global merge lock stays held by a now-dead worker, restarts the + worker(s), and stays idle until the recurring merge-watcher scan flips the branch to MERGE_FAILED. + """ + await client.branch.create(branch_name=branch_name) + for index in range(20): + node = await client.create(kind="BuiltinTag", name=f"{branch_name}-tag-{index}", branch=branch_name) + await node.save() + + # Fire the merge without awaiting completion; we kill the worker while it is in flight, so awaiting + # here would hang. + merge_task = asyncio.create_task( + client.execute_graphql( + query=f'mutation {{ BranchMerge(data: {{name: "{branch_name}"}}) {{ ok }} }}', + branch_name="main", + ) + ) + try: + # Wait until the merge is genuinely in flight, then SIGKILL the worker(s) so the merge dies + # mid-flight with the global merge lock still held by the now-dead worker. + assert await _wait_for_status(client, branch_name, "MERGING", MERGING_TIMEOUT_SECONDS) == "MERGING", ( + "merge never reached MERGING; cannot exercise the mid-merge kill" + ) + base_cmd = list(infrahub_compose.compose_command_property) + infrahub_compose._run_command(cmd=[*base_cmd, "kill", "-s", "SIGKILL", "task-worker"]) + + # Bring the worker(s) back so the recurring merge-watcher can run; issue no writes — the flip + # must come from the idle scan alone. A restarted worker has a fresh identity, so the dead lock + # holder is no longer in the active set. + infrahub_compose.start_container("task-worker") + + status = await _wait_for_status(client, branch_name, "MERGE_FAILED", DETECTION_TIMEOUT_SECONDS) + assert status == "MERGE_FAILED", f"branch {branch_name} was not flagged MERGE_FAILED (status={status})" + finally: + merge_task.cancel() + try: + await merge_task + except asyncio.CancelledError: + pass + except Exception as exc: + # The mutation legitimately errors once its worker is SIGKILLed mid-flight; retrieve the + # exception so an unrelated early failure (validation, connectivity) surfaces in the + # captured test output instead of being lost as an unretrieved-task warning. + print(f"merge mutation task raised: {exc!r}") + + @pytest.mark.skip(reason="test takes too long") -class TestMergeKillDetection(TestInfrahubDockerClient): - """Cross-process detection: SIGKILL a worker mid-merge, stay idle, expect MERGE_FAILED. +class TestMergeKillRecovery(TestInfrahubDockerClient): + """End-to-end cross-process detection and recovery from a killed merge. - This is the detection half of the kill/recovery integration scenario. The recovery half - (`infrahub recover` re-merges) is added with the recovery increment in the same file. + First the detection half: SIGKILL a merge worker mid-flight and stay idle until the recurring + merge-watcher scan flips the branch to MERGE_FAILED. Then the recovery half: running the recover + command rolls back the partial merge, reopens the branch, restores default-branch writes, and lets + the branch merge again. Detection is a strict prerequisite of recovery, so one flow covers both. """ - async def test_killed_merge_is_flagged_failed_while_idle( + async def test_killed_merge_is_detected_then_recovered( self, client: InfrahubClient, infrahub_compose: InfrahubDockerCompose, ) -> None: - branch_name = "merge_kill_detection" - await client.branch.create(branch_name=branch_name) - # Enough changes that the merge stays in MERGING long enough to be caught and killed - # mid-flight rather than completing before the SIGKILL lands. - for index in range(20): - node = await client.create(kind="BuiltinTag", name=f"merge-kill-tag-{index}", branch=branch_name) - await node.save() - - # Fire the merge without awaiting completion (it runs on a task worker); we kill the worker - # while it is in flight, so awaiting here would hang. - merge_task = asyncio.create_task( - client.execute_graphql( - query=f'mutation {{ BranchMerge(data: {{name: "{branch_name}"}}) {{ ok }} }}', - branch_name="main", - ) + branch_name = "merge_kill_recovery" + await _drive_branch_to_merge_failed(client=client, infrahub_compose=infrahub_compose, branch_name=branch_name) + + # Recover from inside the stack: the CLI process reads its configuration from the same + # environment the server container runs with, so it operates on the stack's database. It exits + # non-zero on a failed recovery, which _run_command surfaces as a raised error. + infrahub_compose.exec_in_container( + command=["infrahub", "recover", "merge", "--yes"], + service_name="infrahub-server", ) - try: - # Wait until the merge is genuinely in flight, then SIGKILL the worker(s) so the merge - # dies mid-flight with the global merge lock still held by the now-dead worker. - assert await _wait_for_status(client, branch_name, "MERGING", MERGING_TIMEOUT_SECONDS) == "MERGING", ( - "merge never reached MERGING; cannot exercise the mid-merge kill" - ) - base_cmd = list(infrahub_compose.compose_command_property) - infrahub_compose._run_command(cmd=[*base_cmd, "kill", "-s", "SIGKILL", "task-worker"]) - - # Bring the worker(s) back so the recurring merge-watcher can run; issue no writes — the - # flip must come from the idle scan alone. A restarted worker has a fresh identity, so the - # dead lock holder is no longer in the active set. - infrahub_compose.start_container("task-worker") - - status = await _wait_for_status(client, branch_name, "MERGE_FAILED", DETECTION_TIMEOUT_SECONDS) - assert status == "MERGE_FAILED", f"branch {branch_name} was not flagged MERGE_FAILED (status={status})" - finally: - merge_task.cancel() - try: - await merge_task - except asyncio.CancelledError: - pass - except Exception as exc: - # The mutation legitimately errors once its worker is SIGKILLed mid-flight; retrieve - # the exception so an unrelated early failure (validation, connectivity) surfaces in - # the captured test output instead of being lost as an unretrieved-task warning. - print(f"merge mutation task raised: {exc!r}") - - @pytest.mark.skip(reason="Recovery half is not implemented yet.") - async def test_recover_after_kill_remerges(self, client: InfrahubClient) -> None: # pragma: no cover - ... + + assert await _branch_status(client, branch_name) == "OPEN", "branch was not reopened after recovery" + + # Writes to the default branch are unblocked again once the protection is lifted. + recovered_write = await client.create(kind="BuiltinTag", name="post-recovery-tag", branch="main") + await recovered_write.save() + + # The reopened branch merges cleanly the second time. + remerge = await client.execute_graphql( + query=f'mutation {{ BranchMerge(data: {{name: "{branch_name}"}}) {{ ok }} }}', + branch_name="main", + ) + assert remerge["BranchMerge"]["ok"] is True + status = await _wait_for_status(client, branch_name, "MERGED", MERGING_TIMEOUT_SECONDS) + assert status == "MERGED", f"branch {branch_name} did not reach MERGED after re-merge (status={status})" diff --git a/backend/tests/unit/core/merge/test_failed_merge_predicate.py b/backend/tests/unit/core/merge/test_failed_merge_predicate.py deleted file mode 100644 index 4908ed68774..00000000000 --- a/backend/tests/unit/core/merge/test_failed_merge_predicate.py +++ /dev/null @@ -1,123 +0,0 @@ -from __future__ import annotations - -from dataclasses import dataclass -from typing import TYPE_CHECKING, cast - -import pytest - -from infrahub.core.branch.enums import BranchStatus -from infrahub.core.merge.failure_identifier import MergeFailureIdentifier -from infrahub.core.merge.write_blocker import MergeWriteBlocker -from infrahub.core.timestamp import Timestamp -from tests.adapters.cache import MemoryCache - -if TYPE_CHECKING: - from infrahub.database import InfrahubDatabase - from infrahub.services.component import InfrahubComponent - -GRACE_SECONDS = 180 - - -def _identifier(grace_period_seconds: int) -> MergeFailureIdentifier: - # should_mark_as_failed_merge decides from its arguments and the configured grace period alone, so the - # unused db/component are not needed to exercise it. - return MergeFailureIdentifier( - db=cast("InfrahubDatabase", None), - cache=MemoryCache(), - component=cast("InfrahubComponent", None), - merge_write_blocker=MergeWriteBlocker(cache=MemoryCache()), - grace_period_seconds=grace_period_seconds, - ) - - -@dataclass -class PredicateCase: - name: str - status: BranchStatus - lock_holder_worker_id: str | None - active_worker_ids: set[str] - started_seconds_ago: int | None - expected: bool - grace_seconds: int = GRACE_SECONDS - - -CASES = [ - PredicateCase( - name="dead-holder-past-grace-is-failed", - status=BranchStatus.MERGING, - lock_holder_worker_id="dead-worker", - active_worker_ids={"live-worker"}, - started_seconds_ago=600, - expected=True, - ), - PredicateCase( - name="live-holder-not-failed", - status=BranchStatus.MERGING, - lock_holder_worker_id="live-worker", - active_worker_ids={"live-worker"}, - started_seconds_ago=600, - expected=False, - ), - PredicateCase( - name="absent-lock-not-flagged", - status=BranchStatus.MERGING, - lock_holder_worker_id=None, - active_worker_ids=set(), - started_seconds_ago=600, - expected=False, - ), - PredicateCase( - name="within-grace-not-flagged", - status=BranchStatus.MERGING, - lock_holder_worker_id="dead-worker", - active_worker_ids=set(), - started_seconds_ago=10, - expected=False, - ), - PredicateCase( - name="not-merging-not-flagged", - status=BranchStatus.MERGE_FAILED, - lock_holder_worker_id="dead-worker", - active_worker_ids=set(), - started_seconds_ago=600, - expected=False, - ), - PredicateCase( - name="open-status-not-flagged", - status=BranchStatus.OPEN, - lock_holder_worker_id="dead-worker", - active_worker_ids=set(), - started_seconds_ago=600, - expected=False, - ), - PredicateCase( - name="missing-merge-started-not-flagged", - status=BranchStatus.MERGING, - lock_holder_worker_id="dead-worker", - active_worker_ids=set(), - started_seconds_ago=None, - expected=False, - ), -] - - -@pytest.mark.parametrize("case", CASES, ids=[c.name for c in CASES]) -def test_should_mark_as_failed_merge(case: PredicateCase) -> None: - now = Timestamp() - # Timestamp.add() is typed as the SDK base class; re-wrap so the value is the core Timestamp the - # predicate expects (mirrors how production builds it from the stored string). - merge_started_at = ( - Timestamp(now.add(seconds=-case.started_seconds_ago).to_string()) - if case.started_seconds_ago is not None - else None - ) - - result = _identifier(grace_period_seconds=case.grace_seconds).should_mark_as_failed_merge( - status=case.status, - lock_holder_worker_id=case.lock_holder_worker_id, - active_worker_ids=case.active_worker_ids, - merge_started_at=merge_started_at, - now=now, - ) - - assert result is case.expected diff --git a/backend/tests/unit/core/merge/test_failure_identifier.py b/backend/tests/unit/core/merge/test_failure_identifier.py new file mode 100644 index 00000000000..5a9e499841a --- /dev/null +++ b/backend/tests/unit/core/merge/test_failure_identifier.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, cast + +import pytest + +from infrahub.components import ComponentType +from infrahub.core.branch import Branch +from infrahub.core.branch.enums import BranchStatus +from infrahub.core.merge.failure_identifier import MergeFailureIdentifier +from infrahub.core.merge.merge_locker import MERGE_LOCK_KEY +from infrahub.core.merge.write_blocker import MergeWriteBlocker +from infrahub.core.timestamp import Timestamp +from infrahub.services.component import InfrahubComponent +from infrahub.worker import WORKER_IDENTITY +from tests.adapters.cache import MemoryCache +from tests.adapters.message_bus import BusRecorder + +if TYPE_CHECKING: + from infrahub.database import InfrahubDatabase + +DEAD_WORKER = "dead-worker" +GRACE_PERIOD_SECONDS = 180 + + +def _lock_token(worker_id: str) -> str: + return f"{Timestamp().to_string()}::{worker_id}" + + +@dataclass +class IsFailedMergeCase: + name: str + status: BranchStatus + lock_worker_id: str | None # None means no merge-lock key is set at all + started_seconds_ago: int | None + expected: bool + + +CASES = [ + IsFailedMergeCase( + name="dead-holder-past-grace-is-failed", + status=BranchStatus.MERGING, + lock_worker_id=DEAD_WORKER, + started_seconds_ago=600, + expected=True, + ), + IsFailedMergeCase( + name="live-holder-not-flagged", + status=BranchStatus.MERGING, + lock_worker_id=WORKER_IDENTITY, + started_seconds_ago=600, + expected=False, + ), + IsFailedMergeCase( + name="absent-lock-not-flagged", + status=BranchStatus.MERGING, + lock_worker_id=None, + started_seconds_ago=600, + expected=False, + ), + IsFailedMergeCase( + name="within-grace-not-flagged", + status=BranchStatus.MERGING, + lock_worker_id=DEAD_WORKER, + started_seconds_ago=0, + expected=False, + ), + IsFailedMergeCase( + name="merge-failed-status-not-flagged", + status=BranchStatus.MERGE_FAILED, + lock_worker_id=DEAD_WORKER, + started_seconds_ago=600, + expected=False, + ), + IsFailedMergeCase( + name="open-status-not-flagged", + status=BranchStatus.OPEN, + lock_worker_id=DEAD_WORKER, + started_seconds_ago=600, + expected=False, + ), + IsFailedMergeCase( + name="missing-merge-started-not-flagged", + status=BranchStatus.MERGING, + lock_worker_id=DEAD_WORKER, + started_seconds_ago=None, + expected=False, + ), +] + + +@pytest.mark.parametrize("case", CASES, ids=[c.name for c in CASES]) +async def test_should_mark_as_failed_merge(case: IsFailedMergeCase) -> None: + # should_mark_as_failed_merge reads only the cache (the merge lock plus worker heartbeats) and the branch + # object it is handed, so a cache-backed component needs no database — this stays a unit test. + cache = MemoryCache() + db = cast("InfrahubDatabase", None) # unused by list_workers / should_mark_as_failed_merge + component = InfrahubComponent( + cache=cache, db=db, message_bus=BusRecorder(), component_type=ComponentType.API_SERVER + ) + await component.refresh_heartbeat() # marks WORKER_IDENTITY active in the cache + + if case.lock_worker_id is not None: + await cache.set(MERGE_LOCK_KEY, _lock_token(case.lock_worker_id)) + + identifier = MergeFailureIdentifier( + db=db, + cache=cache, + component=component, + merge_write_blocker=MergeWriteBlocker(cache=cache), + default_branch=Branch(name="main", status=BranchStatus.OPEN, branched_from=Timestamp().to_string()), + grace_period_seconds=GRACE_PERIOD_SECONDS, + ) + + merge_started_at = ( + Timestamp().add(seconds=-case.started_seconds_ago).to_string() if case.started_seconds_ago is not None else None + ) + branch = Branch( + name=f"is-failed-merge-{case.name}", + status=case.status, + branched_from=Timestamp().to_string(), + merge_started_at=merge_started_at, + ) + + assert await identifier.should_mark_as_failed_merge(branch=branch, now=Timestamp()) is case.expected diff --git a/changelog/+merge-recover-command.added.md b/changelog/+merge-recover-command.added.md new file mode 100644 index 00000000000..94ec88ca8b4 --- /dev/null +++ b/changelog/+merge-recover-command.added.md @@ -0,0 +1 @@ +Added the `infrahub recover merge` CLI command to recover from a failed branch merge. It rolls back the partial merge on the default branch, resets the branch and any associated proposed change to `OPEN`, and lifts the write protection so the default branch is writable again. The command is operator-confirmed (skip the prompt with `--yes`) and idempotent: a run with nothing to recover reports so and makes no changes. While a branch is in the failed state, all mutations against it, including its deletion, are refused until recovery has run. diff --git a/dev/knowledge/backend/branch-status.md b/dev/knowledge/backend/branch-status.md index 0a36e63f3b5..dfadecb7c12 100644 --- a/dev/knowledge/backend/branch-status.md +++ b/dev/knowledge/backend/branch-status.md @@ -22,9 +22,9 @@ How Infrahub enforces read-only constraints on branches based on their lifecycle ### Failed merge detection -`backend/infrahub/core/merge/failure_recovery.py` +`backend/infrahub/core/merge/failure_identifier.py` -A merge holds the global `all_branches` merge lock for its whole `MERGING` window; the lock token encodes the holder's `worker_id`. When the holding worker dies, the lock stays held by a `worker_id` that is no longer in the active-worker set. `MergeFailureRecovery.detect_and_mark` flips such a branch `MERGING → MERGE_FAILED` (after a configurable grace period, `INFRAHUB_MERGE_FAILURE_GRACE_PERIOD_SECONDS`, that absorbs a transient heartbeat blip) and updates the `merge:protected` key to `"{branch}::MERGE_FAILED"`. It runs from the recurring `MERGE_WATCHER` workflow (one-minute cron, single-flighted), from a check at worker startup, and the recurring scan also reconciles the cache key against the durable branch status so protection self-heals after a restart or cache flush. A healthy in-progress merge (lock held by a live worker) is never flagged. +A merge holds the global `all_branches` merge lock for its whole `MERGING` window; the lock token encodes the holder's `worker_id`. When the holding worker dies, the lock stays held by a `worker_id` that is no longer in the active-worker set. `MergeFailureIdentifier.scan` flips such a branch `MERGING → MERGE_FAILED` (after a configurable grace period, `INFRAHUB_MERGE_FAILURE_GRACE_PERIOD_SECONDS`, that absorbs a transient heartbeat blip) and updates the `merge:protected` key to `"{branch}::MERGE_FAILED"`. It runs from the recurring `MERGE_WATCHER` workflow (one-minute cron, single-flighted), from a check at server startup, and the recurring scan also reconciles the cache key against the durable branch status so protection self-heals after a restart or cache flush. A healthy in-progress merge (lock held by a live worker) is never flagged. ## BranchStatusChecker @@ -78,7 +78,7 @@ Some mutations need explicit status checks beyond the middleware for richer erro | File | Mutation | Check | |------|----------|-------| | `graphql/mutations/branch.py` | `BranchMerge` | Rejects `MERGED` source; rejects new merges/rebases while any merge is in progress | -| `graphql/mutations/branch.py` | `BranchDelete` | Rejects deleting the branch currently being merged | +| `graphql/mutations/branch.py` | `BranchDelete` | Rejects deleting a `MERGE_FAILED` branch until recovery, enforced against the branch's durable status in the database so it holds regardless of which branch the request targets and even after the volatile write-protection cache key has been dropped; also rejects deleting the branch currently being merged | | `graphql/mutations/proposed_change.py` | `ProposedChangeCreate` | Rejects `MERGED` source branch | ### 3. REST API Endpoints diff --git a/dev/specs/ifc-2437-merge-failure-recovery/tasks.md b/dev/specs/ifc-2437-merge-failure-recovery/tasks.md index 46e3b086b8d..825feb1bd28 100644 --- a/dev/specs/ifc-2437-merge-failure-recovery/tasks.md +++ b/dev/specs/ifc-2437-merge-failure-recovery/tasks.md @@ -87,17 +87,17 @@ Infrahub backend monolith: `backend/infrahub/...`; tests under `backend/tests/{u - [X] T027 [US3] Expose a recovery rollback entry on `DiffMerger` (`backend/infrahub/core/diff/merger/merger.py`) running the range query from `(default branch, merge_started_at)` with **no** `get_affected_node_uuids` list (contracts §3/§4). - [X] T027a [US3] **Unify the in-process rollback onto the range query** (contracts §3 "Decision — unify both paths"). Point `MergeRollbackHandler` (`backend/infrahub/core/merge/rollback_handler.py`) at the T027 range entry keyed on `(destination_branch, merge_started_at)` instead of the UUID-scoped `GraphMerger.rollback()`; keep the handler's in-memory restore (schema registry + branch object + write-blocker, which out-of-process recovery does not need because it reloads from DB). Then **delete** the now-dead UUID-scoped paths: `GraphMerger.rollback()` (`graph_merger.py`), the old `DiffMerger.rollback()` and its `self._affected_node_uuids` instance field (`merger.py:202`, set at `:111`) — **keep** `get_affected_node_uuids` / the local list, still used for the in-merge metadata update (`merger.py:104-124`). The orchestrator already has the only rollback trigger (its `except` → handler); the inner `GraphMerger.merge()` self-rollback was already removed on the refactor branch. **Test impact:** `backend/tests/integration/diff/test_merge_rollback.py` — `BrokenGraphMerger.rollback()` forwarding becomes irrelevant once `GraphMerger.rollback()` is gone; the test must still drive a partial merge and assert the (now range-based) rollback restores pre-merge state. Covers the post-migration in-process-failure gap noted in contracts §3. - [X] T027b [US3] Restore the destination branch's pre-merge `schema_changed_at` during in-process rollback in `backend/infrahub/core/merge/rollback_handler.py`. Today the handler resets the registry schema to `pre_merge_schema` then calls `destination_branch.update_schema_hash()` (no `at`), which — because the in-merge schema update already bumped the branch's hash to the post-merge value — stamps `schema_changed_at` to *rollback time* rather than restoring the pre-merge value (the schema content is correctly restored; only the timestamp is wrong). Capture the pre-merge `schema_changed_at` (and `schema_hash`) at the same point `pre_merge_schema` / `pre_merge_branched_from` are captured in the orchestrator, pass them into `rollback()`, and restore them literally — symmetric with the existing `pre_merge_branched_from` restore. `pre_merge_schema` can also recompute its own hash, so no extra schema load is needed. Low impact (the only consumer, the GraphQL schema-staleness check in `graphql/app.py`, just triggers a harmless reload of the restored schema), so deferred — not blocking the refactor. NB: the analogous pre-existing path `SchemaUpdateCoordinator._restore_registry_state` (`update_coordinator.py`) has the same behavior and should be fixed alongside. -- [ ] T028 [US3] Implement `MergeFailureRecovery.recover(confirmed)` in `backend/infrahub/core/merge/failure_recovery.py`: detect **both** a `MERGE_FAILED` branch **and** a stuck-`MERGING` branch with a dead/absent lock holder (FR-016, clarification) → range rollback → reset branch to `OPEN` → reset PC to `OPEN` → delete `merge:protected`; return a `RecoveryReport`; handle no-failure (FR-023), orphaned/branch-removed (FR-024), idempotent re-run (FR-022). -- [ ] T029 [US3] In `recover()`, find/reset the associated proposed change via node-manager filter (`source_branch__value == `, `state__value == "merging"`) → `state = "open"` + save; proceed without one for a direct branch merge (FR-020; research R7). -- [ ] T030 [US3] Create the `infrahub recover` CLI in `backend/infrahub/cli/recover.py` (AsyncTyper, mirrors `cli/db.py`): config load → init_db → initialize_registry → build `MergeFailureRecovery` → `rich.Console` report → `typer.confirm` unless `--yes/-y` → print `RecoveryOutcome`; close DB in `finally`. Register in `backend/infrahub/cli/__init__.py` (FR-015/016/017). -- [ ] T031 [US3] Enforce the `MERGE_FAILED` **delete block at the mutation gate** in `backend/infrahub/graphql/middleware.py` (`MERGE_FAILED` granted no mutation exception, incl. `BranchDelete`); leave `Branch.delete()` unchanged (FR-014). -- [ ] T032 [P] [US3] Component test in `backend/tests/component/core/merge/test_recovery_rollback.py`: drive a merge, hand-set the marker, run `recover()` → graph diff vs pre-merge empty **and** touched-node `updated_at/by` restored; a variant that raises mid-`merge_graph` recovers the partial graph; re-run is idempotent (SC-008, SC-010). -- [ ] T033 [P] [US3] Component test in `backend/tests/component/core/merge/test_recovery_edge_cases.py`: no-failure → nothing to recover (FR-023); orphaned/branch-removed → cleared without crashing (FR-024); stuck-`MERGING`-dead-lock → recovered under confirmation (FR-016); declined → no changes (US3 #3); **delete of a `MERGE_FAILED` branch rejected, then allowed after recovery** (SC-007). -- [ ] T034 [US3] Integration-docker test (recovery half) in `backend/tests/integration_docker/test_merge_kill_recovery.py`: after `MERGE_FAILED`, `infrahub recover --yes` → default-branch writes succeed, branch re-merges (SC-009). +- [X] T028 [US3] Implement `MergeFailureRecovery.recover(confirmed)` in `backend/infrahub/core/merge/failure_recovery.py`: detect **both** a `MERGE_FAILED` branch **and** a stuck-`MERGING` branch with a dead/absent lock holder (FR-016, clarification) → range rollback → reset branch to `OPEN` → reset PC to `OPEN` → delete `merge:protected`; return a `RecoveryReport`; handle no-failure (FR-023), orphaned/branch-removed (FR-024), idempotent re-run (FR-022). +- [X] T029 [US3] In `recover()`, find/reset the associated proposed change via node-manager filter (`source_branch__value == `, `state__value == "merging"`) → `state = "open"` + save; proceed without one for a direct branch merge (FR-020; research R7). +- [X] T030 [US3] Create the `infrahub recover` CLI in `backend/infrahub/cli/recover.py` (AsyncTyper, mirrors `cli/db.py`): config load → init_db → initialize_registry → build `MergeFailureRecovery` → `rich.Console` report → `typer.confirm` unless `--yes/-y` → print `RecoveryOutcome`; close DB in `finally`. Register in `backend/infrahub/cli/__init__.py` (FR-015/016/017). +- [X] T031 [US3] Enforce the `MERGE_FAILED` **delete block at the mutation gate** in `backend/infrahub/graphql/middleware.py` (`MERGE_FAILED` granted no mutation exception, incl. `BranchDelete`); leave `Branch.delete()` unchanged (FR-014). +- [X] T032 [P] [US3] Component test in `backend/tests/component/core/merge/test_recovery_rollback.py`: drive a merge, hand-set the marker, run `recover()` → graph diff vs pre-merge empty **and** touched-node `updated_at/by` restored; a variant that raises mid-`merge_graph` recovers the partial graph; re-run is idempotent (SC-008, SC-010). +- [X] T033 [P] [US3] Component test in `backend/tests/component/core/merge/test_recovery_edge_cases.py`: no-failure → nothing to recover (FR-023); orphaned/branch-removed → cleared without crashing (FR-024); stuck-`MERGING`-dead-lock → recovered under confirmation (FR-016); declined → no changes (US3 #3); **delete of a `MERGE_FAILED` branch rejected, then allowed after recovery** (SC-007). +- [X] T034 [US3] Integration-docker test (recovery half) in `backend/tests/integration_docker/test_merge_kill_recovery.py`: after `MERGE_FAILED`, `infrahub recover --yes` → default-branch writes succeed, branch re-merges (SC-009). - [ ] T035 [US3] **[ASK-FIRST]** Add RANGE `IndexItem` entries for edge `from`/`to` + a node `updated_at` index in `backend/infrahub/core/graph/index.py` (research R8). **Jira IFC-2715**. - [ ] T036 [US3] **[ASK-FIRST]** Add the graph migration creating the new indexes (under `backend/infrahub/core/migrations/graph/`), bumping the graph version. **Jira IFC-2715**. - [ ] T037 [US3] **[ASK-FIRST]** Update schema-migration queries that bump vertex `updated_at/by` (e.g. `core/migrations/schema/attribute_kind_update.py`, `core/migrations/query/attribute_add.py`, `node_duplicate.py`, `node_remove.py`, …) to co-write `previous_updated_at/by`, mirroring `DiffMergeMetadataQuery` (research R8). **Jira IFC-2716**. -- [ ] T038 [US3] Add a changelog fragment for **`infrahub recover`** under `changelog/`. +- [X] T038 [US3] Add a changelog fragment for **`infrahub recover`** under `changelog/`. **Checkpoint**: An operator can fully recover a failed merge and re-merge; metadata is restored; recovery is idempotent.