From 40535438161e660e971a84ed163d131427930503 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 12:48:37 +0200 Subject: [PATCH] Serialize concurrent datasource publications --- README.md | 5 +- docs/QUALITY_POLICY.md | 6 ++ src/govoplan_datasources/backend/manifest.py | 2 +- src/govoplan_datasources/backend/service.py | 46 +++++++++++++- ...st_postgres_materialization_concurrency.py | 62 ++++++++++++++++++- 5 files changed, 117 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 38f822a..0182a34 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,10 @@ declared pushdown support. Live previews preserve the provider's effective row, serialized-byte, and elapsed-time limits and its redacted diagnostics. Producer modules can append a bounded tabular result or create a new static datasource through an idempotent capability. The publication ledger retains the -producer run, output materialization, provenance, and replay identity. +producer run, output materialization, provenance, and replay identity. On +PostgreSQL, a transaction-scoped advisory lock serializes each tenant, producer, +and idempotency identity before any output side effect, so retries from multiple +application or worker nodes resolve to one durable publication. Tabular staging also evaluates governed quality rules and classifies schema changes before promotion. Blocking stages remain inspectable, and successful diff --git a/docs/QUALITY_POLICY.md b/docs/QUALITY_POLICY.md index be55911..fb2a32f 100644 --- a/docs/QUALITY_POLICY.md +++ b/docs/QUALITY_POLICY.md @@ -90,6 +90,12 @@ changed quality policy. Publication also emits a transactional module stores it in the durable outbox, while reduced installations deliver it through Core after commit. +PostgreSQL deployments serialize publication attempts by tenant, producer, and +idempotency key with a transaction-scoped advisory lock before replay lookup. +This makes concurrent retries from separate API or worker nodes converge on the +same publication and materialization rather than relying on a late uniqueness +failure after output rows have already been persisted. + Approval authority, approval expiry, and retention/deletion execution remain separate work under `govoplan-datasources#2`. Until those contracts are added, no JSON flag is treated as an approval and no stage is deleted automatically. diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index 5e025b1..16d243a 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/src/govoplan_datasources/backend/manifest.py @@ -416,7 +416,7 @@ manifest = ModuleManifest( "promotion. Updates compare the detected schema with the current target and classify each change as compatible, warning, or " "breaking. Diagnostics expose counts and bounded row numbers, never field values. The policy version and hash, diagnostics, and " "schema diff are copied into immutable materialization provenance when promotion succeeds. Producer publication applies the same " - "gate before any catalogue effect, retains validation evidence on the immutable output revision, and emits a transactional terminal event. Approval and retention execution " + "gate before any catalogue effect, retains validation evidence on the immutable output revision, serializes a publication identity across PostgreSQL worker nodes before replay lookup, and emits a transactional terminal event. Approval and retention execution " "are not inferred from arbitrary JSON flags and remain separate governed lifecycle work." ), layer="available", diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index c47a0a3..ff9854c 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/src/govoplan_datasources/backend/service.py @@ -7,7 +7,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from typing import Any, cast -from sqlalchemy import exists, func, or_, select +from sqlalchemy import exists, func, or_, select, text from sqlalchemy.orm import Session from govoplan_core.audit.logging import audit_event @@ -98,6 +98,12 @@ class SqlDatasourceProvider: ) -> DatasourcePublicationResult: db, api_principal = _publication_context(session, principal) prepared = _prepare_publication(request) + _lock_publication_identity( + db, + tenant_id=api_principal.tenant_id, + producer_module=prepared.producer_module, + idempotency_key=prepared.idempotency_key, + ) existing = _existing_publication_result( db, tenant_id=api_principal.tenant_id, @@ -1649,6 +1655,44 @@ def _validate_publication_identity( ) +def _publication_lock_key( + *, + tenant_id: str, + producer_module: str, + idempotency_key: str, +) -> int: + payload = ( + f"govoplan.datasources.publication\0{tenant_id}\0" + f"{producer_module}\0{idempotency_key}" + ).encode("utf-8") + return int.from_bytes( + hashlib.sha256(payload).digest()[:8], + "big", + signed=True, + ) + + +def _lock_publication_identity( + session: Session, + *, + tenant_id: str, + producer_module: str, + idempotency_key: str, +) -> None: + if session.get_bind().dialect.name != "postgresql": + return + session.execute( + text("SELECT pg_advisory_xact_lock(:key)"), + { + "key": _publication_lock_key( + tenant_id=tenant_id, + producer_module=producer_module, + idempotency_key=idempotency_key, + ) + }, + ) + + def _existing_publication_result( session: Session, *, diff --git a/tests/test_postgres_materialization_concurrency.py b/tests/test_postgres_materialization_concurrency.py index d170455..69e6847 100644 --- a/tests/test_postgres_materialization_concurrency.py +++ b/tests/test_postgres_materialization_concurrency.py @@ -9,14 +9,23 @@ from concurrent.futures import ThreadPoolExecutor from sqlalchemy import create_engine, text from sqlalchemy.orm import Session +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.core.datasources import DatasourcePublicationRequest from govoplan_core.db.base import Base from govoplan_datasources.backend.db.models import ( DatasourceMaterializationRecord, DatasourcePayloadRecord, DatasourcePayloadRowRecord, + DatasourcePublicationRecord, DatasourceRecord, ) -from govoplan_datasources.backend.service import _append_materialization +from govoplan_datasources.backend.service import ( + SOURCE_WRITE_SCOPE, + SqlDatasourceProvider, + _append_materialization, +) from govoplan_datasources.backend.tabular import ( encoded_size, field_payload, @@ -45,6 +54,8 @@ class DatasourceMaterializationPostgresTests(unittest.TestCase): DatasourcePayloadRecord.__table__, DatasourcePayloadRowRecord.__table__, DatasourceMaterializationRecord.__table__, + DatasourcePublicationRecord.__table__, + ChangeSequenceEntry.__table__, ] Base.metadata.create_all(self.engine, tables=self.tables) with Session(self.engine) as session: @@ -117,6 +128,55 @@ class DatasourceMaterializationPostgresTests(unittest.TestCase): ), ) + def test_same_publication_identity_is_serialized_and_replayed(self) -> None: + barrier = threading.Barrier(2) + provider = SqlDatasourceProvider() + request = DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref="dataflow-run:concurrent", + idempotency_key="concurrent-publication", + name="Concurrent publication", + source_name="concurrent_publication", + rows=({"id": 1, "result": "match"},), + ) + + def publish(worker: int) -> tuple[str, str, bool]: + api_principal = ApiPrincipal( + principal=PrincipalRef( + account_id=f"worker-{worker}", + membership_id=f"membership-{worker}", + tenant_id="tenant-1", + scopes=frozenset((SOURCE_WRITE_SCOPE,)), + ), + account=object(), + user=object(), + ) + with Session(self.engine) as session: + barrier.wait(timeout=10) + result = provider.publish_rows( + session, + api_principal, + request=request, + ) + session.commit() + return result.ref, result.materialization.ref, result.replayed + + with ThreadPoolExecutor(max_workers=2) as executor: + results = tuple(executor.map(publish, (1, 2))) + + self.assertEqual(1, len({item[0] for item in results})) + self.assertEqual(1, len({item[1] for item in results})) + self.assertEqual([False, True], sorted(item[2] for item in results)) + with Session(self.engine) as session: + self.assertEqual( + 1, + session.query(DatasourcePublicationRecord).count(), + ) + self.assertEqual( + 1, + session.query(DatasourceMaterializationRecord).count(), + ) + if __name__ == "__main__": unittest.main()