feat(reporting-storage): one storage decision with versioned migrations

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.

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"
@@ -85,6 +92,7 @@ adcp_reporting_storage = [
     "production/*.json",
     "production/*.sql",
     "subscriptions/*.sql",
+    "migrations/*.sql",
 ]
 
 # Inside this repository the dependency is the checkout, not PyPI. uv ignores

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

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

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

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

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

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: <path>`` 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

packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/cli.py

@@ -0,0 +1,75 @@
+"""``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
+
+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=<schema>.",
+    )
+    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 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)
+            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
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(main(sys.argv[1:]))