From 017c924b49af4690a37731d19c6573eb4ec5a29d Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 20:09:17 +0000 Subject: [PATCH 1/2] feat(reporting-storage): add PostgreSQL stores for reporting subscriptions AdCP 3.3 reporting subscriptions need their principal records and engine state to survive restarts and agree across workers. PgReportingSubscriptionStateStore implements the SDK's ReportingSubscriptionStateStore and PgPrincipalRecordStore implements PrincipalRecordStore. Each keeps one JSONB document per subject in adcp_reporting_subscription_state or adcp_principal_records and replaces it with one conditional statement on its revision, so the compare-and-swap is atomic across workers. create_schema applies the idempotent DDL, which reporting_subscriptions_schema_sql also returns. Rows are never pruned; the documents keep every subscription version for as long as obligations cite it. The stores need adcp 10.0.0b3 or later. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../adcp-reporting-storage/pyproject.toml | 1 + .../src/adcp_reporting_storage/__init__.py | 12 ++ .../subscriptions/__init__.py | 15 ++ .../subscriptions/pg.py | 172 ++++++++++++++++++ .../subscriptions/reporting_subscriptions.sql | 18 ++ 5 files changed, 218 insertions(+) create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql diff --git a/packages/adcp-reporting-storage/pyproject.toml b/packages/adcp-reporting-storage/pyproject.toml index 2fc517009..b1fac8367 100644 --- a/packages/adcp-reporting-storage/pyproject.toml +++ b/packages/adcp-reporting-storage/pyproject.toml @@ -84,6 +84,7 @@ adcp_reporting_storage = [ "projection/*.json", "production/*.json", "production/*.sql", + "subscriptions/*.sql", ] # Inside this repository the dependency is the checkout, not PyPI. uv ignores diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py index 8979c0afc..e8b9f72bf 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py @@ -205,6 +205,12 @@ from adcp_reporting_storage.submissions.pg import ( PgReportingSubmissionIntentStore as PgReportingSubmissionIntentStore, ) + from adcp_reporting_storage.subscriptions.pg import ( + PgPrincipalRecordStore as PgPrincipalRecordStore, + ) + from adcp_reporting_storage.subscriptions.pg import ( + PgReportingSubscriptionStateStore as PgReportingSubscriptionStateStore, + ) #: Public name -> (module under this package, attribute). Resolved on first use. _LAZY_EXPORTS: dict[str, tuple[str, str]] = { @@ -214,6 +220,11 @@ "PgReportingMaterializerStore": ("materializer.pg", "PgReportingMaterializerStore"), "PgReportingReceiptStore": ("receipts.pg", "PgReportingReceiptStore"), "PgReportingFeedStore": ("feed.pg", "PgReportingFeedStore"), + "PgPrincipalRecordStore": ("subscriptions.pg", "PgPrincipalRecordStore"), + "PgReportingSubscriptionStateStore": ( + "subscriptions.pg", + "PgReportingSubscriptionStateStore", + ), "PgReportingProjectionStore": ("projection.pg", "PgReportingProjectionStore"), "PgReportingStatusProjection": ("projection.pg", "PgReportingStatusProjection"), "PgReportingProjectionOutbox": ("projection.notifications", "PgReportingProjectionOutbox"), @@ -320,6 +331,7 @@ "receipts", "service", "submissions", + "subscriptions", } ) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py new file mode 100644 index 000000000..20ed35401 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/__init__.py @@ -0,0 +1,15 @@ +"""PostgreSQL stores for AdCP 3.3 reporting subscriptions.""" + +from __future__ import annotations + +from adcp_reporting_storage.subscriptions.pg import ( + PgPrincipalRecordStore, + PgReportingSubscriptionStateStore, + reporting_subscriptions_schema_sql, +) + +__all__ = [ + "PgPrincipalRecordStore", + "PgReportingSubscriptionStateStore", + "reporting_subscriptions_schema_sql", +] diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py new file mode 100644 index 000000000..efcf23f49 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py @@ -0,0 +1,172 @@ +"""PostgreSQL stores for AdCP 3.3 reporting subscriptions (experimental). + +:class:`PgReportingSubscriptionStateStore` implements +``adcp.reporting.subscriptions.ReportingSubscriptionStateStore`` and +:class:`PgPrincipalRecordStore` implements +``adcp.server.PrincipalRecordStore``. Together they make a seller's +``sync_principal`` subscriptions and the engine's grants, exceptions and +version snapshots survive restarts and agree across workers:: + + from psycopg_pool import AsyncConnectionPool + from adcp.reporting.subscriptions import ReportingSubscriptionEngine + from adcp.server import PrincipalService + from adcp_reporting_storage import ( + PgPrincipalRecordStore, + PgReportingSubscriptionStateStore, + ) + + async with AsyncConnectionPool(dsn, min_size=2, max_size=10) as pool: + records = PgPrincipalRecordStore(pool=pool) + state = PgReportingSubscriptionStateStore(pool=pool) + await state.create_schema() # creates both tables + principals = PrincipalService(store=records, emit_change=...) + engine = ReportingSubscriptionEngine( + principals=principals, host=..., offerings=..., store=state + ) + +Each store keeps one JSONB document per subject and replaces it with a +single conditional statement on its revision, so the compare-and-swap is +atomic across every worker sharing the database. The pool is borrowed: the +stores never open, own or close it. Requires ``adcp`` 10.0.0b3 or later. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from importlib.resources import files +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + + from adcp.reporting.subscriptions import StoredReportingSubscriptionState + from adcp.server import PrincipalRecord + +__all__ = [ + "PgPrincipalRecordStore", + "PgReportingSubscriptionStateStore", + "reporting_subscriptions_schema_sql", +] + +_REQUIRES = "adcp_reporting_storage.subscriptions requires adcp 10.0.0b3 or later" + + +def reporting_subscriptions_schema_sql() -> str: + """The idempotent DDL for both reporting-subscription tables.""" + return ( + files("adcp_reporting_storage.subscriptions") + .joinpath("reporting_subscriptions.sql") + .read_text() + ) + + +def _document(value: Any) -> dict[str, Any]: + # psycopg decodes JSONB to Python by default; a custom loader may not. + decoded = json.loads(value) if isinstance(value, (str, bytes, bytearray)) else value + if not isinstance(decoded, dict): + raise ValueError("stored reporting-subscription document is not a JSON object") + return decoded + + +class _PgDocumentStore: + def __init__(self, *, pool: AsyncConnectionPool) -> None: + self._pool = pool + + async def create_schema(self) -> None: + """Create both reporting-subscription tables if they do not exist.""" + async with self._pool.connection() as connection: + await connection.execute(reporting_subscriptions_schema_sql()) + + +class PgReportingSubscriptionStateStore(_PgDocumentStore): + """Engine state for reporting subscriptions, one CAS document per principal.""" + + async def load(self, subject: str) -> StoredReportingSubscriptionState | None: + try: + from adcp.reporting.subscriptions import StoredReportingSubscriptionState + except ImportError as error: # pragma: no cover - older adcp + raise ImportError(_REQUIRES) from error + async with self._pool.connection() as connection: + cursor = await connection.execute( + "SELECT document, revision FROM adcp_reporting_subscription_state" + " WHERE subject = %s", + (subject,), + ) + row = await cursor.fetchone() + if row is None: + return None + return StoredReportingSubscriptionState(_document(row[0]), int(row[1])) + + async def compare_and_swap( + self, + subject: str, + expected_revision: int | None, + document: Mapping[str, Any], + ) -> bool: + payload = json.dumps(document, separators=(",", ":"), sort_keys=True) + async with self._pool.connection() as connection: + if expected_revision is None: + cursor = await connection.execute( + "INSERT INTO adcp_reporting_subscription_state (subject, revision, document)" + " VALUES (%s, 1, %s::jsonb) ON CONFLICT (subject) DO NOTHING", + (subject, payload), + ) + else: + cursor = await connection.execute( + "UPDATE adcp_reporting_subscription_state" + " SET revision = revision + 1, document = %s::jsonb, updated_at = now()" + " WHERE subject = %s AND revision = %s", + (payload, subject, expected_revision), + ) + return bool(cursor.rowcount == 1) + + +class PgPrincipalRecordStore(_PgDocumentStore): + """``sync_principal`` records, including reporting subscriptions and their versions.""" + + async def get(self, subject: str) -> PrincipalRecord | None: + try: + from adcp.server import principal_record_from_document + except ImportError as error: # pragma: no cover - older adcp + raise ImportError(_REQUIRES) from error + async with self._pool.connection() as connection: + cursor = await connection.execute( + "SELECT document, store_revision FROM adcp_principal_records WHERE subject = %s", + (subject,), + ) + row = await cursor.fetchone() + if row is None: + return None + return principal_record_from_document(_document(row[0]), store_revision=int(row[1])) + + async def compare_and_swap( + self, + subject: str, + expected_store_revision: int | None, + record: PrincipalRecord, + ) -> bool: + try: + from adcp.server import principal_record_to_document + except ImportError as error: # pragma: no cover - older adcp + raise ImportError(_REQUIRES) from error + if record.subject != subject: + raise ValueError("a principal record is stored under its own subject") + payload = json.dumps( + principal_record_to_document(record), separators=(",", ":"), sort_keys=True + ) + async with self._pool.connection() as connection: + if expected_store_revision is None: + cursor = await connection.execute( + "INSERT INTO adcp_principal_records (subject, store_revision, document)" + " VALUES (%s, %s, %s::jsonb) ON CONFLICT (subject) DO NOTHING", + (subject, record.store_revision, payload), + ) + else: + cursor = await connection.execute( + "UPDATE adcp_principal_records" + " SET store_revision = %s, document = %s::jsonb, updated_at = now()" + " WHERE subject = %s AND store_revision = %s", + (record.store_revision, payload, subject, expected_store_revision), + ) + return bool(cursor.rowcount == 1) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql new file mode 100644 index 000000000..074f5dd02 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/reporting_subscriptions.sql @@ -0,0 +1,18 @@ +-- AdCP 3.3 reporting subscriptions: durable principal records and engine state. +-- Each row is one JSON document per subject, replaced by compare-and-swap on +-- its revision. Rows are never pruned here: the documents keep every +-- subscription version for as long as the obligations that cite it. + +CREATE TABLE IF NOT EXISTS adcp_principal_records ( + subject TEXT PRIMARY KEY, + store_revision BIGINT NOT NULL CHECK (store_revision >= 1), + document JSONB NOT NULL CHECK (jsonb_typeof(document) = 'object'), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS adcp_reporting_subscription_state ( + subject TEXT PRIMARY KEY, + revision BIGINT NOT NULL CHECK (revision >= 1), + document JSONB NOT NULL CHECK (jsonb_typeof(document) = 'object'), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); From 6f55d8028a72c6cc4e0a865d7ad236fb460f039b Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 03:50:08 +0000 Subject: [PATCH 2/2] test(reporting-storage): add the subscription stores to the public API snapshot Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/fixtures/public_api_snapshot.json | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 1dabac09d..ca91211e1 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -8941,6 +8941,7 @@ }, "adcp_reporting_storage": { "resolves": { + "PgPrincipalRecordStore": "adcp_reporting_storage.subscriptions.pg", "PgReportingActivityUnionStore": "adcp_reporting_storage.outbox.status_activity_pg", "PgReportingFeedStore": "adcp_reporting_storage.feed.pg", "PgReportingLedgerStore": "adcp_reporting_storage.ledger.pg", @@ -8957,6 +8958,7 @@ "PgReportingStatusOutbox": "adcp_reporting_storage.outbox.status_pg", "PgReportingStatusProjection": "adcp_reporting_storage.projection.pg", "PgReportingSubmissionIntentStore": "adcp_reporting_storage.submissions.pg", + "PgReportingSubscriptionStateStore": "adcp_reporting_storage.subscriptions.pg", "PgStatusNotificationStore": "adcp_reporting_storage.outbox.status_pg", "ReportingChangeCursorExpiredError": "adcp_reporting_storage.ledger.change_feed", "ReportingChangeKindV1": "union[value:'adjustment',value:'obligation',value:'retirement',value:'revision']",