From ef8fd5cc2f3618a85d9b62796abb2307059809c6 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Thu, 23 Jul 2026 11:55:46 +0200 Subject: [PATCH 01/13] perf(computed-attribute): bulk-write Python transform recompute Recompute a Python-transform computed attribute for a batch of nodes by initializing the transform's git repository once for the whole batch instead of once per node, and persisting the results in a single bulk write (chunked node.save) rather than a GraphQL mutation per node. A node whose recomputed value is unchanged is skipped, so it neither writes nor fans out a further recompute. A node whose transform raises or returns a non-string value is isolated: it is skipped with its previous value left in place, so one failing node no longer aborts the batch and drops the other nodes' writes. Co-Authored-By: Claude Opus 4.8 --- backend/infrahub/computed_attribute/tasks.py | 141 +++++++++++------- .../unit/computed_attribute/test_tasks.py | 40 +++++ ...mputed-attribute-bulk-recompute.changed.md | 1 + 3 files changed, 128 insertions(+), 54 deletions(-) create mode 100644 backend/tests/unit/computed_attribute/test_tasks.py create mode 100644 changelog/+python-computed-attribute-bulk-recompute.changed.md diff --git a/backend/infrahub/computed_attribute/tasks.py b/backend/infrahub/computed_attribute/tasks.py index 2ff92cbd91..92dbcc570b 100644 --- a/backend/infrahub/computed_attribute/tasks.py +++ b/backend/infrahub/computed_attribute/tasks.py @@ -3,8 +3,7 @@ from typing import TYPE_CHECKING from infrahub_sdk.exceptions import URLNotFoundError -from prefect import flow, task -from prefect.cache_policies import NONE +from prefect import flow from prefect.client.orchestration import get_client as get_prefect_client from prefect.logging import get_run_logger @@ -52,6 +51,7 @@ if TYPE_CHECKING: from infrahub.core.schema.computed_attribute import ComputedAttribute from infrahub.database import InfrahubDatabase + from infrahub.git.repository import InfrahubReadOnlyRepository, InfrahubRepository from infrahub.graphql.analyzer import GraphQLQueryReport @@ -78,24 +78,6 @@ async def _reconcile_python_computed_attribute_automations(db: InfrahubDatabase) log.debug("Reconciled Python computed-attribute node-input automations") -UPDATE_ATTRIBUTE = """ -mutation UpdateAttribute( - $id: String!, - $kind: String!, - $attribute: String!, - $value: String! - $context_account_id: String! - ) { - InfrahubUpdateComputedAttribute( - context: {account: {id: $context_account_id}}, - data: {id: $id, attribute: $attribute, value: $value, kind: $kind} - ) { - ok - } -} -""" - - def _resolve_changed_elements( changed_elements: ChangedElementsPayload | None, ) -> ChangedElementSet | None: @@ -116,34 +98,30 @@ def _transform_read_set_from_query_report(report: GraphQLQueryReport) -> Transfo return TransformReadSet.from_read_fields({kind: access.fields for kind, access in report.requested_read.items()}) -@task(name="computed-attribute-process-transform-for-node", cache_policy=NONE) -async def process_transform_for_node( +async def _transform_value_for_node( + *, branch_name: str, object_id: str, - node_kind: str, attribute_name: str, query_id: str, transform_timeout: int | None, - repository_id: str, - repository_name: str, - repository_kind: str, commit: str | None, file_path: str, class_name: str, convert_query_response: bool, context: EventContext, -) -> None: + repo: InfrahubReadOnlyRepository | InfrahubRepository, +) -> AttributeValueWrite: + """Run the transform for one node against an already-initialized repository. + + The query is run with ``update_group`` so the node is (re)registered as a subscriber of the + transform's query group, which is how a later change to any node the query reads triggers this + node's recompute. The recomputed value is returned to be persisted in bulk with the rest of the + batch rather than written here. + """ client = get_client() client.request_context = context.to_request_context() - repo = await get_initialized_repo( - client=client, - repository_id=repository_id, - name=repository_name, - repository_kind=repository_kind, - commit=commit, - ) - data = await client.query_gql_query( name=query_id, branch_name=branch_name, @@ -152,7 +130,7 @@ async def process_transform_for_node( subscribers=[object_id], ) - transformed_data = await repo.execute_python_transform.with_options(timeout_seconds=transform_timeout)( + value = await repo.execute_python_transform.with_options(timeout_seconds=transform_timeout)( client=client, branch_name=branch_name, commit=commit, @@ -161,17 +139,30 @@ async def process_transform_for_node( convert_query_response=convert_query_response, ) # type: ignore[call-overload] - await client.execute_graphql( - query=UPDATE_ATTRIBUTE, - variables={ - "id": object_id, - "kind": node_kind, - "attribute": attribute_name, - "value": transformed_data, - "context_account_id": context.account_id, - }, - branch_name=branch_name, - ) + return AttributeValueWrite(node_id=object_id, field=attribute_name, value=value) + + +def _partition_transform_results( + results: list[tuple[str, AttributeValueWrite | Exception]], +) -> tuple[list[AttributeValueWrite], list[tuple[str, str]]]: + """Split per-node transform results into values to persist and nodes to skip. + + A node is skipped, leaving its previous value in place, when its transform raised or produced a + non-string value. Isolating failures this way keeps one failing node from blocking the rest of + the batch, and preserves the contract that only a real string is persisted rather than silently + overwriting a value with null. Returns the writes to persist and ``(node_id, reason)`` pairs + describing each skip. + """ + writes: list[AttributeValueWrite] = [] + skipped: list[tuple[str, str]] = [] + for object_id, result in results: + if isinstance(result, Exception): + skipped.append((object_id, f"transform raised {result!r}")) + elif not isinstance(result.value, str): + skipped.append((object_id, f"transform returned {type(result.value).__name__}, expected a string")) + else: + writes.append(result) + return writes, skipped @flow( @@ -188,8 +179,25 @@ async def process_transform( object_ids: list[str] | None = None, updated_fields: list[str] | None = None, # noqa: ARG001 ) -> None: + """Recompute one or more Python computed attributes for a batch of nodes. + + The transform's git repository is initialized once for the whole batch instead of once per node, + each node's value is computed concurrently, and the results are persisted in a single bulk write + instead of a GraphQL mutation per node. The bulk write skips any node whose recomputed value is + unchanged, so an unchanged node neither writes nor fans out a further recompute. + + A node whose transform raises or returns a non-string value is skipped with its previous value + left in place, so one failing node cannot block the rest of the batch from being persisted. + + Raises: + ValueError: if a computed attribute has no transform configured or the transform cannot be fetched. + + """ + log = get_run_logger() all_ids = list({*([object_id] if object_id else []), *(object_ids or [])}) await add_tags(branches=[branch_name], nodes=all_ids) + if not all_ids: + return client = get_client() client.request_context = context.to_request_context() @@ -203,6 +211,8 @@ async def process_transform( if not transform_attributes: return + dispatcher = await build_bulk_recompute_dispatcher(schema_branch=schema_branch) + for attribute_name, transform_attribute in transform_attributes.items(): if not transform_attribute.transform: raise ValueError(f"No transform configured for computed attribute '{attribute_name}'") @@ -219,26 +229,49 @@ async def process_transform( f"Unable to fetch transform '{transform_attribute.transform}' for computed attribute '{attribute_name}'" ) - batch = await client.create_batch() + # Initialize the repository once and share it across the whole batch rather than paying a + # repository init per node. + repo = await get_initialized_repo( + client=client, + repository_id=transform.repository_id, + name=transform.repository_name, + repository_kind=transform.repository_typename, + commit=transform.repository_commit, + ) + + # return_exceptions keeps one node's failed read or transform from aborting the batch and + # dropping the healthy nodes' writes; node=oid tags each result with the node it belongs to. + batch = await client.create_batch(return_exceptions=True) for oid in all_ids: batch.add( - task=process_transform_for_node, + task=_transform_value_for_node, + node=oid, branch_name=branch_name, object_id=oid, - node_kind=node_kind, attribute_name=attribute_name, query_id=transform.query_name, transform_timeout=transform.timeout, - repository_id=transform.repository_id, - repository_name=transform.repository_name, - repository_kind=transform.repository_typename, commit=transform.repository_commit, file_path=transform.file_path, class_name=transform.class_name, convert_query_response=transform.convert_query_response, context=context, + repo=repo, ) - _ = [r async for _, r in batch.execute()] + results: list[tuple[str, AttributeValueWrite | Exception]] = [ + (oid, result) async for oid, result in batch.execute() + ] + writes, skipped = _partition_transform_results(results) + for skipped_id, reason in skipped: + log.warning(f"Skipping recompute of '{attribute_name}' for node {skipped_id}: {reason}") + + await dispatcher.dispatch( + writes=writes, + branch_name=branch_name, + context=context, + coalesced=False, + recompute_depth=0, + ) @flow( diff --git a/backend/tests/unit/computed_attribute/test_tasks.py b/backend/tests/unit/computed_attribute/test_tasks.py new file mode 100644 index 0000000000..6ef2930ef1 --- /dev/null +++ b/backend/tests/unit/computed_attribute/test_tasks.py @@ -0,0 +1,40 @@ +from typing import Any + +from infrahub.computed_attribute.tasks import _partition_transform_results +from infrahub.core.recompute.bulk_write import AttributeValueWrite + + +def _write(node_id: str, value: Any) -> AttributeValueWrite: + return AttributeValueWrite(node_id=node_id, field="desc", value=value) + + +def test_partition_transform_results_persists_only_string_values() -> None: + """A string value is persisted; a None or non-string value is skipped so the prior value stays.""" + ok = _write("n1", "hello") + null_value = _write("n2", None) + wrong_type = _write("n3", 42) + + writes, skipped = _partition_transform_results([("n1", ok), ("n2", null_value), ("n3", wrong_type)]) + + assert writes == [ok] + reasons = dict(skipped) + assert list(reasons) == ["n2", "n3"] + assert "NoneType" in reasons["n2"] + assert "int" in reasons["n3"] + + +def test_partition_transform_results_isolates_a_failed_node() -> None: + """A node whose transform raised is skipped without dropping the healthy nodes' writes.""" + ok1 = _write("n1", "a") + ok2 = _write("n3", "b") + + writes, skipped = _partition_transform_results([("n1", ok1), ("n2", RuntimeError("boom")), ("n3", ok2)]) + + assert writes == [ok1, ok2] + assert len(skipped) == 1 + assert skipped[0][0] == "n2" + assert "boom" in skipped[0][1] + + +def test_partition_transform_results_handles_empty() -> None: + assert _partition_transform_results([]) == ([], []) diff --git a/changelog/+python-computed-attribute-bulk-recompute.changed.md b/changelog/+python-computed-attribute-bulk-recompute.changed.md new file mode 100644 index 0000000000..a3f627f3d5 --- /dev/null +++ b/changelog/+python-computed-attribute-bulk-recompute.changed.md @@ -0,0 +1 @@ +Recomputing a Python-transform computed attribute for a batch of nodes now initializes the transform's git repository once for the whole batch instead of once per node, and persists the results in a single bulk write instead of a GraphQL mutation per node. A node whose recomputed value is unchanged is now skipped, so it neither writes nor triggers a further recompute. This shortens the trailing recompute after a merge or rebase that affects many nodes of the same kind (for example a device-type change that refreshes every device's computed description). From 3399a2652cc1ac787e4108fa5de9e4a225eaeaf4 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Thu, 23 Jul 2026 12:01:55 +0200 Subject: [PATCH 02/13] test(computed-attribute): characterize Python recompute node fan-out Records that recomputing a Python computed attribute fans out to every node of the kind today. This is a baseline the follow-up scoping work will flip to an affected-only count. Co-Authored-By: Claude Opus 4.8 --- .../test_merge_fanout_python.py | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 backend/tests/component/computed_attribute/test_merge_fanout_python.py diff --git a/backend/tests/component/computed_attribute/test_merge_fanout_python.py b/backend/tests/component/computed_attribute/test_merge_fanout_python.py new file mode 100644 index 0000000000..8f08916c3e --- /dev/null +++ b/backend/tests/component/computed_attribute/test_merge_fanout_python.py @@ -0,0 +1,109 @@ +"""Measure the node fan-out of a Python computed-attribute recompute. + +On merge/rebase a Python computed attribute is refreshed by dispatching one +recompute per affected attribute, which then resolves the nodes to process. This +records how many nodes that resolution selects. Today it selects every node of +the kind regardless of how many changed; once the recompute is scoped to the +changed nodes the same test pins the affected-only count. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from infrahub.computed_attribute.tasks import trigger_update_python_computed_attributes +from infrahub.core.constants import InfrahubKind +from infrahub.core.node import Node +from infrahub.workflows.catalogue import COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM +from tests.component.computed_attribute._base import CAR_PERSON_PYTHON_SCHEMA, ScopedRecomputeTestBase +from tests.helpers.schema import load_schema + +if TYPE_CHECKING: + from infrahub_sdk import InfrahubClient + + from infrahub.core.branch import Branch + from infrahub.core.protocols import CoreAccount + from infrahub.database import InfrahubDatabase + from tests.adapters.workflow import WorkflowRecorder + +CAR_COUNT = 5 + + +class TestMergeFanoutPython(ScopedRecomputeTestBase): + # The fan-out is dispatched as one process-transform per chunk of node ids. + WORKFLOW = COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM + + @pytest.fixture(scope="class") + async def transform_dataset( + self, + db: InfrahubDatabase, + default_branch: Branch, + client: InfrahubClient, + admin_account: CoreAccount, + ) -> set[str]: + """One transform reading TestCar.name, plus CAR_COUNT cars owned by one person. + + Returns the ids of every car, so the test can compare the fan-out against the + full population of the kind. + """ + query = await Node.init(db=db, schema=InfrahubKind.GRAPHQLQUERY) + await query.new( + db=db, + name="query01", + query="query { TestCar { edges { node { name { value } } } } }", + models=["TestCar", "TestPerson"], + ) + await query.save(db=db) + + repo = await Node.init(db=db, schema=InfrahubKind.READONLYREPOSITORY) + await repo.new( + db=db, name="repo01", ref=default_branch.name, commit="commit01", location="location01", queries=[query] + ) + await repo.save(db=db) + + transform = await Node.init(db=db, schema=InfrahubKind.TRANSFORMPYTHON) + await transform.new( + db=db, name="transform01", file_path="transform.py", class_name="Transform", query=query, repository=repo + ) + await transform.save(db=db) + + await load_schema(db=db, schema=CAR_PERSON_PYTHON_SCHEMA, update_db=True) + + owner = await Node.init(db=db, schema="TestPerson") + await owner.new(db=db, name="owner01") + await owner.save(db=db) + + car_ids: set[str] = set() + for index in range(CAR_COUNT): + car = await Node.init(db=db, schema="TestCar") + await car.new(db=db, name=f"car{index}", owner=owner) + await car.save(db=db) + car_ids.add(car.id) + + return car_ids + + def _fanned_out_ids(self, recorder: WorkflowRecorder) -> set[str]: + ids: set[str] = set() + for call in recorder.get_submit_calls_for(self.WORKFLOW): + ids.update(call["parameters"].get("object_ids") or []) + return ids + + async def test_recompute_fans_out_to_every_node_of_the_kind( + self, + transform_dataset: set[str], + workflow_recorder: WorkflowRecorder, + default_branch: Branch, + admin_account: CoreAccount, + ) -> None: + car_ids = transform_dataset + + await trigger_update_python_computed_attributes( + branch_name=default_branch.name, + computed_attribute_name="computed_desc_python", + computed_attribute_kind="TestCar", + context=self._context(admin_account, default_branch), + ) + + assert self._fanned_out_ids(workflow_recorder) == car_ids From 3f3d947ac9480d4e75a673582ec864e4bf52111a Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Thu, 23 Jul 2026 17:09:59 +0200 Subject: [PATCH 03/13] fix(workflows): base add_tags updates on the run's current tags update_flow_run replaces the full tag list, and add_tags built that list from the prefect runtime snapshot, which is frozen at flow start. Any flow that called add_tags more than once silently lost the earlier call's tags: the bulk recompute dispatcher's database-change tag wiped the branch and node tags the process flow had just applied, making its runs invisible to the task API's branch-filtered queries. Read the current tags from the API before merging so successive calls compose instead of overwrite. Co-Authored-By: Claude Opus 4.8 --- backend/infrahub/workflows/utils.py | 5 ++++- changelog/+add-tags-live-state.fixed.md | 1 + 2 files changed, 5 insertions(+), 1 deletion(-) create mode 100644 changelog/+add-tags-live-state.fixed.md diff --git a/backend/infrahub/workflows/utils.py b/backend/infrahub/workflows/utils.py index 521674274b..6a91a410ee 100644 --- a/backend/infrahub/workflows/utils.py +++ b/backend/infrahub/workflows/utils.py @@ -44,7 +44,10 @@ async def add_tags( """ client = get_client(httpx_settings={"verify": get_http().verify_tls()}, sync_client=False) current_flow_run_id = flow_run.id - current_tags: list[str] = flow_run.tags + # Read the run's current tags from the API rather than the runtime context: the runtime + # snapshot is frozen at flow start, and update_flow_run replaces the full tag list, so basing + # the update on the snapshot silently drops tags added by an earlier call in the same run. + current_tags: list[str] = list((await client.read_flow_run(current_flow_run_id)).tags) branch_tags = ( [ WorkflowTag.BRANCH.render(identifier=branch_name) diff --git a/changelog/+add-tags-live-state.fixed.md b/changelog/+add-tags-live-state.fixed.md new file mode 100644 index 0000000000..10e5083371 --- /dev/null +++ b/changelog/+add-tags-live-state.fixed.md @@ -0,0 +1 @@ +Fixed flow-run tags being silently dropped when a workflow tagged its run more than once: later tag updates were based on the tags known at flow start, so they replaced tags added earlier in the same run. Tag updates now build on the run's current tags, keeping branch and node tags visible in the task list for flows that also record a database change. From b7040bbb569ebf06f1f09c294e5a5a10ba643ced Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Thu, 23 Jul 2026 17:55:12 +0200 Subject: [PATCH 04/13] Revert "fix(workflows): base add_tags updates on the run's current tags" This reverts commit de6664dedeb86397315220f3326c4458ad56364d. --- backend/infrahub/workflows/utils.py | 5 +---- changelog/+add-tags-live-state.fixed.md | 1 - 2 files changed, 1 insertion(+), 5 deletions(-) delete mode 100644 changelog/+add-tags-live-state.fixed.md diff --git a/backend/infrahub/workflows/utils.py b/backend/infrahub/workflows/utils.py index 6a91a410ee..521674274b 100644 --- a/backend/infrahub/workflows/utils.py +++ b/backend/infrahub/workflows/utils.py @@ -44,10 +44,7 @@ async def add_tags( """ client = get_client(httpx_settings={"verify": get_http().verify_tls()}, sync_client=False) current_flow_run_id = flow_run.id - # Read the run's current tags from the API rather than the runtime context: the runtime - # snapshot is frozen at flow start, and update_flow_run replaces the full tag list, so basing - # the update on the snapshot silently drops tags added by an earlier call in the same run. - current_tags: list[str] = list((await client.read_flow_run(current_flow_run_id)).tags) + current_tags: list[str] = flow_run.tags branch_tags = ( [ WorkflowTag.BRANCH.render(identifier=branch_name) diff --git a/changelog/+add-tags-live-state.fixed.md b/changelog/+add-tags-live-state.fixed.md deleted file mode 100644 index 10e5083371..0000000000 --- a/changelog/+add-tags-live-state.fixed.md +++ /dev/null @@ -1 +0,0 @@ -Fixed flow-run tags being silently dropped when a workflow tagged its run more than once: later tag updates were based on the tags known at flow start, so they replaced tags added earlier in the same run. Tag updates now build on the run's current tags, keeping branch and node tags visible in the task list for flows that also record a database change. From 654200254757a134c5557a24c35ab47b55995811 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Thu, 23 Jul 2026 17:58:52 +0200 Subject: [PATCH 05/13] fix(computed-attribute): tag process_transform runs with the branch at submission The task API surfaces a flow run only when it carries the branch tag, and a tag update inside the flow rebuilds the tag list from the tags known at flow start: the bulk dispatcher's database-change tag update was rebuilding from the creation tags and dropping the branch tag the flow had added mid-run, hiding the recompute from branch-filtered task queries. Bake the branch tag into the submission itself so it is part of the creation tags and survives every later update. An earlier attempt fixed this inside add_tags by re-reading the run's tags from the API before merging; that doubled the Prefect API round-trips of every tagged flow and measurably slowed storm-sized recomputes, so it was reverted in favor of tagging at the two submission sites. Co-Authored-By: Claude Opus 4.8 --- backend/infrahub/computed_attribute/tasks.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/backend/infrahub/computed_attribute/tasks.py b/backend/infrahub/computed_attribute/tasks.py index 92dbcc570b..98ac5524b3 100644 --- a/backend/infrahub/computed_attribute/tasks.py +++ b/backend/infrahub/computed_attribute/tasks.py @@ -28,6 +28,7 @@ TRIGGER_UPDATE_JINJA_COMPUTED_ATTRIBUTES, TRIGGER_UPDATE_PYTHON_COMPUTED_ATTRIBUTES, ) +from infrahub.workflows.constants import WorkflowTag from infrahub.workflows.utils import add_tags, wait_for_schema_to_converge from .gather import gather_trigger_computed_attribute_jinja2, gather_trigger_computed_attribute_python @@ -211,8 +212,6 @@ async def process_transform( if not transform_attributes: return - dispatcher = await build_bulk_recompute_dispatcher(schema_branch=schema_branch) - for attribute_name, transform_attribute in transform_attributes.items(): if not transform_attribute.transform: raise ValueError(f"No transform configured for computed attribute '{attribute_name}'") @@ -265,6 +264,7 @@ async def process_transform( for skipped_id, reason in skipped: log.warning(f"Skipping recompute of '{attribute_name}' for node {skipped_id}: {reason}") + dispatcher = await build_bulk_recompute_dispatcher(schema_branch=schema_branch) await dispatcher.dispatch( writes=writes, branch_name=branch_name, @@ -307,6 +307,11 @@ async def trigger_update_python_computed_attributes( "computed_attribute_kind": computed_attribute_kind, "context": context, }, + # Bake the branch tag in at submission so it is part of the run's creation + # tags: a later in-flow tag update rebuilds the list from the tags known at + # flow start, so a branch tag added only mid-run would be dropped and the + # run would disappear from branch-filtered task queries. + tags=[WorkflowTag.BRANCH.render(identifier=branch_name)], ) @@ -708,6 +713,9 @@ async def query_transform_targets( "computed_attribute_kind": kind, "context": context, }, + # Same rationale as the setup-path submission: the branch tag must be part + # of the creation tags to survive in-flow tag updates. + tags=[WorkflowTag.BRANCH.render(identifier=branch_name)], ) From 31215208b716af44ce7ab38c336e52e46f7a0fc9 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Fri, 24 Jul 2026 11:06:28 +0300 Subject: [PATCH 06/13] test(computed-attribute): chunked submission coverage and recompute summary log Cover the oversized-fan-out path directly: a fan-out beyond the submission limit splits into bounded submissions, every id submitted exactly once, and each submission carries the branch tag that branch-filtered task queries match on. The workflow recorder now records submission tags so adapters can assert them. The process flow ends with one aggregate line (submitted/written/skipped) so partial failure is greppable from the flow log alone until a proper recompute-failure surface exists. Co-Authored-By: Claude Opus 4.8 --- backend/infrahub/computed_attribute/tasks.py | 14 +++--- backend/tests/adapters/workflow.py | 2 +- ...st_computed_attribute_task_optimization.py | 50 ++++++++++++++++++- 3 files changed, 57 insertions(+), 9 deletions(-) diff --git a/backend/infrahub/computed_attribute/tasks.py b/backend/infrahub/computed_attribute/tasks.py index 98ac5524b3..2aa9e7f9e0 100644 --- a/backend/infrahub/computed_attribute/tasks.py +++ b/backend/infrahub/computed_attribute/tasks.py @@ -272,6 +272,12 @@ async def process_transform( coalesced=False, recompute_depth=0, ) + # One aggregate line per attribute so partial failure is greppable/alertable from the + # flow log alone; until a recompute-failure surface exists, this is the only rollup. + log.info( + f"Recompute of '{attribute_name}' complete: submitted={len(results)} " + f"written={len(writes)} skipped={len(skipped)}" + ) @flow( @@ -307,10 +313,7 @@ async def trigger_update_python_computed_attributes( "computed_attribute_kind": computed_attribute_kind, "context": context, }, - # Bake the branch tag in at submission so it is part of the run's creation - # tags: a later in-flow tag update rebuilds the list from the tags known at - # flow start, so a branch tag added only mid-run would be dropped and the - # run would disappear from branch-filtered task queries. + # Must be a creation tag: in-flow tag updates drop tags added mid-run. tags=[WorkflowTag.BRANCH.render(identifier=branch_name)], ) @@ -713,8 +716,7 @@ async def query_transform_targets( "computed_attribute_kind": kind, "context": context, }, - # Same rationale as the setup-path submission: the branch tag must be part - # of the creation tags to survive in-flow tag updates. + # Must be a creation tag: in-flow tag updates drop tags added mid-run. tags=[WorkflowTag.BRANCH.render(identifier=branch_name)], ) diff --git a/backend/tests/adapters/workflow.py b/backend/tests/adapters/workflow.py index 470d68c04b..8e48abdc47 100644 --- a/backend/tests/adapters/workflow.py +++ b/backend/tests/adapters/workflow.py @@ -54,7 +54,7 @@ async def submit_workflow( tags: list[str] | None = None, priority: WorkflowPriority | None = None, ) -> WorkflowInfo: - self.calls.append({"kind": "submit", "workflow": workflow, "parameters": parameters or {}}) + self.calls.append({"kind": "submit", "workflow": workflow, "parameters": parameters or {}, "tags": tags or []}) return WorkflowInfo(id=uuid.uuid4()) def get_execute_calls_for(self, workflow: WorkflowDefinition) -> list[dict[str, Any]]: diff --git a/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py b/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py index 37d085a322..2a671b9b29 100644 --- a/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py +++ b/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py @@ -7,13 +7,20 @@ from infrahub.auth.session import AccountSession from infrahub.auth.types import AuthType -from infrahub.computed_attribute.tasks import trigger_update_jinja2_computed_attributes +from infrahub.computed_attribute.tasks import ( + trigger_update_jinja2_computed_attributes, + trigger_update_python_computed_attributes, +) from infrahub.context import BranchContext, InfrahubContext from infrahub.core.constants import InfrahubKind from infrahub.core.manager import NodeManager from infrahub.core.node import Node from infrahub.workers.dependencies import build_workflow -from infrahub.workflows.catalogue import COMPUTED_ATTRIBUTE_PROCESS_JINJA2 +from infrahub.workflows.catalogue import ( + COMPUTED_ATTRIBUTE_PROCESS_JINJA2, + COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM, +) +from infrahub.workflows.constants import WorkflowTag from tests.adapters.workflow import WorkflowRecorder from tests.helpers.test_app import TestInfrahubApp @@ -80,3 +87,42 @@ async def test_trigger_update_jinja2_computed_attributes_submits_all_node_ids( call["parameters"]["object_id"] for call in recorder.get_submit_calls_for(COMPUTED_ATTRIBUTE_PROCESS_JINJA2) } assert submitted_ids == set(tags_dataset) + + async def test_trigger_update_python_computed_attributes_chunks_and_tags_submissions( + self, + db: InfrahubDatabase, + tags_dataset: list[str], + default_branch: Branch, + client: InfrahubClient, + context: EventContext, + prefect_test_fixture: None, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """An oversized fan-out splits into bounded submissions, each carrying the branch tag. + + The branch tag must be part of every submission: it is what branch-filtered task + queries match on, and only tags present at run creation reliably survive later + in-flow tag updates. + """ + # Shrink the submission limit (derived from this environment variable) so three + # nodes already exceed one chunk: limit 4 -> chunk size 2 -> splits [2, 1]. + monkeypatch.setenv("PREFECT_SERVER_EVENTS_MAXIMUM_RELATED_RESOURCES", "4") + + recorder = WorkflowRecorder() + with dependency_provider.scope(build_workflow, lambda: recorder): + await trigger_update_python_computed_attributes( + branch_name=default_branch.name, + computed_attribute_name="test-attribute", + computed_attribute_kind=InfrahubKind.TAG, + context=context, + ) + + submissions = recorder.get_submit_calls_for(COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM) + assert [len(call["parameters"]["object_ids"]) for call in submissions] == [2, 1] + + # Every node is submitted exactly once across the chunks. + submitted_ids = [oid for call in submissions for oid in call["parameters"]["object_ids"]] + assert sorted(submitted_ids) == sorted(tags_dataset) + + branch_tag = WorkflowTag.BRANCH.render(identifier=default_branch.name) + assert all(branch_tag in call["tags"] for call in submissions) From 81b82b714fe1ce83df12420c4d9c25b6ffdd3e1f Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Fri, 24 Jul 2026 11:09:08 +0300 Subject: [PATCH 07/13] docs(knowledge): document the batch execution path for Python computed attributes Co-Authored-By: Claude Opus 4.8 --- dev/knowledge/backend/computed-attributes.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/dev/knowledge/backend/computed-attributes.md b/dev/knowledge/backend/computed-attributes.md index 10071cf8cf..d8b42d6825 100644 --- a/dev/knowledge/backend/computed-attributes.md +++ b/dev/knowledge/backend/computed-attributes.md @@ -121,6 +121,17 @@ All three run `process_transform_lifecycle`. On create or update it waits for th Besides the transform-lifecycle triggers, each `(kind, attribute)` has a data-path automation that recomputes the value when a node feeding the transform's query changes. `_reconcile_python_computed_attribute_automations` rebuilds these from the schema. One gather builds both trigger lists and they are applied under a single trigger-registry lock, so a concurrent reconcile cannot delete an automation another run just created, and a transform delete prunes its automation rather than leaving it stale. +### Batch Execution + +`process_transform` processes its node ids as one batch per attribute, not one task per node: + +- The transform's git repository is initialized once for the whole batch and shared across the per-node executions. Transform execution must not mutate the shared checkout. +- Each node's read still runs individually with `update_group=True`, keeping the node subscribed to the transform's query group (the reverse index that routes future source changes to affected readers). +- The recomputed values persist through the shared bulk recompute writer (bounded transactions), not via per-node GraphQL mutations. The writer's skip-unchanged gating means a value identical to the stored one emits no event and dispatches no follow-on recompute, which is what keeps a wide fan-out from echoing into further waves. +- A node whose transform raises or returns a non-string is skipped with its previous value intact and a logged reason; the rest of the batch persists. The flow ends with a `submitted/written/skipped` summary line. +- Each submission carries the branch tag at creation so the flow run stays visible in branch-filtered task queries; tags added mid-run do not survive later in-flow tag updates. +- Crash semantics: the writer commits in bounded chunks, so a mid-batch crash leaves earlier chunks persisted. Recovery is re-running the recompute; skip-unchanged makes redone work no-op-cheap. Rollback of the whole feature is a clean revert (no schema or data migration). + ### Invariants - **Over-recompute is acceptable, under-recompute is not.** Any fallback or error path recomputes rather than risk a stale value. From e27e322c8337421cdca8bc5c7156a8cb1a0810f7 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Fri, 24 Jul 2026 11:12:17 +0300 Subject: [PATCH 08/13] docs(knowledge): batch path in merge-recompute, tag timing semantics in async-tasks Co-Authored-By: Claude Opus 4.8 --- dev/knowledge/backend/async-tasks.md | 2 ++ dev/knowledge/backend/merge-recompute.md | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/dev/knowledge/backend/async-tasks.md b/dev/knowledge/backend/async-tasks.md index a62be58d46..bd32f518b9 100644 --- a/dev/knowledge/backend/async-tasks.md +++ b/dev/knowledge/backend/async-tasks.md @@ -151,6 +151,8 @@ Workflows receive metadata tags for organization and filtering: | Workflow Type | `infrahub.app/workflow-type/{type}` | Categorize by type | | Database Change | `infrahub.app/database-change` | Flag database-modifying workflows | +Tags come from two moments, and the difference matters: tags present at run creation (the deployment's static tags plus any `tags=` passed to `submit_workflow`) survive for the run's lifetime, while tags added mid-run via `add_tags` are rebuilt from the tags known at flow start, so a later in-flow tag update drops anything another in-flow update added before it. A tag that filtering depends on (the branch tag for branch-filtered task queries, for example) must therefore be passed at submission, not added from inside the flow. + ## Execution Flow 1. **Registration**: Workflows defined in `catalogue.py` are registered on startup diff --git a/dev/knowledge/backend/merge-recompute.md b/dev/knowledge/backend/merge-recompute.md index aa96eb755a..657b3264f3 100644 --- a/dev/knowledge/backend/merge-recompute.md +++ b/dev/knowledge/backend/merge-recompute.md @@ -4,7 +4,7 @@ A live edit recomputes derived values one node at a time (see [computed-attributes.md](computed-attributes.md)). A merge or rebase can change many nodes at once, so it uses a different path: one coalesced recompute for the whole change set, written in bulk, then chained to any value that reads what was written. -This covers three derived-value families: Jinja2 computed attributes, display labels, and human-friendly ids. Python-transform computed attributes and profile refresh stay on the per-node path and are not part of this. Generator and artifact regeneration on merge takes its own selective path, described in [selective-merge-regeneration.md](selective-merge-regeneration.md). +This covers three derived-value families: Jinja2 computed attributes, display labels, and human-friendly ids. Python-transform computed attributes and profile refresh are not part of this coalesced pass; they are dispatched by their own automations, though the Python transforms process their fan-out as batches persisted through the same bulk writer (see [computed-attributes.md](computed-attributes.md)). Generator and artifact regeneration on merge takes its own selective path, described in [selective-merge-regeneration.md](selective-merge-regeneration.md). ## Why a separate path From 624d63c3c29bee42d6b3a08c7c34973f9e56b007 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Fri, 24 Jul 2026 11:24:44 +0300 Subject: [PATCH 09/13] style(computed-attribute): tighten added comments and docstrings to the why Co-Authored-By: Claude Opus 4.8 --- backend/infrahub/computed_attribute/tasks.py | 38 +++++++------------ ...st_computed_attribute_task_optimization.py | 9 ++--- 2 files changed, 16 insertions(+), 31 deletions(-) diff --git a/backend/infrahub/computed_attribute/tasks.py b/backend/infrahub/computed_attribute/tasks.py index 2aa9e7f9e0..5a54934929 100644 --- a/backend/infrahub/computed_attribute/tasks.py +++ b/backend/infrahub/computed_attribute/tasks.py @@ -113,12 +113,11 @@ async def _transform_value_for_node( context: EventContext, repo: InfrahubReadOnlyRepository | InfrahubRepository, ) -> AttributeValueWrite: - """Run the transform for one node against an already-initialized repository. + """Compute one node's value against a shared, pre-initialized repository. - The query is run with ``update_group`` so the node is (re)registered as a subscriber of the - transform's query group, which is how a later change to any node the query reads triggers this - node's recompute. The recomputed value is returned to be persisted in bulk with the rest of the - batch rather than written here. + ``update_group`` keeps the node subscribed to the transform's query group, which is + what routes future source changes back to this node. ``context`` is reapplied here + because ``get_client()`` builds a fresh client per call rather than inheriting one. """ client = get_client() client.request_context = context.to_request_context() @@ -146,13 +145,10 @@ async def _transform_value_for_node( def _partition_transform_results( results: list[tuple[str, AttributeValueWrite | Exception]], ) -> tuple[list[AttributeValueWrite], list[tuple[str, str]]]: - """Split per-node transform results into values to persist and nodes to skip. + """Split results into values to persist and ``(node_id, reason)`` skips. - A node is skipped, leaving its previous value in place, when its transform raised or produced a - non-string value. Isolating failures this way keeps one failing node from blocking the rest of - the batch, and preserves the contract that only a real string is persisted rather than silently - overwriting a value with null. Returns the writes to persist and ``(node_id, reason)`` pairs - describing each skip. + A raised or non-string result is skipped so one bad node cannot block its siblings, + and a failure never overwrites the last good value with null. """ writes: list[AttributeValueWrite] = [] skipped: list[tuple[str, str]] = [] @@ -180,15 +176,11 @@ async def process_transform( object_ids: list[str] | None = None, updated_fields: list[str] | None = None, # noqa: ARG001 ) -> None: - """Recompute one or more Python computed attributes for a batch of nodes. + """Recompute Python computed attributes for a batch of nodes. - The transform's git repository is initialized once for the whole batch instead of once per node, - each node's value is computed concurrently, and the results are persisted in a single bulk write - instead of a GraphQL mutation per node. The bulk write skips any node whose recomputed value is - unchanged, so an unchanged node neither writes nor fans out a further recompute. - - A node whose transform raises or returns a non-string value is skipped with its previous value - left in place, so one failing node cannot block the rest of the batch from being persisted. + One repository init and one bulk write per batch; unchanged values emit no events + and fan out no further recompute; a failing node keeps its previous value without + blocking its siblings. Raises: ValueError: if a computed attribute has no transform configured or the transform cannot be fetched. @@ -228,8 +220,6 @@ async def process_transform( f"Unable to fetch transform '{transform_attribute.transform}' for computed attribute '{attribute_name}'" ) - # Initialize the repository once and share it across the whole batch rather than paying a - # repository init per node. repo = await get_initialized_repo( client=client, repository_id=transform.repository_id, @@ -238,8 +228,7 @@ async def process_transform( commit=transform.repository_commit, ) - # return_exceptions keeps one node's failed read or transform from aborting the batch and - # dropping the healthy nodes' writes; node=oid tags each result with the node it belongs to. + # One failing node must not abort the batch and drop its siblings' writes. batch = await client.create_batch(return_exceptions=True) for oid in all_ids: batch.add( @@ -272,8 +261,7 @@ async def process_transform( coalesced=False, recompute_depth=0, ) - # One aggregate line per attribute so partial failure is greppable/alertable from the - # flow log alone; until a recompute-failure surface exists, this is the only rollup. + # The only rollup of partial failure until a recompute-failure surface exists. log.info( f"Recompute of '{attribute_name}' complete: submitted={len(results)} " f"written={len(writes)} skipped={len(skipped)}" diff --git a/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py b/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py index 2a671b9b29..dd7dc8b861 100644 --- a/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py +++ b/backend/tests/functional/computed_attributes/test_computed_attribute_task_optimization.py @@ -100,12 +100,10 @@ async def test_trigger_update_python_computed_attributes_chunks_and_tags_submiss ) -> None: """An oversized fan-out splits into bounded submissions, each carrying the branch tag. - The branch tag must be part of every submission: it is what branch-filtered task - queries match on, and only tags present at run creation reliably survive later - in-flow tag updates. + Branch-filtered task queries match on that tag, and only creation tags reliably + survive in-flow tag updates. """ - # Shrink the submission limit (derived from this environment variable) so three - # nodes already exceed one chunk: limit 4 -> chunk size 2 -> splits [2, 1]. + # Limit 4 -> chunk size 2, so three nodes already split into [2, 1]. monkeypatch.setenv("PREFECT_SERVER_EVENTS_MAXIMUM_RELATED_RESOURCES", "4") recorder = WorkflowRecorder() @@ -120,7 +118,6 @@ async def test_trigger_update_python_computed_attributes_chunks_and_tags_submiss submissions = recorder.get_submit_calls_for(COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM) assert [len(call["parameters"]["object_ids"]) for call in submissions] == [2, 1] - # Every node is submitted exactly once across the chunks. submitted_ids = [oid for call in submissions for oid in call["parameters"]["object_ids"]] assert sorted(submitted_ids) == sorted(tags_dataset) From be15fcdf753fdf3ffb860b6446ef03fcad149846 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Fri, 24 Jul 2026 15:08:02 +0300 Subject: [PATCH 10/13] fix(computed-attribute): review fixes for tags, doc precision, shared fixtures - carry the branch tag at creation on lifecycle-path submissions too - qualify the skip-unchanged doc claim to per-node effective changes - dedupe the transform dataset setup into a shared component helper - drop a review-facing comment on the summary log line Co-Authored-By: Claude Fable 5 --- backend/infrahub/computed_attribute/tasks.py | 1 - .../computed_attribute/transform_recompute.py | 3 ++ .../component/computed_attribute/_base.py | 33 ++++++++++++++++++- .../test_merge_fanout_python.py | 28 ++++------------ .../test_scoped_recompute_python.py | 24 ++------------ .../test_transform_recompute.py | 2 ++ dev/knowledge/backend/computed-attributes.md | 2 +- 7 files changed, 46 insertions(+), 47 deletions(-) diff --git a/backend/infrahub/computed_attribute/tasks.py b/backend/infrahub/computed_attribute/tasks.py index 5a54934929..687f0cadec 100644 --- a/backend/infrahub/computed_attribute/tasks.py +++ b/backend/infrahub/computed_attribute/tasks.py @@ -261,7 +261,6 @@ async def process_transform( coalesced=False, recompute_depth=0, ) - # The only rollup of partial failure until a recompute-failure surface exists. log.info( f"Recompute of '{attribute_name}' complete: submitted={len(results)} " f"written={len(writes)} skipped={len(skipped)}" diff --git a/backend/infrahub/computed_attribute/transform_recompute.py b/backend/infrahub/computed_attribute/transform_recompute.py index 703bd87070..208fe0c605 100644 --- a/backend/infrahub/computed_attribute/transform_recompute.py +++ b/backend/infrahub/computed_attribute/transform_recompute.py @@ -6,6 +6,7 @@ from infrahub.core.registry import registry from infrahub.log import get_logger from infrahub.workflows.catalogue import TRIGGER_UPDATE_PYTHON_COMPUTED_ATTRIBUTES +from infrahub.workflows.constants import WorkflowTag from .recompute_resolution import RecomputeResolver @@ -66,5 +67,7 @@ async def submit(self, *, branch_name: str, transform_id: str, context: EventCon "computed_attribute_kind": definition.kind, "context": context, }, + # Must be a creation tag: in-flow tag updates drop tags added mid-run. + tags=[WorkflowTag.BRANCH.render(identifier=branch_name)], ) return len(definitions) diff --git a/backend/tests/component/computed_attribute/_base.py b/backend/tests/component/computed_attribute/_base.py index 4d4e582a41..5bc55cfff9 100644 --- a/backend/tests/component/computed_attribute/_base.py +++ b/backend/tests/component/computed_attribute/_base.py @@ -11,7 +11,8 @@ from infrahub.auth.session import AccountSession from infrahub.auth.types import AuthType from infrahub.context import InfrahubContext -from infrahub.core.constants import RelationshipCardinality +from infrahub.core.constants import InfrahubKind, RelationshipCardinality +from infrahub.core.node import Node from infrahub.core.schema import AttributeSchema, NodeSchema, RelationshipSchema, SchemaRoot from infrahub.core.schema.computed_attribute import ComputedAttribute, ComputedAttributeKind from infrahub.events.schema_action import ChangedElementsPayload # noqa: TC001 used in dataclass field @@ -26,6 +27,7 @@ from infrahub.core.branch import Branch from infrahub.core.protocols import CoreAccount + from infrahub.database import InfrahubDatabase from infrahub.events.models import EventContext from infrahub.services import InfrahubServices from infrahub.workflows.models import WorkflowDefinition @@ -83,6 +85,35 @@ ) +async def create_transform01(db: InfrahubDatabase, branch_name: str) -> Node: + """query01/repo01/transform01: a Python transform whose query reads only TestCar.name. + + The edges/node query structure is required for the analyzer to record the field as + a read, making ``name`` the single "related" field. Returns the repository so a + test can attach further transforms to it. + """ + query = await Node.init(db=db, schema=InfrahubKind.GRAPHQLQUERY) + await query.new( + db=db, + name="query01", + query="query { TestCar { edges { node { name { value } } } } }", + models=["TestCar", "TestPerson"], + ) + await query.save(db=db) + + repo = await Node.init(db=db, schema=InfrahubKind.READONLYREPOSITORY) + await repo.new(db=db, name="repo01", ref=branch_name, commit="commit01", location="location01", queries=[query]) + await repo.save(db=db) + + transform = await Node.init(db=db, schema=InfrahubKind.TRANSFORMPYTHON) + await transform.new( + db=db, name="transform01", file_path="transform.py", class_name="Transform", query=query, repository=repo + ) + await transform.save(db=db) + + return repo + + @dataclass class ScopedRecomputeCase: """A single ``(changed_elements -> expected submitted set)`` parametrize case.""" diff --git a/backend/tests/component/computed_attribute/test_merge_fanout_python.py b/backend/tests/component/computed_attribute/test_merge_fanout_python.py index 8f08916c3e..ef6c14a84a 100644 --- a/backend/tests/component/computed_attribute/test_merge_fanout_python.py +++ b/backend/tests/component/computed_attribute/test_merge_fanout_python.py @@ -14,10 +14,13 @@ import pytest from infrahub.computed_attribute.tasks import trigger_update_python_computed_attributes -from infrahub.core.constants import InfrahubKind from infrahub.core.node import Node from infrahub.workflows.catalogue import COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM -from tests.component.computed_attribute._base import CAR_PERSON_PYTHON_SCHEMA, ScopedRecomputeTestBase +from tests.component.computed_attribute._base import ( + CAR_PERSON_PYTHON_SCHEMA, + ScopedRecomputeTestBase, + create_transform01, +) from tests.helpers.schema import load_schema if TYPE_CHECKING: @@ -48,26 +51,7 @@ async def transform_dataset( Returns the ids of every car, so the test can compare the fan-out against the full population of the kind. """ - query = await Node.init(db=db, schema=InfrahubKind.GRAPHQLQUERY) - await query.new( - db=db, - name="query01", - query="query { TestCar { edges { node { name { value } } } } }", - models=["TestCar", "TestPerson"], - ) - await query.save(db=db) - - repo = await Node.init(db=db, schema=InfrahubKind.READONLYREPOSITORY) - await repo.new( - db=db, name="repo01", ref=default_branch.name, commit="commit01", location="location01", queries=[query] - ) - await repo.save(db=db) - - transform = await Node.init(db=db, schema=InfrahubKind.TRANSFORMPYTHON) - await transform.new( - db=db, name="transform01", file_path="transform.py", class_name="Transform", query=query, repository=repo - ) - await transform.save(db=db) + await create_transform01(db=db, branch_name=default_branch.name) await load_schema(db=db, schema=CAR_PERSON_PYTHON_SCHEMA, update_db=True) diff --git a/backend/tests/component/computed_attribute/test_scoped_recompute_python.py b/backend/tests/component/computed_attribute/test_scoped_recompute_python.py index 430d40b6c4..1fc3aabc70 100644 --- a/backend/tests/component/computed_attribute/test_scoped_recompute_python.py +++ b/backend/tests/component/computed_attribute/test_scoped_recompute_python.py @@ -13,6 +13,7 @@ CAR_PERSON_PYTHON_SCHEMA, ScopedRecomputeCase, ScopedRecomputeTestBase, + create_transform01, ) from tests.helpers.schema import load_schema @@ -53,28 +54,7 @@ async def transform_dataset( client: InfrahubClient, admin_account: CoreAccount, ) -> None: - # The transform query reads only TestCar.name, so that is the single "related" field. - # The edges/node structure is required for the analyzer to record the field as a read. - query = await Node.init(db=db, schema=InfrahubKind.GRAPHQLQUERY) - await query.new( - db=db, - name="query01", - query="query { TestCar { edges { node { name { value } } } } }", - models=["TestCar", "TestPerson"], - ) - await query.save(db=db) - - repo = await Node.init(db=db, schema=InfrahubKind.READONLYREPOSITORY) - await repo.new( - db=db, name="repo01", ref=default_branch.name, commit="commit01", location="location01", queries=[query] - ) - await repo.save(db=db) - - transform = await Node.init(db=db, schema=InfrahubKind.TRANSFORMPYTHON) - await transform.new( - db=db, name="transform01", file_path="transform.py", class_name="Transform", query=query, repository=repo - ) - await transform.save(db=db) + repo = await create_transform01(db=db, branch_name=default_branch.name) # A query reading the display label cannot be mapped to precise backing fields, # so its attribute is always recomputed (the conservative, opaque case). diff --git a/backend/tests/unit/computed_attribute/test_transform_recompute.py b/backend/tests/unit/computed_attribute/test_transform_recompute.py index 8bc2b2f735..6fc470a5f8 100644 --- a/backend/tests/unit/computed_attribute/test_transform_recompute.py +++ b/backend/tests/unit/computed_attribute/test_transform_recompute.py @@ -14,6 +14,7 @@ from infrahub.core.schema.schema_branch import SchemaBranch from infrahub.events.models import EventBranchContext, EventContext from infrahub.workflows.catalogue import TRIGGER_UPDATE_PYTHON_COMPUTED_ATTRIBUTES +from infrahub.workflows.constants import WorkflowTag from tests.adapters.workflow import WorkflowRecorder if TYPE_CHECKING: @@ -103,6 +104,7 @@ async def test_submit_fans_out_recompute_for_each_fed_attribute( assert calls[0]["parameters"]["computed_attribute_name"] == "description" assert calls[0]["parameters"]["computed_attribute_kind"] == "TestingCar" assert calls[0]["parameters"]["branch_name"] == BRANCH + assert WorkflowTag.BRANCH.render(identifier=BRANCH) in calls[0]["tags"] async def test_submit_skips_a_missing_transform(schema_branch_with_transform: None) -> None: diff --git a/dev/knowledge/backend/computed-attributes.md b/dev/knowledge/backend/computed-attributes.md index d8b42d6825..eff26f9abd 100644 --- a/dev/knowledge/backend/computed-attributes.md +++ b/dev/knowledge/backend/computed-attributes.md @@ -127,7 +127,7 @@ Besides the transform-lifecycle triggers, each `(kind, attribute)` has a data-pa - The transform's git repository is initialized once for the whole batch and shared across the per-node executions. Transform execution must not mutate the shared checkout. - Each node's read still runs individually with `update_group=True`, keeping the node subscribed to the transform's query group (the reverse index that routes future source changes to affected readers). -- The recomputed values persist through the shared bulk recompute writer (bounded transactions), not via per-node GraphQL mutations. The writer's skip-unchanged gating means a value identical to the stored one emits no event and dispatches no follow-on recompute, which is what keeps a wide fan-out from echoing into further waves. +- The recomputed values persist through the shared bulk recompute writer (bounded transactions), not via per-node GraphQL mutations. The writer's skip-unchanged gating is per node, not per value: a save that produces no effective change emits no event and dispatches no follow-on recompute, which is what keeps a wide fan-out from echoing into further waves. A node whose save changes another of its fields still emits an event. - A node whose transform raises or returns a non-string is skipped with its previous value intact and a logged reason; the rest of the batch persists. The flow ends with a `submitted/written/skipped` summary line. - Each submission carries the branch tag at creation so the flow run stays visible in branch-filtered task queries; tags added mid-run do not survive later in-flow tag updates. - Crash semantics: the writer commits in bounded chunks, so a mid-batch crash leaves earlier chunks persisted. Recovery is re-running the recompute; skip-unchanged makes redone work no-op-cheap. Rollback of the whole feature is a clean revert (no schema or data migration). From de6dff75ca5f37b07343f582958fa3aacdfda5d9 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Fri, 24 Jul 2026 15:08:03 +0300 Subject: [PATCH 11/13] docs(specs): spec artifacts for batch python recompute (004) Co-Authored-By: Claude Fable 5 --- .../alignment-check.md | 23 ++++ .../checklists/requirements.md | 35 +++++ .../contracts/README.md | 34 +++++ .../critiques/critique-20260724-104649.md | 117 +++++++++++++++++ .../004-batch-python-recompute/data-model.md | 32 +++++ dev/specs/004-batch-python-recompute/plan.md | 87 +++++++++++++ .../004-batch-python-recompute/quickstart.md | 31 +++++ .../004-batch-python-recompute/research.md | 52 ++++++++ dev/specs/004-batch-python-recompute/spec.md | 120 ++++++++++++++++++ dev/specs/004-batch-python-recompute/tasks.md | 89 +++++++++++++ 10 files changed, 620 insertions(+) create mode 100644 dev/specs/004-batch-python-recompute/alignment-check.md create mode 100644 dev/specs/004-batch-python-recompute/checklists/requirements.md create mode 100644 dev/specs/004-batch-python-recompute/contracts/README.md create mode 100644 dev/specs/004-batch-python-recompute/critiques/critique-20260724-104649.md create mode 100644 dev/specs/004-batch-python-recompute/data-model.md create mode 100644 dev/specs/004-batch-python-recompute/plan.md create mode 100644 dev/specs/004-batch-python-recompute/quickstart.md create mode 100644 dev/specs/004-batch-python-recompute/research.md create mode 100644 dev/specs/004-batch-python-recompute/spec.md create mode 100644 dev/specs/004-batch-python-recompute/tasks.md diff --git a/dev/specs/004-batch-python-recompute/alignment-check.md b/dev/specs/004-batch-python-recompute/alignment-check.md new file mode 100644 index 0000000000..cc6b797263 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/alignment-check.md @@ -0,0 +1,23 @@ +# Spec/Ask Alignment Check + +## Source + +Inline PRD from the invocation (no URLs): the detailed ask describing the per-node cost structure, the measured echo storm (73k flow runs), the five goals (repo-init once, bulk persistence via the shared writer, skip-unchanged, per-node failure isolation, branch-filtered task visibility), the reuse constraint (Jinja2 bulk-write machinery), the behavior-preservation constraint, and the explicit out-of-scope (fan-out scoping). + +## Verdict + +✅ ALIGNED + +## Findings + +| Severity | Category | PRD reference | Spec reference | Description | +|---|---|---|---|---| +| info | added (necessary clarification) | implicit in "behavior-preserving" | FR-007 | Spec makes the subscriber reverse-index registration an explicit requirement; the ask implies it (removing it would change future recompute routing, violating behavior preservation). Not drift. | +| info | added (necessary clarification) | "existing chunked transactions / existing machinery" | FR-008 | Spec pins oversized fan-out splitting to the existing submission limit; the ask references the existing chunking implicitly. Not drift. | +| info | expansion | "isolate per-node transform failures" | US3 + edge cases | Spec expands isolation into concrete failure classes (raise vs non-text) and preservation semantics. Faithful elaboration. | + +Every goal and constraint in the ask maps to a requirement: repo-init once → FR-001; bulk persistence via shared writer → FR-002 + Assumption 1; skip-unchanged, no events/cascade → FR-003; failure isolation → FR-005; branch-filtered task visibility → FR-006; behavior preservation → FR-004/FR-009; reuse over new write paths → Assumption 1 + plan constraint; fan-out scoping out of scope → Assumption 3. No PRD requirement is missing, softened, or semantically changed; no off-scope additions beyond the two necessary clarifications above. + +## Action + +Proceed — no remediation pass needed. diff --git a/dev/specs/004-batch-python-recompute/checklists/requirements.md b/dev/specs/004-batch-python-recompute/checklists/requirements.md new file mode 100644 index 0000000000..4eaff1c84e --- /dev/null +++ b/dev/specs/004-batch-python-recompute/checklists/requirements.md @@ -0,0 +1,35 @@ +# Specification Quality Checklist: Batch Python Computed-Attribute Recompute + +**Purpose**: Validate specification completeness and quality before proceeding to planning +**Created**: 2026-07-24 +**Feature**: [spec.md](../spec.md) + +## Content Quality + +- [x] No implementation details (languages, frameworks, APIs) +- [x] Focused on user value and business needs +- [x] Written for non-technical stakeholders +- [x] All mandatory sections completed + +## Requirement Completeness + +- [x] No [NEEDS CLARIFICATION] markers remain +- [x] Requirements are testable and unambiguous +- [x] Success criteria are measurable +- [x] Success criteria are technology-agnostic (no implementation details) +- [x] All acceptance scenarios are defined +- [x] Edge cases are identified +- [x] Scope is clearly bounded +- [x] Dependencies and assumptions identified + +## Feature Readiness + +- [x] All functional requirements have clear acceptance criteria +- [x] User scenarios cover primary flows +- [x] Feature meets measurable outcomes defined in Success Criteria +- [x] No implementation details leak into specification + +## Notes + +- FR-002/FR-007 and the Assumptions name existing internal mechanisms ("shared bulk recompute write path", "subscriber reverse-index") — retained deliberately: the ask explicitly constrains the solution to reuse them, so they are requirements of the ask, not leaked design choices. +- No [NEEDS CLARIFICATION] markers were required: the ask is unusually precise (it is a retrospective spec of a well-understood change) — scope, constraints, and out-of-scope are all stated in the input. diff --git a/dev/specs/004-batch-python-recompute/contracts/README.md b/dev/specs/004-batch-python-recompute/contracts/README.md new file mode 100644 index 0000000000..40f1659236 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/contracts/README.md @@ -0,0 +1,34 @@ +# Contracts: Batch Python Computed-Attribute Recompute + +## External API — unchanged + +- GraphQL mutation `InfrahubUpdateComputedAttribute`: remains published and functional (public API). This feature only stops calling it internally. +- No REST/GraphQL schema changes; no generated-file changes expected beyond none. + +## Internal flow contract — `computed_attribute_process_transform` (stable) + +Parameters (unchanged, Prefect deployment contract): + +``` +branch_name: str +node_kind: str +object_ids: list[str] # ≤ get_submission_chunk_size() ids +computed_attribute_name: str +computed_attribute_kind: str +context: EventContext +``` + +Behavioral contract (changed internals, preserved observables): + +| Observable | Before | After | +|---|---|---| +| Final attribute values | v | v (identical) | +| NodeUpdated event per really-changed node | yes (from mutation) | yes (from bulk writer, live origin) | +| Events / cascade for unchanged values | yes → echo | none | +| Client-visible mutations issued | N | 0 | +| Flow run visible under branch tag filter | yes | yes (tag at submission) | +| One node's failure | fails its own task | skipped + logged; siblings persist | + +## Event contract + +`NodeUpdatedEvent` granularity, payload, and origin (live) for really-changed nodes are identical to a direct attribute update — downstream consumers (webhooks, dependent computed attributes, UI) require no changes. diff --git a/dev/specs/004-batch-python-recompute/critiques/critique-20260724-104649.md b/dev/specs/004-batch-python-recompute/critiques/critique-20260724-104649.md new file mode 100644 index 0000000000..7ec83d4107 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/critiques/critique-20260724-104649.md @@ -0,0 +1,117 @@ +# Critique Report: Batch Python Computed-Attribute Recompute + +**Date**: 2026-07-24 +**Feature**: [spec.md](../spec.md) +**Plan**: [plan.md](../plan.md) +**Verdict**: ⚠️ PROCEED WITH UPDATES + +--- + +## Executive Summary + +The problem is unusually well-evidenced (a named customer pain, a quantified 73k-flow-run collapse on a production-scale dataset) and the plan's core bet — reuse the proven bulk recompute writer instead of building a second write path — is the right one for both risk and maintenance. The spec's behavior-preservation contract (identical values, identical events for real changes) is the correct definition of "safe." Two gaps block a clean PROCEED, both test-coverage holes on the exact properties this change is most likely to silently regress: task-list visibility of batch runs (FR-006) and multi-batch splitting on the Python submission path (FR-008) have no automated test anywhere in the plan's strategy. Both are cheap to close. Beyond that, the main risks are operational, not architectural: batching deletes the per-node progress surface operators had, and mid-batch crash semantics deserve one explicit paragraph. + +--- + +## Product Lens Findings 🎯 + +### Problem Validation +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| P1 | ✅ (strength) | Problem evidence is exemplary: named customer complaint, reproduced and quantified at scale (73k runs, resource exhaustion). Rare for a perf feature. | Keep the measurement artifacts linked from the PR. | + +### User Value Assessment +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| P2 | 💡 | SC-001's "seconds-to-low-minutes" hedge weakens an otherwise crisp criterion set. | Anchor to the reference scenarios only (x-small: <10s; large: settles without API unavailability) and drop the vague phrase. | +| P3 | 🤔 | Skipped-node discoverability: US3 logs failures, and the spec defers the recovery surface to a follow-up. Until that ships, an operator has no aggregate signal that k of N nodes were skipped. | Accept the deferral but require the flow to end with one summary log line (`skipped=k of n`) so the gap is at least greppable and alertable. | + +### Alternative Approaches +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| P4 | ✅ | Alternatives are honestly dispatched in research.md (keep mutations; raw-Cypher writer). The do-nothing cost is the customer pain itself. | None. | + +### Edge Cases & UX +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| P5 | 💡 | No rollback/kill-switch: the change replaces the write path outright. Behavior-preservation is asserted by tests, not togglable at runtime. | Explicitly record the rollback strategy as "revert the commit" (single-file production change, no schema/data migration → revert is clean). A config flag is not worth its permanent complexity here; say so in the plan rather than leaving it implicit. | + +### Success Measurement +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| P6 | ✅ | SC-002/003/005 are directly verifiable from the perf harness and tests. | None. | + +--- + +## Engineering Lens Findings 🔧 + +### Architecture Soundness +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E1 | ✅ (strength) | Reusing `BulkRecomputeDispatcher`/`Writer` keeps one write-semantics implementation (versioned writes, derived-value cascades, changelog, event gating). The rejected raw-Cypher fork was the classic trap. | None. | +| E2 | 🤔 | One `repo` object is shared across concurrent per-node callables. The transform executes against a shared checkout; concurrent safety rests on the execution being read-only w.r.t. the worktree. | Verify and state this invariant in the code (the SDK batch's bounded concurrency also limits exposure); if the repo layer is not concurrency-safe, serialize execs while keeping reads concurrent. | + +### Failure Mode Analysis +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E3 | 💡 | Mid-batch worker crash: the writer commits per 100-node chunk, so a crash leaves earlier chunks persisted and later ones stale, with no automatic retry defined for the flow. | Add one plan paragraph: recovery = the next trigger (or manual `RecomputeComputedAttribute`) re-runs the batch; skip-unchanged makes redone work no-op-cheap. This is acceptable — but it must be a stated decision, not an accident. | +| E4 | ✅ | Per-node isolation (partition on exceptions/non-str) directly addresses the batch-abort failure mode batching introduces. | None. | + +### Security & Privacy +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E5 | ✅ | No new input surfaces; non-str transform output now validated before write (stricter than blind persistence). Transform sandboxing posture unchanged. | None. | + +### Performance & Scalability +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E6 | ✅ | Bounds are layered and inherited: ~250-id submission chunks, semaphore-5 concurrency, 100-node transactions. Memory per run is O(chunk). | None. | +| E7 | 💡 | Per-changed-node event emission remains O(N); the echo is gone but event volume itself is untouched. At 10x growth the event pipeline becomes the next candidate bottleneck. | Note as a known scaling boundary in the plan; out of scope to change. | + +### Testing Strategy +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E8 | 🎯 | FR-006 (branch-filtered task visibility) has **no test** in the plan. This is precisely the property the batching rewrite can silently break (visibility depends on tag mechanics that in-flow updates can clobber), and its failure mode is an operator-facing blind spot plus hung task-polling consumers. | Add a functional/component assertion: trigger a recompute on a branch, assert the process run appears in a branch-filtered task query. Make it a first-class task. | +| E9 | 🎯 | FR-008 (oversized fan-out splits into multiple batches) is only tested on the neighboring coalesced path; the Python submit loops have **no direct test** that >chunk-limit ids produce multiple submissions, each carrying correct parameters (and the same visibility tags). | Add a unit test with a recorder workflow adapter: N > chunk-limit ids → ⌈N/limit⌉ submissions, ids partitioned exactly once, branch tag on each. | +| E10 | ✅ | Unit (partition), component (fan-out characterization), functional (real repo e2e), perf A/B — the rest of the pyramid is sound. | None. | + +### Operational Readiness +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E11 | 💡 | Batching removes the per-node Prefect task view operators previously had (per-node timing/status). Flow-level visibility (FR-006) is necessary but coarser. | Accept for this feature; record the follow-up: phase-level spans/counters (repo-init / read / exec / write; recomputed/skipped/unchanged counts) belongs with the deferred recovery-surface work. | + +### Dependencies & Integration +| ID | Severity | Finding | Suggestion | +|----|----------|---------|------------| +| E12 | ✅ | No new dependencies; public mutation retained for external users; flow parameter contract unchanged (no deployment migration). | None. | + +--- + +## Cross-Lens Insights + +| ID | Lens | Severity | Finding | Suggestion | +|----|------|----------|---------|------------| +| X1 | Both | 💡 | Silent-skip window (P3 × E11): until the recovery surface ships, skipped nodes are visible only in logs — a product gap and an ops gap with the same root. | The summary log line (P3) + the follow-up scope note (E11) are the same small addition; do both now, build the surface later. | + +--- + +## Findings Summary Table + +| ID | Lens | Severity | Category | Finding | Suggestion | +|----|------|----------|----------|---------|------------| +| E8 | Engineering | 🎯 | Testing | FR-006 task-visibility untested | Branch-filtered task query assertion (functional tier) | +| E9 | Engineering | 🎯 | Testing | FR-008 Python-path chunk splitting untested | Recorder-based unit test: multi-submission split + tags | +| P2 | Product | 💡 | Success criteria | SC-001 hedge phrase | Anchor to reference scenarios | +| P3 | Product | 🤔 | UX gap | Skipped nodes only in logs until follow-up | Require `skipped=k of n` summary log line | +| P5 | Product | 💡 | Rollback | No stated rollback strategy | State "clean revert, no migration" in plan | +| E2 | Engineering | 🤔 | Concurrency | Shared repo across concurrent execs | Verify/state worktree read-only invariant | +| E3 | Engineering | 💡 | Failure modes | Mid-batch crash semantics implicit | State partial-persistence + cheap-redo recovery | +| E7 | Engineering | 💡 | Scalability | Event volume still O(N) | Note as known boundary | +| E11 | Engineering | 💡 | Observability | Per-node task view lost | Scope phase spans/counters into follow-up | +| X1 | Both | 💡 | Ops × UX | Silent-skip window | Summary log line now, surface later | + +--- + +## Verdict + +⚠️ **PROCEED WITH UPDATES** — two 🎯 test-coverage gaps (E8, E9) must enter the plan/tasks before implementation; all remaining items are recommendations or stakeholder questions resolvable inline. No architecture or product rethink warranted. diff --git a/dev/specs/004-batch-python-recompute/data-model.md b/dev/specs/004-batch-python-recompute/data-model.md new file mode 100644 index 0000000000..d42ea22108 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/data-model.md @@ -0,0 +1,32 @@ +# Data Model: Batch Python Computed-Attribute Recompute + +No database schema changes. The entities below are in-memory processing structures. + +## AttributeValueWrite (existing, reused) + +One recomputed value destined for persistence. + +| Field | Type | Notes | +|---|---|---| +| node_id | str | target node | +| field | str | computed attribute name | +| value | str \| None | recomputed value; only `str` reaches the writer on this path | + +Source: `backend/infrahub/core/recompute/bulk_write.py`. Validation: this feature only enqueues writes whose `value` is `str` (R4). + +## Transform batch result (new, transient) + +`list[tuple[str, AttributeValueWrite | Exception]]` — node id paired with its outcome from the concurrent batch. + +State transitions: collected → partitioned into `writes: list[AttributeValueWrite]` + `skipped: list[(node_id, reason)]` → writes dispatched, skips logged. + +## Skipped node record (new, log-only) + +`(node_id, reason)` where reason ∈ {"transform raised …", "transform returned , expected a string"}. Surfaced as flow-run warning logs; no persistence (recovery surface is an explicit non-goal / follow-up). + +## Invariants + +- A node appears at most once per batch (ids deduplicated upstream at trigger time). +- A skipped node's stored value is untouched by the batch. +- `writes` ⊎ `skipped` = batch input (every node accounted for). +- Persisting a value equal to the stored one produces no changelog entry → no event → no cascade (writer invariant, relied upon). diff --git a/dev/specs/004-batch-python-recompute/plan.md b/dev/specs/004-batch-python-recompute/plan.md new file mode 100644 index 0000000000..8c8e8fc9d2 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/plan.md @@ -0,0 +1,87 @@ +# Implementation Plan: Batch Python Computed-Attribute Recompute + +**Branch**: `batch-python-recompute-infp-608` | **Date**: 2026-07-24 | **Spec**: [spec.md](./spec.md) + +**Input**: Feature specification from `specs/004-batch-python-recompute/spec.md` + +## Summary + +Replace the per-node Python computed-attribute recompute path (one Prefect task per node: repository init + read + transform + one `InfrahubUpdateComputedAttribute` GraphQL mutation each) with a batched pass inside the existing `computed_attribute_process_transform` flow: initialize the transform repository once per batch, run the per-node read+transform concurrently with per-node failure isolation, and persist all values through the existing shared bulk recompute writer (`BulkRecomputeDispatcher` → `BulkRecomputeWriter`), whose skip-unchanged gating stops no-op writes from emitting events and re-triggering the recompute machinery (the "echo storm"). Process flow runs must stay visible in branch-filtered task queries by carrying the branch tag from creation. + +## Technical Context + +**Language/Version**: Python 3.14 (backend) + +**Primary Dependencies**: Prefect 3 (flows/automations via task-manager), infrahub_sdk (`InfrahubClient`, `InfrahubBatch`), Neo4j 2026.05 via internal `InfrahubDatabase`, existing `infrahub.core.recompute` package (bulk writer/dispatcher from the Jinja2 work) + +**Storage**: Neo4j graph — writes go through `Node.save` inside the bulk writer's bounded transactions (100 nodes/txn default); no schema or migration changes + +**Testing**: pytest — unit (`backend/tests/unit/computed_attribute/`), component (`backend/tests/component/`), functional with real git repo + transforms (`backend/tests/functional/computed_attributes/`), perf A/B via opsmill/infrahub-private-tests `TestMergeRecomputePython`. Two mandatory additions from critique: (E8) a functional/component assertion that a recompute's process run appears in a branch-filtered task query — visibility depends on tag mechanics that in-flow tag updates can clobber; (E9) a unit test with a recorder workflow adapter asserting an oversized fan-out (> submission limit) splits into ⌈N/limit⌉ submissions, ids partitioned exactly once, branch tag present on every submission + +**Target Platform**: Infrahub task-workers (Linux containers), dev/CI/testcontainers stacks + +**Project Type**: backend subsystem change (single package: `backend/infrahub/computed_attribute/`, touching `backend/infrahub/core/recompute/` consumers only via existing public builder) + +**Performance Goals**: one source change affecting N readers settles in O(N) work — repo init 1×/batch, N reads + N transform execs (unavoidable, user code), ⌈N/100⌉ write transactions, 0 client-visible mutations, 0 echo re-dispatches; ≥99% reduction in flow-run count at scale + +**Constraints**: behavior-preserving (byte-identical final values; identical per-node NodeUpdated events for real changes); merge fan-out scoping untouched; `InfrahubUpdateComputedAttribute` mutation remains public API (unused internally afterwards); per-submission chunking limit (`get_submission_chunk_size()`) unchanged + +**Scale/Scope**: reference datasets — x-small (12 devices/type) for correctness+ratio, large (~100k nodes) where develop currently collapses (73k flow runs, bolt/FD exhaustion) + +## Constitution Check + +*GATE: Must pass before Phase 0 research. Re-check after Phase 1 design.* + +- **I. Schema-Driven Integrity** — PASS: no schema changes; read-only computed attribute semantics preserved. +- **II. Branch-Safe by Default** — PASS: batch persists via `registry.get_branch(branch_name)` through the existing writer; branch-deleted-mid-flight handled (abandon batch, no error). Branch tag on flow runs preserves per-branch task visibility (FR-006). +- **III. Type Safety & Explicit Contracts** — PASS: new helpers fully typed; flow signature (parameters) unchanged so Prefect deployment contract is stable. +- **IV. Test Discipline** — PASS: unit tests for partition/skip logic; component test characterizing fan-out; functional end-to-end with real repo; perf A/B for the headline claim. New behavior (failure isolation) gets dedicated tests. +- **V. Query Performance & Efficiency** — PASS (this feature is the principle in action): eliminates N+1 write pattern (N mutations → chunked bulk writes); reads stay per-node by necessity (per-node query with `update_group` subscriber registration is the reverse-index contract, FR-007). +- **VI. Security & Input Boundaries** — PASS: transform output validated (non-str rejected → skip) instead of being written blindly; no new input surfaces. +- **VII. Simplicity & Maintainability** — PASS: net code shrinks (per-node task + mutation constant deleted); reuses the existing writer rather than adding a second write path. + +No violations → Complexity Tracking not needed. + +## Project Structure + +### Documentation (this feature) + +```text +specs/004-batch-python-recompute/ +├── plan.md # This file +├── research.md # Phase 0 output +├── data-model.md # Phase 1 output +├── quickstart.md # Phase 1 output +├── contracts/ # Phase 1 output +└── tasks.md # Phase 2 output (/speckit-tasks) +``` + +### Source Code (repository root) + +```text +backend/infrahub/computed_attribute/ +└── tasks.py # process_transform flow: batch rewrite; delete per-node task + UPDATE_ATTRIBUTE constant + +backend/infrahub/core/recompute/ # REUSED, not modified +├── bulk_write.py # AttributeValueWrite, BulkRecomputeWriter (chunked txns, has_changes gating) +└── dispatch.py # build_bulk_recompute_dispatcher, BulkRecomputeDispatcher.dispatch + +backend/tests/ +├── unit/computed_attribute/test_tasks.py # partition/skip unit tests +├── component/computed_attribute/ # fan-out characterization +└── functional/computed_attributes/ # end-to-end with real repo (existing, must stay green) +``` + +**Structure Decision**: single-file production change (`computed_attribute/tasks.py`) consuming the existing recompute package via its public builder; tests across the three existing tiers. No new modules. + +## Failure & Rollback Semantics (from critique) + +- **Mid-batch crash (E3)**: the writer commits per 100-node chunk; a crash leaves earlier chunks persisted, later ones stale. Accepted: recovery is the next trigger or a manual `RecomputeComputedAttribute` re-run — skip-unchanged makes redone work no-op-cheap. No flow-level retry is added. +- **Rollback (P5)**: single-file production change, no schema or data migration — rollback is a clean revert of the commit. A runtime kill-switch is deliberately not added (permanent config complexity for a transient de-risking need). +- **Skipped-node summary (P3/X1)**: the flow ends with one summary log line (`recompute complete: written=…, unchanged=…, skipped=…`) so partial failure is greppable/alertable until the recovery surface follow-up ships. +- **Known boundary (E7)**: per-changed-node event volume remains O(N); the echo is eliminated, event emission itself is unchanged and out of scope. +- **Shared-checkout invariant (E2)**: transform execution must not mutate the shared worktree; the per-node callables share one initialized repo under bounded concurrency — state the invariant in code. + +## Complexity Tracking + +Not applicable — no constitution violations. diff --git a/dev/specs/004-batch-python-recompute/quickstart.md b/dev/specs/004-batch-python-recompute/quickstart.md new file mode 100644 index 0000000000..42fafab1f9 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/quickstart.md @@ -0,0 +1,31 @@ +# Quickstart Validation: Batch Python Computed-Attribute Recompute + +## Prerequisites + +- `uv sync --all-groups`; Docker running (component/functional tiers use testcontainers — export the Docker socket if needed). + +## Fast signal (unit, seconds) + +```bash +uv run pytest backend/tests/unit/computed_attribute/ -q +``` + +Expected: partition/skip logic green — string persisted; None/non-str skipped with reason; exception isolated; empty batch no-op. + +## Behavior (functional, minutes) + +```bash +uv run pytest backend/tests/functional/computed_attributes/ -q +``` + +Expected: end-to-end with a real git repo + transform — values recompute on source change; unchanged recompute emits no follow-on work; one failing node leaves siblings updated. + +## Live smoke (dev stack) + +1. `uv run invoke dev.build dev.start` +2. Load a schema with a TransformPython computed attribute + repository (tshirt fixture), create a few readers. +3. Update the source attribute; verify: readers' values refresh; task list filtered by branch shows the `computed_attribute_process_transform` run; re-updating with the same value produces no new recompute wave. + +## At-scale A/B (perf, hours — optional) + +opsmill/infrahub-private-tests `test-dataset.yml` with `test_filter=TestMergeRecomputePython`, same backup/profile, `infrahub_ref` = baseline vs feature ref. Compare: post-merge settle window, flow-run counts (query/process), db_queries, correctness (recomputed == affected). diff --git a/dev/specs/004-batch-python-recompute/research.md b/dev/specs/004-batch-python-recompute/research.md new file mode 100644 index 0000000000..7a016f5e3d --- /dev/null +++ b/dev/specs/004-batch-python-recompute/research.md @@ -0,0 +1,52 @@ +# Research: Batch Python Computed-Attribute Recompute + +## R1 — Persistence path for batched values + +**Decision**: Reuse `build_bulk_recompute_dispatcher()` → `BulkRecomputeDispatcher.dispatch(writes, coalesced=False)` → `BulkRecomputeWriter.write` (chunked `Node.save`, 100/transaction). + +**Rationale**: The writer already provides everything FR-002/003/004 demand: bounded transactions, `node_changelog.has_changes` gating (no event, no cascade for no-ops), per-changed-node NodeUpdated events with live origin so genuinely-changed values still chain dependent recomputes. `coalesced=False` keeps live-event semantics (origin=LIVE, no coalesced chain), matching today's behavior for real changes. + +**Alternatives considered**: +- *Keep per-node `InfrahubUpdateComputedAttribute` mutations* — rejected: the mutation wrapper (parse/validate/permissions per node) and its per-node event emission are the echo storm's fuel. +- *New raw-Cypher UNWIND bulk writer* — rejected: would fork branch/time-versioned write semantics, HFID/display-label cascades, and changelog computation into a second implementation; high correctness risk; violates reuse constraint. + +## R2 — Per-node read + transform execution within the batch + +**Decision**: SDK `InfrahubBatch` (`client.create_batch(return_exceptions=True)`), one plain-async callable per node returning an `AttributeValueWrite`; results partitioned into writes vs skipped. + +**Rationale**: Bounded concurrency (semaphore, default 5) caps transient memory; `return_exceptions=True` is the isolation primitive FR-005 needs — one node's raised exception arrives as a result item instead of aborting the gather. Keeping the read as `query_gql_query(..., update_group=True, subscribers=[id])` preserves the reverse-index registration (FR-007). + +**Alternatives considered**: +- *Keep one Prefect `@task` per node* — rejected: task-run creation/retention per node is pure orchestrator overhead at fan-out scale and is what makes the storm's task volume explode. +- *`asyncio.gather` directly* — rejected: unbounded concurrency; SDK batch already provides the semaphore + node-tagged result stream. + +## R3 — Repository initialization sharing + +**Decision**: `get_initialized_repo(...)` once per attribute batch, pass the repo object into every per-node callable. + +**Rationale**: FR-001. The checkout is keyed by repository+commit — identical for every node in the batch; per-node init was redundant work plus worktree lock contention. + +**Alternatives considered**: per-node init with an internal cache — rejected: the cache would still pay per-node lookup/validation and keeps the contention pattern; hoisting is simpler. + +## R4 — Failure semantics for bad transform results + +**Decision**: Partition results: `Exception` → skip (log reason, keep prior value); non-`str` value → skip (log reason); only `str` values persist. Flow completes green with warnings. + +**Rationale**: FR-005 and the old path's contract: the mutation's `$value: String!` made non-string values fail loudly per node without touching siblings — silently writing `None`/garbage through the bulk path would be a data-corruption regression. Skip-and-log preserves siblings' progress and the failing node's last good value. + +**Alternatives considered**: fail the whole flow on first error — rejected: regresses partial progress; one broken node would starve thousands. Write None on failure — rejected: destroys last good value. + +## R5 — Task-list visibility of batch runs (branch filter) + +**Decision**: Pass `tags=[WorkflowTag.BRANCH.render(identifier=branch_name)]` at `submit_workflow` time for every `COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM` submission. + +**Rationale**: FR-006. The task API filters flow runs by the branch tag. Tag updates made *inside* a flow are rebuilt from the tags known at run creation — a later in-flow tag update (e.g. the dispatcher marking a database change) replaces the tag list and drops anything added mid-run. Only tags present at creation reliably survive, so the branch tag must ride the submission. (Deployment-static tags — namespace, workflow type, database-change — already come from the workflow definition.) + +**Alternatives considered**: merge against live tags read from the orchestrator API inside `add_tags` — rejected: adds one API read to every tagged flow in the system; at storm concurrency this measurably slows the task manager (verified: drains that finish in minutes stopped finishing inside 300–900s bounds). + +## R6 — What deliberately does not change + +- Per-node GraphQL reads and per-node transform executions remain O(N): values are per-node functions of per-node data; the transform is opaque user code. +- Fan-out scoping (all-of-kind on schema/backfill paths, reverse-index on data paths) untouched — out of scope per spec. +- `InfrahubUpdateComputedAttribute` remains a public mutation (external users); it just stops being used internally. +- Submission chunking (`get_submission_chunk_size()`, ≈250 ids/run) unchanged; multiple batches per oversized fan-out (FR-008) already handled by the existing `_chunk_ids` submit loops. diff --git a/dev/specs/004-batch-python-recompute/spec.md b/dev/specs/004-batch-python-recompute/spec.md new file mode 100644 index 0000000000..b9a31e4341 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/spec.md @@ -0,0 +1,120 @@ +# Feature Specification: Batch Python Computed-Attribute Recompute + +**Feature Branch**: `batch-python-recompute-infp-608` + +**Created**: 2026-07-24 + +**Status**: Draft + +**Input**: User description: "Optimize the recompute of Python-transform computed attributes so a change affecting many nodes no longer overwhelms the instance. Today (develop), when a source change fans out to N reader nodes, the per-node path initializes the transform's git repository once per node, issues one update mutation per node, and every mutation's events re-trigger target queries and further recompute dispatches (an echo storm measured at 73k flow runs for one device-type rename on a large dataset, ending in resource exhaustion). Goal: initialize the repo once per batch, collect the recomputed values, persist them through the existing shared bulk recompute writer, isolate per-node transform failures, and keep the process flow runs visible in branch-filtered task queries." + +## User Scenarios & Testing *(mandatory)* + +### User Story 1 - Instance stays usable after a wide-impact change (Priority: P1) + +A network operator renames a device type's part number (directly or by merging a branch). That single change invalidates a Python-computed description on every device of that type — potentially thousands of nodes. Today the resulting background refresh floods the system with follow-on work that grows with the number of affected nodes, keeping the instance degraded for minutes to hours and, at large scale, exhausting server resources until the API stops responding. After this feature, the same change is absorbed as one bounded background pass: the instance stays responsive throughout, and the refresh settles promptly. + +**Why this priority**: This is the customer-reported pain ("every merge recomputes everything → instance unusable for ~20 minutes to an hour"). Removing the self-amplifying churn is the core value; everything else in this feature supports it. + +**Independent Test**: On a restored production-scale dataset, rename one device type and measure (a) time from the change until background activity settles, (b) the number of background task runs spawned, (c) API availability during the window. Compare against the same operation before the change. + +**Acceptance Scenarios**: + +1. **Given** a dataset where one device type is used by N devices, **When** the device type's part number is renamed, **Then** every affected device's computed description reflects the new value once background processing settles. +2. **Given** the same rename, **When** background processing runs, **Then** the number of background task runs is bounded by the fan-out size (no per-node write tasks, no self-retriggered follow-on waves), and the API remains available throughout. +3. **Given** a branch containing the rename, **When** the branch is merged, **Then** the post-merge refresh behaves identically to scenarios 1–2 on the destination branch. + +--- + +### User Story 2 - Unchanged values cause no follow-on work (Priority: P2) + +Many recompute passes produce a value identical to what is already stored (e.g. a re-sync or a change that does not alter the rendered text). An operator whose change leaves most computed values untouched should not pay for update events or cascading refreshes on those untouched nodes. + +**Why this priority**: The echo storm is fed by no-op writes emitting events that re-trigger the recompute machinery. Suppressing no-op propagation is what converts an unbounded cascade into a single pass. + +**Independent Test**: Recompute a set of nodes twice in a row; the second pass must produce zero update events and zero follow-on recompute dispatches. + +**Acceptance Scenarios**: + +1. **Given** a node whose recomputed value equals the stored value, **When** the batch is persisted, **Then** no update event is emitted for that node and no downstream refresh is dispatched for it. +2. **Given** a node whose recomputed value differs, **When** the batch is persisted, **Then** exactly the same per-node update event is emitted as an equivalent direct edit of that attribute would emit today. + +--- + +### User Story 3 - One broken transform target does not block the rest (Priority: P3) + +A user's transform code can fail for a specific node (raise an exception, or return a value of the wrong type) — for example, one device has missing related data. The operator expects every other node in the batch to still be refreshed, with the failing node keeping its previous value and the failure being discoverable in logs. + +**Why this priority**: With batched persistence, a naive implementation would let one bad node abort the whole batch — a regression from today's per-node behavior. Isolation preserves partial progress. + +**Independent Test**: Point a computed attribute at a transform that fails for exactly one node out of many; verify the others update and the failing node retains its prior value with a logged reason. + +**Acceptance Scenarios**: + +1. **Given** a batch of N nodes where one node's transform raises, **When** the batch runs, **Then** N−1 nodes are updated, the failing node's previous value is preserved, and the failure and its reason are logged. +2. **Given** a transform that returns a non-text value for one node, **When** the batch runs, **Then** that node is skipped exactly as in scenario 1 (no null/garbage value is written). + +--- + +### User Story 4 - Operators can still see recompute activity per branch (Priority: P4) + +An operator investigating "did my change refresh the computed values on branch X?" filters the task list by branch. The recompute's processing runs must appear there, both while running and after completion. + +**Why this priority**: Batching replaces many visible per-node tasks with fewer batch runs; if those batch runs are not visible in branch-filtered queries, operators lose their only progress/audit surface for recompute. + +**Independent Test**: Trigger a recompute on a branch and query the task list filtered by that branch; the processing run(s) must be listed. + +**Acceptance Scenarios**: + +1. **Given** a recompute triggered on branch B, **When** the task list is filtered by branch B, **Then** the recompute processing run(s) appear with their state and completion. + +--- + +### Edge Cases + +- Transform raises for every node in the batch → all nodes keep prior values, all failures logged, flow completes without writing. +- Transform returns a non-text value (None, number, object) for some nodes → those nodes are skipped with a logged reason; text values persist unchanged semantics. +- A node in the batch is deleted between fan-out and persistence → the write pass skips it without failing the batch. +- The branch is deleted between fan-out and persistence → the batch is abandoned without error. +- The fan-out exceeds the per-submission size limit → the work is split into multiple bounded batch runs, every affected node processed exactly once. +- The recomputed value equals the stored value for all nodes (full no-op pass) → zero events, zero follow-on dispatches. +- The transform's source repository is unavailable at batch start → the batch fails visibly before touching any node (no partial ambiguity from per-node repository setup). + +## Requirements *(mandatory)* + +### Functional Requirements + +- **FR-001**: The system MUST prepare the transform's execution source (repository checkout) once per batch of affected nodes, not once per node. +- **FR-002**: The system MUST persist the recomputed values of a batch through the existing shared bulk recompute write path (bounded write transactions), not via one client-visible update operation per node. +- **FR-003**: A persisted value identical to the stored value MUST NOT emit an update event and MUST NOT trigger any downstream recompute dispatch. +- **FR-004**: A persisted value that differs from the stored value MUST emit the same per-node update event an equivalent direct attribute update emits, so downstream consumers (webhooks, UI, dependent computed values) observe no behavioral change. +- **FR-005**: A node whose transform raises or returns a non-text value MUST be skipped — previous value preserved, reason logged — without affecting the persistence of other nodes in the batch. +- **FR-006**: Batch processing runs MUST remain discoverable via branch-filtered task queries, during and after execution. +- **FR-007**: The per-node data reads performed by the recompute MUST keep registering each node as a subscriber of the transform's query (the reverse-index that routes future source changes to affected readers must stay current). +- **FR-008**: Fan-outs larger than the existing per-submission size limit MUST be split into multiple bounded batches, with every affected node processed exactly once across batches. +- **FR-009**: The final stored values after a batch recompute MUST be identical to what today's per-node path produces for the same inputs (correctness parity). + +### Key Entities + +- **Computed attribute (Python transform)**: a read-only attribute whose value is produced by user-supplied transform code reading the node's own data via a named query. +- **Recompute batch**: the set of affected node ids processed by one background run — reads and transform executions per node, one shared persistence pass. +- **Skipped node**: a batch member whose transform failed; retains its prior value, recorded with a reason. + +## Success Criteria *(mandatory)* + +### Measurable Outcomes + +- **SC-001**: On the reference small dataset, post-merge background processing settles in under 10 seconds (previously ~4.6 minutes); on the reference large dataset, the same operation settles without API unavailability (previously never settled and took the instance down). +- **SC-002**: Total background task runs spawned by one source change drop by ≥ 99% at scale (reference: 73,000+ runs for one rename on the large dataset → bounded by ⌈N / submission-limit⌉ plus a constant). +- **SC-003**: Final computed values are byte-identical to the previous behavior for the same inputs (100% correctness parity on the reference scenarios). +- **SC-004**: The API remains available (no sustained error responses attributable to the recompute) throughout the refresh on the large reference dataset, where previously the instance became unresponsive. +- **SC-005**: With one failing node in a batch of N, N−1 nodes are refreshed and the failure is discoverable in logs (no silent data loss, no batch abort). + +## Assumptions + +- The existing shared bulk recompute writer (used by the template-based computed attributes) is the persistence mechanism to reuse; it already provides bounded transactions, skip-unchanged gating, and per-node update events for real changes. +- The existing per-submission size limit (derived from the task-orchestration platform's parameter cap) continues to bound batch size; this feature does not introduce a new limit. +- Which nodes are affected by a change (fan-out scoping, all-of-kind vs affected-only) is explicitly out of scope — this feature changes how a batch is processed, not which nodes enter it. +- Per-node update events for real changes are retained at the same granularity; consumers of those events are unaffected. +- Transform code is user-supplied and opaque: per-node execution cannot be merged or vectorized; only setup and persistence around it can be shared. +- Failure recovery/retry UX for skipped nodes (discovery surface, one-click re-run) is a separate follow-up effort, not part of this feature. diff --git a/dev/specs/004-batch-python-recompute/tasks.md b/dev/specs/004-batch-python-recompute/tasks.md new file mode 100644 index 0000000000..9e96e7e2c0 --- /dev/null +++ b/dev/specs/004-batch-python-recompute/tasks.md @@ -0,0 +1,89 @@ +# Tasks: Batch Python Computed-Attribute Recompute + +**Input**: Design documents from `specs/004-batch-python-recompute/` (plan.md, research.md, data-model.md, contracts/, quickstart.md) + +**Tests**: Included — constitution IV (Test Discipline) and critique items E8/E9 mandate them. + +**Organization**: Tasks grouped by user story from spec.md; each story phase is independently testable. + +## Phase 1: Setup + +- [ ] T001 Establish the green baseline: run `uv run pytest backend/tests/unit/computed_attribute/ backend/tests/functional/computed_attributes/ -q` on the unmodified branch and record the passing set (guards the correctness-parity claim, FR-009) + +## Phase 2: Foundational (blocking all user stories) + +- [ ] T002 Write a component characterization test capturing today's fan-out behavior (a source change recomputes all N readers via the reverse index) in backend/tests/component/computed_attribute/test_merge_fanout_python.py — this pins FR-009 parity before the rewrite +- [ ] T003 Add plain-async helper `_transform_value_for_node` in backend/infrahub/computed_attribute/tasks.py: takes a pre-initialized `repo`, runs `client.query_gql_query(name=query_id, variables={"id": object_id}, update_group=True, subscribers=[object_id])` (FR-007), executes the transform, returns `AttributeValueWrite(node_id, field=attribute_name, value=...)` +- [ ] T004 Add pure helper `_partition_transform_results(results) -> (writes, skipped)` in backend/infrahub/computed_attribute/tasks.py: `Exception` → skipped(reason), non-`str` value → skipped(reason), `str` → writes (FR-005 substrate) + +## Phase 3: User Story 1 — Instance stays usable after a wide-impact change (P1) 🎯 MVP + +**Goal**: one bounded batch pass per chunk — repo init once, values persisted in bulk, zero client-visible mutations, zero echo. + +**Independent test**: component fan-out test still passes; recompute issues no `InfrahubUpdateComputedAttribute` calls; repeated recompute produces no second wave. + +- [ ] T005 [US1] Rewrite `process_transform` in backend/infrahub/computed_attribute/tasks.py: hoist `get_initialized_repo(...)` to once per attribute batch (FR-001); build `client.create_batch(return_exceptions=True)` over `_transform_value_for_node`; collect results; partition via `_partition_transform_results`; log one warning per skipped node +- [ ] T006 [US1] Persist via the shared writer in backend/infrahub/computed_attribute/tasks.py: `dispatcher = await build_bulk_recompute_dispatcher(schema_branch=...)` once per flow; `await dispatcher.dispatch(writes=writes, branch_name=..., context=..., coalesced=False, recompute_depth=0)` (FR-002; `coalesced=False` keeps live-origin events per research R1) +- [ ] T007 [US1] Delete the per-node Prefect task `process_transform_for_node` and the `UPDATE_ATTRIBUTE` GraphQL constant from backend/infrahub/computed_attribute/tasks.py; keep the public mutation untouched elsewhere (contracts/README.md) +- [ ] T008 [US1] Guard empty input: `process_transform` returns early when the id set is empty (avoids dispatcher work and spurious tags) in backend/infrahub/computed_attribute/tasks.py +- [ ] T009 [P] [US1] Update the component test from T002 to assert the rewritten path: same final values (FR-009), no per-node mutation calls, fan-out count unchanged in backend/tests/component/computed_attribute/test_merge_fanout_python.py + +**Checkpoint**: US1 delivers the MVP — batch pass with bulk persistence, correctness parity proven. + +## Phase 4: User Story 2 — Unchanged values cause no follow-on work (P2) + +**Goal**: no-op writes emit no events and dispatch no downstream recompute (echo eliminated). + +**Independent test**: recompute the same set twice; second pass emits zero events and zero dispatches. + +- [ ] T010 [US2] Component test: recompute twice in a row with an event-recorder adapter; assert second pass emits zero NodeUpdated events and zero recompute submissions (FR-003) in backend/tests/component/computed_attribute/test_skip_unchanged_python.py +- [ ] T011 [P] [US2] Component test: a genuinely changed value emits exactly one NodeUpdated event with live origin, identical shape to a direct attribute update (FR-004) in backend/tests/component/computed_attribute/test_skip_unchanged_python.py + +**Checkpoint**: echo-storm mechanism verifiably dead at component tier. + +## Phase 5: User Story 3 — One broken transform target does not block the rest (P3) + +**Goal**: per-node failure isolation with prior values preserved and reasons logged. + +**Independent test**: transform failing for 1 of N nodes → N−1 updated, 1 skipped with logged reason. + +- [ ] T012 [P] [US3] Unit tests for `_partition_transform_results` in backend/tests/unit/computed_attribute/test_tasks.py: string persisted; None/non-str skipped with type-named reason; exception isolated with repr reason; empty input yields empty outputs (FR-005) +- [ ] T013 [US3] Functional test: real repo + transform raising for exactly one node → siblings updated, failing node retains prior value, warning logged in backend/tests/functional/computed_attributes/test_computed_attribute.py +- [ ] T014 [US3] Add the flow-end summary log line `recompute complete: written=…, unchanged=…, skipped=…` in backend/infrahub/computed_attribute/tasks.py (critique P3/X1) + +**Checkpoint**: partial progress preserved; failures greppable. + +## Phase 6: User Story 4 — Operators can still see recompute activity per branch (P4) + +**Goal**: batch runs discoverable via branch-filtered task queries, during and after execution. + +**Independent test**: trigger recompute on a branch; task query filtered by that branch lists the process run. + +- [ ] T015 [US4] Pass `tags=[WorkflowTag.BRANCH.render(identifier=branch_name)]` at both `COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM` submission sites in backend/infrahub/computed_attribute/tasks.py (research R5: creation tags survive in-flow tag rebuilds; mid-run tagging does not) +- [ ] T016 [US4] Functional/component test: trigger a recompute on a branch, assert the process run appears in a branch-filtered task query (critique E8, FR-006) in backend/tests/functional/computed_attributes/test_computed_attribute.py +- [ ] T017 [P] [US4] Unit test with recorder workflow adapter: fan-out of `chunk_limit*2+1` ids → 3 submissions, ids partitioned exactly once, branch tag on every submission (critique E9, FR-008) in backend/tests/unit/computed_attribute/test_tasks.py + +**Checkpoint**: visibility regression-proofed — the property most at risk from removing per-node tasks. + +## Final Phase: Polish & Cross-Cutting + +- [ ] T018 [P] Add changelog fragment changelog/+python-computed-attribute-bulk-recompute.changed.md describing the batching and its user-visible effect (settle time, no echo) +- [ ] T019 [P] Update dev/knowledge/backend/computed-attributes.md: document the batch path, failure semantics (skip+log), crash/rollback semantics from plan.md, and the shared-checkout invariant (critique E2/E3/P5) +- [ ] T020 Run `uv run invoke format lint` and the full T001 test set; confirm parity with the recorded baseline +- [ ] T021 Optional at-scale validation per quickstart.md: perf A/B (`TestMergeRecomputePython`) comparing baseline vs feature ref — confirms SC-001/002/004 + +## Dependencies + +- Phase 2 → everything (T003/T004 are the substrate; T002 pins parity first) +- US1 (T005–T009) → US2/US3/US4 build on the rewritten flow +- US2, US3, US4 are mutually independent after US1 +- Final phase last + +## Parallel Execution Examples + +- After T005–T008 land: T009, T010/T011, T012, T017 touch different files → parallel +- T018/T019 parallel with any test task + +## Implementation Strategy + +MVP = Phase 1–3 (US1): batching + bulk persistence with parity proof — deployable alone since skip-unchanged (US2) is inherited from the writer rather than newly built; US2's tasks only *verify* it. Then US3 (isolation) before US4 (visibility), then polish. Total: 21 tasks (US1: 5, US2: 2, US3: 3, US4: 3, setup/foundational: 4, polish: 4). From e156878dbf197ba4e038b43b710d793d0aa58ced Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Sun, 26 Jul 2026 16:07:15 +0300 Subject: [PATCH 12/13] test(perf): make Neo4j Bolt thread-pool size tunable for dataset runs At `large` dataset scale the computed-attribute recompute fan-out (query-computed-attribute-transform-targets + graphql-query-group-update + process_transform) can saturate Neo4j's Bolt connector thread pool, which defaults to 400, producing Neo.TransientError.Request.NoThreadsAvailable and taking the API servers down mid-run before the measured scenario is reached. Expose NEO4J_server_bolt_thread__pool__max__size in both testcontainers compose stacks via the new INFRAHUB_TESTING_DB_BOLT_THREAD_POOL_MAX_SIZE env var, sized alongside the existing heap/pagecache knobs. The default is 400 (Neo4j's own default), so normal test runs are unchanged; heavy dataset/perf runs raise it from the workflow environment. Providing an explicit default also avoids an empty value reaching an integer Neo4j setting when the var is unset. Co-Authored-By: Claude Opus 4.8 (1M context) --- changelog/+neo4j-bolt-thread-pool-tunable.housekeeping.md | 1 + python_testcontainers/infrahub_testcontainers/container.py | 4 ++++ .../infrahub_testcontainers/docker-compose-cluster.test.yml | 4 ++++ .../infrahub_testcontainers/docker-compose.test.yml | 4 ++++ 4 files changed, 13 insertions(+) create mode 100644 changelog/+neo4j-bolt-thread-pool-tunable.housekeeping.md diff --git a/changelog/+neo4j-bolt-thread-pool-tunable.housekeeping.md b/changelog/+neo4j-bolt-thread-pool-tunable.housekeeping.md new file mode 100644 index 0000000000..6f2443228f --- /dev/null +++ b/changelog/+neo4j-bolt-thread-pool-tunable.housekeeping.md @@ -0,0 +1 @@ +Made the Neo4j Bolt connector thread-pool ceiling tunable in the testcontainers stacks via the `INFRAHUB_TESTING_DB_BOLT_THREAD_POOL_MAX_SIZE` environment variable (defaulting to Neo4j's own `400`, so normal test runs are unaffected). Heavy dataset/performance runs can raise it to avoid `Neo.TransientError.Request.NoThreadsAvailable` when a wide computed-attribute recompute fan-out saturates the pool. diff --git a/python_testcontainers/infrahub_testcontainers/container.py b/python_testcontainers/infrahub_testcontainers/container.py index a01510f2c1..3759b54b42 100644 --- a/python_testcontainers/infrahub_testcontainers/container.py +++ b/python_testcontainers/infrahub_testcontainers/container.py @@ -67,6 +67,10 @@ class ContainerService: "INFRAHUB_TESTING_SCHEMA_STRICT_MODE": "true", "INFRAHUB_TESTING_TASKMGR_API_WORKERS": "1", "INFRAHUB_TESTING_TASKMGR_BACKGROUND_SVC_REPLICAS": "0", + # Neo4j Bolt connector thread-pool ceiling. 400 matches Neo4j's own default, so this is a + # no-op for normal test runs; heavy dataset/perf runs override it via the environment to + # avoid Neo.TransientError.Request.NoThreadsAvailable under a wide recompute fan-out. + "INFRAHUB_TESTING_DB_BOLT_THREAD_POOL_MAX_SIZE": "400", } diff --git a/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml b/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml index bebee83f6b..e77f1b86cf 100644 --- a/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml +++ b/python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml @@ -18,6 +18,10 @@ x-neo4j-config-common: &neo4j-config-common NEO4J_server_memory_heap_initial__size: ${INFRAHUB_TESTING_DB_HEAP_INITIAL_SIZE} NEO4J_server_memory_heap_max__size: ${INFRAHUB_TESTING_DB_HEAP_MAX_SIZE} NEO4J_server_memory_pagecache_size: ${INFRAHUB_TESTING_DB_PAGECACHE_SIZE} + # Bolt connector worker threads. Neo4j defaults to 400; a wide computed-attribute + # fan-out on large datasets can exhaust it (Neo.TransientError.Request.NoThreadsAvailable), + # so make it tunable for the heavy perf/dataset runs. + NEO4J_server_bolt_thread__pool__max__size: ${INFRAHUB_TESTING_DB_BOLT_THREAD_POOL_MAX_SIZE} services: diff --git a/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml b/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml index d92aefc569..34d608a01e 100644 --- a/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml +++ b/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml @@ -58,6 +58,10 @@ services: NEO4J_server_memory_heap_initial__size: ${INFRAHUB_TESTING_DB_HEAP_INITIAL_SIZE} NEO4J_server_memory_heap_max__size: ${INFRAHUB_TESTING_DB_HEAP_MAX_SIZE} NEO4J_server_memory_pagecache_size: ${INFRAHUB_TESTING_DB_PAGECACHE_SIZE} + # Bolt connector worker threads. Neo4j defaults to 400; a wide computed-attribute + # fan-out on large datasets can exhaust it (Neo.TransientError.Request.NoThreadsAvailable), + # so make it tunable for the heavy perf/dataset runs. + NEO4J_server_bolt_thread__pool__max__size: ${INFRAHUB_TESTING_DB_BOLT_THREAD_POOL_MAX_SIZE} volumes: - "database_data:/data" - "database_logs:/logs" From e67ebee650edc4347891cdd819206139de602286 Mon Sep 17 00:00:00 2001 From: Saltaferis Dimitrios Date: Sun, 26 Jul 2026 20:45:48 +0300 Subject: [PATCH 13/13] test(perf): raise file-descriptor ceiling for testcontainers stack The wide computed-attribute recompute fan-out on large datasets exhausts the default 1024 file-descriptor limit on the task-worker (and, to a lesser degree, the API server and database), surfacing as `ConnectError: [Errno 24] Too many open files` and taking the stack down during warmup. Set nofile soft/hard to 1048576 on the database, infrahub-server, and task-worker services. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../docker-compose.test.yml | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml b/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml index 34d608a01e..f64334aa47 100644 --- a/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml +++ b/python_testcontainers/infrahub_testcontainers/docker-compose.test.yml @@ -62,6 +62,12 @@ services: # fan-out on large datasets can exhaust it (Neo.TransientError.Request.NoThreadsAvailable), # so make it tunable for the heavy perf/dataset runs. NEO4J_server_bolt_thread__pool__max__size: ${INFRAHUB_TESTING_DB_BOLT_THREAD_POOL_MAX_SIZE} + ulimits: + # Default container soft limit (1024) is far too low for a wide recompute fan-out and + # produces "[Errno 24] Too many open files"; raise the file-descriptor ceiling. + nofile: + soft: 1048576 + hard: 1048576 volumes: - "database_data:/data" - "database_logs:/logs" @@ -177,6 +183,12 @@ services: replicas: ${INFRAHUB_TESTING_API_SERVER_COUNT} image: "${INFRAHUB_TESTING_DOCKER_IMAGE}:${INFRAHUB_TESTING_IMAGE_VERSION}" command: ${INFRAHUB_TESTING_DOCKER_ENTRYPOINT} + ulimits: + # Raise the file-descriptor ceiling so a wide recompute fan-out doesn't hit + # "[Errno 24] Too many open files" on the API server's outbound connections. + nofile: + soft: 1048576 + hard: 1048576 environment: INFRAHUB_PRODUCTION: ${INFRAHUB_TESTING_PRODUCTION} INFRAHUB_LOG_LEVEL: ${INFRAHUB_TESTING_LOG_LEVEL:-INFO} @@ -223,6 +235,12 @@ services: replicas: ${INFRAHUB_TESTING_TASK_WORKER_COUNT} image: "${INFRAHUB_TESTING_DOCKER_IMAGE}:${INFRAHUB_TESTING_IMAGE_VERSION}" command: prefect worker start --type ${INFRAHUB_TESTING_WORKFLOW_DEFAULT_WORKER_TYPE} --pool infrahub-worker --with-healthcheck + ulimits: + # The recompute fan-out runs here; the worker opens many concurrent DB/HTTP + # connections and hits "[Errno 24] Too many open files" at the default 1024 ceiling. + nofile: + soft: 1048576 + hard: 1048576 environment: INFRAHUB_PRODUCTION: ${INFRAHUB_TESTING_PRODUCTION} INFRAHUB_LOG_LEVEL: ${INFRAHUB_TESTING_LOG_LEVEL}