fix(backend): branch registry updates

backend/infrahub/core/branch/tasks.py

@@ -101,13 +101,15 @@ async def migrate_branch(branch: str, context: InfrahubContext, send_events: boo
             log.info(f"No migrations detected for branch '{obj.name}'")
             obj.graph_version = GRAPH_VERSION
             await obj.save(db=db)
+            registry.branch[obj.name] = obj
             return
 
         # Branch status will remain as so if the migration process fails
         # This will help user to know that a branch is in an invalid state to be used properly and that actions need to be taken
         if obj.status != BranchStatus.NEED_UPGRADE_REBASE:
             obj.status = BranchStatus.NEED_UPGRADE_REBASE
             await obj.save(db=db)
+            registry.branch[obj.name] = obj
 
         try:
             log.info(f"Running migrations for branch '{obj.name}'")
@@ -120,6 +122,7 @@ async def migrate_branch(branch: str, context: InfrahubContext, send_events: boo
             obj.status = BranchStatus.OPEN
         obj.graph_version = GRAPH_VERSION
         await obj.save(db=db)
+        registry.branch[obj.name] = obj
 
     if send_events:
         event_context = context.to_event_context()

backend/infrahub/graphql/mutations/branch.py

@@ -253,6 +253,9 @@ async def mutate(
         async with graphql_context.db.start_transaction() as db:
             await obj.save(db=db, user_id=graphql_context.active_account_session.account_id)
 
+        # update registry after txn commit, so it cannot diverge from db on failure
+        registry.branch[obj.name] = obj
+
         return cls(ok=True)
 
 

backend/infrahub/tasks/registry.py

@@ -7,6 +7,7 @@
 from infrahub.core.constants import GLOBAL_BRANCH_NAME
 from infrahub.graphql.registry import registry as graphql_registry
 from infrahub.log import get_logger
+from infrahub.utils import log_exception_guard
 from infrahub.worker import WORKER_IDENTITY
 
 if TYPE_CHECKING:
@@ -95,10 +96,12 @@ async def refresh_branches(db: InfrahubDatabase) -> None:
                 # have an associated schema
                 continue
 
-            if active_branch.name in registry.branch:
-                await update_branch_registry(db=db, branch=active_branch)
-            else:
-                await create_branch_registry(db=db, branch=active_branch)
+            # Absorb a failure on one branch rather than abandoning the sweep
+            with log_exception_guard(log, f"Failed to refresh branch {active_branch.name!r} in the registry"):
+                if active_branch.name in registry.branch:
+                    await update_branch_registry(db=db, branch=active_branch)
+                else:
+                    await create_branch_registry(db=db, branch=active_branch)
 
         purged_branches = await registry.purge_inactive_branches(db=db, active_branches=active_branches)
         purged_branches.update(

backend/tests/component/core/test_branch_migrate.py

@@ -0,0 +1,46 @@
+from uuid import uuid4
+
+from fast_depends import Provider
+
+from infrahub.auth.session import AccountSession
+from infrahub.auth.types import AuthType
+from infrahub.context import InfrahubContext
+from infrahub.core import registry
+from infrahub.core.branch import Branch
+from infrahub.core.branch.enums import BranchStatus
+from infrahub.core.branch.tasks import migrate_branch
+from infrahub.core.graph import GRAPH_VERSION
+from infrahub.core.initialization import create_branch
+from infrahub.core.schema.schema_branch import SchemaBranch
+from infrahub.database import InfrahubDatabase
+from infrahub.workers.dependencies import build_database
+
+
+async def test_migrate_branch_publishes_migrated_branch(
+    db: InfrahubDatabase,
+    default_branch: Branch,
+    car_person_schema: SchemaBranch,
+    dependency_provider: Provider,
+) -> None:
+    """The flow migrates an instance of its own, so it has to publish that instance itself."""
+    branch = await create_branch(db=db, branch_name="migrate-branch")
+    assert registry.branch[branch.name] is branch
+
+    # A branch as an upgrade leaves it behind: its graph version trails the application
+    branch.graph_version = GRAPH_VERSION - 1
+    await branch.save(db=db)
+
+    context = InfrahubContext.init(
+        branch=default_branch,
+        account=AccountSession(account_id=str(uuid4()), auth_type=AuthType.NONE),
+    )
+    with dependency_provider.scope(build_database, lambda singleton=True: db):  # noqa: ARG005
+        await migrate_branch(branch=branch.name, context=context, send_events=False)
+
+    migrated_branch = await Branch.get_by_name(db=db, name=branch.name)
+    assert migrated_branch.graph_version == GRAPH_VERSION
+    assert migrated_branch.status is BranchStatus.OPEN
+
+    published_branch = registry.branch[branch.name]
+    assert published_branch.graph_version == GRAPH_VERSION
+    assert published_branch.status is BranchStatus.OPEN

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

@@ -588,6 +588,11 @@ async def test_branch_update_description(
 
     branch4_updated = await Branch.get_by_name(db=db, name="branch4")
 
+    # The mutation publishes what it saved, so the cache reflects the committed description
+    cached_branch4 = registry.branch["branch4"]
+    assert cached_branch4.description == "testing"
+    assert cached_branch4.description == branch4_updated.description
+
     assert branch4.updated_at == branch4.created_at
     assert branch4_updated.description == "testing"
     assert branch4.updated_at

backend/tests/component/tasks/test_registry.py

@@ -0,0 +1,51 @@
+import pytest
+
+from infrahub.core import registry
+from infrahub.core.branch import Branch
+from infrahub.core.constants import NULL_VALUE
+from infrahub.core.initialization import create_branch
+from infrahub.core.schema.schema_branch import SchemaBranch
+from infrahub.core.timestamp import Timestamp
+from infrahub.database import InfrahubDatabase
+from infrahub.tasks.registry import refresh_branches
+
+
+async def test_refresh_branches_continues_past_a_branch_it_cannot_refresh(
+    db: InfrahubDatabase,
+    default_branch: Branch,
+    car_person_schema: SchemaBranch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The sweep is the only thing that repairs a stale cache entry, so one bad branch must not end it."""
+    broken_branch = await create_branch(db=db, branch_name="broken-branch")
+    stale_branch = await create_branch(db=db, branch_name="stale-branch")
+
+    # A branch row whose schema hash never reached storage — how the global branch and a graph
+    # predating the field look. Reading it back raises from Branch.active_schema_hash
+    await db.execute_query(
+        query="MATCH (n:Branch {name: $branch_name}) SET n.schema_hash = $null_value",
+        params={"branch_name": broken_branch.name, "null_value": NULL_VALUE},
+    )
+    assert (await Branch.get_by_name(db=db, name=broken_branch.name)).schema_hash is None
+
+    # Something for the sweep to pick up: a rebase timestamp that only exists in the database
+    rebased_branch = await Branch.get_by_name(db=db, name=stale_branch.name)
+    rebased_branch.branched_from = Timestamp().to_string()
+    await rebased_branch.save(db=db)
+    assert registry.branch[stale_branch.name].branched_from != rebased_branch.branched_from
+
+    with caplog.at_level("ERROR", logger="infrahub"):
+        await refresh_branches(db=db)
+
+    # The branch it gave up on has to be reported, with its traceback: absorbed is not silent
+    failures = [
+        record.msg
+        for record in caplog.records
+        if isinstance(record.msg, dict)
+        and record.msg.get("event") == f"Failed to refresh branch '{broken_branch.name}' in the registry"
+    ]
+    assert len(failures) == 1
+    assert failures[0]["level"] == "error"
+    assert failures[0]["exc_info"]
+
+    assert registry.branch[stale_branch.name].branched_from == rebased_branch.branched_from

changelog/+branch-update-registry-cache.fixed.md

@@ -0,0 +1 @@
+Update the in-memory cache for branches after a branch change is saved to prevent the cache diverging from the database. The BranchUpdate mutation and the Prefect task to run database migrations against a branch are both fixed.

changelog/+refresh-branches-per-branch-failure.fixed.md

@@ -0,0 +1 @@
+A branch that cannot be refreshed into a worker's in-memory registry no longer aborts the periodic branch refresh. The failure is logged and the remaining branches are still refreshed.