From 99fe10b2c4e7e434040ac53b1fc7fcf3fd8ce780 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 19:46:04 +0000 Subject: [PATCH 1/2] feat(reporting-storage): one storage decision with versioned migrations ReliableReportingService(storage=...) takes one storage object (DX plan slice P7). memory_storage() and postgres_storage(pool= | dsn=) from adcp.storage each hold the reporting ledger, staged source objects and seals, sync_principal principal records and the reporting-subscription engine state; the managed subscription path uses its principal and state stores. Durability is a property of the storage, never a flag. postgres_storage ships in adcp-reporting-storage and folds in #1724's PostgreSQL principal and engine stores, which gain list_principals() and declare is_durable. - Reporting subscriptions on process-local storage are refused outside development with REPORTING_STORAGE_NOT_DURABLE; the message names postgres_storage and ADCP_ENV=development. pytest and ADCP_ENV=development or test allow it. Without storage=, every store the managed path uses must declare is_durable. - The PostgreSQL schema is managed by numbered migration files in adcp_reporting_storage.migrations, recorded in adcp_schema_migrations. Each file names its version, may include the package's idempotent DDL files, so every statement lives in one place, and may declare a stopped-worker upgrade. Each migration commits on its own under the reporting schema lock. - python -m adcp migrate [--dsn URL] [--status | --dry-run] applies them; --status flags stopped-worker upgrades. - Durable storage checks the schema at startup instead of changing it, and refuses one that is behind with REPORTING_SCHEMA_BEHIND, naming the command and the pending versions. - 0002 reporting_subscription_generations, #1729's widened generation key, carries the stopped-worker note in its header, in --status and in the startup refusal. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 18 ++ .../adcp-reporting-storage/pyproject.toml | 1 + .../migrations/0001_reporting_ledger.sql | 4 + ...002_reporting_subscription_generations.sql | 12 + .../0003_reporting_ledger_extensions.sql | 9 + .../0004_reporting_subscription_state.sql | 4 + .../0005_reporting_inline_storage.sql | 4 + .../migrations/__init__.py | 219 ++++++++++++++++++ .../src/adcp_reporting_storage/storage.py | 110 +++++++++ .../subscriptions/pg.py | 16 ++ src/adcp/__main__.py | 5 + src/adcp/reporting/service.py | 59 ++++- src/adcp/reporting/subscriptions/managed.py | 9 +- src/adcp/storage/__init__.py | 128 ++++++++++ src/adcp/storage/_cli.py | 65 ++++++ src/adcp/storage/postgres.py | 7 + .../test_reporting_storage_migrations_pg.py | 148 ++++++++++++ tests/test_reporting_storage_facade.py | 139 +++++++++++ 18 files changed, 952 insertions(+), 5 deletions(-) create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0001_reporting_ledger.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0002_reporting_subscription_generations.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0003_reporting_ledger_extensions.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0004_reporting_subscription_state.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0005_reporting_inline_storage.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py create mode 100644 src/adcp/storage/__init__.py create mode 100644 src/adcp/storage/_cli.py create mode 100644 src/adcp/storage/postgres.py create mode 100644 tests/conformance/reporting/test_reporting_storage_migrations_pg.py create mode 100644 tests/test_reporting_storage_facade.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 8deda3b1d..227d3a2dd 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -397,6 +397,24 @@ platform = reporting.install(MySeller()) # then serve the platform as usual (from `adcp.server`). It lists only `REPORTING_WIND_DOWN_TASKS` (`get_reporting_status`, `sync_reporting_receipts`, `sync_reporting_status`) and keeps `authorized_at`, so coverage continues. +- **Storage: one decision.** Pass `storage=memory_storage()` or + `storage=postgres_storage(dsn=...)` (or `pool=`) from `adcp.storage`. One + object holds the ledger, staged source objects and seals, principal records + and the subscription engine state; durability is a property of the storage, + never a flag. + - Process-local storage is refused for reporting subscriptions outside + development with `REPORTING_STORAGE_NOT_DURABLE`; pytest and + `ADCP_ENV=development` or `test` allow it. Without `storage=`, every store + the managed path uses must declare `is_durable = True` (the SDK's + PostgreSQL stores do). + - The PostgreSQL schema is managed by versioned migration files, applied in + order by `python -m adcp migrate [--dsn URL] [--status | --dry-run]` + (`--dsn` defaults to `$ADCP_DATABASE_URL`, then `$DATABASE_URL`). Startup + never changes the schema: it refuses one that is behind with + `REPORTING_SCHEMA_BEHIND`, naming the command and the pending versions. + - `0002 reporting_subscription_generations` widens the generation key and + is a stopped-worker upgrade: stop every reporting worker before running + it. `--status` flags it. - **Principals.** Reporting APIs name a principal with `PrincipalRef(tenant_id, principal_id)`; `grants_task` and `authorizes_delivery` take it. One service serves one tenant. diff --git a/packages/adcp-reporting-storage/pyproject.toml b/packages/adcp-reporting-storage/pyproject.toml index b1fac8367..60b4a6a60 100644 --- a/packages/adcp-reporting-storage/pyproject.toml +++ b/packages/adcp-reporting-storage/pyproject.toml @@ -85,6 +85,7 @@ adcp_reporting_storage = [ "production/*.json", "production/*.sql", "subscriptions/*.sql", + "migrations/*.sql", ] # Inside this repository the dependency is the checkout, not PyPI. uv ignores diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0001_reporting_ledger.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0001_reporting_ledger.sql new file mode 100644 index 000000000..1b633cfb9 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0001_reporting_ledger.sql @@ -0,0 +1,4 @@ +-- adcp-migration: 0001 reporting_ledger +-- The reporting ledger: configurations, obligations, revisions, adjustments, +-- consumer statuses and issues. Online-safe: every statement is idempotent. +-- include: ledger/reporting_ledger.sql diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0002_reporting_subscription_generations.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0002_reporting_subscription_generations.sql new file mode 100644 index 000000000..3abd330c6 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0002_reporting_subscription_generations.sql @@ -0,0 +1,12 @@ +-- adcp-migration: 0002 reporting_subscription_generations +-- adcp-upgrade: stopped-worker +-- Widens the owned configuration-generation key to +-- (account_id, consumer_id, delivery_config_id, delivery_config_version, +-- offering_id), so a reporting subscription keeps one generation per resolved +-- offering of a version (adcp-client-python#1729). Existing rows keep +-- offering_id '' and their uniqueness. +-- +-- This is a stopped-worker upgrade: stop every reporting worker before running +-- it, because workers of earlier releases refuse the widened schema. Start the +-- new release only after it commits. On a new schema it changes nothing. +-- include: ledger/reporting_ledger_account_generations.sql diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0003_reporting_ledger_extensions.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0003_reporting_ledger_extensions.sql new file mode 100644 index 000000000..0f680274e --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0003_reporting_ledger_extensions.sql @@ -0,0 +1,9 @@ +-- adcp-migration: 0003 reporting_ledger_extensions +-- Obligation currency, reconciliation, the notification outbox, webhook +-- activity, provisional observations and caller ownership. Online-safe. +-- include: ledger/reporting_ledger_obligation_currency.sql +-- include: ledger/reporting_ledger_reconciliation.sql +-- include: ledger/reporting_notification_outbox.sql +-- include: ledger/reporting_webhook_activity.sql +-- include: ledger/reporting_provisional_observations.sql +-- include: ledger/reporting_caller_ownership.sql diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0004_reporting_subscription_state.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0004_reporting_subscription_state.sql new file mode 100644 index 000000000..90ecab194 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0004_reporting_subscription_state.sql @@ -0,0 +1,4 @@ +-- adcp-migration: 0004 reporting_subscription_state +-- AdCP 3.3 reporting subscriptions: durable principal records and engine +-- state (grant journal, exceptions, version snapshots). Online-safe. +-- include: subscriptions/reporting_subscriptions.sql diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0005_reporting_inline_storage.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0005_reporting_inline_storage.sql new file mode 100644 index 000000000..4578f5637 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0005_reporting_inline_storage.sql @@ -0,0 +1,4 @@ +-- adcp-migration: 0005 reporting_inline_storage +-- Staged source objects and seals for adapters registered on the service. +-- Online-safe: the migration is additive. +-- include: ledger/reporting_inline_storage.sql diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py new file mode 100644 index 000000000..ab5250ca8 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py @@ -0,0 +1,219 @@ +"""Versioned migrations for the SDK's reporting storage (AdCP 3.3, experimental). + +Each migration is a numbered file in this package, ``NNNN_name.sql``, applied +in order and recorded in ``adcp_schema_migrations``. A file's header carries +its metadata: + +* ``-- adcp-migration: NNNN name`` names it; +* ``-- adcp-upgrade: stopped-worker`` marks an upgrade that workers of earlier + releases cannot run beside: stop every reporting worker first; +* ``-- include: `` applies one of the package's idempotent DDL files, in + order, so each statement lives in exactly one place. Any other SQL in the + file runs after its includes. + +Run them with ``python -m adcp migrate`` (``--status`` lists what is applied +and pending, ``--dry-run`` prints the plan). Storage built by +``postgres_storage`` refuses to start on a schema that is behind, naming the +command and the pending versions. + +Every migration runs in its own transaction under the reporting schema lock, +so concurrent runners serialize and a failure commits nothing. +""" + +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from importlib.resources import files +from typing import Any + +__all__ = [ + "MIGRATE_COMMAND", + "Migration", + "MigrationStatus", + "applied_migrations", + "migrate", + "migration_status", + "migrations", + "pending_migrations", +] + +#: The command that applies pending migrations. +MIGRATE_COMMAND = "python -m adcp migrate" + +_TABLE = """ +CREATE TABLE IF NOT EXISTS adcp_schema_migrations ( + version TEXT PRIMARY KEY, + name TEXT NOT NULL, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() +) +""" +_LOCK = ( + "SELECT pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema()))" +) +_FILE = re.compile(r"^(\d{4})_([a-z0-9_]+)\.sql$") + + +@dataclass(frozen=True) +class Migration: + """One numbered migration file.""" + + version: str + name: str + stopped_worker: bool + includes: tuple[str, ...] + body: str + description: str + + def statements(self) -> list[str]: + """The SQL this migration runs, in order.""" + root = files("adcp_reporting_storage") + result = [root.joinpath(path).read_text() for path in self.includes] + if self.body.strip(): + result.append(self.body) + return result + + +@dataclass(frozen=True) +class MigrationStatus: + """A migration and whether, and when, it was applied.""" + + migration: Migration + applied_at: datetime | None + + @property + def pending(self) -> bool: + return self.applied_at is None + + +def _parse(filename: str, text: str) -> Migration: + match = _FILE.match(filename) + if match is None: + raise ValueError(f"migration file {filename!r} is not named NNNN_name.sql") + version, name = match.groups() + includes: list[str] = [] + body: list[str] = [] + description: list[str] = [] + stopped_worker = False + declared: str | None = None + for line in text.splitlines(): + stripped = line.strip() + if stripped.startswith("-- adcp-migration:"): + declared = stripped.removeprefix("-- adcp-migration:").strip() + elif stripped.startswith("-- adcp-upgrade:"): + stopped_worker = stripped.removeprefix("-- adcp-upgrade:").strip() == "stopped-worker" + elif stripped.startswith("-- include:"): + includes.append(stripped.removeprefix("-- include:").strip()) + elif stripped.startswith("--"): + description.append(stripped.removeprefix("--").strip()) + else: + body.append(line) + if declared != f"{version} {name}": + raise ValueError(f"migration file {filename!r} must declare '{version} {name}'") + return Migration( + version=version, + name=name, + stopped_worker=stopped_worker, + includes=tuple(includes), + body="\n".join(body), + description=" ".join(item for item in description if item), + ) + + +def migrations() -> tuple[Migration, ...]: + """Every migration this release ships, in version order.""" + root = files(__name__) + found = [ + _parse(entry.name, entry.read_text()) + for entry in root.iterdir() + if entry.name.endswith(".sql") + ] + found.sort(key=lambda item: item.version) + versions = [item.version for item in found] + if len(set(versions)) != len(versions): + raise ValueError("two migration files share a version") + return tuple(found) + + +async def applied_migrations(connection: Any) -> dict[str, datetime]: + """``version`` -> ``applied_at`` for every recorded migration (empty when none).""" + cursor = await connection.execute("SELECT to_regclass('adcp_schema_migrations') IS NOT NULL") + row = await cursor.fetchone() + if not row or not row[0]: + return {} + cursor = await connection.execute("SELECT version, applied_at FROM adcp_schema_migrations") + return {str(version): applied_at for version, applied_at in await cursor.fetchall()} + + +async def migration_status(connection: Any) -> list[MigrationStatus]: + """Every shipped migration with its ``applied_at``, or ``None`` while pending. + + ``connection`` is a psycopg ``AsyncConnection`` or ``AsyncConnectionPool``. + """ + if hasattr(connection, "getconn"): # a pool + async with connection.connection() as borrowed: + return await migration_status(borrowed) + async with connection.transaction(): + applied = await applied_migrations(connection) + return [MigrationStatus(item, applied.get(item.version)) for item in migrations()] + + +async def pending_migrations(connection: Any) -> list[Migration]: + """The shipped migrations not yet applied, in order.""" + return [item.migration for item in await migration_status(connection) if item.pending] + + +async def migrate(pool: Any, *, dry_run: bool = False) -> list[Migration]: + """Apply every pending migration in order; return what was (or would be) applied. + + ``pool`` is a psycopg ``AsyncConnectionPool`` or ``AsyncConnection`` whose + ``search_path`` selects the schema. Each migration commits on its own, so a + failure leaves the earlier ones applied and the failed one pending. + """ + if hasattr(pool, "getconn"): # a pool + async with pool.connection() as connection: + return await _migrate(connection, dry_run=dry_run) + return await _migrate(pool, dry_run=dry_run) + + +async def _migrate(connection: Any, *, dry_run: bool) -> list[Migration]: + from adcp_reporting_storage.migration import require_owned_schema_or_empty + + pending = await pending_migrations(connection) + if dry_run or not pending: + return pending + applied: list[Migration] = [] + for migration in pending: + async with connection.transaction(): + await connection.execute(_LOCK) + await connection.execute(_TABLE) + if migration.version in await applied_migrations(connection): + continue # another runner applied it while this one waited + await require_owned_schema_or_empty(connection) + for statement in migration.statements(): + await connection.execute(statement) + await connection.execute( + "INSERT INTO adcp_schema_migrations (version, name) VALUES (%s, %s)", + (migration.version, migration.name), + ) + applied.append(migration) + return applied + + +def describe(statuses: Sequence[MigrationStatus]) -> list[str]: + """``--status`` lines: one per migration, flagging stopped-worker upgrades.""" + lines: list[str] = [] + for status in statuses: + item = status.migration + state = "pending" if status.pending else f"applied {status.applied_at:%Y-%m-%dT%H:%M:%SZ}" + note = ( + " [stopped-worker upgrade: stop every reporting worker before migrating]" + if item.stopped_worker and status.pending + else " [stopped-worker upgrade]" + if item.stopped_worker + else "" + ) + lines.append(f"{item.version} {item.name}: {state}{note}") + return lines diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py new file mode 100644 index 000000000..f5887cbf0 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py @@ -0,0 +1,110 @@ +"""``postgres_storage``: every reporting store over one PostgreSQL pool (experimental). + +The durable half of ``adcp.storage``. One object holds the reporting ledger, +staged source objects and seals, ``sync_principal`` principal records, and the +reporting-subscription engine state (grant journal, exceptions, version +snapshots):: + + from adcp.reporting import ReliableReportingService, SubscriptionOptions + from adcp.storage.postgres import postgres_storage + + storage = postgres_storage(dsn=os.environ["DATABASE_URL"]) # or pool=... + reporting = ReliableReportingService( + storage=storage, + account_context=resolve_account_context, + subscriptions=SubscriptionOptions(max_accounts=500), + ) + +The schema is managed by versioned migrations (:mod:`adcp_reporting_storage.migrations`): +run ``python -m adcp migrate`` before starting. Startup never changes the +schema; it refuses one that is behind with ``REPORTING_SCHEMA_BEHIND``, naming +the command and the pending versions. Durability is a property of the +storage, never a flag. +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import datetime +from typing import Any + +__all__ = ["postgres_storage"] + + +def postgres_storage( + *, + pool: Any = None, + dsn: str | None = None, + clock: Callable[[], datetime] | None = None, + max_size: int = 10, +) -> Any: + """Durable :class:`adcp.storage.ReportingStorage` over PostgreSQL. + + Pass a borrowed psycopg ``AsyncConnectionPool`` (``pool=``), whose + ``search_path`` selects the schema, or a ``dsn`` for a pool the storage + opens at startup and closes at shutdown. + """ + from adcp.reporting._errors import ReliableReportingConfigurationError + from adcp.storage import ReportingStorage + + if (pool is None) == (dsn is None): + raise ValueError("postgres_storage needs exactly one of pool= or dsn=") + owned = pool is None + if dsn is not None: + from psycopg_pool import AsyncConnectionPool + + pool = AsyncConnectionPool(dsn, min_size=1, max_size=max_size, open=False) + + from adcp_reporting_storage.inline_storage import PgReportingSealStore, PgReportingStagingStore + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from adcp_reporting_storage.migrations import MIGRATE_COMMAND, pending_migrations + from adcp_reporting_storage.subscriptions.pg import ( + PgPrincipalRecordStore, + PgReportingSubscriptionStateStore, + ) + + staging = PgReportingStagingStore(pool=pool) + opened = False + + async def open_pool() -> None: + nonlocal opened + if owned and not opened: + await pool.open(wait=True) + opened = True + + async def check_schema() -> None: + await open_pool() + pending = await pending_migrations(pool) + if pending: + versions = ", ".join(item.version for item in pending) + stopped = [item.version for item in pending if item.stopped_worker] + note = ( + f"; {', '.join(stopped)} is a stopped-worker upgrade, so stop every reporting " + "worker first" + if stopped + else "" + ) + raise ReliableReportingConfigurationError( + f"the reporting schema is behind this SDK (pending: {versions}). " + f'Run "{MIGRATE_COMMAND}" against this database{note}.', + code="REPORTING_SCHEMA_BEHIND", + ) + await staging.check_ready() + + async def close_pool() -> None: + nonlocal opened + if owned and opened: + await pool.close() + opened = False + + return ReportingStorage( + durability="durable", + ledger=PgReportingLedgerStore(pool=pool, clock=clock), + principals=PgPrincipalRecordStore(pool=pool), + subscriptions=PgReportingSubscriptionStateStore(pool=pool), + staging=staging, + seals=PgReportingSealStore(pool=pool), + check_schema=check_schema, + close=close_pool, + pool=pool, + ) 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 index efcf23f49..dedba4243 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/subscriptions/pg.py @@ -24,6 +24,11 @@ principals=principals, host=..., offerings=..., store=state ) +With the reporting service, prefer ``adcp.storage.postgres.postgres_storage``, +which builds both stores (and the ledger) over one pool and checks the schema +version that ``python -m adcp migrate`` applies; ``create_schema`` remains for +hand-wired compositions. + 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 @@ -70,6 +75,9 @@ def _document(value: Any) -> dict[str, Any]: class _PgDocumentStore: + #: Rows survive restarts and are shared by every worker using the database. + is_durable = True + def __init__(self, *, pool: AsyncConnectionPool) -> None: self._pool = pool @@ -82,6 +90,14 @@ async def create_schema(self) -> None: class PgReportingSubscriptionStateStore(_PgDocumentStore): """Engine state for reporting subscriptions, one CAS document per principal.""" + async def list_principals(self) -> list[str]: + """Every subject with stored state, in subject order (the managed path's roster).""" + async with self._pool.connection() as connection: + cursor = await connection.execute( + "SELECT subject FROM adcp_reporting_subscription_state ORDER BY subject" + ) + return [str(row[0]) for row in await cursor.fetchall()] + async def load(self, subject: str) -> StoredReportingSubscriptionState | None: try: from adcp.reporting.subscriptions import StoredReportingSubscriptionState diff --git a/src/adcp/__main__.py b/src/adcp/__main__.py index c82738541..c38bcabf3 100644 --- a/src/adcp/__main__.py +++ b/src/adcp/__main__.py @@ -683,6 +683,11 @@ def resolve_agent_config(agent_identifier: str) -> dict[str, Any]: def main() -> None: """Main CLI entry point - compatible with JavaScript version.""" + if len(sys.argv) > 1 and sys.argv[1] == "migrate": + # Seller tooling: apply the reporting storage migrations. + from adcp.storage._cli import main as migrate_main + + sys.exit(migrate_main(sys.argv[2:])) parser = argparse.ArgumentParser( description="AdCP Client - Interact with AdCP agents", usage="adcp [options] [tool] [payload]", diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 335b2e7ab..0b745c8a4 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -76,6 +76,7 @@ SubscriptionPlanningFailure, ) from adcp.reporting.subscriptions.managed import ManagedSubscriptions, SubscriptionOptions + from adcp.storage import ReportingStorage __all__ = [ "AdapterRegistration", @@ -492,7 +493,7 @@ class ReliableReportingService: def __init__( self, *, - store: ReportingLedgerStore, + store: ReportingLedgerStore | None = None, account_context: ReportingContextResolver, caller_resolver: ReportingCallerResolver | None = None, consumer_status_enabled: bool = False, @@ -516,6 +517,7 @@ def __init__( initialize_source_storage: Callable[[], Awaitable[None]] | None = None, reporting_subscriptions: ReportingSubscriptionServiceOptions | None = None, subscriptions: SubscriptionOptions | None = None, + storage: ReportingStorage | None = None, ) -> None: """``staging``/``seals`` are the default source stores adapters register with; ``initialize_source_storage`` runs once at startup, after ``store.create_schema()``, @@ -527,7 +529,33 @@ def __init__( runtime, roster, buys, offerings and scheduling from the platform it is installed on. ``reporting_subscriptions`` is the advanced composition, where the host builds and links the engine itself. Without either the - service is unchanged.""" + service is unchanged. + + ``storage`` is the one storage decision (:mod:`adcp.storage`): + ``memory_storage()`` or ``postgres_storage(...)``. It supplies the + ledger ``store``, the default ``staging``/``seals``, and the principal + and subscription-state stores of the managed subscription path. Durable + storage checks its schema version at startup instead of changing it. + Pass ``storage`` or ``store``, not both.""" + if storage is not None: + from adcp.storage import ReportingStorage + + if not isinstance(storage, ReportingStorage): + raise ReliableReportingConfigurationError( + "storage must be memory_storage() or postgres_storage(...)" + ) + if store is not None: + raise ReliableReportingConfigurationError("pass storage or store, not both") + store = storage.ledger + staging = staging if staging is not None else storage.staging + seals = seals if seals is not None else storage.seals + if storage.close is not None: + owned_resources = (*owned_resources, ReportingServiceResource(close=storage.close)) + if store is None: + raise ReliableReportingConfigurationError( + "pass storage=memory_storage() or postgres_storage(...), or a ledger store=" + ) + self._storage = storage if production is not None and ( caller_resolver is not None or producer_factory is not ReportingProducer @@ -621,6 +649,8 @@ def __init__( clock=effective_clock, status_retention_days=status_retention_days, report=self._report_planning_failure, + principal_store=None if storage is None else storage.principals, + state_store=None if storage is None else storage.subscriptions, ) except TypeError as error: raise ReliableReportingConfigurationError(str(error)) from None @@ -1334,10 +1364,33 @@ async def initialize(self) -> None: """Prepare resources once; start/run_worker opens reporting admission.""" await self._lifecycle.initialize() + def _require_durable_subscriptions(self) -> None: + """Refuse process-local reporting-subscription storage outside development.""" + managed = self._managed + if managed is None: + return + from adcp.storage import require_durable + + if self._storage is not None: + require_durable(self._storage.durable, stores="memory_storage()") + return + stores = { + "the reporting ledger": self.store, + "the principal records": getattr(managed.principals, "_store", None), + "the subscription engine state": managed.engine.store, + } + volatile = [name for name, item in stores.items() if not getattr(item, "is_durable", False)] + require_durable(not volatile, stores=", ".join(volatile)) + async def _initialize(self) -> None: self.validate() + self._require_durable_subscriptions() async with self._configuration_lock: - await self.store.create_schema() + if self._storage is not None and self._storage.check_schema is not None: + # Migrations own the schema; startup only refuses one that is behind. + await self._storage.check_schema() + else: + await self.store.create_schema() if self._initialize_source_storage is not None: await self._initialize_source_storage() for configuration in self._pending_configurations.values(): diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index 217b96836..256ca964c 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -482,10 +482,15 @@ def __init__( clock: Callable[[], datetime], status_retention_days: int, report: Callable[[SubscriptionPlanningFailure], Awaitable[None]], + principal_store: PrincipalRecordStore | None = None, + state_store: ReportingSubscriptionStateStore | None = None, ) -> None: if not isinstance(options, SubscriptionOptions): raise TypeError("subscriptions must be SubscriptionOptions") self.options = options + #: The service's storage supplies these when the options do not. + self._principal_store = options.principal_store or principal_store + self._state_store = options.state_store or state_store self._clock = clock self._retention_days = status_retention_days self._report = report @@ -554,7 +559,7 @@ def compose( raise ReliableReportingConfigurationError(str(error)) from None if self._principals is None: self._principals = PrincipalService( - options.principal_store, + self._principal_store, emit_change=options.emit_change, accepted_experimental_features=("protocol.principal",), ) @@ -565,7 +570,7 @@ def compose( offerings=mapped, support=support, status_retention_days=self._retention_days, - store=options.state_store, + store=self._state_store, clock=self._clock, end_account_configs=options.end_account_configs, ) diff --git a/src/adcp/storage/__init__.py b/src/adcp/storage/__init__.py new file mode 100644 index 000000000..86a6e4487 --- /dev/null +++ b/src/adcp/storage/__init__.py @@ -0,0 +1,128 @@ +"""One storage decision for reporting (AdCP 3.3, experimental). + +``memory_storage()`` or ``postgres_storage(pool=... | dsn=...)`` returns a +:class:`ReportingStorage` that holds every reporting store: the ledger, staged +source objects and seals, ``sync_principal`` principal records and the +reporting-subscription engine state (grant journal, exceptions, version +snapshots). Pass it to ``ReliableReportingService(storage=...)``. + +* **Durability is a property of the storage**, never a flag: + ``memory_storage()`` is ``process-local`` and ``postgres_storage`` is + ``durable``. +* **Process-local storage is refused in production** for reporting + subscriptions, with ``REPORTING_STORAGE_NOT_DURABLE``. It is allowed under + pytest and when ``ADCP_ENV`` is ``development`` or ``test``. +* **Migrations are versioned files**, applied by ``python -m adcp migrate``; + durable storage refuses to start on a schema that is behind + (``REPORTING_SCHEMA_BEHIND``). + +``postgres_storage`` ships in the ``adcp-reporting-storage`` distribution and +is re-exported here and from :mod:`adcp.storage.postgres`. +""" + +from __future__ import annotations + +import os +import sys +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Literal + +__all__ = [ + "DEVELOPMENT_ENVIRONMENTS", + "ReportingStorage", + "development_environment", + "memory_storage", + "postgres_storage", + "require_durable", +] + +#: ``ADCP_ENV`` values under which process-local storage is allowed. +DEVELOPMENT_ENVIRONMENTS = frozenset({"development", "test"}) + + +@dataclass(frozen=True) +class ReportingStorage: + """Every reporting store, chosen together. + + Built by :func:`memory_storage` or :func:`postgres_storage`. ``ledger`` + is the reporting ledger store, ``principals`` the principal-record store, + ``subscriptions`` the reporting-subscription engine-state store, and + ``staging``/``seals`` the default source stores adapters register with. + ``check_schema`` runs at startup and refuses a schema that is behind; + ``close`` releases what the storage opened. + """ + + durability: Literal["process-local", "durable"] + ledger: Any + principals: Any + subscriptions: Any + staging: Any = None + seals: Any = None + check_schema: Callable[[], Awaitable[None]] | None = None + close: Callable[[], Awaitable[None]] | None = None + pool: Any = None + + def __post_init__(self) -> None: + if self.durability not in {"process-local", "durable"}: + raise ValueError("durability is process-local or durable") + + @property + def durable(self) -> bool: + return self.durability == "durable" + + +def memory_storage(*, clock: Callable[[], datetime] | None = None) -> ReportingStorage: + """Process-local storage for tests and development; lost on restart.""" + from adcp.reporting.inline_source import InMemorySealStore, InMemoryStagingStore + from adcp.reporting.ledger import InMemoryReportingLedgerStore + from adcp.reporting.subscriptions.engine import InMemoryReportingSubscriptionStore + from adcp.server.principal import InMemoryPrincipalRecordStore + + return ReportingStorage( + durability="process-local", + ledger=InMemoryReportingLedgerStore(clock=clock), + principals=InMemoryPrincipalRecordStore(), + subscriptions=InMemoryReportingSubscriptionStore(), + staging=InMemoryStagingStore(), + seals=InMemorySealStore(), + ) + + +def postgres_storage( + *, + pool: Any = None, + dsn: str | None = None, + clock: Callable[[], datetime] | None = None, + max_size: int = 10, +) -> ReportingStorage: + """Durable storage over PostgreSQL; see :mod:`adcp_reporting_storage.storage`.""" + try: + from adcp_reporting_storage.storage import postgres_storage as build + except ImportError as error: # pragma: no cover - optional distribution + raise ImportError( + 'postgres_storage ships in adcp-reporting-storage: pip install "adcp-reporting-storage"' + ) from error + return build(pool=pool, dsn=dsn, clock=clock, max_size=max_size) # type: ignore[no-any-return] + + +def development_environment() -> bool: + """Whether process-local storage is allowed: pytest, or ``ADCP_ENV`` development/test.""" + if os.environ.get("ADCP_ENV", "").strip().lower() in DEVELOPMENT_ENVIRONMENTS: + return True + return "pytest" in sys.modules or "PYTEST_CURRENT_TEST" in os.environ + + +def require_durable(durable: bool, *, stores: str) -> None: + """Refuse process-local reporting-subscription storage outside development.""" + if durable or development_environment(): + return + from adcp.reporting._errors import ReliableReportingConfigurationError + + raise ReliableReportingConfigurationError( + f"reporting subscriptions keep obligations, grants and frozen coverage in {stores}, " + "which do not survive a restart. Pass storage=postgres_storage(dsn=...) to " + "ReliableReportingService, or set ADCP_ENV=development to run in memory.", + code="REPORTING_STORAGE_NOT_DURABLE", + ) diff --git a/src/adcp/storage/_cli.py b/src/adcp/storage/_cli.py new file mode 100644 index 000000000..1fab46d44 --- /dev/null +++ b/src/adcp/storage/_cli.py @@ -0,0 +1,65 @@ +"""``python -m adcp migrate``: apply the reporting storage migrations.""" + +from __future__ import annotations + +import argparse +import asyncio +import os +import sys +from collections.abc import Sequence + +USAGE = "python -m adcp migrate [--dsn URL] [--status | --dry-run]" + + +def main(argv: Sequence[str]) -> int: + parser = argparse.ArgumentParser( + prog="python -m adcp migrate", + usage=USAGE, + description=( + "Apply the reporting storage migrations (adcp-reporting-storage) to a PostgreSQL " + "database, in order. Each migration commits on its own under the reporting schema " + "lock. A stopped-worker upgrade needs every reporting worker stopped first." + ), + ) + parser.add_argument( + "--dsn", + default=os.environ.get("ADCP_DATABASE_URL") or os.environ.get("DATABASE_URL"), + help="PostgreSQL URL; defaults to $ADCP_DATABASE_URL, then $DATABASE_URL. Select the " + "schema with options=-csearch_path=.", + ) + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--status", action="store_true", help="List applied and pending migrations") + mode.add_argument("--dry-run", action="store_true", help="Print what would be applied") + args = parser.parse_args(list(argv)) + if not args.dsn: + parser.error("pass --dsn or set DATABASE_URL") + try: + return asyncio.run(_run(args.dsn, status=args.status, dry_run=args.dry_run)) + except ImportError as error: + print( + f"{error}. Install the PostgreSQL storage: pip install 'adcp-reporting-storage[pg]'", + file=sys.stderr, + ) + return 2 + + +async def _run(dsn: str, *, status: bool, dry_run: bool) -> int: + from adcp_reporting_storage.migrations import describe, migrate, migration_status + from psycopg import AsyncConnection + + async with await AsyncConnection.connect(dsn, autocommit=True) as connection: + if status: + statuses = await migration_status(connection) + for line in describe(statuses): + print(line) + pending = [item for item in statuses if item.pending] + print(f"{len(pending)} pending" if pending else "up to date") + return 0 + planned = await migrate(connection, dry_run=dry_run) + verb = "would apply" if dry_run else "applied" + for item in planned: + note = " (stopped-worker upgrade)" if item.stopped_worker else "" + print(f"{verb} {item.version} {item.name}{note}") + if not planned: + print("up to date") + return 0 diff --git a/src/adcp/storage/postgres.py b/src/adcp/storage/postgres.py new file mode 100644 index 000000000..3cdb0415f --- /dev/null +++ b/src/adcp/storage/postgres.py @@ -0,0 +1,7 @@ +"""``postgres_storage``, re-exported from the ``adcp-reporting-storage`` distribution.""" + +from __future__ import annotations + +from adcp.storage import postgres_storage + +__all__ = ["postgres_storage"] diff --git a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py new file mode 100644 index 000000000..ae8e74a16 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py @@ -0,0 +1,148 @@ +"""Versioned migrations and ``postgres_storage`` against real PostgreSQL (AdCP 3.3).""" + +from __future__ import annotations + +import pytest + +from adcp.reporting._errors import ReliableReportingConfigurationError +from tests.conformance.reporting._generation_support import isolated_reporting_pool + + +@pytest.mark.asyncio +async def test_migrate_applies_every_version_once_and_status_reads_it() -> None: + from adcp_reporting_storage.migrations import ( + describe, + migrate, + migration_status, + migrations, + pending_migrations, + ) + + async with isolated_reporting_pool(autocommit=True) as pool: + assert [item.version for item in await pending_migrations(pool)] == [ + item.version for item in migrations() + ] + planned = await migrate(pool, dry_run=True) + assert planned and await pending_migrations(pool) == planned # a dry run changes nothing + applied = await migrate(pool) + assert [item.version for item in applied] == [item.version for item in migrations()] + assert await pending_migrations(pool) == [] + assert await migrate(pool) == [] # idempotent + lines = describe(await migration_status(pool)) + assert all(": applied " in line for line in lines) + assert lines[1].endswith("[stopped-worker upgrade]") + + +@pytest.mark.asyncio +async def test_storage_refuses_a_schema_that_is_behind_and_starts_once_migrated() -> None: + from adcp_reporting_storage.migrations import migrate + + from adcp.storage.postgres import postgres_storage + + async with isolated_reporting_pool(autocommit=True) as pool: + storage = postgres_storage(pool=pool) + assert storage.durable + assert storage.check_schema is not None + with pytest.raises(ReliableReportingConfigurationError) as refused: + await storage.check_schema() + error = refused.value + assert error.code == "REPORTING_SCHEMA_BEHIND" and error.recovery == "correctable" + assert "pending: 0001, 0002, 0003, 0004, 0005" in str(error) + assert "python -m adcp migrate" in str(error) + assert "0002 is a stopped-worker upgrade" in str(error) + await migrate(pool) + await storage.check_schema() + + +@pytest.mark.asyncio +async def test_the_subscription_stores_list_principals_and_declare_durability() -> None: + from adcp_reporting_storage.migrations import migrate + + from adcp.storage.postgres import postgres_storage + + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) + storage = postgres_storage(pool=pool) + state = storage.subscriptions + assert state.is_durable and storage.principals.is_durable and storage.ledger.is_durable + assert await state.list_principals() == [] + assert await state.compare_and_swap("principal-b", None, {"document_version": 1}) + assert await state.compare_and_swap("principal-a", None, {"document_version": 1}) + assert await state.list_principals() == ["principal-a", "principal-b"] + + +@pytest.mark.asyncio +async def test_the_cli_migrates_and_reports_status(capsys) -> None: + import os + + from adcp.storage._cli import _run + + async with isolated_reporting_pool(autocommit=True) as pool: + async with pool.connection() as connection: + cursor = await connection.execute("SELECT current_schema()") + schema = (await cursor.fetchone())[0] + base = os.environ["ADCP_PG_TEST_URL"] + dsn = f"{base}{'&' if '?' in base else '?'}options=-csearch_path%3D{schema}" + assert await _run(dsn, status=True, dry_run=False) == 0 + status = capsys.readouterr().out + assert "0002 reporting_subscription_generations: pending [stopped-worker upgrade" in status + assert "5 pending" in status + assert await _run(dsn, status=False, dry_run=False) == 0 + assert "applied 0002 reporting_subscription_generations (stopped-worker upgrade)" in ( + capsys.readouterr().out + ) + assert await _run(dsn, status=True, dry_run=False) == 0 + assert capsys.readouterr().out.strip().endswith("up to date") + + +@pytest.mark.asyncio +async def test_a_managed_service_runs_on_postgres_storage() -> None: + from datetime import timedelta + + from adcp_reporting_storage.migrations import migrate + + from adcp.reporting import PrincipalRef, SubscriptionOptions + from adcp.reporting.service import ReliableReportingService + from adcp.reporting.testing import DeterministicReportingClock + from adcp.storage.postgres import postgres_storage + from tests.test_reliable_reporting_service import DEFINITION + from tests.test_reporting_subscription_managed import ( + AGENCY, + OFFERING, + START, + SUBSCRIPTION, + Seller, + _ByBuySource, + _context, + _ctx, + ) + + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) + clock = DeterministicReportingClock(START) + service = ReliableReportingService( + storage=postgres_storage(pool=pool, clock=clock), + account_context=_context, + clock=clock, + read_jitter_window=timedelta(0), + subscriptions=SubscriptionOptions(definitions={OFFERING["offering_id"]: DEFINITION}), + ) + service.sources.register("gam", _ByBuySource(), capability_offerings=[OFFERING]) + platform = service.install(Seller()) + try: + response = await platform.sync_principal( + { + "idempotency_key": "pg-storage-key-000001", + "configuration": {"reporting_subscriptions": [SUBSCRIPTION]}, + }, + _ctx(), + ) + assert response.result.kind == "applied" + clock.set(START + timedelta(hours=3, minutes=30)) + turn = await service.run_worker() + assert turn.subscription_errors == {} + assert turn.subscription_obligations_written == 2 + engine = service._managed.engine + assert await engine.list_principals() == [PrincipalRef(None, AGENCY)] + finally: + await service.close() diff --git a/tests/test_reporting_storage_facade.py b/tests/test_reporting_storage_facade.py new file mode 100644 index 000000000..66749c409 --- /dev/null +++ b/tests/test_reporting_storage_facade.py @@ -0,0 +1,139 @@ +"""One storage decision: ``memory_storage()`` / ``postgres_storage(...)`` (AdCP 3.3).""" + +from __future__ import annotations + +from datetime import timedelta + +import pytest + +from adcp.reporting import SubscriptionOptions +from adcp.reporting.service import ReliableReportingConfigurationError, ReliableReportingService +from adcp.storage import ReportingStorage, memory_storage +from tests.test_reliable_reporting_service import DEFINITION +from tests.test_reporting_subscription_managed import ( + OFFERING, + START, + SUBSCRIPTION, + Seller, + _ByBuySource, + _context, + _ctx, +) + + +def _service(storage: ReportingStorage, **kwargs) -> ReliableReportingService: + from adcp.reporting.testing import DeterministicReportingClock + + service = ReliableReportingService( + storage=storage, + account_context=_context, + clock=DeterministicReportingClock(START), + read_jitter_window=timedelta(0), + subscriptions=SubscriptionOptions(definitions={OFFERING["offering_id"]: DEFINITION}), + **kwargs, + ) + service.sources.register("gam", _ByBuySource(), capability_offerings=[OFFERING]) + return service + + +def test_memory_storage_is_process_local() -> None: + storage = memory_storage() + assert storage.durability == "process-local" and not storage.durable + assert storage.check_schema is None and storage.close is None + + +@pytest.mark.asyncio +async def test_one_storage_object_backs_every_store() -> None: + storage = memory_storage() + service = _service(storage) + platform = service.install(Seller()) + assert service.store is storage.ledger + response = await platform.sync_principal( + { + "idempotency_key": "storage-facade-key-0001", + "configuration": {"reporting_subscriptions": [SUBSCRIPTION]}, + }, + _ctx(), + ) + assert response.result.kind == "applied" + # The principal record and the engine state live in the storage's stores. + assert await storage.principals.get(_ctx().caller_identity) is not None + assert await storage.subscriptions.list_principals() == [_ctx().caller_identity] + await service.run_worker() + + +@pytest.mark.asyncio +async def test_process_local_storage_is_refused_in_production(monkeypatch) -> None: + monkeypatch.setattr("adcp.storage.development_environment", lambda: False) + service = _service(memory_storage()) + service.install(Seller()) + with pytest.raises(ReliableReportingConfigurationError) as refused: + await service.initialize() + error = refused.value + assert error.code == "REPORTING_STORAGE_NOT_DURABLE" and error.recovery == "correctable" + assert "postgres_storage" in str(error) and "ADCP_ENV=development" in str(error) + + +@pytest.mark.asyncio +async def test_in_memory_stores_without_storage_are_refused_too(monkeypatch) -> None: + monkeypatch.setattr("adcp.storage.development_environment", lambda: False) + service = ReliableReportingService.memory( + account_context=_context, + subscriptions=SubscriptionOptions(definitions={OFFERING["offering_id"]: DEFINITION}), + ) + service.sources.register("gam", _ByBuySource(), capability_offerings=[OFFERING]) + service.install(Seller()) + with pytest.raises(ReliableReportingConfigurationError) as refused: + await service.initialize() + assert refused.value.code == "REPORTING_STORAGE_NOT_DURABLE" + assert "the reporting ledger" in str(refused.value) + + +@pytest.mark.parametrize("env", ["development", "test"]) +def test_adcp_env_allows_process_local_storage(monkeypatch, env: str) -> None: + import sys + + from adcp.storage import development_environment + + monkeypatch.setenv("ADCP_ENV", env) + assert development_environment() + monkeypatch.setenv("ADCP_ENV", "production") + monkeypatch.delitem(sys.modules, "pytest") + monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False) + assert not development_environment() + + +def test_storage_and_store_are_exclusive() -> None: + storage = memory_storage() + with pytest.raises(ReliableReportingConfigurationError, match="not both"): + ReliableReportingService(storage=storage, store=storage.ledger, account_context=_context) + with pytest.raises(ReliableReportingConfigurationError, match="storage="): + ReliableReportingService(account_context=_context) + + +def test_migrations_are_numbered_files_with_their_upgrade_notes() -> None: + pytest.importorskip("adcp_reporting_storage") + from adcp_reporting_storage.migrations import MigrationStatus, describe, migrations + + shipped = migrations() + assert [item.version for item in shipped] == ["0001", "0002", "0003", "0004", "0005"] + widened = shipped[1] + assert widened.name == "reporting_subscription_generations" and widened.stopped_worker + assert widened.includes == ("ledger/reporting_ledger_account_generations.sql",) + assert "stop every reporting worker" in widened.description + assert not any(item.stopped_worker for item in shipped if item is not widened) + for item in shipped: + assert item.statements(), item.version # every include resolves + lines = describe([MigrationStatus(item, None) for item in shipped]) + assert lines[1] == ( + "0002 reporting_subscription_generations: pending [stopped-worker upgrade: stop every " + "reporting worker before migrating]" + ) + + +def test_migrate_cli_requires_a_database(capsys) -> None: + from adcp.storage._cli import main + + with pytest.raises(SystemExit): + main(["--status", "--dsn", ""]) + assert "DATABASE_URL" in capsys.readouterr().err From 7dc6905e75b8d9c4cc346207dc16faaa43dcb115 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 21:35:20 +0000 Subject: [PATCH 2/2] fix(reporting-storage): keep adcp free of imports of the storage distribution The storage boundary says adcp never imports adcp-reporting-storage. The distribution now registers its pieces as entry points that adcp discovers generically: - adcp.storage:postgres resolves adcp.storage.postgres_storage, which adcp.storage.postgres still re-exports; - adcp.commands:migrate serves python -m adcp migrate. The CLI moves to adcp_reporting_storage.migrations.cli, also runnable as python -m adcp_reporting_storage.migrations.cli. postgres_storage reaches adcp only through public names. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../adcp-reporting-storage/pyproject.toml | 7 +++++ .../adcp_reporting_storage/migrations/cli.py | 14 +++++++-- .../src/adcp_reporting_storage/storage.py | 2 +- src/adcp/__main__.py | 10 +++++-- src/adcp/storage/__init__.py | 30 ++++++++++++++----- .../test_reporting_storage_migrations_pg.py | 2 +- tests/test_reporting_storage_facade.py | 25 +++++++++++++++- 7 files changed, 75 insertions(+), 15 deletions(-) rename src/adcp/storage/_cli.py => packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/cli.py (88%) diff --git a/packages/adcp-reporting-storage/pyproject.toml b/packages/adcp-reporting-storage/pyproject.toml index 60b4a6a60..e4dc08975 100644 --- a/packages/adcp-reporting-storage/pyproject.toml +++ b/packages/adcp-reporting-storage/pyproject.toml @@ -63,6 +63,13 @@ azure = [ "azure-storage-blob>=12.20", ] +# adcp discovers these without importing this distribution by name. +[project.entry-points."adcp.storage"] +postgres = "adcp_reporting_storage.storage:postgres_storage" + +[project.entry-points."adcp.commands"] +migrate = "adcp_reporting_storage.migrations.cli:main" + [project.urls] Homepage = "https://github.com/adcontextprotocol/adcp-client-python" Documentation = "https://github.com/adcontextprotocol/adcp-client-python/blob/main/packages/adcp-reporting-storage/README.md" diff --git a/src/adcp/storage/_cli.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/cli.py similarity index 88% rename from src/adcp/storage/_cli.py rename to packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/cli.py index 1fab46d44..0f02de60a 100644 --- a/src/adcp/storage/_cli.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/cli.py @@ -1,4 +1,9 @@ -"""``python -m adcp migrate``: apply the reporting storage migrations.""" +"""``python -m adcp migrate``: apply the reporting storage migrations. + +Registered as the ``migrate`` entry point of the ``adcp.commands`` group, which +``python -m adcp`` dispatches to, so ``adcp`` never imports this distribution. +Also runnable as ``python -m adcp_reporting_storage.migrations.cli``. +""" from __future__ import annotations @@ -44,9 +49,10 @@ def main(argv: Sequence[str]) -> int: async def _run(dsn: str, *, status: bool, dry_run: bool) -> int: - from adcp_reporting_storage.migrations import describe, migrate, migration_status from psycopg import AsyncConnection + from adcp_reporting_storage.migrations import describe, migrate, migration_status + async with await AsyncConnection.connect(dsn, autocommit=True) as connection: if status: statuses = await migration_status(connection) @@ -63,3 +69,7 @@ async def _run(dsn: str, *, status: bool, dry_run: bool) -> int: if not planned: print("up to date") return 0 + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main(sys.argv[1:])) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py index f5887cbf0..7498112f7 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py @@ -44,7 +44,7 @@ def postgres_storage( ``search_path`` selects the schema, or a ``dsn`` for a pool the storage opens at startup and closes at shutdown. """ - from adcp.reporting._errors import ReliableReportingConfigurationError + from adcp.reporting import ReliableReportingConfigurationError from adcp.storage import ReportingStorage if (pool is None) == (dsn is None): diff --git a/src/adcp/__main__.py b/src/adcp/__main__.py index c38bcabf3..04a731a58 100644 --- a/src/adcp/__main__.py +++ b/src/adcp/__main__.py @@ -684,9 +684,15 @@ def resolve_agent_config(agent_identifier: str) -> dict[str, Any]: def main() -> None: """Main CLI entry point - compatible with JavaScript version.""" if len(sys.argv) > 1 and sys.argv[1] == "migrate": - # Seller tooling: apply the reporting storage migrations. - from adcp.storage._cli import main as migrate_main + # Seller tooling: apply the reporting storage migrations, a command the + # adcp-reporting-storage distribution registers. + from adcp.storage import _plugin + try: + migrate_main = _plugin("adcp.commands", "migrate") + except ImportError as error: + print(error, file=sys.stderr) + sys.exit(2) sys.exit(migrate_main(sys.argv[2:])) parser = argparse.ArgumentParser( description="AdCP Client - Interact with AdCP agents", diff --git a/src/adcp/storage/__init__.py b/src/adcp/storage/__init__.py index 86a6e4487..245f3b82f 100644 --- a/src/adcp/storage/__init__.py +++ b/src/adcp/storage/__init__.py @@ -97,14 +97,28 @@ def postgres_storage( clock: Callable[[], datetime] | None = None, max_size: int = 10, ) -> ReportingStorage: - """Durable storage over PostgreSQL; see :mod:`adcp_reporting_storage.storage`.""" - try: - from adcp_reporting_storage.storage import postgres_storage as build - except ImportError as error: # pragma: no cover - optional distribution - raise ImportError( - 'postgres_storage ships in adcp-reporting-storage: pip install "adcp-reporting-storage"' - ) from error - return build(pool=pool, dsn=dsn, clock=clock, max_size=max_size) # type: ignore[no-any-return] + """Durable storage over PostgreSQL, from the ``adcp-reporting-storage`` distribution. + + Resolved through the distribution's ``adcp.storage`` entry point, so + ``adcp`` never imports it by name. + """ + build = _plugin("adcp.storage", "postgres") + storage = build(pool=pool, dsn=dsn, clock=clock, max_size=max_size) + if not isinstance(storage, ReportingStorage): + raise TypeError("the adcp.storage postgres entry point must return ReportingStorage") + return storage + + +def _plugin(group: str, name: str) -> Any: + """Load an entry point a companion distribution registers.""" + from importlib.metadata import entry_points + + for entry in entry_points(group=group): + if entry.name == name: + return entry.load() + raise ImportError( + f"{group}:{name} ships in adcp-reporting-storage: pip install 'adcp-reporting-storage[pg]'" + ) def development_environment() -> bool: diff --git a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py index ae8e74a16..785269eaf 100644 --- a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py @@ -75,7 +75,7 @@ async def test_the_subscription_stores_list_principals_and_declare_durability() async def test_the_cli_migrates_and_reports_status(capsys) -> None: import os - from adcp.storage._cli import _run + from adcp_reporting_storage.migrations.cli import _run async with isolated_reporting_pool(autocommit=True) as pool: async with pool.connection() as connection: diff --git a/tests/test_reporting_storage_facade.py b/tests/test_reporting_storage_facade.py index 66749c409..da623f694 100644 --- a/tests/test_reporting_storage_facade.py +++ b/tests/test_reporting_storage_facade.py @@ -132,8 +132,31 @@ def test_migrations_are_numbered_files_with_their_upgrade_notes() -> None: def test_migrate_cli_requires_a_database(capsys) -> None: - from adcp.storage._cli import main + from adcp_reporting_storage.migrations.cli import main with pytest.raises(SystemExit): main(["--status", "--dsn", ""]) assert "DATABASE_URL" in capsys.readouterr().err + + +def test_python_m_adcp_migrate_dispatches_to_the_storage_distribution(monkeypatch, capsys) -> None: + pytest.importorskip("adcp_reporting_storage") + import sys + + from adcp.__main__ import main + + monkeypatch.delenv("DATABASE_URL", raising=False) + monkeypatch.delenv("ADCP_DATABASE_URL", raising=False) + monkeypatch.setattr(sys, "argv", ["adcp", "migrate", "--status"]) + with pytest.raises(SystemExit): + main() + assert "DATABASE_URL" in capsys.readouterr().err + + +def test_postgres_storage_resolves_through_the_entry_point() -> None: + pytest.importorskip("adcp_reporting_storage") + from adcp_reporting_storage.storage import postgres_storage as direct + + from adcp.storage import _plugin + + assert _plugin("adcp.storage", "postgres") is direct