diff --git a/README.md b/README.md index bce3fc9..a42a626 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,14 @@ logical row disappeared. It never silently applies a correction to business data; a downstream governed transform or Workflow handoff must interpret the recorded action. +Reporting consumers may either evaluate a pinned pipeline revision or pin one +successful published run. An exact run pin is immutable: it cannot be supplied +new parameters, and Dataflow reads only the recorded Datasource materialization +through the provider-neutral catalogue capability. Both Dataflow run authority +and Datasource row access are rechecked for the current principal; the returned +lineage retains the run, publication, datasource, materialization, fingerprint, +and governance snapshot. + ## Development ```bash diff --git a/src/govoplan_dataflow/backend/dataset_output.py b/src/govoplan_dataflow/backend/dataset_output.py index d68e26b..6f48bd9 100644 --- a/src/govoplan_dataflow/backend/dataset_output.py +++ b/src/govoplan_dataflow/backend/dataset_output.py @@ -11,6 +11,11 @@ from govoplan_core.core.dataflows import ( DataflowRunConflictError, DataflowRunUnavailableError, ) +from govoplan_core.core.datasources import ( + DatasourceError, + DatasourceReadRequest, + datasource_catalogue, +) from govoplan_core.security.time import utc_now from govoplan_dataflow.backend.backends.base import ExecutionBudget from govoplan_dataflow.backend.executor import PipelineExecutionError @@ -20,6 +25,7 @@ from govoplan_dataflow.backend.service import ( _execute_pipeline_preview, get_pipeline, get_pipeline_revision, + get_pipeline_run, list_pipelines, ) from govoplan_dataflow.backend.subflows import substitute_parameters @@ -110,6 +116,17 @@ class SqlDataflowDatasetOutputProvider: raise DataflowRunConflictError( "The pinned Dataflow definition hash no longer matches the requested revision." ) + if request.run_ref: + return _read_published_run_output( + typed_session, + typed_principal, + registry=self.registry, + request=request, + pipeline=pipeline, + revision=revision, + policy_decision=decision.to_dict(), + row_limit=row_limit, + ) graph = PipelineGraph.model_validate( substitute_parameters(revision.graph, dict(request.parameters)) ) @@ -179,6 +196,143 @@ def dataset_output_provider(context: object | None = None): return SqlDataflowDatasetOutputProvider(getattr(context, "registry", None)) +def _read_published_run_output( + session, + principal: ApiPrincipal, + *, + registry: object | None, + request: DataflowDatasetRequest, + pipeline, + revision, + policy_decision: dict[str, object], + row_limit: int, +) -> DataflowDatasetResult: + if request.parameters: + raise DataflowRunConflictError( + "An immutable published run cannot be evaluated with new parameters." + ) + run = get_pipeline_run( + session, + tenant_id=principal.tenant_id, + run_ref=request.run_ref or "", + ) + if run.pipeline_id != pipeline.id or run.pipeline_revision_id != revision.id: + raise DataflowRunConflictError( + "The published run does not belong to the pinned Dataflow revision." + ) + if run.definition_hash != revision.content_hash: + raise DataflowRunConflictError( + "The published run definition evidence does not match the pinned revision." + ) + if run.status != "succeeded": + raise DataflowRunUnavailableError( + "Only a successful Dataflow run can be used as an immutable dataset." + ) + if not run.output_datasource_ref or not run.output_materialization_ref: + raise DataflowRunUnavailableError( + "The successful Dataflow run has no immutable Datasource publication." + ) + provider = datasource_catalogue(registry) + if provider is None: + raise DataflowRunUnavailableError( + "The Datasource catalogue required by this published run is not enabled." + ) + + rows: list[dict[str, object]] = [] + materialization = None + total_rows = 0 + try: + while len(rows) < row_limit: + remaining = row_limit - len(rows) + page = provider.read_datasource( + session, + principal, + request=DatasourceReadRequest( + datasource_ref=run.output_datasource_ref, + materialization_ref=run.output_materialization_ref, + limit=min(500, remaining), + offset=len(rows), + expected_fingerprint=( + materialization.fingerprint + if materialization is not None + else None + ), + ), + ) + if ( + page.materialization is None + or page.materialization.ref != run.output_materialization_ref + ): + raise DataflowRunConflictError( + "The Datasource provider returned a different output materialization." + ) + if len(page.rows) > remaining: + raise DataflowRunConflictError( + "The Datasource provider exceeded the requested output window." + ) + materialization = page.materialization + total_rows = page.total_rows + rows.extend(dict(item) for item in page.rows) + if not page.truncated: + break + if not page.rows: + raise DataflowRunUnavailableError( + "The Datasource provider made no progress while reading the published output." + ) + except DatasourceError as exc: + raise DataflowRunUnavailableError( + "The immutable Datasource output is unavailable to the current principal." + ) from exc + + source_fingerprints = tuple(dict(item) for item in run.source_fingerprints) + if request.expected_source_fingerprints and not _fingerprints_match( + request.expected_source_fingerprints, + source_fingerprints, + ): + raise DataflowRunConflictError( + "Dataflow source fingerprints differ from the pinned run evidence." + ) + output_hash = hashlib.sha256( + json.dumps( + rows, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + default=str, + ).encode("utf-8") + ).hexdigest() + if materialization is None: + raise DataflowRunUnavailableError( + "The Datasource provider returned no materialization evidence." + ) + return DataflowDatasetResult( + pipeline_ref=pipeline.id, + revision=revision.revision, + definition_hash=revision.content_hash, + rows=tuple(rows), + total_rows=total_rows, + truncated=len(rows) < total_rows, + output_hash=output_hash, + executor_version=run.executor_version, + run_ref=f"dataflow-run:{run.id}", + source_fingerprints=source_fingerprints, + diagnostics=tuple(dict(item) for item in run.diagnostics), + generated_at=run.finished_at or materialization.created_at, + provenance={ + "module": "dataflow", + "scope_type": pipeline.scope_type, + "scope_id": pipeline.scope_id, + "policy_decision": policy_decision, + "immutable_run": True, + "publication_ref": run.output_publication_ref, + "datasource_ref": run.output_datasource_ref, + "materialization_ref": run.output_materialization_ref, + "materialization_fingerprint": materialization.fingerprint, + "governance": materialization.governance.to_dict(), + }, + ) + + def _fingerprints_match(expected, actual) -> bool: def normalized(values): return sorted( diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 34cbe97..695ea8e 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -270,6 +270,7 @@ DOCUMENTATION = ( "states whether work stopped, completed, failed, or requires operator reconciliation. Scheduled, event, " "and queued execution is partitioned by tenant module entitlement before a run is claimed. Disabling " "Dataflow stops new admission and leaves accepted runs available for an explicit operator decision." + " Reporting may pin a successful published run; Dataflow then rechecks run authority and Datasource access and reads only the exact recorded materialization without reparameterizing or re-executing it." ), layer="available", documentation_types=("admin", "user"), diff --git a/tests/test_dataset_output.py b/tests/test_dataset_output.py index 3e2f5e2..9f5a1ee 100644 --- a/tests/test_dataset_output.py +++ b/tests/test_dataset_output.py @@ -1,14 +1,25 @@ from __future__ import annotations +from datetime import UTC, datetime import unittest from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.dataflows import DataflowDatasetRequest, DataflowRunConflictError +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_CATALOGUE, + DatasourceDescriptor, + DatasourceMaterialization, + DatasourceReadResult, +) from govoplan_core.db.base import Base from govoplan_core.db.session import configure_database, reset_database from govoplan_dataflow.backend.dataset_output import SqlDataflowDatasetOutputProvider -from govoplan_dataflow.backend.db.models import DataflowPipeline, DataflowPipelineRevision +from govoplan_dataflow.backend.db.models import ( + DataflowPipeline, + DataflowPipelineRevision, + DataflowRun, +) from govoplan_dataflow.backend.schemas import PipelineGraph from govoplan_dataflow.backend.service import definition_hash @@ -26,12 +37,71 @@ def principal(tenant_id: str = "tenant-1") -> ApiPrincipal: ) +class PublishedOutputCatalogue: + rows = ( + {"recipient_key": "published", "email": "published@example.test"}, + ) + + def list_datasources(self, *_args, **_kwargs): + return () + + def get_datasource(self, *_args, **_kwargs): + return None + + def list_materializations(self, *_args, **_kwargs): + return () + + def read_datasource(self, _session, _principal, *, request): + materialization = DatasourceMaterialization( + ref="materialization:published-1", + datasource_ref="datasource:published-1", + revision=1, + state="published", + fingerprint="f" * 64, + row_count=len(self.rows), + created_at=datetime(2026, 8, 4, 9, 0, tzinfo=UTC), + ) + rows = self.rows[request.offset : request.offset + request.limit] + return DatasourceReadResult( + datasource=DatasourceDescriptor( + ref="datasource:published-1", + source_name="published_output", + name="Published output", + kind="custom", + mode="static", + shape="tabular", + fingerprint=materialization.fingerprint, + current_materialization_ref=materialization.ref, + row_count=len(self.rows), + ), + materialization=materialization, + rows=rows, + total_rows=len(self.rows), + truncated=request.offset + len(rows) < len(self.rows), + ) + + +class CapabilityRegistry: + def __init__(self, catalogue) -> None: + self.catalogue = catalogue + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_DATASOURCE_CATALOGUE + + def capability(self, name: str): + return self.catalogue if self.has_capability(name) else None + + class DataflowDatasetOutputTests(unittest.TestCase): def setUp(self) -> None: self.database = configure_database("sqlite:///:memory:") Base.metadata.create_all( self.database.engine, - tables=[DataflowPipeline.__table__, DataflowPipelineRevision.__table__], + tables=[ + DataflowPipeline.__table__, + DataflowPipelineRevision.__table__, + DataflowRun.__table__, + ], ) def tearDown(self) -> None: @@ -131,6 +201,67 @@ class DataflowDatasetOutputTests(unittest.TestCase): expected_definition_hash="wrong", ), ) + + run = DataflowRun( + id="run-published-1", + tenant_id="tenant-1", + pipeline_id=pipeline.id, + pipeline_revision_id=revision.id, + run_type="published", + status="succeeded", + execution_backend="duckdb", + environment="production", + executor_version="duckdb-v1", + definition_hash=content_hash, + request_={}, + source_fingerprints=[ + {"node_id": "source", "fingerprint": "source-v1"} + ], + result_schema=[], + diagnostics=[], + output_row_count=1, + output_publication_ref="publication:published-1", + output_datasource_ref="datasource:published-1", + output_materialization_ref="materialization:published-1", + finished_at=datetime(2026, 8, 4, 9, 0, tzinfo=UTC), + ) + session.add(run) + session.flush() + published_provider = SqlDataflowDatasetOutputProvider( + CapabilityRegistry(PublishedOutputCatalogue()) + ) + published = published_provider.read_output( + session, + principal(), + request=DataflowDatasetRequest( + pipeline_ref=pipeline.id, + revision=1, + run_ref="dataflow-run:run-published-1", + expected_definition_hash=content_hash, + expected_source_fingerprints=( + {"node_id": "source", "fingerprint": "source-v1"}, + ), + ), + ) + self.assertEqual("published", published.rows[0]["recipient_key"]) + self.assertEqual("dataflow-run:run-published-1", published.run_ref) + self.assertTrue(published.provenance["immutable_run"]) + self.assertEqual( + "materialization:published-1", + published.provenance["materialization_ref"], + ) + + with self.assertRaises(DataflowRunConflictError): + published_provider.read_output( + session, + principal(), + request=DataflowDatasetRequest( + pipeline_ref=pipeline.id, + revision=1, + run_ref="dataflow-run:run-published-1", + parameters={"changed": True}, + ), + ) with self.assertRaises(DataflowRunConflictError): provider.read_output( session,