From 4175262b8b8f5ec8c581ecb44132bb913b865c75 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 8 Sep 2026 07:47:18 +0200 Subject: [PATCH] fix(security): isolate reference execution and bound source inputs --- .../backend/backends/base.py | 13 +- .../backend/backends/reference.py | 169 ++++++++++++++- src/govoplan_dataflow/backend/manifest.py | 34 +++ src/govoplan_dataflow/backend/service.py | 66 ++++-- tests/test_reference_process.py | 117 +++++++++++ tests/test_reference_subflow_process.py | 194 ++++++++++++++++++ 6 files changed, 574 insertions(+), 19 deletions(-) create mode 100644 tests/test_reference_process.py create mode 100644 tests/test_reference_subflow_process.py diff --git a/src/govoplan_dataflow/backend/backends/base.py b/src/govoplan_dataflow/backend/backends/base.py index fb2ae81..6ef32b9 100644 --- a/src/govoplan_dataflow/backend/backends/base.py +++ b/src/govoplan_dataflow/backend/backends/base.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field +import math from typing import Any, Mapping, Protocol, runtime_checkable from govoplan_dataflow.backend.batches import TypedBatch @@ -33,8 +34,8 @@ class ExecutionBudget: raise ValueError("Execution output row limit must be positive.") if self.max_batch_bytes < 1: raise ValueError("Execution byte limit must be positive.") - if self.max_wall_seconds <= 0: - raise ValueError("Execution time limit must be positive.") + if not math.isfinite(self.max_wall_seconds) or self.max_wall_seconds <= 0: + raise ValueError("Execution time limit must be finite and positive.") if self.max_memory_bytes < 64 * 1024 * 1024: raise ValueError("Execution memory limit must be at least 64 MiB.") if self.max_concurrency < 1: @@ -106,11 +107,19 @@ class BackendExecutionError(RuntimeError): code: str = "backend.execution", node_id: str | None = None, diagnostics: tuple[DataflowDiagnostic, ...] = (), + node_diagnostics: tuple[NodePreviewDiagnostic, ...] = (), + source_fingerprints: tuple[dict[str, Any], ...] = (), + input_row_count: int = 0, + node_preview: NodePreviewResult | None = None, ) -> None: super().__init__(message) self.code = code self.node_id = node_id self.diagnostics = diagnostics + self.node_diagnostics = node_diagnostics + self.source_fingerprints = source_fingerprints + self.input_row_count = input_row_count + self.node_preview = node_preview @runtime_checkable diff --git a/src/govoplan_dataflow/backend/backends/reference.py b/src/govoplan_dataflow/backend/backends/reference.py index b59650c..959e34d 100644 --- a/src/govoplan_dataflow/backend/backends/reference.py +++ b/src/govoplan_dataflow/backend/backends/reference.py @@ -1,10 +1,25 @@ from __future__ import annotations +from dataclasses import asdict +import hashlib +import json +import math + +from govoplan_core.security.bounded_process import ( + ProcessBudgetError, + ProcessLimits, + run_bounded_operation, +) +from govoplan_core.security.worker_payload import ( + decode_worker_payload, + encode_worker_payload, +) from govoplan_dataflow.backend.backends.base import ( BackendExecutionError, BackendExecutionRequest, BackendExecutionResult, BackendSource, + ExecutionBudget, canonical_result_schema, ) from govoplan_dataflow.backend.batches import TypedBatch @@ -14,8 +29,17 @@ from govoplan_dataflow.backend.executor import ( ResolvedSource, execute_preview, ) -from govoplan_dataflow.backend.ir import IrExecutionResult, ir_to_graph -from govoplan_dataflow.backend.schemas import GraphNode +from govoplan_dataflow.backend.ir import IrExecutionResult, IrSchema, TypedGraphIr, ir_to_graph +from govoplan_dataflow.backend.planner import ExecutionPlan +from govoplan_dataflow.backend.schemas import ( + DataflowDiagnostic, + GraphNode, + NodePreviewDiagnostic, + NodePreviewResult, +) + + +_TRANSPORT_BYTES = 32 * 1024 * 1024 class ReferenceExecutionBackend: @@ -32,6 +56,52 @@ class ReferenceExecutionBackend: self, request: BackendExecutionRequest, ) -> BackendExecutionResult: + _validate_source_batches(request) + try: + limits = ProcessLimits( + wall_seconds=request.budget.max_wall_seconds, + cpu_seconds=max(1, math.ceil(request.budget.max_wall_seconds)), + memory_bytes=request.budget.max_memory_bytes, + input_bytes=_TRANSPORT_BYTES, + output_bytes=_TRANSPORT_BYTES, + ) + payload = encode_worker_payload(_request_payload(request), max_bytes=_TRANSPORT_BYTES) + response = decode_worker_payload( + run_bounded_operation(_execute_reference_worker, payload, limits=limits), + max_bytes=_TRANSPORT_BYTES, + ) + except ProcessBudgetError as exc: + raise BackendExecutionError(str(exc), code=f"backend.process.{exc.code}") from exc + except ValueError as exc: + raise BackendExecutionError( + "Reference execution exceeds its supported process/transport budget.", + code="backend.budget", + ) from exc + if "error" in response: + error = response["error"] + raise BackendExecutionError( + error["message"], code=error["code"], node_id=error["node_id"], + diagnostics=tuple(DataflowDiagnostic.model_validate(item) for item in error["diagnostics"]), + node_diagnostics=tuple(NodePreviewDiagnostic.model_validate(item) for item in error["node_diagnostics"]), + source_fingerprints=error["source_fingerprints"], + input_row_count=error["input_row_count"], + node_preview=NodePreviewResult.model_validate(error["node_preview"]) if error["node_preview"] else None, + ) + batch = _batch_from_payload(response["batch"]) + batch.ensure_within( + max_rows=request.budget.max_output_rows, + max_bytes=request.budget.max_batch_bytes, + ) + return BackendExecutionResult( + contract=IrExecutionResult.model_validate(response["contract"]), + batch=batch, + node_diagnostics=tuple(NodePreviewDiagnostic.model_validate(item) for item in response["node_diagnostics"]), + node_preview=NodePreviewResult.model_validate(response["node_preview"]) if response["node_preview"] else None, + metadata=response["metadata"], + ) + + def _execute_in_process(self, request: BackendExecutionRequest) -> BackendExecutionResult: + """Pure reference evaluation, called only inside the disposable worker.""" _validate_source_batches(request) try: result = execute_preview( @@ -50,6 +120,10 @@ class ReferenceExecutionBackend: code="backend.reference", node_id=exc.node_id, diagnostics=tuple(exc.diagnostics), + node_diagnostics=tuple(exc.node_diagnostics), + source_fingerprints=tuple(exc.source_fingerprints), + input_row_count=exc.input_row_count, + node_preview=exc.node_preview, ) from exc observed_batch = TypedBatch.from_rows(result.rows) batch = TypedBatch.from_rows( @@ -97,6 +171,90 @@ class ReferenceExecutionBackend: ) +def _batch_payload(batch: TypedBatch) -> dict: + return { + "schema": batch.schema.model_dump(mode="python"), + "columns": dict(batch.columns), + "row_count": batch.row_count, + "byte_count": batch.byte_count, + } + + +def _batch_from_payload(value: dict) -> TypedBatch: + return TypedBatch( + schema=IrSchema.model_validate(value["schema"]), columns=value["columns"], + row_count=value["row_count"], byte_count=value["byte_count"], + ) + + +def _request_payload(request: BackendExecutionRequest) -> dict: + plan = request.plan + return { + "plan": { + "graph": plan.graph.model_dump(mode="python"), + "ordered_node_ids": plan.ordered_node_ids, + "diagnostics": tuple(item.model_dump(mode="python") for item in plan.diagnostics), + "generated_sql": plan.generated_sql, + "sql_diagnostics": tuple(item.model_dump(mode="python") for item in plan.sql_diagnostics), + "semantic_hash": plan.semantic_hash, + }, + "budget": asdict(request.budget), + "preview_node_id": request.preview_node_id, + "sources": { + key: { + "node_id": source.node_id, "batch": _batch_payload(source.batch), + "source_ref": source.source_ref, "provider": source.provider, + "fingerprint": source.fingerprint, "total_rows": source.total_rows, + "truncated": source.truncated, "source_name": source.source_name, + "kind": source.kind, + } + for key, source in request.sources.items() + }, + } + + +def _execute_reference_worker(payload: bytes) -> bytes: + """Data-only worker boundary; no database, provider callback or principal.""" + value = decode_worker_payload(payload, max_bytes=_TRANSPORT_BYTES) + plan = value["plan"] + request = BackendExecutionRequest( + plan=ExecutionPlan( + graph=TypedGraphIr.model_validate(plan["graph"]), + ordered_node_ids=plan["ordered_node_ids"], + diagnostics=tuple(DataflowDiagnostic.model_validate(item) for item in plan["diagnostics"]), + generated_sql=plan["generated_sql"], + sql_diagnostics=tuple(DataflowDiagnostic.model_validate(item) for item in plan["sql_diagnostics"]), + semantic_hash=plan["semantic_hash"], + ), + budget=ExecutionBudget(**value["budget"]), + preview_node_id=value["preview_node_id"], + sources={ + key: BackendSource(**{**source, "batch": _batch_from_payload(source["batch"])}) + for key, source in value["sources"].items() + }, + ) + try: + result = ReferenceExecutionBackend()._execute_in_process(request) + except BackendExecutionError as exc: + response = {"error": { + "message": str(exc), "code": exc.code, "node_id": exc.node_id, + "diagnostics": tuple(item.model_dump(mode="python") for item in exc.diagnostics), + "node_diagnostics": tuple(item.model_dump(mode="python") for item in exc.node_diagnostics), + "source_fingerprints": exc.source_fingerprints, + "input_row_count": exc.input_row_count, + "node_preview": exc.node_preview.model_dump(mode="python") if exc.node_preview else None, + }} + else: + response = { + "contract": result.contract.model_dump(mode="python"), + "batch": _batch_payload(result.batch), + "node_diagnostics": tuple(item.model_dump(mode="python") for item in result.node_diagnostics), + "node_preview": result.node_preview.model_dump(mode="python") if result.node_preview else None, + "metadata": dict(result.metadata), + } + return encode_worker_payload(response, max_bytes=_TRANSPORT_BYTES) + + def _validate_source_batches(request: BackendExecutionRequest) -> None: for source in request.sources.values(): try: @@ -139,6 +297,7 @@ def _source_for_node( node: GraphNode, ) -> BackendSource | None: candidates = ( + reference_source_key(node), node.id, str(node.config.get("source_ref") or ""), str(node.config.get("source_name") or ""), @@ -153,4 +312,10 @@ def _source_for_node( ) +def reference_source_key(node: GraphNode) -> str: + """Nested graphs may reuse node IDs; bind resolved data to the full config.""" + content = json.dumps(node.config, sort_keys=True, separators=(",", ":"), default=str) + return "reference-config:" + hashlib.sha256(content.encode("utf-8")).hexdigest() + + __all__ = ["ReferenceExecutionBackend"] diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 9b54d79..15650b6 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -154,6 +154,40 @@ ROLE_TEMPLATES = ( ) DOCUMENTATION = localize_documentation_topics(( + DocumentationTopic( + id="dataflow.reference-worker-limits", + title="Reference execution process limits", + summary="Contain expensive expressions and intermediate allocations without changing datasource authority.", + body=( + "Reference previews and reference development runs evaluate in a fresh disposable process, including regex and aggregate intermediate allocations. " + "The existing row, node and per-result byte checks remain. The process additionally enforces the request's wall-clock and memory budgets " + "(default preview: 2 seconds and 256 MiB virtual address space), rounded-up CPU seconds, no regular-file output, and 32 MiB per data-only input/result transport. " + "Supported process budgets are at most 600 seconds and 8 GiB; unsupported controls or exceeded limits fail with structured backend.process diagnostics, never inline fallback. " + "Datasource authorization and bounded source reads remain in the parent, including nested subflow sources; sessions and credentials are not passed to the child. " + "Reference source collection also checks a cumulative 32 MiB typed-data budget before constructing another columnar copy; providers retain their separate per-read limits. " + "Completed node diagnostics survive ordinary evaluation errors; a killed worker returns no partial rows or invented node progress. " + "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY limits shared isolated-work admission per API/worker process, default 1; busy capacity is retryable. " + "It is not a fleet-wide quota or arbitrary-code sandbox. Cancellation checks before/after reference runs remain; hard wall limits stop an unresponsive expression. " + "Staging/production still require the separate DuckDB backend and are not converted to reference execution." + ), + layer="static", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=6, + translations={"de": { + "title": "Prozessgrenzen der Referenzausführung", + "summary": "Aufwendige Ausdrücke und Zwischenspeicher begrenzen, ohne Datenquellenrechte zu verändern.", + "body": ( + "Referenz-Vorschauen und Referenz-Entwicklungsläufe werten Ausdrücke in einem frischen, kurzlebigen Prozess aus, einschließlich regulärer Ausdrücke und großer Zwischenergebnisse. " + "Bestehende Zeilen-, Knoten- und Ergebnis-Bytegrenzen bleiben bestehen. Zusätzlich gelten das Laufzeit- und Speicherbudget der Anfrage " + "(Vorschau standardmäßig 2 Sekunden und 256 MiB virtueller Adressraum), aufgerundete CPU-Sekunden, keine regulären Ausgabedateien und je 32 MiB für den reinen Datentransport. " + "Prozessbudgets unterstützen höchstens 600 Sekunden und 8 GiB. Fehlende Betriebssystemkontrollen oder überschrittene Grenzen erzeugen strukturierte backend.process-Diagnosen, ohne Ausweichbetrieb im Hauptprozess. " + "Datenquellenrechte und begrenzte Quellabrufe werden im Hauptprozess geprüft, auch für verschachtelte Teilflüsse; Sitzungen und Zugangsdaten gelangen nicht in den Kindprozess. " + "Referenz-Quellabrufe prüfen außerdem zusammen höchstens 32 MiB typisierte Daten, bevor eine weitere spaltenweise Kopie entsteht; getrennte Abrufgrenzen der Anbieter bleiben bestehen. " + "Gewöhnliche Auswertungsfehler behalten bereits abgeschlossene Knotendiagnosen. Ein gestoppter Prozess liefert keine Teilzeilen und keinen erfundenen Knotenfortschritt. " + "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY begrenzt gemeinsam genutzte isolierte Arbeit je API-/Worker-Prozess, standardmäßig 1; bei Auslastung ist ein erneuter Versuch möglich. " + "Dies ist weder eine systemweite Quote noch eine Sandbox für beliebigen Code. Abbruchprüfungen vor und nach Referenzläufen bleiben erhalten; harte Laufzeitgrenzen stoppen hängende Ausdrücke. " + "Staging und Produktion benötigen weiterhin das gesonderte DuckDB-Backend und wechseln nicht zur Referenzausführung." + ), + }}, + ), DocumentationTopic( id="dataflow.workspace-layout", title="Dataflow workspace actions", diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index fbb525c..194fc44 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -28,6 +28,7 @@ from govoplan_core.core.datasources import ( datasource_publication, ) from govoplan_core.db.base import utcnow +from govoplan_core.security.worker_payload import WorkerPayloadError, encode_worker_payload from govoplan_dataflow.backend.backends import ( BackendExecutionError, BackendSource, @@ -35,6 +36,7 @@ from govoplan_dataflow.backend.backends import ( execute_typed_graph, ) from govoplan_dataflow.backend.batches import TypedBatch +from govoplan_dataflow.backend.backends.reference import reference_source_key from govoplan_dataflow.backend.db.models import ( DataflowPipeline, DataflowPipelineDeployment, @@ -48,7 +50,6 @@ from govoplan_dataflow.backend.executor import ( PipelineExecutionError, PipelineExecutionResult, ResolvedSource, - execute_preview, ) from govoplan_dataflow.backend.governance import ( definition_governance_payload, @@ -1122,20 +1123,11 @@ def _execute_pipeline_preview( principal=principal, registry=registry, ) - if backend == "reference": - return ( - execute_preview( - graph, - row_limit=row_limit, - source_resolver=source_resolver, - preview_node_id=preview_node_id, - ), - EXECUTOR_VERSION, - ) sources = _typed_backend_sources( graph, source_resolver=source_resolver, - source_limit=max(MAX_SOURCE_ROWS, row_limit), + source_limit=MAX_SOURCE_ROWS if backend == "reference" else max(MAX_SOURCE_ROWS, row_limit), + include_subflows=backend == "reference", ) try: result = execute_typed_graph( @@ -1149,8 +1141,14 @@ def _execute_pipeline_preview( raise PipelineExecutionError( str(exc), node_id=exc.node_id, - diagnostics=tuple(exc.diagnostics), - retryable=exc.code == "backend.capacity", + diagnostics=(*exc.diagnostics, DataflowDiagnostic( + severity="error", code=exc.code, message=str(exc), node_id=exc.node_id, + )), + node_diagnostics=exc.node_diagnostics, + source_fingerprints=exc.source_fingerprints, + input_row_count=exc.input_row_count, + node_preview=exc.node_preview, + retryable=exc.code in {"backend.capacity", "backend.process.busy"}, ) from exc columns = [ PreviewColumn( @@ -1227,13 +1225,51 @@ def _typed_backend_sources( *, source_resolver, source_limit: int = MAX_SOURCE_ROWS, + include_subflows: bool = False, + _depth: int = 0, + _remaining_source_bytes: list[int] | None = None, ) -> dict[str, BackendSource]: + if _depth > 5: + raise PipelineExecutionError("Subflows are limited to five nested levels.") + if _remaining_source_bytes is None: + _remaining_source_bytes = [32 * 1024 * 1024] sources: dict[str, BackendSource] = {} for node in graph.nodes: + if include_subflows and node.type == "subflow": + from govoplan_dataflow.backend.subflows import substitute_parameters + + parameters = node.config.get("parameters") + nested = PipelineGraph.model_validate(substitute_parameters( + node.config.get("graph"), parameters if isinstance(parameters, dict) else {}, + )) + sources.update(_typed_backend_sources( + nested, source_resolver=source_resolver, source_limit=source_limit, + include_subflows=True, _depth=_depth + 1, + _remaining_source_bytes=_remaining_source_bytes, + )) if node.type != "source.reference": continue + if include_subflows and _remaining_source_bytes[0] <= 0: + raise PipelineExecutionError( + "Combined source data exceeds the 32 MiB transfer budget.", node_id=node.id, + ) resolved = source_resolver(node, source_limit) - sources[node.id] = BackendSource( + if include_subflows: + try: + # Check before constructing another columnar copy. The provider + # still owns bounds on its individual authorized read; do not keep + # accumulating individually valid batches before the worker gate. + encoded_size = len(encode_worker_payload( + tuple(dict(row) for row in resolved.rows), max_bytes=_remaining_source_bytes[0], + )) + except WorkerPayloadError as exc: + raise PipelineExecutionError( + "Combined source data exceeds the 32 MiB transfer budget or contains unsupported values.", + node_id=node.id, + ) from exc + _remaining_source_bytes[0] -= encoded_size + key = reference_source_key(node) if include_subflows else node.id + sources[key] = BackendSource( node_id=node.id, batch=TypedBatch.from_rows(resolved.rows), source_ref=resolved.source_ref, diff --git a/tests/test_reference_process.py b/tests/test_reference_process.py new file mode 100644 index 0000000..c2e44a6 --- /dev/null +++ b/tests/test_reference_process.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +from datetime import date, datetime, timezone +from decimal import Decimal +import time +import unittest +from unittest.mock import patch + +from govoplan_core.security.bounded_process import ProcessBudgetError +from govoplan_dataflow.backend.backends import ( + BackendExecutionError, BackendSource, ExecutionBudget, execute_typed_graph, +) +from govoplan_dataflow.backend.backends.reference import ReferenceExecutionBackend +from govoplan_dataflow.backend.batches import TypedBatch +from govoplan_dataflow.backend.executor import PipelineExecutionError, ResolvedSource +from govoplan_dataflow.backend.manifest import get_manifest +from govoplan_dataflow.backend.schemas import GraphEdge, GraphNode, GraphPosition, PipelineGraph +from govoplan_dataflow.backend.service import _execute_pipeline_preview, _typed_backend_sources + + +def graph_for(expression: str = "value", rows: list | None = None) -> PipelineGraph: + nodes = [ + GraphNode(id="source", type="source.inline", label="Source", position=GraphPosition(x=0, y=0), + config={"source_name": "records", "rows": rows or [{"value": "normal"}]}), + GraphNode(id="expression", type="expression", label="Expression", position=GraphPosition(x=100, y=0), + config={"target_column": "result", "expression": expression, "result_type": "unknown"}), + GraphNode(id="output", type="output", label="Output", position=GraphPosition(x=200, y=0), config={}), + ] + return PipelineGraph(nodes=nodes, edges=[ + GraphEdge(id="first", source="source", target="expression"), + GraphEdge(id="second", source="expression", target="output"), + ]) + + +class ReferenceProcessTests(unittest.TestCase): + def test_non_finite_deadlines_are_rejected_before_registry_wait(self) -> None: + for value in (float("nan"), float("inf")): + with self.subTest(value=value), self.assertRaisesRegex(ValueError, "finite"): + ExecutionBudget(max_wall_seconds=value) + + def test_real_child_not_parent_helper_and_typed_result_preserved(self) -> None: + row = {"value": Decimal("1.20"), "date": date(2026, 9, 8), + "when": datetime(2026, 9, 8, tzinfo=timezone.utc), "binary": b"\x00\xff"} + graph = graph_for() + graph.nodes[0] = graph.nodes[0].model_copy(update={"type": "source.reference", "config": { + "source_ref": "datasource:fixture", "source_name": "records", + }}) + source = BackendSource(node_id="source", batch=TypedBatch.from_rows([row]), + source_ref="datasource:fixture", provider="test", fingerprint="pinned", total_rows=1) + with patch.object(ReferenceExecutionBackend, "_execute_in_process", side_effect=AssertionError("parent evaluation")): + result = execute_typed_graph(graph, backend="reference", sources={"source": source}) + self.assertEqual(result.rows, [{**row, "result": Decimal("1.20")}]) + self.assertEqual(result.contract.lineage.source_fingerprints[0]["fingerprint"], "pinned") + + def test_real_pathological_regex_is_stopped_with_structured_failure(self) -> None: + graph = graph_for("regexp_full_match(value, '(a+)+$')", [{"value": "a" * 100 + "!"}]) + started = time.monotonic() + with self.assertRaises(BackendExecutionError) as caught: + execute_typed_graph(graph, backend="reference", budget=ExecutionBudget(max_wall_seconds=2)) + self.assertIn(caught.exception.code, {"backend.process.timeout", "backend.process.cpu_limit"}) + self.assertLess(time.monotonic() - started, 4) + + def test_aggregate_padding_allocation_is_contained_by_child_memory_limit(self) -> None: + graph = graph_for("lpad(value, 900000, '0')", [{"value": "x"} for _ in range(250)]) + with self.assertRaises(BackendExecutionError) as caught: + execute_typed_graph(graph, backend="reference", budget=ExecutionBudget( + max_wall_seconds=5, max_memory_bytes=128 * 1024 * 1024, + )) + self.assertEqual(caught.exception.code, "backend.process.memory_limit") + + def test_busy_preview_is_retryable_and_never_evaluates_inline(self) -> None: + with patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation", side_effect=ProcessBudgetError("busy")): + with self.assertRaises(PipelineExecutionError) as caught: + _execute_pipeline_preview(graph_for(), session=None, principal=None, registry=None, + backend="reference", row_limit=10, preview_node_id=None) + self.assertTrue(caught.exception.retryable) + self.assertEqual(caught.exception.diagnostics[-1].code, "backend.process.busy") + + def test_nested_source_ids_do_not_alias_different_authorized_data(self) -> None: + outer = graph_for() + inner = graph_for() + for graph, ref in ((outer, "datasource:outer"), (inner, "datasource:inner")): + graph.nodes[0] = graph.nodes[0].model_copy(update={"type": "source.reference", "config": { + "source_ref": ref, "source_name": "records", + }}) + outer.nodes[1] = outer.nodes[1].model_copy(update={"type": "subflow", "config": { + "graph": inner.model_dump(mode="python"), "parameters": {}, + }}) + def resolve(node, limit): + return ResolvedSource(rows=({"value": node.config["source_ref"]},), + source_ref=node.config["source_ref"], provider="test", + fingerprint=node.config["source_ref"], total_rows=1) + sources = _typed_backend_sources(outer, source_resolver=resolve, include_subflows=True) + self.assertEqual({source.source_ref for source in sources.values()}, {"datasource:outer", "datasource:inner"}) + self.assertEqual(len(sources), 2) + + def test_static_worker_documentation_is_bilingual(self) -> None: + topic = next(item for item in get_manifest().documentation if item.id == "dataflow.reference-worker-limits") + for body in (topic.body, topic.translations["de"]["body"]): + self.assertIn("GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY", body) + self.assertIn("32 MiB", body) + self.assertEqual(set(topic.documentation_types), {"user", "admin"}) + + def test_cumulative_source_budget_stops_before_reading_further_sources(self) -> None: + graph = graph_for() + graph.nodes = [graph.nodes[0].model_copy(update={ + "id": f"source-{index}", "type": "source.reference", "config": {"source_ref": f"fixture:{index}"}, + }) for index in range(3)] + calls = [] + def resolve(node, limit): + calls.append(node.id) + return ResolvedSource(rows=({"value": "x" * 100},), source_ref=node.config["source_ref"], + provider="test", fingerprint="fixed", total_rows=1) + with self.assertRaisesRegex(PipelineExecutionError, "Combined source data"): + _typed_backend_sources(graph, source_resolver=resolve, include_subflows=True, + _remaining_source_bytes=[200]) + self.assertEqual(calls, ["source-0", "source-1"]) diff --git a/tests/test_reference_subflow_process.py b/tests/test_reference_subflow_process.py new file mode 100644 index 0000000..488f4fd --- /dev/null +++ b/tests/test_reference_subflow_process.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import hashlib +import json +import os +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_CATALOGUE, + DatasourceAccessError, + DatasourceCatalogueProvider, + DatasourceDescriptor, + DatasourceReadRequest, + DatasourceReadResult, +) +from govoplan_core.security.bounded_process import run_bounded_operation +from govoplan_dataflow.backend.backends.reference import ReferenceExecutionBackend +from govoplan_dataflow.backend.executor import EXECUTOR_VERSION, PipelineExecutionError +from govoplan_dataflow.backend.graph import validate_graph +from govoplan_dataflow.backend.schemas import GraphEdge, GraphNode, GraphPosition, PipelineGraph +from govoplan_dataflow.backend.service import _execute_pipeline_preview + + +OUTER_ROWS = ({"id": "outer", "amount": 15}, {"id": "outer-low", "amount": 5}) +INNER_ROWS = ({"id": "inner", "amount": 25}, {"id": "inner-low", "amount": 2}) + + +def node(node_id: str, node_type: str, config: dict) -> GraphNode: + return GraphNode( + id=node_id, type=node_type, label=node_id, + position=GraphPosition(x=0, y=0), config=config, + ) + + +def nested_source_graph() -> PipelineGraph: + # Both external sources deliberately share their node ID and logical name. + # The pinned subflow still has exactly one distinct inline input binding. + nested = PipelineGraph( + nodes=[ + node("input", "source.inline", { + "source_name": "bound_input", "rows": [], "input_binding": True, + }), + node("shared", "source.reference", { + "source_name": "records", "source_ref": {"$parameter": "source_ref"}, + "expected_fingerprint": "inner-pinned", "consistency": "frozen", + }), + node("union", "combine.union", {"mode": "all"}), + node("minimum", "filter.expression", {"expression": "amount >= ${minimum}"}), + node("output", "output", {}), + ], + edges=[ + GraphEdge(id="input-union", source="input", target="union"), + GraphEdge(id="shared-union", source="shared", target="union"), + GraphEdge(id="union-minimum", source="union", target="minimum"), + GraphEdge(id="minimum-output", source="minimum", target="output"), + ], + ) + return PipelineGraph( + nodes=[ + node("shared", "source.reference", { + "source_name": "records", "source_ref": "datasource:outer", + "expected_fingerprint": "outer-pinned", "consistency": "current", + }), + node("nested", "subflow", { + "template_ref": "fixture-nested-source", "template_version": "1", + "parameters": {"source_ref": "datasource:inner", "minimum": 10}, + "graph": nested.model_dump(mode="python"), + }), + node("output", "output", {}), + ], + edges=[ + GraphEdge(id="shared-nested", source="shared", target="nested"), + GraphEdge(id="nested-output", source="nested", target="output"), + ], + ) + + +class ReferenceSubflowProcessTests(unittest.TestCase): + def setUp(self) -> None: + self.graph = nested_source_graph() + self.assertEqual([], [item.model_dump() for item in validate_graph(self.graph) if item.severity == "error"]) + self.session = Mock(spec=Session) + self.principal = ApiPrincipal( + principal=PrincipalRef( + account_id="fixture-account", membership_id="fixture-membership", + tenant_id="fixture-tenant", scopes=frozenset(), + ), + account=object(), user=object(), + ) + self.provider = Mock(spec=DatasourceCatalogueProvider) + self.provider.read_datasource.side_effect = self.read_source + self.registry = SimpleNamespace( + has_capability=lambda name: name == CAPABILITY_DATASOURCE_CATALOGUE, + capability=lambda name: self.provider if name == CAPABILITY_DATASOURCE_CATALOGUE else None, + ) + self.reads: list[tuple[int, DatasourceReadRequest]] = [] + self.denied = False + + def read_source(self, session, principal, *, request: DatasourceReadRequest) -> DatasourceReadResult: + self.assertIs(session, self.session) + self.assertIs(principal, self.principal) + self.reads.append((os.getpid(), request)) + if request.datasource_ref == "datasource:inner" and self.denied: + raise DatasourceAccessError("Current principal cannot read datasource:inner.") + rows, fingerprint = { + "datasource:outer": (OUTER_ROWS, "outer-pinned"), + "datasource:inner": (INNER_ROWS, "inner-pinned"), + }[request.datasource_ref] + self.assertEqual(fingerprint, request.expected_fingerprint) + return DatasourceReadResult( + datasource=DatasourceDescriptor( + ref=request.datasource_ref, source_name="records", name="Fixture source", + kind="custom", mode="static", shape="tabular", + fingerprint=fingerprint, provider="fixture-catalogue", + ), + rows=rows, total_rows=len(rows), truncated=False, + ) + + def preview(self, *, row_limit: int = 10): + return _execute_pipeline_preview( + self.graph, session=self.session, principal=self.principal, + registry=self.registry, backend="reference", row_limit=row_limit, + preview_node_id="nested", + ) + + def test_nested_parameterized_sources_execute_in_real_child_without_id_aliasing(self) -> None: + with ( + patch.object(ReferenceExecutionBackend, "_execute_in_process", side_effect=AssertionError("parent evaluation")), + patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation", wraps=run_bounded_operation) as worker, + ): + result, version = self.preview() + worker.assert_called_once() + self.assertEqual(EXECUTOR_VERSION, version) + self.assertEqual([OUTER_ROWS[0], INNER_ROWS[0]], result.rows) + self.assertEqual(2, result.total_rows) + self.assertFalse(result.truncated) + self.assertEqual(2, result.input_row_count) # Root input count, not the nested binding again. + self.assertIsNotNone(result.node_preview) + self.assertEqual("nested", result.node_preview.node_id) + self.assertEqual(result.rows, result.node_preview.rows) + self.assertEqual(2, result.node_preview.total_rows) + self.assertEqual( + [("shared", "succeeded", 0, 2), ("nested", "succeeded", 2, 2), ("output", "succeeded", 2, 2)], + [(item.node_id, item.status, item.input_rows, item.output_rows) for item in result.node_diagnostics], + ) + binding_hash = hashlib.sha256(json.dumps(list(OUTER_ROWS), sort_keys=True, separators=(",", ":")).encode()).hexdigest() + self.assertEqual([ + {"node_id": "shared", "source_ref": "datasource:outer", "source_name": "records", "kind": "datasource", "provider": "fixture-catalogue", "fingerprint": "outer-pinned", "row_count": 2, "preview_rows": 2, "truncated": False}, + {"node_id": "input", "source_name": "bound_input", "kind": "inline", "fingerprint": binding_hash, "row_count": 2, "subflow_node_id": "nested"}, + {"node_id": "shared", "source_ref": "datasource:inner", "source_name": "records", "kind": "datasource", "provider": "fixture-catalogue", "fingerprint": "inner-pinned", "row_count": 2, "preview_rows": 2, "truncated": False, "subflow_node_id": "nested"}, + ], result.source_fingerprints) + self.assertEqual( + [("datasource:outer", "current", "outer-pinned"), ("datasource:inner", "frozen", "inner-pinned")], + [(request.datasource_ref, request.consistency, request.expected_fingerprint) for _, request in self.reads], + ) + self.assertTrue(all(pid == os.getpid() for pid, _ in self.reads)) + self.assertTrue(all(request.limit <= 500 and request.offset == 0 for _, request in self.reads)) + self.assertEqual([], self.session.mock_calls) + + def test_nested_result_and_node_preview_keep_full_totals_when_output_is_bounded(self) -> None: + result, _ = self.preview(row_limit=1) + self.assertEqual([OUTER_ROWS[0]], result.rows) + self.assertEqual(2, result.total_rows) + self.assertTrue(result.truncated) + self.assertEqual([OUTER_ROWS[0]], result.node_preview.rows) + self.assertEqual(2, result.node_preview.total_rows) + self.assertTrue(result.node_preview.truncated) + self.assertEqual({"outer-pinned", "inner-pinned"}, { + item["fingerprint"] for item in result.source_fingerprints if item["kind"] == "datasource" + }) + self.assertEqual([], self.session.mock_calls) + + def test_denied_nested_datasource_stops_before_worker_or_persistence(self) -> None: + self.denied = True + with ( + patch("govoplan_dataflow.backend.service.execute_typed_graph") as execute, + patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation") as worker, + self.assertRaises(PipelineExecutionError) as caught, + ): + self.preview() + execute.assert_not_called() + worker.assert_not_called() + self.assertEqual("shared", caught.exception.node_id) + self.assertEqual("Current principal cannot read datasource:inner.", str(caught.exception)) + self.assertFalse(caught.exception.retryable) + self.assertIsInstance(caught.exception.__cause__, DatasourceAccessError) + self.assertEqual(["datasource:outer", "datasource:inner"], [request.datasource_ref for _, request in self.reads]) + self.assertEqual([], self.session.mock_calls)