fix(recompute): refresh a reader's derived values when its read peer is deleted

backend/infrahub/core/merge/recompute_coalescing.py

@@ -222,24 +222,64 @@ def build(self, *, changes: Iterable[MergeChange], branch: str) -> CoalescedReco
 
     def _resolve_targets(self, *, signature: ChangeSignature) -> Iterator[_ResolvedTarget]:
         if signature.action == CREATED:
-            include_self, include_cross = True, False
-            fields: frozenset[str] | None = None
-            precise = True
-        elif signature.action == UPDATED:
-            include_self, include_cross = False, True
-            # An update with no recorded fields cannot be scoped, so fall back to every
-            # field rather than risk missing a reader (over-recompute is safe).
-            fields = signature.changed_fields or None
-            precise = bool(signature.changed_fields)
-        elif signature.action == DELETED:
-            include_self, include_cross = False, True
-            fields = None
-            precise = True
-        else:
-            raise ValueError(f"Unknown change action: {signature.action!r}")
+            yield from self._derive_family_targets(
+                kind=signature.kind, fields=None, include_self=True, include_cross=False, precise=True
+            )
+            return
+        if signature.action == DELETED:
+            yield from self._derive_family_targets(
+                kind=signature.kind, fields=None, include_self=False, include_cross=True, precise=True
+            )
+            return
+        if signature.action == UPDATED:
+            if not signature.changed_fields:
+                # No fields to scope on: recompute self and cross, since the unknown change may be a
+                # relationship the node reads and under-recompute is not acceptable.
+                yield from self._derive_family_targets(
+                    kind=signature.kind, fields=None, include_self=True, include_cross=True, precise=False
+                )
+                return
+            # Cross-node readers only; the node refreshed its own values inline on the save.
+            yield from self._derive_family_targets(
+                kind=signature.kind,
+                fields=signature.changed_fields,
+                include_self=False,
+                include_cross=True,
+                precise=True,
+            )
+            # A deleted peer closes the edge without saving the reader, so nothing recomputed it inline;
+            # recompute the reader's own values across the relationship, by its own id.
+            relationship_fields = self._changed_relationship_fields(
+                kind=signature.kind, changed_fields=signature.changed_fields
+            )
+            if relationship_fields:
+                yield from self._derive_family_targets(
+                    kind=signature.kind,
+                    fields=relationship_fields,
+                    include_self=True,
+                    include_cross=False,
+                    precise=True,
+                )
+            return
+        raise ValueError(f"Unknown change action: {signature.action!r}")
 
+    def _changed_relationship_fields(self, *, kind: str, changed_fields: frozenset[str]) -> frozenset[str] | None:
+        """Return the changed fields that name a relationship on ``kind`` (None if none)."""
+        node_schema = self.schema_branch.get_node(name=kind, duplicate=False)
+        matched = changed_fields & {relationship.name for relationship in node_schema.relationships}
+        return frozenset(matched) or None
+
+    def _derive_family_targets(
+        self,
+        *,
+        kind: str,
+        fields: frozenset[str] | None,
+        include_self: bool,
+        include_cross: bool,
+        precise: bool,
+    ) -> Iterator[_ResolvedTarget]:
         yield from self._resolve_computed_targets(
-            kind=signature.kind,
+            kind=kind,
             fields=fields,
             include_self=include_self,
             include_cross=include_cross,
@@ -248,7 +288,7 @@ def _resolve_targets(self, *, signature: ChangeSignature) -> Iterator[_ResolvedT
 
         for display_target in derive_display_label_targets(
             display_labels=self.schema_branch.display_labels,
-            kind=signature.kind,
+            kind=kind,
             changed_fields=fields,
             include_self=include_self,
             include_cross=include_cross,
@@ -264,7 +304,7 @@ def _resolve_targets(self, *, signature: ChangeSignature) -> Iterator[_ResolvedT
 
         for hfid_target in derive_hfid_targets(
             hfids=self.schema_branch.hfids,
-            kind=signature.kind,
+            kind=kind,
             changed_fields=fields,
             include_self=include_self,
             include_cross=include_cross,

backend/tests/component/merge_recompute_coalescing/test_merge_submits_delete_peer.py

@@ -0,0 +1,92 @@
+"""Deleting a read peer must recompute the reader by its own id, not only by a reverse lookup."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+from infrahub import lock
+from infrahub.auth.session import AccountSession
+from infrahub.auth.types import AuthType
+from infrahub.context import InfrahubContext
+from infrahub.core.branch.tasks import merge_branch
+from infrahub.core.diff.coordinator import DiffCoordinator
+from infrahub.core.initialization import create_branch
+from infrahub.core.manager import NodeManager
+from infrahub.core.node import Node
+from infrahub.dependencies.registry import get_component_registry
+from infrahub.workers.dependencies import build_cache, build_database, build_event_service, build_workflow
+from infrahub.workflows.catalogue import COMPUTED_ATTRIBUTE_PROCESS_JINJA2, DISPLAY_LABELS_PROCESS_JINJA2
+from tests.adapters.cache import MemoryCache
+from tests.adapters.event import MemoryInfrahubEvent
+from tests.adapters.workflow import WorkflowRecorder
+from tests.helpers.merge_recompute.dataset import PROFILE_NODE_KIND, PROFILE_PEER_KIND, build_profile_schema
+from tests.helpers.schema import load_schema
+
+if TYPE_CHECKING:
+    from fast_depends import Provider
+
+    from infrahub.core.branch import Branch
+    from infrahub.core.schema.schema_branch import SchemaBranch
+    from infrahub.database import InfrahubDatabase
+
+
+async def test_merge_delete_peer_coalesces_reader_recompute_by_own_id(
+    db: InfrahubDatabase,
+    default_branch: Branch,
+    register_core_models_schema: SchemaBranch,
+    dependency_provider: Provider,
+) -> None:
+    lock.initialize_lock(local_only=True)
+
+    # Optional peer so it can be deleted while the reader survives.
+    schema = build_profile_schema()
+    node_schema = next(node for node in schema.nodes if node.kind == PROFILE_NODE_KIND)
+    node_schema.relationships[0].optional = True
+    await load_schema(db=db, schema=schema, update_db=True)
+
+    peer = await Node.init(db=db, schema=PROFILE_PEER_KIND, branch=default_branch)
+    await peer.new(db=db, name="beta")
+    await peer.save(db=db)
+    # Several readers of the same peer, to prove the deletion coalesces them rather than fanning out.
+    reader_ids: list[str] = []
+    for index in range(3):
+        reader = await Node.init(db=db, schema=PROFILE_NODE_KIND, branch=default_branch)
+        await reader.new(db=db, name=f"reader-{index}", peer=peer)
+        await reader.save(db=db)
+        reader_ids.append(reader.id)
+
+    branch = await create_branch(branch_name="delete-peer-submit", db=db)
+    peer_on_branch = await NodeManager.get_one(id=peer.id, db=db, branch=branch)
+    assert peer_on_branch is not None
+    await peer_on_branch.delete(db=db)
+
+    component_registry = get_component_registry()
+    diff_coordinator = await component_registry.get_component(DiffCoordinator, db=db, branch=branch)
+    await diff_coordinator.update_branch_diff(base_branch=default_branch, diff_branch=branch)
+
+    recorder = WorkflowRecorder()
+    event_recorder = MemoryInfrahubEvent()
+    cache = MemoryCache()
+    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
+        dependency_provider.scope(build_event_service, lambda: event_recorder),
+        dependency_provider.scope(build_workflow, lambda: recorder),
+        dependency_provider.scope(build_cache, lambda: cache),
+    ):
+        await merge_branch(branch=branch.name, context=context)
+
+    # The reverse lookup from the deleted peer finds no readers once its edges close, so the readers
+    # must be recomputed by their own ids, coalesced into one submission per family.
+    for workflow in (COMPUTED_ATTRIBUTE_PROCESS_JINJA2, DISPLAY_LABELS_PROCESS_JINJA2):
+        own_id_submissions = [
+            call
+            for call in recorder.get_submit_calls_for(workflow)
+            if call["parameters"]["node_kind"] == PROFILE_NODE_KIND
+        ]
+        assert len(own_id_submissions) == 1, f"{workflow.name} fanned out instead of coalescing the readers"
+        assert sorted(own_id_submissions[0]["parameters"]["object_ids"]) == sorted(reader_ids)

backend/tests/integration_docker/test_merge_recompute.py

@@ -34,15 +34,6 @@ async def _wait_until(predicate: Callable[[], Awaitable[bool]], *, seconds: int
             await sleep(2)
 
 
-async def _became_true(predicate: Callable[[], Awaitable[bool]], *, seconds: int = 60) -> bool:
-    """Return True if the predicate becomes true within the timeout, False if it never does."""
-    try:
-        await _wait_until(predicate, seconds=seconds)
-    except TimeoutError:
-        return False
-    return True
-
-
 async def _wait_until_merged(*, client: InfrahubClient, branch_name: str, seconds: int = 120) -> None:
     """Wait until the branch reaches the MERGED status, so the test asserts on a merge that applied."""
 
@@ -219,32 +210,23 @@ async def _on_destination() -> bool:
         assert final.display_label == "cnode via gamma"
         assert final.hfid == ["cnode"]
 
-    @pytest.mark.xfail(
-        strict=True,
-        reason=(
-            "the recompute locates readers with a reverse relationship query that returns nothing "
-            "once the deleted peer's edges are closed, so the reader keeps a value that still names "
-            "the deleted peer"
-        ),
-    )
     async def test_deleting_read_peer_refreshes_reader_after_merge(self, client: InfrahubClient) -> None:
-        """After a read peer is deleted and merged, the reader's derived values should stop naming it."""
+        """After a read peer is deleted and merged, the reader's computed attribute stops naming it."""
         schema = build_profile_schema_dict()
         node_schema = schema["nodes"][1]
         node_schema["relationships"][0]["optional"] = True
-        # Self-only display label so the scenario exercises the stale computed value, not the separate
-        # missing-peer diff crash that is fixed on its own path.
+        # Self-only display label: refreshing a cross-relationship display label is handled separately.
         node_schema["display_label"] = "{{ name__value }}"
         await client.schema.load(schemas=[schema], wait_until_converged=True)
 
         peer = await client.create(kind=PROFILE_PEER_KIND, data={"name": "beta"})
         await peer.save()
-        node = await client.create(kind=PROFILE_NODE_KIND, data={"name": "node2", "peer": peer})
+        node = await client.create(kind=PROFILE_NODE_KIND, data={"name": "dnode", "peer": peer})
         await node.save()
 
         async def _reader_initial() -> bool:
             refreshed = await client.get(kind=PROFILE_NODE_KIND, id=node.id)
-            return refreshed.summary.value == "node2 on beta"
+            return refreshed.summary.value == "dnode on beta"
 
         await _wait_until(_reader_initial)
 
@@ -256,11 +238,16 @@ async def _reader_initial() -> bool:
         assert merged
         await _wait_until_merged(client=client, branch_name=branch.name)
 
-        async def _reader_no_longer_names_peer() -> bool:
+        async def _reader_refreshed() -> bool:
             refreshed = await client.get(kind=PROFILE_NODE_KIND, id=node.id)
-            return refreshed.summary.value != "node2 on beta"
+            return refreshed.summary.value == "dnode on None"
 
-        assert await _became_true(_reader_no_longer_names_peer, seconds=60)
+        await _wait_until(_reader_refreshed)
+
+        final = await client.get(kind=PROFILE_NODE_KIND, id=node.id)
+        assert final.summary.value == "dnode on None"
+        # HFID reads only the local name, so the delete leaves it unchanged.
+        assert final.hfid == ["dnode"]
 
     # Multi-level chain: level i reads level i-1 across the source relationship.
 

backend/tests/unit/core/merge/test_build_coalesced_recompute.py

@@ -131,18 +131,32 @@ def test_changes_to_same_target_are_deduplicated() -> None:
     assert _lookups(computed) == {(PROFILE_PEER_KIND, "peer__ids", frozenset({"peer-0", "peer-1"}))}
 
 
-def test_update_without_fields_is_a_bounded_fallback() -> None:
-    """An update with no recorded fields recomputes every cross-node reader and is marked imprecise."""
+def test_unscoped_update_is_a_bounded_fallback() -> None:
+    """An unscoped update imprecisely recomputes cross-node readers and the node's own derived values."""
     builder = CoalescedRecomputeBuilder(schema_branch=_profile_schema_branch())
-    changes = [MergeChange(node_id="peer-0", kind=PROFILE_PEER_KIND, action="updated", changed_fields=frozenset())]
 
-    result = builder.build(changes=changes, branch="main")
-
-    # A peer's readers are the node computed summary and display label that read it across the
-    # relationship; no human-friendly id reads the peer, so even the unscoped fallback omits it.
-    assert set(_by_identity(result)) == {
+    # A peer: its cross-node readers (node summary and display) plus its own derived values.
+    peer_result = builder.build(
+        changes=[MergeChange(node_id="peer-0", kind=PROFILE_PEER_KIND, action="updated", changed_fields=frozenset())],
+        branch="main",
+    )
+    assert set(_by_identity(peer_result)) == {
         (COMPUTED_ATTRIBUTE, PROFILE_NODE_KIND, "summary"),
         (DISPLAY_LABEL, PROFILE_NODE_KIND, None),
+        (DISPLAY_LABEL, PROFILE_PEER_KIND, None),
+        (HFID, PROFILE_PEER_KIND, None),
     }
-    assert result.fallback_used is True
-    assert all(target.precise is False for target in result.targets)
+    assert peer_result.fallback_used is True
+    assert all(target.precise is False for target in peer_result.targets)
+
+    # A node: its own derived values, keyed by its own id.
+    node_result = builder.build(
+        changes=[MergeChange(node_id="node-0", kind=PROFILE_NODE_KIND, action="updated", changed_fields=frozenset())],
+        branch="main",
+    )
+    by_identity = _by_identity(node_result)
+    own = {(PROFILE_NODE_KIND, "ids", frozenset({"node-0"}))}
+    assert _lookups(by_identity[COMPUTED_ATTRIBUTE, PROFILE_NODE_KIND, "summary"]) == own
+    assert _lookups(by_identity[DISPLAY_LABEL, PROFILE_NODE_KIND, None]) == own
+    assert _lookups(by_identity[HFID, PROFILE_NODE_KIND, None]) == own
+    assert node_result.fallback_used is True